跳到内容

如何使用 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() 等)。

注意

如果您同时传递了 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)
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 时必须提供 urlclient

# 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": [("user", "what's the weather in la?")]
}):
    print(chunk)
// invoke the graph
const result = await remoteGraph.invoke({
    messages: [{role: "user", content: "what's the weather in sf"}]
})

// stream outputs from the graph
for await (const chunk of await remoteGraph.stream({
    messages: [{role: "user", content: "what's the weather in la"}]
})):
    console.log(chunk)

同步

注意

要同步使用图,初始化 RemoteGraph 时必须提供 urlsync_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": [("user", "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)
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 });

// create a thread (or use an existing thread instead)
const thread = await client.threads.create();

// invoke the graph with the thread config
const config = { configurable: { thread_id: thread.thread_id }};
const result = await remoteGraph.invoke({
  messages: [{ role: "user", content: "what's the weather in sf" }],
  config
});

// verify that the state was persisted to the thread
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)

# 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)
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 });

// define parent graph and add remote graph directly as a node
const graph = new StateGraph(MessagesAnnotation)
  .addNode("child", remoteGraph)
  .addEdge("START", "child")
  .compile()

// invoke the parent graph
const result = await graph.invoke({
  messages: [{ role: "user", content: "what's the weather in sf" }]
});
console.log(result);