mixle.inference.orchestration module¶

Learned orchestration from telemetry.

The static placement policy decides local-versus-pool execution from rules. LearnedPolicy can improve that decision from historical telemetry rows of the form (features, choice, outcome). For a new feature vector it looks up nearby historical decisions, estimates which choice had lower realized cost, and uses the learned choice only when there is enough comparable evidence.

When the nearby history is sparse or ambiguous, the policy defers to the static fallback. This keeps learned orchestration an incremental optimization around a known policy rather than an unbounded replacement.

class LearnedPolicy(keys, vecs, choices, costs, static, mean=<factory>, scale=<factory>, k=8, min_neighbors=4, margin=0.02)[source]

Bases: object

A history-based placement policy that defers to a static teacher where it lacks evidence.

Parameters:

keys (list[str])
vecs (ndarray)
choices (list[str])
costs (ndarray)
static (Callable[[dict[str, Any]], str])
mean (ndarray)
scale (ndarray)
k (int)
min_neighbors (int)
margin (float)

keys: list[str]

vecs: ndarray

choices: list[str]

costs: ndarray

static: Callable[[dict[str, Any]], str]

mean: ndarray

scale: ndarray

k: int = 8

min_neighbors: int = 4

margin: float = 0.02

decide(features)[source]

Return (choice, learned) – the learned pick when confident, else the static fallback.

Parameters:: features (dict[str, Any])
Return type:: tuple[str, bool]

evaluate(rows, *, cost_key='cost')[source]

Realized-cost comparison on held-out rows: learned policy vs always-static, vs each fixed choice.

Parameters:

rows (list[tuple[dict[str, Any], str, dict[str, Any]]])
cost_key (str)

Return type:

dict[str, Any]

class LearnedAcquisition(keys, vecs, values, static, mean=<factory>, scale=<factory>, k=8, min_neighbors=4)[source]

Bases: object

A history-based action scorer for the reasoner: learns which actions pay off, else defers.

Drop-in for mixle.substrate.act.score_action() (call it as scorer=policy in investigate). From route telemetry – each row a fired action’s (features={kind,cost,overlap}, value) – it estimates the expected yield of an action in a query’s feature region and scores it yield / cost. Where nearby history is too thin, it FALLS BACK to the static lexical scorer: the same never-worse discipline as LearnedPolicy, now on the reasoner’s acquisition decisions (J3).

Parameters:

keys (list[str])
vecs (ndarray)
values (ndarray)
static (Callable[[Any, str], float])
mean (ndarray)
scale (ndarray)
k (int)
min_neighbors (int)

keys: list[str]

vecs: ndarray

values: ndarray

static: Callable[[Any, str], float]

mean: ndarray

scale: ndarray

k: int = 8

min_neighbors: int = 4

expected_yield(features)[source]

Estimated yield of an action with these features, or None when history is too thin to say.

Parameters:: features (dict[str, Any])
Return type:: float | None

learn_action_policy(rows, static_scorer=None, *, value_key='value', k=8, min_neighbors=4)[source]

Learn a reasoner acquisition policy from route telemetry (features, kind, outcome) rows.

static_scorer is the fall-back when history is thin (default mixle.substrate.act.score_action()). value_key names the outcome field to MAXIMIZE (default "value" – did the action yield evidence). Returns a LearnedAcquisition usable directly as investigate(..., scorer=policy).

Parameters:

rows (list[tuple[dict[str, Any], str, dict[str, Any]]])
static_scorer (Callable[[Any, str], float] | None)
value_key (str)
k (int)
min_neighbors (int)

Return type:

LearnedAcquisition

learn_placement_policy(rows, static_policy, *, cost_key='cost', k=8, min_neighbors=4)[source]

Learn a placement policy from telemetry (features, choice, outcome) rows (see module docstring).

static_policy maps a feature dict to a choice and is the fall-back when history is too thin. cost_key names the outcome field to minimize (default "cost"). Feature standardization and the neighbor index are built from the rows; LearnedPolicy.decide() and evaluate follow.

Parameters:

rows (list[tuple[dict[str, Any], str, dict[str, Any]]])
static_policy (Callable[[dict[str, Any]], str])
cost_key (str)
k (int)
min_neighbors (int)

Return type:

LearnedPolicy

learn_schedule_policy(rows, static_policy, *, latency_key='latency', k=8, min_neighbors=4)[source]

Learned pool scheduling (J4): when work is pool-eligible, learn WHERE/WHEN it actually runs fastest.

The same never-worse shape as placement, keyed on realized LATENCY instead of dollar cost: rows are (features, choice, outcome) where features describe the moment (queue depth, job size, local load), choice is the scheduling decision (“run_local” / “queue_pool” / “defer”), and the outcome’s latency is what the decision actually cost in wall-clock. Where nearby history is thin, the returned policy defers to the static scheduler.

Parameters:

rows (list[tuple[dict[str, Any], str, dict[str, Any]]])
static_policy (Callable[[dict[str, Any]], str])
latency_key (str)
k (int)
min_neighbors (int)

Return type:

LearnedPolicy

meta_improve(rows, static_policy, *, cost_key='cost', holdout_frac=0.3, seed=0, k=8, min_neighbors=4)[source]

The meta-improve loop (J5): learn from telemetry, PROMOTE only on a never-worse holdout receipt.

Splits the telemetry into train/holdout, learns a policy on the train slice, and evaluates it against the static policy on the HELD-OUT decisions (realized cost, the same currency the platform pays). The learned policy is promoted iff its held-out mean cost is <= the static policy’s – the receipt is returned either way, so a non-promotion is auditable, not silent. Returns:

{promoted, policy, receipt: {learned_mean_cost, static_mean_cost, deferred_fraction, n}}

policy is the learned policy when promoted, else the static one wrapped for the same call shape – callers can always use the result’s policy and get never-worse behavior by construction.

Parameters:

rows (list[tuple[dict[str, Any], str, dict[str, Any]]])
static_policy (Callable[[dict[str, Any]], str])
cost_key (str)
holdout_frac (float)
seed (int)
k (int)
min_neighbors (int)

Return type:

dict[str, Any]