Plan-and-Execute Agent Pattern: Two-Phase Planning

[yasha-dev1]

Overview

The Plan-and-Execute pattern separates agent execution into two distinct phases: planning (generating a multi-step plan) and execution (carrying out each step). This approach is faster, cheaper, and more performant than traditional ReAct agents for complex tasks. The key insight is that you can use a powerful model for planning and lighter/specialized models for execution.

How It Works

Two-Phase Architecture

Phase 1: PLANNING
User Query → Planner (Strong LLM) → Multi-step Plan

Phase 2: EXECUTION
For each step in plan:
  → Executor (Lighter LLM or specialized model) → Tool calls → Result
  → Update plan if needed (replanning)

Flow Example

User: "Research the top 3 AI companies and create a comparison report"

PLANNING PHASE (Claude Opus 4.6):
Plan:
1. Search for "top AI companies 2026"
2. Extract top 3 companies from search results
3. For each company: search for revenue, funding, products
4. Compile data into comparison table
5. Write executive summary

EXECUTION PHASE (Claude Sonnet 3.7 or Haiku):
Step 1: search_web("top AI companies 2026") → [results]
Step 2: extract_entities(results) → ["OpenAI", "Anthropic", "Google DeepMind"]
Step 3a: search_web("OpenAI revenue 2026") → [data]
Step 3b: search_web("Anthropic revenue 2026") → [data]
Step 3c: search_web("Google DeepMind revenue 2026") → [data]
Step 4: create_table(data) → [table]
Step 5: generate_summary(table) → [summary]

REPLANNING (if needed):
If step 3a fails → Planner generates alternative approach

Reference Implementations

• LangGraph Plan-and-Execute Tutorial - Official guide
• LangChain Blog: Planning Agents - Design patterns
• Build Dynamic Plan-and-Execute Agents with LangGraph - Tutorial
• LangChain AI Agents Complete Guide 2025 - Best practices

Proposed PyWorkflow Implementation

from pyworkflow import workflow, step, agent, hook
from pyworkflow.agents import PlanExecuteAgent, Planner, Executor

# Define executor tools
@step()
async def search_web(query: str) -> str:
    """Search the web for information."""
    return await search_api.query(query)

@step()
async def create_table(data: list[dict]) -> str:
    """Create a markdown table from data."""
    return markdown_table(data)

# Create plan-and-execute agent
@agent(
    pattern="plan_execute",
    planner_model="claude-opus-4-6",        # Strong model for planning
    executor_model="claude-sonnet-3-7-20250219",  # Faster/cheaper for execution
    tools=[search_web, create_table],
    enable_replanning=True,                  # Allow plan updates
    require_approval=False,                  # Set True for human-in-the-loop
)
async def research_agent(query: str):
    """
    Plan-and-execute research agent.
    
    Phase 1: Generate multi-step plan
    Phase 2: Execute each step with tools
    Phase 3: Replan if steps fail or new info emerges
    """
    pass

# Use the agent
result = await research_agent.run(
    "Research top 3 AI companies and compare their products"
)

print(result.plan)          # Initial plan
print(result.executed_steps) # Step-by-step execution log
print(result.answer)        # Final output

Human-in-the-Loop Approval

Use PyWorkflow's hook() primitive for plan approval:

@agent(
    pattern="plan_execute",
    planner_model="claude-opus-4-6",
    executor_model="claude-haiku-3-7",
    tools=[search_web, send_email, charge_credit_card],
    require_approval=True,  # Require human approval before execution
)
async def billing_agent(query: str):
    pass

# Workflow automatically suspends after planning
result = await billing_agent.run("Process refund for order #12345")
# → Agent generates plan and creates approval hook

# Later: Human approves or rejects
await approve_workflow_hook(hook_id, approved=True, feedback="Looks good")
# → Agent resumes and executes approved plan

Internal implementation:

class PlanExecuteAgent:
    async def run(self, query: str):
        # Phase 1: Planning
        plan = await self._generate_plan(query)
        
        if self.require_approval:
            # Suspend workflow and wait for approval
            approval = await hook(
                "plan_approval",
                timeout="24h",
                payload={"plan": plan}
            )
            
            if not approval["approved"]:
                # Replan with feedback
                plan = await self._generate_plan(query, feedback=approval["feedback"])
        
        # Phase 2: Execution
        results = await self._execute_plan(plan)
        return results

Event Types

Plan-and-execute agents record these events:

1. AGENT_STARTED - Agent execution begins

   {"run_id": "abc123", "query": "...", "planner_model": "claude-opus-4-6", "executor_model": "claude-sonnet-3-7"}

2. AGENT_PLAN_GENERATED - Planner creates initial plan

   {
     "plan": [
       {"step": 1, "description": "Search for top AI companies", "tool": "search_web"},
       {"step": 2, "description": "Extract top 3 from results", "tool": "extract_entities"},
       ...
     ],
     "planner_tokens": 1200
   }

3. AGENT_PLAN_APPROVAL_REQUESTED - Waiting for human approval (if enabled)

   {"hook_id": "hook_xyz", "plan": [...]}

4. AGENT_PLAN_APPROVED / AGENT_PLAN_REJECTED - Approval decision

   {"hook_id": "hook_xyz", "approved": true, "feedback": null}

5. AGENT_STEP_STARTED - Executor begins a plan step

   {"step": 1, "description": "Search for top AI companies", "tool": "search_web"}

6. AGENT_STEP_COMPLETED - Step execution finishes

   {"step": 1, "result": "...", "executor_tokens": 300}

7. AGENT_REPLANNING - Plan modified mid-execution

   {
     "reason": "Step 3 failed, searching for alternative approach",
     "original_plan": [...],
     "new_plan": [...]
   }

8. AGENT_EXECUTION_COMPLETED - All steps executed

   {"total_steps": 5, "successful": 5, "failed": 0, "replans": 1}

9. AGENT_COMPLETED / AGENT_FAILED

Implementation Details

Planner Implementation

class Planner:
    async def generate_plan(self, query: str, tools: list, context: dict = None) -> list[Step]:
        """Generate multi-step plan using strong LLM."""
        
        system_prompt = f"""
        You are a planning assistant. Given a user query and available tools, 
        generate a detailed step-by-step plan.
        
        Available tools:
        {self._format_tool_descriptions(tools)}
        
        Output a JSON array of steps:
        [
          {{"step": 1, "description": "...", "tool": "tool_name", "depends_on": []}},
          {{"step": 2, "description": "...", "tool": "tool_name", "depends_on": [1]}},
          ...
        ]
        """
        
        response = await self.llm.generate(
            model=self.planner_model,  # claude-opus-4-6
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": query}
            ]
        )
        
        plan = json.loads(response.text)
        
        # Record plan in event log
        await ctx.record_event(EventType.AGENT_PLAN_GENERATED, {
            "plan": plan,
            "planner_tokens": response.usage.total_tokens
        })
        
        return plan

Executor Implementation

class Executor:
    async def execute_step(self, step: dict, context: dict) -> Any:
        """Execute a single plan step using lighter LLM or direct tool call."""
        
        await ctx.record_event(EventType.AGENT_STEP_STARTED, step)
        
        tool_name = step["tool"]
        tool_function = self.tools[tool_name]
        
        # Option 1: Direct tool execution (no LLM)
        if self._is_deterministic(step):
            result = await tool_function(**step.get("params", {}))
        
        # Option 2: Use lighter LLM to determine tool parameters
        else:
            tool_params = await self._get_tool_params(step, context)
            result = await tool_function(**tool_params)
        
        await ctx.record_event(EventType.AGENT_STEP_COMPLETED, {
            "step": step["step"],
            "result": result
        })
        
        return result

Replanning Logic

class PlanExecuteAgent:
    async def _execute_plan(self, plan: list[Step]) -> dict:
        results = {}
        
        for step in plan:
            try:
                result = await self.executor.execute_step(step, context=results)
                results[step["step"]] = result
            
            except Exception as e:
                if self.enable_replanning:
                    # Replan with failure information
                    new_plan = await self.planner.replan(
                        original_plan=plan,
                        failed_step=step,
                        error=str(e),
                        partial_results=results
                    )
                    
                    await ctx.record_event(EventType.AGENT_REPLANNING, {
                        "reason": f"Step {step['step']} failed: {e}",
                        "original_plan": plan,
                        "new_plan": new_plan
                    })
                    
                    # Continue with new plan
                    plan = new_plan
                else:
                    raise
        
        return results

Trade-offs

Pros

• Cost efficient: Use expensive model only for planning, cheap models for execution
• Faster: Parallel execution of independent steps possible
• Better for complex tasks: Explicit planning improves success rate
• Debuggable: Clear separation of planning vs execution failures
• Human oversight: Easy to inject approval step after planning
• Replanning: Adaptive to failures and new information

Cons

• More complex: Two-phase architecture vs single agent loop
• Latency: Planning phase adds upfront delay
• Rigidity: Plan may become outdated as execution progresses
• Overhead: Not worth it for simple single-step tasks

Comparison to ReAct and Tool-Calling

Aspect	Plan-Execute	ReAct	Tool-Calling
Speed (complex tasks)	Faster	Slower	Medium
Cost (complex tasks)	Lower	Higher	Medium
Explainability	High (plan visible)	High (thoughts visible)	Low
Adaptability	Medium (replanning)	High (per-step)	Low
Setup Complexity	High	Medium	Low

When to Use Plan-and-Execute

Use Plan-and-Execute when:

• Task requires 5+ steps
• You want to use different models for planning vs execution
• Human approval needed before expensive operations
• Cost optimization is important
• Clear upfront plan is valuable

Use ReAct when:

• Task requires adaptive reasoning per step
• You need visible "thinking" traces
• Steps are highly interdependent

Use Tool-Calling when:

• Task is simple (1-3 steps)
• Speed is critical
• Cost of planning phase not justified

Related Issues

• ReAct Agent Pattern: Reasoning + Acting Loop #153 - ReAct Agent (adaptive per-step reasoning)
• Tool-Calling Agent Pattern: Direct Function Calling #156 - Tool-Calling Agent (simpler single-phase execution)
• #[ISSUE_4] - Router/Dispatcher Agent (routes to plan-execute specialists)
• #[ISSUE_6] - Code Generation Agent (uses plan-execute for multi-file changes)

References

• LangChain Blog: Planning Agents
• LangGraph Plan-and-Execute Tutorial
• Build Dynamic Plan-and-Execute Agents (Medium)
• LangChain AI Agents Guide 2025
• LangGraph: Workflows and Agents

Implementation Checklist

• Create pyworkflow/agents/plan_execute.py with PlanExecuteAgent class
• Implement Planner class (multi-step plan generation)
• Implement Executor class (single-step execution)
• Add replanning logic with failure handling
• Integrate hook() for plan approval
• Add event types: AGENT_PLAN_GENERATED, AGENT_STEP_STARTED, AGENT_REPLANNING
• Support different models for planner vs executor
• Add parallel execution for independent steps
• Create @agent(pattern="plan_execute") decorator
• Add tests for planning, execution, and replanning
• Document plan-and-execute pattern in examples/
• Add integration test with human approval workflow