Every stage of a RAG pipeline needs its own evaluation gate

A retrieval-augmented generation (RAG) pipeline is a chain of transformations: documents are parsed, split into chunks, embedded into vectors, indexed, retrieved against a query, and handed to a language model that writes the answer. Each step consumes the output of the one before it, so each can quietly degrade the rest. A parser that drops table cells, a chunker whose overlap setting reverted, an embedding model bumped to a new minor version — none of these raise an exception. The pipeline keeps running and the answers get worse.

The usual response is to measure the answer: run a set of questions, score the responses, watch the number. That is necessary and insufficient. When the score moves, the next question is which stage moved it, and without instrumentation at each boundary that means re-running the pipeline by hand with pieces swapped out. Evaluation hooks are the alternative — a small, always-on gate on every stage, measuring what that stage is responsible for.

The original note

AI RAG — with Evaluation Hooks

• Source Data → Data QA: schema checks, PII scan, dedup
• Ingestion → Chunk QA: length/distribution, overlap histograms
• Chunking → Embedding QA: norm distance, out-of-vocabulary %, drift
• Vectorization → Retrieval Eval: Precision@k, MRR, nDCG, Coverage
• Retrieval → Gen Eval: Factuality, Faithfulness, BLEU/ROUGE, Latency/Cost
• LLM Generation → Human/Online eval: thumbs, issue tags, CTR
• Answer

One end-to-end score cannot localize a regression

An end-to-end score compresses seven independent failure surfaces into one number, and that compression is what makes it useless for diagnosis. When the index is rebuilt overnight, the prompt is edited, and an upstream feed changes its export format in the same window, the score cannot separate those causes.

Stages also mask each other. A strong generator writes a plausible answer from a mediocre retrieval, holding the score flat while retrieval quality erodes underneath it. The reverse happens too: retrieval returns exactly the right passage and the model ignores it. Both produce the same top-line movement and demand opposite fixes.

Stage gates are the unit tests of a data pipeline, and the end-to-end score is the integration test. Integration tests say the system is broken; unit tests say where.

The gates, stage by stage

Data QA

Schema checks assert that every ingested record carries what the rest of the pipeline assumes: non-empty body text, a source identifier, a timestamp, a document type. Assert types and null rates, not just presence. A bad value looks like a null rate jumping from near zero to a few percent overnight, or a field whose type flips from string to list because an exporter changed.

PII scans look for emails, phone numbers, account identifiers, and national ID patterns in text about to be embedded and indexed. Retrieved text is pasted into prompts and usually written into request logs, so one leaked record propagates into several systems. For a corpus that is supposed to be clean, the only acceptable value is zero hits.

Dedup catches near-identical documents, using a normalized hash for exact matches and a similarity-preserving hash such as MinHash or SimHash for the rest. Duplicates compete for the same top-k slots, so a query returns five paraphrases of one paragraph instead of five distinct sources — a retrieval failure that presents as a generation problem.

Chunk QA

Length and distribution means tracking median, p95, and maximum chunk size in both characters and tokens, plus the fraction of chunks below a floor. A healthy distribution is tight and unimodal. A bad one is bimodal (two parsers disagreeing), has a long tail past the context budget, or shows a sudden spike of tiny fragments — the signature of a parser splitting on the wrong delimiter.

Overlap histograms verify that a sliding-window chunker actually produced the configured overlap. Compute the realized token overlap between adjacent chunks of the same document and histogram it. A histogram that collapses to zero means overlap was silently disabled, which surfaces much later as answers that lose any fact stated across a chunk boundary.

Embedding QA

Norm distance tracks the distribution of vector magnitudes, which is stable for a fixed model and preprocessing path. Any zero-norm vector means an empty input reached the encoder; a shifted median norm almost always means the model version, the pooling strategy, or the normalization step changed.

Out-of-vocabulary percentage is the share of tokens the tokenizer cannot map to real vocabulary entries and pushes into unknown or byte-fallback tokens. Rising OOV means a new language entered the corpus, an encoding is mangled, or PDF extraction produced ligature garbage.

Drift compares the current batch against a frozen reference batch: cosine distance between centroids, plus the distribution of pairwise similarities. Timing decides the reading. Drift with no deploy points at upstream content; drift right after a deploy points at the encoder.

Retrieval eval

This gate needs labels — a fixed set of queries, each mapped to the document IDs that should be retrieved.

Precision@k is the fraction of the top k results that are relevant, so it measures how much of the context window is spent on useful material. MRR (mean reciprocal rank) averages 1/rank of the first relevant result across queries, rewarding one good document at the very top — the right metric when the generator leans mostly on the first passage. nDCG (normalized discounted cumulative gain) handles graded relevance: each result contributes its gain divided by a logarithmic position discount, normalized against the best possible ordering. It separates a merely relevant hit from a highly relevant one and penalizes burying the best document deep in the list. Coverage is the share of queries where at least one relevant document appears in the top k, and it caps everything downstream: what is not retrieved cannot be used, and no reranker or prompt fixes it.

Gen eval

Faithfulness asks whether every claim in the answer is supported by the retrieved context. Factuality asks whether the claims are true. The difference is diagnostic: an answer can be perfectly faithful to a retrieved document that is out of date, which is a data problem, not a generation problem. Score both by decomposing an answer into atomic claims and checking each against its cited span.

BLEU and ROUGE measure n-gram overlap against reference answers. They are insensitive to meaning, so they are poor quality scores but useful deterministic tripwires: where reference answers exist and phrasing is constrained, a sharp drop signals that something changed.

Latency and cost are what every quality improvement trades against. Record p50 and p95 latency per stage, input and output token counts, and cost per answered query.

Human and online eval

Thumbs are sparse and skewed toward complaints, so track the rate over time rather than reading it as an absolute score. Issue tags are more valuable: a small fixed taxonomy — wrong document retrieved, right document but wrong answer, missing citation, stale content, unnecessary refusal — where every tag maps to the stage that owns it, so feedback routes to a gate instead of a backlog. Click-through rate on citations is a passive signal; when users stop opening cited sources, the citations have usually stopped matching the answer.

A harness that makes hooks cheap to add

Hooks only get written if adding one takes minutes. A registry keyed by stage name, plus a wrapper that times the stage and fans out to its hooks, is enough.

"""Minimal evaluation-hook harness for a staged pipeline."""
from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable, Protocol


@dataclass
class Metric:
    run_id: str
    stage: str
    name: str
    value: float
    meta: dict = field(default_factory=dict)


class EvalHook(Protocol):
    """Reads a stage output, returns zero or more metrics."""

    def __call__(self, out: Any, ctx: dict) -> list[Metric]: ...


REGISTRY: dict[str, list[EvalHook]] = {}


def hook(stage: str) -> Callable[[EvalHook], EvalHook]:
    """Register an eval function to run after `stage`."""

    def wrap(fn: EvalHook) -> EvalHook:
        REGISTRY.setdefault(stage, []).append(fn)
        return fn

    return wrap


class Pipeline:
    def __init__(self, sink: Callable[[Metric], None]) -> None:
        self.sink = sink
        self.run_id = uuid.uuid4().hex

    def stage(self, name: str, fn: Callable[[Any], Any],
              data: Any) -> Any:
        start = time.perf_counter()
        out = fn(data)
        ms = (time.perf_counter() - start) * 1000.0
        ctx = {"run_id": self.run_id, "stage": name}
        self.sink(Metric(self.run_id, name, "latency_ms", ms))
        for h in REGISTRY.get(name, []):
            for metric in h(out, ctx):
                self.sink(metric)
        return out


@hook("chunking")
def chunk_qa(out: list[str], ctx: dict) -> list[Metric]:
    rid = ctx["run_id"]
    lens = sorted(len(c) for c in out)
    if not lens:
        return [Metric(rid, "chunking", "count", 0.0)]
    p95 = lens[int(0.95 * (len(lens) - 1))]
    empty = sum(1 for n in lens if n < 32) / len(lens)
    return [
        Metric(rid, "chunking", "count", float(len(lens))),
        Metric(rid, "chunking", "len_p95", float(p95)),
        Metric(rid, "chunking", "empty_frac", empty),
    ]

The sink is deliberately a plain callable: a list in development, a row of (run_id, stage, metric, value, timestamp) in production. The embedding gate uses the same signature and needs nothing beyond numpy.

import numpy as np


def cosine(a: np.ndarray, b: np.ndarray) -> float:
    denom = float(np.linalg.norm(a) * np.linalg.norm(b))
    if denom == 0.0:
        return 0.0
    return float(np.dot(a, b) / denom)


@hook("vectorization")
def embedding_qa(out: dict, ctx: dict) -> list[Metric]:
    """out: {'vectors': ndarray, 'oov': int, 'tokens': int}"""
    rid = ctx["run_id"]
    vecs = np.asarray(out["vectors"], dtype=np.float64)
    norms = np.linalg.norm(vecs, axis=1)
    ref = np.load("eval/reference_centroid.npy")
    drift = 1.0 - cosine(vecs.mean(axis=0), ref)
    oov = out["oov"] / max(out["tokens"], 1)
    return [
        Metric(rid, "vectorization", "norm_p50",
               float(np.median(norms))),
        Metric(rid, "vectorization", "zero_norm_frac",
               float(np.mean(norms == 0.0))),
        Metric(rid, "vectorization", "oov_frac", float(oov)),
        Metric(rid, "vectorization", "centroid_drift", drift),
    ]

The retrieval metrics, written out

These are short enough that a dependency is unnecessary, and writing them out removes ambiguity about the definitions in use — particularly the position discount in nDCG, where implementations differ.

import math
from typing import Sequence

Ranked = Sequence[str]
Judged = Sequence[tuple[Ranked, set[str]]]


def precision_at_k(ranked: Ranked, rel: set[str], k: int) -> float:
    """Share of the top k results that are relevant."""
    if k <= 0:
        raise ValueError("k must be positive")
    hits = sum(1 for doc in ranked[:k] if doc in rel)
    return hits / float(k)


def reciprocal_rank(ranked: Ranked, rel: set[str]) -> float:
    """1 / rank of the first relevant hit, else 0."""
    for i, doc in enumerate(ranked, start=1):
        if doc in rel:
            return 1.0 / i
    return 0.0


def mrr(results: Judged) -> float:
    """Mean reciprocal rank across a batch of queries."""
    if not results:
        return 0.0
    total = sum(reciprocal_rank(r, rel) for r, rel in results)
    return total / len(results)


def dcg(gains: Sequence[float]) -> float:
    """Sum of gain / log2(position + 1), positions from 1."""
    return sum(g / math.log2(i + 1)
               for i, g in enumerate(gains, start=1))


def ndcg_at_k(ranked: Ranked, gains: dict[str, float],
              k: int) -> float:
    """DCG of the top k over the DCG of the ideal ordering."""
    actual = [gains.get(doc, 0.0) for doc in ranked[:k]]
    ideal = sorted(gains.values(), reverse=True)[:k]
    best = dcg(ideal)
    if best == 0.0:
        return 0.0
    return dcg(actual) / best


def coverage(results: Judged, k: int) -> float:
    """Share of queries with at least one hit in the top k."""
    if not results:
        return 0.0
    hit = sum(1 for ranked, rel in results
              if any(doc in rel for doc in ranked[:k]))
    return hit / len(results)

Two details matter when replacing an existing implementation. The discount is log2(position + 1) with positions starting at 1, so rank 1 has a discount of exactly 1. The ideal ordering comes from all judged gains for the query, not from the gains of the returned results — otherwise a ranking that retrieved nothing relevant would score 1.0.

How to add hooks to an existing pipeline

1. Freeze a golden query set. Pull real queries from logs, keep a few dozen to a couple hundred, and label the document IDs that should be retrieved for each. Include the awkward ones: multi-hop questions, questions with no good answer, near-duplicate phrasings. Version the set in the repository.
2. Record a baseline before changing anything. Run the current pipeline once and store every metric. Without a baseline, the first alert has no reference point and every threshold is a guess.
3. Give every stage one signature. Refactor stages into fn(input) -> output so the wrapper can time them and attach hooks uniformly. This is usually the only invasive change.
4. Emit to one table. (run_id, stage, metric, value, timestamp) for every hook, everywhere. One shape means one dashboard and the ability to diff two runs by run ID.
5. Put thresholds in config and start warn-only. Keep them beside the code but outside it, so tuning does not require a code review, and let them run without blocking until the noise level is known.
6. Promote the cheap deterministic gates to blocking first. Schema violations, PII hits, and zero-norm vectors are fast and unambiguous. Leave model-judged gates such as faithfulness advisory, re-run the suite on every index rebuild and every prompt or model change, and store the run ID alongside the artifact it produced.

# eval/thresholds.yaml
# Placeholders. Calibrate each value from a recorded baseline
# rather than adopting these as defaults.
source_data:
  pii_hits_max: 0            # blocking
  duplicate_frac_max: 0.02
chunking:
  len_p95_max: 1200          # characters
  empty_frac_max: 0.01
  overlap_zero_frac_max: 0.0
vectorization:
  zero_norm_frac_max: 0.0    # blocking
  oov_frac_max: 0.05
  centroid_drift_max: 0.05   # 1 - cosine vs reference
retrieval:
  precision_at_5_min: 0.40
  ndcg_at_10_min: 0.60
  coverage_at_10_min: 0.95
generation:
  faithfulness_min: 0.90     # advisory
  p95_latency_ms_max: 4000

Failure modes

The golden set rots. Queries drift away from what users ask and the corpus changes, so old labels go stale. Schedule a review, and treat a metric that only ever improves as evidence of overfitting rather than progress.

Labels inherit the current system’s bias. Judging only what the current retriever returned bakes today’s behavior into tomorrow’s target, so a retriever that finds a genuinely better document gets scored down for it. Mix in candidates from another method — lexical BM25 alongside dense retrieval, or a reranked pool — before labeling.

Metrics are computed against a different snapshot than the one served. Scores from a freshly rebuilt index mean nothing while production serves the previous one — stamp the run ID onto the index artifact.

Everything stays warn-only forever. Thresholds that never block become noise, and noise gets muted. Promote the deterministic gates to blocking within a few cycles or delete them.

The judge model changes underneath the metric. Faithfulness and factuality scored by a language model inherit that model’s version. Pin it, record the version in the metric metadata, and keep a small human-audited subset to detect judge drift.

What this buys you

The main return is localization. A regression that used to take a day of manual bisection becomes a glance at which stage’s gate turned red, because each stage is measured against its own expectation rather than the answer’s. That also makes component swaps safe: replacing an embedding model, adding cross-encoder reranking, or fusing lexical and dense results with reciprocal rank fusion becomes a comparison of stage-level numbers instead of an argument about whether the answers feel better.

The secondary return is a shared vocabulary. When someone reports a bad answer, the tag taxonomy routes it to a stage, the stage has a metric, and the metric has a history. That path from complaint to cause is the difference between debugging a pipeline and guessing at one.