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 records a digest-chained transition from the previous accepted model, and a terminal record binds the model actually returned by the optimizer (check it with verify_lineage(); the digests bind the iteration sequence and the model, not the header’s dataset hash, record count, or log-likelihood values, which stay recorded claims). 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]

Verify a complete execution-recorded EM transition chain.

Missing lineage is an unverified result, never a vacuous success. Each iteration must bind its actual model hash to the previous model and transition digests, all records must share one run id, and the final executed model hash must equal the header’s fitted-model hash.

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:
to_dict()[source]

Return the provenance header as a plain dictionary.

Return type:

dict

classmethod from_dict(d)[source]

Reconstruct a provenance header from a dictionary.

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.

Source identity is taken from the ambient repository when there is one, and otherwise from the installed artifact’s embedded build provenance. provenance_source names which, because “no repository here” and “a clean repository” are different facts and were previously reported the same way.

Return type:

dict

detect_drift(model, reference, current, *, psi_threshold=0.25, ks_threshold=0.2, loglik_shift_threshold=-0.5, min_scorable_fraction=0.5, unscorable_shift_threshold=0.1, 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, OR the current sample lost too much scoring coverage to support a no-drift verdict.

Coverage is part of the verdict, not just of the report. score_drift drops unscorable records before comparing distributions, so a current sample of one matching score plus nine records the model cannot score at all used to produce KS 0, mean shift 0, and drift=False – a confident “no drift” from 10% of the evidence. Two coverage conditions now fail closed:

  • fewer than min_scorable_fraction of current records scored finitely, and

  • the current unscorable rate exceeds the reference’s by more than unscorable_shift_threshold (records moving outside the model’s support IS the world moving away from the model).

Both populations are materialized once and reused for scoring and for the per-field columns, so a one-shot iterable is safe: consuming the stream twice previously left the feature pass with empty columns and manufactured a PSI-infinity DRIFT verdict out of two identical inputs.

Parameters:

  • model (Any)

  • reference (Any)

  • current (Any)

  • psi_threshold (float)

  • ks_threshold (float)

  • loglik_shift_threshold (float)

  • min_scorable_fraction (float)

  • unscorable_shift_threshold (float)

  • per_feature (bool)

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).

Both populations are materialized here exactly once, so a one-shot iterable is safe and the reported counts describe the records actually scored. Alongside the shift statistics the report carries the evidence behind them – how many records each population held and what fraction of each was unscorable – because a shift measured over a handful of surviving finite scores is not the same claim as one measured over the whole sample.

Parameters:
Return type:

dict

class DriftReport(drift, score, per_feature=<factory>, thresholds=<factory>, processed_count=0, reasons=<factory>)[source]

Bases: object

Drift decision, aggregate score, feature details, and thresholds.

reasons names every condition that raised the drift flag, so a verdict driven by lost scoring coverage is distinguishable from one driven by a genuine distribution shift.

The fields cannot be rebound (frozen=True) and the containers are copies of what was passed in, so the caller’s own objects cannot reach into a report after it is built. The containers themselves stay writable by whoever holds the report – see __post_init__().

Parameters:
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, expected_tip=_UNSPECIFIED)[source]

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

expected_tip makes the write conditional: the current latest version must equal it (or the name must be empty when it is None) at the moment of allocation, checked under the same lock that allocates the version. A caller extending a lineage passes the tip it adopted, and a concurrent writer that got there first makes this call refuse instead of persisting a sibling successor.

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.

Allocating the next version number is a read-modify-write over versions(name): two concurrent callers (two threads, or two processes) can both read the same existing versions, both compute the same “next” number, and both write it – one write silently clobbers the other, with no error to either caller. To prevent that, the number allocation and the write are serialized per model name with an fcntl.flock on a lock file in the model’s directory (held for both the read and the write, so a second writer waits and is correctly given the NEXT free number rather than racing for the same one). The write itself additionally uses O_CREAT | O_EXCL as an independent, belt-and-suspenders guard: if a file for that version exists anyway (e.g. flock is a no-op on the underlying filesystem), it raises instead of silently overwriting the earlier registration.

The version file is also published atomically: the payload is written to a private temp file in the model’s directory and fsynced before it is ever linked to the final <version>.json path (via os.link, after which the temp name is dropped). A failure anywhere in between – serialization raising partway through json.dump, the disk filling up, the process crashing – therefore never leaves a truncated or corrupt version file on disk: the file only ever exists fully-formed or not at all. This mirrors the temp-file-plus-atomic-publish idiom used by mixle.task.artifact._atomic_json_dump() and mixle.system.registry’s _write_index, except the publish step uses os.link rather than os.replace: this write must still refuse to clobber an existing (or concurrently-claimed) version file, and os.replace would silently overwrite one – reopening the bug the O_CREAT | O_EXCL guard above fixes. Unlike writing directly into the O_CREAT | O_EXCL-opened file, a failed attempt here also leaves nothing behind at path, so the same version number stays cleanly retriable instead of being permanently stuck behind a corrupt file that neither parses nor can be overwritten.

Parameters:
Return type:

str

checkpointer(name, *, every=1, resume=True, trust_code=False)[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

A checkpointer created for a name that already holds a VERIFIABLE chain resumes it: it adopts that chain’s run identity, links its first new checkpoint to the existing tip, and continues the iteration count, so appending recovered work leaves verify_chain true. Pass resume=False to deliberately root a new lineage instead. trust_code is only used to re-load models while checking whether the existing chain is adoptable, and carries the same meaning as in get().

Parameters:
Return type:

Callable[[Any], None]

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

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

A registered model containing a NeuralLeaf-family component embeds its weights as a pickle blob (see mixle.models._neural_serial); deserializing that executes code, so get requires trust_code=True for such an entry – trust the registry root, not just the JSON extension. A pure-statistical entry loads either way.

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

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

Content digest of the exact persisted version envelope, re-verified on load.

A content digest, not a signature: it detects corruption and out-of-band edits of the model alone, not a writer with directory access who rewrites the model and every digest together (the same limit mixle.lifecycle.Model.load() states); integrity against that threat needs a signature or a copy of the digests kept outside the directory.

Parameters:
Return type:

str

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

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

Written via a temp file + os.replace in the same directory (same filesystem, so the rename is atomic): a concurrent reader of current() either sees the old alias target or the new one, never a truncated/partial write, and a crash mid-write leaves the old alias file untouched rather than corrupted.

Parameters:
Return type:

None

current(name, alias=None, *, trust_code=False)[source]

Load the model an alias points at.

Two deliberately different behaviours, split on whether the caller NAMED an alias:

  • current(name) – no alias requested. Resolves the default "production" alias if it exists and falls back to latest if it does not. This is the bootstrap path: a registry with registrations but no promotion yet still serves something.

  • current(name, "production") – an alias was explicitly requested. It must exist; an absent alias raises KeyError. A caller that names an alias is asserting “serve the version promoted to this alias”, and quietly substituting latest answers a different question with an UNPROMOTED model – which also silently defeats a rollback, since rolling an alias back to a known-good version has no effect on a caller whose alias name is misspelled or was never created (SYS-05).

See get() – trust_code is required in the same way and for the same reason.

Parameters:
Return type:

tuple[Any, dict | None]

verify_chain(name, *, trust_code=False)[source]

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

Requires every version to carry a complete digest-chained lineage record. Checks exact parent versions, record digests, model hashes, transition digests, run identity, and iteration order, and re-hashes each loaded model. Missing or partial lineage is unverified and returns False.

Every digest is an unkeyed content digest recomputed by this module’s own helpers, so the chain proves that the stored versions are consistent with each other and unchanged since they were written, not who wrote them: a writer with directory access can replace a model and recompute every digest, and the chain verifies. Guard against that threat with a signature or a copy of the tip digest kept outside the directory.

See get() – trust_code is required in the same way and for the same reason: a chain that contains a NeuralLeaf-family checkpoint must be loaded to be re-hashed, so verifying it is subject to the same code-execution trust gate as loading it directly.

Parameters:
Return type:

bool

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

Bases: object

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

Parameters:
classmethod from_registry(registry, name, *, alias=None, **kw)[source]

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

alias follows Registry.current() exactly: naming one requires it to exist (an absent or misspelled alias raises rather than serving an unpromoted version), while omitting it takes the promoted production model if there is one and the latest registration otherwise.

Parameters:
Return type:

Service

score(records)[source]

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

Records affected by a model_unavailable failure (see module docstring) still score -inf in the returned array – preserving its shape/dtype contract for callers – but are counted separately as n_unavailable in the logged event and surfaced as unavailable_rate by health(), rather than being indistinguishable from a genuinely-scored -inf.

The batch call itself failing is tracked separately again, as batch_unavailable: every record in a batch whose seq_log_density call raised is retried individually, and is counted in n_batch_unavailable / batch_unavailable_rate regardless of whether that per-record retry recovers a real score or itself fails into n_unavailable – so a batch endpoint that is failing over to the (much slower) per-record path on every call remains visible even when every individual retry happens to succeed and no record is actually lost.

Raises:

ValueError – if the model’s batch output does not have exactly one log-density per input record – a cardinality mismatch is a contract violation to reject, not a shape to silently accept. Raised only for a batch call that succeeded with the wrong shape; it is not itself treated as a reason to retry per-record.

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, the unscorable rate (the production problem signal), the unavailable rate (the subset of unscorable records caused by the model being unreachable – an outage – rather than legitimately scored), and the batch-unavailable rate (records whose batch call failed and had to be retried per-record, whether or not that retry itself recovered a score – always >= the unavailable rate; see score()).

mean_loglik is the mean over RECORDS, not over events: it is the total finite log-density divided by scored_records, so it does not move when the same scores arrive in differently sized batches. Averaging each event’s own batch mean made one record with log-likelihood 0 followed by 100 records averaging -10 report -5.0 instead of the true -9.901, which let batch shape alone manufacture or mask a degradation. Unscorable records are excluded from both the numerator and scored_records (they are counted by unscorable_rate instead); an all- unscorable window reports mean_loglik=None rather than a number derived from no evidence.

Parameters:

window (int) – how many of the most recent scoring events to summarize; an exact positive integer. 0 used to select the whole history ([-0:] is a full slice) and a negative value selected a surprising suffix, both silently.

Raises:

  • TypeError – if window is not an exact integer (bool included).

  • ValueError – if window is not positive.

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