去年双十一大促期间，我负责的电商 AI 客服系统遭遇了一次惊心动魄的事故——上游日志系统无意中将用户手机号、身份证号明文传入了 Prompt，险些触发数据合规红线。这个教训让我彻底认识到：在调用任何 AI API 之前，PII（个人身份信息）脱敏预处理是生死线级别的必修课。

本文将从电商促销高并发场景切入，带你完整构建一套可复用的 PII 检测与掩码体系。我会展示用 HolySheep AI 立即注册 作为后端调用的完整方案，涵盖正则匹配、NER 实体识别、API 网关层防护三层机制。

为什么 AI API 调用必须做数据脱敏

很多开发者以为「我把数据发给 AI 服务商，处理完再返回就安全了」。这是一个致命误区：

HolySheep AI 的优势在于国内直连延迟 <50ms，但无论延迟多低，脱敏这一环都不能省——因为真正消耗费用的不是请求时间，而是 token 数量。

常见 PII 类型与检测策略

根据国内法规和实际业务场景，需要重点关注的 PII 类型包括：

类型正则示例风险等级
手机号1[3-9]\d{9}高
身份证号\d{17}[\dXx]极高
银行卡号[1-9]\d{12,18}极高
邮箱地址[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}中
姓名NER 识别中
家庭住址NER 识别高

实战：构建 PII 检测与掩码工具类

方案一：基于正则的快速检测

这种方式适合对延迟敏感的场景，检测速度极快。我在使用 HolySheep AI 时，实测正则扫描 1MB 文本仅需 ~8ms：

"""
PII 检测与掩码工具类
适配 HolySheep AI API v1
"""
import re
from typing import Dict, List, Tuple
from dataclasses import dataclass
from enum import Enum

class PIIType(Enum):
    PHONE = "phone"
    ID_CARD = "id_card"
    BANK_CARD = "bank_card"
    EMAIL = "email"
    ADDRESS = "address"

@dataclass
class PIIDetection:
    pii_type: PIIType
    original: str
    masked: str
    start_pos: int
    end_pos: int

class PIIDetector:
    """PII 检测与掩码核心类"""
    
    # 编译正则表达式提升性能
    PATTERNS: Dict[PIIType, re.Pattern] = {
        PIIType.PHONE: re.compile(
            r'1[3-9]\d{9}', re.IGNORECASE
        ),
        PIIType.ID_CARD: re.compile(
            r'\b([1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx])\b'
        ),
        PIIType.BANK_CARD: re.compile(
            r'\b([1-9]\d{12,18})\b'
        ),
        PIIType.EMAIL: re.compile(
            r'\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b'
        ),
    }
    
    # 掩码策略
    MASK_CONFIG: Dict[PIIType, str] = {
        PIIType.PHONE: lambda m: f"{m[:3]}****{m[-4:]}",
        PIIType.ID_CARD: lambda m: f"{m[:6]}********{m[-4:]}",
        PIIType.BANK_CARD: lambda m: f"****{m[-4:]}",
        PIIType.EMAIL: lambda m: f"{m.split('@')[0][:2]}***@{m.split('@')[1]}",
    }
    
    def detect(self, text: str) -> List[PIIDetection]:
        """检测文本中的所有 PII"""
        detections = []
        for pii_type, pattern in self.PATTERNS.items():
            for match in pattern.finditer(text):
                original = match.group()
                masked = self._apply_mask(original, pii_type)
                detections.append(PIIDetection(
                    pii_type=pii_type,
                    original=original,
                    masked=masked,
                    start_pos=match.start(),
                    end_pos=match.end()
                ))
        # 按位置排序
        return sorted(detections, key=lambda x: x.start_pos)
    
    def mask(self, text: str) -> Tuple[str, List[PIIDetection]]:
        """脱敏并返回处理后的文本"""
        detections = self.detect(text)
        masked_text = text
        offset = 0
        for d in detections:
            original_len = len(d.original)
            masked_len = len(d.masked)
            pos = d.start_pos + offset
            masked_text = (
                masked_text[:pos] 
                + d.masked 
                + masked_text[pos + original_len:]
            )
            offset += masked_len - original_len
        return masked_text, detections
    
    def _apply_mask(self, original: str, pii_type: PIIType) -> str:
        """应用掩码策略"""
        if pii_type in self.MASK_CONFIG:
            return self.MASK_CONFIG[pii_type](original)
        return "***"


使用示例

if __name__ == "__main__":
    detector = PIIDetector()
    
    sample_text = """
    用户王伟，手机号 13812345678，
    身份证号 11010119900307451X，
    邮箱 [email protected]，
    银行卡 6222021234567890123
    """
    
    masked_text, detections = detector.mask(sample_text)
    
    print("脱敏后文本:")
    print(masked_text)
    print("\n检测结果:")
    for d in detections:
        print(f"  [{d.pii_type.value}] {d.original} -> {d.masked}")

方案二：集成 HolySheep AI NER 能力识别姓名和地址

正则能匹配结构化数据，但对姓名、地址等非结构化文本力不从心。我通过 HolySheep AI 的 GPT-4o-mini 接口做 NER 抽取，实测成本仅 $0.15/MToken，比 Claude Haiku 还便宜 40%：

"""
集成 HolySheep AI 的 NER 能力识别复杂 PII
"""
import os
import json
import httpx
from typing import List, Dict, Any

class HolySheepAIPIIRecognizer:
    """调用 HolySheep AI 进行 PII NER 识别"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.client = httpx.Client(
            base_url=self.base_url,
            timeout=30.0,
            headers={"Authorization": f"Bearer {api_key}"}
        )
    
    def extract_pii_ner(self, text: str) -> List[Dict[str, Any]]:
        """
        使用 LLM 提取文本中的 PII 实体
        返回格式: [{"type": "PERSON", "value": "张三", "start": 0, "end": 2}, ...]
        """
        system_prompt = """你是一个 PII 识别专家。请从文本中提取以下类型的敏感信息：
        - PERSON: 人名
        - ADDRESS: 详细地址（包含省市区街道门牌号）
        - PHONE: 电话号码（如已脱敏则跳过）
        - ID_CARD: 身份证号（如已脱敏则跳过）
        
        仅返回 JSON 数组格式，示例：
        [{"type": "PERSON", "value": "王伟", "start": 0, "end": 2}]
        
        如果没有发现 PII，返回空数组 []。"""
        
        response = self.client.post(
            "/chat/completions",
            json={
                "model": "gpt-4o-mini",  # $0.15/MTok，高性价比选择
                "messages": [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": text}
                ],
                "temperature": 0.1,  # 低温度保证稳定性
                "max_tokens": 500
            }
        )
        
        if response.status_code != 200:
            raise Exception(f"HolySheep AI API 错误: {response.status_code} - {response.text}")
        
        result = response.json()
        content = result["choices"][0]["message"]["content"]
        
        # 解析 JSON 返回
        try:
            return json.loads(content)
        except json.JSONDecodeError:
            # 尝试清理 markdown 代码块
            if "```json" in content:
                content = content.split("``json")[1].split("``")[0]
            elif "```" in content:
                content = content.split("``")[1].split("``")[0]
            return json.loads(content)


使用示例

if __name__ == "__main__":
    api_key = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
    recognizer = HolySheepAIPIIRecognizer(api_key)
    
    text = "李明住在北京市朝阳区建国路88号，我的手机号已经脱敏为 138****1234"
    
    pii_entities = recognizer.extract_pii_ner(text)
    print("NER 识别结果:")
    for entity in pii_entities:
        print(f"  {entity['type']}: {entity['value']} (位置: {entity['start']}-{entity['end']})")

方案三：构建完整的预处理管道

实际生产环境中，我建议采用「正则快速扫描 + LLM NER 补充」的双层架构：

"""
PII 预处理完整管道
整合正则检测 + HolySheep AI NER + 审计日志
"""
from typing import Tuple, Optional
import logging
from datetime import datetime
import json

class PIIPreprocessor:
    """PII 预处理管道"""
    
    def __init__(self, holysheep_api_key: str):
        self.regex_detector = PIIDetector()
        self.ner_recognizer = HolySheepAIPIIRecognizer(holysheep_api_key)
        self.logger = logging.getLogger(__name__)
    
    def preprocess(self, text: str, enable_ner: bool = True) -> Tuple[str, dict]:
        """
        完整的 PII 预处理流程
        
        Returns:
            (脱敏后文本, 处理报告)
        """
        report = {
            "timestamp": datetime.now().isoformat(),
            "original_length": len(text),
            "pii_found": [],
            "processing_time_ms": 0
        }
        
        import time
        start_time = time.time()
        
        # 第一层：正则快速扫描
        masked_text, regex_detections = self.regex_detector.mask(text)
        
        for d in regex_detections:
            report["pii_found"].append({
                "type": d.pii_type.value,
                "original": d.original,
                "masked": d.masked,
                "method": "regex"
            })
        
        # 第二层：NER 补充识别（可选）
        if enable_ner:
            try:
                ner_entities = self.ner_recognizer.extract_pii_ner(masked_text)
                # 手动替换 NER 识别的实体
                for entity in ner_entities:
                    entity_type = entity["type"]
                    entity_value = entity["value"]
                    entity_start = entity["start"]
                    entity_end = entity["end"]
                    
                    # 生成掩码
                    if entity_type == "PERSON":
                        masked_value = entity_value[0] + "*" * (len(entity_value) - 1)
                    elif entity_type == "ADDRESS":
                        masked_value = "**[详细地址已脱敏]**"
                    else:
                        masked_value = "***"
                    
                    # 直接替换（简单实现，生产环境需更精确的偏移计算）
                    if masked_value != entity_value:
                        masked_text = masked_text.replace(entity_value, masked_value, 1)
                        report["pii_found"].append({
                            "type": entity_type,
                            "original": entity_value,
                            "masked": masked_value,
                            "method": "ner"
                        })
            except Exception as e:
                self.logger.warning(f"NER 处理失败，降级为正则模式: {e}")
        
        report["processing_time_ms"] = round((time.time() - start_time) * 1000, 2)
        report["pii_count"] = len(report["pii_found"])
        
        return masked_text, report
    
    def preprocess_for_ai_request(
        self, 
        user_input: str, 
        context: Optional[str] = None
    ) -> dict:
        """
        为 AI API 调用准备安全的输入
        
        Returns:
            {
                "safe_input": 脱敏后的用户输入,
                "report": 处理报告,
                "should_block": 是否应阻止请求
            }
        """
        # 合并上下文
        full_text = f"{context}\n{user_input}" if context else user_input
        
        safe_text, report = self.preprocess(full_text)
        
        # 高风险 PII 检查
        high_risk_types = {"id_card", "bank_card"}
        high_risk_found = [
            p for p in report["pii_found"] 
            if p["type"] in high_risk_types and p["method"] == "regex"
        ]
        
        return {
            "safe_input": safe_text,
            "report": report,
            "should_block": len(high_risk_found) > 0 and report["pii_count"] > 5
        }


使用示例：与 HolySheep AI 对接

def chat_with_holysheep(user_message: str, api_key: str):
    """安全的 AI 对话示例"""
    
    preprocessor = PIIPreprocessor(api_key)
    
    # 预处理用户输入
    result = preprocessor.preprocess_for_ai_request(
        user_input=user_message,
        context="你是电商客服小助手，请用友好语气回复用户问题。"
    )
    
    if result["should_block"]:
        return {
            "error": "输入包含过多敏感信息，请通过官方客服渠道处理。",
            "code": "PII_BLOCKED"
        }
    
    # 调用 HolySheep AI
    client = httpx.Client(
        base_url="https://api.holysheep.ai/v1",
        timeout=30.0,
        headers={"Authorization": f"Bearer {api_key}"}
    )
    
    response = client.post(
        "/chat/completions",
        json={
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": "你是电商客服小助手。"},
                {"role": "user", "content": result["safe_input"]}
            ],
            "max_tokens": 500
        }
    )
    
    if response.status_code == 200:
        return {
            "reply": response.json()["choices"][0]["message"]["content"],
            "processing_report": result["report"]
        }
    else:
        return {"error": f"API 错误: {response.status_code}"}


运行示例

if __name__ == "__main__":
    import os
    api_key = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
    
    # 测试各种 PII 场景
    test_cases = [
        "我的订单号是 ORD20240101，收货人张三，电话 13912345678",
        "请帮我查询 11010119900307451X 身份证的积分",
        "我要退货，地址是上海市浦东新区世纪大道100号"
    ]
    
    for user_msg in test_cases:
        print(f"\n用户输入: {user_msg}")
        result = chat_with_holysheep(user_msg, api_key)
        if "error" in result:
            print(f"  -> {result['error']}")
        else:
            print(f"  -> AI 回复: {result['reply'][:50]}...")
            print(f"  -> PII 报告: 检测到 {result['processing_report']['pii_count']} 处敏感信息")

生产环境部署建议

在我司的电商系统中，这套方案经历了双十一 QPS 5000+ 的实战验证，以下是关键配置：

常见报错排查

错误 1：正则匹配消耗过多 CPU

症状：大文本（>100KB）检测时 CPU 占用率飙升至 100%，响应时间 >1s

原因：未编译的正则表达式每次匹配都会重新编译，O(n) 复杂度

解决：在类初始化时预编译所有正则模式：

# 错误写法 - 每次调用都编译
pattern = re.findall(r'1[3-9]\d{9}', text)


正确写法 - 类加载时编译一次

class PIIDetector:
    PATTERNS = {
        PIIType.PHONE: re.compile(r'1[3-9]\d{9}', re.IGNORECASE)
    }
    
    def detect(self, text):
        for pii_type, pattern in self.PATTERNS.items():
            pattern.findall(text)  # 使用预编译的模式

错误 2：NER 返回格式解析失败

症状：json.JSONDecodeError: Expecting value

原因：HolySheep AI 返回的 content 包含 markdown 代码块包裹

解决：增强 JSON 解析容错：

def safe_parse_json(content: str) -> list:
    """安全解析 JSON，处理各种异常格式"""
    content = content.strip()
    
    # 移除 markdown 代码块
    if content.startswith("```"):
        content = content.split("```")[1]
        if content.startswith("json"):
            content = content[4:]
        content = content.strip()
    
    # 移除首尾空白和引号
    content = content.strip('"\'')
    
    try:
        return json.loads(content)
    except json.JSONDecodeError:
        # 尝试修复常见问题
        content = content.replace("'", '"')  # 单引号转双引号
        content = content.replace("None", "null")  # None 转 null
        return json.loads(content)

错误 3：高频调用触发速率限制

症状：429 Too Many Requests

原因：NER 调用过于频繁，触发 HolySheep AI 的 QPS 限制

<

相关资源

相关文章