RAG Implementation Guide 2026: Build Production RAG with pgvector

Published 2026-07-26 · 18 min read

Retrieval-Augmented Generation (RAG) is how you give an LLM your data without retraining it. Done right, it slashes hallucinations, keeps answers grounded in your sources, and lets you update knowledge by editing a database row instead of fine-tuning a model. Done wrong, it returns confidently wrong answers that are harder to debug than the hallucinations it was supposed to fix.

This guide builds a production-grade RAG stack on Postgres + pgvector — the same stack DrAI uses internally for its knowledge base. You'll get the schema, the chunking logic, the hybrid search query, and the evaluation harness. Everything is copy-pasteable.

Why pgvector (and Not a Dedicated Vector DB)?

In 2024 the default answer was "use Pinecone or Milvus." In 2026, for most teams, the answer is pgvector. Here's why:

• You already run Postgres. One database for relational data, metadata, and vectors. No second system to operate.
• Hybrid search is trivial. Full-text search (tsvector) and vector search in the same query, joined on the same row.
• HNSW indexing is fast enough. pgvector's HNSW handles 10M+ vectors with sub-50ms p95 queries on a single instance.
• Transactions. Update a document and its embeddings atomically. No eventual-consistency surprises.

Pinecone and Milvus still win at >100M vectors or for specialized workloads (see our vector DB comparison), but for 95% of RAG apps, pgvector is the right call and the simplest call.

Architecture Overview

Our stack has five stages:

1. Ingest — load documents (PDF, HTML, Markdown, docx).
2. Chunk — split into retrieval-sized units with overlap.
3. Embed — generate vectors via an embedding model.
4. Retrieve — hybrid (BM25 + vector) search with reranking.
5. Generate — pass top-k chunks as context to the LLM.

Each stage has failure modes. We'll cover them all.

Step 1: Database Schema

First, enable pgvector and create the tables:

-- Enable extensions
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;     -- for fuzzy text matching
CREATE EXTENSION IF NOT EXISTS unaccent;    -- accent-insensitive search

-- Documents table: one row per source file/page
CREATE TABLE documents (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    source_url  TEXT NOT NULL,
    title       TEXT,
    doc_type    TEXT,                        -- 'pdf', 'html', 'markdown'
    content_raw TEXT,                        -- full original text
    metadata    JSONB DEFAULT '{}'::jsonb,
    created_at  TIMESTAMPTZ DEFAULT now(),
    updated_at  TIMESTAMPTZ DEFAULT now()
);

-- Chunks table: one row per retrieval-sized unit
CREATE TABLE chunks (
    id           UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id  UUID REFERENCES documents(id) ON DELETE CASCADE,
    chunk_index  INT NOT NULL,
    content      TEXT NOT NULL,
    embedding    vector(1536),               -- matches text-embedding-3-small
    -- BM25-style full-text search column
    tsv          tsvector,
    metadata     JSONB DEFAULT '{}'::jsonb,
    created_at   TIMESTAMPTZ DEFAULT now()
);

-- Indexes
CREATE INDEX idx_chunks_document_id ON chunks(document_id);
CREATE INDEX idx_chunks_embedding ON chunks
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);
CREATE INDEX idx_chunks_tsv ON chunks USING gin(tsv);
CREATE INDEX idx_chunks_metadata ON chunks USING gin(metadata);

-- Auto-update the tsvector on insert/update
CREATE TRIGGER chunks_tsv_trigger BEFORE INSERT OR UPDATE
    ON chunks FOR EACH ROW EXECUTE FUNCTION
    tsvector_update_trigger(tsv, 'pg_catalog.english', content);

The vector(1536) dimension matches OpenAI's text-embedding-3-small. If you use a different embedding model, change the dimension — see the model table below.

Step 2: Pick an Embedding Model

The embedding model is the single biggest factor in retrieval quality. Don't default to whatever the first tutorial used.

For most use cases text-embedding-3-small is the sweet spot — 90% of the quality of the large model at 1/6th the cost. Switch to Voyage-3-large if you're benchmarking MTEB and need every point.

Step 3: Chunking Strategy

Chunking is where most RAG systems quietly fail. The goal is chunks that are self-contained enough to be useful out of context but small enough to fit several in the prompt.

Naive chunking (don't do this in production)

# BAD: fixed-size character chunks lose semantic structure
def naive_chunk(text, size=1000):
    return [text[i:i+size] for i in range(0, len(text), size)]

Recursive character chunking with overlap

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,           # overlap to preserve cross-boundary context
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
)

chunks = splitter.split_text(document_text)

Structure-aware chunking (better)

For Markdown or HTML, split on headers so each chunk is a logical section:

from langchain.text_splitter import MarkdownHeaderTextSplitter

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ]
)
sections = md_splitter.split_text(markdown_text)

# Then apply character splitting within each section
final_chunks = []
for section in sections:
    final_chunks.extend(splitter.split_text(section.page_content))

Rule of thumb: chunk size 400-600 tokens with 10-15% overlap. Smaller chunks = better retrieval precision but worse context. Run the eval harness (Step 8) to find your optimum.

Step 4: Generating Embeddings

import openai, os, asyncio
from typing import List

client = openai.OpenAI(
    base_url="https://ai.dr-ai.top/v1",
    api_key=os.environ["DRAI_KEY"],
)

def embed_batch(texts: List[str], model="text-embedding-3-small") -> List[List[float]]:
    """Embed up to 2048 texts per call. OpenAI truncates >8191 tokens."""
    resp = client.embeddings.create(model=model, input=texts)
    return [d.embedding for d in resp.data]

async def embed_async(corpus, batch_size=512):
    """Parallel embedding for large corpora."""
    sem = asyncio.Semaphore(8)  # 8 concurrent batches
    async def one(batch):
        async with sem:
            return await asyncio.to_thread(embed_batch, batch)
    batches = [corpus[i:i+batch_size] for i in range(0, len(corpus), batch_size)]
    results = await asyncio.gather(*[one(b) for b in batches])
    return [vec for batch in results for vec in batch]

Step 5: Inserting into pgvector

import psycopg, uuid
from psycopg.rows import dict_row_factory

def insert_chunks(conn, document_id, chunks_with_embeddings):
    """chunks_with_embeddings: list of (content, embedding, metadata)"""
    with conn.cursor() as cur:
        for i, (content, emb, meta) in enumerate(chunks_with_embeddings):
            cur.execute(
                """
                INSERT INTO chunks (id, document_id, chunk_index, content, embedding, metadata)
                VALUES (%s, %s, %s, %s, %s, %s)
                """,
                (str(uuid.uuid4()), str(document_id), i, content, emb, psycopg.types.json.Jsonb(meta))
            )
    conn.commit()

Note that the tsv column is populated automatically by the trigger we created — no separate code needed for full-text indexing.

Step 6: Hybrid Retrieval (the Heart of Production RAG)

Pure vector search misses exact-match queries ("what is SKU ABC-123?"). Pure BM25 misses semantic matches ("how do I reset my password" → "credentials"). Hybrid search combines both, and this is where pgvector shines — both indexes live in the same query.

-- Hybrid retrieval: combine vector similarity and BM25 full-text
WITH query_embedding AS (
    SELECT %s::vector(1536) AS emb
),
vector_scores AS (
    SELECT id, 1 - (embedding <=> (SELECT emb FROM query_embedding)) AS score
    FROM chunks
    ORDER BY embedding <=> (SELECT emb FROM query_embedding)
    LIMIT 100                          -- ANN candidate set
),
text_scores AS (
    SELECT id,
        ts_rank(tsv, plainto_tsquery('english', %s), 32) AS score
    FROM chunks
    WHERE tsv @@ plainto_tsquery('english', %s)
    LIMIT 100
)
-- Reciprocal Rank Fusion: combine rankings robustly
SELECT c.id, c.content, c.metadata,
       COALESCE(v.rank, 9999) AS vec_rank,
       COALESCE(t.rank, 9999) AS text_rank,
       -- RRF formula: weight=60 balances the two signals
       COALESCE(1.0 / (60 + v.rank), 0) +
       COALESCE(1.0 / (60 + t.rank), 0) AS fused_score
FROM chunks c
LEFT JOIN (
    SELECT id, ROW_NUMBER() OVER (ORDER BY score DESC) AS rank
    FROM vector_scores
) v ON c.id = v.id
LEFT JOIN (
    SELECT id, ROW_NUMBER() OVER (ORDER BY score DESC) AS rank
    FROM text_scores
) t ON c.id = t.id
WHERE v.id IS NOT NULL OR t.id IS NOT NULL
ORDER BY fused_score DESC
LIMIT 20;

Reciprocal Rank Fusion (RRF) is the production-grade way to combine rankings. Unlike weighted score averaging, it doesn't require normalizing scores across systems — just ranks. The constant 60 is the standard RRF parameter; tune between 20-80 if you're optimizing.

Python wrapper

def hybrid_search(conn, query_text, query_embedding, limit=20):
    with conn.cursor(row_factory=dict_row_factory) as cur:
        cur.execute(HYBRID_SQL, (query_embedding, query_text, query_text))
        return cur.fetchmany(limit)

results = hybrid_search(conn, "how to reset password", q_emb)
for r in results:
    print(r["fused_score"], r["content"][:80])

Step 7: Reranking (the 20% quality boost)

Vector retrieval gets you the top-20 relevant chunks cheaply. A cross-encoder reranker then re-scores those 20 against the query with much higher precision. This two-stage pattern is the single biggest quality lever in modern RAG.

from cohere import Client

cohere = Client(os.environ["COHERE_KEY"])

def rerank(query, candidates, top_n=5):
    resp = cohere.rerank(
        model="rerank-v3.5",
        query=query,
        documents=[c["content"] for c in candidates],
        top_n=top_n,
    )
    return [candidates[r.index] for r in resp.results]

# Pipeline: hybrid retrieve 50 → rerank to 5
candidates = hybrid_search(conn, query, q_emb, limit=50)
final = rerank(query, candidates, top_n=5)

Cross-encoder rerankers (Cohere, Voyage, BGE-reranker) add ~50-150ms latency but typically improve retrieval precision by 15-30% on eval sets. Worth it for almost every production use case.

Step 8: Building the Generation Prompt

Now pass the reranked chunks to the LLM. The prompt structure matters as much as the retrieval:

SYSTEM = """You are a precise assistant. Answer the user's question using
ONLY the provided context. If the context doesn't contain the answer,
say "I don't have enough information" — do not speculate.

For every factual claim, cite the source chunk ID in [brackets]."""

def build_prompt(query, chunks):
    context = "\n\n".join(
        f"[chunk_{c['id'][:8]}] {c['content']}" for c in chunks
    )
    return f"""Context:
{context}

Question: {query}

Answer (cite chunk IDs in brackets):"""

response = client.chat.completions.create(
    model="gpt-5-mini",              # cheap model is enough for grounded QA
    messages=[
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": build_prompt(query, final)},
    ],
    temperature=0.1,                 # low temperature = less hallucination
    response_format={"type": "text"},
)

Three things to notice: the system prompt forces citation (this both improves trust and makes hallucinations visible), the temperature is low, and we use GPT-5-mini rather than the flagship — for grounded QA the cheap model is usually sufficient. See our hallucination prevention guide for more grounding techniques.

Step 9: Evaluation (Without This, You're Flying Blind)

You cannot ship RAG without measuring it. Build an eval set of 50-200 (question, gold-answer, gold-source) triples, then track three metrics:

from ragas import evaluate
from ragas.metrics import context_recall, context_precision, faithfulness, answer_relevancy
from datasets import Dataset

eval_data = Dataset.from_dict({
    "question": [...],
    "answer": [...],          # generated by your pipeline
    "contexts": [[...]],      # chunks retrieved
    "ground_truth": [...],    # gold answers
})

scores = evaluate(eval_data, metrics=[
    context_recall, context_precision, faithfulness, answer_relevancy
])
print(scores)
# {'context_recall': 0.82, 'context_precision': 0.74,
#  'faithfulness': 0.91, 'answer_relevancy': 0.88}

Faithfulness below 0.85 means the LLM is inventing facts. Context recall below 0.7 means your retrieval is broken. Track these on every code change.

Step 10: Cost Optimization

RAG has three cost centers: embeddings (one-time per doc), retrieval (cheap), and generation (expensive). The generation step dominates. Three levers:

1. Use the cheapest LLM that passes eval. GPT-5-mini or Claude Haiku 4 cover 80% of grounded QA. Route hard questions to flagships.
2. Compress context. Rerank to top-3-5 chunks instead of stuffing 10. Each chunk is ~500 tokens; 5 chunks = 2.5k input tokens.
3. Cache embeddings and answers. Embed once. Cache LLM answers keyed on (query hash, top chunk IDs).

For a deeper dive, our AI cost optimization guide covers the math end-to-end.

Common RAG Failure Modes (and Fixes)

Advanced: Agentic RAG with Tool Use

For complex queries ("compare the refund policies of these 3 products"), single-shot retrieval isn't enough. An agentic RAG loop lets the LLM call the retrieval function multiple times, decide what to look up next, and synthesize across calls. The MCP protocol makes this clean — see our MCP guide for the implementation pattern.

tools = [{
    "type": "function",
    "function": {
        "name": "search_knowledge_base",
        "description": "Search the internal docs. Call multiple times if needed.",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        }
    }
}]

# Agent loop: LLM decides when to retrieve and when to answer
messages = [{"role": "user", "content": user_question}]
for _ in range(5):  # max 5 retrieval rounds
    resp = client.chat.completions.create(
        model="claude-sonnet-4-20250514",
        messages=messages, tools=tools, tool_choice="auto"
    )
    if not resp.choices[0].message.tool_calls:
        break  # model is done retrieving
    messages.append(resp.choices[0].message)
    for call in resp.choices[0].message.tool_calls:
        results = hybrid_search(conn, json.loads(call.arguments)["query"], ...)
        messages.append({"role": "tool", "tool_call_id": call.id,
                         "content": json.dumps(results[:5])})

Production Checklist

• [ ] pgvector HNSW index built with ef_search tuned (50-100)
• [ ] Hybrid search (BM25 + vector) with RRF fusion
• [ ] Cross-encoder reranker on top-50 → top-5
• [ ] Low-temperature generation with forced citation
• [ ] Eval harness running on every commit (Ragas or RAGAS-style)
• [ ] Embedding cache so you don't re-embed unchanged docs
• [ ] Query logging to find retrieval gaps in production
• [ ] Document freshness: re-embed when source changes (trigger on updated_at)
• [ ] Fallback: if retrieval returns nothing, the LLM must say "I don't know"

The Bottom Line

Production RAG in 2026 is not exotic — it's Postgres + pgvector + a good embedding model + hybrid retrieval + reranking + a disciplined eval loop. Skip any of those and quality suffers. Get them all right and you'll beat most "AI search" startups on retrieval quality at a fraction of their infrastructure cost.

DrAI exposes the embedding and generation models you need through one endpoint. Sign in, grab an API key, and the code above runs as-is.

Sources & Further Reading

• pgvector — vector similarity search for Postgres (official repo & docs)
• OpenAI embeddings documentation
• Cohere rerank — cross-encoder reranking documentation
• LangChain text-splitting documentation (chunking strategies)
• DrAI: Vector Database Comparison 2026