Vector Databases Explained — How a 23% Recall Drop in Production Cost Us $40k in One Night

import numpy as np
import chromadb
from chromadb.utils import embedding_functions

# Simulate 100k 1536-dim vectors (like text-embedding-3-small)
np.random.seed(42)
num_vectors = 100_000
dim = 1536
raw_vectors = np.random.randn(num_vectors, dim).astype(np.float32)

# Normalize to unit length — critical for cosine similarity
norms = np.linalg.norm(raw_vectors, axis=1, keepdims=True)
normalized_vectors = raw_vectors / norms  # shape: (100000, 1536)

# Initialize ChromaDB with persistent storage
client = chromadb.PersistentClient(path="/tmp/chroma_demo")
collection = client.create_collection(
    name="hnsw_demo",
    metadata={"hnsw:space": "cosine", "hnsw:construction_ef": 200, "hnsw:M": 16}
)

# Add vectors in batches to avoid OOM
batch_size = 10000
for i in range(0, num_vectors, batch_size):
    batch = normalized_vectors[i:i+batch_size]
    ids = [f"vec_{j}" for j in range(i, i+len(batch))]
    collection.add(
        embeddings=batch.tolist(),
        ids=ids,
        metadatas=[{"index": j} for j in range(i, i+len(batch))]
    )

# Query with a random vector
query_vec = normalized_vectors[0].tolist()
results = collection.query(
    query_embeddings=[query_vec],
    n_results=5
)
print("Top 5 results:", results['ids'][0])
print("Distances:", results['distances'][0])
# Note: distances are cosine distances (0 = identical, 2 = opposite)
# If you see distances > 1.0, something is wrong with normalization

import os
import time
import numpy as np
from typing import List, Dict
from openai import OpenAI
import chromadb
from chromadb.config import Settings

# Initialize clients
openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
chroma_client = chromadb.PersistentClient(
    path="/data/chroma",
    settings=Settings(anonymized_telemetry=False)  # disable telemetry in prod
)

def embed_texts(texts: List[str], model: str = "text-embedding-3-small") -> np.ndarray:
    """Get embeddings and normalize them."""
    response = openai_client.embeddings.create(model=model, input=texts)
    embeddings = np.array([d.embedding for d in response.data], dtype=np.float32)
    norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
    return embeddings / norms  # critical: normalize

def create_production_collection(name: str, index_params: Dict = None):
    """Create a collection with production-ready index settings."""
    if index_params is None:
        index_params = {"hnsw:space": "cosine", "hnsw:construction_ef": 200, "hnsw:M": 16}
    return chroma_client.create_collection(
        name=name,
        metadata=index_params
    )

def batch_ingest(collection, texts: List[str], ids: List[str], metadatas: List[Dict], batch_size: int = 100):
    """Ingest texts in batches with retries."""
    for i in range(0, len(texts), batch_size):
        batch_texts = texts[i:i+batch_size]
        batch_ids = ids[i:i+batch_size]
        batch_metadatas = metadatas[i:i+batch_size]
        
        # Embed with retry
        for attempt in range(3):
            try:
                embeddings = embed_texts(batch_texts)
                break
            except Exception as e:
                print(f"Embedding attempt {attempt+1} failed: {e}")
                time.sleep(2 ** attempt)
        else:
            raise RuntimeError(f"Failed to embed batch after 3 attempts")
        
        # Insert into ChromaDB
        collection.add(
            embeddings=embeddings.tolist(),
            ids=batch_ids,
            metadatas=batch_metadatas
        )
        print(f"Ingested batch {i//batch_size + 1}: {len(batch_ids)} vectors")

# Usage example
if __name__ == "__main__":
    texts = ["Product A description", "Product B description"]
    ids = ["prod_a", "prod_b"]
    metadatas = [{"category": "electronics"}, {"category": "books"}]
    
    col = create_production_collection("products_v2")
    batch_ingest(col, texts, ids, metadatas)
    print("Ingestion complete. Run `collection.count()` to verify.")

import numpy as np
from sklearn.neighbors import NearestNeighbors

# Small dataset: 5k vectors, 1536 dims
np.random.seed(42)
X = np.random.randn(5000, 1536).astype(np.float32)
norms = np.linalg.norm(X, axis=1, keepdims=True)
X = X / norms  # normalize

# Brute-force nearest neighbors
nn = NearestNeighbors(n_neighbors=5, metric='cosine')
nn.fit(X)

# Query
query = X[0].reshape(1, -1)
distances, indices = nn.kneighbors(query)
print("Nearest neighbors (indices):", indices[0])
print("Distances:", distances[0])
# This takes ~2ms for 5k vectors. No need for a vector DB.

# For comparison, ChromaDB with HNSW on the same data takes ~1ms,
# but you pay for infrastructure and operational complexity.
# Rule of thumb: <10k vectors -> sklearn, >100k vectors -> vector DB

import chromadb
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict

class ShardedVectorSearch:
    def __init__(self, shard_names: List[str], base_path: str = "/data/chroma"):
        self.shard_names = shard_names
        self.clients = {}
        for name in shard_names:
            client = chromadb.PersistentClient(path=f"{base_path}/{name}")
            self.clients[name] = client.get_or_create_collection(name)
    
    def query(self, query_embedding: List[float], n_results: int = 10, n_per_shard: int = 20) -> List[Dict]:
        """Query all shards in parallel and merge results."""
        all_results = []
        with ThreadPoolExecutor(max_workers=len(self.shard_names)) as executor:
            futures = {
                executor.submit(
                    col.query, query_embeddings=[query_embedding], n_results=n_per_shard
                ): name
                for name, col in self.clients.items()
            }
            for future in as_completed(futures):
                results = future.result()
                for i in range(len(results['ids'][0])):
                    all_results.append({
                        'id': results['ids'][0][i],
                        'distance': results['distances'][0][i],
                        'metadata': results['metadatas'][0][i] if results['metadatas'] else {},
                        'shard': futures[future]
                    })
        # Sort by distance and return top n_results
        all_results.sort(key=lambda x: x['distance'])
        return all_results[:n_results]

# Usage
searcher = ShardedVectorSearch(shard_names=["products_shard_0", "products_shard_1", "products_shard_2"])
results = searcher.query(query_embedding=[0.1]*1536, n_results=10)
print("Top 10 results across shards:", results)

import numpy as np

# Mistake 1: Not normalizing
raw_vec = np.random.randn(1536).astype(np.float32)
# Wrong: using raw_vec directly
# Correct:
normalized = raw_vec / np.linalg.norm(raw_vec)

# Mistake 2: Wrong distance metric
# If you use L2 on unnormalized vectors, magnitude dominates.
# Example: two vectors pointing in same direction but different magnitudes
v1 = np.array([1.0, 0.0])
v2 = np.array([10.0, 0.0])
l2_dist = np.linalg.norm(v1 - v2)  # 9.0 — far apart even though same direction
cos_sim = np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))  # 1.0 — identical direction
print(f"L2: {l2_dist:.2f}, Cosine: {cos_sim:.2f}")

# Mistake 3: Index rebuild blocking
# Never do this on serving node:
# collection.create_index()  # blocks for 45 minutes
# Instead, rebuild on staging and swap.

# Mistake 4: Over-filtering
# Increase n_results to compensate for filter selectivity
results = collection.query(
    query_embeddings=[query_vec],
    n_results=100,  # not 10
    where={'category': 'electronics'}
)
# Then filter in application
filtered = [r for r in results['ids'][0] if r['price'] < 10]

# Mistake 5: Model version drift
# Pin the model in your code:
# openai_client.embeddings.create(model="text-embedding-3-small", ...)
# And log the model version in collection metadata:
# collection.metadata['embedding_model'] = 'text-embedding-3-small-2025-01'

import numpy as np
import faiss
import time

# Generate 1M 1536-dim vectors
np.random.seed(42)
d = 1536
nb = 1_000_000
x = np.random.randn(nb, d).astype(np.float32)
norms = np.linalg.norm(x, axis=1, keepdims=True)
x = x / norms

# Build FAISS index (HNSW)
index = faiss.IndexHNSWFlat(d, 32)  # M=32
index.hnsw.efConstruction = 200
print("Building FAISS index...")
t0 = time.time()
index.add(x)
print(f"FAISS index built in {time.time() - t0:.2f}s")

# Query
query = x[0].reshape(1, -1)
index.hnsw.efSearch = 100
t0 = time.time()
distances, indices = index.search(query, 10)
print(f"FAISS query time: {(time.time() - t0)*1000:.2f}ms")
print("Distances:", distances[0])

# Compare with ChromaDB (conceptual, not run here)
# ChromaDB with same data would take ~45 min to build index and ~5ms per query.
# FAISS builds in ~2 min and queries in ~1ms.
# But FAISS has no persistence, no filtering, no CRUD.
# Choose based on your needs.

import numpy as np
import chromadb
import time
from sklearn.neighbors import NearestNeighbors

def monitor_recall(collection, query_vectors: np.ndarray, k: int = 10):
    """Compare ANN results with brute-force to compute recall."""
    # Get all vectors from collection (assumes small set for monitoring)
    all_data = collection.get(include=['embeddings'])
    all_embeddings = np.array(all_data['embeddings'])
    all_ids = all_data['ids']
    
    # Brute-force search
    nn = NearestNeighbors(n_neighbors=k, metric='cosine')
    nn.fit(all_embeddings)
    
    recall_sum = 0
    for query in query_vectors:
        # ANN search
        ann_results = collection.query(
            query_embeddings=[query.tolist()],
            n_results=k
        )
        ann_ids = set(ann_results['ids'][0])
        
        # Brute-force search
        distances, indices = nn.kneighbors(query.reshape(1, -1))
        bf_ids = set([all_ids[i] for i in indices[0]])
        
        # Compute recall
        intersection = ann_ids.intersection(bf_ids)
        recall = len(intersection) / k
        recall_sum += recall
    
    avg_recall = recall_sum / len(query_vectors)
    print(f"Average recall@{k}: {avg_recall:.3f}")
    if avg_recall < 0.95:
        print("ALERT: Recall below 0.95! Check embedding normalization and model version.")
    return avg_recall

# Usage
client = chromadb.PersistentClient(path="/data/chroma")
collection = client.get_collection("products")
# Generate 10 random query vectors for monitoring
query_vecs = np.random.randn(10, 1536).astype(np.float32)
query_vecs = query_vecs / np.linalg.norm(query_vecs, axis=1, keepdims=True)
monitor_recall(collection, query_vecs)