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:

1. Analysis: Understand the task completely
2. Decomposition: Break into subtasks
3. Planning: Create an execution strategy
4. Execution: Carry out the plan
5. 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:

1. Analyze before acting
2. Decompose complex tasks
3. Identify dependencies and parallelize where possible
4. Adapt when reality differs from the plan
5. 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.