0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

Langfuse・Phoenix・LangSmithで実装するLLMトレースベース評価と本番可観測性

0
Last updated at Posted at 2026-04-26

Langfuse・Phoenix・LangSmithで実装するLLMトレースベース評価と本番可観測性

この記事でわかること

  • LLM可観測性の5層アーキテクチャ(Reliability・Quality・Safety・Cost・Governance)と設計指針
  • Langfuse・Arize Phoenix・LangSmithの3大プラットフォームの技術的差異と選定基準
  • 本番トレースに対するLLM-as-a-Judge自動評価の実装パターン(Python コード付き)
  • トレースベース評価をCI/CDとアラートに組み込む運用設計
  • 3プラットフォームの料金体系・ライセンス・デプロイモデルの定量比較

対象読者

  • 想定読者: LLMアプリケーションを本番運用しているMLエンジニア・SRE
  • 必要な前提知識:
    • Python 3.11+の基礎文法
    • OpenAI APIまたは同等のLLM APIの利用経験
    • 分散トレーシングの基本概念(span、trace IDなど。OpenTelemetryの経験があるとなお良い)

結論・成果

3プラットフォームを比較検証した結果、以下の知見が得られました。

  • Langfuse: フレームワーク非依存で80以上のインテグレーションを持ち、MIT ライセンスによるセルフホスティングが可能。無料枠は月50,000ユニット。2,300社以上が本番採用し、Fortune 50企業のうち19社が利用していると公式ページで報告されています
  • Arize Phoenix: OpenTelemetry/OpenInferenceネイティブで、既存のAPM基盤との統合に適している。GitHubスター7,800以上、月間250万ダウンロードと公式リポジトリで報告されています。完全OSSでセルフホスト可能
  • LangSmith: LangChain/LangGraphとのゼロ設定統合が特徴。環境変数2つの設定で全チェーンコールが自動トレースされ、LangChainエコシステムに深く組み込まれたチームに適している

トレースベース評価を導入することで、「出力を記録するだけの高コストなログ」から「品質劣化を自動検知し改善サイクルを回す可観測性基盤」への転換が可能になります。

LLM可観測性の5層アーキテクチャを理解する

LLMアプリケーションの可観測性は、従来のWebアプリケーション監視とは本質的に異なります。従来の監視が「システムは稼働しているか?」を問うのに対し、LLM可観測性は「モデルの出力は正確で、安全で、コスト効率が良いか?」を問います。

Portkey社のガイドによると、LLM可観測性は以下の5層で構成されます。

5層モデルの全体像

目的 監視対象の例 従来のAPMとの対比
Reliability レイテンシ・レート制限・プロバイダ障害の検知 P95レイテンシ、エラー率、リトライ回数 SLI/SLOに近い
Quality 事実正確性・グラウンディングの測定 Hallucination率、Relevance score テストカバレッジに近い
Safety ジェイルブレイク・有害出力・PII漏洩の検知 Toxicityスコア、PII検出回数 WAFに近い
Cost トークン使用量・リトライ・予算の追跡 月額トークンコスト、リクエスト単価 クラウド課金監視に近い
Governance 全リクエストのトレーサビリティ・監査対応 トレースID、モデルバージョン、プロンプトバージョン 監査ログに近い

MLエンジニアにとって馴染みのある例えで言えば、ReliabilityはMLモデルのサービングSLO(推論レイテンシP99 < 200msなど)、QualityはオフラインのEvalメトリクス(Accuracy, F1など)をオンラインで継続監視するイメージです。

トレースが可観測性の中核になる理由

メトリクス単体では「P95レイテンシが500msを超えた」という事実しかわかりません。しかしトレースがあれば、「Embedding検索に300ms、LLMコールに150ms、後処理に50ms」というブレイクダウンが可能です。

LLM可観測性において、1つのトレースは以下の情報を階層的に保持します。

各Spanには入力・出力・レイテンシ・トークン数・コストが記録され、trace_idで一貫して紐づけられます。このtrace_idが5層すべてを横断するキーとなります。

注意点:

トレーシングなき評価は「高コストなログ」に過ぎないと指摘されています(Portkey社のガイド)。記録だけでなく、記録に対して評価を実行し、その結果をフィードバックループに組み込むことが重要です。

3大プラットフォームの技術比較と選定基準を把握する

Langfuse・Arize Phoenix・LangSmithはいずれもLLM可観測性プラットフォームですが、アーキテクチャ思想・ライセンス・得意領域が大きく異なります。ここでは定量データに基づいて比較します。

アーキテクチャと技術スタックの比較

項目 Langfuse Arize Phoenix LangSmith
ライセンス MIT(完全OSS) Elastic License 2.0 プロプライエタリ
GitHubスター 19,000+ 7,800+ N/A(クローズドソース)
バックエンドDB ClickHouse PostgreSQL(セルフホスト) ClickHouse
標準規格 OpenTelemetryネイティブ OpenTelemetry + OpenInference OpenTelemetry対応
SDKサポート Python, TypeScript, Java Python, TypeScript, Java Python, TypeScript
フレームワーク連携 80以上(フレームワーク非依存) LangChain, LlamaIndex, DSPy他 LangChain/LangGraph特化
セルフホスト フル機能対応(エアギャップ可) Docker/K8s対応 Enterprise契約のみ
非同期トレーシング 対応(アプリにレイテンシ追加なし) 対応 対応

出典: Langfuse公式Firecrawl社比較記事

料金体系の比較

プラン Langfuse Arize Phoenix LangSmith
無料枠 50,000ユニット/月 OSS完全無料(Cloud: 25kスパン/月) 5,000トレース/月
有料プラン Core: $29/月、Pro: $199/月 Cloud: スパン+データ量ベース Plus: $39/ユーザー/月
課金単位 ユニット(Trace, Observation, Scoreそれぞれ1ユニット) スパン + データ取り込み量(GB) ルートトレース(内部ステップは無料)
Enterprise カスタム カスタム カスタム(セルフホスト含む)

出典: Langfuse公式Langfuse vs Arize

課金モデルの落とし穴:

Langfuseはトレース・オブザベーション・スコアそれぞれが1ユニットとしてカウントされます。つまりRAGパイプラインのように1リクエストで5-10個のSpanが発生する場合、1トレースあたり6-11ユニットが消費されます。一方LangSmithはルートトレース単位の課金なので、内部ステップ数に関わらず1リクエスト=1トレースです。ユースケースによってコスト構造が大きく変わるため、無料枠の数字だけで判断しないことが重要です。

選定フローチャート

なぜこの選定基準か:

  • LangSmith: LangChainエコシステムに深く組み込まれている場合、環境変数2つで全チェーンが自動計装されるのは大きなアドバンテージです。ただしLangChain外のフレームワークでは価値が大幅に低下します
  • Phoenix: 既にPrometheus/Grafana/Jaegerなどの既存OTel基盤を持つチームには、OTLP互換のPhoenixが自然に統合できます
  • Langfuse: 上記のどちらにも当てはまらない多くのケースで、フレームワーク非依存・MIT OSS・充実した無料枠という組み合わせが適しています

3プラットフォームでトレースベース評価を実装する

ここでは各プラットフォームで「RAGアプリケーションの本番トレースに対してLLM-as-a-Judge評価を実行する」パターンを実装します。

Langfuseでトレース収集と評価を実装する

Langfuseは@observe()デコレータとOpenAI SDKのドロップインラッパーで、最小限のコード変更でトレースを収集できます。

# langfuse_rag_example.py
# 動作確認: Python 3.11+, langfuse>=3.0, openai>=1.0

from langfuse.openai import openai
from langfuse import observe, get_client

langfuse = get_client()

@observe()
def retrieve_context(query: str) -> list[str]:
    """ベクトルDBからコンテキストを検索(実装は省略、ダミーデータ使用)"""
    return [
        "Langfuseはオープンソースの LLM 可観測性プラットフォームです。",
        "MIT ライセンスで、セルフホスティングに完全対応しています。",
    ]

@observe()
def generate_answer(query: str, contexts: list[str]) -> str:
    """検索コンテキストを用いてLLMで回答を生成"""
    context_text = "\n".join(contexts)
    response = openai.chat.completions.create(
        name="rag-generation",
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": f"以下のコンテキストに基づいて質問に回答してください。\n\nコンテキスト:\n{context_text}",
            },
            {"role": "user", "content": query},
        ],
        temperature=0.1,
    )
    return response.choices[0].message.content

@observe()
def rag_pipeline(query: str) -> str:
    """RAGパイプライン全体をトレースとして記録"""
    contexts = retrieve_context(query)
    answer = generate_answer(query, contexts)
    return answer

result = rag_pipeline("Langfuseのライセンスは?")
langfuse.flush()

@observe()デコレータは関数呼び出しを自動的にSpanとして記録します。MLエンジニアにとっては、MLflowの@mlflow.traceデコレータと同様のアプローチです。openaiのインポート元をlangfuse.openaiに変えるだけで、プロンプト・レスポンス・トークン数・コストが自動的にGenerationとして記録されます。

次に、収集したトレースに対してLLM-as-a-Judge評価を実行します。

# langfuse_evaluation.py
# トレースに対するLLM-as-a-Judge評価の実装例

from langfuse import get_client

langfuse = get_client()

def evaluate_trace_faithfulness(trace_id: str) -> None:
    """トレースの忠実性(Faithfulness)をLLM-as-a-Judgeで評価"""
    trace = langfuse.get_trace(trace_id)

    evaluation_prompt = f"""以下のコンテキストと回答を評価してください。
回答がコンテキストの情報のみに基づいているかを判定します。

コンテキスト: {trace.input}
回答: {trace.output}

評価基準:
- faithful: 回答がコンテキストの情報のみに基づいている
- hallucinated: コンテキストにない情報が含まれている

判定結果を "faithful" または "hallucinated" で回答してください。"""

    from openai import OpenAI
    client = OpenAI()
    judge_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": evaluation_prompt}],
        temperature=0,
    )
    label = judge_response.choices[0].message.content.strip().lower()

    langfuse.score(
        trace_id=trace_id,
        name="faithfulness",
        value=1.0 if label == "faithful" else 0.0,
        comment=f"LLM-as-a-Judge result: {label}",
    )

langfuse.score()でトレースに評価スコアを紐づけることで、ダッシュボード上でスコアの時系列推移やドリフトを監視できます。

Arize Phoenixでトレース収集と評価を実装する

Phoenixはopentelemetryベースの計装とビルトイン評価器で、トレース収集から評価までを一貫して処理します。

# phoenix_rag_example.py
# 動作確認: Python 3.11+, arize-phoenix>=8.0, openinference-instrumentation-openai>=1.0

import phoenix as px
from openinference.instrumentation.openai import OpenAIInstrumentor
from opentelemetry import trace as otel_trace
from openai import OpenAI

# Phoenixサーバーを起動(ローカル開発用)
px.launch_app()

# OpenAI SDKの自動計装を有効化
OpenAIInstrumentor().instrument()

tracer = otel_trace.get_tracer(__name__)
client = OpenAI()

def retrieve_context(query: str) -> list[str]:
    """ベクトルDBからコンテキストを検索(ダミーデータ)"""
    return [
        "Phoenixは Arize AI が開発したOSS可観測性ツールです。",
        "OpenTelemetryとOpenInferenceに基づいて構築されています。",
    ]

def rag_pipeline(query: str) -> str:
    """RAGパイプラインをOTelスパンで計装"""
    with tracer.start_as_current_span("rag-pipeline") as span:
        span.set_attribute("input.value", query)

        with tracer.start_as_current_span("retrieval"):
            contexts = retrieve_context(query)

        context_text = "\n".join(contexts)
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": f"コンテキストに基づいて回答:\n{context_text}",
                },
                {"role": "user", "content": query},
            ],
        )
        answer = response.choices[0].message.content
        span.set_attribute("output.value", answer)
        return answer

result = rag_pipeline("Phoenixの技術基盤は?")

PhoenixのトレーシングはOTelのtracer.start_as_current_span()を直接利用します。MLエンジニアにとっては、torch.profilerrecord_functionコンテキストマネージャと同様の使い方です。OpenAIInstrumentor().instrument()により、OpenAI APIコールは自動的にスパンとして記録されます。

Phoenixのビルトイン評価器でトレースを評価します。

# phoenix_evaluation.py
# Phoenixのトレースに対するLLM-as-a-Judge評価

from phoenix.evals import (
    OpenAIModel,
    llm_classify,
    RAG_RELEVANCY_PROMPT_TEMPLATE,
)
import phoenix as px

# Phoenixからトレースデータをデータフレームとして取得
phoenix_client = px.Client()
spans_df = phoenix_client.get_spans_dataframe()

# 評価用LLMモデルの設定
eval_model = OpenAIModel(model="gpt-4o-mini")

# RAG Relevance評価の実行
relevance_results = llm_classify(
    dataframe=spans_df,
    model=eval_model,
    template=RAG_RELEVANCY_PROMPT_TEMPLATE,
    rails=["relevant", "unrelated"],
    concurrency=10,
    provide_explanation=True,
)

llm_classifyはDataFrameのカラムをテンプレートに自動マッピングし、並行でLLM評価を実行します。provide_explanation=Trueにすることで、単なるラベルだけでなく判定理由も取得できます。

LangSmithでトレース収集と評価を実装する

LangSmithはLangChain/LangGraphを使っている場合、環境変数の設定だけで自動トレーシングが有効になります。

# langsmith_rag_example.py
# 動作確認: Python 3.11+, langchain>=0.3, langsmith>=0.2

# 環境変数の設定(これだけで全チェーンが自動トレース)
# export LANGCHAIN_TRACING_V2=true
# export LANGCHAIN_API_KEY=ls-...

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

prompt = ChatPromptTemplate.from_template(
    "以下のコンテキストに基づいて質問に回答してください。\n\n"
    "コンテキスト: {context}\n\n"
    "質問: {question}"
)

def format_docs(docs: list[str]) -> str:
    return "\n".join(docs)

# LCELチェーンの構築(自動的にトレースされる)
chain = (
    {
        "context": lambda x: format_docs(
            ["LangSmithはLangChainの可観測性プラットフォームです。"]
        ),
        "question": RunnablePassthrough(),
    }
    | prompt
    | llm
    | StrOutputParser()
)

result = chain.invoke("LangSmithとは?")

LangSmithの最大の特徴は、LangChainを使っていればコード変更なしでトレースが有効になる点です。環境変数LANGCHAIN_TRACING_V2=trueLANGCHAIN_API_KEYを設定するだけで、全チェーンの入出力・レイテンシ・トークン数が自動記録されます。

LangSmithの評価は、データセットに対してオフラインとオンラインの両方で実行できます。

# langsmith_evaluation.py
# LangSmithのオンライン評価(本番トレースに対する自動評価)

from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()

def faithfulness_evaluator(run, example) -> dict:
    """LLMの出力がコンテキストに忠実かを評価するカスタム評価器"""
    from openai import OpenAI
    openai_client = OpenAI()

    prediction = run.outputs.get("output", "")
    context = run.inputs.get("context", "")

    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "user",
                "content": (
                    f"回答がコンテキストに忠実か判定してください。\n"
                    f"コンテキスト: {context}\n回答: {prediction}\n"
                    f"'faithful' または 'hallucinated' で回答:"
                ),
            }
        ],
        temperature=0,
    )
    label = response.choices[0].message.content.strip().lower()
    return {"key": "faithfulness", "score": 1.0 if label == "faithful" else 0.0}

# データセットに対する評価実行
results = evaluate(
    lambda inputs: chain.invoke(inputs["question"]),
    data="my-rag-dataset",
    evaluators=[faithfulness_evaluator],
    experiment_prefix="rag-faithfulness-v1",
)

LangSmithのevaluate()は、4種類の評価器(LLM-as-Judge、ヒューリスティック、人手アノテーション、ペアワイズ比較)を統一的なインターフェースで実行できます。experiment_prefixにより、プロンプトやモデルのバージョン変更による品質変化を追跡できます。

実装パターンの比較

観点 Langfuse Phoenix LangSmith
計装方法 @observe()デコレータ + OpenAIラッパー OTelスパン + Instrumentor 環境変数のみ(LangChain)
評価API langfuse.score() llm_classify() + DataFrame evaluate() + カスタム評価器
評価のトリガー SDK呼び出し or バッチ DataFrame操作 データセット or 本番トレース
ML類似パターン MLflow @mlflow.trace torch.profiler W&B Sweeps
コード変更量 小(インポート変更+デコレータ) 中(OTelスパン手動記述) 最小(LangChain利用時)

よくある間違い:
最初は「とりあえず全トレースを評価すれば良い」と考えがちですが、LLM-as-a-Judge評価はトレースあたり追加のLLM APIコールが発生します。本番トラフィックが1日10,000リクエストある場合、全トレース評価ではgpt-4o-miniでも1日あたり約$5-15のコストが追加されます。本番ではサンプリング評価(全トレースの5-10%を評価)が推奨されます。

本番運用の可観測性パイプラインを設計する

開発環境での動作確認とは別に、本番環境では継続的な品質監視とフィードバックループの構築が必要です。

本番可観測性パイプラインのアーキテクチャ

サンプリング評価の実装例

# production_eval_pipeline.py
# 本番トレースのサンプリング評価バッチ(日次実行想定)

import random
from langfuse import get_client

langfuse = get_client()

SAMPLE_RATE = 0.05  # 5%サンプリング

def run_daily_evaluation():
    """日次バッチ: 前日のトレースからサンプリングして評価"""
    traces = langfuse.fetch_traces(
        limit=1000,
        order_by="timestamp",
        order="DESC",
    )

    sampled = [t for t in traces.data if random.random() < SAMPLE_RATE]

    results = {"faithful": 0, "hallucinated": 0, "total": len(sampled)}

    for trace in sampled:
        score = evaluate_single_trace(trace)
        results[score] += 1

        langfuse.score(
            trace_id=trace.id,
            name="daily-faithfulness",
            value=1.0 if score == "faithful" else 0.0,
        )

    faithfulness_rate = results["faithful"] / max(results["total"], 1)

    if faithfulness_rate < 0.85:
        send_alert(
            f"Faithfulness rate dropped to {faithfulness_rate:.1%} "
            f"(threshold: 85%, evaluated: {results['total']} traces)"
        )

    return results

評価メトリクスの設計

本番で継続監視するメトリクスは、目的に応じて3カテゴリに分類できます。

カテゴリ メトリクス 閾値の例 アラート条件
品質 Faithfulness率 > 85% 1時間平均が85%未満
品質 Relevance率 > 80% 1時間平均が80%未満
コスト リクエスト単価 < $0.05 移動平均が閾値の1.5倍超過
コスト 日次トークン消費 < 1M tokens 日次消費が予算の80%超過
安全性 Toxicityフラグ率 < 1% 1件でも即時アラート
安全性 PII漏洩検出 0件 1件でも即時アラート
信頼性 P95レイテンシ < 3s 5分間P95が閾値超過
信頼性 エラー率 < 1% 5分間エラー率が閾値超過

制約条件:

LLM-as-a-Judge評価自体がLLM APIに依存するため、評価用LLMプロバイダの障害時には評価パイプラインが停止します。評価用モデルには本番推論とは異なるプロバイダ(本番がOpenAIならAnthropic、または逆)を使うか、コードベースのヒューリスティック評価(正規表現チェック、出力長チェックなど)をフォールバックとして併用することを推奨します。

CI/CDへの組み込み

プロンプトやモデルの変更をデプロイする前に、オフライン評価をCIで実行するパターンです。

# .github/workflows/llm-eval.yml
# プロンプト変更時の自動評価パイプライン

name: LLM Evaluation
on:
  pull_request:
    paths:
      - "prompts/**"
      - "config/model_config.yaml"

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: pip install langfuse openai

      - name: Run evaluation suite
        env:
          LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SECRET_KEY }}
          LANGFUSE_PUBLIC_KEY: ${{ secrets.LANGFUSE_PUBLIC_KEY }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python scripts/run_eval_suite.py

      - name: Check quality gate
        run: |
          python -c "
          import json
          with open('eval_results.json') as f:
              results = json.load(f)
          assert results['faithfulness'] >= 0.85, f'Faithfulness {results[\"faithfulness\"]:.1%} < 85%'
          assert results['relevance'] >= 0.80, f'Relevance {results[\"relevance\"]:.1%} < 80%'
          print('Quality gate passed')
          "

このパイプラインにより、プロンプト変更がPull Request時に自動評価され、品質基準を満たさない場合はマージがブロックされます。MLエンジニアにとっては、モデルの精度閾値をCIで検証するのと同じパターンです。

よくある問題と解決方法

問題 原因 解決方法
トレースが表示されない SDKのflush()未呼び出し Lambda/サーバーレス環境ではlangfuse.flush()を明示的に呼ぶ。長時間プロセスではatexitで登録
評価スコアがトレースに紐づかない trace_idの不一致 @observe()のreturn値でtrace_idを取得し、評価時に同じIDを使用
トークンコストが実際と乖離 モデル名の不一致 modelパラメータに正確なモデル名(gpt-4o-2024-11-20など)を指定
PhoenixのUI起動に失敗 ポート競合 px.launch_app(port=6007)で別ポートを指定
LangSmithでLangChain外のコールが見えない 自動計装の範囲外 @traceableデコレータを手動で追加、またはOpenTelemetryブリッジを使用
評価コストが想定以上 全トレース評価 サンプリング率を5-10%に設定。安全性チェックのみ全件実行

まとめと次のステップ

まとめ:

  • LLM可観測性は5層(Reliability・Quality・Safety・Cost・Governance)で設計し、トレースがそれら全層を横断する中核データ構造となる
  • Langfuseはフレームワーク非依存・MIT OSS・セルフホスト対応で多くのチームに適する。Phoenixは既存OTel基盤との統合に適する。LangSmithはLangChainエコシステムに深く組み込まれたチームに適する
  • トレースベース評価では全件評価ではなくサンプリング(5-10%)を基本とし、安全性チェックのみ全件実行する
  • CI/CDにオフライン評価を組み込むことで、プロンプト・モデル変更による品質劣化をデプロイ前に検知できる

次にやるべきこと:

  1. 小さく始める: まずLangfuse(無料枠50,000ユニット/月)で既存のLLMアプリに@observe()デコレータを追加し、トレースの可視化から始める
  2. 評価を追加する: 1-2週間のトレースが蓄積したら、Faithfulness評価のバッチジョブを日次で実行し、品質ベースラインを確立する
  3. アラートとCI/CDを統合する: ベースラインが安定したら、閾値ベースのアラートとPR時の自動評価を導入する

参考


注意: この記事はAI(Claude Code)により自動生成されました。内容の正確性については複数の情報源で検証していますが、実際の利用時は公式ドキュメントもご確認ください。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?