トレースの充実化: タグ、コンテキスト、フィードバック

エージェントがトレースを出力するようにインストゥルメントした後は、検索、デバッグ、品質モニタリングで役立つ追加の情報でそれらのトレースを充実させることができます。

タグとメタデータ — キーと値のペアで、トレースを整理、フィルタリング、アノテーションします。
コンテキスト — コホート分析およびデプロイ固有のデバッグのための、ユーザー ID、セッション ID、環境、エージェントのバージョン。
エンドユーザーフィードバック — トレースに対する評価とコメントとしてキャプチャされ、本番運用からのグラウンドトゥルース品質シグナルを提供します。

主要な本番運用パターンの一つは、エンドユーザーからのフィードバックです。誰かが親指を上または下にクリックしたとき、またはデプロイされたアプリにコメントを残したときは、そのインタラクションのトレースに対する評価としてログに記録します。発生元の実行にフィードバックを添付したままにすることで、特定のリクエストのデバッグや、実際の成功と失敗からの評価データセットの構築など、ダウンストリームですぐに役立つようになります。

要件

環境に適したパッケージを選択します:

Production
Development

Bash
pip install --upgrade mlflow-tracing

mlflow-tracing パッケージの依存関係は最小限に抑えられており、本番運用向けに最適化されています。

Bash
pip install --upgrade "mlflow[databricks]>=3.1.0" openai "databricks-connect>=16.1"

環境の設定に従って、MLflowエクスペリメントを作成します。

タグとメタデータ

タグは、トレースがログに記録された後を含め、いつでも設定、更新、または削除できる変更可能なキーと値のペアです。動的な情報（レビューステータス、データ品質ラベル、ユーザーフィードバックシグナルなど）にはタグを使用します。

トレースが記録されると、 メタデータ は変更できなくなります。モデルバージョン、環境、構成など、実行時にキャプチャされた変更されないファクトにはメタデータを使用します。

API	使用する場合
`mlflow.update_current_trace`	実行中にアクティブなトレースにタグまたはメタデータを設定する
`mlflow.set_trace_tag`	完了したトレースのタグを設定または更新する
`mlflow.delete_trace_tag`	完了したトレースからタグを削除する
MLflow の UI	完了したトレースのタグを対話形式で設定または更新する

API	使用する場合
`mlflow.update_current_trace`	実行中にアクティブなトレースにタグまたはメタデータを設定する
`mlflow.set_trace_tag`	完了したトレースのタグを設定または更新する
`mlflow.delete_trace_tag`	完了したトレースからタグを削除する
MLflow の UI	完了したトレースのタグを対話形式で設定または更新する

実行時のタグとメタデータの設定

トレースがアクティブな間にタグやメタデータをアタッチするには、トレースされた関数内で mlflow.update_current_trace を呼び出します。

Python
import mlflow

@mlflow.trace
def my_func(x):
    mlflow.update_current_trace(
        metadata={&quot;model_version&quot;: &quot;v1.2.3&quot;, &quot;environment&quot;: &quot;production&quot;},
        tags={&quot;fruit&quot;: &quot;apple&quot;}
    )
    return x + 1

my_func(10)

注記

update_current_trace タグの新しいキーを追加するか、既存のキーを上書きします。 メタデータ の場合、既存のキーを更新しようとしてもサイレントに無視されます。メタデータは一度設定すると変更できません。

完了したトレースにタグを設定する

トレースがLogsに記録された後にタグを更新または削除するには：

Python
import mlflow

@mlflow.trace
def process_data(data):
    return data.upper()

result = process_data("hello world")

trace_id = mlflow.get_last_active_trace_id()

mlflow.set_trace_tag(trace_id=trace_id, key="review_status", value="approved")
mlflow.set_trace_tag(trace_id=trace_id, key="data_quality", value="high")
mlflow.delete_trace_tag(trace_id=trace_id, key="data_quality")

UI でタグを設定する

トレースに移動し、任意のタグの横にある鉛筆アイコンをクリックして、編集または削除します。

トレースのタグの更新

トレースにコンテキストを追加する

Context links traces to users, sessions, deployments, and code — enabling multi-turn conversation grouping, user cohort analysis, and environment-specific デバッグ.

トレース対象のエージェントロジック内で mlflow.update_current_trace を呼び出してコンテキストをアタッチします:

Python
import mlflow

mlflow.update_current_trace(
    metadata={
        &quot;mlflow.trace.user&quot;: user_id,
        &quot;mlflow.trace.session&quot;: session_id,
    },
    tags={
        &quot;query_category&quot;: &quot;chat&quot;,
    },
)

ログ記録後、 mlflow.search_traces() （返された DataFrame の metadata および tags 列）、または Trace.info.trace_metadata と Trace.info.tags を介して Trace オブジェクトから直接コンテキストにアクセスします。

完全な実行例については、トレースのエンリッチ: タグ、コンテキスト、およびフィードバックを参照してください。

標準コンテキストフィールド

MLflow は、最も一般的なコンテキストタイプに対して標準化されたメタデータフィールドを定義します。これらを使用すると、UIはそれらのフィールドによるフィルタリングとグループ化を自動的に有効にします。

コンテキストタイプ	MLflow フィールド	ユースケース
ユーザーID	`mlflow.trace.user`	パーソナライズ、コホート分析、およびユーザー固有のデバッグのために、トレースを特定のユーザーに関連付ける
セッションID	`mlflow.trace.session`	マルチターン会話のトレースをグループ化して、会話のフロー全体を分析する
クライアントリクエスト ID	`client_request_id` 曜日： `TraceInfo`	Link エンドツーエンドのデバッグのために、トレースを上流の API 呼び出しにリンクします
環境 / バージョン	`mlflow.source.type` + カスタムメタデータ	環境とエージェントのバージョンにわたるデプロイコンテキストの追跡
カスタムフィールド	（メタデータキー）	エージェント固有のコンテキスト (デプロイ ID、リージョン、機能フラグなど)

コンテキストタイプ	MLflow フィールド	ユースケース
ユーザーID	`mlflow.trace.user`	パーソナライズ、コホート分析、およびユーザー固有のデバッグのために、トレースを特定のユーザーに関連付ける
セッションID	`mlflow.trace.session`	マルチターン会話のトレースをグループ化して、会話のフロー全体を分析する
クライアントリクエスト ID	`client_request_id` 曜日： `TraceInfo`	Link エンドツーエンドのデバッグのために、トレースを上流の API 呼び出しにリンクします
環境 / バージョン	`mlflow.source.type` + カスタムメタデータ	環境とエージェントのバージョンにわたるデプロイコンテキストの追跡
カスタムフィールド	（メタデータキー）	エージェント固有のコンテキスト (デプロイ ID、リージョン、機能フラグなど)

自動入力されるフィールド

MLflow は、実行環境からいくつかのメタデータフィールドを自動的に設定します。defaultの検知要件を満たさない場合は、mlflow.update_current_trace を使用していずれかをオーバーライドできます。

メタデータフィールド	説明	自動設定元
`mlflow.source.name`	エントリーポイントまたはスクリプト名	Pythonファイル名、Databricksノートブック名
`mlflow.source.git.commit`	Git commit ハッシュ	現在のGitリポジトリ
`mlflow.source.git.branch`	Git Branch 名	現在のGitリポジトリ
`mlflow.source.git.repoURL`	GitレポジトリのURL	現在のGitリポジトリ
`mlflow.source.type`	実行環境	`NOTEBOOK` (Jupyter/Databricks)、`LOCAL` (Python スクリプト)、それ以外の場合は `UNKNOWN`
`mlflow.sourceRun`	ソースのランID	Active MLflow ラン
`metadata.mlflow.modelId`	MLflow LoggedModel ID	`MLFLOW_ACTIVE_MODEL_ID` 環境変数または `mlflow.set_active_model()`

メタデータフィールド	説明	自動設定元
`mlflow.source.name`	エントリーポイントまたはスクリプト名	Pythonファイル名、Databricksノートブック名
`mlflow.source.git.commit`	Git commit ハッシュ	現在のGitリポジトリ
`mlflow.source.git.branch`	Git Branch 名	現在のGitリポジトリ
`mlflow.source.git.repoURL`	GitレポジトリのURL	現在のGitリポジトリ
`mlflow.source.type`	実行環境	`NOTEBOOK` (Jupyter/Databricks)、`LOCAL` (Python スクリプト)、それ以外の場合は `UNKNOWN`
`mlflow.sourceRun`	ソースのランID	Active MLflow ラン
`metadata.mlflow.modelId`	MLflow LoggedModel ID	`MLFLOW_ACTIVE_MODEL_ID` 環境変数または `mlflow.set_active_model()`

環境やバージョンなどのデプロイメタデータの場合は、ハードコーディングするのではなく、環境変数から値を取得します。

Python
import mlflow
import os

mlflow.update_current_trace(
    metadata={
        &quot;mlflow.source.type&quot;: os.getenv(&quot;APP_ENVIRONMENT&quot;, &quot;development&quot;),
    }
)

ベストプラクティス

一貫したID形式 — エージェント全体でユーザーIDおよびセッションIDに標準化された形式を使用します。
セッションの境界 — セッションの起動および終了のタイミングに関する明確なルールを定義します。
環境変数 — ハードコーディングされた値ではなく、環境変数からメタデータを設定します。
コンテキストタイプの結合 — ユーザー、セッション、および環境のコンテキストを一緒に追跡します。
定期的な分析 — ダッシュボードを設定して、ユーザーの行動、セッションのパターン、バージョンのパフォーマンスをモニターします。
Override default thoughtfully — 自動検出しきい値がデプロイメントに適していない場合にのみ、自動的に設定されたメタデータを上書きします。

ユーザーからのフィードバックを収集する

エンドユーザーからのフィードバックは、エージェントの実環境における品質を示すグラウンドトゥルースシグナルを提供します。MLflow は、フィードバックを評価として取得します。これはトレースに永続的に添付される構造化されたエンティティであり、そのためすべての評価が、それを促した正確なインタラクションに関連付けられたままになります。

トレースの評価

フィードバックの種類

フィードバックの種類	説明	一般的なユースケース
Binary	高評価/低評価または正解/不正解	クイック満足度シグナル
数値	スケールによる評価（たとえば 1–5 つ星など）	詳細な品質評価
分類別	多肢選択式のオプション	課題または応答タイプの分類
テキスト	自由形式のコメント	詳細なユーザー説明

フィードバックの種類	説明	一般的なユースケース
Binary	高評価/低評価または正解/不正解	クイック満足度シグナル
数値	スケールによる評価（たとえば 1–5 つ星など）	詳細な品質評価
分類別	多肢選択式のオプション	課題または応答タイプの分類
テキスト	自由形式のコメント	詳細なユーザー説明

フィードバックデータモデル

ユーザーフィードバックは、トレースまたはスパンにアタッチされた フィードバック エンティティ（アセスメントの一種）として取得されます。各フィードバックエンティティには以下が格納されます。

値 — フィードバック信号（ブール値、数値、テキスト、または構造化データ）
ソース — フィードバックを提供したユーザーを特定する AssessmentSource (詳細は下記を参照)
根拠 — フィードバックに関する任意の解説
Metadata — Timestampやカスタム属性などの追加のコンテキスト

AssessmentSource フィールド

すべてのフィードバック評価上の AssessmentSource オブジェクトは、フィードバックの起源を識別します。

source_type — エンドユーザーフィードバック用 "HUMAN"、自動評価用 "LLM_JUDGE"
source_id — フィードバックを提供した特定のユーザーまたはシステム（たとえば、ユーザーIDの文字列やジャッジ識別子）

mlflow.log_feedback を呼び出すときは、両方のフィールドを渡します。

Python
from mlflow.entities import AssessmentSource

mlflow.log_feedback(
    trace_id=trace_id,
    name="user_feedback",
    value=True,
    source=AssessmentSource(source_type="HUMAN", source_id=user_id),
    rationale="The answer was accurate and helpful.",
)

Link feedbackをトレースにLinkする

フィードバックをログに記録するには、ユーザーのレスポンスを特定のトレースに関連付ける必要があります。2つのアプローチ：

アプローチ 1 — MLflow トレース ID の使用 (よりシンプル): リクエスト中に MLflow で生成されたトレース ID を取得し、それをクライアントに返します。クライアントは、フィードバックとともにそれを送り返します。

アプローチ 2 — クライアントリクエスト ID の使用 （より高度な制御）：リクエストごとに独自の固有 ID を生成し、それをトレースタグとしてアタッチしてから、フィードバックの到着時にそのタグでトレースを検索します。すでにリクエスト追跡システムがある場合に便利です。

警告

エージェントをDatabricks Model Serving Endpointにデプロイする場合は、client_request_idを（属性ではなく）タグとして設定してください。メタデータ属性としてupdate_current_trace(client_request_id=...)を使用すると、サービング環境でのトレースのエクスポートが機能しなくなります。Model Servingを使用する必要がある場合は、アプローチ1（MLflowトレースID）を優先するか、update_current_trace(tags={"client_request_id": ...})経由でclient_request_idを設定してください。

Approach 1: MLflow trace ID
Approach 2: Client request ID

バックエンド

Python
import mlflow
from fastapi import FastAPI, Query
from mlflow.entities import AssessmentSource
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class ChatRequest(BaseModel):
    message: str

class ChatResponse(BaseModel):
    response: str
    trace_id: str  # Return the trace ID so the client can reference it for feedback

@app.post("/chat", response_model=ChatResponse)
def chat(request: ChatRequest):
    response = process_message(request.message)  # Your agent logic here
    trace_id = mlflow.get_current_active_span().trace_id
    return ChatResponse(response=response, trace_id=trace_id)

class FeedbackRequest(BaseModel):
    is_correct: bool
    comment: Optional[str] = None

@app.post("/feedback")
def submit_feedback(
    trace_id: str = Query(..., description="Trace ID from the chat response"),
    feedback: FeedbackRequest = ...,
    user_id: Optional[str] = Query(None)
):
    mlflow.log_feedback(
        trace_id=trace_id,
        name="user_feedback",
        value=feedback.is_correct,
        source=AssessmentSource(source_type="HUMAN", source_id=user_id),
        rationale=feedback.comment
    )
    return {"status": "success", "trace_id": trace_id}

フロントエンド (React)

JavaScript
import React, { useState } from 'react';

function ChatWithFeedback() {
  const [message, setMessage] = useState('');
  const [response, setResponse] = useState('');
  const [traceId, setTraceId] = useState(null);
  const [feedbackSubmitted, setFeedbackSubmitted] = useState(false);

  const sendMessage = async () => {
    const res = await fetch('/chat', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ message }),
    });
    const data = await res.json();
    setResponse(data.response);
    setTraceId(data.trace_id);
    setFeedbackSubmitted(false);
  };

  const submitFeedback = async (isCorrect, comment = null) => {
    if (!traceId || feedbackSubmitted) return;
    const params = new URLSearchParams({ trace_id: traceId });
    await fetch(`/feedback?${params}`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ is_correct: isCorrect, comment }),
    });
    setFeedbackSubmitted(true);
  };

  return (
    <div>
      <input value={message} onChange={(e) =&gt; setMessage(e.target.value)} placeholder="Ask a question..." />
      <button onClick={sendMessage}>Send</button>
      {response && (
        <div>
          <p>{response}</p>
          <div>
            <button onClick={() =&gt; submitFeedback(true)} disabled={feedbackSubmitted}>
              👍
            </button>
            <button onClick={() =&gt; submitFeedback(false)} disabled={feedbackSubmitted}>
              👎
            </button>
          </div>
          {feedbackSubmitted && Thanks for your feedback!</span>}
        </div>
      )}
    </div>
  );
}

リクエスト中にカスタムIDをタグとして添付し、フィードバックの到着時にそのタグでトレースを検索します。

バックエンド

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

app = FastAPI()

class ChatRequest(BaseModel):
    message: str

class ChatResponse(BaseModel):
    response: str
    client_request_id: str

@app.post("/chat", response_model=ChatResponse)
def chat(request: ChatRequest):
    client_request_id = f"req-{uuid.uuid4().hex[:8]}"
    # Must be a tag, not an attribute — required for Model Serving compatibility
    mlflow.update_current_trace(tags={&quot;client_request_id&quot;: client_request_id})
    response = process_message(request.message)
    return ChatResponse(response=response, client_request_id=client_request_id)

class FeedbackRequest(BaseModel):
    is_correct: bool
    comment: Optional[str] = None

@app.post("/feedback")
def submit_feedback(
    request: Request,
    client_request_id: str = Query(..., description="Request ID from the original interaction"),
    feedback: FeedbackRequest = ...
):
    client = MlflowClient()
    experiment = client.get_experiment_by_name("/Shared/production-app")
    traces = client.search_traces(
        experiment_ids=[experiment.experiment_id],
        filter_string=f"tags.client_request_id = '{client_request_id}'",
        max_results=1
    )
    if not traces:
        return {"status": "error", "message": "Unexpected error: request not found"}, 500

    mlflow.log_feedback(
        trace_id=traces[0].info.trace_id,
        name="user_feedback",
        value=feedback.is_correct,
        source=AssessmentSource(
            source_type="HUMAN",
            source_id=request.headers.get("X-User-ID")
        ),
        rationale=feedback.comment
    )
    return {"status": "success", "trace_id": traces[0].info.trace_id}

フロントエンドはアプローチ 1 を反映しています — メッセージを送信し、会話ターンごとに返された client_request_id を保存し、フィードバック送信時に毎回それを返します。

多次元フィードバック

単一のトレースに複数の名前付きアセスメントをLogsして、個別の品質ディメンションをキャプチャします。

Python
from mlflow.entities import AssessmentSource

@app.post("/detailed-feedback")
def submit_detailed_feedback(
    trace_id: str,
    accuracy: int = Query(..., ge=1, le=5, description="Accuracy rating 1–5"),
    helpfulness: int = Query(..., ge=1, le=5, description="Helpfulness rating 1–5"),
    relevance: int = Query(..., ge=1, le=5, description="Relevance rating 1–5"),
    user_id: str = Query(...),
    comment: Optional[str] = None
):
    dimensions = {"accuracy": accuracy, "helpfulness": helpfulness, "relevance": relevance}
    for dimension, score in dimensions.items():
        mlflow.log_feedback(
            trace_id=trace_id,
            name=f"user_{dimension}",
            value=score / 5.0,  # Normalize to 0–1 scale
            source=AssessmentSource(source_type="HUMAN", source_id=user_id),
            rationale=comment if dimension == "accuracy" else None
        )
    return {"status": "success", "trace_id": trace_id, "feedback_recorded": dimensions}

ストリーミング応答

ストリーミング（SSE または WebSockets）を使用する場合、ストリームが完了するまでトレース ID は利用できません。最終的なストリームイベントとして返し、それが到着するまでフィードバックコントロールを無効にします。

バックエンド (FastAPI SSE)

Python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import mlflow, json, asyncio
from typing import AsyncGenerator

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate() -> AsyncGenerator[str, None]:
        try:
            with mlflow.start_span(name="streaming_chat") as span:
                full_response = ""
                async for token in your_llm_stream_function(request.message):
                    full_response += token
                    yield f"data: {json.dumps({'type': 'token', 'content': token})}\n\n"
                    await asyncio.sleep(0.01)  # Prevent overwhelming the client
                span.set_attribute("response", full_response)
                span.set_attribute("token_count", len(full_response.split()))
                # Send trace ID as the final event
                yield f"data: {json.dumps({'type': 'done', 'trace_id': span.trace_id})}\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'error': str(e)})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            &quot;Cache-Control&quot;: &quot;no-cache&quot;,
            &quot;Connection&quot;: &quot;keep-alive&quot;,
            &quot;X-Accel-Buffering&quot;: &quot;no&quot;,  # Disable proxy buffering
        },
    )

フロントエンド側でストリームを読み取り、token個のイベントをレスポンステキストに蓄積し、最後のdoneイベントからトレースIDをキャプチャします。フィードバックコントロールを有効にする条件として traceId && !isStreaming を使用します。

主な実装上の注意点：

トレース ID はストリーミングの完了後にのみ利用可能になるため、到着するまでフィードバックコントロールを無効にするように UI を設計してください。
コンテンツのトークン、完了イベント、エラーを区別するために、 type フィールドを持つ一貫したイベント形式を使用します。
X-Accel-Buffering: no を設定してプロキシのバッファリングを無効にします。
部分的な SSE メッセージを処理するために、フロントエンドにラインバッファリングを実装します。
失敗がトレースに記録されユーザーに表示されるように、エラーイベントをストリームに含めます。

フィードバックの分析

任意のトレースを開いて MLflow UI でフィードバックを表示します。評価はスパンデータと一緒に表示されます。

トレース評価 UI

ユーザーフィードバックのトレース

プログラムによるフィードバックのクエリーと集計:

Python
from mlflow.client import MlflowClient
from datetime import datetime, timedelta

def analyze_user_feedback(experiment_name: str, hours: int = 24):
    client = MlflowClient()
    cutoff_ms = int((datetime.now() - timedelta(hours=hours)).timestamp() * 1000)
    traces = client.search_traces(
        experiment_names=[experiment_name],
        filter_string=f"trace.timestamp_ms > {cutoff_ms}"
    )

    total = len(traces)
    with_feedback = positive = negative = 0

    for trace in traces:
        detail = client.get_trace(trace.info.trace_id)
        if detail.data.assessments:
            with_feedback += 1
            for a in detail.data.assessments:
                if a.name == "user_feedback":
                    if a.value:
                        positive += 1
                    else:
                        negative += 1

    feedback_rate = (with_feedback / total * 100) if total else 0
    positive_rate = (positive / with_feedback * 100) if with_feedback else 0
    print(f"Feedback rate: {feedback_rate:.1f}%  Positive: {positive_rate:.1f}%")
    print(f"Total feedback: {with_feedback} of {total} traces")

analyze_user_feedback("/Shared/production-genai-agent")

同じパターンが多次元フィードバックにも拡張されます。各トレースの評価を反復処理し、a.value を a.name でグループ化して、評価の各ディメンションを個別に平均化します。

その他のリソース

トレースのエンリッチ：タグ、コンテキスト、フィードバック - 完全なチュートリアル：ユーザー、セッション、環境、およびバージョンのコンテキストをトレースに追加する
トレースへのプログラムによるアクセス - タグとメタデータを使用してトレースをフィルター処理および検索します
トレース全体の問題の検索 - トレースのアナリティクスの例
MLflow 評価データセットの構築 - 収集したフィードバックを使用して評価データセットを構築します
本番運用モニタリングのセットアップ - フィードバックに基づく品質モニタリングの監視

次のステップ： OpenTelemetry トレースを Unity Catalog に保存する

要件​

タグとメタデータ​

実行時のタグとメタデータの設定​

完了したトレースにタグを設定する​

UI でタグを設定する​

トレースにコンテキストを追加する​

標準コンテキストフィールド​

自動入力されるフィールド​

ベストプラクティス​

ユーザーからのフィードバックを収集する​

フィードバックの種類​

フィードバック データモデル​

AssessmentSource フィールド​

Link feedbackをトレースにLinkする​

バックエンド​

フロントエンド (React)​

バックエンド​

多次元フィードバック​

ストリーミング応答​

バックエンド (FastAPI SSE)​

フィードバックの分析​

その他のリソース​

要件

タグとメタデータ

実行時のタグとメタデータの設定

完了したトレースにタグを設定する

UI でタグを設定する

トレースにコンテキストを追加する

標準コンテキストフィールド

自動入力されるフィールド

ベストプラクティス

ユーザーからのフィードバックを収集する

フィードバックの種類

フィードバックデータモデル

AssessmentSource フィールド

Link feedbackをトレースにLinkする

バックエンド

フロントエンド (React)

バックエンド

多次元フィードバック

ストリーミング応答

バックエンド (FastAPI SSE)

フィードバックの分析

その他のリソース