Deep Agents 实现

核心概念

Deep Agents 提供了一个基于 LangGraph 构建的开箱即用智能体框架：

• - createdeepagent：创建已配置智能体的工厂函数
• 中间件：注入的能力（文件系统、待办事项、子智能体、摘要）
• 后端：可插拔的文件存储（状态、文件系统、存储、复合）
• 子智能体：通过 task 工具执行的隔离任务

返回的智能体是一个编译后的 LangGraph StateGraph，兼容流式传输、检查点和 LangGraph Studio。

必要导入

python

核心

from deepagents import createdeepagent

子智能体

from deepagents import CompiledSubAgent

后端

from deepagents.backends import ( StateBackend, # 临时（默认） FilesystemBackend, # 真实磁盘 StoreBackend, # 持久化跨线程 CompositeBackend, # 路由路径到后端 )

LangGraph（用于检查点、存储、流式传输）

from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.postgres import PostgresSaver from langgraph.store.memory import InMemoryStore

LangChain（用于自定义模型、工具）

from langchain.chatmodels import initchat_model from langchain_core.tools import tool

基本用法

最小化智能体

python
from deepagents import createdeepagent

默认使用 Claude Sonnet 4

agent = createdeepagent()

result = agent.invoke({messages: [{role: user, content: 你好！}]})

使用自定义工具

python
from langchain_core.tools import tool
from deepagents import createdeepagent

@tool
def web_search(query: str) -> str:
搜索网络获取信息。
return tavily_client.search(query)

agent = createdeepagent(
tools=[web_search],
system_prompt=你是一个研究助手。搜索网络来回答问题。,
)

result = agent.invoke({messages: [{role: user, content: 什么是 LangGraph？}]})

使用自定义模型

python
from langchain.chatmodels import initchat_model
from deepagents import createdeepagent

OpenAI

model = initchatmodel(openai:gpt-4o)

或使用自定义设置的 Anthropic

from langchain_anthropic import ChatAnthropic model = ChatAnthropic(modelname=claude-sonnet-4-5-20250929, maxtokens=8192)

agent = createdeepagent(model=model)

使用检查点（持久化）

python
from langgraph.checkpoint.memory import InMemorySaver
from deepagents import createdeepagent

agent = createdeepagent(checkpointer=InMemorySaver())

使用检查点时必须提供 thread_id

config = {configurable: {thread_id: user-123}} result = agent.invoke({messages: [...]}, config)

继续对话

result = agent.invoke({messages: [{role: user, content: 跟进}]}, config)

流式传输

智能体支持所有 LangGraph 流模式。

流式更新

python
for chunk in agent.stream(
{messages: [{role: user, content: 写一份报告}]},
stream_mode=updates
):
print(chunk) # {node_name: {key: value}}

流式消息（逐 Token）

python
for chunk in agent.stream(
{messages: [{role: user, content: 解释量子计算}]},
stream_mode=messages
):
# 实时 Token 流式传输
print(chunk.content, end=, flush=True)

异步流式传输

python
async for chunk in agent.astream(
{messages: [...]},
stream_mode=updates
):
print(chunk)

多种流模式

python
for mode, chunk in agent.stream(
{messages: [...]},
stream_mode=[updates, messages]
):
if mode == messages:
print(Token:, chunk.content)
else:
print(更新:, chunk)

后端配置

StateBackend（默认 - 临时）

文件存储在智能体状态中，仅在线程内持久化。

python

隐式 - 这是默认值

agent = createdeepagent()

显式

from deepagents.backends import StateBackend agent = createdeepagent(backend=lambda rt: StateBackend(rt))

FilesystemBackend（真实磁盘）

在磁盘上读写实际文件。启用用于 shell 命令的 execute 工具。

python
from deepagents.backends import FilesystemBackend

agent = createdeepagent(
backend=FilesystemBackend(root_dir=/path/to/project),
)

StoreBackend（持久化跨线程）

使用 LangGraph Store 实现跨对话持久化。

python
from langgraph.store.memory import InMemoryStore
from deepagents.backends import StoreBackend

store = InMemoryStore()

agent = createdeepagent(
backend=lambda rt: StoreBackend(rt),
store=store, # StoreBackend 必需
)

CompositeBackend（混合路由）

将不同路径路由到不同后端。

python
from langgraph.store.memory import InMemoryStore
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend

store = InMemoryStore()

agent = createdeepagent(
backend=CompositeBackend(
default=StateBackend(), # /workspace/* → 临时
routes={
/memories/: StoreBackend(store=store), # 持久化
/preferences/: StoreBackend(store=store), # 持久化
},
),
store=store,
)

/memories/ 下的文件在所有对话中持久化

/workspace/ 下的文件在每个线程中临时存在

子智能体

使用默认通用智能体

默认情况下，一个 general-purpose 子智能体可用，包含所有主智能体工具。

python
agent = createdeepagent(tools=[web_search])

智能体现在可以通过 task 工具委托：

task(subagent_type=general-purpose, prompt=深入研究主题 X)

定义自定义子智能体

python
from deepagents import createdeepagent

research_agent = {
name: researcher,
description: 使用网络搜索对复杂主题进行深入研究,
system_prompt: 你是一位专家研究员。
彻底搜索，交叉引用来源，并综合发现。,
tools: [websearch, documentreader],
}

code_agent = {
name: coder,
description: 编写、审查和调试代码,
system_prompt: 你是一位专家程序员。编写干净、经过测试的代码。,
tools: [code_executor, linter],
model: openai:gpt-4o, # 可选：每个子智能体使用不同模型
}

agent = createdeepagent(
subagents=[researchagent, codeagent],
system_prompt=将研究委托给研究员，将编码委托给编码员。,
)

预编译的 LangGraph 子智能体

使用现有的 LangGraph 图作为子智能体。

python
from deepagents import CompiledSubAgent, createdeepagent
from langgraph.prebuilt import createreactagent

现有图

customgraph = createreact_agent( model=anthropic:claude-sonnet-4-5-20250929, tools=[specialized_tool], prompt=自定义工作流指令, )

agent = createdeepagent(
subagents=[CompiledSubAgent(
name=custom-workflow,
description=运行我的专业分析工作流,
runnable=custom_graph,
)]
)

带自定义中间件的子智能体

python
from langchain.agents.middleware import AgentMiddleware

class LoggingMiddleware(AgentMiddleware):
def transform_response(self, response):
print(f子智能体响应：{response})
return response

agent_spec = {
name: logged-agent,
description: 带额外日志记录的智能体,
system_prompt: 你很有帮助。,
tools: [],
middleware: [LoggingMiddleware()], # 在默认中间件之后添加
}

人在回路中

基本中断配置

在特定工具之前暂停执行，等待人工批准。

python
from deepagents import createdeepagent

agent = createdeepagent(
tools=[sendemail, deletefile