The Automated Evaluation Pipeline

Capsule description

By now you have the individual components: RAGAS metrics, a versioned golden dataset, interpretation criteria. But if evaluating your system means opening a Jupyter notebook, copying code from three different places and eyeballing the results, you aren't evaluating — you're running experiments.

The difference between experiments and evaluation is automation. An automated evaluation pipeline is a single, reproducible command with structured outputs that any member of the team can run without knowing the internal details. It's the piece that turns "we want to evaluate" into "we evaluate on every PR".

In this capsule you're going to build that pipeline end-to-end: loading the golden dataset, running the RAG system over every query, collecting the contexts and answers, computing the RAGAS metrics, generating JSON + Markdown reports, and a smoke mode for fast feedback during development.

By the end you'll have python scripts/evaluate.py --mode full as a command anyone can run to produce results that stay comparable over time. It's the foundation on which you'll build regression testing (capsule 06) and CI/CD (capsule 07).

The pipeline's architecture

The pipeline has five clear stages that must stay decoupled so you can test and modify them individually:

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

Each stage has a clear interface:

This separation lets you test each piece in isolation and swap one out without touching the others.

The project's structure

production-rag/
├── scripts/
│   └── evaluate.py                # the entry point
├── eval/
│   ├── __init__.py
│   ├── loader.py                  # loads the golden dataset
│   ├── runner.py                  # runs the RAG over the queries
│   ├── metrics.py                 # the RAGAS wrapper
│   ├── reporter.py                # generates the reports
│   └── thresholds.py              # the quality gates
├── golden_dataset/
│   └── v1.0.0.json
├── eval_reports/                  # gitignored, versioned separately
│   ├── 2026-03-13_a3f29c1.json
│   └── 2026-03-13_a3f29c1.md
└── pyproject.toml

The implementation: the 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 balanced queries: 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

Why filter_by_mode: local development with 100 queries costs time and tokens. Smoke mode gives feedback in 30 seconds without sacrificing type coverage. The PR uses smoke; nightly uses full.

The implementation: the 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])

The key decisions:

• concurrency=5 avoids OpenAI's rate limits while still parallelizing the retrieval (each query consumes embeddings + a chat completion).
• We capture retrieved_source_ids separately from contexts so we can compute precision@k against expected_sources.
• latency_ms gets persisted so we can correlate quality with performance in the report.

The implementation: the 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:
    """Metrics segmented by difficulty and 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

Why segmented metrics: a system can have a global faithfulness=0.90 but faithfulness=0.65 on hard queries. The average hides the problem. Segmenting by difficulty warns you about specific regressions.

The implementation: the 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

The 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())

Usage:

# Local development (fast)
python scripts/evaluate.py --mode smoke

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

# The full nightly run
python scripts/evaluate.py --mode full

Comparison: ad-hoc vs an automated pipeline

Connection with the final project

Your Advanced RAG System must include the eval/ directory with the modules above, and the evaluate.py command documented in the README:

## Evaluation

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

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

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

Reports get generated in eval_reports/{timestamp}_{commit}.{json,md}.

This documentation is what any new team member reads first.

---

## Troubleshooting

### Problem 1: "The pipeline takes too long locally"
**The cause:** you run the full set on every iteration.  
**The fix:** smoke mode (10 queries) for iteration. Reserve full for pre-merge and nightly.

### Problem 2: "Results aren't comparable between runs"
**The cause:** the dataset changes, the models change, the prompts change with no tracking.  
**The fix:** version the dataset, pin the model in the config, record the `commit_sha` in every report. If the scores change for no apparent reason, diff the JSON reports.

### Problem 3: "The pipeline fails with an OpenAI rate limit"
**The cause:** `concurrency` set too high, or an OpenAI tier with low limits.  
**The fix:** lower it to `concurrency=3`, and add retry with exponential backoff in `execute_one`.

### Problem 4: "The reports take up too much space"
**The cause:** you're versioning the reports in git.  
**The fix:** gitignore `eval_reports/` and store them in S3/separate storage. Only the pipeline's code goes into git.

### Problem 5: "A change in RAGAS breaks the historical reports"
**The cause:** RAGAS updates its metric definitions between versions.  
**The fix:** pin RAGAS in `pyproject.toml`. When you upgrade, run the evaluation against the latest version of the dataset and record the "discontinuity point" in the CHANGELOG.

---

## Exercises

### Exercise 1: Implementing the loader with validation

Implement `load_golden_dataset` with schema validation and clear error handling.

<details>
<summary>See the solution</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}")

The explanation: errors with specific messages speed up debugging. Without this, a typo in the JSON turns into an incomprehensible stacktrace.

Exercise 2: Implementing retry with backoff

Add exponential retry to execute_one to handle the 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}")

Exercise 3: A change report between runs

Implement a function that compares two reports and generates a summary of the changes.

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)

Exercise 4: A selective mode by category

Extend filter_by_mode to allow filtering by a specific category.

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

# Usage: evaluate only the hard security queries
records = filter_dataset(dataset, mode="full", categories=["security"], difficulties=["hard"])

Summary

• An automated pipeline turns evaluation from an experiment into a daily practice
• Five decoupled stages: load → execute → collect → metrics → report
• Smoke mode (10 queries) for the PR; full (100+) for nightly
• Reports in JSON (machine-readable) + Markdown (human-readable)
• Metrics segmented by difficulty catch specific regressions
• Exponential retry with jitter handles OpenAI's rate limits
• The diff between runs is the foundation of the regression testing in capsule 06

Additional resources

1. RAGAS Getting Started - A basic run.
2. Pydantic Settings - Configuring pipelines.
3. argparse Tutorial - An ergonomic CLI.
4. MLflow Tracking - An alternative for versioned reports.
5. Async Python Patterns - Controlled concurrency.

Created: March 13, 2026
Version: 2.0