各位同仁，各位对人工智能与软件工程充满热情的专家学者们：
今天，我们齐聚一堂，探讨一个在构建高智能、自主性代理（Agent）时至关重要的模式——“Plan-and-Execute”（规划与执行）。在AI领域，尤其是大型语言模型（LLM）的兴起，我们看到了代理在各种任务中展现出惊人的能力。然而，当任务变得复杂、耗时且需要多步骤协调时，传统的“边走边看”或“ReAct”模式便会暴露出其固有的局限性。此时，“Plan-and-Execute”模式的优势便凸显出来，它为我们提供了一种结构化、高效且更具鲁棒性的解决方案。
ReAct 模式的局限性与魅力：为什么它在长任务中力不从心？
在深入探讨“Plan-and-Execute”之前，我们有必要先回顾一下目前广泛使用的“ReAct”模式。ReAct，全称“Reasoning and Acting”，其核心思想是让代理在一个循环中进行思考（Thought）、行动（Action）和观察（Observation）。代理根据当前的观察和内部状态进行思考，决定下一步要采取什么行动，然后执行该行动，并观察其结果，以此作为下一轮思考的依据。
ReAct 的运作机制
我们来看一个简化的 ReAct 代理的工作流程：
- 思考 (Thought): 代理分析当前任务和环境观察，决定下一步的逻辑推理。
- 行动 (Action): 代理根据思考结果，选择一个工具（Tool）并调用它，例如执行代码、搜索网络、调用API等。
- 观察 (Observation): 代理获取行动的结果，并将其作为下一轮思考的输入。
这个循环不断重复，直到任务完成或达到某个停止条件。
代码示例：一个简单的 ReAct 代理骨架
import time
class Tool:
def __init__(self, name, description, func):
self.name = name
self.description = description
self.func = func
def __call__(self, *args, **kwargs):
print(f"Executing tool: {self.name} with args: {args}, kwargs: {kwargs}")
return self.func(*args, **kwargs)
# 模拟一些工具
def search_web(query):
"""Searches the web for information."""
print(f"Searching for: '{query}'...")
time.sleep(1) # Simulate network latency
if "Python best practices" in query:
return "Found an article: 'PEP 8: Style Guide for Python Code' and 'The Zen of Python'."
return f"Search result for '{query}': Some relevant information."
def execute_code(code_string):
"""Executes a given Python code string."""
print(f"Executing code:n```pythonn{code_string}n```")
try:
# For security, in a real system, you'd use a sandboxed environment.
# Here, we'll just simulate execution and potential output.
if "print('Hello')" in code_string:
return "Output: Hello"
if "result = 10 + 5" in code_string:
return "Execution successful. Variable 'result' might be 15."
return "Code executed successfully (simulated)."
except Exception as e:
return f"Code execution failed: {str(e)}"
# 注册工具
available_tools = {
"search_web": Tool("search_web", "Searches the internet for a query.", search_web),
"execute_code": Tool("execute_code", "Executes Python code in a safe environment.", execute_code),
}
class SimpleReActAgent:
def __init__(self, llm_model, tools):
self.llm = llm_model
self.tools = tools
self.history = []
def _format_tools_for_llm(self):
tool_descriptions = "n".join([
f"- {name}: {tool.description}" for name, tool in self.tools.items()
])
return f"You have access to the following tools:n{tool_descriptions}"
def run(self, task, max_steps=5):
self.history = []
prompt_template = """
You are an AI assistant that can reason and act.
Your goal is to complete the following task: {task}
{tool_descriptions}
Your response should always follow the Thought-Action-Observation format.
Thought: You should always think about what to do.
Action: Use the format `tool_name(arg1=value1, arg2=value2)` to call a tool.
Observation: The result of the action.
Begin!
{history_log}
"""
for step in range(max_steps):
current_prompt = prompt_template.format(
task=task,
tool_descriptions=self._format_tools_for_llm(),
history_log="n".join(self.history)
)
print(f"n--- Step {step + 1} ---")
print(f"Current History:n{self.history}")
print(f"Sending prompt to LLM:n{current_prompt}")
# Simulate LLM response
llm_response = self.llm.generate(current_prompt)
print(f"LLM Raw Response:n{llm_response}")
self.history.append(llm_response)
# Parse LLM response for Action
if "Action:" in llm_response:
action_str = llm_response.split("Action:")[1].split("Observation:")[0].strip()
print(f"Parsed Action String: {action_str}")
try:
# A more robust parser would be needed for real applications
# For simplicity, we'll use a very basic one or just assume specific formats.
# Example: search_web(query='Python best practices')
tool_name_match = action_str.split('(')[0].strip()
args_str = action_str[len(tool_name_match)+1:-1] # Remove tool_name()
if tool_name_match in self.tools:
tool = self.tools[tool_name_match]
# This is a very simplistic arg parser. A real one would use regex or AST.
# For demonstration, let's assume `key='value'` format
kwargs = {}
if args_str:
for arg_pair in args_str.split(','):
if '=' in arg_pair:
key, value = arg_pair.split('=', 1)
kwargs[key.strip()] = value.strip().strip("'"") # remove quotes
observation = tool(**kwargs)
self.history.append(f"Observation: {observation}")
else:
self.history.append(f"Observation: Unknown tool '{tool_name_match}'.")
except Exception as e:
self.history.append(f"Observation: Error parsing/executing action: {str(e)}")
elif "Thought:" in llm_response and "Action:" not in llm_response:
# LLM might just think and not act yet, or think it's done.
print("LLM is thinking or has concluded.")
if "Task complete" in llm_response: # A simple heuristic for task completion
print("Agent believes task is complete.")
break
else:
print("LLM response did not contain a clear Action. Assuming task might be complete or stuck.")
break # Exit if no clear action
print("n--- Task Finished ---")
print(f"Final History:n{self.history}")
# 模拟一个非常简单的LLM
class MockLLM:
def generate(self, prompt):
# This is a highly simplified mock. A real LLM would process the prompt
# and generate a coherent response. Here, we simulate a few steps.
if "Python best practices" in prompt and "search_web" in prompt and "Step 1" in prompt:
return """Thought: The task requires understanding Python best practices. I should start by searching the web for this information.
Action: search_web(query='Python best practices')
"""
elif "Found an article: 'PEP 8'" in prompt and "Step 2" in prompt:
return """Thought: I have found information about Python best practices, specifically PEP 8. Now I should try to demonstrate executing some simple Python code based on these practices, like printing a string.
Action: execute_code(code_string='print(\'Hello, world!\')')
"""
elif "Output: Hello" in prompt and "Step 3" in prompt:
return """Thought: I have successfully executed a simple print statement. The task was to 'learn about and demonstrate Python best practices'. I've learned about PEP 8 and demonstrated a basic execution. This seems sufficient for a simple demonstration.
Task complete.
"""
else:
return "Thought: I am not sure what to do next based on the current prompt. Perhaps I need more information or the task is too complex for this simple agent. Task complete."
# 实例化并运行代理
mock_llm = MockLLM()
agent = SimpleReActAgent(mock_llm, available_tools)
agent.run("Learn about and demonstrate Python best practices.")ReAct 的魅力
- 简单直观： 循环结构易于理解和实现。
- 灵活性： 能够快速响应环境变化和工具的反馈。
- 交互性： 非常适合需要与用户或外部系统进行多次短促交互的任务。
- 动态适应： 代理可以根据每次观察到的结果调整其思考路径。
ReAct 在长任务中的局限性
尽管 ReAct 具有上述优点，但当任务变得复杂、需要多阶段、长链条的推理和行动时，它的不足便显而易见：
- 缺乏长远规划能力： ReAct 代理通常只关注当前步骤和短期目标。它没有一个宏观的计划来指导整个任务的进程，导致容易陷入局部最优解，甚至在循环中反复执行无效操作。
- “迷失”风险高： 随着任务步骤的增加，代理在历史记录中积累的信息越来越多，但缺乏结构化的记忆和对整体目标进度的把握，很容易“迷失”在大量的上下文信息中，忘记最初的目标。
- 效率低下： 每一步都需要重新思考、决策，可能导致重复劳动或不必要的探索。对于已知明确步骤的任务，这种探索性的模式是低效的。
- 难以恢复错误： 当某个中间步骤失败时，ReAct 代理可能不知道如何有效地回溯、修正或重新规划，因为它没有一个预设的“正确路径”来参照。它可能只是尝试不同的行动，直到运气好解决问题，或者彻底失败。
- 上下文窗口限制： LLM 的上下文窗口是有限的。对于长任务，所有的历史记录很快就会填满上下文，导致代理无法“看到”早期的关键信息，从而影响决策质量。
- 缺乏结构化输出： 最终结果可能是一系列行动和观察的堆砌，而非一个清晰、有条理的解决方案或产品。
举个例子，如果任务是“开发一个简单的待办事项列表Web应用”，ReAct 代理可能会尝试编写HTML，然后突然去搜索CSS样式，接着又尝试写后端代码，而没有一个清晰的先数据库、再后端API、再前端界面的开发顺序。它可能在某个环节遇到问题后，无法有效回溯或调整整体策略。
‘Plan-and-Execute’ 模式的核心理念：宏观视野与步步为营
针对 ReAct 在长任务中的局限性，“Plan-and-Execute”模式应运而生。其核心思想在于：将复杂的长任务分解为两个截然不同但又紧密协作的阶段——规划阶段（Planning Phase）和执行阶段（Execution Phase）。
P&E 的基本思想
想象一下人类如何完成一个大型项目：无论是建造一座大楼、开发一个复杂的软件系统，还是组织一场大型活动。我们绝不会“边走边看”。相反，我们会：
- 制定详尽的计划： 确定最终目标，分解为多个子目标，定义每个子目标的具体步骤、所需资源、时间线和责任人。
- 按计划执行： 严格按照既定步骤推进，同时监控进度，处理遇到的问题。
“Plan-and-Execute”模式正是借鉴了这种人类智慧。
P&E 的关键组件
-
规划器 (Planner):
- 职责： 接收高层级的任务目标，并将其分解为一系列逻辑上连续、可执行的子任务或步骤。
- 输出： 一个结构化的计划，通常是一个步骤列表，每个步骤都清晰地定义了其目标和可能的执行方式。
- 智能来源： 通常由一个强大的LLM来担任，利用其强大的理解、推理和知识生成能力来构建计划。
-
执行器 (Executor):
- 职责： 接收规划器生成的计划，并按照计划的步骤逐一执行。
- 输出： 每个步骤的执行结果和状态（成功、失败、部分成功等）。
- 智能来源： 可以是另一个LLM来生成具体执行指令，也可以是调用特定的工具（如代码解释器、API调用器、Web搜索工具等）。执行器通常需要与外部环境进行交互。
-
监控器/反思器 (Monitor/Reflector) – 可选但强烈推荐：
- 职责： 监控执行器的进度和结果，评估当前状态是否与计划一致，或者是否需要修正计划。
- 输出： 评估报告，或触发重新规划的信号。
- 智能来源： 同样可以是LLM，用于进行高层次的判断和反思。
P&E 的工作流程
- 初始化： 代理接收一个高层级的任务目标。
- 规划阶段： 规划器（通常是LLM）分析任务目标，结合其内在知识和可用的工具描述，生成一个详细的、分步的执行计划。这个计划可能包括子目标、依赖关系和每一步的预期产出。
- 执行阶段： 执行器（可能也是LLM，但更侧重于调用工具）接收计划中的第一个步骤。它根据该步骤的描述，选择合适的工具并执行。
- 观察与更新： 执行器获取该步骤的执行结果。
- 监控与反思（可选）： 监控器评估当前步骤的执行结果。
- 如果成功且符合预期，则继续执行计划的下一个步骤。
- 如果失败或不符合预期，监控器可能会尝试进行局部修正，或者将问题反馈给规划器，触发重新规划（Re-planning）。重新规划可能意味着修改原有计划，或者生成一个新的子计划来解决当前问题。
- 迭代： 重复步骤3-5，直到所有计划步骤完成，任务达到最终目标。
P&E 为长任务带来的益处
- 宏观连贯性： 计划提供了整个任务的蓝图，确保所有步骤都服务于最终目标，避免了局部最优陷阱和漫无目的的探索。
- 显著提高效率： 一旦计划确定，执行器可以专注于按部就班地完成任务，减少了每一步的决策开销。对于已知或可预测的步骤，效率提升尤其明显。
- 更强的鲁棒性与错误恢复：
- 计划的结构使得问题更容易定位：哪个步骤失败了？
- 当某个步骤失败时，代理可以利用计划的上下文和监控器的反馈，更智能地进行错误处理（例如，重试、跳过、寻求帮助或重新规划）。
- 中间结果可以作为检查点，方便回溯。
- 模块化与可维护性： 任务被分解为独立的、可管理的步骤，每个步骤都可以使用不同的工具或策略来执行，提高了系统的模块化程度。
- 减轻 LLM 上下文压力： 规划器只需在规划时拥有完整上下文。执行器在执行每个步骤时，只需要关注当前步骤及其相关的上下文，不必载入整个任务的历史，有效缓解了上下文窗口的压力。
- 更好的可解释性： 有了明确的计划，代理的行为路径变得更加透明和可解释。我们可以清晰地知道代理正在做什么，为什么这样做。
深入剖析 ‘Plan-and-Execute’ 的实现细节
现在，让我们更具体地看看如何将“Plan-and-Execute”模式付诸实践。我们将以一个虚拟的“软件开发助手”为例，它需要完成一个相对复杂的任务：“开发一个简单的Web应用，允许用户输入一个待办事项，保存它，并能在一个页面上显示所有已保存的待办事项。”
这个任务涉及前端、后端、数据库等多个技术栈，非常适合展示P&E的优势。
核心组件设计
我们将设计一个 Agent 类，其中包含 Planner 和 Executor 的功能。
可用工具集 (Tools):
在我们的示例中，代理将拥有以下虚拟工具：
code_writer(instructions): 根据指令生成代码。shell_executor(command): 执行shell命令（如创建文件、安装依赖、运行程序）。file_manager(operation, path, content=None): 文件操作（读、写、创建目录）。db_initializer(db_type, schema_definition): 初始化数据库。browser_tester(url, actions): 模拟浏览器行为，测试Web应用。
import time
import json
import subprocess
import os
# --- 模拟工具集 ---
class Tool:
def __init__(self, name, description, func):
self.name = name
self.description = description
self.func = func
def __call__(self, *args, **kwargs):
print(f"[{self.name}] Executing with args: {args}, kwargs: {kwargs}")
try:
return self.func(*args, **kwargs)
except Exception as e:
return f"Error executing {self.name}: {str(e)}"
# 模拟代码生成工具
def _code_writer_func(instructions, language="python"):
print(f"Generating {language} code based on instructions: {instructions}")
time.sleep(1) # Simulate LLM thinking time
# In a real scenario, this would be an LLM call to generate code.
# Here, we'll return placeholder code based on common patterns.
if "Flask app setup" in instructions:
return """
from flask import Flask, render_template, request, redirect, url_for
import sqlite3
app = Flask(__name__)
DATABASE = 'todos.db'
def get_db_connection():
conn = sqlite3.connect(DATABASE)
conn.row_factory = sqlite3.Row
return conn
@app.route('/')
def index():
conn = get_db_connection()
todos = conn.execute('SELECT * FROM todos').fetchall()
conn.close()
return render_template('index.html', todos=todos)
@app.route('/add', methods=['POST'])
def add_todo():
if request.method == 'POST':
todo_text = request.form['todo_text']
conn = get_db_connection()
conn.execute('INSERT INTO todos (text) VALUES (?)', (todo_text,))
conn.commit()
conn.close()
return redirect(url_for('index'))
if __name__ == '__main__':
app.run(debug=True, port=5000)
"""
elif "SQLite schema for todos" in instructions:
return """
CREATE TABLE IF NOT EXISTS todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
text TEXT NOT NULL
);
"""
elif "HTML template for index page" in instructions:
return """
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>My Todo App</title>
<style>
body { font-family: sans-serif; margin: 20px; }
.todo-item { margin-bottom: 10px; padding: 10px; border: 1px solid #eee; }
</style>
</head>
<body>
<h1>Todo List</h1>
<form action="/add" method="post">
<input type="text" name="todo_text" placeholder="Add a new todo" required>
<button type="submit">Add Todo</button>
</form>
<hr>
<h2>Current Todos</h2>
{% if todos %}
{% for todo in todos %}
<div class="todo-item">{{ todo.text }}</div>
{% endfor %}
{% else %}
<p>No todos yet!</p>
{% endif %}
</body>
</html>
"""
return f"// Generated code for: {instructions}n# (Placeholder - real LLM would generate actual code)"
# 模拟 Shell 执行工具
def _shell_executor_func(command):
print(f"Executing shell command: `{command}`")
try:
# For security and stability, in a real agent, this would be sandboxed.
# Here, we just simulate.
if "pip install Flask" in command:
return "Successfully installed Flask (simulated)."
if "python app.py" in command:
print("Starting Flask app in background (simulated).")
# In a real scenario, this would be a non-blocking call
# and the agent would need to monitor the process.
return "Flask app started (simulated). Access at http://127.0.0.1:5000"
if "sqlite3 todos.db" in command and ".schema" in command:
return "SQLite schema displayed (simulated)."
if "python -c" in command: # For initializing DB
return "Database initialized with schema (simulated)."
# Generic command simulation
result = subprocess.run(command, shell=True, capture_output=True, text=True, check=False)
if result.returncode == 0:
return f"Command executed successfully.nSTDOUT:n{result.stdout}nSTDERR:n{result.stderr}"
else:
return f"Command failed with exit code {result.returncode}.nSTDOUT:n{result.stdout}nSTDERR:n{result.stderr}"
except Exception as e:
return f"Shell execution error: {str(e)}"
# 模拟文件管理工具
def _file_manager_func(operation, path, content=None):
print(f"File operation: {operation} on {path}")
if operation == "write":
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, 'w') as f:
f.write(content)
return f"File '{path}' written successfully."
elif operation == "read":
if os.path.exists(path):
with open(path, 'r') as f:
return f.read()
return f"File '{path}' not found."
elif operation == "create_dir":
os.makedirs(path, exist_ok=True)
return f"Directory '{path}' created."
return f"Unknown file operation: {operation}"
# 模拟数据库初始化工具
def _db_initializer_func(db_type, schema_definition):
print(f"Initializing {db_type} database with schema:n{schema_definition}")
if db_type == "sqlite":
db_file = "todos.db"
try:
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
cursor.execute(schema_definition)
conn.commit()
conn.close()
return f"SQLite database '{db_file}' initialized successfully."
except Exception as e:
return f"SQLite database initialization failed: {str(e)}"
return f"Unsupported database type: {db_type}"
# 模拟浏览器测试工具
def _browser_tester_func(url, actions):
print(f"Testing browser at {url} with actions: {actions}")
time.sleep(2) # Simulate browser interaction
# In a real scenario, this would use Selenium/Playwright.
# We'll simulate success/failure based on expected behavior.
if "/add" in actions[0].get("target_url_after_post", ""):
print("Simulating POST request to add todo...")
return "Simulated: Todo 'Test Item' added successfully and redirected to index."
if "http://127.0.0.1:5000" in url and "check_text" in actions[0]:
print(f"Simulating checking for text '{actions[0]['check_text']}' on page...")
if actions[0]['check_text'] in "My Todo App" or actions[0]['check_text'] in "Test Item":
return f"Simulated: Text '{actions[0]['check_text']}' found on page."
return f"Simulated: Text '{actions[0]['check_text']}' NOT found on page."
return "Browser test simulated, outcome indeterminate."
available_tools = {
"code_writer": Tool("code_writer", "Generates code based on instructions.", _code_writer_func),
"shell_executor": Tool("shell_executor", "Executes shell commands.", _shell_executor_func),
"file_manager": Tool("file_manager", "Performs file operations (write, read, create_dir).", _file_manager_func),
"db_initializer": Tool("db_initializer", "Initializes a database schema.", _db_initializer_func),
"browser_tester": Tool("browser_tester", "Simulates browser interactions for web testing.", _browser_tester_func),
}
# --- LLM 模拟器 ---
class MockLLM:
"""A highly simplified mock LLM for demonstration purposes."""
def generate(self, prompt, stop_sequences=None):
print(f"n--- LLM Called ---")
print(f"Prompt (excerpt):n{prompt[:500]}...") # Print first 500 chars of prompt
time.sleep(0.5) # Simulate processing time
# Simulate planning based on task
if "Generate a detailed step-by-step plan" in prompt:
return json.dumps([
{"step": 1, "description": "Set up the project directory and install Flask.", "tool_suggestion": "shell_executor, file_manager"},
{"step": 2, "description": "Define the SQLite database schema for storing todos.", "tool_suggestion": "code_writer"},
{"step": 3, "description": "Initialize the database with the defined schema.", "tool_suggestion": "db_initializer, shell_executor"},
{"step": 4, "description": "Generate the main Flask application code including routes for index and adding todos.", "tool_suggestion": "code_writer"},
{"step": 5, "description": "Create the HTML template for displaying todos and the input form.", "tool_suggestion": "code_writer"},
{"step": 6, "description": "Write the Flask app code to files.", "tool_suggestion": "file_manager"},
{"step": 7, "description": "Create the templates directory and write the HTML template to file.", "tool_suggestion": "file_manager"},
{"step": 8, "description": "Run the Flask application.", "tool_suggestion": "shell_executor"},
{"step": 9, "description": "Test the web application: visit the page, add a todo, and verify it appears.", "tool_suggestion": "browser_tester"}
])
# Simulate execution responses
elif "Set up the project directory and install Flask" in prompt:
return "shell_executor(command='mkdir todo_app && cd todo_app && pip install Flask')"
elif "Define the SQLite database schema for storing todos" in prompt:
return "code_writer(instructions='SQLite schema for todos table')"
elif "Initialize the database with the defined schema" in prompt:
return "db_initializer(db_type='sqlite', schema_definition='CREATE TABLE IF NOT EXISTS todos (id INTEGER PRIMARY KEY AUTOINCREMENT, text TEXT NOT NULL);')"
elif "Generate the main Flask application code" in prompt:
return "code_writer(instructions='Flask app setup with index route, add todo route, and SQLite integration')"
elif "Create the HTML template for displaying todos" in prompt:
return "code_writer(instructions='HTML template for index page with form and todo list')"
elif "Write the Flask app code to files" in prompt:
return "file_manager(operation='write', path='todo_app/app.py', content='<FLASK_APP_CODE_PLACEHOLDER>')" # Content replaced by actual code
elif "Create the templates directory and write the HTML template to file" in prompt:
return "file_manager(operation='create_dir', path='todo_app/templates')"
elif "templates/index.html" in prompt and "file_manager" in prompt:
return "file_manager(operation='write', path='todo_app/templates/index.html', content='<HTML_TEMPLATE_CODE_PLACEHOLDER>')" # Content replaced
elif "Run the Flask application" in prompt:
return "shell_executor(command='cd todo_app && python app.py')"
elif "Test the web application" in prompt:
return """
browser_tester(url='http://127.0.0.1:5000', actions=[
{'type': 'visit'},
{'type': 'input', 'selector': 'input[name="todo_text"]', 'value': 'Buy groceries'},
{'type': 'click', 'selector': 'button[type="submit"]', 'target_url_after_post': 'http://127.00.1:5000'},
{'type': 'check_text', 'text': 'Buy groceries'}
])
"""
elif "Reflect on the outcome" in prompt:
if "Error" in prompt or "Failed" in prompt:
return "Thought: The previous step failed. I need to re-evaluate the plan or try a different approach. Re-planning might be necessary for the failed step."
else:
return "Thought: The step executed successfully. The plan is progressing as expected. Continue to the next step."
return "Thought: I am unable to generate a specific response for this prompt. This might indicate an unhandled scenario or the task is complete. Proceeding with caution."
class PlanAndExecuteAgent:
def __init__(self, llm_model, tools):
self.llm = llm_model
self.tools = tools
self.plan = []
self.task_history = []
self.intermediate_results = {} # Store results from code_writer etc.
def _format_tools_for_llm(self):
tool_descriptions = "n".join([
f"- {name}: {tool.description}" for name, tool in self.tools.items()
])
return f"You have access to the following tools:n{tool_descriptions}"
def _parse_llm_tool_call(self, llm_response):
"""
Parses an LLM response to extract tool name and arguments.
Expected format: tool_name(arg1=value1, arg2='value2')
"""
if '(' not in llm_response or ')' not in llm_response:
return None, None
tool_name = llm_response.split('(')[0].strip()
args_str = llm_response[len(tool_name) + 1:-1] # Extract content inside parentheses
kwargs = {}
if args_str:
# This is a basic parser. For complex cases, consider `ast.literal_eval` or regex.
# It handles simple key=value pairs, including quoted strings.
parts = []
current_part = []
in_quote = False
for char in args_str:
if char == "'" or char == '"':
in_quote = not in_quote
if char == ',' and not in_quote:
parts.append("".join(current_part).strip())
current_part = []
else:
current_part.append(char)
parts.append("".join(current_part).strip())
for part in parts:
if '=' in part:
key, value = part.split('=', 1)
key = key.strip()
value = value.strip()
# Remove quotes if present
if (value.startswith("'") and value.endswith("'")) or
(value.startswith('"') and value.endswith('"')):
value = value[1:-1]
kwargs[key] = value
# else: handle positional args if needed
return tool_name, kwargs
def plan_task(self, task_description):
print("n--- Planning Phase ---")
planning_prompt = f"""
You are an expert project planner. Your goal is to break down the following complex task into a detailed, sequential plan.
The plan should be a list of steps, each with a description and a suggestion for which tools might be useful.
Output your plan as a JSON array of objects, where each object has 'step', 'description', and 'tool_suggestion' keys.
Task: {task_description}
{self._format_tools_for_llm()}
Example JSON output:
[
{{"step": 1, "description": "First thing to do.", "tool_suggestion": "tool_a, tool_b"}},
{{"step": 2, "description": "Next logical step.", "tool_suggestion": "tool_c"}}
]
"""
raw_plan = self.llm.generate(planning_prompt)
try:
self.plan = json.loads(raw_plan)
print("Plan generated successfully:")
for p in self.plan:
print(f" Step {p['step']}: {p['description']} (Tools: {p.get('tool_suggestion', 'N/A')})")
return True
except json.JSONDecodeError as e:
print(f"Error decoding plan JSON: {e}")
self.plan = []
return False
def execute_plan(self, task_description, project_root="todo_app"):
print("n--- Execution Phase ---")
if not self.plan:
print("No plan to execute. Please run plan_task first.")
return
# Ensure project root exists
self.tools["file_manager"]("create_dir", path=project_root)
for i, step_info in enumerate(self.plan):
step_num = step_info['step']
step_desc = step_info['description']
print(f"n--- Executing Step {step_num}/{len(self.plan)}: {step_desc} ---")
# Prepare context for the LLM to decide on execution
current_context = f"""
Task: {task_description}
Current Plan Step: {step_num} - {step_desc}
Previous steps and their results:
{json.dumps(self.task_history[-3:], indent=2)}
Intermediate results (e.g., generated code snippets, paths):
{json.dumps(self.intermediate_results, indent=2)}
Given the current step, its description, and the available tools, what is the best tool to call to execute this step?
Output your action in the format: `tool_name(arg1=value1, arg2='value2')`.
"""
# LLM decides which tool to use and with what arguments
llm_execution_command = self.llm.generate(current_context)
tool_name, kwargs = self._parse_llm_tool_call(llm_execution_command)
if tool_name and tool_name in self.tools:
tool_instance = self.tools[tool_name]
# Special handling for code_writer and file_manager interactions
if tool_name == "file_manager" and "content" in kwargs:
# Replace placeholder content with actual generated code if available
if "Flask_app_code" in self.intermediate_results and "app.py" in kwargs.get("path", ""):
kwargs["content"] = self.intermediate_results["Flask_app_code"]
elif "HTML_template_code" in self.intermediate_results and "index.html" in kwargs.get("path", ""):
kwargs["content"] = self.intermediate_results["HTML_template_code"]
# If the tool is code_writer, store its output for later use.
# If the tool is db_initializer, ensure we use the schema generated previously.
if tool_name == "code_writer":
result = tool_instance(**kwargs)
if "Flask app setup" in kwargs.get("instructions", ""):
self.intermediate_results["Flask_app_code"] = result
elif "HTML template for index page" in kwargs.get("instructions", ""):
self.intermediate_results["HTML_template_code"] = result
elif "SQLite schema" in kwargs.get("instructions", ""):
self.intermediate_results["SQLite_schema"] = result
elif tool_name == "db_initializer" and "schema_definition" in kwargs:
if "SQLite_schema" in self.intermediate_results:
kwargs["schema_definition"] = self.intermediate_results["SQLite_schema"]
result = tool_instance(**kwargs)
else:
result = tool_instance(**kwargs)
print(f"Tool '{tool_name}' output:n{result}")
self.task_history.append({"step": step_num, "description": step_desc, "tool": tool_name, "result": result})
# --- Monitoring and Reflection ---
reflection_prompt = f"""
You are a quality assurance agent. Review the execution result of the last step.
Task: {task_description}
Current Plan Step: {step_num} - {step_desc}
Tool Used: {tool_name}
Tool Output: {result}
Overall Goal: Ensure the web application is developed and functional.
Based on the tool output, did this step execute successfully and move us closer to the overall goal?
If there was an error, describe it and suggest if re-planning or a different approach is needed.
"""
reflection = self.llm.generate(reflection_prompt)
print(f"Reflection:n{reflection}")
if "Error" in reflection or "failed" in reflection.lower():
print(f"!!! Step {step_num} failed or encountered an issue. Re-evaluating plan. !!!")
# In a real system, here we would trigger re-planning or detailed error handling
return False # Stop execution for demonstration
else:
print(f"LLM suggested unknown tool or malformed command: {llm_execution_command}")
self.task_history.append({"step": step_num, "description": step_desc, "tool": "N/A", "result": f"Failed to parse or find tool: {llm_execution_command}"})
return False # Stop on unparseable command
print("n--- Plan Execution Complete ---")
return True
# --- 运行代理 ---
mock_llm_expert = MockLLM() # Using our mock LLM
agent = PlanAndExecuteAgent(mock_llm_expert, available_tools)
task = "Develop a simple web application that allows users to input text (todo items), save them, and view all saved texts on a single page using Flask and SQLite."
# 1. Planning Phase
if agent.plan_task(task):
# 2. Execution Phase
agent.execute_plan(task)
print("n--- Final Task History ---")
for entry in agent.task_history:
print(f"Step {entry['step']}: {entry['description']} -> Result: {entry['result'][:100]}...") # Truncate result for brevity规划阶段 (Planning Phase) 详解
目标： 将高层级任务分解为一系列明确、可执行的步骤。
LLM 作为规划器： 我们利用 LLM 的强大语言理解和生成能力，让它根据任务描述和可用工具，自动生成一个结构化的计划。规划器需要：
- 理解任务目标： 从自然语言描述中提取核心需求。
- 知识储备： 了解完成此类任务的通用流程（例如，Web开发通常涉及项目设置、数据库、后端、前端、部署、测试）。
- 工具感知： 知道哪些工具可以用于哪些类型的子任务。
输出格式： JSON 数组是很好的选择，因为它既结构化又易于解析。
[
{"step": 1, "description": "Set up the project directory and install Flask.", "tool_suggestion": "shell_executor, file_manager"},
{"step": 2, "description": "Define the SQLite database schema for storing todos.", "tool_suggestion": "code_writer"},
// ... 其他步骤 ...
]这样的计划提供了清晰的路线图，规划器不需要知道每个步骤的 具体实现细节，它只需要知道 要做什么 以及 可能用什么。
执行阶段 (Execution Phase) 详解
目标： 按照规划器生成的计划，一步步地执行任务。
LLM 作为执行指令生成器： 在每个计划步骤中，我们再次利用 LLM。但这次，它的角色是根据当前步骤的描述和历史上下文，决定调用哪个具体的工具，以及传递哪些参数。
- 上下文： LLM 需要当前步骤的描述、之前的执行结果（用于延续上下文或修正）、以及可用的工具列表。
- 决策： LLM 基于这些信息，生成一个格式化的工具调用指令（例如
shell_executor(command='pip install Flask')）。 - 工具调用：
PlanAndExecuteAgent解析 LLM 的指令，并实际调用对应的工具函数。
状态管理与中间结果： 在执行过程中，代理需要维护一些状态，例如：
self.task_history： 记录每个已执行步骤的描述、使用的工具和结果，用于后续的调试、回溯或反思。self.intermediate_results： 存储一些重要的中间产物，例如code_writer生成的代码片段、数据库 schema 定义等。这些产物会在后续步骤中被其他工具使用（例如，file_manager会将code_writer生成的代码写入文件）。
监控与反思 (Monitoring and Reflection) 详解
这是 P&E 模式中提升鲁棒性的关键一环。在每个步骤执行后，代理不只是盲目地进行下一步，而是会停下来“反思”：
- 结果评估： 检查当前步骤的执行结果是否符合预期。
- 问题识别： 如果结果包含错误信息、或与预期严重不符，则标识为问题。
- 决策：
- 如果一切顺利，继续执行下一个计划步骤。
- 如果出现问题，可以根据问题的性质采取不同的策略：
- 局部修正： 尝试用不同的参数或工具重新执行当前步骤。
- 重新规划： 如果问题比较严重或需要根本性的策略调整，则可以返回规划阶段，让规划器生成一个新的或修改过的计划。这可能涉及让LLM分析失败原因，并生成一个修复当前问题的子计划。
在我们的 execute_plan 方法中，通过调用 llm.generate(reflection_prompt) 来模拟这一过程。LLM会根据工具的输出，判断该步骤是否成功，并提供一些初步的反思。
Plan-and-Execute 与 ReAct 的深度对比
下表总结了“Plan-and-Execute”与“ReAct”模式在不同维度上的关键差异：
| 特性 | ReAct 模式 | Plan-and-Execute 模式 |
|---|---|---|
| 核心机制 | 思考-行动-观察 循环，即时响应 | 规划阶段 + 执行阶段，分工明确，先谋后动 |
| 任务类型 | 简单、交互性强、短期、探索性任务 | 复杂、多步骤、长周期、结构化、目标明确的任务 |
| 规划能力 | 弱，缺乏长期目标感，容易陷入局部最优 | 强，有明确的全局计划，确保任务连贯性 |
| 效率 | 可能因重复探索而低效 | 高效，按预设路径前进，减少不必要的决策 |
| 鲁棒性 | 面对复杂错误时恢复能力弱，容易“迷失” | 强，错误易定位，支持回溯、局部修正和重新规划 |
| 上下文管理 | 整个任务历史都需维持在上下文，易超限 | 规划阶段需全局上下文，执行阶段可聚焦当前步骤，减轻负担 |
| 可解释性 | 行为路径可能零散，不易理解 | 行为路径清晰，有计划可循，透明度高 |
| 决策粒度 | 细粒度，每一步都是一个完整的思考-行动循环 | 粗粒度（规划），细粒度（执行），分层决策 |
| 实现复杂性 | 相对简单 | 相对复杂，需要设计规划器、执行器、状态管理和反思机制 |
适用场景
-
ReAct 模式：
- 在线客服机器人： 快速响应用户提问，进行简短的查询和回复。
- 实时交互式调试： 根据用户反馈和程序输出，进行即时的问题诊断和修复尝试。
- 探索性数据分析： 在不知道确切路径的情况下，逐步探索数据。
- 简单问答系统： 根据问题进行一次或两次工具调用即可得到答案。
-
Plan-and-Execute 模式：
- 自主软件开发： 从需求分析到代码编写、测试、部署的完整流程。
- 科学实验设计与执行： 设计实验步骤、运行模拟、分析结果。
- 复杂数据处理管道： 数据清洗、转换、分析、可视化等多个串行步骤。
- 项目管理： 自动化项目计划的制定和执行跟踪。
- 多代理协作： 每个代理负责计划中的一个子任务。
混合方法：取长补短
值得注意的是，“Plan-and-Execute”和“ReAct”并非完全互斥。实际上，它们可以很好地结合起来，形成更强大的混合模式：
- P&E 外部框架，ReAct 内部实现： 可以在 P&E 模式的执行阶段，某个具体的步骤内部，使用 ReAct 模式来完成一个特定的子任务。例如，执行器在执行“编写代码”这一步骤时，可能会调用一个“代码生成代理”，而这个代理内部可能就是一个 ReAct 循环，它反复尝试生成代码、编译、测试、修正，直到代码通过。
- 动态规划： 规划器可以生成一个初步的计划，但在执行过程中，如果某个步骤的反馈需要更细致的、探索性的处理，可以临时切换到 ReAct 模式，处理完局部问题后再回到主计划。
这种混合方法能够充分利用两种模式的优势，既保持了宏观的连贯性和效率，又兼顾了局部任务的灵活性和适应性。
挑战与未来展望
“Plan-and-Execute”模式虽然前景广阔，但其实现和优化也面临诸多挑战：
- 规划质量： 计划的质量直接决定了任务的成败。LLM 在生成计划时可能出现“幻觉”、忽略细节、或生成次优计划。如何确保 LLM 生成的计划是逻辑严谨、全面且可执行的，是一个核心问题。未来的研究可能需要结合形式化验证、启发式规则或更复杂的规划算法。
- 上下文与记忆管理： 尽管 P&E 减轻了执行阶段的上下文压力，但规划器在生成计划时仍需要大量的上下文信息和长期记忆。如何有效地管理和检索这些信息，避免上下文窗口的限制，是持续的挑战。向量数据库、长期记忆模块是当前主要的解决方案。
- 工具集成与协调： 代理需要能够无缝地调用各种工具，并且理解它们的输入输出格式和使用限制。工具的数量和复杂性增加，对代理的工具选择和参数填充能力提出了更高要求。
- 错误处理与重新规划的智能性： 当执行失败时，如何智能地诊断问题、选择是局部修正还是全面重新规划，是衡量代理鲁棒性的关键。目前，这很大程度上依赖于 LLM 的反思能力，但仍有提升空间。
- 计算成本： 多个 LLM 调用（规划、执行、反思）会增加计算资源和 API 调用成本。优化 LLM 调用的次数、批处理请求、或使用更高效的小型模型进行特定任务，是重要的研究方向。
- 安全性与伦理： 赋予代理规划和执行复杂任务的能力，也带来了潜在的滥用风险。如何确保代理的行为符合人类价值观和安全规范，是设计P&E系统时必须考虑的伦理问题。
展望未来，“Plan-and-Execute”模式将继续演进。我们可能会看到：
- 更强大的自主规划器： 能够处理更抽象、更不确定的任务，甚至能够进行多层级的层次化规划。
- 自适应执行器： 能够根据环境动态调整执行策略，甚至在执行过程中学习并优化工具使用。
- 更智能的监控与修复机制： 能够主动识别潜在问题，并在问题发生前进行预防性调整。
- 与人类专家的紧密协作： 代理能够生成计划供人类审批，或在关键决策点寻求人类干预，实现人机协同的最高效率。
在构建能够承担更宏大、更复杂的自主任务的智能体时，“规划与执行”模式无疑是通向成功的关键基石。它为我们描绘了一个未来，在这个未来中，智能代理不仅仅是简单的响应者，更是能够深谋远虑、步步为营的强大执行者。