RAG in Python: answers that cite your documents, and how to test them

A RAG app has a simple promise: it answers from your documents, not from whatever the model remembers. Module 4 built the search that finds the passages. This module puts a model on top of it and makes the whole thing correct, which is most of the work. A RAG demo takes an afternoon. A RAG app that a customer can trust needs every step below.

Do you need RAG at all?

RAGRAGA way to answer from your own documents. A search finds the passages that match the question, and they are added to the prompt before the model writes. exists because documents do not fit in a prompt. If yours do, skip it. Anthropic's rule of thumb is a knowledge base under 200,000 tokens, about 500 pages: send all of it with every question, and contextContext WindowEverything a model can see at once while it writes an answer, counted in tokens. When the window is full, something has to be left out. is the only retrieval you need. Nothing is missed, there is no index to keep current, and prompt caching makes the repeated text cheap.

ToolPrompt caching calculatorYour repeated prompt and traffic in; the month with no cache, the 5-minute and the 1-hour cache out.

RAG earns its place when the library is far larger than that, when it changes faster than you want to resend it, or when each user may only see part of it. If what you want is a different tone or format rather than new facts, that is a job for the prompt or for fine-tuning, not retrieval.

ComparisonRAG vs fine-tuningUse RAG or long context to give a model facts, and fine-tuning to change how it behaves. RAG looks up the passages that match each question and puts them in the…

How the answer gets made

Every question goes through 4 steps:

1. Rewrite the message into a question that makes sense on its own, if it depends on the conversation.
2. Search for the passages most likely to answer it, with module 4's hybrid search and reranker.
3. Answer from those passages only, citing the ones it used, or say that they do not answer.
4. Check, before any change ships, that the search finds the right passages and the answers stick to them.

Each step fails in its own way, and the fix for one does nothing for the others. The subject page on data engineering follows a document through the same path, from the file to the index to the answer.

WalkthroughData engineeringEvery AI answer about a business is only as good as the data under it. Follow an order, a document and a card payment from the moment they are made to the…

Measure the search before the answers

When an answer is wrong, either the search missed the passage or the model misused it. You cannot tell which from the answer, so measure the search on its own first. Write down questions with the passage that answers each, run the search, and count how often that passage is in the top k results. That share is recall at kRetrieval EvaluationTesting the search step of a RAG system on its own, with questions whose answering passages you know, and scoring it with recall at k, the share of those passages found in the top k results., and it is the number every retrieval change has to move.

Choose k to match how many passages you send the model. In Anthropic's retrieval tests, sending 20 passages did better than 10 or 5. But more is not free: each passage costs tokens on every question, and research on long inputs found models use information at the start and the end of the prompt better than in the middle. So send the reranked passages in order, best first, and let your eval decide k instead of a rule.

The Build it scores 4 ways of searching on the same questions. On this small set the answer is humbling: meaning search alone already finds every right passage in the top 3. On your documents, with thousands of chunks and product codes, it will differ, and you will only know by measuring.

Give chunks their context back

A chunk cut out of a document forgets what it was about. "It costs £9.95 and is not available for large appliances" does not say it is express delivery. Module 4 put the page title and heading on top of each chunk. Anthropic's method, contextual retrieval, goes further: when a chunk is indexed, a model reads the whole document and writes 1 or 2 sentences placing the chunk in it, and that context is embedded and keyword-indexed with the chunk.

SITUATE = """Here is a chunk from the document above:
<chunk>
{chunk}
</chunk>
In 1 or 2 sentences, say what this chunk is about and where it sits in the document, to help a search
find it. Reply with those sentences only."""


def situate(document: str, chunk: str) -> str:
    """A short context for one chunk, written once when the chunk is indexed."""
    response = client.messages.create(
        model="claude-haiku-5-5",
        max_tokens=200,
        system=[{"type": "text", "text": f"<document>\n{document}\n</document>",
                 "cache_control": {"type": "ephemeral"}}],  # the same document for every chunk: cache it
        messages=[{"role": "user", "content": SITUATE.format(chunk=chunk)}],
    )
    return next(block.text for block in response.content if block.type == "text").strip()

The document sits in a cached system prompt, so for its second chunk onwards it is read from the cache at a fraction of the price. The cache applies only above a minimum length, which the calculator above lists for each model. In Anthropic's tests the method cut failed searches by 35% on its own, by 49% with contextual keyword search too, and by 67% with a reranker, at a one-time cost of about $1 per million document tokens with the model they used in 2024. Run it when you index, never when a customer is waiting, and check that recall goes up on your questions before you pay for it on the whole library.

Rewrite follow-up questions

A customer asks whether the Aster 26 cm pan works on induction, gets an answer, then types "And the copper one?". Searched as typed, those 4 words find nothing. Query rewritingQuery RewritingHaving a model turn what the user typed into a better search query before retrieval, most often turning a follow-up like "and the bigger one?" into a full question the search can match. gives a small model the conversation and asks for the question it means:

REWRITE = """Here is a conversation between a customer and a shop's assistant:
<conversation>
{conversation}
</conversation>
Rewrite the customer's last message as one question that makes sense on its own, naming every
product and detail it refers to. Reply with the question only."""


def standalone(history: list[dict], message: str) -> str:
    """The question to search for: the last message, with what it refers to spelled out."""
    if not history:
        return message
    lines = [f"{turn['role']}: {turn['content']}" for turn in history] + [f"user: {message}"]
    response = client.messages.create(
        model="claude-haiku-5-5",
        max_tokens=200,
        messages=[{"role": "user", "content": REWRITE.format(conversation="\n".join(lines))}],
    )
    return next(block.text for block in response.content if block.type == "text").strip()

The first message of a conversation skips the call. The rewritten question goes to the search; the customer's own words still go to the model that answers, so the reply fits the conversation. Add follow-ups to your retrieval questions and measure recall with and without the rewrite, because each rewrite is an extra call on the way to every answer.

Answer with citations

An answer a customer can check is worth more than one they must take on trust, and a citation makes a missing source visible. You can ask a model to quote its sources in the prompt, but nothing checks the quotes. The Claude API has citations built in: pass each passage as a search_result block with citations turned on, and each part of the answer comes back as a text block carrying the passages it relied on.

{
    "type": "search_result",
    "source": "delivery#express-delivery",          # any stable id or URL; it comes back in the citation
    "title": "Delivery: Express delivery",
    "content": [{"type": "text", "text": "Order by 2pm on a working day and express delivery arrives the next working day."},
                {"type": "text", "text": "It costs £9.95 and is not available for large appliances."}],
    "citations": {"enabled": True},
}

The model cites whole text blocks, so 1 block per sentence lets a citation point at the sentence that matters. Each citation carries the source and title you gave and the cited_text, and the API guarantees it points to a real place in what you sent. The cited text does not count as output tokens. Anthropic found the feature more likely to cite the most relevant quote than citing by prompt.

2 rules come with it. Citations must be on for every search result in a request or for none. And citations cannot be combined with structured outputs: a request with both returns a 400 error. If code needs fields from the answer, build them from the text blocks and their citations, as the Build it does.

Let it say "I don't know"

The search always returns k passages, even when none of them answers the question. Ask a help bot about coffee machines and it gets the 3 passages closest to "coffee machines", which may be about mixers. What the model does next is up to your prompt. A prompt that says "be helpful and confident" gets an answer built from whatever is nearest. That is how a help bot promises a warranty nobody offers.

ScenarioThe warranty the help bot inventedA customer was promised a 2-year warranty on a refurbished laptop. There is no such warranty. Work out why the bot said it, and what stops the next one.

GroundingGroundingMaking a model answer from sources you give it, such as retrieved documents, and only from them, so every statement can be traced to a passage and a missing answer is admitted, not invented. the answer takes 3 lines in the system promptSystem PromptInstructions the builder of an AI app gives the model before you type anything. It sets the role, the tone and the rules, and it outranks what users say.: answer only from the passages given, say plainly when they do not answer the question, and never guess a price, a time or a policy. Then hand the conversation to a person. Test the way out as you test the answers: an eval with only answerable questions rewards a bot that answers everything.

Keep the documents current

A perfectly grounded answer can still be wrong, if the passage it was grounded in is out of date. The price changed on the website and the index still holds last quarter's page. Nothing in the answer looks wrong, and the citation even points to the source. The fix is in the pipeline, not the prompt: re-index a page when it changes, remove pages that are archived, and store the date each chunk was indexed, so a question about prices can be checked against it.

ScenarioThe AI assistant that quotes old pricesThe sales team's AI assistant quoted a customer the old price, 6 weeks after it changed. The document is right. The assistant's copy of it is not.

Test the answers

The search eval tells you the passages arrived. The answer eval tells you the model used them well. For each question, write down what a good answer must do, and grade it like module 7's triage eval, with code where a rule decides and a judge where it cannot:

• Cites the right page. Code: the expected source is among the citations.
• States the fact that matters. Code: "£9.95" is in the answer about express delivery.
• Says nothing the passages do not. A judge: an LLM judgeLLM as a JudgeUsing a language model to grade another model's output against a rubric, for qualities code cannot check. It scales to thousands of answers, but must be checked against a person's verdicts before it is trusted. given the passages and the answer checks that every statement is supported. This catches the answer that cites the right page and then adds a delivery date of its own.
• Declines when it should. Code: a question the pages cannot answer gets no citation, and the judge checks it said so instead of answering from memory.

Run each case more than once, as module 7 did, and grow the set from the questions customers really ask.

Build it: answers from the shop's help pages, with citations

This adds 3 files beside search.py and the docs folder from module 4: a test of the search, the answering function, and a test of the answers.

First the search test. Save the questions and the chunk that answers each as questions.jsonl. The ids are the page name and the section heading, as search.py makes them:

{"question": "How much is next day delivery?", "relevant": ["delivery#express-delivery"]}
{"question": "Can you send my order to France?", "relevant": ["delivery#where-we-deliver"]}
{"question": "Does the CC-28 work on induction?", "relevant": ["cookware#copper-classic-28-cm-pan-cc-28"]}
{"question": "My parcel tracking hasn't updated in days", "relevant": ["delivery#tracking-an-order"]}
{"question": "The bowl arrived cracked, what do I do?", "relevant": ["returns#damaged-or-faulty-on-arrival"]}
{"question": "How long until I get my money back?", "relevant": ["returns#when-your-refund-arrives"]}
{"question": "Is the mixer motor guaranteed?", "relevant": ["warranty#how-long-each-product-is-covered"]}
{"question": "Spots on my knife after the dishwasher", "relevant": ["knives#cleaning-your-knives"]}
{"question": "Can the AST-26 go in the oven?", "relevant": ["cookware#aster-26-cm-frying-pan-ast-26"]}
{"question": "Is the AST-28 dishwasher safe?", "relevant": ["cookware#aster-28-cm-frying-pan-ast-28"]}
{"question": "Which pans can I use on induction?", "relevant": ["cookware#aster-26-cm-frying-pan-ast-26", "cookware#aster-28-cm-frying-pan-ast-28", "cookware#copper-classic-28-cm-pan-cc-28"]}
{"question": "Do you sharpen knives?", "relevant": ["knives#sharpening"]}
{"question": "What does the guarantee not cover?", "relevant": ["warranty#what-the-warranty-covers"]}
{"question": "Can I send back a pan I never used?", "relevant": ["returns#returning-an-item"]}
{"question": "My mixer stopped and shows E04", "relevant": ["mixer#error-e04"]}
{"question": "What does E07 mean on the mixer?", "relevant": ["mixer#error-e07"]}
{"question": "How do I fit the dough hook?", "relevant": ["mixer#attaching-the-tools"]}

Then recall.py, which scores each way of searching on them:

"""How often each way of searching puts the passage that answers the question in its top k."""
import json

from search import Index

index = Index()
cases = [json.loads(line) for line in open("questions.jsonl")]

for k in (1, 3):
    for method in ("by_meaning", "by_keyword", "hybrid", "search"):
        total, missed = 0.0, []
        for case in cases:
            found = {c["id"] for c in getattr(index, method)(case["question"], k)}
            recall = len(found & set(case["relevant"])) / len(case["relevant"])
            total += recall
            if recall < 1:
                missed.append(case["question"])
        print(f"recall@{k} {method:>10}: {total / len(cases):.0%}")
        for question in missed:
            print(f"    missed: {question}")

Run python recall.py. It needs no API key, since the search runs on your machine. When we ran it, the top lines of each block read:

recall@1 by_meaning: 78%
recall@1 by_keyword: 67%
recall@1     hybrid: 67%
recall@1     search: 78%
recall@3 by_meaning: 100%
recall@3 by_keyword: 80%
recall@3     hybrid: 94%
recall@3     search: 100%

Read the missed questions under each line, not only the scores. Keyword search misses when the customer's words are not the page's. The induction question has 3 right chunks, so no method can score above a third on it at k=1. At k=3, the number the assistant uses, meaning search and the full search both find everything; the reranker's job here is order, which matters once k is small. Add questions from real customers until a change shows up in these numbers.

Now the answers. Save this as answer.py:

"""Answer a customer's question from the shop's help pages, citing the passages it used."""
import re
import sys
from dataclasses import dataclass

import anthropic

from search import Index

MODEL = "claude-opus-5-5"
SYSTEM = """You answer customers' questions for an online shop that sells kitchen equipment.
Answer only from the help pages given with the question. If they do not answer it, say that
you cannot find it in the help pages and that a person from the team will reply.
Never guess a price, a time or a policy. Keep the answer short and friendly."""

client = anthropic.Anthropic()
index = Index()


@dataclass
class Answer:
    text: str            # the answer, with [1], [2] after the parts that cite a page
    cited: list[str]     # the chunk ids it cited, in the order of the numbers
    passages: list[str]  # every passage the model was given, for the grader


def as_search_result(chunk: dict) -> dict:
    # One text block per sentence, so a citation points at the sentence, not the whole section.
    sentences = re.split(r"(?<=[.!?])\s+", chunk["text"])
    return {
        "type": "search_result",
        "source": chunk["id"],
        "title": chunk["title"],
        "content": [{"type": "text", "text": s} for s in sentences],
        "citations": {"enabled": True},
    }


def answer(question: str, k: int = 3) -> Answer:
    chunks = index.search(question, k)
    response = client.messages.create(
        model=MODEL,
        max_tokens=1000,
        system=SYSTEM,
        messages=[{"role": "user", "content": [*map(as_search_result, chunks),
                                               {"type": "text", "text": question}]}],
    )
    passages = [f"{c['title']}\n{c['text']}" for c in chunks]
    if response.stop_reason != "end_turn":
        return Answer("Sorry, I can't answer that here. A person from our team will reply.", [], passages)
    text, cited = "", []
    for block in response.content:
        if block.type != "text":
            continue
        text += block.text
        for source in dict.fromkeys(c.source for c in block.citations or []):
            if source not in cited:
                cited.append(source)
            text += f" [{cited.index(source) + 1}]"
    return Answer(text.strip(), cited, passages)


if __name__ == "__main__":
    result = answer(" ".join(sys.argv[1:]))
    print(result.text)
    for number, source in enumerate(result.cited, start=1):
        print(f"[{number}] {source}")

The answer arrives as several text blocks, some with citations and some without. The loop joins them and puts a number after each part that cites a page, with the page ids listed below, which is the shape to show a customer. A stop for any reason other than the end of the answer, a refusal or a cut-off, gets the hand-over to a person, as module 2 taught.

python answer.py "My mixer stopped and shows E04. What should I do?"
python answer.py "Do you sell coffee machines?"

The first should cite mixer#error-e04 and tell the customer to switch off and wait 30 minutes. The second should cite nothing and say a person will reply.

Last, the answer test. Save the cases as rag_cases.jsonl. Each says which page a good answer cites and which facts it must contain; an empty cites means the pages cannot answer it, so the answer must decline:

{"id": "express-price", "question": "How much is express delivery?", "cites": ["delivery#express-delivery"], "facts": ["£9.95"]}
{"id": "france", "question": "Can you deliver to France?", "cites": ["delivery#where-we-deliver"], "facts": []}
{"id": "mixer-e04", "question": "My mixer stopped and shows E04. What should I do?", "cites": ["mixer#error-e04"], "facts": ["30 minutes"]}
{"id": "ast28-dishwasher", "question": "Can the AST-28 go in the dishwasher?", "cites": ["cookware#aster-28-cm-frying-pan-ast-28"], "facts": []}
{"id": "toaster-warranty", "question": "How long is the warranty on a toaster?", "cites": ["warranty#how-long-each-product-is-covered"], "facts": []}
{"id": "refund-tomorrow", "question": "I posted my return yesterday. Will the refund be in my account tomorrow?", "cites": ["returns#when-your-refund-arrives"], "facts": []}
{"id": "coffee", "question": "Do you sell coffee machines?", "cites": [], "facts": []}

The toaster case checks a small inference, since the pages say "electrical appliances", and the refund case checks that the answer does not promise tomorrow. Then rag_evals.py, which reuses the judge's verdict model and the number of runs from module 7's evals.py:

"""Grade the answers: the right page cited, the facts that matter, and nothing the pages do not say."""
import json

import anthropic

from answer import Answer, answer, client
from evals import JUDGE, TRIALS, Verdict

VERDICT_SCHEMA = anthropic.transform_schema(Verdict)
GROUNDED_PROMPT = """You check answers that a shop's assistant wrote from its help pages.
Pass if every statement in the answer is supported by the passages, or if the answer says it
cannot find the information. Fail if the answer states anything the passages do not say,
such as a price, a time, a date or a policy, even if it sounds likely.

<passages>
{passages}
</passages>

<answer>
{answer}
</answer>

Reason about each statement first, then give the verdict."""


def grounded(result: Answer) -> bool:
    prompt = GROUNDED_PROMPT.format(passages="\n\n".join(result.passages), answer=result.text)
    response = client.messages.create(
        model=JUDGE,
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}],
        output_config={"format": {"type": "json_schema", "schema": VERDICT_SCHEMA}},
    )
    if response.stop_reason != "end_turn":
        return False
    text = next(block.text for block in response.content if block.type == "text")
    return Verdict.model_validate_json(text).verdict == "pass"


def grade(case: dict, result: Answer) -> dict[str, bool]:
    checks = {"grounded": grounded(result)}
    if case["cites"]:
        checks["cites"] = set(case["cites"]) <= set(result.cited)
        checks["facts"] = all(fact in result.text for fact in case["facts"])
    else:
        checks["declines"] = not result.cited  # nothing to cite, so the answer must not lean on a page
    return checks


if __name__ == "__main__":
    cases = [json.loads(line) for line in open("rag_cases.jsonl")]
    passed = 0
    for case in cases:
        failures = set()
        for _ in range(TRIALS):
            result = answer(case["question"])
            failures |= {check for check, ok in grade(case, result).items() if not ok}
        passed += not failures
        if failures:
            print(f"  FAIL {case['id']}: {', '.join(sorted(failures))}\n       {result.text}")
    print(f"{passed} of {len(cases)} questions passed every run")

The judge call can use structured outputs because it has no citations in it; only the answer call does. Run python rag_evals.py and read every failure with the answer printed under it. A cites failure with a good answer may mean the answer cited a different page that also holds the fact, so look before you change anything. A grounded failure is the one to fix first: the answer said something your pages do not.

Keep both files growing. Every question a customer asks that the search misses goes into questions.jsonl, and every wrong answer goes into rag_cases.jsonl. Module 11 puts this answer function, the triage script and the agent from module 6 into one assistant, keeps these evals running beside a check of the whole assistant, and ships only when that check passes.