RAG grounding LLM answers is the most reliable way to make language models useful on private or changing data. Instead of hoping the model remembers something from training, you retrieve the relevant passages and force the answer to cite them. This post walks through a complete, minimal pipeline you can run locally — ingestion, chunking, embedding, retrieval, and generation with citations — plus how to verify each stage works.
Step 1: Define the retrieval unit and chunking strategy
Before you write any code, decide what a “document” means for your use case. A 50-page PDF is not one retrieval unit; a single paragraph might be too little context. Most production systems settle on 500–1000 token chunks with 10–20% overlap. That size fits comfortably in an embedding model’s context window and leaves room for the generator’s prompt.
# chunking.py
from langchain_text_splitters import RecursiveCharacterTextSplitter
def chunk_documents(raw_texts: list[str], chunk_size: int = 800, chunk_overlap: int = 100) -> list[str]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = []
for text in raw_texts:
chunks.extend(splitter.split_text(text))
return chunksVerify: Print the first three chunks and their character counts. You should see coherent semantic units, not mid-sentence cuts. If chunks look fragmented, increase chunk_size or adjust separators.
Step 2: Ingest and embed with a local or hosted model
Choose an embedding model that balances latency, cost, and quality. text-embedding-3-small (OpenAI) and bge-small-en-v1.5 (local, via sentence-transformers) are both solid defaults. Store vectors in a lightweight vector database — Chroma runs in-process and requires no infra.
# embed.py
import chromadb
from chromadb.utils import embedding_functions
from chunking import chunk_documents
def build_index(documents: list[dict], persist_dir: str = "./chroma_db"):
"""
documents: list of {"id": str, "text": str, "metadata": dict}
"""
client = chromadb.PersistentClient(path=persist_dir)
# Use OpenAI embeddings; swap for local model if preferred
ef = embedding_functions.OpenAIEmbeddingFunction(
model_name="text-embedding-3-small",
api_key="YOUR_KEY", # or read from env
)
collection = client.get_or_create_collection(
name="docs",
embedding_function=ef,
metadata={"hnsw:space": "cosine"},
)
all_chunks = []
all_metadatas = []
all_ids = []
for doc in documents:
chunks = chunk_documents([doc["text"]])
for i, chunk in enumerate(chunks):
all_chunks.append(chunk)
all_metadatas.append({**doc["metadata"], "chunk_index": i})
all_ids.append(f"{doc['id']}_chunk_{i}")
collection.add(documents=all_chunks, metadatas=all_metadatas, ids=all_ids)
return collectionVerify: Query the collection directly with a known phrase from your corpus. collection.query(query_texts=["exact phrase from doc"], n_results=3) should return that chunk at rank 1 with a high similarity score (>0.75 cosine).
Step 3: Build the retrieval pipeline with metadata filtering
Raw vector search is rarely enough. You need metadata filters (e.g., source == "contract_v3.pdf"), hybrid search (BM25 + vector), and a reranker. Start simple: vector search + metadata filter. Add a cross-encoder reranker once latency budget allows.
# retrieve.py
from chromadb import PersistentClient
from chromadb.utils import embedding_functions
def retrieve(
query: str,
collection_name: str = "docs",
persist_dir: str = "./chroma_db",
k: int = 8,
filter_metadata: dict | None = None,
) -> list[dict]:
client = chromadb.PersistentClient(path=persist_dir)
ef = embedding_functions.OpenAIEmbeddingFunction(
model_name="text-embedding-3-small",
api_key="YOUR_KEY",
)
collection = client.get_collection(name=collection_name, embedding_function=ef)
results = collection.query(
query_texts=[query],
n_results=k,
where=filter_metadata,
include=["documents", "metadatas", "distances"],
)
# Flatten to list of dicts
hits = []
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0],
):
hits.append({
"text": doc,
"metadata": meta,
"score": 1 - dist, # cosine similarity
})
return hitsVerify: Run three test queries: one exact-match, one paraphrase, one out-of-domain. Check that (a) exact-match returns the source chunk at top-1, (b) paraphrase returns semantically relevant chunks, (c) out-of-domain returns low scores (<0.4). If (b) fails, your chunking or embedding model is the bottleneck.
Step 4: Construct a citation-enforced prompt
The generator must cite sources inline. Use a strict instruction format and parse citations post-generation. Number the retrieved chunks and require the model to reference them like [1], [2]. Reject or retry answers that hallucinate citations.
# generate.py
from openai import OpenAI
from retrieve import retrieve
client = OpenAI(api_key="YOUR_KEY")
SYSTEM_PROMPT = """You are a precise question-answering system.
You will be given a question and numbered context passages.
Answer ONLY using the provided context.
Cite every claim by enclosing the passage number in square brackets, e.g., [1].
If the context does not contain the answer, say "I don't know based on the provided documents."
Do not use outside knowledge."""
def build_context(hits: list[dict]) -> str:
lines = []
for i, hit in enumerate(hits, 1):
src = hit["metadata"].get("source", "unknown")
lines.append(f"[{i}] (source: {src}) {hit['text']}")
return "\n\n".join(lines)
def generate_answer(question: str, hits: list[dict], model: str = "gpt-4o-mini") -> str:
context = build_context(hits)
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
]
resp = client.chat.completions.create(
model=model,
messages=messages,
temperature=0.0,
max_tokens=512,
)
return resp.choices[0].message.content.strip()Verify: Feed a question whose answer spans two chunks. The output should contain both [1] and [2]. Feed an unanswerable question — the model must emit the exact “I don’t know” string. If it improvises, lower temperature to 0.0 and sharpen the system prompt.
Step 5: Add a cross-encoder reranker for precision
Vector search optimizes recall; a cross-encoder reranker optimizes precision. cross-encoder/ms-marco-MiniLM-L-6-v2 runs in ~30 ms on CPU and significantly improves top-3 accuracy. Insert it between retrieval and generation.
# rerank.py
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def rerank(query: str, hits: list[dict], top_n: int = 5) -> list[dict]:
pairs = [(query, hit["text"]) for hit in hits]
scores = reranker.predict(pairs)
for hit, score in zip(hits, scores):
hit["rerank_score"] = float(score)
hits.sort(key=lambda h: h["rerank_score"], reverse=True)
return hits[:top_n]Verify: Compare top-3 hits before and after reranking on 20 held-out queries. You should see relevant chunks move up and irrelevant ones drop out. Measure latency — if it exceeds your budget, distill to a smaller model or batch rerank asynchronously.
Step 6: Implement the end-to-end query loop
Wire ingestion, retrieval, reranking, and generation into a single function. Add logging so you can trace every query’s retrieved chunks, rerank scores, and final answer.
# pipeline.py
from retrieve import retrieve
from rerank import rerank
from generate import generate_answer
import json
import time
def answer_question(
question: str,
filter_metadata: dict | None = None,
k_retrieve: int = 12,
k_rerank: int = 5,
) -> dict:
start = time.perf_counter()
hits = retrieve(question, k=k_retrieve, filter_metadata=filter_metadata)
retrieve_ms = (time.perf_counter() - start) * 1000
start = time.perf_counter()
hits = rerank(question, hits, top_n=k_rerank)
rerank_ms = (time.perf_counter() - start) * 1000
start = time.perf_counter()
answer = generate_answer(question, hits)
generate_ms = (time.perf_counter() - start) * 1000
return {
"question": question,
"answer": answer,
"citations": [
{"index": i+1, "source": h["metadata"].get("source"), "text": h["text"][:200]}
for i, h in enumerate(hits)
],
"latency_ms": {
"retrieve": round(retrieve_ms, 1),
"rerank": round(rerank_ms, 1),
"generate": round(generate_ms, 1),
"total": round(retrieve_ms + rerank_ms + generate_ms, 1),
},
}
if __name__ == "__main__":
q = "What is the termination notice period in the vendor agreement?"
result = answer_question(q, filter_metadata={"source": "vendor_agreement.pdf"})
print(json.dumps(result, indent=2))Verify: Run the script. Confirm the JSON output contains (a) an answer with inline [n] citations, (b) a citations array matching those numbers, (c) latency breakdown. Total latency should be <3 s for a typical query on GPT-4o-mini.
Step 7: Evaluate with a labeled test set
You cannot improve what you do not measure. Build a small eval set (30–50 questions) with ground-truth answers and required source documents. Score each run on citation precision, citation recall, and answer correctness.
# eval.py
import json
from pipeline import answer_question
EVAL_SET = [
{
"question": "What is the termination notice period in the vendor agreement?",
"answer": "30 days written notice",
"required_sources": ["vendor_agreement.pdf"],
},
# ... 29 more
]
def evaluate():
results = []
for item in EVAL_SET:
out = answer_question(item["question"])
# Citation precision: fraction of cited sources that are in required_sources
cited_sources = {c["source"] for c in out["citations"]}
required = set(item["required_sources"])
precision = len(cited_sources & required) / len(cited_sources) if cited_sources else 0
# Citation recall: fraction of required sources that were cited
recall = len(cited_sources & required) / len(required) if required else 1
# Answer correctness: simple substring check (replace with LLM judge for production)
correct = item["answer"].lower() in out["answer"].lower()
results.append({
"question": item["question"],
"precision": precision,
"recall": recall,
"correct": correct,
"latency_ms": out["latency_ms"]["total"],
})
avg_precision = sum(r["precision"] for r in results) / len(results)
avg_recall = sum(r["recall"] for r in results) / len(results)
accuracy = sum(r["correct"] for r in results) / len(results)
avg_latency = sum(r["latency_ms"] for r in results) / len(results)
print(f"Citation Precision: {avg_precision:.2f}")
print(f"Citation Recall: {avg_recall:.2f}")
print(f"Answer Accuracy: {accuracy:.2f}")
print(f"Avg Latency: {avg_latency:.0f} ms")
return results
if __name__ == "__main__":
evaluate()Verify: Run eval after every change to chunking, embedding model, reranker, or prompt. Target: citation precision >0.85, citation recall >0.80, answer accuracy >0.90. If precision is low, your retriever returns noise — tighten k_retrieve or improve chunking. If recall is low, you’re missing relevant chunks — increase k_retrieve or check embedding quality.
Step 8: Harden for production
The pipeline above runs locally. To ship it, address three gaps:
- Streaming responses. Replace
generate_answerwith a streaming call so users see tokens incrementally. Parse citations on the fly or buffer until the first citation appears. - Fallback and routing. If the primary embedding provider or generator is degraded, fail over. An inference gateway that honors routing directives and forwards provider cache-control hints can automate this without custom code per provider.
- Observability. Log every query with its retrieved chunk IDs, rerank scores, latency breakdown, and user feedback (thumbs up/down). Build a dashboard that alerts when citation precision drops below threshold.
# streaming_generate.py
from openai import OpenAI
client = OpenAI(api_key="YOUR_KEY")
def stream_answer(question: str, hits: list[dict], model: str = "gpt-4o-mini"):
context = build_context(hits)
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
]
stream = client.chat.completions.create(
model=model,
messages=messages,
temperature=0.0,
max_tokens=512,
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield deltaVerify: Hook the streamer into a simple CLI or web endpoint. Confirm tokens appear incrementally and the final assembled answer still contains valid citations. Load-test with 10 concurrent requests — p95 latency should stay within your SLA.
Common failure modes and fixes
| Symptom | Likely cause | Fix |
|---|---|---|
Model cites [3] but only 2 chunks retrieved |
Hallucinated citation | Enforce citation validation post-generation; retry if invalid |
| Answer contradicts cited chunk | Generator ignored context | Strengthen system prompt; lower temperature; try larger model |
| Relevant chunk not retrieved | Chunking split key info across boundaries | Increase overlap; use semantic chunking (e.g., SemanticChunker) |
| Latency spikes on first query | Cold Chroma / model load | Warm the collection on startup; keep embedding model in memory |
| Citation precision drops after corpus update | Stale index | Rebuild index incrementally or on schedule; version collections |
What to tackle next
- Hybrid search: Combine BM25 (keyword) with vector search using reciprocal rank fusion. Captures exact IDs, error codes, and acronyms that embeddings miss.
- Query rewriting: Expand user queries with synonyms, hypothetical document embeddings (HyDE), or an LLM-generated sub-question decomposition.
- Agentic retrieval: Let the model decide when to search, what filters to apply, and whether to re-search after seeing initial results.
- Fine-tuned embeddings: If domain vocabulary is specialized (legal, medical, code), fine-tune a BERT-style encoder on in-domain pairs. Often yields 5–10 point recall gains.
RAG grounding LLM answers works when you treat retrieval as a first-class engineering problem — not an afterthought. The pipeline above is deliberately minimal: no framework lock-in, no hidden control flow, every component swappable. Start there, measure relentlessly, and add complexity only when the eval numbers demand it.