Skip to main content
RemoteGraph 是一个客户端接口,允许你像与本地图一样与你的部署进行交互。它提供了与 CompiledGraph 相同的 API 接口,这意味着你可以在开发和生产环境中使用相同的方法(invoke()stream()get_state() 等)。本页介绍了如何初始化 RemoteGraph 并与之交互。 RemoteGraph 适用于以下场景:
  • 开发与部署分离:使用 CompiledGraph 在本地构建和测试图,将其部署到 LangSmith,然后在生产环境中使用 RemoteGraph 调用它,同时保持相同的 API 接口。
  • 线程级持久化:通过线程 ID 跨调用持久化和获取对话状态。
  • 子图嵌入:通过将 RemoteGraph 作为子图嵌入到另一个图中,为多智能体工作流构建模块化图。
  • 可重用工作流:将已部署的图用作节点或 工具,以便重用和暴露复杂逻辑。
重要:避免调用同一部署RemoteGraph 设计用于调用其他部署上的图。请勿使用 RemoteGraph 调用自身或同一部署中的另一个图,因为这可能导致死锁和资源耗尽。对于同一部署内的图,应使用本地图组合或子图

前提条件

开始使用 RemoteGraph 前,请确保你拥有:
  • LangSmith 的访问权限,你的图在此处开发和管理。
  • 一个正在运行的 Agent Server,它托管你的已部署图以供远程交互。

初始化图

初始化 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())。
如果同时传入 clientsync_client 以及 url 参数,它们将优先于 url 参数。如果未提供 client / sync_client / url 参数中的任何一个,RemoteGraph 将在运行时引发 ValueError

使用 URL

from langgraph.pregel.remote import RemoteGraph

url = "<DEPLOYMENT_URL>"

# 使用图名称(使用默认助手)
graph_name = "agent"
remote_graph = RemoteGraph(graph_name, url=url)

# 使用助手 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)

# 使用图名称(使用默认助手)
graph_name = "agent"
remote_graph = RemoteGraph(graph_name, client=client, sync_client=sync_client)

# 使用助手 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 时必须提供 urlclient
# 调用图
result = await remote_graph.ainvoke({
    "messages": [{"role": "user", "content": "what's the weather in sf"}]
})

# 流式输出图的结果
async for chunk in remote_graph.astream({
    "messages": [{"role": "user", "content": "what's the weather in la"}]
}):
    print(chunk)

同步调用

要同步使用图,初始化 RemoteGraph 时必须提供 urlsync_client
# 调用图
result = remote_graph.invoke({
    "messages": [{"role": "user", "content": "what's the weather in sf"}]
})

# 流式输出图的结果
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)

# 创建线程(或使用现有线程)
thread = sync_client.threads.create()

# 使用线程配置调用图
config = {"configurable": {"thread_id": thread["thread_id"]}}
result = remote_graph.invoke({
    "messages": [{"role": "user", "content": "what's the weather in sf"}]
}, config=config)

# 验证状态是否已持久化到线程
thread_state = remote_graph.get_state(config)
print(thread_state)

用作子图

如果需要在包含 RemoteGraph 子图节点的图中使用 checkpointer,请确保使用 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)

# 定义父图
builder = StateGraph(MessagesState)
# 直接将远程图添加为节点
builder.add_node("child", remote_graph)
builder.add_edge(START, "child")
graph = builder.compile()

# 调用父图
result = graph.invoke({
    "messages": [{"role": "user", "content": "what's the weather in sf"}]
})
print(result)

# 流式输出父图和子图的结果
for chunk in graph.stream({
    "messages": [{"role": "user", "content": "what's the weather in sf"}]
}, subgraphs=True):
    print(chunk)