mixle.inference.production.registry module

A versioned model registry: store fitted models + their provenance, list versions, promote/swap.

A filesystem-backed store so a production system can register every fitted model (with its Header), list and load any version, and promote a chosen version to an alias (e.g. "production") – the swap point a serving layer reads from. Models serialize through mixle.utils.serialization (the safe registry-keyed JSON); headers are plain JSON dicts.

class Registry(root)[source]

Bases: object

A directory of named models, each with numbered versions and movable aliases.

Parameters:

root (str)

names()[source]

Registered model names.

Return type:

list[str]

versions(name)[source]

Version ids for name in registration order (v1, v2, …).

Parameters:

name (str)

Return type:

list[str]

register(model, name, *, header=None, metadata=None)[source]

Store model under name as a new version; return its version id.

header defaults to model.header if present. The model is serialized with the safe mixle registry; the header (a Header or dict) and metadata are stored alongside.

Parameters:
Return type:

str

checkpointer(name, *, every=1)[source]

Return an optimize(on_step=...) callback that snapshots the model under name every every iterations (recording the iteration + log-density in the version metadata).

Each checkpoint records its model model_hash and the previous checkpoint’s parent_hash, so the saved snapshots form a verifiable chain (see verify_chain()). Resume an interrupted run from the latest checkpoint:

reg = Registry("ckpts")
optimize(data, est, on_step=reg.checkpointer("run", every=5))
model, _ = reg.get("run")              # latest checkpoint
optimize(data, est, prev_estimate=model)   # continue training
Parameters:
Return type:

Callable[[Any], None]

get(name, version='latest')[source]

Load (model, header) for a version ("latest" = highest-numbered).

Parameters:
Return type:

tuple[Any, dict | None]

header(name, version='latest')[source]

Just the provenance header of a version (no model deserialization).

Parameters:
Return type:

dict | None

metadata(name, version='latest')[source]

Just the metadata of a version (no model deserialization) – e.g. a checkpoint’s iteration.

Parameters:
Return type:

dict

promote(name, version, alias='production')[source]

Point alias (e.g. "production") at version – the atomic model swap.

Parameters:
Return type:

None

current(name, alias='production')[source]

Load the model an alias points at (falls back to latest if the alias is unset).

Parameters:
Return type:

tuple[Any, dict | None]

verify_chain(name)[source]

Verify the persisted checkpoint lineage for name (see checkpointer()).

For each version carrying a model_hash, checks that its parent_hash matches the previous such version’s hash and that re-hashing the loaded model reproduces the stored hash (catching corruption or tampering). Returns True when every link holds, or vacuously when no version carries lineage metadata.

Parameters:

name (str)

Return type:

bool