모델 편향 분석

모델이 학습된 후에는 예측에서 편향을 나타내는지 평가해야 합니다. 이 장에서는 학습된 모델을 분석하여 인구통계학적 그룹 전체에 걸쳐 편향을 탐지하고 정량화하는 기술을 다룹니다.

모델 분석 소개

깨끗한 데이터가 있더라도 모델은 학습 프로세스를 통해 편향을 학습하고 증폭시킬 수 있습니다. 모든 사용자에게 공평하게 이익이 되는 공정한 AI 시스템을 보장하려면 체계적인 모델 분석이 필수적입니다.

분리된 성능 평가

모델 편향 탐지의 기초는 집계된 지표만 보는 대신 각 인구통계학적 그룹에 대해 별도로 성능을 평가하는 것입니다.

# 분리된 평가 프레임워크
import numpy as np
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

def disaggregated_evaluation(model, X_test, y_test, protected_attr):
    """
    각 그룹에 대해 별도로 모델 성능 평가
    """
    groups = np.unique(protected_attr)
    y_pred = model.predict(X_test)

    print("분리된 성능 평가")
    print("=" * 80)
    print(f"{'그룹':<15} {'정확도':<12} {'정밀도':<12} {'재현율':<12} {'F1 점수':<12}")
    print("-" * 80)

    results = {}
    for group in groups:
        mask = protected_attr == group

        acc = accuracy_score(y_test[mask], y_pred[mask])
        prec = precision_score(y_test[mask], y_pred[mask], zero_division=0)
        rec = recall_score(y_test[mask], y_pred[mask], zero_division=0)
        f1 = f1_score(y_test[mask], y_pred[mask], zero_division=0)

        results[group] = {'accuracy': acc, 'precision': prec, 'recall': rec, 'f1': f1}
        print(f"{group:<15} {acc:<12.3f} {prec:<12.3f} {rec:<12.3f} {f1:<12.3f}")

    # 불균형 계산
    print("\n" + "=" * 80)
    print("성능 불균형")
    print("-" * 80)

    for metric in ['accuracy', 'precision', 'recall', 'f1']:
        values = [r[metric] for r in results.values()]
        disparity = max(values) - min(values)
        status = "⚠️ 높음" if disparity > 0.1 else "✓ 양호"
        print(f"{metric.capitalize():<15} 불균형: {disparity:.3f} {status}")

    return results

그룹별 혼동 행렬 분석

각 그룹에 대한 혼동 행렬을 검사하면 오류가 어떻게 분산되어 있는지에 대한 패턴이 드러납니다.

# 그룹별 혼동 행렬
from sklearn.metrics import confusion_matrix

def confusion_matrix_by_group(y_true, y_pred, protected_attr):
    """
    각 인구통계학적 그룹에 대한 혼동 행렬 생성
    """
    groups = np.unique(protected_attr)

    print("그룹별 혼동 행렬")
    print("=" * 60)

    for group in groups:
        mask = protected_attr == group
        cm = confusion_matrix(y_true[mask], y_pred[mask])

        tn, fp, fn, tp = cm.ravel() if cm.size == 4 else (0, 0, 0, 0)

        print(f"\n{group}:")
        print(f"  진음성:  {tn:>6}  |  위양성: {fp:>6}")
        print(f"  위음성: {fn:>6}  |  진양성:  {tp:>6}")

        # 비율 계산
        tpr = tp / (tp + fn) if (tp + fn) > 0 else 0
        fpr = fp / (fp + tn) if (fp + tn) > 0 else 0
        tnr = tn / (tn + fp) if (tn + fp) > 0 else 0
        fnr = fn / (fn + tp) if (fn + tp) > 0 else 0

        print(f"  TPR: {tpr:.3f}  FPR: {fpr:.3f}  TNR: {tnr:.3f}  FNR: {fnr:.3f}")

    return True

결정 임계값 분석

확률적 분류기의 경우 결정 임계값이 공정성에 큰 영향을 미칠 수 있습니다. 공정성 기준을 달성하기 위해 다른 그룹에 대해 다른 임계값이 필요할 수 있습니다.

# 공정성을 위한 임계값 최적화
def analyze_thresholds(model, X_test, y_test, protected_attr, metric='tpr'):
    """
    다른 임계값이 공정성에 어떤 영향을 미치는지 분석
    """
    y_proba = model.predict_proba(X_test)[:, 1]
    groups = np.unique(protected_attr)

    thresholds = np.linspace(0, 1, 101)
    results = {group: [] for group in groups}

    for threshold in thresholds:
        y_pred = (y_proba >= threshold).astype(int)

        for group in groups:
            mask = protected_attr == group

            if metric == 'tpr':
                tp = np.sum((y_test[mask] == 1) & (y_pred[mask] == 1))
                fn = np.sum((y_test[mask] == 1) & (y_pred[mask] == 0))
                value = tp / (tp + fn) if (tp + fn) > 0 else 0
            elif metric == 'fpr':
                fp = np.sum((y_test[mask] == 0) & (y_pred[mask] == 1))
                tn = np.sum((y_test[mask] == 0) & (y_pred[mask] == 0))
                value = fp / (fp + tn) if (fp + tn) > 0 else 0

            results[group].append(value)

    # 공정성을 위한 최적 임계값 찾기
    min_disparity = float('inf')
    best_threshold = 0.5

    for i, threshold in enumerate(thresholds):
        values = [results[group][i] for group in groups]
        disparity = max(values) - min(values)
        if disparity < min_disparity:
            min_disparity = disparity
            best_threshold = threshold

    print(f"{metric.upper()} 균등성을 위한 최적 임계값: {best_threshold:.3f}")
    print(f"최소 불균형: {min_disparity:.3f}")

    return best_threshold, results

피처 중요도 및 SHAP 분석

어떤 피처가 예측을 주도하는지 이해하면 잠재적 편향 원인을 식별하는 데 도움이 됩니다.

# 피처 중요도 분석
def analyze_feature_importance_by_group(model, X_test, protected_attr, feature_names):
    """
    피처 중요도가 그룹 간에 다른지 분석
    """
    from sklearn.inspection import permutation_importance

    groups = np.unique(protected_attr)

    print("그룹별 피처 중요도")
    print("=" * 80)

    for group in groups:
        mask = protected_attr == group
        X_group = X_test[mask]
        y_group = y_test[mask]

        # 순열 중요도 계산
        perm_importance = permutation_importance(
            model, X_group, y_group, n_repeats=10, random_state=42
        )

        print(f"\n{group} - 상위 5개 가장 중요한 피처:")
        indices = perm_importance.importances_mean.argsort()[-5:][::-1]

        for idx in indices:
            importance = perm_importance.importances_mean[idx]
            print(f"  {feature_names[idx]:<30} {importance:.4f}")

    print("\n⚠️ 한 그룹에는 중요하지만 다른 그룹에는 중요하지 않은 피처를 찾으세요")
    print("   이는 차별적 대우를 나타낼 수 있습니다")

보정 분석

잘 보정된 모델의 확률 예측은 모든 그룹에 걸쳐 관찰된 빈도와 일치해야 합니다.

# 그룹별 보정 분석
from sklearn.calibration import calibration_curve

def calibration_analysis(model, X_test, y_test, protected_attr, n_bins=10):
    """
    각 그룹에 대해 별도로 보정 평가
    """
    y_proba = model.predict_proba(X_test)[:, 1]
    groups = np.unique(protected_attr)

    print("보정 분석")
    print("=" * 80)

    calibration_errors = {}

    for group in groups:
        mask = protected_attr == group

        prob_true, prob_pred = calibration_curve(
            y_test[mask],
            y_proba[mask],
            n_bins=n_bins,
            strategy='quantile'
        )

        # 보정 오차 계산 (ECE - 예상 보정 오차)
        calibration_error = np.mean(np.abs(prob_true - prob_pred))
        calibration_errors[group] = calibration_error

        print(f"\n{group}:")
        print(f"  예상 보정 오차: {calibration_error:.4f}")

        print(f"  {'예측':<12} {'실제':<12} {'차이':<12}")
        print(f"  {'-'*36}")
        for pred, true in zip(prob_pred, prob_true):
            diff = abs(pred - true)
            print(f"  {pred:<12.3f} {true:<12.3f} {diff:<12.3f}")

    # 보정 불균형 확인
    errors = list(calibration_errors.values())
    disparity = max(errors) - min(errors)

    print(f"\n보정 오차 불균형: {disparity:.4f}")
    if disparity > 0.05:
        print("⚠️ 유의미한 보정 불균형 탐지됨")

    return calibration_errors

하위 그룹 분석

편향에 특히 취약할 수 있는 특정 하위 그룹에 대한 성능을 검토합니다.

# 하위 그룹 성능 분석
def subgroup_analysis(model, X_test, y_test, protected_attrs):
    """
    교차성 하위 그룹 전체에 걸친 성능 분석
    """
    # 하위 그룹 식별자 생성
    subgroups = X_test[protected_attrs].apply(
        lambda x: '_'.join(x.astype(str)), axis=1
    )

    unique_subgroups = subgroups.unique()
    y_pred = model.predict(X_test)

    print("하위 그룹 분석")
    print("=" * 80)
    print(f"{'하위 그룹':<30} {'개수':<10} {'정확도':<12} {'정밀도':<12} {'재현율':<12}")
    print("-" * 80)

    subgroup_results = []

    for subgroup in unique_subgroups:
        mask = subgroups == subgroup
        count = mask.sum()

        if count < 10:  # 매우 작은 하위 그룹 건너뛰기
            continue

        acc = accuracy_score(y_test[mask], y_pred[mask])
        prec = precision_score(y_test[mask], y_pred[mask], zero_division=0)
        rec = recall_score(y_test[mask], y_pred[mask], zero_division=0)

        subgroup_results.append({
            'subgroup': subgroup,
            'count': count,
            'accuracy': acc,
            'precision': prec,
            'recall': rec
        })

        print(f"{subgroup:<30} {count:<10} {acc:<12.3f} {prec:<12.3f} {rec:<12.3f}")

    # 최악의 성능 하위 그룹 식별
    sorted_results = sorted(subgroup_results, key=lambda x: x['accuracy'])

    print("\n⚠️ 최악의 성능 하위 그룹:")
    for result in sorted_results[:3]:
        print(f"  {result['subgroup']}: 정확도 = {result['accuracy']:.3f}")

    return subgroup_results

반사실적 공정성 테스트

보호 속성을 변경하면 (다른 피처는 일정하게 유지) 예측이 변경되는지 테스트합니다.

# 반사실적 공정성 테스트
def counterfactual_test(model, X_test, protected_attr_col, protected_attr_values):
    """
    보호 속성만 변경될 때 예측이 변경되는지 테스트
    """
    X_counterfactual = X_test.copy()

    original_predictions = model.predict_proba(X_test)[:, 1]
    counterfactual_predictions = {}

    for value in protected_attr_values:
        X_counterfactual[protected_attr_col] = value
        counterfactual_predictions[value] = model.predict_proba(X_counterfactual)[:, 1]

    # 예측이 얼마나 자주 변경되는지 계산
    changes = 0
    total = len(X_test)

    for i in range(total):
        preds = [counterfactual_predictions[v][i] for v in protected_attr_values]
        if max(preds) - min(preds) > 0.1:  # 유의미한 변경
            changes += 1

    change_rate = changes / total

    print(f"반사실적 변경률: {change_rate:.3f}")
    if change_rate > 0.1:
        print("⚠️ 모델이 보호 속성에 민감함")
        print("   이는 잠재적 차별을 시사합니다")

    return change_rate

편향 증폭 탐지

모델이 학습 데이터에 존재하는 편향을 증폭시키는지 확인합니다.

# 편향 증폭 분석
def detect_bias_amplification(train_data, model, X_test, protected_attr):
    """
    학습 데이터 대 모델 예측의 편향 비교
    """
    # 학습 데이터의 편향 측정
    train_positive_rates = {}
    for group in train_data[protected_attr].unique():
        mask = train_data[protected_attr] == group
        train_positive_rates[group] = train_data[mask]['label'].mean()

    # 예측의 편향 측정
    y_pred = model.predict(X_test)
    test_protected = X_test[protected_attr]

    pred_positive_rates = {}
    for group in test_protected.unique():
        mask = test_protected == group
        pred_positive_rates[group] = y_pred[mask].mean()

    print("편향 증폭 분석")
    print("=" * 80)
    print(f"{'그룹':<15} {'학습 %':<15} {'예측 %':<15} {'변화':<15}")
    print("-" * 80)

    for group in train_positive_rates.keys():
        train_rate = train_positive_rates.get(group, 0) * 100
        pred_rate = pred_positive_rates.get(group, 0) * 100
        change = pred_rate - train_rate

        status = "⚠️ 증폭됨" if abs(change) > 5 else "✓ 양호"
        print(f"{group:<15} {train_rate:<15.2f} {pred_rate:<15.2f} {change:+.2f}% {status}")

    return train_positive_rates, pred_positive_rates

포괄적인 모델 감사

# 완전한 모델 편향 감사
def comprehensive_model_audit(model, X_train, y_train, X_test, y_test,
                              protected_attrs, feature_names):
    """
    학습된 모델의 완전한 편향 감사 수행
    """
    print("\n" + "=" * 80)
    print("포괄적인 모델 편향 감사")
    print("=" * 80)

    # 1. 분리된 성능
    print("\n1. 분리된 성능")
    print("-" * 80)
    for attr in protected_attrs:
        disaggregated_evaluation(model, X_test, y_test, X_test[attr])

    # 2. 혼동 행렬 분석
    print("\n2. 혼동 행렬 분석")
    print("-" * 80)
    for attr in protected_attrs:
        confusion_matrix_by_group(y_test, model.predict(X_test), X_test[attr])

    # 3. 보정 확인
    print("\n3. 보정 분석")
    print("-" * 80)
    for attr in protected_attrs:
        calibration_analysis(model, X_test, y_test, X_test[attr])

    # 4. 하위 그룹 분석
    if len(protected_attrs) > 1:
        print("\n4. 교차성 하위 그룹 분석")
        print("-" * 80)
        subgroup_analysis(model, X_test, y_test, protected_attrs)

    # 최종 권장 사항
    print("\n" + "=" * 80)
    print("감사 요약 및 권장 사항")
    print("=" * 80)
    print("1. 10% 이상의 모든 성능 불균형 검토")
    print("2. 높은 오류율을 가진 그룹 조사")
    print("3. 편향 증폭 확인")
    print("4. 공정성을 위한 임계값 조정 고려")
    print("5. 필요시 완화 기법 구현")
    print("\n弘益人間 - 모델이 모든 사용자에게 공정하게 서비스하도록 보장하십시오")
    print("=" * 80)

    return True