构建深度研究代理

使用子代理委派机制构建多步骤网络研究代理

概述

本指南演示了如何使用 Deep Agents 从零开始构建一个多步骤网络研究代理。该代理将研究问题分解为聚焦的任务，委派给专门的子代理，并将发现综合成一份全面的报告。

您构建的代理将完成以下工作：

使用待办事项列表规划研究
将聚焦的研究任务委派给具有隔离上下文的子代理
评估搜索结果并根据信息收集情况规划下一步
将发现结果与正确的引用整合到最终报告中

生成的子代理将使用 Tavily 进行网络搜索，并获取完整的网页内容进行分析。

关键概念

本教程涵盖：

子代理（Subagents）——用于并行、上下文隔离的研究
自定义工具（tools）——用于网络搜索
使用内置规划工具进行多步骤规划

前提条件

需要以下 API 密钥：

Anthropic（Claude）或 Google（Gemini）
Tavily——用于网络搜索（可选，免费额度足够）
LangSmith——用于追踪（可选）

环境设置

1. 创建项目目录

bash

mkdir deep-research-agent
cd deep-research-agent

2. 安装依赖

Claude（Anthropic）：

bash

# pip
pip install deepagents tavily-python httpx markdownify langchain-anthropic langchain-core

# uv
uv init
uv add deepagents tavily-python httpx markdownify langchain-anthropic langchain-core
uv sync

Gemini（Google）：

bash

# pip
pip install deepagents tavily-python httpx markdownify langchain-google-genai langchain-core

# uv
uv init
uv add deepagents tavily-python httpx markdownify langchain-google-genai langchain-core
uv sync

3. 设置 API 密钥

Claude：

bash

export ANTHROPIC_API_KEY="your_anthropic_api_key"
export TAVILY_API_KEY="your_tavily_api_key"
export LANGSMITH_API_KEY="your_langsmith_api_key"   # 可选

Gemini：

bash

export GOOGLE_API_KEY="your_google_api_key"
export TAVILY_API_KEY="your_tavily_api_key"
export LANGSMITH_API_KEY="your_langsmith_api_key"   # 可选

构建代理

在项目目录中创建 agent.py：

第一步：添加工具

添加自定义搜索工具。tavily_search 工具使用 Tavily 进行 URL 发现，然后抓取完整的网页内容，使代理能够分析完整的来源信息而非摘要。

python

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:
    """抓取网页并将 HTML 转换为 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:
    """在网络上搜索给定查询的信息。

    使用 Tavily 发现相关 URL，然后抓取并返回完整的网页内容（Markdown 格式）。

    Args:
        query: 要执行的搜索查询
        max_results: 返回的最大结果数（默认：1）
        topic: 主题筛选 - 'general'（通用）、'news'（新闻）或 'finance'（金融）（默认：'general'）

    Returns:
        带有完整网页内容的格式化搜索结果
    """
    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
    )

第二步：添加提示模板

将编排器工作流和子代理的提示模板添加到 agent.py：

研究工作流指令：

python

RESEARCH_WORKFLOW_INSTRUCTIONS = """# 研究工作流

所有研究请求请遵循此工作流：

1. **规划**：使用 write_todos 创建待办事项列表，将研究分解为聚焦的任务
2. **保存请求**：使用 write_file() 将用户的研究问题保存到 `/research_request.md`
3. **研究**：使用 task() 工具将研究任务委派给子代理 - 始终使用子代理进行研究，不要自行研究
4. **综合**：审阅所有子代理的发现，合并引用（每个唯一 URL 在所有发现中分配一个编号）
5. **撰写报告**：将综合的最终报告写入 `/final_report.md`（参见下方报告撰写指南）
6. **验证**：读取 `/research_request.md`，确认你已涵盖所有方面，并带有正确的引用和结构

## 研究规划指南
- 将类似的研究任务合并到一个 TODO 中以减少开销
- 对于简单的事实查询，使用 1 个子代理
- 对于比较或多方面主题，委派给多个并行子代理
- 每个子代理应研究一个特定方面并返回发现

## 报告撰写指南

当撰写最终报告到 `/final_report.md` 时，遵循以下结构模式：

**对于比较类报告：**
1. 引言
2. 主题 A 概述
3. 主题 B 概述
4. 详细比较
5. 结论

**对于列表/排名类报告：**
直接列出条目并附上详情——无需引言：
1. 条目 1 及说明
2. 条目 2 及说明
3. 条目 3 及说明

**对于总结/概述类报告：**
1. 主题概述
2. 关键概念 1
3. 关键概念 2
4. 关键概念 3
5. 结论

**通用指南：**
- 使用清晰的章节标题（## 用于章节，### 用于小节）
- 默认使用段落形式——以文字为主，不要仅是项目符号列表
- 不要使用自我指涉的语言（"我发现……"、"我研究了……"）
- 以专业报告的形式撰写，不要添加元评论
- 每个章节应全面且详细
- 仅当列表形式比散文更合适时使用项目符号

**引用格式：**
- 使用 [1]、[2]、[3] 格式进行内联引用
- 为所有子代理发现中的每个唯一 URL 分配一个引用编号
- 在报告末尾使用 ### 来源 章节列出每个编号的来源
- 来源编号连续无间断（1,2,3,4……）
- 格式：[1] 来源标题: URL（每行一个来源，确保列表渲染正确）
- 示例：

 某些重要发现 [1]。另一个关键见解 [2]。

 ### 来源
 [1] AI Research Paper: https://example.com/paper
 [2] Industry Analysis: https://example.com/analysis
"""

研究员指令：

python

RESEARCHER_INSTRUCTIONS = """你是一名研究助手，正在对用户输入的主题进行研究。作为背景信息，今天的日期是 {date}。

你的工作是使用工具收集关于用户输入主题的信息。
你可以使用 tavily_search 工具来查找有助于回答研究问题的资源。
你可以串行或并行调用它，你的研究是在一个工具调用循环中进行的。

你可以使用 tavily_search 工具进行网络搜索。

像一位时间有限的人类研究员一样思考。遵循以下步骤：

1. **仔细阅读问题**——用户需要哪些具体信息？
2. **从较广泛的搜索开始**——首先使用广泛、全面的查询
3. **每次搜索后，暂停并评估**——我有足够的信息来回答吗？还缺少什么？
4. **在收集信息时执行更精准的搜索**——填补信息空白
5. **当你可以自信地回答时停止**——不要追求完美而不断搜索

**工具调用预算**（防止过度搜索）：
- **简单查询**：最多使用 2-3 次搜索工具调用
- **复杂查询**：最多使用 5 次搜索工具调用
- **始终停止**：如果在 5 次搜索工具调用后仍无法找到正确的来源

**立即停止的条件**：
- 你可以全面回答用户的问题
- 你已有 3 个或更多相关的示例/来源
- 你最近 2 次搜索返回了相似的信息

每次搜索后，在继续之前评估结果：我找到了哪些关键信息？还缺少什么？我有足够的信息来回答吗？我是否应该继续搜索还是提供答案？

当向编排器反馈你的发现时：

1. **结构化你的响应**：使用清晰的标题和详细说明来组织发现
2. **内联引用来源**：在引用来自搜索的信息时使用 [1]、[2]、[3] 格式
3. **包含来源章节**：在末尾使用 ### 来源 列出每个编号的来源及其标题和 URL

示例：
## 主要发现

上下文工程是 AI 代理的一项关键技术 [1]。研究表明，适当的上下文管理可以将性能提升 40% [2]。

### 来源
[1] Context Engineering Guide: https://example.com/context-guide
[2] AI Performance Study: https://example.com/study

编排器将合并所有子代理的引用到最终报告中。
"""

子代理委派指令：

python

SUBAGENT_DELEGATION_INSTRUCTIONS = """# 子代理研究协调

你的角色是通过将 TODO 列表中的任务委派给专门的研究子代理来协调研究。

## 委派策略

**默认：大多数查询从 1 个子代理开始：**
- "什么是量子计算？" -> 1 个子代理（通用概述）
- "列出旧金山排名前 10 的咖啡馆" -> 1 个子代理
- "总结互联网的历史" -> 1 个子代理
- "研究 AI 代理的上下文工程" -> 1 个子代理（涵盖所有方面）

**仅在查询明确需要比较或有明显独立的方面时才使用并行：**

**明确比较** -> 每个元素 1 个子代理：
- "比较 OpenAI vs Anthropic vs DeepMind 的 AI 安全方法" -> 3 个并行子代理
- "比较 Python 与 JavaScript 在 Web 开发中的应用" -> 2 个并行子代理

**明确分离的方面** -> 每个方面 1 个子代理（谨慎使用）：
- "研究欧洲、亚洲和北美的可再生能源采用情况" -> 3 个并行子代理（按地域划分）
- 仅当单个综合搜索无法高效覆盖这些方面时使用此模式

## 关键原则
- **倾向于单个子代理**：一个综合性的研究任务比多个狭窄的任务更加节省 token
- **避免过早分解**：不要将"研究 X"拆分为"研究 X 概述"、"研究 X 技术"、"研究 X 应用"——只需使用 1 个子代理完成所有 X 的研究
- **仅为明确比较进行并行化**：在比较不同实体或地理上分离的数据时使用多个子代理

## 并行执行限制
- 每轮迭代最多使用 {max_concurrent_research_units} 个并行子代理
- 在单次响应中进行多次 task() 调用以启用并行执行
- 每个子代理独立返回发现

## 研究限制
- 如果未找到足够的来源，请在 {max_researcher_iterations} 轮委派后停止
- 当拥有足够的信息可以全面回答时停止
- 倾向于聚焦的研究而非穷举式探索
"""

第三步：创建代理

添加模型初始化和代理创建代码到 agent.py。根据您的提供商选择对应代码：

Claude / OpenAI / Anthropic（使用 init_chat_model）：

python

from datetime import datetime

from deepagents import create_deep_agent
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": "将研究委派给子代理。一次给出一个主题。",
    "system_prompt": RESEARCHER_INSTRUCTIONS.format(date=current_date),
    "tools": [tavily_search],
}

# 选择合适的模型：
# model = init_chat_model(model="google_genai:gemini-3.5-flash", temperature=0.0)
# model = init_chat_model(model="openai:gpt-5.4", temperature=0.0)
# model = init_chat_model(model="anthropic:claude-sonnet-4-6", temperature=0.0)
# model = init_chat_model(model="openrouter:anthropic/claude-sonnet-4-6", temperature=0.0)
# model = init_chat_model(model="fireworks:accounts/fireworks/models/qwen3p5-397b-a17b", temperature=0.0)
# model = init_chat_model(model="baseten:zai-org/GLM-5", temperature=0.0)
# model = init_chat_model(model="ollama:devstral-2", temperature=0.0)

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

Gemini（直接使用 ChatGoogleGenerativeAI）：

python

from datetime import datetime

from langchain_google_genai import ChatGoogleGenerativeAI
from deepagents import create_deep_agent

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": "将研究委派给子代理。一次给出一个主题。",
    "system_prompt": RESEARCHER_INSTRUCTIONS.format(date=current_date),
    "tools": [tavily_search],
}

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview", temperature=0.0)

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

运行代理

您可以同步运行代理（等待完整结果后打印），也可以流式接收更新。在 agent.py 底部添加相应代码。

同步运行

python

from langchain.messages import HumanMessage

if __name__ == "__main__":
    result = agent.invoke(
        {
            "messages": [
                HumanMessage(
                    content="RAG 与微调在 LLM 应用中的主要区别是什么？"
                )
            ]
        }
    )
    for msg in result.get("messages", []):
        if hasattr(msg, "content") and msg.content:
            print(msg.content)

流式更新

python

from langchain.messages import HumanMessage
from langgraph.types import Overwrite

if __name__ == "__main__":
    for chunk in agent.stream(
        {
            "messages": [
                HumanMessage(
                    content="比较 Python 与 JavaScript 在 Web 开发中的应用"
                )
            ]
        },
        stream_mode="updates",
    ):
        for node, update in chunk.items():
            if not update or not (messages := update.get("messages")):
                continue
            msg_list = messages.value if isinstance(messages, Overwrite) else messages
            for msg in msg_list:
                if hasattr(msg, "content") and msg.content:
                    print(msg.content)

执行

从项目根目录运行：

bash

python agent.py

如果您在运行前设置了 LANGSMITH_API_KEY 环境变量，可以在 LangSmith 中查看代理的追踪记录，以调试和监控多步骤行为。

完整代码

查看 GitHub 上的完整 Deep Research 示例。

后续步骤

现在您已经构建了代理，可以通过修改 agent.py 中的提示常量来自定义工作流、委派策略或研究员行为。您还可以调整委派限制以允许更多的并行子代理或委派轮次。

有关本教程中涉及的概念的更多信息，请查阅以下资源：

子代理（Subagents）：了解如何配置具有不同工具和提示的子代理
自定义（Customization）：自定义模型、工具、系统提示和规划行为
LangSmith：追踪研究运行并调试多步骤行为
Deep Research Course：关于使用 LangGraph 进行深度研究的完整课程

下一步

调整 max_concurrent_research_units 和 max_researcher_iterations 参数以控制并行度和研究深度
尝试不同的基础模型（Claude、Gemini、GPT 等）并比较研究质量
扩展 tavily_search 工具以支持更多数据源或自定义搜索参数
连接 MCP 工具，在 Claude、VSCode 等环境中实时获取文档帮助

构建深度研究代理 ​

概述 ​

关键概念 ​

前提条件 ​

环境设置 ​

1. 创建项目目录 ​

2. 安装依赖 ​

3. 设置 API 密钥 ​

构建代理 ​

第一步：添加工具 ​

第二步：添加提示模板 ​

第三步：创建代理 ​

运行代理 ​

同步运行 ​

流式更新 ​

执行 ​

完整代码 ​

后续步骤 ​

下一步 ​

构建深度研究代理

概述

关键概念

前提条件

环境设置

1. 创建项目目录

2. 安装依赖

3. 设置 API 密钥

构建代理

第一步：添加工具

第二步：添加提示模板

第三步：创建代理

运行代理

同步运行

流式更新

执行

完整代码

后续步骤

下一步