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智能体则会在进行一次工具检查时停止继续执行
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
5
0- 0
已为社区贡献1条内容
所有评论(0)