RAG

RAG Pipeline Testing: Retrieval Quality, Grounding, and Faithfulness Metrics

HelpMeTest

HelpMeTest

— 6 min read
RAG Pipeline Testing: Retrieval Quality, Grounding, and Faithfulness Metrics

RAG (Retrieval-Augmented Generation) systems fail in ways that are invisible without the right tests. The LLM might generate confident-sounding output that's completely ungrounded in the retrieved documents. The retriever might fetch irrelevant documents. The response might be faithful to the retrieved context but that context might be wrong. Testing RAG requires evaluating the retrieval step and the generation step separately, then together.

The RAG Failure Taxonomy

Before writing tests, understand how RAG fails:

Failure type Description Detection
Retrieval miss Right documents not retrieved Recall@K against golden corpus
Irrelevant retrieval Retrieved docs don't answer the question Context relevance metric
Faithfulness failure Response claims not in retrieved docs NLI/entailment checking
Grounding failure Response correct but not based on context Faithfulness metric
Hallucination LLM adds information not in retrieved docs LLM judge
Context window failure Too much context dilutes the answer Context utilization metric

Component 1: Testing the Retriever

The retriever is testable independently of the LLM. For a given query, it should retrieve the most relevant documents.

Setting Up Retrieval Evaluation

# tests/rag/test_retrieval.py
import pytest
from typing import List

from src.rag.retriever import Retriever
from src.rag.document_store import DocumentStore

@pytest.fixture(scope="module")
def retriever():
    """Load retriever with test document corpus."""
    store = DocumentStore()
    store.load_documents("tests/fixtures/test_corpus/")  # Your test documents
    return Retriever(document_store=store, top_k=5)

# Golden retrieval dataset: query → expected document IDs
RETRIEVAL_TEST_CASES = [
    {
        "id": "ret-001",
        "query": "How much does the Pro plan cost?",
        "relevant_doc_ids": ["pricing-page", "pro-plan-features"],
        "irrelevant_doc_ids": ["refund-policy", "enterprise-sla"],
    },
    {
        "id": "ret-002",
        "query": "How do I integrate with GitHub Actions?",
        "relevant_doc_ids": ["ci-cd-guide", "github-actions-tutorial"],
        "irrelevant_doc_ids": ["pricing-page", "mobile-testing"],
    },
    {
        "id": "ret-003",
        "query": "What testing frameworks are supported?",
        "relevant_doc_ids": ["supported-frameworks", "framework-integrations"],
        "irrelevant_doc_ids": ["billing-faq"],
    },
]

def test_retrieval_recall_at_5(retriever):
    """At least one relevant document must be in top-5 results."""
    failures = []
    
    for case in RETRIEVAL_TEST_CASES:
        results = retriever.retrieve(case["query"])
        retrieved_ids = {r.doc_id for r in results}
        
        found_relevant = any(doc_id in retrieved_ids for doc_id in case["relevant_doc_ids"])
        
        if not found_relevant:
            failures.append({
                "id": case["id"],
                "query": case["query"],
                "expected": case["relevant_doc_ids"],
                "retrieved": list(retrieved_ids),
            })
    
    assert not failures, f"Retrieval failures:\n{failures}"

def test_retrieval_precision_no_irrelevant_in_top_3(retriever):
    """Top 3 results should not contain known-irrelevant documents."""
    failures = []
    
    for case in RETRIEVAL_TEST_CASES:
        results = retriever.retrieve(case["query"])
        top_3_ids = {r.doc_id for r in results[:3]}
        
        irrelevant_in_top_3 = [
            doc_id for doc_id in case["irrelevant_doc_ids"]
            if doc_id in top_3_ids
        ]
        
        if irrelevant_in_top_3:
            failures.append({
                "id": case["id"],
                "query": case["query"],
                "irrelevant_found": irrelevant_in_top_3,
            })
    
    assert not failures, f"Irrelevant docs in top-3:\n{failures}"

def test_retrieval_consistency(retriever):
    """Same query run twice should return same top documents."""
    query = "How do I run parallel tests?"
    
    results_1 = [r.doc_id for r in retriever.retrieve(query)]
    results_2 = [r.doc_id for r in retriever.retrieve(query)]
    
    assert results_1[:3] == results_2[:3], \
        f"Retrieval is non-deterministic: {results_1[:3]} vs {results_2[:3]}"

Measuring MRR and NDCG

For teams that need precision metrics:

def compute_mrr(retriever, test_cases: list[dict], k: int = 10) -> float:
    """Mean Reciprocal Rank — measures rank of first relevant document."""
    reciprocal_ranks = []
    
    for case in test_cases:
        results = retriever.retrieve(case["query"])
        
        for rank, result in enumerate(results[:k], start=1):
            if result.doc_id in case["relevant_doc_ids"]:
                reciprocal_ranks.append(1 / rank)
                break
        else:
            reciprocal_ranks.append(0)  # No relevant doc in top-k
    
    return sum(reciprocal_ranks) / len(reciprocal_ranks)

def test_mrr_above_baseline(retriever):
    mrr = compute_mrr(retriever, RETRIEVAL_TEST_CASES)
    assert mrr >= 0.7, f"MRR {mrr:.3f} below threshold 0.7"

Component 2: Testing the Generation (Context Faithfulness)

The LLM generation step must be faithful to retrieved context. This is where hallucination happens.

Using RAGAS for Faithfulness

RAGAS is a framework specifically for RAG evaluation:

pip install ragas
# tests/rag/test_generation_faithfulness.py
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision,
)

def test_rag_faithfulness():
    """Response must not contain claims outside the retrieved context."""
    
    # Build test dataset in RAGAS format
    test_data = {
        "question": [
            "What is the cost of the Pro plan?",
            "Does HelpMeTest support parallel test execution?",
            "How long is data retained on the free plan?",
        ],
        "answer": [
            # Actual RAG pipeline outputs
            rag_pipeline.query("What is the cost of the Pro plan?").answer,
            rag_pipeline.query("Does HelpMeTest support parallel test execution?").answer,
            rag_pipeline.query("How long is data retained on the free plan?").answer,
        ],
        "contexts": [
            # Retrieved context for each query
            rag_pipeline.query("What is the cost of the Pro plan?").retrieved_docs,
            rag_pipeline.query("Does HelpMeTest support parallel test execution?").retrieved_docs,
            rag_pipeline.query("How long is data retained on the free plan?").retrieved_docs,
        ],
        "ground_truth": [
            "The Pro plan costs $100 per month.",
            "Yes, parallel test execution is available on the Pro plan.",
            "The free plan does not include data retention.",
        ],
    }
    
    dataset = Dataset.from_dict(test_data)
    
    results = evaluate(
        dataset,
        metrics=[faithfulness, answer_relevancy, context_recall, context_precision],
    )
    
    print(f"\nRAGAS Scores:")
    print(f"  Faithfulness:         {results['faithfulness']:.3f}")
    print(f"  Answer Relevancy:     {results['answer_relevancy']:.3f}")
    print(f"  Context Recall:       {results['context_recall']:.3f}")
    print(f"  Context Precision:    {results['context_precision']:.3f}")
    
    # Thresholds
    assert results["faithfulness"] >= 0.8, \
        f"Faithfulness {results['faithfulness']:.3f} below threshold 0.8"
    assert results["answer_relevancy"] >= 0.75, \
        f"Answer relevancy {results['answer_relevancy']:.3f} below threshold 0.75"
    assert results["context_recall"] >= 0.7, \
        f"Context recall {results['context_recall']:.3f} below threshold 0.7"

Manual Faithfulness Testing

For simpler setups without RAGAS:

def test_response_does_not_hallucinate_pricing(rag_pipeline):
    """
    Critical test: pricing information must come from context, not LLM's training data.
    The LLM might "know" about a pricing tier that no longer exists.
    """
    query = "What are the pricing plans?"
    result = rag_pipeline.query(query)
    
    response = result.answer
    retrieved_docs = " ".join(result.retrieved_docs)
    
    # Extract price mentions from response
    import re
    prices_in_response = re.findall(r'\$[\d,]+', response)
    prices_in_context = re.findall(r'\$[\d,]+', retrieved_docs)
    
    for price in prices_in_response:
        assert price in prices_in_context, (
            f"Response mentions {price} which is not in retrieved context.\n"
            f"Response: {response}\n"
            f"Context prices: {prices_in_context}"
        )

End-to-End RAG Evaluation

With individual components tested, run end-to-end evaluation:

# tests/rag/test_rag_end_to_end.py
import pytest

GOLDEN_QA_PAIRS = [
    {
        "id": "qa-001",
        "question": "What does the Pro plan include?",
        "expected_answer_contains": ["$100", "unlimited tests", "parallel"],
        "must_not_contain": ["free", "enterprise", "SSO"],
    },
    {
        "id": "qa-002",
        "question": "Is there a free plan?",
        "expected_answer_contains": ["free", "10 tests"],
        "must_not_contain": ["credit card required", "payment"],
    },
    {
        "id": "qa-003",
        "question": "What's included in Enterprise?",
        "expected_answer_contains": ["SSO", "priority support"],
        "must_not_contain": ["$100"],  # Should not conflate with Pro pricing
    },
]

@pytest.mark.parametrize("case", GOLDEN_QA_PAIRS)
def test_rag_end_to_end(rag_pipeline, case):
    result = rag_pipeline.query(case["question"])
    answer = result.answer.lower()
    
    missing = [
        expected for expected in case["expected_answer_contains"]
        if expected.lower() not in answer
    ]
    
    present = [
        must_not for must_not in case.get("must_not_contain", [])
        if must_not.lower() in answer
    ]
    
    assert not missing, (
        f"Test {case['id']}: Missing expected content: {missing}\n"
        f"Answer: {result.answer}"
    )
    
    assert not present, (
        f"Test {case['id']}: Found forbidden content: {present}\n"
        f"Answer: {result.answer}"
    )

Testing Context Window Management

When retrieved documents exceed the context window, some are dropped. Test that this doesn't silently degrade quality:

def test_context_overflow_handling(rag_pipeline):
    """
    When more context is retrieved than fits in the context window,
    the most relevant documents should be prioritized.
    """
    # Query that returns many results
    result = rag_pipeline.query("Tell me everything about the product")
    
    # Verify the answer covers the most important topics
    # even if some context was truncated
    answer = result.answer.lower()
    
    critical_info = ["pricing", "free plan", "pro plan"]
    missing = [info for info in critical_info if info not in answer]
    
    assert not missing, \
        f"Context overflow dropped critical info: {missing}\nAnswer: {result.answer[:500]}"

Monitoring RAG in Production

Tests catch regressions before deployment. In production, monitor:

# src/rag/monitoring.py
import time
from dataclasses import dataclass

@dataclass
class RAGTrace:
    query: str
    retrieved_docs: list[str]
    retrieval_latency_ms: float
    generation_latency_ms: float
    answer: str
    num_docs_retrieved: int
    context_length: int

def instrument_rag_pipeline(pipeline):
    """Wrapper that logs traces for offline analysis."""
    original_query = pipeline.query
    
    def traced_query(question: str) -> RAGTrace:
        t0 = time.perf_counter()
        docs = pipeline.retriever.retrieve(question)
        retrieval_time = (time.perf_counter() - t0) * 1000
        
        t1 = time.perf_counter()
        answer = pipeline.generator.generate(question, docs)
        generation_time = (time.perf_counter() - t1) * 1000
        
        trace = RAGTrace(
            query=question,
            retrieved_docs=[d.content for d in docs],
            retrieval_latency_ms=retrieval_time,
            generation_latency_ms=generation_time,
            answer=answer,
            num_docs_retrieved=len(docs),
            context_length=sum(len(d.content) for d in docs),
        )
        
        # Send to your observability system
        log_rag_trace(trace)
        
        return trace
    
    pipeline.query = traced_query
    return pipeline

Key Takeaways

  • Test retrieval and generation separately — they fail independently
  • Use Recall@K to measure retrieval coverage; use MRR or NDCG for ranking quality
  • Use RAGAS faithfulness metric to detect hallucinations introduced by the LLM
  • Run end-to-end golden QA pairs to catch failures that only appear when retrieval and generation interact
  • Monitor context overflow — most RAG systems silently drop documents when the context window fills
  • Log RAG traces in production to build evaluation datasets from real queries
  • Update your golden dataset as your document corpus changes — stale tests give false confidence

Read more

Start now free