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

評価用ハーネス

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

クイックリファレンス​

詳細については、 mlflow.genai.evaluate()を参照してください。

パラメーター

Type

説明

data

MLflow EvaluationDataset、List[Dict]、Pandas DataFrame、Spark DataFrame

テストデータ

predict_fn

呼び出し

アプリ (直接評価のみ)

scorers

リスト[スコアラー]

Quality メトリクス

model_id

str

オプションのバージョン追跡

仕組み​

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

前提 条件​

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

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

  2. MLflow エクスペリメントを作成するには、環境のセットアップに関するクイックスタートに従ってください。

評価モード​

評価モードには、次の 2 つがあります。

  • (推奨) 直接評価。MLflow はアプリを直接呼び出して、評価用のトレースを生成します。
  • 解答用紙の評価。コンピュート前の出力または既存のトレースを評価用に提供します。

直接評価(推奨)​

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

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

評価 とトレースの連携

次のコードは、評価を実行する方法の例を示しています。

Python
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()]
)

その後、UI で結果を表示できます。

評価結果

解答用紙の評価​

生成 AI アプリを直接実行できない場合は、既存のトレースまたはコンピュート前の出力を評価用に提供できます。 ユースケースの例としては、外部システムからの出力のテスト、履歴トレースの評価、異なるプラットフォーム間での出力の比較などがあります。

important

本番運用環境と異なるトレースを持つ解答用紙を使用している場合、 本番運用 モニタリングに利用するためには、スコアラー関数の書き換えが必要になることがあります。

評価と解答用紙の連携

入力と出力の使用例​

次のコードは、評価を実行する方法の例を示しています。

Python
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()]
)

その後、UI で結果を表示できます。

評価結果

既存のトレースの使用例​

次のコードは、既存のトレースを使用して評価を実行する方法を示しています。

Python
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()]
)

主要なパラメーター​

Python
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 で使用されるスキーマと一致しています。Databricks では、各レコードのリネージを追跡することに加えて、スキーマの検証を強制する EvaluationDataset を使用することをお勧めします。

フィールド

データ型

説明

直接評価で使用

解答用紙と併用

inputs

dict[Any, Any]

を使用してpredict_fnに渡されるdict **kwargs。JSON シリアル化可能である必要があります。各キーは、 predict_fnの名前付き引数に対応する必要があります。

必須

inputs+outputsまたはtraceのいずれかが必要です。両方をパスすることはできません。 指定されていない場合は trace から派生します。

outputs

dict[Any, Any]

対応するinputの 生成AI アプリの出力を含むdict。JSON シリアル化可能である必要があります。

トレースから MLflow によって生成された、指定 しないでください 。

inputs+outputsまたはtraceのいずれかが必要です。両方をパスすることはできません。 指定されていない場合は trace から派生します。

expectations

dict[str, Any]

に対応するグラウンドトゥルースラベルを持つ dict input。scorersが品質をチェックするために使用されます。JSON シリアル化可能であり、各キーは strである必要があります。

オプション

オプション

trace

mlflow.entities.Trace

要求のトレース オブジェクト。traceが指定されている場合、expectationsは、個別の列としてではなく、trace上のAssessmentsとして提供できます。

トレースから MLflow によって生成された、指定 しないでください 。

inputs+outputsまたはtraceのいずれかが必要です。両方をパスすることはできません。

predict_fn​

生成 AI アプリのエントリ ポイント。このパラメータは 、直接評価でのみ使用されます。 predict_fn 次の要件を満たす必要があります。

  • data の inputs ディクショナリのキーをキーワード引数として受け入れます。
  • JSON シリアル化可能なディクショナリを返します。
  • MLflow Tracingでインストルメント化されています。
  • 呼び出しごとに 1 つのトレースを出力します。

scorers​

適用する品質メトリクスの一覧です。 次のものを提供できます。

詳細は スコアラー をご覧ください。

model_id​

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

データ形式​

直接評価の場合​

フィールド

必須

説明

inputs

✅

辞書があなたに渡されます predict_fn

expectations

オプション

スコアラーのためのオプションのグラウンドトゥルース

解答用紙の評価に​

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

フィールド

必須

説明

inputs

✅

生成AI アプリへの元の入力

outputs

✅

アプリからのプレコンピュート出力

expectations

オプション

スコアラーのためのオプションのグラウンドトゥルース

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

フィールド

必須

説明

trace

✅

MLflow Trace オブジェクトと入力/出力

expectations

オプション

スコアラーのためのオプションのグラウンドトゥルース

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

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

MLflow 評価データセットは、バージョン管理、リネージ追跡、および本番運用対応評価のための Unity Catalog との統合を提供します。 これらは、評価データのバージョン管理とリネージ追跡が必要な場合や、トレースを評価レコードに変換する必要がある場合に役立ちます。

Python
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()],
)

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

辞書のリストを使用して評価する​

辞書の単純なリストを使用して、正式な評価データセットを作成せずに迅速なプロトタイピングを行います。これは、迅速なプロトタイポ作成、小規模なデータセット (100 例未満)、および非公式の開発テストに役立ちます。

Python
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()],
)

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

Pandas DataFrame を使用した評価​

CSV ファイルまたは既存のデータサイエンスワークフローを操作する場合の評価に Pandas DataFrames を使用します。これは、迅速なプロトタイポ作成、小規模なデータセット (100 例未満)、および非公式の開発テストに役立ちます。

Python
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()],
)

Spark DataFrame を使用した評価​

Spark DataFrames は、大規模な評価や、データが既に Delta Lake または Unity Catalog にある場合に使用します。これは、データが Delta Lake または Unity Catalog に既に存在する場合、または評価を実行する前に MLflow 評価データセット内のレコードをフィルター処理する必要がある場合に便利です。

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

Python
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()],
)

一般的な predict_fn パターン​

アプリを直接呼び出す​

パラメーター名が評価データセットのキーと一致する場合に、 predict_fn としてアプリを直接渡します。これは、評価データセットの inputs と一致するパラメーター名を持つアプリに役立ちます。

Python
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()]
)

アプリを callable でラップする​

アプリで評価データセットの inputsとは異なるパラメーター名またはデータ構造が想定されている場合は、呼び出し可能な関数でラップします。 これは、アプリのパラメーターと評価データセットの input キーの間にパラメーター名の不一致がある場合 (例: user_input vs question)、またはデータ形式の変換が必要な場合 (例: 文字列からリストまたは JSON 解析へ)。

Python
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()]
)

デプロイされたエンドポイントの評価​

to_predict_fn 関数を使用して、エージェント フレームワーク、モデルサービング チャット エンドポイント、およびカスタム エンドポイントMosaic AI評価します。

この関数は、これらのエンドポイントと互換性のある予測関数を作成し、完全な可観測性のためにトレースが有効なエンドポイントからトレースを自動的に抽出します。

注記

to_predict_fn 関数は、エンドポイントへのkwargsパススルーを直接実行します。評価データは、エンドポイントが想定する入力形式と一致する必要があります。形式が一致しない場合、評価は失敗し、認識されない入力キーに関するエラー メッセージが表示されます。

モデルサービング チャット エンドポイントには、 messages キーでフォーマットされたデータが必要です。

Python
import mlflow
from mlflow.genai.scorers import RelevanceToQuery

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

# Evaluate the chat endpoint
results = mlflow.genai.evaluate(
    data=[{"inputs": {"messages": [{"role": "user", "content": "How does MLflow work?"}]}}],
    predict_fn=predict_fn,
    scorers=[RelevanceToQuery()]
)

記録済みモデルの評価​

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

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

Python
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()]
)

次のステップ​

に最終更新