Models

The model gateway routes completion requests to the right provider adapter, handles retries, and tracks usage and cost.

Gateway

arcana.models.gateway.ModelGateway

ModelGateway(
    connections,
    *,
    providers=None,
    retry=None,
    pricing=None,
    on_cost=None,
    unhealthy_cooldown=_UNHEALTHY_COOLDOWN,
)

Single entry point the Agent uses to talk to any model.

Routes provider/model_id strings to the correct adapter, pools adapter instances per connection, retries transient failures with exponential backoff, and emits a CostEvent per completed call.

Token usage is recorded on the session regardless. Per-call cost is emitted only if on_cost is provided at construction; without it, no CostEvent fires.

Usage::

async with ModelGateway(connections=ConnectionStore()) as gw:
    response = await gw.complete("ollama/hermes-3", request)

# Or with a cost sink:
def record(event: CostEvent) -> None:
    ...

gw = ModelGateway(connections=store, on_cost=record)

Source code in packages/arcana-core/arcana/models/gateway.py

def __init__(
    self,
    connections: ConnectionStore,
    *,
    providers: ProviderRegistry | None = None,
    retry: RetryPolicy | None = None,
    pricing: PricingTable | None = None,
    on_cost: Callable[[CostEvent], Any] | None = None,
    unhealthy_cooldown: float = _UNHEALTHY_COOLDOWN,
) -> None:
    self._connections = connections
    self._providers = providers or DEFAULT_PROVIDERS
    self._retry = retry or RetryPolicy()
    self._pricing = pricing or DEFAULT_PRICING
    self._on_cost = on_cost
    self._unhealthy_cooldown = unhealthy_cooldown
    self._cache: dict[str, _CacheEntry] = {}
    self._cache_locks: dict[str, asyncio.Lock] = {}

complete async

complete(model, request)

Dispatch a completion request, retrying transient errors with backoff.

Source code in packages/arcana-core/arcana/models/gateway.py

async def complete(self, model: str, request: CompletionRequest) -> CompletionResponse:
    """Dispatch a completion request, retrying transient errors with backoff."""
    conn = self.resolve(model)
    entry = await self._get_cache_entry(conn)

    if entry.is_open(self._unhealthy_cooldown):
        raise ModelUnavailableError(f"Connection {model!r} is in cooldown after repeated failures.")

    req = replace(request, model_id=conn.default_model or "")
    session_id = (req.metadata or {}).get("session_id", "")

    with get_tracer().start_as_current_span("model.complete") as span:
        span.set_attribute("arcana.model", model)
        if session_id:
            span.set_attribute("arcana.session_id", session_id)

        last_exc: Exception | None = None
        start = time.monotonic()
        for attempt in range(self._retry.max_retries + 1):
            attempt_start = time.monotonic()
            try:
                response = await entry.adapter.complete(req)
                entry.mark_healthy()
                latency_ms = int((time.monotonic() - attempt_start) * 1000)
                span.set_attribute("arcana.input_tokens", response.input_tokens)
                span.set_attribute("arcana.output_tokens", response.output_tokens)
                span.set_attribute("arcana.attempts", attempt + 1)
                await self._emit_cost(model, response, conn, req.metadata)
                _emit_model_call(
                    session_id, model, latency_ms, response.input_tokens, response.output_tokens, attempt + 1, True
                )
                return response
            except _RETRYABLE as exc:
                latency_ms = int((time.monotonic() - attempt_start) * 1000)
                _emit_model_call(session_id, model, latency_ms, 0, 0, attempt + 1, False, str(exc))
                last_exc = exc
                if attempt == self._retry.max_retries:
                    break
                delay = self._retry.backoff(attempt, retry_after=getattr(exc, "retry_after", None))
                if self._retry.total_timeout is not None:
                    elapsed = time.monotonic() - start
                    remaining = self._retry.total_timeout - elapsed
                    if remaining <= 0:
                        break
                    delay = min(delay, remaining)
                await asyncio.sleep(delay)
            except ModelError as exc:
                span.record_exception(exc)
                raise

        # Every path that leaves the retry loop without returning or raising
        # assigns last_exc first; the None case is unreachable, but guard it
        # explicitly rather than raise a possibly-None value.
        if last_exc is None:
            raise ModelUnavailableError(f"Connection {model!r} failed without a captured error.")
        span.record_exception(last_exc)
        if isinstance(last_exc, ModelUnavailableError):
            entry.mark_unhealthy()
        raise last_exc

stream async

stream(model, request)

Stream a response as ModelChunk deltas.

Retry applies only before the first token arrives — mid-stream failures are surfaced immediately since output cannot be cleanly replayed. Emits one CostEvent after the stream completes.

Source code in packages/arcana-core/arcana/models/gateway.py

async def stream(self, model: str, request: CompletionRequest) -> AsyncGenerator[ModelChunk, None]:
    """Stream a response as ``ModelChunk`` deltas.

    Retry applies only before the first token arrives — mid-stream failures
    are surfaced immediately since output cannot be cleanly replayed.
    Emits one ``CostEvent`` after the stream completes.
    """
    conn = self.resolve(model)
    entry = await self._get_cache_entry(conn)

    if entry.is_open(self._unhealthy_cooldown):
        raise ModelUnavailableError(f"Connection {model!r} is in cooldown after repeated failures.")

    req = replace(request, model_id=conn.default_model or "")
    async with aclosing(self._retry_stream(model, conn, entry, req)) as gen:
        async for chunk in gen:
            yield chunk

health async

health(model=None)

Check health for one model string or all cached adapters.

A successful check resets the unhealthy flag so the connection is allowed back into the request path without waiting for cooldown.

Source code in packages/arcana-core/arcana/models/gateway.py

async def health(self, model: str | None = None) -> dict[str, ModelHealth]:
    """Check health for one model string or all cached adapters.

    A successful check resets the unhealthy flag so the connection is
    allowed back into the request path without waiting for cooldown.
    """
    if model is not None:
        conn = self.resolve(model)
        entry = await self._get_cache_entry(conn)
        result = await entry.adapter.health_check()
        if result.healthy:
            entry.mark_healthy()
        return {model: result}

    results: dict[str, ModelHealth] = {}
    for key, entry in self._cache.items():
        result = await entry.adapter.health_check()
        if result.healthy:
            entry.mark_healthy()
        results[key] = result
    return results

resolve

resolve(model)

Parse a model reference and return a ModelConnection with default_model set.

Accepted forms: - provider/model_id - provider:connection_name/model_id - provider (bare — uses connection's default_model) - provider:connection_name (bare named — uses connection's default_model)

When a connection name is given, looks it up by name in ConnectionStore and raises ValueError if it doesn't exist. Without a name, checks by provider first then falls back to ProviderRegistry defaults so out-of-the-box usage requires no config file.

Source code in packages/arcana-core/arcana/models/gateway.py

def resolve(self, model: str) -> ModelConnection:
    """Parse a model reference and return a ModelConnection with default_model set.

    Accepted forms:
    - ``provider/model_id``
    - ``provider:connection_name/model_id``
    - ``provider`` (bare — uses connection's default_model)
    - ``provider:connection_name`` (bare named — uses connection's default_model)

    When a connection name is given, looks it up by name in ConnectionStore and raises
    ValueError if it doesn't exist. Without a name, checks by provider first then falls
    back to ProviderRegistry defaults so out-of-the-box usage requires no config file.
    """
    provider, conn_name, model_id = self._parse_model_string(model)

    if conn_name is not None:
        conn = self._connections.get_by_name(conn_name)
        if conn is None:
            raise ValueError(
                f"No connection named {conn_name!r} found. "
                f"Add it with `arcana providers add` or check your connections file."
            )
        effective = model_id or conn.default_model
        if not effective:
            raise ModelNotConfiguredError(
                f"Connection {conn_name!r} has no default_model and the reference omits model_id. "
                f"Set a default or use '{provider}:{conn_name}/<model_id>'."
            )
        return conn.model_copy(update={"default_model": effective})

    entry = self._providers.get(provider)
    if entry is not None:
        conn = self._connections.get_by_provider(entry.provider)
        if conn is not None:
            effective = model_id or conn.default_model
            if not effective:
                raise ModelNotConfiguredError(
                    f"Connection for {provider!r} has no default_model. "
                    f"Specify: '{provider}/<model_id>' or set a default with `arcana providers edit`."
                )
            return conn.model_copy(update={"default_model": effective})

    if model_id is None:
        raise ModelNotConfiguredError(
            f"No configured connection for {provider!r} and no model_id in reference {model!r}. "
            f"Add a connection with `arcana providers add` or specify: '{provider}/<model_id>'."
        )
    return self._providers.build_default_connection(provider, model_id)

aclose async

aclose()

Close all cached adapters. Called automatically by the context manager.

Source code in packages/arcana-core/arcana/models/gateway.py

async def aclose(self) -> None:
    """Close all cached adapters. Called automatically by the context manager."""
    for entry in self._cache.values():
        await entry.adapter.aclose()
    self._cache.clear()
    self._cache_locks.clear()

arcana.models.gateway.RetryPolicy dataclass

RetryPolicy(
    max_retries=3,
    base=0.5,
    factor=2.0,
    cap=8.0,
    total_timeout=None,
)

Exponential backoff with full jitter. No fallback to a different model.

arcana.models.gateway.ProviderRegistry

ProviderRegistry(entries=None)

Maps provider strings to adapter factories.

Adding a provider = one register() call; no gateway changes needed.

Source code in packages/arcana-core/arcana/models/gateway.py

def __init__(self, entries: dict[str, ProviderEntry] | None = None) -> None:
    self._entries: dict[str, ProviderEntry] = dict(_DEFAULT_ENTRIES) if entries is None else entries

arcana.models.gateway.ProviderEntry dataclass

ProviderEntry(factory, default_endpoint, provider)

Maps a provider string to an adapter factory, its default endpoint, and canonical enum.

Connection store

arcana.models.connection_store.ConnectionStore

ConnectionStore(path=None)

Reads ModelConnection records from disk and credentials from the OS keyring.

Connections are loaded lazily on first access; call reload() to invalidate the cache if the file changes at runtime.

Usage::

store = ConnectionStore()
conn = store.get_by_provider(ModelProvider.ANTHROPIC)
key  = store.get_api_key(conn.id)

Source code in packages/arcana-core/arcana/models/connection_store.py

def __init__(self, path: Path | None = None) -> None:
    self._path = path or _default_path()
    self._connections: list[ModelConnection] | None = None

upsert

upsert(conn)

Insert conn, or replace the existing provider connection with the same name.

Always updates updated_at to now on write.

Source code in packages/arcana-core/arcana/models/connection_store.py

def upsert(self, conn: ModelConnection) -> None:
    """Insert conn, or replace the existing provider connection with the same name.

    Always updates ``updated_at`` to now on write.
    """
    connections = list(self._load())
    idx = next((i for i, c in enumerate(connections) if c.name == conn.name), None)
    updated = conn.model_copy(update={"updated_at": now_utc()})
    if idx is not None:
        updated = updated.model_copy(update={"id": connections[idx].id})
        connections[idx] = updated
    else:
        connections.append(updated)
    self._save(connections)

delete

delete(name)

Remove the connection with the given name and delete its keyring credential.

Source code in packages/arcana-core/arcana/models/connection_store.py

def delete(self, name: str) -> None:
    """Remove the connection with the given name and delete its keyring credential."""
    all_conns = list(self._load())
    to_delete = next((c for c in all_conns if c.name == name), None)
    remaining = [c for c in all_conns if c.name != name]
    self._save(remaining)
    if to_delete is not None:
        ref = to_delete.credential_ref or f"{to_delete.id}_api_key"
        try:
            self.delete_credential(ref)
        except Exception:
            pass

set_credential

set_credential(ref, secret)

Write a secret to the OS keyring under the given ref key.

Source code in packages/arcana-core/arcana/models/connection_store.py

def set_credential(self, ref: str, secret: str) -> None:
    """Write a secret to the OS keyring under the given ref key."""
    import keyring

    keyring.set_password("arcana", ref, secret)

delete_credential

delete_credential(ref)

Delete a secret from the OS keyring. No-op if the entry does not exist.

Source code in packages/arcana-core/arcana/models/connection_store.py

def delete_credential(self, ref: str) -> None:
    """Delete a secret from the OS keyring. No-op if the entry does not exist."""
    import keyring

    try:
        keyring.delete_password("arcana", ref)
    except Exception:
        pass

Adapters

arcana.models.adapters.base.ModelAdapter

Bases: ABC

Every LLM backend implements this interface.

connect async

connect()

Called once by the gateway after adapter construction. Default: no-op.

Source code in packages/arcana-core/arcana/models/adapters/base.py

async def connect(self) -> None:  # noqa: B027
    """Called once by the gateway after adapter construction. Default: no-op."""

aclose async

aclose()

Close underlying connections. Called by the gateway on shutdown. Default: no-op.

Source code in packages/arcana-core/arcana/models/adapters/base.py

async def aclose(self) -> None:  # noqa: B027
    """Close underlying connections. Called by the gateway on shutdown. Default: no-op."""

arcana.models.adapters.base.CompletionRequest dataclass

CompletionRequest(
    system,
    messages,
    temperature=0.7,
    max_tokens=4096,
    tools=None,
    stream=False,
    model_id="",
    metadata=None,
)

arcana.models.adapters.base.CompletionResponse dataclass

CompletionResponse(
    content,
    input_tokens=0,
    output_tokens=0,
    tool_calls=None,
    stop_reason="end_turn",
)

arcana.models.adapters.base.MessageParam

Bases: TypedDict

A single chat message in the canonical adapter wire format.

arcana.models.adapters.base.ModelChunk dataclass

ModelChunk(text, input_tokens=0, output_tokens=0)

A single streaming text delta from the gateway.

input_tokens and output_tokens are non-zero only on the final chunk (providers differ on when they send usage information).

arcana.models.adapters.base.ModelHealth dataclass

ModelHealth(healthy, model_id, message='')

Embedding adapters

Embedding generation is separate from both vector storage and the completion ModelAdapter above: an EmbeddingAdapter turns text into a dense vector. Concrete adapters target one provider each.

• OllamaEmbeddingAdapter — Tier 1, nomic-embed-text (768d) via Ollama's /api/embed; health probed through /api/tags.
• FastEmbedEmbeddingAdapter — Tier 2, nomic-embed-text-v1.5 (768d) via fastembed's in-process ONNX runtime. Optional dependency (arcana-core[embed]), imported lazily — health_check() reports unhealthy when the package is absent rather than failing at import.

Both report a shared model_family, so a database pinned to one can fall back to the other when its vectors are interchangeable. See model pinning.

arcana.models.adapters.embedding.EmbeddingAdapter

Bases: ABC

Converts text into a dense embedding vector. One concrete adapter per provider.

model_name and dimensions are the source of truth for model identity: callers read them to decide — before any vector is generated — whether the resolved model is the one a store is locked to. They are properties rather than fields on the returned vector so that decision needs no embedding call.

model_name abstractmethod property

model_name

Stable identifier for the embedding model.

Must be identical across process restarts: callers use it as a persistence key to detect when a store's model has changed.

dimensions abstractmethod property

dimensions

Length of the vectors this adapter produces.

model_family property

model_family

Identifier shared by models that embed into the same vector space.

Two adapters reporting the same model_family produce interchangeable vectors, so one may serve a database the other pinned. Defaults to model_name — each model is its own family unless an adapter explicitly declares membership in a shared one.

embed abstractmethod async

embed(text)

Embed a single string into a dense vector of length dimensions.

Raises EmbeddingError (or a subclass) on backend failure — never returns an empty or wrong-length vector silently. Callers that want graceful degradation gate on health_check first.

Source code in packages/arcana-core/arcana/models/adapters/embedding.py

@abstractmethod
async def embed(self, text: str) -> list[float]:
    """Embed a single string into a dense vector of length ``dimensions``.

    Raises ``EmbeddingError`` (or a subclass) on backend failure — never
    returns an empty or wrong-length vector silently. Callers that want
    graceful degradation gate on ``health_check`` first.
    """
    ...

embed_batch async

embed_batch(texts)

Embed many strings, order- and length-preserving (result[i] ↔ texts[i]).

Default implementation calls embed serially. Override for providers with a native batch endpoint.

Source code in packages/arcana-core/arcana/models/adapters/embedding.py

async def embed_batch(self, texts: list[str]) -> list[list[float]]:
    """Embed many strings, order- and length-preserving (``result[i]`` ↔ ``texts[i]``).

    Default implementation calls ``embed`` serially. Override for providers
    with a native batch endpoint.
    """
    return [await self.embed(text) for text in texts]

health_check abstractmethod async

health_check()

Probe whether the provider is reachable and usable.

Must not raise — return AdapterHealth(adapter_id=..., healthy=False, message=...) on failure, so callers can probe several adapters without exception handling.

Source code in packages/arcana-core/arcana/models/adapters/embedding.py

@abstractmethod
async def health_check(self) -> AdapterHealth:
    """Probe whether the provider is reachable and usable.

    Must not raise — return ``AdapterHealth(adapter_id=..., healthy=False,
    message=...)`` on failure, so callers can probe several adapters without
    exception handling.
    """
    ...

ensure_model async

ensure_model()

Download or warm up the model on first use; no-op if already present.

Default is a no-op for providers that need no local model.

Source code in packages/arcana-core/arcana/models/adapters/embedding.py

async def ensure_model(self) -> None:  # noqa: B027
    """Download or warm up the model on first use; no-op if already present.

    Default is a no-op for providers that need no local model.
    """

arcana.models.adapters.embedding.EmbeddingError

Bases: Exception

Base for embedding-backend failures.

Concrete adapters raise this (or a subclass) when generation fails, so callers can except EmbeddingError without knowing the provider.

arcana.models.adapters.ollama_embedding.OllamaEmbeddingAdapter

OllamaEmbeddingAdapter(
    model=_DEFAULT_MODEL,
    dimensions=_DEFAULT_DIMENSIONS,
    endpoint=_DEFAULT_ENDPOINT,
    timeout=60.0,
)

Bases: EmbeddingAdapter

Generates embeddings from a local Ollama instance.

Defaults to nomic-embed-text (768-dimensional). Talks to Ollama's /api/embed endpoint, which embeds a single string or a batch in one request; embed_batch uses that batch form directly.

Source code in packages/arcana-core/arcana/models/adapters/ollama_embedding.py

def __init__(
    self,
    model: str = _DEFAULT_MODEL,
    dimensions: int = _DEFAULT_DIMENSIONS,
    endpoint: str = _DEFAULT_ENDPOINT,
    timeout: float = 60.0,
) -> None:
    self._model = model
    self._dimensions = dimensions
    self.endpoint = endpoint.rstrip("/")
    self._client = httpx.AsyncClient(timeout=timeout)

arcana.models.adapters.fastembed_embedding.FastEmbedEmbeddingAdapter

FastEmbedEmbeddingAdapter(
    model_name=_DEFAULT_MODEL_NAME,
    repo_id=_DEFAULT_REPO_ID,
    dimensions=_DEFAULT_DIMENSIONS,
    cache_dir=None,
)

Bases: EmbeddingAdapter

In-process embeddings via fastembed's ONNX runtime.

Defaults to nomic-embed-text-v1.5 (768-dimensional), the same model family as the Ollama tier, so vectors are dimension-compatible. The ONNX model (~130 MB) downloads to ~/.arcana/models/ on first use.

Source code in packages/arcana-core/arcana/models/adapters/fastembed_embedding.py

def __init__(
    self,
    model_name: str = _DEFAULT_MODEL_NAME,
    repo_id: str = _DEFAULT_REPO_ID,
    dimensions: int = _DEFAULT_DIMENSIONS,
    cache_dir: Path | None = None,
) -> None:
    self._model_name = model_name
    self._repo_id = repo_id
    self._dimensions = dimensions
    self._cache_dir = cache_dir or _default_cache_dir()
    self._backend: Any = None

Pricing

arcana.models.pricing.Usage dataclass

Usage(prompt_tokens, completion_tokens, total, cost=None)

arcana.models.pricing.CostEvent dataclass

CostEvent(
    model,
    usage,
    estimated=False,
    priced=True,
    timestamp=(lambda: now(UTC))(),
    metadata=None,
)

Emitted by the gateway after every completed call. Not persisted — sink aggregates it.

metadata is copied verbatim from the originating CompletionRequest. The gateway never reads or validates it; callers use it to attribute cost to a session or agent.

arcana.models.pricing.PricingTable

PricingTable(data=None)

Token cost lookup by provider/model_id. Local (Ollama) models are always $0.

Source code in packages/arcana-core/arcana/models/pricing.py

def __init__(self, data: dict[str, tuple[float, float]] | None = None) -> None:
    self._data: dict[str, tuple[float, float]] = dict(_DEFAULT_PRICES) if data is None else data

cost

cost(model_key, input_tokens, output_tokens, conn=None)

Return the cost in USD, or None if the model has no known price.

Lookup priority: 1. Per-connection cost_per_1k_* overrides (if both are set). 2. Global pricing table entry. 3. 0.0 for local/Ollama models (known to be free). 4. None — price is unknown; callers should flag the event as unpriced.

Source code in packages/arcana-core/arcana/models/pricing.py

def cost(
    self,
    model_key: str,
    input_tokens: int,
    output_tokens: int,
    conn: ModelConnection | None = None,
) -> float | None:
    """Return the cost in USD, or ``None`` if the model has no known price.

    Lookup priority:
    1. Per-connection ``cost_per_1k_*`` overrides (if both are set).
    2. Global pricing table entry.
    3. ``0.0`` for local/Ollama models (known to be free).
    4. ``None`` — price is unknown; callers should flag the event as unpriced.
    """
    if conn is not None:
        if conn.cost_per_1k_input_tokens is not None and conn.cost_per_1k_output_tokens is not None:
            return (
                conn.cost_per_1k_input_tokens * input_tokens + conn.cost_per_1k_output_tokens * output_tokens
            ) / 1000

    entry = self._data.get(model_key)
    if entry:
        return (entry[0] * input_tokens + entry[1] * output_tokens) / 1000

    if conn is not None and conn.is_local:
        return 0.0

    return None

Errors

arcana.models.errors.ModelError

Bases: Exception

Base class for all model errors.

arcana.models.errors.ModelTransientError

ModelTransientError(message, *, retry_after=None)

Bases: ModelError

Retryable error: timeout, connection reset, 429, 500/502/503.

Source code in packages/arcana-core/arcana/models/errors.py

def __init__(self, message: str, *, retry_after: float | None = None) -> None:
    super().__init__(message)
    self.retry_after = retry_after

arcana.models.errors.ModelUnavailableError

ModelUnavailableError(message, *, retry_after=None)

Bases: ModelTransientError

Connection refused — server not running or still cold (Ollama, local endpoints).

Source code in packages/arcana-core/arcana/models/errors.py

def __init__(self, message: str, *, retry_after: float | None = None) -> None:
    super().__init__(message)
    self.retry_after = retry_after

arcana.models.errors.ModelAuthError

Bases: ModelError

Fatal: 401 / 403. Retrying is pointless until credentials change.

arcana.models.errors.ModelBadRequestError

Bases: ModelError

Fatal: 400, malformed request, context-length exceeded.

arcana.models.errors.ModelNotFoundError

Bases: ModelError

Fatal: model not pulled or unknown model ID.

The message should tell the user how to fix it (e.g. ollama pull <model>).

arcana.models.errors.ModelNotConfiguredError

Bases: ModelError

No effective model_id: the reference omitted it and no default_model is set on the connection.