ReAct vs Plan-and-Execute: Agent Reasoning Patterns Compared

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool

llm = ChatOpenAI(model="gpt-4")

@tool
async def search_database(query: str) -> dict:
    """Search database for information."""
    return {"results": [...]}

@tool
async def calculate(expression: str) -> float:
    """Perform calculation."""
    return eval(expression)  # Sandboxed in production

REACT_PROMPT = """You solve problems by thinking step-by-step and using tools.

Available tools:
{tools}

Use this format:

Thought: [Your reasoning about what to do next]
Action: [Tool name]
Action Input: [Tool input]
Observation: [Tool output will appear here]

... repeat Thought/Action/Observation as needed ...

Thought: I now know the final answer
Final Answer: [Your answer]

Task: {task}

Begin!"""

async def react_agent(task: str, max_iterations: int = 10):
    """ReAct agent implementation."""
    
    tools = [search_database, calculate]
    tool_map = {tool.name: tool for tool in tools}
    
    prompt = ChatPromptTemplate.from_template(REACT_PROMPT)
    
    # Initialize conversation
    messages = []
    tool_descriptions = "\n".join([
        f"- {tool.name}: {tool.description}" for tool in tools
    ])
    
    current_text = prompt.format(tools=tool_descriptions, task=task)
    
    for iteration in range(max_iterations):
        # Reasoning step
        response = await llm.ainvoke([
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": current_text}
        ])
        
        response_text = response.content
        current_text += f"\n{response_text}"
        
        # Check for final answer
        if "Final Answer:" in response_text:
            final_answer = response_text.split("Final Answer:")[-1].strip()
            return {
                "answer": final_answer,
                "iterations": iteration + 1,
                "reasoning_trace": current_text
            }
        
        # Extract action
        if "Action:" in response_text and "Action Input:" in response_text:
            action_block = response_text.split("Action:")[1].split("Action Input:")
            action_name = action_block[0].strip()
            action_input = action_block[1].split("\n")[0].strip()
            
            # Execute tool
            if action_name in tool_map:
                try:
                    result = await tool_map[action_name].ainvoke(action_input)
                    observation = f"\nObservation: {result}"
                except Exception as e:
                    observation = f"\nObservation: Error executing {action_name}: {str(e)}"
            else:
                observation = f"\nObservation: Unknown tool {action_name}"
            
            current_text += observation
        else:
            # No valid action found - prompt for proper format
            current_text += "\n\nRemember to use the Thought/Action/Action Input format."
    
    return {
        "answer": "Max iterations reached without final answer",
        "iterations": max_iterations,
        "reasoning_trace": current_text
    }

# Usage
result = await react_agent(
    "Find the total revenue for Q2 2026 and calculate the 15% tax"
)

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent
from typing import TypedDict, Annotated
import operator

class ReactState(TypedDict):
    """State for ReAct agent."""
    messages: Annotated[list, operator.add]
    iteration: int

def create_react_graph():
    """ReAct agent as LangGraph."""
    
    # Use prebuilt ReAct agent
    react_agent = create_react_agent(
        llm,
        tools=[search_database, calculate],
        state_modifier="You are a helpful assistant solving tasks step-by-step."
    )
    
    return react_agent

# Or build custom ReAct graph
def create_custom_react_graph():
    """Custom ReAct implementation with explicit control."""
    
    workflow = StateGraph(ReactState)
    
    def reasoning_node(state: ReactState):
        """Think about next action."""
        response = llm.invoke(state["messages"])
        return {
            "messages": [response],
            "iteration": state.get("iteration", 0) + 1
        }
    
    def tool_execution_node(state: ReactState):
        """Execute tools based on reasoning."""
        last_message = state["messages"][-1]
        
        # Parse tool calls from message
        tool_calls = extract_tool_calls(last_message)
        
        results = []
        for call in tool_calls:
            result = execute_tool(call["name"], call["args"])
            results.append({
                "role": "tool",
                "content": str(result),
                "tool_call_id": call["id"]
            })
        
        return {"messages": results}
    
    def should_continue(state: ReactState) -> str:
        """Decide if more iterations needed."""
        last_message = state["messages"][-1]
        
        # Check for final answer
        if "final answer" in last_message.content.lower():
            return "end"
        
        # Check iteration limit
        if state.get("iteration", 0) >= 10:
            return "end"
        
        # Check for tool calls
        if extract_tool_calls(last_message):
            return "execute_tools"
        
        return "reason"
    
    workflow.add_node("reasoning", reasoning_node)
    workflow.add_node("tool_execution", tool_execution_node)
    
    workflow.add_conditional_edges(
        "reasoning",
        should_continue,
        {
            "execute_tools": "tool_execution",
            "reason": "reasoning",
            "end": END
        }
    )
    
    workflow.add_edge("tool_execution", "reasoning")
    workflow.set_entry_point("reasoning")
    
    return workflow.compile()

from typing import List, Dict
from dataclasses import dataclass

@dataclass
class Step:
    """Single execution step."""
    id: int
    description: str
    tool: str
    args: Dict
    dependencies: List[int]  # Step IDs that must complete first

class PlanExecuteAgent:
    """Plan-and-Execute agent implementation."""
    
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
    
    async def create_plan(self, task: str) -> List[Step]:
        """Planning phase - create complete plan."""
        
        planning_prompt = f"""Create a detailed plan to accomplish this task:
        
        Task: {task}
        
        Available tools:
        {self._format_tools()}
        
        Output a JSON list of steps:
        [
          {{
            "id": 1,
            "description": "Step description",
            "tool": "tool_name",
            "args": {{"arg1": "value1"}},
            "dependencies": []
          }},
          ...
        ]
        
        Ensure steps are in logical order with dependencies specified."""
        
        response = await self.llm.ainvoke([
            {"role": "system", "content": "You are an expert planner."},
            {"role": "user", "content": planning_prompt}
        ])
        
        # Parse JSON plan
        plan_json = extract_json(response.content)
        
        plan = [
            Step(
                id=step["id"],
                description=step["description"],
                tool=step["tool"],
                args=step["args"],
                dependencies=step.get("dependencies", [])
            )
            for step in plan_json
        ]
        
        return plan
    
    async def execute_plan(self, plan: List[Step]) -> Dict:
        """Execution phase - run steps in order."""
        
        results = {}
        
        # Topological sort for dependency order
        sorted_steps = self._topological_sort(plan)
        
        for step in sorted_steps:
            # Wait for dependencies
            for dep_id in step.dependencies:
                if dep_id not in results:
                    raise ValueError(f"Dependency {dep_id} not satisfied for step {step.id}")
            
            logger.info(
                "Executing step",
                extra={
                    "step_id": step.id,
                    "description": step.description
                }
            )
            
            # Execute step
            try:
                # Inject dependency results into args if needed
                args = self._resolve_dependencies(step.args, results)
                
                result = await self.tools[step.tool].ainvoke(args)
                results[step.id] = {
                    "step": step.description,
                    "result": result,
                    "status": "success"
                }
            
            except Exception as e:
                logger.error(
                    "Step failed",
                    extra={"step_id": step.id, "error": str(e)}
                )
                results[step.id] = {
                    "step": step.description,
                    "error": str(e),
                    "status": "failed"
                }
                
                # Abort on failure
                return {
                    "status": "failed",
                    "completed_steps": len(results),
                    "total_steps": len(plan),
                    "results": results
                }
        
        return {
            "status": "success",
            "completed_steps": len(results),
            "total_steps": len(plan),
            "results": results
        }
    
    async def run(self, task: str) -> Dict:
        """Full plan-and-execute cycle."""
        
        # Planning phase
        logger.info("Planning phase started")
        plan = await self.create_plan(task)
        logger.info(f"Plan created with {len(plan)} steps")
        
        # Execution phase
        logger.info("Execution phase started")
        results = await self.execute_plan(plan)
        
        return {
            "task": task,
            "plan": [{"id": s.id, "description": s.description} for s in plan],
            **results
        }
    
    def _topological_sort(self, plan: List[Step]) -> List[Step]:
        """Sort steps by dependencies."""
        sorted_steps = []
        remaining = plan.copy()
        
        while remaining:
            # Find steps with satisfied dependencies
            ready = [
                step for step in remaining
                if all(dep in [s.id for s in sorted_steps] for dep in step.dependencies)
            ]
            
            if not ready:
                raise ValueError("Circular dependency detected in plan")
            
            sorted_steps.extend(ready)
            for step in ready:
                remaining.remove(step)
        
        return sorted_steps
    
    def _resolve_dependencies(self, args: Dict, results: Dict) -> Dict:
        """Replace dependency placeholders with actual results."""
        resolved = {}
        
        for key, value in args.items():
            if isinstance(value, str) and value.startswith("$step_"):
                # Reference to previous step result
                step_id = int(value.replace("$step_", ""))
                resolved[key] = results[step_id]["result"]
            else:
                resolved[key] = value
        
        return resolved
    
    def _format_tools(self) -> str:
        """Format tool descriptions."""
        return "\n".join([
            f"- {name}: {tool.description}"
            for name, tool in self.tools.items()
        ])

# Usage
agent = PlanExecuteAgent(llm, [search_database, calculate])

result = await agent.run(
    "Find the total revenue for Q2 2026 and calculate the 15% tax"
)

from langgraph.graph import StateGraph, END

class PlanExecuteState(TypedDict):
    """State for plan-and-execute agent."""
    task: str
    plan: List[Step]
    current_step: int
    step_results: Dict[int, any]
    status: str

def create_plan_execute_graph():
    """Plan-and-execute as LangGraph."""
    
    workflow = StateGraph(PlanExecuteState)
    
    def planning_node(state: PlanExecuteState):
        """Create execution plan."""
        planning_agent = PlanExecuteAgent(llm, tools)
        plan = await planning_agent.create_plan(state["task"])
        
        return {
            "plan": plan,
            "current_step": 0,
            "step_results": {}
        }
    
    def execution_node(state: PlanExecuteState):
        """Execute current step."""
        step = state["plan"][state["current_step"]]
        
        # Check dependencies satisfied
        for dep_id in step.dependencies:
            if dep_id not in state["step_results"]:
                return {
                    "status": "failed",
                    "error": f"Dependency {dep_id} not satisfied"
                }
        
        # Execute step
        result = await execute_step(step, state["step_results"])
        
        # Update results
        new_results = state["step_results"].copy()
        new_results[step.id] = result
        
        return {
            "step_results": new_results,
            "current_step": state["current_step"] + 1
        }
    
    def verification_node(state: PlanExecuteState):
        """Verify plan completion."""
        all_completed = len(state["step_results"]) == len(state["plan"])
        all_successful = all(
            r.get("status") == "success"
            for r in state["step_results"].values()
        )
        
        return {
            "status": "complete" if (all_completed and all_successful) else "failed"
        }
    
    def should_continue(state: PlanExecuteState) -> str:
        """Check if more steps remain."""
        if state["current_step"] < len(state["plan"]):
            return "execute_step"
        return "verify"
    
    workflow.add_node("planning", planning_node)
    workflow.add_node("execution", execution_node)
    workflow.add_node("verification", verification_node)
    
    workflow.add_edge("planning", "execution")
    
    workflow.add_conditional_edges(
        "execution",
        should_continue,
        {
            "execute_step": "execution",
            "verify": "verification"
        }
    )
    
    workflow.add_edge("verification", END)
    workflow.set_entry_point("planning")
    
    return workflow.compile()

class AdaptivePlanExecuteAgent:
    """Plan-and-execute with replanning on errors."""
    
    async def execute_with_replanning(self, task: str):
        """Execute plan, replan on failures."""
        
        # Initial plan
        plan = await self.create_plan(task)
        
        completed_steps = []
        
        for step in plan:
            try:
                result = await self.execute_step(step)
                completed_steps.append((step, result))
            
            except Exception as e:
                logger.warning(
                    "Step failed, replanning",
                    extra={"step": step.id, "error": str(e)}
                )
                
                # Replan from current state
                remaining_task = self._formulate_remaining_task(
                    original_task=task,
                    completed_steps=completed_steps,
                    failed_step=step
                )
                
                new_plan = await self.create_plan(remaining_task)
                plan = new_plan  # Replace remaining plan
        
        return {"completed_steps": completed_steps}

class HierarchicalAgent:
    """Plan-and-execute at high level, ReAct for complex steps."""
    
    async def execute(self, task: str):
        """High-level planning, low-level ReAct."""
        
        # Create high-level plan
        plan = await self.create_high_level_plan(task)
        
        results = []
        
        for step in plan:
            if self._is_complex_step(step):
                # Use ReAct for complex steps
                logger.info(f"Using ReAct for complex step: {step.description}")
                result = await self.react_agent.run(step.description)
            else:
                # Direct execution for simple steps
                result = await self.execute_step(step)
            
            results.append(result)
        
        return {"results": results}
    
    def _is_complex_step(self, step: Step) -> bool:
        """Determine if step needs ReAct."""
        complexity_indicators = [
            "research",
            "investigate",
            "troubleshoot",
            "explore",
            "find best"
        ]
        
        return any(
            indicator in step.description.lower()
            for indicator in complexity_indicators
        )

def create_checkpointed_plan_execute():
    """Plan-and-execute with mid-execution validation."""
    
    workflow = StateGraph(PlanExecuteState)
    
    # ... planning and execution nodes ...
    
    def validation_checkpoint(state: PlanExecuteState):
        """Validate progress mid-execution."""
        completed = state["step_results"]
        
        # Check if plan is still valid given results
        validation_prompt = f"""Given these completed steps:
        {json.dumps(completed, indent=2)}
        
        Is the remaining plan still appropriate:
        {json.dumps(state["plan"][state["current_step"]:], indent=2)}
        
        Respond: "yes" or "no" with explanation."""
        
        response = llm.invoke(validation_prompt)
        
        if "no" in response.content.lower():
            # Replan needed
            return {"status": "needs_replanning"}
        
        return {"status": "continue"}
    
    # Add validation after every 3 steps
    workflow.add_node("validation", validation_checkpoint)
    
    return workflow

from langgraph.prebuilt import create_react_agent

# ReAct agent in one line
react_agent = create_react_agent(
    llm=ChatOpenAI(model="gpt-4"),
    tools=[search_database, calculate],
    state_modifier="You are a helpful assistant."
)

# Use it
result = react_agent.invoke({
    "messages": [{"role": "user", "content": "Calculate tax on Q2 revenue"}]
})

# Build custom plan-and-execute
def create_plan_execute_agent():
    workflow = StateGraph(PlanExecuteState)
    
    # Add planning, execution, verification nodes
    workflow.add_node("plan", planning_node)
    workflow.add_node("execute", execution_node)
    workflow.add_node("verify", verification_node)
    
    # Sequential flow
    workflow.add_edge("plan", "execute")
    workflow.add_conditional_edges(
        "execute",
        should_continue_executing,
        {"continue": "execute", "done": "verify"}
    )
    
    workflow.set_entry_point("plan")
    
    return workflow.compile(
        checkpointer=PostgresCheckpointer(DATABASE_URL)  # Stateful execution
    )

# BAD: No loop detection
while not done:
    action = reason()
    result = execute(action)

# GOOD: Iteration limit and loop detection
for iteration in range(MAX_ITERATIONS):
    action = reason()
    
    if action in recent_actions[-3:]:
        logger.warning("Loop detected, breaking")
        break
    
    result = execute(action)

# BAD: Sequential when parallel possible
for step in plan:
    await execute(step)

# GOOD: Parallel execution of independent steps
independent_steps = find_independent_steps(plan)
await asyncio.gather(*[execute(step) for step in independent_steps])

import pytest

@pytest.mark.asyncio
async def test_react_reaches_answer():
    """Test ReAct completes task."""
    result = await react_agent("What is 2 + 2?")
    
    assert "4" in result["answer"]
    assert result["iterations"] < 10

@pytest.mark.asyncio
async def test_react_uses_tools():
    """Test ReAct calls appropriate tools."""
    with patch('tools.calculate') as mock_calc:
        mock_calc.return_value = 4
        
        result = await react_agent("Calculate 2 + 2")
        
        assert mock_calc.called
        assert result["answer"] == "4"

@pytest.mark.asyncio
async def test_react_recovers_from_errors():
    """Test ReAct handles tool failures."""
    with patch('tools.search_database') as mock_search:
        # First call fails, second succeeds
        mock_search.side_effect = [
            Exception("Connection error"),
            {"results": ["data"]}
        ]
        
        result = await react_agent("Search for user data")
        
        assert result["iterations"] > 1  # Retried
        assert "data" in str(result)

@pytest.mark.asyncio
async def test_plan_execute_creates_valid_plan():
    """Test planning creates executable plan."""
    agent = PlanExecuteAgent(llm, tools)
    
    plan = await agent.create_plan("Generate report")
    
    # Verify plan structure
    assert len(plan) > 0
    assert all(hasattr(step, 'id') for step in plan)
    assert all(hasattr(step, 'tool') for step in plan)
    
    # Verify dependencies are valid
    step_ids = {step.id for step in plan}
    for step in plan:
        assert all(dep in step_ids for dep in step.dependencies)

@pytest.mark.asyncio
async def test_plan_execute_handles_failures():
    """Test failure handling in execution."""
    agent = PlanExecuteAgent(llm, tools)
    
    with patch.object(agent.tools['search_database'], 'ainvoke') as mock:
        mock.side_effect = Exception("Database error")
        
        result = await agent.run("Search and analyze data")
        
        assert result["status"] == "failed"
        assert "error" in str(result)

recent_actions = deque(maxlen=5)

for iteration in range(MAX_ITERATIONS):
    action = get_next_action()
    
    # Check for repeated actions
    if action in recent_actions:
        logger.warning("Repeated action detected")
        break
    
    recent_actions.append(action)
    execute(action)

def hybrid_agent(task):
    # Analyze task complexity
    if is_well_defined(task):
        return plan_execute_agent(task)
    else:
        return react_agent(task)