本页提供 MLflow 评估和监视概念的参考文档。 有关指南和教程,请参阅 “评估和监视”。
小窍门
有关 MLflow 3 评估和监视 API 文档,请参阅 API 参考。
快速参考
概念 | 目的 | Usage |
---|---|---|
记分器 | 评估跟踪质量 |
@scorer 修饰器或 Scorer 类 |
法官 | 基于 LLM 的评估 | 包裹在计分工具中以便使用 |
评估工具 | 运行脱机评估 | mlflow.genai.evaluate() |
评估数据集 | 测试数据管理 | mlflow.genai.datasets |
评估测试 | 存储评估结果 | 由线束创建 |
生产监控 | 实时质量跟踪 |
Scorer.register 、Scorer.start |
记分器: mlflow.genai.scorers
from mlflow.genai.scorers import scorer
from mlflow.entities import Feedback
from typing import Optional, Dict, Any, List
@scorer
def my_custom_scorer(
*, # MLflow calls your scorer with named arguments
inputs: Optional[Dict[Any, Any]], # App's input from trace
outputs: Optional[Dict[Any, Any]], # App's output from trace
expectations: Optional[Dict[str, Any]], # Ground truth (offline only)
trace: Optional[mlflow.entities.Trace] # Complete trace
) -> int | float | bool | str | Feedback | List[Feedback]:
# Your evaluation logic
return Feedback(value=True, rationale="Explanation")
法官: mlflow.genai.judges
必须被封装在评分器中的基于 LLM 的质量评估器。
from mlflow.genai.judges import is_safe, is_relevant
from mlflow.genai.scorers import scorer
# Direct usage
feedback = is_safe(content="Hello world")
# Wrapped in scorer
@scorer
def safety_scorer(outputs):
return is_safe(content=outputs["response"])
评估工具: mlflow.genai.evaluate(...)
在开发过程中协调脱机评估。
import mlflow
from mlflow.genai.scorers import Safety, RelevanceToQuery
results = mlflow.genai.evaluate(
data=eval_dataset, # Test data
predict_fn=my_app, # Your app
scorers=[Safety(), RelevanceToQuery()], # Quality metrics
model_id="models:/my-app/1" # Optional version tracking
)
评估数据集: mlflow.genai.datasets.EvaluationDataset
版本化测试数据(可选的真实数据)。
import mlflow.genai.datasets
# Create from production traces
dataset = mlflow.genai.datasets.create_dataset(
uc_table_name="catalog.schema.eval_data"
)
# Add traces
traces = mlflow.search_traces(filter_string="trace.status = 'OK'")
dataset.insert(traces)
# Use in evaluation
results = mlflow.genai.evaluate(data=dataset, ...)
评估运行: mlflow.entities.Run
评估结果中包含反馈痕迹。
# Access evaluation results
traces = mlflow.search_traces(run_id=results.run_id)
# Filter by feedback
good_traces = traces[traces['assessments'].apply(
lambda x: all(a.value for a in x if a.name == 'Safety')
)]
生产监控
重要
此功能在 Beta 版中。
持续评估已部署的应用程序。
import mlflow
from mlflow.genai.scorers import Safety, ScorerSamplingConfig
# Register the scorer with a name and start monitoring
safety_scorer = Safety().register(name="my_safety_scorer") # name must be unique to experiment
safety_scorer = safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.7))