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

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

class RiskLevel(Enum):
    UNACCEPTABLE = "Unacceptable (Prohibited)"
    HIGH = "High-Risk (Strict Regulation)"
    LIMITED = "Limited Risk (Transparency Obligations)"
    MINIMAL = "Minimal Risk (Self-Regulation)"

@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 risk classifier.
    Returns (RiskLevel, list_of_applicable_obligations)
    """
    obligations = []

    # Step 1: Check for unacceptable risk
    if (system.uses_biometric and system.is_real_time and system.public_space):
        return RiskLevel.UNACCEPTABLE, ["Cease operations immediately", "Legally prohibited"]

    # Step 2: Check for high-risk domains
    HIGH_RISK_DOMAINS = {
        "healthcare", "hiring", "credit",
        "education", "judiciary", "critical_infrastructure"
    }
    if system.domain in HIGH_RISK_DOMAINS:
        obligations = [
            "Mandatory Conformity Assessment",
            "Technical documentation obligation",
            "Human oversight mechanisms required",
            "Transparency and logging requirements",
            "Bias testing and data governance",
            "Registration in EU database",
        ]
        return RiskLevel.HIGH, obligations

    # Step 3: Limited risk
    if system.interacts_with_humans or system.generates_synthetic_content:
        obligations = [
            "Disclose AI system status to users",
            "Watermark or label synthetic content",
        ]
        return RiskLevel.LIMITED, obligations

    # Step 4: Minimal risk
    return RiskLevel.MINIMAL, ["Voluntary code of conduct recommended"]


# Usage example
credit_scoring_system = AISystemProfile(
    name="Automated Credit Scoring 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"System: {credit_scoring_system.name}")
print(f"Risk Level: {risk_level.value}")
print("Obligations:")
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. Prepare data (loan approval scenario)
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)
})
# Inject artificial bias: males have higher approval probability
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. Create AIF360 dataset
aif_dataset = BinaryLabelDataset(
    df=data,
    label_names=['loan_approved'],
    protected_attribute_names=['gender'],
    favorable_label=1,
    unfavorable_label=0,
)

# 3. Measure bias
privileged_groups = [{'gender': 1}]    # male
unprivileged_groups = [{'gender': 0}]  # female

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

print("=== Original Data Bias Analysis ===")
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% rule violation (bias detected)

# 4. Preprocessing bias mitigation with 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=== After 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

# Train model
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: optimize decision thresholds per group
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)

# Measure fairness metrics
mf = MetricFrame(
    metrics={"selection_rate": selection_rate},
    y_true=y,
    y_pred=y_pred_fair,
    sensitive_features=sensitive,
)
print("\n=== Fairlearn Post-processing Results ===")
print(f"Selection rate by group:\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

# Train model
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 (tree-specific, fast)
explainer = shap.TreeExplainer(rf_model)
shap_values = explainer.shap_values(X_train)

# Individual prediction explanation (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,
    )
)

# Global importance (Summary Plot)
shap.summary_plot(shap_values[1], X_train, feature_names=feature_names)

# SHAP interaction effects
shap_interaction = explainer.shap_interaction_values(X_train[:100])
print(f"Income-CreditScore interaction SHAP: {shap_interaction[0, 0, 1]:.4f}")

import lime
import lime.lime_tabular
import numpy as np

# Create LIME explainer
lime_explainer = lime.lime_tabular.LimeTabularExplainer(
    training_data=X_train,
    feature_names=feature_names,
    class_names=['Rejected', 'Approved'],
    mode='classification',
    discretize_continuous=True,
)

# Explain individual sample
explanation = lime_explainer.explain_instance(
    data_row=X_train[0],
    predict_fn=rf_model.predict_proba,
    num_features=6,
    num_samples=1000,
)

print("=== LIME Explanation (Sample #0) ===")
for feature, weight in explanation.as_list():
    direction = "increases" if weight > 0 else "decreases"
    print(f"  {feature}: {weight:+.4f} ({direction} approval probability)")

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:
    """Standard model card generator (based on 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": ["Credit officers", "Financial regulators"],
            "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": [
            "Regular drift monitoring recommended",
            "Quarterly bias re-evaluation required",
            "Human review required for high-stakes decisions",
        ],
    }
    return model_card

card = generate_model_card(
    model_name="Personal Loan Approval Model v2.1",
    version="2.1.0",
    intended_use="Automated initial screening for personal loan applications",
    out_of_scope_uses=["Corporate loan assessment", "Insurance pricing", "Employment decisions"],
    training_data={"size": 50000, "period": "2020-2024", "source": "Internal loan history"},
    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=["Pre-2020 data not included", "Rural region underrepresentation"],
    ethical_considerations=["Final decisions must be reviewed by human officers", "Mandatory disclosure of rejection reasons"],
)
print(json.dumps(card, indent=2))

1. SFT (Supervised Fine-Tuning): Fine-tune base model on high-quality demonstration data
2. Reward Modeling: Train reward model on human preference pairs (preferred vs. rejected)
3. RL Optimization: Maximize reward with PPO algorithm (with KL divergence constraint)

import hashlib
import random

def green_red_watermark(text: str, key: str, gamma: float = 0.25) -> dict:
    """
    Green/red list watermarking based on Kirchenbauer et al. (2023).
    Uses the previous token as a seed to classify tokens as green or red,
    preferring green tokens during generation to embed a watermark.
    """
    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

# Define model
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)

# Synthetic data
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)

# Apply 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,   # epsilon: smaller = stronger privacy
    target_delta=1e-5,    # delta: probability of epsilon violation
    max_grad_norm=1.0,    # gradient clipping threshold
    epochs=10,
)

# Training loop
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"Training complete: epsilon = {epsilon:.2f}, delta = 1e-5")
print(f"Privacy guarantee: individual data contribution bounded by e^{epsilon:.2f}")

import copy
import numpy as np

def federated_averaging(global_model_weights, client_updates, client_data_sizes):
    """
    FedAvg algorithm: weighted average aggregation based on data sizes.
    """
    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 compliance checklist
gdpr_ai_checklist = {
    "Data Minimization": "Collect only the minimum data necessary for model training",
    "Purpose Limitation": "Prohibit use of training data beyond its stated purpose",
    "Data Subject Rights": "Guarantee the right to explanation for automated decisions (Article 22)",
    "Profiling Restrictions": "Human review required for significant automated profiling decisions",
    "Data Portability": "Right to receive personal data in a portable format",
    "Right to Erasure": "Remove the influence of personal data from models (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"

# Example risk register
risks = [
    AIRisk(
        risk_id="RISK-001",
        description="Gender bias in credit model leading to discriminatory loan rejections",
        severity=Severity.HIGH,
        likelihood=Likelihood.POSSIBLE,
        affected_groups=["Women", "Non-binary individuals"],
        mitigation="Reweighing + quarterly disparate impact monitoring",
        owner="AI Ethics Team",
    ),
    AIRisk(
        risk_id="RISK-002",
        description="Inability to explain model decisions violating GDPR Article 22",
        severity=Severity.CRITICAL,
        likelihood=Likelihood.LIKELY,
        affected_groups=["All loan applicants"],
        mitigation="Build SHAP-based decision explanation system",
        owner="Compliance Team",
    ),
]

print("=== AI Risk Register ===")
for risk in risks:
    print(f"\n[{risk.risk_id}] {risk.description}")
    print(f"  Risk Level: {risk.risk_level} (Score: {risk.risk_score})")
    print(f"  Mitigation: {risk.mitigation}")