评估工具

通过将 mlflow.genai.evaluate() 函数与测试数据(评估数据集)结合使用，并应用 记分器，系统地对 GenAI 应用进行质量测试。

快速参考

工作原理

1. 在测试输入上运行应用，捕获跟踪
2. 应用评分器 来评估质量，创建 反馈
3. 将结果存储在评估过程中

先决条件

1. 安装 MLflow 和所需包

   pip install --upgrade "mlflow[databricks]>=3.1.0" openai "databricks-connect>=16.1"
   

2. 请按照 设置环境快速指南 创建 MLflow 试验。

两种评估模式

模式 1：直接评估（建议）

MLflow 直接调用 GenAI 应用程序，以生成和评估轨迹。 可以传递包装在 Python 函数中的应用程序的入口点（predict_fn），或者，如果您的应用程序被部署为 Databricks 模型服务终结点，则传递包装在 to_predict_fn中的终结点。

优点：

• 允许在脱机评估与生产监视之间轻松重复使用评分器
• 应用的执行自动并行化，以加快评估速度

通过直接调用应用，此模式使你能够重复使用在 生产监视 中为脱机评估定义的记分器，因为生成的跟踪将相同。

步骤 1：运行评估

import mlflow
from mlflow.genai.scorers import RelevanceToQuery, Safety

# Your GenAI app with MLflow tracing
@mlflow.trace
def my_chatbot_app(question: str) -> dict:
    # Your app logic here
    if "MLflow" in question:
        response = "MLflow is an open-source platform for managing ML and GenAI workflows."
    else:
        response = "I can help you with MLflow questions."

    return {"response": response}

# Evaluate your app
results = mlflow.genai.evaluate(
    data=[
        {"inputs": {"question": "What is MLflow?"}},
        {"inputs": {"question": "How do I get started?"}}
    ],
    predict_fn=my_chatbot_app,
    scorers=[RelevanceToQuery(), Safety()]
)

步骤 2：在 UI 中查看结果

模式 2：答案表评估

当无法直接运行 GenAI 应用时，请提供预计算输出或现有跟踪进行评估。

用例：

• 测试外部系统的输出
• 评估历史痕迹
• 比较不同平台的输出

示例（带有输入/输出）：

步骤 1：运行评估

import mlflow
from mlflow.genai.scorers import Safety, RelevanceToQuery

# Pre-computed results from your GenAI app
results_data = [
    {
        "inputs": {"question": "What is MLflow?"},
        "outputs": {"response": "MLflow is an open-source platform for managing machine learning workflows, including tracking experiments, packaging code, and deploying models."},
    },
    {
        "inputs": {"question": "How do I get started?"},
        "outputs": {"response": "To get started with MLflow, install it using 'pip install mlflow' and then run 'mlflow ui' to launch the web interface."},
    }
]

# Evaluate pre-computed outputs
evaluation = mlflow.genai.evaluate(
    data=results_data,
    scorers=[Safety(), RelevanceToQuery()]
)

步骤 2：在 UI 中查看结果

现有轨迹的示例：

import mlflow

# Retrieve traces from production
traces = mlflow.search_traces(
    filter_string="trace.status = 'OK'",
)

# Evaluate problematic traces
evaluation = mlflow.genai.evaluate(
    data=traces,
    scorers=[Safety(), RelevanceToQuery()]
)

关键参数

def mlflow.genai.evaluate(
    data: Union[pd.DataFrame, List[Dict], mlflow.genai.datasets.EvaluationDataset],
    scorers: list[mlflow.genai.scorers.Scorer],
    predict_fn: Optional[Callable[..., Any]] = None,
    model_id: Optional[str] = None,
) -> mlflow.models.evaluation.base.EvaluationResult:

data

以下格式之一的评估数据集：

• EvaluationDataset（推荐）
• 字典、Pandas 数据帧或 Spark 数据帧的列表

如果数据参数作为数据帧或字典列表提供，则它必须遵循以下架构。 这与 EvaluationDataset 使用的架构一致。 除了跟踪每个记录的世系之外，我们建议使用一个 EvaluationDataset ，因为它将强制实施架构验证。

predict_fn

GenAI 应用的入口点（仅限模式 1）。 必须：

• 接受inputs字典中的data键作为关键字参数
• 返回 JSON 可序列化字典
• 使用 MLflow 跟踪进行检测
• 每次调用精确地发出一个跟踪

scorers

要应用的质量指标列表。 可以提供：

• 预定义的评分器
• 自定义评分器

有关更多详细信息，请参阅 记分器 。

model_id

用于将结果链接到应用版本的可选模型标识符（例如 "models:/my-app/1"）。 有关更多详细信息，请参阅 版本跟踪 。

数据格式

直接评估（模式 1）

答题卡评估（模式 2）

选项 A - 提供输入和输出：

选项 B - 提供现有跟踪：

常见数据输入模式

使用 MLflow 评估数据集进行评估（建议）

MLflow 评估数据集提供版本控制、世系跟踪和 Unity 目录集成，用于生产就绪评估。

import mlflow
from mlflow.genai.scorers import Correctness, Safety
from my_app import agent  # Your GenAI app with tracing

# Load versioned evaluation dataset
dataset = mlflow.genai.datasets.get_dataset("catalog.schema.eval_dataset_name")

# Run evaluation
results = mlflow.genai.evaluate(
    data=dataset,
    predict_fn=agent,
    scorers=[Correctness(), Safety()],
)

用于：

• 需要具有版本控制和世系跟踪的评估数据
• 轻松将跟踪转换为评估记录

请参阅 生成评估数据集 ，从跟踪或暂存创建数据集。

使用字典列表进行评估

使用简单的字典列表进行快速原型制作，而无需创建正式的评估数据集。

import mlflow
from mlflow.genai.scorers import Correctness, RelevanceToQuery
from my_app import agent  # Your GenAI app with tracing

# Define test data as a list of dictionaries
eval_data = [
    {
        "inputs": {"question": "What is MLflow?"},
        "expectations": {"expected_facts": ["open-source platform", "ML lifecycle management"]}
    },
    {
        "inputs": {"question": "How do I track experiments?"},
        "expectations": {"expected_facts": ["mlflow.start_run()", "log metrics", "log parameters"]}
    },
    {
        "inputs": {"question": "What are MLflow's main components?"},
        "expectations": {"expected_facts": ["Tracking", "Projects", "Models", "Registry"]}
    }
]

# Run evaluation
results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=agent,
    scorers=[Correctness(), RelevanceToQuery()],
)

用于：

• 快速原型制作
• 小型数据集（< 100 示例）
• 即席开发测试

对于生产，请转换为 MLflow 评估数据集。

使用 Pandas 数据帧进行评估

使用 Pandas 数据帧在处理 CSV 文件或现有数据科学工作流时进行评估。

import mlflow
import pandas as pd
from mlflow.genai.scorers import Correctness, Safety
from my_app import agent  # Your GenAI app with tracing

# Create evaluation data as a Pandas DataFrame
eval_df = pd.DataFrame([
    {
        "inputs": {"question": "What is MLflow?"},
        "expectations": {"expected_response": "MLflow is an open-source platform for ML lifecycle management"}
    },
    {
        "inputs": {"question": "How do I log metrics?"},
        "expectations": {"expected_response": "Use mlflow.log_metric() to log metrics"}
    }
])

# Run evaluation
results = mlflow.genai.evaluate(
    data=eval_df,
    predict_fn=agent,
    scorers=[Correctness(), Safety()],
)

用于：

• 快速原型制作
• 小型数据集（< 100 示例）
• 即席开发测试

使用 Spark 数据帧进行评估

使用 Spark 数据帧进行大规模评估，或在数据已存在于 Delta Lake/Unity Catalog 中时使用。

import mlflow
from mlflow.genai.scorers import Safety, RelevanceToQuery
from my_app import agent  # Your GenAI app with tracing

# Load evaluation data from a Delta table in Unity Catalog
eval_df = spark.table("catalog.schema.evaluation_data")

# Or load from any Spark-compatible source
# eval_df = spark.read.parquet("path/to/evaluation/data")

# Run evaluation
results = mlflow.genai.evaluate(
    data=eval_df,
    predict_fn=agent,
    scorers=[Safety(), RelevanceToQuery()],
)

用于：

• 数据已存在于 Delta Lake 或 Unity 目录中
• 如果需要在运行评估之前筛选 MLflow 评估数据集中的记录

注意： DataFrame 必须符合 评估数据集架构。

常见 predict_fn 模式

直接调用应用

当参数名称与评估数据集密钥匹配时，直接传递你的应用 predict_fn。

import mlflow
from mlflow.genai.scorers import RelevanceToQuery, Safety

# Your GenAI app that accepts 'question' as a parameter
@mlflow.trace
def my_chatbot_app(question: str) -> dict:
    # Your app logic here
    response = f"I can help you with: {question}"
    return {"response": response}

# Evaluation data with 'question' key matching the function parameter
eval_data = [
    {"inputs": {"question": "What is MLflow?"}},
    {"inputs": {"question": "How do I track experiments?"}}
]

# Pass your app directly since parameter names match
results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=my_chatbot_app,  # Direct reference, no wrapper needed
    scorers=[RelevanceToQuery(), Safety()]
)

用于：

• 具有与评估数据集匹配的参数名称的应用 inputs

将你的应用封装成一个可调用的对象

当您的应用程序需要不同于评估数据集 inputs 的参数名称或数据结构时，请进行包装。

import mlflow
from mlflow.genai.scorers import RelevanceToQuery, Safety

# Your existing GenAI app with different parameter names
@mlflow.trace
def customer_support_bot(user_message: str, chat_history: list = None) -> dict:
    # Your app logic here
    context = f"History: {chat_history}" if chat_history else "New conversation"
    return {
        "bot_response": f"Helping with: {user_message}. {context}",
        "confidence": 0.95
    }

# Wrapper function to translate evaluation data to your app's interface
def evaluate_support_bot(question: str, history: str = None) -> dict:
    # Convert evaluation dataset format to your app's expected format
    chat_history = history.split("|") if history else []

    # Call your app with the translated parameters
    result = customer_support_bot(
        user_message=question,
        chat_history=chat_history
    )

    # Translate output to standard format if needed
    return {
        "response": result["bot_response"],
        "confidence_score": result["confidence"]
    }

# Evaluation data with different key names
eval_data = [
    {"inputs": {"question": "Reset password", "history": "logged in|forgot email"}},
    {"inputs": {"question": "Track my order"}}
]

# Use the wrapper function for evaluation
results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=evaluate_support_bot,  # Wrapper handles translation
    scorers=[RelevanceToQuery(), Safety()]
)

用于：

• 应用参数与评估数据集 input 键之间的参数名称不匹配（例如， user_input vs question）
• 数据格式转换（字符串到列表，JSON 分析）

评估已部署的终结点

对于 Databricks 代理框架或模型服务终结点，使用 to_predict_fn 来创建兼容的预测函数。

import mlflow
from mlflow.genai.scorers import RelevanceToQuery

# Create predict function for your endpoint
predict_fn = mlflow.genai.to_predict_fn("endpoints:/my-chatbot-endpoint")

# Evaluate
results = mlflow.genai.evaluate(
    data=[{"inputs": {"question": "How does MLflow work?"}}],
    predict_fn=predict_fn,
    scorers=[RelevanceToQuery()]
)

效益： 自动从已启用跟踪的终结点中提取跟踪，以实现完全可观测性。

评估已记录的模型

将已经记录的 MLflow 模型进行包装，以在评估中的命名参数和模型的单参数接口之间进行转换。

大多数已记录的模型（例如使用 PyFunc 或记录格式如 LangChain）接受一个单一输入参数（例如对于 PyFunc 是 model_inputs），而 predict_fn 则需要命名参数，这些参数应与您的评估数据集中的键对应。

import mlflow
from mlflow.genai.scorers import Safety

# Make sure to load your logged model outside of the predict_fn so MLflow only loads it once!
model = mlflow.pyfunc.load_model("models:/chatbot/staging")

def evaluate_model(question: str) -> dict:
    return model.predict({"question": question})

results = mlflow.genai.evaluate(
    data=[{"inputs": {"question": "Tell me about MLflow"}}],
    predict_fn=evaluate_model,
    scorers=[Safety()]
)

后续步骤

• 评估应用 - 运行首次评估的分步指南
• 生成评估数据集 - 从生产日志或暂存创建结构化测试数据
• 定义自定义记分器 - 生成针对特定用例定制的指标