Langfuseで始めるAIエージェント運用監視ガイド

2026.03.09 公開 2026.03.13 更新 19分で読める

この記事の結論

AIエージェントの本番運用で起きるサイレント障害をLangfuseで可視化。セットアップからコスト監視・品質評価まで、コピペ可能なPythonコード付きで解説。

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

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

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

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デコレータを使えば、この流れを構造化してトレースできる。

# 動作環境: 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：コスト監視とアラートの仕組みを作る

AIエージェントの運用で地味に怖いのが、コストの暴走だ。エージェントが自律的にツール呼び出しを繰り返す設計の場合、無限ループに入ると一晩でAPIコストが跳ね上がる。

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 5：評価パイプラインで品質を継続的に改善する

トレースが取れるようになったら、次は「回答品質の評価」を仕組み化する。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に記録することで、プロンプト変更前後の品質比較が可能になる。

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

失敗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ならプロンプトバージョンごとのスコア比較がダッシュボードで即座に確認できる。

LangSmith vs Langfuse — どちらを選ぶべきか

オブザーバビリティツールの選定で必ず比較されるのが、LangChain公式のLangSmithだ。正直、どちらも優秀。選ぶ基準はシンプルで、技術スタックと運用方針で決まる。

比較項目	Langfuse	LangSmith
ライセンス	MIT（オープンソース）	プロプライエタリ（クローズド）
セルフホスト	⭕ フル機能でセルフホスト可能	△ エンタープライズライセンス必要
フレームワーク	フレームワーク非依存	LangChain/LangGraph最適化
無料枠	50,000ユニット/月	5,000トレース/月
有料プラン	$29/月〜（Coreプラン）	$39/seat/月〜（Plusプラン）
データ保持	30日（Hobby）〜3年（Pro）	14日（Free）〜400日（Enterprise）
ユーザー数制限	有料プランは無制限	プランごとに上限あり

筆者のおすすめ：LangChain/LangGraphで全スタックを統一しているチームはLangSmith一択。それ以外（OpenAI SDK直接、独自オーケストレーション、複数フレームワーク併用）ならLangfuseの方が柔軟性が高い。データ主権やコンプライアンスが厳しい環境ではLangfuseのセルフホスト一択になる。

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

セルフホストで始める場合の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インスタンス以上での冗長化が推奨されている。

参考・出典

Langfuse公式ドキュメント — トークン・コストトラッキング — Langfuse（参照日: 2026-03-09）
Langfuse Self-Hosting Guide — Langfuse（参照日: 2026-03-09）
Langfuse Pricing — Langfuse（参照日: 2026-03-09）
Langfuse OpenAI Python Integration — Langfuse（参照日: 2026-03-09）
Best AI Observability Tools 2026 — Braintrust（参照日: 2026-03-09）

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

今日やること：Langfuse Cloudの無料アカウントを作り、Step 1〜2のコードを手元で動かす。最初のトレースがダッシュボードに表示されるのを確認しよう
今週中：既存のAIエージェントに@observeデコレータを追加し、マルチステップのトレースを取得する。ユーザーID・セッションIDの付与も忘れずに
今月中：コスト監視スクリプトを本番に導入し、評価用データセットを10件以上作成する。プロンプト変更時の品質比較を始めよう

あわせて読みたい:

LangGraph・CrewAI・AutoGen徹底比較 — フレームワーク選定後のオブザーバビリティ設計にも活きる
AIエージェントのセキュリティリスクとOWASP対策ガイド — オブザーバビリティと合わせて本番運用の安全性を確保

AIエージェントの導入・運用についてのご質問・ご相談はお問い合わせフォームからお気軽にどうぞ。

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

Need help moving from reading to rollout?

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

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

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

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