《Deep Agents 实战笔记》第 11 篇 | 基于 deepagents (Python) 官方文档 原文:https://docs.langchain.com/oss/python/deepagents/deep-research

前 10 篇把机制拆了个遍:文件系统、Backend、子代理、上下文工程、权限。从这篇开始进入实战——第一个项目,就是把官方 Deep Research 教程完整精读一遍,亲手复刻一个"多子代理分工 + 联网搜索 + 结构化报告"的深度研究代理。

我特意说"精读"而不是"照抄":这个教程的代码量不大,真正值钱的是提示词的设计。主代理怎么被约束成"只协调不动手"、研究员子代理怎么被教会"像人类研究员一样止损",这些细节才是 deep 与不深的分水岭。

一、总体架构:一个监督者 + 一群一次性研究员

先看全局。整个系统只有两类角色:一个主代理(编排者),和一类可被反复派发的 research-agent 子代理。分工非常清晰——

flowchart TB
    U["用户研究请求"] --> M
    subgraph M["主代理:编排者"]
        direction TB
        P["TodoListMiddleware<br/>write_todos 拆解任务"]
        W["工作流指令<br/>规划 / 保存请求 / 综合 / 写报告"]
        D["委派指令<br/>何时并行、几轮上限"]
    end
    M -->|"task:聚焦子问题"| SA1
    M -->|"task:聚焦子问题"| SA2
    subgraph ISO["独立上下文:research-agent"]
        SA1["研究员 A<br/>tavily_search 多轮搜索"]
        SA2["研究员 B<br/>tavily_search 多轮搜索"]
    end
    SA1 -->|"结构化发现 + 引用"| M
    SA2 -->|"结构化发现 + 引用"| M
    M --> R["综合引用去重<br/>写入 final_report.md"]
    style M fill:#eef4ff,color:#1a3a8b,stroke:#4a72d9
    style ISO fill:#f7f7f7,color:#333333,stroke:#999999
    style SA1 fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a
    style SA2 fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a

正在渲染图表…

注意一个容易被忽略的点:主代理的工具列表里也有 tavily_search,但提示词明确要求它"永远用子代理做研究,自己绝不研究"。这是刻意设计的角色纪律——编排者一旦亲自下场搜索,它的上下文就会被原始网页内容污染,监督和综合的质量随即崩塌。

二、搜索工具精读:不是"搜索",是"搜索 + 抓全文"

教程的 tavily_search 和第 2 篇 Quickstart 里的简版有两个关键差异,值得逐行看:

import os
from typing import Annotated, Literal

import httpx
from langchain.tools import InjectedToolArg, tool
from markdownify import markdownify
from tavily import TavilyClient

tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

def fetch_webpage_content(url: str, timeout: float = 10.0) -> str:
    """Fetch webpage and convert HTML to markdown."""
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    try:
        response = httpx.get(url, headers=headers, timeout=timeout)
        response.raise_for_status()
        return markdownify(response.text)
    except Exception as e:
        return f"Error fetching {url}: {e!s}"

@tool(parse_docstring=True)
def tavily_search(
    query: str,
    max_results: Annotated[int, InjectedToolArg] = 1,
    topic: Annotated[
        Literal["general", "news", "finance"], InjectedToolArg
    ] = "general",
) -> str:
    """Search the web for information on a given query.

    Uses Tavily to discover relevant URLs, then fetches and returns full webpage content as markdown.
    """
    search_results = tavily_client.search(
        query,
        max_results=max_results,
        topic=topic,
    )
    result_texts = []
    for result in search_results.get("results", []):
        url = result["url"]
        title = result["title"]
        content = fetch_webpage_content(url)
        result_texts.append(f"## {title}\n**URL:** {url}\n\n{content}\n---")

    return f"Found {len(result_texts)} result(s) for '{query}':\n\n" + "\n".join(
        result_texts
    )

差异一:Tavily 只负责发现 URL,工具随后用 httpx 抓取完整网页并转成 markdown。研究员拿到的是原始资料全文,而不是搜索引擎给的摘要——分析的深度完全不同。

差异二:max_results 和 topic 用 InjectedToolArg 标注,从模型的参数空间里"藏"起来。模型只需要给 query,其余由代码控制。这是控制模型行为发散度的小技巧:每条查询返回几条结果,不该由模型每次临时起意。

代价也很直白:返回的是全文,单次调用可能就是几千 token。所以这个工具必须和子代理搭配使用——原始内容留在子代理的隔离上下文里消化,主代理永远只收摘要。

三、三段提示词的分工设计

教程把提示词拆成三段,用 format() 注入运行参数后拼接。这是本篇最值得学的部分。

第一段:主代理的工作流指令(RESEARCH_WORKFLOW_INSTRUCTIONS)。六步固定流程,一步都不许自由发挥:

1. Plan:用 write_todos 把研究拆成聚焦任务
2. Save the request:把用户问题写入 /research_request.md
3. Research:用 task() 委派子代理——永远委派,绝不亲自研究
4. Synthesize:汇总所有发现,合并引用(每个 URL 全局只编一个号)
5. Write Report:把最终报告写入 /final_report.md
6. Verify:重读 /research_request.md,确认每个方面都覆盖、引用规范

两个细节有观点。其一,把用户原始问题存进文件再在最后一步读回来校验——这是用文件系统当"验收清单",防跑偏;其二,提示词里内置了报告结构模板(对比类怎么写、清单类怎么写、综述类怎么写)和引用格式([1] [2] 内联标注 + 文末 ### Sources 列表),把"写作规范"也固化了。

第二段:委派策略(SUBAGENT_DELEGATION_INSTRUCTIONS)。这段解决"什么时候该并行",立场非常明确:

默认:绝大多数问题只用 1 个子代理
只有查询明确要求对比、或各方面显然独立时才并行
最多 {max_concurrent_research_units} 个并行子代理
最多 {max_researcher_iterations} 轮委派
宁可聚焦,不做穷举

反直觉但正确:并行是有成本的——每个子代理都要从头建立认知,一个全面的任务往往比多个狭窄任务更省 token。教程甚至专门警告"不要把'研究 X'拆成'研究 X 概述''研究 X 技术''研究 X 应用'",这是典型的过早分解。

第三段:研究员子代理指令(RESEARCHER_INSTRUCTIONS)。这段是我见过写得最好的"搜索止损"教学:

像时间有限的人类研究员一样思考:
1. 仔细读问题——用户到底需要什么?
2. 先宽后窄——先用宽泛查询,再逐步收窄
3. 每次搜索后停下来评估——够了吗?还缺什么?
4. 能自信回答时立即停止——不要追求完美

工具调用预算:
- 简单问题最多 2-3 次搜索
- 复杂问题最多 5 次
- 5 次之后无论如何停止

立即停止的信号:
- 已能全面回答 / 已有 3 个以上相关来源
- 最近 2 次搜索返回的信息雷同

没有这段,子代理的经典失败模式是无限搜索——模型默认"再多搜一次总没坏处",直到把预算烧光。给明确的数字预算和停止信号,比说"要节约"有效一百倍。

四、组装与运行:同步 / 流式两种姿势

三段提示词拼装成主代理的 system_prompt,子代理用字典定义(name + description + system_prompt + tools):

from datetime import datetime

from deepagents import create_deep_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain.chat_models import init_chat_model

max_concurrent_research_units = 3
max_researcher_iterations = 3

current_date = datetime.now().strftime("%Y-%m-%d")

INSTRUCTIONS = (
    RESEARCH_WORKFLOW_INSTRUCTIONS
    + "\n\n"
    + "=" * 80
    + "\n\n"
    + SUBAGENT_DELEGATION_INSTRUCTIONS.format(
        max_concurrent_research_units=max_concurrent_research_units,
        max_researcher_iterations=max_researcher_iterations,
    )
)

research_sub_agent = {
    "name": "research-agent",
    "description": "Delegate research to the sub-agent. Give one topic at a time.",
    "system_prompt": RESEARCHER_INSTRUCTIONS.format(date=current_date),
    "tools": [tavily_search],
}

model = init_chat_model(model="anthropic:claude-sonnet-4-5-20250929", temperature=0.0)

agent = create_deep_agent(
    model=model,
    tools=[tavily_search],
    system_prompt=INSTRUCTIONS,
    subagents=[research_sub_agent],
    middleware=[TodoListMiddleware()],
)

注意 TodoListMiddleware 是可选启用的(任务规划并非默认开启),工作流第一步要用 write_todos,所以必须显式传入。子代理的提示词里 format(date=current_date) 注入了当天日期——搜索类任务对时效敏感,这个细节别省。

同步运行:等完整结果,适合脚本和批处理。

from langchain.messages import HumanMessage

if __name__ == "__main__":
    result = agent.invoke(
        {
            "messages": [
                HumanMessage(
                    content="What are the main differences between RAG and fine-tuning for LLM applications?"
                )
            ]
        }
    )

    for msg in result.get("messages", []):
        if hasattr(msg, "content") and msg.content:
            print(msg.content)

流式运行:stream_events(version="v3") 逐 token 实时输出,长任务的观感完全不同。

from langchain.messages import HumanMessage

if __name__ == "__main__":
    stream = agent.stream_events(
        {
            "messages": [
                HumanMessage(content="Compare Python vs JavaScript for web development")
            ]
        },
        version="v3",
    )
    for message in stream.messages:
        for token in message.text:
            print(token, end="", flush=True)

一次完整运行的时序长这样:

sequenceDiagram
    participant U as 用户
    participant M as 主代理
    participant SA as research-agent
    participant T as tavily_search
    participant FS as 文件系统

    U->>M: 研究问题
    M->>FS: write_todos 拆解任务
    M->>FS: 保存问题到 research_request.md
    M->>SA: task:聚焦子问题
    SA->>T: 宽泛搜索
    T-->>SA: 网页全文(markdown)
    SA->>T: 收窄搜索(评估缺口后)
    SA-->>M: 结构化发现 + 编号引用
    M->>SA: 第二轮 task(如证据不足,最多 3 轮)
    SA-->>M: 发现 + 引用
    M->>FS: 引用合并去重,写 final_report.md
    M->>U: 报告完成

正在渲染图表…

五、效果分析与调优

跑通只是及格,这一节聊聊怎么调。

调优点一:子代理 description 的迭代。 research_sub_agent 里的 description 是主代理决定"何时、如何调用这个子代理"的唯一依据。官方给它的写法是 "Delegate research to the sub-agent. Give one topic at a time."——最后一句话不是废话,它在约束主代理一次只派一个主题,防止把三个问题打包塞进一个子代理导致研究失焦。description 写坏了,整个委派行为就歪了:写得含糊,主代理可能自己动手研究;写得太宽,可能一次委派塞进多个主题。调优的姿势就是"跑一次、看 LangSmith trace、改一版",本质是个迭代循环:

flowchart LR
    A["编写 description<br/>与委派指令"] --> B["运行一次研究任务"]
    B --> C["查看 LangSmith trace"]
    C --> D{"委派行为符合预期?"}
    D -->|"是"| E["固化提示词版本"]
    D -->|"否"| F["修订 description /<br/>委派策略 / 预算参数"]
    F --> B
    style D fill:#fdf3ff,color:#504b5f,stroke:#7e65ae
    style E fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a

正在渲染图表…

调优点二:两个数字阀门。 max_concurrent_research_units = 3 控制并行宽度,max_researcher_iterations = 3 控制委派轮数。这两个数字直接写进提示词,模型会严格遵守。预算紧张就调小,追求覆盖面就调大——但每加一个单位都是真金白银的 token,建议从默认值出发按需加。

调优点三:报告质量看引用。 教程要求"每个唯一 URL 在全部发现中只占一个编号",主代理要跨子代理去重合并引用。如果报告出现引用断号或同一来源两个编号,优先检查综合阶段的提示词有没有被你改坏。

六、实战作业

把这套骨架改造成你自己的垂直研究代理,二选一:

  1. 竞品分析 Agent:工作流指令改为"每个竞品一个子代理"(这正是教程里"显式对比才并行"的合法场景),报告模板改为产品对比结构:概述 → 逐个竞品 → 详细对比 → 结论。
  2. 论文调研 Agent:把 tavily_search 换成 arXiv / Semantic Scholar 的 API 工具,研究员指令的"止损信号"改为"已找到 3 篇高度相关论文即停止",报告模板加上"方法、数据集、结论"三段式。

改造时只动提示词和工具,create_deep_agent 的骨架一行不用变——这就是这个架构的可迁移性。

本篇踩坑点

  • 主代理亲自搜索是事故之源:工具列表里保留 tavily_search 是为了兜底,但如果 trace 里看到主代理大量直接调用搜索,说明工作流指令没镇住它,优先强化"永远委派"那句;
  • write_todos 报工具不存在:任务规划是可选启用的,忘了传 TodoListMiddleware() 就会这样;
  • 抓全文导致超时:fetch_webpage_content 默认 10 秒超时,慢站点会返回错误字符串而非崩溃,但大量失败会让研究员"以为没有资料"——trace 里注意甄别;
  • 子代理不停止:如果发现单个子代理搜了十几次,回到研究员指令检查"工具调用预算"那几行是否被精简掉了。

下一篇预告:Deep Research 解决"信息从哪来",第 12 篇解决"答案凭什么可信"——Agentic RAG,当检索遇上深度智能体:索引、分块卸载、子代理并行分析,再加一个用评分标准(rubric)自我迭代的评审判官。


参考资料: