人机交互功能可让我们让user参与到graph的决策过程中。下面操作指南展示了如何在graph中实现人机交互工作流程。人机交互工作流将user input 集成到自动化流程中，允许在关键阶段做出决策、验证或更正。这在基于 LLM 的应用程序中尤其有用，因为底层模型可能会偶尔产生不准确之处(幻觉)。在合规性、决策或内容生成等低容错场景中，人工参与可通过审查、更正或覆盖模型输出来确保可靠性。基于 LLM 的应用程序中的人机交互工作流程的主要用例包括：🛠️ 审查工具调用：人类可以在工具执行之前审查、编辑或批准 LLM 请求的工具调用。✅ 验证 LLM 输出：人类可以审查、编辑或批准 LLM 生成的内容。💡 提供背景信息：使 LLM 能够明确请求人工输入以澄清或其他详细信息或支持多轮对话。
LangGraph 中的interrupt function通过在特定节点暂停graph、向人类呈现信息并根据他们的输入恢复放入graph来实现人机交互工作流程。interrupt function与 Command 对象结合使用，以使用人工提供的值恢复graph运行。Command是实现条件边的另一种方式。下面是一个核心代码展示from typing import TypedDictimport uuidfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.constants import STARTfrom langgraph.graph import StateGraphfrom langgraph.types import interrupt, Commandclass State(TypedDict): some_text: strdef human_node(state: State): value = interrupt( # 任何可序列化为JSON的值以展示给人类 # 例如，一个问题或一段文字或状态中的一组键 { "text_to_revise": state["some_text"] } ) return { # 使用人类的输入更新状态 "some_text": value }# 构建图graph_builder = StateGraph(State)# 将人类节点添加到图中graph_builder.add_node("human_node", human_node)graph_builder.add_edge(START, "human_node")# `interrupt`需要一个检查点checkpointer = MemorySaver()graph = graph_builder.compile( checkpointer=checkpointer)# 传递一个线程ID给图以运行它thread_config = {"configurable": {"thread_id": uuid.uuid4()}}# 使用`stream()`直接展示`__interrupt__`信息for chunk in graph.stream({"some_text": "原始文本"}, config=thread_config): print(chunk)# 使用Command恢复for chunk in graph.stream(Command(resume="编辑后的文本"), config=thread_config): print(chunk)
{'__interrupt__': (Interrupt(value={'text_to_revise': '原始文本'}, resumable=True, ns=['human_node:1cee88f8-54f0-5c9e-a5b9-f67ae9b81448'], when='during'),)}{'human_node': {'some_text': '编辑后的文本'}}
⚠️：中断既强大又符合人机工程学。然而，虽然从开发人员体验来看，它们可能类似于 Python 的 input() 函数。但需要注意的是，它们不会自动从中断点恢复执行。相反，它们会重新运行使用中断的整个节点。因此，中断通常最好放置在一个node的开头或专用node中。要想在graph中使用interrupt，必须要满足以下条件：- 指定一个checkpoint来保存每个步骤之后的图形状态。
- 使用thread id 运行graph，直到发生中断。
- 使用invoke/ainvoke/stream/astream恢复执行
通常，我们可以通过人机交互工作流程执行三种不同的操作：- 批准或拒绝：在关键步骤（例如 API 调用）之前暂停图表，以审查并批准操作。如果操作被拒绝，您可以阻止图表执行该步骤，并可能采取替代操作。这种模式通常涉及根据人类的输入来路由graph。
- 编辑graph state：暂停图表以查看和编辑图表状态。这对于纠正错误或使用附加信息更新状态很有用。此模式通常涉及使用人工输入更新状态。
- 获取input：在图表中的特定步骤明确请求人工输入。这对于收集其他信息或背景信息以告知代理的决策过程或支持多轮对话非常有用。
下面我们展示了可以使用这些操作实现的不同设计模式。
根据人类的批准或拒绝，graph可以继续执行操作或采取替代路径。from typing import Literalfrom langgraph.types import interrupt, Command
def human_approval(state: State) -> Command[Literal["some_node", "another_node"]]: # 提出问题并中断，等待人工批准 is_approved = interrupt( { "question": "这是否正确？", # 显示需要人工审核和批准的输出 "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)
人类可以查看和编辑graph的状态。这对于纠正错误或用附加信息更新状态很有用。from langgraph.types import interruptdef human_editing(state: State): ... result = interrupt( # 中断信息浮现到客户端。 # 可以是任何可序列化为JSON的值。 { "task": "查看LLM的输出并进行必要的编辑。", "llm_generated_summary": state["llm_generated_summary"] } ) # 使用编辑后的文本更新状态 return { "llm_generated_summary": result["edited_text"] }# 在适当的位置将节点添加到图中# 并将其连接到相关node。graph_builder.add_node("human_editing", human_editing)graph = graph_builder.compile(checkpointer=checkpointer)...# 运行图并触发中断后，graph将暂停。# 使用编辑后的文本恢复它。thread_config = {"configurable": {"thread_id": "some_id"}}graph.invoke( Command(resume={"edited_text": "编辑后的文本"}), config=thread_config)
人类可以在继续之前检查和编辑 LLM 的输出。这在 LLM 请求的工具调用可能很敏感或需要人类监督的应用中尤其重要。def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]: # 这是我们将通过 Command(resume=<human_review>) 提供的值 human_review = interrupt( { "question": "这是否正确？", # 提供工具调用以供审查 "tool_call": tool_call } ) review_action, review_data = human_review # 批准工具调用并继续 if review_action == "continue": return Command(goto="run_tool") # 手动修改工具调用然后继续 elif review_action == "update": ... updated_msg = get_updated_msg(review_data) # 请记住，要修改现有消息，您需要传递具有匹配 ID 的消息。 return Command(goto="run_tool", update={"messages": [updated_message]}) # 提供自然语言反馈，然后将其传递回代理 elif review_action == "feedback": ... feedback_msg = get_feedback_msg(review_data) return Command(goto="call_llm", update={"messages": [feedback_msg]})
一种多轮对话架构，其中代理和人类节点来回循环，直到代理决定将对话交给另一个代理或系统的另一部分。多轮对话涉及代理和人类之间的多次来回交互，这可以让代理以对话的方式从人类那里收集更多信息。这种设计模式在由多个代理组成的 LLM 应用程序中非常有用。一个或多个代理可能需要与人类进行多轮对话，其中人类在对话的不同阶段提供输入或反馈。为简单起见，下面的代理实现被示出为单个节点，但实际上它可能是由多个节点组成的更大图的一部分并包含条件边。在这种模式中，每个代理都有自己的人工节点来收集用户输入，这可以通过使用唯一名称命名人类节点（例如，“代理 1 的人类”、“代理 2 的人类”）或使用子图（其中子图包含人类节点和代理节点）来实现。from langgraph.types import interrupt
def human_input(state: State): human_message = interrupt("human_input") return { "messages": [ { "role": "human", "content": human_message } ] }
def agent(state: State): # 代理逻辑 ...
graph_builder.add_node("human_input", human_input)graph_builder.add_edge("human_input", "agent")graph = graph_builder.compile(checkpointer=checkpointer)
# 运行图并触发中断后，图将暂停。# 使用人类的输入恢复它。graph.invoke( Command(resume="hello!"), config=thread_config)
在此模式中，单个人工节点用于收集多个代理的用户输入。根据状态确定活动代理，因此在收集人工输入后，graph可以路由到正确的代理。from langgraph.types import interrupt
def human_node(state: MessagesState) -> Command[Literal["agent_1", "agent_2", ...]]: """用于收集用户输入的节点。""" user_input = interrupt(value="准备好接收用户输入。")
# 以便在收集输入后可以路由到正确的代理。 # 例如，向状态添加一个字段或使用最后一个活动代理。 # 或者填写由代理生成的AI消息的`name`属性。 active_agent = ...
return Command( update={ "messages": [{ "role": "human", "content": user_input, }] }, goto=active_agent, )
如果需要验证graph本身内（而不是在客户端）人类提供的输入，可以通过在单个节点内使用多个中断调用来实现这一点。from langgraph.types import interrupt
def human_node(state: State): """带有验证的人工节点。""" question = "你的年龄是多少？"
while True: answer = interrupt(question)
if not isinstance(answer, int) or answer < 0: question = f"'{answer}' 不是一个有效的年龄。你的年龄是多少？" answer = None continue else: # 如果答案有效，我们可以继续。 break
print(f"循环中的人类年龄是 {answer} 岁。") return { "age": answer }
当使用中断功能时，graph会在中断处暂停并等待用户输入。可以使用可通过invoke、ainvoke、stream或astream方法传递的Command原语来恢复图形执行。Command 原语提供了几个选项来在恢复期间控制和修改graph的状态：Pass a value to the interrupt：使用 Command(resume=value) 向graph提供数据，例如用户的响应。从发生interrupt的节点开始处恢复执行，但是，这次interrupt(...)调用将返回在 Command(resume=value)中传递的值，而不是暂停graph。使用Command(update=update) 改变graph state的值。从发生interrupt的节点开始处恢复执行 ，但使用更新的状态。# Update the graph state and resume.# 你必须提供一个 `resume` value，如果使用了 `interrupt`.graph.invoke(Command(update={"foo": "bar"}, resume="Let's go!!!"), thread_config)
通过利用命令，您可以恢复graph形执行，处理用户输入，并动态调整graph的状态。
从中断恢复不同于 Python 的 input() 函数，后者执行从调用 input() 函数的确切位置恢复。使用中断的一个关键方面是了解恢复的工作原理。当在中断后恢复执行时，graph执行将从上次触发中断的图形节点的开头开始。从该node的 开始到中断的之间所有代码都将被重新执行（这也就是说为什么建议将中断放在某个node的最开始位置）。counter = 0def node(state: State): # 当图恢复时，从节点开始到中断的所有代码将被重新执行。 global counter counter += 1 print(f"> 进入节点的次数: {counter}") # 暂停图并等待用户输入。 answer = interrupt() print("counter的值是:", counter) ...
> Entered the node: 2 The value of counter is: 2
将具有副作用（非幂等操作）的代码（例如 API 调用）放置在中断之后以避免重复，因为每次节点恢复时都会重新触发这些代码。当将子图作为函数调用时，父图将从调用子图的节点（以及触发中断的节点）的开头恢复执行，同理，子图将从调用interrupt（）函数的节点的开头恢复。在单个节点中使用多个中断对于验证人工输入等模式很有帮助。但是，如果处理不当，在同一个节点中使用多个中断可能会导致意外行为。当一个节点包含多个中断调用时，LangGraph 会保留一个特定于执行该节点的任务的恢复值列表。每当执行恢复时，它都会从节点的开头开始。对于遇到的每个中断，LangGraph 都会检查任务的恢复列表中是否存在匹配值。匹配严格基于index，因此节点内中断调用的顺序至关重要。为避免出现问题，请不要在执行期间动态更改节点的结构。这包括添加、删除或重新排序中断调用，因为此类更改可能会导致index不匹配。这些问题通常源于非常规模式，例如通过 Command(resume=..., update=SOME_STATE_MUTATION) 改变状态或依赖全局变量动态修改节点的结构。
