Part II · Chapter 3

Build a RAG System From Scratch

RAG (Retrieval-Augmented Generation) grounds the model in your data so it answers from facts, not guesses. You'll build the entire pipeline by hand — chunking, embedding, a tiny vector store, hybrid retrieval, re-ranking, and cited generation — so you understand every moving part before reaching for a framework.

⏱️ ~2 hours🧪 5 labs🎯 Intermediate📦 numpy, rank-bm25

⚙️ To run this for realThe code here is complete and correct as written. To actually execute it you'll need:

an Anthropic API key (ANTHROPIC_API_KEY) + pip install anthropic

Reading and learning works without any of this — run when you're ready. IDs/ARNs in examples are placeholders; swap in your own.

Learning objectives

Explain when RAG beats fine-tuning or a bigger context window.
Chunk documents on semantic boundaries with useful metadata.
Embed text and build a minimal in-memory vector store with cosine similarity.
Retrieve with hybrid (vector + keyword) search and re-rank the results.
Generate grounded answers with inline citations, and refuse when context is missing.

What RAG is, and when to use it intermediate

An LLM only knows what was in its training data. RAG lets it answer questions about your private/current documents by retrieving relevant snippets at query time and putting them in the prompt. The model then answers from that supplied context.

Approach	Best when	Trade-off
RAG	Knowledge changes often; needs citations; large corpus	Retrieval quality is the bottleneck
Long context (paste it all)	Small, stable doc set that fits the window	Expensive per call; doesn't scale to millions of docs
Fine-tuning	Teaching style/behavior, not facts	Costly to update; can't cite; facts go stale

The truth about "hallucination" in RAGMost wrong answers in a RAG system are retrieval failures, not generation failures — the model answered correctly from the wrong (or missing) context. That's why we test retrieval separately and first.

The two-phase pipeline intermediate

🗺️ How to read this diagram

This is the whole RAG system on one page. Read it as two separate timelines: the top row happens once, ahead of time (building a searchable index of your documents), and the bottom row happens every time a user asks a question. The dashed line between them is that split.

Top row — OFFLINE, build the index once. Documents → Chunk + metadata (cut each doc into small passages) → Embed (turn each passage into a list of numbers that captures its meaning) → Vector store (save all those number-lists so you can search them). You do this once and reuse it.
Bottom row — ONLINE, per user question. A Question comes in and flows left to right through Retrieve → Re-rank → Build context → LLM → Answer.
Retrieve (vec + BM25) finds the handful of stored chunks most related to the question — using both meaning-based (vector) and exact-word (BM25) search. Re-rank reorders those candidates so the best ones rise to the top.
Build context pastes the winning chunks into the prompt; the LLM reads that prompt and writes an Answer + citations, using only the supplied text. This is the core RAG idea: retrieve, then augment the prompt, then generate.
The purple dashed arrow from Vector store down to Retrieve shows the two timelines meeting: the index you built offline is exactly what the online question searches against.

In short: RAG = look things up first, then answer from what you found. The offline row is a librarian shelving books; the online row is you asking the librarian a question and getting the right pages handed to you before you write your reply.

Sample corpusCreate a folder docs/ with 3–4 small .md or .txt files — a product FAQ, a policy doc, a how-to. Real content you know well makes it obvious when retrieval works or fails.

Lab 3.1 · Chunking intermediate

Models retrieve chunks, not whole documents. Good chunking is the highest-leverage decision in RAG. Too big → irrelevant text dilutes the answer and wastes tokens. Too small → facts get split across chunks.

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.

Lab 3.1

chunk.pyimport os, glob, re

def chunk_text(text, source, target_words=120, overlap=25):
    """Split on paragraphs, then pack into ~target_words chunks with overlap."""
    paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
    chunks, buf = [], []
    for para in paras:
        buf.append(para)
        if sum(len(p.split()) for p in buf) >= target_words:
            chunks.append(" ".join(buf))
            buf = buf[-1:] if overlap else []   # carry last para as overlap
    if buf: chunks.append(" ".join(buf))
    return [{"text": c, "source": source, "id": f"{source}#{i}"}
            for i, c in enumerate(chunks)]

def load_corpus(folder="docs"):
    out = []
    for path in glob.glob(f"{folder}/*"):
        with open(path) as f:
            out += chunk_text(f.read(), os.path.basename(path))
    return out

if __name__ == "__main__":
    c = load_corpus()
    print(f"{len(c)} chunks"); print(c[0])

▶ How this works

Models search over chunks (small passages), not whole documents. This file cuts each document into paragraph-based chunks of roughly a target size, and keeps a little overlap so a fact sitting near a cut isn't lost. Getting this right is the single highest-leverage decision in RAG.

chunk_text first splits the document on blank lines into paragraphs (paras). It then walks paragraph by paragraph, adding each to a buffer buf.
When the buffer reaches target_words (~120 words), it flushes: it joins the buffered paragraphs into one chunk and starts a new buffer. The line marked carry last para as overlap keeps the last paragraph in the new buffer so the next chunk overlaps the previous one — that overlap protects facts near a boundary.
Each chunk is stored as a small dictionary with text, its source filename, and a unique id like faq.md#0. That metadata is what lets you cite sources later.
load_corpus loops over every file in the docs/ folder, reads it, and collects all chunks into one flat list — your whole searchable corpus.

What the output means: Running it prints how many chunks were produced (e.g. 7 chunks) and shows the first chunk dictionary, so you can eyeball whether the splitting looks sensible.

Try this: Set target_words=40 and re-run: you'll get many more, smaller chunks. Then set overlap=0 and notice the chunks no longer share a paragraph. Smaller chunks are more precise but split facts more often — that trade-off is the heart of chunking.

Chunking principles

Split on semantic boundaries (paragraphs, headings) — not blind fixed windows.
Keep 10–20% overlap so a fact near a boundary survives in one chunk.
Carry metadata (source, section, URL, date) — you need it for citations and filtering.
Aim ~100–250 words for precision; add a "parent document" fallback for context.

Lab 3.2 · Embed & store intermediate

An embedding turns text into a vector of numbers where similar meanings sit close together. We store each chunk's vector; at query time we embed the question and find the nearest chunks by cosine similarity.

Embedding model noteUse a dedicated embeddings model/endpoint (e.g. from your provider or an open one like sentence-transformers). The rule that matters: embed queries and documents with the same model, or similarity is meaningless. Below we wrap it behind one embed() function so you can swap providers freely.

Lab 3.2

store.pyimport numpy as np

# --- swap this body for your embeddings provider; keep the signature ---
from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
def embed(texts: list[str]) -> np.ndarray:
    return np.asarray(_model.encode(texts, normalize_embeddings=True))
# -----------------------------------------------------------------------

class VectorStore:
    def __init__(self):
        self.vecs = None; self.chunks = []
    def add(self, chunks):
        self.chunks = chunks
        self.vecs = embed([c["text"] for c in chunks])   # (N, d), normalized
    def search(self, query, k=5):
        q = embed([query])[0]                        # (d,)
        sims = self.vecs @ q                        # cosine (vectors are normalized)
        top = np.argsort(-sims)[:k]
        return [(self.chunks[i], float(sims[i])) for i in top]

▶ How this works

This is the tiny vector store — the searchable index at the heart of RAG. An embedding turns a piece of text into a fixed list of numbers (a vector) where texts with similar meaning get similar numbers. Store one vector per chunk, and at question time you can find the closest chunks by comparing vectors.

The embed() function is the one place text becomes numbers. It's wrapped behind a single function on purpose (the comment says swap this body for your embeddings provider) so you can change providers without touching anything else. The golden rule: embed questions and documents with the same model, or the comparison is meaningless.
normalize_embeddings=True scales every vector to length 1. That's a shortcut that makes cosine similarity (an angle comparison) collapse into a plain dot product later — simpler and faster.
VectorStore.add(chunks) embeds all chunk texts at once and keeps the resulting matrix self.vecs (N chunks × d numbers each) alongside the chunks themselves.
search(query, k) embeds the query, then self.vecs @ q computes the similarity of the query to every stored chunk in one step. np.argsort(-sims)[:k] picks the indices of the top-k highest scores and returns those chunks with their scores.

What the output means: Nothing prints on its own — this file defines the store. Once populated, search() hands back the k most similar chunks and a similarity score (higher = more related) for each.

Try this: This is genuinely what a production vector database (Pinecone, FAISS, pgvector) does under the hood; they just add persistence, scale, and speed. Print self.vecs.shape after add() to see the (number-of-chunks × dimensions) matrix you built.

This IS a vector databaseA production vector DB (Pinecone, Weaviate, pgvector, FAISS…) adds persistence, scale, filtering, and approximate-nearest-neighbor speed. But the core operation is exactly what you just wrote: cosine similarity between a query vector and stored vectors. Now you know what they do under the hood.

Lab 3.3 · Retrieve advanced

Retrieval = nearest vectors by cosine similarity. The query becomes a vector; each chunk's score is the cosine of the angle between them (a dot product on normalized vectors). Take the top-k with a heap (D4) — that's the context you feed the model.

🗺️ How to read this diagram

This picture explains how retrieval actually decides which chunks are "closest". Every chunk and the query are arrows (vectors) pointing out from the same origin; the angle between two arrows measures how related their meanings are.

On the left, the blue query arrow and the chunk arrows all start at the same corner. A small angle between the query and a chunk means they point the same way — i.e. similar meaning. chunk A hugs the query closely, so it's the best match (✓).
Cosine similarity is just the cosine of that angle: 1.0 for arrows pointing the exact same way, down toward 0 as they diverge. Because the vectors were normalized in Lab 3.2, this cosine is simply the dot product q·c.
On the right, the same chunks are shown as a ranked list with their scores — A · 0.91, C · 0.78, B · 0.55. Higher score = smaller angle = more relevant.
top-k means: keep only the highest-scoring few (here using a heap, from the data-structures chapter). Those winning chunks are exactly the context you paste into the prompt in Lab 3.4.

In short: "Nearest by cosine" is the entire retrieval trick: turn text into arrows, then keep the arrows pointing most like the question. Everything else (hybrid, re-ranking) just improves which arrows win.

Lab 3.3

retrieve.pyfrom chunk import load_corpus
from store import VectorStore

store = VectorStore()
store.add(load_corpus())

for chunk, score in store.search("how do I reset my password?", k=3):
    print(f"[{score:.3f}] {chunk['source']}: {chunk['text'][:80]}...")

[0.612] faq.md: To reset your password, go to Settings > Security and click...
[0.341] faq.md: Two-factor authentication can be enabled from the same...
[0.208] policy.md: Account security is governed by...

▶ How this works

This is the payoff of Labs 3.1–3.2: give the store a real question and see which chunks come back. This is the Retrieve step of RAG, running on its own — before any answer-generation exists.

It builds a VectorStore, fills it with load_corpus() (your chunked docs), and then calls store.search(...) with a plain-English question and k=3 (return the 3 closest chunks).
The loop prints each result as [score] source: first-80-characters…. The score is the cosine similarity — closer to 1.0 means the chunk's meaning is nearer to the question.
Notice the top result is the actual password-reset passage from faq.md — retrieval found the right text purely from meaning, even though the question wording differs from the document.

What the output means: Three lines, best first. The top line ([0.612] faq.md: To reset your password…) is the chunk that answers the question; scores fall off for the less-relevant chunks below it.

Try this: Before writing any generation code, run 10 real questions through search() and check the right chunk lands in the top-k. If it doesn't, fix chunking or embedding now — no clever prompt can rescue context that was never retrieved.

Test retrieval alone, firstBefore writing a single line of generation code, run 10 real questions through search() and eyeball whether the right chunk appears in the top-k. If it doesn't, fix chunking/embedding now — no prompt can rescue missing context.

Lab 3.4 · Generate a grounded, cited answer advanced

Now assemble the retrieved chunks into a prompt that forces the model to answer only from context and cite sources.

Lab 3.4

rag.pyfrom anthropic import Anthropic
from store import VectorStore
from chunk import load_corpus
client = Anthropic()

store = VectorStore(); store.add(load_corpus())

SYSTEM = (
    "You answer strictly from the numbered context. "
    "Cite the sources you use inline as [n]. "
    "If the context does not contain the answer, say: "
    "'I don't have that information.' Never use outside knowledge."
)

def answer(question, k=4):
    hits = store.search(question, k=k)
    context = "\n\n".join(
        f"[{i}] (source: {c['source']})\n{c['text']}"
        for i, (c, _) in enumerate(hits, 1)
    )
    resp = client.messages.create(
        model="claude-opus-4-8", max_tokens=600, system=SYSTEM,
        messages=[{"role":"user",
                   "content": f"Context:\n{context}\n\nQuestion: {question}"}],
    )
    text = next(b.text for b in resp.content if b.type=="text")
    return text, hits

ans, sources = answer("How do I reset my password?")
print(ans)

Go to Settings > Security and click "Reset password"; a link is emailed
to your registered address [1]. You can also enable 2FA there [1].

▶ How this works

Here the pieces join into full RAG: retrieve → augment the prompt → generate. The retrieved chunks are pasted into the prompt as numbered context, and the model is told to answer only from that context and cite it. This grounding is what stops the model from making things up.

The SYSTEM prompt is the guardrail. It orders the model to answer strictly from the numbered context, cite sources inline as [n], and — critically — say "I don't have that information." when the answer isn't there instead of inventing one.
answer() first calls store.search(question, k) to get the top chunks (the retrieve step).
It then builds context by numbering each chunk [1] (source: …) and joining them (the augment step) — the numbers are what the model cites.
client.messages.create(...) sends the system rules plus a user message containing the context and the question (the generate step). The final line pulls the text out of the reply's content blocks and returns it along with the source chunks.

What the output means: A short answer that quotes only the retrieved facts and tags them with citations, e.g. Go to Settings > Security… [1]. The [1] points back to the first numbered chunk you fed in.

Try this: Ask something that is definitely NOT in your docs/ and confirm it replies "I don't have that information." Those two prompt rules — answer only from context, and admit when it's missing — are the whole anti-hallucination recipe.

The two rules that prevent hallucination

"Answer only from context" — constrains the model to supplied facts.
"Say you don't know if it's missing" — gives it an honest escape hatch instead of inventing. Test this: ask something NOT in your docs and confirm it declines.

Lab 3.5 · Hybrid search + re-ranking advanced

Vector search understands meaning but misses exact terms (error codes, product names, acronyms). Keyword search (BM25) nails exact terms but misses paraphrase. Combine both, then re-rank. This is the biggest quality jump you'll make.

Lab 3.5

Add BM25 keyword search.

hybrid.pyfrom rank_bm25 import BM25Okapi

class Hybrid:
    def __init__(self, store):
        self.store = store
        self.bm25 = BM25Okapi([c["text"].lower().split() for c in store.chunks])
    def search(self, query, k=10):
        # dense results (id -> rank)
        dense = [c["id"] for c,_ in self.store.search(query, k=k)]
        # sparse results
        scores = self.bm25.get_scores(query.lower().split())
        sparse = [self.store.chunks[i]["id"]
                  for i in sorted(range(len(scores)), key=lambda j:-scores[j])[:k]]
        return self._rrf(dense, sparse)

Fuse with Reciprocal Rank Fusion — a simple, robust way to merge two ranked lists.

hybrid.py (cont.)    def _rrf(self, *lists, k=60):
        scores = {}
        for lst in lists:
            for rank, cid in enumerate(lst):
                scores[cid] = scores.get(cid, 0) + 1/(k+rank)
        by_id = {c["id"]: c for c in self.store.chunks}
        ranked = sorted(scores, key=lambda c:-scores[c])
        return [by_id[cid] for cid in ranked]

Re-rank the top candidates for the final cut. Over-fetch ~10, then let a strong model pick the best 3–4. This is the single highest-leverage quality lever in RAG.

rerank.pyimport re
from anthropic import Anthropic
client = Anthropic()  # set ANTHROPIC_API_KEY

def rerank(question, candidates, keep=4):
    """LLM re-ranker: score each candidate's relevance 0-10, keep the best."""
    listing = "\n".join(f"[{i}] {c['text'][:200]}"
                        for i,c in enumerate(candidates))
    r = client.messages.create(
        model="claude-haiku-4-5", max_tokens=200,
        messages=[{"role":"user","content":
            f"Question: {question}\n\nPassages:\n{listing}\n\n"
            "Return the indices of the {keep} most relevant, comma-separated."}],
    )
    idxs = [int(x) for x in re.findall(r"\d+",
             next(b.text for b in r.content if b.type=="text"))][:keep]
    return [candidates[i] for i in idxs if i < len(candidates)]

▶ How this works

Pure meaning-based (vector) search misses exact terms like error codes and product names; pure keyword search (BM25) misses paraphrases. This lab runs both, merges their rankings, then uses a model to pick the very best few. This is usually the biggest single quality jump you can make in a RAG system.

Step 1 — Hybrid.search. It gets dense results (your vector store.search, good at meaning) and sparse results (BM25Okapi, good at exact words) — two ranked lists of chunk ids for the same query.
Step 2 — _rrf (Reciprocal Rank Fusion). A simple, robust way to merge ranked lists: each list gives a chunk points based on its position (1/(k+rank)), the points are summed per chunk, and chunks are re-sorted by total. Something ranked highly by either method rises — you don't have to tune weights.
Step 3 — rerank. After over-fetching ~10 candidates cheaply, a fast model reads each passage against the question and returns the indices of the best few to keep. This precise final cut is the highest-leverage lever in RAG.
The overall pattern is a funnel: fetch wide and cheap (hybrid, top-10–30), then narrow precisely (re-rank to top-3–5), then generate — giving you both recall and precision without spending tokens on junk.

What the output means: Hybrid.search returns chunks ordered by fused score; rerank trims them to the few most relevant passages, which become the context for the generation step from Lab 3.4.

Try this: Search for an exact term that appears verbatim in one doc (an ID or code) and compare hybrid vs. the pure-vector store.search. The keyword half of hybrid is what reliably surfaces exact matches that vectors alone can blur.

The retrieval funnelOver-fetch cheaply (hybrid, top-10–30) → re-rank precisely (top-3–5) → generate. Cheap-and-wide then expensive-and-narrow gives you both recall and precision without blowing the token budget.

Measuring your RAG system expert

You can't improve what you don't measure. Track retrieval and generation separately:

Stage	Metric	Question it answers
Retrieval	Recall@k	Did the answer-bearing chunk make the top-k?
Retrieval	Context precision	How much of what we retrieved was actually relevant?
Generation	Groundedness	Is every claim supported by the retrieved context?
Generation	Answer relevance	Did it actually address the question?

In Chapter 5 you'll build an automated harness for exactly these metrics. For now, keep a spreadsheet of 15–20 questions with the chunk you expect to be retrieved, and check Recall@k by hand after any change.

Common pitfalls expert

Symptom	Real cause	Fix
"It makes things up"	Answer chunk never retrieved	Fix chunking/hybrid search; test retrieval alone
Answers are vague	Chunks too large, signal diluted	Smaller chunks + re-ranking
Misses exact IDs/codes	Pure vector search	Add BM25 (hybrid)
Leaks other tenants' data	No metadata filter	Filter by ACL/tenant before semantic search
Query & docs embedded differently	Two embedding models	One `embed()` function everywhere

Exercises expert

Exercise 3.1 — Refusal test

Context: A grounded assistant's most important behaviour is knowing when to say nothing. The fastest way to catch hallucination is to ask something you know is absent from the corpus and watch what happens.

Your task: Probe your Lab 3.4 system with a question that is definitely not covered by anything in docs/, and confirm it refuses instead of inventing an answer.

Requirements:

Pick a question with no support in docs/ at all
A correct run replies with the refusal (e.g. "I don't have that information"), not a guess
If it invents an answer, strengthen the system prompt and re-test
Try two or three phrasings — refusal shouldn't depend on exact wording

💡 Hint: Adversarial phrasings that sound in-domain but aren't (a plausible but nonexistent feature) are the ones that expose a weak refusal instruction.

Exercise 3.2 — Metadata filter

Context: The same retriever that powers search is a data-leak vector the moment you have more than one customer. Scoping results to the current tenant is the canonical multi-tenant RAG safeguard.

Your task: Add a tenant field to each chunk and a filter in search() so only the current tenant's chunks are considered, then prove a user can't reach another tenant's docs.

Requirements:

Every chunk carries a tenant value
search() restricts candidates to the current tenant before ranking
A query as tenant A returns zero tenant-B chunks — demonstrate it
The restriction is enforced in code, not asked for in the prompt

💡 Hint: Filter before similarity, or mask out-of-tenant rows so they can never enter the top-k — don't rank everything and trim afterward.

Show hint

Filter the candidate list before computing similarity, or mask out-of-tenant rows by setting their similarity to -inf. Never rely on the model to respect permissions — enforce it in code.

Exercise 3.3 — Measure Recall@k

Context: "It feels better" is not an evaluation. A tiny labelled set turns retrieval tuning into a number you can move, and lets you settle dense-vs-hybrid with evidence instead of vibes.

Your task: Build 10 (question, expected_chunk_id) pairs and a loop that measures how often the expected chunk lands in the top-5, then compare pure-vector retrieval against hybrid on your own corpus.

Requirements:

At least 10 labelled pairs drawn from your actual docs/
Compute Recall@5 = fraction of questions whose expected chunk is in the top-5
Run the identical question set through both retrievers
Report both numbers and state which wins on your corpus
Note one query where they disagree and why

💡 Hint: Reuse the recall_at_k idea from the ladder's Professional rung; keep the pair list in a small list of tuples so re-running after a tweak is one call.

Show solution sketch

Illustrative fragment — defines demo values / files are needed before this runs standalone.

hits = 0
for q, expected in testset:
    ids = [c["id"] for c in hybrid.search(q, k=5)[:5]]
    hits += expected in ids
print("recall@5:", hits/len(testset))

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · Cosine similarity on normalized vectorsBeginner

Context: A retriever ranks chunks by how close their embedding is to the query's. When vectors are unit-length (the lesson normalizes them), that closeness is just a dot product — the cheapest possible similarity.

Your task: Write cosine(a, b) that returns the cosine similarity of two already-normalized vectors, then prove it behaves on a few hand-made vectors.

Requirements:

Use numpy; no explicit Python loop over the dimensions
Return a plain float, not a 0-d numpy array
Assert two identical (same-direction) vectors score ~1.0
Show that an orthogonal vector scores ~0.0 and an opposite one ~-1.0
Work for a 1-D vector of any length

💡 Hint: Once both inputs are unit-length, np.dot is the whole answer — no division needed. Use a helper that normalizes raw input first.

Show solution

With unit-length vectors, cos = a @ b. Runnable (needs numpy, used throughout the lesson):

import numpy as np

def unit(v):
    v = np.asarray(v, dtype=float)
    return v / np.linalg.norm(v)

a = unit([1.0, 2.0, 2.0])
b = unit([2.0, 4.0, 4.0])   # same direction
c = unit([2.0, 0.0, 0.0])
print(round(float(a @ b), 4))   # 1.0 -- identical direction
print(round(float(a @ c), 4))   # < 1.0 -- different direction

Exercise 2 · Chunk with overlap + carry metadataIntermediate

Context: Retrieval quality lives or dies on chunking: too big and you dilute the match, too small and you lose context. Overlap keeps ideas that straddle a boundary findable, and per-chunk metadata is what lets you cite a source later.

Your task: Write a chunker that splits text on paragraph boundaries into ~N-word chunks with a word overlap, attaching source and a unique id to each chunk. Run it on a short multi-paragraph string.

Requirements:

Target chunk size and overlap are parameters (e.g. 60 words, 15 overlap)
Never split inside a word; break on whitespace
Each chunk is a dict/dataclass with text, source, and a stable unique id
Consecutive chunks share overlap trailing words
Print the chunks and eyeball that the overlap is actually present

💡 Hint: Tokenize to a flat word list first, then slide a window of size N advancing by N−overlap each step. The id can be f"{source}#{index}".

Show solution

This is the lesson's chunk_text, self-contained and runnable (no files needed):

import re

def chunk_text(text, source, target_words=20, overlap=True):
    paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
    chunks, buf = [], []
    for para in paras:
        buf.append(para)
        if sum(len(p.split()) for p in buf) >= target_words:
            chunks.append(" ".join(buf))
            buf = buf[-1:] if overlap else []     # carry last para as overlap
    if buf:
        chunks.append(" ".join(buf))
    return [{"text": c, "source": source, "id": f"{source}#{i}"}
            for i, c in enumerate(chunks)]

doc = "Reset your password in Settings then Security.\n\n" \
      "Enable two-factor authentication in the same place.\n\n" \
      "Contact support if you are locked out for good."
out = chunk_text(doc, "faq.md")
print(len(out), "chunks")
for c in out:
    print(c["id"], "::", c["text"][:45])

Exercise 3 · A tiny VectorStore with top-k searchAdvanced

Context: Every RAG system needs a store that can ingest chunks and answer nearest-neighbour queries. Real ones call an embedding API, but you want yours to run offline and deterministically in tests — so the embedding step must be a swappable seam.

Your task: Implement a tiny VectorStore with add(chunks) (embeds & stores) and search(query, k) (returns the top-k chunks by cosine). Inject the embedder so a fake deterministic one is used in the demo.

Requirements:

embed() is passed in (constructor arg or attribute), not hard-coded
The fake embedder is deterministic — same text always yields the same vector
add() stores the vector alongside the chunk, embedding once
search() returns k chunks ordered best-first
Ties and k > len(store) don't crash
Demonstrate a query returning the intuitively-closest chunk

💡 Hint: A deterministic fake: hash the text into a seed, seed numpy, return a random unit vector. Keep vectors and chunks in parallel lists (or a list of pairs) and sort by the cosine you wrote in Exercise 1.

Show solution

The real lesson swaps embed() for a provider; here a toy hashing embedder keeps it offline and deterministic while the store logic is identical. Runnable:

import numpy as np

def embed(texts):                         # toy stand-in; same signature as real one
    vs = []
    for t in texts:
        v = np.zeros(16)
        for w in t.lower().split():
            v[hash(w) % 16] += 1.0
        n = np.linalg.norm(v) or 1.0
        vs.append(v / n)                  # normalize -> cosine is a dot product
    return np.asarray(vs)

class VectorStore:
    def __init__(self):
        self.vecs = None; self.chunks = []
    def add(self, chunks):
        self.chunks = chunks
        self.vecs = embed([c["text"] for c in chunks])
    def search(self, query, k=3):
        q = embed([query])[0]
        sims = self.vecs @ q
        top = np.argsort(-sims)[:k]
        return [(self.chunks[i], float(sims[i])) for i in top]

docs = [{"text":"reset your password in settings security", "id":"a"},
        {"text":"enable two factor authentication", "id":"b"},
        {"text":"billing invoices and refunds", "id":"c"}]
s = VectorStore(); s.add(docs)
for c, score in s.search("how do I reset my password", k=2):
    print(round(score, 3), c["id"])

Exercise 4 · Fuse two rankings with Reciprocal Rank FusionExpert

Context: Hybrid search runs a dense (vector) retriever and a sparse (keyword/BM25) retriever and fuses their results. Reciprocal Rank Fusion is the workhorse: it combines ranks, not raw scores, so the two systems' incompatible score scales never have to be reconciled.

Your task: Implement _rrf(rankings, k=60) that fuses two ranked lists of chunk ids into one, and show it surfaces a chunk that either retriever ranked highly — with no score tuning.

Requirements:

Score each id as Σ 1/(k + rank) across the lists it appears in
Use rank position (0- or 1-based, stated), never the original scores
An id present in only one list still gets its contribution
Return ids sorted by fused score, highest first
Construct an example where RRF's #1 was #2 in both lists — a consensus win

💡 Hint: Iterate each ranking with enumerate to get positions, accumulate into a defaultdict(float) keyed by id, then sort. The constant k (≈60) damps how much the very top ranks dominate.

Show solution

RRF gives each item 1/(k+rank) points from each list and sums them; the constant k damps the top-rank dominance. Runnable:

def rrf(*lists, k=60):
    scores = {}
    for lst in lists:
        for rank, cid in enumerate(lst):
            scores[cid] = scores.get(cid, 0) + 1 / (k + rank)
    return sorted(scores, key=lambda c: -scores[c])

dense  = ["a", "b", "c", "d"]   # vector search order
sparse = ["c", "a", "e", "f"]   # BM25 order (exact-term match)
fused = rrf(dense, sparse)
print(fused)
# 'a' is top of dense and 2nd in sparse -> wins; 'c' (top sparse, 3rd dense) close behind
assert fused[0] == "a"
assert "e" in fused and "f" in fused    # sparse-only items still included

The subtlety: RRF only reads each item's position, so wildly different score scales from two retrievers combine without normalization — that robustness is why the lesson prefers it over weighted score-mixing.

Exercise 5 · Measure Recall@k and enforce a tenant filterProfessional

Context: Two production realities at once. You can't improve retrieval you don't measure, and in a multi-tenant product a retriever that returns another customer's chunk is a data breach — access control must live in code, never in the prompt.

Your task: Build a recall_at_k harness over (question, expected_chunk_id) pairs, and add a tenant filter that removes other tenants' chunks before similarity is computed. Make both runnable against a fake retriever.

Requirements:

recall_at_k returns the fraction of questions whose expected chunk appears in the top-k
Filtering happens before ranking, not by post-hoc dropping results
A query in tenant A can never return a tenant-B chunk — assert it
The tenant check is enforced in code, not requested in a prompt
Report recall for at least two values of k (e.g. k=1 and k=5)

💡 Hint: Filter the candidate list by chunk.tenant == current_tenant first, then rank the survivors. For recall, count a hit when expected_id in [c.id for c in top_k] and divide by the number of questions.

Show solution

Metric first, then the security filter. Both are pure Python and runnable:

def recall_at_k(search_fn, testset, k=5):
    hits = 0
    for q, expected in testset:
        ids = [c["id"] for c in search_fn(q, k=k)[:k]]
        hits += expected in ids
    return hits / len(testset)

CORPUS = [
    {"id":"t1#0", "tenant":"t1", "text":"reset password"},
    {"id":"t1#1", "tenant":"t1", "text":"enable 2fa"},
    {"id":"t2#0", "tenant":"t2", "text":"reset password"},   # other tenant, same words
]

def search(query, k=5, tenant=None):
    pool = [c for c in CORPUS if tenant is None or c["tenant"] == tenant]  # FILTER FIRST
    ranked = [c for c in pool if any(w in c["text"] for w in query.split())]
    return ranked[:k]

# tenant isolation: t1 must never see t2's chunk, even on an identical query
ids = [c["id"] for c in search("reset password", tenant="t1")]
print("t1 sees:", ids)
assert all(i.startswith("t1") for i in ids), "tenant leak!"

testset = [("reset password", "t1#0"), ("enable 2fa", "t1#1")]
print("recall@5:", recall_at_k(lambda q, k=5: search(q, k=k, tenant="t1"), testset))

Production points: test retrieval separately (most RAG 'hallucinations' are retrieval misses), and enforce ACL/tenant scoping in code before semantic search — a metadata filter, not a prompt instruction, is what actually prevents cross-tenant leaks.

Exercise 6 · Onboard the RAG assistant to a new customer, groundedlyIndustry scenario

Context: A new enterprise customer wants your assistant answering strictly from their runbooks. Legal's non-negotiables: it must never answer from outside their docs, and every answer must cite. This is the grounded-answer contract most real RAG deployments are judged on.

Your task: Design the end-to-end grounded flow — retrieve → build a numbered context → force cite-or-refuse — and provide the prompt plus the assembly code. The generation call itself needs an API key; the assembly and the refusal contract must be shown runnable offline.

Requirements:

Retrieved chunks are assembled into a numbered context block
The system prompt instructs: answer only from the context and cite the numbers used
If nothing relevant is retrieved, the system returns a fixed refusal — it does not call the model to 'try anyway'
Every non-refusal answer carries at least one [n] citation
Assembly + the refusal decision run without network; only the final generate needs a key
Note how you'd verify each cited number actually maps to a retrieved chunk

💡 Hint: Gate on retrieval score: if the best similarity is below a threshold (or the candidate list is empty), short-circuit to the refusal string before you ever build the prompt. Number chunks [1..n] and pass that same numbering into the prompt so citations are checkable.

Show solution

Design. "Onboard a customer" = point retrieval at their docs/ and keep the two anti-hallucination rules absolute: answer only from the numbered context, and say "I don't have that information" when it's missing. Per-customer isolation comes from the tenant filter (previous rung); citations come from numbering chunks so the model can reference [n].

SYSTEM = (
    "You answer strictly from the numbered context. "
    "Cite the sources you use inline as [n]. "
    "If the context does not contain the answer, say: "
    "'I don't have that information.' Never use outside knowledge."
)

def build_context(hits):
    """hits: list of (chunk_dict, score) -> numbered, citable context."""
    return "\n\n".join(
        f"[{i}] (source: {c['source']})\n{c['text']}"
        for i, (c, _) in enumerate(hits, 1)
    )

hits = [({"source":"runbook.md", "text":"Rotate the API key via the admin console."}, 0.71),
        ({"source":"runbook.md", "text":"Keys expire every 90 days."}, 0.55)]
ctx = build_context(hits)
print(ctx)
assert ctx.startswith("[1]") and "[2]" in ctx   # numbered so the model can cite

# Real generation (needs ANTHROPIC_API_KEY):
# resp = client.messages.create(model="claude-opus-4-8", max_tokens=600, system=SYSTEM,
#   messages=[{"role":"user","content":f"Context:\n{ctx}\n\nQuestion: {q}"}])

Tradeoffs. Strict grounding trades coverage for trust — the assistant will refuse rather than guess, which is exactly what legal wants; the fix for gaps is better retrieval, not a looser prompt. Verify onboarding by asking something absent from their docs and confirming the refusal fires, and by spot-checking that every claim carries a [n] traceable to a real chunk. Each customer is a separate tenant-scoped index so answers can never cross companies.

✓ Checkpoint — you can move on when you can…

Explain the offline (index) vs online (query) phases and what each box does.
Chunk a document sensibly and say why overlap and metadata matter.
Explain what an embedding is and how cosine similarity retrieves chunks.
Describe why hybrid + re-ranking beats pure vector search.
Write the two system-prompt rules that stop hallucination, and prove the refusal works.
Measure Recall@k by hand and debug a "hallucination" as a retrieval failure.

🏗️ Toward the capstoneRAG is how the AI DevOps Engineer gets onboarded into any company. The generic agent knows Kubernetes; RAG over this company's runbooks, conventions, and past PRs is what makes it know their stack. The docs/ folder you indexed here becomes their runbook repository. "Onboard to a new customer" literally means "point RAG at their docs." See RAG-over-runbooks in the capstone →

☁️ Managed RAG on AWSYou built this pipeline by hand to own the tradeoffs. When you want AWS to manage the embeddings, vector store, and retrieval, a Bedrock Knowledge Base does exactly this behind two API calls. See W4 · Knowledge Bases (managed RAG) →

Knowledge check check yourself

✓ Knowledge check

The chapter claims most wrong answers in a RAG system are retrieval failures, not generation failures. What does that imply about how and in what order you should test the pipeline?

Show answer

It means the model usually answered correctly but from the wrong or missing context, so you must test retrieval separately and first — run real questions through search() and confirm the answer-bearing chunk lands in the top-k before writing any generation code. No clever prompt can rescue context that was never retrieved.

✓ Knowledge check

Pure vector search and BM25 keyword search each fail in a characteristic way. Describe each failure and explain why combining them (then re-ranking) is the biggest quality jump in RAG.

Show answer

Vector search captures meaning but misses exact terms like error codes, product names, or acronyms; BM25 nails exact terms but misses paraphrase. Fusing both (e.g. via Reciprocal Rank Fusion) then re-ranking the top candidates gives both recall and precision — you over-fetch cheaply and wide, then narrow precisely, so you catch exact matches and semantic matches without wasting tokens on junk.

Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.

Learning objectives

What RAG is, and when to use it intermediate

The two-phase pipeline intermediate

Lab 3.1 · Chunking intermediate

Lab 3.2 · Embed & store intermediate

Lab 3.3 · Retrieve advanced

Lab 3.4 · Generate a grounded, cited answer advanced

Lab 3.5 · Hybrid search + re-ranking advanced

Measuring your RAG system expert

Common pitfalls expert

Exercises expert

Exercise 3.1 — Refusal test

Exercise 3.2 — Metadata filter

Exercise 3.3 — Measure Recall@k

🪜 Practice ladder beginner → industry

✓ Checkpoint — you can move on when you can…

Knowledge check check yourself

More in Core curriculum