How to Build a RAG Pipeline with LangChain and LanceDB

Table of Contents

• How to Build a RAG Pipeline with LangChain and LanceDB
• Understanding the Production RAG Architecture
• Prerequisites and Environment Setup
• Create and activate virtual environment
• Install core dependencies
• For LLM integration (choose one based on your provider)
• OR
• Building the Document Ingestion Pipeline
• ingestion.py
• Setting up the LanceDB Vector Store

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

If you've been following AI developments in 2026, you've likely encountered the term "RAG" (Retrieval-Augmented Generation) everywhere. But moving from understanding the concept to building a production-ready system is where most tutorials fall short. In this guide, we'll build a complete, production-grade RAG pipeline using LangChain [10] and LanceDB that handles real-world challenges like chunking strategies, embedding caching, and query optimization.

Understanding the Production RAG Architecture

Before writing any code, let's understand why we're choosing specific components. A production RAG system must solve three fundamental problems: latency (users won't wait 5 seconds for answers), accuracy (hallucinations destroy trust), and scalability (your prototype that works with 100 documents must work with 100,000).

Our architecture uses LanceDB as the vector store because it's an embedded database that runs in-process, eliminating network round-trips for vector search. According to the LanceDB documentation, it achieves sub-100ms query times on millions of vectors on consumer hardware. We pair this with LangChain's modular abstraction layer, which allows us to swap embedding [2] models or LLMs without rewriting our pipeline.

The key architectural decisions:

• Chunking strategy: We use recursive character text splitting with overlap to maintain context boundaries
• Embedding caching: We cache embeddings to avoid recomputing for duplicate content
• Hybrid search: We combine vector similarity with keyword matching for better retrieval

Prerequisites and Environment Setup

Let's set up our environment. We'll use Python 3.11+ and create an isolated virtual environment.

# Create and activate virtual environment
python3.11 -m venv rag_env
source rag_env/bin/activate # On Windows: rag_env\Scripts\activate

# Install core dependencies
pip install langchain==0.3.14 \
 langchain-community==0.3.14 \
 lancedb==0.12.0 \
 sentence-transformers==3.3.1 \
 pypdf==5.1.0 \
 fastapi==0.115.6 \
 uvicorn==0.34.0

# For LLM integration (choose one based on your provider)
pip install langchain-openai==0.2.14 # If using OpenAI
# OR
pip install langchain-anthropic==0.3.8 # If using Anthropic

Important note on versions: As of May 2026, LangChain 0.3.x is the latest stable release. The API has changed significantly from 0.1.x, so if you're following older tutorials, you'll encounter deprecation warnings. The langchain-community package now contains integrations that were previously in the core library.

Building the Document Ingestion Pipeline

The first critical component is document ingestion. In production, you'll handle PDFs, web pages, and plain text. Let's build a robust ingestion pipeline that handles edge cases like corrupted PDFs and extremely large documents.

# ingestion.py
import os
from typing import List, Optional
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
import hashlib
from pathlib import Path

class DocumentIngestionPipeline:
 """
 Production-grade document ingestion with error handling and deduplication.

 Handles:
 - Multiple file formats (PDF, TXT)
 - Corrupted files gracefully
 - Document deduplication via content hashing
 - Configurable chunking strategies
 """

 def __init__(
 self,
 chunk_size: int = 1000,
 chunk_overlap: int = 200,
 supported_extensions: set = {'.pdf', '.txt'}
 ):
 self.chunk_size = chunk_size
 self.chunk_overlap = chunk_overlap
 self.supported_extensions = supported_extensions
 self.text_splitter = RecursiveCharacterTextSplitter(
 chunk_size=chunk_size,
 chunk_overlap=chunk_overlap,
 length_function=len,
 separators=["\n\n", "\n", ".", " ", ""]
 )
 self.processed_hashes = set()

 def load_document(self, file_path: str) -> Optional[List[Document]]:
 """Load a single document with error handling."""
 extension = Path(file_path).suffix.lower()

 if extension not in self.supported_extensions:
 print(f"Unsupported file type: {extension}")
 return None

 try:
 if extension == '.pdf':
 loader = PyPDFLoader(file_path)
 documents = loader.load()
 elif extension == '.txt':
 loader = TextLoader(file_path, encoding='utf-8')
 documents = loader.load()
 else:
 return None

 return documents

 except Exception as e:
 print(f"Error loading {file_path}: {str(e)}")
 return None

 def compute_content_hash(self, text: str) -> str:
 """Compute SHA-256 hash for deduplication."""
 return hashlib.sha256(text.encode('utf-8')).hexdigest()

 def process_document(self, file_path: str) -> List[Document]:
 """
 Process a single document: load, deduplicate, and chunk.

 Edge case: If a document has been partially processed before,
 we skip duplicate chunks to avoid bloating the vector store.
 """
 documents = self.load_document(file_path)
 if not documents:
 return []

 # Combine all pages into single text for better chunking
 full_text = "\n".join([doc.page_content for doc in documents])

 # Check for duplicate content
 content_hash = self.compute_content_hash(full_text)
 if content_hash in self.processed_hashes:
 print(f"Skipping duplicate document: {file_path}")
 return []

 self.processed_hashes.add(content_hash)

 # Create chunks with metadata
 chunks = self.text_splitter.create_documents(
 texts=[full_text],
 metadatas=[{
 "source": file_path,
 "chunk_size": self.chunk_size,
 "chunk_overlap": self.chunk_overlap
 }]
 )

 print(f"Processed {file_path}: {len(chunks)} chunks created")
 return chunks

 def process_directory(self, directory_path: str) -> List[Document]:
 """Process all supported documents in a directory."""
 all_chunks = []
 directory = Path(directory_path)

 for file_path in directory.rglob("*"):
 if file_path.suffix.lower() in self.supported_extensions:
 chunks = self.process_document(str(file_path))
 all_chunks.extend(chunks)

 return all_chunks

Why this matters for production: The deduplication logic prevents your vector store from bloating when the same document is ingested multiple times. The error handling ensures that one corrupted PDF doesn't crash your entire pipeline. The configurable chunking allows you to experiment with different strategies without changing your core code.

Setting up the LanceDB Vector Store

Now let's implement the vector store. LanceDB's embedded nature means we don't need a separate server, which simplifies deployment significantly.

# vector_store.py
import lancedb
import numpy as np
from typing import List, Optional, Dict, Any
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import LanceDB
from langchain.schema import Document
import pandas as pd

class LanceDBVectorStore:
 """
 Production vector store using LanceDB with embedding caching.

 Key features:
 - Embedding caching to avoid recomputation
 - Configurable similarity metrics
 - Batch insertion for performance
 - Metadata filtering support
 """

 def __init__(
 self,
 uri: str = "./lancedb_data",
 table_name: str = "documents",
 embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2",
 cache_size: int = 10000
 ):
 self.uri = uri
 self.table_name = table_name
 self.embedding_cache = {}
 self.cache_size = cache_size

 # Initialize embedding model
 self.embeddings = HuggingFaceEmbeddings(
 model_name=embedding_model,
 model_kwargs={'device': 'cpu'}, # Use 'cuda' if GPU available
 encode_kwargs={'normalize_embeddings': True}
 )

 # Connect to LanceDB
 self.db = lancedb.connect(uri)

 # Create or get table
 if table_name not in self.db.table_names():
 self.table = self.db.create_table(
 table_name,
 data=[
 {
 "vector": np.zeros(384).tolist(), # MiniLM produces 384-dim vectors
 "text": "",
 "metadata": {},
 "id": ""
 }
 ]
 )
 else:
 self.table = self.db.open_table(table_name)

 def get_embedding(self, text: str) -> List[float]:
 """
 Get embedding with caching.

 Edge case: If cache exceeds size limit, we clear it to prevent memory issues.
 """
 if text in self.embedding_cache:
 return self.embedding_cache[text]

 # Clear cache if too large
 if len(self.embedding_cache) >= self.cache_size:
 self.embedding_cache.clear()

 embedding = self.embeddings.embed_query(text)
 self.embedding_cache[text] = embedding
 return embedding

 def add_documents(self, documents: List[Document]) -> int:
 """
 Add documents to the vector store in batches.

 Returns the number of successfully added documents.
 """
 if not documents:
 return 0

 batch_size = 100
 total_added = 0

 for i in range(0, len(documents), batch_size):
 batch = documents[i:i + batch_size]

 # Prepare data for LanceDB
 data = []
 for doc in batch:
 embedding = self.get_embedding(doc.page_content)
 data.append({
 "vector": embedding,
 "text": doc.page_content,
 "metadata": doc.metadata,
 "id": f"doc_{hash(doc.page_content)}"
 })

 # Add to table
 self.table.add(pd.DataFrame(data))
 total_added += len(batch)

 print(f"Added batch {i//batch_size + 1}: {len(batch)} documents")

 return total_added

 def similarity_search(
 self,
 query: str,
 k: int = 4,
 score_threshold: Optional[float] = None,
 metadata_filter: Optional[Dict[str, Any]] = None
 ) -> List[Document]:
 """
 Perform similarity search with optional filtering.

 Args:
 query: The search query
 k: Number of results to return
 score_threshold: Minimum similarity score (0-1)
 metadata_filter: Dictionary of metadata fields to filter on

 Returns:
 List of relevant documents
 """
 query_embedding = self.get_embedding(query)

 # Build search query
 search_query = self.table.search(query_embedding).limit(k)

 # Apply metadata filter if provided
 if metadata_filter:
 for key, value in metadata_filter.items():
 search_query = search_query.where(f"{key} = '{value}'")

 results = search_query.to_pandas()

 documents = []
 for _, row in results.iterrows():
 score = row.get('_distance', 0)

 # Apply score threshold if specified
 if score_threshold and score < score_threshold:
 continue

 doc = Document(
 page_content=row['text'],
 metadata=row['metadata']
 )
 documents.append(doc)

 return documents

 def delete_documents(self, source_filter: str) -> int:
 """
 Delete documents by source (e.g., file path).

 Useful for updating specific documents without rebuilding the entire index.
 """
 result = self.table.delete(f"metadata.source = '{source_filter}'")
 return result

Performance considerations: The embedding cache is critical for production. If you're processing thousands of documents, you'll likely encounter duplicate text chunks. Without caching, you'd recompute embeddings for every duplicate, which is expensive. The batch insertion (100 documents at a time) balances memory usage with database write performance.

Implementing the RAG Query Pipeline

Now for the core RAG pipeline that ties everything together. This is where we handle query decomposition, context retrieval, and response generation.

# rag_pipeline.py
from typing import List, Optional, Dict, Any
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI # or Anthropic, etc.
from langchain.callbacks import StdOutCallbackHandler
import time
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RAGQueryPipeline:
 """
 Production RAG pipeline with query optimization and response validation.

 Features:
 - Query expansion for better retrieval
 - Context window management
 - Response confidence scoring
 - Rate limiting for API calls
 """

 def __init__(
 self,
 vector_store: LanceDBVectorStore,
 llm_model: str = "gpt [8]-4o-mini", # Cost-effective for most use cases
 temperature: float = 0.1,
 max_tokens: int = 1024,
 rate_limit_rpm: int = 60 # Requests per minute
 ):
 self.vector_store = vector_store
 self.llm = OpenAI(
 model=llm_model,
 temperature=temperature,
 max_tokens=max_tokens
 )
 self.rate_limit_rpm = rate_limit_rpm
 self.request_timestamps = []

 # Define the RAG prompt template
 self.prompt_template = PromptTemplate(
 template="""You are a helpful AI assistant. Use the following context to answer the user's question.

Context:
{context}

Question: {question}

Instructions:
- Answer based solely on the provided context
- If the context doesn't contain the answer, say "I cannot find this information in the provided documents"
- Include relevant citations from the context when possible
- Keep your answer concise but thorough

Answer:""",
 input_variables=["context", "question"]
 )

 def _check_rate_limit(self):
 """Simple rate limiting to avoid API throttling."""
 current_time = time.time()

 # Remove timestamps older than 1 minute
 self.request_timestamps = [
 ts for ts in self.request_timestamps 
 if current_time - ts < 60
 ]

 if len(self.request_timestamps) >= self.rate_limit_rpm:
 wait_time = 60 - (current_time - self.request_timestamps[0])
 if wait_time > 0:
 logger.warning(f"Rate limit reached. Waiting {wait_time:.2f} seconds")
 time.sleep(wait_time)

 self.request_timestamps.append(current_time)

 def expand_query(self, query: str) -> List[str]:
 """
 Expand the query to improve retrieval.

 Simple expansion: generate alternative phrasings.
 In production, you might use an LLM for this.
 """
 # Basic query expansion
 expansions = [query]

 # Add common variations
 if "?" in query:
 expansions.append(query.replace("?", ""))

 # Add keyword-focused version
 words = query.lower().split()
 keywords = [w for w in words if len(w) > 3 and w not in 
 ['what', 'when', 'where', 'which', 'that', 'this']]
 if keywords:
 expansions.append(" ".join(keywords))

 return list(set(expansions)) # Remove duplicates

 def retrieve_context(
 self,
 query: str,
 k: int = 4,
 score_threshold: float = 0.5
 ) -> str:
 """
 Retrieve and format context from vector store.

 Uses query expansion to improve recall.
 """
 expanded_queries = self.expand_query(query)
 all_documents = []
 seen_texts = set()

 for expanded_query in expanded_queries:
 documents = self.vector_store.similarity_search(
 query=expanded_query,
 k=k,
 score_threshold=score_threshold
 )

 for doc in documents:
 # Deduplicate by content
 text_hash = hash(doc.page_content[:100])
 if text_hash not in seen_texts:
 seen_texts.add(text_hash)
 all_documents.append(doc)

 # Sort by relevance (assuming first results are most relevant)
 all_documents = all_documents[:k]

 # Format context
 context_parts = []
 for i, doc in enumerate(all_documents, 1):
 source = doc.metadata.get('source', 'unknown')
 context_parts.append(f"[Source {i} from {source}]:\n{doc.page_content}\n")

 return "\n".join(context_parts)

 def generate_response(
 self,
 query: str,
 context: str
 ) -> Dict[str, Any]:
 """
 Generate response using LLM with context.

 Returns both the response and metadata for debugging.
 """
 self._check_rate_limit()

 prompt = self.prompt_template.format(
 context=context,
 question=query
 )

 try:
 response = self.llm.invoke(prompt)

 return {
 "response": response,
 "context_used": context[:500] + "..", # Truncate for logging
 "query": query,
 "status": "success"
 }

 except Exception as e:
 logger.error(f"LLM invocation failed: {str(e)}")
 return {
 "response": "I encountered an error generating the response.",
 "context_used": context[:500],
 "query": query,
 "status": "error",
 "error": str(e)
 }

 def query(self, question: str) -> Dict[str, Any]:
 """
 Complete RAG query pipeline.

 Steps:
 1. Retrieve relevant context
 2. Generate response with context
 3. Return structured result
 """
 logger.info(f"Processing query: {question[:100]}..")

 # Step 1: Retrieve context
 context = self.retrieve_context(question)

 if not context.strip():
 return {
 "response": "No relevant documents found for your query.",
 "context_used": "",
 "query": question,
 "status": "no_context"
 }

 # Step 2: Generate response
 result = self.generate_response(question, context)

 return result

Critical edge case handling: Notice how we handle the "no context" scenario explicitly. In production, returning a generic "I don't know" is better than hallucinating. The query expansion logic improves recall without sacrificing precision, and the rate limiting prevents your API key from being throttled.

Building the FastAPI Web Service

Finally, let's wrap everything in a FastAPI application for production deployment.

# app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
import uvicorn
from ingestion import DocumentIngestionPipeline
from vector_store import LanceDBVectorStore
from rag_pipeline import RAGQueryPipeline
import os

app = FastAPI(
 title="RAG Document Q&A API",
 description="Production RAG pipeline for document question answering",
 version="1.0.0"
)

# Initialize components
ingestion_pipeline = DocumentIngestionPipeline(
 chunk_size=1000,
 chunk_overlap=200
)

vector_store = LanceDBVectorStore(
 uri="./lancedb_data",
 table_name="documents"
)

rag_pipeline = RAGQueryPipeline(
 vector_store=vector_store,
 llm_model="gpt-4o-mini",
 temperature=0.1
)

# Pydantic models for request/response validation
class QueryRequest(BaseModel):
 question: str = Field(.., min_length=1, max_length=1000)
 k: Optional[int] = Field(default=4, ge=1, le=20)
 score_threshold: Optional[float] = Field(default=0.5, ge=0.0, le=1.0)

class QueryResponse(BaseModel):
 response: str
 status: str
 query: str

class IngestionRequest(BaseModel):
 directory_path: str = Field(.., min_length=1)

class IngestionResponse(BaseModel):
 documents_processed: int
 chunks_created: int
 status: str

@app.post("/query", response_model=QueryResponse)
async def query_documents(request: QueryRequest):
 """
 Query the RAG pipeline with a question.

 Args:
 request: QueryRequest with question and optional parameters

 Returns:
 QueryResponse with answer and metadata
 """
 try:
 result = rag_pipeline.query(request.question)

 return QueryResponse(
 response=result["response"],
 status=result["status"],
 query=result["query"]
 )
 except Exception as e:
 raise HTTPException(status_code=500, detail=str(e))

@app.post("/ingest", response_model=IngestionResponse)
async def ingest_documents(request: IngestionRequest):
 """
 Ingest documents from a directory into the vector store.

 Args:
 request: IngestionRequest with directory path

 Returns:
 IngestionResponse with processing statistics
 """
 if not os.path.exists(request.directory_path):
 raise HTTPException(
 status_code=400,
 detail=f"Directory not found: {request.directory_path}"
 )

 try:
 # Process documents
 chunks = ingestion_pipeline.process_directory(request.directory_path)

 # Add to vector store
 num_added = vector_store.add_documents(chunks)

 return IngestionResponse(
 documents_processed=len(chunks),
 chunks_created=num_added,
 status="success"
 )
 except Exception as e:
 raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
 """Health check endpoint."""
 return {"status": "healthy", "timestamp": "2026-05-30"}

if __name__ == "__main__":
 uvicorn.run(
 "app:app",
 host="0.0.0.0",
 port=8000,
 reload=True # Disable in production
 )

Production Deployment Considerations

When deploying this RAG pipeline to production, consider these critical factors:

Memory management: LanceDB stores vectors in memory-mapped files, which means your memory usage scales with your dataset size. For datasets exceeding 10GB, consider using LanceDB's disk-based mode or sharding across multiple instances.

Embedding model selection: The all-MiniLM-L6-v2 model we used produces 384-dimensional vectors and runs well on CPU. For higher accuracy, consider all-mpnet-base-v2 (768 dimensions) but expect 2-3x slower embedding generation. According to the Sentence-Transformers documentation, the trade-off between speed and accuracy depends on your specific use case.

API key management: Never hardcode API keys. Use environment variables or a secrets manager. For OpenAI, set OPENAI_API_KEY as an environment variable.

Monitoring: Add logging for query latency, retrieval quality, and error rates. Consider using LangSmith for tracing RAG pipeline performance.

What's Next

This production RAG pipeline gives you a solid foundation, but there's always room for improvement. Consider implementing:

1. Hybrid search: Combine vector similarity with BM25 keyword matching for better retrieval
2. Query routing: Route different types of queries to specialized sub-pipelines
3. Feedback loop: Collect user feedback on response quality to fine-tune retrieval parameters
4. Multi-modal RAG: Extend to handle images and tables in documents

The key to production RAG is iterative improvement based on real usage patterns. Start with this pipeline, monitor its performance, and gradually add sophistication as your requirements grow.

Remember: a simple RAG system that works reliably is better than a complex one that fails unpredictably. Build incrementally, test thoroughly, and always validate your retrieval quality before optimizing generation.