6. Performance Baselines

Descripción

"El sistema es rápido" no es información útil. "p50 latency: 1.8s, p99: 6.2s, cost per request: $0.015, error rate: 0.3%" — eso sí lo es. Los performance baselines son compromisos numéricos que tú defines antes del lanzamiento y verificas continuamente. En esta cápsula vas a aprender cómo medir, documentar, y monitorear las métricas que realmente importan en tu AI app. Al terminar, vas a tener un benchmark script funcional y un archivo BASELINES.md con números reales. Piensa en los baselines como tu contrato personal con tus usuarios: tú prometes ciertos niveles de latencia y costo, y el benchmark te dice si los estás cumpliendo antes de cada deploy.

Las métricas que importan en AI apps

Las apps AI tienen métricas de performance distintas a las apps web tradicionales:

Apps web tradicionales:
  - Latency p50/p99
  - Requests per second
  - Error rate

Apps AI (además de las anteriores):
  - Cost per request (el LLM cuesta dinero por token)
  - Token usage (input + output tokens por request)
  - Guardrail activation rate (¿con qué frecuencia se bloquean requests?)
  - Fallback activation rate (¿con qué frecuencia se usa el secondary provider?)
  - Parse failure rate (¿con qué frecuencia el LLM da output malformado?)
  - Model-specific latency (gpt-4o vs gpt-4o-mini tienen latencias muy distintas)

El script de benchmark

# scripts/benchmark.py
"""
Ejecuta un benchmark del sistema y establece performance baselines.

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

# ─── Test inputs variados para un benchmark representativo ───
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]:
    """
    Hace un request al endpoint y retorna las métricas.
    Returns None si el request falla.
    """
    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),  # Si el API lo expone
        }
    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:
    """Calcula el percentil p de una lista de datos."""
    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:
    """
    Ejecuta el benchmark completo y retorna las métricas calculadas.
    """
    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]}")
    
    # Separar successes y 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),
        })
    
    # Distribución de sentimientos
    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]:
    """
    Compara las métricas actuales con el baseline documentado.
    Retorna lista de violaciones.
    """
    violations = []
    
    # Leer baselines del archivo Markdown
    baseline_path = Path(baseline_file)
    if not baseline_path.exists():
        return ["BASELINES.md not found — cannot compare"]
    
    content = baseline_path.read_text()
    
    # Extraer targets del markdown (formato simple)
    # En un sistema real, los baselines estarían en JSON
    # Aquí usamos valores hardcoded como ejemplo
    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):
    """Imprime las métricas de forma legible."""
    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)
    
    # Guardar resultados
    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}")
    
    # Comparar con 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")

Qué esperar según el modelo

DESARROLLO (con USE_MOCK_PROVIDER=true):
  p50 latency:    10-50ms (solo overhead de FastAPI + guardrails)
  cost/request:   $0.00 (no hay llamada real)
  error rate:     0%

STAGING/PRODUCCIÓN con gpt-4o-mini:
  p50 latency:    500-1500ms
  p99 latency:    3000-8000ms
  cost/request:   ~$0.001-0.003 (depende del tamaño del prompt)
  error rate:     < 0.5% (sin retries fallidos)

STAGING/PRODUCCIÓN con gpt-4o:
  p50 latency:    1000-3000ms
  p99 latency:    5000-15000ms
  cost/request:   ~$0.010-0.030 (depende del tamaño del prompt)
  error rate:     < 0.5%

FACTORES QUE AUMENTAN LA LATENCIA:
  - Prompt más largo (más tokens de input = más tiempo)
  - max_tokens alto (el modelo genera más texto)
  - Modelo más grande (gpt-4o vs gpt-4o-mini)
  - Alta carga en OpenAI (varía por hora del día)
  - Retries por errors transitorios

Ejercicios

Ejercicio 1: Establecer tus baselines

Ejecuta el benchmark con tu sistema en modo mock y luego (si tienes API key) con el modelo real:

# Con mock (rápido, gratuito):
USE_MOCK_PROVIDER=true python scripts/benchmark.py --n 20

# Con API real (usa créditos, más representativo):
python scripts/benchmark.py --n 20

Documenta los resultados en tu docs/BASELINES.md.

Ejercicio 2: Detectar regresiones de performance

Modifica tu scripts/benchmark.py para que acepte un flag --fail-if-regression que:

1. Lee los baselines de docs/BASELINES.md (o un JSON)
2. Ejecuta el benchmark
3. Compara cada métrica contra el baseline
4. Retorna exit code 1 si alguna métrica excede el límite de alerta

Pruébalo inyectando un time.sleep(2) en tu mock provider para simular una regresión.

# Añadir al final de 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]:
    """
    Compara métricas actuales contra thresholds de alerta.
    Retorna lista de violaciones encontradas.
    """
    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

# En el bloque if __name__ == "__main__":
# Añadir después de 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")

# En mock_provider.py (temporalmente):
import time

class MockProvider:
    def complete(self, messages, **kwargs) -> str:
        time.sleep(2)  # Simular latencia de 2 segundos
        return '{"sentiment": "positive", "score": 0.9, "confidence": 0.85}'

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

Ejercicio 3: Proyección de costo mensual

Escribe una función project_monthly_cost(cost_per_request: float, requests_per_day: int) -> dict que calcule:

1. Costo diario estimado
2. Costo semanal estimado
3. Costo mensual estimado (30 días)
4. Costo anual estimado
5. Un flag budget_warning si el costo mensual supera $500

Usa los datos de tu benchmark para alimentar esta función.

def project_monthly_cost(cost_per_request: float, requests_per_day: int) -> dict:
    """
    Proyecta costos basándose en el costo promedio por request
    y el volumen de tráfico esperado.
    """
    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,
    }

# Ejemplo de uso con datos del benchmark:
if __name__ == "__main__":
    benchmark_cost = 0.002  # $0.002 por 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}")

# Output esperado:
# 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

Ejercicio 4: Dashboard de baselines en la terminal

Crea un script scripts/baselines_dashboard.py que lea tu docs/benchmark_latest.json y lo muestre como un dashboard ASCII con colores (verde si está dentro del target, rojo si excede el límite de alerta):

╔══════════════════════════════════════════╗
║       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

# scripts/baselines_dashboard.py
"""
Muestra un dashboard de baselines en la terminal.

Uso:
  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)

Troubleshooting

Problema: El benchmark muestra latencia de 0-5ms (sospechosamente rápido)

Síntomas:

• p50: 3ms, p99: 8ms — demasiado rápido para una llamada real a un LLM

Causa más probable: Estás ejecutando el benchmark contra el mock provider sin darte cuenta. Verifica que USE_MOCK_PROVIDER no esté seteado en tu .env.

Solución:

# Verificar si el mock está activo:
grep USE_MOCK .env
# Si dice USE_MOCK_PROVIDER=true, ese es tu problema

# Ejecutar con provider real:
USE_MOCK_PROVIDER=false python scripts/benchmark.py --n 5
# Ahora deberías ver latencias de 500-3000ms

Si quieres medir solo el overhead de tu stack (sin LLM), el mock es correcto — pero documenta que esos baselines son "stack-only", no "end-to-end".

Problema: El campo cost_usd aparece como $0.00 en los resultados

Síntomas:

• El benchmark corre bien pero la sección de costos muestra todo en cero

Causa más probable: Tu endpoint no está exponiendo el campo cost_usd en la respuesta JSON. El benchmark script lee body.get("cost_usd", 0) — si tu API no lo incluye, siempre será 0.

Solución: Verifica que tu endpoint incluye el costo en la respuesta:

# En tu 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),  # ← Asegúrate de incluir esto
    "request_id": get_request_id(),
}

Si no quieres exponer el costo al usuario final, puedes loguearlo internamente y leerlo desde los logs en vez de la respuesta HTTP.

Problema: El benchmark se cuelga o da timeout en algunos requests

Síntomas:

• Algunos requests terminan rápido, otros se cuelgan por 30+ segundos
• El script eventualmente falla con TimeoutError

Causa más probable: OpenAI está experimentando latencia alta, o tu rate limiter está rechazando requests y el retry está esperando.

Solución:

# Reducir el timeout del benchmark:
# En benchmark.py, cambiar el timeout de run_request:
# timeout=30 → timeout=15

# Ejecutar con menos requests para diagnosticar:
python scripts/benchmark.py --n 5

# Verificar si el rate limiter está bloqueando:
tail -20 logs/app.json | jq 'select(.event == "client_rate_limit_rejected")'

Si el problema es consistente, verifica https://status.openai.com para ver si hay un incidente activo.

Problema: Los resultados del benchmark varían mucho entre ejecuciones

Síntomas:

• Primera ejecución: p50: 800ms, segunda ejecución: p50: 2500ms
• Los números nunca son estables

Causa más probable: Es normal. Las LLM APIs tienen variabilidad inherente por carga del servidor, routing, y tamaño variable del output generado.

Solución:

# Ejecutar con más requests para promedios más estables:
python scripts/benchmark.py --n 50

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

La regla práctica: usa al menos 30 requests por benchmark y ejecuta 3 veces. Toma la mediana de las 3 ejecuciones como tu baseline. El p99 siempre tendrá alta varianza — es normal que fluctúe 2-3x entre ejecuciones.

Resumen

• Baselines = compromisos: no son aspiracionales — son lo que prometes a tus usuarios
• Las 6 métricas clave: p50/p99 latency, cost/request, error rate, guardrail rate, fallback rate
• Benchmark script: ejecutable antes de cada deploy para detectar regresiones
• El p99 importa: el usuario con el request más lento también existe — no solo promedios
• Documentar en BASELINES.md: para comparar evolución a lo largo del tiempo

Recursos adicionales

1. Why Percentiles — Por qué p99 importa más que el promedio
2. OpenAI Usage Dashboard — Ver costos reales en el dashboard de OpenAI
3. Locust — Para load testing con cientos de usuarios concurrentes
4. statistics module (Python) — Para cálculos estadísticos