人在回路中¶
请改用 interrupt 函数。
从 LangGraph 0.2.57 版本开始，推荐使用 interrupt 函数来设置断点，因为它简化了人在回路中的模式。
请参阅修订后的 人在回路中指南，以获取使用 interrupt 函数的最新版本。
人在回路中（或“环内”）通过几种常见的用户交互模式增强了代理的能力。
常见的交互模式包括
(1) 审批 - 我们可以中断我们的代理，将当前状态呈现给用户，并允许用户接受一个操作。
(2) 编辑 - 我们可以中断我们的代理，将当前状态呈现给用户，并允许用户编辑代理状态。
(3) 输入 - 我们可以显式创建一个图节点来收集人工输入，并将该输入直接传递给代理状态。
这些交互模式的用例包括
(1) 审查工具调用 - 我们可以中断代理以审查和编辑工具调用的结果。
(2) 时间旅行 - 我们可以手动回放和/或分支代理的过去操作。
持久化¶
所有这些交互模式都由 LangGraph 内置的 持久化层启用，该层将在每个步骤写入图状态的检查点。持久化允许图停止，以便人类可以审查和/或编辑图的当前状态，然后使用人类的输入恢复。
断点¶
在图流程中的特定位置添加 断点 是启用人在回路中的一种方法。在这种情况下，开发人员知道工作流程中哪里需要人工输入，只需在该特定图节点之前或之后放置一个断点。
在这里，我们使用检查点和断点编译我们的图，断点位于我们想要在 step_for_human_in_the_loop 之前中断的节点。然后我们执行上述交互模式之一，如果人类编辑图状态，这将创建一个新的检查点。新的检查点将保存到 thread，我们可以通过传入 None 作为输入，从那里恢复图的执行。
# Compile our graph with a checkpointer and a breakpoint before "step_for_human_in_the_loop"
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["step_for_human_in_the_loop"])
# Run the graph up to the breakpoint
thread_config = {"configurable": {"thread_id": "1"}}
for event in graph.stream(inputs, thread_config, stream_mode="values"):
print(event)
# Perform some action that requires human in the loop
# Continue the graph execution from the current checkpoint
for event in graph.stream(None, thread_config, stream_mode="values"):
print(event)
动态断点¶
或者，开发人员可以定义一些必须满足的条件才能触发断点。当开发人员想要在特定条件下停止图时，动态断点 的概念非常有用。这使用了 NodeInterrupt，这是一种特殊的异常类型，可以根据某些条件从节点内部引发。例如，我们可以定义一个动态断点，当输入长度超过 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 作为输入来恢复图的执行。
# Attempt to continue the graph execution with no change to state after we hit the dynamic breakpoint
for event in graph.stream(None, thread_config, stream_mode="values"):
print(event)
图将再次中断，因为此节点将使用相同的图状态重新运行。我们需要更改图状态，以便不再满足触发动态断点的条件。因此，我们可以简单地将图状态编辑为满足我们的动态断点条件（< 5 个字符）的输入，并重新运行节点。
# Update the state to pass the dynamic breakpoint
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 运行，从而有效地跳过节点并绕过动态断点。
# This update will skip the node `my_node` altogether
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
# Compile our graph with a checkpointer and a breakpoint before the step to approve
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["node_2"])
# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
print(event)
# ... Get human approval ...
# If approved, continue the graph execution from the last saved checkpoint
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
请参阅 我们的指南，了解有关如何执行此操作的详细说明！
编辑¶
有时我们想审查和编辑代理的状态。
与批准一样，我们可以在我们想要检查的步骤之前的 断点 处中断我们的代理。
我们可以将当前状态呈现给用户，并允许用户编辑代理状态。
例如，这可以用于纠正代理是否犯了错误（例如，请参阅下面的工具调用部分）。
我们可以通过分支当前检查点来编辑图状态，该检查点已保存到 thread。
然后我们可以像以前一样从我们的分支检查点继续进行图。
# Compile our graph with a checkpointer and a breakpoint before the step to review
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["node_2"])
# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
print(event)
# Review the state, decide to edit it, and create a forked checkpoint with the new state
graph.update_state(thread, {"state": "new state"})
# Continue the graph execution from the forked checkpoint
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
请参阅 本指南，了解有关如何执行此操作的详细说明！
输入¶
有时我们想在图中的特定步骤显式获取人工输入。
我们可以为此创建一个指定的图节点（例如，在我们的示例图中为 human_input）。
与批准和编辑一样，我们可以在此节点之前的 断点 处中断我们的代理。
然后我们可以执行包含人工输入的状态更新，就像我们编辑状态一样。
但是，我们添加一件事
我们可以将 as_node=human_input 与状态更新一起使用，以指定状态更新应被视为节点。
这很微妙，但很重要
通过编辑，用户决定是否编辑图状态。
通过输入，我们在图中显式定义一个用于收集人工输入的节点！
包含人工输入的状态更新然后作为此节点运行。
# Compile our graph with a checkpointer and a breakpoint before the step to to collect human input
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["human_input"])
# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
print(event)
# Update the state with the user input as if it was the human_input node
graph.update_state(thread, {"user_input": user_input}, as_node="human_input")
# Continue the graph execution from the checkpoint created by the human_input node
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
请参阅 本指南，了解有关如何执行此操作的详细说明！
用例¶
审查工具调用¶
一些用户交互模式结合了上述想法。
例如，许多代理使用 工具调用 来做出决策。
工具调用提出了一个挑战，因为代理必须做对两件事
(1) 要调用的工具的名称
(2) 传递给工具的参数
即使工具调用是正确的，我们也可能想要应用酌处权
(3) 工具调用可能是一个我们想要批准的敏感操作
考虑到这些要点，我们可以结合上述想法来创建工具调用的人在回路中审查。
# Compile our graph with a checkpointer and a breakpoint before the step to to review the tool call from the LLM
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["human_review"])
# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
print(event)
# Review the tool call and update it, if needed, as the human_review node
graph.update_state(thread, {"tool_call": "updated tool call"}, as_node="human_review")
# Otherwise, approve the tool call and proceed with the graph execution with no edits
# Continue the graph execution from either:
# (1) the forked checkpoint created by human_review or
# (2) the checkpoint saved when the tool call was originally made (no edits in human_review)
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
请参阅 本指南，了解有关如何执行此操作的详细说明！
时间旅行¶
当与代理一起工作时，我们经常希望仔细检查他们的决策过程
(1) 即使当他们达到期望的最终结果时，导致该结果的推理也通常很重要，需要检查。
(2) 当代理犯错时，了解原因通常很有价值。
(3) 在上述任何一种情况下，手动探索替代的决策路径都很有用。
总的来说，我们将这些调试概念称为 时间旅行，它们由 回放 和 分支 组成。
回放¶
有时我们只想简单地回放代理的过去操作。
上面，我们展示了从图的当前状态（或检查点）执行代理的情况。
我们通过简单地为带有 thread 的输入传入 None 来实现。
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)
请参阅 此额外的概念指南，以获取有关分支的相关背景。
请参阅 本指南，了解有关如何执行时间旅行的详细说明！