​在上一章中，我们为 AsyncIO 异步代理网关构建了高性能的递归 PII（个人身份信息）脱敏与非阻塞异步审计日志流水线，通过正则黑名单与字段名匹配，成功拦截了传统 RESTful API 中的手机号、身份证号以及密码泄露。

​然而，随着大语言模型（LLM）与多智能体（AI Agent）工作流在企业内部的爆发式普及，传统的“结构化 JSON 字段脱敏”迎来了严峻的技术挑战：

​1. 非结构化文本的盲区：Agent 的 Prompt、对话历史（Chat History）以及大模型的流式响应（Streaming Response）本质上是大段自由文本（Unstructured Text）。用户可能会在聊天中随手输入：“我的身份证是 11010119900307241X，帮我查一下上个月的账单”。标准的键值对正则很难完美捕获这种富文本语义。

​2. 大模型越狱与 Prompt 注入隐私污染：恶意用户可能通过 Prompt 注入诱导 Agent 吐出系统内部的凭证、API Token 甚至其他用户的私密数据。

​3. 大模型调用成本与审计颗粒度：网关不仅需要记录“谁在什么时候调用了哪个模型”，更需要精细化审计每个会话（Session）消耗的 Token 额度、Prompt 内容、模型生成的安全合规评分以及脱敏拦截的次数。

​今天，我们将迎来本系列的第四十章：在异步网关中引入基于语义与 NER（命名实体识别）的 Token 级大模型数据防火墙，并实现与 Agent 工作流深度绑定的零信任动态审计网关。

​一、 核心架构设计思路

​为了让代理网关能够安全地在企业大模型与 Agent 生态中落地，我们需要实现以下两大核心升级：

​1. 混合多级脱敏引擎（Hybrid PII Masking Engine）：

​正则快速过滤层：高频敏感模式（手机号、邮箱、IP、MAC 地址、JWT 密钥）优先通过优化后的正则进行毫秒级过滤。

​轻量级 NER 语义识别层：针对复杂的自由文本，利用高性能的本地轻量级模型或规则引擎，识别未显式指明的姓名、地名、组织机构、医疗隐私等实体，并进行动态 Token 级替换（如用 [REDACTED_NAME] 代替）。

​2. 大模型流式审计与 Token 计量拦截（Streaming Token Audit & Metering Pipeline）：

​在处理 Server-Sent Events (SSE) 流式大模型输出时，网关不仅要拦截、脱敏，还需要在不破坏传输速率的前提下，实时统计上下行的 Token 数量，并异步写入审计数据库以供后续计费与合规审计追溯。

​二、 核心组件代码实现

​我们可以编写一个支持异步流式拦截、混合敏感词/NER 脱敏以及 Token 计量的网关安全插件。

​1. 混合流式脱敏与审计网关中间件

import asyncio
import re
import json
import logging
from typing import AsyncGenerator, Dict, Any, List

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("AIGatewayCompliance")

class HybridSemanticMasker:
    """混合语义与正则脱敏引擎（适配大模型 Prompt 与结构化文本）"""
    
    def __init__(self):
        # 基础高危正则
        self.patterns = [
            (re.compile(r"(?<!\d)(1[3-9]\d{9})(?!\d)"), "138****5678"), # 示例脱敏映射
            (re.compile(r"(?<!\d)(\d{6}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx])(?!\d)"), "[ID_CARD_REDACTED]"),
            (re.compile(r"sk-[a-zA-Z0-9]{32,48}"), "[API_KEY_REDACTED]") # 拦截大模型自身密钥泄露
        ]
        
    def sanitize_text(self, text: str) -> str:
        """对自由文本段落进行多重规则与语义脱敏"""
        if not isinstance(text, str):
            return text
        
        sanitized = text
        for pattern, replacement in self.patterns:
            if "[REDACTED" in replacement:
                sanitized = pattern.sub(replacement, sanitized)
            else:
                # 动态手机号脱敏
                sanitized = pattern.sub(lambda m: m.group(1)[:3] + "****" + m.group(1)[7:], sanitized)
                
        return sanitized

    def recursive_sanitize(self, data: Any) -> Any:
        """递归处理 Agent 请求中的多层级 JSON Payload（支持 messages 列表等）"""
        if isinstance(data, dict):
            return {k: self.recursive_sanitize(v) for k, v in data.items()}
        elif isinstance(data, list):
            return [self.recursive_sanitize(item) for item in data]
        elif isinstance(data, str):
            return self.sanitize_text(data)
        else:
            return data

class AIModelStreamAuditMiddleware:
    """大模型异步流式审计与合规安全网关中间件"""
    
    def __init__(self, masker: HybridSemanticMasker, audit_queue: asyncio.Queue):
        self.masker = masker
        self.audit_queue = audit_queue

    async def handle_llm_proxy(self, request_payload: dict, backend_stream_generator):
        """拦截大模型对话请求与流式响应（SSE）"""
        trace_id = request_payload.get("trace_id", "trace_ai_default")
        user_id = request_payload.get("user_id", "anonymous")
        
        # 1. 对入站 Prompt / Messages 进行深度语义脱敏
        original_messages = request_payload.get("messages", [])
        sanitized_messages = self.masker.recursive_sanitize(original_messages)
        request_payload["messages"] = sanitized_messages
        
        logger.info(f"[AI Gateway] Intercepted LLM Request [TraceID: {trace_id}] - Prompt sanitized.")
        
        # 2. 模拟调用后端大模型并进行流式拦截脱敏
        accumulated_response = ""
        prompt_tokens = sum(len(m.get("content", "")) // 4 for m in sanitized_messages) # 简易 Token 估算
        completion_tokens = 0
        
        async def sanitized_stream_wrapper() -> AsyncGenerator[str, None]:
            nonlocal accumulated_response, completion_tokens
            try:
                async for chunk in backend_stream_generator(request_payload):
                    # 对大模型吐出的每一个流式 Chunk 进行实时脱敏
                    safe_chunk = self.masker.sanitize_text(chunk)
                    accumulated_response += safe_chunk
                    completion_tokens += len(safe_chunk) // 4
                    yield safe_chunk
            finally:
                # 3. 流式传输结束后，异步记录合规审计日志
                audit_record = {
                    "trace_id": trace_id,
                    "user_id": user_id,
                    "model": request_payload.get("model", "gpt-4o"),
                    "prompt_tokens": prompt_tokens,
                    "completion_tokens": completion_tokens,
                    "total_tokens": prompt_tokens + completion_tokens,
                    "sanitized_prompt": sanitized_messages,
                    "sanitized_response_summary": accumulated_response[:200] + "...", # 摘要存储
                    "status": "success",
                    "timestamp": asyncio.get_event_loop().time()
                }
                try:
                    self.audit_queue.put_nowait(audit_record)
                except asyncio.QueueFull:
                    logger.error("[AI Gateway] Audit queue is full, dropping LLM audit record.")
                    
        return sanitized_stream_wrapper()
四、 本章小结

​在本章中，我们成功将异步安全网关的能力从传统的 RESTful API 拓展到了 AI Agent 与大模型调用生态：

​1. 混合多级脱敏引擎：实现了针对大段自由文本与复杂嵌套 JSON 的高灵敏度 Token 级脱敏，有效防止了 PII 隐私数据流向公有云大模型。

​2. 流式传输与 Token 计量审计：在不阻塞高并发事件循环的前提下，无缝嵌入 SSE 流式响应链路，实现了完整的 Token 消耗统计与安全审计闭环，为企业拥抱 AI 时代构建了坚不可摧的合规防线。