Python Implementation of Reasoning and Planning
This chapter introduces the core intelligent capabilities of AI Agent, namely reasoning and planning abilities.
Unlike traditional instruction execution patterns, an Agent with reasoning capabilities can engage in complex thinking processes.
It can decompose complex tasks into executable steps and make decisions based on context.
ReAct Framework
ReAct (Reasoning + Acting) is a framework that combines reasoning and acting.
In the ReAct pattern, the Agent alternates between reasoning and acting, forming a closed loop.
Reasoning guides action, and the results of actions feed back into the reasoning process, repeating in this cycle.
ReAct simulates the human problem-solving process: think before taking action, observe results after acting, and then continue reasoning based on new information.
Working Principle
The core idea of ReAct can be summarized as a three-phase loop:
First, the reasoning phase. The Agent generates the next action thought based on the current context.
Second, the action phase. The Agent executes the selected tool or action.
Third, the observation phase. The Agent receives the action result, updates the context, and returns to the first step to continue reasoning.
Code Implementation
Below is a simplified implementation of a ReAct Agent:
Basic Implementation of ReAct Agent
"""
Agent Implementation of the ReAct Framework
Core mechanism: reasoning -> action -> observation -> reasoning (loop)
"""
def __init__(self, llm, tools, max_iterations=5):
# LLM instance, used for reasoning and generation
self.llm = llm
# List of available tools
self.tools = tools
# Maximum number of iterations to prevent infinite loops
self.max_iterations = max_iterations
def run(self, task):
"""
Execute ReAct loop
:param task: User task description
:return: Execution result
"""
# Initialize context, including task and history information
context = {
"task": task,
"steps": [], # Record of executed steps
"observations": [] # Record of observation results
}
for i in range(self.max_iterations):
# Phase 1: Reasoning - Generate the next action based on the current context
reasoning = self.llm.reason(context)
# Determine whether a tool needs to be executed, or return the answer directly
if reasoning.needs_action:
# Select the tool to use
action = reasoning.select_tool(self.tools)
# Execute the tool and obtain the observation result
observation = action.execute(reasoning.tool_input)
# Add the observation result to the context
context["observations"].append(observation)
else:
# No action needed, directly return the answer obtained from reasoning
return reasoning.answer
# Reached the maximum number of iterations without completing
return "Maximum iteration limit reached"
Typical Application Scenarios
The ReAct pattern is particularly suitable for the following scenarios:
Tasks that need to be completed after exploring the environment and collecting information.
Intelligent search scenario: The Agent needs to search for information first, then reason and summarize based on the search results.
Conversational Q&A: The Agent needs to clarify requirements and obtain information through multi-turn dialogue.
Complex problem solving: The problem cannot be solved in one step and requires multiple intermediate steps.
Note: The advantage of the ReAct pattern lies in its flexibility, allowing it to dynamically adjust subsequent actions based on the execution result of each step. But this also means the execution path may be unstable, making it suitable for tasks that require exploration.
Chain of Thought (CoT)
Chain of Thought is a technique that prompts the model to demonstrate a step-by-step reasoning process.
CoT does not ask the model to give the answer directly, but guides it to first show the reasoning steps and then reach the final conclusion.
Why Chain of Thought is Needed
Directly making the model output the answer has several problems:
Intermediate steps in complex reasoning are easily overlooked or skipped.
It is difficult to pinpoint where the error occurred.
Users cannot understand the model's thinking process.
Chain-of-Thought solves these issues by requiring the model to show its reasoning process.
Zero-shot CoT
Zero-shot CoT is a method that can elicit step-by-step reasoning ability without requiring examples.
Simply add a trigger phrase like "Let's think step by step" at the end of the prompt.
Zero-shot CoT Example
prompt = """
Question: Xiao Ming has 5 apples. Xiao Hong gives him 3. Xiao Ming eats 2. How many are left?
Let's think step by step:
"""
# Call the language model
response = llm.generate(prompt)
# Example model output:
# Step 1: Calculate the total number of apples after Xiao Ming receives them
# Xiao Ming originally had 5 apples, Xiao Hong gave him 3
# 5 + 3 = 8
#
# Step 2: Calculate the number after eating
# Xiao Ming ate 2
# 8 - 2 = 6
#
# Conclusion: There are 6 apples left
Few-shot CoT
Few-shot CoT helps the model learn specific reasoning patterns by providing examples that include detailed reasoning processes.
When Zero-shot CoT is not effective, you can try Few-shot CoT.
Few-shot CoT Example
prompt = """
Example 1:
Question: Xiao Zhang has 10 yuan. He buys 3 books at 2 yuan each. How much is left?
Let's think step by step:
- Xiao Zhang originally had 10 yuan
- Each book costs 2 yuan, buys 3 books, spending 3 × 2 = 6 yuan
- 10 - 6 = 4 yuan
Answer: 4 yuan left
Example 2:
Question: A cat catches 2 mice per hour. How many does it catch in 8 hours?
Let's think step by step:
- Catches 2 mice per hour
- In 8 hours, catches 8 × 2 = 16
Answer: Caught 16
Question: {user_question}
Let's think step by step:
"""
response = llm.generate(prompt)
Advantages of Chain of Thought
Chain-of-Thought technology brings three core values:
Traceability: Decomposes complex reasoning into traceable intermediate steps, making the reasoning process easier to understand.
Interpretability: Enhance the model's interpretability, letting users know where answers come from.
Accuracy improvement: Significantly improves accuracy on complex reasoning tasks, especially effective in mathematical and logical tasks.
Tree of Thoughts（ToT）
Tree of Thoughts is an extension of Chain of Thought; it is no longer limited to linear reasoning.
ToT explores multiple possible paths at each reasoning node, forming a tree-like structure.
This enables the Agent to perform multi-path exploration, backtracking, and global evaluation.
Differences from Chain of Thought
CoT uses linear reasoning, where each step depends on the conclusion of the previous step, suitable for problems with clear paths.
ToT uses spatial reasoning, considering multiple possible branches simultaneously, suitable for problems that require exploration and planning.
Code Implementation
Basic Implementation of ToT Agent
"""
Tree of Thoughts Agent Implementation
Core mechanism: Generate multiple candidate branches at each node, evaluate them, then choose the best to continue.
"""
def __init__(self, llm, max_depth=4, beam_size=3):
# LLM instance
self.llm = llm
# Maximum search depth
self.max_depth = max_depth
# Number of candidate nodes retained per level (beam width)
self.beam_size = beam_size
def solve(self, problem):
"""
Solving problems using the ToT framework
:param problem: Problem description
:return: Optimal solution
"""
# Create root node, containing the problem as the initial state
root = ThoughtNode(problem, depth=0)
# Initialize frontier node list (nodes to be expanded)
frontier = [root]
# Expand layer by layer
for depth in range(self.max_depth):
# Store all candidate nodes
all_candidates = []
# Iterate over all frontier nodes
for node in frontier:
# Generate multiple candidate next steps for the current node
candidates = self.llm.generate_thoughts(
node.content, # Current node content
n=self.beam_size # Generation count
)
# Create a new node and add it to the candidate list
for cand in candidates:
all_candidates.append(
ThoughtNode(cand, depth + 1, parent=node)
)
# Evaluate all candidate nodes
evaluated = self.evaluator.rank(all_candidates)
# Select the best beam_size nodes as the next frontier
frontier = evaluated[:self.beam_size]
# Check if a solution has been found
if self.is_solution(frontier):
break
# Backtrack to find the optimal solution
return self.backtrack_best(frontier)
Application Scenarios
ToT is particularly suitable for scenarios that require making choices or planning:
Creative writing: Generate multiple story development directions, evaluate them, then choose the best.
Strategic planning: Evaluate the potential outcomes of multiple action plans.
Complex decision-making: Decision problems that require considering multiple possibilities.
Task Planning and MCTS
Monte Carlo Tree Search (MCTS) is a heuristic search algorithm used for complex decision-making problems.
MCTS has wide applications in Agent planning, especially suitable for game AI and complex task planning.
Core Idea
MCTS evaluates the potential value of each decision node by simulating random games.
It does not need to evaluate all possible paths, but uses sampling and statistics to guide the search direction.
Four Steps of MCTS
Step 1, Selection: Starting from the root node, recursively select the optimal child node until reaching a leaf node.
During selection, the UCB (Upper Confidence Bound) formula is used to balance exploration and exploitation.
Step 2, Expansion: Add one or more child nodes to the leaf node.
Step 3, Simulation: Starting from the new node, randomly simulate the game until the end.
Step 4, Backpropagation: Update the statistics of all nodes along the simulation path.
Code Implementation
MCTS Planner Implementation
class MCTSNode:
"""
MCTS Tree Node
Stores node state, statistics, and child node relationships
"""
def __init__(self, state, parent=None, action=None):
# Current state
self.state = state
# Parent node reference
self.parent = parent
# Action from the parent node to this node
self.action = action
# List of child nodes
self.children = []
# Visit count of this node
self.visit_count = 0
# Cumulative reward value of this node
self.reward = 0.0
def is_fully_expanded(self):
"""Check whether all possible child nodes have been expanded"""
return len(self.children) > 0
def is_terminal(self):
"""Check whether it is a terminal node (game over or goal achieved)"""
return self.state.is_terminal()
def uct_child(self):
"""
Use the UCT formula to select the optimal child node
UCT = reward/visits + C * sqrt(ln(parent_visits)/visits)
C is the exploration constant, usually set to sqrt(2)
"""
# Exploration constant, balancing exploration and exploitation
C = math.sqrt(2)
return max(
self.children,
key=lambda c: c.reward / c.visit_count +
C * math.sqrt(math.log(self.visit_count) / c.visit_count)
)
class MCTSPlanner:
"""
MCTS Planner
Uses Monte Carlo Tree Search to generate action plans for the Agent
"""
def __init__(self, simulation_limit=1000, exploration_constant=1.41):
# Maximum number of simulations
self.simulation_limit = simulation_limit
# Exploration constant
self.exploration_constant = exploration_constant
def plan(self, initial_state):
"""
Begin planning from the initial state
:param initial_state: initial state
:return: optimal action
"""
# Create the root node
root = MCTSNode(initial_state)
# Perform multiple simulations
for _ in range(self.simulation_limit):
# 1. Selection: Select the optimal child node from the root node downward until a leaf node is reached
node = self._selection(root)
# 2. Expansion: If it is not a terminal node, expand a new node
if not node.is_terminal():
node = self._expansion(node)
# 3. Simulation: Randomly simulate from the new node to the terminal
reward = self._simulation(node)
# 4. Backpropagation: Update the statistics of all nodes on the path
self._backpropagation(node, reward)
# Return the action corresponding to the optimal child node of the root node
return root.best_child().action
def _selection(self, node):
"""Selection phase: select the optimal child node"""
while node.is_fully_expanded() and not node.is_terminal():
node = node.uct_child()
return node
def _expansion(self, node):
"""Expansion phase: add a new child node"""
# Generate all possible actions
possible_actions = node.state.get_possible_actions()
# Create a child node for each action
for action in possible_actions:
new_state = node.state.apply_action(action)
child = MCTSNode(new_state, parent=node, action=action)
node.children.append(child)
# Return a random child node (or use a deterministic strategy)
return node.children[0]
def _simulation(self, node):
"""Simulation phase: randomly simulate until the game ends"""
state = node.state
while not state.is_terminal():
# Randomly choose an action
actions = state.get_possible_actions()
action = random.choice(actions)
state = state.apply_action(action)
# Return the final reward
return state.get_reward()
def _backpropagation(self, node, reward):
"""Backpropagation: update statistics"""
while node is not None:
node.visit_count += 1
node.reward += reward
node = node.parent
Note: MCTS has high computational cost and is suitable for scenarios that require deep planning but have clear termination conditions. For tasks with high real-time requirements, it may be necessary to limit the number of simulations or use other methods.
Reflexion (Self-Reflection)
Reflexion is a technique that gives an Agent self-reflection capabilities.
By adding a reflection mechanism to the Agent, it can analyze the causes of failure after a failure, adjust its strategy, and retry.
Core Idea
The Agent not only executes actions, but also observes results and reflects: Why did I fail? How should I improve next time?
This capability is crucial for continuous learning and self-improvement.
Code Implementation
Reflexion Agent Implementation
"""
An Agent with self-reflection capability
Core mechanism: Execute -> Evaluate -> Reflect -> Retry
"""
def __init__(self, actor, reviewer, max_retries=3):
# Executor: responsible for executing specific tasks
self.actor = actor
# Evaluator: responsible for evaluating execution results
self.reviewer = reviewer
# Maximum retry count
self.max_retries = max_retries
def run(self, task):
"""
Execute task with self-reflection
:param task: task description
:return: execution result
"""
# Maintain execution history
history = []
for attempt in range(self.max_retries):
# Phase 1: Attempt to execute the task
result = self.actor.execute(task, history)
# Record this attempt
history.append({
"attempt": attempt,
"result": result
})
# Phase 2: Evaluate the result
feedback = self.reviewer.evaluate(task, result)
# Phase 3: Check whether it succeeded
if feedback.is_success:
return result
# Phase 4: Reflect on the cause of failure
# Generate a new strategy prompt to guide the next attempt
reflection = self.reviewer.reflect(
task, # Original task
result, # Failed result
feedback # Evaluation feedback
)
# Add the reflection result to history for reference in the next attempt
history.append({
"type": "reflection",
"content": reflection
})
# If all retries fail, return the history
return history
class Reviewer:
"""Evaluator: evaluate execution results and generate reflections"""
def evaluate(self, task, result):
"""
Evaluate execution result
:return: an object containing whether it succeeded and detailed feedback
"""
# Check whether the result meets the task requirements
is_success = self.check_success(task, result)
if is_success:
return EvaluationResult(is_success=True)
else:
# Generate failure cause analysis
failure_reasons = self.analyze_failures(task, result)
return EvaluationResult(
is_success=False,
reasons=failure_reasons
)
def reflect(self, task, result, feedback):
"""
Generate reflection content
Help the Actor understand the cause of failure and improve its strategy
"""
prompt = f"""
Task: {task}
Execution result: {result}
Failure reasons: {feedback.reasons}
Please analyze the causes of failure and provide suggestions for improvement.
Key points:
1. What went wrong?
2. How should we avoid it next time?
3. What strategy needs to change?
"""
return self.llm.generate(prompt)
Application Scenarios
Reflexion is particularly suitable for the following scenarios:
Tasks that require continuous improvement, such as dialogue systems, code generation, etc.
Scenarios where the cost of errors is high but the cost of retrying is low.
Situations where it is necessary to learn from failure.
Task Decomposition Strategies
Complex tasks usually need to be decomposed into manageable subtasks.
Effective task decomposition is the foundation of planning ability.
Recursive Task Decomposition
Recursively decompose a task into smaller subtasks until the subtasks can be executed directly.
Recursive Task Decomposition
"""
Recursively decompose tasks
:param task: The task to decompose
:param is_executable_fn: Function to determine whether the task can be executed directly
:param decompose_fn: Function to decompose the task
:return: List of executable tasks
"""
# If the task can be executed directly, return it directly
if is_executable_fn(task):
return [task]
# Decompose the task into subtasks
subtasks = decompose_fn(task)
# Recursively decompose each subtask
result = []
for subtask in subtasks:
# Recursively call the decomposition function on subtasks
result.extend(
decompose(subtask, is_executable_fn, decompose_fn)
)
return result
# Example: Determine whether the task is executable
def is_code_complete(task):
"""Check whether the task can directly execute code"""
return task.get("type") == "code" and len(task.get("dependencies", [])) == 0
# Example: Decompose a complex task
def decompose_programming_task(task):
"""Decompose a programming task"""
if "write_function" in task["type"]:
return [
{"type": "understand_requirements", "task": task["spec"]},
{"type": "write_code", "spec": task["spec"]},
{"type": "write_tests", "function": task["function_name"]},
{"type": "verify_tests", "function": task["function_name"]}
]
return [task]
Parallel Task Decomposition
Identify independent subtasks that can be executed in parallel to improve execution efficiency.
This is a key strategy for accelerating task execution.
Hierarchical Task Decomposition
Divide tasks into different levels of abstraction; high-level tasks call low-level tasks, forming a task hierarchy tree.
Suitable for complex systems that require multiple levels of abstraction.
Plan-and-Execute Pattern
Plan-and-Execute is an architectural pattern that separates planning from execution.
The Agent first fully plans the entire task flow, then executes according to the plan.
Differences from ReAct
ReAct interleaves reasoning and execution, which is more flexible but the path may be unstable.
Plan-and-Execute plans first and then executes, which is more stable but lacks dynamic adjustment capability.
Code Implementation
Plan-and-Execute Agent Implementation
"""
Agent with Plan-and-Execute architecture
Core mechanism: plan completely first, then execute according to the plan
"""
def __init__(self, planner, executor):
# Planner: responsible for creating the execution plan
self.planner = planner
# Executor: responsible for executing specific steps
self.executor = executor
def run(self, task):
"""
Execute task
Divided into planning phase and execution phase
"""
# ==================== Planning Phase ====================
# Generate the complete execution plan at once
plan = self.planner.create_plan(task)
# ==================== Execution Phase ====================
results = []
# Execute in order according to the plan
for step in plan.steps:
# Execute the current step
result = self.executor.execute(step)
results.append(result)
# Check whether replanning is needed
# For example: execution results do not match expectations, or unexpected situations occur
if self.needs_replan(results):
# Replan based on current results
plan = self.planner.replan(task, results)
# Integrate all results
return self.summarize(results)
def needs_replan(self, results):
"""
Determine whether replanning is needed
Check whether execution results match expectations
"""
# Get the last result
last_result = results[-1]
# If the result clearly does not match expectations, replanning is needed
if last_result.is_unexpected():
return True
# If the result makes subsequent steps impossible to execute, replanning is needed
if last_result.blocks_future():
return True
return False
class Planner:
"""Planner: generate task execution plan"""
def create_plan(self, task):
"""Create initial execution plan"""
prompt = f"""
Task: {task}
Please create a detailed execution plan, including:
1. The order of steps to execute
2. The specific operations for each step
3. The dependency relationships between steps
Output format:
Step 1: [Operation description]
Step 2: [Operation description]
...
"""
plan_text = self.llm.generate(prompt)
return Plan.parse(plan_text)
def replan(self, task, results):
"""Replan based on execution results"""
# Analyze the completed results
completed = [r for r in results if r.is_success]
failed = [r for r in results if not r.is_success]
# Generate an adjusted plan
prompt = f"""
Original task: {task}
Completed steps: {completed}
Failed steps: {failed}
Please create an adjusted execution plan:
"""
plan_text = self.llm.generate(prompt)
return Plan.parse(plan_text)
class Executor:
"""Executor: execute specific plan steps"""
def execute(self, step):
"""Execute a single step"""
# Choose the execution method based on the step type
if step.type == "tool_call":
return self.execute_tool_call(step)
elif step.type == "code":
return self.execute_code(step)
elif step.type == "query":
return self.execute_query(step)
return Result(success=False, error="Unknown step type")
Note: The Plan-and-Execute mode is suitable for scenarios where the task structure is relatively stable and can be planned in advance. For environments that require flexible adaptation, the ReAct mode may be more suitable.
Chapter Summary
This chapter introduces the core reasoning and planning capabilities of AI Agents.
ReActThe framework enables the Agent to adjust its strategy while executing through a reasoning-and-action loop.
Chain of Thought (CoT)Enhance the accuracy of executing complex tasks by demonstrating the step-by-step reasoning process.
Tree of Thoughts (ToT)Expanded the chain of thought, supporting multi-path exploration and backtracking.
MCTSIt is a heuristic search algorithm, suitable for complex decision-making and planning problems.
ReflexionEndow the Agent with self-reflection capabilities to learn and improve from failures.
Task decomposition strategyIt is the foundation of planning ability, including recursive decomposition, parallel decomposition, and hierarchical decomposition.
Plan-and-ExecuteAdopts a plan-first-then-execute approach, suitable for tasks with stable structures.
These technologies can be used alone or in combination to build more powerful Agent systems.
Other extensions