Planning & Task Decomposition

Complex tasks require planning. Without a plan, agents attempt to solve everything in a single linear pass, often losing track of sub-goals or repeating work. This lesson covers the plan-and-execute pattern, hierarchical planning, task graphs, re-planning on failure, and how to validate plans before execution.

---

Why Planning Matters

Consider a complex request:

"Research the top 5 AI startups in healthcare, compare their funding, products, and team sizes, then write a summary report."

A naive agent would try to do everything in one long chain. A planning agent first breaks this into sub-tasks:

Plan:
  1. Search for top AI startups in healthcare
  2. For each startup, gather:
     a. Funding information
     b. Product descriptions
     c. Team size data
  3. Compile all data into a comparison table
  4. Write a summary report

---

The Plan-and-Execute Pattern

graph TD
  U["USER TASK"] --> P["PLANNER (LLM Call 1) — "Break this task into numbered steps.""]
  P --> E["EXECUTOR (Loop) — for each step in the plan: 1. Execute the step (LLM + tools); 2. Record the result; 3. Check if re-planning is needed"]
  E --> F["FINAL ANSWER"]

Implementation

from openai import OpenAI
import json

client = OpenAI()

def create_plan(task: str) -> list[str]:
    """Use the LLM to generate a plan."""
    response = client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": (
                "You are a planning agent. Given a task, break it into a numbered "
                "list of concrete steps. Return JSON: {\"steps\": [\"step1\", \"step2\", ...]}"
            )},
            {"role": "user", "content": f"Task: {task}"},
        ],
    )
    plan = json.loads(response.choices[0].message.content)
    return plan["steps"]

def execute_step(step: str, context: str) -> str:
    """Execute a single step of the plan."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": (
                "You are an execution agent. Complete the given step using the "
                "context from previous steps. Be concise and factual."
            )},
            {"role": "user", "content": (
                f"Step to execute: {step}\n\n"
                f"Context from previous steps:\n{context}"
            )},
        ],
    )
    return response.choices[0].message.content

def plan_and_execute(task: str) -> str:
    """Full plan-and-execute agent."""
    print("Creating plan...")
    steps = create_plan(task)
    for i, step in enumerate(steps):
        print(f"  Step {i + 1}: {step}")

    context = ""
    results = []

    for i, step in enumerate(steps):
        print(f"\nExecuting step {i + 1}/{len(steps)}: {step}")
        result = execute_step(step, context)
        results.append(f"Step {i + 1} ({step}): {result}")
        context = "\n".join(results)
        print(f"  Result: {result[:100]}...")

    return context

---

Hierarchical Planning

For very complex tasks, use a hierarchy of plans where high-level steps are themselves decomposed into sub-plans:

graph TD
  G["Level 0 (Goal): Write a market analysis report"]
  G --> R["Research market"]
  G --> A["Analyse data"]
  G --> C["Compare competitors"]
  G --> W["Write report"]
  R --> S["Search sources"]
  R --> FI["Filter results"]
  R --> SU["Summarise findings"]

def hierarchical_plan(task: str, depth: int = 0, max_depth: int = 2) -> dict:
    """Recursively plan with sub-tasks."""
    plan = create_plan(task)

    result = {"task": task, "steps": []}
    for step in plan:
        step_info = {"description": step}

        # Check if this step is complex enough to warrant sub-planning
        if depth < max_depth and is_complex(step):
            step_info["sub_plan"] = hierarchical_plan(step, depth + 1, max_depth)

        result["steps"].append(step_info)

    return result

def is_complex(step: str) -> bool:
    """Heuristic: a step is complex if it contains multiple verbs or 'and'."""
    complexity_indicators = ["and", "then", "also", "additionally"]
    return any(indicator in step.lower() for indicator in complexity_indicators)

---

Task Graphs

Not all plans are linear. Some steps can run in parallel, and some have dependencies.