图 API 概述

图

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

1. 状态：一个共享数据结构，表示应用程序的当前快照。它可以是任何数据类型，但通常使用共享的状态模式来定义。

2. 节点：编码智能体逻辑的函数。它们接收当前状态作为输入，执行某些计算或副作用，并返回更新后的状态。

3. 边：根据当前状态决定下一步执行哪个节点的函数。它们可以是条件分支或固定转换。

通过组合节点和边，你可以创建复杂的、可循环的工作流，使状态随时间演化。但真正强大之处在于 LangGraph 如何管理该状态。

需要强调的是：节点和边不过是函数——它们可以包含大语言模型(LLM)或普通代码。

简而言之：节点执行工作，边决定接下来做什么。

LangGraph 的底层图算法使用消息传递来定义通用程序。当一个节点完成操作时，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将生成的消息传递给下一组节点，整个过程如此循环。受到 Google Pregel 系统的启发，程序以离散的”超级步”方式运行。

超级步可以被理解为对图节点的一次迭代。并行运行的节点属于同一个超级步，而顺序运行的节点属于不同的超级步。在图执行开始时，所有节点都处于 inactive 状态。当节点在其任何传入边（或”通道”）上接收到新消息（状态）时，它变为 active 状态。活跃节点随后运行其函数并以更新作为响应。在每个超级步结束时，没有传入消息的节点通过将自己标记为 inactive 来投票 halt。当所有节点都处于 inactive 状态且没有消息在传输中时，图执行终止。

StateGraph

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

编译你的图

要构建图，你首先定义状态，然后添加节点和边，最后编译它。编译图究竟是什么，为什么需要它？

编译是一个相当简单的步骤。它对图的结构进行一些基本检查（如没有孤立节点等）。编译也是你可以指定运行时参数（如检查点器和断点）的地方。你只需调用 .compile 方法即可编译图：

graph = graph_builder.compile(...)

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

状态

定义图时首先要做的是定义图的状态。状态由图的模式以及指定如何将更新应用于状态的reducer函数组成。模式将作为图中所有节点和边的输入模式，可以是 TypedDict 或 Pydantic 模型。所有节点将发出对状态的更新，这些更新随后使用指定的 reducer 函数应用。

模式

指定图模式的主要文档化方式是使用 TypedDict。如果你想在状态中提供默认值，请使用 dataclass。如果你需要递归数据验证，我们也支持使用 Pydantic BaseModel 作为图状态（但请注意 Pydantic 的性能不如 TypedDict 或 dataclass）。

默认情况下，图的输入和输出模式相同。如果你想更改此设置，也可以直接指定显式的输入和输出模式。当你有很多键，其中一些明确用于输入，另一些用于输出时，这很有用。更多信息请参阅指南。

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

多重模式

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

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

节点可以写入图内部的私有状态通道进行内部节点通信。我们可以简单地定义一个私有模式 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:
    # 写入 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'}

这里有两个微妙但重要的要点：

1. 我们将 state: InputState 作为输入模式传递给 node_1。但我们写入了 foo，这是 OverallState 中的一个通道。我们如何能写入一个不包含在输入模式中的状态通道？这是因为节点_可以写入图状态中的任何状态通道_。图状态是在初始化时定义的状态通道的联合，包括 OverallState 以及过滤器 InputState 和 OutputState。

2. 我们用以下方式初始化图：

   StateGraph(
       OverallState,
       input_schema=InputState,
       output_schema=OutputState
   )
   

   我们如何能在 node_2 中写入 PrivateState？如果它没有在 StateGraph 初始化中传入，图如何访问这个模式？

   我们可以这样做，因为节点也可以声明额外的状态通道，只要状态模式定义存在。在这种情况下，PrivateState 模式已定义，因此我们可以将 bar 添加为图中的新状态通道并写入它。

Reducer

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

默认 reducer

以下两个示例展示了如何使用默认 reducer：

Example A

from typing_extensions import TypedDict

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

在此示例中，没有为任何键指定 reducer 函数。假设图的输入为：

{"foo": 1, "bar": ["hi"]}。然后假设第一个节点返回 {"foo": 2}。这被视为对状态的更新。请注意，节点不需要返回完整的状态模式——只需返回更新即可。应用此更新后，状态将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则状态将变为 {"foo": 2, "bar": ["bye"]}。

Example 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"]}。然后假设第一个节点返回 {"foo": 2}。这被视为对状态的更新。请注意，节点不需要返回完整的状态模式——只需返回更新即可。应用此更新后，状态将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则状态将变为 {"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 之外，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]

节点

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

1. state——图的状态
2. config——一个 RunnableConfig 对象，包含配置信息（如 thread_id）和追踪信息（如 tags）
3. 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"` 来创建到/从此节点的边

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]:
    # 耗时计算
    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}}]

1. 第一次运行需要两秒（由于模拟的耗时计算）。
2. 第二次运行利用缓存并快速返回。

边

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

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

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

对于每个节点，选择一种路由机制：使用普通边进行静态路由，或使用条件边/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 接受图的当前状态并返回一个值。

默认情况下，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

默认情况下，节点和边是预先定义的，并在相同的共享状态上运行。然而，在某些情况下，确切的边可能事先未知，和/或你可能希望不同版本的状态同时存在。一个常见的例子是 map-reduce 设计模式。在此设计模式中，第一个节点可能生成一个对象列表，你可能希望将某个其他节点应用于所有这些对象。对象的数量可能事先未知（意味着边的数量可能未知），并且下游节点的输入状态应该不同（每个生成的对象一个）。

为了支持此设计模式，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 模式。不要使用 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 langgraph.types import Command, interrupt

def human_review(state: State):
    # 暂停图并等待一个值
    answer = interrupt("Do you approve?")
    return {"messages": [{"role": "user", "content": answer}]}

# 第一次调用 - 命中中断并暂停
result = graph.invoke({"messages": [...]}, config)

# 使用一个值恢复 - interrupt() 调用返回 "yes"
result = graph.invoke(Command(resume="yes"), config)

查看中断概念指南了解中断模式的完整详情，包括多重中断和验证循环。

从工具返回

你可以从工具返回 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 步。递归限制可以在运行时对任何图设置，并通过 config 字典传递给 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  # 管理值 - 跟踪直到限制的步数

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。

了解更多

• 如何使用图 API
• 函数式 API 概念概述
• 选择图 API 还是函数式 API

---

通过 MCP 连接这些文档到 Claude、VSCode 等工具，获取实时答案。

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