向跟踪添加元数据和用户反馈

2025-09-05

MLflow 跟踪允许生产应用使用其他元数据和上下文（例如请求 ID、会话和用户 ID 以及自定义标记）来增强跟踪。应用还可以记录用户反馈以及跟踪。然后，可以使用此元数据来组织、分析和调试跟踪。

将元数据添加到跟踪

基本跟踪工作后，将元数据或上下文添加到跟踪，以便更好地调试和见解。生产应用程序需要同时跟踪多个上下文片段：用于调试的客户端请求 ID、用于多轮次对话的会话 ID、用于个性化和分析的用户 ID，以及用于作见解的环境元数据。 MLflow 具有以下标准化标记和属性，用于捕获重要的上下文信息：

Metadata	用例	MLflow 属性或标记
客户端请求 ID	将跟踪链接到用于端到端调试的特定客户端请求或 API 调用	`TraceInfo` 参数 `client_request_id`
用户会话 ID	从多轮次对话对跟踪进行分组，使你能够分析完整的聊天流	`mlflow.trace.session` 标记
用户 ID	将跟踪与特定用户相关联，以便进行个性化、队列分析和用户特定的调试	`mlflow.trace.user` 标记
环境数据	跟踪部署上下文（环境、版本、区域），以便跨不同部署进行作见解和调试	自动标记和常见自定义标记
自定义标记	添加自定义元数据，尤其是用于组织、搜索和筛选跟踪	（标记）

下面是一个全面示例，演示如何跟踪 FastAPI 应用程序中的所有这些内容。

import mlflow
import os
from fastapi import FastAPI, Request, HTTPException # HTTPException might be needed depending on full app logic
from pydantic import BaseModel

# Initialize FastAPI app
app = FastAPI()

class ChatRequest(BaseModel):
    message: str

@mlflow.trace # Ensure @mlflow.trace is the outermost decorator
@app.post("/chat") # FastAPI decorator should be 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
    # The @mlflow.trace decorator ensures an active trace is available
    mlflow.update_current_trace(
        client_request_id=client_request_id,
        metadata={
            # 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,
            # Override automatically populated environment metadata
            "mlflow.source.type": os.getenv("APP_ENVIRONMENT", "development"),  # Override default LOCAL/NOTEBOOK
            # Add customer environment metadata
            "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"),
            # Add custom tags
            "my_custom_tag": "custom tag value",
        }
    )

    # --- Your application logic for processing the chat message ---
    # For example, calling a language model with context
    # response_text = my_llm_call(
    #     message=chat_request.message,
    #     session_id=session_id,
    #     user_id=user_id
    # )
    response_text = f"Processed message: '{chat_request.message}'"
    # --- End of application logic ---

    # Return response
    return {
        "response": response_text
    }

# To run this example (requires uvicorn and fastapi):
# uvicorn your_file_name:app --reload
#
# Example curl request with context headers:
# curl -X POST "http://127.0.0.1:8000/chat" \
#      -H "Content-Type: application/json" \
#      -H "X-Request-ID: req-abc-123-xyz-789" \
#      -H "X-Session-ID: session-def-456-uvw-012" \
#      -H "X-User-ID: user-jane-doe-12345" \
#      -d '{"message": "What is my account balance?"}'

有关向跟踪添加上下文的详细信息，请参阅：

跟踪用户和会话
跟踪版本和环境
附加自定义标记和元数据
mlflow.update_current_trace() 用于添加元数据的 API
自动填充标记和保留标准标记列表的 MLflow 跟踪文档

收集用户反馈

捕获有关特定交互的用户反馈对于了解质量和改进 GenAI 应用程序至关重要。在添加元数据到跟踪中显示的客户端请求 ID 跟踪的基础上，下面的 FastAPI 示例演示了如何：

使用客户端请求 ID 查找确切的跟踪并附加反馈，将反馈链接到特定交互。
使用 API log_feedbacklog_expectation，以创建在 MLflow UI 中可见的结构化反馈对象。
通过查询跟踪及其关联的反馈来分析质量模式，以确定哪种类型的交互收到正面或负面评级。

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

# Initialize FastAPI app
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()
    # Get the experiment by name (using Databricks workspace path)
    experiment = client.get_experiment_by_name("/Shared/production-app")
    traces = client.search_traces(
        experiment_ids=[experiment.experiment_id],
        filter_string=f"attributes.client_request_id = '{client_request_id}'",
        max_results=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
    feedback = 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 feedback

# Example usage:
# After a chat interaction returns a response, the client can submit feedback:
#
# curl -X POST "http://127.0.0.1:8000/chat_feedback?client_request_id=req-abc-123-xyz-789" \
#      -H "Content-Type: application/json" \
#      -H "X-User-ID: user-jane-doe-12345" \
#      -d '{
#        "is_correct": true,
#        "comment": "The response was accurate and helpful"
#      }'

生产跟踪反馈

有关记录用户反馈的详细信息，请参阅：

收集用户反馈
mlflow.log_feedback() 用于存储结构化反馈的 API 和 log_expectation API

使用元数据查询跟踪

将元数据添加到跟踪后，可以使用该上下文信息分析生产行为。具体而言，该方法 MLflowClient.search_traces() 允许按标记和元数据进行筛选。以下示例查找特定用户和特定用户会话的跟踪。

import mlflow
from mlflow.client import MlflowClient
import pandas as pd

client = MlflowClient()
experiment = client.get_experiment_by_name("/Shared/production-app")

# Query traces by user
user_traces = client.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 = client.search_traces(
    experiment_ids=[experiment.experiment_id],
    filter_string="tags.`mlflow.trace.session` = 'session-123'",
    max_results=100
)

有关许多示例用例 mlflow.search_traces()，请参阅 “搜索和分析跟踪”。

后续步骤

特性参考

有关本指南中的概念和功能的详细信息，请参阅：

跟踪数据模型 - 了解跟踪、范围和属性
日志记录评估 - 了解反馈的存储和使用方式
搜索和分析跟踪 - 查看许多常见用例的示例查询 mlflow.search_traces()

Compartir a través de