RemoteGraph 是一个客户端接口，允许您像与本地图一样与您的部署交互。它提供了与CompiledGraph相同的 API 对等性，这意味着您可以在开发和生产环境中使用相同的方法（invoke()、stream()、get_state()等）。本页面描述了如何初始化 RemoteGraph 并与之交互。 RemoteGraph 对于以下方面很有用：
- 开发与部署分离：使用
CompiledGraph 在本地构建和测试图，将其部署到 LangSmith，然后使用 RemoteGraph 在生产环境中调用它，同时使用相同的 API 接口。
- 线程级别持久化：通过线程 ID 在多次调用中持久化并获取对话状态。
- 子图嵌入：通过将
RemoteGraph 作为子图嵌入到另一个图中，为多代理工作流构建模块化图。
- 可重用工作流：将已部署的图用作节点或工具，从而可以重用和公开复杂的逻辑。
重要提示：避免调用相同的部署RemoteGraph 旨在调用其他部署上的图。请勿使用 RemoteGraph 调用自身或同一部署上的另一个图，因为这可能导致死锁和资源耗尽。相反，对于同一部署中的图，请使用本地图组合或子图。 先决条件
在使用 RemoteGraph 之前，请确保您拥有
初始化图
初始化 RemoteGraph 时，您必须始终指定
name：您要与之交互的图的名称或助手 ID。如果您指定图名称，将使用默认助手。如果您指定助手 ID，将使用该特定助手。图名称与您在部署的 langgraph.json 配置文件中使用的名称相同。api_key：有效的 LangSmith API 密钥。您可以将其设置为环境变量（LANGSMITH_API_KEY）或直接通过 api_key 参数传递。如果 LangGraphClient / SyncLangGraphClient 使用 api_key 参数初始化，您也可以在 client / sync_client 参数中提供 API 密钥。
此外，您必须提供以下之一
url：您要与之交互的部署的 URL。如果您传递 url 参数，则同步和异步客户端都将使用提供的 URL、标头（如果提供）和默认配置值（例如，超时）创建。client：一个 LangGraphClient 实例，用于异步与部署交互（例如，使用 .astream()、.ainvoke()、.aget_state()、.aupdate_state()）。sync_client：一个 SyncLangGraphClient 实例，用于同步与部署交互（例如，使用 .stream()、.invoke()、.get_state()、.update_state()）。
如果您同时传递 client 或 sync_client 以及 url 参数，它们将优先于 url 参数。如果未提供 client / sync_client / url 参数，RemoteGraph 将在运行时引发 ValueError。
使用 URL
from langgraph.pregel.remote import RemoteGraph
url = "<DEPLOYMENT_URL>"
# Using graph name (uses default assistant)
graph_name = "agent"
remote_graph = RemoteGraph(graph_name, url=url)
# Using assistant ID
assistant_id = "<ASSISTANT_ID>"
remote_graph = RemoteGraph(assistant_id, url=url)
使用客户端
from langgraph_sdk import get_client, get_sync_client
from langgraph.pregel.remote import RemoteGraph
url = "<DEPLOYMENT_URL>"
client = get_client(url=url)
sync_client = get_sync_client(url=url)
# Using graph name (uses default assistant)
graph_name = "agent"
remote_graph = RemoteGraph(graph_name, client=client, sync_client=sync_client)
# Using assistant ID
assistant_id = "<ASSISTANT_ID>"
remote_graph = RemoteGraph(assistant_id, client=client, sync_client=sync_client)
调用图
RemoteGraph 实现了与 CompiledGraph 相同的 Runnable 接口，因此您可以像使用编译图一样使用它。它支持全套标准方法，包括 .invoke()、.stream()、.get_state() 和 .update_state()，以及它们的异步变体。
要异步使用图，您必须在初始化 RemoteGraph 时提供 url 或 client。
# invoke the graph
result = await remote_graph.ainvoke({
"messages": [{"role": "user", "content": "what's the weather in sf"}]
})
# stream outputs from the graph
async for chunk in remote_graph.astream({
"messages": [{"role": "user", "content": "what's the weather in la"}]
}):
print(chunk)
要同步使用图，您必须在初始化 RemoteGraph 时提供 url 或 sync_client。
# invoke the graph
result = remote_graph.invoke({
"messages": [{"role": "user", "content": "what's the weather in sf"}]
})
# stream outputs from the graph
for chunk in remote_graph.stream({
"messages": [{"role": "user", "content": "what's the weather in la"}]
}):
print(chunk)
在线程级别持久化状态
默认情况下，图运行（例如，使用 .invoke() 或 .stream() 进行的调用）是无状态的，这意味着中间检查点和最终状态在运行后不会持久化。 如果您想保留运行的输出——例如，为了支持人机协作工作流——您可以创建一个线程并通过 config 参数传递其 ID。这与常规编译图的工作方式相同：from langgraph_sdk import get_sync_client
url = "<DEPLOYMENT_URL>"
graph_name = "agent"
sync_client = get_sync_client(url=url)
remote_graph = RemoteGraph(graph_name, url=url)
# create a thread (or use an existing thread instead)
thread = sync_client.threads.create()
# invoke the graph with the thread config
config = {"configurable": {"thread_id": thread["thread_id"]}}
result = remote_graph.invoke({
"messages": [{"role": "user", "content": "what's the weather in sf"}]
}, config=config)
# verify that the state was persisted to the thread
thread_state = remote_graph.get_state(config)
print(thread_state)
用作子图
如果需要将 checkpointer 与包含 RemoteGraph 子图节点的图一起使用，请确保使用 UUID 作为线程 ID。
RemoteGraph 实例作为子图节点调用。这允许模块化、可扩展的工作流，其中不同的职责分配给不同的图。 RemoteGraph 公开了与常规 CompiledGraph 相同的接口，因此您可以直接将其用作另一个图中的子图。例如：from langgraph_sdk import get_sync_client
from langgraph.graph import StateGraph, MessagesState, START
from typing import TypedDict
url = "<DEPLOYMENT_URL>"
graph_name = "agent"
remote_graph = RemoteGraph(graph_name, url=url)
# define parent graph
builder = StateGraph(MessagesState)
# add remote graph directly as a node
builder.add_node("child", remote_graph)
builder.add_edge(START, "child")
graph = builder.compile()
# invoke the parent graph
result = graph.invoke({
"messages": [{"role": "user", "content": "what's the weather in sf"}]
})
print(result)
# stream outputs from both the parent graph and subgraph
for chunk in graph.stream({
"messages": [{"role": "user", "content": "what's the weather in sf"}]
}, subgraphs=True):
print(chunk)
