如何使用 RemoteGraph 与部署进行交互¶
RemoteGraph 是一个接口，它允许你与 LangGraph 平台部署进行交互，就好像它是一个常规的、本地定义的 LangGraph 图（例如 CompiledGraph）一样。本指南将向你展示如何初始化 RemoteGraph 并与之交互。
初始化图¶
初始化 RemoteGraph 时，你必须始终指定：
- name：你想要与之交互的图的名称。这与你在 langgraph.json 配置文件中用于部署的图名称相同。
- api_key：有效的 LangSmith API 密钥。可以将其设置为环境变量（LANGSMITH_API_KEY），也可以通过 api_key 参数直接传递。如果 LangGraphClient / SyncLangGraphClient 是使用 api_key 参数初始化的，那么 API 密钥也可以通过 client / sync_client 参数提供。
此外，你还必须提供以下其中一项：
- 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¶
使用客户端¶
调用图¶
由于 RemoteGraph 是一个 Runnable，它实现了与 CompiledGraph 相同的方法，因此你可以像通常与已编译图进行交互那样与它进行交互，即通过调用 .invoke()、.stream()、.get_state()、.update_state() 等方法（以及它们的异步对应方法）。
异步调用¶
注意
若要异步使用该图，你必须在初始化 RemoteGraph 时提供 url 或 client。
同步调用¶
注意
若要同步使用该图，你必须在初始化 RemoteGraph 时提供 url 或 sync_client。
线程级持久化¶
默认情况下，图运行（即 .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)
import { Client } from "@langchain/langgraph-sdk";
import { RemoteGraph } from "@langchain/langgraph/remote";
const url = `<DEPLOYMENT_URL>`;
const graphName = "agent";
const client = new Client({ apiUrl: url });
const remoteGraph = new RemoteGraph({ graphId: graphName, url });
// 创建一个线程（或者使用现有的线程）
const thread = await client.threads.create();
// 使用线程配置调用图
const config = { configurable: { thread_id: thread.thread_id }};
const result = await remoteGraph.invoke({
messages: [{ role: "user", content: "what's the weather in sf" }],
}, config);
// 验证状态是否已持久化到线程
const threadState = await remoteGraph.getState(config);
console.log(threadState);
作为子图使用¶
注意
如果你需要在包含 RemoteGraph 子图节点的图中使用 checkpointer，请确保使用 UUID 作为线程 ID。
由于 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)
import { MessagesAnnotation, StateGraph, START } from "@langchain/langgraph";
import { RemoteGraph } from "@langchain/langgraph/remote";
const url = `<DEPLOYMENT_URL>`;
const graphName = "agent";
const remoteGraph = new RemoteGraph({ graphId: graphName, url });
// 定义父图并直接将远程图作为节点添加
const graph = new StateGraph(MessagesAnnotation)
.addNode("child", remoteGraph)
.addEdge(START, "child")
.compile()
// 调用父图
const result = await graph.invoke({
messages: [{ role: "user", content: "what's the weather in sf" }]
});
console.log(result);
// 流式输出父图和子图的结果
for await (const chunk of await graph.stream({
messages: [{ role: "user", content: "what's the weather in la" }]
}, { subgraphs: true })) {
console.log(chunk);
}