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 comprehensive

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.