5.5 Agent 规划与执行
学习时长：3-4 周
Agent 规划与执行是实现复杂任务自动化的核心能力。本节覆盖从简单的任务分解到高级的多步骤规划、自我反思和动态调整策略。
5.5.1 任务规划基础
核心概念
任务规划（Task Planning）是将复杂目标分解为可执行的子任务序列，并确定执行顺序和依赖关系。
规划范式对比
| 范式 | 特点 | 优势 | 劣势 | 适用场景 |
|---|---|---|---|---|
| ReAct | 推理-行动循环 | 灵活、可解释 | 可能陷入循环 | 需要工具调用的任务 |
| Plan-and-Execute | 先规划后执行 | 结构清晰、高效 | 难以动态调整 | 明确的多步骤任务 |
| Tree of Thoughts | 树状搜索 | 探索多种方案 | 计算成本高 | 需要最优解的问题 |
| Reflexion | 反思-改进循环 | 自我优化 | 需要多轮迭代 | 需要高质量输出 |
1. ReAct 模式（Reasoning + Acting）
ReAct 是最经典的 Agent 规划模式，通过"思考-行动-观察"循环完成任务。
核心流程
1. Thought（思考）：分析当前状态，决定下一步
2. Action（行动）：调用工具或执行操作
3. Observation（观察）：获取行动结果
4. 重复 1-3，直到任务完成实现示例
python
from openai import OpenAI
from typing import List, Dict, Callable, Optional
import json
client = OpenAI()
class ReActAgent:
"""ReAct 模式 Agent"""
def __init__(self, tools: Dict[str, Callable], max_iterations: int = 10):
"""
Args:
tools: 可用工具字典 {tool_name: tool_function}
max_iterations: 最大迭代次数（防止无限循环）
"""
self.tools = tools
self.max_iterations = max_iterations
self.history: List[Dict] = []
def _build_prompt(self, task: str) -> str:
"""构建 ReAct 提示词"""
tool_descriptions = "\n".join([
f"- {name}: {func.__doc__}"
for name, func in self.tools.items()
])
prompt = f"""你是一个能够使用工具解决问题的 AI Agent。
可用工具：
{tool_descriptions}
任务：{task}
请使用以下格式进行推理和行动：
Thought: [分析当前情况，决定下一步]
Action: [工具名称]
Action Input: [工具输入，JSON格式]
你会收到：
Observation: [工具执行结果]
然后继续思考，直到完成任务。完成时输出：
Thought: 任务已完成
Final Answer: [最终答案]
开始！
"""
return prompt
def run(self, task: str) -> str:
"""执行任务"""
self.history = []
# 初始提示
messages = [
{"role": "system", "content": "你是一个使用 ReAct 模式的 AI Agent"},
{"role": "user", "content": self._build_prompt(task)}
]
for iteration in range(self.max_iterations):
print(f"\n{'='*50}")
print(f"迭代 {iteration + 1}/{self.max_iterations}")
print(f"{'='*50}")
# 1. LLM 推理
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.3,
max_tokens=500
)
agent_response = response.choices[0].message.content
print(f"\nAgent 输出：\n{agent_response}")
# 2. 解析输出
if "Final Answer:" in agent_response:
# 任务完成
final_answer = agent_response.split("Final Answer:")[-1].strip()
print(f"\n✅ 任务完成！")
print(f"最终答案：{final_answer}")
return final_answer
# 3. 提取 Action 和 Action Input
action, action_input = self._parse_action(agent_response)
if not action:
print("⚠️ 无法解析 Action，重试...")
messages.append({"role": "assistant", "content": agent_response})
messages.append({
"role": "user",
"content": "请明确指定 Action 和 Action Input（JSON格式）"
})
continue
# 4. 执行工具
observation = self._execute_tool(action, action_input)
print(f"\nObservation: {observation}")
# 5. 更新对话历史
messages.append({"role": "assistant", "content": agent_response})
messages.append({
"role": "user",
"content": f"Observation: {observation}"
})
# 记录历史
self.history.append({
"iteration": iteration + 1,
"thought": agent_response,
"action": action,
"action_input": action_input,
"observation": observation
})
return "❌ 达到最大迭代次数，任务未完成"
def _parse_action(self, text: str) -> tuple[Optional[str], Optional[str]]:
"""解析 Action 和 Action Input"""
try:
# 提取 Action
if "Action:" not in text:
return None, None
action_line = text.split("Action:")[1].split("\n")[0].strip()
# 提取 Action Input
if "Action Input:" not in text:
return action_line, "{}"
input_text = text.split("Action Input:")[1].split("\n")[0].strip()
return action_line, input_text
except Exception as e:
print(f"解析错误: {e}")
return None, None
def _execute_tool(self, tool_name: str, tool_input: str) -> str:
"""执行工具"""
if tool_name not in self.tools:
return f"错误：工具 '{tool_name}' 不存在。可用工具：{list(self.tools.keys())}"
try:
# 解析 JSON 输入
if tool_input.strip().startswith("{"):
input_dict = json.loads(tool_input)
result = self.tools[tool_name](**input_dict)
else:
result = self.tools[tool_name](tool_input)
return str(result)
except Exception as e:
return f"工具执行错误: {str(e)}"
# 定义工具
def search_web(query: str) -> str:
"""在网上搜索信息"""
# 模拟搜索（实际应调用搜索 API）
mock_results = {
"Python": "Python 是一门高级编程语言，由 Guido van Rossum 于 1991 年创建。",
"机器学习": "机器学习是人工智能的一个分支，通过数据和算法让计算机自动学习。",
"深度学习": "深度学习是机器学习的子领域，使用多层神经网络处理复杂数据。"
}
for key, value in mock_results.items():
if key in query:
return value
return f"未找到关于 '{query}' 的相关信息"
def calculate(expression: str) -> str:
"""计算数学表达式"""
try:
result = eval(expression)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误: {e}"
def get_current_time() -> str:
"""获取当前时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 使用示例
tools = {
"search_web": search_web,
"calculate": calculate,
"get_current_time": get_current_time
}
agent = ReActAgent(tools=tools, max_iterations=5)
# 测试任务
result = agent.run("搜索 Python 的相关信息，然后计算 2024 - 1991 等于多少年")输出示例
==================================================
迭代 1/5
==================================================
Agent 输出：
Thought: 我需要先搜索 Python 的信息，然后计算年份差
Action: search_web
Action Input: {"query": "Python"}
Observation: Python 是一门高级编程语言，由 Guido van Rossum 于 1991 年创建。
==================================================
迭代 2/5
==================================================
Agent 输出：
Thought: 我已经知道 Python 创建于 1991 年，现在需要计算 2024 - 1991
Action: calculate
Action Input: {"expression": "2024 - 1991"}
Observation: 2024 - 1991 = 33
==================================================
迭代 3/5
==================================================
Agent 输出：
Thought: 任务已完成
Final Answer: Python 是一门高级编程语言，由 Guido van Rossum 于 1991 年创建。从 1991 年到 2024 年，Python 已经发展了 33 年。
✅ 任务完成！
最终答案：Python 是一门高级编程语言，由 Guido van Rossum 于 1991 年创建。从 1991 年到 2024 年，Python 已经发展了 33 年。2. Plan-and-Execute 模式
先生成完整计划，再逐步执行，适合结构化任务。
核心流程
1. Planning（规划）：生成完整的任务分解计划
2. Execution（执行）：按顺序执行每个子任务
3. Verification（验证）：检查执行结果
4. Re-planning（重新规划）：如果失败，调整计划实现示例
python
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
class TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class SubTask:
"""子任务"""
id: int
description: str
tool: str
input: Dict
status: TaskStatus = TaskStatus.PENDING
result: Optional[str] = None
dependencies: List[int] = None # 依赖的任务 ID
def __post_init__(self):
if self.dependencies is None:
self.dependencies = []
class PlanAndExecuteAgent:
"""Plan-and-Execute 模式 Agent"""
def __init__(self, tools: Dict[str, Callable]):
self.tools = tools
self.plan: List[SubTask] = []
def run(self, task: str) -> str:
"""执行任务"""
print(f"📋 任务：{task}\n")
# 1. 生成计划
print("🧠 生成执行计划...")
self.plan = self._generate_plan(task)
if not self.plan:
return "❌ 无法生成有效计划"
self._print_plan()
# 2. 执行计划
print("\n🚀 开始执行计划...\n")
results = []
for subtask in self.plan:
# 检查依赖是否完成
if not self._check_dependencies(subtask):
subtask.status = TaskStatus.FAILED
print(f"❌ 子任务 {subtask.id} 失败：依赖未满足")
continue
# 执行子任务
subtask.status = TaskStatus.IN_PROGRESS
print(f"▶️ 执行子任务 {subtask.id}: {subtask.description}")
result = self._execute_subtask(subtask)
subtask.result = result
if "错误" in result or "失败" in result:
subtask.status = TaskStatus.FAILED
print(f"❌ 失败：{result}")
# 尝试重新规划
if self._should_replan(subtask):
print("\n🔄 重新规划...")
return self.run(task) # 递归重试
else:
return f"任务失败：{result}"
else:
subtask.status = TaskStatus.COMPLETED
print(f"✅ 完成：{result}\n")
results.append(result)
# 3. 生成最终答案
final_answer = self._synthesize_results(task, results)
print(f"\n🎉 任务完成！\n最终答案：{final_answer}")
return final_answer
def _generate_plan(self, task: str) -> List[SubTask]:
"""生成任务计划"""
tool_descriptions = "\n".join([
f"- {name}: {func.__doc__}"
for name, func in self.tools.items()
])
prompt = f"""将以下任务分解为可执行的子任务序列。
任务：{task}
可用工具：
{tool_descriptions}
请以 JSON 数组格式返回计划，每个子任务包含：
- id: 子任务编号（从1开始）
- description: 子任务描述
- tool: 使用的工具名称
- input: 工具输入（JSON对象）
- dependencies: 依赖的子任务 ID 列表（如果有）
示例：
[
{{"id": 1, "description": "搜索信息", "tool": "search_web", "input": {{"query": "Python"}}, "dependencies": []}},
{{"id": 2, "description": "计算年份", "tool": "calculate", "input": {{"expression": "2024-1991"}}, "dependencies": [1]}}
]
只返回 JSON 数组，不要其他文字。"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.3
)
try:
plan_data = json.loads(response.choices[0].message.content)
# 处理可能的包装
if isinstance(plan_data, dict) and "plan" in plan_data:
plan_data = plan_data["plan"]
elif isinstance(plan_data, dict) and "tasks" in plan_data:
plan_data = plan_data["tasks"]
plan = []
for item in plan_data:
subtask = SubTask(
id=item["id"],
description=item["description"],
tool=item["tool"],
input=item["input"],
dependencies=item.get("dependencies", [])
)
plan.append(subtask)
return plan
except Exception as e:
print(f"计划生成错误: {e}")
return []
def _check_dependencies(self, subtask: SubTask) -> bool:
"""检查依赖任务是否完成"""
for dep_id in subtask.dependencies:
dep_task = next((t for t in self.plan if t.id == dep_id), None)
if not dep_task or dep_task.status != TaskStatus.COMPLETED:
return False
return True
def _execute_subtask(self, subtask: SubTask) -> str:
"""执行子任务"""
if subtask.tool not in self.tools:
return f"错误：工具 '{subtask.tool}' 不存在"
try:
result = self.tools[subtask.tool](**subtask.input)
return str(result)
except Exception as e:
return f"执行错误: {str(e)}"
def _should_replan(self, failed_task: SubTask) -> bool:
"""判断是否需要重新规划（简化版）"""
# 实际应用中应该更智能地判断
return False
def _synthesize_results(self, task: str, results: List[str]) -> str:
"""综合所有结果生成最终答案"""
results_text = "\n".join([f"- {r}" for r in results])
prompt = f"""基于以下子任务的执行结果，回答原始问题。
原始问题：{task}
执行结果：
{results_text}
请给出简洁明了的最终答案："""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content
def _print_plan(self):
"""打印计划"""
print("\n📝 执行计划：")
for subtask in self.plan:
deps = f" (依赖: {subtask.dependencies})" if subtask.dependencies else ""
print(f" {subtask.id}. {subtask.description} [{subtask.tool}]{deps}")
# 使用示例
agent = PlanAndExecuteAgent(tools=tools)
result = agent.run("搜索深度学习的定义，然后计算 100 * 50 的结果")输出示例
📋 任务：搜索深度学习的定义，然后计算 100 * 50 的结果
🧠 生成执行计划...
📝 执行计划：
1. 搜索深度学习定义 [search_web]
2. 计算乘法 [calculate] (依赖: [1])
🚀 开始执行计划...
▶️ 执行子任务 1: 搜索深度学习定义
✅ 完成：深度学习是机器学习的子领域，使用多层神经网络处理复杂数据。
▶️ 执行子任务 2: 计算乘法
✅ 完成：100 * 50 = 5000
🎉 任务完成！
最终答案：深度学习是机器学习的子领域，使用多层神经网络处理复杂数据。100 乘以 50 等于 5000。5.5.2 高级规划技术
1. Hierarchical Planning（层次化规划）
将复杂任务分解为多层子任务，适合大型项目。
python
from typing import List, Dict, Optional
from dataclasses import dataclass, field
@dataclass
class HierarchicalTask:
"""层次化任务节点"""
id: str
description: str
level: int # 层级（0=根任务）
subtasks: List['HierarchicalTask'] = field(default_factory=list)
status: TaskStatus = TaskStatus.PENDING
result: Optional[str] = None
class HierarchicalPlanner:
"""层次化规划器"""
def __init__(self, max_depth: int = 3):
self.max_depth = max_depth
def decompose(self, task: str, level: int = 0) -> HierarchicalTask:
"""递归分解任务"""
if level >= self.max_depth:
# 达到最大深度，返回原子任务
return HierarchicalTask(
id=f"task_{level}_{hash(task) % 1000}",
description=task,
level=level
)
# 调用 LLM 分解任务
subtasks_desc = self._llm_decompose(task, level)
root = HierarchicalTask(
id=f"task_{level}_root",
description=task,
level=level
)
for i, subtask_desc in enumerate(subtasks_desc):
# 递归分解子任务
subtask = self.decompose(subtask_desc, level + 1)
root.subtasks.append(subtask)
return root
def _llm_decompose(self, task: str, level: int) -> List[str]:
"""使用 LLM 分解任务"""
prompt = f"""将以下任务分解为 2-4 个子任务（第 {level+1} 层）：
任务：{task}
要求：
1. 子任务应该是可独立执行的
2. 子任务之间有逻辑顺序
3. 如果任务已经足够简单，返回空列表
以 JSON 数组格式返回子任务描述：
["子任务1", "子任务2", ...]
只返回 JSON，不要其他文字。"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
try:
content = response.choices[0].message.content.strip()
# 移除可能的 markdown 代码块标记
if content.startswith("```"):
content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
subtasks = json.loads(content)
return subtasks if isinstance(subtasks, list) else []
except:
return []
def print_tree(self, task: HierarchicalTask, indent: int = 0):
"""打印任务树"""
prefix = " " * indent
status_icon = {
TaskStatus.PENDING: "⏳",
TaskStatus.IN_PROGRESS: "▶️",
TaskStatus.COMPLETED: "✅",
TaskStatus.FAILED: "❌"
}[task.status]
print(f"{prefix}{status_icon} [{task.level}] {task.description}")
for subtask in task.subtasks:
self.print_tree(subtask, indent + 1)
# 使用示例
planner = HierarchicalPlanner(max_depth=2)
task_tree = planner.decompose("开发一个 AI 聊天机器人")
print("📊 层次化任务分解：\n")
planner.print_tree(task_tree)输出示例
📊 层次化任务分解：
⏳ [0] 开发一个 AI 聊天机器人
⏳ [1] 设计系统架构
⏳ [2] 选择技术栈
⏳ [2] 设计数据库模型
⏳ [1] 实现核心功能
⏳ [2] 集成 LLM API
⏳ [2] 实现对话管理
⏳ [1] 部署与测试
⏳ [2] 编写单元测试
⏳ [2] 部署到云服务器2. Dynamic Re-planning（动态重新规划）
根据执行结果动态调整计划。
python
class DynamicAgent:
"""支持动态重新规划的 Agent"""
def __init__(self, tools: Dict[str, Callable], max_replans: int = 3):
self.tools = tools
self.max_replans = max_replans
self.replan_count = 0
def run(self, task: str) -> str:
"""执行任务（支持动态重新规划）"""
plan = self._generate_plan(task)
execution_history = []
while True:
result, success = self._execute_plan(plan, execution_history)
if success:
return result
# 执行失败，判断是否重新规划
if self.replan_count >= self.max_replans:
return f"❌ 达到最大重新规划次数，任务失败"
print(f"\n🔄 执行失败，进行第 {self.replan_count + 1} 次重新规划...")
# 基于执行历史重新规划
plan = self._replan(task, execution_history)
self.replan_count += 1
def _execute_plan(
self,
plan: List[SubTask],
history: List[Dict]
) -> tuple[str, bool]:
"""执行计划，返回 (结果, 是否成功)"""
results = []
for subtask in plan:
print(f"\n▶️ 执行：{subtask.description}")
result = self._execute_subtask(subtask)
history.append({
"subtask": subtask.description,
"result": result,
"success": "错误" not in result
})
if "错误" in result:
print(f"❌ 失败：{result}")
return result, False
print(f"✅ 成功：{result}")
results.append(result)
final_answer = " ".join(results)
return final_answer, True
def _replan(self, task: str, history: List[Dict]) -> List[SubTask]:
"""基于执行历史重新规划"""
history_text = "\n".join([
f"- {h['subtask']}: {'成功' if h['success'] else '失败'} ({h['result']})"
for h in history
])
prompt = f"""之前的执行计划失败了，请生成新的计划。
原始任务：{task}
执行历史：
{history_text}
请分析失败原因，生成改进的计划。以 JSON 数组格式返回。"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.5 # 增加温度以探索新方案
)
# 解析新计划（简化处理）
return self._generate_plan(task)
def _generate_plan(self, task: str) -> List[SubTask]:
"""生成计划（复用之前的实现）"""
# ... 同 PlanAndExecuteAgent
pass
def _execute_subtask(self, subtask: SubTask) -> str:
"""执行子任务（复用之前的实现）"""
# ... 同 PlanAndExecuteAgent
pass3. Multi-Agent Planning（多 Agent 协作规划）
多个专业 Agent 协作完成复杂任务。
python
from typing import List, Dict, Optional
from abc import ABC, abstractmethod
class SpecializedAgent(ABC):
"""专业 Agent 基类"""
def __init__(self, name: str, expertise: str):
self.name = name
self.expertise = expertise
@abstractmethod
def can_handle(self, task: str) -> bool:
"""判断是否能处理该任务"""
pass
@abstractmethod
def execute(self, task: str) -> str:
"""执行任务"""
pass
class ResearchAgent(SpecializedAgent):
"""研究型 Agent（擅长信息检索）"""
def __init__(self):
super().__init__("研究员", "信息检索与分析")
def can_handle(self, task: str) -> bool:
keywords = ["搜索", "查找", "研究", "调查", "信息"]
return any(kw in task for kw in keywords)
def execute(self, task: str) -> str:
# 调用搜索工具
return f"[{self.name}] 已完成研究：{task}"
class CalculatorAgent(SpecializedAgent):
"""计算型 Agent（擅长数学计算）"""
def __init__(self):
super().__init__("计算器", "数学计算")
def can_handle(self, task: str) -> bool:
keywords = ["计算", "求和", "乘法", "除法", "数学"]
return any(kw in task for kw in keywords)
def execute(self, task: str) -> str:
# 调用计算工具
return f"[{self.name}] 计算完成：{task}"
class WriterAgent(SpecializedAgent):
"""写作型 Agent（擅长内容生成）"""
def __init__(self):
super().__init__("作家", "内容创作")
def can_handle(self, task: str) -> bool:
keywords = ["写", "生成", "创作", "总结", "报告"]
return any(kw in task for kw in keywords)
def execute(self, task: str) -> str:
# 调用 LLM 生成内容
return f"[{self.name}] 已完成写作：{task}"
class MultiAgentCoordinator:
"""多 Agent 协调器"""
def __init__(self, agents: List[SpecializedAgent]):
self.agents = agents
def run(self, task: str) -> str:
"""协调多个 Agent 完成任务"""
print(f"📋 任务：{task}\n")
# 1. 任务分解
subtasks = self._decompose_task(task)
print("📝 任务分解：")
for i, st in enumerate(subtasks, 1):
print(f" {i}. {st}")
# 2. 任务分配
print("\n👥 Agent 分配：")
assignments = self._assign_tasks(subtasks)
# 3. 执行任务
print("\n🚀 开始执行：\n")
results = []
for subtask, agent in assignments:
print(f"▶️ {agent.name} 执行：{subtask}")
result = agent.execute(subtask)
print(f"✅ {result}\n")
results.append(result)
# 4. 综合结果
final_answer = self._synthesize(task, results)
print(f"🎉 任务完成！\n{final_answer}")
return final_answer
def _decompose_task(self, task: str) -> List[str]:
"""分解任务"""
prompt = f"""将以下任务分解为可由专业 Agent 执行的子任务：
任务：{task}
可用 Agent：
- 研究员：信息检索与分析
- 计算器：数学计算
- 作家：内容创作
以 JSON 数组格式返回子任务列表。"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
try:
content = response.choices[0].message.content.strip()
if content.startswith("```"):
content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
subtasks = json.loads(content)
return subtasks if isinstance(subtasks, list) else [task]
except:
return [task]
def _assign_tasks(
self,
subtasks: List[str]
) -> List[tuple[str, SpecializedAgent]]:
"""将子任务分配给合适的 Agent"""
assignments = []
for subtask in subtasks:
# 找到能处理该任务的 Agent
suitable_agents = [a for a in self.agents if a.can_handle(subtask)]
if suitable_agents:
agent = suitable_agents[0] # 选择第一个匹配的
else:
agent = self.agents[0] # 默认分配给第一个 Agent
assignments.append((subtask, agent))
print(f" • {subtask} → {agent.name}")
return assignments
def _synthesize(self, task: str, results: List[str]) -> str:
"""综合所有结果"""
results_text = "\n".join(results)
return f"针对任务「{task}」，各 Agent 协作完成：\n{results_text}"
# 使用示例
agents = [
ResearchAgent(),
CalculatorAgent(),
WriterAgent()
]
coordinator = MultiAgentCoordinator(agents)
result = coordinator.run("研究 Python 的历史，计算它诞生多少年了，然后写一篇 200 字的总结")输出示例
📋 任务：研究 Python 的历史，计算它诞生多少年了，然后写一篇 200 字的总结
📝 任务分解：
1. 研究 Python 的历史和诞生年份
2. 计算从诞生到现在的年数
3. 写一篇 200 字的总结
👥 Agent 分配：
• 研究 Python 的历史和诞生年份 → 研究员
• 计算从诞生到现在的年数 → 计算器
• 写一篇 200 字的总结 → 作家
🚀 开始执行：
▶️ 研究员 执行：研究 Python 的历史和诞生年份
✅ [研究员] 已完成研究：研究 Python 的历史和诞生年份
▶️ 计算器 执行：计算从诞生到现在的年数
✅ [计算器] 计算完成：计算从诞生到现在的年数
▶️ 作家 执行：写一篇 200 字的总结
✅ [作家] 已完成写作：写一篇 200 字的总结
🎉 任务完成！
针对任务「研究 Python 的历史，计算它诞生多少年了，然后写一篇 200 字的总结」，各 Agent 协作完成：
[研究员] 已完成研究：研究 Python 的历史和诞生年份
[计算器] 计算完成：计算从诞生到现在的年数
[作家] 已完成写作：写一篇 200 字的总结5.5.3 自我反思与优化
1. Reflexion 模式
通过自我反思不断改进输出质量。
python
from typing import List, Dict, Optional
class ReflexionAgent:
"""Reflexion 模式 Agent：执行-反思-改进循环"""
def __init__(self, max_iterations: int = 3):
self.max_iterations = max_iterations
self.history: List[Dict] = []
def run(self, task: str) -> str:
"""执行任务（带自我反思）"""
print(f"📋 任务：{task}\n")
current_solution = None
for iteration in range(1, self.max_iterations + 1):
print(f"{'='*50}")
print(f"迭代 {iteration}/{self.max_iterations}")
print(f"{'='*50}\n")
# 1. 生成解决方案
solution = self._generate_solution(task, current_solution)
print(f"💡 解决方案：\n{solution}\n")
# 2. 自我评估
evaluation = self._self_evaluate(task, solution)
print(f"🔍 自我评估：\n{evaluation}\n")
# 3. 判断是否满意
if self._is_satisfactory(evaluation):
print("✅ 解决方案满意，任务完成！")
return solution
# 4. 生成反思
reflection = self._reflect(task, solution, evaluation)
print(f"💭 反思：\n{reflection}\n")
# 记录历史
self.history.append({
"iteration": iteration,
"solution": solution,
"evaluation": evaluation,
"reflection": reflection
})
current_solution = solution
print("⚠️ 达到最大迭代次数")
return current_solution
def _generate_solution(
self,
task: str,
previous_solution: Optional[str]
) -> str:
"""生成解决方案"""
if previous_solution is None:
# 首次生成
prompt = f"请解决以下任务：\n\n{task}"
else:
# 基于反思改进
last_reflection = self.history[-1]["reflection"]
prompt = f"""请改进之前的解决方案。
任务：{task}
之前的方案：
{previous_solution}
反思意见：
{last_reflection}
请生成改进后的方案："""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.7
)
return response.choices[0].message.content
def _self_evaluate(self, task: str, solution: str) -> str:
"""自我评估解决方案"""
prompt = f"""请评估以下解决方案的质量。
任务：{task}
解决方案：
{solution}
评估维度：
1. 完整性：是否完整解决了任务？
2. 准确性：信息是否准确？
3. 清晰度：表达是否清晰易懂？
4. 创新性：是否有独特见解？
请给出评分（1-10分）和详细评价："""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content
def _is_satisfactory(self, evaluation: str) -> bool:
"""判断是否满意（简化版：检查评分）"""
# 提取评分
import re
scores = re.findall(r'(\d+)/10|(\d+)分', evaluation)
if scores:
# 取第一个评分
score = int(scores[0][0] or scores[0][1])
return score >= 8 # 8分以上认为满意
# 如果没有明确评分，检查关键词
positive_keywords = ["优秀", "很好", "满意", "完美"]
return any(kw in evaluation for kw in positive_keywords)
def _reflect(self, task: str, solution: str, evaluation: str) -> str:
"""生成反思"""
prompt = f"""基于评估结果，反思如何改进解决方案。
任务：{task}
当前方案：
{solution}
评估结果：
{evaluation}
请提出具体的改进建议："""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.5
)
return response.choices[0].message.content
# 使用示例
agent = ReflexionAgent(max_iterations=3)
result = agent.run("写一篇关于 AI Agent 的技术博客，要求通俗易懂且有深度")2. Self-Consistency with Voting（投票自洽）
生成多个方案，投票选出最佳。
python
from collections import Counter
from typing import List, Dict
class SelfConsistencyAgent:
"""自洽性 Agent：生成多个方案并投票"""
def __init__(self, num_samples: int = 5):
self.num_samples = num_samples
def run(self, task: str) -> str:
"""执行任务（多样本投票）"""
print(f"📋 任务：{task}\n")
print(f"🎲 生成 {self.num_samples} 个候选方案...\n")
# 1. 生成多个候选方案
candidates = []
for i in range(self.num_samples):
solution = self._generate_solution(task, temperature=0.8)
candidates.append(solution)
print(f"方案 {i+1}：{solution[:100]}...")
print(f"\n🗳️ 投票选出最佳方案...\n")
# 2. 提取核心答案（用于投票）
answers = [self._extract_answer(sol) for sol in candidates]
# 3. 投票
vote_counts = Counter(answers)
most_common = vote_counts.most_common()
print("投票结果：")
for answer, count in most_common:
print(f" • {answer}: {count} 票")
# 4. 选择得票最多的答案
best_answer = most_common[0][0]
best_solution = candidates[answers.index(best_answer)]
print(f"\n✅ 最终方案（{most_common[0][1]} 票）：")
print(best_solution)
return best_solution
def _generate_solution(self, task: str, temperature: float) -> str:
"""生成解决方案"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": task}],
temperature=temperature,
max_tokens=300
)
return response.choices[0].message.content
def _extract_answer(self, solution: str) -> str:
"""提取核心答案（用于投票）"""
# 简化：提取最后一句话作为核心答案
sentences = solution.split("。")
return sentences[-2] if len(sentences) > 1 else solution[:50]
# 使用示例
agent = SelfConsistencyAgent(num_samples=5)
result = agent.run("2024 年是闰年吗？请给出判断依据。")5.5.4 实战案例：复杂任务自动化
案例：自动化研究报告生成
python
"""
任务：给定一个主题，自动生成研究报告
步骤：
1. 信息检索（搜索相关资料）
2. 内容分析（提取关键信息）
3. 结构规划（设计报告大纲）
4. 内容生成（撰写各章节）
5. 质量检查（自我评估与优化）
"""
class ResearchReportAgent:
"""研究报告生成 Agent"""
def __init__(self):
self.report_structure = []
self.content = {}
def generate_report(self, topic: str) -> str:
"""生成研究报告"""
print(f"📊 生成研究报告：{topic}\n")
# 1. 信息检索
print("🔍 步骤 1/5：信息检索")
research_data = self._research(topic)
print(f"✅ 检索到 {len(research_data)} 条相关信息\n")
# 2. 内容分析
print("📖 步骤 2/5：内容分析")
key_points = self._analyze(research_data)
print(f"✅ 提取 {len(key_points)} 个关键点\n")
# 3. 结构规划
print("📝 步骤 3/5：结构规划")
self.report_structure = self._plan_structure(topic, key_points)
print("✅ 报告大纲：")
for i, section in enumerate(self.report_structure, 1):
print(f" {i}. {section}")
print()
# 4. 内容生成
print("✍️ 步骤 4/5：内容生成")
for section in self.report_structure:
content = self._generate_section(section, key_points)
self.content[section] = content
print(f"✅ 完成：{section}")
print()
# 5. 质量检查
print("🔍 步骤 5/5：质量检查")
final_report = self._assemble_report(topic)
quality_score = self._quality_check(final_report)
print(f"✅ 质量评分：{quality_score}/10\n")
if quality_score < 7:
print("⚠️ 质量不达标，进行优化...")
final_report = self._optimize_report(final_report)
print("🎉 报告生成完成！\n")
return final_report
def _research(self, topic: str) -> List[str]:
"""模拟信息检索"""
# 实际应调用搜索 API
return [
f"{topic} 的定义和背景",
f"{topic} 的核心技术",
f"{topic} 的应用场景",
f"{topic} 的发展趋势"
]
def _analyze(self, data: List[str]) -> List[str]:
"""分析提取关键点"""
prompt = f"""分析以下信息，提取关键要点：
{chr(10).join(f'- {d}' for d in data)}
以 JSON 数组格式返回关键点列表。"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
try:
content = response.choices[0].message.content.strip()
if content.startswith("```"):
content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
return json.loads(content)
except:
return data
def _plan_structure(self, topic: str, key_points: List[str]) -> List[str]:
"""规划报告结构"""
return [
"摘要",
"引言",
"核心内容",
"应用案例",
"未来展望",
"结论"
]
def _generate_section(self, section: str, key_points: List[str]) -> str:
"""生成章节内容"""
prompt = f"""为研究报告撰写「{section}」章节。
参考要点：
{chr(10).join(f'- {kp}' for kp in key_points[:3])}
要求：
- 字数：200-300字
- 语言专业、逻辑清晰
- 包含具体数据或案例
内容："""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.7
)
return response.choices[0].message.content
def _assemble_report(self, topic: str) -> str:
"""组装完整报告"""
report = f"# {topic} 研究报告\n\n"
for section in self.report_structure:
report += f"## {section}\n\n"
report += self.content[section] + "\n\n"
return report
def _quality_check(self, report: str) -> int:
"""质量检查（返回评分）"""
prompt = f"""评估以下研究报告的质量，给出 1-10 分的评分。
{report[:500]}...
评估标准：
- 结构完整性
- 内容专业性
- 逻辑清晰度
- 语言流畅度
只返回数字评分（如：8）"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
try:
return int(response.choices[0].message.content.strip())
except:
return 7
def _optimize_report(self, report: str) -> str:
"""优化报告"""
# 简化：这里只返回原报告
# 实际应该调用 LLM 进行改进
return report
# 使用示例
agent = ResearchReportAgent()
report = agent.generate_report("大语言模型的应用")
print("="*60)
print(report[:500])
print("...")5.5.5 学习资源
推荐论文
- ReAct: ReAct: Synergizing Reasoning and Acting in Language Models
- Reflexion: Reflexion: Language Agents with Verbal Reinforcement Learning
- Tree of Thoughts: Tree of Thoughts: Deliberate Problem Solving with Large Language Models
开源框架
- LangGraph: 状态机式 Agent 框架（LangChain 出品）
- AutoGPT: 自主 Agent 先驱项目
- BabyAGI: 任务驱动的自主 Agent
实战练习
- 实现一个 ReAct Agent，完成多步骤数学推理
- 构建 Plan-and-Execute Agent，自动化数据分析流程
- 开发多 Agent 系统，模拟软件开发团队协作
- 实现 Reflexion Agent，自动优化代码质量
关键要点
- ✅ ReAct 是基础：理解"思考-行动-观察"循环
- ✅ 规划要灵活：支持动态调整，避免僵化执行
- ✅ 自我反思很重要：通过反思提升输出质量
- ✅ 多 Agent 协作：复杂任务分工合作更高效
- ✅ 防止无限循环：设置最大迭代次数和超时机制
- ✅ 记录执行历史：便于调试和优化规划策略
常见陷阱
- ❌ 规划过于复杂，执行效率低
- ❌ 缺少错误处理，一步失败导致全盘崩溃
- ❌ 无限循环，Agent 陷入重复行动
- ❌ 忽视成本控制，频繁调用 LLM 导致费用激增
下一步 完成本节学习后，结合前面的工具调用（5.3）和记忆管理（5.4），你已经具备构建完整 AI Agent 系统的能力。建议实战项目：
- 自动化客服系统（多轮对话 + 工具调用 + 记忆）
- 代码审查助手（规划 + 执行 + 反思）
- 个人助理 Agent（任务管理 + 日程安排 + 信息检索）