Databricksにおけるユーザーと環境のトレースと分析

明けましておめでとうございます。今年もよろしくお願いします。

こちらにトレースを分析するチュートリアルが追加されていたのでウォークスルーします。

チュートリアル：ユーザーと環境のトレース・分析

このチュートリアルでは、トレースにコンテキストを追加し、ユーザー、セッション、デプロイメントを追跡・分析する方法を説明します。

• シンプルなチャットアプリで mlflow.update_current_trace() を使い、トレースにカスタムメタデータやタグを追加します。
• mlflow.search_traces() を使って、関連するトレースを抽出し、ユーザー、セッション、環境、アプリバージョンごとの統計を計算します。

環境セットアップ

必要なパッケージをインストールします：

• mlflow[databricks]: 最新バージョンのMLflowを使用して、より多くの機能と改善を利用します。
• openai: このアプリはOpenAI APIクライアントを使ってDatabricksホストのモデルを呼び出します。

%pip install -qq --upgrade "mlflow[databricks]>=3.1.0" openai
dbutils.library.restartPython()

MLflowのエクスペリメントを作成します。Databricksノートブックを使用している場合、このステップはスキップしてデフォルトのノートブックエクスペリメントを利用できます。それ以外の場合は、環境セットアップクイックスタート（AWS | Azure | GCP）に従い、エクスペリメントを作成しMLflow Trackingサーバーに接続してください。

アプリケーションの定義とトレース

以下のシンプルなチャットアプリケーションは、Databricksホストのファウンデーションモデル（AWS | Azure | GCP）を呼び出してユーザーの質問に回答します。

トレースは以下で行います：

• mlflow.openai.autolog() でOpenAIクライアント呼び出しを自動記録
• @mlflow.trace で my_app() のアプリケーションロジックをトレース
• mlflow.update_current_trace() で my_app() 内のトレースにコンテキストを追加：
  • ユーザー・セッションコンテキスト：ユーザーIDなどの情報を引数として渡します。
  • デプロイメントコンテキスト：環境やアプリバージョンなどの情報を環境変数で渡します。

MLflowは一部のメタデータを自動でトレースに記録しますが、デフォルト値の上書きやカスタムメタデータの追加も可能です。下記の例で両方を示します。

import mlflow
import os
from databricks.sdk import WorkspaceClient

mlflow.openai.autolog()

@mlflow.trace
def my_app(user_id: str, session_id: str, message: str) -> str:
    """チャットメッセージを処理し、トレースに追加情報を記録します。"""

    # 現在のトレースにユーザーとセッションのコンテキストを追加
    # @mlflow.traceデコレータによりアクティブなトレースが保証されます
    mlflow.update_current_trace(
        metadata={
            "mlflow.trace.user": user_id,
            "mlflow.trace.session": session_id,
        },
        tags={
          "query_category": "chat",  # カスタムタグの例
        },
    )

    app_environment = os.getenv("APP_ENVIRONMENT", "development")
    mlflow.update_current_trace(
        metadata={
            # 自動記録されるメタデータの上書き
            "mlflow.source.type": app_environment,  # デフォルトのLOCAL/NOTEBOOKを上書き
            # カスタムメタデータの追加
            "app_version": os.getenv("APP_VERSION", "1.0.0"),
            "deployment_id": os.getenv("DEPLOYMENT_ID", "unknown"),
        }
    )

    # トレースは実行時間、入力、出力、エラーも記録します
    # ここにチャットロジックを記述
    response = chat_completion(message)
    return response

# シンプルなチャットロジック

def chat_completion(message: str) -> str:
    # DatabricksホストのLLMに接続するOpenAIクライアントを作成
    w = WorkspaceClient()
    client = w.serving_endpoints.get_open_ai_client()

    response = client.chat.completions.create(
        model="databricks-claude-sonnet-4",
        messages=[
            {
                "role": "system",
                "content": "あなたは親切なアシスタントです。1～2文で簡潔に回答してください。",
            },
            {
                "role": "user",
                "content": message,
            },
        ]
    )
    return response.choices[0].message.content

上記のアプリケーションロジックは、ユーザーやセッションなどのメタデータを関数引数として受け取ります。実際の運用アプリケーションでは、リクエストオブジェクトのヘッダーからメタデータを抽出する実装になる場合もあります。例えば、Databricksアプリとしてデプロイした場合（AWS | Azure | GCP）、アプリはHTTPヘッダーからメタデータを取得できます（AWS | Azure | GCP）。

次に、異なるユーザー・セッションをシミュレートし、それぞれで1回以上のチャットを行います。デプロイメント情報は環境変数で設定します。

# 環境変数を設定し、トレースにデプロイメント固有のメタデータを記録します。
os.environ["APP_ENVIRONMENT"] = "staging"
os.environ["APP_VERSION"] = "1.0.0"
os.environ["DEPLOYMENT_ID"] = "deployment-123"

# ユーザーとセッションのコンテキストでチャットを実行し、サンプルトレースを生成
for session in range(2):
  # このユーザーの各セッションで2回チャット
  result = my_app(
      user_id="user-123",
      session_id=f"session-abc-{session}",
      message="MLflowとは何ですか？GenAIでどのように役立ちますか？"
  )
  result = my_app(
      user_id="user-123",
      session_id=f"session-abc-{session}",
      message="機械学習とAIの違いは何ですか？"
  )

os.environ["APP_VERSION"] = "1.1.0"
os.environ["DEPLOYMENT_ID"] = "deployment-456"

for session in range(2):
  # このユーザーの各セッションで1回チャット
  result = my_app(
      user_id="user-456",
      session_id=f"session-def-{session}",
      message="MLflowとは何ですか？機械学習でどのように役立ちますか？"
  )

複数のトレースが記録されます。

ノートブックエクスペリメントからもトレースを確認できます。

トレースの検索

以下の分析はすべて mlflow.search_traces() を使って関連するトレースを収集し、分析します：

import mlflow
# トレースを検索します
traces = mlflow.search_traces()
traces

各トレースには、アプリで記録した追加のコンテキスト（ユーザーIDなど）が付与されています：

first_trace = traces.iloc[0]
first_trace.trace_metadata['mlflow.trace.user']  # ユーザー

'user-456'

ユーザー行動の分析

まず、特定のユーザーの行動を分析します。

import pandas as pd
import time

def analyze_user_behavior(user_id: str, days: int = 7):
    """特定ユーザーのアクティビティパターンを分析します。"""

    cutoff_ms = int((time.time() - days * 86400) * 1000)

    traces = mlflow.search_traces(
        filter_string=f"metadata.`mlflow.trace.user` = '{user_id}' AND "
                      f"trace.timestamp_ms > {cutoff_ms}",
        order_by=["trace.timestamp_ms DESC"],
    )

    if len(traces) == 0:
        print(f"ユーザー {user_id} のアクティビティは見つかりませんでした")
        return

    # 主要な指標を計算
    total_interactions = len(traces)
    unique_sessions = set(row.trace_metadata.get("mlflow.trace.session", "") for index, row in traces.iterrows())
    unique_sessions.discard("")

    print(f"ユーザー {user_id} のアクティビティレポート（過去{days}日間）")
    print("=" * 50)
    print(f"総インタラクション数: {total_interactions}")
    print(f"ユニークセッション数: {len(unique_sessions)}")

    # 日別アクティビティ
    traces['date'] = pd.to_datetime(traces['request_time'], unit='ms').dt.date
    daily_activity = traces.groupby('date').size()
    print(f"\n日別アクティビティ:")
    print(daily_activity.to_string())

    # クエリカテゴリ
    query_categories = traces['tags'].apply(lambda tags: tags.get('query_category'))
    unique_categories = set(query_categories.dropna())
    category_counts = query_categories.value_counts()
    print(f"\nクエリカテゴリ:")
    print(category_counts.to_string())

    # パフォーマンス統計
    print(f"\nパフォーマンス:")
    print(f"平均応答時間: {traces['execution_duration'].mean():.1f}ms")
    print(f"エラー率: {(traces['state'] == 'ERROR').mean() * 100:.1f}%")

    return traces

analyze_user_behavior(user_id="user-123")

ユーザー user-123 のアクティビティレポート（過去7日間）
==================================================
総インタラクション数: 4
ユニークセッション数: 2

日別アクティビティ:
date
2026-01-01    4

クエリカテゴリ:
tags
chat    4

パフォーマンス:
平均応答時間: 3655.2ms
エラー率: 0.0%

セッションフローの分析

ユーザーは複数ターンの会話を行う場合があります。セッションごとにターンを分析することで、ユーザー体験を可視化できます。下記ではトレースのタイムスタンプ順に会話の流れを並べています。

def analyze_session_flow(session_id: str):
    """セッション内の会話フローを分析します。"""

    # セッション内の全トレースを時系列順に取得
    session_traces = mlflow.search_traces(
        filter_string=f"metadata.`mlflow.trace.session` = '{session_id}'",
        order_by=["timestamp ASC"]
    )

    # 会話のタイムラインを作成
    conversation_turns = []
    for index, row in session_traces.iterrows():
        conversation_turns.append({
            "ターン": index + 1,
            "タイムスタンプ": int(row.request_time),
            "処理時間_ms": int(row.execution_duration),
            "ステータス": str(row.state),
            "応答": row.response,
        })

    return conversation_turns

analyze_session_flow(session_id="session-abc-0")

[{'ターン': 1,
  'タイムスタンプ': 1767240954404,
  '処理時間_ms': 4706,
  'ステータス': 'OK',
  '応答': 'MLflowは機械学習のライフサイクル管理を行うオープンソースプラットフォームで、実験追跡、モデル管理、デプロイメントなどの機能を提供します。GenAIにおいては、大規模言語モデルの学習実験の管理、プロンプトエンジニアリングの結果追跡、モデルのバージョン管理、そして本番環境への安全なデプロイメントを支援することで、AI開発の効率性と再現性を大幅に向上させます。'},
 {'ターン': 2,
  'タイムスタンプ': 1767240959786,
  '処理時間_ms': 3528,
  'ステータス': 'OK',
  '応答': 'AIは人間の知能を模倣する技術全般を指し、機械学習はその実現手法の一つです。機械学習はデータから自動的にパターンを学習してタスクを実行する手法で、現在のAIの主要な技術基盤となっています。'}]

環境・バージョンごとの分析

環境やアプリバージョンなどのデプロイメントメタデータも、ユーザーやセッションと同様に分析できます。デプロイメントごとの品質やレイテンシ、その他重要な指標の変化を追跡できます。

traces = mlflow.search_traces()

traces['app_version'] = traces['trace_metadata'].apply(lambda meta: meta.get('app_version'))
traces['user_id'] = traces['trace_metadata'].apply(lambda meta: meta.get('mlflow.trace.user'))
traces['app_environment'] = traces['trace_metadata'].apply(lambda meta: meta.get('mlflow.source.type'))

interactions_per_version = traces.groupby('app_version').size()
print(f"アプリバージョンごとのインタラクション数:")
print(interactions_per_version.to_string())

users_per_version = traces.groupby('app_version')['user_id'].nunique()
print(f"\nアプリバージョンごとのユニークユーザー数:")
print(users_per_version.to_string())

interactions_per_environment = traces.groupby('app_environment').size()
print(f"\nアプリ環境ごとのインタラクション数:")
print(interactions_per_environment.to_string())

アプリバージョンごとのインタラクション数:
app_version
1.0.0    4
1.1.0    2

アプリバージョンごとのユニークユーザー数:
app_version
1.0.0    1
1.1.0    1

アプリ環境ごとのインタラクション数:
app_environment
staging    6

次のステップ

• トレースにコンテキストを追加（AWS）- トレースにカスタムメタデータやタグを追加する方法を学びます。
• プログラムからトレースを検索（AWS）- mlflow.search_traces() の使い方を学びます。
• トレースを分析（AWS）- 他のトレース分析例を参照できます。

はじめてのDatabricks

はじめてのDatabricks

Databricks無料トライアル

Databricks無料トライアル