建立生产监控

重要

此功能在 Beta 版中。

生产监控可让你自动对来自代理的追踪运行 MLflow 3 评分器，以持续评估其质量。为一个 MLflow 实验安排评分器，监控服务会评估传入的跟踪数据的可配置样本。结果作为评估反馈附加到每个跟踪记录。

生产监视包括以下内容：

使用内置或自定义评分器进行自动化质量评估，包括用于评估整个对话的多轮次评委。
可配置的采样率，以便控制覆盖率与计算成本之间的权衡。
在开发和生产中使用相同的评分器来确保一致的评估。
持续质量评估，后台运行监控。

注释

MLflow 3 生产监控与来自 MLflow 2 的跟踪记录兼容。

先决条件

在设置生产监视之前，请确保具备：

MLflow 试验：记录跟踪的 MLflow 试验。如果未指定试验，则使用活动试验。
带仪器的生产应用：您的代理必须使用 MLflow 追踪来记录痕迹。请参阅生产跟踪指南。
定义的记分器：已测试过可与您的应用程序跟踪格式一同使用的记分器。如果你在开发过程中将生产应用用作 predict_fn 中的 mlflow.genai.evaluate()，那么你的评分器可能已经兼容。
SQL 仓库 ID（适用于 Unity Catalog 跟踪）：如果跟踪存储在 Unity Catalog 中，则必须配置 SQL 仓库 ID，以使监控功能正常工作。

为 Unity Catalog 追踪配置 SQL 仓库

如果你的追踪存储在 Unity 目录中，监控作业会通过 SQL 仓库对 Unity 目录表进行评分查询。在注册评分员之前，先在实验中配置仓库ID，否则监控作业会因错误提示 mlflow.monitoring.sqlWarehouseId 标签丢失而失败。

用 set_databricks_monitoring_sql_warehouse_id(). 设置 SQL 仓库 ID 该辅助工具将 ID 存储在 mlflow.monitoring.sqlWarehouseId 实验标签中，监控作业从该标签读取：

from mlflow.tracing import set_databricks_monitoring_sql_warehouse_id

# Set the SQL warehouse ID for monitoring
set_databricks_monitoring_sql_warehouse_id(
    sql_warehouse_id="<SQL_WAREHOUSE_ID>",
    experiment_id="<EXPERIMENT_ID>"  # Optional, uses active experiment if not specified
)

注释

在笔记本或应用中设置 MLFLOW_TRACING_SQL_WAREHOUSE_ID 环境变量并不能替代。它只适用于你设置的过程。监控作业单独运行，读取实验标签中的仓库ID。使用 set_databricks_monitoring_sql_warehouse_id() 时，仓库ID会在实验中保持保留。

配置 Unity 目录跟踪监控需要以下工作区级权限：

CAN USE 在 SQL 仓库中。
CAN EDIT 关于MLflow实验。
对监控作业的权限（注册第一个评分器时自动授予）。

监视作业在首次在试验中注册记分器的用户的标识下运行。此用户的权限确定监视作业可以访问的内容。

开始

监控会在你启动评分器的那一刻开始。为你的实验注册一个评分器，然后使用抽样配置按 .register() 和 .start() 模式启动它：

from mlflow.genai.scorers import Safety, ScorerSamplingConfig

# Register and start a built-in judge
safety_judge = Safety().register(name="safety")
safety_judge = safety_judge.start(sampling_config=ScorerSamplingConfig(sample_rate=0.7))

这种两步模式适用于任何评分者类型：内置裁判、自定义裁判、基于代码的记分员和多回合裁判。

有关记分员和裁判类型，请参见记分员和裁判。

注释

在任何给定时间，最多 20 名评分者都可以与持续质量监视的实验相关联。

查看结果

计划记分器后，允许 15-20 分钟进行初始处理。然后：

进入您的 MLflow 实验。
打开 Traces 选项卡，查看附加到追踪的评估。
使用监视仪表板跟踪质量趋势。

对于多轮对话评审，评估会附在每个会话中的第一个轨迹上。有关详细信息，请参阅评估的存储方式。

最佳做法

采样策略

对于安全性和安保检查等关键评分标准，请使用 sample_rate=1.0。
对于昂贵的评分系统，例如复杂的 LLM 模型，应使用较低的采样率（0.05-0.2）。
若要在开发期间进行迭代改进，请使用中等速率（0.3-0.5）。

平衡覆盖范围与成本，如以下示例所示：

# High-priority scorers: higher sampling
safety_judge = Safety().register(name="safety")
safety_judge = safety_judge.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))  # 100% coverage for critical safety

# Expensive scorers: lower sampling
complex_scorer = ComplexCustomScorer().register(name="complex_analysis")
complex_scorer = complex_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.05))  # 5% for expensive operations

筛选跟踪

在filter_string中使用ScorerSamplingConfig参数以控制评分器评估哪些跟踪。此使用与mlflow.search_traces()相同的筛选器语法。

from mlflow.genai.scorers import Safety, ScorerSamplingConfig

# Only evaluate traces that completed successfully
safety_judge = Safety().register(name="safety")
safety_judge = safety_judge.start(
    sampling_config=ScorerSamplingConfig(
        sample_rate=1.0,
        filter_string="attributes.status = 'OK'"
    ),
)

可以组合多个条件：

import time

# Evaluate successful traces from the last 24 hours
one_day_ago = int((time.time() - 86400) * 1000)
safety_judge = safety_judge.start(
    sampling_config=ScorerSamplingConfig(
        sample_rate=0.5,
        filter_string=f"attributes.status = 'OK' AND attributes.timestamp_ms > {one_day_ago}"
    ),
)

自定义记分器设计

使自定义评分器保持独立，如以下示例所示：

@scorer
def well_designed_scorer(inputs, outputs):
    # All imports inside the function
    import re
    import json

    # Handle missing data gracefully
    response = outputs.get("response", "")
    if not response:
        return 0.0

    # Return consistent types
    return float(len(response) > 100)

Troubleshooting

记分器未运行

如果评分人员没有执行任务，请检查以下内容：

检查实验：确保将日志记录到实验，而不是记录到单个运行。
采样率：采样率较低，可能需要时间才能查看结果。
验证筛选器字符串：确保 filter_string 匹配实际跟踪信息。

序列化问题

用于生产监视的自定义记分器已序列化，以便监视服务远程执行这些评分器。这会施加多个约束：

笔记本要求：必须从 Databricks 笔记本定义和注册自定义 @scorer 函数。序列化机制依赖于笔记本环境。
自包含函数：所有导入都必须在函数体内内内联。在序列化期间不会捕获对在函数外部定义的外部变量、模块或对象的引用。
没有基于类的评分器：只能注册@scorer 基于修饰器的评分器。无法序列化基于 Scorer 类的子类以便进行远程执行。
不需要导入的类型提示：函数签名中的类型提示如果需要导入语句（例如，从 List 导入 typing），会导致序列化失败。

创建自定义评分器时，在函数定义中包含导入。

# Avoid external dependencies
import external_library  # Outside function

@scorer
def bad_scorer(outputs):
    return external_library.process(outputs)

# Include imports in the function definition
@scorer
def good_scorer(outputs):
    import json  # Inside function
    return len(json.dumps(outputs))

# Avoid using type hints in scorer function signature that requires imports
from typing import List

@scorer
def scorer_with_bad_types(outputs: List[str]):
    return False

# Class-based scorers are not supported for production monitoring
class MyScorer(Scorer):
    name: str = "my_scorer"
    def __call__(self, outputs):
        return len(outputs) > 10

下一步：管理生产评分员

其他资源

基于代码的评分器 - 生成根据需求定制的评分器。
评估对话 - 了解多轮次对话评估和多轮次评委。
生成 MLflow 评估数据集 - 使用监视结果提高质量。

参考指南

管理生产评分器 ——用于计分器生命周期管理的API参考。
记分器和 LLM 评委 - 了解电源监视的指标。
评估你的代理 ——线下评估如何与生产相关。

上次更新时间为 2026-10-26