LangGraph篇-人机交互

SuifengFlying

公告

欢迎来到我的博客，这里记录生活与技术。

了解更多

1774 字

9 分钟

LangGraph篇-人机交互

2026-07-29

LangGraph

Human-in-the-loop(人机交互)#

人机交互（或称“在循环中”）工作流将人类输入整合到自动化过程中，在关键阶段允许决策、验证或修正。这在基于 LLM 的应用中尤其有用，因为基础模型可能会产生偶尔的不准确性。在合规、决策或内容生成等低误差容忍场景中，人类的参与通过允许审查、修正或覆盖模型输出来确保可靠性。

使用案例#

基于 LLM 应用中的人机交互工作流的关键使用案例包括：

🛠️ 审查工具调用：人类可以在工具执行之前审查、编辑或批准 LLM 请求的工具调用。
✅ 验证 LLM 输出：人类可以审查、编辑或批准 LLM 生成的内容。
💡 提供上下文：使 LLM 能够明确请求人类输入以进行澄清或提供额外细节，或支持多轮对话。

`interrupt`#

interrupt函数在 LangGraph 中通过在特定节点暂停图形、向人类展示信息以及使用他们的输入恢复图形，从而启用人工干预工作流。该函数对于批准、编辑或收集额外输入等任务非常有用。interrupt函数与 Command 对象结合使用，以人类提供的值恢复图形。

1
from langgraph.types import interrupt
2

3
def human_node(state: State):
4
    value = interrupt(
5
        # 任何可序列化为 JSON 的值，供人类查看。
6
        # 例如，一个问题、一段文本或状态中的一组键
7
       {
8
          "text_to_revise": state["some_text"]
9
       }
10
    )
11
    # 使用人类的输入更新状态或根据输入调整图形。
12
    return {
13
        "some_text": value
14
    }
15

16
graph = graph_builder.compile(
17
    checkpointer=checkpointer # `interrupt` 工作所需
18
)
19

20
# 运行图形直到遇到中断
21
thread_config = {"configurable": {"thread_id": "some_id"}}
22
graph.invoke(some_input, config=thread_config)
23

24
# 用人类的输入恢复图形
25
graph.invoke(Command(resume=value_from_human), config=thread_config)

1
{'some_text': '编辑后的文本'}

Warning

中断非常强大且易于使用。然而，尽管它们在开发者体验上可能类似于 Python 的 input() 函数，但重要的是要注意，它们不会自动从中断点恢复执行。相反，它们会重新运行使用中断的整个节点。因此，中断通常最好放置在节点的开头或一个专用的节点中。请阅读从中断恢复部分以获取更多细节。

完整代码

要求#

要在图形中使用 interrupt，您需要：

指定一个检查点，以在每一步后保存图形状态。
在适当位置调用**** interrupt()。有关示例，请参见设计模式部分。
使用**** 线程 ID ****运行图形，直到触发 interrupt。
使用**** invoke/ainvoke/stream/astream ****恢复执行（见 Command原语）。

设计模式#

您通常可以采取三种不同的行动来实现人工干预工作流：

批准或拒绝：在关键步骤之前暂停图形，例如 API 调用，以审查和批准该操作。如果拒绝该操作，您可以阻止图形执行该步骤，并可能采取替代行动。此模式通常涉及根据人类的输入路由图形。
编辑图形状态：暂停图形以审查和编辑图形状态。这对于纠正错误或使用额外信息更新状态非常有用。此模式通常涉及使用人类的输入更新状态。
获取输入：在图形的特定步骤明确请求人类输入。这对于收集额外信息或上下文以帮助代理的决策过程或支持多轮对话非常有用。

下面我们展示可以使用这些行动实现的不同设计模式。

批准或拒绝#

根据人类的批准或拒绝，图形可以继续执行操作或采取替代路径。

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

1
from typing import Literal
2
from langgraph.types import interrupt, Command
3

4
def human_approval(state: State) -> Command[Literal["some_node", "another_node"]]:
5
    is_approved = interrupt(
6
        {
7
            "question": "这是正确的吗？",
8
            # 展示应由人类审查和批准的输出。
9
            "llm_output": state["llm_output"]
10
        }
11
    )
12

13
    if is_approved:
14
        return Command(goto="some_node")
15
    else:
16
        return Command(goto="another_node")
17

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

22
# 在运行图形并触发中断后，图形将暂停。
23
# 用批准或拒绝恢复。
24
thread_config = {"configurable": {"thread_id": "some_id"}}
25
graph.invoke(Command(resume=True), config=thread_config)

请参见如何审查工具调用获取更详细的示例

添加断点#

在某些节点执行之前或之后设置断点通常很有用。这可以用来在继续之前等待人工批准。当您“编译”图形时，可以设置这些断点。您可以在节点执行之前（使用interrupt_before）或节点执行之后（使用interrupt_after）设置断点。

使用断点时，您必须使用检查点。这是因为您的图形需要能够恢复执行。

为了恢复执行，您可以使用 None 作为输入调用您的图。

1
# Initial run of graph
2
graph.invoke(inputs, config=config)
3

4
# Let's assume it hit a breakpoint somewhere, you can then resume by passing in None
5
graph.invoke(None, config=config)

有关如何添加断点的完整演练，请参阅本指南。

人机交互 (HIL) 在代理系统中至关重要。断点是一种常见的 HIL 交互模式，允许图在特定步骤停止并寻求人为批准后再继续执行（例如，对于敏感操作）。

断点建立在 LangGraph 检查点之上，检查点在每个节点执行后保存图的状态。检查点保存在线程中，这些线程保存图状态，并且可以在图执行完成后访问。这使得图执行可以在特定点暂停，等待人为批准，然后从最后一个检查点恢复执行。

让我们看看它的基本用法。

下面，我们做两件事

我们使用 interrupt_before 指定步骤，来指定断点。
我们设置一个检查点来保存图的状态。

1
from typing import TypedDict
2
from langgraph.graph import StateGraph, START, END
3
from langgraph.checkpoint.memory import MemorySaver
4
from IPython.display import Image, display
5

6

7
class State(TypedDict):
8
    input: str
9

10

11
def step_1(state):
12
    print("---Step 1---")
13
    pass
14

15

16
def step_2(state):
17
    print("---Step 2---")
18
    pass
19

20

21
def step_3(state):
22
    print("---Step 3---")
23
    pass
24

25

26
builder = StateGraph(State)
27
builder.add_node("step_1", step_1)
28
builder.add_node("step_2", step_2)
29
builder.add_node("step_3", step_3)
30
builder.add_edge(START, "step_1")
31
builder.add_edge("step_1", "step_2")
32
builder.add_edge("step_2", "step_3")
33
builder.add_edge("step_3", END)
34

35
# Set up memory
36
memory = MemorySaver()
37

38
# Add
39
graph = builder.compile(checkpointer=memory, interrupt_before=["step_3"])
40

41
# 将生成的图片保存到文件
42
graph_png = graph.get_graph().draw_mermaid_png()
43
with open("breakpoints_case.png", "wb") as f:
44
    f.write(graph_png)

我们为检查点创建一个线程 ID。

我们运行到步骤 3，如 interrupt_before 中定义。

在用户输入/批准后，我们恢复执行，方法是用 None 调用图。

1
from typing import TypedDict
2
from langgraph.checkpoint.memory import MemorySaver
3
from langgraph.graph import StateGraph, START, END
4

5
class State(TypedDict):
6
    input: str
7

8
def step_1(state):
9
    print("---Step 1---")
10
    pass
11

12
def step_2(state):
13
    print("---Step 2---")
14
    pass
15

16
def step_3(state):
17
    print("---Step 3---")
18
    pass
19

20
builder = StateGraph(State)
21
builder.add_node("step_1", step_1)
22
builder.add_node("step_2", step_2)
23
builder.add_node("step_3", step_3)
24
builder.add_edge(START, "step_1")
25
builder.add_edge("step_1", "step_2")
26
builder.add_edge("step_2", "step_3")
27
builder.add_edge("step_3", END)
28

29
# Set up memory
30
memory = MemorySaver()
31

32
# Add
33
graph = builder.compile(checkpointer=memory, interrupt_before=["step_3"])
34

35
# Input
36
initial_input = {"input": "hello world"}
37

38
# Thread
39
thread = {"configurable": {"thread_id": "1"}}
40

41
# 运行graph，直到第一次中断
42
for event in graph.stream(initial_input, thread, stream_mode="values"):
43
    print(event)
44

45
user_approval = input("Do you want to go to Step 3? (yes/no): ")
46

47
if user_approval.lower() == "yes":
48
    # If approved, continue the graph execution
49
    for event in graph.stream(None, thread, stream_mode="values"):
50
        print(event)
51
else:
52
    print("Operation cancelled by user.")

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

LangGraph篇-人机交互

https://mizuki.mysqil.com/posts/langgraph篇-人机交互/

作者

SuifengFlying

发布于

2026-07-29

许可协议

CC BY-NC-SA 4.0

欢迎来到 SuifengFlying!

Human-in-the-loop(人机交互)#

使用案例#

`interrupt`#

要求#

设计模式#

批准或拒绝#

添加断点#

Related articles

欢迎来到 SuifengFlying!

Human-in-the-loop(人机交互)#

使用案例#

interrupt#

要求#

设计模式#

批准或拒绝#

添加断点#

Related articles

`interrupt`#