`yggdrasil.loki.engine`¶

engine ¶

TokenEngine — the LLM reasoning contract Loki agents run on.

A :class:TokenEngine turns a chat message list into a completion. It's the single seam between Loki and whatever model backs it: the OpenAI API, the Anthropic (Claude) API, or a Databricks serving endpoint. Engines declare whether they're :meth:available (credentials/config present) so an agent can pick the best reachable brain, and implement :meth:complete.

Messages use the portable [{"role": ..., "content": ...}] shape shared by every provider; system is passed separately (Anthropic keeps it out of the message list, and the others accept a leading system message).

Adaptive model selection. Each engine declares a small :attr:MODELS tier map — a "fast" model and a "deep" (more capable) one. When the caller pins neither a model nor a tier, the engine adapts: light, short requests resolve to the fast model; long or reasoning-heavy ones resolve to the deep model (:meth:choose_tier). Pinning a model= always wins, and passing tier="deep" / "fast" forces the choice — adaptivity is only the default, never an override of an explicit decision.

Completion `dataclass` ¶

Completion(
    text: str,
    model: Optional[str] = None,
    usage: dict[str, Any] = dict(),
    raw: Any = None,
)

The result of a single engine turn.

TokenEngine ¶

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

Bases: ABC

A pluggable LLM backend Loki reasons with.

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.

available `abstractmethod` ¶

available() -> bool

True when this engine has the credentials/config to run.

complete `abstractmethod` ¶

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

Run one chat completion and return a :class:Completion.

tier forces "fast" / "deep" model selection for this call; None (the default) lets the engine adapt.

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.

yggdrasil.loki.engine¶