Vectorizer Python SDK

A comprehensive Python SDK for the Vectorizer semantic search service.

Package: vectorizer_sdk (PEP 625 compliant) Version: 3.5.0 PyPI: https://pypi.org/project/vectorizer-sdk/

v3.5 — server alignment (no client API changes)

Version tracks the Vectorizer 3.5.0 server release: non-blocking search during batch inserts, PQ/Binary quantization wiring, SIMD quantize kernels, BM25-after-restart and WAL-durability fixes, and a security/dependency refresh. All server-internal — the client API is unchanged since v3.3 (REST control-surface parity + dashboard metrics). See CHANGELOG.md for the full method surface.

v3.2 — backpressure-aware client (HTTP 429 + Retry-After)

The REST VectorizerClient honors server-side bulk-upsert backpressure shipped in Vectorizer 3.2.0 (#263). On HTTP 429 Too Many Requests the client parses Retry-After (seconds form, 1 s default, 30 s cap), sleeps, and retries up to 3 times before raising a typed RateLimitError. Pre-3.2.0 clients bounced 429s into a generic 5xx and lost the retry budget. Identical semantics ship in every first-party SDK (Rust, Python, TypeScript, Go, C#) — see tests/test_retry_after_parse.py.

v3.1 — /insert_vectors + stable client-id upserts

• insert_vectors(collection, vectors, public_key=None) — bulk- insert pre-computed embeddings with caller-supplied vector ids. Skips the embedding pipeline entirely.
• insert / insert_texts: the request id is now used verbatim as the stored Vector.id (non-chunked) or as <id>#<chunk_index> (chunked). Re-running the same payload upserts in place instead of duplicating.
• Chunked vectors expose a flat payload layout ({content, file_path, chunk_index, parent_id, ...user_metadata}). Legacy nested payloads from ≤ 3.0.x stay readable during the deprecation window.

Client-id contract: non-empty, length ≤ 256, no leading/trailing whitespace, must not contain #.

v3.0 — VectorizerRPC is the default transport

Starting with v3.0, the recommended transport is VectorizerRPC: a binary, length-prefixed MessagePack protocol over raw TCP (port 15503 by default). It replaces JSON parsing on the hot path with a single msgpack.unpackb, removes per-request HTTP framing, and supports multiplexed call/response on a single long-lived TCP connection. The spec is at docs/specs/VECTORIZER_RPC.md in the parent repo.

The legacy REST VectorizerClient (over aiohttp) stays available for browsers, ops scripts, and anything that already targets HTTP.

import asyncio
import vectorizer_sdk

async def main():
    client = await vectorizer_sdk.connect_async("vectorizer://127.0.0.1:15503")
    # `hello` and `search_basic` are RPC-only (not available on the legacy
    # REST `VectorizerClient`).
    await client.hello(vectorizer_sdk.HelloPayload(client_name="my-app"))
    print(await client.list_collections())
    hits = await client.search_basic("docs", "vector database", limit=5)
    for hit in hits:
        print(hit.id, hit.score)
    await client.close()

asyncio.run(main())

A synchronous RpcClient is also exported for blocking scripts and notebooks; see examples/rpc_quickstart.py for a runnable end-to-end example.

Switching transports

Features

• VectorizerRPC (default in v3.x): binary, low-latency, multiplexed
• Multiple Transport Protocols: HTTP/HTTPS and UMICP support
• UMICP Protocol: High-performance protocol using umicp-sdk package (v0.3.2+)
• Vector Operations: Insert, search, update, delete vectors
• Collection Management: Create, delete, and monitor collections
• Semantic Search: Find similar content using embeddings
• Intelligent Search: AI-powered search with query expansion, MMR diversification, and domain expansion
• Semantic Search: Advanced semantic search with reranking and similarity thresholds
• Contextual Search: Context-aware search with metadata filtering
• Multi-Collection Search: Cross-collection search with intelligent aggregation
• Hybrid Search: Combine dense and sparse vectors for improved search quality
• Discovery Operations: Collection filtering, query expansion, and intelligent discovery
• File Operations: File content retrieval, chunking, project outlines, and related files
• Graph Relationships: Automatic relationship discovery, path finding, and edge management
• Summarization: Text and context summarization with multiple methods
• Workspace Management: Multi-workspace support for project organization
• Backup & Restore: Collection backup and restore operations
• Batch Operations: Efficient bulk insert, update, delete, and search
• Qdrant Compatibility: Full Qdrant 1.14.x REST API compatibility for easy migration
  • Snapshots API (create, list, delete, recover)
  • Sharding API (create shard keys, distribute data)
  • Cluster Management API (status, recovery, peer management, metadata)
  • Query API (query, batch query, grouped queries with prefetch)
  • Search Groups and Matrix API (grouped results, similarity matrices)
  • Named Vectors support (partial)
  • Quantization configuration (PQ and Binary)
• Error Handling: Comprehensive exception handling
• Async Support: Full async/await support for high performance
• Type Safety: Full type hints and validation

Installation

# Install from PyPI
pip install vectorizer-sdk

# Or specific version
pip install vectorizer-sdk==3.6.1

Package Layout (v3.x)

The flat 2,907-line client.py was split per API surface in the phase4_split-sdk-python-client refactor. Everything that used to hang off VectorizerClient still works — it's now composed from focused sub-clients:

sdks/python/
├── client.py              # compat shim → vectorizer.client
└── vectorizer/
    ├── __init__.py        # re-exports VectorizerClient + sub-clients
    ├── _base.py           # Transport ABC + RestTransport + TransportRouter
    ├── collections.py     # CollectionsClient
    ├── vectors.py         # VectorsClient
    ├── search.py          # SearchClient
    ├── graph.py           # GraphClient
    ├── admin.py           # AdminClient
    └── auth.py            # AuthClient

The legacy flat import still works:

from vectorizer import VectorizerClient  # recommended
from client import VectorizerClient       # legacy, still supported

Advanced users can pull a single surface:

from vectorizer import RestTransport
from vectorizer.collections import CollectionsClient

transport = RestTransport("http://localhost:15002", api_key="...")
collections = CollectionsClient(transport)
info = await collections.list_collections()

_base.Transport is an abstract base class. The concrete RestTransport ships here; the RpcTransport from the phase6_sdk-python-rpc work plugs in by subclassing the same ABC. That's why VectorizerClient("vectorizer://host:15503") will be the canonical default URL scheme once RPC lands — per-surface modules already route every call through Transport, never through aiohttp or httpx directly. See docs/specs/VECTORIZER_RPC.md for the RPC URL/port conventions.

Quick Start

import asyncio
from vectorizer import VectorizerClient, Vector

async def main():
    async with VectorizerClient() as client:
        # Create a collection
        await client.create_collection("my_collection", dimension=512)

        # Generate embedding
        embedding = await client.embed_text("Hello, world!")

        # Create vector
        vector = Vector(
            id="doc1",
            data=embedding,
            metadata={"text": "Hello, world!"}
        )

        # Insert text
        await client.insert_texts("my_collection", [{
            "id": "doc1",
            "text": "Hello, world!",
            "metadata": {"source": "example"}
        }])

        # Search for similar vectors — a list of SearchResult, not a
        # {"results": [...]} envelope (changed in 3.6.0; see CHANGELOG)
        results = await client.search_vectors(
            collection="my_collection",
            query="greeting",
            limit=5
        )
        for hit in results:
            print(hit.id, hit.score)

        # Intelligent search with multi-query expansion
        from models import IntelligentSearchRequest
        intelligent_results = await client.intelligent_search(
            IntelligentSearchRequest(
                query="machine learning algorithms",
                collections=["my_collection", "research"],
                max_results=15,
                domain_expansion=True,
                technical_focus=True,
                mmr_enabled=True,
                mmr_lambda=0.7
            )
        )

        # Semantic search with reranking
        from models import SemanticSearchRequest
        semantic_results = await client.semantic_search(
            SemanticSearchRequest(
                query="neural networks",
                collection="my_collection",
                max_results=10,
                semantic_reranking=True,
                similarity_threshold=0.6
            )
        )

        # Graph Operations (requires graph enabled in collection config)
        # List all graph nodes
        nodes = await client.list_graph_nodes("my_collection")
        print(f"Graph has {nodes.count} nodes")

        # Get neighbors of a node
        neighbors = await client.get_graph_neighbors("my_collection", "document1")
        print(f"Node has {len(neighbors.neighbors)} neighbors")

        # Find related nodes within 2 hops
        from models import FindRelatedRequest
        related = await client.find_related_nodes(
            "my_collection",
            "document1",
            FindRelatedRequest(max_hops=2, relationship_type="SIMILAR_TO")
        )
        print(f"Found {len(related.related)} related nodes")

        # Find shortest path between two nodes
        from models import FindPathRequest
        path = await client.find_graph_path(
            FindPathRequest(
                collection="my_collection",
                source="document1",
                target="document2"
            )
        )
        if path.found:
            print(f"Path found: {' -> '.join([n.id for n in path.path])}")

        # Create explicit relationship
        from models import CreateEdgeRequest
        edge = await client.create_graph_edge(
            CreateEdgeRequest(
                collection="my_collection",
                source="document1",
                target="document2",
                relationship_type="REFERENCES",
                weight=0.9
            )
        )
        print(f"Created edge: {edge.edge_id}")

        # Discover SIMILAR_TO edges for entire collection
        from models import DiscoverEdgesRequest
        discovery_result = await client.discover_graph_edges(
            "my_collection",
            DiscoverEdgesRequest(
                similarity_threshold=0.7,
                max_per_node=10
            )
        )
        print(f"Discovered {discovery_result.edges_created} edges")

        # Discover edges for a specific node
        node_discovery = await client.discover_graph_edges_for_node(
            "my_collection",
            "document1",
            DiscoverEdgesRequest(
                similarity_threshold=0.7,
                max_per_node=10
            )
        )
        print(f"Discovered {node_discovery.edges_created} edges for node")

        # Get discovery status
        status = await client.get_graph_discovery_status("my_collection")
        print(
            f"Discovery status: {status.total_nodes} nodes, "
            f"{status.total_edges} edges, "
            f"{status.progress_percentage:.1f}% complete"
        )

        # Contextual search with metadata filtering
        from models import ContextualSearchRequest
        contextual_results = await client.contextual_search(
            ContextualSearchRequest(
                query="deep learning",
                collection="my_collection",
                context_filters={"category": "AI", "year": 2023},
                max_results=10,
                context_weight=0.4
            )
        )

        # Multi-collection search
        from models import MultiCollectionSearchRequest
        multi_results = await client.multi_collection_search(
            MultiCollectionSearchRequest(
                query="artificial intelligence",
                collections=["my_collection", "research", "tutorials"],
                max_per_collection=5,
                max_total_results=20,
                cross_collection_reranking=True
            )
        )

        # Hybrid search (dense + sparse vectors)
        from models import HybridSearchRequest, SparseVector

        sparse_query = SparseVector(
            indices=[0, 5, 10, 15],
            values=[0.8, 0.6, 0.9, 0.7]
        )

        hybrid_results = await client.hybrid_search(
            HybridSearchRequest(
                collection="my_collection",
                query="search query",
                query_sparse=sparse_query,
                alpha=0.7,
                algorithm="rrf",  # "rrf", "weighted", or "alpha"
                dense_k=20,
                sparse_k=20,
                final_k=10
            )
        )

        print(f"Found {len(hybrid_results.results)} similar vectors")

        # Qdrant-compatible API usage
        # List collections
        qdrant_collections = await client.qdrant_list_collections()
        print(f"Qdrant collections: {qdrant_collections}")

        # Search points (Qdrant format)
        qdrant_results = await client.qdrant_search_points(
            collection="my_collection",
            vector=embedding,
            limit=10,
            with_payload=True
        )
        print(f"Qdrant search results: {qdrant_results}")

asyncio.run(main())

Advanced Features

Discovery Operations

Filter Collections

Filter collections based on query relevance:

filtered = await client.filter_collections(
    query="machine learning",
    min_score=0.5
)

Expand Queries

Expand queries with related terms:

expanded = await client.expand_queries(
    query="neural networks",
    max_expansions=5
)

Discover

Intelligent discovery across collections:

discovery = await client.discover(
    query="authentication methods",
    max_results=10
)

File Operations

Get File Content

Retrieve file content from collection:

content = await client.get_file_content(
    collection="docs",
    file_path="src/client.py"
)

List Files

List all files in a collection:

files = await client.list_files_in_collection(
    collection="docs"
)

Get File Chunks

Get ordered chunks of a file:

chunks = await client.get_file_chunks_ordered(
    collection="docs",
    file_path="README.md",
    chunk_size=1000
)

Get Project Outline

Get project structure outline:

outline = await client.get_project_outline(
    collection="codebase"
)

Get Related Files

Find files related to a specific file:

related = await client.get_related_files(
    collection="codebase",
    file_path="src/client.py",
    max_results=5
)

Summarization Operations

WARNING: The /summarize/* REST endpoints are documented but not yet wired server-side (see DOC_GAP_ANALYSIS). The SDK methods below (summarize_text, summarize_context) will fail until server wiring is complete.

Summarize Text

Summarize text using various methods:

from models import SummarizeTextRequest

summary = await client.summarize_text(
    SummarizeTextRequest(
        text="Long document text...",
        method="extractive",  # 'extractive', 'abstractive', 'hybrid'
        max_length=200
    )
)

Summarize Context

Summarize context with metadata:

from models import SummarizeContextRequest

summary = await client.summarize_context(
    SummarizeContextRequest(
        context="Document context...",
        method="abstractive",
        focus="key_points"
    )
)

Workspace Management

WARNING: add_workspace, list_workspaces, and remove_workspace are exposed via the REST transport through dynamic __getattr__ delegation, but they are not first-class SDK methods yet. They work at runtime but won't autocomplete in IDEs. A future release will add explicit methods.

Add Workspace

Add a new workspace:

await client.add_workspace(
    name="my-project",
    path="/path/to/project"
)

List Workspaces

List all workspaces:

workspaces = await client.list_workspaces()

Remove Workspace

Remove a workspace:

await client.remove_workspace(
    name="my-project"
)

Backup Operations

WARNING: create_backup, list_backups, and restore_backup are exposed via the REST transport through dynamic __getattr__ delegation, but they are not first-class SDK methods yet. They work at runtime but won't autocomplete in IDEs. A future release will add explicit methods.

Create Backup

Create a backup of collections:

backup = await client.create_backup(
    name="backup-2024-11-24"
)

List Backups

List all available backups:

backups = await client.list_backups()

Restore Backup

Restore from a backup:

await client.restore_backup(
    filename="backup-2024-11-24.vecdb"
)

Configuration

HTTP Configuration (Default)

from vectorizer import VectorizerClient

# Default HTTP configuration
client = VectorizerClient(
    base_url="http://localhost:15002",
    api_key="your-api-key",
    timeout=30
)

UMICP Configuration (High Performance)

UMICP (Universal Messaging and Inter-process Communication Protocol) provides significant performance benefits using the official umicp-python package.

Using Connection String

from vectorizer import VectorizerClient

client = VectorizerClient(
    connection_string="umicp://localhost:15003",
    api_key="your-api-key"
)

print(f"Using protocol: {client.get_protocol()}")  # Output: umicp

Using Explicit Configuration

from vectorizer import VectorizerClient

client = VectorizerClient(
    protocol="umicp",
    api_key="your-api-key",
    umicp={
        "host": "localhost",
        "port": 15003
    },
    timeout=60
)

When to Use UMICP

Use UMICP when:

• Large Payloads: Inserting or searching large batches of vectors
• High Throughput: Need maximum performance for production workloads
• Low Latency: Need minimal protocol overhead

Use HTTP when:

• Development: Quick testing and debugging
• Firewall Restrictions: Only HTTP/HTTPS allowed
• Simple Deployments: No need for custom protocol setup

Protocol Comparison

Installing with UMICP Support

pip install vectorizer-sdk umicp-python

Master/Slave Configuration (Read/Write Separation)

Vectorizer supports Master-Replica replication for high availability and read scaling. The SDK provides automatic routing - writes go to master, reads are distributed across replicas.

Basic Setup

from vectorizer import VectorizerClient

# Configure with master and replicas - SDK handles routing automatically
client = VectorizerClient(
    hosts={
        "master": "http://master-node:15002",
        "replicas": ["http://replica1:15002", "http://replica2:15002"]
    },
    api_key="your-api-key",
    read_preference="replica"  # "master" | "replica" | "nearest"
)

# Writes automatically go to master
await client.create_collection("documents", dimension=768)
await client.insert_texts("documents", [
    {"id": "doc1", "text": "Sample document", "metadata": {"source": "api"}}
])
# Update-via-reinsert: re-call `insert_texts` with the same id to replace the record.
await client.insert_texts("documents", [
    {"id": "doc1", "text": "Sample document (updated)", "metadata": {"updated": True}}
])
report = await client.delete_vectors("documents", ["doc1"])
print(f"deleted={report.deleted} failed={report.failed}")

# Tier demotion: move vectors between collections without re-embedding
# (issue #265). Insert into dst lands BEFORE delete from src so a
# mid-batch crash leaves a recoverable duplicate, never data loss.
mv = await client.move_to_collection("hot", "warm", ["vec-1", "vec-2"])
for row in mv.results:
    if row.status != "ok":
        print(f"move failed id={row.id!r} status={row.status} err={row.error!r}")

# Reads automatically go to replicas (load balanced)
results = await client.search_vectors("documents", query="sample", limit=10)  # List[SearchResult]
collections = await client.list_collections()
vector = await client.get_vector("documents", "doc1")  # Vector — vector.data, vector.metadata

Control surface (3.4)

Admin / observability

import asyncio
from vectorizer_sdk import VectorizerClient

async def main():
    client = VectorizerClient(base_url="http://localhost:15002")

    # Server health, uptime, collection/vector counts
    stats = await client.get_stats()
    print(f"Total vectors: {stats['total_vectors']}")

    status = await client.get_status()
    print(f"Server v{status['version']}, uptime: {status['uptime']}s")

    # Recent logs
    logs = await client.get_logs(lines=50, level="INFO")
    for entry in logs:
        print(f"{entry['timestamp']}: {entry['message']}")

    # Per-collection indexing progress
    progress = await client.get_indexing_progress()
    for collection, pct in progress.items():
        print(f"{collection}: {pct:.1f}% complete")

    # Force flush one collection
    await client.force_save_collection("my_docs")

    # List and clean empty collections
    empty = await client.list_empty_collections()
    if empty:
        report = await client.cleanup_empty_collections()
        print(f"Cleaned up {report['deleted']} empty collections")

    # List workspaces
    workspaces = await client.list_workspaces()
    print(f"Workspaces: {workspaces}")

asyncio.run(main())

Auth

import asyncio
from vectorizer_sdk import VectorizerClient

async def main():
    client = VectorizerClient(base_url="http://localhost:15002")

    # Current user info
    me = await client.me()
    print(f"Logged in as: {me['username']} (roles: {', '.join(me['roles'])})")

    # Refresh token with extended TTL
    token = await client.refresh_token()
    print(f"Token refreshed, expires in: {token['expires_in']} seconds")

    # Validate password before creating account
    report = await client.validate_password("MySecure123!")
    print(f"Valid: {report['valid']}, feedback: {', '.join(report['feedback'])}")

    # Create API key for programmatic access
    api_key = await client.create_api_key(
        name="integration-key",
        expires_in=86400 * 365  # 1 year
    )
    print(f"API Key: {api_key['api_key']}")

    # List and revoke API keys
    keys = await client.list_api_keys()
    for key in keys:
        print(f"Key: {key['id']} (expires: {key['expires_at']})")
    await client.revoke_api_key(keys[0]['id'])

    # Change password
    await client.change_password("newPassword123!")

    # Logout
    await client.logout()

asyncio.run(main())

Replication

import asyncio
from vectorizer_sdk import VectorizerClient

async def main():
    client = VectorizerClient(base_url="http://localhost:15002")

    # Check replication role and status
    status = await client.get_replication_status()
    print(f"Role: {status['role']}, enabled: {status['enabled']}")

    # Get replication statistics (lag, bytes synced)
    stats = await client.get_replication_stats()
    print(f"Bytes synced: {stats['bytes_synced']}")

    # List all replicas connected to this master
    replicas = await client.list_replicas()
    for replica in replicas:
        print(f"Replica: {replica['address']} (lag: {replica['lag_ms']}ms)")

asyncio.run(main())

Discovery pipeline

The discovery pipeline chains six stages from broad search to final LLM-ready prompt:

import asyncio
from vectorizer_sdk import VectorizerClient

async def main():
    client = VectorizerClient(base_url="http://localhost:15002")

    # Stage 1: Broad discovery — multi-query search across all collections
    broad = await client.broad_discovery(
        query="machine learning algorithms",
        max_results=20
    )
    print(f"Found {len(broad['results'])} broad results")

    # Stage 2: Semantic focus — narrow search to top collection
    focused = await client.semantic_focus(
        query="neural networks",
        collection="research",
        max_results=10
    )
    print(f"Focused results: {len(focused['results'])}")

    # Stage 3: Promote README — elevate high-quality chunks
    promoted = await client.promote_readme(
        results=focused['results'],
        readme_boost=2.0
    )

    # Stage 4: Compress evidence — distill to bullet points
    bullets = await client.compress_evidence(
        chunks=promoted['results'],
        max_bullets=15
    )
    print(f"Evidence bullets: {bullets['bullets']}")

    # Stage 5: Build answer plan — organize bullets into sections
    plan = await client.build_answer_plan(
        evidence=bullets['bullets'],
        max_sections=5
    )
    print(f"Sections: {plan['sections']}")

    # Stage 6: Render LLM prompt — final markdown string for LLM
    llm_prompt = await client.render_llm_prompt(
        plan=plan,
        style="formal"
    )
    print(f"LLM prompt:\n{llm_prompt['markdown']}")

asyncio.run(main())

Hub backups

import asyncio
from vectorizer_sdk import VectorizerClient

async def main():
    client = VectorizerClient(base_url="http://localhost:15002")

    user_id = "user-123"

    # List user's backups
    backups = await client.list_user_backups(user_id)
    for backup in backups:
        print(f"Backup: {backup['id']} (size: {backup['size_bytes']} bytes)")

    # Create a new backup
    new_backup = await client.create_user_backup(
        user_id=user_id,
        name="full-backup-2024-01",
        description="January full backup",
        collections=None  # backup all
    )
    print(f"Created backup: {new_backup['id']}")

    # Restore a backup
    await client.restore_user_backup(
        user_id=user_id,
        backup_id=new_backup['id']
    )
    print("Restore started")

    # Delete old backup
    await client.delete_user_backup(user_id, backups[0]['id'])

asyncio.run(main())

Read Preferences

Read-Your-Writes Consistency

For operations that need to immediately read what was just written:

# Option 1: Override read preference for specific operation
await client.insert_texts("docs", [new_doc])
result = await client.get_vector("docs", new_doc["id"], read_preference="master")

# Option 2: Use context manager for a block of operations
async with client.with_master() as master_client:
    await master_client.insert_texts("docs", [new_doc])
    result = await master_client.get_vector("docs", new_doc["id"])

Automatic Operation Routing

The SDK automatically classifies operations:

Standalone Mode (Single Node)

For development or single-node deployments, use the simple configuration:

# Single node - no replication
client = VectorizerClient(
    base_url="http://localhost:15002",
    api_key="your-api-key"
)

Testing

The SDK includes a comprehensive test suite with 73+ tests covering all functionality:

Running Tests

# Run basic tests (recommended)
python3 test_simple.py

# Run comprehensive tests
python3 test_sdk_comprehensive.py

# Run all tests with detailed reporting
python3 run_tests.py

# Run specific test
python3 -m unittest test_simple.TestBasicFunctionality

Test Coverage

• Data Models: 100% coverage (Vector, Collection, CollectionInfo, SearchResult)
• Exceptions: 100% coverage (all 12 custom exceptions)
• Client Operations: 95% coverage (all CRUD operations)
• Edge Cases: 100% coverage (Unicode, large vectors, special data types)
• Validation: Complete input validation testing
• Error Handling: Comprehensive exception testing

Test Results

🧪 Basic Tests: ✅ 18/18 (100% success)
🧪 Comprehensive Tests: ⚠️ 53/55 (96% success)
🧪 Syntax Validation: ✅ 7/7 (100% success)
🧪 Import Validation: ✅ 5/5 (100% success)

📊 Overall Success Rate: 75%
⏱️ Total Execution Time: <0.4 seconds

Test Categories

1. Unit Tests: Individual component testing
2. Integration Tests: Mock-based workflow testing
3. Validation Tests: Input validation and error handling
4. Edge Case Tests: Unicode, large data, special scenarios
5. Syntax Tests: Code compilation and import validation

Qdrant Feature Parity

The SDK provides full compatibility with Qdrant 1.14.x REST API:

Snapshots API

# List collection snapshots
snapshots = await client.qdrant_list_collection_snapshots("my_collection")

# Create snapshot
snapshot = await client.qdrant_create_collection_snapshot("my_collection")

# Delete snapshot
await client.qdrant_delete_collection_snapshot("my_collection", "snapshot_name")

# Recover from snapshot
await client.qdrant_recover_collection_snapshot("my_collection", "snapshots/backup.snapshot")

# Full snapshot (all collections)
full_snapshot = await client.qdrant_create_full_snapshot()

Sharding API

# List shard keys
shard_keys = await client.qdrant_list_shard_keys("my_collection")

# Create shard key
await client.qdrant_create_shard_key("my_collection", {"shard_key": "tenant_id"})

# Delete shard key
await client.qdrant_delete_shard_key("my_collection", {"shard_key": "tenant_id"})

Cluster Management API

# Get cluster status
status = await client.qdrant_get_cluster_status()

# Recover current peer
await client.qdrant_cluster_recover()

# Remove peer
await client.qdrant_remove_peer("peer_123")

# Metadata operations
metadata_keys = await client.qdrant_list_metadata_keys()
key_value = await client.qdrant_get_metadata_key("my_key")
await client.qdrant_update_metadata_key("my_key", {"config": "value"})

Query API

# Basic query
results = await client.qdrant_query_points("my_collection", {
    "query": [0.1, 0.2, 0.3],
    "limit": 10,
    "with_payload": True
})

# Query with prefetch (multi-stage retrieval)
results = await client.qdrant_query_points("my_collection", {
    "prefetch": [{"query": [0.1, 0.2, 0.3], "limit": 100}],
    "query": {"fusion": "rrf"},
    "limit": 10
})

# Batch query
results = await client.qdrant_batch_query_points("my_collection", {
    "searches": [
        {"query": [0.1, 0.2, 0.3], "limit": 5},
        {"query": [0.3, 0.4, 0.5], "limit": 5}
    ]
})

# Query groups
results = await client.qdrant_query_points_groups("my_collection", {
    "query": [0.1, 0.2, 0.3],
    "group_by": "category",
    "group_size": 3,
    "limit": 10
})

Search Groups & Matrix API

# Search groups
groups = await client.qdrant_search_points_groups("my_collection", {
    "vector": [0.1, 0.2, 0.3],
    "group_by": "category",
    "group_size": 3,
    "limit": 5
})

# Search matrix pairs (pairwise similarity)
pairs = await client.qdrant_search_matrix_pairs("my_collection", {
    "sample": 100,
    "limit": 500
})

# Search matrix offsets (compact format)
offsets = await client.qdrant_search_matrix_offsets("my_collection", {
    "sample": 100,
    "limit": 500
})

Documentation

• Full Documentation
• API Reference
• Examples
• Test Documentation

License

MIT License - see LICENSE file for details.

Support

• GitHub Issues: https://github.com/cmmv-hive/vectorizer/issues
• Email: team@hivellm.org