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:

Variable	Default	Description
ACTIAN_VECTORAI_URL	localhost:6574	gRPC server address
ACTIAN_VECTORAI_REST_URL	http://localhost:6573	REST API base URL
ACTIAN_VECTORAI_API_KEY	—	API key for authentication
ACTIAN_VECTORAI_TLS	false	Enable TLS
ACTIAN_VECTORAI_TLS_CA_CERT	—	CA certificate path (verify the server)
ACTIAN_VECTORAI_TLS_CLIENT_CERT	—	Client certificate path (mTLS)
ACTIAN_VECTORAI_TLS_CLIENT_KEY	—	Client private-key path (mTLS)
ACTIAN_VECTORAI_ALLOW_INSECURE	false	Permit credentials over plaintext to a remote host
ACTIAN_VECTORAI_TIMEOUT	30.0	Default per-RPC timeout (seconds)
ACTIAN_VECTORAI_MAX_RETRIES	3	Max retry attempts
ACTIAN_VECTORAI_POOL_SIZE	1	gRPC connection-pool size

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:

Namespace	Access	Description
Collections	client.collections	create, list, get, update, delete, exists
Points	client.points	upsert, get, delete, payload ops, search, query, scroll, count
VDE	client.vde	engine lifecycle, rebuild, optimize, compact, import
Auth	client.auth	admin login/JWT, API-key management

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