Skip to content
←

필사 모드: Langfuse SDK計測 — 何が自動で取れて何を手で入れるか

日本語
0%
정확도 0%
💡 왼쪽 원문을 읽으면서 오른쪽에 따라 써보세요. Tab 키로 힌트를 받을 수 있습니다.

はじめに — 計測は境界を引く作業

第一回ではデータモデルを見ました。trace一つがobservationのツリーを収め、traceレベルの属性が下へ伝播するという構造でした。今回はそのツリーを実際に作る側です。

計測で失敗が出る場所はいつも同じです。自動で取れる範囲を過大に見るか、過小に見るかです。過大に見れば本当に必要だった中間ステップが空になり、過小に見ればすでに取れているものを手で入れ直してツリーが二重になります。

構成と設定名は2026-08-15に公式ドキュメントで確認しました。Langfuseはバージョンによってアーキテクチャが変わるため、利用中のバージョンのドキュメントを再確認してください。本記事はサーバーv4とPython SDK v4を基準にしています。バージョンポリシーのドキュメントは、サーバーv4ではPython SDK v4とJS/TS SDK v5が正式提供だと示しています。

出発点 — 三つのキーとシングルトンのクライアント

入門ドキュメントがすべての統合に共通して要求する環境変数は三つです。

pip install langfuse

export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"

LANGFUSE_BASE_URLの既定値はEUリージョンです。ドキュメントは米国、日本、HIPAAの各リージョンに別のホストがあると明示しています。セルフホストならここに自分のインスタンスのアドレスを入れます。

クライアントはシングルトンです。Python SDK概要は、アプリケーションのどこからでもget_client()でアクセスできると説明します。設定をコードで直接渡したいときだけLangfuse()コンストラクタを使います。

方式その一 — デコレータ

もっとも少ないコードで始める方法です。計測ドキュメントは、observe()が関数の入力、出力、所要時間、エラーを自動で取ると説明します。

from langfuse import observe

@observe()
def my_data_processing_function(data, parameter):
    return {"processed_data": data, "status": "ok"}

@observe(name="llm-call", as_type="generation")
async def my_async_llm_call(prompt_text):
    return "LLM response"

引数はドキュメントに四つ挙がっています。nameでobservation名を変え、as_typeでspanかgenerationかを決め、capture_inputとcapture_outputで入出力の収集を切ります。既定値はどちらも真です。

capture_inputを切る場面は思ったより多く来ます。関数の引数に生のプロンプトや利用者の個人情報が入る経路がそれです。ただしこれは部分的な対策で、本格的なマスキングは第五回で扱います。

方式その二 — コンテキストマネージャ

関数単位ではなくブロック単位で切りたいときに使います。ドキュメントはstart_as_current_observation()を、有効なOpenTelemetryコンテキストを更新しながらobservationを作る基本の方法だと説明します。

from langfuse import get_client, propagate_attributes

langfuse = get_client()

with langfuse.start_as_current_observation(
    as_type="span",
    name="user-request-pipeline",
    input={"user_query": "Tell me a joke"},
) as root_span:
    with propagate_attributes(user_id="user_123", session_id="session_abc"):
        with langfuse.start_as_current_observation(
            as_type="generation",
            name="joke-generation",
            model="gpt-4o",
        ) as generation:
            generation.update(output="Why did the span cross the road?")

    root_span.update(output={"final_joke": "..."})

入れ子がそのままツリーになります。withブロックのインデントが、第一回で描いた親子関係と一対一で対応します。ここで出てくるmodel="gpt-4o"が重要です。generationとして作り、モデル名を与えて初めてコストの軸ができます。

現在有効なobservationを参照なしで更新する方法もあります。ドキュメントはupdate_current_span()とupdate_current_generation()をこの用途で案内しています。深い呼び出しスタックの下から値を足したいときに便利です。

方式その三 — コンテキストを動かさない手動生成

ドキュメントは、有効なコンテキストを変えずに直接制御したいときはstart_observation()を使うよう案内しています。

from langfuse import get_client

langfuse = get_client()

span = langfuse.start_observation(name="manual-span")
span.update(input="Data for side task")

child = span.start_observation(name="child-span", as_type="generation")
child.end()
span.end()

この方式が必要になるのは、非同期処理、バックグラウンドタスク、コールバックのように開始と終了が同じスコープにないコードです。その代わりend()を呼ぶ責任が完全に人の側に来ます。例外経路でend()が抜ければ、そのobservationは終わらないまま残ります。

trace全体の入出力をルートのobservationとは別に決めたいときはset_trace_io()またはset_current_trace_io()を使います。ルートが内部のパイプラインオブジェクトを扱っていても、trace一覧には利用者が実際に打った文が見えるようにするためです。

属性を下へ流す — propagate_attributes

第一回でtraceレベルの属性が下へ伝播すると述べました。その伝播をコードで有効にする場所がpropagate_attributes()です。

from langfuse import propagate_attributes

with propagate_attributes(
    user_id="user_123",
    session_id="session_abc",
    metadata={"experiment": "variant_a"},
    version="1.0",
    environment="staging",
    trace_name="user-workflow",
    as_baggage=True,
):
    run_pipeline()

ドキュメントに挙がっている引数が上のとおりです。as_baggage=TrueはHTTPヘッダを通じたサービス間の伝播を有効にします。ゲートウェイとワーカーが分かれた構成で一つの流れをつなぐには、この指定が必要です。

ここで実務上よく間違える点が一つあります。これらの値はブロックの中で作られるobservationに適用されます。ブロックの外ですでに作られたobservationには付きません。ですからpropagate_attributesはリクエストを受けた最も外側で開くべきです。

フレームワーク統合 — 自動で取れる範囲

入門ドキュメントが並べる統合経路は、OpenAI SDK、Vercel AI SDK、LangChain、ネイティブSDK、OpenTelemetry、そしてLlamaIndexやCrewAI、LiteLLM、AutoGen、Google ADKといったフレームワークです。インストールコマンドは統合ごとに異なります。

# Python: OpenAI SDK ラッパー
pip install langfuse

# Python: LangChain
pip install langfuse langchain-openai

# JS/TS: OpenAI ラッパー
npm install @langfuse/openai

# JS/TS: LangChain
npm install @langfuse/core @langfuse/langchain

# JS/TS: ネイティブSDK
npm install @langfuse/tracing @langfuse/otel @opentelemetry/sdk-node

OpenAI統合はインポートを変えるだけで終わります。

from langfuse.openai import openai

completion = openai.chat.completions.create(
    name="test-chat",
    model="gpt-4o",
    messages=[{"role": "user", "content": "1 + 1 = "}],
)

import OpenAI from "openai";
import { observeOpenAI } from "@langfuse/openai";

const openai = observeOpenAI(new OpenAI());

こうした統合が自動で取ってくれるのはモデル呼び出しそのものです。モデル名、プロンプト、応答、トークン使用量、遅延です。取ってくれないのは、その呼び出しの間にあるあなたのコードです。どの文書を何件取得したか、リランキングで順序がどう変わったか、どの分岐を通ったかは手で入れる必要があります。

境界を決める基準は単純です。障害の振り返りで「では、どの段階が問題だったのか」に答えなければならない段階は、すべてobservationであるべきです。

OpenTelemetryとの関係 — v4が変えたこと

Python SDK概要は、Langfuse SDKがOpenTelemetryの上に作られていると明示しています。そしてバージョンポリシーのドキュメントは、v3からv4への移行で従来のバッチ取り込みをOpenTelemetryに置き換えたと述べています。これは内部実装の細部ではなく取り込み経路そのものの変更なので、v3当時の資料を見ているとここでずれます。

OpenTelemetry統合ドキュメントが示す取り込み経路は次のとおりです。

  • エンドポイントのパスは/api/public/otelです。クラウドのリージョンごとにホストが異なります。
  • プロトコルはHTTP上のOTLPで、HTTP/JSONとHTTP/protobufに対応します。ドキュメントはgRPCをまだ未対応と明示しています。
  • 認証ヘッダが二つ必要です。
# 例: OTLPエクスポータに渡すヘッダ
OTEL_EXPORTER_OTLP_ENDPOINT="https://cloud.langfuse.com/api/public/otel"
OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic ${AUTH_STRING},x-langfuse-ingestion-version=4"

AUTH_STRINGは公開キーと秘密キーをbase64で符号化した値です。二つ目のヘッダであるx-langfuse-ingestion-versionは、v4のリアルタイム取り込みを有効にする値だとドキュメントが説明しています。

属性の対応規則もドキュメントにあります。model属性を持つスパンは自動的にgeneration型のobservationになります。GenAIセマンティック規約の側ではgen_ai.system、gen_ai.request.model、gen_ai.response.model、gen_ai.prompt、gen_ai.completion、gen_ai.usage.*、gen_ai.usage.costが対応づけられます。Langfuse独自の名前空間にはlangfuse.observation.type、langfuse.observation.model.name、langfuse.observation.usage_details、langfuse.trace.metadata.*があります。

この対応を知ると選択肢が一つ増えます。すでにOpenTelemetryで計測済みのサービスなら、Langfuse SDKを入れずにエクスポート先を追加するだけでトレースが届きます。ただしその場合、スパンが正しい属性を持つように自分で面倒を見る必要があります。

出さないものを絞り込む

すべてのスパンをLangfuseへ送る必要はありません。すでにOpenTelemetryを使っているアプリケーションなら、HTTPサーバーのスパンやデータベースのスパンまで一緒に流れ込みます。高度な利用のドキュメントは、should_export_spanで出すスパンを選ぶよう案内しています。

from langfuse import Langfuse

langfuse = Langfuse(should_export_span=lambda span: True)

ドキュメントは、組み合わせて使える組み込みの判定関数としてis_default_export_span、is_langfuse_span、is_genai_spanを紹介しています。また、以前使われていたblocked_instrumentation_scopes引数は非推奨であり、拒否規則はshould_export_spanで表現するよう明示しています。

短命なプロセスでデータが消える理由

第一回で見たとおり、Langfuseはデータをローカルでバッチにまとめ、バックグラウンドで送ります。リクエスト経路を塞がない代わりに、バッチを吐き出す前にプロセスが終われば、そのデータは消えます。

バッチ処理、サーバーレス関数、CLIスクリプトがすべてここに当たります。ドキュメントは、バッファにあるすべてのobservationとscore、メディアのメタデータを今すぐ送る動作としてflush()を、フラッシュに加えてバックグラウンドスレッドの終了まで待つ動作としてshutdown()を区別しています。

from langfuse import get_client

langfuse = get_client()

try:
    run_batch_job()
finally:
    langfuse.flush()      # 今すぐ送信
    langfuse.shutdown()   # バックグラウンドスレッドの終了まで待機

長時間動くサーバーでは気にする必要がありません。問題になるのはいつでも短命なプロセスです。

おわりに — 計測の順番

まとめると順番はこうなります。まずフレームワーク統合でモデル呼び出しを自動で取ります。次に最も外側でpropagate_attributesを開き、利用者とセッション、環境を付けます。そして障害の振り返りで名指しされそうな中間段階にだけ手動のspanを入れます。最後に短命な実行経路へflush()を付けます。

この順番を守ればツリーが二重にならず、後で答えるべき問いに必要な軸が残ります。次回は、こうして作られたデータが実際にどこへどんな形で保存されるのか、Langfuseがなぜ ClickHouse を使うのかを見ます。

試してみる

シリーズ

参考資料

현재 단락 (1/113)

第一回ではデータモデルを見ました。trace一つがobservationのツリーを収め、traceレベルの属性が下へ伝播するという構造でした。今回はそのツリーを実際に作る側です。

작성 글자: 0자원문 글자: 8,284자작성 단락: 0/113