Vector Databases for AI: Qdrant, Pinecone, Weaviate & Milvus Compared

Deep dive into vector databases — architecture, indexing, performance, scaling, and production deployment for AI applications.

Introduction

Vector databases are the backbone of modern AI applications — RAG, semantic search, recommendations, and embeddings storage. This guide compares the four major players with production insights.

Vector Database Landscape

Database	Type	License	Best For
Qdrant	Purpose-built	Apache 2.0	High performance, filtering, Rust
Pinecone	Managed service	Proprietary	Zero-ops, enterprise
Weaviate	Purpose-built	BSD-3	GraphQL, modules, hybrid search
Milvus	Purpose-built	Apache 2.0	Massive scale, Kubernetes native

Part 1: Core Concepts

Vector Index Types

# 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, {}),
}

Distance Metrics

# 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()

Part 2: Qdrant Deep Dive

Architecture & Features

# 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
        )

Qdrant Performance Tuning

# 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",
    }
}

Part 3: Pinecone Deep Dive

Serverless Architecture

# 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
        )

Pinecone Cost Optimization

# 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
        }
    }

Part 4: Weaviate Deep Dive

GraphQL & Modules

# 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_host=url.replace("http://", "").replace("https://", ""),
            http_port=8080,
            http_secure=False,
            auth_credentials=auth_credentials
        )

    def create_collection(
        self,
        name: str,
        properties: List[Property],
        vectorizer_config: Configure.Vectorizer = None,
        vector_index_config: Configure.VectorIndex = None
    ):
        """Create collection with vectorizer module."""

        if vectorizer_config is None:
            vectorizer_config = Configure.Vectorizer.text2vec_openai(
                model="text-embedding-3-small"
            )

        if vector_index_config is None:
            vector_index_config = Configure.VectorIndex.hnsw(
                distance_metric=VectorDistances.COSINE,
                quantizer=Configure.VectorIndex.Quantizer.bq()
            )

        self.client.collections.create(
            name=name,
            properties=properties,
            vectorizer_config=vectorizer_config,
            vector_index_config=vector_index_config
        )

    def hybrid_search(
        self,
        collection_name: str,
        query: str,
        vector: List[float] = None,
        alpha: float = 0.5,  # 0 = keyword, 1 = vector
        limit: int = 10,
        filters = None
    ):
        """Hybrid search with GraphQL-like flexibility."""
        collection = self.client.collections.get(collection_name)

        return collection.query.hybrid(
            query=query,
            vector=vector,
            alpha=alpha,
            limit=limit,
            filters=filters,
            return_metadata=MetadataQuery(
                score=True,
                explain_score=True
            )
        )

    def bm25_search(
        self,
        collection_name: str,
        query: str,
        limit: int = 10,
        properties: List[str] = None
    ):
        """Pure keyword search (BM25)."""
        collection = self.client.collections.get(collection_name)
        return collection.query.bm25(
            query=query,
            limit=limit,
            query_properties=properties
        )

Weaviate Modules

# vectordb/weaviate_modules.py
WEAVIATE_MODULES = {
    "text2vec_openai": {
        "description": "OpenAI embeddings",
        "models": ["text-embedding-3-small", "text-embedding-3-large", "ada-002"],
        "params": ["model", "dimensions", "base_url"]
    },
    "text2vec_cohere": {
        "description": "Cohere embeddings",
        "models": ["embed-english-v3.0", "embed-multilingual-v3.0"],
        "params": ["model", "truncate"]
    },
    "text2vec_huggingface": {
        "description": "Local HuggingFace models",
        "models": ["sentence-transformers/all-MiniLM-L6-v2", "BAAI/bge-large-en-v1.5"],
        "params": ["model", "endpoint"]
    },
    "generative_openai": {
        "description": "RAG generation with OpenAI",
        "models": ["gpt-4o", "gpt-4o-mini", "gpt-3.5-turbo"],
        "params": ["model", "temperature"]
    },
    "generative_cohere": {
        "description": "RAG generation with Cohere",
        "models": ["command-r-plus", "command-r"],
        "params": ["model", "temperature"]
    },
    "reranker_cohere": {
        "description": "Cohere reranking",
        "models": ["rerank-english-v3.0", "rerank-multilingual-v3.0"]
    },
}

# Multi-vector collections
MULTI_VECTOR_CONFIG = {
    "dense": Configure.Vectorizer.text2vec_openai(model="text-embedding-3-small"),
    "sparse": Configure.Vectorizer.text2vec_contextionary()  # BM25-like
}

Part 5: Milvus Deep Dive

Kubernetes-Native Architecture

# vectordb/milvus_client.py
from pymilvus import (
    MilvusClient, DataType, Function, FunctionType,
    AnnSearchRequest, RRFRanker, WeightedRanker
)
from typing: List, Dict, Optional

class MilvusManager:
    def __init__(self, uri: str = "http://localhost:19530", token: str = ""):
        self.client = MilvusClient(uri=uri, token=token)

    def create_collection(
        self,
        name: str,
        dimension: int,
        primary_field: str = "id",
        vector_field: str = "vector",
        enable_dynamic_field: bool = True,
        auto_id: bool = True
    ):
        schema = self.client.create_schema(
            auto_id=auto_id,
            enable_dynamic_field=enable_dynamic_field
        )

        schema.add_field(
            field_name=primary_field,
            datatype=DataType.INT64,
            is_primary=True
        )
        schema.add_field(
            field_name=vector_field,
            datatype=DataType.FLOAT_VECTOR,
            dim=dimension
        )
        schema.add_field(
            field_name="text",
            datatype=DataType.VARCHAR,
            max_length=65535
        )
        schema.add_field(
            field_name="metadata",
            datatype=DataType.JSON
        )

        # Add sparse vector field for hybrid search
        schema.add_field(
            field_name="sparse_vector",
            datatype=DataType.SPARSE_FLOAT_VECTOR
        )

        # Add BM25 function for sparse vectors
        bm25_function = Function(
            name="text_bm25",
            function_type=FunctionType.BM25,
            input_field_names=["text"],
            output_field_names=["sparse_vector"]
        )
        schema.add_function(bm25_function)

        index_params = self.client.prepare_index_params()
        index_params.add_index(
            field_name=vector_field,
            index_type="HNSW",
            metric_type="COSINE",
            params={"M": 16, "efConstruction": 200}
        )
        index_params.add_index(
            field_name="sparse_vector",
            index_type="SPARSE_INVERTED_INDEX",
            metric_type="BM25"
        )

        self.client.create_collection(
            collection_name=name,
            schema=schema,
            index_params=index_params
        )

    def hybrid_search(
        self,
        collection_name: str,
        dense_vector: List[float],
        sparse_query: str,
        limit: int = 10,
        dense_weight: float = 0.7,
        sparse_weight: float = 0.3,
        output_fields: List[str] = None
    ):
        """Hybrid dense + sparse search with re-ranking."""

        dense_req = AnnSearchRequest(
            data=[dense_vector],
            anns_field="vector",
            param={"metric_type": "COSINE", "params": {"ef": 128}},
            limit=limit * 2
        )

        sparse_req = AnnSearchRequest(
            data=[sparse_query],
            anns_field="sparse_vector",
            param={"metric_type": "BM25"},
            limit=limit * 2
        )

        # RRF fusion (default)
        rerank = RRFRanker(k=60)

        return self.client.hybrid_search(
            collection_name=collection_name,
            reqs=[dense_req, sparse_req],
            rerank=rerank,
            limit=limit,
            output_fields=output_fields or ["text", "metadata"]
        )

Milvus Scaling

# vectordb/milvus_scaling.py
MILVUS_SCALING = {
    "components": {
        "proxy": "Stateless, handles requests, auto-scale",
        "query_node": "Executes searches, scale with query load",
        "data_node": "Handles ingestion, scale with write load",
        "index_node": "Builds indexes, batch scale",
        "root_coord": "Metadata, single active",
        "query_coord": "Load balancing, single active",
        "data_coord": "Segment management, single active",
    },
    "deployment": {
        "standalone": "All-in-one, dev/test only",
        "cluster": "Separate components, production",
        "k8s_operator": "Milvus Operator for Kubernetes",
        "helm_chart": "Official Helm chart with all components",
    },
    "partitioning": {
        "partition_key": "Hash-based partitioning for multi-tenancy",
        "shard_num": "Number of shards per collection (default: 1-64)",
    }
}

Part 6: Production Comparison

Benchmark Results (Typical)

# vectordb/benchmarks.py
BENCHMARK_RESULTS = {
    "dataset": "1M vectors, 768-dim, COSINE",
    "queries": 1000,
    "results": {
        "Qdrant": {
            "p50_latency_ms": 12,
            "p99_latency_ms": 45,
            "recall_at_10": 0.98,
            "memory_gb": 3.2,
            "index_build_min": 8
        },
        "Pinecone": {
            "p50_latency_ms": 18,
            "p99_latency_ms": 55,
            "recall_at_10": 0.97,
            "memory_gb": "managed",
            "index_build_min": "managed"
        },
        "Weaviate": {
            "p50_latency_ms": 22,
            "p99_latency_ms": 68,
            "recall_at_10": 0.96,
            "memory_gb": 4.1,
            "index_build_min": 12
        },
        "Milvus": {
            "p50_latency_ms": 15,
            "p99_latency_ms": 52,
            "recall_at_10": 0.98,
            "memory_gb": 3.8,
            "index_build_min": 10
        }
    }
}

Feature Matrix

# vectordb/feature_matrix.py
FEATURE_MATRIX = {
    "Feature": ["Qdrant", "Pinecone", "Weaviate", "Milvus"],
    "Deployment": ["Self-hosted/Cloud", "Managed only", "Self-hosted/Cloud", "Self-hosted/K8s"],
    "License": ["Apache 2.0", "Proprietary", "BSD-3", "Apache 2.0"],
    "Hybrid Search": ["✅ Native", "✅ Sparse+Dense", "✅ Native", "✅ Native"],
    "Filtering": ["✅ Payload indexes", "✅ Metadata filter", "✅ GraphQL where", "✅ Scalar index"],
    "Multi-tenancy": ["✅ Sharding", "✅ Namespaces", "✅ Tenants", "✅ Partition keys"],
    "Backups": ["✅ Snapshots", "✅ Collections", "✅ Backups", "✅ Backups"],
    "Rollback": ["✅ Point-in-time", "❌", "✅", "✅"],
    "GraphQL": ["❌", "❌", "✅ Native", "❌"],
    "REST API": ["✅", "✅", "✅", "✅"],
    "gRPC": ["✅", "✅", "✅", "✅"],
    "Python SDK": ["✅ Excellent", "✅ Good", "✅ Good", "✅ Good"],
    "Rust Core": ["✅", "❌", "❌ (Go)", "❌ (C++)"],
    "Quantization": ["✅ Scalar/Binary", "❌", "✅ BQ/PQ", "✅ PQ/SQ"],
    "DiskANN": ["✅", "❌", "❌", "✅"],
    "RBAC": ["✅ Cloud only", "✅", "✅", "✅"],
    "Cost (1M vecs)": ["~$50/mo self-host", "~$200/mo serverless", "~$80/mo self-host", "~$100/mo K8s"],
}

Part 7: Migration & Interoperability

Universal Client Interface

# vectordb/universal.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing: List, Dict, Optional

@dataclass
class VectorRecord:
    id: str
    vector: List[float]
    payload: Dict
    sparse_vector: Dict[int, float] = None

@dataclass
class SearchResult:
    id: str
    score: float
    payload: Dict
    vector: List[float] = None

class VectorDBClient(ABC):
    @abstractmethod
    def create_collection(self, name: str, dimension: int, config: dict): pass

    @abstractmethod
    def upsert(self, collection: str, records: List[VectorRecord]): pass

    @abstractmethod
    def search(
        self,
        collection: str,
        query_vector: List[float],
        limit: int = 10,
        filter: Dict = None
    ) -> List[SearchResult]: pass

    @abstractmethod
    def hybrid_search(
        self,
        collection: str,
        dense_vector: List[float],
        sparse_query: str,
        limit: int = 10
    ) -> List[SearchResult]: pass

    @abstractmethod
    def delete(self, collection: str, ids: List[str]): pass

    @abstractmethod
    def get_stats(self, collection: str) -> dict: pass

# Factory
def create_vector_db(provider: str, **kwargs) -> VectorDBClient:
    providers = {
        "qdrant": QdrantClient,
        "pinecone": PineconeClient,
        "weaviate": WeaviateClient,
        "milvus": MilvusClient,
    }
    return providers[provider.lower()](**kwargs)

Migration Strategy

# vectordb/migration.py
async def migrate_collection(
    source: VectorDBClient,
    target: VectorDBClient,
    collection_name: str,
    batch_size: int = 1000
):
    """Migrate collection between vector databases."""

    # 1. Get source schema/stats
    stats = source.get_stats(collection_name)
    dimension = stats["dimension"]

    # 2. Create target collection
    target.create_collection(collection_name, dimension, {})

    # 3. Stream migrate in batches
    offset = 0
    total = stats["vector_count"]

    while offset < total:
        # Scroll source
        records = source.scroll(collection_name, offset, batch_size)

        # Transform if needed (dimension change, etc.)
        transformed = transform_records(records, dimension)

        # Upsert to target
        target.upsert(collection_name, transformed)

        offset += batch_size
        print(f"Migrated {min(offset, total)}/{total}")

    # 4. Verify
    target_stats = target.get_stats(collection_name)
    assert target_stats["vector_count"] == total

Part 8: Production Checklist

Pre-Production

# vectordb/production_checklist.py
PRODUCTION_CHECKLIST = {
    "capacity_planning": [
        "Estimate vector count growth (3-5x current)",
        "Calculate memory: vectors * dim * 4 bytes * 1.5 (overhead)",
        "Plan for quantization (INT8 = 4x savings)",
        "Reserve 30% headroom for indexing",
    ],
    "index_tuning": [
        "HNSW: m=16, ef_construct=200 for balanced",
        "ef_search=128 for production queries",
        "Enable quantization for >100k vectors",
        "Set appropriate nlist/nprobe for IVF",
    ],
    "reliability": [
        "Enable WAL (write-ahead log)",
        "Configure replication factor >= 2",
        "Set up automated backups (daily + incremental)",
        "Test restore procedure monthly",
        "Monitor disk usage, set alerts at 70%",
    ],
    "security": [
        "Enable TLS for all connections",
        "Use API keys, rotate quarterly",
        "Network policies: restrict to app subnets",
        "Audit logs enabled",
    ],
    "monitoring": [
        "Latency p50/p95/p99 per collection",
        "Query throughput (QPS)",
        "Index build queue depth",
        "Memory/CPU/disk per node",
        "Recall@K sampling (weekly)",
        "Error rates by type",
    ],
    "scaling": [
        "Horizontal pod autoscaler (K8s)",
        "Read replicas for query-heavy workloads",
        "Sharding strategy for >10M vectors",
        "Separate indexing nodes from query nodes",
    ]
}

Decision Framework

VECTOR DATABASE SELECTION:

Team has DevOps capacity?
├─→ No → Pinecone (managed) or Qdrant Cloud
└─→ Yes
    ├─→ Need GraphQL / semantic modules?
    │   └─→ Yes → Weaviate
    │
    ├─→ Kubernetes-native, massive scale (>100M)?
    │   └─→ Yes → Milvus
    │
    ├─→ Rust performance, filtering, hybrid search?
    │   └─→ Yes → Qdrant
    │
    ├─→ Budget sensitive, self-hosted?
    │   └─→ Yes → Qdrant or Milvus
    │
    └─→ Enterprise support required?
        └─→ Yes → Pinecone or Weaviate Cloud

Migration Paths

CURRENT → RECOMMENDED MIGRATION

Elasticsearch/OpenSearch → Qdrant (similar filter model)
pgvector → Qdrant/Milvus (better scaling)
Chroma → Qdrant (compatible API, better perf)
Pinecone → Qdrant Cloud (API compatible, cost savings)
Weaviate → Qdrant (if GraphQL not needed)
Custom FAISS → Milvus (managed FAISS at scale)

Resources

• Qdrant Documentation
• Pinecone Documentation
• Weaviate Documentation
• Milvus Documentation
• Vector Database Benchmarks
• ANN Benchmarks
• Choosing a Vector Database

---

Vector databases are not interchangeable — each has distinct strengths. Benchmark with YOUR data, YOUR queries, and YOUR scale before committing. The "best" database is the one your team can operate reliably.