Módulo 7: Evaluación de Prompts
7. Evaluation Pipelines Automatizadas
Descripción
Automatizar la evaluación end-to-end: desde cargar el dataset hasta generar reportes. Scheduling con cron/CI. Trend tracking con historial de métricas. Alertas cuando la calidad cae. Dashboard con visualizaciones.
¿Qué es una Evaluation Pipeline?
Una evaluation pipeline automatizada es el sistema que ejecuta evaluaciones de forma regular sin intervención manual. Es el equivalente de una batería de tests de regresión, pero para LLMs.
Pipeline Manual (sin automatización):
- Cada vez que alguien piensa en hacerlo
- Resultados en hojas de cálculo o notas
- Sin historial ni tendencias
- Equipo no se entera cuando falla
Pipeline Automatizada:
Dataset → Run prompts → Evaluate → Store results → Report → Alert if needed
↑ ↓
Cron/CI ←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←← Slack/Email
Arquitectura de la Pipeline
┌─────────────────────────────────────────────────────┐
│ EVALUATION PIPELINE │
│ │
│ 1. LOADER → Carga golden set │
│ 2. RUNNER → Ejecuta prompt(s) │
│ 3. EVALUATOR → Calcula métricas │
│ 4. COMPARATOR → Compara con baseline │
│ 5. REPORTER → Genera reporte markdown │
│ 6. NOTIFIER → Envía alertas si hay fallo │
│ 7. STORAGE → Persiste historial │
└─────────────────────────────────────────────────────┘
Implementación Completa
Pipeline Base
import json
import time
import asyncio
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import Callable, Any
from openai import OpenAI, AsyncOpenAI
client = OpenAI()
async_client = AsyncOpenAI()
@dataclass
class EvaluationRun:
"""Resultado de una ejecución completa de la pipeline."""
run_id: str
prompt_name: str
prompt_version: str
timestamp: str
golden_set_size: int
metricas: dict[str, float] = field(default_factory=dict)
fallos: list[dict] = field(default_factory=list)
metadata: dict = field(default_factory=dict)
duracion_segundos: float = 0.0
costo_estimado: float = 0.0
class EvaluationPipeline:
"""
Pipeline de evaluación modular y extensible.
Soporta:
- Múltiples métricas (accuracy, faithfulness, format, latencia)
- Evaluación asíncrona para mejor performance
- Comparación con baseline
- Historial persistente
- Alertas automáticas
"""
def __init__(
self,
golden_set_path: str,
history_path: str = "eval_history.jsonl",
baseline_path: str = "baseline.json",
notificaciones: bool = True
):
self.golden_set_path = Path(golden_set_path)
self.history_path = Path(history_path)
self.baseline_path = Path(baseline_path)
self.notificaciones = notificaciones
# Cargar golden set
with open(self.golden_set_path) as f:
self.golden_set = json.load(f)
print(f"Pipeline inicializada: {len(self.golden_set)} ejemplos en golden set")
async def _run_prompt_async(
self,
prompt_template: str,
input_text: str,
semaphore: asyncio.Semaphore
) -> tuple[str, dict]:
"""Ejecuta un prompt de forma asíncrona con control de concurrencia."""
async with semaphore:
inicio = time.time()
response = await async_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": prompt_template.format(input=input_text)
}],
temperature=0
)
return (
response.choices[0].message.content.strip(),
{
"latencia_ms": (time.time() - inicio) * 1000,
"tokens": response.usage.total_tokens,
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens
}
)
def _calcular_accuracy(self, outputs: list[str]) -> float:
"""Calcula accuracy contra expected_outputs del golden set."""
if not outputs:
return 0.0
correctos = sum(
output.strip().lower() == str(ej["expected_output"]).strip().lower()
for output, ej in zip(outputs, self.golden_set)
)
return correctos / len(outputs)
def _calcular_format_compliance(
self,
outputs: list[str],
formato: str = "text"
) -> float:
"""Calcula format compliance según el formato esperado."""
if not outputs:
return 0.0
compliant = 0
for output in outputs:
if formato == "json":
try:
json.loads(output)
compliant += 1
except json.JSONDecodeError:
pass
elif formato == "text":
compliant += 1 # Texto siempre válido
elif formato in ["POSITIVO", "NEGATIVO", "NEUTRO"]:
if output.strip().upper() in ["POSITIVO", "NEGATIVO", "NEUTRO"]:
compliant += 1
return compliant / len(outputs)
async def _calcular_faithfulness_batch(
self,
outputs: list[str],
sample_rate: float = 0.2
) -> float:
"""
Calcula faithfulness con LLM-as-judge en un sample.
Solo evalúa el sample_rate para reducir costo.
"""
import random
indices = random.sample(
range(len(outputs)),
int(len(outputs) * sample_rate) + 1
)
scores = []
semaphore = asyncio.Semaphore(5)
async def evaluar_fidelidad(idx):
ej = self.golden_set[idx]
output = outputs[idx]
prompt = f"""¿El OUTPUT usa solo información del INPUT?
INPUT: {ej['input']}
OUTPUT: {output}
Responde "1" (fiel) o "0" (inventa). Solo el número."""
resp_text, _ = await self._run_prompt_async(prompt, "", semaphore)
return 1.0 if "1" in resp_text else 0.0
tasks = [evaluar_fidelidad(i) for i in indices]
scores = await asyncio.gather(*tasks)
return sum(scores) / len(scores) if scores else 0.5
def _identificar_fallos(
self,
outputs: list[str],
max_fallos: int = 20
) -> list[dict]:
"""Identifica los casos donde el prompt falló."""
fallos = []
for i, (output, ej) in enumerate(zip(outputs, self.golden_set)):
esperado = str(ej["expected_output"]).strip().lower()
actual = output.strip().lower()
if esperado != actual:
fallos.append({
"id": ej.get("id", str(i)),
"input": str(ej["input"])[:100],
"esperado": str(ej["expected_output"]),
"actual": output[:100],
"categoria": ej.get("categoria", "unknown"),
"dificultad": ej.get("dificultad", "unknown")
})
if len(fallos) >= max_fallos:
break
return fallos
async def ejecutar(
self,
prompt_template: str,
prompt_name: str,
prompt_version: str,
metricas: list[str] = None,
max_concurrent: int = 10
) -> EvaluationRun:
"""
Ejecuta la pipeline completa de evaluación.
metricas: Lista de métricas a calcular.
Opciones: "accuracy", "format", "faithfulness", "latencia", "costo"
"""
if metricas is None:
metricas = ["accuracy", "format", "latencia", "costo"]
run_id = f"{prompt_name}_{prompt_version}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
inicio_total = time.time()
print(f"\n{'='*50}")
print(f"Ejecutando evaluación: {prompt_name} {prompt_version}")
print(f"Golden set: {len(self.golden_set)} ejemplos")
print(f"{'='*50}")
# 1. Ejecutar todos los prompts en paralelo
semaphore = asyncio.Semaphore(max_concurrent)
tasks = [
self._run_prompt_async(prompt_template, str(ej["input"]), semaphore)
for ej in self.golden_set
]
print(f"Ejecutando {len(tasks)} prompts en paralelo (max_concurrent={max_concurrent})...")
resultados = await asyncio.gather(*tasks)
outputs = [r[0] for r in resultados]
metadatas = [r[1] for r in resultados]
# 2. Calcular métricas
print("Calculando métricas...")
metricas_calculadas = {}
if "accuracy" in metricas:
metricas_calculadas["accuracy"] = self._calcular_accuracy(outputs)
if "format" in metricas:
formato = self.golden_set[0].get("formato_esperado", "text") if self.golden_set else "text"
metricas_calculadas["format"] = self._calcular_format_compliance(outputs, formato)
if "faithfulness" in metricas:
print(" Calculando faithfulness (LLM-as-judge, sample 20%)...")
metricas_calculadas["faithfulness"] = await self._calcular_faithfulness_batch(outputs)
if "latencia" in metricas:
latencias = [m["latencia_ms"] for m in metadatas]
sorted_lat = sorted(latencias)
n = len(sorted_lat)
metricas_calculadas["latencia_p50"] = sorted_lat[n // 2]
metricas_calculadas["latencia_p95"] = sorted_lat[int(n * 0.95)]
metricas_calculadas["latencia_mean"] = sum(latencias) / n
if "costo" in metricas:
tokens_total = sum(m["tokens"] for m in metadatas)
# GPT-4o-mini: ~$0.15/1M input tokens, ~$0.60/1M output tokens
prompt_tokens = sum(m["prompt_tokens"] for m in metadatas)
completion_tokens = sum(m["completion_tokens"] for m in metadatas)
costo = prompt_tokens * 0.15 / 1e6 + completion_tokens * 0.60 / 1e6
metricas_calculadas["tokens_total"] = tokens_total
metricas_calculadas["costo_usd"] = costo
metricas_calculadas["costo_por_request"] = costo / len(outputs)
# 3. Identificar fallos
fallos = self._identificar_fallos(outputs)
# 4. Crear run result
duracion = time.time() - inicio_total
run = EvaluationRun(
run_id=run_id,
prompt_name=prompt_name,
prompt_version=prompt_version,
timestamp=datetime.now().isoformat(),
golden_set_size=len(self.golden_set),
metricas=metricas_calculadas,
fallos=fallos,
duracion_segundos=duracion,
costo_estimado=metricas_calculadas.get("costo_usd", 0.0)
)
# 5. Persistir en historial
self._guardar_historial(run)
print(f"\n✅ Evaluación completada en {duracion:.1f}s")
self._imprimir_resumen(run)
return run
def _guardar_historial(self, run: EvaluationRun) -> None:
"""Persiste el run en el historial JSONL."""
with open(self.history_path, "a") as f:
f.write(json.dumps(asdict(run)) + "\n")
def _imprimir_resumen(self, run: EvaluationRun) -> None:
"""Imprime resumen en consola."""
print(f"\n📊 Métricas ({run.prompt_name} {run.prompt_version}):")
for metrica, valor in run.metricas.items():
if isinstance(valor, float):
print(f" {metrica:25s}: {valor:.4f}")
else:
print(f" {metrica:25s}: {valor}")
if run.fallos:
print(f"\n❌ Fallos: {len(run.fallos)}/{run.golden_set_size}")
for fallo in run.fallos[:3]: # Mostrar solo los primeros 3
print(f" [{fallo['id']}] Expected: {fallo['esperado']} | Got: {fallo['actual'][:40]}")Trend Tracking
El historial permite detectar tendencias: ¿la calidad mejora o empeora con el tiempo?
class TrendTracker:
"""Analiza tendencias en el historial de evaluaciones."""
def __init__(self, history_path: str = "eval_history.jsonl"):
self.history_path = Path(history_path)
def cargar_historial(
self,
prompt_name: str,
last_n: int = 30
) -> list[dict]:
"""Carga los últimos N runs de un prompt."""
if not self.history_path.exists():
return []
runs = []
with open(self.history_path) as f:
for line in f:
run = json.loads(line.strip())
if run.get("prompt_name") == prompt_name:
runs.append(run)
return runs[-last_n:]
def analizar_tendencia(
self,
prompt_name: str,
metrica: str = "accuracy",
ventana: int = 7 # últimos N runs
) -> dict:
"""
Analiza la tendencia de una métrica en el tiempo.
Usa regresión lineal para detectar si está mejorando/empeorando.
"""
from scipy import stats as scipy_stats
historial = self.cargar_historial(prompt_name, last_n=ventana)
if len(historial) < 2:
return {"tendencia": "INSUFICIENTE_DATOS", "n": len(historial)}
valores = []
for run in historial:
m = run.get("metricas", {}).get(metrica)
if m is not None:
valores.append(float(m))
if len(valores) < 2:
return {"tendencia": "METRICA_NO_DISPONIBLE"}
x = list(range(len(valores)))
slope, intercept, r_value, p_value, std_err = scipy_stats.linregress(x, valores)
tendencia = "ESTABLE"
if slope > 0.005 and p_value < 0.05:
tendencia = "MEJORANDO ↑"
elif slope < -0.005 and p_value < 0.05:
tendencia = "EMPEORANDO ↓"
return {
"tendencia": tendencia,
"slope": slope,
"r_squared": r_value ** 2,
"p_value": p_value,
"ultimo_valor": valores[-1],
"primer_valor": valores[0],
"delta_total": valores[-1] - valores[0],
"n_puntos": len(valores),
"valores": valores
}
def generar_reporte_tendencias(self, prompt_name: str) -> str:
"""Genera reporte de tendencias para todas las métricas."""
lineas = [
f"# Reporte de Tendencias: {prompt_name}",
f"Generado: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
""
]
metricas = ["accuracy", "faithfulness", "format", "latencia_p95"]
for metrica in metricas:
analisis = self.analizar_tendencia(prompt_name, metrica)
if "tendencia" not in analisis:
continue
emoji = {"MEJORANDO ↑": "📈", "EMPEORANDO ↓": "📉", "ESTABLE": "➡️"}.get(
analisis["tendencia"], "❓"
)
lineas.append(f"## {emoji} {metrica.upper()}")
lineas.append(f"- Tendencia: **{analisis['tendencia']}**")
if "ultimo_valor" in analisis:
ultimo = analisis["ultimo_valor"]
if metrica in ["accuracy", "faithfulness", "format"]:
lineas.append(f"- Último valor: {ultimo:.2%}")
else:
lineas.append(f"- Último valor: {ultimo:.1f}ms")
if "delta_total" in analisis:
delta = analisis["delta_total"]
lineas.append(f"- Delta (total): {delta:+.4f}")
lineas.append("")
return "\n".join(lineas)Sistema de Alertas
class AlertaManager:
"""Gestiona alertas cuando métricas caen por debajo de umbrales."""
def __init__(self, umbrales: dict[str, float] | None = None):
# Umbrales default — personalizar según tu caso de uso
self.umbrales = umbrales or {
"accuracy": 0.85, # Alertar si accuracy < 85%
"faithfulness": 0.80, # Alertar si faithfulness < 80%
"format": 0.95, # Alertar si format compliance < 95%
"latencia_p95": 5000.0, # Alertar si p95 > 5000ms
"costo_por_request": 0.01 # Alertar si costo > $0.01/request
}
self.alertas_activas: list[dict] = []
def verificar(self, run: EvaluationRun) -> list[dict]:
"""
Verifica si alguna métrica supera los umbrales.
Retorna lista de alertas generadas.
"""
alertas = []
for metrica, umbral in self.umbrales.items():
valor = run.metricas.get(metrica)
if valor is None:
continue
# Para latencia y costo: alertar si SUPERA el umbral (mayor es peor)
# Para el resto: alertar si está POR DEBAJO del umbral (menor es peor)
if metrica in ["latencia_p95", "costo_por_request"]:
if valor > umbral:
alertas.append({
"run_id": run.run_id,
"metrica": metrica,
"valor": valor,
"umbral": umbral,
"tipo": "EXCEDE_UMBRAL",
"mensaje": f"⚠️ {metrica}={valor:.4f} excede umbral {umbral}"
})
else:
if valor < umbral:
alertas.append({
"run_id": run.run_id,
"metrica": metrica,
"valor": valor,
"umbral": umbral,
"tipo": "BAJO_UMBRAL",
"mensaje": f"🚨 {metrica}={valor:.2%} por debajo del umbral {umbral:.2%}"
})
self.alertas_activas.extend(alertas)
return alertas
def notificar_slack(self, alertas: list[dict], webhook_url: str) -> None:
"""Envía alertas a Slack via webhook."""
if not alertas:
return
import urllib.request
import json
texto = f"🚨 *Alertas de Evaluation Pipeline*\n\n"
for alerta in alertas:
texto += f"• {alerta['mensaje']}\n"
texto += f" Run ID: `{alerta['run_id']}`\n\n"
payload = {"text": texto}
data = json.dumps(payload).encode()
req = urllib.request.Request(
webhook_url,
data=data,
headers={"Content-Type": "application/json"}
)
try:
urllib.request.urlopen(req, timeout=5)
print(f"✅ {len(alertas)} alertas enviadas a Slack")
except Exception as e:
print(f"❌ Error enviando a Slack: {e}")
def notificar_email(
self,
alertas: list[dict],
to: str,
from_email: str,
smtp_config: dict
) -> None:
"""Envía alertas por email."""
if not alertas:
return
import smtplib
from email.mime.text import MIMEText
cuerpo = "Alertas de Evaluation Pipeline:\n\n"
for alerta in alertas:
cuerpo += f"• {alerta['mensaje']}\n"
msg = MIMEText(cuerpo)
msg["Subject"] = f"🚨 Alerta: {len(alertas)} métricas fuera de umbral"
msg["From"] = from_email
msg["To"] = to
with smtplib.SMTP(smtp_config["host"], smtp_config["port"]) as server:
server.sendmail(from_email, to, msg.as_string())
print(f"✅ Email de alerta enviado a {to}")Reporte Automático
def generar_reporte_markdown(
run: EvaluationRun,
baseline_metricas: dict | None = None,
tendencias: dict | None = None
) -> str:
"""Genera un reporte completo en markdown para el run de evaluación."""
timestamp = datetime.fromisoformat(run.timestamp).strftime("%Y-%m-%d %H:%M:%S")
lineas = [
f"# Evaluation Report: {run.prompt_name}",
f"**Versión:** {run.prompt_version} ",
f"**Fecha:** {timestamp} ",
f"**Run ID:** `{run.run_id}` ",
"",
f"## 📊 Métricas Principales",
"",
"| Métrica | Valor |" + (" Baseline | Delta |" if baseline_metricas else ""),
"|---------|-------|" + (" ---------|-------|" if baseline_metricas else ""),
]
for metrica, valor in run.metricas.items():
if isinstance(valor, float):
val_str = f"{valor:.2%}" if valor <= 1.0 else f"{valor:.1f}"
else:
val_str = str(valor)
if baseline_metricas and metrica in baseline_metricas:
baseline_val = baseline_metricas[metrica]
delta = valor - baseline_val if isinstance(valor, float) else 0
delta_str = f"{delta:+.2%}" if abs(delta) <= 1 else f"{delta:+.1f}"
estado = "✅" if delta >= 0 else "⚠️" if delta > -0.05 else "🚨"
lineas.append(f"| {metrica} | {val_str} | {baseline_val:.2%} | {estado} {delta_str} |")
else:
lineas.append(f"| {metrica} | {val_str} |")
# Tendencias
if tendencias:
lineas.extend(["", "## 📈 Tendencias (últimos 7 runs)", ""])
for metrica, t in tendencias.items():
if isinstance(t, dict) and "tendencia" in t:
lineas.append(f"- **{metrica}:** {t['tendencia']}")
# Fallos
if run.fallos:
lineas.extend([
"",
f"## ❌ Casos Fallidos ({len(run.fallos)}/{run.golden_set_size})",
"",
"| ID | Input | Esperado | Obtenido | Categoría |",
"|----|-------|----------|----------|-----------|",
])
for fallo in run.fallos[:10]: # Máximo 10 en el reporte
input_corto = fallo["input"][:40].replace("|", "\\|")
lineas.append(
f"| {fallo['id']} | {input_corto}... | `{fallo['esperado']}` | `{fallo['actual'][:30]}` | {fallo['categoria']} |"
)
else:
lineas.extend(["", "## ✅ Sin Fallos", ""])
# Metadata
lineas.extend([
"",
"## ℹ️ Metadata",
f"- **Duración:** {run.duracion_segundos:.1f}s",
f"- **Costo estimado:** ${run.costo_estimado:.4f}",
f"- **Golden set:** {run.golden_set_size} ejemplos",
])
return "\n".join(lineas)Scheduling: Ejecución Programada
Con Cron (Linux/Mac)
# Editar crontab: crontab -e
# Formato: minuto hora dia_mes mes dia_semana comando
# Ejecutar evaluation pipeline todos los días a las 2am
0 2 * * * cd /app && python run_evaluation.py >> /var/log/eval_pipeline.log 2>&1
# Ejecutar cada lunes a las 9am (reporte semanal)
0 9 * * 1 cd /app && python run_weekly_report.pyScript de Ejecución (run_evaluation.py)
#!/usr/bin/env python3
"""Script principal para ejecutar la evaluation pipeline."""
import asyncio
import sys
from pathlib import Path
# Configuración
GOLDEN_SET_PATH = "datasets/clasificador_sentimiento.json"
PROMPT_FILE = "prompts/clasificador_v2.txt"
PROMPT_NAME = "clasificador_sentimiento"
PROMPT_VERSION = "v2.1"
SLACK_WEBHOOK = "https://hooks.slack.com/services/xxx/yyy/zzz" # Opcional
async def main():
print(f"Iniciando evaluation pipeline: {PROMPT_NAME} {PROMPT_VERSION}")
# Cargar prompt
with open(PROMPT_FILE) as f:
prompt_template = f.read()
# Inicializar pipeline
pipeline = EvaluationPipeline(
golden_set_path=GOLDEN_SET_PATH,
history_path="eval_history.jsonl"
)
# Ejecutar evaluación
run = await pipeline.ejecutar(
prompt_template=prompt_template,
prompt_name=PROMPT_NAME,
prompt_version=PROMPT_VERSION,
metricas=["accuracy", "format", "faithfulness", "latencia", "costo"]
)
# Verificar alertas
alertas_mgr = AlertaManager(umbrales={
"accuracy": 0.88,
"faithfulness": 0.80,
"format": 0.95
})
alertas = alertas_mgr.verificar(run)
if alertas and SLACK_WEBHOOK:
alertas_mgr.notificar_slack(alertas, SLACK_WEBHOOK)
# Generar reporte
tracker = TrendTracker()
tendencias = {}
for metrica in ["accuracy", "faithfulness"]:
tendencias[metrica] = tracker.analizar_tendencia(PROMPT_NAME, metrica)
reporte = generar_reporte_markdown(run, tendencias=tendencias)
# Guardar reporte
reporte_path = f"reports/{run.run_id}.md"
Path("reports").mkdir(exist_ok=True)
with open(reporte_path, "w") as f:
f.write(reporte)
print(f"\nReporte guardado: {reporte_path}")
# Exit code 1 si hay alertas críticas (para CI/CD)
if any(a["tipo"] == "BAJO_UMBRAL" for a in alertas):
print(f"\n❌ {len(alertas)} alertas críticas — exit code 1")
sys.exit(1)
print("\n✅ Pipeline completada exitosamente")
if __name__ == "__main__":
asyncio.run(main())Dashboard Simple con Tablas
def generar_dashboard_texto(
prompt_name: str,
history_path: str = "eval_history.jsonl",
last_n: int = 10
) -> str:
"""
Genera un dashboard en texto/markdown con los últimos N runs.
"""
tracker = TrendTracker(history_path)
historial = tracker.cargar_historial(prompt_name, last_n)
if not historial:
return f"Sin datos para {prompt_name}"
lineas = [
f"# Dashboard: {prompt_name}",
f"Últimos {len(historial)} runs",
"",
"| Fecha | Versión | Accuracy | Faithfulness | Format | Costo |",
"|-------|---------|----------|-------------|--------|-------|",
]
for run_data in historial:
timestamp = run_data.get("timestamp", "")[:16]
version = run_data.get("prompt_version", "?")
metricas = run_data.get("metricas", {})
accuracy = f"{metricas.get('accuracy', 0):.2%}"
faith = f"{metricas.get('faithfulness', 0):.2%}" if "faithfulness" in metricas else "N/A"
format_c = f"{metricas.get('format', 0):.2%}" if "format" in metricas else "N/A"
costo = f"${metricas.get('costo_usd', 0):.4f}" if "costo_usd" in metricas else "N/A"
lineas.append(f"| {timestamp} | {version} | {accuracy} | {faith} | {format_c} | {costo} |")
return "\n".join(lineas)Troubleshooting
Problema 1: Pipeline lenta con golden set grande
Síntoma: Evaluar 500 ejemplos tarda 15+ minutos.
Solución:
# Usar concurrencia más alta (verificar rate limits de la API)
run = await pipeline.ejecutar(
prompt_template=prompt,
prompt_name="clasificador",
prompt_version="v2",
max_concurrent=20 # Aumentar desde el default de 10
)
# Para golden sets muy grandes: dividir en batches
async def evaluar_con_batches(golden_set, batch_size=100):
for i in range(0, len(golden_set), batch_size):
batch = golden_set[i:i+batch_size]
print(f"Procesando batch {i//batch_size + 1}/{len(golden_set)//batch_size + 1}")
# Evaluar batch...
await asyncio.sleep(1) # Pausa entre batchesProblema 2: Alertas muy ruidosas (muchos false positives)
Síntoma: Recibes alertas todos los días aunque el sistema está bien.
Causa: Umbrales muy estrictos o variación natural del modelo.
Solución:
# 1. Usar ventana deslizante (promedio de últimos N runs)
# en lugar de comparar run individual
def umbral_con_ventana(historial: list[dict], metrica: str, ventana: int = 3) -> float:
"""Promedia los últimos ventana runs para suavizar variaciones."""
ultimos = historial[-ventana:]
valores = [run.get("metricas", {}).get(metrica, 0) for run in ultimos]
return sum(valores) / len(valores) if valores else 0
# 2. Alertar solo si cae 3 runs consecutivos (no solo 1)
# 3. Ajustar umbrales basándote en datos históricos realesProblema 3: Historial crece demasiado
Síntoma: eval_history.jsonl tiene cientos de megabytes.
Solución:
def comprimir_historial(
history_path: str = "eval_history.jsonl",
max_runs: int = 90 # Mantener últimos 90 runs (3 meses aprox.)
) -> None:
"""Mantiene solo los últimos max_runs en el historial."""
with open(history_path) as f:
lineas = f.readlines()
if len(lineas) <= max_runs:
return
# Respaldar antes de truncar
backup_path = f"{history_path}.bak"
with open(backup_path, "w") as f:
f.writelines(lineas)
# Mantener solo los últimos N
with open(history_path, "w") as f:
f.writelines(lineas[-max_runs:])
print(f"Historial comprimido: {len(lineas)} → {max_runs} runs")Ejercicios
Ejercicio 1: Crear una pipeline mínima viable
Implementa una pipeline que evalúe un prompt de clasificación contra 20 ejemplos y guarde los resultados en JSONL.
Ver solución
import json
import time
from datetime import datetime
from openai import OpenAI
client = OpenAI()
def pipeline_minima(
prompt_template: str,
golden_set: list[dict],
prompt_name: str = "mi_prompt",
history_path: str = "eval_history.jsonl"
) -> dict:
"""Pipeline mínima viable."""
inicio = time.time()
outputs = []
for ej in golden_set:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt_template.format(input=ej["input"])}],
temperature=0
)
outputs.append(response.choices[0].message.content.strip())
# Calcular accuracy
correctos = sum(
o.lower() == str(e["expected_output"]).lower()
for o, e in zip(outputs, golden_set)
)
accuracy = correctos / len(golden_set)
# Guardar resultado
resultado = {
"timestamp": datetime.now().isoformat(),
"prompt_name": prompt_name,
"golden_set_size": len(golden_set),
"metricas": {"accuracy": accuracy},
"duracion_s": time.time() - inicio
}
with open(history_path, "a") as f:
f.write(json.dumps(resultado) + "\n")
print(f"✅ Accuracy: {accuracy:.2%} | Guardado en {history_path}")
return resultado
# Uso:
PROMPT = "Clasifica como POSITIVO, NEGATIVO o NEUTRO: {input}. Solo la categoría."
golden = [
{"input": "Excelente producto", "expected_output": "POSITIVO"},
{"input": "Terrible servicio", "expected_output": "NEGATIVO"},
{"input": "El paquete llegó", "expected_output": "NEUTRO"},
]
resultado = pipeline_minima(PROMPT, golden)Ejercicio 2: Agregar trend tracking
Dado el siguiente historial, implementa una función que detecte si la accuracy está mejorando o empeorando:
Ver solución
def analizar_tendencia_simple(historial: list[dict], metrica: str = "accuracy") -> str:
"""Analiza tendencia simple con últimos 5 valores."""
from scipy import stats
valores = [r.get("metricas", {}).get(metrica) for r in historial]
valores = [v for v in valores if v is not None][-5:] # Últimos 5
if len(valores) < 3:
return "INSUFICIENTE_DATOS"
x = list(range(len(valores)))
slope, _, _, p_value, _ = stats.linregress(x, valores)
if abs(slope) < 0.002 or p_value > 0.1:
return f"ESTABLE ({valores[-1]:.2%})"
elif slope > 0:
return f"MEJORANDO ↑ ({valores[0]:.2%} → {valores[-1]:.2%})"
else:
return f"EMPEORANDO ↓ ({valores[0]:.2%} → {valores[-1]:.2%})"
# Test con historial simulado:
historial = [
{"metricas": {"accuracy": 0.88}},
{"metricas": {"accuracy": 0.89}},
{"metricas": {"accuracy": 0.91}},
{"metricas": {"accuracy": 0.92}},
{"metricas": {"accuracy": 0.93}},
]
print(analizar_tendencia_simple(historial)) # MEJORANDO ↑ (88% → 93%)Resumen
- Pipeline: Dataset → Run prompts → Evaluate → Store → Report → Alert — automatizado
- Async: Paralelizar llamadas API para evaluar golden sets grandes en segundos en vez de minutos
- Historial: JSONL persistente con timestamps para trend tracking
- Trend tracking: Regresión lineal sobre historial para detectar degradación gradual
- Alertas: Umbrales por métrica con notificación a Slack/email/CI
- Scheduling: Cron para evaluación diaria/semanal; CI/CD para cada commit
- Dashboard: Tablas de historial para visibilidad del equipo
Recursos adicionales
- LangSmith — Pipeline de evaluación managed
- Weights & Biases — Experiment tracking con dashboards
- Prometheus + Grafana — Monitoring y alertas para producción
- Apache Airflow — Scheduling de pipelines complejas
- asyncio Documentation — Para paralelización