はじめに
こんにちはあんどおです。
Langfuse便利ですね。僕も毎日のように使っていますがアップデートが早くてついていくのに必死です。
今回は大量のトレースログからdatasetsを一括作成する方法についてまとめました。
サマリ
この記事のサマリです。
大量の本番トレースログを一括でdatasetsに変換
- UI経由の手作業でのdatasets作成で対応できない、本番環境での大規模ログをlangfuse python SDKで一括登録
Session APIを用いて会話単位のトレースログをまとめて取得し整形
- SessionIDで会話全体を一括取得し、評価用に整形
独自のvalidationでトレースログをフィルタし、データセットの質を担保
- 会話完了チェックやフローvalidationで無効なログを除外し、評価データセットの質を保持
前提
Langfuseにはトレースログをdatasetsに登録する方法が2種類ある
Langfuseでトレースログからdatasetsを作成する方法は、私の知っている限りでは以下の2つの方法があります。
- UI経由でdatasetsに手動登録
- SDK経由でdatasetsに一括登録
"UI経由で手動登録"ではlangfuseのUI(トレース画面)で該当のトレースを選択した上で、"add to datasets"ボタンを押下すれば簡単に特定のトレースログを指定のdatasetsに登録することができます。
一方で"SDK経由で一括登録"ではpythonファイルなどでスクリプトを作成し、指定したトレースログをdatasetsに登録できます。
トレースログのdatasetsへの登録方法の比較
| 方法 | 適用シーン | メリット | デメリット | 学習コスト |
|---|---|---|---|---|
| UI経由 | 〜数十件 | 直感的、簡単 | 大量データに不向き | 低い |
| SDK/API | 数百件〜 | 効率的、自動化可能 | コーディング必要 | 中程度 |
なぜSDKを利用するのか？
一言で言えば、大量かつ一定基準以上のトレースログを一括でdatasetsに登録したいからです。
実際にAgentの運用を開始すると大量のトレースログが生成されます。それらの大量ログを全てUIから手作業でdatasetsに登録していくことは不可能ではないですが、かなり非効率です。アプリケーションの規模が大きくなれば全て手作業で実施することは非現実的でしょう。
大量のトレースログを一括で登録
Langfuseは本番運用を考慮して設計されており、SDKを利用したスクリプトで以下のようなことができます。
- トレースログの一括取得
- datasetsへの一括登録
一定基準以上のトレースログのみフィルタ
さらにスクリプト内にvalidationをかませておけば無効なログがdatasetsに紛れ込むことも防げるので実験の品質も上がります。validationはpythonなどで自分で実装する必要があります。
上記を踏まえると、本格的なAgent運用のにおいてはSDKでのトレースログ一括datasets作成・登録機能は強力な武器になります。
実装：本番トレースログからdatasetsを作成
全体の構成
今回の例ではpython SDKを想定して実装します。会話形式のチャットボットのトレースログを利用してdatasetsを作るケースを想定します。
全体の作業工程は以下のとおりです。
[本番環境: Langfuse Sessions]
↓
[① Session情報取得] #sessioｎIDを利用
get_session(session_id)
↓
[② 会話履歴抽出・整形] #Session情報から会話履歴を取得
extract_conversation_history()
↓
[③ データ品質検証] #特定の基準で会話履歴をフィルタ
- is_conversation_completed()
- is_conversation_flow_valid()
↓
[④ Dataset作成・登録] #ログを一括でdatasetsに登録
langfuse.create_dataset_item()
↓
[Langfuse Dataset]
ポイントは以下の4つです。
- SessionIDを指定して、ひとまとまりの会話を取得する(取得データはjsonのlistになります)
- 必要に応じて会話を整形し、その後の処理で利用しやすい形に変換
- 任意のvalidationをかけ、評価に利用するログのみ抽出
(以下のコードではmetaデータにvalidationの結果を付与しフィルタできるようにしてます) - 処理済みのログをdatasetsに一括登録 (datasetsがなければ作成する)
個人的には評価用のデータセットにはvalidationを噛ませるのが大事だなと思っていますが、langfuseではその点を自由度高く実装できます。
サンプルコード
では具体的にどうなっているのか、サンプルコードを用いて説明します。
① Session情報取得
import os
import requests
def get_session(session_id: str) -> dict:
"""
LangfuseからセッションIDを使ってセッション情報を取得
Args:
session_id: 取得したいセッションのID
Returns:
セッション情報を含む辞書
Raises:
ValueError: 認証情報が設定されていない場合
requests.HTTPError: APIリクエストが失敗した場合
"""
# 環境変数から認証情報を取得
public_key = os.environ.get("LANGFUSE_PUBLIC_KEY")
secret_key = os.environ.get("LANGFUSE_SECRET_KEY")
host = os.environ.get("LANGFUSE_HOST")
if not all([public_key, secret_key, host]):
raise ValueError("Langfuse認証情報が設定されていません")
# APIエンドポイントを構築
url = f"{host}/api/public/sessions/{session_id}"
# Basic認証でGETリクエストを送信
response = requests.get(url, auth=(public_key, secret_key))
response.raise_for_status()
return response.json()
② 会話履歴抽出・整形
from typing import TypedDict
class ConversationItem(TypedDict):
"""会話履歴の1項目"""
input: str | None
output: dict | None
def extract_conversation_history(session_data: dict) -> list[ConversationItem]:
"""
セッションデータから会話履歴を時系列順に抽出
Args:
session_data: get_session関数で取得したセッション情報
Returns:
input, outputを含む辞書のリスト(時系列順にソート済み)
"""
conversation_history = []
# tracesからinputとoutputのみを取り出す
traces = session_data.get("traces", [])
for trace in traces:
conversation_history.append({
"input": trace.get("input"),
"output": trace.get("output"),
"timestamp": trace.get("timestamp"),
})
# timestampで時系列順(古い順)にソート
conversation_history.sort(key=lambda x: x.get("timestamp", ""))
# input/outputのみ返却
return [
{"input": item["input"], "output": item["output"]}
for item in conversation_history
]
③ データ品質検証
# 会話フェーズの順序定義(実際のユースケースに応じて定義)
CONVERSATION_PHASES = [
"greeting_agent",
"main_interaction_agent",
"summary_agent",
"closing_agent",
]
def is_conversation_completed(conversation_history: list[ConversationItem]) -> bool:
"""
会話が正常に完了しているかチェック
条件:
- 最後のagent_typeが終了エージェント(closing_agent)
- 終了エージェントが1回だけ出現
Args:
conversation_history: 抽出した会話履歴
Returns:
会話が正常に完了していればTrue
"""
if not conversation_history:
return False
# 終了エージェントの出現回数をカウント
closing_count = sum(
1 for item in conversation_history
if item.get("output", {}).get("agent_type") == "closing_agent"
)
# 終了エージェントが1つだけ存在するかチェック
if closing_count != 1:
return False
# 最後のagent_typeが終了エージェントかチェック
last_agent = conversation_history[-1].get("output", {}).get("agent_type")
return last_agent == "closing_agent"
def is_conversation_flow_valid(conversation_history: list[ConversationItem]) -> bool:
"""
会話フローの順序が正しいかチェック
条件:
- フェーズが定義順に進行(同じフェーズの連続はOK)
- 前のフェーズに戻るのはNG
Args:
conversation_history: 抽出した会話履歴
Returns:
会話フローが正しければTrue
"""
if not conversation_history:
return False
max_phase_index = -1 # 現在到達している最大のフェーズインデックス
for item in conversation_history:
agent_type = item.get("output", {}).get("agent_type")
# agent_typeが定義されたフェーズに存在するかチェック
if not agent_type or agent_type not in CONVERSATION_PHASES:
return False
current_index = CONVERSATION_PHASES.index(agent_type)
# フェーズが前に戻っている場合はNG
if current_index < max_phase_index:
return False
max_phase_index = max(max_phase_index, current_index)
return True
④ Dataset作成・登録
from langfuse import get_client
import logging
logger = logging.getLogger(__name__)
def create_dataset_from_sessions(
session_ids: list[str],
dataset_name: str,
evaluation_map: dict[str, dict] # session_id -> expected_outputのマッピング
) -> None:
"""
複数のセッションからDatasetを作成
Args:
session_ids: 処理対象のセッションIDリスト
dataset_name: 作成するDataset名
evaluation_map: 各セッションの期待される出力(expected_output)のマッピング
例: {"session_001": {"metric_1": True, "metric_2": False}}
※このデータは別途準備が必要(人手評価やルールベース評価など)
"""
langfuse = get_client()
# Dataset作成(存在しない場合のみ)
try:
langfuse.get_dataset(dataset_name)
logger.info(f"Dataset '{dataset_name}' は既に存在します")
except Exception:
langfuse.create_dataset(
name=dataset_name,
description="本番データから作成した評価用データセット"
)
logger.info(f"Dataset '{dataset_name}' を作成しました")
logger.info(f"処理対象のセッション数: {len(session_ids)}")
# 各セッションを処理
success_count = 0
error_count = 0
for idx, session_id in enumerate(session_ids, 1):
logger.info(f"[{idx}/{len(session_ids)}] セッション: {session_id}")
try:
# 1. セッション取得
session_data = get_session(session_id)
# 2. 会話履歴抽出
conversation_history = extract_conversation_history(session_data)
# 3. 評価結果取得(expected_output) ※このデータは事前に別途準備が必要
evaluation_result = evaluation_map.get(session_id, {})
# 4. データ品質検証
is_completed = is_conversation_completed(conversation_history)
is_flow_valid = is_conversation_flow_valid(conversation_history)
logger.info(f" 会話完了: {'OK' if is_completed else 'NG'}")
logger.info(f" フロー検証: {'OK' if is_flow_valid else 'NG'}")
# 5. Datasetアイテムとして追加
langfuse.create_dataset_item(
dataset_name=dataset_name,
input={
"session_id": session_id,
"conversation_history": conversation_history,
},
expected_output={
# 期待される評価結果(正解データ)
"evaluation_results": evaluation_result
},
metadata={
"session_id": session_id,
"conversation_count": len(conversation_history),
"is_completed": is_completed,
"is_flow_valid": is_flow_valid,
}
)
logger.info("追加成功")
success_count += 1
except Exception as e:
logger.error(f" エラー: {e}")
error_count += 1
continue
# 最終結果のサマリー
logger.info(f"\n{'='*50}")
logger.info(f"処理完了")
logger.info(f"成功: {success_count}件")
logger.info(f"失敗: {error_count}件")
logger.info(f"{'='*50}")
datasetsの活用
datasetsは基本的に評価に利用することになると思います。
langfuseにはrun_experimentというdatasetsと評価器を効率的に繋げる機能がありこれがとても便利です。
run_experiment機能はこちらの記事でまとめてますのでぜひ見てみてください！
まとめ
本番環境にある大量のログからdatasetsを一括で作れるのはとても便利ですね！
明らかに気になる目立った異常ログはUIからぽちぽちdatasetsに登録して、それが頻発するようならSDKで範囲を広げて一括でdatasetsに登録するみたいな感じで両方を上手く使い分けていきたいなと思いました！
もっとこんな使い方あるよ、ちょっと理解間違ってるよとかがあればぜひ教えてください！
今度はプロンプト管理についても書きたいなと思ってます。
Discussion