本指南涵盖了将 Deep Agent 从本地原型转变为生产部署的考量因素。它将逐步介绍记忆作用域划分、配置执行环境、添加护栏以及连接前端。
智能体利用来自记忆和执行环境的信息来完成任务。
在生产环境中,有几个原语决定了信息如何共享和访问:
线程 (Thread) :单次对话。消息历史和暂存文件默认作用于线程,不会延续。
用户 (User) :与您的智能体交互的人。记忆和文件可以是用户私有的,也可以在用户之间共享。身份和授权来自您的 认证层 。
助手 (Assistant) :配置的智能体实例。记忆和文件可以绑定到一个助手,也可以在所有助手之间共享。
本页涵盖:
LangSmith 部署
将 Deep Agent 投入生产的最快方法是 deepagents deploy ,它打包您的智能体配置并通过一条命令将其部署为 LangSmith 部署。或者,您可以直接配置 LangSmith 部署 。任一路径都会配置您的智能体所需的基础设施:助手 、线程 、运行 、存储和检查点器,因此您无需自行设置这些。它还为您提供开箱即用的 认证 、Webhook 、Cron 任务 和 可观测性 ,并可以通过 MCP 或 A2A 暴露您的智能体。
对于基于 CLI 的方法,请参阅 使用 CLI 部署 。对于手动设置,请参阅 LangSmith 部署快速入门 。
除非另有说明,本页上的所有代码片段均使用以下 langgraph.json:
{
" dependencies " : [ "." ],
" graphs " : {
" agent " : "./agent.py:agent"
},
" env " : ".env"
}
langgraph.json 是告诉 LangGraph 平台如何构建和运行您的应用的配置文件。它位于项目的根目录,对于本地开发(使用 langgraph dev)和生产部署都是必需的。关键字段如下:
字段 描述 dependencies要安装的包。["."] 将当前目录作为包安装(从 requirements.txt、pyproject.toml 或 package.json 读取)。 graphs将图 ID 映射到其代码位置。每个条目为 "<id>": "./<file>:<variable>",其中 <id> 是您通过 API 调用图时使用的名称,<variable> 是从 <file> 导出的已编译图或构造函数。 env包含环境变量(API 密钥、秘密)的 .env 文件路径。这些在构建时设置并在运行时可用。
有关完整配置选项(自定义 Docker 步骤、存储索引、认证处理程序等)的列表,请参阅 应用结构 。
生产考量
多租户
当您的智能体服务于多个用户时,您需要处理三个问题:验证每个用户是谁、控制他们可以访问的内容,以及管理智能体代表他们行事时使用的凭据。
用户身份和访问控制
LangSmith 部署 支持 自定义认证 以建立用户身份,以及 授权处理程序 以控制对线程、助手和存储命名空间等资源的访问。授权处理程序在认证成功后运行,可以:
使用所有权元数据标记资源(例如,owner: user_id)
返回过滤器,以便用户只能看到自己的资源
对未经授权的操作返回 HTTP 403 拒绝访问
有关分步教程,请参阅 使对话私有 。有关演练,请观看 自定义认证视频 。
您如何 划分记忆作用域 和 执行环境 决定了用户之间共享哪些数据。详见下文各节。
团队访问控制 (RBAC)
LangSmith 的 基于角色的访问控制 管理您的团队中谁可以部署、配置和监控智能体。这与上述最终用户授权是分开的。
角色 访问权限 工作区管理员 (Workspace Admin) 包括设置和成员管理在内的完全权限 工作区编辑 (Workspace Editor) 创建和修改资源,但不能删除运行或管理成员 工作区查看者 (Workspace Viewer) 只读访问
具有细粒度权限的自定义角色在企业计划中可用。请参阅 RBAC 参考 了解完整的权限模型。
最终用户凭据
当您的智能体需要代表用户调用外部 API 时(例如,读取他们的 GitHub 仓库、发送 Slack 消息、查询他们的数据仓库),您需要一种方法将用户的凭据传递给智能体,而无需硬编码它们。
通过 Agent Auth 进行 OAuth。 Agent Auth 提供托管的 OAuth 2.0 流程。配置 OAuth 提供商,智能体可以请求针对每个用户作用域的 Token。首次使用时,智能体会 中断 执行并呈现 OAuth 同意 URL。用户认证后,智能体使用有效 Token 恢复执行。Token 会自动存储和刷新。
from langchain_auth import Client
from langchain . tools import tool , ToolRuntime
auth_client = Client ()
# 在您的智能体工具内:
@tool
async def github_action ( runtime : ToolRuntime ):
"""代表用户通过 GitHub 执行操作。"""
auth_result = await auth_client . authenticate (
provider = "github" ,
scopes = [ "repo" , "read:org" ],
user_id = runtime . server_info . user . identity ,
)
# 使用 auth_result.token 代表用户进行 GitHub API 调用
沙盒的凭据注入。 如果您的智能体在 沙盒 内运行调用外部 API 的代码,沙盒认证代理 可以自动将凭据注入出站请求,因此沙盒代码永远不会接收原始 API 密钥。有关设置详情,请参阅 管理密钥 。
工作区秘密。 对于所有用户共享的 API 密钥(例如您组织的 LLM 提供商密钥、搜索 API 密钥),将它们作为 工作区秘密 存储在 LangSmith 中。有关详情,请参阅 管理密钥 。
基于 LLM 的应用程序 heavily I/O-bound:调用语言模型、数据库和外部服务。异步编程让这些操作并发运行而不是阻塞,从而提高吞吐量和响应能力。
LangChain 遵循在异步方法名前缀 a 的约定(例如,ainvoke、abefore_agent、astream)。同步和异步变体存在于同一个类或命名空间中。
在生产环境中构建时:
创建异步工具。 LangChain 在单独的线程中运行同步工具以避免阻塞,但原生异步完全避免了线程开销。
使用异步中间件方法。 自定义 中间件 应实现异步钩子(例如,abefore_agent 而不是 before_agent)。
对外部资源生命周期使用异步。 创建 沙盒 或连接到 MCP 服务器 涉及网络调用,应该被 await。这就是为什么配置这些资源的 图工厂 是异步的。
持久性
Deep Agents 运行在 LangGraph 上,后者开箱即用地提供 持久执行 。持久化 层在每一步检查点化状态,因此因失败、超时或 人机协同 暂停而中断的运行可以从其最后记录的状态恢复,而无需重新处理之前的步骤。对于产生许多子代理的长时间运行的 Deep Agents,这意味着运行中的失败不会丢失已完成的工作。
检查点化还启用:
无限 中断 。 人机协同工作流可以暂停几分钟或几天,并确切地从离开的位置恢复。
时间旅行 。 每个检查点化的步骤都是一个您可以倒回的快照,如果出现问题,可以让您从早期状态重放。
安全处理敏感操作。 对于涉及支付或其他不可逆操作的工作流,检查点提供审计跟踪和恢复点,以检查导致操作的确切状态。
没有记忆,每次对话都从头开始。记忆让您的智能体跨对话保留信息(用户偏好、学到的指令、过去的经验),以便它可以随时间个性化其行为。有关记忆类型的概述,请参阅 记忆概念指南 。
作用域划分
记忆在对话之间始终是持久的。主要问题是它如何跨用户和助手边界划分作用域。正确的作用域取决于谁应该查看和修改数据:
作用域 命名空间 用例 示例 用户 (推荐默认)(user_id)每用户偏好和上下文 ”我偏好简洁的回复” 助手 (assistant_id)一个助手的共享指令 ”将帖子限制在 280 个字符” 全局 (org_id)所有用户和助手的只读策略 ”永不披露内部定价”
共享记忆(助手、用户或组织作用域)是 Prompt 注入的载体。如果一个用户可以写入另一个用户的对话读取的记忆,恶意用户可以将指令注入该共享状态。在适当的地方强制只读访问。例如,使组织范围的策略只能通过应用代码写入,而不是由智能体本身写入。使用 权限 声明式拒绝写入共享路径,或使用 后端策略钩子 进行自定义验证逻辑。
在 Deep Agents 中,记忆作为文件存储在虚拟文件系统中。默认情况下,文件仅持续单次对话。要持久化它们,将像 /memories/ 这样的路径路由到写入 LangGraph Store 的 StoreBackend 。使用 CompositeBackend 为智能体提供临时暂存空间和持久的 长期记忆 。
下面显示的 rt.server_info 和 rt.execution_info 命名空间模式需要 deepagents>=0.5.0。
按 user_id 划分命名空间。每个用户拥有自己的私有记忆。这是推荐的默认值,因为大多数应用部署单个助手。 from deepagents import create_deep_agent
from deepagents . backends import CompositeBackend , StateBackend , StoreBackend
agent = create_deep_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
backend = CompositeBackend (
default = StateBackend (),
routes = {
"/memories/" : StoreBackend (
namespace = lambda rt : (
rt . server_info . assistant_id ,
rt . server_info . user . identity ,
),
),
},
),
system_prompt = """您在 /memories/ 处拥有持久记忆。
在每次对话开始时读取 /memories/instructions.txt 以获取
积累的知识和偏好。当您学到应该
持久化的内容时,更新该文件。""" ,
)
按 assistant_id 划分命名空间。记忆在同一助手的所有用户之间共享,因此任何用户都可以读取或更新它。用于共享指令或适用于使用给定助手的每个人的知识(例如,“总是以正式语气回复”)。 from deepagents import create_deep_agent
from deepagents . backends import CompositeBackend , StateBackend , StoreBackend
agent = create_deep_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
backend = CompositeBackend (
default = StateBackend (),
routes = {
"/memories/" : StoreBackend (
namespace = lambda rt : (
rt . server_info . assistant_id ,
),
),
},
),
)
仅按 user_id 划分命名空间。记忆跟随用户跨越所有助手。用于全局用户 profile(名称、时区、通信偏好),无论用户与哪个助手交谈都应适用。 from deepagents import create_deep_agent
from deepagents . backends import CompositeBackend , StateBackend , StoreBackend
agent = create_deep_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
backend = CompositeBackend (
default = StateBackend (),
routes = {
"/memories/" : StoreBackend (
namespace = lambda rt : ( rt . server_info . user . identity ,),
),
},
),
)
按 org_id 划分命名空间。记忆在所有用户和所有助手之间共享。通常用于组织范围的策略(合规规则、品牌指南),智能体应为只读。写入权限应限制为应用代码以防止 Prompt 注入。 from deepagents import create_deep_agent
from deepagents . backends import CompositeBackend , StateBackend , StoreBackend
agent = create_deep_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
backend = CompositeBackend (
default = StateBackend (),
routes = {
"/memories/" : StoreBackend (
namespace = lambda rt : ( rt . context . org_id ,),
),
},
),
)
您也可以使用 Store API 从应用代码读取和写入存储。有关示例,请参阅 高级用法 。
有关完整的命名空间工厂 API,请参阅 命名空间工厂 。有关自我改进指令和知识库等记忆模式,请参阅 长期记忆 。
执行环境
在本地,智能体可以在磁盘上读写文件并直接运行 shell 命令。在生产环境中,您需要考虑隔离和持久性。正确的设置取决于您的智能体是否需要执行代码:
文件系统后端 如果您的智能体仅读写文件则足够。选择匹配您持久性需求的后端:临时暂存空间、持久存储或两者混合。
沙盒 添加一个带有 execute 工具的隔离容器用于运行 shell 命令。如果您的智能体需要运行代码、安装包或执行文件 I/O 以外的任何操作,请使用沙盒。
文件系统
根据需要持久化的内容选择后端:
有关后端完整列表及如何构建自定义后端,请参阅 后端 。
FilesystemBackend 和 LocalShellBackend 直接访问主机。不要在已部署的智能体中使用它们。
如果您的智能体需要运行代码(不仅仅是读写文件),请使用 沙盒 。沙盒在隔离容器内提供文件系统和用于运行 shell 命令的 execute 工具。这种隔离也保护您的主机:如果智能体的代码耗尽内存或崩溃,只有沙盒受影响。您的服务器保持运行。
生命周期
关键决策是沙盒存活多久。每次对话获得一个全新的沙盒,还是对话共享一个持久环境?
作用域 沙盒 ID 存储于 生命周期 示例用例 线程作用域 线程 元数据每对话全新,TTL 时清理 每次对话干净开始的数据分析机器人 助手作用域 助手 配置所有对话共享 跨对话维护克隆仓库的编码助手
下面的示例使用异步 图工厂 而不是静态图,因为沙盒需要 thread_id 或 assistant_id 来查找或创建正确的沙盒。图工厂不接收完整的 Runtime(没有 server_info 或 execution_info);相反,接受 RunnableConfig 并从 config["configurable"] 读取 thread_id 和 assistant_id。工厂是异步的,因为沙盒创建是 I/O 绑定操作,需要仅在调用时可用的每运行信息。
每次对话获得自己的沙盒。图工厂 从运行配置读取 thread_id,因此每个 线程 自动获得自己的隔离环境。提供商的基于标签的查找处理跨运行的去重。当沙盒 TTL 过期时清理。 from daytona import CreateSandboxFromSnapshotParams , Daytona
from deepagents import create_deep_agent
from langchain_core . runnables import RunnableConfig
from langchain_daytona import DaytonaSandbox
client = Daytona ()
async def agent ( config : RunnableConfig ):
thread_id = config [ " configurable " ][ "thread_id" ]
try :
sandbox = await client . find_one ( labels = { "thread_id" : thread_id })
except Exception :
sandbox = await client . create (
CreateSandboxFromSnapshotParams (
labels = { "thread_id" : thread_id },
auto_delete_interval = 3600 , # TTL: 空闲时清理
)
)
return create_deep_agent ( model = "google_genai:gemini-3.1-pro-preview" , backend = DaytonaSandbox ( sandbox = sandbox ))
所有对话共享一个沙盒。图工厂 从 config["configurable"] 读取 助手 ID,因此同一助手上的每个线程返回到相同的环境。文件、已安装的包和克隆的仓库跨对话持久化。 from daytona import CreateSandboxFromSnapshotParams , Daytona
from deepagents import create_deep_agent
from langchain_core . runnables import RunnableConfig
from langchain_daytona import DaytonaSandbox
client = Daytona ()
async def agent ( config : RunnableConfig ):
assistant_id = config [ " configurable " ][ "assistant_id" ]
try :
sandbox = await client . find_one ( labels = { "assistant_id" : assistant_id })
except Exception :
sandbox = await client . create (
CreateSandboxFromSnapshotParams ( labels = { "assistant_id" : assistant_id })
)
return create_deep_agent ( model = "google_genai:gemini-3.1-pro-preview" , backend = DaytonaSandbox ( sandbox = sandbox ))
助手作用域沙盒随时间积累文件、已安装的包和其他沙盒内状态。使用沙盒提供商配置 TTL,使用快照定期重置,或实现清理逻辑以防止沙盒的磁盘和内存无限增长。
因为 agent 变量是异步函数(而不是已编译的图),服务器将其视为 图工厂 并在每次运行时调用它,注入配置。工厂通过提供商的基于标签的搜索查找或创建沙盒,并返回连接到该沙盒的全新智能体图。
一旦使用 langgraph deploy 部署,使用 SDK 从您的应用代码调用智能体。客户端代码无论作用域如何都是相同的。作用域完全在上述智能体工厂中处理,但行为不同:
每个线程获得自己的沙盒。同一线程内的后续消息重用相同的沙盒,但新线程总是全新开始,没有来自之前对话的遗留文件或已安装的包。 from langgraph_sdk import get_client
client = get_client ( url = "<DEPLOYMENT_URL>" , api_key = "<LANGSMITH_API_KEY>" )
# 对话 1:安装 pandas 并分析数据
thread_1 = await client . threads . create ()
async for chunk in client . runs . stream (
thread_1 [ " thread_id " ],
"agent" ,
input = { "messages" : [{ "role" : "human" , "content" : "Install pandas and analyze sales_data.csv" }]},
stream_mode = "updates" ,
):
print ( chunk . data )
# 同一对话中的后续消息 — pandas 仍已安装
async for chunk in client . runs . stream (
thread_1 [ " thread_id " ],
"agent" ,
input = { "messages" : [{ "role" : "human" , "content" : "Now plot the results" }]},
stream_mode = "updates" ,
):
print ( chunk . data )
# 对话 2:全新沙盒 — pandas 未安装,没有来自对话 1 的文件
thread_2 = await client . threads . create ()
async for chunk in client . runs . stream (
thread_2 [ " thread_id " ],
"agent" ,
input = { "messages" : [{ "role" : "human" , "content" : "What packages are installed?" }]},
stream_mode = "updates" ,
):
print ( chunk . data )
所有线程共享一个沙盒。当沙盒具有重建成本高的状态时(例如克隆的仓库、已安装的依赖或构建产物),这很有用。同一助手上的任何对话都接续上一次结束的地方,无需重复设置。 from langgraph_sdk import get_client
client = get_client ( url = "<DEPLOYMENT_URL>" , api_key = "<LANGSMITH_API_KEY>" )
# 对话 1:克隆并设置项目
thread_1 = await client . threads . create ()
async for chunk in client . runs . stream (
thread_1 [ " thread_id " ],
"agent" ,
input = { "messages" : [{ "role" : "human" , "content" : "Clone https://github.com/org/repo and install dependencies" }]},
stream_mode = "updates" ,
):
print ( chunk . data )
# 对话 2:仓库和依赖仍在那里
thread_2 = await client . threads . create ()
async for chunk in client . runs . stream (
thread_2 [ " thread_id " ],
"agent" ,
input = { "messages" : [{ "role" : "human" , "content" : "Run the test suite and fix any failures" }]},
stream_mode = "updates" ,
):
print ( chunk . data )
文件传输
沙盒是隔离容器,因此您的应用代码无法直接访问其中的文件。使用 upload_files() 和 download_files() 在沙盒边界移动数据:
在智能体运行前播种沙盒 :上传用户文件、技能 脚本、配置或 持久记忆 ,以便智能体从一开始就拥有所需内容
在智能体完成后检索结果 :下载生成的产物(报告、图表、导出)并同步更新的记忆回未来对话
有关特定提供商的文件传输示例,请参阅 处理文件 。有关提供商设置、安全和生命周期模式,请参阅完整的 沙盒指南 。
智能体需要执行的 技能 脚本必须在智能体运行前上传到沙盒中。您可能还希望同步 记忆 ,以便智能体可以在容器内读取和更新它们。使用带有 before_agent 和 after_agent 钩子的 自定义中间件 在沙盒边界移动文件: from deepagents import create_deep_agent
from langchain . agents . middleware import AgentMiddleware , AgentState
from langgraph . runtime import Runtime
def _safe_filename ( key : str ) -> str :
"""拒绝包含路径遍历或通配符的键。"""
name = key . split ( "/" )[ - 1 ]
if ".." in name or any ( c in name for c in ( "*" , "?" )):
raise ValueError ( f "Invalid key: { key } " )
return name
class SandboxSyncMiddleware ( AgentMiddleware ):
"""在存储和沙盒之间同步技能和记忆。"""
def __init__ ( self , backend : CompositeBackend ):
super (). __init__ ()
self . backend = backend
async def abefore_agent ( self , state : AgentState , runtime : Runtime ) -> None :
"""上传技能脚本和记忆到沙盒。"""
user_id = runtime . server_info . user . identity
store = runtime . store
files = []
for item in await store . asearch (( "skills" , user_id )):
name = _safe_filename ( item . key )
files . append (( f "/skills/ { name } " , item . value [ " content " ]. encode ()))
for item in await store . asearch (( "memories" , user_id )):
name = _safe_filename ( item . key )
files . append (( f "/memories/ { name } " , item . value [ " content " ]. encode ()))
if files :
await self . backend . upload_files ( files )
async def aafter_agent ( self , state : AgentState , runtime : Runtime ) -> None :
"""同步更新的记忆回存储。"""
user_id = runtime . server_info . user . identity
store = runtime . store
items = await store . asearch (( "memories" , user_id ))
results = await self . backend . download_files (
[ f "/memories/ { item . key } " for item in items ]
)
for result in results :
if result . content is not None :
await store . aput (
( "memories" , user_id ),
result . path . split ( "/" )[ - 1 ],
{ "content" : result . content . decode ()},
)
backend = CompositeBackend (
default = DaytonaSandbox ( sandbox = sandbox ),
routes = {
"/skills/" : StoreBackend (
rt ,
namespace = lambda rt : ( "skills" , rt . server_info . user . identity ),
),
"/memories/" : StoreBackend (
rt ,
namespace = lambda rt : ( "memories" , rt . server_info . user . identity ),
),
},
)
agent = create_deep_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
backend = backend ,
middleware = [ SandboxSyncMiddleware ( backend )],
)
管理密钥
沙盒是隔离容器,因此来自您主机的环境变量在内部不可用。有两种方法为沙盒代码提供 API 密钥和其他秘密:
认证代理 (推荐)。 沙盒认证代理 拦截来自沙盒的出站请求并自动注入认证头。沙盒代码正常调用外部 API,代理根据目标主机添加正确的凭据。这意味着 API 密钥从不出现在沙盒代码、环境变量或日志中。
{
" proxy_config " : {
" rules " : [
{
" name " : "openai-api" ,
" match_hosts " : [ "api.openai.com" ],
" inject_headers " : {
" Authorization " : "Bearer ${OPENAI_API_KEY}"
}
},
{
" name " : "anthropic-api" ,
" match_hosts " : [ "api.anthropic.com" ],
" inject_headers " : {
" x-api-key " : "${ANTHROPIC_API_KEY}"
}
}
]
}
}
${SECRET_KEY} 引用解析为您 LangSmith 工作区设置 中存储的秘密。在创建引用它们的模板之前,在那里配置秘密。
工作区秘密。 对于不需要基于代理注入的 API 密钥(例如,由智能体服务器本身使用,而不是沙盒代码使用的密钥),将它们作为 工作区秘密 存储在 LangSmith 中。这些在运行时作为环境变量对工作区中的所有代理可用。
避免通过环境变量或文件上传将秘密传递到沙盒。智能体可以读取沙盒内任何可访问的文件或环境变量,包括凭据。认证代理将秘密完全保持在沙盒之外。
生产中的智能体自主运行,这意味着它们可以无限循环、命中速率限制或处理包含敏感信息的用户数据。Deep Agents 提供两层保护:
权限 :声明式允许/拒绝规则,控制智能体可以读取或写入哪些文件和目录。使用权限将智能体隔离到工作目录、保护敏感文件或强制只读记忆。
中间件 :包装模型和工具调用的钩子,用于速率限制、错误处理和数据隐私。
速率限制
这里的速率限制指的是限制智能体自己在一次运行内的 LLM 和工具使用,而不是传入请求的 API 网关速率限制。
没有限制,困惑的智能体可以通过在同一工具调用上循环或进行数百次模型调用在几分钟内耗尽您的 LLM API 预算。为每次运行的模型调用和工具执行设置上限:
from deepagents import create_deep_agent
from langchain . agents . middleware import ModelCallLimitMiddleware , ToolCallLimitMiddleware
agent = create_deep_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
middleware = [
ModelCallLimitMiddleware ( run_limit = 50 ),
ToolCallLimitMiddleware ( run_limit = 200 ),
],
)
使用 run_limit 限制单次调用内的调用(每轮重置)。使用 thread_limit 限制整个对话中的调用(需要检查点器)。有关完整配置,请参阅 ModelCallLimitMiddleware 和 ToolCallLimitMiddleware 。
处理错误
并非所有错误都应以相同方式处理。暂时性失败(网络超时、速率限制)应自动重试。LLM 可以恢复的错误(坏工具输出、解析失败)应反馈给模型。需要人工输入的错误应暂停智能体。有关带代码示例的完整分解,请参阅 适当处理错误 。
中间件处理暂时性情况。模型调用和工具调用各有自己的带指数退避的重试中间件。如果您的主要模型提供商完全宕机,回退中间件切换到替代方案:
from deepagents import create_deep_agent
from langchain . agents . middleware import (
ModelFallbackMiddleware ,
ModelRetryMiddleware ,
ToolRetryMiddleware ,
)
agent = create_deep_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
middleware = [
# 在速率限制、超时和 5xx 错误时重试模型调用
ModelRetryMiddleware ( max_retries = 3 , backoff_factor = 2.0 , initial_delay = 1.0 ),
# 如果主要模型完全宕机,回退到替代方案
ModelFallbackMiddleware ( "gpt-4.1" ),
# 重试命中外部 API 的特定工具(不是所有工具)
ToolRetryMiddleware (
max_retries = 2 ,
tools = [ "search" , "fetch_url" ],
retry_on = ( TimeoutError , ConnectionError ),
),
],
)
将 ToolRetryMiddleware 作用域限定到特定工具,而不是重试一切。失败的 filesystem read_file 不会从重试中受益,但超时的 web 搜索可能会。有关完整配置,请参阅 ModelRetryMiddleware 和 ModelFallbackMiddleware 。
数据隐私
如果您的智能体处理可能包含电子邮件、信用卡号或其他 PII 的用户输入,您可以在其到达模型或存储在日志中之前检测和处理它:
from deepagents import create_deep_agent
from langchain . agents . middleware import PIIMiddleware
agent = create_deep_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
middleware = [
PIIMiddleware ( "email" , strategy = "redact" , apply_to_input = True ),
PIIMiddleware ( "credit_card" , strategy = "mask" , apply_to_input = True ),
],
)
策略包括 redact(替换为 [REDACTED_EMAIL])、mask(部分掩码,如 ****-****-****-1234)、hash(确定性哈希)和 block(抛出错误)。您也可以为领域特定模式编写自定义检测器。
有关完整配置,请参阅 PIIMiddleware 。
有关可用中间件的完整列表,请参阅 预构建中间件 。
Deep Agents 使用 useStream 将您的 UI 连接到智能体后端。useStream 是一个前端钩子(可用于 React、Vue、Svelte 和 Angular),实时流式传输来自您的智能体的消息、子代理进度和自定义状态。
在本地,useStream 指向 http://localhost:2024。在生产环境中,将其指向您的 LangSmith 部署 并配置重连,以便用户在连接断开时不丢失进度。
import { useStream } from "@langchain/react" ;
function App () {
const stream = useStream < typeof agent > ( {
apiUrl : "https://your-deployment.langsmith.dev" ,
assistantId : "agent" ,
reconnectOnMount : true , // 页面刷新或导航后恢复流
fetchStateHistory : true , // 挂载时加载完整线程历史
} ) ;
}
reconnectOnMount 自动接续进行中的运行。如果用户在智能体工作时刷新,他们将看到它继续而不是空白屏幕。fetchStateHistory 加载线程的完整对话历史,因此返回用户可以看到之前的消息。
对于产生许多子代理的 Deep Agent 工作流,提交时设置高 recursionLimit 以避免切断长时间运行的执行:
stream . submit (
{ messages : [ { type : "human" , content : text } ] },
{
streamSubgraphs : true ,
config : { recursionLimit : 10000 },
},
) ;
有关特定于 Deep Agents 的 UI 模式,例如子代理卡片、待办列表和自定义状态渲染,请参阅 前端指南 。
连接这些文档 到 Claude、VSCode 等 via MCP 以获取实时答案。