M09: RAG — Retrieval-Augmented Generation

Claude is brilliant but has a knowledge cutoff. RAG solves this by giving Claude access to your documents at query time — no fine-tuning required. You'll build a complete "Chat with Your Docs" system.

Learning Objectives

• Explain why RAG is needed and what problem it solves that prompt engineering cannot
• Describe the end-to-end RAG pipeline: load, chunk, embed, store, retrieve, generate
• Explain how text embeddings and cosine similarity enable semantic search
• Compare chunking strategies and their impact on retrieval quality
• Build a complete RAG system with ChromaDB and Claude

The Knowledge Problem

What Is RAG? (End-to-End Pipeline)

Here's what this looks like in practice. When a user asks a question, the retrieval step returns actual chunk objects like this:

Those chunk texts get pasted directly into Claude's prompt as context. Claude reads them, writes an answer, and cites [Source 1: filing_guide.md]. The user never sees the retrieval machinery — they just get an accurate, sourced answer.

Embeddings Explained from Scratch

So what IS an embedding? In plain English, it's a way of representing text as a point in space. Each word, sentence, or paragraph gets converted into a list of numbers — and those numbers encode what the text means, not just what words it contains. Two sentences that say the same thing in different words will end up at nearby points, even if they share zero words in common.

How does this actually work under the hood? An embedding model (a specialized neural network, much smaller than Claude) reads your text and outputs a fixed-size array of floats — typically 1,536 numbers. During training, the model learned to place semantically similar text at nearby coordinates. It doesn't follow rules you wrote; it learned these associations from millions of text pairs. The result is that "cancel my subscription" and "I want to stop paying" land near each other, even though they share no keywords.

If you're familiar with keyword search (like SQL's LIKE '%term%'), here's the key difference: keyword search requires an exact character match, so "heart attack" won't find "myocardial infarction." Embedding-based search doesn't care about specific words at all — it compares meaning. This is why RAG systems use embeddings instead of traditional search: your users won't always use the same vocabulary as your documents.

So what does an embedding actually look like? It's just an array of floating-point numbers — typically 1,536 of them:

Notice how the first two vectors start with almost identical numbers — they mean the same thing, so their coordinates are nearly the same. The pizza vector looks completely different. That's the whole trick: meaning becomes geometry, and "similar meaning" becomes "short distance."

Chunking Strategies

Chunking is the process of splitting your documents into smaller pieces before embedding. Why not just embed the whole document? Because embedding models work best with short text (under ~512 tokens). A 5,000-word document crammed into a single vector loses nuance — the embedding becomes a blurry average of everything the document discusses, making it hard to match specific questions.

The core challenge is finding the right granularity. Too large, and each chunk covers multiple topics, so the retriever finds the chunk but Claude has to hunt for the relevant sentence inside a wall of text. Too small, and each chunk loses context — "the treatment was effective" tells you nothing without knowing which treatment or which patient. Most teams start with 300–500 character chunks and tune from there based on retrieval quality.

If you've used database pagination (LIMIT 50 OFFSET 100), chunking feels similar — you're dividing data into pages. But there's a crucial difference: database pages don't overlap, while chunks should. Without overlap, a sentence that spans a chunk boundary gets split across two chunks, and neither chunk alone answers the question. The overlap parameter (typically 10–20% of chunk size) duplicates a few sentences at each boundary to prevent this.

Here's what actual chunks look like after splitting a short document. Notice how the overlap zone duplicates text at chunk boundaries to preserve context:

The overlap ensures "The filing office assigns a unique file number" appears in both chunks, so a query about file numbers will match regardless of which chunk the retriever finds.

Vector Databases

Here's what a "row" in a vector database actually contains — the vector, the original text, and metadata sitting side by side:

When you query the database, it compares your question's vector against every stored vector, finds the closest matches, and returns the document text and metadata for each. You never have to work with the raw vectors yourself — the database handles that.

Under the hood, vector databases use approximate nearest neighbor (ANN)An algorithm that finds vectors "close to" a query vector without checking every single vector in the database. ANN trades a tiny amount of accuracy for massive speed gains, using index structures like HNSW (Hierarchical Navigable Small World) graphs. algorithms to avoid comparing your query against every single stored vector. The most common algorithm is called HNSW (Hierarchical Navigable Small World). It works by building a graph structure that connects similar vectors to each other. When a query arrives, instead of checking all 100,000 vectors, the algorithm navigates the graph and checks maybe 200 — returning results in milliseconds with 95–99% accuracy. This small trade-off in precision makes the difference between a search that takes 50ms and one that takes 30 seconds.

If you already know SQL databases, here's the key contrast: in PostgreSQL, you'd write SELECT * WHERE debtor_name = 'Acme Corp' — an exact match. A vector database lets you say "find me everything semantically similar to 'Acme Corporation'" and it would also find "ACME CORP," "Acme Corp Inc," and related entities — because their embedding vectors are close together. It's the difference between exact lookup and meaning-based search.

Popular options include: ChromaDB (local, Python-native, great for prototyping), Pinecone (managed cloud service, scales to billions of vectors), and pgvector (a PostgreSQL extension that adds vector search to your existing database).

Citations — Native Provenance From Claude

The RAG pipeline you just built returns retrieved chunks and Claude's answer, but stitching those two together — "which sentence in the answer came from which chunk?" — is your responsibility. For audit-grade applications (legal, medical, regulated finance), that stitching is fragile: Claude can reword, summarize, or merge sources, and matching back to the original passages is error-prone.

Anthropic ships a built-in Citations feature that solves this at the API level. You pass your source documents into the request as document content blocks; Claude returns its answer with explicit citations arrays attached to each sentence, telling you exactly which document and which character span supports each claim. No regex, no fuzzy matching, no hallucinated quotes.

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "document",
                "source": {"type": "text", "media_type": "text/plain", "data": chunk_1},
                "title": "Filing UCC-2024-001",
                "citations": {"enabled": True}
            },
            # ...more documents...
            {"type": "text", "text": "Summarize the secured party's collateral interest."}
        ]
    }]
)
# response.content[0].citations -> [{document_index: 0, document_title: "...", cited_text: "..."}]

RAG vs Citations — Pick Per Job

The strongest production pattern combines them: use your RAG pipeline to find the top ~10 candidate chunks, then pass those chunks as citation-enabled documents in the request. You get the scale of vector search and the audit-grade provenance of Citations in the same call.

Multimodal Inputs — PDFs, Images, & the Files API

The retrieval pipeline you built so far assumes your documents arrive as plain text. In production they don't. They arrive as PDFs with embedded tables and figures, scanned images, charts, and screenshots. Claude has three native primitives for handling these without a separate OCR step or vision model: PDF support, vision, and the Files API. Knowing which to reach for — and when to combine them with your RAG pipeline — closes the last gap between a toy chatbot and a production document agent.

PDF Support — Skip the OCR Pipeline

Pass a PDF directly as a document content block (base64-encoded or via Files API reference) and Claude reads it natively — including tables, headings, and the layout structure that OCR'd text would lose. Limits to know: up to ~100 pages per document, ~32MB per file. For longer documents, split into chunks and use your RAG pipeline to shortlist before passing the most relevant pages.

import anthropic, base64

with open("ucc_filing.pdf", "rb") as f:
    pdf_data = base64.standard_b64encode(f.read()).decode("utf-8")

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "document",
                "source": {"type": "base64", "media_type": "application/pdf", "data": pdf_data},
                "title": "UCC-1 Filing 2024-001",
                "citations": {"enabled": True}  # ← pairs with Citations
            },
            {"type": "text", "text": "Extract debtor, secured party, and collateral. Cite each field."}
        ]
    }]
)

Vision — Images as First-Class Inputs

Same content-block paradigm with type: "image". Useful for: extracting data from charts/graphs, reading screenshots from agents that monitor dashboards, classifying scanned forms when you don't have native PDFs, and processing diagram-heavy documents (architecture diagrams, flowcharts). Cost: image tokens are calculated by image dimensions — ~1.15 tokens per pixel after resizing. A 1280x800 screenshot is roughly 1500 tokens.

with open("dashboard_screenshot.png", "rb") as f:
    image_data = base64.standard_b64encode(f.read()).decode("utf-8")

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {"type": "image",
             "source": {"type": "base64", "media_type": "image/png", "data": image_data}},
            {"type": "text", "text": "Read the latency chart and report the p95 value for the last 24h."}
        ]
    }]
)

Files API — Upload Once, Reference Many Times

If you reference the same document across many requests — an agent that answers questions about a 50-page contract over a multi-turn session, or a batch job that runs 1000 questions against the same PDF — uploading via the Files API beats re-encoding the file in every request. You upload once, get a file_id, and reference it by ID in subsequent messages.

# Step 1: upload (do this once)
file = client.files.create(file=open("contract.pdf", "rb"), purpose="user_data")
file_id = file.id  # e.g., "file_abc123"

# Step 2: reference by ID in any request
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {"type": "document", "source": {"type": "file", "file_id": file_id}},
            {"type": "text", "text": "What is the termination clause?"}
        ]
    }]
)

When to Use Each — Decision Matrix

Code Walkthrough: Chat with Your Docs

Step 1: Document Loading & Chunking

Let's start by getting our documents into the system. The load_documents function is straightforward — it reads every markdown and text file from a folder and returns them as a list. Nothing clever here, just file I/O with proper error handling (we skip files with encoding issues rather than crashing the whole pipeline).

The interesting part is chunk_text. Here's the dilemma: embedding models work best with short text — typically under 512 tokens — but our documents might be thousands of words. We could just chop the text every 500 characters, but that might cut a sentence in half. So instead, the function tries to find a natural break point — a paragraph boundary, a period, or at worst a space — before making the cut. This small detail makes a real difference in retrieval quality.

One thing to watch for: the 50-character overlap between chunks. Without overlap, information that spans a chunk boundary gets split across two chunks, and neither chunk alone contains the full thought. The overlap duplicates a small amount of text at each boundary, acting as insurance that context isn't lost.

# pip install chromadb>=0.4.0 anthropic>=0.30.0
import os
import glob

def load_documents(docs_dir: str) -> list[dict]:
    """Load all .md and .txt files from a directory."""
    docs = []
    for pattern in ["*.md", "*.txt"]:
        for path in glob.glob(os.path.join(docs_dir, pattern)):
            try:
                with open(path, "r", encoding="utf-8") as f:
                    docs.append({
                        "content": f.read(),
                        "source": os.path.basename(path),
                    })
            except (IOError, UnicodeDecodeError) as e:
                print(f"Skipping {path}: {e}")
    if not docs:
        raise FileNotFoundError(f"No documents found in {docs_dir}")
    return docs

def chunk_text(
    text: str, chunk_size: int = 500, overlap: int = 50
) -> list[str]:
    """Recursive character splitter with overlap."""
    if len(text) <= chunk_size:
        return [text]
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        # Try to break at a paragraph or sentence boundary
        if end < len(text):
            for sep in ["\n\n", "\n", ". ", " "]:
                last_sep = chunk.rfind(sep)
                if last_sep > chunk_size * 0.5:
                    end = start + last_sep + len(sep)
                    chunk = text[start:end]
                    break
        chunks.append(chunk.strip())
        start = end - overlap  # overlap for continuity
    return [c for c in chunks if c]

# Usage
docs = load_documents("./docs")
all_chunks = []
for doc in docs:
    chunks = chunk_text(doc["content"], chunk_size=500, overlap=50)
    for i, chunk in enumerate(chunks):
        all_chunks.append({
            "text": chunk,
            "source": doc["source"],
            "chunk_index": i,
        })
print(f"Loaded {len(docs)} docs, {len(all_chunks)} chunks")

// npm install chromadb@^1.7.0 @anthropic-ai/sdk@^0.30.0
import { readFileSync, readdirSync } from 'fs';
import { join, basename } from 'path';

function loadDocuments(docsDir) {
  const files = readdirSync(docsDir).filter(
    f => f.endsWith('.md') || f.endsWith('.txt')
  );
  if (files.length === 0)
    throw new Error(`No documents found in ${docsDir}`);
  return files.map(f => {
    try {
      return {
        content: readFileSync(join(docsDir, f), 'utf-8'),
        source: f,
      };
    } catch (e) {
      console.warn(`Skipping ${f}: ${e.message}`);
      return null;
    }
  }).filter(Boolean);
}

function chunkText(text, chunkSize = 500, overlap = 50) {
  if (text.length <= chunkSize) return [text];
  const chunks = [];
  let start = 0;
  while (start < text.length) {
    let end = start + chunkSize;
    let chunk = text.slice(start, end);
    if (end < text.length) {
      for (const sep of ['\n\n', '\n', '. ', ' ']) {
        const lastSep = chunk.lastIndexOf(sep);
        if (lastSep > chunkSize * 0.5) {
          end = start + lastSep + sep.length;
          chunk = text.slice(start, end);
          break;
        }
      }
    }
    chunks.push(chunk.trim());
    start = end - overlap;
  }
  return chunks.filter(Boolean);
}

const docs = loadDocuments('./docs');
const allChunks = [];
for (const doc of docs) {
  const chunks = chunkText(doc.content, 500, 50);
  chunks.forEach((chunk, i) => {
    allChunks.push({ text: chunk, source: doc.source, chunkIndex: i });
  });
}
console.log(`Loaded ${docs.length} docs, ${allChunks.length} chunks`);

Step 2: Embed & Store in ChromaDB

Now we need somewhere to store our chunks as searchable vectors. Here's the pleasant surprise: ChromaDB handles the embedding step for you. When you call collection.add() with text documents, ChromaDB automatically runs them through its built-in embedding model and stores the resulting vectors. You never touch a separate embedding API. The "hnsw:space": "cosine" setting tells ChromaDB to use cosine similarity when searching — the standard for text embeddings, as we discussed in the Embeddings section.

One gotcha that trips up every beginner: chromadb.Client() stores everything in memory only. When your Python script exits, all your vectors disappear. For this lab that's fine — we re-ingest each run. But for anything beyond a quick experiment, switch to chromadb.PersistentClient(path="./chroma_db"), which writes to disk and survives restarts. Also worth knowing: ChromaDB's default embedding model is good enough for prototyping, but for production you'll want Voyage AI or OpenAI embeddings, which produce more accurate similarity scores for domain-specific content like legal or medical text.

import chromadb

# ChromaDB uses its own built-in embedding model by default
client = chromadb.Client()  # in-memory; use PersistentClient for disk
collection = client.get_or_create_collection(
    name="my_docs",
    metadata={"hnsw:space": "cosine"},  # use cosine similarity
)

# Add chunks with metadata
try:
    collection.add(
        ids=[f"chunk_{i}" for i in range(len(all_chunks))],
        documents=[c["text"] for c in all_chunks],
        metadatas=[{"source": c["source"], "index": c["chunk_index"]}
                   for c in all_chunks],
    )
    print(f"Stored {collection.count()} chunks in ChromaDB")
except Exception as e:
    print(f"Error storing chunks: {e}")
    raise

import { ChromaClient } from 'chromadb';

const chroma = new ChromaClient();
const collection = await chroma.getOrCreateCollection({
  name: "my_docs",
  metadata: { "hnsw:space": "cosine" },
});

try {
  await collection.add({
    ids: allChunks.map((_, i) => `chunk_${i}`),
    documents: allChunks.map(c => c.text),
    metadatas: allChunks.map(c => ({
      source: c.source, index: c.chunkIndex
    })),
  });
  const count = await collection.count();
  console.log(`Stored ${count} chunks in ChromaDB`);
} catch (e) {
  console.error(`Error storing chunks: ${e.message}`);
  throw e;
}

Step 3: Retrieve & Generate with Claude

This is where everything comes together — the "Retrieve" and "Generate" stages working as a pair. When a user asks a question, two things happen in sequence. First, we query ChromaDB to find the top-k most semantically similar chunks. Then we format those chunks into a context block and send them to Claude along with the original question. The key insight: we inject only the most relevant chunks (typically 3–5), not the entire corpus. This keeps the prompt small enough for Claude's context window while providing exactly the information needed to answer.

Pay close attention to the system prompt — it's the most important design decision in this entire function. We tell Claude to answer "based ONLY on the provided context" and to explicitly say so if the context doesn't contain the answer. Why so strict? Without this grounding instruction, Claude will cheerfully fall back to its training data and make up plausible-sounding answers — defeating the entire purpose of RAG. Also notice we wrap retrieval and generation in separate try/except blocks. These are independent failure points: ChromaDB might be down, or the Claude API might return a rate limit error. Catching them separately means you get a clear error message pointing to exactly which stage failed.

import anthropic

claude = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY env var

def query_rag(question: str, top_k: int = 3) -> str:
    """Retrieve relevant chunks and generate an answer with citations."""
    # Step 1: Retrieve
    try:
        results = collection.query(
            query_texts=[question],
            n_results=top_k,
        )
    except Exception as e:
        return f"Retrieval error: {e}"

    if not results["documents"] or not results["documents"][0]:
        return "No relevant documents found. I don't have enough information."

    # Check relevance threshold
    distances = results["distances"][0] if results.get("distances") else []
    chunks = results["documents"][0]
    sources = results["metadatas"][0]

    # Build context from retrieved chunks
    context_parts = []
    for i, (chunk, meta) in enumerate(zip(chunks, sources)):
        context_parts.append(
            f"[Source {i+1}: {meta['source']}]\n{chunk}"
        )
    context = "\n\n---\n\n".join(context_parts)

    # Step 2: Generate with Claude
    try:
        response = claude.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=(
                "You are a helpful assistant that answers questions "
                "based ONLY on the provided context. If the context "
                "doesn't contain the answer, say so. Always cite "
                "your sources using [Source N] format."
            ),
            messages=[{
                "role": "user",
                "content": (
                    f"Context:\n{context}\n\n"
                    f"Question: {question}\n\n"
                    "Answer based on the context above, citing sources:"
                ),
            }],
        )
        return response.content[0].text
    except anthropic.APIError as e:
        return f"Generation error: {e.status_code} - {e.message}"

# Test it
answer = query_rag("What are the key features of the product?")
print(answer)

import Anthropic from '@anthropic-ai/sdk';

const claude = new Anthropic(); // reads ANTHROPIC_API_KEY env var

async function queryRag(question, topK = 3) {
  // Step 1: Retrieve
  let results;
  try {
    results = await collection.query({
      queryTexts: [question],
      nResults: topK,
    });
  } catch (e) {
    return `Retrieval error: ${e.message}`;
  }

  if (!results.documents?.[0]?.length) {
    return "No relevant documents found. I don't have enough information.";
  }

  const chunks = results.documents[0];
  const sources = results.metadatas[0];

  const context = chunks.map((chunk, i) =>
    `[Source ${i + 1}: ${sources[i].source}]\n${chunk}`
  ).join('\n\n---\n\n');

  // Step 2: Generate with Claude
  try {
    const response = await claude.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 1024,
      system:
        "You are a helpful assistant that answers questions " +
        "based ONLY on the provided context. If the context " +
        "doesn't contain the answer, say so. Always cite " +
        "your sources using [Source N] format.",
      messages: [{
        role: "user",
        content:
          `Context:\n${context}\n\n` +
          `Question: ${question}\n\n` +
          "Answer based on the context above, citing sources:",
      }],
    });
    return response.content[0].text;
  } catch (e) {
    return `Generation error: ${e.status} - ${e.message}`;
  }
}

const answer = await queryRag("What are the key features of the product?");
console.log(answer);

Hands-On Exercise

Environment Setup

mkdir rag-lab && cd rag-lab
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\activate
pip install "anthropic>=0.30.0" "chromadb>=0.4.0"
export ANTHROPIC_API_KEY=your-key-here             # Windows: set ANTHROPIC_API_KEY=your-key-here

Step 1: Create Sample Documents & Ingestion Pipeline

First, we need documents to search. We'll create a docs/ folder with 3 sample files, then build the ingestion pipeline that loads, chunks, and stores them in ChromaDB. This covers the first 4 stages of the RAG pipeline: Load → Chunk → Embed → Store.

Create a docs/ folder and add these 3 files:

# UCC Filing Guide

## Initial Filing (UCC-1)
A UCC-1 financing statement must be filed in the state where the debtor is located. For individuals, this is their principal residence. For registered organizations (corporations, LLCs), this is the state of organization.

The filing office assigns a unique file number and timestamps the filing. A properly filed UCC-1 is effective for 5 years from the date of filing.

## Continuation Statements
To maintain perfection beyond 5 years, a continuation statement (UCC-3) must be filed within 6 months before the 5-year lapse date. Missing this window means the filing lapses and the secured party loses priority.

## Amendments
Amendments can change the collateral description, add or remove debtors, or assign the financing statement to a new secured party. Each amendment receives its own file number but references the original UCC-1.

# Risk Assessment Criteria

## High Risk Indicators
- Debtor has multiple UCC filings from different secured parties (competing liens)
- Collateral description uses broad terms like "all assets" (blanket lien)
- Filing is within 90 days of a bankruptcy petition (preference risk)
- Debtor name on filing doesn't exactly match legal name (perfection defect)

## Medium Risk Indicators
- Continuation statement filed close to the deadline (less than 30 days before lapse)
- Collateral type is inventory or accounts receivable (high turnover)
- Secured party is an individual rather than an institution

## Low Risk Indicators
- Single secured party with specific collateral description
- Filing has been continued at least once (established relationship)
- Debtor is a well-established registered organization

Q: What happens if a continuation statement is filed late?
A: If filed after the 5-year lapse date, the original filing is no longer effective. The secured party must file a new UCC-1, and their priority date resets to the new filing date. Any liens filed by other parties in the gap period will have priority.

Q: Can a UCC filing be terminated?
A: Yes. The secured party can file a UCC-3 termination statement. Once filed, the financing statement is no longer effective. The debtor can also demand termination if the obligation has been satisfied.

Q: What is the difference between a UCC-1 and a UCC-3?
A: A UCC-1 is the initial financing statement that creates the public record. A UCC-3 is used for all subsequent changes: continuations, amendments, assignments, and terminations. The UCC-3 always references the original UCC-1 file number.

Now create rag_pipeline.py with the ingestion pipeline:

import os
import glob
import json
import anthropic
import chromadb

# ── Document Loading ─────────────────────────────────────────
def load_documents(docs_dir: str) -> list[dict]:
    """Load all .md and .txt files from a directory."""
    docs = []
    for pattern in ["*.md", "*.txt"]:
        for path in glob.glob(os.path.join(docs_dir, pattern)):
            try:
                with open(path, "r", encoding="utf-8") as f:
                    docs.append({"content": f.read(), "source": os.path.basename(path)})
            except (IOError, UnicodeDecodeError) as e:
                print(f"  Skipping {path}: {e}")
    if not docs:
        raise FileNotFoundError(f"No documents found in {docs_dir}")
    return docs

# ── Chunking ─────────────────────────────────────────────────
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split text into chunks with overlap at natural boundaries."""
    if len(text) <= chunk_size:
        return [text]
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if end < len(text):
            for sep in ["\n\n", "\n", ". ", " "]:
                last_sep = chunk.rfind(sep)
                if last_sep > chunk_size * 0.5:
                    end = start + last_sep + len(sep)
                    chunk = text[start:end]
                    break
        chunks.append(chunk.strip())
        start = end - overlap
    return [c for c in chunks if c]

# ── Ingestion (Load → Chunk → Embed → Store) ────────────────
def ingest(docs_dir: str = "./docs") -> chromadb.Collection:
    """Load docs, chunk them, and store in ChromaDB."""
    print("── Ingestion Pipeline ──")
    docs = load_documents(docs_dir)
    print(f"  Loaded {len(docs)} documents")

    all_chunks = []
    for doc in docs:
        chunks = chunk_text(doc["content"], chunk_size=500, overlap=50)
        for i, chunk in enumerate(chunks):
            all_chunks.append({"text": chunk, "source": doc["source"], "index": i})
    print(f"  Created {len(all_chunks)} chunks")

    client = chromadb.Client()  # in-memory for simplicity
    collection = client.get_or_create_collection(
        name="rag_lab", metadata={"hnsw:space": "cosine"}
    )
    collection.add(
        ids=[f"chunk_{i}" for i in range(len(all_chunks))],
        documents=[c["text"] for c in all_chunks],
        metadatas=[{"source": c["source"], "index": c["index"]} for c in all_chunks],
    )
    print(f"  Stored {collection.count()} chunks in ChromaDB")
    return collection

# ── Run ingestion ────────────────────────────────────────────
if __name__ == "__main__":
    collection = ingest()
    print("\n✓ Ingestion complete. Ready for queries.")

Run the ingestion:

Step 2: Add Retrieval & Generation with Claude

Depends on: Step 1 (this step uses the collection object and ingest() function from Step 1. If you're starting fresh, complete Step 1 first.)

Now we wire up the last two pipeline stages — Retrieve and Generate. This function searches ChromaDB for the top-3 most relevant chunks, formats them into a context block with source labels, and sends everything to Claude with a grounding prompt.

Append the query_rag function to rag_pipeline.py and replace the existing if __name__ == "__main__" block with the code below (keep the imports and the load_documents, chunk_text, and ingest functions from Step 1):

# ── Query (Retrieve → Generate) ──────────────────────────────
def query_rag(collection, question: str, top_k: int = 3, verbose: bool = True) -> str:
    """Search for relevant chunks and generate an answer with Claude."""
    client = anthropic.Anthropic()

    # Retrieve
    results = collection.query(query_texts=[question], n_results=top_k)
    chunks = results["documents"][0]
    sources = results["metadatas"][0]
    distances = results["distances"][0] if results.get("distances") else []

    if verbose:
        print(f"\n  Query: {question}")
        print(f"  Retrieved {len(chunks)} chunks:")
        for i, (chunk, meta) in enumerate(zip(chunks, sources)):
            dist = f" (distance: {distances[i]:.3f})" if distances else ""
            print(f"    [{i+1}] {meta['source']}{dist}: {chunk[:80]}...")

    # Format context
    context = "\n\n---\n\n".join(
        f"[Source {i+1}: {meta['source']}]\n{chunk}"
        for i, (chunk, meta) in enumerate(zip(chunks, sources))
    )

    # Generate
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=(
            "You are a helpful assistant that answers questions "
            "based ONLY on the provided context. If the context "
            "doesn't contain the answer, say so explicitly. "
            "Always cite sources using [Source N] format."
        ),
        messages=[{
            "role": "user",
            "content": f"Context:\n{context}\n\nQuestion: {question}\n\nAnswer based on the context above, citing sources:",
        }],
    )
    return response.content[0].text


# ── Test Queries ─────────────────────────────────────────────
if __name__ == "__main__":
    collection = ingest()

    print("\n" + "=" * 55)
    print("RAG QUERY TESTS")
    print("=" * 55)

    questions = [
        "How long is a UCC-1 filing effective?",
        "What are the high risk indicators for a UCC filing?",
        "What happens if a continuation statement is filed late?",
        "What is the difference between a UCC-1 and a UCC-3?",
        "What is the weather in Tokyo?",  # Should say "not in context"
    ]

    for q in questions:
        answer = query_rag(collection, q)
        print(f"\n  Answer: {answer[:200]}...")
        print(f"  {'─' * 50}")

Run the complete pipeline:

Verify Everything Works

Run the complete pipeline end-to-end. All 5 queries should return answers, with the first 4 citing sources and the 5th correctly declining to answer:

If all queries complete and you see cited answers for domain questions and a "not in context" response for the weather question, you've built a working RAG pipeline.

Knowledge Check

Module Summary

References & Resources

• RAG Guide — Anthropic Documentation
• ChromaDB Documentation
• RAG Cookbook — Anthropic GitHub
• Voyage AI Embeddings — Documentation
• Chunking Strategies — Pinecone Learning