LLMアプリを運用していると、「回答が悪かった」という結果だけでは原因を特定できません。実際には、プロンプトのどのバージョンを使ったのか、入力トークン数はいくつか、処理に時間がかかった箇所はどこか、という情報が必要になります。
この記事では、LangfuseのPython SDK v4を使い、プロンプト・LLM呼び出し・回答を1つのトレースとして記録し、さらに評価スコアを付与する方法を試します。結論として、Langfuseを「ログ保存先」ではなく、プロンプト改善と回帰テストの基盤として使うのがポイントです。
1. 環境を準備する
Langfuseでプロジェクトを作成し、SettingsからAPIキーを発行します。Langfuse Cloudを使う場合は、次の環境変数を設定しました。セルフホストの場合は LANGFUSE_BASE_URL を自分のURLに変更します。
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export OPENAI_API_KEY="sk-..."
Python SDKとLLMクライアントをインストールします。今回のコードは、現在のOpenTelemetryベースのPython SDK v4を前提にしています。
python -m venv .venv
source .venv/bin/activate
pip install langfuse openai
PowerShellの場合は、次のように設定できます。
$env:LANGFUSE_PUBLIC_KEY="pk-lf-..."
$env:LANGFUSE_SECRET_KEY="sk-lf-..."
$env:LANGFUSE_BASE_URL="https://cloud.langfuse.com"
$env:OPENAI_API_KEY="sk-..."
SDK v4では、従来の trace() や generation() よりも、get_client() と start_as_current_observation()、または @observe を使う構成が基本です。Langfuse公式ドキュメント
2. Langfuseでプロンプトを管理する
まずLangfuseのPrompt Managementで、次のプロンプトを作成しました。
- 名前:
qa-answer - 種類: Text
- ラベル:
production - 本文:
次の質問に日本語で回答してください。
根拠が不明な場合は、推測せず「わかりません」と回答してください。
質問:
アプリケーションコードにプロンプトを直接埋め込まず、Langfuseからラベル付きで取得します。これにより、プロンプトを変更してもアプリケーションを再デプロイせず、バージョンごとの結果を比較できます。
3. プロンプトと回答をトレースする
次のコードでは、1回のユーザーリクエストをルートのspan、その中のLLM呼び出しをgenerationとして記録します。prompt=prompt を指定することで、どのプロンプトバージョンが使われたかもgenerationに紐づきます。
from openai import OpenAI
from langfuse import get_client, propagate_attributes
langfuse = get_client()
llm = OpenAI()
def answer_question(question: str, expected_keyword: str | None = None) -> str:
prompt = langfuse.get_prompt("qa-answer", label="production")
prompt_text = prompt.compile(question=question)
with langfuse.start_as_current_observation(
as_type="span",
name="qa-request",
input={"question": question},
) as root:
with propagate_attributes(
trace_name="qa-answer",
user_id="demo-user",
session_id="demo-session",
environment="development",
):
with langfuse.start_as_current_observation(
as_type="generation",
name="answer-generation",
model="gpt-5-mini",
prompt=prompt,
input=[{"role": "user", "content": prompt_text}],
) as generation:
response = llm.chat.completions.create(
model="gpt-5-mini",
messages=[
{"role": "user", "content": prompt_text}
],
)
answer = response.choices[0].message.content or ""
if response.usage:
generation.update(
output=answer,
usage_details={
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
},
)
else:
generation.update(output=answer)
root.update(output={"answer": answer})
# 簡易的なオンライン評価
if expected_keyword is not None:
score = 1.0 if expected_keyword in answer else 0.0
langfuse.score_current_trace(
name="keyword_match",
value=score,
data_type="NUMERIC",
comment="期待キーワードの有無",
)
return answer
if __name__ == "__main__":
try:
print(answer_question("Pythonで辞書を作る方法は？", "dict"))
finally:
# 短時間で終了するプロセスでは必ず送信を待つ
langfuse.flush()
実行後、LangfuseのTraces画面で qa-request を開くと、入力、回答、generation、モデル名、トークン使用量、プロンプトのバージョンを確認できます。
特に便利だったのは、プロンプトを修正した後に、トレースをプロンプトのバージョン別に比較できる点です。回答だけを保存するよりも、「どの指示で回答品質が変わったか」を追いやすくなります。プロンプトとトレースを紐づける公式ガイド
4. スコアを付けて回答品質を確認する
本番環境では、ユーザー評価やルールベースのチェック結果をスコアとして送信できます。例えば、回答が空でないかを評価する場合は次のように書けます。
from langfuse import get_client
langfuse = get_client()
trace_id = "トレースID"
langfuse.create_score(
name="not_empty",
value=1.0,
trace_id=trace_id,
data_type="NUMERIC",
comment="回答が空でない",
)
langfuse.flush()
スコアには数値だけでなく、真偽値、カテゴリ、テキストも利用できます。人手評価、ユーザーの「役に立った」ボタン、NGワード検出、LLM-as-a-Judgeの結果などを同じ仕組みに集約できます。Scores via API/SDK
ただし、キーワード一致だけでは意味的な正しさを評価できません。最初は決定的なルールで始め、必要になった段階で評価用LLMや人手レビューを追加するのが扱いやすいと感じました。
5. Datasetでプロンプト変更を回帰テストする
本番で見つかった質問と期待回答をDatasetに保存しておくと、プロンプト変更前後を同じ入力で比較できます。まず一度だけDatasetとテストケースを登録します。
from langfuse import get_client
langfuse = get_client()
dataset_name = "qa-cases"
langfuse.create_dataset(
name=dataset_name,
description="日本語QAの回帰テスト",
)
items = [
{
"id": "python-dict",
"input": {"question": "Pythonで辞書を作る方法は？"},
"expected_output": "dict",
},
{
"id": "unknown-answer",
"input": {"question": "存在しない仕様の正確な数値は？"},
"expected_output": "わかりません",
},
]
for item in items:
langfuse.create_dataset_item(
dataset_name=dataset_name,
**item,
)
langfuse.flush()
次に、Datasetを取得して実験を実行します。answer_question には、本番で使っているアプリケーション関数をそのまま渡すのが重要です。
from langfuse import Evaluation, get_client
from app import answer_question
langfuse = get_client()
def application_task(*, item, **kwargs):
return answer_question(item.input["question"])
def contains_expected(*, output, expected_output, **kwargs):
return Evaluation(
name="contains_expected",
value=1.0 if expected_output in output else 0.0,
)
try:
dataset = langfuse.get_dataset("qa-cases")
result = dataset.run_experiment(
name="qa-prompt-test",
run_name="qa-prompt-production-v1",
description="本番プロンプトの回帰テスト",
task=application_task,
evaluators=[contains_expected],
)
print(result.format())
finally:
langfuse.flush()
Dataset実験では、各入力に対する出力と評価値を確認できます。プロンプトを変更したら run_name を変えて再実行し、同じテストケースでスコアや回答内容を比較します。Datasetを使った評価の公式ガイド
まとめ
記事で紹介した、プロンプト管理から回答評価・回帰テストまでの流れを図にすると次の通りです。
今回試した構成では、次の流れを作れました。
- Langfuseからプロンプトのバージョンを取得する
- ユーザー入力、LLM呼び出し、回答をトレースする
- トークン使用量やモデル情報を記録する
- ルールベースや人手評価をスコアとして保存する
- Datasetでプロンプト変更を回帰テストする
最初から複雑な評価基盤を作る必要はありません。まずはLLM呼び出しをgenerationとして記録し、失敗した入力をDatasetに追加するところから始めると、プロンプト改善のサイクルを回しやすくなります。