---
source_url: "https://pypi.org/project/vectorizer-sdk/?utm_source=openai"
title: vectorizer-sdk · PyPI
mirrored_at: 2026-08-05T13:02:24.544Z
host: pypi.org
cited_in_42a: true
mirror_canonical: "https://index.42a.ai/pypi.org/project/vectorizer-sdk/index__q__utm_source_openai"
---

> **Original source:** https://pypi.org/project/vectorizer-sdk/?utm_source=openai

## 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/](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](https://github.com/hivellm/vectorizer/issues/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`](https://pypi.org/project/vectorizer-sdk/examples/rpc_quickstart.py) for a runnable end-to-end example.

### Switching transports

Goal

API

Default RPC, async

`await vectorizer_sdk.connect_async("vectorizer://host:15503")`

Default RPC, sync

`vectorizer_sdk.connect("vectorizer://host:15503")`

Legacy REST

`vectorizer_sdk.VectorizerClient(host="...", port=15002)`

## 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.5.0

## 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
        results \= await client.search\_vectors(
            collection\="my\_collection",
            query\="greeting",
            limit\=5
        )

        \# 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)](https://pypi.org/project/umicp-python/) 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

Feature

HTTP/HTTPS

UMICP

Transport

aiohttp (standard HTTP)

umicp-python package

Performance

Standard

Optimized for large payloads

Latency

Standard

Lower overhead

Firewall

Widely supported

May require configuration

Installation

Default

Requires umicp-python

#### 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)
collections \= await client.list\_collections()
vector \= await client.get\_vector("documents", "doc1")

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

Preference

Description

Use Case

`"replica"`

Route reads to replicas (round-robin)

Default for high read throughput

`"master"`

Route all reads to master

When you need read-your-writes consistency

`"nearest"`

Route to the node with lowest latency

Geo-distributed deployments

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

Operation Type

Routed To

Methods

**Writes**

Always Master

`insert_texts`, `insert_vectors`, `delete_vectors`, `create_collection`, `delete_collection`

**Reads**

Based on `read_preference`

`search_vectors`, `get_vector`, `list_collections`, `intelligent_search`, `semantic_search`, `hybrid_search`

#### 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](https://docs.cmmv-hive.org/vectorizer)
-   [API Reference](https://docs.cmmv-hive.org/vectorizer/api)
-   [Examples](https://pypi.org/project/vectorizer-sdk/examples.py)
-   [Test Documentation](https://pypi.org/project/vectorizer-sdk/TESTES_RESUMO.md)

## License

MIT License - see LICENSE file for details.

## Support

-   GitHub Issues: [https://github.com/cmmv-hive/vectorizer/issues](https://github.com/cmmv-hive/vectorizer/issues)
-   Email: [team@hivellm.org](mailto:team@hivellm.org)