阶段 3：任务分解与多步推理 Agent
📖 本阶段目标
- 理解任务分解的原理和方法
- 掌握 ReAct 模式的实现
- 学习让 Agent 进行多步骤推理
- 实践构建能够自主规划的工作流 Agent
🧩 任务分解核心概念
为什么需要任务分解？
复杂任务通常无法一步完成，需要： 1. 分解：将大任务拆解为小步骤 2. 规划：确定执行顺序 3. 执行：逐步完成各个子任务 4. 整合：汇总结果
任务分解模式
1. 线性分解
任务：准备一份市场报告
↓
步骤1：收集市场数据 → 步骤2：分析数据 → 步骤3：生成报告
2. 并行分解
任务：准备项目方案
↓
├─ 技术方案（并行）
├─ 成本估算（并行）
└─ 时间规划（并行）
↓
整合为完整方案
3. 层级分解
任务：开发新功能
├─ 需求分析
│ ├─ 用户调研
│ └─ 竞品分析
├─ 技术设计
│ ├─ 架构设计
│ └─ 接口设计
└─ 实施计划
🔄 ReAct 模式深入
ReAct 循环
# examples/stage3/react_agent.py
from typing import List, Dict, Any
from openai import OpenAI
import json
import re
client = OpenAI()
class ReActAgent:
"""ReAct 模式 Agent"""
def __init__(self, max_iterations=10):
self.max_iterations = max_iterations
self.tools = self._define_tools()
self.thought_history = []
def _define_tools(self):
"""定义可用工具"""
return {
"search": self.search_tool,
"calculate": self.calculate_tool,
"lookup": self.lookup_tool,
}
def search_tool(self, query: str) -> str:
"""搜索工具（模拟）"""
# 实际应该调用真实搜索API
mock_results = {
"Python": "Python是一种高级编程语言，由Guido van Rossum创建于1991年。",
"北京": "北京是中国的首都，人口超过2000万。",
"AI": "人工智能是计算机科学的一个分支，研究如何让机器模拟人类智能。"
}
for key in mock_results:
if key in query:
return mock_results[key]
return f"关于'{query}'的搜索结果：暂无相关信息。"
def calculate_tool(self, expression: str) -> str:
"""计算工具"""
try:
result = eval(expression, {"__builtins__": {}}, {})
return str(result)
except:
return "计算错误"
def lookup_tool(self, keyword: str) -> str:
"""查找工具（在之前的搜索结果中查找）"""
# 简化实现
return f"查找到关于'{keyword}'的信息"
def parse_action(self, text: str) -> tuple:
"""解析 LLM 输出的行动"""
# 匹配 Action: tool_name[arguments]
action_pattern = r"Action:\s*(\w+)\[(.*?)\]"
match = re.search(action_pattern, text)
if match:
tool_name = match.group(1)
arguments = match.group(2).strip('"\'')
return tool_name, arguments
return None, None
def run(self, question: str) -> str:
"""运行 ReAct 循环"""
prompt = f"""
你是一个使用 ReAct 模式的智能助手。
可用工具：
- search[query]: 搜索信息
- calculate[expression]: 计算数学表达式
- lookup[keyword]: 在之前的结果中查找关键词
请按以下格式思考和行动：
Thought: [你的思考过程]
Action: [工具名称][参数]
Observation: [工具返回的结果，由系统填充]
... (重复 Thought/Action/Observation)
Thought: 我现在知道最终答案了
Final Answer: [最终答案]
问题: {question}
开始！
Thought:
"""
messages = [{"role": "user", "content": prompt}]
full_response = ""
for i in range(self.max_iterations):
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
temperature=0.2,
max_tokens=500
)
thought = response.choices[0].message.content
full_response += thought + "\n"
# 检查是否得到最终答案
if "Final Answer:" in thought:
final_answer = thought.split("Final Answer:")[-1].strip()
return {
"answer": final_answer,
"process": full_response,
"iterations": i + 1
}
# 解析行动
tool_name, arguments = self.parse_action(thought)
if tool_name and tool_name in self.tools:
# 执行工具
observation = self.tools[tool_name](arguments)
full_response += f"Observation: {observation}\n"
# 更新消息
messages.append({"role": "assistant", "content": thought})
messages.append({
"role": "user",
"content": f"Observation: {observation}\nThought:"
})
else:
# 如果没有找到有效的行动，提示继续
messages.append({"role": "assistant", "content": thought})
messages.append({
"role": "user",
"content": "请继续思考或给出最终答案。\nThought:"
})
return {
"answer": "达到最大迭代次数，未能得到答案",
"process": full_response,
"iterations": self.max_iterations
}
# 测试
if __name__ == "__main__":
agent = ReActAgent(max_iterations=5)
question = "Python语言是什么时候创建的？创建者是谁？"
result = agent.run(question)
print("=" * 50)
print("问题:", question)
print("=" * 50)
print("推理过程:")
print(result["process"])
print("=" * 50)
print("最终答案:", result["answer"])
print("迭代次数:", result["iterations"])
🎯 实践项目
项目 1：自动资料收集 Agent
# examples/stage3/research_agent.py
from openai import OpenAI
import json
from typing import List, Dict
client = OpenAI()
class ResearchAgent:
"""自动研究和资料收集 Agent"""
def __init__(self):
self.research_results = []
self.sources = []
def decompose_research_task(self, topic: str) -> List[str]:
"""分解研究任务为子问题"""
prompt = f"""
研究主题: {topic}
请将这个研究主题分解为3-5个具体的子问题，这些问题的答案能够全面覆盖该主题。
以JSON格式返回：
{{
"subtopics": [
"子问题1",
"子问题2",
...
]
}}
"""
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result.get("subtopics", [])
def research_subtopic(self, subtopic: str) -> Dict:
"""研究单个子主题"""
# 这里应该调用真实的搜索API或知识库
# 模拟搜索过程
prompt = f"""
请简要回答以下问题（100-200字）：
{subtopic}
请给出准确、专业的回答。
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
answer = response.choices[0].message.content
return {
"question": subtopic,
"answer": answer,
"source": "GPT-4 知识库"
}
def synthesize_results(self, topic: str, results: List[Dict]) -> str:
"""综合研究结果"""
results_text = "\n\n".join([
f"Q: {r['question']}\nA: {r['answer']}"
for r in results
])
prompt = f"""
研究主题: {topic}
以下是对该主题的分项研究结果：
{results_text}
请将这些信息综合成一份连贯、结构化的研究报告（300-500字），包括：
1. 概述
2. 主要发现
3. 总结
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.5
)
return response.choices[0].message.content
def conduct_research(self, topic: str) -> Dict:
"""执行完整的研究流程"""
print(f"📚 开始研究: {topic}\n")
# 步骤1: 任务分解
print("步骤1: 分解研究任务...")
subtopics = self.decompose_research_task(topic)
print(f"已分解为 {len(subtopics)} 个子问题：")
for i, st in enumerate(subtopics, 1):
print(f" {i}. {st}")
print()
# 步骤2: 逐个研究子主题
print("步骤2: 研究各个子主题...")
results = []
for i, subtopic in enumerate(subtopics, 1):
print(f" 研究中 ({i}/{len(subtopics)}): {subtopic[:50]}...")
result = self.research_subtopic(subtopic)
results.append(result)
print()
# 步骤3: 综合报告
print("步骤3: 生成综合报告...")
report = self.synthesize_results(topic, results)
print()
return {
"topic": topic,
"subtopics": subtopics,
"detailed_results": results,
"final_report": report
}
# 测试
if __name__ == "__main__":
agent = ResearchAgent()
topic = "AI Agent的核心技术"
result = agent.conduct_research(topic)
print("=" * 60)
print("📊 研究报告")
print("=" * 60)
print(result["final_report"])
print("=" * 60)
项目 2：工作流助手
# examples/stage3/workflow_agent.py
from openai import OpenAI
from typing import List, Dict, Any
from datetime import datetime
import json
client = OpenAI()
class WorkflowAgent:
"""工作流自动化助手"""
def __init__(self):
self.workflow_history = []
self.available_actions = {
"send_email": self.send_email,
"create_task": self.create_task,
"schedule_meeting": self.schedule_meeting,
"generate_document": self.generate_document,
"analyze_data": self.analyze_data
}
def send_email(self, recipient: str, subject: str, body: str) -> Dict:
"""发送邮件（模拟）"""
print(f"📧 发送邮件到: {recipient}")
print(f" 主题: {subject}")
return {"status": "sent", "timestamp": datetime.now().isoformat()}
def create_task(self, title: str, description: str, assignee: str) -> Dict:
"""创建任务（模拟）"""
print(f"✅ 创建任务: {title}")
print(f" 负责人: {assignee}")
return {"task_id": f"TASK-{len(self.workflow_history)+1}", "status": "created"}
def schedule_meeting(self, title: str, participants: List[str], duration: int) -> Dict:
"""安排会议（模拟）"""
print(f"📅 安排会议: {title}")
print(f" 参与者: {', '.join(participants)}")
return {"meeting_id": f"MEET-{len(self.workflow_history)+1}", "status": "scheduled"}
def generate_document(self, doc_type: str, content: str) -> Dict:
"""生成文档（模拟）"""
print(f"📄 生成文档: {doc_type}")
return {"doc_id": f"DOC-{len(self.workflow_history)+1}", "status": "generated"}
def analyze_data(self, data_source: str, analysis_type: str) -> Dict:
"""分析数据（模拟）"""
print(f"📊 分析数据: {data_source}")
return {"analysis_id": f"ANAL-{len(self.workflow_history)+1}", "status": "completed"}
def plan_workflow(self, user_request: str) -> List[Dict]:
"""规划工作流步骤"""
prompt = f"""
用户请求: {user_request}
可用操作:
1. send_email(recipient, subject, body) - 发送邮件
2. create_task(title, description, assignee) - 创建任务
3. schedule_meeting(title, participants, duration) - 安排会议
4. generate_document(doc_type, content) - 生成文档
5. analyze_data(data_source, analysis_type) - 分析数据
请将用户请求分解为具体的操作步骤，以JSON格式返回：
{{
"steps": [
{{
"step_number": 1,
"action": "操作名称",
"parameters": {{"param1": "value1"}},
"description": "步骤描述"
}},
...
]
}}
注意：
- 步骤要有逻辑顺序
- 参数要具体明确
- 每个步骤要有清晰的描述
"""
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
plan = json.loads(response.choices[0].message.content)
return plan.get("steps", [])
def execute_workflow(self, user_request: str) -> Dict:
"""执行完整工作流"""
print(f"🚀 开始执行工作流")
print(f"📝 用户请求: {user_request}\n")
# 步骤1: 规划
print("=" * 60)
print("步骤1: 规划工作流")
print("=" * 60)
steps = self.plan_workflow(user_request)
print(f"已规划 {len(steps)} 个步骤：")
for step in steps:
print(f" {step['step_number']}. {step['description']}")
print()
# 步骤2: 执行
print("=" * 60)
print("步骤2: 执行工作流")
print("=" * 60)
execution_results = []
for step in steps:
print(f"\n执行步骤 {step['step_number']}: {step['description']}")
print("-" * 60)
action_name = step['action']
if action_name in self.available_actions:
try:
# 执行操作
action_func = self.available_actions[action_name]
result = action_func(**step['parameters'])
execution_results.append({
"step": step['step_number'],
"action": action_name,
"status": "success",
"result": result
})
print(f"✓ 完成")
except Exception as e:
execution_results.append({
"step": step['step_number'],
"action": action_name,
"status": "failed",
"error": str(e)
})
print(f"✗ 失败: {e}")
else:
print(f"⚠ 未知操作: {action_name}")
print("\n" + "=" * 60)
print("工作流执行完成")
print("=" * 60)
return {
"request": user_request,
"planned_steps": steps,
"execution_results": execution_results,
"overall_status": "completed"
}
# 测试
if __name__ == "__main__":
agent = WorkflowAgent()
request = """
我们需要启动一个新项目：
1. 给团队成员发邮件通知项目启动
2. 创建项目任务列表
3. 安排项目启动会议
4. 生成项目计划文档
"""
result = agent.execute_workflow(request)
print("\n\n📊 执行总结:")
print(f"总步骤数: {len(result['planned_steps'])}")
print(f"成功: {sum(1 for r in result['execution_results'] if r['status'] == 'success')}")
print(f"失败: {sum(1 for r in result['execution_results'] if r['status'] == 'failed')}")
🎨 高级技巧
1. 自适应规划
让 Agent 根据中间结果调整计划：
def adaptive_planning(self, goal: str, current_state: Dict) -> List[Dict]:
"""根据当前状态自适应调整计划"""
prompt = f"""
目标: {goal}
当前状态: {json.dumps(current_state, ensure_ascii=False)}
基于当前状态，规划接下来需要执行的步骤。
如果目标已达成，返回空列表。
"""
# ... 实现
2. 任务优先级排序
def prioritize_tasks(self, tasks: List[Dict]) -> List[Dict]:
"""对任务进行优先级排序"""
prompt = f"""
以下是待执行的任务列表：
{json.dumps(tasks, ensure_ascii=False)}
请根据以下因素排序：
1. 依赖关系（先决条件）
2. 紧急程度
3. 重要性
返回排序后的任务列表。
"""
# ... 实现
3. 错误恢复
def handle_step_failure(self, step: Dict, error: str) -> Dict:
"""处理步骤失败"""
prompt = f"""
步骤执行失败：
步骤: {step['description']}
错误: {error}
请提供：
1. 失败原因分析
2. 可能的替代方案
3. 是否应该继续执行后续步骤
"""
# ... 实现
✅ 阶段完成检查清单
- [ ] 理解任务分解的基本原理
- [ ] 掌握 ReAct 模式的实现
- [ ] 能够让 Agent 进行多步推理
- [ ] 实现了自动规划和执行的工作流
- [ ] 了解如何处理复杂任务的依赖关系
- [ ] 掌握错误处理和恢复机制
- [ ] 完成至少一个多步推理项目
🎯 下一步
完成本阶段后，进入 阶段 4：记忆、知识库系统与长期状态，学习如何赋予 Agent 记忆能力。
💡 小贴士：多步推理是 Agent 的"大脑"，好的任务分解能力让 Agent 更智能。记住：从简单任务开始，逐步增加复杂度。