How to Build a Production RAG System Step by Step (Python, pgvector, Hybrid Search, Reranking)

#ai #python #rag #tutorial

This tutorial builds the retrieval core of a production RAG system. Not a notebook toy, but the pieces you actually need: tenant-safe vector search, hybrid retrieval, reranking and grounded answers.

Stack: Python, Postgres + pgvector, OpenAI embeddings, rank_bm25, sentence-transformers.

Step 1: Schema with tenant isolation

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
    id          BIGSERIAL PRIMARY KEY,
    tenant_id   TEXT NOT NULL,
    doc_title   TEXT,
    section     TEXT,
    content     TEXT NOT NULL,
    embedding   VECTOR(1536)
);

CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON chunks (tenant_id);

tenant_id lives in the table from day one. Permissions are enforced in SQL, not in the prompt.

Step 2: Structure-aware chunking

import re

def chunk_by_headings(text: str, doc_title: str, max_chars: int = 1500):
    sections = re.split(r"\n(?=#+ |\d+\.\s)", text)
    chunks = []
    for sec in sections:
        heading = sec.strip().split("\n")[0][:120]
        for i in range(0, len(sec), max_chars):
            chunks.append({
                "doc_title": doc_title,
                "section": heading,
                "content": f"{doc_title} > {heading}\n{sec[i:i + max_chars]}",
            })
    return chunks

Prefixing each chunk with its title and heading gives the embedding context that a raw slice of text does not have.

Step 3: Embed and store

import numpy as np
import psycopg
from pgvector.psycopg import register_vector
from langchain_openai import OpenAIEmbeddings

emb = OpenAIEmbeddings(model="text-embedding-3-small")  # 1536 dims
conn = psycopg.connect("postgresql://localhost/rag")
register_vector(conn)

def ingest(tenant_id, chunks):
    vectors = emb.embed_documents([c["content"] for c in chunks])
    with conn.cursor() as cur:
        for c, v in zip(chunks, vectors):
            cur.execute(
                "INSERT INTO chunks (tenant_id, doc_title, section, content, embedding) "
                "VALUES (%s, %s, %s, %s, %s)",
                (tenant_id, c["doc_title"], c["section"], c["content"], np.array(v)),
            )
    conn.commit()

Step 4: Hybrid retrieval with Reciprocal Rank Fusion

from rank_bm25 import BM25Okapi

def vector_search(tenant_id, query, k=30):
    qv = np.array(emb.embed_query(query))
    rows = conn.execute(
        "SELECT id, content FROM chunks WHERE tenant_id = %s "
        "ORDER BY embedding <=> %s LIMIT %s",
        (tenant_id, qv, k),
    ).fetchall()
    return rows

def keyword_search(tenant_id, query, k=30):
    rows = conn.execute(
        "SELECT id, content FROM chunks WHERE tenant_id = %s", (tenant_id,)
    ).fetchall()
    bm25 = BM25Okapi([r[1].lower().split() for r in rows])
    scores = bm25.get_scores(query.lower().split())
    ranked = sorted(zip(rows, scores), key=lambda x: x[1], reverse=True)
    return [r for r, _ in ranked[:k]]

def rrf(*result_lists, k=60):
    scores, docs = {}, {}
    for results in result_lists:
        for rank, (doc_id, content) in enumerate(results):
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
            docs[doc_id] = content
    return [(i, docs[i]) for i in sorted(scores, key=scores.get, reverse=True)]

In production, use Postgres full-text search or a search engine for the keyword side instead of loading rows into memory. The fusion logic stays the same.

Step 5: Rerank with a cross-encoder

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query, candidates, top_n=5):
    scores = reranker.predict([(query, c[1]) for c in candidates])
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [c for c, _ in ranked[:top_n]]

Step 6: Generate a grounded, cited answer

from langchain.chat_models import init_chat_model

llm = init_chat_model("openai:gpt-4.1-mini", temperature=0)

def answer(tenant_id, question):
    fused = rrf(vector_search(tenant_id, question), keyword_search(tenant_id, question))
    top = rerank(question, fused[:40])
    context = "\n\n".join(f"[{i}] {c}" for i, c in top)
    prompt = (
        "Answer ONLY from the sources below. Cite sources like [id]. "
        "If the answer is not in the sources, say you don't know.\n\n"
        f"Sources:\n{context}\n\nQuestion: {question}"
    )
    return llm.invoke(prompt).content

Step 7: Evaluate before you ship

Create 50 to 100 real questions with expected sources. Track faithfulness, answer relevance and context recall using RAGAS or DeepEval, and fail your CI pipeline when scores fall below your baseline.

What's next

The production version adds Docling for PDFs and tables, background ingestion workers, query rewriting, answer caching in Redis and observability with Langfuse.

That full system, RegRadar, is Project 2 in Vector 2.0, the live Gen-AI developer cohort by TechSimPlus taught by Prateek Mishra

https://vector.techsimplus.com

Visit the website and see the full curriculum.

DEV Community