Skip to content

如何使用RemoteGraph与部署进行交互

先决条件

RemoteGraph 是一个接口，允许您像操作本地定义的 LangGraph 图（例如 CompiledGraph）一样与您的 LangGraph 平台部署进行交互。本指南将向您展示如何初始化一个 RemoteGraph 并与其进行交互。

初始化图

当初始化一个 RemoteGraph 时，您必须始终指定：

  • name：您要与其交互的图的名称。这是您在部署的 langgraph.json 配置文件中使用的相同图名称。
  • api_key：有效的 LangSmith API 密钥。可以设置为环境变量（LANGSMITH_API_KEY）或直接通过 api_key 参数传递。API 密钥也可以通过 client / sync_client 参数提供，如果 LangGraphClient / SyncLangGraphClient 是使用 api_key 参数初始化的。

此外，您必须提供以下之一：

  • 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>
graph_name = "agent"
remote_graph = RemoteGraph(graph_name, url=url)

import { RemoteGraph } from "@langchain/langgraph/remote";

const url = `<DEPLOYMENT_URL>`;
const graphName = "agent";
const remoteGraph = new RemoteGraph({ graphId: graphName, url });

使用客户端

from langgraph_sdk import get_client, get_sync_client
from langgraph.pregel.remote import RemoteGraph

url = <DEPLOYMENT_URL>
graph_name = "agent"
client = get_client(url=url)
sync_client = get_sync_client(url=url)
remote_graph = RemoteGraph(graph_name, client=client, sync_client=sync_client)

import { Client } from "@langchain/langgraph-sdk";
import { RemoteGraph } from "@langchain/langgraph/remote";

const client = new Client({ apiUrl: `<DEPLOYMENT_URL>` });
const graphName = "agent";
const remoteGraph = new RemoteGraph({ graphId: graphName, client });

调用图

由于 RemoteGraph 是一个实现了与 CompiledGraph 相同方法的 Runnable，因此你可以像调用编译后的图一样来调用它，即通过调用 .invoke()、.stream()、.get_state()、.update_state() 等方法（以及它们的异步版本）。

异步调用

注意

要异步使用图，初始化 RemoteGraph 时必须提供 url 或 client。

# 调用图
result = await remote_graph.ainvoke({
    "messages": [{"role": "user", "content": "旧金山的天气怎么样"}]
})

# 从图中流式传输输出
async for chunk in remote_graph.astream({
    "messages": [{"role": "user", "content": "洛杉矶的天气怎么样"}]
}):
    print(chunk)

// 调用图
const result = await remoteGraph.invoke({
    messages: [{role: "user", content: "旧金山的天气怎么样"}]
})

// 从图中流式传输输出
for await (const chunk of await remoteGraph.stream({
    messages: [{role: "user", content: "洛杉矶的天气怎么样"}]
})):
    console.log(chunk)

同步调用

注意

要同步使用图，初始化 RemoteGraph 时必须提供 url 或 sync_client。

# 调用图
result = remote_graph.invoke({
    "messages": [{"role": "user", "content": "旧金山的天气怎么样"}]
})

# 从图中流式传输输出
for chunk in remote_graph.stream({
    "messages": [{"role": "user", "content": "洛杉矶的天气怎么样"}]
}):
    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": "旧金山的天气怎么样"}]
}, 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: "旧金山的天气怎么样" }],
}, config);

// 验证状态是否已持久化到线程中
const threadState = await remoteGraph.getState(config);
console.log(threadState);

作为子图使用

Note

如果你需要在一个包含 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);
}

Comments