Build hybrid RAG search on PostgreSQL with pgvector

Approach

Store chunks with an embedding column and a full-text tsvector. At query time, run a vector search and a keyword search and merge them with reciprocal rank fusion, filtering by tenant inside the query. Pass the top chunks, labelled with IDs, to the model with instructions to answer only from them.

Solution

import anthropic
import psycopg

client = anthropic.Anthropic()

SCHEMA = '''
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE IF NOT EXISTS chunks (
  id bigserial PRIMARY KEY, tenant_id int NOT NULL, doc_id text NOT NULL, heading text, body text NOT NULL,
  embedding vector(1024) NOT NULL,
  tsv tsvector GENERATED ALWAYS AS (to_tsvector('english', coalesce(heading, '') || ' ' || body)) STORED);
CREATE INDEX IF NOT EXISTS chunks_embedding ON chunks USING hnsw (embedding vector_cosine_ops);
CREATE INDEX IF NOT EXISTS chunks_tsv ON chunks USING gin (tsv);
'''

HYBRID = '''
WITH semantic AS (
  SELECT id, row_number() OVER (ORDER BY embedding <=> %(qvec)s::vector) AS r
  FROM chunks WHERE tenant_id = %(tenant)s ORDER BY embedding <=> %(qvec)s::vector LIMIT 40),
keyword AS (
  SELECT id, row_number() OVER (ORDER BY ts_rank_cd(tsv, q) DESC) AS r
  FROM chunks, websearch_to_tsquery('english', %(question)s) q
  WHERE tenant_id = %(tenant)s AND tsv @@ q ORDER BY ts_rank_cd(tsv, q) DESC LIMIT 40)
SELECT c.id, c.doc_id, c.heading, c.body, sum(1.0 / (60 + r)) AS score
FROM (SELECT * FROM semantic UNION ALL SELECT * FROM keyword) ranks JOIN chunks c USING (id)
GROUP BY c.id ORDER BY score DESC LIMIT 6;
'''


def answer(conn: psycopg.Connection, tenant: int, question: str) -> str:
    qvec = embed(question)                         # your embedding model, returns a list of floats
    rows = conn.execute(HYBRID, {"qvec": str(qvec), "tenant": tenant, "question": question}).fetchall()
    if not rows:
        return "I couldn't find that in your documents."
    docs = "\n".join(f'<doc id="{r[0]}" source="{r[1]}" heading="{r[2]}">{r[3]}</doc>' for r in rows)
    response = client.messages.create(
        model="claude-sonnet-5-5", max_tokens=800,
        system="Answer only from the documents. Cite doc ids like [12]. If they don't answer it, say so.",
        messages=[{"role": "user", "content": f"{docs}\n\nQuestion: {question}"}],
    )
    return response.content[0].text

Pitfalls

• Filtering by tenant after retrieval instead of in the query (leaks and empty results).
• A tokenizer config that splits identifiers like E4012; test keyword search on your real queries.
• Mixing vectors from different embedding models in one column.

Variations

• Add a reranker between retrieval and generation for better precision.
• Use row-level security on chunks so a missing tenant filter fails closed.

Further reading: pgvector