人机交互¶
请使用interrupt函数替代。
自LangGraph 0.2.57版本以来，推荐使用interrupt函数来设置断点，因为它简化了**人机交互**模式。
请参阅最新版本的人机交互指南，该版本使用了interrupt函数。
人机交互（或“人在回路中”）通过几种常见的用户交互模式增强了代理的能力。
常见的交互模式包括：
(1) 批准 - 我们可以中断代理，向用户展示当前状态，并允许用户接受某个操作。
(2) 编辑 - 我们可以中断代理，向用户展示当前状态，并允许用户编辑代理的状态。
(3) 输入 - 我们可以显式创建一个图节点以收集人类输入，并直接将输入传递给代理状态。
这些交互模式的应用场景包括：
(1) 审查工具调用 - 我们可以中断代理以审查和编辑工具调用的结果。
(2) 时间旅行 - 我们可以手动重新播放和/或分叉代理过去的动作。
持久化¶
所有这些交互模式都是由LangGraph内置的persistence层支持的，该层会在每一步写入图状态的检查点。持久化允许图暂停，以便人类可以审查和/或编辑当前图的状态，然后根据人类的输入继续执行。
断点¶
在图流程中的特定位置添加一个断点是启用“人在环中”（human-in-the-loop）的一种方式。在这种情况下，开发人员知道在工作流中哪个地方需要人类输入，并且只需在特定图节点之前或之后放置一个断点即可。
这里，我们使用一个检查点器和一个断点，在我们想要中断的节点之前，即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"收到的输入比5个字符长：{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": "新状态"})
# 从分叉的检查点继续图的执行
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) 工具调用可能是敏感操作，我们希望批准
考虑到这些要点，我们可以结合上述想法来创建一个包含人类审查的工具调用流程。
# 编译我们的图，并在LLM中审查工具调用之前添加检查点器和断点
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)
参见此附加概念指南以了解有关分叉的相关背景。
参见此指南以了解详细的操作方法！