评估工具

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

快速参考

有关详细信息,请参阅 mlflow.genai.evaluate()

参数 类型 Description
data MLflow EvaluationDataset、List[Dict]、Pandas DataFrame、Spark DataFrame 测试数据
predict_fn Callable 你的应用(仅限直接评估)
scorers List[Scorer] 质量指标
model_id str 可选版本跟踪

工作原理

  1. 在测试输入上运行应用,捕获跟踪数据

  2. 应用评分器 来评估质量,创建 反馈

  3. 将结果存储在评估执行中。

先决条件

  1. 安装 MLflow 和所需的包。

    pip install --upgrade "mlflow[databricks]>=3.1.0" openai "databricks-connect>=16.1"
    
  2. 请按照设置环境快速入门创建 MLflow 试验。

评估模式

有两种评估模式:

  • (推荐) 直接评估。 MLflow 直接调用应用以生成用于评估的跟踪信息。
  • 答题卡评估。 您提供预先计算的输出或现有的跟踪以进行评估。

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

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

评估如何与跟踪配合使用

以下代码演示了如何运行评估的示例:

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()]
)

然后,可以在 UI 中查看结果:

评估结果

答案表评估

当无法直接运行 GenAI 应用时,可以提供现有的跟踪或预计算输出进行评估。 示例用例包括测试外部系统的输出、评估历史跟踪以及比较不同平台的输出。

重要

如果您使用的答案表中的跟踪信息与生产环境不同,则可能需要重新编写记分器函数,以便用于生产监视

如何使用答案表进行评估

使用输入和输出的示例

以下代码演示了如何运行评估的示例:

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()]
)

然后,可以在 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 DataFrame 或 Spark DataFrame 的列表。

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

领域 数据类型 Description 直接评估配合使用 答案表一起使用
inputs dict[Any, Any] 一个使用 dict 传递给 predict_fn**kwargs。 必须是 JSON 可序列化的。 每个键必须对应于predict_fn中的命名参数。 必选 要么inputs + outputs,要么trace是必需的。 不能同时传递这两者。
未提供时派生自 trace
outputs dict[Any, Any] 含对应 dict 的 GenAI 应用输出的 input。 必须是 JSON 可序列化的。 不得提供,由 MLflow 根据 Trace 生成。 要么inputs + outputs,要么trace是必需的。 不能同时传递这两者。
未提供时派生自 trace
expectations dict[str, Any] 含与 dict 对应的基本事实标签的 inputscorers 用于质量检查。 必须是 JSON 可序列化的,并且每个密钥必须是一个 str 可选 可选
trace mlflow.entities.Trace 请求的跟踪对象。 如果提供了trace,那么expectations可以在Assessments上以trace的形式提供,而不是作为一个单独的列。 不得提供,由 MLflow 根据 Trace 生成。 要么inputs + outputs,要么trace是必需的。 不能同时传递这两者。

predict_fn

GenAI 应用的入口点。 此参数仅用于 直接计算predict_fn 必须满足以下要求:

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

scorers

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

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

model_id

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

数据格式

用于 直接评估

领域 必选 Description
inputs 传递给 predict_fn 的字典
expectations 可选 评分器的可选基本事实

适用于答题卡评估

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

领域 必选 Description
inputs GenAI 应用的原始输入
outputs 应用的预计算输出
expectations 可选 评分器的可选基本事实

选项 B - 提供现有跟踪:

领域 必选 Description
trace 具有输入/输出的 MLflow 跟踪对象
expectations 可选 评分器的可选基本事实

常见数据输入模式

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()],
)

若要从跟踪或从头开始创建数据集,请参阅 生成评估数据集

使用字典列表进行评估

使用简单的字典列表进行快速原型制作,而无需创建正式的评估数据集。 这对于快速原型制作、小型数据集(少于 100 个示例)和非正式开发测试非常有用。

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()],
)

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

使用 Pandas 数据帧进行评估

使用 Pandas 数据帧在处理 CSV 文件或现有数据科学工作流时进行评估。 这对于快速原型制作、小型数据集(少于 100 个示例)和非正式开发测试非常有用。

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()],
)

使用 Spark 数据帧进行评估

使用 Spark 数据帧进行大规模评估,或者当数据已在 Delta Lake 或 Unity 目录中时。 当 Delta Lake 或 Unity 目录中已存在数据时,或者在运行评估之前需要筛选 MLflow 评估数据集中的记录时,这非常有用。

DataFrame 必须符合 评估数据集架构

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()],
)

常见 predict_fn 模式

直接调用应用

当参数名称与评估数据集密钥匹配时,直接将应用作为 predict_fn 传递。 这对于具有与评估数据集中匹配 inputs 的参数名称的应用非常有用。

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不同的参数名称或数据结构,请将其包装在可调用函数中。 当您的应用程序参数与评估数据集 input 键(例如,user_inputquestion)之间存在参数名称不匹配,或者需要进行数据格式转换(例如,将字符串转换为列表,或进行 JSON 解析)时,这会非常有用。

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()]
)

评估已部署的终结点

使用该 to_predict_fn 函数评估马赛克 AI 代理框架、模型服务聊天终结点和自定义终结点。

此函数创建一个与这些终结点兼容的预测函数,并自动从已启用跟踪的终结点中提取跟踪以实现完全可观测性。

注释

to_predict_fn 函数直接执行 kwargs 的透传至终结点。 评估数据必须与终结点所需的输入格式匹配。 如果格式不匹配,则评估失败,并显示有关无法识别的输入键的错误消息。

模型部署聊天

模型服务聊天终结点需要使用 messages 密钥格式化的数据。

import mlflow
from mlflow.genai.scorers import RelevanceToQuery

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

# Evaluate the chat endpoint
results = mlflow.genai.evaluate(
    data=[{"inputs": {"messages": [{"role": "user", "content": "How does MLflow work?"}]}}],
    predict_fn=predict_fn,
    scorers=[RelevanceToQuery()]
)

代理框架

代理框架终结点可以具有不同的输入接口。 以下示例显示了一个 input 键:

import mlflow
from mlflow.genai.scorers import RelevanceToQuery

# Create a predict function for a Knowledge Assistant agent endpoint
predict_fn = mlflow.genai.to_predict_fn("endpoints:/ka-56a301ab-endpoint")

# Evaluate the agent endpoint
results = mlflow.genai.evaluate(
    data=[{"inputs": {"input": [{"role": "user", "content": "How do I use the Models from Code feature in MLflow?"}]}}],
    predict_fn=predict_fn,
    scorers=[RelevanceToQuery()]
)

自定义终结点

自定义终结点可能具有完全不同的访问模式,用于向它们提交数据。 确保 data 输入格式与用于评估的终结点兼容。

如果评估数据格式与端点不兼容,请对模型的接口进行封装。 转换层可以确保将正确的有效负载提交到评估终结点。

import mlflow
from mlflow.genai.scorers import RelevanceToQuery

def custom_predict_fn(inputs):
    # Transform inputs to match your endpoint's expected format
    # For example, if your endpoint expects a 'query' key instead of 'messages'
    transformed_inputs = {
        "query": inputs["messages"][0]["content"],
        "context": inputs.get("context", "")
    }

    # Call your endpoint with the transformed data
    original_predict_fn = mlflow.genai.to_predict_fn("endpoints:/my-custom-endpoint")
    return original_predict_fn(transformed_inputs)

# Use your wrapper function for evaluation
results = mlflow.genai.evaluate(
    data=[{"inputs": {"messages": [{"role": "user", "content": "What is machine learning?"}], "context": "technical documentation"}}],
    predict_fn=custom_predict_fn,
    scorers=[RelevanceToQuery()]
)

评估记录的模型

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

大多数已记录的模型(例如使用 PyFunc 或日志变体,如 LangChain)接受一个单一输入参数(例如 model_inputs 对于 PyFunc),而 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()]
)

后续步骤