たきびAIラボ TAKIBI · AI · LAB
🛡️サイバーセキュリティ ハウツー 公開 2026.06.07

LLMエージェントのトレース異常検知でリソース悪用を早期発見する

OpenTelemetry × Langfuse × Prometheus によるセキュリティ監視設計ガイド

OWASP LLM10:2025の緩和策として明示される「ロギング・モニタリングと異常検知」を実装する。OpenTelemetry Gen AI spans × Langfuse セッション集計 × Prometheus Z スコアアラートの3層パイプラインで、LLMエージェントのリソース悪用を事後検知する方法をコード例付きで解説。

読了 約25分
LLMエージェントのトレース異常検知でリソース悪用を早期発見する:OpenTelemetry × Langfuse × Prometheus によるセキュリティ監視設計ガイド

LLMエージェントを本番運用していると、「なんかいつもよりトークン消費量が多い気がする」という感覚を持ったことはないだろうか。予算上限は設定してある、サーキットブレーカーもある——しかし、それらは遮断の仕組みだ。いつ・どのユーザーが・どのツールを何回呼んで・どれだけトークンを消費したかを時系列で把握し、統計的に異常なセッションをインシデントとして検知する監視基盤は、別に作る必要がある。

本記事が扱うのは「事後検知・監視」という切り口だ。OWASP LLM10:2025「Unbounded Consumption(無制限の消費)」は、緩和策の一つとして「ロギングとモニタリング、異常検知の実装」を明示しているが、具体的な実装ガイドは少ない。ここでは次の3ステップを実装コード付きで解説する。

  1. OpenTelemetry Gen AI semantic conventions によるエージェントトレースの構造化
  2. Langfuse Session/Trace API を使ったセッション横断のトークン累積集計
  3. Prometheus × Grafana へのエクスポートと Z スコアベースのアラートルール設定

「既存のゲートウェイがあるのに、なぜ別途トレース監視が必要か」という疑問への答えも最初に整理する。


なぜゲートウェイのレート制限だけでは不十分なのか

APIゲートウェイでトークン消費量ベースのレート制限を設定すると、「1ユーザーが1分あたり100,000トークンを超えたらブロック」のような静的ルールが動く。これは有効だ——が、次の問題には対応できない。

緩やかな消費パターンの異常を見逃す 攻撃者やバグが「静的な上限を超えない速度」でじわじわリソースを消費し続けると、レート制限は発動しない。ある正規ユーザーが平均1セッション2,000トークンのところ、今日のセッションは18,000トークンを使っている——この種の「基準値からの統計的逸脱」は、固定閾値では捕捉できない。

ツール呼び出しパターンの異常が見えない ReAct型エージェントは「Thought→Action→Observation」を繰り返す。正常なタスクであれば3〜5サイクルで完了するところ、あるセッションが同一ツールを50回呼び出し続けている——これはループバグの可能性が高い。しかしゲートウェイはHTTPレイヤの流量しか見ておらず、エージェント内部の行動パターンは追跡できない。

マルチエージェント構成で消費経路が分散する オーケストレーターからサブエージェントへタスクが委譲されると、各エージェントが個別にAPIを呼ぶため、1エンティティ単位のゲートウェイ制限は簡単に分散逃れされる。セッション全体を縦断するトレースを持たなければ、合算消費量を把握できない。

ここでOWASP LLM10:2025が緩和策として挙げる「ログとモニタリングの実装」が意味を持ってくる。構造化されたトレースとセッション集計、そして統計的異常検知がそろってはじめて、多層防御の「検知層」が機能する。


Step 1:OpenTelemetry Gen AI spans でトレースを構造化する

Gen AI セマンティクス規約の概要

OpenTelemetry の Gen AI Semantic Conventions(2024年末から1.x系で安定化が進んでいる)は、LLM推論呼び出しに関する標準属性名を定めている。監視基盤を構築するうえで押さえるべき主要属性を示す。

属性名説明
gen_ai.systemstringモデルプロバイダ(openai, anthropic, azure.openai 等)
gen_ai.operation.namestring操作種別(chat, text_completion, embeddings
gen_ai.request.modelstring要求モデル名(gpt-4o, claude-opus-4 等)
gen_ai.response.modelstring実際に使われたモデル名
gen_ai.usage.input_tokensint消費された入力トークン数
gen_ai.usage.output_tokensint消費された出力トークン数
gen_ai.usage.total_tokensint合計トークン数

エージェントフレームワーク(LangChain、LlamaIndex、Haystack など)の多くは、OpenTelemetry エクスポーターを有効にするだけでこれらの属性を自動付与する。カスタム実装の場合は手動でスパンに記録する必要がある。

Python での実装例(LangChain + OpenTelemetry)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.langchain import LangchainInstrumentor

# Langfuse の OTLP エンドポイントに直接送出する設定
LANGFUSE_PUBLIC_KEY = "pk-lf-..."
LANGFUSE_SECRET_KEY = "sk-lf-..."
LANGFUSE_HOST = "https://cloud.langfuse.com"

import base64
auth = base64.b64encode(
    f"{LANGFUSE_PUBLIC_KEY}:{LANGFUSE_SECRET_KEY}".encode()
).decode()

exporter = OTLPSpanExporter(
    endpoint=f"{LANGFUSE_HOST}/api/public/otel/v1/traces",
    headers={"Authorization": f"Basic {auth}"},
)

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

# LangChain の全LLM呼び出しを自動インストルメント
LangchainInstrumentor().instrument()

この設定だけで、LangChain を通じたすべての LLM 呼び出しが gen_ai.usage.input_tokens などの属性付きスパンとして Langfuse に送られる。

セッションとユーザーの紐付け

トレースに session.iduser.id を付与すると、後でセッション横断の集計が可能になる。

tracer = trace.get_tracer("my-agent")

def run_agent_session(user_id: str, session_id: str, user_input: str):
    with tracer.start_as_current_span("agent.session") as span:
        span.set_attribute("session.id", session_id)
        span.set_attribute("user.id", user_id)
        span.set_attribute("user.input", user_input[:200])  # 先頭200文字だけ記録
        # ... エージェントの実行処理

Step 2:Langfuse でセッション横断のトークン累積を集計する

Langfuse の Session API を使った集計

Langfuse は同一 session_id を持つトレースをグループ化し、セッション単位のトークン合計を参照できる。これを Python SDK で取得して集計する。

import langfuse
from datetime import datetime, timedelta, timezone
from collections import defaultdict

lf = langfuse.Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="https://cloud.langfuse.com",
)

def get_user_token_stats(
    lookback_hours: int = 24,
    page_size: int = 100,
) -> dict[str, dict]:
    """直近N時間のユーザー別トークン消費統計を返す"""
    since = datetime.now(timezone.utc) - timedelta(hours=lookback_hours)
    stats: dict[str, dict] = defaultdict(
        lambda: {"total_tokens": 0, "session_count": 0, "trace_count": 0}
    )

    page = 1
    while True:
        traces = lf.fetch_traces(
            from_timestamp=since,
            page=page,
            limit=page_size,
        ).data

        if not traces:
            break

        for t in traces:
            uid = t.user_id or "anonymous"
            usage = t.usage  # UsageDetails オブジェクト
            if usage:
                stats[uid]["total_tokens"] += usage.total or 0
                stats[uid]["trace_count"] += 1
            # session 単位の重複排除
            if t.session_id:
                stats[uid].setdefault("sessions", set()).add(t.session_id)

        page += 1

    # sessions set を件数に変換
    for v in stats.values():
        v["session_count"] = len(v.pop("sessions", set()))

    return dict(stats)

ツール呼び出し回数の集計

ReAct エージェントのアクションステップはスパンとして記録される。ツール呼び出し回数を集計するには、スパン名や gen_ai.operation.nametool のものを数える。

def get_tool_call_counts(session_id: str) -> dict[str, int]:
    """セッション内のツール別呼び出し回数を返す"""
    traces = lf.fetch_traces(session_id=session_id).data
    counts: dict[str, int] = defaultdict(int)

    for trace in traces:
        for obs in lf.fetch_observations(trace_id=trace.id).data:
            # LangChain の tool 呼び出しは type == "tool" で記録される
            if obs.type == "TOOL":
                tool_name = obs.name or "unknown"
                counts[tool_name] += 1

    return dict(counts)

これにより「あるセッションで web_search が47回呼ばれている」「execute_code が通常の10倍の頻度で呼ばれている」といった異常パターンを検出できる。


Step 3:Prometheus へのエクスポートと Z スコアアラート設計

Prometheus Pushgateway へのメトリクス送出

集計した統計を Prometheus に送るには、prometheus_client ライブラリの Pushgateway 機能を使う。

from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
import math

PUSHGATEWAY_URL = "http://prometheus-pushgateway:9091"

def push_user_metrics(stats: dict[str, dict]) -> None:
    registry = CollectorRegistry()

    g_tokens = Gauge(
        "llm_agent_user_total_tokens_24h",
        "直近24時間のユーザー別トークン消費合計",
        labelnames=["user_id"],
        registry=registry,
    )
    g_sessions = Gauge(
        "llm_agent_user_session_count_24h",
        "直近24時間のユーザー別セッション数",
        labelnames=["user_id"],
        registry=registry,
    )

    for uid, s in stats.items():
        g_tokens.labels(user_id=uid).set(s["total_tokens"])
        g_sessions.labels(user_id=uid).set(s["session_count"])

    push_to_gateway(PUSHGATEWAY_URL, job="llm_agent_monitor", registry=registry)

このスクリプトを cron で5〜15分ごとに実行するか、FastAPI などの軽量サービスとして常駐させる。

Z スコアベースの異常検知アラートルール

静的閾値ではなく Z スコア(標準偏差に基づく逸脱度)で異常を検知すると、ユーザーごとの通常利用量の差異を吸収しつつ異常を捕捉できる。

Grafana の Alerting Rules(PromQL)で Z スコア計算を行う例を示す。

# 直近7日間のユーザー別トークン消費の平均(μ)と標準偏差(σ)を計算
# ※ recording rule として事前に定義しておく

# μ: 7日間の1日あたり平均
avg_over_time(llm_agent_user_total_tokens_24h[7d])

# σ: 7日間の標準偏差
stddev_over_time(llm_agent_user_total_tokens_24h[7d])

# Z スコア = (現在値 - μ) / σ
# アラート条件: Z スコアが 3.0 を超えたユーザーを検知
(
  llm_agent_user_total_tokens_24h
  - avg_over_time(llm_agent_user_total_tokens_24h[7d])
) /
(
  stddev_over_time(llm_agent_user_total_tokens_24h[7d]) + 1
)
> 3.0

分母に + 1 を加えるのは、σ=0(全日同じ消費量)でのゼロ除算を防ぐためだ。

検知すべき3つの異常パターン

BUYING GUIDE

監視設計の前に確認すること

  1. 01

    ベースライン期間

    Z スコア計算には最低7日分(理想は30日分)のデータが必要。新規ユーザーや新規エージェントには静的閾値を先に適用する

  2. 02

    ノイズ許容度

    Z スコア閾値を低くすると誤検知が増える。まず3σ(99.7%信頼区間)で様子を見て、誤検知率に応じて調整する

  3. 03

    セッション定義

    Langfuse の session_id は呼び出し元で付与する必要がある。エージェントのセッション管理コードと整合していることを確認

  4. 04

    Pushgateway の可用性

    Prometheus Pushgateway が落ちると集計が途切れる。アラート基盤そのものの死活監視も設定すること

  5. 05

    個人情報の取り扱い

    user_id がメール等の個人情報の場合、Prometheus ラベルに直接使わずハッシュ化する

パターン1:per-user トークン消費の Z スコア異常 上記PromQLが検知する。通常の3倍以上のトークンを消費しているユーザーをアラート。誤検知を減らすため、Z スコア > 3.0 かつ絶対値 > 10,000 トークン の両条件を組み合わせる。

パターン2:ツール呼び出し回数スパイク 単一セッションでの同一ツール呼び出し回数が閾値(例: 20回)を超えた場合。これはエージェントのループバグや、エージェントを操作してツールを大量実行させるプロンプトインジェクション攻撃の兆候になる。

llm_agent_tool_call_count_per_session > 20

パターン3:コンテキスト長パーセンタイル逸脱 input_tokens が全セッションの99パーセンタイルを超えている。非常に長いプロンプトを送ることでLLMに大量の処理をさせる「sponge attack」(次節で詳述)の検知に有効。

histogram_quantile(0.99,
  sum by (le) (rate(llm_agent_input_tokens_bucket[1h]))
)

SHIELD 論文と sponge attack の検知

2026年1月に公開された SHIELD(arXiv 2601.19174)は、LLMエージェントへのsponge attack(スポンジ攻撃)に対するセキュリティフレームワークを提案している。

sponge attack とは

sponge attack は、計算資源(CPU・GPU・メモリ・トークン)を最大限に吸収させることを目的とした入力を用いる攻撃だ。具体的には次のようなパターンが知られている。

  • プロンプト長の爆発: 再帰的な自己参照や反復フレーズを含む入力で、コンテキスト長を意図的に膨張させる
  • ツールチェーン操作: エージェントが無限に近いツール呼び出しループに入るよう誘導する
  • 間接的プロンプトインジェクション: 外部ソース(Webページ・ドキュメント)経由でエージェントに長大なタスクを実行させる

SHIELD 論文は、これらの攻撃を検知するための特徴量としてトークン消費速度の変化率ツール呼び出し頻度の統計的逸脱を提案している(査読状況は2026年6月時点で未確定。arXiv プレプリント段階)。

静的ルールベース vs. 統計ベース異常検知の使い分け

検知アプローチ向いている場面限界
静的ルール(閾値超過)明確な上限を設けたい場合(例: 1セッション最大50,000トークン)緩やかな逸脱・新パターンは見逃す
Z スコア統計検知ユーザーの通常行動から大きく外れた消費を検知したい場合ベースラインデータが必要。急激な正常利用増加でも誤検知する可能性
パーセンタイル監視全体の傾向の中での外れ値を見つけたい場合スロースタート型の逸脱は検知しにくい
ツール呼び出し系列分析ループ・sponge attack の行動パターンを検知したい場合実装コストが高い。十分なトレースデータが必要

実務では、まず静的ルールとZ スコアを組み合わせた二層検知から始め、十分なデータが蓄積した段階でツール呼び出しパターン分析を追加するのが現実的だ。


実務での使い方:インシデント対応フローへの統合

アラート受信後の調査手順

Grafana からアラートが発火した後、Langfuse で対象セッションを深掘りする流れを示す。

def investigate_anomalous_session(user_id: str, session_id: str) -> dict:
    """異常セッションの詳細調査レポートを生成する"""
    report = {
        "user_id": user_id,
        "session_id": session_id,
    }

    # 1. セッション内のトレース一覧を取得
    traces = lf.fetch_traces(session_id=session_id).data
    report["trace_count"] = len(traces)
    report["total_tokens"] = sum(
        (t.usage.total or 0) for t in traces if t.usage
    )

    # 2. ツール呼び出し回数を集計
    report["tool_calls"] = get_tool_call_counts(session_id)

    # 3. 時系列でのトークン増加パターン(最初のトレースと最後のトレースを比較)
    if len(traces) >= 2:
        first_ts = traces[0].start_time
        last_ts = traces[-1].start_time
        elapsed_minutes = (last_ts - first_ts).total_seconds() / 60
        report["session_duration_minutes"] = elapsed_minutes
        report["tokens_per_minute"] = (
            report["total_tokens"] / elapsed_minutes if elapsed_minutes > 0 else 0
        )

    return report

この調査レポートをもとに、対応レベルを判断する。

調査結果推定原因推奨対応
特定ツールが20回以上呼ばれているエージェントのループバグエージェントのプロンプト・ツール定義の見直し
tokens_per_minute が全体99%ile超sponge attack または悪意ある入力セッションの即時停止、入力フィルタリングの強化
複数ユーザーで同時に Z スコア異常新機能リリース・システム変更ベースラインのリセット
特定ユーザーのみで数日継続異常な利用パターンアカウント確認・利用停止の検討

プロンプト例:異常調査時の ChatGPT/Claude への質問

トレースデータをエクスポートして LLM に分析させる場面では、次のプロンプトが役立つ。

以下のLLMエージェントのトレースデータを分析してください。

セッションID: {session_id}
総トークン消費: {total_tokens}
ツール呼び出し回数: {tool_calls}
セッション時間: {duration_minutes}分

このセッションで次の観点から異常の有無を評価してください:
1. ツール呼び出し回数が通常パターン(3〜5回)と比較して著しく多くないか
2. トークン消費速度が急上昇・急下降していないか
3. 繰り返しのツール呼び出し(ループ兆候)がないか

分析結果を「問題なし / 要調査 / 即対応必要」の3段階で示し、
根拠とともに推奨アクションを提案してください。

注意点と設計上の落とし穴

トレースデータの保持期間とコスト

Langfuse Cloud は全トレースを保存するが、大量のエージェントが動くシステムでは月次のストレージコストが無視できなくなる。セキュリティ目的の保持には最低30日を推奨するが、詳細スパン(個別ステップのログ)は7日、集計メトリクスは90日という階層別保持ポリシーを設計すると現実的だ。

Prometheus の高カーディナリティ問題

user_id をそのままラベルにすると、ユーザー数の増加に比例して Prometheus のメモリ消費が増大する(高カーディナリティ問題)。ユーザー数が1万を超えるシステムでは、Prometheus ではなく Victoria Metrics や Grafana Mimir の採用を検討するか、ユーザー粒度の集計は Langfuse や専用の分析DBに委ねる設計にする。

Pushgateway の Staleness 問題

Pushgateway はスクレイプ型ではなくプッシュ型のため、スクリプトが停止してもメトリクスが残り続ける。--push.disable-consistency-check を使わず、定期的に古いグループを DELETE するジョブを設けること。


よくある質問(FAQ)

FAQ

よくある質問

クリックで展開。

Datadog や New Relic でも同様の監視はできますか?

はい。Datadog は OpenTelemetry のネイティブ取り込みをサポートしており、gen_ai.* スパン属性をそのままメトリクスに変換できます。New Relic も OTLP エンドポイントを提供しています。Langfuse の代わりに直接これらへ送出することも可能です。ただし、エージェントフレームワーク固有のトレース表示(親スパン・子スパンの視覚的な階層表示)は Langfuse の方が充実しているため、調査用UIとしてLangfuseを残しつつ、アラート基盤は既存のAPMツールに統合するというハイブリッド構成も現実的です。

OpenAI の公式SDKは Gen AI セマンティクス規約に対応していますか?

2026年6月時点で、OpenAI の公式 Python SDK は opentelemetry-instrumentation-openai サードパーティパッケージを通じてインストルメントできます。LangChain・LlamaIndex・Haystack などの主要フレームワークは独自の OTel インストルメンテーションを内包しており、gen_ai.usage.input_tokens などの属性を自動付与します。直接 openai パッケージを使っている場合は opentelemetry-instrumentation-openai の追加が必要です。

Z スコアのベースラインデータが不足している新規サービスではどうすればいいですか?

サービス開始直後はデータが少なく Z スコア計算が不安定になります。最初の2〜4週間は静的閾値(例: 1セッションで50,000トークンを超えたらアラート)で運用し、データが蓄積したら Z スコアに切り替えるのが実用的です。また、類似サービスや社内ベンチマークがあれば、その統計をシードとして初期μ・σを設定する方法もあります。

Langfuse のセルフホスト版と Cloud 版のどちらを選ぶべきですか?

セキュリティ要件が高い(ユーザー入力を外部に出せない)場合はセルフホスト版を選びます。Langfuse はオープンソース(MIT/ELv2ライセンス)で Docker Compose や Kubernetes でデプロイできます。Cloud 版はセットアップが簡単で、Prometheus エクスポーターは組み込みで提供されているため、スモールスタートに向いています。本番環境でのデータ主権が求められる場合は、入力・出力のマスキング機能を有効にしたうえでセルフホストを推奨します。

マルチエージェント構成でセッション IDを伝播させる方法は?

LangChain の場合、RunnableConfigcallbacks を通じてトレーサーを子エージェントに引き継ぎます。OpenTelemetry のコンテキスト伝播(W3C TraceContext または Baggage)を使う場合、traceparent ヘッダーをエージェント間のHTTP呼び出しに付与することで、親トレースと子トレースが同一セッションとして紐づきます。セッションIDは独自のBaggageエントリとして伝播するか、すべてのエージェント実行時に共通の session_id を環境変数や引数として渡す設計が扱いやすいです。


まとめ

OWASP LLM10:2025「Unbounded Consumption」の緩和策として、本記事では次の3層監視パイプラインを解説した。

  1. OpenTelemetry Gen AI semantic conventions でトレースを構造化し、gen_ai.usage.input_tokens / output_tokens / total_tokens をスパン属性として記録
  2. Langfuse Session/Trace API でセッション横断のトークン累積とツール呼び出しパターンを集計
  3. Prometheus × Grafana に送出し、Z スコアベースの per-user 異常・ツール呼び出し回数スパイク・コンテキスト長パーセンタイル逸脱を検知するアラートルールを設定

静的なレート制限・予算上限・サーキットブレーカーが「遮断」の仕組みであるのに対し、本記事のアプローチは統計的逸脱として異常を検知し、インシデント調査の手がかりを提供する。二つのアプローチは相補的であり、多層防御の「検知層」を厚くすることで、バグ・ループ・sponge attack・内部不正のいずれに対しても対応力が上がる。

最初のステップとして、LangchainInstrumentor().instrument() の一行を追加するだけで OpenTelemetry との統合は始まる。まずトレースを可視化し、ベースラインデータを蓄積することから始めてほしい。


次に読むおすすめ

セキュリティ監視の設計をさらに深めたい方に、関連コンテンツを紹介する。

たきびAIラボ note:主要AIツール5つの実務比較 — LLMエージェント導入時のツール・スタック選定で迷っている方に。実費課金ベースでの比較が、コスト最適化の観点から監視設計の優先度を判断する参考になります。


関連記事


参考リンク