如何添加动态断点

注意

对于**人机协作**工作流，请使用新的 interrupt() 函数。请查阅 人机协作概念指南，了解更多关于 interrupt 设计模式的信息。

先决条件

本指南假定您熟悉以下概念

• 断点
• LangGraph 术语表

人机协作 (HIL) 交互对于智能体系统至关重要。断点是一种常见的 HIL 交互模式，它允许图在特定步骤停止，并在继续之前寻求人类的批准（例如，对于敏感操作）。

在 LangGraph 中，您可以在节点执行之前/之后添加断点。但很多时候，根据某些条件在给定节点内部**动态地**中断图可能会很有帮助。这样做时，包含有关中断**原因**的信息也可能有所帮助。

本指南展示了如何使用 NodeInterupt（一个可以从节点内部抛出的特殊异常）动态地中断图。让我们看看它的实际应用！

定义图

import {
  Annotation,
  MemorySaver,
  NodeInterrupt,
  StateGraph,
} from "@langchain/langgraph";

const StateAnnotation = Annotation.Root({
  input: Annotation<string>,
});

const step1 = async (state: typeof StateAnnotation.State) => {
  console.log("---Step 1---");
  return state;
};

const step2 = async (state: typeof StateAnnotation.State) => {
  // Let's optionally raise a NodeInterrupt
  // if the length of the input is longer than 5 characters
  if (state.input?.length > 5) {
    throw new NodeInterrupt(`Received input that is longer than 5 characters: ${state.input}`);
  }
  console.log("---Step 2---");
  return state;
};

const step3 = async (state: typeof StateAnnotation.State) => {
  console.log("---Step 3---");
  return state;
};

const checkpointer = new MemorySaver();

const graph = new StateGraph(StateAnnotation)
  .addNode("step1", step1)
  .addNode("step2", step2)
  .addNode("step3", step3)
  .addEdge("__start__", "step1")
  .addEdge("step1", "step2")
  .addEdge("step2", "step3")
  .addEdge("step3", "__end__")
  .compile({ checkpointer });

import * as tslab from "tslab";

const representation = graph.getGraph();
const image = await representation.drawMermaidPng();
const arrayBuffer = await image.arrayBuffer();

await tslab.display.png(new Uint8Array(arrayBuffer));

使用动态中断运行图

首先，让我们使用一个长度小于等于 5 个字符的输入来运行图。这应该会安全地忽略我们定义的中断条件，并在图执行结束时返回原始输入。

const initialInput = { input: "hello" };
const config = {
  configurable: {
    thread_id: "1",
  },
  streamMode: "values" as const,
};

const stream = await graph.stream(initialInput, config);

for await (const event of stream) {
  console.log(event);
}

{ input: 'hello' }
---Step 1---
{ input: 'hello' }
---Step 2---
{ input: 'hello' }
---Step 3---
{ input: 'hello' }

此时检查图，我们可以看到没有剩余的任务需要运行，并且图确实已经完成了执行。

const state = await graph.getState(config);
console.log(state.next);
console.log(state.tasks);

[]
[]

现在，让我们使用一个长度大于 5 个字符的输入来运行图。这应该会通过在 step2 节点内部抛出 NodeInterrupt 错误来触发我们定义的动态中断。

const longInput = { input: "hello world" };
const config2 = {
  configurable: {
    thread_id: "2",
  },
  streamMode: "values" as const,
};

const streamWithInterrupt = await graph.stream(longInput, config2);

for await (const event of streamWithInterrupt) {
  console.log(event);
}

{ input: 'hello world' }
---Step 1---
{ input: 'hello world' }

我们可以看到图现在在执行 step2 时停止了。此时检查图状态，我们可以看到有关下一个要执行的节点（step2）、引发中断的节点（也是 step2）以及有关中断的其他信息。

const state2 = await graph.getState(config2);
console.log(state2.next);
console.log(JSON.stringify(state2.tasks, null, 2));

[ 'step2' ]
[
  {
    "id": "c91a38f7-2aec-5c38-a3f0-60fba6efe73c",
    "name": "step2",
    "interrupts": [
      {
        "value": "Received input that is longer than 5 characters: hello world",
        "when": "during"
      }
    ]
  }
]

如果尝试从断点处恢复图，由于我们的输入和图状态没有改变，将再次中断。

// NOTE: to resume the graph from a dynamic interrupt we use the same syntax as
// regular interrupts -- we pass null as the input
const resumedStream = await graph.stream(null, config2);

for await (const event of resumedStream) {
  console.log(event);
}

const state3 = await graph.getState(config2);
console.log(state3.next);
console.log(JSON.stringify(state2.tasks, null, 2));

[ 'step2' ]
[
  {
    "id": "c91a38f7-2aec-5c38-a3f0-60fba6efe73c",
    "name": "step2",
    "interrupts": [
      {
        "value": "Received input that is longer than 5 characters: hello world",
        "when": "during"
      }
    ]
  }
]

更新图状态

为了解决这个问题，我们可以做几件事。

首先，我们可以像一开始那样，在不同的线程上用更短的输入运行图。或者，如果想从断点处恢复图的执行，可以更新状态，使其输入长度小于 5 个字符（这是我们中断的条件）。

// NOTE: this update will be applied as of the last successful node before the interrupt,
// i.e. `step1`, right before the node with an interrupt
await graph.updateState(config2, { input: "short" });

const updatedStream = await graph.stream(null, config2);

for await (const event of updatedStream) {
  console.log(event);
}

const state4 = await graph.getState(config2);
console.log(state4.next);
console.log(state4.values);

---Step 2---
{ input: 'short' }
---Step 3---
{ input: 'short' }
[]
{ input: 'short' }

您还可以**以节点 step2 的身份**（中断的节点）更新状态，这将完全跳过该节点。

const config3 = {
  configurable: {
    thread_id: "3",
  },
  streamMode: "values" as const,
};

const skipStream = await graph.stream({ input: "hello world" }, config3);

// Run the graph until the first interruption
for await (const event of skipStream) {
  console.log(event);
}

{ input: 'hello world' }
---Step 1---
{ input: 'hello world' }

// NOTE: this update will skip the node `step2` entirely
await graph.updateState(config3, undefined, "step2");

// Resume the stream
for await (const event of await graph.stream(null, config3)) {
  console.log(event);
}

const state5 = await graph.getState(config3);
console.log(state5.next);
console.log(state5.values);

---Step 3---
{ input: 'hello world' }
[]
{ input: 'hello world' }