【LangGraph】Use RemoteGraph：如何使用 RemoteGraph 与平台部署交互 示例

作者：彬彬侠
如何使用 RemoteGraph 与部署交互（How to interact with the deployment using RemoteGraph）

预备知识

LangGraph PlatformLangGraph Server

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

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

name：要交互的图的名称。这与你在部署的 langgraph.json 配置文件中使用的图名称相同。api_key：有效的 LangSmith API 密钥。可以通过环境变量 LANGSMITH_API_KEY 设置，或直接通过 api_key 参数传递。如果通过 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 参数，client 或 sync_client 将优先于 url 参数。如果未提供 client、sync_client 或 url 参数，RemoteGraph 将在运行时抛出 ValueError。
使用 URL

• python
  1. from langgraph.pregel.remote import RemoteGraph
  3. url =<DEPLOYMENT_URL>
  4. graph_name ="agent"
  5. remote_graph = RemoteGraph(graph_name, url=url)

  复制代码
  javascript
  1. import{ RemoteGraph }from"@langchain/langgraph/remote";const url =`<DEPLOYMENT_URL>`;const graphName ="agent";const remoteGraph =newRemoteGraph({graphId: graphName, url });

  复制代码

使用客户端

• python
  1. from langgraph_sdk import get_client, get_sync_client
  2. from langgraph.pregel.remote import RemoteGraph
  4. url =<DEPLOYMENT_URL>
  5. graph_name ="agent"
  6. client = get_client(url=url)
  7. sync_client = get_sync_client(url=url)
  8. remote_graph = RemoteGraph(graph_name, client=client, sync_client=sync_client)

  复制代码
  javascript
  1. import{ Client }from"@langchain/langgraph-sdk";import{ RemoteGraph }from"@langchain/langgraph/remote";const client =newClient({apiUrl:`<DEPLOYMENT_URL>`});const graphName ="agent";const remoteGraph =newRemoteGraph({graphId: graphName, client });

  复制代码

调用图

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

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

• python
  1. # 调用图
  2. result =await remote_graph.ainvoke({"messages":[{"role":"user","content":"旧金山的天气如何？"}]})# 从图中流式传输输出asyncfor chunk in remote_graph.astream({"messages":[{"role":"user","content":"洛杉矶的天气如何？"}]}):print(chunk)

  复制代码
• javascript
  1. // 调用图const result =await remoteGraph.invoke({messages:[{role:"user",content:"旧金山的天气如何？"}]})// 从图中流式传输输出forawait(const chunk ofawait remoteGraph.stream({messages:[{role:"user",content:"洛杉矶的天气如何？"}]})):
  2.     console.log(chunk)

  复制代码

同步调用

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

1. # 调用图
2. result = remote_graph.invoke({"messages":[{"role":"user","content":"旧金山的天气如何？"}]})# 从图中流式传输输出for chunk in remote_graph.stream({"messages":[{"role":"user","content":"洛杉矶的天气如何？"}]}):print(chunk)

复制代码

线程级持久化

默认情况下，图的运行（即 .invoke() 或 .stream() 调用）是无状态的——检查点和图的最终状态不会持久化。如果你希望持久化图运行的输出（例如，支持人机交互功能），可以创建一个线程并通过 config 参数提供线程 ID，就像操作常规已编译图一样：

• python
  1. from langgraph_sdk import get_sync_client
  2. url =<DEPLOYMENT_URL>
  3. graph_name ="agent"
  4. sync_client = get_sync_client(url=url)
  5. remote_graph = RemoteGraph(graph_name, url=url)# 创建一个线程（或使用现有线程）
  6. thread = sync_client.threads.create()# 使用线程配置调用图
  7. config ={"configurable":{"thread_id": thread["thread_id"]}}
  8. result = remote_graph.invoke({"messages":[{"role":"user","content":"旧金山的天气如何？"}]}, config=config)# 验证状态是否已持久化到线程
  9. thread_state = remote_graph.get_state(config)print(thread_state)

  复制代码
• javascript
  1. import{ Client }from"@langchain/langgraph-sdk";import{ RemoteGraph }from"@langchain/langgraph/remote";const url =`<DEPLOYMENT_URL>`;const graphName ="agent";const client =newClient({apiUrl: url });const remoteGraph =newRemoteGraph({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);
  2. console.log(threadState);

  复制代码

作为子图使用

注意
如果需要在使用 RemoteGraph 子图节点的图中使用检查点器，请确保使用 UUID 作为线程 ID。
由于 RemoteGraph 的行为与常规 CompiledGraph 相同，它也可以作为另一个图的子图使用。例如：

• python
  1. from langgraph_sdk import get_sync_client
  2. from langgraph.graph import StateGraph, MessagesState, START
  3. from typing import TypedDict
  5. url =<DEPLOYMENT_URL>
  6. graph_name ="agent"
  7. remote_graph = RemoteGraph(graph_name, url=url)# 定义父图
  8. builder = StateGraph(MessagesState)# 直接将远程图添加为节点
  9. builder.add_node("child", remote_graph)
  10. builder.add_edge(START,"child")
  11. graph = builder.compile()# 调用父图
  12. result = graph.invoke({"messages":[{"role":"user","content":"旧金山的天气如何？"}]})print(result)# 从父图和子图流式传输输出for chunk in graph.stream({"messages":[{"role":"user","content":"旧金山的天气如何？"}]}, subgraphs=True):print(chunk)

  复制代码
• javascript
  1. import{ MessagesAnnotation, StateGraph,START}from"@langchain/langgraph";import{ RemoteGraph }from"@langchain/langgraph/remote";const url =`<DEPLOYMENT_URL>`;const graphName ="agent";const remoteGraph =newRemoteGraph({graphId: graphName, url });// 定义父图并直接将远程图添加为节点const graph =newStateGraph(MessagesAnnotation).addNode("child", remoteGraph).addEdge(START,"child").compile()// 调用父图const result =await graph.invoke({messages:[{role:"user",content:"旧金山的天气如何？"}]});
  2. console.log(result);// 从父图和子图流式传输输出forawait(const chunk ofawait graph.stream({messages:[{role:"user",content:"洛杉矶的天气如何？"}]},{subgraphs:true})){
  3.   console.log(chunk);}

  复制代码

---

总结

RemoteGraph 机制：本指南展示了如何使用 RemoteGraph 接口与 LangGraph 平台部署交互，使其像本地定义的 CompiledGraph 一样，通过同步或异步方法（如 .invoke()、.stream() 等）调用远程图。

• 核心步骤：
  
  
  
  • 初始化：
    
    必须提供 name（图名称，与 langgraph.json 中的一致）和 api_key（通过 LANGSMITH_API_KEY 或参数）。至少提供 url（部署 URL）、client（异步 LangGraphClient）或 sync_client（同步 SyncLangGraphClient）之一。如果同时提供 client 或 sync_client 和 url，前者优先。
    
    
    
  • 调用：
    
    异步调用（.ainvoke()、.astream() 等）需要 url 或 client。同步调用（.invoke()、.stream() 等）需要 url 或 sync_client。支持标准 Runnable 方法，与本地图一致。
    
    
    
  • 线程持久化：
    
    默认无状态运行，可通过 config 指定 thread_id 持久化状态。使用 sync_client.threads.create() 创建线程，并通过 get_state() 验证状态持久化。
    
    
    
  • 子图使用：
    
    RemoteGraph 可作为子图节点嵌入父图，需使用 UUID 作为线程 ID 以支持检查点器。支持父图和子图的流式输出（通过 subgraphs=True）。
    
    
    
  
  
• 使用建议：
  
  配置：在 .env 中设置 LANGSMITH_API_KEY，或通过 api_key 参数传递，避免硬编码。线程 ID：使用 UUID 作为线程 ID，确保检查点器兼容性，尤其在子图场景中。调试：结合 LangSmith 跟踪运行，分析远程图的调用和状态。生产部署：确保 DEPLOYMENT_URL 指向正确的 LangGraph 平台部署，验证 API 密钥和网络可达性。跨语言支持：Python 和 JavaScript 示例展示了相似的 API，适合不同技术栈。
  
  
  
• 注意事项：
  
  未提供 url、client 或 sync_client 将导致运行时 ValueError。异步和同步方法需要相应的客户端支持（client 或 sync_client）。确保 langgraph.json 中的图名称与 RemoteGraph 的 name 参数一致。如果需要更复杂的交互（如动态配置或多线程管理），可扩展 config 或结合 LangGraph 平台功能。
  
  
  

---

参考资料：

https://langchain-ai.github.io/langgraph/how-tos/use-remote-graph/

原文地址：https://blog.csdn.net/u013172930/article/details/148140797