mixle.task package

Small, local, task-specific models – train or distill one, save a durable artifact, call it as a function.

The unit is a TaskModel: a fitted model scoped to one task (classify, extract, recommend a model shape, …), small enough to run locally and fast, with a durable artifact (artifact) so a plain Python program can load it in a fresh process and just call it. Producers:

  • distill() – a teacher (any callable LM) labels data, a tiny 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>)[source]

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='')[source]

Bases: object

One request -> what the agent did: the ordered tool calls and the final text reply.

Parameters:
request: str
plan: list[dict]
reply: str = ''
conversation_id: str = ''
class AgentTraces(traces=<factory>)[source]

Bases: object

The harvested corpus plus the teacher views the distillers consume.

Parameters:

traces (list[AgentTrace])

traces: list[AgentTrace]
requests(*, min_steps=0)[source]

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

Parameters:

min_steps (int)

Return type:

list[str]

tool_specs()[source]

Infer ToolSpecs from observed usage: args = union of keys, required = keys in EVERY call.

Return type:

list[ToolSpec]

call_teacher()[source]

A distill_tool_caller teacher: request -> the FIRST tool call the agent made (or no-op).

Return type:

Any

plan_teacher()[source]

A distill_planner / sft_planner teacher: request -> the agent’s full tool-call plan.

Return type:

Any

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

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]
calibrate(texts, teacher_labels)[source]

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

Parameters:
Return type:

CalibratedTaskModel

predict_sets(texts)[source]

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

text (Any)

Return type:

list[str]

decide(text)[source]

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

Parameters:

text (Any)

Return type:

Any

batch_decide(texts)[source]
Parameters:

texts (Sequence[Any])

Return type:

list[Any]

escalation_rate(texts)[source]

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

Parameters:

texts (Sequence[Any])

Return type:

float

save(path)[source]

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')[source]

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

Parameters:
Return type:

CalibratedTaskModel

class CallableLLM(fn)[source]

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

  • prompt (str)

  • system (str | None)

  • kwargs (Any)

Return type:

str

class Cascade(model, teacher, *, cost=None)[source]

Bases: object

Serve text -> label cheaply: local model when confident, teacher otherwise; track spend, harvest labels.

Parameters:

  • model (CalibratedTaskModel)

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

  • cost (CostModel | None)

serve(texts)[source]
Parameters:

texts (Sequence[Any])

Return type:

list[Any]

harvested()[source]

The escalated (texts, teacher_labels) – targeted training data to re-distill a cheaper model.

Return type:

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

realized_cost()[source]

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

Return type:

float

report()[source]

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

Project the cheapest route at volume using the realized escalation rate (needs a CostModel).

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

Bases: object

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

Parameters:
n_requests: int = 0
n_escalated: int = 0
escalated_texts: list[Any]
escalated_labels: list[Any]
property realized_escalation_rate: float
class CostModel(c_frontier, c_local=0.0, c_label=0.0, train_cost=0.0)[source]

Bases: object

Unit costs (any consistent currency). c_local is the amortized local per-request cost (~0 on CPU).

Parameters:
c_frontier: float
c_local: float = 0.0
c_label: float = 0.0
train_cost: float = 0.0
setup_cost(n_label)[source]

One-time cost to stand up a local model: label n_label examples + train.

Parameters:

n_label (int)

Return type:

float

class DensityGate(featurizer, density=None, log_threshold=None)[source]

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

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

Parameters:
Return type:

DensityGate

log_density(texts)[source]

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

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

Parameters:

text (str)

Return type:

bool

ood_mask(texts)[source]
Parameters:

texts (Sequence[str])

Return type:

ndarray

to_spec()[source]
Return type:

dict[str, Any]

classmethod from_spec(spec)[source]
Parameters:

spec (dict[str, Any])

Return type:

DensityGate

class DesignModel(signature, n_constraints, n_fingerprint=0)[source]

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)[source]
Parameters:
Return type:

None

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

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

prefilter closes the designer loop: pass a design judge – typically the tiny 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-bad 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:
Return type:

ndarray

predict(points, *, fingerprint=None)[source]

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:
Return type:

dict[str, ndarray]

to_json()[source]
Return type:

dict[str, Any]

classmethod from_json(d)[source]
Parameters:

d (dict[str, Any])

Return type:

DesignModel

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

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)[source]
Parameters:
Return type:

Any

class DeviceSpec(max_bytes=None, max_ops=None, torch_free=False)[source]

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)

max_bytes: int | None = None
max_ops: int | None = None
torch_free: bool = False
classmethod for_latency(max_ms, ops_per_second, *, max_bytes=None, torch_free=False)[source]

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 honest built-in constant.

Parameters:
Return type:

DeviceSpec

violations(fp)[source]

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

Parameters:

fp (EdgeFootprint)

Return type:

list[float]

feasible(fp)[source]
Parameters:

fp (EdgeFootprint)

Return type:

bool

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

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

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

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, ...] = ('mlp', 'structured')
dim_choices: Sequence[int] = (64, 128, 256, 512)
hidden_range: tuple[int, int] = (4, 96)
epochs_range: tuple[int, int] = (40, 320)
log10_lr_range: tuple[float, float] = (-3.0, -1.0)
bits_choices: Sequence[int] = (32, 8)
ngram: int = 3
components_range: tuple[int, int] = (1, 4)
bins_range: tuple[int, int] = (2, 8)
max_its_range: tuple[int, int] = (10, 60)
min_gain_range: tuple[float, float] = (0.0, 5.0)
dims()[source]
Return type:

int

bounds()[source]
Return type:

list[tuple[float, float]]

signature()[source]

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

Return type:

str

decode(point)[source]

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

Parameters:

point (ndarray)

Return type:

tuple[str, dict[str, Any]]

class ExtractionIO(vocab, fields, *, max_len=128)[source]

Bases: object

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

Parameters:
kind = 'extraction'
predict(module, text)[source]
Parameters:
Return type:

dict[str, str]

predict_batch(module, texts)[source]
Parameters:
Return type:

list[dict[str, str]]

predict_with_confidence(module, texts)[source]

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:
Return type:

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

to_spec()[source]
Return type:

dict[str, Any]

classmethod from_spec(spec)[source]
Parameters:

spec (dict[str, Any])

Return type:

ExtractionIO

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

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 = 0
n_requests: int = 0
report()[source]
Return type:

dict[str, Any]

save(path)[source]
Parameters:

path (str)

Return type:

str

class MatcherHarness(solution, teacher)[source]

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
report()[source]
Return type:

dict[str, Any]

class FieldChoice(path, kind, family, runner_up, gap_bits)[source]

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

Bases: object

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

Parameters:
kind = 'generative_text'
logits_batch(model, raw_inputs)[source]

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

Parameters:
Return type:

ndarray

proba_batch(model, raw_inputs)[source]

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

Parameters:
Return type:

ndarray

log_evidence(model, raw_inputs)[source]

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:
Return type:

ndarray

predict_batch(model, raw_inputs)[source]
Parameters:
Return type:

list[str]

predict(model, raw_input)[source]
Parameters:
Return type:

str

to_spec()[source]
Return type:

dict[str, Any]

classmethod from_spec(spec)[source]
Parameters:

spec (dict[str, Any])

Return type:

GenerativeTextIO

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

Bases: object

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

Deterministic and dependency-free (stdlib hashlib), so it serializes as three numbers and rebuilds identically anywhere – no fitted vocabulary, no external tokenizer. Counts are L2-normalized per row.

Parameters:
transform(texts)[source]
Parameters:

texts (list[str])

Return type:

ndarray

to_spec()[source]
Return type:

dict[str, Any]

classmethod from_spec(spec)[source]
Parameters:

spec (dict[str, Any])

Return type:

HashedNGram

class HashedRecord(dim=256, seed=0)[source]

Bases: object

Map a heterogeneous record (tuple or dict) to a fixed vector by the hashing trick – tabular tasks.

Each field is hashed by key: a categorical/string/bool value contributes a 1 at hash("key=value"); a numeric value contributes a bounded tanh(value) at hash("num:key") (and its presence at a second bucket). Stateless and deterministic – no fitted encoder or vocabulary – so it serializes as two numbers and rebuilds identically. The shape of a “classify this record / route this ticket / flag this transaction” model.

Parameters:
transform(records)[source]
Parameters:

records (list[Any])

Return type:

ndarray

to_spec()[source]
Return type:

dict[str, Any]

classmethod from_spec(spec)[source]
Parameters:

spec (dict[str, Any])

Return type:

HashedRecord

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

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:
kind = 'lns_structured_classifier'
int_logits_batch(model, raw_inputs)[source]

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

Parameters:
Return type:

ndarray

logits_batch(model, raw_inputs)[source]

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

Parameters:
Return type:

ndarray

proba_batch(model, raw_inputs)[source]

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:
Return type:

ndarray

predict_batch(model, raw_inputs)[source]
Parameters:
Return type:

list[str]

to_spec()[source]
Return type:

dict[str, Any]

classmethod from_spec(spec)[source]
Parameters:

spec (dict[str, Any])

Return type:

LNSStructuredClassifierIO

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

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 = None
low_confidence_fields()[source]

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

Return type:

list[FieldChoice]

fit(data, **kwargs)[source]

Fit the recommended estimator on data and return the model.

Parameters:
Return type:

Any

explain()[source]

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

Bases: object

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

Parameters:
complete(prompt, *, system=None, **kwargs)[source]
Parameters:

  • prompt (str)

  • system (str | None)

  • kwargs (Any)

Return type:

str

class QuantizedClassifierIO(featurizer, labels)[source]

Bases: _ClassifierIO

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

Parameters:

  • featurizer (Any)

  • labels (list[str])

kind = 'quantized_classifier'
logits_batch(model, raw_inputs)[source]
Parameters:
Return type:

ndarray

to_spec()[source]
Return type:

dict[str, Any]

classmethod from_spec(spec)[source]
Parameters:

spec (dict[str, Any])

Return type:

QuantizedClassifierIO

class QuantizedMLP(layers, *, bits=8)[source]

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:
logits(feats)[source]
Parameters:

feats (ndarray)

Return type:

ndarray

nbytes()[source]

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

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

Return type:

int

to_arrays()[source]
Return type:

dict[str, ndarray]

classmethod from_arrays(arrays)[source]
Parameters:

arrays (dict[str, ndarray])

Return type:

QuantizedMLP

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

Bases: object

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

Parameters:
dim_choices: Sequence[int] = (128, 256, 512, 1024)
hidden_range: tuple[int, int] = (16, 128)
epochs_range: tuple[int, int] = (50, 400)
log10_lr_range: tuple[float, float] = (-3.0, -1.0)
n: int = 4
dims()[source]
Return type:

int

decode(point)[source]
Parameters:

point (ndarray)

Return type:

dict[str, Any]

cost(recipe)[source]

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

list[tuple[float, float]]

class RecordClassifierIO(featurizer, labels)[source]

Bases: _ClassifierIO

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

Parameters:

  • featurizer (HashedRecord)

  • labels (list[str])

kind = 'record_classifier'
class RoutePlan(route, volume, per_request, total, savings_vs_frontier, p_escalate, break_even, options)[source]

Bases: object

The costed comparison of routes at a given volume, with the recommended choice and its savings.

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

Bases: object

Route each request to the cheapest tier whose calibrated model is confident; the last tier always answers.

Parameters:

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

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

Build from Solution objects (cheapest-first) + the frontier callable.

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

Parameters:
Return type:

Router

serve(xs)[source]
Parameters:

xs (Any)

Return type:

list[Any]

harvested()[source]

The frontier-answered (inputs, labels) — targeted training data for the cheaper tiers.

Return type:

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

report()[source]

Per-tier traffic and REALIZED economics vs sending everything to the final tier.

Return type:

dict[str, Any]

summary()[source]
Return type:

str

class Scorecard(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')[source]

Bases: object

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)

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

dict[str, Any]

table()[source]
Return type:

str

class RouterStats(tiers: 'list[TierStats]' = <factory>, harvested_inputs: 'list[Any]' = <factory>, harvested_labels: 'list[Any]' = <factory>)[source]

Bases: object

Parameters:
tiers: list[TierStats]
harvested_inputs: list[Any]
harvested_labels: list[Any]
property n_requests: int
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>)[source]

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 = (64,)
epochs: int = 300
lr: float = 0.01
seed: int = 0
n_requests: int = 0
n_escalated: int = 0
harvested_inputs: list
harvested_ys: list
interval(x)[source]

(yhat, lo, hi) with calibrated >= 1 - alpha coverage of the teacher’s answer.

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

report()[source]
Return type:

dict[str, Any]

save(path)[source]

Persist net (safetensors via the mixle.mlp builder) + featurizer + calibration; load() restores.

Parameters:

path (str)

Return type:

str

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

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

Parameters:
Return type:

RegressionSolution

improve()[source]

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

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 = (64,)
epochs: int = 300
lr: float = 0.01
seed: int = 0
n_requests: int = 0
n_escalated: int = 0
harvested_inputs: list
harvested_sets: list
try_local(x)[source]

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

Parameters:

x (Any)

Return type:

list[str] | None

report()[source]
Return type:

dict[str, Any]

save(path)[source]

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

Parameters:

path (str)

Return type:

str

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

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

Parameters:
Return type:

MultiLabelSolution

improve()[source]

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

Return type:

bool

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

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 = 0
n_escalated: int = 0
harvested_inputs: list
harvested_outputs: list
property schema: dict[str, str]
try_local(x)[source]

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

Parameters:

x (Any)

Return type:

dict[str, Any] | None

report()[source]
Return type:

dict[str, Any]

save(path)[source]

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')[source]

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

Parameters:
Return type:

StructuredSolution

improve()[source]

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

Return type:

bool

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

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 = 160
constrained: bool = True
conf_floor: float | None = None
lm_config: dict
n_requests: int = 0
n_escalated: int = 0
harvested: list[tuple[str, list[dict]]]
try_plan(request)[source]

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

dict[str, Any]

save(path)[source]

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')[source]

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

Parameters:
Return type:

GenerativePlanner

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

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 = 8
n_requests: int = 0
n_escalated: int = 0
harvested: list[tuple[str, list[dict]]]
try_plan(request, *, execute=None)[source]

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

No teacher, no stats — this is what a server without the frontier can run.

Parameters:
Return type:

dict[str, Any] | None

report()[source]
Return type:

dict[str, Any]

save(path)[source]

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

Parameters:

path (str)

Return type:

str

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

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

Parameters:
Return type:

Planner

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

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 (an honest failure, never a silently bad model).

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 = None
seed: int = 0
synthesized: int = 0
gate_inputs: list
edge: Any = None
report()[source]

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

Return type:

dict

improve()[source]

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 stay honest across rounds.

Return type:

bool

health(recent_inputs=None, *, p_threshold=0.01)[source]

Is the calibration still holding on live traffic? The guarantee-watcher.

The conformal escalate-or-answer rate was calibrated under exchangeability; when the input distribution drifts, the live escalation rate moves away from the verified baseline — which is exactly the observable to alarm on. This compares the live rate against the baseline with an exact binomial test, and (when recent_inputs are supplied and an OOD gate exists) also checks the gate’s out-of-distribution hit-rate against its design quantile.

Returns a dict with drifted (bool), the rates, and p-values. Honest caveat: a drift alarm means “the world changed, escalations (and cost) will rise, re-solve soon” — the SYSTEM stays correct throughout because unsure inputs still go to the teacher.

Parameters:

  • recent_inputs (Any)

  • p_threshold (float)

Return type:

dict[str, Any]

save(path)[source]

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')[source]

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

Parameters:
Return type:

str

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

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:
Return type:

Solution

class StructuredClassifierIO(field_keys, label_index, labels)[source]

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:
kind = 'structured_classifier'
logits_batch(model, raw_inputs)[source]

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

Parameters:
Return type:

ndarray

proba_batch(model, raw_inputs)[source]

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

Parameters:
Return type:

ndarray

predict_batch(model, raw_inputs)[source]
Parameters:
Return type:

list[str]

predict(model, raw_input)[source]
Parameters:
Return type:

str

to_spec()[source]
Return type:

dict[str, Any]

classmethod from_spec(spec)[source]
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='')[source]

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 = None
config: dict[str, Any]
task: str = ''
io: dict[str, Any]
meta: dict[str, Any]
schema_version: str = '1'
created_at: str = ''
to_dict()[source]
Return type:

dict[str, Any]

classmethod from_dict(d)[source]
Parameters:

d (dict[str, Any])

Return type:

TaskManifest

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

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

raw_inputs (list[Any])

Return type:

list[Any]

save(path)[source]

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')[source]

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

Parameters:
Return type:

TaskModel

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

Bases: object

A distilled function-caller: emit a call only when selection AND arguments are trustworthy.

Parameters:
selector: Solution
extractors: dict[str, Any]
tools: dict[str, ToolSpec]
teacher: Callable[[str], dict]
selection_agreement: float
n_requests: int = 0
n_escalated: int = 0
harvested: list[tuple[str, dict]]
try_local(request)[source]

The local decision alone: a trustworthy call / no-op, or None (= must escalate).

No teacher, no stats — this is what a server without the frontier can run.

Parameters:

request (str)

Return type:

dict[str, Any] | None

report()[source]
Return type:

dict[str, Any]

save(path)[source]

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

Parameters:

path (str)

Return type:

str

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

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

Parameters:
Return type:

ToolCaller

class ToolSpec(name, args, required=None)[source]

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 = None
property required_args: list[str]
class TextClassifierIO(featurizer, labels)[source]

Bases: _ClassifierIO

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

Parameters:

  • featurizer (HashedNGram)

  • labels (list[str])

kind = 'text_classifier'
class TuneResult(model, recipe, agreement, score, cost, history=None)[source]

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 = None
acquisition_scores(student, texts, method='margin')[source]

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

Parameters:
Return type:

ndarray

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

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:
Return type:

ActiveResult

adapter_from_spec(spec)[source]

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

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

Parameters:
Return type:

float

break_even_volume(cost, n_label, *, p_escalate=0.0)[source]

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

cascade_cost_per_request(cost, p_escalate)[source]

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

Parameters:

  • cost (CostModel)

  • p_escalate (float)

Return type:

float

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

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, bad JSON, off-allowlist family, fit error) yields the heuristic mixle.task.recommend.recommend_model() estimator when fallback is set.

Parameters:
Return type:

DesignedModel

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

Label texts with teacher, fit a small student to match, 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.

Parameters:
Return type:

TaskModel

distill_designer(design, *, quality_quantile=0.5, seed=0)[source]

Distill the design ledger into a tiny 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='')[source]

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

Parameters:
Return type:

TaskModel

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

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:
Return type:

Planner

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

Distill the teacher’s function-calling into a tiny selector + 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='')[source]

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 (cheap), 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 tiny 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:
Return type:

EdgeDistillResult

footprint(student)[source]

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_from_labels(texts, teacher_labels, *, labels=None, n=3, dim=256, hidden=(64,), epochs=200, lr=1e-2, seed=0, task='', device='cpu')[source]

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:
Return type:

TaskModel

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

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

Parameters:
Return type:

TaskModel

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

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

Parameters:
Return type:

TaskModel

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

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:
Return type:

TaskModel

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

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

Parameters:
Return type:

TaskModel

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

Distill a teacher into a tiny 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:
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='')[source]

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

Parameters:
Return type:

TaskModel

extraction_f1(model, gold, texts)[source]

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

Parameters:
Return type:

float

harvest_agent_traces(directory=None)[source]

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

Parameters:

directory (str | Path | None)

Return type:

AgentTraces

parse_conversation(doc)[source]

Split one stored conversation into traces: user request -> assistant tool calls until the next user turn.

Parameters:

doc (dict)

Return type:

list[AgentTrace]

get_arrays_builder(name)[source]

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

Parameters:

name (str | None)

Return type:

Callable[[…], Any]

get_builder(name)[source]

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

Parameters:

name (str)

Return type:

Callable[[…], Any]

llm_extractor(llm, fields, *, instruction=None, system=None)[source]

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:
Return type:

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

llm_labeler(llm, labels, *, instruction=None, system=None)[source]

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:
Return type:

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

load_harvested(path)[source]

Read the harvested.jsonl a serving endpoint accumulates into (inputs, answers).

Two line formats are understood: {"input": ..., "label": ...} (the mixle-mlops /v1/tasks/{name}/feedback classification format — labels are str-coerced) and {"input": ..., "answer": ...} (the /v1/solutions/{name}/feedback format for every solve shape — the answer keeps its shape: a number for regression, a list of labels for multilabel, a dict for structured). Input JSON lists coerce back to tuples — the record shape the solve trained on — so the pairs feed straight into solve*(..., prelabeled=load_harvested(p)).

Parameters:

path (str)

Return type:

tuple[list, list]

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

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-tiny JSON payload.

Parameters:

  • student (TaskModel)

  • step (float)

Return type:

TaskModel

pick_label(text, labels)[source]

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

Parameters:
Return type:

str

recommend_model(data, *, fit=False, **analyze_kwargs)[source]

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:
Return type:

ModelRecommendation

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

Pick the cheapest route over volume requests; honor an optional cap on tolerated escalation.

local_only is treated as a cascade with no escalation only when its quality is acceptable to the caller – here it is offered whenever max_escalation is not exceeded by p_escalate (the cascade already answers the easy cases locally and escalates the rest, so it dominates pure local-only on accuracy; local-only is kept as the floor cost when escalation is disallowed entirely, max_escalation == 0).

Parameters:

  • cost (CostModel)

  • volume (int)

  • n_label (int)

  • p_escalate (float)

  • max_escalation (float | None)

Return type:

RoutePlan

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

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:
Return type:

Solution

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

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:
Return type:

ExtractorHarness

replace_matcher(teacher, pairs, **solve_kw)[source]

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:
Return type:

MatcherHarness

route_stack(solutions, teacher, *, costs)[source]

Convenience: Router.from_solutions() with tiers sorted cheapest-first by cost.

Parameters:
Return type:

Router

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

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

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:
Return type:

GenerativePlanner

spec_to_estimator(spec)[source]

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

load_arrays(path)[source]

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

Parameters:

path (str)

Return type:

tuple[Any, TaskManifest]

load_json(path)[source]

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')[source]

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

Parameters:
Return type:

tuple[Any, TaskManifest]

quantize_mlp(student, *, bits=8, clip_percentile=None)[source]

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

Read just the manifest of an artifact directory (cheap: no weights loaded).

Parameters:

path (str)

Return type:

TaskManifest

register_adapter(kind, from_spec)[source]

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

Parameters:
Return type:

None

register_arrays_builder(name, builder)[source]

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:
Return type:

None

register_builder(name, builder)[source]

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:
Return type:

None

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

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

Parameters:
Return type:

str

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

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

Parameters:
Return type:

str

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

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

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 is not sure about.

  • 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 — an honest failure). 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)[source]

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

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:
Return type:

MultiLabelSolution

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

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

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:
Return type:

float

measure_ops_per_second(student, inputs, *, repeats=5)[source]

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:
Return type:

float

task_fingerprint(data, labels)[source]

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) – cheap 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:
Return type:

list[float]

tokenize(text)[source]

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='')[source]

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 cheapest 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:
Return type:

TuneResult

Submodules