Planning and breaking a goal into subgoals

An agent without a plan takes one step at a time and can go in circles. An agent with a plan breaks the goal down into subgoals first, and then keeps track of which are done.

The plan should be data, not prose — then your code can validate it, show it to the user and abort if it looks wrong:

{"steps": [
  {"id": 1, "goal": "fetch the order data", "tool": "search_orders", "dependencies": []},
  {"id": 2, "goal": "check the delivery status", "tool": "search_deliveries", "dependencies": [1]},
  {"id": 3, "goal": "write the answer to the customer", "tool": null, "dependencies": [1, 2]}
]}

Replanning happens when a step fails or gives an unexpected result — but with a cap, otherwise the agent replans for ever.

from dataclasses import dataclass, field

@dataclass
class Step:
    id: int
    goal: str
    tool: str | None
    dependencies: list[int] = field(default_factory=list)
    status: str = "waiting"        # waiting|done|failed
    result: str | None = None

def validate_plan(steps: list[Step], allowed: set[str], max_steps: int = 10) -> None:
    if not 1 <= len(steps) <= max_steps:
        raise ValueError(f"the plan has {len(steps)} steps (1-{max_steps} allowed)")
    ids = {s.id for s in steps}
    for s in steps:
        if s.tool and s.tool not in allowed:
            raise ValueError(f"unknown tool: {s.tool}")
        if any(d not in ids or d >= s.id for d in s.dependencies):
            raise ValueError(f"step {s.id} has an invalid dependency")   # also catches cycles

def run_plan(steps, execute, llm, max_replans=2):
    replans = 0
    while (nxt := next((s for s in steps if s.status == "waiting"
                        and all(any(x.id == d and x.status == "done" for x in steps) for d in s.dependencies)), None)):
        nxt.result = execute(nxt)
        nxt.status = "done" if nxt.result is not None else "failed"
        if nxt.status == "failed":
            if replans >= max_replans:
                return {"status": "gave_up", "steps": steps}
            steps = llm.replan(steps, failed=nxt)
            replans += 1
    return {"status": "done" if all(s.status == "done" for s in steps) else "partial", "steps": steps}

The validation is not a formality: it catches hallucinated tool names and circular dependencies before anything is run.