Graph API 概述 - 以编程方式使用文档

图

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

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

Nodes：对代理逻辑进行编码的函数。它们接收当前状态作为输入，执行一些计算或副作用，并返回更新后的状态。

Edges：确定下一步执行哪个 Node 的函数，它们基于当前状态。它们可以是条件分支或固定转换。

通过组合 Nodes 和 Edges，您可以创建随时间演化状态的复杂循环工作流。不过，真正的力量来自于 LangGraph 如何管理该状态。

需要强调的是： Nodes 和 Edges 不过是函数而已——它们可以包含 LLM 或只是普通的代码。

简而言之： _节点做工作，边决定下一步做什么_.

LangGraph 的底层图算法使用消息传递来定义通用程序。当节点完成其操作时，它会沿一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将结果消息传递给下一组节点，过程继续进行。受 Google 的 Pregel 系统启发，程序以离散的"超级步"进行。

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

StateGraph

StateGraph 类是使用的主要图类。它由用户定义的 State object.

编译您的图

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

编译是一个非常简单的步骤。它对图的结构提供一些基本检查（没有孤立节点等）。这也是您可以指定运行时参数（如检查点和断点）的地方。您只需调用 .compile method:

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

模式

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

- A **标准模式** 用于简单字段（成为"最新值"通道，更新时覆盖）
- A **ReducedValue** 用于需要自定义归约函数的字段（当节点并行运行时）
- A **MessagesValue** 用于聊天消息列表（预构建的消息感知归约器）
- An **UntrackedValue** 用于不应被检查点保存的临时状态

  StateSchema,
  ReducedValue,
  MessagesValue,
  UntrackedValue
} from "@langchain/langgraph";


const AgentState = new StateSchema({
  // Prebuilt messages value with built-in reducer
  messages: MessagesValue,

  // Simple fields use Zod schemas directly
  currentStep: z.string(),

  // Fields with defaults
  retryCount: z.number().default(0),

  // Custom reducer for accumulating values
  allSteps: new ReducedValue(
    z.array(z.string()).default(() => []),
    {
      inputSchema: z.string(),
      reducer: (current, newStep) => [...current, newStep],
    }
  ),

  // Transient state not saved to checkpoints
  tempCache: new UntrackedValue(z.record(z.string(), z.unknown())),
});

// Type extraction
type State = typeof AgentState.State;   // Full state type
type Update = typeof AgentState.Update; // Partial update type

// Use in graph
const graph = new StateGraph(AgentState)
  .addNode("myNode", ...)
  .compile();

默认情况下，图的输入和输出模式相同。如果你想更改这一点，你也可以直接指定明确的输入和输出模式。当你有大量键，且有些键专门用于输入而其他用于输出时，这很有用。

多个模式

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

- Internal nodes can pass information that is not required in the graph's input / output.
- We may also want to use different input / output schemas for the graph. The output might, for example, only contain a single relevant output key.

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

还可以为图定义明确的输入和输出模式。在这种情况下，我们定义一个包含 _所有_ 与图操作相关的键的"内部"模式。但是，我们还定义 input 和 output 模式作为"内部"模式的子集来约束图的输入和输出。参见定义输入和输出模式了解更多详情。

让我们看一个例子：

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

我们传递 state 作为输入模式给 node1。但是，我们写入 foo，它是 OverallState中的一个通道。如何写入不在输入模式中的状态通道？这是因为节点 _可以写入图状态中的任何状态通道。_ 图状态是初始化时定义的状态通道的并集，其中包括 OverallState 和过滤器 InputState 和 OutputState.

我们用 StateGraph({ state: OverallState, input: InputState, output: OutputState })初始化图。如何写入 PrivateState in node2？如果模式没有在 StateGraph 初始化时传递，图如何访问这个模式？我们可以这样做，因为 _节点也可以声明额外的状态通道_ 只要状态模式定义存在。在这种情况下， PrivateState 模式已定义，因此我们可以添加 bar 作为图中的一个新状态通道并向其写入。

警告

私有通道在流式传输时不会被删除。

输入、输出和私有模式约束每个节点 _读取_ （其输入模式）以及什么 invoke _返回_ （输出模式）。它们会 **不** 从...隐藏通道 stream.

当你使用流式传输时 streamMode: "values"，图会发出 **所有** 的状态通道（默认情况下包括私有通道），因为值流式传输默认为完整的状态通道集而不是输出模式。这就是为什么像这样的私有通道 bar 被隐藏 invoke ，但在流式传输时可见：

要限制流式传输的值到特定的通道集（例如仅输出模式），请传递 outputKeys:

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

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

归约器

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

默认归约器

这两个示例展示了如何使用默认归约器：

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

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

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

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 用于在图形执行期间存在但 **永远不应被检查点化**的状态字段。当图形从检查点恢复时，未跟踪的值将重置为其初始状态（或不可用）。

这对于以下场景很有用： - **无法序列化的** 数据库连接 - **临时缓存** 应该在恢复时重新构建 - **大对象** 你不想持久化的 - **仅运行时配置** 每次应重新传递

const State = new StateSchema({
  messages: MessagesValue,

  // Untracked: throws if multiple nodes write in same step (guard: true is default)
  dbConnection: new UntrackedValue(),

  // Untracked with guard: false allows multiple writes, keeps last value
  tempCache: new UntrackedValue(
    z.record(z.string(), z.unknown()),
    { guard: false }
  ),

  // Untracked without a schema (for maximum flexibility)
  runtimeConfig: new UntrackedValue(),
});

Behavior: - 执行期间：值像普通状态一样存储和访问 - 在检查点：未跟踪的值 **被排除** 在检查点数据之外 - 恢复时：未跟踪的值从头开始（空或使用默认值） - 使用 guard: true （默认）：如果多个节点在同一步骤写入则抛出错误 - 使用 guard: false：允许多次写入，最后的值获胜

类型工具

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

`GraphNode`

使用 GraphNode 来为在图构建器外部定义的节点函数添加类型：

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

// Basic node - receives state, returns partial update
const incrementNode: GraphNode<typeof State> = (state) => {
  return { count: state.count + 1 };
};

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

// Node with Command routing - specify valid destinations
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(),
});

// These are equivalent:
const myNode1: GraphNode<typeof State> = (state) => ({ step: "done" });
const myNode2: typeof State.Node = (state) => ({ step: "done" });

`ConditionalEdgeRouter`

使用 ConditionalEdgeRouter 用于条件边中的路由函数（不更新状态，仅做路由）：

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

// Router returns node name(s) or END
const router: ConditionalEdgeRouter<typeof State, "process" | "summarize"> = (state) => {
  if (!state.shouldContinue) {
    return END;
  }
  return state.step === "initial" ? "process" : "summarize";
};

// Use in graph
graph.addConditionalEdges("check", router);

`StateSchema.State` 和 `StateSchema.Update`

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

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

// Extract the full state type
type MyState = typeof MyStateSchema.State;
// { messages: BaseMessage[], count: number }

// Extract the update type (partial, with reducer input types)
type MyUpdate = typeof MyStateSchema.Update;
// { messages?: Messages, count?: number }

在图状态中处理消息

为什么要使用消息？

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

要了解更多关于消息对象的信息，请参阅消息概念指南.

在图中使用消息

在很多情况下，将之前的对话历史作为消息列表存储在您的图状态中是非常有用的。要做到这一点，您可以使用预构建的 MessagesValue 它提供了一个支持消息的reducer，能够自动处理消息ID、更新和删除。

这个 MessagesValue reducer对于告诉图如何更新状态中的 Message 对象列表至关重要，每次状态更新时都会根据reducer的指示进行更新。如果没有指定reducer，每次状态更新都会用最近提供的值覆盖整个消息列表。 MessagesValue 正确处理这种情况：对于全新的消息，它会追加到现有列表；对于现有消息（通过ID匹配），它会在原地更新它们。

序列化

除了跟踪消息ID之外， MessagesValue 还会尝试将消息反序列化为LangChain Message 对象，只要在 messages channel. This allows sending graph inputs / state updates in the following format:

// this is supported
{
  messages: [new HumanMessage("message")];
}

// and this is also supported
{
  messages: [{ role: "human", content: "message" }];
}

由于状态更新总是被反序列化为LangChain Messages 当使用 MessagesValue时，您应该使用点符号来访问消息属性，例如 state.messages.at(-1).content。以下是使用 MessagesValue:

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

const graph = new StateGraph(State)
  ...

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

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

节点

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

state—图表的状态状态
config—A RunnableConfig 对象，包含配置信息如 thread_id 和追踪信息如 tags

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

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

// Option 1: Use GraphNode type utility
const myNode: GraphNode<typeof State> = (state, config) => {
  console.log("In node: ", config?.configurable?.user_id);
  return { results: `Hello, ${state.input}!` };
};

// Option 2: Use State.Node shorthand
const otherNode: typeof State.Node = (state) => {
  return state;
};

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

在底层，函数会被转换为 RunnableLambda，这会为您的函数添加批量和异步支持，以及原生追踪和调试.

如果您向图表添加节点时未指定名称，则会使用与函数名相同的默认名称。

builder.addNode(myNode);
// You can then create edges to/from this node by referencing it as `"myNode"`

重新执行和幂等性

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

Idempotency. 设计 **节点** 逻辑，使重新执行不会破坏状态。如果节点插入数据库行，运行它两次不应创建重复行，除非是有意的。使用幂等性键、upserts 或读写前检查。对于周围的副作用 interrupt()，参见图更改之前调用的副作用 interrupt 必须是幂等的.

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

节点内的任务和中断。 If a **节点** 调用 **任务** or interrupt, 恢复时适用更严格的确定性规则。LangGraph 恢复已完成的 **任务** 从检查点的结果，但更改 **任务** or interrupt 恢复点之前代码中的顺序可能导致缓存值不匹配。A Functional API **入口点** 编译为单个 **节点** 以这种方式运行整个入口点方法。参见确定性, 幂等性，以及在节点中使用任务.

在节点中使用任务

If a 节点包含多个操作，您可能会发现将每个操作实现为 **任务** 而不是将逻辑分散到多个节点。当图使用检查点时，任务结果会被保存，因此恢复线程可以跳过已完成的 **任务** 节点内的工作。

Original

With task

`START` 节点

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

graph.addEdge(START, "nodeA");

`END` 节点

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

graph.addEdge("nodeA", END);

节点缓存

LangGraph supports caching of tasks/nodes based on the input to the node. To use caching:

- 在编译图时指定缓存（或指定入口点）
- 为节点指定缓存策略。每个缓存策略支持：
- keyFunc，用于根据节点的输入生成缓存键。
- ttl，缓存的生存时间（秒）。如果不指定，缓存永不过期。

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

const expensiveNode: GraphNode<typeof State> = async (state) => {
  // Simulate an expensive operation
  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" });   // [!code highlight]
// [{"expensive_node": {"result": 10}}]
await graph.invoke({ x: 5 }, { streamMode: "updates" });   // [!code highlight]
// [{"expensive_node": {"result": 10}, "__metadata__": {"cached": true}}]

边

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

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

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

普通边

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

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

条件边

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

graph.addConditionalEdges("nodeA", routingFunction);

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

默认情况下，返回值 routingFunction 用作下一个要发送状态的节点（或节点列表）的名称。所有这些节点将作为下一个超级步的一部分并行运行。

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

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

入口点

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

graph.addEdge(START, "nodeA");

条件入口点

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

graph.addConditionalEdges(START, routingFunction);

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

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

`Send`

默认情况下， Nodes 和 Edges are defined ahead of time and operate on the same shared state. However, there can be cases where the exact edges are not known ahead of time and/or you may want different versions of State 同时存在。一个常见的例子是map-reduce设计模式。在这种设计模式中，第一个节点可能生成一个对象列表，而你可能希望将其他节点应用到所有这些对象上。对象的数量可能事先未知（这意味着边的数量可能未知），输入 State 到下游 Node 应该不同（每个生成的对象一个）。

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

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

`Command`

Command是一个用于控制图执行的多功能原语。它接受四个参数：

- update：应用状态更新（类似于从节点返回更新）。
- goto：导航到特定节点（类似于条件边).
- graph：从子图.
- resume：在中断.

Command 后在三种上下文中使用：

- **从节点返回**：使用 update, goto，和 graph 将状态更新与控制流结合。
- **输入到 invoke or stream**：使用 resume 在中断后继续执行。
- **从工具返回**：与从节点返回类似，从工具内部结合状态更新和控制流。

从节点返回

`update` 和 `goto`

返回Command从节点函数更新状态并一步路由到下一个节点：

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

使用 Command 你还可以实现动态控制流行为（与条件边):

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

使用 Command 当你需要 **同时** 更新状态 **和** 路由到不同的节点。如果只需要路由而不更新状态，请使用条件边 instead.

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

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

查看这个操作指南，了解如何使用 Command.

`graph`

如果你正在使用子图，你可以通过指定 graph: Command.PARENT in Command:

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

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

输入到 `invoke` or `stream`

警告

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

// WRONG - graph resumes from the latest checkpoint
// (last step that ran), appears stuck
await graph.invoke(new Command({ update: { messages: [{ role: "user", content: "follow up" }] } }), config);  // [!code --]

// CORRECT - plain object restarts from __start__
await graph.invoke({ messages: [{ role: "user", content: "follow up" }] }, config);  // [!code ++]

`resume`

使用 new Command({ resume: ... }) 在中断后提供值并恢复图执行中断传递给 resume 的值成为 interrupt() 调用的返回值：

const humanReview = async (state: typeof StateAnnotation.State) => {
  // Pauses the graph and waits for a value
  const answer = interrupt("Do you approve?");
  return { messages: [{ role: "user", content: answer }] };
};

// First invocation - hits the interrupt and pauses
const result = await graph.invoke({ messages: [...] }, config);

// Resume with a value - the interrupt() call returns "yes"
const resumed = await graph.invoke(new Command({ resume: "yes" }), config);

请参阅中断概念指南了解更多关于中断模式的详细信息，包括多个中断和验证循环。

从工具返回值

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

请参阅在工具内使用了解更多。

图迁移

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

- 对于位于图末尾的线程（即未被中断的线程），您可以更改整个图的拓扑结构（即所有节点和边，删除、添加、重命名等）
- For threads currently interrupted, we support all topology changes other than renaming / removing nodes (as that thread could now be about to enter a node that no longer exists) -- if this is a blocker please reach out and we can prioritize a solution.
- 对于状态修改，我们对键的添加和删除具有完整的向后和向前兼容性
- 重命名的状态键会丢失其在现有线程中的保存状态
- 以不兼容方式更改类型的状态键可能会导致在包含更改前状态的线程中出现一些问题——如果这成为阻碍，请联系我们，我们可以优先解决。

运行时上下文

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

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

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

await graph.invoke(inputs, config);

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

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}!` };
});

递归限制

递归限制设置图在单次执行中可执行的最大 super-steps 步数。一旦达到限制，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 在图形执行时递增此计数器，并在 GraphRecursionError 一旦超过配置的 recursionLimit 时抛出异常。

访问当前步骤计数器

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

const myNode: GraphNode<typeof State> = async (state, config) => {
  const currentStep = config.metadata?.langgraph_step;
  console.log(`Currently on step: ${currentStep}`);
  return state;
}

设计您的图形时加入明确的终止条件，并将捕获 GraphRecursionError 作为安全网：

  StateGraph,
  StateSchema,
  ReducedValue,
  GraphNode,
  ConditionalEdgeRouter,
  END,
  GraphRecursionError
} from "@langchain/langgraph";


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

// Build graph with explicit termination logic
const graph = new StateGraph(State)
  .addNode("reasoning", async (state) => {
    // Normal processing - design your graph with explicit termination conditions
    return {
      messages: ["thinking..."]
    };
  })
  .addConditionalEdges("reasoning", (state) => {
    // Add your termination condition here
    if (state.messages.length >= 5) {
      return END;
    }
    return "reasoning";
  });

const app = graph.compile();

// Catch GraphRecursionError as a safety net
try {
  const result = await app.invoke(
    { messages: [] },
    { recursionLimit: 10 }
  );
} catch (error) {
  if (error instanceof GraphRecursionError) {
    console.log("Recursion limit reached, handling gracefully");
    // Handle the error - return partial results, notify user, etc.
  }
}

主动式与反应式方法

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

  StateGraph,
  StateSchema,
  ReducedValue,
  GraphNode,
  ConditionalEdgeRouter,
  END,
  GraphRecursionError
} from "@langchain/langgraph";


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


// Build graph with explicit termination logic
const builder = new StateGraph(State)
  .addNode("agent", async (state) => {
    return {
      messages: ["Processing..."]
    };
  })
  .addConditionalEdges("agent", (state) => {
    // Design termination conditions into your graph
    if (state.messages.length >= 5) {
      return END;
    }
    return "agent";
  });

const graph = builder.compile();

// Reactive Approach - catch GraphRecursionError as safety net
try {
  const result = await graph.invoke(
    { messages: [] },
    { recursionLimit: 10 }
  );
} catch (error) {
  if (error instanceof GraphRecursionError) {
    // Handle externally after graph execution fails
    console.log("Recursion limit exceeded, handling gracefully");
  }
}

反应式方法在超过限制后捕获 GraphRecursionError 。设计您的图形时加入明确的终止条件，以首先避免达到限制。

方法	检测	处理	控制流
反应式（捕获 `GraphRecursionError`)	After limit exceeded	Outside graph in try/catch	Graph execution terminated

反应式优势：

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

其他可用元数据

连同 langgraph_step，以下元数据也可在 config.metadata:

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

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

  return state;
}

可视化

能够可视化图形通常很方便，尤其是当它们变得更加复杂时。LangGraph 提供了多种内置的可视化方式。请参阅可视化您的图了解更多

可观测性与追踪

要追踪、调试和评估您的代理，请使用 LangSmith.

图

StateGraph

编译您的图

状态

模式

多个模式

归约器

默认归约器

未跟踪的值

类型工具

GraphNode

State.Node 简写

ConditionalEdgeRouter

StateSchema.State 和 StateSchema.Update

在图状态中处理消息

为什么要使用消息？

在图中使用消息

序列化

节点

重新执行和幂等性

在节点中使用任务

Original

With task

START 节点

END 节点

节点缓存

边

普通边

条件边

入口点

条件入口点

Send

Command

从节点返回

update 和 goto

graph

输入到 invoke or stream

resume

从工具返回值

图迁移

运行时上下文

递归限制

访问和处理递归计数器

工作原理

访问当前步骤计数器

主动式与反应式方法

其他可用元数据

可视化

可观测性与追踪

了解更多

`GraphNode`

`State.Node` 简写

`ConditionalEdgeRouter`

`StateSchema.State` 和 `StateSchema.Update`

`START` 节点

`END` 节点

`Send`

`Command`

`update` 和 `goto`

`graph`

输入到 `invoke` or `stream`

`resume`