1. はじめに
生成AIを活用したアプリケーション開発において、以下のような課題に直面したことはありませんか？
- 「本番環境でLLMが変な応答を返したけど、何が原因か分からない」
- 「プロンプトを変更するたびにコードをデプロイするのが面倒」
- 「どのプロンプトバージョンがパフォーマンス良いか分からない」
- 「トークン使用量やコストが把握できていない」
- 「非エンジニアのチームメンバーがプロンプトを編集できない」
これらの問題を解決するためにLangfuseを紹介したいと思います
Langfuse is an open-source LLM engineering platform that helps teams collaboratively debug, analyze, and iterate on their LLM applications. All platform features are natively integrated to accelerate the development workflow. Langfuse is open, self-hostable, and extensible
引用元:(https://langfuse.com/docs)
Langfuseは、単なる監視ツールではなく、LLMアプリケーション開発のライフサイクル全体を支援する、オープンソースの「LLMエンジニアリングプラットフォーム」です。大きな特徴としては以下になります。
- トレーシング (Tracing): アプリケーションの動作を詳細にデバッグ・可視化します
- メトリクス (Metrics): コスト、レイテンシ、品質をダッシュボードで分析します
- 評価 (Evaluation): 自動または手動でLLMの出力品質をテスト・評価します
- プロンプト管理 (Prompt Management): プロンプトのバージョン管理と本番デプロイを行います
本記事では、Vertex AI Geminiを例に、Langfuseの導入戦略からセットアップ、そしてトレーシング、プロンプト管理まで、公式ドキュメントを引用しつつ実体験を踏まえながら書いていきます。
まだローカルな環境で実行したに過ぎず、より良い使い方を模索している最中です...「こういうこと出来るよ」とか「こうしたら良くなるよ」等あるかと思います,,,その際はコメントでそっと教えていただけると嬉しいです！
2. 導入戦略：Langfuse Cloud vs セルフホスティング
Langfuseの導入にあたり、最初の戦略的な岐路は「Langfuse Cloud」を利用するか、「セルフホスティング（OSS版）」を自社インフラに構築するかです。
この選択は、主に「運用負荷（インフラ管理を事業者様にお任せするか）」と「データ主権（トレースデータを自社のインフラ内に保持する必要があるか）」という2つのトレードオフによって決定されます。
-
Langfuse Cloud: フルマネージドサービスになるので、インフラの運用・保守負荷を手放せます。プランによってはSOC2, ISO27001, HIPAA準拠など、コンプライアンス要件にも対応しているようです（https://langfuse.com/pricing）
-
Self-Hosted (OSS版): すべてのデータが自社のインフラストラクチャ内に留まるため、データ主権を維持できます。Docker Compose, Kubernetes (Helm), Terraformなど多様な導入方法がサポートされていますが、インフラのセットアップと継続的なメンテナンス（アップデート、スケーリング）は当然自己責任となります。
今回はCloud版を使って遊んでみました。OSS版だとローカルで遊ぶ分には簡単ですが、クラウド運用は環境だけでちょこっと費用がかかるので個人で利用する分にはCloud版で良いなと感じました
因みに、OSS版で、トレーシング、プロンプト管理、評価といったコア機能は、スケーラビリティの制限なく無料で利用できます（すごい）
3. セットアップと基本のトレーシング（「何が」見えるのか）
3.1. セットアップ
Langfuse Cloud にアクセスして、無料アカウントを作成します
プロジェクトを作成したら、Settings → API Keysから以下を取得します
- Public Key (pk-lf-...)
- Secret Key (sk-lf-...)
3.2. Pythonパッケージのインストール
Vertex AI（google-cloud-aiplatform）とLangfuse（langfuse）をインストールします。
pip install langfuse
pip install google-cloud-aiplatform
pip install python-dotenv
3.3 環境変数の設定
.envファイルを作成し、APIキーを設定します。LANGFUSE_HOSTは、Cloud版のデフォルト（USリージョン）では不要ですが、EUリージョンやセルフホスティングの場合は設定します。
# Langfuse設定
LANGFUSE_PUBLIC_KEY=pk-lf-xxxxxxxxxx
LANGFUSE_SECRET_KEY=sk-lf-xxxxxxxxxx
LANGFUSE_HOST=https://us.cloud.langfuse.com # USリージョンの場合
# GCP設定 (Vertex AI用)
GCP_PROJECT_ID=your-project-id
GCP_LOCATION=us-central1
3.4 デコレータ
Vertex AIのようなサポートされていないSDKや、独自の前処理・後処理ロジックをトレースしたい場合は、@observeデコレータが便利です
import os
from dotenv import load_dotenv
import vertexai
from vertexai.generative_models import GenerativeModel
from langfuse import observe, Langfuse
#.envファイルから環境変数を読み込む
load_dotenv()
# Vertex AI初期化
vertexai.init(
project=os.getenv("GCP_PROJECT_ID"),
location=os.getenv("GCP_LOCATION")
)
# Langfuse初期化
# 環境変数 LANGFUSE_SECRET_KEY, LANGFUSE_PUBLIC_KEY, LANGFUSE_HOST が設定されていれば、引数なしで初期化可能
langfuse = Langfuse(
secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
host=os.getenv("LANGFUSE_HOST")
)
@observe(name="text-generation", as_type="generation")
def generate_text(prompt: str) -> str:
model = GenerativeModel("gemini-2.5-flash")
response = model.generate_content(prompt)
# 重要 トークン使用量を送信する
# Vertex AI SDKは自動でLangfuseに送信しないため、手動で更新する
try:
input_tokens = response.usage_metadata.prompt_token_count
output_tokens = response.usage_metadata.candidates_token_count
total_tokens = response.usage_metadata.total_token_count
langfuse.update_current_observation(
input=prompt,
output=response.text,
model="gemini-2.5-flash",
# usage引数に渡すことで、コストが自動計算される
usage={
"input": input_tokens,
"output": output_tokens,
"total": total_tokens,
}
)
except Exception as e:
print(f"Error updating langfuse observation: {e}")
langfuse.update_current_observation(
input=prompt,
output=response.text,
model="gemini-2.5-flash",
metadata={ "warning": "Usage metadata not available" }
)
return response.text
# 実行
result = generate_text("Pythonの特徴を3つ教えてください")
print(result)
# プログラム終了時にキューイングされたデータを送信
langfuse.flush()
たったこれだけで、Langfuse Cloudのダッシュボードでトレースを確認できます！（すごい）
@observeデコレータは、その関数の入出力、実行時間、エラーを自動的にトレースします。
If either usage or cost are not ingested, Langfuse will attempt to infer the missing values based on the model parameter of the generation at the time of ingestion. This is especially useful for some model providers or self-hosted models which do not include usage or cost in the response.
https://langfuse.com/docs/observability/features/token-and-cost-tracking
4. 実践:プロンプト管理（即時デプロイとロールバック）
Langfuseの便利すぎる機能が、プロンプト管理です。これにより、プロンプトとアプリケーションコードを完全に分離できます。
4.1. バージョンとラベルの概念
Langfuseのプロンプト管理は、Gitに似た「バージョン」と「ラベル」の概念で動作します。
- ラベル (Labels): 特定のバージョン（例：v2）に付けられる、人間が読めるエイリアス（例：production, staging）
4.2. Configによるモデルパラメータの同時管理
Langfuseは、プロンプトテキストと同時に、これらのモデルパラメータをconfig（JSON形式）として一括でバージョン管理できます。
Langfuse Cloudのダッシュボードから作成する場合（[Prompts] → [New Prompt]）
- name を code-reviewer に設定
- prompt (テキスト) に {{code}} や {{criteria}} といったテンプレート変数を記述
- config (JSON) にモデル設定を記述
{
"model": "gemini-2.5-flash",
"temperature": 0.2
}
こんな感じで書いて保存します。（そしてproduction ラベルを付ける とかしてください）
4.3. Langfuseを使った実装
アプリケーションコード側では、ラベル（例：production）でプロンプトを取得します。
@observe(name="code-review", as_type="generation")
def review_code(code: str) -> str:
# Langfuseから "production" ラベルが付いたプロンプトを取得
try:
prompt = langfuse.get_prompt("code-reviewer", label="production")
except Exception as e:
print(f"Failed to get prompt from Langfuse, using fallback: {e}")
return "Error: Could not retrieve prompt."
# 【重要】現在のトレース（generation）に、どのプロンプトバージョンを使ったかを紐付ける
# これにより、UIでプロンプトバージョンごとの性能（コスト、レイテンシ、スコア）を分析できる
langfuse.update_current_observation(prompt=prompt)
# 変数をプロンプトに埋め込む
compiled_prompt_text = prompt.compile(
code=code,
criteria="セキュリティ、パフォーマンス"
)
# configからモデル名や設定を取得
model_name = prompt.config.get("model", "gemini-2.5-pro") # デフォルト値を設定
temperature = prompt.config.get("temperature", 0.2)
model = GenerativeModel(model_name)
response = model.generate_content(
compiled_prompt_text,
generation_config={"temperature": temperature}
)
#... update_current_observationでusageなどを記録...
return response.text
何が嬉しいのか
「ソースコードとプロンプトの分離」と「即時ロールバック」の点で嬉しいです。
- デプロイ: アプリコードは
label="production"を参照するようにデプロイ - 改善: チームがUIでプロンプトを改善し、新しいバージョンとして保存
- 新デプロイ: 本番に適用するため、ソースコードの再デプロイは不要です！ Langfuse UIにログインし、production ラベルのポインターを付け替えるだけです!
- インシデント発生: 新しいプロダクトに問題があり、品質スコア（後述）が急落したことをダッシュボードで検知したとします
- 即時ロールバック: 修正は、Langfuse UIで productionラベルをv2から安全なv1に付け戻すだけです。CI/CDを回す必要も、hotfixをデプロイする必要もありません！
この「監視→分析→即時ロールバック」のサイクルの速さが、Langfuseのプロンプト管理機能がもたらす最大の運用メリットだと思います
（productionをv3からv2に簡単に戻せます）
（バージョンごとにそのプロンプトでの出入力も閲覧可能です）
5. 実践: 継続的改善ループ（評価とフィードバック）
トレーシングで「何が起きたか」を可視化し、プロンプト管理で「改善」の手段を得ました。次はその「いつ改善すべきか」を判断する「評価 (Evaluation)」です。
Langfuseは、開発中の「オフライン評価」（既知のデータセットに対するテスト）と、本番環境の「オンライン評価」（未知のライブトラフィックに対する監視）の両方をサポートします。（ここも普通にすごい笑）
ここでは、本番環境での「オンライン評価」の3つの主要な手法と、それらを組み合わせた「継続的改善ループ」を紹介します
5.1. 方法1：エンドユーザーフィードバック (オンライン・手動)
最も重要な品質メトリクスは、エンドユーザーの満足度です。アプリケーションのUIに「👍/👎」ボタンを実装し、ユーザーの生の評価を収集できます
import { LangfuseWeb } from "langfuse";
export function UserFeedbackComponent(props: { traceId: string }) {
const langfuseWeb = new LangfuseWeb({
publicKey: env.NEXT_PUBLIC_LANGFUSE_PUBLIC_KEY,
baseUrl: "https://cloud.langfuse.com", // 🇪🇺 EU region
// baseUrl: "https://us.cloud.langfuse.com", // 🇺🇸 US region
});
const handleUserFeedback = async (value: number) =>
await langfuseWeb.score({
traceId: props.traceId,
name: "user_feedback",
value,
});
return (
<div>
<button onClick={() => handleUserFeedback(1)}>👍</button>
<button onClick={() => handleUserFeedback(0)}>👎</button>
</div>
);
}
5.2. 方法2：LLM-as-a-Judge (オンライン・自動)
LLMを「裁判官（Judge）」として使い、本番の応答が「ハルシネーションを起こしていないか」「RAGの文脈に基づいているか」などを自動でスコアリングする手法です。
5.3. 方法3：人間によるアノテーションと「改善ループ」の確立
人間が直接見たって良いのです。「Annotation Queues（アノテーションキュー）」という機能があります。
6. 実践4: ABテスト
6.1. A/Bテスト（データドリブンな意思決定）
2つの異なるプロンプト（またはモデル）を本番環境で競わせ、どちらが優れたメトリクス（コスト、品質）を出すかをデータに基づいて判断することも出来そうです。
- 準備 (UI): Langfuse UIで、code-reviewerプロンプトに対し、prod-a ラベル（v1）と prod-b ラベル（v2）の2つのバージョンを準備します。
- 実装: アプリケーションコード側で、リクエストごとにランダム（またはユーザーIDで固定）でどちらかのプロンプトを取得します。
import random
@observe(name="ab-test-generation", as_type="generation")
def review_code_ab_test(code: str):
# 50%の確率で A か B を選択
variant = random.choice(["prod-a", "prod-b"])
prompt = langfuse.get_prompt("code-reviewer", label=variant)
# トレースにA/Bテストの情報を記録
langfuse.update_current_observation(
prompt=prompt,
metadata={"ab_test_variant": variant}
)
#... LLM呼び出し...
分析としては以下のような感じになるのだと思います。
LangfuseのMetricsダッシュボードで、2つをフィルターして
- Cost（prod-b は安いか？）
- Latency（prod-b は速いか？）
- user_feedback スコア（prod-b はユーザー満足度が高いか？）
上記の観点で比較して「prod-b が優れている」と判断したら、production ラベルを prod-b に切り替え、A/Bテストを終了します。
「何となく」ではなく「根拠を持った」意思決定が出来るのはとても便利なポイントだと思います！
9. まとめ
ここまで読んでいただきありがとうございました。
出来ることは他にも色々あるのは承知していますが、プロンプトの評価やその管理をソースコードと切り離して行えるのはかなり便利だなと思いました。それまで感覚に近かったものを定量で表現してカイゼンをしていけるのも嬉しいポイントだと思います。
理解が間違っていたり、逆にこんなことも出来るよ等あれば是非コメントで教えていただけると嬉しいです！