前面四节分别讨论了ReAct、规划、反思和多Agent协作。这些能力各自解决特定的问题，但它们组合在一起时，才能让Agent真正应对复杂任务。
本节将构建一个完整的复杂任务拆解Agent。这个Agent能够：
构建一个市场调研Agent。用户给它一个调研需求，它需要：
这个任务天然需要规划（多步骤）、ReAct（每步灵活执行）、反思（遇到困难时调整方向）。
这个Agent需要以下工具：
import json
import time
from typing import Any, Dict, List, Optional
from datetime import datetime
class MarketResearchTools:
"""市场调研Agent的工具集"""
def __init__(self):
self.search_history = []
self.analysis_results = {}
def search_market_data(self, keyword: str, source: str = "all") -> str:
"""
搜索市场数据。
参数keyword是搜索关键词，source是数据来源（all/industry_report/news/social_media）。
"""
# 模拟搜索结果
mock_results = {
"智能手表": {
"industry_report": "2026年全球智能手表市场规模预计达到1200亿美元，年增长率15%。",
"news": "苹果、华为、三星占据智能手表市场前三名，合计市场份额65%。",
"social_media": "用户讨论热点：健康监测、续航、外观设计。",
},
"无线耳机": {
"industry_report": "2026年无线耳机市场预计达到800亿美元，增速放缓至8%。",
"news": "价格战加剧，中低端市场增长迅速。",
"social_media": "用户讨论热点：降噪效果、佩戴舒适度、连接稳定性。",
},
}
self.search_history.append({
"keyword": keyword,
"source": source,
"timestamp": datetime.now().isoformat(),
})
# 查找匹配结果
if keyword in mock_results:
data = mock_results[keyword]
if source == "all":
return "\n".join(f"[{k}] {v}" for k, v in data.items())
return data.get(source, f"未找到{source}来源的数据")
# 部分匹配
for key in mock_results:
if keyword in key or key in keyword:
data = mock_results[key]
return f"找到相关数据（{key}）：\n" + "\n".join(
f"[{k}] {v}" for k, v in data.items()
)
return f"未找到与'{keyword}'相关的市场数据。建议尝试更宽泛的关键词。"
def analyze_competitor(self, competitor_name: str) -> str:
"""
分析竞争对手。
参数competitor_name是竞争对手名称。
"""
mock_competitors = {
"苹果": {
"strengths": "品牌影响力强、生态系统完善、用户忠诚度高",
"weaknesses": "价格偏高、创新节奏放缓",
"market_share": "28%",
"strategy": "高端市场领导者，注重用户体验和生态整合",
},
"华为": {
"strengths": "技术实力强、本土化优势、渠道覆盖广",
"weaknesses": "海外市场受限、供应链挑战",
"market_share": "22%",
"strategy": "技术创新驱动，积极拓展可穿戴设备生态",
},
"三星": {
"strengths": "供应链完整、全球渠道、产品线丰富",
"weaknesses": "软件体验一般、品牌差异化不足",
"market_share": "15%",
"strategy": "全价位覆盖，侧重硬件创新",
},
}
if competitor_name in mock_competitors:
info = mock_competitors[competitor_name]
self.analysis_results[competitor_name] = info
return (
f"竞争对手分析：{competitor_name}\n"
f"优势：{info['strengths']}\n"
f"劣势：{info['weaknesses']}\n"
f"市场份额：{info['market_share']}\n"
f"战略：{info['strategy']}"
)
return f"未找到'{competitor_name}'的竞争分析数据。"
def collect_user_feedback(self, product: str, limit: int = 10) -> str:
"""
收集用户反馈。
参数product是产品名称，limit是收集数量。
"""
mock_feedback = [
"电池续航太短，一天一充不够用",
"健康监测功能很实用，但有时数据不准",
"外观设计不错，佩戴舒适",
"价格有点高，希望有优惠",
"和手机的连接偶尔不稳定",
"睡眠监测功能很惊喜，数据详细",
"屏幕在阳光下看不清",
"更新后系统更流畅了",
"充电速度快，半小时充到80%",
"运动模式识别不够智能",
]
feedback = mock_feedback[:limit]
return f"收集到{len(feedback)}条用户反馈：\n" + "\n".join(
f"- {f}" for f in feedback
)
def generate_report(self, content: str, format_type: str = "summary") -> str:
"""
生成报告。
参数content是报告内容，format_type是报告格式（summary/detailed）。
"""
if format_type == "summary":
return f"[调研摘要]\n{content[:500]}"
else:
return f"[详细调研报告]\n{content}"
整合ReAct、规划、反思能力：
class ComplexTaskAgent:
"""能拆解复杂任务的Agent"""
def __init__(self, llm_client, tool_registry,
max_iterations: int = 20,
verbose: bool = True):
self.client = llm_client
self.tools = tool_registry
self.max_iterations = max_iterations
self.verbose = verbose
self.system_prompt = """
你是一个市场调研Agent。你能够帮助用户完成市场调研任务。
在接到复杂任务时，你需要：
1. 先分析任务，制定执行计划
2. 按计划逐步执行，使用工具收集数据和分析
3. 如果遇到困难，停下来反思并调整策略
4. 完成所有步骤后，生成完整的调研报告
可用的工具：
- search_market_data：搜索市场数据
- analyze_competitor：分析竞争对手
- collect_user_feedback：收集用户反馈
- generate_report：生成报告
执行原则：
- 先搜索市场概况，再深入分析具体方面
- 数据分析要基于工具返回的真实数据，不要编造
- 生成报告前确保已收集足够的信息
- 如果工具返回"未找到"，尝试调整关键词而非放弃
"""
def run(self, user_task: str) -> Dict[str, Any]:
"""执行复杂任务"""
trace = {
"task": user_task,
"plan": None,
"steps": [],
"reflections": [],
"final_result": None,
}
# 初始化记忆
memory = WorkingMemory(max_tokens=12000)
memory.set_system_prompt(self.system_prompt)
memory.add_message("user", user_task, importance=5)
# 第1阶段：制定计划
if self.verbose:
print("\n" + "="*60)
print("阶段1：制定计划")
print("="*60)
plan_text = self._generate_plan(memory, user_task)
trace["plan"] = plan_text
memory.add_message("system", f"[执行计划]\n{plan_text}", importance=5)
if self.verbose:
print(plan_text)
# 第2阶段：执行循环
if self.verbose:
print("\n" + "="*60)
print("阶段2：执行")
print("="*60)
consecutive_failures = 0
recent_tool_calls = []
for iteration in range(self.max_iterations):
if self.verbose:
print(f"\n--- 第{iteration + 1}轮 ---")
# 检查是否需要反思
if consecutive_failures >= 3 or len(set(recent_tool_calls[-3:])) == 1 and len(recent_tool_calls) >= 3:
if self.verbose:
print(" [触发反思]")
reflection_prompt = (
"\n[反思] 你已经连续多次遇到问题。"
"请停下来思考：\n"
"1. 当前的方法是否有问题？\n"
"2. 是否需要换一种策略？\n"
"3. 是否需要调整搜索关键词或改用其他工具？\n"
"在思考后，继续执行。"
)
memory.add_message("system", reflection_prompt, importance=4)
trace["reflections"].append(iteration)
consecutive_failures = 0
recent_tool_calls = []
# 调用LLM
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=memory.get_context(),
tools=self.tools.get_schemas(),
temperature=0.0,
)
message = response.choices[0].message
if message.tool_calls:
memory.add_message("assistant",
f"[调用工具] {message.tool_calls[0].function.name}",
importance=3)
for tool_call in message.tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if self.verbose:
print(f" 工具：{name}({args})")
result = self.tools.execute(name, args)
if self.verbose:
status = "✓" if result.success else "✗"
print(f" {status} 结果：{result.content[:150]}...")
# 更新失败计数
if result.success:
consecutive_failures = 0
else:
consecutive_failures += 1
recent_tool_calls.append(name)
# 记录步骤
trace["steps"].append({
"iteration": iteration,
"tool": name,
"arguments": args,
"success": result.success,
"result_preview": result.content[:200],
})
memory.add_message("tool", result.content,
importance=4,
tool_call_id=tool_call.id)
else:
# 最终答案
final_answer = message.content or ""
trace["final_result"] = final_answer
if self.verbose:
print(f"\n最终答案：\n{final_answer}")
return trace
trace["final_result"] = "达到最大迭代次数，任务未完成。"
return trace
def _generate_plan(self, memory, task: str) -> str:
"""生成执行计划"""
planning_prompt = (
f"请为以下任务制定一个详细的执行计划：\n\n{task}\n\n"
"计划要求：\n"
"1. 分析任务需要哪些信息\n"
"2. 确定执行步骤的顺序\n"
"3. 为每个步骤指定使用的工具\n"
"4. 考虑步骤之间的依赖关系\n\n"
"输出格式：\n"
"[计划]\n"
"1. 步骤描述（使用什么工具）\n"
"2. ...\n"
"[开始执行]"
)
memory.add_message("system", planning_prompt, importance=5)
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=memory.get_context(),
temperature=0.0,
)
plan = response.choices[0].message.content or ""
# 从记忆中移除规划提示（保持上下文整洁）
# 实际实现中可以根据需要保留或移除
return plan
def create_market_research_agent(api_key: str) -> ComplexTaskAgent:
"""创建市场调研Agent"""
from openai import OpenAI
from harness.tools import ToolRegistry, ToolCategory, ToolPermission
client = OpenAI(api_key=api_key)
registry = ToolRegistry()
# 创建工具实例
tools = MarketResearchTools()
# 注册工具
registry.register(
name="search_market_data",
description=(
"搜索市场数据。当需要了解市场规模、趋势、用户讨论热点时使用。"
"参数keyword是搜索关键词（如'智能手表'、'无线耳机'）。"
"参数source可选：all=所有来源，industry_report=行业报告，"
"news=新闻，social_media=社交媒体。"
),
func=tools.search_market_data,
parameters={
"type": "object",
"properties": {
"keyword": {"type": "string", "description": "搜索关键词"},
"source": {"type": "string", "description": "数据来源", "default": "all"},
},
"required": ["keyword"],
},
category=ToolCategory.READ,
permission=ToolPermission.AUTO,
)
registry.register(
name="analyze_competitor",
description=(
"分析竞争对手。当需要了解特定竞争对手的优势、劣势、市场份额时使用。"
"参数competitor_name是竞争对手名称（如'苹果'、'华为'）。"
),
func=tools.analyze_competitor,
parameters={
"type": "object",
"properties": {
"competitor_name": {"type": "string", "description": "竞争对手名称"},
},
"required": ["competitor_name"],
},
category=ToolCategory.READ,
permission=ToolPermission.AUTO,
)
registry.register(
name="collect_user_feedback",
description=(
"收集用户反馈。当需要了解用户对产品的评价、痛点、建议时使用。"
"参数product是产品名称，limit是收集数量（默认10）。"
),
func=tools.collect_user_feedback,
parameters={
"type": "object",
"properties": {
"product": {"type": "string", "description": "产品名称"},
"limit": {"type": "integer", "description": "收集数量", "default": 10},
},
"required": ["product"],
},
category=ToolCategory.READ,
permission=ToolPermission.AUTO,
)
registry.register(
name="generate_report",
description=(
"生成调研报告。当已收集足够数据并需要生成最终报告时使用。"
"参数content是报告内容，format_type是格式（summary/detailed）。"
),
func=tools.generate_report,
parameters={
"type": "object",
"properties": {
"content": {"type": "string", "description": "报告内容"},
"format_type": {"type": "string", "description": "报告格式", "default": "summary"},
},
"required": ["content"],
},
category=ToolCategory.WRITE,
permission=ToolPermission.AUTO,
)
return ComplexTaskAgent(client, registry, verbose=True)
def test_complex_task_agent():
"""测试复杂任务拆解Agent"""
agent = create_market_research_agent(os.getenv("OPENAI_API_KEY", ""))
task = "请调研一下智能手表市场，分析主要竞争对手，收集用户反馈，并生成一份调研报告。"
result = agent.run(task)
print("\n" + "="*60)
print("执行总结")
print("="*60)
print(f"计划：{result['plan'][:200]}...")
print(f"执行步骤数：{len(result['steps'])}")
print(f"反思触发次数：{len(result['reflections'])}")
print(f"工具调用成功：{sum(1 for s in result['steps'] if s['success'])}")
print(f"工具调用失败：{sum(1 for s in result['steps'] if not s['success'])}")
============================================================
阶段1：制定计划
============================================================
[计划]
1. 搜索智能手表市场数据（使用search_market_data工具）
2. 分析主要竞争对手苹果和华为（使用analyze_competitor工具）
3. 收集用户反馈（使用collect_user_feedback工具）
4. 汇总数据生成调研报告（使用generate_report工具）
[开始执行]
============================================================
阶段2：执行
============================================================
--- 第1轮 ---
工具：search_market_data({"keyword": "智能手表", "source": "all"})
✓ 结果：[industry_report] 2026年全球智能手表市场规模预计达到1200亿美元...
--- 第2轮 ---
工具：analyze_competitor({"competitor_name": "苹果"})
✓ 结果：竞争对手分析：苹果...
--- 第3轮 ---
工具：analyze_competitor({"competitor_name": "华为"})
✓ 结果：竞争对手分析：华为...
--- 第4轮 ---
工具：collect_user_feedback({"product": "智能手表", "limit": 10})
✓ 结果：收集到10条用户反馈...
--- 第5轮 ---
工具：generate_report({"content": "..."})
✓ 结果：[调研摘要]...
最终答案：
根据市场调研结果，智能手表市场呈现以下情况：
【市场概况】
2026年全球智能手表市场规模预计达1200亿美元，年增长率15%。
【竞争格局】
- 苹果：28%市场份额，优势在于品牌和生态，劣势是价格偏高
- 华为：22%市场份额，优势在于技术和本土化，劣势是海外受限
【用户反馈】
主要关注点：电池续航、健康监测准确性、价格...
【建议】
（Agent根据数据生成的建议）
在这个实战中，Agent展现了几个关键的"拆解"能力：
第一：任务理解与分解。 从"调研智能手表市场，分析竞争对手，收集用户反馈，生成报告"这个复合任务中，Agent识别出了4个明确的子任务，并为每个子任务选择了正确的工具。
第二：有序执行。 Agent按照逻辑依赖顺序执行：先搜索市场概况，再分析竞争对手，然后收集用户反馈，最后生成报告。这个顺序是合理的——前一步的结果为后一步提供基础。
第三：反思的触发。 如果某个工具返回"未找到"，Agent会触发反思，调整搜索关键词或改用其他工具，而不是机械地重复相同操作。
第四：最终汇总。 Agent将各步骤的结果整合为一份结构化的调研报告，而非简单拼接工具输出。
本节构建了一个完整的复杂任务拆解Agent，整合了第6章的所有能力：
第6章到此结束。在下一章中，我们将进入评估体系设计——如何度量Agent的表现，用数据而非感觉来判断Agent的质量。
评论专区
评论加载中...登录后即可发表评论