Pipeline Automatizado de Evaluation

Descripción de la cápsula

Hasta aquí tienes los componentes individuales: métricas RAGAS, golden dataset versionado, criterios de interpretación. Pero si para evaluar tu sistema necesitas abrir un Jupyter notebook, copiar código de tres lugares distintos y mirar resultados a ojo, no estás evaluando — estás haciendo experimentos.

La diferencia entre experimentos y evaluación es automatización. Un pipeline de evaluación automatizada es un comando único, reproducible, con outputs estructurados, que cualquier miembro del equipo puede correr sin saber los detalles internos. Es la pieza que convierte "queremos evaluar" en "evaluamos en cada PR".

En esta cápsula vas a construir ese pipeline end-to-end: carga del golden dataset, ejecución del sistema RAG sobre cada query, recolección de contextos y respuestas, cálculo de métricas RAGAS, generación de reportes JSON + Markdown, y modo smoke para feedback rápido durante desarrollo.

Al terminar tendrás python scripts/evaluate.py --mode full como comando que cualquiera puede correr y producir resultados comparables a través del tiempo. Es la base sobre la cual construirás regression testing (cápsula 06) y CI/CD (cápsula 07).

Arquitectura del pipeline

El pipeline tiene cinco etapas claras que deben estar desacopladas para poder testearlas y modificarlas individualmente:

┌─────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│ Load Golden     │────▶│ Execute RAG      │────▶│ Collect Traces   │
│ Dataset         │     │ over each query  │     │ (answer+context) │
└─────────────────┘     └──────────────────┘     └──────────────────┘
                                                          │
                                                          ▼
┌─────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│ Export Reports  │◀────│ Compare against  │◀────│ Run RAGAS        │
│ JSON + MD       │     │ thresholds       │     │ metrics          │
└─────────────────┘     └──────────────────┘     └──────────────────┘

Cada etapa tiene una interfaz clara:

Esta separación te permite testear cada pieza en isolation y reemplazar una sin tocar las demás.

Estructura del proyecto

production-rag/
├── scripts/
│   └── evaluate.py                # entry point
├── eval/
│   ├── __init__.py
│   ├── loader.py                  # carga golden dataset
│   ├── runner.py                  # ejecuta RAG sobre queries
│   ├── metrics.py                 # wrapper RAGAS
│   ├── reporter.py                # genera reportes
│   └── thresholds.py              # quality gates
├── golden_dataset/
│   └── v1.0.0.json
├── eval_reports/                  # gitignored, versionados aparte
│   ├── 2026-03-13_a3f29c1.json
│   └── 2026-03-13_a3f29c1.md
└── pyproject.toml

Implementación: loader

# eval/loader.py
import json
from pathlib import Path
from pydantic import BaseModel

class GoldenRecord(BaseModel):
    id: str
    query: str
    ground_truth_answer: str
    expected_sources: list[str]
    difficulty: str
    query_type: str
    category: str

class GoldenDataset(BaseModel):
    version: str
    records: list[GoldenRecord]

def load_golden_dataset(path: Path) -> GoldenDataset:
    raw = json.loads(path.read_text())
    return GoldenDataset(
        version=raw["meta"]["version"],
        records=[GoldenRecord(**r) for r in raw["records"]],
    )

def filter_by_mode(dataset: GoldenDataset, mode: str = "full") -> list[GoldenRecord]:
    if mode == "smoke":
        # 10 queries balanceadas: 4 easy, 4 medium, 2 hard
        easy = [r for r in dataset.records if r.difficulty == "easy"][:4]
        medium = [r for r in dataset.records if r.difficulty == "medium"][:4]
        hard = [r for r in dataset.records if r.difficulty == "hard"][:2]
        return easy + medium + hard
    return dataset.records

Por qué filter_by_mode: desarrollo local con 100 queries cuesta tiempo y tokens. Modo smoke da feedback en 30 segundos sin sacrificar cobertura de tipos. PR usa smoke; nightly usa full.

Implementación: runner

# eval/runner.py
import asyncio
from time import perf_counter
from pydantic import BaseModel

class EvalSample(BaseModel):
    query: str
    answer: str
    contexts: list[str]
    ground_truth: str
    expected_sources: list[str]
    retrieved_source_ids: list[str]
    latency_ms: float
    record_id: str

async def execute_one(record: GoldenRecord, rag_app) -> EvalSample:
    start = perf_counter()
    response = await rag_app.answer_with_context(record.query)
    elapsed = (perf_counter() - start) * 1000

    return EvalSample(
        query=record.query,
        answer=response.answer,
        contexts=[c.content for c in response.contexts],
        ground_truth=record.ground_truth_answer,
        expected_sources=record.expected_sources,
        retrieved_source_ids=[c.doc_id for c in response.contexts],
        latency_ms=elapsed,
        record_id=record.id,
    )

async def execute_batch(records: list[GoldenRecord], rag_app, concurrency: int = 5) -> list[EvalSample]:
    sem = asyncio.Semaphore(concurrency)
    async def with_sem(r):
        async with sem:
            return await execute_one(r, rag_app)
    return await asyncio.gather(*[with_sem(r) for r in records])

Decisiones clave:

• concurrency=5 evita rate limits de OpenAI mientras paraleliza retrieval (cada query consume embeddings + chat completion).
• Capturamos retrieved_source_ids separado de contexts para calcular precision@k contra expected_sources.
• latency_ms se persiste para correlacionar calidad con performance en el reporte.

Implementación: metrics

# eval/metrics.py
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

def compute_ragas_metrics(samples: list[EvalSample]) -> dict:
    rows = [
        {
            "question": s.query,
            "answer": s.answer,
            "contexts": s.contexts,
            "ground_truth": s.ground_truth,
        }
        for s in samples
    ]
    ds = Dataset.from_list(rows)
    result = evaluate(
        dataset=ds,
        metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    )
    return {
        "faithfulness": float(result["faithfulness"]),
        "answer_relevancy": float(result["answer_relevancy"]),
        "context_precision": float(result["context_precision"]),
        "context_recall": float(result["context_recall"]),
    }

def compute_retrieval_metrics(samples: list[EvalSample], k: int = 5) -> dict:
    precisions = []
    recalls = []
    mrrs = []
    for s in samples:
        expected = set(s.expected_sources)
        retrieved = s.retrieved_source_ids[:k]
        if not expected:
            continue
        # precision@k
        precisions.append(sum(1 for d in retrieved if d in expected) / k)
        # recall@k
        recalls.append(len(set(retrieved) & expected) / len(expected))
        # MRR
        rr = 0.0
        for idx, d in enumerate(retrieved, 1):
            if d in expected:
                rr = 1.0 / idx
                break
        mrrs.append(rr)
    return {
        f"precision_at_{k}": sum(precisions) / len(precisions) if precisions else 0,
        f"recall_at_{k}": sum(recalls) / len(recalls) if recalls else 0,
        "mrr": sum(mrrs) / len(mrrs) if mrrs else 0,
    }

def compute_segmented_metrics(samples: list[EvalSample], dataset: GoldenDataset) -> dict:
    """Métricas segmentadas por difficulty y query_type"""
    record_meta = {r.id: r for r in dataset.records}
    by_difficulty = {"easy": [], "medium": [], "hard": []}
    for s in samples:
        meta = record_meta.get(s.record_id)
        if meta:
            by_difficulty[meta.difficulty].append(s)
    segmented = {}
    for diff, subset in by_difficulty.items():
        if subset:
            segmented[diff] = compute_ragas_metrics(subset)
    return segmented

Por qué métricas segmentadas: un sistema puede tener faithfulness=0.90 global pero faithfulness=0.65 en queries hard. El promedio esconde el problema. Segmentar por difficulty te avisa de regresiones específicas.

Implementación: reporter

# eval/reporter.py
import json
from datetime import datetime
from pathlib import Path

def export_json_report(report: dict, output_dir: Path, commit_sha: str) -> Path:
    timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
    filename = f"{timestamp}_{commit_sha[:7]}.json"
    path = output_dir / filename
    path.write_text(json.dumps(report, indent=2, ensure_ascii=False))
    return path

def export_markdown_report(report: dict, output_dir: Path, commit_sha: str) -> Path:
    timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
    filename = f"{timestamp}_{commit_sha[:7]}.md"
    path = output_dir / filename

    lines = [
        f"# Evaluation Report - {timestamp}",
        f"",
        f"**Commit**: `{commit_sha}`",
        f"**Dataset**: {report['dataset_version']}",
        f"**Mode**: {report['mode']}",
        f"**Samples**: {report['n_samples']}",
        f"",
        f"## Generation Metrics (RAGAS)",
        f"",
        f"| Metric | Score |",
        f"|--------|-------|",
        f"| Faithfulness | {report['ragas']['faithfulness']:.3f} |",
        f"| Answer Relevancy | {report['ragas']['answer_relevancy']:.3f} |",
        f"| Context Precision | {report['ragas']['context_precision']:.3f} |",
        f"| Context Recall | {report['ragas']['context_recall']:.3f} |",
        f"",
        f"## Retrieval Metrics",
        f"",
        f"| Metric | Score |",
        f"|--------|-------|",
        f"| Precision@5 | {report['retrieval']['precision_at_5']:.3f} |",
        f"| Recall@5 | {report['retrieval']['recall_at_5']:.3f} |",
        f"| MRR | {report['retrieval']['mrr']:.3f} |",
        f"",
        f"## Performance",
        f"",
        f"- Avg latency: {report['performance']['avg_latency_ms']:.0f} ms",
        f"- p95 latency: {report['performance']['p95_latency_ms']:.0f} ms",
        f"",
    ]
    path.write_text("\n".join(lines))
    return path

Entry point: scripts/evaluate.py

# scripts/evaluate.py
import argparse
import asyncio
import subprocess
from pathlib import Path
import statistics

from eval.loader import load_golden_dataset, filter_by_mode
from eval.runner import execute_batch
from eval.metrics import compute_ragas_metrics, compute_retrieval_metrics, compute_segmented_metrics
from eval.reporter import export_json_report, export_markdown_report
from eval.thresholds import check_thresholds, DEFAULT_THRESHOLDS
from app.pipeline import build_rag_app

def get_git_commit() -> str:
    result = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True)
    return result.stdout.strip()

async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", choices=["smoke", "full"], default="full")
    parser.add_argument("--dataset", default="golden_dataset/v1.0.0.json")
    parser.add_argument("--output-dir", default="eval_reports")
    parser.add_argument("--enforce-thresholds", action="store_true")
    args = parser.parse_args()

    dataset = load_golden_dataset(Path(args.dataset))
    records = filter_by_mode(dataset, args.mode)
    print(f"Evaluating {len(records)} queries from {dataset.version} (mode: {args.mode})")

    rag_app = build_rag_app()
    samples = await execute_batch(records, rag_app)

    ragas_scores = compute_ragas_metrics(samples)
    retrieval_scores = compute_retrieval_metrics(samples, k=5)
    segmented = compute_segmented_metrics(samples, dataset)

    latencies = [s.latency_ms for s in samples]
    performance = {
        "avg_latency_ms": statistics.mean(latencies),
        "p95_latency_ms": sorted(latencies)[int(0.95 * len(latencies))],
    }

    report = {
        "mode": args.mode,
        "dataset_version": dataset.version,
        "commit_sha": get_git_commit(),
        "n_samples": len(samples),
        "ragas": ragas_scores,
        "retrieval": retrieval_scores,
        "segmented_by_difficulty": segmented,
        "performance": performance,
    }

    output_dir = Path(args.output_dir)
    output_dir.mkdir(exist_ok=True)
    json_path = export_json_report(report, output_dir, report["commit_sha"])
    md_path = export_markdown_report(report, output_dir, report["commit_sha"])
    print(f"Reports: {json_path}, {md_path}")

    if args.enforce_thresholds:
        passed, failures = check_thresholds(report, DEFAULT_THRESHOLDS)
        if not passed:
            print("THRESHOLD FAILURES:")
            for f in failures:
                print(f"  - {f}")
            exit(1)

if __name__ == "__main__":
    asyncio.run(main())

Uso:

# Desarrollo local (rápido)
python scripts/evaluate.py --mode smoke

# Pre-merge en CI
python scripts/evaluate.py --mode smoke --enforce-thresholds

# Nightly completo
python scripts/evaluate.py --mode full

Comparación: ad-hoc vs pipeline automatizado

Conexión con el proyecto final

Tu Advanced RAG System debe incluir el directorio eval/ con los módulos arriba, y el comando evaluate.py documentado en el README:

## Evaluación

```bash
# Smoke test (10 queries, 30 segundos)
python scripts/evaluate.py --mode smoke

# Full evaluation (100 queries, 5-10 minutos)
python scripts/evaluate.py --mode full

# Pre-merge con quality gates
python scripts/evaluate.py --mode smoke --enforce-thresholds

Reports se generan en eval_reports/{timestamp}_{commit}.{json,md}.

Esta documentación es lo que cualquier nuevo miembro del equipo lee primero.

---

## Troubleshooting

### Problema 1: "Pipeline tarda demasiado en local"
**Causa:** corres el set completo cada iteración.  
**Solución:** modo smoke (10 queries) para iteración. Reserva full para pre-merge y nightly.

### Problema 2: "Resultados no comparables entre runs"
**Causa:** dataset cambia, modelos cambian, prompts cambian sin tracking.  
**Solución:** versiona dataset, fija modelo en config, registra `commit_sha` en cada reporte. Si scores cambian sin razón aparente, diff los reportes JSON.

### Problema 3: "Pipeline falla con OpenAI rate limit"
**Causa:** `concurrency` muy alto o tier de OpenAI con límites bajos.  
**Solución:** baja a `concurrency=3`, agrega retry con backoff exponencial en `execute_one`.

### Problema 4: "Reportes ocupan mucho espacio"
**Causa:** versionas reportes en git.  
**Solución:** gitignore `eval_reports/` y guarda en S3/storage separado. En git solo va el código del pipeline.

### Problema 5: "Cambio en RAGAS rompe reportes históricos"
**Causa:** RAGAS actualiza definiciones de métricas entre versiones.  
**Solución:** pin RAGAS en `pyproject.toml`. Cuando actualices, corre evaluación contra última versión del dataset y registra el "punto de discontinuidad" en CHANGELOG.

---

## Ejercicios

### Ejercicio 1: Implementar el loader con validación

Implementa `load_golden_dataset` con validación de schema y manejo de errores claro.

<details>
<summary>Ver solución</summary>

```python
import json
from pathlib import Path
from pydantic import ValidationError

def load_golden_dataset(path: Path) -> GoldenDataset:
    if not path.exists():
        raise FileNotFoundError(f"Golden dataset not found: {path}")
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON in {path}: {e}")
    if "meta" not in raw or "records" not in raw:
        raise ValueError(f"Dataset missing required keys: meta, records")
    try:
        return GoldenDataset(
            version=raw["meta"]["version"],
            records=[GoldenRecord(**r) for r in raw["records"]],
        )
    except ValidationError as e:
        raise ValueError(f"Schema validation failed: {e}")

Explicación: errores con mensaje específico aceleran debug. Sin esto, un typo en el JSON se convierte en stacktrace incomprensible.

Ejercicio 2: Implementar retry con backoff

Agrega retry exponencial en execute_one para manejar rate limits.

import asyncio
import random

async def execute_one_with_retry(record: GoldenRecord, rag_app, max_attempts: int = 3) -> EvalSample:
    last_error = None
    for attempt in range(max_attempts):
        try:
            return await execute_one(record, rag_app)
        except Exception as e:
            last_error = e
            if "rate" in str(e).lower() or "429" in str(e):
                wait = (2 ** attempt) + random.uniform(0, 1)
                await asyncio.sleep(wait)
                continue
            raise
    raise RuntimeError(f"Failed after {max_attempts} attempts: {last_error}")

Ejercicio 3: Reporte de cambio entre runs

Implementa función que compara dos reportes y genera resumen de cambios.

def diff_reports(prev: dict, curr: dict) -> dict:
    diffs = {}
    for category in ["ragas", "retrieval"]:
        diffs[category] = {}
        for metric, current_value in curr[category].items():
            previous_value = prev[category].get(metric, 0)
            delta = current_value - previous_value
            pct = (delta / previous_value) * 100 if previous_value > 0 else 0
            diffs[category][metric] = {
                "previous": previous_value,
                "current": current_value,
                "delta": delta,
                "pct_change": pct,
                "direction": "↑" if delta > 0 else "↓" if delta < 0 else "=",
            }
    return diffs

def format_diff_markdown(diffs: dict) -> str:
    lines = ["## Changes vs previous run", ""]
    for category, metrics in diffs.items():
        lines.append(f"### {category}")
        for metric, change in metrics.items():
            lines.append(
                f"- **{metric}**: {change['previous']:.3f} → {change['current']:.3f} "
                f"({change['direction']} {change['pct_change']:+.1f}%)"
            )
    return "\n".join(lines)

Ejercicio 4: Modo selectivo por categoría

Extiende filter_by_mode para permitir filtrar por categoría específica.

def filter_dataset(
    dataset: GoldenDataset,
    mode: str = "full",
    categories: list[str] | None = None,
    difficulties: list[str] | None = None,
) -> list[GoldenRecord]:
    records = dataset.records
    if categories:
        records = [r for r in records if r.category in categories]
    if difficulties:
        records = [r for r in records if r.difficulty in difficulties]
    if mode == "smoke":
        records = records[:10]
    return records

# Uso: evaluar solo queries de seguridad hard
records = filter_dataset(dataset, mode="full", categories=["security"], difficulties=["hard"])

Resumen

• Pipeline automatizado convierte evaluación de experimento a práctica diaria
• Cinco etapas desacopladas: load → execute → collect → metrics → report
• Modo smoke (10 queries) para PR; full (100+) para nightly
• Reportes en JSON (machine-readable) + Markdown (human-readable)
• Métricas segmentadas por difficulty detectan regresiones específicas
• Retry exponencial con jitter maneja rate limits de OpenAI
• Diff entre runs es la base de regression testing en cápsula 06

Recursos adicionales

1. RAGAS Getting Started - Ejecución básica.
2. Pydantic Settings - Configuración de pipelines.
3. argparse Tutorial - CLI ergonómico.
4. MLflow Tracking - Alternativa para reports versionados.
5. Async Python Patterns - Concurrencia controlada.

Creado: Marzo 13, 2026
Versión: 2.0