【2026年版】AI Governance & Compliance完全ガイド：企業のリスク管理と法規制対応の実装戦略

# リスクマネジメントシステム実装例
class AIRiskManagementSystem:
    def __init__(self):
        self.risk_register = {}
        self.mitigation_measures = {}
        self.monitoring_metrics = {}

    def assess_risk(self, ai_system):
        # リスク評価ロジック
        return risk_score

    def implement_mitigation(self, risk_id, measures):
        # リスク軽減策の実装
        pass

    def monitor_continuously(self):
        # 継続的監視
        pass

# データガバナンス実装例
class DataGovernance:
    def validate_training_data(self, dataset):
        # 学習データの品質・代表性検証
        checks = [
            self.check_bias_indicators(dataset),
            self.verify_data_quality(dataset),
            self.validate_consent_compliance(dataset)
        ]
        return all(checks)

    def ensure_data_lineage(self, data_source):
        # データ系譜の記録・追跡
        pass

def statistical_parity_check(predictions, sensitive_attribute):
    """統計的パリティの検証"""
    groups = np.unique(sensitive_attribute)
    positive_rates = []

    for group in groups:
        group_mask = sensitive_attribute == group
        positive_rate = np.mean(predictions[group_mask])
        positive_rates.append(positive_rate)

    # 最大と最小の差が閾値以下かチェック
    parity_difference = max(positive_rates) - min(positive_rates)
    return parity_difference <= PARITY_THRESHOLD

def equalized_opportunity_check(y_true, y_pred, sensitive_attribute):
    """機会均等性の評価"""
    from sklearn.metrics import confusion_matrix

    fairness_metrics = {}
    groups = np.unique(sensitive_attribute)

    for group in groups:
        group_mask = sensitive_attribute == group
        y_true_group = y_true[group_mask]
        y_pred_group = y_pred[group_mask]

        cm = confusion_matrix(y_true_group, y_pred_group)
        # 真陽性率（感度）の計算
        tpr = cm[1,1] / (cm[1,1] + cm[1,0])
        fairness_metrics[group] = tpr

    return fairness_metrics

def bias_mitigation_preprocessing(X, y, sensitive_attr):
    """前処理段階でのバイアス軽減"""
    from aif360.algorithms.preprocessing import Reweighing

    # データセットの重み付け調整
    reweighing = Reweighing(
        unprivileged_groups=[{sensitive_attr: 0}],
        privileged_groups=[{sensitive_attr: 1}]
    )

    transformed_dataset = reweighing.fit_transform(dataset)
    return transformed_dataset

def bias_mitigation_postprocessing(y_pred, y_true, sensitive_attr):
    """後処理段階での公平性調整"""
    from aif360.algorithms.postprocessing import CalibratedEqOddsPostprocessing

    postprocessor = CalibratedEqOddsPostprocessing(
        unprivileged_groups=[{sensitive_attr: 0}],
        privileged_groups=[{sensitive_attr: 1}]
    )

    adjusted_predictions = postprocessor.fit_predict(
        dataset_pred, dataset_true
    )
    return adjusted_predictions

import shap
import matplotlib.pyplot as plt

def generate_shap_explanations(model, X_test, feature_names):
    """SHAP値による特徴量重要度の可視化"""

    # Explainerの初期化
    explainer = shap.Explainer(model)
    shap_values = explainer(X_test)

    # 個別予測の説明
    shap.waterfall_plot(shap_values[0])

    # 特徴量重要度サマリー
    shap.summary_plot(shap_values, X_test, feature_names=feature_names)

    # 部分依存プロット
    shap.partial_dependence_plot(
        "feature_name", model.predict, X_test, ice=False
    )

    return shap_values

from lime.lime_tabular import LimeTabularExplainer
import pandas as pd

def generate_lime_explanations(model, X_train, X_test, feature_names, instance_idx=0):
    """LIMEによる局所的説明の生成"""

    explainer = LimeTabularExplainer(
        X_train.values,
        feature_names=feature_names,
        mode='classification',
        discretize_continuous=True
    )

    # 特定インスタンスの説明
    explanation = explainer.explain_instance(
        X_test.iloc[instance_idx].values,
        model.predict_proba,
        num_features=len(feature_names)
    )

    # HTML形式で可視化
    explanation.save_to_file('explanation.html')

    return explanation

import streamlit as st
import plotly.graph_objects as go
from datetime import datetime, timedelta

def create_monitoring_dashboard():
    """AIシステム監視ダッシュボードの構築"""

    st.title("AI システム監視ダッシュボード")

    # パフォーマンス監視
    col1, col2, col3 = st.columns(3)

    with col1:
        accuracy = get_current_accuracy()
        st.metric("予測精度", f"{accuracy:.2%}", f"{accuracy - 0.85:.2%}")

    with col2:
        fairness_score = get_fairness_score()
        st.metric("公平性スコア", f"{fairness_score:.2f}", f"{fairness_score - 0.8:.2f}")

    with col3:
        drift_score = get_data_drift_score()
        st.metric("データドリフト", f"{drift_score:.3f}", f"{drift_score - 0.05:.3f}")

    # アラート一覧
    st.subheader("アクティブアラート")
    alerts = get_active_alerts()
    for alert in alerts:
        st.error(f"🚨 {alert['type']}: {alert['message']}")

class AIMonitoringSystem:
    def __init__(self):
        self.thresholds = {
            'accuracy_drop': 0.05,
            'fairness_violation': 0.1,
            'data_drift': 0.2
        }

    def monitor_performance(self, current_metrics, baseline_metrics):
        """パフォーマンス監視とアラート"""
        alerts = []

        # 精度低下の検出
        if current_metrics['accuracy'] < baseline_metrics['accuracy'] - self.thresholds['accuracy_drop']:
            alerts.append({
                'type': 'PERFORMANCE_DEGRADATION',
                'message': f"Accuracy dropped to {current_metrics['accuracy']:.2%}",
                'severity': 'HIGH'
            })

        # 公平性違反の検出
        if current_metrics['fairness_score'] < baseline_metrics['fairness_score'] - self.thresholds['fairness_violation']:
            alerts.append({
                'type': 'FAIRNESS_VIOLATION',
                'message': f"Fairness score dropped to {current_metrics['fairness_score']:.2f}",
                'severity': 'CRITICAL'
            })

        return alerts

    def detect_data_drift(self, reference_data, current_data):
        """データドリフトの検出"""
        from scipy.stats import ks_2samp

        drift_detected = False
        feature_drifts = {}

        for feature in reference_data.columns:
            ks_statistic, p_value = ks_2samp(
                reference_data[feature],
                current_data[feature]
            )

            feature_drifts[feature] = {
                'ks_statistic': ks_statistic,
                'p_value': p_value,
                'drift_detected': p_value < 0.05
            }

            if p_value < 0.05:
                drift_detected = True

        return drift_detected, feature_drifts

class FairCreditScoring:
    def __init__(self):
        self.protected_attributes = ['race', 'gender', 'age']
        self.fairness_constraints = {
            'demographic_parity': 0.05,
            'equal_opportunity': 0.05
        }

    def train_fair_model(self, X, y, sensitive_features):
        """公平性制約付きモデルの訓練"""
        from fairlearn.reductions import ExponentiatedGradient
        from fairlearn.reductions import DemographicParity

        # 公平性制約の設定
        constraint = DemographicParity()

        # 制約付き最適化
        mitigator = ExponentiatedGradient(
            estimator=LogisticRegression(),
            constraints=constraint
        )

        mitigator.fit(X, y, sensitive_features=sensitive_features)
        return mitigator

def generate_credit_explanation(model, customer_data, feature_names):
    """融資判定の説明生成"""

    # SHAP値の計算
    explainer = shap.Explainer(model)
    shap_values = explainer([customer_data])

    # 主要影響要因の抽出
    feature_impact = dict(zip(feature_names, shap_values.values[0]))
    sorted_features = sorted(
        feature_impact.items(),
        key=lambda x: abs(x[1]),
        reverse=True
    )[:5]

    explanation = {
        'decision': model.predict([customer_data])[0],
        'confidence': model.predict_proba([customer_data]).max(),
        'key_factors': sorted_features,
        'explanation_text': generate_natural_language_explanation(sorted_features)
    }

    return explanation

class FairHiringAuditor:
    def __init__(self):
        self.protected_classes = ['gender', 'race', 'age', 'disability']

    def audit_hiring_algorithm(self, candidates, predictions, demographics):
        """採用アルゴリズムの公平性監査"""
        audit_results = {}

        for protected_attr in self.protected_classes:
            if protected_attr in demographics.columns:
                # 群間での選考通過率の比較
                groups = demographics[protected_attr].unique()
                pass_rates = {}

                for group in groups:
                    group_mask = demographics[protected_attr] == group
                    group_pass_rate = predictions[group_mask].mean()
                    pass_rates[group] = group_pass_rate

                # 最小群と最大群の比率
                min_rate = min(pass_rates.values())
                max_rate = max(pass_rates.values())
                adverse_impact_ratio = min_rate / max_rate if max_rate > 0 else 0

                audit_results[protected_attr] = {
                    'pass_rates_by_group': pass_rates,
                    'adverse_impact_ratio': adverse_impact_ratio,
                    'passes_80_percent_rule': adverse_impact_ratio >= 0.8
                }

        return audit_results

def implement_hiring_bias_mitigation(training_data, protected_attributes):
    """採用データでのバイアス軽減"""

    # 1. データバランシング
    balanced_data = balance_protected_groups(training_data, protected_attributes)

    # 2. 特徴選択の見直し
    # バイアスの原因となる可能性のある特徴量を除外
    biased_features = ['university_ranking', 'previous_company_prestige']
    cleaned_features = [col for col in training_data.columns if col not in biased_features]

    # 3. 交差検証における公平性評価
    cv_fairness_scores = []
    kfold = KFold(n_splits=5, shuffle=True, random_state=42)

    for train_idx, val_idx in kfold.split(balanced_data):
        train_data = balanced_data.iloc[train_idx]
        val_data = balanced_data.iloc[val_idx]

        # モデル訓練
        model = train_model(train_data[cleaned_features])

        # 公平性評価
        val_predictions = model.predict(val_data[cleaned_features])
        fairness_score = evaluate_fairness(
            val_predictions,
            val_data[protected_attributes]
        )
        cv_fairness_scores.append(fairness_score)

    return np.mean(cv_fairness_scores)

# MLflowを使用したモデル管理例
import mlflow
import mlflow.sklearn
from mlflow.tracking import MlflowClient

def register_governed_model(model, model_name, fairness_metrics, explainability_report):
    """ガバナンス情報付きモデル登録"""

    with mlflow.start_run():
        # モデル性能メトリクス
        mlflow.log_metric("accuracy", fairness_metrics['accuracy'])
        mlflow.log_metric("precision", fairness_metrics['precision'])
        mlflow.log_metric("recall", fairness_metrics['recall'])

        # 公平性メトリクス
        mlflow.log_metric("demographic_parity", fairness_metrics['demographic_parity'])
        mlflow.log_metric("equalized_odds", fairness_metrics['equalized_odds'])

        # ガバナンス文書
        mlflow.log_artifact(explainability_report, "governance/explainability")
        mlflow.log_artifact("risk_assessment.pdf", "governance/risk_assessment")

        # モデル登録
        mlflow.sklearn.log_model(
            model,
            model_name,
            registered_model_name=model_name
        )

        # メタデータの追加
        client = MlflowClient()
        model_version = client.get_latest_versions(
            model_name,
            stages=["None"]
        )[0]

        client.update_model_version(
            model_name,
            model_version.version,
            description="Governance-compliant model with fairness validation"
        )

# Evidentlyを使用した監視システム例
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_suite import MetricSuite
from evidently.metrics import *

def setup_continuous_monitoring(reference_data, current_data, target_column):
    """継続監視システムの設定"""

    # カラムマッピング
    column_mapping = ColumnMapping()
    column_mapping.target = target_column
    column_mapping.prediction = 'prediction'
    column_mapping.numerical_features = ['feature1', 'feature2']
    column_mapping.categorical_features = ['category1', 'category2']

    # レポートの設定
    report = Report(metrics=[
        DataQualityPreset(),
        DataDriftPreset(),
        TargetDriftPreset(),
        RegressionPreset() if is_regression else ClassificationPreset()
    ])

    # レポート生成
    report.run(
        reference_data=reference_data,
        current_data=current_data,
        column_mapping=column_mapping
    )

    # HTML形式で保存
    report.save_html("monitoring_report.html")

    # アラート条件の設定
    drift_detected = report.get_metric_result('DataDriftTable').drift_detected
    if drift_detected:
        send_alert("Data drift detected in production model")

    return report