評価ハーネス

mlflow.genai.evaluate()関数は、テスト データ (評価データセット) に対して GenAI アプリの品質を実行し、スコア付け子を適用することで、GenAI アプリの品質を体系的にテストします。

クイック リファレンス

動作方法

1. テスト入力でアプリを実行し、トレースをキャプチャする
2. スコアラーを適用して品質を評価し、フィードバックを作成します
3. 結果を評価実行に格納します

[前提条件]

1. MLflow と必要なパッケージをインストールする

   pip install --upgrade "mlflow[databricks]>=3.1.0" openai "databricks-connect>=16.1"
   

2. 環境のセットアップのクイックスタートに従って、MLflow 実験を作成します。

2 つの評価モード

モード 1: 直接評価 (推奨)

MLflow は GenAI アプリを直接呼び出してトレースを生成および評価します。 Python 関数 (predict_fn) にラップされたアプリケーションのエントリ ポイントを渡すか、アプリが Databricks Model Serving エンドポイントとしてデプロイされている場合は、 to_predict_fnでラップされたエンドポイントを渡すことができます。

メリット:

• オフライン評価と運用監視の間でスコアラーを簡単に再利用できます
• アプリの実行の自動並列化による評価の高速化

このモードでは、アプリを直接呼び出すことで、運用環境の 監視 でオフライン評価用に定義されたスコアラーを再利用できます。結果のトレースは同じになるためです。

手順 1: 評価を実行する

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

# Your GenAI app with MLflow tracing
@mlflow.trace
def my_chatbot_app(question: str) -> dict:
    # Your app logic here
    if "MLflow" in question:
        response = "MLflow is an open-source platform for managing ML and GenAI workflows."
    else:
        response = "I can help you with MLflow questions."

    return {"response": response}

# Evaluate your app
results = mlflow.genai.evaluate(
    data=[
        {"inputs": {"question": "What is MLflow?"}},
        {"inputs": {"question": "How do I get started?"}}
    ],
    predict_fn=my_chatbot_app,
    scorers=[RelevanceToQuery(), Safety()]
)

手順 2: UI で結果を表示する

モード 2: 回答シートの評価

GenAI アプリを直接実行できない場合は、事前に計算された出力または既存のトレースを評価用に提供します。

ユース ケース:

• 外部システムからの出力のテスト
• 歴史的痕跡の評価
• 異なるプラットフォーム間での出力の比較

(入力/出力あり) の例:

手順 1: 評価を実行する

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

# Pre-computed results from your GenAI app
results_data = [
    {
        "inputs": {"question": "What is MLflow?"},
        "outputs": {"response": "MLflow is an open-source platform for managing machine learning workflows, including tracking experiments, packaging code, and deploying models."},
    },
    {
        "inputs": {"question": "How do I get started?"},
        "outputs": {"response": "To get started with MLflow, install it using 'pip install mlflow' and then run 'mlflow ui' to launch the web interface."},
    }
]

# Evaluate pre-computed outputs
evaluation = mlflow.genai.evaluate(
    data=results_data,
    scorers=[Safety(), RelevanceToQuery()]
)

手順 2: UI で結果を表示する

既存のトレースの例:

import mlflow

# Retrieve traces from production
traces = mlflow.search_traces(
    filter_string="trace.status = 'OK'",
)

# Evaluate problematic traces
evaluation = mlflow.genai.evaluate(
    data=traces,
    scorers=[Safety(), RelevanceToQuery()]
)

キー パラメーター

def mlflow.genai.evaluate(
    data: Union[pd.DataFrame, List[Dict], mlflow.genai.datasets.EvaluationDataset],
    scorers: list[mlflow.genai.scorers.Scorer],
    predict_fn: Optional[Callable[..., Any]] = None,
    model_id: Optional[str] = None,
) -> mlflow.models.evaluation.base.EvaluationResult:

data

次のいずれかの形式の評価データセットです。

• EvaluationDataset (推奨)
• ディクショナリ、Pandas DataFrame、または Spark DataFrame の一覧

データ引数が DataFrame またはディクショナリの一覧として指定されている場合は、次のスキーマに従う必要があります。 これは、 EvaluationDataset で使用されるスキーマと一致します。 各レコードの系列を追跡するだけでなく、スキーマ検証を適用するため、 EvaluationDataset を使用することをお勧めします。

predict_fn

GenAI アプリのエントリ ポイント (モード 1 のみ)。 必須：

• キーワード引数としてinputsのdataディクショナリからキーを受け入れる
• JSON シリアル化可能なディクショナリを返す
• MLflow トレースを使用してインストルメント化する
• 呼び出しごとに 1 つのトレースを正確に出力する

scorers

適用する品質メトリックの一覧。 次の情報を指定できます。

• 定義済みのスコアラー
• カスタム スコアラー

詳細については、 スコアラー を参照してください。

model_id

結果をアプリのバージョン ( "models:/my-app/1" など) にリンクするためのオプションのモデル識別子。 詳細については、「 バージョン追跡」 を参照してください。

データ形式

直接評価の場合 (モード 1)

回答シート評価用 (モード 2)

オプション A - 入力と出力を指定します。

オプション B - 既存のトレースを指定します。

一般的なデータ入力パターン

MLflow 評価データセットを使用して評価する (推奨)

MLflow 評価データセットでは、バージョン管理、系列追跡、Unity カタログの統合が提供され、実稼働対応の評価が可能になります。

import mlflow
from mlflow.genai.scorers import Correctness, Safety
from my_app import agent  # Your GenAI app with tracing

# Load versioned evaluation dataset
dataset = mlflow.genai.datasets.get_dataset("catalog.schema.eval_dataset_name")

# Run evaluation
results = mlflow.genai.evaluate(
    data=dataset,
    predict_fn=agent,
    scorers=[Correctness(), Safety()],
)

次の用途に使用します。

• バージョン管理と系列追跡を使用して評価データを取得する必要がある
• トレースを評価レコードに簡単に変換する

評価データセットの構築について、トレースからまたは一からデータセットを作成する手順を参照してください。

辞書の一覧を使用して評価する

正式な評価データセットを作成せずに簡単なプロトタイプ作成を行うには、簡単な辞書の一覧を使用します。

import mlflow
from mlflow.genai.scorers import Correctness, RelevanceToQuery
from my_app import agent  # Your GenAI app with tracing

# Define test data as a list of dictionaries
eval_data = [
    {
        "inputs": {"question": "What is MLflow?"},
        "expectations": {"expected_facts": ["open-source platform", "ML lifecycle management"]}
    },
    {
        "inputs": {"question": "How do I track experiments?"},
        "expectations": {"expected_facts": ["mlflow.start_run()", "log metrics", "log parameters"]}
    },
    {
        "inputs": {"question": "What are MLflow's main components?"},
        "expectations": {"expected_facts": ["Tracking", "Projects", "Models", "Registry"]}
    }
]

# Run evaluation
results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=agent,
    scorers=[Correctness(), RelevanceToQuery()],
)

次の用途に使用します。

• 迅速なプロトタイプ作成
• 小さなデータセット (< 100 例)
• アドホック開発テスト

運用環境の場合は、 MLflow 評価データセットに変換します。

Pandas DataFrame を使用して評価する

CSV ファイルまたは既存のデータ サイエンス ワークフローを使用する場合は、Pandas DataFrames を使用して評価します。

import mlflow
import pandas as pd
from mlflow.genai.scorers import Correctness, Safety
from my_app import agent  # Your GenAI app with tracing

# Create evaluation data as a Pandas DataFrame
eval_df = pd.DataFrame([
    {
        "inputs": {"question": "What is MLflow?"},
        "expectations": {"expected_response": "MLflow is an open-source platform for ML lifecycle management"}
    },
    {
        "inputs": {"question": "How do I log metrics?"},
        "expectations": {"expected_response": "Use mlflow.log_metric() to log metrics"}
    }
])

# Run evaluation
results = mlflow.genai.evaluate(
    data=eval_df,
    predict_fn=agent,
    scorers=[Correctness(), Safety()],
)

次の用途に使用します。

• 迅速なプロトタイプ作成
• 小さなデータセット (< 100 例)
• アドホック開発テスト

Spark DataFrame を使用して評価する

大規模な評価、またはデータが Delta Lake/Unity カタログに既にある場合は、Spark DataFrames を使用します。

import mlflow
from mlflow.genai.scorers import Safety, RelevanceToQuery
from my_app import agent  # Your GenAI app with tracing

# Load evaluation data from a Delta table in Unity Catalog
eval_df = spark.table("catalog.schema.evaluation_data")

# Or load from any Spark-compatible source
# eval_df = spark.read.parquet("path/to/evaluation/data")

# Run evaluation
results = mlflow.genai.evaluate(
    data=eval_df,
    predict_fn=agent,
    scorers=[Safety(), RelevanceToQuery()],
)

次の用途に使用します。

• Delta Lake または Unity カタログにデータが既に存在する
• 評価を実行する前に MLflow 評価データセット内のレコードをフィルター処理する必要がある場合

手記： DataFrame は 、評価データセット スキーマに準拠している必要があります。

一般的な predict_fn パターン

アプリを直接呼び出す

パラメーター名が評価データセット キーと一致する場合は、アプリを predict_fn として直接渡します。

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

# Your GenAI app that accepts 'question' as a parameter
@mlflow.trace
def my_chatbot_app(question: str) -> dict:
    # Your app logic here
    response = f"I can help you with: {question}"
    return {"response": response}

# Evaluation data with 'question' key matching the function parameter
eval_data = [
    {"inputs": {"question": "What is MLflow?"}},
    {"inputs": {"question": "How do I track experiments?"}}
]

# Pass your app directly since parameter names match
results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=my_chatbot_app,  # Direct reference, no wrapper needed
    scorers=[RelevanceToQuery(), Safety()]
)

次の用途に使用します。

• 評価データセットに一致するパラメーター名を持つアプリ inputs

アプリを呼び出し可能にパッケージ化する

評価データセットの inputsとは異なるパラメーター名またはデータ構造が必要な場合は、アプリをラップします。

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

# Your existing GenAI app with different parameter names
@mlflow.trace
def customer_support_bot(user_message: str, chat_history: list = None) -> dict:
    # Your app logic here
    context = f"History: {chat_history}" if chat_history else "New conversation"
    return {
        "bot_response": f"Helping with: {user_message}. {context}",
        "confidence": 0.95
    }

# Wrapper function to translate evaluation data to your app's interface
def evaluate_support_bot(question: str, history: str = None) -> dict:
    # Convert evaluation dataset format to your app's expected format
    chat_history = history.split("|") if history else []

    # Call your app with the translated parameters
    result = customer_support_bot(
        user_message=question,
        chat_history=chat_history
    )

    # Translate output to standard format if needed
    return {
        "response": result["bot_response"],
        "confidence_score": result["confidence"]
    }

# Evaluation data with different key names
eval_data = [
    {"inputs": {"question": "Reset password", "history": "logged in|forgot email"}},
    {"inputs": {"question": "Track my order"}}
]

# Use the wrapper function for evaluation
results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=evaluate_support_bot,  # Wrapper handles translation
    scorers=[RelevanceToQuery(), Safety()]
)

次の用途に使用します。

• アプリのパラメーターと評価データセット input キーのパラメーター名の不一致 ( user_input と questionなど)
• データ形式の変換 (文字列からリストへの変換、JSON 解析)

デプロイされたエンドポイントを評価する

Databricks Agent Framework または Model Serving エンドポイントの場合は、 to_predict_fn を使用して互換性のある予測関数を作成します。

import mlflow
from mlflow.genai.scorers import RelevanceToQuery

# Create predict function for your endpoint
predict_fn = mlflow.genai.to_predict_fn("endpoints:/my-chatbot-endpoint")

# Evaluate
results = mlflow.genai.evaluate(
    data=[{"inputs": {"question": "How does MLflow work?"}}],
    predict_fn=predict_fn,
    scorers=[RelevanceToQuery()]
)

益： 完全な監視のために、トレースが有効なエンドポイントからトレースを自動的に抽出します。

ログに記録されたモデルを評価する

ログに記録された MLflow モデルをラップして、評価の名前付きパラメーターとモデルの単一パラメーター インターフェイスの間で変換します。

ほとんどのログ記録モデル (PyFunc を使用するモデルや LangChain などのログフレーバー) では、1 つの入力パラメーター (PyFunc の model_inputs など) を受け取りますが、 predict_fn は評価データセット内のキーに対応する名前付きパラメーターを想定しています。

import mlflow
from mlflow.genai.scorers import Safety

# Make sure to load your logged model outside of the predict_fn so MLflow only loads it once!
model = mlflow.pyfunc.load_model("models:/chatbot/staging")

def evaluate_model(question: str) -> dict:
    return model.predict({"question": question})

results = mlflow.genai.evaluate(
    data=[{"inputs": {"question": "Tell me about MLflow"}}],
    predict_fn=evaluate_model,
    scorers=[Safety()]
)

次のステップ

• アプリを評価する - 最初の評価を実行するためのステップ バイ ステップ ガイド
• 評価データセットの構築 - 運用ログまたはスクラッチから構造化されたテスト データを作成する
• カスタム スコアラーを定義 する - 特定のユース ケースに合わせて調整されたメトリックを構築する