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

[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 fecho 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, shellTrue, capture_outputTrue, textTrue, timeout5) 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: fError: {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)# 编译图启用持久化MemorySaverapp workflow.compile(checkpointerMemorySaver())### 代码示例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_modevalues): 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_nodehuman_review) # 第二次执行从中断点继续直到结束 print(\n第二次执行从审核节点继续...) for event in app.stream(None, config, stream_modevalues): 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的自动化能力又保留了人类的判断力实现了安全与效率的平衡。