コード ベースのスコアラーを開発する

AI の MLflow 評価では、 カスタム コードベースのスコアラー を使用して、エージェントまたはアプリケーションの柔軟な評価メトリックを定義できます。

スコアラーを開発するときは、多くの場合、迅速に反復処理する必要があります。 この開発者ワークフローを使用して、毎回アプリ全体を再実行せずにスコアラーを更新します。

  1. 評価データを定義する
  2. アプリからトレースを生成する
  3. トレースをクエリして結果を格納する
  4. スコアラーを反復処理するときは、格納されているトレースを使用して評価します

サンプル ノートブックには、このチュートリアルのすべてのコードが含まれています。

前提条件: MLflow を設定し、アプリケーションを定義する

mlflow[databricks]を最新バージョンに更新して、GenAI エクスペリエンスを最大限に高め、次のサンプル アプリでは OpenAI クライアントを使用しているため、openaiをインストールします。

%pip install -q --upgrade "mlflow[databricks]>=3.14.0" "openai>=1.0.0"
dbutils.library.restartPython()

このページの例はUnityカタログに保存されたトレースにアクセスできます。 SQLウェアハウスを実行前に設定してください:

import os

os.environ["MLFLOW_TRACING_SQL_WAREHOUSE_ID"] = "<SQL_WAREHOUSE_ID>"

以下の mlflow.openai.autolog() 呼び出しでは、MLflow Tracing を使用してアプリケーションが自動的にインストルメント化されます。 記録されたトレースは、評価中のスコアラーへの入力になります。

import mlflow
from mlflow.entities.trace_location import UnityCatalog

mlflow.openai.autolog()

# Set up MLflow tracking to Databricks.
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment(
    experiment_name="/Shared/docs-demo",
    trace_location=UnityCatalog(
        catalog_name="<UC_CATALOG_NAME>",
        schema_name="<UC_SCHEMA_NAME>",
        table_prefix="<UC_TABLE_PREFIX>",
    ),
)

databricks-openai パッケージを使用して、Databricks でホストされる LLM に接続する OpenAI クライアントを取得します。 使用可能な基礎モデルからモデルを選択します。

from databricks_openai import DatabricksOpenAI

# Create an OpenAI client that is connected to Databricks-hosted LLMs
client = DatabricksOpenAI()

# Select an LLM
model_name = "databricks-claude-sonnet-4"

このチュートリアル用の簡単な質問に回答するアシスタント アプリを作成します。

@mlflow.trace
def sample_app(messages: list[dict[str, str]]):
    # 1. Prepare messages for the LLM
    messages_for_llm = [
        {"role": "system", "content": "You are a helpful assistant."},
        *messages,
    ]

    # 2. Call LLM to generate a response
    response = client.chat.completions.create(
        model= model_name,
        messages=messages_for_llm,
    )
    return response.choices[0].message.content


sample_app([{"role": "user", "content": "What is the capital of France?"}])

手順 1: 評価データを定義する

以下の評価データは、LLM が応答するための要求の一覧です。 このアプリでは、要求は単純な質問や複数のメッセージとの会話にすることができます。

eval_dataset = [
    {
        "inputs": {
            "messages": [
                {"role": "user", "content": "How much does a microwave cost?"},
            ]
        },
    },
    {
        "inputs": {
            "messages": [
                {
                    "role": "user",
                    "content": "Can I return the microwave I bought 2 months ago?",
                },
            ]
        },
    },
    {
        "inputs": {
            "messages": [
                {
                    "role": "user",
                    "content": "I'm having trouble with my account.  I can't log in.",
                },
                {
                    "role": "assistant",
                    "content": "I'm sorry to hear that you're having trouble with your account.  Are you using our website or mobile app?",
                },
                {"role": "user", "content": "Website"},
            ]
        },
    },
]

手順 2: アプリからトレースを生成する

mlflow.genai.evaluate()を使用して、アプリからトレースを生成します。 evaluate()には少なくとも 1 つのスコアラーが必要であるため、この初期トレース生成用にプレースホルダー スコアラーを定義します。

from mlflow.genai.scorers import scorer

@scorer
def placeholder_metric() -> int:
    # placeholder return value
    return 1

プレースホルダー スコアラーを使用して評価を実行します。

eval_results = mlflow.genai.evaluate(
    data=eval_dataset,
    predict_fn=sample_app,
    scorers=[placeholder_metric]
)

上記のコードを実行した後、評価データセットの各行に対して実験に 1 つのトレースが必要です。 Databricks Notebook では、セルの結果の一部としてトレースの視覚化も表示されます。 評価中に sample_app によって生成された LLM の応答は、ノートブックトレース UI の [出力] フィールドと MLflow 実験 UI の [応答 ] 列に表示されます。

生成されたサンプル トレース

手順 3: 結果のトレースに対してクエリを実行して格納する

生成されたトレースをローカル変数に格納します。 mlflow.search_traces()関数は、トレースの Pandas データフレームを返します。

generated_traces = mlflow.search_traces(run_id=eval_results.run_id)
generated_traces

手順 4: スコアラーを反復処理するときに、格納されているトレースを使用して evaluate() を呼び出す

トレースの Pandas DataFrame を入力データセットとして evaluate() に直接渡します。 これにより、アプリを再実行することなく、メトリックをすばやく反復処理できます。 次のコードは、事前計算済みの generated_tracesで新しいスコアラーを実行します。

from mlflow.genai.scorers import scorer

@scorer
def response_length(outputs: str) -> int:
    # Example metric.
    # Implement your actual metric logic here.
    return len(outputs)

# Note the lack of a predict_fn parameter.
mlflow.genai.evaluate(
    data=generated_traces,
    scorers=[response_length],
)

ノートブックの例

次のノートブックには、このページのすべてのコードが含まれています。

MLflow 評価用のコード ベーススコアラーの開発者ワークフロー

ノートブックを入手

その他のリソース

  • 最終更新: