Python API Reference

Functions

set_log_level(_level: str)

Set the engine log level.

Currently a no-op at runtime. Use the RUST_LOG environment variable before starting the engine (e.g. RUST_LOG=debug).

• level: One of "error", "warn", "info", "debug", "trace".

Returns: None

import os
os.environ["RUST_LOG"] = "debug"  # set before creating Engine
set_log_level("debug")  # no effect at runtime

repair_json(input, default=None) -> str

Repair a possibly-truncated or malformed JSON string.

The common case is a structured-generation response cut off by a token limit (finish_reason == "length"), e.g. {"a": 1, "b": "hel — valid up to the cut but unparseable as a whole. Completes it best-effort: keeps complete fields, closes and keeps a truncated string value, drops a dangling key with no value, and closes open objects/arrays. Never raises.

• input: the JSON string to repair.
• default: optional JSON string whose fields fill any missing/null fields in the result (fields present in input win). Pass None for no default.

Returns: the repaired JSON string.

Caveat: a value truncated mid-token is kept as-is. Repair is a safety net — prefer checking finish_reason and raising the token budget.

from atelico import repair_json
repair_json('{"a": 1, "b": "hel', '{"a": 0, "c": 3}')
**'{"a":1,"b":"hel","c":3}'**

RecommendedSampling

A model's advisory sampling parameters, if it declares any.

Each field is None when the model has no recommendation for that knob. The engine applies these only when the request omits the corresponding field (explicit request values always win); they are surfaced here for introspection/display. Mirrors the engine's nested recommended_sampling object — additive: unknown keys in the JSON are ignored.

Methods

from_dict(cls, d: dict[str, Any]) -> 'RecommendedSampling'

ModelCapabilities

A loaded model's static capabilities, as returned by :meth:Engine.model_capabilities.

Use :meth:from_json to parse the JSON string that method returns into a typed object. The mapping is additive: fields absent from an older engine's JSON fall back to their defaults, and unknown keys are ignored, so a newer engine never breaks an older binding.

• thinking_style: one of "unsupported", "thinktags", "native".
• device: e.g. "Cuda(0)" / "Cpu" / "Metal(0)" (empty on engines that predate the field).
• native_tool_format: e.g. "gemma4", or None when the model has no native tool block (tools are prose-injected).
• recommended_sampling: nested advisory sampling knobs (see :class:RecommendedSampling).

Methods

from_dict(cls, d: dict[str, Any]) -> 'ModelCapabilities'

from_json(cls, s: str) -> 'ModelCapabilities'

Parse the JSON string from :meth:Engine.model_capabilities.

Backend

Helper for constructing backend configuration JSON strings.

Each method returns a JSON string that can be passed to Engine configuration. Two backend types are supported:

• in_memory: Local on-device inference (Metal/CUDA/CPU).
• proxy: Forward requests to a remote OpenAI-compatible API.

local = Backend.in_memory(name="local")
remote = Backend.proxy(name="openai", api_key="sk-...")

Methods

@staticmethod in_memory(name="local", models=vec![], model_directory=None) -> str

Return a JSON configuration string for a local in-memory backend.

The returned JSON contains the backend name, type, list of models to pre-load, and an optional model cache directory.

• name: Logical name for this backend (default "local").
• models: List of HuggingFace model IDs to pre-load at startup.
• model_directory: Filesystem path where model weights are cached. Pass None to use the default cache.

Returns: A JSON string with the backend configuration.

{
  "name": "local",
  "backend_type": "in_memory",
  "models": ["meta-llama/Llama-3.2-1B-Instruct-GGUF"],
  "model_directory": "/models"
}

cfg = Backend.in_memory(
    name="local",
    models=["meta-llama/Llama-3.2-1B-Instruct-GGUF"],
    model_directory="/models",
)

@staticmethod proxy(name="openai", base_url="https://api.openai.com/v1", api_key=None) -> str

Return a JSON configuration string for a remote proxy backend.

The proxy backend forwards all requests to an OpenAI-compatible HTTP API.

• name: Logical name for this backend (default "openai").
• base_url: Base URL of the remote API (default "https://api.openai.com/v1").
• api_key: API key for authentication. Pass None if the remote API does not require one.

Returns: A JSON string with the backend configuration.

{
  "name": "openai",
  "backend_type": "proxy",
  "base_url": "https://api.openai.com/v1",
  "api_key": "sk-..."
}

cfg = Backend.proxy(name="openai", api_key="sk-...")

TokenStream

Iterator that yields streaming LLM chat completion chunks as JSON strings.

Implements Python's iterator protocol (__iter__ / __next__). Each call to __next__ blocks (releasing the GIL) until the next token chunk arrives from the inference engine.

Each yielded string is a JSON object following the OpenAI ChatCompletionChunk schema:

{
  "id": "chatcmpl-...",
  "object": "chat.completion.chunk",
  "model": "...",
  "choices": [{
    "index": 0,
    "delta": {"role": "assistant", "content": "token"},
    "finish_reason": null
  }]
}

The last chunk has finish_reason set to "stop" (or "length"), and after that __next__ raises StopIteration.

stream = engine.llm_chat_completion_stream(request_json)
for chunk_json in stream:
    chunk = json.loads(chunk_json)
    token = chunk["choices"][0]["delta"].get("content", "")
    print(token, end="", flush=True)

AudioSpeechChunkStream

Iterator that yields streaming TTS speech chunks as JSON strings.

Implements Python's iterator protocol (__iter__ / __next__). Each yielded string is an AudioSpeechChunk JSON object:

{
  "sequence": 0,
  "audio": "<base64 WAV bytes>",
  "duration_seconds": 1.42,
  "text": "First sentence."
}

__next__ raises StopIteration once the stream is exhausted.

stream = engine.audio_synthesize_stream(request_json)
for chunk_json in stream:
    chunk = json.loads(chunk_json)
    wav_bytes = base64.b64decode(chunk["audio"])
    # play wav_bytes as soon as it arrives

JsonResultStream

Handle to a non-blocking, non-streaming call started by one of the *_async methods. The full response is computed off the calling thread (on the engine's runtime) and delivered here exactly once as a JSON string.

Two ways to consume it: poll() — non-blocking. Returns the JSON string once ready, or None while still pending. Ideal for a game / UI loop that ticks each frame:

handle = engine.llm_chat_completion_async(request)
while True:
    result = handle.poll()
    if result is not None:
        response = json.loads(result)
        break
    do_one_frame_of_work()

result() — blocks (releasing the GIL) until the response is ready, then returns the JSON string. Equivalent to the synchronous call but the work runs on the engine runtime:

handle = engine.image_generate_async(request)
image = json.loads(handle.result())

Unlike *_stream (token-by-token deltas), this yields the COMPLETE response in one shot — the right shape for schema-constrained / structured output, which streaming can't serve cleanly. Backed by the same Engine::spawn_result primitive (one shared Rust codepath) the C ABI and Lua async bindings use.

Methods

poll() -> Optional[Strin]

Non-blocking poll. Returns the full JSON result string once the async call completes, or None if it is still in flight. Raises on failure.

result() -> str

Block (releasing the GIL) until the async call completes, then return the full JSON result string. Raises on failure.

TranscriptionStream

A live transcription session from Engine.audio_transcribe_stream.

Feed microphone audio with feed; read events with poll (non-blocking, for a game or UI loop) or by iterating (blocking). Each event is a JSON string tagged by type:

{"type": "speech_started", "start_time": 1.2}
{"type": "speech_ended", "start_time": 1.2, "end_time": 2.0}
{"type": "transcription", "text": "Stormfly, come", "is_final": true,
 "start_time": 1.2, "end_time": 2.0}

A failed transcription raises InferenceError from that one poll / next; the stream keeps listening. Call finish() to transcribe the utterance in progress and end the stream.

stream = engine.audio_transcribe_stream(json.dumps({"model": "in-memory::whisper"}))
stream.feed(samples, 48000)          # from your audio callback
while (event := stream.poll()) is not None:
    event = json.loads(event)
    if event["type"] == "transcription" and event["is_final"]:
        print(event["text"])

Methods

feed(samples: list[float], sample_rate: int)

Queue mono audio (floats in -1..1) at sample_rate Hz. Returns immediately, and is safe to call from an audio-callback thread while another thread reads events.

Raises: ValueError: If sample_rate is 0. AtelicoError: If the stream has been finished, or is more than 30 s of audio behind (poll its events so it can catch up).

stream.feed([0.0] * 480, 48000)

finish()

Stop listening: the utterance in progress is transcribed, then the stream ends once its remaining events are read.

stream.finish()
while not stream.done:
    stream.poll()

poll() -> Optional[Strin]

Next event as a JSON string, or None if none is ready (or the stream has ended — check done). Never blocks.

Raises: InferenceError: For one failed transcription; keep polling.

event = stream.poll()

done() -> bool

Whether the stream has ended (after finish, once every event was read).

while not stream.done:
    stream.poll()

Engine

The main Atelico AI Engine.

Thread-safe. Supports context manager protocol (with statement).

engine = Engine(device="auto")
engine.load_model("meta-llama/Llama-3.2-1B-Instruct-GGUF")
response = engine.llm_chat_completion(request_json)

Methods

__init__(config_json=None, device="auto", scheduling_mode="balance", vram_budget_mb=0, data_logging_json=None) -> Self

Create a new Atelico AI Engine instance.

Either pass a full config_json string, or use the individual keyword arguments for simple setups.

• config_json: Full engine configuration as a JSON string. When provided, all other arguments are ignored.
• device: Device selection -- "auto" (default, tries CUDA then Metal then CPU), "cpu", "metal", or "cuda".
• scheduling_mode: GPU scheduling mode when sharing with a game renderer -- "balance" (default), "prioritize_compute", or "prioritize_graphics".
• vram_budget_mb: Maximum VRAM the engine may use, in megabytes. 0 means unlimited (default).
• data_logging_json: Optional JSON object configuring asynchronous input/output capture. Ignored when config_json is supplied.

Returns: A new Engine instance configured with the specified device and settings.

Raises: AtelicoError: If engine initialization fails (e.g. device unavailable).

engine = Engine(device="metal")

**Context manager for automatic cleanup:**
with Engine(device="auto") as engine:
    engine.load_model("meta-llama/Llama-3.2-1B-Instruct-GGUF")

close()

Shut down the engine and release all resources.

Unloads every model, frees GPU memory, and stops background threads. After calling this, the engine instance must not be used again. Equivalent to exiting the context manager.

Returns: None

engine = Engine()
engine.close()

load_model(model_id: str)

Load a model into memory (blocking).

Downloads model weights from HuggingFace Hub if not already cached, then loads them onto the configured device. Blocks until fully loaded.

• model_id: HuggingFace model ID (e.g. "meta-llama/Llama-3.2-1B-Instruct-GGUF").

Returns: None

Raises: ModelLoadError: If the model cannot be found or loaded.

engine.load_model("meta-llama/Llama-3.2-1B-Instruct-GGUF")

load_model_async(model_id: str) -> JsonResultStream

Start a non-blocking model load — the "preload during a loading screen" primitive (first-use download + memory load run off the caller). Returns a JsonResultStream; on completion it delivers {"model_id": "...", "loaded": true}.

handle = engine.load_model_async("meta-llama/Llama-3.2-1B-Instruct-GGUF")
while handle.poll() is None:
    render_loading_screen_frame()

unload_model(model_id: str)

Unload a model and free its resources (GPU memory, caches).

• model_id: The model ID that was previously passed to load_model.

Returns: None

Raises: ModelLoadError: If the model is not currently loaded.

engine.unload_model("meta-llama/Llama-3.2-1B-Instruct-GGUF")

llm_chat_completion(request_json: str) -> str

Run a chat completion request (blocking, non-streaming).

• request_json: A JSON string following the OpenAI ChatCompletionRequest schema. Engine extensions include speculative, enable_thinking, priority, and prefix_cache ({"scope": "auto"} reuses the KV of a shared prompt prefix instead of re-prefilling it; byte-identical greedy output, lower prefill cost — ideal for a fixed system prompt across many calls).

  priority is "interactive", "normal" (the default) or "background", and decides how this request is scheduled against everything else competing for the GPU. A long generation hands the GPU over at token boundaries, so an "interactive" request does not wait for a "background" one to finish.

  seed sets the sampler's RNG. Omitting it does not randomise: the engine uses one fixed default seed, so identical requests repeat.

  adapter_scale (float, optional) is the strength of the LoRA adapter this request runs with (from the model::adapter suffix or the model's default): 1.0 as trained, 0.0 the base model, values between blend. Absent uses the model's default scale, else 1.0.

Expected request JSON structure:

{
  "model": "meta-llama/Llama-3.2-1B-Instruct-GGUF",
  "messages": [
    {"role": "system", "content": "You are helpful."},
    {"role": "user", "content": "Hello!"}
  ],
  "max_tokens": 128,
  "temperature": 0.7,
  "top_p": 0.9,
  "response_format": null,
  "speculative": {"proposer": "prompt_lookup", "k": 5}
}

speculative is optional — multi-token (speculative) decoding to speed up generation (proposer: "prompt_lookup" or "lookahead"). Greedy output is unchanged.

Returns: A JSON string with the ChatCompletionResponse.

{
  "id": "chatcmpl-...",
  "choices": [{
    "message": {"role": "assistant", "content": "Hi!"},
    "finish_reason": "stop"
  }],
  "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8}
}

Raises: ValueError: If request_json is not valid JSON or is missing required fields. InferenceError: If token generation fails.

import json
request = json.dumps({
    "model": "meta-llama/Llama-3.2-1B-Instruct-GGUF",
    "messages": [{"role": "user", "content": "Hello!"}],
    "max_tokens": 128,
})
response = json.loads(engine.llm_chat_completion(request))
print(response["choices"][0]["message"]["content"])

llm_chat_completion_stream(request_json: str) -> TokenStream

Start a streaming chat completion and return a TokenStream iterator.

Each iteration yields a JSON string with one ChatCompletionChunk (see TokenStream for the chunk schema). The GIL is released while waiting for each chunk, so other Python threads can run.

• request_json: JSON ChatCompletionRequest (same schema as llm_chat_completion). The stream field is set automatically.

Returns: A TokenStream iterator that yields JSON chunk strings per token.

Raises: ValueError: If request_json is invalid. InferenceError: If stream setup fails.

import json
request = json.dumps({
    "model": "meta-llama/Llama-3.2-1B-Instruct-GGUF",
    "messages": [{"role": "user", "content": "Tell me a joke"}],
})
for chunk_json in engine.llm_chat_completion_stream(request):
    chunk = json.loads(chunk_json)
    token = chunk["choices"][0]["delta"].get("content", "")
    print(token, end="", flush=True)

llm_chat_completion_async(request_json: str) -> JsonResultStream

Start a non-blocking, non-streaming chat completion.

Returns immediately with a JsonResultStream handle; the full ChatCompletionResponse is computed on the engine runtime and delivered once (poll for it, or block with .result()). Unlike llm_chat_completion_stream it yields the COMPLETE response — the right shape for schema-constrained / structured output. Same request schema as llm_chat_completion.

handle = engine.llm_chat_completion_async(request)
**... do other work ...**
response = json.loads(handle.result())

llm_text_completion_async(request_json: str) -> JsonResultStream

Start a non-blocking, non-streaming text completion. See llm_chat_completion_async; same request schema as llm_text_completion.

llm_respond_async(request_json: str) -> JsonResultStream

Start a non-blocking, non-streaming Responses-API request. See llm_chat_completion_async; same request schema as llm_respond.

llm_text_completion(request_json: str) -> str

Run a text completion request (blocking, non-streaming).

Unlike chat completion, text completion continues a raw prompt without chat template formatting.

• request_json: A JSON string with TextCompletionRequest fields.

Expected request JSON structure:

{
  "model": "meta-llama/Llama-3.2-1B-Instruct-GGUF",
  "prompt": "Once upon a time",
  "max_tokens": 50,
  "temperature": 0.7
}

Returns: A JSON string with the TextCompletionResponse.

{
  "id": "cmpl-...",
  "object": "text_completion",
  "model": "...",
  "choices": [{"text": " there was a...", "index": 0, "finish_reason": "stop"}],
  "usage": {"prompt_tokens": 4, "completion_tokens": 50, "total_tokens": 54}
}

Raises: ValueError: If request_json is invalid. InferenceError: If token generation fails.

import json
request = json.dumps({
    "model": "meta-llama/Llama-3.2-1B-Instruct-GGUF",
    "prompt": "Once upon a time",
    "max_tokens": 50,
})
response = json.loads(engine.llm_text_completion(request))
print(response["choices"][0]["text"])

llm_respond(request_json: str) -> str

Run a Responses API request (blocking).

The Responses API is a higher-level conversational interface that manages conversation state internally.

• request_json: A JSON string with ResponseRequest fields.

Expected request JSON structure:

{
  "model": "meta-llama/Llama-3.2-1B-Instruct-GGUF",
  "input": "What is 2+2?",
  "instructions": "You are a math tutor.",
  "max_output_tokens": 100,
  "temperature": 0.7
}

Returns: A JSON string with the ResponseResponse.

{
  "id": "resp-...",
  "object": "response",
  "output": [{"type": "message", "content": [{"type": "output_text", "text": "4"}]}],
  "usage": {"input_tokens": 5, "output_tokens": 1}
}

Raises: ValueError: If request_json is invalid. InferenceError: If generation fails.

import json
request = json.dumps({
    "model": "meta-llama/Llama-3.2-1B-Instruct-GGUF",
    "input": "What is 2+2?",
})
response = json.loads(engine.llm_respond(request))

tokenize(model_id: str, text: str) -> list[u3]

Encode text into token IDs using a loaded model's tokenizer.

Loads the model on demand if it isn't already resident. Useful for budgeting prompt length against the context window before sending a request, or for pre-tokenizing prefixes that will be reused.

• model_id: Model identifier in "backend::model" format.
• text: UTF-8 text to tokenize.

Returns: List of token IDs (list[int]).

Raises: AtelicoError if the model fails to load or tokenize.

tokens = engine.tokenize(
    "in-memory::meta-llama/Llama-3.2-3B-Instruct-Q4_K_M",
    "Hello, world!",
)
print(len(tokens), tokens[:5])

detokenize(model_id, tokens, skip_special_tokens=true) -> str

Decode a token ID sequence back into text using a loaded model's tokenizer.

• model_id: Model identifier in "backend::model" format.
• tokens: Token ID sequence as list[int].
• skip_special_tokens: If True (default), drops tokens like <|begin_of_text|> from the output.

Returns: The decoded text (str).

Raises: AtelicoError if the model fails to load or detokenize.

text = engine.detokenize(
    "in-memory::meta-llama/Llama-3.2-3B-Instruct-Q4_K_M",
    [128000, 9906, 11, 1917, 0],
    skip_special_tokens=True,
)
**text == "Hello, world!"**

model_capabilities(model_id: str) -> str

Inspect a loaded model's static capabilities.

Returns vocab size, context window, dtype, and active backend (metal/cuda/cpu) for budgeting, telemetry, and debugging. Loads the model on demand.

• model_id: Model identifier in "backend::model" format.

Returns: A JSON string with the ModelCapabilities:

{
  "model_id": "in-memory::meta-llama/Llama-3.2-3B-Instruct-Q4_K_M",
  "vocab_size": 128256,
  "max_position_embeddings": 131072,
  "dtype": "Q4_K_M",
  "device": "Cuda(0)",
  "backend_kind": "metal",
  "thinking_style": "native",
  "native_tool_format": "gemma4",
  "recommended_sampling": { "temperature": 1.0, "top_p": 0.95 }
}

device is the device the weights are on ("Cuda(0)"/"Cpu"/"Metal(0)"; empty on older engines). thinking_style is one of "unsupported"/"thinktags"/"native". native_tool_format (e.g. "gemma4") and each recommended_sampling knob (temperature, top_p, repetition_penalty) are omitted when the model declares none. The typed atelico.ModelCapabilities.from_json(...) dataclass wraps this string.

import atelico
caps = atelico.ModelCapabilities.from_json(
    engine.model_capabilities("in-memory::meta-llama/Llama-3.2-3B-Instruct-Q4_K_M")
)
print(f"context window = {caps.max_position_embeddings} tokens")

llm_call_function_with_policy(function_json, model_id, variables_json, system_prompt, policy_json) -> str

Execute a templated LM function under a generation policy (blocking). Validators, repair, retries, and fallbacks are layered on top of the simple llm_call_function path.

• function_json: JSON LM function definition (same shape as llm_call_function).

• model_id: Model ID in "backend::model" format.

• variables_json: JSON object with template variable values.

• system_prompt: Optional system message. Pass None to omit.

• policy_json: JSON generation policy. All fields optional:

  {
    "max_attempts": 3,
    "validators": [
      {"type": "json_parse"},
      {"type": "json_schema_subset", "schema": {"type": "object"}},
      {"type": "regex", "pattern": "^[A-Z].*"},
      {"type": "choice", "options": ["yes", "no"]}
    ],
    "repair": "local_json_repair",
    "fallbacks": [
      {"type": "model", "model": "in-memory::stronger-model"},
      {"type": "static", "content": "{\"label\": \"unknown\"}"}
    ],
    "guardrails": "off",
    "timeout_ms": 30000
  }

  repair: "none", "local_json_repair", "retry_with_error_message", "retry_with_original_prompt_and_error".

  guardrails: "off", "monitor", "strict", "retry_on_block". Function-level guardrails require a Rust-side handle and are a no-op when invoked through Python — only router-level guardrails apply.

  Fallback type: "none", "prompt" (with "template"), "model" (with "model"), "generation" (with "max_tokens" and/or "temperature"), "static" (with "content" and optional "parsed"), "cache_only".

Returns: A JSON string with the result and trace:

{
  "content": "...",
  "parsed":  null,
  "trace": {
    "attempts": [{"attempt": 1, "model": "...", "outcome": "Pass"}],
    "fallback_used": null,
    "fallback_kind": null,
    "final_from_fallback": false,
    "cache_status": null
  }
}

image_generate(request_json: str) -> str

Generate an image from a text prompt (blocking).

• request_json: A JSON string with ImageGenerationRequest fields.

Expected request JSON structure:

{
  "model": "PixArt-alpha/PixArt-Sigma-XL-2-512-MS",
  "prompt": "A sunset over mountains",
  "n": 1,
  "size": "512x512",
  "response_format": "b64_json"
}

Returns: A JSON string with the ImageGenerationResponse.

{
  "created": 1234567890,
  "data": [{"b64_json": "iVBORw0KGgo...", "revised_prompt": "A sunset over mountains"}]
}

Raises: ValueError: If request_json is invalid. InferenceError: If image generation fails.

import json
request = json.dumps({
    "model": "PixArt-alpha/PixArt-Sigma-XL-2-512-MS",
    "prompt": "A sunset over mountains",
    "size": "512x512",
})
response = json.loads(engine.image_generate(request))
b64_image = response["data"][0]["b64_json"]

video_generate(request_json: str) -> str

Generate a video from a text prompt (blocking).

• request_json: A JSON string with VideoGenerationRequest fields (model, prompt, optional num_frames, width, height, fps, seed, response_format = "b64_mp4" or "b64_raw_nhwc").

Returns: A JSON string with the response (data, num_frames, format, ...).

Raises: ValueError: If request_json is invalid JSON. InferenceError: If video generation fails.

image_remove_background(request_json: str) -> str

Remove the background from an image (blocking).

• request_json: A JSON string with BackgroundRemovalRequest fields.

Expected request JSON structure:

{
  "model": "briaai/RMBG-1.4",
  "image": "iVBORw0KGgo..."
}

Returns: A JSON string with the processed image.

{
  "data": [{"b64_json": "iVBORw0KGgo..."}]
}

Raises: ValueError: If request_json is invalid. InferenceError: If background removal fails.

import json
request = json.dumps({"model": "briaai/RMBG-1.4", "image": b64_image_str})
response = json.loads(engine.image_remove_background(request))
result_image = response["data"][0]["b64_json"]

image_generate_async(request_json: str) -> JsonResultStream

Start a non-blocking image generation. Returns a JsonResultStream; generation runs on the engine runtime (seconds-long) without blocking the caller. Same request schema as image_generate. See llm_chat_completion_async for the poll/result pattern.

image_remove_background_async(request_json: str) -> JsonResultStream

Start a non-blocking background removal. Returns a JsonResultStream. Same request schema as image_remove_background.

audio_synthesize(request_json: str) -> str

Synthesize speech from text (blocking).

• request_json: A JSON string with AudioSpeechRequest fields.

Expected request JSON structure:

{
  "model": "in-memory::tts",
  "input": "Hello from Atelico.",
  "voice": "af_heart"
}

Accepted model ids after the in-memory:: prefix: tts, kokoro, kokoro-82m, pocket, pocket-tts. Default is kokoro-82m.

Returns: A JSON string with the synthesized audio:

{
  "audio_b64": "UklGRn...",
  "duration_seconds": 1.42,
  "format": "wav",
  "sample_rate": 24000
}

audio_b64 decodes to a complete WAV file.

import base64, json
req = json.dumps({
    "model": "in-memory::tts",
    "input": "Hello from Atelico.",
    "voice": "af_heart",
})
resp = json.loads(engine.audio_synthesize(req))
open("hello.wav", "wb").write(base64.b64decode(resp["audio_b64"]))

audio_transcribe(request_json: str) -> str

Transcribe audio to text (blocking).

• request_json: A JSON string with the transcription request:

{
  "model": "in-memory::whisper",
  "audio_b64": "UklGRn...",
  "language": "en"
}

audio_b64 must be a base64-encoded WAV file (RIFF header + PCM data); the binding decodes the WAV and reads the sample rate from the header. language is optional (auto-detect if omitted).

Accepted model ids after the in-memory:: prefix: whisper (default — resolves to whisper-base.en), whisper-tiny[.en], whisper-base[.en], whisper-small[.en], whisper-medium[.en], whisper-large-v3[-turbo], distil-large-v3.

Returns: A JSON string AudioTranscriptionResponse:

{"text": "Hello from Atelico.", "language": "en", "duration": 1.4}

import base64, json
req = json.dumps({
    "model": "in-memory::whisper",
    "audio_b64": base64.b64encode(open("speech.wav", "rb").read()).decode(),
})
resp = json.loads(engine.audio_transcribe(req))
print(resp["text"])

audio_synthesize_stream(request_json: str) -> AudioSpeechChunkStream

Start a streaming TTS synthesis and return an iterator of chunks.

• request_json: A JSON AudioSpeechRequest (same schema as audio_synthesize).

Returns: An AudioSpeechChunkStream that yields JSON AudioSpeechChunk strings, one per sentence/clause.

import base64, json
stream = engine.audio_synthesize_stream(json.dumps({
    "model": "in-memory::pocket-tts",
    "input": "First sentence. Second one comes right after.",
    "voice": "alba",
}))
for chunk_json in stream:
    chunk = json.loads(chunk_json)
    wav_bytes = base64.b64decode(chunk["audio"])
    # play wav_bytes through your audio backend immediately

audio_transcribe_stream(config_json: str) -> TranscriptionStream

Open a live transcription stream for microphone audio.

• config_json: A JSON TranscriptionStreamConfig: \{"model": "in-memory::whisper", "endpointing": "command" | "dialogue" | "dictation", "energy_threshold": 0.02, "language": "en", "partial_interval_ms": 700\} — only model is required.

Returns: A TranscriptionStream to feed audio into and poll events from. Utterances are found by voice-activity detection and transcribed through the engine's Whisper as they end.

Raises: ValueError: If config_json is malformed or invalid.

stream = engine.audio_transcribe_stream(json.dumps({
    "model": "in-memory::whisper",
    "endpointing": "command",
}))

audio_synthesize_async(request_json: str) -> JsonResultStream

Start a non-blocking full-clip TTS synthesis. Returns a JsonResultStream; the complete clip is delivered once as the same {audio_b64, duration_seconds, format, sample_rate} JSON that audio_synthesize returns. Same request schema as audio_synthesize. For per-sentence streaming use audio_synthesize_stream instead.

audio_transcribe_async(request_json: str) -> JsonResultStream

Start a non-blocking transcription. Returns a JsonResultStream delivering the AudioTranscriptionResponse JSON. Same request schema (base64 WAV via audio_b64) as audio_transcribe.

embed_text(request_json: str) -> str

Generate text embeddings (blocking).

• request_json: A JSON string with EmbeddingRequest fields.

Expected request JSON structure:

{
  "model": "sentence-transformers/all-MiniLM-L6-v2",
  "input": ["Hello world", "Goodbye world"]
}

Returns: A JSON string with the EmbeddingResponse.

{
  "object": "list",
  "model": "sentence-transformers/all-MiniLM-L6-v2",
  "data": [
    {"object": "embedding", "index": 0, "embedding": [0.1, -0.2, ...]},
    {"object": "embedding", "index": 1, "embedding": [0.3, 0.4, ...]}
  ],
  "usage": {"prompt_tokens": 8, "total_tokens": 8}
}

Raises: ValueError: If request_json is invalid. InferenceError: If embedding generation fails.

import json
request = json.dumps({
    "model": "sentence-transformers/all-MiniLM-L6-v2",
    "input": ["Hello world", "Goodbye world"],
})
response = json.loads(engine.embed_text(request))
vectors = [d["embedding"] for d in response["data"]]

embed_async(request_json: str) -> JsonResultStream

Start a non-blocking embedding request. Returns a JsonResultStream delivering the EmbeddingResponse JSON. Same request schema as embed_text.

embed_image(request_json: str) -> str

Embed a single image into a vector with a cross-modal (CLIP/SigLIP) or vision (DINOv2) model. With a cross-modal model the vector shares the text model's space, so a text query can retrieve images by similarity.

• request_json: JSON with model (e.g. "clip", "siglip", "dinov2-small") and image (base64-encoded PNG/JPEG bytes).

Returns: EmbeddingResponse JSON with one data entry.

import json, base64
req = json.dumps({"model": "clip", "image": base64.b64encode(open("cat.png","rb").read()).decode()})
vec = json.loads(engine.embed_image(req))["data"][0]["embedding"]

embed_image_async(request_json: str) -> JsonResultStream

Start a non-blocking image-embedding request. Returns a JsonResultStream delivering the EmbeddingResponse JSON. Same request schema as embed_image.

embed_audio(request_json: str) -> str

Embed a single audio clip into a vector with the CLAP model. The vector shares CLAP's text space, so a text query like "dog barking" retrieves matching audio by similarity.

• request_json: JSON with model ("clap") and audio (base64-encoded WAV bytes).

Returns: EmbeddingResponse JSON with one data entry.

import json, base64
req = json.dumps({"model": "clap", "audio": base64.b64encode(open("bark.wav","rb").read()).decode()})
vec = json.loads(engine.embed_audio(req))["data"][0]["embedding"]

embed_audio_async(request_json: str) -> JsonResultStream

Start a non-blocking audio-embedding request. Returns a JsonResultStream delivering the EmbeddingResponse JSON. Same request schema as embed_audio.

embed_similarity(model_id: str, text_a: str, text_b: str) -> float

Compute cosine similarity between two texts using an embedding model.

A convenience method that embeds both texts and returns their cosine similarity as a float in the range -1.0 to 1.0.

• model_id: Embedding model to use (e.g. "sentence-transformers/all-MiniLM-L6-v2").
• text_a: First text.
• text_b: Second text.

Returns: Cosine similarity score (1.0 = identical, 0.0 = orthogonal, -1.0 = opposite).

Raises: InferenceError: If embedding generation fails.

score = engine.embed_similarity(
    "sentence-transformers/all-MiniLM-L6-v2",
    "Hello world",
    "Hi there",
)
**score is approximately 0.72 (varies by model)**

classifier_predict_text(classifier_id: str, text: str, top_k: Optional[int]) -> str

Predict the class of a text string using a loaded classifier.

• classifier_id: ID of a previously loaded classifier.
• text: The input text to classify.
• top_k: Number of top predictions to return. Pass None for all classes.

Returns: A JSON string with the prediction result.

{
  "label": "positive",
  "probability": 0.95,
  "top": [
    {"label": "positive", "probability": 0.95},
    {"label": "neutral", "probability": 0.04}
  ]
}

Raises: AtelicoError: If the classifier is not loaded or prediction fails.

import json
result = json.loads(engine.classifier_predict_text("sentiment", "I love this!", top_k=3))
print(result["label"], result["probability"])

classifier_predict_async(classifier_id: str, text: str, top_k: Optional[int]) -> JsonResultStream

Start a non-blocking text classification. Returns a JsonResultStream delivering the prediction JSON (same schema as classifier_predict).

classifier_predict_image(classifier_id: str, image_path: str, top_k: Optional[int]) -> str

Predict the class of an image using a multi-modal classifier.

The classifier referenced by classifier_id must have been trained with an image embedder (e.g. DINOv2). Calling this against a text-only classifier raises AtelicoError.

• classifier_id: ID of a previously loaded image classifier.
• image_path: Filesystem path to the image (PNG / JPEG).
• top_k: Number of top predictions to return. Pass None for all classes.

Returns: A JSON string with the prediction result (same schema as classifier_predict).

import json
result = json.loads(engine.classifier_predict_image(
    "animals",
    "/abs/path/to/cat.jpg",
    top_k=3,
))
print(result["label"], result["probability"])

classifier_predict_audio(classifier_id: str, audio_path: str, top_k: Optional[int]) -> str

Predict the class of an audio clip using a multi-modal classifier.

The classifier referenced by classifier_id must have been trained with an audio embedder (e.g. CLAP). Calling this against a classifier trained on a different modality raises AtelicoError.

• classifier_id: ID of a previously loaded audio classifier.
• audio_path: Filesystem path to the audio clip (WAV).
• top_k: Number of top predictions to return. Pass None for all classes.

Returns: A JSON string with the prediction result (same schema as classifier_predict_text).

import json
result = json.loads(engine.classifier_predict_audio(
    "sounds",
    "/abs/path/to/bark.wav",
    top_k=3,
))
print(result["label"], result["probability"])

classifier_load(classifier_id: str, directory: str)

Load a classifier from a directory so it can serve predictions.

The directory must contain a classifier exported by the engine (the classifier kind plus, for embedding-based kinds, its embedder reference). Every kind is handled -- centroid, KNN/HNSW, SetFit, and the embedder-free n-gram perceptron, which loads offline from a bundled directory with no model download. Once loaded, the classifier is addressable by classifier_id from the classifier_predict_* methods.

• classifier_id: ID to register the loaded classifier under.
• directory: Filesystem path to the classifier directory.

Raises: AtelicoError: If the directory is missing or the classifier fails to load.

engine.classifier_load("facing_lr", "/abs/path/to/facing_lr")
result = engine.classifier_predict_image("facing_lr", "/abs/path/to/creature.png")

classifier_create(classifier_id: str, config_json: str)

Create an empty classifier to teach during play with classifier_train_incremental_text.

The classifier starts with no labels; each label is created by its first example. Persist it with classifier_save and restore it with classifier_load.

• classifier_id: ID to register it under; must not already exist.
• config_json: JSON object with kind ("centroid", "knn" or "ngram_perceptron"), embedder (embedding model id, required for centroid/knn, absent for ngram_perceptron; weights must be cached), optional multilabel (bool) and k_neighbors (knn only, default 7).

Raises: ValueError: If the config is malformed, does not fit the kind, or the id is taken.

import json
engine.classifier_create("dragon_cues", json.dumps({"kind": "centroid", "embedder": "all-MiniLM-L6-v2"}))

classifier_train_incremental_text(classifier_id: str, label: str, text: str)

Teach a classifier one example while it runs: folds text into label, creating the label if it is new. No retraining pass.

• classifier_id: a created or loaded classifier.
• label: the example's label (non-empty).
• text: the example.

Raises: ValueError: On an empty label. AtelicoError: If the classifier cannot learn incrementally (see classifier_info).

engine.classifier_train_incremental_text("dragon_cues", "come", "Stormfly, come")

classifier_train_incremental_embedding(classifier_id: str, label: str, embedding: list[float])

Teach one example from an embedding you already computed, skipping a second embedding of the same text. The vector must come from the classifier's own embedder model and have embedding_dim entries (see classifier_info).

Raises: ValueError: On a wrong-length vector, or a classifier with no embedding space.

vector = json.loads(engine.embed(...))["data"][0]["embedding"]
engine.classifier_train_incremental_embedding("dragon_cues", "come", vector)

classifier_predict_embedding(classifier_id, embedding, top_k=None) -> str

Classify an embedding you already computed. Same rules for the vector as classifier_train_incremental_embedding; returns the same JSON as classifier_predict_text.

result = json.loads(engine.classifier_predict_embedding("dragon_cues", vector, top_k=3))
print(result["label"])

classifier_remove_label(classifier_id: str, label: str)

Forget a label and everything learned for it.

Raises: ValueError: If the label does not exist.

engine.classifier_remove_label("dragon_cues", "spin")

classifier_info(classifier_id: str) -> str

Describe a classifier: kind, labels and examples per label, as JSON.

{"classifier_id": "dragon_cues", "kind": "centroid", "modality": "text",
 "embedding_dim": 384, "supports_incremental_training": true,
 "labels": [{"label": "come", "example_count": 6}]}

embedding_dim is null for classifiers without an embedding space; example_count is null for classifiers saved before counts were recorded.

info = json.loads(engine.classifier_info("dragon_cues"))
for entry in info["labels"]:
    print(entry["label"], entry["example_count"])

classifier_save(classifier_id: str, directory: str)

Save a classifier to a directory, in the format classifier_load reads back. The directory is created if missing; a previous save is replaced.

engine.classifier_save("dragon_cues", "/saves/slot1/dragon_cues")

guardrail_check_input(text: str) -> str

Check user input text against safety guardrails.

• text: The user-provided input text to validate.

Returns: A JSON string with the SafetyVerdict.

When allowed:

{"action": "Allow", "checker_name": "keyword", "score": null}

When blocked:

{"action": {"Block": {"reason": "profanity detected"}}, "checker_name": "keyword", "score": 0.95}

Fields:

• action: "Allow", or {"Block": {"reason": "..."}}, or {"Rewrite": {"original": "...", "rewritten": "...", "reason": "..."}}.
• checker_name: which guardrail checker produced the verdict.
• score: confidence score (0.0-1.0), or null.

Raises: AtelicoError if guardrails are not configured.

import json
verdict = json.loads(engine.guardrail_check_input("Tell me a joke"))
if verdict["action"] != "Allow":
    print(f"Blocked by {verdict['checker_name']}")

guardrail_check_output(text: str) -> str

Check model output text against safety guardrails.

• text: The model-generated output to validate before displaying.

Returns: A JSON string with the SafetyVerdict (same schema as guardrail_check_input).

Raises: AtelicoError: If guardrails are not configured.

import json
verdict = json.loads(engine.guardrail_check_output("Here is a helpful answer."))
if not verdict["allowed"]:
    print(f"Blocked: {verdict['category']}")

lora_load(model_id: str, adapter_path: str)

Make an adapter the default for a model: requests for model_id without a ::adapter suffix run with it from the next request on. A request can still pick its own adapter ("model::adapter") and strength ("adapter_scale").

• model_id: The base model ID.
• adapter_path: An asset-store adapter id ("pirate"), or a local adapter directory written as a path — absolute, or starting with ./ / ../ — holding adapter_config.json and adapter_model.safetensors.

Returns: None

Raises: ModelLoadError: If the adapter cannot be resolved.

engine.lora_load("meta-llama/Llama-3.2-1B-Instruct-GGUF", "./adapters/my-lora")

lora_unload(model_id: str)

Clear a model's default adapter and strength: requests without a suffix run on the base weights again.

• model_id: The base model ID.

Returns: None

engine.lora_unload("meta-llama/Llama-3.2-1B-Instruct-GGUF")

lora_set_scale(model_id: str, scale: float)

Set the strength a model's adapter runs at, for requests without their own adapter_scale: 1.0 as trained, 0.0 the base model, values between blend.

• model_id: The base model ID.
• scale: Finite and non-negative; above 1.0 exaggerates the adapter.

Returns: None

Raises: ValueError: For a negative or non-finite scale.

engine.lora_set_scale("meta-llama/Llama-3.2-1B-Instruct-GGUF", 0.5)

lora_list_active() -> str

List the models that have a default adapter.

Returns: A JSON array of {"model_id": ..., "adapter": ...} objects, [] when no model has one.

import json
for entry in json.loads(engine.lora_list_active()):
    print(entry["model_id"], "->", entry["adapter"])
**meta-llama/Llama-3.2-1B-Instruct-GGUF -> ./adapters/my-lora**

guardrail_check_image_prompt(text: str) -> str

Check an image generation prompt against safety guardrails.

• text: The image generation prompt to validate.

Returns: A JSON string with the SafetyVerdict (same schema as guardrail_check_input).

Raises: AtelicoError: If guardrails are not configured.

import json
verdict = json.loads(engine.guardrail_check_image_prompt("A sunset over mountains"))
if not verdict["allowed"]:
    print(f"Blocked: {verdict['category']}")

model_list() -> str

List all loaded models.

Returns: A JSON array of ModelInfo objects.

[
  {"id": "meta-llama/Llama-3.2-1B-Instruct-GGUF", "object": "model", "owned_by": "meta-llama"}
]

import json
models = json.loads(engine.model_list())
for m in models:
    print(m["id"])

model_is_loaded(model_id: str) -> bool

Check if a model is loaded and ready for inference.

• model_id: The model ID to check.

Returns: True if the model is loaded and ready, False otherwise.

is_ready = engine.model_is_loaded("meta-llama/Llama-3.2-1B-Instruct-GGUF")

kvstore_create(store_id: str, db_path: str, config_json: str) -> int

Create a new semantic key-value store backed by LanceDB.

• store_id: Unique identifier for this store instance.
• db_path: Filesystem path for the LanceDB database directory.
• config_json: A JSON string with KvStoreConfig fields.

Expected config JSON structure:

{
  "table_name": "entries",
  "embed_dim": 384,
  "has_priority": false,
  "similarity_weight": 0.5,
  "priority_weight": 0.5
}

Returns: The store handle ID (int).

Raises: ValueError: If config_json is invalid. StoreError: If store creation fails.

import json
config = json.dumps({"table_name": "game_lore", "embed_dim": 384})
handle = engine.kvstore_create("lore", "/data/lore_db", config)

kvstore_insert(store_id: str, entries_json: str)

Insert entries into a KV store.

• store_id: The store to insert into.
• entries_json: A JSON array of KvEntry objects.

Expected entries JSON structure (each object is a KvEntry):

[
  {
    "id": "greeting-1",
    "key_text": "how do I greet the innkeeper?",
    "key_embedding": [0.1, 0.2, 0.3],
    "facets": {"location": "tavern"},
    "values": {"answer": {"Utf8": "Hello!"}, "mood": {"Int32": 2}},
    "priority": 0.5
  }
]

key_embedding must be embed_dim long. facets and values are required (use {} when the store has none); every facet the store was created with must be present. Each value is tagged with its column type (Utf8, Int32, Float32). priority is optional.

Returns: None

Raises: ValueError: If entries_json is invalid. StoreError: If the store is not found or insertion fails.

import json
entries = json.dumps([{
    "id": "greeting-1",
    "key_text": "how do I greet the innkeeper?",
    "key_embedding": [0.1, 0.2, 0.3],
    "facets": {},
    "values": {"answer": {"Utf8": "Hello!"}},
}])
engine.kvstore_insert("lore", entries)

kvstore_query(store_id: str, query_json: str) -> str

Query a KV store using vector similarity search.

• store_id: The store to query.
• query_json: A JSON string with KvQuery fields.

Expected query JSON structure:

{
  "query_embedding": [0.1, 0.2, 0.3],
  "query_text": "optional text filter",
  "limit": 5,
  "vector_search_limit": 20,
  "use_prefilter": true
}

Returns: A JSON array of result objects.

[
  {
    "id": "...",
    "key_text": "greeting",
    "similarity": 0.95,
    "priority": 1.0,
    "combined_score": 0.975
  }
]

Raises: ValueError: If query_json is invalid. StoreError: If the store is not found or the query fails.

import json
query = json.dumps({"query_embedding": embedding_vec, "limit": 5})
results = json.loads(engine.kvstore_query("lore", query))
for r in results:
    print(r["key_text"], r["similarity"])

kvstore_create_fts_index(store_id: str, column: str)

Create a full-text-search (FTS) index on a string column of a KV store.

Required before calling kvstore_lexical_query or kvstore_hybrid_query. Typical column choices are "key_text" (the search keys) or any string-typed value column.

• store_id: The store to index.
• column: Column name to index (string-typed).

Raises: StoreError if the store is not found.

engine.kvstore_create_fts_index("lore", "key_text")

kvstore_lexical_query(store_id: str, query_json: str) -> str

Lexical (full-text) retrieval against an FTS-indexed column.

The target column must have an FTS index — see kvstore_create_fts_index. Returns rows ranked by BM25 relevance.

• store_id: The store to query.

• query_json: JSON LexicalSearchQuery:

  {"text": "dragon attack", "column": "key_text", "filter": null, "limit": 10}

Returns: A JSON array of KvResult rows. The combined_score field carries the BM25 score.

Raises: StoreError if the store is not found, or InternalError if the column does not have an FTS index.

import json
rows = json.loads(engine.kvstore_lexical_query(
    "lore",
    json.dumps({"text": "dragon", "column": "key_text", "limit": 5}),
))
for r in rows: print(r["key_text"], r["combined_score"])

kvstore_hybrid_query(store_id: str, query_json: str) -> str

Hybrid (vector + lexical) retrieval with weighted-sum reranking.

Each branch returns up to effective_per_branch_limit candidates; scores are min-max normalised independently and combined as vector_weight * vec_norm + lexical_weight * lex_norm. The FTS column must have an FTS index.

• store_id: The store to query.

• query_json: JSON HybridSearchQuery:

  {
    "embeddings": [0.0, 0.1, 0.2],
    "text": "dragon attack",
    "fts_column": "key_text",
    "filter": null,
    "limit": 5,
    "per_branch_limit": 0,
    "vector_weight": 0.5,
    "lexical_weight": 0.5
  }

  per_branch_limit: 0 defaults to 2 * limit.

Returns: A JSON object:

{
  "results": [{ ...KvResult... }],
  "scores":  [{"vector_score": 0.8, "lexical_score": 0.6, "merged_score": 0.7}]
}

combined_score on each result equals the corresponding merged_score.

kvstore_scan(store_id, filter=None, limit=100) -> str

Scan entries in a KV store with an optional SQL-like filter.

Unlike kvstore_query, this does not use vector similarity -- it returns entries matching the filter, ordered by insertion.

• store_id: The store to scan.
• filter: SQL-like filter expression (e.g. "category = 'lore'"). Pass None to return all entries.
• limit: Maximum number of entries to return (default 100).

Returns: A JSON array of KvEntry objects.

[
  {"key": "greeting", "value": "Hello!", "embedding": [0.1, 0.2, 0.3]}
]

Raises: StoreError: If the store is not found.

import json
entries = json.loads(engine.kvstore_scan("lore", limit=10))
for entry in entries:
    print(entry["key"], entry["value"])

kvstore_count(store_id, filter=None) -> int

Count entries in a KV store, optionally filtered.

• store_id: The store to count entries in.
• filter: SQL-like filter expression. Pass None to count all entries.

Returns: The number of matching entries (int).

Raises: StoreError: If the store is not found.

count = engine.kvstore_count("lore")
filtered = engine.kvstore_count("lore", filter="category = 'quest'")

kvstore_delete(store_id: str)

Delete a KV store and remove its database files.

• store_id: The store to delete.

Returns: None

Raises: StoreError: If the store is not found.

engine.kvstore_delete("lore")

set_scheduling_mode(mode: str)

Set the GPU scheduling mode at runtime.

Controls how GPU time is shared between inference and rendering when running alongside a game engine.

• mode: One of "balance" (default), "prioritize_compute" (maximize inference speed), or "prioritize_graphics" (minimize rendering impact).

Returns: None

engine.set_scheduling_mode("prioritize_graphics")

set_vram_budget_mb(mb: int)

Set the VRAM budget in megabytes at runtime.

• mb: Maximum VRAM to use in megabytes. 0 means unlimited.

Returns: None

engine.set_vram_budget_mb(4096)  # 4 GB budget

set_target_tps(tps: int)

Set the target tokens per second for inference throttling.

When set to a non-zero value, the engine will pace token generation to leave GPU headroom for rendering.

• tps: Target tokens per second. 0 means unlimited (default).

Returns: None

engine.set_target_tps(30)  # throttle to ~30 tok/s

@staticmethod is_cig_d3d12_supported(device_index=0) -> bool

Check whether D3D12 Compute-in-Graphics (CiG) is supported on a GPU.

CiG allows sharing GPU scheduling context with a D3D12 renderer, avoiding OS-level context switching. Requires NVIDIA R570+ driver, CUDA 12.8+, and Ada Lovelace+ GPU.

• device_index: CUDA device index (default 0).

Returns: True if D3D12 CiG is supported, False otherwise. Always returns False when built without CUDA.

supported = Engine.is_cig_d3d12_supported(0)

@staticmethod is_cig_vulkan_supported(device_index=0) -> bool

Check whether Vulkan Compute-in-Graphics (CiG) is supported on a GPU.

CiG allows sharing GPU scheduling context with a Vulkan renderer. Requires NVIDIA R570+ driver, CUDA 12.9+, and Ada Lovelace+ GPU.

• device_index: CUDA device index (default 0).

Returns: True if Vulkan CiG is supported, False otherwise. Always returns False when built without CUDA.

supported = Engine.is_cig_vulkan_supported(0)

AnnIndex

Approximate Nearest Neighbor index for vector search.

Backed by HNSW (Hierarchical Navigable Small World) graph. This is a pure data structure -- no GPU or models required.

index = AnnIndex(dim=384, max_elements=1000)
index.insert([0.1, 0.2, ...], label_id=42)
index.build()
results = index.search([0.1, 0.2, ...], k=5)

Methods

__init__(dim, max_elements=10000, m=16, ef_construction=200, ef_search=50) -> Self

Create a new empty ANN index.

• dim: Dimensionality of vectors.
• max_elements: Maximum number of elements the index can hold (default 10000).
• m: HNSW M parameter -- connectivity degree (default 16).
• ef_construction: Search scope during construction (default 200).
• ef_search: Search scope during query (default 50).

Returns: A new empty AnnIndex instance.

index = AnnIndex(dim=384, max_elements=1000)

dim() -> int

Return the vector dimensionality of this index.

Returns: The dimensionality (int) passed to the constructor.

index = AnnIndex(dim=384)
assert index.dim() == 384

insert(vector: list[float], label_id: int)

Insert a vector with an associated label ID.

Call build() after all insertions are complete before searching.

• vector: A list of floats with length matching dim.
• label_id: An integer label associated with this vector (used to identify results returned by search).

Returns: None

index.insert([0.1, 0.2, 0.3], label_id=42)

build()

Build the HNSW index graph. Must be called after all insertions and before search.

Returns: None

index.insert([0.1, 0.2, 0.3], label_id=1)
index.build()
**Index is now ready for search**

search(query: list[float], k: int) -> list[(usize, f32)]

Search for the k nearest neighbors of a query vector.

• query: Query vector with length matching dim.
• k: Number of nearest neighbors to return.

Returns: A list of (label_id, distance) tuples sorted by ascending cosine distance (lower = more similar).

results = index.search([0.1, 0.2, 0.3], k=5)
for label_id, distance in results:
    print(f"Label {label_id}: distance={distance:.4f}")

save(path: str)

Save the index to a file on disk.

• path: Filesystem path to write the index to.

Returns: None

Raises: AtelicoError: If the file cannot be written.

index.save("/data/my_index.bin")

@staticmethod load(path: str) -> Self

Load a previously saved index from disk.

• path: Filesystem path to read the index from.

Returns: A new AnnIndex instance with the loaded data.

Raises: AtelicoError: If the file cannot be read or is corrupted.

index = AnnIndex.load("/data/my_index.bin")
results = index.search([0.1, 0.2, 0.3], k=5)

A2f

Audio-to-Face (A2F / lip-sync): drive ARKit-52 facial blendshapes from speech.

Standalone — has its own ~741 MB model bundle (NVIDIA Audio2Face-3D, downloaded on first use), independent of Engine. Build one and reuse it across calls.

import json
from atelico import A2f
a2f = A2f()                                            # auto device; downloads bundle once
frames = json.loads(a2f.process(pcm, 24000, "claire")) # pcm: list[float] in [-1, 1]
**frames[i]["weights"] is a list of 52 ARKit blendshape values in [0, 1]**

Methods

__init__() -> Self

Load A2F on the fastest available device, downloading the model bundle on first use. Reuse the instance — loading is expensive, inference is cheap.

process(pcm, sample_rate, character, noise_seed=None) -> str

Run A2F over a mono PCM waveform → ARKit-52 blendshape frames (60 fps), as JSON.

• pcm: mono samples in [-1.0, 1.0].
• sample_rate: the PCM sample rate (e.g. 24000 for Kokoro TTS; resampled internally).
• character: "claire", "james", or "mark".
• noise_seed: optional int seed for deterministic output (default random).

Returns: a JSON array [{"time_seconds": .., "weights": [52 floats]}, ...].