Vector Database Guide: Pinecone, Weaviate, Chroma, and pgvector Compared

# pip install chromadb

import chromadb
from chromadb.utils import embedding_functions

# In-memory client (lost when process ends)
client = chromadb.Client()

# Persistent client
client = chromadb.PersistentClient(path="./chroma_db")

# Create collection with embedding function
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="your-key",
    model_name="text-embedding-3-small"
)

collection = client.get_or_create_collection(
    name="documents",
    embedding_function=openai_ef,
    metadata={"hnsw:space": "cosine"}  # Distance metric
)

# Add documents (Chroma handles embedding)
collection.add(
    documents=[
        "The quick brown fox jumps over the lazy dog.",
        "Machine learning is a subset of artificial intelligence.",
        "Python is a versatile programming language.",
    ],
    metadatas=[
        {"source": "sample.txt", "category": "text"},
        {"source": "ml_intro.txt", "category": "ai"},
        {"source": "python_intro.txt", "category": "programming"},
    ],
    ids=["doc1", "doc2", "doc3"]
)

# Query
results = collection.query(
    query_texts=["What is AI?"],
    n_results=2,
    where={"category": "ai"}  # Metadata filtering
)

for doc, meta, distance in zip(
    results["documents"][0],
    results["metadatas"][0],
    results["distances"][0]
):
    print(f"Distance: {distance:.3f} | Source: {meta['source']}")
    print(f"Content: {doc[:100]}\n")

# Update and delete
collection.update(ids=["doc1"], documents=["Updated text content"])
collection.delete(ids=["doc1"])
print(f"Collection count: {collection.count()}")

# pip install pinecone-client

from pinecone import Pinecone, ServerlessSpec
import numpy as np

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

# Create index
if "my-index" not in [idx.name for idx in pc.list_indexes()]:
    pc.create_index(
        name="my-index",
        dimension=1536,  # Must match your embedding model
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

index = pc.Index("my-index")

# Upsert vectors (with metadata)
from openai import OpenAI

client = OpenAI()

def embed(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [item.embedding for item in response.data]

documents = [
    {"id": "doc1", "text": "Introduction to machine learning concepts."},
    {"id": "doc2", "text": "Python data structures and algorithms."},
    {"id": "doc3", "text": "Deep learning with PyTorch tutorial."},
]

texts = [d["text"] for d in documents]
embeddings = embed(texts)

# Batch upsert
vectors = [
    {
        "id": doc["id"],
        "values": emb,
        "metadata": {"text": doc["text"], "category": "tech"}
    }
    for doc, emb in zip(documents, embeddings)
]

index.upsert(vectors=vectors, namespace="production")

# Query
query_embedding = embed(["how does machine learning work?"])[0]

results = index.query(
    namespace="production",
    vector=query_embedding,
    top_k=3,
    include_values=False,
    include_metadata=True,
    filter={"category": {"$eq": "tech"}}  # Metadata filter
)

for match in results.matches:
    print(f"Score: {match.score:.3f} | ID: {match.id}")
    print(f"Text: {match.metadata['text']}\n")

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

# pip install psycopg2-binary pgvector

import psycopg2
from pgvector.psycopg2 import register_vector
import numpy as np

# Connect to PostgreSQL with pgvector
conn = psycopg2.connect("postgresql://user:password@localhost/mydb")
register_vector(conn)

cur = conn.cursor()

# Enable extension and create table
cur.execute("CREATE EXTENSION IF NOT EXISTS vector")

cur.execute("""
    CREATE TABLE IF NOT EXISTS documents (
        id SERIAL PRIMARY KEY,
        content TEXT NOT NULL,
        category VARCHAR(50),
        embedding vector(1536)  -- Match your model's dimensions
    )
""")

# Create HNSW index for fast approximate search
cur.execute("""
    CREATE INDEX IF NOT EXISTS documents_embedding_idx
    ON documents USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64)
""")

conn.commit()

# Insert with embeddings
def insert_document(content: str, category: str, embedding: list[float]):
    cur.execute(
        "INSERT INTO documents (content, category, embedding) VALUES (%s, %s, %s)",
        (content, category, np.array(embedding))
    )
    conn.commit()

# Similarity search
def search(query_embedding: list[float], top_k: int = 5, category: str | None = None):
    if category:
        cur.execute("""
            SELECT content, category, 1 - (embedding <=> %s) AS similarity
            FROM documents
            WHERE category = %s
            ORDER BY embedding <=> %s
            LIMIT %s
        """, (np.array(query_embedding), category, np.array(query_embedding), top_k))
    else:
        cur.execute("""
            SELECT content, category, 1 - (embedding <=> %s) AS similarity
            FROM documents
            ORDER BY embedding <=> %s
            LIMIT %s
        """, (np.array(query_embedding), np.array(query_embedding), top_k))
    
    return cur.fetchall()

# Hybrid search (semantic + keyword)
def hybrid_search(query: str, query_embedding: list[float], top_k: int = 5):
    cur.execute("""
        SELECT content, category,
               ts_rank(to_tsvector(content), plainto_tsquery(%s)) AS keyword_score,
               1 - (embedding <=> %s) AS semantic_score,
               -- Combine: 50% keyword + 50% semantic
               (ts_rank(to_tsvector(content), plainto_tsquery(%s)) * 0.5 +
                (1 - (embedding <=> %s)) * 0.5) AS combined_score
        FROM documents
        WHERE to_tsvector(content) @@ plainto_tsquery(%s)
           OR (1 - (embedding <=> %s)) > 0.7
        ORDER BY combined_score DESC
        LIMIT %s
    """, (query, np.array(query_embedding), query, np.array(query_embedding), 
          query, np.array(query_embedding), top_k))
    
    return cur.fetchall()

# pip install qdrant-client

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

# Connect to local Qdrant (docker run -p 6333:6333 qdrant/qdrant)
client = QdrantClient(host="localhost", port=6333)

# Or cloud
# client = QdrantClient(url="https://your-cluster.qdrant.io", api_key="your-key")

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

# Upsert points
points = [
    PointStruct(
        id=1,
        vector=[0.1] * 1536,  # Your actual embedding
        payload={"text": "Article about ML", "category": "ai", "views": 1500}
    ),
    PointStruct(
        id=2,
        vector=[0.2] * 1536,
        payload={"text": "Python tutorial", "category": "programming", "views": 3000}
    ),
]

client.upsert(collection_name="articles", points=points)

# Search with filters
results = client.search(
    collection_name="articles",
    query_vector=[0.1] * 1536,
    limit=5,
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(value="ai")),
            FieldCondition(key="views", range={"gte": 1000})
        ]
    ),
    with_payload=True
)

for r in results:
    print(f"Score: {r.score:.3f} | Text: {r.payload['text']}")