Graph API 概览 - Docs by LangChain中文

图

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

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

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

StateGraph

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

编译你的图

要构建你的图，你首先定义状态，然后添加节点和边，接着再编译它。编译你的图到底意味着什么，为什么需要这样做？编译是一个相当简单的步骤。它会对你的图结构进行一些基本的检查（例如没有孤立节点等）。在这里你也可以指定运行时参数，如检查点和断点。你只需调用 .compile 方法来编译你的图：

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

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

状态

当你定义一个图时，首先要做的是定义图的 State。State 由图的模式和 reducer 函数组成，后者指定了如何对状态应用更新。State 的模式将是图中所有 Nodes 和 Edges 的输入模式。你使用 StateSchema 类来定义状态，该类接受任何标准模式（如 Zod）用于单个字段，以及像 ReducedValue 和 MessagesValue 这样的特殊值类型。所有 Nodes 都将发出对 State 的更新，这些更新随后会使用指定的 reducer 函数进行应用。

模式

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

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

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

const AgentState = new StateSchema({
  // 内置消息值的预构建消息值
  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 { StateSchema, GraphNode } 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 作为新的状态通道并向其写入。

私有通道在流式传输时不会被编辑（redacted）。输入、输出和私有模式限制了每个节点读取的内容（其输入模式）以及 invoke 返回的内容（输出模式）。它们不会对 stream 隐藏通道。当你使用 streamMode: "values" 进行流式传输时，图默认会发出所有状态通道——包括私有通道——因为 values 流默认使用完整的通道集合，而不是输出模式。这就是为什么像 bar 这样的私有通道会被 invoke 隐藏，但在流式传输时可见：

for await (const chunk of await graph.stream(
  { userInput: "My" },
  { streamMode: "values" }
)) {
  console.log(chunk);
}
// { userInput: 'My' }
// { userInput: 'My', foo: 'My name' }
// { userInput: 'My', foo: 'My name', bar: 'My name is' }            // <-- 私有通道
// { userInput: 'My', foo: 'My name', bar: 'My name is', graphOutput: 'My name is Lance' }

要将流式传输的值限制到特定的通道集合（例如，仅输出模式），可以传入 outputKeys：

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

如果你只需要每个步骤中节点实际生成的通道（而不是完整的累积状态），可以改用 streamMode: "updates"。

Reducers

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

默认 reducer

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

Example A

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

Example B

import { StateSchema, ReducedValue } 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 外，每当 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()),
});

节点

在 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 { v7 as uuid7 } from "uuid";
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 = uuid7();
const config = { configurable: { thread_id: threadId } };

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

import { v7 as uuid7 } from "uuid";
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 = uuid7();
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}}]

边

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

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

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

对于每个节点，只选择一种路由机制：使用普通边进行静态路由，或使用条件边 / 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");

条件入口点

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

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 模式。不要使用 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: ["思考中..."]
    };
  })
  .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("达到递归限制，正在优雅地处理");
    // 处理错误 - 返回部分结果，通知用户等。
  }
}

主动式 vs. 响应式方法

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

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: ["处理中..."]
    };
  })
  .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("达到递归限制，正在优雅地处理");
  }
}

响应式方法在超过限制后捕获 GraphRecursionError。设计你的图时要有明确的终止条件，以避免一开始就达到限制。

方法	检测时机	处理方式	控制流
响应式（捕获 `GraphRecursionError`）	超过限制之后	在图外的 try/catch 中	图执行终止

响应式的优势：

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

其他可用的元数据

除了 langgraph_step，config.metadata 中还提供了以下元数据：

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

  console.log(`步骤: ${metadata?.langgraph_step}`);
  console.log(`节点: ${metadata?.langgraph_node}`);
  console.log(`触发器: ${metadata?.langgraph_triggers}`);
  console.log(`路径: ${metadata?.langgraph_path}`);
  console.log(`检查点命名空间: ${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

​编译你的图

​状态

​模式

​多种模式

​Reducers

​默认 reducer

​非跟踪值

​类型工具

​GraphNode

​State.Node 简写

​ConditionalEdgeRouter

​StateSchema.State 和 StateSchema.Update

​在图状态中处理消息

​为什么要使用消息？

​在你的图中使用消息

​序列化

​节点

​重新执行和幂等性

​在节点中使用任务

​START 节点

​END 节点

​节点缓存

​边

​普通边

​条件边

​入口点

​条件入口点

​Send

​Command

​从节点返回

​update 和 goto

​graph

​作为 invoke 或 stream 的输入

​resume

​从工具返回

​图迁移

​运行时上下文

​递归限制

​访问和处理递归计数器

​它是如何工作的

​访问当前步骤计数器

​主动式 vs. 响应式方法

​其他可用的元数据

​可视化

​可观察性与追踪

​了解更多

图

StateGraph

编译你的图

状态

模式

多种模式

Reducers

默认 reducer

非跟踪值

类型工具

`GraphNode`

`State.Node` 简写

`ConditionalEdgeRouter`

`StateSchema.State` 和 `StateSchema.Update`

在图状态中处理消息

为什么要使用消息？

在你的图中使用消息

序列化

节点

重新执行和幂等性

在节点中使用任务

`START` 节点

`END` 节点

节点缓存

边

普通边

条件边

入口点

条件入口点

`Send`

`Command`

从节点返回

`update` 和 `goto`

`graph`

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

`resume`

从工具返回

图迁移

运行时上下文

递归限制

访问和处理递归计数器

它是如何工作的

访问当前步骤计数器

主动式 vs. 响应式方法

其他可用的元数据

可视化

可观察性与追踪

了解更多