Capstone: chat with your documents (RAG)

The big idea

You will build the most common AI product there is: a chat box that answers questions from a set of documents, such as a help center, an employee handbook or a folder of contracts, with a source link on every claim and an honest "I could not find this" when the documents do not say.

The finished app does five jobs. It ingests documents into chunks and indexes them, finds the chunks that match a question with two kinds of search, puts the best ones in order with a reranker, writes an answer that cites those chunks or refuses, and measures all of that against a small eval set so you know whether the next change helped.

Think of a good reference librarian. They have catalogued every book (ingestion), know how to look things up both by exact title and by topic (hybrid search), pull a few books and skim them to pick the best (reranking), answer with page numbers you can check (citations), and say "we do not have anything on that" rather than invent a book (refusal). Your app does the same, and the eval set is the librarian's performance review.

How it works

What you need

• Python 3.10 or newer and a folder of 20 to 200 documents you know well. Your own notes, a product's public docs or a handbook all work. Knowing the content is what lets you write eval questions.
• Access to a model API for the final answer, and an embedding model (an API or a small open-weights model run locally). Until Checkpoint 4 you do not need either: the runnable core below uses a fake model and a toy embedding.
• A place for vectors. For a few thousand chunks, a NumPy array or SQLite is enough; move to a vector database or a Postgres extension when you have far more or need filtering at scale ( Vector databases and ANN indexes ).

Lay the project out so each stage can be tested alone:

rag-app/
  ingest.py      # documents -> chunks.jsonl (+ embeddings)
  search.py      # bm25(), vector_search(), hybrid(), rerank()
  answer.py      # prompt template, model call, citation check, refusal
  evals/
    questions.jsonl   # question, gold chunk id, required fact (or "refuse")
    run_evals.py
  app.py         # the chat endpoint or a command-line loop

Checkpoint 1: ingest and chunk

Parse each document to plain text but keep its structure: headings, lists and tables. Split along headings first and only then by length ( Document ingestion and chunking ), aiming for chunks of a few hundred tokens with a small overlap when a section is long. Give every chunk a stable id such as returns#exceptions and store its source path and heading.

Prefix each chunk with its context before indexing, for example "returns > Exceptions:". A chunk that says "Gift cards cannot be returned" is much easier to find, and to cite, when it also says which policy it belongs to. This is a cheap version of contextual retrieval ( Contextual retrieval and query transformation ).

Done when: you can print every chunk with its id and source, and no chunk cuts a table or a sentence in half.

Checkpoint 2: two indexes and hybrid search

Build a keyword index with BM25 and a vector index with your embedding model. Keyword search finds exact strings, such as error codes, product names and order numbers. Vector search finds paraphrases, such as "send it back" for "return". Run both, then merge the rankings with reciprocal rank fusionA way to merge several ranked lists by summing 1 / (k + rank) for each item across lists, so items ranked high by any list rise to the top.: each chunk scores the sum of 1 / (60 + rank) over the two lists, so you never have to compare raw scores from different systems.

Done when: for 10 questions you write by hand, the right chunk is in the fused top 20, and you have at least one question that only keyword search gets right and one that only vector search gets right.

Checkpoint 3: rerank

Fusion finds candidates; it does not order them well. A rerankerA model that reads the query and one candidate passage together and outputs a relevance score, used to reorder a short list of search results. (a cross-encoder) reads the question and each candidate together, which is slower but much more precise. Rerank the top 20 and keep the top 3 to 5. Keep the reranker's top score too: it is your evidence strength for the refusal decision.

Done when: the right chunk is ranked first for most of your hand-written questions, and you have looked at the cases where it is not.

Checkpoint 4: cited answers and refusal

Now call a real model. Put the instructions first and the chunks with their ids in tags, so the model can refer to them:

Answer the question using only the documents below.
After every sentence that uses a document, cite it as [id].
If the documents do not contain the answer, reply exactly:
"I could not find this in the documents."

<doc id="returns#returns">returns > Returns: Items can be returned within 30 days ...</doc>
<doc id="shipping#shipping">shipping > Shipping: Standard delivery takes 3 to 5 ...</doc>

Question: How many days do I have to send a jacket back?

Then check the answer in code before showing it. Every cited id must be one you actually retrieved, and an answer without any citation is treated as unsupported. Refuse in two places: before the model call when the best reranker score is below a threshold, and after it when the model itself says it cannot find the answer. In the UI, show citations as clickable markers that open the chunk ( UX patterns for AI ).

Done when: answers carry valid citations, an invented id is caught by your check, and a question about something not in your documents gets the refusal.

Checkpoint 5: a small eval set

Write 30 to 50 questions from your documents ( Building an eval set ). For each, record the chunk id that holds the answer and a fact the answer must contain. Add 5 to 10 questions the documents cannot answer, marked "refuse". Score four things separately: recall@kThe share of questions for which the chunk holding the answer appears in the top k retrieved results. of retrieval, whether the answer cites the right chunk, whether it contains the required fact, and whether it refuses exactly when it should. Separate scores tell you which stage to fix ( Evaluating RAG ).

Done when: one command prints all four scores, and you have read every failure.

Try it

Here is the whole pipeline in one file, small enough to read in a few minutes. It uses a toy concept-based embedding and a fake model that quotes the best sentence, so it runs anywhere without keys:

# Chat with your documents, the whole pipeline in one file: ingest, hybrid search, rerank,
# cited answers and refusal. Toy embedding and fake model, so it runs offline.
import math, re
from collections import Counter

DOCS = {  # a made-up store's help center: doc id -> markdown
    "returns": "# Returns\nItems can be returned within 30 days of delivery.\nRefunds go to the "
               "original payment method within 5 business days.\n# Exceptions\nGift cards and "
               "opened earbuds cannot be returned.",
    "shipping": "# Shipping\nStandard delivery takes 3 to 5 business days.\nExpress delivery "
                "arrives the next business day if ordered before 2 pm.",
    "warranty": "# Warranty\nElectronics have a 2 year warranty against defects.\nWarranty "
                "claims need the order number and a photo of the fault.",
}
SYN = {"refund": "money", "money": "money", "back": "money", "return": "return", "returned": "return",
       "send": "return", "refunds": "money", "delivery": "ship", "arrive": "ship", "arrives": "ship",
       "shipping": "ship"}
STOP = set("a an the to of and or i you my do does can is are how many what when have for be".split())
words = lambda t: [w for w in re.findall(r"[a-z0-9]+", t.lower()) if w not in STOP]

def ingest(docs):  # chunk by heading; prefix each chunk with its doc and section (context)
    chunks = []
    for doc, text in docs.items():
        for sec in text.split("# ")[1:]:
            head, body = sec.split("\n", 1)
            chunks.append({"id": f"{doc}#{head.lower()}", "text": f"{doc} > {head}: {body}"})
    return chunks

CHUNKS = ingest(DOCS)
DF = Counter(w for c in CHUNKS for w in set(words(c["text"])))
AVG = sum(len(words(c["text"])) for c in CHUNKS) / len(CHUNKS)

def bm25(q, c, k1=1.2, b=0.75):
    tf, n = Counter(words(c["text"])), len(words(c["text"]))
    return sum(math.log(1 + (len(CHUNKS) - DF[w] + 0.5) / (DF[w] + 0.5)) * tf[w] * (k1 + 1)
               / (tf[w] + k1 * (1 - b + b * n / AVG)) for w in words(q) if w in tf)

def embed(t):  # toy "embedding": words mapped to concepts, as a unit vector
    v = Counter(SYN.get(w, w) for w in words(t))
    norm = math.sqrt(sum(x * x for x in v.values()))
    return {k: x / norm for k, x in v.items()}

VECS = {c["id"]: embed(c["text"]) for c in CHUNKS}  # the "vector store"

def retrieve(q, k=3):  # hybrid: reciprocal rank fusion of BM25 and vector rankings
    qv = embed(q)
    by_bm25 = sorted(CHUNKS, key=lambda c: -bm25(q, c))
    by_vec = sorted(CHUNKS, key=lambda c: -sum(qv.get(w, 0) * x for w, x in VECS[c["id"]].items()))
    rrf = Counter()
    for ranking in (by_bm25, by_vec):
        for rank, c in enumerate(ranking):
            rrf[c["id"]] += 1 / (60 + rank + 1)
    return [c for c in CHUNKS if c["id"] in dict(rrf.most_common(k))]

def rerank(q, chunks):  # stand-in for a cross-encoder: reads query and chunk together
    qc = set(embed(q))
    score = lambda c: len(qc & set(embed(c["text"]))) / len(qc)
    return sorted(((score(c), c) for c in chunks), key=lambda x: -x[0])

def answer(q, threshold=0.6):  # share of the question's concepts the best chunk must cover
    best_score, best = rerank(q, retrieve(q))[0]
    if best_score < threshold:  # refuse instead of guessing
        return "I could not find this in the documents.", None
    qc = set(embed(q))  # fake model: quote the best-supported sentence and cite it
    sentences = re.split(r"(?<=\.)\s+", best["text"].split(": ", 1)[1])
    sent = max(sentences, key=lambda s: len(qc & set(embed(s))))
    return f"{sent} [{best['id']}]", best["id"]

if __name__ == "__main__":
    print(len(CHUNKS), "chunks:", ", ".join(c["id"] for c in CHUNKS))
    for q in ["How many days do I have to send a jacket back?", "When does express delivery arrive?",
              "Do you ship to Mars?"]:
        print(f"Q: {q}\nA: {answer(q)[0]}")

4 chunks: returns#returns, returns#exceptions, shipping#shipping, warranty#warranty
Q: How many days do I have to send a jacket back?
A: Items can be returned within 30 days of delivery. [returns#returns]
Q: When does express delivery arrive?
A: Express delivery arrives the next business day if ordered before 2 pm. [shipping#shipping]
Q: Do you ship to Mars?
A: I could not find this in the documents.

And the eval set, which loads the app above and scores it:

# A small eval set for the RAG app: retrieval recall, citation, answer and refusal checks.
import os, runpy

app = runpy.run_path(os.path.join(os.path.dirname(__file__), "AI-15.01-rag.py"))  # load the app
retrieve, answer = app["retrieve"], app["answer"]

EVAL = [  # question, chunk that holds the answer (None = should refuse), fact the answer must contain
    ("How many days do I have to send a jacket back?", "returns#returns", "30 days"),
    ("When will my refund money arrive?", "returns#returns", "5 business days"),
    ("Can I return opened earbuds?", "returns#exceptions", "cannot be returned"),
    ("How long is standard delivery?", "shipping#shipping", "3 to 5"),
    ("What do I need for a warranty claim?", "warranty#warranty", "photo"),
    ("Do you ship to Mars?", None, None),
    ("What is the CEO's phone number?", None, None),
]

rows, scores = [], {"recall@3": [], "cited right": [], "answer ok": [], "refusal ok": []}
for q, gold, fact in EVAL:
    text, cited = answer(q)
    if gold is None:
        scores["refusal ok"].append(cited is None)
    else:
        scores["recall@3"].append(gold in [c["id"] for c in retrieve(q)])
        scores["cited right"].append(cited == gold)
        scores["answer ok"].append(fact in text)
    rows.append(f"{'PASS' if (cited is None) == (gold is None) and (gold is None or fact in text) else 'FAIL'}  {q}")

print("\n".join(rows))
for name, s in scores.items():
    print(f"{name:<12} {sum(s)}/{len(s)}")

PASS  How many days do I have to send a jacket back?
FAIL  When will my refund money arrive?
PASS  Can I return opened earbuds?
PASS  How long is standard delivery?
PASS  What do I need for a warranty claim?
PASS  Do you ship to Mars?
PASS  What is the CEO's phone number?
recall@3     5/5
cited right  5/5
answer ok    4/5
refusal ok   2/2

1. Read the one failure. Retrieval found the right chunk and the citation is right, but the answer quotes the wrong sentence. Which stage would you fix, and how do the separate scores tell you that without reading code?
2. Change the refusal threshold in answer() from 0.6 to 0.4 and run both files. What happens to "Do you ship to Mars?" and why is a threshold a trade-off, not a constant?
3. Add a question that only exact matching can answer, such as one about "2 pm", and check whether BM25 or the toy embedding finds it.

Going deeper

Swapping in real parts

Replace one stage at a time and re-run the evals after each: the embedding function, then the vector store, then the reranker, then the fake model. If a score drops after a swap, you know exactly which part caused it. Log every request as a trace with the retrieved ids, scores, prompt and answer ( LLM observability and tracing ), because those traces become your next eval cases.

Keeping the index fresh

Documents change. Store a content hash per document and re-ingest only what changed; delete chunks of removed documents, or the app will cite a policy that no longer exists. Record the index version in each trace so an answer can be matched to the documents it saw.

Permissions

If users may only see some documents, filter by permission in the search itself, before ranking, using metadata stored with each chunk. Never retrieve everything and ask the model to ignore what the user may not see: whatever is in the prompt can end up in the answer.

When not to use RAG

If all your documents fit comfortably in the model's context window, try putting them all in the prompt, with prompt caching to keep the cost down ( Prompt caching and cost control ), before building retrieval. Compare both on your eval set. Retrieval wins when the collection is large, changes often or needs per-user permissions.

Final checklist

• Every chunk has a stable id, its source and its heading prefix.
• Hybrid search (BM25 plus vectors, fused with RRF) and a reranker on the top candidates.
• Answers cite chunk ids; code rejects any id that was not retrieved.
• Refusal when evidence is weak, tested with questions the documents cannot answer.
• An eval set of 30 or more questions that reports recall@k, citation accuracy, answer correctness and refusal accuracy separately.
• Traces for every request, and re-ingestion that handles changed and deleted documents.

Trade-offs

Common mistakes

• Writing no eval set, so every change to chunking, search or prompts is judged by trying a few questions by hand.
• Scoring only the final answer, so you cannot tell whether retrieval or generation failed.
• Trusting the model's citations without checking that each cited id was actually retrieved.
• Chunking by a fixed number of characters across headings and tables, which separates facts from their context.
• Forgetting deleted documents in the index, so the app confidently cites a retired policy.

In the wild

Check yourself

Recap