人工介入¶
请改用 interrupt 函数。
从 LangGraph 0.2.57 版本开始，推荐使用 interrupt 函数 来设置断点，因为它简化了**人工介入**模式。
请参阅修订后的人工介入指南，以了解使用 interrupt 函数的最新版本。
人工介入（或“循环内人工”）通过几种常见的用户交互模式增强了智能体的能力。
常见的交互模式包括：
(1) 审批 - 我们可以中断智能体，向用户展示当前状态，并允许用户批准某个操作。
(2) 编辑 - 我们可以中断智能体，向用户展示当前状态，并允许用户编辑智能体的状态。
(3) 输入 - 我们可以显式地创建一个图节点来收集人类输入，并将该输入直接传递给智能体状态。
这些交互模式的用例包括：
(1) 审查工具调用 - 我们可以中断智能体，以审查和编辑工具调用的结果。
(2) 时间回溯 - 我们可以手动重播和/或分叉智能体过去的操作。
持久化¶
所有这些交互模式都由 LangGraph 的内置持久化层支持，该层会在图状态的每一步写入一个检查点。持久化允许图暂停，以便人类可以审查和/或编辑图的当前状态，然后根据人类的输入恢复执行。
断点¶
在图流程的特定位置添加断点是实现人工介入循环的一种方式。在这种情况下，开发者知道工作流中*何处*需要人工输入，只需在特定图节点之前或之后放置一个断点即可。
在这里，我们使用检查点器和一个断点来编译我们的图，断点位于我们想要中断的节点 step_for_human_in_the_loop 之前。然后，我们执行上述交互模式之一，如果人类编辑了图状态，这将创建一个新的检查点。新的检查点会保存到 thread 中，我们可以通过传入 None 作为输入从那里恢复图的执行。
# 使用检查点器和在 "step_for_human_in_the_loop" 之前的断点来编译我们的图
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["step_for_human_in_the_loop"])
# 运行图直到断点处
thread_config = {"configurable": {"thread_id": "1"}}
for event in graph.stream(inputs, thread_config, stream_mode="values"):
print(event)
# 执行一些需要人工介入的操作
# 从当前检查点继续图的执行
for event in graph.stream(None, thread_config, stream_mode="values"):
print(event)
动态断点¶
或者，开发者可以定义一些必须满足的*条件*，以便触发断点。动态断点这一概念在开发者希望在*特定条件*下暂停图的执行时非常有用。这使用了 NodeInterrupt，它是一种特殊类型的异常，可以根据某些条件在节点内部抛出。例如，我们可以定义一个动态断点，当 input 的长度超过 5 个字符时触发。
def my_node(state: State) -> State:
if len(state['input']) > 5:
raise NodeInterrupt(f"Received input that is longer than 5 characters: {state['input']}")
return state
假设我们使用一个会触发动态断点的输入来运行图，然后尝试通过简单地传入 None 作为输入来恢复图的执行。
# 在触发动态断点后，尝试在不改变状态的情况下继续图的执行
for event in graph.stream(None, thread_config, stream_mode="values"):
print(event)
图将再次*中断*，因为这个节点将使用相同的图状态*重新运行*。我们需要更改图状态，使得触发动态断点的条件不再满足。因此，我们可以简单地将图状态编辑为满足动态断点条件（长度 < 5 个字符）的输入，然后重新运行该节点。
# 更新状态以通过动态断点
graph.update_state(config=thread_config, values={"input": "foo"})
for event in graph.stream(None, thread_config, stream_mode="values"):
print(event)
或者，如果我们想保留当前输入并跳过执行检查的节点 (my_node) 该怎么办？为此，我们可以简单地使用 as_node="my_node" 执行图更新，并传入 None 作为值。这不会更新图状态，但会以 my_node 的身份运行更新，从而有效地跳过该节点并绕过动态断点。
# 此更新将完全跳过节点 `my_node`
graph.update_state(config=thread_config, values=None, as_node="my_node")
for event in graph.stream(None, thread_config, stream_mode="values"):
print(event)
有关详细的操作指南，请参阅我们的指南！
交互模式¶
审批¶
有时我们希望对智能体执行过程中的某些步骤进行审批。
我们可以在想要审批的步骤之前的断点处中断智能体。
通常建议对敏感操作（例如，使用外部 API 或写入数据库）采用这种方式。
借助持久化功能，我们可以将当前智能体状态以及下一步操作展示给用户进行审核和审批。
如果获得批准，图将从最后保存的检查点继续执行，该检查点会保存到 thread 中：
# 使用检查点和在要审批的步骤之前的断点来编译我们的图
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["node_2"])
# 运行图直到断点处
for event in graph.stream(inputs, thread, stream_mode="values"):
print(event)
# ... 获取人工审批 ...
# 如果获得批准，从最后保存的检查点继续执行图
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
有关详细操作方法，请参阅我们的指南！
编辑¶
有时我们希望审核并编辑智能体的状态。
与审批一样，我们可以在想要检查的步骤之前的断点处中断智能体。
我们可以将当前状态展示给用户，并允许用户编辑智能体状态。
例如，如果智能体犯了错误，这可以用于纠正它（例如，见下面关于工具调用的部分）。
我们可以通过分叉当前检查点来编辑图状态，该检查点会保存到 thread 中。
然后我们可以像之前一样从分叉的检查点继续执行图。
# 使用检查点和在要审核的步骤之前的断点来编译我们的图
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["node_2"])
# 运行图直到断点处
for event in graph.stream(inputs, thread, stream_mode="values"):
print(event)
# 审核状态，决定进行编辑，并使用新状态创建一个分叉的检查点
graph.update_state(thread, {"state": "new state"})
# 从分叉的检查点继续执行图
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
有关详细操作方法，请参阅本指南！
输入¶
有时我们希望在图中的特定步骤显式获取人工输入。
我们可以为此创建一个指定的图节点（例如，在我们的示例图中为 human_input）。
与审批和编辑一样，我们可以在该节点之前的断点处中断智能体。
然后我们可以执行包含人工输入的状态更新，就像我们编辑状态时所做的那样。
但是，我们要添加一件事：
我们可以在状态更新时使用 as_node=human_input 来指定该状态更新*应被视为一个节点*。
这很微妙，但很重要：
在编辑时，用户决定是否编辑图状态。
在输入时，我们在图中显式定义一个节点来收集人工输入！
包含人工输入的状态更新随后将*作为该节点*运行。
# 使用检查点和在要收集人工输入的步骤之前的断点来编译我们的图
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["human_input"])
# 运行图直到断点处
for event in graph.stream(inputs, thread, stream_mode="values"):
print(event)
# 使用用户输入更新状态，就好像它是 human_input 节点一样
graph.update_state(thread, {"user_input": user_input}, as_node="human_input")
# 从 human_input 节点创建的检查点继续执行图
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
有关详细操作方法，请参阅本指南！
用例¶
审查工具调用¶
一些用户交互模式结合了上述理念。
例如，许多智能体使用工具调用来做出决策。
工具调用带来了一个挑战，因为智能体必须确保两件事正确： (1) 要调用的工具名称 (2) 传递给工具的参数
即使工具调用是正确的，我们可能还需要进行斟酌： (3) 工具调用可能是一个敏感操作，我们需要进行审批
考虑到这些要点，我们可以结合上述理念，对工具调用进行人工审查。
# 使用检查点器和断点编译我们的图，以便在审查大语言模型的工具调用步骤之前中断
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["human_review"])
# 运行图直到断点
for event in graph.stream(inputs, thread, stream_mode="values"):
print(event)
# 审查工具调用，并在需要时更新它，作为 human_review 节点
graph.update_state(thread, {"tool_call": "updated tool call"}, as_node="human_review")
# 否则，批准工具调用并继续执行图，不做任何编辑
# 从以下两种情况之一继续执行图：
# (1) human_review 创建的分支检查点
# (2) 最初进行工具调用时保存的检查点（human_review 中未做编辑）
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
有关详细操作指南，请参阅本指南！
时光回溯¶
在使用智能体时，我们通常希望仔细检查它们的决策过程： (1) 即使它们得出了期望的最终结果，导致该结果的推理过程通常也值得仔细审查。 (2) 当智能体犯错时，了解原因通常很有价值。 (3) 在上述任何一种情况下，手动探索其他决策路径都很有用。
我们将这些调试概念统称为“时光回溯”，它们由“重放”和“分支”组成。
重放¶
有时我们只想简单地重放智能体过去的操作。
上面我们展示了从图的当前状态（或检查点）执行智能体的情况。
我们只需通过传入 None 作为输入和一个 thread 即可。
thread = {"configurable": {"thread_id": "1"}}
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
现在，我们可以通过传入检查点 ID 来修改此操作，从而从*特定*检查点重放过去的操作。
要获取特定的检查点 ID，我们可以轻松获取线程中的所有检查点并筛选出我们想要的那个。
每个检查点都有一个唯一的 ID，我们可以使用它从特定检查点进行重放。
假设通过审查检查点，我们想从一个名为 xxx 的检查点进行重放。
我们只需在运行图时传入检查点 ID。
config = {'configurable': {'thread_id': '1', 'checkpoint_id': 'xxx'}}
for event in graph.stream(None, config, stream_mode="values"):
print(event)
重要的是，图知道哪些检查点之前已经执行过。
因此，它将重放任何之前执行过的节点，而不是重新执行它们。
有关重放的相关背景信息，请参阅此附加概念指南。
有关时光回溯的详细操作指南，请参阅本指南！
分支¶
有时我们想对智能体过去的操作进行分支，并探索图中的不同路径。
如上文所述，“编辑”*正是*我们对图的*当前*状态进行此操作的方式！
但是，如果我们想对图的*过去*状态进行分支呢？
例如，假设我们想编辑一个特定的检查点 xxx。
我们在更新图的状态时传入这个 checkpoint_id。
config = {"configurable": {"thread_id": "1", "checkpoint_id": "xxx"}}
graph.update_state(config, {"state": "updated state"}, )
这将创建一个新的分支检查点 xxx-fork，然后我们可以从该检查点运行图。
config = {'configurable': {'thread_id': '1', 'checkpoint_id': 'xxx-fork'}}
for event in graph.stream(None, config, stream_mode="values"):
print(event)
有关分支的相关背景信息，请参阅此附加概念指南。
有关时光回溯的详细操作指南，请参阅本指南！