mixle.inference.production package

Production / MLOps layer for fitted mixle models.

The deployment-facing half of inference, kept out of the core mixle.inference namespace: reproducible model artifacts (provenance headers + verifiable training lineage), drift detection, a versioned model registry, a scoring service, and a drift-triggered monitor.

  • provenance: fit_with_provenance -> a Header (config, data hash, model hash, convergence trace, timing, env); verify_lineage checks the per-iteration model-hash chain.

  • drift: detect_drift -> DriftReport (feature PSI/KS + model score drift).

  • registry / serving / monitor: Registry (versioned store + alias swap + checkpointer), Service (batch scoring + activity logging), Monitor (drift -> retrain -> swap).

The container/Kubernetes serving layer that wraps Service lives in the separate mixle-deploy package.

fit_with_provenance(data, estimator, *, seed=None, lineage=True, **optimize_kw)[source]

Fit estimator on data via EM (mixle.inference.optimize()) and return (model, header), the model carrying a .header Header with the data hash, the final model hash, schema, training settings + per-iteration convergence trace, timing, final log-likelihood, and environment. Pass your own out= to print iterations (then the trace is not captured).

With lineage=True (default) each iteration in the convergence trace also records the accepted model’s model_hash and the previous iteration’s parent_hash, forming a verifiable hash chain (check it with verify_lineage()). This fingerprints the model every iteration; pass lineage=False to skip it for very large models. Any user on_step= is still called.

Parameters:
verify_lineage(header)[source]

Check that a header’s per-iteration convergence trace is an intact model-hash chain.

Returns True when every iteration that recorded a model_hash names the previous such iteration’s hash as its parent_hash (so iteration i+1 provably descends from i), and True vacuously when the trace carries no lineage (fit_with_provenance(lineage=False) or a custom out). Returns False on the first broken link. Accepts a Header or its to_dict.

Parameters:

header (Any)

Return type:

bool

class Header(model_type, model_summary, schema, n_records, dataset_hash, final_loglik, model_hash=None, training=<factory>, timing=<factory>, resources=<factory>, environment=<factory>, created_at='')[source]

Bases: object

A descriptive, serializable provenance record for a fitted model.

Parameters:
model_type: str
model_summary: str
schema: list[tuple[str, str]]
n_records: int | None
dataset_hash: str
final_loglik: float | None
model_hash: str | None = None
training: dict
timing: dict
resources: dict
environment: dict
created_at: str = ''
to_dict()[source]
Return type:

dict

classmethod from_dict(d)[source]
Parameters:

d (dict)

Return type:

Header

build_header(model, data, *, training=None, started=None, finished=None, final_loglik='auto', resources=None, hash_sort=False, hash_max_records=None)[source]

Build a Header for model trained on data (does not run any fitting).

Parameters:

  • model (Any)

  • data (Any)

  • training (dict | None)

  • started (float | None)

  • finished (float | None)

  • final_loglik (Any)

  • resources (dict | None)

  • hash_sort (bool)

  • hash_max_records (int | None)

Return type:

Header

environment_info()[source]

Snapshot of the software/hardware environment for reproducibility.

Return type:

dict

detect_drift(model, reference, current, *, psi_threshold=0.25, ks_threshold=0.2, loglik_shift_threshold=-0.5, per_feature=True)[source]

Combine score drift and per-feature drift into a single DriftReport.

drift is flagged if the score-distribution KS exceeds ks_threshold, OR the mean log-likelihood drops by more than -loglik_shift_threshold (i.e. mean_loglik_shift < loglik_shift_threshold), OR any feature’s PSI exceeds psi_threshold.

Parameters:
Return type:

DriftReport

score_drift(model, reference, current)[source]

The model-native drift signal: how the model’s log-density distribution shifts from reference to current data. Returns the KS statistic between the two log-likelihood samples and their mean shift (mean current log-density minus mean reference; negative => current data is less likely under the model).

Parameters:
Return type:

dict

class DriftReport(drift: 'bool', score: 'dict', per_feature: 'dict' = <factory>, thresholds: 'dict' = <factory>)[source]

Bases: object

Parameters:
drift: bool
score: dict
per_feature: dict
thresholds: dict
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

class Service(model, *, name=None, reference=None, log_path=None, keep=1000)[source]

Bases: object

A deployed model that scores batches and logs each computation for monitoring.

Parameters:

  • model (Any)

  • name (str | None)

  • reference (Any)

  • log_path (str | None)

  • keep (int)

classmethod from_registry(registry, name, *, alias='production', **kw)[source]

Load the model an alias points at in registry and serve it (carrying its provenance header).

Parameters:
Return type:

Service

score(records)[source]

Return per-record log-densities and log the computation (timing, mean log-lik, unscorable count).

Parameters:

records (Any)

Return type:

ndarray

check_drift(records)[source]

Drift of records versus the service’s reference sample (requires a reference).

Parameters:

records (Any)

Return type:

Any

health(*, window=100)[source]

Summary of the most recent window scoring events – throughput, mean log-likelihood, and the unscorable rate (the production problem signal).

Parameters:

window (int)

Return type:

dict

class Monitor(model, estimator, reference, *, psi_threshold=0.25, ks_threshold=0.2, loglik_shift_threshold=-0.5)[source]

Bases: object

Stateful monitor for one deployed model: drift detection + retrain-and-swap + DOE-driven sampling.

Parameters:

  • model (Any)

  • estimator (Any)

  • reference (Any)

  • psi_threshold (float)

  • ks_threshold (float)

  • loglik_shift_threshold (float)

check(current)[source]

Drift report of current (production) data against the reference under the current model.

Parameters:

current (Any)

Return type:

DriftReport

update(current, *, retrain=True, combine_reference=True, **fit_kw)[source]

Check drift on current and, if drift is flagged and retrain, fit a fresh model (with a new provenance header) and swap it in. Returns {report, action, model, header} and appends to history. combine_reference retrains on reference + current (else current only).

Parameters:
Return type:

dict

suggest_samples(bounds, n=10, *, method='lhs', objective=None, seed=None)[source]

Use mixle.doe to propose where to collect new data (e.g. after drift, or to meet an objective). method='lhs'/'sobol' gives a space-filling batch over bounds (list of (lo, hi)); pass an objective(x)->float to switch to active learning (ALC/ALM) that targets where the model is most informative.

Parameters:
Return type:

Any

Submodules