[LangGraph] Human-in-the-loop示例之人工干预shell命令执行

引言在现代AI应用中，智能体（Agent）自主执行shell命令的能力越来越常见，例如自动化运维、代码生成与执行、数据分析等场景。然而，完全自主的shell执行存在巨大的安全风险：删除文件、修改系统配置、执行恶意命令等操作一旦失控，后果不堪设想。因此，Human-in-the-loop（人机协同）模式成为解决这一问题的关键方案。LangGraph是LangChain生态中专注于构建有状态、多步骤、图结构智能体的框架。它天然支持状态持久化和“中断-恢复”机制，非常适合实现人工审核与干预的场景。本文将深入剖析LangGraph中Human-in-the-loop的原理，并通过一个可运行的示例，展示如何让智能体在执行shell命令前等待人工确认或修改。## Human-in-the-loop的核心机制在LangGraph中，Human-in-the-loop的底层实现依赖于两个关键概念：1.节点函数（Node Function）与中断（Interrupt）：每个节点可以调用interrupt()函数，挂起当前图的执行，等待外部输入。执行状态会被持久化到存储后端（如内存、数据库）。2.状态更新与恢复：外部系统（如用户界面）可以获取当前状态，并提交更新。LangGraph会从断点处继续执行，注入新的输入。这种设计使得我们可以在任意节点插入人工审核逻辑，而不需要修改整个图的拓扑结构。## 示例：人工干预shell命令执行假设我们有一个智能体，它可以根据用户的需求生成并执行shell命令。我们希望在命令执行前，让用户审查命令内容：同意执行、修改命令、或拒绝执行。### 系统架构我们将构建一个简单的LangGraph图，包含三个节点：-parse_input：解析用户输入，生成shell命令。-human_review：中断点，等待人工审核。用户通过输入“Y”、“N”或修改后的命令来响应。-execute_command：根据审核结果执行或跳过命令。### 代码示例1：定义状态与节点函数pythonfrom typing import TypedDict, Optionalfrom langgraph.graph import StateGraph, ENDfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.constants import interrupt # 关键：中断函数# 定义状态类型class AgentState(TypedDict): user_input: str # 用户原始输入 generated_command: Optional[str] # 生成的shell命令 human_decision: Optional[str] # 人工决策：'Y'表示执行，'N'表示拒绝，其他字符串视为修改后的命令 execution_result: Optional[str] # 命令执行结果# 节点1：解析输入，生成shell命令def parse_input_node(state: AgentState) -> AgentState: """根据用户输入生成shell命令（模拟）""" user_input = state["user_input"] # 这里简化处理：直接根据关键词生成命令 if "list" in user_input.lower(): command = "ls -la" elif "disk" in user_input.lower(): command = "df -h" else: command = f"echo 'Unknown command for: {user_input}'" print(f"[Agent] 生成命令: {command}") return {"generated_command": command}# 节点2：人工审核节点（中断点）def human_review_node(state: AgentState) -> AgentState: """等待人工输入，返回决策结果""" command = state["generated_command"] # 中断，等待外部输入 # interrupt() 会挂起图执行，并等待用户通过外部接口提交数据 user_response = interrupt( f"请审核即将执行的命令:\n{command}\n输入 'Y' 执行，'N' 拒绝，或直接输入修改后的命令:" ) # 用户响应可以是字符串 return {"human_decision": user_response}# 节点3：根据决策执行命令def execute_command_node(state: AgentState) -> AgentState: """根据人工决策执行或修改命令""" decision = state["human_decision"] original_command = state["generated_command"] if decision.strip().upper() == "Y": # 同意执行原始命令 cmd_to_execute = original_command print(f"[Agent] 人工同意，执行命令: {cmd_to_execute}") elif decision.strip().upper() == "N": # 拒绝执行 print("[Agent] 人工拒绝，跳过执行") return {"execution_result": "Command rejected by human"} else: # 用户提供了修改后的命令 cmd_to_execute = decision.strip() print(f"[Agent] 人工修改命令，执行: {cmd_to_execute}") # 模拟执行命令（实际环境请使用subprocess等） import subprocess try: result = subprocess.run(cmd_to_execute, shell=True, capture_output=True, text=True, timeout=5) output = result.stdout + result.stderr print(f"[Agent] 执行结果:\n{output}") return {"execution_result": output} except Exception as e: print(f"[Agent] 执行失败: {str(e)}") return {"execution_result": f"Error: {str(e)}"}# 构建图workflow = StateGraph(AgentState)# 添加节点workflow.add_node("parse_input", parse_input_node)workflow.add_node("human_review", human_review_node)workflow.add_node("execute_command", execute_command_node)# 设置边：从开始到parse_input，然后到human_review，最后到execute_commandworkflow.set_entry_point("parse_input")workflow.add_edge("parse_input", "human_review")workflow.add_edge("human_review", "execute_command")workflow.add_edge("execute_command", END)# 编译图，启用持久化（MemorySaver）app = workflow.compile(checkpointer=MemorySaver())### 代码示例2：运行智能体并注入人工干预pythonimport uuiddef run_agent_with_human(user_input: str): """运行智能体，模拟人工审核过程""" # 初始化状态 initial_state = {"user_input": user_input} # 生成唯一线程ID（用于状态持久化） thread_id = str(uuid.uuid4()) config = {"configurable": {"thread_id": thread_id}} # 第一次执行：直到中断点 print("="*50) print(f"用户输入: {user_input}") print("第一次执行（到达审核节点前）...") # 使用stream模式，逐节点产出结果 events = [] for event in app.stream(initial_state, config, stream_mode="values"): events.append(event) if "generated_command" in event: print(f"当前状态: {event}") # 获取当前状态（包含中断信息） current_state = app.get_state(config) print(f"中断信息: {current_state.tasks}") # 显示中断任务详情 # 模拟人工审核：这里我们让用户输入决策 # 实际应用中可通过Web UI、CLI等交互 print("\n--- 人工审核环节 ---") human_input = input("请输入您的决策 (Y/N/修改后的命令): ").strip() # 更新状态：提交人工决策，恢复执行 # 注意：update_state会从断点继续执行 app.update_state(config, {"human_decision": human_input}, as_node="human_review") # 第二次执行：从中断点继续，直到结束 print("\n第二次执行（从审核节点继续）...") for event in app.stream(None, config, stream_mode="values"): events.append(event) if "execution_result" in event: print(f"最终状态: {event}") # 打印最终结果 final_state = app.get_state(config) print(f"\n最终执行结果: {final_state.values.get('execution_result', '无结果')}") print("="*50)# 运行示例if __name__ == "__main__": # 示例1：用户请求列出文件 run_agent_with_human("请列出当前目录的文件") # 示例2：用户请求查看磁盘信息 # run_agent_with_human("查看磁盘使用情况")### 运行效果演示假设用户输入“请列出当前目录的文件”，智能体生成命令ls -la。程序会输出：==================================================用户输入: 请列出当前目录的文件第一次执行（到达审核节点前）...[Agent] 生成命令: ls -la当前状态: {'user_input': '请列出当前目录的文件', 'generated_command': 'ls -la', ...}中断信息: [...]--- 人工审核环节 ---请输入您的决策 (Y/N/修改后的命令): Y第二次执行（从审核节点继续）...[Agent] 人工同意，执行命令: ls -la[Agent] 执行结果:total 24drwxr-xr-x 5 user staff 160 Aug 10 12:00 ....最终执行结果: Command output...如果用户输入“N”，则命令被跳过；如果用户输入“ls -al”，则执行修改后的命令。## 原理深入剖析### 中断与状态持久化的协同工作LangGraph的interrupt()函数本质上是一个特殊的“状态标记”。当节点调用它时，框架会：1. 将当前图的执行上下文（包括局部变量、节点指针）序列化并保存到checkpointer中。2. 返回一个控制权给外部调用者，后者可以通过get_state()获取中断信息。3. 当外部调用update_state()时，框架反序列化状态，从断点处恢复执行，并将新数据注入到中断节点。这种机制类似于操作系统的“进程挂起与恢复”，但专门为图计算设计。在我们的示例中，human_review_node调用interrupt()后，图的状态中会记录tasks字段，包含中断时传递给用户的消息。### 为什么不是简单的条件分支？传统方法可能通过条件判断（if-else）实现人工审核，例如：pythonif need_review: wait_for_input()else: execute()但这种方式的缺陷是：-状态丢失：如果程序崩溃，审核状态丢失。-并发困难：多个会话共享状态时容易混乱。-难以回放：无法回溯审核历史。LangGraph的图结构天然解决了这些问题：每个会话有独立thread_id，状态持久化到存储后端，可以随时中断、恢复、回放。## 总结本文通过一个具体的shell命令执行审核示例，详细展示了如何在LangGraph中实现Human-in-the-loop。核心收获包括：1.interrupt()函数：在任意节点挂起图执行，等待外部输入，是实现人工干预的基石。2.状态持久化：通过MemorySaver或其他checkpointer，确保会话状态不丢失，支持断点恢复。3.灵活的审核逻辑：用户不仅可以同意/拒绝，还可以修改命令内容，体现了人机协同的灵活性。LangGraph的Human-in-the-loop机制不仅适用于shell命令审核，还可广泛应用于：- AI辅助编程中的代码审查- 金融交易中的风险控制- 自动驾驶中的紧急决策通过将“人”引入循环，我们既利用了AI的自动化能力，又保留了人类的判断力，实现了安全与效率的平衡。