Vector Database Comparison & Benchmarks 2025: Pinecone vs Weaviate vs Milvus vs Qdrant vs pgvector

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue, Range,
    HnswConfigDiff, OptimizersConfigDiff
)
from openai import OpenAI
import uuid
import time
from typing import Optional


class EnterpriseVectorStore:
    """
    Production-grade Qdrant wrapper for enterprise RAG workloads.
    Handles collection creation, upsert, and filtered search.
    """
    COLLECTION = "enterprise_docs"
    VECTOR_DIM = 768  # OpenAI ada-002 / text-embedding-3-small at 768

    def __init__(self, host: str = "localhost", port: int = 6333):
        self.client = QdrantClient(host=host, port=port, timeout=30)
        self.openai = OpenAI()
        self._ensure_collection()

    def _ensure_collection(self):
        existing = [c.name for c in self.client.get_collections().collections]
        if self.COLLECTION not in existing:
            self.client.create_collection(
                collection_name=self.COLLECTION,
                vectors_config=VectorParams(
                    size=self.VECTOR_DIM,
                    distance=Distance.COSINE,
                ),
                hnsw_config=HnswConfigDiff(
                    m=16,
                    ef_construct=128,
                    full_scan_threshold=10_000,  # HNSW below this, full scan above
                    on_disk=True,  # Critical at 100M+ vectors
                ),
                optimizers_config=OptimizersConfigDiff(
                    indexing_threshold=20_000,  # Batch index, not per-upsert
                    memmap_threshold=50_000,
                ),
            )
            # Create payload index for filtered search performance
            for field in ["tenant_id", "status", "doc_type"]:
                self.client.create_payload_index(
                    collection_name=self.COLLECTION,
                    field_name=field,
                    field_schema="keyword"
                )
            self.client.create_payload_index(
                collection_name=self.COLLECTION,
                field_name="created_at",
                field_schema="float"
            )
            print(f"Collection '{self.COLLECTION}' created with payload indexes.")

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

    def upsert_document(self, doc_id: str, text: str, metadata: dict) -> None:
        vector = self.embed(text)
        self.client.upsert(
            collection_name=self.COLLECTION,
            points=[PointStruct(
                id=str(uuid.uuid5(uuid.NAMESPACE_DNS, doc_id)),
                vector=vector,
                payload={**metadata, "doc_id": doc_id, "text": text[:500]}
            )]
        )

    def search(
        self,
        query: str,
        tenant_id: str,
        doc_type: Optional[str] = None,
        min_created_at: Optional[float] = None,
        top_k: int = 10,
    ) -> list[dict]:
        query_vector = self.embed(query)

        must_conditions = [
            FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id)),
            FieldCondition(key="status", match=MatchValue(value="published")),
        ]
        if doc_type:
            must_conditions.append(
                FieldCondition(key="doc_type", match=MatchValue(value=doc_type))
            )
        if min_created_at:
            must_conditions.append(
                FieldCondition(key="created_at", range=Range(gte=min_created_at))
            )

        start = time.perf_counter()
        results = self.client.search(
            collection_name=self.COLLECTION,
            query_vector=query_vector,
            query_filter=Filter(must=must_conditions),
            limit=top_k,
            with_payload=True,
        )
        latency_ms = (time.perf_counter() - start) * 1000
        print(f"Search latency: {latency_ms:.1f}ms, results: {len(results)}")

        return [
            {"doc_id": r.payload["doc_id"], "score": r.score, "text": r.payload.get("text")}
            for r in results
        ]


# Usage
if __name__ == "__main__":
    store = EnterpriseVectorStore(host="localhost", port=6333)
    store.upsert_document(
        doc_id="doc-001",
        text="Q4 2025 revenue increased 18% YoY driven by enterprise segment growth.",
        metadata={"tenant_id": "acme-corp", "status": "published",
                  "doc_type": "financial", "created_at": 1735000000.0}
    )
    results = store.search(
        query="What was revenue growth in Q4?",
        tenant_id="acme-corp",
        doc_type="financial",
    )
    print(results)