Measure RAG retrieval properly: recall@k, MRR and a golden set that catches regressions

#When to use this

Users say the assistant "misses things", and nobody can say whether that's the model or the search. Or you're changing a chunk size, an embedding model or a reranker, and you have no numbers to show whether it helped.

Retrieval failures limit everything after them: if the right chunk never reaches the model, no prompt can recover it. So measure retrieval by itself. It's also fast and cheap to test, because no LLM call is needed.

#Step 1: build a golden set

A golden set is a list of real questions, each with the documents that should be found.

{"id": "q001", "query": "How do I rotate the VPN gateway certificate?", "relevant": ["runbook-vpn-cert", "kb-1042"], "tags": ["networking"]}
{"id": "q002", "query": "error 9002 transaction log full", "relevant": ["sql-log-full"], "tags": ["exact-term", "database"]}
{"id": "q003", "query": "who approves production firewall changes", "relevant": ["policy-change-mgmt"], "tags": ["policy"]}

How to build a useful one:

Match at the level of document or stable section ID, not chunk ID. Chunk IDs change whenever you re-chunk, and that would invalidate the set just when you need to compare.

Warning

Real questions may contain personal or sensitive data. Clean them before you add them to a repository, and apply the same access rules as the source data. See PII redaction.

#Step 2: choose metrics

Choose k to match what your app really sends to the model. If you pass 5 chunks, report recall@5. A good recall@50 doesn't help you if only 5 get used, though it tells you a reranker has something to work with.

#Step 3: compute them

This has no dependency on any vector store. You supply a retrieve function:

import json, math, statistics
from typing import Callable

def load_golden(path):
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]

def dedupe(doc_ids):
    """Several chunks can come from one document; keep first-seen order."""
    return list(dict.fromkeys(doc_ids))

def recall_at_k(ranked, relevant, k):
    return len(set(ranked[:k]) & relevant) / len(relevant)

def hit_at_k(ranked, relevant, k):
    return float(bool(set(ranked[:k]) & relevant))

def reciprocal_rank(ranked, relevant):
    for i, doc in enumerate(ranked, start=1):
        if doc in relevant:
            return 1.0 / i
    return 0.0

def ndcg_at_k(ranked, relevant, k):
    dcg = sum(1 / math.log2(i + 1) for i, d in enumerate(ranked[:k], 1) if d in relevant)
    ideal = sum(1 / math.log2(i + 1) for i in range(1, min(len(relevant), k) + 1))
    return dcg / ideal if ideal else 0.0

def evaluate(golden, retrieve: Callable[[str, int], list[str]], k=5, fetch=20):
    per_query = []
    for item in golden:
        relevant = set(item["relevant"])
        if not relevant:
            continue          # score no-answer questions separately
        ranked = dedupe(retrieve(item["query"], fetch))
        per_query.append({
            "id": item["id"],
            "tags": item.get("tags", []),
            "hit": hit_at_k(ranked, relevant, k),
            "recall": recall_at_k(ranked, relevant, k),
            "mrr": reciprocal_rank(ranked, relevant),
            "ndcg": ndcg_at_k(ranked, relevant, k),
            "top": ranked[:k],
        })
    summary = {m: round(statistics.mean(q[m] for q in per_query), 3)
               for m in ("hit", "recall", "mrr", "ndcg")}
    return summary, per_query

Adapter for pgvector, as an example:

import psycopg

def make_retriever(conn, embed):
    def retrieve(query, n):
        vector = embed(query)
        rows = conn.execute(
            """SELECT doc_id FROM chunks
               ORDER BY embedding <=> %s::vector LIMIT %s""",
            (str(vector), n),
        ).fetchall()
        return [r[0] for r in rows]
    return retrieve

Fetch more chunks than k, because deduplicating by document shortens the list.

#Step 4: read the failures

The average tells you whether there's a problem. The failing queries tell you why.

summary, rows = evaluate(golden, retrieve, k=5)
print(summary)

for row in sorted(rows, key=lambda r: r["recall"])[:15]:
    print(row["id"], row["recall"], row["top"])

Break it down by tag too, so a good average can't hide a weak area:

from collections import defaultdict

by_tag = defaultdict(list)
for row in rows:
    for tag in row["tags"]:
        by_tag[tag].append(row["recall"])

for tag, values in sorted(by_tag.items()):
    print(f"{tag:15} n={len(values):3} recall={statistics.mean(values):.2f}")

Then match the pattern to a fix:

#Step 5: compare changes fairly

Change one thing at a time, and compare per query, not just on averages. With small sets, differences can be pure noise.

import random

def paired_bootstrap(base_rows, new_rows, metric="recall", n=5000, seed=7):
    rng = random.Random(seed)
    base = {r["id"]: r[metric] for r in base_rows}
    new = {r["id"]: r[metric] for r in new_rows}
    diffs = [new[i] - base[i] for i in base if i in new]
    means = sorted(
        statistics.mean(rng.choices(diffs, k=len(diffs))) for _ in range(n)
    )
    return {
        "mean_diff": round(statistics.mean(diffs), 3),
        "ci_95": (round(means[int(n * .025)], 3), round(means[int(n * .975)], 3)),
        "better": sum(d > 0 for d in diffs),
        "worse": sum(d < 0 for d in diffs),
    }

If the 95% interval includes zero, you haven't shown an improvement. Also look at which queries got worse, as a net gain might still break something critical. Measure latency and cost at the same time, because a small recall gain may not justify a large increase in either.

#Step 6: make it a CI gate

Store a baseline with the code, and fail the build on a real drop:

import json, sys

baseline = json.load(open("eval/baseline.json"))
summary, _ = evaluate(golden, retrieve, k=5)

failed = [m for m in ("recall", "mrr")
          if summary[m] < baseline[m] - 0.02]

print(summary)
if failed:
    print("Regression in:", failed)
    sys.exit(1)

Run it whenever the embedding model, chunking, index settings or retrieval code changes. Use a fixed snapshot of the index for repeatable results, or expect some drift when content changes.

Caution

Don't update the baseline just to get a build to pass. Require a person to review the change, and to say why it's acceptable. Otherwise, your gate slowly lowers itself.

#After retrieval: measure answers separately

Retrieval metrics don't tell you if the model used the context correctly. Run these as a separate stage, using LLM-as-judge evaluation. That way, you can tell where a failure comes from:

#Prevent it next time

• Grow the golden set from production failures every month.
• Track recall by topic over time, not just one overall number.
• Rerun after every re-index. See quality regression after re-index.
• Related: evaluation strategy and retrieval budgets.