自訂程式碼型 評分器 提供最極致的彈性,能精確定義您 GenAI 應用程式的品質衡量標準。 您可以定義針對特定業務使用案例量身打造的評估指標,無論是基於簡單的啟發式、進階邏輯或程式設計評估。
針對下列案例使用自定義計分器:
- 定義自訂啟發式方法或程式碼型評估指標。
- 如何自訂應用程式追蹤資料的映射方式,以便對應至 Databricks 研究支持的 LLM 評估者。
- 使用您自己的 LLM 進行評估,而不是使用 Databricks 提供的 LLM 進行判斷。
- 有沒有其他需要比自訂 LLM 法官提供的更多彈性和控制的使用案例?
如需包含許多範例的教學課程,請參閱 程式碼型評分器範例。
自訂計分器的運作方式
自訂計分器是以 Python 撰寫,並讓您完全掌控分析和評估應用程式追蹤中的任何數據。 定義自訂計分器之後,您可以像 內建 LLM Judge 一樣使用它。 與其他評分器一樣,相同的自訂評分器可用於 開發中的評估 ,並重複用於 生產環境中的監視。
例如,假設您想要一個評分器來檢查 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 應用程式的輸入、中間過程和輸出。 - 記分器邏輯可以完全自訂。 您可以致電法學碩士或其他評分員。
- 此評分器的輸出是具有值和解釋的豐富
Feedback物件。 -
度量名稱 為
llm_response_time_good,符合評分器函數名稱。
這種模式只是基於代碼的評分器的一種可能性。 本文的其餘部分說明定義自訂計分器的選項。
使用 @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 顯示 | 用例 |
|---|---|---|
"yes"/"no" |
通過/失敗 | 二進位評估 |
True/False |
正確/錯誤 | 布爾值檢查 |
int/float |
數值 | 分數、計數 |
Feedback |
值 + 理由 | 詳細評量 |
List[Feedback] |
多個指標 | 多層面評估 |
簡單值
輸出原始值,以進行簡單的通過/失敗或數值評估。 以下是傳回字串作為回應的 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")
]
指標命名方式
當您定義計分器時,請使用清晰、一致的名稱來指出計分器的目的。 這些名稱會在您的評估、監控結果及儀表板中顯示為指標名稱。 遵循 MLflow 命名慣例,例如 safety_check 或 relevance_monitor。
當您使用@scorer裝飾器或Scorer類別來定義評分器時,評估和監測所建立的評估執行中的度量名稱會遵循簡單的規則:
- 如果評分器傳回一或多個
Feedback物件,Feedback.name則欄位優先 (如果已指定)。 - 對於原始傳回值或未命名的
Feedback,會使用函數名稱(@scorer裝飾器)或Scorer.name欄位(Scorer類別)。
將這些規則擴展到所有可能性,形成以下度量命名行為的表格:
| 返回值 |
@scorer 裝飾器行為 |
Scorer 類別行為 |
|---|---|---|
基本值 (int, float, str) |
函式名稱 |
name 欄位 |
| 沒有名字的反饋 | 函式名稱 |
name 欄位 |
| 具名反饋 |
Feedback 名字 |
Feedback 名字 |
List[Feedback] 含有名字的 |
Feedback 名稱 |
Feedback 名稱 |
對於評估和監控,所有指標都必須具有不同的名稱。 如果記分員傳回 List[Feedback],則 Feedback 中的每個List都必須有一個獨特的名稱。
請參閱教學課程中的 命名行為範例 。
存取計分器中的秘密
自訂評分器可以存取 Databricks 秘密 ,以安全地使用 API 金鑰和認證。 這在整合外部服務時非常有用,例如需要身份驗證的自訂 LLM 端點,例如 Azure OpenAI、AWS Bedrock 等。 這種方法適用於 開發評估 和 生產監控
根據預設, dbutils 在評分器執行階段環境中無法使用。 若要存取評分器執行階段環境中的秘密,請從評分器函式內部呼叫 from databricks.sdk.runtime import dbutils 。
下列範例顯示如何存取自訂評分器中的密碼:
import mlflow
from mlflow.genai.scorers import scorer, ScorerSamplingConfig
from mlflow.entities import Trace, Feedback
@scorer
def custom_llm_scorer(trace: Trace) -> Feedback:
# Explicitly import dbutils to access secrets
from databricks.sdk.runtime import dbutils
# Retrieve your API key from Databricks secrets
api_key = dbutils.secrets.get(scope='my-scope', key='api-key')
# Use the API key to call your custom LLM endpoint
# ... your custom evaluation logic here ...
return Feedback(
value="yes",
rationale="Evaluation completed using custom endpoint"
)
# Register and start the scorer
custom_llm_scorer.register()
custom_llm_scorer.start(sampling_config = ScorerSamplingConfig(sample_rate=1))
錯誤處理
當評分器遇到追蹤錯誤時,MLflow 可以擷取該追蹤的錯誤詳細數據,然後繼續正常執行。 為了擷取錯誤詳細數據,MLflow 提供兩種方法:
- 讓例外狀況傳播 (建議) ,讓 MLflow 可以為您擷取錯誤訊息。
- 明確處理例外狀況。
讓例外狀況傳播(建議)
最簡單的方法是讓例外狀況自然擲回。 MLflow 會自動擷取該例外並建立 Feedback 包含以下錯誤細節的物件:
-
value:None -
error:例外狀況詳細數據,例如例外狀況對象、錯誤訊息和堆棧追蹤
誤差資訊會顯示在評估結果中。 開啟對應的數據列以查看錯誤詳細數據。
明確處理例外狀況
若要自訂錯誤處理或提供特定錯誤訊息,請攔截例外狀況,並傳回具有 Feedback 值和錯誤詳細資料的 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:用於具有錯誤代碼的結構化錯誤報告
使用 Scorer 類別定義計分器
上面描述的 @scorer 裝飾器 很簡單,一般推薦,但當它不夠用時,可以改用 Scorer 基類。 以類別為基礎的定義允許更複雜的評分器,尤其是需要狀態的評分器。 該 Scorer 類別是一個 Pydantic 對象,因此您可以定義其他欄位並在方法中使用 __call__ 它們。
您必須將 name 欄位定義為度量名稱。 如果您傳回物件清單Feedback,則必須在每個name物件中設定Feedback欄位,以避免命名衝突。
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()使用你的評分器 - 監控生產環境中的生成式人工智慧 - 部署評分器以持續監控
API 參考資料
本指南中使用的 MLflow API 包括: