Building Production RAG with PostgreSQL and pgvector in Python

Retrieval-Augmented Generation (RAG) is the industry standard for grounding Large Language Models in private domain data. While specialized vector databases like Pinecone, Chroma, and Qdrant are popular, managing a separate vector database creates data sync headaches, dual authentication configurations, and expensive infrastructure overhead.

In 2026, PostgreSQL with the pgvector extension has emerged as the premier choice for production RAG pipelines. It allows you to store relational customer data, transactions, and high-dimensional semantic embeddings side-by-side in a single ACID-compliant database.

In this deep guide, we implement an enterprise RAG pipeline using PostgreSQL, pgvector, and modern Python with psycopg3.

1. System Architecture: Relational + Vector Hybrid Workflow

┌─────────────────────────────────┐
│     Inbound User Question       │
└────────────────┬────────────────┘
                 │ 1. Compute Query Embedding (384-d / 1536-d)
                 ▼
┌─────────────────────────────────┐
│     pgvector HNSW Index Query   │ ◄─── Metadata Filter (tenant_id = 42)
│   (Cosine Distance: <=> )       │
└────────────────┬────────────────┘
                 │ 2. Top-k Relevant Chunks
                 ▼
┌─────────────────────────────────┐
│    Prompt Augmentation & LLM    │ ───► Accurate Grounded Answer
└─────────────────────────────────┘

By leveraging PostgreSQL’s HNSW (Hierarchical Navigable Small World) indexing, query latencies stay under 15ms even over millions of vectors.

2. Environment Setup

Install the required production libraries:

pip install psycopg[binary,pool] pgvector sentence-transformers pydantic

Enable the extension inside your PostgreSQL instance:

CREATE EXTENSION IF NOT EXISTS vector;

3. Database Schema Design with HNSW Indexing

Create schema.sql:

CREATE TABLE document_chunks (
    id BIGSERIAL PRIMARY KEY,
    document_title VARCHAR(255) NOT NULL,
    chunk_index INT NOT NULL,
    content TEXT NOT NULL,
    metadata JSONB DEFAULT '{}'::jsonb,
    embedding VECTOR(384), -- 384 dimensions for all-MiniLM-L6-v2
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Build HNSW index using Cosine Distance
CREATE INDEX ON document_chunks 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

4. Python Implementation: Chunking, Ingestion, and Semantic Retrieval

Create rag_engine.py:

import psycopg
from pgvector.psycopg import register_vector
from sentence_transformers import SentenceTransformer
from typing import List, Dict, Any

class PostgresRAGEngine:
    def __init__(self, db_url: str):
        self.db_url = db_url
        self.model = SentenceTransformer('all-MiniLM-L6-v2')

    def get_connection(self):
        conn = psycopg.connect(self.db_url)
        register_vector(conn)
        return conn

    def ingest_document(self, title: str, chunks: List[str]):
        # Encodes text chunks and batch inserts into PostgreSQL.
        embeddings = self.model.encode(chunks)

        with self.get_connection() as conn:
            with conn.cursor() as cur:
                data = [
                    (title, idx, chunk, emb.tolist())
                    for idx, (chunk, emb) in enumerate(zip(chunks, embeddings))
                ]
                cur.executemany(
                    "INSERT INTO document_chunks (document_title, chunk_index, content, embedding) VALUES (%s, %s, %s, %s)",
                    data
                )
            conn.commit()

    def search_similar(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
        # Queries the HNSW index using cosine similarity (<=>).
        query_vector = self.model.encode(query).tolist()

        with self.get_connection() as conn:
            with conn.cursor() as cur:
                # <=> is cosine distance; lower distance means higher similarity
                cur.execute(
                    "SELECT id, document_title, content, (1 - (embedding <=> %s)) AS similarity FROM document_chunks ORDER BY embedding <=> %s LIMIT %s;",
                    (query_vector, query_vector, limit)
                )
                rows = cur.fetchall()
                return [
                    {
                        "id": r[0],
                        "title": r[1],
                        "content": r[2],
                        "similarity": round(float(r[3]), 4)
                    }
                    for r in rows
                ]

if __name__ == "__main__":
    DB_URL = "postgresql://postgres:postgres@localhost:5432/ragdb"
    rag = PostgresRAGEngine(DB_URL)

    # Test Ingestion
    docs = [
        "PostgreSQL pgvector allows fast vector similarity searches directly inside SQL.",
        "LangGraph enables developers to build cyclic, stateful multi-agent workflows.",
        "Model Context Protocol standardizes LLM integration with external tools."
    ]
    rag.ingest_document("AI Tech Specs 2026", docs)

    # Test Query
    results = rag.search_similar("How does pgvector handle vector search in databases?")
    for res in results:
        print(f"[{res['similarity']}] {res['content']}")

5. Troubleshooting & Common Errors

6. Frequently Asked Questions (FAQ)

When should I use HNSW vs IVFFlat index?

Use HNSW for nearly all production use cases. It delivers 10x-50x faster query performance with higher recall at the expense of slightly slower build times and memory footprint. IVFFlat should only be considered for very constrained memory environments.

Can I filter by user ID or tenant ID during vector search?

Yes! Unlike standalone vector databases where metadata filtering is awkward, with pgvector you simply add WHERE tenant_id = 123 directly into the query, and PostgreSQL’s query optimizer combines the relational index and vector index automatically.