使用图 API - 以编程方式使用文档

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

设置

安装 langgraph:

npm install @langchain/langgraph

定义和更新状态

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

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

定义状态

状态在 LangGraph 中使用 StateSchema 类来定义。这提供了一个统一的 API，接受标准模式（如 Zod）作为各个字段以及特殊值类型如 ReducedValue, MessagesValue和 UntrackedValue.

默认情况下，图的输入和输出模式相同，状态决定该模式。参见定义输入和输出模式了解如何定义不同的输入和输出模式。

让我们考虑一个使用消息。这代表了多种 LLM 应用程序的状态的通用公式。请参阅我们的概念页面了解更多详情。

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

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

更新状态

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

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

此节点只是将消息追加到我们的消息列表中（归约器处理连接），并填充一个额外字段。

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

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

LangGraph 提供了内置的图形可视化工具。让我们检查一下我们的图。请参阅可视化您的图了解可视化的详细信息。

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({ 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 它已经有一个内置的 reducer。对于自定义字段，你可以使用 ReducedValue 来定义如何应用更新。

在之前的示例中，我们的节点更新了 "messages" 通过向其中追加消息来更新状态中的键。该 MessagesValue reducer 自动处理此操作：

// MessagesValue already has a built-in reducer
const State = new StateSchema({
  messages: MessagesValue,  // [!code highlight]
  extraField: z.number(),
});

我们的节点可以简单地返回新消息（reducer 处理连接）：

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

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 来处理这些考虑事项：

const State = new StateSchema({  // [!code highlight]
  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" };  // [!code highlight]

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 以方便使用，我们可以这样：

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

定义输入和输出模式

默认情况下， StateGraph 使用单一模式运行，所有节点都期望使用该模式进行通信。但是，也可以为图定义不同的输入和输出模式。

当指定了不同的模式时，节点之间的通信仍会使用内部模式。输入模式确保提供的输入符合预期结构，而输出模式则根据定义的输出模式过滤内部数据，仅返回相关信息。

下面，我们将了解如何定义不同的输入和输出模式。

// Define the schema for the input
const InputState = new StateSchema({
  question: z.string(),
});

// Define the schema for the output
const OutputState = new StateSchema({
  answer: z.string(),
});

// Define the overall schema, combining both input and output
const OverallState = new StateSchema({
  question: z.string(),
  answer: z.string(),
});

// Define the node that processes the input
const answerNode: GraphNode<typeof OverallState> = (state) => {
  // Example answer and an extra key
  return { answer: "bye", question: state.question };
};

// Build the graph with input and output schemas specified
const graph = new StateGraph({
  input: InputState,
  output: OutputState,
  state: OverallState,
})
  .addNode("answerNode", answerNode)
  .addEdge(START, "answerNode")
  .addEdge("answerNode", END)
  .compile();

// Invoke the graph with an input and print the result
console.log(await graph.invoke({ question: "hi" }));

{ answer: 'bye' }

请注意，invoke 的输出仅包含输出模式。

在节点之间传递私有状态

In some cases, you may want nodes to exchange information that is crucial for intermediate logic but doesn't need to be part of the main schema of the graph. This private data is not relevant to the overall input/output of the graph and should only be shared between certain nodes.

下面，我们将创建一个由三个节点（节点_1、节点_2 和节点_3）组成的有序图示例，其中私有数据在前两个步骤（节点_1 和节点_2）之间传递，而第三个步骤（节点_3）只能访问公开的总体状态。

// The overall state of the graph (this is the public state shared across nodes)
const OverallState = new StateSchema({
  a: z.string(),
});

// Output from node1 contains private data that is not part of the overall state
const Node1Output = new StateSchema({
  privateData: z.string(),
});

// Node 2 input only requests the private data available after node1
const Node2Input = new StateSchema({
  privateData: z.string(),
});

// The private data is only shared between node1 and 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 only has access to the overall state (no access to private data from 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;
};

// Connect nodes in a sequence
// node2 accepts private data from node1, whereas
// node3 does not see the private data.
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();

// Invoke the graph with the initial state
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 支持多种其他方法。本节涵盖所有可用选项。

Channels API

channels API 提供了对状态管理的底层控制。LangGraph 提供了多种内置 channel 类型：

Channel 类型	行为	使用场景
`LastValue`	存储最新的值	被覆盖的简单字段
`BinaryOperatorAggregate`	使用 reducer 函数合并值	累积值（计数器、列表）
`Topic`	将所有值收集到序列中	事件流、审计日志
`EphemeralValue`	在 supersteps 之间重置的值	临时计算状态

使用对象简写语法：

当你传递一个带有 reducer 和 default的对象时，它会创建一个 BinaryOperatorAggregate channel。传递 null 会创建一个 LastValue channel:

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

const workflow = new StateGraph({
  channels: {
    // BinaryOperatorAggregate: combines values with a reducer
    messages: {
      reducer: (current, update) => current.concat(update),
      default: () => [],
    },
    // LastValue: stores the most recent value (null = no reducer)
    question: null,
    answer: null,
  },
});

直接使用 channel 类：

为了更精细的控制，你可以直接实例化 channel 类：

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

const workflow = new StateGraph({
  channels: {
    messages: new BinaryOperatorAggregate<BaseMessage[]>(
      (current, update) => current.concat(update),
      () => []
    ),
    question: new LastValue<string>(),
    // Topic collects all values pushed during execution
    events: new Topic<string>(),
  },
});

Annotation.Root

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

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() schema 来定义状态。LangGraph 通过 .langgraph 插件扩展了 Zod v3，提供了 .reducer() 和 .metadata() methods:

const State = z.object({
  // Use .langgraph.reducer() to attach a reducer function
  messages: z
    .array(z.custom())
    .default([])
    .langgraph.reducer(messagesStateReducer),
  // Simple fields work directly (last-write-wins)
  question: z.string().optional(),
  answer: z.string().optional(),
  // Custom reducer for accumulating values
  count: z
    .number()
    .default(0)
    .langgraph.reducer((current, update) => current + update),
});

const graph = new StateGraph(State);

Zod v4 中的 Zod 对象

Zod v4 使用基于注册表的方式。使用 LangGraph 注册表为 schema 字段附加元数据：

const State = z.object({
  // Use .register() with the LangGraph registry and MessagesZodMeta
  messages: z
    .array(z.custom())
    .default([])
    .register(registry, MessagesZodMeta),
  // Simple fields work directly (last-write-wins)
  question: z.string().optional(),
  answer: z.string().optional(),
  // Custom reducer via registry metadata
  count: z
    .number()
    .default(0)
    .register(registry, { reducer: (current: number, update: number) => current + update }),
});

const graph = new StateGraph(State);

对比表

方法	Reducer	类型安全	Zod 版本	推荐
`StateSchema`	✅ 内置	✅ 完整	v3 或 v4	✅ 是
Channels API	✅ Manual	⚠️ Partial	N/A	For advanced cases
`Annotation.Root`	✅ Built-in	✅ Full	N/A	Legacy
Zod v3 + `.langgraph`	✅ 通过插件	✅ 完整	仅 v3	旧版
Zod v4 + 注册表	✅ 通过注册表	✅ 完整	仅 v4	旧版

添加运行时配置

有时你可能希望能够在调用图时对其进行配置。例如，你可能希望在运行时指定要使用的 LLM 或系统提示， _而不污染图状态中的这些参数_.

添加运行时配置的方法：

为配置指定 schema
将配置添加到节点或条件边的函数签名中
将配置传入图中。

下面是简单示例：

// 1. Specify config schema
const ContextSchema = z.object({
  myRuntimeValue: z.string(),
});

// 2. Define a graph that accesses the config in a node
const State = new StateSchema({
  myStateValue: z.number(),
});

const node: GraphNode<typeof State> = (state, runtime) => {
  if (runtime?.context?.myRuntimeValue === "a") {  // [!code highlight]
    return { myStateValue: 1 };
  } else if (runtime?.context?.myRuntimeValue === "b") {  // [!code highlight]
    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. Pass in configuration at runtime:
console.log(await graph.invoke({}, { context: { myRuntimeValue: "a" } }));  // [!code highlight]
console.log(await graph.invoke({}, { context: { myRuntimeValue: "b" } }));  // [!code highlight]

{ myStateValue: 1 }
{ myStateValue: 2 }

Extended example: specifying LLM at runtime

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

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

  // Usage
  const inputMessage = { role: "user", content: "hi" };
  // With no configuration, uses default (Anthropic)
  const response1 = await graph.invoke({ messages: [inputMessage] });
  // Or, can set 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

Extended example: specifying model and system message at runtime

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

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

  // Usage
  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 对象及其默认参数，并将其与一个节点关联：

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

默认情况下，重试策略在任何异常时重试，除了以下情况：

* TypeError
* SyntaxError
* ReferenceError

Extended example: customizing retry policies

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

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

  // Create an in-memory database
  const db: typeof Database.prototype = new Database(":memory:");

  const model = new ChatAnthropic({ model: "claude-3-5-sonnet-20240620" });

  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)
    // Define the two nodes we will cycle between
    .addNode("call_model", callModel, { retryPolicy: { maxAttempts: 5 } })
    .addNode("query_database", queryDatabase, {
      retryPolicy: {
        retryOn: (e: any): boolean => {
          if (e instanceof Database.SqliteError) {
            // Retry on "SQLITE_BUSY" error
            return e.code === "SQLITE_BUSY";
          }
          return false; // Don't retry on other errors
        },
      },
    })
    .addEdge(START, "call_model")
    .addEdge("call_model", "query_database")
    .addEdge("query_database", END);

  const graph = workflow.compile();

在节点内访问执行信息

您可以通过以下方式访问执行标识和重试信息 runtime.executionInfo。这会显示线程、运行和检查点标识符以及重试状态，无需从 config directly.

属性	类型	描述
`threadId`	`string \	undefined`
`runId`	`string \	undefined`
`checkpointId`	`string`	当前执行的检查点 ID。
`checkpointNs`	`string`	当前执行的检查点命名空间。
`taskId`	`string`	当前执行的任务 ID。
`nodeAttempt`	`number`	当前执行尝试次数（从 1 开始）。
`nodeFirstAttemptTime`	`number \	undefined`

访问线程和运行 ID

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

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}`);  // [!code highlight]
  return { result: "done" };
};

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

根据重试状态调整行为

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

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

const myNode: GraphNode<typeof State> = async (state, runtime) => {
  const info = runtime.executionInfo;
  if (info.nodeAttempt > 1) {  // [!code highlight]
    // use a fallback on retries
    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}`);  // [!code highlight]
    if (server.user != null) {
      console.log(`User: ${server.user.identity}`);
    }
  }
  return { result: "done" };
};

serverInfo is null 当图不在 LangGraph Server 上运行时。

创建步骤序列

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

如何构建顺序图
用于构造类似图的内置简写方法。

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

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

Why split application steps into a sequence with LangGraph?

LangGraph 让为您的应用程序添加底层持久层变得很容易。这允许在节点执行之间对状态进行检查点保存，因此您的 LangGraph 节点可以控制：

* 状态更新如何进行检查点保存
* 如何在 human-in-the-loop 工作流
* 中恢复中断我们如何使用 LangGraph 的时间旅行功能特性

它们还决定了执行步骤的流式传输方式，以及如何使用 Studio.

让我们演示一个端到端的示例。我们将创建一个包含三个步骤的序列：

在状态的键中填充一个值
更新相同的值
填充一个不同的值

让我们首先定义我们的状态。这控制着图的模式，也可以指定如何应用更新。请参阅使用 reducer 处理状态更新了解更多详情。

在我们的例子中，我们只跟踪两个值：

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

最后，我们定义图。我们使用 StateGraph 来定义一个在此状态上操作的图。

然后我们将使用 addNode 和 addEdge 来填充我们的图并定义其控制流。

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 节点.
* 当没有更多节点可执行时，图停止运行。

接下来我们编译我们的图。这会对图的结构进行一些基本检查（例如，识别孤立的节点）。如果我们要通过 checkpointer为应用程序添加持久化功能，也会在此处传入。

LangGraph 提供了内置的图形可视化工具。让我们检查一下我们的序列。参见可视化您的图了解可视化详情。

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 原生支持节点的并行执行，这可以显著增强基于图的工作流程的性能。这种并行化通过扇出和扇入机制实现，利用标准边和条件_边。以下是一些示例，展示如何创建适合您的分支数据流。

并行运行图节点

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

const State = new StateSchema({
  // The reducer makes this append-only
  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();

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

Exception handling?

LangGraph 在 superstep中执行节点，这意味着虽然并行分支是并行执行的，但整个 superstep 是 **事务性的**。如果这些分支中的任何一个抛出异常， **所有** 更新都不会应用到状态（整个 superstep 报错）。

重要的是，当使用 checkpointer时，superstep 内成功节点的结果会被保存，恢复时不会重复执行。

如果您有容易出错的节点（也许想处理不稳定的 API 调用），LangGraph 提供了两种方式来解决这个问题：

您可以在节点内编写常规 Python 代码来捕获和处理异常。
您可以设置 **retry_策略** 来指示图重试抛出某些类型异常的节点。只有失败的分支会被重试，因此您无需担心执行冗余工作。

结合使用，您可以进行并行执行并完全控制异常处理。

建议

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

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

条件分支

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

const State = new StateSchema({
  aggregate: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
  // Add a key to the state. We will set this key to determine
  // how we branch.
  which: z.string(),  // [!code highlight]
});

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"] };  // [!code highlight]
};

const conditionalEdge: ConditionalEdgeRouter<typeof State, "b" | "c"> = (state) => {
  // Fill in arbitrary logic here that uses the state
  // to determine the next node
  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();

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 和其他高级分支模式。下面是一个如何使用它的示例：

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

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

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

// Call the graph: here we call it to generate a list of jokes
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 节点。

您还可以在调用或流式传输图时设置图的递归限制。递归限制设置了图在引发错误之前允许执行的 super-steps 数量。了解更多关于递归限制概念.

让我们考虑一个带有循环的简单图，以便更好地理解这些机制是如何工作的。

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

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，您可以捕获并处理它：

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

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

const State = new StateSchema({
  // The reducer makes this append-only
  aggregate: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});

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

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

// Define edges
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();

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

Node A sees []
Node B sees ['A']
Node A sees ['A', 'B']
Node B sees ['A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B']
Node B sees ['A', 'B', 'A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B', 'A', 'B']
{ aggregate: ['A', 'B', 'A', 'B', 'A', 'B', 'A'] }

设置递归限制

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

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

Node A sees []
Node B sees ['A']
Node A sees ['A', 'B']
Node B sees ['A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B']
Recursion Error

使用以下方式结合控制流和状态更新 `Command`

结合控制流（边）和状态更新（节点）可能很有用。例如，您可能希望在同一节点中同时执行状态更新并决定下一步进入哪个节点。LangGraph 通过从节点函数返回 Command 对象来提供一种方式：

const myNode = (state: State): Command => {
  return new Command({
    // state update
    update: { foo: "bar" },
    // control flow
    goto: "myOtherNode"
  });
};

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

// Define graph state
const State = new StateSchema({
  foo: z.string(),
});

// Define the nodes

const nodeA: GraphNode<typeof State, "nodeB" | "nodeC"> = (state) => {
  console.log("Called A");
  const value = Math.random() > 0.5 ? "b" : "c";
  // this is a replacement for a conditional edge function
  const goto = value === "b" ? "nodeB" : "nodeC";

  // note how Command allows you to BOTH update the graph state AND route to the next node
  return new Command({
    // this is the state update
    update: { foo: value },
    // this is a replacement for an edge
    goto,
  });
};

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

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

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

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

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

Called A
Called C
{ foo: 'cc' }

导航到父图中的节点

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

const myNode = (state: State): Command => {
  return new Command({
    update: { foo: "bar" },
    goto: "otherSubgraph",  // where `otherSubgraph` is a node in the parent graph
    graph: Command.PARENT
  });
};

让我们使用上面的例子来演示。我们将通过更改 nodeA 在上面的例子中，将其转换为一个单节点图，然后作为子图添加到我们的父图中。

const State = new StateSchema({
  // NOTE: we define a reducer here
  foo: new ReducedValue(  // [!code highlight]
    z.string().default(""),
    { reducer: (x, y) => x + y }
  ),
});

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

  // note how Command allows you to BOTH update the graph state AND route to the next node
  return new Command({
    update: { foo: "a" },  // [!code highlight]
    goto: value,
    // this tells LangGraph to navigate to nodeB or nodeC in the parent graph
    // NOTE: this will navigate to the closest parent graph relative to the subgraph
    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("Called B");  // [!code highlight]
  // NOTE: since we've defined a reducer, we don't need to manually append
  // new characters to existing 'foo' value. instead, reducer will append these
  // automatically
  return { foo: "b" };
};  // [!code highlight]

const nodeC: GraphNode<typeof State> = (state) => {
  console.log("Called 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);

Called A
Called C
{ foo: 'ac' }

在工具内部使用

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

const lookupUserInfo = tool(
  async (input, runtime) => {
    const userId = runtime.serverInfo?.user?.identity;  // [!code highlight]
    const userInfo = getUserInfo(userId);
    return new Command({
      update: {
        // update the state keys
        userInfo: userInfo,
        // update the message history
        messages: [{
          role: "tool",
          content: "Successfully looked up user information",
          tool_call_id: runtime.toolCall.id
        }]
      }
    });
  },
  {
    name: "lookupUserInfo",
    description: "Use this to look up user information to better assist them with their questions.",
    schema: z.object({}),
  }
);

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

可视化您的图

这里我们演示如何可视化您创建的图。

您可以可视化任意图，包括状态图.

让我们创建一个简单的示例图来演示可视化。

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

如果需要，我们可以将图渲染为 .png. 这使用 Mermaid.ink API 来生成图表。

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

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

设置

定义和更新状态

定义状态

更新状态

使用归约器处理状态更新

MessagesValue

定义输入和输出模式

在节点之间传递私有状态

替代状态定义

Channels API

Annotation.Root

Zod v3 中的 Zod 对象

Zod v4 中的 Zod 对象

对比表

添加运行时配置

Extended example: specifying LLM at runtime

Extended example: specifying model and system message at runtime

添加重试策略

Extended example: customizing retry policies

在节点内访问执行信息

访问线程和运行 ID

根据重试状态调整行为

在节点内访问服务器信息

创建步骤序列

Why split application steps into a sequence with LangGraph?

创建分支

并行运行图节点

Exception handling?

条件分支

Map-Reduce 和 Send API

创建和控制循环

设置递归限制

使用以下方式结合控制流和状态更新 Command

导航到父图中的节点

在工具内部使用

可视化您的图

Mermaid

PNG

使用以下方式结合控制流和状态更新 `Command`