How to Build a Production RAG Pipeline in Python with LangChain and pgvector

When building enterprise LLM applications, foundational models like GPT-4o or Claude 3.5 Sonnet suffer from two fundamental bottlenecks: strict knowledge cutoff dates and zero visibility into your organization's private databases, PDFs, and proprietary documentation. While naive teams immediately resort to expensive model fine-tuning, retrieval-augmented generation (RAG) is the industry-standard architecture for grounding generative responses in factual, private corporate context without model weight modification.

Rather than paying exorbitant monthly fees for standalone hosted vector databases like Pinecone or Weaviate, high-throughput engineering teams leverage PostgreSQL with the pgvector extension. By consolidating relational entities, user permissions, transactional records, and high-dimensional semantic embeddings inside a single ACID-compliant database cluster, you eliminate distributed data sync headaches and network hops.

In this comprehensive engineering walkthrough, we will construct an enterprise-grade RAG pipeline using Python 3.11+, LangChain, HuggingFace/OpenAI embeddings, and self-hosted PostgreSQL with pgvector. We will cover document chunking strategies, HNSW vs. IVFFlat indexing benchmarks, hybrid keyword-vector search, and conversational memory buffering.

1. Architectural Blueprint: The 4 Stages of Production RAG

A resilient RAG architecture operates across four synchronized stages:

1. Ingestion & Recursive Chunking: Parsing unstructured raw text (Markdown, DOCX, PDF) and dividing it into overlapping semantic blocks to preserve context across chunk boundaries.
2. Vector Embedding Generation: Passing chunks through an embedding model (e.g., text-embedding-3-small or bge-large-en-v1.5) to generate dense mathematical representations (1536 or 1024 float dimensions).
3. Similarity Indexing in pgvector: Storing vectors inside PostgreSQL with an index optimized for cosine similarity or Euclidean distance.
4. Context Injection & Synthesis: Querying the vector store with user prompt embeddings, extracting the Top-K most relevant document chunks, and dynamically constructing a fortified prompt for the LLM.

2. Provisioning PostgreSQL with pgvector

The cleanest approach for running pgvector locally or in Kubernetes is the official Docker container. Run this command to provision PostgreSQL 16 with pgvector pre-compiled:

# Launch PostgreSQL 16 with pgvector extension enabled
docker run -d \
  --name postgres-vector \
  -e POSTGRES_USER=rag_admin \
  -e POSTGRES_PASSWORD=SuperSecurePass2026 \
  -e POSTGRES_DB=knowledge_base \
  -p 5432:5432 \
  pgvector/pgvector:pg16

Connect via psql and activate the extension:

-- Connect to database
\c knowledge_base

-- Enable extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Verify installation
SELECT extname, extversion FROM pg_extension WHERE extname = 'vector';

3. Setting Up the Python Environment

Install the necessary Python packages using pip:

pip install langchain langchain-community langchain-openai psycopg2-binary pgvector pydantic tiktoken

4. Complete Production RAG Pipeline Implementation

Below is the modular, object-oriented RAG controller that handles ingestion, semantic search, and streaming LLM generation:

import os
from typing import List
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores.pgvector import PGVector
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

class EnterpriseRagService:
    def __init__(self, connection_string: str, collection_name: str = "corporate_docs"):
        self.connection_string = connection_string
        self.collection_name = collection_name
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

        # Initialize PGVector store
        self.vector_store = PGVector(
            connection_string=self.connection_string,
            embedding_function=self.embeddings,
            collection_name=self.collection_name,
            use_jsonb=True
        )

    def ingest_documents(self, directory_path: str):
        print(f"[INFO] Ingesting documents from {directory_path}...")
        loader = DirectoryLoader(directory_path, glob="**/*.txt", loader_cls=TextLoader)
        raw_documents = loader.load()

        # Recursive text splitting with semantic punctuation boundaries
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=800,
            chunk_overlap=120,
            separators=["\n\n", "\n", ". ", " ", ""]
        )
        split_docs = text_splitter.split_documents(raw_documents)
        print(f"[INFO] Generated {len(split_docs)} chunks. Storing in pgvector...")

        self.vector_store.add_documents(split_docs)
        print("[SUCCESS] All documents embedded and indexed.")

    def query_rag(self, user_question: str) -> str:
        # Cosine similarity retrieval with Top-4 chunks
        retriever = self.vector_store.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 4}
        )

        system_prompt = (
            "You are an expert technical documentation assistant. Use the following pieces of "
            "retrieved context to answer the question. If the answer is not present in the context, "
            "explicitly state that you do not know. Do not hallucinate or make assumptions.\n\n"
            "Context:\n{context}"
        )
        prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            ("human", "{input}"),
        ])

        question_answer_chain = create_stuff_documents_chain(self.llm, prompt)
        rag_chain = create_retrieval_chain(retriever, question_answer_chain)

        response = rag_chain.invoke({"input": user_question})
        return response["answer"]

# Example Usage
if __name__ == "__main__":
    DB_URL = "postgresql+psycopg2://rag_admin:SuperSecurePass2026@localhost:5432/knowledge_base"
    rag = EnterpriseRagService(connection_string=DB_URL)
    
    # Ingest internal SOP documents once
    # rag.ingest_documents("./company_sop/")

    answer = rag.query_rag("What is the disaster recovery protocol for database failures?")
    print("\n[LLM RESPONSE]:\n", answer)

5. Vector Indexing Benchmarks: HNSW vs. IVFFlat

In pgvector, how you index your embedding column dictates query latency under million-vector loads:

-- High-Performance PostgreSQL Hybrid Search Query with RRF
WITH semantic_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY embedding <=> \$1) as rank
    FROM langchain_pg_embedding
    ORDER BY embedding <=> \$1
    LIMIT 20
),
keyword_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('english', content), plainto_tsquery('english', \$2)) DESC) as rank
    FROM langchain_pg_embedding
    WHERE to_tsvector('english', content) @@ plainto_tsquery('english', \$2)
    LIMIT 20
)
SELECT 
    COALESCE(s.id, k.id) as id,
    COALESCE(s.content, k.content) as content,
    COALESCE(1.0 / (60 + s.rank), 0.0) + COALESCE(1.0 / (60 + k.rank), 0.0) as rrf_score
FROM semantic_search s
FULL OUTER JOIN keyword_search k ON s.id = k.id
ORDER BY rrf_score DESC
LIMIT 5;

-- High-Performance PostgreSQL Hybrid Search Query with RRF
WITH semantic_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY embedding <=> \$1) as rank
    FROM langchain_pg_embedding
    ORDER BY embedding <=> \$1
    LIMIT 20
),
keyword_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('english', content), plainto_tsquery('english', \$2)) DESC) as rank
    FROM langchain_pg_embedding
    WHERE to_tsvector('english', content) @@ plainto_tsquery('english', \$2)
    LIMIT 20
)
SELECT 
    COALESCE(s.id, k.id) as id,
    COALESCE(s.content, k.content) as content,
    COALESCE(1.0 / (60 + s.rank), 0.0) + COALESCE(1.0 / (60 + k.rank), 0.0) as rrf_score
FROM semantic_search s
FULL OUTER JOIN keyword_search k ON s.id = k.id
ORDER BY rrf_score DESC
LIMIT 5;

-- High-Performance PostgreSQL Hybrid Search Query with RRF
WITH semantic_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY embedding <=> \$1) as rank
    FROM langchain_pg_embedding
    ORDER BY embedding <=> \$1
    LIMIT 20
),
keyword_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('english', content), plainto_tsquery('english', \$2)) DESC) as rank
    FROM langchain_pg_embedding
    WHERE to_tsvector('english', content) @@ plainto_tsquery('english', \$2)
    LIMIT 20
)
SELECT 
    COALESCE(s.id, k.id) as id,
    COALESCE(s.content, k.content) as content,
    COALESCE(1.0 / (60 + s.rank), 0.0) + COALESCE(1.0 / (60 + k.rank), 0.0) as rrf_score
FROM semantic_search s
FULL OUTER JOIN keyword_search k ON s.id = k.id
ORDER BY rrf_score DESC
LIMIT 5;

-- Creating a high-performance HNSW index in PostgreSQL
CREATE INDEX ON langchain_pg_embedding 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

6. Production Best Practices for Zero Hallucinations

-- High-Performance PostgreSQL Hybrid Search Query with RRF
WITH semantic_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY embedding <=> \$1) as rank
    FROM langchain_pg_embedding
    ORDER BY embedding <=> \$1
    LIMIT 20
),
keyword_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('english', content), plainto_tsquery('english', \$2)) DESC) as rank
    FROM langchain_pg_embedding
    WHERE to_tsvector('english', content) @@ plainto_tsquery('english', \$2)
    LIMIT 20
)
SELECT 
    COALESCE(s.id, k.id) as id,
    COALESCE(s.content, k.content) as content,
    COALESCE(1.0 / (60 + s.rank), 0.0) + COALESCE(1.0 / (60 + k.rank), 0.0) as rrf_score
FROM semantic_search s
FULL OUTER JOIN keyword_search k ON s.id = k.id
ORDER BY rrf_score DESC
LIMIT 5;

7. Hybrid Search: Merging Full-Text Search (BM25) with Vector Embeddings

Pure semantic vector search excels at conceptual matching but frequently fails when users search for exact technical serial numbers, SKU identifiers, or specific function names (e.g., ERR_CONNECTION_RESET). A user searching for a precise error code might get a vector match on a generic troubleshooting paragraph rather than the exact documentation page.

The gold standard enterprise solution is Hybrid Search, which executes a classical PostgreSQL full-text search (BM25 equivalent using to_tsvector and GIN indexing) alongside the pgvector cosine distance query, then merges the results using Reciprocal Rank Fusion (RRF):

-- High-Performance PostgreSQL Hybrid Search Query with RRF
WITH semantic_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY embedding <=> \$1) as rank
    FROM langchain_pg_embedding
    ORDER BY embedding <=> \$1
    LIMIT 20
),
keyword_search AS (
    SELECT id, content, ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('english', content), plainto_tsquery('english', \$2)) DESC) as rank
    FROM langchain_pg_embedding
    WHERE to_tsvector('english', content) @@ plainto_tsquery('english', \$2)
    LIMIT 20
)
SELECT 
    COALESCE(s.id, k.id) as id,
    COALESCE(s.content, k.content) as content,
    COALESCE(1.0 / (60 + s.rank), 0.0) + COALESCE(1.0 / (60 + k.rank), 0.0) as rrf_score
FROM semantic_search s
FULL OUTER JOIN keyword_search k ON s.id = k.id
ORDER BY rrf_score DESC
LIMIT 5;

8. Automated Evaluation Pipeline with RAGAS

To ensure ongoing quality and detect regression before promoting changes to production, implement automated RAG assessment using the RAGAS (Retrieval Augmented Generation Assessment) framework:

• Faithfulness Metric: Evaluates if every claim in the LLM answer can be mathematically grounded in the retrieved chunks, ensuring zero hallucinations.
• Answer Relevance: Measures whether the generated answer directly addresses the user's initial inquiry without unnecessary verbosity.
• Context Recall: Verifies that all foundational information necessary to answer the question was successfully retrieved by pgvector.