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

HumanInTheLoopMiddleware的意义

HumanInTheLoopMiddleware 是 LangChain 中用于在 Agent 执行过程中插入人工审核节点的中间件。它允许你在工具被实际调用前或调用后暂停流程，等待人工批准、修改或拒绝，是构建安全、可控智能体的核心组件。

HumanInTheLoopMiddleware的作用时机

HumanInTheLoopMiddleware 的默认介入时机是在工具实际执行之前。换句话说，当 Agent 决定调用某个工具并生成了工具调用的参数后，中间件会立刻中断流程，把“待审批的工具调用请求”交给你，此时工具还没真正执行。

在 create_agent 构建的图结构中，执行流程大致为：

模型节点（生成工具调用）→ 中断点 → 工具节点（执行工具）

代码实战

step1 配置模型API

from langchain_openai import ChatOpenAI import json with open(r"config/model_call.json", 'r', encoding='utf-8') as f: config = json.load(f) # ================= 模型初始化 ================= llm = ChatOpenAI( api_key=config["api_key"], base_url=config["base_url"], model=config["model"], temperature=0.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', encoding='utf-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', encoding='utf-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( model=llm, tools=[read_file, open_url, create_py_file, get_filenames_pathlib, copy_file, get_desktop_path], system_prompt=system_prompt, middleware=[hitl_middleware], checkpointer=checkpointer )

在创建create_agent之前，需要提前写好有关智能体的提示词部分，在提示词中澄清模型调用工具的规则

InMemorySaver 是 LangGraph（LangChain 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(r"config/model_call.json", 'r', encoding='utf-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', encoding='utf-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', encoding='utf-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_key=config["api_key"], base_url=config["base_url"], model=config["model"], temperature=0.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( model=llm, tools=[read_file, create_py_file, get_filenames_pathlib, get_desktop_path], system_prompt=system_prompt, middleware=[hitl_middleware], checkpointer=checkpointer ) 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}]}, config=config ) 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, config=config) 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="我的桌面放了几本小说？")]}, config=config ): 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(resume=resume_value) for e in agent2.stream(cmd, config=config): print_stream_event(e) else: print_stream_event(event)

如果智能体调用没有写成whileTrue循环并重新调用的形式，则create_agent智能体则会在进行一次工具检查时停止继续执行