Measuring retrieval precision and recall in production RAG

Most RAG systems ship blind. Measuring retrieval precision and recall RAG in production requires more than offline eval sets—you need instrumentation that captures what users actually retrieved versus what they needed, then computes metrics against labeled relevance. When you treat retrieval as a measurable component instead of a black box, you can catch regressions before they erode answer quality.

This guide walks through a concrete pipeline to instrument, compute, and monitor those metrics from real traffic. The steps assume a Python service and a vector store, but the schema and math transfer to any stack.

Step 1: Establish ground truth from production signals

You cannot compute retrieval precision and recall RAG metrics without a relevance label per retrieved chunk. In production, explicit user feedback is sparse but gold. Capture thumbs up/down on answers and map them to the retrieved context via a session ID. If a user downvotes an answer that quoted three chunks, treat those three as suspect; if they upvote, treat them as relevant only if the answer is correct (add a second signal for answer correctness).

Implicit signals help fill the gap. Log dwell time on the rendered answer and whether the user copied a retrieved snippet. A short dwell time followed by a reformulated query often means the retrieval missed. These are weak labels, so keep them in a separate weight class from explicit ratings.

For queries without feedback, synthesize pseudo-labels: pick a document chunk, ask an LLM to write a realistic user question that chunk answers, then treat that chunk as the sole relevant item. This gives you a steady stream of labeled pairs for shadow evaluation. Keep the synthetic set disjoint from training data to avoid inflated numbers.

import uuid, json, time

def log_feedback(session_id: str, chunk_ids: list[str], rating: int, implicit: bool = False):
    # rating: 1 positive, 0 negative; implicit flags weak labels
    event = {
        "type": "retrieval_feedback",
        "session_id": session_id,
        "chunk_ids": chunk_ids,
        "rating": rating,
        "implicit": implicit,
        "ts": time.time()
    }
    # push to your Kafka/topic or warehouse
    print(json.dumps(event))

Store these events alongside the original retrieval logs keyed by session_id. Retention of 90 days is enough to spot seasonal drift.

Step 2: Instrument the retriever to emit structured traces

Wrap your vector search call so every query records the candidate set. Include the score, the rank, and the index version. Without rank data you can only compute binary precision, not ranked variants like mean reciprocal rank. A retriever that returns five chunks but logs none of them is undebuggable; don’t ship that.

def retrieve(query: str, top_k: int = 5, index_version: str = "v2") -> list[dict]:
    results = vector_store.search(query, top_k=top_k)
    trace = {
        "session_id": uuid.uuid4().hex,
        "query": query,
        "top_k": top_k,
        "index_version": index_version,
        "retrieved": [
            {"chunk_id": r.id, "score": r.score, "rank": i}
            for i, r in enumerate(results)
        ],
        "ts": time.time()
    }
    emit_trace(trace)  # to same sink as feedback
    return results

Keep the trace lightweight. A 5 KB JSON per query is fine at moderate traffic; sample at 10% if you ingest millions of queries per day. Tag the trace with the experiment bucket if you run A/B retrieval strategies—otherwise you will average away a winning variant.

In the warehouse, define a table retrieval_traces(session_id, query, retrieved ARRAY, ts) and a table feedback(session_id, chunk_ids, rating, implicit). Joining them is a 30-second SQL task.

Step 3: Join traces with labels and compute metrics

Precision at k is the fraction of retrieved chunks in the top k that are relevant. Recall at k is the fraction of all relevant chunks (per query) that appear in the top k. If you only have one relevant chunk from synthetic data, recall is either 0 or 1. For multi-chunk relevant sets from explicit feedback, recall becomes meaningful.

def precision_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
    top = retrieved[:k]
    hits = sum(1 for c in top if c in relevant)
    return hits / k

def recall_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
    if not relevant:
        return 0.0
    top = retrieved[:k]
    hits = sum(1 for c in top if c in relevant)
    return hits / len(relevant)

Join the feedback table to traces on session_id. For synthetic labels, you already know the relevant chunk ID, so join on the generated query ID. Compute per-session, then average.

def evaluate_session(trace: dict, relevant: set[str], k: int = 5):
    retrieved_ids = [r["chunk_id"] for r in trace["retrieved"]]
    return {
        "p@k": precision_at_k(retrieved_ids, relevant, k),
        "r@k": recall_at_k(retrieved_ids, relevant, k),
    }

Run this over a rolling 24-hour window. Expect precision@5 to start low (0.3–0.6) on real feedback and near 0.8 on synthetic single-chunk tasks if your embedding model is decent. If synthetic precision is below 0.5, your chunking or embedding is broken—fix that before trusting any production number.

Step 4: Aggregate with bootstrap confidence intervals

A single day may yield only dozens of explicit feedback sessions. Use bootstrap resampling to estimate the uncertainty of your retrieval precision and recall RAG numbers. Point estimates from small samples lie.

import random

def bootstrap_ci(scores: list[float], n=1000, alpha=0.05):
    if len(scores) < 2:
        return (0.0, 0.0)
    means = []
    for _ in range(n):
        sample = random.choices(scores, k=len(scores))
        means.append(sum(sample) / len(sample))
    means.sort()
    lo = means[int(alpha/2 * n)]
    hi = means[int((1-alpha/2) * n)]
    return lo, hi

If the 95% CI of precision drops below your SLO (say 0.5), page the owning team. Do not trust point estimates from fewer than 30 sessions; instead, lean on the synthetic stream for daily signal and use explicit feedback as a weekly audit.

Step 5: Track drift and segment by query class

Overall metrics hide failures. Segment by query intent (navigational, factual, long-tail) using a simple classifier or the first token. A drop in recall on long-tail queries often signals embedding drift after a corpus update, while a precision drop on navigational queries means your metadata filter is misrouting.

def segment_metrics(rows: list[dict]):
    segs = {}
    for row in rows:
        seg = row.get("intent", "unknown")
        segs.setdefault(seg, {"p": [], "r": []})
        segs[seg]["p"].append(row["p@k"])
        segs[seg]["r"].append(row["r@k"])
    return {
        k: {"p_ci": bootstrap_ci(v["p"]), "r_ci": bootstrap_ci(v["r"])}
        for k, v in segs.items()
    }

Plot these weekly. If factual recall falls from 0.7 to 0.4 after a vector index rebuild, you have a concrete rollback trigger. Set the alert on the lower CI bound, not the mean, to avoid false positives from noise.

Step 6: Scale label generation without stalling

Synthetic labeling at scale needs an LLM to draft questions from chunks. If you batch thousands of chunks nightly, provider rate limits will bite. Routing those judgment calls through n4n.ai gives you automatic fallback when a provider is degraded, so the eval pipeline keeps producing labels instead of timing out. The gateway also meters per-token usage, so you can attribute eval cost separately from serving cost.

Use a fixed model directive and forward cache-control to avoid re-paying for stable chunks:

curl https://api.n4n.ai/v1/chat/completions \
  -H "Authorization: Bearer $KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "openai/gpt-4o-mini",
    "messages": [{"role":"user","content":"Write a query for: {{chunk}}"}],
    "cache_control": {"type":"ephemeral"}
  }'

Honor the client routing directive if you need to pin a specific provider for reproducibility. The fallback only triggers when the primary is rate-limited or erroring, which is exactly what you want for a background job.

Verify success

You have a working measurement loop when:

1. Every production query emits a trace with session_id and ranked chunk IDs.
2. At least one labeled relevance signal (feedback or synthetic) exists for >5% of daily queries.
3. A scheduled job computes precision@5 and recall@5 with bootstrap CIs and writes them to a dashboard.
4. An alert fired at least once in staging when you artificially degraded the retriever (e.g., shuffled results) to prove the signal works.
5. Segmented metrics show stable CIs over two weeks of normal traffic.

If you can meet those five conditions, you are measuring retrieval precision and recall RAG in production rather than guessing. The numbers will be noisy, but they beat a silent vector store that breaks without anyone noticing.