RAGセキュリティ監視パイプライン設計 RAG Security

RAGセキュリティ監視パイプライン — 本番ゼロからの設計ガイド

2026年6月27日 #03 監視・可観測性 対象: SRE / セキュリティ担当

RAGセキュリティ監視パイプラインを本番環境でゼロから設計する場合、「何をどのタイミングでログに出力し、どのルールで検知するか」が最大の課題になる。 一般的なAPMツールはRAG特有のイベント構造（Ingestion・Retrieval・Generation）を把握しておらず、そのまま流用すると権限外参照やガードレール違反を見落とす。

本記事では、RAG監査ログ設計の具体的なJSONスキーマ、セキュリティイベントスキーマの必須フィールド定義、検知ルール20選、Pythonによる簡易パイプライン実装、そして品質KPIとセキュリティKPIを統合したRAGモニタリングダッシュボード構成案を解説する。

この記事で分かること

  • RAG専用の監査ログスキーマ（必須フィールド・機密度ラベル）の設計方法
  • 権限外参照・異常頻度・ガードレール違反を検知する20のルール定義
  • 品質監視（RAGASメトリクス）とセキュリティ監視を1つのパイプラインに統合する設計パターン

対象読者 / 前提知識

対象読者：RAGを本番運用するSRE・セキュリティ監視担当

前提知識：RAGの基本アーキテクチャ（Embedding・Vector Store・LLM呼び出し）の概要、PythonでのJSON操作、SIEM/ログ集約ツールの基礎知識

目次

  1. なぜRAGには専用の監視パイプラインが必要なのか
  2. RAGパイプラインのイベントモデル
  3. セキュリティ監査ログスキーマの設計
  4. 検知ルール20選
  5. Pythonでの簡易監視パイプライン実装例
  6. ダッシュボード構成案
  7. 監視の運用コスト見積もり観点

なぜRAGには専用の監視パイプラインが必要なのか

一般的なAPM（Datadog・New Relic等）はレイテンシ・エラーレート・スループットの3指標を中心に設計されており、誰がどのドキュメントを取得したかというRAG固有のアクセス制御イベントをキャプチャできない。 RAGはドキュメントをチャンク化してVector Storeに格納する時点で、元ドキュメントのアクセス権限がチャンク単位に引き継がれないケースが多く、これが本番でのコンプライアンス違反につながる。 OWASP RAG Security Cheat Sheetは「RAGはリスクを削減するのではなく、データパイプライン全体に再分配する」と明示している。

一般APMとRAG専用監視の差異を整理すると以下のとおりだ。

観点一般APMRAG専用監視
イベント粒度HTTP req/res単位Ingestion・Retrieval・Generationの3フェーズ
アクセス制御APIキー・JWT認証のみチャンク単位のACL・テナント隔離の検証
品質KPI非対応Faithfulness・Answer Relevancy等RAGASメトリクス
検知対象レイテンシ異常・5xxエラー権限外取得・ガードレール違反・異常クエリ頻度
ログ保全任意取得ドキュメントIDをトレーサブルに保存（監査要件）
インシデント対応汎用プレイブック毒入りドキュメント隔離・キャッシュ無効化専用手順

品質監視とセキュリティ監視は別のシステムで運用されがちだが、RAGASメトリクス（Faithfulness・Context Recall等）の悪化はドキュメント汚染攻撃の早期シグナルになりうるため、両者を同一パイプラインに統合するアーキテクチャが本番運用では有効だ。

ここまでのまとめ

  • 一般APMはRAGのアクセス制御・品質KPIイベントをカバーしない
  • OWASPはIngestionからGenerationまでの全フェーズにわたる可観測性を要求している
  • 品質劣化シグナルとセキュリティアラートを統合することで、攻撃の早期検知が可能になる

RAGパイプラインのイベントモデル

RAGの監視設計で最初に行うべきは、パイプライン上のどの地点でどのイベントが発生するかを明確にすることだ。 以下のASCII図でイベント発生点を俯瞰する。 

[ユーザー/エージェント]
       │ query
       ▼
┌──────────────────────────────────────────────────────────┐
│  [E1] QueryReceived                                      │
│         │                                                │
│  ┌──────▼──────┐    ┌───────────────┐    ┌───────────┐  │
│  │  Retrieval  │───▶│  Vector Store │───▶│ LLM (Gen) │  │
│  │  Engine     │    │  ACL Check    │    │           │  │
│  └─────────────┘    └───────────────┘    └───────────┘  │
│  [E2] QueryNormalized                                    │
│  [E3] RetrievalHit / [E4] RetrievalMiss                  │
│  [E5] ACLCheckResult (permit/deny)                       │
│  [E6] ChunkScoreRecorded                                 │
│                          [E7] PromptAssembled            │
│                          [E8] GenerationCompleted        │
│                          [E9] GuardrailEvaluated         │
└──────────────────────────────────────────────────────────┘
       ▲
[Ingestion Pipeline]
  [E0-A] DocumentIngested
  [E0-B] DocumentUpdated
  [E0-C] DocumentDeleted
  [E0-D] HashVerificationFailed

Ingestionイベント（追加・更新・削除）

Ingestionフェーズでは、ドキュメントの追加・更新・削除の3種類のイベントに加え、ハッシュ検証失敗（E0-D）を独立したセキュリティイベントとして扱う必要がある。 OWASP RAG Security Cheat Sheet Section 1は「すべてのドキュメントをIngestion時点でSHA-256でハッシュし、Retrieval前に検証すること」を要求している。 ドキュメントの出所（アップロード者・ソースURL・承認フロー通過有無）をprovenanceフィールドに記録することで、汚染ドキュメントのフォレンジックが可能になる。

Retrievalイベント（クエリ・ヒット・スコア・権限チェック）

RetrievalフェーズはRAGセキュリティで最もインシデント頻度が高い領域だ。 クエリ正規化後のテキスト（E2）、取得されたチャンクのIDとコサイン類似度スコア（E6）、ACLチェック結果（E5）の3点が最低限の記録項目になる。 OWASP Section 4が強調するとおり、権限チェックはIngestion時ではなくRetrieval時にも毎回実施する必要があり、その結果を監査ログに残すことが前提条件となる。

Generationイベント（プロンプト・応答・ガードレール判定結果）

Generationフェーズでは、LLMに渡したプロンプト全体（または安全なハッシュ）、生成された応答、そしてガードレール違反検知の結果（pass/block/redact）を記録する。 ガードレール結果（E9）はセキュリティイベントとして扱い、連続違反のカウントをリアルタイムで追跡することが重要だ。 応答にPII・機密ラベル付きコンテンツが含まれていないかの出力バリデーション結果もこのイベントに統合する。

ここまでのまとめ

  • イベントはIngestion（E0系）・Retrieval（E2〜E6）・Generation（E7〜E9）の3グループに分類する
  • ハッシュ検証失敗とACLチェック結果は独立したセキュリティイベントとして扱う
  • ガードレール判定結果はGenerationイベントに統合し、連続違反を追跡可能にする

セキュリティ監査ログスキーマの設計

RAG品質管理とセキュリティ監視を統合するには、単一のJSONスキーマで両者のフィールドを収容することが最も実装コストが低い。 スキーマはイベントタイプによってフィールドの必須/任意を切り替えられるようevent_typeを起点とした条件分岐で定義する。

必須フィールド一覧

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "RAGAuditEvent",
  "type": "object",
  "required": [
    "event_id", "event_type", "timestamp", "trace_id",
    "user_id", "role", "tenant",
    "query_text_hash", "retrieved_doc_ids",
    "guardrail_result", "sensitivity_label"
  ],
  "properties": {
    "event_id":          { "type": "string", "format": "uuid" },
    "event_type":        {
      "type": "string",
      "enum": [
        "ingestion.added", "ingestion.updated", "ingestion.deleted",
        "ingestion.hash_failed",
        "retrieval.query", "retrieval.hit", "retrieval.miss",
        "retrieval.acl_deny",
        "generation.prompt_assembled", "generation.completed",
        "generation.guardrail_block", "generation.guardrail_redact"
      ]
    },
    "timestamp":         { "type": "string", "format": "date-time" },
    "trace_id":          { "type": "string", "description": "end-to-end request correlation ID" },
    "user_id":           { "type": "string" },
    "role":              { "type": "string", "description": "RBAC role at request time" },
    "tenant":            { "type": "string", "description": "tenant/org identifier" },
    "query_text_hash":   { "type": "string", "description": "SHA-256 of normalized query" },
    "retrieved_doc_ids": { "type": "array", "items": { "type": "string" } },
    "chunk_scores":      { "type": "array", "items": { "type": "number" } },
    "acl_check_result":  { "type": "string", "enum": ["permit", "deny", "partial_deny"] },
    "guardrail_result":  {
      "type": "object",
      "properties": {
        "status":       { "type": "string", "enum": ["pass", "block", "redact"] },
        "rule_ids":     { "type": "array", "items": { "type": "string" } },
        "pii_detected": { "type": "boolean" }
      }
    },
    "sensitivity_label": {
      "type": "string",
      "enum": ["public", "internal", "confidential", "restricted"]
    },
    "doc_provenance": {
      "type": "object",
      "properties": {
        "source_url":  { "type": "string" },
        "uploader_id": { "type": "string" },
        "sha256_hash": { "type": "string" },
        "approved":    { "type": "boolean" }
      }
    },
    "ragas_metrics": {
      "type": "object",
      "description": "RAG品質管理メトリクス（generation eventのみ）",
      "properties": {
        "faithfulness":     { "type": "number", "minimum": 0, "maximum": 1 },
        "answer_relevancy": { "type": "number", "minimum": 0, "maximum": 1 },
        "context_recall":   { "type": "number", "minimum": 0, "maximum": 1 }
      }
    },
    "response_text_hash": { "type": "string" },
    "latency_ms":         { "type": "integer" }
  }
}

query_text_hashにはSHA-256ハッシュを格納し、生クエリテキストは別の暗号化ストレージに保存する。 これにより、ログ基盤にアクセスできるオペレーターがクエリ内容を直接参照できない構造を実現し、プライバシーとフォレンジック要件を両立できる。 ragas_metricsを同一スキーマに含めることで、FaithfulnessスコアとGuardrail違反数を同じダッシュボードで相関分析できるのがこの設計の最大の差別化ポイントだ。

機密度ラベルと取り扱いポリシー

sensitivity_labelログ保存期間アクセス可能ロール暗号化要件削除ポリシー
public90日全SRE転送時TLSTTL後自動削除
internal1年SRE・セキュリティ担当転送時TLS＋保存時AES-256保持期限後削除
confidential3年セキュリティ担当のみテナント別鍵で暗号化手動承認後削除
restricted7年（法的要件依存）Security Lead＋法務HSM管理鍵＋WORM法的保持期間まで削除不可

OWASP RAG Security Cheat Sheet Section 4に従い、ソースドキュメントが削除または権限変更された場合はVector Store内の関連チャンクも連鎖削除し、その削除イベントをログに記録する必要がある。 restrictedラベルを持つドキュメントへの照会が集中した場合は、後述の検知ルール（R-18）が自動でアラートを発火する。

ここまでのまとめ

  • 単一JSONスキーマにRAGAS品質メトリクスとセキュリティフィールドを統合することで、両者の相関分析が可能になる
  • 生クエリテキストはハッシュのみをログに記録し、原文は暗号化ストアに分離保管する
  • 機密度ラベルごとに保存期間・暗号化要件・削除ポリシーを定義し、コンプライアンスを担保する

検知ルール20選

以下のルール一覧は、権限外参照・異常クエリ頻度・ガードレール連続違反・大量取得・特定ラベル集中照会を網羅する。 優先度Highは即時アラート、Mediumはダッシュボードでのトレンド監視、Lowは定期バッチレポートを推奨する。

#ルール名対象イベント検知条件アクション優先度
R-01ACL権限外参照retrieval.acl_deny同一user_idで5分以内に3回以上denyセッション一時停止・SlackアラートHigh
R-02テナント境界越えretrieval.hitretrieved_doc_idsにリクエスト元tenant外のchunk_idが含まれる即時ブロック・インシデント起票High
R-03クエリ異常頻度retrieval.query同一user_idで60秒以内に30クエリ超レート制限・アラートHigh
R-04ガードレール連続違反generation.guardrail_block同一user_idで10分以内に5回以上blockユーザーロック・エスカレーションHigh
R-05ハッシュ検証失敗ingestion.hash_failed1件でも発生当該ドキュメント隔離・セキュリティ通知High
R-06大量チャンク取得retrieval.hit1リクエストでretrieved_doc_ids件数が閾値（デフォルト10）超アラート・リクエスト記録High
R-07Restricted集中照会retrieval.hitsensitivity_label=restrictedのchunkが10分以内に50件超取得自動ブロック・インシデント起票High
R-08プロンプトインジェクション試行generation.prompt_assembledプロンプトに"ignore previous"・"SYSTEM:"等のパターンを検出Generationブロック・ログ記録High
R-09PII出力検出generation.completedguardrail_result.pii_detected=true応答を返さずredact・アラートHigh
R-10削除済みドキュメント参照retrieval.hitretrieved_doc_idsに削除済みchunk_idが含まれる即時ブロック・Vector Store整合性チェック起動High
R-11夜間異常クエリretrieval.query業務時間外（0:00〜5:00 JST）にクエリ頻度が通常の300%超アラート・ユーザー確認Medium
R-12新規ロールによるConfidential参照retrieval.hitroleが過去30日以内に初めてconfidentialドキュメントに触れた管理者通知・監査ログフラグMedium
R-13Faithfulness急落generation.completedragas_metrics.faithfulness が直近100件の平均から0.2以上低下品質アラート・ドキュメント汚染調査トリガーMedium
R-14スコア分布異常retrieval.hitchunk_scoresの最大値が0.98超（過適合の可能性）埋め込み分布チェック起動Medium
R-15クエリパターン偵察retrieval.query同一user_idで類似クエリを系統的にバリエーション展開（Levenshtein距離≦5）Corpus偵察疑いフラグ・アラートMedium
R-16Ingestion権限外実行ingestion.addedingestionを実行したuser_idが承認済みパイプラインロール外Ingestionロールバック・通知Medium
R-17キャッシュ境界違反retrieval.hitキャッシュhit時のsensitivity_labelがリクエスタのクリアランスを超過キャッシュ無効化・アラートMedium
R-18Restricted集中照会（緩）retrieval.hitsensitivity_label=restrictedのchunkが1時間以内に200件超ダッシュボードフラグ・次営業日レポートLow
R-19ガードレール迂回試行generation.guardrail_blockblockされた直後の再クエリが5秒以内（自動化ツール疑い）レート制限強化・フラグMedium
R-20Context Recall長期低下generation.completedragas_metrics.context_recall が7日移動平均で0.15以上低下Vector Store再インデックス提案・品質アラートLow

R-13（Faithfulness急落）とR-20（Context Recall低下）はRAGパイプライン評価のRAGAS的な品質指標をセキュリティアラートのトリガーとして使用する点で、従来の監視設計にない視点だ。 ドキュメント汚染攻撃はまず品質メトリクスの悪化として観測されることが多く、これらのルールが早期警戒システムとして機能する。

ここまでのまとめ

  • High優先度10件は即時アラート・自動ブロックと連動させる
  • RAGASメトリクスの急落をセキュリティルールに組み込むことで品質劣化と攻撃を同時検知できる
  • R-15（偵察パターン）のようにクエリ意図レベルの検知を行うにはLevenshtein距離等の類似度計算が必要になる

Pythonでの簡易監視パイプライン実装例

以下は「ログ出力 → ルール評価 → アラート送信」の3ステップを最小限のコードで実現した実装例だ。 本番では各ステップをマイクロサービス化または既存のログ基盤（OpenTelemetry Collector等）に差し込む形で拡張する。

Step 1: 監査ログ出力

# filename: rag_audit_logger.py
import json
import hashlib
import uuid
from datetime import datetime, timezone

def make_audit_event(
    event_type: str,
    user_id: str,
    role: str,
    tenant: str,
    query_text: str,
    retrieved_doc_ids: list[str],
    chunk_scores: list[float],
    acl_check_result: str,
    guardrail_result: dict,
    sensitivity_label: str,
    ragas_metrics: dict | None = None,
    trace_id: str | None = None,
) -> dict:
    return {
        "event_id": str(uuid.uuid4()),
        "event_type": event_type,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "trace_id": trace_id or str(uuid.uuid4()),
        "user_id": user_id,
        "role": role,
        "tenant": tenant,
        # Store hash only; raw text goes to encrypted store
        "query_text_hash": hashlib.sha256(query_text.encode()).hexdigest(),
        "retrieved_doc_ids": retrieved_doc_ids,
        "chunk_scores": chunk_scores,
        "acl_check_result": acl_check_result,
        "guardrail_result": guardrail_result,
        "sensitivity_label": sensitivity_label,
        "ragas_metrics": ragas_metrics or {},
    }

def emit_event(event: dict) -> None:
    # Output as structured JSON log; shipped to SIEM by log agent
    print(json.dumps(event, ensure_ascii=False))

出力例（retrieval.hitイベント）：

{
  "event_id": "a3f2c1d4-...",
  "event_type": "retrieval.hit",
  "timestamp": "2026-06-27T12:00:00+00:00",
  "trace_id": "9b8e7f6a-...",
  "user_id": "user_042",
  "role": "analyst",
  "tenant": "org_acme",
  "query_text_hash": "e3b0c44298fc1c...",
  "retrieved_doc_ids": ["doc_101_chunk_3", "doc_205_chunk_1"],
  "chunk_scores": [0.91, 0.87],
  "acl_check_result": "permit",
  "guardrail_result": {"status": "pass", "rule_ids": [], "pii_detected": false},
  "sensitivity_label": "confidential",
  "ragas_metrics": {}
}

Step 2: ルール評価エンジン

# filename: rag_rule_engine.py
from collections import defaultdict, deque
from datetime import datetime, timezone
import time

class RuleEngine:
    def __init__(self):
        # Sliding window: user_id -> deque of timestamps
        self._acl_deny_windows: dict[str, deque] = defaultdict(deque)
        self._query_windows: dict[str, deque] = defaultdict(deque)
        self._guardrail_windows: dict[str, deque] = defaultdict(deque)
        self.DELETED_CHUNK_IDS: set[str] = set()  # populated from document lifecycle events

    def _sliding_window_count(self, window: deque, window_sec: int) -> int:
        now = time.time()
        while window and window[0] < now - window_sec:
            window.popleft()
        return len(window)

    def evaluate(self, event: dict) -> list[dict]:
        alerts = []
        uid = event.get("user_id", "")
        etype = event.get("event_type", "")
        now = time.time()

        # R-01: ACL deny 3+ times in 5 min
        if etype == "retrieval.acl_deny":
            self._acl_deny_windows[uid].append(now)
            if self._sliding_window_count(self._acl_deny_windows[uid], 300) >= 3:
                alerts.append({"rule": "R-01", "severity": "high", "user_id": uid,
                                "event_id": event["event_id"]})

        # R-03: Query rate > 30 in 60 sec
        if etype == "retrieval.query":
            self._query_windows[uid].append(now)
            if self._sliding_window_count(self._query_windows[uid], 60) > 30:
                alerts.append({"rule": "R-03", "severity": "high", "user_id": uid,
                                "event_id": event["event_id"]})

        # R-04: Guardrail block 5+ times in 10 min
        if etype == "generation.guardrail_block":
            self._guardrail_windows[uid].append(now)
            if self._sliding_window_count(self._guardrail_windows[uid], 600) >= 5:
                alerts.append({"rule": "R-04", "severity": "high", "user_id": uid,
                                "event_id": event["event_id"]})

        # R-06: Bulk chunk retrieval (>10 chunks)
        if etype == "retrieval.hit":
            doc_ids = event.get("retrieved_doc_ids", [])
            if len(doc_ids) > 10:
                alerts.append({"rule": "R-06", "severity": "high", "user_id": uid,
                                "chunk_count": len(doc_ids), "event_id": event["event_id"]})

        # R-09: PII detected in output
        if etype == "generation.completed":
            gr = event.get("guardrail_result", {})
            if gr.get("pii_detected"):
                alerts.append({"rule": "R-09", "severity": "high", "user_id": uid,
                                "event_id": event["event_id"]})

        # R-13: Faithfulness drop (simplified check)
        if etype == "generation.completed":
            metrics = event.get("ragas_metrics", {})
            if metrics.get("faithfulness", 1.0) < 0.5:
                alerts.append({"rule": "R-13", "severity": "medium", "user_id": uid,
                                "faithfulness": metrics.get("faithfulness"),
                                "event_id": event["event_id"]})

        return alerts

Step 3: アラート送信

# filename: rag_alert_dispatcher.py
import json
import urllib.request

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"

def dispatch_alert(alert: dict) -> None:
    severity_emoji = {"high": ":red_circle:", "medium": ":large_yellow_circle:", "low": ":large_blue_circle:"}
    emoji = severity_emoji.get(alert.get("severity", "low"), ":white_circle:")
    payload = {
        "text": f"{emoji} *RAG Security Alert* [{alert['rule']}]\n"
                f"User: `{alert.get('user_id', 'unknown')}` | "
                f"Event: `{alert.get('event_id', '-')}`\n"
                f"```{json.dumps(alert, ensure_ascii=False, indent=2)}```"
    }
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        SLACK_WEBHOOK_URL,
        data=data,
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req) as resp:
        _ = resp.read()

# --- Wiring all 3 steps together ---
def process_rag_event(raw_event: dict) -> None:
    engine = RuleEngine()
    alerts = engine.evaluate(raw_event)
    emit_event(raw_event)           # Step 1: structured log
    for alert in alerts:            # Step 2+3: rule match -> alert
        dispatch_alert(alert)

ここまでのまとめ

  • 3ステップ（emit → evaluate → dispatch）を分離することで、各層の差し替えが容易になる
  • スライディングウィンドウカウンタをメモリ上で管理するだけでR-01/03/04の大半は実装可能
  • RAGASメトリクスはgeneration.completedイベント内に埋め込み、R-13等の品質検知ルールと統合する

ダッシュボード構成案

品質KPIとセキュリティKPIを1画面に収めることで、RAGの異常を多角的に把握できる。 以下はGrafanaやDatadogのカスタムダッシュボードを想定したレイアウト案だ。 

┌──────────────────────────────────────────────────────────────────────────────┐
│  RAG Unified Monitoring Dashboard                    [tenant: all] [1h ▼]   │
├───────────────────────────┬──────────────────────────┬───────────────────────┤
│  🟢 QUALITY KPIs           │  🔴 SECURITY KPIs          │  🔔 ACTIVE ALERTS      │
│                           │                          │                       │
│  Faithfulness    0.87 ▲   │  ACL Denies (5min)   2   │  R-03 [HIGH] user042  │
│  Answer Relevancy 0.91 ─  │  Guardrail Blocks   14   │  R-04 [HIGH] user007  │
│  Context Recall  0.78 ▼   │  Hash Failures       0   │  R-13 [MED] trace9f   │
│                           │  Tenant Violations   0   │                       │
├───────────────────────────┴──────────────────────────┴───────────────────────┤
│  📈 Query Volume & Anomaly (time series, 1h)                                  │
│  ████░░░░░█████████░░░░████████████░░░░░░░░░░░░░░░░░░ [threshold ---]        │
│  00:00        15:00        30:00        45:00        60:00                   │
├──────────────────────────────────────────────────────────────────────────────┤
│  📊 Sensitivity Label Distribution    │  🥧 Event Type Breakdown              │
│                                       │                                       │
│  public     ██████████████████ 68%    │  retrieval.hit    45%                 │
│  internal   ████████ 24%              │  retrieval.query  30%                 │
│  confidential ████ 7%                 │  generation.*     20%                 │
│  restricted  █ 1%                     │  ingestion.*       5%                 │
├──────────────────────────────────────────────────────────────────────────────┤
│  🔎 Top Retrieved Doc IDs (1h)        │  ⚠️ Rule Trigger Heatmap               │
│  doc_205  ████████████████ 342        │       R01 R03 R04 R06                 │
│  doc_101  ████████████ 218            │  u042  ·   ●   ·   ·                  │
│  doc_089  ████████ 145                │  u007  ·   ·   ●   ·                  │
│  doc_334  ████ 78                     │  u019  ·   ·   ·   ●                  │
└──────────────────────────────────────────────────────────────────────────────┘

左上ブロックにRAGASメトリクス（Faithfulness・Answer Relevancy・Context Recall）を配置し、右上にセキュリティKPI（ACL Deny件数・Guardrail Block数・Hash失敗数）を並べることで、品質低下とセキュリティ事象を同一の視野で確認できる。 右下のRule Trigger HeatmapはユーザーIDとルールIDのクロス集計で、特定ユーザーが複数ルールに同時抵触しているかを即座に把握するために有効だ。 ダッシュボードのデータソースはStep 1で出力した構造化JSONログをElasticsearch/OpenSearchに取り込むか、OpenTelemetry Collector経由でPrometheusにエクスポートする構成が一般的だ。

ここまでのまとめ

  • 品質KPIとセキュリティKPIを左右に並べる構成で、異常の相関を即座に読み取れる
  • Rule Trigger Heatmapにより、複数ルールに同時抵触するユーザーを視覚的に特定できる
  • データソースは構造化JSONログを集約ストア（Elasticsearch等）に取り込むだけで実現可能

監視の運用コスト見積もり観点

RAG監視パイプラインの運用コストは主に「ログ量」と「アラート疲れ」の2軸で決まる。 設計段階でこの2点を定量的に見積もらないと、本番稼働後に収集コストの増大やアラート無視（Alert Fatigue）が発生する。

見積もり項目計算式の目安チューニング方針
1日あたりのログ量 クエリ数/日 × 平均イベント数/クエリ（3〜5件） × 1イベント平均2KB sensitivity_label=publicはサンプリング率10%に落として保存コスト削減
アラート発火率 ルール数 × 誤検知率 × クエリ数/日 High優先度ルールは閾値を2週間のベースラインから設定（固定値禁止）
ルール評価CPU スライディングウィンドウはO(1)。Redis利用でステートレスに水平スケール RedisのkeyはユーザーID:rule_id、TTLをウィンドウ幅に設定して自動削除
RAGASメトリクス計算コスト LLMベースのRAGAS評価は高コスト。全件評価は不要 全クエリの5〜10%をランダムサンプリングして評価（重要ユーザーは100%）
アラート疲れ防止 Mediumアラートの重複抑制：同一rule+user_idで30分以内の再発火は抑制 週次でFP/TP比を集計し、FP率30%超のルールは閾値を自動引き上げ

アラート疲れを防ぐ最も効果的な対策は、閾値を固定値ではなくベースライン相対値で設定することだ。 たとえばR-03（クエリ異常頻度）の閾値「60秒に30クエリ超」は業務システムの種類によって最適値が大きく異なるため、2週間の移動平均の3σを閾値として動的に更新する設計を採用する。 またRAGASメトリクスの計算にはLLM APIコストが発生するため、全リクエストの5〜10%をランダムサンプリングし、suspicious_flagが立ったリクエストのみ100%評価するハイブリッド戦略が現実的だ。

ここまでのまとめ

  • ログコストはsensitivity_labelごとのサンプリング率で制御し、restrictedのみ全量保存する
  • 閾値は固定値ではなくベースライン相対値（移動平均±3σ）で動的設定する
  • RAGASメトリクスの計算はサンプリング＋フラグ付きリクエスト全件の2段階戦略で費用を最適化する

シリーズ内リンク