Graph API 概述 - Docs by LangChain中文

图

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

State：一个共享的数据结构，代表应用程序的当前快照。它可以是任何数据类型，但通常使用共享状态模式来定义。
Nodes：编码智能体逻辑的函数。它们接收当前状态作为输入，执行一些计算或副作用，并返回一个更新的状态。
Edges：根据当前状态决定接下来执行哪个 Node 的函数。它们可以是条件分支或固定转换。

通过组合 Nodes 和 Edges，您可以创建复杂的、循环的工作流，这些工作流会随时间推移演变状态。然而，真正的威力来自于 LangGraph 如何管理该状态。需要强调的是：Nodes 和 Edges 不过是函数——它们可以包含一个大语言模型（LLM），也可以只是普通的代码。简而言之：节点执行工作，边指示下一步做什么。 LangGraph 底层的图算法使用消息传递来定义一个通用程序。当一个节点完成其操作时，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行它们的函数，将结果消息传递给下一组节点，这个过程持续进行。该程序受到 Google 的 Pregel 系统的启发，以离散的”超步骤”进行。一个超步骤可以被视为对图节点的一次单次迭代。并行运行的节点属于同一个超步骤，而顺序运行的节点则属于不同的超步骤。在图执行开始时，所有节点都处于 inactive 状态。当一个节点在其任何传入边（或”通道”）上接收到新消息（状态）时，它就会变为 active 状态。然后，活动节点运行其函数并返回更新。在每个超步骤结束时，没有传入消息的节点通过将自己标记为 inactive 来投票”停止”。当所有节点都处于 inactive 状态且没有消息在传递时，图执行终止。

StateGraph

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

编译您的图

要构建您的图，您首先定义状态，然后添加节点和边，最后编译它。编译您的图具体是什么意思？为什么需要它？编译是一个相当简单的步骤。它对图的结构进行一些基本检查（例如，没有孤立的节点等）。这也是您可以指定运行时参数（如检查点器和断点）的地方。您只需调用 .compile 方法即可编译图：

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

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

State

定义图时，您要做的第一件事就是定义图的 State。State 由图的模式和 reducer 函数组成，后者指定了如何将更新应用到状态。State 的模式将是图中所有 Nodes 和 Edges 的输入模式。您可以使用 StateSchema 类定义状态，该类接受任何标准模式（如 Zod）用于单个字段，以及特殊值类型，如 ReducedValue 和 MessagesValue。所有 Nodes 将发出对 State 的更新，然后使用指定的 reducer 函数应用这些更新。

Schema

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

一个用于简单字段的标准模式（变成一个在更新时覆盖的”最后值”通道）
一个 ReducedValue，用于需要自定义 reducer 函数的字段（当节点并行运行时）
一个 MessagesValue，用于聊天消息列表（预构建了消息感知的 reducer）
一个 UntrackedValue，用于不应被检查点的瞬态状态

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

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

  // 简单字段直接使用 Zod 模式
  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();

默认情况下，图将具有相同的输入和输出模式。如果您想更改这一点，也可以直接指定显式的输入和输出模式。当您有很多键，并且某些键明确用于输入，另一些用于输出时，这很有用。

多个模式

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

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

可以让节点在图中写入私有状态通道，用于内部节点通信。我们可以简单地定义一个私有模式 PrivateState。也可以为图定义显式的输入和输出模式。在这些情况下，我们定义一个包含与图操作相关的所有键的”内部”模式。但是，我们也定义 input 和 output 模式，它们是”内部”模式的子集，以约束图的输入和输出。有关更多详细信息，请参阅定义输入和输出模式。让我们看一个例子：

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

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

我们将 state 作为输入模式传递给 node1。但是，我们写入了 foo，这是 OverallState 中的一个通道。我们如何写入一个未包含在输入模式中的状态通道？这是因为节点_可以写入图状态中的任何状态通道。_图状态是在初始化时定义的状态通道的并集，其中包括 OverallState 以及过滤后的 InputState 和 OutputState。
我们使用 StateGraph({ state: OverallState, input: InputState, output: OutputState }) 初始化图。我们如何在 node2 中写入 PrivateState？如果它没有在 StateGraph 初始化时传递，图是如何获得对这个模式的访问权限的？我们能做到这一点是因为_节点还可以声明额外的状态通道_，只要状态模式定义存在即可。在这种情况下，PrivateState 模式已经定义，因此我们可以将 bar 添加为图中的新状态通道并写入其中。

流式传输时私有通道不会被编辑。输入、输出和私有模式约束了每个节点_读取_的内容（其输入模式）以及 invoke _返回_的内容（输出模式）。它们不会对 stream 隐藏通道。当您使用 streamMode: "values" 进行流式传输时，图默认会发出其所有状态通道——包括私有通道——因为值流式传输默认使用完整状态通道集，而不是输出模式。这就是为什么像 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 + " 名字" };
  })
  .addNode("node2", (state) => {
    return { bar: state.foo + " 是" };
  })
  .addNode(
    "node3",
    (state) => {
      return { graphOutput: state.bar + " 兰斯" };
    },
    { input: PrivateState },
  )
  .addEdge(START, "node1")
  .addEdge("node1", "node2")
  .addEdge("node2", "node3")
  .addEdge("node3", END)
  .compile();

const stream = await graph.streamEvents({ userInput: "我的" }, { version: "v3" });
for await (const snapshot of stream.values) {
  console.log(snapshot);
}
// { userInput: '我的' }
// { foo: '我的 名字', userInput: '我的' }
// { foo: '我的 名字', userInput: '我的', bar: '我的 名字 是' }            // <-- 私有通道
// { foo: '我的 名字', userInput: '我的', graphOutput: '我的 名字 是 兰斯', bar: '我的 名字 是' }

要将流式传输的值限制在特定的一组通道（例如仅输出模式），请传递 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，并将返回值保存为新的状态值：

const newValue = reducer(currentState[key], nodeUpdate[key]); // left, right

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

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"]。自定义 reducers 组合左参数和右参数。默认 reducer 丢弃左参数，只保留右参数。

默认 reducer

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

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 模式——只需返回一个更新。应用此更新后，State 将变为 { foo: 2, bar: ["hi"] }。如果第二个节点返回 { bar: ["bye"] }，那么 State 将变为 { foo: 2, bar: ["bye"] }

自定义 reducers

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

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 模式——只需返回一个更新。应用此更新后，State 将变为 { foo: 2, bar: ["hi"] }。如果第二个节点返回 { bar: ["bye"] }，那么 State 将变为 { foo: 2, bar: ["hi", "bye"] }。请注意，这里的 bar 键是通过将两个数组连接在一起来更新的。

未跟踪的值

UntrackedValue 用于在图执行期间存在但永远不应被检查点的状态字段。当图从检查点恢复时，未跟踪的值将重置为其初始状态（或不可用）。这对于以下情况很有用：

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

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

const State = new StateSchema({
  messages: MessagesValue,

  // 未跟踪：如果多个节点在同一步骤中写入则抛出异常（guard: true 是默认值）
  dbConnection: new UntrackedValue<DatabaseConnection>(),

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

  // 没有模式的未跟踪（为了最大灵活性）
  runtimeConfig: new UntrackedValue(),
});

行为：

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

不要将 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`

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

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 响应）。要了解更多关于消息对象的信息，请参阅消息概念指南。

在图中使用消息

在许多情况下，将先前的对话历史作为消息列表存储在图状态中是有帮助的。为此，您可以使用预构建的 MessagesValue，它提供了一个消息感知的 reducer，可自动处理消息 ID、更新和删除。 MessagesValue reducer 对于告诉图如何在每次状态更新时更新状态中的 Message 对象列表至关重要。如果您不指定 reducer，每次状态更新都会用最近提供的值覆盖消息列表。MessagesValue 正确地处理了这一点：对于全新的消息，它追加到现有列表；对于现有消息（通过 ID 匹配），它就地更新它们。

MessagesValue 实际上是 ReducedValue 的一个特例，预配置了一个内部的 messagesStateReducer 来处理消息列表和更新。这为 LangGraph 图中的聊天消息历史提供了方便、消息感知的状态管理。

序列化

除了跟踪消息 ID 之外，MessagesValue 还会尝试在 messages 通道上接收到状态更新时，将消息反序列化为 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，它是一个 BaseMessage 对象的列表，带有内置的 reducer。通常，除了消息之外，还有更多状态需要跟踪，所以我们看到人们扩展这个状态并添加更多字段，例如：

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

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

Nodes

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

state—图的状态
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.addNode(myNode);
// 然后您可以通过引用 `"myNode"` 来创建指向/来自此节点的边

重新执行和幂等性

当您使用检查点器进行编译时，LangGraph 在超步骤边界保存检查点，而不是在节点函数内部保存。如果执行停止并稍后恢复（例如在中断或重试之后），受影响的节点将从其函数的开头重新运行。暂停前的代码和副作用会再次运行。 幂等性。 设计节点逻辑时，确保重新执行不会破坏状态。如果一个节点插入了一行数据库记录，运行它两次不应该创建重复的行，除非这是有意为之的。使用幂等键、upsert 或读写检查。有关 interrupt() 周围的副作用，请参阅调用 interrupt 之前的副作用必须是幂等的。 图更改。 关于代码更改的确定性规则不适用于图结构。您可以添加或删除节点和边，而不会中断现有线程的恢复。恢复运行使用保存的状态，并执行您现在编译的任何图。 节点内的任务和中断。 如果一个节点调用任务或 interrupt，则在恢复时应用更严格的确定性规则。LangGraph 从检查点器恢复已完成的任务结果，但在恢复点之前更改代码中的任务或 interrupt 顺序可能导致缓存值不匹配。功能 API 入口点编译为单个节点，以这种方式运行整个入口点方法。请参阅确定性、幂等性以及在节点中使用任务。

在节点中使用任务

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

原始
带任务

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

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 节点是一个特殊的节点，表示将用户输入发送到图的节点。引用此节点的主要目的是确定首先应调用哪些节点。

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

graph.addEdge(START, "nodeA");

`END` 节点

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

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

graph.addEdge("nodeA", END);

节点缓存

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

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

Edges

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

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

一个节点可以有多个传出边。如果一个节点有多个传出边，所有这些目标节点都将在下一个超步骤中并行执行。

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

普通边

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

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

条件边

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

graph.addConditionalEdges("nodeA", routingFunction);

类似于节点，routingFunction 接受图的当前 state 并返回一个值。默认情况下，routingFunction 的返回值用作将状态发送到的下一个节点（或节点列表）的名称。所有这些节点都将在下一个超步骤中并行运行。您可以选择提供一个对象，将 routingFunction 的输出映射到下一个节点的名称。

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

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

入口点

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

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

graph.addEdge(START, "nodeA");

条件入口点

条件入口点允许您根据自定义逻辑从不同的节点开始。您可以使用来自虚拟 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 接受两个参数：第一个是节点的名称，第二个是传递给该节点的状态。

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 以在单个步骤中更新状态并路由到下一个节点：

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],
});

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，从子图中的节点导航到父图中的不同节点：

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

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

将 graph 设置为 Command.PARENT 将导航到最近的父图。当您将更新从子图节点发送到父图节点以更新一个父图和子图状态模式共享的键时，您必须在父图状态中为您要更新的键定义一个 reducer。

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

作为 `invoke` 或 `stream` 的输入

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`

使用 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 在工具完成后路由到特定节点。

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

详情请参考在工具内使用。

图迁移

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

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

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

运行时上下文

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

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

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

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。默认情况下，此值设置为 25 步。递归限制可以在运行时在任何图上设置，并通过配置对象传递给 invoke/stream。重要的是，recursionLimit 是一个独立的 config 键，不应像所有其他用户定义的配置一样传递到 configurable 键内。请参阅下面的示例：

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

访问和处理递归计数器

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

工作原理

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

访问当前步骤计数器

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

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

设计您的图时包含明确的终止条件，并捕获 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");
    // 处理错误 - 返回部分结果、通知用户等。
  }
}

主动方法与被动方法

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

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");
  }
}

被动方法在超过限制后捕获 GraphRecursionError。设计您的图时包含明确的终止条件，以首先避免达到限制。

方法	检测	处理	控制流
被动（捕获 `GraphRecursionError`）	超过限制之后	在 try/catch 的图外部	图执行终止

被动方法的优势：

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

其他可用的元数据

除了 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。

了解更多

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

​图

​StateGraph

​编译您的图

​State

​Schema

​多个模式

​Reducers

​Reducer 参数

​默认 reducer

​自定义 reducers

​未跟踪的值

​类型工具

​GraphNode

​State.Node 简写

​ConditionalEdgeRouter

​StateSchema.State 和 StateSchema.Update

​在图状态中处理消息

​为什么使用消息？

​在图中使用消息

​序列化

​Nodes

​重新执行和幂等性

​在节点中使用任务

​START 节点

​END 节点

​节点缓存

​Edges

​普通边

​条件边

​入口点

​条件入口点

​Send

​Command

​从节点返回

​update 和 goto

​graph

​作为 invoke 或 stream 的输入

​resume

​从工具返回

​图迁移

​运行时上下文

​递归限制

​访问和处理递归计数器

​工作原理

​访问当前步骤计数器

​主动方法与被动方法

​其他可用的元数据

​可视化

​可观测性和追踪

​了解更多

图

StateGraph

编译您的图

State

Schema

多个模式

Reducers

Reducer 参数

默认 reducer

自定义 reducers

未跟踪的值

类型工具

`GraphNode`

`State.Node` 简写

`ConditionalEdgeRouter`

`StateSchema.State` 和 `StateSchema.Update`

在图状态中处理消息

为什么使用消息？

在图中使用消息

序列化

Nodes

重新执行和幂等性

在节点中使用任务

`START` 节点

`END` 节点

节点缓存

Edges

普通边

条件边

入口点

条件入口点

`Send`

`Command`

从节点返回

`update` 和 `goto`

`graph`

作为 `invoke` 或 `stream` 的输入

`resume`

从工具返回

图迁移

运行时上下文

递归限制

访问和处理递归计数器

工作原理

访问当前步骤计数器

主动方法与被动方法

其他可用的元数据

可视化

可观测性和追踪

了解更多