mixle.task.orchestrate module

orchestrate – the minimal controller loop: plan a step, execute it against a world, re-plan on a failed or atypical step, and stop on low confidence, world completion, or budget exhaustion.

plan_model is any (request, history) -> step | None callable – a Planner step, a GenerativePlanner decode, or a test double; None (or a step whose tool is None/"__stop__") means STOP. world is kept behind the World protocol rather than importing a concrete environment directly. Any object with step/done/score can plug in.

Every executed (or failed) step is appended to the returned trace as a TraceStep, so mixle.task.replay.replay() can later re-run the same episode against the same world.step for a bit-identical-replay check.

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

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