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}'**

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

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

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.

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_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])

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",
  "backend_kind": "metal"
}

import json
caps = json.loads(engine.model_capabilities("in-memory::meta-llama/Llama-3.2-3B-Instruct-Q4_K_M"))
print(f"context window = {caps['max_position_embeddings']} tokens")

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_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(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(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.

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.

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)

Load a LoRA adapter onto an already-loaded model.

• model_id: The base model ID to attach the adapter to.
• adapter_path: Filesystem path to the adapter directory (must contain adapter_config.json and weight files).

Returns: None

Raises: ModelLoadError: If the model is not loaded or the adapter path is invalid.

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

lora_unload(model_id: str)

Unload a LoRA adapter from a model, reverting to base model weights.

• model_id: The model whose adapter should be removed.

Returns: None

Raises: ModelLoadError: If the model is not loaded or has no adapter.

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

lora_set_scale(model_id: str, scale: float)

Set the LoRA runtime scale factor for a model's loaded adapter.

A scale of 1.0 applies the full adapter effect; 0.0 effectively disables it without unloading.

• model_id: The model with a loaded LoRA adapter.
• scale: Scale factor (typically 0.0 to 1.0).

Returns: None

Raises: ModelLoadError: If the model is not loaded or has no adapter.

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

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_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:

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

Returns: None

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

import json
entries = json.dumps([
    {"key": "greeting", "value": "Hello!", "embedding": [0.1, 0.2, 0.3]},
])
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_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

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.