Test RAG Retrieval Recall at K

TL;DR

Label the relevant chunk IDs for each evaluation query, retrieve the top K results, and compute the fraction of labeled IDs that were returned. Test several K values, inspect zero-recall queries, and fail CI when critical queries miss or the aggregate score drops beyond an agreed tolerance.

Key Takeaways

• Recall at K measures how much labeled relevant evidence appears in the first K retrieved results.
• Stable document or chunk IDs are essential because text matching makes evaluation brittle.
• Report both macro recall and query-level results so easy queries cannot hide complete misses.
• Evaluate several K values to expose the tradeoff between retrieval coverage and context budget.
• Use category slices and failure diagnostics before changing embeddings, chunking, or search settings.
• Gate regressions with a baseline delta and a critical-query rule, not an arbitrary universal score.

To test RAG retrieval recall at K, create queries with known relevant document or chunk IDs, run your real retriever, and divide the number of relevant IDs found in the top K by the total number of relevant IDs. This isolates retrieval coverage before generation can hide a missing-evidence problem.

This tutorial builds a runnable Python evaluation harness, adds regression assertions, and explains how to read failures. For the end-to-end evaluation strategy around retrieval, generation, faithfulness, and operations, use the RAG application evaluation complete guide.

You will use a deterministic in-memory retriever first, then connect the same evaluator to your production adapter. The small fixture makes the metric easy to verify, while the interface keeps the test useful when your vector database, embedding model, or reranker changes.

What You Will Build

By the end, you will have:

• A JSONL dataset containing queries, relevant chunk IDs, categories, and criticality.
• A typed retriever contract that any search backend can implement.
• Recall at K calculations for K values of 1, 3, and 5.
• Query-level diagnostics that show missing evidence and retrieved rankings.
• Pytest regression gates for aggregate quality and critical queries.
• A JSON report suitable for CI artifacts and release comparison.

Recall at K answers a narrow question: did retrieval surface the evidence that a correct answer requires? It does not judge ranking order within K, context cleanliness, answer quality, or citation validity. Keep that boundary clear so every failed test points to the right subsystem.

Prerequisites

Use Python 3.11 or newer. The tutorial relies only on the standard library for the evaluator and pytest 8 or newer for regression tests. Create an empty folder and run:

python -m venv .venv
source .venv/bin/activate
python -m pip install "pytest>=8,<10"
python --version
pytest --version

On Windows PowerShell, activate with .venv\Scripts\Activate.ps1. You also need access to the same retrieval path used by your RAG application. That path should return stable IDs plus scores, not only concatenated text.

Use frozen evaluation data for regression runs. If the corpus changes, record its version because deleted, merged, or re-chunked evidence can make a valid test label impossible to retrieve.

Verification: python --version prints 3.11 or later, and pytest --version prints an installed pytest version.

Step 1: Test RAG Retrieval Recall at K with a Clear Definition

For a query q, let G(q) be the set of ground-truth relevant IDs and R_k(q) be the IDs in the first K retrieval results. Compute:

Recall@K(q) = |G(q) intersect R_k(q)| / |G(q)|

If a query needs chunks refund-01 and refund-02, but the top three results contain only refund-01, Recall@3 is 1 / 2 = 0.5. If both appear, it is 1.0. If neither appears, it is 0.0.

This definition treats every labeled relevant item equally. It is appropriate when all listed chunks contain necessary or acceptable evidence. Decide your labeling policy before collecting data:

For alternatives, model groups rather than putting every substitute into one required set. This tutorial uses required evidence because its meaning is unambiguous. Do not include queries with an empty relevant-ID list in recall aggregation. They need a separate unanswerable-query test.

Verification: For ground truth {a, b} and retrieved IDs [a, x, b], manually confirm Recall@1 is 0.5 and Recall@3 is 1.0.

Step 2: Create a Labeled Retrieval Dataset

Create eval_dataset.jsonl with one JSON object per line:

{"query_id":"q001","query":"How long do I have to request a refund?","relevant_ids":["refund-window"],"category":"refunds","critical":true}
{"query_id":"q002","query":"Can an admin export an audit log and which format is used?","relevant_ids":["audit-export","audit-format"],"category":"security","critical":true}
{"query_id":"q003","query":"Where can I change notification frequency?","relevant_ids":["notification-settings"],"category":"settings","critical":false}
{"query_id":"q004","query":"What happens when a subscription payment fails?","relevant_ids":["payment-retry","payment-suspension"],"category":"billing","critical":false}

Build labels by asking subject-matter experts to identify evidence in a versioned corpus. Store the smallest stable unit your retriever returns, usually a chunk ID. Avoid copying document titles into relevant_ids when the system ranks chunks because document-level matching can count the wrong passage as a hit.

Include realistic wording from users, not only text copied from source headings. Add categories for slices such as product area, language, question type, freshness, or tenant. Mark a small set as critical when a complete miss would create material risk. For broader data design, follow the adversarial RAG evaluation dataset tutorial.

Validate labels independently. A second reviewer should check ambiguous cases and confirm that every labeled ID exists in the corpus snapshot. Store dataset and corpus versions beside each report.

Verification: Run python -m json.tool against each line, or load the file in Step 4. Confirm every query_id is unique, every relevant_ids array is nonempty, and every ID exists in the indexed corpus.

Step 3: Implement a Retriever Contract

Create retrieval_eval.py. Start with typed result and retriever definitions, then add a deterministic fixture that simulates ranked search:

from __future__ import annotations

from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Protocol
import argparse
import json


@dataclass(frozen=True)
class SearchResult:
    item_id: str
    score: float


class Retriever(Protocol):
    def search(self, query: str, limit: int) -> list[SearchResult]: ...


class FixtureRetriever:
    def __init__(self) -> None:
        self.rankings = {
            "How long do I have to request a refund?": [
                SearchResult("refund-window", 0.94),
                SearchResult("payment-retry", 0.61),
            ],
            "Can an admin export an audit log and which format is used?": [
                SearchResult("audit-export", 0.91),
                SearchResult("security-overview", 0.83),
                SearchResult("audit-format", 0.79),
            ],
            "Where can I change notification frequency?": [
                SearchResult("profile-settings", 0.88),
                SearchResult("notification-settings", 0.84),
            ],
            "What happens when a subscription payment fails?": [
                SearchResult("payment-retry", 0.93),
                SearchResult("billing-overview", 0.82),
                SearchResult("payment-suspension", 0.77),
            ],
        }

    def search(self, query: str, limit: int) -> list[SearchResult]:
        return self.rankings.get(query, [])[:limit]

The evaluator depends only on search(query, limit). Your production adapter can call Pinecone, Qdrant, Elasticsearch, OpenSearch, PostgreSQL with pgvector, or an internal API, then map each hit to SearchResult. Preserve the backend ranking and return at most limit items.

Deduplicate IDs before scoring if hybrid retrieval can return the same chunk through multiple channels. Record the exact query after any application-side rewriting when you want to test the full retrieval pipeline. Use the raw query when you specifically want to isolate the search backend.

Verification: Run python -c "from retrieval_eval import FixtureRetriever; print(FixtureRetriever().search('How long do I have to request a refund?', 1))". The output contains refund-window and exactly one result.

Step 4: Calculate Query and Aggregate Recall

Append the evaluation models and functions to retrieval_eval.py:

@dataclass(frozen=True)
class EvalCase:
    query_id: str
    query: str
    relevant_ids: list[str]
    category: str
    critical: bool


@dataclass(frozen=True)
class QueryScore:
    query_id: str
    category: str
    critical: bool
    k: int
    recall: float
    relevant_ids: list[str]
    retrieved_ids: list[str]
    missing_ids: list[str]


def load_cases(path: Path) -> list[EvalCase]:
    cases = []
    for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
        if not line.strip():
            continue
        data = json.loads(line)
        case = EvalCase(**data)
        if not case.relevant_ids:
            raise ValueError(f"{path}:{line_number} has no relevant_ids")
        cases.append(case)
    if len({case.query_id for case in cases}) != len(cases):
        raise ValueError("query_id values must be unique")
    return cases


def recall_at_k(relevant_ids: list[str], retrieved_ids: list[str], k: int) -> float:
    if k < 1:
        raise ValueError("k must be at least 1")
    relevant = set(relevant_ids)
    if not relevant:
        raise ValueError("relevant_ids must not be empty")
    retrieved = set(retrieved_ids[:k])
    return len(relevant & retrieved) / len(relevant)


def evaluate(retriever: Retriever, cases: list[EvalCase], k: int) -> list[QueryScore]:
    scores = []
    for case in cases:
        results = retriever.search(case.query, limit=k)
        retrieved_ids = list(dict.fromkeys(result.item_id for result in results))
        relevant = set(case.relevant_ids)
        scores.append(QueryScore(
            query_id=case.query_id,
            category=case.category,
            critical=case.critical,
            k=k,
            recall=recall_at_k(case.relevant_ids, retrieved_ids, k),
            relevant_ids=case.relevant_ids,
            retrieved_ids=retrieved_ids,
            missing_ids=sorted(relevant - set(retrieved_ids[:k])),
        ))
    return scores


def macro_recall(scores: list[QueryScore]) -> float:
    if not scores:
        raise ValueError("scores must not be empty")
    return sum(score.recall for score in scores) / len(scores)

Macro recall averages query scores, so every query has equal influence. Micro recall instead sums all hits and divides by all labeled items, giving more weight to queries with more labels. Report macro recall as the primary regression metric and add micro recall only when that weighting is intentional.

Sets prevent duplicate retrieved IDs from creating false credit. The slice retrieved_ids[:k] enforces K even if a faulty adapter returns too many hits.

Verification: Run python -c "from retrieval_eval import recall_at_k; assert recall_at_k(['a','b'], ['a','x','b'], 1) == 0.5; assert recall_at_k(['a','b'], ['a','x','b'], 3) == 1.0; print('ok')". It prints ok.

Step 5: Test Multiple K Values and Produce a Report

Append this runner to retrieval_eval.py:

def run(dataset: Path, output: Path, ks: list[int]) -> dict:
    cases = load_cases(dataset)
    retriever = FixtureRetriever()
    runs = []
    for k in sorted(set(ks)):
        scores = evaluate(retriever, cases, k)
        runs.append({
            "k": k,
            "macro_recall": round(macro_recall(scores), 4),
            "zero_recall_count": sum(score.recall == 0 for score in scores),
            "queries": [asdict(score) for score in scores],
        })
    report = {
        "dataset": str(dataset),
        "metric": "recall_at_k",
        "runs": runs,
    }
    output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    return report


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dataset", type=Path, default=Path("eval_dataset.jsonl"))
    parser.add_argument("--output", type=Path, default=Path("recall_report.json"))
    parser.add_argument("--k", type=int, nargs="+", default=[1, 3, 5])
    args = parser.parse_args()
    report = run(args.dataset, args.output, args.k)
    for item in report["runs"]:
        print(f"Recall@{item['k']}: {item['macro_recall']:.4f} "
              f"(zero-recall queries: {item['zero_recall_count']})")


if __name__ == "__main__":
    main()

Run the evaluation:

python retrieval_eval.py --dataset eval_dataset.jsonl --output recall_report.json --k 1 3 5

Choose K values that reflect real context limits. Recall should be nondecreasing as K grows for a fixed ranking. A falling value indicates unstable retrieval, inconsistent queries, filtering, or evaluator defects. Higher K is not automatically better because extra passages consume tokens and may distract generation. Measure context quality separately with the RAG context precision with Ragas tutorial.

Verification: The fixture prints Recall@1 of 0.5000 and Recall@3 of 1.0000. Recall@5 is also 1.0000 because no additional required evidence is missing. Open recall_report.json and confirm it contains four query records for each K.

Step 6: Test RAG Retrieval Recall at K in Pytest

Create test_retrieval_recall.py:

from pathlib import Path

import pytest

from retrieval_eval import FixtureRetriever, evaluate, load_cases, macro_recall, recall_at_k


def test_recall_at_k_boundaries() -> None:
    assert recall_at_k(["a", "b"], ["a", "x", "b"], 1) == 0.5
    assert recall_at_k(["a", "b"], ["a", "x", "b"], 3) == 1.0
    with pytest.raises(ValueError):
        recall_at_k([], ["a"], 1)


def test_retrieval_recall_at_3() -> None:
    cases = load_cases(Path("eval_dataset.jsonl"))
    scores = evaluate(FixtureRetriever(), cases, k=3)
    assert macro_recall(scores) >= 0.95
    assert all(score.recall > 0 for score in scores if score.critical)


def test_recall_is_nondecreasing() -> None:
    cases = load_cases(Path("eval_dataset.jsonl"))
    retriever = FixtureRetriever()
    by_k = [macro_recall(evaluate(retriever, cases, k)) for k in (1, 3, 5)]
    assert by_k == sorted(by_k)

Run pytest -q. The first test protects metric behavior. The second enforces an illustrative project threshold and forbids complete misses for critical queries. The third catches inconsistent ranking or adapter behavior.

Do not copy the 0.95 threshold blindly. Establish a reviewed baseline on representative data, then set a minimum score and allowed regression delta based on risk. A useful gate can require candidate >= minimum and candidate >= baseline - tolerance. Version changes to labels, corpus, embedding model, filters, and rerankers so reviewers can explain score movement.

Verification: The command prints 3 passed. Temporarily move audit-format below rank 3 in the fixture and confirm test_retrieval_recall_at_3 fails. Restore it after proving the gate detects a regression.

Step 7: Diagnose Retrieval Recall Failures

A scalar score tells you that coverage changed, not why. Start with query records where recall == 0, then partial misses, then the largest category regression. Compare missing IDs with the retrieved IDs and scores.

Use this diagnosis table:

Change one retrieval variable at a time. Re-run the identical dataset and compare query-level deltas, not only the mean. Inspect whether gains come from hard queries or merely reorder already successful cases.

Recall can be perfect while the answer cites the wrong source or makes unsupported claims. After retrieval passes, use RAG citation correctness examples to test the generated answer and its source mapping.

Verification: Pick one deliberately failing query, write a one-sentence root-cause hypothesis, make one controlled change, and confirm the report shows the expected query-level movement without unrelated category regressions.

Step 8: Run the Test in CI

Use the same command locally and in automation. A minimal GitHub Actions job is:

name: Retrieval evaluation

on:
  pull_request:
  workflow_dispatch:

jobs:
  recall-at-k:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: python -m pip install "pytest>=8,<10"
      - run: pytest -q test_retrieval_recall.py
      - run: python retrieval_eval.py --k 1 3 5
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: retrieval-recall-report
          path: recall_report.json

For a remote retriever, pin an index snapshot or use a dedicated evaluation environment. Cache neither stale credentials nor unversioned query results. Protect secrets with the CI platform and avoid logging private query text. If remote search is nondeterministic, run repeated trials and gate a documented aggregate, or configure deterministic search where supported.

Keep fast, deterministic retrieval tests on pull requests. Run larger multilingual, adversarial, and production-sampled suites nightly or before releases. Archive reports with commit, dataset, corpus, and retrieval configuration identifiers.

Verification: Open a pull request and confirm the job passes, the report artifact is downloadable, and a deliberately lowered fixture ranking makes the job fail. Confirm if: always() still uploads diagnostics on that failed run.

Troubleshooting

Recall is always zero -> Check the identifier level first. Ground truth may contain document IDs while retrieval returns chunk IDs. Print both lists for one query and normalize only documented prefixes, never approximate text.

Recall changes between identical runs -> Freeze the index, filters, query rewriting, and reranker configuration. Check whether approximate nearest-neighbor search or a remote service introduces variability. Record repeated-run distributions if determinism is unavailable.

Recall falls after re-chunking -> Migrate labels to the new stable IDs through a reviewed mapping, or relabel against the new corpus. Do not compare incompatible ID spaces and call the difference a model regression.

A valid alternative answer is marked wrong -> Improve the labeling model. Represent acceptable evidence groups and award a hit when at least one member of each required group is returned. Do not add all alternatives to a flat required set.

Recall improves only when K is huge -> Inspect rank positions and precision. Try hybrid retrieval or reranking instead of sending excessive context to the generator. Large K can move the retrieval problem downstream.

The aggregate passes while important queries fail -> Add critical-query assertions and category floors. Always report zero-recall counts because an average can hide catastrophic misses.

Best Practices and Common Mistakes

Do use stable IDs, representative queries, reviewed labels, corpus versioning, category slices, several K values, and query-level diagnostics. Keep the evaluator separate from generation so retrieval failures remain attributable.

Do not use the answer generated by the same RAG system as ground truth. Do not judge recall through string similarity between chunks. Do not tune on the only evaluation dataset, and do not celebrate higher recall without checking added noise, latency, and context cost.

Treat thresholds as product decisions supported by evidence. A medical policy assistant and an internal FAQ bot do not share the same consequence of a missed source. Critical cases may require 1.0 recall at the production K even when the broader suite allows partial recall.

Where To Go Next

You now have a retrieval coverage gate. Place it inside the broader complete RAG application evaluation workflow, then add complementary tests:

• Measure signal-to-noise with RAG context precision using Ragas.
• Verify source claims with RAG citation correctness examples.
• Expand hard cases with an adversarial RAG evaluation dataset.

Next, replace FixtureRetriever with a thin adapter around your real search API. Freeze the dataset and index version, save the first accepted report as a baseline, and review every query-level regression before changing the threshold.

Interview Questions and Answers

Q: What does Recall at K measure in a RAG system?

It measures the fraction of labeled relevant evidence retrieved within the first K results. It evaluates coverage, not answer correctness or ranking quality inside the cutoff.

Q: Why should you test several K values?

Several cutoffs reveal whether evidence is highly ranked or appears only after many distractors. The curve helps select a production K that balances coverage with context size, latency, and noise.

Q: What is the difference between macro and micro recall?

Macro recall averages per-query recall and weights every query equally. Micro recall combines hits and labels across queries, so queries with more relevant items carry more weight.

Q: How do you handle multiple acceptable evidence chunks?

Represent alternatives as groups and require a hit from each necessary group. A flat set incorrectly treats every alternative as mandatory and understates valid recall.

Q: Can Recall at K evaluate an entire RAG application?

No. It isolates retrieval coverage. Add context precision, answer correctness, faithfulness, citation correctness, latency, and safety evaluations for end-to-end confidence.

Q: How should Recall at K be gated in CI?

Use a representative frozen dataset, a minimum acceptable score, a tolerated delta from a reviewed baseline, and zero-miss rules for critical queries. Store query-level reports as artifacts.

Conclusion

To test RAG retrieval recall at K reliably, label stable evidence IDs, run the real retrieval path at meaningful cutoffs, calculate query-level recall, and preserve diagnostics. Aggregate scores are useful release signals only when category slices and critical misses remain visible.

Start with the deterministic harness, verify its expected scores, and then swap in your production retriever. A versioned dataset plus explicit regression gates turns retrieval quality from a subjective demo into a repeatable QA check.

Interview Questions and Answers

Frequently Asked Questions

Related Guides

• Evaluate RAG Citation Correctness with Examples
• Evaluating RAG retrieval precision (2026)
• Selenium Test Web Components Slots Tutorial: Test Web Component Slots with Selenium
• Test scenarios vs test cases (2026)
• TypeScript Decorators Test Metadata Tutorial: Create Test Metadata
• A/B test validation: A Complete Guide for QA (2026)