Planning › Implementation
Language AI HandbookAll chapters Planning › ImplementationOne section · Sections in this chapter7 / 12 ↓
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
We will build a working planning agent in Python that demonstrates task decomposition, plan representation, sequential execution with monitoring, and replanning on failure. The implementation uses simulated tools that occasionally fail to demonstrate recovery logic without requiring external API calls.
The architecture has four layers: data structures that represent plans and steps, tool implementations that do the actual work, an execution engine that dispatches steps in dependency order, and a recovery layer that handles failures. Each layer can be inspected and modified independently, which mirrors the separation of concerns in production planning agents.
Setup and Data Structures
import random
random.seed(42)We define the data structures that represent a plan and its execution state. The design prioritizes clarity over performance: every field is explicitly named, and status transitions are tracked as an enum rather than as boolean flags. This makes the execution logic easier to reason about and debug.
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class StepStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class PlanStep:
step_id: int
description: str
tool: str
tool_input: str
depends_on: list = field(default_factory=list)
status: StepStatus = StepStatus.PENDING
output: Optional[str] = None
error: Optional[str] = None
attempts: int = 0
@dataclass
class Plan:
goal: str
steps: list = field(default_factory=list)
context: dict = field(default_factory=dict)
def get_step(self, step_id: int) -> Optional[PlanStep]:
for step in self.steps:
if step.step_id == step_id:
return step
return None
def pending_steps(self) -> list:
return [s for s in self.steps if s.status == StepStatus.PENDING]
def completed_steps(self) -> list:
return [s for s in self.steps if s.status == StepStatus.COMPLETED]
def failed_steps(self) -> list:
return [s for s in self.steps if s.status == StepStatus.FAILED]
def is_complete(self) -> bool:
return all(
s.status in (StepStatus.COMPLETED, StepStatus.SKIPPED)
for s in self.steps
)The PlanStep dataclass encodes everything the executor needs for a single step: the tool to call, the input to pass it, the steps it depends on, its current status, and its output once it completes. The depends_on list is the key field for dependency management. The Plan class wraps a list of steps and provides convenience methods for querying subsets of them by status. Note that is_complete treats SKIPPED as a terminal state alongside COMPLETED, which allows the planner to mark unreachable downstream steps as skipped after a plan-invalidating failure rather than leaving them permanently pending.
Tool Definitions
Each tool has a small probability of failing to demonstrate recovery logic. In a real system, these would be actual API calls or code execution environments.
# Simulated tool implementations.
# fail_prob controls how often a tool raises a transient error.
def search_tool(query: str, fail_prob: float = 0.15) -> dict:
if random.random() < fail_prob:
raise RuntimeError(f"Search API timeout for query: '{query}'")
results = {
"climate change temperature": "Global mean temperature has risen 1.1 C above pre-industrial levels.",
"renewable energy capacity": "Global renewable capacity reached 3,372 GW in 2023, up 295 GW from 2022.",
"carbon emissions 2023": "Global CO2 emissions hit 36.8 billion tonnes in 2023, a record high.",
"sea level rise rate": "Sea levels are rising at 3.7 mm per year on average since 1993.",
}
for key, value in results.items():
if any(word in query.lower() for word in key.split()):
return {"query": query, "result": value, "source": "simulated_db"}
return {
"query": query,
"result": f"No specific data found for: {query}",
"source": "simulated_db",
}
def compute_tool(expression: str, fail_prob: float = 0.05) -> dict:
if random.random() < fail_prob:
raise RuntimeError(f"Compute service unavailable for: '{expression}'")
# Parse simple arithmetic expressions manually to avoid security concerns
import ast
import operator
ops = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
}
def ast_compute(node):
if isinstance(node, ast.Constant):
return node.n
if isinstance(node, ast.BinOp) and type(node.op) in ops:
return ops[type(node.op)](
ast_compute(node.left), ast_compute(node.right)
)
raise ValueError(f"Unsupported expression: {ast.dump(node)}")
try:
tree = ast.parse(expression, mode="eval")
result = ast_compute(tree.body)
return {"expression": expression, "result": result}
except Exception as exc:
raise RuntimeError(f"Computation failed: {exc}") from exc
def summarize_tool(text: str, fail_prob: float = 0.05) -> dict:
if random.random() < fail_prob:
raise RuntimeError("Summarization model unavailable")
words = text.split()
summary = " ".join(words[: min(20, len(words))]) + (
"..." if len(words) > 20 else ""
)
return {"original_length": len(words), "summary": summary}
TOOLS = {
"search": search_tool,
"compute": compute_tool,
"summarize": summarize_tool,
}The tools dictionary maps string names to callable functions. The execution engine uses string lookup at runtime rather than direct function references, which mirrors how real tool-calling agents work: the planner specifies tools by name in the plan, and the executor resolves names to implementations when each step runs.
Plan Generation
The planner takes a goal and generates a structured plan. In a real system, this function calls an LLM with a planning prompt. Here we hardcode two example plans to focus on demonstrating execution and recovery logic without requiring an API connection.
def generate_plan(goal: str) -> Plan:
"""
Generate a structured plan for the given goal.
Returns a Plan with PlanStep objects encoding dependencies.
"""
plan = Plan(goal=goal)
if "climate" in goal.lower():
plan.steps = [
PlanStep(
step_id=1,
description="Search for current global temperature anomaly data",
tool="search",
tool_input="climate change temperature",
depends_on=[],
),
PlanStep(
step_id=2,
description="Search for renewable energy capacity data",
tool="search",
tool_input="renewable energy capacity",
depends_on=[],
),
PlanStep(
step_id=3,
description="Search for current CO2 emissions data",
tool="search",
tool_input="carbon emissions 2023",
depends_on=[],
),
PlanStep(
step_id=4,
description="Search for sea level rise data",
tool="search",
tool_input="sea level rise rate",
depends_on=[],
),
PlanStep(
step_id=5,
description="Summarize all gathered climate data into a brief report",
tool="summarize",
tool_input="AGGREGATE", # Filled at runtime from steps 1-4
depends_on=[1, 2, 3, 4],
),
]
else:
plan.steps = [
PlanStep(
step_id=1,
description=f"Search for information about: {goal}",
tool="search",
tool_input=goal[:50],
depends_on=[],
),
PlanStep(
step_id=2,
description="Summarize the gathered information",
tool="summarize",
tool_input="AGGREGATE",
depends_on=[1],
),
]
return planExecution Engine
The executor runs each step in dependency order, handling retries for transient failures and storing outputs for downstream steps.
import time
MAX_RETRIES = 2
RETRY_DELAY = 0.1 # seconds
def can_execute(step: PlanStep, plan: Plan) -> bool:
"""Return True if all dependencies for this step have completed."""
for dep_id in step.depends_on:
dep = plan.get_step(dep_id)
if dep is None or dep.status != StepStatus.COMPLETED:
return False
return True
def build_tool_input(step: PlanStep, plan: Plan) -> str:
"""
For steps with tool_input == 'AGGREGATE', concatenate the outputs
of all dependency steps as the tool input.
"""
if step.tool_input != "AGGREGATE":
return step.tool_input
parts = []
for dep_id in step.depends_on:
dep = plan.get_step(dep_id)
if dep and dep.output:
parts.append(dep.output)
return " | ".join(parts)
def execute_step(step: PlanStep, plan: Plan) -> bool:
"""
Execute a single plan step. Retries up to MAX_RETRIES times on failure.
Returns True on success, False if all retries are exhausted.
"""
tool_fn = TOOLS.get(step.tool)
if tool_fn is None:
step.status = StepStatus.FAILED
step.error = f"Unknown tool: {step.tool}"
return False
tool_input = build_tool_input(step, plan)
step.status = StepStatus.IN_PROGRESS
for attempt in range(1, MAX_RETRIES + 2):
step.attempts = attempt
try:
result = tool_fn(tool_input)
output_val = result.get("result") or result.get("summary", "")
step.output = str(output_val)
step.status = StepStatus.COMPLETED
return True
except RuntimeError as exc:
step.error = str(exc)
if attempt <= MAX_RETRIES:
time.sleep(RETRY_DELAY)
else:
step.status = StepStatus.FAILED
return False
step.status = StepStatus.FAILED
return False
def execute_plan(plan: Plan, verbose: bool = False) -> list:
"""
Execute all steps in dependency order.
Returns an execution log (list of dicts, one per step executed).
"""
execution_log = []
max_rounds = len(plan.steps) * 2 # Safety limit
for _ in range(max_rounds):
ready = [s for s in plan.pending_steps() if can_execute(s, plan)]
if not ready:
break
for step in ready:
success = execute_step(step, plan)
execution_log.append(
{
"step_id": step.step_id,
"description": step.description,
"tool": step.tool,
"status": step.status.value,
"attempts": step.attempts,
"output_preview": step.output[:80] if step.output else None,
"error": step.error,
}
)
if verbose:
status_label = "COMPLETED" if success else "FAILED"
attempts_label = f" ({step.attempts} attempt{'s' if step.attempts > 1 else ''})"
detail = step.output[:60] if success else step.error
print(
f" Step {step.step_id} [{status_label}{attempts_label}]: {detail}"
)
if plan.is_complete():
break
return execution_logThe execute_plan function implements a simple round-robin scheduler: at each iteration, it finds all pending steps whose dependencies have all completed and dispatches them. In a real parallel executor, each ready step would be dispatched to a separate worker and the loop would wait on futures. In this sequential implementation, ready steps within a single round are executed one after another, but across rounds the dependency structure is respected.
The build_tool_input function handles the AGGREGATE sentinel value, which signals that the step's input should be constructed by concatenating all dependency outputs. This is how the summarization step in the climate plan receives the combined results of the four preceding search steps.
Replanning Logic
When steps fail, the agent first attempts local repair by modifying the tool input. If that fails, it escalates to suffix replanning.
def local_repair(failed_step: PlanStep, plan: Plan) -> bool:
"""
Attempt to repair a single failed step by modifying its tool input.
In a real system this calls the LLM with failure context.
Returns True if a repair was applied.
"""
if failed_step.tool == "search":
# Broaden the query as a fallback
failed_step.tool_input = failed_step.tool_input + " overview"
failed_step.status = StepStatus.PENDING
failed_step.error = None
failed_step.attempts = 0
return True
return False
def suffix_replan(plan: Plan) -> Plan:
"""
Suffix replanning: given completed steps and their outputs, generate
a continuation plan for the remaining subgoal.
Returns a new Plan that extends the completed steps.
"""
completed_outputs = [
f"Step {s.step_id}: {s.output}"
for s in plan.completed_steps()
if s.output
]
new_plan = Plan(goal=plan.goal)
new_plan.steps = [s for s in plan.steps if s.status == StepStatus.COMPLETED]
max_existing_id = max((s.step_id for s in new_plan.steps), default=0)
aggregated_input = (
" | ".join(completed_outputs) if completed_outputs else plan.goal
)
new_plan.steps.append(
PlanStep(
step_id=max_existing_id + 1,
description="Synthesize available data into partial report",
tool="summarize",
tool_input=aggregated_input,
depends_on=[s.step_id for s in new_plan.steps],
)
)
new_plan.context["replanned"] = True
return new_plan
def execute_with_recovery(goal: str) -> dict:
"""
Full pipeline: generate plan, execute, recover from failures.
Returns a summary dict with counts and outputs.
"""
plan = generate_plan(goal)
all_logs = execute_plan(plan, verbose=False)
# Local repair pass
failed = plan.failed_steps()
if failed:
for step in failed:
local_repair(step, plan)
logs2 = execute_plan(plan, verbose=False)
all_logs.extend(logs2)
# Suffix replanning if still failing
still_failed = plan.failed_steps()
if still_failed:
plan = suffix_replan(plan)
logs3 = execute_plan(plan, verbose=False)
all_logs.extend(logs3)
return {
"goal": goal,
"plan_steps": len(plan.steps),
"completed": len(plan.completed_steps()),
"failed": len(plan.failed_steps()),
"replanned": plan.context.get("replanned", False),
"final_outputs": {
s.step_id: s.output for s in plan.completed_steps() if s.output
},
"execution_log": all_logs,
}Running the Planner
Let's execute the planner on a climate research task and inspect the results.
result = execute_with_recovery(
goal="Research the current state of climate change and prepare a summary report"
)Goal: Research the current state of climate change and prepare a s... Plan steps: 5 Completed: 5 Failed: 0 Replanning used: False Final outputs (5 steps produced output): Step 1: Global mean temperature has risen 1.1 C above pre-industrial levels. Step 2: Global renewable capacity reached 3,372 GW in 2023, up 295 GW from 202... Step 3: Global CO2 emissions hit 36.8 billion tonnes in 2023, a record high. Step 4: Sea levels are rising at 3.7 mm per year on average since 1993. Step 5: Global mean temperature has risen 1.1 C above pre-industrial levels. |...
The output shows the plan's final state: how many steps completed, whether replanning was triggered, and what each step produced. The dependency structure ensures step 5 (summarization) only runs after all four search steps have produced their outputs, even though the four search steps run in sequence in this single-threaded implementation.
Tracking Planning Metrics
To understand aggregate planner performance, we run the pipeline across many trials and collect statistics.
def run_planning_experiment(n_trials: int = 30) -> dict:
"""Run the planning pipeline n_trials times and return aggregate statistics."""
first_pass_rates = []
total_attempts_per_run = []
failures_per_run = []
for _ in range(n_trials):
plan = generate_plan(
"Research the current state of climate change and prepare a summary report"
)
execute_plan(plan, verbose=False)
total_steps = len(plan.steps)
first_pass = sum(
1
for s in plan.steps
if s.status == StepStatus.COMPLETED and s.attempts == 1
)
total_attempts = sum(s.attempts for s in plan.steps)
failed_count = sum(
1 for s in plan.steps if s.status == StepStatus.FAILED
)
first_pass_rates.append(
first_pass / total_steps if total_steps else 0.0
)
total_attempts_per_run.append(total_attempts)
failures_per_run.append(failed_count)
n = n_trials
return {
"n_trials": n,
"avg_first_pass_rate": sum(first_pass_rates) / n,
"avg_attempts_per_run": sum(total_attempts_per_run) / n,
"avg_failures_per_run": sum(failures_per_run) / n,
"raw_first_pass_rates": first_pass_rates,
"raw_total_attempts": total_attempts_per_run,
}
experiment_results = run_planning_experiment(n_trials=30)Planning Experiment Results (30 trials, 5-step climate plan) Avg first-pass completion rate: 86.0% Avg tool call attempts per run: 5.7 Avg failures per run: 0.03
Most runs complete with every step succeeding on the first attempt. Occasionally a search step times out and requires a retry, adding one or two extra attempts. The summarization step (step 5) very rarely fails because it only runs after the four search steps have completed and its input is always a non-empty concatenation of their outputs. This illustrates an important design principle: dependent steps tend to have higher success rates because their inputs are pre-validated by the successful completion of their predecessor steps. A well-structured dependency graph naturally concentrates failure risk in the early, independent steps and shields later synthesis steps from missing-input failures.
Key Parameters
The key parameters controlling planning agent behavior are:
- MAX_RETRIES: Number of times to retry a failed tool call before marking the step as failed. Higher values improve resilience to transient failures but increase latency.
- depends_on (step parameter): The list of step IDs that must complete before this step executes. Correctly specifying dependencies is the most important factor in plan correctness.
- fail_prob (tool parameter): The probability that any individual tool call raises a transient error. In production this is determined by infrastructure reliability.
- AGGREGATE sentinel: The special
tool_inputvalue that signals the execution engine to concatenate the outputs of all dependency steps before passing them to the tool.