Building Production-Grade RAG Systems with FastAPI & PostgreSQL Vector

While dedicated standalone vector databases have received significant attention, PostgreSQL equipped with the pgvector extension remains one of the most reliable and cost-effective solutions for enterprise RAG systems. It allows developers to keep relational data, metadata filters, user access rules, and vector embeddings within a unified, ACID-compliant database architecture.

This guide demonstrates how to construct a production-ready RAG search service combining PostgreSQL, pgvector HNSW indexing, and an asynchronous FastAPI backend.

1. PostgreSQL Schema with HNSW Vector Indexing

To deliver fast similarity search across hundreds of thousands of document embeddings, we use an HNSW (Hierarchical Navigable Small World) index in PostgreSQL. Here is the SQL schema definition:

-- Enable required PostgreSQL extensions
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;

-- Create documents table storing text content, full-text tsvector, and vector embeddings
CREATE TABLE documents (
    id BIGSERIAL PRIMARY KEY,
    title VARCHAR(500) NOT NULL,
    content TEXT NOT NULL,
    content_tsv TSVECTOR GENERATED ALWAYS AS (to_tsvector('english', content)) STORED,
    embedding vector(1536) NOT NULL, -- Standard dimension for embeddings
    metadata JSONB NOT NULL DEFAULT '{}',
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- Full-text GIN index for sparse keyword search
CREATE INDEX idx_documents_tsv ON documents USING gin(content_tsv);

-- Metadata JSONB GIN index for fast property filtering
CREATE INDEX idx_documents_metadata ON documents USING gin(metadata);

-- HNSW Vector Index for fast cosine similarity search
CREATE INDEX idx_documents_embedding_hnsw ON documents 
USING hnsw (embedding vector_cosine_ops) 
WITH (m = 16, ef_construction = 64);

2. Async FastAPI Service Implementation

Below is an async Python service built with FastAPI and asyncpg to execute hybrid vector and text searches efficiently:

# main.py - Asynchronous FastAPI Search Service
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
import asyncpg

app = FastAPI(title="Production RAG Search API", version="1.0.0")

class SearchQueryRequest(BaseModel):
    query_text: str
    embedding: List[float] = Field(..., min_items=1536, max_items=1536)
    top_k: int = Field(default=10, ge=1, le=50)

class SearchResultResponse(BaseModel):
    id: int
    title: str
    content: str
    similarity: float

# Connection pool setup
db_pool: Optional[asyncpg.Pool] = None

@app.on_event("startup")
async def startup():
    global db_pool
    db_pool = await asyncpg.create_pool(
        "postgresql://postgres:password@localhost:5432/rag_db",
        min_size=5,
        max_size=20
    )

@app.on_event("shutdown")
async def shutdown():
    if db_pool:
        await db_pool.close()

@app.post("/api/v1/search", response_model=List[SearchResultResponse])
async def search_documents(request: SearchQueryRequest):
    if not db_pool:
        raise HTTPException(status_code=500, detail="Database pool not initialized.")

    async with db_pool.acquire() as conn:
        # Perform cosine similarity query utilizing the HNSW index
        query_sql = """
            SELECT 
                id, 
                title, 
                content, 
                1 - (embedding <=> $1::vector) as similarity
            FROM documents
            WHERE is_active = TRUE
            ORDER BY embedding <=> $1::vector
            LIMIT $2;
        """
        rows = await conn.fetch(query_sql, str(request.embedding), request.top_k)
        
        return [
            SearchResultResponse(
                id=row["id"],
                title=row["title"],
                content=row["content"][:300] + "...",
                similarity=float(row["similarity"])
            )
            for row in rows
        ]

3. Performance Optimization Best Practices

When running PostgreSQL pgvector in high-traffic production environments, consider the following optimization strategies:

• Tune ef_search at Query Time: Increase the ef_search parameter (e.g., SET hnsw.ef_search = 100;) during complex queries to trade a few milliseconds of latency for higher recall accuracy.
• Keep Connection Pools Warm: Use async connection pools like asyncpg to avoid connection overhead on every API invocation.
• Implement Reciprocal Rank Fusion: Combine vector similarity scores with full-text search scores to ensure key terminology matches are prioritized alongside semantic search.

"PostgreSQL with pgvector gives developers transactional reliability and vector search capabilities in a single database, keeping systems simpler and easier to operate."

4. Conclusion

Building production RAG systems doesn't require complex infrastructure stacks. PostgreSQL with pgvector provides a reliable, high-performance foundation that scales gracefully alongside your application.