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

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

Langfuseを構成して、OpenTelemetry ベースのトレース範囲をDatabricks MLflow OTLP エンドポイントに送信し、トレースが他のMLflowトレースと共にUnity Catalogテーブルに保存されるようにします。

Databricks でトレースを統合すると、次の利点が得られます。

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

要件​

  • 「OpenTelemetry on Databricks」プレビューが有効になっている Databricks ワークスペース。「Databricks プレビューの管理」を参照してください。
  • 新しいMLflowエクスペリメント。

ステップ 1: パッケージをインストールする​

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

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

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

Langfuse では、SDK を初期化するためにLANGFUSE_HOST 、 LANGFUSE_PUBLIC_KEY 、およびLANGFUSE_SECRET_KEY環境変数が必要です。トレースが Langfuse サーバーに送信されないようにするには、これらの変数をダミー値に設定します。「OTLP エクスポータの追加」で設定された OTEL エクスポータのみがスパンを受信します。

Python
import os

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

これは、Langfuse が独自のサーバー上でトレースを収集するのを防ぐ最も簡単な方法です。あるいは、 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カタログとスキーマを定義し、 set_experiment_trace_locationを使用してMLflow拡張機能にリンクします。 これにより、Databricks に着信トレースを保存する場所が指示されます。

Python
import mlflow
from mlflow.entities import UCSchemaLocation
from mlflow.tracing.enablement import set_experiment_trace_location

CATALOG_NAME = "<UC_CATALOG_NAME>"
SCHEMA_NAME = "<UC_SCHEMA_NAME>"
EXPERIMENT_ID = "<EXPERIMENT_ID>"

trace_location = set_experiment_trace_location(
    location=UCSchemaLocation(catalog_name=CATALOG_NAME, schema_name=SCHEMA_NAME),
    experiment_id=EXPERIMENT_ID,
)

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

Langfuse が使用するTracerProviderを取得し、Databricks OTLP エンドポイントを指すOTLPSpanExporterを持つBatchSpanProcessorをアタッチします。

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 OpenTelemetry TracerProvider.
langfuse = get_client()

# Retrieve the global TracerProvider.
# Use this to attach an additional span processor that forwards Langfuse spans to Databricks.
provider = otel_trace.get_tracer_provider()

# Configure the OTLP exporter to send spans to the Databricks MLflow endpoint.
# The X-Databricks-UC-Table-Name header routes incoming spans to the Unity Catalog table defined by the trace location.
databricks_exporter = OTLPSpanExporter(
    endpoint=f"{DATABRICKS_HOST}/api/2.0/otel/v1/traces",
    headers={
        "content-type": "application/x-protobuf",
        "Authorization": f"Bearer {DATABRICKS_TOKEN}",
        "X-Databricks-UC-Table-Name": trace_location.full_otel_spans_table_name,
    },
)

# Attach the Databricks exporter to Langfuse's provider. Because the Langfuse
# env vars are set to dummy values, this processor is the only active exporter,
# so all spans are sent 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] タブをクリックします。 my_pipeline呼び出しからのトレースが表示されます。

次のステップ​