Langfuse v4 で LLM アプリを計測・改善する — Sessions / Users / Scores 実践ガイド

はじめに

LLM アプリケーションを作っていると、必ず遭遇する壁があります。

• プロンプトを変えたけど、本当に精度が上がったのか分からない
• ユーザーから「なんか最近おかしい」と言われたが、どの呼び出しが悪かったか追えない
• コストが膨らんでいるが、どのユーザー・どの機能が重いのか見えない

最初は print() でログを出していたのが、logging に移り、そのうち SQLite に書き始め、気づけば自前の「なんちゃって LLMOps 基盤」を作っている……というのがよくある成り行きです。筆者もそうでした。

この記事では、Langfuse v4 を使って LLM アプリに「計測と改善のループ」を最小コストで入れる方法を、Python SDK を中心に解説します。

この記事の読者：

• Python or TypeScript で LLM アプリを書いている
• 自前ログから卒業したい
• プロンプト改善を「なんとなく」から「データで判断」に変えたい

押さえる 4 機能：

1. Tracing — LLM 呼び出しの入出力・コスト・レイテンシ
2. Sessions — 関連する呼び出しをグルーピング
3. Users — ユーザー別の観測
4. Scores — 品質の数値化

この 4 つを押さえると、計測と改善のループに必要な土台が整います。

---

なぜ自前実装じゃなく Langfuse なのか

LLM アプリのログ取りは「簡単そう」に見えて、実際にやると以下を全部自分で作ることになります：

やること	自前実装の難しさ
入力・出力・モデル名・トークン数・コスト・レイテンシを揃えて保存	スキーマ設計とマイグレーション地獄
「チェーン」呼び出しを階層化してタイムライン表示	親子関係を持つ span の DB 設計 + 可視化 UI
ユーザー別・セッション別の集計ダッシュボード	SQL + フロント開発
プロンプトバージョン別の精度比較	バージョン管理 + 評価パイプライン
スコア（品質指標）の時系列プロット	時系列 DB または集計バッチ + グラフ UI
SDK ベンダーごとのレスポンス差異の吸収	OpenAI / Anthropic / Gemini のトークン数フィールド名がバラバラ

これを自前で作ろうとすると、本来作りたいアプリの開発時間がそこに取られていきます。

Langfuse は上記を全部カバーしてくれる OSS の LLMOps プラットフォームです。Cloud 版は無料枠（Hobby プラン）があり、個人開発でも始められます。Self-hosted もできるので、取引情報のような機微データは手元で運用できます。

Langfuse を使うメリット（具体）

1. SDK を呼ぶだけで trace が残る（@observe デコレータ 1 行）
2. OpenAI / Anthropic / Gemini / LangChain の自動計装がある（ラッパー or インテグレーション）
3. コスト計算がモデル名から自動（token 数と model 名で自動計算）
4. バージョン別比較が UI で完結（プロンプト v1 vs v2 の Score 分布を並べて表示）
5. Dataset + Experiments で golden case に対する自動評価ができる

---

セットアップ（最小例）

1. プロジェクト作成 & キー取得

1. https://cloud.langfuse.com でアカウント作成
2. Project を作って Settings → API Keys から Public Key と Secret Key を取得
3. .env に以下を追加

LANGFUSE_PUBLIC_KEY=pk-lf-xxxxxxxx
LANGFUSE_SECRET_KEY=sk-lf-xxxxxxxx
LANGFUSE_HOST=https://us.cloud.langfuse.com  # or eu.cloud.langfuse.com

2. SDK インストール

# Python (v4 系を推奨、OpenTelemetry ベース)
pip install "langfuse>=4.3"

# Node.js / TypeScript
npm install langfuse

3. 初回の trace を送る

Python 最小例：

from langfuse import Langfuse

langfuse = Langfuse()  # 環境変数から自動読み込み

with langfuse.start_as_current_observation(
    name="hello.first_trace",
    as_type="span",
    input={"question": "What is Langfuse?"},
) as span:
    # ここに LLM 呼び出しや処理を書く
    answer = "LLM observability platform"
    span.update(output={"answer": answer})

langfuse.flush()  # プログラム終了前に flush

これだけで Langfuse UI の Tracing タブに 1 件出てきます。

---

1. Tracing — LLM 呼び出しを見える化する

基本：@observe デコレータでラップ

関数を trace に自動変換する最も簡単な方法は @observe デコレータです。

from langfuse import observe, Langfuse
from openai import OpenAI

langfuse = Langfuse()
openai_client = OpenAI()

@observe(as_type="generation")
def ask_llm(question: str) -> str:
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
    )
    return response.choices[0].message.content

@observe()
def main_flow(question: str):
    # この関数全体が 1 つの trace になる
    # 中で呼ばれる ask_llm は子 span として階層化される
    answer = ask_llm(question)
    return answer

main_flow("Langfuse とは？")

Langfuse UI でこう見えます：

Trace: main_flow
  └── ask_llm (generation)
        ├── input:  "Langfuse とは？"
        ├── output: "LLM 可観測性プラットフォーム..."
        ├── model:  gpt-4o-mini
        ├── tokens: input=12 / output=45
        └── cost:   $0.000027

OpenAI / Anthropic の自動計装

@observe を書かなくても、Langfuse がラップした SDK を使えば自動で trace になります：

# OpenAI のラッパー
from langfuse.openai import OpenAI

client = OpenAI()  # これを使うだけで、全 chat.completions.create が trace される
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
)

同様に Anthropic、Google Gemini、LangChain、LlamaIndex も自動計装が提供されています。詳細は Langfuse Integrations を参照。

ネストした処理（チェーン）

@observe()
def rag_pipeline(query: str):
    docs = retrieve(query)        # 子 span 1
    context = rerank(docs, query)  # 子 span 2
    answer = generate(context, query)  # 子 span 3（LLM 呼び出し）
    return answer

@observe()
def retrieve(query): ...

@observe()
def rerank(docs, query): ...

@observe(as_type="generation")
def generate(context, query): ...

1 つの trace の中に階層化された span が並び、「どこで時間がかかっているか」「どこで失敗したか」 が一目で分かります。

---

2. Sessions — 関連する trace をグルーピング

いつ使うか

• マルチターン会話: 1 つの会話の中で複数回 LLM を呼ぶ（チャットボット）
• ワークフロー: 1 つのドキュメント処理で複数ステージを経る（分類 → 抽出 → 検証）
• ユーザーセッション: 1 回のブラウザ訪問中の全呼び出しを束ねる

書き方

Session は session_id 属性を trace に付けるだけです。v4 Python SDK では propagate_attributes() を使います：

from langfuse import Langfuse, propagate_attributes

langfuse = Langfuse()

with propagate_attributes(session_id="conv-20260421-abc"):
    # この中で作られる全 trace / span に session_id が付く
    answer1 = chat("こんにちは")
    answer2 = chat("今日の天気は？")
    answer3 = chat("じゃあ傘は必要？")

Langfuse UI での見え方

Sessions タブ（左サイドバー）に conv-20260421-abc が 1 行で表示され、クリックすると：

• 時系列での全 trace
• セッション全体のトータルコスト・トータルレイテンシ
• 会話の流れ（input / output が順に並ぶ）

マルチターン会話の実装例

def handle_message(conversation_id: str, user_message: str) -> str:
    with propagate_attributes(session_id=conversation_id):
        history = load_history(conversation_id)
        history.append({"role": "user", "content": user_message})

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=history,
        )

        assistant_message = response.choices[0].message.content
        save_history(conversation_id, assistant_message)
        return assistant_message

conversation_id を session_id にそのまま流用するだけ。これで 1 会話の全ターンが Langfuse 上でグループ化されます。

---

3. Users — ユーザー別の観測

いつ使うか

• コスト按分: 有料プランのユーザー・フリーミアムでヘビーユーザーを特定
• エラー分析: 特定ユーザーでだけ高頻度のエラーが出るパターン
• パーソナライゼーション効果検証: ユーザーごとの Score 分布

書き方

propagate_attributes() に user_id を渡します：

with propagate_attributes(
    session_id="conv-20260421-abc",
    user_id="user_12345",
):
    answer = chat("質問内容")

Langfuse UI での見え方

Users タブに user_12345 が表示され、以下が見えます：

• そのユーザーの総 trace 数
• 累計コスト・累計トークン数
• エラー率
• 最後のアクティビティ日時
• 紐付く全 trace のタイムライン

どの ID を user_id にするか

サービスタイプ	推奨 user_id
認証ありの SaaS	認証システムのユーザー ID（Firebase UID、Auth0 sub、Better Auth id 等）
匿名チャットボット	セッション Cookie or ブラウザフィンガープリント
Slack Bot	Slack user ID
API サービス	API キーに紐付く顧客 ID

GDPR 注意: user_id に個人情報（メールアドレス・実名）を入れないこと。ハッシュ化した内部 ID を使います。

---

4. Scores — 品質を数値化する

ここが LLMOps の核です。Score = その trace が「どれだけ良かったか」の数値。Score を軸に時系列プロットすると、プロンプト変更・モデル変更の効果が視覚的に見えます。

Score の 3 大ソース

ソース	例	実装コスト
ユーザーフィードバック	チャット UI の 👍 / 👎 ボタン	低
自動評価（LLM-as-a-judge）	GPT-4 で出力をスコアリング	中
運用者の編集行動	Human-In-The-Loop（HITL）ワークフローで人間がどれだけ修正したか	中

実装 1: ユーザーフィードバック

チャット UI に「👍 良い回答 / 👎 違う」ボタンを置き、API で Langfuse に送信：

# FastAPI 例
from fastapi import FastAPI
from langfuse import Langfuse

app = FastAPI()
langfuse = Langfuse()

@app.post("/feedback")
async def feedback(trace_id: str, value: int):  # value = 1 or 0
    langfuse.create_score(
        trace_id=trace_id,
        name="user.thumbs",
        value=value,
        data_type="BOOLEAN",
    )
    return {"ok": True}

フロント側では trace_id を start_as_current_observation の返り値から取り出してレスポンスに含めておき、ユーザーがボタンを押したらそれを送信するのみ。

実装 2: LLM-as-a-judge

別の LLM に「この出力が query に対して的確か 0〜1 で評価して」と聞き、スコアを保存：

@observe(as_type="generation")
def evaluate_answer(query: str, answer: str) -> float:
    judge_prompt = f"""
以下の質問に対する回答の的確さを 0〜1 で評価してください。
数値のみ返答。

質問: {query}
回答: {answer}
スコア:
"""
    response = judge_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": judge_prompt}],
    )
    return float(response.choices[0].message.content.strip())

# 本番回答のあとに非同期で走らせる
def ask_and_score(query: str):
    with langfuse.start_as_current_observation(name="qa") as span:
        answer = llm_answer(query)
        score = evaluate_answer(query, answer)
        span.score(name="llm_judge.accuracy", value=score)
        return answer

実装 3: 人間によるレビュー編集量（HITL）

人間が AI 出力をレビューして修正する UI がある場合、「どれだけ修正したか」を Score にする。プロンプト改善の定量評価ループになります。

Score 名（hitl.edit_rate など）は Langfuse 側で決まっているわけではなく、自由に命名できます。ここでは <カテゴリ>.<指標名> という命名規則で揃えています。

def on_approve(trace_id: str, ai_output: dict, edited: dict):
    diff_rate = compute_edit_rate(ai_output, edited)  # 0〜1
    langfuse.create_score(
        trace_id=trace_id,
        name="hitl.edit_rate",
        value=diff_rate,
        comment=f"editedFields={count_changed(ai_output, edited)}",
    )

def compute_edit_rate(a: dict, b: dict) -> float:
    """a と b を比較して、変更フィールド数 / 全フィールド数 を返す"""
    keys = set(a) | set(b)
    changed = sum(1 for k in keys if a.get(k) != b.get(k))
    return changed / len(keys) if keys else 0.0

この Score が 低ければ低いほど AI 出力の精度が高い（人間がほとんど修正しなかった）ので、プロンプト改善の北極星指標になります。

---

5. プロンプトチューニングの実践ループ

Langfuse を入れただけでは精度は上がりません。これら 4 機能を組み合わせて「観測 → 仮説 → 実験 → 監視」のループを回すことがゴールです。

ステップ 1: 弱点発見（Tracing + Scores）

Langfuse UI の Tracing タブで、Score が低い trace だけをフィルタします。

Filter: score.name = "hitl.edit_rate" AND score.value > 0.5

Score が高い（= 人間が大幅に修正した）trace を並べて、共通パターンを探します。

• 特定のドキュメント形式で精度が悪い？
• 特定のフィールドだけ誤抽出が多い？
• プロンプトに含まれていない指示が必要なケース？

ステップ 2: 仮説検証（Datasets + Experiments）

弱かった入力をまとめて Langfuse の Dataset として保存します（UI から右クリック → "Add to dataset"）。

Dataset に対して新しいプロンプトで Experiment を回すと、Score の分布が並べて比較できます：

from langfuse import Langfuse

langfuse = Langfuse()
dataset = langfuse.get_dataset("weak-cases-2026-04")

for item in dataset.items:
    with item.observe(run_name="prompt-v3") as trace:
        output = new_prompt_llm(item.input)
        trace.update(output=output)
        # 自動評価 or 人手評価でスコア付け

Experiments タブで v2 と v3 の Score 中央値・分布が見られます。

ステップ 3: デプロイと監視（Sessions + Scores）

本番にデプロイしたら、Langfuse の Dashboards タブで Score の時系列を監視：

• プロンプト v3 にリリースした直後に hitl.edit_rate が下がった → 改善成功
• 逆に上がった → リグレッション。v2 にロールバック

時系列プロットと session / user フィルタを組み合わせれば、「特定のユーザー層だけ悪化した」のような細かい発見もできます。

このループで大事なこと

• Score の設計が全て。何を測るかを決めずに始めない
• Score は複数持っていい。hitl.edit_rate（編集率）と llm_judge.relevance（関連性）は別々に追跡する
• 自動評価と人手評価を使い分ける。自動は量、手動は質

---

よくある詰まりポイント

metadata と trace-level 属性の違い

初見で一番混乱するポイントです。Langfuse には属性の置き場所が 2 つあります：

置き場所	設定方法	UI での見え方
trace-level 属性	propagate_attributes(session_id, user_id)	Trace 詳細上部のバッジ、Sessions / Users 画面に集計される
metadata	start_as_current_observation(metadata={...})	Trace 詳細の Metadata セクションにのみ表示

「metadata に session_id を入れたのに Sessions 画面に出ない」というのは、propagate_attributes を使わないといけないのに metadata に入れている状態です。

UI が分からないときは API で確認

「本当に送信されているのか？」を確認したいとき、Langfuse Public API を直接叩くのが最速です：

TRACE_ID="a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6"
curl -s "https://us.cloud.langfuse.com/api/public/traces/${TRACE_ID}" \
  -H "Authorization: Basic $(echo -n 'pk-lf-xxx:sk-lf-xxx' | base64)" \
  | jq '{sessionId, userId, tags, scores: [.scores[] | {name, value}]}'

出力：

{
  "sessionId": "conv-20260421-abc",
  "userId": "user_12345",
  "scores": [
    { "name": "hitl.edit_rate", "value": 0.12 }
  ]
}

UI がもっさりしているとき、キャッシュで反映が遅いとき、API 経由なら直接中身を確認できます。

flush を忘れない

短命プロセス（スクリプト・Lambda・バッチ）では 終了前に langfuse.flush() を呼ばないと trace が送信されません。

# スクリプトの最後
langfuse.flush()

long-running なサーバーでは flush は不要ですが、graceful shutdown で呼んでおくと安全です。

---

まとめ

Langfuse v4 の 4 機能を組み合わせると、計測と改善のループに必要な土台が整います。

機能	用途	実装コスト
Tracing	何が起きたか見る	@observe 1 行
Sessions	関連する trace を束ねる	session_id 属性を渡す
Users	ユーザー別の観測	user_id 属性を渡す
Scores	品質を数値化	create_score() or span.score()

最初の一歩：

1. pip install "langfuse>=4.3" して .env に API キー
2. 一番使っている LLM 呼び出し関数に @observe(as_type="generation") を付ける
3. UI の Tracing タブで 1 件 trace が出るのを確認
4. チャットがあるなら propagate_attributes(session_id=..., user_id=...) で包む
5. フィードバックボタンがあるなら create_score() で送る

改善ループ：

観測（Tracing + Scores）
  ↓
弱点特定（低スコアの trace をフィルタ）
  ↓
仮説検証（Dataset + Experiments で A/B 比較）
  ↓
デプロイ（新プロンプト本番投入）
  ↓
監視（Dashboards で Score 時系列）
  ↓ （リグレッションなら戻す）
観測 に戻る

自前で LLMOps 基盤を組む時間を、Langfuse に任せて 「LLM アプリそのもの」の改善に振り向ける という選び方もあります。まずは最小構成で 1 件 trace を送ってみて、慣れてきたら Sessions / Users / Scores を順に足していくのが進めやすい順番です。

---

参考：

• Langfuse 公式ドキュメント
• Python SDK v4 リファレンス
• Tracing: @observe デコレータ
• Scores API

使用したバージョン：

• Langfuse Cloud v3.169（2026年4月時点）
• Python SDK: langfuse>=4.3.1（OpenTelemetry ベース）