RAG Pipeline Template
WorkflowProduction-ready RAG pipeline with LangChain.
RAG Pipeline Template
Overview
Production-ready RAG (Retrieval-Augmented Generation) pipeline template for building question-answering systems over your own data. This template provides a modular, extensible foundation for implementing retrieval-augmented generation with LangChain.
What is RAG?
Retrieval-Augmented Generation (RAG) is an AI architecture that combines:
- Retrieval: Finding relevant documents from a knowledge base
- Augmentation: Adding retrieved context to the prompt
- Generation: Using an LLM to generate answers based on the context
RAG enables LLMs to answer questions about private data, stay up-to-date with recent information, and provide source citations for their answers.
Template Structure
rag-template/
├── src/
│ ├── __init__.py
│ ├── config.py # Configuration management
│ ├── ingestion/
│ │ ├── __init__.py
│ │ ├── loaders.py # Document loaders
│ │ ├── chunkers.py # Chunking strategies
│ │ └── pipeline.py # Ingestion pipeline
│ ├── retrieval/
│ │ ├── __init__.py
│ │ ├── embeddings.py # Embedding models
│ │ ├── vector_store.py # Vector store setup
│ │ ├── retrievers.py # Retrieval strategies
│ │ └── ranking.py # Re-ranking (optional)
│ ├── generation/
│ │ ├── __init__.py
│ │ ├── prompts.py # Prompt templates
│ │ ├── chains.py # QA chains
│ │ └── conversation.py # Conversational QA
│ └── evaluation/
│ ├── __init__.py
│ ├── metrics.py # Evaluation metrics
│ └── pipeline.py # Evaluation pipeline
├── data/
│ ├── documents/ # Source documents
│ └── processed/ # Processed chunks
├── tests/
│ ├── test_ingestion.py
│ ├── test_retrieval.py
│ └── test_chains.py
├── main.py # Entry point
├── requirements.txt
└── README.md
Installation
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install core dependencies
pip install langchain langchain-core langchain-community langchain-openai
# Install vector store
pip install langchain-chroma # or faiss-cpu, pinecone-client, etc.
# Install optional dependencies
pip install langchain-huggingface # For HuggingFace embeddings
pip install langchain-cohere # For Cohere re-ranking
pip install llama-index # For advanced retrieval
# Or install all
pip install -r requirements.txt
requirements.txt
# Core
langchain>=0.2.0
langchain-core>=0.2.0
langchain-community>=0.2.0
langchain-openai>=0.1.0
# Vector stores
langchain-chroma>=0.1.0
chromadb>=0.4.0
# Embeddings
langchain-huggingface>=0.0.3
# Document processing
pypdf>=4.0.0
python-docx>=1.1.0
markdown>=3.6
beautifulsoup4>=4.12.0
# Evaluation
langsmith>=0.1.0
ragas>=0.1.0
# Utilities
pydantic>=2.0.0
python-dotenv>=1.0.0
typer>=0.9.0
Configuration
Environment Variables (.env)
# OpenAI API
OPENAI_API_KEY=sk-...
OPENAI_API_BASE=https://api.openai.com/v1
# Embedding model
EMBEDDING_MODEL=text-embedding-3-small
EMBEDDING_DIMENSIONS=1536
# LLM model
LLM_MODEL=gpt-4o
LLM_TEMPERATURE=0.0
LLM_MAX_TOKENS=4096
# Vector store
VECTOR_STORE=chroma
CHROMA_PERSIST_DIRECTORY=./chroma_db
# Chunking
CHUNK_SIZE=1000
CHUNK_OVERLAP=200
# Retrieval
SEARCH_TYPE=similarity
SEARCH_K=4
# Logging
LOG_LEVEL=INFO
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=lsv2_...
Configuration Class
# src/config.py
from pydantic import BaseModel, Field
from typing import Literal
import os
from dotenv import load_dotenv
load_dotenv()
class EmbeddingConfig(BaseModel):
model: str = Field(default="text-embedding-3-small", description="Embedding model name")
dimensions: int = Field(default=1536, description="Embedding dimensions")
provider: Literal["openai", "huggingface", "cohere", "google"] = Field(default="openai")
class ChunkingConfig(BaseModel):
strategy: Literal["character", "recursive_character", "token", "semantic"] = Field(default="recursive_character")
chunk_size: int = Field(default=1000, ge=100, le=5000)
chunk_overlap: int = Field(default=200, ge=0, le=500)
class RetrievalConfig(BaseModel):
vector_store: Literal["chroma", "faiss", "pinecone", "qdrant", "weaviate"] = Field(default="chroma")
search_type: Literal["similarity", "mmr", "similarity_score_threshold"] = Field(default="similarity")
search_k: int = Field(default=4, ge=1, le=20)
score_threshold: float = Field(default=0.7, ge=0.0, le=1.0)
class LLMConfig(BaseModel):
model: str = Field(default="gpt-4o", description="LLM model name")
temperature: float = Field(default=0.0, ge=0.0, le=1.0)
max_tokens: int = Field(default=4096, ge=100, le=8192)
provider: Literal["openai", "anthropic", "google", "cohere", "ollama"] = Field(default="openai")
class RAGConfig(BaseModel):
embedding: EmbeddingConfig = Field(default_factory=EmbeddingConfig)
chunking: ChunkingConfig = Field(default_factory=ChunkingConfig)
retrieval: RetrievalConfig = Field(default_factory=RetrievalConfig)
llm: LLMConfig = Field(default_factory=LLMConfig)
@classmethod
def from_env(cls) -> "RAGConfig":
return cls(
embedding=EmbeddingConfig(
model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-small"),
dimensions=int(os.getenv("EMBEDDING_DIMENSIONS", "1536")),
),
chunking=ChunkingConfig(
chunk_size=int(os.getenv("CHUNK_SIZE", "1000")),
chunk_overlap=int(os.getenv("CHUNK_OVERLAP", "200")),
),
retrieval=RetrievalConfig(
search_k=int(os.getenv("SEARCH_K", "4")),
score_threshold=float(os.getenv("SCORE_THRESHOLD", "0.7")),
),
llm=LLMConfig(
model=os.getenv("LLM_MODEL", "gpt-4o"),
temperature=float(os.getenv("LLM_TEMPERATURE", "0.0")),
),
)
config = RAGConfig.from_env()
Document Ingestion
Document Loaders
# src/ingestion/loaders.py
from typing import List
from langchain_core.documents import Document
from langchain_community.document_loaders import (
PyPDFLoader,
Docx2txtLoader,
TextLoader,
UnstructuredMarkdownLoader,
WebBaseLoader,
DirectoryLoader,
)
from pathlib import Path
def load_pdf(file_path: str) -> List[Document]:
"""Load a PDF document."""
loader = PyPDFLoader(file_path)
return loader.load()
def load_docx(file_path: str) -> List[Document]:
"""Load a Word document."""
loader = Docx2txtLoader(file_path)
return loader.load()
def load_text(file_path: str) -> List[Document]:
"""Load a text document."""
loader = TextLoader(file_path, encoding="utf-8")
return loader.load()
def load_markdown(file_path: str) -> List[Document]:
"""Load a Markdown document."""
loader = UnstructuredMarkdownLoader(file_path)
return loader.load()
def load_webpage(url: str) -> List[Document]:
"""Load a webpage."""
loader = WebBaseLoader(url)
return loader.load()
def load_directory(directory: str, recursive: bool = True) -> List[Document]:
"""Load all documents from a directory."""
loader = DirectoryLoader(
directory,
glob="**/*.{pdf,docx,txt,md}",
recursive=recursive,
show_progress=True,
)
return loader.load()
def load_documents(sources: List[str]) -> List[Document]:
"""Load documents from various sources."""
documents = []
for source in sources:
path = Path(source)
if path.is_file():
suffix = path.suffix.lower()
if suffix == ".pdf":
documents.extend(load_pdf(source))
elif suffix == ".docx":
documents.extend(load_docx(source))
elif suffix == ".txt":
documents.extend(load_text(source))
elif suffix in [".md", ".markdown"]:
documents.extend(load_markdown(source))
elif path.is_dir():
documents.extend(load_directory(source))
elif source.startswith("http"):
documents.extend(load_webpage(source))
else:
raise ValueError(f"Unsupported source: {source}")
return documents
Chunking Strategies
# src/ingestion/chunkers.py
from typing import List
from langchain_core.documents import Document
from langchain_text_splitters import (
RecursiveCharacterTextSplitter,
CharacterTextSplitter,
TokenTextSplitter,
MarkdownHeaderTextSplitter,
)
from langchain_community.document_transformers import (
EmbeddingsClusteringFilter,
EmbeddingsRedundantFilter,
)
from src.config import ChunkingConfig
def create_recursive_chunker(config: ChunkingConfig):
"""Create a recursive character chunker."""
return RecursiveCharacterTextSplitter(
chunk_size=config.chunk_size,
chunk_overlap=config.chunk_overlap,
length_function=len,
separators=[
"\n\n", # Paragraphs
"\n", # Lines
"。", # Chinese sentences
"！",
"？",
".", # English sentences
" ", # Words
"", # Characters
],
)
def create_character_chunker(config: ChunkingConfig):
"""Create a simple character chunker."""
return CharacterTextSplitter(
chunk_size=config.chunk_size,
chunk_overlap=config.chunk_overlap,
separator="\n\n",
)
def create_token_chunker(config: ChunkingConfig):
"""Create a token-based chunker."""
return TokenTextSplitter(
chunk_size=config.chunk_size // 4, # Approximate chars to tokens
chunk_overlap=config.chunk_overlap // 4,
)
def split_markdown(documents: List[Document]) -> List[Document]:
"""Split Markdown documents by headers."""
splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
],
)
split_docs = []
for doc in documents:
split_docs.extend(splitter.split_text(doc.page_content))
return split_docs
def remove_redundant_chunks(
documents: List[Document],
embeddings,
threshold: float = 0.95
) -> List[Document]:
"""Remove redundant chunks based on embedding similarity."""
redundant_filter = EmbeddingsRedundantFilter(
embeddings=embeddings,
similarity_fn=lambda x, y: 1 - x.distance(y),
threshold=threshold,
)
return redundant_filter.transform(documents)
def cluster_chunks(
documents: List[Document],
embeddings,
n_clusters: int = 10
) -> List[Document]:
"""Cluster chunks for better retrieval."""
clustering_filter = EmbeddingsClusteringFilter(
embeddings=embeddings,
num_clusters=n_clusters,
num_closest=1,
sorted=True,
)
return clustering_filter.transform(documents)
def chunk_documents(
documents: List[Document],
config: ChunkingConfig
) -> List[Document]:
"""Chunk documents according to configuration."""
if config.strategy == "recursive_character":
chunker = create_recursive_chunker(config)
elif config.strategy == "character":
chunker = create_character_chunker(config)
elif config.strategy == "token":
chunker = create_token_chunker(config)
else:
chunker = create_recursive_chunker(config)
return chunker.split_documents(documents)
Ingestion Pipeline
# src/ingestion/pipeline.py
from typing import List, Optional
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
from src.config import RAGConfig
from src.ingestion.loaders import load_documents
from src.ingestion.chunkers import chunk_documents
def ingest_documents(
sources: List[str],
config: RAGConfig,
clean: bool = False
) -> List[Document]:
"""Run the complete ingestion pipeline."""
# Load documents
print(f"Loading documents from {len(sources)} sources...")
documents = load_documents(sources)
print(f"Loaded {len(documents)} documents")
# Optional: Clean documents
if clean:
documents = clean_documents(documents)
# Chunk documents
print(f"Chunking documents with strategy: {config.chunking.strategy}...")
chunks = chunk_documents(documents, config.chunking)
print(f"Created {len(chunks)} chunks")
# Add metadata
for chunk in chunks:
if "source" not in chunk.metadata:
chunk.metadata["source"] = "unknown"
if "chunk_id" not in chunk.metadata:
chunk.metadata["chunk_id"] = hash(chunk.page_content) % 1000000
return chunks
def clean_documents(documents: List[Document]) -> List[Document]:
"""Clean and normalize documents."""
cleaned = []
for doc in documents:
# Remove extra whitespace
content = " ".join(doc.page_content.split())
# Remove common boilerplate
content = remove_boilerplate(content)
# Truncate if too long
if len(content) > 100000:
content = content[:100000]
cleaned.append(Document(
page_content=content,
metadata=doc.metadata
))
return cleaned
def remove_boilerplate(text: str) -> str:
"""Remove common boilerplate text."""
import re
# Remove common patterns
patterns = [
r'\n\s*\n', # Multiple newlines
r'\[.*?\]', # Bracketed text (often navigation)
r'http[s]?://\S+', # URLs
]
for pattern in patterns:
text = re.sub(pattern, '\n', text)
return text.strip()
Vector Store Setup
Embeddings
# src/retrieval/embeddings.py
from typing import Optional
from langchain_openai import OpenAIEmbeddings
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_cohere import CohereEmbeddings
from src.config import EmbeddingConfig
def get_embeddings(config: EmbeddingConfig):
"""Get embedding model based on configuration."""
if config.provider == "openai":
return OpenAIEmbeddings(
model=config.model,
dimensions=config.dimensions,
)
elif config.provider == "huggingface":
return HuggingFaceEmbeddings(
model_name=config.model,
model_kwargs={"device": "cpu"},
)
elif config.provider == "cohere":
return CohereEmbeddings(
model=config.model,
)
else:
raise ValueError(f"Unsupported embedding provider: {config.provider}")
Vector Store
# src/retrieval/vector_store.py
from typing import List, Optional
from langchain_core.documents import Document
from langchain_core.vectorstores import VectorStore
from langchain_chroma import Chroma
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from src.config import RetrievalConfig, EmbeddingConfig
def create_vector_store(
documents: List[Document],
embeddings: OpenAIEmbeddings,
config: RetrievalConfig,
persist_directory: Optional[str] = None
) -> VectorStore:
"""Create and populate vector store."""
if config.vector_store == "chroma":
return create_chroma(
documents=documents,
embeddings=embeddings,
persist_directory=persist_directory,
)
elif config.vector_store == "faiss":
return create_faiss(
documents=documents,
embeddings=embeddings,
)
else:
raise ValueError(f"Unsupported vector store: {config.vector_store}")
def create_chroma(
documents: List[Document],
embeddings: OpenAIEmbeddings,
persist_directory: Optional[str] = None
) -> Chroma:
"""Create Chroma vector store."""
if persist_directory:
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory=persist_directory,
)
else:
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings,
)
return vectorstore
def create_faiss(
documents: List[Document],
embeddings: OpenAIEmbeddings,
) -> FAISS:
"""Create FAISS vector store."""
vectorstore = FAISS.from_documents(
documents=documents,
embedding=embeddings,
)
return vectorstore
def load_vector_store(
embeddings: OpenAIEmbeddings,
config: RetrievalConfig,
persist_directory: str
) -> VectorStore:
"""Load existing vector store."""
if config.vector_store == "chroma":
return Chroma(
persist_directory=persist_directory,
embedding_function=embeddings,
)
elif config.vector_store == "faiss":
return FAISS.load_local(
folder_path=persist_directory,
embeddings=embeddings,
allow_dangerous_deserialization=True,
)
else:
raise ValueError(f"Unsupported vector store: {config.vector_store}")
Retrievers
# src/retrieval/retrievers.py
from typing import List
from langchain_core.vectorstores import VectorStore
from langchain_core.retrievers import BaseRetriever
from langchain_openai import OpenAIEmbeddings
from src.config import RetrievalConfig
def create_retriever(
vectorstore: VectorStore,
config: RetrievalConfig
) -> BaseRetriever:
"""Create retriever based on configuration."""
if config.search_type == "similarity":
return vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": config.search_k},
)
elif config.search_type == "mmr":
return vectorstore.as_retriever(
search_type="mmr",
search_kwargs={
"k": config.search_k,
"fetch_k": config.search_k * 3,
"lambda_mult": 0.5,
},
)
elif config.search_type == "similarity_score_threshold":
return vectorstore.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"k": config.search_k,
"score_threshold": config.score_threshold,
},
)
else:
return vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": config.search_k},
)
def create_hybrid_retriever(
vectorstore: VectorStore,
documents: List,
alpha: float = 0.7
) -> BaseRetriever:
"""Create hybrid retriever combining vector and BM25 search."""
from langchain_community.retrievers import BM25Retriever
# Create BM25 retriever
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5
# Create vector retriever
vector_retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5},
)
# Combine with Reciprocal Rank Fusion
from langchain_community.retrievers import EnsembleRetriever
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[1 - alpha, alpha],
)
return ensemble_retriever
def create_self_query_retriever(
vectorstore: VectorStore,
embeddings: OpenAIEmbeddings,
metadata_field_info: List[dict],
document_content_description: str
) -> BaseRetriever:
"""Create self-querying retriever with metadata filtering."""
from langchain.chains.query_constructor.base import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain_openai import ChatOpenAI
attribute_info = [
AttributeInfo(
name=field["name"],
description=field["description"],
type=field["type"],
)
for field in metadata_field_info
]
retriever = SelfQueryRetriever.from_llm(
ChatOpenAI(model="gpt-4o"),
vectorstore,
document_content_description,
attribute_info,
embeddings,
)
return retriever
Re-ranking
# src/retrieval/ranking.py
from typing import List
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from langchain_cohere import CohereRerank
from langchain_community.cross_encoders import CrossEncoder
def add_cohere_reranker(
retriever: BaseRetriever,
top_n: int = 5,
model: str = "rerank-english-v3.0"
) -> BaseRetriever:
"""Add Cohere re-ranking to retriever."""
from langchain.retrievers import ContextualCompressionRetriever
compressor = CohereRerank(
model=model,
top_n=top_n,
)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever,
)
return compression_retriever
def add_cross_encoder_reranker(
retriever: BaseRetriever,
model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2",
top_n: int = 5
) -> BaseRetriever:
"""Add cross-encoder re-ranking to retriever."""
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
cross_encoder = CrossEncoder(
model_name=model_name,
)
compressor = CrossEncoderReranker(
model=cross_encoder,
top_n=top_n,
)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever,
)
return compression_retriever
Generation Chains
Prompt Templates
# src/generation/prompts.py
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage
# Standard QA prompt
QA_PROMPT = ChatPromptTemplate.from_messages([
("system", """You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question.
If you don't know the answer, say that you don't know.
Use three sentences maximum and keep the answer concise.
Context: {context}"""),
("human", "{input}"),
])
# Detailed QA prompt
DETAILED_QA_PROMPT = ChatPromptTemplate.from_messages([
("system", """You are a helpful assistant that answers questions based on provided context.
Instructions:
1. Use ONLY the provided context to answer the question
2. If the answer is not in the context, say "I don't have enough information to answer this question"
3. Cite the source documents when possible
4. Be concise but thorough
5. If there are conflicting pieces of information, mention both
Context:
{context}
Question: {input}
Answer:"""),
])
# Conversational QA prompt
CONVERSATIONAL_PROMPT = ChatPromptTemplate.from_messages([
("system", """You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question.
If you don't know the answer, say that you don't know.
Use three sentences maximum and keep the answer concise.
Context: {context}
Previous conversation:
{chat_history}"""),
("human", "{input}"),
])
# Citations prompt
CITATIONS_PROMPT = ChatPromptTemplate.from_messages([
("system", """You are a helpful assistant. Answer the question based on the context.
For each claim you make, cite the relevant source using [1], [2], etc.
Context:
{context}
Question: {input}
Format your answer with citations like this:
- The sky is blue [1].
- Water boils at 100°C [2].
Answer:"""),
])
QA Chains
# src/generation/chains.py
from typing import Any, Dict
from langchain_core.language_models import BaseChatModel
from langchain_core.retrievers import BaseRetriever
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from src.generation.prompts import QA_PROMPT, DETAILED_QA_PROMPT
def create_qa_chain(
llm: BaseChatModel,
retriever: BaseRetriever,
prompt=QA_PROMPT,
return_source_documents: bool = False
):
"""Create a basic QA chain."""
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
chain = (
{"context": retriever | format_docs, "input": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
if return_source_documents:
def wrap_output(answer: str, docs) -> Dict[str, Any]:
return {
"answer": answer,
"source_documents": docs,
}
chain = (
{"context": retriever | format_docs, "input": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
| lambda x: {"answer": x}
)
return chain
def create_refine_chain(
llm: BaseChatModel,
retriever: BaseRetriever,
):
"""Create a Refine chain for long documents."""
from langchain.chains import RefineChain
return RefineChain.from_llm(
llm=llm,
retriever=retriever,
return_intermediate_steps=True,
)
def create_map_reduce_chain(
llm: BaseChatModel,
retriever: BaseRetriever,
):
"""Create a Map-Reduce chain."""
from langchain.chains import MapReduceDocumentsChain, ReduceDocumentsChain
from langchain.chains.question_answering import load_qa_chain
# Map chain
map_chain = load_qa_chain(llm, chain_type="map_reduce")
# Reduce chain
reduce_documents_chain = ReduceDocumentsChain(
combine_chain=map_chain,
collapse_chain=map_chain,
)
# Map-reduce chain
chain = MapReduceDocumentsChain(
retriever=retriever,
document_variable_name="context",
reduce_documents_chain=reduce_documents_chain,
)
return chain
Conversational QA
# src/generation/conversation.py
from typing import List, Tuple
from langchain_core.language_models import BaseChatModel
from langchain_core.retrievers import BaseRetriever
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import BaseMessage
from src.generation.prompts import CONVERSATIONAL_PROMPT
def format_chat_history(history: List[Tuple[str, str]]) -> str:
"""Format chat history for the prompt."""
formatted = []
for human, ai in history:
formatted.append(f"Human: {human}")
formatted.append(f"Assistant: {ai}")
return "\n".join(formatted)
def create_conversational_chain(
llm: BaseChatModel,
retriever: BaseRetriever,
memory=None,
):
"""Create a conversational QA chain."""
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
def format_history(history):
if not history:
return ""
return format_chat_history(history)
chain = (
{
"context": retriever | format_docs,
"input": RunnablePassthrough(),
"chat_history": RunnableLambda(lambda x: x.get("chat_history", [])) | format_history,
}
| CONVERSATIONAL_PROMPT
| llm
| StrOutputParser()
)
return chain
def create_conversation_with_memory(
llm: BaseChatModel,
retriever: BaseRetriever,
memory_store,
):
"""Create a conversational QA chain with persistent memory."""
from langchain_core.chat_history import BaseChatMessageHistory
def get_session_history(session_id: str) -> BaseChatMessageHistory:
return memory_store.get(session_id)
# Use MemorySaver or other checkpointer
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
# Create graph-based conversation
from langgraph.graph import StateGraph, MessagesState
def retrieve_and_answer(state: MessagesState):
messages = state["messages"]
last_message = messages[-1].content
docs = retriever.invoke(last_message)
context = "\n\n".join(d.page_content for d in docs)
response = llm.invoke([
("system", f"Use this context: {context}"),
*messages,
])
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node("answer", retrieve_and_answer)
builder.set_entry_point("answer")
builder.add_edge("answer", "__end__")
graph = builder.compile(checkpointer=checkpointer)
return graph
Main Pipeline
Complete RAG Pipeline
# main.py
from src.config import RAGConfig
from src.ingestion.pipeline import ingest_documents
from src.retrieval.embeddings import get_embeddings
from src.retrieval.vector_store import create_vector_store, load_vector_store
from src.retrieval.retrievers import create_retriever
from src.generation.chains import create_qa_chain
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
def build_rag_pipeline(
sources: list[str],
config: RAGConfig,
fresh: bool = True
):
"""Build the complete RAG pipeline."""
# Get embeddings
print("Loading embedding model...")
embeddings = get_embeddings(config.embedding)
if fresh:
# Ingest documents
print("Ingesting documents...")
documents = ingest_documents(sources, config)
# Create vector store
print("Creating vector store...")
vectorstore = create_vector_store(
documents=documents,
embeddings=embeddings,
config=config.retrieval,
persist_directory="./chroma_db",
)
else:
# Load existing vector store
print("Loading existing vector store...")
vectorstore = load_vector_store(
embeddings=embeddings,
config=config.retrieval,
persist_directory="./chroma_db",
)
# Create retriever
print("Creating retriever...")
retriever = create_retriever(vectorstore, config.retrieval)
# Create LLM
print("Loading LLM...")
llm = ChatOpenAI(
model=config.llm.model,
temperature=config.llm.temperature,
max_tokens=config.llm.max_tokens,
)
# Create QA chain
print("Creating QA chain...")
qa_chain = create_qa_chain(
llm=llm,
retriever=retriever,
return_source_documents=True,
)
return qa_chain
def main():
"""Main entry point."""
import argparse
parser = argparse.ArgumentParser(description="RAG Pipeline")
parser.add_argument("--sources", nargs="+", required=True, help="Document sources")
parser.add_argument("--query", help="Query to answer")
parser.add_argument("--fresh", action="store_true", help="Fresh ingestion")
args = parser.parse_args()
# Load config
config = RAGConfig.from_env()
# Build pipeline
qa_chain = build_rag_pipeline(
sources=args.sources,
config=config,
fresh=args.fresh,
)
# Run query
if args.query:
result = qa_chain.invoke(args.query)
print(f"\nAnswer: {result['answer']}")
print(f"\nSources: {len(result['source_documents'])} documents")
for i, doc in enumerate(result['source_documents'], 1):
print(f" [{i}] {doc.metadata.get('source', 'unknown')}: {doc.page_content[:100]}...")
if __name__ == "__main__":
main()
Usage Examples
Basic Question Answering
from main import build_rag_pipeline
# Build pipeline
qa_chain = build_rag_pipeline(
sources=["./data/documents/*.pdf"],
fresh=True,
)
# Ask a question
result = qa_chain.invoke("What is the return policy?")
print(result["answer"])
print("Sources:", result["source_documents"])
Conversational QA
from src.generation.conversation import create_conversational_chain
from langchain_openai import ChatOpenAI
# Create conversational chain
conversational_chain = create_conversational_chain(
llm=ChatOpenAI(model="gpt-4o"),
retriever=retriever,
)
# First question
response1 = conversational_chain.invoke({
"input": "What is the return policy?",
"chat_history": [],
})
print(response1)
# Follow-up question
response2 = conversational_chain.invoke({
"input": "What about exchanges?",
"chat_history": [("What is the return policy?", response1)],
})
print(response2)
API Server
# api/server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from main import build_rag_pipeline
app = FastAPI()
# Build pipeline once
qa_chain = build_rag_pipeline(
sources=["./data/documents/*.pdf"],
fresh=False,
)
class QueryRequest(BaseModel):
query: str
conversation_id: str | None = None
class QueryResponse(BaseModel):
answer: str
sources: list[dict]
@app.post("/ask", response_model=QueryResponse)
async def ask_question(request: QueryRequest):
try:
result = qa_chain.invoke(request.query)
return QueryResponse(
answer=result["answer"],
sources=[
{"content": doc.page_content[:200], "source": doc.metadata.get("source")}
for doc in result["source_documents"]
],
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
Evaluation
RAGAS Evaluation
# src/evaluation/pipeline.py
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from datasets import Dataset
def evaluate_rag(
qa_chain,
test_dataset: list[dict],
):
"""Evaluate RAG pipeline using RAGAS."""
# Run queries
answers = []
contexts = []
questions = []
for item in test_dataset:
result = qa_chain.invoke(item["question"])
answers.append(result["answer"])
contexts.append([doc.page_content for doc in result["source_documents"]])
questions.append(item["question"])
# Create dataset
dataset = Dataset.from_dict({
"question": questions,
"answer": answers,
"contexts": contexts,
"ground_truth": [item["answer"] for item in test_dataset],
})
# Evaluate
results = evaluate(
dataset=dataset,
metrics=[faithfulness, answer_relevancy, context_precision],
)
return results
def calculate_metrics(results):
"""Calculate evaluation metrics."""
return {
"faithfulness": results["faithfulness"],
"answer_relevancy": results["answer_relevancy"],
"context_precision": results["context_precision"],
}
Troubleshooting
Low Retrieval Quality
- Check embeddings: Ensure same embedding model for indexing and retrieval
- Adjust chunk size: Try 500-2000 characters
- Increase k: Retrieve more documents
- Add re-ranking: Use Cohere or cross-encoder re-ranking
- Try hybrid search: Combine vector and BM25 search
Slow Performance
- Use smaller embeddings: text-embedding-3-small
- Enable caching: Cache embeddings
- Use FAISS: Faster than Chroma for local search
- Async processing: For ingestion
- Batch operations: For vector store
Memory Issues
- Process in batches: Don't load all documents at once
- Smaller chunks: Reduce chunk size
- Clear store: Between runs
- Streaming: For long documents
Resources
- LangChain RAG Guide: https://python.langchain.com/docs/tutorials/rag/
- LangChain Documentation: https://python.langchain.com/
- RAGAS: https://github.com/explodinggradients/ragas
- Chroma: https://www.trychroma.com/
- FAISS: https://github.com/facebookresearch/faiss
Last updated: May 2026