【翻译】LangGraph 1.0 Graph API 设计思想概览

说明: 阅读一些 Langchain 等大模型框架的官方文档时, 对于一些感觉不错的涉及到框架设计核心思想的文档进行翻译, 基本忠于原文内容, 但也加入一些个人理解的部分以及个人的经验案例进行二次创作, 属于个人学习笔记, 分享正在学习 Langchain 框架的朋友!!!

原文链接

目录

Graph

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

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

  2. Nodes：编码 Agent 逻辑的函数。它们接收当前State作为输入，执行某些计算或副作用，并返回更新后的State。

  3. Edges：根据当前State确定接下来执行哪个 Node 的函数。它们可以是条件分支或固定转换。

通过组合 Nodes 和 Edges，你可以创建随时间发生State迭代的复杂循环工作流。然而, 使用LangGraph框架真正的优势就来自 LangGraph 如何帮助你去轻松的管理这些共享State的复杂变更关系。

强调一下：Nodes 和 Edges 本质还是函数 - 它们可以包含 LLM 或普通代码。

一句话来说：节点是处理数据或消息的基本单元，边决定了数据或消息的流转方向。

LangGraph 的底层Graph算法使用消息传递来定义通用程序。当节点完成其操作时，它沿着一条或多条边向其他节点发送消息。这些接收节点然后执行它们的函数，将结果消息传递给下一组节点，如此循环这个过程, 直至消息在Graph 中流转了所有的节点, 最后进行输出。受到 Google 的 Pregel 系统启发，程序以一个个的super-step(中文不知道怎么翻译, 超步吗? 怪怪的, 所以还是保留原文)进行。

一个super-step可以被认为是对Graph节点的单次迭代。并行运行的节点属于同一super-step，而顺序运行的节点属于不同的super-step。在Graph执行开始时，所有节点都处于 不活跃 态。当节点通过任意一条输入边（或"通道"）上接收到新消息（即 state）时，会从不活跃态变为 活跃态。活跃的节点会触发去执行函数代码, 并将输出更新到State。在每个super-step的末尾，没有传入消息的节点通过将自己标记为 inactive 来投票"停止"。当所有节点都是 不活跃 且没有消息进行传输时，Graph执行结束。

Super-Step ：LangGraph 的核心执行模型

首先, 什么是 Super-Step？ Super-Step = 图执行中的一个"轮次"或"迭代"

在每个 super-step 中：

  • 所有可以并行运行的节点会同时执行
  • 这些并行节点构成一个 super-step
  • 当这一轮节点都完成后，才进入下一个 super-step

在执行中, 每个节点都经历一轮状态的转换, INACTIVE (默认/休眠) -> 收到消息 (来自上游节点或 START) -> ACTIVE (执行中) -> 执行完毕，发送消息到下游 -> INACTIVE (投票停止), 从INACTIVE开始 到回归到INACTIVE结束. 中间完成一轮完整的执行轮次.

以一个客服 Agent为例, 图解 Super-Step:
在这里插入图片描述

总结: Super-Step 是 LangGraph 的核心执行模型：

  • 它将图的执行分解为多个轮次, Super-step 是执行的基本单位, 同一 super-step 内的节点可以并行, 提升执行效率;不同 super-step 的节点必须顺序执行, 控制依赖顺序关系,保证执行的正确性; 通过分层执行的机制,在并行性能和依赖关系的处理之间达到平衡.
  • 每个轮次内，通过消息驱动的架构, 来实现节点的激活和执行
  • 每个节点都有明确清晰的状态转换规则, 最终整个图可以通过"投票停止"机制自动判断何时终止, 实现了自动终止检测方式.
  • 背后的设计思想来自 Google 的 Pregel 系统（大规模图处理）

这种设计使得 LangGraph 既能处理节点之间复杂的依赖关系，又能支持动态的拓扑变化，适合复杂Agent这类需要灵活控制流的应用。

StateGraph

StateGraph 类是编排Graph 时最常使用的类, 由用户指定 State 对象来参数化。

编译(Compile)工作流Graph

要构建Graph，你首先定义好state，然后添加nodes和edges，最后编译Graph。编译Graph 到底指的是什么，为什么需要编译这个过程？

编译是一个相当简单的步骤。它对Graph的结构进行一些基础检查（比如是否存在孤立的节点等）。在编译时, 也可以指定一些运行时参数(runtime args)，如设置检查点和断点。你通过调用 .compile 方法来编译你的Graph：

graph = graph_builder.compile(...)

注意, 在使用Graph之前, 必须先进行编译。

三大组件之一: State

构建Graph, 首先要做的是定义Graph的 State。State 中包含了Graph的schema以及reducer 函数，两者决定了如何将节点的输出更新到State中。State 的 Schema 默认会在Graph中所有 Nodes 和 Edges 全局生效，结构类型一般是 TypedDict 或 Pydantic 模型。同样, 所有 Nodes 将输出的结果经过 Reducer函数处理后, 也会合并到 State 中。

Schema

定义Schema的主要方式是使用 TypedDict。如果你想在State中设置默认值，那么可以使用 dataclass。如果对字段的取值设置一些验证逻辑, Langraph 也支持使用 Pydantic BaseModel 定义Graph State的结构（但注意 Pydantic 的性能不如 TypedDict 或 dataclass）。

默认情况下，Graph的输入和输出的 Schema结构都是一样。如果你需要使用多Schema的场景，LangGraph 允许显式指定输入和输出各自的Schema结构。当在 State 中包含了很多key，其中一些key明确用于输入，其他用于输出时，这种分开指定chema的方式就很灵活。有关更多信息，请参阅指南。

ps: 整理下 Langgraph 中针对 State 的几种常用定义方式:

  1. TypeDict(主流方式)
from typing import TypedDict, Annotated
	class AgentState(TypedDict):
		messages: Annotated[list[BaseMessage], add_messages]
		user_id: str

这种方式的优点就是性能最好, 也是最轻量简单的方式, 而且是官方最为提倡的方式, 也是使用 langchain.agent.create_agent来创建轻量级 agent 支持的state 定义方式. 但是这种方式不支持默认值的设置, 如果想要指定默认值, 可以使用 @dataclass
2. dataclass(支持默认值)

from dataclasses import dataclass
@dataclass
class AgentState:
	messages: Annotated[list[BaseMessage], add_messages]
	user_id: str = "default_user"

这种方式就是适用于需要默认值的场景. 但如果想要对参数字段的类型添加一些验证逻辑怎么办呢? 前两种都不支持, 此时就可以使用Pydantic BaseModel.
3. Pydantic BaseModel(支持参数验证)

from pydantic import BaseModel
class AgentState(BaseModel):
	messages: list[BaseMessage]
	user_id: str

这种方式虽然为参数设置一些约束条件, 并且在运行时进行验证, 但性能相比于 TypedDict更差, 一般场景不适用这种定义方式.

多Schema

通常，所有Graph节点都通过一个Schema进行数据交互。这意味着它们都读写都使用相同的State结构。但是，在某些情况下，我们希望对State有更多控制：

  • 内部节点想透传Graph的输入/输出中不需要的信息字段。
  • 为Graph使用不同的输入/输出Schema。例如，输出Schema可能只包含相关的key 字段。

为了内部节点的交互, 在Graph内写入私有的State Channel(这里的 Channel, 我理解是 State 中定义的 key)。也就是, 简单地定义一个私有Schema: private-state。

同样, 也可以为Graph显式定义输入和输出各自的Schema。在这些情况下，我们定义一个internal-schema，包含与Graph操作相关的所有 的键。但是，我们也定义了 input 和 output Schema，它们是internal-schema的子集，以方便控制Graph的输入和输出的信息。更多详细信息，请参阅指南。

让我们看一个例子：

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 作为输入Schema传递给 node_1。但是，输出却放到 foo字段里，这是 OverallState 中的一个通道(Channel, 也就是 State 中的一个 key )。Langgraph 是如何能写入到不包括在输入Schema中的State通道(Channel)？这是因为节点 可以写入GraphState中的任何的State通道。 在Graph初始化时, Graph State是所有State通道的并集，包括 OverallState 和 InputState 和 OutputState。

  2. 在初始化Graph, 要指定所有图内定义的 StateSchema：

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

    那么，我们如何能在 node_2 中写入 PrivateState？如果它没有在 StateGraph 初始化中设置，Graph如何对该Schema进行访问？

    我们可以这样做是因为_nodes也可以声明额外的State channels_，只要StateSchema定义存在。在这种情况下，PrivateState Schema已定义，所以我们可以在Graph中添加 bar 作为新的State通道并写入它。(这句原文也很绕, 说实在的, 没太看懂, 不过开发一般的 Agent 的也用不到, 可以先作为了解的知识点)

    So, how can we write to `PrivateState` in `node_2`? How does the graph gain access to this schema if it was not passed in the `StateGraph` initialization?

We can do this because `_nodes` can also declare additional state `channels_` as long as the state schema definition exists. In this case, the `PrivateState` schema is defined, so we can add `bar` as a new state channel in the graph and write to it.

Reducers

Reducers 是理解如何将节点的更新合并到 State 的关键。State 中的每个Key都有其自己独立的 reducer 函数。如果没有明确指定 reducer 函数，则默认指定对该Key的所有更新都是覆盖操作, 即State 中的同一Key 的"新值"替换"旧值"。除了"覆盖"操作类型的 Reducer, 还有几种不同类型的 reducers，从默认类型的 reducer 开始：

默认 Reducer

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

from typing_extensions import TypedDict

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

在这个例子中，没有为任何键指定 reducer 函数。假设Graph的输入是：

{"foo": 1, "bar": ["hi"]}。然后让我们假设第一个 Node 返回 {"foo": 2}。这被视为是对State内容的更新。注意 Node 不需要返回整个 State Schema的完整数据 - 只需要一个待变更的Key即可。将该变更合并到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 类型为第二个键（bar）指定一个 reducer 函数（operator.add）。注意第一个键保持不变。假设Graph的输入是 {"foo": 1, "bar": ["hi"]}。然后让我们假设第一个 Node 返回 {"foo": 2}。这被视为对State的更新。注意 Node 不需要返回完整 State Schema 字段- 只需要一个待更新的字段。将该字段的内容更新到 State后，State的值变成 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，那么 State 将是 {"foo": 2, "bar": ["hi", "bye"]}。注意这里 bar 键通过将新旧两个列表相加在一起来更新, 也就是reducer 函数（operator.add）的含义, 将新值追加到原始数据中。

覆盖型reducer
在某些情况下，你可能想绕过 reducer 并直接覆盖State值。LangGraph 为此目的提供了 [`Overwrite`](https://reference.langchain.com/python/langgraph/types/) 类型。[在此处了解如何使用 `Overwrite`](/oss/python/langgraph/use-graph-api#bypass-reducers-with-overwrite)。

在GraphState中使用Messages(消息)

为什么使用Messages？

大多数LLM 服务提供商都提供了聊天对话模型, 接受Messages(消息列表)作为输入。具体来说, LangChain 的chat model都会接收Message对象列表作为输入。这些消息对象有多种形式，例如 HumanMessage（用户输入）或 AIMessage（LLM 响应）。

要了解更多关于消息对象是什么，请参考消息概念指南。

在Graph中使用Messages(消息)

在大部分情况下，为了方便的存储对话历史,实现短期记忆的效果, 可以将历史对话转换为一组消息, 然后放在Graph 的 State 中。因此，可以向Graph State添加一个Key（Channel 也是一个所谓的消息通道），该Key中存放了一组 Message 对象，并指定了 reducer 函数（请参阅下面示例中的 messages 键）。在每次State更新时（例如，当节点send 更新时）, reducer 函数将告诉Graph如何更新State中的 Message 对象列表。如果没指定 reducer，那么默认情况下, 每次State更新都会用最近提供的值来覆盖消息列表。如果想简单地将消息附加到现有列表，你可以使用 operator.add 作为 reducer。

但是，你也可能想在你的Graph State中手动更新消息（例如在human-in-the-loop场景中）。如果你使用 operator.add，手动的State变更内容将被追加到现有消息列表，而不是更新现有消息。为了避免这种情况，你需要一个可以跟踪消息 ID 并覆盖现有消息（如果更新）的 reducer。为了实现这一点，你可以使用预构建的 add_messages 函数。对于全新消息，它将简单地附加到现有列表，但它也将正确处理现有消息的更新。

Serialization(序列化)

除了跟踪消息 ID，当对 messages 通道(原文中用 Channel 我就很费解, 所谓 Channel 实际就是指的 State 中的一个字段而已, 非得整这种乱七八糟的术语, 增加理解成本. )进行State更新时, add_messages 函数会尝试将消息反序列化为 LangChain 的 Message 对象。

有关 LangChain 序列化/反序列化的更多信息，请参阅此处。这允许通过以下格式来指定Graph输入/State更新：

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

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

ps: 所以在 agent.invoke 的时候两种写法都可以的原因, 1. agent.invoke({“messages”: [HumanMessage(content=“message”)]}) 2. agent.invoke({“messages”: [{“type”: “human”, “content”: “message”}]})

由于在使用 add_messages 时, State更新总是被反序列化为 LangChain Messages，你应该使用点符号来访问消息属性，如 state["messages"][-1].content(ps: 就是我们输出模型生成结果时, 经常用的一种写法)。

下面是一个使用 add_messages 作为其 reducer 函数的Graph示例。

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

由于经常要在State中包含消息列表字段，所以 LangGraph 就预构建了包含消息列表的State, 称为 MessagesState，这让使用消息变得容易。MessagesState 定义了一个单一的 messages 键，它是 AnyMessage 对象的列表，并使用 add_messages reducer。通常，由于在 Graph 中流转的State不仅仅只有消息，所以可以对该State进行子类继承, 然后扩展更多字段，如：

from langgraph.graph import MessagesState

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

三大组件之一:Nodes(节点)

LangGraph针对节点的设计哲是: 节点就是函数. 可以把一个普通的 Python 函数（同步或异步）直接添加到工作流中作为一个节点, 无需任何包装，节点可以接受以下参数：

  1. state – Graph的State
  2. config – 一个 RunnableConfig 对象，包含配置信息如 thread_id 和跟踪信息如 tags
  3. runtime – 一个 Runtime 对象，包含运行时context和其他信息如 store 和 stream_writer

类似于 NetworkX，你使用 add_node 方法将这些节点添加到Graph：

from dataclasses import dataclass
from typing_extensions import TypedDict

from langchain_core.runnables import RunnableConfig
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("在节点中: ", runtime.context.user_id)
    return {"results": f"Hello, {state['input']}!"}

def node_with_config(state: State, config: RunnableConfig):
    print("在节点中，thread_id: ", config["configurable"]["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_config", node_with_config)
...

在LangGraph内部，python函数被转换为 RunnableLambda，它为你的函数添加批处理和异步支持，以及原生跟踪和调试功能。

如果你向Graph添加节点而不指定名称，它将被给予一个等于函数名称的默认名称。

builder.add_node(my_node)
# 然后你可以通过将其引用为 `"my_node"` 来创建到/从该节点的边

ps:
LangGragh 中其实有两类节点, 普通节点和工具节点(ToolNode). 普通节点是直接写函数, 不需要做额外的封装, 但ToolNode是对Tool函数的执行节点包装器, 将一个工具函数转换成了一个可以被调度执行的图节点.
ToolNode中除了绑定代表工具的函数, 还可以指定工具名称工具调用失败的异常处理, 所以在实际Agent 工作流执行中, ToolNode 负责解析大模型生成的 ToolCall对象(包含工具名/参数等), 然后调用具体的工具函数, 最后将结果转换成 ToolMessage, 并追加到 state 中的消息列表里.

# retriever_tool 是 LangChain 的 Tool 对象
# 需要 ToolNode 来处理 LLM 生成的 ToolCall
graph.add_node('retrieve', ToolNode([retriever_tool]))

具体执行流程: Agent 节点 → 决定调用 rag_retriever → ToolNode 节点 → 执行检索，返回结果 → Agent 节点 → 看到检索结果，继续推理
另外, ToolNode是从langgraph.prebuilt模块中导入, 我看了下该模块下, 只定义了一种在用特殊的节点类(另一个ValidationNode已被废弃了), 即ToolNode, 所以LangGraph 预设的 Node 扩展类当前也就只有 ToolNode了这一种了,
在这里插入图片描述

START 节点

START 节点是一个特殊节点，代表接收用户输入的Graph节点。引用此节点的主要目的是确定应该首先调用哪些节点。

from langgraph.graph import START

graph.add_edge(START, "node_a")

END 节点

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

from langgraph.graph import END

graph.add_edge("node_a", END)

Node Caching(节点缓存)

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

  • 在编译Graph时指定缓存（或指定入口点）
  • 为节点指定缓存策略。每个缓存策略支持：
    • 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'))    # [!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. 第二次运行利用缓存并快速返回。

三大组件之一:Edges(边)

边定义了数据如何路由的逻辑以及整个工作流Graph如何结束。边是实现Agent的流程编排和节点间进行消息交互的重要组件。有几种关键类型的边：

  • 普通边(Normal Edges)：直接从一个节点转到下一个。
  • 条件边(Conditional Edges)：调用函数, 根据函数结果有条件的确定接下来要流转的节点。
  • 入口点(Entry Point)：当用户输入到达时首先调用哪个节点。
  • 条件入口点(Conditional Entry Point)：当用户输入到达时, 调用函数, 根据函数输出有条件的确定首先调用哪个节点。

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

普通边

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

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

条件边

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

graph.add_conditional_edges("node_a", routing_function)

类似于节点，routing_function 接受Graph的当前 state 并返回一个值(ps: 这个值其实是下一个节点的名称)。

默认情况下，routing_function 的返回值是下一个要路由的节点名称（或节点列表）。所有这些节点将作为下一个super-step的一部分并行运行。

此外, add_conditional_edges也可以指定一个节点映射字典，将 routing_function 的输出结果映射到下一个节点的名称。

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

注意: 如果你想在单个函数中同时实现对State更新和节点路由，请使用 Command 而不是条件边。(ps: 节点条件路由实现有两种方式, 一种是条件边, 另外一种是 Command 命令的方式)

入口点

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

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 是提前定义的，并在相同的全局共享State上运行。但是，在某些情况下，节点之间的边关系可能无法提前知道，或者你可能想要不同版本的 State 同时存在。这种情况的一个常见例子是map-reduce的设计模式。在这种设计模式中，第一个节点可能生成一组对象，你可能想将某个节点去接收所有这些对象的结果, 即所有这些对象都汇总到某个节点上。但对象的数量可能提前不知道（即, 边的数量也不知道），而且下游 Node 的输入 State 也可能会不同（每个生成的对象一个）。
在这里插入图片描述

为了支持这种设计模式，LangGraph 支持从条件边(graph.add_conditional_edges)返回 Send 对象。Send 接受两个参数：第一个是节点的名称，第二个是要传递给该节点的State。

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

前面介绍了如何在节点中实现条件边的动态创建和 State 的更新机制。那么，能否将数据流转和 State 更新这两个操作结合在一起呢？例如，你可能想在同一个节点中既执行State更新又决定接下来要转到哪个节点。LangGraph 允许从节点返回 Command 对象, 来将边创建和 State 更新两个操作封装在一起：

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # State更新
        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[Literal["my_other_node"]]。这对于Graph编译很重要，可以告诉 LangGraph my_node 可以导航到 my_other_node。

查看这个操作指南以获得关于如何使用 Command 的端到端示例。

何时应该使用 Command 而不是条件边？

  • 当你需要既更新GraphState又路由到不同的节点时，使用 Command。例如，在实现Multi-Agent切换时，核心点就是路由到不同的Agent, 并向该Agent透传一些信息。
  • 使用条件边可以实现节点之间的条件路由, 但不能更新State。

导航到父Graph中的节点

如果你使用子图，你可能想从子图中的节点导航到不同的子图（即父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中的节点
        graph=Command.PARENT
    )
将 `graph` 设置为 `Command.PARENT` 将导航到最接近的父Graph。

当你从子图节点向父Graph节点发送更新时，对于父Graph和子图StateSchema共享的Key，你必须在父GraphState中为你正在更新的键定义一个reducer。参见此示例。

这在实现Multi-Agent切换时特别有用。

查看本指南以获得详细信息。

内部工具State 更新

一个常见的用例是: 在工具内部更新GraphState。例如，在客户支持应用程序中，你可能想在对话开始时根据客户的帐号或 ID 查找客户信息。

有关详细信息，请参考本指南。

Human-in-the-loop

Command 是人在循环中工作流的重要部分：当使用 interrupt() 收集用户输入时，Command 然后用于提供输入并通过 Command(resume="User input") 恢复执行。查看本概念指南以获得更多信息。

Graph迁移

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

  • 对于在Graph末尾的线程（即未中断），你可以更改Graph的整个拓扑（即所有节点和边、删除、添加、重命名等）
  • 对于当前中断的线程，我们支持除了重命名/删除节点之外的所有拓扑更改（因为该线程现在可能即将进入不再存在的节点）-- 如果这是一个阻塞，请联系我们，我们可以优先考虑解决方案。
  • 对于修改State，我们对添加和删除键有完整的向后和向前兼容性
  • 重命名的State键会在现有线程中丢失其保存的State
  • 类型更改不兼容的State键可能会在具有更改前State的线程中导致问题 – 如果这是一个阻塞，请联系我们，我们可以优先考虑解决方案。

运行时上下文(Runtime context)

创建Graph时，你可以为传递给节点的运行时上下文(runtime context)指定 context_schema。这可以帮助传递非GraphState的数据。例如，你可能想传递依赖项，如模型名或数据库连接。

ps: 这里的运行时环境是存放的一些与会话无关的全局环境信息, 侧重点这些环境信息是静态不变的,比如用户身份/API 秘钥等, 这些信息一般只会读, 不会被修改, 这是与 State会被不同节点动态修改更新的最大的区别. 所以你如果要是考虑存放一些全局静态变量, 可以考虑 RuntimeContext.
和 State 的异同点:

  • 同: 都是节点的输入, 如def my_node(state: AgentState, context: RuntimeContext):, 同时接收 State 和 Context
  • 异: State 是Graph 的内部节点之间流转的数据载体, 动态可变; Context 是全局静态变量, 可读不可写.
@dataclass
class ContextSchema:
    llm_provider: str = "openai"

graph = StateGraph(State, context_schema=ContextSchema)

然后你可以使用 invoke 方法的 context 参数将此上下文传递到Graph中。

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

有关配置的完整分解，请参阅本指南。

递归限制

递归限制设置Graph在单次执行期间可以执行的最大super-step数。一旦达到限制，LangGraph 将引发 GraphRecursionError。默认情况下，此值设置为 25 步。递归限制可以在运行时在任何Graph上设置，并通过配置字典传递给 invoke/stream。重要的是，recursion_limit 是一个独立的 config 键，不应该像所有其他用户定义的配置那样在 configurable 键内传递。请参阅下面的示例：

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

阅读本操作指南以了解更多关于递归限制如何工作的信息。

访问和处理递归计数器

当前步骤计数器可在任何节点内的 config["metadata"]["langgraph_step"] 中访问，允许在达到递归限制之前进行主动递归处理。这使你能够在Graph逻辑中实现优雅的降级策略。

它如何工作

步骤计数器存储在 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"当前在步骤：{current_step}")
    return state
主动递归处理

你可以检查步骤计数器并在达到限制之前主动路由到不同的节点。这允许在你的Graph中进行优雅的降级。

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

def reasoning_node(state: dict, config: RunnableConfig) -> dict:
    current_step = config["metadata"]["langgraph_step"]
    recursion_limit = config["recursion_limit"]  # 总是存在，默认为 25

    # 检查是否接近限制（例如，80% 阈值）
    if current_step >= recursion_limit * 0.8:
        return {
            **state,
            "route_to": "fallback",
            "reason": "接近递归限制"
        }

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

def fallback_node(state: dict, config: RunnableConfig) -> dict:
    """处理递归限制接近的情况"""
    return {
        **state,
        "messages": state["messages"] + ["达到复杂性限制，提供最佳努力答案"]
    }

def route_based_on_state(state: dict) -> str:
    if state.get("route_to") == "fallback":
        return "fallback"
    elif state.get("done"):
        return END
    return "reasoning"

# 构建Graph
graph = StateGraph(dict)
graph.add_node("reasoning", reasoning_node)
graph.add_node("fallback", fallback_node)
graph.add_conditional_edges("reasoning", route_based_on_state)
graph.add_edge("fallback", END)
graph.set_entry_point("reasoning")

app = graph.compile()
主动与被动方法

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

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, END
from langgraph.errors import GraphRecursionError

# 主动方法（推荐）
def agent_with_monitoring(state: dict, config: RunnableConfig) -> dict:
    """在Graph内主动监控和处理递归"""
    current_step = config["metadata"]["langgraph_step"]
    recursion_limit = config["recursion_limit"]

    # 早期检测 - 路由到内部处理
    if current_step >= recursion_limit - 2:  # 限制前 2 步
        return {
            **state,
            "status": "recursion_limit_approaching",
            "final_answer": "达到迭代限制，返回部分结果"
        }

    # 正常处理
    return {"messages": state["messages"] + [f"步骤 {current_step}"]}

# 被动方法（回退）
try:
    result = graph.invoke(initial_state, {"recursion_limit": 10})
except GraphRecursionError as e:
    # 在Graph执行失败后外部处理
    result = fallback_handler(initial_state)

这些方法之间的关键区别是：

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

主动优势：

  • Graph内的优雅降级
  • 可以在检查点中保存中间State
  • 更好的用户体验，具有部分结果
  • Graph正常完成（无异常）

被动优势：

  • 更简单的实现
  • 无需修改Graph逻辑
  • 集中式错误处理
其他可用的元数据

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

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

    print(f"步骤：{metadata['langgraph_step']}")
    print(f"节点：{metadata['langgraph_node']}")
    print(f"触发器：{metadata['langgraph_triggers']}")
    print(f"路径：{metadata['langgraph_path']}")
    print(f"检查点命名空间：{metadata['langgraph_checkpoint_ns']}")

    return state

可视化

能够可视化Graph通常很好，特别是当它们变得更复杂时。LangGraph 附带了几种内置的Graph可视化方式。有关更多信息，请参阅本操作指南。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

AI多Agent协作系统实战（二十九）：“小虾真的能分析出根因吗？“——当AI团队里没有人会“看病“

avatar AI Agent技术社区

AI Agent可观测性：破解多步推理黑盒的技术实践

6.1 给开发者的建议在项目早期就将可观测性纳入设计，而非事后补救从核心指标和关键路径开始，逐步完善建立“观测-分析-优化”的闭环文化6.2 推荐工具与资源开源工具：LangSmith, Weights & Biases, MLflow, Prometheus + Grafana学习资源：相关论文、开源项目、技术博客。

avatar AI Agent技术社区
cover

AI Agent从无到有9: 掌握提示词工程与思维链模式

avatar AI Agent技术社区