mixle.evolve.closed_loop module

L1: closed-loop self-evolution with operator credit – the loop at five altitudes, wired end to end.

Router-harvested failures -> acquisition (A5’s mixle.task.acquire.acquire()) -> challenger production (a distill/refine/evolve operator, all reused ImprovementOperator instances, not reimplemented) -> the held-out challenger_beats_champion() gate -> deploy, as ONE budgeted background loop (ClosedLoopSelfEvolution). A per-context meta-bandit (OperatorCreditBandit, wrapping mixle.task.bandit.UCB1 – the same UCB1 machinery reused across this codebase for meta-bandits) learns which challenger-production operator wins for which kind of failure, and every ADOPTED champion gets a genealogy receipt (GenealogyLedger, built directly on EvolutionLedger) recording its parent, the operator that produced it, and the measured gap – a real, walk-backable lineage.

Which subsystems this wires, not rebuilds:

  • mixle.task.router – harvested_from_router() reads a real Router’s .harvested() (the frontier-answered cases the cheap tiers could not handle) as the harvested-failure source when a caller already runs one. For domains without a live Router (e.g. this module’s own tests, which evolve a marginal label model rather than a routed per-input classifier), harvest_failures() is the same idea generalized to any Objective: the observations where the current champion did NOT score at its own best-attainable pointwise value – literally “the cases it got wrong.”

  • mixle.task.acquire.acquire() (A5) ranks the harvested pool before it is spent on challenger production. acquire needs a “scoreable” model (predict_proba-shaped); a champion that is a bare marginal distribution (does not condition on the item) is wrapped by _ConstantProbaAdapter so the real acquire() code path runs (entropy strategy) instead of being skipped – honestly, this degenerates to “prioritize while the champion overall is unconfident” since the per-item probability is constant, but it is the real primitive, not a mock.

  • mixle.evolve.operators – three EXISTING operators stand in for L1’s “distill / refine / evolve ops” triad (see default_challenger_operators()): AutoSelect (cold-start refit – distillation’s “train a fresh model from the teacher-labeled pool” shape) as "distill", Refit (warm-started refit of the champion’s own parameters) as "refine", and Mutate (the genetic-programming structure-edit operator) as "evolve".

  • mixle.evolve.verify.challenger_beats_champion() is the held-out gate, used exactly as-is.

  • mixle.evolve.ledger.EvolutionLedger is the genealogy substrate, wrapped (not replaced) by GenealogyLedger.

  • mixle.task.bandit.UCB1 is the meta-bandit’s arm-selection machinery, one instance per context.

Principled crossover. If a challenger-production operator ever combines two existing champions (a “crossover” in evolutionary-algorithm terms), it MUST do so via mixture composition (mixle.ops.mixture(), exactly what Mutate’s grow move and Recompose already do) – never by cutting and pasting incompatible weight sub-blocks from two unrelated models. principled_crossover() is the explicit primitive for that case.

accuracy_objective()[source]

Higher-is-better accuracy for a model exposing a single “most likely label” prediction (a fitted CategoricalDistribution’s argmax(pmap)): per-observation 1{y == predicted_label}.

Return type:

Objective

harvest_failures(champion, batch, objective)[source]

The objective-generic analog of mixle.task.router.Router.harvested(): the observations in batch where champion did NOT score at its own best-attainable pointwise value under objective – for accuracy_objective() this is exactly “the cases it got wrong.” A scalar-only objective (no honest per-observation vector) harvests the whole batch – there is no finer signal to rank on.

Parameters:
Return type:

list[Any]

harvested_from_router(router)[source]

Pull the REAL harvested failure pool from a live mixle.task.router.Router: every input that escalated all the way to the frontier, paired with its frontier label – exactly router.harvested(), just zipped into one pool for mixle.task.acquire.acquire().

Parameters:

router (Any)

Return type:

list[tuple[Any, Any]]

default_challenger_operators()[source]

The three L1 challenger-production operators, each a reused ImprovementOperator (no bespoke fitting logic):

  • "distill" -> AutoSelect – cold-start: fit a fresh model from the harvested pool, the “train a new model from teacher-labeled data” shape of distillation.

  • "refine" -> Refit – warm-started refit of the champion’s own parameters on the harvested pool.

  • "evolve" -> Mutate – genetic-programming structure edit (grow/shrink/perturb), mixle.evolve’s own structure-search operator.

Return type:

dict[str, ImprovementOperator]

principled_crossover(model_a, model_b, *, weight_a=0.5)[source]

Combine two champions the ONLY principled way this framework allows: mixture composition (mixle.ops.mixture()), never gene-splicing (cutting/pasting incompatible weight sub-blocks). Returns an unfitted mixture prototype; refit it against data (e.g. via its own .estimator()) before treating it as a challenger.

Parameters:
Return type:

Any

class OperatorCreditBandit(operator_names, *, c=1.0, seed=0)[source]

Bases: object

A per-context meta-bandit over challenger-production operators, wrapping one mixle.task.bandit.UCB1 per context (e.g. a failure type/domain) – so the loop learns, independently for each context, which operator actually produces winning challengers there.

Parameters:
select(context)[source]

The bandit-chosen operator name for context.

Parameters:

context (str)

Return type:

str

reward(context, operator, reward)[source]

Fold an observed (non-negative, anti-regression) reward back into operator’s arm.

Parameters:
Return type:

None

report()[source]

Per-context, per-operator mean reward and pull count.

Return type:

dict[str, dict[str, float]]

class GenealogyLedger(ledger=<factory>, _model_ids=<factory>, _counter=0)[source]

Bases: object

Genealogy receipts for adopted (gate-passing) champions, built directly on EvolutionLedger (never storing model objects in the ledger rows themselves – only their operator, measured gap, and a stable id – exactly the ledger’s own JSON-serializability discipline).

Parameters:

  • ledger (EvolutionLedger)

  • _model_ids (dict[int, str])

  • _counter (int)

record_adoption(*, parent, child, operator, gap, context, meta=None)[source]

Record ONE adoption: child replaced parent via operator, measured gap (the verified challenger-beats-champion delta). parent=None marks the root of a lineage.

Parameters:
Return type:

dict[str, Any]

lineage(model)[source]

Reconstruct model’s full lineage: the ordered (root-first) chain of adoption receipts {operator, delta (the measured gap), parent_hash, meta: {child_hash, context, ...}} back to the first recorded ancestor. Returns [] if model was never recorded as an adoption.

Parameters:

model (Any)

Return type:

list[dict[str, Any]]

class LoopStepResult(context, operator, promoted, delta, champion_score, champion)[source]

Bases: object

One ClosedLoopSelfEvolution.step() outcome.

Parameters:
class ClosedLoopSelfEvolution(champion, *, objective, operators=None, context_fn=None, acquire_k=32, acquire_strategy='entropy', bandit_c=1.0, seed=0)[source]

Bases: object

The whole L1 loop, one budgeted step at a time: harvest -> acquire -> propose -> gate -> deploy, with per-context operator credit and genealogy receipts on every adoption.

Parameters:

  • champion (Any) – the initial fitted model (e.g. a CategoricalDistribution).

  • objective (Objective) – the Objective the loop optimizes (e.g. accuracy_objective()).

  • operators (dict[str, ImprovementOperator] | None) – challenger-production operators by name; defaults to default_challenger_operators().

  • context_fn (Callable[[Sequence[Any]], str] | None) – batch -> context key for the per-context bandit; defaults to a single global context ("default") when the caller has no domain/failure-type signal to condition on.

  • acquire_k (int) – how many harvested failures A5’s acquire prioritizes per step.

  • acquire_strategy (str) – the acquire() ranking strategy (default "entropy", the one that works for a marginal/non-conditional champion via _ConstantProbaAdapter).

  • seed (int) – RNG seed threaded through the bandit and the gate.

  • bandit_c (float)

step(batch, *, verify=None, context=None)[source]

One cycle of the loop on one arriving batch of held-out-shaped observations.

  1. harvest failures under the CURRENT champion,

  2. rank them with A5’s acquire,

  3. pick a challenger-production operator via the per-context bandit,

  4. propose + gate the challenger,

  5. reward the bandit with the verified (anti-regression) delta,

  6. deploy + record a genealogy receipt iff the gate promotes.

Parameters:
Return type:

LoopStepResult

run(stream, *, verify_batches=None, contexts=None, budget=None)[source]

Run the loop over a SEQUENCE of batches (a stream) – the budgeted background loop’s outer iteration. budget (if given) is a ceiling on the total cost_hint of operators actually applied (proposal attempts that were skipped as inapplicable don’t spend budget); the loop stops early once it is exhausted, leaving the champion at whatever it last became.

Parameters:
Return type:

list[LoopStepResult]