-- Create table with vector column
CREATE TABLE items (
    id SERIAL PRIMARY KEY,
    name TEXT,
    embedding VECTOR(3)  -- Vector with 3 dimensions
);

-- Insert vector
INSERT INTO items (name, embedding) VALUES
    ('item1', '[1, 2, 3]'),
    ('item2', '[4, 5, 6]'),
    ('item3', '[1, 2, 4]');

-- Query vector
SELECT  FROM items;

2. Distance Metrics

pgvector supports several distance metrics:

| Operator | Distance | Use Case |

|----------|----------|----------|

| <-> | L2 (Euclidean) | Default, general purpose |

| <#> | Inner Product (Negative) | Dot product similarity |

| <=> | Cosine Distance | Normalized vectors |

| <+> | L1 (Manhattan) | Sparse vectors |

-- L2 Distance (Euclidean)
SELECT name, embedding <-> '[1, 2, 3]' AS distance
FROM items
ORDER BY distance
LIMIT 5;

-- Cosine Distance
SELECT name, embedding <=> '[1, 2, 3]' AS distance
FROM items
ORDER BY distance
LIMIT 5;

-- Inner Product
SELECT name, embedding <#> '[1, 2, 3]' AS distance
FROM items
ORDER BY distance
LIMIT 5;

3. Vector Dimensions

-- Create IVFFlat index
CREATE INDEX ON items USING ivfflat (embedding vectorl2ops)

WITH (lists = 100);

-- For cosine distance
CREATE INDEX ON items USING ivfflat (embedding vectorcosineops)

WITH (lists = 100);

-- For inner product
CREATE INDEX ON items USING ivfflat (embedding vectoripops)

WITH (lists = 100);

• Number of clusters for the index
• Rule of thumb: sqrt(numrows) for < 1M rows
• For > 1M rows: sqrt(numrows) up to numrows / 1000

-- Set probes (number of clusters to search)
SET ivfflat.probes = 10;  -- Default: 1

-- More probes = more accurate but slower
SET ivfflat.probes = 50;

2. HNSW Index

HNSW (Hierarchical Navigable Small World) has faster queries but slower build time.

-- Create HNSW index
CREATE INDEX ON items USING hnsw (embedding vectorl2ops)
WITH (m = 16, efconstruction = 64);


-- For cosine distance
CREATE INDEX ON items USING hnsw (embedding vectorcosineops)

WITH (m = 16, efconstruction = 64);

• m: Number of connections per node (default: 16)
• efconstruction: Size of dynamic candidate list during construction (default: 64)

-- Set efsearch for query time
SET hnsw.efsearch = 100;  -- Default: 40


-- Higher = more accurate but slower
SET hnsw.efsearch = 200;

3. Choosing the Right Index

| Criteria | IVFFlat | HNSW |

|----------|---------|------|

| Build time | Fast | Slow |

| Query time | Medium | Fast |

| Memory | Low | High |

| Recall | Good with tuning | Very good |

| Use case | Large datasets | Speed critical |

Practice with Python

1. Setup Python Environment

pip install psycopg2-binary pgvector numpy openai sentence-transformers

2. Basic Operations

import psycopg2
from pgvector.psycopg2 import registervector

import numpy as np

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

Register vector type
registervector(conn)

Create table
cur = conn.cursor()
cur.execute("""
    CREATE TABLE IF NOT EXISTS documents (
        id SERIAL PRIMARY KEY,
        content TEXT,
        embedding VECTOR(384)
    )
""")
conn.commit()

Insert vector
embedding = np.random.rand(384).tolist()
cur.execute(
    "INSERT INTO documents (content, embedding) VALUES (%s, %s)",
    ("Sample document", embedding)
)
conn.commit()

Query similar vectors
queryembedding = np.random.rand(384).tolist()

cur.execute("""
    SELECT id, content, embedding <-> %s AS distance
    FROM documents
    ORDER BY distance
    LIMIT 5
""", (queryembedding,))

results = cur.fetchall()
for row in results:
    print(f"ID: {row[0]}, Content: {row[1]}, Distance: {row[2]:.4f}")

conn.close()

3. With Sentence Transformers

import psycopg2
from pgvector.psycopg2 import registervector

from sentencetransformers import SentenceTransformer

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

Connect
conn = psycopg2.connect(
    host="localhost",
    database="vectordb",
    user="postgres",
    password="mysecretpassword"
)
registervector(conn)

cur = conn.cursor()

Create table
cur.execute("""
    CREATE TABLE IF NOT EXISTS articles (
        id SERIAL PRIMARY KEY,
        title TEXT,
        content TEXT,
        embedding VECTOR(384)
    )
""")
conn.commit()

Sample documents
documents = [
    {"title": "Introduction to Machine Learning", "content": "Machine learning is a subset of artificial intelligence..."},
    {"title": "Deep Learning Basics", "content": "Deep learning uses neural networks with multiple layers..."},
    {"title": "Natural Language Processing", "content": "NLP enables computers to understand human language..."},
    {"title": "Computer Vision Applications", "content": "Computer vision allows machines to interpret images..."},
    {"title": "Reinforcement Learning", "content": "RL is about learning through interaction with environment..."},
]

Insert with embeddings
for doc in documents:
    embedding = model.encode(doc["content"]).tolist()
    cur.execute(
        "INSERT INTO articles (title, content, embedding) VALUES (%s, %s, %s)",
        (doc["title"], doc["content"], embedding)
    )
conn.commit()

Semantic search
def semanticsearch(query, limit=5):
    queryembedding = model.encode(query).tolist()


    cur.execute("""
        SELECT title, content, 1 - (embedding <=> %s) AS similarity
        FROM articles
        ORDER BY embedding <=> %s
        LIMIT %s
    """, (queryembedding, queryembedding, limit))


    return cur.fetchall()

Test search
results = semanticsearch("How do neural networks learn?")
print("\nSearch results for: 'How do neural networks learn?'")
for title, content, similarity in results:
    print(f"\n{title} (Similarity: {similarity:.4f})")
    print(f"  {content[:100]}...")

conn.close()

4. With OpenAI Embeddings

import psycopg2
from pgvector.psycopg2 import registervector

from openai import OpenAI

Initialize OpenAI client
client = OpenAI(apikey="your-api-key")

def getembedding(text, model="text-embedding-3-small"):

    """Get embedding from OpenAI API"""
    response = client.embeddings.create(
        input=text,
        model=model
    )
    return response.data[0].embedding

Connect
conn = psycopg2.connect(
    host="localhost",
    database="vectordb",
    user="postgres",
    password="mysecretpassword"
)
registervector(conn)
cur = conn.cursor()

Create table for OpenAI embeddings (1536 dimensions)
cur.execute("""
    CREATE TABLE IF NOT EXISTS knowledgebase (

        id SERIAL PRIMARY KEY,
        title TEXT,
        content TEXT,
        embedding VECTOR(1536),
        metadata JSONB
    )
""")
conn.commit()

Create HNSW index
cur.execute("""
    CREATE INDEX IF NOT EXISTS knowledgebaseembeddingidx
    ON knowledgebase USING hnsw (embedding vectorcosineops)

    WITH (m = 16, efconstruction = 64)
""")
conn.commit()

Insert document
def adddocument(title, content, metadata=None):

    embedding = getembedding(content)
    cur.execute(
        """INSERT INTO knowledgebase (title, content, embedding, metadata)

           VALUES (%s, %s, %s, %s) RETURNING id""",
        (title, content, embedding, metadata or {})
    )
    conn.commit()
    return cur.fetchone()[0]

Search
def searchdocuments(query, limit=5, threshold=0.7):
    queryembedding = getembedding(query)

    cur.execute("""
        SELECT
            id,
            title,
            content,
            1 - (embedding <=> %s) AS similarity
        FROM knowledgebase

        WHERE 1 - (embedding <=> %s) > %s
        ORDER BY embedding <=> %s
        LIMIT %s
    """, (queryembedding, queryembedding, threshold, queryembedding, limit))

    return cur.fetchall()

Example usage
adddocument(

    "PostgreSQL pgvector Guide",
    "pgvector is an extension for PostgreSQL that enables vector similarity search...",
    {"category": "database", "author": "Ruby"}
)

results = searchdocuments("How to do vector search in PostgreSQL?")
for docid, title, content, similarity in results:

    print(f"[{similarity:.4f}] {title}")

from langchaincommunity.vectorstores import PGVector
from langchainopenai import OpenAIEmbeddings

from langchain.textsplitter import RecursiveCharacterTextSplitter
from langchainopenai import ChatOpenAI

from langchain.chains import RetrievalQA

Connection string
CONNECTIONSTRING = "postgresql://postgres:mysecretpassword@localhost:5432/vectordb"

Initialize embeddings
embeddings = OpenAIEmbeddings()

Create vector store
vectorstore = PGVector(
    connectionstring=CONNECTIONSTRING,
    embeddingfunction=embeddings,

    collectionname="langchaindocs",

)

Add documents
documents = [
    "pgvector is a PostgreSQL extension for vector similarity search.",
    "It supports exact and approximate nearest neighbor search.",
    "HNSW and IVFFlat are the two main index types in pgvector.",
    "pgvector can store vectors up to 16,000 dimensions.",
]

Split documents
textsplitter = RecursiveCharacterTextSplitter(
    chunksize=500,

    chunkoverlap=50
)
texts = textsplitter.createdocuments(documents)

Add to vector store
vectorstore.adddocuments(texts)


Create retriever
retriever = vectorstore.asretriever(
    searchtype="similarity",

    searchkwargs={"k": 3}
)

Create QA chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
qachain = RetrievalQA.fromchaintype(

    llm=llm,
    chaintype="stuff",
    retriever=retriever,
    returnsourcedocuments=True
)

Query
query = "What index types does pgvector support?"
result = qachain.invoke({"query": query})


print(f"Question: {query}")
print(f"Answer: {result['result']}")
print("\nSources:")
for doc in result['sourcedocuments']:
    print(f"  - {doc.pagecontent[:100]}...")

import psycopg2
from pgvector.psycopg2 import registervector
from openai import OpenAI

class RAGSystem:
    def init(self, connectionstring, openaiapikey):

        self.client = OpenAI(apikey=openaiapikey)
        self.conn = psycopg2.connect(connectionstring)

        registervector(self.conn)
        self.cur = self.conn.cursor()
        self.setuptable()

    def setuptable(self):
        self.cur.execute("""
            CREATE TABLE IF NOT EXISTS ragdocuments (

                id SERIAL PRIMARY KEY,
                content TEXT,
                embedding VECTOR(1536),
                metadata JSONB,
                createdat TIMESTAMP DEFAULT CURRENTTIMESTAMP

            )
        """)
        self.cur.execute("""
            CREATE INDEX IF NOT EXISTS ragdocumentsembeddingidx
            ON ragdocuments USING hnsw (embedding vectorcosineops)

        """)
        self.conn.commit()

    def getembedding(self, text):

        response = self.client.embeddings.create(
            input=text,
            model="text-embedding-3-small"
        )
        return response.data[0].embedding

    def adddocument(self, content, metadata=None):
        embedding = self.getembedding(content)
        self.cur.execute(
            """INSERT INTO ragdocuments (content, embedding, metadata)

               VALUES (%s, %s, %s) RETURNING id""",
            (content, embedding, metadata or {})
        )
        self.conn.commit()
        return self.cur.fetchone()[0]

    def adddocuments(self, documents):
        """Add multiple documents efficiently"""
        for doc in documents:
            content = doc.get("content", doc) if isinstance(doc, dict) else doc
            metadata = doc.get("metadata", {}) if isinstance(doc, dict) else {}
            self.adddocument(content, metadata)


    def retrieve(self, query, k=5):
        queryembedding = self.getembedding(query)

        self.cur.execute("""
            SELECT content, 1 - (embedding <=> %s) AS similarity
            FROM ragdocuments

            ORDER BY embedding <=> %s
            LIMIT %s
        """, (queryembedding, queryembedding, k))


        return self.cur.fetchall()

    def generateresponse(self, query, k=5):
        # Retrieve relevant documents
        docs = self.retrieve(query, k)

        # Build context
        context = "\n\n".join([f"[Relevance: {sim:.2f}] {content}"
                               for content, sim in docs])

        # Generate response
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": f"""You are a helpful assistant.
                Answer the question based on the following context:

                {context}

                If the context doesn't contain enough information, say so."""},
                {"role": "user", "content": query}
            ],
            temperature=0.7
        )

        return {
            "answer": response.choices[0].message.content,
            "sources": [{"content": c, "similarity": s} for c, s in docs]
        }

Usage
rag = RAGSystem(
    "postgresql://postgres:mysecretpassword@localhost:5432/vectordb",
    "your-openai-api-key"
)

Add knowledge
rag.adddocuments([

    "pgvector supports L2 distance, inner product, and cosine distance.",
    "The HNSW index provides faster queries but slower build times.",
    "IVFFlat is recommended for large datasets where some accuracy loss is acceptable.",
])

Query
result = rag.generateresponse("What distance metrics does pgvector support?")
print(f"Answer: {result['answer']}")

Advanced Features

1. Hybrid Search (Vector + Full-text)

-- Table with metadata
CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name TEXT,
    category TEXT,
    price DECIMAL,
    embedding VECTOR(384)
);

CREATE INDEX ON products USING hnsw (embedding vectorcosineops);

CREATE INDEX ON products (category);
CREATE INDEX ON products (price);

-- Vector search with filter
SELECT name, price, 1 - (embedding <=> $1) AS similarity
FROM products
WHERE category = 'electronics'
  AND price BETWEEN 100 AND 500
ORDER BY embedding <=> $1
LIMIT 10;

import psycopg2
from pgvector.psycopg2 import registervector
import psycopg2.extras

conn = psycopg2.connect(...)
registervector(conn)


Batch insert
def batchinsert(documents):
    """Insert documents in batch for better performance"""
    data = [(doc['content'], doc['embedding']) for doc in documents]

    with conn.cursor() as cur:
        psycopg2.extras.executevalues(

            cur,
            "INSERT INTO documents (content, embedding) VALUES %s",
            data,
            template="(%s, %s::vector)"
        )
    conn.commit()

Usage
docs = [
    {"content": "Doc 1", "embedding": [0.1, 0.2, ...]},
    {"content": "Doc 2", "embedding": [0.3, 0.4, ...]},
    # ... more documents
]
batchinsert(docs)

Performance Tips

1. Index Maintenance

-- Reindex for performance
REINDEX INDEX CONCURRENTLY itemsembeddingidx;

-- Analyze table for query planner
ANALYZE items;

-- Vacuum for cleanup
VACUUM ANALYZE items;

2. Connection Pooling

from psycopg2 import pool

Create connection pool
connectionpool = pool.ThreadedConnectionPool(

    minconn=5,
    maxconn=20,
    host="localhost",
    database="vectordb",
    user="postgres",
    password="mysecretpassword"
)

def getconnection():
    return connectionpool.getconn()


def releaseconnection(conn):
    connectionpool.putconn(conn)