PostgreSQL + pgvector.

Why Postgres Over a Dedicated Vector DB

Setting Up pgvector

docker run -d --name pg-vector \
  -e POSTGRES_PASSWORD=devpassword \
  -p 5432:5432 \
  ankane/pgvector:latest

-- run once, inside the database
CREATE EXTENSION IF NOT EXISTS vector;

Schema Design

A vector(n) column stores fixed-length embeddings — n must match your embedding model's output size exactly (1536 for OpenAI's text-embedding-3-small, 768 for Gemini's default). Store the embedding right alongside the text and metadata it came from, in one row.

CREATE TABLE document_chunks (
    id SERIAL PRIMARY KEY,
    document_id TEXT NOT NULL,
    content TEXT NOT NULL,
    embedding VECTOR(1536) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now()
);

Inserting Embeddings

The ingestion phase from Topic 7 Section 0, now writing to a real table instead of an in-memory list.

import psycopg
from openai import OpenAI

client = OpenAI()

def get_embedding(text: str) -> list[float]:
    return client.embeddings.create(model="text-embedding-3-small", input=text).data[0].embedding

def ingest_chunk(conn: psycopg.Connection, document_id: str, content: str) -> None:
    embedding = get_embedding(content)
    conn.execute(
        "INSERT INTO document_chunks (document_id, content, embedding) VALUES (%s, %s, %s)",
        (document_id, content, embedding),
    )
    conn.commit()

Similarity Search & Indexes

pgvector adds distance operators directly to SQL — <=> for cosine distance (1 minus Topic 6's cosine similarity, so smaller is more similar). Without an index this is Topic 6's brute-force scan, just running inside the database instead of Python.

SELECT content, 1 - (embedding <=> %(query_embedding)s) AS similarity
FROM document_chunks
ORDER BY embedding <=> %(query_embedding)s
LIMIT 4;

For real scale, add an ANN index — this is where Topic 6 Section 4's "brute force stops scaling" gets solved concretely.

CREATE INDEX ON document_chunks
USING hnsw (embedding vector_cosine_ops);
-- HNSW: better query speed and recall, slower to build — the default good choice
-- IVFFlat is the faster-to-build, slightly lower-recall alternative for very large tables

Combining with Metadata Filters

Because this is still SQL, you get something an in-memory vector search doesn't hand you for free: filtering by ordinary columns before or alongside the vector search — "find similar chunks, but only from documents this user can access."

SELECT content, 1 - (embedding <=> %(query_embedding)s) AS similarity
FROM document_chunks
WHERE document_id = ANY(%(allowed_document_ids)s)   -- ordinary SQL filter
ORDER BY embedding <=> %(query_embedding)s
LIMIT 4;

Connecting from Python

pip install psycopg[binary] pgvector

import psycopg
from pgvector.psycopg import register_vector

conn = psycopg.connect("postgresql://postgres:devpassword@localhost:5432/postgres")
register_vector(conn)  # lets psycopg send/receive Python lists as VECTOR directly

def search(conn: psycopg.Connection, query: str, top_k: int = 4) -> list[tuple[str, float]]:
    query_embedding = get_embedding(query)
    rows = conn.execute(
        """SELECT content, 1 - (embedding <=> %s) AS similarity
           FROM document_chunks ORDER BY embedding <=> %s LIMIT %s""",
        (query_embedding, query_embedding, top_k),
    ).fetchall()
    return rows

Capstone: Migrate Topic 7's RAG Store to pgvector

Replace Topic 7's TinySearchIndex with a Postgres-backed version — same public interface, real persistence underneath.

import psycopg
from pgvector.psycopg import register_vector
from openai import OpenAI

client = OpenAI()

def get_embedding(text: str) -> list[float]:
    return client.embeddings.create(model="text-embedding-3-small", input=text).data[0].embedding

class PgVectorIndex:
    def __init__(self, dsn: str):
        self.conn = psycopg.connect(dsn)
        register_vector(self.conn)
        self.conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS document_chunks (
                id SERIAL PRIMARY KEY,
                content TEXT NOT NULL,
                embedding VECTOR(1536) NOT NULL
            )""")
        self.conn.commit()

    def add_documents(self, documents: list[str]) -> None:
        for doc in documents:
            embedding = get_embedding(doc)
            self.conn.execute(
                "INSERT INTO document_chunks (content, embedding) VALUES (%s, %s)",
                (doc, embedding),
            )
        self.conn.commit()

    def search(self, query: str, top_k: int = 4) -> list[tuple[str, float]]:
        query_embedding = get_embedding(query)
        rows = self.conn.execute(
            """SELECT content, 1 - (embedding <=> %s) AS similarity
               FROM document_chunks ORDER BY embedding <=> %s LIMIT %s""",
            (query_embedding, query_embedding, top_k),
        ).fetchall()
        return [(content, float(score)) for content, score in rows]


if __name__ == "__main__":
    index = PgVectorIndex("postgresql://postgres:devpassword@localhost:5432/postgres")
    index.add_documents([
        "Reset your password from the account settings page.",
        "Our refund policy allows returns within 30 days of purchase.",
        "The mobile app supports offline mode since version 4.2.",
    ])
    for content, score in index.search("can I get my money back"):
        print(f"{score:.3f}  {content}")