AI EngineeringMay 20, 20269 min read

Building Production RAG Systems

A complete RAG service in Python built on FastAPI, Postgres with pgvector, OpenAI, and a cross-encoder reranker, plus the engineering work that makes it safe to rely on.

Retrieval-augmented generation is the easiest impressive demo in AI and one of the harder things to run in production. I've spent the last few years building RAG into enterprise products like knowledge search, AI assistants, and document-heavy workflows. The gap between "it answered my test question" and "thousands of employees use it every day" comes almost entirely from the parts you never see in a demo. This post is the implementation I wish I'd had when I started: a complete, hand-rolled RAG service, along with the engineering that makes it trustworthy.

The mental model

Strip away the jargon and RAG is three steps: store your knowledge in a searchable form, retrieve the most relevant pieces for a question, and generate an answer grounded in what you retrieved. None of this makes the model smarter. Your job is to put better information in front of it at the right moment.

The three stages of a RAG pipeline: ingest and index documents, retrieve relevant context for a query, then generate a grounded answer

The stack

I'll build this with Python + FastAPI, Postgres with the pgvector extension, OpenAI for embeddings and generation, and a cross-encoder for reranking. Two deliberate choices:

One datastore. Postgres holds the documents, the vectors (dense search), and a full-text index (sparse search). One system to operate instead of three. You don't need a dedicated vector database until you actually do.
No framework. Nothing is hidden behind a Chain abstraction, so every moving part is visible and tunable. The whole service is about 150 lines.

pip install "openai>=1.0" "psycopg[binary,pool]" pgvector tiktoken \
    sentence-transformers fastapi uvicorn tenacity

The data model

Everything starts with the schema. One table holds chunks with both a vector column (for semantic search) and a generated tsvector column (for keyword search), each with its own index.

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
    id          BIGSERIAL PRIMARY KEY,
    document_id TEXT         NOT NULL,
    ordinal     INT          NOT NULL,
    content     TEXT         NOT NULL,
    embedding   VECTOR(1536) NOT NULL,            -- text-embedding-3-small
    metadata    JSONB        NOT NULL DEFAULT '{}',
    tsv         TSVECTOR GENERATED ALWAYS AS (to_tsvector('english', content)) STORED
);

-- Dense (approximate nearest neighbour) and sparse (full-text) indexes.
CREATE INDEX chunks_embedding_idx ON chunks USING hnsw (embedding vector_cosine_ops);
CREATE INDEX chunks_tsv_idx       ON chunks USING gin  (tsv);
CREATE INDEX chunks_document_idx  ON chunks (document_id);

The clients and connection pool, shared across the app:

import os
from openai import AsyncOpenAI
from psycopg_pool import AsyncConnectionPool
from pgvector.psycopg import register_vector_async

EMBED_MODEL = "text-embedding-3-small"
GEN_MODEL   = "gpt-4o-mini"

openai = AsyncOpenAI()  # reads OPENAI_API_KEY from the environment

async def _configure(conn):
    await register_vector_async(conn)  # teaches psycopg how to send/receive VECTOR

pool = AsyncConnectionPool(
    conninfo=os.environ["DATABASE_URL"],
    open=False,
    configure=_configure,
)

Chunking

This is the part that took me the longest to get right. Slicing documents into fixed-size blocks is easy and usually wrong, because it cuts ideas in half. Split on paragraphs, pack them up to a token budget, and carry a little overlap between chunks so a fact never gets stranded across two chunks where neither one can be retrieved on its own.

import tiktoken

_enc = tiktoken.get_encoding("cl100k_base")

def _ntokens(text: str) -> int:
    return len(_enc.encode(text))

def chunk_text(text: str, *, max_tokens: int = 512, overlap: float = 0.15) -> list[str]:
    """Pack paragraphs into ~max_tokens chunks, carrying overlap across boundaries.

    (Pre-split any single paragraph longer than max_tokens before calling this.)
    """
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    overlap_budget = int(max_tokens * overlap)
    chunks: list[str] = []
    window: list[str] = []

    for para in paragraphs:
        window.append(para)
        if sum(_ntokens(p) for p in window) < max_tokens:
            continue
        chunks.append("\n\n".join(window))
        # Carry the trailing paragraphs that fit the overlap budget into the next window.
        carried, used = [], 0
        for p in reversed(window):
            used += _ntokens(p)
            if used > overlap_budget:
                break
            carried.insert(0, p)
        window = carried

    if window:
        chunks.append("\n\n".join(window))
    return chunks

Embedding and indexing

Embed a document's chunks in one request, then write them in a transaction. Deleting the old chunks first makes re-indexing idempotent, which also gives you incremental updates without a full rebuild.

import json
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(5), wait=wait_exponential(min=1, max=30))
async def embed(texts: list[str]) -> list[list[float]]:
    """Embed a batch in a single request, with backoff on transient API errors."""
    resp = await openai.embeddings.create(model=EMBED_MODEL, input=texts)
    return [item.embedding for item in resp.data]

async def index_document(document_id: str, text: str, metadata: dict | None = None) -> int:
    """Chunk, embed, and (re)index one document. Re-indexing replaces its old chunks."""
    chunks = chunk_text(text)
    if not chunks:
        return 0
    vectors = await embed(chunks)  # for very large docs, batch this in groups of ~256
    metadata = metadata or {}

    async with pool.connection() as conn, conn.cursor() as cur:
        async with conn.transaction():
            await cur.execute("DELETE FROM chunks WHERE document_id = %s", (document_id,))
            await cur.executemany(
                """INSERT INTO chunks (document_id, ordinal, content, embedding, metadata)
                   VALUES (%s, %s, %s, %s, %s)""",
                [
                    (document_id, i, content, vector, json.dumps(metadata))
                    for i, (content, vector) in enumerate(zip(chunks, vectors))
                ],
            )
    return len(chunks)

Hybrid retrieval

If the right context never reaches the model, no prompt will save the answer. Most of the "the AI is wrong" bugs I've debugged turned out to be retrieval failures. Pure semantic search is great at meaning and bad at exact terms like error codes and names, and keyword search is the opposite. So I run both and merge the results with Reciprocal Rank Fusion. Since everything lives in Postgres, that's a single query.

Hybrid retrieval with reranking: run dense (semantic) and sparse (keyword) search in parallel, merge the candidates, then rerank with a cross-encoder

from dataclasses import dataclass
import numpy as np
from psycopg.rows import dict_row

@dataclass
class Chunk:
    id: int
    content: str
    metadata: dict

CANDIDATES = 50   # how many to pull from each retriever before fusing
RRF_K      = 60   # reciprocal-rank-fusion constant

async def hybrid_search(query: str, *, candidates: int = CANDIDATES) -> list[Chunk]:
    """Dense + sparse retrieval, fused with RRF, in one round-trip."""
    [query_vector] = await embed([query])
    # pgvector's adapter sends numpy arrays as VECTOR. A plain list would go as
    # double precision[], and Postgres has no <=> operator for that.
    query_vector = np.array(query_vector)

    async with pool.connection() as conn:
        async with conn.cursor(row_factory=dict_row) as cur:
            await cur.execute(
                """
                WITH dense AS (
                    SELECT id, content, metadata,
                           ROW_NUMBER() OVER (ORDER BY embedding <=> %(vec)s) AS rank
                    FROM chunks
                    ORDER BY embedding <=> %(vec)s
                    LIMIT %(n)s
                ),
                sparse AS (
                    SELECT id, content, metadata,
                           ROW_NUMBER() OVER (
                               ORDER BY ts_rank_cd(tsv, websearch_to_tsquery('english', %(q)s)) DESC
                           ) AS rank
                    FROM chunks
                    WHERE tsv @@ websearch_to_tsquery('english', %(q)s)
                    LIMIT %(n)s
                )
                SELECT COALESCE(d.id, s.id)             AS id,
                       COALESCE(d.content, s.content)   AS content,
                       COALESCE(d.metadata, s.metadata) AS metadata,
                       COALESCE(1.0 / (%(rrf)s + d.rank), 0)
                     + COALESCE(1.0 / (%(rrf)s + s.rank), 0) AS score
                FROM dense d
                FULL OUTER JOIN sparse s USING (id)
                ORDER BY score DESC
                LIMIT %(n)s
                """,
                {"vec": query_vector, "q": query, "n": candidates, "rrf": RRF_K},
            )
            rows = await cur.fetchall()

    return [Chunk(id=r["id"], content=r["content"], metadata=r["metadata"]) for r in rows]

Reranking

Vector search is fast but approximate. I take the fused candidates and pass them through a cross-encoder, a model that reads the query and the passage together, and reorder them by actual relevance. It costs a little latency and buys a lot of precision.

import asyncio
from sentence_transformers import CrossEncoder

# Load once at startup. The model is hundreds of MB, so don't load it per request.
_reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

@dataclass
class Scored:
    chunk: Chunk
    score: float

def _rerank_sync(query: str, chunks: list[Chunk], top_k: int) -> list[Scored]:
    if not chunks:
        return []
    scores = _reranker.predict([(query, c.content) for c in chunks])
    ranked = sorted(
        (Scored(chunk, float(score)) for chunk, score in zip(chunks, scores)),
        key=lambda s: s.score,
        reverse=True,
    )
    return ranked[:top_k]

async def rerank(query: str, chunks: list[Chunk], *, top_k: int = 6) -> list[Scored]:
    # Cross-encoder inference is CPU-bound; keep it off the event loop.
    return await asyncio.to_thread(_rerank_sync, query, chunks, top_k)

Grounded generation

This is the easy part, with two details I always insist on: every answer cites its sources, and a confidence gate lets the system say "I don't know" instead of making something up. In an enterprise setting, a confident wrong answer does more damage than no answer at all.

@dataclass
class Answer:
    text: str
    sources: list[Scored]

MIN_RELEVANCE = 0.0  # cross-encoder logit floor; tune this on your eval set

SYSTEM_PROMPT = (
    "You are a precise assistant. Answer the question using ONLY the numbered sources. "
    "Cite the sources you use inline as [1], [2], and so on. "
    "If the sources do not contain the answer, say you don't know. Never guess."
)

def _format_sources(sources: list[Scored]) -> str:
    return "\n\n".join(f"[{i + 1}] {s.chunk.content}" for i, s in enumerate(sources))

async def answer(query: str) -> Answer:
    candidates = await hybrid_search(query)
    ranked = await rerank(query, candidates)

    # Confidence gate: refuse instead of hallucinating when nothing is clearly relevant.
    if not ranked or ranked[0].score < MIN_RELEVANCE:
        return Answer("I don't have enough information to answer that.", sources=[])

    resp = await openai.chat.completions.create(
        model=GEN_MODEL,
        temperature=0.1,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user",
             "content": f"Question: {query}\n\nSources:\n{_format_sources(ranked)}"},
        ],
    )
    return Answer(resp.choices[0].message.content or "", sources=ranked)

Tying it together: the API

A single endpoint takes a question, runs the pipeline, and returns the answer with the sources it used, so the client can render citations and the user can check them.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import BaseModel

@asynccontextmanager
async def lifespan(app: FastAPI):
    await pool.open()
    yield
    await pool.close()

app = FastAPI(lifespan=lifespan)

class AskRequest(BaseModel):
    question: str

class SourceOut(BaseModel):
    id: int
    score: float
    content: str
    metadata: dict

class AskResponse(BaseModel):
    answer: str
    sources: list[SourceOut]

@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest) -> AskResponse:
    result = await answer(req.question)
    return AskResponse(
        answer=result.text,
        sources=[
            SourceOut(id=s.chunk.id, score=s.score,
                      content=s.chunk.content, metadata=s.chunk.metadata)
            for s in result.sources
        ],
    )

Knowing it works: evaluation

"It feels better" is not a good enough reason to deploy. Build a golden set of real questions with known-relevant chunk IDs and track retrieval quality on every change. recall@k answers the most important question first: is the right context being found at all?

@dataclass
class EvalCase:
    query: str
    relevant_ids: set[int]

async def recall_at_k(cases: list[EvalCase], k: int = 6) -> float:
    """Fraction of queries where a known-relevant chunk appears in the top-k."""
    hits = 0
    for case in cases:
        retrieved = await hybrid_search(case.query, candidates=k)
        if {c.id for c in retrieved} & case.relevant_ids:
            hits += 1
    return hits / len(cases)

Wire that into CI and a regression becomes a failing check instead of a user complaint.

What makes this production-grade

It isn't much code, but the details are what separate it from a weekend demo:

Hybrid retrieval + RRF instead of vector search alone, so it catches exact terms and meaning.
Reranking to turn approximate recall into real precision.
Citations + a confidence gate so answers are verifiable and the system can decline.
Batched embeds, retries, a connection pool, and transactional re-indexing. This is the boring reliability work that keeps it running at scale.
One datastore to operate, and no framework hiding the behavior you need to tune.

One extension is worth calling out because every enterprise product I've built has needed it: permission-aware retrieval. The query carries the asking user's allowed scopes and the SQL filters on them, so the system can only retrieve chunks that user is allowed to see. It's a one-line change to each retrieval CTE, and if you get it wrong you have a security incident on your hands.

-- inside both the dense and sparse CTEs:
WHERE metadata->>'acl' = ANY(%(user_scopes)s)

Conclusion

Most of the work in RAG is ordinary engineering. The demo takes three lines. The product takes careful retrieval, honest grounding, real evaluation, and the same unglamorous reliability work you'd put into any other service. I kept everything here in plain Python and SQL on purpose: when you can see every moving part, you can measure it, tune it, and trust it.

So work in this order: retrieval and evaluation first, the prompt after. Get the right context in front of the model and hold yourself to a number, and the rest gets much easier.