AI Agent Planning Patterns

Real-World Scenario

An agent’s job is “email the monthly sales report to the sales team”. That is four steps with an order: load the data, summarize it, write the report, send it. A good planning agent writes those steps down, checks the plan makes sense, runs the steps in dependency order, and changes course when the September export turns out not to exist, instead of emailing a report about nothing.

The model’s part is played by planner, a stand-in that returns a JSON plan the way an LLM would when asked for one. Everything around it is the real code an agent needs, and it runs without an API key. The planning lives in that code. The model proposes a plan, and your code decides whether it is valid, what runs when, and what happens on failure.

The Tools

Four small tools the plans can use. Save them as tools.py.

SALES = {"2026-08": [("north", 41200), ("south", 38950), ("west", 27400)]}
OUTBOX = []                                              # emails "sent" land here

def load_sales(month):
    if month not in SALES:
        raise LookupError(f"no sales export for {month}")
    return SALES[month]

def summarize(rows):
    total = sum(amount for _, amount in rows)
    best = max(rows, key=lambda r: r[1])[0]
    return {"total": total, "best_region": best, "regions": len(rows)}

def write_report(summary, month):
    return (f"Sales {month}: total {summary['total']:,} across {summary['regions']} regions; "
            f"best region {summary['best_region']}.")

def send_email(to, body):
    OUTBOX.append((to, body))
    return f"sent to {to}"

TOOLS = {f.__name__: f for f in (load_sales, summarize, write_report, send_email)}

TOOLS is the list of names a plan may use. Anything else is a hallucinated tool.

Validating a Plan Before Anything Runs

A plan is data: steps with a tool and the steps they depend on. Check it the way you’d check any untrusted input.

import json
from graphlib import TopologicalSorter, CycleError
from tools import TOOLS

def check_plan(plan_json):
    """Reject a model-written plan before anything runs."""
    steps = {s["id"]: s for s in json.loads(plan_json)}
    problems = [f"step {i}: unknown tool {s['tool']!r}" for i, s in steps.items() if s["tool"] not in TOOLS]
    problems += [f"step {i}: depends on missing step {d}" for i, s in steps.items()
                 for d in s["after"] if d not in steps]
    if problems:
        return None, problems
    try:
        order = list(TopologicalSorter({i: s["after"] for i, s in steps.items()}).static_order())
    except CycleError as e:
        return None, [f"cycle: {e.args[1]}"]
    return [steps[i] for i in order], []

if __name__ == "__main__":
    bad = '''[{"id": 1, "tool": "load_sales", "after": []},
              {"id": 2, "tool": "summarize", "after": [1, 7]},
              {"id": 3, "tool": "post_to_slack", "after": [2]}]'''
    loop = '''[{"id": 1, "tool": "summarize", "after": [2]}, {"id": 2, "tool": "write_report", "after": [1]}]'''
    for plan in (bad, loop):
        print(check_plan(plan)[1])

$ python validate.py
["step 3: unknown tool 'post_to_slack'", 'step 2: depends on missing step 7']
['cycle: [1, 2, 1]']

The first plan has two errors a model really makes: a tool that doesn’t exist (post_to_slack was never offered) and a dependency on a step that was never written. The second is a cycle: summarize needs the report, and the report needs the summary, so no order can satisfy it. TopologicalSorter finds the cycle and names the steps in it. For a valid plan, it returns the order to run the steps in.

These messages are written for the model as well as for you. Sending them back with “fix this plan” usually gets a corrected plan on the next try.

Plan, Execute, Replan

The executor runs validated steps in order and passes each step’s result to later steps. When a step fails, it stops and asks for a new plan.

import json

def planner(goal, month, feedback=None):
    """Stands in for an LLM asked to return a JSON plan; replans when given feedback."""
    if feedback is None:
        steps = [(1, "load_sales", {"month": month}, []),
                 (2, "summarize", {"rows": "$1"}, [1]),
                 (3, "write_report", {"summary": "$2", "month": month}, [2]),
                 (4, "send_email", {"to": "sales-team", "body": "$3"}, [3])]
    else:   # the failure is visible to the model, so it can choose a different route
        steps = [(1, "send_email", {"to": "sales-team",
                  "body": f"The {month} report is delayed: {feedback}."}, [])]
    return json.dumps([{"id": i, "tool": t, "args": a, "after": d} for i, t, a, d in steps])

from planner import planner
from tools import TOOLS, OUTBOX
from validate import check_plan

def execute(plan):
    results = {}
    for step in plan:
        args = {k: results[int(v[1:])] if isinstance(v, str) and v.startswith("$") else v
                for k, v in step["args"].items()}           # "$2" means "step 2's result"
        try:
            results[step["id"]] = TOOLS[step["tool"]](**args)
        except Exception as e:
            print(f"  step {step['id']} {step['tool']} FAILED: {e}")
            return None, str(e)
        print(f"  step {step['id']} {step['tool']} -> {results[step['id']]}")
    return results, None

if __name__ == "__main__":
    for month in ("2026-08", "2026-09"):
        print(f"goal: email the {month} sales report")
        plan, _ = check_plan(planner("monthly sales report", month))
        results, error = execute(plan)
        if error:                                     # replan with the failure, don't push on
            print("  replanning")
            execute(check_plan(planner("monthly sales report", month, feedback=error))[0])
    print("outbox:", *OUTBOX, sep="\n  ")

$ python execute.py
goal: email the 2026-08 sales report
  step 1 load_sales -> [('north', 41200), ('south', 38950), ('west', 27400)]
  step 2 summarize -> {'total': 107550, 'best_region': 'north', 'regions': 3}
  step 3 write_report -> Sales 2026-08: total 107,550 across 3 regions; best region north.
  step 4 send_email -> sent to sales-team
goal: email the 2026-09 sales report
  step 1 load_sales FAILED: no sales export for 2026-09
  replanning
  step 1 send_email -> sent to sales-team
outbox:
  ('sales-team', 'Sales 2026-08: total 107,550 across 3 regions; best region north.')
  ('sales-team', 'The 2026-09 report is delayed: no sales export for 2026-09.')

August runs straight through: each $n argument is replaced by step n’s result, so summarize gets the loaded rows and send_email gets the finished report. September fails at step 1, because there’s no export. The executor does not push on to summarize nothing and email an empty report. It stops and gives the error to the planner, which produces a different plan: tell the team the report is delayed, and why. The outbox shows both outcomes.

Replanning with the error is what makes this an agent rather than a script. A fixed script can only crash or carry on regardless. With the failure in view, the model can pick a route the script’s author never wrote down.

Running Independent Steps in Parallel

Dependencies also say what doesn’t depend on what. A regional report loads three regions, merges them, and produces a chart and a text report before emailing both.

import time
from concurrent.futures import ThreadPoolExecutor
from graphlib import TopologicalSorter

AFTER = {"load_north": [], "load_south": [], "load_west": [],      # plan: step -> depends on
         "merge": ["load_north", "load_south", "load_west"],
         "chart": ["merge"], "report": ["merge"], "email": ["chart", "report"]}

def run(step):
    time.sleep(0.2)                                   # every step is a 0.2 s API call
    return step

if __name__ == "__main__":
    graph = TopologicalSorter(AFTER)
    graph.prepare()
    start = time.perf_counter()
    with ThreadPoolExecutor() as pool:
        while graph.is_active():
            ready = graph.get_ready()                     # everything whose inputs are done
            print(f"{time.perf_counter() - start:4.1f}s  running {sorted(ready)}")
            for done in pool.map(run, ready):
                graph.done(done)
    print(f"{time.perf_counter() - start:4.1f}s  finished; one at a time would take {0.2 * len(AFTER):.1f}s")

$ python batches.py
 0.0s  running ['load_north', 'load_south', 'load_west']
 0.2s  running ['merge']
 0.4s  running ['chart', 'report']
 0.6s  running ['email']
 0.8s  finished; one at a time would take 1.4s

get_ready() returns every step whose dependencies are finished. The three loads start together, then the merge, then the chart and report together, then the email. Seven 0.2-second steps finish in 0.8 seconds instead of 1.4, and the dependency graph guarantees the email never goes out before both attachments exist. This loop runs in rounds: pool.map waits for a whole batch before asking for the next, so with uneven step times a round waits for its slowest step. Submitting each step as soon as it is ready, with as_completed, removes that wait at the cost of a slightly longer loop. A plan written as a numbered list can’t be parallelized safely, which is one reason to ask the model for dependencies, not just an order.

Checking a Result Before Acting on It

Verify outputs against their source before a step with side effects, such as sending an email.

import re
from tools import load_sales, summarize, write_report

def verify_report(report, rows):
    """Check the report against the data before it is sent, not after."""
    problems = []
    total = sum(amount for _, amount in rows)
    stated = [int(n.replace(",", "")) for n in re.findall(r"total ([\d,]+)", report)]
    if stated != [total]:
        problems.append(f"total says {stated}, data says {total}")
    best = max(rows, key=lambda r: r[1])[0]
    if f"best region {best}" not in report:
        problems.append(f"best region should be {best}")
    return problems or ["ok"]

rows = load_sales("2026-08")
tool_report = write_report(summarize(rows), "2026-08")
model_draft = "Sales 2026-08: total 107,450 across 3 regions; best region south."  # a typical slip
for name, report in [("tool-built report", tool_report), ("model's own draft", model_draft)]:
    print(f"{name:18} {verify_report(report, rows)}")

$ python verify.py
tool-built report  ['ok']
model's own draft  ['total says [107450], data says 107550', 'best region should be north']

The report built by the tools passes: its total and best region were computed from the rows. The draft is the kind of report a model writes when it does the arithmetic itself, and it gets both facts wrong. The check recomputes from the same rows and catches both before anything is sent. An agent that runs this check between write_report and send_email can send the failure back to the planner, just as execute.py does with a failed tool.

The general rule is to let the model decide what to do, and let code compute and check the numbers. Verification costs a few lines, and a wrong figure in an email to the whole sales team costs far more.

Practice

1. Some tools should run at most once per plan: two send_email steps means two emails. Extend check_plan to reject a plan that uses send_email more than once.

   Solution

   import json
   from validate import check_plan
   
   def check_side_effects(plan_json, once=("send_email",)):
       plan, problems = check_plan(plan_json)
       if plan:
           for tool in once:
               count = sum(step["tool"] == tool for step in plan)
               if count > 1:
                   problems.append(f"{tool} appears {count} times; side effects must run once")
       return (None if problems else plan), problems
   
   steps = [{"id": 1, "tool": "write_report", "after": []},
            {"id": 2, "tool": "send_email", "after": [1]},
            {"id": 3, "tool": "send_email", "after": [1]}]
   print(check_side_effects(json.dumps(steps))[1])
   print(check_side_effects(json.dumps(steps[:2]))[1])

   $ python sol_one_email.py
   ['send_email appears 2 times; side effects must run once']
   []

   The duplicate-send plan is rejected with a reason the model can act on, and the single-send plan passes. Checks like this are cheap insurance on any tool with an external effect: payments, emails, deletions, deployments.

2. In batches.py, make load_south fail. Keep running steps that don’t depend on it, and skip, without running, everything downstream of the failure.

   Solution

   from graphlib import TopologicalSorter
   from batches import AFTER
   
   def run(step):
       if step == "load_south":
           raise TimeoutError("south region API timed out")
       return step
   
   graph = TopologicalSorter(AFTER)
   graph.prepare()
   failed, skipped = set(), set()
   while graph.is_active():
       for step in graph.get_ready():
           if any(dep in failed | skipped for dep in AFTER[step]):
               skipped.add(step)                     # an input is missing: don't run it
           else:
               try:
                   run(step)
               except Exception as e:
                   failed.add(step)
                   print(f"{step} failed: {e}")
           graph.done(step)
   print("failed:", sorted(failed), " skipped:", sorted(skipped))

   $ python sol_skip.py
   load_south failed: south region API timed out
   failed: ['load_south']  skipped: ['chart', 'email', 'merge', 'report']

   North and west still load, and everything that needed the south region is skipped rather than run with missing input. Skipping propagates: email is skipped because chart and report were, even though it never touched south directly. The failed and skipped lists are what the planner needs to decide whether to retry, report partial results, or tell someone.

3. A planner can keep proposing the same broken plan. Cap replanning at three attempts, then escalate to a person with the last error.

   Solution

   from execute import execute
   from validate import check_plan
   import json
   
   def stubborn_planner(feedback=None):
       """A model that keeps proposing the same broken plan."""
       return json.dumps([{"id": 1, "tool": "load_sales", "args": {"month": "2026-09"}, "after": []}])
   
   feedback, attempts = None, 0
   while attempts < 3:
       attempts += 1
       results, feedback = execute(check_plan(stubborn_planner(feedback))[0])
       if feedback is None:
           break
   else:
       print(f"gave up after {attempts} plans; escalating to a person with: {feedback}")

   $ python sol_replan_cap.py
     step 1 load_sales FAILED: no sales export for 2026-09
     step 1 load_sales FAILED: no sales export for 2026-09
     step 1 load_sales FAILED: no sales export for 2026-09
   gave up after 3 plans; escalating to a person with: no sales export for 2026-09

   Three identical failures, then a clear hand-off with the reason. Python’s while ... else runs the else only when the loop ends without break, which here means that no plan succeeded. Without the cap, an agent with a stubborn planner loops until something times out, spending tokens on each attempt.