以编程方式使用文档

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

RemoteGraph 适用于以下场景：

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

前提条件

在开始使用 RemoteGraph之前，请确保您拥有：

  • - 访问 LangSmith，这是您的图开发和管理的平台。
  • - 一个运行中的 Agent Server，用于托管已部署的图以进行远程交互。

初始化图

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

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

初始化。此外，您必须提供以下之一：

  • - url：要与之交互的部署的 URL。如果你传入 url 参数，sync 和 async 客户端都将使用提供的 URL、headers（如果提供）和默认配置值（例如 timeout）来创建。
  • - client: A LangGraphClient 实例用于异步交互部署（例如使用 .astream(), .ainvoke(), .aget_state(), .aupdate_state()).
  • - sync_client: A SyncLangGraphClient 实例用于同步交互部署（例如使用 .stream(), .invoke(), .get_state(), .update_state()).

使用 URL

from langgraph.pregel.remote import RemoteGraph

url = ""

# Using graph name (uses default assistant)
graph_name = "agent"
remote_graph = RemoteGraph(graph_name, url=url)

# Using assistant ID
assistant_id = ""
remote_graph = RemoteGraph(assistant_id, url=url)
const url = "";

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

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

使用客户端

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

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 = ""
remote_graph = RemoteGraph(assistant_id, client=client, sync_client=sync_client)
const client = new Client({ apiUrl: "" });

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

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

调用图

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

异步

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

同步

# 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() or .stream()进行的调用）是无状态的，这意味着运行后不会持久化中间检查点和最终状态。

如果你想保留运行的输出（例如支持人机协作工作流），可以创建一个线程并通过 config 参数传递其 ID。这与普通编译图的工作方式相同：

from langgraph_sdk import get_sync_client

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)
const 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 实例作为 子图 节点。这允许创建模块化、可扩展的工作流，其中不同的职责被拆分到单独的图中。

RemoteGraph 公开了与普通 CompiledGraph相同的接口，因此你可以直接将其作为子图用在另一个图中。例如：

from langgraph_sdk import get_sync_client
from langgraph.graph import StateGraph, MessagesState, START
from typing import TypedDict

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