跳到主要内容

跟踪版本与环境

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

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

将此元数据附加到您的跟踪中，可提供以下关键洞察：

环境特定分析：比较 开发、预生产 和 生产 环境之间的行为

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

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

部署验证：确保不同部署目标之间行为一致

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

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

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

自动填充标签​

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

  • mlflow.source.name：生成跟踪的入口点或脚本（Python 脚本自动填充文件名，Jupyter 笔记本自动填充笔记本名称）
  • mlflow.source.git.commit：如果从 Git 仓库运行，则自动检测并填充提交哈希
  • mlflow.source.type：如果在 Jupyter 笔记本中运行则为 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（跟踪选项卡）中，使用搜索功能按上下文标签筛选跟踪

  • 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 查询跟踪：构建自定义分析和监控工作流

手动跟踪：通过上下文感知跨度添加详细的检测

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