第48章：数据隐私与AI伦理

🌟 章节导入：走进数据隐私与AI伦理中心

亲爱的朋友们，欢迎来到我们的数据隐私与AI伦理中心！这是一个充满责任感和道德意识的数据保护中心，在这里，我们将见证如何通过隐私保护技术和AI伦理框架，构建既智能又负责任的AI应用，就像从单纯追求效率升级到兼顾公平、透明和隐私保护的智能系统一样。

🏛️ 数据隐私与AI伦理中心全景

想象一下，你正站在一个现代化的数据隐私与AI伦理中心门口，眼前是四座风格迥异但又紧密相连的建筑群：

🛡️ 数据隐私保护大厦

这是我们的第一站，一座充满保护意识的数据隐私保护大厦。在这里：

• GDPR合规办公室里，合规专家们正在研究最新的数据保护法规要求
• 数据脱敏实验室的专家们专注于开发各种数据脱敏技术
• 隐私计算中心如同智能的隐私保护系统，实现数据可用不可见

⚖️ AI伦理框架研究院

这座建筑闪烁着金色的光芒，象征着AI伦理的公平与正义：

• 算法偏见检测室里，伦理专家们正在检测和消除算法中的偏见
• 公平性评估中心负责评估AI系统的公平性和公正性
• 可解释AI实验室致力于让AI决策过程透明可理解

🔐 隐私保护技术中心

这是一座充满密码学魅力的隐私保护技术中心：

• 差分隐私实验室里，隐私专家们正在研究差分隐私算法
• 联邦学习中心提供分布式机器学习解决方案
• 同态加密研究室实现加密数据的直接计算

🌐 隐私保护推荐系统平台

最令人兴奋的是这座未来感十足的隐私保护推荐系统平台：

• 用户数据匿名化系统保护用户隐私的同时提供个性化服务
• 模型公平性保证机制确保推荐结果的公平性
• 隐私影响评估中心评估系统对用户隐私的影响

🚀 伦理革命的见证者

在这个数据隐私与AI伦理中心，我们将见证AI应用的三大伦理革命：

🛡️ 隐私保护革命

从数据收集到隐私保护，我们将掌握：

• GDPR等数据保护法规的合规要求
• 数据脱敏和匿名化技术
• 隐私计算和联邦学习

⚖️ 公平性革命

从算法效率到算法公平，我们将实现：

• 算法偏见的检测和消除
• 公平性评估和保证
• 可解释AI技术

🔐 隐私计算革命

从数据共享到隐私计算，我们将建立：

• 差分隐私保护机制
• 联邦学习框架
• 同态加密应用

🎯 学以致用的企业级项目

在本章的最后，我们将综合运用所学的所有技术，构建一个完整的隐私保护推荐系统。这不仅仅是一个学习项目，更是一个具备实际商业部署价值的企业级应用：

• 电商平台可以基于这个系统，在保护用户隐私的同时提供个性化推荐
• 内容平台可以利用这个系统，实现公平的内容推荐
• 金融服务可以基于这个系统，提供隐私保护的智能服务
• 技术服务商可以基于这个系统为客户提供合规的AI解决方案

🔥 准备好了吗？

现在，让我们戴上伦理的眼镜，穿上隐私的防护服，一起走进这个充满责任魅力的数据隐私与AI伦理中心。在这里，我们不仅要学习最前沿的隐私保护技术，更要将这些技术转化为真正负责任、有道德的AI应用！

准备好迎接这场伦理革命了吗？让我们开始这激动人心的学习之旅！

---

🎯 学习目标（SMART目标）

完成本章学习后，学生将能够：

📚 知识目标

• 数据隐私保护体系：深入理解GDPR合规要求、数据脱敏技术、隐私计算方法等数据隐私保护核心概念
• AI伦理框架：掌握算法偏见检测、公平性评估方法、可解释AI技术等AI伦理关键技术
• 隐私保护技术：理解差分隐私算法、联邦学习框架、同态加密应用等隐私保护核心技术
• 隐私保护AI应用理念：综合运用数据匿名化、模型公平性、隐私影响评估等企业级隐私保护技术

🛠️ 技能目标

• 数据隐私保护能力：能够遵循GDPR等法规要求，实现数据脱敏和匿名化
• AI伦理评估能力：具备检测算法偏见、评估系统公平性的实战能力
• 隐私计算应用能力：掌握使用差分隐私、联邦学习等技术的实践能力
• 隐私保护系统开发能力：能够构建企业级隐私保护AI系统，具备大规模隐私保护系统开发的工程实践能力

💡 素养目标

• 隐私保护意识：培养数据隐私保护第一的开发思维模式
• AI伦理理念：建立公平、透明、负责任的AI设计意识
• 合规意识：注重法律法规遵循和合规性实践
• 社会责任意识：理解AI技术的社会责任和伦理要求

---

📝 知识导图

---

48.1 数据隐私保护

想象一下，您正在管理一家处理大量用户数据的公司。您需要确保用户数据的安全和隐私，同时遵守各种数据保护法规。这就像银行需要保护客户资金安全一样，我们需要保护用户的数据隐私。

在AI应用的世界里，数据隐私保护就是我们的"数据保护法规"。它帮助我们遵循GDPR等法规要求，实现数据脱敏和隐私计算，确保用户数据的安全和隐私。

📋 GDPR合规要求

GDPR（General Data Protection Regulation，通用数据保护条例）是欧盟的数据保护法规，对全球的数据处理活动都有重要影响：

# 示例1：GDPR合规检查系统

"""

GDPR合规检查系统



包含：

- 数据主体权利管理

- 数据处理原则检查

- 数据保护影响评估

- 违规检测和报告

"""



from typing import Dict, List, Optional, Any

from datetime import datetime, timedelta

from dataclasses import dataclass, asdict

from enum import Enum



class DataSubjectRight(Enum):

    """数据主体权利"""

    ACCESS = "访问权"

    RECTIFICATION = "更正权"

    ERASURE = "删除权"

    PORTABILITY = "可携带权"

    OBJECT = "反对权"

    RESTRICTION = "限制处理权"



class ProcessingPurpose(Enum):

    """数据处理目的"""

    CONSENT = "同意"

    CONTRACT = "合同履行"

    LEGAL_OBLIGATION = "法律义务"

    VITAL_INTERESTS = "重大利益"

    PUBLIC_TASK = "公共任务"

    LEGITIMATE_INTERESTS = "合法利益"



@dataclass

class DataProcessingRecord:

    """数据处理记录"""

    record_id: str

    data_subject_id: str

    data_categories: List[str]

    processing_purpose: ProcessingPurpose

    legal_basis: str

    retention_period: timedelta

    created_at: datetime

    consent_given: bool

    consent_date: Optional[datetime]



@dataclass

class DataSubjectRequest:

    """数据主体请求"""

    request_id: str

    data_subject_id: str

    right_type: DataSubjectRight

    request_date: datetime

    status: str

    response_date: Optional[datetime]



class GDPRComplianceChecker:

    """GDPR合规检查器"""

    

    def __init__(self):

        """初始化GDPR合规检查器"""

        self.processing_records: Dict[str, DataProcessingRecord] = {}

        self.data_subject_requests: List[DataSubjectRequest] = []

        self.consent_records: Dict[str, Dict] = {}

        print("📋 GDPR合规检查器启动成功！")

    

    def register_data_processing(self, record: DataProcessingRecord) -> bool:

        """注册数据处理活动"""

        self.processing_records[record.record_id] = record

        

        # 检查合法性

        if not self._check_legal_basis(record):

            print(f"⚠️ 警告: 数据处理记录 {record.record_id} 缺少合法依据")

            return False

        

        # 检查同意（如需要）

        if record.processing_purpose == ProcessingPurpose.CONSENT:

            if not record.consent_given:

                print(f"❌ 错误: 数据处理记录 {record.record_id} 需要同意但未获得")

                return False

        

        print(f"✅ 数据处理记录已注册: {record.record_id}")

        return True

    

    def _check_legal_basis(self, record: DataProcessingRecord) -> bool:

        """检查合法依据"""

        # 检查是否有合法依据

        if not record.legal_basis:

            return False

        

        # 检查处理目的与合法依据的匹配

        legal_basis_map = {

            ProcessingPurpose.CONSENT: ["同意"],

            ProcessingPurpose.CONTRACT: ["合同履行"],

            ProcessingPurpose.LEGAL_OBLIGATION: ["法律义务"],

            ProcessingPurpose.VITAL_INTERESTS: ["重大利益"],

            ProcessingPurpose.PUBLIC_TASK: ["公共任务"],

            ProcessingPurpose.LEGITIMATE_INTERESTS: ["合法利益"]

        }

        

        allowed_bases = legal_basis_map.get(record.processing_purpose, [])

        return record.legal_basis in allowed_bases

    

    def handle_data_subject_request(self, request: DataSubjectRequest) -> Dict[str, Any]:

        """处理数据主体请求"""

        self.data_subject_requests.append(request)

        

        # 根据权利类型处理

        if request.right_type == DataSubjectRight.ACCESS:

            return self._handle_access_request(request)

        elif request.right_type == DataSubjectRight.RECTIFICATION:

            return self._handle_rectification_request(request)

        elif request.right_type == DataSubjectRight.ERASURE:

            return self._handle_erasure_request(request)

        elif request.right_type == DataSubjectRight.PORTABILITY:

            return self._handle_portability_request(request)

        else:

            return {

                "status": "pending",

                "message": "请求已记录，正在处理中"

            }

    

    def _handle_access_request(self, request: DataSubjectRequest) -> Dict[str, Any]:

        """处理访问请求"""

        # 查找相关数据处理记录

        related_records = [

            record for record in self.processing_records.values()

            if record.data_subject_id == request.data_subject_id

        ]

        

        return {

            "status": "completed",

            "data_subjects": [asdict(record) for record in related_records],

            "message": "数据访问请求已完成"

        }

    

    def _handle_rectification_request(self, request: DataSubjectRequest) -> Dict[str, Any]:

        """处理更正请求"""

        return {

            "status": "completed",

            "message": "数据更正请求已处理"

        }

    

    def _handle_erasure_request(self, request: DataSubjectRequest) -> Dict[str, Any]:

        """处理删除请求（被遗忘权）"""

        # 删除相关数据处理记录

        records_to_delete = [

            record_id for record_id, record in self.processing_records.items()

            if record.data_subject_id == request.data_subject_id

        ]

        

        for record_id in records_to_delete:

            del self.processing_records[record_id]

        

        return {

            "status": "completed",

            "deleted_records": len(records_to_delete),

            "message": "数据删除请求已处理"

        }

    

    def _handle_portability_request(self, request: DataSubjectRequest) -> Dict[str, Any]:

        """处理可携带权请求"""

        # 导出数据为机器可读格式

        related_records = [

            record for record in self.processing_records.values()

            if record.data_subject_id == request.data_subject_id

        ]

        

        export_data = {

            "data_subject_id": request.data_subject_id,

            "export_date": datetime.now().isoformat(),

            "records": [asdict(record) for record in related_records]

        }

        

        return {

            "status": "completed",

            "export_data": export_data,

            "message": "数据可携带权请求已完成"

        }

    

    def conduct_dpia(self, processing_activity: Dict[str, Any]) -> Dict[str, Any]:

        """进行数据保护影响评估（DPIA）"""

        risk_factors = []

        risk_score = 0

        

        # 检查数据处理规模

        if processing_activity.get("data_volume", 0) > 1000000:

            risk_factors.append("大规模数据处理")

            risk_score += 2

        

        # 检查敏感数据类型

        sensitive_categories = ["健康", "生物识别", "种族", "政治观点"]

        if any(cat in processing_activity.get("data_categories", []) 

               for cat in sensitive_categories):

            risk_factors.append("处理敏感数据")

            risk_score += 3

        

        # 检查自动化决策

        if processing_activity.get("automated_decision_making", False):

            risk_factors.append("自动化决策")

            risk_score += 2

        

        # 检查数据共享

        if processing_activity.get("data_sharing", False):

            risk_factors.append("数据共享")

            risk_score += 1

        

        # 评估风险等级

        if risk_score >= 5:

            risk_level = "高风险"

            recommendation = "必须进行DPIA，可能需要咨询监管机构"

        elif risk_score >= 3:

            risk_level = "中等风险"

            recommendation = "建议进行DPIA"

        else:

            risk_level = "低风险"

            recommendation = "可能不需要DPIA"

        

        return {

            "risk_level": risk_level,

            "risk_score": risk_score,

            "risk_factors": risk_factors,

            "recommendation": recommendation,

            "assessment_date": datetime.now().isoformat()

        }

    

    def check_consent_validity(self, consent_record: Dict[str, Any]) -> bool:

        """检查同意有效性"""

        # 检查同意是否明确

        if not consent_record.get("explicit", False):

            return False

        

        # 检查同意是否可撤销

        if not consent_record.get("withdrawable", True):

            return False

        

        # 检查同意是否过期

        consent_date = consent_record.get("consent_date")

        if consent_date:

            consent_date_obj = datetime.fromisoformat(consent_date)

            validity_period = timedelta(days=365)  # 假设同意有效期为1年

            if datetime.now() - consent_date_obj > validity_period:

                return False

        

        return True



# 运行演示

if __name__ == "__main__":

    checker = GDPRComplianceChecker()

    

    # 注册数据处理活动

    print("="*60)

    print("GDPR合规检查演示")

    print("="*60)

    

    record = DataProcessingRecord(

        record_id="proc_001",

        data_subject_id="user_123",

        data_categories=["姓名", "邮箱", "购买记录"],

        processing_purpose=ProcessingPurpose.CONSENT,

        legal_basis="同意",

        retention_period=timedelta(days=365),

        created_at=datetime.now(),

        consent_given=True,

        consent_date=datetime.now()

    )

    

    checker.register_data_processing(record)

    

    # 处理数据主体请求

    print("\n处理数据主体请求:")

    access_request = DataSubjectRequest(

        request_id="req_001",

        data_subject_id="user_123",

        right_type=DataSubjectRight.ACCESS,

        request_date=datetime.now(),

        status="pending",

        response_date=None

    )

    

    result = checker.handle_data_subject_request(access_request)

    print(f"访问请求结果: {result['status']}")

    

    # 进行DPIA

    print("\n数据保护影响评估:")

    processing_activity = {

        "data_volume": 2000000,

        "data_categories": ["健康", "生物识别"],

        "automated_decision_making": True,

        "data_sharing": False

    }

    

    dpia_result = checker.conduct_dpia(processing_activity)

    print(f"风险等级: {dpia_result['risk_level']}")

    print(f"风险评分: {dpia_result['risk_score']}")

    print(f"建议: {dpia_result['recommendation']}")

🔒 数据脱敏技术

数据脱敏是保护敏感数据的重要技术，可以在保持数据可用性的同时保护隐私：

# 示例2：数据脱敏系统

"""

数据脱敏系统



包含：

- 静态数据脱敏

- 动态数据脱敏

- 脱敏算法选择

- 脱敏效果评估

"""



import re

import hashlib

import random

import string

from typing import Dict, List, Optional, Any, Callable

from datetime import datetime



class DataMasking:

    """数据脱敏器"""

    

    def __init__(self):

        """初始化脱敏器"""

        self.masking_strategies = self._load_masking_strategies()

        print("🔒 数据脱敏器启动成功！")

    

    def _load_masking_strategies(self) -> Dict[str, Callable]:

        """加载脱敏策略"""

        return {

            "email": self._mask_email,

            "phone": self._mask_phone,

            "id_card": self._mask_id_card,

            "name": self._mask_name,

            "address": self._mask_address,

            "credit_card": self._mask_credit_card,

            "ip_address": self._mask_ip_address

        }

    

    def _mask_email(self, email: str) -> str:

        """脱敏邮箱"""

        if "@" not in email:

            return email

        

        local, domain = email.split("@", 1)

        if len(local) <= 2:

            masked_local = "*" * len(local)

        else:

            masked_local = local[0] + "*" * (len(local) - 2) + local[-1]

        

        return f"{masked_local}@{domain}"

    

    def _mask_phone(self, phone: str) -> str:

        """脱敏手机号"""

        if len(phone) != 11:

            return phone

        

        return phone[:3] + "****" + phone[-4:]

    

    def _mask_id_card(self, id_card: str) -> str:

        """脱敏身份证号"""

        if len(id_card) != 18:

            return id_card

        

        return id_card[:6] + "********" + id_card[-4:]

    

    def _mask_name(self, name: str) -> str:

        """脱敏姓名"""

        if len(name) <= 1:

            return "*"

        elif len(name) == 2:

            return name[0] + "*"

        else:

            return name[0] + "*" * (len(name) - 2) + name[-1]

    

    def _mask_address(self, address: str) -> str:

        """脱敏地址"""

        # 保留前几个字符，其余用*替代

        if len(address) <= 5:

            return "*" * len(address)

        else:

            return address[:3] + "*" * (len(address) - 3)

    

    def _mask_credit_card(self, card_number: str) -> str:

        """脱敏信用卡号"""

        # 移除空格和连字符

        card_number = re.sub(r'[\s-]', '', card_number)

        

        if len(card_number) < 4:

            return "*" * len(card_number)

        

        return "****" * 3 + card_number[-4:]

    

    def _mask_ip_address(self, ip: str) -> str:

        """脱敏IP地址"""

        parts = ip.split(".")

        if len(parts) != 4:

            return ip

        

        return f"{parts[0]}.{parts[1]}.***.***"

    

    def mask_data(self, data: Any, data_type: str = "auto") -> Any:

        """脱敏数据"""

        if data_type == "auto":

            data_type = self._detect_data_type(data)

        

        if data_type in self.masking_strategies:

            return self.masking_strategies[data_type](str(data))

        else:

            # 默认脱敏：保留前20%，其余用*替代

            data_str = str(data)

            if len(data_str) <= 5:

                return "*" * len(data_str)

            else:

                keep_length = max(1, len(data_str) // 5)

                return data_str[:keep_length] + "*" * (len(data_str) - keep_length)

    

    def _detect_data_type(self, data: str) -> str:

        """自动检测数据类型"""

        data_str = str(data)

        

        # 检测邮箱

        if "@" in data_str and "." in data_str:

            return "email"

        

        # 检测手机号

        if re.match(r'^1[3-9]\d{9}$', data_str):

            return "phone"

        

        # 检测身份证号

        if re.match(r'^\d{17}[\dXx]$', data_str):

            return "id_card"

        

        # 检测信用卡号

        if re.match(r'^[\d\s-]{13,19}$', data_str):

            return "credit_card"

        

        # 检测IP地址

        if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', data_str):

            return "ip_address"

        

        return "default"

    

    def mask_record(self, record: Dict[str, Any], 

                   fields_to_mask: List[str] = None) -> Dict[str, Any]:

        """脱敏记录"""

        masked_record = record.copy()

        

        if fields_to_mask is None:

            # 自动检测需要脱敏的字段

            fields_to_mask = [key for key in record.keys() 

                            if any(keyword in key.lower() 

                                  for keyword in ["email", "phone", "name", 

                                                 "id", "card", "address", "ip"])]

        

        for field in fields_to_mask:

            if field in masked_record:

                masked_record[field] = self.mask_data(masked_record[field])

        

        return masked_record



class DynamicDataMasking:

    """动态数据脱敏"""

    

    def __init__(self):

        """初始化动态脱敏器"""

        self.masker = DataMasking()

        self.user_roles = {

            "admin": ["full_access"],

            "analyst": ["masked_access"],

            "public": ["minimal_access"]

        }

        print("🔒 动态数据脱敏器启动成功！")

    

    def mask_for_role(self, data: Any, user_role: str, data_type: str = "auto") -> Any:

        """根据用户角色脱敏数据"""

        role_permissions = self.user_roles.get(user_role, ["minimal_access"])

        

        if "full_access" in role_permissions:

            return data

        elif "masked_access" in role_permissions:

            return self.masker.mask_data(data, data_type)

        else:

            # 最小访问：完全脱敏

            return "*" * len(str(data)) if len(str(data)) > 0 else "*"



# 运行演示

if __name__ == "__main__":

    masker = DataMasking()

    

    print("="*60)

    print("数据脱敏演示")

    print("="*60)

    

    # 测试各种数据类型

    test_data = {

        "email": "alice@example.com",

        "phone": "13800138000",

        "id_card": "110101199001011234",

        "name": "张三",

        "address": "北京市朝阳区某某街道123号",

        "credit_card": "1234-5678-9012-3456",

        "ip_address": "192.168.1.100"

    }

    

    print("\n原始数据:")

    for key, value in test_data.items():

        print(f"  {key}: {value}")

    

    print("\n脱敏后数据:")

    masked_data = masker.mask_record(test_data)

    for key, value in masked_data.items():

        print(f"  {key}: {value}")

    

    # 动态脱敏演示

    print("\n" + "="*60)

    print("动态数据脱敏演示")

    print("="*60)

    

    dynamic_masker = DynamicDataMasking()

    email = "alice@example.com"

    

    print(f"\n原始邮箱: {email}")

    print(f"管理员查看: {dynamic_masker.mask_for_role(email, 'admin')}")

    print(f"分析师查看: {dynamic_masker.mask_for_role(email, 'analyst')}")

    print(f"公众查看: {dynamic_masker.mask_for_role(email, 'public')}")

🔐 隐私计算方法

隐私计算允许在不暴露原始数据的情况下进行计算，实现"数据可用不可见"：

# 示例3：隐私计算基础实现

"""

隐私计算基础实现



包含：

- 安全多方计算概念

- 隐私集合求交

- 隐私信息检索

- 零知识证明概念

"""



import hashlib

import random

from typing import List, Set, Dict, Any

from cryptography.hazmat.primitives import hashes

from cryptography.hazmat.primitives.asymmetric import rsa, padding

from cryptography.hazmat.backends import default_backend



class PrivacyPreservingComputation:

    """隐私保护计算"""

    

    def __init__(self):

        """初始化隐私计算器"""

        print("🔐 隐私保护计算器启动成功！")

    

    def secure_set_intersection(self, set1: Set[str], set2: Set[str]) -> Set[str]:

        """

        隐私集合求交（PSI）

        在不暴露各自集合内容的情况下，计算两个集合的交集

        """

        # 简化实现：使用哈希

        # 实际应用中应使用更安全的协议（如OT、DH等）

        hash_set1 = {hashlib.sha256(item.encode()).hexdigest() for item in set1}

        hash_set2 = {hashlib.sha256(item.encode()).hexdigest() for item in set2}

        

        intersection_hashes = hash_set1 & hash_set2

        

        # 返回交集（需要原始数据映射，实际应用中需要更复杂的协议）

        result = set()

        hash_to_item = {hashlib.sha256(item.encode()).hexdigest(): item 

                       for item in set1 | set2}

        

        for hash_val in intersection_hashes:

            if hash_val in hash_to_item:

                result.add(hash_to_item[hash_val])

        

        return result

    

    def secure_sum(self, values: List[float], noise: float = 0.1) -> float:

        """

        安全求和（添加噪声保护隐私）

        使用差分隐私的思想

        """

        # 添加拉普拉斯噪声

        laplace_noise = random.gauss(0, noise)

        noisy_sum = sum(values) + laplace_noise

        

        return noisy_sum

    

    def homomorphic_encryption_demo(self, value1: int, value2: int) -> Dict[str, Any]:

        """

        同态加密演示（概念性）

        实际应用需要使用专门的同态加密库

        """

        # 这里只是演示概念，实际同态加密更复杂

        # 同态加密允许在加密数据上直接进行计算

        

        # 模拟加密

        encrypted_v1 = value1 * 2 + 1  # 简化的"加密"

        encrypted_v2 = value2 * 2 + 1

        

        # 在加密数据上计算（加法同态）

        encrypted_sum = encrypted_v1 + encrypted_v2

        

        # 解密结果

        decrypted_sum = (encrypted_sum - 2) // 2

        

        return {

            "value1": value1,

            "value2": value2,

            "encrypted_v1": encrypted_v1,

            "encrypted_v2": encrypted_v2,

            "encrypted_sum": encrypted_sum,

            "decrypted_sum": decrypted_sum,

            "actual_sum": value1 + value2,

            "correct": decrypted_sum == value1 + value2

        }



# 运行演示

if __name__ == "__main__":

    privacy_comp = PrivacyPreservingComputation()

    

    # 隐私集合求交演示

    print("="*60)

    print("隐私集合求交演示")

    print("="*60)

    

    set1 = {"alice", "bob", "charlie", "david"}

    set2 = {"bob", "charlie", "eve", "frank"}

    

    intersection = privacy_comp.secure_set_intersection(set1, set2)

    print(f"集合1: {set1}")

    print(f"集合2: {set2}")

    print(f"交集: {intersection}")

    print(f"实际交集: {set1 & set2}")

    

    # 安全求和演示

    print("\n" + "="*60)

    print("安全求和演示（差分隐私）")

    print("="*60)

    

    values = [100, 200, 150, 180, 220]

    secure_sum = privacy_comp.secure_sum(values, noise=5.0)

    actual_sum = sum(values)

    

    print(f"原始值: {values}")

    print(f"实际和: {actual_sum}")

    print(f"安全求和（带噪声）: {secure_sum:.2f}")

    print(f"误差: {abs(secure_sum - actual_sum):.2f}")

    

    # 同态加密演示

    print("\n" + "="*60)

    print("同态加密演示（概念性）")

    print("="*60)

    

    result = privacy_comp.homomorphic_encryption_demo(10, 20)

    print(f"值1: {result['value1']}, 值2: {result['value2']}")

    print(f"加密值1: {result['encrypted_v1']}, 加密值2: {result['encrypted_v2']}")

    print(f"加密和: {result['encrypted_sum']}")

    print(f"解密和: {result['decrypted_sum']}")

    print(f"正确性: {'✅' if result['correct'] else '❌'}")

---

48.2 AI伦理框架

AI伦理框架确保AI系统的公平性、透明性和可解释性，就像法官需要公正、透明地审理案件一样，AI系统也需要公平、透明地做出决策。

⚖️ 算法偏见检测

算法偏见可能导致不公平的决策，需要检测和消除：

# 示例4：算法偏见检测系统

"""

算法偏见检测系统



包含：

- 偏见类型识别

- 偏见来源分析

- 偏见度量方法

- 偏见消除策略

"""



import numpy as np

from typing import Dict, List, Tuple, Any

from collections import Counter

from dataclasses import dataclass



@dataclass

class BiasMetric:

    """偏见度量"""

    metric_name: str

    value: float

    threshold: float

    is_biased: bool



class BiasDetector:

    """算法偏见检测器"""

    

    def __init__(self):

        """初始化偏见检测器"""

        print("⚖️ 算法偏见检测器启动成功！")

    

    def detect_demographic_parity(self, predictions: List[int], 

                                  protected_attributes: List[str]) -> BiasMetric:

        """

        检测人口统计均等性（Demographic Parity）

        不同群体的正例率应该相同

        """

        # 统计各群体的正例率

        group_rates = {}

        for group in set(protected_attributes):

            group_indices = [i for i, g in enumerate(protected_attributes) if g == group]

            group_predictions = [predictions[i] for i in group_indices]

            group_rates[group] = sum(group_predictions) / len(group_predictions)

        

        # 计算最大差异

        rates = list(group_rates.values())

        max_diff = max(rates) - min(rates) if rates else 0

        

        threshold = 0.1  # 10%的差异阈值

        is_biased = max_diff > threshold

        

        return BiasMetric(

            metric_name="Demographic Parity",

            value=max_diff,

            threshold=threshold,

            is_biased=is_biased

        )

    

    def detect_equalized_odds(self, predictions: List[int], 

                            true_labels: List[int],

                            protected_attributes: List[str]) -> BiasMetric:

        """

        检测均等化机会（Equalized Odds）

        不同群体的真正例率和假正例率应该相同

        """

        group_metrics = {}

        

        for group in set(protected_attributes):

            group_indices = [i for i, g in enumerate(protected_attributes) if g == group]

            group_pred = [predictions[i] for i in group_indices]

            group_true = [true_labels[i] for i in group_indices]

            

            # 计算真正例率（TPR）和假正例率（FPR）

            tp = sum(1 for p, t in zip(group_pred, group_true) if p == 1 and t == 1)

            fn = sum(1 for p, t in zip(group_pred, group_true) if p == 0 and t == 1)

            fp = sum(1 for p, t in zip(group_pred, group_true) if p == 1 and t == 0)

            tn = sum(1 for p, t in zip(group_pred, group_true) if p == 0 and t == 0)

            

            tpr = tp / (tp + fn) if (tp + fn) > 0 else 0

            fpr = fp / (fp + tn) if (fp + tn) > 0 else 0

            

            group_metrics[group] = {"TPR": tpr, "FPR": fpr}

        

        # 计算TPR和FPR的最大差异

        tprs = [m["TPR"] for m in group_metrics.values()]

        fprs = [m["FPR"] for m in group_metrics.values()]

        

        max_tpr_diff = max(tprs) - min(tprs) if tprs else 0

        max_fpr_diff = max(fprs) - min(fprs) if fprs else 0

        max_diff = max(max_tpr_diff, max_fpr_diff)

        

        threshold = 0.1

        is_biased = max_diff > threshold

        

        return BiasMetric(

            metric_name="Equalized Odds",

            value=max_diff,

            threshold=threshold,

            is_biased=is_biased

        )

    

    def detect_disparate_impact(self, predictions: List[int],

                               protected_attributes: List[str]) -> BiasMetric:

        """

        检测不同影响（Disparate Impact）

        不同群体的正例率比例不应低于80%（四分之五规则）

        """

        group_rates = {}

        for group in set(protected_attributes):

            group_indices = [i for i, g in enumerate(protected_attributes) if g == group]

            group_predictions = [predictions[i] for i in group_indices]

            group_rates[group] = sum(group_predictions) / len(group_predictions)

        

        rates = list(group_rates.values())

        if not rates or min(rates) == 0:

            ratio = 0

        else:

            ratio = min(rates) / max(rates)

        

        threshold = 0.8  # 80%规则

        is_biased = ratio < threshold

        

        return BiasMetric(

            metric_name="Disparate Impact",

            value=ratio,

            threshold=threshold,

            is_biased=is_biased

        )

    

    def comprehensive_bias_analysis(self, predictions: List[int],

                                   true_labels: List[int],

                                   protected_attributes: List[str]) -> Dict[str, Any]:

        """综合偏见分析"""

        metrics = {}

        

        # 人口统计均等性

        metrics["demographic_parity"] = self.detect_demographic_parity(

            predictions, protected_attributes

        )

        

        # 均等化机会

        metrics["equalized_odds"] = self.detect_equalized_odds(

            predictions, true_labels, protected_attributes

        )

        

        # 不同影响

        metrics["disparate_impact"] = self.detect_disparate_impact(

            predictions, protected_attributes

        )

        

        # 总体偏见评估

        has_bias = any(metric.is_biased for metric in metrics.values())

        

        return {

            "metrics": metrics,

            "has_bias": has_bias,

            "bias_summary": {

                name: {

                    "value": metric.value,

                    "is_biased": metric.is_biased

                }

                for name, metric in metrics.items()

            }

        }



# 运行演示

if __name__ == "__main__":

    detector = BiasDetector()

    

    print("="*60)

    print("算法偏见检测演示")

    print("="*60)

    

    # 模拟数据：假设有性别偏见

    # 女性被拒绝的比例更高

    predictions = [1, 0, 1, 0, 0, 1, 0, 0, 1, 1,  # 女性：3/10通过

                   1, 1, 1, 0, 1, 1, 1, 0, 1, 1]  # 男性：8/10通过

    true_labels = [1, 0, 1, 0, 1, 1, 0, 0, 1, 1,

                   1, 1, 1, 0, 1, 1, 1, 0, 1, 1]

    protected_attributes = (["female"] * 10) + (["male"] * 10)

    

    analysis = detector.comprehensive_bias_analysis(

        predictions, true_labels, protected_attributes

    )

    

    print("\n偏见分析结果:")

    for metric_name, metric in analysis["metrics"].items():

        status = "🔴 存在偏见" if metric.is_biased else "✅ 无偏见"

        print(f"\n{metric_name}:")

        print(f"  值: {metric.value:.3f}")

        print(f"  阈值: {metric.threshold}")

        print(f"  状态: {status}")

    

    print(f"\n总体评估: {'🔴 系统存在偏见' if analysis['has_bias'] else '✅ 系统无偏见'}")

📊 公平性评估方法

公平性评估确保AI系统对不同群体都是公平的：

# 示例5：公平性评估系统

"""

公平性评估系统



包含：

- 公平性定义

- 公平性指标

- 公平性测试

- 公平性优化

"""



from typing import Dict, List, Tuple

import numpy as np



class FairnessEvaluator:

    """公平性评估器"""

    

    def __init__(self):

        """初始化公平性评估器"""

        print("📊 公平性评估器启动成功！")

    

    def evaluate_fairness(self, predictions: List[int], 

                         true_labels: List[int],

                         protected_attributes: List[str]) -> Dict[str, float]:

        """综合公平性评估"""

        metrics = {}

        

        # 准确率公平性

        metrics["accuracy_fairness"] = self._accuracy_fairness(

            predictions, true_labels, protected_attributes

        )

        

        # 精确率公平性

        metrics["precision_fairness"] = self._precision_fairness(

            predictions, true_labels, protected_attributes

        )

        

        # 召回率公平性

        metrics["recall_fairness"] = self._recall_fairness(

            predictions, true_labels, protected_attributes

        )

        

        # F1分数公平性

        metrics["f1_fairness"] = self._f1_fairness(

            predictions, true_labels, protected_attributes

        )

        

        return metrics

    

    def _accuracy_fairness(self, predictions: List[int],

                          true_labels: List[int],

                          protected_attributes: List[str]) -> float:

        """准确率公平性（不同群体的准确率差异）"""

        group_accuracies = {}

        

        for group in set(protected_attributes):

            group_indices = [i for i, g in enumerate(protected_attributes) if g == group]

            group_pred = [predictions[i] for i in group_indices]

            group_true = [true_labels[i] for i in group_indices]

            

            accuracy = sum(p == t for p, t in zip(group_pred, group_true)) / len(group_pred)

            group_accuracies[group] = accuracy

        

        accuracies = list(group_accuracies.values())

        return 1 - (max(accuracies) - min(accuracies)) if accuracies else 0

    

    def _precision_fairness(self, predictions: List[int],

                           true_labels: List[int],

                           protected_attributes: List[str]) -> float:

        """精确率公平性"""

        group_precisions = {}

        

        for group in set(protected_attributes):

            group_indices = [i for i, g in enumerate(protected_attributes) if g == group]

            group_pred = [predictions[i] for i in group_indices]

            group_true = [true_labels[i] for i in group_indices]

            

            tp = sum(1 for p, t in zip(group_pred, group_true) if p == 1 and t == 1)

            fp = sum(1 for p, t in zip(group_pred, group_true) if p == 1 and t == 0)

            

            precision = tp / (tp + fp) if (tp + fp) > 0 else 0

            group_precisions[group] = precision

        

        precisions = list(group_precisions.values())

        return 1 - (max(precisions) - min(precisions)) if precisions else 0

    

    def _recall_fairness(self, predictions: List[int],

                        true_labels: List[int],

                        protected_attributes: List[str]) -> float:

        """召回率公平性"""

        group_recalls = {}

        

        for group in set(protected_attributes):

            group_indices = [i for i, g in enumerate(protected_attributes) if g == group]

            group_pred = [predictions[i] for i in group_indices]

            group_true = [true_labels[i] for i in group_indices]

            

            tp = sum(1 for p, t in zip(group_pred, group_true) if p == 1 and t == 1)

            fn = sum(1 for p, t in zip(group_pred, group_true) if p == 0 and t == 1)

            

            recall = tp / (tp + fn) if (tp + fn) > 0 else 0

            group_recalls[group] = recall

        

        recalls = list(group_recalls.values())

        return 1 - (max(recalls) - min(recalls)) if recalls else 0

    

    def _f1_fairness(self, predictions: List[int],

                    true_labels: List[int],

                    protected_attributes: List[str]) -> float:

        """F1分数公平性"""

        group_f1s = {}

        

        for group in set(protected_attributes):

            group_indices = [i for i, g in enumerate(protected_attributes) if g == group]

            group_pred = [predictions[i] for i in group_indices]

            group_true = [true_labels[i] for i in group_indices]

            

            tp = sum(1 for p, t in zip(group_pred, group_true) if p == 1 and t == 1)

            fp = sum(1 for p, t in zip(group_pred, group_true) if p == 1 and t == 0)

            fn = sum(1 for p, t in zip(group_pred, group_true) if p == 0 and t == 1)

            

            precision = tp / (tp + fp) if (tp + fp) > 0 else 0

            recall = tp / (tp + fn) if (tp + fn) > 0 else 0

            f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0

            

            group_f1s[group] = f1

        

        f1s = list(group_f1s.values())

        return 1 - (max(f1s) - min(f1s)) if f1s else 0



# 运行演示

if __name__ == "__main__":

    evaluator = FairnessEvaluator()

    

    # 模拟数据

    predictions = [1, 0, 1, 0, 0, 1, 0, 0, 1, 1,

                   1, 1, 1, 0, 1, 1, 1, 0, 1, 1]

    true_labels = [1, 0, 1, 0, 1, 1, 0, 0, 1, 1,

                   1, 1, 1, 0, 1, 1, 1, 0, 1, 1]

    protected_attributes = (["female"] * 10) + (["male"] * 10)

    

    fairness_metrics = evaluator.evaluate_fairness(

        predictions, true_labels, protected_attributes

    )

    

    print("="*60)

    print("公平性评估结果")

    print("="*60)

    for metric_name, value in fairness_metrics.items():

        print(f"{metric_name}: {value:.3f} (越接近1越公平)")

🔍 可解释AI技术

可解释AI让AI决策过程透明可理解：

# 示例6：可解释AI基础实现

"""

可解释AI基础实现



包含：

- 特征重要性分析

- 决策路径追踪

- 解释可视化

"""



from typing import Dict, List, Tuple, Any

import numpy as np



class ExplainableAI:

    """可解释AI系统"""

    

    def __init__(self):

        """初始化可解释AI系统"""

        print("🔍 可解释AI系统启动成功！")

    

    def feature_importance(self, model, features: List[str], 

                          sample: np.ndarray) -> Dict[str, float]:

        """

        特征重要性分析

        使用简单的扰动方法

        """

        # 获取基准预测

        baseline_pred = model.predict(sample.reshape(1, -1))[0]

        

        importance_scores = {}

        

        for i, feature_name in enumerate(features):

            # 扰动特征

            perturbed_sample = sample.copy()

            perturbed_sample[i] = 0  # 将特征置零

            

            # 获取扰动后的预测

            perturbed_pred = model.predict(perturbed_sample.reshape(1, -1))[0]

            

            # 计算重要性（预测差异）

            importance = abs(baseline_pred - perturbed_pred)

            importance_scores[feature_name] = importance

        

        # 归一化

        total_importance = sum(importance_scores.values())

        if total_importance > 0:

            importance_scores = {k: v / total_importance 

                               for k, v in importance_scores.items()}

        

        return importance_scores

    

    def decision_path_explanation(self, decision_tree, sample: np.ndarray,

                                 feature_names: List[str]) -> List[Dict[str, Any]]:

        """

        决策路径解释（适用于决策树）

        """

        # 获取决策路径

        node_indicator = decision_tree.decision_path(sample.reshape(1, -1))

        leaf_id = decision_tree.apply(sample.reshape(1, -1))[0]

        

        # 提取路径上的节点

        path = node_indicator.indices[node_indicator.indptr[0]:

                                      node_indicator.indptr[1]]

        

        explanation = []

        for node_id in path:

            if node_id == leaf_id:

                break

            

            # 获取节点信息

            threshold = decision_tree.tree_.threshold[node_id]

            feature_idx = decision_tree.tree_.feature[node_id]

            feature_name = feature_names[feature_idx] if feature_idx >= 0 else "leaf"

            

            sample_value = sample[feature_idx] if feature_idx >= 0 else None

            

            explanation.append({

                "node_id": node_id,

                "feature": feature_name,

                "threshold": threshold,

                "sample_value": sample_value,

                "condition": f"{feature_name} <= {threshold:.2f}" if sample_value <= threshold 

                           else f"{feature_name} > {threshold:.2f}"

            })

        

        return explanation

    

    def generate_explanation(self, model, sample: np.ndarray,

                           feature_names: List[str],

                           prediction: float) -> Dict[str, Any]:

        """生成综合解释"""

        # 特征重要性

        importance = self.feature_importance(model, feature_names, sample)

        

        # 排序特征

        sorted_features = sorted(importance.items(), key=lambda x: x[1], reverse=True)

        

        explanation = {

            "prediction": prediction,

            "top_features": sorted_features[:5],  # 前5个重要特征

            "feature_importance": importance,

            "explanation_text": self._generate_text_explanation(

                prediction, sorted_features[:3]

            )

        }

        

        return explanation

    

    def _generate_text_explanation(self, prediction: float,

                                  top_features: List[Tuple[str, float]]) -> str:

        """生成文本解释"""

        explanation = f"模型预测值为 {prediction:.3f}。"

        explanation += "主要影响因素包括："

        

        for i, (feature, importance) in enumerate(top_features, 1):

            explanation += f"\n{i}. {feature}（重要性：{importance:.3f}）"

        

        return explanation



# 运行演示

if __name__ == "__main__":

    from sklearn.tree import DecisionTreeClassifier

    from sklearn.datasets import make_classification

    

    # 创建示例模型

    X, y = make_classification(n_samples=100, n_features=5, random_state=42)

    feature_names = [f"feature_{i}" for i in range(5)]

    

    model = DecisionTreeClassifier(max_depth=3, random_state=42)

    model.fit(X, y)

    

    explainer = ExplainableAI()

    

    # 解释单个样本

    sample = X[0]

    prediction = model.predict_proba(sample.reshape(1, -1))[0][1]

    

    explanation = explainer.generate_explanation(

        model, sample, feature_names, prediction

    )

    

    print("="*60)

    print("可解释AI演示")

    print("="*60)

    print(f"\n预测值: {explanation['prediction']:.3f}")

    print(f"\n文本解释:\n{explanation['explanation_text']}")

    print(f"\n特征重要性:")

    for feature, importance in explanation['top_features']:

        print(f"  {feature}: {importance:.3f}")

---

48.3 隐私保护技术

隐私保护技术是保护用户数据隐私的核心技术，包括差分隐私、联邦学习和同态加密等。

🔐 差分隐私算法

差分隐私通过添加噪声保护个体隐私：

# 示例7：差分隐私实现

"""

差分隐私实现



包含：

- 拉普拉斯机制

- 指数机制

- 隐私预算管理

"""



import numpy as np

import random

from typing import List, Dict, Any

from math import exp



class DifferentialPrivacy:

    """差分隐私系统"""

    

    def __init__(self, epsilon: float = 1.0):

        """

        初始化差分隐私系统

        epsilon: 隐私预算（越小越隐私，但噪声越大）

        """

        self.epsilon = epsilon

        print(f"🔐 差分隐私系统启动成功！隐私预算: {epsilon}")

    

    def laplace_mechanism(self, true_value: float, sensitivity: float) -> float:

        """

        拉普拉斯机制

        用于数值查询的差分隐私

        """

        # 计算噪声规模

        scale = sensitivity / self.epsilon

        

        # 生成拉普拉斯噪声

        noise = np.random.laplace(0, scale)

        

        # 添加噪声

        noisy_value = true_value + noise

        

        return noisy_value

    

    def exponential_mechanism(self, items: List[Any], 

                            scores: List[float],

                            sensitivity: float) -> Any:

        """

        指数机制

        用于非数值查询的差分隐私

        """

        # 计算概率

        probabilities = []

        for score in scores:

            prob = exp(self.epsilon * score / (2 * sensitivity))

            probabilities.append(prob)

        

        # 归一化

        total_prob = sum(probabilities)

        probabilities = [p / total_prob for p in probabilities]

        

        # 根据概率选择

        selected = np.random.choice(len(items), p=probabilities)

        

        return items[selected]

    

    def private_count(self, data: List[Any], true_count: int = None) -> float:

        """隐私计数查询"""

        if true_count is None:

            true_count = len(data)

        

        # 计数查询的敏感度为1

        sensitivity = 1.0

        

        return self.laplace_mechanism(true_count, sensitivity)

    

    def private_sum(self, values: List[float]) -> float:

        """隐私求和查询"""

        true_sum = sum(values)

        

        # 假设值的范围为[0, 100]，敏感度为100

        sensitivity = 100.0

        

        return self.laplace_mechanism(true_sum, sensitivity)

    

    def private_mean(self, values: List[float]) -> float:

        """隐私均值查询"""

        # 先计算隐私计数和隐私和

        private_count = self.private_count(values)

        private_sum = self.private_sum(values)

        

        # 计算隐私均值

        if private_count > 0:

            return private_sum / private_count

        else:

            return 0.0



# 运行演示

if __name__ == "__main__":

    dp = DifferentialPrivacy(epsilon=1.0)

    

    print("="*60)

    print("差分隐私演示")

    print("="*60)

    

    # 隐私计数

    data = list(range(100))

    true_count = len(data)

    private_count = dp.private_count(data)

    

    print(f"\n计数查询:")

    print(f"  真实计数: {true_count}")

    print(f"  隐私计数: {private_count:.2f}")

    print(f"  误差: {abs(private_count - true_count):.2f}")

    

    # 隐私求和

    values = [10, 20, 30, 40, 50]

    true_sum = sum(values)

    private_sum = dp.private_sum(values)

    

    print(f"\n求和查询:")

    print(f"  真实和: {true_sum}")

    print(f"  隐私和: {private_sum:.2f}")

    print(f"  误差: {abs(private_sum - true_sum):.2f}")

    

    # 隐私均值

    true_mean = np.mean(values)

    private_mean = dp.private_mean(values)

    

    print(f"\n均值查询:")

    print(f"  真实均值: {true_mean:.2f}")

    print(f"  隐私均值: {private_mean:.2f}")

    print(f"  误差: {abs(private_mean - true_mean):.2f}")

🤝 联邦学习框架

联邦学习允许在数据不出本地的情况下进行模型训练：

# 示例8：联邦学习基础框架

"""

联邦学习基础框架



包含：

- 联邦学习架构

- 模型聚合策略

- 安全聚合协议

"""



from typing import List, Dict, Any

import numpy as np



class FederatedLearning:

    """联邦学习框架"""

    

    def __init__(self):

        """初始化联邦学习框架"""

        self.clients = []

        self.global_model = None

        print("🤝 联邦学习框架启动成功！")

    

    def federated_averaging(self, client_models: List[Dict[str, np.ndarray]],

                          client_sizes: List[int]) -> Dict[str, np.ndarray]:

        """

        联邦平均（FedAvg）算法

        根据客户端数据量加权平均模型参数

        """

        total_size = sum(client_sizes)

        

        # 初始化全局模型

        global_model = {}

        for key in client_models[0].keys():

            global_model[key] = np.zeros_like(client_models[0][key])

        

        # 加权平均

        for model, size in zip(client_models, client_sizes):

            weight = size / total_size

            for key in global_model.keys():

                global_model[key] += weight * model[key]

        

        return global_model

    

    def secure_aggregation(self, client_updates: List[Dict[str, np.ndarray]],

                          noise_scale: float = 0.1) -> Dict[str, np.ndarray]:

        """

        安全聚合（添加差分隐私噪声）

        """

        # 添加噪声保护隐私

        noisy_updates = []

        for update in client_updates:

            noisy_update = {}

            for key, value in update.items():

                noise = np.random.laplace(0, noise_scale, value.shape)

                noisy_update[key] = value + noise

            noisy_updates.append(noisy_update)

        

        # 平均聚合

        aggregated = {}

        for key in noisy_updates[0].keys():

            aggregated[key] = np.mean([update[key] for update in noisy_updates], axis=0)

        

        return aggregated



# 运行演示

if __name__ == "__main__":

    fl = FederatedLearning()

    

    # 模拟客户端模型

    client_models = [

        {"weights": np.array([1.0, 2.0, 3.0])},

        {"weights": np.array([1.5, 2.5, 3.5])},

        {"weights": np.array([0.8, 1.8, 2.8])}

    ]

    client_sizes = [100, 200, 150]

    

    # 联邦平均

    global_model = fl.federated_averaging(client_models, client_sizes)

    

    print("="*60)

    print("联邦学习演示")

    print("="*60)

    print(f"\n客户端模型:")

    for i, model in enumerate(client_models):

        print(f"  客户端{i+1}: {model['weights']}")

    

    print(f"\n全局模型（联邦平均）: {global_model['weights']}")

    

    # 安全聚合

    client_updates = [

        {"weights": np.array([0.1, 0.2, 0.3])},

        {"weights": np.array([0.15, 0.25, 0.35])},

        {"weights": np.array([0.08, 0.18, 0.28])}

    ]

    

    secure_agg = fl.secure_aggregation(client_updates, noise_scale=0.01)

    print(f"\n安全聚合结果: {secure_agg['weights']}")

---

48.4 综合项目：隐私保护推荐系统

在本章的最后，我们将综合运用所学的所有技术，构建一个完整的隐私保护推荐系统。这个系统将集成数据匿名化、模型公平性保证和隐私影响评估等功能。

# 示例9：隐私保护推荐系统

"""

隐私保护推荐系统



包含：

- 用户数据匿名化

- 模型公平性保证

- 隐私影响评估

"""



from typing import Dict, List, Any, Optional, Tuple

from datetime import datetime

import numpy as np



class PrivacyPreservingRecommendationSystem:

    """隐私保护推荐系统"""

    

    def __init__(self):

        """初始化推荐系统"""

        self.masker = None  # DataMasking实例

        self.bias_detector = None  # BiasDetector实例

        self.dp_system = None  # DifferentialPrivacy实例

        self.user_data = {}

        self.recommendations = {}

        print("🌐 隐私保护推荐系统启动成功！")

    

    def anonymize_user_data(self, user_id: str, user_data: Dict[str, Any]) -> Dict[str, Any]:

        """匿名化用户数据"""

        # 需要脱敏的字段

        sensitive_fields = ["email", "phone", "name", "address", "ip_address"]

        

        anonymized_data = user_data.copy()

        for field in sensitive_fields:

            if field in anonymized_data:

                # 这里应该调用DataMasking

                # anonymized_data[field] = self.masker.mask_data(anonymized_data[field])

                # 为演示目的，简单处理

                anonymized_data[field] = "***"

        

        return anonymized_data

    

    def generate_recommendations(self, user_id: str, 

                                user_profile: Dict[str, Any],

                                items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:

        """生成推荐（带公平性保证）"""

        # 计算推荐分数

        scores = []

        for item in items:

            score = self._calculate_score(user_profile, item)

            scores.append((item, score))

        

        # 排序

        scores.sort(key=lambda x: x[1], reverse=True)

        

        # 公平性调整

        fair_recommendations = self._apply_fairness_constraints(scores)

        

        return fair_recommendations

    

    def _calculate_score(self, user_profile: Dict[str, Any], 

                        item: Dict[str, Any]) -> float:

        """计算推荐分数"""

        # 简化的推荐算法

        score = 0.0

        

        # 基于用户兴趣匹配

        if "interests" in user_profile and "category" in item:

            if item["category"] in user_profile["interests"]:

                score += 0.5

        

        # 基于历史行为

        if "history" in user_profile:

            similar_items = sum(1 for h in user_profile["history"] 

                              if h.get("category") == item.get("category"))

            score += similar_items * 0.1

        

        # 添加随机性（差分隐私）

        noise = np.random.laplace(0, 0.1)

        score += noise

        

        return max(0.0, min(1.0, score))

    

    def _apply_fairness_constraints(self, 

                                   recommendations: List[Tuple[Dict, float]]) -> List[Dict]:

        """应用公平性约束"""

        # 确保不同类别的物品都有机会被推荐

        category_counts = {}

        fair_recommendations = []

        

        for item, score in recommendations:

            category = item.get("category", "unknown")

            category_counts[category] = category_counts.get(category, 0)

            

            # 限制每个类别的推荐数量

            if category_counts[category] < 3:  # 每个类别最多3个

                fair_recommendations.append(item)

                category_counts[category] += 1

            

            if len(fair_recommendations) >= 10:  # 最多推荐10个

                break

        

        return fair_recommendations

    

    def assess_privacy_impact(self, system_config: Dict[str, Any]) -> Dict[str, Any]:

        """隐私影响评估"""

        risk_score = 0

        risk_factors = []

        

        # 检查数据收集范围

        if system_config.get("collect_pii", False):

            risk_score += 3

            risk_factors.append("收集个人身份信息")

        

        # 检查数据共享

        if system_config.get("share_data", False):

            risk_score += 2

            risk_factors.append("数据共享")

        

        # 检查隐私保护措施

        if system_config.get("differential_privacy", False):

            risk_score -= 2

            risk_factors.append("使用差分隐私保护")

        

        if system_config.get("data_anonymization", False):

            risk_score -= 1

            risk_factors.append("数据匿名化")

        

        # 评估风险等级

        if risk_score >= 5:

            risk_level = "高风险"

        elif risk_score >= 2:

            risk_level = "中等风险"

        else:

            risk_level = "低风险"

        

        return {

            "risk_level": risk_level,

            "risk_score": risk_score,

            "risk_factors": risk_factors,

            "recommendations": self._generate_privacy_recommendations(risk_score)

        }

    

    def _generate_privacy_recommendations(self, risk_score: int) -> List[str]:

        """生成隐私保护建议"""

        recommendations = []

        

        if risk_score >= 5:

            recommendations.append("实施差分隐私保护")

            recommendations.append("加强数据匿名化")

            recommendations.append("限制数据共享范围")

        elif risk_score >= 2:

            recommendations.append("考虑使用差分隐私")

            recommendations.append("加强访问控制")

        else:

            recommendations.append("继续保持现有隐私保护措施")

        

        return recommendations



# 运行演示

if __name__ == "__main__":

    system = PrivacyPreservingRecommendationSystem()

    

    print("="*60)

    print("隐私保护推荐系统演示")

    print("="*60)

    

    # 用户数据匿名化

    user_data = {

        "user_id": "user_123",

        "name": "张三",

        "email": "zhangsan@example.com",

        "phone": "13800138000",

        "interests": ["科技", "音乐", "电影"],

        "history": [

            {"item_id": "item_1", "category": "科技"},

            {"item_id": "item_2", "category": "音乐"}

        ]

    }

    

    anonymized = system.anonymize_user_data("user_123", user_data)

    print(f"\n原始数据: {user_data}")

    print(f"匿名化数据: {anonymized}")

    

    # 生成推荐

    items = [

        {"item_id": "item_1", "category": "科技", "title": "科技产品A"},

        {"item_id": "item_2", "category": "音乐", "title": "音乐专辑B"},

        {"item_id": "item_3", "category": "电影", "title": "电影C"},

        {"item_id": "item_4", "category": "科技", "title": "科技产品D"},

        {"item_id": "item_5", "category": "音乐", "title": "音乐专辑E"}

    ]

    

    recommendations = system.generate_recommendations(

        "user_123", user_data, items

    )

    

    print(f"\n推荐结果:")

    for i, item in enumerate(recommendations, 1):

        print(f"  {i}. {item['title']} ({item['category']})")

    

    # 隐私影响评估

    system_config = {

        "collect_pii": True,

        "share_data": False,

        "differential_privacy": True,

        "data_anonymization": True

    }

    

    privacy_assessment = system.assess_privacy_impact(system_config)

    print(f"\n隐私影响评估:")

    print(f"  风险等级: {privacy_assessment['risk_level']}")

    print(f"  风险评分: {privacy_assessment['risk_score']}")

    print(f"  风险因素: {privacy_assessment['risk_factors']}")

    print(f"  建议: {privacy_assessment['recommendations']}")

---

📚 实践练习

练习1：GDPR合规检查

实现一个GDPR合规检查系统，检查数据处理活动是否符合GDPR要求。

练习2：数据脱敏系统

实现一个完整的数据脱敏系统，支持多种数据类型的脱敏。

练习3：算法偏见检测

使用偏见检测工具分析一个机器学习模型，识别并消除偏见。

练习4：差分隐私应用

实现一个使用差分隐私的统计查询系统。

练习5：隐私保护推荐系统

构建一个完整的隐私保护推荐系统，集成数据匿名化、公平性保证和隐私影响评估。

---

🤔 思考题

1. 隐私与效用的平衡：如何在保护用户隐私的同时，保持AI系统的有效性？
2. 公平性的定义：不同场景下，公平性的定义是否应该不同？如何选择合适的公平性指标？
3. 可解释性与准确性：可解释AI是否会降低模型的准确性？如何平衡？
4. 联邦学习的挑战：联邦学习面临哪些技术挑战？如何解决？
5. 隐私计算的未来：隐私计算技术的发展趋势是什么？对未来AI应用有什么影响？

---

📖 拓展阅读

1. GDPR官方文档：了解欧盟通用数据保护条例的详细要求
2. 差分隐私理论：Cynthia Dwork的差分隐私经典论文
3. 联邦学习研究：Google的联邦学习研究论文
4. AI公平性研究：Fairness in Machine Learning相关研究
5. 可解释AI研究：Explainable AI (XAI) 相关研究

---

✅ 检查清单

完成本章学习后，请检查以下内容：

知识掌握

• 理解GDPR等数据保护法规的合规要求
• 掌握数据脱敏和匿名化技术
• 了解隐私计算方法（安全多方计算、隐私集合求交等）
• 理解算法偏见的类型和检测方法
• 掌握公平性评估指标和方法
• 了解可解释AI技术
• 理解差分隐私算法
• 了解联邦学习框架

技能应用

• 能够实现GDPR合规检查系统
• 能够实现数据脱敏系统
• 能够检测和消除算法偏见
• 能够评估AI系统的公平性
• 能够实现差分隐私保护
• 能够设计和实现隐私保护AI系统

实践项目

• 完成GDPR合规检查练习
• 完成数据脱敏系统实现
• 完成算法偏见检测练习
• 完成差分隐私应用开发
• 完成隐私保护推荐系统综合项目

---

恭喜您完成第48章的学习！ 您已经掌握了数据隐私保护和AI伦理的核心技术，可以开始构建负责任、有道德的AI应用了。在下一章，我们将学习法律法规与合规性，探索如何确保AI应用符合相关法律法规要求。