AIエージェント入門

Langfuse AIエージェント可観測性ガイド｜ログ・トレース・評価の実装

2026.03.27 更新 25分で読める
Langfuse AIエージェント可観測性ガイド｜ログ・トレース・評価の実装

この記事の結論

LangfuseでAIエージェントのトレース・ログ・評価を実装する方法を解説。LangSmith・Arize・W&Bとの比較、Python SDKコード例5個以上、本番監視ベストプラクティス付き。

「デモでは完璧に動いていたのに、本番に出したら原因不明のエラーが頻発する…」

正直、これはAIエージェント開発で最もよくある落とし穴だ。マルチステップの推論、外部ツール呼び出し、RAG検索――これらが複雑に絡み合うと、従来のログだけでは何が起きているのか把握できなくなる。エラーメッセージすら出ずに、ただ「微妙に間違った回答」が返り続ける。いわゆるサイレント障害というやつだ。

この記事では、オープンソースのLLMオブザーバビリティプラットフォーム「Langfuse」を使って、AIエージェントの挙動を可視化・監視・改善するための具体的な手順を、コピペ可能なコード付きで解説する。セットアップから本番運用のアラート設計まで、今日から始められる内容になっている。なお、OpenAI Agents SDKやLangGraph・CrewAI・AutoGenなど各フレームワークにLangfuseを接続する方法も後述する。

Step 1：Langfuseの環境構築（10分で完了）

最初の一歩は驚くほど簡単だ。Langfuse Cloudを使えば、アカウント作成から最初のトレース送信まで10分もかからない。

1-1. Langfuse Cloudにサインアップ

cloud.langfuse.com でアカウントを作成し、プロジェクトを新規作成する。ダッシュボードの Settings → API Keys から、Public KeyとSecret Keyを取得しよう。

1-2. Python SDKのインストールと初期設定

以下のコマンドでSDKをインストールする。

# 動作環境: Python 3.10+
# 必要パッケージ: langfuse, openai
pip install langfuse openai

環境変数を設定する。APIキーは絶対にコードにハードコードしないこと。

# .envファイルに記述（.gitignoreに追加必須）
LANGFUSE_PUBLIC_KEY="pk-lf-xxxxxxxxxxxx"
LANGFUSE_SECRET_KEY="sk-lf-xxxxxxxxxxxx"
LANGFUSE_HOST="https://cloud.langfuse.com"
OPENAI_API_KEY="sk-xxxxxxxxxxxx"

1-3. 疎通確認コード

まずは接続が成功するか確認してみよう。

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

from langfuse import get_client

langfuse = get_client()

if langfuse.auth_check():
    print("✅ Langfuse接続成功！")
else:
    print("❌ 認証エラー。API Keyとホスト名を確認してください。")

ここまでできたら、次はAIエージェントの処理をトレースしていく。

Step 2：OpenAI呼び出しのトレースを自動化する

Langfuseの便利なところは、import文を1行変えるだけで、OpenAI API呼び出しのトレースが全自動になる点だ。

# 動作環境: Python 3.10+, langfuse>=3.0.0, openai>=1.30.0
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

# ❌ 通常のimport
# from openai import OpenAI

# ⭕ Langfuseラッパー経由のimport（これだけで自動トレース）
from langfuse.openai import openai

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "あなたはカスタマーサポートエージェントです。"},
        {"role": "user", "content": "注文のキャンセル方法を教えてください。"}
    ],
    temperature=0.3,
    # Langfuse固有のメタデータ（任意）
    langfuse_extra={
        "user_id": "customer-789",
        "session_id": "support-session-001",
        "tags": ["customer-support", "order-cancel"]
    }
)

print(response.choices[0].message.content)

このコードを実行すると、Langfuseダッシュボードに以下の情報が自動記録される。

記録される項目 具体的な内容
入力プロンプト system + user メッセージ全文
出力レスポンス assistant メッセージ全文
モデル名 gpt-4o
トークン使用量 入力/出力/合計トークン
コスト USD換算の推定コスト
レイテンシ リクエスト〜レスポンスの所要時間
ユーザーID langfuse_extraで指定した値

Step 3：マルチステップエージェントのトレース設計

実際のAIエージェントは、1回のLLM呼び出しで完結しない。ユーザー質問の分析 → ナレッジ検索 → 回答生成 → 回答の検証、という複数のステップを踏む。Langfuseの@observeデコレータを使えば、この流れを構造化してトレースできる。

AIエージェントのメモリ設計と組み合わせると、どのメモリ参照がどのトレーススパンで発生したかも追跡できて便利だ。

# 動作環境: Python 3.10+, langfuse>=3.0.0, openai>=1.30.0
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

from langfuse.openai import openai
from langfuse.decorators import observe, langfuse_context

@observe()
def classify_intent(user_message: str) -> str:
    """ユーザーの意図を分類する"""
    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "ユーザーの意図を以下から1つ選べ: FAQ, 注文確認, クレーム, その他"},
            {"role": "user", "content": user_message}
        ],
        temperature=0
    )
    intent = response.choices[0].message.content.strip()
    langfuse_context.update_current_observation(
        metadata={"detected_intent": intent}
    )
    return intent

@observe()
def search_knowledge_base(query: str, intent: str) -> str:
    """意図に応じたナレッジベース検索（簡略化）"""
    # 実際にはベクトルDB検索やRAGパイプラインを実装する
    knowledge = {
        "FAQ": "返品は商品到着後14日以内に受け付けています。",
        "注文確認": "注文状況はマイページからご確認いただけます。",
        "クレーム": "ご不便をおかけし申し訳ございません。担当部署にお繋ぎします。",
    }
    return knowledge.get(intent, "該当する情報が見つかりませんでした。")

@observe()
def generate_response(user_message: str, context: str) -> str:
    """検索結果をもとに回答を生成する"""
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"以下のコンテキストに基づいて回答してください:n{context}"},
            {"role": "user", "content": user_message}
        ],
        temperature=0.3
    )
    return response.choices[0].message.content

@observe(name="cs-agent-workflow")
def run_cs_agent(user_message: str) -> str:
    """カスタマーサポートエージェントのメインフロー"""
    # Step 1: 意図分類
    intent = classify_intent(user_message)

    # Step 2: ナレッジ検索
    context = search_knowledge_base(user_message, intent)

    # Step 3: 回答生成
    answer = generate_response(user_message, context)

    # トレースにスコアを付与（後でフィルタリング・分析に活用）
    langfuse_context.score_current_trace(
        name="intent_confidence",
        value=1.0 if intent != "その他" else 0.5,
        comment=f"Detected intent: {intent}"
    )

    return answer

# 実行
result = run_cs_agent("先週注文した商品を返品したいのですが")
print(result)

このコードを実行すると、Langfuseのダッシュボードに「cs-agent-workflow」という親トレースの下に、classify_intent→search_knowledge_base→generate_responseの3つの子スパンがネストされた形で表示される。各ステップの所要時間、トークン消費、入出力が一目でわかる。

Step 4：OpenTelemetryでフレームワーク横断トレースを統合する

2026年現在、AIエージェントの業界標準はOpenTelemetry（OTEL）に収束しつつある。Pydantic AI、smolagents、Strands Agents など主要フレームワークがOTELトレースを標準サポートしており、LangfuseもOTELエンドポイントを提供している。

LangGraph・CrewAI・AutoGenなど複数フレームワークを使い分けているチームには、OTEL統合が特に有効だ。フレームワークごとに異なるSDKを入れなくても、統一フォーマットでLangfuseに送れる。

# 動作環境: Python 3.10+, opentelemetry-sdk>=1.20.0, langfuse>=3.0.0
# 必要パッケージ: pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http langfuse
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

import os
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
import base64

# Langfuse OTELエンドポイントの認証設定
public_key = os.environ["LANGFUSE_PUBLIC_KEY"]
secret_key = os.environ["LANGFUSE_SECRET_KEY"]
auth_token = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()

# TracerProviderの設定
exporter = OTLPSpanExporter(
    endpoint="https://cloud.langfuse.com/api/public/otel/v1/traces",
    headers={"Authorization": f"Basic {auth_token}"}
)

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

# トレーサーの取得
tracer = trace.get_tracer("my-agent")

# カスタムスパンの作成
with tracer.start_as_current_span("tool-execution") as span:
    span.set_attribute("tool.name", "web_search")
    span.set_attribute("tool.input", "最新のLangfuse価格")
    # ツール実行ロジック
    result = "Langfuse Coreプランは月$29〜"
    span.set_attribute("tool.output", result)

print("OTELトレースをLangfuseに送信しました")

OTEL統合のポイントは、BatchSpanProcessorを使う点だ。SimpleSpanProcessorは同期処理でレイテンシが増えるため、本番では必ずバッチ処理を使うこと。

Step 5：コスト監視とアラートの仕組みを作る

AIエージェントの運用で地味に怖いのが、コストの暴走だ。エージェントが自律的にツール呼び出しを繰り返す設計の場合、無限ループに入ると一晩でAPIコストが跳ね上がる。AIエージェントのセキュリティリスクの観点からも、コスト監視はインシデント検知の早期警戒システムとして機能する。

Langfuseのダッシュボードでは、トークン使用量とコストをリアルタイムで確認できる。さらに、APIを使えばプログラマティックにコストデータを取得し、閾値超えでSlack通知を飛ばすことも可能だ。

# 動作環境: Python 3.10+, langfuse>=3.0.0, requests
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

import os
import requests
from datetime import datetime, timedelta

LANGFUSE_HOST = os.environ["LANGFUSE_HOST"]
LANGFUSE_PUBLIC_KEY = os.environ["LANGFUSE_PUBLIC_KEY"]
LANGFUSE_SECRET_KEY = os.environ["LANGFUSE_SECRET_KEY"]
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL", "")

DAILY_COST_THRESHOLD_USD = 50.0  # 1日あたりの上限

def check_daily_cost():
    """Langfuse APIから直近24時間のコストを集計する"""
    today = datetime.utcnow().strftime("%Y-%m-%d")
    url = f"{LANGFUSE_HOST}/api/public/metrics/daily"
    params = {"traceName": None, "fromTimestamp": today}

    resp = requests.get(
        url,
        auth=(LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY),
        params=params,
        timeout=10
    )
    resp.raise_for_status()
    data = resp.json()

    total_cost = sum(d.get("totalCost", 0) for d in data.get("data", []))
    return total_cost

def send_slack_alert(cost: float):
    """コスト超過時にSlack通知"""
    if not SLACK_WEBHOOK_URL:
        print(f"⚠️ コスト超過: ${cost:.2f} (閾値: ${DAILY_COST_THRESHOLD_USD})")
        return
    requests.post(SLACK_WEBHOOK_URL, json={
        "text": f"🚨 AIエージェントのAPI利用コストが閾値を超えましたn"
                f"本日のコスト: ${cost:.2f} / 閾値: ${DAILY_COST_THRESHOLD_USD}n"
                f"Langfuseダッシュボードで詳細を確認してください。"
    })

if __name__ == "__main__":
    cost = check_daily_cost()
    print(f"本日のコスト: ${cost:.2f}")
    if cost > DAILY_COST_THRESHOLD_USD:
        send_slack_alert(cost)

このスクリプトをcronで1時間ごとに実行すれば、コスト暴走を早期に検知できる。

Step 6：評価パイプラインで品質を継続的に改善する

トレースが取れるようになったら、次は「回答品質の評価」を仕組み化する。Langfuseのデータセット機能とLLM-as-a-Judge評価を組み合わせると、CI/CDパイプラインに品質チェックを組み込める。

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

from langfuse import get_client

langfuse = get_client()

# 評価用データセットの作成
dataset = langfuse.create_dataset(
    name="cs-agent-eval-v1",
    description="カスタマーサポートエージェントの品質評価用データセット"
)

# テストケースの追加
test_cases = [
    {
        "input": {"message": "返品したいです"},
        "expected_output": "返品ポリシーの案内を含む回答"
    },
    {
        "input": {"message": "注文番号12345の配送状況を教えて"},
        "expected_output": "注文確認方法の案内を含む回答"
    },
    {
        "input": {"message": "商品が壊れていました。最悪です"},
        "expected_output": "謝罪と担当部署への取次を含む回答"
    }
]

for case in test_cases:
    langfuse.create_dataset_item(
        dataset_name="cs-agent-eval-v1",
        input=case["input"],
        expected_output=case["expected_output"]
    )

print(f"✅ データセット作成完了: {len(test_cases)}件のテストケース")

データセットを作ったら、run_cs_agentを各テストケースで実行し、結果をLangfuseに記録することで、プロンプト変更前後の品質比較が可能になる。RAGパイプラインを使っている場合は、検索クオリティの評価もこの仕組みで統一できる。

主要4ツール比較：Langfuse vs LangSmith vs Arize vs W&B Weave

オブザーバビリティツールの選定で必ず比較されるのが、LangChain公式のLangSmith、MLOps老舗のWeights & Biases（W&B Weave）、エンタープライズ向けのArize Phoenixだ。2026年現在の比較をまとめた。

比較項目 Langfuse LangSmith Arize Phoenix W&B Weave
ライセンス MIT（OSS） プロプライエタリ OSS+エンタープライズ プロプライエタリ
セルフホスト ⭕ 無料・全機能 △ 要エンタープライズ ⭕ 無料 ❌ クラウドのみ
フレームワーク最適化 フレームワーク非依存 LangChain/LangGraph OpenTelemetry準拠 既存W&Bユーザー向け
無料枠 50,000ユニット/月 5,000トレース/月 無制限（セルフホスト） 月100GBまで
有料プラン開始価格 $29/月（Core） $39/seat/月（Plus） 要問合せ $50/seat/月〜
マルチステップトレース ⭕ エージェントグラフ対応 ⭕ LangGraphと深連携 △ 補完ロギングが必要 ⭕ 基本対応
プロンプト管理 ⭕ UIバージョン管理 ⭕ ハブ連携 △ 限定的 △ 実験管理中心
OTEL対応 ⭕ ネイティブ対応 △ 部分的 ⭕ OTEL中心設計 ❌ 独自SDK

料金情報の最終確認: 2026-03-27

用途別おすすめの選び方

  • LangChain/LangGraphを使っている → LangSmith（インテグレーションがほぼゼロ設定）
  • データを社内に置きたい・コンプライアンス要件がある → Langfuse（セルフホスト）またはArize Phoenix
  • 既存のW&Bインフラがある → W&B Weave（MLとLLMを統合管理）
  • フレームワーク非依存・OSS優先・コスト重視 → Langfuse
  • OpenTelemetryを既に使っている → Arize Phoenix（OTEL中心設計）

正直にお伝えすると、どのツールも2026年現在では急速に機能追加が進んでいる。競合比較表は半年で陳腐化するため、実際の選定では必ず最新の公式ドキュメントを確認してほしい。

【要注意】オブザーバビリティ導入でよくある失敗パターン

失敗1：全てのリクエストをフルトレースして料金爆発

❌ 開発時の設定のまま本番に持ち込み、全リクエストの全スパンを記録

⭕ 本番ではサンプリングレートを設定し、10〜30%のリクエストのみトレース。異常検知時だけフル記録に切り替える

なぜ重要か：Langfuseの料金はユニット数（トレース+観測+スコア）に比例する。月間100万リクエストのサービスで全トレースすると、Coreプラン（月$29）の無料枠10万ユニットを初日で使い切る計算になる。

失敗2：トレースIDとユーザーセッションを紐づけない

❌ デフォルト設定のまま使い、「このエラーはどのユーザーの操作で発生したのか」が追えない

⭕ user_idとsession_idを必ず付与する。ユーザー問い合わせ時にセッション単位でトレースを検索できるようにする

なぜ重要か：ユーザーから「昨日の夕方、AIの回答がおかしかった」と問い合わせがあった時、セッションIDがなければ数万件のトレースから該当箇所を探す羽目になる。

失敗3：評価指標を決めずに「とりあえずトレース」で終わる

❌ トレースデータだけが溜まり、誰も見ない、何も改善しないダッシュボードの完成

⭕ 導入時に「何を計測し、どの数値が悪化したらアクションを取るか」を3つ以内で決める

なぜ重要か：オブザーバビリティツールは目的ではなく手段。KPIの例として、タスク完了率、ハルシネーション率、平均レイテンシの3つから始めるのが現実的だ。

失敗4：プロンプトのバージョン管理をコード側だけでやる

❌ プロンプトをPythonファイル内にハードコードし、gitのdiffでしか変更履歴を追えない

⭕ Langfuseのプロンプト管理機能を使い、バージョン管理・A/Bテスト・ロールバックをUI上で完結させる

なぜ重要か：プロンプトの変更が回答品質に与える影響は、コードの変更より予測が難しい。Langfuseならプロンプトバージョンごとのスコア比較がダッシュボードで即座に確認できる。

失敗5：セキュリティを後回しにする

❌ トレースデータに個人情報・APIキーが混入したまま記録される

⭕ 本番投入前に、トレースデータのマスキング設定とアクセス制御（プロジェクトごとのAPIキー分割）を必ず実施する

なぜ重要か：Langfuseには入出力プロンプトが丸ごと記録される。AIエージェントのIDとセキュリティ管理の観点から、観測データの扱いは慎重に設計すること。

本番環境でのエージェント監視ベストプラクティス5選

数十社のAIエージェント導入支援を通じて得た、本番環境での実践的なベストプラクティスをまとめる。

1. アラート設計は「3段階」で構成する

レベル 条件例 通知先 対応時間
Warning エラー率 >5% / レイテンシ >5s Slack #dev 翌営業日
Critical エラー率 >20% / コスト閾値超過 Slack #ops + PagerDuty 1時間以内
Emergency エージェント無限ループ検知 SMS + 電話 即時

2. 本番リリース前に「ゴールデンセット」評価を通す

最低30件の代表的なユーザーリクエストを「ゴールデンセット」として定義し、プロンプト変更・モデル変更のたびにLangfuseのデータセット評価を実行する。これでリグレッションを事前に検知できる。

3. トレースIDをアプリケーションログと紐づける

Langfuseのトレースに、アプリケーション側のリクエストID（例: Nginxのrequest_id、AWS X-RayのTraceID）をmetadataとして付与しておく。障害発生時に複数のログシステムを横断して調査できる。

4. 「サイレント障害」専用の評価器を設ける

エラーは出ないが回答が間違っている「サイレント障害」は、Langfuseの評価スコアで定期的に検知する。LLM-as-a-Judge評価器を設定し、スコアが一定以下のトレースを自動でレビューキューに入れる仕組みが効果的だ。

5. 定期的な「トレース棚卸し」をスケジュールする

月1回、過去30日間のトレースデータを分析し、「よく失敗するリクエストパターン」と「コストが異常に高いリクエスト」を特定する。これを改善サイクルのインプットにすると、エージェントの品質が継続的に向上する。

セルフホストで始める場合のDocker Compose構成

データを外部に出したくないチーム向けに、セルフホストの構成も紹介しておく。Langfuse v3はDocker Composeで比較的簡単にデプロイできる。

# docker-compose.yml（Langfuse v3セルフホスト）
# 動作環境: Docker 24+, Docker Compose v2+
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

version: "3.9"
services:
  langfuse-web:
    image: langfuse/langfuse:latest
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/langfuse
      - CLICKHOUSE_URL=http://clickhouse:8123
      - REDIS_HOST=redis
      - NEXTAUTH_SECRET=your-secret-here  # 必ず変更すること
      - SALT=your-salt-here               # 必ず変更すること
      - NEXTAUTH_URL=http://localhost:3000
    depends_on:
      - db
      - clickhouse
      - redis

  langfuse-worker:
    image: langfuse/langfuse-worker:latest
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/langfuse
      - CLICKHOUSE_URL=http://clickhouse:8123
      - REDIS_HOST=redis
    depends_on:
      - db
      - clickhouse
      - redis

  db:
    image: postgres:16
    environment:
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=langfuse
    volumes:
      - pgdata:/var/lib/postgresql/data

  clickhouse:
    image: clickhouse/clickhouse-server:latest
    volumes:
      - chdata:/var/lib/clickhouse

  redis:
    image: redis:7-alpine
    volumes:
      - redisdata:/data

volumes:
  pgdata:
  chdata:
  redisdata:

docker compose up -d で起動後、http://localhost:3000 にアクセスすれば、セルフホスト版のLangfuseが使える。本番環境では、最低2CPU・4GBメモリ、Web containerは2インスタンス以上での冗長化が推奨されている（公式ドキュメント参照）。

参考・出典

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

  1. 今日やること：Langfuse Cloudの無料アカウントを作り、Step 1〜2のコードを手元で動かす。最初のトレースがダッシュボードに表示されるのを確認しよう
  2. 今週中：既存のAIエージェントに@observeデコレータを追加し、マルチステップのトレースを取得する。ユーザーID・セッションIDの付与も忘れずに
  3. 今月中：コスト監視スクリプトを本番に導入し、評価用データセットを30件作成する。ゴールデンセット評価をCI/CDパイプラインに組み込もう

あわせて読みたい:

この記事はAIgent Lab編集部がお届けしました。

Need help moving from reading to rollout?

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

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

30分無料相談を予約 支援実績を見る

この記事をシェア

X Facebook LINE

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

関連記事

AIエージェント最小権限設計｜AWS/GCP/Azure IAM実装ガイド AIエージェント入門

AIエージェント最小権限設計｜AWS/GCP/Azure IAM実装ガイド

AIエージェントへの過剰権限付与はセキュリティリスクの根本原因です。AWS IA...

13分で読める
Claude Code Agent Teams Q&A｜詰まりポイント完全解消 AIエージェント入門

Claude Code Agent Teams Q&A｜詰まりポイント完全解消

CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1の使...

8分で読める
AIエージェント入門

AIエージェントの認証設計入門：APIキー・OAuth・Vaultの使い分け

AIエージェントが外部APIを安全に呼び出すための認証設計を解説。環境変数→JW...

12分で読める