Most RAG tutorials give you 30 lines that work once on a sample PDF, then leave you alone when your pipeline starts hallucinating on the docs that actually matter.
This is the other tutorial. We're going to build a RAG pipeline in Python that handles the parts tutorials skip: chunking strategy, retrieval quality, evaluation, and the failure modes that show up when your document set grows past 50 files.
Total code: under 200 lines. No managed services required. No paid vector database subscription.
What RAG Is and What It Actually Fixes
RAG — Retrieval-Augmented Generation — is the pattern for giving an LLM access to your documents without putting your documents in the system prompt.
The naive approach is context stuffing:
# Don't do this
with open("all_docs.txt") as f:
context = f.read() # 200,000 tokens
response = client.messages.create(
model="claude-sonnet-5-5",
messages=[{"role": "user", "content": f"{context}\n\nUser question: {query}"}]
)
This breaks at scale. Token limits hit. Cost goes up linearly. Retrieval quality goes down as the LLM struggles to attend to relevant content buried in 50 pages of irrelevant context.
RAG fixes this by inverting the flow: instead of sending all documents to the LLM, you retrieve only the relevant chunks, then send only those:
User query → Embed query → Vector search → Top-k chunks → LLM → Answer
The hard parts are: chunking well, embedding the right things, retrieving accurately, and knowing when your retrieval is failing.
Dependencies
pip install anthropic chromadb sentence-transformers tiktoken
What each does:
-
anthropic— Claude API for generation -
chromadb— local vector database (no subscription, runs in-process) -
sentence-transformers— open-source embedding model (no API cost for embeddings) -
tiktoken— token counting for chunk size control
Versions tested (2026-10-27, Python 3.14): anthropic==2.3.0, chromadb==1.5.9, sentence-transformers==6.2.0, tiktoken==0.14.0
Part 1: Chunking — Where Most RAG Pipelines Fail
Bad chunking is the most common cause of retrieval failures. If you split at 500 characters regardless of content structure, you will split sentences mid-thought, lose context, and surface chunks that technically match the query but contain nothing useful.
The rule: chunk by semantic unit, not character count.
import tiktoken
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class Chunk:
text: str
source: str
chunk_index: int
token_count: int
metadata: dict = field(default_factory=dict)
def count_tokens(text: str, model: str = "cl100k_base") -> int:
enc = tiktoken.get_encoding(model)
return len(enc.encode(text))
def chunk_document(
text: str,
source: str,
target_tokens: int = 400,
overlap_tokens: int = 50,
) -> list[Chunk]:
"""
Chunk by paragraph boundary, respecting token limits.
Overlap prevents context loss at chunk edges.
"""
paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
chunks = []
current_paras: list[str] = []
current_tokens = 0
chunk_index = 0
for para in paragraphs:
para_tokens = count_tokens(para)
# If adding this paragraph exceeds limit, flush current chunk
if current_tokens + para_tokens > target_tokens and current_paras:
chunk_text = "\n\n".join(current_paras)
chunks.append(Chunk(
text=chunk_text,
source=source,
chunk_index=chunk_index,
token_count=current_tokens,
))
chunk_index += 1
# Overlap: keep last paragraph(s) up to overlap_tokens
overlap_paras = []
overlap_count = 0
for p in reversed(current_paras):
p_tokens = count_tokens(p)
if overlap_count + p_tokens <= overlap_tokens:
overlap_paras.insert(0, p)
overlap_count += p_tokens
else:
break
current_paras = overlap_paras
current_tokens = overlap_count
current_paras.append(para)
current_tokens += para_tokens
# Flush remaining
if current_paras:
chunks.append(Chunk(
text="\n\n".join(current_paras),
source=source,
chunk_index=chunk_index,
token_count=current_tokens,
))
return chunks
Why target_tokens=400? Claude's effective context window is large, but retrieval quality degrades when you return chunks that are too large. 300-500 tokens per chunk is the sweet spot for most document types. Adjust based on your document structure.
Part 2: Embedding and Storing
from sentence_transformers import SentenceTransformer
import chromadb
EMBEDDING_MODEL = "all-MiniLM-L6-v2" # 384 dimensions, fast, runs locally
COLLECTION_NAME = "docs"
def build_index(chunks: list[Chunk], persist_path: str = "./chroma_db") -> chromadb.Collection:
"""Embed chunks and store in ChromaDB."""
model = SentenceTransformer(EMBEDDING_MODEL)
client = chromadb.PersistentClient(path=persist_path)
# Delete and recreate for clean indexing
try:
client.delete_collection(COLLECTION_NAME)
except Exception:
pass
collection = client.create_collection(COLLECTION_NAME)
texts = [chunk.text for chunk in chunks]
embeddings = model.encode(texts, show_progress_bar=True).tolist()
collection.add(
ids=[f"{chunk.source}::{chunk.chunk_index}" for chunk in chunks],
embeddings=embeddings,
documents=texts,
metadatas=[{
"source": chunk.source,
"chunk_index": chunk.chunk_index,
"token_count": chunk.token_count,
} for chunk in chunks],
)
print(f"Indexed {len(chunks)} chunks from {len({c.source for c in chunks})} documents")
return collection
A note on embedding models: all-MiniLM-L6-v2 is fast and free, with solid retrieval performance for English technical documentation. If your documents are multilingual or highly specialized, consider all-mpnet-base-v2 or a domain-specific model.
Part 3: Retrieval
def retrieve(
query: str,
collection: chromadb.Collection,
top_k: int = 5,
min_relevance: float = 0.3,
) -> list[dict]:
"""
Retrieve top-k chunks for a query.
Filter by minimum relevance score to avoid noise.
"""
model = SentenceTransformer(EMBEDDING_MODEL)
query_embedding = model.encode([query]).tolist()
results = collection.query(
query_embeddings=query_embedding,
n_results=top_k,
include=["documents", "metadatas", "distances"],
)
chunks = []
for doc, meta, distance in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0],
):
# ChromaDB returns L2 distance — convert to similarity score
similarity = 1 / (1 + distance)
if similarity >= min_relevance:
chunks.append({
"text": doc,
"source": meta["source"],
"chunk_index": meta["chunk_index"],
"similarity": round(similarity, 3),
})
return chunks
The min_relevance filter matters. Without it, you will return chunks for queries that have no good answer in your corpus — and the LLM will hallucinate rather than admit ignorance.
Part 4: Generation with Retrieved Context
import anthropic
import os
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
def generate_answer(
query: str,
chunks: list[dict],
model: str = "claude-sonnet-5-5",
) -> dict:
"""Generate an answer grounded in the retrieved chunks."""
if not chunks:
return {
"answer": "I don't have relevant information in the provided documents to answer this question.",
"sources": [],
"chunks_used": 0,
}
# Build context block with source attribution
context_parts = []
for i, chunk in enumerate(chunks, 1):
context_parts.append(
f"[Source {i}: {chunk['source']}, relevance={chunk['similarity']}]\n{chunk['text']}"
)
context = "\n\n---\n\n".join(context_parts)
system_prompt = """You are a precise assistant that answers questions based only on the provided context.
Rules:
- Answer only from the context provided. Do not use prior knowledge.
- If the context doesn't contain enough information, say so explicitly.
- Cite sources by reference number when making factual claims.
- Keep answers concise and direct."""
response = client.messages.create(
model=model,
max_tokens=1024,
system=system_prompt,
messages=[{
"role": "user",
"content": f"Context:\n\n{context}\n\nQuestion: {query}",
}],
)
return {
"answer": response.content[0].text,
"sources": [c["source"] for c in chunks],
"chunks_used": len(chunks),
}
Part 5: Putting It Together
from pathlib import Path
def load_and_chunk_directory(docs_dir: str) -> list[Chunk]:
"""Load all .txt and .md files from a directory."""
chunks = []
docs_path = Path(docs_dir)
for file_path in sorted(docs_path.rglob("*")):
if file_path.suffix not in {".txt", ".md"}:
continue
text = file_path.read_text(encoding="utf-8")
file_chunks = chunk_document(
text=text,
source=str(file_path.relative_to(docs_path)),
)
chunks.extend(file_chunks)
print(f" {file_path.name}: {len(file_chunks)} chunks")
return chunks
def run_rag(docs_dir: str, query: str) -> None:
print("Loading and chunking documents...")
chunks = load_and_chunk_directory(docs_dir)
print(f"\nBuilding vector index ({len(chunks)} chunks)...")
collection = build_index(chunks)
print(f"\nQuery: {query}")
retrieved = retrieve(query, collection)
print(f"Retrieved {len(retrieved)} chunks")
for chunk in retrieved:
print(f" [{chunk['similarity']:.3f}] {chunk['source']} chunk {chunk['chunk_index']}")
result = generate_answer(query, retrieved)
print(f"\nAnswer ({result['chunks_used']} chunks used):")
print(result["answer"])
print(f"\nSources: {', '.join(set(result['sources']))}")
if __name__ == "__main__":
import sys
docs_dir = sys.argv[1] if len(sys.argv) > 1 else "./docs"
query = sys.argv[2] if len(sys.argv) > 2 else "What are the main features?"
run_rag(docs_dir, query)
The Failure Modes That Matter
1. Retrieval returning nothing (similarity too low)
Lower min_relevance or check that your documents are indexed. A common mistake: running the script without calling build_index first, or using a stale index after adding new documents.
2. Retrieval returning the wrong things
Check your chunk boundaries. If a relevant answer spans two chunks and those chunks are separated by overlap, retrieval may surface neither. Lower target_tokens to create smaller chunks, or increase overlap_tokens.
3. LLM ignoring the retrieved context
This is usually a prompt problem. The system prompt must explicitly constrain the model to the context. "Use only the provided context" is not enough — add the negative constraint: "Do not use prior knowledge."
4. Answers that are technically in context but wrong
Run retrieve() and print the chunks before calling generate_answer(). If the retrieved chunks don't contain the right answer, you have a retrieval problem, not a generation problem. Fix retrieval before tuning the prompt.
5. No way to know when you're failing
This is the one that actually bites in production. You need evaluation — not intuition.
Evaluation: Knowing When Your RAG Is Failing
The LLM Evaluation Workflow Pack (link) includes a RAG evaluation harness that measures:
- Context precision: Are the retrieved chunks actually relevant to the question?
- Context recall: Did you retrieve all the relevant chunks?
- Answer faithfulness: Is the generated answer grounded in the retrieved context, or is the LLM confabulating?
- Answer relevance: Does the answer actually address the question?
Without systematic evaluation, you're operating blind. A RAG pipeline that works on your test questions will fail on production queries you haven't thought of.
The prompt templates for evaluating RAG quality are in the AI Dev Toolkit (link) under the Evaluation category — specifically: "Evaluate RAG pipeline context precision", "Evaluate answer faithfulness to retrieved context", and "Generate diverse test queries for RAG evaluation."
What to Do Next
The pipeline above handles the 80% case. Four directions from here:
Choosing a vector store: ChromaDB is the easiest start, but it is not the only option. For a side-by-side of pgvector, Chroma and Qdrant, see Vector Databases for Python Developers.
Persistence: The ChromaDB index is rebuilt every run. For production, call build_index() once and persist to disk. Load with chromadb.PersistentClient(path="./chroma_db") on subsequent runs.
Hybrid retrieval: Pure semantic search misses exact matches. Add BM25 keyword retrieval alongside vector search and merge results. rank_bm25 (pure Python, no deps) handles this. Hybrid retrieval consistently outperforms either approach alone.
Reranking: After retrieval, use a cross-encoder model (e.g., cross-encoder/ms-marco-MiniLM-L-6-v2) to rerank the top-20 results and keep the top-5. This adds latency but improves precision significantly for ambiguous queries.
Full pipeline code: copy-paste from this article — everything runs standalone.
The AI Dev Toolkit (link) includes prompt templates for RAG architecture decisions, evaluation, and debugging.
The LLM Evaluation Workflow Pack (link) includes the full RAG evaluation harness.
Top comments (0)