How to Build a Production RAG Pipeline with LangChain and LanceDB

Table of Contents

• How to Build a Production RAG Pipeline with LangChain and LanceDB
  • Architecture Overview and Why LanceDB Matters
  • Prerequisites and Environment Setup
  • Core Implementation: Ingestion Pipeline
  • Core Implementation: Retrieval and Query Service
  • FastAPI Endpoint with Async Support
• CORS for frontend integration
• Global service instance
  • Pitfalls and Production Tips
• Process in batches of 100

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

You're building a retrieval-augmented generation system for a customer support application. The naive approach—stuffing documents into a prompt—breaks at 10 pages. The standard approach—using Chroma or FAISS—works until you need to update documents without rebuilding the entire index. This is where LanceDB changes the game.

LanceDB is a columnar vector database [1] built on the Lance columnar format. Unlike FAISS which stores vectors in memory or Chroma which uses SQLite, LanceDB persists vectors to disk with zero-copy reads and supports incremental updates without full re-indexing. According to the LanceDB documentation, it handles up to 100x more vectors per node than in-memory solutions for the same hardware cost.

We're building a production RAG [3] pipeline that ingests technical documentation, chunks it intelligently, embeds it with OpenAI's text-embedding-3-small, stores vectors in LanceDB, and serves queries through a FastAPI endpoint. The system handles concurrent users, document updates, and maintains sub-200ms query latency.

Architecture Overview and Why LanceDB Matters

The pipeline has four stages: ingestion, embedding, storage, and retrieval. Each stage must handle failure gracefully. Here's the stack:

• Ingestion: LangChain [8] document loaders with recursive chunking
• Embedding: OpenAI [7] text-embedding-3-small (1536 dimensions)
• Storage: LanceDB with IVF-PQ indexing for approximate nearest neighbor search
• Serving: FastAPI with async endpoints and connection pooling

LanceDB's key advantage is its columnar storage format. When you query for similar vectors, it only reads the vector columns, not the entire row. This means metadata fields like "source document" or "chunk index" don't slow down vector search. In production, this translates to 3-5x faster queries compared to row-based stores.

The database also supports multi-modal data natively. You can store vectors, text, images, and structured data in the same table. For our use case, this means we can attach metadata directly to vectors without a separate SQL database.

Prerequisites and Environment Setup

You need Python 3.10+ and an OpenAI API key. Install the dependencies:

pip install lancedb langchain langchain-openai fastapi uvicorn pydantic

Set your environment variables:

export OPENAI_API_KEY="sk-your-key-here"
export LANCEDB_URI="/data/vector-store"

The LanceDB URI points to a directory on disk. Unlike in-memory databases, LanceDB persists everything to this location. If the directory doesn't exist, LanceDB creates it on first write.

Core Implementation: Ingestion Pipeline

The ingestion pipeline loads documents, splits them into chunks, embeds each chunk, and stores the vectors. Here's the production-ready implementation:

import os
import hashlib
from typing import List, Dict, Any
from datetime import datetime

import lancedb
import pyarrow as pa
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import DirectoryLoader, TextLoader

class DocumentIngestionPipeline:
    """Handles document ingestion with deduplication and incremental updates."""

    def __init__(self, db_uri: str, table_name: str = "documents"):
        self.db = lancedb.connect(db_uri)
        self.table_name = table_name
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len,
            separators=["\n\n", "\n", " ", ""]
        )

    def compute_chunk_hash(self, text: str, metadata: Dict[str, Any]) -> str:
        """Generate a deterministic hash for deduplication."""
        content = f"{text}{metadata.get('source', '')}{metadata.get('page', '')}"
        return hashlib.sha256(content.encode()).hexdigest()

    def load_documents(self, directory_path: str) -> List[Dict[str, Any]]:
        """Load documents from directory with error handling."""
        loader = DirectoryLoader(
            directory_path,
            glob="/*.md",
            loader_cls=TextLoader,
            loader_kwargs={"encoding": "utf-8"},
            show_progress=True
        )

        try:
            documents = loader.load()
            print(f"Loaded {len(documents)} documents from {directory_path}")
            return documents
        except Exception as e:
            print(f"Failed to load documents: {e}")
            return []

    def chunk_documents(self, documents: List[Any]) -> List[Dict[str, Any]]:
        """Split documents into overlapping chunks with metadata."""
        chunks = []

        for doc in documents:
            doc_chunks = self.text_splitter.split_documents([doc])

            for chunk in doc_chunks:
                chunk_hash = self.compute_chunk_hash(
                    chunk.page_content, 
                    chunk.metadata
                )

                chunks.append({
                    "vector": None,  # Will be filled during embedding
                    "text": chunk.page_content,
                    "source": chunk.metadata.get("source", "unknown"),
                    "chunk_hash": chunk_hash,
                    "created_at": datetime.utcnow().isoformat()
                })

        print(f"Created {len(chunks)} chunks from {len(documents)} documents")
        return chunks

    def embed_chunks(self, chunks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Generate embeddings for all chunks with batching."""
        texts = [chunk["text"] for chunk in chunks]

        # OpenAI embeddings API has rate limits; batch in groups of 20
        batch_size = 20
        all_embeddings = []

        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            try:
                embeddings = self.embeddings.embed_documents(batch)
                all_embeddings.extend(embeddings)
                print(f"Embedded batch {i//batch_size + 1}/{(len(texts)-1)//batch_size + 1}")
            except Exception as e:
                print(f"Embedding failed for batch starting at index {i}: {e}")
                # Fill with zeros for failed embeddings
                all_embeddings.extend([[0.0] * 1536] * len(batch))

        for chunk, embedding in zip(chunks, all_embeddings):
            chunk["vector"] = embedding

        return chunks

    def store_vectors(self, chunks: List[Dict[str, Any]]) -> int:
        """Store vectors in LanceDB with deduplication."""
        if not chunks:
            return 0

        # Create PyArrow schema for the table
        schema = pa.schema([
            pa.field("vector", pa.list_(pa.float32(), 1536)),
            pa.field("text", pa.string()),
            pa.field("source", pa.string()),
            pa.field("chunk_hash", pa.string()),
            pa.field("created_at", pa.string())
        ])

        # Check if table exists
        try:
            table = self.db.open_table(self.table_name)
            existing_hashes = set(
                table.search().limit(0).to_pandas()["chunk_hash"].tolist()
            )
        except Exception:
            table = self.db.create_table(self.table_name, schema=schema)
            existing_hashes = set()

        # Filter out duplicates
        new_chunks = [
            chunk for chunk in chunks 
            if chunk["chunk_hash"] not in existing_hashes
        ]

        if not new_chunks:
            print("No new chunks to add")
            return 0

        # Convert to PyArrow table and add
        data = {
            "vector": [chunk["vector"] for chunk in new_chunks],
            "text": [chunk["text"] for chunk in new_chunks],
            "source": [chunk["source"] for chunk in new_chunks],
            "chunk_hash": [chunk["chunk_hash"] for chunk in new_chunks],
            "created_at": [chunk["created_at"] for chunk in new_chunks]
        }

        arrow_table = pa.Table.from_pydict(data, schema=schema)
        table.add(arrow_table)

        print(f"Added {len(new_chunks)} new vectors to {self.table_name}")
        return len(new_chunks)

    def run(self, directory_path: str) -> int:
        """Execute the full ingestion pipeline."""
        documents = self.load_documents(directory_path)
        if not documents:
            return 0

        chunks = self.chunk_documents(documents)
        chunks = self.embed_chunks(chunks)
        count = self.store_vectors(chunks)

        return count

The deduplication logic is critical. Without it, re-running ingestion creates duplicate vectors. The chunk hash combines text content with source metadata, so updating a document changes its hash and creates a new entry. Old entries remain until garbage collection.

Core Implementation: Retrieval and Query Service

The retrieval service handles semantic search and returns context for the LLM. This implementation uses LanceDB's built-in vector search with configurable parameters:

from typing import List, Optional, Tuple
import numpy as np
from pydantic import BaseModel

class SearchResult(BaseModel):
    text: str
    source: str
    score: float
    chunk_hash: str

class RetrievalService:
    """Handles vector search and context retrieval."""

    def __init__(self, db_uri: str, table_name: str = "documents"):
        self.db = lancedb.connect(db_uri)
        self.table = self.db.open_table(table_name)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    def search(
        self, 
        query: str, 
        k: int = 5,
        score_threshold: float = 0.5
    ) -> List[SearchResult]:
        """
        Perform semantic search with score filtering.

        Args:
            query: Natural language query
            k: Number of results to return
            score_threshold: Minimum similarity score (0-1)

        Returns:
            List of SearchResult objects sorted by relevance
        """
        # Embed the query
        query_vector = self.embeddings.embed_query(query)

        # Perform search with LanceDB
        results = (
            self.table.search(query_vector)
            .limit(k * 2)  # Fetch more for post-filtering
            .to_pandas()
        )

        # Post-filter by score threshold
        filtered_results = []
        for _, row in results.iterrows():
            score = row.get("_distance", 0)
            # LanceDB returns distance; convert to similarity
            similarity = 1.0 / (1.0 + score)

            if similarity >= score_threshold:
                filtered_results.append(SearchResult(
                    text=row["text"],
                    source=row["source"],
                    score=similarity,
                    chunk_hash=row["chunk_hash"]
                ))

        # Return top k after filtering
        return filtered_results[:k]

    def build_context(self, results: List[SearchResult]) -> str:
        """Format search results as context for LLM."""
        context_parts = []

        for i, result in enumerate(results, 1):
            context_parts.append(
                f"[Source {i}: {result.source}]\n{result.text}\n"
            )

        return "\n---\n".join(context_parts)

The score threshold prevents irrelevant chunks from polluting the context. A threshold of 0.5 works well for technical documentation; you might lower it to 0.3 for creative tasks where relevance is less strict.

FastAPI Endpoint with Async Support

The API layer wraps the retrieval service with proper error handling and rate limiting:

from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import asyncio
from typing import Optional

app = FastAPI(title="RAG Pipeline API")

# CORS for frontend integration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# Global service instance
retrieval_service: Optional[RetrievalService] = None

@app.on_event("startup")
async def startup_event():
    global retrieval_service
    db_uri = os.getenv("LANCEDB_URI", "/data/vector-store")
    retrieval_service = RetrievalService(db_uri)

class QueryRequest(BaseModel):
    query: str
    k: int = 5
    score_threshold: float = 0.5

class QueryResponse(BaseModel):
    results: List[SearchResult]
    context: str
    query_time_ms: float

@app.post("/query", response_model=QueryResponse)
async def query_endpoint(request: QueryRequest):
    """Main query endpoint with timing."""
    if not retrieval_service:
        raise HTTPException(status_code=503, detail="Service not initialized")

    start_time = asyncio.get_event_loop().time()

    try:
        results = retrieval_service.search(
            query=request.query,
            k=request.k,
            score_threshold=request.score_threshold
        )

        context = retrieval_service.build_context(results)

        query_time = (asyncio.get_event_loop().time() - start_time) * 1000

        return QueryResponse(
            results=results,
            context=context,
            query_time_ms=round(query_time, 2)
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/ingest")
async def ingest_endpoint(directory: str):
    """Trigger document ingestion."""
    pipeline = DocumentIngestionPipeline(
        db_uri=os.getenv("LANCEDB_URI", "/data/vector-store")
    )

    try:
        count = pipeline.run(directory)
        return {"status": "success", "documents_ingested": count}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

The async endpoint uses asyncio.get_event_loop().time() for precise timing. This matters when you're monitoring SLAs—FastAPI's built-in timing doesn't account for network overhead.

Pitfalls and Production Tips

1. Embedding Rate Limits

OpenAI's embedding API has rate limits of 3,000 RPM for tier 1 accounts. Our batch size of 20 keeps us under this limit, but you should add exponential backoff for production. The tenacity library works well here:

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(self, texts):
    return self.embeddings.embed_documents(texts)

2. Memory Management with Large Corpora

Loading 10,000 documents into memory simultaneously will OOM a 16GB machine. Stream documents using LangChain's lazy loading:

loader = DirectoryLoader(
    directory_path,
    loader_cls=TextLoader,
    loader_kwargs={"encoding": "utf-8"},
    show_progress=True,
    use_multithreading=True
)

# Process in batches of 100
for batch in loader.lazy_load():
    chunks = self.chunk_documents([batch])
    # Process immediately instead of accumulating

3. LanceDB Index Configuration

By default, LanceDB uses brute-force search. For tables with more than 10,000 vectors, create an IVF-PQ index:

table.create_index(
    metric="cosine",
    num_partitions=256,  # IVF centroids
    num_sub_vectors=96,  # PQ compression
    index_cache_size=100_000_000  # 100MB cache
)

This reduces search latency from O(n) to O(log n) with minimal accuracy loss. The trade-off is 2-5% recall degradation for 10x speed improvement.

4. Concurrent Write Conflicts

LanceDB supports multiple readers but single writer. If you have multiple ingestion workers, use a file lock:

import fcntl

with open("/tmp/lancedb.lock", "w") as lock_file:
    fcntl.flock(lock_file, fcntl.LOCK_EX)
    table.add(new_data)
    fcntl.flock(lock_file, fcntl.LOCK_UN)

5. Context Window Management

The LLM's context window limits how many chunks you can include. For GPT-4 with 8K context, limit to 5 chunks of 1000 tokens each. Track token usage:

import tiktoken

def count_tokens(text: str) -> int:
    encoding = tiktoken.encoding_for_model("gpt-4")
    return len(encoding.encode(text))

# In build_context, enforce token limit
total_tokens = sum(count_tokens(r.text) for r in results)
if total_tokens > 6000:  # Leave room for system prompt and query
    results = results[:3]  # Reduce to fit

What's Next

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

• Hybrid search: Combine vector search with BM25 keyword search for better recall on exact matches. LanceDB supports full-text search natively.
• Document versioning: Track which version of a document produced each chunk. When documents update, mark old chunks as stale and exclude them from search.
• Monitoring: Log query latency, retrieval scores, and user feedback. Use this data to tune chunk sizes and score thresholds.

The complete code is available on GitHub. For more on vector database architecture, see our guide on choosing between vector databases. If you're scaling beyond 10 million vectors, check out our production RAG patterns.

The key insight from building this system: vector search is not a magic bullet. The quality of your chunks, the relevance of your metadata, and the robustness of your error handling matter more than the database you choose. LanceDB makes the storage part easy—the hard work is in the preprocessing and monitoring.

---