BrainPad Advent Calendar 2025 8日目の記事になります
--
株式会社ブレインパッド プロダクトユニットの津久井です。
弊社は「データ活用の促進を通じて持続可能な未来をつくる」をミッションに、データ分析支援やSaaSプロダクトの提供を通じて、企業の「データ活用の日常化」を推進しております。
現在私は、AI SaaSのエージェントリンリーの開発リードを担当しています。
Langfuseのインフラを立てたくない侍と申す
(若干語弊がありますが、) 以前記事にも書いたLangfuseですが、
LLMのトレースには Langfuse webのエンドポイントに対してアプリケーション側から直接トレースを送信するのが基本です。
しかし、ネットワーク環境の都合・管理リソースを増やしたくない...等で
Langfuseエンドポイントを常時起動したくないという想いもあり、
トレースを一旦ファイルに書き出し、後からローカルの Langfuse インスタンスにインポートする(オレオレ過ぎない)方法を実現してみました。
実現手順
1. LiteLLMのコールバック先をOpenTelemetryに設定
以前の記事同様、今回もLiteLLMを使います。
LiteLLMはコールバックが非常に簡潔に記載可能で、かつOpenTelemetry (以下、OTEL) のエンドポイントにも対応しています。
具体的には、LiteLLMの実行環境でOTELのエンドポイントを環境変数で設定 + コールバックを"otel"に設定するだけで、LLMへの問い合わせ時に自動的にトレースが送信されます。
OTELのエンドポイントを環境変数で設定
...
# Collector への エンドポイント
- OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318
# のプロトコル (HTTP/Protobuf)
- OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
...
コールバックを"otel"に設定
import os
import litellm
# Litellm の OpenTelemetry コールバックを有効化します
litellm.callbacks = ["otel"]
# このサンプルは、Gemini 2.5 Flash 呼び出します。
# プロジェクトIDを環境変数から取得（設定されていない場合はデフォルト値を使用）
vertex_project = os.getenv("GOOGLE_CLOUD_PROJECT", "xxxxxxx")
response = litellm.completion(
model="vertex_ai/gemini-2.5-flash",
messages=[{"role": "user", "content": "東京は日本のどの地方にありますか？"}],
vertex_project=vertex_project,
)
2. OpenTelemetry Collectorで、受け取ったトレースログをテキスト出力
受け取ったトレースをファイルに出力します。OpenTelemetry Collector には様々なエクスポーターが用意されており、その中の file exporter を使うと、トレースデータを行単位のJSONL (OTLP フォーマット) として保存できます。
(以下、Collector の設定ファイル例)
receivers:
otlp:
protocols:
http:
endpoint: 0.0.0.0:4318
processors:
batch: {}
exporters:
file:
path: "./traces.jsonl" # 出力先のファイル
service:
pipelines:
traces:
receivers: [otlp]
processors: [batch]
exporters: [file]
このCollectorを起動すると、アプリから送られてきたトレースが traces.jsonl に 1 行ずつ記録されます。
ファイルは JSON Lines 形式なので各行が 1 つの TracesData オブジェクトになっており、Langfuse エンドポイントでもそのまま受け取れる形式です。
{"resourceSpans":[{"resource":{"attributes":[{"key":"telemetry.sdk.language","value":{"stringValue":"python"}},{ ...
たとえばログを保存するディレクトリを gcsfuse でマウントし、Sidecar等でOTELエンドポイントを起動しておけば、そのまま GCS に保存することも可能です。
( Cloud Runなどの場合は、GCSへの保存完了までコンテナ起動を維持する必要あり )
3. トレースデータを Langfuse にインポート
保存した traces.jsonl をローカルの Langfuse にインポートします。(GCS等に保存したファイルを、ダウンロードしておく)
Langfuse は Basic 認証つきの OTLP/HTTP エンドポイントを持っており、/api/public/otel/v1/traces に POST するとトレースを取り込んでくれます。(OTLP JSON Lines ファイルの各行を、そのまま HTTP ボディに投げればOK)
以下の Python スクリプトでは ファイルを 1 行ずつ読み込み、Langfuse に POST しています。
import base64
import requests
# Langfuse API キー
public_key = "pk-lf-XXXXXXXXXXX"
secret_key = "sk-lf-YYYYYYYYYYY"
auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
# ヘッダ設定
headers = {
"Authorization": f"Basic {auth}",
"Content-Type": "application/json",
}
# ローカル Langfuse のエンドポイント
url = "http://localhost:3000/api/public/otel/v1/traces"
with open("./traces.jsonl", "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
res = requests.post(url, headers=headers, data=line)
if res.status_code != 200:
print("Error", res.status_code, res.text)
このスクリプトを実行すると、ファイルのトレースが Langfuse に一気に取り込まれ、Web UI で閲覧できるようになります。
注意
インポート重複排除等はもちろん無いので、
jsonlファイルの持ち方やインポート指針は ある程度工夫が必要です。(ローカルの場合は、一度環境を捨ててインポートし直す、もアリかもしれません)
余談
OpenTelemetry Collector には File Log Receiver も存在するので、おそらくhttp送信スクリプトを書かなくともOTELのエコシステムでjsonから送信は可能なのではと考えています。
(まだ試していないので、もしうまくいったらまた記載します)