yggdrasil.loki

loki

Loki — the global yggdrasil agent.

Loki adapts to wherever it runs: it detects the backends it can reach (Databricks session, node, local), acts as a token/credential provider for them, and dispatches pluggable :class:LokiSkill actions. Driven from code via :class:Loki or from the terminal via ygg loki.

from yggdrasil.loki import Loki
Loki.current().card()

Loki

Loki()

The global yggdrasil agent — capability-aware, token-providing.

agent_id property

agent_id: int

Stable int64 id derived from user@host (xxhash, not crypto); cached.

databricks property

databricks: 'Optional[DatabricksClient]'

The authenticated Databricks client when a session is present.

This is Loki acting as a token provider: skills and downstream code take this client to reach Databricks service endpoints (SQL, Genie, jobs, serving, …) under the agent's resolved credentials. Returns None when no Databricks session is detected.

aws property

aws: Any

The configured :class:~yggdrasil.aws.AWSClient when AWS is reachable.

Loki as an AWS token provider — the AWS skill fleet rides this client (its resolved credentials / region / role). None when no AWS session is detected.

current classmethod

current() -> 'Loki'

The process-global Loki (created on first use).

backends

backends(*, refresh: bool = False) -> list[Backend]

Detected backends (cached; pass refresh=True to re-sniff).

load_specialists

load_specialists() -> list[str]

Import the specialized skill fleets for every reachable backend.

Databricks problems get the databricks-* skills, AWS problems the aws-* skills — registered only when their backend is detected, so ygg loki skills shows the fleet that actually applies here. Returns the backends whose fleet loaded.

token_info

token_info() -> dict[str, Any]

Non-secret summary of the Databricks credentials Loki provides.

whoami

whoami(*, probe: bool = False) -> 'Optional[str]'

The Databricks user, if reachable. probe allows one network call.

engines

engines() -> 'list[TokenEngine]'

Every known reasoning engine (call .available() to filter).

available_engines

available_engines(*, refresh: bool = False) -> 'dict[str, TokenEngine]'

Reachable engines (name → instance), availability probed in parallel.

Several engines gate on a network round-trip — the Ollama liveness probe, the Databricks backend check — so probing them one after another stacks up their latencies on the startup path. Fanning the available() checks across a small thread pool collapses that to the slowest single probe. Each engine memoizes its own result, so this also warms the caches that later serial available() calls (the status line, :meth:engine, :meth:select) reuse for free.

engine

engine(name: 'Optional[str]' = None) -> 'Optional[TokenEngine]'

Resolve a reasoning engine by name, or the best available one.

select

select(
    text: "Optional[str]" = None,
    *,
    tier: "Optional[str]" = None,
    base: "Optional[str]" = None,
    confirm: "Optional[Callable[[TokenEngine, Optional[str]], bool]]" = None
) -> "Optional[TokenEngine]"

Resource-aware engine choice: keep light work cheap, escalate heavy work to a capable remote — asking first.

Two axes drive the pick:

• Complexity — an explicit tier (deep → heavy), else the prompt itself (long or reasoning-heavy text → heavy).
• Resources — a CUDA GPU, or enough CPU + RAM (≥ 4 cores, ≥ 8 GB), decides whether this box can comfortably run a local model.

A session pins a base provider, but complexity moves the choice both ways:

• remote → local (demote): light work on a capable box drops to a free local model even when the base is a remote API — saving money, silently (no downside to ask about).
• local → remote (escalate): heavy work climbs to the most capable remote (the base remote if it is one). When this means switching from a free local model up to a paid remote one, confirm(engine, model) is asked first; a falsy answer keeps the work on the cheap/local path.

With no local engine the base simply stands; with no remote the local engine carries even heavy work. Returns an available engine, or None when nothing is reachable.

can_run_local

can_run_local() -> bool

Whether this box can comfortably host a local model — cached.

Wraps :func:yggdrasil.loki.resources.can_run_local but memoizes the result for the process: the probe imports torch to check for a CUDA GPU (slow on the first call, on a box that has it), and engine selection asks this on every turn. Hardware doesn't change within a session.

bootstrap_local

bootstrap_local(
    *,
    model: "Optional[str]" = None,
    pull: bool = True,
    on_progress: "Optional[Callable[[dict[str, Any]], None]]" = None
) -> dict[str, Any]

Ready a free local reasoning engine, lazily installing on demand.

Prefers a reachable Ollama server — ensures the model sized to this workstation (the more RAM/GPU, the larger the default) is pulled, only if missing. Falls back to the HF transformers engine (weights lazy-download on first use). When neither is present, returns what to install. This is the "free local brain" entry point — sized to the box, smart enough for basic setup/config, and able to hand heavier work up to a remote model.

reason

reason(
    prompt: str,
    *,
    system: "Optional[str]" = None,
    engine: "Optional[str]" = None,
    tier: "Optional[str]" = None,
    base: "Optional[str]" = None,
    confirm: "Optional[Callable[[TokenEngine, Optional[str]], bool]]" = None,
    **options: Any
) -> str

Reason about prompt with the best (or named) engine → reply text.

tier ("fast" / "deep") forces the model tier; the default (None) lets the engine pick adaptively from the prompt. A pinned engine is used as-is; otherwise the choice is resource-aware (:meth:select), sticking to the session base and asking confirm before escalating to a paid remote model.

reason_stream

reason_stream(
    prompt: str,
    *,
    system: "Optional[str]" = None,
    engine: "Optional[str]" = None,
    tier: "Optional[str]" = None,
    base: "Optional[str]" = None,
    confirm: "Optional[Callable[[TokenEngine, Optional[str]], bool]]" = None,
    **options: Any
) -> "Iterator[str]"

Stream a reply to prompt — yields text chunks as they arrive.

Same engine/tier resolution as :meth:reason, but live: the chosen engine streams token deltas so the terminal prints them as they come.

act

act(
    task: str,
    *,
    root: str = ".",
    engine: "Optional[str]" = None,
    tier: "Optional[str]" = None,
    max_steps: int = 12,
    read_only: bool = False,
    allow_shell: bool = False,
    allow_web: bool = False,
    confirm: "Optional[Callable[[str], bool]]" = None,
    toolbox: "Optional[Toolbox]" = None,
    on_think: "Optional[Callable[[int], None]]" = None,
    on_step: "Optional[Callable[[dict[str, Any]], None]]" = None
) -> dict[str, Any]

Pursue task autonomously: discover, decide, and modify files.

This is Loki acting on its own — the reason→act→observe loop. The agent's engine plans against a tool catalog (filesystem discovery + edits, optionally a shell), emits one JSON tool call per turn, and Loki runs it and feeds the observation back, until the engine declares it's done or max_steps is hit. The tools are confined to root (the working tree the agent was pointed at).

tier pins the model tier for every turn; left None the engine adapts per turn — cheap early scouting turns, the capable model once the transcript (and the reasoning) grows.

Returns a transcript: the resolved engine, every step (its thought/tool/args/observation), the final answer, whether it completed, and the files_changed list. Pass on_step to stream each completed turn, and on_think(n) to learn when turn n is about to call the (slow) model — the CLI uses the pair to keep a live spinner + step-budget bar running through the otherwise silent reasoning between tool calls.

delegate

delegate(
    tasks: list[str],
    *,
    root: str = ".",
    engine: "Optional[str]" = None,
    tier: "Optional[str]" = None,
    max_steps: int = 8,
    allow_web: bool = True,
    allow_shell: bool = False,
    read_only: bool = False,
    timeout: "Optional[float]" = None,
    on_update: "Optional[Callable[[list[Any]], None]]" = None
) -> list[dict[str, Any]]

Fan tasks out to background process agents and wait for them.

Each task runs as its own ygg loki do subprocess (an isolated act loop), so independent work proceeds in parallel while Loki monitors them — its autonomy multiplier. Returns one summary row per agent (status, elapsed, answer, files_changed). on_update(agents) streams live progress (the CLI renders the dashboard from it).

decompose

decompose(
    goal: str,
    *,
    engine: "Optional[str]" = None,
    tier: "Optional[str]" = None,
    max_tasks: int = 6
) -> list[str]

Break goal into independent subtasks an agent fleet can run in parallel.

Asks the reasoning engine for a JSON array of self-contained tasks (parallel-safe — no ordering between them). Returns the parsed list, capped at max_tasks; falls back to [goal] if nothing parses.

plan

plan(text: str) -> 'AgentPlan'

Classify a request into a structured :class:AgentPlan.

Loki reasoning, optimized: rather than throwing every prompt at one engine, it classifies the problem — a category and solution action (answer / act on files / fetch tabular / ask Genie), the persona to embody (data engineer, analyst, software engineer, trader, confessor, companion, …), the skills likely required, and whether to isolate the work to a specialist (the "databricks on databricks" scheme). Returns an :class:AgentPlan (mapping-compatible).

classify_data

classify_data(text: str) -> dict[str, Any]

Global context: is this request data- or time-series-shaped?

Drives the data path — a positive classification routes a sourced request to tabular fetching + caching (:class:TabularSkill) instead of a plain page fetch. Returns {"data", "timeseries", "why"}.

specialist

specialist(name: str) -> 'Optional[Loki]'

A specialized agent to isolate a category of work, or None.

"databricks" resolves the workspace-bound :class:~yggdrasil.databricks.loki.DatabricksLoki when the SDK and a session are present; otherwise falls back to self.

Cached per name: the REPL asks for the same specialist on every databricks turn, and the resolution (import + singleton lookup + backend check) is stable for the process.

run

run(skill_name: str, **kwargs: Any) -> Any

Dispatch skill skill_name with kwargs, using self as the provider.

The first parameter is skill_name (not name) so a skill that itself takes a name= kwarg — e.g. scaffold(name=…) — can be dispatched as loki.run("scaffold", name="acme") without a clash.

card

card(*, refresh: bool = False) -> dict[str, Any]

Everything Loki knows about itself — identity, reach, skills.

LokiSkill

Bases: ABC

One discoverable, environment-aware capability Loki can perform.

available

available(agent: 'Loki') -> bool

True when this skill can run in agent's environment.

Default: available everywhere, unless :attr:requires names a backend that isn't detected. Override for finer checks.

run abstractmethod

run(agent: 'Loki', **kwargs: Any) -> Any

Perform the skill, using agent as the capability/token provider.

Backend dataclass

Backend(name: str, available: bool, detail: dict[str, Any] = dict())

A capability surface Loki can drive (e.g. Databricks, a node).

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.

Tool dataclass

Tool(
    name: str,
    description: str,
    params: dict[str, str],
    run: Callable[..., str],
    mutates: bool = False,
)

One named capability the agent can invoke by emitting a JSON call.

run takes the decoded args as keyword arguments and returns a string observation the agent reads on its next turn. params maps each argument name to a one-line description — it's rendered into the system prompt so the model knows the tool's shape.

Toolbox dataclass

Toolbox(tools: dict[str, Tool] = dict(), changed: list[str] = list())

A named set of tools plus the run state the loop reports on.

spec

spec() -> str

The tool catalog as prompt text — one block per tool.

call

call(name: str, args: dict[str, Any]) -> str

Run a tool by name; surface any error as a readable observation.

ModelPricing dataclass

ModelPricing(input_usd_per_mtok: float, output_usd_per_mtok: float)

USD price per million tokens, split input/output.

ModelUsage dataclass

ModelUsage(
    engine: str,
    model: str,
    calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
)

Running consumption for one (engine, model) pair (or the global roll-up).

TokenMeter

TokenMeter()

Accumulates per-model token usage and enforces a cost budget (USD).

Recording never raises — it only counts. Enforcement is explicit: :meth:check_budget raises :class:TokenBudgetExceeded when the USD spend crosses the cap, so the caller (the REPL) can stop between actions and offer to raise it. The cap is money, not tokens — a fixed budget across models of wildly different per-token prices.

total_cost property

total_cost: float

Global USD — summed per row so each is priced at its own rate.

record

record(
    engine: str, model: str, input_tokens: int, output_tokens: int
) -> ModelUsage

Add one completion's tokens to the (engine, model) row.

rows

rows() -> list[ModelUsage]

Per-model usage rows, busiest first.

total

total() -> ModelUsage

The global roll-up across every engine and model.

set_limit

set_limit(usd: Optional[float]) -> None

Set (or clear, with None) the total-spend cap in USD.

raise_limit

raise_limit(by: Optional[float] = None) -> float

Bump the cap by by USD (or one :attr:cost_step); returns the new cap.

remaining

remaining() -> Optional[float]

USD left under the cap (None when uncapped; may go negative).

check_budget

check_budget() -> None

Raise :class:TokenBudgetExceeded if the spend cap is set and crossed.

LokiSession dataclass

LokiSession(
    id: str,
    dir: Path,
    user: str,
    name: str = "",
    first_prompt: str = "",
    created_at: float = time.time(),
    last_used_at: float = time.time(),
)

One isolated, named, per-user session directory tree.

start classmethod

start(*, user: Optional[str] = None, purge: bool = True) -> 'LokiSession'

Create a fresh session for the user (purging stale ones first).

list classmethod

list(*, user: Optional[str] = None) -> 'list[LokiSession]'

The user's sessions, most-recently-used first.

resume classmethod

resume(id: str, *, user: Optional[str] = None) -> 'Optional[LokiSession]'

Reopen an existing session by id and reset its purge clock.

purge classmethod

purge(
    *,
    user: Optional[str] = None,
    keep: int = KEEP,
    max_age_days: float = MAX_AGE_DAYS,
    exclude: "Optional[Path]" = None
) -> "list[str]"

Delete the user's stale session trees; return the ids removed.

Stale = untouched for > max_age_days (by last_used_at) OR outside the keep most-recently-used. exclude (the live session) is spared.

touch

touch() -> None

Mark the session used now — resets its purge clock.

name_from_prompt

name_from_prompt(prompt: str) -> str

Name the session from its first user prompt (slug); persist + return.

LokiMemory

LokiMemory(
    path: "str | Path | None" = None,
    *,
    keep_recent: int = 8,
    compress_chars: int = 6000
)

Recent turns + a rolling synthesis of older context, auto-compressed.

add

add(role: str, content: str) -> None

Append a turn (role is "user" / "assistant").

system_context

system_context() -> Optional[str]

The memory rendered as a system note for the next reasoning call.

Synthesis first (the long tail, compressed), then the recent turns verbatim — bounded and token-efficient. None when empty.

should_compress

should_compress() -> bool

True when the raw context is large enough to fold into the synthesis.

The cheap gate :meth:maybe_compress checks first — exposed so a caller (the CLI) can show a waiting animation only when a compression (a real, slow model call) is actually about to run, not on every turn.

maybe_compress

maybe_compress(agent: 'Loki', *, engine: Optional[str] = None) -> bool

Fold older turns into the synthesis when context grows too large.

Keeps the last :attr:keep_recent turns verbatim and summarizes the rest via the agent's fast engine. Returns whether it compressed. Reasoning failures (no engine) leave the raw turns intact.

register

register(skill: 'type[LokiSkill] | LokiSkill') -> 'type[LokiSkill] | LokiSkill'

Register a skill (class or instance) by its name.

Usable as a decorator on a :class:LokiSkill subclass::

@register
class Echo(LokiSkill):
    name = "echo"
    def run(self, agent, **kw): return kw

registry

registry() -> list['LokiSkill']

All registered skills, sorted by name.

detect

detect() -> list[Backend]

Every backend Loki can see from here, in priority order.

filesystem_toolbox

filesystem_toolbox(
    root: str | Path = ".",
    *,
    read_only: bool = False,
    allow_shell: bool = False,
    allow_web: bool = False,
    confirm: Optional[Callable[[str], bool]] = None
) -> Toolbox

Build the default toolbox rooted at root.

The read tools (list_dir, read_file, find, grep, read_table) are always present — discovery, including parsing local tabular files (CSV/Parquet/Arrow/XLSX/JSON) through the io handlers. The write tools — write_file, edit_file, run_python (write & run Python), and run (a shell command) — are added unless read_only; the agent can code and shell out by default. allow_shell is kept for back-compat but no longer gates run. Network tools (web_*) are added only with allow_web.

confirm (fn(action) -> bool) gates destructive ops on non-temporary assets — overwriting/editing an existing file outside the system temp dir asks first; new files and scratch/temp files don't.

price_for

price_for(engine: str, model: str) -> ModelPricing

Resolve pricing: exact (engine, model) → (engine, "*") → default.