AI倫理と責任あるAI開発ガイド2025：バイアス、公平性、透明性、規制の全て

┌─────────────────────────────────────────────────────┐
│           グローバルAI規制タイムライン                    │
├─────────────────────────────────────────────────────┤
│  2021.04  EU AI Act草案発表                           │
│  2023.12  EU AI Act最終合意                           │
│  2024.08  EU AI Act発効開始                           │
│  2025.02  禁止条項の適用                               │
│  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

# 代理変数（proxy）検出例
from sklearn.metrics import mutual_information_score

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)

┌──────────────────────────────────────────────┐
│        フィードバックループの例                    │
│                                              │
│  バイアスのある予測 ──▶ バイアスのある決定        │
│       ▲                     │                │
│       │                     ▼                │
│  バイアスのあるデータ ◀── バイアスのある結果      │
│                                              │
│  例：予測的治安（Predictive Policing）          │
│  - 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']

    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
    }

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)
        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:
        target_tpr = 0.8
        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

def explain_with_shap(model, X_train, X_explain, feature_names=None):
    """SHAPを使用してモデル予測を説明します。"""
    explainer = shap.Explainer(model, X_train)
    shap_values = explainer(X_explain)

    # グローバル重要度
    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}")

    # ローカル説明
    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

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

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]
    avg_attention = attention.mean(dim=1).squeeze()

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

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

    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 "既に望ましいクラスとして予測されています。"

    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）                  │
│  - 社会的スコアリングシステム                              │
│  - リアルタイム遠隔生体認証（一部例外あり）                  │
│  - 職場/学校での感情認識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,
        },
        'transparency': {
            'description': '透明性と説明可能性リスク',
            'weight': 0.20,
        },
        'safety': {
            'description': '安全性と堅牢性リスク',
            'weight': 0.25,
        },
        'privacy': {
            'description': 'プライバシーとデータ保護リスク',
            'weight': 0.15,
        },
        'accountability': {
            'description': '責任とガバナンスリスク',
            'weight': 0.15,
        }
    }

    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
        }

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

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_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)
                    })

        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)

┌─────────────────────────────────────────────────┐
│         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')
                ]
            }
        ]

        results = []
        for probe in bias_templates:
            results.append({
                'category': probe['category'],
                'status': 'NEEDS_MANUAL_REVIEW'
            })

        return results

    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')),
            '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_patterns = self._check_pii(text)
        if pii_patterns:
            results['flags'].append({
                'type': 'PII_DETECTED',
                'patterns': pii_patterns,
                'action': 'REDACT'
            })
            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_jp': r'0\d{1,4}-?\d{1,4}-?\d{4}',
        }

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

        return found

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 get_status(self):
        """全体チェックリストの状態を返します。"""
        total = sum(len(items) for items in self.CHECKLIST_ITEMS.values())
        completed = len(self.completed)

        return {
            'total': total,
            'completed': completed,
            'remaining': total - completed,
            'ready_to_deploy': completed == total,
        }

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の経験
├── 論文発表またはオープンソース貢献
└── 政策助言またはガバナンスフレームワークの設計