Picture this: your company has just deployed an internal chatbot powered by GPT-4. The first week is impressive. Then a product manager asks it about the pricing structure you updated last quarter, and it confidently returns the old numbers. A compliance officer asks about the updated data retention policy, and the model hallucinates a policy that never existed. By week three, engineers have stopped trusting it. The chatbot becomes a cautionary tale in your next all-hands.
This is the failure mode that Retrieval-Augmented Generation (RAG) was designed to solve — and it's not just about stale training data. It's about the fundamental tension between what large language models are good at (reasoning, synthesis, language generation) and what they're structurally incapable of (knowing your internal documents, your proprietary data, or anything that happened after their training cutoff). RAG bridges that gap by making retrieval a first-class citizen in your inference pipeline. Instead of asking a model to recall facts from parametric memory, you retrieve the relevant facts at query time and inject them into the prompt. The model becomes a reasoning engine over evidence you supply.
By the end of this lesson, you'll be able to design, build, and evaluate a production-grade RAG pipeline. You'll understand every layer of the stack — from document ingestion and chunking strategy through vector database selection, retrieval tuning, and context assembly — well enough to make deliberate architectural decisions rather than cargo-culting tutorials. You'll also understand where RAG breaks down and what to reach for when it does.
What you'll learn:
This lesson assumes you're comfortable with Python, have worked with LLM APIs (OpenAI or equivalent), and understand what embeddings are conceptually — that is, you know a vector is a numerical representation of meaning and that similar vectors represent semantically similar content. You should also have basic familiarity with API authentication, environment variables, and working in a virtual environment. No prior RAG experience is required, but you'll get the most out of this lesson if you've felt the pain of a model confidently answering questions with wrong information.
Before we write a single line of code, we need an honest mental model of what a RAG system actually is. The term gets thrown around loosely, so let's be precise.
A RAG pipeline has two major phases: indexing (which runs offline or incrementally) and retrieval-augmented inference (which runs at query time). Here's what each phase involves:
Indexing phase:
Inference phase:
This sounds simple, and in toy demos it is. The complexity emerges at scale — when your document corpus is 500,000 chunks, when documents get updated daily, when users ask multi-hop questions that require synthesizing information across three different source documents, and when your legal team needs to know exactly which document a claim came from.
Let's build this up properly.
The quality of your RAG system is bounded above by the quality of your ingested content. Garbage in, garbage out applies here with unusual severity, because the retrieval step can't compensate for missing or malformed content.
For enterprise workflows, your documents will come from heterogeneous sources. LangChain's document loaders are a practical starting point, but it's worth knowing what they're doing under the hood so you can override or replace them when necessary.
from langchain.document_loaders import (
PyPDFLoader,
UnstructuredWordDocumentLoader,
ConfluenceLoader,
S3FileLoader,
)
from langchain.schema import Document
import os
def load_policy_documents(policy_dir: str) -> list[Document]:
"""
Load all policy documents from a directory.
Returns a list of LangChain Document objects.
"""
documents = []
for filename in os.listdir(policy_dir):
filepath = os.path.join(policy_dir, filename)
if filename.endswith(".pdf"):
loader = PyPDFLoader(filepath)
elif filename.endswith(".docx"):
loader = UnstructuredWordDocumentLoader(filepath)
else:
continue
loaded_docs = loader.load()
# Attach source metadata — this will be critical for citations later
for doc in loaded_docs:
doc.metadata["source_file"] = filename
doc.metadata["ingestion_timestamp"] = "2024-10-15T09:00:00Z"
documents.extend(loaded_docs)
return documents
Notice we're attaching metadata at ingest time. This is not optional housekeeping — it's how you'll enable source attribution, access control filtering, and incremental re-indexing later.
Real enterprise documents are not clean markdown. They contain headers, footers, page numbers, legal boilerplate repeated on every page, navigation menus embedded in HTML exports, and tables that don't survive plaintext extraction intact.
You have three choices when you encounter this:
Option 1: Accept the noise. Sometimes the retrieval model is robust enough that footer text ("Page 3 of 47 | CONFIDENTIAL") doesn't meaningfully affect retrieval quality. This is often fine for large corpora where the relevant content swamps the noise.
Option 2: Pre-process aggressively. Write targeted cleaning functions for each document type. This is expensive to maintain but critical for structured content like financial reports or regulatory filings.
Option 3: Use a structured extraction pipeline. Tools like Unstructured.io or Azure Document Intelligence can parse PDFs into structured elements (titles, narrative text, tables, figures) that you treat as first-class chunks. Tables in particular are notoriously hard to embed well as plaintext — a table row like "Q3 Revenue | $4.2M | +12% YoY" loses its meaning without its column headers.
def clean_pdf_text(raw_text: str) -> str:
"""
Remove common PDF artifacts that pollute embedding quality.
"""
import re
# Remove page number patterns like "Page 3 of 47"
text = re.sub(r"Page \d+ of \d+", "", raw_text)
# Remove excessive whitespace and newlines
text = re.sub(r"\n{3,}", "\n\n", text)
text = re.sub(r" {2,}", " ", text)
# Remove running headers/footers (company-specific — you'll need to customize)
text = re.sub(r"ACME CORP CONFIDENTIAL \| Q[1-4] \d{4}", "", text)
return text.strip()
Warning: Be careful about removing content you think is noise. Legal document boilerplate sometimes contains binding language. Table headers stripped from their rows are useless. Always validate your cleaning functions against a representative sample of actual user queries.
Chunking is the single most underappreciated decision in RAG system design, and it's where most tutorial-driven implementations fail in production. Let's understand why chunking matters before discussing strategy.
An embedding model compresses text into a fixed-size vector (typically 768 or 1536 dimensions). If you embed a 10,000-word document as a single chunk, that vector must represent everything in the document. When you retrieve it for a narrow query, you're retrieving a lot of irrelevant content along with what you need. If you chunk too small (individual sentences), you lose the context that gives each sentence meaning — a sentence like "This supersedes all previous agreements" is meaningless without knowing what document it came from and what agreements it refers to.
Fixed-size chunking splits text every N characters or tokens, optionally with overlap. It's simple, deterministic, and works reasonably well for homogeneous documents.
from langchain.text_splitter import RecursiveCharacterTextSplitter
def create_fixed_chunks(
documents: list,
chunk_size: int = 512,
chunk_overlap: int = 64
) -> list:
"""
Split documents into overlapping fixed-size chunks.
chunk_size of 512 tokens is a practical starting point.
chunk_overlap of 64 ensures context isn't lost at boundaries.
"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_documents(documents)
# Preserve and augment metadata
for i, chunk in enumerate(chunks):
chunk.metadata["chunk_index"] = i
chunk.metadata["chunk_size"] = len(chunk.page_content)
return chunks
The RecursiveCharacterTextSplitter is smarter than it looks — it tries to split at paragraph boundaries first, then sentences, then words, only falling back to arbitrary character cuts as a last resort. This means your chunks tend to end at semantically reasonable boundaries.
Semantic chunking uses an embedding model to detect topic shifts and split at semantic boundaries rather than character counts. This produces more coherent chunks but adds latency to the indexing pipeline and can be inconsistent.
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
def create_semantic_chunks(documents: list) -> list:
"""
Split documents at semantic boundaries using embedding similarity.
More expensive than fixed-size but produces more coherent chunks
for narrative documents like policy text or research reports.
"""
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
splitter = SemanticChunker(
embeddings=embeddings,
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=95, # Split when similarity drops below 95th percentile
)
return splitter.split_documents(documents)
This is the architecture pattern that most production RAG systems eventually converge on. The idea is:
You retrieve based on child chunks (better semantic precision), but you inject the parent chunk into the prompt (better context for the LLM). This is sometimes called the "Small-to-Big" or "child-to-parent" retrieval pattern.
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
def build_parent_child_retriever(documents: list):
"""
Build a retriever that indexes child chunks but returns parent chunks.
This gives you the precision of small-chunk retrieval with the
context richness of large-chunk injection.
"""
# Parent splitter — large chunks stored in docstore
parent_splitter = RecursiveCharacterTextSplitter(
chunk_size=2000,
chunk_overlap=200,
)
# Child splitter — small chunks used for embedding and retrieval
child_splitter = RecursiveCharacterTextSplitter(
chunk_size=200,
chunk_overlap=20,
)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
collection_name="policy_documents",
embedding_function=embeddings,
)
# InMemoryStore works for dev; swap for Redis or a persistent KV store in prod
docstore = InMemoryStore()
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=docstore,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
retriever.add_documents(documents)
return retriever
Tip: The parent-child pattern is particularly valuable for regulatory and compliance documents where a specific clause only makes sense in the context of the surrounding section. Retrieving just the clause gives the LLM insufficient context to reason about it correctly.
Embeddings are the bridge between natural language and vector space, and your choice of embedding model has a direct impact on retrieval quality. The key properties you care about:
paraphrase-multilingual-mpnet-base-v2).from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
def get_embedding_model(model_type: str = "openai"):
"""
Return an embedding model appropriate for the use case.
"""
if model_type == "openai":
return OpenAIEmbeddings(
model="text-embedding-3-small",
dimensions=1536, # Can reduce to 512 with MRL — saves cost, minor quality drop
)
elif model_type == "local":
# Good choice for air-gapped environments or sensitive data
return HuggingFaceEmbeddings(
model_name="BAAI/bge-large-en-v1.5",
model_kwargs={"device": "cuda"},
encode_kwargs={"normalize_embeddings": True},
)
elif model_type == "multilingual":
return HuggingFaceEmbeddings(
model_name="sentence-transformers/paraphrase-multilingual-mpnet-base-v2",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True},
)
raise ValueError(f"Unknown model type: {model_type}")
One subtle issue worth understanding: most embedding models are trained for symmetric similarity (comparing two passages). But in RAG, you're comparing a short query to longer document chunks. This asymmetry can hurt retrieval quality.
Models like BAAI/bge-large-en-v1.5 address this with instruction-prefixed embeddings — you prefix queries with "Represent this sentence: Query: " and documents with a different prefix. This signals to the model that the embedding is being used asymmetrically.
def embed_for_retrieval(texts: list[str], is_query: bool = False) -> list[list[float]]:
"""
Embed text with appropriate prefix for asymmetric retrieval.
BGE models use instruction prefixes to improve cross-encoder alignment.
"""
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-large-en-v1.5")
if is_query:
# Query instruction tells model to expect retrieval context
texts = [f"Represent this sentence for searching relevant passages: {t}"
for t in texts]
return model.encode(texts, normalize_embeddings=True).tolist()
Your vector store is doing two jobs simultaneously: storing vectors for ANN search and acting as a metadata filter engine. Most developers optimize for the first and ignore the second until it causes production incidents.
| Vector Store | Best For | Watch Out For |
|---|---|---|
| Chroma | Local dev, small corpora | Doesn't scale past ~1M vectors well |
| Pinecone | Managed, enterprise scale | Cost at high query volume, vendor lock-in |
| Weaviate | Hybrid search (vector + keyword) | Complex self-hosted ops |
| pgvector | Teams already on Postgres | Slower ANN than dedicated stores at scale |
| Qdrant | High performance, self-hosted | Less ecosystem maturity |
| Redis | Low latency, existing Redis infra | Limited metadata filtering |
For enterprise workflows where you're already running Postgres (and most enterprises are), pgvector is worth serious consideration. You get ACID transactions, row-level security for access control, and the ability to JOIN vector search results with your existing relational data.
from langchain_community.vectorstores.pgvector import PGVector
from langchain_openai import OpenAIEmbeddings
CONNECTION_STRING = "postgresql+psycopg2://user:password@localhost:5432/enterprise_rag"
def build_pgvector_store(chunks: list, collection_name: str = "policy_docs"):
"""
Build a vector store in Postgres using pgvector.
Advantages over Chroma for enterprise:
- Row-level security can enforce document access permissions
- Atomic updates when documents are revised
- Query results can be JOINed with user permission tables
"""
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = PGVector.from_documents(
documents=chunks,
embedding=embeddings,
collection_name=collection_name,
connection_string=CONNECTION_STRING,
pre_delete_collection=False, # Set True to rebuild from scratch
)
return vectorstore
Pure vector similarity works for homogeneous corpora. In enterprise environments, you almost always need to filter by metadata alongside vector search — for example, "only return documents from the Legal department" or "only retrieve content with an effective date after 2023-01-01."
def retrieve_with_metadata_filter(
vectorstore,
query: str,
department: str = None,
effective_after: str = None,
k: int = 5
) -> list:
"""
Perform metadata-filtered vector retrieval.
Metadata filters run BEFORE ANN search in most vector stores,
which means filtering to a small subset can significantly
reduce recall. Size your filtered subsets carefully.
"""
filter_dict = {}
if department:
filter_dict["department"] = department
if effective_after:
filter_dict["effective_date"] = {"$gte": effective_after}
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={
"k": k,
"filter": filter_dict if filter_dict else None,
}
)
return retriever.get_relevant_documents(query)
Warning: When you apply a metadata filter that reduces your searchable corpus to a small subset, ANN search quality degrades. If you're filtering to 200 documents out of 500,000, the ANN index was built for the full corpus — you may get worse results than a brute-force search over the filtered set. Pinecone, Weaviate, and Qdrant handle this differently; understand your vector store's filter-then-search vs. search-then-filter behavior.
The default "embed query, find nearest vectors, return top-5" approach works surprisingly well, but it has well-documented failure modes. Let's cover the advanced retrieval techniques that address them.
Dense retrieval (embeddings) is excellent at semantic matching but struggles with exact term matching. If a user asks "What does clause 8.3.2 say?", the semantic embedding of that query doesn't capture the critical information ("clause 8.3.2"). Sparse retrieval (BM25, TF-IDF) handles exact keyword matches but misses paraphrase and synonym relationships.
The production-grade solution is hybrid search: run both dense and sparse retrieval, then merge the result sets using Reciprocal Rank Fusion (RRF).
from langchain.retrievers import BM25Retriever, EnsembleRetriever
def build_hybrid_retriever(chunks: list, vectorstore, k: int = 5):
"""
Combine dense vector retrieval with sparse BM25 retrieval.
RRF fusion score: 1 / (rank + 60) summed across retrievers.
The 60 is a smoothing constant — lower values favor top-ranked results more.
"""
# Sparse retriever for exact keyword matching
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = k
# Dense retriever for semantic similarity
dense_retriever = vectorstore.as_retriever(
search_kwargs={"k": k}
)
# Ensemble with equal weights — tune weights based on your query distribution
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, dense_retriever],
weights=[0.5, 0.5],
)
return ensemble_retriever
LLMs can preprocess queries to improve retrieval before the actual search happens. Two patterns here:
HyDE (Hypothetical Document Embeddings): Ask the LLM to generate a hypothetical answer to the query, then embed that hypothetical answer and use it for retrieval. The idea is that an answer-shaped text is more similar to relevant document chunks than a question-shaped text.
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
def hyde_query_expansion(query: str) -> str:
"""
Generate a hypothetical answer to use as a retrieval query.
HyDE works well for factual queries where the answer format
is predictable. It can hurt for exploratory or ambiguous queries.
"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
prompt = PromptTemplate(
input_variables=["query"],
template="""Write a detailed answer to the following question as if you were
an expert reviewing internal company documentation. Do not hedge or qualify.
Write in the style of a policy document excerpt.
Question: {query}
Hypothetical answer excerpt:"""
)
chain = prompt | llm
hypothetical_doc = chain.invoke({"query": query}).content
return hypothetical_doc
Multi-query retrieval: Rewrite the original query into 3-5 different formulations, retrieve for each, and deduplicate. This compensates for the fact that your query might miss chunks that would be retrieved if phrased differently.
from langchain.retrievers.multi_query import MultiQueryRetriever
def build_multiquery_retriever(vectorstore, llm):
"""
Automatically generate query variants to improve recall.
Useful when users write queries in varied ways or when
the right terminology isn't obvious from the query alone.
"""
retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
llm=llm,
)
return retriever
Even with good retrieval, your top-5 chunks may not all be relevant. A cross-encoder reranker takes each retrieved chunk and scores it against the query using a more expensive but more accurate model than the bi-encoder used for retrieval. You retrieve broadly (top-20) and rerank to select the top-5 to inject into your prompt.
from sentence_transformers import CrossEncoder
def rerank_chunks(
query: str,
chunks: list,
top_n: int = 5,
model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
) -> list:
"""
Rerank retrieved chunks using a cross-encoder model.
Cross-encoders are ~10x more accurate than bi-encoders at relevance
scoring because they see both query and document simultaneously,
but they're too slow to use across your full corpus at query time.
Retrieve broadly with bi-encoder, rerank narrowly with cross-encoder.
"""
reranker = CrossEncoder(model_name)
# Create query-chunk pairs for the cross-encoder
pairs = [(query, chunk.page_content) for chunk in chunks]
# Score all pairs
scores = reranker.predict(pairs)
# Sort chunks by score descending and return top_n
scored_chunks = list(zip(chunks, scores))
scored_chunks.sort(key=lambda x: x[1], reverse=True)
return [chunk for chunk, score in scored_chunks[:top_n]]
Once you have your retrieved chunks, you have to turn them into a prompt that gives the LLM the best chance of producing an accurate, grounded response. This is less about prompt magic and more about information architecture.
Every LLM has a context window limit. In a RAG prompt, you're competing for that budget across: system instructions, retrieved context, conversation history, and the user query. Be deliberate.
def assemble_rag_prompt(
query: str,
retrieved_chunks: list,
conversation_history: list[dict] = None,
max_context_tokens: int = 3000,
) -> dict:
"""
Assemble a structured prompt from retrieved chunks.
Enforces a token budget to prevent context overflow.
Places most relevant content first (primacy effect).
"""
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")
system_prompt = """You are an internal knowledge assistant for ACME Corporation.
Answer questions using ONLY the information provided in the context sections below.
If the context does not contain sufficient information to answer the question,
say so explicitly — do not infer or extrapolate beyond the provided content.
Always cite the source document name when you use information from the context."""
# Build context string within token budget
context_parts = []
used_tokens = 0
for i, chunk in enumerate(retrieved_chunks):
source = chunk.metadata.get("source_file", "Unknown")
chunk_text = f"[Source: {source}]\n{chunk.page_content}"
chunk_tokens = len(enc.encode(chunk_text))
if used_tokens + chunk_tokens > max_context_tokens:
break
context_parts.append(chunk_text)
used_tokens += chunk_tokens
context_string = "\n\n---\n\n".join(context_parts)
user_message = f"""Context:
{context_string}
Question: {query}
Answer:"""
messages = [
{"role": "system", "content": system_prompt},
]
if conversation_history:
messages.extend(conversation_history[-4:]) # Keep last 2 turns
messages.append({"role": "user", "content": user_message})
return messages
Generic instructions like "only answer from the context" don't work reliably. LLMs will still confabulate when the context is ambiguous or incomplete. More effective patterns:
Here's a production-ready RAG pipeline class that integrates the components we've built:
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.retrievers import EnsembleRetriever, BM25Retriever
import tiktoken
import logging
logger = logging.getLogger(__name__)
class EnterpriseRAGPipeline:
"""
Production-grade RAG pipeline for enterprise knowledge bases.
Design decisions made explicit:
- Hybrid retrieval (dense + sparse) for robustness
- Cross-encoder reranking for precision
- Token-budgeted context assembly
- Structured metadata for citation and filtering
"""
def __init__(
self,
vectorstore: Chroma,
chunks: list,
llm_model: str = "gpt-4o",
embedding_model: str = "text-embedding-3-small",
retrieval_k: int = 20,
rerank_top_n: int = 5,
):
self.vectorstore = vectorstore
self.chunks = chunks
self.llm = ChatOpenAI(model=llm_model, temperature=0.0)
self.retrieval_k = retrieval_k
self.rerank_top_n = rerank_top_n
self.encoder = tiktoken.encoding_for_model(llm_model)
# Build hybrid retriever
bm25 = BM25Retriever.from_documents(chunks)
bm25.k = retrieval_k
dense = vectorstore.as_retriever(search_kwargs={"k": retrieval_k})
self.retriever = EnsembleRetriever(
retrievers=[bm25, dense],
weights=[0.4, 0.6],
)
# Load cross-encoder for reranking
from sentence_transformers import CrossEncoder
self.reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def retrieve_and_rerank(self, query: str) -> list:
"""Retrieve broadly, then rerank to top-n."""
candidates = self.retriever.get_relevant_documents(query)
if not candidates:
logger.warning(f"No candidates retrieved for query: {query[:50]}")
return []
# Rerank
pairs = [(query, c.page_content) for c in candidates]
scores = self.reranker.predict(pairs)
scored = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, score in scored[:self.rerank_top_n]]
def assemble_context(self, chunks: list, max_tokens: int = 3000) -> str:
"""Build a token-budgeted context string."""
parts = []
used = 0
for chunk in chunks:
source = chunk.metadata.get("source_file", "Unknown Document")
section = chunk.metadata.get("section", "")
header = f"[Source: {source}" + (f", Section: {section}" if section else "") + "]"
text = f"{header}\n{chunk.page_content}"
tokens = len(self.encoder.encode(text))
if used + tokens > max_tokens:
logger.info(f"Context budget reached. Using {len(parts)} of {len(chunks)} chunks.")
break
parts.append(text)
used += tokens
return "\n\n---\n\n".join(parts)
def query(
self,
user_query: str,
department_filter: str = None,
conversation_history: list = None,
) -> dict:
"""
End-to-end RAG query with structured response.
Returns answer text and source citations.
"""
# Retrieve with optional metadata filter
if department_filter:
filtered_retriever = self.vectorstore.as_retriever(
search_kwargs={
"k": self.retrieval_k,
"filter": {"department": department_filter}
}
)
candidates = filtered_retriever.get_relevant_documents(user_query)
else:
candidates = self.retriever.get_relevant_documents(user_query)
# Rerank
pairs = [(user_query, c.page_content) for c in candidates]
scores = self.reranker.predict(pairs)
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
top_chunks = [doc for doc, _ in ranked[:self.rerank_top_n]]
# Assemble prompt
context = self.assemble_context(top_chunks)
system = """You are a knowledgeable assistant with access to internal company documentation.
Answer the user's question using only the provided context.
If the context is insufficient, say so clearly.
Always cite source documents by name."""
messages = [{"role": "system", "content": system}]
if conversation_history:
messages.extend(conversation_history[-4:])
messages.append({
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {user_query}"
})
response = self.llm.invoke(messages)
# Extract unique sources cited
sources = list({c.metadata.get("source_file", "Unknown") for c in top_chunks})
return {
"answer": response.content,
"sources": sources,
"chunks_used": len(top_chunks),
}
Building a RAG system without evaluation is guesswork. You need to know if your changes (different chunk size, different retrieval strategy, different reranker) actually improve or degrade quality.
RAGAS (Retrieval Augmented Generation Assessment) provides four key metrics:
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
def evaluate_rag_pipeline(
pipeline: EnterpriseRAGPipeline,
test_cases: list[dict],
) -> dict:
"""
Evaluate RAG pipeline quality using RAGAS metrics.
test_cases format:
[
{
"question": "What is the data retention policy for customer records?",
"ground_truth": "Customer records must be retained for 7 years...",
},
...
]
"""
questions = []
answers = []
contexts = []
ground_truths = []
for case in test_cases:
result = pipeline.query(case["question"])
# For RAGAS, we need the raw retrieved texts
retrieved_docs = pipeline.retrieve_and_rerank(case["question"])
context_texts = [doc.page_content for doc in retrieved_docs]
questions.append(case["question"])
answers.append(result["answer"])
contexts.append(context_texts)
ground_truths.append(case.get("ground_truth", ""))
dataset = Dataset.from_dict({
"question": questions,
"answer": answers,
"contexts": contexts,
"ground_truth": ground_truths,
})
results = evaluate(
dataset,
metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)
return results
Don't just evaluate against synthetic questions. Capture real user queries from production, have domain experts annotate good answers, and use that as your gold standard. A test set of 50-100 expert-annotated questions is worth more than 10,000 synthetic ones.
Tip: Set up A/B evaluation when you make pipeline changes. Route 10% of traffic to the new configuration, log query/response pairs, and have a weekly human review cadence. Quantitative metrics will tell you something changed; human review tells you if it changed for the better.
You'll build a RAG pipeline for a fictional company's HR policy documents. You can use the provided sample documents or adapt this to a real document set you have access to.
Setup:
pip install langchain langchain-openai langchain-community langchain-experimental \
sentence-transformers ragas chromadb tiktoken pypdf rank-bm25
Step 1: Create sample policy documents
Create three text files in a policies/ directory:
data_retention_policy.txt — A 500-word document describing data retention rulesremote_work_policy.txt — A 500-word document describing remote work guidelinesexpense_reimbursement_policy.txt — A 500-word document describing expense rulesWrite realistic policy content. Don't use placeholder text — the quality of your test is proportional to the realism of your documents.
Step 2: Build the indexing pipeline
# exercise_rag.py
import os
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.retrievers import BM25Retriever, EnsembleRetriever
# 1. Load documents
documents = []
for filename in os.listdir("policies"):
loader = TextLoader(f"policies/{filename}")
docs = loader.load()
for doc in docs:
doc.metadata["source_file"] = filename
doc.metadata["policy_type"] = filename.replace(".txt", "")
documents.extend(docs)
# 2. Chunk
splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
chunks = splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks from {len(documents)} documents")
# 3. Build vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings, collection_name="hr_policies")
# 4. Build hybrid retriever
bm25 = BM25Retriever.from_documents(chunks)
bm25.k = 5
dense = vectorstore.as_retriever(search_kwargs={"k": 5})
retriever = EnsembleRetriever(retrievers=[bm25, dense], weights=[0.4, 0.6])
# 5. Test retrieval
test_query = "How long do we need to keep employee expense records?"
results = retriever.get_relevant_documents(test_query)
for r in results:
print(f"\nSource: {r.metadata['source_file']}")
print(r.page_content[:200])
Step 3: Add the LLM generation layer
Extend the pipeline to use ChatOpenAI to generate an answer from the retrieved context. Use the prompt structure from the earlier sections.
Step 4: Design 10 test questions
Write 10 questions ranging from straightforward (answerable from a single chunk) to challenging (requiring synthesis across multiple documents, or deliberately unanswerable from the provided docs). Run your pipeline against each and evaluate qualitatively.
Challenge task: Implement the cross-encoder reranking step. Compare the top-5 results from the ensemble retriever before and after reranking. Do the rankings change meaningfully?
Mistake 1: Chunk size mismatch with your embedding model's context window
If your chunks are larger than your embedding model's max token count, the excess gets truncated silently. You'll never get an error — your embeddings will just be incomplete. Fix: validate chunk sizes against your model's limit before indexing.
Mistake 2: Not rebuilding the BM25 index when documents change
BM25 indexes are built from the document corpus in memory. If you add documents to your vector store but don't rebuild the BM25 retriever, your hybrid search will be inconsistent. In production, use a proper incremental update strategy or rebuild nightly.
Mistake 3: Over-relying on top-1 retrieval
Users often ask questions that require information from multiple document sections. Retrieving only 1-2 chunks and expecting the LLM to synthesize a complete answer will produce incomplete responses. Start with top-5 to top-10 before reranking.
Mistake 4: Ignoring context order
LLMs exhibit primacy and recency bias — they pay more attention to content at the beginning and end of the context window. Don't assume the model weights all retrieved chunks equally. Put your highest-scored chunks first and last, with lower-confidence chunks in the middle.
Mistake 5: Using the same embedding model for different languages or domains without validation
A model trained predominantly on English text will produce meaningfully worse embeddings for French, German, or technical jargon. If your corpus includes non-English content or highly specialized terminology (medical, legal, financial), benchmark retrieval quality explicitly rather than assuming your embedding model handles it.
Mistake 6: Not handling the "I don't know" case
Without explicit grounding instructions, LLMs will answer questions even when the retrieved context is insufficient. Users will trust these answers. Test your pipeline with questions that are deliberately outside your document corpus and verify that the model refuses gracefully rather than confabulating.
Mistake 7: Token count budgeting without considering the full prompt
You budget 3000 tokens for context, but you forget that your system prompt is 500 tokens, conversation history is 800 tokens, and the user query is 200 tokens. You've just exceeded a 4K context limit or cut your context budget to 500 tokens without realizing it. Always calculate token budgets against the total prompt, not just the context portion.
You now have the conceptual framework and working code to build production-grade RAG pipelines. Let's recap the key architectural decisions:
What RAG doesn't solve: RAG improves groundedness but it doesn't eliminate hallucination entirely, and it doesn't fix reasoning errors. If a user asks a question that requires multi-step logical inference across five documents, even perfect retrieval won't guarantee a correct answer from the LLM. For complex reasoning tasks, consider agentic RAG (where the LLM iteratively decides what to retrieve) or graph RAG (where relationships between documents are explicitly represented).
Next learning areas:
The gap between a working demo and a production RAG system is wider than most tutorials suggest. What makes the difference is exactly what we covered here: deliberate design decisions at each layer, rigorous evaluation, and understanding the failure modes deeply enough to build around them.
Learning Path: Intro to AI & Prompt Engineering