How to Build a RAG Pipeline with LanceDB and LangChain

Table of Contents

• How to Build a RAG Pipeline with LanceDB and LangChain
  • Why LanceDB for Production RAG
  • Prerequisites and Environment Setup
  • Core Implementation: Document Ingestion Pipeline
  • Retrieval with Hybrid Search and Metadata Filtering
  • Pitfalls and Production Tips
  • What's Next

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

Building production retrieval-augmented generation (RAG) systems requires careful consideration of vector storage, retrieval speed, and cost management. After spending the last year deploying RAG pipelines for enterprise clients, I've found that most tutorials skip the hard parts: handling concurrent writes, managing embedding costs at scale, and dealing with real-world data quality issues.

This tutorial walks through building a production-grade RAG pipeline using LanceDB as the vector store and LangChain [7] for orchestration. LanceDB's columnar storage format and zero-copy reads make it particularly suitable for large-scale deployments where you need to iterate on embeddings without re-ingesting data.

Why LanceDB for Production RAG

LanceDB stores vectors in Apache Lance format, which gives you two critical advantages over traditional vector database [2]s. First, you can add new embedding columns without rewriting existing data - this matters when you upgrade from OpenAI's text-embedding-ada-002 to text-embedding-3-large and don't want to re-embed 10 million documents. Second, LanceDB supports blob storage backends (S3, GCS, Azure Blob) natively, meaning your vector store can scale to hundreds of terabytes without managing separate infrastructure.

The trade-off is that LanceDB is relatively new compared to Pinecone or Weaviate [8]. As of July 2026, the Python SDK is at version 0.12.x and the Rust core is still evolving. You'll encounter breaking changes between minor versions, so pin your dependencies.

Prerequisites and Environment Setup

You need Python 3.10+ and at least 8GB RAM for the embedding models we'll use. Start with a clean virtual environment:

python -m venv rag-lancedb
source rag-lancedb/bin/activate
pip install lancedb==0.12.0 langchain==0.3.1 langchain-openai==0.2.0 pypdf==4.2.0 tiktoken==0.7.0

Set your OpenAI API key as an environment variable. The embedding model we'll use costs $0.13 per 1M tokens for text-embedding-3-small. For a 10,000 document corpus averaging 500 tokens each, that's roughly $0.65 in embedding costs.

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

Core Implementation: Document Ingestion Pipeline

The ingestion pipeline handles three failure modes that crash naive implementations: PDF parsing errors, rate limiting from embedding APIs, and partial writes to the vector store.

import lancedb
import pyarrow as pa
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
import hashlib
from typing import List, Dict
import time

class ProductionRAGIngestor:
    def __init__(self, db_path: str = "./lancedb", table_name: str = "documents"):
        self.db = lancedb.connect(db_path)
        self.table_name = table_name
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            dimensions=1536  # Explicit dimension for schema enforcement
        )
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", ".", " ", ""],
            length_function=len
        )

    def create_schema(self):
        """Define schema upfront to catch type mismatches at write time."""
        schema = pa.schema([
            pa.field("vector", pa.list_(pa.float32(), 1536)),
            pa.field("text", pa.string()),
            pa.field("source", pa.string()),
            pa.field("chunk_id", pa.string()),
            pa.field("doc_hash", pa.string()),
            pa.field("created_at", pa.timestamp('us'))
        ])

        try:
            self.db.create_table(self.table_name, schema=schema, mode="overwrite")
        except Exception as e:
            # Table already exists - this is fine in production
            if "already exists" in str(e):
                pass
            else:
                raise e

    def process_pdf(self, pdf_path: str) -> List[Dict]:
        """Extract text with error handling for corrupted PDFs."""
        try:
            loader = PyPDFLoader(pdf_path)
            documents = loader.load()
        except Exception as e:
            print(f"Failed to parse {pdf_path}: {e}")
            return []

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

        chunks = self.text_splitter.split_text(full_text)
        doc_hash = hashlib.sha256(full_text.encode()).hexdigest()[:16]

        records = []
        for i, chunk in enumerate(chunks):
            records.append({
                "text": chunk,
                "source": pdf_path,
                "chunk_id": f"{doc_hash}_{i}",
                "doc_hash": doc_hash,
                "created_at": int(time.time())
            })
        return records

    def embed_with_retry(self, texts: List[str], max_retries: int = 3) -> List[List[float]]:
        """Handle OpenAI rate limits with exponential backoff."""
        for attempt in range(max_retries):
            try:
                return self.embeddings.embed_documents(texts)
            except Exception as e:
                if "rate limit" in str(e).lower():
                    wait_time = (2 ** attempt) * 5
                    print(f"Rate limited, waiting {wait_time}s..")
                    time.sleep(wait_time)
                else:
                    raise e
        raise Exception("Max retries exceeded for embedding")

    def ingest_batch(self, records: List[Dict], batch_size: int = 100):
        """Batch ingestion with progress tracking and partial failure handling."""
        table = self.db.open_table(self.table_name)

        for i in range(0, len(records), batch_size):
            batch = records[i:i + batch_size]
            texts = [r["text"] for r in batch]

            try:
                vectors = self.embed_with_retry(texts)
                for j, vec in enumerate(vectors):
                    batch[j]["vector"] = vec

                # Convert to Arrow table for LanceDB
                import pyarrow as pa
                arrow_table = pa.Table.from_pylist(batch)
                table.add(arrow_table)

                print(f"Ingested {min(i + batch_size, len(records))}/{len(records)} chunks")

            except Exception as e:
                print(f"Batch {i} failed: {e}")
                # Don't crash - log and continue
                continue

The key design decisions here are explicit schema enforcement and batch-level error handling. LanceDB doesn't enforce schema by default - if you accidentally pass a string where a float array is expected, you'll get a cryptic Arrow error at write time. The create_schema method prevents that.

The embed_with_retry method handles OpenAI's rate limits, which return HTTP 429 errors. The exponential backoff starts at 5 seconds and caps at 20 seconds. For production systems handling millions of documents, you'd want to use a queue-based architecture instead, but this pattern works for datasets up to about 100,000 documents.

Retrieval with Hybrid Search and Metadata Filtering

Pure vector search fails on exact keyword matches. If someone searches for "Python 3.12 release date" and your documents contain "Python 3.12 was released on October 2, 2023", the vector similarity might miss the exact date. LanceDB supports full-text search via Tantivy integration, but as of version 0.12.0, this requires building from source with the tantivy feature flag.

Instead, we'll implement a hybrid approach: vector search for semantic similarity, then re-rank results using keyword overlap.

from langchain_openai import ChatOpenAI
from langchain.schema import Document
import numpy as np
from typing import List, Tuple

class ProductionRAGRetriever:
    def __init__(self, db_path: str = "./lancedb", table_name: str = "documents"):
        self.db = lancedb.connect(db_path)
        self.table = self.db.open_table(table_name)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)

    def hybrid_search(self, query: str, k: int = 10, alpha: float = 0.7) -> List[Document]:
        """
        Hybrid search combining vector similarity and keyword overlap.
        alpha controls the weight: 1.0 = pure vector, 0.0 = pure keyword.
        """
        # Vector search
        query_vector = self.embeddings.embed_query(query)
        vector_results = self.table.search(query_vector).limit(k * 2).to_list()

        # Keyword scoring using simple token overlap
        query_tokens = set(query.lower().split())
        scored_docs = []

        for doc in vector_results:
            vector_score = doc["_distance"]  # LanceDB returns distance, not similarity
            doc_tokens = set(doc["text"].lower().split())

            # Jaccard similarity for keyword overlap
            if len(query_tokens) > 0 and len(doc_tokens) > 0:
                keyword_score = len(query_tokens & doc_tokens) / len(query_tokens | doc_tokens)
            else:
                keyword_score = 0.0

            # Normalize vector distance to similarity (LanceDB uses L2 by default)
            vector_similarity = 1.0 / (1.0 + vector_score)

            combined_score = alpha * vector_similarity + (1 - alpha) * keyword_score
            scored_docs.append((combined_score, doc))

        # Sort by combined score and take top k
        scored_docs.sort(key=lambda x: x[0], reverse=True)
        top_docs = scored_docs[:k]

        return [
            Document(
                page_content=doc["text"],
                metadata={
                    "source": doc["source"],
                    "chunk_id": doc["chunk_id"],
                    "relevance_score": score
                }
            )
            for score, doc in top_docs
        ]

    def query_with_sources(self, query: str, k: int = 5) -> Tuple[str, List[Document]]:
        """Retrieve documents and generate answer with citations."""
        docs = self.hybrid_search(query, k=k)

        context = "\n\n".join([
            f"[Source: {doc.metadata['source']}]\n{doc.page_content}"
            for doc in docs
        ])

        prompt = f"""Answer the question based on the provided context. 
        If the context doesn't contain enough information, say so.
        Always cite the source document for each claim.

        Context:
        {context}

        Question: {query}

        Answer:"""

        response = self.llm.invoke(prompt)
        return response.content, docs

The alpha parameter in hybrid_search lets you tune the balance between semantic and keyword matching. In my testing, 0.7 works well for technical documentation where exact terms matter, while 0.5 is better for conversational queries. You'll want to A/B test this on your specific dataset.

One gotcha: LanceDB's _distance field uses L2 distance by default. The conversion to similarity (1 / (1 + distance)) is a heuristic - for production systems, you should normalize distances across your dataset to get consistent similarity scores.

Pitfalls and Production Tips

Memory management with large datasets: LanceDB loads the entire index into memory by default. For a 10 million document corpus with 1536-dimensional vectors, that's roughly 60GB of RAM. Use the read_consistency_interval parameter to control how often LanceDB refreshes from disk:

table = self.db.open_table(
    "documents",
    read_consistency_interval=60  # Refresh from disk every 60 seconds
)

Embedding cost optimization: OpenAI's text-embedding-3-small costs $0.13/1M tokens. For a 100,000 document corpus averaging 1000 tokens per chunk, that's $13 per full re-embed. Cache embeddings by document hash to avoid re-embedding unchanged documents:

def get_cached_embedding(self, text: str) -> List[float]:
    text_hash = hashlib.sha256(text.encode()).hexdigest()
    # Check if we already have this embedding
    existing = self.table.search().where(f"text_hash = '{text_hash}'").limit(1).to_list()
    if existing:
        return existing[0]["vector"]
    return None

Concurrent write conflicts: LanceDB uses optimistic concurrency control. If two processes write to the same table simultaneously, one will fail with a version conflict. Use a write lock or queue writes through a single process:

import fcntl

def safe_write(self, records):
    with open("/tmp/lancedb_write.lock", "w") as lockfile:
        fcntl.flock(lockfile, fcntl.LOCK_EX)
        try:
            table = self.db.open_table(self.table_name)
            arrow_table = pa.Table.from_pylist(records)
            table.add(arrow_table)
        finally:
            fcntl.flock(lockfile, fcntl.LOCK_UN)

Schema evolution: LanceDB supports adding columns without rewriting data, but you cannot remove columns. Plan your schema carefully. If you need to remove a column, create a new table and copy the data.

What's Next

This pipeline handles the core RAG workflow, but production systems need monitoring, A/B testing frameworks, and cost tracking. Consider adding:

• Embedding cost tracking per query using tiktoken token counting
• Query latency monitoring with OpenTelemetry instrumentation
• Automated re-indexing when source documents change
• Feedback collection loop for retrieval quality metrics

The complete code for this tutorial is available on GitHub. For more on optimizing retrieval pipelines, check out our guide on vector database benchmarking and production RAG architectures.

Remember that RAG systems degrade silently - your retrieval accuracy can drop from 90% to 60% without any error messages. Monitor your retrieval metrics in production and set up alerts for significant changes.