Python pgvector

A PostgreSQL extension and Python library for storing, indexing, and querying vector embeddings for similarity search applications.

Overview

pgvector extends PostgreSQL with vector data types and similarity search capabilities, making it ideal for AI/ML applications like semantic search, recommendation systems, and RAG (Retrieval-Augmented Generation). The pgvector Python package provides seamless integration with psycopg2, SQLAlchemy, and other database adapters.

⤢ zoom

Extension Setup and Connection

Configure PostgreSQL with the pgvector extension and establish Python connections.

Key Concepts

• pgvector extension: Must be installed in PostgreSQL and enabled per database
• Connection adapters: Works with psycopg2, psycopg3, asyncpg, and SQLAlchemy
• Vector registration: Register vector type with your connection adapter
• Version compatibility: Requires PostgreSQL 11+ and pgvector 0.4.0+

Common Patterns

# Install the Python package
# pip install pgvector

# ============================================
# psycopg2 Connection
# ============================================
import psycopg2
from pgvector.psycopg2 import register_vector

# Connect to PostgreSQL
conn = psycopg2.connect(
    host="localhost",
    database="vectordb",
    user="postgres",
    password="password"
)

# Enable pgvector extension (once per database)
cur = conn.cursor()
cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
conn.commit()

# Register vector type with connection
register_vector(conn)

# ============================================
# psycopg3 Connection
# ============================================
import psycopg
from pgvector.psycopg import register_vector

conn = psycopg.connect(
    "host=localhost dbname=vectordb user=postgres password=password"
)

# Enable extension and register type
conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
register_vector(conn)

# ============================================
# asyncpg Connection (async)
# ============================================
import asyncpg
from pgvector.asyncpg import register_vector

async def main():
    conn = await asyncpg.connect(
        host="localhost",
        database="vectordb",
        user="postgres",
        password="password"
    )

    await conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
    await register_vector(conn)

    return conn

# ============================================
# SQLAlchemy Connection
# ============================================
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://postgres:password@localhost:5432/vectordb"
engine = create_engine(DATABASE_URL)

# Enable extension
with engine.connect() as conn:
    conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
    conn.commit()

SessionLocal = sessionmaker(bind=engine)

Examples

Connection with environment variables:

import os
import psycopg2
from pgvector.psycopg2 import register_vector

conn = psycopg2.connect(
    host=os.getenv("PGHOST", "localhost"),
    port=os.getenv("PGPORT", "5432"),
    database=os.getenv("PGDATABASE", "vectordb"),
    user=os.getenv("PGUSER", "postgres"),
    password=os.getenv("PGPASSWORD", "")
)

cur = conn.cursor()
cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
conn.commit()
register_vector(conn)

FastAPI dependency injection:

from fastapi import Depends
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker

DATABASE_URL = "postgresql://postgres:password@localhost/vectordb"
engine = create_engine(DATABASE_URL)

# Ensure extension is enabled at startup
with engine.connect() as conn:
    conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
    conn.commit()

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/search")
def search_vectors(query: str, db: Session = Depends(get_db)):
    # Use db for vector operations
    pass

Vector Column Definition

Define vector columns with specific dimensions for storing embeddings.

⤢ zoom

Key Concepts

• Vector dimensions: Must match your embedding model output (e.g., 384, 768, 1536)
• VECTOR type: PostgreSQL column type with fixed dimensions
• SQLAlchemy integration: Use Vector type from pgvector.sqlalchemy
• Dimension limits: Up to 16,000 dimensions for storage; HNSW/IVFFlat indexes support up to 2,000 dimensions (use halfvec cast for larger vectors)

Common Patterns

# ============================================
# Raw SQL Table Creation
# ============================================
import psycopg2
from pgvector.psycopg2 import register_vector

conn = psycopg2.connect("...")
register_vector(conn)
cur = conn.cursor()

# Create table with vector column
# Common embedding dimensions:
# - OpenAI text-embedding-3-small: 1536
# - sentence-transformers all-MiniLM-L6-v2: 384
# - Cohere embed-english-v3.0: 1024

cur.execute("""
    CREATE TABLE IF NOT EXISTS documents (
        id SERIAL PRIMARY KEY,
        title VARCHAR(255) NOT NULL,
        content TEXT,
        embedding VECTOR(1536)
    )
""")

# Multiple vector columns
cur.execute("""
    CREATE TABLE IF NOT EXISTS products (
        id SERIAL PRIMARY KEY,
        name VARCHAR(255) NOT NULL,
        price DECIMAL(10, 2),
        title_embedding VECTOR(384),
        description_embedding VECTOR(384),
        image_embedding VECTOR(512)
    )
""")

conn.commit()

# ============================================
# SQLAlchemy Model Definition
# ============================================
from sqlalchemy import Column, Integer, String, Text, Float
from sqlalchemy.orm import declarative_base
from pgvector.sqlalchemy import Vector

Base = declarative_base()

class Document(Base):
    __tablename__ = "documents"

    id = Column(Integer, primary_key=True)
    title = Column(String(255), nullable=False)
    content = Column(Text)
    embedding = Column(Vector(1536))  # OpenAI dimensions

    def __repr__(self):
        return f"<Document(id={self.id}, title='{self.title}')>"

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    price = Column(Float)
    title_embedding = Column(Vector(384))
    description_embedding = Column(Vector(384))

# Create tables
Base.metadata.create_all(engine)

Examples

SQLAlchemy 2.0 style with Mapped:

from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
from pgvector.sqlalchemy import Vector
from typing import Optional
import numpy as np

class Base(DeclarativeBase):
    pass

class Document(Base):
    __tablename__ = "documents"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(255))
    content: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    embedding: Mapped[list] = mapped_column(Vector(1536))

Altering existing tables:

# Add vector column to existing table
cur.execute("""
    ALTER TABLE articles
    ADD COLUMN embedding VECTOR(768)
""")

# Change vector dimensions (requires dropping and recreating)
cur.execute("ALTER TABLE documents DROP COLUMN embedding")
cur.execute("ALTER TABLE documents ADD COLUMN embedding VECTOR(384)")
conn.commit()

Vector Operations (Insert, Update, Query)

Perform CRUD operations with vector data.

⤢ zoom

Key Concepts

• Vector format: Pass vectors as Python lists or NumPy arrays
• Batch operations: Use executemany() or bulk insert for efficiency
• Type conversion: pgvector automatically converts between Python lists and PostgreSQL vectors
• NULL handling: Vector columns can be NULL if not specified as NOT NULL

Common Patterns

import numpy as np
from pgvector.psycopg2 import register_vector

# ============================================
# INSERT Operations
# ============================================

# Single insert with list
embedding = [0.1, 0.2, 0.3, 0.4, 0.5]  # Simplified example
cur.execute(
    "INSERT INTO documents (title, content, embedding) VALUES (%s, %s, %s)",
    ("My Document", "Document content here", embedding)
)
conn.commit()

# Insert with NumPy array
embedding = np.random.rand(1536).tolist()
cur.execute(
    "INSERT INTO documents (title, content, embedding) VALUES (%s, %s, %s)",
    ("NumPy Document", "Content", embedding)
)
conn.commit()

# Batch insert with executemany
documents = [
    ("Doc 1", "Content 1", np.random.rand(1536).tolist()),
    ("Doc 2", "Content 2", np.random.rand(1536).tolist()),
    ("Doc 3", "Content 3", np.random.rand(1536).tolist()),
]
cur.executemany(
    "INSERT INTO documents (title, content, embedding) VALUES (%s, %s, %s)",
    documents
)
conn.commit()

# Insert with RETURNING
cur.execute(
    """
    INSERT INTO documents (title, content, embedding)
    VALUES (%s, %s, %s)
    RETURNING id
    """,
    ("New Doc", "Content", embedding)
)
doc_id = cur.fetchone()[0]
conn.commit()

# ============================================
# UPDATE Operations
# ============================================

# Update single vector
new_embedding = np.random.rand(1536).tolist()
cur.execute(
    "UPDATE documents SET embedding = %s WHERE id = %s",
    (new_embedding, doc_id)
)
conn.commit()

# Update with condition
cur.execute(
    """
    UPDATE documents
    SET embedding = %s
    WHERE embedding IS NULL AND title = %s
    """,
    (new_embedding, "My Document")
)
conn.commit()

# ============================================
# SELECT Operations
# ============================================

# Select all vectors
cur.execute("SELECT id, title, embedding FROM documents")
rows = cur.fetchall()
for row in rows:
    doc_id, title, embedding = row
    print(f"ID: {doc_id}, Title: {title}, Dim: {len(embedding)}")

# Select specific document
cur.execute(
    "SELECT embedding FROM documents WHERE id = %s",
    (doc_id,)
)
result = cur.fetchone()
if result:
    embedding = np.array(result[0])
    print(f"Vector shape: {embedding.shape}")

# ============================================
# DELETE Operations
# ============================================

cur.execute("DELETE FROM documents WHERE id = %s", (doc_id,))
conn.commit()

Examples

SQLAlchemy CRUD operations:

from sqlalchemy.orm import Session
import numpy as np

def create_document(db: Session, title: str, content: str, embedding: list):
    doc = Document(
        title=title,
        content=content,
        embedding=embedding
    )
    db.add(doc)
    db.commit()
    db.refresh(doc)
    return doc

def get_document(db: Session, doc_id: int):
    return db.query(Document).filter(Document.id == doc_id).first()

def update_embedding(db: Session, doc_id: int, new_embedding: list):
    doc = db.query(Document).filter(Document.id == doc_id).first()
    if doc:
        doc.embedding = new_embedding
        db.commit()
        db.refresh(doc)
    return doc

def delete_document(db: Session, doc_id: int):
    doc = db.query(Document).filter(Document.id == doc_id).first()
    if doc:
        db.delete(doc)
        db.commit()
        return True
    return False

# Bulk insert
def bulk_create_documents(db: Session, documents_data: list):
    docs = [Document(**data) for data in documents_data]
    db.add_all(docs)
    db.commit()
    return docs

Generating embeddings with OpenAI:

import openai
import numpy as np

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

# Insert document with real embedding
title = "Machine Learning Basics"
content = "Machine learning is a subset of artificial intelligence..."
embedding = get_embedding(f"{title} {content}")

cur.execute(
    "INSERT INTO documents (title, content, embedding) VALUES (%s, %s, %s)",
    (title, content, embedding)
)
conn.commit()

Similarity Search

Query vectors using distance metrics: L2 (Euclidean), cosine similarity, and inner product.

⤢ zoom

Key Concepts

• L2 distance (<->): Euclidean distance, smaller is more similar
• Cosine distance (<=>): 1 - cosine similarity, smaller is more similar
• Inner product (<#>): Negative dot product, larger (less negative) is more similar
• Normalised vectors: Cosine and inner product give same results for normalised vectors
• K-nearest neighbours: Use ORDER BY and LIMIT for top-K queries

Common Patterns

import numpy as np

# Query vector (from your embedding model)
query_embedding = np.random.rand(1536).tolist()

# ============================================
# L2 Distance (Euclidean)
# ============================================
# Best for: General similarity when magnitude matters
# Smaller distance = more similar

cur.execute(
    """
    SELECT id, title, embedding <-> %s AS distance
    FROM documents
    ORDER BY distance
    LIMIT 10
    """,
    (query_embedding,)
)
results = cur.fetchall()

for doc_id, title, distance in results:
    print(f"{title}: {distance:.4f}")

# ============================================
# Cosine Distance
# ============================================
# Best for: Text embeddings, normalised vectors
# Range: 0 (identical) to 2 (opposite)

cur.execute(
    """
    SELECT id, title, embedding <=> %s AS distance
    FROM documents
    ORDER BY distance
    LIMIT 10
    """,
    (query_embedding,)
)
results = cur.fetchall()

# Convert to cosine similarity
for doc_id, title, distance in results:
    similarity = 1 - distance
    print(f"{title}: similarity={similarity:.4f}")

# ============================================
# Inner Product (Dot Product)
# ============================================
# Best for: Maximum inner product search (MIPS)
# Note: Returns negative inner product, so ORDER BY ASC for highest

cur.execute(
    """
    SELECT id, title, (embedding <#> %s) * -1 AS inner_product
    FROM documents
    ORDER BY embedding <#> %s
    LIMIT 10
    """,
    (query_embedding, query_embedding)
)
results = cur.fetchall()

for doc_id, title, ip in results:
    print(f"{title}: inner_product={ip:.4f}")

# ============================================
# Filtered Search
# ============================================

# Search within category
cur.execute(
    """
    SELECT id, title, embedding <=> %s AS distance
    FROM documents
    WHERE category = %s
    ORDER BY distance
    LIMIT 10
    """,
    (query_embedding, "technology")
)

# Search with minimum similarity threshold
cur.execute(
    """
    SELECT id, title, 1 - (embedding <=> %s) AS similarity
    FROM documents
    WHERE embedding <=> %s < 0.3
    ORDER BY embedding <=> %s
    LIMIT 10
    """,
    (query_embedding, query_embedding, query_embedding)
)

# Date-filtered search
cur.execute(
    """
    SELECT id, title, embedding <=> %s AS distance
    FROM documents
    WHERE created_at > NOW() - INTERVAL '30 days'
    ORDER BY distance
    LIMIT 10
    """,
    (query_embedding,)
)

Examples

SQLAlchemy similarity search:

from sqlalchemy import func, text
from sqlalchemy.orm import Session
from pgvector.sqlalchemy import Vector

def search_similar(db: Session, query_embedding: list, limit: int = 10):
    # Using cosine distance
    results = db.query(
        Document,
        Document.embedding.cosine_distance(query_embedding).label("distance")
    ).order_by(
        Document.embedding.cosine_distance(query_embedding)
    ).limit(limit).all()

    return [(doc, distance) for doc, distance in results]

def search_l2(db: Session, query_embedding: list, limit: int = 10):
    # Using L2 distance
    results = db.query(
        Document,
        Document.embedding.l2_distance(query_embedding).label("distance")
    ).order_by(
        Document.embedding.l2_distance(query_embedding)
    ).limit(limit).all()

    return results

def search_inner_product(db: Session, query_embedding: list, limit: int = 10):
    # Using inner product (max inner product search)
    results = db.query(
        Document,
        Document.embedding.max_inner_product(query_embedding).label("score")
    ).order_by(
        Document.embedding.max_inner_product(query_embedding)
    ).limit(limit).all()

    return results

Semantic search with sentence-transformers:

from sentence_transformers import SentenceTransformer
import psycopg2
from pgvector.psycopg2 import register_vector

# Load embedding model
model = SentenceTransformer('all-MiniLM-L6-v2')

# Connect to database
conn = psycopg2.connect("...")
register_vector(conn)
cur = conn.cursor()

def semantic_search(query: str, limit: int = 5):
    # Generate query embedding
    query_embedding = model.encode(query).tolist()

    # Search for similar documents
    cur.execute(
        """
        SELECT id, title, content, 1 - (embedding <=> %s) AS similarity
        FROM documents
        ORDER BY embedding <=> %s
        LIMIT %s
        """,
        (query_embedding, query_embedding, limit)
    )

    results = cur.fetchall()
    return [
        {
            "id": r[0],
            "title": r[1],
            "content": r[2],
            "similarity": float(r[3])
        }
        for r in results
    ]

# Usage
results = semantic_search("How does machine learning work?")
for r in results:
    print(f"{r['title']}: {r['similarity']:.3f}")

Hybrid search with keyword and vector:

def hybrid_search(query: str, query_embedding: list, limit: int = 10):
    cur.execute(
        """
        SELECT id, title, content,
               embedding <=> %s AS vector_distance,
               ts_rank(to_tsvector('english', content), plainto_tsquery(%s)) AS text_rank
        FROM documents
        WHERE to_tsvector('english', content) @@ plainto_tsquery(%s)
        ORDER BY vector_distance * 0.5 - text_rank * 0.5
        LIMIT %s
        """,
        (query_embedding, query, query, limit)
    )
    return cur.fetchall()

Index Types (IVFFlat, HNSW)

Create indexes to accelerate similarity searches on large datasets.

⤢ zoom

Key Concepts

• IVFFlat: Inverted file index, faster to build, uses less memory
• HNSW: Hierarchical Navigable Small World, better recall and speed
• Distance operators: Index must match query operator (<->, <=>, <#>)
• Recall vs speed: Trade-off between accuracy and query performance
• Build time: HNSW takes longer to build than IVFFlat

Common Patterns

# ============================================
# IVFFlat Index
# ============================================
# Faster to build, less memory
# Good for: Moderate datasets, memory constraints

# Create index for L2 distance
cur.execute("""
    CREATE INDEX ON documents
    USING ivfflat (embedding vector_l2_ops)
    WITH (lists = 100)
""")

# Create index for cosine distance
cur.execute("""
    CREATE INDEX ON documents
    USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 100)
""")

# Create index for inner product
cur.execute("""
    CREATE INDEX ON documents
    USING ivfflat (embedding vector_ip_ops)
    WITH (lists = 100)
""")

conn.commit()

# Set probes for search (default is 1)
# Higher probes = better recall, slower search
cur.execute("SET ivfflat.probes = 10")

# ============================================
# HNSW Index
# ============================================
# Better recall and speed, more memory and build time
# Good for: Production systems, high recall requirements

# Create index for L2 distance
cur.execute("""
    CREATE INDEX ON documents
    USING hnsw (embedding vector_l2_ops)
    WITH (m = 16, ef_construction = 64)
""")

# Create index for cosine distance
cur.execute("""
    CREATE INDEX ON documents
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64)
""")

# Create index for inner product
cur.execute("""
    CREATE INDEX ON documents
    USING hnsw (embedding vector_ip_ops)
    WITH (m = 16, ef_construction = 64)
""")

conn.commit()

# Set ef_search for queries (default is 40)
# Higher ef_search = better recall, slower search
cur.execute("SET hnsw.ef_search = 100")

# ============================================
# Index Parameters Guide
# ============================================

# IVFFlat lists parameter:
# - Recommended: rows / 1000 for up to 1M rows
# - Recommended: sqrt(rows) for over 1M rows
# - Example: 1M rows -> 1000 lists

# HNSW parameters:
# - m: Max connections per node (default 16)
#   Higher = better recall, more memory
# - ef_construction: Size of dynamic candidate list (default 64)
#   Higher = better index quality, slower build

Examples

Creating indexes with SQLAlchemy:

from sqlalchemy import Index, text
from pgvector.sqlalchemy import Vector

# Create HNSW index in model definition
class Document(Base):
    __tablename__ = "documents"

    id = Column(Integer, primary_key=True)
    title = Column(String(255))
    embedding = Column(Vector(1536))

    __table_args__ = (
        Index(
            'ix_documents_embedding_hnsw',
            'embedding',
            postgresql_using='hnsw',
            postgresql_with={'m': 16, 'ef_construction': 64},
            postgresql_ops={'embedding': 'vector_cosine_ops'}
        ),
    )

# Or create after table exists
with engine.connect() as conn:
    conn.execute(text("""
        CREATE INDEX CONCURRENTLY ix_documents_embedding
        ON documents
        USING hnsw (embedding vector_cosine_ops)
        WITH (m = 16, ef_construction = 64)
    """))
    conn.commit()

Choosing between IVFFlat and HNSW:

# Check your dataset size
cur.execute("SELECT COUNT(*) FROM documents")
row_count = cur.fetchone()[0]

if row_count < 10000:
    # Small dataset: exact search may be fine
    print("Consider exact search without index")
elif row_count < 100000:
    # Medium dataset: IVFFlat is efficient
    lists = max(row_count // 1000, 10)
    cur.execute(f"""
        CREATE INDEX ON documents
        USING ivfflat (embedding vector_cosine_ops)
        WITH (lists = {lists})
    """)
    print(f"Created IVFFlat index with {lists} lists")
else:
    # Large dataset: HNSW for better performance
    cur.execute("""
        CREATE INDEX ON documents
        USING hnsw (embedding vector_cosine_ops)
        WITH (m = 16, ef_construction = 100)
    """)
    print("Created HNSW index")

conn.commit()

Monitoring index build progress:

# For large indexes, build concurrently
cur.execute("""
    CREATE INDEX CONCURRENTLY idx_docs_embedding
    ON documents
    USING hnsw (embedding vector_cosine_ops)
""")

# Check index size
cur.execute("""
    SELECT pg_size_pretty(pg_relation_size('idx_docs_embedding'))
""")
print(f"Index size: {cur.fetchone()[0]}")

# Verify index is being used
cur.execute("EXPLAIN ANALYZE SELECT * FROM documents ORDER BY embedding <=> %s LIMIT 10",
            (query_embedding,))
for row in cur.fetchall():
    print(row[0])

Performance Optimisation

Optimise vector search performance through configuration, indexing, and query tuning.

⤢ zoom

Key Concepts

• Index tuning: Adjust parameters based on recall requirements and dataset size
• Query planning: Use EXPLAIN to understand query execution
• Memory allocation: Configure work_mem and maintenance_work_mem
• Parallel operations: Enable parallel index builds and queries
• Filtering strategy: Filter before or after vector search based on selectivity

Common Patterns

# ============================================
# Session Configuration
# ============================================

# Increase probes for better IVFFlat recall
cur.execute("SET ivfflat.probes = 20")  # Default: 1

# Increase ef_search for better HNSW recall
cur.execute("SET hnsw.ef_search = 200")  # Default: 40

# Increase work memory for sorting
cur.execute("SET work_mem = '256MB'")

# Enable parallel query execution
cur.execute("SET max_parallel_workers_per_gather = 4")

# ============================================
# Partial Indexes
# ============================================

# Index only active documents
cur.execute("""
    CREATE INDEX ON documents
    USING hnsw (embedding vector_cosine_ops)
    WHERE is_active = true
""")

# Index by category for filtered searches
cur.execute("""
    CREATE INDEX ON documents
    USING hnsw (embedding vector_cosine_ops)
    WHERE category = 'technology'
""")

# ============================================
# Optimised Queries
# ============================================

# Use explicit LIMIT (always required for index usage)
cur.execute(
    """
    SELECT id, title, embedding <=> %s AS distance
    FROM documents
    ORDER BY embedding <=> %s
    LIMIT 10
    """,
    (query_embedding, query_embedding)
)

# Pre-filter highly selective conditions
cur.execute(
    """
    SELECT id, title, embedding <=> %s AS distance
    FROM documents
    WHERE user_id = %s
      AND created_at > %s
    ORDER BY embedding <=> %s
    LIMIT 10
    """,
    (query_embedding, user_id, cutoff_date, query_embedding)
)

# ============================================
# Batch Processing
# ============================================

# Bulk embedding generation and insert
def bulk_embed_and_insert(documents: list, batch_size: int = 100):
    from sentence_transformers import SentenceTransformer
    model = SentenceTransformer('all-MiniLM-L6-v2')

    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        texts = [doc['content'] for doc in batch]

        # Batch encode for efficiency
        embeddings = model.encode(texts, batch_size=batch_size)

        # Prepare insert data
        data = [
            (doc['title'], doc['content'], embedding.tolist())
            for doc, embedding in zip(batch, embeddings)
        ]

        cur.executemany(
            "INSERT INTO documents (title, content, embedding) VALUES (%s, %s, %s)",
            data
        )
        conn.commit()
        print(f"Inserted {min(i + batch_size, len(documents))}/{len(documents)}")

Examples

Analysing query performance:

def analyse_query(query_embedding: list):
    # Use EXPLAIN ANALYZE to check index usage
    cur.execute(
        """
        EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)
        SELECT id, title
        FROM documents
        ORDER BY embedding <=> %s
        LIMIT 10
        """,
        (query_embedding,)
    )

    plan = cur.fetchone()[0]

    # Check if index is being used
    import json
    plan_data = json.loads(json.dumps(plan))

    execution_time = plan_data[0]['Execution Time']
    planning_time = plan_data[0]['Planning Time']

    print(f"Planning: {planning_time:.2f}ms")
    print(f"Execution: {execution_time:.2f}ms")

    return plan_data

# Compare index performance
def benchmark_search(query_embedding: list, num_queries: int = 100):
    import time

    start = time.time()
    for _ in range(num_queries):
        cur.execute(
            """
            SELECT id FROM documents
            ORDER BY embedding <=> %s
            LIMIT 10
            """,
            (query_embedding,)
        )
        cur.fetchall()

    elapsed = time.time() - start
    qps = num_queries / elapsed

    print(f"Queries per second: {qps:.2f}")
    print(f"Average latency: {elapsed/num_queries*1000:.2f}ms")

Connection pooling for high concurrency:

from psycopg2 import pool
from pgvector.psycopg2 import register_vector

# Create connection pool
connection_pool = pool.ThreadedConnectionPool(
    minconn=5,
    maxconn=20,
    host="localhost",
    database="vectordb",
    user="postgres",
    password="password"
)

def get_connection():
    conn = connection_pool.getconn()
    register_vector(conn)
    return conn

def release_connection(conn):
    connection_pool.putconn(conn)

# Usage
conn = get_connection()
try:
    cur = conn.cursor()
    cur.execute(
        "SELECT id FROM documents ORDER BY embedding <=> %s LIMIT 10",
        (query_embedding,)
    )
    results = cur.fetchall()
finally:
    release_connection(conn)

Approximate nearest neighbour tuning:

def tune_recall(query_embeddings: list, ground_truth: list):
    """
    Tune index parameters for target recall.
    ground_truth: list of exact nearest neighbour IDs per query
    """

    # Test different ef_search values
    for ef_search in [10, 20, 40, 80, 160, 320]:
        cur.execute(f"SET hnsw.ef_search = {ef_search}")

        total_recall = 0
        for query_emb, true_ids in zip(query_embeddings, ground_truth):
            cur.execute(
                """
                SELECT id FROM documents
                ORDER BY embedding <=> %s
                LIMIT 10
                """,
                (query_emb,)
            )
            result_ids = [r[0] for r in cur.fetchall()]

            # Calculate recall
            matches = len(set(result_ids) & set(true_ids))
            total_recall += matches / len(true_ids)

        avg_recall = total_recall / len(query_embeddings)
        print(f"ef_search={ef_search}: recall={avg_recall:.3f}")

Quick Reference

Task	Code
Enable extension	CREATE EXTENSION IF NOT EXISTS vector
Register type (psycopg2)	register_vector(conn)
Create vector column	embedding VECTOR(1536)
Insert vector	INSERT INTO t (emb) VALUES (%s)
L2 distance	embedding <-> query_vector
Cosine distance	embedding <=> query_vector
Inner product	embedding <#> query_vector
Cosine similarity	1 - (embedding <=> query_vector)
K-nearest search	ORDER BY embedding <=> %s LIMIT 10
Create IVFFlat index	CREATE INDEX ON t USING ivfflat (emb vector_cosine_ops)
Create HNSW index	CREATE INDEX ON t USING hnsw (emb vector_cosine_ops)
Set IVFFlat probes	SET ivfflat.probes = 10
Set HNSW ef_search	SET hnsw.ef_search = 100

Distance Operator Reference

Operator	Name	Use Case	Order
<->	L2 distance	General similarity	ASC (smaller = similar)
<=>	Cosine distance	Text embeddings	ASC (smaller = similar)
<#>	Inner product	MIPS	ASC (more negative = similar)

Index Operations Reference

Index Type	Operator	Operation Class
IVFFlat	<->	vector_l2_ops
IVFFlat	<=>	vector_cosine_ops
IVFFlat	<#>	vector_ip_ops
HNSW	<->	vector_l2_ops
HNSW	<=>	vector_cosine_ops
HNSW	<#>	vector_ip_ops

Common Embedding Dimensions

Model	Dimensions
OpenAI text-embedding-3-small	1536 (default; supports 64–1536 via dimensions parameter)
OpenAI text-embedding-3-large	3072
Cohere embed-english-v3.0	1024
sentence-transformers all-MiniLM-L6-v2	384
sentence-transformers all-mpnet-base-v2	768
BGE large	1024

Common Issues and Solutions

Issue	Solution
"type vector does not exist"	Run CREATE EXTENSION IF NOT EXISTS vector
"expected X dimensions, not Y"	Ensure embedding dimensions match column definition
Index not being used	Add LIMIT clause; check operator matches index type
Slow index build	Increase maintenance_work_mem; use CONCURRENTLY
Poor recall with IVFFlat	Increase ivfflat.probes (e.g., 10-20)
Poor recall with HNSW	Increase hnsw.ef_search (e.g., 100-200)
Out of memory on insert	Batch inserts; reduce maintenance_work_mem
Cannot install extension	Ensure pgvector is installed in PostgreSQL server
Cosine similarity always 1	Vectors may be zero or identical; check embedding generation
Index build takes too long	Use IVFFlat instead of HNSW; reduce ef_construction

Debugging Tips

# Check pgvector version
cur.execute("SELECT extversion FROM pg_extension WHERE extname = 'vector'")
print(f"pgvector version: {cur.fetchone()[0]}")

# Check vector dimensions
cur.execute("""
    SELECT column_name, udt_name, character_maximum_length
    FROM information_schema.columns
    WHERE table_name = 'documents' AND column_name = 'embedding'
""")
print(cur.fetchone())

# List all vector indexes
cur.execute("""
    SELECT indexname, indexdef
    FROM pg_indexes
    WHERE tablename = 'documents'
    AND indexdef LIKE '%vector%'
""")
for row in cur.fetchall():
    print(f"{row[0]}: {row[1]}")

# Check if index is used
cur.execute("""
    EXPLAIN SELECT id FROM documents
    ORDER BY embedding <=> %s LIMIT 10
""", (query_embedding,))
for row in cur.fetchall():
    print(row[0])

# Inspect vector values
cur.execute("SELECT id, embedding[:5] FROM documents LIMIT 1")
doc_id, partial = cur.fetchone()
print(f"First 5 dimensions of doc {doc_id}: {partial}")

Performance Troubleshooting

# Check table and index sizes
cur.execute("""
    SELECT
        pg_size_pretty(pg_total_relation_size('documents')) as total,
        pg_size_pretty(pg_relation_size('documents')) as table,
        pg_size_pretty(pg_indexes_size('documents')) as indexes
""")
sizes = cur.fetchone()
print(f"Total: {sizes[0]}, Table: {sizes[1]}, Indexes: {sizes[2]}")

# Monitor connection pool
print(f"Pool connections used: {connection_pool._used}")
print(f"Pool connections available: {connection_pool._pool}")

# Profile slow queries
cur.execute("SET log_min_duration_statement = 100")  # Log queries > 100ms

Related Topics

The following topics complement Python pgvector development and would make useful additions to your reference collection:

1. Python - SQLAlchemy: ORM patterns for pgvector models
2. PostgreSQL: Database administration and tuning
3. Python - FastAPI: Building vector search APIs
4. OpenTelemetry: Monitoring vector search performance
5. Database Patterns: Indexing strategies and query optimisation
6. Python - Redis: Caching embeddings and search results