How to Build a RAG Pipeline with LanceDB and LangChain

Table of Contents

• How to Build a RAG Pipeline with LanceDB and LangChain
  • Understanding the LanceDB Advantage for Production RAG
  • Prerequisites and Environment Setup
  • Building the Document Ingestion Pipeline
• ingestion.py
  • Implementing the Retrieval and Generation System
• rag_pipeline.py
  • Exposing the RAG Pipeline as a Production API
• api.py
• Global instances (initialized at startup)

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

Retrieval-Augmented Generation (RAG) has become the standard architecture for production AI systems that need to answer questions based on proprietary data. While vector databases like Pinecone [8] and Weaviate dominate the conversation, LanceDB offers a compelling alternative: an embedded, columnar vector database built on Lance columnar format that eliminates network latency and reduces operational complexity.

In this tutorial, we'll build a production-ready RAG pipeline using LanceDB as our vector store and LangChain for orchestration. By the end, you'll have a system that can ingest PDF documents, chunk them intelligently, generate embeddings, and answer questions with source citations—all running locally without external dependencies.

Understanding the LanceDB Advantage for Production RAG

Before diving into code, let's understand why LanceDB deserves consideration in your architecture decisions. Traditional vector databases operate as separate services requiring network calls for every query. This introduces latency, operational overhead, and potential failure points. LanceDB takes a different approach: it's an embedded database that stores vectors and metadata in Lance columnar format files on disk.

The key architectural implications are significant:

• Zero network latency: All operations happen in-process, reducing query latency by 30-50ms per call compared to client-server architectures
• No separate infrastructure: No need to manage a database cluster, reducing DevOps burden
• Columnar storage: Lance format enables efficient filtering and projection, meaning you can store rich metadata alongside vectors without performance penalties
• Disk-based storage: Unlike in-memory solutions, LanceDB can handle datasets that exceed available RAM

According to the LanceDB documentation, the database supports up to 100K vectors per index by default, with configurable limits for larger datasets. For production workloads exceeding this, you can partition data across multiple tables or use the IVF-PQ index configuration.

Prerequisites and Environment Setup

We'll need Python 3.10+ and several libraries. Create a fresh virtual environment to avoid dependency conflicts:

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

pip install lancedb langchain langchain-community pypdf sentence-transformers [4] fastapi uvicorn

The key dependencies serve specific purposes:

• lancedb: The vector database itself, version 0.6.0 or later
• langchain: Orchestration framework for chaining LLM calls with retrieval
• pypdf: PDF parsing for document ingestion
• sentence-transformers: Local embedding model (all-MiniLM-L6-v2) that runs on CPU
• fastapi/uvicorn: For exposing our RAG pipeline as an API endpoint

For the embedding model, we'll use all-MiniLM-L6-v2 from SentenceTransformers. This model produces 384-dimensional embeddings and runs efficiently on CPU, making it suitable for production deployments without GPU requirements. According to the SentenceTransformers documentation, this model achieves a solid balance between speed and accuracy for semantic search tasks.

Building the Document Ingestion Pipeline

The first component of our RAG system is the ingestion pipeline. This handles PDF parsing, intelligent chunking, embedding generation, and storage in LanceDB. Let's build it step by step.

# ingestion.py
import os
import uuid
from typing import List, Dict, Any
from pathlib import Path

import lancedb
import pyarrow as pa
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader
from sentence_transformers import SentenceTransformer
import numpy as np

class DocumentIngestor:
    """
    Production-grade document ingestion pipeline for LanceDB RAG.

    Handles PDF parsing, intelligent chunking, embedding generation,
    and vector storage with metadata preservation.
    """

    def __init__(self, db_path: str = "./lancedb_data", 
                 table_name: str = "documents",
                 chunk_size: int = 512,
                 chunk_overlap: int = 64,
                 model_name: str = "all-MiniLM-L6-v2"):

        self.db_path = db_path
        self.table_name = table_name
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

        # Initialize embedding model - loads once and caches
        self.embedding_model = SentenceTransformer(model_name)

        # Connect to LanceDB (creates directory if not exists)
        self.db = lancedb.connect(db_path)

        # Initialize text splitter with sensible defaults
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            separators=["\n\n", "\n", ".", " ", ""]
        )

    def _create_table_schema(self):
        """
        Define the schema for our vector store.
        LanceDB requires explicit schema definition for typed columns.
        """
        schema = pa.schema([
            pa.field("id", pa.string()),
            pa.field("vector", pa.list_(pa.float32(), 384)),  # 384-dim embeddings
            pa.field("text", pa.string()),
            pa.field("source", pa.string()),
            pa.field("page_number", pa.int32()),
            pa.field("chunk_index", pa.int32()),
            pa.field("total_chunks", pa.int32()),
        ])
        return schema

    def ingest_pdf(self, pdf_path: str) -> int:
        """
        Ingest a single PDF file into LanceDB.

        Args:
            pdf_path: Path to the PDF file

        Returns:
            Number of chunks ingested
        """
        if not os.path.exists(pdf_path):
            raise FileNotFoundError(f"PDF not found: {pdf_path}")

        # Load PDF with page-level metadata
        loader = PyPDFLoader(pdf_path)
        documents = loader.load()

        # Extract source filename for metadata
        source = Path(pdf_path).name

        all_chunks = []

        for doc in documents:
            page_num = doc.metadata.get("page", 1)
            page_text = doc.page_content

            # Skip empty pages
            if not page_text.strip():
                continue

            # Split page into chunks
            chunks = self.text_splitter.split_text(page_text)

            for idx, chunk_text in enumerate(chunks):
                all_chunks.append({
                    "text": chunk_text,
                    "source": source,
                    "page_number": page_num,
                    "chunk_index": idx,
                    "total_chunks": len(chunks)
                })

        if not all_chunks:
            print(f"Warning: No content extracted from {pdf_path}")
            return 0

        # Generate embeddings in batch for efficiency
        texts = [chunk["text"] for chunk in all_chunks]
        embeddings = self.embedding_model.encode(
            texts, 
            show_progress_bar=True,
            normalize_embeddings=True  # Important for cosine similarity
        )

        # Prepare records for LanceDB
        records = []
        for chunk, embedding in zip(all_chunks, embeddings):
            records.append({
                "id": str(uuid.uuid4()),
                "vector": embedding.tolist(),
                "text": chunk["text"],
                "source": chunk["source"],
                "page_number": chunk["page_number"],
                "chunk_index": chunk["chunk_index"],
                "total_chunks": chunk["total_chunks"]
            })

        # Create or append to table
        try:
            table = self.db.open_table(self.table_name)
            table.add(records)
        except Exception:
            # Table doesn't exist yet, create it
            schema = self._create_table_schema()
            table = self.db.create_table(
                self.table_name, 
                data=records,
                schema=schema
            )

        print(f"Ingested {len(records)} chunks from {source}")
        return len(records)

    def ingest_directory(self, directory_path: str) -> Dict[str, int]:
        """
        Ingest all PDFs in a directory.

        Args:
            directory_path: Path to directory containing PDF files

        Returns:
            Dictionary mapping filename to chunk count
        """
        pdf_files = list(Path(directory_path).glob("*.pdf"))
        results = {}

        for pdf_file in pdf_files:
            try:
                chunk_count = self.ingest_pdf(str(pdf_file))
                results[pdf_file.name] = chunk_count
            except Exception as e:
                print(f"Error ingesting {pdf_file.name}: {e}")
                results[pdf_file.name] = -1

        return results

The ingestion pipeline handles several edge cases that appear in production:

• Empty pages: PDFs often contain blank pages or scanned images with no extractable text. We skip these gracefully.
• Batch embedding: Generating embeddings one at a time is slow. We batch all chunks from a document for efficient GPU/CPU utilization.
• Schema enforcement: LanceDB requires consistent schema across all inserts. We define this explicitly to catch type mismatches early.
• UUID-based deduplication: Each chunk gets a unique ID, preventing duplicate entries if ingestion is run multiple times.

The chunk size of 512 characters with 64-character overlap is a starting point. For technical documentation, you might increase to 1024 characters. For conversational data, 256 characters often works better. The overlap ensures that semantically related content isn't split across chunk boundaries.

Implementing the Retrieval and Generation System

With documents ingested, we need a retrieval system that finds relevant chunks and an LLM that generates answers. Let's build the query pipeline.

# rag_pipeline.py
from typing import List, Tuple, Optional, Dict, Any
import lancedb
from sentence_transformers import SentenceTransformer
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import Document
import numpy as np

class RAGPipeline:
    """
    Production RAG pipeline with LanceDB retrieval and LLM generation.

    Handles query embedding, vector search, context assembly,
    and answer generation with source attribution.
    """

    def __init__(self, 
                 db_path: str = "./lancedb_data",
                 table_name: str = "documents",
                 model_name: str = "all-MiniLM-L6-v2",
                 llm_model: str = "gpt [5]-4o-mini",
                 top_k: int = 5,
                 temperature: float = 0.3):

        self.db_path = db_path
        self.table_name = table_name
        self.top_k = top_k

        # Initialize embedding model (same as ingestion)
        self.embedding_model = SentenceTransformer(model_name)

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

        try:
            self.table = self.db.open_table(table_name)
        except Exception as e:
            raise RuntimeError(
                f"Table '{table_name}' not found. Run ingestion first. Error: {e}"
            )

        # Initialize LLM with conservative temperature for factual answers
        self.llm = ChatOpenAI(
            model=llm_model,
            temperature=temperature,
            max_tokens=1024
        )

        # Define the RAG prompt template
        self.prompt_template = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful assistant that answers questions based on provided context.

Rules:
1. Only use information from the provided context to answer questions.
2. If the context doesn't contain enough information, say "I don't have enough information to answer this question."
3. Always cite the source document and page number for each piece of information.
4. Be concise but thorough in your answers.
5. If the question is ambiguous, ask for clarification.

Context:
{context}"""),
            ("human", "{question}")
        ])

    def retrieve(self, query: str) -> List[Dict[str, Any]]:
        """
        Retrieve relevant chunks from LanceDB using vector similarity.

        Args:
            query: Natural language query string

        Returns:
            List of relevant chunks with metadata
        """
        # Generate query embedding
        query_embedding = self.embedding_model.encode(
            query,
            normalize_embeddings=True
        )

        # Perform vector search in LanceDB
        # Using cosine similarity (since embeddings are normalized)
        results = (
            self.table.search(query_embedding.tolist())
            .limit(self.top_k)
            .to_list()
        )

        return results

    def generate(self, query: str, context_chunks: List[Dict[str, Any]]) -> str:
        """
        Generate an answer using the LLM with retrieved context.

        Args:
            query: Original user query
            context_chunks: Retrieved chunks from vector search

        Returns:
            Generated answer with citations
        """
        # Format context for the prompt
        context_parts = []
        for i, chunk in enumerate(context_chunks, 1):
            source_info = f"[Source: {chunk['source']}, Page {chunk['page_number']}]"
            context_parts.append(f"Document {i} {source_info}:\n{chunk['text']}\n")

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

        # Generate response
        messages = self.prompt_template.format_messages(
            context=context,
            question=query
        )

        response = self.llm.invoke(messages)
        return response.content

    def query(self, query: str) -> Dict[str, Any]:
        """
        End-to-end RAG query: retrieve then generate.

        Args:
            query: Natural language question

        Returns:
            Dictionary with answer and source documents
        """
        # Retrieve relevant chunks
        chunks = self.retrieve(query)

        if not chunks:
            return {
                "answer": "No relevant documents found in the knowledge base.",
                "sources": []
            }

        # Generate answer
        answer = self.generate(query, chunks)

        # Extract source information for transparency
        sources = [
            {
                "source": chunk["source"],
                "page": chunk["page_number"],
                "text_preview": chunk["text"][:200] + ".."
            }
            for chunk in chunks
        ]

        return {
            "answer": answer,
            "sources": sources
        }

The retrieval pipeline uses normalized embeddings for cosine similarity search. LanceDB's default metric is cosine similarity when embeddings are normalized, which is why we normalize during both ingestion and query. This ensures consistent similarity calculations.

The top_k parameter of 5 is a starting point. In production, you might adjust this based on your chunk size and the complexity of questions. For technical documentation, 3-5 chunks usually provide sufficient context. For broader questions, 7-10 chunks might be necessary.

Exposing the RAG Pipeline as a Production API

For real-world use, we need to expose our RAG pipeline as a REST API. FastAPI provides the performance and validation we need.

# api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
import uvicorn

from ingestion import DocumentIngestor
from rag_pipeline import RAGPipeline

app = FastAPI(
    title="LanceDB RAG API",
    description="Production RAG pipeline with LanceDB vector store",
    version="1.0.0"
)

# Global instances (initialized at startup)
ingestor: Optional[DocumentIngestor] = None
rag: Optional[RAGPipeline] = None

class QueryRequest(BaseModel):
    question: str = Field(.., min_length=1, max_length=2000)
    top_k: Optional[int] = Field(default=5, ge=1, le=20)

class IngestRequest(BaseModel):
    pdf_path: str = Field(.., min_length=1)

class QueryResponse(BaseModel):
    answer: str
    sources: List[Dict[str, Any]]

@app.on_event("startup")
async def startup_event():
    """Initialize the RAG pipeline on server start."""
    global ingestor, rag
    try:
        ingestor = DocumentIngestor()
        rag = RAGPipeline()
        print("RAG pipeline initialized successfully")
    except Exception as e:
        print(f"Failed to initialize RAG pipeline: {e}")
        raise

@app.post("/ingest", status_code=201)
async def ingest_document(request: IngestRequest):
    """
    Ingest a PDF document into the vector store.

    Args:
        request: Contains pdf_path to the document

    Returns:
        Status of ingestion with chunk count
    """
    if not ingestor:
        raise HTTPException(status_code=503, detail="Ingestor not initialized")

    try:
        chunk_count = ingestor.ingest_pdf(request.pdf_path)
        return {
            "status": "success",
            "chunks_ingested": chunk_count,
            "pdf_path": request.pdf_path
        }
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail=f"PDF not found: {request.pdf_path}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

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

    Args:
        request: Contains question and optional top_k parameter

    Returns:
        Answer with source citations
    """
    if not rag:
        raise HTTPException(status_code=503, detail="RAG pipeline not initialized")

    try:
        result = rag.query(request.question)
        return QueryResponse(
            answer=result["answer"],
            sources=result["sources"]
        )
    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",
        "ingestor_ready": ingestor is not None,
        "rag_ready": rag is not None
    }

if __name__ == "__main__":
    uvicorn.run(
        "api:app",
        host="0.0.0.0",
        port=8000,
        reload=False,  # Disable in production
        workers=1  # LanceDB is not thread-safe for writes
    )

The API design addresses several production concerns:

• Startup initialization: The RAG pipeline initializes once at startup, avoiding repeated model loading overhead
• Input validation: Pydantic models enforce constraints like minimum query length and maximum top_k
• Graceful error handling: Each endpoint returns appropriate HTTP status codes and error messages
• Health check: The /health endpoint enables load balancers to verify service availability

One critical note: LanceDB is not thread-safe for write operations. In production, you should either use a single worker process for writes or implement a write queue. The API above uses a single worker, which is acceptable for moderate traffic but may need scaling considerations for high-throughput scenarios.

Edge Cases and Production Considerations

Building a production RAG system requires handling numerous edge cases that simple tutorials ignore. Here are the critical ones we've addressed and others you'll need to consider:

Empty or unparseable documents: PDFs can be scanned images, corrupted, or password-protected. Our ingestion pipeline catches these with try-except blocks, but you should add explicit checks for PDF validity before attempting to parse.

Memory management: The embedding model loads into memory and stays there. For production systems handling multiple document types, consider using a model server like Hugging Face Text Generation Inference to offload model memory.

Query ambiguity: Users often ask vague questions. Our prompt template instructs the LLM to ask for clarification, but you might implement a query rewriting step that expands abbreviations or corrects typos before retrieval.

Cold start problem: When the system has no documents ingested, queries return empty results. The health check endpoint helps monitoring systems detect this state, but you should also implement a minimum document threshold before enabling the query endpoint.

Vector index maintenance: LanceDB builds indexes lazily. For large datasets (100K+ vectors), you should trigger index creation explicitly after bulk ingestion to maintain query performance.

What's Next

This tutorial provides a production-ready foundation for RAG with LanceDB, but real-world systems require additional components:

1. Document versioning: Implement a versioning system that tracks document updates and invalidates old embeddings
2. Hybrid search: Combine vector similarity with keyword search (BM25) for better retrieval accuracy on exact matches
3. Query decomposition: Break complex questions into sub-questions and aggregate results
4. Feedback loop: Collect user feedback on answer quality to fine-tune chunking and retrieval parameters

For further reading, explore LanceDB's documentation on indexing strategies and LangChain's guide on advanced retrieval techniques. These resources will help you optimize your pipeline for production workloads with millions of documents.

The complete code for this tutorial is available in a single repository structure: ingestion.py, rag_pipeline.py, and api.py. Run python api.py to start the server, then use curl or any HTTP client to ingest documents and query your knowledge base.

---