Vector Databases — Pinecone vs pgvector vs Chroma

import numpy as np
import time

def exact_knn(query: np.ndarray, vectors: np.ndarray, k: int = 5) -> list[int]:
    """Exact KNN — compare against every vector. O(n)."""
    distances = np.linalg.norm(vectors - query, axis=1)
    return np.argsort(distances)[:k].tolist()

# With 1M vectors of dimension 1536:
# Exact KNN: ~500ms per query
# HNSW (ANN): ~2ms per query, 99%+ recall

Layer 3: [A] -------------- [F]           (few nodes, big jumps)
Layer 2: [A] --- [C] --- [F] --- [H]      (more nodes, smaller jumps)  
Layer 1: [A]-[B]-[C]-[D]-[E]-[F]-[G]-[H]  (all nodes, fine-grained)

pip install chromadb

import chromadb

# In-memory (for testing)
client = chromadb.Client()

# Persistent (for production)
client = chromadb.PersistentClient(path="./chroma_data")

# Create a collection
collection = client.get_or_create_collection(
    name="documents",
    metadata={
        "hnsw:space": "cosine",     # Distance metric: cosine, l2, or ip
        "hnsw:M": 16,               # HNSW connections per node (default 16)
        "hnsw:construction_ef": 100, # Build-time search width (default 100)
        "hnsw:search_ef": 50,        # Query-time search width (default 10)
    },
)

# CREATE — add documents
collection.add(
    ids=["doc1", "doc2", "doc3"],
    documents=[
        "How to reset your password",
        "Billing FAQ and payment methods",
        "API rate limits and quotas",
    ],
    metadatas=[
        {"category": "account", "priority": "high"},
        {"category": "billing", "priority": "medium"},
        {"category": "technical", "priority": "low"},
    ],
)

# READ — get by ID
results = collection.get(ids=["doc1", "doc2"])
print(results["documents"])  # ["How to reset your password", "Billing FAQ..."]

# UPDATE — upsert (update if exists, insert if not)
collection.upsert(
    ids=["doc1"],
    documents=["Updated: How to reset your password — new 2FA flow"],
    metadatas=[{"category": "account", "priority": "high", "updated": True}],
)

# DELETE
collection.delete(ids=["doc3"])

# DELETE with filter
collection.delete(where={"category": "billing"})

# Basic similarity search
results = collection.query(
    query_texts=["I forgot my login credentials"],
    n_results=5,
)

# Search with metadata filter
results = collection.query(
    query_texts=["payment issue"],
    n_results=5,
    where={"category": "billing"},  # Only search billing documents
)

# Complex filters
results = collection.query(
    query_texts=["urgent problem"],
    n_results=10,
    where={
        "$and": [
            {"category": {"$in": ["account", "billing"]}},
            {"priority": {"$eq": "high"}},
        ]
    },
)

# Access results
for i in range(len(results["documents"][0])):
    print(f"Score: {1 - results['distances'][0][i]:.3f}")
    print(f"  Doc: {results['documents'][0][i][:100]}")
    print(f"  Meta: {results['metadatas'][0][i]}")

from chromadb.utils import embedding_functions

# OpenAI embeddings
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="sk-...",
    model_name="text-embedding-3-small",
)

# Sentence Transformers (local, free)
st_ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="all-MiniLM-L6-v2",
)

# Use with collection
collection = client.get_or_create_collection(
    name="documents",
    embedding_function=openai_ef,  # Applied automatically on add/query
)

-- Enable the extension (requires pgvector installed on the server)
CREATE EXTENSION IF NOT EXISTS vector;

docker run -d \
  --name pgvector \
  -e POSTGRES_PASSWORD=password \
  -p 5432:5432 \
  pgvector/pgvector:pg16

-- Create table with vector column
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    embedding vector(1536),  -- Match your embedding model's dimensions
    metadata JSONB DEFAULT '{}',
    source VARCHAR(500),
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Metadata index for filtering
CREATE INDEX idx_documents_metadata ON documents USING GIN (metadata);
CREATE INDEX idx_documents_source ON documents (source);

-- HNSW index (recommended for most workloads)
-- Slower to build, faster queries, higher recall
CREATE INDEX idx_documents_embedding_hnsw 
ON documents 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- IVFFlat index (for very large datasets)
-- Faster to build, requires training data, lower recall
CREATE INDEX idx_documents_embedding_ivf 
ON documents 
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);  -- Rule of thumb: sqrt(num_rows) for lists

import psycopg2
import json
from openai import OpenAI

oai_client = OpenAI()

def get_connection():
    return psycopg2.connect(
        host="localhost",
        port=5432,
        dbname="ragdb",
        user="postgres",
        password="password",
    )

def insert_document(content: str, metadata: dict, source: str):
    """Embed and store a document."""
    # Generate embedding
    response = oai_client.embeddings.create(
        model="text-embedding-3-small",
        input=content,
    )
    embedding = response.data[0].embedding
    
    conn = get_connection()
    cur = conn.cursor()
    
    cur.execute(
        """
        INSERT INTO documents (content, embedding, metadata, source)
        VALUES (%s, %s::vector, %s, %s)
        RETURNING id
        """,
        (content, str(embedding), json.dumps(metadata), source),
    )
    
    doc_id = cur.fetchone()[0]
    conn.commit()
    cur.close()
    conn.close()
    
    return doc_id


def search_documents(
    query: str,
    n_results: int = 5,
    source_filter: str = None,
    metadata_filter: dict = None,
) -> list[dict]:
    """Search for similar documents using cosine similarity."""
    # Embed the query
    response = oai_client.embeddings.create(
        model="text-embedding-3-small",
        input=query,
    )
    query_embedding = response.data[0].embedding
    
    conn = get_connection()
    cur = conn.cursor()
    
    # Build query with optional filters
    sql = """
        SELECT id, content, metadata, source,
               1 - (embedding <=> %s::vector) as similarity
        FROM documents
        WHERE 1=1
    """
    params = [str(query_embedding)]
    
    if source_filter:
        sql += " AND source = %s"
        params.append(source_filter)
    
    if metadata_filter:
        for key, value in metadata_filter.items():
            sql += f" AND metadata->>'{key}' = %s"
            params.append(str(value))
    
    sql += " ORDER BY embedding <=> %s::vector LIMIT %s"
    params.extend([str(query_embedding), n_results])
    
    cur.execute(sql, params)
    
    results = []
    for row in cur.fetchall():
        results.append({
            "id": row[0],
            "content": row[1],
            "metadata": row[2],
            "source": row[3],
            "similarity": float(row[4]),
        })
    
    cur.close()
    conn.close()
    
    return results


# Usage
doc_id = insert_document(
    content="Our refund policy allows full refunds within 30 days of purchase.",
    metadata={"category": "policy", "version": "2.1"},
    source="policies/refund.md",
)

results = search_documents("Can I get my money back?", n_results=3)
for r in results:
    print(f"[{r['similarity']:.3f}] {r['content'][:100]}")

-- Increase HNSW search width for higher recall (slower queries)
SET hnsw.ef_search = 100;  -- Default is 40

-- Increase IVF probes for higher recall (slower queries)
SET ivfflat.probes = 10;  -- Default is 1

-- Check index usage
EXPLAIN ANALYZE
SELECT id, content, 1 - (embedding <=> '[0.1, 0.2, ...]'::vector) as similarity
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 5;

pip install pinecone

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")

# Create an index
pc.create_index(
    name="knowledge-base",
    dimension=1536,             # Must match your embedding model
    metric="cosine",            # cosine, euclidean, or dotproduct
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1",
    ),
)

# Connect to the index
index = pc.Index("knowledge-base")

from openai import OpenAI

oai_client = OpenAI()

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

# UPSERT — insert or update vectors
index.upsert(
    vectors=[
        {
            "id": "doc1",
            "values": embed("How to reset your password"),
            "metadata": {
                "content": "How to reset your password",
                "category": "account",
                "source": "faq.md",
            },
        },
        {
            "id": "doc2",
            "values": embed("Billing FAQ and payment methods"),
            "metadata": {
                "content": "Billing FAQ and payment methods",
                "category": "billing",
                "source": "faq.md",
            },
        },
    ],
    namespace="production",  # Namespaces isolate data within an index
)

# QUERY — similarity search
results = index.query(
    vector=embed("forgot my login"),
    top_k=5,
    namespace="production",
    include_metadata=True,
    filter={
        "category": {"$in": ["account", "technical"]},
    },
)

for match in results.matches:
    print(f"[{match.score:.3f}] {match.metadata['content'][:100]}")

# FETCH — get by ID
fetched = index.fetch(ids=["doc1"], namespace="production")

# DELETE
index.delete(ids=["doc2"], namespace="production")

# DELETE by filter
index.delete(
    filter={"category": "billing"},
    namespace="production",
)

# Index statistics
stats = index.describe_index_stats()
print(f"Total vectors: {stats.total_vector_count}")
print(f"Namespaces: {stats.namespaces}")

def batch_upsert(
    documents: list[dict],
    namespace: str = "production",
    batch_size: int = 100,
):
    """Upsert documents in batches with error handling."""
    import time
    
    total = len(documents)
    
    for i in range(0, total, batch_size):
        batch = documents[i:i + batch_size]
        
        # Embed batch
        texts = [doc["content"] for doc in batch]
        response = oai_client.embeddings.create(
            model="text-embedding-3-small",
            input=texts,
        )
        
        # Build vectors
        vectors = []
        for j, doc in enumerate(batch):
            vectors.append({
                "id": doc["id"],
                "values": response.data[j].embedding,
                "metadata": {
                    "content": doc["content"][:1000],  # Pinecone metadata limit: 40KB
                    **doc.get("metadata", {}),
                },
            })
        
        # Upsert with retry
        for attempt in range(3):
            try:
                index.upsert(vectors=vectors, namespace=namespace)
                break
            except Exception as e:
                if attempt == 2:
                    raise
                time.sleep(2 ** attempt)
        
        print(f"Upserted {min(i + batch_size, total)}/{total}")

# Docker
docker run -d \
  --name qdrant \
  -p 6333:6333 \
  -p 6334:6334 \
  -v $(pwd)/qdrant_storage:/qdrant/storage:z \
  qdrant/qdrant

pip install qdrant-client

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue,
)

client = QdrantClient(host="localhost", port=6333)

# Create collection
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,
        distance=Distance.COSINE,
    ),
)

# Insert vectors
from openai import OpenAI
oai = OpenAI()

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

client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=embed("Password reset instructions"),
            payload={
                "content": "Password reset instructions",
                "category": "account",
                "source": "faq.md",
            },
        ),
        PointStruct(
            id=2,
            vector=embed("Billing and payment FAQ"),
            payload={
                "content": "Billing and payment FAQ",
                "category": "billing",
                "source": "faq.md",
            },
        ),
    ],
)

# Search with filter
results = client.query_points(
    collection_name="documents",
    query=embed("I can't log in"),
    limit=5,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="category",
                match=MatchValue(value="account"),
            ),
        ]
    ),
)

for point in results.points:
    print(f"[{point.score:.3f}] {point.payload['content']}")

import chromadb
import psycopg2
import json

def migrate_chroma_to_pgvector(
    chroma_path: str,
    collection_name: str,
    pg_connection_string: str,
):
    """Migrate all vectors from ChromaDB to pgvector."""
    
    # Connect to ChromaDB
    chroma = chromadb.PersistentClient(path=chroma_path)
    collection = chroma.get_collection(collection_name)
    
    # Get all data
    total = collection.count()
    print(f"Migrating {total} vectors from ChromaDB to pgvector...")
    
    batch_size = 1000
    conn = psycopg2.connect(pg_connection_string)
    cur = conn.cursor()
    
    for offset in range(0, total, batch_size):
        # Fetch batch from ChromaDB
        results = collection.get(
            limit=batch_size,
            offset=offset,
            include=["documents", "metadatas", "embeddings"],
        )
        
        # Insert into pgvector
        for i in range(len(results["ids"])):
            cur.execute(
                """
                INSERT INTO documents (content, embedding, metadata, source)
                VALUES (%s, %s::vector, %s, %s)
                ON CONFLICT DO NOTHING
                """,
                (
                    results["documents"][i],
                    str(results["embeddings"][i]),
                    json.dumps(results["metadatas"][i]),
                    results["metadatas"][i].get("source", "chromadb_migration"),
                ),
            )
        
        conn.commit()
        print(f"  Migrated {min(offset + batch_size, total)}/{total}")
    
    cur.close()
    conn.close()
    print("Migration complete. Now build the HNSW index:")
    print("  CREATE INDEX idx_embedding ON documents USING hnsw (embedding vector_cosine_ops);")

import chromadb
from pinecone import Pinecone

def migrate_chroma_to_pinecone(
    chroma_path: str,
    collection_name: str,
    pinecone_api_key: str,
    pinecone_index_name: str,
    namespace: str = "default",
):
    """Migrate all vectors from ChromaDB to Pinecone."""
    
    chroma = chromadb.PersistentClient(path=chroma_path)
    collection = chroma.get_collection(collection_name)
    
    pc = Pinecone(api_key=pinecone_api_key)
    index = pc.Index(pinecone_index_name)
    
    total = collection.count()
    batch_size = 100  # Pinecone recommends smaller batches
    
    for offset in range(0, total, batch_size):
        results = collection.get(
            limit=batch_size,
            offset=offset,
            include=["documents", "metadatas", "embeddings"],
        )
        
        vectors = []
        for i in range(len(results["ids"])):
            metadata = results["metadatas"][i] or {}
            metadata["content"] = (results["documents"][i] or "")[:1000]
            
            vectors.append({
                "id": results["ids"][i],
                "values": results["embeddings"][i],
                "metadata": metadata,
            })
        
        index.upsert(vectors=vectors, namespace=namespace)
        print(f"  Migrated {min(offset + batch_size, total)}/{total}")
    
    print("Migration complete.")

METRICS = {
    "query_latency_p50": "Should be < 50ms for good UX",
    "query_latency_p99": "Should be < 200ms",
    "insert_latency_p50": "Track for batch ingestion pipelines",
    "index_size_bytes": "Monitor for capacity planning",
    "total_vectors": "Track growth rate",
    "recall_at_k": "Measure with a ground truth test set",
    "filter_overhead_ms": "Time added by metadata filters",
    "error_rate": "Failed queries / total queries",
}

Are you prototyping or in early development?
├── YES → ChromaDB (zero setup, iterate fast)
└── NO → Do you already run PostgreSQL?
    ├── YES → Will you exceed 5M vectors?
    │   ├── YES → Consider Pinecone or Qdrant
    │   └── NO → pgvector (no new infrastructure)
    └── NO → Is operational simplicity your top priority?
        ├── YES → Pinecone (fully managed, zero ops)
        └── NO → Do you need self-hosted for compliance/privacy?
            ├── YES → Qdrant (best self-hosted performance)
            └── NO → Pinecone (default choice when not on Postgres)

from abc import ABC, abstractmethod

class VectorStore(ABC):
    """Unified interface for vector databases."""
    
    @abstractmethod
    def add(self, ids: list[str], texts: list[str], metadatas: list[dict]) -> None:
        """Add documents to the store."""
        pass
    
    @abstractmethod
    def search(self, query: str, n_results: int = 5, where: dict = None) -> list[dict]:
        """Search for similar documents."""
        pass
    
    @abstractmethod
    def delete(self, ids: list[str]) -> None:
        """Delete documents by ID."""
        pass
    
    @abstractmethod
    def count(self) -> int:
        """Return total document count."""
        pass


class ChromaVectorStore(VectorStore):
    def __init__(self, path: str, collection_name: str):
        import chromadb
        self.client = chromadb.PersistentClient(path=path)
        self.collection = self.client.get_or_create_collection(collection_name)
    
    def add(self, ids, texts, metadatas):
        self.collection.add(ids=ids, documents=texts, metadatas=metadatas)
    
    def search(self, query, n_results=5, where=None):
        params = {"query_texts": [query], "n_results": n_results}
        if where:
            params["where"] = where
        results = self.collection.query(**params)
        return [
            {"content": results["documents"][0][i], "metadata": results["metadatas"][0][i],
             "score": 1 - results["distances"][0][i]}
            for i in range(len(results["documents"][0]))
        ]
    
    def delete(self, ids):
        self.collection.delete(ids=ids)
    
    def count(self):
        return self.collection.count()


class PgVectorStore(VectorStore):
    def __init__(self, connection_string: str):
        import psycopg2
        self.conn_string = connection_string
        self._embed_client = OpenAI()
    
    def _embed(self, text: str) -> list[float]:
        resp = self._embed_client.embeddings.create(
            model="text-embedding-3-small", input=text
        )
        return resp.data[0].embedding
    
    def _get_conn(self):
        import psycopg2
        return psycopg2.connect(self.conn_string)
    
    def add(self, ids, texts, metadatas):
        conn = self._get_conn()
        cur = conn.cursor()
        for text_id, text, meta in zip(ids, texts, metadatas):
            embedding = self._embed(text)
            cur.execute(
                "INSERT INTO documents (id, content, embedding, metadata) "
                "VALUES (%s, %s, %s::vector, %s) ON CONFLICT (id) DO UPDATE "
                "SET content = EXCLUDED.content, embedding = EXCLUDED.embedding",
                (text_id, text, str(embedding), json.dumps(meta)),
            )
        conn.commit()
        cur.close()
        conn.close()
    
    def search(self, query, n_results=5, where=None):
        embedding = self._embed(query)
        conn = self._get_conn()
        cur = conn.cursor()
        sql = (
            "SELECT content, metadata, 1 - (embedding <=> %s::vector) as score "
            "FROM documents ORDER BY embedding <=> %s::vector LIMIT %s"
        )
        cur.execute(sql, (str(embedding), str(embedding), n_results))
        results = [
            {"content": row[0], "metadata": row[1], "score": float(row[2])}
            for row in cur.fetchall()
        ]
        cur.close()
        conn.close()
        return results
    
    def delete(self, ids):
        conn = self._get_conn()
        cur = conn.cursor()
        cur.execute("DELETE FROM documents WHERE id = ANY(%s)", (ids,))
        conn.commit()
        cur.close()
        conn.close()
    
    def count(self):
        conn = self._get_conn()
        cur = conn.cursor()
        cur.execute("SELECT COUNT(*) FROM documents")
        result = cur.fetchone()[0]
        cur.close()
        conn.close()
        return result


# Usage — swap implementations without changing RAG code
# store = ChromaVectorStore("./chroma_db", "knowledge_base")
# store = PgVectorStore("postgresql://user:pass@localhost/ragdb")

# Your RAG pipeline doesn't care which one it is
# results = store.search("How do I reset my password?", n_results=5)