得点
スコアラーは、出力を分析し、構造化されたフィードバックを生成することで、生成AI アプリの品質を評価します。一度書けば、開発中や本番運用中など、どこでも使えます。
クイックリファレンス
戻り値の型 | UIディスプレイ | ユースケース |
---|---|---|
| 合格/不合格 | バイナリ評価 |
| true/false | Boolean チェック |
| 数値 | スコア、カウント |
| 価値 + 根拠 | 詳細な評価 |
| 複数のメトリクス | 多面評価 |
一度書けば、どこでも使える
MLflow スコアラーの主要な設計原則は 、一度記述すれば、どこでも使用できる ことです。同じスコアラー機能は、次の場合にシームレスに機能します。
- 開発 :
mlflow.genai.evaluate()
を使用してアプリのさまざまなバージョンを評価します - 本番運用 : Monitor live traffic quality with MLflow's 本番運用 モニタリング サービス
この統一されたアプローチにより、品質メトリクスをローカルで開発およびテストし、まったく同じロジックを変更せずに本番運用にデプロイできます。
from mlflow.genai.scorers import scorer
from mlflow.entities import Feedback
# Define your scorer once
@scorer
def response_completeness(outputs: str) -> Feedback:
# Outputs is return value of your app. Here we assume it's a string.
if len(outputs.strip()) < 10:
return Feedback(
value=False,
rationale="Response too short to be meaningful"
)
if outputs.lower().endswith(("...", "etc", "and so on")):
return Feedback(
value=False,
rationale="Response appears incomplete"
)
return Feedback(
value=True,
rationale="Response appears complete"
)
# Directly call the scorer function for spot testing
response_completeness(outputs="This is a test response...")
# Use in development evaluation
mlflow.genai.evaluate(
data=test_dataset,
predict_fn=my_app,
scorers=[response_completeness]
)
スコアラーの仕事方法
スコアラーは、生成AI アプリケーションからのトレースを分析し、品質評価を生成します。フローは次のとおりです。
- アプリが実行され 、その実行をキャプチャした トレース が生成されます
- MLflow はトレース をスコアラー関数に渡します
- スコアラーは 、カスタムロジックを使用してトレースの入力、出力、および中間実行ステップを分析します
- フィードバックは スコアと説明で生成されます
- フィードバックは解析のためにトレース に添付されます
入力
スコアラーは、すべてのスパン、属性、出力を含む完全な MLflow トレース を受け取ります。便宜上、MLflow では、一般的に必要なデータも抽出され、名前付き引数として渡されます。
@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 metadata
) -> Union[int, float, bool, str, Feedback, List[Feedback]]:
# Your evaluation logic here
すべてのパラメーターは省略可能であり、スコアラーが必要とするものだけを宣言します。
- inputs : アプリに送信されたリクエスト(ユーザークエリ、コンテキストなど)。
- outputs : アプリからのレスポンス(生成されたテキスト、ツール呼び出しなど)
- 期待 値: グラウンド トゥルースまたはラベル (期待される応答、ガイドラインなど)
- trace :すべてのスパンを含む完全な実行トレースで、中間ステップ、レイテンシ、ツールの使用状況などの分析が可能です。
mlflow.genai.evaluate()
を実行する場合、inputs
、outputs
、および expectations
パラメーターを data
引数で指定するか、トレースから解析できます。
mlflow.genai.create_monitor()
を実行すると、inputs
パラメーターと outputs
パラメーターは常にトレースから解析されます。expectations
はご利用いただけません。
出力
スコアラーは、評価のニーズに応じてさまざまなタイプを返すことができます。
単純な値
単純な合格/不合格または数値評価のプリミティブ値を返します。
- 合格/不合格文字列 :
"yes"
または"no"
UI で "Pass" または "Fail" としてレンダリングされます - Boolean値 : バイナリ評価の場合は
True
またはFalse
- 数値 : スコア、カウント、または測定値の整数または浮動小数点数
# These example assumes your app returns a string as a response.
@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
オブジェクトを返し、説明を含めます。
from mlflow.entities import Feedback, AssessmentSource
@scorer
def content_quality(outputs):
return Feedback(
value=0.85, # Can be numeric, boolean, or string
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",
}
)
複数のフィードバックオブジェクトをリストとして返すことができます。各フィードバックは、評価結果に個別のメトリクスとして表示されます。
@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")
]
メトリクスの命名動作
@scorer
デコレータを使用する場合、評価結果のメトリクス名は次のルールに従います。
-
プリミティブ値または名前のない 1 つのフィードバック : スコアラー関数の名前がフィードバック名になります
Python@scorer
def word_count(outputs: str) -> int:
# "word_count" will be used as a metric name
return len(outputs).split()
@scorer
def response_quality(outputs: Any) -> Feedback:
# "response_quality" will be used as a metric name
return Feedback(value=True, rationale="Good quality") -
explcit 名を持つ 1 つのフィードバック : Feedback オブジェクトで指定された名前がメトリクス名として使用されます
Python@scorer
def assess_factualness(outputs: Any) -> Feedback:
# Name "factual_accuracy" is explicitly specfied, it will be used as a metric name
return Feedback(name="factual_accuracy", value=True, rationale="Factual accuracy is high") -
複数のフィードバック : 各フィードバック オブジェクトで指定された名前は保持されます。フィードバックごとに一意の名前を指定する必要があります。
Python@scorer
def multi_aspect_check(outputs) -> list[Feedback]:
# These names ARE used since multiple feedbacks are returned
return [
Feedback(name="grammar", value=True, rationale="No errors"),
Feedback(name="clarity", value=0.9, rationale="Very clear"),
Feedback(name="completeness", value="yes", rationale="All points addressed")
]
この命名動作により、評価結果とダッシュボードで一貫したメトリクス名が保証されます。
エラー処理
スコアラーでエラーが発生した場合、MLflow には 2 つのアプローチが用意されています。
例外を伝播させる (推奨)
最も簡単な方法は、例外が自然にスローされるようにすることです。MLflow は例外を自動的にキャプチャし、エラーの詳細を含む Feedback オブジェクトを作成します。
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 は次の情報を使用してフィードバックを作成します。
value
:None
error
: 例外オブジェクト、エラーメッセージ、スタックトレースなどの例外の詳細
評価結果にはエラー情報が表示されます。対応する行を開いて、エラーの詳細を確認します。
例外を明示的に処理する
カスタム エラー処理の場合、または特定のエラー メッセージを提供する場合は、例外をキャッチし、 None
値とエラーの詳細を含むフィードバックを返します。
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 : エラーコードを含む構造化エラーレポートの場合
期待が利用可能な場合
期待 値 (グラウンド トゥルースまたはラベル) は、通常、オフライン評価にとって重要です。mlflow.genai.evaluate()
の実行時に 2 つの方法で指定できます。
- 入力
data
引数に列 (またはフィールド)expectations
を含めます。 Expectation
を Traces に関連付け、それらをdata
引数に渡します。
@scorer
def exact_match(outputs: str, expectations: dict[str, Any]) -> Feedback:
expected = expectations.get("expected_response")
is_correct = outputs == expected
return Feedback(
value=is_correct,
rationale=f"Response {'matches' if is_correct else 'differs from'} expected"
)
data = [
{
"inputs": {"question": "What is the capital of France?"},
"outputs": "Paris",
# Specify expected response in the expectations field
"expectations": {
"expected_response": "Paris"
}
},
]
mlflow.genai.evaluate(
data=data,
scorers=[exact_match],
)
本番運用 モニタリングは、グラウンドトゥルースなしでライブトラフィックを評価するため、通常は期待値を持ちません。 オフラインとオンラインの両方で同じスコアラーを使用する場合は、期待を適切に処理するように設計します
トレース・データの使用
スコアラーは、複雑なアプリケーションの動作を評価するために、完全なトレースにアクセスできます。
from mlflow.entities import Feedback, Trace
from mlflow.genai.scorers import scorer
@scorer
def tool_call_efficiency(trace: Trace) -> Feedback:
"""Evaluate how effectively the app uses tools"""
# Retrieve all tool call spans from the trace
tool_calls = trace.search_spans(span_type="TOOL")
if not tool_calls:
return Feedback(
value=None,
rationale="No tool usage to evaluate"
)
# Check for redundant calls
tool_names = [span.name for span in tool_calls]
if len(tool_names) != len(set(tool_names)):
return Feedback(
value=False,
rationale=f"Redundant tool calls detected: {tool_names}"
)
# Check for errors
failed_calls = [s for s in tool_calls if s.status.status_code != "OK"]
if failed_calls:
return Feedback(
value=False,
rationale=f"{len(failed_calls)} tool calls failed"
)
return Feedback(
value=True,
rationale=f"Efficient tool usage: {len(tool_calls)} successful calls"
)
mlflow.genai.evaluate()
を使用してオフライン評価を実行する場合、トレースは次のようになります。
data
引数で指定されます (既に使用可能な場合)。data
引数のinputs
に対してpredict_fn
を実行することによって生成されます。
mlflow.genai.create_monitor()
を使用して本番運用 モニタリングを実行すると、モニターによって収集されたトレースは、指定されたサンプリング基準とフィルタリング基準を使用して、スコアラー関数に直接渡されます。
スコアラーの実装アプローチ
MLflow には、スコアラーを実装する方法が 2 つあります。
デコレーターアプローチ(推奨)
@scorer
デコレータを使用して、シンプルな関数ベースのスコアラーを作成します。
from mlflow.genai.scorers import scorer
from mlflow.entities import Feedback
@scorer
def response_tone(outputs: str) -> Feedback:
"""Check if response maintains professional tone"""
informal_phrases = ["hey", "gonna", "wanna", "lol", "btw"]
found = [p for p in informal_phrases if p in outputs.lower()]
if found:
return Feedback(
value=False,
rationale=f"Informal language detected: {', '.join(found)}"
)
return Feedback(
value=True,
rationale="Professional tone maintained"
)
クラスベースのアプローチ
Scorer
基本クラスは、状態を必要とするより複雑なスコアラーに使用します。Scorer
クラスはPydanticオブジェクトであるため、追加のフィールドを定義して__call__
メソッドで使用できます。
from mlflow.genai.scorers import Scorer
from mlflow.entities import Feedback
from typing import Optional
# Scorer class is a Pydantic object
class ResponseQualityScorer(Scorer):
# The `name` field is mandatory
name: str = "response_quality"
# Define additiona lfields
min_length: int = 50
required_sections: Optional[list[str]] = None
# Override the __call__ method to implement the scorer logic
def __call__(self, outputs: str) -> Feedback:
issues = []
# Check length
if len(outputs.split()) < self.min_length:
issues.append(f"Too short (minimum {self.min_length} words)")
# Check required sections
missing = [s for s in self.required_sections if s not in outputs]
if missing:
issues.append(f"Missing sections: {', '.join(missing)}")
if issues:
return Feedback(
value=False,
rationale="; ".join(issues)
)
return Feedback(
value=True,
rationale="Response meets all quality criteria"
)
カスタムスコアラー開発ワークフロー
カスタム スコアラーを開発する場合、多くの場合、アプリケーションを毎回再実行せずに迅速に反復する必要があります。MLflow は、効率的なワークフローをサポートします。
- トレースを一度生成 するには、次のようにアプリを実行します。
mlflow.genai.evaluate()
- トレースの保存方法
mlflow.search_traces()
- アプリを再実行せずに、保存されたトレースを
evaluate()
に渡すことで 、スコアラーを反復処理 します
このアプローチにより、スコアラーの開発にかかる時間とリソースを節約できます。
# Step 1: Generate traces with a placeholder scorer
initial_results = mlflow.genai.evaluate(
data=test_dataset,
predict_fn=my_app,
scorers=[lambda **kwargs: 1] # Placeholder scorer
)
# Step 2: Store traces for reuse
traces = mlflow.search_traces(run_id=initial_results.run_id)
# Step 3: Iterate on your scorer without re-running the app
@scorer
def my_custom_scorer(outputs):
# Your evaluation logic here
pass
# Test scorer on stored traces (no predict_fn needed)
results = mlflow.genai.evaluate(
data=traces,
scorers=[my_custom_scorer]
)
よくある落とし穴
デコレーターによるスコアラーの命名
from mlflow.genai.scorers import scorer
from mlflow.entities import Feedback
# GOTCHA: Function name becomes feedback name for single returns
@scorer
def quality_check(outputs):
# This 'name' parameter is IGNORED
return Feedback(name="ignored", value=True)
# Feedback will be named "quality_check"
# CORRECT: Use function name meaningfully
@scorer
def response_quality(outputs):
return Feedback(value=True, rationale="Good quality")
# Feedback will be named "response_quality"
# EXCEPTION: Multiple feedbacks preserve their names
@scorer
def multi_check(outputs):
return [
Feedback(name="grammar", value=True), # Name preserved
Feedback(name="spelling", value=True), # Name preserved
Feedback(name="clarity", value=0.9) # Name preserved
]
スコアラーの状態管理
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!
def __call__(self, outputs, **kwargs):
self.results.append(outputs) # Causes issues
return Feedback(value=True)
# CORRECT: Use instance attributes
class GoodScorer(Scorer):
def __init__(self):
super().__init__(name="good_scorer")
self.results = [] # Per-instance state
def __call__(self, outputs, **kwargs):
self.results.append(outputs) # Safe
return Feedback(value=True)
次のステップ
- カスタムコードベースのスコアラーを作成する - Python関数を使用してドメイン固有のスコアラーを構築します
- LLM審査員による評価 - 一般的な品質メトリクスのための事前構築済みのLLMベースのスコアラーを使用する
- 本番運用のトラフィックを 監視するために同じスコアラーを適用する