图

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

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

1. Nodes：对代理逻辑进行编码的函数。它们接收当前状态作为输入，执行一些计算或副作用，并返回更新后的状态。

1. Edges：确定下一步执行哪个 Node 的函数，它们基于当前状态。它们可以是条件分支或固定转换。

通过组合 Nodes 和 Edges，您可以创建随时间演化状态的复杂循环工作流。不过，真正的力量来自于 LangGraph 如何管理该状态。

需要强调的是： Nodes 和 Edges 不过是函数而已——它们可以包含 LLM 或只是普通的代码。

简而言之： _节点做工作，边决定下一步做什么_.

LangGraph 的底层图算法使用 消息传递 来定义通用程序。当节点完成其操作时，它会沿一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将结果消息传递给下一组节点，过程继续进行。受 Google 的 Pregel 系统启发，程序以离散的"超级步"进行。

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

StateGraph

StateGraph 类是使用的主要图类。它由用户定义的 State object.

编译您的图

要构建您的图，首先定义 状态，然后添加 节点 和 边，然后编译它。到底什么是编译图，为什么需要编译？

编译是一个非常简单的步骤。它对图的结构提供一些基本检查（没有孤立节点等）。这也是您可以指定运行时参数（如 检查点 和断点）的地方。您只需调用 .compile method:

graph = graph_builder.compile(...)

状态

定义图时，你要做的第一件事是定义 State 的图。 State 由 图的模式 以及 reducer 函数组成 指定如何将更新应用到状态。的架构将是所有 State 的输入模式 Nodes 和 Edges 在图中，可以是 TypedDict or a Pydantic 模型。所有 Nodes 将向发出更新 State 然后使用指定的 reducer function.

模式

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

默认情况下，图将具有相同的输入和输出模式。如果想更改此设置，可以直接指定输入和输出模式。当有很多键且某些明确用于输入而其他用于输出时，这很有用。请参阅 指南 了解更多

多个模式

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

• - Internal nodes can pass information that is not required in the graph's input / output.
• - We may also want to use different input / output schemas for the graph. The output might, for example, only contain a single relevant output key.

可以让节点在图内写入私有状态通道以进行内部节点通信。我们可以简单地定义一个私有模式， PrivateState.

还可以为图定义明确的输入和输出模式。在这种情况下，我们定义一个包含 _所有_ 与图操作相关的键的"内部"模式。但是，我们还定义 input 和 output 模式作为"内部"模式的子集来约束图的输入和输出。参见 定义输入和输出模式 了解更多详情。

让我们看一个例子：

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

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

1. 我们按以下方式初始化图：

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

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

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

归约器

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

默认归约器

这两个示例展示了如何使用默认归约器：

from typing_extensions import TypedDict

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

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

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

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

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

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

覆盖

在图状态中处理消息

为什么要使用消息？

大多数现代 LLM 提供商都有接受消息列表作为输入的聊天模型接口。LangChain 的 聊天模型接口 特别接受消息对象列表作为输入。这些消息有多种形式，如 HumanMessage（用户输入）或 AIMessage（LLM 响应）。

要了解更多关于消息对象的信息，请参阅 消息概念指南.

在图中使用消息

在许多情况下，将之前的对话历史作为消息列表存储在图状态中是有帮助的。为此，我们可以向图状态添加一个存储 Message 对象列表的键（通道），并用归约函数注解它（请参阅下面的示例中的 messages 键）。归约函数对于告诉图如何用每次状态更新来更新状态中的 Message 对象列表至关重要（例如，当节点发送更新时）。如果你没有指定归约函数，每次状态更新都会用最近提供的值覆盖消息列表。如果你只想将消息追加到现有列表，可以使用 operator.add 作为归约函数。

但是，你可能还希望在图状态中手动更新消息（例如人工介入）。如果你要使用 operator.add，您发送给图的manual state updates将被追加到现有的消息列表中，而不是更新现有消息。为了避免这种情况，您需要一个reducer来跟踪消息ID并在更新时覆盖现有消息。为此，您可以使用预构建的 add_messages 函数。对于全新的消息，它会简单地将消息追加到现有列表中，但也会正确处理现有消息的更新。

序列化

除了跟踪消息ID之外，add_messages 函数还会在接收到状态更新时尝试将消息反序列化为LangChain Message 对象，只要在 messages channel.

更多信息，请参见 LangChain serialization/deserialization. This allows sending graph inputs / state updates in the following format:

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

# and this is also supported
{"messages": [{"type": "human", "content": "message"}]}

由于使用 @[ Messages ] 时，状态更新总是被反序列化为LangChainadd_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—A RunnableConfig 对象，包含配置信息如 thread_id 和追踪信息如 tags
3. runtime—A 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)  # [!code highlight]
    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"`

重新执行和幂等性

当您使用 检查点保存器编译时，LangGraph 会在 super-step 边界处保存检查点，而不是在节点内部的函数中间。如果执行停止并稍后恢复（例如在 中断 or a 重试之后），受影响的 **节点** 会从头开始重新运行函数。暂停前的代码和副作用会再次执行。

Idempotency. 设计 **节点** 逻辑时应确保重新执行不会破坏状态。如果节点插入数据库行，运行两次不应创建重复行，除非是有意为之。使用幂等键、upsert 操作或先读后写检查。关于 interrupt()的副作用，请参阅 在 interrupt 之前调用的副作用必须是幂等的.

图表变更。 确定性 关于代码变更的规则不适用于图表结构。您可以添加或删除 **节点** 和边而不会破坏现有线程的恢复。恢复的运行使用保存的状态并执行当前编译的图表。

节点内部的任务和中断。 If a **节点** 调用 **任务** or interrupt, 恢复时适用更严格的确定性规则。LangGraph 恢复已完成的 **任务** 从检查点的结果，但更改 **任务** or interrupt 恢复点之前代码中的顺序可能导致缓存值不匹配。A Functional API **入口点** 编译为单个 **节点** 以这种方式运行整个入口点方法。参见 确定性, 幂等性，以及 在节点中使用任务.

在节点中使用任务

If a 节点 包含多个操作，您可能会发现将每个操作实现为 **任务** 而不是将逻辑分散到多个节点。当图使用检查点时，任务结果会被保存，因此恢复线程可以跳过已完成的 **任务** 节点内的工作。

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 supports caching of tasks/nodes based on the input to the node. To use caching:

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

例如：

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'))    # [!code highlight]
# [{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates'))    # [!code highlight]
# [{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]

1. 首次运行需要两秒（由于模拟的昂贵计算）。
2. 第二次运行使用缓存并快速返回。

边

边定义了逻辑如何路由以及图如何决定停止。这是您的代理如何工作的重要部分，也是不同节点之间如何相互通信的重要部分。有几种关键类型的边：

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

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

普通边

如果您 **始终** 想从节点 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"})

入口点

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

from langgraph.graph import START

graph.add_edge(START, "node_a")

条件入口点

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

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 are defined ahead of time and operate on the same shared state. However, there can be cases where the exact edges are not known ahead of time and/or you may want different versions of State 同时存在。一个常见的例子是 map-reduce 设计模式。在这种设计模式中，第一个节点可能生成一个对象列表，而你可能希望将其他节点应用到所有这些对象上。对象的数量可能事先未知（这意味着边的数量可能未知），输入 State 到下游 Node 应该不同（每个生成的对象一个）。

为了支持这种设计模式，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 or 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当你需要 **同时** 更新状态 **和** 路由到不同的节点。如果只需要路由而不更新状态，请使用 条件边 instead.

查看这个 操作指南 ，了解如何使用 Command.

graph

的端到端示例。如果你正在使用 子图，你可以通过指定 graph=Command.PARENT in Command:

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
    )

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

输入到 invoke or stream

resume

使用 Command(resume=...) 在中断后提供值并恢复图执行 中断传递给 resume 的值成为 interrupt() 调用的返回值：

请参阅 中断概念指南 了解更多关于中断模式的详细信息，包括多个中断和验证循环。

从工具返回值

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

请参阅 在工具内使用 了解更多。

图迁移

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

• - 对于位于图末尾的线程（即未被中断的线程），您可以更改整个图的拓扑结构（即所有节点和边，删除、添加、重命名等）
• - For threads currently interrupted, we support all topology changes other than renaming / removing nodes (as that thread could now be about to enter a node that no longer exists) -- if this is a blocker please reach out and we can prioritize a solution.
• - 对于状态修改，我们对键的添加和删除具有完整的向后和向前兼容性
• - 重命名的状态键会丢失其在现有线程中的保存状态
• - 以不兼容方式更改类型的状态键可能会导致在包含更改前状态的线程中出现一些问题——如果这成为阻碍，请联系我们，我们可以优先解决。

运行时上下文

创建图时，您可以指定一个 context_schema 作为传递给节点的运行时上下文。这对于传递 不属于图状态的信息给节点很有用。例如，您可能想要传递依赖项，如模型名称或数据库连接。

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

graph = StateGraph(State, context_schema=ContextSchema)

然后您可以使用 context 的 invoke method.

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)
    # ...

参见 添加运行时配置 获取完整的配置说明。

递归限制

递归限制设置图在单次执行中可以执行的最大 super-steps 步数。一旦达到限制，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 在图形执行时递增此计数器，并在 GraphRecursionError 一旦超过配置的 recursion_limit 时抛出异常。

访问当前步骤计数器

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

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

这些方法之间的主要区别是：

主动式优势：

• - 图形内优雅降级
• - 可在检查点保存中间状态
• - 通过部分结果获得更好的用户体验
• - 图形正常完成（无异常）

反应式优势：

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

其他可用元数据

连同 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 的选择