Deep Agents
基础介绍:
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、execute | FilesystemMiddleware |
| 子 Agent | 通过 task 工具把子任务委派给独立上下文的专用 Agent | SubAgentMiddleware |
| 上下文压缩 | 自动总结历史消息,避免长任务超出上下文窗口 | 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 |
| DeepSeek | ChatOpenAI(..., base_url="https://api.deepseek.com") | DeepSeek 兼容 OpenAI 接口,建议传模型实例 |
| OpenRouter | 使用对应 LangChain provider 或 OpenAI 兼容模型实例 | 注意模型是否支持工具调用 |
| 本地模型 | 通过 OpenAI 兼容服务或对应 LangChain 集成接入 | 必须确认 Tool Calling 能正常工作 |
参数说明:
create_deep_agent 是创建 Deep Agent 的核心函数,常用参数如下:
| 参数 | 类型 | 是否必填 | 说明 | 建议 |
|---|---|---|---|---|
model | str 或模型实例 | 建议必填 | 模型字符串,或初始化好的 Chat Model | 生产中显式传入 |
tools | 列表 | 否 | 自定义工具函数、LangChain Tool 或 tool schema | 只放业务工具 |
system_prompt | str / SystemMessage / 配置字典 | 否 | 自定义系统提示词 | 用来定义角色和行为 |
middleware | 列表 | 否 | 追加自定义中间件 | 高级场景使用 |
subagents | 列表 | 否 | 同步、异步或已编译子 Agent 定义 | 复杂任务拆分时使用 |
skills | 路径列表 | 否 | Skills 目录路径 | 领域知识按需加载 |
memory | 路径列表 | 否 | AGENTS.md 文件路径 | 项目规则和长期偏好 |
permissions | 权限规则列表 | 否 | 文件系统工具的路径权限规则 | 生产或敏感路径必须配置 |
backend | Backend 实例或工厂 | 否 | 控制文件存储和执行环境 | 默认 StateBackend() |
interrupt_on | 字典 | 否 | 指定哪些工具调用需要人工审批 | 需配合 checkpointer |
response_format | Pydantic / TypedDict / Schema | 否 | 结构化输出格式 | 报告、表单、JSON 输出时使用 |
state_schema | DeepAgentState 子类 | 否 | 扩展 Agent 状态结构 | 高级状态管理使用 |
context_schema | 类型定义 | 否 | 定义运行时上下文 | 多租户或请求级上下文使用 |
checkpointer | Checkpointer | 否 | 保存对话和中断状态 | 多轮对话、HITL 必须配置 |
store | BaseStore | 否 | 长期存储后端 | 持久化文件或跨线程共享 |
debug | bool | 否 | 输出调试信息 | 本地排错时开启 |
name | str | 否 | Agent 名称 | 多 Agent 观测时有用 |
cache | BaseCache | 否 | 缓存配置 | 高级性能优化 |
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 | 适用场景 |
|---|---|---|---|---|
StateBackend | LangGraph 图状态内 | 否,同一 thread 内有效 | 否 | 默认后端,本地学习和简单 Demo |
FilesystemBackend | 本机磁盘 | 是 | 否 | 可信本地开发、需要真实读写文件 |
StoreBackend | LangGraph 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 可以帮助查看完整调用链,包括模型请求、工具调用、状态变化、错误堆栈和耗时;
更多推荐

所有评论(0)