AI偏见检测与缓解：从数据到推理的全链路方案

AI偏见
├── 数据层偏见
│   ├── 历史偏见（Historical Bias）
│   ├── 表征偏见（Representation Bias）
│   ├── 测量偏见（Measurement Bias）
│   └── 聚合偏见（Aggregation Bias）
├── 算法层偏见
│   ├── 优化目标偏见（Objective Bias）
│   ├── 特征选择偏见（Feature Bias）
│   └── 反馈循环偏见（Feedback Loop Bias）
└── 应用层偏见
    ├── 部署上下文偏见
    ├── 用户交互偏见
    └── 解释性偏见

BIAS_TYPES = {
    "historical_bias": {
        "definition": "历史数据反映了历史上的歧视和不平等",
        "example": "用过去100年CEO数据训练的模型学习到"CEO=男性"",
        "detection": "分析训练数据中敏感属性的分布",
        "mitigation": "重新采样、数据增强、fairness constraints",
    },
    "representation_bias": {
        "definition": "某些群体在数据集中代表性不足",
        "example": "训练数据中老年人面孔占2%，但实际人口占18%",
        "detection": "子群体覆盖率分析",
        "mitigation": "过采样、合成数据、数据收集改进",
    },
    "measurement_bias": {
        "definition": "对不同群体使用不同的测量方式或标准",
        "example": "用"贷款偿还时间"作为信用指标，但对某些群体更宽松",
        "detection": "测量方式与结果的相关性分析",
        "mitigation": "标准化测量、公平测量设计",
    },
    "aggregation_bias": {
        "definition": "将不同群体混为一谈，忽视群体间真实差异",
        "example": "用统一模型预测所有地区的购房能力，忽视地区差异",
        "detection": "子群体性能差异分析",
        "mitigation": "分层建模、个性化模型",
    },
    "feedback_loop_bias": {
        "definition": "模型预测影响未来数据，形成自我强化循环",
        "example": "AI拒绝某些群体贷款，该群体违约数据少，模型继续高估风险",
        "detection": "时序数据分析、干预影响评估",
        "mitigation": "介入干预、重新平衡、多样性采样",
    }
}

import numpy as np
from dataclasses import dataclass

@dataclass
class BiasMetrics:
    """偏见检测指标"""
    demographic_parity_diff: float      # 统计奇偶性差异
    equalized_odds_diff: float          # 均等化几率差异
    disparate_impact_ratio: float       #  Disparate Impact
    correlation_ratio: float            # 相关比率

class DataBiasDetector:
    def __init__(self, sensitive_attributes: list[str]):
        self.sensitive_attrs = sensitive_attributes
    
    def analyze(self, dataset, label_col, protected_col):
        """全面分析数据偏见"""
        results = {}
        
        # 1. 描述性统计
        results["distribution"] = self.analyze_distribution(
            dataset, protected_col
        )
        
        # 2. Disparate Impact分析
        results["disparate_impact"] = self.compute_disparate_impact(
            dataset, protected_col, label_col
        )
        
        # 3. 相关性分析
        results["correlations"] = self.analyze_correlations(
            dataset, protected_col
        )
        
        # 4. 代理变量检测
        results["proxy_variables"] = self.detect_proxy_variables(
            dataset, protected_col
        )
        
        return results
    
    def compute_disparate_impact(self, df, protected_col, outcome_col):
        """
        Disparate Impact（不同影响）分析
        4/5规则：某一群体的正向结果率不应低于
        最优群体的80%
        """
        rates = {}
        for group in df[protected_col].unique():
            group_data = df[df[protected_col] == group]
            rates[group] = group_data[outcome_col].mean()
        
        max_rate = max(rates.values())
        min_rate = min(rates.values())
        
        impact_ratio = min_rate / max_rate
        
        return {
            "rates": rates,
            "impact_ratio": impact_ratio,
            "passes_4_5_rule": impact_ratio >= 0.8,
            "severity": "high" if impact_ratio < 0.5 else 
                       "medium" if impact_ratio < 0.8 else "low"
        }
    
    def detect_proxy_variables(self, df, protected_col):
        """
        检测代理变量（与受保护属性高度相关但非直接相关）
        """
        protected_binary = self.binarize_protected(df[protected_col])
        
        proxy_candidates = []
        for col in df.columns:
            if col == protected_col or df[col].dtype == 'object':
                continue
            
            # 计算相关性
            corr = np.corrcoef(protected_binary, df[col].astype(float))[0, 1]
            
            if abs(corr) > 0.7:  # 高度相关
                proxy_candidates.append({
                    "variable": col,
                    "correlation": corr,
                    "risk": "high" if abs(corr) > 0.85 else "medium"
                })
        
        return proxy_candidates

class FairnessMetrics:
    """公平性指标计算"""
    
    @staticmethod
    def demographic_parity(y_true, y_pred, sensitive_attr):
        """统计均等（Demographic Parity）"""
        groups = np.unique(sensitive_attr)
        rates = []
        for g in groups:
            mask = sensitive_attr == g
            rates.append(y_pred[mask].mean())
        return abs(rates[0] - rates[1])
    
    @staticmethod
    def equalized_odds(y_true, y_pred, sensitive_attr):
        """均等化几率（Equalized Odds）"""
        groups = np.unique(sensitive_attr)
        tpr_diffs = []
        fpr_diffs = []
        
        for g in groups:
            mask = sensitive_attr == g
            tp = ((y_true[mask] == 1) & (y_pred[mask] == 1)).sum()
            fn = ((y_true[mask] == 1) & (y_pred[mask] == 0)).sum()
            fp = ((y_true[mask] == 0) & (y_pred[mask] == 1)).sum()
            tn = ((y_true[mask] == 0) & (y_pred[mask] == 0)).sum()
            
            tpr = tp / (tp + fn) if (tp + fn) > 0 else 0
            fpr = fp / (fp + tn) if (fp + tn) > 0 else 0
            
            tpr_diffs.append(tpr)
            fpr_diffs.append(fpr)
        
        return {
            "tpr_diff": abs(tpr_diffs[0] - tpr_diffs[1]),
            "fpr_diff": abs(fpr_diffs[0] - fpr_diffs[1])
        }
    
    @staticmethod
    def calibration(y_true, y_prob, sensitive_attr, n_bins=10):
        """校准公平性"""
        groups = np.unique(sensitive_attr)
        calibrations = []
        
        for g in groups:
            mask = sensitive_attr == g
            group_metrics = []
            
            for i in range(n_bins):
                bin_mask = mask & (y_prob >= i/n_bins) & (y_prob < (i+1)/n_bins)
                if bin_mask.sum() > 0:
                    bin_prob = y_prob[bin_mask].mean()
                    bin_true = y_true[bin_mask].mean()
                    group_metrics.append({
                        "bin": i,
                        "predicted": bin_prob,
                        "actual": bin_true,
                        "diff": abs(bin_prob - bin_true)
                    })
            
            calibrations.append({g: group_metrics})
        
        return calibrations

class PreprocessingDebiasing:
    """数据预处理偏见缓解"""
    
    def resample_for_fairness(self, df, protected_col, label_col,
                               target_fairness="demographic_parity"):
        """
        重采样以平衡受保护属性
        """
        if target_fairness == "demographic_parity":
            return self.upsample_minority(df, protected_col, label_col)
        elif target_fairness == "equalized_odds":
            return self.stratified_resample(df, protected_col, label_col)
    
    def upsample_minority(self, df, protected_col, label_col):
        """上采样少数群体"""
        groups = df[protected_col].unique()
        max_size = max(df[protected_col].value_counts())
        
        resampled = []
        for g in groups:
            group_data = df[df[protected_col] == g]
            # 多次采样达到最大值
            n_copies = max_size // len(group_data)
            remainder = max_size % len(group_data)
            
            resampled.append(pd.concat([group_data] * n_copies +
                                       [group_data.sample(remainder)]))
        
        return pd.concat(resampled).sample(frac=1)
    
    def reweight_samples(self, df, protected_col, label_col):
        """
        样本重加权
        为不同群体-标签组合分配不同权重
        """
        group_label_counts = df.groupby([protected_col, label_col]).size()
        total = len(df)
        
        weights = {}
        for (g, l), count in group_label_counts.items():
            # 计算期望的比例（公平比例）
            expected = 0.5  # 假设二分类标签应该是1:1
            # 计算实际的比例
            actual = count / total
            # 权重 = 期望/实际
            expected_count = total * expected / len(groups)
            weights[(g, l)] = expected_count / count
        
        df_copy = df.copy()
        df_copy['weight'] = df_copy.apply(
            lambda x: weights.get((x[protected_col], x[label_col]), 1.0), 
            axis=1
        )
        
        return df_copy

import torch
import torch.nn as nn

class FairClassifier(nn.Module):
    """带公平性约束的分类器"""
    
    def __init__(self, input_dim, fair_constraints=None):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(input_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, 1),
            nn.Sigmoid()
        )
        self.fair_constraints = fair_constraints or []
    
    def forward(self, x):
        return self.net(x)
    
    def fairness_loss(self, outputs, labels, sensitive_attrs, 
                      constraint_type="demographic_parity"):
        """计算公平性损失项"""
        if constraint_type == "demographic_parity":
            # 最小化预测率在受保护属性上的差异
            mask_0 = sensitive_attrs == 0
            mask_1 = sensitive_attrs == 1
            pred_rate_0 = outputs[mask_0].mean()
            pred_rate_1 = outputs[mask_1].mean()
            return (pred_rate_0 - pred_rate_1).square()
        
        elif constraint_type == "equalized_odds":
            # 分别对TPR和FPR施加约束
            # ... 实现细节
            pass
        
        elif constraint_type == "individual_fairness":
            # 相似的个体应该有相似的预测
            # 需要定义"相似性"度量
            pass
        
        return 0.0

def train_fair_model(model, train_loader, sensitive_train, 
                     lambda_fair=0.1, epochs=100):
    """训练带公平性约束的模型"""
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    criterion = nn.BCELoss()
    
    for epoch in range(epochs):
        for batch_x, batch_y in train_loader:
            # 获取对应的敏感属性
            # 假设batch中包含敏感属性
            sensitive_batch = batch_sensitive[batch_x_index]
            
            optimizer.zero_grad()
            
            outputs = model(batch_x)
            class_loss = criterion(outputs, batch_y)
            fair_loss = model.fairness_loss(
                outputs.squeeze(), batch_y, sensitive_batch
            )
            
            # 总损失 = 分类损失 + λ × 公平性损失
            total_loss = class_loss + lambda_fair * fair_loss
            
            total_loss.backward()
            optimizer.step()
        
        if epoch % 10 == 0:
            print(f"Epoch {epoch}, Class Loss: {class_loss.item():.4f}, "
                  f"Fair Loss: {fair_loss.item():.4f}")

class PostProcessingDebias:
    """推理后处理偏见缓解"""
    
    def threshold_adjustment(self, y_prob, sensitive_attrs,
                             target_metric="equalized_odds"):
        """
        为不同群体设置不同的决策阈值
        以实现公平性目标
        """
        groups = np.unique(sensitive_attrs)
        thresholds = {}
        
        if target_metric == "equalized_odds":
            # 调整阈值使各群体的TPR和FPR更接近
            for g in groups:
                mask = sensitive_attrs == g
                group_probs = y_prob[mask]
                
                # 使用网格搜索找最优阈值
                best_threshold = 0.5
                best_score = float('inf')
                
                for thresh in np.linspace(0.1, 0.9, 50):
                    # 计算当前阈值下的TPR和FPR
                    # 选择使总差异最小的阈值
                    score = self._compute_odds_diff(
                        y_prob, y_true, mask, thresh
                    )
                    if score < best_score:
                        best_score = score
                        best_threshold = thresh
                
                thresholds[g] = best_threshold
        
        return thresholds
    
    def calibrate_by_group(self, y_prob, sensitive_attrs, y_true):
        """
        按群体校准预测概率
        确保预测值在各群体上都是良好校准的
        """
        from sklearn.isotonic import IsotonicRegression
        
        calibrated = y_prob.copy()
        groups = np.unique(sensitive_attrs)
        
        for g in groups:
            mask = sensitive_attrs == g
            calibrator = IsotonicRegression(out_of_bounds='clip')
            calibrator.fit(y_prob[mask], y_true[mask])
            calibrated[mask] = calibrator.predict(y_prob[mask])
        
        return calibrated

class EndToEndBiasGovernance:
    """
    端到端偏见治理框架
    覆盖数据、训练、推理全流程
    """
    
    def __init__(self):
        self.data_detector = DataBiasDetector()
        self.preprocessor = PreprocessingDebiasing()
        self.trainer = FairClassifier()
        self.postprocessor = PostProcessingDebias()
        self.audit_logger = BiasAuditLogger()
    
    def full_pipeline(self, data, sensitive_attrs, label):
        """完整偏见治理流程"""
        
        # 阶段1: 数据审计
        print("阶段1: 数据偏见审计")
        data_report = self.data_detector.analyze(
            data, label, sensitive_attrs[0]
        )
        self.audit_logger.log(data_report)
        
        # 阶段2: 数据层缓解
        print("阶段2: 数据预处理")
        if data_report["disparate_impact"]["impact_ratio"] < 0.8:
            data = self.preprocessor.resample_for_fairness(
                data, sensitive_attrs[0], label
            )
        
        # 阶段3: 训练层缓解
        print("阶段3: 公平性训练")
        fair_lambda = self._determine_fairness_weight(data_report)
        # 训练带公平性约束的模型
        
        # 阶段4: 推理层缓解
        print("阶段4: 后处理校准")
        # 应用后处理偏见缓解
        
        # 阶段5: 审计报告
        print("阶段5: 生成审计报告")
        return self.generate_audit_report()
    
    def continuous_monitoring(self, deployed_model, production_data):
        """生产环境持续监控"""
        # 定期检查模型在不同群体上的表现
        # 监控公平性指标漂移
        # 触发再训练当偏见超出容忍度
        pass