BytesRouter

主题

LangChain 文档版本 跟随上游 源快照 d7ebdf1d 更新于 2026-10-08

Graph API 概述

图

从根本上说，LangGraph 将智能体工作流建模为图。你使用三个关键组件来定义智能体的行为：

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

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

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

通过组合 Nodes 和 Edges，你可以创建随时间演进状态、带有复杂循环的工作流。不过，真正的威力在于 LangGraph 管理该状态的方式。

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

简而言之：节点负责工作，边指明接下来做什么。

LangGraph 的底层图算法使用消息传递来定义通用程序。当一个节点完成其操作时，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行各自的函数，将产生的消息传递给下一组节点，如此继续。受 Google 的 Pregel 系统启发，程序以离散的「super-step」推进。

一个 super-step 可以看作是对图节点的一次迭代。并行运行的节点属于同一个 super-step，而顺序运行的节点则属于不同的 super-step。在图执行开始时，所有节点都处于 inactive 状态。当节点在其任意入边（或「通道」）上收到一条新消息（状态）时，它就会变为 active。随后该活动节点运行其函数并以更新作为响应。在每个 super-step 结束时，没有收到入消息的节点通过将自身标记为 inactive 来投票 halt。当所有节点都处于 inactive 且没有消息在传输时，图执行便终止。

StateGraph

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

编译你的图

要构建你的图，首先要定义状态，然后添加节点和边，接着对其进行编译。编译你的图究竟是什么，又为什么需要它？

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

python
graph = graph_builder.compile(...)

WARNING

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

状态

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

Schema

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

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

INFO

langchain 中更高级的 create_agent 工厂不支持 Pydantic 状态 schema。

多个 schema

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

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

可以让节点写入图内的私有状态通道，以实现内部节点通信。我们可以简单地定义一个私有 schema，即 PrivateState。

也可以为图定义显式的输入和输出 schema。在这些情况下，我们定义一个包含与图操作相关的所有键的「内部」schema。但是，我们还会定义 input 和 output schema，它们是「内部」schema 的子集，用于约束图的输入和输出。更多详情请参见定义输入和输出 schema。

让我们看一个示例：

python
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:
    # 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'}

View example trace

打开本示例的一个公开 LangSmith 运行记录。

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

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

  2. 我们使用以下内容初始化图：

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

    我们如何在 node_2 中写入 PrivateState？如果该 schema 没有在 StateGraph 初始化中传入，图如何获得对该 schema 的访问权限？

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

WARNING

流式传输时不会对私有通道进行脱敏。

输入、输出和私有 schema 约束了每个节点读取什么（其输入 schema）以及 invoke 返回什么（输出 schema）。它们不会向 stream 隐藏通道。

当你使用 stream_mode="values" 进行流式传输时，图默认会发出所有状态通道，包括私有通道，因为值流式传输默认使用完整的状态通道集合，而不是输出 schema。这就是为什么像 bar 这样的私有通道会被 invoke 隐藏，但在流式传输时可见：

python
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'}        # <-- private channel
# {'foo': 'My name', 'user_input': 'My', 'graph_output': 'My name is Lance', 'bar': 'My name is'}

查看示例追踪

打开此示例的公开 LangSmith 运行记录。

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

python
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"。

归约器

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

归约器参数

每个归约器都是一个带有两个位置参数的二元函数：

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

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

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

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

python
from typing import Annotated

from typing_extensions import TypedDict

def append_strings(left: list[str], right: list[str]) -> list[str]:
    """Combine the existing state value (left) with a node update (right)."""
    return left + right

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

假设状态为 {"tags": ["draft"]}，并且某个节点返回 {"tags": ["review"]}。LangGraph 会调用：

python
append_strings(left=["draft"], right=["review"])  # returns ["draft", "review"]

tags 的新状态值为 ["draft", "review"]。

自定义归约器会组合左参数和右参数。默认归约器会丢弃左参数，只保留右参数。

默认归约器

默认归约器忽略左参数，并用右参数替换状态值。以下示例展示了如何使用默认归约器：

python
from typing_extensions import TypedDict

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

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

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

自定义归约器

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

python
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）指定了一个归约器函数（operator.add）。请注意，第一个键保持不变。假设图的输入是 {"foo": 1, "bar": ["hi"]}。接着假设第一个 Node 返回 {"foo": 2}。这被视为对状态的一次更新。请注意，Node 不需要返回整个 State schema——只需返回一次更新。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，那么 State 将变为 {"foo": 2, "bar": ["hi", "bye"]}。请注意这里 bar 键是通过将两个列表相加来更新的。

覆盖

TIP

在某些情况下，你可能希望绕过归约器并直接覆盖一个状态值。LangGraph 为此提供了 Overwrite 类型。在此处了解如何使用 Overwrite。

重置归约器字段

使用归约器时一个常见的困惑来源：对于合并型归约器，返回空值不会清除该字段。因为归约器将右参数合并到左参数中，空更新会被合并进去，之前累积的值会被保留。

这种模式对于必须在重试尝试之间清除的错误缓冲区或重试计数器很重要：

python
from operator import add
from typing import Annotated

from typing_extensions import TypedDict

class State(TypedDict):
    errors: Annotated[list[str], add]

# node A returns {"errors": ["bad sql"]}
# node B returns {"errors": []}
# state["errors"] is still ["bad sql"]; the empty list is merged in, not cleared

要在保留合并型归约器的同时清除该字段，请用 Overwrite 包装该更新：

python
from operator import add
from typing import Annotated

from langgraph.types import Overwrite
from typing_extensions import TypedDict

class State(TypedDict):
    errors: Annotated[list[str], add]

def clear_errors(state: State):
    # Bypass the merging reducer and clear the field
    return {"errors": Overwrite([])}

更多信息请参见使用 Overwrite 绕过归约器。

未追踪值

UntrackedValue 用于那些在图执行期间应存在但绝不应保存到检查点的状态字段。当图从检查点恢复时，未追踪值会被重置为其初始状态（或不可用）。

这在以下场景中很有用：

  • 无法序列化的数据库连接
  • 应在恢复时重建的临时缓存
  • 你不想持久化的大型对象
  • 每次都应全新传入的仅运行时配置
typescript
import { StateSchema, UntrackedValue, MessagesValue } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: MessagesValue,

  // Untracked: throws if multiple nodes write in same step (guard: true is default)
  dbConnection: new UntrackedValue<DatabaseConnection>(),

  // Untracked with guard: false allows multiple writes, keeps last value
  tempCache: new UntrackedValue(
    z.record(z.string(), z.unknown()),
    { guard: false }
  ),

  // Untracked without a schema (for maximum flexibility)
  runtimeConfig: new UntrackedValue(),
});

行为：

  • 执行期间：值会像普通状态一样被存储并可访问
  • 检查点化时：未追踪值会被排除在检查点数据之外
  • 恢复时：未追踪值从头开始（为空或使用其默认值）
  • 使用 guard: true（默认）：如果多个节点在同一步写入则抛出错误
  • 使用 guard: false：允许多次写入，最后一个值生效

WARNING

不要将 UntrackedValue 用于需要在中断或时间旅行之间持久化的数据。对于持久化数据，请使用常规状态字段或 ReducedValue。

类型工具

LangGraph 提供了若干类型工具，以便在定义节点和条件边时获得更好的 TypeScript 类型安全性。

GraphNode

使用 GraphNode 为在图构建器之外定义的节点函数标注类型：

typescript
import { GraphNode, StateSchema, Command } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  count: z.number().default(0),
  result: z.string(),
});

// Basic node - receives state, returns partial update
const incrementNode: GraphNode<typeof State> = (state) => {
  return { count: state.count + 1 };
};

// Async node
const fetchNode: GraphNode<typeof State> = async (state, config) => {
  const response = await fetch(`/api/data/${state.count}`);
  return { result: await response.text() };
};

// Node with Command routing - specify valid destinations
const routerNode: GraphNode<{ InputSchema: typeof State; Nodes: "process" | "done" }> = (state) => {
  if (state.count >= 10) {
    return new Command({ goto: "done" });
  }
  return new Command({
    update: { count: state.count + 1 },
    goto: "process"
  });
};

State.Node 简写

每个 StateSchema 实例都有一个 Node 属性，为节点类型标注提供了一种简写：

typescript
const State = new StateSchema({
  messages: MessagesValue,
  step: z.string(),
});

// These are equivalent:
const myNode1: GraphNode<typeof State> = (state) => ({ step: "done" });
const myNode2: typeof State.Node = (state) => ({ step: "done" });

ConditionalEdgeRouter

将 ConditionalEdgeRouter 用于条件边中的路由函数（不更新状态，仅路由）：

typescript
import { ConditionalEdgeRouter, END } from "@langchain/langgraph";

const State = new StateSchema({
  shouldContinue: z.boolean(),
  step: z.string(),
});

// Router returns node name(s) or END
const router: ConditionalEdgeRouter<{ InputSchema: typeof State; Nodes: "process" | "summarize" }> = (state) => {
  if (!state.shouldContinue) {
    return END;
  }
  return state.step === "initial" ? "process" : "summarize";
};

// Use in graph
graph.addConditionalEdges("check", router);

StateSchema.State 和 StateSchema.Update

从 schema 中提取状态类型和更新类型，以便在自定义类型定义中使用：

typescript
import { StateSchema } from "@langchain/langgraph";

const MyStateSchema = new StateSchema({
  messages: MessagesValue,
  count: z.number().default(0),
});

// Extract the full state type
type MyState = typeof MyStateSchema.State;
// { messages: BaseMessage[], count: number }

// Extract the update type (partial, with reducer input types)
type MyUpdate = typeof MyStateSchema.Update;
// { messages?: Messages, count?: number }

在图状态中使用消息

为什么使用消息？

大多数现代 LLM 服务商都有一个聊天模型接口，它接受消息列表作为输入。尤其是 LangChain 的聊天模型接口，它接受消息对象列表作为输入。这些消息有多种形式，例如 HumanMessage（用户输入）或 AIMessage（LLM 响应）。

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

在你的图中使用消息

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

然而，你可能还想手动更新图状态中的消息（例如人在回路）。如果你使用 operator.add，你发送给图的手动状态更新会被追加到现有的消息列表，而不是更新已有消息。为了避免这种情况，你需要一个能够跟踪消息 ID 并在消息被更新时覆盖已有消息的归约器。为此，你可以使用预构建的 add_messages 函数。对于全新的消息，它只会追加到现有列表，但它也会正确处理对已有消息的更新。

序列化

除了跟踪消息 ID 之外，每当在 messages 通道上收到状态更新时，add_messages 函数还会尝试将消息反序列化为 LangChain Message 对象。

更多信息请参见 LangChain 序列化/反序列化。这允许以以下格式发送图输入/状态更新：

python
# 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 作为其归约器函数的图示例。

python
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 归约器。通常，除了消息之外还有更多状态需要跟踪，因此我们看到人们对该状态进行子类化并添加更多字段，例如：

python
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 方法将这些节点添加到图中：

python
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，它为你的函数增加了批处理和异步支持，以及原生追踪和调试。

如果你向图中添加节点时未指定名称，它将被赋予一个等同于函数名的默认名称。

python
builder.add_node(my_node)
# You can then create edges to/from this node by referencing it as `"my_node"`

重新执行与幂等性

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

幂等性。 设计节点逻辑，使重新执行不会破坏状态。如果某个节点插入了一行数据库记录，那么运行两次不应产生重复行，除非这是有意为之。请使用幂等键、upsert 或写前读取检查。关于 interrupt() 周围的副作用，请参见在 interrupt 之前调用的副作用必须是幂等的。

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

节点内的任务与中断。 如果节点调用了任务或 interrupt，则恢复时会适用更严格的确定性规则。LangGraph 会从检查点器恢复已完成的任务结果，但在恢复点之前的代码中更改任务或 interrupt 的顺序可能导致与缓存的值不匹配。一个函数式 API entrypoint 会编译成一个单一的节点，以这种方式运行整个 entrypoint 方法。请参见确定性、幂等性和在节点中使用任务。

在节点中使用任务

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

python
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)

查看示例追踪

打开此示例的公开 LangSmith 运行记录。

START 节点

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

python
from langgraph.graph import START

graph.add_edge(START, "node_a")

END 节点

END 节点是一个特殊节点，表示终止节点。当你想要表示哪些边在完成后没有后续操作时，就会引用该节点。

python
from langgraph.graph import END

graph.add_edge("node_a", END)

节点缓存

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

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

例如：

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

NOTE

set_entry_point(node) 定义图将执行的第一个节点。 它等价于 builder.add_edge(START, node)。

set_finish_point(node) 定义图中的最后一个节点。 它等价于 builder.add_edge(node, END)。

两种方法都有效，但 add_edge(START, ...) 和 add_edge(..., END) 是推荐的现代语法。

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

边

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

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

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

WARNING

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

普通边

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

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

条件边

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

python
graph.add_conditional_edges("node_a", routing_function)

与节点类似，routing_function 接受图的当前 state 并返回一个值。

默认情况下，routing_function 的返回值被用作接下来要向其发送状态的节点名称（或节点列表）。所有这些节点都将作为下一个 superstep 的一部分并行运行。

你可以选择提供一个字典，将 routing_function 的输出映射到下一个节点的名称。

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

TIP

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

入口点

入口点是图启动时首先运行的一个或多个节点。你可以使用从虚拟 START 节点到要执行的第一个节点的 add_edge 方法来指定从何处进入图。

python
from langgraph.graph import START

graph.add_edge(START, "node_a")

条件入口点

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

python
from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

你可以选择提供一个字典，将 routing_function 的输出映射到下一个节点的名称。

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

Send

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

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

python
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

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

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

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

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

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

NOTE

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

WARNING

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，从子图内的某个节点导航到父图中的另一个节点：

python
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
    )

NOTE

将 graph 设置为 Command.PARENT 将导航到最近的父图。

当你从子图节点向父图节点发送更新，且该键为父图和子图状态 schema共享时，你必须为你在父图状态中更新的该键定义一个归约器。请参见此示例。

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

作为 invoke 或 stream 的输入

WARNING

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

python
# 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() 调用的返回值：

python
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

View example trace

打开本示例的一个公开 LangSmith 运行记录。

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

从工具返回

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

WARNING

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

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

图迁移

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

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

TIP

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

运行时上下文

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

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

graph = StateGraph(State, context_schema=ContextSchema)

然后你可以使用 invoke 方法的 context 参数将此上下文传入图。

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

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

python
from langgraph.runtime import Runtime

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

有关配置的完整说明，请参见添加运行时配置。

递归限制

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

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

阅读递归限制以了解更多关于递归限制如何工作的内容。

访问和处理递归计数器

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

工作原理

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

访问当前步骤计数器

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

python
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 托管值，用于跟踪在达到递归限制之前还剩多少步。这允许在你的图中实现优雅降级。

python
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})

主动式与响应式方法

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

python
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"] 中还提供以下元数据：

python
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。

了解更多

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可