Production RAG
RAG Production Project â End-to-End Pipeline
Build a complete, production-ready RAG system from scratch. This tutorial covers every step: document ingestion, intelligent chunking, embedding, vector storage, retrieval, reranking, generation, and evaluation.
- Document Pipeline â Load PDFs, HTML, and markdown with automatic parsing
- Smart Chunking â Semantic splitting with overlap for optimal retrieval
- Hybrid Search â Combine vector similarity with keyword search
- Production Eval â Measure faithfulness, relevance, and answer quality
"The difference between a demo and a product is the pipeline between them."
RAG Production Project
This is a complete, runnable RAG system built with LangChain. Every component is production-ready with error handling, logging, and evaluation.
System Architecture
Phase 1: Document Ingestion
from langchain_community.document_loaders import (
PyPDFLoader, DirectoryLoader,
WebBaseLoader, TextLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.documents import Document
import hashlib
from pathlib import Path
class DocumentIngestor:
"""Production document ingestion pipeline."""
def __init__(self, chunk_size=500, chunk_overlap=50):
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", ". ", " ", ""],
length_function=len,
)
self.vectorstore = None
def load_directory(self, path: str) -> list[Document]:
"""Load all documents from a directory."""
loaders = {
".pdf": lambda p: DirectoryLoader(p, glob="**/*.pdf", loader_cls=PyPDFLoader),
".md": lambda p: DirectoryLoader(p, glob="**/*.md", loader_cls=TextLoader),
".html": lambda p: DirectoryLoader(p, glob="**/*.html", loader_cls=WebBaseLoader),
}
all_docs = []
dir_path = Path(path)
for ext, loader_fn in loaders.items():
try:
loader = loader_fn(str(dir_path))
docs = loader.load()
all_docs.extend(docs)
print(f"Loaded {len(docs)} {ext} files")
except Exception as e:
print(f"Error loading {ext}: {e}")
return all_docs
def add_metadata(self, docs: list[Document]) -> list[Document]:
"""Add useful metadata to documents."""
for doc in docs:
# Source file hash for deduplication
content_hash = hashlib.md5(doc.page_content.encode()).hexdigest()
doc.metadata["content_hash"] = content_hash
# Chunk index
doc.metadata["chunk_index"] = len(all_docs) if 'all_docs' in dir() else 0
# Word count
doc.metadata["word_count"] = len(doc.page_content.split())
return docs
def chunk_documents(self, docs: list[Document]) -> list[Document]:
"""Split documents into optimal chunks."""
chunks = self.splitter.split_documents(docs)
print(f"Split {len(docs)} docs into {len(chunks)} chunks")
# Add chunk metadata
for i, chunk in enumerate(chunks):
chunk.metadata["chunk_id"] = f"chunk_{i}"
chunk.metadata["chunk_size"] = len(chunk.page_content)
return chunks
def ingest(self, path: str) -> Chroma:
"""Full ingestion pipeline."""
print("Loading documents...")
docs = self.load_directory(path)
print("Adding metadata...")
docs = self.add_metadata(docs)
print("Chunking documents...")
chunks = self.chunk_documents(docs)
print("Creating vector store...")
self.vectorstore = Chroma.from_documents(
chunks, self.embeddings,
collection_name="knowledge_base",
persist_directory="./chroma_db"
)
print(f"Ingested {len(chunks)} chunks into vector store")
return self.vectorstore
# Usage
ingestor = DocumentIngestor(chunk_size=500, chunk_overlap=50)
vectorstore = ingestor.ingest("./documents/")
Phase 2: Hybrid Retrieval
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
class HybridRetriever:
"""Combine dense vector search with sparse keyword search."""
def __init__(self, vectorstore: Chroma, docs: list):
self.vectorstore = vectorstore
# Dense retriever (vector similarity)
self.dense_retriever = vectorstore.as_retriever(
search_type="mmr", # Maximal marginal relevance
search_kwargs={"k": 5, "fetch_k": 20}
)
# Sparse retriever (BM25 keywords)
self.sparse_retriever = BM25Retriever.from_documents(
docs, k=5
)
# Ensemble: 60% dense + 40% sparse
self.ensemble = EnsembleRetriever(
retrievers=[self.dense_retriever, self.sparse_retriever],
weights=[0.6, 0.4]
)
def retrieve(self, query: str) -> list:
"""Retrieve relevant documents."""
return self.ensemble.invoke(query)
def retrieve_with_scores(self, query: str) -> list[dict]:
"""Retrieve with relevance scores."""
results = self.ensemble.invoke(query)
return [
{
"content": doc.page_content,
"metadata": doc.metadata,
"relevance_score": getattr(doc, "relevance_score", None)
}
for doc in results
]
# --- Query Rewriter ---
class QueryRewriter:
"""Improve retrieval with query transformation."""
def __init__(self):
self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def rewrite(self, query: str) -> str:
"""Rewrite query for better retrieval."""
prompt = ChatPromptTemplate.from_template(
"Rewrite this query to be more specific and searchable: {query}"
)
chain = prompt | self.llm | StrOutputParser()
return chain.invoke({"query": query})
def generate_sub_questions(self, query: str) -> list[str]:
"""Break complex query into sub-questions."""
prompt = ChatPromptTemplate.from_template(
"Break this into 3 specific sub-questions:\n{query}"
)
chain = prompt | self.llm | StrOutputParser()
result = chain.invoke({"query": query})
return [q.strip() for q in result.split("\n") if q.strip()]
Phase 3: RAG Chain with Citations
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
class RAGChain:
"""Production RAG chain with citations."""
def __init__(self, retriever: HybridRetriever):
self.retriever = retriever
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
def format_docs(self, docs) -> str:
"""Format retrieved documents with citations."""
formatted = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "Unknown")
page = doc.metadata.get("page", "N/A")
formatted.append(
f"[{i}] Source: {source}, Page: {page}\n{doc.page_content}"
)
return "\n\n".join(formatted)
def build_chain(self):
"""Build the RAG chain."""
prompt = ChatPromptTemplate.from_template("""
You are a helpful assistant. Use ONLY the provided context to answer the question.
If the context doesn't contain enough information, say so.
Always cite your sources using [1], [2], etc.
Context:
{context}
Question: {question}
Answer:""")
chain = (
{"context": self.retriever.ensemble | self.format_docs, "question": RunnablePassthrough()}
| prompt
| self.llm
| StrOutputParser()
)
return chain
def query(self, question: str) -> dict:
"""Query the RAG system."""
chain = self.build_chain()
answer = chain.invoke(question)
# Get sources for citation
docs = self.retriever.retrieve(question)
sources = [
{"content": doc.page_content[:200], "source": doc.metadata.get("source")}
for doc in docs
]
return {"answer": answer, "sources": sources}
# Usage
retriever = HybridRetriever(vectorstore, docs)
rag = RAGChain(retriever)
result = rag.query("What is the main difference between RAG and fine-tuning?")
print(result["answer"])
print("Sources:", result["sources"])
Phase 4: Evaluation with Ragas
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
def evaluate_rag_system(rag_chain, test_questions):
"""Evaluate RAG system with standard metrics."""
# Run test questions
test_data = {
"question": [],
"answer": [],
"contexts": [],
"ground_truth": [],
}
for q in test_questions:
result = rag_chain.query(q["question"])
test_data["question"].append(q["question"])
test_data["answer"].append(result["answer"])
test_data["contexts"].append([s["content"] for s in result["sources"]])
test_data["ground_truth"].append(q["ground_truth"])
# Create dataset
dataset = Dataset.from_dict(test_data)
# Evaluate
result = evaluate(
dataset,
metrics=[
faithfulness, # Is answer grounded in context?
answer_relevancy, # Does answer address the question?
context_precision, # Are retrieved contexts relevant?
context_recall, # Did we retrieve all needed info?
],
)
return result
# Test dataset
test_questions = [
{
"question": "What is RAG?",
"ground_truth": "Retrieval-Augmented Generation combines LLMs with external knowledge retrieval."
},
{
"question": "How does chunking affect retrieval quality?",
"ground_truth": "Chunk size and overlap determine how well semantic units are preserved."
},
]
# Run evaluation
metrics = evaluate_rag_system(rag, test_questions)
print(f"Faithfulness: {metrics['faithfulness']:.2f}")
print(f"Answer Relevancy: {metrics['answer_relevancy']:.2f}")
print(f"Context Precision: {metrics['context_precision']:.2f}")
print(f"Context Recall: {metrics['context_recall']:.2f}")
Phase 5: Production Deployment
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_core.messages import HumanMessage
import uvicorn
import logging
app = FastAPI(title="RAG API", version="1.0.0")
logging.basicConfig(level=logging.INFO)
class QueryRequest(BaseModel):
question: str
max_sources: int = 5
class QueryResponse(BaseModel):
answer: str
sources: list[dict]
latency_ms: float
@app.post("/query", response_model=QueryResponse)
async def query_rag(request: QueryRequest):
"""Query the RAG system."""
import time
start = time.time()
try:
result = rag.query(request.question)
latency = (time.time() - start) * 1000
logging.info(f"Query: {request.question[:50]}... | Latency: {latency:.0f}ms")
return QueryResponse(
answer=result["answer"],
sources=result["sources"][:request.max_sources],
latency_ms=latency,
)
except Exception as e:
logging.error(f"Error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health():
return {"status": "healthy", "vector_store": "connected"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
Interview Prep
Q: How do you choose the right chunking strategy?
Answer:
- Fixed-size (500 tokens): Simple, works well for uniform documents
- Recursive (by headers): Preserves document structure
- Semantic: Best for topic-heavy content (uses embedding similarity)
- Parent-child: Retrieve small chunks, return larger context
- Rule of thumb: Start with 500 tokens + 50 overlap, then tune based on eval metrics
Q: How do you handle hallucination in RAG?
Answer:
- Faithfulness prompting: "Use ONLY the provided context"
- Citation requirements: Force source attribution
- Context precision: Rerank retrieved docs
- Self-RAG: Let the model decide when to retrieve
- Evaluation: Use Ragas faithfulness metric to measure
Q: When would you use hybrid search vs pure vector search?
Answer:
- Pure vector: Semantic similarity, fuzzy queries ("explain like I'm five")
- Hybrid (vector + BM25): When keywords matter ("Python 3.12 features")
- BM25 only: Exact keyword matching (error codes, IDs)
- Recommended: Start with hybrid (60% vector + 40% BM25), tune weights
Q: How do you measure RAG quality in production?
Answer: Use a combination of:
- Offline metrics (Ragas): faithfulness, relevancy, precision, recall
- Online metrics: User satisfaction, click-through rate, follow-up rate
- Latency: p50 < 2s, p95 < 5s
- Cost: Cost per query (embedding + generation + retrieval)
- A/B testing: Compare different retrieval strategies