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

GenAIの本番運用モニタリング

備考

ベータ版

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

Databricksの GenAI の本番運用モニタリングを使用すると、本番運用 GenAI アプリからのトレースに対してMLflow 3 スコアラーを自動的に実行して、品質を継続的に監視できます。

本番運用トラフィックのサンプルを自動的に評価するスコアラーをスケジュールできます。 スコアラーの評価結果は、評価されたトレースにフィードバックとして自動的に添付されます。

本番運用モニタリングには次の内容が含まれます。

  • 組み込みまたはカスタムのスコアラーを使用した自動品質評価。
  • サンプリング レートを設定できるため、カバレッジと計算コストのトレードオフを制御できます。
  • 一貫した評価を確保するために、開発と本番運用で同じスコアラーを使用します。
  • バックグラウンドで実行されるモニタリングによる継続的な品質評価。
注記

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

レガシー本番運用モニタリングの詳細については、 「本番運用モニタリングAPIリファレンス (レガシー)」を参照してください。

前提 条件​

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

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

デフォルトでは、各審査員は GenAI 品質評価を実行するために特別に調整された、Databricks がホストする LLM モデルを使用します。スコアラー定義でmodel引数を使用すると、代わりにDatabricksモデルサービング エンドポイントを使用するように判定モデルを変更できます。 モデルはdatabricks:/<databricks-serving-endpoint-name>形式で指定する必要があります。

Python
safety_scorer = Safety(model="databricks:/databricks-gpt-oss-20b").register(name="my_custom_safety_scorer")

ガイドラインベースの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))

事前定義されたスコアラーと同様に、代わりにDatabricksモデルサービング エンドポイントを使用するように審査員モデルを変更できます。

Python
english_scorer = Guidelines(
  name="english",
  guidelines=["The response must be in English"],
  model="databricks:/databricks-gpt-oss-20b",
).register(name="custom_is_english")

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

ガイドラインベースの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,
        },
        model="databricks:/databricks-gpt-oss-20b",  # optional
    )

    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, Guidelines, ScorerSamplingConfig, list_scorers

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

guidelines_scorer = Guidelines(
    name="english",
    guidelines=["Response must be in English"]
).register(name="english_check")
guidelines_scorer = guidelines_scorer.start(
    sampling_config=ScorerSamplingConfig(sample_rate=0.5), # Sample 50%
)

# List and manage all scorers
all_scorers = list_scorers()
for scorer in all_scorers:
    if scorer.sample_rate > 0:
        print(f"{scorer.name} is active")
    else:
        print(f"{scorer.name} is stopped")

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

スコアラーのライフサイクルはMLflowエクスペリメントを中心としています。 スコアラーのライフサイクルの状態は次のとおりです。

  1. 未登録 : スコアラー関数は定義されていますが、サーバーに認識されていません。
  2. Registered : スコアラーはアクティブなMLflowエクスペリメントに登録されています。 スコアラーを登録するには、 .register()を使用します。
  3. アクティブ : スコアラーはサンプル レート > 0 で実行されています。スコアラーを開始するには、 .start()を使用します。
  4. 停止 : スコアラーは登録されていますが、実行されていません (サンプル レート = 0)。得点者を止めるには、 .stop()を使用します。
  5. 削除済み : スコアラーはサーバーから削除され、 MLflowエクスペリメントとの関連がなくなりました。 スコアラーを削除するには、 .delete()を使用します。

すべてのライフサイクル操作は 不変 です。つまり、各操作によって元のスコアラーは変更されません。代わりに、新しいスコアラー インスタンスを返します。

基本的なスコアラーライフサイクル​

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

# Built-in scorer lifecycle
safety_scorer = Safety().register(name="safety_check")
safety_scorer = safety_scorer.start(
    sampling_config=ScorerSamplingConfig(sample_rate=1.0),
)
safety_scorer = safety_scorer.update(
    sampling_config=ScorerSamplingConfig(sample_rate=0.8),
)
safety_scorer = safety_scorer.stop()
safety_scorer.delete()

# Custom scorer lifecycle
@scorer
def response_length(outputs):
    return len(str(outputs.get("response", "")))

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

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

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

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
# Demonstrate immutability
original_scorer = Safety().register(name="safety")
original_scorer = original_scorer.start(
   sampling_config=ScorerSamplingConfig(sample_rate=0.3),
)

# Update returns new instance
updated_scorer = original_scorer.update(
    sampling_config=ScorerSamplingConfig(sample_rate=0.8),
)

# Original remains unchanged
print(f"Original: {original_scorer.sample_rate}")  # 0.3
print(f"Updated: {updated_scorer.sample_rate}")    # 0.8

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

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

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

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. モニタリングダッシュボードを使用して、品質の傾向を追跡します。

おすすめの方法​

スコアラーの状態管理​

  • sample_rateプロパティを使用して操作の前にスコアラーの状態を確認します。
  • 不変パターンを使用します。.start() 、 .update() 、 .stop()の結果を変数に割り当てます。
  • スコアラーのライフサイクルを理解します。.stop()登録を保持し、 .delete()スコアラーを完全に削除します。

メトリクス バックフィル​

  • 小さく始めましょう。ジョブの期間とリソースの使用量を見積もるには、まず短い時間範囲から始めます。
  • 適切なサンプル レートを使用します。高いサンプル レートを使用する場合のコストと時間の影響を考慮してください。

サンプリング戦略​

  • 安全性やセキュリティのチェックなどの重要なスコアラーには、 sample_rate=1.0使用します。

  • 複雑な LLM 審査員などの高価な採点者の場合は、より低いサンプル レート (0.05 ～ 0.2) を使用します。

  • 開発中の反復的な改善には、中程度のレート (0.3 ～ 0.5) を使用します。

  • 次の例に示すように、カバレッジとコストのバランスをとってください。

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

次のステップ​

リファレンスガイド​

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

に最終更新