メインコンテンツまでスキップ
に最終更新

Langfuse のトレースを Databricks にエクスポートする

Langfuseを構成して、Databricks OTLP EndpointにOTelスパンを送信します。トレースは他のMLflowトレースとともにUnity Catalogテーブルに保存され、SQLでクエリーして比較したり、MLflow UIで表示したりできます。

Databricks上でトレースを集約すると、次のことが可能です：

  • Langfuseで計測された呼び出しを、他のフレームワークからのトレースと一緒に1か所でクエリーして比較します。
  • Databricks SQL を使用して、トレースデータを大規模に分析します。
  • アクセス制御やリネージなど、Unity Catalogのガバナンスをすべてのトレースに適用します。

要件​

  • Unity Catalog が有効なワークスペース。
  • Unity Catalog でカタログとスキーマを作成する権限。
  • トレースを表示するための CAN USE 権限を持つ Databricks SQL warehouse。
  • サポートされているリージョンのワークスペース。リージョン限定機能を参照してください。
  • UC トレース保存場所を持つ MLflow エクスペリメント。セットアップの詳細については、要件を参照してください。
  • トレースの保存に使用されるカタログおよびスキーマに対する、次の Unity Catalog 権限:
    • USE_CATALOG また、カタログおよびスキーマにUSE_SCHEMAを設定します。
    • MODIFY および <table_prefix>_otel_* テーブル上の SELECT。「権限の付与」を参照してください。
    • CREATE TABLE スキーマ上で実行することで、セットアップステップで新しいエクスペリメント用のトレーステーブルを作成できるようにします。

ステップ 1: パッケージのインストール​

Databricksノートブックに必要なパッケージをインストールします:

Python
%pip install "langfuse>=3.14.5" "mlflow[databricks]>=3.14.0" opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
%restart_python

ステップ 2: Langfuse トレース収集を無効にする​

Langfuse は、LANGFUSE_HOST、LANGFUSE_PUBLIC_KEY、および LANGFUSE_SECRET_KEY の各環境変数から SDK を初期化します。スパンが ステップ 5 で追加された Databricks エクスポーターにのみ送信されるように、これらをダミー値に設定します。

Python
import os

os.environ["LANGFUSE_HOST"] = "localhost"
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
注記

または、LANGFUSE_TRACING_ENABLED=False を設定して Langfuse の組み込みトレース収集を無効にします。

ステップ 3: Databricks 接続を構成する​

ワークスペースのホストURLとAPIトークンを取得します。Databricksノートブックでは、ノートブックのコンテキストから取得します。

Python
# If running outside a Databricks notebook, set DATABRICKS_HOST and DATABRICKS_TOKEN environment variables manually.
context = dbutils.notebook.entry_point.getDbutils().notebook().getContext()
DATABRICKS_HOST = context.apiUrl().get().rstrip("/")
DATABRICKS_TOKEN = context.apiToken().get()

ステップ 4: エクスペリメントを Unity Catalog にリンクする​

Unity Catalog のカタログ、スキーマ、およびテーブル プレフィックスを MLflow エクスペリメントにバインドします。これにより、受信したトレースの保存先が Databricks に通知されます。

Python
import mlflow
from mlflow.entities.trace_location import UnityCatalog

experiment = mlflow.set_experiment(
    experiment_name="<MLFLOW_EXPERIMENT_NAME>",
    trace_location=UnityCatalog(
        catalog_name="<UC_CATALOG_NAME>",
        schema_name="<UC_SCHEMA_NAME>",
        table_prefix="<UC_TABLE_PREFIX>",
    ),
)

詳細なセットアップ手順については、Unity Catalog のトレース場所を指定したエクスペリメントの作成を参照してください。

ステップ 5: Databricks OTLP エクスポーターを追加する​

Langfuse がグローバル OTel プロバイダーとして登録する TracerProvider を取得し、Databricks OTLP Endpoint を指す BatchSpanProcessor をアタッチします。X-Databricks-UC-Table-Name ヘッダーは、トレースの場所によって定義された Unity Catalog テーブルに受信スパンをルーティングします。

Python
from langfuse import get_client
from opentelemetry import trace as otel_trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Initialize the Langfuse client. Langfuse registers its own TracerProvider as the global OTel TracerProvider.
langfuse = get_client()

# Retrieve the global TracerProvider to attach an additional span processor.
provider = otel_trace.get_tracer_provider()

databricks_exporter = OTLPSpanExporter(
    endpoint=f"{DATABRICKS_HOST}/api/2.0/otel/v1/traces",
    headers={
        &quot;content-type&quot;: &quot;application/x-protobuf&quot;,
        &quot;Authorization&quot;: f&quot;Bearer {DATABRICKS_TOKEN}&quot;,
        &quot;X-Databricks-UC-Table-Name&quot;: experiment.trace_location.full_otel_spans_table_name,
    },
)

# Because the Langfuse env vars are set to dummy values, this processor is the only
# active exporter, so all spans go exclusively to Databricks.
provider.add_span_processor(BatchSpanProcessor(databricks_exporter))

ステップ 6: トレースされた関数を実行する​

Langfuse の @observe() デコレーターを使用してエージェントを計測します。デコレーターは、エクスポーターが Databricks に送信する OTel スパンを作成します。

Python
from langfuse import observe

@observe()
def my_llm_call(prompt: str) -> str:
    # Replace with your LLM logic (OpenAI, Anthropic, etc.)
    return f"Response to: {prompt}"

@observe()
def my_pipeline(user_input: str) -> str:
    result = my_llm_call(user_input)
    return result

my_pipeline("Hello, world!")

ステップ 7: トレースの表示​

Databricks ワークスペースで MLflow エクスペリメントを開き、 [Traces] tab をクリックします。my_pipeline 呼び出しからのトレースが表示されます。

取り込まれた Langfuse トレースを OTel スパン属性値で検索するには、OTel スパン属性によるトレースの検索を参照してください。

このページの見出し