Skip to content
LangSmith 文档版本 跟随上游 源快照 e0051c9c 更新于 2026-10-19

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 在线程的沙箱中读写文件。参见从代码读写沙箱文件。

另请参阅


BytesRouter 是社区翻译站点,所有译文仅供参考,一切以官方英文原文为准。