PlanAndExecute

Ask the model for a plan (a dependency graph of steps), run the steps as parallel child runs, repair the plan if a step fails, then synthesize a final answer.

class PlanAndExecute:
    def __init__(
        self,
        executor,
        *,
        model=None,
        max_parallel=3,
        max_replans=1,
        synthesize=True,
        planner_system=...,
    ): ...

"""Plan once, run the steps as parallel child runs honoring dependencies, then synthesize."""

import asyncio
import json

from aaa.algos import PlanAndExecute, ToolLoop
from aaa.llm import FakeLLMClient, LLMHandler
from aaa.llm.fake import text, tool_call
from aaa.runtime import Engine
from aaa.tools import ToolHandler, ToolRegistry, tool


@tool
def population(city: str) -> int:
    """Population of a city (millions)."""
    return {"Tokyo": 14, "Delhi": 32, "Lima": 10}[city]


PLAN = {
    "steps": [
        {"id": "a", "description": "population of Tokyo", "depends_on": []},
        {"id": "b", "description": "population of Delhi", "depends_on": []},
        {"id": "c", "description": "add the two populations", "depends_on": ["a", "b"]},
    ]
}


def model(request):
    system = request.messages[0].content if request.messages[0].role == "system" else ""
    last = request.messages[-1]
    if "planner" in system:
        return text(json.dumps(PLAN))
    if "Combine" in system:
        return text("Tokyo + Delhi = 46 million people")
    if last.role == "user":  # an executor child: do its one step
        step = last.content.split("Your step (")[1].split(")")[0]
        if step in ("a", "b"):
            return tool_call("population", city="Tokyo" if step == "a" else "Delhi")
        return text("46")
    return text(last.content)


async def main() -> None:
    registry = ToolRegistry([population])
    async with Engine() as engine:
        LLMHandler(FakeLLMClient(model)).attach(engine)
        ToolHandler(registry).attach(engine)
        agent = PlanAndExecute(ToolLoop(registry), max_parallel=2)
        run = await engine.start(agent, {"task": "Total population of Tokyo and Delhi?"})
        outcome = await run.result()
        log = await engine.store.read(run.run_id)
    print(outcome.value["text"])
    print("steps run:", [e.data["step_id"] for e in log if e.type == "aaa.task.started"])


asyncio.run(main())

Tokyo + Delhi = 46 million people
steps run: ['a', 'b', 'c']

How it works

flowchart LR
    T[task] --> PL[Planner LLM<br/>structured JSON plan]
    PL --> V{valid DAG?}
    V -->|no| F1[failed: invalid plan]
    V -->|yes| S[Scheduler]
    S --> A[step a]
    S --> B[step b]
    A --> C[step c<br/>depends on a, b]
    B --> C
    C --> SY[Synthesis LLM]
    SY --> OUT[success]
    A -.fails.-> RP[Replan<br/>max_replans]
    RP --> S

1. Plan. One LLM call with a JSON schema asks for {"steps": [{"id", "description", "depends_on": []}]}.
2. Validate. Duplicate ids, unknown dependencies and cycles are rejected (failed, reason invalid plan: ...). A plan is code-checked before any step runs.
3. Execute. Ready steps (all dependencies done) start as child runs of executor (any algorithm, typically a ToolLoop), at most max_parallel at once. Each child receives the overall task, its step, and the results of its prerequisites.
4. Replan. If a step fails and replans remain, the planner is asked for the remaining steps given what is done and what failed. New step ids get an r<n>- prefix. Beyond max_replans, the run fails with the failing step's reason.
5. Synthesize. One more call combines all step results into the answer.

The plan itself is in the log (aaa.plan.created, aaa.plan.revised, aaa.task.started), so it is inspectable, resumable and replayable.

Parameters

Parameter	Default
executor	required	The algorithm that runs each step. Its name must be unique and registered on every worker.
model	None	Model for planning and synthesis.
max_parallel	3	Concurrent step runs.
max_replans	1	How many times a failure may trigger replanning. 0 disables.
synthesize	True	False skips the final call and returns {"results": {...}}.
planner_system	built-in	Override the planner prompt. Keep the JSON shape.

Result: {"text": <synthesis>, "results": {step_id: result}}.

Budgets and children

Child runs have their own outcomes and their spend rolls up into the parent's budget, so one Budget(max_cost=1.0) on the parent bounds the whole tree. If the parent finishes or is cancelled, running children are cancelled.

When not to use it

For a task needing two or three tool calls in sequence, a ToolLoop is simpler and cheaper: planning and synthesis add two LLM calls. Choose PlanAndExecute when steps are independent and can run in parallel, are individually long, or when you want the plan itself as an auditable artifact.