How to Build a RAG Pipeline with LangChain and LanceDB

Table of Contents

• How to Build a RAG Pipeline with LangChain and LanceDB
  • Real-World Use Case and Architecture
  • Prerequisites and Environment Setup
  • Core Implementation: Document Ingestion Pipeline
  • Core Implementation: Query Pipeline with FastAPI
• Global vector store instance (initialized at startup)
  • Pitfalls and Production Tips
    • 1. Chunk Size and Overlap Trade-offs
    • 3. Memory Usage with Large PDFs
    • 4. API Rate Limits

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

You're building a retrieval-augmented generation (RAG) pipeline for a production document search system. The problem is straightforward: you have thousands of PDFs, and you need to answer questions about their content without fine-tuning [3] a model. The standard approach—embedding chunks into a vector store and retrieving relevant context—works, but most tutorials skip the hard parts: handling chunk overlap correctly, managing vector store schema migrations, and dealing with real-world query failures.

This tutorial walks through building a RAG pipeline using LangChain 0.3.x and LanceDB 0.12.x, deployed behind a FastAPI endpoint. We'll cover the architecture decisions that matter when you're not just prototyping on a Jupyter notebook.

Real-World Use Case and Architecture

Consider a legal document review system. A law firm has 10,000 contract PDFs. Associates need to ask questions like "What is the indemnification clause in contract 4523?" or "Which contracts have a force majeure clause exceeding 30 days?" A naive approach—loading all text into a single LLM context window—fails because GPT-4's 128k token context isn't large enough for 10,000 documents, and the cost per query would be prohibitive.

The RAG architecture solves this by splitting documents into chunks, embedding them into a vector database [1], and retrieving only relevant chunks per query. The key production considerations are:

• Chunking strategy: Overlapping chunks prevent information loss at boundaries, but too much overlap wastes storage and retrieval time.
• Vector store choice: LanceDB stores vectors as Apache Arrow tables, which means zero-copy reads and no separate database server to manage. This matters when you're deploying on a single machine or in a serverless environment.
• Retrieval pipeline: You need to handle cases where no relevant chunks exist, or where the retrieved chunks exceed the LLM's context window.

Prerequisites and Environment Setup

You'll need Python 3.10+ and the following packages. Install them in a virtual environment to avoid dependency conflicts:

python -m venv rag_env
source rag_env/bin/activate  # On Windows: rag_env\Scripts\activate

pip install langchain==0.3.14 langchain-community==0.3.14 langchain-openai [8]==0.2.14 \
            lancedb==0.12.0 pypdf==5.1.0 fastapi==0.115.6 uvicorn==0.34.0 \
            pydantic==2.10.3 python-multipart==0.0.19

The versions above are the latest stable releases as of January 2026. LangChain 0.3.x introduced breaking changes to the document loader and vector store interfaces, so using the exact versions matters. If you're on an older LangChain version, the LanceDB vector store class might not exist—it was added in LangChain 0.2.5.

Set your OpenAI API key as an environment variable:

export OPENAI_API_KEY="sk-your-key-here"

For production, use a secrets manager like AWS Secrets Manager or HashiCorp Vault. Hardcoding keys in environment variables is acceptable for local development but not for deployment.

Core Implementation: Document Ingestion Pipeline

The ingestion pipeline handles PDF parsing, chunking with overlap, and vector store population. Here's the complete implementation with error handling for malformed PDFs and duplicate documents:

import os
import hashlib
from typing import List, Optional
from pathlib import Path

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import LanceDB
import lancedb

def compute_document_hash(content: str) -> str:
    """Generate SHA-256 hash for deduplication."""
    return hashlib.sha256(content.encode('utf-8')).hexdigest()

def load_and_chunk_pdfs(
    pdf_directory: str,
    chunk_size: int = 1000,
    chunk_overlap: int = 200
) -> List[dict]:
    """
    Load all PDFs from a directory, split into overlapping chunks.

    Args:
        pdf_directory: Path to directory containing PDF files
        chunk_size: Target size of each chunk in characters
        chunk_overlap: Number of overlapping characters between chunks

    Returns:
        List of dictionaries with 'text', 'source', 'page', and 'hash' keys

    Raises:
        FileNotFoundError: If pdf_directory doesn't exist
        ValueError: If chunk_overlap >= chunk_size
    """
    if chunk_overlap >= chunk_size:
        raise ValueError(
            f"chunk_overlap ({chunk_overlap}) must be less than chunk_size ({chunk_size})"
        )

    pdf_path = Path(pdf_directory)
    if not pdf_path.exists():
        raise FileNotFoundError(f"Directory not found: {pdf_directory}")

    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ".", " ", ""],
    )

    documents = []
    pdf_files = list(pdf_path.glob("*.pdf"))

    if not pdf_files:
        print(f"Warning: No PDF files found in {pdf_directory}")
        return documents

    for pdf_file in pdf_files:
        try:
            loader = PyPDFLoader(str(pdf_file))
            pages = loader.load()
        except Exception as e:
            print(f"Error loading {pdf_file.name}: {e}")
            continue

        # Combine all pages into a single text for chunking
        # This preserves page boundaries as metadata
        full_text = "\n".join([page.page_content for page in pages])

        chunks = text_splitter.split_text(full_text)

        for i, chunk_text in enumerate(chunks):
            doc_hash = compute_document_hash(chunk_text)
            documents.append({
                "text": chunk_text,
                "source": pdf_file.name,
                "page": i // (len(chunks) // len(pages) + 1) + 1,  # Approximate page number
                "hash": doc_hash
            })

    print(f"Loaded {len(documents)} chunks from {len(pdf_files)} PDFs")
    return documents

def populate_vector_store(
    documents: List[dict],
    db_path: str = "./lancedb_store",
    table_name: str = "documents",
    embedding_model: str = "text-embedding-3-small"
) -> LanceDB:
    """
    Create or update a LanceDB vector store with document chunks.

    Args:
        documents: List of document dictionaries from load_and_chunk_pdfs
        db_path: Directory path for LanceDB database
        table_name: Name of the table to store vectors
        embedding_model: OpenAI embedding model name

    Returns:
        LanceDB vector store instance
    """
    embeddings = OpenAIEmbeddings(model=embedding_model)

    # Initialize LanceDB connection
    db = lancedb.connect(db_path)

    # Check if table exists and handle schema changes
    try:
        existing_table = db.open_table(table_name)
        # In production, you'd check schema compatibility here
        # LanceDB 0.12.x doesn't support schema migration automatically
        print(f"Table '{table_name}' exists with {existing_table.count_rows()} rows")
    except Exception:
        print(f"Creating new table '{table_name}'")

    # Extract texts and metadata for LangChain
    texts = [doc["text"] for doc in documents]
    metadatas = [
        {"source": doc["source"], "page": doc["page"], "hash": doc["hash"]}
        for doc in documents
    ]

    # Create vector store with deduplication
    # LangChain's LanceDB wrapper handles embedding and insertion
    vector_store = LanceDB.from_texts(
        texts=texts,
        embedding=embeddings,
        metadatas=metadatas,
        connection=db,
        table_name=table_name,
        mode="overwrite"  # Replace existing data; use "append" for incremental updates
    )

    print(f"Vector store populated with {len(texts)} vectors")
    return vector_store

The RecursiveCharacterTextSplitter with separators= ensures chunks break at natural boundaries first. The 200-character overlap means that if a sentence spans two chunks, the second chunk includes the tail of the previous sentence. This prevents the LLM from missing context like "The indemnification clause in Section 4.2 states that.." when the clause definition starts in the previous chunk.

The compute_document_hash function enables deduplication. In production, you'd check this hash against existing entries before inserting. LanceDB doesn't enforce unique constraints on metadata fields, so you need to handle this at the application level.

Core Implementation: Query Pipeline with FastAPI

The query pipeline retrieves relevant chunks and generates answers. This implementation handles edge cases like empty results and context overflow:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import Document
from typing import Optional

app = FastAPI(title="RAG Document Query API")

# Global vector store instance (initialized at startup)
vector_store: Optional[LanceDB] = None

class QueryRequest(BaseModel):
    question: str = Field(.., min_length=1, max_length=500)
    k: int = Field(default=4, ge=1, le=20, description="Number of chunks to retrieve")
    temperature: float = Field(default=0.0, ge=0.0, le=2.0)

class QueryResponse(BaseModel):
    answer: str
    sources: List[str]
    confidence: float

@app.on_event("startup")
async def load_vector_store():
    """Initialize the vector store on application startup."""
    global vector_store
    db_path = os.getenv("DB_PATH", "./lancedb_store")
    table_name = os.getenv("TABLE_NAME", "documents")

    try:
        db = lancedb.connect(db_path)
        table = db.open_table(table_name)
        embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        vector_store = LanceDB(
            connection=db,
            table_name=table_name,
            embedding=embeddings
        )
        print(f"Loaded vector store with {table.count_rows()} vectors")
    except Exception as e:
        print(f"Failed to load vector store: {e}")
        # In production, you might want to exit or retry
        vector_store = None

@app.post("/query", response_model=QueryResponse)
async def query_documents(request: QueryRequest):
    """
    Answer a question using RAG.

    Args:
        request: QueryRequest with question and optional parameters

    Returns:
        QueryResponse with answer and source metadata

    Raises:
        HTTPException 503: If vector store is not loaded
        HTTPException 400: If no relevant documents found
    """
    if vector_store is None:
        raise HTTPException(
            status_code=503,
            detail="Vector store not initialized. Check server logs."
        )

    # Step 1: Retrieve relevant documents
    try:
        docs_with_scores = vector_store.similarity_search_with_score(
            request.question,
            k=request.k
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Retrieval failed: {str(e)}"
        )

    if not docs_with_scores:
        raise HTTPException(
            status_code=400,
            detail="No relevant documents found for the query."
        )

    # Step 2: Filter by relevance score (OpenAI embeddings return cosine similarity)
    # Scores range from 0 (dissimilar) to 1 (identical)
    relevant_docs = [
        (doc, score) for doc, score in docs_with_scores
        if score > 0.5  # Threshold for relevance
    ]

    if not relevant_docs:
        raise HTTPException(
            status_code=400,
            detail="No documents met the relevance threshold (0.5)."
        )

    # Step 3: Prepare context (handle token limits)
    context_parts = []
    total_chars = 0
    max_context_chars = 12000  # Approx 3000 tokens for GPT-4

    for doc, score in relevant_docs:
        chunk_text = doc.page_content
        if total_chars + len(chunk_text) > max_context_chars:
            break
        context_parts.append(chunk_text)
        total_chars += len(chunk_text)

    context = "\n\n---\n\n".join(context_parts)

    # Step 4: Generate answer
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant. Answer the question based on the provided context. "
                    "If the context doesn't contain the answer, say 'I cannot find this information in the documents.' "
                    "Do not make up information."),
        ("human", "Context:\n{context}\n\nQuestion: {question}")
    ])

    llm = ChatOpenAI(
        model="gpt-4o-mini",  # Cheaper than GPT-4, sufficient for RAG
        temperature=request.temperature,
        max_tokens=500
    )

    chain = prompt | llm
    response = chain.invoke({
        "context": context,
        "question": request.question
    })

    # Step 5: Extract sources
    sources = list(set([
        doc.metadata.get("source", "unknown")
        for doc, _ in relevant_docs
    ]))

    # Confidence is the average score of retrieved documents
    avg_confidence = sum(score for _, score in relevant_docs) / len(relevant_docs)

    return QueryResponse(
        answer=response.content,
        sources=sources,
        confidence=round(avg_confidence, 3)
    )

@app.get("/health")
async def health_check():
    """Simple health check endpoint."""
    return {"status": "healthy", "vector_store_loaded": vector_store is not None}

The similarity_search_with_score method returns both documents and their cosine similarity scores. The threshold of 0.5 is arbitrary—you should tune this based on your embedding model and data. For text-embedding-3-small, scores above 0.7 typically indicate strong relevance, but this varies by domain.

The context truncation logic (max_context_chars = 12000) prevents exceeding the LLM's token limit. GPT-4o-mini has a 128k token context window, but sending 100k tokens per query would be slow and expensive. Truncating to ~3000 tokens keeps response times under 2 seconds in production.

Pitfalls and Production Tips

1. Chunk Size and Overlap Trade-offs

The 1000-character chunk size with 200-character overlap works for general text, but legal documents with dense clauses might need smaller chunks (500 characters) to capture specific provisions. Test with your actual documents: if the LLM frequently says "### 2. LanceDB Schema Changes

LanceDB 0.12.x stores vectors in a fixed schema. If you change the embedding model (e.g., from text-embedding-3-small to text-embedding-3-large), the vector dimensions change (1536 vs 3072), and you cannot append to the existing table. You must create a new table or delete the old one. In production, version your tables:

table_name = f"documents_v{EMBEDDING_DIMENSION}"

3. Memory Usage with Large PDFs

PyPDFLoader loads entire PDFs into memory. A 500-page PDF with images can consume 2GB of RAM. For production, stream PDFs page-by-page using pypdf.PdfReader directly:

from pypdf import PdfReader

def stream_pdf_pages(pdf_path):
    reader = PdfReader(pdf_path)
    for page_num, page in enumerate(reader.pages):
        text = page.extract_text()
        if text.strip():  # Skip empty pages
            yield text, page_num + 1

4. API Rate Limits

OpenAI's embedding API has rate limits (typically 3,000 RPM for Tier 1 accounts). If you're ingesting 100,000 chunks, you'll hit this limit. Implement exponential backoff:

import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def embed_with_retry(texts, embeddings):
    return embeddings.embed_documents(texts)

5. Handling Empty or Malformed PDFs

The try-except block in load_and_chunk_pdfs catches errors per file. In production, log these failures to a separate file for manual review. Some PDFs are scanned images with no extractable text—you'll need OCR (e.g., Tesseract) for those.

What's Next

This pipeline handles the core RAG workflow, but production systems need more:

• Hybrid search: Combine vector similarity with keyword search (BM25) for better recall on exact matches. LangChain supports this via EnsembleRetriever.
• Caching: Cache frequent queries using Redis to reduce LLM costs. The langchain.cache module provides a RedisCache implementation.
• Monitoring: Track retrieval latency, LLM response times, and user satisfaction scores. Tools like LangSmith or Weights & Biases integrate with LangChain.

For further reading, check out our guides on optimizing vector store performance and handling multi-turn conversations in RAG.

The complete code is available on GitHub (note: this is a placeholder—the actual repository is not publicly documented).

---