How to Test a RAG Pipeline: Two Surfaces, Not One Score
Tom PiaggioCo-Founder at Autonoma
How to test a RAG pipeline starts with treating retrieval and generation as two separate test surfaces instead of scoring the whole answer at once: assert that the retriever pulled the right chunks, then separately assert that the generator's answer is faithful to those chunks and actually answers the question. Combine deterministic retrieval checks with threshold-gated generation metrics, faithfulness, answer relevancy, and context precision, computed with Ragas or DeepEval against an eval set built from your own documents, and gate the aggregate score in CI on every pull request.
Someone drops a screenshot in Slack. A user asked the "ask our docs" bot about the refund window, and the bot answered with total confidence. The answer is wrong. Not garbled, not a timeout, not a stack trace. Wrong, stated with the same even tone as everything else in the response.
Now you're staring at the two questions everyone stares at the first time this happens. Did the retriever pull the wrong chunk, so the generator was reasoning from garbage? Or did the retriever do its job, and the generator ignored the context, or quietly filled in a gap with something it half-remembered from pretraining? You cannot tell from the transcript alone. The retrieved chunks and the generated answer both read as plausible in isolation, and the only way to actually know which stage broke is to have already been asserting on both of them separately, before this ticket ever landed in your inbox.
The Two Failure Surfaces an End-to-End Score Can't Localize
A RAG pipeline has exactly one job from the outside: take a question, return a grounded answer. That framing is exactly why an end-to-end pass or fail is the wrong metric to build a test suite around. It tells you the pipeline produced a bad answer. It has nothing to say about which of the two stages inside that pipeline produced it, and those two stages fail for completely different reasons, with completely different fixes.
A retrieval failure means the retriever handed the generator the wrong material. Right question, wrong chunks, and the generator is now confidently reasoning from context that never had the answer in it. The fix lives in chunking strategy, embedding model choice, top-k, or reranking, nowhere near the prompt. A generation failure means the retriever did its job. The right chunks were sitting right there in context, and the generator still produced a claim that isn't in them, either by embellishing past what the context supports or by quietly discounting it in favor of something the model already "knew." The fix for that lives in the prompt, the faithfulness threshold, or the refusal logic, and touching chunking or embeddings does nothing for it.
Score the whole answer as one unit and both failures look identical from the outside: a wrong answer. Split the assertion into two gates, one on what the retriever returned and one on what the generator did with it, and the failure tells you exactly which team meeting you need to have.
Retrieval and generation fail for different reasons and need different fixes. Testing them as one pass or fail hides which one you actually have.
Signal
Retrieval Failure
Generation Failure
Symptom
Right question, wrong chunks
Right chunks, unsupported claim
Root cause
Chunking, embeddings, low top-k
Model ignores or embellishes context
Fix lever
Chunking, embeddings, reranking
Prompt, threshold, refusal logic
Test surface
Context the retriever returned
Answer text given fixed context
Determinism
Deterministic given a fixed index
Non-deterministic, needs N-run gating
If you're mapping the wider surface a shipped genAI feature needs covered, testing generative AI applications is the broader pillar this retrieval and generation split sits inside.
Testing the Retrieval Layer
Retrieval is the deterministic half of a RAG pipeline, and that's exactly why it deserves its own test surface. Given a fixed index and a fixed query, the retriever returns the same chunks every time. There's no sampling, no temperature, no model deciding how to phrase anything. If a retrieval test flakes, it isn't the retriever being probabilistic, it's your index or your query changing underneath the test.
The core assertion is simple to state and easy to get wrong in practice: for a question with a known-correct chunk, is that chunk in the top-k results, and at what rank. Rank matters because "somewhere in the top 50" and "the first result" are very different production experiences once you factor in how many chunks actually make it into the generator's context window. Context precision asks what fraction of what you retrieved was actually relevant. Context recall asks what fraction of everything relevant you managed to retrieve, in app-builder terms: precision punishes a retriever that buries the right answer under noise, and recall punishes a retriever that leaves the right answer out of the top-k entirely.
The three fix levers, in the order most teams reach for them: chunk size and overlap first (chunks too large dilute the embedding, too small lose context), embedding model choice second (a domain mismatch between your embedding model and your actual content shows up here first), and top-k or reranking third (retrieving more candidates and reranking them is usually cheaper than re-chunking your entire index). If you're seeing a specific "confidently wrong answer" pattern and want to isolate whether it's a retrieval problem before touching generation at all, how to test if your RAG pipeline is retrieving the right context goes deeper into hit-rate, MRR, and vector search accuracy specifically.
Testing the Generation Layer Given Retrieved Context
Once retrieval is asserted separately, generation testing gets to ask a narrower, more answerable question: given exactly this context, is this answer any good? Two metrics carry almost all of the weight here. Faithfulness asks whether every claim in the answer is actually supported by the retrieved context, the generation-side counterpart to the hallucination problem: an answer can be fluent, confident, and completely unsupported by anything the model was given. Answer relevancy asks the opposite failure mode: an answer that's fully grounded in the context but doesn't actually address what was asked, a faithful non-answer.
Both metrics need the same underlying mechanism to be trustworthy: don't score the whole answer as one grounded-or-not unit. Decompose it into atomic claims and check each one against the retrieved chunks individually, then aggregate. A single ungrounded sentence buried inside an otherwise solid answer is exactly the failure a whole-answer score is built to miss. How to test for AI hallucinations covers that claim-decomposition mechanism in more depth if you want the generation-only version of this problem outside of a RAG context specifically.
There's a third case generation testing has to cover that faithfulness and relevancy alone don't: what happens when the context genuinely doesn't contain the answer. A well-behaved pipeline should refuse or hedge rather than generate a confident answer from insufficient context, and that refusal behavior is itself a testable assertion, not a hope. Build refusal test cases directly into your eval set, questions your corpus genuinely cannot answer, and assert the generator says so instead of inventing something plausible-sounding to fill the gap.
Building Your Eval Set From Your Own Documents
None of the metrics above mean anything without an eval set, and this is the part almost every RAG testing guide skips straight past, as if a labeled dataset just appears once you've read enough about faithfulness. It doesn't. You have to build it, and the good news is it's a mechanical process once you know the shape of it.
Start by sampling real chunks out of your own index, not synthetic documents, not a public benchmark corpus. For each sampled chunk, generate a candidate question that chunk should be able to answer, plus the ground-truth answer, using an LLM to draft the pair and a human to review it before it goes in the set. That review step is not optional: an LLM-drafted question can accidentally be answerable from general knowledge rather than from the chunk specifically, which quietly breaks the point of the test. Layer in real user queries mined from your logs, deduplicated and sampled, because the questions your actual users ask are reliably weirder and more specific than anything an LLM generates from a chunk in isolation.
The subset everyone skips is the adversarial one: questions your corpus genuinely cannot answer. Deliberately write a handful of questions that sound reasonable but fall outside what your documents cover, and mark them so the expected behavior is a refusal, not an answer. Without that subset, you have no way to test whether your pipeline knows the difference between "I don't have this" and "I'll guess."
On size: twenty to fifty curated examples, each one reviewed and something you'd defend individually, will catch more real regressions than five hundred synthetic ones nobody has looked at. A small set you trust means every red result gets investigated. A large set nobody has reviewed means red results get shrugged off as noise, which defeats the entire point of having a test suite. Here's the script that handles the sampling, the LLM-assisted question generation, the log mining, and the adversarial subset:
"""Build a candidate RAG eval set from your own documents and query logs.This script produces a *draft*. Every generated question/answer pair needsa human review pass before it's trusted in eval_set/eval_set.json, which iswhy this writes to a separate `.generated.json` file instead of overwritingthe reviewed set directly.Expected input formats: --corpus a JSONL file, one chunk per line: {"id": ..., "source": ..., "text": ...} --logs a JSONL file, one query per line: {"query": "..."}"""import argparseimport jsonimport randomfrom collections import Counterfrom pathlib import Pathfrom openai import OpenAIADVERSARIAL_SEED_TOPICS = [ "a loyalty rewards or points program", "carbon offsets or sustainability commitments for shipping", "a price-match guarantee against competitors", "student or military discounts",]def load_jsonl(path: Path) -> list[dict]: with open(path) as f: return [json.loads(line) for line in f if line.strip()]def sample_chunks(corpus: list[dict], n: int, seed: int = 42) -> list[dict]: rng = random.Random(seed) return rng.sample(corpus, k=min(n, len(corpus)))def generate_qa_pair(client: OpenAI, chunk: dict) -> dict: prompt = ( "You are building a test set for a retrieval-augmented generation system. " "Given the passage below, write one question that can be answered fully and " "only from this passage, plus the correct answer using only facts stated in " "the passage. Respond as JSON: {\"question\": ..., \"answer\": ...}\n\n" f"Passage:\n{chunk['text']}" ) response = client.chat.completions.create( model="gpt-4o-mini", temperature=0, response_format={"type": "json_object"}, messages=[{"role": "user", "content": prompt}], ) payload = json.loads(response.choices[0].message.content) return { "id": f"gen-{chunk['id']}", "question": payload["question"], "ground_truth_answer": payload["answer"], "expected_chunk_ids": [chunk["id"]], "category": "standard", "needs_human_review": True, }def mine_queries_from_logs(logs: list[dict], n: int, seed: int = 42) -> list[str]: queries = [row["query"].strip() for row in logs if row.get("query", "").strip()] counts = Counter(queries) deduped = list(counts.keys()) rng = random.Random(seed) rng.shuffle(deduped) return deduped[:n]def build_adversarial_set(client: OpenAI, topics: list[str]) -> list[dict]: examples = [] for i, topic in enumerate(topics): prompt = ( "Write one realistic customer support question about " f"{topic}. It should sound like a question a real customer would ask, " "in one sentence, no preamble." ) response = client.chat.completions.create( model="gpt-4o-mini", temperature=0, messages=[{"role": "user", "content": prompt}], ) question = response.choices[0].message.content.strip() examples.append({ "id": f"adv-{i:03d}", "question": question, "ground_truth_answer": "insufficient_context", "expected_chunk_ids": [], "category": "adversarial", "needs_human_review": True, }) return examplesdef main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--corpus", required=True, type=Path) parser.add_argument("--logs", type=Path, default=None) parser.add_argument("--out", required=True, type=Path) parser.add_argument("--sample-size", type=int, default=30) parser.add_argument("--query-sample-size", type=int, default=20) args = parser.parse_args() client = OpenAI() corpus = load_jsonl(args.corpus) sampled = sample_chunks(corpus, n=args.sample_size) generated_examples = [generate_qa_pair(client, chunk) for chunk in sampled] mined_queries = [] if args.logs and args.logs.exists(): logs = load_jsonl(args.logs) mined_queries = mine_queries_from_logs(logs, n=args.query_sample_size) adversarial_examples = build_adversarial_set(client, ADVERSARIAL_SEED_TOPICS) output = { "generated_examples": generated_examples, "mined_queries_for_manual_labeling": mined_queries, "adversarial_examples": adversarial_examples, } args.out.parent.mkdir(parents=True, exist_ok=True) with open(args.out, "w") as f: json.dump(output, f, indent=2) print(f"Wrote {len(generated_examples)} candidate examples, " f"{len(mined_queries)} mined queries needing ground truth, and " f"{len(adversarial_examples)} adversarial examples to {args.out}") print("Review every entry before merging into eval_set/eval_set.json. " "Mined queries still need a human-written ground_truth_answer.")if __name__ == "__main__": main()
And here's the shape the resulting eval set takes once it's reviewed and merged, a corpus of chunks plus a list of examples, each one tagged standard or adversarial:
{ "corpus": [ { "id": "chunk-001", "source": "refund-policy.md", "text": "Refunds are issued within 30 days of purchase for unopened items. Opened items are eligible for store credit only, issued within 14 days of purchase." }, { "id": "chunk-002", "source": "refund-policy.md", "text": "Digital products, gift cards, and personalized items are non-refundable under any circumstance." }, { "id": "chunk-003", "source": "shipping-policy.md", "text": "Standard shipping takes 5 to 7 business days within the continental United States. Expedited shipping takes 2 business days for an additional fee." }, { "id": "chunk-004", "source": "shipping-policy.md", "text": "We do not currently ship outside the United States and Canada. Orders placed with an unsupported address are automatically canceled and refunded." }, { "id": "chunk-005", "source": "account-policy.md", "text": "Accounts are automatically flagged for review after five failed login attempts within one hour. A flagged account can be unlocked by verifying the email on file." }, { "id": "chunk-006", "source": "account-policy.md", "text": "Deleting an account is permanent after a 30 day grace period, during which a user can restore it by logging back in." } ], "examples": [ { "id": "ex-001", "question": "How long do I have to return an unopened item?", "ground_truth_answer": "You have 30 days from the purchase date to return an unopened item for a full refund.", "expected_chunk_ids": ["chunk-001"], "category": "standard" }, { "id": "ex-002", "question": "Can I get a refund on a gift card I bought by mistake?", "ground_truth_answer": "No, gift cards are non-refundable under any circumstance.", "expected_chunk_ids": ["chunk-002"], "category": "standard" }, { "id": "ex-003", "question": "How long does standard shipping take within the US?", "ground_truth_answer": "Standard shipping takes 5 to 7 business days within the continental United States.", "expected_chunk_ids": ["chunk-003"], "category": "standard" }, { "id": "ex-004", "question": "What happens to my order if I ship to a country you don't support?", "ground_truth_answer": "The order is automatically canceled and refunded because you do not currently ship outside the United States and Canada.", "expected_chunk_ids": ["chunk-004"], "category": "standard" }, { "id": "ex-005", "question": "How many failed logins before my account gets flagged?", "ground_truth_answer": "Your account is flagged for review after five failed login attempts within one hour.", "expected_chunk_ids": ["chunk-005"], "category": "standard" }, { "id": "ex-006", "question": "What is your company's policy on carbon offsets for shipping?", "ground_truth_answer": "insufficient_context", "expected_chunk_ids": [], "category": "adversarial" }, { "id": "ex-007", "question": "Do you offer a loyalty rewards program with points on every purchase?", "ground_truth_answer": "insufficient_context", "expected_chunk_ids": [], "category": "adversarial" } ]}
Shipping the Runnable Harness
With an eval set in hand, the harness is just pytest wired up to run both gates against it. Start with a shared fixture layer so every test file works off the same corpus, the same eval examples, and the same retriever instance instead of each test file re-loading its own copy:
import jsonimport mathimport osimport refrom collections import Counterfrom pathlib import Pathimport pytestEVAL_SET_PATH = Path(__file__).resolve().parent.parent / "eval_set" / "eval_set.json"# Words too common to carry meaning in a lookup. This list has to be explicit.# At this corpus size inverse document frequency cannot identify them on its# own: a stopword that happens to appear in exactly one chunk looks just as# rare, and therefore just as important, as a real keyword. See the note in# SimpleRetriever for why that distinction decides the ranking.STOPWORDS = frozenset( """ a an and any are as at be been by can do does for from get got have how i if in is it its me my no not of on one only or our out so than that the their them then there these they this to under up us was we what when where which who will with would you your """.split())def _tokenize(text: str) -> set[str]: """Lowercase, split into word characters, and drop stopwords.""" words = re.findall(r"[a-z0-9]+", text.lower()) return {word for word in words if word not in STOPWORDS}class SimpleRetriever: """Deterministic IDF-weighted keyword retriever over a fixed corpus. Swap this for your real vector-search client (pgvector, Pinecone, Chroma, etc). The important property this class demonstrates is the one the article leans on: given a fixed corpus and a fixed query, retrieve() always returns the same ranked list. There is no sampling here at all. Scoring stays purely lexical, so the retrieval suite runs offline with no API key. Each token is weighted by inverse document frequency, computed once over the corpus at construction time: idf(token) = log((N + 1) / (df(token) + 1)) + 1 A chunk's score is the share of the query's total token weight that the chunk matches, which keeps scores in [0, 1] and comparable across queries of different lengths. Two details do the real work: 1. Stopwords come out before scoring. Weighting alone is not enough here, because "to" appearing in one chunk earns exactly the same idf as "unopened" appearing in one chunk. Filtering first is what stops a chunk that shares two throwaway words from outranking the one chunk that actually answers the question. 2. Tokens absent from the corpus get the maximum idf. They can never be matched, so they only inflate the denominator, which pushes questions the corpus cannot answer toward a low score instead of letting them latch onto whichever chunk happens to share a word. """ def __init__(self, corpus: list[dict]): self.corpus = corpus self._tokens = {c["id"]: _tokenize(c["text"]) for c in corpus} document_frequency = Counter() for tokens in self._tokens.values(): document_frequency.update(tokens) n_docs = len(corpus) self._idf = { token: math.log((n_docs + 1) / (df + 1)) + 1 for token, df in document_frequency.items() } self._unseen_idf = math.log(n_docs + 1) + 1 def _weight(self, token: str) -> float: return self._idf.get(token, self._unseen_idf) def retrieve(self, query: str, top_k: int = 3) -> list[dict]: query_tokens = _tokenize(query) query_weight = sum(self._weight(token) for token in query_tokens) scored = [] for chunk in self.corpus: matched = query_tokens & self._tokens[chunk["id"]] score = sum(self._weight(token) for token in matched) / max(query_weight, 1e-9) scored.append((score, chunk)) # Descending score, then ascending chunk id. The explicit second key # means ties resolve the same way on every run, instead of depending # on the order the corpus happened to arrive in. scored.sort(key=lambda pair: (-pair[0], pair[1]["id"])) results = [] for score, chunk in scored[:top_k]: results.append({**chunk, "score": round(score, 4)}) return resultsclass RagPipeline: """Thin wrapper combining a retriever with a generation call. generate() calls the OpenAI API. Replace the body of generate() with whichever model client your pipeline actually uses; keep the signature (query, context_chunks) -> str so the tests don't need to change. """ def __init__(self, retriever: SimpleRetriever): self.retriever = retriever def generate(self, query: str, context_chunks: list[dict]) -> str: from openai import OpenAI client = OpenAI() context_text = "\n\n".join(c["text"] for c in context_chunks) prompt = ( "Answer the question using only the context below. " "If the context does not contain the answer, respond exactly with " "'insufficient_context'.\n\n" f"Context:\n{context_text}\n\nQuestion: {query}\n\nAnswer:" ) response = client.chat.completions.create( model="gpt-4o-mini", temperature=0, messages=[{"role": "user", "content": prompt}], ) return response.choices[0].message.content.strip()@pytest.fixture(scope="session")def eval_set() -> dict: with open(EVAL_SET_PATH) as f: return json.load(f)@pytest.fixture(scope="session")def corpus(eval_set: dict) -> list[dict]: return eval_set["corpus"]@pytest.fixture(scope="session")def eval_examples(eval_set: dict) -> list[dict]: return eval_set["examples"]@pytest.fixture(scope="session")def retriever(corpus: list[dict]) -> SimpleRetriever: return SimpleRetriever(corpus)@pytest.fixture(scope="session")def rag_pipeline(retriever: SimpleRetriever) -> RagPipeline: return RagPipeline(retriever)@pytest.fixture(scope="session")def requires_openai_key(): if not os.environ.get("OPENAI_API_KEY"): pytest.skip("OPENAI_API_KEY not set; skipping generation tests that need a live model call")
The retrieval tests run entirely offline against your fixed index, no LLM call, no API key, no non-determinism to manage. They compute hit-rate@k and mean reciprocal rank across the whole eval set and check the adversarial examples come back with low retrieval confidence rather than a false-positive match:
"""Retrieval-layer tests. Deterministic: no API key required.Every assertion here gates on an aggregate across the eval set, not on anysingle example, which is safe to do even though retrieval is deterministic:it keeps the threshold meaningful as you add more examples over time."""HIT_RATE_THRESHOLD = 0.8MRR_THRESHOLD = 0.7ADVERSARIAL_MAX_SCORE = 0.2TOP_K = 3def test_hit_rate_at_k(retriever, eval_examples): standard = [ex for ex in eval_examples if ex["category"] == "standard"] hits = 0 for example in standard: results = retriever.retrieve(example["question"], top_k=TOP_K) retrieved_ids = {r["id"] for r in results} if retrieved_ids & set(example["expected_chunk_ids"]): hits += 1 hit_rate = hits / len(standard) assert hit_rate >= HIT_RATE_THRESHOLD, ( f"hit-rate@{TOP_K} was {hit_rate:.2f} across {len(standard)} examples, " f"below the {HIT_RATE_THRESHOLD} threshold" )def test_mean_reciprocal_rank(retriever, eval_examples): standard = [ex for ex in eval_examples if ex["category"] == "standard"] reciprocal_ranks = [] for example in standard: results = retriever.retrieve(example["question"], top_k=TOP_K) rank = None for position, chunk in enumerate(results, start=1): if chunk["id"] in example["expected_chunk_ids"]: rank = position break reciprocal_ranks.append(1 / rank if rank else 0.0) mrr = sum(reciprocal_ranks) / len(reciprocal_ranks) assert mrr >= MRR_THRESHOLD, f"MRR was {mrr:.2f}, below the {MRR_THRESHOLD} threshold"def test_adversarial_examples_get_low_retrieval_confidence(retriever, eval_examples): adversarial = [ex for ex in eval_examples if ex["category"] == "adversarial"] assert adversarial, "eval set has no adversarial examples; add some before trusting this test" false_positive_count = 0 for example in adversarial: results = retriever.retrieve(example["question"], top_k=1) top_score = results[0]["score"] if results else 0.0 if top_score > ADVERSARIAL_MAX_SCORE: false_positive_count += 1 false_positive_rate = false_positive_count / len(adversarial) assert false_positive_rate <= 0.2, ( f"{false_positive_count}/{len(adversarial)} adversarial questions matched a chunk " f"with score above {ADVERSARIAL_MAX_SCORE}; the corpus may be too broad or the " "adversarial set may not be adversarial enough" )
The generation tests need an actual model call, and this is where the two dominant libraries earn their keep. Ragas scores a batch of question, answer, and context triples at once, which is the shape you want once you're evaluating a full eval set instead of asserting on one example:
"""Generation-layer tests using Ragas. Needs OPENAI_API_KEY.Each metric is scored per example by Ragas, then this file asserts on themean across the whole set, never on a single row, so one noisy generationdoesn't fail the build by itself."""import pytestragas = pytest.importorskip("ragas")datasets = pytest.importorskip("datasets")from ragas import evaluatefrom ragas.metrics import answer_relevancy, context_precision, faithfulnessfrom datasets import DatasetFAITHFULNESS_THRESHOLD = 0.8ANSWER_RELEVANCY_THRESHOLD = 0.75CONTEXT_PRECISION_THRESHOLD = 0.7TOP_K = 3@pytest.fixture(scope="module")def ragas_dataset(requires_openai_key, rag_pipeline, eval_examples): standard = [ex for ex in eval_examples if ex["category"] == "standard"] rows = {"question": [], "answer": [], "contexts": [], "ground_truth": []} for example in standard: retrieved = rag_pipeline.retriever.retrieve(example["question"], top_k=TOP_K) answer = rag_pipeline.generate(example["question"], retrieved) rows["question"].append(example["question"]) rows["answer"].append(answer) rows["contexts"].append([chunk["text"] for chunk in retrieved]) rows["ground_truth"].append(example["ground_truth_answer"]) return Dataset.from_dict(rows)@pytest.fixture(scope="module")def ragas_scores(ragas_dataset): result = evaluate( ragas_dataset, metrics=[faithfulness, answer_relevancy, context_precision], ) return result.to_pandas()def test_aggregate_faithfulness(ragas_scores): mean_score = ragas_scores["faithfulness"].mean() assert mean_score >= FAITHFULNESS_THRESHOLD, ( f"mean faithfulness was {mean_score:.2f}, below {FAITHFULNESS_THRESHOLD}" )def test_aggregate_answer_relevancy(ragas_scores): mean_score = ragas_scores["answer_relevancy"].mean() assert mean_score >= ANSWER_RELEVANCY_THRESHOLD, ( f"mean answer relevancy was {mean_score:.2f}, below {ANSWER_RELEVANCY_THRESHOLD}" )def test_aggregate_context_precision(ragas_scores): mean_score = ragas_scores["context_precision"].mean() assert mean_score >= CONTEXT_PRECISION_THRESHOLD, ( f"mean context precision was {mean_score:.2f}, below {CONTEXT_PRECISION_THRESHOLD}" )
DeepEval scores case by case with the same metric names, faithfulness, answer relevancy, contextual precision, and plugs directly into pytest's assertion model if you prefer per-test-case granularity while still aggregating before you gate:
"""Generation-layer tests using DeepEval. Needs OPENAI_API_KEY.DeepEval's assert_test() is convenient for a single case, but this filedeliberately measures each metric manually and aggregates, because gatingon any one example's pass/fail is exactly the single-sample noise thearticle warns about."""import pytestdeepeval = pytest.importorskip("deepeval")from deepeval.metrics import AnswerRelevancyMetric, FaithfulnessMetricfrom deepeval.test_case import LLMTestCaseFAITHFULNESS_THRESHOLD = 0.8ANSWER_RELEVANCY_THRESHOLD = 0.75REFUSAL_RATE_THRESHOLD = 0.8TOP_K = 3REFUSAL_MARKER = "insufficient_context"def _build_test_case(rag_pipeline, example): retrieved = rag_pipeline.retriever.retrieve(example["question"], top_k=TOP_K) answer = rag_pipeline.generate(example["question"], retrieved) return LLMTestCase( input=example["question"], actual_output=answer, retrieval_context=[chunk["text"] for chunk in retrieved], expected_output=example["ground_truth_answer"], ), answerdef test_aggregate_faithfulness(requires_openai_key, rag_pipeline, eval_examples): standard = [ex for ex in eval_examples if ex["category"] == "standard"] metric = FaithfulnessMetric(threshold=FAITHFULNESS_THRESHOLD) scores = [] for example in standard: test_case, _ = _build_test_case(rag_pipeline, example) metric.measure(test_case) scores.append(metric.score) mean_score = sum(scores) / len(scores) assert mean_score >= FAITHFULNESS_THRESHOLD, ( f"mean faithfulness was {mean_score:.2f} across {len(scores)} examples, " f"below {FAITHFULNESS_THRESHOLD}" )def test_aggregate_answer_relevancy(requires_openai_key, rag_pipeline, eval_examples): standard = [ex for ex in eval_examples if ex["category"] == "standard"] metric = AnswerRelevancyMetric(threshold=ANSWER_RELEVANCY_THRESHOLD) scores = [] for example in standard: test_case, _ = _build_test_case(rag_pipeline, example) metric.measure(test_case) scores.append(metric.score) mean_score = sum(scores) / len(scores) assert mean_score >= ANSWER_RELEVANCY_THRESHOLD, ( f"mean answer relevancy was {mean_score:.2f} across {len(scores)} examples, " f"below {ANSWER_RELEVANCY_THRESHOLD}" )def test_adversarial_examples_trigger_refusal(requires_openai_key, rag_pipeline, eval_examples): adversarial = [ex for ex in eval_examples if ex["category"] == "adversarial"] assert adversarial, "eval set has no adversarial examples; add some before trusting this test" refusals = 0 for example in adversarial: _, answer = _build_test_case(rag_pipeline, example) if REFUSAL_MARKER in answer.lower(): refusals += 1 refusal_rate = refusals / len(adversarial) assert refusal_rate >= REFUSAL_RATE_THRESHOLD, ( f"only {refusals}/{len(adversarial)} adversarial questions triggered a refusal, " f"below the {REFUSAL_RATE_THRESHOLD} threshold" )
Notice both generation test files compute a mean across the eval set and assert on that mean, never on any single example. That's deliberate, and it's the difference between a harness a team trusts and one they route around after the third false alarm.
Eval set in, aggregate score out. The threshold gates on the mean across the set, which is what keeps the harness usable in CI instead of noisy.
Wiring this into CI is the same shape as any other regression suite: run it on every pull request, fail the merge if the aggregate drops below threshold, and give the team a real signal instead of a manual step someone forgets before a release.
Worth being clear about what that gate does and doesn't cover, because it's easy to read a green check as broader than it is: it gates the pipeline, not the feature. A pull request that changes the chat component, the citation link builder, or the fallback screen passes this suite untouched, because none of those files can move a faithfulness score. We built Autonoma for that other half: our Planner reads the application code around a RAG feature and plans end-to-end cases from it, including the database state each one needs, so the two gates cover the pipeline and the product surface separately instead of one standing in for the other.
Why RAG Tests Flake (and the Fix)
Generation scores are sampled measurements, not fixed properties of an answer, because both the generator and, if you're using one, the judge model scoring it are non-deterministic. Exact-match assertions were never the right tool here: two correct answers can use completely different words, so assert on properties and score thresholds instead of string equality.
Run the eval set multiple times rather than once, and gate on the distribution instead of a single pass. Five runs with four clearing threshold is a defensible floor for most generation metrics, and it's the aggregate across the set that matters, never any single example, for the same reason a batting average means more than one at-bat. If you're relying on an LLM as the judge for faithfulness or relevancy, pin its model version, set temperature to zero where the provider allows it, and treat disagreement between repeated judge calls as signal about your rubric, not noise to average away.
The retrieval layer stays out of all of this. Given a fixed index and a fixed query, retrieval is reproducible, which is exactly why testing it as its own surface gives you a stable signal while generation stays fuzzy around it. The practitioner rule worth pinning above your desk: a flaky RAG test is either an assertion that's too strict for normal variance, or a prompt and query that are genuinely too ambiguous to have one right answer. It is never "the model is just being random today," and treating it as such is how thresholds quietly drift until the suite stops meaning anything.
"""Wrap a non-deterministic check so it gates on a pass rate, not one run.Usage: from lib.repeat_and_threshold import repeat_and_gate def one_run() -> bool: score = my_faithfulness_check(answer, context) return score >= 0.8 result = repeat_and_gate(one_run, n=5, min_pass_rate=0.8) assert result.passed, f"pass rate was {result.pass_rate:.2f} across {result.n} runs""""from dataclasses import dataclassfrom typing import Callable@dataclassclass GateResult: n: int passes: int pass_rate: float passed: bool runs: list[bool]def repeat_and_gate(check: Callable[[], bool], n: int = 5, min_pass_rate: float = 0.8) -> GateResult: """Run `check` n times and gate on the fraction of runs that pass. This is the pattern the article's non-determinism section leans on: a single run of a faithfulness or judge check is a sample, not a fixed property of the answer. Running it once and trusting that one result is how a threshold quietly drifts until the suite stops meaning anything. Running it N times and gating on the pass rate turns a noisy single sample into a distribution you can actually threshold. """ runs = [bool(check()) for _ in range(n)] passes = sum(runs) pass_rate = passes / n return GateResult( n=n, passes=passes, pass_rate=pass_rate, passed=pass_rate >= min_pass_rate, runs=runs, )
There's a slower failure mode hiding behind flake, and it's the more dangerous of the two: a suite that stops describing the system. Chunking changes, a document set gets restructured, a prompt is rewritten, and the eval set keeps passing against expectations that no longer match what the pipeline is for. Nothing goes red, which is exactly the problem. On the behavioral side we handed that upkeep to our Diffs Agent, which reads each pull request's code diff and adds, updates, or deprecates the affected end-to-end cases, so drift shows up as a changed test rather than as coverage that quietly stopped meaning anything.
How Autonoma Fits Above the RAG Pipeline
Every check in this guide, retrieval, faithfulness, relevancy, refusal, answers the same underlying question: did the pipeline produce a grounded, relevant answer. Green scores across the board prove exactly that and nothing more. They don't prove the citation link under that answer actually resolved to the right document, that the streaming response didn't truncate mid-sentence in the UI, that the "I don't have enough information" path rendered the fallback screen instead of a blank div, or that a follow-up question kept the retrieved context instead of starting the conversation over. Those are failures your eval harness will never see, because they happen one layer up, in the running application the pipeline is wired into.
That's the layer Autonoma operates in, and it's worth being precise about the boundary: our platform runs behavioral end-to-end tests against the real, running app, it doesn't score a faithfulness metric or grade a model's output, that job belongs entirely to the harness above. Concretely, the Planner turns the routes and components around a RAG feature into test cases, the Executor drives those cases against the running application, and the Reviewer classifies each result as a real bug, an agent error, or a plan that no longer matches the code. Your eval set proves the pipeline returned a grounded answer; behavioral E2E on the feature the pipeline powers confirms the answer actually helped the user in the product they were using.
The Practitioner Takeaway
Separate the two surfaces before you write a single assertion. Retrieval is deterministic and testable with hard thresholds today; generation is fuzzy and needs aggregate gating across a set, run more than once. Build that set from your own documents, not a benchmark, and keep it small enough that every red result gets a human's attention. This piece is the hub for testing the pipeline itself; if you want the metrics deep dive specifically, RAG evaluation metrics that matter covers the app-builder's practical subset in more depth, and if the failure you're chasing is specifically about what the retriever returned, how to test RAG retrieval is the layer below this one. Own the eval set, gate the aggregate, and the next confidently wrong answer will come with an answer to which stage actually broke.
That leaves one boundary worth naming plainly, since it decides what you build versus what you add. Everything above is the pipeline's own test surface, and it should stay yours: your documents, your thresholds, your judgment about what counts as grounded. Autonoma is what we'd put above it, not inside it. It scores no metric and grades no model output, it drives the feature the pipeline powers and asserts the application ended up in the right state, which is the one failure a perfect faithfulness score is structurally unable to report.
Frequently Asked Questions
Test retrieval and generation as two separate surfaces instead of one end-to-end score. Assert that the retriever returns the known-correct chunk within the top-k results, then separately assert that the generator's answer is faithful to whatever context it was given and actually answers the question, using threshold-gated metrics computed with Ragas or DeepEval against an eval set built from your own documents.
Retrieval testing checks whether the right chunks came back from the index, and it's deterministic: given a fixed index and query, you get the same result every time. Generation testing checks whether the answer produced from those chunks is faithful and relevant, and it's non-deterministic, since the same context can produce different wording on different runs. They fail for different reasons and need different fixes, which is why they need separate assertions.
Twenty to fifty curated examples, each reviewed by a human, catches more real regressions than several hundred synthetic ones nobody has looked at. A small, trusted set means every failure gets investigated. A large, unreviewed set means failures get dismissed as noise, which defeats the point of having the suite.
Yes, for the reference-free checks. Faithfulness only needs the retrieved context and the generated answer, no labeled ground truth required, which is why it scales to live traffic. Answer relevancy works the same way. Context precision and recall against a known-correct chunk do need a labeled eval set, which is why building one from your own documents is still worth doing even if you lean on reference-free checks elsewhere.
An eval harness proves the pipeline returns a grounded, relevant answer given a fixed input. It doesn't prove the feature works for a user in the running app: whether a citation link resolves, whether a streaming answer renders without truncating, whether the fallback UI appears when context is insufficient, or whether a follow-up question keeps the retrieved context. That's a separate, behavioral test layer that sits above the pipeline and needs its own coverage in the actual application.
