The Complete Production SOP: How to Deploy and Run Qdrant Vector Database Payload-Filtered HNSW in Autonomous AI Agents

---
name: "qdrant-payload-filtered-hnsw"
description: "Perform vector similarity search with strict metadata payload filtering using Qdrant."
metadata:
  author: "Qdrant"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "setup qdrant collection"
  - "payload filter qdrant vector search"
  - "tune qdrant hnsw"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Qdrant Vector Database Payload-Filtered HNSW

## Execution SOP & Rules
1. Initialize client: client = QdrantClient(url=os.getenv('QDRANT_URL', 'http://localhost:6333')).
2. Create collection with explicit HnswConfigDiff and VectorParams.
3. Create payload indexes before massive ingestion using `client.create_payload_index(...)`.
4. Query using `client.search` or `client.query_points` with strict `models.Filter`.

import os
from qdrant_client import QdrantClient
from qdrant_client.http import models

def get_qdrant_client() -> QdrantClient:
    url = os.getenv("QDRANT_URL", "http://localhost:6333")
    api_key = os.getenv("QDRANT_API_KEY", None)
    client = QdrantClient(url=url, api_key=api_key, timeout=10.0)
    # Test connection health
    client.get_collections()
    return client

from qdrant_client import QdrantClient
from qdrant_client.http import models
from typing import List, Dict, Any

def setup_knowledge_base(client: QdrantClient, collection_name: str = "articles"):
    if not client.collection_exists(collection_name):
        client.create_collection(
            collection_name=collection_name,
            vectors_config=models.VectorParams(
                size=768,
                distance=models.Distance.COSINE,
                on_disk=False
            ),
            hnsw_config=models.HnswConfigDiff(
                m=16,
                ef_construct=100,
                full_scan_threshold=10000,
                on_disk=False
            ),
            quantization_config=models.ScalarQuantization(
                scalar=models.ScalarQuantizationConfig(
                    type=models.ScalarType.INT8,
                    quantile=0.99,
                    always_ram=True
                )
            )
        )
        # Mandatory payload index for deterministic HNSW filtering
        client.create_payload_index(
            collection_name=collection_name,
            field_name="tenant_id",
            field_schema=models.PayloadSchemaType.KEYWORD
        )

def search_tenant_documents(
    client: QdrantClient,
    collection_name: str,
    query_vector: List[float],
    tenant_id: str,
    limit: int = 5
) -> List[models.ScoredPoint]:
    payload_filter = models.Filter(
        must=[
            models.FieldCondition(
                key="tenant_id",
                match=models.MatchValue(value=tenant_id)
            )
        ]
    )
    return client.query_points(
        collection_name=collection_name,
        query=query_vector,
        query_filter=payload_filter,
        limit=limit,
        search_params=models.SearchParams(
            hnsw_ef=64,
            exact=False
        )
    ).points

import pytest
from qdrant_client import QdrantClient
from qdrant_client.http import models
from src.vector_service import setup_knowledge_base, search_tenant_documents

@pytest.fixture
def in_memory_client():
    client = QdrantClient(":memory:")
    setup_knowledge_base(client, collection_name="test_articles")
    return client

def test_filtered_hnsw_isolation(in_memory_client):
    client = in_memory_client
    # Ingest two distinct tenants
    client.upsert(
        collection_name="test_articles",
        points=[
            models.PointStruct(id=1, vector=[0.1] * 768, payload={"tenant_id": "tenant_a"}),
            models.PointStruct(id=2, vector=[0.1] * 768, payload={"tenant_id": "tenant_b"}),
        ]
    )
    
    results = search_tenant_documents(
        client=client,
        collection_name="test_articles",
        query_vector=[0.1] * 768,
        tenant_id="tenant_a",
        limit=10
    )
    
    assert len(results) == 1
    assert results[0].id == 1
    assert results[0].payload["tenant_id"] == "tenant_a"