mixle.task package

Local task-specific models with durable artifacts and calibrated serving.

The unit is a TaskModel: a fitted model scoped to one task (classify, extract, recommend a model shape, …), scoped to one operational behavior and saved as a durable artifact (artifact) so a plain Python program can load it in a fresh process and call it. Producers:

• distill() – a teacher callable labels data and a local student is fit to match;

• tune_recipe() – mixle.doe searches the student recipe to minimize train cost.

This module’s public surface re-exports the artifact contract; the model/distill/tune layers land on top.

class ActiveResult(model, labels_used, history=<factory>, labeled_texts=<factory>, labeled_labels=<factory>)

Bases: object

The actively-distilled student plus an audit trail of labels spent vs. quality reached each round.

Parameters:

• model (TaskModel)

• labels_used (int)

• history (list[dict[str, Any]])

• labeled_texts (list[str])

• labeled_labels (list[Any])

class AgentTrace(request, plan, reply='', conversation_id='')

Bases: object

One request, ordered tool calls, and final text reply.

Parameters:

• request (str)

• plan (list[dict])

• reply (str)

• conversation_id (str)

class AgentTraces(traces=<factory>)

Bases: object

The harvested corpus plus the teacher views the distillers consume.

Parameters:

traces (list[AgentTrace])

requests(*, min_steps=0)

The request texts (optionally only those whose plan has at least min_steps calls).

Parameters:

min_steps (int)

Return type:

list[str]

tool_specs()

Infer tool specs from observed argument usage.

Return type:

list[ToolSpec]

call_teacher()

Return a distill_tool_caller teacher over the first tool call.

Return type:

Any

plan_teacher()

Return a planner teacher over the full harvested tool-call plan.

Return type:

Any

class CalibratedTaskModel(task, *, alpha=0.1, qhat=None, density_gate=None)

Bases: object

A TaskModel plus a conformal threshold: predicts label sets and decides answer-vs-escalate.

Parameters:

• task (TaskModel)

• alpha (float)

• qhat (float | None)

• density_gate (Any)

property labels: list[str]

Return labels in the probability-vector order used by the adapter.

calibrate(texts, teacher_labels)

Set the conformal threshold from held-out (texts, teacher_labels) for 1 - alpha set coverage.

Parameters:

• texts (Sequence[Any])

• teacher_labels (Sequence[Any])

Return type:

CalibratedTaskModel

predict_sets(texts)

Conformal label set per input (the classes whose score clears the calibrated threshold).

Parameters:

texts (Sequence[Any])

Return type:

list[list[str]]

predict_set(text)

Return the conformal label set for one input.

Parameters:

text (Any)

Return type:

list[str]

decide(text)

Return the label if the input is a confident, in-distribution singleton, else ESCALATE (None).

Parameters:

text (Any)

Return type:

Any

batch_decide(texts)

Return local labels or ESCALATE for a batch of inputs.

Parameters:

texts (Sequence[Any])

Return type:

list[Any]

escalation_rate(texts)

Empirical p_escalate – the fraction of inputs escalated (ambiguous set or, if gated, OOD).

Parameters:

texts (Sequence[Any])

Return type:

float

save(path)

Persist the underlying model, the calibration (alpha, qhat), and any density gate in the artifact.

qhat can legitimately be +inf (a small calibration set / tight alpha: too little data to admit any confident singleton, so every input escalates). That is a real, callable threshold, so it is persisted as the JSON-safe sentinel "inf" and reloads back to float('inf') – a loaded model stays callable instead of raising “call calibrate”.

Parameters:

path (str)

Return type:

str

classmethod load(path, *, device='cpu')

Rebuild a calibrated model (with its density gate, if any) from an artifact; decisions match exactly.

Parameters:

• path (str)

• device (str)

Return type:

CalibratedTaskModel

class CallableLLM(fn)

Bases: object

Wrap a plain fn(prompt) -> str (or fn(prompt, system)) as an LLM – local models and tests.

Parameters:

fn (Callable[..., str])

complete(prompt, *, system=None, **kwargs)

Call the wrapped Python function and return its text output.

Parameters:

• prompt (str)

• system (str | None)

• kwargs (Any)

Return type:

str

class CapabilitySuite(corruptions=<factory>, invariances=<factory>, probes=<factory>)

Bases: object

The behavioral spec an example distillation is checked against.

corruptions maps a named severity level (e.g. "typo_10") to a text -> text corruption; insertion order is the intended severity order (mild first) so callers can read the profile’s ordering directly. invariances maps a name to a meaning-preserving rewrite (case jitter, whitespace, a synonym swap) – a well-behaved model’s prediction should not change under it. probes are fixed edge-case inputs whose raw predictions are recorded without assuming ground truth.

Parameters:

• corruptions (dict[str, Callable[[str], str]])

• invariances (dict[str, Callable[[str], str]])

• probes (list[str])

class Cascade(model, teacher, *, cost=None)

Bases: object

Serve text -> label through a confident local model, escalating to the teacher when needed.

Parameters:

• model (CalibratedTaskModel)

• teacher (Callable[..., Any])

• cost (CostModel | None)

serve(texts)

Serve a batch of requests through the cascade.

Parameters:

texts (Sequence[Any])

Return type:

list[Any]

harvested()

Return escalated (texts, teacher_labels) as targeted retraining data.

Return type:

tuple[list[Any], list[Any]]

realized_cost()

Actual spend so far: c_local per request plus c_frontier per escalation (requires a CostModel).

Return type:

float

report()

Realized economics: requests, escalation rate, spend, and savings vs serving everything on the frontier.

Return type:

dict[str, Any]

plan(*, volume, n_label, max_escalation=None)

Project the lowest-cost route at volume using the realized escalation rate.

Parameters:

• volume (int)

• n_label (int)

• max_escalation (float | None)

Return type:

RoutePlan

class CascadeStats(n_requests=0, n_escalated=0, escalated_texts=<factory>, escalated_labels=<factory>)

Bases: object

Running tally of how a cascade served traffic – the basis for realized cost and the harvest.

Parameters:

• n_requests (int)

• n_escalated (int)

• escalated_texts (list[Any])

• escalated_labels (list[Any])

property realized_escalation_rate: float

Return the observed fraction of requests escalated to the teacher.

class ComposedAnswer(answer, intermediate, stages, total_contribution)

Bases: object

A composed x -> z answer plus the per-stage receipt that attributes it to both stages.

Parameters:

• answer (Any)

• intermediate (Any)

• stages (list[tuple[str, Any, float]])

• total_contribution (float)

check(tol=1e-9)

sum(contributions) == total_contribution – the ledger is exact by construction.

Parameters:

tol (float)

Return type:

bool

class ComposedModel(a, b, *, name_a='stage_a', name_b='stage_b')

Bases: object

Chain a: x -> y then b: y -> z as one callable x -> z.

composed(x) returns the bare answer z (so a ComposedModel can stand in anywhere a plain teacher callable is expected – including as the a or b of another compose(), chaining further). composed.answer(x) returns the ledger-carrying ComposedAnswer instead.

Parameters:

• a (Callable[[Any], Any])

• b (Callable[[Any], Any])

• name_a (str)

• name_b (str)

answer(x)

Return the composed answer with each stage’s contribution record.

Parameters:

x (Any)

Return type:

ComposedAnswer

class SyntheticDomain(name, vocab, period=8, noise_p=0.0, pattern_seed=0)

Bases: object

One synthetic “domain”: a fixed periodic token pattern, optionally corrupted by noise.

pattern_seed fixes a length-period sequence of token ids (drawn once, from 0..vocab)) that repeats forever – the domain’s learnable structure. Each sampled token then has independent probability noise_p of being replaced by a uniform-random token, so noise_p=0 is a perfectly learnable domain and noise_p=1 (or period=None) is pure, irreducible noise: no amount of training data lowers a model’s achievable loss on it below log(vocab). Distinct (period, pattern_seed, noise_p) triples give genuinely different data-generating distributions, standing in for e.g. “web text” vs “code” vs “books” without needing real corpora.

Parameters:

• name (str)

• vocab (int)

• period (int | None)

• noise_p (float)

• pattern_seed (int)

sample(n_tokens, *, seed=0)

Draw n_tokens ids (int64 array) from this domain’s distribution.

Parameters:

• n_tokens (int)

• seed (int)

Return type:

ndarray

estimate_near_duplicate_rate(corpus, *, shingle_size=5, num_hashes=64, threshold=0.8, seed=0)

Estimate the fraction of documents in corpus that have a near-duplicate elsewhere in it.

A minimal, honest MinHash quality/dedup receipt: each document is reduced to its set of word-shingle_size shingles, each shingle set to a num_hashes-entry MinHash signature (an unbiased estimator of Jaccard similarity), and two documents are called near-duplicates when their signatures agree on at least threshold of their entries. Returns |{documents with >= 1 near-duplicate partner}| / |corpus|. O(n^2) in the corpus size – fine for the receipt-sized corpora this is meant for, not a production LSH dedup pipeline.

Parameters:

• corpus (Sequence[str])

• shingle_size (int)

• num_hashes (int)

• threshold (float)

• seed (int)

Return type:

float

optimize_mixture(domains, proxy_steps, budget, *, method='bandit', proxy_kwargs=None, seed=0)

Learn domain mixture weights via repeated short proxy runs (DoReMi-style search).

budget proxy runs (each proxy_run_score() at proxy_steps gradient steps) are used to search the mixture-weight simplex. method="bandit" (default) discretizes the simplex into a lattice of candidate mixtures (mixle.doe.mixture.simplex_lattice) and searches them with mixle.task.bandit.ThompsonGaussian (reward = negative held-out loss); method="doe" searches continuously via mixle.doe.optimizer.BayesianOptimizer over a softmax-reparameterized simplex. Returns the learned weight vector (one entry per domain, summing to 1).

Parameters:

• domains (Sequence[SyntheticDomain])

• proxy_steps (int)

• budget (int)

• method (str)

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

• seed (int)

Return type:

ndarray

proxy_run_score(mixture_weights, domains, proxy_steps, *, batch_size=16, d_model=16, n_layer=1, n_head=2, block=8, lr=3.0e-3, eval_tokens=512, seed=0, eval_seed=999_000, return_detail=False)

Run one short proxy training and return the mean held-out NLL across domains (lower is better).

Builds a training token stream by drawing mixture_weights[i]-proportional tokens from each domain (concatenated; the number of tokens is chosen so training runs roughly proxy_steps gradient steps at batch_size), trains a real (tiny) mixle.models.language_model.LM on it for one epoch, then scores held-out NLL on eval_tokens fresh tokens from EACH domain (independent of the mixture) and returns the unweighted mean across domains – the DoReMi objective is generalizing to every domain, not just the ones the mixture over-samples. return_detail=True also returns the per-domain NLL dict, keyed by domain name.

seed controls the training-data draw (and so varies across repeated proxy runs, e.g. inside optimize_mixture()’s search loop); eval_seed controls the held-out draw and is fixed by default so different mixtures proposed during a search are scored against the SAME held-out set – comparing candidate mixtures on a moving eval target would swamp the (often small) between-mixture signal in eval-sampling noise.

Parameters:

• mixture_weights (Sequence[float])

• domains (Sequence[SyntheticDomain])

• proxy_steps (int)

• batch_size (int)

• d_model (int)

• n_layer (int)

• n_head (int)

• block (int)

• lr (float)

• eval_tokens (int)

• seed (int)

• eval_seed (int)

• return_detail (bool)

Return type:

float | tuple[float, dict[str, float]]

class EstimatorBandit(estimators, *, n_boot=32, mean_fn=None, mc_draws=64, seed=None)

Bases: _BanditBase

Thompson sampling for ARBITRARY mixle reward models, via the online bootstrap.

Each arm keeps n_boot accumulator replicates of its estimator; update adds the reward to every replicate with an independent Poisson(1) weight (Eckles & Kaptein’s online bootstrap), so the replicate ensemble approximates the sampling distribution of the fitted reward model with no conjugate structure required. select plays each arm once, then draws one non-empty replicate per arm, fits it (estimator.estimate), scores it with mean_fn (default: Monte-Carlo mean of estimate.sampler(...).sample(mc_draws)), and plays the argmax – posterior-sample-then-maximize, exactly Thompson’s rule with a bootstrap posterior.

estimators is one mixle ParameterEstimator per arm (Gamma for waiting times, Gaussian for margins, a mixture for multi-modal rewards – anything with the accumulator contract).

Parameters:

• n_boot (int)

• mc_draws (int)

• seed (int | None)

class ThompsonBernoulli(n_arms, *, alpha=1.0, beta=1.0, seed=None)

Bases: _BanditBase

Beta-Bernoulli Thompson sampling. Rewards live in [0, 1]; fractional rewards contribute fractional pseudo-counts (the standard Bernoulli-moment update).

Parameters:

• n_arms (int)

• alpha (float)

• beta (float)

• seed (int | None)

class ThompsonGaussian(n_arms, *, mu0=0.0, kappa0=1.0e-2, alpha0=0.5, beta0=0.5, seed=None)

Bases: _BanditBase

Normal-Inverse-Gamma Thompson sampling: unknown mean AND variance per arm, so early optimism comes from honest posterior width rather than a tuned exploration constant.

Parameters:

• n_arms (int)

• mu0 (float)

• kappa0 (float)

• alpha0 (float)

• beta0 (float)

• seed (int | None)

class UCB1(n_arms, *, c=1.0, seed=None)

Bases: _BanditBase

The deterministic optimism baseline: play each arm once, then argmax mean_k + c * sqrt(2 ln t / n_k). Ties break to the lowest index; with no randomness anywhere, two UCB1 runs on the same reward sequence are identical.

Parameters:

• n_arms (int)

• c (float)

• seed (int | None)

class CollapseVerdict(ok, reason, scores=<factory>, diversities=<factory>, failed_round=None)

Bases: object

The result of collapse_monitor(): ok plus which check failed, and the raw series.

Parameters:

• ok (bool)

• reason (str | None)

• scores (list[float])

• diversities (list[float])

• failed_round (int | None)

collapse_monitor(history, *, score_key='score', candidates_key='candidates', diversity_fn=distinct_count_diversity, score_tol=0.0, diversity_tol=0.0)

Check a self-improvement round history for collapse: score non-decreasing and diversity not shrinking.

Each entry of history supplies the round’s held-out verified score under score_key and either its candidate pool under candidates_key (diversity computed via diversity_fn) or, when candidates_key is absent, a precomputed diversity number directly under "diversity". score_tol/diversity_tol allow a small, explicitly-named amount of round-to-round noise before a decrease/shrink counts as a real regression (0.0 = strict non-decreasing). The first round to violate either check ends the scan – reason names which check failed, failed_round where.

Parameters:

• history (Sequence[Mapping[str, Any]])

• score_key (str)

• candidates_key (str)

• diversity_fn (Callable[[Sequence[Any]], float])

• score_tol (float)

• diversity_tol (float)

Return type:

CollapseVerdict

distinct_count_diversity(candidates)

Diversity proxy: the number of distinct candidates (by str identity) in the round’s pool.

Parameters:

candidates (Sequence[Any])

Return type:

float

entropy_diversity(candidates)

Diversity proxy: Shannon entropy (nats) of the candidate-frequency distribution in the round’s pool.

Parameters:

candidates (Sequence[Any])

Return type:

float

class CostModel(c_frontier, c_local=0.0, c_label=0.0, train_cost=0.0)

Bases: object

Unit costs in any consistent currency.

Parameters:

• c_frontier (float)

• c_local (float)

• c_label (float)

• train_cost (float)

setup_cost(n_label)

Return the one-time label and training cost for a local model.

Parameters:

n_label (int)

Return type:

float

class DensityGate(featurizer, density=None, log_threshold=None)

Bases: object

A generative density over featurized inputs with a calibrated out-of-distribution floor on log p(x).

The featurizer is any transform(list) -> matrix: HashedNGram for text, or HashedRecord for dict/tuple records (so record models get the same OOD protection).

Parameters:

• featurizer (Any)

• density (Any)

• log_threshold (float | None)

fit(texts, *, n_components=4, alpha=0.02, max_its=60, min_covar=1e-3, seed=0)

Fit a diagonal-Gaussian mixture to the features and set the OOD floor at the alpha density quantile.

Parameters:

• texts (Sequence[str])

• n_components (int)

• alpha (float)

• max_its (int)

• min_covar (float)

• seed (int)

Return type:

DensityGate

log_density(texts)

log p(x) of each input under the fitted density (higher = more typical of training data).

Parameters:

texts (Sequence[str])

Return type:

ndarray

is_ood(text)

True when the input is atypical: log p(x) below the calibrated floor.

Parameters:

text (str)

Return type:

bool

ood_mask(texts)

Return a boolean mask marking inputs below the calibrated density floor.

Parameters:

texts (Sequence[str])

Return type:

ndarray

to_spec()

Serialize the featurizer, fitted density, and threshold for task artifacts.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Rebuild a density gate from to_spec() output.

Parameters:

spec (dict[str, Any])

Return type:

DensityGate

class DesignModel(signature, n_constraints, n_fingerprint=0)

Bases: object

A probabilistic model of the design space itself: design point -> (quality, budget violations).

Every evaluated design is a row; GP surrogates fitted on the rows drive propose() (the next design worth training, by feasibility-weighted expected improvement) and predict() (mean, sd, and probability-of-fitting-the-device for untrained designs). It serializes, so what was learned designing students for one task warm-starts the next – the design knowledge is itself a model artifact. (And distill_designer() compresses it into a student – models all the way down, each level a real artifact.)

Parameters:

• signature (str)

• n_constraints (int)

• n_fingerprint (int)

add(point, quality, violations, *, fingerprint=None, **tag)

Append one evaluated design point and its feasibility metadata.

Parameters:

• point (Any)

• quality (float)

• violations (Sequence[float])

• fingerprint (Sequence[float] | None)

• tag (Any)

Return type:

None

propose(bounds, *, seed=None, n_candidates=256, prefilter=None, max_tries=8, fingerprint=None)

The next design worth training: feasibility-weighted EI over everything seen so far.

prefilter closes the designer loop: pass a design judge – typically the compact student from distill_designer(), called as prefilter(point_tuple) -> label – and any proposal it labels "weak" is vetoed and re-drawn (fresh acquisition seed), up to max_tries. The distilled design knowledge thus skips known weak designs before a single training run is spent; if every retry is vetoed the last proposal is returned anyway (the judge advises, the surrogate decides). fingerprint conditions the proposal on the current task (see _fingerprint_bounds()); the returned point has design coords only.

Parameters:

• bounds (Sequence[tuple[float, float]])

• seed (Any)

• n_candidates (int)

• prefilter (Callable[[tuple], Any] | None)

• max_tries (int)

• fingerprint (Sequence[float] | None)

Return type:

ndarray

predict(points, *, fingerprint=None)

For untrained designs: predicted quality (mean, sd) and P(fits the device).

points carry design coords only; fingerprint (required when the ledger is fingerprinted) selects which task’s slice the prediction conditions on.

Parameters:

• points (Any)

• fingerprint (Sequence[float] | None)

Return type:

dict[str, ndarray]

to_json()

Serialize the design ledger for reuse across search runs.

Return type:

dict[str, Any]

classmethod from_json(d)

Reconstruct a design ledger from serialized JSON data.

Parameters:

d (dict[str, Any])

Return type:

DesignModel

class DesignedModel(estimator, spec, source, note='')

Bases: object

The model an LLM (or the fallback) designed: the estimator, the spec it built from, and the source.

Parameters:

• estimator (Any)

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

• source (str)

• note (str)

fit(data, **kwargs)

Fit the designed estimator with mixle.inference.optimize.

Parameters:

• data (Sequence[Any])

• kwargs (Any)

Return type:

Any

class DisagreementGate(classifier, threshold=0.5)

Bases: object

A fitted agree/disagree classifier over the student’s feature space, plus an escalation threshold.

Parameters:

• classifier (TaskModel)

• threshold (float)

disagreement_proba(texts)

P(disagree | x) under the fitted classifier.

Parameters:

texts (Sequence[str])

Return type:

ndarray

is_ood(text)

Return whether one input is predicted to disagree with the teacher.

Parameters:

text (str)

Return type:

bool

ood_mask(texts)

Same duck-typed shape as mixle.task.density.DensityGate.ood_mask() – drops straight into CalibratedTaskModel(..., density_gate=this).

Parameters:

texts (Sequence[str])

Return type:

ndarray

class UnionGate(*gates)

Bases: object

Escalate if ANY constituent gate flags an input – composes a DisagreementGate with a real DensityGate (or any other ood_mask-exposing gate) with no changes to either gate’s own code.

Parameters:

gates (Any)

ood_mask(texts)

Return the elementwise OR of all constituent gate masks.

Parameters:

texts (Sequence[str])

Return type:

ndarray

best_family(design, *, tag_key='family')

The single top-ranked recorded family, or None if nothing has been recorded yet.

Parameters:

• design (DesignModel)

• tag_key (str)

Return type:

str | None

rank_design_families(design, *, tag_key='family', candidates=None, default_score=float('-inf'))

Rank every family tag recorded in design by its mean quality, best first.

candidates, if given, are ALSO included in the ranking even if never recorded – an untried family gets default_score (-inf by default: no evidence ranks strictly below any recorded family, however weak, rather than being silently omitted or tied with a proven winner).

Parameters:

• design (DesignModel)

• tag_key (str)

• candidates (Sequence[str] | None)

• default_score (float)

Return type:

list[tuple[str, float]]

record_accepted_recipe(design, point, quality, violations, *, family, fingerprint=None, **tag)

Record an accepted structural recipe under its family tag – the training signal for rank_design_families(). A thin, named wrapper over DesignModel.add so callers do not have to remember which tag key the prior reads.

Parameters:

• design (DesignModel)

• point (Any)

• quality (float)

• violations (Sequence[float])

• family (str)

• fingerprint (Sequence[float] | None)

• tag (Any)

Return type:

None

class DeviceSpec(max_bytes=None, max_ops=None, torch_free=False)

Bases: object

A hard deployment budget. None leaves an axis unconstrained.

max_bytes: model size on flash/disk. max_ops: per-inference op budget (a latency proxy – calibrate ops/sec once per device to turn a latency target into this number). torch_free: the device cannot run torch, so only pure-mixle students qualify.

Parameters:

• max_bytes (int | None)

• max_ops (int | None)

• torch_free (bool)

classmethod for_latency(max_ms, ops_per_second, *, max_bytes=None, torch_free=False)

A budget from a latency target: max_ops = ops_per_second * max_ms / 1000.

ops_per_second must come from a probe run on the target device for the student kind you deploy (measure_ops_per_second() measures it for a representative student) – throughput differs by orders of magnitude across devices and student kinds, so there is no portable built-in constant.

Parameters:

• max_ms (float)

• ops_per_second (float)

• max_bytes (int | None)

• torch_free (bool)

Return type:

DeviceSpec

violations(fp)

Normalized constraint values, feasible when <= 0 (the form constrained BO consumes).

Parameters:

fp (EdgeFootprint)

Return type:

list[float]

feasible(fp)

Return whether a measured footprint satisfies device constraints.

Parameters:

fp (EdgeFootprint)

Return type:

bool

class EdgeDistillResult(model, family, recipe, agreement, footprint, feasible, pareto, design, trials=<factory>)

Bases: object

Outcome of an edge search: the winning student (with measured footprint), whether it truly fits the device, the (bytes, agreement) Pareto front over everything trained, and the updated DesignModel carrying the accumulated design knowledge.

Parameters:

• model (TaskModel)

• family (str)

• recipe (dict[str, Any])

• agreement (float)

• footprint (EdgeFootprint)

• feasible (bool)

• pareto (list[dict[str, Any]])

• design (DesignModel)

• trials (list[dict[str, Any]])

class EdgeFootprint(bytes, ops, torch_free)

Bases: object

A student’s measured deployment cost: serialized bytes, per-inference ops (multiply- accumulates for an MLP; factor evaluations for a structured classifier), and torch_free.

Parameters:

• bytes (int)

• ops (int)

• torch_free (bool)

class EdgeSpace(families=('mlp', 'structured'), dim_choices=(64, 128, 256, 512), hidden_range=(4, 96), epochs_range=(40, 320), log10_lr_range=(-3.0, -1.0), bits_choices=(32, 8), ngram=3, components_range=(1, 4), bins_range=(2, 8), max_its_range=(10, 60), min_gain_range=(0.0, 5.0))

Bases: object

One unit-cube design space over family (structure) and each family’s recipe (process).

Coordinate 0 selects the family; 1..4 decode family-specifically. families defaults to what the input kind and device allow: hashed-MLP students need torch; structured students need fixed-schema records. Decode is deterministic, so a design point is a reproducible recipe.

Parameters:

• families (tuple[str, ...])

• dim_choices (Sequence[int])

• hidden_range (tuple[int, int])

• epochs_range (tuple[int, int])

• log10_lr_range (tuple[float, float])

• bits_choices (Sequence[int])

• ngram (int)

• components_range (tuple[int, int])

• bins_range (tuple[int, int])

• max_its_range (tuple[int, int])

• min_gain_range (tuple[float, float])

dims()

Return the normalized design-space dimensionality.

Return type:

int

bounds()

Return normalized design-space bounds for DOE search.

Return type:

list[tuple[float, float]]

signature()

Fingerprint of the space so persisted design knowledge is only reused where it applies.

Return type:

str

decode(point)

Unit-cube point -> (family, recipe kwargs); the mlp recipe carries a bits precision.

Parameters:

point (ndarray)

Return type:

tuple[str, dict[str, Any]]

class Emulator(gp, x_train, y_train, bounds, target_fidelity, receipt)

Bases: object

A fitted forward surrogate: .predict, .escalate_mask, .receipt. Built by emulate().

Parameters:

• gp (Any)

• x_train (np.ndarray)

• y_train (np.ndarray)

• bounds (np.ndarray)

• target_fidelity (float | None)

• receipt (EmulatorReceipt)

predict(x)

Return (mean, std) of the surrogate’s posterior at x (always at the target fidelity).

Parameters:

x (Any)

Return type:

tuple[ndarray, ndarray]

escalate_mask(x, tol)

Return a boolean mask: True where the surrogate’s std at x exceeds tol (escalate).

Parameters:

• x (Any)

• tol (float)

Return type:

ndarray

class EmulatorReceipt(held_out_rmse, coverage, nominal_coverage, n_holdout, n_train, cost_spent, fidelities)

Bases: object

A measured, not asserted, report of an Emulator’s own quality.

held_out_rmse and coverage are computed against true-simulator calls that were not used to fit the surrogate (n_holdout of them, carved out of budget before training starts). coverage is the empirical fraction of holdout points whose true value falls within the emulator’s own mean +/- 1 std; nominal_coverage is what that fraction should be if the error bars are calibrated (~0.6827 for a Gaussian posterior). cost_spent is the total simulator cost actually used (holdout + training; each single-fidelity call costs 1, each multi-fidelity call costs its fidelity’s entry in costs).

Parameters:

• held_out_rmse (float)

• coverage (float)

• nominal_coverage (float)

• n_holdout (int)

• n_train (int)

• cost_spent (float)

• fidelities (tuple[float, ...] | None)

emulate(simulator, bounds, *, budget, fidelities=None, costs=None, seed=None, n_init=None, n_candidates=256, n_reference=128, holdout_frac=0.2, method='alc', fit_kwargs=None)

Fit a budget-limited GP surrogate of simulator over bounds, placing calls by acquisition.

simulator(x) (single fidelity) or simulator(x, s) (fidelities given, s one of them) returns the true response at x; budget is the total simulator cost available (single fidelity: 1 unit per call; multi-fidelity: costs per fidelity, default the fidelity value itself, mirroring mixle.doe.multifidelity.multi_fidelity_minimize()). A holdout_frac slice of the budget is spent up front on Latin-hypercube points evaluated at the target (highest) fidelity and held out of training, purely to compute EmulatorReceipt; the remainder trains the surrogate: single fidelity via mixle.doe.active.active_learning_design() (method "alc" or "alm"; "random" places a plain Latin-hypercube design instead, for comparison), multi-fidelity via ALC-at-target-fidelity point choice plus BOCA-style cost-aware fidelity choice (see the module docstring). Returns a fitted Emulator.

Parameters:

• simulator (Callable[[...], float])

• bounds (Sequence[tuple[float, float]])

• budget (int)

• fidelities (Sequence[float] | None)

• costs (Sequence[float] | None)

• seed (int | RandomState | None)

• n_init (int | None)

• n_candidates (int)

• n_reference (int)

• holdout_frac (float)

• method (str)

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

Return type:

Emulator

class ExecutionTrace(request, steps=<factory>)

Bases: object

An ordered list of TraceStep – JSON-serializable, so it can be stored (e.g. as a mixle.substrate "trace" item) and replayed in a fresh process.

Parameters:

• request (str)

• steps (list[TraceStep])

to_json()

Serialize the full execution trace to JSON-compatible data.

Return type:

dict[str, Any]

classmethod from_json(d)

Reconstruct an execution trace from JSON-compatible data.

Parameters:

d (dict[str, Any])

Return type:

ExecutionTrace

dumps()

Serialize the execution trace to a stable JSON string.

Return type:

str

class EmbeddingHeadIO(featurizer, labels)

Bases: _ClassifierIO

str -> label classifier over WordEmbeddingFeaturizer features – the “embedding_head” rung.

Parameters:

• featurizer (WordEmbeddingFeaturizer)

• labels (list[str])

class ExtractionIO(vocab, fields, *, max_len=128)

Bases: object

text -> {field: value}: tokenize, tag (BIO), decode spans back to substrings of the original text.

Parameters:

• vocab (dict[str, int])

• fields (list[str])

• max_len (int)

predict(module, text)

Extract fields from a single text record.

Parameters:

• module (Any)

• text (str)

Return type:

dict[str, str]

predict_batch(module, texts)

Extract fields from a batch of text records.

Parameters:

• module (Any)

• texts (list[str])

Return type:

list[dict[str, str]]

predict_with_confidence(module, texts)

Extract each record and a confidence in [0, 1]: the min per-token tag probability over tagged tokens.

A low confidence or a missing field is an explicit signal that the format may be unfamiliar and should be escalated. Returns 0.0 when nothing was tagged.

Parameters:

• module (Any)

• texts (list[str])

Return type:

list[tuple[dict[str, str], float]]

to_spec()

Serialize the extraction vocabulary, fields, and maximum sequence length.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Reconstruct extraction IO from an artifact spec.

Parameters:

spec (dict[str, Any])

Return type:

ExtractionIO

class Environment(*args, **kwargs)

Bases: Protocol

Generic act-observe world.

reset starts (or restarts) an episode from a seed and returns an initial observation; step applies one action and returns (observation, cost); action_space lists the actions currently legal to take. Costs are returned per step (not tracked internally) so interact() can enforce ONE budget semantics uniformly across arbitrary environments.

class ExplorationEnvironment(n_cells, n_targets, budget)

Bases: object

Thin Environment wrapper over ExplorationWorld.

Holds the episode config (cell/target/budget counts); reset(seed) builds a fresh ExplorationWorld and keeps it as self.world (so a caller – or the "eig" policy below, which reads ExplorationWorld internals exactly the way myopic_eig_policy() already does – can still get at the raw world). ExplorationWorld’s own public API is unmodified; this class only adapts it.

Parameters:

• n_cells (int)

• n_targets (int)

• budget (int)

class GaussianStreamingBelief(prior_mu=0.0, prior_sigma2=4.0, min_covar=0.05, belief_pseudo_count=0.05)

Bases: object

Per-cell streaming posterior over a scalar continuous latent (ExplorationWorld’s per-cell “geology” value), folded in one accepted survey observation at a time via mixle.inference.streaming.StreamingEstimator – the generic online sufficient- statistic machinery M0’s condition() is built to consume once a fitted model exists. One independent GaussianDistribution per cell; an unsurveyed cell reports the shared prior.

Parameters:

• prior_mu (float)

• prior_sigma2 (float)

• min_covar (float)

• belief_pseudo_count (float)

update(obs)

Fold one accepted survey observation’s prospectivity read into that cell’s belief. Drill/rejected/other observations carry no continuous read and are not folded in here – a drill resolves ground truth directly, it needs no posterior (v1 scope).

Parameters:

obs (dict[str, Any])

Return type:

None

credible_interval(cell, level=0.9)

A level-credible interval for the cell’s latent: the running Gaussian’s own mean, and a standard error of the mean built from a prior/sample-variance blend (see belief_pseudo_count) over this belief’s own read count – not the raw per-cell sample variance alone, which is degenerate (zero, before min_covar clamps it) at a single read and undercovers badly until several reads accumulate.

Parameters:

• cell (int)

• level (float)

Return type:

tuple[float, float]

class InteractionLog(seed, budget, policy, trace, total_cost, n_actions)

Bases: object

One episode’s action/observation/cost trace, replayable via mixle.task.replay.

Each recorded "act" step bundles POLICY DECISION + env.step + belief update as one unit (rather than recording the chosen action alone and replaying it against a bare env.step) because a world-peeking policy like "eig" (myopic_eig_policy() reads ExplorationWorld’s own RNG-backed prospectivity() while DECIDING) consumes the same environment randomness the eventual observation depends on – replaying only the action list would silently desync that RNG stream and stop reproducing bit-for-bit. Bundling the policy call into the replayed unit keeps the two draws in the same relative order both times.

Parameters:

• seed (int | None)

• budget (float)

• policy (str)

• trace (ExecutionTrace)

• total_cost (float)

• n_actions (int)

is_deterministic(env, belief_model)

Replay this log against a fresh env/belief_model pair (same policy name, same seed) and confirm every recorded step reproduces exactly – the M1 replay receipt. Only named policies ("eig", "greedy") can be reconstructed for replay; a log built from an arbitrary callable policy cannot (the callable itself is not serialized).

Parameters:

• env (Environment)

• belief_model (Any)

Return type:

bool

interact(env, belief_model, *, policy='eig', budget, seed=None)

Drive the act-observe-update loop.

Resets env, then repeatedly: pick an action over env.action_space() (EIG / belief- greedy / a caller callable), execute it via env.step, fold the observation into belief_model.update(obs), until the summed action cost would exceed budget or the policy/environment stops (action_space() empty, policy returns None, or the environment refuses the action). Every reset/act is recorded as a TraceStep (see InteractionLog for why policy decision + step are bundled into one "act" unit) so the returned InteractionLog replays deterministically via mixle.task.replay.

Parameters:

• env (Environment)

• belief_model (Any)

• policy (str | Callable[[Environment, Any, list[Any]], Any])

• budget (float)

• seed (int | None)

Return type:

InteractionLog

class EpisodeResult(score, n_actions, trace=<factory>)

Bases: object

Score, action count, and trace captured from one exploration episode.

Parameters:

• score (int)

• n_actions (int)

• trace (list[dict[str, Any]])

class ExplorationWorld(n_cells, n_targets, budget, seed=0)

Bases: object

One episode over a synthetic mineral-style exploration world: n_cells candidate sites, n_targets of them hidden true targets, each cell’s TRUE target status correlated with a latent “geology” feature that a survey partially reveals as a noisy prospectivity reading.

Parameters:

• n_cells (int)

• n_targets (int)

• budget (int)

• seed (int)

prospectivity(cell)

The world’s own current noisy read of cell – what a policy actually gets to see.

Parameters:

cell (int)

Return type:

float

step(action)

Apply one typed action (plain dict, so a fitted plan model can score/sample over the same action vocabulary): {"type": "survey", "cell": i} or {"type": "drill", "cell": i}. Returns a plain-dict observation. Raises nothing on an over-budget action – it is simply refused (recorded, zero effect) once done, so a policy that keeps acting past budget exhaustion degrades gracefully rather than crashing.

Parameters:

action (dict[str, Any])

Return type:

dict[str, Any]

score()

Targets correctly identified so far: distinct true-target cells actually drilled.

Return type:

int

action_menu()

Every action a policy could take right now (undrilled cells only, for drills).

Return type:

list[dict[str, Any]]

greedy_prospectivity_policy(world)

Survey every undrilled cell once (low-cost information), then drill highest-read-prospectivity cells first – a fixed heuristic baseline for learned or diagnosis-directed policies.

Parameters:

world (ExplorationWorld)

Return type:

dict[str, Any] | None

random_policy(world)

Choose a random currently valid action from the world’s action menu.

Parameters:

world (ExplorationWorld)

Return type:

dict[str, Any] | None

run_episode(policy, *, n_cells, n_targets, budget, seed)

Drive policy(world) -> action (a plain dict, or None to end early) until the world’s budget is exhausted or the policy stops itself.

Parameters:

• n_cells (int)

• n_targets (int)

• budget (int)

• seed (int)

Return type:

EpisodeResult

class ExtractorHarness(model, teacher, fields, required, holdout_f1, n_fallback=0, n_requests=0)

Bases: object

A distilled extractor in front of the parser it replaces: local extraction or teacher fallback.

Parameters:

• model (Any)

• teacher (Callable[[str], dict])

• fields (list[str])

• required (list[str])

• holdout_f1 (float)

• n_fallback (int)

• n_requests (int)

report()

Return extraction holdout quality and fallback metrics.

Return type:

dict[str, Any]

save(path)

Persist the wrapped extraction model artifact.

Parameters:

path (str)

Return type:

str

class MatcherHarness(solution, teacher)

Bases: object

A calibrated pair-matcher in front of the rule it replaces.

Parameters:

• solution (Solution)

• teacher (Callable[[Any, Any], Any])

property holdout_agreement: float

Return held-out agreement of the pairwise matcher solution.

report()

Return the underlying matcher solution report.

Return type:

dict[str, Any]

class CeilingReport(held_out_score, target, met)

Bases: object

Whether the CURRENT structural class meets target on held-out data – the capacity ladder’s verdict, computed once before any new structure is proposed.

Parameters:

• held_out_score (float)

• target (float)

• met (bool)

class ImagineResult(ceiling, verdicts=<factory>, breaks_ceiling=None)

Bases: object

Capacity ceiling result plus candidate verdicts from structural imagination.

Parameters:

• ceiling (CeilingReport)

• verdicts (list[ProposalVerdict])

• breaks_ceiling (str | None)

class ProposalVerdict(name, accepted, train_score, held_out_score, reason='')

Bases: object

Evaluation verdict for one proposed structural candidate.

Parameters:

• name (str)

• accepted (bool)

• train_score (float)

• held_out_score (float)

• reason (str)

class StructuralCandidate(name, fit, new_information='')

Bases: object

One proposed richer structure. new_information MUST name the specific capability the starting class provably lacks (e.g. “2-component mixture: represents a bimodal posterior a single Gaussian cannot”) – empty/None means “no new information source” and the candidate is rejected regardless of any measured improvement.

Parameters:

• name (str)

• fit (Callable[[Sequence[Any]], Any])

• new_information (str)

ceiling_report(held_out_score, target)

Return whether held-out score reaches the requested target.

Parameters:

• held_out_score (float)

• target (float)

Return type:

CeilingReport

propose_structure(candidates, train, held_out, ceiling)

Fit and verify each candidate in order. A candidate is accepted only if it names a genuine new information source and improves held-out score over the ceiling’s own held-out score (never train alone, since a richer family can always fit train better without a real capability gain). The first accepted candidate that also reaches ceiling.target breaks the ceiling.

Parameters:

• candidates (Sequence[StructuralCandidate])

• train (Sequence[Any])

• held_out (Sequence[Any])

• ceiling (CeilingReport)

Return type:

ImagineResult

class InverseModel(*, module, prior, simulator, family, theta_dim, y_dim, receipts, seed=None)

Bases: object

A fitted amortized posterior q(theta | y) plus its InverseReceipts.

Parameters:

• module (Any)

• prior (Any)

• simulator (Callable[[Any], Any])

• family (str)

• theta_dim (int)

• y_dim (int)

• receipts (InverseReceipts)

• seed (int | None)

posterior(y)

Wrap q(theta | y) as an M0 Posterior: sample(n) / log_density(theta) / mean(field) / .receipt – so downstream condition/do composition treats a learned inverse like an exactly-conditioned one, modulo the amortization warning on .receipt and the InverseReceipts pointer at .receipt.inverse_receipts.

Parameters:

y (Any)

Return type:

Posterior

class InverseReceipts(sbc_statistic, sbc_pvalue, sbc_bins, sbc_replications, sbc_pass, coverage, coverage_pass, prior_predictive, rounds_trained, sharpness_by_round=<factory>, ess=None, ess_ratio=None, warnings=<factory>)

Bases: object

The calibration report that ships with every InverseModel – tells the caller whether to trust q(theta | y), not just a point estimate.

Parameters:

• sbc_statistic (float)

• sbc_pvalue (float)

• sbc_bins (int)

• sbc_replications (int)

• sbc_pass (bool)

• coverage (dict[float, float])

• coverage_pass (dict[float, bool])

• prior_predictive (dict[str, Any])

• rounds_trained (int)

• sharpness_by_round (list[float])

• ess (float | None)

• ess_ratio (float | None)

• warnings (list[str])

learn_inverse(simulator, prior, *, family='flow', n_sims=2000, rounds=1, n_sbc_replications=200, coverage_levels=(0.5, 0.9), reweight=False, true_log_likelihood=None, y_obs=None, seed=None, m_steps=200, lr=5e-3, max_its=1, hidden=32, n_posterior_samples=200, n_reweight_samples=500)

Learn an amortized posterior q(theta | y) for simulator g: theta -> y under prior p(theta). See the module docstring for the full algorithm and the calibration receipts computed unconditionally.

family="flow" (build_conditional_flow) requires theta (the quantity being inferred, the student’s y-argument) to be >= 2-dimensional – build_conditional_flow needs y_dim >= 2 for its coupling layers to be non-trivial (see its own docstring). A 1-D theta (e.g. a scalar-parameter inverse problem) must use family="mdn", which has no such restriction (a mixture of per-component Gaussians is well-defined for scalar theta too, and is the more direct fit for asserting multimodality component-by-component).

rounds > 1 (SNPE-style sequential refinement toward a SPECIFIC observation) requires y_obs: round 1 alone (unconditional pair generation) is the only round that has meaning without an observation to sharpen against.

Parameters:

• simulator (Callable[[Any], Any])

• prior (Any)

• family (str)

• n_sims (int)

• rounds (int)

• n_sbc_replications (int)

• coverage_levels (tuple[float, ...])

• reweight (bool)

• true_log_likelihood (Callable[[Any, Any], float] | None)

• y_obs (Any)

• seed (int | None)

• m_steps (int)

• lr (float)

• max_its (int)

• hidden (int)

• n_posterior_samples (int)

• n_reweight_samples (int)

Return type:

InverseModel

class FieldChoice(path, kind, family, runner_up, gap_bits)

Bases: object

The family chosen for one field, the runner-up, and how decisive the choice was (bits/obs).

Parameters:

• path (str)

• kind (str)

• family (str)

• runner_up (str | None)

• gap_bits (float | None)

property confident: bool

Confident when the family is type-determined (no real contender) or clears the runner-up by a margin.

class GenerativeTextIO(labels, vocab, log_prior)

Bases: object

Adapter over {label: fitted p(tokens|label)} + log-priors: exact posteriors and log p(x).

Parameters:

• labels (list[str])

• vocab (list[str])

• log_prior (list[float])

logits_batch(model, raw_inputs)

log P(tokens, label) per label – an (m, K) matrix (multinomial: sum of token logs).

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

proba_batch(model, raw_inputs)

The exact class posterior (softmax of log-joints; the shared evidence cancels).

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

log_evidence(model, raw_inputs)

Per-token log p(x) (length-normalized) – the built-in typicality/OOD score.

Raw document evidence scales with length (a short gibberish string would outrank a long in-domain one), so typicality is reported per token: mean log-probability under the full generative model.

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

predict_batch(model, raw_inputs)

Return the highest-scoring generative class for each input.

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

list[str]

predict(model, raw_input)

Return the highest-scoring generative class for one input.

Parameters:

• model (Any)

• raw_input (Any)

Return type:

str

to_spec()

Serialize the generative text adapter.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Reconstruct the generative text adapter from a spec.

Parameters:

spec (dict[str, Any])

Return type:

GenerativeTextIO

extractive_capture_profile(student, teacher, texts, suite, *, fields)

The extraction-student capture profile: F1-against-gold and schema validity, not exact-match agreement.

gold is the teacher’s own extraction on the clean texts – the true answer a corruption should not change. Reports, JSON-serializable:

• "clean_f1" – student F1 against gold on clean text (teacher’s own clean F1 against its own gold is trivially 1.0 and omitted);

• "corruptions" – per corruption name, {"student_f1", "teacher_f1"} against the same fixed gold – both sides scored against the same ground truth, so a comparison is meaningful;

• "invariances" – per invariance name, {"student_f1", "teacher_f1"} between each side’s clean prediction and its prediction on the rewritten text (1.0 = perfectly invariant);

• "schema_validity" – {"student", "teacher"} fraction of clean-text extractions that are complete and grounded (validate_extraction_schema()) – an executable check, never eyeballed;

• "abstention" – as in capture_profile(), if either side exposes a decision API.

Parameters:

• student (Any)

• teacher (Any)

• texts (Sequence[str])

• suite (CapabilitySuite)

• fields (Sequence[str])

Return type:

dict[str, Any]

validate_extraction_schema(record, source_text, fields)

Executable schema check for one extracted record: complete (every expected field present) and grounded (every non-empty value is an actual substring of source_text, not hallucinated).

Returns a plain dict (complete, grounded, missing, ungrounded) – never a single pass/fail bit, so a caller can see exactly what failed.

Parameters:

• record (dict[str, Any])

• source_text (str)

• fields (Sequence[str])

Return type:

dict[str, Any]

class HashedNGram(n=3, dim=256, seed=0)

Bases: object

Map a string to a fixed-width float vector by hashing its character n-grams into dim buckets.

The featurizer is deterministic and dependency-free. It serializes as three scalar settings and rebuilds without a fitted vocabulary or external tokenizer. Counts are L2-normalized per row.

Parameters:

• n (int)

• dim (int)

• seed (int)

transform(texts)

Return L2-normalized hashed n-gram feature rows for texts.

Parameters:

texts (list[str])

Return type:

ndarray

to_spec()

Return the serializable featurizer configuration.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Rebuild a featurizer from to_spec() output.

Parameters:

spec (dict[str, Any])

Return type:

HashedNGram

class HashedRecord(dim=256, seed=0)

Bases: object

Map a heterogeneous record to a fixed-width hashed feature vector.

Each tuple position or dictionary key owns a hashed namespace. Categorical, string, and boolean values contribute an indicator feature; numeric values contribute a bounded value feature and a presence feature. The transform is stateless and deterministic, so it serializes as two scalar settings and rebuilds without a fitted encoder or vocabulary.

Parameters:

• dim (int)

• seed (int)

transform(records)

Return L2-normalized hashed feature rows for heterogeneous records.

Parameters:

records (list[Any])

Return type:

ndarray

to_spec()

Return the serializable record-featurizer configuration.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Rebuild a record featurizer from to_spec() output.

Parameters:

spec (dict[str, Any])

Return type:

HashedRecord

class LNSStructuredClassifierIO(field_keys, label_index, labels, step=1e-2)

Bases: StructuredClassifierIO

The structured classifier executed in the log-number system: integers above the leaf boundary.

A structured student’s per-label score is a sum of factor log-densities – in log-space that is products of probabilities, which is exactly what LogNumberSystem runs on integers: each factor’s log-density is quantized once at the leaf boundary (k = round(logp / step)), then the per-label accumulation is integer ADDs, mixture components fold with the integer logadd LUT, the classification is an integer argmax, and the posterior is the integer log-softmax of mixle.engines.lns_nn – no exp/log anywhere above the leaves (one exp only if you ask for linear-scale probabilities). The dequantized scores match the float classifier within the engine’s documented bound (~``1.5 * step`` per fold), so step is a dial between integer-width and fidelity.

Categorical factors are pre-quantized to integer tables at first use, so their leaves are pure integer lookups – on an all-discrete schema inference touches no floats at all. Continuous leaves evaluate in float and quantize at the boundary, the same contract as the engine’s SumProductCircuit.

Parameters:

• field_keys (list[str] | None)

• label_index (int)

• labels (list[str])

• step (float)

int_logits_batch(model, raw_inputs)

Per-label INTEGER log-joint scores (m, K) – the whole combination is integer math.

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

logits_batch(model, raw_inputs)

Return floating logit values decoded from integer log-space scores.

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

proba_batch(model, raw_inputs)

Posterior via the INTEGER log-softmax (max + LUT); one exp at the very end for linear scale.

The LUT rounds each log-probability to ~``step``, so the raw exp sums to 1 +/- K*step/2; the final float renormalization (free – we already left integer space for the exp) removes that systematic drift without touching the integer pipeline.

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

predict_batch(model, raw_inputs)

Return integer-logit argmax labels for a batch of raw inputs.

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

list[str]

to_spec()

Serialize the LNS structured-classifier adapter.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Reconstruct the LNS structured-classifier adapter from a spec.

Parameters:

spec (dict[str, Any])

Return type:

LNSStructuredClassifierIO

class LadderResult(target, rungs, winner)

Bases: object

The ladder’s outcome: every rung’s measured score, and the smallest rung meeting target (or None).

Parameters:

• target (float)

• rungs (list[RungResult])

• winner (str | None)

ceiling(rung)

The measured score of rung, or None if that rung was unavailable in this environment.

Parameters:

rung (str)

Return type:

float | None

class ModelRecommendation(estimator, fields, dependencies, warnings, profile=None)

Bases: object

A model shape recommended from data: estimator, per-field choices+confidence, dependencies, and notes.

Parameters:

• estimator (Any)

• fields (list[FieldChoice])

• dependencies (list[tuple[str, str, float]])

• warnings (list[str])

• profile (Any)

low_confidence_fields()

Fields whose family choice is not yet decisive – where more data would most sharpen the model.

Return type:

list[FieldChoice]

fit(data, **kwargs)

Fit the recommended estimator on data and return the model.

Parameters:

• data (Sequence[Any])

• kwargs (Any)

Return type:

Any

explain()

Plain-language lines: the underlying profile’s explanation (families, bits, dependencies, warnings).

Return type:

list[str]

class OpenAICompatLLM(base_url, model, *, api_key=None, temperature=0.0, max_tokens=512, timeout=60.0)

Bases: object

An LLM backed by any OpenAI-compatible /v1/chat/completions endpoint (stdlib HTTP only).

Parameters:

• base_url (str)

• model (str)

• api_key (str | None)

• temperature (float)

• max_tokens (int)

• timeout (float)

complete(prompt, *, system=None, **kwargs)

Call an OpenAI-compatible chat-completions endpoint and return message text.

Parameters:

• prompt (str)

• system (str | None)

• kwargs (Any)

Return type:

str

class OpenAICompatVLM(base_url, model, *, api_key=None, top_logprobs=20, timeout=60.0, continue_key='continue_final_message', continue_value=True, extra_body=None)

Bases: object

A VLM backed by an OpenAI-compatible /v1/chat/completions endpoint that returns real per-token logprobs for an open-weight vision-language model (a vLLM- or TGI-served LLaVA / Qwen-VL / … deployment). See the module docstring for why this deliberately does not target proprietary hosted vision APIs.

Continuing a partial completion (every next_logprobs call after the first token of a decode) needs the server to prefill the given prefix rather than start generation fresh; this uses vLLM’s continue_final_message extension by default (append the prefix as a partial assistant message, set that flag). Pass continue_key/continue_value to target a server with a different convention.

Parameters:

• base_url (str)

• model (str)

• api_key (str | None)

• top_logprobs (int)

• timeout (float)

• continue_key (str)

• continue_value (Any)

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

next_logprobs(image, prefix, *, prompt, system=None)

One image-conditioned next-token distribution given the tokens generated so far (prefix).

Parameters:

• image (Any)

• prefix (tuple[str, ...])

• prompt (str)

• system (str | None)

Return type:

list[tuple[str, float]]

next_logprobs_for(image, prompt, *, system=None)

Bind image/prompt into the next_logprobs(prefix) -> [(token, log_prob), ...] shape mixle.enumeration.best_first_decode() / mixle.enumeration.quantized_best_first_decode() expect directly – the whole bridge from “an image and a question” to “enumerate the top-k answers”.

Parameters:

• image (Any)

• prompt (str)

• system (str | None)

Return type:

Callable[[tuple[str, …]], Iterable[tuple[str, float]]]

class CallableVLM(fn)

Bases: object

Wrap a plain fn(image, prefix) -> [(token, log_prob), ...] as a VLM – local models and tests.

Parameters:

fn (Callable[[Any, tuple[str, ...]], Iterable[tuple[str, float]]])

next_logprobs_for(image)

Bind image into the next_logprobs(prefix) shape mixle.enumeration expects directly.

Parameters:

image (Any)

Return type:

Callable[[tuple[str, …]], Iterable[tuple[str, float]]]

score_candidate(next_logprobs_fn, candidate_tokens)

Teacher-forced total log-probability of candidate_tokens under next_logprobs_fn.

Walks one token at a time, reading off the ACTUAL log-probability of the candidate’s own next token at each step – never approximated or guessed. If a step’s returned continuations do not include the candidate’s token (e.g. it fell outside top_logprobs), returns -inf rather than silently dropping or padding the score with a made-up value: that is a real “this candidate wasn’t even considered by the model at that step” outcome, not a bug to hide.

Parameters:

• next_logprobs_fn (Callable[[tuple[str, ...]], Iterable[tuple[str, float]]])

• candidate_tokens (Sequence[str])

Return type:

float

score_fn_for(next_logprobs_fn)

Bind a next_logprobs function into the score(candidate) -> float shape mixle.enumeration.top_k_scored() expects directly, for ranking a fixed candidate set.

Parameters:

next_logprobs_fn (Callable[[tuple[str, ...]], Iterable[tuple[str, float]]])

Return type:

Callable[[Sequence[str]], float]

class QuantizedClassifierIO(featurizer, labels)

Bases: _ClassifierIO

The classifier IO for quantized students: same featurize -> logits -> label contract, no torch.

Parameters:

• featurizer (Any)

• labels (list[str])

logits_batch(model, raw_inputs)

Featurize raw inputs and return quantized-model logits.

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

to_spec()

Serialize the quantized classifier IO adapter.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Reconstruct the quantized classifier IO adapter from a spec.

Parameters:

spec (dict[str, Any])

Return type:

QuantizedClassifierIO

class QuantizedMLP(layers, *, bits=8)

Bases: object

A quantized-weight MLP with a pure-numpy forward pass.

layers is [(W_int (out, in), scale fp32, bias fp32 (out,)), ...] with weights in the symmetric bits range (int8: [-127, 127]; int4: [-7, 7], stored nibble-packed on disk); the forward is x @ (W * s).T + b with ReLU between layers – exactly the dequantized version of the trained torch stack, so its logits match torch-on-dequantized-weights to float tolerance.

Parameters:

• layers (list[tuple[np.ndarray, float, np.ndarray]])

• bits (int)

logits(feats)

Compute dequantized logits for a feature matrix.

Parameters:

feats (ndarray)

Return type:

ndarray

nbytes()

Deployable payload bytes: packed weights (1 B/weight at int8, 1/2 B at int4) + fp32 biases + one fp32 scale per layer.

Return type:

int

macs()

Per-inference multiply-accumulates (integer x fp32 dequant multiplies count the same).

Return type:

int

to_arrays()

Serialize the quantized layers into artifact-ready NumPy arrays.

Return type:

dict[str, ndarray]

classmethod from_arrays(arrays)

Reconstruct a quantized MLP from artifact array payloads.

Parameters:

arrays (dict[str, ndarray])

Return type:

QuantizedMLP

class EditTrial(edge, held_out_score, verified)

Bases: object

Held-out result for one proposed graph edit.

Parameters:

• edge (tuple[int, int])

• held_out_score (float)

• verified (bool)

class SearchOutcome(trials, found_edge, final_model, history=<factory>)

Bases: object

Final refinement state plus the verified edit-search history.

Parameters:

• trials (int)

• found_edge (tuple[int, int] | None)

• final_model (HeterogeneousBayesianNetwork)

• history (list[EditTrial])

apply_edge(model, edge, train_data)

Refit the named child factor as a linear-Gaussian conditional.

Every other factor is kept unchanged, so the returned model represents only the proposed edge edit.

Parameters:

• model (HeterogeneousBayesianNetwork)

• edge (tuple[int, int])

• train_data (Sequence[tuple])

Return type:

HeterogeneousBayesianNetwork

blind_structure_search(model, train_data, held_out, edit_space, *, target)

Try candidate edges in order and accept only verified held-out gains.

Parameters:

• model (HeterogeneousBayesianNetwork)

• train_data (Sequence[tuple])

• held_out (Sequence[tuple])

• edit_space (Sequence[tuple[int, int]])

• target (float)

Return type:

SearchOutcome

diagnosis_directed_correction(model, train_data, failing_cases, held_out, *, background=None, target)

Diagnose the fault from failing_cases, apply only its suggested edge (trying both parent-child orientations of the named pair, since diagnose reports an undirected co-anomaly), and verify held-out improvement before accepting – one trial if the diagnosis names the right pair and orientation, honestly more (or a refusal) if it does not.

Parameters:

• model (HeterogeneousBayesianNetwork)

• train_data (Sequence[tuple])

• failing_cases (Sequence[tuple])

• held_out (Sequence[tuple])

• background (Sequence[tuple] | None)

• target (float)

Return type:

SearchOutcome

fit_independent_baseline(train_data)

Fit an independent network with one marginal Gaussian per field.

Parameters:

train_data (Sequence[tuple])

Return type:

HeterogeneousBayesianNetwork

class RecipeSpace(dim_choices=(128, 256, 512, 1024), hidden_range=(16, 128), epochs_range=(50, 400), log10_lr_range=(-3.0, -1.0), n=4)

Bases: object

The tunable axes of a distillation recipe and how a unit-cube point decodes into concrete knobs.

Parameters:

• dim_choices (Sequence[int])

• hidden_range (tuple[int, int])

• epochs_range (tuple[int, int])

• log10_lr_range (tuple[float, float])

• n (int)

dims()

Return the normalized recipe-search dimensionality.

Return type:

int

decode(point)

Decode a normalized design point into a distillation recipe.

Parameters:

point (ndarray)

Return type:

dict[str, Any]

cost(recipe)

Relative training cost of a recipe in [0, 1] (params x steps, normalized by the space’s max).

Parameters:

recipe (dict[str, Any])

Return type:

float

bounds()

Return normalized DOE bounds for recipe search.

Return type:

list[tuple[float, float]]

class RecordClassifierIO(featurizer, labels)

Bases: _ClassifierIO

record -> label: hashed-record features into a small classifier (tuples/dicts of mixed fields).

Parameters:

• featurizer (HashedRecord)

• labels (list[str])

class RoutePlan(route, volume, per_request, total, savings_vs_frontier, p_escalate, break_even, options)

Bases: object

Costed route comparison for a fixed request volume.

Parameters:

• route (str)

• volume (int)

• per_request (float)

• total (float)

• savings_vs_frontier (float)

• p_escalate (float)

• break_even (float)

• options (dict[str, float])

class Router(tiers)

Bases: object

Route each request to the lowest-cost tier whose calibrated model is confident.

Parameters:

tiers (list[tuple[str, Any, float]])

classmethod from_solutions(solutions, teacher, *, costs, names=None)

Build from Solution objects ordered by cost plus the teacher callable.

costs has one entry per solution plus one for the teacher (per-request).

Parameters:

• solutions (list)

• teacher (Any)

• costs (list[float])

• names (list[str] | None)

Return type:

Router

serve(xs)

Route a batch of requests and return the tier-selected answers.

Parameters:

xs (Any)

Return type:

list[Any]

harvested()

Return teacher-answered (inputs, labels) for retraining lower-cost tiers.

Return type:

tuple[list[Any], list[Any]]

report()

Return per-tier traffic and realized economics.

Return type:

dict[str, Any]

summary()

Render a compact human-readable traffic and cost summary.

Return type:

str

class RungResult(rung, score, model, note='')

Bases: object

One rung’s measured outcome: its held-out agreement score, the fitted student (if built), and a note.

Parameters:

• rung (str)

• score (float | None)

• model (TaskModel | None)

• note (str)

class Scorecard(task, n_test, end_to_end_accuracy, local_agreement, escalation_rate, student_p50_ms, student_p95_ms, teacher_p50_ms, artifact_bytes, student_cost_per_1k, teacher_cost_per_1k)

Bases: object

Evaluation summary for a distilled task service.

Parameters:

• task (str)

• n_test (int)

• end_to_end_accuracy (float)

• local_agreement (float)

• escalation_rate (float)

• student_p50_ms (float)

• student_p95_ms (float)

• teacher_p50_ms (float)

• artifact_bytes (int | None)

• student_cost_per_1k (float | None)

• teacher_cost_per_1k (float | None)

as_dict()

Return the scorecard fields as a plain dictionary.

Return type:

dict[str, Any]

table()

Render a compact comparison table for local and teacher service metrics.

Return type:

str

class RouterStats(tiers=<factory>, harvested_inputs=<factory>, harvested_labels=<factory>, degraded=<factory>)

Bases: object

Mutable accounting for routed requests, harvested labels, and degraded tier calls.

Parameters:

• tiers (list[TierStats])

• harvested_inputs (list[Any])

• harvested_labels (list[Any])

• degraded (list[DegradedResult])

property n_requests: int

Return the total number of requests answered across all tiers.

class HarvestResolveResult(accepted, n_harvested, escalation_before, escalation_after, escalation_drop, agreement, router=None, tier_name='')

Bases: object

Receipt from resolve_from_harvest().

escalation_before is exactly 1.0: every harvested input, by definition, escalated all the way to the teacher under the current router. escalation_after is the new tier’s own calibrated escalation rate on a held-out split of that same harvested set; escalation_drop is the difference. router is the new stack with the tier inserted (None when nothing was accepted because there is too little harvested data to fit/calibrate or the new tier does not escalate measurably less often than always-escalate, in which case it buys nothing and is rejected).

Parameters:

• accepted (bool)

• n_harvested (int)

• escalation_before (float)

• escalation_after (float)

• escalation_drop (float)

• agreement (float)

• router (Router | None)

• tier_name (str)

resolve_from_harvest(router, *, cost_per_request, name='resolved', alpha=0.1, holdout=0.25, min_drop=0.05, distill_kw=None, seed=0)

Train a new router tier from harvested teacher labels.

Every harvested input escalated through the existing tiers, so the baseline escalation rate on that set is 1.0. A new tier is fit and calibrated on a held-out split of the same harvested set without re-calling the teacher. It is inserted only if its calibrated escalation rate drops by at least min_drop below 1.0 on that split.

Parameters:

• router (Router)

• cost_per_request (float)

• name (str)

• alpha (float)

• holdout (float)

• min_drop (float)

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

• seed (int)

Return type:

HarvestResolveResult

class RegressionSolution(net, featurizer, teacher, qhat, alpha, tol, holdout_mae, y_mean, y_scale, train_inputs=<factory>, train_ys=<factory>, cal_inputs=<factory>, cal_ys=<factory>, hidden=(64, ), epochs=300, lr=0.01, seed=0, n_requests=0, n_escalated=0, harvested_inputs=<factory>, harvested_ys=<factory>)

Bases: object

A calibrated numeric student in front of the routine it replaces.

Parameters:

• net (Any)

• featurizer (Any)

• teacher (Callable[[...], Any])

• qhat (float)

• alpha (float)

• tol (float)

• holdout_mae (float)

• y_mean (float)

• y_scale (float)

• train_inputs (list)

• train_ys (list)

• cal_inputs (list)

• cal_ys (list)

• hidden (tuple)

• epochs (int)

• lr (float)

• seed (int)

• n_requests (int)

• n_escalated (int)

• harvested_inputs (list)

• harvested_ys (list)

interval(x)

Return (yhat, lo, hi) with calibrated teacher-answer coverage.

Parameters:

x (Any)

Return type:

tuple[float, float, float]

property answers_locally: bool

Whether the calibrated precision meets the tolerance at all (else everything escalates).

decide(x)

Return the calibrated point estimate when local precision is sufficient.

If answers_locally is false, return None to signal escalation. Unlike __call__, this method never falls through to the teacher itself, so a Router tier can decide whether to escalate to the next tier.

Parameters:

x (Any)

Return type:

float | None

report()

Return calibration, precision, request, and harvest metrics.

Return type:

dict[str, Any]

save(path)

Persist the network, featurizer, and calibration metadata.

Parameters:

path (str)

Return type:

str

classmethod load(path, teacher, *, device='cpu')

Reconstitute a serving RegressionSolution (no training/calibration data; improve() raises).

Parameters:

• path (str)

• teacher (Callable[[...], Any])

• device (str)

Return type:

RegressionSolution

improve()

Re-fit with harvested pairs; promote only if the calibrated width shrinks (anti-regression).

Return type:

bool

class MultiLabelSolution(net, featurizer, labels, teacher, upper_absent, lower_present, alpha, holdout_set_agreement, train_inputs=<factory>, train_sets=<factory>, cal_inputs=<factory>, cal_sets=<factory>, hidden=(64, ), epochs=300, lr=0.01, seed=0, n_requests=0, n_escalated=0, harvested_inputs=<factory>, harvested_sets=<factory>)

Bases: object

A per-label-calibrated tagger in front of the routine it replaces.

Parameters:

• net (Any)

• featurizer (Any)

• labels (list[str])

• teacher (Callable[[...], Any])

• upper_absent (ndarray)

• lower_present (ndarray)

• alpha (float)

• holdout_set_agreement (float)

• train_inputs (list)

• train_sets (list)

• cal_inputs (list)

• cal_sets (list)

• hidden (tuple)

• epochs (int)

• lr (float)

• seed (int)

• n_requests (int)

• n_escalated (int)

• harvested_inputs (list)

• harvested_sets (list)

try_local(x)

The decided label set, or None when any label is ambiguous (= must escalate).

Parameters:

x (Any)

Return type:

list[str] | None

decide(x)

Return the local multilabel decision, or None when the example should escalate.

Parameters:

x (Any)

Return type:

list[str] | None

report()

Return multi-label agreement, escalation, and harvest metrics.

Return type:

dict[str, Any]

save(path)

Persist net + featurizer + per-label bars; load() restores a serving tagger.

Parameters:

path (str)

Return type:

str

classmethod load(path, teacher, *, device='cpu')

Reconstitute a serving MultiLabelSolution (no training/calibration data; improve() raises).

Parameters:

• path (str)

• teacher (Callable[[...], Any])

• device (str)

Return type:

MultiLabelSolution

improve()

Re-fit with harvested sets; promote only if held-out set agreement does not regress.

Return type:

bool

class OrchestrationResult(answer, trace, stopped_reason)

Bases: object

Final answer, execution trace, and stop reason from an orchestration run.

Parameters:

• answer (Any)

• trace (ExecutionTrace)

• stopped_reason (str)

class World(*args, **kwargs)

Bases: Protocol

The minimal environment contract orchestrate needs.

step(action)

Apply one action and return the environment’s step result.

Parameters:

action (dict[str, Any])

Return type:

Any

property done: bool

Whether the environment has reached a terminal state.

score()

Return the environment’s current score or outcome metric.

Return type:

Any

class OutcomeTrainedDecomposer(plan_model, imitation_model, rounds=<factory>)

Bases: object

Outcome-trained plan model, baseline imitation model, and per-round statistics.

Parameters:

• plan_model (PlanModel)

• imitation_model (PlanModel)

• rounds (list[RoundStats])

class RoundStats(round, mean_score, n_candidates, n_kept)

Bases: object

Candidate-generation statistics for one outcome-decomposition round.

Parameters:

• round (int)

• mean_score (float)

• n_candidates (int)

• n_kept (int)

evaluate_greedy_heuristic(*, seeds, n_cells, n_targets, budget)

Return the mean score of the built-in greedy policy across held-out seeds.

Parameters:

• n_cells (int)

• n_targets (int)

• budget (int)

Return type:

float

evaluate_plan_model(model, *, seeds, n_cells, n_targets, budget, rng_seed=0)

Mean world score of model’s sampled plan, executed once per held-out seed.

Parameters:

• model (PlanModel)

• n_cells (int)

• n_targets (int)

• budget (int)

• rng_seed (int)

Return type:

float

execute_plan(plan_types, *, n_cells, n_targets, budget, seed)

Execute a plan (a sequence of action types, e.g. ["survey", "survey", "drill", ...]) in a fresh seeded world: at each step, “survey” targets the undrilled cell with the noisiest current read (most to gain), “drill” targets the undrilled cell with the highest current prospectivity read – the plan model decides the order and mix of action types; this fixed rule decides which cell, the same division of labor the plan/tool-name abstraction uses everywhere else in this plan. Returns the world’s final score.

Parameters:

• plan_types (list[str])

• n_cells (int)

• n_targets (int)

• budget (int)

• seed (int)

Return type:

int

imitation_traces(policy, *, n_worlds, n_cells, n_targets, budget, seed_offset=0)

Run policy over n_worlds seeded episodes and return each episode’s ACCEPTED action-type sequence used to fit the round-0 imitation model.

Parameters:

• n_worlds (int)

• n_cells (int)

• n_targets (int)

• budget (int)

• seed_offset (int)

train_outcome_decomposer(*, seed_worlds, n_cells, n_targets, budget, k_candidates=30, success_quantile=0.6, rounds=3, seed=0)

Train a plan model by sampling, executing, keeping high-outcome plans, and refitting.

Parameters:

• seed_worlds (int)

• n_cells (int)

• n_targets (int)

• budget (int)

• k_candidates (int)

• success_quantile (float)

• rounds (int)

• seed (int)

Return type:

OutcomeTrainedDecomposer

class StructuredSolution(fields_cat, fields_num, teacher, n_requests=0, n_escalated=0, harvested_inputs=<factory>, harvested_outputs=<factory>)

Bases: object

Per-field calibrated students in front of the dict-valued routine they replace.

Parameters:

• fields_cat (dict[str, Solution])

• fields_num (dict[str, RegressionSolution])

• teacher (Callable[[...], Any])

• n_requests (int)

• n_escalated (int)

• harvested_inputs (list)

• harvested_outputs (list)

property schema: dict[str, str]

categorical or numeric.

Type:

Return each output field’s inferred kind

try_local(x)

The fully-decided output dict, or None when ANY field is unsure (= must escalate).

Parameters:

x (Any)

Return type:

dict[str, Any] | None

decide(x)

Return the local structured-output decision, or None when the example should escalate.

Parameters:

x (Any)

Return type:

dict[str, Any] | None

report()

Return per-field calibration details and aggregate serving/harvest counts.

Return type:

dict[str, Any]

save(path)

Persist every field’s sub-artifact under one directory; load() restores the whole schema.

Parameters:

path (str)

Return type:

str

classmethod load(path, teacher, *, device='cpu')

Reconstitute a serving StructuredSolution (fields serve locally; escalation runs teacher).

Parameters:

• path (str)

• teacher (Callable[[...], Any])

• device (str)

Return type:

StructuredSolution

improve()

Push the harvested dicts down into every field’s buffer; each sub improves anti-regressively.

Return type:

bool

class DecompositionProposer(plan_model, log=<factory>)

Bases: object

An outcome-trained proposer over decompositions: plan_model scores/samples which intermediates (in what order) to route a task’s output through, and shifts toward higher-outcome decompositions as they get logged – train_outcome_decomposer()’s refit-on-successes loop, applied to decomposition proposals instead of tool-call plans.

Parameters:

• plan_model (PlanModel)

• log (list[tuple[list[str], float]])

class DependencyForest(chosen, edge_gains, mdl_gain, edges=<factory>)

Bases: object

A discovered decomposition of output: the ordered list of parent fields it was routed through (chosen, e.g. ["m1", "m2"]; empty means monolithic – no candidate intermediate or input cleared min_gain), each step’s own gain, and the total mdl_gain – the description-length gain (nats) of this decomposition over solving output directly from the raw inputs. Positive mdl_gain means the decomposition COMPRESSES; by construction it is the sum of the chosen edges’ own dependency_gain()/ regression_gain() scores.

Parameters:

• chosen (list[str])

• edge_gains (list[float])

• mdl_gain (float)

• edges (list[LinearGaussianEdge | None])

predict(inputs, candidate_intermediates)

Sum of each chosen edge’s prediction from its own parent field – the decomposed model’s point estimate, used to compare predictive accuracy against the monolithic baseline.

Parameters:

• inputs (Mapping[str, float])

• candidate_intermediates (Mapping[str, Callable[[Mapping[str, float]], float]])

Return type:

float

class TaskExample(inputs, output)

Bases: object

One observed instance of a task: named inputs and the realized output. The joint this module reasons over is (inputs, proposed_intermediates, output) – proposed_intermediates are not stored here, they are RECOMPUTED per candidate by discover_decomposition() (a candidate intermediate is a function of inputs, not a fixed observed field).

Parameters:

• inputs (Mapping[str, float])

• output (float)

discover_decomposition(task_examples, candidate_intermediates, *, max_parents=4, min_gain=0.0, max_its=30, seed=0)

Discover which candidate intermediates output should be routed through, by greedy forward selection scored with regression_gain() / dependency_gain() – the SAME model-based description-length test learn_structure() uses for record fields, applied here per step against the current RESIDUAL so multiple intermediates (output = f(g(a), h(b))) can each earn their own edge, not just the single best one (a plain DependencyTreeDistribution forest allows one parent per field; a task’s output routinely needs several).

Every raw input is itself a candidate parent, so a task with NO real decomposable structure correctly comes back with chosen == [] (monolithic: the raw inputs already explain output as well as anything) rather than inventing intermediates that do not pay for themselves.

Parameters:

• task_examples (Sequence[TaskExample])

• candidate_intermediates (Mapping[str, Callable[[Mapping[str, float]], float]])

• max_parents (int)

• min_gain (float)

• max_its (int)

• seed (int)

Return type:

DependencyForest

fit_decomposition(task_examples, decomposition, candidate_intermediates, *, max_its=30, seed=0)

Fit a SPECIFIC, given decomposition (in order) rather than discovering one – every named field is forced in, in order, scored and residualized the same way discover_decomposition()’s forward selection does. Lets a caller both score (DependencyForest.mdl_gain) and predict with (DependencyForest.predict()) a decomposition it did not necessarily search for – e.g. a deliberately-worse candidate, for the MDL-gain/outcome correlation check.

Parameters:

• task_examples (Sequence[TaskExample])

• decomposition (Sequence[str])

• candidate_intermediates (Mapping[str, Callable[[Mapping[str, float]], float]])

• max_its (int)

• seed (int)

Return type:

DependencyForest

init_decomposition_proposer(seed_decompositions)

Fit the round-0 (imitation) proposer on a seed corpus of decompositions – e.g. every chosen a few discover_decomposition() calls returned on early task instances.

Parameters:

seed_decompositions (Sequence[Sequence[str]])

Return type:

DecompositionProposer

log_decomposition_recipe(design, mdl_gain, *, family)

Record one decomposition attempt’s MDL gain into the existing design ledger under family ("decomposed" / MONOLITHIC) – a thin wrapper over record_accepted_recipe() so rank_design_families() and best_family() answer “has decomposing this kind of task actually paid off” from real history, the same what-worked prior every other structural family search uses.

Parameters:

• design (DesignModel)

• mdl_gain (float)

• family (str)

Return type:

None

mdl_score(task_examples, decomposition, candidate_intermediates, *, max_its=30, seed=0)

The MDL gain (nats) of routing output through a SPECIFIC, given decomposition – a thin accessor over fit_decomposition() for callers that only want the score (e.g. ranking several candidate decompositions for the MDL-gain/outcome correlation check).

Parameters:

• task_examples (Sequence[TaskExample])

• decomposition (Sequence[str])

• candidate_intermediates (Mapping[str, Callable[[Mapping[str, float]], float]])

• max_its (int)

• seed (int)

Return type:

float

monolithic_predict(train, test)

OLS fit of output on the raw inputs (every field jointly, closed form) – the “solve as one black box” baseline discover_decomposition() is compared against. Matched compute against the decomposed model: both are single closed-form linear solves over the same n examples.

Parameters:

• train (Sequence[TaskExample])

• test (Sequence[TaskExample])

Return type:

list[float]

record_decomposition_outcome(proposer, decomposition, outcome, *, success_quantile=0.6, min_log=4)

Log one (decomposition, outcome) pair from a REAL task instance, and once at least min_log outcomes are on file, refit plan_model on the decompositions scoring at or above this round’s own success_quantile – literally train_outcome_decomposer()’s keep-the-successes-and-refit step, so future sample() calls favor what actually worked, not just what the seed corpus imitated.

Parameters:

• proposer (DecompositionProposer)

• decomposition (Sequence[str])

• outcome (float)

• success_quantile (float)

• min_log (int)

Return type:

DecompositionProposer

class PilotLadderResult(outcomes, halted_at, journal)

Bases: object

The whole ladder’s outcome: every attempted rung, where (if anywhere) it halted, and the journal.

Parameters:

• outcomes (list[RungOutcome])

• halted_at (str | None)

• journal (EpistemicJournal)

class Rung(name, real_target, decision_pieces, vocab=64, d_model=16, n_layer=2, n_head=2, block=8, n_workers=1, steps=40, switch_step=None, batch_size=8, lr=0.01, seed=0, max_final_loss=3.0, max_forgetting_gap=1.5, require_continuity=True, exercise_mup_transfer=False, mup_base_width=None, exercise_moe_decision=False, moe_experts=4, moe_max_relative_diff=1.0, exercise_fault_tolerance=False, exercise_eval_suite=False, exercise_context_parallel=False, exercise_scaling_law_fit=False)

Bases: object

One pilot-ladder rung: a tiny simulated stand-in for a REAL roadmap rung’s size/context/GPU count.

real_target documents the real rung this stands in for (e.g. "1B params / 8k context / 8 GPUs") purely for the record – nothing here can measure that scale, so the actual training below runs at vocab/d_model/n_layer/n_head/block sizes chosen to finish in seconds on a laptop. n_workers is a documented stand-in for the real rung’s GPU count; this module does not spawn n_workers real processes (see the F1 note in the module docstring) but records it as part of the rung’s identity for the decision journal.

Parameters:

• name (str)

• real_target (str)

• decision_pieces (tuple[str, ...])

• vocab (int)

• d_model (int)

• n_layer (int)

• n_head (int)

• block (int)

• n_workers (int)

• steps (int)

• switch_step (int | None)

• batch_size (int)

• lr (float)

• seed (int)

• max_final_loss (float)

• max_forgetting_gap (float | None)

• require_continuity (bool)

• exercise_mup_transfer (bool)

• mup_base_width (int | None)

• exercise_moe_decision (bool)

• moe_experts (int)

• moe_max_relative_diff (float)

• exercise_fault_tolerance (bool)

• exercise_eval_suite (bool)

• exercise_context_parallel (bool)

• exercise_scaling_law_fit (bool)

class RungArtifacts(rung, health_report, loss_curve, forgetting_curve, forgetting_gap, final_loss, mfu_mean, skipped_pieces=<factory>, exercised_receipts=<factory>)

Bases: object

The roadmap’s per-rung artifacts: MFU, loss curve, forgetting curve, plus this pilot’s bookkeeping.

Parameters:

• rung (str)

• health_report (dict[str, Any])

• loss_curve (list[float])

• forgetting_curve (list[float])

• forgetting_gap (float | None)

• final_loss (float)

• mfu_mean (float | None)

• skipped_pieces (dict[str, str])

• exercised_receipts (dict[str, Any])

class RungOutcome(artifacts, passed, reason, decision_record)

Bases: object

One rung’s full outcome: its artifacts, the GO/NO-GO verdict, why, and its journal entry.

Parameters:

• artifacts (RungArtifacts)

• passed (bool)

• reason (str)

• decision_record (DecisionRecord)

run_pilot_ladder(rungs, *, peak_flops_per_sec=1.0e12)

Run each of rungs in order, gating progression on a real GO/NO-GO check of its own artifacts.

For every rung: train the rung’s tiny simulated model, collect MFU / loss-curve / forgetting-curve artifacts (reusing mixle.utils.parallel.training_health’s exact machinery), exercise whichever of F9/H2 the rung opted into for real, append one Bayesian decision-journal entry (mixle.epistemic.journal.EpistemicJournal) recording the belief update and the actual GO/NO-GO action taken, and – this is the gate – stop the ladder the first time a rung’s measured receipts fail its own stated criteria. peak_flops_per_sec is an arbitrary stand-in “hardware peak” (no real hardware backs any MFU number this produces); it only needs to be a fixed positive constant for MFU to be comparable across this ladder’s own rungs, which is all the GO/NO-GO gate uses it for.

Parameters:

• rungs (Sequence[Rung])

• peak_flops_per_sec (float)

Return type:

PilotLadderResult

class GenerativePlanner(lm, codec, tools, teacher, plan_agreement, max_new=160, constrained=True, conf_floor=None, lm_config=<factory>, n_requests=0, n_escalated=0, harvested=<factory>)

Bases: object

A plan-writing LM behind a parse-and-validate gate: only verified plans leave; the rest escalate.

Parameters:

• lm (Any)

• codec (_CharCodec)

• tools (dict[str, ToolSpec])

• teacher (Callable[[str], list[dict]])

• plan_agreement (float)

• max_new (int)

• constrained (bool)

• conf_floor (float | None)

• lm_config (dict)

• n_requests (int)

• n_escalated (int)

• harvested (list[tuple[str, list[dict]]])

try_plan(request)

Generate, parse, validate (grammar + specs + copy-fidelity); None = must escalate.

With constrained=True (default) the decode itself runs inside the plan grammar (mixle.task.constrained.constrained_plan_decode()): malformed text and copy-drifted values are unrepresentable, and the parse/validate below is a pure backstop.

Parameters:

request (str)

Return type:

list[dict] | None

report()

Return plan agreement, escalation, and harvested-trace metrics.

Return type:

dict[str, Any]

save(path)

Persist the plan-writing LM (weights + builder config), codec, specs, and gates; load() restores.

Parameters:

path (str)

Return type:

str

classmethod load(path, teacher, *, device='cpu')

Reconstitute a serving GenerativePlanner from save() output plus the teacher fallback.

Parameters:

• path (str)

• teacher (Callable[[str], list[dict]])

• device (str)

Return type:

GenerativePlanner

class Planner(selector, extractors, tools, teacher, plan_agreement, max_steps=8, n_requests=0, n_escalated=0, harvested=<factory>)

Bases: object

A distilled decomposer: emit verified steps until STOP, or escalate the whole problem.

Parameters:

• selector (Any)

• extractors (dict[str, Any])

• tools (dict[str, ToolSpec])

• teacher (Callable[[str], list[dict]])

• plan_agreement (float)

• max_steps (int)

• n_requests (int)

• n_escalated (int)

• harvested (list[tuple[str, list[dict]]])

try_plan(request, *, execute=None)

The local decomposition alone: a complete verified plan, or None (= must escalate).

This method does not call the teacher.

Parameters:

• request (str)

• execute (dict[str, Callable[[...], Any]] | None)

Return type:

dict[str, Any] | None

report()

Return plan agreement, escalation, and harvested-trace metrics.

Return type:

dict[str, Any]

save(path)

Persist selector + per-tool extractors + specs as one artifact directory; load() restores.

Parameters:

path (str)

Return type:

str

classmethod load(path, teacher, *, device='cpu')

Reconstitute a serving Planner from save() output plus the teacher fallback.

Parameters:

• path (str)

• teacher (Callable[[str], list[dict]])

• device (str)

Return type:

Planner

class PlanModel(dist, training_log_probs)

Bases: object

A fitted Markov chain over tool-name sequences, plus the training traces’ own log-prob spread.

Parameters:

• dist (Any)

• training_log_probs (ndarray)

log_prob(plan)

Exact log-probability of plan (a tool-name list, or the [{"tool":...}, ...] shape).

Parameters:

plan (Sequence[Any])

Return type:

float

sample(rng=None)

Draw one plausible tool-name sequence from the fitted chain.

The underlying sampler draws a length from len_dist first, then walks the chain; once the walk reaches an absorbing state (no fitted outgoing transition – typically the tool that always ends a workflow), the remaining, unreachable slots are returned as None. Truncate there rather than exposing that padding: only known, actually-reached tool names are emitted.

Parameters:

rng (RandomState | None)

Return type:

list[str]

is_typical(plan, *, quantile=0.05)

False when plan scores below the training traces’ own quantile log-prob – the escalation signal: a plan that does not look like what this agent usually does.

Parameters:

• plan (Sequence[Any])

• quantile (float)

Return type:

bool

fit_plan_model(traces, *, smoothing=0.5, init_p=1.0)

Fit a PlanModel on harvested traces’ tool-name sequences.

smoothing is the Markov chain’s Dirichlet pseudo-count (higher = smoother transition estimates, matters most with few traces). Fits via mixle.inference.optimize() on the existing MarkovChainEstimator – the same declare-an-estimator/call-optimize path every other mixle model uses, not hand-rolled counting.

init_p defaults to 1.0 (use every trace for the init pass), not optimize’s own init_p=0.1 default: that Bernoulli-subsamples observations for a low-cost init estimate, sized for large corpora, but a trace corpus here is typically tens to a few hundred sequences – with that few, a 10% subsample has a real chance of drawing ZERO sequences, which crashes MarkovChainEstimator.estimate1 (all_keys ends up empty, dividing by zero). Using the full corpus for this small an init pass is low-overhead and more reliable; override down only for corpora large enough that subsampling actually matters.

Parameters:

• traces (AgentTraces | Sequence[AgentTrace])

• smoothing (float)

• init_p (float)

Return type:

PlanModel

class Solution(cascade, teacher, kind, train_inputs, train_labels, cal_inputs, cal_labels, holdout_agreement, escalation_rate, promoted, target_agreement, distill_kw=<factory>, ood=None, seed=0, synthesized=0, gate_inputs=<factory>, edge=None)

Bases: object

A deployed task: a calibrated student in front of the teacher, plus the loop to improve it.

Call it like the original function. promoted says whether the student passed verification – when False the callable simply runs the teacher instead of deploying an unverified student.

Parameters:

• cascade (Cascade)

• teacher (Callable[[...], Any])

• kind (str)

• train_inputs (list)

• train_labels (list)

• cal_inputs (list)

• cal_labels (list)

• holdout_agreement (float)

• escalation_rate (float)

• promoted (bool)

• target_agreement (float | None)

• distill_kw (dict)

• ood (float | None)

• seed (int)

• synthesized (int)

• gate_inputs (list)

• edge (Any)

report()

What you would want on a dashboard: verification, live escalation, realized cost.

Return type:

dict

improve()

Re-distill with the harvested (escalated) labels; promote only if it verifies at least as well.

Returns True when a better student was promoted. The calibration slice is never trained on, so the conformal guarantee and the agreement comparison remain valid across rounds.

Return type:

bool

health(recent_inputs=None, *, p_threshold=0.01)

Check whether live escalation behavior has drifted from calibration.

The conformal answer-or-escalate rule is calibrated under an exchangeability assumption. When the input distribution shifts, the live escalation rate may move away from the verified baseline. This method compares the live rate with the baseline using an exact binomial test and, when recent_inputs and an OOD gate are available, compares the gate hit rate with its design quantile.

Returns a dictionary with drifted, live and baseline rates, and p-values where enough observations are available. A drift alarm means traffic has changed and retraining or review may be needed; abstained inputs still route to the teacher.

Parameters:

• recent_inputs (Any)

• p_threshold (float)

Return type:

dict[str, Any]

save(path)

Persist the calibrated student as a load-anywhere artifact, with its verification record.

Every deployed artifact carries how it was verified — held-out agreement with the teacher, the escalation rate, the conformal alpha, and how much of its training data was synthetic — so “is this model trustworthy” is answerable from the artifact alone.

Parameters:

path (str)

Return type:

str

deploy(name, root='./mixle_data/registry')

Save into the serving layout — {root}/tasks/{name} — the directory the mixle-mlops /v1/tasks routes serve from. Returns the artifact path.

Parameters:

• name (str)

• root (str)

Return type:

str

classmethod load(path, teacher, *, cost=None, device='cpu')

Reconstitute a serving Solution from a saved artifact — the deploy path for a fresh process.

The loaded Solution answers locally / escalates to teacher and harvests labels exactly like the original. It carries no training or calibration data, so improve() raises — collect the harvested pairs and re-solve (real + harvested inputs) to train the next round.

Parameters:

• path (str)

• teacher (Callable[[...], Any])

• cost (Any)

• device (str)

Return type:

Solution

class StructuredClassifierIO(field_keys, label_index, labels)

Bases: object

record -> label through a structured probabilistic model instead of a neural net.

The model is a fitted joint over (field_1, ..., field_m, label) – a DependencyTreeDistribution (or mixture) discovered by mixle.inference.structure.learn_structure(). Classification is the generative rule argmax_label P(features, label): score each candidate label and pick the best. Because softmax_label log P(features, label) = P(label | features) exactly (the feature evidence is a shared constant across labels), proba_batch() returns the true posterior – not a softmax over arbitrary logits – so conformal calibration (mixle.task.calibrate) and the density gate operate on a real probability.

The student is interpretable (model.edges() shows the discovered dependencies), kilobytes on disk, and round-trips through the json artifact path. It assumes a fixed schema: every record exposes the same fields (field_keys for dicts, positional for tuples) – the variable set a Bayesian network is defined over.

Parameters:

• field_keys (list[str] | None)

• label_index (int)

• labels (list[str])

logits_batch(model, raw_inputs)

Per-label log-joint log P(features, label) as an (m, K) score matrix (the classifier logits).

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

proba_batch(model, raw_inputs)

The exact posterior P(label | features) – softmax of the per-label log-joints (shared evidence cancels).

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

ndarray

predict_batch(model, raw_inputs)

Predict labels for raw inputs by maximizing the per-label joint score.

Parameters:

• model (Any)

• raw_inputs (list[Any])

Return type:

list[str]

predict(model, raw_input)

Predict the label for one raw input.

Parameters:

• model (Any)

• raw_input (Any)

Return type:

str

to_spec()

Return the serializable structured-classifier adapter specification.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Rebuild a structured-classifier adapter from its artifact io specification.

Parameters:

spec (dict[str, Any])

Return type:

StructuredClassifierIO

class TaskManifest(payload, builder=None, config=<factory>, task='', io=<factory>, meta=<factory>, schema_version='1', created_at='')

Bases: object

The self-describing header of a task artifact: enough to rebuild and call the model, plus provenance.

Parameters:

• payload (str)

• builder (str | None)

• config (dict[str, Any])

• task (str)

• io (dict[str, Any])

• meta (dict[str, Any])

• schema_version (str)

• created_at (str)

to_dict()

Return the strict-JSON manifest representation written to manifest.json.

Return type:

dict[str, Any]

classmethod from_dict(d)

Parse a manifest dictionary into a TaskManifest.

Parameters:

d (dict[str, Any])

Return type:

TaskManifest

class TaskModel(model, adapter, *, builder=None, config=None, payload='torch', task='', meta=None)

Bases: object

A fitted small model plus its I/O adapter, callable as task(raw) -> result and saveable to a directory.

Parameters:

• model (Any)

• adapter (Any)

• builder (str | None)

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

• payload (str)

• task (str)

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

batch(raw_inputs)

Run the wrapped model on a batch of raw inputs through its adapter.

Parameters:

raw_inputs (list[Any])

Return type:

list[Any]

save(path)

Persist as a task artifact: the model payload plus the adapter’s io spec and metadata.

Parameters:

path (str)

Return type:

str

classmethod load(path, *, device='cpu')

Rebuild a TaskModel (model + adapter) from a saved artifact directory.

Parameters:

• path (str)

• device (str)

Return type:

TaskModel

class TraceStep(tool, args=<factory>, seed=None, result=None)

Bases: object

One recorded step: the tool name, the args it ran with, the seed (if any), and its result.

Parameters:

• tool (str)

• args (dict[str, Any])

• seed (int | None)

• result (Any)

to_json()

Serialize this trace step to JSON-compatible data.

Return type:

dict[str, Any]

classmethod from_json(d)

Reconstruct a trace step from JSON-compatible data.

Parameters:

d (dict[str, Any])

Return type:

TraceStep

class ToolCaller(selector, extractors, tools, teacher, selection_agreement, n_requests=0, n_escalated=0, harvested=<factory>)

Bases: object

Distilled function caller with calibrated selection and argument extraction.

Parameters:

• selector (Solution)

• extractors (dict[str, Any])

• tools (dict[str, ToolSpec])

• teacher (Callable[[str], dict])

• selection_agreement (float)

• n_requests (int)

• n_escalated (int)

• harvested (list[tuple[str, dict]])

try_local(request)

Return the local decision, or None when the request must escalate.

This method does not call the teacher.

Parameters:

request (str)

Return type:

dict[str, Any] | None

report()

Return serving counts, escalation rate, and selector agreement diagnostics.

Return type:

dict[str, Any]

save(path)

Persist selector, per-tool extractors, and tool specs.

Parameters:

path (str)

Return type:

str

classmethod load(path, teacher, *, device='cpu')

Reconstitute a serving ToolCaller from save() output plus the teacher fallback.

Parameters:

• path (str)

• teacher (Callable[[str], dict])

• device (str)

Return type:

ToolCaller

class ToolSpec(name, args, required=None)

Bases: object

One callable tool: its name and the argument fields to extract from the request text.

Parameters:

• name (str)

• args (list[str])

• required (list[str] | None)

property required_args: list[str]

Return required argument names, defaulting to all declared arguments.

class TextClassifierIO(featurizer, labels)

Bases: _ClassifierIO

str -> label: hashed character n-gram features into a small classifier.

Parameters:

• featurizer (HashedNGram)

• labels (list[str])

class CalibratedTuneResult(model, recipe, agreement, score, cost, history=None)

Bases: object

The outcome of a routing-ready recipe search: the calibrated winner, its recipe and scores, and history.

Parameters:

• model (CalibratedTaskModel)

• recipe (dict[str, Any])

• agreement (float)

• score (float)

• cost (float)

• history (Any)

class TuneResult(model, recipe, agreement, score, cost, history=None)

Bases: object

The outcome of a recipe search: the winning model, its recipe and scores, and the full BO history.

Parameters:

• model (TaskModel)

• recipe (dict[str, Any])

• agreement (float)

• score (float)

• cost (float)

• history (Any)

class WordEmbeddingFeaturizer(vectors, dim, seed=0)

Bases: object

Average per-word embedding vectors from a fixed lookup table – a dependency-free “embedding head” featurizer.

Unlike HashedNGram (which treats distinct surface tokens as unrelated hash buckets), two words given nearby vectors in vectors produce nearby features regardless of their spelling – the property a synonym-generalizing rule needs. A word missing from vectors falls back to a deterministic hashed sub-vector, so out-of-vocabulary text still produces a valid feature; it just earns no semantic generalization it was never given a vector for.

Parameters:

• vectors (dict[str, Sequence[float]] | None)

• dim (int)

• seed (int)

transform(texts)

Map texts to normalized embedding features with hashed fallback rows.

Parameters:

texts (list[str])

Return type:

ndarray

to_spec()

Serialize embedding vectors and fallback hashing settings.

Return type:

dict[str, Any]

classmethod from_spec(spec)

Reconstruct the embedding featurizer from an artifact spec.

Parameters:

spec (dict[str, Any])

Return type:

WordEmbeddingFeaturizer

acquisition_scores(student, texts, method='margin')

Informativeness of each unlabeled text under the student (higher = more worth labeling).

Parameters:

• student (TaskModel)

• texts (Sequence[str])

• method (str)

Return type:

ndarray

active_distill(teacher, pool, *, budget, seed_size=20, rounds=5, acquisition='margin', labels=None, recipe=None, val_texts=None, seed=0)

Distill from pool under a labeling budget, querying the teacher only for the most informative items.

Labels a seed_size random seed, then over rounds adds the top-scoring unlabeled examples (by acquisition) until budget labels are spent, refitting the student each round. If val_texts is given, the teacher labels it once and each round’s agreement on it is logged.

Parameters:

• teacher (Callable[[...], Any])

• pool (Sequence[str])

• budget (int)

• seed_size (int)

• rounds (int)

• acquisition (str)

• labels (Sequence[str] | None)

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

• val_texts (Sequence[str] | None)

• seed (int)

Return type:

ActiveResult

adapter_from_spec(spec)

Rebuild an adapter from its io spec (the kind field selects the factory).

Parameters:

spec (dict[str, Any])

Return type:

Any

agreement(student, teacher_labels, texts)

Fraction of texts where the student’s label matches the teacher’s – distillation fidelity.

Parameters:

• student (TaskModel)

• teacher_labels (Sequence[Any])

• texts (Sequence[str])

Return type:

float

break_even_volume(cost, n_label, *, p_escalate=0.0)

Requests after which a distilled route undercuts frontier-only (inf if it never does).

Setup is amortized against the per-request saving c_frontier - per_request(route). With p_escalate=0 this is the local-only break-even; pass the model’s escalation rate for the cascade break-even.

Parameters:

• cost (CostModel)

• n_label (int)

• p_escalate (float)

Return type:

float

capture_profile(student, teacher, texts, suite)

Run student and teacher through suite and return a profile.

Returns a plain, json.dumps-safe dict:

• "clean_agreement" – student/teacher label agreement on the uncorrupted texts;

• "corruptions" – per corruption name, student/teacher agreement on the corrupted texts (in the suite’s insertion order, mild-to-severe by convention);

• "invariances" – per invariance name, {"student_violation_rate", "teacher_violation_rate"}: how often each side’s prediction changes under a rewrite that should not change it. A student must not be penalized for an invariance the teacher itself violates – both rates are reported, never one diff;

• "probes" – {"student": [...], "teacher": [...]} raw predictions on the fixed probe inputs, or omitted if the suite has no probes;

• "abstention" – present only if student or teacher exposes a decision API (decide / batch_decide): each side’s escalation rate on texts (None for a side with no decision API).

There is deliberately no single aggregate score field.

Parameters:

• student (Any)

• teacher (Any)

• texts (Sequence[str])

• suite (CapabilitySuite)

Return type:

dict[str, Any]

cascade_cost_per_request(cost, p_escalate)

Expected per-request cost of the cascade: always run local, escalate the p_escalate fraction.

Parameters:

• cost (CostModel)

• p_escalate (float)

Return type:

float

case_jitter_invariance(text)

A meaning-preserving rewrite: swap the case of every letter.

Parameters:

text (str)

Return type:

str

capacity_ladder(teacher_or_labels, texts, *, target, rungs=DEFAULT_RUNGS, val_texts=None, val_labels=None, labels=None, word_vectors=None, calibration_frac=0.3, n=3, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, device='cpu')

Fit a student at each rung of rungs (increasing representation family) and measure held-out agreement.

teacher_or_labels is either a callable teacher (labels texts and, if given separately, val_texts) or a sequence of labels already aligned with texts – mirroring the distill/distill_from_labels duality. When val_texts/val_labels are not given, a calibration_frac held-out slice of (texts, teacher labels) is used (same split machinery as routing calibration), so a paraphrase/synonym generalization gap between train and held-out is measurable even with a single corpus.

word_vectors (word -> dense vector) is the only thing that makes the "embedding_head" rung semantically richer than "hashed_ngram" – without it, that rung still builds (never skipped, it is one of the two minimum rungs) but falls back to hashed features per out-of-vocabulary word, so it will not beat "hashed_ngram". "strong_encoder"/"small_lm" are recognized rung names with no estimator wired in this environment: they are skipped with a note, never raised as an error.

Returns a LadderResult with every rung’s measured score and either the smallest rung meeting target or winner=None with every built rung’s ceiling attached – “target unmet” is a valid result, never an exception.

Parameters:

• teacher_or_labels (Callable[[...], Any] | Sequence[Any])

• texts (Sequence[str])

• target (float)

• rungs (Sequence[str])

• val_texts (Sequence[str] | None)

• val_labels (Sequence[Any] | None)

• labels (Sequence[str] | None)

• word_vectors (dict[str, Sequence[float]] | None)

• calibration_frac (float)

• n (int)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• device (str)

Return type:

LadderResult

climb_to(fault, *, rungs=KNOWN_RUNGS)

Given a refinement-loop fault localized to a saturated leaf’s current rung, return the next rung up.

fault is either a bare rung name or an object naming its current rung via a rung or dominant attribute (the shape diagnose()’s FaultReport will eventually carry) – this lets a caller climb straight to the next rung for the one saturated leaf, without re-running the whole ladder. Raises ValueError if the current rung is already the top of rungs.

Parameters:

• fault (Any)

• rungs (Sequence[str])

Return type:

str

compose(a, b, *, name_a='stage_a', name_b='stage_b')

Chain a: x -> y and b: y -> z into one ledger-carrying x -> z callable.

Parameters:

• a (Callable[[Any], Any])

• b (Callable[[Any], Any])

• name_a (str)

• name_b (str)

Return type:

ComposedModel

design_model(data, llm, *, fallback=True, validate_rows=200)

Ask llm to design a model for data; build, fit-validate, and fall back to the heuristic on failure.

The LLM sees a compact data_profile() and returns a JSON spec; the spec is built into a real estimator and fit on a sample to prove it works. Any failure (no LLM, invalid JSON, off-allowlist family, fit error) yields the heuristic mixle.task.recommend.recommend_model() estimator when fallback is set.

Parameters:

• data (Sequence[Any])

• llm (Any)

• fallback (bool)

• validate_rows (int)

Return type:

DesignedModel

fit_disagreement_gate(student, texts, teacher_labels, *, dim=256, hidden=(32,), epochs=150, lr=1e-2, seed=0, threshold=0.5)

Fit a DisagreementGate from a labeled sample: run student on texts, label each example "disagree" where it differs from teacher_labels and "agree" otherwise, and distill a compact binary classifier of that target over the same hashed n-gram feature family the student itself uses (a different, wider/deeper recipe is fine – what matters is the classifier learns a decision surface over the input text, not that it matches the student’s exact recipe).

Parameters:

• student (TaskModel)

• texts (Sequence[str])

• teacher_labels (Sequence[Any])

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• threshold (float)

Return type:

DisagreementGate

measure_disagreement_mass(student, texts, teacher_labels)

Fraction of texts where the student’s label differs from the teacher’s.

Parameters:

• student (TaskModel)

• texts (Sequence[str])

• teacher_labels (Sequence[Any])

Return type:

float

distill(teacher, texts, *, labels=None, n=3, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu', n_jobs=1)

Label texts with teacher, fit a local student, and return a callable TaskModel.

n/dim size the hashed n-gram featurizer; hidden the student MLP. labels fixes the label set (else inferred from the teacher’s outputs). The student’s train-set agreement with the teacher is recorded in meta. n_jobs > 1 fans teacher labeling across that many threads (order-preserving; the win is parallel in-flight requests against a network-bound teacher) – every distill_* teacher entry point takes the same knob.

Parameters:

• teacher (Callable[[...], Any])

• texts (Sequence[str])

• labels (Sequence[str] | None)

• n (int)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

• n_jobs (int)

Return type:

TaskModel

distill_designer(design, *, quality_quantile=0.5, seed=0)

Distill the design ledger into a compact torch-free student that judges designs: point -> good/weak.

The DesignModel’s rows are records (the design coordinates); a row is labeled good when it was feasible on the device and its quality reached the ledger’s quality_quantile. distill_structured_from_labels() – the same machinery the design model tunes – compresses that knowledge into a kilobyte Bayesian-network student usable as a zero-cost pre-filter for future searches. Teacher trains student; a model designs the students; the designer is distilled into a student: each level of the tower is a real artifact.

Parameters:

• design (DesignModel)

• quality_quantile (float)

• seed (int)

Return type:

TaskModel

distill_extractor(teacher, texts, fields, *, max_vocab=5000, d_model=64, hidden=64, epochs=60, lr=5e-3, max_len=128, seed=0, device='cpu', task='')

Distill a teacher’s extractions into a local sequence tagger; return model(text) -> {field: value}.

Parameters:

• teacher (Callable[[...], Any])

• texts (Sequence[str])

• fields (Sequence[str])

• max_vocab (int)

• d_model (int)

• hidden (int)

• epochs (int)

• lr (float)

• max_len (int)

• seed (int)

• device (str)

• task (str)

Return type:

TaskModel

distill_planner(teacher, requests, tools, *, holdout=0.2, seed=0, max_steps=8, selector_kw=None, extractor_kw=None)

Distill the teacher’s multi-step plans into next-step students (see module docstring).

Plan-level verification is measured on held-out requests the students never trained on: a plan agrees when every step’s tool and required arguments match the teacher’s plan exactly, in order.

Parameters:

• teacher (Callable[[str], list[dict]])

• requests (Sequence[str])

• tools (Sequence[ToolSpec])

• holdout (float)

• seed (int)

• max_steps (int)

• selector_kw (dict | None)

• extractor_kw (dict | None)

Return type:

Planner

distill_tool_caller(teacher, requests, tools, *, seed=0, selector_kw=None, extractor_kw=None)

Distill the teacher’s function-calling into a local selector plus per-tool argument extractors.

Parameters:

• teacher (Callable[[str], dict]) – teacher(request) -> {"tool": name-or-None, "args": {field: value}} — the frontier LLM / agent / rule currently doing the calling. It labels everything; it remains the fallback.

• requests (Sequence[str]) – example request texts covering the tools.

• tools (Sequence[ToolSpec]) – the tool specs (names + argument fields; required defaults to all).

• extractor_kw (dict | None) – knobs forwarded to solve() and distill_extractor().

• seed (int)

• selector_kw (dict | None)

• extractor_kw

Return type:

ToolCaller

distill_for_edge(teacher, train_data, val_data, device, *, labels=None, train_labels=None, val_labels=None, space=None, design=None, designer=None, n_init=4, n_iter=6, screen_fidelity=0.3, promote=2, seed=0, task='')

Search structure x training-process for the best student that fits device.

The teacher labels train_data/val_data once (cached) – or pass train_labels/ val_labels when the labels already exist (a harvested dataset, an upstream solve split) and the teacher is then never called (it may be None). Candidates proposed by the DesignModel are trained at screen_fidelity (reduced cost), scored by held-out agreement, and measured (footprint()); the top promote feasible screens are re-trained at full fidelity and the best feasible one wins (ties -> smaller). Pass a previous search’s design (same space + device shape) to warm-start: the surrogate already knows which regions blow the budget. Pass designer (the compact judge from distill_designer()) to veto known-weak proposals before any training is spent. If nothing fits the device, the least-infeasible student is returned with feasible=False – inspect result.pareto for the real trade-off frontier.

Parameters:

• teacher (Callable[[...], Any] | None)

• train_data (Sequence[Any])

• val_data (Sequence[Any])

• device (DeviceSpec)

• labels (Sequence[str] | None)

• train_labels (Sequence[Any] | None)

• val_labels (Sequence[Any] | None)

• space (EdgeSpace | None)

• design (DesignModel | None)

• designer (Callable[[tuple], Any] | None)

• n_init (int)

• n_iter (int)

• screen_fidelity (float)

• promote (int)

• seed (int)

• task (str)

Return type:

EdgeDistillResult

footprint(student)

Measure a student’s deployment cost. Bytes are real (fp32 weights, int8 arrays for a quantized student, or serialized JSON for a structured one); ops are the closed-form per-inference count for the student’s kind.

Parameters:

student (TaskModel)

Return type:

EdgeFootprint

distill_for_routing(teacher, texts, *, labels=None, calibration_frac=0.2, alpha=0.1, n=3, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu', density_gate=False, density_gate_alpha=0.05, n_jobs=1)

Label texts with teacher, fit a student, and calibrate it for routing – all in one call.

A calibration_frac slice of the (teacher-)labeled data is held out from training and used to set a conformal threshold, so the returned CalibratedTaskModel is immediately decide()-able: confident, in-distribution inputs get the student’s label; everything else is ESCALATE. Pass it straight to Cascade (with teacher) or Router for tiered serving – no separate calibration split to manage by hand. Deterministic given seed; the calibration slice is disjoint from the student’s training data.

density_gate=True additionally escalates inputs a softmax cannot see are atypical: see distill_from_labels_for_routing().

Parameters:

• teacher (Callable[[...], Any])

• texts (Sequence[str])

• labels (Sequence[str] | None)

• calibration_frac (float)

• alpha (float)

• n (int)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

• density_gate (bool)

• density_gate_alpha (float)

• n_jobs (int)

Return type:

CalibratedTaskModel

distill_from_labels(texts, teacher_labels, *, labels=None, n=3, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu')

Fit a student from already-labeled (texts, teacher_labels) – the teacher-free training core of distill.

Active labeling (mixle.task.active) uses this to avoid re-querying the teacher: it controls exactly which examples were paid for and passes their labels straight in. labels fixes the label set so a student trained on a partial sample still spans every class.

Parameters:

• texts (Sequence[str])

• teacher_labels (Sequence[Any])

• labels (Sequence[str] | None)

• n (int)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

Return type:

TaskModel

distill_from_labels_for_routing(texts, teacher_labels, *, labels=None, calibration_frac=0.2, alpha=0.1, n=3, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu', density_gate=False, density_gate_alpha=0.05)

Teacher-free training core of distill_for_routing(): fit + calibrate from labels already in hand.

Splits (texts, teacher_labels) into a training slice and a held-out calibration_frac slice (fixed by seed), trains the student on the former via distill_from_labels(), then calibrates (calibrate()) on the latter. labels (if given, else inferred from all of teacher_labels before the split) is shared by both slices so a class that lands entirely on one side of the split doesn’t shrink the label set out from under the other.

density_gate=True fits a DensityGate on the training slice’s features (reusing the student’s own featurizer, so there is no second feature space to keep in sync), calibrates its OOD floor (density_gate_alpha) on the disjoint calibration slice, and wires it into the returned model – an input whose log p(x) falls below that floor escalates even if the conformal set is a confident singleton.

Parameters:

• texts (Sequence[str])

• teacher_labels (Sequence[Any])

• labels (Sequence[str] | None)

• calibration_frac (float)

• alpha (float)

• n (int)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

• density_gate (bool)

• density_gate_alpha (float)

Return type:

CalibratedTaskModel

distill_text_generative(teacher, texts, *, labels=None, pseudo_count=0.5, min_count=2, task='')

Distill a teacher into the generative text student (the teacher labels; see module docstring).

Parameters:

• teacher (Callable[[...], Any])

• texts (Sequence[str])

• labels (Sequence[str] | None)

• pseudo_count (float)

• min_count (int)

• task (str)

Return type:

TaskModel

distill_text_generative_from_labels(texts, teacher_labels, *, labels=None, pseudo_count=1.0, min_count=2, task='')

Fit the per-class token models from already-labeled texts (the teacher-free training core).

Parameters:

• texts (Sequence[str])

• teacher_labels (Sequence[Any])

• labels (Sequence[str] | None)

• pseudo_count (float)

• min_count (int)

• task (str)

Return type:

TaskModel

distill_records(teacher, records, *, labels=None, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu', n_jobs=1)

Distill a teacher into a record classifier (record -> label over tuples/dicts of mixed fields).

The structured-data sibling of distill(): classify a transaction, route a ticket, categorize a record. Uses the hashing-trick HashedRecord featurizer, so it needs no fitted encoder.

Parameters:

• teacher (Callable[[...], Any])

• records (Sequence[Any])

• labels (Sequence[str] | None)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

• n_jobs (int)

Return type:

TaskModel

distill_records_for_routing(teacher, records, *, labels=None, calibration_frac=0.2, alpha=0.1, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu', density_gate=False, density_gate_alpha=0.05, n_jobs=1)

The structured-record sibling of distill_for_routing(): fit + calibrate a record classifier in one call, returning a routing-ready CalibratedTaskModel.

Parameters:

• teacher (Callable[[...], Any])

• records (Sequence[Any])

• labels (Sequence[str] | None)

• calibration_frac (float)

• alpha (float)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

• density_gate (bool)

• density_gate_alpha (float)

• n_jobs (int)

Return type:

CalibratedTaskModel

distill_records_from_labels(records, teacher_labels, *, labels=None, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu')

Teacher-free record-classifier training core (mirrors distill_from_labels() for structured records).

Parameters:

• records (Sequence[Any])

• teacher_labels (Sequence[Any])

• labels (Sequence[str] | None)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

Return type:

TaskModel

distill_records_from_labels_for_routing(records, teacher_labels, *, labels=None, calibration_frac=0.2, alpha=0.1, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu', density_gate=False, density_gate_alpha=0.05)

Teacher-free training core of distill_records_for_routing() (mirrors distill_from_labels_for_routing() for structured records). density_gate=True fits the OOD gate on the training slice’s record features and calibrates it on the calibration slice, same as the text path.

Parameters:

• records (Sequence[Any])

• teacher_labels (Sequence[Any])

• labels (Sequence[str] | None)

• calibration_frac (float)

• alpha (float)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

• density_gate (bool)

• density_gate_alpha (float)

Return type:

CalibratedTaskModel

distill_structured(teacher, records, *, labels=None, n_components=1, min_gain=0.0, n_bins=4, max_its=30, seed=0, task='', n_jobs=1)

Distill a teacher into a structured probabilistic classifier – a learned Bayesian network, not an MLP.

The teacher labels records; mixle.inference.structure.learn_structure() then discovers the dependency forest over the joint (field_1, ..., field_m, label) and fits it. The student classifies by the generative rule argmax_label P(features, label) – and because softmax_label log P(features, label) = P(label | features) exactly, its confidence is a real posterior the cascade/calibration stack can trust. Unlike distill_records() (a hashed-feature MLP), this student is interpretable (model.edges() lists the discovered dependencies), a few kilobytes on disk, and needs no torch to run.

n_components > 1 fits a MixtureOfDependencyTrees – a latent-cluster student whose sub-structures differ by regime. Assumes a fixed record schema (see StructuredClassifierIO).

Parameters:

• teacher (Callable[[...], Any])

• records (Sequence[Any])

• labels (Sequence[str] | None)

• n_components (int)

• min_gain (float)

• n_bins (int)

• max_its (int)

• seed (int)

• task (str)

• n_jobs (int)

Return type:

TaskModel

distill_structured_from_labels(records, teacher_labels, *, labels=None, n_components=1, min_gain=0.0, n_bins=4, max_its=30, seed=0, task='')

Teacher-free core of distill_structured(): fit a structured classifier from labeled records.

Parameters:

• records (Sequence[Any])

• teacher_labels (Sequence[Any])

• labels (Sequence[str] | None)

• n_components (int)

• min_gain (float)

• n_bins (int)

• max_its (int)

• seed (int)

• task (str)

Return type:

TaskModel

distill_from_soft_labels(texts, teacher_probs, *, labels, temperature=2.0, hard_weight=0.0, n=3, dim=256, hidden=(64,), epochs=300, lr=1e-2, seed=0, task='', device='cpu')

Fit a student to per-example teacher probabilities over labels.

teacher_probs is (N, C) with rows summing to 1 (renormalized if not), column j the teacher’s probability of labels[j]. The student minimizes the temperature-softened T^2 * KL(teacher || student) (Hinton’s scaling, so the soft gradients keep magnitude as T grows), optionally mixed with hard_weight times the hard cross-entropy against the teacher’s argmax. temperature > 1 softens both sides so runner-up structure influences the fit. The result is deterministic given seed and returns a TaskModel whose proba_batch approximates the teacher’s full distribution.

Parameters:

• texts (Sequence[str])

• teacher_probs (Sequence[Any])

• labels (Sequence[str])

• temperature (float)

• hard_weight (float)

• n (int)

• dim (int)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• seed (int)

• task (str)

• device (str)

Return type:

TaskModel

distill_soft(teacher_proba, texts, *, labels, **kwargs)

Query a probability-returning teacher once over texts and soft-distill it (see distill_from_soft_labels()). teacher_proba(texts) -> (N, C) returns each example’s class distribution over labels (e.g. an LLM’s normalized top-k logprobs).

Parameters:

• teacher_proba (Callable[[list[str]], Any])

• texts (Sequence[str])

• labels (Sequence[str])

• kwargs (Any)

Return type:

TaskModel

soft_agreement(student, teacher_probs, texts)

Mean KL divergence KL(teacher || student) over texts – how faithfully the student matches the teacher’s full soft distribution (0 = identical), the soft-distillation analog of mixle.task.distill.agreement(). Lower is better; use it to compare soft vs hard students.

Parameters:

• student (TaskModel)

• teacher_probs (Sequence[Any])

• texts (Sequence[str])

Return type:

float

extraction_f1(model, gold, texts)

Micro-averaged field-level F1: a field counts as correct when the extracted value exactly matches gold.

Parameters:

• model (TaskModel)

• gold (Sequence[dict[str, str]])

• texts (Sequence[str])

Return type:

float

harvest_agent_traces(directory=None)

Read every stored mixle-agent conversation and return the trace corpus (skips unreadable files).

Parameters:

directory (str | Path | None)

Return type:

AgentTraces

keyboard_typo_corruption(rate, *, seed=0)

A corruption: replace each letter with a random lowercase letter independently with probability rate.

Deterministic given seed – the same corruption function always maps the same text to the same output.

Parameters:

• rate (float)

• seed (int)

Return type:

Callable[[str], str]

parse_conversation(doc)

Split one stored conversation into request-to-tool-plan traces.

Parameters:

doc (dict)

Return type:

list[AgentTrace]

get_arrays_builder(name)

Look up a registered arrays builder, triggering native self-registration on first call.

Parameters:

name (str | None)

Return type:

Callable[[…], Any]

get_builder(name)

Look up a registered builder, triggering native-builder self-registration on first call.

Parameters:

name (str)

Return type:

Callable[[…], Any]

is_bit_identical_replay(trace, tools)

Replay trace and return whether every step reproduces exactly.

Parameters:

• trace (ExecutionTrace)

• tools (dict[str, Callable[[...], Any]])

Return type:

bool

llm_extractor(llm, fields, *, instruction=None, system=None)

Turn an LLM into a field-extraction teacher texts -> [{field: value}] for mixle.task.extract.distill_extractor().

Each text is extracted into a JSON object over fields (values must be verbatim substrings so they align to token spans during distillation). The returned callable has the batched-teacher shape the extractor expects.

Parameters:

• llm (LLM)

• fields (Sequence[str])

• instruction (str | None)

• system (str | None)

Return type:

Callable[[list[str]], list[dict[str, str]]]

llm_labeler(llm, labels, *, instruction=None, system=None)

Turn an LLM into a label-constrained teacher texts -> [label] for distillation / active labeling.

Each item is classified into labels by a constrained prompt; the reply is mapped back with pick_label(). The returned callable has the batched-teacher shape the rest of mixle.task expects.

Parameters:

• llm (LLM)

• labels (Sequence[str])

• instruction (str | None)

• system (str | None)

Return type:

Callable[[list[str]], list[str]]

load_harvested(path)

Read harvested serving feedback into (inputs, answers).

Two JSONL formats are supported: {"input": ..., "label": ...} for classification feedback and {"input": ..., "answer": ...} for solution feedback. Classification labels are string-coerced; solution answers keep their JSON shape. Input JSON lists are restored as tuples so record-shaped examples can be passed back into solve/distillation workflows.

Parameters:

path (str)

Return type:

tuple[list, list]

lns_classifier(student, *, step=1e-2)

Re-execute a structured student in integer log-space (the LNS rung for structured students).

The fitted model is unchanged (same factors, same JSON artifact); what changes is how inference runs: factor log-densities are quantized once at the leaf boundary, and everything above – per-label accumulation, mixture folding, the argmax decision, the posterior’s log-softmax – is integer add/max/LUT arithmetic (LNSStructuredClassifierIO). step trades fidelity for integer width; the dequantized scores match the float classifier within ~``1.5 * step`` per fold. This is compute quantization (transcendental-free combination), not weight compression – pair it with the structured student’s already compact JSON payload.

Parameters:

• student (TaskModel)

• step (float)

Return type:

TaskModel

orchestrate(question, plan_model, world, *, budget, confidence_threshold=None)

Plan one step at a time against plan_model, execute it on world, re-plan once on a failed step, and stop on an explicit STOP, low confidence, world completion, or budget exhaustion.

Parameters:

• question (str)

• plan_model (Callable[[str, list[dict[str, Any]]], dict[str, Any] | None])

• world (World)

• budget (int)

• confidence_threshold (float | None)

Return type:

OrchestrationResult

pick_label(text, labels)

Map a free-text LLM reply to one of labels (exact, then substring, else the first label).

Parameters:

• text (str)

• labels (Sequence[str])

Return type:

str

record_step(tools, tool, args, *, seed=None)

Run tools[tool] once with args (and seed, if the tool accepts one), recording the result.

Parameters:

• tools (dict[str, Callable[[...], Any]])

• tool (str)

• args (dict[str, Any])

• seed (int | None)

Return type:

TraceStep

recommend_model(data, *, fit=False, **analyze_kwargs)

Recommend a model shape for data (and optionally fit it); see ModelRecommendation.

analyze_kwargs pass through to mixle.utils.automatic.analyze_structure() (sampling, pairwise budget, validation). With fit=True the returned recommendation’s estimator is also fit and the model is attached as .model.

Parameters:

• data (Sequence[Any])

• fit (bool)

• analyze_kwargs (Any)

Return type:

ModelRecommendation

replay(trace, tools)

Re-execute every step of trace against tools with the exact same args and seed.

Parameters:

• trace (ExecutionTrace)

• tools (dict[str, Callable[[...], Any]])

Return type:

ExecutionTrace

recommend_route(cost, *, volume, n_label, p_escalate, max_escalation=None)

Pick the lowest-cost route over volume requests.

local_only is offered only when the caller explicitly disallows escalation by setting max_escalation == 0. Otherwise the cascade route keeps local answers for calibrated inputs and escalates the remaining traffic to the teacher.

Parameters:

• cost (CostModel)

• volume (int)

• n_label (int)

• p_escalate (float)

• max_escalation (float | None)

Return type:

RoutePlan

select_alpha_for_cost(model, cal_texts, cal_labels, probe_texts, cost, *, volume, n_label, alphas=(0.01, 0.05, 0.1, 0.15, 0.2, 0.3))

Select alpha from a CostModel target.

The sweep connects recommend_route() to the calibration step so threshold selection reflects both model behavior and the caller’s cost assumptions.

model is anything with the CalibratedTaskModel shape: a mutable alpha attribute, calibrate(texts, labels), and escalation_rate(texts). For each candidate in alphas, this recalibrates model and measures its realized escalation rate on probe_texts (a held-out slice disjoint from cal_texts), then scores that escalation rate with recommend_route() over volume requests. The winner is the alpha whose recommended route is lowest-cost overall; model is left calibrated at that winning alpha. Returns (best_alpha, best_plan, plan_by_alpha) so the full sweep remains auditable.

Parameters:

• model (Any)

• cal_texts (Sequence[Any])

• cal_labels (Sequence[Any])

• probe_texts (Sequence[Any])

• cost (CostModel)

• volume (int)

• n_label (int)

• alphas (Sequence[float])

Return type:

tuple[float, RoutePlan, dict[float, RoutePlan]]

replace_alerter(teacher, series, *, window=16, stride=1, **solve_kw)

Replace a threshold/heuristic alert rule over a sliding window with a calibrated model.

teacher(window) -> label (e.g. "alert"/"ok") labels every window of the historical series; the returned Solution is called with a window (the latest window samples) and answers locally only when conformally confident and in-distribution — otherwise it runs the rule.

Parameters:

• teacher (Callable[[Sequence[float]], Any])

• series (Sequence[float])

• window (int)

• stride (int)

• solve_kw (Any)

Return type:

Solution

replace_extractor(teacher, texts, fields, *, required=None, holdout=0.25, seed=0, **distill_kw)

Replace a regex/parser scraper with a distilled token-level extractor + teacher fallback.

The teacher labels the training texts; a held-out slice measures field-level F1 against the teacher. At call time a prediction missing any required field (default: all fields) falls back to the teacher — the same never-silently-wrong shape as solve().

Parameters:

• teacher (Callable[[str], dict])

• texts (Sequence[str])

• fields (Sequence[str])

• required (Sequence[str] | None)

• holdout (float)

• seed (int)

• distill_kw (Any)

Return type:

ExtractorHarness

replace_matcher(teacher, pairs, **solve_kw)

Replace a record-matching/dedup rule with a calibrated model over encoded pairs.

teacher(a, b) -> label (e.g. "match"/"no-match") labels the example pairs; each pair is encoded as one record (both sides plus numeric-difference and same-value features, which is where matchers earn their keep). Call the result with (a, b): confident pairs answer locally, everything else runs the rule.

Parameters:

• teacher (Callable[[Any, Any], Any])

• pairs (Sequence[tuple[Any, Any]])

• solve_kw (Any)

Return type:

MatcherHarness

route_stack(solutions, teacher, *, costs)

Convenience: Router.from_solutions() with tiers sorted by ascending cost.

Parameters:

• solutions (list)

• teacher (Any)

• costs (list[float])

Return type:

Router

class RefinementReport(tasks, verified_gain_pairs, solve_rate_before, solve_rate_after)

Bases: object

Measured account of one outcome-refinement round.

Parameters:

• tasks (int)

• verified_gain_pairs (int)

• solve_rate_before (float)

• solve_rate_after (float)

class ProbeHeadToHead(non_myopic_score, myopic_score, non_myopic_wins)

Bases: object

Held-out comparison between the non-myopic probe policy and a myopic baseline.

Parameters:

• non_myopic_score (float)

• myopic_score (float)

• non_myopic_wins (bool)

head_to_head_probe(plan_model, *, held_out_seeds, n_cells, n_targets, budget)

Compare the non-myopic (outcome-trained) plan model against the myopic EIG policy on the same held-out seeds at matched budget.

Parameters:

• plan_model (PlanModel)

• n_cells (int)

• n_targets (int)

• budget (int)

Return type:

ProbeHeadToHead

myopic_eig_policy(world)

One-step-lookahead policy, explicitly information-theoretic: EXPLOIT (drill) the most confident current target-candidate once its belief clears _DRILL_CONFIDENCE; otherwise EXPLORE (survey) the single most UNCERTAIN cell – maximum current entropy, i.e. the read closest to the decision boundary relative to its own noise, the textbook expected-information-gain target. No lookahead beyond this one step – by construction, it cannot see that an apparently-mediocre probe now sets up a better probe later.

Parameters:

world (ExplorationWorld)

Return type:

dict | None

outcome_refine_planner(planner, tasks, verify_fn, *, k=5, temperature=0.8, epochs=15, lr=1e-3, seed=0)

Run one propose-verify-retrain round and return the planner plus report.

For each task: sample k candidate plans (sample_plans()), keep the ones verify_fn accepts, and for tasks with at least one verified success – add the highest-scoring verified candidate as a new supervised-fine-tuning pair. Fine-tunes the LM on every such pair in one fit_pairs call. solve_rate_before/_after are measured on the same held-out tasks via the planner’s own single-shot try_plan (matched budget), before and after the retrain – not an aggregate over the k samples used to harvest the training signal.

Parameters:

• planner (GenerativePlanner)

• tasks (Sequence[str])

• verify_fn (Callable[[str, list[dict]], bool])

• k (int)

• temperature (float)

• epochs (int)

• lr (float)

• seed (int)

Return type:

tuple[GenerativePlanner, RefinementReport]

class ProposeVerifyResult(proposal, rounds=<factory>, best_candidate=None, best_result=None)

Bases: object

The full receipted history of a propose-verify-retrain run.

Parameters:

• proposal (SequenceProposal)

• rounds (list[RoundLog])

• best_candidate (tuple | None)

• best_result (OracleResult | None)

property oracle_calls: int

Return the total number of candidate evaluations sent to the oracle.

all_candidates()

Every candidate tried across every round, in order – dead ends included, none dropped.

Return type:

list[tuple[tuple, OracleResult]]

class RoundLog(round_index, candidates, results, kept_indices)

Bases: object

One round’s full record: every candidate tried and its oracle result, plus which were kept.

Parameters:

• round_index (int)

• candidates (list[tuple])

• results (list[OracleResult])

• kept_indices (list[int])

class SequenceProposal(alphabet, length, pseudo_count=1.0, position_models=<factory>)

Bases: object

A position-independent categorical proposal over fixed-length sequences from alphabet.

Parameters:

• alphabet (tuple[Any, ...])

• length (int)

• pseudo_count (float)

• position_models (list[CategoricalDistribution])

sample(k, rng)

Draw k i.i.d. sequences (each length symbols) from the current proposal.

Parameters:

• k (int)

• rng (Generator)

Return type:

list[tuple]

refit(sequences, weights)

Reweighted MLE: refit each position’s categorical on sequences, replicated in that position’s training multiset proportional to weights, through the shared optimize EM driver – never a hand-rolled frequency count.

Parameters:

• sequences (list[tuple])

• weights (ndarray)

Return type:

SequenceProposal

propose_verify_retrain(proposal, oracle, *, k_per_round, rounds, keep_frac=0.25, seed=None)

Sample, verify, keep, and refit under a fixed oracle-call budget.

Each round draws k_per_round candidates from proposal, verifies every one with oracle, keeps the top keep_frac by oracle score, and refits proposal on the kept winners weighted by score. The exact oracle-call budget is k_per_round * rounds. oracle=None raises immediately because this routine requires a verifiable objective rather than fabricating one.

Parameters:

• proposal (SequenceProposal)

• oracle (VerifiableOracle | None)

• k_per_round (int)

• rounds (int)

• keep_frac (float)

• seed (int | None)

Return type:

ProposeVerifyResult

class GridWorld(size, goal, obstacles=<factory>, step_cost=-1.0, goal_reward=10.0, max_steps=100)

Bases: object

A deterministic size x size grid MDP: a goal cell worth goal_reward, a per-step cost of step_cost, and optional impassable obstacles (moving into a wall or obstacle leaves the agent in place, still paying the step cost). The optimal policy is the shortest obstacle-free path to the goal – computable independently via optimal_path_length() (BFS), which is what makes this a closed-form-known-optimum test environment.

Parameters:

• size (int)

• goal (tuple[int, int])

• obstacles (frozenset[tuple[int, int]])

• step_cost (float)

• goal_reward (float)

• max_steps (int)

property n_states: int

Return the number of states in the square grid.

state_index(state)

Map a (row, column) state to its row-major integer index.

Parameters:

state (tuple[int, int])

Return type:

int

index_state(index)

Map a row-major integer state index back to (row, column).

Parameters:

index (int)

Return type:

tuple[int, int]

states()

Return every grid state in row-major order.

Return type:

list[tuple[int, int]]

transition(state, action)

The deterministic next state for action at state (walls/obstacles are a no-op).

Parameters:

• state (tuple[int, int])

• action (str)

Return type:

tuple[int, int]

reset(start=(0, 0))

Reset the environment to start and return the initial state.

Parameters:

start (tuple[int, int])

Return type:

tuple[int, int]

step(action)

Apply one action and return (next_state, reward, done).

Parameters:

action (str)

Return type:

tuple[tuple[int, int], float, bool]

optimal_path_length(start=(0, 0))

BFS shortest obstacle-free path length from start to goal – ground truth for tests, computed independently of any learning algorithm in this module.

Parameters:

start (tuple[int, int])

Return type:

int

class QLearningResult(q_table, rewards_per_episode)

Bases: object

The fitted Q-table plus the per-episode return trace (the learning curve).

Parameters:

• q_table (ndarray)

• rewards_per_episode (list[float])

greedy_action_index(state_index)

Return the index of the highest-valued action for state_index.

Parameters:

state_index (int)

Return type:

int

greedy_policy(env)

The recovered deterministic policy: the argmax action at every non-goal state.

Parameters:

env (GridWorld)

Return type:

dict[tuple[int, int], str]

tabular_q_learning(env, *, episodes=500, alpha=0.3, gamma=0.95, epsilon=0.2, seed=None)

Epsilon-greedy tabular Q-learning: episodes full rollouts from env.reset(), each step updating Q(s, a) toward the observed one-step Bellman target.

Parameters:

• env (GridWorld)

• episodes (int)

• alpha (float)

• gamma (float)

• epsilon (float)

• seed (int | None)

Return type:

QLearningResult

rollout(env, policy, *, start=(0, 0))

Roll out a deterministic state -> action policy from start; the (state, action) trace (stops at the goal or env.max_steps, whichever first).

Parameters:

• env (GridWorld)

• policy (dict[tuple[int, int], str])

• start (tuple[int, int])

Return type:

list[tuple[tuple[int, int], str]]

class MaxEntIRLResult(reward_weights, policy, history)

Bases: object

The recovered reward, its induced Boltzmann-rational policy, and the convergence trace (||expert_feature_expectation - policy_feature_expectation|| per iteration – should decrease toward zero as the algorithm’s own certificate of fit).

Parameters:

• reward_weights (ndarray)

• policy (ndarray)

• history (list[float])

reward(features)

Evaluate the learned linear reward on feature rows.

Parameters:

features (ndarray)

Return type:

ndarray

max_ent_irl(env, expert_trajectories, *, start=(0, 0), gamma=0.9, iterations=150, lr=0.5, features=None)

Recover linear reward weights whose maximum-entropy-optimal policy matches the expert’s empirical feature expectations, via gradient ascent on trajectory likelihood: weights += lr * (expert_feature_expectation - policy_feature_expectation). Requires only expert_trajectories (state sequences); never sees the expert’s true reward or the actions that produced them.

Parameters:

• env (GridWorld)

• expert_trajectories (list[list[tuple[int, int]]])

• start (tuple[int, int])

• gamma (float)

• iterations (int)

• lr (float)

• features (ndarray | None)

Return type:

MaxEntIRLResult

rollout_states(env, policy, *, start=(0, 0))

The state-only trace of a deterministic policy from start (the demonstration format max_ent_irl() expects: what the expert visited, not what it was thinking).

Parameters:

• env (GridWorld)

• policy (dict[tuple[int, int], str])

• start (tuple[int, int])

Return type:

list[tuple[int, int]]

state_features(env)

Default feature map: a one-hot indicator per grid cell (n_states x n_states) – a fully expressive tabular basis, so recovering per-feature weights is equivalent to recovering the per-state reward directly.

Parameters:

env (GridWorld)

Return type:

ndarray

sample_plans(planner, request, n=5, *, temperature=1.0, seed=0)

Draw n stochastic candidate plans from the trained LM, each scored by score_plan().

Sorted highest-score first. A draw that fails to parse or validate (the grammar is not enforced during stochastic sampling, unlike the constrained decode path) is returned as (None, -inf) – an undefined score IS the escalation signal: a generative decomposition model that cannot produce a coherent plan for a request should say so, never guess silently.

Parameters:

• planner (GenerativePlanner)

• request (str)

• n (int)

• temperature (float)

• seed (int)

Return type:

list[tuple[list[dict] | None, float]]

score_plan(planner, request, plan)

Mean per-character teacher-forced log-probability of a candidate plan under the trained LM.

This is not a decode: it scores a plan supplied by the CALLER (a candidate to rank against alternatives, or an already-taken plan to flag as low-probability after the fact) – the same confidence metric constrained_plan_decode() computes for its own greedy path, generalized to any plan text. Higher (less negative) is more probable; a plan scoring below the planner’s calibrated conf_floor is exactly the “low-probability plan” escalation signal used by plan-quality checks, computed explicitly here rather than left implicit in the decode loop.

Parameters:

• planner (GenerativePlanner)

• request (str)

• plan (Sequence[dict])

Return type:

float

scorecard(student, teacher, test_inputs, *, student_cost=None, teacher_cost=None, task='task')

Measure a deployed student against the teacher it replaces on held-out inputs (see module docstring).

Parameters:

• student (Any) – any solve shape — Solution, RegressionSolution, MultiLabelSolution, StructuredSolution (or anything exposing cascade.model.decide) — the escalate-aware system under test.

• teacher (Any) – the callable being replaced; also the accuracy reference.

• test_inputs (Any) – held-out inputs (the teacher is called once per input for the reference labels).

• teacher_cost (float | None) – optional per-request costs for the $/1k rows. The blended student cost prices escalated requests at teacher_cost.

• task (str) – a label for the table header.

• student_cost (float | None)

• teacher_cost

Return type:

Scorecard

sft_planner(teacher, requests, tools, *, holdout=0.2, seed=0, d_model=96, n_layer=3, n_head=4, block=192, epochs=30, lr=3e-3, device='cpu', constrained=True)

Trace-SFT a small causal LM into a plan writer, verified on held-out requests.

Traces serialize as request\n=> tool(k=v; ...) | ... \n pairs; LM.fit_pairs trains with the prompt masked so only plan tokens carry loss; generation stops at newline. Held-out agreement is plan-level exact match (tools + required args, in order) on requests the LM never saw.

Parameters:

• teacher (Callable[[str], list[dict]])

• requests (Sequence[str])

• tools (Sequence[ToolSpec])

• holdout (float)

• seed (int)

• d_model (int)

• n_layer (int)

• n_head (int)

• block (int)

• epochs (int)

• lr (float)

• device (str)

• constrained (bool)

Return type:

GenerativePlanner

spec_to_estimator(spec)

Build a mixle estimator from an allowlisted spec dict (recursively); raise on anything off the allowlist.

Specs:

• {"family": "<name>"} – a scalar leaf (see ALLOWED_FAMILIES);

• {"type": "composite", "fields": [spec, ...]} – a tuple record of sub-models;

• {"type": "mixture", "k": K, "component": spec} – a K-component mixture of the component model.

Parameters:

spec (dict[str, Any])

Return type:

Any

whitespace_invariance(text)

A meaning-preserving rewrite: collapse all whitespace runs to single spaces.

Parameters:

text (str)

Return type:

str

load_arrays(path)

Rebuild a torch-free model from an arrays-payload artifact; return (model, manifest).

Parameters:

path (str)

Return type:

tuple[Any, TaskManifest]

load_json(path)

Rebuild a pure mixle distribution from a json-payload artifact; return (model, manifest).

Parameters:

path (str)

Return type:

tuple[Any, TaskManifest]

load_module(path, *, device='cpu')

Rebuild a torch module from its manifest alone and load weights; return (module, manifest).

Parameters:

• path (str)

• device (str)

Return type:

tuple[Any, TaskManifest]

quantize_mlp(student, *, bits=8, clip_percentile=None)

Quantize a trained torch MLP student to an int8/int4, numpy-inference TaskModel.

Per-tensor symmetric weight quantization (scale = max|W| / qmax with qmax 127 for int8, 7 for int4); biases stay fp32 (they are a negligible byte fraction and quantizing them buys nothing). The returned student reuses the same featurizer and label list, reports payload="arrays" (int4 weights nibble-packed on disk: two per byte), and – having no torch dependence at inference – qualifies for torch_free devices. LNS needs LUT matmul kernels (mixle.engines.lns) and is left explicitly unimplemented.

clip_percentile guards heavy-tailed weights. Plain max-scaling lets one outlier set the whole layer’s scale: at int4 (qmax=7) a single weight 30x the rest quantizes everything else to 0, collapsing the layer. When set (e.g. 99.9), the scale is derived from that percentile of |W| instead of the max, and weights above it saturate at +/-qmax – the bulk of the distribution keeps its resolution at the cost of clipping a few outliers. Default None keeps the exact max-scale behavior (bit-identical on well-behaved weights).

Parameters:

• student (TaskModel)

• bits (int)

• clip_percentile (float | None)

Return type:

TaskModel

read_manifest(path)

Read only the manifest of an artifact directory without loading weights.

Parameters:

path (str)

Return type:

TaskManifest

register_adapter(kind, from_spec)

Register an adapter’s from_spec factory under kind so a saved io block can rebuild it.

Parameters:

• kind (str)

• from_spec (Callable[[dict[str, Any]], Any])

Return type:

None

register_arrays_builder(name, builder)

Register builder(arrays: dict[str, ndarray], **config) -> model for arrays-payload artifacts.

The arrays payload is for torch-free numeric students (e.g. an int8-quantized MLP): weights live in one .npz, and the builder reconstructs the runnable model from them in a fresh process.

Parameters:

• name (str)

• builder (Callable[[...], Any])

Return type:

None

register_builder(name, builder)

Register builder under name so an artifact carrying builder=name can reconstruct its module.

builder(**config) must return a fresh (untrained) nn.Module whose parameter shapes match the saved weights. Re-registering the same name with the same callable is a no-op; a conflicting one raises.

Parameters:

• name (str)

• builder (Callable[[...], Any])

Return type:

None

save_arrays(path, arrays, builder, config=None, *, task='', io=None, meta=None)

Persist a dict of numpy arrays as an artifact directory (arrays.npz); return path.

Parameters:

• path (str)

• arrays (dict[str, Any])

• builder (str)

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

• task (str)

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

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

Return type:

str

save_json(path, model, *, task='', io=None, meta=None)

Persist a pure (torch-free) mixle distribution via the safe serialization registry; return path.

Parameters:

• path (str)

• model (Any)

• task (str)

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

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

Return type:

str

save_module(path, module, builder, config, *, task='', io=None, meta=None)

Persist a torch module as an artifact directory and return path.

builder/config must reconstruct an architecturally identical module (get_builder(builder)(**config)); weights go through safetensors.torch.save_model so tied parameters (e.g. the LM’s tied head) round-trip.

Parameters:

• path (str)

• module (Any)

• builder (str)

• config (dict[str, Any])

• task (str)

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

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

Return type:

str

solve(teacher, inputs, *, alpha=0.1, target_agreement=None, holdout=0.25, kind=None, ood=0.02, propose=None, propose_budget=8, synthesize=0, prelabeled=None, device=None, device_space=None, cost=None, seed=0, **distill_kw)

Replace teacher (the code currently doing the job) with a calibrated, self-improving model.

Parameters:

• teacher (Callable[[...], Any]) – The callable performing the task today (per-item or batched). It labels the dataset and remains the fallback for inputs the student does not handle confidently.

• inputs (Sequence[Any]) – Example inputs (text, or tuple/dict records) covering the task. The teacher labels them.

• alpha (float) – Escalation honesty – answer locally only when a single label is conformally covered at >= 1 - alpha; otherwise fall back to the teacher.

• target_agreement (float | None) – Optional gate. If the student’s held-out agreement with the teacher misses it, the returned Solution routes everything to the teacher (promoted=False).

• holdout (float) – Fraction reserved for calibration + verification (never trained on).

• kind (str | None) – Force the student path, 'text' or 'record'; default sniffs the first input.

• ood (float | None) – Fit a p(x) gate over the training inputs and escalate inputs whose log p(x) falls below this quantile floor — so a wildly novel input escalates even when the softmax looks confident. On by default (0.02); None disables.

• propose (str | None) – "auto" searches the student recipe (dim/hidden/epochs/lr, Bayesian-optimized on a val slice carved from the training split) instead of using the defaults. Teacher-free — the labels are already computed, so candidates cost only student fits.

• propose_budget (int) – Total candidate recipes tried when propose="auto".

• synthesize (int) – When example inputs are scarce, sample this many synthetic inputs from a generative model fit to the real training inputs (record inputs only) and have the teacher label them. Labels are always real (teacher-produced); the calibration slice and the OOD gate stay real-inputs-only, so the conformal guarantee and the p(x) floor reflect the true distribution.

• prelabeled (tuple[Sequence[Any], Sequence[Any]] | None) – Already-teacher-labeled (inputs, labels) pairs — typically load_harvested("harvested.jsonl") from a serving deployment — folded into the TRAINING split (and the OOD gate: they are real traffic) but never into calibration, which stays a fresh split of inputs. This is the re-solve half of the serving loop.

• device (Any) – A DeviceSpec makes this “give me this capability on that device”: the student is found by distill_for_edge() — a structure x precision x recipe search under the device’s hard byte/ops/torch-free budget (reusing the already-computed labels; the teacher is not re-called) — and the result’s footprint, Pareto front, and design ledger land on Solution.edge. If nothing fits the budget the Solution is demoted (everything routes to the teacher). Incompatible with propose="auto" (the device search subsumes it). A plain string (e.g. "cpu") keeps its old meaning: the torch training device.

• device_space (Any) – Optional EdgeSpace constraining the device search (families, size ranges, precisions); default spans the standard space.

• cost (Any) – Optional CostModel for realized-savings reporting.

• seed (int) – Split + fit determinism.

• **distill_kw (Any) – Student knobs forwarded to distillation (dim, hidden, epochs, lr, …). student="generative" swaps the hashed-feature MLP for mixle’s generative student — per-class token models for text (mixle.task.generative_text) or the structure-learned joint for records (distill_structured_from_labels()): exact posteriors, no torch needed at inference, and a built-in log p(x).

Returns:

A Solution – call it like the original function; report() / improve() / save().

Return type:

Solution

solve_regression(teacher, inputs, *, tol, alpha=0.1, holdout=0.25, kind=None, hidden=(64,), epochs=300, lr=1e-2, dim=256, prelabeled=None, seed=0)

Replace a numeric routine with a conformally-calibrated student (see module docstring).

Parameters:

• teacher (Callable[[...], Any]) – the numeric routine (teacher(x) -> float); labels the dataset, remains the fallback.

• inputs (Sequence[Any]) – example inputs (text or dict/tuple records).

• tol (float) – the caller’s precision requirement — answer locally only when the calibrated qhat <= tol.

• alpha (float) – interval miscoverage level (1 - alpha coverage of the teacher’s answer).

• prelabeled (tuple[Sequence[Any], Sequence[float]] | None) – already-teacher-labeled (inputs, values) — typically harvested escalations from a serving deployment — folded into the TRAINING split only, never calibration (which stays a fresh split of inputs, so qhat keeps its finite-sample guarantee). The re-solve half of the serving loop.

• holdout (float)

• kind (str | None)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• dim (int)

• seed (int)

Return type:

RegressionSolution

solve_multilabel(teacher, inputs, *, alpha=0.1, holdout=0.25, kind=None, hidden=(64,), epochs=300, lr=1e-2, dim=256, prelabeled=None, seed=0)

Replace a set-of-labels routine with a per-label-calibrated student (see module docstring).

prelabeled — already-teacher-labeled (inputs, label_sets), typically harvested escalations from a serving deployment — folds into the TRAINING split only, never calibration (which stays a fresh split of inputs, so the per-label bars keep their finite-sample rank guarantee). Labels seen only in prelabeled still enter the label space.

Parameters:

• teacher (Callable[[...], Any])

• inputs (Sequence[Any])

• alpha (float)

• holdout (float)

• kind (str | None)

• hidden (Sequence[int])

• epochs (int)

• lr (float)

• dim (int)

• prelabeled (tuple[Sequence[Any], Sequence[Sequence[str]]] | None)

• seed (int)

Return type:

MultiLabelSolution

solve_structured(teacher, inputs, *, tol=None, alpha=0.1, prelabeled=None, seed=0, **sub_kw)

Replace a dict-valued routine with per-field calibrated students (see module docstring).

Parameters:

• teacher (Callable[[...], Any]) – teacher(x) -> dict with a consistent schema; called once per example input.

• inputs (Sequence[Any]) – example inputs (text or dict/tuple records).

• tol (dict[str, float] | float | None) – the precision requirement for numeric fields — a scalar for all, or {field: tol}. Required when the schema has numeric fields.

• alpha (float) – shared miscoverage level for every field’s calibration.

• prelabeled (tuple[Sequence[Any], Sequence[dict]] | None) – already-teacher-labeled (inputs, output_dicts) — typically harvested escalations from a serving deployment — fanned down into every field’s TRAINING split only, never calibration (each sub-solution’s guarantee stays a fresh split of inputs). The schema stays authoritative from the inputs pass; a pair missing a field is simply skipped for that field.

• **sub_kw (Any) – knobs forwarded to every sub-solve (epochs, hidden, dim, …).

• seed (int)

• **sub_kw

Return type:

StructuredSolution

measure_inference_seconds(student, inputs, *, repeats=5)

Median measured wall-clock seconds per single-input inference for student on this host.

Runs student.batch(inputs) repeats times (after one untimed warm-up) and reports the median per-item time. Measured, not modeled – run it on the machine whose latency you care about (the deploy device, not the dev laptop) for a number that means anything there.

Parameters:

• student (TaskModel)

• inputs (Sequence[Any])

• repeats (int)

Return type:

float

measure_ops_per_second(student, inputs, *, repeats=5)

Measured throughput (footprint ops / measured second) for this student kind on this host.

The calibration constant that turns a latency budget into DeviceSpec max_ops (DeviceSpec.for_latency()): probe once per (device, student kind), reuse across searches.

Parameters:

• student (TaskModel)

• inputs (Sequence[Any])

• repeats (int)

Return type:

float

task_fingerprint(data, labels)

A fixed small vector describing which task this is: the coords cross-task warm start keys on.

(log10 #examples, #labels, #fields, fraction of categorical fields, normalized label entropy) – low-overhead invariants of the dataset, O(1)-scaled so the design surrogate’s default lengthscale treats similar tasks as informative neighbors and dissimilar ones as weakly coupled.

Parameters:

• data (Sequence[Any])

• labels (Sequence[Any])

Return type:

list[float]

tokenize(text)

Split text into (token, start, end) triples: runs of digits, letters, or single punctuation.

Parameters:

text (str)

Return type:

list[tuple[str, int, int]]

tune_recipe(teacher, train_texts, val_texts, *, labels=None, space=None, n_init=4, n_iter=8, cost_weight=0.0, seed=0, task='')

Bayesian-optimize the distillation recipe; return the best re-distilled TaskModel.

Maximizes held-out agreement(student, teacher, val_texts) minus cost_weight * relative_train_cost. Set cost_weight > 0 to prefer the lowest-cost recipe that still matches the teacher. teacher is called once per candidate on val_texts (cached across the search) and once per candidate on train_texts.

Parameters:

• teacher (Callable[[...], Any])

• train_texts (Sequence[str])

• val_texts (Sequence[str])

• labels (Sequence[str] | None)

• space (RecipeSpace | None)

• n_init (int)

• n_iter (int)

• cost_weight (float)

• seed (int)

• task (str)

Return type:

TuneResult

tune_recipe_for_routing(teacher, train_texts, val_texts, *, labels=None, space=None, n_init=4, n_iter=8, cost_weight=0.0, calibration_frac=0.3, alpha=0.1, seed=0, task='', density_gate=False, density_gate_alpha=0.05)

Optimize a distillation recipe and calibrate the winning model for routing.

The search holds back a calibration_frac slice of val_texts before evaluating candidate recipes. That slice does not score candidates or influence the search; it is used afterward to calibrate the winning model into a CalibratedTaskModel. The result is a task-specific recipe whose complexity and epoch budget were selected from data and whose model can be passed directly to a Cascade or Router.

Teacher calls are shared through one cache. train_texts are queried once for the whole search rather than once per trial, and validation inputs that appear in both calibration and search slices are not queried twice. Every distinct input is priced once, no matter how many candidate recipes the search evaluates.

density_gate=True wires the same OOD escalation as distill_for_routing(): a gate fit on train_texts, its floor calibrated on the disjoint cal_texts slice.

Parameters:

• teacher (Callable[[...], Any])

• train_texts (Sequence[str])

• val_texts (Sequence[str])

• labels (Sequence[str] | None)

• space (RecipeSpace | None)

• n_init (int)

• n_iter (int)

• cost_weight (float)

• calibration_frac (float)

• alpha (float)

• seed (int)

• task (str)

• density_gate (bool)

• density_gate_alpha (float)

Return type:

CalibratedTuneResult

Submodules

• mixle.task.acquire module
• mixle.task.active module
• mixle.task.artifact module
• mixle.task.bandit module
• mixle.task.calibrate module
• mixle.task.calibrated_generator module
• mixle.task.capability module
• mixle.task.capacity module
• mixle.task.cascade module
• mixle.task.checkpoint_family_ladder module
• mixle.task.collapse module
• mixle.task.compose module
• mixle.task.constrained module
• mixle.task.data_mixture module
• mixle.task.density module
• mixle.task.deploy_family module
• mixle.task.design module
• mixle.task.design_prior module
• mixle.task.disagreement module
• mixle.task.discrepancy_invention_loop module
• mixle.task.distill module
• mixle.task.distill_methods module
• mixle.task.distill_soft module
• mixle.task.economics module
• mixle.task.edge module
• mixle.task.emulate module
• mixle.task.environment module
• mixle.task.explore_world module
• mixle.task.extract module
• mixle.task.frontier_to_native module
• mixle.task.generative_capability module
• mixle.task.generative_text module
• mixle.task.harness module
• mixle.task.imagine module
• mixle.task.inverse module
• mixle.task.irl module
• mixle.task.llm module
• mixle.task.model module
• mixle.task.multilabel module
• mixle.task.orchestrate module
• mixle.task.outcome_decomposer module
• mixle.task.pilot_ladder module
• mixle.task.plan module
• mixle.task.plan_model module
• mixle.task.plan_refine module
• mixle.task.probe_policy module
• mixle.task.propose module
• mixle.task.quantize module
• mixle.task.recommend module
• mixle.task.refine module
• mixle.task.regress module
• mixle.task.replay module
• mixle.task.rl module
• mixle.task.router module
• mixle.task.scorecard module
• mixle.task.sft_plan module
• mixle.task.solve module
• mixle.task.structured_out module
• mixle.task.task_decomposition module
• mixle.task.toolcall module
• mixle.task.traces module
• mixle.task.tune module
• mixle.task.vlm module