コードベースのカスタムスコアラーを作成する
カスタム コード ベースのスコアラーは、GenAI アプリケーションの品質を測定する方法を正確に定義するための究極の柔軟性を提供します。単純なヒューリスティック、高度なロジック、プログラムによる評価など、特定のビジネスユースケースに合わせてカスタマイズされた評価メトリックを定義できます。
カスタム スコアラーは、次のシナリオで使用します。
- カスタムヒューリスティックまたはコードベースの評価の定義 メトリクス.
- アプリのトレースのデータが、Databricks の研究に基づいた LLM 審査員にどのようにマップされるかをカスタマイズします。
- 評価には (Databricks でホストされている LLM ジャッジ モデルではなく) 独自の LLM モデルを使用します。
- カスタム LLM スコアラーで提供される以上の柔軟性と制御が必要なその他のユースケース。
多くの例を含むチュートリアルについては、 「コードベースのスコアラーの例」を参照してください。
カスタムスコアラーの仕組み
カスタム スコアラーは Python で記述されており、アプリのトレースのあらゆるデータを評価するための完全な制御を提供します。カスタム スコアラーを定義したら、それを定義済みスコアラーとまったく同じように使用できます。他のスコアラーと同様に、同じカスタム スコアラーを開発時の評価に使用したり、本番運用のモニタリングに再利用したりできます。
たとえば、LLM の応答時間が許容範囲内であるかどうかをチェックするスコアラーが必要だとします。以下のMLflow UI の画像は、このカスタム メトリクスによってスコア付けされたトレースを示しています。
以下のコード スニペットは、このカスタム スコアラーを定義し、それをmlflow.genai.evaluate()
で使用します。
import mlflow
from mlflow.genai.scorers import scorer
from mlflow.entities import Trace, Feedback, SpanType
@scorer
def llm_response_time_good(trace: Trace) -> Feedback:
# Search particular span type from the trace
llm_span = trace.search_spans(span_type=SpanType.CHAT_MODEL)[0]
response_time = (llm_span.end_time_ns - llm_span.start_time_ns) / 1e9 # second
max_duration = 5.0
if response_time <= max_duration:
return Feedback(
value="yes",
rationale=f"LLM response time {response_time:.2f}s is within the {max_duration}s limit."
)
else:
return Feedback(
value="no",
rationale=f"LLM response time {response_time:.2f}s exceeds the {max_duration}s limit."
)
# Evaluate the scorer using pre-generated traces
span_check_eval_results = mlflow.genai.evaluate(
data=generated_traces,
scorers=[llm_response_time_good]
)
上記の例は、コードベースのスコアラーの一般的なパターンを示しています。
@scorer
デコレータはスコアラーを定義するために使用されます。- このスコアラーへの入力は完全な
trace
であり、AI アプリの入力、中間スパン、および出力にアクセスできます。 - スコアラーロジックは完全にカスタマイズできます。LLM や他の採点者に電話することもできます。
- このスコアラーの出力は、値と説明を含む豊富な
Feedback
オブジェクトです。 - メトリクス名は
llm_response_time_good
で、スコアラー関数名と一致します。
このパターンは、コードベースのスコアラーの可能性の 1 つにすぎません。この記事の残りの部分では、カスタム スコアラーを定義するためのオプションについて説明します。
@scorer
デコレータでスコアラーを定義する
ほとんどのコードベースのスコアラーは、 @scorer
デコレータを使用して定義する必要があります。以下は、このようなスコアラーの署名であり、可能な入力と出力を示しています。
from mlflow.genai.scorers import scorer
from typing import Optional, Any
from mlflow.entities import Feedback
@scorer
def my_custom_scorer(
*, # All arguments are keyword-only
inputs: Optional[dict[str, Any]], # App's raw input, a dictionary of input argument names and values
outputs: Optional[Any], # App's raw output
expectations: Optional[dict[str, Any]], # Ground truth, a dictionary of label names and values
trace: Optional[mlflow.entities.Trace] # Complete trace with all spans and metadata
) -> Union[int, float, bool, str, Feedback, List[Feedback]]:
# Your evaluation logic here
@scorer
デコレータよりも柔軟性を高めるには、代わりにScorer
クラスを使用してスコアラーを定義することもできます。
入力
スコアラーは、すべてのスパン、属性、および出力を含む完全なMLflow トレースを受け取ります。利便性のため、MLflow は一般的に必要なデータを抽出し、名前付き引数として渡します。すべての入力引数はオプションなので、スコアラーに必要なものだけを宣言します。
inputs
: アプリに送信されたリクエスト (例: ユーザークエリ、コンテキスト)。outputs
: アプリからの応答(生成されたテキスト、ツールの呼び出しなど)expectations
: グラウンドトゥルースまたはラベル (例: 予想される応答、ガイドラインなど)trace
: すべてのスパンを含む完全なMLflow トレースで、中間ステップ、レイテンシ、ツールの使用状況などを分析できます。トレースは、インスタンス化されたmlflow.entities.trace
クラスとしてカスタム スコアラーに渡されます。
mlflow.genai.evaluate()
を実行する場合、 inputs
、 outputs
、およびexpectations
引数をdata
引数で指定するか、トレースから解析することができます。
本番運用モニタリングに登録されているスコアラーは、常にトレースからinputs
とoutputs
を解析します。 expectations
は使用できません。
出力
スコアラーは、評価のニーズに応じて、さまざまな種類の単純な値または豊富なフィードバック オブジェクトを返すことができます。
戻り値の型 | MLflow UI ディスプレイ | ユースケース |
---|---|---|
| 合格/不合格 | バイナリ評価 |
| true/false | Booleanチェック |
| 数値 | スコア、カウント |
価値 + 根拠 | 詳細な評価 | |
| 複数のメトリクス | 多面的な評価 |
単純な値
簡単な合格/不合格または数値評価のためのプリミティブ値を出力します。以下は、応答として文字列を返す AI アプリのシンプルなスコアラーです。
@scorer
def response_length(outputs: str) -> int:
# Return a numeric metric
return len(outputs.split())
@scorer
def contains_citation(outputs: str) -> str:
# Return pass/fail string
return "yes" if "[source]" in outputs else "no"
豊富なフィードバック
スコア、根拠、メタデータを含む詳細な評価のFeedback
オブジェクトまたはFeedback
オブジェクトのリストを返します。
from mlflow.entities import Feedback, AssessmentSource
@scorer
def content_quality(outputs):
return Feedback(
value=0.85, # Can be numeric, boolean, string, or other types
rationale="Clear and accurate, minor grammar issues",
# Optional: source of the assessment. Several source types are supported,
# such as "HUMAN", "CODE", "LLM_JUDGE".
source=AssessmentSource(
source_type="HUMAN",
source_id="grammar_checker_v1"
),
# Optional: additional metadata about the assessment.
metadata={
"annotator": "me@example.com",
}
)
複数のフィードバック オブジェクトをリストとして返すことができます。各フィードバックにはname
フィールドを指定する必要があり、それらの名前は評価結果に個別のメトリックとして表示されます。
@scorer
def comprehensive_check(inputs, outputs):
return [
Feedback(name="relevance", value=True, rationale="Directly addresses query"),
Feedback(name="tone", value="professional", rationale="Appropriate for audience"),
Feedback(name="length", value=150, rationale="Word count within limits")
]
メトリクスの命名動作
スコアラーを定義するときは、スコアラーの目的を示す明確で一貫性のある名前を使用します。これらの名前は、評価、モニタリングの結果、ダッシュボードにメトリクス名として表示されます。 safety_check
やrelevance_monitor
などの MLflow 命名規則に従ってください。
@scorer
デコレータまたはScorer
クラスのいずれかを使用してスコアラーを定義すると、評価およびモニタリングによって作成された評価実行内のメトリクス名は、単純なルールに従います。
- スコアラーが 1 つ以上の
Feedback
オブジェクトを返す場合、指定されていればFeedback.name
フィールドが優先されます。 - プリミティブな戻り値または名前のない
Feedback
の場合、関数名 (@scorer
デコレータの場合) またはScorer.name
フィールド (Scorer
クラスの場合) が使用されます。
これらのルールをあらゆる可能性に拡張すると、メトリクスの命名動作に関する次の表が得られます。
戻り値 |
|
|
---|---|---|
プリミティブ値 ( | 関数名 |
|
名前なしのフィードバック | 関数名 |
|
名前付きのフィードバック |
|
|
|
|
|
評価とモニタリングでは、すべてのメトリクスに個別の名前が付いていることが重要です。 スコアラーがList[Feedback]
返す場合、 List
内の各Feedback
には異なる名前が必要です。
命名動作の例についてはチュートリアルを参照してください。
エラー処理
スコアラーがトレースのエラーに遭遇すると、MLflow はそのトレースのエラーの詳細をキャプチャし、正常に実行を続行できます。エラーの詳細をキャプチャするために、MLflow は次の 2 つの方法を提供します。
- 例外を伝播させて (推奨)、MLflow がエラー メッセージをキャプチャできるようにします。
- 例外を明示的に処理します。
例外を伝播させる(推奨)
最も単純なアプローチは、例外を自然にスローさせることです。MLflow は自動的に例外をキャプチャし、エラーの詳細を含むFeedback
オブジェクトを作成します。以下の例では、スコアラーは特定のフィールドを持つ JSON を想定しています。
import mlflow
from mlflow.entities import Feedback
from mlflow.genai.scorers import scorer
@scorer
def is_valid_response(outputs: str) -> Feedback:
import json
# Let json.JSONDecodeError propagate if response isn't valid JSON
data = json.loads(outputs)
# Let KeyError propagate if required fields are missing
summary = data["summary"]
confidence = data["confidence"]
return Feedback(
value=True,
rationale=f"Valid JSON with confidence: {confidence}"
)
# Run the scorer on invalid data that triggers exceptions
invalid_data = [
{
# Valid JSON
"outputs": '{"summary": "this is a summary", "confidence": 0.95}'
},
{
# Invalid JSON
"outputs": "invalid json",
},
{
# Missing required fields
"outputs": '{"summary": "this is a summary"}'
},
]
mlflow.genai.evaluate(
data=invalid_data,
scorers=[is_valid_response],
)
例外が発生すると、MLflow は次の内容でFeedback
を作成します。
value
:None
error
: 例外オブジェクト、エラーメッセージ、スタックトレースなどの例外の詳細
評価結果にエラー情報が表示されます。対応する行を開いてエラーの詳細を表示します。
例外を明示的に処理する
カスタム エラー処理や特定のエラー メッセージを提供するには、例外をキャッチし、 None
値とエラーの詳細を含むFeedback
を返します。
from mlflow.entities import AssessmentError, Feedback
@scorer
def is_valid_response(outputs):
import json
try:
data = json.loads(outputs)
required_fields = ["summary", "confidence", "sources"]
missing = [f for f in required_fields if f not in data]
if missing:
return Feedback(
error=AssessmentError(
error_code="MISSING_REQUIRED_FIELDS",
error_message=f"Missing required fields: {missing}",
),
)
return Feedback(
value=True,
rationale="Valid JSON with all required fields"
)
except json.JSONDecodeError as e:
return Feedback(error=e) # Can pass exception object directly to the error parameter
error
問題は以下を受け入れます:
- Python 例外 : 例外オブジェクトを直接渡す
AssessmentError
: エラーコードを含む構造化されたエラーレポート用
Scorer
クラスでスコアラーを定義する
上で説明した@scorer
デコレータはシンプルで、一般的に推奨されますが、不十分な場合は代わりにScorer
基本クラスを使用できます。クラスベースの定義により、より複雑なスコアラー、特に状態を必要とするスコアラーが可能になります。Scorer
クラスはPydantic オブジェクトなので、追加のフィールドを定義して__call__
メソッドで使用することができます。
メトリクス名を設定するには、 name
フィールドを定義する必要があります。 Feedback
オブジェクトのリストを返す場合は、名前の競合を避けるために、各Feedback
のname
フィールドを設定する必要があります。
from mlflow.genai.scorers import Scorer
from mlflow.entities import Feedback
from typing import Optional
# Scorer class is a Pydantic object
class CustomScorer(Scorer):
# The `name` field is mandatory
name: str = "response_quality"
# Define additional fields
my_custom_field_1: int = 50
my_custom_field_2: Optional[list[str]] = None
# Override the __call__ method to implement the scorer logic
def __call__(self, outputs: str) -> Feedback:
# Your logic here
return Feedback(
value=True,
rationale="Response meets all quality criteria"
)
国家管理
Scorer
クラスを使用してスコアラーを作成するときは、Python クラスで状態を管理するためのルールに注意してください。特に、変更可能なクラス属性ではなく、インスタンス属性を使用するようにしてください。以下の例は、スコアラーインスタンス間で状態を誤って共有していることを示しています。
from mlflow.genai.scorers import Scorer
from mlflow.entities import Feedback
# WRONG: Don't use mutable class attributes
class BadScorer(Scorer):
results = [] # Shared across all instances!
name: str = "bad_scorer"
def __call__(self, outputs, **kwargs):
self.results.append(outputs) # Causes issues
return Feedback(value=True)
# CORRECT: Use instance attributes
class GoodScorer(Scorer):
results: list[str] = None
name: str = "good_scorer"
def __init__(self):
self.results = [] # Per-instance state
def __call__(self, outputs, **kwargs):
self.results.append(outputs) # Safe
return Feedback(value=True)
次のステップ
- コードベースのスコアラーの例- コードベースのスコアラーの多くの例を参照してください
- コードベースのスコアラーを開発する- カスタム スコアラーの開発ワークフローをステップスルーする
- 評価ハーネス-
mlflow.genai.evaluate()
がスコアラーをどのように使用しているかを理解する - GenAI の本番運用モニタリング- 継続的なモニタリングのためにスコアラーを展開します
APIリファレンス
このガイドで使用されるMLflow APIsには以下が含まれます。