# vectordb/indexes.py from enum import Enum from dataclasses import dataclass from typing: List class IndexType(Enum): FLAT = "flat" # Exact search, O(n)  IVF_FLAT = "ivf_flat" # Inverted file, balanced  IVF_PQ = "ivf_pq" # Product quantization, compressed  HNSW = "hnsw" # Hierarchical NSW, fast approximate  ANNOY = "annoy" # Trees, good for static data  SCANN = "scann" # Google's ScaNN, high recall  DISKANN = "diskann" # Disk-based, large scale  @dataclass class IndexConfig: index_type: IndexType params: dict # HNSW params  m: int = 16 # Connections per layer  ef_construct: int = 100 # Build-time search width  ef_search: int = 50 # Query-time search width  # IVF params  nlist: int = 100 # Number of clusters  nprobe: int = 10 # Clusters to search  # PQ params  m_pq: int = 8 # Subquantizers  nbits: int = 8 # Bits per subquantizer  # Recommendations by use case INDEX_RECOMMENDATIONS = { "small_exact": IndexConfig(IndexType.FLAT, {}), "medium_fast": IndexConfig(IndexType.HNSW, {"m": 16, "ef_construct": 200}), "large_scale": IndexConfig(IndexType.IVF_PQ, {"nlist": 4096, "m_pq": 16}), "massive": IndexConfig(IndexType.DISKANN, {}), } 

# vectordb/metrics.py class DistanceMetric(Enum): COSINE = "cosine" # Normalized dot product, most common  EUCLIDEAN = "euclidean" # L2 distance  DOT = "dot" # Inner product (requires normalized vectors)  MANHATTAN = "manhattan" # L1 distance  HAMMING = "hamming" # Binary vectors  JACCARD = "jaccard" # Set similarity  # Choose metric based on embedding model METRIC_BY_MODEL = { "text-embedding-3-large": DistanceMetric.COSINE, "text-embedding-3-small": DistanceMetric.COSINE, "bge-large-en-v1.5": DistanceMetric.COSINE, "e5-large-v2": DistanceMetric.COSINE, "instructor-xl": DistanceMetric.COSINE, # Some models work with dot product (vectors pre-normalized) } def normalize_vectors(vectors: List[List[float]]) -> List[List[float]]: """Normalize for cosine similarity with dot product index.""" import numpy as np arr = np.array(vectors) norms = np.linalg.norm(arr, axis=1, keepdims=True) return (arr / (norms + 1e-8)).tolist() 

# vectordb/qdrant_client.py from qdrant_client import QdrantClient from qdrant_client.models import * from typing: List, Dict, Optional import numpy as np class QdrantManager: def __init__(self, url: str = "http://localhost:6333", api_key: str = None): self.client = QdrantClient(url=url, api_key=api_key) def create_collection( self, name: str, vector_size: int, distance: Distance = Distance.COSINE, hnsw_config: HnswConfigDiff = None, quantization_config: QuantizationConfig = None, on_disk_payload: bool = True ): """Create optimized collection.""" vectors_config = VectorParams( size=vector_size, distance=distance, hnsw_config=hnsw_config or HnswConfigDiff( m=16, ef_construct=200, full_scan_threshold=10000 ), quantization_config=quantization_config or ScalarQuantizationConfig( type=ScalarQuantizationType.INT8, quantile=0.99, always_ram=True ), on_disk=on_disk_payload ) self.client.create_collection( collection_name=name, vectors_config=vectors_config, optimizers_config=OptimizersConfigDiff( default_segment_number=2, max_segment_size=20000, memmap_threshold=50000, indexing_threshold=20000, flush_interval_sec=5 ), wal_config=WalConfigDiff(wal_capacity_mb=32), ) def upsert_batch( self, collection_name: str, points: List[PointStruct], batch_size: int = 1000 ) -> UpdateResult: """Efficient batch upsert.""" for i in range(0, len(points), batch_size): batch = points[i:i + batch_size] self.client.upsert( collection_name=collection_name, points=batch, wait=True ) return UpdateResult(status=UpdateStatus.COMPLETED) def search( self, collection_name: str, query_vector: List[float], limit: int = 10, score_threshold: float = None, query_filter: Filter = None, search_params: SearchParams = None ) -> List[ScoredPoint]: """High-performance search with filtering.""" return self.client.search( collection_name=collection_name, query_vector=query_vector, limit=limit, score_threshold=score_threshold, query_filter=query_filter, search_params=search_params or SearchParams( hnsw_ef=128, exact=False ) ) def hybrid_search( self, collection_name: str, query_vector: List[float], query_text: str, limit: int = 10, vector_weight: float = 0.7, text_weight: float = 0.3 ) -> List[ScoredPoint]: """Hybrid vector + keyword search (requires sparse vectors).""" # Requires sparse vector index setup  return self.client.query_points( collection_name=collection_name, prefetch=[ Prefetch( query=query_vector, using="dense", limit=limit * 2 ), Prefetch( query=query_text, using="sparse", limit=limit * 2 ) ], query=FusionQuery(fusion=Fusion.RRF), limit=limit ) 

# vectordb/qdrant_tuning.py QDRANT_TUNING_GUIDE = { "memory_optimization": { "quantization": "Use scalar (INT8) or binary quantization for 4-8x memory reduction", "on_disk_payload": "Store payload on disk, keep only vectors in RAM", "memmap_threshold": "Enable memmap for segments > 50k vectors", "hnsw_m": "Lower m (8-16) for memory, higher (32-64) for recall", }, "search_speed": { "hnsw_ef_search": "Increase for recall (50-256), decrease for speed", "exact_search": "Use exact=True for small collections (<10k)", "parallel_queries": "Batch multiple queries in single request", "filter_optimization": "Put selective filters first, use payload indexes", }, "indexing_speed": { "max_segment_size": "Larger segments = faster indexing, more memory", "indexing_threshold": "Delay indexing until segment reaches threshold", "default_segment_number": "Match CPU cores for parallel indexing", "flush_interval_sec": "Balance durability vs performance", }, "scaling": { "sharding": "Auto-shard by setting shard_number > 1", "replication": "Set replication_factor >= 2 for HA", "distributed": "Use Qdrant Cluster for multi-node", } } 

# vectordb/pinecone_client.py from pinecone import Pinecone, ServerlessSpec from typing: List, Dict, Optional import time class PineconeManager: def __init__(self, api_key: str): self.pc = Pinecone(api_key=api_key) def create_index( self, name: str, dimension: int, metric: str = "cosine", spec: ServerlessSpec = None, deletion_protection: str = "disabled" ): """Create serverless index.""" if spec is None: spec = ServerlessSpec(cloud="aws", region="us-east-1") self.pc.create_index( name=name, dimension=dimension, metric=metric, spec=spec, deletion_protection=deletion_protection ) # Wait for ready  while not self.pc.describe_index(name).status["ready"]: time.sleep(1) def upsert( self, index_name: str, vectors: List[Dict], namespace: str = "", batch_size: int = 100 ): """Batch upsert with automatic retry.""" index = self.pc.Index(index_name) for i in range(0, len(vectors), batch_size): batch = vectors[i:i + batch_size] index.upsert(vectors=batch, namespace=namespace) def query( self, index_name: str, vector: List[float], top_k: int = 10, namespace: str = "", filter: Dict = None, include_metadata: bool = True, include_values: bool = False ): index = self.pc.Index(index_name) return index.query( vector=vector, top_k=top_k, namespace=namespace, filter=filter, include_metadata=include_metadata, include_values=include_values ) def query_by_id( self, index_name: str, id: str, top_k: int = 10, namespace: str = "" ): """Query using existing vector ID (efficient for recommendations).""" index = self.pc.Index(index_name) return index.query( id=id, top_k=top_k, namespace=namespace ) 

# vectordb/pinecone_costs.py PINECONE_COSTS = { "serverless": { "storage_gb_month": 0.33, # $0.33/GB/month  "read_units": 0.000065, # Per 1000 read units  "write_units": 0.000013, # Per 1000 write units  }, "pod_based": { "p1_x1": 0.096, # $0.096/hour  "p1_x2": 0.192, "p1_x4": 0.384, "p1_x8": 0.768, "p2_x1": 0.346, "p2_x2": 0.692, "p2_x4": 1.384, "p2_x8": 2.768, } } def estimate_monthly_cost( index_type: str, vector_count: int, dimension: int, queries_per_month: int, writes_per_month: int, pod_type: str = "p1_x1" ) -> dict: """Estimate Pinecone costs.""" # Storage: ~4 bytes per dimension per vector (float32)  storage_gb = (vector_count * dimension * 4) / (1024 ** 3) if index_type == "serverless": storage_cost = storage_gb * PINECONE_COSTS["serverless"]["storage_gb_month"] read_cost = (queries_per_month / 1000) * PINECONE_COSTS["serverless"]["read_units"] write_cost = (writes_per_month / 1000) * PINECONE_COSTS["serverless"]["write_units"] total = storage_cost + read_cost + write_cost else: pod_cost = PINECONE_COSTS["pod_based"][pod_type] * 24 * 30 total = pod_cost return { "storage_gb": storage_gb, "monthly_cost_usd": total, "breakdown": { "storage": storage_cost if index_type == "serverless" else pod_cost, "reads": read_cost if index_type == "serverless" else 0, "writes": write_cost if index_type == "serverless" else 0 } } 

# vectordb/weaviate_client.py import weaviate from weaviate.classes.init import Auth from weaviate.classes.config import Configure, Property, DataType from typing: List, Dict, Optional class WeaviateManager: def __init__( self, url: str = "http://localhost:8080", auth_credentials: Auth = None ): self.client = weaviate.connect_to_custom( http_hostSource：https://dev.to/ipsphil_digital/vector-databases-for-ai-qdrant-pinecone-weaviate-milvus-compared-2i56
My favorites