图

从核心层面看，LangGraph 将智能体工作流建模为图。你可以使用三个关键组件来定义智能体的行为：

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

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

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

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

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

简而言之：节点负责执行工作，边负责决定下一步做什么。

LangGraph 的底层图算法使用消息传递来定义通用程序。当一个 Node 完成其操作时，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行各自的函数，将产生的消息传递给下一组节点，如此循环往复。受 Google 的 Pregel 系统的启发，程序以离散的"超级步骤（superstep）"推进。

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

StateGraph

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

编译图

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

编译是一个非常简单的步骤。它会对图的结构做一些基本检查（例如没有孤立节点等）。你还可以在这里指定运行时参数，例如检查点和断点。只需调用 .compile 方法即可编译图：

graph = graph_builder.compile(...)

const graph = new StateGraph(StateAnnotation)
  .addNode("nodeA", nodeA)
  .addEdge(START, "nodeA")
  .addEdge("nodeA", END)
  .compile();

WARNING

在可以使用图之前，你必须先编译它。

状态

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

定义图时你要做的第一件事就是定义图的 State。State 由图的 schema 以及指定如何对状态应用更新的 reducer 函数 组成。State 的 schema 将是图中所有 Nodes 和 Edges 的输入 schema。你使用 StateSchema 类定义状态，该类接受任何标准 schema（如 Zod）作为单个字段，同时接受像 ReducedValue 和 MessagesValue 这样的特殊值类型。所有 Nodes 都会向 State 发出更新，这些更新随后使用指定的 reducer 函数来应用。

Schema

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

默认情况下，图的输入和输出 schema 相同。如果你想改变这一点，也可以直接指定显式的输入和输出 schema。当你有大量键，其中一些专门用于输入、另一些专门用于输出时，这非常有用。更多信息请参阅指南。

INFO

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

指定图 schema 的主要方式是使用 StateSchema 类。schema 中的每个字段可以是：

• 用于简单字段的 Standard schema（成为一个"末值"通道，更新时覆盖）
• 用于需要自定义 reducer 函数的字段的 ReducedValue（当节点并行运行时）
• 用于对话消息列表的 MessagesValue（内置感知消息的 reducer）
• 用于不应被检查点持久化的临时状态的 UntrackedValue

import {
  StateSchema,
  ReducedValue,
  MessagesValue,
  UntrackedValue
} from "@langchain/langgraph";
import { z } from "zod/v4";

const AgentState = new StateSchema({
  // 预构建的消息值，带有内置 reducer
  messages: MessagesValue,

  // 简单字段直接使用 Zod schema
  currentStep: z.string(),

  // 带有默认值的字段
  retryCount: z.number().default(0),

  // 用于累积值的自定义 reducer
  allSteps: new ReducedValue(
    z.array(z.string()).default(() => []),
    {
      inputSchema: z.string(),
      reducer: (current, newStep) => [...current, newStep],
    }
  ),

  // 不会保存到检查点的临时状态
  tempCache: new UntrackedValue(z.record(z.string(), z.unknown())),
});

// 类型提取
type State = typeof AgentState.State;   // 完整的状态类型
type Update = typeof AgentState.Update; // 部分更新类型

// 在图中使用
const graph = new StateGraph(AgentState)
  .addNode("myNode", ...)
  .compile();

默认情况下，图的输入和输出 schema 相同。如果你想改变这一点，也可以直接指定显式的输入和输出 schema。当你有大量键，其中一些专门用于输入、另一些专门用于输出时，这非常有用。

多个 schema

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

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

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

也可以为图定义显式的输入和输出 schema。在这些情况下，我们定义一个包含图操作相关的_所有_键的"内部"schema。同时，我们还定义作为"内部"schema 子集的 input 和 output schema，以约束图的输入和输出。更多细节请参阅定义输入和输出 schema。

让我们看一个例子：

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

import { END, START, StateGraph, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const InputState = new StateSchema({
  userInput: z.string(),
});

const OutputState = new StateSchema({
  graphOutput: z.string(),
});

const OverallState = new StateSchema({
  foo: z.string(),
  userInput: z.string(),
  graphOutput: z.string(),
});

const PrivateState = new StateSchema({
  bar: z.string(),
});

const graph = new StateGraph({
  state: OverallState,
  input: InputState,
  output: OutputState,
})
  .addNode("node1", (state) => {
    // 写入 OverallState
    return { foo: state.userInput + " name" };
  })
  .addNode("node2", (state) => {
    // 从 OverallState 读取，写入 PrivateState
    return { bar: state.foo + " is" };
  })
  .addNode(
    "node3",
    (state) => {
      // 从 PrivateState 读取，写入 OutputState
      return { graphOutput: state.bar + " Lance" };
    },
    { input: PrivateState },
  )
  .addEdge(START, "node1")
  .addEdge("node1", "node2")
  .addEdge("node2", "node3")
  .addEdge("node3", END)
  .compile();

await graph.invoke({ userInput: "My" });
// { graphOutput: 'My name is Lance' }

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

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

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

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

我们如何能在 `node_2` 中写入 `PrivateState`？如果这个 schema 没有传入 `StateGraph` 初始化，图是如何获得访问它的权限的呢？

我们可以这样做，因为 `_nodes` 只要存在状态 schema 定义，就可以声明额外的状态 `channels_`。在本例中，`PrivateState` schema 已定义，因此我们可以在图中添加 `bar` 作为新的状态通道并写入它。

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

2. 我们使用 StateGraph({ state: OverallState, input: InputState, output: OutputState }) 初始化图。我们如何能在 node2 中写入 PrivateState？如果这个 schema 没有传入 StateGraph 初始化，图是如何获得访问它的权限的呢？我们可以这样做，因为_节点只要存在状态 schema 定义，就可以声明额外的状态通道。_在本例中，PrivateState schema 已定义，因此我们可以在图中添加 bar 作为新的状态通道并写入它。

WARNING

私有通道在流式输出时不会被隐藏。

输入、输出和私有 schema 会约束每个节点_读取_的内容（其输入 schema）以及 invoke _返回_的内容（输出 schema）。它们不会从 stream 中隐藏通道。

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

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'}

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

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

WARNING

私有通道在流式输出时不会被隐藏。

输入、输出和私有 schema 会约束每个节点_读取_的内容（其输入 schema）以及 invoke _返回_的内容（输出 schema）。它们不会从 stream 中隐藏通道。

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

import { END, START, StateGraph, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const InputState = new StateSchema({
  userInput: z.string(),
});

const OutputState = new StateSchema({
  graphOutput: z.string(),
});

const OverallState = new StateSchema({
  foo: z.string(),
  userInput: z.string(),
  graphOutput: z.string(),
});

const PrivateState = new StateSchema({
  bar: z.string(),
});

const graph = new StateGraph({
  state: OverallState,
  input: InputState,
  output: OutputState,
})
  .addNode("node1", (state) => {
    return { foo: state.userInput + " name" };
  })
  .addNode("node2", (state) => {
    return { bar: state.foo + " is" };
  })
  .addNode(
    "node3",
    (state) => {
      return { graphOutput: state.bar + " Lance" };
    },
    { input: PrivateState },
  )
  .addEdge(START, "node1")
  .addEdge("node1", "node2")
  .addEdge("node2", "node3")
  .addEdge("node3", END)
  .compile();

const stream = await graph.streamEvents({ userInput: "My" }, { version: "v3" });
for await (const snapshot of stream.values) {
  console.log(snapshot);
}
// { userInput: 'My' }
// { foo: 'My name', userInput: 'My' }
// { foo: 'My name', userInput: 'My', bar: 'My name is' }            // <-- private channel
// { foo: 'My name', userInput: 'My', graphOutput: 'My name is Lance', bar: 'My name is' }

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

const stream = await graph.streamEvents(
  { userInput: "My" },
  { version: "v3", outputKeys: ["graphOutput"] }  
);
for await (const snapshot of stream.values) {
  console.log(snapshot);
}
// { graphOutput: 'My name is Lance' }

如果你只需要节点每一步实际产生的通道（而不是完整累积的状态），请改用 streamMode: "updates"。

Reducers

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

Reducer 参数

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

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

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

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

const newValue = reducer(currentState[key], nodeUpdate[key]); // 左参数、右参数

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

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 会调用：

append_strings(left=["draft"], right=["review"])  # 返回 ["draft", "review"]

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

import { ReducedValue, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  tags: new ReducedValue(
    z.array(z.string()).default(() => []),
    {
      reducer: (left: string[], right: string[]) => {
        // left：现有状态；right：来自节点的更新
        return left.concat(right);
      },
    }
  ),
});

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

const reducer = (left: string[], right: string[]) => left.concat(right);

reducer(["draft"], ["review"]); // left、right → ["draft", "review"]

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

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

默认 reducer

默认 reducer 会忽略左参数，并用右参数替换状态值。下面的示例展示了如何使用默认 reducer：

from typing_extensions import TypedDict

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

import { StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  foo: z.number(),
  bar: z.array(z.string()),
});

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

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

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

自定义 reducer

自定义 reducer 组合左参数和右参数，而不是替换状态值，这对于累积值非常有用，例如将更新追加到列表中。下面的示例展示了如何指定自定义 reducer：

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）指定 reducer 函数（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 键是通过将两个列表相加来更新的。

import { ReducedValue, StateSchema } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  foo: z.number(),
  bar: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});

在此示例中，我们使用了 ReducedValue 为第二个键（bar）指定 reducer 函数。请注意，第一个键保持不变。假设图的输入是 { foo: 1, bar: ["hi"] }。然后假设第一个 Node 返回 { foo: 2 }。这被视为对状态的更新。请注意，Node 不需要返回完整的 State schema——只需要一个更新。应用此更新后，State 将变为 { foo: 2, bar: ["hi"] }。如果第二个节点返回 { bar: ["bye"] }，那么 State 将变为 { foo: 2, bar: ["hi", "bye"] }。请注意，这里的 bar 键是通过连接两个数组来更新的。

Overwrite

TIP

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

未跟踪的值

UntrackedValue 用于那些在图的执行过程中应当存在，但绝不应该被检查点持久化的状态字段。当图从检查点恢复时，未跟踪的值会被重置为初始状态（或变得不可用）。

它适用于：

• 无法序列化的数据库连接
• 应在恢复时重建的临时缓存
• 你不想持久化的大型对象
• 每次都应全新传入的仅运行时配置

import { StateSchema, UntrackedValue, MessagesValue } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: MessagesValue,

  // Untracked：如果多个节点在同一一步写入，则抛出错误（guard: true 是默认值）
  dbConnection: new UntrackedValue<DatabaseConnection>(),

  // 带 guard: false 的 Untracked 允许多次写入，保留最后一个值
  tempCache: new UntrackedValue(
    z.record(z.string(), z.unknown()),
    { guard: false }
  ),

  // 不带 schema 的 Untracked（以实现最大灵活性）
  runtimeConfig: new UntrackedValue(),
});

行为：

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

WARNING

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

类型工具

LangGraph 提供了几个类型工具，用于在定义节点和条件边时获得更好的 TypeScript 类型安全性。

GraphNode

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

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

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

// 基本节点 - 接收状态，返回部分更新
const incrementNode: GraphNode<typeof State> = (state) => {
  return { count: state.count + 1 };
};

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

// 带 Command 路由的节点 - 指定有效的目标
const routerNode: GraphNode<typeof State, "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 属性，为标注节点类型提供了简写：

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

// 这两者是等价的：
const myNode1: GraphNode<typeof State> = (state) => ({ step: "done" });
const myNode2: typeof State.Node = (state) => ({ step: "done" });

ConditionalEdgeRouter

在条件边中使用 ConditionalEdgeRouter 作为路由函数（不更新状态，仅进行路由）：

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

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

// 路由函数返回节点名称或 END
const router: ConditionalEdgeRouter<typeof State, "process" | "summarize"> = (state) => {
  if (!state.shouldContinue) {
    return END;
  }
  return state.step === "initial" ? "process" : "summarize";
};

// 在图中使用
graph.addConditionalEdges("check", router);

StateSchema.State 和 StateSchema.Update

从 schema 中提取状态和更新类型，用于自定义类型定义：

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

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

// 提取完整的状态类型
type MyState = typeof MyStateSchema.State;
// { messages: BaseMessage[], count: number }

// 提取更新类型（部分，带有 reducer 输入类型）
type MyUpdate = typeof MyStateSchema.Update;
// { messages?: Messages, count?: number }

在图状态中使用消息

为什么使用消息？

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

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

在图中使用消息

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

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

在许多情况下，将之前的对话历史以消息列表的形式存储在图中是很有用的。为此，你可以使用预构建的 MessagesValue，它提供了一个感知消息的 reducer，可自动处理消息 ID、更新和删除。

MessagesValue reducer 对于告诉图在每次状态更新时如何更新状态中的 Message 对象列表至关重要。如果你不指定 reducer，每次状态更新都会用最近提供的值覆盖消息列表。MessagesValue 能正确处理这一点：对于全新消息，它会追加到现有列表；对于现有消息（按 ID 匹配），它会在原位置更新。

TIP

MessagesValue 实际上是 ReducedValue 的一种特殊情况，它预先配置了一个内部 messagesStateReducer，用于处理消息列表和更新。这为 LangGraph 图中的对话消息历史提供了便捷的、感知消息的状态管理。

序列化

除了跟踪消息 ID 之外，只要在 messages 通道上收到状态更新，add_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]

除了跟踪消息 ID 之外，只要在 messages 通道上收到状态更新，MessagesValue 也会尝试将消息反序列化为 LangChain Message 对象。这样可以按以下格式发送图输入/状态更新：

// 这是支持的
{
  messages: [new HumanMessage("message")];
}

// 这也被支持
{
  messages: [{ role: "human", content: "message" }];
}

由于使用 MessagesValue 时状态更新总是会被反序列化为 LangChain Messages，因此你应该使用点号记法来访问消息属性，例如 state.messages.at(-1).content。下面是使用 MessagesValue 的图示例：

import { StateGraph, StateSchema, MessagesValue } from "@langchain/langgraph";

const State = new StateSchema({
  messages: MessagesValue,
});

const graph = new StateGraph(State)
  ...

messages 字段被定义为一个 MessagesValue，它是带有内置 reducer 的 BaseMessage 对象列表。通常，需要跟踪的状态不止消息，因此我们看到人们会扩展此状态并添加更多字段，例如：

import { StateSchema, MessagesValue } from "@langchain/langgraph";
import * as z from "zod";

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

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

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

1. state——图的状态
2. config——一个 RunnableConfig 对象，包含诸如 thread_id 之类的配置信息以及诸如 tags 之类的追踪信息

你可以使用 addNode 方法向图中添加节点。为了获得更好的类型安全性，请使用 GraphNode 类型工具或 State.Node 为节点函数标注类型：

import { StateGraph, StateSchema, GraphNode } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  input: z.string(),
  results: z.string(),
});

// 方案 1：使用 GraphNode 类型工具
const myNode: GraphNode<typeof State> = (state, config) => {
  console.log("In node: ", config?.configurable?.user_id);
  return { results: `Hello, ${state.input}!` };
};

// 方案 2：使用 State.Node 简写
const otherNode: typeof State.Node = (state) => {
  return state;
};

const builder = new StateGraph(State)
  .addNode("myNode", myNode)
  .addNode("otherNode", otherNode)
  ...

在后台，函数会被转换为 RunnableLambda，为你的函数添加批处理和异步支持，以及原生追踪和调试。

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

builder.add_node(my_node)
# 然后你可以通过引用 `"my_node"` 来创建进出该节点的边

builder.addNode(myNode);
// 然后你可以通过引用 `"myNode"` 来创建进出该节点的边

重新执行与幂等性

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

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

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

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

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

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

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

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

在节点中使用 task

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

Original

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)

With task

from typing import NotRequired

import requests
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import task
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

class State(TypedDict):
urls: list[str]
results: NotRequired[list[str]]

@task
def _make_request(url: str):
"""Make a request."""
return requests.get(url).text[:100]  

def call_api(state: State):
"""Example node that makes API requests as checkpointed tasks."""
futures = [_make_request(url) for url in state["urls"]]  
results = [f.result() for f in futures]
return {"results": results}

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({"urls": ["https://www.example.com"]}, config)

Original

import * as z from "zod";

import {
END,
MemorySaver,
START,
StateGraph,
StateSchema,
} from "@langchain/langgraph";
import type { GraphNode } from "@langchain/langgraph";

const State = new StateSchema({
url: z.string(),
result: z.string().optional(),
});

const callApi: GraphNode<typeof State> = async (state) => {
const response = await fetch(state.url); 
const text = await response.text();
const result = text.slice(0, 100);
return { result };
};

const builder = new StateGraph(State)
.addNode("callApi", callApi)
.addEdge(START, "callApi")
.addEdge("callApi", END);

const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

const threadId = crypto.randomUUID();
const config = { configurable: { thread_id: threadId } };

await graph.invoke({ url: "https://www.example.com" }, config);

With task

import * as z from "zod";

import {
END,
MemorySaver,
START,
StateGraph,
StateSchema,
task,
} from "@langchain/langgraph";
import type { GraphNode } from "@langchain/langgraph";

const State = new StateSchema({
urls: z.array(z.string()),
results: z.array(z.string()).optional(),
});

const makeRequest = task("makeRequest", async (url: string) => {
const response = await fetch(url); 
const text = await response.text();
return text.slice(0, 100);
});

const callApi: GraphNode<typeof State> = async (state) => {
const pending = state.urls.map((url) => makeRequest(url)); 
const results = await Promise.all(pending);
return { results };
};

const builder = new StateGraph(State)
.addNode("callApi", callApi)
.addEdge(START, "callApi")
.addEdge("callApi", END);

const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

const threadId = crypto.randomUUID();
const config = { configurable: { thread_id: threadId } };

await graph.invoke({ urls: ["https://www.example.com"] }, config);

START 节点

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

from langgraph.graph import START

graph.add_edge(START, "node_a")

import { START } from "@langchain/langgraph";

graph.addEdge(START, "nodeA");

END 节点

END 节点是一个表示终点的特殊节点。当你想指明哪些边完成后没有后续动作时，会引用此节点。

from langgraph.graph import END

graph.add_edge("node_a", END)

import { END } from "@langchain/langgraph";

graph.addEdge("nodeA", END);

节点缓存

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

• 在编译图（或指定 entrypoint）时指定一个缓存
• 为节点指定缓存策略。每种缓存策略支持：
  • 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}}]

INFO

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. 第二次运行会利用缓存并快速返回。

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

• 在编译图（或指定 entrypoint）时指定一个缓存
• 为节点指定缓存策略。每种缓存策略支持：
  • keyFunc，用于根据节点的输入生成缓存键。
  • ttl，缓存的存活时间，以秒为单位。如果未指定，缓存将永不过期。

import { StateGraph, StateSchema, GraphNode, START } from "@langchain/langgraph";
import { InMemoryCache } from "@langchain/langgraph-checkpoint";
import { z } from "zod/v4";

const State = new StateSchema({
  x: z.number(),
  result: z.number(),
});

const expensiveNode: GraphNode<typeof State> = async (state) => {
  // 模拟一个开销大的操作
  await new Promise((resolve) => setTimeout(resolve, 3000));
  return { result: state.x * 2 };
};

const graph = new StateGraph(State)
  .addNode("expensive_node", expensiveNode, { cachePolicy: { ttl: 3 } })
  .addEdge(START, "expensive_node")
  .compile({ cache: new InMemoryCache() });

await graph.invoke({ x: 5 }, { streamMode: "updates" });   
// [{"expensive_node": {"result": 10}}]
await graph.invoke({ x: 5 }, { streamMode: "updates" });   
// [{"expensive_node": {"result": 10}, "__metadata__": {"cached": true}}]

边

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

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

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

WARNING

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

普通边

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

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

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

graph.addEdge("nodeA", "nodeB");

条件边

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

graph.add_conditional_edges("node_a", routing_function)

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

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

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

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

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

graph.addConditionalEdges("nodeA", routingFunction);

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

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

你可以选择提供一个对象，将 routingFunction 的输出映射到下一个节点的名称。

graph.addConditionalEdges("nodeA", routingFunction, {
  true: "nodeB",
  false: "nodeC",
});

TIP

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

入口点

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

from langgraph.graph import START

graph.add_edge(START, "node_a")

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

import { START } from "@langchain/langgraph";

graph.addEdge(START, "nodeA");

条件入口点

条件入口点允许你根据自定义逻辑从不同的节点开始。你可以使用从虚拟 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"})

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

import { START } from "@langchain/langgraph";

graph.addConditionalEdges(START, routingFunction);

你可以选择提供一个对象，将 routingFunction 的输出映射到下一个节点的名称。

graph.addConditionalEdges(START, routingFunction, {
  true: "nodeB",
  false: "nodeC",
});

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)

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

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

import { Send } from "@langchain/langgraph";

graph.addConditionalEdges("nodeA", (state) => {
  return state.subjects.map(
    (subject) => new Send("generateJoke", { subject })
  );
});

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。如果只需要路由而不更新状态，请改用条件边。

INFO

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

import { Command } from "@langchain/langgraph";

graph.addNode("myNode", (state) => {
  return new Command({
    update: { foo: "bar" },
    goto: "myOtherNode",
  });
});

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

import { Command } from "@langchain/langgraph";

graph.addNode("myNode", (state) => {
  if (state.foo === "bar") {
    return new Command({
      update: { foo: "baz" },
      goto: "myOtherNode",
    });
  }
});

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

在节点函数中使用 Command 时，你必须在添加节点时添加 ends 参数，以指定它可以路由到哪些节点：

builder.addNode("myNode", myNode, {
  ends: ["myOtherNode", END],
});

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

def my_node(state: State) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # 其中 `other_subgraph` 是父图中的一个节点
        graph=Command.PARENT
    )

INFO

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

当你从子图节点向父图节点发送更新时，如果更新的键同时存在于父图和子图的状态 schema 中，你必须在父图状态中为要更新的键定义一个reducer。请参阅此示例。

如果你使用子图，可以通过在 Command 中指定 graph: Command.PARENT，从子图内的节点导航到父图中的另一个节点：

import { Command } from "@langchain/langgraph";

graph.addNode("myNode", (state) => {
  return new Command({
    update: { foo: "bar" },
    goto: "otherSubgraph", // 其中 `otherSubgraph` 是父图中的一个节点
    graph: Command.PARENT,
  });
});

INFO

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

当你从子图节点向父图节点发送更新时，如果更新的键同时存在于父图和子图的状态 schema 中，你必须在父图状态中为要更新的键定义一个reducer。

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

作为 invoke 或 stream 的输入

WARNING

Command(resume=...) 是唯一用作 invoke()/stream() 输入的 Command 模式（可选地与 update=... 组合，以便在恢复的同时应用状态更改）。不要单独使用 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)  

WARNING

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

// 错误 - 图从最新的检查点恢复
// （最后运行的一步），看起来像是卡住了
await graph.invoke(new Command({ update: { messages: [{ role: "user", content: "follow up" }] } }), config);  

// 正确 - 普通对象从 __start__ 重新开始
await graph.invoke({ messages: [{ role: "user", content: "follow up" }] }, config);  

resume

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

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):
    # 暂停图并等待一个值
    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"}}

# 首次调用 - 触发中断并暂停
stream = graph.stream_events({"messages": []}, config, version="v3")
_ = stream.output  # 驱动流直到完成
print(stream.interrupts)

# 使用一个值恢复 - interrupt() 调用返回 "yes"
resumed = graph.stream_events(Command(resume="yes"), config, version="v3")
final = resumed.output

请参阅中断概念指南，了解中断模式的完整细节，包括多个中断和验证循环。

使用 new Command({ resume: ... }) 提供一个值，并在中断后恢复图执行。传给 resume 的值会成为被暂停节点内 interrupt() 调用的返回值：

import { Command, interrupt } from "@langchain/langgraph";

const humanReview = async (state: typeof StateAnnotation.State) => {
  // 暂停图并等待一个值
  const answer = interrupt("Do you approve?");
  return { messages: [{ role: "user", content: answer }] };
};

// 首次调用 - 触发中断并暂停
const result = await graph.invoke({ messages: [...] }, config);

// 使用一个值恢复 - interrupt() 调用返回 "yes"
const resumed = await graph.invoke(new Command({ resume: "yes" }), config);

请参阅中断概念指南，了解中断模式的完整细节，包括多个中断和验证循环。

从工具返回

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

WARNING

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

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

图迁移

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

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

TIP

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

运行时上下文

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

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

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

graph = StateGraph(State, context_schema=ContextSchema)

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

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

import { StateGraph, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  input: z.string(),
  output: z.string(),
});

const ContextSchema = z.object({
  llm: z.union([z.literal("openai"), z.literal("anthropic")]),
});

const graph = new StateGraph(State, ContextSchema);

然后你可以使用 context 属性将该配置传入图中。

const config = { context: { llm: "anthropic" } };

await graph.invoke(inputs, config);

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

from langgraph.runtime import Runtime

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

import { Runtime, GraphNode } from "@langchain/langgraph";
import * as z from "zod";

const nodeA: GraphNode<typeof State> = (state, config) => {
  const llm = getLLM(runtime.context?.llm);
  // ...
  return {};
};

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

graph.addNode("myNode", (state, config) => {
  const llmType = config.context?.llm || "openai";
  const llm = getLLM(llmType);
  return { results: `Hello, ${state.input}!` };
});

递归限制

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

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

请阅读递归限制了解更多关于递归限制工作原理的内容。

递归限制设置了图在单次执行中最多可以执行的超级步骤数量。一旦达到该限制，LangGraph 将抛出 GraphRecursionError。默认情况下，该值设置为 25 步。递归限制可以在运行时在任何图上设置，并通过 config 对象传递给 invoke/stream。重要的是，recursionLimit 是一个独立的 config 键，不应像所有其他用户自定义配置那样放在 configurable 键内部。请参阅下面的示例：

await graph.invoke(inputs, {
  recursionLimit: 5,
  context: { llm: "anthropic" },
});

访问和处理递归计数器

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

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

工作原理

步计数器存储在 config["metadata"]["langgraph_step"] 中。LangGraph 在图执行时递增该计数器，一旦超过配置的 recursion_limit 就会抛出 GraphRecursionError。

步计数器存储在 config.metadata.langgraph_step 中。LangGraph 在图执行时递增该计数器，一旦超过配置的 recursionLimit 就会抛出 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

import { RunnableConfig } from "@langchain/core/runnables";
import { StateGraph } from "@langchain/langgraph";

const myNode: GraphNode<typeof State> = async (state, config) => {
  const currentStep = config.metadata?.langgraph_step;
  console.log(`Currently on step: ${currentStep}`);
  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"]:
    """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"]}

# 构建图
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})

为你的图设计显式的终止条件，并捕获 GraphRecursionError 作为安全网：

import {
  StateGraph,
  StateSchema,
  ReducedValue,
  GraphNode,
  ConditionalEdgeRouter,
  END,
  GraphRecursionError
} from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});

// 使用显式终止逻辑构建图
const graph = new StateGraph(State)
  .addNode("reasoning", async (state) => {
    // 正常处理 - 为你的图设计显式终止条件
    return {
      messages: ["thinking..."]
    };
  })
  .addConditionalEdges("reasoning", (state) => {
    // 在这里添加你的终止条件
    if (state.messages.length >= 5) {
      return END;
    }
    return "reasoning";
  });

const app = graph.compile();

// 捕获 GraphRecursionError 作为安全网
try {
  const result = await app.invoke(
    { messages: [] },
    { recursionLimit: 10 }
  );
} catch (error) {
  if (error instanceof GraphRecursionError) {
    console.log("Recursion limit reached, handling gracefully");
    // 处理错误 - 返回部分结果、通知用户等
  }
}

主动与被动方法

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

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:
    """Proactively monitor and handle recursion within the graph"""
    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"]}

import {
  StateGraph,
  StateSchema,
  ReducedValue,
  GraphNode,
  ConditionalEdgeRouter,
  END,
  GraphRecursionError
} from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});

// 使用显式终止逻辑构建图
const builder = new StateGraph(State)
  .addNode("agent", async (state) => {
    return {
      messages: ["Processing..."]
    };
  })
  .addConditionalEdges("agent", (state) => {
    // 为你的图设计终止条件
    if (state.messages.length >= 5) {
      return END;
    }
    return "agent";
  });

const graph = builder.compile();

// 被动方法 - 捕获 GraphRecursionError 作为安全网
try {
  const result = await graph.invoke(
    { messages: [] },
    { recursionLimit: 10 }
  );
} catch (error) {
  if (error instanceof GraphRecursionError) {
    // 在图执行失败后在外部处理
    console.log("Recursion limit exceeded, handling gracefully");
  }
}

这两种方法的关键区别是：

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

主动方法的优点：

• 在图内进行优雅降级
• 可以在检查点中保存中间状态
• 使用部分结果时用户体验更好
• 图正常完成（无异常）

被动方法的优点：

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

被动方法会在超过限制后捕获 GraphRecursionError。为你的图设计显式的终止条件，从一开始就避免达到限制。

方法	检测	处理	控制流
被动（捕获 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_step 之外，config.metadata 中还提供以下元数据：

const inspectMetadata: GraphNode<typeof State> = async (state, config) => {
  const metadata = config.metadata;

  console.log(`Step: ${metadata?.langgraph_step}`);
  console.log(`Node: ${metadata?.langgraph_node}`);
  console.log(`Triggers: ${metadata?.langgraph_triggers}`);
  console.log(`Path: ${metadata?.langgraph_path}`);
  console.log(`Checkpoint NS: ${metadata?.langgraph_checkpoint_ns}`);

  return state;
}

可视化

能够可视化图通常很不错，尤其是在图变得复杂时。LangGraph 内置了多种可视化图的方法。更多信息请参阅可视化你的图。

可观测性与追踪

要对你的智能体进行追踪、调试和评估，请使用 LangSmith。

了解更多

• 如何使用 Graph API
• Functional API 概念概述
• 在 Graph API 与 Functional API 之间选择