Building a Production-Ready RAG System with LangChain, PostgreSQL pgvector, and OpenAI Embeddings

Introduction

Retrieval-Augmented Generation (RAG) has become the de facto standard for grounding large language models in domain-specific, up-to-date, or proprietary knowledge. While the concept is straightforward — retrieve relevant context and feed it to an LLM — building a production-ready RAG system involves numerous engineering decisions that directly impact latency, accuracy, cost, and maintainability.

This guide provides a complete, end-to-end implementation of a production-grade RAG system using LangChain for orchestration, PostgreSQL with pgvector for vector storage and similarity search, and OpenAI embeddings for semantic representation. We'll cover architecture, data ingestion pipelines, chunking strategies, retrieval optimization, prompt engineering, evaluation, and deployment considerations.

By the end, you'll have a working codebase you can adapt for your own use cases — whether that's internal knowledge bases, customer support automation, legal document analysis, or any scenario where an LLM needs reliable access to specific information.

Table of Contents

• Introduction
• Core Concepts
• Architecture Overview
• Step-by-Step Guide
• Real-World Examples
• Production Code Examples
• Comparison Table: Vector Store Options
• Best Practices
• Common Mistakes
• Performance Tips
• Security Considerations
• Deployment Notes
• Debugging Tips
• FAQ
• Conclusion

Core Concepts

What Is RAG?

RAG combines two distinct capabilities: retrieval (finding relevant documents from a knowledge base) and generation (using an LLM to synthesize an answer grounded in that context). The retrieval step typically uses vector similarity search — converting both documents and queries into high-dimensional embeddings and finding the nearest neighbors.

Why PostgreSQL + pgvector?

PostgreSQL is battle-tested, ACID-compliant, and already ubiquitous in production stacks. The pgvector extension adds first-class vector similarity search (IVFFlat, HNSW indexes) without requiring a separate vector database. This reduces operational complexity, leverages existing PostgreSQL expertise, and supports hybrid queries (vector + metadata + full-text search) in a single transaction.

Embedding Models

OpenAI's text-embedding-3-small (1536 dimensions) and text-embedding-3-large (3072 dimensions) offer state-of-the-art semantic representations. The smaller model is 5x cheaper and often sufficient; the larger model improves recall on complex queries. Both are accessible via a simple API call.

LangChain's Role

LangChain provides abstractions for document loaders, text splitters, embedding integrations, vector store connectors, retrievers, and prompt templates. It standardizes the plumbing so you can focus on business logic rather than glue code.

Architecture Overview

The production RAG pipeline consists of two primary data flows:

1. Ingestion Pipeline (Offline / Batch)

1. Source Connectors pull raw documents (PDFs, HTML, Markdown, Notion, Confluence, etc.).
2. Text Splitters chunk documents into semantically coherent segments (typically 500-1500 tokens with overlap).
3. Embedding Service calls OpenAI API to generate vector representations for each chunk.
4. Vector Store Writer upserts chunks + embeddings + metadata into PostgreSQL/pgvector.
5. Index Maintenance rebuilds or updates HNSW/IVFFlat indexes periodically.

2. Query Pipeline (Online / Real-time)

1. Query Rewriting (optional) expands or decomposes the user query.
2. Embedding converts the query to a vector using the same model.
3. Vector Search retrieves top-K similar chunks from pgvector.
4. Re-ranking (optional) uses a cross-encoder or LLM to re-score candidates.
5. Context Assembly packs retrieved chunks into a prompt template with citations.
6. LLM Generation calls the chat model (e.g., GPT-4o) with the augmented prompt.
7. Post-processing validates citations, formats output, streams response.

Both pipelines share the same embedding model and vector store schema, ensuring consistency.

Step-by-Step Guide

Prerequisites

• Python 3.10+
• PostgreSQL 15+ with pgvector extension enabled
• OpenAI API key with embeddings and chat completions access
• Virtual environment tool (venv, poetry, or uv)

1. Database Setup

Enable the extension and create the schema:

-- Enable pgvectorCREATE EXTENSION IF NOT EXISTS vector;-- Documents tableCREATE TABLE documents (    id BIGSERIAL PRIMARY KEY,    source_id TEXT NOT NULL,    source_type TEXT NOT NULL,    title TEXT,    content TEXT NOT NULL,    metadata JSONB DEFAULT '{}',    created_at TIMESTAMPTZ DEFAULT NOW(),    updated_at TIMESTAMPTZ DEFAULT NOW());-- Chunks table with vector columnCREATE TABLE document_chunks (    id BIGSERIAL PRIMARY KEY,    document_id BIGINT REFERENCES documents(id) ON DELETE CASCADE,    chunk_index INT NOT NULL,    content TEXT NOT NULL,    embedding vector(1536), -- matches text-embedding-3-small    metadata JSONB DEFAULT '{}',    created_at TIMESTAMPTZ DEFAULT NOW());-- HNSW index for fast approximate nearest neighbor searchCREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops)WITH (m = 16, ef_construction = 64);-- Metadata filter indexesCREATE INDEX ON document_chunks ((metadata->>'category'));CREATE INDEX ON document_chunks ((metadata->>'author'));

2. Project Structure

rag-system/├── config/│   ├── __init__.py│   └── settings.py├── ingestion/│   ├── __init__.py│   ├── loaders.py│   ├── splitters.py│   └── pipeline.py├── retrieval/│   ├── __init__.py│   ├── vector_store.py│   ├── retriever.py│   └── reranker.py├── generation/│   ├── __init__.py│   ├── prompts.py│   └── chain.py├── evaluation/│   ├── __init__.py│   └── metrics.py├── api/│   ├── __init__.py│   ├── routes.py│   └── main.py├── tests/│   ├── test_ingestion.py│   ├── test_retrieval.py│   └── test_generation.py├── pyproject.toml└── README.md

3. Configuration Management

# config/settings.pyfrom pydantic_settings import BaseSettingsfrom functools import lru_cacheclass Settings(BaseSettings):    # Database    database_url: str = "postgresql+asyncpg://user:pass@localhost:5432/ragdb"        # OpenAI    openai_api_key: str    embedding_model: str = "text-embedding-3-small"    embedding_dimension: int = 1536    chat_model: str = "gpt-4o-mini"        # Chunking    chunk_size: int = 1000    chunk_overlap: int = 200        # Retrieval    top_k: int = 10    rerank_top_k: int = 5    similarity_threshold: float = 0.7        # Ingestion    batch_size: int = 100        class Config:        env_file = ".env"        env_file_encoding = "utf-8"@lru_cachedef get_settings() -> Settings:    return Settings()

4. Document Loaders

# ingestion/loaders.pyfrom pathlib import Pathfrom typing import List, Dict, Anyfrom langchain_core.documents import Documentfrom langchain_community.document_loaders import (    PyPDFLoader,    TextLoader,    UnstructuredMarkdownLoader,    UnstructuredHTMLLoader,    CSVLoader,)LOADER_MAP = {    ".pdf": PyPDFLoader,    ".txt": TextLoader,    ".md": UnstructuredMarkdownLoader,    ".html": UnstructuredHTMLLoader,    ".csv": CSVLoader,}def load_documents(source_path: str) -> List[Document]:    """Load documents from a file or directory."""    path = Path(source_path)    documents = []        if path.is_file():        files = [path]    else:        files = list(path.rglob("*"))        for file_path in files:        if file_path.suffix.lower() not in LOADER_MAP:            continue        loader_class = LOADER_MAP[file_path.suffix.lower()]        loader = loader_class(str(file_path))        docs = loader.load()        # Add source metadata        for i, doc in enumerate(docs):            doc.metadata.update({                "source_id": file_path.stem,                "source_type": file_path.suffix.lower()[1:],                "file_path": str(file_path),                "chunk_index": i,            })        documents.extend(docs)        return documents

5. Text Splitting Strategy

# ingestion/splitters.pyfrom langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain_core.documents import Documentfrom typing import Listfrom config.settings import get_settingsdef get_text_splitter() -> RecursiveCharacterTextSplitter:    settings = get_settings()    return RecursiveCharacterTextSplitter(        chunk_size=settings.chunk_size,        chunk_overlap=settings.chunk_overlap,        length_function=len,        separators=[            "",  # Paragraphs            "",     # Lines            ". ",     # Sentences            "? ",            "! ",            "; ",            ": ",            ", ",            " ",      # Words            "",       # Characters        ],        keep_separator=True,    )def split_documents(documents: List[Document]) -> List[Document]:    splitter = get_text_splitter()    return splitter.split_documents(documents)

6. Vector Store Integration

# retrieval/vector_store.pyfrom typing import List, Optional, Dict, Anyfrom langchain_core.documents import Documentfrom langchain_core.embeddings import Embeddingsfrom langchain_openai import OpenAIEmbeddingsfrom langchain_postgres import PGVectorfrom sqlalchemy.ext.asyncio import create_async_enginefrom config.settings import get_settingsclass VectorStore:    def __init__(self):        self.settings = get_settings()        self.embeddings: Embeddings = OpenAIEmbeddings(            model=self.settings.embedding_model,            api_key=self.settings.openai_api_key,        )        self._store: Optional[PGVector] = None        @property    def store(self) -> PGVector:        if self._store is None:            self._store = PGVector(                embeddings=self.embeddings,                collection_name="document_chunks",                connection=self.settings.database_url,                use_jsonb=True,                async_mode=True,            )        return self._store        async def add_documents(self, documents: List[Document]) -> List[str]:        """Add documents to the vector store."""        return await self.store.aadd_documents(documents)        async def similarity_search(        self,        query: str,        k: int = 10,        filter: Optional[Dict[str, Any]] = None,    ) -> List[Document]:        """Search for similar documents."""        return await self.store.asimilarity_search(            query, k=k, filter=filter        )        async def similarity_search_with_score(        self,        query: str,        k: int = 10,        filter: Optional[Dict[str, Any]] = None,    ) -> List[tuple[Document, float]]:        """Search with similarity scores."""        return await self.store.asimilarity_search_with_score(            query, k=k, filter=filter        )        async def delete_by_metadata(self, filter: Dict[str, Any]) -> None:        """Delete documents matching metadata filter."""        await self.store.adelete(filter=filter)

7. Advanced Retriever with Hybrid Search

# retrieval/retriever.pyfrom typing import List, Optional, Dict, Anyfrom langchain_core.documents import Documentfrom langchain_core.retrievers import BaseRetrieverfrom langchain_core.callbacks import CallbackManagerForRetrieverRunfrom retrieval.vector_store import VectorStorefrom retrieval.reranker import Rerankerfrom config.settings import get_settingsclass HybridRetriever(BaseRetriever):    """Combines vector similarity search with metadata filtering and optional re-ranking."""        vector_store: VectorStore    reranker: Optional[Reranker] = None        def __init__(self, **kwargs):        super().__init__(**kwargs)        self.settings = get_settings()        def _get_relevant_documents(        self,        query: str,        *,        run_manager: CallbackManagerForRetrieverRun,        filter: Optional[Dict[str, Any]] = None,        k: Optional[int] = None,    ) -> List[Document]:        # Vector search        k = k or self.settings.top_k        docs_with_scores = self.vector_store.similarity_search_with_score(            query, k=k, filter=filter        )                # Filter by similarity threshold        docs = [            doc for doc, score in docs_with_scores            if score >= self.settings.similarity_threshold        ]                # Re-rank if reranker available        if self.reranker and docs:            docs = self.reranker.rerank(query, docs, top_k=self.settings.rerank_top_k)                return docs        async def _aget_relevant_documents(        self,        query: str,        *,        run_manager: CallbackManagerForRetrieverRun,        filter: Optional[Dict[str, Any]] = None,        k: Optional[int] = None,    ) -> List[Document]:        k = k or self.settings.top_k        docs_with_scores = await self.vector_store.asimilarity_search_with_score(            query, k=k, filter=filter        )                docs = [            doc for doc, score in docs_with_scores            if score >= self.settings.similarity_threshold        ]                if self.reranker and docs:            docs = await self.reranker.arerank(query, docs, top_k=self.settings.rerank_top_k)                return docs

8. Cross-Encoder Re-ranker

# retrieval/reranker.pyfrom typing import List, Optionalfrom langchain_core.documents import Documentfrom sentence_transformers import CrossEncoderimport torchfrom config.settings import get_settingsclass Reranker:    """Cross-encoder re-ranker for improved retrieval precision."""        def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):        self.settings = get_settings()        self.device = "cuda" if torch.cuda.is_available() else "cpu"        self.model = CrossEncoder(model_name, device=self.device)        def rerank(self, query: str, documents: List[Document], top_k: int = 5) -> List[Document]:        if not documents:            return []                pairs = [[query, doc.page_content] for doc in documents]        scores = self.model.predict(pairs)                # Combine scores with original order        scored_docs = list(zip(documents, scores))        scored_docs.sort(key=lambda x: x[1], reverse=True)                return [doc for doc, _ in scored_docs[:top_k]]        async def arerank(self, query: str, documents: List[Document], top_k: int = 5) -> List[Document]:        # Run in thread pool for async compatibility        import asyncio        loop = asyncio.get_event_loop()        return await loop.run_in_executor(None, self.rerank, query, documents, top_k)

9. Prompt Engineering for Grounded Generation

# generation/prompts.pyfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_core.messages import SystemMessage, HumanMessageRAG_SYSTEM_PROMPT = """You are a precise, citation-focused assistant. Answer the user's question using ONLY the provided context.Guidelines:1. If the context doesn't contain the answer, say "I cannot answer based on the provided documents."2. Cite sources using [doc_id] notation at the end of each sentence.3. Be concise but complete.4. Distinguish between facts from the documents and your own reasoning.5. If multiple documents contradict, mention the discrepancy."""RAG_HUMAN_TEMPLATE = """Context documents:{context}Question: {question}Answer with citations:"""def format_context(documents: list) -> str:    """Format retrieved documents for prompt injection."""    formatted = []    for i, doc in enumerate(documents):        source = doc.metadata.get("source_id", f"doc_{i}")        content = doc.page_content.strip()        formatted.append(f"[doc_{i}] Source: {source}{content}")    return "---".join(formatted)def get_rag_prompt() -> ChatPromptTemplate:    return ChatPromptTemplate.from_messages([        SystemMessage(content=RAG_SYSTEM_PROMPT),        HumanMessage(content=RAG_HUMAN_TEMPLATE),    ])

10. Generation Chain with Streaming

# generation/chain.pyfrom typing import AsyncIterator, List, Dict, Anyfrom langchain_core.documents import Documentfrom langchain_core.runnables import RunnablePassthrough, RunnableLambdafrom langchain_openai import ChatOpenAIfrom langchain_core.output_parsers import StrOutputParserfrom generation.prompts import get_rag_prompt, format_contextfrom retrieval.retriever import HybridRetrieverfrom config.settings import get_settingsclass RAGChain:    def __init__(self):        self.settings = get_settings()        self.retriever = HybridRetriever()        self.llm = ChatOpenAI(            model=self.settings.chat_model,            api_key=self.settings.openai_api_key,            temperature=0.1,            streaming=True,        )        self.prompt = get_rag_prompt()        self.chain = (            {                "context": RunnableLambda(self._retrieve_and_format),                "question": RunnablePassthrough(),            }            | self.prompt            | self.llm            | StrOutputParser()        )        def _retrieve_and_format(self, question: str) -> str:        docs = self.retriever.invoke(question)        return format_context(docs)        async def _aretrieve_and_format(self, question: str) -> str:        docs = await self.retriever.ainvoke(question)        return format_context(docs)        async def astream(self, question: str) -> AsyncIterator[str]:        """Stream the response token by token."""        async for chunk in self.chain.astream(question):            yield chunk        async def ainvoke(self, question: str) -> Dict[str, Any]:        """Invoke and return full response with metadata."""        docs = await self.retriever.ainvoke(question)        context = format_context(docs)        response = await self.chain.ainvoke({"context": context, "question": question})        return {            "answer": response,            "sources": [                {                    "content": doc.page_content[:200] + "...",                    "metadata": doc.metadata,                }                for doc in docs            ],        }

11. Ingestion Pipeline Orchestration

# ingestion/pipeline.pyfrom typing import List, Dict, Anyfrom langchain_core.documents import Documentfrom ingestion.loaders import load_documentsfrom ingestion.splitters import split_documentsfrom retrieval.vector_store import VectorStorefrom config.settings import get_settingsimport asyncioclass IngestionPipeline:    def __init__(self):        self.settings = get_settings()        self.vector_store = VectorStore()        async def run(self, source_path: str) -> Dict[str, Any]:        """Execute the full ingestion pipeline."""        # 1. Load        documents = load_documents(source_path)        print(f"Loaded {len(documents)} raw documents")                # 2. Split        chunks = split_documents(documents)        print(f"Split into {len(chunks)} chunks")                # 3. Batch embed and store        batch_size = self.settings.batch_size        total_stored = 0                for i in range(0, len(chunks), batch_size):            batch = chunks[i:i + batch_size]            ids = await self.vector_store.add_documents(batch)            total_stored += len(ids)            print(f"Stored batch {i//batch_size + 1}: {len(ids)} chunks")                return {            "source_path": source_path,            "documents_loaded": len(documents),            "chunks_created": len(chunks),            "chunks_stored": total_stored,        }        async def update_document(self, source_id: str, source_path: str) -> Dict[str, Any]:        """Update an existing document (delete old, ingest new)."""        # Delete existing chunks for this source        await self.vector_store.delete_by_metadata({"source_id": source_id})        # Re-ingest        return await self.run(source_path)

12. FastAPI Server

# api/routes.pyfrom fastapi import APIRouter, HTTPException, Queryfrom pydantic import BaseModel, Fieldfrom typing import Optional, List, Dict, Anyfrom generation.chain import RAGChainfrom ingestion.pipeline import IngestionPipelinerouter = APIRouter()rag_chain = RAGChain()ingestion_pipeline = IngestionPipeline()class QueryRequest(BaseModel):    question: str = Field(..., min_length=1, max_length=2000)    filter: Optional[Dict[str, Any]] = None    top_k: Optional[int] = Field(None, ge=1, le=50)class QueryResponse(BaseModel):    answer: str    sources: List[Dict[str, Any]]class IngestRequest(BaseModel):    source_path: strclass IngestResponse(BaseModel):    documents_loaded: int    chunks_created: int    chunks_stored: int@router.post("/query", response_model=QueryResponse)async def query_rag(request: QueryRequest):    """Query the RAG system."""    try:        result = await rag_chain.ainvoke(request.question)        return result    except Exception as e:        raise HTTPException(status_code=500, detail=str(e))@router.post("/query/stream")async def query_rag_stream(request: QueryRequest):    """Stream the RAG response."""    from fastapi.responses import StreamingResponse        async def generate():        async for chunk in rag_chain.astream(request.question):            yield f"data: {chunk}"        yield "data: [DONE]"        return StreamingResponse(generate(), media_type="text/event-stream")@router.post("/ingest", response_model=IngestResponse)async def ingest_documents(request: IngestRequest):    """Ingest documents from a path."""    try:        result = await ingestion_pipeline.run(request.source_path)        return result    except Exception as e:        raise HTTPException(status_code=500, detail=str(e))@router.post("/ingest/update", response_model=IngestResponse)async def update_document(source_id: str = Query(...), source_path: str = Query(...)):    """Update a specific document."""    try:        result = await ingestion_pipeline.update_document(source_id, source_path)        return result    except Exception as e:        raise HTTPException(status_code=500, detail=str(e))

# api/main.pyfrom fastapi import FastAPIfrom api.routes import routerfrom contextlib import asynccontextmanager@asynccontextmanagerasync def lifespan(app: FastAPI):    # Startup    print("RAG API starting up...")    yield    # Shutdown    print("RAG API shutting down...")app = FastAPI(    title="Production RAG API",    description="RAG system with LangChain, pgvector, and OpenAI",    version="1.0.0",    lifespan=lifespan,)app.include_router(router, prefix="/api/v1")@app.get("/health")async def health_check():    return {"status": "healthy"}

Real-World Examples

Example 1: Internal Knowledge Base for Engineering Team

A 50-person engineering team maintains 2,000+ markdown files in a Git repository (architecture decisions, API specs, runbooks, postmortems). The RAG system indexes the repo on every push via CI/CD, enabling developers to ask questions like "What's the retry policy for the payments service?" or "How do we rotate database credentials?" and receive cited answers instantly.

Example 2: Customer Support Automation

A SaaS company ingests 10,000+ historical support tickets, product documentation, and release notes. The RAG system handles 40% of Tier-1 queries automatically, escalating only when confidence is low or the user requests a human. Average response time drops from 15 minutes to 3 seconds.

Example 3: Legal Contract Analysis

A law firm uploads 500+ contracts (PDFs). Lawyers query "What are the termination clauses in contracts with Company X?" The system retrieves relevant sections across all contracts, cites exact pages, and summarizes differences — reducing review time from hours to minutes.

Production Code Examples

Async Batch Embedding with Rate Limiting

# utils/embedding_batch.pyimport asynciofrom typing import Listfrom langchain_openai import OpenAIEmbeddingsfrom tenacity import retry, stop_after_attempt, wait_exponential_jitterfrom config.settings import get_settingsclass BatchedEmbeddings:    """OpenAI embeddings with async batching and rate limiting."""        def __init__(self, batch_size: int = 100, max_concurrent: int = 5):        self.settings = get_settings()        self.embeddings = OpenAIEmbeddings(            model=self.settings.embedding_model,            api_key=self.settings.openai_api_key,        )        self.batch_size = batch_size        self.semaphore = asyncio.Semaphore(max_concurrent)        @retry(        wait=wait_exponential_jitter(initial=1, max=60),        stop=stop_after_attempt(3),    )    async def _embed_batch(self, texts: List[str]) -> List[List[float]]:        async with self.semaphore:            return await self.embeddings.aembed_documents(texts)        async def embed_documents(self, texts: List[str]) -> List[List[float]]:        """Embed documents in batches with controlled concurrency."""        all_embeddings = []        for i in range(0, len(texts), self.batch_size):            batch = texts[i:i + self.batch_size]            embeddings = await self._embed_batch(batch)            all_embeddings.extend(embeddings)        return all_embeddings        async def embed_query(self, text: str) -> List[float]:        async with self.semaphore:            return await self.embeddings.aembed_query(text)

Hybrid Search with Full-Text + Vector

# retrieval/hybrid_search.pyfrom typing import List, Optional, Dict, Anyfrom langchain_core.documents import Documentfrom sqlalchemy.ext.asyncio import AsyncSessionfrom sqlalchemy import text, selectfrom sqlalchemy.orm import selectinloadfrom database.models import DocumentChunkfrom config.settings import get_settingsclass HybridSearch:    """Combines pgvector similarity search with PostgreSQL full-text search."""        def __init__(self, session: AsyncSession):        self.session = session        self.settings = get_settings()        async def search(        self,        query: str,        vector: List[float],        k: int = 10,        filter: Optional[Dict[str, Any]] = None,        full_text_weight: float = 0.3,        vector_weight: float = 0.7,    ) -> List[Document]:        """Reciprocal Rank Fusion (RRF) of vector and full-text results."""        # Build filter conditions        filter_clauses = []        params = {            "query_vector": vector,            "query_text": query,            "k": k * 2,  # Fetch more for fusion            "vector_weight": vector_weight,            "full_text_weight": full_text_weight,        }                if filter:            for key, value in filter.items():                filter_clauses.append(f"metadata->>'{key}' = :{key}")                params[key] = value                filter_sql = " AND ".join(filter_clauses) if filter_clauses else "1=1"                # RRF query: combines vector and full-text ranks        sql = f"""        WITH vector_results AS (            SELECT id, content, metadata,                   ROW_NUMBER() OVER (ORDER BY embedding <=> :query_vector) as vector_rank            FROM document_chunks            WHERE {filter_sql}            LIMIT :k        ),        fts_results AS (            SELECT id, content, metadata,                   ROW_NUMBER() OVER (ORDER BY ts_rank_cd(to_tsvector('english', content), plainto_tsquery('english', :query_text)) DESC) as fts_rank            FROM document_chunks            WHERE {filter_sql}              AND to_tsvector('english', content) @@ plainto_tsquery('english', :query_text)            LIMIT :k        ),        fused AS (            SELECT id, content, metadata,                   COALESCE(:vector_weight / (60 + vector_rank), 0) +                   COALESCE(:full_text_weight / (60 + fts_rank), 0) as fused_score            FROM vector_results            FULL OUTER JOIN fts_results USING (id, content, metadata)        )        SELECT id, content, metadata        FROM fused        ORDER BY fused_score DESC        LIMIT :k        """                result = await self.session.execute(text(sql), params)        rows = result.fetchall()                return [            Document(page_content=row.content, metadata=row.metadata or {})            for row in rows        ]

Evaluation Harness

# evaluation/metrics.pyfrom typing import List, Dict, Anyfrom dataclasses import dataclassfrom langchain_core.documents import Documentimport numpy as np@dataclassclass RetrievalMetrics:    recall_at_k: Dict[int, float]    mrr: float    ndcg_at_k: Dict[int, float]    latency_ms: float@dataclassclass GenerationMetrics:    faithfulness: float  # Does answer match retrieved context?    relevance: float     # Does answer address the question?    citation_accuracy: float  # Are citations correct?    latency_ms: floatclass RAGEvaluator:    def __init__(self, retriever, generator):        self.retriever = retriever        self.generator = generator        def evaluate_retrieval(        self,        test_cases: List[Dict[str, Any]],        k_values: List[int] = [1, 3, 5, 10]    ) -> RetrievalMetrics:        """Evaluate retrieval against ground truth."""        # test_cases: [{question, relevant_doc_ids}]        all_recall = {k: [] for k in k_values}        reciprocal_ranks = []        ndcg_scores = {k: [] for k in k_values}        latencies = []                for case in test_cases:            import time            start = time.perf_counter()            retrieved = self.retriever.invoke(case["question"])            latencies.append((time.perf_counter() - start) * 1000)                        retrieved_ids = [doc.metadata.get("doc_id") for doc in retrieved]            relevant = set(case["relevant_doc_ids"])                        # Recall@K            for k in k_values:                top_k = set(retrieved_ids[:k])                recall = len(top_k & relevant) / len(relevant) if relevant else 1.0                all_recall[k].append(recall)                        # MRR            for rank, doc_id in enumerate(retrieved_ids, 1):                if doc_id in relevant:                    reciprocal_ranks.append(1.0 / rank)                    break            else:                reciprocal_ranks.append(0.0)                        # NDCG@K (simplified binary relevance)            for k in k_values:                dcg = sum(                    1.0 / np.log2(i + 2)                    for i, doc_id in enumerate(retrieved_ids[:k])                    if doc_id in relevant                )                ideal_dcg = sum(1.0 / np.log2(i + 2) for i in range(min(len(relevant), k)))                ndcg = dcg / ideal_dcg if ideal_dcg > 0 else 1.0                ndcg_scores[k].append(ndcg)                return RetrievalMetrics(            recall_at_k={k: np.mean(v) for k, v in all_recall.items()},            mrr=np.mean(reciprocal_ranks),            ndcg_at_k={k: np.mean(v) for k, v in ndcg_scores.items()},            latency_ms=np.mean(latencies),        )        async def evaluate_generation(        self,        test_cases: List[Dict[str, Any]]    ) -> GenerationMetrics:        """Evaluate generation quality using LLM-as-judge."""        # Implementation would use a separate LLM to score faithfulness, relevance, etc.        # Placeholder for brevity        pass

Comparison Table: Vector Store Options

Best Practices

1. Chunk semantically, not just by size. Use structure-aware splitters (Markdown headers, HTML tags, PDF sections) to preserve context boundaries.
2. Store rich metadata. Include source, author, date, category, version, and hierarchical paths. This enables powerful filtering and citation.
3. Use the same embedding model for ingestion and query. Mismatched models destroy retrieval quality.
4. Implement incremental updates. Track file hashes (SHA256) to only re-embed changed documents.
5. Monitor retrieval metrics continuously. Log recall@K, MRR, and latency per query. Set alerts for degradation.
6. Version your prompts and models. Treat prompt templates as code — version, test, and rollback.
7. Separate ingestion and query compute. Ingestion is batch/async; query is latency-sensitive. Scale independently.
8. Use async throughout. Async I/O for embeddings, database, and LLM calls maximizes throughput.
9. Implement graceful degradation. If vector search fails, fall back to keyword search. If LLM fails, return retrieved snippets.

Common Mistakes

1. Chunking too small (< 200 tokens). Loses context, increases noise, bloats index. Aim for 500-1500 tokens with 10-20% overlap.
2. Ignoring metadata filtering. Searching the entire corpus for every query kills latency and precision. Filter by category, date, author first.
3. Using cosine similarity without normalization. pgvector's vector_cosine_ops expects normalized vectors. OpenAI embeddings are normalized; custom models may not be.
4. No re-ranking. Raw vector search retrieves semantically similar but topically irrelevant chunks. A cross-encoder re-ranker typically improves precision@5 by 15-30%.
5. Stuffing all retrieved chunks into context. Exceeds context window, dilutes attention, increases cost. Use re-ranking + top-K (3-5) + summarization if needed.
6. Hardcoding prompts. Prompts need iteration. A/B test prompt variants; track which templates yield higher citation accuracy.
7. Skipping evaluation. Without ground-truth test sets, you cannot measure regression when changing models, chunk sizes, or prompts.
8. Embedding PII or secrets. Never send sensitive data to embedding APIs. Redact or hash before ingestion.
9. Single index for all use cases. Different query types (fact lookup vs. summarization vs. comparison) benefit from different chunk sizes and retrieval strategies.

Performance Tips

Database Level

• Use HNSW indexes for production (m=16, ef_construction=64); tune ef_search at query time (40-200).
• Partition large tables by time (monthly) or tenant for faster index scans and easier maintenance.
• Enable parallel_leader_participation and tune max_parallel_workers_per_gather for large scans.
• Use pg_stat_statements to identify slow queries; add partial indexes for common filters.
• Consider pgvector's halfvec (float16) or bit quantization for 2-4x storage savings with minimal recall loss.

Embedding Level

• Batch embeddings (100-500 texts per request) to amortize API overhead.
• Cache embeddings for repeated queries (e.g., common questions) using Redis.
• Use text-embedding-3-small for most cases; upgrade to -large only if evaluation proves it's worth the 5x cost.
• Implement exponential backoff with jitter for OpenAI rate limits (429 errors).

Retrieval Level

• Pre-filter with metadata before vector search using PostgreSQL's partial indexes.
• Use LIMIT in the vector search query; don't fetch 1000 rows to re-rank 5.
• Async re-ranking: run cross-encoder in a thread pool while streaming LLM tokens.
• Implement query rewriting (HyDE, step-back prompting) for complex questions.

Generation Level

• Stream tokens to reduce perceived latency (first token in < 500ms).
• Use smaller chat models (GPT-4o-mini, Claude Haiku) for simple QA; route complex reasoning to larger models.
• Cache frequent answers with semantic similarity matching (e.g., Redis + embedding).

Security Considerations

1. Data Privacy: OpenAI API does not train on your data (per policy), but data leaves your network. For strict compliance, use self-hosted embeddings (e.g., BGE, E5) and local LLMs (Ollama, vLLM).
2. Injection Attacks: User queries go directly into vector search and LLM prompts. Sanitize inputs; use parameterized queries for metadata filters; limit prompt template interpolation.
3. Access Control: Implement row-level security (RLS) in PostgreSQL so users only retrieve documents they're authorized to see.
4. Audit Logging: Log every query, retrieved document IDs, and generated answer (hash) for compliance and debugging.
5. Rate Limiting: Protect embedding and generation endpoints with per-user/per-IP limits to prevent abuse and cost runaway.
6. Secrets Management: Store API keys in Vault, AWS Secrets Manager, or Kubernetes secrets — never in code or .env files in production.

Deployment Notes

Containerization

# DockerfileFROM python:3.11-slimWORKDIR /app# Install system dependencies for pgvector, pdf parsingRUN apt-get update && apt-get install -y \    gcc \    libpq-dev \    poppler-utils \    tesseract-ocr \    && rm -rf /var/lib/apt/lists/*# Install Python dependenciesCOPY pyproject.toml poetry.lock* ./ RUN pip install --no-cache-dir poetry && \    poetry config virtualenvs.create false && \    poetry install --only=mainCOPY . .EXPOSE 8000CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Kubernetes Deployment

# k8s/deployment.yamlapiVersion: apps/v1kind: Deploymentmetadata:  name: rag-apispec:  replicas: 3  selector:    matchLabels:      app: rag-api  template:    metadata:      labels:        app: rag-api    spec:      containers:      - name: api        image: your-registry/rag-api:latest        ports:        - containerPort: 8000        envFrom:        - secretRef:            name: rag-secrets        resources:          requests:            memory: "1Gi"            cpu: "500m"          limits:            memory: "2Gi"            cpu: "1000m"        livenessProbe:          httpGet:            path: /health            port: 8000          initialDelaySeconds: 30          periodSeconds: 10        readinessProbe:          httpGet:            path: /health            port: 8000          initialDelaySeconds: 5          periodSeconds: 5---apiVersion: v1kind: Servicemetadata:  name: rag-apispec:  selector:    app: rag-api  ports:  - port: 80    targetPort: 8000  type: ClusterIP

Database Migration Strategy

• Use Alembic for schema migrations.
• For pgvector index changes (e.g., IVFFlat → HNSW), create new index concurrently, then drop old.
• Backfill embeddings in batches during low-traffic windows.

Debugging Tips

1. Log the exact query vector. Verify it's normalized and matches the embedding model's output dimension.
2. Inspect retrieved chunks. Add a debug endpoint that returns raw chunks with scores before re-ranking.
3. Visualize embedding space. Use UMAP/t-SNE on a sample of embeddings to check cluster separation.
4. Test with known answers. Maintain a "golden set" of 50+ question-answer pairs for regression testing.
5. Profile database queries. Use EXPLAIN ANALYZE on the vector search query; verify index usage.
6. Monitor OpenAI latency. Track p50/p95/p99 for embedding and chat completions separately.
7. Check for silent failures. Empty retrieval results, truncated context, or malformed citations often produce plausible but wrong answers.

FAQ

Q: Why choose pgvector over a dedicated vector database like Pinecone?

A: pgvector eliminates a separate infrastructure component, supports ACID transactions, enables hybrid SQL + vector queries, and leverages existing PostgreSQL operational expertise. For most teams already running Postgres, it's the pragmatic choice until scale or specialized features (e.g., multi-tenancy with strict isolation) demand a dedicated solution.

Q: What chunk size should I use for my documents?

A: Start with 1000 tokens and 200 token overlap for general text. For code, use smaller chunks (300-500 tokens) with structure-aware splitting (functions, classes). For legal/regulatory text, larger chunks (1500-2000) preserve clause context. Always evaluate with your specific corpus.

Q: How do I handle documents that change frequently?

A: Compute a content hash (SHA256) for each document at ingestion. Store the hash in metadata. On re-ingestion, compare hashes; only re-process and re-embed changed documents. Delete old chunks by source_id before inserting new ones.

Q: Can I use a local embedding model instead of OpenAI?

A: Yes. Replace OpenAIEmbeddings with HuggingFaceEmbeddings (e.g., BAAI/bge-base-en-v1.5 or intfloat/e5-base-v2). Ensure the vector dimension in your schema matches the model (768 for base models). Self-hosted embeddings eliminate API costs and data egress but require GPU inference infrastructure.

Q: How do I evaluate RAG quality without human labels?

A: Use synthetic evaluation: generate questions from your documents using an LLM, then measure retrieval recall against the source documents. For generation, use "LLM-as-judge" with a rubric (faithfulness, relevance, citation accuracy). Tools like RAGAS or LangSmith automate this.

Q: What's the cost of running this in production?

A> Rough estimate for 1M documents (avg 10 chunks each = 10M embeddings): OpenAI embedding cost ~$200 (text-embedding-3-small at $0.02/1M tokens). Chat completion cost depends on query volume — 10K queries/day with 2K context tokens each ≈ $600/month (GPT-4o-mini). PostgreSQL hosting: $100-500/month depending on instance size. Total: ~$1K-2K/month for moderate scale.

Q: How do I handle multi-language documents?

A: Use a multilingual embedding model (e.g., text-embedding-3-large supports 100+ languages, or intfloat/multilingual-e5-large for self-hosted). Store language metadata per chunk. At query time, detect query language and filter by language metadata, or rely on the embedding model's cross-lingual alignment.

Q: Can this architecture support multi-tenancy?

A: Yes. Add a tenant_id column to document_chunks and create a partial HNSW index per tenant (or use row-level security). Filter every query by tenant_id. For strict isolation, use separate schemas or databases per tenant.

Conclusion

Building a production RAG system is as much about engineering discipline as it is about AI. The stack we've covered — LangChain for orchestration, PostgreSQL/pgvector for storage, OpenAI for embeddings and generation — provides a robust, scalable foundation that you can iterate on confidently.

The key takeaways:

• Invest in evaluation early. You cannot improve what you don't measure.
• Chunk intelligently. Semantic boundaries beat fixed-size windows.
• Re-rank aggressively. Cross-encoders are the highest-ROI addition to any retrieval pipeline.
• Monitor everything. Latency, recall, costs, error rates — treat them as first-class metrics.
• Design for change. Models, prompts, and requirements will evolve. Modular, tested code lets you swap components without rewrites.

Ready to deploy your own RAG system? Clone the reference implementation, adapt the chunking and prompts to your domain, and start measuring. The gap between a demo and a production system is bridged by iteration — and you now have the blueprint to iterate fast.

Next steps: Set up a CI/CD pipeline that re-indexes your knowledge base on every merge, add a feedback loop (thumbs up/down on answers) to collect training data, and schedule a monthly evaluation review. Your future self will thank you.

Have questions about scaling, custom embeddings, or hybrid search strategies? Drop a comment below or reach out on Twitter — I'm always happy to discuss RAG architecture.