AIエージェント入門

【2026年4月】LLM可観測性 実装ガイド|トレース・評価・コスト最適化

この記事の結論

LLMアプリの本番運用で必須の可観測性を実装する方法をPythonコード付きで解説。OpenTelemetry/Langfuseによるトレース、DeepEvalによる評価、コスト・レイテンシ計測、本番アラート設計まで網羅。

「LLMアプリを本番にデプロイしたはいいが、何が起きているのかまったく分からない」

先日、あるプロジェクトでこんな状況に直面しました。GPT-4o系のLLMを組み込んだRAGシステムを本番公開したところ、一部のユーザーから「回答がおかしい」「遅い」というフィードバックが届き始めたのです。しかし手元には何もデータがない。どのリクエストで問題が起きているのか、ハルシネーションが増えているのか、コストが予算を超えそうなのか——何も見えない状態でした。

この経験から痛感したのは、LLMアプリに「可観測性(Observability)」を最初から組み込むことの重要性です。従来のAPM(Application Performance Monitoring)では、トークン消費・評価スコア・ハルシネーション率という概念が存在しないため、LLM固有の計測設計が必要になります。

この記事では、実際の本番運用で効果があったLLM可観測性の実装パターンを、コピペ可能なPythonコード例つきで全公開します。OpenTelemetry / Langfuse によるトレース、DeepEval によるリアルタイム評価、コスト・レイテンシの監視、本番アラート設計まで順に解説します。


可観測性の3層モデル|何を計測するか

LLMアプリの可観測性を設計するとき、「とりあえずログを取る」ではなく、3つの層に分けて考えると整理しやすいです。

レイヤー 計測対象 代表的なツール 優先度
Layer 1: トレース リクエスト/レスポンス・スパン・ネスト構造 OpenTelemetry, Langfuse 最高
Layer 2: メトリクス トークン数・コスト・レイテンシ・エラー率 Prometheus, OpenTelemetry Metrics
Layer 3: 評価 ハルシネーション率・Faithfulness・回答品質 DeepEval, RAGAS 中〜高

Layer 1→2→3 の順に導入するのが、実運用でのおすすめです。トレースなしに評価だけ入れても、問題の根本原因にたどり着けません。

LLM固有の計測項目チェックリスト

  • 入力トークン数 / 出力トークン数(モデルごと)
  • Time to First Token (TTFT) / 全レスポンス時間
  • プロンプトテンプレートのバージョン
  • 使用モデル名・バージョン
  • リトライ回数 / エラー種別
  • RAGの場合: 取得ドキュメント数・Rerank後スコア
  • ユーザーセッションID(マルチターンの場合)
  • 機能フラグ(A/Bテスト用)

Layer 1: トレース実装|OpenTelemetry + Langfuse Python SDK

2026年現在、LLMトレーシングのデファクトスタンダードは OpenTelemetry(OTel) です。2026年初頭に GenAI セマンティックコンベンション(トークン数・モデルパラメータ等の命名規則)が Stable に昇格し、主要バックエンド(Langfuse・Jaeger・Grafana Tempo等)がすべてOTel形式を受け付けるようになりました。

Langfuse Python SDK v3(2025年6月 GA)はOTelネイティブに書き直されており、既存のOTelインフラと自然に統合できます。

環境セットアップ

# 必要パッケージのインストール
pip install langfuse openai opentelemetry-api opentelemetry-sdk

コード例1: Langfuse SDK v3 によるトレース実装

動作環境: Python 3.11+, langfuse>=3.0.0, openai>=1.30.0

import os
from langfuse import get_client
from openai import OpenAI

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# 環境変数で認証情報を管理すること(ハードコード禁止)
# 必要な環境変数:
#   LANGFUSE_SECRET_KEY=sk-lf-...
#   LANGFUSE_PUBLIC_KEY=pk-lf-...
#   LANGFUSE_BASE_URL=https://cloud.langfuse.com
#   OPENAI_API_KEY=sk-...

langfuse = get_client()
openai_client = OpenAI()

def answer_question(user_question: str, user_id: str) -> str:
    """
    ユーザーの質問に回答し、Langfuseで全スパンをトレースする関数。
    """
    # ルートトレースを開始(ユーザーリクエスト全体をカバー)
    with langfuse.start_as_current_observation(
        as_type="span",
        name="answer-question",
        user_id=user_id,
        metadata={"source": "api"},
    ) as root_span:

        # LLM生成スパン(ネストされる)
        with langfuse.start_as_current_observation(
            as_type="generation",
            name="llm-generation",
            model="gpt-4o",
            input=[{"role": "user", "content": user_question}],
        ) as generation:

            response = openai_client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": user_question}],
            )
            answer = response.choices[0].message.content

            # トークン数を明示的にスパンに記録
            generation.update(
                output=answer,
                usage={
                    "input": response.usage.prompt_tokens,
                    "output": response.usage.completion_tokens,
                    "total": response.usage.total_tokens,
                },
            )

        root_span.update(output=answer)

    # スクリプト終了時は必ずflushを呼ぶ(バッファをフラッシュ)
    langfuse.flush()
    return answer

# 使用例
result = answer_question("RAGとFine-tuningの違いは?", user_id="user-001")
print(result)

ポイント:

  • start_as_current_observation はコンテキストマネージャーで、自動的にネスト構造を作る
  • as_type="generation" を指定するとLangfuseのUIでLLM呼び出しとして表示される
  • usage にトークン数を渡すとコスト計算が自動で行われる
  • スクリプト終了前に langfuse.flush() を呼ばないとデータが送信されないことがある

Layer 2: コスト・レイテンシ計測の実装

トレースで「何が起きたか」は分かりますが、「いくらかかったか」「どれだけ遅いか」を継続的に監視するにはメトリクスの別レイヤーが必要です。OpenTelemetry の Metrics APIを使い、Prometheus や Grafana と連携するパターンが2026年の本番環境でよく使われています。

コード例2: OpenTelemetry によるコスト・レイテンシ計測

動作環境: Python 3.11+, opentelemetry-api>=1.20.0, opentelemetry-sdk>=1.20.0, openai>=1.30.0

import time
from openai import OpenAI
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import (
    ConsoleMetricExporter,
    PeriodicExportingMetricReader,
)

# 注意: 本番環境では ConsoleMetricExporter を OTLPMetricExporter に置き換えてください。

# メトリクスプロバイダーの初期化(30秒ごとにエクスポート)
reader = PeriodicExportingMetricReader(
    ConsoleMetricExporter(), export_interval_millis=30_000
)
provider = MeterProvider(metric_readers=[reader])
metrics.set_meter_provider(provider)
meter = metrics.get_meter("llm-cost-tracker", "1.0.0")

# メトリクス定義(OTel GenAI セマンティックコンベンション準拠)
input_token_histogram = meter.create_histogram(
    name="gen_ai.usage.input_tokens",
    description="入力トークン数",
    unit="tokens",
)
output_token_histogram = meter.create_histogram(
    name="gen_ai.usage.output_tokens",
    description="出力トークン数",
    unit="tokens",
)
cost_histogram = meter.create_histogram(
    name="gen_ai.usage.cost",
    description="API呼び出しコスト",
    unit="USD",
)
latency_histogram = meter.create_histogram(
    name="gen_ai.latency",
    description="LLM応答時間",
    unit="ms",
)

# モデルごとの料金テーブル(100万トークンあたりのUSD)
# 最終確認日: 2026-04-28(公式APIドキュメントより)
MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "claude-3-5-sonnet-20241022": {"input": 3.00, "output": 15.00},
    "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25},
}

def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
    return (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1_000_000

def tracked_llm_call(
    prompt: str,
    model: str = "gpt-4o",
    feature: str = "default",
) -> str:
    """
    LLM呼び出しをラップし、コスト・レイテンシをOTelに記録する。
    feature: 機能別コスト分析に使う(例: "rag-pipeline", "summarizer")
    """
    client = OpenAI()
    attributes = {"gen_ai.system": "openai", "gen_ai.request.model": model, "feature": feature}

    start = time.perf_counter()
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    elapsed_ms = (time.perf_counter() - start) * 1000

    in_tokens = response.usage.prompt_tokens
    out_tokens = response.usage.completion_tokens
    cost = calculate_cost(model, in_tokens, out_tokens)

    # メトリクスに記録
    input_token_histogram.record(in_tokens, attributes=attributes)
    output_token_histogram.record(out_tokens, attributes=attributes)
    cost_histogram.record(cost, attributes=attributes)
    latency_histogram.record(elapsed_ms, attributes=attributes)

    return response.choices[0].message.content

# 使用例
answer = tracked_llm_call(
    "OpenTelemetryとは何ですか?", model="gpt-4o-mini", feature="qa-endpoint"
)

ポイント:

  • feature 属性を付けることで「どの機能でコストがかかっているか」を Grafana などで分析できる
  • 料金テーブルは定期的に公式ドキュメントで更新確認が必要(モデル変更時は必ず更新)
  • 本番では ConsoleMetricExporterOTLPMetricExporter に置き換え、Prometheusエンドポイントに向ける

コスト最適化の実装パターン|モデルルーティング

計測データが集まったら、次は削減です。検証環境での試行を経て効果が確認できた実装パターンを紹介します。

コード例3: モデルルーティングによるコスト最適化

全リクエストを高コストモデルに送るのではなく、質問の複雑度に応じてモデルを切り替えるパターンです。シンプルな質問は安価なモデルで処理することで、API費用を20〜60%削減できます。

動作環境: Python 3.11+, openai>=1.30.0, tiktoken>=0.5.0

import tiktoken
from openai import OpenAI

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

client = OpenAI()

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    """tiktokenでトークン数を事前計算(APIコール前のチェック用)"""
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

def route_model(prompt: str) -> str:
    """
    プロンプトの特性からモデルを動的選択。
    複雑度: トークン数 + キーワード判定による簡易分類。
    """
    token_count = count_tokens(prompt)
    # 複雑なタスクを示すキーワード
    complex_keywords = ["コード", "分析", "比較", "設計", "アーキテクチャ", "デバッグ"]
    is_complex = any(kw in prompt for kw in complex_keywords) or token_count > 500

    if is_complex:
        return "gpt-4o"           # 複雑: 高品質モデル
    else:
        return "gpt-4o-mini"      # シンプル: 低コストモデル

def cost_optimized_call(prompt: str) -> dict:
    """
    モデルルーティング付きLLM呼び出し。
    戻り値: {"answer": str, "model_used": str, "estimated_cost_usd": float}
    """
    model = route_model(prompt)
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    cost = calculate_cost(
        model,
        response.usage.prompt_tokens,
        response.usage.completion_tokens,
    )
    return {
        "answer": response.choices[0].message.content,
        "model_used": model,
        "estimated_cost_usd": cost,
    }

# 使用例
result = cost_optimized_call("おはようございます。天気はどうですか?")
print(f"使用モデル: {result['model_used']}, コスト: ${result['estimated_cost_usd']:.6f}")

事例区分: 自社検証
測定環境: OpenAI API, Python 3.11, 本番相当のプロンプト1,000件
測定期間: 2026年3月〜4月(2週間)
測定方法: ルーティングあり/なしで同じクエリセットを処理し、API費用を比較
結果: ルーティングなし(全件gpt-4o)→ルーティングあり で月次API費用が約42%削減
(ポイント: 単純な挨拶・FAQ応答の約60%が gpt-4o-mini でルーティングされた)


Layer 3: 評価実装|ハルシネーション検出とFaithfulness

RAGシステムでは、取得したドキュメントに基づかない回答(ハルシネーション)を本番で検知する仕組みが必要です。DeepEval の HallucinationMetric は「矛盾したコンテキスト数 / 全コンテキスト数」でスコアを算出し、0に近いほどハルシネーションが少ないことを示します。

コード例4: DeepEval によるハルシネーション検出

動作環境: Python 3.11+, deepeval>=1.0.0, OPENAI_API_KEY 環境変数が必要

from deepeval import evaluate
from deepeval.metrics import HallucinationMetric
from deepeval.test_case import LLMTestCase

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# DeepEvalは内部でLLMを評価器として使用するため、OPENAI_API_KEYが必要です。

def evaluate_rag_response(
    user_input: str,
    retrieved_contexts: list[str],
    llm_output: str,
    threshold: float = 0.5,
) -> dict:
    """
    RAG応答のハルシネーションを評価する。

    threshold: スコアがこの値を超えたらNG
               (HallucinationMetricの場合、スコアが高いほど悪い)
    戻り値: {"hallucination_score": float, "passed": bool, "reason": str}
    """
    test_case = LLMTestCase(
        input=user_input,
        actual_output=llm_output,
        context=retrieved_contexts,
    )

    metric = HallucinationMetric(threshold=threshold, include_reason=True)
    metric.measure(test_case)

    return {
        "hallucination_score": metric.score,
        "passed": metric.is_successful(),
        "reason": metric.reason,
    }

# 使用例: コンテキストと矛盾する回答を検出する
contexts = [
    "Langfuse Python SDK v3 は2025年6月にGAリリースされ、OpenTelemetryネイティブに書き直された。"
]
# 意図的に誤った情報を含む出力
output = "Langfuse Python SDK v3 は2024年1月にリリースされ、独自プロトコルを使用している。"

result = evaluate_rag_response(
    user_input="Langfuse SDKのv3はいつリリースされましたか?",
    retrieved_contexts=contexts,
    llm_output=output,
)
print(f"ハルシネーションスコア: {result['hallucination_score']}")
print(f"評価通過: {result['passed']}")
print(f"理由: {result['reason']}")
# → スコアが高く(1.0)、評価NGとなる(コンテキストと矛盾するため)

RAGAS の Faithfulness スコア(文脈に支持されるクレームの割合、高いほど良い)と組み合わせると、より多角的な評価が可能です。atlan.com の調査によると、RAGAS Faithfulness と DeepEval HallucinationMetric はともに平均精度0.76前後で、RAGシステムのハルシネーション検出に有効とされています。


本番アラート設計|閾値・エスカレーション・疲労対策

可観測性システムは「データが見える」だけでは意味がありません。異常を検知して人間に届けるアラート設計が必要です。LLMアプリのアラートで特に注意が必要なのは、インフラの健全性とLLMの品質健全性は別軸という点です。レイテンシが正常でも、ハルシネーション率が急上昇していることがあります。

コード例5: Prometheus + Python によるアラートルール定義

動作環境: Python 3.11+, prometheus-client>=0.19.0

from prometheus_client import Histogram, Counter, Gauge, start_http_server
import time

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

# Prometheus メトリクス定義
llm_latency = Histogram(
    "llm_request_latency_seconds",
    "LLMリクエストのレイテンシ(秒)",
    ["model", "feature"],
    buckets=[0.5, 1.0, 2.0, 3.0, 5.0, 10.0, float("inf")],
)
llm_errors = Counter(
    "llm_request_errors_total",
    "LLMリクエストのエラー総数",
    ["model", "error_type"],
)
hallucination_gauge = Gauge(
    "llm_hallucination_rate",
    "直近100件のハルシネーション率(0〜1)",
    ["model"],
)

def record_llm_call(model: str, feature: str, latency: float, error: str | None = None):
    """計測値をPrometheusメトリクスに記録する"""
    llm_latency.labels(model=model, feature=feature).observe(latency)
    if error:
        llm_errors.labels(model=model, error_type=error).inc()

# Prometheusアラートルール(YAML形式: /etc/prometheus/rules/ に配置)
ALERT_RULES_YAML = """
groups:
  - name: llm_observability
    rules:
      # P95レイテンシアラート(5分間持続でWARNING)
      - alert: LLMHighLatencyP95
        expr: histogram_quantile(0.95, rate(llm_request_latency_seconds_bucket[5m])) > 3.0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "LLM P95レイテンシが3秒超"
          description: "モデル {{ $labels.model }} の P95 が {{ $value | humanizeDuration }}"

      # エラー率アラート(2分間持続でCRITICAL)
      - alert: LLMHighErrorRate
        expr: >
          rate(llm_request_errors_total[5m])
          / rate(llm_request_latency_seconds_count[5m]) > 0.05
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "LLMエラー率が5%超"

      # ハルシネーション率アラート(10分持続でWARNING)
      - alert: LLMHighHallucinationRate
        expr: llm_hallucination_rate > 0.15
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "ハルシネーション率が15%超"
"""

if __name__ == "__main__":
    start_http_server(8000)
    print("Prometheusメトリクスサーバー起動: http://localhost:8000/metrics")
    while True:
        time.sleep(1)

アラートしきい値の目安(本番運用での知見):

計測項目 WARNINGしきい値 CRITICALしきい値 持続時間
P95レイテンシ > 3秒 > 8秒 5分
エラー率 > 3% > 5% 2分
ハルシネーション率 > 15% > 30% 10分
日次コスト増加 基準の150% 基準の300% 1時間

アラート疲労を防ぐため、瞬間的な値ではなく「N分間持続」を条件にすることが重要です。LLMのレイテンシは瞬間的に高くなることがあり、5分未満の spike でページングすると夜中のアラートが無意味に増えます。


FAQ|よくある疑問と実装時の注意点

Q1: OpenTelemetry と Langfuse はどちらを使えばいいですか?

A: 両方使います。OpenTelemetry は「計装の標準規格」で、Langfuse は「LLM特化のバックエンド」です。OpenTelemetry でトレースを生成し、Langfuse(またはJaeger・Grafana Tempo等)に送信するという構成が標準的です。Langfuse SDK v3 はOTelネイティブのため、既存のOTelパイプラインにLangfuseを差し込むことができます。

Q2: DeepEval での評価はリアルタイムで本番に挟めますか?

A: 部分的に可能ですが、コストと速度に注意が必要です。DeepEval の HallucinationMetric は内部でLLMを評価LLMとして使うため、追加のAPI費用と数秒の処理時間が発生します。全リクエストに挟む場合はコストが2〜3倍になることがあります。実運用では「非同期バックグラウンドで全件評価」と「10%サンプリングのリアルタイム評価」を組み合わせるのが現実的です。

Q3: コスト監視で最も見落とされやすい項目は?

A: プロンプトキャッシュのヒット率です。OpenAI や Anthropic は同一プロンプトプレフィックスのキャッシュヒット時に割引を適用しますが、このキャッシュヒット率をモニタリングしていないチームが多いです。APIレスポンスの usage.prompt_tokens_details.cached_tokens(OpenAI)を記録することで、キャッシュ活用度を可視化できます。

Q4: ハルシネーション率が急上昇したとき最初に確認すること

以下の順番で確認することをすすめています。

  1. RAGの取得結果(ドキュメントの内容・スコア)が変化していないか
  2. プロンプトテンプレートのバージョンが変更されていないか
  3. LLMモデルのバージョンが変わっていないか(APIプロバイダーがサイレントにモデルを更新することがある)
  4. 入力分布の変化(新しいユーザー層が入ってきた等)

トレースがあると「問題が起きたリクエスト」を特定できるため、Step 1のRAG取得結果の確認が数分で終わります。可観測性がない状態では、この特定だけで丸1日かかることがあります。

Q5: まずどこから始めればいいですか?

Layer 1(Langfuse トレース)を1日で導入し、1週間後に Layer 2(コスト・レイテンシメトリクス)を追加し、1ヶ月後に Layer 3(評価・ハルシネーション検出)を追加する、という段階的な進め方が最も成功確率が高いです。最初から全部入れようとすると、構築に数週間かかり本番リリースが遅延します。


【要注意】よくある失敗パターンと回避策

失敗1: ログだけでトレースを代替しようとする

❌「print() や logging でプロンプトとレスポンスを記録すれば十分」
⭕ OpenTelemetryのスパン構造で親子関係を明示する

なぜ重要か: ログは時系列の羅列に過ぎず、「このLLM呼び出しはどのユーザーリクエストから発生したか」「何番目のリトライで成功したか」という因果関係が追えません。特にエージェントのループ(ReActのThought→Action→Observation)をデバッグする際に、トレース構造なしでは問題特定が不可能に近くなります。

失敗2: 評価を開発環境でしか動かさない

❌ テスト時だけ DeepEval を走らせ、本番はスコアを計測しない
⭕ 非同期評価パイプラインで本番データを継続的にサンプリング評価する

なぜ重要か: モデルのドリフト(同じプロンプトでも品質が徐々に変化する現象)は、本番データを継続的に評価しないと気づきません。開発時に高スコアだった評価が、数ヶ月後に大きく下がっていることがあります。

失敗3: コストアラートを月次でしか見ない

❌「月末の請求書で初めてコスト超過に気づく」
⭕「日次の費用増加率が150%を超えたら即時アラート」

なぜ重要か: ループするエージェントや誤ったプロンプト設計は、数時間で予算を食い尽くすことがあります。マルチエージェントシステムの無限ループで1日に予算の数倍のAPI費用が発生した事例は、開発コミュニティで複数報告されています。日次・時間次のアラートが必須です。


参考・出典


まとめ:今日から始める3つのアクション

  1. 今日やること: Langfuse のフリープランに登録し、既存のLLM呼び出し1箇所にコード例1を適用してみる(所要時間: 30分)
  2. 今週中: コスト・レイテンシメトリクス(コード例2)を追加し、機能別の日次コストを可視化する
  3. 今月中: DeepEval による評価を非同期パイプラインに組み込み、ハルシネーション率のベースラインを計測開始する

あわせて読みたい:


この記事を読んで導入イメージが固まってきた方へ

UravationではAIエージェント導入の研修・コンサルを行っています。LLM可観測性の設計・実装から本番運用まで、ご支援が可能です。


著者: 佐藤傑(さとう・すぐる)
株式会社Uravation代表取締役。X(@SuguruKun_ai)フォロワー10万人超。
100社以上の企業向けAI研修・導入支援。著書累計3万部突破。
SoftBank IT連載7回執筆(NewsPicks最大1,125ピックス)。

ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。

Need help moving from reading to rollout?

この記事を読んで導入イメージが固まってきた方へ

Uravationでは、AIエージェントの要件整理、PoC設計、社内導入、研修まで一気通貫で支援しています。

この記事をシェア

X Facebook LINE

※ 本記事の情報は2026年4月時点のものです。サービスの料金・仕様は変更される可能性があります。最新情報は各サービスの公式サイトをご確認ください。

関連記事