本教程提供了一个简单的示例来开始使用 mlflow.search_traces（） 。 有关搜索跟踪的详细信息，请参阅 以编程方式搜索跟踪。
环境设置
安装所需程序包：
-
mlflow[databricks]：使用最新版本的 MLflow 获取更多功能和改进。 -
openai：此应用将使用 OpenAI API 客户端调用 Databricks 托管的模型。
%pip install -qq --upgrade "mlflow[databricks]>=3.1.0" openai
dbutils.library.restartPython()
创建 MLflow 实验。 如果使用 Databricks 笔记本，则可以跳过此步骤并使用默认笔记本试验。 否则，请按照 环境设置快速指南 创建实验并连接到 MLflow 跟踪服务器。
生成用于分析的跟踪记录
此简单的应用生成用于 search_traces() 的跟踪。
import mlflow
from databricks.sdk import WorkspaceClient
mlflow.openai.autolog()
@mlflow.trace
def my_app(message: str) -> str:
# Create an OpenAI client that is connected to Databricks-hosted LLMs
w = WorkspaceClient()
client = w.serving_endpoints.get_open_ai_client()
response = client.chat.completions.create(
model="databricks-claude-sonnet-4",
messages=[
{
"role": "system",
"content": "You are a helpful assistant. Give brief, 1-2 sentence responses.",
},
{
"role": "user",
"content": message,
},
]
)
# Add examples of custom metadata and tags
mlflow.update_current_trace(
metadata={
"mlflow.trace.user": 'name@my_company.com',
},
tags={
"environment": "production",
},
)
return response.choices[0].message.content
my_app("What is MLflow and how does it help with GenAI?")
my_app("What is ML vs. AI?")
my_app("What is MLflow and how does it help with machine learning?")
快速参考
# Search by status
mlflow.search_traces(filter_string="attributes.status = 'OK'")
mlflow.search_traces(filter_string="attributes.status = 'ERROR'")
# Search by time
mlflow.search_traces(filter_string="attributes.timestamp_ms > 1749006880539")
mlflow.search_traces(filter_string="attributes.execution_time_ms > 2500")
# Search by metadata
mlflow.search_traces(filter_string="metadata.`mlflow.trace.user` = 'name@my_company.com'")
# Search by tags
mlflow.search_traces(filter_string="tags.environment = 'production'")
mlflow.search_traces(filter_string="tags.`mlflow.traceName` = 'my_app'")
# Combined filters (AND only)
mlflow.search_traces(
filter_string="attributes.status = 'OK' AND tags.environment = 'production'"
)
traces = mlflow.search_traces()
traces
mlflow.search_traces() 返回具有以下字段的 Trace pandas 数据帧或对象列表：
list(traces.columns)
['trace_id',
'trace',
'client_request_id',
'state',
'request_time',
'execution_duration',
'request',
'response',
'trace_metadata',
'tags',
'spans',
'assessments']
搜索示例
运行本教程时，下面的代码单元将显示搜索结果。
按状态搜索
通过状态进行搜索，可以查找成功、失败或正在进行的跟踪记录。
mlflow.search_traces(filter_string="attributes.status = 'OK'")
mlflow.search_traces(filter_string="attributes.status != 'ERROR'")
按时间戳搜索
必须使用 Unix 时间戳以毫秒为单位指定时间。
查找过去 5 分钟内的痕迹：
import time
from datetime import datetime
current_time_ms = int(time.time() * 1000)
five_minutes_ago = current_time_ms - (5 * 60 * 1000)
mlflow.search_traces(
filter_string=f"attributes.timestamp_ms > {five_minutes_ago}"
)
在日期范围内搜索：
start_date = int(datetime(2026, 1, 1).timestamp() * 1000)
end_date = int(datetime(2026, 1, 31).timestamp() * 1000)
mlflow.search_traces(
filter_string=f"attributes.timestamp_ms > {start_date} AND attributes.timestamp_ms < {end_date}"
)
还可以使用“timestamp”别名，而不是“timestamp_ms”：
mlflow.search_traces(filter_string=f"attributes.timestamp > {five_minutes_ago}")
按执行时间搜索
查找缓慢的跟踪：
mlflow.search_traces(filter_string="attributes.execution_time_ms > 2500")
还可以使用“延迟”别名，而不是“execution_time_ms”：
mlflow.search_traces(filter_string="attributes.latency > 1000")
按元数据搜索
请记住，对于带有点的元数据名称使用反引号。
搜索特定用户的自定义元数据：
mlflow.search_traces(filter_string="metadata.`mlflow.trace.user` = 'name@my_company.com'")
按标记搜索
请记得对带点的标记名称使用反引号。
搜索系统标记：
mlflow.search_traces(
filter_string="tags.`mlflow.traceName` = 'my_app'"
)
使用 mlflow.update_current_trace() 搜索自定义标签集：
mlflow.search_traces(filter_string="tags.environment = 'production'")
复杂筛选器
仅支持 AND，不支持 OR。
查找最近成功的生产过程追踪记录。
current_time_ms = int(time.time() * 1000)
one_hour_ago = current_time_ms - (60 * 60 * 1000)
mlflow.search_traces(
filter_string=f"attributes.status = 'OK' AND "
f"attributes.timestamp_ms > {one_hour_ago} AND "
f"tags.environment = 'production'"
)
寻找特定用户的快速追踪：
mlflow.search_traces(
filter_string="attributes.execution_time_ms < 2500 AND "
"metadata.`mlflow.trace.user` = 'name@my_company.com'"
)
从超出性能阈值的特定函数查找跟踪：
mlflow.search_traces(
filter_string="tags.`mlflow.traceName` = 'my_app' AND "
"attributes.execution_time_ms > 1000"
)
后续步骤
通常，你将调用 mlflow.search_traces() 提取一组跟踪，然后对返回的数据帧或对象列表 Trace 执行进一步分析或处理。
有关更高级的示例，请参阅：
- 教程：跟踪和分析用户和环境 - 运行将上下文元数据添加到跟踪和分析结果的示例
- 示例：跟踪分析 - 查看各种跟踪分析示例