Skip to content

yggdrasil.loki.engines

engines

Concrete :class:~yggdrasil.loki.engine.TokenEngine backends.

Remote (hosted APIs — fast, capable, metered):

  • :class:OpenAIEngine — the OpenAI API.
  • :class:ClaudeEngine — the Anthropic (Claude) API.
  • :class:DatabricksServingEngine — a Databricks model-serving endpoint.

Local (run on this workstation — free, private, resource-bound):

  • :class:TransformersEngine — an open HuggingFace model via transformers (CPU, or an Intel GPU through the XPU torch build).
  • :class:OpenVINOEngine — a model on the Intel NPU (AI Boost) via OpenVINO / optimum-intel, falling back to the Intel GPU then CPU.
  • :class:OllamaEngine — a model served by a local Ollama server.

ClaudeEngine 

ClaudeEngine(
    *,
    model: Optional[str] = None,
    tier: Optional[str] = None,
    api_key: Optional[str] = None,
    auth_token: Optional[str] = None
)

Bases: TokenEngine

Reason via the Anthropic Messages API (anthropic SDK).

model_label property 

model_label: str

Human label for status output — the pin, or the adaptive ceiling.

uses_oauth property 

uses_oauth: bool

True when this engine will authenticate with a subscription token.

choose_tier 

choose_tier(
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
) -> str

Adaptive tier for this request: "deep" or "fast".

Sizes on the message content (the actual work, not the fixed system boilerplate) and scans both message and system text for reasoning signals. Long or signalled requests get the deep tier; the rest stay fast. Override for a smarter policy.

resolve_model 

resolve_model(
    *,
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
    tier: Optional[str] = None
) -> Optional[str]

The model id to use for this request.

An explicit self.model pin wins. Otherwise a forced tier (arg or self.tier) selects from :attr:MODELS; with neither, the tier is chosen adaptively. Falls back to :attr:default_model when the tier isn't in the map.

usage 

usage() -> list[Any]

This engine's per-model usage rows from the global meter.

generate 

generate(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> str

Convenience: complete a single user prompt → reply text.

generate_stream 

generate_stream(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> "Iterator[str]"

Convenience: stream a single user prompt → text chunks.

DatabricksServingEngine 

DatabricksServingEngine(
    *,
    client: Any = None,
    endpoint: Optional[str] = None,
    model: Optional[str] = None,
    available: Optional[bool] = None
)

Bases: TokenEngine

Reason via a Databricks serving endpoint.

model_label property 

model_label: str

Human label for status output — the pin, or the adaptive ceiling.

choose_tier 

choose_tier(
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
) -> str

Adaptive tier for this request: "deep" or "fast".

Sizes on the message content (the actual work, not the fixed system boilerplate) and scans both message and system text for reasoning signals. Long or signalled requests get the deep tier; the rest stay fast. Override for a smarter policy.

resolve_model 

resolve_model(
    *,
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
    tier: Optional[str] = None
) -> Optional[str]

The model id to use for this request.

An explicit self.model pin wins. Otherwise a forced tier (arg or self.tier) selects from :attr:MODELS; with neither, the tier is chosen adaptively. Falls back to :attr:default_model when the tier isn't in the map.

usage 

usage() -> list[Any]

This engine's per-model usage rows from the global meter.

generate 

generate(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> str

Convenience: complete a single user prompt → reply text.

generate_stream 

generate_stream(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> "Iterator[str]"

Convenience: stream a single user prompt → text chunks.

warm 

warm() -> None

Best-effort: build + cache the OpenAI-compatible client ahead of the first completion so the first submit isn't slowed by client setup.

A no-op when the openai dep isn't present yet (the first real call installs it) — the warmer must never trigger a background pip install.

OllamaEngine 

OllamaEngine(
    *,
    model: Optional[str] = None,
    tier: Optional[str] = None,
    host: Optional[str] = None
)

Bases: LocalEngine

Reason with a local model served by Ollama, sized to the workstation.

bootstrap_model property 

bootstrap_model: str

The default model for this box — the resource-sized row, or the engine's :attr:default_model fallback.

choose_tier 

choose_tier(
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
) -> str

Adaptive tier for this request: "deep" or "fast".

Sizes on the message content (the actual work, not the fixed system boilerplate) and scans both message and system text for reasoning signals. Long or signalled requests get the deep tier; the rest stay fast. Override for a smarter policy.

resolve_model 

resolve_model(
    *,
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
    tier: Optional[str] = None
) -> Optional[str]

An explicit pin wins; otherwise the model sized to this workstation.

Local models are resource-bound, so the remote fast/deep cost tier doesn't apply — the box, not the prompt, picks the size.

usage 

usage() -> list[Any]

This engine's per-model usage rows from the global meter.

generate 

generate(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> str

Convenience: complete a single user prompt → reply text.

stream 

stream(
    messages: list[dict[str, Any]],
    *,
    system: Optional[str] = None,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    tier: Optional[str] = None,
    **options: Any
) -> "Iterator[str]"

Yield reply text incrementally as it is produced.

The default has no real streaming — it runs :meth:complete and yields the whole reply once. Engines whose SDK streams override this to yield token deltas live (and still record usage on completion).

generate_stream 

generate_stream(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> "Iterator[str]"

Convenience: stream a single user prompt → text chunks.

installed_models 

installed_models() -> list[str]

Models already pulled onto this Ollama server (empty if unreachable).

has_model 

has_model(model: str) -> bool

True when model (with or without a :tag) is already pulled.

pull 

pull(
    model: Optional[str] = None,
    *,
    timeout: float = 1800.0,
    on_progress: "Optional[Callable[[dict[str, Any]], None]]" = None
) -> str

Download model onto the Ollama server (lazy — the heavy bit).

Defaults to :attr:bootstrap_model, the lightweight brain. Without on_progress this does the non-streaming pull (one final status object). With on_progress it streams Ollama's NDJSON progress events — {"status", "completed", "total"} per chunk — so a caller can render a live download bar; the final status string is returned either way.

ensure 

ensure(
    model: Optional[str] = None,
    on_progress: "Optional[Callable[[dict[str, Any]], None]]" = None,
) -> dict[str, Any]

Make sure model is available, pulling it only if missing.

Returns {"model", "was_present", "status"} — the lazy-install receipt so a caller (the setup skill) can report what it did. An on_progress callback streams the pull's download progress.

OpenAIEngine 

OpenAIEngine(
    *,
    model: Optional[str] = None,
    tier: Optional[str] = None,
    api_key: Optional[str] = None,
    base_url: Optional[str] = None
)

Bases: TokenEngine

Reason via the OpenAI Chat Completions API (openai SDK).

model_label property 

model_label: str

Human label for status output — the pin, or the adaptive ceiling.

choose_tier 

choose_tier(
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
) -> str

Adaptive tier for this request: "deep" or "fast".

Sizes on the message content (the actual work, not the fixed system boilerplate) and scans both message and system text for reasoning signals. Long or signalled requests get the deep tier; the rest stay fast. Override for a smarter policy.

resolve_model 

resolve_model(
    *,
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
    tier: Optional[str] = None
) -> Optional[str]

The model id to use for this request.

An explicit self.model pin wins. Otherwise a forced tier (arg or self.tier) selects from :attr:MODELS; with neither, the tier is chosen adaptively. Falls back to :attr:default_model when the tier isn't in the map.

usage 

usage() -> list[Any]

This engine's per-model usage rows from the global meter.

generate 

generate(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> str

Convenience: complete a single user prompt → reply text.

generate_stream 

generate_stream(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> "Iterator[str]"

Convenience: stream a single user prompt → text chunks.

OpenVINOEngine 

OpenVINOEngine(
    *,
    model: Optional[str] = None,
    tier: Optional[str] = None,
    device: Optional[str] = None
)

Bases: TransformersEngine

Reason with a local model on the Intel NPU via OpenVINO / optimum-intel.

bootstrap_model property 

bootstrap_model: str

The default model for this box — the resource-sized row, or the engine's :attr:default_model fallback.

choose_tier 

choose_tier(
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
) -> str

Adaptive tier for this request: "deep" or "fast".

Sizes on the message content (the actual work, not the fixed system boilerplate) and scans both message and system text for reasoning signals. Long or signalled requests get the deep tier; the rest stay fast. Override for a smarter policy.

resolve_model 

resolve_model(
    *,
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
    tier: Optional[str] = None
) -> Optional[str]

An explicit pin wins; otherwise the model sized to this workstation.

Local models are resource-bound, so the remote fast/deep cost tier doesn't apply — the box, not the prompt, picks the size.

usage 

usage() -> list[Any]

This engine's per-model usage rows from the global meter.

generate 

generate(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> str

Convenience: complete a single user prompt → reply text.

stream 

stream(
    messages: list[dict[str, Any]],
    *,
    system: Optional[str] = None,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    tier: Optional[str] = None,
    **options: Any
) -> Iterator[str]

Generate live, token by token, via TextIteratorStreamer.

Without this the base :meth:stream runs the whole generation in one blocking :meth:complete and yields it at the end — so a slow CPU run prints nothing until it finishes. Here the pipeline runs on a worker thread and feeds a streamer the terminal drains as tokens arrive.

generate_stream 

generate_stream(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> "Iterator[str]"

Convenience: stream a single user prompt → text chunks.

ready 

ready(model: Optional[str] = None) -> bool

True when the pipeline for model (resolved if omitted) is loaded.

Lets a caller (the CLI) warn that a turn is about to trigger the slow first load — download weights + build the pipeline — instead of going silent on a CPU box.

warm 

warm(model: Optional[str] = None) -> None

Build the model's pipeline ahead of the first turn — best-effort.

Loading a local model is slow and silent (download weights → build the pipeline); the ygg loki REPL calls this on a background thread so the wait overlaps the user picking a session and typing, instead of stalling the first submit. Swallows failures — they're cached in :attr:_FAILED and surfaced on the first real turn.

available 

available() -> bool

True when OpenVINO + optimum are installed and an NPU/GPU is present.

A CPU-only box is left to the transformers / ollama engines — this engine exists for the accelerators a torch pipeline can't reach (chiefly the NPU). Cheap: the package check is find_spec; the device list is memoized.

resolve_device 

resolve_device() -> str

The OpenVINO device to run on: an explicit pin wins, else the best present accelerator — NPU first (the whole point), then GPU.

TransformersEngine 

TransformersEngine(
    *,
    model: Optional[str] = None,
    tier: Optional[str] = None,
    device: Optional[str] = None
)

Bases: LocalEngine

Reason with a local HuggingFace model (transformers pipeline).

bootstrap_model property 

bootstrap_model: str

The default model for this box — the resource-sized row, or the engine's :attr:default_model fallback.

choose_tier 

choose_tier(
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
) -> str

Adaptive tier for this request: "deep" or "fast".

Sizes on the message content (the actual work, not the fixed system boilerplate) and scans both message and system text for reasoning signals. Long or signalled requests get the deep tier; the rest stay fast. Override for a smarter policy.

resolve_model 

resolve_model(
    *,
    messages: Optional[list[dict[str, Any]]] = None,
    system: Optional[str] = None,
    tier: Optional[str] = None
) -> Optional[str]

An explicit pin wins; otherwise the model sized to this workstation.

Local models are resource-bound, so the remote fast/deep cost tier doesn't apply — the box, not the prompt, picks the size.

usage 

usage() -> list[Any]

This engine's per-model usage rows from the global meter.

generate 

generate(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> str

Convenience: complete a single user prompt → reply text.

generate_stream 

generate_stream(
    prompt: str,
    *,
    system: Optional[str] = None,
    tier: Optional[str] = None,
    **options: Any
) -> "Iterator[str]"

Convenience: stream a single user prompt → text chunks.

resolve_device 

resolve_device() -> Optional[str]

The device to load the pipeline on: an explicit pin (ctor arg / YGG_LOKI_HF_DEVICE) wins; otherwise the best auto-detected accelerator — NVIDIA cuda, Intel GPU xpu, Apple mps — or None (CPU). Lets a local model use the GPU without configuration.

ready 

ready(model: Optional[str] = None) -> bool

True when the pipeline for model (resolved if omitted) is loaded.

Lets a caller (the CLI) warn that a turn is about to trigger the slow first load — download weights + build the pipeline — instead of going silent on a CPU box.

warm 

warm(model: Optional[str] = None) -> None

Build the model's pipeline ahead of the first turn — best-effort.

Loading a local model is slow and silent (download weights → build the pipeline); the ygg loki REPL calls this on a background thread so the wait overlaps the user picking a session and typing, instead of stalling the first submit. Swallows failures — they're cached in :attr:_FAILED and surfaced on the first real turn.

stream 

stream(
    messages: list[dict[str, Any]],
    *,
    system: Optional[str] = None,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    tier: Optional[str] = None,
    **options: Any
) -> Iterator[str]

Generate live, token by token, via TextIteratorStreamer.

Without this the base :meth:stream runs the whole generation in one blocking :meth:complete and yields it at the end — so a slow CPU run prints nothing until it finishes. Here the pipeline runs on a worker thread and feeds a streamer the terminal drains as tokens arrive.