Exporter les traces Langfuse vers Databricks

Configurez Langfuse pour envoyer des étendues OTel au endpoint OTLP Databricks. Les traces sont stockées dans les tables Unity Catalog avec vos autres traces MLflow, où vous pouvez les interroger et les comparer à l’aide de SQL, ou les afficher dans l’interface utilisateur MLflow.

La consolidation des traces sur Databricks vous permet de :

• Query et comparez les appels instrumentés par Langfuse ainsi que les traces provenant d’autres frameworks en un seul endroit.
• Utilisez Databricks SQL pour analyser les données de trace à l'échelle de la montée en charge.
• Appliquez la gouvernance Unity Catalog, telle que les contrôles d’accès et la traçabilité, à toutes vos traces.

Prérequis

• Un workspace compatible avec Unity Catalog.
• Autorisations pour créer des catalogues et des schémas dans Unity Catalog.
• Un warehouse Databricks SQL avec l’autorisation CAN USE pour afficher les traces.
• Un Workspace dans une région prise en charge. Voir Features with limited regional availability.
• Une expérience MLflow avec un emplacement de trace UC. Pour en savoir plus sur la configuration, voir Requirements.
• Les autorisations Unity Catalog suivantes sur le catalogue et le schéma utilisés pour stocker les traces :
  • USE_CATALOG et USE_SCHEMA sur le catalogue et le schéma.
  • MODIFY et SELECT sur les tables <table_prefix>_otel_*. Consultez l’octroi des autorisations.
  • CREATE TABLE sur le schéma, afin que l’étape de configuration puisse créer les tables de traces pour la nouvelle expérimentation.

Étape 1 : installer les packages

Installez les packages requis dans votre notebook Databricks :

Python

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

Étape 2 : Désactiver la collecte des traces Langfuse

Langfuse initialise son SDK à partir des variables d’environnement LANGFUSE_HOST, LANGFUSE_PUBLIC_KEY et LANGFUSE_SECRET_KEY. Définissez-les sur des valeurs factices pour que les étendues soient acheminées uniquement vers l'exportateur Databricks ajouté à l'étape 5.

Python

import os

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

remarque

Vous pouvez également définir LANGFUSE_TRACING_ENABLED=False pour désactiver la collecte de traces intégrée de Langfuse.

Étape 3 : Configurez la connexion Databricks

Récupérez l'URL de l'hôte de votre workspace et le jeton API. Dans un notebook Databricks, récupérez-les à partir du contexte du notebook :

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()

Étape 4 : Link une expérimentation à Unity Catalog

Associer un catalogue, un schéma et un préfixe de table Unity Catalog à une expérimentation MLflow. Ceci indique à Databricks où stocker les traces entrantes.

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>",
    ),
)

Pour obtenir tous les détails de configuration, consultez Créer une expérimentation avec un emplacement de trace Unity Catalog.

Étape 5 : ajouter l’exportateur OTLP Databricks

Récupérez le TracerProvider que Langfuse enregistre en tant que fournisseur OTel global, puis associez un BatchSpanProcessor qui pointe vers l'endpoint OTLP de Databricks. L'en-tête X-Databricks-UC-Table-Name achemine les plages entrantes vers la table Unity Catalog définie par l'emplacement de la trace.

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={
        "content-type": "application/x-protobuf",
        "Authorization": f"Bearer {DATABRICKS_TOKEN}",
        "X-Databricks-UC-Table-Name": 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))

Étape 6 : Exécutez une fonction tracée

Utilisez le décorateur @observe() de Langfuse pour instrumenter votre agent. Le décorateur crée des spans OTel que l’exportateur envoie à Databricks.

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!")

Étape 7 : Afficher les traces

Ouvrez l’expérimentation MLflow dans votre workspace Databricks et cliquez sur la tab Traces. La trace de l’appel my_pipeline apparaît.

Pour rechercher des traces Langfuse ingérées par les valeurs d’attribut de span OTel, consultez Rechercher des traces par attributs de span OTel.