actian-vectorai-client

Official Python client for Actian VectorAI DB

Actian VectorAI Python Client

The official Python SDK for Actian VectorAI DB — a fully typed client with synchronous and asynchronous APIs, a namespaced surface, a type-safe filter DSL, hybrid-search fusion, and first-class VDE engine operations.

Features

• Async & sync clients — AsyncVectorAIClient and a synchronous VectorAIClient
• Namespaced API — client.collections, client.points, client.vde, client.auth
• Fully typed — ships py.typed; Pydantic models and hints throughout
• Type-safe filter DSL — fluent Field / FilterBuilder payload filtering
• Hybrid fusion — client-side RRF and DBSF for merging multi-query results
• Index selection — HNSW, Flat, and the IVF family with nlist / nprobe tuning
• VDE operations — engine lifecycle, online rebuilds, compaction, dataset import
• Authentication — admin login, JWT, and API-key management
• Resilient transport — gRPC primary with REST secondary, retries, and smart batching
• Bring your own embeddings — store and search any list[float] vectors

Installation

pip install actian-vectorai-client

Requires Python 3.10+ (tested on 3.10–3.14).

Quick start

Sync

from actian_vectorai import VectorAIClient, VectorParams, Distance, PointStruct

with VectorAIClient() as client:
    info = client.health_check()
    print(f"Connected to {info['title']} v{info['version']}")

    client.collections.create(
        "products",
        vectors_config=VectorParams(size=128, distance=Distance.Cosine),
    )
    client.points.upsert("products", [
        PointStruct(id=1, vector=[0.1] * 128, payload={"name": "Widget"}),
        PointStruct(id=2, vector=[0.2] * 128, payload={"name": "Gadget"}),
    ])
    results = client.points.search("products", vector=[0.15] * 128, limit=5)
    for r in results:
        print(f"  id={r.id}  score={r.score:.4f}  payload={r.payload}")

    client.collections.delete("products")

Async

import asyncio
from actian_vectorai import AsyncVectorAIClient, VectorParams, Distance, PointStruct

async def main():
    async with AsyncVectorAIClient() as client:
        await client.collections.create(
            "demo",
            vectors_config=VectorParams(size=128, distance=Distance.Cosine),
        )
        await client.points.upsert("demo", [
            PointStruct(id=1, vector=[0.1] * 128, payload={"tag": "hello"}),
        ])
        results = await client.points.search("demo", vector=[0.1] * 128, limit=5)
        print(results)
        await client.collections.delete("demo")

asyncio.run(main())

Authentication

Credentials are sent on every request. Provide them explicitly or via the ACTIAN_VECTORAI_* environment (constructor kwargs take priority):

client = VectorAIClient(api_key="vdai_...")        # explicit
# or: export ACTIAN_VECTORAI_API_KEY=vdai_...      # environment / .env

Admin and API-key management is available under client.auth (admin login, JWT, and create / list / rotate / delete API keys).

Configuration

Configuration is read from ACTIAN_VECTORAI_* environment variables (and a local .env, if present). .env is git-ignored; start from the template:

cp .env.example .env    # then set the server address and any credentials

Variables:

from actian_vectorai import Settings, settings

print(settings.url)                              # global, lazily loaded
cfg = Settings(url="remote:6574", timeout=60.0)  # explicit overrides

TLS & secure connections

Enable TLS and, optionally, mutual TLS:

client = VectorAIClient(
    "vectorai.example.com:6574",
    tls=True,
    tls_ca_cert="/path/ca.pem",           # verify the server
    tls_client_cert="/path/client.pem",   # mTLS (optional)
    tls_client_key="/path/client-key.pem",
    api_key="vdai_...",
)

When credentials would be sent over an unencrypted connection to a non-loopback host, the client logs a warning (it never blocks the connection). Use tls=True for production, or pass allow_insecure=True to acknowledge the risk and silence the warning on a trusted network.

Retries

Transient failures are retried with exponential backoff. Tune the policy:

from actian_vectorai import RetryConfig, VectorAIClient

client = VectorAIClient(
    retry_config=RetryConfig(max_retries=5, initial_backoff_ms=200),
)

API overview

The client is organized into namespaces:

# Collections
client.collections.create("col", vectors_config=VectorParams(size=128, distance=Distance.Cosine))
client.collections.list()

# Points
client.points.upsert("col", [PointStruct(id=1, vector=[...], payload={...})])
client.points.get("col", ids=[1, 2, 3])
client.upload_points("col", points, batch_size=256)     # bulk with auto-batching

# Search & query
results = client.points.search("col", vector=[...], limit=10)
results = client.points.query("col", query=[...], limit=10)
points, next_offset = client.points.scroll("col", limit=100)

# VDE
client.vde.rebuild_index("col")
client.vde.compact_collection("col")

Filter DSL

from actian_vectorai import Field, FilterBuilder

f = (
    FilterBuilder()
    .must(Field("category").eq("electronics"))
    .must(Field("price").between(100.0, 500.0))
    .must_not(Field("deleted").eq(True))
    .build()
)
results = client.points.search("products", vector=[...], limit=10, filter=f)

Hybrid fusion

Merge results from multiple queries client-side:

from actian_vectorai import reciprocal_rank_fusion, distribution_based_score_fusion

dense  = client.points.search("col", vector=dense_query,  limit=50)
sparse = client.points.search("col", vector=sparse_query, limit=50)

fused = reciprocal_rank_fusion([dense, sparse], limit=10, weights=[0.7, 0.3])
fused = distribution_based_score_fusion([dense, sparse], limit=10)

Documentation

• Python SDK Quickstart — install, authenticate, first client
• API Reference — endpoints, schemas, examples
• Academy — tutorials and guided walkthroughs

License

Proprietary — © 2026 Actian Corporation. All rights reserved.