面向 GenAI 应用的生产监控

机器学习项目并非在首次发布后即告结束。持续监控和增量增强对于长期成功至关重要。MLflow Tracing 为您的生产应用程序提供全面的可观测性，支持持续改进的迭代过程，同时确保向用户高质量交付。

GenAI 应用面临独特的挑战，使得生产监控变得必不可少。由于模型更新、数据分布变化或新的用户交互模式，质量漂移可能随时间发生。涉及 LLM、向量数据库和检索系统的多步骤工作流的操作复杂性，会产生多个潜在的故障点，需要持续监督。成本管理变得至关重要，因为令牌使用量和 API 成本会根据用户行为和模型性能而显著变化。

关键监控区域

了解要监控什么有助于您专注于真正影响用户体验和业务成果的指标。与其尝试监控所有内容，不如专注于为您的特定应用和用户群提供可操作见解的领域。

操作指标
质量指标
业务影响

性能和可靠性：监控从用户请求到最终响应的端到端响应时间，包括 LLM 推理延迟、检索系统性能和组件级瓶颈。跟踪整体错误率、LLM API 故障、超时发生和依赖故障，以维护系统可靠性。

资源利用率：监控令牌消耗模式、API 成本跟踪、请求吞吐量和系统资源使用情况，以优化性能和控制成本。

业务指标：跟踪用户参与率、会话完成率、功能采用率和用户满意度分数，以了解应用程序的业务影响。

为生产端点设置追踪

将 GenAI 应用程序部署到生产环境时，您需要配置 MLflow Tracing 以将追踪发送到您的 MLflow 追踪服务器。此配置为所有生产可观测性功能奠定了基础。

专业提示：使用轻量级追踪 SDK

MLflow Tracing SDK mlflow-tracing 是一个轻量级软件包，仅包含最少量的依赖项，用于使用 MLflow Tracing 检测您的代码/模型/代理。

⚡️ 更快的部署：显着更小的软件包大小和更少的依赖项，可在容器和无服务器环境中实现更快的部署

🔧 简单的依赖管理：减少依赖项意味着更少的维护开销和更少的潜在冲突

📦 增强的便携性：通过最小的兼容性问题，轻松部署到不同的平台

🔒 改进的安全性：更少的依赖项导致更小的攻击面，从而降低安全风险

🚀 性能优化：针对生产环境中的高容量追踪进行了优化

兼容性警告

安装 MLflow Tracing SDK 时，请确保环境中未安装完整的 MLflow 软件包。在同一环境中同时拥有这两个软件包可能会导致冲突和意外行为。

环境变量配置

在您的生产环境中配置以下环境变量

# Required: Set MLflow Tracking URI
export MLFLOW_TRACKING_URI="http://your-mlflow-server:5000"

# Optional: Configure the experiment name for organizing traces
export MLFLOW_EXPERIMENT_NAME="production-genai-app"

# Optional: Configure async logging (recommended for production)
export MLFLOW_ENABLE_ASYNC_TRACE_LOGGING=true
export MLFLOW_ASYNC_TRACE_LOGGING_MAX_WORKERS=10
export MLFLOW_ASYNC_TRACE_LOGGING_MAX_QUEUE_SIZE=1000

自托管追踪服务器

您可以使用 MLflow 追踪服务器来存储生产追踪数据。但是，追踪服务器是为离线体验而优化的，通常不适合处理超大规模流量。对于高容量生产工作负载，请考虑使用 OpenTelemetry 与专用可观测性平台集成。

如果您选择在生产环境中使用追踪服务器，我们强烈建议

使用基于 SQL 的追踪服务器，其上运行可伸缩的数据库和工件存储
配置适当的索引，用于追踪表以获得更好的查询性能
设置定期删除，用于追踪数据管理
监控服务器性能并进行适当扩展

有关更多详细信息，请参阅追踪服务器设置指南。

性能考量

数据库：对于生产部署，请使用 PostgreSQL 或 MySQL，而不是 SQLite，以获得更好的并发写入性能。

存储：使用云存储（S3、Azure Blob、GCS）进行工件存储，以确保可伸缩性和可靠性。

索引：确保对 timestamp_ms、status 和频繁查询的标签列进行适当的索引，以随着追踪量的增长保持查询性能。

保留：实施数据保留策略，以管理存储成本并随时间推移维护系统性能。

Docker 部署示例

使用 Docker 部署时，通过容器配置传递环境变量

# Dockerfile
FROM python:3.9-slim

# Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy application code
COPY . /app
WORKDIR /app

# Set default environment variables (can be overridden at runtime)
ENV MLFLOW_TRACKING_URI=""
ENV MLFLOW_EXPERIMENT_NAME="production-genai-app"
ENV MLFLOW_ENABLE_ASYNC_TRACE_LOGGING=true

CMD ["python", "app.py"]

使用环境变量运行容器

docker run -d \
  -e MLFLOW_TRACKING_URI="http://your-mlflow-server:5000" \
  -e MLFLOW_EXPERIMENT_NAME="production-genai-app" \
  -e MLFLOW_ENABLE_ASYNC_TRACE_LOGGING=true \
  -e APP_VERSION="1.0.0" \
  your-app:latest

Kubernetes 部署示例

对于 Kubernetes 部署，使用 ConfigMaps 和 Secrets

# configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: mlflow-config
data:
  MLFLOW_TRACKING_URI: 'http://mlflow-server:5000'
  MLFLOW_EXPERIMENT_NAME: 'production-genai-app'
  MLFLOW_ENABLE_ASYNC_TRACE_LOGGING: 'true'

---
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: genai-app
spec:
  template:
    spec:
      containers:
        - name: app
          image: your-app:latest
          envFrom:
            - configMapRef:
                name: mlflow-config
          env:
            - name: APP_VERSION
              value: '1.0.0'

OpenTelemetry 集成

MLflow 生成的追踪与 OpenTelemetry 追踪规范兼容。因此，MLflow 追踪可以导出到支持 OpenTelemetry 的各种可观测性平台。

默认情况下，MLflow 将追踪导出到 MLflow Tracking Server。要启用将追踪导出到 OpenTelemetry Collector，请在启动任何追踪之前设置 OTEL_EXPORTER_OTLP_ENDPOINT 环境变量

pip install opentelemetry-exporter-otlp

import mlflow
import os

# Set the endpoint of the OpenTelemetry Collector
os.environ["OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"] = "https://:4317/v1/traces"
# Optionally, set the service name to group traces
os.environ["OTEL_SERVICE_NAME"] = "your-service-name"

# Trace will be exported to the OTel collector
with mlflow.start_span(name="foo") as span:
    span.set_inputs({"a": 1})
    span.set_outputs({"b": 2})

支持的可观测性平台

单击以下图标了解如何为您的特定可观测性平台设置 OpenTelemetry Collector

OpenTelemetry 配置

MLflow 使用标准 OTLP Exporter 将追踪导出到 OpenTelemetry Collector 实例。您可以使用 OpenTelemetry 支持的所有配置

export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT="https://:4317/v1/traces"
export OTEL_EXPORTER_OTLP_TRACES_PROTOCOL="http/protobuf"
export OTEL_EXPORTER_OTLP_TRACES_HEADERS="api_key=12345"

追踪导出

MLflow 仅将追踪导出到单个目的地。当配置了 OTEL_EXPORTER_OTLP_ENDPOINT 环境变量时，MLflow 将不会将追踪导出到 MLflow Tracking Server，您将无法在 MLflow UI 中看到追踪。

使用 Databricks 进行托管监控

Databricks 还提供了一个托管解决方案，用于监控与 MLflow Tracing 集成的 GenAI 应用程序。

Monitoring Hero

功能包括

跟踪操作指标，例如请求量、延迟、错误和成本。
使用托管评估监控质量指标，例如正确性、安全性、上下文充分性等。
使用 Python 函数配置自定义指标。
通过查看 MLflow Tracing 记录的追踪进行根本原因分析。
支持托管在 Databricks 之外的应用程序

生产监控配置

异步追踪日志记录

对于生产应用程序，MLflow 默认异步记录追踪以防止阻塞您的应用程序

环境变量	描述	默认值
`MLFLOW_ENABLE_ASYNC_TRACE_LOGGING`	是否异步记录追踪。设置为 `False` 时，追踪将以阻塞方式记录。	`True`
`MLFLOW_ASYNC_TRACE_LOGGING_MAX_WORKERS`	每个进程用于异步追踪日志记录的最大工作线程数。增加此值可提高追踪日志记录的吞吐量，但也会增加 CPU 使用率和内存消耗。	`10`
`MLFLOW_ASYNC_TRACE_LOGGING_MAX_QUEUE_SIZE`	在工作线程将追踪记录到后端之前，可排队的最大追踪数量。当队列已满时，新追踪将被丢弃。增加此值可提高追踪日志记录的持久性，但也会增加内存消耗。	`1000`
`MLFLOW_ASYNC_TRACE_LOGGING_RETRY_TIMEOUT`	重试失败追踪日志记录的超时时间（秒）。当追踪日志记录失败时，它将在此超时时间内以退避方式重试，之后将被丢弃。	`500`

高容量应用程序的示例配置

export MLFLOW_ENABLE_ASYNC_TRACE_LOGGING=true
export MLFLOW_ASYNC_TRACE_LOGGING_MAX_WORKERS=20
export MLFLOW_ASYNC_TRACE_LOGGING_MAX_QUEUE_SIZE=2000
export MLFLOW_ASYNC_TRACE_LOGGING_RETRY_TIMEOUT=600

为生产追踪添加上下文

在生产环境中，使用上下文信息丰富追踪对于理解用户行为、调试问题和改进 GenAI 应用程序至关重要。此上下文使您能够分析用户交互，跨不同细分市场跟踪质量，并识别导致更好或更差结果的模式。

跟踪请求、会话和用户上下文

生产应用程序需要同时跟踪多个上下文。这是一个全面的示例，展示了如何在 FastAPI 应用程序中跟踪所有这些

import mlflow
import os
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel

# Initialize FastAPI app
app = FastAPI()


class ChatRequest(BaseModel):
    message: str


@app.post("/chat")  # FastAPI decorator should be outermost
@mlflow.trace  # Ensure @mlflow.trace is the inner decorator
def handle_chat(request: Request, chat_request: ChatRequest):
    # Retrieve all context from request headers
    client_request_id = request.headers.get("X-Request-ID")
    session_id = request.headers.get("X-Session-ID")
    user_id = request.headers.get("X-User-ID")

    # Update the current trace with all context and environment metadata
    mlflow.update_current_trace(
        client_request_id=client_request_id,
        tags={
            # Session context - groups traces from multi-turn conversations
            "mlflow.trace.session": session_id,
            # User context - associates traces with specific users
            "mlflow.trace.user": user_id,
            # Environment metadata - tracks deployment context
            "environment": "production",
            "app_version": os.getenv("APP_VERSION", "1.0.0"),
            "deployment_id": os.getenv("DEPLOYMENT_ID", "unknown"),
            "region": os.getenv("REGION", "us-east-1"),
        },
    )

    # Your application logic for processing the chat message
    response_text = f"Processed message: '{chat_request.message}'"

    return {"response": response_text}

反馈收集

捕获用户对特定交互的反馈对于理解质量和改进 GenAI 应用程序至关重要

import mlflow
from mlflow.client import MlflowClient
from fastapi import FastAPI, Query, Request
from pydantic import BaseModel
from typing import Optional
from mlflow.entities import AssessmentSource

app = FastAPI()


class FeedbackRequest(BaseModel):
    is_correct: bool  # True for correct, False for incorrect
    comment: Optional[str] = None


@app.post("/chat_feedback")
def handle_chat_feedback(
    request: Request,
    client_request_id: str = Query(
        ..., description="The client request ID from the original chat request"
    ),
    feedback: FeedbackRequest = ...,
):
    """
    Collect user feedback for a specific chat interaction identified by client_request_id.
    """
    # Search for the trace with the matching client_request_id
    client = MlflowClient()
    experiment = client.get_experiment_by_name("production-genai-app")
    traces = client.search_traces(experiment_ids=[experiment.experiment_id])
    traces = [
        trace for trace in traces if trace.info.client_request_id == client_request_id
    ][:1]

    if not traces:
        return {
            "status": "error",
            "message": f"Unable to find data for client request ID: {client_request_id}",
        }, 500

    # Log feedback using MLflow's log_feedback API
    mlflow.log_feedback(
        trace_id=traces[0].info.trace_id,
        name="response_is_correct",
        value=feedback.is_correct,
        source=AssessmentSource(
            source_type="HUMAN", source_id=request.headers.get("X-User-ID")
        ),
        rationale=feedback.comment,
    )

    return {
        "status": "success",
        "message": "Feedback recorded successfully",
        "trace_id": traces[0].info.trace_id,
    }

查询带上下文的追踪

使用上下文信息分析生产行为

import mlflow

# Query traces by user
user_traces = mlflow.search_traces(
    experiment_ids=[experiment.experiment_id],
    filter_string="tags.`mlflow.trace.user` = 'user-jane-doe-12345'",
    max_results=100,
)

# Query traces by session
session_traces = mlflow.search_traces(
    experiment_ids=[experiment.experiment_id],
    filter_string="tags.`mlflow.trace.session` = 'session-def-456'",
    max_results=100,
)

# Query traces by environment
production_traces = mlflow.search_traces(
    experiment_ids=[experiment.experiment_id],
    filter_string="tags.environment = 'production'",
    max_results=100,
)

总结

使用 MLflow Tracing 进行生产监控为您的 GenAI 应用程序提供全面的可观测性。了解用户如何实际与您的应用程序交互，在真实条件下监控质量和性能，以及跟踪 GenAI 举措的业务影响，这些对于长期成功都至关重要。

成功进行生产部署的关键建议包括：为生产部署使用 mlflow-tracing 以最小化依赖项并优化性能；为高容量应用程序配置异步日志记录以防止阻塞；使用标签和元数据添加丰富的上下文以进行有效调试和分析；实施反馈收集以进行质量监控和持续改进；考虑 OpenTelemetry 集成以用于企业可观测性平台；以及在实施适当的错误处理的同时监控性能。

无论您是使用自托管 MLflow、通过 OpenTelemetry 与企业可观测性平台集成，还是利用 Databricks Mosaic AI 的高级功能，MLflow Tracing 都为理解和改进您的生产 GenAI 应用程序提供了基础。

后续步骤

使用追踪调试和观察您的应用：学习基础可观测性概念和技术

通过 SDK 查询追踪：了解如何以编程方式访问追踪数据进行分析

跟踪用户和会话：实施用户和会话上下文跟踪以获得更好的监控洞察

关键监控区域​

为生产端点设置追踪​

专业提示：使用轻量级追踪 SDK​

环境变量配置​

自托管追踪服务器​

性能考量​

Docker 部署示例​

Kubernetes 部署示例​

OpenTelemetry 集成​

支持的可观测性平台​

OpenTelemetry 配置​

使用 Databricks 进行托管监控​

生产监控配置​

异步追踪日志记录​

为生产追踪添加上下文​

跟踪请求、会话和用户上下文​

反馈收集​

查询带上下文的追踪​

总结​

后续步骤​