Production-ish RAG with pgvector
The numpy-array approach from the previous lesson stores vectors in process memory with no persistence, no filtering, and no way to update individual documents. Postgres with the pgvector extension solves all three: vectors live in a durable, indexed, SQL-queryable store that supports metadata filtering and incremental updates through standard INSERT/UPDATE/DELETE.
Setup: Postgres + pgvector
pgvector adds a vector column type and distance operators to Postgres. Install it as an extension:
CREATE EXTENSION IF NOT EXISTS vector;Create a table that stores chunks alongside their embeddings and metadata:
CREATE TABLE chunks (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
doc_id TEXT NOT NULL,
chunk_index INTEGER NOT NULL,
content TEXT NOT NULL,
embedding vector(1536),
content_hash TEXT NOT NULL,
metadata JSONB DEFAULT '{}',
indexed_at TIMESTAMPTZ DEFAULT now(),
UNIQUE (doc_id, chunk_index)
);vector(1536)matchestext-embedding-3-smalloutput dimensions. Change to 1024 forembed-v3or 3072 fortext-embedding-3-large.content_hashenables incremental ingestion — skip chunks whose content hasn't changed.metadata(JSONB) holds arbitrary key-value pairs for filtered search: source file, section title, document date, access tier.- The
UNIQUE (doc_id, chunk_index)constraint allows upserts viaON CONFLICT.
HNSW index
Brute-force sequential scan works up to ~100K vectors. Beyond that, create an HNSW index for approximate nearest neighbor search:
CREATE INDEX ON chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);vector_cosine_ops— use cosine distance (appropriate for normalized embeddings). Other options:vector_l2_opsfor Euclidean,vector_ip_opsfor inner product.m = 16— each node in the HNSW graph connects to 16 neighbors. Higher values increase recall but use more memory. 16 is the default and works well for most workloads.ef_construction = 200— controls index build quality. Higher = better recall at the cost of slower index builds. 200 is aggressive but worth it — you build the index once and query it thousands of times.
At query time, set the search beam width:
SET hnsw.ef_search = 100;ef_search = 100 gives recall@10 > 0.99 for most datasets with pgvector's HNSW. Lower it for faster queries at the cost of recall; raise it if you need recall@10 above 0.995.
Ingestion pipeline with content hashing
Production corpora change incrementally — a few documents added or updated per day. Re-embedding everything on each update wastes time and API credits. Content hashing lets you upsert only what changed:
import hashlib
import psycopg2
from psycopg2.extras import execute_values
from openai import OpenAI
openai_client = OpenAI()
def content_hash(text):
return hashlib.sha256(text.encode()).hexdigest()[:16]
def chunk_text(text, chunk_size=500, overlap=50):
words = text.split()
chunks = []
start = 0
while start < len(words):
end = start + chunk_size
chunks.append(" ".join(words[start:end]))
start += chunk_size - overlap
return chunks
def get_embeddings(texts, model="text-embedding-3-small"):
response = openai_client.embeddings.create(input=texts, model=model)
return [e.embedding for e in response.data]
def ingest_document(conn, doc_id, text, metadata=None):
metadata = metadata or {}
chunks = chunk_text(text)
with conn.cursor() as cur:
cur.execute(
"SELECT chunk_index, content_hash FROM chunks WHERE doc_id = %s",
(doc_id,)
)
existing = {row[0]: row[1] for row in cur.fetchall()}
changed = []
for i, chunk in enumerate(chunks):
h = content_hash(chunk)
if existing.get(i) != h:
changed.append((i, chunk, h))
if not changed:
return {"doc_id": doc_id, "skipped": len(chunks), "upserted": 0}
texts_to_embed = [c[1] for c in changed]
embeddings = get_embeddings(texts_to_embed)
rows = []
for (chunk_idx, chunk_text_val, h), emb in zip(changed, embeddings):
rows.append((doc_id, chunk_idx, chunk_text_val, emb, h,
psycopg2.extras.Json(metadata)))
with conn.cursor() as cur:
execute_values(cur, """
INSERT INTO chunks (doc_id, chunk_index, content, embedding,
content_hash, metadata)
VALUES %s
ON CONFLICT (doc_id, chunk_index) DO UPDATE SET
content = EXCLUDED.content,
embedding = EXCLUDED.embedding,
content_hash = EXCLUDED.content_hash,
metadata = EXCLUDED.metadata,
indexed_at = now()
""", rows, template="(%s, %s, %s, %s::vector, %s, %s)")
conn.commit()
return {"doc_id": doc_id, "skipped": len(chunks) - len(changed),
"upserted": len(changed)}The flow: split the document into chunks, hash each chunk, compare against stored hashes, embed only the changed chunks, upsert. On a 200-page document where 3 pages changed, this embeds 3 chunks instead of 400 — a 99% reduction in API calls.
Stale-document detection
When documents are deleted from the source corpus, their chunks linger in the database. The indexed_at timestamp catches this:
def remove_stale_documents(conn, known_doc_ids, before_timestamp=None):
with conn.cursor() as cur:
if before_timestamp:
cur.execute(
"DELETE FROM chunks WHERE indexed_at < %s RETURNING doc_id",
(before_timestamp,)
)
else:
cur.execute(
"DELETE FROM chunks WHERE doc_id != ALL(%s) RETURNING doc_id",
(list(known_doc_ids),)
)
deleted = cur.fetchall()
conn.commit()
return [row[0] for row in deleted]Run this after a full ingestion pass: anything not touched (its indexed_at is old) or not in the current source list gets pruned. This prevents the vector store from returning context from documents that no longer exist.
Pruning orphan chunks
When a document shrinks (e.g., a section is deleted), the old chunks beyond the new chunk count become orphans:
def prune_orphan_chunks(conn, doc_id, current_chunk_count):
with conn.cursor() as cur:
cur.execute(
"DELETE FROM chunks WHERE doc_id = %s AND chunk_index >= %s",
(doc_id, current_chunk_count)
)
conn.commit()Call this at the end of ingest_document after upserting the current chunks.
Querying with metadata filters
The power of SQL-backed vector search: combine semantic similarity with structured metadata filters in a single query.
def search_pgvector(conn, query, top_k=5, filters=None):
query_embedding = get_embeddings([query])[0]
where_clauses = []
params = [query_embedding, top_k]
if filters:
for key, value in filters.items():
where_clauses.append(f"metadata->>%s = %s")
params.insert(-1, key)
params.insert(-1, value)
where_sql = ""
if where_clauses:
where_sql = "WHERE " + " AND ".join(where_clauses)
with conn.cursor() as cur:
cur.execute(f"""
SELECT doc_id, chunk_index, content, metadata,
1 - (embedding <=> %s::vector) AS similarity
FROM chunks
{where_sql}
ORDER BY embedding <=> %s::vector
LIMIT %s
""", [query_embedding] + params[2:-1] + [query_embedding, top_k])
return [
{"doc_id": row[0], "chunk_index": row[1], "content": row[2],
"metadata": row[3], "similarity": float(row[4])}
for row in cur.fetchall()
]The <=> operator is pgvector's cosine distance. 1 - distance = similarity. The HNSW index accelerates this — Postgres uses the index automatically when the ORDER BY matches the index's distance operator.
A cleaner approach uses parameterized filtering directly:
def search_by_source(conn, query, source_type, top_k=5):
query_embedding = get_embeddings([query])[0]
with conn.cursor() as cur:
cur.execute("""
SELECT doc_id, chunk_index, content,
1 - (embedding <=> %s::vector) AS similarity
FROM chunks
WHERE metadata->>'source_type' = %s
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (query_embedding, source_type, query_embedding, top_k))
return [
{"doc_id": row[0], "chunk_index": row[1], "content": row[2],
"similarity": float(row[3])}
for row in cur.fetchall()
]
# Search only API documentation, ignoring blog posts and changelogs
results = search_by_source(conn, "authentication flow", "api_docs", top_k=5)Metadata filtering is the main reason to use a real database instead of a flat vector index. Without it, every query searches your entire corpus — including irrelevant document types, outdated versions, and access-restricted content.
Full ingestion script
Putting it together — a script that ingests a directory of text files:
import os
import psycopg2
conn = psycopg2.connect(
host="localhost", dbname="ragdb", user="rag", password="rag"
)
docs_dir = "./documents"
ingested_ids = set()
for filename in os.listdir(docs_dir):
if not filename.endswith(".txt"):
continue
doc_id = filename
text = open(os.path.join(docs_dir, filename)).read()
metadata = {
"source_type": "documentation",
"filename": filename,
}
result = ingest_document(conn, doc_id, text, metadata)
print(f"{doc_id}: upserted={result['upserted']}, skipped={result['skipped']}")
ingested_ids.add(doc_id)
chunks = chunk_text(text)
prune_orphan_chunks(conn, doc_id, len(chunks))
stale = remove_stale_documents(conn, ingested_ids)
if stale:
print(f"Removed stale documents: {stale}")
conn.close()Run this on a cron schedule or trigger it from a CI pipeline when documentation updates. The content-hash check makes it cheap to run frequently — unchanged documents cost zero API calls.
Query path: the same RAG loop
The query side is identical to the numpy version, with search_pgvector replacing the numpy dot product:
import anthropic
anthropic_client = anthropic.Anthropic()
def rag_query(conn, query, top_k=5, filters=None):
results = search_pgvector(conn, query, top_k, filters)
context = "\n\n".join(
f"[{i+1}] (similarity: {r['similarity']:.3f}, source: {r['doc_id']})\n{r['content']}"
for i, r in enumerate(results)
)
prompt = f"""Answer the question based on the provided context. If the context
doesn't contain enough information, say so. Cite sources by their block number.
Context:
{context}
Question: {query}"""
response = anthropic_client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}]
)
return {
"answer": response.content[0].text,
"sources": results,
}
conn = psycopg2.connect(host="localhost", dbname="ragdb", user="rag", password="rag")
result = rag_query(conn, "How do I configure authentication?",
filters={"source_type": "api_docs"})
print(result["answer"])Performance characteristics
pgvector with HNSW on a standard Postgres instance (4 vCPU, 16GB RAM):
- 100K vectors, 1536 dims — p99 query latency ~5ms, recall@10 > 0.99
- 1M vectors, 1536 dims — p99 query latency ~15ms, recall@10 > 0.98, index uses ~6GB RAM
- 10M vectors, 1536 dims — p99 query latency ~50ms, recall@10 > 0.95, index uses ~60GB RAM — requires tuning
mandef_constructionor moving to a dedicated vector database
For most production workloads under 5M vectors, pgvector on a managed Postgres instance (RDS, Cloud SQL, Supabase) is the right choice. You get vector search plus full SQL — transactions, joins, access control, backups — without adding a separate vector database to your infrastructure.