LLMアプリのモニタリング・オブザーバビリティ：本番運用で必要な監視設計

制作&開発

LLMアプリを本番環境にリリースした後、何を監視すればいいか把握できていますか。従来のWebアプリとは異なり、LLMアプリにはAPIコスト・レイテンシ・ハルシネーション率・プロンプトの品質など、固有の指標を継続的に追う必要があります。この記事では、LLMアプリのオブザーバビリティ設計を具体的に解説します。

LLMモニタリングで追跡すべき指標

LLMアプリの監視は「コスト」「パフォーマンス」「品質」の3軸で設計します。それぞれ何を見ればいいか整理しておきましょう。

コスト指標

  • 入力トークン数 / 出力トークン数（1リクエストあたり・1日あたり）
  • API費用の推移（モデル・機能・ユーザーセグメント別）
  • コスト異常の検知（突然のスパイクはプロンプトインジェクション等の兆候）

パフォーマンス指標

  • TTFT（Time to First Token）：ストリーミング開始までの時間
  • 全体レイテンシ（P50・P95・P99）
  • エラー率（429 Rate Limit・500 Server Error等）

品質指標

  • ユーザーフィードバック（👍👎）の比率
  • LLM-as-judgeによる自動品質評価スコア
  • ハルシネーション検出率
  • コンテキスト利用率（RAGの場合：取得ドキュメントが実際に使われているか）

Langfuseの導入

LangfuseはオープンソースのLLMオブザーバビリティプラットフォームです。セルフホスティングまたはクラウドサービスとして使えます。デコレータを一つ付けるだけでトレースが自動的に記録される点が便利です。

pip install langfuse anthropic

# .env
LANGFUSE_PUBLIC_KEY=pk-...
LANGFUSE_SECRET_KEY=sk-...
LANGFUSE_HOST=https://cloud.langfuse.com  # セルフホストの場合は自前のURL
import anthropic
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse()
client = anthropic.Anthropic()

@observe()  # このデコレータでLangfuseにトレースが送られる
def generate_response(user_message: str, session_id: str) -> str:
    langfuse_context.update_current_trace(
        session_id=session_id,
        tags=["production", "chat"]
    )

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}]
    )

    text = response.content[0].text if response.content else ""

    # メタデータを記録
    langfuse_context.update_current_observation(
        usage={
            "input": response.usage.input_tokens,
            "output": response.usage.output_tokens,
        }
    )

    return text

構造化ログの実装

Langfuse等のツールを使わない場合は、構造化ログでLLMの動作を記録しましょう。JSON形式で出力すればCloudWatch LogsやDatadogで検索しやすくなります。

import json
import time
import logging
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class LLMLog:
    timestamp: str
    model: str
    user_id: Optional[str]
    session_id: str
    input_tokens: int
    output_tokens: int
    latency_ms: int
    cost_usd: float
    error: Optional[str] = None
    quality_score: Optional[float] = None

logger = logging.getLogger("llm")
logger.setLevel(logging.INFO)

# JSON形式でログ出力（CloudWatch Logs・Datadogで検索しやすい）
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(message)s'))
logger.addHandler(handler)

def log_llm_call(log: LLMLog):
    logger.info(json.dumps(asdict(log)))

# 使用例
start = time.time()
try:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}]
    )
    latency = int((time.time() - start) * 1000)

    # Claude Sonnet 4.6の概算コスト（実際は公式料金を参照）
    input_cost = response.usage.input_tokens * 3 / 1_000_000
    output_cost = response.usage.output_tokens * 15 / 1_000_000

    log_llm_call(LLMLog(
        timestamp=time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        model="claude-sonnet-4-6",
        user_id=user_id,
        session_id=session_id,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        latency_ms=latency,
        cost_usd=input_cost + output_cost,
    ))
except Exception as e:
    log_llm_call(LLMLog(
        timestamp=time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        model="claude-sonnet-4-6",
        user_id=user_id,
        session_id=session_id,
        input_tokens=0,
        output_tokens=0,
        latency_ms=int((time.time() - start) * 1000),
        cost_usd=0,
        error=str(e),
    ))
    raise

LLM-as-judgeによる品質自動評価

実は、LLMを使ってLLMの出力品質を自動評価できます。評価用の安いモデル（例：claude-haiku）を使えば、コストを抑えながら継続的な品質監視が実現します。

def evaluate_response_quality(
    question: str,
    context: str,
    answer: str,
    eval_model: str = "claude-haiku-4-5-20251001"
) -> dict:
    """回答品質を0-10のスコアで評価する"""
    prompt = f"""以下のQAペアの品質を評価してください。

質問: {question}
コンテキスト: {context}
回答: {answer}

以下の観点で各10点満点で評価し、JSONで返してください：
- faithfulness: コンテキストに基づいた回答か（ハルシネーションがないか）
- relevance: 質問に的確に答えているか
- completeness: 必要な情報を網羅しているか

JSON形式のみ出力してください：
{{"faithfulness": 数値, "relevance": 数値, "completeness": 数値, "overall": 数値}}"""

    response = client.messages.create(
        model=eval_model,
        max_tokens=256,
        messages=[{"role": "user", "content": prompt}]
    )

    text = response.content[0].text
    return json.loads(text)

アラートの設定

監視の仕上げは、異常を自動検知するアラートの設定です。次の4つを最低限設定しておきましょう。

  • コストアラート：1時間のAPI費用が閾値を超えたら通知
  • エラー率アラート：5分間のエラー率が5%を超えたら通知
  • レイテンシアラート：P95レイテンシが10秒を超えたら通知
  • 品質アラート：LLM-as-judgeのスコアが閾値を下回ったら通知

Datadog・CloudWatch・Grafana等の監視ツールのメトリクスにログを送り、アラートを設定します。

まず構造化ログとコストアラートから始めよう

LLMアプリのモニタリングはコスト・パフォーマンス・品質の3軸で設計します。Langfuseなどの専用ツールを使えばトレーシングが自動化されます。まず取り組むなら、構造化ログの出力とコスト異常・エラー率のアラート設定から始めるのが現実的です。品質監視はLLM-as-judgeで自動化することで、継続的な改善サイクルを回せます。

よくある質問

Langfuseのセルフホスティングとクラウド版の違いは？

クラウド版は無料枠（月50,000イベント）で始められ、セットアップが不要です。セルフホスティングはDockerで簡単に立ち上げられ、データを自社インフラに保持できます。機密性の高いプロンプト・ユーザーデータを扱う場合はセルフホスティングをお勧めします。

モニタリングのオーバーヘッドはLLMのレイテンシに影響しますか？

Langfuseは非同期でデータを送信するため、通常レイテンシへの影響は1ms未満です。構造化ログも同様に非同期処理することでオーバーヘッドを最小化できます。

スポンサーリンク

コメント

タイトルとURLをコピーしました