The Planning Pattern: Strategic AI Agent Design
Learn how the planning pattern enables agents to tackle complex tasks through strategic decomposition, goal setting, and adaptive execution.
The Planning Pattern: Strategic AI Agent Design
Some tasks are too complex for step-by-step reactive execution. They require stepping back, understanding the full scope, breaking the problem into parts, and creating a coherent strategy before taking action.
The planning pattern enables this. It transforms agents from reactive responders into strategic thinkers that can tackle multi-step, complex tasks with greater success and efficiency.
What is the Planning Pattern?
The planning pattern separates thinking from doing:
- Analysis: Understand the task completely
- Decomposition: Break into subtasks
- Planning: Create an execution strategy
- Execution: Carry out the plan
- Adaptation: Adjust when things change
Task → Analyze → Plan → Execute Step 1 → Execute Step 2 → ... → Complete
↑ |
└──────── Replan if needed ←──────────────────┘
Unlike ReAct (which thinks and acts in tight loops), planning creates a comprehensive strategy upfront—though it can adapt as needed.
Why Planning Matters
Complex Task Success
Research shows planning improves success rates on complex tasks by 20-40%. Without a plan, agents often:
- Miss important steps
- Execute in wrong order
- Get stuck in unproductive loops
- Forget the overall goal
Efficiency
A good plan reduces wasted effort:
- Identifies parallelizable work
- Avoids redundant actions
- Allocates resources appropriately
- Prevents dead ends
Transparency
Plans are explainable:
- Users can review before execution
- Progress is trackable
- Failures are diagnosable
- Modifications are possible
Basic Planning Implementation
Here's a foundational planning agent:
from anthropic import Anthropic
from dataclasses import dataclass
from typing import List, Optional
import json
@dataclass
class PlanStep:
id: str
description: str
dependencies: List[str]
status: str = "pending" # pending, in_progress, completed, failed
result: Optional[str] = None
@dataclass
class Plan:
goal: str
steps: List[PlanStep]
current_step: int = 0
class PlanningAgent:
def __init__(self):
self.client = Anthropic()
self.plan: Optional[Plan] = None
def create_plan(self, task: str) -> Plan:
"""Generate a plan for the task"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[{
"role": "user",
"content": f"""Create a detailed plan to accomplish this task:
Task: {task}
For each step, provide:
- id: unique identifier (step_1, step_2, etc.)
- description: what needs to be done
- dependencies: list of step ids that must complete first
Output as JSON:
{{
"goal": "the overall goal",
"steps": [
{{"id": "step_1", "description": "...", "dependencies": []}},
{{"id": "step_2", "description": "...", "dependencies": ["step_1"]}}
]
}}
Keep steps atomic and actionable. Include 3-10 steps depending on complexity."""
}]
)
plan_data = json.loads(response.content[0].text)
steps = [
PlanStep(
id=s["id"],
description=s["description"],
dependencies=s.get("dependencies", [])
)
for s in plan_data["steps"]
]
self.plan = Plan(goal=plan_data["goal"], steps=steps)
return self.plan
def execute_plan(self) -> str:
"""Execute the plan step by step"""
if not self.plan:
raise ValueError("No plan created")
completed_steps = {}
while not self._is_complete():
# Find next executable step
step = self._get_next_step(completed_steps)
if not step:
break
step.status = "in_progress"
# Execute the step
result = self._execute_step(step, completed_steps)
if result.get("success"):
step.status = "completed"
step.result = result.get("output")
completed_steps[step.id] = step.result
else:
step.status = "failed"
# Attempt replanning
if not self._handle_failure(step, result.get("error")):
return f"Plan failed at step: {step.description}"
return self._summarize_results(completed_steps)
def _get_next_step(self, completed: dict) -> Optional[PlanStep]:
"""Find next step whose dependencies are satisfied"""
for step in self.plan.steps:
if step.status != "pending":
continue
deps_satisfied = all(d in completed for d in step.dependencies)
if deps_satisfied:
return step
return None
def _is_complete(self) -> bool:
"""Check if all steps are done"""
return all(s.status == "completed" for s in self.plan.steps)
def _execute_step(self, step: PlanStep, context: dict) -> dict:
"""Execute a single step"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[{
"role": "user",
"content": f"""Execute this step:
Step: {step.description}
Context from previous steps:
{json.dumps(context, indent=2)}
Overall goal: {self.plan.goal}
Provide the result of this step."""
}]
)
return {
"success": True,
"output": response.content[0].text
}
def _handle_failure(self, step: PlanStep, error: str) -> bool:
"""Try to recover from a failed step"""
# Could implement replanning logic here
return False
def _summarize_results(self, results: dict) -> str:
"""Create final summary"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"""Summarize the results of this completed plan:
Goal: {self.plan.goal}
Step results:
{json.dumps(results, indent=2)}
Provide a concise summary of what was accomplished."""
}]
)
return response.content[0].text
Planning Strategies
Strategy 1: Hierarchical Planning
Break complex plans into nested sub-plans:
class HierarchicalPlanner:
def __init__(self, llm):
self.llm = llm
def create_hierarchical_plan(self, task: str, depth: int = 2) -> dict:
"""Create a multi-level plan"""
# Level 1: High-level phases
high_level = self._plan_level(task, "phases")
if depth > 1:
# Level 2: Steps within each phase
for phase in high_level["phases"]:
phase["steps"] = self._plan_level(
phase["description"],
"steps"
)["steps"]
if depth > 2:
# Level 3: Actions within each step
for phase in high_level["phases"]:
for step in phase["steps"]:
step["actions"] = self._plan_level(
step["description"],
"actions"
)["actions"]
return high_level
def _plan_level(self, task: str, level_name: str) -> dict:
response = self.llm.generate(f"""
Break down this task into {level_name}:
{task}
Return as JSON with "{level_name}" array.
""")
return json.loads(response)
Strategy 2: Goal-Oriented Planning
Plan backward from the goal:
class GoalOrientedPlanner:
def __init__(self, llm):
self.llm = llm
def plan_backward(self, goal: str) -> list:
"""Plan by working backward from the goal"""
steps = []
current_goal = goal
while True:
# What's needed to achieve current goal?
prerequisite = self.llm.generate(f"""
Goal to achieve: {current_goal}
What is the immediate prerequisite for this goal?
If no prerequisite (this is the starting point), say "START".
Respond with just the prerequisite description or "START".
""")
if "START" in prerequisite:
break
steps.insert(0, {
"goal": current_goal,
"prerequisite": prerequisite
})
current_goal = prerequisite
return steps
Strategy 3: Constraint-Based Planning
Plan while respecting constraints:
class ConstraintPlanner:
def __init__(self, llm):
self.llm = llm
def plan_with_constraints(
self,
task: str,
constraints: list[str]
) -> dict:
"""Create a plan that respects all constraints"""
# Generate initial plan
plan = self._generate_plan(task)
# Validate against constraints
violations = self._check_constraints(plan, constraints)
# Iterate until valid
max_iterations = 3
for _ in range(max_iterations):
if not violations:
break
plan = self._fix_violations(plan, violations, constraints)
violations = self._check_constraints(plan, constraints)
return {
"plan": plan,
"constraints_satisfied": len(violations) == 0,
"remaining_violations": violations
}
def _check_constraints(self, plan: dict, constraints: list) -> list:
"""Check which constraints are violated"""
violations = []
for constraint in constraints:
response = self.llm.generate(f"""
Plan: {json.dumps(plan)}
Constraint: {constraint}
Does this plan violate this constraint?
Respond with YES or NO and explain.
""")
if "YES" in response.upper():
violations.append({
"constraint": constraint,
"explanation": response
})
return violations
Strategy 4: Resource-Aware Planning
Plan considering available resources:
class ResourceAwarePlanner:
def __init__(self, llm, resources: dict):
self.llm = llm
self.resources = resources # e.g., {"time": 60, "api_calls": 100}
def plan_within_resources(self, task: str) -> dict:
"""Create a plan that fits within resource limits"""
# Estimate resource needs for each potential step
initial_plan = self._generate_plan(task)
for step in initial_plan["steps"]:
step["resource_estimates"] = self._estimate_resources(step)
# Calculate total resource usage
total_usage = self._calculate_total_usage(initial_plan)
# Check if within limits
if self._within_limits(total_usage):
return initial_plan
# Trim or optimize plan
return self._optimize_for_resources(initial_plan)
def _estimate_resources(self, step: dict) -> dict:
response = self.llm.generate(f"""
Step: {step['description']}
Estimate resource usage:
- time_minutes: how long this will take
- api_calls: how many API calls needed
Respond as JSON.
""")
return json.loads(response)
def _optimize_for_resources(self, plan: dict) -> dict:
"""Reduce plan to fit within resources"""
response = self.llm.generate(f"""
This plan exceeds resource limits:
Plan: {json.dumps(plan)}
Limits: {json.dumps(self.resources)}
Modify the plan to fit within limits while still achieving
the core goal. Remove or combine steps as needed.
Return the modified plan as JSON.
""")
return json.loads(response)
Adaptive Replanning
Plans rarely survive contact with reality. Adaptive replanning handles changes:
class AdaptivePlanner:
def __init__(self, llm):
self.llm = llm
self.original_plan = None
self.current_plan = None
self.execution_history = []
def execute_with_adaptation(self, task: str) -> str:
"""Execute plan with dynamic adaptation"""
self.original_plan = self._create_plan(task)
self.current_plan = self.original_plan.copy()
while not self._is_complete():
step = self._get_next_step()
# Execute step
result = self._execute_step(step)
self.execution_history.append({
"step": step,
"result": result
})
# Check if replanning needed
if self._needs_replanning(result):
self.current_plan = self._replan(
self.original_plan["goal"],
self.execution_history
)
return self._summarize()
def _needs_replanning(self, result: dict) -> bool:
"""Determine if we need to adapt the plan"""
response = self.llm.generate(f"""
Original plan: {json.dumps(self.current_plan)}
Latest result: {json.dumps(result)}
Execution history: {json.dumps(self.execution_history)}
Does this result require changing the plan?
Consider:
1. Did the step fail?
2. Did we learn something that changes the approach?
3. Are remaining steps still relevant?
Respond YES or NO with brief explanation.
""")
return "YES" in response.upper()
def _replan(self, goal: str, history: list) -> dict:
"""Create a new plan based on what we've learned"""
response = self.llm.generate(f"""
Original goal: {goal}
What we've done so far:
{json.dumps(history)}
Create a new plan to complete the goal from this point.
Build on what's been accomplished and adjust for what we've learned.
Return as JSON with remaining steps.
""")
return json.loads(response)
Plan Visualization
Help users understand and monitor plans:
class PlanVisualizer:
def to_text(self, plan: dict) -> str:
"""Convert plan to readable text format"""
output = [f"Goal: {plan['goal']}\n"]
for i, step in enumerate(plan["steps"], 1):
status_icon = {
"pending": "○",
"in_progress": "◐",
"completed": "●",
"failed": "✗"
}.get(step.get("status", "pending"), "○")
deps = f" (after: {', '.join(step['dependencies'])})" if step.get("dependencies") else ""
output.append(f"{status_icon} Step {i}: {step['description']}{deps}")
return "\n".join(output)
def to_mermaid(self, plan: dict) -> str:
"""Convert plan to Mermaid diagram"""
lines = ["graph TD"]
for step in plan["steps"]:
# Add node
lines.append(f' {step["id"]}["{step["description"][:30]}..."]')
# Add edges for dependencies
for dep in step.get("dependencies", []):
lines.append(f" {dep} --> {step['id']}")
return "\n".join(lines)
def progress_bar(self, plan: dict) -> str:
"""Show plan progress"""
total = len(plan["steps"])
completed = sum(1 for s in plan["steps"] if s.get("status") == "completed")
percentage = int(completed / total * 100)
bar_length = 20
filled = int(bar_length * completed / total)
bar = "█" * filled + "░" * (bar_length - filled)
return f"Progress: [{bar}] {percentage}% ({completed}/{total} steps)"
Planning for Parallel Execution
Identify and execute independent steps in parallel:
class ParallelPlanner:
def __init__(self, llm, max_parallel: int = 3):
self.llm = llm
self.max_parallel = max_parallel
def create_parallel_plan(self, task: str) -> dict:
"""Create a plan with parallel execution batches"""
# Get all steps with dependencies
plan = self._create_plan(task)
# Group into execution batches
batches = self._create_batches(plan["steps"])
plan["batches"] = batches
return plan
def _create_batches(self, steps: list) -> list:
"""Group steps into parallel execution batches"""
batches = []
completed = set()
remaining = {s["id"]: s for s in steps}
while remaining:
# Find all steps whose dependencies are complete
batch = []
for step_id, step in list(remaining.items()):
deps = set(step.get("dependencies", []))
if deps.issubset(completed):
batch.append(step)
if len(batch) >= self.max_parallel:
break
if not batch:
raise ValueError("Circular dependency detected")
batches.append(batch)
# Mark as complete for next iteration
for step in batch:
completed.add(step["id"])
del remaining[step["id"]]
return batches
async def execute_parallel(self, plan: dict) -> dict:
"""Execute plan with parallel batches"""
results = {}
for batch in plan["batches"]:
# Execute batch in parallel
batch_results = await asyncio.gather(*[
self._execute_step(step, results)
for step in batch
])
# Store results
for step, result in zip(batch, batch_results):
results[step["id"]] = result
return results
When to Use Planning
Use Planning When:
- Task has multiple interdependent steps
- Order of operations matters
- Resources need allocation
- Progress tracking is important
- User review before execution is valuable
Skip Planning When:
- Task is simple and straightforward
- Speed is critical (planning adds latency)
- Task is highly dynamic (plan becomes obsolete quickly)
- Single-step execution suffices
Conclusion
The planning pattern transforms agents from reactive responders into strategic actors. By analyzing tasks, decomposing them into steps, and creating coherent execution strategies, agents can tackle complex challenges more reliably and efficiently.
Key principles:
- Analyze before acting
- Decompose complex tasks
- Identify dependencies and parallelize where possible
- Adapt when reality differs from the plan
- Make plans visible and trackable
Planning isn't overhead—it's an investment that pays dividends in reliability, efficiency, and transparency.
Ready to coordinate multiple planning agents? Check out The Multi-Agent Pattern for orchestrating agent teams.