Why use RAG instead of fine-tuning?

RAG retrieves up-to-date facts at inference time — fine-tuning bakes knowledge into weights at training time. RAG is cheaper (no GPU training), keeps knowledge current without retraining, and lets you cite sources. Fine-tune when you need to change the model's behavior or style, not just its knowledge.

What chunk size should I use?

Start with 512–1024 tokens with 10–20% overlap. Larger chunks preserve more context but reduce retrieval precision. Smaller chunks are more precise but may lack context for the answer. The right size depends on your document structure — dense technical docs often need smaller chunks than narrative prose.

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

Issue	Symptom	Fix
Wrong chunks retrieved	Irrelevant answers	Better chunking, denser embeddings (Voyage AI)
Context too long	Model ignores parts	Reduce k, smaller chunks, re-rank results
Hallucination despite context	Model adds info not in docs	Tighten system prompt, set temperature = 0
Stale knowledge	Outdated answers	Add document timestamps, filter by recency
Too slow	High latency	Cache embeddings, async retrieval, smaller model

Practice

Every task here exercises the retrieval half of the pipeline, which runs entirely on your machine. None of them calls an LLM API. Save Steps 1-4 above as rag_base.py, minus the import anthropic and client = anthropic.Anthropic() lines, which retrieval doesn’t use. Each solution imports from it. Importing it re-runs its Step 2-4 print calls, so every output below starts with the same five lines.

Only 4 chunks came out of 4 documents. Find out why, then re-chunk the Refund Policy with chunk_size=20, overlap=5 and print where each chunk starts and ends.

Solution

from rag_base import DOCUMENTS, chunk_text

refund = DOCUMENTS[0]["content"]
print(f"\nRefund Policy: {len(refund.split())} words")
for size, overlap in [(200, 30), (20, 5)]:
    pieces = chunk_text(refund, chunk_size=size, overlap=overlap)
    print(f"chunk_size={size}, overlap={overlap}: {len(pieces)} chunk(s)")

for i, piece in enumerate(chunk_text(refund, chunk_size=20, overlap=5)):
    words = piece.split()
    print(f"  {i}: {len(words):2d} words  starts '{' '.join(words[:3])}'  ends '{' '.join(words[-5:])}'")

$ python small_chunks.py
Total chunks: 4
Embedding shape: (4, 92)
[0.318] Refund Policy: Customers may request a full refund within 30 days of purchase. After 30 days and up to 90 days, a 5...
[0.000] Warranty Policy: Electronics carry a 1-year manufacturer warranty. Extended warranty (3 years) is available at 15% of...
[0.000] Account Security Policy: Passwords must be at least 12 characters with uppercase, lowercase, numbers, and symbols. Two-factor...

Refund Policy: 50 words
chunk_size=200, overlap=30: 1 chunk(s)
chunk_size=20, overlap=5: 4 chunk(s)
  0: 20 words  starts 'Customers may request'  ends 'up to 90 days, a'
  1: 20 words  starts 'up to 90'  ends 'are available. Digital products are'
  2: 20 words  starts 'are available. Digital'  ends 'any time with photo evidence.'
  3:  5 words  starts 'any time with'  ends 'any time with photo evidence.'

At 50 words, each policy fits in one 200-word chunk, so the default settings never split anything. With 20/5, each chunk begins with the previous chunk’s last 5 words, which is the overlap. Look at chunk 3, though: its 5 words are already the end of chunk 2. chunk_text only stops once start passes the end, so it emits a final chunk that is pure overlap. That chunk would be embedded and retrieved as a near-duplicate.

The lesson tests retrieval with one question. Run retrieve(q, k=1) on all four questions from Step 5 and check whether the top chunk is the right policy.

Solution

from rag_base import retrieve

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?",
]
print()
for q in questions:
    best = retrieve(q, k=1)[0]
    print(f"[{best['similarity']:.3f}] {best['title']:24s} <- {q}")

$ python top1.py
Total chunks: 4
Embedding shape: (4, 92)
[0.318] Refund Policy: Customers may request a full refund within 30 days of purchase. After 30 days and up to 90 days, a 5...
[0.000] Warranty Policy: Electronics carry a 1-year manufacturer warranty. Extended warranty (3 years) is available at 15% of...
[0.000] Account Security Policy: Passwords must be at least 12 characters with uppercase, lowercase, numbers, and symbols. Two-factor...

[0.254] Shipping Policy          <- Can I return a product I bought 45 days ago?
[0.501] Shipping Policy          <- How much does overnight shipping cost?
[0.000] Warranty Policy          <- What happens if I enter my password wrong 5 times?
[0.000] Warranty Policy          <- Is my laptop covered if I drop it?

Only one of four is right. TF-IDF matches exact words: “password” doesn’t match “Passwords”, “laptop” and “drop” appear in no policy, and “days” pulls the refund question toward shipping’s “business days”. The last two scored 0.000 against every chunk, yet retrieve still returned a “top” result, because argsort always ranks something first.

Retrieval that matches nothing shouldn’t hand the model a context anyway. Wrap retrieve so it drops chunks below a similarity of 0.1 and returns an empty list when nothing is left. Run it on the four questions with k=3.

Solution

from rag_base import retrieve

MIN_SIMILARITY = 0.1

def retrieve_or_refuse(query: str, k: int = 3) -> list[dict]:
    """Like retrieve(), but drop chunks that don't clear the similarity floor."""
    return [r for r in retrieve(query, k=k) if r["similarity"] >= MIN_SIMILARITY]

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?",
]
print()
for q in questions:
    hits = retrieve_or_refuse(q)
    titles = [h["title"] for h in hits] or "NO CONTEXT: answer 'I don't know'"
    print(f"{q}\n    -> {titles}")

$ python threshold.py
Total chunks: 4
Embedding shape: (4, 92)
[0.318] Refund Policy: Customers may request a full refund within 30 days of purchase. After 30 days and up to 90 days, a 5...
[0.000] Warranty Policy: Electronics carry a 1-year manufacturer warranty. Extended warranty (3 years) is available at 15% of...
[0.000] Account Security Policy: Passwords must be at least 12 characters with uppercase, lowercase, numbers, and symbols. Two-factor...

Can I return a product I bought 45 days ago?
    -> ['Shipping Policy', 'Refund Policy']
How much does overnight shipping cost?
    -> ['Shipping Policy']
What happens if I enter my password wrong 5 times?
    -> NO CONTEXT: answer 'I don't know'
Is my laptop covered if I drop it?
    -> NO CONTEXT: answer 'I don't know'

The floor removes the zero-score padding, and with k=3 the refund question now carries the Refund Policy too. But the password question is refused even though the Account Security Policy answers it. A threshold can only stop retrieval from returning junk. It can’t make retrieval find a document it scored at zero.

Context costs tokens. Build the exact prompt string Step 5 sends, without calling the API, and count its tokens with tiktoken’s local cl100k_base encoding for k = 1 to 4. Print the k=1 prompt too.

Solution

import tiktoken
from rag_base import retrieve

enc = tiktoken.get_encoding("cl100k_base")   # a local tokenizer: no API call

def build_prompt(question: str, k: int) -> str:
    """The exact context + prompt string Step 5 sends, without sending it."""
    retrieved = retrieve(question, k=k)
    context = "\n\n".join([f"[Source: {r['title']}]\n{r['content']}" for r in retrieved])
    return f"""Context:
{context}

Question: {question}

Answer based only on the context above:"""

question = "How much does overnight shipping cost?"
print()
for k in [1, 2, 3, 4]:
    print(f"k={k}: {len(enc.encode(build_prompt(question, k))):4d} prompt tokens")
print("\n" + build_prompt(question, k=1))

$ python prompt_cost.py
Total chunks: 4
Embedding shape: (4, 92)
[0.318] Refund Policy: Customers may request a full refund within 30 days of purchase. After 30 days and up to 90 days, a 5...
[0.000] Warranty Policy: Electronics carry a 1-year manufacturer warranty. Extended warranty (3 years) is available at 15% of...
[0.000] Account Security Policy: Passwords must be at least 12 characters with uppercase, lowercase, numbers, and symbols. Two-factor...

k=1:   90 prompt tokens
k=2:  153 prompt tokens
k=3:  220 prompt tokens
k=4:  294 prompt tokens

Context:
[Source: Shipping Policy]
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.

Question: How much does overnight shipping cost?

Answer based only on the context above:

The $35 answer is entirely in the k=1 prompt. Going to k=3, the lesson’s default, more than doubles the prompt (90 to 220 tokens) to add chunks that scored 0.000 for this question. cl100k_base is OpenAI’s tokenizer, not Claude’s, so treat the counts as estimates. The growth with k is the point. That’s the “Context too long → reduce k” row of the checklist, in numbers.

Swap TF-IDF for a dense embedding model that runs locally, all-MiniLM-L6-v2 from sentence-transformers (pip install sentence-transformers; the model downloads on first use). Embed the same chunks and rerun task 2’s four questions.

Solution

from sentence_transformers import SentenceTransformer
from rag_base import chunks

model = SentenceTransformer("all-MiniLM-L6-v2")      # runs locally, no API key
chunk_vecs = model.encode([c["content"] for c in chunks], normalize_embeddings=True)

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?",
]
print()
for q in questions:
    sims = chunk_vecs @ model.encode(q, normalize_embeddings=True)   # cosine similarity
    best = int(sims.argmax())
    print(f"[{sims[best]:.3f}] {chunks[best]['title']:24s} <- {q}")

$ python dense.py 2>/dev/null
Total chunks: 4
Embedding shape: (4, 92)
[0.318] Refund Policy: Customers may request a full refund within 30 days of purchase. After 30 days and up to 90 days, a 5...
[0.000] Warranty Policy: Electronics carry a 1-year manufacturer warranty. Extended warranty (3 years) is available at 15% of...
[0.000] Account Security Policy: Passwords must be at least 12 characters with uppercase, lowercase, numbers, and symbols. Two-factor...

[0.570] Refund Policy            <- Can I return a product I bought 45 days ago?
[0.676] Shipping Policy          <- How much does overnight shipping cost?
[0.512] Account Security Policy  <- What happens if I enter my password wrong 5 times?
[0.297] Warranty Policy          <- Is my laptop covered if I drop it?

All four are right now. Dense embeddings match meaning, so “password wrong 5 times” lands on “5 failed login attempts” and “laptop … drop it” on “electronics … physical damage” with no shared words. (2>/dev/null hides the model-loading progress bar, which goes to stderr.) The laptop match scores only 0.297, so task 3’s floor of 0.1 would need re-tuning for this model: dense similarity scores sit on a different scale from TF-IDF’s.

Real-World Scenario

How RAG Works

Building a RAG Pipeline from Scratch

Production RAG with a Real Vector Database

RAG Quality Checklist

Practice

Frequently Asked Questions