在运行时重建图 - Docs by LangChain

你可能需要为新的运行使用不同配置重建图。例如，你可能希望根据用户的凭证加载不同的工具。本指南将展示如何使用 ServerRuntime 来实现这一点。

在大多数情况下，自定义最好通过在各个节点内根据配置进行条件判断来处理，而不是动态改变整个图结构。这使得测试和管理更加容易。

前置条件

请确保首先查看此操作指南以设置你的应用进行部署。
ServerRuntime 需要 langgraph-api >= 0.7.31 和 langgraph-sdk >= 0.3.5。在此之前，图工厂函数只接受单个 config: RunnableConfig 参数。

定义图

假设你有一个应用，其中包含一个简单的图，该图调用一个 LLM 并将响应返回给用户。应用文件目录结构如下：

my-app/
|-- langgraph.json
|-- my_project/
|   |-- __init__.py
|   |-- agents.py     # 你的图代码
|-- pyproject.toml

其中图定义在 agents.py 中。

不重建

部署你的 Agent Server 最常见的方法是引用一个在文件顶层定义的已编译图实例。示例如下：

# my_project/agents.py
from langgraph.graph import StateGraph, MessagesState, START

async def model(state: MessagesState):
    return {"messages": [{"role": "assistant", "content": "Hi, there!"}]}

graph_workflow = StateGraph(MessagesState)
graph_workflow.add_node("model", model)
graph_workflow.add_edge(START, "model")
agent = graph_workflow.compile()

要让服务器知道你的图，你需要在 LangGraph API 配置（langgraph.json）中指定包含 CompiledStateGraph 实例的变量路径，例如：

{
    "$schema": "https://langgra.ph/schema.json",
    "dependencies": ["."],
    "graphs": {
        "chat_agent": "my_project.agents:agent",
    }
}

重建

要在每次新运行时重建你的图，请提供一个返回（或产出）图的工厂函数。该工厂函数可以选择性地接受一个 ServerRuntime 参数或一个 RunnableConfig。服务器会检查你函数的类型注解以确定注入哪些参数，因此请确保包含正确的类型提示。服务器的队列工作进程在需要处理运行时会调用你的工厂函数。该函数也会在某些其他端点被调用，以更新状态、读取状态或获取助手模式。ServerRuntime 会告诉你哪个上下文触发了调用。

ServerRuntime 处于测试阶段，未来版本可能会更改。

简单工厂函数

最简单的形式是一个返回已编译图的普通 async def 函数：

from langchain_openai import ChatOpenAI
from langgraph.graph import START, StateGraph
from langchain_core.runnables import RunnableConfig
from langgraph_sdk.runtime import ServerRuntime

from my_agent.utils.state import AgentState

model = ChatOpenAI(model="gpt-5.4")


def make_graph_for_user(user_id: str):
    """为每个用户构建一个定制的图。"""
    graph_workflow = StateGraph(AgentState)

    async def call_model(state):
        return {"messages": [await model.ainvoke(state["messages"])]}

    graph_workflow.add_node("agent", call_model)
    graph_workflow.add_edge(START, "agent")
    return graph_workflow.compile()


async def make_graph(config: RunnableConfig, runtime: ServerRuntime):
    user = runtime.ensure_user()
    return make_graph_for_user(user.identity)

上下文管理器工厂函数

如果你需要设置和清理资源（数据库连接、加载 MCP 工具等），请使用异步上下文管理器。使用 runtime.execution_runtime 来检查图是被调用用于实际执行还是仅用于内省（模式、可视化）：

import contextlib

from langchain_openai import ChatOpenAI
from langgraph.graph import START, StateGraph
from langchain_core.runnables import RunnableConfig
from langgraph_sdk.runtime import ServerRuntime

from my_agent.utils.state import AgentState

model = ChatOpenAI(model="gpt-5.4")


def make_agent_graph(tools: list):
    """制作一个简单的 LLM 代理。"""
    graph_workflow = StateGraph(AgentState)
    bound = model.bind_tools(tools)

    async def call_model(state):
        return {"messages": [await bound.ainvoke(state["messages"])]}

    graph_workflow.add_node("agent", call_model)
    graph_workflow.add_edge(START, "agent")
    return graph_workflow.compile()


@contextlib.asynccontextmanager
async def make_graph(runtime: ServerRuntime):
    if ert := runtime.execution_runtime:
        # 仅在实际执行期间设置昂贵的资源。
        # 内省调用（get_schema, get_graph, ...）会跳过此步骤。
        mcp_tools = await connect_mcp(ert.ensure_user())  # 你的设置逻辑
        yield make_agent_graph(tools=mcp_tools)
        await disconnect_mcp()  # 你的清理逻辑
    else:
        # 对于模式/状态读取，返回一个具有相同拓扑结构
        # 但不设置昂贵资源的图。
        yield make_agent_graph(tools=[])

最后，在 langgraph.json 中指定你的工厂函数路径：

{
    "$schema": "https://langgra.ph/schema.json",
    "dependencies": ["."],
    "graphs": {
        "chat_agent": "my_project.agents:make_graph",
    }
}

ServerRuntime 参考

你的工厂函数接收一个具有以下属性的 ServerRuntime 实例：

属性	类型	描述
`access_context`	`str`	工厂函数被调用的原因：`"threads.create_run"`、`"threads.update"`、`"threads.read"` 或 `"assistants.read"`。
`user`	`BaseUser \| None`	已认证的用户，如果未配置自定义认证则为 `None`。
`store`	`BaseStore`	用于持久化和记忆的存储实例。

方法：

方法	描述
`ensure_user()`	返回已认证的用户。如果未提供用户，则引发 `PermissionError`。
`execution_runtime`	当 `access_context` 为 `"threads.create_run"` 时返回执行运行时，否则返回 `None`。使用此方法仅在执行期间有条件地设置昂贵资源。

访问上下文

服务器不仅在执行运行时调用你的工厂函数，还在其他几种上下文中调用。在所有上下文中，返回的图应具有相同的拓扑结构（节点、边、状态模式）。在写入上下文（threads.create_run、threads.update）中拓扑结构不匹配可能导致状态更新不正确。在读取上下文（threads.read、assistants.read）中，不匹配会影响报告的待处理任务、模式和可视化，但不会损坏数据。使用 execution_runtime 有条件地设置昂贵资源，而不改变图结构。

上下文	描述
`threads.create_run`	完整的图执行。`execution_runtime` 可用。
`threads.update`	通过 `aupdate_state` 进行状态更新。不执行节点函数，但可以更改待处理任务。
`threads.read`	通过 `aget_state` / `aget_state_history` 读取状态。
`assistants.read`	用于可视化、MCP、A2A 等的模式和图内省。

为每个图自定义跟踪

你可以使用工厂函数来为特定图自定义或禁用跟踪。有关示例，请参阅条件跟踪：在部署的代理中自定义跟踪。有关更多信息，请参阅 LangGraph API 配置文件。

将这些文档通过 MCP 连接到 Claude、VSCode 等，以获取实时答案。

在 GitHub 上编辑此页面或提交问题。

​前置条件

​定义图

​不重建

​重建

​简单工厂函数

​上下文管理器工厂函数

​ServerRuntime 参考

​访问上下文

​为每个图自定义跟踪

前置条件

定义图

不重建

重建

简单工厂函数

上下文管理器工厂函数

ServerRuntime 参考

访问上下文

为每个图自定义跟踪