基于中间件组合的安全防护

什么是基于中间件组合的安全防护

核心定义

一套通过中间件组合构建的安全防护体系，嵌入 AI 应用与 Agent 执行流程，在关键节点做安全检查、内容过滤、风险拦截，提前阻止安全问题发生。

核心能力

• 检测敏感信息、违规内容、恶意请求
• 拦截提示词注入、PII 泄露、不当输出
• 验证输出合规性与质量
• 提前阻断风险，避免安全事件

常见防护场景

• 防 PII（个人信息）泄露
• 检测并阻止提示词注入攻击
• 拦截有害、违规内容
• 输出合规与质量校验

实现方式

• 内置中间件自动实现防护
• 开发者可自定义中间件，灵活扩展防护规则
• 部署在 Agent 执行前后、模型调用、工具调用等关键环节

一句话总结：用中间件做多层安全拦截，给 AI Agent 和应用穿上 “安全防护衣”。

安全防护的实现方式

AI 安全防护的两种实现方式：确定性安全防护和基于模型的安全防护：

对比维度	确定性安全防护	基于模型的安全防护
实现原理	预设规则、正则、关键词、黑名单	大模型（LLM）语义理解、内容判别
响应速度	快	慢
成本	低	高
可预测性	可预测、可追溯	不可控、不可预测
覆盖能力	规则内有效，易遗漏复杂场景	泛化性强，能识别规则外风险
优缺点	优点：快、稳、省；缺点：扩展性差、易绕过	优点：智能、灵活；缺点：延迟高、成本高
典型场景	关键词过滤、PII 检测、正则拦截	语义风险判断、复杂内容合规校验
工具支持	LangChain 内置中间件、规则引擎	LangChain 支持 LLM 分类、自定义判别

内置安全防护功能

PII检测

LangChain 提供PII 中间件，用于识别 / 处理文本中的个人身份信息（PII），如邮箱、电话、IP、证件号等，保护用户隐私，满足合规要求。

策略	作用	效果示例	适用场景
redact	检测到 PII 直接删除 / 替换	[REDACTED]	完全隐藏、不留痕迹
mask	部分掩码、保留部分字符	****1234	保留格式、不影响识别
hash	不可逆哈希、值不变	a1f5c6...	保留数据结构、不可还原
block	检测到直接拦截、报错	抛出异常	禁止含 PII 内容通过

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

# 定义工具列表（客服工具、邮箱工具）
tools = [customer_service_tool, email_tool]

# 创建Agent
agent = create_agent(
    tools=tools,
    # 中间件配置
    middleware=[
        # PII中间件：策略=redact，应用到输入
        PIIMiddleware(
            strategy="redact",
            apply_to_input=True
        ),
        # PII中间件：策略=mask，应用到输入
        PIIMiddleware(
            strategy="mask",
            apply_to_input=True
        ),
        # PII中间件：策略=block，拦截API Key
        PIIMiddleware(
            strategy="block",
            detector=r"sk-[a-zA-Z0-9]{20,32}",
            apply_to_input=True
        )
    ]
)

# 测试输入：包含邮箱、手机号
messages = [
    {
        "role": "user",
        "content": "My email is john.doe@example.com and card is 4111-1111-1111-1110"
    }
]

# 执行：中间件自动处理PII
response = agent.invoke(messages)
print(response)

输入含邮箱 / 手机号 / 证件号时，中间件按策略：

• 脱敏（mask/redact）
• 拦截（block 并报错）

实现PII 隐私保护。

Human-in-the-loop（人工介入）

LangChain 内置中间件，在执行敏感操作前强制人工审批，是高风险场景的核心安全防护手段。

适用场景

• 输入 / 输出包含敏感数据
• 执行可能产生破坏性操作
• 接收外部方通信请求
• 任何有重大业务影响的操作

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

# 工具列表：发邮件、删除数据库（敏感操作）
tools = [send_email_tool, delete_database_tool]

# 创建Agent
agent = create_agent(
    tools=tools,
    middleware=[
        # 人工介入中间件
        HumanInTheLoopMiddleware(
            # 对发邮件、数据库操作要求人工审批
            require_approval_for_sensitive_ops=True,
            send_email=True,
            delete_database=True
        )
    ],
    # 启用检查点持久化
    checkpointer=MemorySaver()
)

# 配置：必须指定 thread_id 用于持久化
config = {"configurable": {"thread_id": "user-123"}}

# 执行：敏感工具会暂停，等待人工 approval
result = agent.invoke(
    {"messages": [{"role": "user", "content": "发邮件到团队邮箱"}]},
    config=config
)

# 人工决策：批准/拒绝
# 恢复执行：传入 {"approved": true} 或 false
result = agent.invoke(
    Command(resume={"approved": "true"}),
    config=config
)

核心流程

1. 执行敏感操作前自动暂停
2. 人工审批（批准 / 拒绝）
3. 仅批准后才继续执行工具
4. 支持持久化会话（thread_id）

自定义安全防护措施

LangChain 支持自定义中间件，可在 Agent 运行的前、中、后任意阶段插入安全逻辑，实现定制化检测、过滤、拦截，满足个性化安全要求。

核心能力：

• 自定义钩子（hook）嵌入 Agent 执行流程
• 输入前检测、执行中监控、输出后校验
• 灵活拦截违规内容、注入风险、敏感信息
• 不依赖内置规则，完全自主控制防护策略

Agent执行前的安全防护

使用 before_agent 钩子在每次调用开始时进行一次验证，这对于会检查验证规则特别有用，例如快速控制在任何处理开始前阻止不适当的查询。

from typing import Any, Dict
from langgraph.runtime import import_ware
from before_agent_middleware import before_AgentState_Hook_config

# 定义 before_agent 钩子函数
def before_agent(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    对用户输入进行安全防护
    """
    # 过滤敏感词
    sensitive_keywords = ["赌博", "毒品", "暴力"]
    for keyword in sensitive_keywords:
        if keyword in state["messages"][-1].content:
            return {
                "messages": [
                    {"role": "assistant", "content": "内容不安全，无法回复"}
                ]
            }
    # 未命中敏感词，返回原状态
    return state

# 创建 Agent 时注册中间件
model.bind(
    search_tool={},
    middleware_content_filter=before_agent
)

代码讲解

1. 导入模块
   • 导入类型注解工具 typing
   • 导入 LangGraph 运行时相关模块
   • 导入自定义中间件配置
2. 钩子函数 before_agent
   • 接收状态字典 state，处理后返回新状态
   • 核心功能：敏感词过滤
   • 遍历敏感词列表，检测用户最后一条输入
   • 命中敏感词 → 返回安全提示
   • 未命中 → 原样返回状态
3. 注册中间件
   • 将 before_agent 作为内容过滤器绑定到模型
   • 实现 Agent 执行前的安全拦截

Agent执行后的安全防护

使用 after_agent 钩子在 Agent 完成执行后验证结果，这对于 Agent 的输出是否符合安全、规范、质量要求非常有用。

from typing import Any
from langchain_agents.middleware import after_agent, AgentState
from langgraph.runtime import Runtime

# 定义 after_agent 钩子函数
def after_agent_verification(state: AgentState, runtime: Runtime) -> AgentState:
    """Agent 输出内容验证"""
    agent_output = state["messages"][-1].content
    # 检查是否为"我不知道"
    if "我不清楚" in agent_output:
        # 修改状态消息，增加更多内容，使其更具体
        state["messages"][-1].content = agent_output + "\n\n请提供更多内容，以便我更准确地回答"
    # 检查输出消息长度是否不足
    if len(state["messages"]) < 1:
        # 追加系统消息提示输出不足
        state["messages"].append(
            {"role": "system", "content": "输出内容不足，可能不准确，请完善输出"}
        )
    return state

# 创建带验证的 Agent
agent = create_agent(
    model,
    tools=[search_tool],
    middleware_output_verification=after_agent_verification,
)

代码解读

1. 功能定位
   • 在 Agent 执行完成后 对输出结果进行安全与质量校验。
   • 用于确保输出合规、完整、准确。
2. 核心函数 after_agent_verification
   • 参数：接收当前状态 state 和运行时 runtime。
   • 获取输出：提取最后一条消息内容 agent_output。
   • 检查 “我不知道” 类表述：若包含 “我不清楚”，则补充提示，让回答更具体。
   • 检查消息长度：若消息数量不足，追加系统提示，说明输出可能不准确。
   • 返回：处理后的新状态。
3. Agent 注册
   • 创建 Agent 时，将输出验证中间件 middleware_output_verification 绑定到模型，实现后置安全校验。

组合多个中间件实现安全防护措施

你可以通过多个安全防护中间件添加到 middleware 数组中堆叠它们，按顺序执行。请你构建分层防护：

from langchain.agents import create_agent
from langchain.agents.middleware import PiIMiddleware, HumanInTheLoopMiddleware

agent = create_agent(
    model,
    tools=[search_tool, send_email_tool],
    middleware=[
        # Layer 1: Deterministic input filter before Agent
        deterministic_input_filter,
        ContentFilterMiddleware(banned_keywords=["hack", "exploit"]),
        # Layer 2: PII Protection before and after model
        PiIMiddleware(email, strategy="redact", apply_to_input=True),
        PiIMiddleware(email, strategy="sensitive", apply_to_output=True),
        HumanInTheLoopMiddleware(interrupt_on="send_email", True),
        # Layer 4: Model based safety check after Agent
        SafetyGuardrailMiddleware(check_agent_output=True),
    ],
)

核心原理

这是 LangGraph / LangChain 智能体的多层安全防护架构：

• 把多个中间件放进 middleware 列表
• 执行时从上到下按顺序依次运行
• 实现输入→处理→输出全流程防护

整体执行流程

1. 用户输入 → 第一层：关键词拦截
2. 输入 → 第二层：隐私脱敏
3. 执行工具 → 第三层：发邮件必须人工审核
4. 输出结果 → 第四层：内容安全校验
5. 最终返回给用户

安全防护最佳实践

AI 智能体安全防护的最佳实践，核心是分层防御、多维度防护、可观测、可更新四大原则，确保系统安全可控。

1. 分层防御（Defense in Depth）

   • 多环节、多阶段防护：输入、模型、输出、工具调用全链路拦截。
   • 层层设防，降低单一漏洞导致的风险。

2. 多维度安全防护

   • 内容安全：过滤违规、敏感、有害内容。
   • 隐私保护：PII 脱敏、数据加密、权限管控。
   • 滥用防护：限制高频调用、防止资源耗尽。
   • 身份验证：严格鉴权，防止未授权访问。

3. 用户可控

   • 开放配置开关，允许用户根据场景调整安全策略。
   • 关键操作（如发邮件、调用工具）需要人工确认。

4. 自动化与监控

   • 持续监控输入输出、日志、告警。
   • 及时发现异常、攻击、数据泄露。

5. 可观测

   • 完整记录请求、响应、拦截、错误。
   • 可追溯、可审计、可复盘安全事件。

6. 定期更新

   • 模型、规则、防护策略持续迭代。
   • 应对新型攻击、新风险，保持防护有效性。

一句话总结

这套实践强调全方位、多层次、自动化、可监控、可更新的安全体系，从输入到输出全链路防护，确保 AI 智能体安全、可控、合规。

总结

核心是：用中间件构建可定制、分层、可控的 AI 安全体系，在保障安全的同时不牺牲体验。

关键能力总结

• 丰富安全功能：包括 PII 识别、人工审核、输入 / 输出过滤、内容安全等，快速落地安全防护。

• 自定义安全策略：可根据业务需求定制规则、拦截逻辑、审核流程，适配不同场景。

• Agent 全流程防护： 在 Agent 执行的关键节点（输入、调用、输出）实现安全控制与内容过滤。

• 平衡安全与体验：保证安全性的同时，保持良好用户体验与功能能力。

最终目标

通过完整的安全防护体系，降低 AI 风险，同时维持用户体验与功能可用性，强调安全是持续迭代的过程。