actian-vectorai-client 1.0.0

Python SDK for Actian VectorAI DB

 

Python client library for Actian VectorAI DB

Actian VectorAI Python Client

Official Python SDK for Actian VectorAI DB. Features async/sync support, namespaced API, type-safe Filter DSL, hybrid fusion, VDE engine operations, and smart batching.

Features

• Async & Sync clients — Full async/await with AsyncVectorAIClient, synchronous wrapper with VectorAIClient
• Namespaced API — client.collections, client.points, client.vde
• Environment Configuration — Auto-loading .env files, ACTIAN_VECTORAI_* env vars, Settings class
• Authentication — Access token / API key Bearer auth for gRPC and REST
• Type-safe Filter DSL — Fluent Field / FilterBuilder API for payload filtering
• Hybrid Fusion — Client-side RRF for multi-query result merging
• VDE Operations — Engine lifecycle, rebuilds, compaction, dataset import
• Smart Batching — Automatic request batching with SmartBatcher
• Embedding Providers — Built-in OpenAI embeddings support with OpenAIEmbedder
• Pydantic models — Type hints and validation throughout
• gRPC + REST transport — gRPC primary transport with REST fallback transport

Installation

pip install actian-vectorai-client

With OpenAI embeddings support:

pip install actian-vectorai-client[openai]

For development:

cd clients/python
uv venv .venv
uv pip install -e ".[dev]"

Requires: Python 3.10+ (validated on 3.10 through 3.14), grpcio 1.80+, protobuf 5.29+, pydantic 2.10+, numpy 1.26+

Quick Start

Sync Client

from actian_vectorai import VectorAIClient, VectorParams, Distance, PointStruct

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

    # Create collection
    client.collections.create(
        "products",
        vectors_config=VectorParams(size=128, distance=Distance.Cosine),
    )

    # Insert points
    client.points.upsert("products", [
        PointStruct(id=1, vector=[0.1] * 128, payload={"name": "Widget"}),
        PointStruct(id=2, vector=[0.2] * 128, payload={"name": "Gadget"}),
        PointStruct(id=3, vector=[0.3] * 128, payload={"name": "Gizmo"}),
    ])

    # Search
    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}")

    # Clean up
    client.collections.delete("products")

Async Client

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

Actian VectorAI DB supports access token authentication. Tokens are passed as Bearer headers on every gRPC and REST request.

# Method 1: Explicit constructor argument
client = VectorAIClient(access_token="your-token")

# Method 2: Environment variable
# export ACTIAN_VECTORAI_ACCESS_TOKEN=your-token
client = VectorAIClient()  # auto-resolved

# Method 3: .env file (auto-loaded from cwd)
# ACTIAN_VECTORAI_ACCESS_TOKEN=your-token
client = VectorAIClient()

Priority: constructor kwargs > environment variables / .env > defaults.

Environment Configuration

The SDK loads configuration from ACTIAN_VECTORAI_* environment variables and .env files automatically. Copy .env.example to .env to get started:

cp .env.example .env

Key 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_ACCESS_TOKEN	—	Access token for authentication
ACTIAN_VECTORAI_TLS	false	Enable TLS
ACTIAN_VECTORAI_TIMEOUT	30.0	Default per-RPC timeout (seconds)
ACTIAN_VECTORAI_MAX_RETRIES	3	Max retry attempts
ACTIAN_VECTORAI_POOL_SIZE	1	Connection pool size
ACTIAN_VECTORAI_LOG_LEVEL	WARNING	SDK log level

See .env.example for the complete list.

Access settings programmatically:

from actian_vectorai import Settings, settings

# Global singleton (lazy, auto-loaded)
print(settings.url)  # "localhost:6574"

# Custom instance with overrides
cfg = Settings(url="remote:6574", timeout=60.0)

API Overview

The client uses a namespaced architecture — operations are grouped by domain:

Namespace	Access	Description
Collections	client.collections	Create, list, delete, update collections
Points	client.points	CRUD, search, query, payload, indexing
VDE	client.vde	Engine lifecycle, rebuilds, compaction, import

Collection Management

client.collections.create("my_col", vectors_config=VectorParams(size=128, distance=Distance.Cosine))
client.collections.list()                     # -> ["my_col", ...]
client.collections.get_info("my_col")         # -> CollectionInfo
client.collections.exists("my_col")           # -> True
client.collections.delete("my_col")

Point Operations

# Upsert
client.points.upsert("col", [PointStruct(id=1, vector=[...], payload={...})])

# Get by ID
points = client.points.get("col", ids=[1, 2, 3])

# Delete
client.points.delete("col", ids=[1, 2])

# Count
n = client.points.count("col")

# Bulk upload with auto-batching
total = client.upload_points("col", points, batch_size=256)

Search

# Vector similarity search
results = client.points.search("col", vector=[...], limit=10)

# Filtered search
from actian_vectorai import Field, FilterBuilder
f = FilterBuilder().must(Field("price").lte(100.0)).build()
results = client.points.search("col", vector=[...], limit=10, filter=f)

# Batch search
batch = client.points.search_batch("col", [
    {"vector": query1, "limit": 5},
    {"vector": query2, "limit": 10},
])

Query & Advanced Search

# Universal query — nearest-neighbor search
results = client.points.query("col", query=[0.1] * 128, limit=10)

# Batch query
batched = client.points.query_batch("col", [
    {"query": vec1, "limit": 5},
    {"query": vec2, "limit": 10},
])

Scroll & Pagination

# Manual cursor-based pagination
points, next_offset = client.points.scroll("col", limit=100)

# Automatic async iteration over all matching points
async for point in client.points.scroll_all("col", limit=100):
    process(point)

VDE Operations

client.vde.open_collection("col")
client.vde.get_state("col")          # -> CollectionState.READY
client.vde.get_stats("col")          # -> CollectionStats
client.vde.flush("col")
client.vde.rebuild_index("col")
client.vde.compact_collection("col")
client.vde.close_collection("col")

Filter DSL

Type-safe filter building for payload queries:

from actian_vectorai import Field, FilterBuilder, has_id, is_empty

# Equality
f = FilterBuilder().must(Field("category").eq("electronics")).build()

# Range
f = FilterBuilder().must(Field("price").between(100.0, 500.0)).build()

# Combined conditions
f = (
    FilterBuilder()
    .must(Field("category").eq("electronics"))
    .must(Field("price").lte(500.0))
    .must_not(Field("deleted").eq(True))
    .build()
)

# Operator syntax
cond = Field("category").eq("electronics") & Field("price").lte(500.0)
f = cond.build()

# Use in search
results = client.points.search("products", vector=[...], limit=10, filter=f)

Hybrid Fusion

Combine results from multiple search queries:

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)

# Reciprocal Rank Fusion
fused = reciprocal_rank_fusion([dense, sparse], limit=10, weights=[0.7, 0.3])

# Distribution-Based Score Fusion
fused = distribution_based_score_fusion([dense, sparse], limit=10)

Testing

# Run all unit tests
python -m pytest tests/unit/ -v

# Run integration tests (requires running server)
python -m pytest tests/integration/ -v

# Run all tests
python -m pytest tests/ -v

# With coverage
python -m pytest tests/ --cov=actian_vectorai --cov-report=term-missing

Examples

38 runnable examples covering every feature of the SDK:

python examples/01_hello_world.py
python examples/02_async_hello_world.py
python examples/03_collection_management.py
python examples/04_point_crud.py
python examples/05_vector_search.py
python examples/06_filtered_search.py
python examples/07_payload_management.py
python examples/08_batch_upload.py
python examples/09_query_api.py
python examples/10_field_indexes.py
python examples/11_advanced_filters.py
python examples/12_search_params.py
python examples/13_rest_transport.py
python examples/14_resilience.py
python examples/15_hybrid_fusion.py
python examples/16_vde_operations.py
python examples/17_semantic_search.py
python examples/18_error_handling.py
python examples/19_tls_connection.py
python examples/20_connection_pool.py
python examples/21_search_batch.py
python examples/22_interceptors.py
python examples/23_delete_operations.py
python examples/24_async_concurrent.py
python examples/25_comprehensive_api.py
python examples/26_advanced_vde.py
python examples/27_exception_handling.py
python examples/28_uuid_point_ids.py
python examples/29_named_vectors.py
python examples/30_smart_batcher.py
python examples/31_delete_points.py
python examples/32_field_indexes.py
python examples/33_sparse_vectors.py
python examples/34_quantization.py
python examples/35_rag_integration.py
python examples/36_telemetry.py
python examples/37_scroll_pagination.py
python examples/38_openai_embeddings.py

Dependencies

Package	Version	Purpose
grpcio	≥ 1.80.0	gRPC transport
protobuf	≥ 5.29.2	Protocol buffer serialisation
pydantic	≥ 2.10.0	Data models and validation
numpy	≥ 1.26.0	Vector array operations

Documentation

• API Reference — Complete method signatures, parameters, types, models, enums, and exceptions
• Server Availability Status — What's working vs under server development
• Examples — 51 runnable examples covering every SDK feature

License

Proprietary — Actian Corporation

Current Status (v1.0.0)

43 SDK methods are fully available against Actian VectorAI DB v1.0.0. See the full Server Availability Status for details.

Verified on live server (Actian VectorAI DB 1.0.0 / VDE 1.0.0)

• has_id filters validated for both numeric IDs and UUID IDs
• Named-vector collections validated end-to-end
• Payload field indexes declared via payload_schema in extra_params_json at collection creation time
• Scroll-based pagination validated end-to-end
• Capacity limit errors (CapacityExceededError) properly handled as non-retryable
• Sparse-vector and multi-dense-vector write paths remain under server development

---

Copyright © 2025–2026 Actian Corporation. All Rights Reserved.