Planning and Execution

1def simple_agent_loop(goal: str):
2    """Agent without explicit planning."""
3    state = AgentState(goal=goal)
4
5    while not state.completed:
6        # Decide next action based on current context
7        context = gather_context(state)
8        action = llm.decide_action(context)
9
10        # Execute and observe
11        result = execute(action)
12        state.update(action, result)
13
14        # Check completion
15        if is_goal_complete(state, result):
16            state.completed = True
17
18    return state.final_result

1def planning_agent(goal: str):
2    """Agent with full upfront planning."""
3
4    # Create complete plan first
5    plan = create_plan(goal)
6    print(f"Plan with {len(plan.steps)} steps created")
7
8    # Execute each step
9    for step in plan.steps:
10        action = step_to_action(step)
11        result = execute(action)
12
13        if not result.success:
14            # Plan failed - need to replan
15            plan = replan(goal, completed_steps, result.error)
16
17    return plan.final_result
18
19
20def create_plan(goal: str) -> Plan:
21    """Use LLM to create a complete plan."""
22    prompt = f"""
23Create a detailed step-by-step plan to accomplish this goal:
24{goal}
25
26For each step, specify:
271. What to do
282. Expected outcome
293. How to verify success
30
31Return as a numbered list.
32"""
33    response = llm.generate(prompt)
34    return parse_plan(response.text)

1@dataclass
2class HierarchicalPlan:
3    """Multi-level plan structure."""
4    goal: str
5    phases: list["Phase"]
6
7
8@dataclass
9class Phase:
10    """High-level phase of a plan."""
11    name: str
12    description: str
13    steps: list["Step"]
14    status: str = "pending"
15
16
17@dataclass
18class Step:
19    """Concrete step within a phase."""
20    description: str
21    tool_hint: str | None = None
22    status: str = "pending"
23
24
25class HierarchicalPlanner:
26    """Creates and manages multi-level plans."""
27
28    def create_plan(self, goal: str, context: str = "") -> HierarchicalPlan:
29        """Create a hierarchical plan."""
30
31        # First, identify major phases
32        phases_prompt = f"""
33Break this goal into 3-5 major phases:
34Goal: {goal}
35Context: {context}
36
37Return phase names and descriptions.
38"""
39        phases_response = self.llm.generate(phases_prompt)
40        phases = self._parse_phases(phases_response.text)
41
42        # Then, detail each phase
43        for phase in phases:
44            steps_prompt = f"""
45Goal: {goal}
46Current phase: {phase.name} - {phase.description}
47
48List the specific steps needed to complete this phase.
49"""
50            steps_response = self.llm.generate(steps_prompt)
51            phase.steps = self._parse_steps(steps_response.text)
52
53        return HierarchicalPlan(goal=goal, phases=phases)
54
55    def get_current_step(self, plan: HierarchicalPlan) -> Step | None:
56        """Get the next step to execute."""
57        for phase in plan.phases:
58            if phase.status == "pending":
59                for step in phase.steps:
60                    if step.status == "pending":
61                        return step
62                # All steps in phase done, mark phase complete
63                phase.status = "completed"
64        return None  # All done

1class PlanExecutor:
2    """Executes plans step by step."""
3
4    def __init__(self, agent, planner):
5        self.agent = agent
6        self.planner = planner
7
8    def execute(self, plan: Plan) -> ExecutionResult:
9        """Execute a plan to completion."""
10
11        for i, step in enumerate(plan.steps):
12            # Skip completed steps
13            if step.status == "completed":
14                continue
15
16            step.status = "in_progress"
17            print(f"Step {i+1}/{len(plan.steps)}: {step.description}")
18
19            # Translate step to action(s)
20            actions = self._step_to_actions(step)
21
22            # Execute each action
23            for action in actions:
24                result = self.agent.execute(action)
25
26                if not result.success:
27                    step.status = "failed"
28                    return ExecutionResult(
29                        success=False,
30                        failed_step=step,
31                        error=result.error,
32                    )
33
34            step.status = "completed"
35
36        return ExecutionResult(success=True)
37
38    def _step_to_actions(self, step: Step) -> list[Action]:
39        """Convert a plan step to concrete actions."""
40        prompt = f"""
41Convert this plan step into specific tool calls:
42Step: {step.description}
43Tool hint: {step.tool_hint or "None"}
44
45Available tools: {self.agent.tool_descriptions}
46
47Return the tool name and parameters for each action needed.
48"""
49        response = self.agent.llm.generate_with_tools(
50            prompt=prompt,
51            tools=self.agent.tools,
52        )
53        return self._parse_actions(response)

1class AdaptivePlanner:
2    """Planner that adapts to failures and changes."""
3
4    def replan(
5        self,
6        original_plan: Plan,
7        completed_steps: list[Step],
8        failure: str,
9    ) -> Plan:
10        """Create a new plan after failure."""
11
12        prompt = f"""
13The original plan failed. Create a new plan.
14
15Original goal: {original_plan.goal}
16
17What worked:
18{self._format_completed(completed_steps)}
19
20What failed:
21{failure}
22
23Lessons learned:
24- The previous approach didn't work
25- We need to try something different
26
27Create a revised plan that:
281. Builds on what was already completed
292. Avoids the approach that failed
303. Takes a different path to the goal
31"""
32        response = self.llm.generate(prompt)
33        return self._parse_plan(response.text)
34
35    def should_replan(self, state: AgentState) -> bool:
36        """Determine if replanning is needed."""
37        # Replan after multiple consecutive failures
38        if state.consecutive_failures >= 3:
39            return True
40
41        # Replan if we're stuck (same action repeated)
42        if self._is_stuck(state):
43            return True
44
45        # Replan if major assumption proved wrong
46        if state.has_major_error():
47            return True
48
49        return False
50
51    def _is_stuck(self, state: AgentState) -> bool:
52        """Detect if agent is repeating the same action."""
53        recent = state.actions[-5:] if len(state.actions) >= 5 else []
54        if len(recent) < 3:
55            return False
56        # Check if last 3 actions are the same
57        return len(set(a.name for a in recent[-3:])) == 1

1class ReActAgent:
2    """Agent using the ReAct pattern."""
3
4    def run(self, goal: str) -> str:
5        """Execute using ReAct pattern."""
6
7        state = AgentState(goal=goal)
8
9        while not state.completed:
10            # Generate thought + action together
11            react_response = self._react_step(state)
12
13            # Log the reasoning
14            print(f"Thought: {react_response.thought}")
15
16            if react_response.action:
17                print(f"Action: {react_response.action.name}")
18
19                # Execute action
20                result = self.execute(react_response.action)
21                print(f"Observation: {result.output[:200]}")
22
23                # Update state
24                state.add_step(
25                    thought=react_response.thought,
26                    action=react_response.action,
27                    observation=result.output,
28                )
29
30            if react_response.final_answer:
31                state.completed = True
32                state.final_result = react_response.final_answer
33
34        return state.final_result
35
36    def _react_step(self, state: AgentState) -> ReActResponse:
37        """Generate one ReAct step."""
38
39        prompt = f"""
40Goal: {state.goal}
41
42{self._format_history(state)}
43
44Respond in this format:
45Thought: [Your reasoning about what to do next]
46Action: [tool_name(param1="value1", param2="value2")]
47
48OR if you have the final answer:
49Thought: [Your reasoning]
50Final Answer: [Your final response to the goal]
51"""
52        response = self.llm.generate(prompt)
53        return self._parse_react(response.text)
54
55
56@dataclass
57class ReActResponse:
58    """Parsed ReAct response."""
59    thought: str
60    action: Action | None = None
61    final_answer: str | None = None