Qdrant Vector Database Best Practices

Production patterns for Qdrant vector search. Covers collections, search, filtering, and performance optimization.

When to Apply

• Creating or modifying Qdrant collections
• Implementing semantic search for news/documents
• Building RAG pipelines with vector retrieval
• Optimizing search performance and accuracy
• Managing payload indexes and filtering

1. Collection Configuration (CRITICAL)

Create optimized collection

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, OptimizersConfigDiff,
    ScalarQuantization, ScalarQuantizationConfig, ScalarType,
    HnswConfigDiff, PayloadSchemaType
)

client = QdrantClient(host="localhost", port=6333)

# High-speed with low memory
client.create_collection(
    collection_name="telegram_news_v2",
    vectors_config=VectorParams(
        size=384,  # Match your embedding model
        distance=Distance.COSINE,
        on_disk=True  # Keep vectors on disk
    ),
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantizationConfig(
            type=ScalarType.INT8,
            always_ram=True  # Quantized in RAM for speed
        )
    ),
    hnsw_config=HnswConfigDiff(
        m=16,              # Number of links per node
        ef_construct=100,  # Build-time precision
        on_disk=False      # HNSW graph in RAM
    ),
    optimizers_config=OptimizersConfigDiff(
        memmap_threshold=20000
    )
)

HNSW tuning guide

Distance metrics

• COSINE - Best for normalized embeddings (most common)
• DOT - For non-normalized, when magnitude matters
• EUCLID - For spatial/geographic data

2. Payload Indexing (CRITICAL)

ALWAYS create indexes BEFORE uploading data

# Index text fields for exact match filtering
client.create_payload_index(
    collection_name="telegram_news_v2",
    field_name="source",
    field_schema=PayloadSchemaType.KEYWORD
)

# Index datetime fields for range queries
client.create_payload_index(
    collection_name="telegram_news_v2",
    field_name="timestamp",
    field_schema=PayloadSchemaType.DATETIME
)

# Index numeric fields for range filtering
client.create_payload_index(
    collection_name="telegram_news_v2",
    field_name="relevance_score",
    field_schema=PayloadSchemaType.FLOAT
)

Why: Without indexes, Qdrant does full scan on payloads = SLOW

3. Search Patterns (HIGH)

Basic semantic search

from qdrant_client.models import Filter, FieldCondition, MatchValue, Range

results = client.search(
    collection_name="telegram_news_v2",
    query_vector=embedding,
    limit=10,
    score_threshold=0.7,  # Minimum similarity
    with_payload=True
)

Filtered search (combine semantic + metadata)

from datetime import datetime, timedelta

results = client.search(
    collection_name="telegram_news_v2",
    query_vector=embedding,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="timestamp",
                range=Range(
                    gte=(datetime.utcnow() - timedelta(hours=24)).isoformat()
                )
            ),
            FieldCondition(
                key="source",
                match=MatchValue(value="crypto_news")
            )
        ]
    ),
    limit=5,
    with_payload=["title", "text", "timestamp", "source"]
)

Search with payload selection (reduce transfer)

# Only return specific fields
results = client.search(
    collection_name="telegram_news_v2",
    query_vector=embedding,
    limit=10,
    with_payload=["title", "summary", "timestamp"]  # Not full text
)

4. Upsert Patterns (HIGH)

Batch upsert

from qdrant_client.models import PointStruct

points = [
    PointStruct(
        id=hash(article["url"]) & 0xFFFFFFFF,  # Deterministic ID
        vector=embedding,
        payload={
            "title": article["title"],
            "text": article["text"],
            "source": article["source"],
            "timestamp": article["timestamp"].isoformat(),
            "symbols": article.get("symbols", []),
        }
    )
    for article, embedding in zip(articles, embeddings)
]

# Batch upload (max ~100 per batch)
client.upsert(
    collection_name="telegram_news_v2",
    points=points
)

Idempotent upsert (use deterministic IDs)

• Hash URL or unique identifier for point ID
• Re-upserting same ID overwrites cleanly
• Prevents duplicates

5. Advanced Filtering

Complex filters

# Must match ANY of these sources AND be recent
filter = Filter(
    must=[
        FieldCondition(
            key="timestamp",
            range=Range(gte="2026-02-07T00:00:00Z")
        )
    ],
    should=[  # OR logic
        FieldCondition(key="source", match=MatchValue(value="bloomberg")),
        FieldCondition(key="source", match=MatchValue(value="reuters")),
        FieldCondition(key="source", match=MatchValue(value="coindesk"))
    ]
)

Nested object filtering

# Filter by nested payload field
FieldCondition(
    key="metadata.language",
    match=MatchValue(value="en")
)

6. Performance Optimization

Query tuning

# Override HNSW ef at query time (higher = more accurate, slower)
results = client.search(
    collection_name="telegram_news_v2",
    query_vector=embedding,
    limit=10,
    search_params=SearchParams(
        hnsw_ef=128,  # Default usually 100
        exact=False   # Set True for small collections (<1000)
    )
)

Collection stats

info = client.get_collection("telegram_news_v2")
print(f"Points: {info.points_count}")
print(f"Indexed: {info.indexed_vectors_count}")
print(f"Status: {info.status}")

Maintenance

• Monitor indexed_vectors_count vs points_count
• Wait for indexing after large uploads before searching
• Use client.update_collection() to change optimizer settings
• Compact collection after many deletes

7. Embedding Best Practices

• Use consistent embedding model across insert and search
• Normalize embeddings for cosine similarity
• Cache embeddings for frequently searched queries
• Batch embedding generation for efficiency
• Model size: 384 dims (fast) vs 768/1024 dims (more accurate)

8. Error Handling

from qdrant_client.http.exceptions import UnexpectedResponse

try:
    results = client.search(...)
except UnexpectedResponse as e:
    logger.error(f"Qdrant error: {e.status_code} - {e.content}")
    # Fallback to cached results
except ConnectionError:
    logger.error("Qdrant connection lost")
    # Retry with backoff