14.3.1 Agent 职责划分
在深度研究助手中，我们设计了三个专门的 Agent，每个 Agent 负责一个特定的任务。这使得每个 Agent 都很简单，易于理解和维护。
在第七章中，我们学习了如何使用SimpleAgent来构建智能体。SimpleAgent的设计理念是简单直接：每次调用run()方法时，Agent 会分析用户的问题，决定是否需要调用工具，然后返回结果。这种设计在处理简单任务时非常有效，但当面对深度研究这样的复杂任务时，就需要我们继续采用多智能体协作的方案进行。
如表 14.1 所示，三个 Agent 分别负责规划、总结和报告生成。
表 14.1 三个 Agent 的职责划分
让我们详细介绍每个 Agent 的设计。
Agent 1：研究规划专家（TODO Planner）
职责：将研究主题分解为 3-5 个子任务
设计理念：研究规划专家的核心任务是理解用户的研究主题，分析主题的关键方面，然后生成一系列子任务。这个过程类似于人类研究者在开始研究前的"头脑风暴"阶段。
Prompt 设计：
todo_planner_instructions = """
你是一个研究规划专家。你的任务是将用户的研究主题分解为3-5个子任务。
当前日期：{current_date}
研究主题：{research_topic}
请分析这个研究主题，将其分解为3-5个子任务。每个子任务应该：
1. 涵盖主题的一个重要方面
2. 有明确的研究目标
3. 可以通过搜索引擎找到相关资料
请以JSON格式返回子任务列表，每个子任务包含：
- title：任务标题（简洁明了）
- intent：任务意图（为什么要研究这个）
- query：搜索查询（用于搜索引擎的查询字符串，可以使用英文以获得更好的搜索结果）
示例输出：
[
{{
"title": "什么是多模态模型",
"intent": "了解多模态模型的基础概念，为后续研究打下基础",
"query": "multimodal model definition concept 2024"
}},
...
]
请确保：
1. 子任务数量在3-5个之间
2. 子任务之间有逻辑关系（如从基础到应用，从现状到趋势）
3. 搜索查询能够准确找到相关资料
4. 只返回JSON，不要包含其他文本
"""
关键设计点：提示词包含当前日期以获取最新信息，明确要求 JSON 格式输出便于解析，通过示例帮助 Agent 理解期望输出，并强调子任务数量、逻辑关系等约束。
实现代码：
这里的 ToolAwareSimpleAgent 是根据 SimpleAgent 拓展实现，可以在 14.3.2 了解，这里不用深究。
class PlanningService:
def __init__(self, llm: HelloAgentsLLM):
self._agent = ToolAwareSimpleAgent(
name="TODO Planner",
system_prompt="你是一个研究规划专家",
llm=llm,
tool_call_listener=self._on_tool_call
)
def plan_todo_list(self, state: SummaryState) -> List[TodoItem]:
prompt = todo_planner_instructions.format(
current_date=get_current_date(),
research_topic=state.research_topic,
)
response = self._agent.run(prompt)
tasks_payload = self._extract_tasks(response)
todo_items = []
for idx, item in enumerate(tasks_payload, start=1):
task = TodoItem(
id=idx,
title=item["title"],
intent=item["intent"],
query=item["query"],
)
todo_items.append(task)
return todo_items
def _extract_tasks(self, response: str) -> List[dict]:
"""从Agent响应中提取JSON"""
# 使用正则表达式提取JSON部分
json_match = re.search(r'\[.*\]', response, re.DOTALL)
if json_match:
json_str = json_match.group(0)
return json.loads(json_str)
else:
raise ValueError("无法从响应中提取JSON")
Agent 2：任务总结专家（Task Summarizer）
职责：总结搜索结果，提取关键信息
设计理念：任务总结专家的核心任务是阅读搜索结果，提取关键信息，并以结构化的方式呈现。这个过程类似于人类研究者在阅读文献后做笔记的过程。
Prompt 设计：
task_summarizer_instructions = """
你是一个任务总结专家。你的任务是总结搜索结果，提取关键信息。
任务标题：{task_title}
任务意图：{task_intent}
搜索查询：{task_query}
搜索结果：
{search_results}
请仔细阅读以上搜索结果，提取关键信息，并以Markdown格式返回总结。
总结应该包含：
1. **核心观点**：搜索结果中的核心观点和结论
2. **关键数据**：重要的数字、日期、名称等
3. **来源引用**：为每个观点添加来源引用（使用[1]、[2]等标记）
请确保：
1. 总结简洁明了，避免冗余
2. 保留重要的细节和数据
3. 为每个观点添加来源引用
4. 使用Markdown格式（标题、列表、加粗等）
示例输出：
## 核心观点
多模态模型是一种能够处理多种类型数据的AI模型[1]。与传统的单模态模型不同，多模态模型可以同时理解文本、图像、音频等[2]。
**关键特点：**
- 跨模态理解[1]
- 统一表示[3]
- 端到端训练[2]
## 来源
[1] https://example.com/source1
[2] https://example.com/source2
[3] https://example.com/source3
"""
关键设计点：提示词包含任务标题、意图、查询等上下文帮助 Agent 理解任务，明确要求输出包含核心观点、关键数据、来源引用，强调为每个观点添加来源引用，并通过示例帮助 Agent 理解期望的输出格式。
实现代码：
class SummarizationService:
def __init__(self, llm: HelloAgentsLLM):
self._agent = ToolAwareSimpleAgent(
name="Task Summarizer",
system_prompt="你是一个任务总结专家",
llm=llm,
tool_call_listener=self._on_tool_call
)
def summarize_task(
self,
task: TodoItem,
search_results: List[dict]
) -> str:
# 格式化搜索结果
formatted_sources = self._format_sources(search_results)
prompt = task_summarizer_instructions.format(
task_title=task.title,
task_intent=task.intent,
task_query=task.query,
search_results=formatted_sources,
)
summary = self._agent.run(prompt)
return summary
def _format_sources(self, search_results: List[dict]) -> str:
"""格式化搜索结果"""
formatted = []
for idx, result in enumerate(search_results, start=1):
formatted.append(
f"[{idx}] {result['title']}\n"
f"URL: {result['url']}\n"
f"摘要: {result['snippet']}\n"
)
return "\n".join(formatted)
Agent 3：报告撰写专家（Report Writer）
职责：整合所有子任务的总结，生成最终报告
设计理念：报告撰写专家的核心任务是将所有子任务的总结整合成一份结构化的报告。这个过程类似于人类研究者在完成所有调研后撰写研究报告的过程。
Prompt 设计：
report_writer_instructions = """
你是一个报告撰写专家。你的任务是整合所有子任务的总结，生成一份结构化的研究报告。
研究主题：{research_topic}
子任务总结：
{task_summaries}
请整合以上所有子任务的总结，生成一份结构化的研究报告。
报告应该包含：
1. **标题**：研究主题
2. **概述**：简要介绍研究主题和报告结构（2-3段）
3. **各个子任务的详细分析**：按照逻辑顺序组织（使用二级标题）
4. **总结**：总结研究的主要发现（1-2段）
5. **参考文献**：所有来源引用（按照子任务分组）
请确保：
1. 报告结构清晰，逻辑连贯
2. 消除重复的信息
3. 保留所有来源引用
4. 使用Markdown格式
示例输出：
# 多模态大模型的最新进展
## 概述
本报告系统地研究了多模态大模型的最新进展...
## 1. 什么是多模态模型
（此处插入子任务1的总结）
## 2. 最新的多模态模型有哪些
（此处插入子任务2的总结）
...
## 总结
通过本次研究，我们了解了...
## 参考文献
### 任务1：什么是多模态模型
[1] https://example.com/source1
...
"""
关键设计点：提示词明确要求报告包含标题、概述、详细分析、总结、参考文献等结构，强调按逻辑顺序组织内容，要求合并重复信息消除冗余，并保留所有来源引用。
实现代码：
class ReportingService:
def __init__(self, llm: HelloAgentsLLM):
self._agent = ToolAwareSimpleAgent(
name="Report Writer",
system_prompt="你是一个报告撰写专家",
llm=llm,
tool_call_listener=self._on_tool_call
)
def generate_report(
self,
research_topic: str,
task_summaries: List[Tuple[TodoItem, str]]
) -> str:
# 格式化子任务总结
formatted_summaries = self._format_summaries(task_summaries)
prompt = report_writer_instructions.format(
research_topic=research_topic,
task_summaries=formatted_summaries,
)
report = self._agent.run(prompt)
return report
def _format_summaries(
self,
task_summaries: List[Tuple[TodoItem, str]]
) -> str:
"""格式化子任务总结"""
formatted = []
for idx, (task, summary) in enumerate(task_summaries, start=1):
formatted.append(
f"## 任务{idx}：{task.title}\n"
f"意图：{task.intent}\n\n"
f"{summary}\n"
)
return "\n".join(formatted)
14.3.2 ToolAwareSimpleAgent 的设计
在第七章中，我们实现了SimpleAgent，它是 HelloAgents 框架的基础 Agent。但在深度研究助手中，我们需要一个能够记录工具调用的 Agent。这就是ToolAwareSimpleAgent的由来。
在深度研究助手中，我们需要记录每个 Agent 的工具调用情况，用于：
- 调试：查看 Agent 调用了哪些工具，传入了什么参数
- 日志：记录研究过程中的所有操作
- 分析：分析 Agent 的行为模式
- 进度展示：实时显示 Agent 正在做什么
SimpleAgent本身不支持工具调用监听，因此我们需要扩展它。
ToolAwareSimpleAgent在SimpleAgent的基础上增加了一个tool_call_listener参数，这是一个回调函数，每次工具调用时都会被调用。
使用示例：
from hello_agents import ToolAwareSimpleAgent
def tool_listener(call_info):
print(f"Agent: {call_info['agent_name']}")
print(f"工具: {call_info['tool_name']}")
print(f"参数: {call_info['parsed_parameters']}")
print(f"结果: {call_info['result']}")
agent = ToolAwareSimpleAgent(
name="研究助手",
system_prompt="你是一个研究助手",
llm=llm,
tool_call_listener=tool_listener
)
ToolAwareSimpleAgent继承自SimpleAgent，重写了_execute_tool_call方法：
class ToolAwareSimpleAgent(SimpleAgent):
def __init__(
self,
name: str,
system_prompt: str,
llm: HelloAgentsLLM,
tool_registry: Optional[ToolRegistry] = None,
tool_call_listener: Optional[Callable] = None,
):
super().__init__(
name=name,
system_prompt=system_prompt,
llm=llm,
tool_registry=tool_registry,
)
self._tool_call_listener = tool_call_listener
def _execute_tool_call(self, tool_name: str, parameters: str) -> str:
"""执行工具调用，并通知监听器"""
# 解析参数
parsed_parameters = self._parse_parameters(parameters)
# 调用工具
result = super()._execute_tool_call(tool_name, parameters)
# 通知监听器
if self._tool_call_listener:
self._tool_call_listener({
"agent_name": self.name,
"tool_name": tool_name,
"parsed_parameters": parsed_parameters,
"result": result,
})
return result
在深度研究助手中，我们使用ToolAwareSimpleAgent来记录所有 Agent 的工具调用：
class DeepResearchAgent:
def __init__(self, config: Configuration):
self.config = config
self.llm = HelloAgentsLLM(...)
# 创建工具调用监听器
def tool_listener(call_info):
self._emit_event({
"type": "tool_call",
"agent": call_info["agent_name"],
"tool": call_info["tool_name"],
"parameters": call_info["parsed_parameters"],
})
# 创建三个Agent，都使用相同的监听器
self.planner = PlanningService(self.llm, tool_listener)
self.summarizer = SummarizationService(self.llm, tool_listener)
self.reporter = ReportingService(self.llm, tool_listener)
这样，所有 Agent 的工具调用都会被记录，并通过 SSE 推送到前端，实时显示给用户。
14.3.3 Agent 协作模式
三个 Agent 之间是顺序协作的关系，如图 14.6 所示。
图 14.6 Agent 协作流程
顺序协作模式的特点是：
- 线性流程：Agent 按照固定的顺序执行
- 明确的输入输出：每个 Agent 的输入来自上一个 Agent 的输出
- 无并发：同一时间只有一个 Agent 在工作
DeepResearchAgent是整个系统的核心协调器，负责调度三个 Agent：
class DeepResearchAgent:
def run(self, research_topic: str) -> str:
# 1. 规划阶段
self._emit_event({"type": "status", "message": "正在规划研究任务..."})
todo_list = self.planner.plan_todo_list(research_topic)
self._emit_event({"type": "tasks", "tasks": todo_list})
# 2. 执行阶段
task_summaries = []
for task in todo_list:
self._emit_event({
"type": "status",
"message": f"正在研究：{task.title}"
})
# 搜索
search_results = self.search_service.search(task.query)
# 总结
summary = self.summarizer.summarize_task(task, search_results)
task_summaries.append((task, summary))
self._emit_event({
"type": "task_completed",
"task_id": task.id
})
# 3. 报告阶段
self._emit_event({"type": "status", "message": "正在生成报告..."})
report = self.reporter.generate_report(research_topic, task_summaries)
self._emit_event({"type": "report", "content": report})
return report