Agent Planning & Task Decomposition

Why Impulsive Agents Fail at Complex Tasks

Think about how you tackle a really complex project, say, planning a product launch. You don’t just start randomly doing things. You step back, break the project into phases, identify which tasks depend on which, then execute in a logical order.

Agents that act “impulsively”, deciding the next action only based on the last observation, struggle with complex tasks for the same reason: they lose the big picture. They might go down productive-looking paths that don’t contribute to the goal, repeat work, miss dependencies, or optimise locally (this step) while failing globally (the overall task).

Planning before execution fixes this. Separate the agent into two modes:

1. Planner: think carefully about the goal, break it into concrete steps, identify dependencies
2. Executor: follow the plan, step by step, with the ability to flag problems

Plan-and-Execute: The Pattern

The plan-and-execute pattern works like this:

User gives goal
       ↓
PLANNER creates a numbered step-by-step plan
       ↓
EXECUTOR runs step 1, gets result
       ↓
EXECUTOR runs step 2, gets result
       ↓
... (repeat for all steps)
       ↓
SYNTHESISER assembles final answer from all results

The plan is created before any tools are called. It serves as the “spec” for the execution phase.

Implementing the Planner

The planner is a specialised LLM call that produces a structured plan:

from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

class PlanStep(BaseModel):
    step_num:    int
    description: str
    tool:        str            # which tool to use (or 'none')
    inputs:      dict           # what inputs to provide
    depends_on:  list[int]      # which step numbers must complete first

class ExecutionPlan(BaseModel):
    goal:    str
    steps:   list[PlanStep]
    rationale: str   # why this approach

def create_plan(goal: str, available_tools: list[str]) -> ExecutionPlan:
    """Ask the LLM to create a detailed execution plan."""
    
    response = client.beta.chat.completions.parse(
        model='gpt-5.6-sol',
        messages=[{
            'role': 'system',
            'content': f"""You are a planning specialist. Break goals into concrete steps.
Available tools: {', '.join(available_tools)}
Create specific, executable steps. Each step should use exactly one tool or be a reasoning step."""
        }, {
            'role': 'user',
            'content': f'Create a detailed execution plan for: {goal}'
        }],
        response_format=ExecutionPlan,
        temperature=0.1,
    )
    
    return response.choices[0].message.parsed

Executing the Plan

The executor runs each step, handling tools and passing results forward:

def execute_plan(plan: ExecutionPlan) -> dict:
    """Execute each step in the plan, in order of dependencies."""
    results = {}   # step_num → result
    
    for step in sorted(plan.steps, key=lambda s: s.step_num):
        print(f"\nStep {step.step_num}: {step.description}")
        
        # Wait for dependencies
        for dep in step.depends_on:
            if dep not in results:
                raise RuntimeError(f"Step {dep} not completed before step {step.step_num}")

Now execute the step: either call a tool or use the LLM to reason:

        if step.tool == 'none':
            # This is a reasoning/synthesis step
            # Build context from dependency results
            dep_context = '\n'.join([
                f"Step {d} result: {results[d]}"
                for d in step.depends_on
            ])
            
            response = client.chat.completions.create(
                model='gpt-5.6-sol',
                messages=[{
                    'role': 'user',
                    'content': f"""Complete this step: {step.description}
                    
Context from prior steps:
{dep_context}"""
                }],
            )
            result = response.choices[0].message.content
        
        else:
            # Execute the specified tool
            tool_func = TOOL_REGISTRY.get(step.tool)
            if not tool_func:
                result = f"Error: tool '{step.tool}' not found"
            else:
                try:
                    result = tool_func(**step.inputs)
                    result = str(result)
                except Exception as e:
                    result = f"Error: {e}"
        
        results[step.step_num] = result
        print(f"  Result: {str(result)[:200]}...")
    
    return results

Parallel Execution with a Task Graph

When steps have depends_on: [] (no dependencies), they can run in parallel. Let’s implement that:

import concurrent.futures
from collections import defaultdict

def execute_plan_parallel(plan: ExecutionPlan) -> dict:
    """Execute independent steps in parallel."""
    results = {}
    completed = set()
    
    steps_by_num = {s.step_num: s for s in plan.steps}
    
    def step_ready(step: PlanStep) -> bool:
        return all(dep in completed for dep in step.depends_on)
    
    remaining = list(plan.steps)
    
    while remaining:
        # Find all steps that are ready to run
        ready = [s for s in remaining if step_ready(s)]
        
        if not ready:
            # No progress possible: circular dependency or error
            raise RuntimeError("No executable steps found, possible circular dependency")
        
        # Remove ready steps from remaining
        remaining = [s for s in remaining if s not in ready]
        
        # Execute all ready steps in parallel
        with concurrent.futures.ThreadPoolExecutor() as executor:
            future_to_step = {
                executor.submit(execute_single_step, step, results): step
                for step in ready
            }
            for future in concurrent.futures.as_completed(future_to_step):
                step = future_to_step[future]
                results[step.step_num] = future.result()
                completed.add(step.step_num)
    
    return results

Replanning When Things Go Wrong

The original plan is based on assumptions. When those assumptions break, the agent needs to replan:

def execute_with_replanning(goal: str) -> str:
    """Execute a plan, replanning if steps fail."""
    
    available_tools = list(TOOL_REGISTRY.keys())
    plan = create_plan(goal, available_tools)
    
    print(f"Plan created: {len(plan.steps)} steps")
    
    results = {}
    for step in sorted(plan.steps, key=lambda s: s.step_num):
        result = execute_single_step(step, results)
        
        # Check if this step failed significantly
        if isinstance(result, str) and result.startswith('Error:'):
            print(f"Step {step.step_num} failed: {result}")
            print("Replanning remaining steps...")
            
            # What's left to do?
            completed_steps = [plan.steps[i] for i in results.keys()]
            remaining_goal = f"""Original goal: {goal}
            
Completed steps so far:
{chr(10).join(f'Step {i}: {results[i][:200]}' for i in sorted(results.keys()))}

Failed step: {step.description}
Error: {result}

What's the best way to proceed?"""
            
            # Generate a new partial plan for the remaining work
            new_plan = create_plan(remaining_goal, available_tools)
            plan = new_plan   # replace with new plan
            break
        
        results[step.step_num] = result
    
    return str(results)

Putting It Together: A Research Pipeline

def research_and_report(topic: str) -> str:
    goal = f"""Research '{topic}' and produce a structured report with:
    1. Overview (what it is)
    2. Current state / recent developments
    3. Key players / organisations
    4. Future outlook
    5. Recommendations for someone interested in this field"""
    
    # Plan
    plan = create_plan(goal, ['search_web', 'read_webpage'])
    print(f"\nPlan ({len(plan.steps)} steps):")
    for step in plan.steps:
        deps = f" (after steps {step.depends_on})" if step.depends_on else ""
        print(f"  {step.step_num}. {step.description}{deps}")
    
    # Execute
    results = execute_plan_parallel(plan)
    
    # Synthesise
    synthesis = client.chat.completions.create(
        model='gpt-5.6-sol',
        messages=[{
            'role': 'user',
            'content': f"""Write a comprehensive report on '{topic}' using these research results:

{chr(10).join(f'Step {k}: {v[:500]}' for k, v in sorted(results.items()))}

Write a 600-word report with clear headings."""
        }],
    ).choices[0].message.content
    
    return synthesis

print(research_and_report("LangGraph for production AI agents"))

Exercise: Implement a plan-and-execute agent for a multi-step task of your choice: booking a travel itinerary (mock the tools), preparing a competitive analysis, or setting up a project structure. Print the full plan before execution so you can verify it makes sense. What happens if you add a new constraint mid-execution, does replanning help?