actian-vectorai-client 1.0.0.post1

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 — REST endpoint reference, schemas, and examples
• Python SDK Quickstart — Install, authenticate, and build your first client
• Academy — Tutorials and guided walkthroughs

License

Proprietary — Actian Corporation

---

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