在运行时重建图

使用 ServerRuntime 为每次运行以不同配置重建图。

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

NOTE

在大多数情况下，最好通过对单个节点内的配置进行条件判断来处理自定义，而不是动态更改整个图结构。这样更易于测试和管理。

前置条件

• 请务必先查看这篇关于为部署设置应用的操作指南。
• ServerRuntime 需要 langgraph-api >= 0.7.31 和 langgraph-sdk >= 0.3.5。在此之前，图工厂只接受单个 config: RunnableConfig 参数。

定义图

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

my-app/
|-- langgraph.json
|-- my_project/
|   |-- __init__.py
|   |-- agents.py     # code for your graph
|-- 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。服务器会检查函数的类型注解以确定要注入哪些参数，因此请确保包含正确的类型提示。服务器的队列工作进程会在需要处理运行时调用你的工厂函数。对于某些其他端点（更新状态、读取状态或获取 assistant schema），该函数也会被调用。ServerRuntime 会告诉你哪个上下文触发了调用。

NOTE

ServerRuntime 处于 beta 阶段，未来版本可能会发生变化。

简单工厂

最简单的形式是一个返回已编译图的普通 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.5")

def make_graph_for_user(user_id: str):
    """Build a graph customized per user."""
    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 检查图是正在被调用以进行实际执行，还是仅用于内省（schema、可视化）：

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.5")

def make_agent_graph(tools: list):
    """Make a simple LLM agent."""
    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:
        # Only set up expensive resources during actual execution.
        # Introspection calls (get_schema, get_graph, ...) skip this.
        mcp_tools = await connect_mcp(ert.ensure_user())  # your setup logic
        yield make_agent_graph(tools=mcp_tools)
        await disconnect_mcp()  # your teardown logic
    else:
        # For schema/state reads, return a graph with the same
        # topology but no expensive resource setup.
        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	用于持久化和内存的 store 实例。

方法：

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

访问上下文

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

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

为每个图自定义追踪

你可以使用工厂函数为特定图自定义或禁用追踪。有关示例，请参阅条件追踪：在已部署的 agent 中自定义追踪。

请参阅 LangGraph API 配置文件的更多信息。

---