启用人工干预¶

要在智能代理或工作流中审查、编辑和批准工具调用，可以使用中断来暂停图并等待人工输入。中断使用 LangGraph 的持久化层（保存图状态），可以无限期暂停图执行直到你恢复。

Info

有关人机协同工作流的更多信息，请参阅人机协同概念指南。

使用 `interrupt` 暂停¶

Python

动态中断（也称为动态断点）根据图的当前状态触发。你可以通过在适当的位置调用 @interrupt 函数来设置动态中断。图将暂停，允许人工干预，然后使用他们的输入恢复图。这对于批准、编辑或收集额外上下文等任务很有用。

Note

从 v1.0 开始，interrupt 是暂停图的推荐方式。NodeInterrupt 已弃用，将在 v2.0 中移除。

JavaScript

动态中断（也称为动态断点）根据图的当前状态触发。你可以通过在适当的位置调用 @interrupt 函数来设置动态中断。图将暂停，允许人工干预，然后使用他们的输入恢复图。这对于批准、编辑或收集额外上下文等任务很有用。

要在图中使用 interrupt，你需要：

指定检查点器 以在每一步后保存图状态。
在适当的位置调用 interrupt()。有关示例，请参阅常见模式部分。
使用 线程 ID 运行图 直到命中 interrupt。
使用 invoke/stream 恢复执行（参见 Command 原语）。

Python

# highlight-next-line
from langgraph.types import interrupt, Command

def human_node(state: State):
    # highlight-next-line
    value = interrupt( # (1)!
        {
            "text_to_revise": state["some_text"] # (2)!
        }
    )
    return {
        "some_text": value # (3)!
    }


graph = graph_builder.compile(checkpointer=checkpointer) # (4)!

# 运行图直到命中中断。
config = {"configurable": {"thread_id": "some_id"}}
result = graph.invoke({"some_text": "original text"}, config=config) # (5)!
print(result['__interrupt__']) # (6)!
# > [
# >    Interrupt(
# >       value={'text_to_revise': 'original text'},
# >       resumable=True,
# >       ns=['human_node:6ce9e64f-edef-fe5d-f7dc-511fa9526960']
# >    )
# > ]

# highlight-next-line
print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)!
# > {'some_text': 'Edited text'}

interrupt(...) 在 human_node 暂停执行，将给定的载荷呈现给人类。
任何 JSON 可序列化的值都可以传递给 interrupt 函数。这里是一个包含要修改文本的字典。
恢复后，interrupt(...) 的返回值是人类提供的输入，用于更新状态。
需要检查点器来持久化图状态。在生产中，这应该是持久的（例如由数据库支持）。
图使用一些初始状态调用。
当图命中中断时，它返回一个带有载荷和元数据的 Interrupt 对象。
图使用 Command(resume=...) 恢复，注入人类的输入并继续执行。

JavaScript

// highlight-next-line
import { interrupt, Command } from "@langchain/langgraph";

const graph = graphBuilder
  .addNode("humanNode", (state) => {
    // highlight-next-line
    const value = interrupt(
      // (1)!
      {
        textToRevise: state.someText, // (2)!
      }
    );
    return {
      someText: value, // (3)!
    };
  })
  .addEdge(START, "humanNode")
  .compile({ checkpointer }); // (4)!

// 运行图直到命中中断。
const config = { configurable: { thread_id: "some_id" } };
const result = await graph.invoke({ someText: "original text" }, config); // (5)!
console.log(result.__interrupt__); // (6)!
// > [
// >   {
// >     value: { textToRevise: 'original text' },
// >     resumable: true,
// >     ns: ['humanNode:6ce9e64f-edef-fe5d-f7dc-511fa9526960'],
// >     when: 'during'
// >   }
// > ]

// highlight-next-line
console.log(await graph.invoke(new Command({ resume: "Edited text" }), config)); // (7)!
// > { someText: 'Edited text' }

interrupt(...) 在 humanNode 暂停执行，将给定的载荷呈现给人类。
任何 JSON 可序列化的值都可以传递给 interrupt 函数。这里是一个包含要修改文本的对象。
恢复后，interrupt(...) 的返回值是人类提供的输入，用于更新状态。
需要检查点器来持久化图状态。在生产中，这应该是持久的（例如由数据库支持）。
图使用一些初始状态调用。
当图命中中断时，它返回一个包含 __interrupt__ 的对象，其中有载荷和元数据。
图使用 Command({ resume: ... }) 恢复，注入人类的输入并继续执行。

扩展示例：使用 interrupt

Python

from typing import TypedDict
import uuid
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph

# highlight-next-line
from langgraph.types import interrupt, Command


class State(TypedDict):
    some_text: str


def human_node(state: State):
    # highlight-next-line
    value = interrupt(  # (1)!
        {
            "text_to_revise": state["some_text"]  # (2)!
        }
    )
    return {
        "some_text": value  # (3)!
    }


# 构建图
graph_builder = StateGraph(State)
graph_builder.add_node("human_node", human_node)
graph_builder.add_edge(START, "human_node")
checkpointer = InMemorySaver()  # (4)!
graph = graph_builder.compile(checkpointer=checkpointer)
# 传递线程 ID 给图来运行它。
config = {"configurable": {"thread_id": uuid.uuid4()}}
# 运行图直到命中中断。
result = graph.invoke({"some_text": "original text"}, config=config)  # (5)!

print(result['__interrupt__']) # (6)!
# > [
# >    Interrupt(
# >       value={'text_to_revise': 'original text'},
# >       resumable=True,
# >       ns=['human_node:6ce9e64f-edef-fe5d-f7dc-511fa9526960']
# >    )
# > ]
print(result["__interrupt__"])  # (6)!
# > [Interrupt(value={'text_to_revise': 'original text'}, id='6d7c4048049254c83195429a3659661d')]

# highlight-next-line
print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)!
# > {'some_text': 'Edited text'}

interrupt(...) 在 human_node 暂停执行，将给定的载荷呈现给人类。
任何 JSON 可序列化的值都可以传递给 interrupt 函数。这里是一个包含要修改文本的字典。
恢复后，interrupt(...) 的返回值是人类提供的输入，用于更新状态。
需要检查点器来持久化图状态。在生产中，这应该是持久的（例如由数据库支持）。
图使用一些初始状态调用。
当图命中中断时，它返回一个带有载荷和元数据的 Interrupt 对象。
图使用 Command(resume=...) 恢复，注入人类的输入并继续执行。

JavaScript

import { z } from "zod";
import { v4 as uuidv4 } from "uuid";
import { MemorySaver, StateGraph, START, interrupt, Command } from "@langchain/langgraph";

const StateAnnotation = z.object({
  someText: z.string(),
});

// 构建图
const graphBuilder = new StateGraph(StateAnnotation)
  .addNode("humanNode", (state) => {
    // highlight-next-line
    const value = interrupt( // (1)!
      {
        textToRevise: state.someText // (2)!
      }
    );
    return {
      someText: value // (3)!
    };
  })
  .addEdge(START, "humanNode");

const checkpointer = new MemorySaver(); // (4)!

const graph = graphBuilder.compile({ checkpointer });

// 传递线程 ID 给图来运行它。
const config = { configurable: { thread_id: uuidv4() } };

// 运行图直到命中中断。
const result = await graph.invoke({ someText: "original text" }, config); // (5)!

console.log(result.__interrupt__); // (6)!
// > [
// >   {
// >     value: { textToRevise: 'original text' },
// >     resumable: true,
// >     ns: ['humanNode:6ce9e64f-edef-fe5d-f7dc-511fa9526960'],
// >     when: 'during'
// >   }
// > ]

// highlight-next-line
console.log(await graph.invoke(new Command({ resume: "Edited text" }), config)); // (7)!
// > { someText: 'Edited text' }

interrupt(...) 在 humanNode 暂停执行，将给定的载荷呈现给人类。
任何 JSON 可序列化的值都可以传递给 interrupt 函数。这里是一个包含要修改文本的对象。
恢复后，interrupt(...) 的返回值是人类提供的输入，用于更新状态。
需要检查点器来持久化图状态。在生产中，这应该是持久的（例如由数据库支持）。
图使用一些初始状态调用。
当图命中中断时，它返回一个包含 __interrupt__ 的对象，其中有载荷和元数据。
图使用 Command({ resume: ... }) 恢复，注入人类的输入并继续执行。

0.4.0 新增

Python

__interrupt__ 是一个特殊的键，当图被中断时运行图会返回它。在 0.4.0 版本中添加了对 invoke 和 ainvoke 中 __interrupt__ 的支持。如果你使用的是旧版本，只有在使用 stream 或 astream 时才能在结果中看到 __interrupt__。你也可以使用 graph.get_state(thread_id) 来获取中断值。

JavaScript

__interrupt__ 是一个特殊的键，当图被中断时运行图会返回它。在 0.4.0 版本中添加了对 invoke 中 __interrupt__ 的支持。如果你使用的是旧版本，只有在使用 stream 时才能在结果中看到 __interrupt__。你也可以使用 graph.getState(config) 来获取中断值。

Warning

Python

中断在开发者体验方面类似于 Python 的 input() 函数，但它们不会自动从中断点恢复执行。相反，它们会重新运行使用中断的整个节点。因此，中断通常最好放在节点的开头或专门的节点中。

JavaScript

中断功能强大且符合人体工程学，但重要的是要注意它们不会自动从中断点恢复执行。相反，它们会重新运行使用中断的整个节点。因此，中断通常最好放在节点的开头或专门的节点中。

使用 `Command` 原语恢复¶

Python

Warning

从 interrupt 恢复与 Python 的 input() 函数不同，后者从调用 input() 函数的确切位置恢复执行。

当在图中使用 interrupt 函数时，执行在该点暂停并等待用户输入。

Python

要恢复执行，使用 @[Command][] 原语，可以通过 invoke 或 stream 方法提供。图从最初调用 interrupt(...) 的节点开头恢复执行。这次，interrupt 函数将返回 Command(resume=value) 中提供的值，而不是再次暂停。从节点开头到 interrupt 的所有代码都将重新执行。

# 通过提供用户输入来恢复图执行。
graph.invoke(Command(resume={"age": "25"}), thread_config)

JavaScript

要恢复执行，使用 @[Command][] 原语，可以通过 invoke 或 stream 方法提供。图从最初调用 interrupt(...) 的节点开头恢复执行。这次，interrupt 函数将返回 Command(resume=value) 中提供的值，而不是再次暂停。从节点开头到 interrupt 的所有代码都将重新执行。

// 通过提供用户输入来恢复图执行。
await graph.invoke(new Command({ resume: { age: "25" } }), threadConfig);

用一次调用恢复多个中断¶

当带有中断条件的节点并行运行时，任务队列中可能有多个中断。例如，以下图有两个并行运行的节点需要人工输入：

Python

一旦图被中断并停滞，你可以使用 Command.resume 一次恢复所有中断，传递中断 ID 到恢复值的字典映射。

from typing import TypedDict
import uuid
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command


class State(TypedDict):
    text_1: str
    text_2: str


def human_node_1(state: State):
    value = interrupt({"text_to_revise": state["text_1"]})
    return {"text_1": value}


def human_node_2(state: State):
    value = interrupt({"text_to_revise": state["text_2"]})
    return {"text_2": value}


graph_builder = StateGraph(State)
graph_builder.add_node("human_node_1", human_node_1)
graph_builder.add_node("human_node_2", human_node_2)

# 从 START 并行添加两个节点
graph_builder.add_edge(START, "human_node_1")
graph_builder.add_edge(START, "human_node_2")

checkpointer = InMemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

thread_id = str(uuid.uuid4())
config: RunnableConfig = {"configurable": {"thread_id": thread_id}}
result = graph.invoke(
    {"text_1": "original text 1", "text_2": "original text 2"}, config=config
)

# 使用中断 ID 到值的映射恢复
resume_map = {
    i.id: f"edited text for {i.value['text_to_revise']}"
    for i in graph.get_state(config).interrupts
}
print(graph.invoke(Command(resume=resume_map), config=config))
# > {'text_1': 'edited text for original text 1', 'text_2': 'edited text for original text 2'}

JavaScript

const state = await parentGraph.getState(threadConfig);
const resumeMap = Object.fromEntries(
  state.interrupts.map((i) => [
    i.interruptId,
    `human input for prompt ${i.value}`,
  ])
);

await parentGraph.invoke(new Command({ resume: resumeMap }), threadConfig);

常见模式¶

下面我们展示可以使用 interrupt 和 Command 实现的不同设计模式。

批准或拒绝¶

在关键步骤（如 API 调用）之前暂停图，以审查和批准操作。如果操作被拒绝，你可以阻止图执行该步骤，并可能采取替代操作。

Python

from typing import Literal
from langgraph.types import interrupt, Command

def human_approval(state: State) -> Command[Literal["some_node", "another_node"]]:
    is_approved = interrupt(
        {
            "question": "Is this correct?",
            # 呈现应该由人类审查和批准的输出。
            "llm_output": state["llm_output"]
        }
    )

    if is_approved:
        return Command(goto="some_node")
    else:
        return Command(goto="another_node")

# 在适当的位置将节点添加到图中
# 并将其连接到相关节点。
graph_builder.add_node("human_approval", human_approval)
graph = graph_builder.compile(checkpointer=checkpointer)

# 运行图并命中中断后，图将暂停。
# 使用批准或拒绝来恢复它。
thread_config = {"configurable": {"thread_id": "some_id"}}
graph.invoke(Command(resume=True), config=thread_config)

JavaScript

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

// 在适当的位置将节点添加到图中
// 并将其连接到相关节点。
graphBuilder.addNode("humanApproval", (state) => {
  const isApproved = interrupt({
    question: "Is this correct?",
    // 呈现应该由人类审查和批准的输出。
    llmOutput: state.llmOutput,
  });

  if (isApproved) {
    return new Command({ goto: "someNode" });
  } else {
    return new Command({ goto: "anotherNode" });
  }
});
const graph = graphBuilder.compile({ checkpointer });

// 运行图并命中中断后，图将暂停。
// 使用批准或拒绝来恢复它。
const threadConfig = { configurable: { thread_id: "some_id" } };
await graph.invoke(new Command({ resume: true }), threadConfig);

审查和编辑状态¶

Python

from langgraph.types import interrupt

def human_editing(state: State):
    ...
    result = interrupt(
        # 要呈现给客户端的中断信息。
        # 可以是任何 JSON 可序列化的值。
        {
            "task": "Review the output from the LLM and make any necessary edits.",
            "llm_generated_summary": state["llm_generated_summary"]
        }
    )

    # 使用编辑后的文本更新状态
    return {
        "llm_generated_summary": result["edited_text"]
    }

# 在适当的位置将节点添加到图中
# 并将其连接到相关节点。
graph_builder.add_node("human_editing", human_editing)
graph = graph_builder.compile(checkpointer=checkpointer)

...

# 运行图并命中中断后，图将暂停。
# 使用编辑后的文本恢复它。
thread_config = {"configurable": {"thread_id": "some_id"}}
graph.invoke(
    Command(resume={"edited_text": "The edited text"}),
    config=thread_config
)

JavaScript

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

function humanEditing(state: z.infer<typeof StateAnnotation>) {
  const result = interrupt({
    // 要呈现给客户端的中断信息。
    // 可以是任何 JSON 可序列化的值。
    task: "Review the output from the LLM and make any necessary edits.",
    llmGeneratedSummary: state.llmGeneratedSummary,
  });

  // 使用编辑后的文本更新状态
  return {
    llmGeneratedSummary: result.editedText,
  };
}

// 在适当的位置将节点添加到图中
// 并将其连接到相关节点。
graphBuilder.addNode("humanEditing", humanEditing);
const graph = graphBuilder.compile({ checkpointer });

// 运行图并命中中断后，图将暂停。
// 使用编辑后的文本恢复它。
const threadConfig = { configurable: { thread_id: "some_id" } };
await graph.invoke(
  new Command({ resume: { editedText: "The edited text" } }),
  threadConfig
);

审查工具调用¶

人类可以在继续之前审查和编辑 LLM 的输出。这在 LLM 请求的工具调用可能敏感或需要人工监督的应用中特别重要。

要向工具添加人工批准步骤：

在工具中使用 interrupt() 暂停执行。
使用 Command 恢复以根据人工输入继续。

Python

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
from langgraph.prebuilt import create_react_agent

# 需要人工审查/批准的敏感工具示例
def book_hotel(hotel_name: str):
    """预订酒店"""
    # highlight-next-line
    response = interrupt(  # (1)!
        f"Trying to call `book_hotel` with args {{'hotel_name': {hotel_name}}}. "
        "Please approve or suggest edits."
    )
    if response["type"] == "accept":
        pass
    elif response["type"] == "edit":
        hotel_name = response["args"]["hotel_name"]
    else:
        raise ValueError(f"Unknown response type: {response['type']}")
    return f"Successfully booked a stay at {hotel_name}."

# highlight-next-line
checkpointer = InMemorySaver() # (2)!

agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_hotel],
    # highlight-next-line
    checkpointer=checkpointer, # (3)!
)

@interrupt 函数在特定节点暂停智能代理图。在这种情况下，我们在工具函数开头调用 interrupt()，这会在执行工具的节点暂停图。interrupt() 中的信息（例如工具调用）可以呈现给人类，图可以使用用户输入（工具调用批准、编辑或反馈）恢复。
InMemorySaver 用于在工具调用循环的每一步存储智能代理状态。这启用了短期记忆和人机协同功能。在此示例中，我们使用 InMemorySaver 在内存中存储智能代理状态。在生产应用中，智能代理状态将存储在数据库中。
使用 checkpointer 初始化智能代理。

JavaScript

import { MemorySaver } from "@langchain/langgraph";
import { interrupt } from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

// 需要人工审查/批准的敏感工具示例
const bookHotel = tool(
  async ({ hotelName }) => {
    // highlight-next-line
    const response = interrupt(
      // (1)!
      `Trying to call \`bookHotel\` with args {"hotelName": "${hotelName}"}. ` +
        "Please approve or suggest edits."
    );
    if (response.type === "accept") {
      // 继续使用原始参数
    } else if (response.type === "edit") {
      hotelName = response.args.hotelName;
    } else {
      throw new Error(`Unknown response type: ${response.type}`);
    }
    return `Successfully booked a stay at ${hotelName}.`;
  },
  {
    name: "bookHotel",
    description: "Book a hotel",
    schema: z.object({
      hotelName: z.string(),
    }),
  }
);

// highlight-next-line
const checkpointer = new MemorySaver(); // (2)!

const agent = createReactAgent({
  llm: model,
  tools: [bookHotel],
  // highlight-next-line
  checkpointSaver: checkpointer, // (3)!
});

@interrupt 函数在特定节点暂停智能代理图。在这种情况下，我们在工具函数开头调用 interrupt()，这会在执行工具的节点暂停图。interrupt() 中的信息（例如工具调用）可以呈现给人类，图可以使用用户输入（工具调用批准、编辑或反馈）恢复。
MemorySaver 用于在工具调用循环的每一步存储智能代理状态。这启用了短期记忆和人机协同功能。在此示例中，我们使用 MemorySaver 在内存中存储智能代理状态。在生产应用中，智能代理状态将存储在数据库中。
使用 checkpointSaver 初始化智能代理。

使用 stream() 方法运行智能代理，传递 config 对象以指定线程 ID。这允许智能代理在将来的调用中恢复同一对话。

Python

config = {
   "configurable": {
      # highlight-next-line
      "thread_id": "1"
   }
}

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
    # highlight-next-line
    config
):
    print(chunk)
    print("\n")

JavaScript

const config = {
  configurable: {
    // highlight-next-line
    thread_id: "1",
  },
};

const stream = await agent.stream(
  { messages: [{ role: "user", content: "book a stay at McKittrick hotel" }] },
  // highlight-next-line
  config
);

for await (const chunk of stream) {
  console.log(chunk);
  console.log("\n");
}

你应该看到智能代理运行直到到达 interrupt() 调用，此时它暂停并等待人工输入。

使用 Command 恢复智能代理以根据人工输入继续。

Python

from langgraph.types import Command

for chunk in agent.stream(
    # highlight-next-line
    Command(resume={"type": "accept"}),  # (1)!
    # Command(resume={"type": "edit", "args": {"hotel_name": "McKittrick Hotel"}}),
    config
):
    print(chunk)
    print("\n")

@interrupt 函数与 @[Command][] 对象一起使用，以使用人类提供的值恢复图。

JavaScript

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

const resumeStream = await agent.stream(
  // highlight-next-line
  new Command({ resume: { type: "accept" } }), // (1)!
  // new Command({ resume: { type: "edit", args: { hotelName: "McKittrick Hotel" } } }),
  config
);

for await (const chunk of resumeStream) {
  console.log(chunk);
  console.log("\n");
}

@interrupt 函数与 @[Command][] 对象一起使用，以使用人类提供的值恢复图。

向任何工具添加中断¶

你可以创建一个包装器来向任何工具添加中断。以下示例提供了与 Agent Inbox UI 和 Agent Chat UI 兼容的参考实现。

Python

向任何工具添加人机协同的包装器

from typing import Callable
from langchain_core.tools import BaseTool, tool as create_tool
from langchain_core.runnables import RunnableConfig
from langgraph.types import interrupt
from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt

def add_human_in_the_loop(
    tool: Callable | BaseTool,
    *,
    interrupt_config: HumanInterruptConfig = None,
) -> BaseTool:
    """包装工具以支持人机协同审查。"""
    if not isinstance(tool, BaseTool):
        tool = create_tool(tool)

    if interrupt_config is None:
        interrupt_config = {
            "allow_accept": True,
            "allow_edit": True,
            "allow_respond": True,
        }

    @create_tool(  # (1)!
        tool.name,
        description=tool.description,
        args_schema=tool.args_schema
    )
    def call_tool_with_interrupt(config: RunnableConfig, **tool_input):
        request: HumanInterrupt = {
            "action_request": {
                "action": tool.name,
                "args": tool_input
            },
            "config": interrupt_config,
            "description": "Please review the tool call"
        }
        # highlight-next-line
        response = interrupt([request])[0]  # (2)!
        # 批准工具调用
        if response["type"] == "accept":
            tool_response = tool.invoke(tool_input, config)
        # 更新工具调用参数
        elif response["type"] == "edit":
            tool_input = response["args"]["args"]
            tool_response = tool.invoke(tool_input, config)
        # 用用户反馈回复 LLM
        elif response["type"] == "response":
            user_feedback = response["args"]
            tool_response = user_feedback
        else:
            raise ValueError(f"Unsupported interrupt response type: {response['type']}")

        return tool_response

    return call_tool_with_interrupt

此包装器创建一个新工具，在执行被包装的工具之前调用 interrupt()。
interrupt() 使用 Agent Inbox UI 期望的特殊输入和输出格式：- @[HumanInterrupt][] 对象列表发送给 AgentInbox 以向最终用户呈现中断信息 - 恢复值由 AgentInbox 作为列表提供（即 Command(resume=[...])）

JavaScript

向任何工具添加人机协同的包装器

import { StructuredTool, tool } from "@langchain/core/tools";
import { RunnableConfig } from "@langchain/core/runnables";
import { interrupt } from "@langchain/langgraph";

interface HumanInterruptConfig {
  allowAccept?: boolean;
  allowEdit?: boolean;
  allowRespond?: boolean;
}

interface HumanInterrupt {
  actionRequest: {
    action: string;
    args: Record<string, any>;
  };
  config: HumanInterruptConfig;
  description: string;
}

function addHumanInTheLoop(
  originalTool: StructuredTool,
  interruptConfig: HumanInterruptConfig = {
    allowAccept: true,
    allowEdit: true,
    allowRespond: true,
  }
): StructuredTool {
  // 包装原始工具以支持人机协同审查
  return tool(
    // (1)!
    async (toolInput: Record<string, any>, config?: RunnableConfig) => {
      const request: HumanInterrupt = {
        actionRequest: {
          action: originalTool.name,
          args: toolInput,
        },
        config: interruptConfig,
        description: "Please review the tool call",
      };

      // highlight-next-line
      const response = interrupt([request])[0]; // (2)!

      // 批准工具调用
      if (response.type === "accept") {
        return await originalTool.invoke(toolInput, config);
      }
      // 更新工具调用参数
      else if (response.type === "edit") {
        const updatedArgs = response.args.args;
        return await originalTool.invoke(updatedArgs, config);
      }
      // 用用户反馈回复 LLM
      else if (response.type === "response") {
        return response.args;
      } else {
        throw new Error(
          `Unsupported interrupt response type: ${response.type}`
        );
      }
    },
    {
      name: originalTool.name,
      description: originalTool.description,
      schema: originalTool.schema,
    }
  );
}

此包装器创建一个新工具，在执行被包装的工具之前调用 interrupt()。
interrupt() 使用 Agent Inbox UI 期望的特殊输入和输出格式：- [HumanInterrupt] 对象列表发送给 AgentInbox 以向最终用户呈现中断信息 - 恢复值由 AgentInbox 作为列表提供（即 Command({ resume: [...] })）

你可以使用包装器向任何工具添加 interrupt()，而无需在工具内部添加它：

Python

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

# highlight-next-line
checkpointer = InMemorySaver()

def book_hotel(hotel_name: str):
   """预订酒店"""
   return f"Successfully booked a stay at {hotel_name}."


agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[
        # highlight-next-line
        add_human_in_the_loop(book_hotel), # (1)!
    ],
    # highlight-next-line
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "1"}}

# 运行智能代理
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
    # highlight-next-line
    config
):
    print(chunk)
    print("\n")

add_human_in_the_loop 包装器用于向工具添加 interrupt()。这允许智能代理暂停执行并在继续工具调用之前等待人工输入。

JavaScript

import { MemorySaver } from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

// highlight-next-line
const checkpointer = new MemorySaver();

const bookHotel = tool(
  async ({ hotelName }) => {
    return `Successfully booked a stay at ${hotelName}.`;
  },
  {
    name: "bookHotel",
    description: "Book a hotel",
    schema: z.object({
      hotelName: z.string(),
    }),
  }
);

const agent = createReactAgent({
  llm: model,
  tools: [
    // highlight-next-line
    addHumanInTheLoop(bookHotel), // (1)!
  ],
  // highlight-next-line
  checkpointSaver: checkpointer,
});

const config = { configurable: { thread_id: "1" } };

// 运行智能代理
const stream = await agent.stream(
  { messages: [{ role: "user", content: "book a stay at McKittrick hotel" }] },
  // highlight-next-line
  config
);

for await (const chunk of stream) {
  console.log(chunk);
  console.log("\n");
}

addHumanInTheLoop 包装器用于向工具添加 interrupt()。这允许智能代理暂停执行并在继续工具调用之前等待人工输入。

你应该看到智能代理运行直到到达 interrupt() 调用，此时它暂停并等待人工输入。

使用 Command 恢复智能代理以根据人工输入继续。

Python

from langgraph.types import Command

for chunk in agent.stream(
    # highlight-next-line
    Command(resume=[{"type": "accept"}]),
    # Command(resume=[{"type": "edit", "args": {"args": {"hotel_name": "McKittrick Hotel"}}}]),
    config
):
    print(chunk)
    print("\n")

JavaScript

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

const resumeStream = await agent.stream(
  // highlight-next-line
  new Command({ resume: [{ type: "accept" }] }),
  // new Command({ resume: [{ type: "edit", args: { args: { hotelName: "McKittrick Hotel" } } }] }),
  config
);

for await (const chunk of resumeStream) {
  console.log(chunk);
  console.log("\n");
}

验证人工输入¶

如果你需要在图本身（而不是客户端）内验证人类提供的输入，可以通过在单个节点中使用多个中断调用来实现。

Python

from langgraph.types import interrupt

def human_node(state: State):
    """带验证的人工节点。"""
    question = "What is your age?"

    while True:
        answer = interrupt(question)

        # 验证答案，如果答案无效则再次请求输入。
        if not isinstance(answer, int) or answer < 0:
            question = f"'{answer} is not a valid age. What is your age?"
            answer = None
            continue
        else:
            # 如果答案有效，我们可以继续。
            break

    print(f"The human in the loop is {answer} years old.")
    return {
        "age": answer
    }

JavaScript

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

graphBuilder.addNode("humanNode", (state) => {
  // 带验证的人工节点。
  let question = "What is your age?";

  while (true) {
    const answer = interrupt(question);

    // 验证答案，如果答案无效则再次请求输入。
    if (typeof answer !== "number" || answer < 0) {
      question = `'${answer}' is not a valid age. What is your age?`;
      continue;
    } else {
      // 如果答案有效，我们可以继续。
      break;
    }
  }

  console.log(`The human in the loop is ${answer} years old.`);
  return {
    age: answer,
  };
});

Python

使用中断调试¶

要调试和测试图，使用静态中断（也称为静态断点）逐节点执行图或在特定节点暂停图执行。静态中断在节点执行之前或之后的定义点触发。你可以通过在编译时或运行时指定 interrupt_before 和 interrupt_after 来设置静态中断。

Warning

静态中断不推荐用于人机协同工作流。请改用动态中断。

编译时运行时

# highlight-next-line
graph = graph_builder.compile( # (1)!
    # highlight-next-line
    interrupt_before=["node_a"], # (2)!
    # highlight-next-line
    interrupt_after=["node_b", "node_c"], # (3)!
    checkpointer=checkpointer, # (4)!
)

config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}

# 运行图直到断点
graph.invoke(inputs, config=thread_config) # (5)!

# 恢复图
graph.invoke(None, config=thread_config) # (6)!

断点在 compile 时设置。
interrupt_before 指定在节点执行之前暂停执行的节点。
interrupt_after 指定在节点执行之后暂停执行的节点。
需要检查点器来启用断点。
图运行直到命中第一个断点。
通过为输入传入 None 来恢复图。这将运行图直到命中下一个断点。

# highlight-next-line
graph.invoke( # (1)!
    inputs,
    # highlight-next-line
    interrupt_before=["node_a"], # (2)!
    # highlight-next-line
    interrupt_after=["node_b", "node_c"] # (3)!
    config={
        "configurable": {"thread_id": "some_thread"}
    },
)

config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}

# 运行图直到断点
graph.invoke(inputs, config=config) # (4)!

# 恢复图
graph.invoke(None, config=config) # (5)!

graph.invoke 使用 interrupt_before 和 interrupt_after 参数调用。这是运行时配置，可以为每次调用更改。
interrupt_before 指定在节点执行之前暂停执行的节点。
interrupt_after 指定在节点执行之后暂停执行的节点。
图运行直到命中第一个断点。
通过为输入传入 None 来恢复图。这将运行图直到命中下一个断点。

Note

你不能在运行时为子图设置静态断点。如果你有子图，必须在编译时设置断点。

在 LangGraph Studio 中使用静态中断¶

你可以使用 LangGraph Studio 调试你的图。你可以在 UI 中设置静态断点，然后运行图。你还可以使用 UI 在执行的任何时刻检查图状态。

LangGraph Studio 对使用 langgraph dev 本地部署的应用免费。

注意事项¶

使用人机协同时，有一些注意事项需要牢记。

与有副作用的代码一起使用¶

将有副作用的代码（如 API 调用）放在 interrupt 之后或单独的节点中以避免重复，因为每次恢复节点时都会重新触发这些代码。

中断后的副作用单独节点中的副作用

Python

from langgraph.types import interrupt

def human_node(state: State):
    """带验证的人工节点。"""

    answer = interrupt(question)

    api_call(answer) # 没问题，因为在中断之后

JavaScript

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

function humanNode(state: z.infer<typeof StateAnnotation>) {
  // 带验证的人工节点。

  const answer = interrupt(question);

  apiCall(answer); // 没问题，因为在中断之后
}

Python

from langgraph.types import interrupt

def human_node(state: State):
    """带验证的人工节点。"""

    answer = interrupt(question)

    return {
        "answer": answer
    }

def api_call_node(state: State):
    api_call(...) # 没问题，因为在单独的节点中

JavaScript

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

function humanNode(state: z.infer<typeof StateAnnotation>) {
  // 带验证的人工节点。

  const answer = interrupt(question);

  return {
    answer
  };
}

function apiCallNode(state: z.infer<typeof StateAnnotation>) {
  apiCall(state.answer); // 没问题，因为在单独的节点中
}

与作为函数调用的子图一起使用¶

当作为函数调用子图时，父图将从触发 interrupt 的子图调用所在节点的开头恢复执行。同样，子图将从调用 interrupt() 函数的节点的开头恢复。

Python

def node_in_parent_graph(state: State):
    some_code()  # <-- 当子图恢复时这将重新执行。
    # 作为函数调用子图。
    # 子图包含 `interrupt` 调用。
    subgraph_result = subgraph.invoke(some_input)
    ...

JavaScript

async function nodeInParentGraph(state: z.infer<typeof StateAnnotation>) {
  someCode(); // <-- 当子图恢复时这将重新执行。
  // 作为函数调用子图。
  // 子图包含 `interrupt` 调用。
  const subgraphResult = await subgraph.invoke(someInput);
  // ...
}

在单个节点中使用多个中断¶

在单个节点中使用多个中断对于像验证人工输入这样的模式很有帮助。但是，如果处理不当，在同一节点中使用多个中断可能导致意外行为。

当节点包含多个中断调用时，LangGraph 会保留一个特定于执行节点的任务的恢复值列表。每当恢复执行时，它都从节点开头开始。对于遇到的每个中断，LangGraph 检查任务的恢复列表中是否存在匹配值。匹配 严格基于索引，因此节点内中断调用的顺序至关重要。

为避免问题，请避免在执行之间动态更改节点的结构。这包括添加、删除或重新排序中断调用，因为这些更改可能导致索引不匹配。这些问题通常来自非常规模式，例如通过 Command(resume=..., update=SOME_STATE_MUTATION) 变更状态或依赖全局变量动态修改节点的结构。

启用人工干预¶

使用 interrupt 暂停¶

使用 Command 原语恢复¶

用一次调用恢复多个中断¶

常见模式¶

批准或拒绝¶

审查和编辑状态¶

审查工具调用¶

向任何工具添加中断¶

验证人工输入¶

使用中断调试¶

在 LangGraph Studio 中使用静态中断¶

注意事项¶

与有副作用的代码一起使用¶

与作为函数调用的子图一起使用¶

在单个节点中使用多个中断¶

使用 `interrupt` 暂停¶

使用 `Command` 原语恢复¶