GenAI 应用的生产监控

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

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

关键监控领域

了解要监控的内容可以帮助您专注于真正影响用户体验和业务成果的指标。与其试图监控一切，不如关注能够为您的特定应用程序和用户群提供可操作见解的领域。

运营指标
质量指标
业务影响

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

资源利用率：监控 token 消耗模式、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

# Optional: Configure trace sampling ratio (default is 1.0)
export MLFLOW_TRACE_SAMPLING_RATIO=0.1

自托管跟踪服务器

您可以使用 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 环境变量。您还可以启用双向导出，同时将跟踪发送到 MLflow 和 OpenTelemetry。

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})

指标导出

当配置了指标端点时，MLflow 可以导出 OpenTelemetry 指标。这使您能够在兼容的监控系统中监控 span 持续时间和其他与跟踪相关的指标。

先决条件：必须安装 opentelemetry-exporter-otlp 库才能启用指标导出。

pip install opentelemetry-exporter-otlp

启用指标导出

配置 OpenTelemetry 指标端点:

# For OpenTelemetry Collector (gRPC endpoint)
export OTEL_EXPORTER_OTLP_METRICS_ENDPOINT="https://:4317"
export OTEL_EXPORTER_OTLP_METRICS_PROTOCOL="grpc"

# OR for OpenTelemetry Collector (HTTP endpoint)
export OTEL_EXPORTER_OTLP_METRICS_ENDPOINT="https://:4318/v1/metrics"
export OTEL_EXPORTER_OTLP_METRICS_PROTOCOL="http/protobuf"

直接 Prometheus 导出

Prometheus 可以直接接收 MLflow 导出的 OpenTelemetry 指标。

# Configure MLflow to send metrics directly to Prometheus
export OTEL_EXPORTER_OTLP_METRICS_ENDPOINT="https://:9090/api/v1/otlp/v1/metrics"
export OTEL_EXPORTER_OTLP_METRICS_PROTOCOL="http/protobuf"

Prometheus 配置：使用 --web.enable-otlp-receiver 和 --enable-feature=otlp-deltatocumulative 标志启动 Prometheus，以直接接受 OTLP 指标。

导出的指标

启用后，MLflow 将导出以下 OpenTelemetry 直方图指标：

mlflow.trace.span.duration：一个直方图，测量 span 执行持续时间（毫秒）
- 单位：ms（毫秒）
- 标签/属性:
  - root：对于根 span 为 "true"，对于子 span 为 "false"
  - span_type：span 的类型（例如，“LLM”、“CHAIN”、“AGENT”或“unknown”）
  - span_status：span 的状态（“OK”、“ERROR”或“UNSET”）
  - experiment_id：与跟踪关联的 MLflow 实验 ID
  - tags.*：所有跟踪标签（例如，tags.mlflow.traceName、tags.mlflow.evalRequestId）
  - metadata.*：所有跟踪元数据（例如，metadata.mlflow.sourceRun、metadata.mlflow.modelId、metadata.mlflow.trace.tokenUsage）

此直方图使您能够分析：

不同 span 类型的响应时间分布
根 span 和子 span 之间的性能差异
按 span 状态（“ERROR”）对错误率进行监控
按 MLflow 实验分组的性能指标
按跟踪标签（例如，tags.mlflow.traceName、tags.mlflow.evalRequestId）分段的性能指标
按模型 ID 或源运行的性能分析（例如，metadata.mlflow.modelId、metadata.mlflow.sourceRun）
服务性能随时间变化的趋势

完整示例

import mlflow
import os

# Enable metrics collection and export
os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = "https://:4317"
os.environ["OTEL_EXPORTER_OTLP_METRICS_PROTOCOL"] = "grpc"

# Metrics will be exported to OpenTelemetry Collector
with mlflow.start_span(name="process_request", span_type="CHAIN") as span:
    span.set_inputs({"query": "What is MLflow?"})
    # Your application logic here
    span.set_outputs({"response": "MLflow is an open source platform..."})

支持的可观测性平台

点击以下图标，了解如何为您的特定可观测性平台设置 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"

双向导出

默认情况下，配置 OTLP 导出时，MLflow 只将跟踪发送到 OpenTelemetry Collector。要同时将跟踪发送到 MLflow Tracking Server 和 OpenTelemetry Collector，请设置 MLFLOW_TRACE_ENABLE_OTLP_DUAL_EXPORT=true。

import mlflow
import os

# Enable dual export
os.environ["MLFLOW_TRACE_ENABLE_OTLP_DUAL_EXPORT"] = "true"
os.environ["OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"] = "https://:4317/v1/traces"

# Configure MLflow tracking
mlflow.set_tracking_uri("https://:5000")
mlflow.set_experiment("my-experiment")

# Traces will be sent to both MLflow and OTel collector
with mlflow.start_span(name="foo") as span:
    span.set_inputs({"a": 1})
    span.set_outputs({"b": 2})

跟踪导出行为

默认（MLFLOW_TRACE_ENABLE_OTLP_DUAL_EXPORT=false）：跟踪仅发送到 OpenTelemetry Collector。
双向导出（MLFLOW_TRACE_ENABLE_OTLP_DUAL_EXPORT=true）：跟踪将发送到 MLflow Tracking Server 和 OpenTelemetry Collector。

Databricks 托管监控

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

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

采样跟踪

对于高吞吐量应用程序，您可能希望减少导出到后端的跟踪数量。您可以配置采样率来控制导出的跟踪数量。

环境变量	描述	默认值
`MLFLOW_TRACE_SAMPLING_RATIO`	跟踪的采样率。设置为 `0.0` 时，将不导出任何跟踪。设置为 `1.0` 时，将导出所有跟踪。	`1.0`

默认值为 1.0，表示所有跟踪都将被导出。当设置为小于 1.0 时，例如 0.1，将只导出 10% 的跟踪。采样是在跟踪级别进行的，这意味着某些跟踪中的所有 span 都将一起导出或丢弃。

为生产跟踪添加上下文

在生产环境中，用上下文信息丰富跟踪对于理解用户行为、调试问题和改进您的 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 集成​

指标导出​

直接 Prometheus 导出​

导出的指标​

完整示例​

支持的可观测性平台​

OpenTelemetry 配置​

双向导出​

Databricks 托管监控​

生产监控配置​

异步跟踪日志记录​

采样跟踪​

为生产跟踪添加上下文​

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

反馈收集​

使用上下文查询跟踪​

总结​

后续步骤​