チュートリアル: ユーザーと環境の追跡と分析
このチュートリアルでは、ユーザー、セッション、デプロイメントを追跡および分析するために、トレースにコンテキストを追加する方法を説明します。
- シンプルなチャット アプリでは、
mlflow.update_current_trace()を使用して、カスタム メタデータとタグをトレースに追加します。 - トレースを分析するには、
mlflow.search_traces()を使用して、ユーザー、セッション、環境、アプリのバージョンに関する関連するトレースと統計情報を抽出します。
環境設定
必要なパッケージをインストールします。
mlflow[databricks]: より多くの機能と改善を得るには、最新バージョンの MLflow を使用してください。openai: このアプリは、OpenAI API クライアントを使用して、Databricks でホストされるモデルを呼び出します。
Python
%pip install -qq --upgrade "mlflow[databricks]>=3.1.0" openai
dbutils.library.restartPython()
MLflowエクスペリメントを作成します。 Databricksノートブックを使用している場合は、このステップをスキップして、デフォルトのノートブック体験を使用できます。 それ以外の場合は、環境設定のクイックスタートに従ってエクスペリメントを作成し、 MLflowトラッキング サーバーに接続します。
アプリケーションの定義と追跡
以下のシンプルなチャット アプリケーションは、Databricks がホストする基盤モデルを呼び出して、ユーザーのクエリに回答します。
トレースは以下を使用して行われます:
mlflow.openai.autolog()OpenAIクライアントの呼び出しを自動ログする@mlflow.traceアプリケーションロジックをトレースするにはmy_app()mlflow.update_current_trace()my_app()内のトレースにコンテキストを追加します:- ユーザーとセッションのコンテキスト: ユーザー ID などのクエリ固有の情報は、引数としてアプリケーション ロジックに渡すことができます。
- デプロイメント コンテキスト: 環境やアプリのバージョンなどのデプロイメント固有の情報は、環境変数を通じてアプリケーションに渡されるため、デプロイメント間での構成の変更が簡素化されます。
MLflow はトレース内の一部のメタデータを自動的に入力しますが、デフォルト値を上書きできます。カスタム メタデータを定義することもできます。以下の例は両方を示しています。
Python
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:
"""Process a chat message with extra content logging for traces."""
# Add user and session context to the current trace.
# The @mlflow.trace decorator ensures there is an active trace.
mlflow.update_current_trace(
metadata={
"mlflow.trace.user": user_id,
"mlflow.trace.session": session_id,
},
tags={
"query_category": "chat", # Example of a custom tag
},
)
app_environment = os.getenv("APP_ENVIRONMENT", "development")
mlflow.update_current_trace(
metadata={
# Override automatically populated metadata
"mlflow.source.type": app_environment, # Override default LOCAL/NOTEBOOK
# Add custom metadata
"app_version": os.getenv("APP_VERSION", "1.0.0"),
"deployment_id": os.getenv("DEPLOYMENT_ID", "unknown"),
}
)
# The trace will capture the execution time, inputs, outputs, and any errors
# Your chat logic here
response = chat_completion(message)
return response
# Basic chat logic
def chat_completion(message: str) -> str:
# Create an OpenAI client that is connected to Databricks-hosted LLMs
w = WorkspaceClient()
client = w.serving_endpoints.get_open_ai_client()
response = client.chat.completions.create(
model="databricks-claude-sonnet-4",
messages=[
{
"role": "system",
"content": "You are a helpful assistant. Give brief, 1-2 sentence responses.",
},
{
"role": "user",
"content": message,
},
]
)
return response.choices[0].message.content
上記のアプリケーション ロジックは、ユーザー、セッション、およびその他のメタデータを関数の引数として受け取ります。本番運用アプリケーションでは、実装によってリクエスト オブジェクトのヘッダーからメタデータが抽出される場合があります。 たとえば、アプリケーションがDatabricks Appとしてデプロイされている場合、アプリはメタデータを使用して HTTP ヘッダーにアクセスできます。
次に、それぞれ 1 つ以上のチャット インタラクションを含む、いくつかの異なるユーザーとセッションをシミュレートします。環境変数を使用してデプロイメント情報を設定します。
Python
# Set environment variables to log deployment-specific metadata with traces.
os.environ["APP_ENVIRONMENT"] = "staging"
os.environ["APP_VERSION"] = "1.0.0"
os.environ["DEPLOYMENT_ID"] = "deployment-123"
# Run the chat completion with user and session context to generate example traces:
for session in range(2):
# 2 chat interactions per session for this user
result = my_app(
user_id="user-123",
session_id=f"session-abc-{session}",
message="What is MLflow and how does it help with GenAI?"
)
result = my_app(
user_id="user-123",
session_id=f"session-abc-{session}",
message="What is ML vs. AI?"
)
os.environ["APP_VERSION"] = "1.1.0"
os.environ["DEPLOYMENT_ID"] = "deployment-456"
for session in range(2):
# 1 chat interaction per session for this user
result = my_app(
user_id="user-456",
session_id=f"session-def-{session}",
message="What is MLflow and how does it help with machine learning?"
)
痕跡を探す
以下の分析はすべて、 mlflow.search_traces()を使用して分析に関連するトレースを収集することに基づいています。
Python
import mlflow
traces = mlflow.search_traces()
traces
各トレースには、ユーザー ID など、アプリに記録された追加のコンテキストが注釈として付けられます。
Python
first_trace = traces.iloc[0]
first_trace.trace_metadata['mlflow.trace.user']
Output
'user-456'
ユーザーの行動を分析する
まず、特定のユーザーの行動を分析します。
Python
import pandas as pd
import time
def analyze_user_behavior(user_id: str, days: int = 7):
"""Analyze activity patterns for a specific user."""
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"No activity found for user {user_id}")
return
# Calculate key metrics
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 {user_id} Activity Report ({days} days)")
print("=" * 50)
print(f"Total interactions: {total_interactions}")
print(f"Unique sessions: {len(unique_sessions)}")
# Daily activity
traces['date'] = pd.to_datetime(traces['request_time'], unit='ms').dt.date
daily_activity = traces.groupby('date').size()
print(f"\nDaily activity:")
print(daily_activity.to_string())
# Query categories
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"\nQuery categories:")
print(category_counts.to_string())
# Performance stats
print(f"\nPerformance:")
print(f"Average response time: {traces['execution_duration'].mean():.1f}ms")
print(f"Error rate: {(traces['state'] == 'ERROR').mean() * 100:.1f}%")
return traces
Python
analyze_user_behavior(user_id="user-123")
Output
User user-123 Activity Report (7 days)
==================================================
Total interactions: 4
Unique sessions: 2
Daily activity:
date
2025-12-12 4
Query categories:
tags
chat 4
Performance:
Average response time: 2177.5ms
Error rate: 0.0%
セッションフローを分析する
ユーザーは、複数回の会話でアプリケーションとやり取りする場合があります。セッションをターンバイターンで分析すると、ユーザーのエクスペリエンスを説明するのに役立ちます。以下では、トレースのタイムスタンプを使用して会話の順番を並べています。
Python
def analyze_session_flow(session_id: str):
"""Analyze conversation flow within a session."""
# Get all traces from a session, ordered chronologically
session_traces = mlflow.search_traces(
filter_string=f"metadata.`mlflow.trace.session` = '{session_id}'",
order_by=["timestamp ASC"]
)
# Build a timeline of the conversation
conversation_turns = []
for index, row in session_traces.iterrows():
conversation_turns.append({
"turn": index + 1,
"timestamp": int(row.request_time),
"duration_ms": int(row.execution_duration),
"status": str(row.state),
"response": row.response,
})
return conversation_turns
Python
analyze_session_flow(session_id="session-abc-0")
Output
[{'turn': 1,
'timestamp': 1765560306051,
'duration_ms': 2570,
'status': 'OK',
'response': 'MLflow is an open-source platform for managing the machine learning lifecycle, including experiment tracking, model packaging, and deployment. For GenAI, it helps by providing tools to track experiments with large language models, manage model versions, log prompts and responses, and deploy AI models at scale while maintaining reproducibility and governance.'},
{'turn': 2,
'timestamp': 1765560308943,
'duration_ms': 2644,
'status': 'OK',
'response': 'AI (Artificial Intelligence) is the broader field focused on creating machines that can perform tasks requiring human-like intelligence, while ML (Machine Learning) is a subset of AI that specifically uses algorithms to learn patterns from data without being explicitly programmed for each task. Think of AI as the goal and ML as one of the main methods to achieve it.'}]
環境とバージョンを分析する
環境やアプリのバージョンなどのデプロイメント メタデータは、ユーザーやセッションと同様に分析できます。デプロイメントを分析すると、アプリケーションを反復処理するときに、品質、レイテンシ、その他の重要なメトリックの改善または低下を追跡するのに役立ちます。
Python
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"Interactions per app version:")
print(interactions_per_version.to_string())
users_per_version = traces.groupby('app_version')['user_id'].nunique()
print(f"\nDistinct users per app version:")
print(users_per_version.to_string())
interactions_per_environment = traces.groupby('app_environment').size()
print(f"\nInteractions per app environment:")
print(interactions_per_environment.to_string())
Output
Interactions per app version:
app_version
1.0.0 4
1.1.0 4
Distinct users per app version:
app_version
1.0.0 1
1.1.0 1
Interactions per app environment:
app_environment
staging 8
次のステップ
- トレースにコンテキストを追加する- トレースにカスタム メタデータとタグを追加する方法の詳細について説明します。
- プログラムでトレースを検索する- mlflow.search_traces()の使用について詳しく学習します。
- トレースの分析- トレース分析の他の例を参照してください。