If you’ve built a RAG system that works in a notebook but falls apart under real traffic, the problem usually isn’t the model — it’s how the RAG pipeline stages connect. Retrieval, augmentation, and generation each have distinct failure modes, and the boundaries between them are where latency spikes, hallucinations sneak in, and costs balloon. This guide walks through each stage with production-grade patterns, not toy examples.
Retrieval: finding the right context
Retrieval is the only stage where you have full control over what the model sees. Get this wrong and nothing downstream can recover.
Choose the right index for your access pattern
Vector search gets all the attention, but it’s not always the right tool. Match the index to your query type:
# Hybrid retrieval: dense + sparse + exact match
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
vectorstore = Chroma(
persist_directory="./chroma_db",
embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})
# BM25 for keyword-heavy queries (error codes, model numbers, proper nouns)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 20
# Combine with reciprocal rank fusion
retriever = EnsembleRetriever(
retrievers=[dense_retriever, bm25_retriever],
weights=[0.6, 0.4]
)Pitfall: Using only dense vectors for queries like “error code 429” or “GPT-4 Turbo pricing.” Dense embeddings collapse these into generic “error” or “pricing” neighborhoods. Always layer a sparse retriever (BM25, SPLADE) or exact-match filter for structured identifiers.
Rerank before you truncate
Sending 20 raw chunks to the generator wastes tokens and introduces noise. A cross-encoder reranker costs ~50ms but dramatically improves precision:
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def retrieve_and_rerank(query: str, k: int = 5, initial_k: int = 20):
docs = retriever.invoke(query)[:initial_k]
pairs = [(query, d.page_content) for d in docs]
scores = reranker.predict(pairs)
ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
return [d for d, _ in ranked[:k]]Tradeoff: Reranking adds latency. For high-throughput paths, cache reranker scores for repeated queries or use a smaller model (MiniLM-L-6-v2 runs ~15ms on CPU). Skip reranking entirely for navigational queries where the top-1 dense result is usually correct.
Filter by metadata before vector search
If your documents have tenant IDs, document versions, or access control tags, filter before ANN search — not after. Post-filtering wastes ANN probes on documents you’ll discard:
# Good: pre-filter in the vector store query
results = vectorstore.similarity_search(
query,
k=10,
filter={"tenant_id": tenant_id, "version": "v2.1"}
)
# Bad: fetch 50 then filter in Python
results = vectorstore.similarity_search(query, k=50)
results = [d for d in results if d.metadata["tenant_id"] == tenant_id]Augmentation: shaping context for the model
Augmentation is where you decide what the model actually sees. This stage determines whether the model answers from your data or hallucinates.
Chunking strategy determines retrieval granularity
Fixed-size chunks (512 tokens, 100 overlap) are the default, but they break semantic boundaries. Consider structure-aware chunking:
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_experimental.text_splitter import SemanticChunker
# For code, legal, or technical docs: split on structural boundaries
code_splitter = RecursiveCharacterTextSplitter.from_language(
language="python",
chunk_size=1000,
chunk_overlap=100
)
# For narrative text: semantic chunking preserves topic coherence
semantic_splitter = SemanticChunker(
OpenAIEmbeddings(model="text-embedding-3-small"),
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=95
)Pitfall: Chunking too small loses context; chunking too large exceeds the model’s effective attention window. For most RAG pipeline stages, 500-1000 token chunks with 10-15% overlap hit the sweet spot. Test with your actual queries — measure recall@k on a labeled eval set.
Include metadata in the context window
The model needs to know where information came from. Inject source metadata inline:
def format_context(docs: list[Document]) -> str:
parts = []
for i, doc in enumerate(docs):
source = doc.metadata.get("source", "unknown")
page = doc.metadata.get("page", "?")
section = doc.metadata.get("section", "")
header = f"[Doc {i+1} | {source} | p.{page}"
if section:
header += f" | {section}"
header += "]"
parts.append(f"{header}\n{doc.page_content}")
return "\n\n---\n\n".join(parts)This lets the model cite sources natively: “According to Doc 3 (pricing.md, p.2), the enterprise tier includes…”
Compress context when you hit token limits
When retrieved context exceeds your budget, don’t just truncate — compress:
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
compressor = CohereRerank(model="rerank-english-v3.0")
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever
)
# Returns compressed, query-relevant passages instead of full chunks
compressed_docs = compression_retriever.invoke(query)Tradeoff: Compression adds a model call (~200-400ms). Use it selectively — only when len(context) > 0.7 * context_window. For n4n.ai users, the gateway’s per-token metering makes this tradeoff visible in real time: you see exactly what compression saves on the generation call.
Generation: constrained synthesis
Generation is where the model produces the final answer. The prompt template and decoding parameters matter more than the model choice.
Use a strict instruction template
RAG_PROMPT = """You are a precise technical assistant. Answer the user's question using ONLY the provided context.
Rules:
- If the context doesn't contain the answer, say "I don't have enough information to answer this."
- Cite sources inline using [Doc N] format.
- Do not use external knowledge.
- Be concise. Prefer bullet points for multi-part answers.
Context:
{context}
Question: {question}
Answer:"""Pitfall: Soft instructions like “use the context when possible” invite hallucination. The model will supplement from training data unless explicitly forbidden. The “I don’t know” rule is your primary hallucination guard.
Control decoding for factual consistency
generation_config = {
"temperature": 0.0, # Deterministic for factual QA
"top_p": 0.1, # Nucleus sampling off
"max_tokens": 512, # Cap output length
"stop": ["\n\nQuestion:", "Context:"], # Prevent runaway
}Temperature 0.0 isn’t optional for RAG. Any stochasticity introduces variance in whether the model sticks to context. If you need “creative” responses, build a separate chain — don’t mix modes.
Stream with citation validation
Streaming improves perceived latency, but you lose the ability to validate the full response before sending. Compromise: stream tokens but validate citations post-hoc:
async def stream_with_validation(chain, inputs):
full_response = ""
async for chunk in chain.astream(inputs):
full_response += chunk
yield chunk
# Validate after stream completes
citations = extract_citations(full_response)
invalid = [c for c in citations if not is_valid_citation(c, inputs["context"])]
if invalid:
# Log for monitoring, optionally trigger fallback
logger.warning(f"Invalid citations detected: {invalid}")Wiring it together: a minimal production pipeline
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI
def build_rag_chain():
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.0,
max_tokens=512,
)
chain = (
{"context": retriever | format_context, "question": RunnablePassthrough()}
| RAG_PROMPT
| llm
)
return chain.with_config(
run_name="rag_pipeline",
tags=["production", "rag"]
)Observability hooks you’ll wish you had
from langchain_core.callbacks import BaseCallbackHandler
class RAGMetricsHandler(BaseCallbackHandler):
def on_retriever_end(self, documents, **kwargs):
latency = kwargs.get("run_id") # track retrieval latency
num_docs = len(documents)
avg_score = sum(d.metadata.get("score", 0) for d in documents) / max(num_docs, 1)
metrics.gauge("rag.retrieval.latency", latency)
metrics.gauge("rag.retrieval.num_docs", num_docs)
metrics.gauge("rag.retrieval.avg_score", avg_score)
def on_llm_end(self, response, **kwargs):
usage = response.llm_output.get("token_usage", {})
metrics.increment("rag.generation.prompt_tokens", usage.get("prompt_tokens", 0))
metrics.increment("rag.generation.completion_tokens", usage.get("completion_tokens", 0))Instrument every stage boundary. You need to know: retrieval latency, reranker latency, tokens sent to generation, tokens generated, and citation validity rate. Without these, you’re debugging blind.
Common failure patterns and fixes
| Symptom | Likely Stage | Fix |
|---|---|---|
| “I don’t know” on answerable questions | Retrieval | Add BM25, check chunking, verify metadata filters |
| Hallucinated details not in sources | Generation | Lower temperature, strengthen “only use context” instruction |
| Correct answer but wrong citation | Augmentation | Include source metadata in context, validate citations post-generation |
| Latency spikes at 95th percentile | Retrieval | Cache embeddings, use smaller reranker, add ANN index |
| Cost grows faster than traffic | Generation | Compress context, cap max_tokens, route simple queries to smaller model |
Evaluation: the stage you’re probably skipping
You cannot improve what you don’t measure. Build a minimal eval harness:
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
eval_dataset = [
{
"question": "What's the rate limit for the enterprise API?",
"ground_truth": "10,000 requests/minute",
"contexts": ["Enterprise tier includes 10,000 RPM rate limit..."],
},
# ... 50+ realistic queries from production logs
]
results = evaluate(
dataset=eval_dataset,
metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
llm=ChatOpenAI(model="gpt-4o", temperature=0),
)
print(results.to_pandas())Run this on every deploy. Track the four RAGAS metrics over time — they correlate with user satisfaction better than any single metric.
The three RAG pipeline stages — retrieval, augmentation, generation — each demand different optimization strategies. Retrieval is about recall and precision tradeoffs. Augmentation is about context engineering: chunking, metadata, compression. Generation is about constraint: strict prompts, deterministic decoding, citation discipline. Treat them as independent services with clear contracts, instrument the boundaries, and you’ll build a RAG system that survives contact with production traffic.