以编程方式访问追踪数据

通过程序搜索、读取和分析痕迹。 使用 mlflow.search_traces() 查询存储在 Unity Catalog 表、MLflow 跟踪服务器或推理表中的跟踪；对于存储在 Unity Catalog 中的跟踪，还可以直接使用 SQL 查询 Delta 表。 一旦你有了追踪，就阅读其对象模型——元数据、跨度、评估和令牌使用情况——以检查发生了什么。 可以选择用于分析或创建评估数据集的跟踪子集。

mlflow.search_traces() API（应用程序接口）

def mlflow.search_traces(
    experiment_ids: list[str] | None = None,
    filter_string: str | None = None,
    max_results: int | None = None,
    order_by: list[str] | None = None,
    extract_fields: list[str] | None = None,
    run_id: str | None = None,
    return_type: Literal['pandas', 'list'] | None = None,
    model_id: str | None = None,
    sql_warehouse_id: str | None = None,
    include_spans: bool = True,
    locations: list[str] | None = None,
) -> pandas.DataFrame | list[Trace]

mlflow.search_traces() 允许按几个维度筛选和选择数据：

• 按查询字符串进行筛选

• 按位置进行筛选：试验、运行、模型或 Unity 目录架构

• 限制数据：最大结果、包括或排除区间

• 调整返回值格式：数据格式、数据顺序

search_traces() 返回 pandas 数据帧或对象列表 Trace ，然后可以进一步分析这些对象或重新调整为评估数据集。 请参阅这些返回类型的 架构详细信息 。

有关完整详细信息，mlflow.search_traces()请参阅 API 文档。

mlflow.search_traces() 参数

搜索查询语法

该 filter_string 参数使用类似于 SQL 的查询语言来筛选跟踪。 字符串值必须用单引号（例如）trace.status = 'OK'包装，数值不得用引号括起来（例如）。 trace.execution_time_ms > 1000 将条件组合与 AND。 运算符OR不受支持。

支持的筛选器和比较器

Databricks 管理的 MLflow 支持以下字段和比较器。

与 OSS MLflow 的差异

Databricks 托管 MLflow 上的搜索查询语法密切跟踪 OSS MLflow，但存在以下差异：

搜索第三方 OpenTelemetry 范围

若要搜索从第三方 OpenTelemetry 工具（如 Langfuse）引入的跟踪，请改用 span.attributes.* 前缀。 请参阅按 OTel span 属性搜索追踪。

使用 SQL 查询跟踪表

当 Trace 存储在 Unity 目录中时，除了 SDK 之外，还可以用 Databricks SQL 查询它们。 MLflow服务将跨度数据存储在兼容OpenTelemetry的表中，并自动创建Databricks的SQL视图，将数据转换为MLflow格式。

Azure Databricks 建议查询视图（或使用 SDK），而不是底层的 OpenTelemetry 表，因为其模式可能会随时间变化。 对于大型跟踪卷，视图性能可能会下降：可以创建一个对视图的 具体化视图 并逐步更新，或者使用 SDK 以获得近期数据的最佳性能。

{table_prefix}_trace_unified

所有痕迹数据的统一视图，按痕迹ID分组。 每一行包含原始的范围数据以及跟踪元数据（MLflow标签、元数据和评估）。 顶层列：

trace_id: STRING
client_request_id: STRING
request_time: TIMESTAMP
state: STRING
execution_duration_ms: DOUBLE
request: STRING
response: STRING
trace_metadata: VARIANT
tags: MAP<STRING, STRING>
spans: LIST<STRUCT>          # per-span records: name, kind, timing, attributes, status, events, links
assessments: LIST<STRUCT>    # feedback and expectation records with source, value, rationale, metadata

trace_metadata列和 span attributes字段为 VARIANT。 用冒号路径语法读取，并转换为你需要的类型，而不是用地图键查找：

SELECT spans[0].attributes:`mlflow.spanInputs`::STRING FROM my_catalog.my_schema.my_prefix_trace_unified

{table_prefix}_trace_metadata

仅包含 MLflow 标签、元数据和按跟踪 ID 分组的评估结果。 当你需要MLflow注释数据但不需要跨度数据时，它比统一视图更高效。 列： trace_id、 client_request_id、 tags、 trace_metadata、 （ assessments 与统一视图中的结构相同）。

注释数据格式

MLflow 注释实体（元数据、标签、评估和运行链接）也存储在 {table_prefix}_otel_annotations 表中，每个实体对应一行，并具有类型化的 annotation_type（METADATA、TAG、FEEDBACK、EXPECTATION 或 RUN_LINK）。 该表仅追加并采用软删除，因此在检索时需要去重：对每个 annotation_id 取最新的一行（按 updated_at 降序排序），并排除已设置 deleted_at 的行。 value列和metadata列是VARIANT（JSON）。 对于评估，用户提供的元数据与内部MLflow密钥（前缀 mlflow.）并置;读取用户元数据时忽略内部密钥。

分析查询性能

要诊断查询慢，可以在 SQL 仓库 查询历史中查看查询配置文件：打开 SQL 仓库 页面，选择你的仓库，点击 查询历史。 筛选出来源为 MLflow 的查询，然后打开其中一个查询，查看其概况并检查：

• 调度时间：较长的调度时间意味着查询会因仓库负载过重而排队等待。 在MLflow UI中切换到不同的仓库，或者在客户端中配置不同的仓库。

• 整体查询性能：对于持续运行缓慢的查询，使用 更大的SQL仓库，收紧 的 trace.timestamp_ms边界，并尽可能移除其他过滤谓词。

读取跟踪数据

MLflow Trace 有两个组件：

• TraceInfo，有关跟踪的元数据：

  • ID 和状态

  • 预览数据

  • 时序

  • Tags

  • 令牌用法

  • 评估

• TraceData，执行数据：

  • 跨度

  • 完整请求和响应数据

基本元数据属性

# Primary identifiers
print(f"Trace ID: {trace.info.trace_id}")
print(f"Client Request ID: {trace.info.client_request_id}")

# Status information
print(f"State: {trace.info.state}")  # OK, ERROR, IN_PROGRESS

# Request/response previews (truncated)
print(f"Request preview: {trace.info.request_preview}")
print(f"Response preview: {trace.info.response_preview}")

存储位置和试验

location = trace.info.trace_location
print(f"Location type: {location.type}")

# Stored in Unity Catalog (recommended)
if location.uc_table_prefix:
    print(f"UC location: {location.uc_table_prefix.full_table_prefix}")

# Stored in an MLflow experiment
if location.mlflow_experiment:
    print(f"Experiment ID: {trace.info.experiment_id}")

# Stored in a Databricks inference table
if location.inference_table:
    print(f"Table: {location.inference_table.full_table_name}")

实验是 UI 入口，与后端无关。 使用 trace.info.experiment_id 可在 MLflow UI 中打开追踪，即使其存储在 Unity Catalog 中。

请求和响应预览

response_preview 和 request_preview 属性提供完整请求和响应数据的截断摘要，因此您无需加载完整的有效负载，也能了解发生了什么情况。

request_preview = trace.info.request_preview
response_preview = trace.info.response_preview

# Full request/response data (see below)
full_request = trace.data.request
full_response = trace.data.response

与时间相关的属性

# Timestamps (milliseconds since epoch)
print(f"Start time (ms): {trace.info.request_time}")
print(f"Timestamp (ms): {trace.info.timestamp_ms}")  # Alias for request_time

# Duration
print(f"Execution duration (ms): {trace.info.execution_duration}")

# Convert to human-readable format
import datetime
start_time = datetime.datetime.fromtimestamp(trace.info.request_time / 1000)
print(f"Started at: {start_time}")

标记和元数据

# Tags (mutable, can be updated after creation)
for key, value in trace.info.tags.items():
    print(f"  {key}: {value}")

print(f"Environment: {trace.info.tags.get('environment')}")

# Trace metadata (immutable, set at creation)
for key, value in trace.info.trace_metadata.items():
    print(f"  {key}: {value}")

令牌使用情况信息

MLflow 跟踪功能可以通过使用 LLM 提供商 API 返回的令牌计数来记录 LLM 调用的令牌使用情况。

# Get aggregated token usage (if available)
token_usage = trace.info.token_usage
if token_usage:
    print(f"Input tokens: {token_usage.get('input_tokens')}")
    print(f"Output tokens: {token_usage.get('output_tokens')}")
    print(f"Total tokens: {token_usage.get('total_tokens')}")

Assessments

查找带有 search_assessments() 的评估：

# Get all assessments
all_assessments = trace.search_assessments()

# Search by name
helpfulness = trace.search_assessments(name="helpfulness")
if helpfulness:
    assessment = helpfulness[0]
    print(f"Helpfulness: {assessment.value}")
    print(f"Source: {assessment.source.source_type} - {assessment.source.source_id}")
    print(f"Rationale: {assessment.rationale}")

# Search by type
feedback_only = trace.search_assessments(type="feedback")
expectations_only = trace.search_assessments(type="expectation")

# Search by span ID
span_assessments = trace.search_assessments(span_id=retriever_span.span_id)

# Include overridden assessments
all_including_invalid = trace.search_assessments(all=True)

查看评估详细信息：

for assessment in trace.info.assessments:
    print(f"Assessment: {assessment.name}")
    print(f"  Type: {type(assessment).__name__}")
    print(f"  Value: {assessment.value}")
    print(f"  Source: {assessment.source.source_type.value}")

    if assessment.rationale:
        print(f"  Rationale: {assessment.rationale}")
    if assessment.metadata:
        print(f"  Metadata: {assessment.metadata}")
    if assessment.error:
        print(f"  Error: {assessment.error}")

使用 span

跨度是跟踪的基础单元，表示单个操作或工作单位。 该 Span 类表示从跟踪中提取到的不可变已结束的跨度。

访问范围属性

# Access all spans from a trace
spans = trace.data.spans
print(f"Total spans: {len(spans)}")

span = spans[0]

# Basic properties
print(f"Span ID: {span.span_id}")
print(f"Name: {span.name}")
print(f"Type: {span.span_type}")
print(f"Parent ID: {span.parent_id}")  # None for root spans

# Timing (nanoseconds)
duration_ms = (span.end_time_ns - span.start_time_ns) / 1_000_000
print(f"Duration: {duration_ms:.2f}ms")

# Status
print(f"Status code: {span.status.status_code}")

# Inputs and outputs
print(f"Inputs: {span.inputs}")
print(f"Outputs: {span.outputs}")

查找特定跨度

用于 search_spans() 查找与特定条件匹配的跨度：

import re
from mlflow.entities import SpanType

# Search by exact name
retriever_spans = trace.search_spans(name="retrieve_documents")

# Search by regex pattern
tool_spans = trace.search_spans(name=re.compile(r".*_tool$"))

# Search by span type
chat_spans = trace.search_spans(span_type=SpanType.CHAT_MODEL)
llm_spans = trace.search_spans(span_type="CHAT_MODEL")  # String also works

# Search by span ID
specific_span = trace.search_spans(span_id=retriever_spans[0].span_id)

# Combine criteria
tool_fact_check = trace.search_spans(
    name="fact_check_tool",
    span_type=SpanType.TOOL,
)

Span 属性

from mlflow.tracing.constant import SpanAttributeKey

chat_span = trace.search_spans(span_type=SpanType.CHAT_MODEL)[0]

# Get all attributes
for key, value in chat_span.attributes.items():
    print(f"  {key}: {value}")

# Get a specific attribute
specific_attr = chat_span.get_attribute("custom_attribute")

# Access chat-specific attributes using SpanAttributeKey
messages = chat_span.get_attribute(SpanAttributeKey.CHAT_MESSAGES)
tools = chat_span.get_attribute(SpanAttributeKey.CHAT_TOOLS)

# Access per-span token usage
input_tokens = chat_span.get_attribute("llm.token_usage.input_tokens")
output_tokens = chat_span.get_attribute("llm.token_usage.output_tokens")

请求和响应数据

import json

# Get root span request/response
request_json = trace.data.request
response_json = trace.data.response

# Parse JSON strings
if request_json:
    request_data = json.loads(request_json)
if response_json:
    response_data = json.loads(response_json)

最佳做法

关键字参数

始终使用关键字（命名）参数与 mlflow.search_traces()。 它允许位置参数，但函数参数正在演变。

最佳做法： mlflow.search_traces(filter_string="trace.status = 'OK'")

不良做法： mlflow.search_traces([], "trace.status = 'OK'")

filter_string 陷阱

使用 filter_string 参数进行 mlflow.search_traces() 搜索时，请记住以下几点：

• 使用前缀：trace.、或 tag.metadata.

• 如果标记或属性名称中有点，请使用反引号：tag.`mlflow.traceName`

• 仅使用单引号：'value'，而不是"value"

• 使用 Unix 时间戳（以毫秒为单位）来表示时间，1749006880539 而非日期。

• 仅使用 AND：无 OR 支持

有关支持字段和运算符的完整列表，请参阅 搜索查询语法 。

SQL 仓库集成

要读取存储在 Unity Catalog 实验中的跟踪，Databricks SQL 仓库是必需的。 在调用由 Unity Catalog 支持的实验上的 mlflow.search_traces() 或 mlflow.get_trace() 之前，先设置 MLFLOW_TRACING_SQL_WAREHOUSE_ID。 如果未设置此环境变量，读取会失败 SQL warehouse ID is required for accessing traces in UC tables.

对于大型非 Unity 目录数据集，如推理表，SQL 仓库是可选的，并提升查询性能。

import os

os.environ['MLFLOW_TRACING_SQL_WAREHOUSE_ID'] = 'fa92bea7022e81fb'

# Required for UC-backed experiments. Improves performance for large non-UC datasets.
traces = mlflow.search_traces(
    filter_string="trace.status = 'OK'",
    locations=['my_catalog.my_schema'],
)

分页

mlflow.search_traces() 返回内存中的结果，这适用于较小的结果集。 若要处理大型结果集，请使用 MlflowClient.search_traces() ，因为它支持分页。

其他资源

• 检测并监控 MLflow 追踪中的问题 ——让 MLflow 检测你各个追踪中的问题。

• 构建评估数据集 ——将查询的痕迹转换为测试数据集。

下一步：收集反馈并构建数据集