您可能需要为新的运行使用不同配置重建您的图。例如,您可能想根据用户凭证加载不同的工具。本指南展示了如何使用 ServerRuntime.
前提条件
- - 请务必查看 本操作指南 关于设置应用程序以进行部署的内容。
- -
ServerRuntime需要langgraph-api >= 0.7.31和langgraph-sdk >= 0.3.5。在此之前,图工厂只接受单个config: RunnableConfigargument.
定义图
假设您有一个简单的图应用程序,该图调用 LLM 并将响应返回给用户。应用程序文件目录如下:
my-app/
|-- langgraph.json
|-- my_project/
| |-- __init__.py
| |-- agents.py # code for your graph
|-- pyproject.toml
其中图定义在 agents.py.
不重建
部署代理服务器最常见的方式是引用在文件顶层定义的已编译图实例。示例如下:
# 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()
为了让服务器识别您的图,您需要指定包含 CompiledStateGraph 实例的变量路径到您的 LangGraph API 配置(langgraph.json), e.g.:
{
"$schema": "https://langgra.ph/schema.json",
"dependencies": ["."],
"graphs": {
"chat_agent": "my_project.agents:agent",
}
}
重建
若要每次新运行时重建图,请提供一个 **工厂函数** 返回(或生成)图。工厂可选择接受 ServerRuntime 参数或 RunnableConfig。服务器检查函数的类型注解以确定要注入哪些参数,因此请确保包含正确的类型提示。服务器的队列工作线程在需要处理运行时随时调用工厂函数。该函数还会在某些其他端点被调用,用于更新状态、读取状态或获取助手架构。 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.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 检查图是被调用执行还是仅被用于自省(架构、可视化):
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", or "assistants.read". |
user | `BaseUser \ | None` |
store | BaseStore | 用于持久化和内存的存储实例。 |
Methods:
| 方法 | 说明 |
|---|---|
ensure_user() | 返回已认证的用户。如果没有提供用户则抛出 PermissionError 错误。 |
execution_runtime | 返回执行运行时,当 access_context is "threads.create_run", or 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 配置文件.