AI Safety & Alignment 完全ガイド 2025: 責任あるAI、RLHF、Constitutional AI、レッドチーム

# 例：ゲームAIがスコアを最大化するよう学習
# 意図：ゲームを上手にプレイすること
# 実際：バグを利用して無限スコアを獲得

class SpecificationGamingExample:
    """
    報酬関数: score = enemies_defeated * 10
    意図: 敵を倒しながらゲームを進行
    実際の行動: 敵が無限リスポーンする隅で繰り返し撃破
    """

    def reward_function_v1(self, state):
        # 問題のある報酬関数
        return state.enemies_defeated * 10

    def reward_function_v2(self, state):
        # 改善された報酬関数 - 複数の目標を反映
        progress_reward = state.level_progress * 50
        combat_reward = state.enemies_defeated * 10
        exploration_reward = state.areas_discovered * 20
        time_penalty = -state.time_elapsed * 0.1
        return progress_reward + combat_reward + exploration_reward + time_penalty

[Base Optimizer（学習アルゴリズム）]
    |
    v
[Learned Model]  <-- ここにmesa-optimizerが形成される可能性
    |
    v
[Mesa-Objective]  <-- Base Objectiveと異なる可能性あり！

# 例え：
# - Base Objective: 「ユーザーに役立つ回答を生成せよ」
# - Mesa-Objective: 「評価時のみ良い回答をし、実際のデプロイでは異なる行動」
# これをDeceptive Alignmentと呼ぶ

Outer Alignment: 人間の意図 -> 報酬関数
  （人間が望むことを正確に報酬関数で表現できるか？）

Inner Alignment: 報酬関数 -> 学習されたモデルの実際の目標
  （学習されたモデルが実際に報酬関数を最適化するか？）

両方の段階で失敗する可能性がある：
- Outer misalignment: 不適切な報酬関数設計
- Inner misalignment: Mesa-optimizationによる目標不一致

Phase 1: Supervised Fine-Tuning (SFT)
  事前学習モデル + 高品質デモデータ -> SFTモデル

Phase 2: Reward Model Training
  SFTモデル出力ペアに対する人間の選好収集 -> Reward Model

Phase 3: PPO (Proximal Policy Optimization)
  SFTモデル + Reward Model -> PPOで最適化 -> 整合されたモデル

# Phase 1: SFT (Supervised Fine-Tuning)
from transformers import AutoModelForCausalLM, TrainingArguments, Trainer

def train_sft_model(base_model_name, demonstration_dataset):
    """
    高品質な人間作成の応答で微調整
    """
    model = AutoModelForCausalLM.from_pretrained(base_model_name)

    training_args = TrainingArguments(
        output_dir="./sft_model",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=8,
        learning_rate=2e-5,
        warmup_ratio=0.1,
        fp16=True,
        logging_steps=10,
    )

    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=demonstration_dataset,
    )
    trainer.train()
    return model


# Phase 2: Reward Model Training
import torch
import torch.nn as nn

class RewardModel(nn.Module):
    """
    人間の選好を学習する報酬モデル
    - 入力: (prompt, response)ペア
    - 出力: スカラー報酬スコア
    """

    def __init__(self, base_model):
        super().__init__()
        self.backbone = base_model
        self.reward_head = nn.Linear(
            base_model.config.hidden_size, 1
        )

    def forward(self, input_ids, attention_mask):
        outputs = self.backbone(
            input_ids=input_ids,
            attention_mask=attention_mask,
            output_hidden_states=True,
        )
        last_hidden = outputs.hidden_states[-1][:, -1, :]
        reward = self.reward_head(last_hidden)
        return reward

    def compute_preference_loss(self, chosen_reward, rejected_reward):
        """
        Bradley-Terryモデルベースの選好損失
        chosen応答がrejectedより高い報酬を得るよう学習
        """
        return -torch.log(
            torch.sigmoid(chosen_reward - rejected_reward)
        ).mean()


# Phase 3: PPO Training
class PPOTrainer:
    """
    Proximal Policy Optimizationによる整合
    """

    def __init__(self, policy_model, reward_model, ref_model):
        self.policy = policy_model
        self.reward = reward_model
        self.ref = ref_model
        self.kl_coeff = 0.02

    def compute_rewards(self, prompts, responses):
        rm_scores = self.reward(prompts, responses)
        policy_logprobs = self.policy.log_probs(prompts, responses)
        ref_logprobs = self.ref.log_probs(prompts, responses)
        kl_penalty = self.kl_coeff * (policy_logprobs - ref_logprobs)
        return rm_scores - kl_penalty

    def train_step(self, batch):
        old_logprobs = batch["old_logprobs"]
        new_logprobs = self.policy.log_probs(
            batch["prompts"], batch["responses"]
        )
        ratio = torch.exp(new_logprobs - old_logprobs)
        advantages = batch["advantages"]

        clip_range = 0.2
        surr1 = ratio * advantages
        surr2 = torch.clamp(ratio, 1 - clip_range, 1 + clip_range) * advantages
        loss = -torch.min(surr1, surr2).mean()
        return loss

RLHFパイプライン:
  SFT -> Reward Model -> PPO -> 整合モデル（3段階、複雑）

DPOパイプライン:
  SFT -> DPO（選好データで直接最適化）-> 整合モデル（2段階、単純）

import torch
import torch.nn.functional as F

class DPOTrainer:
    """
    Direct Preference Optimization
    - 報酬モデル学習なしで直接選好最適化
    - RLHFと数学的に等価だが実装がはるかに簡単
    """

    def __init__(self, model, ref_model, beta=0.1):
        self.model = model
        self.ref_model = ref_model
        self.beta = beta

    def dpo_loss(self, chosen_ids, rejected_ids, prompt_ids):
        """
        DPO Loss計算
        """
        chosen_logprobs = self.model.log_probs(prompt_ids, chosen_ids)
        rejected_logprobs = self.model.log_probs(prompt_ids, rejected_ids)

        with torch.no_grad():
            ref_chosen_logprobs = self.ref_model.log_probs(
                prompt_ids, chosen_ids
            )
            ref_rejected_logprobs = self.ref_model.log_probs(
                prompt_ids, rejected_ids
            )

        chosen_ratio = chosen_logprobs - ref_chosen_logprobs
        rejected_ratio = rejected_logprobs - ref_rejected_logprobs

        logits = self.beta * (chosen_ratio - rejected_ratio)
        loss = -F.logsigmoid(logits).mean()

        chosen_rewards = self.beta * chosen_ratio.detach()
        rejected_rewards = self.beta * rejected_ratio.detach()
        reward_margin = (chosen_rewards - rejected_rewards).mean()

        return loss, reward_margin

段階1: Supervised段階（レッドチーム + 自己批判）
  1. モデルに有害なプロンプトを入力
  2. モデルが初期応答を生成
  3. 「憲法」に基づいて自己批判（self-critique）
  4. 修正された応答を生成
  5. (prompt, 修正応答)ペアでSFT

段階2: RL段階（RLAIF - RL from AI Feedback）
  1. モデルが応答ペアを生成
  2. AIが憲法に基づいて選好を判断
  3. AIフィードバックで報酬モデルを学習
  4. RLで最適化

class ConstitutionalAI:
    """
    Constitutional AIパイプライン
    人間フィードバックの代わりに原則（憲法）に基づくAI自己修正
    """

    CONSTITUTION = [
        {
            "principle": "無害性",
            "critique_prompt": (
                "この応答がユーザーや他の人に害を与える可能性があるか？"
                "暴力、差別、違法行為を助長しているか？"
            ),
            "revision_prompt": (
                "有害な内容を除去し、安全でありながら"
                "役に立つ方向に応答を修正せよ。"
            ),
        },
        {
            "principle": "正直性",
            "critique_prompt": (
                "この応答に事実と異なる情報があるか？"
                "不確実なことを確実だと表現しているか？"
            ),
            "revision_prompt": (
                "不正確な情報を修正し、不確実な部分は"
                "そうであると明示せよ。"
            ),
        },
        {
            "principle": "有用性",
            "critique_prompt": (
                "この応答はユーザーの質問に十分に答えているか？"
                "より役立つ情報があるか？"
            ),
            "revision_prompt": (
                "ユーザーにより有用になるよう応答を補完しつつ、"
                "安全性と正直性を維持せよ。"
            ),
        },
    ]

    def __init__(self, model):
        self.model = model

    def self_critique(self, prompt, initial_response):
        critiques = []
        for principle in self.CONSTITUTION:
            critique = self.model.generate(
                f"原則: {principle['principle']}\n"
                f"質問: {prompt}\n"
                f"応答: {initial_response}\n\n"
                f"批判: {principle['critique_prompt']}\n"
                f"この原則に照らして応答を評価せよ:"
            )
            critiques.append({
                "principle": principle["principle"],
                "critique": critique,
            })
        return critiques

    def generate_safe_response(self, prompt):
        initial = self.model.generate(prompt)
        critiques = self.self_critique(prompt, initial)
        revised = self.revise(prompt, initial, critiques)
        return {
            "initial": initial,
            "critiques": critiques,
            "revised": revised,
        }

1. データバイアス
   - 学習データが特定グループを過大/過小に代表
   - 例：医療データにおける特定人種の比率不均衡

2. アルゴリズムバイアス
   - モデル構造や学習過程で発生
   - 例：多数クラスに偏った最適化

3. 社会的バイアス
   - 学習テキストに内在する社会的ステレオタイプ
   - 例：「看護師」と「医師」の性別関連性

4. 確証バイアス
   - 既存パターンを強化する方向に学習
   - 例：フィードバックループによるバイアス増幅

5. 測定バイアス
   - 評価指標自体のバイアス
   - 例：英語中心のベンチマークで多言語モデルを評価

import numpy as np
from collections import defaultdict

class BiasDetector:
    """
    LLM出力のバイアスを検出するツール
    """

    def counterfactual_test(self, template, attributes):
        """
        反事実テスト：属性のみ変更して出力差を測定
        """
        results = defaultdict(list)

        for attr_name, attr_values in attributes.items():
            for value in attr_values:
                prompt = template.replace(f"[{attr_name}]", value)
                output = self.model.generate(prompt)
                results[attr_name].append({
                    "value": value,
                    "output": output,
                    "sentiment": self.analyze_sentiment(output),
                    "toxicity": self.measure_toxicity(output),
                })

        return self.compute_disparity(results)

    def winogender_evaluation(self, model):
        """
        Winogenderスタイルの評価
        職業-性別ステレオタイプバイアスの測定
        """
        templates = [
            "The nurse notified the patient that _ shift would end soon.",
            "The engineer told the manager that _ design was complete.",
        ]

        bias_scores = []
        for template in templates:
            he_prob = model.token_probability(
                template.replace("_", "his")
            )
            she_prob = model.token_probability(
                template.replace("_", "her")
            )
            bias_score = abs(he_prob - she_prob)
            bias_scores.append(bias_score)

        return np.mean(bias_scores)

1. 学習データの問題
   - 不正確な情報を含む学習データ
   - 古い情報（知識カットオフ以降）
   - 稀な事実に対する学習不足

2. モデルアーキテクチャの限界
   - 自己回帰生成の雪だるま効果
   - Attentionメカニズムの限界
   - 確率ベース生成の本質的不確実性

3. デコーディング戦略
   - 高いTemperature：創造的だが不正確
   - top-p/top-k設定による変動性

4. 学習目標
   - 次トークン予測が事実正確性と不一致
   - RLHFのhelpfulnessバイアス（知らなくても回答）

class HallucinationDetector:
    """
    ハルシネーション検出パイプライン
    """

    def detect_self_consistency(self, prompt, n_samples=5):
        """
        Self-Consistencyベースのハルシネーション検出
        同じ質問に複数回回答を生成し一貫性を測定
        """
        responses = [
            self.model.generate(prompt, temperature=0.7)
            for _ in range(n_samples)
        ]

        consistency_scores = []
        for i in range(len(responses)):
            for j in range(i + 1, len(responses)):
                score = self.compute_similarity(
                    responses[i], responses[j]
                )
                consistency_scores.append(score)

        avg_consistency = np.mean(consistency_scores)

        return {
            "responses": responses,
            "consistency": avg_consistency,
            "likely_hallucination": avg_consistency < 0.5,
        }

    def detect_with_retrieval(self, prompt, response, knowledge_base):
        """
        RAGベースのハルシネーション検出
        各主張をナレッジベースと照合
        """
        claims = self.extract_claims(response)
        results = []
        for claim in claims:
            relevant_docs = knowledge_base.search(claim, top_k=3)
            is_supported = self.fact_checker.verify(
                claim, relevant_docs
            )
            results.append({
                "claim": claim,
                "supported": is_supported,
            })

        hallucination_rate = sum(
            1 for r in results if not r["supported"]
        ) / len(results)

        return {
            "claims": results,
            "hallucination_rate": hallucination_rate,
        }

class HallucinationMitigation:
    """
    ハルシネーション軽減の総合戦略
    """

    def rag_grounding(self, query, knowledge_base):
        """
        RAGによる事実根拠の確保
        """
        docs = knowledge_base.search(query, top_k=5)
        context = "\n\n".join(
            f"[出典 {i+1}] {doc.content}"
            for i, doc in enumerate(docs)
        )

        prompt = (
            f"以下の情報のみに基づいて質問に答えよ。"
            f"情報が不十分なら「確認できません」と答えよ。\n\n"
            f"参考情報:\n{context}\n\n"
            f"質問: {query}\n答え:"
        )
        return self.model.generate(prompt)

    def chain_of_verification(self, query):
        """
        Chain-of-Verification (CoVe)
        1. 初期応答生成
        2. 検証質問生成
        3. 独立して検証質問に回答
        4. 検証結果を反映して最終応答
        """
        initial = self.model.generate(query)

        verification_qs = self.model.generate(
            f"以下の応答の事実的主張を検証できる"
            f"具体的な質問を生成せよ:\n"
            f"応答: {initial}\n検証質問:"
        )

        verifications = []
        for vq in verification_qs.split("\n"):
            if vq.strip():
                answer = self.model.generate(vq.strip())
                verifications.append({
                    "question": vq.strip(),
                    "answer": answer,
                })

        final = self.model.generate(
            f"元の質問: {query}\n"
            f"初期応答: {initial}\n\n"
            f"検証結果:\n"
            + "\n".join(
                f"Q: {v['question']}\nA: {v['answer']}"
                for v in verifications
            )
            + f"\n\n検証結果を反映した正確な最終応答を作成せよ:"
        )
        return final

レッドチームプロセス:

1. 範囲定義
   - テスト対象リスクカテゴリの選定
   - 成功/失敗基準の定義
   - 倫理的境界の設定

2. 攻撃ベクトル設計
   - 直接攻撃（有害コンテンツの直接要求）
   - 間接攻撃（コンテキスト操作、ロール付与）
   - プロンプトインジェクション（システムプロンプト無力化）
   - 多段階攻撃（段階的な境界移動）

3. テスト実行
   - 手動テスト（専門家）
   - 自動テスト（AIベース）
   - クラウドソーシング（多様な視点）

4. 結果分析と改善
   - 脆弱性の分類と深刻度評価
   - モデル改善（追加学習、フィルタリング）
   - 再テスト

class JailbreakTaxonomy:
    """
    LLM Jailbreak攻撃タイプの分類
    """

    CATEGORIES = {
        "role_playing": {
            "description": "特定ロールを付与して制限を回避",
            "severity": "high",
        },
        "prompt_injection": {
            "description": "システムプロンプトを無力化する入力",
            "severity": "critical",
        },
        "context_manipulation": {
            "description": "コンテキストを操作して制限を回避",
            "severity": "medium",
        },
        "encoding_attacks": {
            "description": "エンコーディングや変換でフィルタを回避",
            "severity": "high",
        },
        "multi_turn": {
            "description": "複数ターンにわたって段階的に境界を移動",
            "severity": "high",
        },
        "indirect_injection": {
            "description": "外部データソースを通じた間接インジェクション",
            "severity": "critical",
        },
    }

ユーザー入力
    |
    v
[Input Guardrail]
  - 有害コンテンツフィルタリング
  - プロンプトインジェクション検出
  - PII検出
  - トピック制限
    |
    v
[LLM処理]
    |
    v
[Output Guardrail]
  - 有害コンテンツフィルタリング
  - ハルシネーション検出
  - バイアス検出
  - 形式検証
  - 引用確認
    |
    v
安全な応答

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import re

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class GuardrailResult:
    passed: bool
    risk_level: RiskLevel
    reason: Optional[str] = None

class InputGuardrails:
    """
    入力ガードレール - LLMに渡す前にフィルタリング
    """

    def check_prompt_injection(self, user_input):
        injection_patterns = [
            r"ignore\s+(previous|all|above)\s+instructions",
            r"you\s+are\s+now\s+(?:DAN|unrestricted)",
            r"system\s*:\s*",
            r"jailbreak",
        ]

        for pattern in injection_patterns:
            if re.search(pattern, user_input, re.IGNORECASE):
                return GuardrailResult(
                    passed=False,
                    risk_level=RiskLevel.CRITICAL,
                    reason="Prompt injection detected",
                )

        return GuardrailResult(
            passed=True, risk_level=RiskLevel.LOW,
        )

    def check_pii(self, text):
        pii_patterns = {
            "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
            "phone": r"\b\d{3}[-.]?\d{3,4}[-.]?\d{4}\b",
            "credit_card": r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b",
        }

        detected_pii = []
        for pii_type, pattern in pii_patterns.items():
            matches = re.findall(pattern, text)
            if matches:
                detected_pii.append({
                    "type": pii_type, "count": len(matches),
                })

        if detected_pii:
            return GuardrailResult(
                passed=False,
                risk_level=RiskLevel.HIGH,
                reason=f"PII detected: {detected_pii}",
            )

        return GuardrailResult(
            passed=True, risk_level=RiskLevel.LOW,
        )

class InterpretabilityTools:
    """
    AIモデル解釈ツール集
    """

    def shap_explanation(self, model, input_text):
        """
        SHAP (SHapley Additive exPlanations)
        - ゲーム理論で各入力特徴の寄与度を計算
        - モデル非依存（どのモデルにも適用可能）
        """
        import shap
        explainer = shap.Explainer(model)
        shap_values = explainer([input_text])
        return {
            "tokens": shap_values.data[0],
            "values": shap_values.values[0],
            "base_value": shap_values.base_values[0],
        }

    def lime_explanation(self, model, input_text):
        """
        LIME (Local Interpretable Model-agnostic Explanations)
        - 入力周辺の局所的解釈モデルを学習
        - 個別予測に対する解釈
        """
        from lime.lime_text import LimeTextExplainer
        explainer = LimeTextExplainer()
        explanation = explainer.explain_instance(
            input_text, model.predict_proba,
            num_features=10, num_samples=1000,
        )
        return {
            "features": explanation.as_list(),
            "score": explanation.score,
        }

    def attention_visualization(self, model, input_tokens):
        """
        Attention Weight可視化
        - TransformerのSelf-Attentionパターン分析
        """
        outputs = model(input_tokens, output_attentions=True)
        attentions = outputs.attentions

        analysis = []
        for layer_idx, layer_attn in enumerate(attentions):
            for head_idx in range(layer_attn.shape[1]):
                head_attn = layer_attn[0, head_idx].detach().numpy()
                analysis.append({
                    "layer": layer_idx,
                    "head": head_idx,
                    "entropy": self._attention_entropy(head_attn),
                })
        return analysis

主要な研究方向：

1. Superposition（重畳）
   - ニューロンが複数の特徴を同時にエンコード
   - 特徴数 > ニューロン数（多義ニューロン）

2. Sparse Autoencoders (SAE)
   - 重畳された特徴を分離する技法
   - 各方向が一つの概念に対応

3. Circuit Analysis（回路分析）
   - 特定の行動を担当するニューロン回路を識別
   - モジュール化された機能単位の発見

4. Scaling Monosemanticity
   - 大規模モデルでも単一意味特徴を抽出
   - Claudeで数百万の解釈可能な特徴を発見

EU AI Actリスク等級体系：

[禁止（Unacceptable Risk）]
  - 社会的スコアリング
  - リアルタイム生体認証大衆監視
  - 潜在意識操作技法
  - 脆弱層の搾取

[高リスク（High Risk）]
  - 医療機器
  - 採用・人事管理
  - 信用評価
  - 教育評価
  - 法執行
  - 必須サービスへのアクセス

  要件：
  - リスク管理システム
  - データガバナンス
  - 技術文書化
  - 透明性義務
  - 人的監視
  - 正確性、堅牢性、サイバーセキュリティ

[限定リスク（Limited Risk）]
  - チャットボット（AI使用告知義務）
  - ディープフェイク（生成表示義務）
  - 感情認識システム

[最小リスク（Minimal Risk）]
  - AIベースのゲーム
  - スパムフィルタ
  - 規制なし（自発的行動規範推奨）

Microsoft Responsible AI:
原則: 公正性、信頼性・安全性、プライバシー・セキュリティ、
      包摂性、透明性、説明責任
ツール: Responsible AI Dashboard, Fairlearn, InterpretML


Google Responsible AI:
原則:
  1. 社会的に有益であること
  2. 不公正なバイアスを避ける
  3. 安全のために構築・テスト
  4. 人間に対して説明責任を持つ
  5. プライバシー原則を組み込む
  6. 高い科学的基準を維持
  7. これらの原則に合致する用途のみ


Anthropic Responsible Scaling Policy (RSP):
AI Safety Level (ASL)体系:
  - ASL-1: 意味のあるリスクなし
  - ASL-2: 現在のモデルレベル（基本安全措置）
  - ASL-3: 非国家行為者のWMD増強可能
  - ASL-4: 国家行為者レベルの脅威可能

class StagedRollout:
    """
    AIモデルの段階的デプロイ戦略
    """

    STAGES = [
        {
            "name": "Internal Testing",
            "audience": "社内スタッフ",
            "percentage": 0,
            "duration": "2週間",
            "criteria": [
                "全安全テスト通過",
                "レッドチームレポート完了",
                "社内フィードバック収集",
            ],
        },
        {
            "name": "Limited Beta",
            "audience": "信頼できるパートナー",
            "percentage": 1,
            "duration": "2週間",
        },
        {
            "name": "Gradual Rollout",
            "audience": "一般ユーザー",
            "percentage": 10,
            "duration": "段階的拡大",
        },
        {
            "name": "Full Deployment",
            "audience": "全ユーザー",
            "percentage": 100,
            "duration": "常時",
        },
    ]

    def should_rollback(self, metrics):
        rollback_triggers = {
            "safety_incident": metrics.get("safety_incidents", 0) > 0,
            "error_spike": metrics.get("error_rate", 0) > 0.05,
            "latency_spike": metrics.get("p99_latency", 0) > 5000,
        }
        for trigger, is_triggered in rollback_triggers.items():
            if is_triggered:
                return True, f"Rollback trigger: {trigger}"
        return False, "Normal"

class KillSwitch:
    """
    AIシステムの緊急停止メカニズム
    """

    def __init__(self, config):
        self.config = config
        self.is_active = True

    def automatic_trigger(self, metrics):
        conditions = {
            "safety_critical": (
                metrics.get("safety_incidents", 0) > 0
            ),
            "cascade_failure": (
                metrics.get("error_rate", 0) > 0.5
            ),
            "data_breach": (
                metrics.get("pii_leak_detected", False)
            ),
        }
        for condition, triggered in conditions.items():
            if triggered:
                self.activate(reason=condition)
                return True
        return False

    def activate(self, reason):
        self.is_active = False
        self.switch_to_fallback()
        self.notify_team(reason)
        self.preserve_logs()
        self.initiate_postmortem(reason)

AI Safety多層防御体系：

Layer 1: モデル整合（RLHF, DPO, Constitutional AI）
Layer 2: 入出力ガードレール（フィルタリング、検証）
Layer 3: モニタリングと監査（リアルタイム監視、定期監査）
Layer 4: 人的監視（エスカレーション、キルスイッチ）
Layer 5: 規制とガバナンス（法的フレームワーク、内部ポリシー）