メインコンテンツまでスキップ

本番運用のモニタリング

備考

ベータ版

この機能は ベータ版です。

概要​

MLflowによって、計測的に品質を監視するためにあなたの生成AIアプリケーションのトレースに対して自動的にスコアラーを実行することができます。任意の スコアラー (カスタム メトリクス と組み込み/カスタム LLM ジャッジを含む) をスケジュールして、本番運用トラフィックのサンプルを自動的に評価できます。

主な利点:

  • 手動介入なしの 自動の品質評価 。
  • カバレッジと計算コストのバランスをとるための 柔軟なサンプリング 。
  • 開発から同じスコアラーを使用した 一貫した評価 。
  • 定期的なバックグラウンド実行による 継続的なモニタリング 。
注記

MLflow 3 本番運用 モニタリングは、 MLflow 2 からログに記録されたトレースと互換性があります。

前提 条件​

品質モニタリングを設定する前に、次のことを確認してください。

  1. MLflow エクスペリメント : トレースがログに記録されているMLflowエクスペリメント。指定しない場合は、アクティブなエクスペリメントが使用されます。
  2. 計測可能になっている本番運用アプリケーション : 生成 AI アプリは、 MLflow Tracingを使用してトレースをログに記録する必要があります。 本番運用のトレーシングガイドを参照してください。
  3. 定義されたスコアラー : アプリケーションのトレース形式で動作するテスト済みの スコアラー 。
ヒント

開発時に本番運用アプリをmlflow.genai.evaluate()のpredict_fnとして使用していた場合は、スコアラーはすでに互換性がある可能性があります。

本番運用 モニタリングを始める​

このセクションには、さまざまな種類のスコアラーを作成する方法を示すコード例が含まれています。

注記

いつでも、最大20個のスコアラーをエクスペリメントに関連付けて、継続的な品質モニタリングを行うことができます。

定義済みスコアラーを使用する​

MLflow には、すぐに使用できるいくつかの 定義済みスコアラー が用意されています モニタリング。

Python
from mlflow.genai.scorers import Safety, ScorerSamplingConfig

# Register the scorer with a name and start monitoring
safety_scorer = Safety().register(name="my_safety_scorer")  # name must be unique to experiment
safety_scorer = safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.7))

ガイドラインベースのLLMスコアラーを使用する​

ガイドラインベースのLLMスコアラー は、合格/不合格の自然言語基準を使用して入力と出力を評価できます。

Python
from mlflow.genai.scorers import Guidelines

# Create and register the guidelines scorer
english_scorer = Guidelines(
  name="english",
  guidelines=["The response must be in English"]
).register(name="is_english")  # name must be unique to experiment

# Start monitoring with the specified sample rate
english_scorer = english_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.7))

プロンプトベースのスコアラーを使用する​

ガイドラインベースのLLMスコアラーよりも柔軟性を高めるために、カスタマイズ可能な選択カテゴリ(優れた/良い/悪いなど)とオプションの数値スコアリングを使用してマルチレベルの品質評価を可能にする プロンプトベースのスコアラー を使用できます。

Python
from mlflow.genai.scorers import scorer, ScorerSamplingConfig


@scorer
def formality(inputs, outputs, trace):
    # Must be imported inline within the scorer function body
    from mlflow.genai.judges.databricks import custom_prompt_judge
    from mlflow.entities.assessment import DEFAULT_FEEDBACK_NAME

    formality_prompt = """
    You will look at the response and determine the formality of the response.

    <request>{{request}}</request>
    <response>{{response}}</response>

    You must choose one of the following categories.

    [[formal]]: The response is very formal.
    [[semi_formal]]: The response is somewhat formal. The response is somewhat formal if the response mentions friendship, etc.
    [[not_formal]]: The response is not formal.
    """

    my_prompt_judge = custom_prompt_judge(
        name="formality",
        prompt_template=formality_prompt,
        numeric_values={
            "formal": 1,
            "semi_formal": 0.5,
            "not_formal": 0,
        },
    )

    result = my_prompt_judge(request=inputs, response=inputs)
    if hasattr(result, "name"):
        result.name = DEFAULT_FEEDBACK_NAME
    return result

# Register the custom scorer and start monitoring
formality_scorer = formality.register(name="my_formality_scorer")  # name must be unique to experiment
formality_scorer = formality_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.1))

カスタムスコアラー関数を使用する​

LLMベースのスコアリングを省略するオプションなど、最大限の柔軟性を得るために、モニタリング用のカスタムスコアラー機能を定義して使用できます。

カスタムスコアラーを定義するときは、関数シグネチャにインポートする必要がある型ヒントを使用しないでください。スコアラー関数本体がインポートする必要があるパッケージを使用する場合は、適切なシリアル化のために、これらのパッケージを関数内にインラインでインポートします。

一部のパッケージは、インラインインポートを必要とせずにデフォルトで使用できます。これには、 databricks-agents、 mlflow-skinny、 openai、 およびサーバレス環境バージョン 2 に含まれるすべてのパッケージが含まれます。

Python
from mlflow.genai.scorers import scorer, ScorerSamplingConfig


# Custom metric: Check if response mentions Databricks
@scorer
def mentions_databricks(outputs):
    """Check if the response mentions Databricks"""
    return "databricks" in str(outputs.get("response", "")).lower()

# Custom metric: Response length check
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(str(outputs.get("response", "")))

# Custom metric with multiple inputs
@scorer
def response_relevance_score(inputs, outputs):
    """Score relevance based on keyword matching"""
    query = str(inputs.get("query", "")).lower()
    response = str(outputs.get("response", "")).lower()

    # Simple keyword matching (replace with your logic)
    query_words = set(query.split())
    response_words = set(response.split())

    if not query_words:
        return 0.0

    overlap = len(query_words & response_words)
    return overlap / len(query_words)

# Register and start monitoring custom scorers
databricks_scorer = mentions_databricks.register(name="databricks_mentions")
databricks_scorer = databricks_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

length_scorer = response_length.register(name="response_length")
length_scorer = length_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

relevance_scorer = response_relevance_score.register(name="response_relevance_score")  # name must be unique to experiment
relevance_scorer = relevance_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

複数のスコアラー構成​

包括的なモニタリング設定のために、複数のスコアラーを個別に登録、開始できます。

Python
from mlflow.genai.scorers import Safety, RelevanceToQuery, ScorerSamplingConfig

# Configure multiple scorers for comprehensive monitoring
safety_scorer = Safety().register(name="safety_check")  # name must be unique within an MLflow experiment
safety_scorer = safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))  # Check all traces

relevance_scorer = RelevanceToQuery().register(name="relevance_check")
relevance_scorer = relevance_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))  # Sample 50%

length_scorer = response_length.register(name="length_analysis")
length_scorer = length_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.3))

スケジュールされたスコアラーを管理する​

現在の得点者を一覧表示する​

エクスペリメントに登録されているすべてのスコアラーを表示するには:

Python
from mlflow.genai.scorers import list_scorers

# List all registered scorers
scorers = list_scorers()
for scorer in scorers:
    print(f"Name: {scorer._server_name}")
    print(f"Sample rate: {scorer.sample_rate}")
    print(f"Filter: {scorer.filter_string}")
    print("---")

スコアラーを更新する​

既存のスコアラー構成を変更するには:

Python
from mlflow.genai.scorers import get_scorer

# Get existing scorer and update its configuration (immutable operation)
safety_scorer = get_scorer(name="safety_monitor")
updated_scorer = safety_scorer.update(sampling_config=ScorerSamplingConfig(sample_rate=0.8))  # Increased from 0.5

# Note: The original scorer remains unchanged; update() returns a new scorer instance
print(f"Original sample rate: {safety_scorer.sample_rate}")  # Original rate
print(f"Updated sample rate: {updated_scorer.sample_rate}")   # New rate

スコアラーの停止と削除​

監視を停止するか、スコアラーを完全に削除するには:

Python
from mlflow.genai.scorers import get_scorer, delete_scorer

# Get existing scorer
databricks_scorer = get_scorer(name="databricks_mentions")

# Stop monitoring (sets sample_rate to 0, keeps scorer registered)
stopped_scorer = databricks_scorer.stop()
print(f"Sample rate after stop: {stopped_scorer.sample_rate}")  # 0

# Remove scorer entirely from the server
delete_scorer(name=databricks_scorer.name)

# Or restart monitoring from a stopped scorer
restarted_scorer = stopped_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

履歴トレースの評価 (メトリクス バックフィル)​

新しいメトリクスまたは更新されたメトリクスを履歴トレースに遡及的に適用できます。

現在のサンプルレートを使用した基本的なメトリクスのバックフィル​

Python
from databricks.agents.scorers import backfill_scorers

safety_scorer = Safety()
safety_scorer.register(name="safety_check")
safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

#custom scorer
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(outputs)

response_length.register(name="response_length")
response_length.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

# Use existing sample rates for specified scorers
job_id = backfill_scorers(
    scorers=["safety_check", "response_length"]
)

カスタムサンプルレートと時間範囲を使用したメトリクスバックフィル​

Python
from databricks.agents.scorers import backfill_scorers, BackfillScorerConfig
from datetime import datetime
from mlflow.genai.scorers import Safety, Correctness

safety_scorer = Safety()
safety_scorer.register(name="safety_check")
safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

#custom scorer
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(outputs)

response_length.register(name="response_length")
response_length.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

# Define custom sample rates for backfill
custom_scorers = [
    BackfillScorerConfig(scorer=safety_scorer, sample_rate=0.8),
    BackfillScorerConfig(scorer=response_length, sample_rate=0.9)
]

job_id = backfill_scorers(
    experiment_id=YOUR_EXPERIMENT_ID,
    scorers=custom_scorers,
    start_time=datetime(2024, 6, 1),
    end_time=datetime(2024, 6, 30)
)

最近のデータのバックフィル​

Python
from datetime import datetime, timedelta

# Backfill last week's data with higher sample rates
one_week_ago = datetime.now() - timedelta(days=7)

job_id = backfill_scorers(
    scorers=[
        BackfillScorerConfig(scorer=safety_scorer, sample_rate=0.8),
        BackfillScorerConfig(scorer=response_length, sample_rate=0.9)
    ],
    start_time=one_week_ago
)

結果を見る​

スコアラーをスケジュールした後、初期処理に15〜20分かかります。そうしたら：

  1. MLflowエクスペリメントに移動します。
  2. トレース タブを開いて、トレースに添付された評価を確認します。
  3. モニタリングダッシュボードを使用して、品質の傾向を追跡します。

おすすめの方法​

サンプリング戦略​

次の例に示すように、カバレッジとコストのバランスをとります。

Python
# High-priority scorers: higher sampling
safety_scorer = Safety().register(name="safety")
safety_scorer = safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))  # 100% coverage for critical safety

# Expensive scorers: lower sampling
complex_scorer = ComplexCustomScorer().register(name="complex_analysis")
complex_scorer = complex_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.05))  # 5% for expensive operations

カスタムスコアラーのデザイン​

次の例に示すように、カスタムスコアラーを自己完結型のままにします。

Python
@scorer
def well_designed_scorer(inputs, outputs):
    # ✅ All imports inside the function
    import re
    import json

    # ✅ Handle missing data gracefully
    response = outputs.get("response", "")
    if not response:
        return 0.0

    # ✅ Return consistent types
    return float(len(response) > 100)

トラブルシューティング​

スコアラーが実行されない​

スコアラーが実行されていない場合は、次の点を確認してください。

  1. エクスペリメントのチェック : トレースが個々の実行ではなくエクスペリメントに記録されていることを確認します。
  2. サンプリングレート :サンプルレートが低いと、結果が表示されるまでに時間がかかる場合があります。

シリアル化の問題​

カスタムスコアラーを作成するときは、関数定義にインポートを含めます。

Python
# ❌ Avoid external dependencies
import external_library  # Outside function

@scorer
def bad_scorer(outputs):
    return external_library.process(outputs)

# ✅ Include imports in the function definition
@scorer
def good_scorer(outputs):
    import json  # Inside function
    return len(json.dumps(outputs))

# ❌ Avoid using type hints in scorer function signature that requires imports
from typing import List

@scorer
def scorer_with_bad_types(outputs: List[str]):
    return False

メトリクスのバックフィルの問題​

「エクスペリメントに予定されたスコアラー 'X' が見つかりません」

  • スコアラー名がエクスペリメントに登録されているスコアラーと一致していることを確認してください
  • list_scorers方法を使用して利用可能なスコアラーを確認する

アーカイブトレース​

トレースとそれに関連する評価を Unity Catalog Delta テーブルに保存して、長期保存と高度な分析を行うことができます。 これは、カスタムダッシュボードの構築、詳細な実行に役立ちます アナリティクス トレースデータに対して、アプリケーションの動作の永続的な記録を維持します。

注記

指定した Unity Catalog Delta テーブルに書き込むために必要なアクセス許可が必要です。ターゲットテーブルがまだ存在しない場合は、作成されます。

テーブルが既に存在する場合は、トレースが追加されます。

アーカイブトレースの有効化​

エクスペリメントのトレースのアーカイブを開始するには、 enable_databricks_trace_archival 関数を使用します。 ターゲット Delta テーブルのフルネーム (カタログやスキーマなど) を指定する必要があります。experiment_idを指定しない場合、現在アクティブなエクスペリメントに対してトレースのアーカイブが有効になります。

Python
from mlflow.tracing.archival import enable_databricks_trace_archival

# Archive traces from a specific experiment to a Unity Catalog Delta table
enable_databricks_trace_archival(
    delta_table_fullname="my_catalog.my_schema.archived_traces",
    experiment_id="YOUR_EXPERIMENT_ID",
)

アーカイブトレースの無効化​

エクスペリメントのトレースのアーカイブは、 disable_databricks_trace_archival 機能を使用していつでも停止できます。

Python
from mlflow.tracing.archival import disable_databricks_trace_archival

# Stop archiving traces for the specified experiment
disable_databricks_trace_archival(experiment_id="YOUR_EXPERIMENT_ID")

次のステップ​

次のチュートリアルでジャーニーを続けてください。

リファレンスガイド​

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

に最終更新