Capstone: Build a Cited RAG Assistant

Project brief

Your company's engineers lose hours searching a large body of technical documentation. You are asked to build an assistant that answers questions from those documents, cites the exact passages it used, and says "I don't know" when the documents do not contain the answer. Leadership will only approve rollout if you can show measured retrieval quality and answer quality on a fixed evaluation set, not a demo of five hand-picked questions.

This is retrieval-augmented generation (RAG): retrieve relevant passages, put them in the prompt, and have a language model answer from them. The capstone uses the Python 3 documentation as a stand-in corpus because it is large, technical, freely downloadable and licensed for reuse. Everything you build transfers to an internal wiki, a policy library or a product manual.

Learning goals

• Build an ingestion pipeline that cleans documents, splits them into chunks with stable IDs and keeps source metadata.
• Index chunks with an open embedding model (sentence-transformers) in FAISS, and implement BM25 keyword search.
• Fuse dense and keyword results with reciprocal rank fusion (RRF) and measure whether fusion helps.
• Generate grounded answers with numbered citations through an OpenAI-compatible endpoint, served locally by Ollama or by a hosted provider.
• Write a 30-question golden evaluation set and report retrieval metrics (Recall@k, MRR), answer correctness, citation validity and abstention accuracy.

Required background

Review embeddings-and-rag, vector-databases-and-indexing, document-processing-and-chunking, advanced-retrieval and evaluating-rag. For the generation step, apis-tokens-and-pricing and local-runtimes-and-formats help.

Setup

Environment. Python 3.10+ with numpy, sentence-transformers, faiss-cpu and openai (the client library, used here only as a generic HTTP client for any OpenAI-compatible server). BM25 and RRF are implemented below in plain Python so you can see every step; pgvector in PostgreSQL is an acceptable alternative to FAISS if you already run Postgres.

Hardware. The embedding model below runs acceptably on a laptop CPU for a corpus of a few thousand chunks; a GPU or a free hosted notebook GPU makes indexing several times faster. For generation, a local 3B to 8B model quantized to about 4 bits needs roughly 3 to 7 GB of memory by the site's AI Index method; a hosted API needs no local hardware but sends your documents to a third party.

Generator. As of November 2026, Ollama's documentation describes an OpenAI-compatible API at http://localhost:11434/v1; the client library requires an API key value but a local Ollama server ignores it. Any model you have pulled works; the code reads the model name from an environment variable. Hosted providers with OpenAI-compatible endpoints work by changing the base URL and supplying a real key from the environment, never hard-coded.

Corpus. The Python 3 documentation, downloadable as an archive of plain-text files from https://docs.python.org/3/download.html. It is distributed under the terms on https://docs.python.org/3/license.html (the PSF licence, with code examples also under a permissive licence); read those terms and keep the attribution in your report. If you use your own documents instead, confirm you have the right to process them, and with a hosted generator, the right to send them to that provider.

Milestone 1: Ingestion and chunking

A chunk is the unit you retrieve and cite. Chunks that are too large dilute the embedding and waste prompt space; chunks that are too small lose the context needed to answer. Start with about 150 words and a 30-word overlap, keep the source file and a stable chunk ID, and inspect the results.

import pathlib, re, json

def chunk_text(doc_id, text, max_words=150, overlap=30):
    """Sliding word window; returns chunks with stable IDs."""
    words, chunks, start = text.split(), [], 0
    step = max_words - overlap
    while start < len(words):
        piece = " ".join(words[start:start + max_words])
        chunks.append({"id": f"{doc_id}#{len(chunks)}", "doc": doc_id, "text": piece})
        if start + max_words >= len(words):
            break
        start += step
    return chunks

def ingest(folder):
    chunks = []
    for path in sorted(pathlib.Path(folder).rglob("*.txt")):
        text = path.read_text(encoding="utf-8", errors="replace")
        text = re.sub(r"[=\-*~^]{4,}", " ", text)   # drop reST underline rules
        doc_id = str(path.relative_to(folder))
        chunks += chunk_text(doc_id, text)
    return chunks

chunks = ingest("python-docs-text")
json.dump(chunks, open("chunks.json", "w"))
print(len(chunks), chunks[100]["id"], chunks[100]["text"][:200])

Why 150 words? The embedding model used next, sentence-transformers/all-MiniLM-L6-v2, truncates input at 256 word-piece tokens, and English prose averages roughly 1.3 tokens per word, so a 200-word chunk would often be silently cut off. Check this by tokenizing a sample of chunks with the model's tokenizer. A stronger design splits on section headings first and only then by length, so that a chunk never straddles two topics; implement that as a stretch goal and compare.

Milestone 2: Dense and keyword retrieval

Dense retrieval embeds query and chunks into vectors and ranks by cosine similarity. With unit-normalized vectors, cosine similarity equals the inner product, so FAISS's exact inner-product index is correct and simple.

import numpy as np, faiss
from sentence_transformers import SentenceTransformer

emb = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
X = emb.encode([c["text"] for c in chunks], batch_size=64,
               normalize_embeddings=True, show_progress_bar=True).astype("float32")
index = faiss.IndexFlatIP(X.shape[1])      # exact search; 384 dimensions here
index.add(X)
faiss.write_index(index, "chunks.faiss")

def dense_search(query, k=20):
    q = emb.encode([query], normalize_embeddings=True).astype("float32")
    scores, ids = index.search(q, k)
    return [chunks[i]["id"] for i in ids[0] if i != -1]

BM25 is the classic keyword ranking function. For query terms $q_1,\dots,q_n$ and a chunk $d$:

Here $f(q_i, d)$ is how often term $q_i$ occurs in $d$, $|d|$ is the chunk length in tokens, avgdl is the average chunk length, $N$ is the number of chunks, $n(q_i)$ is the number of chunks containing $q_i$, and $k_1 \approx 1.2$ to $2$ and $b = 0.75$ are the usual constants. Term frequency saturates (the tenth occurrence adds little), rare terms count more, and long chunks are penalized.

import math
from collections import Counter

TOKEN = re.compile(r"[a-z0-9_]+")
def tokenize(s): return TOKEN.findall(s.lower())

class BM25:
    def __init__(self, texts, k1=1.5, b=0.75):
        self.k1, self.b = k1, b
        self.tf = [Counter(tokenize(t)) for t in texts]
        self.len = [sum(c.values()) for c in self.tf]
        self.N, self.avgdl = len(texts), sum(self.len) / len(texts)
        df = Counter(term for c in self.tf for term in c)
        self.idf = {t: math.log(1 + (self.N - n + 0.5) / (n + 0.5)) for t, n in df.items()}

    def search(self, query, k=20):
        q = tokenize(query)
        scores = []
        for i, (tf, dl) in enumerate(zip(self.tf, self.len)):
            norm = self.k1 * (1 - self.b + self.b * dl / self.avgdl)
            s = sum(self.idf[t] * tf[t] * (self.k1 + 1) / (tf[t] + norm) for t in q if t in tf)
            scores.append((s, i))
        return [chunks[i]["id"] for s, i in sorted(scores, reverse=True)[:k] if s > 0]

bm25 = BM25([c["text"] for c in chunks])

This loops over every chunk per query, which is fine for tens of thousands of chunks and a learning project; production keyword search uses an inverted index (for example in a search engine or PostgreSQL full-text search). Note the tokenizer keeps underscores, so identifiers like __init__ and dict_keys survive as single tokens (a dotted name such as os.path becomes os and path), which matters for technical documents. It does no stemming, so "dogs" will not match "dog"; dense retrieval covers that gap.

Milestone 3: Hybrid retrieval with reciprocal rank fusion

Dense retrieval is good at paraphrase ("make a copy of a list" finds copy.deepcopy); BM25 is good at exact identifiers, error messages and version numbers. Their scores live on different scales, so instead of adding scores, reciprocal rank fusion adds a function of each result's rank:

where $R$ is the set of rankings, $\text{rank}_r(d)$ is the 1-based position of $d$ in ranking $r$ (a ranking that does not contain $d$ contributes nothing) and $k = 60$ is the constant used by Cormack, Clarke and Büttcher (2009).

Worked example. Dense returns [A, B, C]; BM25 returns [C, A, D].

• A: $1/(60+1) + 1/(60+2) = 0.016393 + 0.016129 = 0.032522$
• C: $1/(60+3) + 1/(60+1) = 0.015873 + 0.016393 = 0.032266$
• B: $1/(60+2) = 0.016129$
• D: $1/(60+3) = 0.015873$

Fused order: A, C, B, D. Documents that both retrievers like rise to the top, and the large $k$ keeps a single first place from dominating.

def rrf(rankings, k=60, top_n=6):
    score = Counter()
    for ranked in rankings:
        for rank, doc_id in enumerate(ranked, start=1):
            score[doc_id] += 1.0 / (k + rank)
    return [d for d, _ in score.most_common(top_n)]

by_id = {c["id"]: c for c in chunks}
def hybrid_search(query, k=6):
    return [by_id[i] for i in rrf([dense_search(query), bm25.search(query)], top_n=k)]

Milestone 4: Grounded generation with citations

Number the retrieved chunks, instruct the model to answer only from them and to cite by number, and require an explicit refusal string when the context is insufficient. Then validate the citations in code: every cited number must exist, and an answer with no citations is treated as a failure unless it is the refusal.

import os
from openai import OpenAI

client = OpenAI(base_url=os.environ.get("LLM_BASE_URL", "http://localhost:11434/v1"),
                api_key=os.environ.get("LLM_API_KEY", "unused-for-local"))
MODEL = os.environ["LLM_MODEL"]          # e.g. a model you have pulled in Ollama
REFUSAL = "I can't find this in the documentation."

SYSTEM = ("Answer the question using ONLY the numbered sources. Cite every claim "
          "with its source number in square brackets, like [2]. If the sources do "
          f"not contain the answer, reply exactly: {REFUSAL}")

def answer(question, k=6):
    hits = hybrid_search(question, k)
    context = "\n\n".join(f"[{n}] ({h['id']})\n{h['text']}" for n, h in enumerate(hits, 1))
    resp = client.chat.completions.create(model=MODEL, temperature=0, messages=[
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": f"Sources:\n{context}\n\nQuestion: {question}"}])
    text = resp.choices[0].message.content.strip()
    cited = sorted({int(n) for n in re.findall(r"\[(\d+)\]", text)})
    valid = [n for n in cited if 1 <= n <= len(hits)]
    return {"answer": text, "citations": [hits[n - 1]["id"] for n in valid],
            "invalid_citations": [n for n in cited if n not in valid],
            "refused": text.startswith(REFUSAL), "retrieved": [h["id"] for h in hits]}

Temperature 0 makes evaluation repeatable (not perfectly deterministic on every server, so run the evaluation twice and report variation). Putting the sources before the question and keeping $k$ small reduces the chance that the model ignores passages in the middle of a long prompt.

Milestone 5: The golden evaluation set and metrics

Write 30 questions by hand, in a JSONL file, before you tune anything. Mix types so the set exercises the whole system: about 15 single-fact questions, 6 questions whose answer combines two passages, 4 questions that require an exact identifier (a function name or error message), and 5 questions whose answer is not in the corpus. For each, record the reference answer and the IDs of the chunks that contain the evidence (the evaluation code compares chunk IDs).

{"id": "q07", "type": "fact", "question": "Which exception does dict.pop raise when the key is missing and no default is given?", "answer": "KeyError", "relevant": ["library/stdtypes.txt#412"]}
{"id": "q28", "type": "unanswerable", "question": "What is the default port of the Python package index mirror at my company?", "answer": null, "relevant": []}

(The chunk ID shown is illustrative; use the IDs your own ingestion produces, and re-check them whenever you change chunking.)

Retrieval metrics over the answerable questions, for a ranked list $L$ and relevant set $G$:

with the reciprocal rank taken as 0 when nothing relevant is retrieved. For example, if the retrieved list is [x, a, y, b] and $G = \{a, b\}$, Recall@3 is $1/2$ and the reciprocal rank is $1/2$.

def recall_at_k(retrieved, relevant, k):
    return len(set(retrieved[:k]) & set(relevant)) / len(relevant)

def reciprocal_rank(retrieved, relevant):
    return next((1 / i for i, d in enumerate(retrieved, 1) if d in relevant), 0.0)

def evaluate(gold, search_fn, k=5):
    answerable = [g for g in gold if g["relevant"]]
    rec = sum(recall_at_k(search_fn(g["question"]), g["relevant"], k) for g in answerable)
    mrr = sum(reciprocal_rank(search_fn(g["question"]), g["relevant"]) for g in answerable)
    return {"recall@k": rec / len(answerable), "mrr": mrr / len(answerable)}

ids = lambda f: (lambda q: [x if isinstance(x, str) else x["id"] for x in f(q)])
for name, fn in [("dense", dense_search), ("bm25", bm25.search), ("hybrid", hybrid_search)]:
    print(name, evaluate(gold, ids(fn)))

Dense and BM25 return 20 results here while hybrid returns 6, so MRR is computed over lists of different depths; truncate every list to the same depth (for example 10) when you compare retrievers.

Answer metrics. Grade each generated answer as correct, partially correct or incorrect against the reference; you may use a strong model as a judge with a written rubric, but hand-check at least 10 of its verdicts and report the agreement rate. Also report: citation validity (share of cited numbers that exist), citation support (share of answers whose cited passages actually contain the claim, checked by hand), abstention accuracy (unanswerable questions refused, and answerable questions not refused), and latency at the median and 95th percentile.

Evaluation criteria and target metrics

These are course thresholds for a 30-question set, not published benchmarks. With only 30 questions, one question is 3.3 percentage points, so treat differences of a few points as noise and say so.

Grading rubric

Stretch goals

• Add a cross-encoder re-ranker over the top 20 fused results and measure the change in MRR and latency.
• Replace word-window chunking with heading-aware chunking and compare Recall@5.
• Store vectors in PostgreSQL with pgvector and filter by document section with metadata.
• Add query rewriting for multi-part questions and evaluate on the multi-passage subset only.
• Track cost and latency per answer for a local and a hosted generator and recommend one.

Write-up template

1. Use case and users. Who asks what, and what a wrong answer costs.
2. Corpus. Source, licence, size (documents, chunks, tokens), cleaning steps.
3. System. Diagram in words: ingestion, chunking parameters, embedding model, index type, BM25 settings, fusion, prompt, generator and where it runs.
4. Evaluation set. How questions were written, the type mix, and who verified the references.
5. Results. Retrieval table (dense, BM25, hybrid), answer and citation metrics, latency; note run-to-run variation.
6. Failure analysis. Five failures, each traced to a stage: chunking, retrieval, ranking, generation or the question itself.
7. Risks and next steps. Data exposure, stale documents, prompt injection inside documents, and the next improvement you would fund.

Common pitfalls

• Tuning on the evaluation questions until they pass, then reporting those numbers as if they were unseen.
• Silently truncated chunks that are longer than the embedding model's limit.
• Forgetting to normalize embeddings before inner-product search, which turns cosine similarity into a length-biased score.
• Trusting citations without checking that the cited passage supports the claim.
• Treating text inside retrieved documents as trusted instructions; a document can contain a prompt-injection attack.

In practice

Production RAG teams spend most of their time on ingestion quality and evaluation rather than on the model. A typical baseline is exactly this design: structure-aware chunking, a hybrid of dense and keyword retrieval fused by rank, an optional re-ranker, and a prompt that demands citations and allows refusal. The golden set grows from real user questions and failure reports, every change is evaluated against it before release, and citations are rendered as links so users can verify answers themselves.