构建深度研究代理 - Docs by LangChain

概述

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

使用待办事项列表规划研究
将专注的研究任务委派给具有隔离上下文的子代理
在收集信息时评估搜索结果并规划下一步
将带有适当引用的发现综合成最终报告

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

关键概念

本教程涵盖：

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

先决条件

需要以下 API 密钥：

Anthropic (Claude) 或 Google (Gemini)
Tavily 用于网络搜索（可选 - 免费套餐足够）
LangSmith 用于跟踪（可选）

设置

创建项目目录

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

安装依赖

Claude
Gemini

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

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

设置 API 密钥

Claude
Gemini

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

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 发现，然后获取完整的网页内容，以便代理可以分析完整的来源，而不是摘要。

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"获取 {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"找到 {len(result_texts)} 个关于 '{query}' 的结果：\n\n" + "\n".join(
        result_texts
    )

添加提示词

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

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 研究论文：https://example.com/paper
 [2] 行业分析：https://example.com/analysis
"""

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] 上下文工程指南：https://example.com/context-guide
[2] AI 性能研究：https://example.com/study

协调器将把所有子代理的引用整合到最终报告中。
"""

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

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

## 委派策略

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

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

**明确的比较** -> 每个元素 1 个子代理：
- “比较 OpenAI、Anthropic 和 DeepMind 的 AI 安全方法” -> 3 个并行子代理
- “比较 Python 和 JavaScript 用于 Web 开发” -> 2 个并行子代理

**明显分离的方面** -> 每个方面 1 个子代理（谨慎使用）：
- “研究欧洲、亚洲和北美的可再生能源采用情况” -> 3 个并行子代理（地理分离）
- 仅当单个全面搜索无法有效涵盖这些方面时才使用此模式

## 关键原则
- **倾向于单个子代理**：一个全面的研究任务比多个狭窄的任务更节省令牌
- **避免过早分解**：不要将“研究 X”分解为“研究 X 概述”、“研究 X 技术”、“研究 X 应用” - 只需使用 1 个子代理研究所有 X
- **仅在明确比较时并行化**：在比较不同实体或地理分离的数据时使用多个子代理

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

## 研究限制
- 如果在 {max_researcher_iterations} 轮委派后仍未找到足够的来源，则停止
- 当你有足够信息可以全面回答时停止
- 倾向于专注研究而非穷尽探索"""

创建代理

将模型初始化和代理创建添加到 agent.py。选择你的提供商：

Claude
Gemini

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.1-pro-preview", temperature=0.0)

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

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 底部：

同步运行
流式更新

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)

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)

从项目根目录运行代理：

python agent.py

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

完整代码

在 GitHub 上查看完整的深度研究示例。

后续步骤

现在你已经构建了代理，可以通过更改代理文件中的提示常量来自定义它，以调整工作流、委派策略或研究员行为。你还可以调整委派限制，以允许更多并行子代理或委派轮次。有关本教程中概念的更多信息，请查看以下资源：

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

将这些文档通过 MCP 连接到 Claude、VSCode 等，以获取实时答案。

在 GitHub 上编辑此页面或提交问题。

​概述

​关键概念

​先决条件

​设置

​构建代理

​运行代理

​完整代码

​后续步骤

概述

关键概念

先决条件

设置

构建代理

运行代理

完整代码

后续步骤