PIIMiddleware 使用说明

护栏（Guardrails）体系中负责隐私合规这一层的确定性护栏，通过 middleware 机制在 agent 执行链路的关键节点（输入、输出、工具结果、流式传输）上自动拦截和处置 PII（Personally Identifiable Information），无需改动业务节点代码。

构造函数签名

PIIMiddleware(
    pii_type,            # 必填，positional
    *,                   # 以下均为 keyword-only
    strategy="redact",
    detector=None,
    apply_to_input=True,
    apply_to_output=False,
    apply_to_tool_results=False,
)

参数详解

pii_type（必填）— 要检测的 PII 类型。内置五种："email"、"credit_card"（Luhn 校验）、"ip"（IPv4/IPv6）、"mac_address"、"url"。也可传任意自定义字符串，但此时必须同时提供 detector，否则抛 ValueError。

strategy（默认 "redact"）— 检测到 PII 后的处理方式：

• "block" — 直接抛出 PIIDetectionError，终止整个 run
• "redact" — 替换为 [REDACTED_email] 这样的占位符
• "mask" — 部分遮蔽，保留末几位（如 ****-****-****-1234）
• "hash" — 替换为确定性哈希（如 <email_hash:a1b2c3d4>），同一值每次哈希结果相同，可用于分析

detector（默认 None）— 自定义检测器，两种形式：

• 传字符串 → 视为正则表达式
• 传 callable → 接受 str，返回 list[PIIMatch]
• 传 None → 使用内置检测器（仅内置 pii_type 有内置检测器）

apply_to_input（默认 True）— 是否在 before_model 钩子中检测用户最后一条 HumanMessage。

apply_to_output（默认 False）— 是否在 after_model 钩子中检测 AI 最后一条 AIMessage，同时启用流式 transformer 实时脱敏。

apply_to_tool_results（默认 False）— 是否在 before_model 钩子中检测最新一批 ToolMessage（即 AI 发出工具调用后返回的结果），同时启用流式 transformer 处理 tools 通道事件。

下面是参数选择决策图，以及完整的运行时工作流程图。

接下来是运行时的完整工作流程，展示各参数在 agent 生命周期中分别在哪个阶段介入。

典型用法速查

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

# 场景 1：只保护用户输入（最常见，也是默认行为）
agent = create_agent("openai:gpt-4o", middleware=[
    PIIMiddleware("email"),                        # apply_to_input=True 默认开启
    PIIMiddleware("credit_card", strategy="mask"), # 信用卡号末四位可见
])

# 场景 2：同时保护 AI 输出（防止模型"复读"用户的 PII）
agent = create_agent("openai:gpt-4o", middleware=[
    PIIMiddleware("email", apply_to_output=True),
])

# 场景 3：工具返回结果中包含 PII（如搜索结果、数据库查询）
agent = create_agent("openai:gpt-4o", middleware=[
    PIIMiddleware("ip", strategy="hash", apply_to_tool_results=True),
])

# 场景 4：自定义 PII 类型（API Key 等内部凭证）
agent = create_agent("openai:gpt-4o", middleware=[
    PIIMiddleware("api_key",
        detector=r"sk-[a-zA-Z0-9]{32,}",
        strategy="block"),              # 发现即终止，不允许泄露
])

# 场景 5：全面防护（输入 + 输出 + 工具结果 + 流式）
agent = create_agent("openai:gpt-4o", middleware=[
    PIIMiddleware("email",
        strategy="redact",
        apply_to_input=True,
        apply_to_output=True,
        apply_to_tool_results=True),
])

几个容易踩的坑

apply_to_output=True 时流式脱敏（_PIIStreamTransformer）才会启用，否则流式输出不受保护，只有最终 state 里的 AIMessage 会被处理。多个 PIIMiddleware 实例可以叠加，每种 PII 类型独立检测、独立策略，互不干扰。自定义 pii_type 不提供 detector 时构造期就会抛 ValueError，不是运行时才报错。