使用图 API

本指南演示了 LangGraph 图 API 的基础知识。它将引导你了解状态，以及如何构建常见的图结构，如序列、分支和循环。它还涵盖了 LangGraph 的控制功能，包括用于 map-reduce 工作流的 Send API，以及用于将状态更新与跨节点的“跳转”相结合的 Command API。

设置

安装 langgraph：

npm install @langchain/langgraph

设置 LangSmith 以便更好地调试注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 让你能够使用跟踪数据来调试、测试和监控你使用 LangGraph 构建的 LLM 应用程序——阅读文档了解更多入门信息。

定义和更新状态

这里我们将展示如何在 LangGraph 中定义和更新状态。我们将演示：

如何使用状态来定义图的架构
如何使用归约器来控制状态更新的处理方式。

定义状态

LangGraph 中的状态使用 StateSchema 类定义。它提供了一个统一的 API，可以接受字段的标准架构（如 Zod），以及特殊值类型如 ReducedValue、MessagesValue 和 UntrackedValue。默认情况下，图将具有相同的输入和输出架构，而状态决定了该架构。有关如何定义不同的输入和输出架构，请参阅定义输入和输出架构。让我们考虑一个使用消息的简单例子。这代表了许多 LLM 应用程序中一种通用的状态表示方式。更多详情请参阅我们的概念页面。

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

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

此状态跟踪一个消息对象列表，以及一个额外的整数字段。

更新状态

让我们构建一个包含单个节点的示例图。我们的节点只是一个 TypeScript 函数，它读取图的状态并对其进行更新。此函数的第一个参数始终是状态：

import { AIMessage } from "@langchain/core/messages";
import { GraphNode } from "@langchain/langgraph";

const node: GraphNode<typeof State> = (state) => {
  const messages = state.messages;
  const newMessage = new AIMessage("Hello!");
  return { messages: [newMessage], extraField: 10 };
};

这个节点只是简单地向我们的消息列表追加一条消息（归约器负责处理拼接），并填充一个额外字段。

节点应直接返回对状态的更新，而不是直接修改状态本身。

接下来，让我们定义一个包含此节点的简单图。我们使用 StateGraph 来定义一个在此状态上操作的图。然后使用 addNode 来填充我们的图。

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

const graph = new StateGraph(State)
  .addNode("node", node)
  .addEdge("__start__", "node")
  .compile();

LangGraph 提供了内置工具来可视化你的图。让我们检查一下我们的图。有关可视化的详情，请参阅可视化你的图。

import * as fs from "node:fs/promises";

const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());

await fs.writeFile("graph.png", imageBuffer);

在这种情况下，我们的图只执行一个节点。让我们进行一次简单的调用：

import { HumanMessage } from "@langchain/core/messages";

const result = await graph.invoke({ messages: [new HumanMessage("Hi")], extraField: 0 });
console.log(result);

{ messages: [HumanMessage { content: 'Hi' }, AIMessage { content: 'Hello!' }], extraField: 10 }

请注意：

我们通过更新状态的单个键来启动调用。
我们在调用结果中接收整个状态。

为方便起见，我们通常通过日志记录来检查消息对象的内容：

for (const message of result.messages) {
  console.log(`${message.getType()}: ${message.content}`);
}

human: Hi
ai: Hello!

使用归约器处理状态更新

状态中的每个键都可以拥有自己独立的归约器函数，该函数控制来自节点的更新如何被应用。如果没有显式指定归约器函数，则假定对该键的所有更新都应覆盖它。在前面的例子中，我们使用了 MessagesValue，它已经内置了一个归约器。对于自定义字段，你可以使用 ReducedValue 来定义如何应用更新。在前面的例子中，我们的节点通过向 "messages" 键追加一条消息来更新状态。MessagesValue 归约器会自动处理：

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

// MessagesValue 已经内置了归约器
const State = new StateSchema({
  messages: MessagesValue,
  extraField: z.number(),
});

我们的节点可以简单地返回新消息（归约器处理拼接）：

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

const node: GraphNode<typeof State> = (state) => {
  const newMessage = new AIMessage("Hello!");
  return { messages: [newMessage], extraField: 10 };
};

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

const graph = new StateGraph(State)
  .addNode("node", node)
  .addEdge(START, "node")
  .compile();

const result = await graph.invoke({ messages: [new HumanMessage("Hi")] });

for (const message of result.messages) {
  console.log(`${message.getType()}: ${message.content}`);
}

human: Hi
ai: Hello!

MessagesValue

在实践中，更新消息列表还有其他注意事项：

我们可能希望更新状态中的现有消息。
我们可能希望接受消息格式的简写形式，例如 OpenAI 格式。

LangGraph 包含内置的 MessagesValue 来处理这些考虑：

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

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

const node: GraphNode<typeof State> = (state) => {
  const newMessage = new AIMessage("Hello!");
  return { messages: [newMessage], extraField: 10 };
};

const graph = new StateGraph(State)
  .addNode("node", node)
  .addEdge(START, "node")
  .compile();

const inputMessage = { role: "user", content: "Hi" };

const result = await graph.invoke({ messages: [inputMessage] });

for (const message of result.messages) {
  console.log(`${message.getType()}: ${message.content}`);
}

human: Hi
ai: Hello!

对于涉及聊天模型的应用程序来说，这是一种通用的状态表示。LangGraph 为了方便起见，包含了预构建的 MessagesValue，因此我们可以这样写：

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

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

定义输入和输出架构

默认情况下，StateGraph 使用单一架构操作，并且所有节点都预期使用该架构进行通信。然而，也可以为图定义不同的输入和输出架构。当指定了不同的架构时，内部仍会使用一个架构在节点之间通信。输入架构确保提供的输入符合预期的结构，而输出架构则过滤内部数据，仅根据定义的输出架构返回相关信息。下面，我们将看到如何定义不同的输入和输出架构。

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

// 定义输入的架构
const InputState = new StateSchema({
  question: z.string(),
});

// 定义输出的架构
const OutputState = new StateSchema({
  answer: z.string(),
});

// 定义整体架构，结合输入和输出
const OverallState = new StateSchema({
  question: z.string(),
  answer: z.string(),
});

// 定义处理输入的节点
const answerNode: GraphNode<typeof OverallState> = (state) => {
  // 示例答案和一个额外的键
  return { answer: "bye", question: state.question };
};

// 使用指定的输入和输出架构构建图
const graph = new StateGraph({
  input: InputState,
  output: OutputState,
  state: OverallState,
})
  .addNode("answerNode", answerNode)
  .addEdge(START, "answerNode")
  .addEdge("answerNode", END)
  .compile();

// 使用输入调用图并打印结果
console.log(await graph.invoke({ question: "hi" }));

{ answer: 'bye' }

请注意，invoke 的输出只包含输出架构。

在节点之间传递私有状态

在某些情况下，你可能希望节点交换对中间逻辑至关重要但不需要成为图主架构一部分的信息。这些私有数据与图的总体输入/输出无关，并且只应在特定节点之间共享。下面，我们将创建一个由三个节点（node_1、node_2 和 node_3）组成的顺序图示例，其中私有数据在前两个步骤（node_1 和 node_2）之间传递，而第三步（node_3）只能访问公共的整体状态。

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

// 图的整体状态（这是节点间共享的公共状态）
const OverallState = new StateSchema({
  a: z.string(),
});

// node1 的输出包含不属于整体状态的私有数据
const Node1Output = new StateSchema({
  privateData: z.string(),
});

// Node 2 输入只请求 node1 之后可用的私有数据
const Node2Input = new StateSchema({
  privateData: z.string(),
});

// 私有数据仅在 node1 和 node2 之间共享
const node1: GraphNode<typeof OverallState> = (state) => {
  const output = { privateData: "set by node1" };
  console.log(`Entered node 'node1':\n\tInput: ${JSON.stringify(state)}.\n\tReturned: ${JSON.stringify(output)}`);
  return output;
};

const node2: GraphNode<typeof Node2Input> = (state) => {
  const output = { a: "set by node2" };
  console.log(`Entered node 'node2':\n\tInput: ${JSON.stringify(state)}.\n\tReturned: ${JSON.stringify(output)}`);
  return output;
};

// Node 3 只能访问整体状态（无法访问 node1 的私有数据）
const node3: GraphNode<typeof OverallState> = (state) => {
  const output = { a: "set by node3" };
  console.log(`Entered node 'node3':\n\tInput: ${JSON.stringify(state)}.\n\tReturned: ${JSON.stringify(output)}`);
  return output;
};

// 按顺序连接节点
// node2 接受来自 node1 的私有数据，而
// node3 看不到私有数据。
const graph = new StateGraph(OverallState)
  .addNode("node1", node1)
  .addNode("node2", node2, { input: Node2Input })
  .addNode("node3", node3)
  .addEdge(START, "node1")
  .addEdge("node1", "node2")
  .addEdge("node2", "node3")
  .addEdge("node3", END)
  .compile();

// 使用初始状态调用图
const response = await graph.invoke({ a: "set at start" });

console.log(`\nOutput of graph invocation: ${JSON.stringify(response)}`);

Entered node 'node1':
	Input: {"a":"set at start"}.
	Returned: {"privateData":"set by node1"}
Entered node 'node2':
	Input: {"privateData":"set by node1"}.
	Returned: {"a":"set by node2"}
Entered node 'node3':
	Input: {"a":"set by node2"}.
	Returned: {"a":"set by node3"}

Output of graph invocation: {"a":"set by node3"}

替代状态定义

虽然 StateSchema 是定义状态的推荐方法，但 LangGraph 也支持其他几种方法。本节涵盖所有可用选项。

通道 API

通道 API 提供了对状态管理的底层控制。LangGraph 提供了几种内置通道类型：

通道类型	行为	用例
`LastValue`	存储最近的值	会被覆盖的简单字段
`BinaryOperatorAggregate`	使用归约器函数组合值	累积值（计数器、列表）
`Topic`	将所有值收集到序列中	事件流、审计日志
`EphemeralValue`	在超步之间重置的值	临时计算状态

使用对象简写： 当你传递一个包含 reducer 和 default 的对象时，它会创建一个 BinaryOperatorAggregate 通道。传递 null 会创建一个 LastValue 通道：

import { BaseMessage } from "@langchain/core/messages";
import { StateGraph } from "@langchain/langgraph";

interface WorkflowState {
  messages: BaseMessage[];
  question: string;
  answer: string;
}

const workflow = new StateGraph<WorkflowState>({
  channels: {
    // BinaryOperatorAggregate：使用归约器组合值
    messages: {
      reducer: (current, update) => current.concat(update),
      default: () => [],
    },
    // LastValue：存储最近的值（null = 无归约器）
    question: null,
    answer: null,
  },
});

直接使用通道类： 为了获得更多控制，你可以直接实例化通道类：

import { BaseMessage } from "@langchain/core/messages";
import { StateGraph, LastValue, BinaryOperatorAggregate, Topic } from "@langchain/langgraph";

interface WorkflowState {
  messages: BaseMessage[];
  question: string;
  events: string[];
}

const workflow = new StateGraph<WorkflowState>({
  channels: {
    messages: new BinaryOperatorAggregate<BaseMessage[]>(
      (current, update) => current.concat(update),
      () => []
    ),
    question: new LastValue<string>(),
    // Topic 收集执行期间推送的所有值
    events: new Topic<string>(),
  },
});

Annotation.Root

Annotation.Root 提供了一种声明式的方式来定义带有归约器的状态。它类似于 StateSchema，但使用不同的语法：

import { BaseMessage } from "@langchain/core/messages";
import { Annotation, StateGraph, messagesStateReducer } from "@langchain/langgraph";

const State = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: messagesStateReducer,
    default: () => [],
  }),
  question: Annotation<string>(),
  count: Annotation<number>({
    reducer: (current, update) => current + update,
    default: () => 0,
  }),
});

const graph = new StateGraph(State);

使用 Zod v3 的 Zod 对象

当使用 Zod v3 时，你可以使用普通的 z.object() 架构来定义状态。LangGraph 通过一个 .langgraph 插件扩展了 Zod v3，该插件提供了 .reducer() 和 .metadata() 方法：

import { z } from "zod/v3";
import { BaseMessage } from "@langchain/core/messages";
import { StateGraph, messagesStateReducer } from "@langchain/langgraph";

const State = z.object({
  // 使用 .langgraph.reducer() 来附加归约器函数
  messages: z
    .array(z.custom<BaseMessage>())
    .default([])
    .langgraph.reducer(messagesStateReducer),
  // 简单字段直接工作（最后写入获胜）
  question: z.string().optional(),
  answer: z.string().optional(),
  // 用于累积值的自定义归约器
  count: z
    .number()
    .default(0)
    .langgraph.reducer((current, update) => current + update),
});

const graph = new StateGraph(State);

使用 Zod v4 的 Zod 对象

Zod v4 使用基于注册表的方法。使用 LangGraph 注册表来向架构字段附加元数据：

import * as z from "zod";
import { BaseMessage } from "@langchain/core/messages";
import { StateGraph, MessagesZodMeta, messagesStateReducer } from "@langchain/langgraph";
import { registry } from "@langchain/langgraph/zod";

const State = z.object({
  // 使用 .register() 配合 LangGraph 注册表和 MessagesZodMeta
  messages: z
    .array(z.custom<BaseMessage>())
    .default([])
    .register(registry, MessagesZodMeta),
  // 简单字段直接工作（最后写入获胜）
  question: z.string().optional(),
  answer: z.string().optional(),
  // 通过注册表元数据自定义归约器
  count: z
    .number()
    .default(0)
    .register(registry, { reducer: (current: number, update: number) => current + update }),
});

const graph = new StateGraph(State);

比较表

方法	归约器	类型安全	Zod 版本	推荐
`StateSchema`	✅ 内置	✅ 完整	v3 或 v4	✅ 是
通道 API	✅ 手动	⚠️ 部分	不适用	用于高级场景
`Annotation.Root`	✅ 内置	✅ 完整	不适用	遗留
Zod v3 + `.langgraph`	✅ 通过插件	✅ 完整	仅 v3	遗留
Zod v4 + 注册表	✅ 通过注册表	✅ 完整	仅 v4	遗留

添加运行时配置

有时你希望在调用图时能够对其进行配置。例如，你可能希望能够在运行时指定使用哪个 LLM 或系统提示符，而不会用这些参数污染图状态。要添加运行时配置：

为你的配置指定一个架构
将配置添加到节点或条件边的函数签名中
将配置传递给图。

下面是一个简单的示例：

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

// 1. 指定配置架构
const ContextSchema = z.object({
  myRuntimeValue: z.string(),
});

// 2. 定义一个在节点中访问配置的图
const State = new StateSchema({
  myStateValue: z.number(),
});

const node: GraphNode<typeof State> = (state, runtime) => {
  if (runtime?.context?.myRuntimeValue === "a") {
    return { myStateValue: 1 };
  } else if (runtime?.context?.myRuntimeValue === "b") {
    return { myStateValue: 2 };
  } else {
    throw new Error("Unknown values.");
  }
};

const graph = new StateGraph(State, ContextSchema)
  .addNode("node", node)
  .addEdge(START, "node")
  .addEdge("node", END)
  .compile();

// 3. 在运行时传入配置：
console.log(await graph.invoke({}, { context: { myRuntimeValue: "a" } }));
console.log(await graph.invoke({}, { context: { myRuntimeValue: "b" } }));

{ myStateValue: 1 }
{ myStateValue: 2 }

扩展示例：在运行时指定 LLM

下面我们演示一个实际示例，我们在运行时配置要使用的 LLM。我们将使用 OpenAI 和 Anthropic 的模型。

import { ChatOpenAI } from "@langchain/openai";
import { ChatAnthropic } from "@langchain/anthropic";
import { StateGraph, StateSchema, MessagesValue, GraphNode, START, END } from "@langchain/langgraph";
import * as z from "zod";

const ConfigSchema = z.object({
  modelProvider: z.string().default("anthropic"),
});

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

const MODELS = {
  anthropic: new ChatAnthropic({ model: "claude-haiku-4-5-20251001" }),
  openai: new ChatOpenAI({ model: "gpt-5.4-mini" }),
};

const callModel: GraphNode<typeof State> = async (state, config) => {
  const modelProvider = config?.configurable?.modelProvider || "anthropic";
  const model = MODELS[modelProvider as keyof typeof MODELS];
  const response = await model.invoke(state.messages);
  return { messages: [response] };
};

const graph = new StateGraph(State, ConfigSchema)
  .addNode("model", callModel)
  .addEdge(START, "model")
  .addEdge("model", END)
  .compile();

// 用法
const inputMessage = { role: "user", content: "hi" };
// 无配置时，使用默认值（Anthropic）
const response1 = await graph.invoke({ messages: [inputMessage] });
// 或者，可以设置为 OpenAI
const response2 = await graph.invoke(
  { messages: [inputMessage] },
  { configurable: { modelProvider: "openai" } },
);

console.log(response1.messages.at(-1)?.response_metadata?.model);
console.log(response2.messages.at(-1)?.response_metadata?.model);

claude-haiku-4-5-20251001
gpt-5.4-mini

扩展示例：在运行时指定模型和系统消息

下面我们演示一个实际示例，我们配置两个参数：运行时要使用的 LLM 和系统消息。

import { ChatOpenAI } from "@langchain/openai";
import { ChatAnthropic } from "@langchain/anthropic";
import { SystemMessage } from "@langchain/core/messages";
import { StateGraph, StateSchema, MessagesValue, GraphNode, START, END } from "@langchain/langgraph";
import * as z from "zod";

const ConfigSchema = z.object({
  modelProvider: z.string().default("anthropic"),
  systemMessage: z.string().optional(),
});

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

const MODELS = {
  anthropic: new ChatAnthropic({ model: "claude-haiku-4-5-20251001" }),
  openai: new ChatOpenAI({ model: "gpt-5.4-mini" }),
};

const callModel: GraphNode<typeof State> = async (state, config) => {
  const modelProvider = config?.configurable?.modelProvider || "anthropic";
  const systemMessage = config?.configurable?.systemMessage;

  const model = MODELS[modelProvider as keyof typeof MODELS];
  let messages = state.messages;

  if (systemMessage) {
    messages = [new SystemMessage(systemMessage), ...messages];
  }

  const response = await model.invoke(messages);
  return { messages: [response] };
};

const graph = new StateGraph(State, ConfigSchema)
  .addNode("model", callModel)
  .addEdge(START, "model")
  .addEdge("model", END)
  .compile();

// 用法
const inputMessage = { role: "user", content: "hi" };
const response = await graph.invoke(
  { messages: [inputMessage] },
  {
    configurable: {
      modelProvider: "openai",
      systemMessage: "Respond in Italian."
    }
  }
);

for (const message of response.messages) {
  console.log(`${message.getType()}: ${message.content}`);
}

human: hi
ai: Ciao! Come posso aiutarti oggi?

添加重试策略

在许多用例中，你可能希望你的节点具有自定义的重试策略，例如，如果你正在调用 API、查询数据库或调用 LLM 等。LangGraph 允许你为节点添加重试策略。要配置重试策略，请将 retryPolicy 参数传递给 addNode。retryPolicy 参数接受一个 RetryPolicy 对象。下面我们将使用默认参数实例化一个 RetryPolicy 对象，并将其与一个节点关联：

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

const graph = new StateGraph(State)
  .addNode("nodeName", nodeFunction, { retryPolicy: {} })
  .compile();

默认情况下，重试策略会对除以下异常之外的任何异常进行重试：

TypeError
SyntaxError
ReferenceError

扩展示例：自定义重试策略

考虑一个从 SQL 数据库读取的示例。下面我们向节点传递两个不同的重试策略：

import Database from "better-sqlite3";
import { ChatAnthropic } from "@langchain/anthropic";
import { StateGraph, StateSchema, MessagesValue, GraphNode, START, END } from "@langchain/langgraph";
import { AIMessage } from "@langchain/core/messages";

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

// 创建一个内存数据库
const db: typeof Database.prototype = new Database(":memory:");

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" });

const callModel: GraphNode<typeof State> = async (state) => {
  const response = await model.invoke(state.messages);
  return { messages: [response] };
};

const queryDatabase: GraphNode<typeof State> = async (state) => {
  const queryResult: string = JSON.stringify(
    db.prepare("SELECT * FROM Artist LIMIT 10;").all(),
  );

  return { messages: [new AIMessage({ content: "queryResult" })] };
};

const workflow = new StateGraph(State)
  // 定义我们将在其间循环的两个节点
  .addNode("call_model", callModel, { retryPolicy: { maxAttempts: 5 } })
  .addNode("query_database", queryDatabase, {
    retryPolicy: {
      retryOn: (e: any): boolean => {
        if (e instanceof Database.SqliteError) {
          // 对 "SQLITE_BUSY" 错误进行重试
          return e.code === "SQLITE_BUSY";
        }
        return false; // 不对其他错误进行重试
      },
    },
  })
  .addEdge(START, "call_model")
  .addEdge("call_model", "query_database")
  .addEdge("query_database", END);

const graph = workflow.compile();

在节点内部访问执行信息

你可以通过 runtime.executionInfo 访问执行标识和重试信息。这会显示线程、运行和检查点标识符以及重试状态，而无需直接从 config 读取。

属性	类型	描述
`threadId`	`string \| undefined`	当前执行的线程 ID。
`runId`	`string \| undefined`	当前执行的运行 ID。
`checkpointId`	`string`	当前执行的检查点 ID。
`checkpointNs`	`string`	当前执行的检查点命名空间。
`taskId`	`string`	当前执行的任务 ID。
`nodeAttempt`	`number`	当前执行尝试次数（从 1 开始）。
`nodeFirstAttemptTime`	`number \| undefined`	第一次尝试开始时的 Unix 时间戳（秒）。在重试期间保持不变。

访问线程和运行 ID

使用 executionInfo 在节点内部访问线程 ID、运行 ID 和其他标识字段：

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

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

const myNode: GraphNode<typeof State> = async (state, runtime) => {
  const info = runtime.executionInfo;
  console.log(`Thread: ${info.threadId}, Run: ${info.runId}`);
  return { result: "done" };
};

const graph = new StateGraph(State)
  .addNode("my_node", myNode)
  .addEdge(START, "my_node")
  .addEdge("my_node", END)
  .compile();

根据重试状态调整行为

当节点具有重试策略时，使用 executionInfo 检查当前尝试次数，并在第一次尝试失败后切换到回退方案：

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

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

const myNode: GraphNode<typeof State> = async (state, runtime) => {
  const info = runtime.executionInfo;
  if (info.nodeAttempt > 1) {
    // 在重试时使用回退方案
    return { result: await callFallbackApi() };
  }
  return { result: await callPrimaryApi() };
};

const graph = new StateGraph(State)
  .addNode("my_node", myNode, { retryPolicy: { maxAttempts: 3 } })
  .addEdge(START, "my_node")
  .addEdge("my_node", END)
  .compile();

即使没有重试策略，executionInfo 也可在 Runtime 对象上使用——nodeAttempt 默认为 1，nodeFirstAttemptTime 设置为节点开始执行的时间。

在节点内部访问服务器信息

当你的图在 LangGraph Server 上运行时，你可以通过 runtime.serverInfo 访问服务器特定的元数据。

属性	类型	描述
`assistantId`	`string`	当前部署的助手 ID。
`graphId`	`string`	当前部署的图 ID。
`user`	`BaseUser \| null`	已认证用户，如果配置了自定义认证。

const myNode: GraphNode<typeof State> = async (state, runtime) => {
  const server = runtime.serverInfo;
  if (server != null) {
    console.log(`Assistant: ${server.assistantId}, Graph: ${server.graphId}`);
    if (server.user != null) {
      console.log(`User: ${server.user.identity}`);
    }
  }
  return { result: "done" };
};

当图未在 LangGraph Server 上运行时，serverInfo 为 null。

需要 deepagents>=1.9.0（或 @langchain/langgraph>=1.2.8）才能使用 runtime.executionInfo 和 runtime.serverInfo。

创建步骤序列

前提条件 本指南假设你熟悉上面关于状态的部分。

这里我们演示如何构建一个简单的步骤序列。我们将展示：

如何构建一个顺序图
构建类似图的内置简写方式。

要添加一系列节点，我们使用图的 .addNode 和 .addEdge 方法：

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

const builder = new StateGraph(State)
  .addNode("step1", step1)
  .addNode("step2", step2)
  .addNode("step3", step3)
  .addEdge(START, "step1")
  .addEdge("step1", "step2")
  .addEdge("step2", "step3");

为什么使用 LangGraph 将应用程序步骤拆分为序列？

LangGraph 可以轻松地为你的应用程序添加底层持久化层。这允许在节点执行之间对状态进行检查点，因此你的 LangGraph 节点控制着：

状态更新如何被检查点
如何在人机交互工作流中恢复中断
如何使用 LangGraph 的时间旅行功能“回退”和分支执行

它们还决定执行步骤如何被流式传输，以及你的应用程序如何使用 Studio 进行可视化和调试。让我们演示一个端到端的示例。我们将创建一个三个步骤的序列：

在状态的某个键中填入一个值
更新同一个值
填入一个不同的值

首先，我们来定义我们的状态。这控制着图的模式，并可以指定如何应用更新。更多详情请参见使用 Reducer 处理状态更新。在我们的例子中，我们只跟踪两个值：

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

const State = new StateSchema({
  value1: z.string(),
  value2: z.number(),
});

我们的节点只是读取图的状态并对其进行更新的 TypeScript 函数。该函数的第一个参数始终是状态：

const step1: GraphNode<typeof State> = (state) => {
  return { value1: "a" };
};

const step2: GraphNode<typeof State> = (state) => {
  const currentValue1 = state.value1;
  return { value1: `${currentValue1} b` };
};

const step3: GraphNode<typeof State> = (state) => {
  return { value2: 10 };
};

请注意，在向状态发出更新时，每个节点只需指定它希望更新的键的值。默认情况下，这会覆盖相应键的值。你也可以使用 Reducer 来控制更新的处理方式——例如，你可以将连续的更新追加到某个键后面。更多详情请参见使用 Reducer 处理状态更新。

最后，我们定义图。我们使用 StateGraph 来定义一个操作此状态的图。然后我们将使用 addNode 和 addEdge 来填充我们的图并定义其控制流。

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

const graph = new StateGraph(State)
  .addNode("step1", step1)
  .addNode("step2", step2)
  .addNode("step3", step3)
  .addEdge(START, "step1")
  .addEdge("step1", "step2")
  .addEdge("step2", "step3")
  .compile();

指定自定义名称 你可以使用 .addNode 为节点指定自定义名称：

const graph = new StateGraph(State)
.addNode("myNode", step1)
.compile();

请注意：

.addEdge 接受节点的名称，对于函数，默认为 node.name。
我们必须指定图的入口点。为此，我们添加一条边连接到 START 节点。
当没有更多节点可执行时，图会停止。

接下来我们编译我们的图。这会提供一些基本的图结构检查（例如，识别孤立的节点）。如果我们通过检查点器为应用程序添加持久化功能，它也会在这里传入。LangGraph 提供了内置的工具来可视化你的图。让我们检查一下我们的序列。有关可视化的详细信息，请参见可视化你的图。

import * as fs from "node:fs/promises";

const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());

await fs.writeFile("graph.png", imageBuffer);

让我们进行一次简单的调用：

const result = await graph.invoke({ value1: "c" });
console.log(result);

{ value1: 'a b', value2: 10 }

请注意：

我们通过为单个状态键提供一个值来启动调用。我们必须始终为至少一个键提供一个值。
我们传入的值被第一个节点覆盖了。
第二个节点更新了该值。
第三个节点填入了一个不同的值。

创建分支

节点的并行执行对于加速整体图运行至关重要。LangGraph 原生支持节点的并行执行，这可以显著提高基于图的工作流的性能。这种并行化是通过扇出和扇入机制实现的，利用标准边和 conditional_edges。下面是一些示例，展示了如何添加适合你的创建分支数据流。

并行运行图节点

在此示例中，我们从 Node A 扇出到 B 和 C，然后扇入到 D。对于我们的状态，我们指定了 reducer add 操作。这将为状态中的特定键组合或累积值，而不是简单地覆盖现有值。对于列表，这意味着将新列表与现有列表连接起来。有关使用 Reducer 更新状态的更多详情，请参见上面关于状态 Reducer 的部分。

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

const State = new StateSchema({
  // reducer 使其为只追加
  aggregate: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});

const nodeA: GraphNode<typeof State> = (state) => {
  console.log(`Adding "A" to ${state.aggregate}`);
  return { aggregate: ["A"] };
};

const nodeB: GraphNode<typeof State> = (state) => {
  console.log(`Adding "B" to ${state.aggregate}`);
  return { aggregate: ["B"] };
};

const nodeC: GraphNode<typeof State> = (state) => {
  console.log(`Adding "C" to ${state.aggregate}`);
  return { aggregate: ["C"] };
};

const nodeD: GraphNode<typeof State> = (state) => {
  console.log(`Adding "D" to ${state.aggregate}`);
  return { aggregate: ["D"] };
};

const graph = new StateGraph(State)
  .addNode("a", nodeA)
  .addNode("b", nodeB)
  .addNode("c", nodeC)
  .addNode("d", nodeD)
  .addEdge(START, "a")
  .addEdge("a", "b")
  .addEdge("a", "c")
  .addEdge("b", "d")
  .addEdge("c", "d")
  .addEdge("d", END)
  .compile();

import * as fs from "node:fs/promises";

const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());

await fs.writeFile("graph.png", imageBuffer);

通过 Reducer，你可以看到在每个节点中添加的值被累加了。

const result = await graph.invoke({
  aggregate: [],
});
console.log(result);

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']
{ aggregate: ['A', 'B', 'C', 'D'] }

在上面的示例中，节点 "b" 和 "c" 在同一个超步中并发执行。因为它们在同一个步骤中，节点 "d" 在 "b" 和 "c" 都完成后才执行。重要的是，并行超步的更新可能不会一致排序。如果你需要来自并行超步的更新具有一致、预定的顺序，你应该将输出与用于排序的值一起写入状态的单独字段中。

异常处理？

LangGraph 在超步内执行节点，这意味着虽然并行分支是并行执行的，但整个超步是事务性的。如果这些分支中的任何一个引发异常，则不会将任何更新应用到状态（整个超步入错）。重要的是，当使用检查点器时，超步内成功节点的结果会被保存，并且在恢复时不会重复执行。如果你有容易出错的节点（也许想处理不稳定的 API 调用），LangGraph 提供两种方式来解决：

你可以在节点内编写常规的 Python 代码来捕获和处理异常。
你可以设置一个 retry_policy 来指示图重试引发特定类型异常的节点。只有失败的分支会被重试，因此你不必担心执行冗余工作。

这些功能共同让你可以执行并行执行并完全控制异常处理。

设置最大并发数 你可以在调用图时通过在配置中设置 max_concurrency 来控制最大并发任务数。

const result = await graph.invoke({ value1: "c" }, {configurable: {max_concurrency: 10}});

条件分支

如果你的扇出应该在运行时根据状态变化，你可以使用 addConditionalEdges 来使用图状态选择一个或多个路径。参见下面的示例，其中节点 a 生成一个状态更新，决定后续节点。

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

const State = new StateSchema({
  aggregate: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
  // 在状态中添加一个键。我们将设置这个键来决定
  // 如何分支。
  which: z.string(),
});

const nodeA: GraphNode<typeof State> = (state) => {
  console.log(`Adding "A" to ${state.aggregate}`);
  return { aggregate: ["A"], which: "c" };
};

const nodeB: GraphNode<typeof State> = (state) => {
  console.log(`Adding "B" to ${state.aggregate}`);
  return { aggregate: ["B"] };
};

const nodeC: GraphNode<typeof State> = (state) => {
  console.log(`Adding "C" to ${state.aggregate}`);
  return { aggregate: ["C"] };
};

const conditionalEdge: ConditionalEdgeRouter<typeof State, "b" | "c"> = (state) => {
  // 在这里填写任意逻辑，使用状态
  // 来确定下一个节点
  return state.which as "b" | "c";
};

const graph = new StateGraph(State)
  .addNode("a", nodeA)
  .addNode("b", nodeB)
  .addNode("c", nodeC)
  .addEdge(START, "a")
  .addEdge("b", END)
  .addEdge("c", END)
  .addConditionalEdges("a", conditionalEdge)
  .compile();

import * as fs from "node:fs/promises";

const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());

await fs.writeFile("graph.png", imageBuffer);

const result = await graph.invoke({ aggregate: [] });
console.log(result);

Adding "A" to []
Adding "C" to ['A']
{ aggregate: ['A', 'C'], which: 'c' }

你的条件边可以路由到多个目标节点。例如：

const routeBcOrCd: ConditionalEdgeRouter<typeof State, "b" | "c" | "d"> = (state) => {
  if (state.which === "cd") {
    return ["c", "d"];
  }
  return ["b", "c"];
};

Map-Reduce 和 Send API

LangGraph 使用 Send API 支持 Map-Reduce 和其他高级分支模式。以下是如何使用它的一个示例：

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

const OverallState = new StateSchema({
  topic: z.string(),
  subjects: z.array(z.string()),
  jokes: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
  bestSelectedJoke: z.string(),
});

const generateTopics: GraphNode<typeof OverallState> = (state) => {
  return { subjects: ["lions", "elephants", "penguins"] };
};

const generateJoke: GraphNode<typeof OverallState> = (state) => {
  const jokeMap: Record<string, string> = {
    lions: "Why don't lions like fast food? Because they can't catch it!",
    elephants: "Why don't elephants use computers? They're afraid of the mouse!",
    penguins: "Why don't penguins like talking to strangers at parties? Because they find it hard to break the ice."
  };
  return { jokes: [jokeMap[state.subject]] };
};

const continueToJokes: ConditionalEdgeRouter<typeof OverallState, "generateJoke"> = (state) => {
  return state.subjects.map((subject) => new Send("generateJoke", { subject }));
};

const bestJoke: GraphNode<typeof OverallState> = (state) => {
  return { bestSelectedJoke: "penguins" };
};

const graph = new StateGraph(OverallState)
  .addNode("generateTopics", generateTopics)
  .addNode("generateJoke", generateJoke)
  .addNode("bestJoke", bestJoke)
  .addEdge(START, "generateTopics")
  .addConditionalEdges("generateTopics", continueToJokes)
  .addEdge("generateJoke", "bestJoke")
  .addEdge("bestJoke", END)
  .compile();

import * as fs from "node:fs/promises";

const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());

await fs.writeFile("graph.png", imageBuffer);

// 调用图：这里我们调用它来生成一个笑话列表
const stream = await graph.streamEvents({ topic: "animals" }, { version: "v3" });
for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

{ generateTopics: { subjects: [ 'lions', 'elephants', 'penguins' ] } }
{ generateJoke: { jokes: [ "Why don't lions like fast food? Because they can't catch it!" ] } }
{ generateJoke: { jokes: [ "Why don't elephants use computers? They're afraid of the mouse!" ] } }
{ generateJoke: { jokes: [ "Why don't penguins like talking to strangers at parties? Because they find it hard to break the ice." ] } }
{ bestJoke: { bestSelectedJoke: 'penguins' } }

创建和控制循环

创建带有循环的图时，我们需要一种机制来终止执行。最常见的方式是添加一个条件边，一旦我们达到某个终止条件，就路由到 END 节点。你也可以在调用或流式传输图时设置图的递归限制。递归限制设置了图在引发错误之前允许执行的超步数量。阅读更多关于递归限制概念的信息。让我们考虑一个带有循环的简单图，以更好地理解这些机制的工作原理。

要返回状态的最后一个值而不是收到递归限制错误，请参阅下一节。

创建循环时，您可以包含一个指定终止条件的条件边：

const route: ConditionalEdgeRouter<typeof State, "b"> = (state) => {
  if (terminationCondition(state)) {
    return END;
  } else {
    return "b";
  }
};

const graph = new StateGraph(State)
  .addNode("a", nodeA)
  .addNode("b", nodeB)
  .addEdge(START, "a")
  .addConditionalEdges("a", route)
  .addEdge("b", "a")
  .compile();

要控制递归限制，请在配置中指定 "recursionLimit"。这将引发 GraphRecursionError，您可以捕获并处理它：

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

try {
  await graph.invoke(inputs, { recursionLimit: 3 });
} catch (error) {
  if (error instanceof GraphRecursionError) {
    console.log("Recursion Error");
  }
}

让我们定义一个带有简单循环的图。请注意，我们使用条件边来实现终止条件。

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

const State = new StateSchema({
  // reducer 使其仅支持追加操作
  aggregate: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});

const nodeA: GraphNode<typeof State> = (state) => {
  console.log(`节点 A 看到 ${state.aggregate}`);
  return { aggregate: ["A"] };
};

const nodeB: GraphNode<typeof State> = (state) => {
  console.log(`节点 B 看到 ${state.aggregate}`);
  return { aggregate: ["B"] };
};

// 定义边
const route: ConditionalEdgeRouter<typeof State, "b"> = (state) => {
  if (state.aggregate.length < 7) {
    return "b";
  } else {
    return END;
  }
};

const graph = new StateGraph(State)
  .addNode("a", nodeA)
  .addNode("b", nodeB)
  .addEdge(START, "a")
  .addConditionalEdges("a", route)
  .addEdge("b", "a")
  .compile();

import * as fs from "node:fs/promises";

const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());

await fs.writeFile("graph.png", imageBuffer);

此架构类似于一个 ReAct 智能体，其中节点 "a" 是一个工具调用模型，节点 "b" 代表工具。在我们的 route 条件边中，我们指定当状态中的 "aggregate" 列表长度超过阈值后应终止。调用图时，我们可以看到节点 "a" 和 "b" 交替执行，直到满足终止条件。

const result = await graph.invoke({ aggregate: [] });
console.log(result);

节点 A 看到 []
节点 B 看到 ['A']
节点 A 看到 ['A', 'B']
节点 B 看到 ['A', 'B', 'A']
节点 A 看到 ['A', 'B', 'A', 'B']
节点 B 看到 ['A', 'B', 'A', 'B', 'A']
节点 A 看到 ['A', 'B', 'A', 'B', 'A', 'B']
{ aggregate: ['A', 'B', 'A', 'B', 'A', 'B', 'A'] }

施加递归限制

在某些应用中，我们可能无法保证会达到给定的终止条件。在这些情况下，我们可以设置图的递归限制。这会在给定数量的超级步骤后引发 GraphRecursionError。然后我们可以捕获并处理此异常：

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

try {
  await graph.invoke({ aggregate: [] }, { recursionLimit: 4 });
} catch (error) {
  if (error instanceof GraphRecursionError) {
    console.log("Recursion Error");
  }
}

节点 A 看到 []
节点 B 看到 ['A']
节点 A 看到 ['A', 'B']
节点 B 看到 ['A', 'B', 'A']
节点 A 看到 ['A', 'B', 'A', 'B']
递归错误

使用 `Command` 组合控制流和状态更新

将控制流（边）和状态更新（节点）组合起来可能很有用。例如，您可能希望在同一个节点中同时执行状态更新并决定下一步要跳转到哪个节点。LangGraph 提供了一种方法，通过从节点函数返回一个 Command 对象来实现：

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

const myNode = (state: State): Command => {
  return new Command({
    // 状态更新
    update: { foo: "bar" },
    // 控制流
    goto: "myOtherNode"
  });
};

下面我们展示一个端到端的示例。让我们创建一个包含 3 个节点的简单图：A、B 和 C。我们将首先执行节点 A，然后根据节点 A 的输出决定接下来是跳转到节点 B 还是节点 C。

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

// 定义图状态
const State = new StateSchema({
  foo: z.string(),
});

// 定义节点

const nodeA: GraphNode<typeof State, "nodeB" | "nodeC"> = (state) => {
  console.log("调用 A");
  const value = Math.random() > 0.5 ? "b" : "c";
  // 这相当于一个条件边函数
  const goto = value === "b" ? "nodeB" : "nodeC";

  // 注意 Command 如何让你既能更新图状态，又能路由到下一个节点
  return new Command({
    // 这是状态更新
    update: { foo: value },
    // 这相当于一条边
    goto,
  });
};

const nodeB: GraphNode<typeof State> = (state) => {
  console.log("调用 B");
  return { foo: state.foo + "b" };
};

const nodeC: GraphNode<typeof State> = (state) => {
  console.log("调用 C");
  return { foo: state.foo + "c" };
};

现在我们可以使用上述节点创建 StateGraph。请注意，该图没有用于路由的条件边！这是因为控制流是在 nodeA 内部使用 Command 定义的。

const graph = new StateGraph(State)
  .addNode("nodeA", nodeA, {
    ends: ["nodeB", "nodeC"],
  })
  .addNode("nodeB", nodeB)
  .addNode("nodeC", nodeC)
  .addEdge(START, "nodeA")
  .compile();

您可能注意到我们使用 ends 来指定 nodeA 可以导航到哪些节点。这对于图渲染是必要的，并告诉 LangGraph nodeA 可以导航到 nodeB 和 nodeC。

import * as fs from "node:fs/promises";

const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());

await fs.writeFile("graph.png", imageBuffer);

如果我们多次运行该图，我们会看到它根据节点 A 中的随机选择走不同的路径（A -> B 或 A -> C）。

const result = await graph.invoke({ foo: "" });
console.log(result);

调用 A
调用 C
{ foo: 'cc' }

导航到父图中的节点

如果您正在使用子图，您可能希望从子图中的一个节点导航到另一个子图（即父图中的不同节点）。为此，您可以在 Command 中指定 graph=Command.PARENT：

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

让我们用上面的例子来演示。我们将通过把上面示例中的 nodeA 改成一个单节点图，并将其作为子图添加到我们的父图中来实现。

使用 Command.PARENT 的状态更新 当您从子图节点向父图节点发送更新，并且该键由父图和子图的状态模式共享时，您必须为父图状态中正在更新的键定义一个 reducer。请参见下面的示例。

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

const State = new StateSchema({
  // 注意：我们在这里定义了一个 reducer
  foo: new ReducedValue(  
    z.string().default(""),
    { reducer: (x, y) => x + y }
  ),
});

const nodeA: GraphNode<typeof State, "nodeB" | "nodeC"> = (state) => {
  console.log("调用 A");
  const value = Math.random() > 0.5 ? "nodeB" : "nodeC";

  // 注意 Command 如何让你既能更新图状态，又能路由到下一个节点
  return new Command({
    update: { foo: "a" },
    goto: value,
    // 这告诉 LangGraph 导航到父图中的 nodeB 或 nodeC
    // 注意：这将导航到相对于子图的最近父图
    graph: Command.PARENT,
  });
};

const subgraph = new StateGraph(State)
  .addNode("nodeA", nodeA, { ends: ["nodeB", "nodeC"] })
  .addEdge(START, "nodeA")
  .compile();

const nodeB: GraphNode<typeof State> = (state) => {
  console.log("调用 B");
  // 注意：由于我们定义了 reducer，我们不需要手动将
  // 新字符追加到现有的 'foo' 值。相反，reducer 会自动
  // 追加
  return { foo: "b" };
};

const nodeC: GraphNode<typeof State> = (state) => {
  console.log("调用 C");
  return { foo: "c" };
};

const graph = new StateGraph(State)
  .addNode("subgraph", subgraph, { ends: ["nodeB", "nodeC"] })
  .addNode("nodeB", nodeB)
  .addNode("nodeC", nodeC)
  .addEdge(START, "subgraph")
  .compile();

const result = await graph.invoke({ foo: "" });
console.log(result);

调用 A
调用 C
{ foo: 'ac' }

在工具内部使用

一个常见的用例是从工具内部更新图状态。例如，在一个客户支持应用中，您可能希望在对话开始时根据客户的账号或 ID 查询客户信息。要从工具更新图状态，您可以从工具返回 Command(update={"my_custom_key": "foo", "messages": [...]})：

import { tool } from "@langchain/core/tools";
import { Command } from "@langchain/langgraph";
import * as z from "zod";

const lookupUserInfo = tool(
  async (input, runtime) => {
    const userId = runtime.serverInfo?.user?.identity;
    const userInfo = getUserInfo(userId);
    return new Command({
      update: {
        // 更新状态键
        userInfo: userInfo,
        // 更新消息历史
        messages: [{
          role: "tool",
          content: "成功查询用户信息",
          tool_call_id: runtime.toolCall.id
        }]
      }
    });
  },
  {
    name: "lookupUserInfo",
    description: "使用此工具查询用户信息，以便更好地协助他们的问题。",
    schema: z.object({}),
  }
);

当从工具返回 Command 时，您必须在 Command.update 中包含 messages（或用于消息历史记录的任何状态键），并且 messages 中的消息列表必须包含一个 ToolMessage。这对于最终消息历史记录的有效性是必要的（LLM 提供商要求带有工具调用的 AI 消息之后必须跟随工具结果消息）。

如果您使用通过 Command 更新状态的工具，我们建议使用预构建的 ToolNode，它可以自动处理返回 Command 对象的工具并将其传播到图状态。如果您正在编写一个调用工具的自定义节点，则需要手动将工具返回的 Command 对象作为该节点的更新进行传播。

可视化您的图

这里我们演示如何可视化您创建的图。您可以可视化任意 Graph，包括 StateGraph。让我们创建一个简单的示例图来演示可视化。

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

const State = new StateSchema({
  messages: MessagesValue,
  value: new ReducedValue(
    z.number().default(0),
    { reducer: (x, y) => x + y }
  ),
});

const node1: GraphNode<typeof State> = (state) => {
  return { value: state.value + 1 };
};

const node2: GraphNode<typeof State> = (state) => {
  return { value: state.value * 2 };
};

const router: ConditionalEdgeRouter<typeof State, "node2"> = (state) => {
  if (state.value < 10) {
    return "node2";
  }
  return END;
};

const app = new StateGraph(State)
  .addNode("node1", node1)
  .addNode("node2", node2)
  .addEdge(START, "node1")
  .addConditionalEdges("node1", router)
  .addEdge("node2", "node1")
  .compile();

Mermaid

我们也可以将图类转换为 Mermaid 语法。

const drawableGraph = await app.getGraphAsync();
console.log(drawableGraph.drawMermaid());

%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
    tart__([<p>__start__</p>]):::first
    e1(node1)
    e2(node2)
    nd__([<p>__end__</p>]):::last
    tart__ --> node1;
    e1 -.-> node2;
    e1 -.-> __end__;
    e2 --> node1;
    ssDef default fill:#f2f0ff,line-height:1.2
    ssDef first fill-opacity:0
    ssDef last fill:#bfb6fc

PNG

如果愿意，我们可以将 Graph 渲染成 .png。这里使用 Mermaid.ink API 来生成图表。

import * as fs from "node:fs/promises";

const drawableGraph = await app.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());

await fs.writeFile("graph.png", imageBuffer);

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

Edit this page on GitHub or file an issue.

Get started

Capabilities

Production

Frontend

LangGraph APIs

设置

定义和更新状态

定义状态

更新状态

使用归约器处理状态更新

MessagesValue

定义输入和输出架构

在节点之间传递私有状态

替代状态定义

通道 API

Annotation.Root

使用 Zod v3 的 Zod 对象

使用 Zod v4 的 Zod 对象

比较表

添加运行时配置

添加重试策略

在节点内部访问执行信息

访问线程和运行 ID

根据重试状态调整行为

在节点内部访问服务器信息

创建步骤序列

创建分支

并行运行图节点

条件分支

Map-Reduce 和 Send API

创建和控制循环

施加递归限制

使用 `Command` 组合控制流和状态更新

导航到父图中的节点

在工具内部使用

可视化您的图

Mermaid

PNG

​设置

​定义和更新状态

​定义状态

​更新状态

​使用归约器处理状态更新

​MessagesValue

​定义输入和输出架构

​在节点之间传递私有状态

​替代状态定义

​通道 API

​Annotation.Root

​使用 Zod v3 的 Zod 对象

​使用 Zod v4 的 Zod 对象

​比较表

​添加运行时配置

​添加重试策略

​在节点内部访问执行信息

​访问线程和运行 ID

​根据重试状态调整行为

​在节点内部访问服务器信息

​创建步骤序列

​创建分支

​并行运行图节点

​条件分支

​Map-Reduce 和 Send API

​创建和控制循环

​施加递归限制

​使用 Command 组合控制流和状态更新

​导航到父图中的节点

​在工具内部使用

​可视化您的图

​Mermaid

​PNG

设置

定义和更新状态

定义状态

更新状态

使用归约器处理状态更新

MessagesValue

定义输入和输出架构

在节点之间传递私有状态

替代状态定义

通道 API

Annotation.Root

使用 Zod v3 的 Zod 对象

使用 Zod v4 的 Zod 对象

比较表

添加运行时配置

添加重试策略

在节点内部访问执行信息

访问线程和运行 ID

根据重试状态调整行为

在节点内部访问服务器信息

创建步骤序列

创建分支

并行运行图节点

条件分支

Map-Reduce 和 Send API

创建和控制循环

施加递归限制

使用 `Command` 组合控制流和状态更新

导航到父图中的节点

在工具内部使用

可视化您的图

Mermaid

PNG