06 · 项目二:深度研究智能体从 0 到 1

目标:跑通并理解 code/chapter14/helloagents-deepresearch —— 输入任意研究主题,自动拆解成 TODO 子任务、并发搜索全网、逐任务总结、最终生成带引用来源的 Markdown 研究报告,全程进度实时"直播"到浏览器。

技术要点:TODO 驱动的规划-执行模式 + ToolAwareSimpleAgent + 搜索引擎集成 + SSE 流式推送 + 多线程并发。

预计 3~4 天。

0. 成品长什么样

输入"量子计算对密码学的影响",页面上实时出现:

一张 TODO 清单(模型自动拆的 3~5 个子任务),每项带状态灯
每个任务依次点亮:搜索来源列表滚动出现 → 摘要文字逐字蹦出
最后生成完整研究报告(Markdown 渲染,带引用来源),同时存一份笔记到本地

1. 架构总览

浏览器(Vue, :5173)
   │ POST /research/stream {topic: "..."}     ← SSE长连接
   ▼
FastAPI (:8000)  src/main.py
   ▼
DeepResearchAgent  src/agent.py(总编排)
   │
   ├─ ① 规划: todo_agent(研究规划专家) → 输出TODO列表
   │
   ├─ ② 执行: 每个TODO一个线程,并发跑
   │     └─ dispatch_search() 搜索(DuckDuckGo/Tavily/Perplexity)
   │        → summarizer(任务总结专家) 流式生成摘要
   │
   ├─ ③ 汇总: report_agent(报告撰写专家) → Markdown报告
   │
   └─ 全程: NoteTool记笔记 + ToolCallTracker把工具事件推给前端
   │
   ▼ 每个进度都作为SSE事件 yield 出去
浏览器边收边渲染

和项目一对比着记:项目一的任务结构是写死的四步;项目二的任务是模型现场规划出来的 N 步,而且并发执行、全程直播。

2. 里程碑一:跑起来(半天)

2.1 后端(注意:这个项目用 uv 管理)

cd hello-agents/code/chapter14/helloagents-deepresearch/backend
uv sync            # 读取pyproject.toml + uv.lock,精确复现依赖环境
cp .env.example .env

.env 最小配置(默认 DuckDuckGo 搜索,不用搜索 Key):

SEARCH_API='duckduckgo'
LLM_PROVIDER=custom
LLM_MODEL_ID=Qwen/Qwen2.5-72B-Instruct
LLM_API_KEY=你的key
LLM_BASE_URL=https://api-inference.modelscope.cn/v1/
MAX_WEB_RESEARCH_LOOPS=3
FETCH_FULL_PAGE=True

LLM_PROVIDER 是这个项目特有的:custom = 走 API;ollama / lmstudio = 走本地模型(想完全免费离线跑,装个 Ollama 拉个 qwen 模型,改两行配置即可,这也是该项目"fully local"设计初衷)。

启动:

cd src
uv run python main.py     # 或 uv run uvicorn main:app --reload

Checkpoint 1: curl http://localhost:8000/healthz 返回 {"status":"ok"};后端日志出现一行完整的配置摘要(provider/model/search_api/api_key 打码显示)——这行日志排错时非常有用。

2.2 前端

cd ../frontend
npm install
npm run dev

Checkpoint 2: 打开 http://localhost:5173,输入一个简单主题(如"咖啡因的作用机制"),完整看到 TODO 列表 → 逐任务完成 → 最终报告的全过程。

3. 里程碑二:读懂编排核心 agent.py(1~2 天)

src/agent.py 的 DeepResearchAgent 是全项目的心脏。建议按"三个 Agent → run() → run_stream()"的顺序读。

3.1 三个专家 Agent,职责各一句话

self.todo_agent   = ...  # 研究规划专家: 主题 → TODO任务列表(JSON)
self.report_agent = ...  # 报告撰写专家: 所有摘要 → 结构化Markdown报告
self._summarizer_factory = lambda: ...  # 任务总结专家: 搜索结果 → 单任务摘要

细节一:总结专家不是一个实例,而是一个工厂函数(每次调用新建一个)。为什么?因为任务是多线程并发执行的,共享一个 Agent 实例会让不同任务的对话历史串味,每线程一个实例最干净。

名词补课:线程(Thread)= 一个程序内部可以同时推进的多条"执行流",好比一家餐厅的多个服务员。Python 有一个特殊限制叫 GIL(全局解释器锁):同一时刻只有一条线程真正在执行 Python 代码——但线程在等待网络响应(比如等 LLM 回复、等搜索结果)时会自动让出执行权给别的线程。所以"每个任务都在等外部服务"的场景(正是本项目),多线程的并发提速是实打实的。

细节二:它们都是 ToolAwareSimpleAgent,统一挂了 NoteTool(笔记工具)和 tool_call_listener(工具调用监听器)。Agent 每次记笔记,监听器都会记录事件——这是前端能显示"正在保存笔记 xxx"的数据来源。

3.2 同步版 run():先看简单的

def run(self, topic: str) -> SummaryStateOutput:
    state = SummaryState(research_topic=topic)          # 全程共享的状态对象
    state.todo_items = self.planner.plan_todo_list(state)   # ① 规划
    if not state.todo_items:
        state.todo_items = [self.planner.create_fallback_task(state)]  # 兜底:至少一个任务

    for task in state.todo_items:                        # ② 顺序执行每个任务
        self._execute_task(state, task, emit_stream=False)

    report = self.reporting.generate_report(state)       # ③ 汇总成报告
    self._persist_final_report(state, report)            #    存为笔记
    return SummaryStateOutput(report_markdown=report, todo_items=state.todo_items)

_execute_task 内部,每个任务的生命周期:

in_progress → dispatch_search(搜索) → 无结果? → skipped
                     ↓有结果
      prepare_research_context(整理正文+来源列表)
                     ↓
      summarizer 总结 → task.summary 写入 → completed

注意 SummaryState 上有把线程锁(self._state_lock),往 web_research_results 等共享列表追加时都要拿锁——多线程写共享数据的基本功。

3.3 流式版 run_stream():全项目最难也最值得学的函数

它要解决的问题:N 个任务并发跑,每个任务随时产生事件(状态变更/来源/摘要片段),如何汇成一条有序的事件流发给浏览器?

答案是经典的生产者-消费者队列(下面是骨架示意,... 处省略了参数,不能直接运行,看结构就行):

event_queue: Queue[dict] = Queue()          # 所有线程往这一个队列扔事件

def worker(task, step):                     # 每个任务一个线程(生产者)
    enqueue({"type": "task_status", "status": "in_progress", ...})
    for event in self._execute_task(state, task, emit_stream=True, step=step):
        enqueue(event, task=task)
    enqueue({"type": "__task_done__", "task_id": task.id})   # 内部哨兵事件

for task in state.todo_items:               # 全部线程一起启动 → 并发!
    Thread(target=worker, args=(task, step), daemon=True).start()

while finished_workers < active_workers:    # 主生成器(消费者)
    event = event_queue.get()               # 阻塞等任意线程的事件
    if event.get("type") == "__task_done__":
        finished_workers += 1               # 数哨兵,数够了说明全部干完
        continue
    yield event                             # ← 每yield一次,SSE就推一条给浏览器

几个精妙设计,读源码时对号入座:

stream_token / step:规划完就给每个任务分配了频道号(task_1、task_2…),每个事件都带上它,前端才知道"这条摘要片段该追加到哪张卡片"。并发的乱序事件流,靠它复原成 N 条有序时间线。
哨兵事件 __task_done__:只用来计数,不发给前端。比 thread.join() 优雅——join 会阻塞,期间无法转发别的线程的事件。
摘要流式化:summarizer.stream_task_summary 返回一个 chunk 迭代器,每个 chunk 包装成 task_summary_chunk 事件 yield 出去,前端逐字追加,形成"打字机"效果。
工具事件转发:_set_tool_event_sink(enqueue) 让 Agent 每次调工具立刻入队,而不是攒到最后。

事件类型速查(前端 App.vue 就是对着这张表写的 switch):

type	含义
`status`	全局状态文本
`todo_list`	规划完成,任务清单
`task_status`	某任务状态变更(in_progress/completed/skipped/failed)
`sources`	某任务的搜索来源
`task_summary_chunk`	摘要文字片段(打字机)
`report_note` / `final_report` / `done`	报告笔记 / 最终报告 / 结束

3.4 搜索服务 services/search.py

dispatch_search(query, config, loop_count) 是统一入口,按 SEARCH_API 分发到 DuckDuckGo(免费无Key)/ Tavily / Perplexity / SearXNG。返回统一结构 {"results": [{title, url, content}, ...]}。FETCH_FULL_PAGE=True 时还会抓取网页正文全文,摘要质量更高但更慢。

想换搜索源,不用改代码:请求体里就能覆盖({"topic": "...", "search_api": "tavily"}),或改 .env。

4. 里程碑三:动手改造(1 天)

调任务数量:找到 src/prompts.py 的 todo_planner_system_prompt,把任务数上限从默认改成"最多 3 个",观察规划行为变化。体会:行为改变优先改提示词,而不是改代码。
换搜索后端:申请 Tavily Key,.env 切到 tavily,对比同一主题下来源质量差异。
加事件类型:在 _execute_task 里,任务开始搜索前 yield {"type": "status", "message": f"开始搜索: {task.query}"},前端确认能收到(浏览器 F12 → Network → stream 请求 → EventStream 标签可以直接看原始事件流)。
挑战:限制并发。任务全部同时开线程,任务多时容易触发 LLM 限流。用 threading.Semaphore(2) 限制同时最多 2 个任务在跑。

5. 本项目的可迁移经验

规划-执行模式:LLM 拆任务(输出 JSON TODO)+ 代码循环执行,是"深度研究类"产品(OpenAI Deep Research、Manus 等)的通用骨架
生产者-消费者队列:多线程事件汇流成单一 SSE 流的标准解法
哨兵事件计数:不阻塞地等待 N 个并发单元结束
每线程独立 Agent 实例:避免对话历史被并发污染
搜索后端可插拔:统一返回结构 + 配置分发,换供应商不动业务代码

自测清单

项目一是"固定流水线",项目二是"规划-执行"。用一句话说清区别,以及各自适合什么场景。
run_stream 里为什么用 Queue 而不是直接在 worker 线程里 yield?(提示:生成器不能跨线程 yield)
__task_done__ 事件前端能收到吗?它的作用是什么?
摘要的"打字机效果"从后端到前端经过了哪些环节?
为什么总结专家用工厂函数每次新建实例?
已完成至少两个改造练习。

下一篇:07 · 项目三:赛博小镇从 0 到 1