LESSON 30分

ストーリー

田
田中VPoE
ポリシーを策定しても、「誰が」「いつ」「何を」AIで処理したかの記録がなければ、問題発生時に原因を追跡できない
あなた
監査証跡（Audit Trail）ですね。金融システムなどでは必須の仕組みですが、AIシステムでも必要なんですか？
あ
田
田中VPoE
AIシステムでは特に重要だ。LLMの出力は非決定的で、同じ入力でも異なる結果が返ることがある。問題が起きた時に「その時の入力と出力の組み合わせ」を再現できなければ、原因究明ができない
あなた
AI特有の課題があるんですね
あ

AI監査証跡の要件

なぜAIシステムで監査証跡が重要か

理由説明
非決定性同じ入力でも出力が異なる可能性がある
説明責任AIの判断根拠を後から説明する必要がある
法規制対応EU AI Act等でトレーサビリティが要求される
品質改善過去の入出力データが品質改善の基盤になる
インシデント対応問題発生時の原因特定に不可欠

記録すべき情報

カテゴリ記録項目目的
リクエストタイムスタンプ、ユーザーID、入力データ誰がいつ何を要求したか
モデルモデル名、バージョン、パラメータどのモデルで処理したか
プロンプトシステムプロンプト、ユーザープロンプトどのような指示で処理したか
レスポンス出力データ、トークン数、レイテンシ何が返されたか
メタデータ信頼度スコア、コスト、エラー情報処理の品質はどうだったか

監査証跡の設計

ログスキーマ

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

@dataclass
class AuditLogEntry:
    """AI処理の監査ログエントリ"""
    # 識別情報
    log_id: str
    request_id: str
    timestamp: datetime

    # ユーザー情報
    user_id: str
    department: str
    use_case: str

    # 入力情報
    input_type: str  # text, image, audio, document
    input_data_hash: str  # 入力データのハッシュ（データ自体は別途保存）
    input_metadata: dict = field(default_factory=dict)

    # モデル情報
    model_name: str = ""
    model_version: str = ""
    model_parameters: dict = field(default_factory=dict)

    # プロンプト情報
    system_prompt_hash: str = ""
    prompt_template_id: str = ""

    # 出力情報
    output_data_hash: str = ""
    output_summary: str = ""
    tokens_input: int = 0
    tokens_output: int = 0

    # 品質情報
    confidence_score: float = 0.0
    latency_ms: int = 0
    cost_usd: float = 0.0

    # 判定情報
    human_review_required: bool = False
    human_review_result: str = ""  # approved, rejected, modified
    final_decision: str = ""

    # エラー情報
    error_occurred: bool = False
    error_type: str = ""
    error_message: str = ""

ストレージ設計

保存先データ保持期間アクセス頻度
リアルタイムDB監査ログメタデータ90日高（監視・検索）
オブジェクトストレージ入出力データ本体1年中（調査時）
長期アーカイブ全データ（圧縮）5年低（法規制対応）
分析基盤集約メトリクス無期限中（トレンド分析）

トレーサビリティの実装

リクエストトレーシング

import uuid
import hashlib
import json
from datetime import datetime

class AuditTracer:
    """AI処理の監査トレーシング"""

    def __init__(self, storage_backend):
        self.storage = storage_backend

    def start_trace(self, user_id: str, department: str, use_case: str) -> str:
        """トレーシングを開始"""
        request_id = str(uuid.uuid4())
        self.current_entry = AuditLogEntry(
            log_id=str(uuid.uuid4()),
            request_id=request_id,
            timestamp=datetime.now(),
            user_id=user_id,
            department=department,
            use_case=use_case
        )
        return request_id

    def log_input(self, input_data: Any, input_type: str):
        """入力データを記録"""
        data_str = json.dumps(input_data, ensure_ascii=False, default=str)
        self.current_entry.input_type = input_type
        self.current_entry.input_data_hash = hashlib.sha256(
            data_str.encode()
        ).hexdigest()
        # 入力データ本体はオブジェクトストレージに保存
        self.storage.store_data(
            self.current_entry.request_id, "input", input_data
        )

    def log_model(self, model_name: str, version: str, params: dict):
        """使用モデル情報を記録"""
        self.current_entry.model_name = model_name
        self.current_entry.model_version = version
        self.current_entry.model_parameters = params

    def log_output(self, output_data: Any, confidence: float, latency_ms: int):
        """出力結果を記録"""
        data_str = json.dumps(output_data, ensure_ascii=False, default=str)
        self.current_entry.output_data_hash = hashlib.sha256(
            data_str.encode()
        ).hexdigest()
        self.current_entry.confidence_score = confidence
        self.current_entry.latency_ms = latency_ms
        self.storage.store_data(
            self.current_entry.request_id, "output", output_data
        )

    def finalize(self) -> AuditLogEntry:
        """トレーシングを完了し、ログを永続化"""
        self.storage.store_log(self.current_entry)
        return self.current_entry

改ざん防止

対策方法
ハッシュチェーン各ログエントリに前エントリのハッシュを含める
書き込み専用ログストレージは追記のみ許可、削除・更新を禁止
デジタル署名ログエントリにサーバー署名を付与
定期的な整合性チェックバッチでハッシュチェーンの検証を実行

監査クエリの実装

よくある監査クエリ

class AuditQueryService:
    """監査ログの検索・分析"""

    def __init__(self, storage_backend):
        self.storage = storage_backend

    def find_by_user(self, user_id: str, start_date: datetime, end_date: datetime):
        """ユーザー別の利用履歴"""
        return self.storage.query(
            user_id=user_id,
            timestamp_range=(start_date, end_date)
        )

    def find_errors(self, start_date: datetime, end_date: datetime):
        """エラー発生履歴"""
        return self.storage.query(
            error_occurred=True,
            timestamp_range=(start_date, end_date)
        )

    def find_low_confidence(self, threshold: float = 0.7):
        """低信頼度の処理結果"""
        return self.storage.query(
            confidence_score_lt=threshold
        )

    def usage_summary(self, department: str, period: str):
        """部門別利用サマリー"""
        return self.storage.aggregate(
            group_by=["use_case", "model_name"],
            filter_department=department,
            period=period,
            metrics=["count", "total_cost", "avg_latency", "avg_confidence"]
        )

まとめ

要素内容
記録項目リクエスト、モデル、プロンプト、レスポンス、メタデータ
ストレージ短期（DB）+ 中期（オブジェクト）+ 長期（アーカイブ）の3層
トレーサビリティリクエストIDによる全処理の追跡可能性
改ざん防止ハッシュチェーン、書き込み専用、デジタル署名

チェックリスト

  • AI監査証跡で記録すべき情報を理解した
  • ログスキーマの設計ができる
  • ストレージ設計（保持期間、アクセスパターン）を把握した
  • 改ざん防止の仕組みを理解した
  • 監査クエリのユースケースを把握した

次のステップへ

次は法規制対応とコンプライアンスを学びます。

推定読了時間: 30分