langchain中间件之人工审核件：HumanInTheLoopMiddleware

HumanInTheLoopMiddleware的意义HumanInTheLoopMiddleware 是 LangChain 中用于在 Agent 执行过程中插入人工审核节点的中间件。它允许你在工具被实际调用前或调用后暂停流程等待人工批准、修改或拒绝是构建安全、可控智能体的核心组件。HumanInTheLoopMiddleware的作用时机HumanInTheLoopMiddleware 的默认介入时机是在工具实际执行之前。换句话说当 Agent 决定调用某个工具并生成了工具调用的参数后中间件会立刻中断流程把“待审批的工具调用请求”交给你此时工具还没真正执行。在 create_agent 构建的图结构中执行流程大致为模型节点生成工具调用→ 中断点 → 工具节点执行工具代码实战step1 配置模型APIfrom langchain_openai import ChatOpenAI import json with open(rconfig/model_call.json, r, encodingutf-8) as f: config json.load(f) # 模型初始化 llm ChatOpenAI( api_keyconfig[api_key], base_urlconfig[base_url], modelconfig[model], temperature0.7, )本人使用的是deepseek的API并将密钥信息保存在了本地的config/model_call.json文件之中如果为ollama调用本地模型等其他方式调用只需更改与之适配的模型调用方式即可step2 配置模型工作所需的工具from langchain_core.tools import tool import os from pathlib import Path tool def create_py_file(file_path, code): 创建py文件 try: dir_name os.path.dirname(file_path) if dir_name and not os.path.exists(dir_name): os.makedirs(dir_name) with open(file_path, w, encodingutf-8) as f: f.write(code) return f✅ 文件创建成功{file_path} except Exception as e: return f创建py文件失败: {e} tool def get_filenames_pathlib(folder_path: str) - list: 获取指定文件夹下的所有文件和文件夹名称。当用户想查看某个目录下有哪些文件时使用。 try: path Path(folder_path) all_items [item.name for item in path.iterdir()] return all_items except Exception as e: return [f错误: {e}] tool def read_file(file_path: str) - str: 读取指定路径的文本文件内容。当用户要求查看文件内容时使用。 try: with open(file_path, r, encodingutf-8) as f: return f.read() except Exception as e: return f读取出错: {e} tool def get_desktop_path(): 获取当前用户的桌面文件夹路径。支持 Windows、macOS、Linux。 try: # 方法1使用 pathlib 获取桌面路径 desktop Path.home() / Desktop # 如果 Desktop 不存在尝试中文桌面部分 Linux 发行版 if not desktop.exists(): desktop_cn Path.home() / 桌面 if desktop_cn.exists(): desktop desktop_cn # 如果还不存在使用 os.path 方式 if not desktop.exists(): desktop_os os.path.join(os.path.expanduser(~), Desktop) if os.path.exists(desktop_os): desktop Path(desktop_os) else: desktop_cn_os os.path.join(os.path.expanduser(~), 桌面) if os.path.exists(desktop_cn_os): desktop Path(desktop_cn_os) desktop_path str(desktop) return f桌面路径: {desktop_path} except Exception as e: return f获取桌面路径失败: {e}使用 LangChain 的 tool 装饰器来定义函数tool 装饰器的作用是将一个普通的 Python 函数转换为 Agent 能够理解并使用的“工具”对象。本人代码以几个简单的功能函数为例获取桌面位置、查看文件夹内文件、读取文件内容的功能函数step3 将工具集成到agent中from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver from langchain.agents.middleware import HumanInTheLoopMiddleware system_prompt 你是一个智能助手可以访问桌面、读取文件、打开网页、创建py文件 你可以在model_tools文件夹下自由创建你需要用到的py文件 model_tools添加好的工具需要重写auto_agent_prompt.py才会生效 打开软件先从桌面找没有的话再从网页打开 hitl_middleware HumanInTheLoopMiddleware( interrupt_on{get_filenames_pathlib: {allowed_decisions: [approve, reject]}}, description_prefix获取指定文件夹下的所有文件和文件夹名称需要人工审核 ) # 创建检查点器 checkpointer InMemorySaver() agent2 create_agent( modelllm, tools[read_file, open_url, create_py_file, get_filenames_pathlib, copy_file, get_desktop_path], system_promptsystem_prompt, middleware[hitl_middleware], checkpointercheckpointer )在创建create_agent之前需要提前写好有关智能体的提示词部分在提示词中澄清模型调用工具的规则InMemorySaver 是 LangGraphLangChain Agent 的底层框架中提供的一种检查点存储器。它的核心作用可以总结为让 Agent 具备“记忆”和“断点续传”的能力。HumanInTheLoopMiddleware内的interrupt_on定义模型调用时需要检查的工具step4 全部代码import json from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, SystemMessage from pathlib import Path import os import webbrowser from langchain.agents.middleware import HumanInTheLoopMiddleware,InterruptOnConfig from langchain.agents import create_agent from langchain_core.callbacks import BaseCallbackHandler from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command with open(rconfig/model_call.json, r, encodingutf-8) as f: config json.load(f) # 工具初始化 tool def create_py_file(file_path, code): 创建py文件 try: dir_name os.path.dirname(file_path) if dir_name and not os.path.exists(dir_name): os.makedirs(dir_name) with open(file_path, w, encodingutf-8) as f: f.write(code) return f✅ 文件创建成功{file_path} except Exception as e: return f创建py文件失败: {e} tool def get_filenames_pathlib(folder_path: str) - list: 获取指定文件夹下的所有文件和文件夹名称。当用户想查看某个目录下有哪些文件时使用。 try: path Path(folder_path) all_items [item.name for item in path.iterdir()] return all_items except Exception as e: return [f错误: {e}] tool def read_file(file_path: str) - str: 读取指定路径的文本文件内容。当用户要求查看文件内容时使用。 try: with open(file_path, r, encodingutf-8) as f: return f.read() except Exception as e: return f读取出错: {e} tool def get_desktop_path(): 获取当前用户的桌面文件夹路径。支持 Windows、macOS、Linux。 try: # 方法1使用 pathlib 获取桌面路径 desktop Path.home() / Desktop # 如果 Desktop 不存在尝试中文桌面部分 Linux 发行版 if not desktop.exists(): desktop_cn Path.home() / 桌面 if desktop_cn.exists(): desktop desktop_cn # 如果还不存在使用 os.path 方式 if not desktop.exists(): desktop_os os.path.join(os.path.expanduser(~), Desktop) if os.path.exists(desktop_os): desktop Path(desktop_os) else: desktop_cn_os os.path.join(os.path.expanduser(~), 桌面) if os.path.exists(desktop_cn_os): desktop Path(desktop_cn_os) desktop_path str(desktop) return f桌面路径: {desktop_path} except Exception as e: return f获取桌面路径失败: {e} # 模型初始化 llm ChatOpenAI( api_keyconfig[api_key], base_urlconfig[base_url], modelconfig[model], temperature0.7, ) # 构建 Agent system_prompt 你是一个智能助手可以访问桌面、读取文件、打开网页、创建py文件 你可以在model_tools文件夹下自由创建你需要用到的py文件 model_tools添加好的工具需要重写auto_agent_prompt.py才会生效 打开软件先从桌面找没有的话再从网页打开 # interrupt_on{get_filenames_pathlib: {allowed_decisions: [approve, reject]}}, hitl_middleware HumanInTheLoopMiddleware( interrupt_on{get_filenames_pathlib: True}, description_prefix获取指定文件夹下的所有文件和文件夹名称需要人工审核 ) # 创建检查点器 checkpointer InMemorySaver() agent2 create_agent( modelllm, tools[read_file, create_py_file, get_filenames_pathlib, get_desktop_path], system_promptsystem_prompt, middleware[hitl_middleware], checkpointercheckpointer ) def extract_interrupt_data(result): 从返回结果中提取人工审核信息 if isinstance(result, dict) and __interrupt__ in result: interrupts result[__interrupt__] if interrupts: interrupt_obj interrupts[0] value getattr(interrupt_obj, value, None) if value and action_requests in value: return value[action_requests][0] return None def test_main(user_input我的桌面放了几本小说): config {configurable: {thread_id: user-session-1}} # 第一次调用 result agent2.invoke( {messages: [{role: user, content: user_input}]}, configconfig ) while True: review_data extract_interrupt_data(result) if not review_data: # 没有中断输出最终结果并退出 print(最终结果) for msg in result.get(messages, []): if hasattr(msg, content) and msg.content: print(f{msg.type}: {msg.content}) break # 有中断等待用户决策 tool_name review_data.get(name, 未知工具) tool_args review_data.get(args, {}) print(f\n触发人工审核{tool_name}({tool_args})) decision input(请输入审核结果approve/reject).strip().lower() if decision approve: cmd Command(resume{decisions: [{type: approve}]}) elif decision reject: reason input(请输入拒绝原因).strip() or 人工审核未通过 cmd Command(resume{decisions: [{type: reject, message: reason}]}) else: print(输入无效操作已取消。) break # 恢复执行并获取新的结果继续循环检查 result agent2.invoke(cmd, configconfig) if __name__ __main__: test_main()注意点def print_stream_event(event): 从 stream event 中提取并打印工具调用和 AI 回复 for node_name, node_output in event.items(): # 过滤掉 None 或非字典的输出 if not isinstance(node_output, dict): continue messages node_output.get(messages, []) for msg in messages: # AI 文本回复 if getattr(msg, content, None): print(f {msg.content}) # 工具调用 if getattr(msg, tool_calls, None): for tc in msg.tool_calls: print(f 调用工具: {tc[name]}({tc[args]})) # 工具返回结果通常是 ToolMessage if getattr(msg, type, ) tool: print(f 工具结果: {msg.content}) if __name__ __main__: config {configurable: {thread_id: test-1}} for event in agent2.stream( {messages: [HumanMessage(content我的桌面放了几本小说)]}, configconfig ): if __interrupt__ in event: print_stream_event(event) print(中断需要审批, event[__interrupt__]) decision input(请输入 approve 或 reject: ).strip().lower() # 修复使用正确的决策格式 if decision approve: resume_value { decisions: [ { type: approve # 注意这里是approve不是approved } ] } else: # reject resume_value { decisions: [ { type: reject # 注意这里是reject不是rejected } ] } cmd Command(resumeresume_value) for e in agent2.stream(cmd, configconfig): print_stream_event(e) else: print_stream_event(event)如果智能体调用没有写成whileTrue循环并重新调用的形式则create_agent智能体则会在进行一次工具检查时停止继续执行