RAG — Retrieval-Augmented Generation

Real-World Scenario

A fintech company has 50,000 internal policy documents. Customer support agents spend 30 minutes per inquiry searching through them. A RAG system lets agents ask natural language questions and get answers grounded in the actual policy documents — with citations. Implementation takes one afternoon; the ROI is immediate.

How RAG Works

Query → Embed query → Search vector DB → Retrieve top-k chunks
  → Inject chunks into prompt → LLM generates grounded answer

The key insight: instead of asking the LLM to memorize facts, you retrieve the relevant facts at query time and hand them to the LLM in the prompt. The LLM only needs to reason over and synthesize the retrieved context.

Building a RAG Pipeline from Scratch

# pip install anthropic numpy scikit-learn tiktoken
import anthropic
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
import json

client = anthropic.Anthropic()

# Step 1: Define documents (replace with real document loading)
DOCUMENTS = [
    {
        "id": "policy_001",
        "title": "Refund Policy",
        "content": """Customers may request a full refund within 30 days of purchase.
        After 30 days and up to 90 days, a 50% store credit is issued.
        After 90 days, no refunds are available. Digital products are non-refundable
        once downloaded. Damaged items qualify for full refund at any time with photo evidence."""
    },
    {
        "id": "policy_002",
        "title": "Shipping Policy",
        "content": """Standard shipping takes 5-7 business days. Express shipping (2-3 days)
        costs $15. Overnight shipping costs $35. Free standard shipping on orders over $50.
        International shipping is available to 45 countries, taking 10-21 business days.
        Orders are processed within 24 hours on business days."""
    },
    {
        "id": "policy_003",
        "title": "Account Security Policy",
        "content": """Passwords must be at least 12 characters with uppercase, lowercase,
        numbers, and symbols. Two-factor authentication is required for accounts with
        more than $500 in transactions. After 5 failed login attempts, the account is
        locked for 30 minutes. Security questions are deprecated; use TOTP apps instead."""
    },
    {
        "id": "policy_004",
        "title": "Warranty Policy",
        "content": """Electronics carry a 1-year manufacturer warranty. Extended warranty
        (3 years) is available at 15% of product price. Warranty covers defects in
        materials and workmanship. Physical damage, water damage, and unauthorized
        modifications void the warranty. Warranty service is provided via mail-in only."""
    },
]

# Step 2: Chunk documents into smaller pieces
def chunk_text(text: str, chunk_size: int = 200, overlap: int = 30) -> list[str]:
    """Split text into overlapping chunks by word count."""
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = min(start + chunk_size, len(words))
        chunks.append(" ".join(words[start:end]))
        start += chunk_size - overlap
    return chunks


# Create a flat list of chunks with metadata
chunks = []
for doc in DOCUMENTS:
    for i, chunk_text_content in enumerate(chunk_text(doc["content"])):
        chunks.append({
            "chunk_id":   f"{doc['id']}_chunk_{i}",
            "doc_id":     doc["id"],
            "title":      doc["title"],
            "content":    chunk_text_content,
        })

print(f"Total chunks: {len(chunks)}")

# Step 3: Embed all chunks using Anthropic's embedding model
# For production, use a dedicated embedding API (Voyage AI, OpenAI, etc.)
# This example uses a simple TF-IDF approach for illustration;
# replace embed_texts() with real embedding API calls in production.

from sklearn.feature_extraction.text import TfidfVectorizer

# Real embedding code with Voyage AI (recommended with Claude):
# import voyageai
# vo = voyageai.Client()
# embeddings = vo.embed([c["content"] for c in chunks], model="voyage-3").embeddings

# Using TF-IDF as a stand-in (not as good as dense embeddings):
vectorizer = TfidfVectorizer(max_features=1000, stop_words="english")
chunk_texts = [c["content"] for c in chunks]
chunk_embeddings = vectorizer.fit_transform(chunk_texts).toarray()

print(f"Embedding shape: {chunk_embeddings.shape}")

# Step 4: Retrieval function
def retrieve(query: str, k: int = 3) -> list[dict]:
    """Find the k most relevant chunks for a query."""
    query_embedding = vectorizer.transform([query]).toarray()
    similarities = cosine_similarity(query_embedding, chunk_embeddings)[0]
    top_k_indices = np.argsort(similarities)[::-1][:k]
    
    results = []
    for idx in top_k_indices:
        results.append({
            **chunks[idx],
            "similarity": float(similarities[idx]),
        })
    return results


# Test retrieval
results = retrieve("How long do I have to get a refund?")
for r in results:
    print(f"[{r['similarity']:.3f}] {r['title']}: {r['content'][:100]}...")

# Step 5: Generation with retrieved context
def rag_answer(question: str, k: int = 3) -> dict:
    """Retrieve relevant context and generate a grounded answer."""
    retrieved = retrieve(question, k=k)
    
    # Build context string with source attribution
    context = "\n\n".join([
        f"[Source: {r['title']}]\n{r['content']}"
        for r in retrieved
    ])
    
    system = """You are a helpful customer support agent.
Answer questions using ONLY the provided context.
If the context doesn't contain enough information, say so clearly.
Always cite which policy document your answer comes from."""

    prompt = f"""Context:
{context}

Question: {question}

Answer based only on the context above:"""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        system=system,
        messages=[{"role": "user", "content": prompt}]
    )

    return {
        "answer":    response.content[0].text,
        "sources":   [r["title"] for r in retrieved],
        "chunks":    retrieved,
    }


# Test the full pipeline
questions = [
    "Can I return a product I bought 45 days ago?",
    "How much does overnight shipping cost?",
    "What happens if I enter my password wrong 5 times?",
    "Is my laptop covered if I drop it?",
]

for q in questions:
    result = rag_answer(q)
    print(f"\nQ: {q}")
    print(f"A: {result['answer']}")
    print(f"Sources: {result['sources']}")
    print("-" * 60)

Production RAG with a Real Vector Database

# pip install chromadb anthropic
import chromadb
import anthropic
from typing import Optional

client = anthropic.Anthropic()
chroma = chromadb.Client()   # use chromadb.PersistentClient("./db") for persistence

collection = chroma.create_collection("company_docs")

def ingest_documents(documents: list[dict]) -> None:
    """Chunk and store documents in ChromaDB."""
    ids, docs, metadatas = [], [], []
    
    for doc in documents:
        chunks = chunk_text(doc["content"])
        for i, chunk in enumerate(chunks):
            ids.append(f"{doc['id']}_{i}")
            docs.append(chunk)
            metadatas.append({"title": doc["title"], "doc_id": doc["id"]})
    
    # ChromaDB handles embedding internally with its default model
    collection.add(documents=docs, ids=ids, metadatas=metadatas)
    print(f"Ingested {len(ids)} chunks from {len(documents)} documents")


def query_and_answer(question: str, n_results: int = 3) -> str:
    """RAG pipeline using ChromaDB retrieval."""
    results = collection.query(query_texts=[question], n_results=n_results)
    
    context_pieces = []
    for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
        context_pieces.append(f"[{meta['title']}]\n{doc}")
    
    context = "\n\n---\n\n".join(context_pieces)
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="Answer using only the provided context. Cite sources.",
        messages=[{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}]
    )
    return response.content[0].text


ingest_documents(DOCUMENTS)
print(query_and_answer("What warranty do electronics have?"))

RAG Quality Checklist