如何使用 RemoteGraph 与部署交互

RemoteGraph 是一个客户端接口，允许你像使用本地图一样与部署交互。它提供与 CompiledGraph 的 API 对等性，这意味着你可以在开发和生产环境中使用相同的方法（invoke()、stream()、get_state() 等）。本页介绍如何初始化 RemoteGraph 并与之交互。

RemoteGraph 适用于以下场景：

• 开发与部署分离：在本地使用 CompiledGraph 构建和测试图，将其部署到 LangSmith，然后在生产环境中使用 RemoteGraph 调用它，同时使用相同的 API 接口。
• 线程级持久化：使用线程 ID 在多次调用之间持久化和获取对话状态。
• 子图嵌入：通过将 RemoteGraph 作为子图嵌入另一个图中，为多 agent 工作流组合模块化图。
• 可复用工作流：将部署的图用作节点或工具，以便复用和公开复杂逻辑。

WARNING

重要：避免调用同一个部署

RemoteGraph 旨在调用其他部署上的图。不要使用 RemoteGraph 调用自身或同一部署上的另一个图，因为这可能导致死锁和资源耗尽。对于同一部署内的图，请改用本地图组合或子图。

前置条件

在开始使用 RemoteGraph 之前，请确保你具备：

• 对 LangSmith 的访问权限，你的图在那里开发和管理。
• 一个正在运行的 Agent Server，它托管你部署的图以进行远程交互。

初始化图

初始化 RemoteGraph 时，你必须始终指定：

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

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

• url：要交互的部署 URL。如果传递 url 参数，将使用提供的 URL、header（如果提供）和默认配置值（例如超时）创建同步和异步客户端。
• client：用于异步与部署交互的 LangGraphClient 实例（例如使用 .astream()、.ainvoke()、.aget_state()、.aupdate_state()）。
• sync_client：用于同步与部署交互的 SyncLangGraphClient 实例（例如使用 .stream()、.invoke()、.get_state()、.update_state()）。

NOTE

如果你同时传递 client 或 sync_client 以及 url 参数，它们将优先于 url 参数。如果未提供 client / sync_client / url 中的任何一个，RemoteGraph 将在运行时引发 ValueError。

使用 URL

Python

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)

JavaScript

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

const url = "<DEPLOYMENT_URL>";

// Using graph name (uses default assistant)
const graphName = "agent";
const remoteGraph = new RemoteGraph({ graphId: graphName, url });

// Using assistant ID
const assistantId = "<ASSISTANT_ID>";
const remoteGraph = new RemoteGraph({ graphId: assistantId, url });

使用客户端

Python

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)

JavaScript

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

const client = new Client({ apiUrl: "<DEPLOYMENT_URL>" });

// Using graph name (uses default assistant)
const graphName = "agent";
const remoteGraph = new RemoteGraph({ graphId: graphName, client });

// Using assistant ID
const assistantId = "<ASSISTANT_ID>";
const remoteGraph = new RemoteGraph({ graphId: assistantId, client });

调用图

RemoteGraph 实现了与 CompiledGraph 相同的 Runnable 接口，因此你可以像使用编译图一样使用它。它支持全套标准方法，包括 .invoke()、.stream()、.get_state() 和 .update_state()，以及它们的异步变体。

异步

NOTE

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

Python

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

JavaScript

// 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)

同步

NOTE

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

Python

# 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。这与常规编译图的工作方式相同：

Python

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)

JavaScript

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

用作子图

NOTE

如果你需要将 checkpointer 与具有 RemoteGraph 子图节点的图一起使用，请确保使用 UUID 作为线程 ID。

一个图还可以作为子图节点调用多个 RemoteGraph 实例。这允许实现模块化、可扩展的工作流，将不同职责拆分到单独的图中。

RemoteGraph 公开与常规 CompiledGraph 相同的接口，因此你可以直接将其用作另一个图内的子图。例如：

Python

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)

JavaScript

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

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

---