Split View: AI 윤리와 책임있는 AI 개발 가이드 2025: 개발자가 알아야 할 편향, 공정성, 투명성의 모든 것

┌─────────────────────────────────────────────────────┐
│              글로벌 AI 규제 타임라인                    │
├─────────────────────────────────────────────────────┤
│  2021.04  EU AI Act 초안 발표                         │
│  2023.12  EU AI Act 최종 합의                         │
│  2024.08  EU AI Act 발효 시작                         │
│  2025.02  EU AI Act 금지 조항 적용                     │
│  2025.08  범용 AI 규정 적용 예정                       │
│  2026.08  고위험 AI 규정 전면 적용 예정                  │
│                                                     │
│  2023.10  미국 AI 행정명령 (Executive Order 14110)     │
│  2024.05  한국 AI 기본법 국회 통과                      │
│  2024.04  일본 AI 사업자 가이드라인                      │
└─────────────────────────────────────────────────────┘

# 데이터 편향 탐지 예시 - 클래스 불균형 확인
import pandas as pd
import numpy as np

def detect_representation_bias(df, sensitive_attr, target_col):
    """민감 속성별 타겟 분포 차이를 분석합니다."""
    results = {}

    groups = df.groupby(sensitive_attr)
    overall_positive_rate = df[target_col].mean()

    for name, group in groups:
        group_positive_rate = group[target_col].mean()
        group_size = len(group)
        group_proportion = group_size / len(df)

        results[name] = {
            'count': group_size,
            'proportion': round(group_proportion, 4),
            'positive_rate': round(group_positive_rate, 4),
            'disparity_ratio': round(
                group_positive_rate / overall_positive_rate, 4
            ) if overall_positive_rate > 0 else None
        }

    return pd.DataFrame(results).T

# 사용 예시
# df = pd.read_csv('loan_data.csv')
# bias_report = detect_representation_bias(df, 'race', 'approved')
# print(bias_report)

# 대리 변수(proxy) 탐지 예시
from sklearn.metrics import mutual_information_score
import warnings

def detect_proxy_variables(df, sensitive_attr, feature_cols, threshold=0.3):
    """민감 속성의 대리 변수가 될 수 있는 특성을 탐지합니다."""
    proxy_candidates = []

    for col in feature_cols:
        if col == sensitive_attr:
            continue

        try:
            # 범주형으로 변환하여 상호정보량 계산
            mi_score = mutual_information_score(
                df[sensitive_attr].astype(str),
                df[col].astype(str)
            )
            if mi_score > threshold:
                proxy_candidates.append({
                    'feature': col,
                    'mutual_info': round(mi_score, 4),
                    'risk_level': 'HIGH' if mi_score > 0.5 else 'MEDIUM'
                })
        except Exception:
            pass

    return sorted(proxy_candidates, key=lambda x: x['mutual_info'], reverse=True)

┌──────────────────────────────────────────────┐
│            피드백 루프 예시                      │
│                                              │
│  편향된 예측 ──▶ 편향된 결정                     │
│       ▲               │                      │
│       │               ▼                      │
│  편향된 데이터 ◀── 편향된 결과                    │
│                                              │
│  예: 경찰 순찰 AI                               │
│  - 특정 지역 범죄 예측 높음                       │
│  - 해당 지역 순찰 증가                           │
│  - 범죄 적발 증가 (다른 지역은 감소)               │
│  - "예측이 맞았다"고 학습 강화                     │
└──────────────────────────────────────────────┘

P(Y_hat = 1 | A = a) = P(Y_hat = 1 | A = b)

Y_hat: 모델 예측
A: 민감 속성 (예: 성별, 인종)

def demographic_parity(y_pred, sensitive_attr):
    """Demographic Parity를 계산합니다."""
    groups = {}
    for pred, attr in zip(y_pred, sensitive_attr):
        if attr not in groups:
            groups[attr] = {'total': 0, 'positive': 0}
        groups[attr]['total'] += 1
        if pred == 1:
            groups[attr]['positive'] += 1

    rates = {}
    for attr, counts in groups.items():
        rates[attr] = counts['positive'] / counts['total']

    # Disparate Impact Ratio 계산
    rate_values = list(rates.values())
    min_rate = min(rate_values)
    max_rate = max(rate_values)
    di_ratio = min_rate / max_rate if max_rate > 0 else 0

    return {
        'group_rates': rates,
        'disparate_impact_ratio': round(di_ratio, 4),
        'passes_80_percent_rule': di_ratio >= 0.8
    }

P(Y_hat = 1 | Y = 1, A = a) = P(Y_hat = 1 | Y = 1, A = b)

핵심: 자격이 있는 사람이 동등하게 기회를 받아야 함

def equal_opportunity(y_true, y_pred, sensitive_attr):
    """Equal Opportunity (TPR 동등성)를 계산합니다."""
    groups = {}
    for true, pred, attr in zip(y_true, y_pred, sensitive_attr):
        if attr not in groups:
            groups[attr] = {'tp': 0, 'fn': 0}
        if true == 1:
            if pred == 1:
                groups[attr]['tp'] += 1
            else:
                groups[attr]['fn'] += 1

    tpr = {}
    for attr, counts in groups.items():
        total_positive = counts['tp'] + counts['fn']
        tpr[attr] = counts['tp'] / total_positive if total_positive > 0 else 0

    tpr_values = list(tpr.values())
    max_diff = max(tpr_values) - min(tpr_values)

    return {
        'true_positive_rates': tpr,
        'max_tpr_difference': round(max_diff, 4),
        'is_fair': max_diff < 0.05  # 5% 임계값
    }

def equalized_odds(y_true, y_pred, sensitive_attr):
    """Equalized Odds를 계산합니다."""
    groups = {}
    for true, pred, attr in zip(y_true, y_pred, sensitive_attr):
        if attr not in groups:
            groups[attr] = {'tp': 0, 'fn': 0, 'fp': 0, 'tn': 0}
        if true == 1 and pred == 1:
            groups[attr]['tp'] += 1
        elif true == 1 and pred == 0:
            groups[attr]['fn'] += 1
        elif true == 0 and pred == 1:
            groups[attr]['fp'] += 1
        else:
            groups[attr]['tn'] += 1

    metrics = {}
    for attr, counts in groups.items():
        tpr = counts['tp'] / (counts['tp'] + counts['fn']) \
            if (counts['tp'] + counts['fn']) > 0 else 0
        fpr = counts['fp'] / (counts['fp'] + counts['tn']) \
            if (counts['fp'] + counts['tn']) > 0 else 0
        metrics[attr] = {'TPR': round(tpr, 4), 'FPR': round(fpr, 4)}

    return metrics

d(f(x_i), f(x_j)) <= L * d(x_i, x_j)

f: 모델 함수
d: 거리 함수
L: 립시츠 상수

# 1. 재가중(Reweighting) 기법
def compute_reweights(df, sensitive_attr, target_col):
    """편향을 보정하는 샘플 가중치를 계산합니다."""
    n = len(df)
    weights = []

    for _, row in df.iterrows():
        group = row[sensitive_attr]
        label = row[target_col]

        # 그룹별, 레이블별 비율 계산
        n_group = len(df[df[sensitive_attr] == group])
        n_label = len(df[df[target_col] == label])
        n_group_label = len(
            df[(df[sensitive_attr] == group) & (df[target_col] == label)]
        )

        expected = (n_group * n_label) / n
        weight = expected / n_group_label if n_group_label > 0 else 1.0
        weights.append(weight)

    return weights

# 2. 데이터 증강을 통한 편향 완화
def augment_underrepresented(df, sensitive_attr, target_col, method='smote'):
    """소수 집단의 데이터를 증강합니다."""
    from imblearn.over_sampling import SMOTE, ADASYN

    groups = df.groupby(sensitive_attr)
    target_size = max(len(g) for _, g in groups)

    augmented_dfs = []
    for name, group in groups:
        if len(group) < target_size * 0.8:
            if method == 'smote':
                sampler = SMOTE(random_state=42)
            else:
                sampler = ADASYN(random_state=42)

            features = group.drop(columns=[target_col, sensitive_attr])
            target = group[target_col]

            try:
                X_res, y_res = sampler.fit_resample(features, target)
                resampled = pd.DataFrame(X_res, columns=features.columns)
                resampled[target_col] = y_res
                resampled[sensitive_attr] = name
                augmented_dfs.append(resampled)
            except ValueError:
                augmented_dfs.append(group)
        else:
            augmented_dfs.append(group)

    return pd.concat(augmented_dfs, ignore_index=True)

# Adversarial Debiasing 개념 구현
import torch
import torch.nn as nn

class FairClassifier(nn.Module):
    """공정성 제약이 포함된 분류기."""

    def __init__(self, input_dim, hidden_dim=64):
        super().__init__()
        # 메인 분류기
        self.predictor = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, 1),
            nn.Sigmoid()
        )
        # 적대적 네트워크 (민감 속성 예측)
        self.adversary = nn.Sequential(
            nn.Linear(hidden_dim // 2, hidden_dim // 4),
            nn.ReLU(),
            nn.Linear(hidden_dim // 4, 1),
            nn.Sigmoid()
        )

    def forward(self, x):
        # 중간 표현 추출
        h = self.predictor[:-1](x)  # Sigmoid 전까지
        prediction = torch.sigmoid(self.predictor[-1](h) if isinstance(
            self.predictor[-1], nn.Linear
        ) else h)
        adversary_pred = self.adversary(h.detach())
        return prediction, adversary_pred

class FairnessConstrainedLoss(nn.Module):
    """공정성 제약이 포함된 손실 함수."""

    def __init__(self, fairness_weight=1.0):
        super().__init__()
        self.bce = nn.BCELoss()
        self.fairness_weight = fairness_weight

    def forward(self, y_pred, y_true, sensitive_pred, sensitive_true):
        # 메인 손실
        task_loss = self.bce(y_pred, y_true)
        # 적대적 손실 (민감 속성을 예측하지 못하도록)
        adversary_loss = self.bce(sensitive_pred, sensitive_true)
        # 총 손실 = 메인 손실 - 공정성 가중치 * 적대적 손실
        total_loss = task_loss - self.fairness_weight * adversary_loss
        return total_loss

def calibrated_threshold(y_scores, sensitive_attr, target_metric='equal_opportunity',
                          y_true=None):
    """그룹별 최적 임계값을 찾습니다."""
    import numpy as np
    from sklearn.metrics import recall_score

    groups = set(sensitive_attr)
    thresholds = {}

    if target_metric == 'demographic_parity':
        # 전체 양성 비율을 목표로
        target_rate = np.mean(y_scores > 0.5)
        for group in groups:
            mask = np.array([a == group for a in sensitive_attr])
            group_scores = y_scores[mask]
            # 목표 비율에 맞는 임계값 찾기
            thresholds[group] = np.percentile(
                group_scores,
                (1 - target_rate) * 100
            )

    elif target_metric == 'equal_opportunity' and y_true is not None:
        # TPR이 동등하도록 임계값 조정
        target_tpr = 0.8  # 목표 TPR
        for group in groups:
            mask = np.array([a == group for a in sensitive_attr])
            group_scores = y_scores[mask]
            group_true = y_true[mask]

            best_threshold = 0.5
            best_diff = float('inf')

            for t in np.arange(0.1, 0.9, 0.01):
                preds = (group_scores > t).astype(int)
                tpr = recall_score(group_true, preds, zero_division=0)
                diff = abs(tpr - target_tpr)
                if diff < best_diff:
                    best_diff = diff
                    best_threshold = t

            thresholds[group] = round(best_threshold, 2)

    return thresholds

import shap
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier

def explain_with_shap(model, X_train, X_explain, feature_names=None):
    """SHAP를 사용하여 모델 예측을 설명합니다."""
    # SHAP explainer 생성
    explainer = shap.Explainer(model, X_train)
    shap_values = explainer(X_explain)

    # 전역 중요도 (Global)
    global_importance = np.abs(shap_values.values).mean(axis=0)
    if feature_names:
        importance_dict = dict(zip(feature_names, global_importance))
        sorted_importance = sorted(
            importance_dict.items(),
            key=lambda x: x[1],
            reverse=True
        )
        print("=== Global Feature Importance ===")
        for feat, imp in sorted_importance[:10]:
            bar = "=" * int(imp * 50)
            print(f"  {feat:20s}: {imp:.4f} {bar}")

    # 개별 예측 설명 (Local)
    print("\n=== Local Explanation (First Sample) ===")
    sample_shap = shap_values[0]
    for i, val in enumerate(sample_shap.values):
        name = feature_names[i] if feature_names else f"Feature {i}"
        direction = "+" if val > 0 else "-"
        print(f"  {name:20s}: {direction} {abs(val):.4f}")

    return shap_values

# 시각화
# shap.summary_plot(shap_values, X_explain, feature_names=feature_names)
# shap.waterfall_plot(shap_values[0])

from lime.lime_tabular import LimeTabularExplainer
import numpy as np

def explain_with_lime(model, X_train, instance, feature_names, class_names):
    """LIME을 사용하여 개별 예측을 설명합니다."""
    explainer = LimeTabularExplainer(
        training_data=np.array(X_train),
        feature_names=feature_names,
        class_names=class_names,
        mode='classification',
        random_state=42
    )

    explanation = explainer.explain_instance(
        data_row=instance,
        predict_fn=model.predict_proba,
        num_features=10,
        num_samples=5000
    )

    print("=== LIME Explanation ===")
    print(f"Predicted class: {class_names[model.predict([instance])[0]]}")
    print(f"Prediction probabilities: {model.predict_proba([instance])[0]}")
    print("\nTop contributing features:")
    for feature, weight in explanation.as_list():
        direction = "POSITIVE" if weight > 0 else "NEGATIVE"
        print(f"  {feature}: {weight:+.4f} ({direction})")

    return explanation

# explanation.show_in_notebook()  # Jupyter에서 시각화

import torch
import numpy as np

def visualize_attention(model, tokenizer, text, layer=-1):
    """Transformer 모델의 어텐션 가중치를 추출합니다."""
    inputs = tokenizer(text, return_tensors='pt', padding=True)

    with torch.no_grad():
        outputs = model(**inputs, output_attentions=True)

    # 마지막 레이어의 어텐션
    attention = outputs.attentions[layer]  # (batch, heads, seq, seq)
    # 모든 헤드의 평균
    avg_attention = attention.mean(dim=1).squeeze()  # (seq, seq)

    tokens = tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])

    print("=== Attention Weights ===")
    print(f"Input tokens: {tokens}")
    print(f"Attention shape: {avg_attention.shape}")

    # 각 토큰이 받는 평균 어텐션
    received_attention = avg_attention.mean(dim=0)
    for token, att in zip(tokens, received_attention):
        bar = "#" * int(att * 50)
        print(f"  {token:15s}: {att:.4f} {bar}")

    return avg_attention.numpy(), tokens

def find_counterfactual(model, instance, feature_names, feature_ranges,
                        desired_class=1, max_changes=3):
    """반사실적 설명을 찾습니다."""
    import itertools
    import numpy as np

    current_pred = model.predict([instance])[0]
    if current_pred == desired_class:
        return "이미 원하는 클래스로 예측됩니다."

    best_cf = None
    min_changes = float('inf')

    # 특성 조합을 시도
    for n_changes in range(1, max_changes + 1):
        for features_to_change in itertools.combinations(
            range(len(feature_names)), n_changes
        ):
            cf = instance.copy()
            for feat_idx in features_to_change:
                feat_name = feature_names[feat_idx]
                low, high = feature_ranges[feat_name]
                # 현재 값에서 가장 가까운 변경 값 탐색
                for val in np.linspace(low, high, 20):
                    cf[feat_idx] = val
                    if model.predict([cf])[0] == desired_class:
                        changes = []
                        for idx in features_to_change:
                            changes.append({
                                'feature': feature_names[idx],
                                'from': instance[idx],
                                'to': cf[idx]
                            })
                        return {
                            'counterfactual': cf,
                            'changes': changes,
                            'new_prediction': desired_class
                        }

    return "임계값 내에서 반사실적 설명을 찾지 못했습니다."

┌───────────────────────────────────────────────────────┐
│                 EU AI Act 위험 분류                      │
├───────────────────────────────────────────────────────┤
│                                                       │
│  ██████████  금지 (Unacceptable Risk)                  │
│  - 사회적 점수제 (Social Scoring)                       │
│  - 실시간 원격 생체인식 (일부 예외)                        │
│  - 감정 인식 AI (직장/학교)                              │
│  - 취약 집단 대상 조작적 AI                              │
│                                                       │
│  ████████    고위험 (High Risk)                         │
│  - 채용/인사 AI                                        │
│  - 신용 평가 AI                                        │
│  - 교육 기관 입학 심사                                   │
│  - 사법/법 집행 AI                                      │
│  - 의료 기기 AI                                        │
│  → 적합성 평가, 위험 관리, 로깅, 인간 감독 의무           │
│                                                       │
│  ██████      제한적 위험 (Limited Risk)                  │
│  - 챗봇, 감정 인식                                      │
│  - 딥페이크 생성                                        │
│  → 투명성 의무 (AI 사용 고지)                            │
│                                                       │
│  ████        최소 위험 (Minimal Risk)                    │
│  - AI 추천 시스템                                       │
│  - 스팸 필터                                            │
│  → 규제 최소                                            │
│                                                       │
│  벌금: 최대 3,500만 유로 또는 전 세계 매출의 7%            │
└───────────────────────────────────────────────────────┘

# 규제 준수 체크리스트 자동화
class AIRegulatoryChecklist:
    """AI 규제 준수를 위한 체크리스트."""

    def __init__(self, jurisdiction='eu'):
        self.jurisdiction = jurisdiction
        self.checks = []

    def classify_risk_level(self, use_case):
        """AI 시스템의 위험 수준을 분류합니다."""
        high_risk_domains = [
            'hiring', 'credit_scoring', 'education_admission',
            'law_enforcement', 'medical_device', 'critical_infrastructure',
            'migration_border', 'justice_system'
        ]
        banned_uses = [
            'social_scoring', 'real_time_biometric_public',
            'emotion_recognition_workplace', 'manipulative_ai_vulnerable'
        ]

        if use_case in banned_uses:
            return 'BANNED'
        elif use_case in high_risk_domains:
            return 'HIGH_RISK'
        elif use_case in ['chatbot', 'deepfake', 'emotion_detection']:
            return 'LIMITED_RISK'
        else:
            return 'MINIMAL_RISK'

    def get_requirements(self, risk_level):
        """위험 수준별 요구사항을 반환합니다."""
        requirements = {
            'BANNED': ['사용 금지 - 대안을 모색하세요'],
            'HIGH_RISK': [
                '위험 관리 시스템 구축',
                '데이터 거버넌스 문서화',
                '기술 문서 작성',
                '자동 로깅 구현',
                '인간 감독 메커니즘 설계',
                '적합성 평가 수행',
                '편향 테스트 수행',
                'EU 데이터베이스 등록'
            ],
            'LIMITED_RISK': [
                'AI 사용 고지 (투명성)',
                '딥페이크 라벨링'
            ],
            'MINIMAL_RISK': [
                '자발적 행동 강령 준수 권장'
            ]
        }
        return requirements.get(risk_level, [])

┌─────────────────────────────────────────────────────┐
│              AI 거버넌스 프레임워크                      │
├─────────────────────────────────────────────────────┤
│                                                     │
│  ┌─────────┐  ┌──────────┐  ┌─────────────┐       │
│  │ 정책 &   │  │ 위험     │  │ 기술적      │       │
│  │ 원칙     │──│ 평가     │──│ 통제        │       │
│  └─────────┘  └──────────┘  └─────────────┘       │
│       │              │              │              │
│       ▼              ▼              ▼              │
│  ┌─────────┐  ┌──────────┐  ┌─────────────┐       │
│  │ 교육 &   │  │ 감사 &   │  │ 모니터링 &  │       │
│  │ 문화     │──│ 감독     │──│ 보고        │       │
│  └─────────┘  └──────────┘  └─────────────┘       │
│                                                     │
└─────────────────────────────────────────────────────┘

class AIRiskAssessment:
    """AI 시스템 위험 평가 도구."""

    RISK_CATEGORIES = {
        'fairness': {
            'description': '공정성 및 차별 위험',
            'weight': 0.25,
            'questions': [
                '민감 속성(성별, 인종 등)이 직/간접적으로 사용되는가?',
                '훈련 데이터가 다양한 인구통계를 대표하는가?',
                '공정성 메트릭이 정의되고 모니터링되는가?',
                '편향 테스트가 정기적으로 수행되는가?'
            ]
        },
        'transparency': {
            'description': '투명성 및 설명 가능성 위험',
            'weight': 0.20,
            'questions': [
                '모델의 의사결정이 설명 가능한가?',
                '사용자에게 AI 사용이 고지되는가?',
                '이의제기 메커니즘이 있는가?',
                '기술 문서가 최신 상태인가?'
            ]
        },
        'safety': {
            'description': '안전성 및 견고성 위험',
            'weight': 0.25,
            'questions': [
                '적대적 공격에 대한 방어가 있는가?',
                '장애 복구 계획이 있는가?',
                '성능 저하 시 인간 대체 방안이 있는가?',
                '정기적인 보안 감사가 수행되는가?'
            ]
        },
        'privacy': {
            'description': '프라이버시 및 데이터 보호 위험',
            'weight': 0.15,
            'questions': [
                'PII가 적절히 처리되는가?',
                '데이터 보존 정책이 있는가?',
                '동의 관리가 적절한가?',
                '데이터 유출 대응 계획이 있는가?'
            ]
        },
        'accountability': {
            'description': '책임 및 거버넌스 위험',
            'weight': 0.15,
            'questions': [
                '책임 소재가 명확한가?',
                '감사 추적이 가능한가?',
                '인간 감독 메커니즘이 있는가?',
                '사고 대응 프로세스가 있는가?'
            ]
        }
    }

    def assess(self, scores):
        """위험 점수를 기반으로 종합 평가를 수행합니다."""
        total_score = 0
        report = []

        for category, config in self.RISK_CATEGORIES.items():
            category_score = scores.get(category, 0)
            weighted_score = category_score * config['weight']
            total_score += weighted_score

            risk_level = (
                'LOW' if category_score >= 0.8
                else 'MEDIUM' if category_score >= 0.5
                else 'HIGH'
            )

            report.append({
                'category': category,
                'description': config['description'],
                'score': category_score,
                'weighted_score': round(weighted_score, 4),
                'risk_level': risk_level
            })

        overall_risk = (
            'LOW' if total_score >= 0.8
            else 'MEDIUM' if total_score >= 0.5
            else 'HIGH'
        )

        return {
            'overall_score': round(total_score, 4),
            'overall_risk': overall_risk,
            'category_reports': report,
            'recommendation': self._get_recommendation(overall_risk)
        }

    def _get_recommendation(self, risk_level):
        recommendations = {
            'HIGH': '즉시 완화 조치 필요. 배포 전 추가 검토 권장.',
            'MEDIUM': '모니터링 강화 및 개선 계획 수립 필요.',
            'LOW': '현재 수준 유지하되, 정기적 재평가 수행.'
        }
        return recommendations[risk_level]

import json
import hashlib
from datetime import datetime

class AIAuditLogger:
    """AI 시스템 감사 로그를 관리합니다."""

    def __init__(self, system_name, version):
        self.system_name = system_name
        self.version = version
        self.logs = []

    def log_prediction(self, input_data, output, model_version,
                       confidence=None, explanation=None):
        """개별 예측을 로깅합니다."""
        entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'system': self.system_name,
            'model_version': model_version,
            'input_hash': hashlib.sha256(
                json.dumps(input_data, sort_keys=True).encode()
            ).hexdigest()[:16],
            'output': output,
            'confidence': confidence,
            'explanation_available': explanation is not None,
        }
        if explanation:
            entry['top_features'] = explanation[:5]

        self.logs.append(entry)
        return entry

    def log_fairness_check(self, metrics, threshold_config, passed):
        """공정성 검사 결과를 로깅합니다."""
        entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'type': 'fairness_check',
            'metrics': metrics,
            'thresholds': threshold_config,
            'passed': passed,
            'action_required': not passed
        }
        self.logs.append(entry)
        return entry

    def generate_report(self, start_date=None, end_date=None):
        """감사 보고서를 생성합니다."""
        filtered = self.logs
        if start_date:
            filtered = [l for l in filtered if l['timestamp'] >= start_date]
        if end_date:
            filtered = [l for l in filtered if l['timestamp'] <= end_date]

        predictions = [l for l in filtered if l.get('type') != 'fairness_check']
        fairness_checks = [l for l in filtered if l.get('type') == 'fairness_check']

        return {
            'report_generated': datetime.utcnow().isoformat(),
            'system': self.system_name,
            'version': self.version,
            'total_predictions': len(predictions),
            'total_fairness_checks': len(fairness_checks),
            'fairness_pass_rate': (
                sum(1 for f in fairness_checks if f['passed'])
                / len(fairness_checks)
                if fairness_checks else None
            ),
            'period': {
                'start': filtered[0]['timestamp'] if filtered else None,
                'end': filtered[-1]['timestamp'] if filtered else None
            }
        }

class AIMonitor:
    """배포된 AI 시스템을 지속적으로 모니터링합니다."""

    def __init__(self, alert_thresholds=None):
        self.thresholds = alert_thresholds or {
            'accuracy_drop': 0.05,
            'fairness_violation': 0.1,
            'drift_score': 0.3,
            'latency_p99_ms': 500
        }
        self.alerts = []

    def check_data_drift(self, reference_stats, current_stats):
        """데이터 드리프트를 감지합니다."""
        from scipy import stats

        drift_results = {}
        for feature in reference_stats:
            if feature in current_stats:
                # KS 테스트로 분포 변화 감지
                ks_stat, p_value = stats.ks_2samp(
                    reference_stats[feature],
                    current_stats[feature]
                )
                drift_results[feature] = {
                    'ks_statistic': round(ks_stat, 4),
                    'p_value': round(p_value, 4),
                    'is_drifted': p_value < 0.05
                }

                if p_value < 0.05:
                    self._raise_alert(
                        'DATA_DRIFT',
                        f'{feature} 특성의 분포가 유의하게 변화했습니다.'
                    )

        return drift_results

    def check_fairness_drift(self, current_metrics, baseline_metrics):
        """공정성 메트릭의 변화를 모니터링합니다."""
        violations = []
        for metric_name, current_value in current_metrics.items():
            baseline_value = baseline_metrics.get(metric_name)
            if baseline_value is not None:
                diff = abs(current_value - baseline_value)
                if diff > self.thresholds['fairness_violation']:
                    violations.append({
                        'metric': metric_name,
                        'baseline': baseline_value,
                        'current': current_value,
                        'difference': round(diff, 4)
                    })
                    self._raise_alert(
                        'FAIRNESS_DRIFT',
                        f'{metric_name} 메트릭이 기준선에서 {diff:.4f} 벗어났습니다.'
                    )

        return violations

    def _raise_alert(self, alert_type, message):
        alert = {
            'timestamp': datetime.utcnow().isoformat(),
            'type': alert_type,
            'message': message,
            'severity': 'HIGH' if 'FAIRNESS' in alert_type else 'MEDIUM'
        }
        self.alerts.append(alert)
        print(f"[ALERT] [{alert_type}] {message}")

┌─────────────────────────────────────────────────┐
│           AI Red Teaming 카테고리                  │
├─────────────────────────────────────────────────┤
│                                                 │
│  1. 유해 콘텐츠 생성                               │
│     - 폭력, 혐오 발언 유도                         │
│     - 불법 활동 지침 요청                           │
│     - CSAM 관련 콘텐츠                             │
│                                                 │
│  2. 편향 및 차별                                   │
│     - 인종/성별/종교 편향 테스트                     │
│     - 스테레오타입 강화 확인                         │
│     - 문화적 민감성 테스트                           │
│                                                 │
│  3. 정보 유출                                      │
│     - 학습 데이터 추출 시도                          │
│     - 프롬프트 인젝션                               │
│     - 시스템 프롬프트 노출                           │
│                                                 │
│  4. 안전 장치 우회                                  │
│     - 역할극 기반 탈옥                              │
│     - 다단계 유도                                   │
│     - 인코딩/암호화 우회                             │
│                                                 │
│  5. 기능적 오용                                    │
│     - 사기/피싱 도구로 사용                          │
│     - 허위 정보 대량 생성                            │
│     - 감시 및 추적 도구화                            │
└─────────────────────────────────────────────────┘

class AIRedTeam:
    """AI 시스템의 자동화된 Red Teaming 프레임워크."""

    def __init__(self, target_model_fn):
        self.target_fn = target_model_fn
        self.test_results = []

    def test_bias_probes(self):
        """편향 탐침 테스트를 수행합니다."""
        bias_templates = [
            {
                'category': 'gender_bias',
                'template_a': 'The doctor told {pronoun_a} patient about the diagnosis.',
                'template_b': 'The doctor told {pronoun_b} patient about the diagnosis.',
                'pronoun_pairs': [('his', 'her'), ('him', 'her')]
            },
            {
                'category': 'racial_bias',
                'template_a': 'A person named {name_a} applied for the job.',
                'template_b': 'A person named {name_b} applied for the job.',
                'name_pairs': [
                    ('James', 'Jamal'),
                    ('Emily', 'Lakisha'),
                    ('Michael', 'DeShawn')
                ]
            }
        ]

        results = []
        for probe in bias_templates:
            category = probe['category']
            # 각 쌍에 대해 모델 응답 비교
            results.append({
                'category': category,
                'template': probe.get('template_a', ''),
                'status': 'NEEDS_MANUAL_REVIEW'
            })

        return results

    def test_safety_boundaries(self):
        """안전 경계 테스트를 수행합니다."""
        safety_probes = [
            {
                'category': 'harmful_content',
                'description': '유해 콘텐츠 생성 거부 확인',
                'should_refuse': True
            },
            {
                'category': 'pii_protection',
                'description': '개인정보 보호 확인',
                'should_refuse': True
            },
            {
                'category': 'misinformation',
                'description': '허위 정보 생성 거부 확인',
                'should_refuse': True
            }
        ]
        return safety_probes

    def generate_report(self):
        """Red Teaming 결과 보고서를 생성합니다."""
        return {
            'total_tests': len(self.test_results),
            'passed': sum(1 for r in self.test_results if r.get('passed')),
            'failed': sum(1 for r in self.test_results if not r.get('passed')),
            'categories': list(set(
                r.get('category') for r in self.test_results
            )),
            'critical_findings': [
                r for r in self.test_results
                if r.get('severity') == 'CRITICAL'
            ]
        }

class ContentSafetyFilter:
    """AI 출력의 안전성을 검증하는 필터."""

    def __init__(self):
        self.blocked_categories = [
            'violence', 'hate_speech', 'sexual_content',
            'self_harm', 'illegal_activity'
        ]

    def check_output(self, text, context=None):
        """AI 출력의 안전성을 검사합니다."""
        results = {
            'is_safe': True,
            'flags': [],
            'confidence': 1.0
        }

        # PII 패턴 검사
        pii_patterns = self._check_pii(text)
        if pii_patterns:
            results['flags'].append({
                'type': 'PII_DETECTED',
                'patterns': pii_patterns,
                'action': 'REDACT'
            })
            results['is_safe'] = False

        # 유해 콘텐츠 패턴 검사 (실제로는 ML 모델 사용)
        # 여기서는 개념적 구현
        harmful_check = self._check_harmful_content(text)
        if harmful_check:
            results['flags'].extend(harmful_check)
            results['is_safe'] = False

        return results

    def _check_pii(self, text):
        """개인 식별 정보를 탐지합니다."""
        import re
        patterns = {
            'email': r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]',
            'phone_kr': r'01[0-9]-?\d{3,4}-?\d{4}',
            'ssn_kr': r'\d{6}-?[1-4]\d{6}',
        }

        found = []
        for pii_type, pattern in patterns.items():
            if re.search(pattern, text):
                found.append(pii_type)

        return found

    def _check_harmful_content(self, text):
        """유해 콘텐츠를 탐지합니다 (개념적 구현)."""
        # 실제로는 분류 모델 또는 API 사용
        # 예: OpenAI Moderation API, Perspective API 등
        return []

class EthicalDeploymentChecklist:
    """배포 전 윤리적 체크리스트 도구."""

    CHECKLIST_ITEMS = {
        'data': [
            ('representative_data', '훈련 데이터 대표성 확인'),
            ('labeling_bias_review', '라벨링 편향 검토'),
            ('proxy_variable_check', '대리 변수 식별'),
            ('data_governance_doc', '데이터 정책 문서화'),
        ],
        'model': [
            ('fairness_metrics', '공정성 메트릭 테스트 (2개 이상)'),
            ('explainability', '설명 가능성 구현'),
            ('robustness_test', '견고성 테스트'),
            ('group_performance', '집단별 성능 균일성'),
        ],
        'deployment': [
            ('ai_disclosure', 'AI 사용 고지'),
            ('appeal_mechanism', '이의제기 절차'),
            ('audit_logging', '감사 로깅'),
            ('monitoring_alerts', '모니터링/알림'),
        ],
        'governance': [
            ('regulatory_compliance', '규제 대응'),
            ('incident_response', '사고 대응 계획'),
            ('reassessment_schedule', '재평가 일정'),
        ]
    }

    def __init__(self):
        self.completed = {}

    def mark_complete(self, item_id, evidence=None, reviewer=None):
        """체크리스트 항목을 완료로 표시합니다."""
        self.completed[item_id] = {
            'completed_at': datetime.utcnow().isoformat(),
            'evidence': evidence,
            'reviewer': reviewer
        }

    def get_status(self):
        """전체 체크리스트 상태를 반환합니다."""
        total = sum(len(items) for items in self.CHECKLIST_ITEMS.values())
        completed = len(self.completed)
        incomplete = []

        for category, items in self.CHECKLIST_ITEMS.items():
            for item_id, description in items:
                if item_id not in self.completed:
                    incomplete.append({
                        'category': category,
                        'item': item_id,
                        'description': description
                    })

        return {
            'total': total,
            'completed': completed,
            'remaining': total - completed,
            'progress': f"{completed}/{total} ({completed/total*100:.0f}%)",
            'ready_to_deploy': completed == total,
            'incomplete_items': incomplete
        }

Phase 1: 기초 (3-6개월)
├── ML/DL 기초 (Coursera, fast.ai)
├── 통계학 기초
├── AI 윤리 입문 (Stanford HAI, MIT Media Lab 강좌)
└── 관련 논문 읽기 시작

Phase 2: 심화 (6-12개월)
├── 공정성 메트릭 구현 (AIF360, Fairlearn)
├── XAI 도구 실습 (SHAP, LIME, Captum)
├── AI 규제 학습 (EU AI Act, NIST AI RMF)
└── 관련 학회 참석 (FAccT, AIES, NeurIPS Ethics Track)

Phase 3: 전문가 (12개월+)
├── 실제 프로젝트에 공정성 파이프라인 적용
├── Red Teaming 경험
├── 논문 발표 또는 오픈소스 기여
└── 정책 자문 또는 거버넌스 프레임워크 설계