Graph API 概览 - LangChain 文档

图

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

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

通过组合 Nodes 和 Edges，你可以创建复杂的循环工作流，随着时间的推移不断演化状态。然而，真正的力量来自于 LangGraph 管理状态的方式。强调一下：Nodes 和 Edges 只不过是函数——它们可以包含大语言模型（LLM），也可以只是普通的程序代码。简而言之：节点负责干活，边负责告诉下一步干什么。 LangGraph 的底层图算法使用消息传递来定义通用程序。当一个节点完成其操作时，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将生成的消息传递给下一组节点，以此类推。受到 Google 的 Pregel 系统的启发，该程序以离散的“超步 (super-steps)”进行。一个超步可以被视为对图节点的一次迭代。并行运行的节点属于同一个超步，而顺序运行的节点属于不同的超步。在图执行开始时，所有节点最初都处于 inactive（空闲）状态。当一个节点在其任何入边（或“通道”）上收到新消息（状态）时，它会变为 active（激活）状态。激活的节点随后运行其函数并响应更新。在每个超步结束时，没有收到消息的节点通过将自己标记为 inactive 来投票决定 halt（停止）。当所有节点都处于 inactive 且没有消息正在传输时，图执行终止。

StateGraph

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

编译你的图

要构建你的图，首先定义状态 (state)，然后添加节点 (nodes)和边 (edges)，最后进行编译。究竟什么是“编译你的图”，为什么需要它？编译是一个非常简单的步骤。它对图的结构进行一些基本检查（确保没有孤立节点等）。这也是你可以指定运行时参数（如检查点记录器 checkpointers 和断点）的地方。你只需调用 .compile 方法即可编译图：

graph = graph_builder.compile(...)

在使用图之前，你必须先对其进行编译。

状态

定义图时做的第一件事就是定义图的 State。State 由图的模式 (schema) 以及指定如何将更新应用于状态的 reducer 函数组成。State 的模式将作为图中所有 Nodes 和 Edges 的输入模式，可以是一个 TypedDict 或 Pydantic 模型。所有 Nodes 都会发出对 State 的更新，然后使用指定的 reducer 函数应用这些更新。

Schema

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

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

多模式 (Multiple schemas)

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

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

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

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:
    # Write to OverallState
    return {"foo": state["user_input"] + " name"}

def node_2(state: OverallState) -> PrivateState:
    # Read from OverallState, write to PrivateState
    return {"bar": state["foo"] + " is"}

def node_3(state: PrivateState) -> OutputState:
    # Read from PrivateState, write to 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 的输入模式。但是，我们向 OverallState 中的通道 foo 写入数据。我们如何能向一个不包含在输入模式中的状态通道写入数据呢？这是因为节点可以写入图状态中的任何状态通道。图状态是初始化时定义的状态通道的并集，包括 OverallState 以及过滤器 InputState 和 OutputState。
我们使用以下方式初始化图
```
StateGraph(
    OverallState,
    input_schema=InputState,
    output_schema=OutputState
)
```
我们如何在 node_2 中向 PrivateState 写入数据？如果 PrivateState 在 StateGraph 初始化时没有传入，图是如何获得对该模式的访问权限的？我们可以这样做是因为只要状态模式定义存在，_nodes 就可以声明额外的状态 channels_（通道）。在这个例子中，PrivateState 模式已经定义，所以我们可以将 bar 作为图中一个新的状态通道添加并向其写入。

Reducers

Reducer 是理解节点更新如何应用于 State 的关键。State 中的每个键都有自己独立的 reducer 函数。如果未显式指定 reducer 函数，则默认对该键的所有更新都将执行覆盖（override）操作。有几种不同类型的 reducer，从默认类型开始

默认 reducer

这两个例子展示了如何使用默认 reducer

示例 A

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"]}

示例 B

from typing import Annotated
from typing_extensions import TypedDict
from operator import add

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

在此示例中，我们使用 Annotated 类型为第二个键 (bar) 指定了 reducer 函数 (operator.add)。请注意，第一个键保持不变。假设图的输入是 {"foo": 1, "bar": ["hi"]}。接着假设第一个 Node 返回 {"foo": 2}。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则 State 将变为 {"foo": 2, "bar": ["hi", "bye"]}。请注意，这里通过将两个列表相加来更新 bar 键。

重写 (Overwrite)

在某些情况下，你可能希望绕过 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 之外，每当在 messages 通道上收到状态更新时，add_messages 函数还会尝试将消息反序列化为 LangChain Message 对象。更多信息请参见 LangChain 序列化/反序列化。这允许以以下格式发送图输入/状态更新：

# this is supported
{"messages": [HumanMessage(content="message")]}

# and this is also supported
{"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)
# You can then create edges to/from this node by referencing it as `"my_node"`

`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 用于根据节点输入生成缓存键，默认情况下是对输入使用 pickle 进行 hash。
- 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]:
    # expensive computation
    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}}]

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

边 (Edges)

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

普通边 (Normal Edges)：直接从一个节点进入下一个节点。
条件边 (Conditional Edges)：调用一个函数来确定接下来去哪个（些）节点。
入口点 (Entry Point)：用户输入到达时首先调用的节点。
条件入口点 (Conditional Entry Point)：调用一个函数来确定用户输入到达时首先调用哪个（些）节点。

一个节点可以有多个出边。如果一个节点有多个出边，那么所有这些目标节点都将作为下一个超步的一部分并行执行。

对于每个节点，选择一种路由机制：使用普通边进行静态路由，或使用条件边 / 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 是一个用于控制图执行的多功能原语。它接受四个参数：

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(
        # state update
        update={"foo": "bar"},
        # control flow
        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",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT
    )

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

这在实现多智能体接力 (handoffs) 时特别有用。详情请参阅导航到父图中的节点。

`invoke` 或 `stream` 的输入

Command(resume=...) 是唯一打算作为 invoke()/stream() 输入的 Command 模式。不要使用 Command(update=...) 作为输入来继续多轮对话——因为传递任何 Command 作为输入都会从最新的检查点（即最后运行的步骤，而不是 __start__）恢复，如果图已经运行结束，它看起来会卡住。要在现有线程上继续对话，请传递一个普通的输入字典：

# WRONG - graph resumes from the latest checkpoint
# (last step that ran), appears stuck
graph.invoke(Command(update={
    "messages": [{"role": "user", "content": "follow up"}]
}), config)

# CORRECT - plain dict restarts from __start__
graph.invoke( {
    "messages": [{"role": "user", "content": "follow up"}]
}, config)

`resume`

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

from langgraph.types import Command, interrupt

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

# First invocation - hits the interrupt and pauses
result = graph.invoke({"messages": [...]}, config)

# Resume with a value - the interrupt() call returns "yes"
result = graph.invoke(Command(resume="yes"), config)

有关中断模式（包括多个中断和验证循环）的完整详细信息，请查看中断概念指南。

从工具返回

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

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

详情请参阅在工具内部使用。

图迁移

即使在使用检查点记录器（checkpointer）跟踪状态时，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"] 中。递归限制检查遵循以下逻辑：step > stop，其中 stop = step + recursion_limit + 1。当超过限制时，LangGraph 会引发 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  # Managed value - tracks steps until limit

def reasoning_node(state: State) -> dict:
    # RemainingSteps is automatically populated by LangGraph
    remaining = state["remaining_steps"]

    # Check if we're running low on steps
    if remaining <= 2:
        return {"messages": ["Approaching limit, wrapping up..."]}

    # Normal processing
    return {"messages": ["thinking..."]}

def route_decision(state: State) -> Literal["reasoning_node", "fallback_node"]:
    """Route based on remaining steps"""
    if state["remaining_steps"] <= 2:
        return "fallback_node"
    return "reasoning_node"

def fallback_node(state: State) -> dict:
    """Handle cases where recursion limit is approaching"""
    return {"messages": ["Reached complexity limit, providing best effort answer"]}

# Build graph
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 works with any 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

# Proactive Approach (recommended) - using RemainingSteps
def agent_with_monitoring(state: State) -> dict:
    """Proactively monitor and handle recursion within the graph"""
    remaining = state["remaining_steps"]

    # Early detection - route to internal handling
    if remaining <= 2:
        return {
            "messages": ["Approaching limit, returning partial result"]
        }

    # Normal processing
    return {"messages": [f"Processing... ({remaining} steps remaining)"]}

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

# Build graph
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()

# Proactive: Graph completes gracefully
result = graph.invoke({"messages": []}, {"recursion_limit": 10})

# Reactive Approach (fallback) - catching error externally
try:
    result = graph.invoke({"messages": []}, {"recursion_limit": 10})
except GraphRecursionError as e:
    # Handle externally after graph execution fails
    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。

了解更多

将这些文档连接到 Claude、VSCode 等，以获得实时答案。

在 GitHub 上编辑此页或提交问题。

入门

能力

生产

前端

LangGraph API

文档索引

​图

​StateGraph

​编译你的图

​状态

​Schema

​多模式 (Multiple schemas)

​Reducers

​默认 reducer

​重写 (Overwrite)

​在图状态中使用消息

​为什么要使用消息？

​在你的图中使用消息

​序列化

​MessagesState

​节点 (Nodes)

​START 节点

​END 节点

​节点缓存

​边 (Edges)

​普通边

​条件边

​入口点

​条件入口点

​Send (发送)

​命令

​从节点返回

​update 和 goto

​graph

​invoke 或 stream 的输入

​resume

​从工具返回

​图迁移

​运行时上下文

​递归限制

​访问和处理递归计数器

​工作原理

​访问当前步数计数器

​主动递归处理

​主动与被动方法

​其他可用的元数据

​可视化

​可观测性与追踪

​了解更多

图

StateGraph

编译你的图

状态

Schema

多模式 (Multiple schemas)

Reducers

默认 reducer

重写 (Overwrite)

在图状态中使用消息

为什么要使用消息？

在你的图中使用消息

序列化

MessagesState

节点 (Nodes)

`START` 节点

`END` 节点

节点缓存

边 (Edges)

普通边

条件边

入口点

条件入口点

`Send (发送)`

`命令`

从节点返回

`update` 和 `goto`

`graph`

`invoke` 或 `stream` 的输入

`resume`

从工具返回

图迁移

运行时上下文

递归限制

访问和处理递归计数器

工作原理

访问当前步数计数器

主动递归处理

主动与被动方法

其他可用的元数据

可视化

可观测性与追踪

了解更多