自动化深度研究智能体
一句话
构建一个能够自动化执行深度研究任务的智能体助手，通过TODO驱动的研究范式将复杂主题分解为可执行的子任务。
什么时候翻这页
- 需要快速了解新主题、技术或事件
- 需要系统化地收集、整理和分析信息
- 需要生成结构化研究报告并保留引用来源
- 想了解多智能体协作在知识密集型应用中的应用
核心概念
- TODO驱动的研究范式：将复杂研究主题分解为3-5个子任务，通过规划、执行、报告三阶段完成
- 三阶段研究流程：
- 规划阶段：将研究主题分解为子任务
- 执行阶段：搜索资料并总结关键信息
- 报告阶段：整合所有子任务结果生成最终报告
- ToolAwareSimpleAgent：扩展SimpleAgent，增加工具调用监听能力
- 多智能体协作：三个专门Agent各司其职（规划、总结、报告）
- SSE（Server-Sent Events）：服务器推送技术，实现实时进度展示
怎么做
-
系统架构设计
- 前后端分离架构（Vue3+TypeScript + FastAPI）
- 四层架构：前端层、后端层、智能体层、外部服务层
-
实现TODO驱动研究
- 创建研究规划专家（TODO Planner）分解主题
- 创建任务总结专家（Task Summarizer）处理搜索结果
- 创建报告撰写专家（Report Writer）整合结果
-
集成工具系统
- 扩展SearchTool支持多种搜索引擎
- 使用NoteTool持久化研究进度
-
设计前端交互
- 全屏模态对话框UI
- SSE实时进度展示
- Markdown结果可视化
命令 / 代码速查
启动后端：
cd helloagents-deepresearch/backend
uv sync # 或 pip install -e .
cp .env.example .env
# 编辑.env文件配置API密钥
python src/main.py
启动前端：
cd helloagents-deepresearch/frontend
npm install
npm run dev
研究规划Agent提示词：
todo_planner_instructions = """
你是一个研究规划专家。你的任务是将用户的研究主题分解为3-5个子任务。
当前日期：{current_date}
研究主题：{research_topic}
请分析这个研究主题，将其分解为3-5个子任务。每个子任务应该：
1. 涵盖主题的一个重要方面
2. 有明确的研究目标
3. 可以通过搜索引擎找到相关资料
请以JSON格式返回子任务列表，每个子任务包含：
- title：任务标题（简洁明了）
- intent：任务意图（为什么要研究这个）
- query：搜索查询（用于搜索引擎的查询字符串，可以使用英文以获得更好的搜索结果）
"""
ToolAwareSimpleAgent实现：
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
SSE后端端点：
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from typing import AsyncGenerator
import asyncio
import json
app = FastAPI()
async def research_stream(topic: str) -> AsyncGenerator[str, None]:
"""研究流式生成器"""
try:
# 1. 规划阶段
yield f"data: {json.dumps({'type': 'progress', 'stage': 'planning', 'percentage': 10, 'text': '正在规划研究任务...'})}\n\n"
# 调用PlanningService
todo_items = await planning_service.plan_todo_list(topic)
yield f"data: {json.dumps({'type': 'plan', 'data': [item.dict() for item in todo_items]})}\n\n"
# 2. 执行阶段
for idx, task in enumerate(todo_items, start=1):
percentage = 10 + (idx / len(todo_items)) * 70
yield f"data: {json.dumps({'type': 'progress', 'stage': 'executing', 'percentage': percentage, 'text': f'正在研究任务{idx}/{len(todo_items)}：{task.title}'})}\n\n"
# 搜索和总结
search_results = await search_service.search(task.query)
summary, source_urls = await summarization_service.summarize_task(task, search_results)
yield f"data: {json.dumps({'type': 'task_summary', 'task_id': task.id, 'summary': summary})}\n\n"
# 3. 报告阶段
yield f"data: {json.dumps({'type': 'progress', 'stage': 'reporting', 'percentage': 90, 'text': '正在生成最终报告...'})}\n\n"
# 生成报告
report = await reporting_service.generate_report(topic, task_summaries)
yield f"data: {json.dumps({'type': 'report', 'data': report})}\n\n"
# 完成
yield f"data: {json.dumps({'type': 'progress', 'stage': 'completed', 'percentage': 100, 'text': '研究完成！'})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
@app.post("/api/research")
async def research(request: ResearchRequest):
"""研究端点（SSE）"""
return StreamingResponse(
research_stream(request.topic),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
}
)
与 Python / Claude Code 手册的联系
- SimpleAgent扩展：基于第七章的SimpleAgent实现ToolAwareSimpleAgent
- 工具系统集成：扩展第七章的SearchTool，集成第九章的NoteTool
- 多智能体协作：延续第十三章的多智能体产品设计理念
- SSE技术：应用协议章节讲解的服务器推送技术
初学者易错点
- JSON解析错误：Agent返回的JSON可能包含额外文本，需要使用正则表达式提取JSON部分
- 搜索引擎选择不当：不同搜索引擎适合不同场景，需根据研究主题选择合适的搜索引擎
- 子任务数量控制：子任务太少会遗漏信息，太多会导致冗余，控制在3-5个为宜
- 来源引用缺失：总结时必须为每个观点添加来源引用，确保可追溯性
- SSE连接管理：前端需要正确处理SSE连接的建立、消息接收和错误处理