13 · 实战手敲三:深度研究智能体

手敲系列第三篇,对应项目二(深度研究智能体)的同架构自研版:Plan-and-Execute 模式——规划师把大问题拆成子任务、研究员并行联网搜索、报告员汇总成 Markdown 报告,全过程用 SSE 实时推送到页面。

全部 8 个文件、约 450 行,每一行都在下面给出。预计 2~3 天。

核心新技能:任务分解(Planning) 和 给 Agent 接上真实的互联网。

0. 成品长什么样

页面输入"2025 年国产大模型的发展情况",点开始:

1. 页面先出现研究计划:4 个子任务(比如"主要厂商和模型盘点""开源生态""商业化落地""与国际对比")
2. 每个子任务旁边的 ⏳ 陆续变成 ✅(它们在并行搜索和总结)
3. 最后输出一篇带章节结构的 Markdown 研究报告

1. 架构:Plan-and-Execute(和第 11 篇的 ReAct 对比着理解)

用户问题
   │
   ▼
规划师Agent ──── "拆成3~5个可独立搜索的子任务" ──▶ ["子任务1", "子任务2", ...]
   │
   ▼ 并行(每个子任务一个研究员)
研究员Agent = web_search(子任务) → LLM只依据搜索结果总结(带来源)
   │
   ▼ 全部完成后
报告员Agent ──── 把所有研究材料整合成结构化Markdown报告

ReAct vs Plan-and-Execute,面试必考的对比:

2. 建项目(10 分钟)

PyCharm New Project → deep-research。requirements.txt:

openai>=1.40.0
fastapi>=0.115.0
uvicorn[standard]>=0.32.0
python-dotenv>=1.0.0
ddgs>=9.0.0

pip install -r requirements.txt,.env 三件套同前。新建 static 目录。

ddgs 是 DuckDuckGo 搜索的免费 Python 库,不需要申请 Key。国内网络访问 DuckDuckGo 可能不稳定——没关系,代码里做了降级:搜不到就让模型用自身知识回答并注明。这也是 Agent 开发的重要一课:外部依赖永远可能挂,每个依赖都要想好挂了怎么办。

3. llm.py(30 分钟)

与第 12 篇的 llm.py 一字不差(chat + _extract_json + chat_json),照第 12 篇第 3 节再敲一遍——chat_json 这种核心工具函数,敲两遍你就永远记住了。

✅ Checkpoint 1:__main__ 测试同第 12 篇,打印出字典。

4. search.py —— 给 Agent 接上互联网(1 小时)

"""联网搜索: DuckDuckGo免key。失败不致命——返回空列表,研究员会降级用模型自身知识"""


def web_search(query: str, max_results: int = 5) -> list[dict]:
    try:
        from ddgs import DDGS
        with DDGS() as d:
            raw = list(d.text(query, max_results=max_results))
        return [
            {
                "title": r.get("title", ""),
                "url": r.get("href", ""),
                "snippet": r.get("body", ""),
            }
            for r in raw
        ]
    except Exception as e:
        print(f"[search] 搜索失败,降级处理: {e}")
        return []


def format_results(results: list[dict]) -> str:
    """把搜索结果拼成给LLM看的文本,带编号方便引用来源"""
    if not results:
        return "(本次搜索无结果。请基于你已有的知识回答,并在开头注明:信息可能不是最新。)"
    return "\n\n".join(
        f"[{i + 1}] {r['title']}\n{r['snippet']}\n来源: {r['url']}"
        for i, r in enumerate(results)
    )


if __name__ == "__main__":
    results = web_search("Python 3.13 新特性")
    print(f"拿到 {len(results)} 条结果")
    print(format_results(results)[:500])

✅ Checkpoint 2:Run,拿到若干条真实搜索结果;如果网络不通,打印降级提示——两种结果都算通过,因为降级本身就是设计的一部分。

5. agents.py —— 三个角色(2 小时)

"""规划-执行-汇总三段式(Plan-and-Execute)"""
from llm import chat, chat_json
from search import format_results, web_search

PLANNER_PROMPT = """你是研究规划师。把用户的研究问题拆解成3~5个可以独立搜索的子任务。
只输出JSON: {"tasks": ["子任务1", "子任务2", ...]}
要求: 每个子任务具体、可搜索;所有子任务合起来能完整覆盖原问题;不要重复。"""


def make_plan(question: str) -> list[str]:
    data = chat_json([
        {"role": "system", "content": PLANNER_PROMPT},
        {"role": "user", "content": question},
    ])
    tasks = data.get("tasks", [])
    # 防御: 模型偶尔会拆出0个或20个任务,超出合理范围直接报错(宁可失败也不要默默跑偏)
    if not (1 <= len(tasks) <= 8):
        raise ValueError(f"规划结果异常: {tasks}")
    return tasks


RESEARCHER_PROMPT = """你是研究员。根据提供的搜索结果回答子任务,要求:
1. 只依据搜索结果,不编造;信息不足要明说
2. 保留关键数据、时间、事实
3. 结尾用 [1][2] 的形式标注用到的来源编号"""


def research_task(task: str) -> str:
    results = web_search(task)
    return chat([
        {"role": "system", "content": RESEARCHER_PROMPT},
        {"role": "user", "content": f"子任务: {task}\n\n搜索结果:\n{format_results(results)}"},
    ])


REPORTER_PROMPT = """你是报告撰写人。把各子任务的研究材料整合成一篇结构清晰的Markdown研究报告:
# 报告标题
## 摘要(3句话以内)
## 若干章节(对应子任务,内容相近的可合并)
## 结论
只依据提供的材料写作,保持客观,保留材料中的来源标注。"""


def write_report(question: str, findings: list[tuple[str, str]]) -> str:
    material = "\n\n".join(f"### 子任务: {t}\n{r}" for t, r in findings)
    return chat([
        {"role": "system", "content": REPORTER_PROMPT},
        {"role": "user", "content": f"研究问题: {question}\n\n研究材料:\n{material}"},
    ], temperature=0.5)


if __name__ == "__main__":
    tasks = make_plan("2025年国产大模型的发展情况")
    print("研究计划:")
    for i, t in enumerate(tasks, 1):
        print(f"  {i}. {t}")

✅ Checkpoint 3:Run,打印出 3~5 个具体、不重复的子任务。多换几个问题试,观察拆解质量——规划的质量直接决定报告的质量,这就是 Plan-and-Execute 的命门。

6. main.py —— SSE 推送研究进度(2 小时)

复用第 11 篇的"后台线程 + 队列 + SSE"套路,但事件更丰富:

"""FastAPI入口: 研究接口(SSE推送全过程) + 静态页面"""
import json
import queue
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel

from agents import make_plan, research_task, write_report

app = FastAPI(title="Deep Research")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


class ResearchRequest(BaseModel):
    question: str


def run_research(question: str, emit):
    """完整流水线: 规划 → 并行研究 → 汇总报告,每个节点都emit事件"""
    tasks = make_plan(question)
    emit({"type": "plan", "tasks": tasks})

    findings: list = [None] * len(tasks)
    with ThreadPoolExecutor(max_workers=3) as pool:
        future_to_index = {}
        for i, t in enumerate(tasks):
            emit({"type": "task_start", "index": i})
            future_to_index[pool.submit(research_task, t)] = i
        for fut in as_completed(future_to_index):
            i = future_to_index[fut]
            findings[i] = fut.result()
            emit({"type": "task_done", "index": i, "preview": findings[i][:60]})

    emit({"type": "reporting"})
    report = write_report(question, list(zip(tasks, findings)))
    emit({"type": "report", "content": report})


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/api/research")
def research(req: ResearchRequest):
    q: queue.Queue = queue.Queue()

    def worker():
        try:
            run_research(req.question, q.put)
        except Exception as e:
            q.put({"type": "error", "detail": str(e)})
        q.put(None)

    threading.Thread(target=worker, daemon=True).start()

    def events():
        while True:
            event = q.get()
            if event is None:
                break
            yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"

    return StreamingResponse(events(), media_type="text/event-stream")


@app.get("/")
def index():
    return FileResponse("static/index.html")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

注意 findings: list = [None] * len(tasks) + future_to_index:并行任务完成顺序是乱的(哪个先搜完哪个先回来),但报告要按计划顺序写——用索引把结果放回原位。这是并发编程最常见的坑之一。

✅ Checkpoint 4:Run main.py,用 curl 看事件流(-N 表示不缓冲):

curl -N -X POST http://127.0.0.1:8000/api/research \
  -H "Content-Type: application/json" \
  -d '{"question": "2025年国产大模型的发展情况"}'

依次看到 plan → task_start×N → task_done×N(顺序可能乱) → reporting → report。

7. static/index.html —— 实时进度页(1.5 小时)

<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="utf-8">
<title>深度研究智能体</title>
<style>
  body { font-family: system-ui, sans-serif; max-width: 760px; margin: 32px auto; padding: 0 16px; }
  form { display: flex; gap: 8px; margin-bottom: 16px; }
  input { flex: 1; padding: 8px 10px; border: 1px solid #ccc; border-radius: 6px; }
  button { padding: 8px 16px; background: #0a58ca; color: #fff; border: none; border-radius: 6px; cursor: pointer; }
  #tasks li { margin: 6px 0; }
  #report { border: 1px solid #ddd; border-radius: 8px; padding: 16px; margin-top: 16px;
            white-space: pre-wrap; line-height: 1.7; display: none; }
  .err { color: #d63384; }
</style>
</head>
<body>
<h3>深度研究智能体</h3>
<form id="f">
  <input id="q" placeholder="输入研究问题,如: 2025年国产大模型的发展情况" required>
  <button>开始研究</button>
</form>
<ol id="tasks"></ol>
<div id="report"></div>

<script>
const $ = (id) => document.getElementById(id);

$("f").onsubmit = async (e) => {
  e.preventDefault();
  $("tasks").innerHTML = "";
  $("report").style.display = "none";

  const resp = await fetch("/api/research", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ question: $("q").value }),
  });

  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buf = "";
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });
    const parts = buf.split("\n\n");
    buf = parts.pop();
    for (const part of parts) {
      if (!part.startsWith("data: ")) continue;
      handle(JSON.parse(part.slice(6)));
    }
  }
};

function handle(ev) {
  if (ev.type === "plan") {
    for (const [i, t] of ev.tasks.entries()) {
      const li = document.createElement("li");
      li.id = "task-" + i;
      li.textContent = "⏳ " + t;
      $("tasks").appendChild(li);
    }
  }
  if (ev.type === "task_done") {
    const li = $("task-" + ev.index);
    li.textContent = "✅ " + li.textContent.slice(2);
  }
  if (ev.type === "reporting") {
    const li = document.createElement("li");
    li.textContent = "✍️ 正在撰写报告…";
    $("tasks").appendChild(li);
  }
  if (ev.type === "report") {
    $("report").textContent = ev.content;
    $("report").style.display = "block";
  }
  if (ev.type === "error") {
    const li = document.createElement("li");
    li.className = "err";
    li.textContent = "出错: " + ev.detail;
    $("tasks").appendChild(li);
  }
}
</script>
</body>
</html>

✅ 最终验收:

• [ ] 页面提交问题后先出现任务清单,⏳ 逐个变 ✅(注意变绿的顺序常常不是从上到下——这就是并行)
• [ ] 最后出现完整的 Markdown 报告,含摘要、章节、结论
• [ ] 断网(或搜索被墙)时,系统不崩,报告开头出现"信息可能不是最新"的降级声明
• [ ] 对照第 06 篇官方项目二:planner-executor、SSE 事件流、并发研究——骨架完全一致

8. 进阶练习

1. 二层深挖:研究员发现某个子任务信息不足时,让它自己再发起一轮追加搜索(把"是否需要补搜"作为一个 JSON 字段让模型输出)
2. 报告评审:加一个"评审 Agent",对报告按完整性/客观性打分,低于阈值就带着意见让报告员重写一版(Reflection)
3. 来源可点击:研究员的输出里带了 URL,把报告里的 [1] 渲染成超链接

9. 面试怎么讲

"我实现了一个 Plan-and-Execute 架构的研究 Agent:规划师用受约束的 JSON 输出拆解任务(带数量合法性校验),研究员并行联网搜索并只依据搜索结果作答以抑制幻觉,搜索源故障有降级路径;进度用 SSE 事件流实时推送,前端用索引对齐乱序完成的并行结果。"

下一篇:14 · 实战手敲四:记忆小镇(多NPC) | 返回目录:README