トレースによる本番運用の可観測性

MLflow Tracing は、 Databricks 外部にデプロイされた本番運用GenAIアプリを、実行の詳細をキャプチャして Databricks ワークスペースに送信し、 MLflow UIで表示できるようにすることで、包括的なオブザーバビリティを提供します。

本番運用トレースの仕組み:

1. アプリがトレースを生成する - API 呼び出しごとにトレース データが作成されます
2. トレースは Databricks MLflow 追跡サーバーに移動します - ワークスペースの資格情報を使用
3. MLflow UI で表示 - Databricks ワークスペースでトレースを分析する

このページでは、Databricks の外部にデプロイされたアプリのトレースについて説明します。アプリがDatabricksモデルサービングを使用してデプロイされている場合は、「Databricksモデルサービングを使用したトレース」を参照してください。

前提 条件

注記

本番運用トレースには MLflow 3が必要です。 MLflow 2.x は、本番運用トレースではサポートされていません。

必要なパッケージをインストールします。次の表では、オプションについて説明します。

トピック	mlflow-tracing	mlflow[databricks]
推奨されるユースケース	本番運用のデプロイメント	開発と実験
利点	最小限の依存関係で無駄のない迅速なデプロイを実現 大容量トレーシングに最適化されたパフォーマンス 本番運用 モニタリングのためのクライアント側のトレースに注力	完全な MLflow 実験機能セット (UI、LLM-as-a-judge、開発ツールなど) すべての開発ツールとユーティリティが含まれています

Python

## Install mlflow-tracing for production deployment tracing
%pip install --upgrade mlflow-tracing

## Install mlflow for experimentation and development
%pip install --upgrade "mlflow[databricks]>=3.1"

基本的なトレース設定

Databricks がトレースを収集できるように、Databricks ワークスペースに接続するようにアプリケーションのデプロイを構成します。

次の環境変数を設定します。

Bash

# Required: Set the Databricks workspace host and authentication token
export DATABRICKS_HOST="https://your-workspace.cloud.databricks.com"
export DATABRICKS_TOKEN="your-databricks-token"

# Required: Set MLflow Tracking URI to "databricks" to log to Databricks
export MLFLOW_TRACKING_URI=databricks

# Required: Configure the experiment name for organizing traces (must be a workspace path)
export MLFLOW_EXPERIMENT_NAME="/Shared/production-genai-app"

デプロイ例

環境変数を設定したら、それらをアプリケーションに渡します。タブをクリックして、接続の詳細をさまざまなフレームワークに渡す方法を確認します。

Docker

Docker デプロイの場合は、コンテナ設定を通じて環境変数を渡します。

Dockerfile

# Dockerfile
FROM python:3.9-slim

# Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy application code
COPY . /app
WORKDIR /app

# Set default environment variables (can be overridden at runtime)
ENV DATABRICKS_HOST=""
ENV DATABRICKS_TOKEN=""
ENV MLFLOW_TRACKING_URI=databricks
ENV MLFLOW_EXPERIMENT_NAME="/Shared/production-genai-app"

CMD ["python", "app.py"]

環境変数を使用してコンテナを実行します。

Bash

docker run -d \
  -e DATABRICKS_HOST="https://your-workspace.cloud.databricks.com" \
  -e DATABRICKS_TOKEN="your-databricks-token" \
  -e MLFLOW_TRACKING_URI=databricks \
  -e MLFLOW_EXPERIMENT_NAME="/Shared/production-genai-app" \
  -e APP_VERSION="1.0.0" \
  your-app:latest

Kubernetes

Kubernetes デプロイの場合は、ConfigMaps と Secrets を使用して環境変数を渡します。

YAML

# configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: databricks-config
data:
  DATABRICKS_HOST: 'https://your-workspace.cloud.databricks.com'
  MLFLOW_TRACKING_URI: databricks
  MLFLOW_EXPERIMENT_NAME: '/Shared/production-genai-app'

---
# secret.yaml
apiVersion: v1
kind: Secret
metadata:
  name: databricks-secrets
type: Opaque
stringData:
  DATABRICKS_TOKEN: 'your-databricks-token'

---
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: genai-app
spec:
  template:
    spec:
      containers:
        - name: app
          image: your-app:latest
          envFrom:
            - configMapRef:
                name: databricks-config
            - secretRef:
                name: databricks-secrets
          env:
            - name: APP_VERSION
              value: '1.0.0'

トレース収集の確認

アプリをデプロイした後、トレースが適切に収集されていることを確認します。

Python

import mlflow
from mlflow.client import MlflowClient
import os

# Ensure MLflow is configured for Databricks
mlflow.set_tracking_uri("databricks")

# Check connection to MLflow server
client = MlflowClient()
try:
    # List recent experiments to verify connectivity
    experiments = client.search_experiments()
    print(f"Connected to MLflow. Found {len(experiments)} experiments.")

    # Check if traces are being logged
    traces = mlflow.search_traces(
        experiment_names=[os.getenv("MLFLOW_EXPERIMENT_NAME", "/Shared/production-genai-app")],
        max_results=5
    )
    print(f"Found {len(traces)} recent traces.")
except Exception as e:
    print(f"Error connecting to MLflow: {e}")
    print(f"Check your authentication and connectivity")

トレースにコンテキストを追加する

基本的なトレースが機能したら、デバッグと知見を向上させるためにコンテキストを追加します。 MLflow には、重要なコンテキスト情報をキャプチャするための次の標準化されたタグと属性があります。

• 要求の追跡 - エンドツーエンドのデバッグのためにトレースを特定の API 呼び出しにリンクします
• ユーザーセッション - ユーザージャーニーを理解するために、関連するインタラクションをグループ化します
• 環境データ - 各トレースを生成したデプロイ、バージョン、またはリージョンを追跡します
• ユーザーフィードバック - 品質評価を収集し、特定のインタラクションにリンクします

要求、セッション、およびユーザー コンテキストの追跡

本番運用アプリケーションは、デバッグ用のクライアントリクエストID、マルチターンカンバセーション用のセッションID、パーソナライゼーションと分析用のユーザーID、運用上の知見のための環境メタデータなど、複数のコンテキストを同時に追跡する必要があります。 以下は、FastAPIアプリケーションでこれらすべてを追跡する方法を示す包括的な例です。

Python

import mlflow
import os
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel

# Initialize FastAPI app
app = FastAPI()

class ChatRequest(BaseModel):
    message: str

@mlflow.trace # Ensure @mlflow.trace is the outermost decorator
@app.post("/chat") # FastAPI decorator should be inner
def handle_chat(request: Request, chat_request: ChatRequest):
    # Retrieve all context from request headers
    client_request_id = request.headers.get("X-Request-ID")
    session_id = request.headers.get("X-Session-ID")
    user_id = request.headers.get("X-User-ID")

    # Update the current trace with all context and environment metadata
    # The @mlflow.trace decorator ensures an active trace is available
    mlflow.update_current_trace(
        client_request_id=client_request_id,
        tags={
            # Session context - groups traces from multi-turn conversations
            "mlflow.trace.session": session_id,
            # User context - associates traces with specific users
            "mlflow.trace.user": user_id,
            # Environment metadata - tracks deployment context
            "environment": "production",
            "app_version": os.getenv("APP_VERSION", "1.0.0"),
            "deployment_id": os.getenv("DEPLOYMENT_ID", "unknown"),
            "region": os.getenv("REGION", "us-east-1")
        }
    )

    # --- Your application logic for processing the chat message ---
    # For example, calling a language model with context
    # response_text = my_llm_call(
    #     message=chat_request.message,
    #     session_id=session_id,
    #     user_id=user_id
    # )
    response_text = f"Processed message: '{chat_request.message}'"
    # --- End of application logic ---

    # Return response
    return {
        "response": response_text
    }

# To run this example (requires uvicorn and fastapi):
# uvicorn your_file_name:app --reload
#
# Example curl request with all context headers:
# curl -X POST "http://127.0.0.1:8000/chat" \
#      -H "Content-Type: application/json" \
#      -H "X-Request-ID: req-abc-123-xyz-789" \
#      -H "X-Session-ID: session-def-456-uvw-012" \
#      -H "X-User-ID: user-jane-doe-12345" \
#      -d '{"message": "What is my account balance?"}'

この組み合わせたアプローチには、いくつかの利点があります。

• クライアント要求 ID : システム全体の特定のクライアント要求とトレースを関連付けることで、エンドツーエンドのデバッグを可能にします
• セッションID (タグ: mlflow.trace.session):複数ターンの会話からのトレースをグループ化し、会話全体のフローを分析できるようにします
• ユーザーID (タグ: mlflow.trace.user):パーソナライゼーション、コホート分析、およびユーザー固有のデバッグのために、トレースを特定のユーザーに関連付けます
• 環境メタデータ : デプロイ コンテキスト (環境、バージョン、リージョン) を追跡し、さまざまなデプロイ間での運用の知見とデバッグを行います

トレースへのコンテキストの追加に関する詳細については、 ユーザーとセッションの追跡 および 環境とコンテキストの追跡に関するドキュメントを参照してください。

ユーザーからのフィードバックを収集する

特定のインタラクションに関するユーザーフィードバックを収集することは、品質を理解し、生成AIアプリケーションを改善するために不可欠です。この例では、前のセクションで示したクライアント要求 ID 追跡に基づいて、その ID を使用してフィードバックを特定のトレースにリンクする方法を示します。

以下は、FastAPIでフィードバック収集を実装する例です。

Python

import mlflow
from mlflow.client import MlflowClient
from fastapi import FastAPI, Query, Request
from pydantic import BaseModel
from typing import Optional
from mlflow.entities import AssessmentSource

# Initialize FastAPI app
app = FastAPI()

class FeedbackRequest(BaseModel):
    is_correct: bool  # True for correct, False for incorrect
    comment: Optional[str] = None

@app.post("/chat_feedback")
def handle_chat_feedback(
    request: Request,
    client_request_id: str = Query(..., description="The client request ID from the original chat request"),
    feedback: FeedbackRequest = ...
):
    """
    Collect user feedback for a specific chat interaction identified by client_request_id.
    """
    # Search for the trace with the matching client_request_id
    client = MlflowClient()
    # Get the experiment by name (using Databricks workspace path)
    experiment = client.get_experiment_by_name("/Shared/production-app")
    traces = client.search_traces(
        experiment_ids=[experiment.experiment_id],
        filter_string=f"attributes.client_request_id = '{client_request_id}'",
        max_results=1
    )

    if not traces:
        return {
            "status": "error",
            "message": f"Unable to find data for client request ID: {client_request_id}"
        }, 500

    # Log feedback using MLflow's log_feedback API
    mlflow.log_feedback(
        trace_id=traces[0].info.trace_id,
        name="response_is_correct",
        value=feedback.is_correct,
        source=AssessmentSource(
            source_type="HUMAN",
            source_id=request.headers.get("X-User-ID")
        ),
        rationale=feedback.comment
    )

    return {
        "status": "success",
        "message": "Feedback recorded successfully",
        "trace_id": traces[0].info.trace_id,
        "client_request_id": client_request_id,
        "feedback_by": request.headers.get("X-User-ID")
    }

# Example usage:
# After a chat interaction returns a response, the client can submit feedback:
#
# curl -X POST "http://127.0.0.1:8000/chat_feedback?client_request_id=req-abc-123-xyz-789" \
#      -H "Content-Type: application/json" \
#      -H "X-User-ID: user-jane-doe-12345" \
#      -d '{
#        "is_correct": true,
#        "comment": "The response was accurate and helpful"
#      }'

このフィードバック収集アプローチにより、次のことが可能になります。

• フィードバックを特定のインタラクションにリンク する: クライアント要求 ID を使用して正確なトレースを検索し、フィードバックを添付します
• 構造化されたフィードバックを格納する : log_feedback API は、MLflow UI に表示される適切な評価オブジェクトを作成します
• 品質パターンの分析 : トレースとそれに関連するフィードバックをクエリして、肯定的または否定的な評価を受けるインタラクションのタイプを特定します

後で MLflow UI またはプログラムを使用してフィードバックを使用してトレースのクエリを実行し、パターンを分析してアプリケーションを改善できます。

コンテキスト付きのクエリトレース

コンテキスト情報を使用して、本番運用の動作を分析します。

Python

import mlflow
from mlflow.client import MlflowClient
import pandas as pd

client = MlflowClient()
experiment = client.get_experiment_by_name("/Shared/production-app")

# Query traces by user
user_traces = client.search_traces(
    experiment_ids=[experiment.experiment_id],
    filter_string="tags.`mlflow.trace.user` = 'user-jane-doe-12345'",
    max_results=100
)

# Query traces by session
session_traces = client.search_traces(
    experiment_ids=[experiment.experiment_id],
    filter_string="tags.`mlflow.trace.session` = 'session-123'",
    max_results=100
)

次のステップ

これらの推奨アクションとチュートリアルで旅を続けてください。

• 本番運用トラフィック の自動品質評価を設定する

リファレンスガイド

このガイドで説明されている概念と機能の詳細なドキュメントをご覧ください。

• 本番運用 モニタリングのコンセプト - MLflow が継続的な品質モニタリングをどのように実現するかを理解する
• トレーシング・データ・モデル - トレース、スパン、属性について学習します
• ログ評価 - フィードバックの保存方法と使用方法を理解する