主题
自定义 Deep Agents
了解如何使用系统提示词、工具、子智能体等来自定义 Deep Agents
围绕你的目标构建 harness。create_deep_agent 为你提供了生产就绪的基础:将其连接到你的数据、塑造其行为,并添加你的用例所需的能力。
python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
system_prompt="You are a helpful assistant.",
tools=[search, fetch_url],
memory=["./AGENTS.md"],
skills=["./skills/"],
)python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="openai:gpt-5.5",
system_prompt="You are a helpful assistant.",
tools=[search, fetch_url],
memory=["./AGENTS.md"],
skills=["./skills/"],
)python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
system_prompt="You are a helpful assistant.",
tools=[search, fetch_url],
memory=["./AGENTS.md"],
skills=["./skills/"],
)python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
system_prompt="You are a helpful assistant.",
tools=[search, fetch_url],
memory=["./AGENTS.md"],
skills=["./skills/"],
)python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
system_prompt="You are a helpful assistant.",
tools=[search, fetch_url],
memory=["./AGENTS.md"],
skills=["./skills/"],
)python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
system_prompt="You are a helpful assistant.",
tools=[search, fetch_url],
memory=["./AGENTS.md"],
skills=["./skills/"],
)python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
system_prompt="You are a helpful assistant.",
tools=[search, fetch_url],
memory=["./AGENTS.md"],
skills=["./skills/"],
)| 参数 | 作用 |
|---|---|
model= | 使用哪个模型 |
system_prompt= | 智能体的自定义指令 |
tools= | 智能体可调用的领域工具 |
memory= | 启动时加载的 AGENTS.md 文件 |
skills= | 用于按需知识的技能目录 |
backend= | 文件系统后端(默认为 StateBackend) |
permissions= | 文件系统的路径级访问控制 |
subagents= | 用于委派任务的自定义子智能体 |
middleware= | 合并到 Deep Agents 栈中的额外 middleware;一个 .name 与内置条目匹配的实例会就地替换该条目,其他任何内容则落在最后一个核心 middleware 条目之后,并在 profile、prompt-caching 和 memory 之前 |
interrupt_on= | 在工具调用前暂停以待人工批准 |
response_format= | 结构化输出 schema |
state_schema= | 自定义图状态 schema |
context_schema= | 每次运行的运行时上下文 schema(用户 ID、API 密钥、功能标志) |
| profiles | 作为可复用捆绑包的按模型默认值 |
完整函数签名
python
create_deep_agent(
model: str | BaseChatModel | None = None,
tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
*,
system_prompt: str | SystemMessage | None = None,
middleware: Sequence[AgentMiddleware[StateT_co, ContextT]] = (),
subagents: Sequence[SubAgent | CompiledSubAgent | AsyncSubAgent] | None = None,
skills: list[str] | None = None,
memory: list[str] | None = None,
permissions: list[FilesystemPermission] | None = None,
backend: BackendProtocol | None = None,
interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,
response_format: ResponseFormat[ResponseT] | type[ResponseT] | dict[str, Any] | None = None,
state_schema: type[DeepAgentState] | None = None,
context_schema: type[ContextT] | None = None,
checkpointer: Checkpointer | None = None,
store: BaseStore | None = None,
debug: bool = False,
name: str | None = None,
cache: BaseCache | None = None
) -> CompiledStateGraph[AgentState[ResponseT], ContextT, InputAgentState, OutputAgentState[ResponseT]]有关完整参数列表,请参阅 create_deep_agent API 参考。要从零开始组合完全自定义的 harness,请参阅配置 harness,或按照分步指南从零构建 deep agent 操作。
TIP
当你添加工具、子智能体和后端时,使用 LangSmith 来追踪各部分如何协同运作。按照可观测性快速入门进行设置,并参阅投入生产了解在 LangSmith 上部署。
我们还建议你设置 LangSmith Engine,它会监控你的追踪、检测问题并提出修复建议。
Model
以 provider:model 格式传入 model 字符串,或传入已初始化的模型实例。有关所有服务商,请参阅支持的模型;有关经过测试的推荐,请参阅建议的模型。
TIP
使用 provider:model 格式(例如 openai:gpt-5.5)在模型之间快速切换。
👉 阅读 OpenAI 聊天模型集成文档
bash
pip install -U "langchain[openai]"bash
uv add "langchain[openai]"python
import os
from deepagents import create_deep_agent
os.environ["OPENAI_API_KEY"] = "sk-..."
agent = create_deep_agent(model="openai:gpt-5.5")
# this calls init_chat_model for the specified model with default parameters
# to use specific model parameters, use init_chat_model directlypython
import os
from langchain.chat_models import init_chat_model
from deepagents import create_deep_agent
os.environ["OPENAI_API_KEY"] = "sk-..."
model = init_chat_model(model="openai:gpt-5.5")
agent = create_deep_agent(model=model)python
import os
from langchain_openai import ChatOpenAI
from deepagents import create_deep_agent
os.environ["OPENAI_API_KEY"] = "sk-..."
model = ChatOpenAI(model="gpt-5.5")
agent = create_deep_agent(model=model)TIP
聊天模型会自动重试瞬时的 API 故障(采用指数退避)。有关调整 max_retries / timeout 的默认值、限制和代码示例,请参阅 LangChain 的模型页面。
工具
除了用于文件管理和子智能体生成的内置工具之外,你还可以提供自定义工具:
python
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=[internet_search],
)python
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
agent = create_deep_agent(
model="openai:gpt-5.5",
tools=[internet_search],
)python
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
tools=[internet_search],
)python
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
tools=[internet_search],
)python
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
tools=[internet_search],
)python
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
tools=[internet_search],
)python
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
tools=[internet_search],
)MCP 工具
TIP
Deep Agents 完全支持 Model Context Protocol (MCP) 工具。你可以从任何 MCP 服务器——数据库、API、文件系统等——加载工具,并将它们直接传给 create_deep_agent。
安装带 mcp extra 的 LangChain 以连接到 MCP 服务器:
bash
pip install "langchain[mcp]"python
import asyncio
from deepagents import create_deep_agent
from langchain.mcp import MCPAdapter
async def main():
config = {"mcpServers": {"my_server": {"url": "http://localhost:8000/mcp"}}}
async with MCPAdapter(config) as adapter:
tools = await adapter.list_tools()
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=tools,
)
await agent.ainvoke(
{
"messages": [
{"role": "user", "content": "Use the MCP server to help me."}
]
},
config={"configurable": {"thread_id": "1"}},
)python
import asyncio
from deepagents import create_deep_agent
from langchain.mcp import MCPAdapter
async def main():
config = {"mcpServers": {"my_server": {"url": "http://localhost:8000/mcp"}}}
async with MCPAdapter(config) as adapter:
tools = await adapter.list_tools()
agent = create_deep_agent(
model="openai:gpt-5.5",
tools=tools,
)
await agent.ainvoke(
{
"messages": [
{"role": "user", "content": "Use the MCP server to help me."}
]
},
config={"configurable": {"thread_id": "1"}},
)python
import asyncio
from deepagents import create_deep_agent
from langchain.mcp import MCPAdapter
async def main():
config = {"mcpServers": {"my_server": {"url": "http://localhost:8000/mcp"}}}
async with MCPAdapter(config) as adapter:
tools = await adapter.list_tools()
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
tools=tools,
)
await agent.ainvoke(
{
"messages": [
{"role": "user", "content": "Use the MCP server to help me."}
]
},
config={"configurable": {"thread_id": "1"}},
)python
import asyncio
from deepagents import create_deep_agent
from langchain.mcp import MCPAdapter
async def main():
config = {"mcpServers": {"my_server": {"url": "http://localhost:8000/mcp"}}}
async with MCPAdapter(config) as adapter:
tools = await adapter.list_tools()
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
tools=tools,
)
await agent.ainvoke(
{
"messages": [
{"role": "user", "content": "Use the MCP server to help me."}
]
},
config={"configurable": {"thread_id": "1"}},
)python
import asyncio
from deepagents import create_deep_agent
from langchain.mcp import MCPAdapter
async def main():
config = {"mcpServers": {"my_server": {"url": "http://localhost:8000/mcp"}}}
async with MCPAdapter(config) as adapter:
tools = await adapter.list_tools()
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
tools=tools,
)
await agent.ainvoke(
{
"messages": [
{"role": "user", "content": "Use the MCP server to help me."}
]
},
config={"configurable": {"thread_id": "1"}},
)python
import asyncio
from deepagents import create_deep_agent
from langchain.mcp import MCPAdapter
async def main():
config = {"mcpServers": {"my_server": {"url": "http://localhost:8000/mcp"}}}
async with MCPAdapter(config) as adapter:
tools = await adapter.list_tools()
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
tools=tools,
)
await agent.ainvoke(
{
"messages": [
{"role": "user", "content": "Use the MCP server to help me."}
]
},
config={"configurable": {"thread_id": "1"}},
)python
import asyncio
from deepagents import create_deep_agent
from langchain.mcp import MCPAdapter
async def main():
config = {"mcpServers": {"my_server": {"url": "http://localhost:8000/mcp"}}}
async with MCPAdapter(config) as adapter:
tools = await adapter.list_tools()
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
tools=tools,
)
await agent.ainvoke(
{
"messages": [
{"role": "user", "content": "Use the MCP server to help me."}
]
},
config={"configurable": {"thread_id": "1"}},
)有关包括 stdio 服务器、OAuth 身份认证、工具过滤和有状态会话在内的详细配置选项,请参阅完整的 MCP 指南。
系统提示词
传入 system_prompt= 来为智能体提供你自己的指令:
python
from deepagents import create_deep_agent
research_instructions = """\
You are an expert researcher. Your job is to conduct \
thorough research, and then write a polished report. \
"""
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
system_prompt=research_instructions,
)python
from deepagents import create_deep_agent
research_instructions = """\
You are an expert researcher. Your job is to conduct \
thorough research, and then write a polished report. \
"""
agent = create_deep_agent(
model="openai:gpt-5.5",
system_prompt=research_instructions,
)python
from deepagents import create_deep_agent
research_instructions = """\
You are an expert researcher. Your job is to conduct \
thorough research, and then write a polished report. \
"""
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
system_prompt=research_instructions,
)python
from deepagents import create_deep_agent
research_instructions = """\
You are an expert researcher. Your job is to conduct \
thorough research, and then write a polished report. \
"""
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
system_prompt=research_instructions,
)python
from deepagents import create_deep_agent
research_instructions = """\
You are an expert researcher. Your job is to conduct \
thorough research, and then write a polished report. \
"""
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
system_prompt=research_instructions,
)python
from deepagents import create_deep_agent
research_instructions = """\
You are an expert researcher. Your job is to conduct \
thorough research, and then write a polished report. \
"""
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
system_prompt=research_instructions,
)python
from deepagents import create_deep_agent
research_instructions = """\
You are an expert researcher. Your job is to conduct \
thorough research, and then write a polished report. \
"""
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
system_prompt=research_instructions,
)NOTE
除了字符串之外,主智能体还接受带有结构化内容块的 SystemMessage;Deep Agents 会保留这些块(子智能体的字典规范仍为字符串)。
子智能体提示词
声明式子智能体会针对其自身的模型解析 profile 覆盖,然后将解析后的 profile 的 base_system_prompt / system_prompt_suffix 应用到子智能体编写的 system_prompt。仅附带 system_prompt_suffix 的 profile(内置 Anthropic / OpenAI profile 的常见情况)会追加到编写的提示词之后。设置 base_system_prompt 的 profile 则会直接替换它。
通用子智能体提示词
自动添加的通用子智能体按以下顺序解析其基础提示词:general_purpose_subagent.system_prompt(若设置)-> HarnessProfile.base_system_prompt(若设置)-> SDK 通用默认值,并在其上叠加 profile 后缀。当两个覆盖字段都设置时,通用子智能体专用的那个优先,因此同时调整两个字段的调用方绝不会看到其 GP 覆盖被静默丢弃:
python
from deepagents import (
GeneralPurposeSubagentProfile,
HarnessProfile,
register_harness_profile,
)
register_harness_profile(
"anthropic",
HarnessProfile(
base_system_prompt="You are ACME's support orchestrator.", # main agent
general_purpose_subagent=GeneralPurposeSubagentProfile(
system_prompt="You are a research subagent. Cite sources.", # GP subagent
),
system_prompt_suffix="Always think step by step.",
),
)| 栈 | 最终系统提示词 |
|---|---|
| 主智能体 | "You are ACME's support orchestrator." + SUFFIX |
| GP 子智能体 | "You are a research subagent. Cite sources." + SUFFIX |
中间件
Deep Agents 支持任何 middleware,包括下面列出的内置 middleware、LangChain 的预构建 middleware、服务商特定的 middleware,以及你自己编写的自定义 middleware。
将 middleware 传给 create_deep_agent 的 middleware 参数。每个实例都会通过将其 .name 与栈中已有的内置条目进行匹配,从而合并到 Deep Agents 栈中:匹配的实例会就地替换该实例,任何不匹配的实例则会被插入到 PatchToolCallsMiddleware 之后。请参阅覆盖默认 middleware 实例。
Deep Agents 栈
create_deep_agent 会按固定的顺序构建 middleware。裸栈是仅使用模型时得到的结果。完整栈是完整的组装顺序,包括仅当你传入可选参数或解析后的 harness profile 贡献它们时才会出现的槽位。
裸栈
仅有一个 model(没有其他可选参数)时,主智能体通常包括:
FilesystemMiddlewareSubAgentMiddleware(因为通用子智能体会被自动添加,除非 harness profile 将其禁用)SummarizationMiddlewarePatchToolCallsMiddleware- 提示词缓存 middleware(始终注册;每个条目在不支持的模型上不执行任何操作)
- Harness profile 附加项和排除工具过滤,前提是解析后的模型 profile 定义了它们
完整栈
从第一个到最后一个:
SkillsMiddleware:仅当你传入skills时。它被注入到文件系统 middleware 之前,以便在文件工具运行之前技能元数据就可用。FilesystemMiddleware:处理文件系统操作,例如读取、写入和浏览目录。当你传入permissions时,文件系统权限强制执行会包含在此处,以便它能够评估智能体可能调用的每一个工具。SubAgentMiddleware:仅当至少有一个同步子智能体可用时。它生成并协调子智能体以委派任务。由于通用子智能体默认会被自动添加,因此它包含在裸栈中;通过禁用该子智能体且不传入任何同步subagents即可省略它。请参阅不使用子智能体运行。SummarizationMiddleware:当对话变长时,压缩消息历史以保持在上下文限制之内(通过 create_summarization_middleware)。PatchToolCallsMiddleware:当运行在中断后恢复或收到格式错误的工具调用参数时,修复消息历史中悬空的工具调用。它运行在 Anthropic 提示词缓存和下方尾部栈之前。AsyncSubAgentMiddleware:仅当你配置了异步子智能体时。你的 middleware 参数:你作为
middleware参数传入的可选 middleware 会在 Patch 之后、但栈的其余部分之前被合并。.name与上述某个内置条目匹配的实例会就地替换该实例,而不是重复;其他任何内容则落在此处。请参阅覆盖默认 middleware 实例。Harness profile 附加项:来自解析后的模型 profile 的服务商特定 middleware(如果有)。
排除工具过滤:当 harness profile 列出被排除的工具时,middleware 会将这些工具从智能体中移除。
提示词缓存(
AnthropicPromptCachingMiddleware和BedrockPromptCachingMiddleware):两者都会始终注册,并运行在 Patch 之后以及你的 middleware 之后,以使缓存的 prefix 与实际发送给模型的内容匹配。每个都会在不支持的模型上不执行任何操作(unsupported_model_behavior="ignore"),因此 Anthropic middleware 适用于 Anthropic 模型,而 Bedrock middleware 适用于支持缓存的 AWS Bedrock 模型。MemoryMiddleware:仅当你传入memory时。
NOTE
`MemoryMiddleware` 被放置在 profile 附加项和提示词缓存 middleware **之后**,这样对注入记忆的更新就不太可能使缓存 prefix 失效。`create_deep_agent` 实现注释中也提到了同样的顺序问题。
HumanInTheLoopMiddleware:仅当你传入interrupt_on时。在已配置的工具调用处暂停,以等待人工批准或输入。
同步子智能体栈
内置的通用子智能体和每个声明式同步 SubAgent 图都使用一个由 create_deep_agent 在代码中构建的栈。它在整体形态上与主智能体一致(文件系统、摘要、Patch、profile 附加项、Anthropic 和 Bedrock 缓存、可选权限),但有两处不同:
- 技能在这些内部智能体上运行于
PatchToolCallsMiddleware之后(在主智能体上,当设置了skills时,技能运行于文件系统 middleware 之前)。 - 子智能体图内部没有
SubAgentMiddleware(只有父智能体暴露task工具)。
当声明式子智能体设置 interrupt_on 时,该值会被转发给该子智能体的 create_agent,后者会为已配置的工具调用接入人在回路处理。
预构建 middleware
LangChain 还提供额外的预构建 middleware,让你可以附加各种功能,例如重试、回退或 PII 检测。更多内容请参阅预构建 middleware。
deepagents 库还提供 create_summarization_tool_middleware,使智能体能够在恰当的时机触发摘要——例如在任务之间——而不是按固定的 token 间隔触发。有关更多细节,请参阅摘要。
服务商特定的 middleware
有关针对特定 LLM 服务商优化的服务商特定 middleware,请参阅 Middleware 集成。
自定义 middleware
你可以提供额外的 middleware 来扩展功能、添加工具或实现自定义钩子:
python
from langchain.agents.middleware import wrap_tool_call
from langchain.tools import tool
from deepagents import create_deep_agent
@tool
def get_weather(city: str) -> str:
"""Get the weather in a city."""
return f"The weather in {city} is sunny."
call_count = [0] # Use list to allow modification in nested function
@wrap_tool_call
def log_tool_calls(request, handler):
"""Intercept and log every tool call - demonstrates cross-cutting concern."""
call_count[0] += 1
tool_name = request.name if hasattr(request, "name") else str(request)
print(f"[Middleware] Tool call #{call_count[0]}: {tool_name}")
print(f"[Middleware] Arguments: {request.args if hasattr(request, 'args') else 'N/A'}")
# Execute the tool call
result = handler(request)
# Log the result
print(f"[Middleware] Tool call #{call_count[0]} completed")
return result
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=[get_weather],
middleware=[log_tool_calls],
)python
from langchain.agents.middleware import wrap_tool_call
from langchain.tools import tool
from deepagents import create_deep_agent
@tool
def get_weather(city: str) -> str:
"""Get the weather in a city."""
return f"The weather in {city} is sunny."
call_count = [0] # Use list to allow modification in nested function
@wrap_tool_call
def log_tool_calls(request, handler):
"""Intercept and log every tool call - demonstrates cross-cutting concern."""
call_count[0] += 1
tool_name = request.name if hasattr(request, "name") else str(request)
print(f"[Middleware] Tool call #{call_count[0]}: {tool_name}")
print(f"[Middleware] Arguments: {request.args if hasattr(request, 'args') else 'N/A'}")
# Execute the tool call
result = handler(request)
# Log the result
print(f"[Middleware] Tool call #{call_count[0]} completed")
return result
agent = create_deep_agent(
model="openai:gpt-5.5",
tools=[get_weather],
middleware=[log_tool_calls],
)python
from langchain.agents.middleware import wrap_tool_call
from langchain.tools import tool
from deepagents import create_deep_agent
@tool
def get_weather(city: str) -> str:
"""Get the weather in a city."""
return f"The weather in {city} is sunny."
call_count = [0] # Use list to allow modification in nested function
@wrap_tool_call
def log_tool_calls(request, handler):
"""Intercept and log every tool call - demonstrates cross-cutting concern."""
call_count[0] += 1
tool_name = request.name if hasattr(request, "name") else str(request)
print(f"[Middleware] Tool call #{call_count[0]}: {tool_name}")
print(f"[Middleware] Arguments: {request.args if hasattr(request, 'args') else 'N/A'}")
# Execute the tool call
result = handler(request)
# Log the result
print(f"[Middleware] Tool call #{call_count[0]} completed")
return result
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
tools=[get_weather],
middleware=[log_tool_calls],
)python
from langchain.agents.middleware import wrap_tool_call
from langchain.tools import tool
from deepagents import create_deep_agent
@tool
def get_weather(city: str) -> str:
"""Get the weather in a city."""
return f"The weather in {city} is sunny."
call_count = [0] # Use list to allow modification in nested function
@wrap_tool_call
def log_tool_calls(request, handler):
"""Intercept and log every tool call - demonstrates cross-cutting concern."""
call_count[0] += 1
tool_name = request.name if hasattr(request, "name") else str(request)
print(f"[Middleware] Tool call #{call_count[0]}: {tool_name}")
print(f"[Middleware] Arguments: {request.args if hasattr(request, 'args') else 'N/A'}")
# Execute the tool call
result = handler(request)
# Log the result
print(f"[Middleware] Tool call #{call_count[0]} completed")
return result
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
tools=[get_weather],
middleware=[log_tool_calls],
)python
from langchain.agents.middleware import wrap_tool_call
from langchain.tools import tool
from deepagents import create_deep_agent
@tool
def get_weather(city: str) -> str:
"""Get the weather in a city."""
return f"The weather in {city} is sunny."
call_count = [0] # Use list to allow modification in nested function
@wrap_tool_call
def log_tool_calls(request, handler):
"""Intercept and log every tool call - demonstrates cross-cutting concern."""
call_count[0] += 1
tool_name = request.name if hasattr(request, "name") else str(request)
print(f"[Middleware] Tool call #{call_count[0]}: {tool_name}")
print(f"[Middleware] Arguments: {request.args if hasattr(request, 'args') else 'N/A'}")
# Execute the tool call
result = handler(request)
# Log the result
print(f"[Middleware] Tool call #{call_count[0]} completed")
return result
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
tools=[get_weather],
middleware=[log_tool_calls],
)python
from langchain.agents.middleware import wrap_tool_call
from langchain.tools import tool
from deepagents import create_deep_agent
@tool
def get_weather(city: str) -> str:
"""Get the weather in a city."""
return f"The weather in {city} is sunny."
call_count = [0] # Use list to allow modification in nested function
@wrap_tool_call
def log_tool_calls(request, handler):
"""Intercept and log every tool call - demonstrates cross-cutting concern."""
call_count[0] += 1
tool_name = request.name if hasattr(request, "name") else str(request)
print(f"[Middleware] Tool call #{call_count[0]}: {tool_name}")
print(f"[Middleware] Arguments: {request.args if hasattr(request, 'args') else 'N/A'}")
# Execute the tool call
result = handler(request)
# Log the result
print(f"[Middleware] Tool call #{call_count[0]} completed")
return result
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
tools=[get_weather],
middleware=[log_tool_calls],
)python
from langchain.agents.middleware import wrap_tool_call
from langchain.tools import tool
from deepagents import create_deep_agent
@tool
def get_weather(city: str) -> str:
"""Get the weather in a city."""
return f"The weather in {city} is sunny."
call_count = [0] # Use list to allow modification in nested function
@wrap_tool_call
def log_tool_calls(request, handler):
"""Intercept and log every tool call - demonstrates cross-cutting concern."""
call_count[0] += 1
tool_name = request.name if hasattr(request, "name") else str(request)
print(f"[Middleware] Tool call #{call_count[0]}: {tool_name}")
print(f"[Middleware] Arguments: {request.args if hasattr(request, 'args') else 'N/A'}")
# Execute the tool call
result = handler(request)
# Log the result
print(f"[Middleware] Tool call #{call_count[0]} completed")
return result
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
tools=[get_weather],
middleware=[log_tool_calls],
)WARNING
初始化后不要修改属性
如果你需要在多次钩子调用之间追踪值(例如计数器或累积数据),请使用图状态。 图状态在设计上限定在线程范围内,因此在并发下更新是安全的。
这样做:
python
from langchain.agents.middleware import AgentMiddleware
class CustomMiddleware(AgentMiddleware):
def __init__(self):
pass
def before_agent(self, state, runtime):
return {"x": state.get("x", 0) + 1} # Update graph state instead不要这样做:
python
class CustomMiddlewareBad(AgentMiddleware):
def __init__(self):
self.x = 1
def before_agent(self, state, runtime):
self.x += 1 # Mutation causes race conditions就地修改,例如在 before_agent 中修改 self.x,或在钩子中更改其他共享值,可能导致难以察觉的缺陷和竞态条件,因为许多操作会并发运行(子智能体、并行工具,以及不同线程上的并行调用)。
有关使用自定义属性扩展状态的完整细节,请参阅自定义 middleware - 自定义状态 schema。
如果你必须在自定义 middleware 中使用修改,请考虑当子智能体、并行工具或并发的智能体调用同时运行时会发生什么。
覆盖默认 middleware 实例
NOTE
通过匹配 .name 覆盖默认 middleware 需要 deepagents>=0.7。
传入一个 .name 与 Deep Agents 栈中某个条目匹配的 middleware 实例(例如 SummarizationMiddleware),即可就地替换该内置实例,而不是追加一个重复项。你传入的任何 .name 不匹配内置条目的 middleware 都不会被替换,它会落在最后一个核心 middleware 条目之后,并在 profile、prompt-caching 和 memory 之前。完整的顺序请参阅完整栈。
python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.middleware import SummarizationMiddleware
backend = StateBackend()
model = "openai:gpt-5.5"
custom_summarization = SummarizationMiddleware(
model=model,
backend=backend,
summary_prompt="Your custom summary prompt here.",
)
agent = create_deep_agent(
model=model,
middleware=[custom_summarization], # replaces the default SummarizationMiddleware
)NOTE
覆盖会替换默认 middleware 实例,而不会与之合并。这意味着你的替代实例必须完整配置好它所需的任何设置。这对 FilesystemMiddleware 尤为重要:如果你覆盖它,必须将 backend(以及适用的 permissions)直接传给你的自定义实例,因为它不会继承传给 create_deep_agent() 的 backend= 和 permissions=。要限制可用的文件系统工具,请将 tools 允许列表传给你的自定义 FilesystemMiddleware 实例;相关示例请参阅虚拟文件系统访问中的“限制文件系统工具”。
由 Deep Agents 自动添加的通用子智能体会从主智能体继承其默认 middleware 的覆盖,但不会沿用主智能体专有的 middleware。
通过 subagents= 定义的声明式子智能体不会继承主智能体的 middleware 自定义。若要在此处应用覆盖,请直接在该子智能体自己的 middleware 字段中传入;该字段会与同步子智能体栈进行匹配,其方式与 middleware= 与主智能体相匹配的方式相同。
示例
调整摘要触发时机
使用自定义的 trigger 和 keep 阈值覆盖 SummarizationMiddleware,以比默认更早或更晚地压缩对话历史,并控制每次压缩后保留多少条最近消息。
python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.middleware import SummarizationMiddleware
backend = StateBackend()
model = "anthropic:claude-sonnet-4-6"
agent = create_deep_agent(
model=model,
middleware=[
SummarizationMiddleware(
model=model,
backend=backend,
trigger=("tokens", 100000), # summarize once the conversation exceeds 100k tokens
keep=("messages", 20), # keep the most recent 20 messages verbatim
),
],
)trigger 还接受 ("fraction", ...) 以表示模型上下文窗口的百分比,而阈值列表会用 OR 语义将它们组合起来。完整选项集请参阅 SummarizationMiddleware 参考。
更新提示词缓存 TTL
覆盖 AnthropicPromptCachingMiddleware 以将缓存生存期延长到默认 5m TTL 之上,这对各轮之间间隔较长的智能体很有用。关于默认情况下缓存如何应用,请参阅提示词缓存。
python
from deepagents import create_deep_agent
from langchain_anthropic.middleware import AnthropicPromptCachingMiddleware
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
middleware=[
AnthropicPromptCachingMiddleware(ttl="1h"), # replaces the default 5m TTL
],
)限制启用的文件系统工具
NOTE
FilesystemMiddleware 上的 tools 允许列表需要 deepagents>=0.7。
使用 tools 允许列表覆盖 FilesystemMiddleware,以仅向模型暴露文件系统工具的一个子集,而不是完整的默认集合。
python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.middleware import FilesystemMiddleware
backend = StateBackend()
# Read-only agent: write_file, edit_file, delete, and execute are never shown
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
backend=backend,
middleware=[
FilesystemMiddleware(backend=backend, tools=["read_file", "ls", "glob", "grep"]),
],
)有关更多细节,请参阅限制文件系统工具。
自定义技能目录在系统提示词中的呈现方式
使用自定义 system_prompt= 模板覆盖 SkillsMiddleware,以控制 自动生成的技能目录如何被组织——例如添加周边上下文、 重新排列各节,或将路由提示(优先级、类别偏好)放在与 目录相同的块中。该模板必须包含 {skills_locations}、 {skills_list} 和 {skills_load_warnings} 槽位,middleware 会在 每一轮替换它们——因此随着技能被添加、 重命名或移除,目录会自动保持同步。
下面的示例将路由提示与目录放在一起:
python
from deepagents import create_deep_agent
from deepagents.backends.filesystem import FilesystemBackend
from deepagents.middleware.skills import SkillsMiddleware
backend = FilesystemBackend(root_dir="/path/to/skills")
custom_skills_prompt = """## Skills
Routing hints:
- For research topics → prefer arxiv-search first
- For general questions → use web-research
- For LangGraph documentation → langgraph-docs
{skills_locations}
{skills_list}
{skills_load_warnings}
"""
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
middleware=[
SkillsMiddleware(
backend=backend,
sources=["/path/to/skills"],
system_prompt=custom_skills_prompt,
),
],
)在这种情况下,不要同时向 create_deep_agent 传入 skills=:该覆盖实例 已经拥有 backend 和 sources。任何缺失的槽位({skills_locations}、 {skills_list}、{skills_load_warnings})都会在构造时引发 ValueError。
抑制自动生成的技能目录
使用 system_prompt=None 覆盖 SkillsMiddleware,从而加载技能时 完全不向系统提示词追加技能节。技能仍留在 state["skills_metadata"] 中,技能文件仍可通过文件系统 工具访问,因此倾向于在 create_deep_agent(system_prompt=...) 中自行编写描述块(例如 一份紧凑的手写路由摘要)的调用方,可以 添加他们所需的确切指引,而无需内置的那份。
python
from deepagents import create_deep_agent
from deepagents.backends.filesystem import FilesystemBackend
from deepagents.middleware.skills import SkillsMiddleware
backend = FilesystemBackend(root_dir="/path/to/skills")
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
system_prompt="""You are a research assistant.
Skills (read the SKILL.md file when relevant):
- /path/to/skills/arxiv-search/SKILL.md — papers
- /path/to/skills/web-research/SKILL.md — general web lookups
- /path/to/skills/langgraph-docs/SKILL.md — LangGraph documentation
""",
middleware=[
SkillsMiddleware(
backend=backend,
sources=["/path/to/skills"],
system_prompt=None,
),
],
)create_deep_agent 的 system_prompt 中的技能名称、路径以及任何手写提示 都由调用方负责在技能变化时保持同步。
解释器
使用解释器添加一个 eval 工具,它在限定作用域的 QuickJS 运行时中运行 JavaScript。当智能体需要以编程方式组合工具、批处理工作、在代码中处理错误,或在不使用完整 shell 环境的情况下转换结构化数据时,解释器很有用。
python
from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
middleware=[CodeInterpreterMiddleware()],
)python
from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware
agent = create_deep_agent(
model="openai:gpt-5.5",
middleware=[CodeInterpreterMiddleware()],
)python
from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
middleware=[CodeInterpreterMiddleware()],
)python
from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
middleware=[CodeInterpreterMiddleware()],
)python
from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
middleware=[CodeInterpreterMiddleware()],
)python
from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
middleware=[CodeInterpreterMiddleware()],
)python
from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
middleware=[CodeInterpreterMiddleware()],
)有关设置、以编程方式进行工具调用、子智能体编排和限制,请参阅解释器。
子智能体
要隔离细节工作并避免上下文膨胀,请使用子智能体:
python
import os
from typing import Literal
from deepagents import create_deep_agent
from tavily import TavilyClient
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
research_subagent = {
"name": "research-agent",
"description": "Used to research more in depth questions",
"system_prompt": "You are a great researcher",
"tools": [internet_search],
"model": "openai:gpt-5.5", # Optional override, defaults to main agent model
}
subagents = [research_subagent]
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
subagents=subagents,
)有关更多信息,请参阅 子智能体。
后端
deep agent 的工具可以利用虚拟文件系统来存储、访问和编辑文件。默认情况下,deep agents 使用 StateBackend。
如果你使用技能或记忆,则必须在创建智能体之前将所需的技能或记忆文件添加到后端。
一种线程范围的文件系统后端,存储在 langgraph 状态中。
文件在一个线程内跨轮次持久保留(通过你的检查点器),且不跨线程共享。
python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
# By default we provide a StateBackend
agent = create_deep_agent(model="google_genai:gemini-3.6-flash")
# Under the hood, it looks like
agent2 = create_deep_agent(
model="google_genai:gemini-3.6-flash",
backend=StateBackend(),
)python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
# By default we provide a StateBackend
agent = create_deep_agent(model="openai:gpt-5.5")
# Under the hood, it looks like
agent2 = create_deep_agent(
model="openai:gpt-5.5",
backend=StateBackend(),
)python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
# By default we provide a StateBackend
agent = create_deep_agent(model="anthropic:claude-sonnet-5")
# Under the hood, it looks like
agent2 = create_deep_agent(
model="anthropic:claude-sonnet-5",
backend=StateBackend(),
)python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
# By default we provide a StateBackend
agent = create_deep_agent(model="openrouter:z-ai/glm-5.2")
# Under the hood, it looks like
agent2 = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
backend=StateBackend(),
)python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
# By default we provide a StateBackend
agent = create_deep_agent(model="fireworks:accounts/fireworks/models/glm-5p2")
# Under the hood, it looks like
agent2 = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
backend=StateBackend(),
)python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
# By default we provide a StateBackend
agent = create_deep_agent(model="baseten:zai-org/GLM-5.2")
# Under the hood, it looks like
agent2 = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
backend=StateBackend(),
)python
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
# By default we provide a StateBackend
agent = create_deep_agent(model="ollama:north-mini-code-1.0")
# Under the hood, it looks like
agent2 = create_deep_agent(
model="ollama:north-mini-code-1.0",
backend=StateBackend(),
)有关更多信息,请参阅后端。
沙箱
沙箱是一种专门的后端,它们在带有自身文件系统的隔离环境中运行智能体代码,并提供一个用于执行 shell 命令的 execute 工具。 当你希望你的 deep agent 能够写入文件、安装依赖并运行命令,而不更改本地机器上的任何内容时,请使用沙箱后端。
你在创建 deep agent 时,通过将沙箱后端传给 backend 来配置沙箱:
bash
pip install "langsmith[sandbox]"bash
uv add "langsmith[sandbox]"python
from deepagents import create_deep_agent
from deepagents.backends import LangSmithSandbox
from langchain_anthropic import ChatAnthropic
from langsmith.sandbox import SandboxClient
client = SandboxClient()
ls_sandbox = client.create_sandbox()
backend = LangSmithSandbox(sandbox=ls_sandbox)
agent = create_deep_agent(
model=ChatAnthropic(model="claude-sonnet-4-6"),
system_prompt="You are a Python coding assistant with sandbox access.",
backend=backend,
)
try:
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Create a small Python package and run pytest",
}
]
}
)
finally:
client.delete_sandbox(ls_sandbox.name)更多信息,请参阅沙箱。
人在回路
某些工具操作可能较为敏感,需要人工在执行前予以批准。 你可以为每个工具配置审批:
python
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)python
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="openai:gpt-5.5",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)python
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)python
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)python
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)python
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)python
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)你可以为智能体和子智能体在工具调用时以及从工具调用内部配置中断。 更多信息,请参阅人在回路。
技能
你可以使用技能为你的深度智能体提供新的能力与专长。 工具通常涵盖较底层功能,如原生文件系统操作,而技能可以包含关于如何完成任务的详细说明、参考信息,以及其他资源(例如模板)。 只有当智能体判定该技能对当前提示词有用时,才会加载这些文件。 这种渐进式披露减少了智能体在启动时需要考量的 token 与上下文数量。
示例技能请参见 Deep Agents 示例技能。
要为你的深度智能体添加技能,请将它们作为参数传给 create_deep_agent:
python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
backend = StateBackend()
skill_url = "https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/libs/code/examples/skills/langgraph-docs/SKILL.md"
with urlopen(skill_url) as response:
skill_content = response.read().decode('utf-8')
skills_files = {
"/skills/langgraph-docs/SKILL.md": create_file_data(skill_content),
}
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
backend=backend,
skills=["/skills/"],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [{"role": "user", "content": "What is langgraph?"}],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": skills_files,
},
config={"configurable": {"thread_id": "12345"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
backend = StateBackend()
skill_url = "https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/libs/code/examples/skills/langgraph-docs/SKILL.md"
with urlopen(skill_url) as response:
skill_content = response.read().decode('utf-8')
skills_files = {
"/skills/langgraph-docs/SKILL.md": create_file_data(skill_content),
}
agent = create_deep_agent(
model="openai:gpt-5.5",
backend=backend,
skills=["/skills/"],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [{"role": "user", "content": "What is langgraph?"}],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": skills_files,
},
config={"configurable": {"thread_id": "12345"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
backend = StateBackend()
skill_url = "https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/libs/code/examples/skills/langgraph-docs/SKILL.md"
with urlopen(skill_url) as response:
skill_content = response.read().decode('utf-8')
skills_files = {
"/skills/langgraph-docs/SKILL.md": create_file_data(skill_content),
}
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
backend=backend,
skills=["/skills/"],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [{"role": "user", "content": "What is langgraph?"}],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": skills_files,
},
config={"configurable": {"thread_id": "12345"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
backend = StateBackend()
skill_url = "https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/libs/code/examples/skills/langgraph-docs/SKILL.md"
with urlopen(skill_url) as response:
skill_content = response.read().decode('utf-8')
skills_files = {
"/skills/langgraph-docs/SKILL.md": create_file_data(skill_content),
}
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
backend=backend,
skills=["/skills/"],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [{"role": "user", "content": "What is langgraph?"}],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": skills_files,
},
config={"configurable": {"thread_id": "12345"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
backend = StateBackend()
skill_url = "https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/libs/code/examples/skills/langgraph-docs/SKILL.md"
with urlopen(skill_url) as response:
skill_content = response.read().decode('utf-8')
skills_files = {
"/skills/langgraph-docs/SKILL.md": create_file_data(skill_content),
}
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
backend=backend,
skills=["/skills/"],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [{"role": "user", "content": "What is langgraph?"}],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": skills_files,
},
config={"configurable": {"thread_id": "12345"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
backend = StateBackend()
skill_url = "https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/libs/code/examples/skills/langgraph-docs/SKILL.md"
with urlopen(skill_url) as response:
skill_content = response.read().decode('utf-8')
skills_files = {
"/skills/langgraph-docs/SKILL.md": create_file_data(skill_content),
}
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
backend=backend,
skills=["/skills/"],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [{"role": "user", "content": "What is langgraph?"}],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": skills_files,
},
config={"configurable": {"thread_id": "12345"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
backend = StateBackend()
skill_url = "https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/libs/code/examples/skills/langgraph-docs/SKILL.md"
with urlopen(skill_url) as response:
skill_content = response.read().decode('utf-8')
skills_files = {
"/skills/langgraph-docs/SKILL.md": create_file_data(skill_content),
}
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
backend=backend,
skills=["/skills/"],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [{"role": "user", "content": "What is langgraph?"}],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": skills_files,
},
config={"configurable": {"thread_id": "12345"}},
)记忆
使用 AGENTS.md 文件为你的深度智能体提供额外上下文。
TIP
要生成编码智能体可通过 AGENTS.md 发现的仓库 wiki,请参阅 OpenWiki。
在创建深度智能体时,你可以向 memory 参数传入一个或多个文件路径:
python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
with urlopen(
"https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/examples/text-to-sql-agent/AGENTS.md"
) as response:
agents_md = response.read().decode("utf-8")
checkpointer = MemorySaver()
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
memory=[
"/AGENTS.md"
],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Please tell me what's in your memory files.",
}
],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": {"/AGENTS.md": create_file_data(agents_md)},
},
config={"configurable": {"thread_id": "123456"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
with urlopen(
"https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/examples/text-to-sql-agent/AGENTS.md"
) as response:
agents_md = response.read().decode("utf-8")
checkpointer = MemorySaver()
agent = create_deep_agent(
model="openai:gpt-5.5",
memory=[
"/AGENTS.md"
],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Please tell me what's in your memory files.",
}
],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": {"/AGENTS.md": create_file_data(agents_md)},
},
config={"configurable": {"thread_id": "123456"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
with urlopen(
"https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/examples/text-to-sql-agent/AGENTS.md"
) as response:
agents_md = response.read().decode("utf-8")
checkpointer = MemorySaver()
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
memory=[
"/AGENTS.md"
],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Please tell me what's in your memory files.",
}
],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": {"/AGENTS.md": create_file_data(agents_md)},
},
config={"configurable": {"thread_id": "123456"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
with urlopen(
"https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/examples/text-to-sql-agent/AGENTS.md"
) as response:
agents_md = response.read().decode("utf-8")
checkpointer = MemorySaver()
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
memory=[
"/AGENTS.md"
],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Please tell me what's in your memory files.",
}
],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": {"/AGENTS.md": create_file_data(agents_md)},
},
config={"configurable": {"thread_id": "123456"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
with urlopen(
"https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/examples/text-to-sql-agent/AGENTS.md"
) as response:
agents_md = response.read().decode("utf-8")
checkpointer = MemorySaver()
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
memory=[
"/AGENTS.md"
],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Please tell me what's in your memory files.",
}
],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": {"/AGENTS.md": create_file_data(agents_md)},
},
config={"configurable": {"thread_id": "123456"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
with urlopen(
"https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/examples/text-to-sql-agent/AGENTS.md"
) as response:
agents_md = response.read().decode("utf-8")
checkpointer = MemorySaver()
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
memory=[
"/AGENTS.md"
],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Please tell me what's in your memory files.",
}
],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": {"/AGENTS.md": create_file_data(agents_md)},
},
config={"configurable": {"thread_id": "123456"}},
)python
from urllib.request import urlopen
from deepagents import create_deep_agent
from deepagents.backends.utils import create_file_data
from langgraph.checkpoint.memory import MemorySaver
with urlopen(
"https://raw.githubusercontent.com/langchain-ai/deepagents/refs/heads/main/examples/text-to-sql-agent/AGENTS.md"
) as response:
agents_md = response.read().decode("utf-8")
checkpointer = MemorySaver()
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
memory=[
"/AGENTS.md"
],
checkpointer=checkpointer,
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Please tell me what's in your memory files.",
}
],
# Seed the default StateBackend's in-state filesystem (virtual paths must start with "/").
"files": {"/AGENTS.md": create_file_data(agents_md)},
},
config={"configurable": {"thread_id": "123456"}},
)配置档案
一个 harness profile 是一组可复用的、按模型划分的配置,当选中匹配的模型时,create_deep_agent 会自动应用它。当你希望行为跟随模型——而非调用处——时,profiles 就是合适的工具,例如为 Claude 的指令风格调优的系统提示词后缀、为 GPT 重写的工具描述,或仅在特定服务商下才有意义的额外 middleware。
单个 profile 可以携带:自定义基础系统提示词(base_system_prompt)、追加的后缀(system_prompt_suffix)、工具描述覆盖、要排除的工具或 middleware、要注入的额外 middleware,以及对自动添加的通用子智能体所做的修改。
python
from deepagents import HarnessProfile, register_harness_profile
# Append a system-prompt suffix whenever gpt-5.5 is selected.
register_harness_profile(
"openai:gpt-5.5",
HarnessProfile(system_prompt_suffix="Respond in under 100 words."),
)有关注册键、合并语义和插件打包,请参阅 Profiles。一个范围更窄的配套 API——服务商 profiles——用于为某个服务商打包模型构建参数(API 密钥、超时、重试设置)。
结构化输出
Deep Agents 支持结构化输出。 你可以通过在调用 create_deep_agent() 时把所需的结构化输出 schema 作为 response_format 参数传入来设置它。 当模型生成结构化数据后,它会被捕获、校验,并返回在深度智能体状态的 'structured_response' 键中。
python
import os
from typing import Literal
from pydantic import BaseModel, Field
from tavily import TavilyClient
from deepagents import create_deep_agent
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""Run a web search"""
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
class WeatherReport(BaseModel):
"""A structured weather report with current conditions and forecast."""
location: str = Field(description="The location for this weather report")
temperature: float = Field(description="Current temperature in Celsius")
condition: str = Field(
description="Current weather condition (e.g., sunny, cloudy, rainy)"
)
humidity: int = Field(description="Humidity percentage")
wind_speed: float = Field(description="Wind speed in km/h")
forecast: str = Field(description="Brief forecast for the next 24 hours")
agent = create_deep_agent(
model=model,
response_format=WeatherReport,
tools=[internet_search],
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "What's the weather like in San Francisco?",
}
]
}
)
print(result["structured_response"])
# location='San Francisco, California' temperature=18.3 condition='Sunny' humidity=48 wind_speed=7.6 forecast='Pleasant sunny conditions expected to continue with temperatures around 64°F (18°C) during the day, dropping to around 52°F (11°C) at night. Clear skies with minimal precipitation expected.'View example trace
打开本示例的一个公开 LangSmith 运行记录。
更多信息与示例,请参阅响应格式。
高级
create_deep_agent 会在 create_agent 之上预先组装一套 middleware 栈。若要构建完全自定义的智能体——精确选择要包含哪些能力——请参阅配置 harness。