10. Planning Systems

Planning transforms a vague goal into a sequence of executable steps. Without a planner, agents flail—making random tool calls, repeating actions, or giving up when the path isn't obvious.

Two planning modes:

1. Zero-shot: The LLM plans on the fly, deciding each step based on current state.
2. Deliberate: The agent generates a full plan upfront, then executes step by step.

Zero-shot is simpler but less reliable for complex tasks. Deliberate planning is more reliable but requires the agent to commit to a sequence before execution.

Deliberate planner implementation:

@dataclass
class PlanStep:
    step_id: int
    description: str
    tool_name: Optional[str]
    tool_args: Optional[dict[str, Any]]
    status: str = "pending"  # pending, executing, completed, failed
    result: Optional[Any] = None

@dataclass
class Plan:
    goal: str
    steps: list[PlanStep]
    current_step: int = 0
    
    def is_complete(self) -> bool:
        return all(s.status == "completed" for s in self.steps)
    
    def next_step(self) -> Optional[PlanStep]:
        if self.current_step < len(self.steps):
            step = self.steps[self.current_step]
            step.status = "executing"
            return step
        return None

class DeliberatePlanner:
    def __init__(self, llm: LLMInterface):
        self.llm = llm
    
    async def create_plan(self, goal: str, tools: list[dict]) -> Plan:
        prompt = f"""You are a task planner. Given the goal and available tools, create a step-by-step plan.
        
Goal: {goal}

Available tools:
{json.dumps(tools, indent=2)}

Respond with a JSON plan:
{{
  "steps": [
    {{"description": "Step 1 description", "tool_name": "tool_name", "tool_args": {{}}}},
    ...
  ]
}}"""

        response = await self.llm.generate(prompt)
        plan_data = json.loads(response)
        
        steps = [
            PlanStep(
                step_id=i,
                description=s["description"],
                tool_name=s.get("tool_name"),
                tool_args=s.get("tool_args", {})
            )
            for i, s in enumerate(plan_data["steps"])
        ]
        
        return Plan(goal=goal, steps=steps)

Failure mode: plan-step mismatch. The plan assumes a tool exists with specific arguments, but the actual tool has different parameters. Always validate the plan against the tool registry before execution.

async def validate_plan(self, plan: Plan, registry: ToolRegistry) -> list[str]:
    errors = []
    for step in plan.steps:
        if step.tool_name:
            try:
                tool = registry.get(step.tool_name)
                # Check required parameters
                for param in tool.parameters.get("required", []):
                    if param not in step.tool_args:
                        errors.append(
                            f"Step {step.step_id}: Missing required parameter '{param}' "
                            f"for tool '{step.tool_name}'"
                        )
            except KeyError:
                errors.append(f"Step {step.step_id}: Unknown tool '{step.tool_name}'")
    return errors