Module 8: RAG — Retrieval-Augmented Generation

LLMs have a knowledge cutoff — they only know what was in their training data. They cannot answer questions about events after their cutoff date, your company's internal documents, or any private data they were never trained on. They also hallucinate confidently when they do not know something.

Retrieval-Augmented Generation (RAG) solves this by retrieving relevant documents at query time and injecting them into the LLM's context. Instead of relying on memorised knowledge, the model reads the actual source material and generates an answer grounded in that material.

RAG is the most widely deployed pattern in production AI systems because it is simpler and cheaper than fine-tuning, can be updated instantly (just add or remove documents), and provides citations back to source material.

A RAG system has two phases: an indexing phase (offline, run once or periodically) and a query phase (online, run for every user question). Understanding both phases is essential to building an effective pipeline.

Indexing Phase (Offline)

1. Load — Read documents from files, databases, APIs, or web pages
2. Chunk — Split documents into smaller, meaningful pieces
3. Embed — Convert each chunk into a vector (a list of numbers) using an embedding model
4. Store — Save the vectors and original text in a vector database

Query Phase (Online)

1. Embed Query — Convert the user's question into a vector using the same embedding model
2. Retrieve — Find the most similar document chunks using vector similarity search
3. Augment — Insert retrieved chunks into the LLM prompt as context
4. Generate — The LLM reads the context and produces a grounded answer

Chunking is arguably the most important step in a RAG pipeline. If your chunks are too large, they waste context tokens and dilute relevance. If they are too small, they lose context and the model cannot synthesise a coherent answer.

The goal is to create chunks that are self-contained (make sense on their own), focused (about one topic), and appropriately sized (typically 200-1000 tokens).

Chunking with Overlap

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split text into overlapping chunks by character count."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        # Try to end at a sentence boundary
        last_period = chunk.rfind(". ")
        if last_period > chunk_size * 0.5:
            end = start + last_period + 1
            chunk = text[start:end]
        chunks.append(chunk.strip())
        start = end - overlap  # overlap for context continuity
    return chunks

# Example
text = open("company_handbook.txt").read()
chunks = chunk_text(text, chunk_size=800, overlap=100)
print(f"Created {len(chunks)} chunks from {len(text)} characters")

Generating Embeddings

# Using OpenAI embeddings (most popular choice)
import openai

client = openai.OpenAI()

def get_embeddings(texts: list[str]) -> list[list[float]]:
    """Generate embeddings for a batch of texts."""
    response = client.embeddings.create(
        model="text-embedding-3-small",  # 1536 dimensions, cheap
        input=texts
    )
    return [item.embedding for item in response.data]

# Embed a single query
query_embedding = get_embeddings(["What is the refund policy?"])[0]

# Embed all chunks
chunk_embeddings = get_embeddings(chunks)

Vector databases are purpose-built for storing and searching high-dimensional vectors. They use specialised indexing algorithms (like HNSW or IVF) that make similarity search fast even over millions of vectors.

Setting Up ChromaDB

import chromadb

# Persistent storage (survives restarts)
client = chromadb.PersistentClient(path="./chroma_db")

# Create a collection
collection = client.get_or_create_collection(
    name="company_docs",
    metadata={"hnsw:space": "cosine"}  # cosine similarity
)

# Index documents
collection.add(
    documents=chunks,                          # raw text
    ids=[f"chunk_{i}" for i in range(len(chunks))],  # unique IDs
    metadatas=[{"source": "handbook.pdf", "page": i // 5}
               for i in range(len(chunks))]     # metadata for filtering
)

# Query
results = collection.query(
    query_texts=["What is the vacation policy?"],
    n_results=5,
    where={"source": "handbook.pdf"}  # optional metadata filter
)

for doc, score in zip(results["documents"][0], results["distances"][0]):
    print(f"[{score:.3f}] {doc[:100]}...")

Let us build a production-quality RAG pipeline that indexes a collection of documents and answers questions from them. This example uses ChromaDB for storage and Anthropic's Claude for generation.

import anthropic
import chromadb
import os
import json

class RAGPipeline:
    """Complete RAG pipeline: index, retrieve, generate."""

    def __init__(self, collection_name: str = "knowledge_base"):
        self.ai = anthropic.Anthropic()
        self.db = chromadb.PersistentClient(path="./rag_db")
        self.collection = self.db.get_or_create_collection(collection_name)

    def index_document(self, text: str, source: str, chunk_size: int = 800):
        """Chunk and index a document."""
        chunks = chunk_text(text, chunk_size=chunk_size, overlap=100)
        self.collection.add(
            documents=chunks,
            ids=[f"{source}_{i}" for i in range(len(chunks))],
            metadatas=[{"source": source, "chunk_index": i}
                       for i in range(len(chunks))]
        )
        print(f"Indexed {len(chunks)} chunks from '{source}'")

    def index_directory(self, dir_path: str):
        """Index all .txt and .md files in a directory."""
        for fname in os.listdir(dir_path):
            if fname.endswith((".txt", ".md")):
                path = os.path.join(dir_path, fname)
                text = open(path, "r", encoding="utf-8").read()
                self.index_document(text, source=fname)

    def retrieve(self, query: str, n_results: int = 5) -> list[dict]:
        """Retrieve relevant chunks for a query."""
        if self.collection.count() == 0:
            return []
        results = self.collection.query(
            query_texts=[query],
            n_results=min(n_results, self.collection.count())
        )
        return [
            {"text": doc, "source": meta["source"], "score": dist}
            for doc, meta, dist in zip(
                results["documents"][0],
                results["metadatas"][0],
                results["distances"][0]
            )
        ]

    def query(self, question: str, n_results: int = 5) -> dict:
        """Full RAG: retrieve context, then generate an answer."""
        chunks = self.retrieve(question, n_results)

        if not chunks:
            return {"answer": "No documents indexed yet.", "sources": []}

        # Build context string with source attribution
        context_parts = []
        for i, c in enumerate(chunks, 1):
            context_parts.append(f"[Source {i}: {c['source']}]\n{c['text']}")
        context = "\n\n---\n\n".join(context_parts)

        system_prompt = f"""You are a helpful assistant that answers questions
based ONLY on the provided context. If the context does not contain
the answer, say "I don't have enough information to answer that."
Always cite which source(s) you used, e.g., [Source 1].

Context:
{context}"""

        response = self.ai.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=system_prompt,
            messages=[{"role": "user", "content": question}]
        )

        return {
            "answer": response.content[0].text,
            "sources": [c["source"] for c in chunks],
            "chunks_used": len(chunks)
        }

# Usage
rag = RAGPipeline()
rag.index_directory("./company_docs")
result = rag.query("What is our remote work policy?")
print(result["answer"])
print("Sources:", result["sources"])

Measuring RAG quality requires evaluating both the retrieval step and the generation step separately. A common mistake is only evaluating the final answer without understanding whether failures come from bad retrieval or bad generation.

Simple Evaluation Script

def evaluate_rag(pipeline: RAGPipeline, test_cases: list[dict]) -> dict:
    """Evaluate RAG pipeline on test cases.
    Each test case: {"question": str, "expected_source": str, "expected_keywords": [str]}
    """
    results = {"retrieval_hits": 0, "answer_relevant": 0, "total": len(test_cases)}

    for tc in test_cases:
        result = pipeline.query(tc["question"])

        # Check retrieval: did we find the right source?
        if tc["expected_source"] in result["sources"]:
            results["retrieval_hits"] += 1

        # Check answer: does it contain expected keywords?
        answer_lower = result["answer"].lower()
        if all(kw.lower() in answer_lower for kw in tc["expected_keywords"]):
            results["answer_relevant"] += 1

    results["retrieval_accuracy"] = results["retrieval_hits"] / results["total"]
    results["answer_accuracy"] = results["answer_relevant"] / results["total"]
    return results

# Example test cases
test_cases = [
    {"question": "What is the vacation policy?",
     "expected_source": "hr_handbook.txt",
     "expected_keywords": ["days", "annual"]},
    {"question": "How do I submit expenses?",
     "expected_source": "finance_guide.txt",
     "expected_keywords": ["receipt", "submit"]},
]
metrics = evaluate_rag(rag, test_cases)
print(f"Retrieval accuracy: {metrics['retrieval_accuracy']:.0%}")
print(f"Answer accuracy: {metrics['answer_accuracy']:.0%}")