Module 8: Capstone Project — Production-Ready AI System

6. Performance Baselines

Next →

Description

"The system is fast" is not useful information. "p50 latency: 1.8s, p99: 6.2s, cost per request: $0.015, error rate: 0.3%" — that is. Performance baselines are numeric commitments that you define before launch and verify continuously. In this capsule you'll learn how to measure, document, and monitor the metrics that really matter in your AI app. When you finish, you'll have a working benchmark script and a BASELINES.md file with real numbers. Think of the baselines as your personal contract with your users: you promise certain levels of latency and cost, and the benchmark tells you whether you're meeting them before each deploy.

The metrics that matter in AI apps

AI apps have different performance metrics than traditional web apps:

Traditional web apps:
  - Latency p50/p99
  - Requests per second
  - Error rate

AI apps (in addition to the above):
  - Cost per request (the LLM costs money per token)
  - Token usage (input + output tokens per request)
  - Guardrail activation rate (how often are requests blocked?)
  - Fallback activation rate (how often is the secondary provider used?)
  - Parse failure rate (how often does the LLM give malformed output?)
  - Model-specific latency (gpt-4o vs gpt-4o-mini have very different latencies)

The benchmark script

# scripts/benchmark.py
"""
Runs a benchmark of the system and establishes performance baselines.

Usage:
  python scripts/benchmark.py
  python scripts/benchmark.py --n 50 --url http://staging.example.com
  python scripts/benchmark.py --compare docs/BASELINES.md  # Compare with a previous baseline
"""
import time
import json
import statistics
import argparse
import sys
from typing import Optional
from pathlib import Path
import urllib.request
import urllib.error

# ─── Varied test inputs for a representative benchmark ───
BENCHMARK_INPUTS = [
    "This product is absolutely amazing! I love everything about it.",
    "Terrible experience. The product broke after two days.",
    "It's okay, I guess. Not great, not bad.",
    "I'm so happy with this purchase! Best decision ever!",
    "Waste of money. Don't buy this.",
    "Pretty good for the price. Would recommend.",
    "The worst customer service I've ever experienced.",
    "Arrived on time and works as described.",
    "Completely disappointed with the quality.",
    "Exceeded my expectations in every way!",
]

def run_request(base_url: str, text: str, timeout: int = 30) -> Optional[dict]:
    """
    Makes a request to the endpoint and returns the metrics.
    Returns None if the request fails.
    """
    data = json.dumps({"text": text}).encode()
    req = urllib.request.Request(
        f"{base_url}/api/v1/analyze",
        data=data,
        headers={"Content-Type": "application/json"}
    )
    
    start = time.monotonic()
    try:
        response = urllib.request.urlopen(req, timeout=timeout)
        elapsed_ms = (time.monotonic() - start) * 1000
        body = json.loads(response.read())
        
        return {
            "success": True,
            "latency_ms": elapsed_ms,
            "status_code": response.status,
            "sentiment": body.get("sentiment"),
            "score": body.get("score"),
            "confidence": body.get("confidence"),
            "degraded": body.get("degraded", False),
            "cost_usd": body.get("cost_usd", 0),  # If the API exposes it
        }
    except urllib.error.HTTPError as e:
        elapsed_ms = (time.monotonic() - start) * 1000
        return {
            "success": False,
            "latency_ms": elapsed_ms,
            "status_code": e.code,
            "error": str(e),
        }
    except Exception as e:
        elapsed_ms = (time.monotonic() - start) * 1000
        return {
            "success": False,
            "latency_ms": elapsed_ms,
            "status_code": 0,
            "error": str(e),
        }

def calculate_percentile(data: list[float], p: int) -> float:
    """Calculates the p percentile of a list of data."""
    if not data:
        return 0.0
    sorted_data = sorted(data)
    index = int(len(sorted_data) * p / 100)
    return sorted_data[min(index, len(sorted_data) - 1)]

def run_benchmark(
    base_url: str = "http://localhost:8000",
    n_requests: int = 20,
    verbose: bool = True
) -> dict:
    """
    Runs the complete benchmark and returns the calculated metrics.
    """
    if verbose:
        print(f"\n🔍 Running benchmark: {n_requests} requests to {base_url}")
        print(f"   Using {len(BENCHMARK_INPUTS)} different input variations\n")
    
    results = []
    
    for i in range(n_requests):
        text = BENCHMARK_INPUTS[i % len(BENCHMARK_INPUTS)]
        result = run_request(base_url, text)
        results.append(result)
        
        if verbose:
            if result["success"]:
                print(f"  [{i+1}/{n_requests}] ✅ {result['latency_ms']:.0f}ms — {result.get('sentiment', '?')} (score: {result.get('score', '?')})")
            else:
                print(f"  [{i+1}/{n_requests}] ❌ {result['latency_ms']:.0f}ms — Error {result['status_code']}: {result.get('error', '')[:60]}")
    
    # Separate successes and failures
    successes = [r for r in results if r["success"]]
    failures = [r for r in results if not r["success"]]
    
    latencies = [r["latency_ms"] for r in successes]
    costs = [r.get("cost_usd", 0) for r in successes]
    degraded = [r for r in successes if r.get("degraded")]
    
    metrics = {
        "total_requests": n_requests,
        "successful_requests": len(successes),
        "failed_requests": len(failures),
        "error_rate_percent": round(len(failures) / n_requests * 100, 2),
        "degraded_rate_percent": round(len(degraded) / max(len(successes), 1) * 100, 2),
    }
    
    if latencies:
        metrics.update({
            "latency_p50_ms": round(calculate_percentile(latencies, 50), 1),
            "latency_p90_ms": round(calculate_percentile(latencies, 90), 1),
            "latency_p99_ms": round(calculate_percentile(latencies, 99), 1),
            "latency_min_ms": round(min(latencies), 1),
            "latency_max_ms": round(max(latencies), 1),
            "latency_mean_ms": round(statistics.mean(latencies), 1),
        })
    
    if any(c > 0 for c in costs):
        metrics.update({
            "cost_per_request_usd": round(statistics.mean(costs), 6),
            "cost_total_usd": round(sum(costs), 6),
            "cost_p99_usd": round(calculate_percentile(costs, 99), 6),
        })
    
    # Sentiment distribution
    sentiments = [r.get("sentiment") for r in successes if r.get("sentiment")]
    if sentiments:
        from collections import Counter
        sentiment_dist = dict(Counter(sentiments))
        metrics["sentiment_distribution"] = sentiment_dist
    
    return metrics

def compare_with_baseline(metrics: dict, baseline_file: str) -> list[str]:
    """
    Compares the current metrics with the documented baseline.
    Returns a list of violations.
    """
    violations = []
    
    # Read the baselines from the Markdown file
    baseline_path = Path(baseline_file)
    if not baseline_path.exists():
        return ["BASELINES.md not found — cannot compare"]
    
    content = baseline_path.read_text()
    
    # Extract the targets from the markdown (simple format)
    # In a real system, the baselines would be in JSON
    # Here we use hardcoded values as an example
    targets = {
        "latency_p50_ms": 2000,
        "latency_p99_ms": 10000,
        "error_rate_percent": 1.0,
        "degraded_rate_percent": 10.0,
    }
    
    for metric, target in targets.items():
        if metric in metrics:
            value = metrics[metric]
            if value > target:
                violations.append(
                    f"⚠️  {metric}: {value} exceeds target {target}"
                )
    
    return violations

def print_metrics(metrics: dict):
    """Prints the metrics in a readable way."""
    print("\n" + "=" * 50)
    print("BENCHMARK RESULTS")
    print("=" * 50)
    
    print(f"\n📊 Volume:")
    print(f"   Total requests:  {metrics['total_requests']}")
    print(f"   Successful:      {metrics['successful_requests']}")
    print(f"   Failed:          {metrics['failed_requests']}")
    print(f"   Error rate:      {metrics['error_rate_percent']}%")
    print(f"   Degraded rate:   {metrics['degraded_rate_percent']}%")
    
    if "latency_p50_ms" in metrics:
        print(f"\n⏱️  Latency:")
        print(f"   p50:  {metrics['latency_p50_ms']}ms")
        print(f"   p90:  {metrics['latency_p90_ms']}ms")
        print(f"   p99:  {metrics['latency_p99_ms']}ms")
        print(f"   mean: {metrics['latency_mean_ms']}ms")
        print(f"   min:  {metrics['latency_min_ms']}ms")
        print(f"   max:  {metrics['latency_max_ms']}ms")
    
    if "cost_per_request_usd" in metrics:
        print(f"\n💰 Cost:")
        print(f"   Per request (avg): ${metrics['cost_per_request_usd']:.6f}")
        print(f"   Total for benchmark: ${metrics['cost_total_usd']:.4f}")
        daily_estimate = metrics['cost_per_request_usd'] * 10000  # 10k requests/day
        print(f"   Daily estimate (10k req): ${daily_estimate:.2f}")
    
    if "sentiment_distribution" in metrics:
        print(f"\n🎭 Sentiment distribution:")
        for sentiment, count in metrics["sentiment_distribution"].items():
            print(f"   {sentiment}: {count}")
    
    print("=" * 50)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Performance benchmark")
    parser.add_argument("--url", default="http://localhost:8000")
    parser.add_argument("--n", type=int, default=20, help="Number of requests")
    parser.add_argument("--compare", help="Path to BASELINES.md to compare against")
    parser.add_argument("--quiet", action="store_true")
    args = parser.parse_args()
    
    metrics = run_benchmark(
        base_url=args.url,
        n_requests=args.n,
        verbose=not args.quiet
    )
    
    print_metrics(metrics)
    
    # Save results
    output_path = Path("docs/benchmark_latest.json")
    output_path.parent.mkdir(exist_ok=True)
    with open(output_path, "w") as f:
        json.dump(metrics, f, indent=2)
    print(f"\nResults saved to {output_path}")
    
    # Compare with baseline
    if args.compare:
        violations = compare_with_baseline(metrics, args.compare)
        if violations:
            print("\n⚠️  BASELINE VIOLATIONS:")
            for v in violations:
                print(f"  {v}")
            sys.exit(1)
        else:
            print("\n✅ All metrics within baseline targets")

What to expect depending on the model

DEVELOPMENT (with USE_MOCK_PROVIDER=true):
  p50 latency:    10-50ms (only FastAPI + guardrails overhead)
  cost/request:   $0.00 (there's no real call)
  error rate:     0%

STAGING/PRODUCTION with gpt-4o-mini:
  p50 latency:    500-1500ms
  p99 latency:    3000-8000ms
  cost/request:   ~$0.001-0.003 (depends on the prompt size)
  error rate:     < 0.5% (without failed retries)

STAGING/PRODUCTION with gpt-4o:
  p50 latency:    1000-3000ms
  p99 latency:    5000-15000ms
  cost/request:   ~$0.010-0.030 (depends on the prompt size)
  error rate:     < 0.5%

FACTORS THAT INCREASE LATENCY:
  - Longer prompt (more input tokens = more time)
  - High max_tokens (the model generates more text)
  - Larger model (gpt-4o vs gpt-4o-mini)
  - High load on OpenAI (varies by time of day)
  - Retries on transient errors

Exercises

Exercise 1: Establish your baselines

Run the benchmark with your system in mock mode and then (if you have an API key) with the real model:

# With mock (fast, free):
USE_MOCK_PROVIDER=true python scripts/benchmark.py --n 20

# With real API (uses credits, more representative):
python scripts/benchmark.py --n 20

Document the results in your docs/BASELINES.md.

See interpretation guide

Mock results: shows the overhead of your stack (FastAPI + middleware + guardrails + parsing). If it's > 100ms, there's something slow in your code that isn't the LLM.

Real results: a high p99 (e.g., 8000ms) can indicate:

  • Very long prompts → consider truncating the input
  • Very high max_tokens → check if it's necessary
  • Latency spikes on OpenAI → normal, that's why p99 exists

If the p99 is 10x the p50 (e.g., p50=800ms, p99=8000ms), there's high variability. This is normal for LLM APIs under shared load.

Exercise 2: Detect performance regressions

Modify your scripts/benchmark.py so it accepts a --fail-if-regression flag that:

  1. Reads the baselines from docs/BASELINES.md (or a JSON)
  2. Runs the benchmark
  3. Compares each metric against the baseline
  4. Returns exit code 1 if any metric exceeds the alert limit

Test it by injecting a time.sleep(2) in your mock provider to simulate a regression.

See solution 
# Add to the end of scripts/benchmark.py:

ALERT_THRESHOLDS = {
    "latency_p50_ms": 3000,
    "latency_p99_ms": 15000,
    "error_rate_percent": 3.0,
    "degraded_rate_percent": 10.0,
}

def check_regression(metrics: dict, thresholds: dict = ALERT_THRESHOLDS) -> list[str]:
    """
    Compares current metrics against alert thresholds.
    Returns a list of violations found.
    """
    violations = []
    for metric, threshold in thresholds.items():
        value = metrics.get(metric)
        if value is not None and value > threshold:
            violations.append(
                f"REGRESSION: {metric} = {value} (threshold: {threshold})"
            )
    return violations

# In the if __name__ == "__main__": block
# Add after print_metrics(metrics):
if args.fail_if_regression:
    violations = check_regression(metrics)
    if violations:
        print("\n❌ REGRESSION DETECTED:")
        for v in violations:
            print(f"  {v}")
        sys.exit(1)
    else:
        print("\n✅ No regressions detected — all metrics within thresholds")

To simulate the regression:

# In mock_provider.py (temporarily):
import time

class MockProvider:
    def complete(self, messages, **kwargs) -> str:
        time.sleep(2)  # Simulate 2 seconds of latency
        return '{"sentiment": "positive", "score": 0.9, "confidence": 0.85}'

Run:

python scripts/benchmark.py --n 10 --fail-if-regression
# Must return exit code 1 if p50 > 3000ms
echo $?  # → 1

This pattern is exactly what's used in CI/CD: the benchmark runs automatically and the pipeline fails if there's a regression.

Exercise 3: Monthly cost projection

Write a function project_monthly_cost(cost_per_request: float, requests_per_day: int) -> dict that calculates:

  1. Estimated daily cost
  2. Estimated weekly cost
  3. Estimated monthly cost (30 days)
  4. Estimated yearly cost
  5. A budget_warning flag if the monthly cost exceeds $500

Use the data from your benchmark to feed this function.

See solution 
def project_monthly_cost(cost_per_request: float, requests_per_day: int) -> dict:
    """
    Projects costs based on the average cost per request
    and the expected traffic volume.
    """
    daily = cost_per_request * requests_per_day
    weekly = daily * 7
    monthly = daily * 30
    yearly = daily * 365

    return {
        "cost_per_request_usd": round(cost_per_request, 6),
        "requests_per_day": requests_per_day,
        "daily_usd": round(daily, 2),
        "weekly_usd": round(weekly, 2),
        "monthly_usd": round(monthly, 2),
        "yearly_usd": round(yearly, 2),
        "budget_warning": monthly > 500,
    }

# Example usage with benchmark data:
if __name__ == "__main__":
    benchmark_cost = 0.002  # $0.002 per request (gpt-4o-mini)
    
    scenarios = {
        "Low traffic (1k/day)": project_monthly_cost(benchmark_cost, 1_000),
        "Medium traffic (10k/day)": project_monthly_cost(benchmark_cost, 10_000),
        "High traffic (100k/day)": project_monthly_cost(benchmark_cost, 100_000),
    }
    
    for name, projection in scenarios.items():
        warning = " ⚠️ BUDGET WARNING" if projection["budget_warning"] else ""
        print(f"{name}: ${projection['monthly_usd']}/month{warning}")

# Expected output:
# Low traffic (1k/day): $60.0/month
# Medium traffic (10k/day): $600.0/month ⚠️ BUDGET WARNING
# High traffic (100k/day): $6000.0/month ⚠️ BUDGET WARNING

The projection with gpt-4o (10x more expensive) would be:

  • Low (1k/day): $600/month
  • Medium (10k/day): $6,000/month
  • High (100k/day): $60,000/month

This is why many teams use gpt-4o-mini for most of the traffic and reserve gpt-4o for requests that need higher quality.

Exercise 4: Baselines dashboard in the terminal

Create a script scripts/baselines_dashboard.py that reads your docs/benchmark_latest.json and displays it as an ASCII dashboard with colors (green if within target, red if it exceeds the alert limit):

╔══════════════════════════════════════════╗
║       PRODUCTION AI SYSTEM BASELINES     ║
╠══════════════════════════════════════════╣
║  p50 latency:    1,234 ms    ✅ OK      ║
║  p99 latency:   12,345 ms    ⚠️ ALERT   ║
║  cost/request:  $0.0023      ✅ OK      ║
║  error rate:    0.3%         ✅ OK      ║
║  degraded rate: 1.2%         ✅ OK      ║
╚══════════════════════════════════════════╝
  Last updated: 2026-03-08 14:30:00
See solution 
# scripts/baselines_dashboard.py
"""
Displays a baselines dashboard in the terminal.

Usage:
  python scripts/baselines_dashboard.py
  python scripts/baselines_dashboard.py --file docs/benchmark_latest.json
"""
import json
import sys
from pathlib import Path
from datetime import datetime

TARGETS = {
    "latency_p50_ms": {"target": 2000, "alert": 3000, "label": "p50 latency", "unit": "ms", "fmt": ",.0f"},
    "latency_p99_ms": {"target": 10000, "alert": 15000, "label": "p99 latency", "unit": "ms", "fmt": ",.0f"},
    "cost_per_request_usd": {"target": 0.002, "alert": 0.005, "label": "cost/request", "unit": "$", "fmt": ".4f"},
    "error_rate_percent": {"target": 1.0, "alert": 3.0, "label": "error rate", "unit": "%", "fmt": ".1f"},
    "degraded_rate_percent": {"target": 2.0, "alert": 10.0, "label": "degraded rate", "unit": "%", "fmt": ".1f"},
}

GREEN = "\033[92m"
YELLOW = "\033[93m"
RED = "\033[91m"
RESET = "\033[0m"

def status_color(value: float, target: float, alert: float) -> tuple[str, str]:
    if value <= target:
        return f"{GREEN}✅ OK{RESET}", GREEN
    elif value <= alert:
        return f"{YELLOW}⚠️  WARN{RESET}", YELLOW
    else:
        return f"{RED}❌ ALERT{RESET}", RED

def show_dashboard(metrics_file: str = "docs/benchmark_latest.json"):
    path = Path(metrics_file)
    if not path.exists():
        print(f"❌ File not found: {metrics_file}")
        print("   Run 'python scripts/benchmark.py' first to generate baselines.")
        sys.exit(1)

    metrics = json.loads(path.read_text())

    print("\n╔══════════════════════════════════════════════╗")
    print("║       PRODUCTION AI SYSTEM BASELINES         ║")
    print("╠══════════════════════════════════════════════╣")

    for key, config in TARGETS.items():
        value = metrics.get(key)
        if value is None:
            print(f"║  {config['label']:18s}  {'N/A':>12s}    {'—':8s}  ║")
            continue

        status, color = status_color(value, config["target"], config["alert"])

        if config["unit"] == "$":
            formatted = f"${value:{config['fmt']}}"
        elif config["unit"] == "%":
            formatted = f"{value:{config['fmt']}}%"
        else:
            formatted = f"{value:{config['fmt']}} {config['unit']}"

        print(f"║  {config['label']:18s}  {formatted:>12s}    {status}  ║")

    print("╚══════════════════════════════════════════════╝")

    mod_time = datetime.fromtimestamp(path.stat().st_mtime)
    print(f"  Last updated: {mod_time.strftime('%Y-%m-%d %H:%M:%S')}")
    print()

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--file", default="docs/benchmark_latest.json")
    args = parser.parse_args()
    show_dashboard(args.file)

This dashboard is useful for quickly reviewing before a deploy or at the start of your workday. The key is the colors: green, yellow, and red let you spot problems instantly without having to read numbers.

Troubleshooting

Problem: The benchmark shows latency of 0-5ms (suspiciously fast)

Symptoms:

  • p50: 3ms, p99: 8ms — too fast for a real call to an LLM

Most likely cause: You're running the benchmark against the mock provider without realizing it. Verify that USE_MOCK_PROVIDER isn't set in your .env.

Solution:

# Check if the mock is active:
grep USE_MOCK .env
# If it says USE_MOCK_PROVIDER=true, that's your problem

# Run with the real provider:
USE_MOCK_PROVIDER=false python scripts/benchmark.py --n 5
# Now you should see latencies of 500-3000ms

If you want to measure only your stack's overhead (without the LLM), the mock is correct — but document that those baselines are "stack-only", not "end-to-end".

Problem: The cost_usd field shows as $0.00 in the results

Symptoms:

  • The benchmark runs fine but the cost section shows everything as zero

Most likely cause: Your endpoint isn't exposing the cost_usd field in the JSON response. The benchmark script reads body.get("cost_usd", 0) — if your API doesn't include it, it will always be 0.

Solution: Verify that your endpoint includes the cost in the response:

# In your router (src/app/routers/sentiment.py):
return {
    "sentiment": result["sentiment"],
    "score": result["score"],
    "confidence": result["confidence"],
    "degraded": result.get("degraded", False),
    "cost_usd": result.get("cost_usd", 0),  # ← Make sure to include this
    "request_id": get_request_id(),
}

If you don't want to expose the cost to the end user, you can log it internally and read it from the logs instead of the HTTP response.

Problem: The benchmark hangs or times out on some requests

Symptoms:

  • Some requests finish quickly, others hang for 30+ seconds
  • The script eventually fails with TimeoutError

Most likely cause: OpenAI is experiencing high latency, or your rate limiter is rejecting requests and the retry is waiting.

Solution:

# Reduce the benchmark's timeout:
# In benchmark.py, change the timeout of run_request:
# timeout=30 → timeout=15

# Run with fewer requests to diagnose:
python scripts/benchmark.py --n 5

# Check if the rate limiter is blocking:
tail -20 logs/app.json | jq 'select(.event == "client_rate_limit_rejected")'

If the problem is consistent, check https://status.openai.com to see if there's an active incident.

Problem: The benchmark results vary a lot between runs

Symptoms:

  • First run: p50: 800ms, second run: p50: 2500ms
  • The numbers are never stable

Most likely cause: It's normal. LLM APIs have inherent variability due to server load, routing, and the variable size of the generated output.

Solution:

# Run with more requests for more stable averages:
python scripts/benchmark.py --n 50

# Run 3 times and average:
for i in 1 2 3; do
  python scripts/benchmark.py --n 30 --quiet 2>/dev/null
  echo "--- Run $i complete ---"
done

The rule of thumb: use at least 30 requests per benchmark and run 3 times. Take the median of the 3 runs as your baseline. The p99 will always have high variance — it's normal for it to fluctuate 2-3x between runs.

Summary

  • Baselines = commitments: they're not aspirational — they're what you promise your users
  • The 6 key metrics: p50/p99 latency, cost/request, error rate, guardrail rate, fallback rate
  • Benchmark script: executable before each deploy to detect regressions
  • The p99 matters: the user with the slowest request also exists — not just averages
  • Document in BASELINES.md: to compare evolution over time

Additional resources

  1. Why Percentiles — Why p99 matters more than the average
  2. OpenAI Usage Dashboard — See real costs in the OpenAI dashboard
  3. Locust — For load testing with hundreds of concurrent users
  4. statistics module (Python) — For statistical calculations
← Previous5. Project: Production AI SystemNext →7. Operational Runbook