mlflow
mlflow 模块提供了一个高级的“流式”（fluent）API，用于启动和管理 MLflow 运行。例如
import mlflow
mlflow.start_run()
mlflow.log_param("my", "param")
mlflow.log_metric("score", 100)
mlflow.end_run()
您还可以使用如下的上下文管理器语法
with mlflow.start_run() as run:
mlflow.log_param("my", "param")
mlflow.log_metric("score", 100)
它会在 with 块结束时自动终止运行。
流式跟踪 API 目前不是线程安全的。任何并发调用跟踪 API 的操作都必须手动实现互斥。
如需更底层的 API，请参阅 mlflow.client 模块。
- class mlflow.ActiveModel(logged_model: LoggedModel, set_by_user: bool)[源代码]
mlflow.entities.LoggedModel的包装器，以便启用 Python 的with语法。
- class mlflow.ActiveRun(run)[源代码]
mlflow.entities.Run的包装器，以便启用 Python 的with语法。
- class mlflow.Image(image: Union[numpy.ndarray, PIL.Image.Image, str, list[typing.Any]])[源代码]
mlflow.Image 是一种图像媒体对象，为在 MLflow 中处理图像提供了一种轻量级选项。图像可以是 numpy 数组、PIL 图像或图像的文件路径。该图像以 PIL 图像形式存储，并可使用 mlflow.log_image 或 mlflow.log_table 记录到 MLflow 中。
- 参数
image – 图像可以是 numpy 数组、PIL 图像或图像的文件路径。
import mlflow import numpy as np from PIL import Image # Create an image as a numpy array image = np.zeros((100, 100, 3), dtype=np.uint8) image[:, :50] = [255, 128, 0] # Create an Image object image_obj = mlflow.Image(image) # Convert the Image object to a list of pixel values pixel_values = image_obj.to_list()
- resize(size: tuple[int, int])[源代码]
将图像调整为指定大小。
- 参数
size – 调整后的图像大小。
- 返回
调整大小后的图像对象副本。
- save(path: str)[源代码]
将图像保存到文件。
- 参数
path – 保存图像的文件路径。
- to_array()[源代码]
将图像转换为 numpy 数组。
- 返回
像素值的 numpy 数组。
- to_list()[源代码]
将图像转换为像素值列表。
- 返回
像素值列表。
- to_pil()[源代码]
将图像转换为 PIL 图像。
- 返回
PIL 图像。
- exception mlflow.MlflowException(message: str, error_code: int = 1, sqlstate: Optional[str] = None, error_class: Optional[str] = None, **kwargs)[源代码]
用于表面化外部操作失败信息的通用异常。与此异常关联的错误消息可能会出现在 HTTP 响应中，供客户端进行调试。如果错误文本包含敏感信息，请改用通用的 Exception 对象。
- get_http_status_code()[源代码]
- classmethod invalid_parameter_value(message: str, sqlstate: Optional[str] = None, error_class: Optional[str] = None, **kwargs)[源代码]
构建一个带有 INVALID_PARAMETER_VALUE 错误代码的 MlflowException 对象。
- 参数
message – 描述所发生错误的详细信息。这将包含在异常的序列化 JSON 表示中。
sqlstate – 用于错误分类的 5 字符 SQLSTATE 代码。
error_class – 描述性的错误类名称。
kwargs – 要包含在 MlflowException 序列化 JSON 表示中的其他键值对。
- serialize_as_json()[源代码]
- mlflow.active_run() ActiveRun | None[源代码]
获取当前活动的
Run，如果不存在则返回 None。注意
此 API 是线程本地的（thread-local），仅返回当前线程中的活动运行。如果您的应用程序是多线程的，并且在另一个线程中启动了运行，则此 API 无法检索该运行。
注意：您无法通过
mlflow.active_run返回的运行对象访问当前活动的运行属性（参数、指标等）。要访问这些属性，请按照以下方式使用mlflow.client.MlflowClientimport mlflow mlflow.start_run() run = mlflow.active_run() print(f"Active run_id: {run.info.run_id}") mlflow.end_run()
- mlflow.autolog(log_input_examples: bool = False, log_model_signatures: bool = True, log_models: bool = True, log_datasets: bool = True, log_traces: bool = True, disable: bool = False, exclusive: bool = False, disable_for_unsupported_versions: bool = False, silent: bool = False, extra_tags: dict[str, str] | None = None, exclude_flavors: list[str] | None = None) None[源代码]
为所有支持的集成启用（或禁用）并配置自动日志记录。
参数将传递给任何支持它们的自动日志记录集成。
有关支持的自动日志记录集成列表，请参阅 跟踪文档。
请注意，在任何时候设置的框架特定配置将优先于此函数设置的任何配置。例如
import mlflow mlflow.autolog(log_models=False, exclusive=True) import sklearn
将为 sklearn 启用自动日志记录，且 log_models=False，exclusive=True，但是
import mlflow mlflow.autolog(log_models=False, exclusive=True) import sklearn mlflow.sklearn.autolog(log_models=True)
将为 sklearn 启用自动日志记录，且 log_models=True，exclusive=False（后者源于 mlflow.sklearn.autolog 中 exclusive 的默认值）；其他框架的自动日志函数（例如 mlflow.tensorflow.autolog）将使用 mlflow.autolog 设置的配置（在本例中为 log_models=False，exclusive=True），直到它们被用户显式调用。
- 参数
log_input_examples – 如果为
True，则在训练期间收集训练数据集的输入示例，并将其与模型工件一起记录。如果为False，则不记录输入示例。注意：输入示例是 MLflow 模型属性，仅在log_models也为True时才会收集。log_model_signatures – 如果为
True，则收集描述模型输入和输出的ModelSignatures，并将其与模型工件一起记录。如果为False，则不记录签名。注意：模型签名是 MLflow 模型属性，仅在log_models也为True时才会收集。log_models – 如果为
True，则训练好的模型将作为 MLflow 模型工件进行记录。如果为False，则不记录训练好的模型。输入样本和模型签名（MLflow 模型的属性）在log_models为False时也会被省略。log_datasets – 如果为
True，则数据集信息将被记录到 MLflow Tracking。如果为False，则不记录数据集信息。log_traces – 如果为
True，则收集集成的跟踪信息。如果为False，则不收集跟踪。disable – 如果为
True，则禁用所有支持的自动日志记录集成。如果为False，则启用所有支持的自动日志记录集成。exclusive – 如果为
True，则自动记录的内容不会记录到用户创建的流畅运行中。如果为False，则自动记录的内容将记录到活动的流畅运行中，该运行可能是用户创建的。disable_for_unsupported_versions – 如果为
True，则对所有未针对当前版本 MLflow 客户端进行测试或不兼容的集成库版本禁用自动日志记录。silent – 如果为
True，则在自动日志记录设置和训练执行期间抑制所有来自 MLflow 的事件日志和警告。如果为False，则显示自动日志记录设置和训练执行期间的所有事件和警告。extra_tags – 要为自动日志记录创建的每个托管运行设置的额外标签的字典。
exclude_flavors – 从自动日志记录中排除的风格（flavor）名称列表。例如：tensorflow, pyspark.ml
import numpy as np import mlflow.sklearn from mlflow import MlflowClient from sklearn.linear_model import LinearRegression def print_auto_logged_info(r): tags = {k: v for k, v in r.data.tags.items() if not k.startswith("mlflow.")} artifacts = [f.path for f in MlflowClient().list_artifacts(r.info.run_id, "model")] print(f"run_id: {r.info.run_id}") print(f"artifacts: {artifacts}") print(f"params: {r.data.params}") print(f"metrics: {r.data.metrics}") print(f"tags: {tags}") # prepare training data X = np.array([[1, 1], [1, 2], [2, 2], [2, 3]]) y = np.dot(X, np.array([1, 2])) + 3 # Auto log all the parameters, metrics, and artifacts mlflow.autolog() model = LinearRegression() with mlflow.start_run() as run: model.fit(X, y) # fetch the auto logged parameters and metrics for ended run print_auto_logged_info(mlflow.get_run(run_id=run.info.run_id))
run_id: fd10a17d028c47399a55ab8741721ef7 artifacts: ['model/MLmodel', 'model/conda.yaml', 'model/model.pkl'] params: {'copy_X': 'True', 'normalize': 'False', 'fit_intercept': 'True', 'n_jobs': 'None'} metrics: {'training_score': 1.0, 'training_root_mean_squared_error': 4.440892098500626e-16, 'training_r2_score': 1.0, 'training_mean_absolute_error': 2.220446049250313e-16, 'training_mean_squared_error': 1.9721522630525295e-31} tags: {'estimator_class': 'sklearn.linear_model._base.LinearRegression', 'estimator_name': 'LinearRegression'}
- mlflow.context(metadata: dict[str, str] | None = None, tags: dict[str, str] | None = None, enabled: bool | None = None, session_id: str | None = None, user: str | None = None) Generator[None, None, None][源代码]
一种上下文管理器，可将元数据和/或标签注入到其作用域内创建的任何跟踪中，而无需创建包装跨度（span）。它还可以用于选择性地禁用该作用域内的跟踪。
当您需要将跟踪级信息（例如会话 ID）附加到非您控制的代码（如自动插桩库）生成的跟踪，或者当您想在不影响全局跟踪状态的情况下禁止特定代码块的跟踪时，此功能非常有用。
import mlflow # Enable auto-tracing for LangChain mlflow.langchain.autolog() with mlflow.tracing.context( session_id="session-123", user="user-456", tags={"project": "my-project"}, ): # Any trace created inside this block will carry the metadata and tags. agent.invoke("What is the capital of France?") # Disable tracing within a specific block with mlflow.tracing.context(enabled=False): # No traces will be created inside this block. agent.invoke("This call will not be traced")
该上下文管理器可以嵌套以组合多组元数据和标签。当在多个级别指定相同的键时，内部级别的值优先。
import mlflow with mlflow.tracing.context(metadata={"foo": "bar", "baz": "qux"}): with mlflow.tracing.context(metadata={"foo": "baz", "qux": "quux"}): my_func() # Trace created by my_func will have metadata={"foo": "baz", "baz": "qux", "qux": "quux"}
- 参数
metadata – 要注入到跟踪的
request_metadata中的键值对（在跟踪创建后不可变）。tags – 要注入到跟踪的
tags中的键值对。enabled – 该作用域内是否启用了跟踪。如果为
False，则作用域内的所有跟踪调用将返回NoOpSpan而不创建任何跟踪。如果为None（默认值），则值将继承自外部作用域。这不会影响通过mlflow.tracing.disable()设置的全局跟踪状态。session_id – 与此作用域内创建的跟踪相关联的会话 ID。内部存储为
mlflow.trace.session键下的元数据。user – 与此作用域内创建的跟踪相关联的用户标识符。内部存储为
mlflow.trace.user键下的元数据。
- mlflow.create_experiment(name: str, artifact_location: Optional[str] = None, tags: Optional[dict[str, typing.Any]] = None, trace_location: Optional[UnityCatalog] = None) str[源代码]
创建一个实验。
- 参数
name – 实验名称，必须是一个非空且唯一的字符串。
artifact_location – 存储运行工件的位置。如果未提供，服务器将选择合适的默认值。
tags – 可选字典，包含要设置为实验标签的字符串键和值。
trace_location – 可选的 UC 跟踪位置，用于链接到该实验。必须是
mlflow.entities.trace_location.UnityCatalog(...)的实例。如果未设置table_prefix，则默认为实验 ID。注意：事后调用mlflow.set_experiment以激活该实验并同步跟踪提供程序。
- 返回
所创建实验的字符串 ID。
import mlflow from pathlib import Path # Create an experiment name, which must be unique and case sensitive experiment_id = mlflow.create_experiment( "Social NLP Experiments", artifact_location=Path.cwd().joinpath("mlruns").as_uri(), tags={"version": "v1", "priority": "P1"}, ) experiment = mlflow.get_experiment(experiment_id) print(f"Name: {experiment.name}") print(f"Experiment_id: {experiment.experiment_id}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Tags: {experiment.tags}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}") print(f"Creation timestamp: {experiment.creation_time}")
- mlflow.create_workspace(name: str, description: Optional[str] = None, default_artifact_root: Optional[str] = None, trace_archival_config: Optional[TraceArchivalConfig] = None) Workspace[源代码]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
创建一个新的工作空间。
- 参数
name: 工作空间名称（小写字母数字，内部可包含连字符）。description: 可选的工作空间描述。default_artifact_root: 可选的工件根 URI；回退到服务器默认值。trace_archival_config: 可选的归档设置；字段保留为
None即表示忽略它们。- 返回
新创建的工作空间。
- 引发
MlflowException: 如果名称无效、已存在或没有可用的工件根目录。
- mlflow.delete_experiment(experiment_id: str) None[源代码]
从后端存储中删除实验。
- 参数
experiment_id –
create_experiment返回的字符串形式的实验 ID。
import mlflow experiment_id = mlflow.create_experiment("New Experiment") mlflow.delete_experiment(experiment_id) # Examine the deleted experiment details. experiment = mlflow.get_experiment(experiment_id) print(f"Name: {experiment.name}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}") print(f"Last Updated timestamp: {experiment.last_update_time}")
- mlflow.delete_experiment_tag(key: str) None[源代码]
删除当前实验的标签。
- 参数
key – 要删除的标签名称。
- mlflow.delete_run(run_id: str) None[源代码]
删除具有给定 ID 的运行。
- 参数
run_id – 要删除的运行的唯一标识符。
import mlflow with mlflow.start_run() as run: mlflow.log_param("p", 0) run_id = run.info.run_id mlflow.delete_run(run_id) lifecycle_stage = mlflow.get_run(run_id).info.lifecycle_stage print(f"run_id: {run_id}; lifecycle_stage: {lifecycle_stage}")
- mlflow.delete_tag(key: str) None[源代码]
删除运行的标签。此操作不可逆。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
key – 标签的名称
- mlflow.delete_trace_tag(trace_id: str, key: str) None[源代码]
注意
参数
request_id已弃用。请改用trace_id。删除具有给定跟踪 ID 的跟踪上的标签。
跟踪可以是活动的，也可以是已经结束并记录在后端中的。以下是在活动跟踪上删除标签的示例。您可以替换
trace_id参数以删除已结束跟踪上的标签。import mlflow with mlflow.start_span("my_span") as span: mlflow.set_trace_tag(span.trace_id, "key", "value") mlflow.delete_trace_tag(span.trace_id, "key")
- 参数
trace_id – 要从其删除标签的跟踪 ID。
key – 标签的字符串键。长度最多为 250 个字符，否则在存储时将被截断。
- mlflow.delete_workspace(name: str, *, mode: str = WorkspaceDeletionMode.RESTRICT) None[源代码]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
删除现有的工作空间。
- 参数
name: 要删除的工作空间名称。mode: 删除模式。可以是 SET_DEFAULT、CASCADE 或 RESTRICT 之一。
- mlflow.disable_system_metrics_logging()[源代码]
全局禁用系统指标日志记录。
调用此函数将全局禁用系统指标日志记录，但用户仍可通过 mlflow.start_run(log_system_metrics=True) 为单个运行选择加入系统指标日志记录。
- mlflow.doctor(mask_envs=False)[源代码]
打印出有助于调试 MLflow 相关问题的有用信息。
- 参数
mask_envs – 如果为 True，则在输出中屏蔽 MLflow 环境变量值（例如 “MLFLOW_ENV_VAR”: “***”），以防止泄露敏感信息。
警告
此 API 仅应用于调试目的。
输出可能包含敏感信息，例如包含密码的数据库 URI。
System information: Linux #58~20.04.1-Ubuntu SMP Thu Oct 13 13:09:46 UTC 2022 Python version: 3.8.13 MLflow version: 2.0.1 MLflow module location: /usr/local/lib/python3.8/site-packages/mlflow/__init__.py Tracking URI: sqlite:///mlflow.db Registry URI: sqlite:///mlflow.db MLflow environment variables: MLFLOW_TRACKING_URI: sqlite:///mlflow.db MLflow dependencies: Flask: 2.2.2 Jinja2: 3.0.3 alembic: 1.8.1 click: 8.1.3 cloudpickle: 2.2.0 databricks-cli: 0.17.4.dev0 docker: 6.0.0 entrypoints: 0.4 gitpython: 3.1.29 gunicorn: 20.1.0 importlib-metadata: 5.0.0 markdown: 3.4.1 matplotlib: 3.6.1 numpy: 1.23.4 packaging: 21.3 pandas: 1.5.1 protobuf: 3.19.6 pyarrow: 9.0.0 pytz: 2022.6 pyyaml: 6.0 querystring-parser: 1.2.4 requests: 2.28.1 scikit-learn: 1.1.3 scipy: 1.9.3 shap: 0.41.0 sqlalchemy: 1.4.42 sqlparse: 0.4.3
- mlflow.enable_system_metrics_logging()[源代码]
全局启用系统指标日志记录。
调用此函数将全局启用系统指标日志记录，但用户仍可通过 mlflow.start_run(log_system_metrics=False) 为单个运行选择退出系统指标日志记录。
- mlflow.end_run(status: str = 'FINISHED') None[源代码]
结束一个活动的 MLflow 运行（如果存在的话）。
import mlflow # Start run and get status mlflow.start_run() run = mlflow.active_run() print(f"run_id: {run.info.run_id}; status: {run.info.status}") # End run and get status mlflow.end_run() run = mlflow.get_run(run.info.run_id) print(f"run_id: {run.info.run_id}; status: {run.info.status}") print("--") # Check for any active runs print(f"Active run: {mlflow.active_run()}")
- mlflow.evaluate(model=None, data=None, *, model_type=None, targets=None, predictions=None, dataset_path=None, feature_names=None, evaluators=None, evaluator_config=None, extra_metrics=None, custom_artifacts=None, env_manager='local', model_config=None, inference_params=None, model_id=None, _called_from_genai_evaluate=False)[源代码]
在给定的数据和选定的指标上评估模型性能。
此函数使用指定的
evaluators在指定数据集上评估 PyFunc 模型或自定义可调用对象，并将产生的指标和工件记录到 MLflow 跟踪服务器。用户也可以跳过设置model，直接将模型输出放入data中进行评估。有关详细信息，请阅读 模型评估文档。- 默认评估器行为
默认评估器（可以通过
evaluators="default"或evaluators=None调用）支持下面列出的模型类型。对于每种预定义模型类型，默认评估器会在选定的一组指标上评估您的模型，并生成绘图等工件。请在下方查找更多详细信息。对于
"regressor"和"classifier"模型类型，默认评估器都会使用 SHAP 生成模型摘要图和特征重要性图。- 对于回归模型（regressor），默认评估器还会记录
指标: example_count, mean_absolute_error, mean_squared_error, root_mean_squared_error, sum_on_target, mean_on_target, r2_score, max_error, mean_absolute_percentage_error。
- 对于二分类模型（binary classifiers），默认评估器还会记录
指标: true_negatives, false_positives, false_negatives, true_positives, recall, precision, f1_score, accuracy_score, example_count, log_loss, roc_auc, precision_recall_auc。
工件: 提升曲线图（lift curve plot）、准确率-召回率曲线图、ROC 曲线图。
- 对于多分类模型（multiclass classifiers），默认评估器还会记录
指标: accuracy_score, example_count, f1_score_micro, f1_score_macro, log_loss
工件: 一个用于 “per_class_metrics” 的 CSV 文件（每类指标包括 true_negatives/false_positives/false_negatives/true_positives/recall/precision/roc_auc, precision_recall_auc）、准确率-召回率合并曲线图、ROC 合并曲线图。
- 对于问答模型（question-answering），默认评估器记录
指标:
exact_match,token_count, toxicity (需要 evaluate, torch), flesch_kincaid_grade_level (需要 textstat) 和 ari_grade_level。工件: 一个 JSON 文件，包含以表格格式呈现的输入、输出、目标（如果提供了
targets参数）以及模型的每行指标。
- 对于文本摘要模型（text-summarization），默认评估器记录
指标:
token_count, ROUGE (需要安装 evaluate, nltk 和 rouge_score), toxicity (需要 evaluate, torch, transformers), ari_grade_level (需要 textstat), flesch_kincaid_grade_level (需要 textstat)。工件: 一个 JSON 文件，包含以表格格式呈现的输入、输出、目标（如果提供了
targets参数）以及模型的每行指标。
- 对于文本模型（text），默认评估器记录
指标:
token_count, toxicity (需要 evaluate, torch, transformers), ari_grade_level (需要 textstat), flesch_kincaid_grade_level (需要 textstat)。工件: 一个 JSON 文件，包含以表格格式呈现的输入、输出、目标（如果提供了
targets参数）以及模型的每行指标。
- 对于检索模型（retriever），默认评估器记录
指标:
precision_at_k(k),recall_at_k(k)和ndcg_at_k(k)- 默认值均为retriever_k= 3。工件: 一个 JSON 文件，包含以表格格式呈现的输入、输出、目标以及模型的每行指标。
对于 sklearn 模型，默认评估器还会记录由 model.score 方法计算出的模型评估标准（例如分类器的平均准确率）。
上述列出的指标/工件被记录到活动的 MLflow 运行中。如果不存在活动运行，则会创建一个新的 MLflow 运行来记录这些指标和工件。
此外，有关指定数据集的信息——哈希值、名称（如果指定）、路径（如果指定）以及评估该数据集的模型 UUID——会被记录到
mlflow.datasets标签中。- 对于默认评估器，可用的
evaluator_config选项包括 log_model_explainability: 一个布尔值，指定是否记录模型可解释性见解，默认值为 True。
- log_explainer: 如果为 True，则记录用于计算模型可解释性
见解的解释器作为模型。默认值为 False。
explainability_algorithm: 指定模型可解释性 SHAP 解释器算法的字符串。支持的算法包括：‘exact’, ‘permutation’, ‘partition’, ‘kernel’。如果未设置，则
shap.Explainer将使用 “auto” 算法，该算法会根据模型自动选择最佳解释器。explainability_nsamples: 用于计算模型可解释性见解的样本行数。默认值为 2000。
explainability_kernel_link: SHAP 内核解释器使用的内核链接函数。可用值为 “identity” 和 “logit”。默认值为 “identity”。
max_classes_for_multiclass_roc_pr: 对于多分类任务，记录每类 ROC 曲线和准确率-召回率曲线的最大类数。如果类数大于配置的最大值，则不记录这些曲线。
metric_prefix: 一个可选的前缀，添加到评估期间生成的每个指标和工件的名称之前。
log_metrics_with_dataset_info: 一个布尔值，指定是否在评估期间记录到 MLflow Tracking 的每个指标名称中包含评估数据集的信息，默认值为 True。
pos_label: 如果指定，则为计算二分类模型的分类指标（如准确率、召回率、f1 等）时使用的正标签。对于多分类和回归模型，此参数将被忽略。
average: 计算多分类模型的分类指标（如准确率、召回率、f1 等）时使用的平均方法（默认值：
'weighted'）。对于二分类和回归模型，此参数将被忽略。sample_weights: 计算模型性能指标时应用的每个样本的权重。
col_mapping: 一个将输入数据集或输出预测中的列名映射到调用评估函数时所使用列名的字典。
retriever_k: 当
model_type="retriever"时使用的参数，作为计算内置指标precision_at_k(k),recall_at_k(k)和ndcg_at_k(k)时所使用的最高排名检索文档数。默认值为 3。对于所有其他模型类型，此参数将被忽略。
- 对于默认评估器，可用的
- 评估数据集的限制
对于分类任务，数据集标签用于推断总类数。
对于二分类任务，负标签值必须为 0、-1 或 False，正标签值必须为 1 或 True。
- 指标/工件计算的限制
对于分类任务，某些指标和工件的计算要求模型输出类概率。目前，对于 scikit-learn 模型，默认评估器会调用底层模型的
predict_proba方法来获取概率。对于其他模型类型，默认评估器不会计算需要概率输出的指标/工件。
- 默认评估器记录模型可解释性见解的限制
shap.Explainer的auto算法对线性模型使用Linear解释器，对树模型使用Tree解释器。由于 SHAP 的Linear和Tree解释器不支持多分类，因此默认评估器在多分类任务中回退使用Exact或Permutation解释器。目前不支持为 PySpark 模型记录模型可解释性见解。
评估数据集的标签值必须是数字或布尔值，所有特征值必须是数字，并且每个特征列只能包含标量值。
- 启用环境恢复时的限制
当为被评估模型启用环境恢复（即指定了非本地的
env_manager）时，模型将作为客户端加载，并在具有模型训练时依赖项的独立 Python 环境中调用 MLflow 模型评分服务器进程。因此，模型的predict_proba（用于概率输出）或score（计算 sklearn 模型的评估标准）等方法变得无法访问，默认评估器不会计算需要这些方法的指标或工件。由于模型是 MLflow 模型服务器进程，SHAP 解释计算较慢。因此，除非明确将
evaluator_config选项 log_model_explainability 设置为True，否则在指定了非本地env_manager时，模型可解释性将被禁用。
- 参数
model –
可选。如果指定，它应为以下内容之一
Pyfunc 模型实例
指向 pyfunc 模型的 URI
指向 MLflow Deployments 端点的 URI，例如
"endpoints:/my-chat"可调用函数：此函数应能够接收模型输入并返回预测结果。它应遵循
predict方法的签名。以下是有效函数的示例model = mlflow.pyfunc.load_model(model_uri) def fn(model_input): return model.predict(model_input)
如果省略，表示将使用静态数据集进行评估，而不是模型。在这种情况下，
data参数必须是包含模型输出的 Pandas DataFrame 或 mlflow PandasDataset，并且predictions参数必须是data中包含模型输出的列名。data –
以下内容之一
评估特征的 numpy 数组或列表，排除标签。
- 包含评估特征、标签以及可选模型输出的 Pandas DataFrame。
当模型未指定时，需要提供模型输出。如果未指定
feature_names参数，除标签列和预测列之外的所有列均被视为特征列。否则，仅feature_names中存在的列名被视为特征列。
- 包含评估特征和标签的 Spark DataFrame。如果
未指定
feature_names参数，除标签列之外的所有列均被视为特征列。否则，仅feature_names中存在的列名被视为特征列。Spark DataFrame 中仅前 10000 行将被用作评估数据。
- 包含评估
特征、标签以及可选模型输出的
mlflow.data.dataset.Dataset实例。仅 PandasDataset 支持模型输出。当未指定模型时，需要模型输出，并应通过 PandasDataset 的predictions属性指定。
model_type –
(可选) 描述模型类型的字符串。默认评估器支持以下模型类型
'classifier''regressor''question-answering''text-summarization''text''retriever'
如果未指定
model_type，则必须通过extra_metrics参数提供要计算的指标列表。注意
'question-answering','text-summarization','text'和'retriever'属于实验性质，在未来版本中可能会更改或删除。targets – 如果
data是 numpy 数组或列表，则为评估标签的 numpy 数组或列表。如果data是 DataFrame，则为data中包含评估标签的列名字符串。分类器和回归模型必需，但对于问答、文本摘要和文本模型是可选的。如果data是定义了目标的mlflow.data.dataset.Dataset，则targets为可选。predictions –
可选。包含模型输出的列名。
当指定了
model且输出多个列时，predictions可用于指定将用于存储模型评估输出的列名。当未指定
model且data是 pandas dataframe 时，predictions可用于指定data中包含模型输出的列名。
# Evaluate a model that outputs multiple columns data = pd.DataFrame({"question": ["foo"]}) def model(inputs): return pd.DataFrame({"answer": ["bar"], "source": ["baz"]}) results = evaluate( model=model, data=data, predictions="answer", # other arguments if needed ) # Evaluate a static dataset data = pd.DataFrame({"question": ["foo"], "answer": ["bar"], "source": ["baz"]}) results = evaluate( data=data, predictions="answer", # other arguments if needed )
dataset_path – (可选) 数据存储的路径。不得包含双引号 (
")。如果指定，该路径将记录到mlflow.datasets标签中，用于血缘追踪。feature_names – (可选) 列表。如果
data参数是 numpy 数组或列表，则feature_names是每个特征的特征名称列表。如果feature_names=None，则使用feature_{feature_index}格式生成特征名称。如果data参数是 Pandas DataFrame 或 Spark DataFrame，则feature_names是 DataFrame 中特征列的名称列表。如果feature_names=None，则除标签列和预测列之外的所有列均被视为特征列。evaluators – 用于模型评估的评估器名称，或评估器名称列表。如果未指定，则使用所有能够评估指定数据集上指定模型的评估器。默认评估器可通过名称
"default"引用。要查看所有可用评估器，请调用mlflow.models.list_evaluators()。evaluator_config – 提供给评估器的附加配置字典。如果指定了多个评估器，则每个配置应作为嵌套字典提供，其键为评估器名称。
extra_metrics –
(可选)
EvaluationMetric对象列表。这些指标将在与预定义 model_type 关联的默认指标之外计算，设置 model_type=None 将仅计算 extra_metrics 中指定的指标。有关内置指标及如何定义额外指标的更多信息，请参阅 mlflow.metrics 模块。import mlflow import numpy as np def root_mean_squared_error(eval_df, _builtin_metrics): return np.sqrt((np.abs(eval_df["prediction"] - eval_df["target"]) ** 2).mean()) rmse_metric = mlflow.models.make_metric( eval_fn=root_mean_squared_error, greater_is_better=False, ) mlflow.evaluate(..., extra_metrics=[rmse_metric])
custom_artifacts –
(可选) 具有以下签名的自定义工件函数列表
def custom_artifact( eval_df: Union[pandas.Dataframe, pyspark.sql.DataFrame], builtin_metrics: Dict[str, float], artifacts_dir: str, ) -> Dict[str, Any]: """ Args: eval_df: A Pandas or Spark DataFrame containing ``prediction`` and ``target`` column. The ``prediction`` column contains the predictions made by the model. The ``target`` column contains the corresponding labels to the predictions made on that row. builtin_metrics: A dictionary containing the metrics calculated by the default evaluator. The keys are the names of the metrics and the values are the scalar values of the metrics. Refer to the DefaultEvaluator behavior section for what metrics will be returned based on the type of model (i.e. classifier or regressor). artifacts_dir: A temporary directory path that can be used by the custom artifacts function to temporarily store produced artifacts. The directory will be deleted after the artifacts are logged. Returns: A dictionary that maps artifact names to artifact objects (e.g. a Matplotlib Figure) or to artifact paths within ``artifacts_dir``. """ ...
工件可表示的对象类型
表示工件文件路径的字符串 uri。MLflow 将根据文件扩展名推断工件类型。
JSON 对象的字符串表示形式。这将保存为 .json 工件。
Pandas DataFrame。这将解析为 CSV 工件。
Numpy 数组。这将保存为 .npy 工件。
Matplotlib 图形。这将保存为图像工件。注意，
matplotlib.pyplot.savefig在幕后使用默认配置进行调用。要自定义，请使用所需的配置保存图形并返回其文件路径，或通过matplotlib.rcParams中的环境变量定义自定义。将尝试使用默认协议对其他对象进行 pickle 处理。
import mlflow import matplotlib.pyplot as plt def scatter_plot(eval_df, builtin_metrics, artifacts_dir): plt.scatter(eval_df["prediction"], eval_df["target"]) plt.xlabel("Targets") plt.ylabel("Predictions") plt.title("Targets vs. Predictions") plt.savefig(os.path.join(artifacts_dir, "example.png")) plt.close() return {"pred_target_scatter": os.path.join(artifacts_dir, "example.png")} def pred_sample(eval_df, _builtin_metrics, _artifacts_dir): return {"pred_sample": pred_sample.head(10)} mlflow.evaluate(..., custom_artifacts=[scatter_plot, pred_sample])
env_manager –
指定环境管理器以在隔离的 Python 环境中加载候选
model并恢复其依赖项。默认值为local，并支持以下值virtualenv: (推荐) 使用 virtualenv 恢复训练模型时所使用的 python 环境。conda: 使用 Conda 恢复训练模型时使用的软件环境。local: 使用当前 Python 环境进行模型推理，这可能与训练模型时使用的环境不同，并可能导致错误或无效的预测。
model_config – 用于使用 pyfunc 加载模型的模型配置。检查模型的 pyfunc 风格以了解您的特定模型支持哪些键。如果未指示，则使用模型的默认模型配置（如果有）。
inference_params – (可选) 进行预测时传递给模型的推理参数字典，例如
{"max_tokens": 100}。仅在model为 MLflow Deployments 端点 URI（例如"endpoints:/my-chat"）时使用。model_id – (可选) 评估结果（例如指标和跟踪）将链接到的 MLflow LoggedModel 或模型版本的 ID。如果未指定 model_id 但指定了 model，则将使用来自 model 的 ID。
_called_from_genai_evaluate – (可选) 仅在内部使用。
- 返回
一个
mlflow.models.EvaluationResult实例，包含使用给定数据集评估模型后的指标。
- mlflow.flush_artifact_async_logging() None[源代码]
刷新所有待处理的工件异步日志记录。
- mlflow.flush_async_logging() None[源代码]
刷新所有待处理的异步日志记录。
- mlflow.flush_trace_async_logging(terminate=False) None[源代码]
刷新所有待处理的跟踪异步日志记录。
- 参数
terminate – 如果为 True，则在刷新后关闭日志记录线程。
- mlflow.get_active_model_id() str | None[源代码]
获取活动模型 ID。如果未通过
set_active_model()设置活动模型，则默认活动模型将使用来自环境变量MLFLOW_ACTIVE_MODEL_ID或遗留环境变量_MLFLOW_ACTIVE_MODEL_ID的模型 ID 进行设置。如果两者都未设置，则返回 None。请注意，此函数仅从当前线程获取活动模型 ID。- 返回
如果已设置，则返回活动模型 ID，否则返回 None。
- mlflow.get_active_trace_id() str | None[源代码]
获取当前进程中的活动跟踪 ID。
此函数是线程安全的。
示例
import mlflow @mlflow.trace def f(): trace_id = mlflow.get_active_trace_id() print(trace_id) f()
- 返回
如果存在，则返回当前活动的跟踪 ID，否则返回 None。
- mlflow.get_artifact_uri(artifact_path: Optional[str] = None) str[源代码]
获取当前活动运行中指定工件的绝对 URI。
如果未指定 path，将返回当前活动运行的工件根 URI；对
log_artifact和log_artifacts的调用会将工件写入工件根 URI 的子目录中。如果当前没有运行活动，此方法将创建一个新的活动运行。
- 参数
artifact_path – 用于获取绝对 URI 的运行相对工件路径。例如，“path/to/artifact”。如果未指定，将返回当前活动运行的工件根 URI。
- 返回
指向指定工件或当前活动运行工件根目录的绝对 URI。例如，如果提供了工件路径且当前活动运行使用 S3 后端存储，则这可能是形式为
s3://<bucket_name>/path/to/artifact/root/path/to/artifact的 URI。如果未提供工件路径且当前活动运行使用 S3 后端存储，则这可能是形式为s3://<bucket_name>/path/to/artifact/root的 URI。
import tempfile import mlflow features = "rooms, zipcode, median_price, school_rating, transport" with tempfile.NamedTemporaryFile("w") as tmp_file: tmp_file.write(features) tmp_file.flush() # Log the artifact in a directory "features" under the root artifact_uri/features with mlflow.start_run(): mlflow.log_artifact(tmp_file.name, artifact_path="features") # Fetch the artifact uri root directory artifact_uri = mlflow.get_artifact_uri() print(f"Artifact uri: {artifact_uri}") # Fetch a specific artifact uri artifact_uri = mlflow.get_artifact_uri(artifact_path="features/features.txt") print(f"Artifact uri: {artifact_uri}")
- mlflow.get_assessment(trace_id: str, assessment_id: str) Assessment[source]
从后端存储中获取评估实体。
- 参数
trace_id – 追踪记录（trace）的 ID。
assessment_id – 要获取的评估（assessment）的 ID。
- 返回
Assessment 对象。
- 返回类型
- mlflow.get_experiment(experiment_id: str) Experiment[source]
通过 experiment_id 从后端存储中检索实验。
- 参数
experiment_id –
create_experiment返回的字符串形式的实验 ID。- 返回
import mlflow experiment = mlflow.get_experiment("0") print(f"Name: {experiment.name}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Tags: {experiment.tags}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}") print(f"Creation timestamp: {experiment.creation_time}")
- mlflow.get_experiment_by_name(name: str) Experiment | None[source]
通过实验名称从后端存储中检索实验。
- 参数
name – 区分大小写的实验名称。
- 返回
如果存在具有指定名称的实验，则返回一个
mlflow.entities.Experiment实例，否则返回 None。
import mlflow # Case sensitive name experiment = mlflow.get_experiment_by_name("Default") print(f"Experiment_id: {experiment.experiment_id}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Tags: {experiment.tags}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}") print(f"Creation timestamp: {experiment.creation_time}")
- mlflow.get_parent_run(run_id: str) Run | None[source]
如果存在，获取给定运行 ID 的父运行。
- 参数
run_id – 子运行的唯一标识符。
- 返回
如果存在父运行，则返回一个
mlflow.entities.Run对象。否则，返回 None。
import mlflow # Create nested runs with mlflow.start_run(): with mlflow.start_run(nested=True) as child_run: child_run_id = child_run.info.run_id parent_run = mlflow.get_parent_run(child_run_id) print(f"child_run_id: {child_run_id}") print(f"parent_run_id: {parent_run.info.run_id}")
- mlflow.get_registry_uri() str[source]
获取当前的注册表 URI。如果未指定，则默认为跟踪 URI。
- 返回
注册表 URI。
# Get the current model registry uri mr_uri = mlflow.get_registry_uri() print(f"Current model registry uri: {mr_uri}") # Get the current tracking uri tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}") # They should be the same assert mr_uri == tracking_uri
Current model registry uri: file:///.../mlruns Current tracking uri: file:///.../mlruns
- mlflow.get_run(run_id: str) Run[source]
从后端存储中获取运行记录。生成的 Run 包含运行元数据集合（RunInfo）以及运行参数、标签和指标集合（RunData）。它还包含运行输入集合（实验性功能），包括有关运行使用的数据集的信息（RunInputs）。如果运行中记录了多个具有相同键的指标，RunData 将包含每个指标在最大步长处最近记录的值。
- 参数
run_id – 运行的唯一标识符。
- 返回
如果运行存在，则返回单个 Run 对象。否则，引发异常。
import mlflow with mlflow.start_run() as run: mlflow.log_param("p", 0) run_id = run.info.run_id print(f"run_id: {run_id}; lifecycle_stage: {mlflow.get_run(run_id).info.lifecycle_stage}")
- mlflow.get_tracking_uri() str[source]
获取当前的跟踪 URI。这可能与当前活动运行的跟踪 URI 不对应，因为跟踪 URI 可以通过
set_tracking_uri进行更新。- 返回
跟踪 URI。
import mlflow # Get the current tracking uri tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}")
Current tracking uri: sqlite:///mlflow.db
- mlflow.import_checkpoints(checkpoint_path: str, source_run_id: Optional[str] = None, model_prefix: Optional[str] = None, overwrite_checkpoints: bool = False) list[LoggedModel][source]
为指定检查点路径下的所有顶级文件和目录创建外部模型。
此 API 目前仅支持 Databricks 运行时。
- 参数
checkpoint_path – 包含检查点的路径。目前仅支持 Databricks Unity Catalog Volume 路径。它必须遵循 https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-volumes#volume-naming-and-reference 中指定的“/Volumes/<catalog_identifier>/<schema_identifier>/<volume_identifier>/<path_to_checkpoints_directory>”格式。注意：每个路径必须与其他模型和运行隔离。
source_run_id – 这些检查点训练所使用的 MLflow 源运行的 ID。如果未提供，则在可用时使用当前的活动运行。
model_prefix – 为从每个检查点创建的每个外部模型名称添加前缀的字符串。如果未提供，则不应用前缀。
overwrite_checkpoints – 如果为 True 且在关联的实验中发现了同名的现有模型，它们将被删除并重新创建以指向最新的检查点。默认为 False。
- 返回
- 已导入模型的列表。如果 ‘overwrite_checkpoints’ 为 True，则列表仅包含
新创建的模型，否则列表包含新名称的新创建模型以及现有名称的现有模型。
示例
import mlflow # Optionally start a run so `source_run_id` can be inferred with mlflow.start_run() as run: # ... training code that writes checkpoints to a UC Volume ... logged_models = mlflow.import_checkpoints( checkpoint_path=( "/Volumes/mycatalog/myschema/myvolume/mytrainingmodel/trainingrun1/checkpoints" ), # You can omit `source_run_id` if there is an active run. # source_run_id=run.info.run_id, model_prefix="my_model_", overwrite_checkpoints=True, )
- mlflow.is_tracking_uri_set()[source]
如果已设置跟踪 URI，则返回 True，否则返回 False。
- mlflow.last_active_run() Run | None[source]
获取最近的活动运行。
示例
import mlflow from sklearn.model_selection import train_test_split from sklearn.datasets import load_diabetes from sklearn.ensemble import RandomForestRegressor mlflow.autolog() db = load_diabetes() X_train, X_test, y_train, y_test = train_test_split(db.data, db.target) # Create and train models. rf = RandomForestRegressor(n_estimators=100, max_depth=6, max_features=3) rf.fit(X_train, y_train) # Use the model to make predictions on the test dataset. predictions = rf.predict(X_test) autolog_run = mlflow.last_active_run()
- 返回
如果存在，则返回活动运行（这等同于
mlflow.active_run()）。否则，返回当前 Python 进程中启动的、已达到终止状态（即 FINISHED、FAILED 或 KILLED）的最后一次运行。
- mlflow.load_table(artifact_file: str, run_ids: list[str] | None = None, extra_columns: list[str] | None = None) pandas.DataFrame[source]
将 MLflow 跟踪中的表作为 pandas.DataFrame 加载。该表从指定的 run_ids 中的指定 artifact_file 加载。extra_columns 是不在原表中但通过运行信息进行扩充并添加到 DataFrame 中的列。
- 参数
artifact_file – 以 posixpath 格式表示的、相对于运行的构件文件路径，即要加载的表（例如“dir/file.json”）。
run_ids – 可选的 run_ids 列表，用于从中加载表。如果未指定 run_ids，则从当前实验中的所有运行中加载该表。
extra_columns – 可选的额外列列表，用于添加到返回的 DataFrame 中。例如，如果 extra_columns=[“run_id”]，则返回的 DataFrame 将有一个名为 run_id 的列。
- 返回
如果构件存在，则返回包含已加载表的 pandas.DataFrame，否则抛出 MlflowException。
import mlflow table_dict = { "inputs": ["What is MLflow?", "What is Databricks?"], "outputs": ["MLflow is ...", "Databricks is ..."], "toxicity": [0.0, 0.0], } with mlflow.start_run() as run: # Log the dictionary as a table mlflow.log_table(data=table_dict, artifact_file="qabot_eval_results.json") run_id = run.info.run_id loaded_table = mlflow.load_table( artifact_file="qabot_eval_results.json", run_ids=[run_id], # Append a column containing the associated run ID for each row extra_columns=["run_id"], )
# Loads the table with the specified name for all runs in the given # experiment and joins them together import mlflow table_dict = { "inputs": ["What is MLflow?", "What is Databricks?"], "outputs": ["MLflow is ...", "Databricks is ..."], "toxicity": [0.0, 0.0], } with mlflow.start_run(): # Log the dictionary as a table mlflow.log_table(data=table_dict, artifact_file="qabot_eval_results.json") loaded_table = mlflow.load_table( "qabot_eval_results.json", # Append the run ID and the parent run ID to the table extra_columns=["run_id"], )
- mlflow.log_artifact(local_path: str, artifact_path: Optional[str] = None, run_id: Optional[str] = None) None[source]
将本地文件或目录记录为当前活动运行的构件。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
local_path – 要写入的文件的路径。
artifact_path – 如果提供，则为
artifact_uri中要写入的目录。run_id – 如果指定，则将构件记录到指定的运行中。如果未指定，则将构件记录到当前活动运行中。
import tempfile from pathlib import Path import mlflow # Create a features.txt artifact file features = "rooms, zipcode, median_price, school_rating, transport" with tempfile.TemporaryDirectory() as tmp_dir: path = Path(tmp_dir, "features.txt") path.write_text(features) # With artifact_path=None write features.txt under # root artifact_uri/artifacts directory with mlflow.start_run(): mlflow.log_artifact(path)
- mlflow.log_artifacts(local_dir: str, artifact_path: Optional[str] = None, run_id: Optional[str] = None) None[source]
将本地目录的所有内容作为运行的构件进行记录。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
local_dir – 要写入的文件目录的路径。
artifact_path – 如果提供，则为
artifact_uri中要写入的目录。run_id – 如果指定，则将构件记录到指定的运行中。如果未指定，则将构件记录到当前活动运行中。
import json import tempfile from pathlib import Path import mlflow # Create some files to preserve as artifacts features = "rooms, zipcode, median_price, school_rating, transport" data = {"state": "TX", "Available": 25, "Type": "Detached"} with tempfile.TemporaryDirectory() as tmp_dir: tmp_dir = Path(tmp_dir) with (tmp_dir / "data.json").open("w") as f: json.dump(data, f, indent=2) with (tmp_dir / "features.json").open("w") as f: f.write(features) # Write all files in `tmp_dir` to root artifact_uri/states with mlflow.start_run(): mlflow.log_artifacts(tmp_dir, artifact_path="states")
- mlflow.log_dict(dictionary: dict[str, typing.Any], artifact_file: str, run_id: Optional[str] = None) None[source]
将 JSON/YAML 可序列化对象（例如 dict）记录为构件。序列化格式（JSON 或 YAML）会根据 artifact_file 的扩展名自动推断。如果文件扩展名不存在或不匹配 [“.json”, “.yml”, “.yaml”] 中的任何一个，则使用 JSON 格式。
- 参数
dictionary – 要记录的字典。
artifact_file – 以 posixpath 格式表示的、相对于运行的构件文件路径，用于保存字典（例如“dir/data.json”）。
run_id – 如果指定，则将字典记录到指定的运行中。如果未指定，则将字典记录到当前活动运行中。
import mlflow dictionary = {"k": "v"} with mlflow.start_run(): # Log a dictionary as a JSON file under the run's root artifact directory mlflow.log_dict(dictionary, "data.json") # Log a dictionary as a YAML file in a subdirectory of the run's root artifact directory mlflow.log_dict(dictionary, "dir/data.yml") # If the file extension doesn't exist or match any of [".json", ".yaml", ".yml"], # JSON format is used. mlflow.log_dict(dictionary, "data") mlflow.log_dict(dictionary, "data.txt")
- mlflow.log_figure(figure: Union[matplotlib.figure.Figure, plotly.graph_objects.Figure], artifact_file: str, *, save_kwargs: dict[str, typing.Any] | None = None) None[source]
将图形（figure）记录为构件。支持以下图形对象：
- 参数
figure – 要记录的图形。
artifact_file – 以 posixpath 格式表示的、相对于运行的构件文件路径，用于保存图形（例如“dir/file.png”）。
save_kwargs – 传递给保存图形的方法的其他关键字参数。
import mlflow import matplotlib.pyplot as plt fig, ax = plt.subplots() ax.plot([0, 1], [2, 3]) with mlflow.start_run(): mlflow.log_figure(fig, "figure.png")
- mlflow.log_image(image: Union[numpy.ndarray, PIL.Image.Image, mlflow.Image], artifact_file: str | None = None, key: str | None = None, step: int | None = None, timestamp: int | None = None, synchronous: bool | None = False) None[source]
在 MLflow 中记录图像，支持两种使用场景：
- 1. 基于时间步的图像记录
非常适合跟踪迭代过程中的变化或进展（例如，在模型训练阶段）。
用法：
log_image(image, key=key, step=step, timestamp=timestamp)
- 2. 构件文件图像记录
最适用于静态图像记录，图像直接保存为文件构件。
用法：
log_image(image, artifact_file)
- 支持以下图像格式：
-
mlflow.Image：对 PIL 图像的 MLflow 包装器，便于进行图像记录。
- Numpy 数组支持
数据类型：
bool（适用于记录图像遮罩）
整数 [0, 255]
无符号整数 [0, 255]
浮点数 [0.0, 1.0]
警告
超出范围的整数值将引发 ValueError。
超出范围的浮点值将自动按最小值/最大值缩放并发出警告。
形状（H: 高度, W: 宽度）：
H x W（灰度）
H x W x 1（灰度）
H x W x 3（假定为 RGB 通道顺序）
H x W x 4（假定为 RGBA 通道顺序）
- 参数
image – 要记录的图像对象。
artifact_file – 指定图像作为构件存储的路径（以 POSIX 格式），相对于运行的根目录（例如，“dir/image.png”）。此参数保留用于向后兼容，不应与 key、step 或 timestamp 同时使用。
key – 基于时间步记录图像的图像名称。此字符串只能包含字母数字、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。
step – 保存图像时的整数训练步数（迭代）。默认为 0。
timestamp – 保存图像的时间。默认为当前系统时间。
synchronous – 实验性功能 如果为 True，则阻塞直到图像成功记录为止。
import mlflow import numpy as np image = np.random.randint(0, 256, size=(100, 100, 3), dtype=np.uint8) with mlflow.start_run(): mlflow.log_image(image, key="dogs", step=3)
import mlflow from PIL import Image image = Image.new("RGB", (100, 100)) with mlflow.start_run(): mlflow.log_image(image, key="dogs", step=3)
import mlflow from PIL import Image # If you have a preexisting saved image Image.new("RGB", (100, 100)).save("image.png") image = mlflow.Image("image.png") with mlflow.start_run() as run: mlflow.log_image(run.info.run_id, image, key="dogs", step=3)
import mlflow import numpy as np image = np.random.randint(0, 256, size=(100, 100, 3), dtype=np.uint8) with mlflow.start_run(): mlflow.log_image(image, "image.png")
- mlflow.log_input(dataset: Optional[mlflow.data.dataset.Dataset] = None, context: Optional[str] = None, tags: Optional[dict[str, str]] = None, model: Optional[LoggedModelInput] = None) None[source]
记录当前运行中使用的数据集。
- 参数
dataset – 要记录的
mlflow.data.dataset.Dataset对象。context – 数据集使用的上下文。例如：“training”、“testing”。这将设置为键为 mlflow.data.context 的输入标签。
tags – 要与数据集关联的标签。标签字典 tag_key -> tag_value。
model – 要作为运行输入进行记录的
mlflow.entities.LoggedModelInput实例。
- mlflow.log_inputs(datasets: Optional[list[typing.Optional[mlflow.data.dataset.Dataset]]] = None, contexts: Optional[list[str | None]] = None, tags_list: Optional[list[dict[str, str] | None]] = None, models: Optional[list[LoggedModelInput | None]] = None) None[source]
记录当前运行中使用的一批数据集。
datasets、contexts、tags_list 列表必须具有相同的长度。这些列表中的条目可以是
None，表示对应输入的空值。- 参数
datasets – 要记录的
mlflow.data.dataset.Dataset对象列表。contexts – 数据集使用的上下文列表。例如：“training”、“testing”。这将设置为键为 mlflow.data.context 的输入标签。
tags_list – 要与数据集关联的标签列表。标签字典 tag_key -> tag_value。
models – 要作为运行输入进行记录的
mlflow.entities.LoggedModelInput实例列表。目前仅 Databricks 管理的 MLflow 支持此参数。
import numpy as np import mlflow array = np.asarray([[1, 2, 3], [4, 5, 6], [7, 8, 9]]) dataset = mlflow.data.from_numpy(array, source="data.csv") array2 = np.asarray([[-1, 2, 3], [-4, 5, 6]]) dataset2 = mlflow.data.from_numpy(array2, source="data2.csv") # Log 2 input datasets used for training and test, # the training dataset has no tag. # the test dataset has tags `{"my_tag": "tag_value"}`. with mlflow.start_run(): mlflow.log_inputs( [dataset, dataset2], contexts=["training", "test"], tags_list=[None, {"my_tag": "tag_value"}], models=None, )
- mlflow.log_issue(*, trace_id: str, issue_id: str, issue_name: Optional[str] = None, source: Optional[AssessmentSource] = None, run_id: Optional[str] = None, rationale: Optional[str] = None, metadata: Optional[dict[str, str]] = None, span_id: Optional[str] = None) Assessment[source]
记录到追踪记录（Trace）的问题引用。这会将一个追踪记录与发现的问题链接起来。此 API 仅接受关键字参数。
- 参数
trace_id – 追踪记录（trace）的 ID。
issue_id – 要引用的问题的 ID。
issue_name – 问题的名称。如果未提供，将使用问题 ID 获取问题名称。
source – 来源表示此问题在追踪记录中是如何被检测到的，例如人工审查或自动 AI 扫描。必须是
AssessmentSource的实例。如果未提供，默认为 LLM_JUDGE 源类型。run_id – 在给定追踪记录上检测到问题的运行 ID。
rationale – 问题引用的理由/正当化说明。
metadata – 问题引用的附加元数据。
span_id – 如果需要与追踪记录中的特定 Span 关联，则为与问题关联的 Span 的 ID。
- 返回
创建的问题引用评估。
- 返回类型
示例
import mlflow from mlflow.entities import AssessmentSource, AssessmentSourceType # Log an issue reference to link a trace to a discovered issue issue_ref = mlflow.log_issue( trace_id="tr-1234567890abcdef", issue_id="iss-abcdef123456", source=AssessmentSource( source_type=AssessmentSourceType.LLM_JUDGE, source_id="issue-discovery-agent" ), run_id="run-123", rationale="Response time exceeded 2 seconds threshold", )
- mlflow.log_metric(key: str, value: float, step: Optional[int] = None, synchronous: Optional[bool] = None, timestamp: Optional[int] = None, run_id: Optional[str] = None, model_id: Optional[str] = None, dataset: Optional[Union[mlflow.data.dataset.Dataset, Dataset]] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
在当前运行下记录一个指标。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
key – 指标名称。此字符串只能包含字母数字、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储都支持最长 250 个字符的键，但有些可能支持更长的键。
value – 指标值。请注意，根据存储的不同，某些特殊值（如 +/- Infinity）可能会被其他值替换。例如，SQLAlchemy 存储将 +/- Infinity 替换为最大/最小浮点值。所有后端存储都支持最长 5000 个字符的值，但有些可能支持更大的值。
step – 指标步长。如果未指定，默认为零。
synchronous – 实验性功能 如果为 True，则阻塞直到指标成功记录为止。如果为 False，则异步记录指标并返回表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 中读取，如果未设置，则默认为 False。
timestamp – 计算此指标的时间。默认为当前系统时间。
run_id – 如果指定，则将指标记录到指定的运行中。如果未指定，则将指标记录到当前活动运行中。
model_id – 与指标关联的模型 ID。如果未指定，则使用
mlflow.set_active_model()设置的当前活动模型 ID。如果不存在活动模型，则将使用与指定或活动运行关联的模型 ID。dataset – 与指标关联的数据集。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回表示记录操作 future 的 RunOperations。
- mlflow.log_metrics(metrics: dict[str, float], step: Optional[int] = None, synchronous: Optional[bool] = None, run_id: Optional[str] = None, timestamp: Optional[int] = None, model_id: Optional[str] = None, dataset: Optional[Union[mlflow.data.dataset.Dataset, Dataset]] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
在当前运行下记录多个指标。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
metrics – 指标名称：字符串 -> 值：浮点数 的字典。请注意，根据存储的不同，某些特殊值（如 +/- Infinity）可能会被其他值替换。例如，基于 SQL 的存储可能将 +/- Infinity 替换为最大/最小浮点值。
step – 记录指定指标的单个整数步长。如果未指定，每个指标将在零步长处被记录。
synchronous – 实验性功能 如果为 True，则阻塞直到指标成功记录为止。如果为 False，则异步记录指标并返回表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 中读取，如果未设置，则默认为 False。
run_id – 运行 ID。如果指定，则将指标记录到指定的运行中。如果未指定，则将指标记录到当前活动运行中。
timestamp – 计算这些指标的时间。默认为当前系统时间。
model_id – 与指标关联的模型 ID。如果未指定，则使用
mlflow.set_active_model()设置的当前活动模型 ID。如果不存在活动模型，则将使用与指定或活动运行关联的模型 ID。dataset – 与指标关联的数据集。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回表示记录操作 future 的
mlflow.utils.async_logging.run_operations.RunOperations实例。
- mlflow.log_outputs(models: Optional[list[LoggedModelOutput]] = None)[source]
将输出（例如模型）记录到活动运行中。如果没有活动运行，将创建一个新的运行。
- 参数
models – 要作为运行输出进行记录的
mlflow.entities.LoggedModelOutput实例列表。- 返回
None。
- mlflow.log_param(key: str, value: Any, synchronous: Optional[bool] = None) Any[source]
在当前运行下记录一个参数（例如模型超参数）。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
key – 参数名称。此字符串只能包含字母数字、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储都支持最长 250 个字符的键，但有些可能支持更长的键。
value – 参数值，如果不是字符串，则会被转换为字符串。所有内置后端存储都支持最长 6000 个字符的值，但有些可能支持更大的值。
synchronous – 实验性功能 如果为 True，则阻塞直到参数成功记录为止。如果为 False，则异步记录参数并返回表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 中读取，如果未设置，则默认为 False。
- 返回
当 synchronous=True 时，返回参数值。当 synchronous=False 时，返回表示记录操作 future 的
mlflow.utils.async_logging.run_operations.RunOperations实例。
- mlflow.log_params(params: dict[str, typing.Any], synchronous: Optional[bool] = None, run_id: Optional[str] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
在当前运行下记录一批参数。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
params – 参数名称：字符串 -> 值：(字符串，如果不是则会被转换为字符串) 的字典。
synchronous – 实验性功能 如果为 True，则阻塞直到参数成功记录为止。如果为 False，则异步记录参数并返回表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 中读取，如果未设置，则默认为 False。
run_id – 运行 ID。如果指定，则将参数记录到指定的运行中。如果未指定，则将参数记录到当前活动运行中。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回表示记录操作 future 的
mlflow.utils.async_logging.run_operations.RunOperations实例。
- mlflow.log_stream(stream: io.BufferedIOBase | io.RawIOBase, artifact_file: str, run_id: Optional[str] = None) None[source]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
将二进制类文件对象（例如
io.BytesIO）记录为构件。- 参数
stream – 一个支持
.read()方法的二进制类文件对象（例如io.BytesIO）。artifact_file – 以 posixpath 格式表示的、相对于运行的构件文件路径，用于保存流内容（例如“dir/file.bin”）。
run_id – 如果指定，则将构件记录到指定的运行中。如果未指定，则将构件记录到当前活动运行中。
- mlflow.log_table(data: Union[dict[str, typing.Any], pandas.DataFrame], artifact_file: str, run_id: str | None = None) None[source]
将表记录到 MLflow 跟踪中作为 JSON 构件。如果 artifact_file 在运行中已存在，数据将追加到现有的 artifact_file 中。
- 参数
data – 要记录的字典或 pandas.DataFrame。
artifact_file – 以 posixpath 格式表示的、相对于运行的构件文件路径，用于保存表格（例如“dir/file.json”）。
run_id – 如果指定，则将表记录到指定的运行中。如果未指定，则将表记录到当前活动运行中。
import mlflow table_dict = { "inputs": ["What is MLflow?", "What is Databricks?"], "outputs": ["MLflow is ...", "Databricks is ..."], "toxicity": [0.0, 0.0], } with mlflow.start_run(): # Log the dictionary as a table mlflow.log_table(data=table_dict, artifact_file="qabot_eval_results.json")
import mlflow import pandas as pd table_dict = { "inputs": ["What is MLflow?", "What is Databricks?"], "outputs": ["MLflow is ...", "Databricks is ..."], "toxicity": [0.0, 0.0], } df = pd.DataFrame.from_dict(table_dict) with mlflow.start_run(): # Log the df as a table mlflow.log_table(data=df, artifact_file="qabot_eval_results.json")
- mlflow.log_text(text: str, artifact_file: str, run_id: Optional[str] = None) None[source]
将文本记录为构件。
- 参数
text – 包含要记录的文本的字符串。
artifact_file – 以 posixpath 格式表示的、相对于运行的构件文件路径，用于保存文本（例如“dir/file.txt”）。
run_id – 如果指定，则将构件记录到指定的运行中。如果未指定，则将构件记录到当前活动运行中。
import mlflow with mlflow.start_run(): # Log text to a file under the run's root artifact directory mlflow.log_text("text1", "file1.txt") # Log text in a subdirectory of the run's root artifact directory mlflow.log_text("text2", "dir/file2.txt") # Log HTML text mlflow.log_text("<h1>header</h1>", "index.html")
- mlflow.log_trace(name: str = 'Task', request: Optional[Any] = None, response: Optional[[Any] = None, intermediate_outputs: dict[str, typing.Any] | None = None, attributes: dict[str, typing.Any] | None = None, tags: dict[str, str] | None = None, start_time_ms: int | None = None, execution_time_ms: int | None = None) str[source]
警告
mlflow.tracing.fluent.log_trace自 3.6.0 版本起已被弃用。此方法将在未来的版本中删除。创建一个具有单个根 Span 的追踪记录。当您想要记录任意的（请求，响应）对而无需结构化的 OpenTelemetry Span 时，此 API 非常有用。该追踪记录链接到活动实验。
- 参数
name – 追踪记录（以及根 Span）的名称。默认为“Task”。
request – 整个追踪记录的输入数据。这也设置在追踪记录的根 Span 上。
response – 整个追踪记录的输出数据。这也设置在追踪记录的根 Span 上。
intermediate_outputs – 模型或代理在处理请求时产生的中间输出的字典。键是输出的名称，值是输出本身。值必须是 JSON 可序列化的。
attributes – 要在追踪根 span 上设置的属性字典。
tags – 要在追踪上设置的标签字典。
start_time_ms – 追踪的开始时间（自 UNIX 纪元起的毫秒数）。若未指定，则使用当前时间作为追踪的开始和结束时间。
execution_time_ms – 追踪的执行时间（自 UNIX 纪元起的毫秒数）。
- 返回
已记录追踪的 ID。
示例
import time import mlflow trace_id = mlflow.log_trace( request="Does mlflow support tracing?", response="Yes", intermediate_outputs={ "retrieved_documents": ["mlflow documentation"], "system_prompt": ["answer the question with yes or no"], }, start_time_ms=int(time.time() * 1000), execution_time_ms=5129, ) trace = mlflow.get_trace(trace_id, flush=True) print(trace.data.intermediate_outputs)
- mlflow.login(backend: str = 'databricks', interactive: bool = True) None[source]
配置 MLflow 服务器身份验证并将 MLflow 连接到追踪服务器。
此方法提供了一种将 MLflow 连接到其追踪服务器的简便方式。目前仅支持 Databricks 追踪服务器。如果找不到现有的 Databricks 配置文件，系统将提示用户输入凭据，凭据将保存到 ~/.databrickscfg 中。
- 参数
backend – 字符串，追踪服务器的后端。目前仅支持 “databricks”。
interactive – 布尔值，控制在缺少凭据时是否请求用户输入。如果为 true，当找不到凭据时将请求用户输入；否则，如果找不到凭据，将引发异常。
- mlflow.override_feedback(*, trace_id: str, assessment_id: str, value: float | int | str | bool | dict[str, float | int | str | bool] | list[float | int | str | bool], rationale: Optional[str] = None, source: Optional[AssessmentSource] = None, metadata: Optional[dict[str, typing.Any]] = None) Assessment[source]
使用新的评估覆盖现有的反馈评估。此 API 会记录一个新的评估，并将 overrides 字段设置为提供的评估 ID。原始评估将被标记为无效，但除此之外保持不变。当你想要纠正由 LLM 判断器生成的评估，但又希望保留原始评估以供后续判断器微调时，这非常有用。
如果你想原地修改评估，请改用
update_assessment()。- 参数
trace_id – 追踪记录（trace）的 ID。
assessment_id – 要覆盖的评估的 ID。
value – 评估的新值。
rationale – 新评估的基本原理。
source – 新评估的来源。
metadata – 新评估的附加元数据。
- 返回
创建后的评估。
- 返回类型
示例
import mlflow from mlflow.entities import AssessmentSource, AssessmentSourceType # First, log an initial LLM-generated feedback as a simulation llm_feedback = mlflow.log_feedback( trace_id="tr-1234567890abcdef", name="relevance", value=0.6, source=AssessmentSource( source_type=AssessmentSourceType.LLM_JUDGE, source_id="gpt-4" ), rationale="Response partially addresses the question", ) # Later, a human reviewer disagrees and wants to override corrected_assessment = mlflow.override_feedback( trace_id="tr-1234567890abcdef", assessment_id=llm_feedback.assessment_id, value=0.9, rationale="Response fully addresses the question with good examples", source=AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="expert_reviewer@company.com" ), metadata={ "override_reason": "LLM underestimated relevance", "review_date": "2024-01-15", "confidence": "high", }, )
- mlflow.register_model(model_uri, name, await_registration_for=300, *, tags: Optional[dict[str, typing.Any]] = None, env_pack: Optional[Union[Literal['databricks_model_serving'], mlflow.utils.env_pack.EnvPackConfig]] = None) ModelVersion[source]
在模型注册表中为
model_uri指定的模型文件创建一个新的模型版本。请注意，此方法假定模型注册表后端 URI 与追踪后端 URI 相同。
- 参数
model_uri –
指向 MLmodel 目录的 URI。支持的 URI 方案包括
runs:/URIs（例如runs:/<run_id>/<artifact_path>），用于从特定运行中注册模型。运行 ID 会与模型版本一起记录。models:/URIs，支持两种形式models:/<model_name>/<version>用于推广现有的已注册模型版本。当引用的模型版本具有关联的源运行时，将保留源运行谱系。models:/<model_id>用于从已记录的模型（例如log_model返回的模型）创建新的已注册模型版本。保留源运行谱系。
本地文件系统路径，用于注册先前使用
save_model保存的本地持久化 MLflow 模型。
name – 在其下创建新模型版本的已注册模型的名称。如果给定名称的已注册模型不存在，则会自动创建它。
await_registration_for – 等待模型版本完成创建并处于
READY状态的秒数。默认情况下，函数等待五分钟。指定 0 或 None 可跳过等待。tags – 键值对字典，转换为
mlflow.entities.model_registry.ModelVersionTag对象。env_pack –
字符串或 EnvPackConfig。如果指定，模型依赖项会先可选地安装到当前的 Python 环境中，然后整个环境将被打包并包含在已注册的模型工件中。如果使用快捷方式字符串 “databricks_model_serving”，则模型依赖项将安装在当前环境中。这在将模型部署到 Databricks Model Serving 等服务环境时非常有用。
注意
Experimental: This parameter may change or be removed in a future release without warning. (实验性：此参数可能在未来的版本中更改或移除，恕不另行通知。)
- 返回
由后端创建的单个
mlflow.entities.model_registry.ModelVersion对象。
import mlflow.sklearn from mlflow.models import infer_signature from sklearn.datasets import make_regression from sklearn.ensemble import RandomForestRegressor mlflow.set_tracking_uri("sqlite:////tmp/mlruns.db") params = {"n_estimators": 3, "random_state": 42} X, y = make_regression(n_features=4, n_informative=2, random_state=0, shuffle=False) # Log MLflow entities with mlflow.start_run() as run: rfr = RandomForestRegressor(**params).fit(X, y) signature = infer_signature(X, rfr.predict(X)) mlflow.log_params(params) mlflow.sklearn.log_model(rfr, name="sklearn-model", signature=signature) model_uri = f"runs:/{run.info.run_id}/sklearn-model" mv = mlflow.register_model(model_uri, "RandomForestRegressionModel") print(f"Name: {mv.name}") print(f"Version: {mv.version}")
- mlflow.run(uri, entry_point='main', version=None, parameters=None, docker_args=None, experiment_name=None, experiment_id=None, backend='local', backend_config=None, storage_dir=None, synchronous=True, run_id=None, run_name=None, env_manager=None, build_image=False, docker_auth=None)[source]
运行 MLflow 项目。项目可以是本地的，也可以存储在 Git URI 中。
MLflow 内置支持在本地运行项目，或在 Databricks 或 Kubernetes 集群上远程运行。你还可以通过安装适当的第三方插件在其他目标上运行项目。有关详细信息，请参阅 社区插件。
有关在链式工作流中使用此方法的信息，请参阅 构建多步骤工作流。
- 引发
mlflow.exceptions.ExecutionException – 执行不成功。
- 参数
uri – 要运行的项目的 URI。指向包含 MLproject 文件的项目目录的本地文件系统路径或 Git 仓库 URI（例如 https://github.com/mlflow/mlflow-example）。
entry_point – 项目中要运行的入口点。如果找不到指定名称的入口点，则会将项目文件
entry_point作为脚本运行，使用 “python” 运行.py文件，使用默认 shell（由环境变量$SHELL指定）运行.sh文件。version – 对于基于 Git 的项目，可以是 commit 哈希值或分支名称。
parameters – 入口点命令的参数（字典）。
docker_args – docker 命令的参数（字典）。
experiment_name – 用于启动运行的实验名称。
experiment_id – 用于启动运行的实验 ID。
backend – 运行的执行后端：MLflow 内置支持 “local”、“databricks” 和 “kubernetes”（实验性）后端。如果在 Databricks 上运行，将针对确定的 Databricks 工作区运行：如果已设置
databricks://profile形式的 Databricks 追踪 URI（例如通过设置 MLFLOW_TRACKING_URI 环境变量），将针对 <profile> 指定的工作区运行。否则，将针对默认 Databricks CLI 配置文件指定的工作区运行。backend_config – 一个字典或 JSON 文件路径（必须以 ‘.json’ 结尾），将作为配置传递给后端。应提供的内容对于每个执行后端都是不同的，并在 https://www.mlflow.org/docs/latest/projects.html 中有详细说明。
storage_dir – 仅在
backend为 “local” 时使用。MLflow 将传递给path类型参数的分布式 URI 中的工件下载到storage_dir的子目录中。synchronous – 是否在等待运行完成时阻塞。默认为 True。请注意，如果
synchronous为 False 且backend为 “local”，此方法将返回，但当前进程在退出时会阻塞直到本地运行完成。如果当前进程被中断，通过此方法启动的任何异步运行都将被终止。如果synchronous为 True 且运行失败，当前进程也会报错。run_id – 注意：此参数由 MLflow 项目 API 内部使用，不应指定。如果指定，将使用该运行 ID 而不是创建一个新的运行。
run_name – 与项目执行关联的 MLflow 运行的名称。如果为
None，则不设置 MLflow 运行名称。env_manager –
指定环境管理器以创建新的运行环境并在其中安装项目依赖项。支持以下值
local: 使用本地环境
virtualenv: 使用 virtualenv（以及用于 Python 版本管理的 pyenv）
uv: 使用 uv
conda: 使用 conda
如果未指定，MLflow 会通过检查项目目录中的文件自动确定要使用的环境管理器。例如，如果存在
python_env.yaml，则将使用 virtualenv。build_image – 是否构建项目的新 docker 镜像或重用现有镜像。默认：False（重用现有镜像）
docker_auth – 表示用于 Docker 注册表身份验证信息的字典。有关可用选项，请参阅 docker.client.DockerClient.login。
- 返回
mlflow.projects.SubmittedRun，公开关于启动运行的信息（例如运行 ID）。
import mlflow project_uri = "https://github.com/mlflow/mlflow-example" params = {"alpha": 0.5, "l1_ratio": 0.01} # Run MLflow project and create a reproducible conda environment # on a local host mlflow.run(project_uri, parameters=params)
- mlflow.search_experiments(view_type: int = 1, max_results: Optional[int] = None, filter_string: Optional[str] = None, order_by: Optional[list[str]] = None) list[Experiment][source]
搜索与指定搜索查询匹配的实验。
- 参数
view_type –
mlflow.entities.ViewType中定义的枚举值之一：ACTIVE_ONLY、DELETED_ONLY或ALL。max_results – 如果传递，则指定期望的最大实验数量。如果未传递，将返回所有实验。
filter_string –
过滤查询字符串（例如
"name = 'my_experiment'"），默认为搜索所有实验。支持以下标识符、比较器和逻辑运算符。- 标识符
name: 实验名称creation_time: 实验创建时间last_update_time: 实验最后更新时间tags.<tag_key>: 实验标签。如果tag_key包含空格，则必须用反引号包裹（例如"tags.`extra key`"）。
- 字符串属性和标签的比较器
=: 等于!=: 不等于LIKE: 区分大小写的模式匹配ILIKE: 不区分大小写的模式匹配
- 数值属性的比较器
=: 等于!=: 不等于<: 小于<=: 小于等于>: 大于>=: 大于等于
- 逻辑运算符
AND: 组合两个子查询，如果两者都为真，则返回 True。
order_by –
要排序的列列表。
order_by列可以包含可选的DESC或ASC值（例如"name DESC"）。默认排序为ASC，因此"name"等同于"name ASC"。如果未指定，默认为["last_update_time DESC"]，即优先列出最近更新的实验。支持以下字段experiment_id: 实验 IDname: 实验名称creation_time: 实验创建时间last_update_time: 实验最后更新时间
- 返回
Experiment对象列表。
import mlflow def assert_experiment_names_equal(experiments, expected_names): actual_names = [e.name for e in experiments if e.name != "Default"] assert actual_names == expected_names, (actual_names, expected_names) mlflow.set_tracking_uri("sqlite:///:memory:") # Create experiments for name, tags in [ ("a", None), ("b", None), ("ab", {"k": "v"}), ("bb", {"k": "V"}), ]: mlflow.create_experiment(name, tags=tags) # Search for experiments with name "a" experiments = mlflow.search_experiments(filter_string="name = 'a'") assert_experiment_names_equal(experiments, ["a"]) # Search for experiments with name starting with "a" experiments = mlflow.search_experiments(filter_string="name LIKE 'a%'") assert_experiment_names_equal(experiments, ["ab", "a"]) # Search for experiments with tag key "k" and value ending with "v" or "V" experiments = mlflow.search_experiments(filter_string="tags.k ILIKE '%v'") assert_experiment_names_equal(experiments, ["bb", "ab"]) # Search for experiments with name ending with "b" and tag {"k": "v"} experiments = mlflow.search_experiments(filter_string="name LIKE '%b' AND tags.k = 'v'") assert_experiment_names_equal(experiments, ["ab"]) # Sort experiments by name in ascending order experiments = mlflow.search_experiments(order_by=["name"]) assert_experiment_names_equal(experiments, ["a", "ab", "b", "bb"]) # Sort experiments by ID in descending order experiments = mlflow.search_experiments(order_by=["experiment_id DESC"]) assert_experiment_names_equal(experiments, ["bb", "ab", "b", "a"])
- mlflow.search_model_versions(max_results: Optional[int] = None, filter_string: Optional[str] = None, order_by: Optional[list[str]] = None) list[ModelVersion][source]
搜索满足过滤条件的模型版本。
- 参数
max_results – 如果传递，则指定期望的最大模型数量。如果未传递，将返回所有模型。
filter_string –
过滤查询字符串（例如
"name = 'a_model_name' and tag.key = 'value1'"），默认为搜索所有模型版本。支持以下标识符、比较器和逻辑运算符。- 标识符
name: 模型名称。source_path: 模型版本源路径。run_id: 生成该模型版本的 mlflow 运行 ID。tags.<tag_key>: 模型版本标签。如果tag_key包含空格，则必须用反引号包裹（例如"tags.`extra key`"）。
- 比较器
=: 等于。!=: 不等于。LIKE: 区分大小写的模式匹配。ILIKE: 不区分大小写的模式匹配。IN: 在值列表中。只有run_id标识符支持IN比较器。
- 逻辑运算符
AND: 组合两个子查询，如果两者都为真，则返回 True。
order_by – 带有 ASC|DESC 注释的列名列表，用于对匹配的搜索结果进行排序。
- 返回
- 满足搜索表达式的
mlflow.entities.model_registry.ModelVersion对象列表。 满足搜索表达式。
- 满足搜索表达式的
import mlflow from sklearn.linear_model import LogisticRegression for _ in range(2): with mlflow.start_run(): mlflow.sklearn.log_model( LogisticRegression(), name="Cordoba", registered_model_name="CordobaWeatherForecastModel", ) # Get all versions of the model filtered by name filter_string = "name = 'CordobaWeatherForecastModel'" results = mlflow.search_model_versions(filter_string=filter_string) print("-" * 80) for res in results: print(f"name={res.name}; run_id={res.run_id}; version={res.version}") # Get the version of the model filtered by run_id filter_string = "run_id = 'ae9a606a12834c04a8ef1006d0cff779'" results = mlflow.search_model_versions(filter_string=filter_string) print("-" * 80) for res in results: print(f"name={res.name}; run_id={res.run_id}; version={res.version}")
-------------------------------------------------------------------------------- name=CordobaWeatherForecastModel; run_id=ae9a606a12834c04a8ef1006d0cff779; version=2 name=CordobaWeatherForecastModel; run_id=d8f028b5fedf4faf8e458f7693dfa7ce; version=1 -------------------------------------------------------------------------------- name=CordobaWeatherForecastModel; run_id=ae9a606a12834c04a8ef1006d0cff779; version=2
- mlflow.search_registered_models(max_results: Optional[int] = None, filter_string: Optional[str] = None, order_by: Optional[list[str]] = None) list[RegisteredModel][source]
搜索满足过滤条件的已注册模型。
- 参数
max_results – 如果传递，则指定期望的最大模型数量。如果未传递，将返回所有模型。
filter_string –
过滤查询字符串（例如 “name = ‘a_model_name’ and tag.key = ‘value1’”），默认为搜索所有已注册模型。支持以下标识符、比较器和逻辑运算符。
- 标识符
”name”: 已注册模型名称。
”tags.<tag_key>”: 已注册模型标签。如果 “tag_key” 包含空格，则必须用反引号包裹（例如 “tags.`extra key`”）。
- 比较器
”=”: 等于。
”!=”: 不等于。
”LIKE”: 区分大小写的模式匹配。
”ILIKE”: 不区分大小写的模式匹配。
- 逻辑运算符
”AND”: 组合两个子查询，如果两者都为真，则返回 True。
order_by – 带有 ASC|DESC 注释的列名列表，用于对匹配的搜索结果进行排序。
- 返回
满足搜索表达式的
mlflow.entities.model_registry.RegisteredModel对象列表。
import mlflow from sklearn.linear_model import LogisticRegression with mlflow.start_run(): mlflow.sklearn.log_model( LogisticRegression(), name="Cordoba", registered_model_name="CordobaWeatherForecastModel", ) mlflow.sklearn.log_model( LogisticRegression(), name="Boston", registered_model_name="BostonWeatherForecastModel", ) # Get search results filtered by the registered model name filter_string = "name = 'CordobaWeatherForecastModel'" results = mlflow.search_registered_models(filter_string=filter_string) print("-" * 80) for res in results: for mv in res.latest_versions: print(f"name={mv.name}; run_id={mv.run_id}; version={mv.version}") # Get search results filtered by the registered model name that matches # prefix pattern filter_string = "name LIKE 'Boston%'" results = mlflow.search_registered_models(filter_string=filter_string) print("-" * 80) for res in results: for mv in res.latest_versions: print(f"name={mv.name}; run_id={mv.run_id}; version={mv.version}") # Get all registered models and order them by ascending order of the names results = mlflow.search_registered_models(order_by=["name ASC"]) print("-" * 80) for res in results: for mv in res.latest_versions: print(f"name={mv.name}; run_id={mv.run_id}; version={mv.version}")
-------------------------------------------------------------------------------- name=CordobaWeatherForecastModel; run_id=248c66a666744b4887bdeb2f9cf7f1c6; version=1 -------------------------------------------------------------------------------- name=BostonWeatherForecastModel; run_id=248c66a666744b4887bdeb2f9cf7f1c6; version=1 -------------------------------------------------------------------------------- name=BostonWeatherForecastModel; run_id=248c66a666744b4887bdeb2f9cf7f1c6; version=1 name=CordobaWeatherForecastModel; run_id=248c66a666744b4887bdeb2f9cf7f1c6; version=1
- mlflow.search_runs(experiment_ids: list[str] | None = None, filter_string: str = '', run_view_type: int = 1, max_results: int = 100000, order_by: list[str] | None = None, output_format: str = 'pandas', search_all_experiments: bool = False, experiment_names: list[str] | None = None) Union[list[Run], pandas.DataFrame][source]
搜索符合指定条件的运行（Runs）。
- 参数
experiment_ids – 实验 ID 列表。搜索可以使用实验 ID 或实验名称，但不能在同一次调用中同时使用两者。如果
experiment_names同时也不为None或[]，则使用除None或[]以外的值将导致错误。如果experiment_names为None或[]，则默认使用当前激活的实验。filter_string – 过滤查询字符串，默认为搜索所有运行。
run_view_type –
mlflow.entities.ViewType中定义的ACTIVE_ONLY、DELETED_ONLY或ALL运行枚举值之一。max_results – 放入数据框中的最大运行数量。默认为 100,000，以避免在用户机器上导致内存不足问题。
order_by – 要排序的列列表（例如 “metrics.rmse”）。
order_by列可以包含可选的DESC或ASC值。默认为ASC。默认排序顺序为先按start_time DESC排序，然后按run_id排序。output_format – 要返回的输出格式。如果为
pandas，则返回一个pandas.DataFrame；如果为list，则返回一个mlflow.entities.Run列表。search_all_experiments – 指定是否搜索所有实验的布尔值。仅在
experiment_ids为[]或None时生效。experiment_names – 实验名称列表。搜索可以使用实验 ID 或实验名称，但不能在同一次调用中同时使用两者。如果
experiment_ids同时也不为None或[]，则使用除None或[]以外的值将导致错误。如果experiment_ids为None或[]，则默认使用当前激活的实验。
- 返回
一个
mlflow.entities.Run列表。如果 output_format 为pandas：返回运行的pandas.DataFrame，其中每个指标、参数和标签分别展开到名为 metrics.*、params.* 或 tags.* 的独立列中。对于没有特定指标、参数或标签的运行，相应列的值分别为（NumPy）Nan、None或None。- 返回类型
如果 output_format 为
listimport mlflow # Create an experiment and log two runs under it experiment_name = "Social NLP Experiments" experiment_id = mlflow.create_experiment(experiment_name) with mlflow.start_run(experiment_id=experiment_id): mlflow.log_metric("m", 1.55) mlflow.set_tag("s.release", "1.1.0-RC") with mlflow.start_run(experiment_id=experiment_id): mlflow.log_metric("m", 2.50) mlflow.set_tag("s.release", "1.2.0-GA") # Search for all the runs in the experiment with the given experiment ID df = mlflow.search_runs([experiment_id], order_by=["metrics.m DESC"]) print(df[["metrics.m", "tags.s.release", "run_id"]]) print("--") # Search the experiment_id using a filter_string with tag # that has a case insensitive pattern filter_string = "tags.s.release ILIKE '%rc%'" df = mlflow.search_runs([experiment_id], filter_string=filter_string) print(df[["metrics.m", "tags.s.release", "run_id"]]) print("--") # Search for all the runs in the experiment with the given experiment name df = mlflow.search_runs(experiment_names=[experiment_name], order_by=["metrics.m DESC"]) print(df[["metrics.m", "tags.s.release", "run_id"]])
metrics.m tags.s.release run_id 0 2.50 1.2.0-GA 147eed886ab44633902cc8e19b2267e2 1 1.55 1.1.0-RC 5cc7feaf532f496f885ad7750809c4d4 -- metrics.m tags.s.release run_id 0 1.55 1.1.0-RC 5cc7feaf532f496f885ad7750809c4d4 -- metrics.m tags.s.release run_id 0 2.50 1.2.0-GA 147eed886ab44633902cc8e19b2267e2 1 1.55 1.1.0-RC 5cc7feaf532f496f885ad7750809c4d4
- mlflow.search_sessions(max_results: int = 100, run_id: Optional[str] = None, model_id: Optional[str] = None, include_spans: bool = True, locations: Optional[list[str]] = None) list[Session][source]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
返回与给定搜索条件匹配的完整会话。
会话是共享相同会话 ID 的一组追踪，通常代表多轮对话或一系列相关交互。此 API 通过首先从追踪中识别唯一会话 ID，然后并行获取每个会话所属的所有追踪来检索完整会话。
- 参数
max_results – 要返回的最大会话数。默认为 100。
run_id – 用于限定搜索范围的运行 ID。当在一个活动的运行下创建追踪时，它将与该运行关联，你可以通过运行 ID 进行过滤以检索追踪。
model_id – 如果指定，则搜索与给定模型 ID 关联的追踪。
include_spans – 如果为
True，则在返回的追踪中包含 span。否则，仅返回追踪元数据。默认为True。locations – 要搜索的位置列表。要在实验中搜索，请提供实验 ID 列表。要在 Databricks 上的 UC 表中搜索，请以 <catalog_name>.<schema_name>[.<table_prefix>] 格式提供位置列表。如果未提供，搜索将在当前活动的实验中执行。
- 返回
Session对象列表，其中每个会话包含共享相同会话 ID 的Trace对象。会话按其首次追踪的时间戳排序（最近的在前）。每个 Session 对象通过session.id提供方便的访问，并支持使用for trace in session进行迭代。
import mlflow # Get all sessions from the current experiment sessions = mlflow.search_sessions() # Each session provides convenient access to ID and traces for session in sessions: print(f"Session {session.id} has {len(session)} traces") for trace in session: print(f" Trace: {trace.info.trace_id}")
- mlflow.set_experiment(experiment_name: Optional[str] = None, experiment_id: Optional[str] = None, trace_location: Optional[UnityCatalog] = None) Experiment[source]
将给定的实验设置为活动实验。必须通过 experiment_name 指定名称或通过 experiment_id 指定 ID 来指定实验。实验名称和 ID 不能同时指定。
注意
如果按名称设置的实验不存在，将创建一个具有给定名称的新实验。创建实验后，它将被设置为活动实验。在某些平台（如 Databricks）上，实验名称必须是绝对路径，例如
"/Users/<username>/my-experiment"。- 参数
experiment_name – 要激活的实验的区分大小写的名称。
experiment_id – 要激活的实验的 ID。如果不存在具有此 ID 的实验，则会抛出异常。
trace_location – 用于配置实验衍生追踪目的地的可选 UC 追踪位置。必须是
mlflow.entities.trace_location.UnityCatalog(...)的实例。
- 返回
表示新的活动实验的
mlflow.entities.Experiment实例。
import mlflow # Set an experiment name, which must be unique and case-sensitive. experiment = mlflow.set_experiment("Social NLP Experiments") # Get Experiment Details print(f"Experiment_id: {experiment.experiment_id}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Tags: {experiment.tags}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}")
- mlflow.set_experiment_tag(key: str, value: Any) None[source]
在当前实验上设置一个标签。值会被转换为字符串。
- 参数
key – 标签名称。此字符串只能包含字母数字、下划线 (_)、短横线 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储都支持长度最大为 250 的键，但有些可能支持更大的键。
value – 标签值，如果不是字符串则会被转换为字符串。所有后端存储都支持长度最大为 5000 的值，但有些可能支持更大的值。
- mlflow.set_experiment_tags(tags: dict[str, typing.Any]) None[source]
为当前活动的实验设置标签。
- 参数
tags – 包含标签名称和相应值的字典。
- mlflow.set_model_version_tag(name: str, version: Optional[str] = None, key: Optional[str] = None, value: Optional[Any] = None) None[source]
为模型版本设置一个标签。
- 参数
name – 已注册模型名称。
version – 已注册模型版本。
key – 要记录的标签键。key 是必需的。
value – 要记录的标签值。value 是必需的。
- mlflow.set_registry_uri(uri: str) None[source]
设置注册表服务器 URI。如果你的注册表服务器与追踪服务器不同，此方法特别有用。
- 参数
uri – 空字符串，或带有
file:/前缀的本地文件路径。数据存储在提供的文件的本地（如果为空，则为./mlruns）。HTTP URI，如https://my-tracking-server:5000或http://my-oss-uc-server:8080。Databricks 工作区，作为字符串 “databricks” 提供，或者为了使用 Databricks CLI 配置文件，提供 “databricks://<profileName>”。
import mflow # Set model registry uri, fetch the set uri, and compare # it with the tracking uri. They should be different mlflow.set_registry_uri("sqlite:////tmp/registry.db") mr_uri = mlflow.get_registry_uri() print(f"Current registry uri: {mr_uri}") tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}") # They should be different assert tracking_uri != mr_uri
- mlflow.set_system_metrics_node_id(node_id)[source]
设置系统指标节点 ID。
node_id 是收集指标的机器的标识符。这在多节点（分布式训练）设置中很有用。
- mlflow.set_system_metrics_samples_before_logging(samples)[source]
设置记录系统指标之前的采样数。
每次收集了 samples 个样本后，系统指标将记录到 mlflow。默认情况下 samples=1。
- mlflow.set_system_metrics_sampling_interval(interval)[source]
设置系统指标采样间隔。
每 interval 秒，将收集系统指标。默认情况下 interval=10。
- mlflow.set_tag(key: str, value: Any, synchronous: Optional[bool] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
在当前运行下设置一个标签。如果没有活动的运行，此方法将创建一个新的活动运行。
- 参数
key – 标签名称。此字符串只能包含字母数字、下划线 (_)、短横线 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储都支持长度最大为 250 的键，但有些可能支持更大的键。
value – 标签值，如果不是字符串则会被转换为字符串。所有后端存储都支持长度最大为 5000 的值，但有些可能支持更大的值。
synchronous – 实验性 如果为 True，则阻塞直到标签成功记录。如果为 False，则异步记录标签并返回表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 读取，如果未设置，则默认为 False。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回表示记录操作 future 的
mlflow.utils.async_logging.run_operations.RunOperations实例。
- mlflow.set_tags(tags: dict[str, typing.Any], synchronous: Optional[bool] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
为当前运行记录一批标签。如果没有活动的运行，此方法将创建一个新的活动运行。
- 参数
tags – 标签名称: 字符串 -> 值: 字典（字符串，如果不是字符串则会被转换为字符串）
synchronous – 实验性 如果为 True，则阻塞直到标签成功记录。如果为 False，则异步记录标签并返回表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 读取，如果未设置，则默认为 False。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回表示记录操作 future 的
mlflow.utils.async_logging.run_operations.RunOperations实例。
- mlflow.set_trace_tag(trace_id: str, key: str, value: str)[source]
注意
参数
request_id已弃用。请改用trace_id。在具有给定追踪 ID 的追踪上设置一个标签。
追踪可以是活动的，也可以是已经结束并记录在后端中的追踪。以下是在活动追踪上设置标签的示例。你可以替换
trace_id参数以在已经结束的追踪上设置标签。import mlflow with mlflow.start_span(name="span") as span: mlflow.set_trace_tag(span.trace_id, "key", "value")
- 参数
trace_id – 要设置标签的追踪的 ID。
key – 标签的字符串键。长度最多为 250 个字符，否则在存储时将被截断。
value – 标签的字符串值。长度最多为 250 个字符，否则在存储时将被截断。
- mlflow.set_tracking_uri(uri: str | pathlib.Path) None[source]
设置追踪服务器 URI。这不会影响当前活动的运行（如果存在），但会对后续运行生效。
- 参数
uri –
空字符串，或带有
file:/前缀的本地文件路径。数据存储在提供的文件的本地（如果为空，则为./mlruns）。HTTP URI，如
https://my-tracking-server:5000。Databricks 工作区，作为字符串 “databricks” 提供，或者为了使用 Databricks CLI 配置文件，提供 “databricks://<profileName>”。
pathlib.Path实例
import mlflow mlflow.set_tracking_uri("file:///tmp/my_tracking") tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}")
- mlflow.set_workspace(workspace: str | None) None[source]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
为后续的 MLflow 操作设置活动工作区。
- mlflow.start_run(run_id: Optional[str] = None, experiment_id: Optional[str] = None, run_name: Optional[str] = None, nested: bool = False, parent_run_id: Optional[str] = None, tags: Optional[dict[str, typing.Any]] = None, description: Optional[str] = None, log_system_metrics: Optional[bool] = None) ActiveRun[源码]
开启一个新的 MLflow 运行，并将其设置为活动运行，在该运行下将记录指标和参数。返回值可用作
with块内的上下文管理器；否则，您必须调用end_run()来终止当前运行。如果您传递了
run_id或者设置了MLFLOW_RUN_ID环境变量，start_run将尝试恢复具有指定运行 ID 的运行，其他参数将被忽略。run_id的优先级高于MLFLOW_RUN_ID。如果恢复现有的运行，则运行状态将被设置为
RunStatus.RUNNING。MLflow 会在运行上设置各种默认标签，具体定义见 MLflow 系统标签。
- 参数
run_id – 如果指定，获取具有指定 UUID 的运行，并在该运行下记录参数和指标。运行的结束时间不会被重置，其状态设置为运行中，但运行的其他属性（
source_version,source_type等）不会改变。experiment_id – 要在其中创建当前运行的实验 ID（仅在未指定
run_id时适用）。如果未指定experiment_id参数，将按以下顺序查找有效的实验：通过set_experiment激活的实验、MLFLOW_EXPERIMENT_NAME环境变量、MLFLOW_EXPERIMENT_ID环境变量，或由跟踪服务器定义的默认实验。run_name – 新运行的名称，应为非空字符串。仅在未指定
run_id时使用。如果创建了新运行且未指定run_name，则会为该运行生成一个随机名称。nested – 控制运行是否嵌套在父运行中。
True会创建一个嵌套运行。parent_run_id – 如果指定，当前运行将嵌套在具有指定 UUID 的运行下。父运行必须处于 ACTIVE 状态。
tags – 一个可选的字符串键值对字典，用于在运行上设置标签。如果正在恢复运行，则会在已恢复的运行上设置这些标签。如果正在创建新运行，则会在新运行上设置这些标签。
description – 一个可选的字符串，用于填充运行的描述框。如果正在恢复运行，描述将在已恢复的运行上设置。如果正在创建新运行，描述将在新运行上设置。
log_system_metrics – 布尔值，默认为 None。如果为 True，系统指标（如 CPU/GPU 利用率）将被记录到 MLflow 中。如果为 None，我们将检查环境变量 MLFLOW_ENABLE_SYSTEM_METRICS_LOGGING 以确定是否记录系统指标。系统指标记录是 MLflow 2.8 中的一项实验性功能，可能会发生变化。
- 返回
mlflow.ActiveRun对象，它充当封装运行状态的上下文管理器。
import mlflow # Create nested runs experiment_id = mlflow.create_experiment("experiment1") with mlflow.start_run( run_name="PARENT_RUN", experiment_id=experiment_id, tags={"version": "v1", "priority": "P1"}, description="parent", ) as parent_run: mlflow.log_param("parent", "yes") with mlflow.start_run( run_name="CHILD_RUN", experiment_id=experiment_id, description="child", nested=True, ) as child_run: mlflow.log_param("child", "yes") print("parent run:") print(f"run_id: {parent_run.info.run_id}") print("description: {}".format(parent_run.data.tags.get("mlflow.note.content"))) print("version tag value: {}".format(parent_run.data.tags.get("version"))) print("priority tag value: {}".format(parent_run.data.tags.get("priority"))) print("--") # Search all child runs with a parent id query = f"tags.mlflow.parentRunId = '{parent_run.info.run_id}'" results = mlflow.search_runs(experiment_ids=[experiment_id], filter_string=query) print("child runs:") print(results[["run_id", "params.child", "tags.mlflow.runName"]]) # Create a nested run under the existing parent run with mlflow.start_run( run_name="NEW_CHILD_RUN", experiment_id=experiment_id, description="new child", parent_run_id=parent_run.info.run_id, ) as child_run: mlflow.log_param("new-child", "yes")
- mlflow.test(fn: Optional[Callable[[~ _P], mlflow.pytest.decorator._R]] = None) Callable[[~_P], mlflow.pytest.decorator._R][源码]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
为 MLflow pytest 插件标记一个测试函数。
支持裸
@mlflow.test和@mlflow.test()。如果插件未启用，则会在测试时引发异常，而不是在没有 MLflow 运行/跟踪管理的情况下静默运行测试。
- mlflow.update_current_trace(tags: Optional[dict[str, str]] = None, metadata: Optional[dict[str, str]] = None, client_request_id: Optional[str] = None, request_preview: Optional[str] = None, response_preview: Optional[str] = None, state: Optional[Union[TraceState, str]] = None, model_id: Optional[str] = None, session_id: Optional[str] = None, user: Optional[str] = None)[源码]
使用给定选项更新当前活动跟踪。
- 参数
tags – 用于更新跟踪的标签字典。标签专为可变值设计，可以在跟踪创建后通过 MLflow UI 或 API 进行更新。
metadata – 用于更新跟踪的元数据字典。元数据一旦被记录就无法更新。它适合记录不可变的值，如生成跟踪的应用程序版本的 git 哈希值。
client_request_id – 客户端提供的与跟踪关联的请求 ID。这对于将跟踪链接回应用程序或外部系统中的特定请求非常有用。如果为 None，则客户端请求 ID 不会更新。
request_preview – 要在 UI 的跟踪列表视图中显示的请求预览。默认情况下，MLflow 将通过限制长度来简单地截断跟踪请求。此参数允许您指定自定义预览字符串。
response_preview – 要在 UI 的跟踪列表视图中显示的响应预览。默认情况下，MLflow 将通过限制长度来简单地截断跟踪响应。此参数允许您指定自定义预览字符串。
state – 要在跟踪上设置的状态。可以是 TraceState 枚举值或字符串。仅允许“OK”和“ERROR”。这会覆盖整体跟踪状态，而不影响当前跨度（span）的状态。
model_id – 要与跟踪关联的模型 ID。如果未设置，则活动的模型 ID 将与跟踪关联。
session_id – 要与跟踪关联的会话 ID。作为元数据存储在
mlflow.trace.session键下。user – 要与跟踪关联的用户标识符。作为元数据存储在
mlflow.trace.user键下。
示例
您可以在使用
@mlflow.trace装饰的函数内或 with mlflow.start_span 上下文管理器的作用域内使用此函数。如果没有找到活动跟踪，此函数将引发异常。在被 @mlflow.trace 装饰的函数内使用
@mlflow.trace def my_func(x): mlflow.update_current_trace(tags={"fruit": "apple"}, client_request_id="req-12345") return x + 1
在
with mlflow.start_span上下文管理器内使用with mlflow.start_span("span"): mlflow.update_current_trace(tags={"fruit": "apple"}, client_request_id="req-12345")
更新跟踪的用户、会话和来源信息
mlflow.update_current_trace( session_id="session-4f855da00427", user="user-id-cc156f29bcfb", metadata={ "mlflow.source.name": "inference.py", "mlflow.source.git.commit": "1234567890", "mlflow.source.git.repoURL": "https://github.com/mlflow/mlflow", }, )
更新请求预览
import mlflow import openai @mlflow.trace def predict(messages: list[dict]) -> str: # Customize the request preview to show the first and last messages custom_preview = f"{messages[0]['content'][:10]} ... {messages[-1]['content'][:10]}" mlflow.update_current_trace(request_preview=custom_preview) # Call the model response = openai.chat.completions.create( model="o4-mini", messages=messages, ) return response.choices[0].message.content messages = [ {"role": "user", "content": "Hi, how are you?"}, {"role": "assistant", "content": "I'm good, thank you!"}, {"role": "user", "content": "What's your name?"}, # ... (long message history) {"role": "assistant", "content": "Bye!"}, ] predict(messages) # The request preview rendered in the UI will be: # "Hi, how are you? ... Bye!"
- mlflow.update_workspace(name: str, description: Optional[str] = None, default_artifact_root: Optional[str] = None, trace_archival_config: Optional[TraceArchivalConfig] = None) Workspace[源码]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
更新现有工作空间的元数据。
- 参数
name: 要更新的工作空间名称。description: 新描述，或
None表示保持不变。default_artifact_root: 新的构件根 URI，空字符串表示清除，或None。trace_archival_config: 可选的归档设置；使用""清除，None保持不变。- 返回
更新后的工作空间。
- 引发
MlflowException: 如果工作空间不存在或没有可用的构件根路径。
- mlflow.validate_evaluation_results(validation_thresholds: dict[str, mlflow.models.evaluation.validation.MetricThreshold], candidate_result: mlflow.models.evaluation.base.EvaluationResult, baseline_result: Optional[mlflow.models.evaluation.base.EvaluationResult] = None)[源码]
针对另一个模型（基准）验证一个模型（候选）的评估结果。如果候选结果未达到验证阈值，将引发 ModelValidationFailedException。
注意
此 API 是对
mlflow.evaluate()API 中已弃用的模型验证功能的替代。- 参数
validation_thresholds – 用于模型验证的指标名称到
mlflow.models.MetricThreshold的字典。每个指标名称必须是内置指标的名称或extra_metrics参数中定义的指标名称。candidate_result – 候选模型的评估结果。由
mlflow.evaluate()API 返回。baseline_result – 基准模型的评估结果。由
mlflow.evaluate()API 返回。如果设置为 None，候选模型结果将直接与阈值进行比较。
代码示例
import mlflow from mlflow.models import MetricThreshold thresholds = { "accuracy_score": MetricThreshold( # accuracy should be >=0.8 threshold=0.8, # accuracy should be at least 5 percent greater than baseline model accuracy min_absolute_change=0.05, # accuracy should be at least 0.05 greater than baseline model accuracy min_relative_change=0.05, greater_is_better=True, ), } # Get evaluation results for the candidate model candidate_result = mlflow.evaluate( model="<YOUR_CANDIDATE_MODEL_URI>", data=eval_dataset, targets="ground_truth", model_type="classifier", ) # Get evaluation results for the baseline model baseline_result = mlflow.evaluate( model="<YOUR_BASELINE_MODEL_URI>", data=eval_dataset, targets="ground_truth", model_type="classifier", ) # Validate the results mlflow.validate_evaluation_results( thresholds, candidate_result, baseline_result, )
有关更多详细信息，请参阅 模型验证文档。
MLflow 跟踪 API
mlflow 模块提供了一组用于 MLflow 跟踪 的高级 API。有关如何使用这些跟踪 API 的详细指导，请参阅 跟踪 Fluent API 指南。
-
mlflow.trace(func: Callable[[~ _P], mlflow.tracing.fluent._R], name: str | None = None, span_type: str = SpanType.UNKNOWN, attributes: dict[str, typing.Any] | None = None, output_reducer: Optional[Callable[[list[typing.Any]], Any]] = None, trace_destination: mlflow.entities.trace_location.TraceLocationBase | None = None, sampling_ratio_override: float | None = None, log_level: SpanLogLevel | str | None = None) Callable[[~_P], mlflow.tracing.fluent._R][源码]
- mlflow.trace(func: None = None, name: str | None = None, span_type: str = SpanType.UNKNOWN, attributes: dict[str, typing.Any] | None = None, output_reducer: Optional[Callable[[list[typing.Any]], Any]] = None, trace_destination: mlflow.entities.trace_location.TraceLocationBase | None = None, sampling_ratio_override: float | None = None, log_level: SpanLogLevel | str | None = None) Callable[[Callable[[~_P], mlflow.tracing.fluent._R]], Callable[[~_P], mlflow.tracing.fluent._R]]
一个装饰器，为被装饰的函数创建一个新的跨度（span）。
当您使用此
@mlflow.trace()装饰器装饰函数时，将为被装饰函数的范围创建一个跨度。该跨度将自动捕获函数的输入和输出。当应用于方法时，它不会捕获 self 参数。函数内引发的任何异常都将把跨度状态设置为ERROR，并且详细信息（如异常消息和堆栈跟踪）将被记录到跨度的attributes字段中。例如，以下代码将产生一个名称为
"my_function"的跨度，捕获输入参数x和y以及函数输出。import mlflow @mlflow.trace def my_function(x, y): return x + y
这等同于使用
mlflow.start_span()上下文管理器执行的操作，但需要的样板代码更少。import mlflow def my_function(x, y): return x + y with mlflow.start_span("my_function") as span: x = 1 y = 2 span.set_inputs({"x": x, "y": y}) result = my_function(x, y) span.set_outputs({"output": result})
@mlflow.trace 装饰器目前支持以下函数类型
支持的函数类型 函数类型
支持
同步函数
✅
异步
✅ (>= 2.16.0)
生成器 (Generator)
✅ (>= 2.20.2)
异步生成器 (Async Generator)
✅ (>= 2.20.2)
类方法 (ClassMethod)
✅ (>= 3.0.0)
静态方法 (StaticMethod)
✅ (>= 3.0.0)
有关使用 @mlflow.trace 装饰器的更多示例，包括流式处理/异步处理，请参阅 MLflow 跟踪文档。
提示
当您想要跟踪自定义函数时，@mlflow.trace 装饰器非常有用。但是，您可能也想跟踪外部库中的函数。在这种情况下，您可以使用此
mlflow.trace()函数直接包装该函数，而不是将其用作装饰器。这将创建与装饰器创建的完全相同的跨度，即捕获来自函数调用的信息。import math import mlflow mlflow.trace(math.factorial)(5)
- 参数
func – 要装饰的函数。作为装饰器使用时不得提供此项。
name – 跨度的名称。如果未提供，将使用函数的名称。
span_type – 跨度的类型。可以是字符串或
SpanType枚举值。attributes – 要在跨度上设置的属性字典。
output_reducer – 一个将生成器函数的输出减少为单个值以设置为跨度输出的函数。
trace_destination – 记录跟踪的目标，例如 MLflow 实验。如果未提供，目标将是活动的 MLflow 实验或由
mlflow.tracing.set_destination()函数设置的目标。此参数仅应用于根跨度，为非根跨度设置此参数将被忽略并发出警告。sampling_ratio_override – 此特定函数的采样率覆盖值。必须在 0.0 到 1.0 之间。如果提供，它将覆盖全局
MLFLOW_TRACE_SAMPLING_RATIO设置。1.0 表示对所有跟踪进行采样，0.5 表示采样 50%，0.0 表示不进行采样。如果未提供（None），则使用全局采样率。注意：这仅适用于根跨度；如果父级被跟踪，嵌套调用总是会被跟踪。log_level – 要附加到跨度的可选严重性级别。接受
SpanLogLevel或其名称（例如"INFO","DEBUG"）。如果未提供，跨度级别将在结束时从跨度类型解析。
- mlflow.start_span(name: str = 'span', span_type: str | None = 'UNKNOWN', attributes: dict[str, typing.Any] | None = None, trace_destination: mlflow.entities.trace_location.TraceLocationBase | None = None, log_level: SpanLogLevel | str | None = None, run_id: str | None = None) Generator[LiveSpan, None, None][源码]
上下文管理器，用于创建新跨度并将其作为当前上下文中的跨度启动。
此上下文管理器自动管理跨度生命周期和父子关系。上下文管理器退出时，跨度将结束。上下文管理器内引发的任何异常都将把跨度状态设置为
ERROR，并且详细信息（如异常消息和堆栈跟踪）将被记录到跨度的attributes字段中。可以在上下文管理器内创建新跨度，它们将被分配为子跨度。import mlflow with mlflow.start_span("my_span") as span: x = 1 y = 2 span.set_inputs({"x": x, "y": y}) z = x + y span.set_outputs(z) span.set_attribute("key", "value") # do something
当此上下文管理器用于顶层范围（即不在另一个跨度上下文中）时，该跨度将被视为根跨度。根跨度没有父引用，并且整个跟踪将在根跨度结束时记录。
提示
如果您想更明确地控制跟踪生命周期，可以使用 mlflow.start_span_no_context() API。它提供了较低级别的跨度启动方式并明确控制父子关系。但是，通常建议只要满足需求就使用此上下文管理器，因为它需要的样板代码更少且不易出错。
注意
上下文管理器默认不会跨线程传播跨度上下文。有关如何跨线程传播跨度上下文的信息，请参阅 多线程 (Multi Threading)。
- 参数
name – 跨度的名称。
span_type – 跨度的类型。可以是字符串或
SpanType枚举值attributes – 要在跨度上设置的属性字典。
trace_destination – 记录跟踪的目标，例如 MLflow 实验。如果未提供，目标将是活动的 MLflow 实验或由
mlflow.tracing.set_destination()函数设置的目标。此参数仅应用于根跨度，为非根跨度设置此参数将被忽略并发出警告。log_level – 要附加到跨度的可选严重性级别。接受
SpanLogLevel或其名称（例如"INFO","DEBUG"）。如果未提供，跨度级别将在结束时从跨度类型解析。run_id – 要与跟踪关联的 MLflow 运行 ID。此参数仅在创建根跨度时应用。如果提供该参数且没有显式的 trace_destination，则跟踪将记录到运行所属的实验中。如果已通过 mlflow.start_run() 设置了活动的 MLflow 运行，则此参数的优先级高于活动运行。
- 返回
生成一个代表所创建跨度的
mlflow.entities.Span。
- mlflow.start_span_no_context(name: str, span_type: str = 'UNKNOWN', parent_span: Optional[LiveSpan] = None, inputs: Optional[Any] = None, attributes: Optional[dict[str, str]] = None, tags: Optional[dict[str, str]] = None, metadata: Optional[dict[str, str]] = None, experiment_id: Optional[str] = None, start_time_ns: Optional[int] = None, log_level: Optional[Union[SpanLogLevel, str]] = None) LiveSpan[源码]
启动一个不附加到全局跟踪上下文的跨度。
当您想要创建一个不自动链接到父跨度的跨度，并希望手动管理父子关系时，这非常有用。
使用此函数启动的跨度必须使用跨度对象的 end() 方法手动结束。
- 参数
name – 跨度的名称。
span_type – 跨度的类型。可以是字符串或
SpanType枚举值parent_span – 要链接的父跨度。如果为 None，该跨度将被视为根跨度。
inputs – 跨度的输入数据。
attributes – 要在跨度上设置的属性字典。
tags – 要在追踪上设置的标签字典。
metadata – 要在跟踪上设置的元数据字典。
experiment_id – 要与跟踪关联的实验 ID。如果未提供，将使用当前活动实验。
start_time_ns – 跨度的开始时间（纳秒）。如果未提供，将使用当前时间。
log_level – 要附加到跨度的可选严重性级别。接受
SpanLogLevel或其名称（例如"INFO","DEBUG"）。如果未提供，跨度级别将在结束时从跨度类型解析。
- 返回
一个代表所创建跨度的
mlflow.entities.Span。
示例
import mlflow root_span = mlflow.start_span_no_context("my_trace") # Create a child span child_span = mlflow.start_span_no_context( "child_span", # Manually specify the parent span parent_span=root_span, ) # Do something... child_span.end() root_span.end()
- mlflow.get_trace(trace_id: str, silent: bool = False, flush: bool = False) Trace | None[源码]
注意
参数
request_id已弃用。请改用trace_id。如果存在，通过给定的请求 ID 获取跟踪。
此函数首先从内存缓冲区中检索跟踪，如果不存在，则从跟踪存储中获取跟踪。如果跟踪在跟踪存储中未找到，则返回 None。
- 参数
trace_id – 追踪记录（trace）的 ID。
silent – 如果为 True，当找不到跟踪时抑制警告消息。API 将返回 None 而不发出任何警告。默认为 False。
flush – 如果为 True 且未找到跟踪，则刷新任何挂起的异步跟踪写入并重试。对于异步日志记录可能尚未完成的测试或脚本非常有用。默认为 False。
import mlflow with mlflow.start_span(name="span") as span: span.set_attribute("key", "value") trace = mlflow.get_trace(span.trace_id) print(trace)
- 返回
带有给定请求 ID 的
mlflow.entities.Trace对象。
- 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, flush: bool = False) 'pandas.DataFrame' | list[Trace][源码]
注意
参数
experiment_ids已弃用。请改用locations。返回与实验内给定搜索表达式列表匹配的跟踪。
注意
如果预期的搜索结果数量很大，请考虑直接使用 MlflowClient.search_traces API 对结果进行分页。此函数将所有结果返回到内存中，可能不适合大型结果集。
- 参数
experiment_ids – 用于界定搜索范围的实验 ID 列表。
filter_string – 搜索过滤字符串。
max_results – 所需的最大跟踪数。如果为 None，将返回所有与搜索表达式匹配的跟踪。
order_by – order_by 子句列表。
extract_fields –
版本 3.6.0 起弃用: 此参数已弃用，并将在未来版本中删除。
指定要从跟踪中提取的字段，格式为
"span_name.[inputs|outputs].field_name"或"span_name.[inputs|outputs]"。注意
此参数仅在返回类型设置为“pandas”时支持。
例如，
"predict.outputs.result"从名为"predict"的跨度中检索输出"result"字段，而"predict.outputs"获取整个输出字典，包括键"result"和"explanation"。默认情况下，不会有任何字段被提取到 DataFrame 列中。当指定多个字段时，每个字段都作为其自己的列提取。如果提供了无效的字段字符串，函数将静默返回而不添加该字段的列。支持的字段仅限于跨度的
"inputs"和"outputs"。如果跨度名称或字段名称包含点，则必须用反引号括起来。例如：# span name contains a dot extract_fields = ["`span.name`.inputs.field"] # field name contains a dot extract_fields = ["span.inputs.`field.name`"] # span name and field name contain a dot extract_fields = ["`span.name`.inputs.`field.name`"]
run_id – 用于界定搜索范围的运行 ID。当在活动运行下创建跟踪时，它将与该运行关联，您可以按运行 ID 进行过滤以检索跟踪。有关如何按运行 ID 过滤跟踪的示例，请参见下文。
return_type –
返回值的类型。支持以下返回类型。如果安装了 pandas 库，则默认返回类型为“pandas”。否则，默认返回类型为“list”。
- ”pandas”: 返回包含跟踪信息的 Pandas DataFrame
其中每一行代表一个单一的跟踪，每一列代表跟踪的一个字段，例如 trace_id, spans 等。
”list”: 返回
Trace对象列表。
model_id – 如果指定，则搜索与给定模型 ID 关联的追踪。
sql_warehouse_id – 已弃用。请改用 MLFLOW_TRACING_SQL_WAREHOUSE_ID 环境变量。用于在推理表或 UC 表中搜索跟踪的 SQL 仓库 ID。仅在 Databricks 中使用。
include_spans – 如果为
True，则在返回的跟踪中包含跨度。否则，仅返回跟踪元数据，例如 trace ID、开始时间、结束时间等，而不包含任何跨度。默认为True。locations – 要搜索的位置列表。要在实验中搜索，请提供实验 ID 列表。要在 Databricks 上的 UC 表中搜索，请以 <catalog_name>.<schema_name>[.<table_prefix>] 格式提供位置列表。如果未提供，搜索将在当前活动的实验中执行。
flush – 如果为
True，在搜索前刷新任何挂起的异步跟踪写入。对于测试或脚本非常有用，以确保所有跟踪均可见。默认为False。
- 返回
满足搜索表达式的跟踪。根据 return_type 参数的值，作为
Trace对象列表或 Pandas DataFrame 返回。
import mlflow with mlflow.start_span(name="span1") as span: span.set_inputs({"a": 1, "b": 2}) span.set_outputs({"c": 3, "d": 4}) mlflow.search_traces( extract_fields=["span1.inputs", "span1.outputs", "span1.outputs.c"], return_type="pandas", )
import mlflow with mlflow.start_span(name="non_dict_span") as span: span.set_inputs(["a", "b"]) span.set_outputs([1, 2, 3]) mlflow.search_traces( extract_fields=["non_dict_span.inputs", "non_dict_span.outputs"], )
- mlflow.get_current_active_span() LiveSpan | None[源码]
获取全局上下文中的当前活动跨度。
注意
这仅在跨度使用 Fluent API（如 @mlflow.trace 或 with mlflow.start_span）创建时有效。如果使用 mlflow.start_span_no_context API 创建跨度，它将不会附加到全局上下文，因此此函数不会返回它。
import mlflow @mlflow.trace def f(): span = mlflow.get_current_active_span() span.set_attribute("key", "value") return 0 f()
- 返回
当前存在的活动跨度，否则为 None。
- mlflow.get_last_active_trace_id(thread_local: bool = False) str | None[源码]
获取同一进程中存在的最后一个活动跟踪。
警告
此函数默认不是线程安全的，返回同一进程中的最后一个活动跟踪。如果您想获取当前线程中的最后一个活动跟踪，请将 thread_local 参数设置为 True。
- 参数
thread_local – 如果为 True，则返回当前线程中的最后一个活动跟踪。否则，返回同一进程中的最后一个活动跟踪。默认为 False。
- 返回
最后一个活动跟踪的 ID（如果存在），否则为 None。
import mlflow @mlflow.trace def f(): pass f() trace_id = mlflow.get_last_active_trace_id() # Set a tag on the trace mlflow.set_trace_tag(trace_id, "key", "value") # Get the full trace object trace = mlflow.get_trace(trace_id)
- mlflow.add_trace(trace: Trace | dict[str, typing.Any], target: Optional[LiveSpan] = None)[源码]
将完成的跟踪对象添加到另一个跟踪中。
当您调用由 MLflow Tracing 装饰的远程服务时，这特别有用。通过使用此函数，您可以将来自远程服务的跟踪合并到当前的活动本地跟踪中，从而可以看到包括远程服务调用内部发生情况在内的完整跟踪。
以下示例演示了如何使用此函数将远程服务的跟踪合并到函数中的当前活动跟踪中。
@mlflow.trace(name="predict") def predict(input): # Call a remote service that returns a trace in the response resp = requests.get("https://your-service-endpoint", ...) # Extract the trace from the response trace_json = resp.json().get("trace") # Use the remote trace as a part of the current active trace. # It will be merged under the span "predict" and exported together when it is ended. mlflow.add_trace(trace_json)
如果您有特定的目标跨度来合并跟踪，可以传递目标跨度
def predict(input): # Create a local span with mlflow.start_span(name="predict") as span: resp = requests.get("https://your-service-endpoint", ...) trace_json = resp.json().get("trace") # Merge the remote trace under the span created above mlflow.add_trace(trace_json, target=span)
- 参数
trace –
一个
Trace对象或跟踪的字典表示。跟踪必须已经完成，即不应对其进行进一步更新。否则，此函数将引发异常。target –
合并给定跟踪的目标跨度。
如果提供，跟踪将合并在目标跨度下。
如果未提供，跟踪将合并在当前活动跨度下。
如果未提供且没有活动跨度，将创建一个名为“Remote Trace <…>”的新跨度，并将跟踪合并在其下。
- mlflow.log_assessment(trace_id: str, assessment: Assessment) Assessment[源码]
将评估记录到跟踪。评估可以是预期（expectation）或反馈（feedback）。
- 预期：表示特定操作预期值的标签。
例如，聊天机器人对用户问题的预期回答。
- 反馈：表示对操作质量的反馈的标签。
反馈可以来自不同的来源，例如人类评判者、启发式评分器或 LLM-as-a-Judge。
以下代码使用 LLM-as-a-Judge 提供的反馈标注跟踪。
import mlflow from mlflow.entities import Feedback feedback = Feedback( name="faithfulness", value=0.9, rationale="The model is faithful to the input.", metadata={"model": "gpt-4o-mini"}, ) mlflow.log_assessment(trace_id="1234", assessment=feedback)
以下代码使用带有来源信息的人类提供的基本事实（ground truth）标注跟踪。当未提供来源时，默认来源设置为“default”，类型为“HUMAN”。
import mlflow from mlflow.entities import AssessmentSource, AssessmentSourceType, Expectation # Specify the annotator information as a source. source = AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="john@example.com", ) expectation = Expectation( name="expected_answer", value=42, source=source, ) mlflow.log_assessment(trace_id="1234", assessment=expectation)
- 预期值可以是任何可序列化为 JSON 的值。例如，您可以
记录完整 LLM 消息作为预期值。
import mlflow from mlflow.entities.assessment import Expectation expectation = Expectation( name="expected_message", # Full LLM message including expected tool calls value={ "role": "assistant", "content": "The answer is 42.", "tool_calls": [ { "id": "1234", "type": "function", "function": {"name": "add", "arguments": "40 + 2"}, } ], }, ) mlflow.log_assessment(trace_id="1234", assessment=expectation)
您还可以在反馈生成过程中记录错误信息。为此，请向 error 参数提供
AssessmentError的实例，并将 value 参数保留为 None。import mlflow from mlflow.entities import AssessmentError, Feedback error = AssessmentError( error_code="RATE_LIMIT_EXCEEDED", error_message="Rate limit for the judge exceeded.", ) feedback = Feedback( trace_id="1234", name="faithfulness", error=error, ) mlflow.log_assessment(trace_id="1234", assessment=feedback)
- mlflow.log_expectation(*, trace_id: str, name: str, value: Any, source: Optional[AssessmentSource] = None, metadata: Optional[dict[str, typing.Any]] = None, span_id: Optional[str] = None) Assessment[源码]
将预期（例如基本事实标签）记录到跟踪。此 API 仅接受关键字参数。
- 参数
trace_id – 追踪记录（trace）的 ID。
name – 预期评估的名称，例如“expected_answer”
value – 预期的值。它可以是任何可序列化为 JSON 的值。
source – 预期评估的来源。必须是
AssessmentSource的实例。如果未提供，默认为 HUMAN 来源类型。metadata – 预期的额外元数据。
span_id – 与预期关联的跨度 ID（如果需要与跟踪中的特定跨度关联）。
- 返回
创建的预期评估。
- 返回类型
示例
import mlflow from mlflow.entities import AssessmentSource, AssessmentSourceType # Log simple expected answer expectation = mlflow.log_expectation( trace_id="tr-1234567890abcdef", name="expected_answer", value="The capital of France is Paris.", source=AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="annotator@company.com" ), metadata={"question_type": "factual", "difficulty": "easy"}, ) # Log expected classification label mlflow.log_expectation( trace_id="tr-1234567890abcdef", name="expected_category", value="positive", source=AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="data_labeler_001" ), metadata={"labeling_session": "batch_01", "confidence": 0.95}, )
- mlflow.log_feedback(*, trace_id: str, name: str = 'feedback', value: Optional[Union[float, int, str, bool, dict[str, float | int | str | bool], list[float | int | str | bool]]] = None, source: Optional[AssessmentSource] = None, error: Optional[Union[Exception, AssessmentError]] = None, rationale: Optional[str] = None, metadata: Optional[dict[str, typing.Any]] = None, span_id: Optional[str] = None) Assessment[source]
记录反馈到 Trace。此 API 仅接受关键字参数。
- 参数
trace_id – 追踪记录（trace）的 ID。
name – 反馈评估的名称，例如 “faithfulness”（忠实度）。如果未提供，默认为 “feedback”。
value – 反馈的值。可以是
None，或以下类型之一：- float (浮点数) - int (整数) - str (字符串) - bool (布尔值) - 上述类型的值组成的列表 - 以字符串为键、值为上述类型的字典source – 反馈评估的来源。必须是
AssessmentSource的实例。如果未提供，默认为 CODE（代码）来源类型。error – 表示计算反馈时遇到的任何问题的错误对象，例如来自 LLM 评估器的超时错误。接受异常对象或
AssessmentError对象。必须提供此参数或 value。rationale – 反馈的理由/依据。
metadata – 反馈的附加元数据。
span_id – 与反馈关联的 span ID，如果需要将其关联到 trace 中的特定 span。
- 返回
创建的反馈评估。
- 返回类型
示例
import mlflow from mlflow.entities import AssessmentSource, AssessmentSourceType # Log simple feedback score feedback = mlflow.log_feedback( trace_id="tr-1234567890abcdef", name="relevance", value=0.9, source=AssessmentSource( source_type=AssessmentSourceType.LLM_JUDGE, source_id="gpt-4" ), rationale="Response directly addresses the user's question", ) # Log detailed feedback with structured data mlflow.log_feedback( trace_id="tr-1234567890abcdef", name="quality_metrics", value={"accuracy": 0.95, "completeness": 0.88, "clarity": 0.92, "overall": 0.92}, source=AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="expert_evaluator" ), rationale="High accuracy and clarity, slightly incomplete coverage", )
- mlflow.update_assessment(trace_id: str, assessment_id: str, assessment: Assessment) Assessment[source]
更新 Trace 中现有的预期（ground truth）。
- 参数
trace_id – 追踪记录（trace）的 ID。
assessment_id – 要更新的预期或反馈评估的 ID。
assessment – 更新后的评估。
- 返回
更新后的反馈或预期评估。
- 返回类型
示例
以下代码用新值更新现有的预期。要更新其他字段，请提供相应的参数。
import mlflow from mlflow.entities import Expectation, ExpectationValue # Create an expectation with value 42. response = mlflow.log_assessment( trace_id="1234", assessment=Expectation(name="expected_answer", value=42), ) assessment_id = response.assessment_id # Update the expectation with a new value 43. mlflow.update_assessment( trace_id="1234", assessment_id=assessment.assessment_id, assessment=Expectation(name="expected_answer", value=43), )
- mlflow.delete_assessment(trace_id: str, assessment_id: str)[source]
删除与 trace 关联的评估。
- 参数
trace_id – 追踪记录（trace）的 ID。
assessment_id – 要删除的评估 ID。
- mlflow.tracing.configure(span_processors: list[typing.Callable[[ForwardRef('LiveSpan')], NoneType]] | None = None) mlflow.tracing.config.TracingConfigContext[source]
配置 MLflow 追踪。可用作函数或上下文管理器。
仅更新显式提供的参数，保持其他参数不变。
- 参数
span_processors – 在导出前处理 span 的函数列表。这有助于过滤/屏蔽 span 中的特定属性，以防止记录敏感数据或减小 span 大小。每个函数必须接受一个 LiveSpan 类型的参数且不应返回任何值。提供多个函数时，它们将按提供顺序依次应用。
- 返回
- 用于临时配置更改的上下文管理器。
作为函数使用时，配置更改会持久生效。作为上下文管理器使用时，退出时会还原更改。
- 返回类型
TracingConfigContext
示例
def pii_filter(span): """Example PII filter that masks sensitive data in span attributes.""" # Mask sensitive inputs if inputs := span.inputs: for key, value in inputs.items(): if "password" in key.lower() or "token" in key.lower(): span.set_inputs({**inputs, key: "[REDACTED]"}) # Mask sensitive outputs if outputs := span.outputs: if isinstance(outputs, dict): for key in outputs: if "secret" in key.lower(): outputs[key] = "[REDACTED]" span.set_outputs(outputs) # Mask sensitive attributes for attr_key in list(span.attributes.keys()): if "api_key" in attr_key.lower(): span.set_attribute(attr_key, "[REDACTED]") # Permanent configuration change mlflow.tracing.configure(span_processors=[pii_filter]) # Temporary configuration change with mlflow.tracing.configure(span_processors=[pii_filter]): # PII filtering enabled only in this block pass
- mlflow.tracing.context(metadata: dict[str, str] | None = None, tags: dict[str, str] | None = None, enabled: bool | None = None, session_id: str | None = None, user: str | None = None) Generator[None, None, None][source]
一种上下文管理器，可将元数据和/或标签注入到其作用域内创建的任何跟踪中，而无需创建包装跨度（span）。它还可以用于选择性地禁用该作用域内的跟踪。
当您需要将跟踪级信息（例如会话 ID）附加到非您控制的代码（如自动插桩库）生成的跟踪，或者当您想在不影响全局跟踪状态的情况下禁止特定代码块的跟踪时，此功能非常有用。
import mlflow # Enable auto-tracing for LangChain mlflow.langchain.autolog() with mlflow.tracing.context( session_id="session-123", user="user-456", tags={"project": "my-project"}, ): # Any trace created inside this block will carry the metadata and tags. agent.invoke("What is the capital of France?") # Disable tracing within a specific block with mlflow.tracing.context(enabled=False): # No traces will be created inside this block. agent.invoke("This call will not be traced")
该上下文管理器可以嵌套以组合多组元数据和标签。当在多个级别指定相同的键时，内部级别的值优先。
import mlflow with mlflow.tracing.context(metadata={"foo": "bar", "baz": "qux"}): with mlflow.tracing.context(metadata={"foo": "baz", "qux": "quux"}): my_func() # Trace created by my_func will have metadata={"foo": "baz", "baz": "qux", "qux": "quux"}
- 参数
metadata – 要注入到跟踪的
request_metadata中的键值对（在跟踪创建后不可变）。tags – 要注入到跟踪的
tags中的键值对。enabled – 该作用域内是否启用了跟踪。如果为
False，则作用域内的所有跟踪调用将返回NoOpSpan而不创建任何跟踪。如果为None（默认值），则值将继承自外部作用域。这不会影响通过mlflow.tracing.disable()设置的全局跟踪状态。session_id – 与此作用域内创建的跟踪相关联的会话 ID。内部存储为
mlflow.trace.session键下的元数据。user – 与此作用域内创建的跟踪相关联的用户标识符。内部存储为
mlflow.trace.user键下的元数据。
- mlflow.tracing.disable()[source]
禁用追踪。
注意
此函数设置 OpenTelemetry 使用 NoOpTracerProvider，并有效禁用所有追踪操作。
示例
import mlflow @mlflow.trace def f(): return 0 # Tracing is enabled by default f() assert len(mlflow.search_traces(flush=True)) == 1 # Disable tracing mlflow.tracing.disable() f() assert len(mlflow.search_traces(flush=True)) == 1
- mlflow.tracing.disable_notebook_display()[source]
禁止在 notebook 输出单元格中显示 MLflow Trace UI。调用
mlflow.tracing.enable_notebook_display()可重新启用显示。
- mlflow.tracing.enable()[source]
启用追踪。
示例
import mlflow @mlflow.trace def f(): return 0 # Tracing is enabled by default f() assert len(mlflow.search_traces(flush=True)) == 1 # Disable tracing mlflow.tracing.disable() f() assert len(mlflow.search_traces(flush=True)) == 1 # Re-enable tracing mlflow.tracing.enable() f() assert len(mlflow.search_traces(flush=True)) == 2
- mlflow.tracing.enable_notebook_display()[source]
启用 notebook 输出单元格中的 MLflow Trace UI。显示功能默认开启，Trace UI 将在执行以下任何操作时显示：
追踪完成时（即每当导出 trace 时）
调用
mlflow.search_traces()流畅 API 时调用
mlflow.client.MlflowClient.get_trace()或mlflow.client.MlflowClient.search_traces()客户端 API 时
- mlflow.tracing.get_tracing_context_headers_for_http_request() dict[str, str][source]
获取包含追踪上下文信息的 HTTP 请求头。追踪上下文序列化为 W3C TraceContext 规范中定义的 traceparent 头。有关详细信息，请参阅 https://opentelemetry.org.cn/docs/concepts/context-propagation/ 和 https://w3org.cn/TR/trace-context/#traceparent-header
- 返回
包含追踪上下文信息的 HTTP 请求头。
示例（客户端代码）
import mlflow from mlflow.tracing import get_tracing_context_headers_for_http_request with mlflow.start_span("client-root") as client_span: # Get the headers that hold information of the tracing context, # and send request to remote agent with the headers headers = get_tracing_context_headers_for_http_request() resp = requests.post(f"{base_url}/remote_agent_handler", headers=headers)
示例（服务器处理程序代码）
import mlflow from flask import Flask, request from mlflow.tracing import set_tracing_context_from_http_request_headers app = Flask(__name__) @app.post("/agent-handler") def handle(): headers = dict(request.headers) with set_tracing_context_from_http_request_headers(headers): with mlflow.start_span("server-handler") as span: # call agent ... span.set_attribute("key", "value")
- mlflow.tracing.reset()[source]
重置指示 MLflow 追踪提供程序是否已初始化的标志。这确保在执行下一次追踪操作时重新初始化追踪提供程序。
- mlflow.tracing.set_databricks_monitoring_sql_warehouse_id(sql_warehouse_id: str, experiment_id: Optional[str] = None) None[source]
为记录到给定 MLflow 实验的 trace 设置用于 Databricks 生产监控的 SQL 仓库 ID。这仅对以 UC schema 作为 trace 位置的实验有效。
- 参数
sql_warehouse_id – 用于监控的 SQL 仓库 ID。
experiment_id – MLflow 实验 ID。如果未提供，将使用当前活跃的实验。
- mlflow.tracing.set_destination(destination: mlflow.entities.trace_location.TraceLocationBase, *, context_local: bool = False)[source]
设置 MLflow 将导出 trace 的自定义 span 位置。
此函数指定的目的地将优先于其他配置，如追踪 URI、OTLP 环境变量。
- 参数
destination –
指定 trace 数据位置的 trace 位置对象。目前支持以下位置：
MlflowExperimentLocation：将 trace 记录到MLflow 实验。
context_local – 如果为 False（默认），则全局设置目的地。如果为 True，则目的地在每个异步任务或线程中隔离，从而在并发应用中提供隔离。
示例
将 trace 记录到 MLflow 实验
from mlflow.entities.trace_location import MlflowExperimentLocation mlflow.tracing.set_destination(MlflowExperimentLocation(experiment_id="123"))
注意：这与通过
MLFLOW_EXPERIMENT_ID环境变量或mlflow.set_experimentAPI 设置活跃 MLflow 实验的效果相同，但作用域更窄。在异步任务或线程之间隔离目的地
from mlflow.tracing.destination import Databricks mlflow.tracing.set_destination( MlflowExperimentLocation(experiment_id="123"), context_local=True, )
使用
context_local标志设置的目的地仅在当前异步任务或线程中有效。当您希望从多租户应用程序将 trace 发送到不同目的地时，这特别有用。** 重置目的地：**
mlflow.tracing.reset()
- mlflow.tracing.set_experiment_trace_location(location: UCSchemaLocation, experiment_id: Optional[str] = None, sql_warehouse_id: Optional[str] = None) UCSchemaLocation[source]
配置实验 trace 的存储位置。
用于存储 trace 数据的 Unity Catalog 表将在指定的 schema 中创建。当启用追踪时，指定实验的所有 trace 都将存储在提供的 Unity Catalog schema 中。
注意
如果实验已链接到存储位置，这将引发错误。请使用 mlflow.tracing.unset_experiment_trace_location 先删除现有的存储位置，然后再设置新的。
- 参数
location – Unity Catalog 中实验 trace 的存储位置。
experiment_id – 要设置存储位置的 MLflow 实验 ID。如果未指定，将使用当前活跃的实验。
sql_warehouse_id – 用于创建视图和查询的 SQL 仓库 ID。如果未指定，则使用 MLFLOW_TRACING_SQL_WAREHOUSE_ID 中的值，如果未设置环境变量，则回退到默认 SQL 仓库。
- 返回
表示所配置存储位置的 UCSchemaLocation 对象，包括 spans 和 logs 表的表名。
示例
import mlflow from mlflow.entities import UCSchemaLocation location = UCSchemaLocation(catalog_name="my_catalog", schema_name="my_schema") result = mlflow.tracing.set_experiment_trace_location( location=location, experiment_id="12345", ) print(result.full_otel_spans_table_name) # my_catalog.my_schema.otel_spans_table @mlflow.trace def add(x): return x + 1 add(1) # this writes the trace to the storage location set above
- mlflow.tracing.set_span_chat_tools(span: LiveSpan, tools: list[ChatTool])[source]
在指定 span 上设置 mlflow.chat.tools 属性。此属性在 UI 中使用，也被消费 trace 数据的下游应用程序（如 MLflow evaluate）使用。
- 参数
span – 要添加该属性的 LiveSpan
tools – 标准化聊天工具定义的列表（详细信息请参阅 规范）
示例
import mlflow from mlflow.tracing import set_span_chat_tools tools = [ { "type": "function", "function": { "name": "add", "description": "Add two numbers", "parameters": { "type": "object", "properties": { "a": {"type": "number"}, "b": {"type": "number"}, }, "required": ["a", "b"], }, }, } ] @mlflow.trace def f(): span = mlflow.get_current_active_span() set_span_chat_tools(span, tools) return 0 f()
- mlflow.tracing.set_tracing_context_from_http_request_headers(headers: dict[str, str])[source]
上下文管理器，用于从 HTTP 请求头中提取追踪上下文，并将提取的追踪上下文设置为该上下文管理器作用域内的当前追踪上下文。追踪上下文必须序列化为 W3C TraceContext 规范中定义的 “traceparent” 头，请参阅
mlflow.tracing.get_tracing_context_headers_for_http_request()了解如何获取 HTTP 请求头。- 参数
headers – 用于提取追踪上下文的 HTTP 请求头。
示例（客户端代码）
import mlflow from mlflow.tracing import get_tracing_context_headers_for_http_request with mlflow.start_span("client-root") as client_span: # Get the headers that hold information of the tracing context, # and send request to remote agent with the headers headers = get_tracing_context_headers_for_http_request() resp = requests.post(f"{base_url}/remote_agent_handler", headers=headers)
示例（服务器处理程序代码）
import mlflow from flask import Flask, request from mlflow.tracing import set_tracing_context_from_http_request_headers app = Flask(__name__) @app.post("/agent-handler") def handle(): headers = dict(request.headers) with set_tracing_context_from_http_request_headers(headers): with mlflow.start_span("server-handler") as span: # call agent ... span.set_attribute("key", "value")
- mlflow.tracing.unset_experiment_trace_location(location: UCSchemaLocation, experiment_id: Optional[str] = None) None[source]
取消设置实验 trace 的存储位置。
此函数删除实验存储位置配置，包括视图和实验标签。
- 参数
location – 要取消设置的存储位置。
experiment_id – 要取消设置存储位置的 MLflow 实验 ID。如果未提供，将使用当前活跃的实验。
示例
import mlflow from mlflow.entities import UCSchemaLocation mlflow.tracing.unset_experiment_trace_location( location=UCSchemaLocation(catalog_name="my_catalog", schema_name="my_schema"), experiment_id="12345", )
MLflow 已记录模型 (Logged Model) API
mlflow 模块提供了一组高级 API，用于与 MLflow 已记录模型 进行交互。
- mlflow.clear_active_model() None[source]
清除活跃模型。这将清除之前由
mlflow.set_active_model()、MLFLOW_ACTIVE_MODEL_ID环境变量或遗留的_MLFLOW_ACTIVE_MODEL_ID环境变量设置的活跃模型。从当前线程。若要临时切换活跃模型，请改用
with mlflow.set_active_model(...)。import mlflow # Set the active model by name mlflow.set_active_model(name="my_model") # Clear the active model mlflow.clear_active_model() # Check that the active model is None assert mlflow.get_active_model_id() is None # If you want to temporarily set the active model, # use `set_active_model` as a context manager instead with mlflow.set_active_model(name="my_model") as active_model: assert mlflow.get_active_model_id() == active_model.model_id assert mlflow.get_active_model_id() is None
- mlflow.create_external_model(name: Optional[str] = None, source_run_id: Optional[str] = None, tags: Optional[dict[str, str]] = None, params: Optional[dict[str, str]] = None, model_type: Optional[str] = None, experiment_id: Optional[str] = None) LoggedModel[source]
创建一个新的 LoggedModel，其构件存储在 MLflow 外部。这对于跟踪未采用 MLflow 模型格式打包的模型、应用程序或生成式 AI 代理的参数和性能数据（指标、trace 等）非常有用。
- 参数
name – 模型名称。如果未指定，将随机生成一个名称。
source_run_id – 与模型关联的运行 ID。如果未指定且存在活跃运行，则将使用该活跃运行 ID。
tags – 设置为模型标签的字符串键值对字典。
params – 设置为模型参数的字符串键值对字典。
model_type – 模型类型。这是一个用户定义的字符串，可用于搜索和比较相关模型。例如，设置
model_type="agent"可让您将来轻松搜索此模型并将其与“agent”类型的其他模型进行比较。experiment_id – 模型所属实验的实验 ID。
- 返回
状态为
READY的新mlflow.entities.LoggedModel对象。
- mlflow.delete_logged_model_tag(model_id: str, key: str) None[source]
删除指定已记录模型中的标签。
- 参数
model_id – 模型 ID。
key – 要删除的标签键。
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) mlflow.set_logged_model_tags(model_info.model_id, {"key": "value"}) model = mlflow.get_logged_model(model_info.model_id) assert model.tags["key"] == "value" mlflow.delete_logged_model_tag(model_info.model_id, "key") model = mlflow.get_logged_model(model_info.model_id) assert "key" not in model.tags
- mlflow.finalize_logged_model(model_id: str, status: Union[Literal['READY', 'FAILED'], LoggedModelStatus]) LoggedModel[source]
通过更新状态来最终确定模型。
- 参数
model_id – 要最终确定的模型 ID。
status – 要在模型上设置的最终状态。
- 返回
更新后的模型。
示例
import mlflow from mlflow.entities import LoggedModelStatus model = mlflow.initialize_logged_model(name="model") logged_model = mlflow.finalize_logged_model( model_id=model.model_id, status=LoggedModelStatus.READY, ) assert logged_model.status == LoggedModelStatus.READY
- mlflow.get_logged_model(model_id: str) LoggedModel[source]
通过 ID 获取已记录的模型。
- 参数
model_id – 已记录模型的 ID。
- 返回
已记录的模型。
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) logged_model = mlflow.get_logged_model(model_id=model_info.model_id) assert logged_model.model_id == model_info.model_id
- mlflow.initialize_logged_model(name: Optional[str] = None, source_run_id: Optional[str] = None, tags: Optional[dict[str, str]] = None, params: Optional[dict[str, str]] = None, model_type: Optional[str] = None, experiment_id: Optional[str] = None) LoggedModel[source]
初始化 LoggedModel。创建状态为
PENDING且无构件的 LoggedModel。您必须向模型添加构件并将其最终确定为READY状态，例如通过调用特定于风味的log_model()方法，如mlflow.pyfunc.log_model()。- 参数
name – 模型名称。如果未指定，将随机生成一个名称。
source_run_id – 与模型关联的运行 ID。如果未指定且存在活跃运行，则将使用该活跃运行 ID。
tags – 设置为模型标签的字符串键值对字典。
params – 设置为模型参数的字符串键值对字典。
model_type – 模型的类型。
experiment_id – 模型所属实验的实验 ID。
- 返回
状态为
PENDING的新mlflow.entities.LoggedModel对象。
- mlflow.last_logged_model() LoggedModel | None[source]
获取当前会话中最近记录的模型。如果尚未记录任何模型，则返回 None。
- 返回
已记录的模型。
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) last_model = mlflow.last_logged_model() assert last_model.model_id == model_info.model_id
-
mlflow.search_logged_models(experiment_ids: list[str] | None = None, filter_string: str | None = None, datasets: list[dict[str, str]] | None = None, max_results: int | None = None, order_by: list[dict[str, typing.Any]] | None = None, output_format: Literal['pandas'] = 'pandas') pandas.DataFrame[source]
- mlflow.search_logged_models(experiment_ids: list[str] | None = None, filter_string: str | None = None, datasets: list[dict[str, str]] | None = None, max_results: int | None = None, order_by: list[dict[str, typing.Any]] | None = None, output_format: Literal['list'] = 'list') list[LoggedModel]
搜索符合指定搜索条件的已记录模型。
- 参数
experiment_ids – 用于搜索已记录模型的实验 ID 列表。如果未指定，将使用当前活跃实验。
filter_string –
要解析的类 SQL 过滤字符串。过滤字符串语法支持：
- 实体规范
属性：attribute_name（如果未指定前缀，则为默认值）
指标：metrics.metric_name
参数：params.param_name
标签：tags.tag_name
- 比较运算符
对于数值实体（指标和数值属性）：<、<=、>、>=、=、!=
对于字符串实体（参数、标签、字符串属性）：=、!=、IN、NOT IN
多个条件可以使用 ‘AND’ 连接
字符串值必须包含在单引号中
- 示例过滤字符串
creation_time > 100
metrics.rmse > 0.5 AND params.model_type = ‘rf’
tags.release IN (‘v1.0’, ‘v1.1’)
params.optimizer != ‘adam’ AND metrics.accuracy >= 0.9
datasets –
用于指定应用指标过滤的数据集字典列表。例如，具有 metrics.accuracy > 0.9 和名为 “test_dataset” 的数据集的过滤字符串表示我们将返回在 test_dataset 上准确度 > 0.9 的所有已记录模型。考虑匹配标准的 ANY 数据集中的指标值。如果未指定数据集，则过滤中将考虑所有数据集的指标。支持以下字段：
- dataset_name (str)
必需。数据集名称。
- dataset_digest (str)
可选。数据集摘要。
max_results – 要返回的已记录模型的最大数量。
order_by –
用于指定搜索结果排序的字典列表。支持以下字段：
- field_name (str)
必需。用于排序的字段名称，例如 “metrics.accuracy”。
- ascending (bool)
可选。排序是否为升序。
- dataset_name (str)
可选。如果
field_name指向指标，则此字段指定与指标关联的数据集名称。排序时仅考虑与指定数据集名称关联的指标。仅当field_name指向指标时，才能设置此字段。- dataset_digest (str)
可选。如果
field_name指向指标，则此字段指定与指标关联的数据集摘要。排序时仅考虑与指定数据集名称和摘要关联的指标。仅当同时设置了dataset_name时，才能设置此字段。
output_format – 搜索结果的输出格式。支持的值为 ‘pandas’ 和 ‘list’。
- 返回
以指定输出格式返回的搜索结果。
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) another_model_info = mlflow.pyfunc.log_model( name="another_model", python_model=DummyModel() ) models = mlflow.search_logged_models(output_format="list") assert [m.name for m in models] == ["another_model", "model"] models = mlflow.search_logged_models( filter_string="name = 'another_model'", output_format="list" ) assert [m.name for m in models] == ["another_model"] models = mlflow.search_logged_models( order_by=[{"field_name": "creation_time", "ascending": True}], output_format="list" ) assert [m.name for m in models] == ["model", "another_model"]
- mlflow.set_active_model(*, name: Optional[str] = None, model_id: Optional[str] = None) ActiveModel[source]
设置具有指定名称或模型 ID 的活跃模型，它将用于链接在模型生命周期内生成的 trace。返回值可用作
with块中的上下文管理器；否则，您必须调用set_active_model()来更新活跃模型。- 参数
name – 要设置为活跃的
mlflow.entities.LoggedModel的名称。如果具有该名称的 LoggedModel 不存在，它将在当前实验下创建。如果存在多个具有该名称的 LoggedModel，则最新的一个将被设置为活跃。model_id – 要设置为活跃的
mlflow.entities.LoggedModel的 ID。如果不存在具有该 ID 的 LoggedModel，将引发异常。
- 返回
mlflow.ActiveModel对象，该对象充当包装 LoggedModel 状态的上下文管理器。
import mlflow # Set the active model by name mlflow.set_active_model(name="my_model") # Set the active model by model ID model = mlflow.create_external_model(name="test_model") mlflow.set_active_model(model_id=model.model_id) # Use the active model in a context manager with mlflow.set_active_model(name="new_model"): print(mlflow.get_active_model_id()) # Traces are automatically linked to the active model mlflow.set_active_model(name="my_model") @mlflow.trace def predict(model_input): return model_input predict("abc") traces = mlflow.search_traces( model_id=mlflow.get_active_model_id(), return_type="list", flush=True ) assert len(traces) == 1
- mlflow.set_logged_model_tags(model_id: str, tags: dict[str, typing.Any]) None[source]
在指定的已记录模型上设置标签。
- 参数
model_id – 模型 ID。
tags – 要在模型上设置的标签。
- 返回
无
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) mlflow.set_logged_model_tags(model_info.model_id, {"key": "value"}) model = mlflow.get_logged_model(model_info.model_id) assert model.tags["key"] == "value"
- mlflow.log_model_params(params: dict[str, str], model_id: Optional[str] = None) None[source]
将参数记录到指定的已记录模型。
- 参数
params – 要记录在模型上的参数。
model_id – 模型 ID。如果未指定，则使用当前活跃模型 ID。
- 返回
无
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) mlflow.log_model_params(params={"param": "value"}, model_id=model_info.model_id)