主题
Managed Deep Agents 运行时
从智能体工厂、工具与中间件接收的原生运行时类型中,读取运行上下文、已验证的调用方、渠道投递与沙箱。
智能体工厂、工具与中间件钩子各自接收一个运行时(runtime)。它是原生的 LangChain 或 LangGraph 运行时,额外增加了三个托管字段。上下文(context)、状态(state)与存储(store)的用法与 LangChain 中一致,你的类型检查器也能看到你自己定义的上下文类型。
NOTE
Managed Deep Agents 目前处于**公开 beta**阶段,在 LangSmith Cloud 上提供。
理解运行时类型
针对你的代码运行的每个位置,Managed Deep Agents 各扩展了一个原生类型:
| 代码 | 类型 | 扩展自 |
|---|---|---|
| 智能体工厂 | ManagedServerRuntime[Context] | langgraph_sdk.runtime 的 ServerRuntime |
| 工具 | ManagedToolRuntime[Context, State] | langchain.tools 的 ToolRuntime |
| 中间件钩子 | ManagedRuntime[Context] | langgraph.runtime 的 Runtime |
类型参数遵循原生的顺序,因此 ManagedToolRuntime 先接收上下文、再接收状态。两个参数都是可选的。ManagedDeepAgentRuntime 是 ManagedToolRuntime 的别名。
每个类型都保留原生的全部字段与方法,例如 context、state、store、stream_writer 与 tool_call_id。Managed Deep Agents 额外增加了三个字段:
server_info:原生服务器元数据加上已验证的调用方。工具与中间件会收到它。channel:启动本次运行的、经过验证的渠道投递。直连 API 与计划任务运行时为None。backend:该线程的沙箱文件系统。项目未声明沙箱时为None。
区分上下文、调用方与投递
上下文、调用方与渠道投递来自不同来源,信任级别也不同。运行时把它们保存在各自独立的字段中:
- 上下文来自 API 调用的
context参数。LangGraph 会按你的上下文 schema 校验它。用它来选择行为,而不是授予访问权限。 - 服务器信息来自身份,后者从已验证的 token 解析出调用方。Managed Deep Agents 会剥离客户端通过
configurable传入的身份键,因此授权使用这个字段。 - 渠道来自托管渠道 ingress,它在运行开始前验证投递。直连 API 调用上下文里的
channel键不会创建runtime.channel。
API 调用方在输入旁传入上下文:
python
await client.runs.create(
thread_id,
"support",
input={"messages": [{"role": "user", "content": "Where is my order?"}]},
context={"tier": "fast"},
)处理不带上下文的渠道运行
渠道投递不携带应用上下文。Managed Deep Agents 会在 LangGraph 校验上下文之前移除投递数据,因此必填的上下文字段不会让渠道运行被拒绝。直连 API 运行保持原生校验。
在渠道运行中,上下文在智能体工厂、工具与中间件里都是 None。
要区分渠道运行与直连运行,检查 runtime.channel 即可。
在智能体工厂中读取运行时
智能体工厂接收一个 ManagedServerRuntime:原生 ServerRuntime 加上 channel 与 backend。编写工厂参见按运行选择配置。
Agent Server 也会调用工厂来读取 schema 与状态。access_context 标明操作类型。execution_runtime 在 threads.create_run 之外为 None,因此请从 execution_runtime.context 读取运行上下文。每次调用都要返回相同的图结构与 schema。
channel 可用,因此工厂可以按渠道选择配置。backend 恒为 None,因为沙箱由工厂来选定。原生的 user、store 与 ensure_user() 保持可用。
在工具中读取运行时
给运行时参数加上 ManagedToolRuntime 注解。Managed Deep Agents 会为该参数注入托管运行时,无论它叫什么名字:
python
from dataclasses import dataclass
from langchain.tools import tool
from managed_deepagents import ManagedToolRuntime
@dataclass
class AppContext:
region: str = "us"
@tool
async def check_order(order_id: str, runtime: ManagedToolRuntime[AppContext]) -> str:
"""Check the status of an order."""
principal = runtime.server_info.principal if runtime.server_info else None
if principal is None:
return "Sign in to check orders."
if runtime.channel is not None and runtime.channel.post is not None:
await runtime.channel.post({"type": "content", "content": "Checking that order now."})
region = runtime.context.region if runtime.context else "us"
return f"Order {order_id} for {principal.id} ships from {region}."当没有解析出调用方时,principal 缺失,因为身份是可选启用的。post 会发送一条附加消息,智能体的最终回复仍会自动发送。编写工具的更多内容参见自定义工具。
在中间件中读取运行时
生命周期与模型钩子保持原生的 Runtime[Context] 签名。用 isinstance 收窄类型即可读取托管字段:
python
from dataclasses import dataclass
from langchain.agents.middleware import AgentState, before_model
from langgraph.runtime import Runtime
from managed_deepagents import ManagedRuntime
@dataclass
class AppContext:
region: str = "us"
@before_model
def log_caller(state: AgentState, runtime: Runtime[AppContext]) -> None:
if not isinstance(runtime, ManagedRuntime):
return None
principal = runtime.server_info.principal if runtime.server_info else None
source = runtime.channel.provider if runtime.channel else "api"
print(f"{principal.id if principal else 'anonymous'} via {source}")
return None工具调用钩子接收 ManagedToolRuntime。声明式子智能体收到同样的托管字段。两者都不强制要求身份或沙箱声明。编写中间件的更多内容参见自定义中间件。
参考托管字段
服务器信息
runtime.server_info 是一个 ManagedServerInfo。它保留原生的 assistant_id、graph_id 与 user 字段,并新增调用方信息:
principal:已验证的调用方。id是调用方 ID,kind是"person"、"service"或"channel",claims存放 token 声明。claims["groups"]恒为字符串元组,claims["email"]恒为字符串。subject:本次运行所使用的权限主体。authority为"agent"或"user",并带agent_id与user_id。link:账户链接。status当前报告"not_required"。source:运行的入口。provider是诸如"http"、"schedule"、"studio"或"slack"的值,并带可选的thread_id。
从 principal 读取调用方,而不是原生的 user。身份提供方与声明映射参见身份。
渠道
runtime.channel 是一个 RuntimeChannel:
name:配置的渠道名称。provider:提供商标识,例如"slack"。event:类型化的投递。见下方的事件类型。raw_event:原始提供方载荷(当渠道提供时)。对 Slack,这是不含 HTTP 头与 Trigger 信封的内层 Slack 事件。恢复(resume)时可以省略它。post:一个异步函数,向启动本次运行的会话发送消息。渠道无法发送时为None。
event.type 是以下之一:
message:新消息。messages存放传入的消息。user_prompt_response:对交互式提示的回复。携带action、可选的value与可选的correlation_id。参见智能体自有的 interrupt。interrupt_resume:对某个待处理 interrupt 的恢复。resume存放恢复值。
post 接受内容消息 {"type": "content", "content": ...} 或原生消息 {"type": "native", "native": ...}。内置的 Slack 渠道只发送文本内容。回复目标保持私有:post 没有地址或目的地覆盖参数,代码也无法读取或替换目标。
后端
runtime.backend 在线程的沙箱中读写文件。参见从代码读写沙箱文件。