mixle.registry module

Registry – a local directory + index of fitted task models, queryable by capability and fingerprint.

The registry is the local library catalog that orchestrators, routers, capture flows, and accumulation workflows can read from or write into. Deliberately a directory plus a JSON index, not a server: every entry is a saved TaskModel or CalibratedTaskModel artifact directory (see mixle.task.artifact) plus a small index record naming its capabilities, task fingerprint (task_fingerprint()), and capture profile. find_for answers “do I already have something for this task”; tier_stack turns a matching capability into an ascending-cost tier list – the shape Router consumes directly (Router(tiers=stack)), with the frontier appended last as the router’s own fallback tier.

class RegistryEntry(entry_id, path, kind, capabilities=<factory>, fingerprint=None, profile=<factory>, cost=0.0)

Bases: object

One catalog record: where the artifact lives, what it’s registered under, and how much it costs to run.

Parameters:

• entry_id (str)

• path (str)

• kind (str)

• capabilities (list[str])

• fingerprint (list[float] | None)

• profile (dict[str, Any])

• cost (float)

to_dict()

Serialize the registry entry into JSON-compatible fields.

Return type:

dict[str, Any]

classmethod from_dict(d)

Create a registry entry from a JSON index record.

Parameters:

d (dict[str, Any])

Return type:

RegistryEntry

class Registry(dir)

Bases: object

A dir-backed catalog of registered models: register writes an artifact + index entry; find_for/tier_stack query it. Re-opening the same dir in a fresh process sees every entry.

Parameters:

dir (str)

register(model, *, capabilities, fingerprint=None, profile=None, cost=0.0, entry_id=None)

Save model’s artifact under dir and add its index entry; return the entry.

model is a fitted TaskModel or CalibratedTaskModel – the two artifact-saveable task model kinds. capabilities names what this model answers (matched by find_for()); fingerprint is typically task_fingerprint()’s vector for the training data; profile is free-form (e.g. a capture_profile() dict); cost is the per-request cost used to order tier_stack().

Parameters:

• model (TaskModel | CalibratedTaskModel)

• capabilities (Sequence[str])

• fingerprint (Sequence[float] | None)

• profile (dict[str, Any] | None)

• cost (float)

• entry_id (str | None)

Return type:

RegistryEntry

load(entry_id)

Reload a registered model by entry_id (round-trips through the artifact on disk).

Parameters:

entry_id (str)

Return type:

TaskModel | CalibratedTaskModel

find_for(query, *, top_k=None)

Entries matching query: a capability name (str, containment match) or a task fingerprint vector (array-like of floats, nearest-neighbor match). top_k caps how many are returned – every capability match by default, or the single nearest fingerprint match by default.

Parameters:

• query (str | Sequence[float])

• top_k (int | None)

Return type:

list[RegistryEntry]

tier_stack(task, *, frontier, costs=None, names=None)

Ascending-cost (name, model, cost) tiers for capability task, frontier appended last.

Matching entries are loaded (load()) and ordered by their registered cost. The result is exactly the shape Router takes as tiers=: each non-final tier exposes decide(x), the final tier is the callable frontier fallback. costs (one entry per matching solution plus one for frontier, mirroring from_solutions()) overrides the registered per-entry costs when given.

Parameters:

• task (str)

• frontier (Any)

• costs (Sequence[float] | None)

• names (Sequence[str] | None)

Return type:

list[tuple[str, Any, float]]