AI伦理与治理：构建负责任的人工智能体系

class BiasSourceAnalysis:
    sources = {
        "数据偏见": {
            "历史偏见": "训练数据反映的社会不平等",
            "采样偏见": "某些群体在数据中代表不足",
            "标注偏见": "标注者的主观偏见"
        },
        "算法偏见": {
            "特征选择": "选择了与敏感属性相关的特征",
            "模型优化": "优化整体准确率可能牺牲少数群体",
            "阈值设定": "统一阈值对不同群体影响不同"
        },
        "部署偏见": {
            "反馈循环": "AI输出影响现实，加剧原有偏见",
            "场景迁移": "在A场景训练的模型用于B场景",
            "使用者偏见": "使用者有意无意地引导输出"
        }
    }

class FairnessMetrics:
    def demographic_parity(self, y_pred, sensitive_attribute):
        """人口统计平等：不同群体的正例预测率应相同"""
        groups = set(sensitive_attribute)
        rates = {}
        for g in groups:
            mask = sensitive_attribute == g
            rates[g] = y_pred[mask].mean()
        
        # 最大差异
        disparity = max(rates.values()) - min(rates.values())
        return {"rates": rates, "disparity": disparity}
    
    def equal_opportunity(self, y_true, y_pred, sensitive_attribute):
        """机会平等：不同群体的真正例率应相同"""
        groups = set(sensitive_attribute)
        tpr = {}
        for g in groups:
            mask = (sensitive_attribute == g) & (y_true == 1)
            tpr[g] = y_pred[mask].mean()
        
        disparity = max(tpr.values()) - min(tpr.values())
        return {"tpr": tpr, "disparity": disparity}
    
    def intersectional_analysis(self, y_pred, attributes):
        """交叉分析：同时考虑多个敏感属性"""
        # 如：性别×种族×年龄
        results = {}
        for gender in attributes["gender"]:
            for race in attributes["race"]:
                mask = (attributes["gender"] == gender) & (attributes["race"] == race)
                if mask.sum() > 0:
                    results[f"{gender}_{race}"] = y_pred[mask].mean()
        return results

class BiasMitigation:
    def preprocess_reweighing(self, data, sensitive_attr, label):
        """预处理：重新加权训练样本"""
        weights = np.ones(len(data))
        
        # 计算期望概率
        p_y = {y: (label == y).mean() for y in set(label)}
        p_a = {a: (sensitive_attr == a).mean() for a in set(sensitive_attr)}
        
        for a in set(sensitive_attr):
            for y in set(label):
                mask = (sensitive_attr == a) & (label == y)
                p_ay = mask.mean()
                expected = p_a[a] * p_y[y]
                if p_ay > 0:
                    weights[mask] = expected / p_ay
        
        return weights
    
    def postprocess_threshold(self, y_scores, sensitive_attr, y_true):
        """后处理：为不同群体设定不同阈值"""
        thresholds = {}
        for group in set(sensitive_attr):
            mask = sensitive_attr == group
            # 找到使TPR-FPR差最大化的阈值
            thresholds[group] = self._optimize_threshold(
                y_scores[mask], y_true[mask]
            )
        
        y_pred = np.zeros(len(y_scores))
        for group, threshold in thresholds.items():
            mask = sensitive_attr == group
            y_pred[mask] = (y_scores[mask] >= threshold).astype(int)
        
        return y_pred

class ModelCard:
    def __init__(self):
        self.model_details = {
            "name": "SentimentAnalyzer-v2",
            "version": "2.1.0",
            "owner": "AI Team",
            "license": "Apache 2.0"
        }
        self.intended_use = {
            "primary": "产品评论情感分析",
            "users": "产品团队、客服团队",
            "out_of_scope": [
                "不应用于心理健康评估",
                "不用于司法决策"
            ]
        }
        self.training_data = {
            "sources": ["产品评论数据集", "公开情感数据集"],
            "size": "500K samples",
            "demographics": "主要为中文用户评论",
            "preprocessing": "PII脱敏、去重、平衡采样"
        }
        self.performance = {
            "overall_accuracy": 0.92,
            "by_group": {
                "电子产品评论": 0.95,
                "服装评论": 0.89,
                "食品评论": 0.91
            },
            "fairness": {
                "demographic_parity": 0.03,
                "equal_opportunity": 0.05
            }
        }
        self.limitations = [
            "对反讽/讽刺文本识别准确率较低(65%)",
            "多语言混合文本效果下降",
            "长文本(>500字)效果不稳定"
        ]

class ExplainabilityToolkit:
    def feature_importance(self, model, input_instance):
        """特征重要性解释"""
        # SHAP值
        import shap
        explainer = shap.Explainer(model)
        shap_values = explainer(input_instance)
        return shap_values
    
    def counterfactual(self, model, input_instance, target):
        """反事实解释：需要改变什么才能得到不同结果"""
        return llm.generate(f"""
        当前输入：{input_instance}
        当前预测：{model.predict(input_instance)}
        期望预测：{target}
        
        最小化修改输入，使预测变为{target}。
        解释为什么这些修改有效。
        """)
    
    def decision_trace(self, model, input_instance):
        """决策追踪：展示模型的推理过程"""
        return {
            "input_features": extract_features(input_instance),
            "attention_weights": model.get_attention(input_instance),
            "layer_activations": model.get_activations(input_instance),
            "confidence": model.get_confidence(input_instance),
            "similar_training_examples": find_similar_in_training(input_instance)
        }

class AISystemAudit:
    def audit(self, system):
        report = {
            "data_audit": self._audit_data(system),
            "model_audit": self._audit_model(system),
            "deployment_audit": self._audit_deployment(system),
            "impact_audit": self._audit_impact(system),
        }
        return report
    
    def _audit_data(self, system):
        return {
            "data_lineage": trace_data_origin(system.training_data),
            "consent_verification": check_data_consent(system.training_data),
            "bias_assessment": assess_data_bias(system.training_data),
            "freshness": check_data_freshness(system.training_data),
        }
    
    def _audit_model(self, system):
        return {
            "performance": evaluate_performance(system.model),
            "fairness": evaluate_fairness(system.model),
            "robustness": test_robustness(system.model),
            "interpretability": assess_interpretability(system.model),
        }
    
    def _audit_impact(self, system):
        return {
            "stakeholder_analysis": identify_affected_parties(system),
            "risk_assessment": assess_risks(system),
            "benefit_distribution": analyze_benefits(system),
            "feedback_mechanism": check_feedback_channels(system),
        }

class AIIncidentResponse:
    def handle(self, incident):
        # 1. 分类
        severity = self._classify(incident)
        
        # 2. 紧急措施
        if severity == "critical":
            self._pause_system(incident.system_id)
            self._notify_stakeholders(incident)
        
        # 3. 根因分析
        root_cause = self._analyze_root_cause(incident)
        
        # 4. 修复
        fix = self._develop_fix(root_cause)
        
        # 5. 事后报告
        report = self._generate_report(incident, root_cause, fix)
        
        # 6. 流程改进
        self._update_guidelines(report)
        
        return report

class DifferentialPrivacy:
    def __init__(self, epsilon=1.0):
        self.epsilon = epsilon
    
    def add_noise(self, data):
        """在数据上添加拉普拉斯噪声"""
        sensitivity = compute_sensitivity(data)
        noise = np.random.laplace(
            0, sensitivity / self.epsilon, size=data.shape
        )
        return data + noise
    
    def dp_train(self, model, data, epochs=10):
        """差分隐私训练"""
        for epoch in range(epochs):
            for batch in data.batches:
                # 梯度裁剪
                gradients = compute_gradients(model, batch)
                clipped = clip_gradients(gradients, max_norm=1.0)
                # 添加噪声
                noisy = self.add_noise(clipped)
                # 更新模型
                model.update(noisy)

class FederatedLearning:
    def train(self, server_model, clients, rounds=100):
        for round in range(rounds):
            # 1. 分发模型
            for client in clients:
                client.receive_model(server_model.state_dict())
            
            # 2. 本地训练
            client_updates = []
            for client in clients:
                update = client.local_train(epochs=5)
                client_updates.append(update)
            
            # 3. 安全聚合
            aggregated = self._secure_aggregate(client_updates)
            
            # 4. 更新全局模型
            server_model.update(aggregated)

class AIGovernanceCommittee:
    def __init__(self):
        self.members = [
            {"role": "技术负责人", "responsibility": "技术评估"},
            {"role": "法务代表", "responsibility": "合规审查"},
            {"role": "伦理顾问", "responsibility": "伦理评估"},
            {"role": "用户代表", "responsibility": "用户视角"},
            {"role": "业务负责人", "responsibility": "商业价值"}
        ]
    
    def review(self, ai_project):
        """审查AI项目"""
        criteria = {
            "technical_feasibility": self._assess_technical(ai_project),
            "ethical_compliance": self._assess_ethics(ai_project),
            "legal_compliance": self._assess_legal(ai_project),
            "social_impact": self._assess_impact(ai_project),
            "risk_level": self._assess_risk(ai_project),
        }
        
        decision = self._make_decision(criteria)
        return {
            "approved": decision["approved"],
            "conditions": decision.get("conditions", []),
            "monitoring_plan": self._create_monitoring_plan(ai_project),
            "review_date": self._next_review_date()
        }