AI 거버넌스 & 책임감 있는 AI: EU AI Act, XAI, 편향 감지, AI 안전 기술까지

from dataclasses import dataclass
from enum import Enum
from typing import List

class RiskLevel(Enum):
    UNACCEPTABLE = "용납 불가 (금지)"
    HIGH = "고위험 (엄격 규제)"
    LIMITED = "제한적 위험 (투명성 의무)"
    MINIMAL = "최소 위험 (자율 규제)"

@dataclass
class AISystemProfile:
    name: str
    uses_biometric: bool
    is_real_time: bool
    public_space: bool
    domain: str  # healthcare, hiring, credit, education, judiciary, infrastructure
    interacts_with_humans: bool
    generates_synthetic_content: bool

def classify_eu_ai_act_risk(system: AISystemProfile) -> tuple[RiskLevel, List[str]]:
    """
    EU AI Act 위험 등급 분류기.
    Returns (RiskLevel, list_of_applicable_obligations)
    """
    obligations = []

    # 1단계: 용납 불가 위험 확인
    if (system.uses_biometric and system.is_real_time and system.public_space):
        return RiskLevel.UNACCEPTABLE, ["즉시 사용 중단", "법적 금지 대상"]

    # 2단계: 고위험 도메인 확인
    HIGH_RISK_DOMAINS = {
        "healthcare", "hiring", "credit",
        "education", "judiciary", "critical_infrastructure"
    }
    if system.domain in HIGH_RISK_DOMAINS:
        obligations = [
            "적합성 평가(Conformity Assessment) 필수",
            "기술 문서화 의무",
            "인간 감독(Human Oversight) 체계 구축",
            "투명성 및 로깅 요건",
            "편향 테스트 및 데이터 거버넌스",
            "EU 데이터베이스 등록",
        ]
        return RiskLevel.HIGH, obligations

    # 3단계: 제한적 위험
    if system.interacts_with_humans or system.generates_synthetic_content:
        obligations = [
            "AI 시스템임을 사용자에게 고지",
            "합성 콘텐츠 워터마킹 또는 표시",
        ]
        return RiskLevel.LIMITED, obligations

    # 4단계: 최소 위험
    return RiskLevel.MINIMAL, ["자발적 행동 강령 권장"]


# 사용 예시
credit_scoring_system = AISystemProfile(
    name="자동 신용 평가 AI",
    uses_biometric=False,
    is_real_time=False,
    public_space=False,
    domain="credit",
    interacts_with_humans=False,
    generates_synthetic_content=False,
)

risk_level, obligations = classify_eu_ai_act_risk(credit_scoring_system)
print(f"시스템: {credit_scoring_system.name}")
print(f"위험 등급: {risk_level.value}")
print("의무 사항:")
for ob in obligations:
    print(f"  - {ob}")

import numpy as np
import pandas as pd
from aif360.datasets import BinaryLabelDataset
from aif360.metrics import BinaryLabelDatasetMetric, ClassificationMetric
from aif360.algorithms.preprocessing import Reweighing
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler

# 1. 데이터 준비 (대출 승인 시나리오)
np.random.seed(42)
n = 1000
data = pd.DataFrame({
    'income': np.random.normal(50000, 20000, n).clip(10000, 150000),
    'credit_score': np.random.normal(680, 100, n).clip(300, 850),
    'age': np.random.randint(20, 70, n),
    'gender': np.random.choice([0, 1], n, p=[0.5, 0.5]),  # 0=female, 1=male
    'loan_approved': np.zeros(n, dtype=int)
})
# 인위적 편향 주입: 남성이 승인될 확률이 더 높게 설정
prob = 0.3 + 0.2 * data['gender'] + 0.3 * (data['credit_score'] > 700).astype(int)
data['loan_approved'] = (np.random.random(n) < prob).astype(int)

# 2. AIF360 데이터셋 생성
aif_dataset = BinaryLabelDataset(
    df=data,
    label_names=['loan_approved'],
    protected_attribute_names=['gender'],
    favorable_label=1,
    unfavorable_label=0,
)

# 3. 편향 측정
privileged_groups = [{'gender': 1}]    # 남성
unprivileged_groups = [{'gender': 0}]  # 여성

dataset_metric = BinaryLabelDatasetMetric(
    aif_dataset,
    unprivileged_groups=unprivileged_groups,
    privileged_groups=privileged_groups,
)

print("=== 원본 데이터 편향 분석 ===")
print(f"Disparate Impact: {dataset_metric.disparate_impact():.4f}")
print(f"Statistical Parity Difference: {dataset_metric.statistical_parity_difference():.4f}")
# Disparate Impact < 0.8 → 80% 규칙 위반 (편향 존재)

# 4. Reweighing으로 전처리 편향 완화
rw = Reweighing(
    unprivileged_groups=unprivileged_groups,
    privileged_groups=privileged_groups,
)
dataset_reweighed = rw.fit_transform(aif_dataset)

# 완화 후 측정
metric_reweighed = BinaryLabelDatasetMetric(
    dataset_reweighed,
    unprivileged_groups=unprivileged_groups,
    privileged_groups=privileged_groups,
)
print("\n=== Reweighing 후 편향 분석 ===")
print(f"Disparate Impact: {metric_reweighed.disparate_impact():.4f}")
print(f"Statistical Parity Difference: {metric_reweighed.statistical_parity_difference():.4f}")

from fairlearn.postprocessing import ThresholdOptimizer
from fairlearn.metrics import MetricFrame, selection_rate, demographic_parity_difference
from sklearn.ensemble import GradientBoostingClassifier

# 모델 훈련
X = data[['income', 'credit_score', 'age']].values
y = data['loan_approved'].values
sensitive = data['gender'].values

scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

base_model = GradientBoostingClassifier(n_estimators=100, random_state=42)
base_model.fit(X_scaled, y)

# ThresholdOptimizer: 집단별 결정 임계값 최적화
postprocess_est = ThresholdOptimizer(
    estimator=base_model,
    constraints="demographic_parity",
    predict_method="predict_proba",
    objective="balanced_accuracy_score",
)
postprocess_est.fit(X_scaled, y, sensitive_features=sensitive)

y_pred_fair = postprocess_est.predict(X_scaled, sensitive_features=sensitive)

# 공정성 지표 측정
mf = MetricFrame(
    metrics={"selection_rate": selection_rate},
    y_true=y,
    y_pred=y_pred_fair,
    sensitive_features=sensitive,
)
print("\n=== Fairlearn 후처리 결과 ===")
print(f"집단별 선택률:\n{mf.by_group}")
print(f"Demographic Parity Difference: {demographic_parity_difference(y, y_pred_fair, sensitive_features=sensitive):.4f}")

import shap
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification

# 모델 훈련
X_train, y_train = make_classification(
    n_samples=500, n_features=8, n_informative=5, random_state=42
)
feature_names = [
    'income', 'credit_score', 'age', 'debt_ratio',
    'employment_years', 'num_accounts', 'late_payments', 'loan_amount'
]

rf_model = RandomForestClassifier(n_estimators=100, random_state=42)
rf_model.fit(X_train, y_train)

# SHAP TreeExplainer (트리 기반 모델 전용, 빠름)
explainer = shap.TreeExplainer(rf_model)
shap_values = explainer.shap_values(X_train)

# 개별 예측 설명 (Waterfall Plot)
sample_idx = 0
shap.waterfall_plot(
    shap.Explanation(
        values=shap_values[1][sample_idx],
        base_values=explainer.expected_value[1],
        data=X_train[sample_idx],
        feature_names=feature_names,
    )
)

# 전역 중요도 (Summary Plot)
shap.summary_plot(shap_values[1], X_train, feature_names=feature_names)

# SHAP 상호작용 효과
shap_interaction = explainer.shap_interaction_values(X_train[:100])
print(f"Income-CreditScore 상호작용 SHAP: {shap_interaction[0, 0, 1]:.4f}")

import lime
import lime.lime_tabular
import numpy as np

# LIME 설명기 생성
lime_explainer = lime.lime_tabular.LimeTabularExplainer(
    training_data=X_train,
    feature_names=feature_names,
    class_names=['거절', '승인'],
    mode='classification',
    discretize_continuous=True,
)

# 개별 샘플 설명
explanation = lime_explainer.explain_instance(
    data_row=X_train[0],
    predict_fn=rf_model.predict_proba,
    num_features=6,
    num_samples=1000,
)

print("=== LIME 설명 (샘플 #0) ===")
for feature, weight in explanation.as_list():
    direction = "증가" if weight > 0 else "감소"
    print(f"  {feature}: {weight:+.4f} ({direction}에 기여)")

explanation.show_in_notebook(show_table=True)

import json
from datetime import datetime

def generate_model_card(
    model_name: str,
    version: str,
    intended_use: str,
    out_of_scope_uses: list,
    training_data: dict,
    evaluation_results: dict,
    fairness_analysis: dict,
    limitations: list,
    ethical_considerations: list,
) -> dict:
    """표준 모델 카드 생성기 (Mitchell et al. 2019 기반)."""
    model_card = {
        "model_details": {
            "name": model_name,
            "version": version,
            "date": datetime.now().strftime("%Y-%m-%d"),
            "type": "Binary Classifier",
            "paper": "https://arxiv.org/abs/1810.03993",
        },
        "intended_use": {
            "primary_uses": intended_use,
            "primary_users": ["신용 심사 담당자", "금융 규제 기관"],
            "out_of_scope_uses": out_of_scope_uses,
        },
        "factors": {
            "relevant_factors": ["gender", "age_group", "income_bracket"],
            "evaluation_factors": ["demographic_parity", "equal_opportunity"],
        },
        "metrics": {
            "performance_measures": evaluation_results,
            "decision_thresholds": {"default": 0.5, "high_precision": 0.7},
        },
        "training_data": training_data,
        "fairness_analysis": fairness_analysis,
        "limitations": limitations,
        "ethical_considerations": ethical_considerations,
        "caveats_recommendations": [
            "정기적인 드리프트 모니터링 권장",
            "분기별 편향 재평가 필수",
            "고위험 결정 시 인간 검토 병행",
        ],
    }
    return model_card

card = generate_model_card(
    model_name="신용 대출 승인 모델 v2.1",
    version="2.1.0",
    intended_use="개인 신용 대출 신청 초기 심사 자동화",
    out_of_scope_uses=["기업 대출 심사", "보험료 산정", "고용 결정"],
    training_data={"size": 50000, "period": "2020-2024", "source": "내부 대출 이력"},
    evaluation_results={"accuracy": 0.84, "AUC": 0.91, "F1": 0.82},
    fairness_analysis={
        "demographic_parity_diff": 0.03,
        "equal_opportunity_diff": 0.02,
        "disparate_impact": 0.96,
    },
    limitations=["2020년 이전 데이터 미포함", "농촌 지역 대표성 부족"],
    ethical_considerations=["최종 결정은 반드시 인간 담당자가 검토", "거절 사유 의무 고지"],
)
print(json.dumps(card, ensure_ascii=False, indent=2))

1. SFT (Supervised Fine-Tuning): 고품질 데모 데이터로 기반 모델 미세 조정
2. Reward Modeling: 인간 선호도 쌍(preferred vs rejected)으로 보상 모델 훈련
3. RL 최적화: PPO 알고리즘으로 보상 극대화 (KL 발산 제약 포함)

import hashlib
import random

def green_red_watermark(text: str, key: str, gamma: float = 0.25) -> dict:
    """
    Kirchenbauer et al. (2023) 방식의 그린/레드 리스트 워터마킹.
    토큰 시퀀스에서 이전 토큰을 시드로 사용해 토큰을 그린/레드로 분류,
    그린 토큰을 선호하여 생성 시 워터마크를 삽입.
    """
    words = text.split()
    green_count = 0
    total = len(words)

    for i, word in enumerate(words):
        prev_token = words[i - 1] if i > 0 else "<s>"
        seed = int(hashlib.sha256(f"{key}{prev_token}".encode()).hexdigest(), 16) % (2**32)
        random.seed(seed)
        is_green = random.random() > (1 - gamma)
        if is_green:
            green_count += 1

    z_score = (green_count - gamma * total) / ((gamma * (1 - gamma) * total) ** 0.5 + 1e-9)
    return {
        "green_token_ratio": green_count / total,
        "z_score": z_score,
        "is_watermarked": z_score > 4.0,
    }

import torch
import torch.nn as nn
from opacus import PrivacyEngine
from torch.utils.data import DataLoader, TensorDataset

# 모델 정의
class SimpleNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc = nn.Sequential(
            nn.Linear(10, 64),
            nn.ReLU(),
            nn.Linear(64, 2),
        )
    def forward(self, x):
        return self.fc(x)

# 합성 데이터
X = torch.randn(1000, 10)
y = torch.randint(0, 2, (1000,))
dataset = TensorDataset(X, y)
loader = DataLoader(dataset, batch_size=64, shuffle=True)

model = SimpleNet()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

# Opacus PrivacyEngine 적용
privacy_engine = PrivacyEngine()
model, optimizer, loader = privacy_engine.make_private_with_epsilon(
    module=model,
    optimizer=optimizer,
    data_loader=loader,
    target_epsilon=1.0,   # 엡실론: 작을수록 강한 프라이버시
    target_delta=1e-5,    # 델타: 엡실론 위반 확률
    max_grad_norm=1.0,    # 클리핑 임계값
    epochs=10,
)

# 훈련 루프
criterion = nn.CrossEntropyLoss()
for epoch in range(3):
    for batch_X, batch_y in loader:
        optimizer.zero_grad()
        outputs = model(batch_X)
        loss = criterion(outputs, batch_y)
        loss.backward()
        optimizer.step()

epsilon = privacy_engine.get_epsilon(delta=1e-5)
print(f"훈련 완료: epsilon = {epsilon:.2f}, delta = 1e-5")
print(f"프라이버시 보장: 개별 데이터 기여도가 e^{epsilon:.2f}배 이내로 제한")

import copy
import numpy as np

def federated_averaging(global_model_weights, client_updates, client_data_sizes):
    """
    FedAvg 알고리즘: 데이터 크기 가중치 기반 평균 집계.
    """
    total_data = sum(client_data_sizes)
    averaged_weights = {}

    for key in global_model_weights.keys():
        weighted_sum = sum(
            client_updates[i][key] * (client_data_sizes[i] / total_data)
            for i in range(len(client_updates))
        )
        averaged_weights[key] = weighted_sum

    return averaged_weights

# GDPR AI 적용 체크리스트
gdpr_ai_checklist = {
    "데이터 최소화": "모델 훈련에 필요한 최소한의 데이터만 수집",
    "목적 제한": "훈련 데이터를 명시된 목적 외에 사용 금지",
    "정보 주체 권리": "자동화 결정에 대한 설명 요구권 보장 (Article 22)",
    "프로파일링 제한": "자동화 프로파일링 기반 중요 결정에 인간 검토 의무",
    "데이터 이동권": "개인이 자신의 데이터를 포터블 형식으로 받을 권리",
    "삭제권": "모델에서 개인 데이터 영향 제거 (Machine Unlearning)",
}

for right, description in gdpr_ai_checklist.items():
    print(f"[GDPR] {right}: {description}")

from dataclasses import dataclass, field
from typing import List
from enum import IntEnum

class Severity(IntEnum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

class Likelihood(IntEnum):
    RARE = 1
    UNLIKELY = 2
    POSSIBLE = 3
    LIKELY = 4

@dataclass
class AIRisk:
    risk_id: str
    description: str
    severity: Severity
    likelihood: Likelihood
    affected_groups: List[str]
    mitigation: str
    owner: str
    residual_risk: str = "TBD"

    @property
    def risk_score(self) -> int:
        return self.severity * self.likelihood

    @property
    def risk_level(self) -> str:
        score = self.risk_score
        if score >= 12:
            return "CRITICAL"
        elif score >= 8:
            return "HIGH"
        elif score >= 4:
            return "MEDIUM"
        return "LOW"

# 예시 위험 등록부
risks = [
    AIRisk(
        risk_id="RISK-001",
        description="신용 모델의 성별 편향으로 인한 대출 거절 차별",
        severity=Severity.HIGH,
        likelihood=Likelihood.POSSIBLE,
        affected_groups=["여성", "비이진 성별"],
        mitigation="Reweighing + 분기별 disparate impact 모니터링",
        owner="AI 윤리팀",
    ),
    AIRisk(
        risk_id="RISK-002",
        description="모델 결정에 대한 설명 불가로 GDPR Article 22 위반",
        severity=Severity.CRITICAL,
        likelihood=Likelihood.LIKELY,
        affected_groups=["모든 대출 신청자"],
        mitigation="SHAP 기반 결정 설명 시스템 구축",
        owner="컴플라이언스팀",
    ),
]

print("=== AI 위험 등록부 ===")
for risk in risks:
    print(f"\n[{risk.risk_id}] {risk.description}")
    print(f"  위험 수준: {risk.risk_level} (점수: {risk.risk_score})")
    print(f"  완화 조치: {risk.mitigation}")