基础介绍:

Deep Agents 是 LangChain 官方推出的一套开箱即用(Batteries Included)的 Agent 开发框架,专门面向复杂任务、长流程执行、多步骤规划、多 Agent 协作以及需要上下文工程的应用场景而设计;

相比传统 Agent 需要开发者自行组合 Prompt、Tool、Memory、Workflow、Checkpoint、Sub Agent 等组件,Deep Agents 已经内置了任务规划、虚拟文件系统、子 Agent 调度、上下文压缩、长期记忆、人工审批、结构化输出等核心能力;开发者可以把更多精力放在业务逻辑上,而不是从零搭建 Agent 运行框架;

本文基于 Deep Agents 当前官方能力整理,示例统一使用 DeepSeek OpenAI 兼容接口;DeepSeek 官方当前推荐的 OpenAI 兼容 base_url 是 https://api.deepseek.com,示例模型使用 deepseek-v4-flash;如果你使用 OpenAI、Anthropic、Google、OpenRouter 或本地模型,只需要替换模型初始化方式

什么是 Deep Agents:

Deep Agents 可以理解为一个面向复杂任务的高级 Agent Harness

它不是一个简单的 "LLM + Tool 调用循环",而是在 LangGraph 运行时之上封装了一套更完整的 Agent 基础设施

安装完成后,即可直接获得以下能力:

  • 自动任务规划(Planning)
  • 虚拟文件系统读写(Filesystem)
  • 子 Agent 协作(Sub Agents)
  • 上下文压缩与管理(Context Management)
  • 长任务执行(Long Running Tasks)
  • 人工审批(Human-in-the-loop)
  • 持久化记忆(Memory)
  • 按需加载 Skills
  • 结构化输出(Structured Output)
  • 流式输出(Streaming)
  • 可插拔 Middleware

Deep Agents 的核心目标,是把复杂 Agent 中最常见、最容易重复造轮子的工程问题抽象出来;

传统 ReAct Agent 通常是:

用户输入 → LLM 推理 → 调用工具 → LLM 推理 → 调用工具 → 输出结果;

当任务比较小时,这种模式运行良好;但随着任务复杂度提升,例如自动生成市场调研报告、分析多家公司财报、编写完整项目代码、长时间检索与整理资料,就会逐渐暴露出以下问题:

  • 上下文窗口迅速膨胀,工具输出把消息历史撑爆
  • 推理链过长,模型容易忘记前面已经完成的工作
  • 多个子任务混在同一上下文里,相互干扰
  • 工具调用逻辑越来越复杂,难以维护
  • 关键操作缺少人工审批,安全边界不清晰
  • 多轮会话和跨线程记忆需要额外持久化设计

Deep Agents 的价值,正是把这些问题变成默认能力,让开发者从 "把 Agent 跑起来" 进入到 "把业务 Agent 做好"

与 LangChain、LangGraph 的关系:

很多初学者容易混淆 LangChain、LangGraph 和 Deep Agents,它们不是相互替代关系,而是位于不同抽象层级

框架层级定位核心职责适用场景
LangChain基础组件层提供 Model、Prompt、Tool、Retriever、Output Parser、Middleware 等 AI 应用组件构建模型调用、工具调用、检索增强和普通 Agent
LangGraph运行时层提供状态管理、节点与边、Checkpoint、Interrupt、持久化和可恢复执行需要自定义图结构和执行流程的复杂 Agent 或 Workflow
Deep Agents应用框架层在 LangGraph 之上封装规划、文件系统、子 Agent、上下文压缩、记忆和审批快速构建复杂多步骤 Agent,无需从头设计架构

可以这样理解:LangChain 提供积木,LangGraph 负责把积木组织成可持续运行、可恢复、可追踪的流程,Deep Agents 则是在 LangGraph 之上预先搭好一套复杂 Agent 的通用底座

如果应用只是一次模型调用,LangChain 就够用;如果需要精确控制状态、节点和执行路径,应该直接学习 LangGraph;如果希望快速得到一个具备规划、文件系统、子 Agent 和上下文压缩能力的生产级 Agent,则可以从 Deep Agents 入手

核心能力:

Deep Agents 默认会装配一组中间件,开发者不需要手动实现这些基础能力;

能力说明核心组件
任务规划通过 write_todos 工具维护任务清单,将复杂任务拆成有序步骤TodoListMiddleware
虚拟文件系统提供 ls、read_file、write_file、edit_file、glob、grep 等文件工具;部分版本和后端支持 delete、executeFilesystemMiddleware
子 Agent通过 task 工具把子任务委派给独立上下文的专用 AgentSubAgentMiddleware
上下文压缩自动总结历史消息,避免长任务超出上下文窗口SummarizationMiddleware
Tool Call 修补对部分模型的工具调用格式做兼容修补PatchToolCallsMiddleware
异步子 Agent支持远程或后台子 Agent 任务AsyncSubAgentMiddleware
Human-in-the-loop在关键工具调用前暂停,等待人工审批、修改或拒绝HumanInTheLoopMiddleware
长期记忆加载 AGENTS.md 文件作为跨会话上下文MemoryMiddleware
Skills按需加载可复用领域知识、工作流和操作规范SkillsMiddleware

注意:tools 参数传入的是额外工具,它不会移除 Deep Agents 内置工具,如果需要隐藏或禁用内置工具,需要使用更高级的 Harness Profile 配置

环境准备:

Deep Agents 发布在 PyPI,包名为 deepagents;当前 Deep Agents 要求 Python 版本为 >=3.11,<4.0;建议使用 Python 3.11、3.12 或 3.13

使用 uv 安装(推荐):

uv init deepagents-demo
cd deepagents-demo
uv add deepagents langchain-openai langgraph tavily-python python-dotenv

使用 pip 安装:

pip install deepagents langchain-openai langgraph tavily-python python-dotenv

Deep Agents 对模型最重要的要求是:模型必须支持 Tool Calling

如果使用 DeepSeek,需要配置 DeepSeek API Key

macOS / Linux:

export DEEPSEEK_API_KEY="sk-your-deepseek-api-key"
export TAVILY_API_KEY="tvly-your-tavily-api-key"

Windows PowerShell:

$env:DEEPSEEK_API_KEY="sk-your-deepseek-api-key"
$env:TAVILY_API_KEY="tvly-your-tavily-api-key"

DeepSeek 的 API Key 可在 https://platform.deepseek.com 申请,Tavily API Key 在 https://www.tavily.com/ 申请

模型配置:

Deep Agents 的 model 参数支持两种常见写法,第一种是直接传入模型字符串,例如:

agent = create_deep_agent(
    model="openai:gpt-5.5",
)

这种写法适合 LangChain 已经支持的 provider

第二种是传入已经初始化好的模型实例,使用 DeepSeek、OpenRouter、本地兼容 OpenAI 的服务时,这种方式更清晰:

import os
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

常见模型配置可以参考下表:

使用场景推荐写法说明
OpenAI 官方模型model="openai:gpt-5.5" 或 init_chat_model(...)适合直接使用 LangChain provider 字符串
Anthropic 模型model="anthropic:claude-sonnet-4-6"需要配置 Anthropic API Key
DeepSeekChatOpenAI(..., base_url="https://api.deepseek.com")DeepSeek 兼容 OpenAI 接口,建议传模型实例
OpenRouter使用对应 LangChain provider 或 OpenAI 兼容模型实例注意模型是否支持工具调用
本地模型通过 OpenAI 兼容服务或对应 LangChain 集成接入必须确认 Tool Calling 能正常工作

参数说明:

create_deep_agent 是创建 Deep Agent 的核心函数,常用参数如下:

参数类型是否必填说明建议
modelstr 或模型实例建议必填模型字符串,或初始化好的 Chat Model生产中显式传入
tools列表否自定义工具函数、LangChain Tool 或 tool schema只放业务工具
system_promptstr / SystemMessage / 配置字典否自定义系统提示词用来定义角色和行为
middleware列表否追加自定义中间件高级场景使用
subagents列表否同步、异步或已编译子 Agent 定义复杂任务拆分时使用
skills路径列表否Skills 目录路径领域知识按需加载
memory路径列表否AGENTS.md 文件路径项目规则和长期偏好
permissions权限规则列表否文件系统工具的路径权限规则生产或敏感路径必须配置
backendBackend 实例或工厂否控制文件存储和执行环境默认 StateBackend()
interrupt_on字典否指定哪些工具调用需要人工审批需配合 checkpointer
response_formatPydantic / TypedDict / Schema否结构化输出格式报告、表单、JSON 输出时使用
state_schemaDeepAgentState 子类否扩展 Agent 状态结构高级状态管理使用
context_schema类型定义否定义运行时上下文多租户或请求级上下文使用
checkpointerCheckpointer否保存对话和中断状态多轮对话、HITL 必须配置
storeBaseStore否长期存储后端持久化文件或跨线程共享
debugbool否输出调试信息本地排错时开启
namestr否Agent 名称多 Agent 观测时有用
cacheBaseCache否缓存配置高级性能优化

system_prompt 还有一种更精细的配置方式:

agent = create_deep_agent(
    model=model,
    system_prompt={
        "prefix": "你是一位专业的研究助手。",
        "suffix": "回答必须结构清晰,优先给出可验证结论。",
    },
)

prefix 会放在 Deep Agents 默认系统提示词之前,suffix 会放在默认提示词之后;如果希望完全替换默认系统提示词,可以传入 base,或者把 base 设置为 None

文件系统后端对比:

Deep Agents 的文件能力由 Backend 控制,Backend 决定 Agent 读写的文件放在哪里,以及是否能执行命令

后端存储位置跨线程持久化是否能执行 Shell适用场景
StateBackendLangGraph 图状态内否,同一 thread 内有效否默认后端,本地学习和简单 Demo
FilesystemBackend本机磁盘是否可信本地开发、需要真实读写文件
StoreBackendLangGraph Store是否跨 Thread 共享文件、多用户持久化
LocalShellBackend本机磁盘是是,本机直接执行只适合完全受信任的本地开发
CompositeBackend按路径路由到多个后端取决于子后端取决于子后端不同目录使用不同存储策略

初学者直接使用默认的 StateBackend 即可;FilesystemBackend 会让 Agent 直接读写真实文件,LocalShellBackend 会让 Agent 在本机执行 Shell 命令,没有沙箱隔离;除非非常清楚风险,否则不要在生产环境或不可信输入场景中使用

如果必须使用真实文件系统,至少要配合:

  • permissions
  • Human-in-the-loop
  • 独立工作目录
  • 不包含密钥的环境变量
  • 清晰的操作日志和回滚策略

快速开始:

本节演示如何用最少的代码创建一个可以运行的 Deep Agent

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)


def get_weather(city: str) -> str:
    """查询指定城市的天气。"""
    return f"{city} 今天晴天,气温 25 摄氏度,适合出行。"


agent = create_deep_agent(
    model=model,
    tools=[get_weather],
    system_prompt="你是一个友好的助手,能够查询天气信息。",
)

result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "南昌今天天气怎么样?"}
        ]
    }
)

print(result["messages"][-1].content)
# 运行脚本
python hello_agent.py
# 输出内容:
# 南昌今天**晴天**,气温 **25°C**,非常适合出行。

这个示例里,get_weather 是一个普通 Python 函数,Deep Agents 会把它注册为 Agent 可调用的工具,函数参数类型和 docstring 会帮助模型理解这个工具什么时候使用

自定义系统提示词:

通过 system_prompt 可以定义 Agent 的身份、任务目标和行为规范

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

code_review_instructions = """你是一位资深 Python 工程师,专注于代码审查。

你的职责:
- 检查代码的可读性、性能、安全性
- 指出潜在 Bug 和改进建议
- 提供具体的修改示例

回复格式:
1. 总体评分(1-10)
2. 主要问题列表
3. 具体改进建议(附修改后的代码片段)
"""

agent = create_deep_agent(
    model=model,
    system_prompt=code_review_instructions,
)

code_to_review = """
def find_user(users, name):
    for i in range(len(users)):
        if users[i]['name'] == name:
            return users[i]
    return None
"""

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": f"请审查以下代码:\n```python\n{code_to_review}\n```",
            }
        ]
    }
)

print(result["messages"][-1].content)

流式输出:

Deep Agents 底层是 LangGraph Graph,因此可以使用流式输出观察模型生成过程

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

agent = create_deep_agent(
    model=model,
    system_prompt="你是一位技术写作专家,擅长写清晰易懂的技术文档。",
)

for msg, metadata in agent.stream(
    {
        "messages": [
            {"role": "user", "content": "写一篇关于 Python 装饰器的简短介绍。"}
        ]
    },
    stream_mode="messages",
):
    content = getattr(msg, "content", None)
    if isinstance(content, str):
        print(content, end="", flush=True)

print()

stream_mode="messages" 适合把模型 token 逐步打印出来;如果还需要观察状态变化、工具调用和中断事件,可以进一步使用 LangGraph 的事件流能力

多轮对话与 Checkpoint:

如果希望 Agent 在多轮调用之间保留对话状态,需要配置 checkpointer,并在调用时传入相同的 thread_id

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

checkpointer = InMemorySaver()

agent = create_deep_agent(
    model=model,
    system_prompt="你是一个乐于助人的编程助手。",
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "session-001"}}

result1 = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "我叫 ZHU,我在学习 Python。"}
        ]
    },
    config=config,
)

print("第一轮回复:", result1["messages"][-1].content)

result2 = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "我叫什么名字?我在学什么?"}
        ]
    },
    config=config,
)

print("第二轮回复:", result2["messages"][-1].content)

thread_id:同一个 thread_id 下的调用共享对话状态;不同的 thread_id 之间相互隔离,互不干扰,可以把它理解为 "对话房间号"

InMemorySaver 只适合本地测试,进程结束后状态会丢失;生产环境应使用 SQLite、Postgres 或 LangGraph Platform 提供的持久化 checkpointer

虚拟文件系统:

Deep Agents 默认带有虚拟文件系统能力;即使不配置 FilesystemBackend,Agent 也可以通过默认 StateBackend 在图状态里读写文件

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

checkpointer = InMemorySaver()

agent = create_deep_agent(
    model=model,
    system_prompt="你是一个资料整理助手。需要沉淀的信息请写入文件,再基于文件内容回答。",
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "fs-001"}}

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "请在 /workspace/notes.md 写入三条学习 Deep Agents 的要点,然后读出来总结。",
            }
        ]
    },
    config=config,
)

print(result["messages"][-1].content)

这里的 /workspace/notes.md 默认不会写到真实磁盘;在默认 StateBackend 下,文件保存在 LangGraph 图状态中,只要同一个 thread_id 和 checkpointer 还在,就能继续读写
虚拟文件系统的意义不只是保存文件,更重要的是做上下文卸载;当工具返回内容很长时,可以让 Agent 把中间资料写到文件中,只在消息上下文里保留文件路径和摘要,这样可以显著降低上下文膨胀

权限控制:

permissions 用来控制 Deep Agents 内置文件系统工具能访问哪些路径

import os

from deepagents import FilesystemPermission, create_deep_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

permissions = [
    FilesystemPermission(
        operations=["read", "write"],
        paths=["/secrets/**"],
        mode="deny",
    ),
    FilesystemPermission(
        operations=["write"],
        paths=["/workspace/releases/**"],
        mode="interrupt",
    ),
    FilesystemPermission(
        operations=["read", "write"],
        paths=["/workspace/**"],
        mode="allow",
    ),
    FilesystemPermission(
        operations=["read", "write"],
        paths=["/**"],
        mode="deny",
    ),
]

checkpointer = InMemorySaver()

agent = create_deep_agent(
    model=model,
    permissions=permissions,
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "perm-001"}}

result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "在 /workspace/hello.md 写入一行测试内容"}
        ]
    },
    config=config,
)

print(result["messages"][-1].content)

mode 有三种取值:allow:允许执行;deny:拒绝执行,并返回权限错误;interrupt:执行前暂停,等待人工审批
需要注意:permissions 作用在 Deep Agents 内置文件系统工具层面;如果自定义了一个工具,并在工具内部直接读写磁盘,permissions 不会自动拦截这个自定义工具

Human-in-the-loop:

对于敏感操作(例如删除文件、发送邮件),可以配置 Agent 在执行前暂停等待人工确认

import os

from deepagents import create_deep_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)


@tool
def delete_file(path: str) -> str:
    """删除指定路径的文件。"""
    return f"文件 {path} 已删除。"


@tool
def read_file_preview(path: str) -> str:
    """读取指定路径的文件摘要。"""
    return f"文件 {path} 的内容摘要:这里是示例内容。"


checkpointer = InMemorySaver()

agent = create_deep_agent(
    model=model,
    tools=[delete_file, read_file_preview],
    interrupt_on={
        "delete_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
        },
        "read_file_preview": False,
    },
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "hitl-001"}}

paused = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "请删除 /tmp/test.txt 文件。"}
        ]
    },
    config=config,
    version="v2",
)

print("待审批请求:")
print(paused.interrupts)

resumed = agent.invoke(
    Command(
        resume={
            "decisions": [
                {"type": "approve"}
            ]
        }
    ),
    config=config,
    version="v2",
)

print("最终回复:")
print(resumed.value["messages"][-1].content)

HITL 有四种常见决策:

决策作用适用场景
approve按原参数执行工具确认删除、确认发送
edit修改工具参数后执行修改邮件收件人、改文件路径
reject拒绝执行,并把反馈返回给 Agent禁止删除、禁止写库
respond不执行工具,直接把人工回复作为工具结果ask_user 这类人工输入工具

两个关键点:

  • HITL 必须配置 checkpointer,否则 Agent 无法可靠暂停和恢复;
  • 恢复执行时必须使用同一个 thread_id

Memory:

Deep Agents 的 memory 参数用于加载 AGENTS.md 文件
Memory 更适合保存长期有效的信息,例如:项目介绍、代码风格、常用命令、用户偏好、业务术语和不应反复询问的背景信息等
Memory 会在 Agent 启动时加载,并注入系统提示词;它不是按需加载,而是默认进入上下文

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

checkpointer = InMemorySaver()

agent = create_deep_agent(
    model=model,
    memory=["/memory/AGENTS.md"],
    checkpointer=checkpointer,
)

files = {
    "/memory/AGENTS.md": """# 项目记忆

用户偏好:
- 示例代码优先使用中文注释;
- 解释概念时先讲用途,再讲 API;

项目约定:
- DeepSeek 示例统一使用 deepseek-v4-flash;
- 环境变量统一使用 DEEPSEEK_API_KEY;
"""
}

result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "请用我的偏好解释 Deep Agents 的作用。"}
        ],
        "files": files,
    },
    config={"configurable": {"thread_id": "memory-001"}},
)

print(result["messages"][-1].content)

注意:不要把 API Key、访问令牌、密码等敏感信息写入 Memory,且 Memory是长期上下文,不适合保存一次性临时信息

Skills:

Skills 与 Memory 不同,Memory 是始终加载,而 Skills 是按需加载

这意味着可以把比较长的领域知识、操作流程、写作规范、工具说明放进 Skill 文件里,Agent 只有在需要时才读取完整内容

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

agent = create_deep_agent(
    model=model,
    skills=["/skills"],
    system_prompt="你是一个技术资料整理助手。需要专业写作规范时,应优先使用可用 Skills。",
)

files = {
    "/skills/report-writer/SKILL.md": """---
name: report-writer
description: 当需要把资料整理成结构化研究报告时使用
---

# 报告写作规范

写作顺序:
1. 先给结论;
2. 再列依据;
3. 最后给风险和下一步建议;

格式要求:
- 使用 Markdown;
- 标题清晰;
- 不要把未经验证的信息写成确定事实;
"""
}

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "请用报告格式解释 Deep Agents 适合解决什么问题。",
            }
        ],
        "files": files,
    }
)

print(result["messages"][-1].content)

Skills 适合放:

  • 复杂操作流程
  • 代码库约定
  • 写作模板
  • 行业术语
  • 工具使用指南
  • 长篇参考资料

如果某段内容每次都必须被模型看到,用 Memory;如果某段内容只有特定任务才需要,用 Skills

Sub Agents:

Sub Agents 用来把复杂任务拆给不同角色处理,主 Agent 负责规划、协调和最终回答,子 Agent 负责处理某一类专门任务,并拥有独立上下文窗口

这可以减少主 Agent 上下文膨胀,也能降低多任务之间的干扰

一个子 Agent 至少需要:

  • name:唯一名称
  • description:主 Agent 用它判断什么时候委派任务
  • system_prompt:子 Agent 的角色和行为规则
import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)


def summarize_sources(text: str) -> str:
    """把资料压缩成要点摘要。"""
    lines = [line.strip() for line in text.splitlines() if line.strip()]
    return "\n".join(f"- {line}" for line in lines[:8])


summary_subagent = {
    "name": "source-summarizer",
    "description": "当需要压缩长资料、提取要点时使用。",
    "system_prompt": "你是资料摘要专家,只保留核心事实、关键数字和来源线索。",
    "tools": [summarize_sources],
}

agent = create_deep_agent(
    model=model,
    subagents=[summary_subagent],
    system_prompt="你是研究任务协调者。资料过长时,请委派 source-summarizer 子 Agent 先做压缩。",
)

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "请帮我整理这段资料的重点:Deep Agents 提供规划、文件系统、子 Agent、上下文压缩、Memory 和 HITL。",
            }
        ]
    }
)

print(result["messages"][-1].content)

Sub Agent 可以覆盖主 Agent 的部分配置,例如:tools、model、middleware、interrupt_on、skills、permissions、response_format 等
如果没有手动提供名为 general-purpose 的子 Agent,Deep Agents 默认会添加一个通用子 Agent,主 Agent 可以通过 task 工具把子任务委派给它

Context Engineering:

Deep Agents 的一大核心价值,是把上下文工程变成默认能力

在复杂任务里,真正困难的不是 "调用一次模型",而是如何决定哪些信息应该放进上下文,哪些信息应该写入文件,哪些信息应该交给子 Agent,哪些历史应该被总结压缩

一个常见策略如下:用户提出复杂任务,主 Agent 使用 write_todos 拆分步骤,搜索工具或业务工具返回大量资料,Agent 将长资料写入虚拟文件系统,主上下文只保留文件路径、摘要和下一步计划,子 Agent 读取相关文件并处理局部任务,汇总结果写入最终报告,长历史通过 summarization 压缩

这比把所有搜索结果、工具结果和中间草稿都塞进 messages 更稳定

什么时候应该写文件:

  • 工具返回内容很长
  • 后续步骤还要反复引用
  • 中间结果需要被多个子任务共享
  • 需要保存草稿、计划、表格或报告

什么时候应该委派子 Agent:

  • 子任务有明确边界
  • 子任务需要大量上下文
  • 子任务可以独立完成
  • 主 Agent 只需要最终摘要或结构化结果

什么时候应该用 Memory:

  • 信息跨会话仍然有效
  • 用户偏好会反复影响回答
  • 项目规则需要长期遵守

什么时候应该用 Skills:

  • 知识很多,但不是每次都需要
  • 某类任务有固定流程
  • 需要把领域规则从主 prompt 中拆出去

结构化输出:

当希望 Agent 返回稳定 JSON、报告对象或业务字段时,可以使用 response_format 

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


class LearningPlan(BaseModel):
    """学习计划结构。"""

    title: str = Field(description="计划标题")
    goals: list[str] = Field(description="学习目标")
    steps: list[str] = Field(description="执行步骤")
    risks: list[str] = Field(description="可能遇到的问题")


model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

agent = create_deep_agent(
    model=model,
    system_prompt="你是一位学习路径规划助手。",
    response_format=LearningPlan,
)

result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "帮我制定一个 3 天入门 Deep Agents 的学习计划。"}
        ]
    }
)

plan = result["structured_response"]

print(plan.title)
print(plan.goals)
print(plan.steps)
print(plan.risks)

结构化输出适合:研究报告,表单生成,数据抽取,任务计划和 API 返回等

如果模型或 provider 对结构化输出支持不稳定,可以降低 schema 复杂度,或者让 Agent 先输出 Markdown,再由后处理程序解析

综合案例:

本节展示一个完整的研究报告 Agent

场景说明:构建一个 Agent,接收用户的研究主题,自动分解任务,调用 Tavily 搜索资料,委派子 Agent 整理内容,最终输出一份结构化 Markdown 报告

import json
import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from pydantic import BaseModel, Field
from tavily import TavilyClient


main_model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
    temperature=0.3,
    extra_body={"thinking": {"type": "disabled"}},
)

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


def web_search(query: str, max_results: int = 5) -> str:
    """使用 Tavily 搜索互联网信息。"""
    try:
        response = tavily_client.search(
            query=query,
            max_results=max_results,
            search_depth="advanced",
            include_answer=True,
            include_raw_content=False,
        )

        lines = []

        if response.get("answer"):
            lines.append(f"总结:{response['answer']}\n")

        for idx, item in enumerate(response.get("results", []), start=1):
            lines.append(
                f"""
[{idx}] {item.get("title", "")}
URL: {item.get("url", "")}
内容: {item.get("content", "")}
""".strip()
            )

        return "\n\n".join(lines)
    except Exception as exc:
        return f"搜索出错:{exc}"


class ResearchReport(BaseModel):
    """研究报告结构化输出模型。"""

    title: str = Field(description="报告标题")
    markdown: str = Field(description="完整 Markdown 报告正文,必须以一级标题开头")


def format_report(title: str, sections: str) -> str:
    """将研究内容整理为 Markdown 格式的报告。

    Args:
        title: 报告标题
        sections: 各章节内容,以 JSON 字符串表示,格式:{"章节名": "内容"}
    """
    try:
        data = json.loads(sections)
        lines = [f"# {title}\n"]
        for section_name, content in data.items():
            lines.append(f"## {section_name}\n")
            lines.append(f"{content}\n")
        return "\n".join(lines)
    except Exception as exc:
        return f"格式化出错:{exc}"


report_writer_subagent = {
    "name": "report-writer",
    "description": "专业的报告撰写助手,负责将研究资料整合为结构清晰的 Markdown 报告。",
    "system_prompt": """你是一位专业的报告撰写专家。

接收研究资料后,必须直接输出完整 Markdown 报告正文。
报告应包含:背景、核心概念、主要能力、应用场景、风险限制、学习建议。
不要把未经来源支持的信息写成确定事实。
""",
    "tools": [format_report],
}

checkpointer = InMemorySaver()

research_instructions = """你是一位专业的研究助手,擅长深度调研和报告撰写。

当用户提供研究主题时,你应当:
1. 使用 web_search 工具搜索相关信息,至少搜索 2-3 次;
2. 每次搜索使用不同关键词,覆盖概念、官方资料、应用场景和限制;
3. 将搜索结果写入虚拟文件,避免把长资料全部堆在上下文里;
4. 委派 report-writer 子 Agent 撰写最终报告;
5. 最终输出必须符合 ResearchReport 结构。
"""

agent = create_deep_agent(
    model=main_model,
    tools=[web_search],
    system_prompt=research_instructions,
    subagents=[report_writer_subagent],
    checkpointer=checkpointer,
    response_format=ResearchReport,
)

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "请帮我研究 DeepAgents 是什么,以及它的主要应用场景。",
            }
        ]
    },
    config={"configurable": {"thread_id": "research-001"}},
)

report = result["structured_response"]

print("=" * 60)
print(report.title)
print("=" * 60)
print(report.markdown)

这个案例里有几个关键设计:

主 Agent 负责任务规划、搜索、协调和最终结构化输出;

web_search 是外部资料入口;

长资料应写入虚拟文件系统,而不是全部留在 messages;

report-writer 子 Agent 专注报告写作;

response_format=ResearchReport 保证最终结果有稳定字段;

checkpointer 保证长任务可以持续运行并保留状态

调试与观测:

Agent 出问题时,不要只看最终回答,还应该观察:

  • 模型有没有调用工具
  • 调用了哪个工具
  • 工具参数是否正确
  • 工具返回是否过长
  • 是否被权限规则拦截
  • 是否进入了 HITL 中断
  • 是否因为模型不支持 Tool Calling 而退化成普通回答

本地调试可以开启 debug=True:

agent = create_deep_agent(
    model=model,
    tools=[get_weather],
    debug=True,
)

生产或复杂项目建议接入 LangSmith

macOS / Linux:

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="lsv2-your-langsmith-key"
export LANGSMITH_PROJECT="deepagents-demo"

Windows PowerShell:

$env:LANGSMITH_TRACING="true"
$env:LANGSMITH_API_KEY="lsv2-your-langsmith-key"
$env:LANGSMITH_PROJECT="deepagents-demo"

LangSmith 可以帮助查看完整调用链,包括模型请求、工具调用、状态变化、错误堆栈和耗时;

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐