跳到主要内容

跟踪版本和环境

在您的 GenAI 应用程序中跟踪环境、应用程序版本和自定义上下文信息，可以实现跨不同部署阶段、版本和业务特定维度的全面可观测性。MLflow 提供灵活的机制，可以使用标签（tags）为您的跟踪（traces）附加丰富的元数据。

为何跟踪环境和上下文？​

将这些元数据附加到您的跟踪中可以提供关键的洞察，用于：

特定环境的分析：比较 development、staging 和 production 环境中的行为

版本管理：跟踪不同应用程序版本（例如 v1.0.1、v1.2.0）的性能和回归情况

自定义分类：添加业务特定的上下文（例如 customer_tier: "premium"、feature_flag: "new_algorithm"）

部署验证：确保跨不同部署目标的行为一致性

根因分析：快速将问题缩小到特定的环境、版本或配置

用于上下文的标准和自定义标签​

MLflow 使用标签（键值对字符串）来存储跟踪上的上下文信息。

自动填充的标签​

这些标准标签会根据您的执行环境被 MLflow 自动捕获。

  • mlflow.source.name：生成跟踪的入口点或脚本（对于 Python 脚本，自动填充为文件名；对于 Jupyter notebooks，自动填充为 notebook 名称）
  • mlflow.source.git.commit：如果从 Git 仓库运行，会自动检测并填充提交哈希
  • mlflow.source.type：如果在 Jupyter notebook 中运行，则为 NOTEBOOK；如果在本地运行 Python 脚本，则为 LOCAL；否则为 UNKNOWN（自动检测）

如果需要，您可以使用 mlflow.update_current_trace() 或 mlflow.set_trace_tag() 手动覆盖这些自动填充的标签，以获得更精细地控制。

保留的标准标签​

一些标准标签具有特殊含义，但必须手动设置。

  • mlflow.trace.session：将来自多轮对话或用户会话的跟踪分组在一起
  • mlflow.trace.user：将跟踪与特定用户关联，以进行以用户为中心的分析

自定义标签​

您可以定义自定义标签来捕获任何业务特定或应用程序特定的上下文。常见的示例包括：

  • environment：例如 "production"、"staging"（来自 DEPLOY_ENV 环境变量）
  • app_version：例如 "1.0.0"（来自 APP_VERSION 环境变量）
  • deployment_id：例如 "deploy-abc-123"（来自 DEPLOYMENT_ID 环境变量）
  • region：例如 "us-east-1"（来自 REGION 环境变量）
  • 功能标志和 A/B 测试变体

基本实现​

以下是如何将各种类型的上下文作为标签添加到您的跟踪中：

import mlflow
import os
import platform


@mlflow.trace
def process_data_with_context(data: dict, app_config: dict):
    """Process data and add environment, version, and custom context."""

    current_env = os.getenv("APP_ENVIRONMENT", "development")
    current_app_version = app_config.get("version", "unknown")
    current_model_version = app_config.get("model_in_use", "gpt-3.5-turbo")

    # Define custom context tags
    context_tags = {
        "environment": current_env,
        "app_version": current_app_version,
        "model_version": current_model_version,
        "python_version": platform.python_version(),
        "operating_system": platform.system(),
        "data_source": data.get("source", "batch"),
        "processing_mode": "online" if current_env == "production" else "offline",
    }

    # Add tags to the current trace
    mlflow.update_current_trace(tags=context_tags)

    # Your application logic here...
    result = (
        f"Processed '{data['input']}' in {current_env} with app {current_app_version}"
    )

    return result


# Example usage
config = {"version": "1.1.0", "model_in_use": "claude-3-sonnet-20240229"}
input_data = {"input": "Summarize this document...", "source": "realtime_api"}

processed_result = process_data_with_context(input_data, config)
print(processed_result)

要点

  • 使用 os.getenv() 获取环境变量（例如 APP_ENVIRONMENT、APP_VERSION）
  • 将应用程序或模型配置传递到您跟踪的函数中
  • 使用 platform 模块获取系统信息
  • mlflow.update_current_trace() 将所有键值对添加到当前跟踪

查询和分析上下文数据​

使用 MLflow UI​

在 MLflow UI（Traces 选项卡）中，使用搜索功能按上下文标签过滤跟踪。

  • tags.environment = 'production'
  • tags.app_version = '2.1.0'
  • tags.model_used = 'advanced_model' AND tags.client_variant = 'treatment'
  • tags.feature_flag_new_ui = 'true'

您可以按标签对跟踪进行分组，以比较不同上下文下的性能或错误率。

程序化分析​

使用 MLflow SDK 进行更复杂的分析或与其他工具集成。

比较不同应用程序版本的错误率和性能。

import mlflow
from mlflow import MlflowClient


def compare_version_metrics(experiment_id: str, versions: list):
    """Compare error rates and performance across app versions"""

    version_metrics = {}

    for version in versions:
        traces = mlflow.search_traces(
            experiment_ids=[experiment_id],
            filter_string=f"tags.environment = 'production' AND tags.app_version = '{version}'",
        )

        if traces.empty:
            version_metrics[version] = {
                "error_rate": None,
                "avg_latency": None,
                "total_traces": 0,
            }
            continue

        # Calculate metrics
        error_count = len(traces[traces["status"] == "ERROR"])
        total_traces = len(traces)
        error_rate = (error_count / total_traces) * 100

        successful_traces = traces[traces["status"] == "OK"]
        avg_latency = (
            successful_traces["execution_time_ms"].mean()
            if not successful_traces.empty
            else 0
        )

        version_metrics[version] = {
            "error_rate": error_rate,
            "avg_latency": avg_latency,
            "total_traces": total_traces,
        }

    return version_metrics


# Usage
metrics = compare_version_metrics("1", ["1.0.0", "1.1.0", "1.2.0"])
for version, data in metrics.items():
    print(
        f"Version {version}: {data['error_rate']:.1f}% errors, {data['avg_latency']:.1f}ms avg latency"
    )

最佳实践​

标记策略​

标准化标签键：为您的自定义标签使用一致的命名约定（例如 snake_case）。

用于部署上下文的环境变量：在 CI/CD 或部署过程中设置的环境变量用于版本和环境信息。

自动化上下文附加：确保应用程序或部署脚本自动应用上下文标签。

权衡粒度和简洁性：捕获足够的上下文以进行有用的分析，但要避免过多的标记，以免跟踪难以管理。

性能考量​

最小化标签量：虽然添加标签的开销很小，但在高吞吐量系统中避免附加过多的标签。

使用简短的标签值：保持标签值简洁，以减少存储开销。

一致的标记：确保您的标记策略在所有服务和部署环境中都得到一致的应用。

安全和隐私​

避免敏感数据：请勿直接在标签中存储 PII 或敏感信息。

使用匿名标识符：在跟踪用户时，请使用匿名标识符而不是个人信息。

审查标签内容：定期审计您的标签，以确保它们不包含敏感信息。

后续步骤​

MLflow 跟踪 UI：学习使用 UI 按环境和版本过滤和分析跟踪。

搜索跟踪：掌握用于复杂基于上下文查询的高级搜索语法。

通过 SDK 查询跟踪：构建自定义分析和监控工作流。

手动跟踪：通过上下文感知的跨度（spans）添加详细的仪器。

通过实现全面的环境和版本跟踪，您可以为您的 GenAI 应用程序构建强大的可观测性，该可观测性可以从开发扩展到生产部署。