RemoteGraph 是一个客户端接口,允许您与部署交互,就像它是本地图一样。它提供与 CompiledGraph 的 API 对等性,这意味着您可以在开发和生产环境中使用相同的方法(invoke()stream()get_state() 等)。本页描述如何初始化 RemoteGraph 并与其交互。 RemoteGraph 对以下情况很有用:
  • 开发和部署的分离:使用 CompiledGraph 在本地构建和测试图,将其部署到 LangSmith,然后使用 RemoteGraph在生产中调用它,同时使用相同的 API 接口。
  • 线程级持久化:使用线程 ID 持久化和获取对话状态
  • 子图嵌入:通过将 RemoteGraph 作为子图嵌入另一个图中,为多智能体工作流程组合模块化图。
  • 可重用工作流程:将部署的图用作节点或工具,以便您可以重用和公开复杂的逻辑。
重要:避免调用相同的部署RemoteGraph 旨在调用其他部署上的图。不要使用 RemoteGraph 调用自身或同一部署上的另一个图,因为这可能导致死锁和资源耗尽。相反,对同一部署中的图使用本地图组合或子图

Prerequisites

Before getting started with RemoteGraph, make sure you have:
  • Access to LangSmith, where your graphs are developed and managed.
  • A running Agent Server, which hosts your deployed graphs for remote interaction.

Initialize the graph

When initializing a RemoteGraph, you must always specify:
  • name: The name of the graph you want to interact with or an assistant ID. If you specify a graph name, the default assistant will be used. If you specify an assistant ID, that specific assistant will be used. The graph name is the same name you use in the langgraph.json configuration file for your deployment.
  • api_key: A valid LangSmith API key. You can set as an environment variable (LANGSMITH_API_KEY) or pass directly in the api_key argument. You can also provide the API key in the client / sync_client arguments, if LangGraphClient / SyncLangGraphClient was initialized with the api_key argument.
Additionally, you have to provide one of the following:
  • url: The URL of the deployment you want to interact with. If you pass the url argument, both sync and async clients will be created using the provided URL, headers (if provided), and default configuration values (e.g., timeout).
  • client: A LangGraphClient instance for interacting with the deployment asynchronously (e.g., using .astream(), .ainvoke(), .aget_state(), .aupdate_state()).
  • sync_client: A SyncLangGraphClient instance for interacting with the deployment synchronously (e.g., using .stream(), .invoke(), .get_state(), .update_state()).
If you pass both client or sync_client as well as the url argument, they will take precedence over the url argument. If none of the client / sync_client / url arguments are provided, RemoteGraph will raise a ValueError at runtime.

Use a URL

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)

Use a client

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)

Invoke the graph

RemoteGraph implements the same Runnable interface as CompiledGraph, so you can use it in the same way as a compiled graph. It supports the full set of standard methods, including .invoke(), .stream(), .get_state(), and .update_state(), as well as their asynchronous variants.

Asynchronously

To use the graph asynchronously, you must provide either the url or client when initializing the RemoteGraph.
# 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)

Synchronously

To use the graph synchronously, you must provide either the url or sync_client when initializing the RemoteGraph.
# 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)

Persist state at the thread level

By default, graph runs (for example, calls made with .invoke() or .stream()) are stateless, which means that intermediate checkpoints and the final state are not persisted after a run. If you want to preserve the outputs of a run—for example, to support human-in-the-loop workflows—you can create a thread and pass its ID through the config argument. This works the same way as with a regular compiled graph:
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)

Use as a subgraph

If you need to use a checkpointer with a graph that has a RemoteGraph subgraph node, make sure to use UUIDs as thread IDs.
A graph can also call out to multiple RemoteGraph instances as subgraph nodes. This allows for modular, scalable workflows where different responsibilities are split across separate graphs. RemoteGraph exposes the same interface as a regular CompiledGraph, so you can use it directly as a subgraph inside another graph. For example:
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)

Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.