🎯 课程主题

PIIMiddleware（个人身份信息中间件）——在模型调用前检测并保护敏感数据（邮箱、信用卡号、IP 等），防止信息泄露。

📝 核心知识点

1. PII（个人身份信息）的检测类型

  • 概念说明：PIIMiddleware 内置了对多种敏感数据类型的检测能力，通过 pii_type 参数指定。
  • 关键细节：支持的检测类型包括 email（邮箱）、credit_card（信用卡号）、url（网址）、mac_address（MAC 地址）、ip_address（IP 地址）以及 api_key（API 密钥）等。

2. 四种保护策略（strategy）

  • 概念说明：检测到敏感信息后，以不同策略替换或处理。
  • 关键细节：
    • redact：完全替换为描述性占位字符串，如邮箱被替换为 [REDACTED EMAIL]，适合日志清洗和合规输出。
    • mask：部分遮盖，保留最后几位，中间用 *** 替代，适合前端用户界面显示。
    • hash：对敏感值做单向哈希计算，同一输入始终产生相同哈希值，适合匿名统计分析和追踪。
    • block：直接抛出异常，阻断含敏感信息的请求，适合对隐私零容忍的场景。

3. 检测时机参数

  • 概念说明：通过布尔参数控制检测发生的节点。
  • 关键细节：
    • apply_to_input（默认 True）：模型调用前检测输入消息。
    • apply_to_output（默认 False）：模型调用后检测输出。
    • apply_to_tool_output（默认 False）：工具调用后检测工具输出。
    • 通常只开启 apply_to_input=True，目的是防止敏感信息发送给大模型，后续环节无需重复检测。

4. 自定义检测器（Custom Detector）

  • 概念说明：内置检测器覆盖有限，可通过自定义正则函数扩展检测能力。
  • 关键细节：
    • 编写一个生成器函数，接收 content（文本），yield 返回形如 {"text": matched_str, "start": start_idx, "end": end_idx} 的字典。
    • 示例：自定义手机号检测器（11 位数字正则）、自定义 API Key 检测器（sk- 开头的字母数字串）。
    • 在 PIIMiddleware 参数 detector 中传入自定义函数。

🏗️ 架构与工作流

用户消息 → apply_to_input 检测 → 策略替换 → 清洁后的消息 → 调用模型 → 模型回复

Middleware 位于 Agent 的输入管道中，在消息进入模型前对消息内容做正则扫描和替换。block 策略会在检测到敏感信息时直接抛异常中断流程；其余策略则就地替换后继续执行。

💻 代码实战

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain_core.messages import HumanMessage
import re

# 模型初始化（略）

# === 举例一：使用内置检测器 ===
agent = create_agent(
    model,
    middleware=[
        PIIMiddleware(
            pii_type="email", strategy="redact", apply_to_input=True
        ),
        PIIMiddleware(
            pii_type="credit_card", strategy="mask", apply_to_input=True
        ),
        PIIMiddleware(
            pii_type="url", strategy="hash", apply_to_input=True
        ),
        PIIMiddleware(
            pii_type="mac_address", strategy="mask", apply_to_input=True
        ),
        PIIMiddleware(
            pii_type="ip_address", strategy="block", apply_to_input=True
        ),
    ],
)

response = agent.invoke({
    "messages": [
        HumanMessage(content="""
        请发邮件到 test@example.com，
        我的银行卡号是 4111111111111111，
        参考网址 https://example.com/doc，
        MAC 地址是 00:1B:44:11:3A:B7。
        """)
    ]
})

# === 举例二：自定义检测器（手机号 + API Key） ===
def phone_number_detector(content: str):
    pattern = re.compile(r"1[3-9]\d{9}")
    for match in pattern.finditer(content):
        yield {
            "text": match.group(),
            "start": match.start(),
            "end": match.end(),
        }

def api_key_detector(content: str):
    pattern = re.compile(r"sk-[a-zA-Z0-9]+")
    for match in pattern.finditer(content):
        yield {
            "text": match.group(),
            "start": match.start(),
            "end": match.end(),
        }

agent2 = create_agent(
    model,
    middleware=[
        PIIMiddleware(
            detector=phone_number_detector,
            strategy="mask",
            apply_to_input=True,
        ),
        PIIMiddleware(
            detector=api_key_detector,
            strategy="hash",
            apply_to_input=True,
        ),
    ],
)

response2 = agent2.invoke({
    "messages": [
        HumanMessage(content="""
        我的手机号是 13812345678，API Key 是 sk-abc123XYZ，
        另外这个链接 https://example.com（未配置检测，所以不会被替换）。
        """)
    ]
})

# === 举例三：block 策略触发异常处理 ===
try:
    response3 = agent.invoke({
        "messages": [HumanMessage(content="我的 IP 是 192.168.1.1")]
    })
except Exception as e:
    print(f"检测到 IP 抛出异常: {e}")

⚠️ 常见问题与避坑指南

  • 未配置对应 pii_type 的敏感信息不会被检测到，原样传递给模型（如举例二中 URL 未被替换）。
  • block 策略会直接抛异常，生产环境需用 try/except 包裹或换成其他策略。
  • 自定义检测器的函数必须是生成器（generator），通过 yield 而非 return 返回结果。
  • 多个 PIIMiddleware 实例需以列表形式传入 middleware 参数，类型和策略可按需组合。

💡 个人总结与延伸

PIIMiddleware 是 LangChain 提供的数据安全防护层，核心思路是「在数据进模型之前完成脱敏」。四种策略覆盖了从完全抹除（redact）到硬阻断（block）的完整梯度，实际生产建议结合业务合规要求选型——日志审计用 redact，用户界面用 mask，统计分析用 hash，高危场景用 block。

最近更新:
Contributors: yangwenguang