Testing

Testing RAG Pipelines: How to Validate Retrieval-Augmented Generation

HelpMeTest

HelpMeTest

— 6 min read
Testing RAG Pipelines: How to Validate Retrieval-Augmented Generation

Retrieval-Augmented Generation (RAG) systems are notoriously hard to test. Unlike deterministic software where the same input always produces the same output, RAG pipelines combine a retrieval step (vector search, keyword search, or hybrid) with an LLM generation step — both of which introduce variability. A RAG system can appear to work perfectly on your test queries and then fail silently on real user questions.

This guide covers how to systematically test each component of a RAG pipeline, what metrics matter, and how to build a test suite that catches regressions before they reach production.

The RAG Pipeline Components to Test

A typical RAG pipeline has four testable stages:

  1. Indexing: Documents are chunked, embedded, and stored in a vector store
  2. Retrieval: User query is embedded, and the most relevant chunks are fetched
  3. Context assembly: Retrieved chunks are ranked, deduped, and formatted into a prompt
  4. Generation: LLM receives the assembled context and generates a response

Each stage can fail independently. Testing the final answer quality alone tells you something went wrong but not where.

Stage 1: Testing Document Chunking

Chunking strategy determines retrieval quality. Chunks that are too small lose context; chunks that are too large dilute relevance scores. Test that your chunking produces semantically coherent pieces:

import pytest
from your_rag import chunk_document

def test_chunk_preserves_sentence_boundaries():
    text = "Machine learning is powerful. Neural networks are a subset of ML. They learn from data."
    chunks = chunk_document(text, chunk_size=50, overlap=10)
    
    for chunk in chunks:
        # No chunk should start mid-sentence
        assert not chunk.startswith(('.', ',', ';')), f"Chunk starts with punctuation: {chunk}"

def test_chunk_overlap_is_correct():
    text = "A" * 100
    chunks = chunk_document(text, chunk_size=40, overlap=10)
    
    for i in range(len(chunks) - 1):
        assert chunks[i][-10:] == chunks[i+1][:10], f"Overlap mismatch at chunk {i}"

def test_no_empty_chunks():
    text = "Short text.\n\n\n\nAnother section."
    chunks = chunk_document(text, chunk_size=100)
    assert all(len(chunk.strip()) > 0 for chunk in chunks)

def test_metadata_preserved():
    doc = {"text": "Important content.", "source": "policy-doc.pdf", "page": 3}
    chunks = chunk_document(doc["text"], metadata=doc)
    
    for chunk in chunks:
        assert chunk.metadata["source"] == "policy-doc.pdf"
        assert chunk.metadata["page"] == 3

Stage 2: Testing Retrieval Quality

Retrieval quality is measured by whether the right documents are returned for a given query. Build a test set of (query, expected_document) pairs:

import pytest
from your_rag import VectorStore

@pytest.fixture
def populated_store():
    store = VectorStore()
    store.add_documents([
        {"id": "doc-1", "text": "Python is a high-level programming language.", "metadata": {"topic": "python"}},
        {"id": "doc-2", "text": "JavaScript runs in the browser and on Node.js.", "metadata": {"topic": "javascript"}},
        {"id": "doc-3", "text": "Docker containers package applications with their dependencies.", "metadata": {"topic": "devops"}},
    ])
    return store

RETRIEVAL_TEST_CASES = [
    ("What is Python?", "doc-1", "python programming language"),
    ("How does JavaScript work?", "doc-2", "javascript browser"),
    ("What is containerization?", "doc-3", "docker containers"),
]

@pytest.mark.parametrize("query, expected_doc_id, description", RETRIEVAL_TEST_CASES)
def test_retrieval_returns_correct_document(populated_store, query, expected_doc_id, description):
    results = populated_store.search(query, top_k=3)
    retrieved_ids = [r.id for r in results]
    
    assert expected_doc_id in retrieved_ids, (
        f"Expected '{expected_doc_id}' for query '{query}' but got: {retrieved_ids}"
    )

def test_retrieval_rank_quality(populated_store):
    results = populated_store.search("Python programming", top_k=3)
    assert results[0].id == "doc-1", f"Expected doc-1 first, got {results[0].id}"

def test_retrieval_scores_are_ranked(populated_store):
    results = populated_store.search("containerization", top_k=3)
    scores = [r.score for r in results]
    assert scores == sorted(scores, reverse=True), "Results should be sorted by score descending"

Recall@K Metric

Measure what percentage of expected documents appear in the top-K results:

def recall_at_k(store, test_cases, k=5):
    hits = 0
    for query, expected_id, _ in test_cases:
        results = store.search(query, top_k=k)
        if any(r.id == expected_id for r in results):
            hits += 1
    return hits / len(test_cases)

def test_retrieval_recall_meets_threshold(populated_store):
    recall = recall_at_k(populated_store, RETRIEVAL_TEST_CASES, k=3)
    assert recall >= 0.9, f"Retrieval recall@3 is {recall:.2%}, expected >= 90%"

Stage 3: Testing Context Relevance

Even when the right documents are retrieved, the assembled context needs to be relevant to the query. Use an LLM as a judge to score relevance:

import openai

def score_context_relevance(query: str, context: str) -> float:
    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""Rate how relevant this context is for answering the query.
            
Query: {query}
Context: {context}

Rate from 0 (completely irrelevant) to 1 (perfectly relevant). Respond with only a decimal number."""
        }]
    )
    return float(response.choices[0].message.content.strip())

def test_context_relevance(rag_pipeline):
    test_cases = [
        ("How do I reset my password?", 0.8),
        ("What's the weather today?", 0.1),
    ]
    
    for query, min_score in test_cases:
        context = rag_pipeline.retrieve_context(query)
        score = score_context_relevance(query, context)
        assert score >= min_score, f"Context relevance {score:.2f} below threshold {min_score} for: {query}"

Stage 4: Testing Answer Faithfulness

Faithfulness measures whether the generated answer is grounded in the retrieved context — or whether the LLM is hallucinating:

def check_faithfulness(answer: str, context: str) -> dict:
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""Identify any claims in the Answer that are NOT supported by the Context.

Context: {context}

Answer: {answer}

List unsupported claims as a JSON array. If all claims are supported, return [].
Respond with only the JSON array."""
        }]
    )
    unsupported = json.loads(response.choices[0].message.content.strip())
    return {
        "faithful": len(unsupported) == 0,
        "unsupported_claims": unsupported,
    }

@pytest.mark.parametrize("query", [
    "What is our refund policy?",
    "How do I contact support?",
    "What payment methods do you accept?",
])
def test_answers_are_faithful(rag_pipeline, query):
    context = rag_pipeline.retrieve_context(query)
    answer = rag_pipeline.generate_answer(query, context)
    
    result = check_faithfulness(answer, context)
    assert result["faithful"], (
        f"Unfaithful answer for '{query}'.\n"
        f"Unsupported claims: {result['unsupported_claims']}\n"
        f"Answer: {answer}"
    )

End-to-End RAG Evaluation with RAGAS

RAGAS is an open-source library specifically designed for RAG evaluation. It measures faithfulness, answer relevancy, context precision, and context recall:

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

def test_rag_quality_metrics(rag_pipeline, test_dataset):
    questions = [d['question'] for d in test_dataset]
    ground_truths = [d['ground_truth'] for d in test_dataset]
    
    answers = []
    contexts = []
    for question in questions:
        ctx = rag_pipeline.retrieve_context(question)
        ans = rag_pipeline.generate_answer(question, ctx)
        answers.append(ans)
        contexts.append([ctx])
    
    dataset = Dataset.from_dict({
        'question': questions,
        'answer': answers,
        'contexts': contexts,
        'ground_truth': ground_truths,
    })
    
    result = evaluate(dataset, metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ])
    
    assert result['faithfulness'] >= 0.85, f"Faithfulness {result['faithfulness']:.2f} below threshold"
    assert result['answer_relevancy'] >= 0.80, f"Answer relevancy {result['answer_relevancy']:.2f} below threshold"
    assert result['context_recall'] >= 0.75, f"Context recall {result['context_recall']:.2f} below threshold"

Testing Latency

RAG pipelines have multiple I/O operations. Measure and enforce latency budgets at each stage:

import time

def test_retrieval_latency(populated_store):
    query = "How do I configure authentication?"
    
    start = time.perf_counter()
    results = populated_store.search(query, top_k=5)
    elapsed_ms = (time.perf_counter() - start) * 1000
    
    assert elapsed_ms < 200, f"Retrieval took {elapsed_ms:.0f}ms, expected <200ms"

def test_end_to_end_latency(rag_pipeline):
    query = "What is the pricing for the Pro plan?"
    
    start = time.perf_counter()
    answer = rag_pipeline.answer(query)
    elapsed_ms = (time.perf_counter() - start) * 1000
    
    assert elapsed_ms < 5000, f"End-to-end took {elapsed_ms:.0f}ms, expected <5s"
    assert len(answer) > 10, "Answer should not be empty"

Regression Testing After Index Updates

When you add new documents to the index or re-embed with a new model, run a regression suite to ensure existing queries still return correct results:

REGRESSION_CASES = [
    {
        "query": "What is the refund window?",
        "expected_keywords": ["30 days", "refund"],
    },
    {
        "query": "How do I upgrade my plan?",
        "expected_keywords": ["settings", "billing", "upgrade"],
    },
]

def test_rag_regression(rag_pipeline):
    failures = []
    
    for case in REGRESSION_CASES:
        answer = rag_pipeline.answer(case["query"])
        answer_lower = answer.lower()
        
        missing_keywords = [
            kw for kw in case["expected_keywords"]
            if kw.lower() not in answer_lower
        ]
        
        if missing_keywords:
            failures.append({
                "query": case["query"],
                "missing": missing_keywords,
                "answer": answer,
            })
    
    assert not failures, f"Regression failures:\n{json.dumps(failures, indent=2)}"

Building a Golden Dataset

The foundation of reliable RAG testing is a curated golden dataset — a set of questions with known correct answers that you maintain and grow over time:

  1. Start with real user questions from production logs, support tickets, or user interviews
  2. Add adversarial cases — questions the system should decline to answer because the information isn't in the corpus
  3. Include edge cases — ambiguous questions, multi-hop questions requiring information from several documents, questions with numerical answers
  4. Version the dataset — when you change the pipeline, run the full golden dataset and record metrics so you can track improvement or regression over time

A good golden dataset for a documentation RAG might have 200-500 questions covering different topics, difficulty levels, and expected behaviors. It becomes your system's test spec.

RAG testing is harder than testing deterministic software, but it's not optional. A RAG system you can't evaluate is a system you can't safely improve — every change is a gamble. Build the evaluation infrastructure early and it pays dividends every time you update the pipeline.

Read more