跟踪用户和会话

Traces with session IDs

在您的生成式AI（GenAI）应用程序中跟踪用户和会话，为理解用户行为、分析对话流程以及提升个性化提供了重要背景信息。MLflow 提供内置支持，可将跟踪与用户关联并将其分组到会话中。

为何要跟踪用户和会话？

用户和会话跟踪为您的生成式AI应用程序提供了强大的分析和改进能力。了解不同用户如何与您的应用程序交互有助于识别整个用户群体的使用模式，而分析多轮对话则能揭示您的系统在交互中保持上下文的能力。

此跟踪功能使您能够监控不同用户群体的性能指标，从而识别需要有针对性改进的领域。您还可以在同一对话中的多次交互中保持会话连续性，通过改进上下文感知能力来创造更好的用户体验。

标准 MLflow 元数据字段

MLflow 3 新功能

用户和会话跟踪仅在 MLflow 3 及更高版本中可用。要升级，请运行 pip install --upgrade mlflow

MLflow 提供两个用于会话和用户跟踪的标准元数据字段

mlflow.trace.user - 将跟踪与特定用户关联
mlflow.trace.session - 将属于多轮对话的跟踪进行分组

当您使用这些标准元数据字段时，MLflow 会自动在 UI 中启用筛选和分组功能。与标签不同，元数据一旦被记录就无法更新，这使其非常适合用户和会话ID等不可变标识符。

实现示例

基本用法
Web 应用程序

以下是如何为您的应用程序添加用户和会话跟踪：

import mlflow


@mlflow.trace
def chat_completion(user_id: str, session_id: str, message: str):
    """Process a chat message with user and session tracking."""

    # Add user and session context to the current trace
    mlflow.update_current_trace(
        metadata={
            "mlflow.trace.user": user_id,  # Links trace to specific user
            "mlflow.trace.session": session_id,  # Groups trace with conversation
        }
    )

    # Your chat logic here
    response = generate_response(message)
    return response


# Example usage
def handle_user_message(request):
    return chat_completion(
        user_id=request.user_id,  # e.g., "user-123"
        session_id=request.session_id,  # e.g., "session-abc-456"
        message=request.message,
    )

关键实现注意事项

@mlflow.trace 装饰器会自动为函数执行创建跟踪
mlflow.update_current_trace() 将用户 ID 和会话 ID 添加到活动跟踪
在整个应用程序中使用一致的 ID 格式以确保可靠的分析

在生产环境的 Web 应用程序中，从请求头中提取用户和会话信息

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

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):
    # Extract context from headers
    session_id = request.headers.get("X-Session-ID")
    user_id = request.headers.get("X-User-ID")

    # Update trace with user and session context
    mlflow.update_current_trace(
        metadata={
            "mlflow.trace.session": session_id,
            "mlflow.trace.user": user_id,
        }
    )

    # Process the chat message
    response_text = process_chat_message(chat_request.message, user_id, session_id)
    return {"response": response_text}

请求示例

curl -X POST "http://127.0.0.1:8000/chat" \
     -H "Content-Type: application/json" \
     -H "X-Session-ID: session-def-456" \
     -H "X-User-ID: user-jane-doe-12345" \
     -d '{"message": "What is my account balance?"}'

查询与分析

MLflow UI 搜索
用户分析
会话分析

在 MLflow UI 中使用这些搜索查询来筛选跟踪

# Find all traces for a specific user
metadata.`mlflow.trace.user` = 'user-123'

# Find all traces in a session
metadata.`mlflow.trace.session` = 'session-abc-456'

# Find traces for a user within a specific session
metadata.`mlflow.trace.user` = 'user-123' AND metadata.`mlflow.trace.session` = 'session-abc-456'

当您使用标准用户和会话元数据时，MLflow UI 提供强大的筛选和分组功能。

以编程方式分析用户行为模式

import mlflow


def analyze_user_behavior(user_id: str, experiment_id: str):
    """Analyze a specific user's interaction patterns."""

    # Search for all traces from a specific user
    user_traces = mlflow.search_traces(
        experiment_ids=[experiment_id],
        filter_string=f"metadata.`mlflow.trace.user` = '{user_id}'",
        return_type="list",
    )

    if len(user_traces) == 0:
        return {"error": "No traces found for user"}

    # Calculate key metrics
    total_interactions = len(user_traces)
    unique_sessions = len(
        set(
            t.info.request_metadata.get("mlflow.trace.session", "") for t in user_traces
        )
    )

    successful_traces = [trace for trace in user_traces if trace.info.state == "OK"]
    avg_response_time = (
        sum(trace.info.execution_time_ms for trace in successful_traces)
        / len(successful_traces)
        if successful_traces
        else 0
    )

    return {
        "user_id": user_id,
        "total_interactions": total_interactions,
        "unique_sessions": unique_sessions,
        "avg_response_time_ms": avg_response_time,
    }


# Usage
user_stats = analyze_user_behavior("user-123", "1")
print(
    f"User has {user_stats['total_interactions']} interactions across {user_stats['unique_sessions']} sessions"
)

分析会话内的对话流程

def analyze_session_flow(session_id: str, experiment_id: str):
    """Analyze conversation flow within a session."""

    # Get all traces from a session, ordered chronologically
    session_traces = mlflow.search_traces(
        experiment_ids=[experiment_id],
        filter_string=f"metadata.`mlflow.trace.session` = '{session_id}'",
        order_by=["timestamp_ms ASC"],
        return_type="list",
    )

    if len(session_traces) == 0:
        return {"error": "No traces found for session"}

    # Build conversation timeline
    conversation_turns = []
    for i, trace in enumerate(session_traces):
        conversation_turns.append(
            {
                "turn": i + 1,
                "timestamp": trace.info.timestamp_ms,
                "duration_ms": trace.info.execution_time_ms,
                "status": trace.info.status,
            }
        )

    # Session-level metrics
    session_duration = max(trace.info.timestamp_ms for trace in session_traces) - min(
        trace.info.timestamp_ms for trace in session_traces
    )
    total_turns = len(conversation_turns)
    avg_turn_duration = (
        sum(trace.info.execution_time_ms for trace in session_traces) / total_turns
    )

    return {
        "session_id": session_id,
        "total_turns": total_turns,
        "session_duration_ms": session_duration,
        "avg_turn_duration_ms": avg_turn_duration,
        "conversation_flow": conversation_turns,
    }


# Usage
session_analysis = analyze_session_flow(session_id, exp.experiment_id)
print(
    f"Session had {session_analysis['total_turns']} turns over {session_analysis['session_duration_ms']}ms"
)

最佳实践与集成

实施指南

成功的用户和会话跟踪需要在整个应用程序中采用一致的 ID 格式，以确保可靠的分析和筛选。明确定义会话何时开始和结束的规则，例如在一段时间不活动后或用户明确结束对话时。考虑通过用户细分或会话类型等额外上下文来丰富元数据，以增强分析能力。将用户和会话数据与请求 ID 结合起来，可以为您的系统提供完整的可追溯性。

与 MLflow 功能集成

用户和会话跟踪显著增强了 MLflow 的其他功能。您可以在跟踪搜索中使用用户和会话过滤器进行有针对性的分析和调试。当比较不同用户群体的指标以识别改进机会时，质量评估变得更加强大。环境跟踪与用户上下文结合良好，可实现全面的可观测性，而生产监控则能够根据实时环境中的用户群组跟踪性能模式。

总结

MLflow 的用户和会话跟踪提供了强大的功能，可用于理解用户行为并优化用户体验。这种方法通过深入了解应用程序中的用户行为模式和会话动态来提供增强的分析能力。它通过提供用户和会话上下文来改进调试，有助于更快地识别和解决用户特有的问题。

当您能够跟踪不同用户群体的性能时，质量监控将变得更有针对性，从而在最关键的领域实现有重点的改进。通过多轮对话跟踪进行对话分析有助于优化对话系统和用户流程，以提高参与度。

标准 MLflow 元数据 (mlflow.trace.user 和 mlflow.trace.session) 与自定义上下文的结合，为生成式AI应用程序中的以用户为中心的可观测性奠定了全面的基础。

后续步骤

跟踪环境与上下文: 为您的跟踪添加部署和版本上下文

搜索跟踪: 掌握用户和会话分析的高级筛选技术

质量分析: 使用用户和会话数据来提高应用程序质量

生产监控: 通过用户上下文设置全面的生产可观测性

为何要跟踪用户和会话？​

标准 MLflow 元数据字段​

实现示例​

查询与分析​

最佳实践与集成​

实施指南​

与 MLflow 功能集成​

总结​

后续步骤​