Graph API 概述 - Docs by LangChain中文

图

LangGraph 的核心是将智能体工作流建模为图。您使用三个关键组件来定义智能体的行为：

State：一个共享的数据结构，代表应用程序的当前快照。它可以是任何数据类型，但通常使用共享状态模式来定义。
Nodes：编码智能体逻辑的函数。它们接收当前状态作为输入，执行一些计算或副作用，并返回一个更新的状态。
Edges：根据当前状态决定接下来执行哪个 Node 的函数。它们可以是条件分支或固定转换。

通过组合 Nodes 和 Edges，您可以创建复杂的、循环的工作流，这些工作流会随时间推移演变状态。然而，真正的威力来自于 LangGraph 如何管理该状态。需要强调的是：Nodes 和 Edges 不过是函数——它们可以包含一个大语言模型（LLM），也可以只是普通的代码。简而言之：节点执行工作，边指示下一步做什么。 LangGraph 底层的图算法使用消息传递来定义一个通用程序。当一个节点完成其操作时，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行它们的函数，将结果消息传递给下一组节点，这个过程持续进行。该程序受到 Google 的 Pregel 系统的启发，以离散的”超步骤”进行。一个超步骤可以被视为对图节点的一次单次迭代。并行运行的节点属于同一个超步骤，而顺序运行的节点则属于不同的超步骤。在图执行开始时，所有节点都处于 inactive 状态。当一个节点在其任何传入边（或”通道”）上接收到新消息（状态）时，它就会变为 active 状态。然后，活动节点运行其函数并返回更新。在每个超步骤结束时，没有传入消息的节点通过将自己标记为 inactive 来投票”停止”。当所有节点都处于 inactive 状态且没有消息在传递时，图执行终止。

StateGraph

StateGraph 类是要使用的主要图类。它由一个用户定义的 State 对象参数化。

编译您的图

要构建您的图，您首先定义状态，然后添加节点和边，最后编译它。编译您的图具体是什么意思？为什么需要它？编译是一个相当简单的步骤。它对图的结构进行一些基本检查（例如，没有孤立的节点等）。这也是您可以指定运行时参数（如检查点器和断点）的地方。您只需调用 .compile 方法即可编译图：

graph = graph_builder.compile(...)

您必须先编译图，然后才能使用它。

State

定义图时，您要做的第一件事就是定义图的 State。State 由图的模式和 reducer 函数组成，后者指定了如何将更新应用到状态。State 的模式将是图中所有 Nodes 和 Edges 的输入模式，可以是 TypedDict 或 Pydantic 模型。所有 Nodes 将发出对 State 的更新，然后使用指定的 reducer 函数应用这些更新。

Schema

指定图模式的主要文档化方法是使用 TypedDict。如果您想在状态中提供默认值，请使用 dataclass。我们还支持使用 Pydantic BaseModel 作为图状态，如果您需要递归数据验证（但请注意，Pydantic 的性能不如 TypedDict 或 dataclass）。默认情况下，图将具有相同的输入和输出模式。如果您想更改这一点，也可以直接指定显式的输入和输出模式。当您有很多键，并且某些键明确用于输入，另一些用于输出时，这很有用。有关更多信息，请参阅指南。

langchain 中的高级 create_agent 工厂不支持 Pydantic 状态模式。

多个模式

通常，所有图节点都使用单个模式进行通信。这意味着它们将读取和写入相同的状态通道。但是，在某些情况下，我们希望对此有更多的控制：

内部节点可以传递在图的输入/输出中不需要的信息。
我们可能还希望为图使用不同的输入/输出模式。例如，输出可能只包含一个相关的输出键。

可以让节点在图中写入私有状态通道，用于内部节点通信。我们可以简单地定义一个私有模式 PrivateState。也可以为图定义显式的输入和输出模式。在这些情况下，我们定义一个包含与图操作相关的所有键的”内部”模式。但是，我们也定义 input 和 output 模式，它们是”内部”模式的子集，以约束图的输入和输出。有关更多详细信息，请参阅定义输入和输出模式。让我们看一个例子：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class InputState(TypedDict):
    user_input: str


class OutputState(TypedDict):
    graph_output: str


class OverallState(TypedDict):
    foo: str
    user_input: str
    graph_output: str


class PrivateState(TypedDict):
    bar: str


def node_1(state: InputState) -> OverallState:
    # 写入 OverallState
    return {"foo": state["user_input"] + " name"}


def node_2(state: OverallState) -> PrivateState:
    # 读取 OverallState，写入 PrivateState
    return {"bar": state["foo"] + " is"}


def node_3(state: PrivateState) -> OutputState:
    # 读取 PrivateState，写入 OutputState
    return {"graph_output": state["bar"] + " Lance"}


builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)

graph = builder.compile()
graph.invoke({"user_input": "My"})
# {'graph_output': 'My name is Lance'}

这里有两个微妙而重要的点需要注意：

我们将 state: InputState 作为输入模式传递给 node_1。但是，我们写入了 foo，这是 OverallState 中的一个通道。我们如何写入一个未包含在输入模式中的状态通道？这是因为节点_可以写入图状态中的任何状态通道。_图状态是在初始化时定义的状态通道的并集，其中包括 OverallState 以及过滤后的 InputState 和 OutputState。
我们使用以下方式初始化图：
StateGraph( OverallState, input_schema=InputState, output_schema=OutputState )
我们如何在 node_2 中写入 PrivateState？如果它没有在 StateGraph 初始化时传递，图是如何获得对这个模式的访问权限的？我们能做到这一点是因为 _nodes 还可以声明额外的状态 channels_，只要状态模式定义存在即可。在这种情况下，PrivateState 模式已经定义，因此我们可以将 bar 添加为图中的新状态通道并写入其中。

流式传输时私有通道不会被编辑。输入、输出和私有模式约束了每个节点_读取_的内容（其输入模式）以及 invoke _返回_的内容（输出模式）。它们不会对 stream 隐藏通道。当您使用 stream_mode="values" 进行流式传输时，图默认会发出其所有状态通道——包括私有通道——因为值流式传输默认使用完整状态通道集，而不是输出模式。这就是为什么像 bar 这样的私有通道被 invoke 隐藏，但在流式传输时却可见：

stream = graph.stream_events({"user_input": "My"}, version="v3")
for snapshot in stream.values:
    print(snapshot)
# {'user_input': 'My'}
# {'foo': 'My name', 'user_input': 'My'}
# {'foo': 'My name', 'user_input': 'My', 'bar': 'My name is'}        # <-- 私有通道
# {'foo': 'My name', 'user_input': 'My', 'graph_output': 'My name is Lance', 'bar': 'My name is'}

要将流式传输的值限制在特定的一组通道（例如仅输出模式），请传递 output_keys：

stream = graph.stream_events(
    {"user_input": "My"},
    version="v3",
    output_keys=["graph_output"],
)
for snapshot in stream.values:
    print(snapshot)
# {'graph_output': 'My name is Lance'}

如果您只需要节点每步实际产生的通道（而不是完整的累积状态），请改用 stream_mode="updates"。

Reducers

Reducers 是理解来自节点的更新如何应用到 State 的关键。State 中的每个键都有其独立的 reducer 函数。如果没有显式指定 reducer 函数，则假定对该键的所有更新都应覆盖它。有几种不同类型的 reducers，从默认类型的 reducer 开始：

Reducer 参数

每个 reducer 都是一个二元函数，有两个位置参数：

左参数：状态中为该键存储的当前值。
右参数：节点返回的对该键的更新。

当一个节点返回部分更新时，LangGraph 为每个更新的键调用 reducer，并将返回值保存为新的状态值：

new_value = reducer(left=current_state[key], right=node_update[key])

左参数总是来自累积的状态。右参数总是来自最新的节点更新。以下示例显式地命名了这两个参数：

from typing import Annotated

from typing_extensions import TypedDict

def append_strings(left: list[str], right: list[str]) -> list[str]:
    """将现有状态值（left）与节点更新（right）进行合并。"""
    return left + right

class State(TypedDict):
    tags: Annotated[list[str], append_strings]

假设状态是 {"tags": ["draft"]}，一个节点返回 {"tags": ["review"]}。LangGraph 调用：

append_strings(left=["draft"], right=["review"])  # 返回 ["draft", "review"]

tags 的新状态值是 ["draft", "review"]。自定义 reducers 组合左参数和右参数。默认 reducer 丢弃左参数，只保留右参数。

默认 reducer

默认 reducer 忽略左参数，并用右参数替换状态值。此示例展示了如何使用默认 reducer：

from typing_extensions import TypedDict


class State(TypedDict):
    foo: int
    bar: list[str]

在此示例中，没有为任何键指定 reducer 函数。假设图的输入是： {"foo": 1, "bar": ["hi"]}。然后假设第一个 Node 返回 {"foo": 2}。这被视为对状态的更新。请注意，Node 不需要返回整个 State 模式——只需返回一个更新。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，那么 State 将变为 {"foo": 2, "bar": ["bye"]}

自定义 reducers

自定义 reducer 组合左参数和右参数，而不是替换状态值，这对于累积值（例如将更新追加到列表）很有用。此示例展示了如何指定自定义 reducer：

from operator import add
from typing import Annotated

from typing_extensions import TypedDict


class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]

在此示例中，我们使用 Annotated 类型为第二个键（bar）指定了一个 reducer 函数（operator.add）。请注意，第一个键保持不变。假设图的输入是 {"foo": 1, "bar": ["hi"]}。然后假设第一个 Node 返回 {"foo": 2}。这被视为对状态的更新。请注意，Node 不需要返回整个 State 模式——只需返回一个更新。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，那么 State 将变为 {"foo": 2, "bar": ["hi", "bye"]}。请注意，这里的 bar 键是通过将两个列表相加来更新的。

覆盖

在某些情况下，您可能希望绕过 reducer 并直接覆盖状态值。LangGraph 为此目的提供了 Overwrite 类型。此处了解如何使用 Overwrite。

在图状态中处理消息

为什么使用消息？

大多数现代 LLM 提供商都有一个聊天模型接口，该接口接受一个消息列表作为输入。特别是 LangChain 的聊天模型接口接受一个消息对象列表作为输入。这些消息以多种形式出现，例如 HumanMessage（用户输入）或 AIMessage（LLM 响应）。要了解更多关于消息对象的信息，请参阅消息概念指南。

在图中使用消息

在许多情况下，将先前的对话历史作为消息列表存储在图状态中是有帮助的。为此，我们可以向图状态添加一个键（通道），该键存储 Message 对象的列表，并用一个 reducer 函数对其进行注解（参见下面示例中的 messages 键）。reducer 函数对于告诉图如何在每次状态更新（例如，当一个节点发送更新时）时更新状态中的 Message 对象列表至关重要。如果您不指定 reducer，每次状态更新都会用最近提供的值覆盖消息列表。如果您只想将消息追加到现有列表，可以使用 operator.add 作为 reducer。但是，您可能也希望手动更新图状态中的消息（例如人在回路中）。如果您使用 operator.add，您发送给图的手动状态更新将被追加到现有消息列表，而不是更新现有消息。为避免这种情况，您需要一个能够跟踪消息 ID 并在更新时覆盖现有消息的 reducer。为此，您可以使用预构建的 add_messages 函数。对于全新的消息，它将简单地追加到现有列表，但它也会正确处理现有消息的更新。

序列化

除了跟踪消息 ID 之外，add_messages 函数还会尝试在 messages 通道上接收到状态更新时，将消息反序列化为 LangChain Message 对象。有关更多信息，请参阅 LangChain 序列化/反序列化。这允许以下列格式发送图输入/状态更新：

# 这是支持的
{"messages": [HumanMessage(content="message")]}

# 这也是支持的
{"messages": [{"type": "human", "content": "message"}]}

由于使用 add_messages 时，状态更新总是被反序列化为 LangChain Messages，您应该使用点表示法来访问消息属性，例如 state["messages"][-1].content。下面是使用 add_messages 作为其 reducer 函数的图的示例。

from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

class GraphState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

MessagesState

由于在状态中拥有一个消息列表非常常见，因此存在一个名为 MessagesState 的预构建状态，可以轻松使用消息。MessagesState 定义了一个单独的 messages 键，它是一个 AnyMessage 对象的列表，并使用 add_messages reducer。通常，除了消息之外，还有更多状态需要跟踪，所以我们看到人们子类化这个状态并添加更多字段，例如：

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

Nodes

在 LangGraph 中，节点是 Python 函数（同步或异步），接受以下参数：

state—图的状态
config—一个 RunnableConfig 对象，包含配置信息如 thread_id 和跟踪信息如 tags
runtime—一个 Runtime 对象，包含运行时 context 和其他信息，如 store、stream_writer、execution_info、server_info、heartbeat（用于空闲超时刷新）和 control（用于优雅关闭）

类似于 NetworkX，您使用 add_node 方法将这些节点添加到图中：

from dataclasses import dataclass
from typing_extensions import TypedDict

from langgraph.graph import StateGraph
from langgraph.runtime import Runtime

class State(TypedDict):
    input: str
    results: str

@dataclass
class Context:
    user_id: str

builder = StateGraph(State)

def plain_node(state: State):
    return state

def node_with_runtime(state: State, runtime: Runtime[Context]):
    print("In node: ", runtime.context.user_id)
    return {"results": f"Hello, {state['input']}!"}

def node_with_execution_info(state: State, runtime: Runtime):
    print("In node with thread_id: ", runtime.execution_info.thread_id)
    return {"results": f"Hello, {state['input']}!"}


builder.add_node("plain_node", plain_node)
builder.add_node("node_with_runtime", node_with_runtime)
builder.add_node("node_with_execution_info", node_with_execution_info)
...

在后台，函数被转换为 RunnableLambda，它为您的函数添加了批处理和异步支持，以及原生跟踪和调试。如果您在添加节点到图时没有指定名称，它将被赋予一个与函数名称等价的默认名称。

builder.add_node(my_node)
# 然后您可以通过引用 `"my_node"` 来创建指向/来自此节点的边

重新执行和幂等性

当您使用检查点器进行编译时，LangGraph 在超步骤边界保存检查点，而不是在节点函数内部保存。如果执行停止并稍后恢复（例如在中断或重试之后），受影响的节点将从其函数的开头重新运行。暂停前的代码和副作用会再次运行。 幂等性。 设计节点逻辑时，确保重新执行不会破坏状态。如果一个节点插入了一行数据库记录，运行它两次不应该创建重复的行，除非这是有意为之的。使用幂等键、upsert 或读写检查。有关 interrupt() 周围的副作用，请参阅调用 interrupt 之前的副作用必须是幂等的。 图更改。 关于代码更改的确定性规则不适用于图结构。您可以添加或删除节点和边，而不会中断现有线程的恢复。恢复运行使用保存的状态，并执行您现在编译的任何图。 节点内的任务和中断。 如果一个节点调用任务或 interrupt，则在恢复时应用更严格的确定性规则。LangGraph 从检查点器恢复已完成的任务结果，但在恢复点之前更改代码中的任务或 interrupt 顺序可能导致缓存值不匹配。功能 API 入口点编译为单个节点，以这种方式运行整个入口点方法。请参阅确定性、幂等性以及在节点中使用任务。

在节点中使用任务

如果一个节点包含多个操作，您可能会发现将每个操作实现为一个任务比将逻辑分散到多个节点中更容易。当图使用检查点器时，任务结果会被检查点记录，因此恢复线程可以跳过节点内已完成的任务工作。

原始
带任务

from typing import NotRequired

import requests
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class State(TypedDict):
    url: str
    result: NotRequired[str]


def call_api(state: State):
    """Example node that makes an API request."""
    result = requests.get(state["url"]).text[:100]
    return {"result": result}


builder = StateGraph(State)
builder.add_node("call_api", call_api)
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

graph.invoke({"url": "https://www.example.com"}, config)

from typing import NotRequired

import requests
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import task
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class State(TypedDict):
    urls: list[str]
    results: NotRequired[list[str]]


@task
def _make_request(url: str):
    """Make a request."""
    return requests.get(url).text[:100]


def call_api(state: State):
    """Example node that makes API requests as checkpointed tasks."""
    futures = [_make_request(url) for url in state["urls"]]
    results = [f.result() for f in futures]
    return {"results": results}


builder = StateGraph(State)
builder.add_node("call_api", call_api)
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

graph.invoke({"urls": ["https://www.example.com"]}, config)

`START` 节点

START 节点是一个特殊的节点，表示将用户输入发送到图的节点。引用此节点的主要目的是确定首先应调用哪些节点。

from langgraph.graph import START

graph.add_edge(START, "node_a")

`END` 节点

END 节点是一个特殊的节点，表示终止节点。当您想表示哪些边在完成之后没有操作时，会引用此节点。

from langgraph.graph import END

graph.add_edge("node_a", END)

节点缓存

LangGraph 支持基于节点输入的缓存任务/节点。要使用缓存：

在编译图（或指定入口点）时指定缓存
为节点指定缓存策略。每个缓存策略支持：
- key_func，用于根据节点输入生成缓存键，默认为输入的 hash（使用 pickle）。
- ttl，缓存的生存时间（秒）。如果未指定，缓存将永不过期。

例如：

import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy


class State(TypedDict):
    x: int
    result: int


builder = StateGraph(State)


def expensive_node(state: State) -> dict[str, int]:
    # 昂贵的计算
    time.sleep(2)
    return {"result": state["x"] * 2}


builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=3))
builder.set_entry_point("expensive_node")
builder.set_finish_point("expensive_node")

graph = builder.compile(cache=InMemoryCache())

print(graph.invoke({"x": 5}, stream_mode='updates'))
# [{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates'))
# [{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]

set_entry_point(node) 定义了图将执行的第一个节点。它等价于 builder.add_edge(START, node)。set_finish_point(node) 定义了图中的最后一个节点。它等价于 builder.add_edge(node, END)。两种方法都有效，但 add_edge(START, ...) 和 add_edge(..., END) 是推荐的现代语法。

第一次运行需要两秒钟（因为模拟了昂贵的计算）。
第二次运行利用了缓存并快速返回。

Edges

边定义了逻辑如何路由以及图如何决定停止。这是您的智能体工作方式以及不同节点如何相互通信的重要组成部分。有几种关键类型的边：

普通边：直接从一个节点到下一个节点。
条件边：调用一个函数来决定接下来去哪个节点。
入口点：当用户输入到达时，首先调用哪个节点。
条件入口点：调用一个函数来决定当用户输入到达时首先调用哪个节点。

一个节点可以有多个传出边。如果一个节点有多个传出边，所有这些目标节点都将在下一个超步骤中并行执行。

对于每个节点，选择一种路由机制：对于静态路由使用普通边，或对于动态路由使用条件边 / Command。不要从同一个节点混合普通边和动态路由，因为两条路径都可能执行，并使图行为更难推理。

普通边

如果您总是想从节点 A 到节点 B，您可以直接使用 add_edge 方法。

graph.add_edge("node_a", "node_b")

条件边

如果您想选择性地路由到一个或多个边（或选择性地终止），您可以使用 add_conditional_edges 方法。此方法接受一个节点的名称和一个在该节点执行后调用的”路由函数”：

graph.add_conditional_edges("node_a", routing_function)

类似于节点，routing_function 接受图的当前 state 并返回一个值。默认情况下，routing_function 的返回值用作将状态发送到的下一个节点（或节点列表）的名称。所有这些节点都将在下一个超步骤中并行运行。您可以选择提供一个字典，将 routing_function 的输出映射到下一个节点的名称。

graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})

如果您想在单个函数中组合状态更新和路由，请使用 Command 而不是条件边。

入口点

入口点是图启动时运行的第一个节点。您可以使用 add_edge 方法从虚拟 START 节点指向要执行的第一个节点，以指定从哪里进入图。

from langgraph.graph import START

graph.add_edge(START, "node_a")

条件入口点

条件入口点允许您根据自定义逻辑从不同的节点开始。您可以使用来自虚拟 START 节点的 add_conditional_edges 来实现这一点。

from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

您可以通过提供一个字典来可选地将 routing_function 的输出映射到下一个节点的名称。

graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})

`Send`

默认情况下，Nodes 和 Edges 是预先定义好的，并在相同的共享状态上运行。然而，在某些情况下，确切的边可能不是预先知道的，或者您可能希望不同的 State 版本同时存在。一个常见的例子是 map-reduce 设计模式。在这种设计模式中，第一个节点可能生成一个对象列表，而您可能希望将其他一些节点应用于所有这些对象。对象的数量可能预先未知（意味着边的数量可能未知），并且下游 Node 的输入 State 应该是不同的（每个生成的对象对应一个）。为了支持这种设计模式，LangGraph 支持从条件边返回 Send 对象。Send 接受两个参数：第一个是节点的名称，第二个是传递给该节点的状态。

from langgraph.types import Send

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state['subjects']]

graph.add_conditional_edges("node_a", continue_to_jokes)

`Command`

Command 是一个用于控制图执行的多功能原语。它接受四个参数：

update：应用状态更新（类似于从节点返回更新）。
goto：导航到特定节点（类似于条件边）。
graph：在从子图导航时指定父图。
resume：提供一个值以在中断后恢复执行。

Command 在三种上下文中使用：

从节点返回：使用 update、goto 和 graph 将状态更新与控制流结合起来。
作为 invoke 或 stream 的输入：使用 resume 在中断后继续执行。
从工具返回：类似于从节点返回，在工具内部组合状态更新和控制流。

从节点返回

`update` 和 `goto`

从节点函数返回 Command 以在单个步骤中更新状态并路由到下一个节点：

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # 状态更新
        update={"foo": "bar"},
        # 控制流
        goto="my_other_node"
    )

使用 Command，您还可以实现动态控制流行为（与条件边相同）：

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    if state["foo"] == "bar":
        return Command(update={"foo": "baz"}, goto="my_other_node")

当您需要同时更新状态并路由到不同节点时，使用 Command。如果您只需要路由而不更新状态，请使用条件边。

当在您的节点函数中返回 Command 时，您必须添加返回类型注解，并列出该节点要路由到的节点名称列表，例如 Command[Literal["my_other_node"]]。这对于图渲染是必要的，并告知 LangGraph my_node 可以导航到 my_other_node。

Command 只会添加动态边——使用 add_edge / addEdge 定义的静态边仍然会执行。例如，如果 node_a 返回 Command(goto="my_other_node") 并且您还有 graph.add_edge("node_a", "node_b")，那么 node_b 和 my_other_node 都将运行。对于每个节点，请使用 Command 或静态边之一来路由到下一个节点，不要两者都用。

请查看此操作指南了解如何使用 Command 的端到端示例。

`graph`

如果您使用子图，您可以通过在 Command 中指定 graph=Command.PARENT，从子图中的节点导航到父图中的不同节点：

def my_node(state: State) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # 其中 `other_subgraph` 是父图中的一个节点
        graph=Command.PARENT
    )

将 graph 设置为 Command.PARENT 将导航到最近的父图。当您将更新从子图节点发送到父图节点以更新一个父图和子图状态模式共享的键时，您必须在父图状态中为您要更新的键定义一个 reducer。请参阅此示例。

这在实现多智能体交接时特别有用。详情请查看导航到父图中的节点。

作为 `invoke` 或 `stream` 的输入

Command(resume=...) 是唯一旨在作为 invoke()/stream() 输入的 Command 模式（可选地与 update=... 结合，以便在恢复时也应用状态更改）。不要单独使用 Command(update=...) 作为输入来继续多轮对话——因为将任何 Command 作为输入传递都会从最新的检查点恢复（即最后运行的步骤，而不是 __start__），如果图已经完成，它将看起来卡住了。要在现有线程上继续对话，请传递一个普通输入字典：

# 错误 - 图从最新的检查点恢复
# （最后运行的步骤），看起来卡住了
graph.invoke(Command(update={
    "messages": [{"role": "user", "content": "follow up"}]
}), config)

# 正确 - 普通字典从 __start__ 重新启动
graph.invoke( {
    "messages": [{"role": "user", "content": "follow up"}]
}, config)

`resume`

使用 Command(resume=...) 提供一个值并在中断后恢复图执行。传递给 resume 的值将成为暂停节点内部 interrupt() 调用的返回值：

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict):
    messages: list[dict]


def human_review(state: State):
    # Pauses the graph and waits for a value
    answer = interrupt("Do you approve?")
    return {"messages": [{"role": "user", "content": answer}]}


graph = (
    StateGraph(State)
    .add_node("human_review", human_review)
    .add_edge(START, "human_review")
    .add_edge("human_review", END)
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "graph-api-resume"}}

# First run - hits the interrupt and pauses
stream = graph.stream_events({"messages": []}, config, version="v3")
_ = stream.output  # drive the stream to completion
print(stream.interrupts)

# Resume with a value - the interrupt() call returns "yes"
resumed = graph.stream_events(Command(resume="yes"), config, version="v3")
final = resumed.output

有关中断模式的完整详情，请查看中断概念指南，包括多次中断和验证循环。

从工具返回

您可以从工具返回 Command 以更新图状态和控制流。使用 update 修改状态（例如，保存对话中查找的客户信息），使用 goto 在工具完成后路由到特定节点。

在工具内部使用时，goto 会添加一条动态边——在调用该工具的节点上已经定义的任何静态边仍然会执行。对于每个节点，请使用工具驱动的动态路由或静态边之一来路由到下一个节点，不要两者都用。

详情请参考在工具内使用。

图迁移

LangGraph 可以轻松处理图定义（节点、边和状态）的迁移，即使使用检查点跟踪状态时也是如此。

对于处于图末尾（即未中断）的线程，您可以更改图的整个拓扑结构（即所有节点和边，删除、添加、重命名等）
对于当前处于中断状态的线程，我们支持除重命名/删除节点以外的所有拓扑更改（因为该线程可能即将进入一个不再存在的节点）——如果这是阻碍因素，请联系我们，我们可以优先解决。
对于修改状态，我们在添加和删除键方面具有完全的向后和向前兼容性。
被重命名的状态键在现有线程中会丢失其保存的状态。
以不兼容方式更改类型的状态键目前可能会在具有更改前状态的线程中引起问题——如果这是阻碍因素，请联系我们，我们可以优先解决。

对于技术上兼容但会改变业务逻辑的更改，例如重写工具集或重组对话流程，请参阅业务兼容性。该页面介绍了如何在状态中固定行为版本，以便现有线程保留旧路径，而新线程采用最新版本。

运行时上下文

创建图时，您可以为传递给节点的运行时上下文指定 context_schema。这对于传递不属于图状态的信息给节点非常有用。例如，您可能希望传递诸如模型名称或数据库连接之类的依赖项。

@dataclass
class ContextSchema:
    llm_provider: str = "openai"

graph = StateGraph(State, context_schema=ContextSchema)

然后，您可以使用 invoke 方法的 context 参数将此上下文传递给图。

graph.invoke(inputs, context={"llm_provider": "anthropic"})

然后，您可以在节点或条件边内部访问和使用此上下文：

from langgraph.runtime import Runtime

def node_a(state: State, runtime: Runtime[ContextSchema]):
    llm = get_llm(runtime.context.llm_provider)
    # ...

有关配置的完整细分，请参阅添加运行时配置。

递归限制

递归限制设置了图在单次执行期间可以执行的最大超级步骤数。一旦达到限制，LangGraph 将抛出 GraphRecursionError。从 1.0.6 版本开始，默认的递归限制设置为 1000 步。递归限制可以在运行时在任何图上设置，并通过配置字典传递给 invoke/stream。重要的是，recursion_limit 是一个独立的 config 键，不应像所有其他用户定义的配置一样传递到 configurable 键内。请参阅下面的示例：

graph.invoke(inputs, config={"recursion_limit": 5}, context={"llm": "anthropic"})

阅读递归限制以了解有关递归限制工作原理的更多信息。

访问和处理递归计数器

当前的步骤计数器可以在任何节点内的 config["metadata"]["langgraph_step"] 中访问，从而可以在达到递归限制之前主动处理递归。这使您能够在图逻辑中实施优雅降级策略。

工作原理

步骤计数器存储在 config["metadata"]["langgraph_step"] 中。LangGraph 在图执行时递增此计数器，并在超过配置的 recursion_limit 时抛出 GraphRecursionError。

访问当前步骤计数器

您可以在任何节点内访问当前步骤计数器以监控执行进度。

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

def my_node(state: dict, config: RunnableConfig) -> dict:
    current_step = config["metadata"]["langgraph_step"]
    print(f"Currently on step: {current_step}")
    return state

主动递归处理

LangGraph 提供了一个 RemainingSteps 托管值，用于跟踪在达到递归限制之前剩余的步骤数。这允许在图内实现优雅降级。

from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps

class State(TypedDict):
    messages: Annotated[list, lambda x, y: x + y]
    remaining_steps: RemainingSteps  # 托管值 - 跟踪剩余步骤直到限制

def reasoning_node(state: State) -> dict:
    # RemainingSteps 由 LangGraph 自动填充
    remaining = state["remaining_steps"]

    # 检查我们是否接近限制
    if remaining <= 2:
        return {"messages": ["Approaching limit, wrapping up..."]}

    # 正常处理
    return {"messages": ["thinking..."]}

def route_decision(state: State) -> Literal["reasoning_node", "fallback_node"]:
    """根据剩余步骤进行路由"""
    if state["remaining_steps"] <= 2:
        return "fallback_node"
    return "reasoning_node"

def fallback_node(state: State) -> dict:
    """处理接近递归限制的情况"""
    return {"messages": ["Reached complexity limit, providing best effort answer"]}

# 构建图
builder = StateGraph(State)
builder.add_node("reasoning_node", reasoning_node)
builder.add_node("fallback_node", fallback_node)
builder.add_edge(START, "reasoning_node")
builder.add_conditional_edges("reasoning_node", route_decision)
builder.add_edge("fallback_node", END)

graph = builder.compile()

# RemainingSteps 适用于任何 recursion_limit
result = graph.invoke({"messages": []}, {"recursion_limit": 10})

主动方法与被动方法

处理递归限制有两种主要方法：主动方法（在图内监控）和被动方法（在外部捕获错误）。

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps
from langgraph.errors import GraphRecursionError

class State(TypedDict):
    messages: Annotated[list, lambda x, y: x + y]
    remaining_steps: RemainingSteps

# 主动方法（推荐） - 使用 RemainingSteps
def agent_with_monitoring(state: State) -> dict:
    """在图内主动监控和处理递归"""
    remaining = state["remaining_steps"]

    # 提前检测 - 路由到内部处理
    if remaining <= 2:
        return {
            "messages": ["Approaching limit, returning partial result"]
        }

    # 正常处理
    return {"messages": [f"Processing... ({remaining} steps remaining)"]}

def route_decision(state: State) -> Literal["agent", END]:
    if state["remaining_steps"] <= 2:
        return END
    return "agent"

# 构建图
builder = StateGraph(State)
builder.add_node("agent", agent_with_monitoring)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_decision)
graph = builder.compile()

# 主动方法：图优雅地完成
result = graph.invoke({"messages": []}, {"recursion_limit": 10})

# 被动方法（后备） - 在外部捕获错误
try:
    result = graph.invoke({"messages": []}, {"recursion_limit": 10})
except GraphRecursionError as e:
    # 图执行失败后在外部处理
    result = {"messages": ["Fallback: recursion limit exceeded"]}

这两种方法的主要区别在于：

方法	检测	处理	控制流
主动（使用 `RemainingSteps`）	达到限制之前	在图内通过条件路由	图继续运行到完成节点
被动（捕获 `GraphRecursionError`）	超过限制之后	在 try/catch 的图外部	图执行终止

主动方法的优势：

在图内优雅降级
可以在检查点中保存中间状态
提供部分结果，用户体验更好
图正常完成（无异常）

被动方法的优势：

实现更简单
无需修改图逻辑
集中式错误处理

其他可用的元数据

除了 langgraph_step，以下元数据也可以在 config["metadata"] 中使用：

def inspect_metadata(state: dict, config: RunnableConfig) -> dict:
    metadata = config["metadata"]

    print(f"Step: {metadata['langgraph_step']}")
    print(f"Node: {metadata['langgraph_node']}")
    print(f"Triggers: {metadata['langgraph_triggers']}")
    print(f"Path: {metadata['langgraph_path']}")
    print(f"Checkpoint NS: {metadata['langgraph_checkpoint_ns']}")

    return state

可视化

能够可视化图通常很好，尤其是当它们变得更加复杂时。LangGraph 附带了几种内置的图可视化方法。更多信息请参阅可视化您的图。

可观测性和追踪

要追踪、调试和评估您的智能体，请使用 LangSmith。

了解更多

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

Edit this page on GitHub or file an issue.

​图

​StateGraph

​编译您的图

​State

​Schema

​多个模式

​Reducers

​Reducer 参数

​默认 reducer

​自定义 reducers

​覆盖

​在图状态中处理消息

​为什么使用消息？

​在图中使用消息

​序列化

​MessagesState

​Nodes

​重新执行和幂等性

​在节点中使用任务

​START 节点

​END 节点

​节点缓存

​Edges

​普通边

​条件边

​入口点

​条件入口点

​Send

​Command

​从节点返回

​update 和 goto

​graph

​作为 invoke 或 stream 的输入

​resume

​从工具返回

​图迁移

​运行时上下文

​递归限制

​访问和处理递归计数器

​工作原理

​访问当前步骤计数器

​主动递归处理

​主动方法与被动方法

​其他可用的元数据

​可视化

​可观测性和追踪

​了解更多

图

StateGraph

编译您的图

State

Schema

多个模式

Reducers

Reducer 参数

默认 reducer

自定义 reducers

覆盖

在图状态中处理消息

为什么使用消息？

在图中使用消息

序列化

MessagesState

Nodes

重新执行和幂等性

在节点中使用任务

`START` 节点

`END` 节点

节点缓存

Edges

普通边

条件边

入口点

条件入口点

`Send`

`Command`

从节点返回

`update` 和 `goto`

`graph`

作为 `invoke` 或 `stream` 的输入

`resume`

从工具返回

图迁移

运行时上下文

递归限制

访问和处理递归计数器

工作原理

访问当前步骤计数器

主动递归处理

主动方法与被动方法

其他可用的元数据

可视化

可观测性和追踪

了解更多