AI 数据隐私治理：合规、脱敏与全链路防护

AI 需要数据 → 数据包含隐私 → 隐私需要保护
     ↑                              ↓
     └──── 如何破局？←──────────────┘

用户输入 → Agent 处理 → 工具调用 → LLM 推理 → 结果输出
  ↑           ↑           ↑           ↑           ↑
风险1       风险2       风险3       风险4       风险5
敏感信息    上下文泄露  数据过度    模型记忆    输出泄露
            到第三方    访问        训练数据    给未授权方

class InputPrivacyFilter:
    """输入数据自动脱敏"""

    PII_PATTERNS = {
        "phone": r"1[3-9]\d{9}",
        "email": r"[\w.-]+@[\w.-]+\.\w+",
        "id_card": r"\d{17}[\dXx]",
        "bank_card": r"\d{16,19}",
        "address": r"[\u4e00-\u9fa5]{2,}(省|市|区|县|路|街|号)",
    }

    def sanitize(self, text: str) -> str:
        for pii_type, pattern in self.PII_PATTERNS.items():
            text = re.sub(pattern, f"[{pii_type}]", text)
        return text

    def mask_sensitive(self, text: str, user_consent: dict) -> str:
        """根据用户同意级别脱敏"""
        if not user_consent.get("share_location"):
            text = self._remove_location(text)
        if not user_consent.get("share_contact"):
            text = self._remove_contact(text)
        return text

# 使用
filter = InputPrivacyFilter()
safe_input = filter.sanitize("我的手机是13812345678，邮箱是test@qq.com")
# 结果: "我的手机是[phone]，邮箱是[email]"

class ContextIsolator:
    """防止用户间上下文泄露"""

    def __init__(self):
        self.user_contexts = {}  # 用户隔离

    def get_context(self, user_id: str) -> dict:
        """每个用户独立上下文"""
        if user_id not in self.user_contexts:
            self.user_contexts[user_id] = {
                "history": [],
                "preferences": {},
                "knowledge": [],
            }
        return self.user_contexts[user_id]

    def clear_context(self, user_id: str):
        """用户请求删除时，清除所有数据（被遗忘权）"""
        del self.user_contexts[user_id]
        # 同时清除向量数据库中的用户数据
        vector_db.delete(filter={"user_id": user_id})
        # 清除日志
        logs.delete(filter={"user_id": user_id})

class ToolAccessControl:
    """工具调用的数据访问控制"""

    def check_access(self, user_id: str, tool: str, args: dict) -> bool:
        # 1. 检查用户权限
        user_role = self.get_user_role(user_id)

        # 2. 检查工具数据范围
        if tool == "query_db":
            # 确保只查询用户自己的数据
            if "user_id" not in args.get("filter", {}):
                args["filter"]["user_id"] = user_id
            return True

        elif tool == "send_email":
            # 检查收件人是否在授权列表
            allowed = self.get_allowed_recipients(user_id)
            if args["to"] not in allowed:
                return False
            return True

        elif tool == "read_file":
            # 检查文件路径是否在用户目录内
            user_dir = f"/data/{user_id}/"
            if not args["path"].startswith(user_dir):
                return False
            return True

class ModelPrivacyGuard:
    """防止模型输出训练数据中的敏感信息"""

    def post_process(self, output: str, user_id: str) -> str:
        # 1. 检查输出中是否包含其他用户信息
        if self._contains_other_users_data(output, user_id):
            output = self._redact(output)

        # 2. 检查是否输出了训练数据
        if self._matches_training_data(output):
            output = "[输出被过滤：疑似训练数据泄露]"

        # 3. 脱敏处理
        output = InputPrivacyFilter().sanitize(output)

        return output

    def _contains_other_users_data(self, text: str, current_user: str) -> bool:
        """检查输出是否包含其他用户的数据"""
        # 用 NER 识别输出中的个人信息
        entities = ner.extract(text)
        for entity in entities:
            if entity.type == "PERSON" or entity.type == "EMAIL":
                # 检查这个信息属于谁
                owner = self.data_ownership.get(entity.text)
                if owner and owner != current_user:
                    return True
        return False

class OutputAuditor:
    def audit(self, agent_id: str, user_id: str,
              input_text: str, output_text: str, tools_used: list):
        """记录完整审计日志"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": agent_id,
            "user_id": user_id,
            "input_hash": hashlib.sha256(input_text.encode()).hexdigest(),
            "output_hash": hashlib.sha256(output_text.encode()).hexdigest(),
            "tools_called": tools_used,
            "data_accessed": self._trace_data_access(tools_used),
            "privacy_check": self._run_privacy_checks(output_text),
        }
        self.audit_log.insert(entry)

        # 异常告警
        if not entry["privacy_check"]["passed"]:
            self.alert.send(f"""
            ⚠️ 隐私违规告警
            Agent: {agent_id}
            User: {user_id}
            违规类型: {entry['privacy_check']['violations']}
            """)

class CrossBorderDataHandler:
    """处理跨境数据传输合规"""

    REGIONS = {
        "cn": {"allowed": ["cn", "hk"], "restricted": ["us", "eu"]},
        "eu": {"allowed": ["eu", "us_adequate"], "restricted": ["cn", "ru"]},
        "us": {"allowed": ["us", "eu", "uk"], "restricted": []},
    }

    def can_transfer(self, from_region: str, to_region: str,
                     data_type: str) -> bool:
        rules = self.REGIONS.get(from_region, {})
        if to_region in rules.get("restricted", []):
            if data_type == "personal_data":
                return False  # 禁止传输个人数据
            elif data_type == "anonymized":
                return True   # 匿名化数据可以传输
        return True

    def anonymize(self, data: dict) -> dict:
        """数据匿名化"""
        # 差分隐私：添加噪声
        for key in ["age", "income", "location"]:
            if key in data:
                data[key] = self._add_noise(data[key])
        # 去标识化
        for key in ["name", "email", "phone", "id_card"]:
            if key in data:
                data[key] = "ANONYMIZED"
        return data