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)

# 個別予測の説明 (ウォーターフォールプロット)
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,
    )
)

# グローバル重要度 (サマリープロット)
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 (教師あり微調整): 高品質なデモンストレーションデータでベースモデルを微調整
2. 報酬モデリング: 人間の好みペア (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,   # epsilon: 小さいほど強いプライバシー
    target_delta=1e-5,    # delta: epsilon違反の確率
    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}")