mlflow
该 mlflow 模块提供了一个高级的“流式”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 块结束时自动终止运行。
fluent tracking API 目前不是线程安全的。任何并发调用 tracking API 的调用者必须手动实现互斥。
对于更低级别的 API，请参见 mlflow.client 模块。
- class mlflow.ActiveModel(logged_model: LoggedModel, set_by_user: bool)[source]
围绕
mlflow.entities.LoggedModel的包装器，使其可以使用 Pythonwith语法。
- class mlflow.ActiveRun(run)[source]
用于封装
mlflow.entities.Run，以启用使用 Pythonwith语法。
- class mlflow.Image(image: Union[numpy.ndarray, PIL.Image.Image, str, list[typing.Any]])[source]
mlflow.Image 是一个用于图像媒体对象的轻量级选项，用于在 MLflow 中处理图像。图像可以是 numpy 数组、PIL 图像或图像的文件路径。图像以 PIL 图像的形式存储，并且可以使用 mlflow.log_image 或 mlflow.log_table 将其记录到 MLflow。
- Parameters
image – Image 可以是 numpy array、PIL image，或指向图像的文件路径。
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])[source]
将图像调整为指定大小。
- Parameters
size – 要将图像调整到的大小。
- Returns
调整大小后的图像对象的副本。
- save(path: str)[source]
将图像保存到文件。
- Parameters
path – 保存图像的文件路径。
- to_array()[source]
将图像转换为 numpy 数组。
- Returns
Numpy 的像素值数组。
- to_list()[source]
将图像转换为像素值列表。
- Returns
像素值列表。
- to_pil()[source]
将图像转换为 PIL 图像。
- Returns
PIL 图像。
- exception mlflow.MlflowException(message, error_code=1, **kwargs)[source]
通用异常，用于揭示面向外部操作的失败信息。 与该异常相关的错误消息可能会在 HTTP 响应中暴露给客户端以用于调试。 如果错误文本是敏感的，请改为抛出一个通用的 Exception 对象。
- get_http_status_code()[source]
- classmethod invalid_parameter_value(message, **kwargs)[source]
构造一个 MlflowException 对象，并使用 INVALID_PARAMETER_VALUE 错误代码。
- Parameters
message – 描述所发生错误的消息。该消息将包含在异常的序列化 JSON 表示中。
kwargs – 要包含在 MlflowException 的序列化 JSON 表示中的附加键值对。
- serialize_as_json()[source]
- mlflow.active_run() ActiveRun | None[source]
获取当前活动的
Run，如果不存在则返回 None。注意
此 API 为 thread-local，仅返回当前线程中活动的 run。如果您的应用是多线程的，并且在不同线程中启动了 run，则此 API 不会检索到该 run。
注意：您无法通过由
mlflow.active_run返回的运行来访问当前活动运行的属性（参数、指标等）。为了访问这些属性，请按如下方式使用mlflow.client.MlflowClient：import 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[source]
为所有受支持的集成启用（或禁用）并配置 autologging。
这些参数会传递给支持它们的任何自动记录（autologging）集成。
请参阅 tracking docs 以获取受支持的 autologging 集成列表。
请注意，在任何时点设置的特定于框架的配置将优先于由此函数设置的任何配置。例如：
import mlflow mlflow.autolog(log_models=False, exclusive=True) import sklearn
将为 sklearn 启用 autologging，使用 log_models=False 和 exclusive=True，但
import mlflow mlflow.autolog(log_models=False, exclusive=True) import sklearn mlflow.sklearn.autolog(log_models=True)
将为 sklearn 启用自动日志记录（autologging），并设置 log_models=True 和 exclusive=False，后者来自 mlflow.sklearn.autolog 中 exclusive 的默认值；其他框架的自动日志记录函数（例如 mlflow.tensorflow.autolog）将使用由 mlflow.autolog 设置的配置（在此示例中，log_models=False，exclusive=True），直到用户显式调用它们。
- Parameters
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，则禁用所有受支持的 autologging integrations。如果False，则启用所有受支持的 autologging integrations。exclusive – 如果
True，自动记录的内容不会记录到用户创建的 fluent 运行。 如果False，自动记录的内容会记录到活动的 fluent 运行，该运行可能是用户创建的。disable_for_unsupported_versions – 如果
True，则对所有集成库中那些未针对此版本的 MLflow 客户端进行测试或与之不兼容的版本禁用自动日志记录（autologging）。silent – 如果
True，在 autologging 设置和训练执行期间抑制 MLflow 的所有事件日志和警告。如果False，在 autologging 设置和训练执行期间显示所有事件和警告。extra_tags – 一个字典，用于在 autologging 创建的每个托管运行上设置额外标签。
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.create_experiment(name: str, artifact_location: Optional[str] = None, tags: Optional[dict[str, typing.Any]] = None) str[source]
创建一个实验。
- Parameters
name – 实验名称，必须是非空且唯一的字符串。
artifact_location – 存储运行产物的位置。如果未提供，服务器会选择一个合适的默认值。
tags – 一个可选的字典，键和值均为字符串，用于在实验上设置标签。
- Returns
已创建实验的字符串 ID。
import mlflow from pathlib import Path # 创建一个实验名称，该名称必须唯一且区分大小写 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.delete_experiment(experiment_id: str) None[source]
从后端存储中删除一个实验。
- Parameters
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[source]
从当前实验中删除一个标签。
- Parameters
key – 要删除的标签的名称。
- mlflow.delete_run(run_id: str) None[source]
删除具有给定 ID 的运行。
- Parameters
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[source]
从 run 中删除一个标签。此操作不可逆。如果没有活动的 run，此方法将创建一个新的活动 run。
- Parameters
key – 标签的名称
- mlflow.delete_trace_tag(trace_id: str, key: str) None[source]
在具有给定 trace ID 的 trace 上删除一个标签。
该跟踪可以是正在活动的，也可以是已经结束并记录在后端。下面是一个在活动跟踪上删除标签的示例。您可以替换
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")
- Parameters
trace_id – 要从中删除标签的 trace 的 ID。
key – 标签的字符串键。必须最多为 250 个字符，否则在存储时会被截断。
- mlflow.disable_system_metrics_logging()[source]
在全局范围内禁用系统指标日志记录。
调用此函数将全局禁用系统指标记录，但用户仍然可以通过mlflow.start_run(log_system_metrics=True)为单次运行选择启用系统指标记录。
- mlflow.doctor(mask_envs=False)[source]
打印出用于调试 MLflow 问题的有用信息。
- Parameters
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()[source]
在全局启用系统指标日志记录。
调用此函数将全局启用系统指标记录，但用户仍然可以通过 mlflow.start_run(log_system_metrics=False) 为单个运行选择退出系统指标记录。
- mlflow.end_run(status: str = 'FINISHED') None[source]
结束一个活动的 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)[source]
在给定数据和所选指标上评估模型性能。
此函数使用指定的
evaluators在指定的数据集上评估 PyFunc 模型或自定义可调用对象，并将产生的指标 & 工件记录到 MLflow 跟踪服务器。用户也可以跳过设置model，直接将模型输出放入data中进行评估。有关详细信息，请阅读 the Model Evaluation documentation。- Default Evaluator behavior:
默认评估器，可通过
evaluators="default"或evaluators=None调用，支持下面列出的模型类型。对于每种预定义的模型类型，默认评估器会对您的模型在所选的一组指标上进行评估并生成诸如图表之类的工件。详情如下。对于
"regressor"和"classifier"两种模型类型，默认评估器使用 SHAP 生成模型摘要图和特征重要性图。- For regressor models, the default evaluator additionally logs:
metrics: 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.
- For binary classifiers, the default evaluator additionally logs:
metrics: true_negatives, false_positives, false_negatives, true_positives, recall, precision, f1_score, accuracy_score, example_count, log_loss, roc_auc, precision_recall_auc.
工件：提升曲线图、精确率-召回率曲线图、ROC 曲线图。
- For multiclass classifiers, the default evaluator additionally logs:
指标: accuracy_score, example_count, f1_score_micro, f1_score_macro, log_loss
artifacts: 一个用于“per_class_metrics”的CSV文件（每类指标包括 true_negatives/false_positives/false_negatives/true_positives/recall/precision/roc_auc, precision_recall_auc），精确率-召回率合并曲线图，ROC合并曲线图。
- For question-answering models, the default evaluator logs:
指标:
exact_match,token_count, toxicity（需要 evaluate、torch、flesch_kincaid_grade_level（需要 textstat）和 ari_grade_level。artifacts: 一个 JSON 文件，包含 inputs、outputs、targets（如果提供了
targets参数），以及以表格格式表示的模型每行指标。
- For text-summarization models, the default evaluator logs:
指标:
token_count, ROUGE (需要安装 evaluate, nltk, 和 rouge_score), toxicity (需要安装 evaluate, torch, transformers), ari_grade_level (需要安装 textstat), flesch_kincaid_grade_level (需要安装 textstat).artifacts: 一个 JSON 文件，包含模型在表格格式中的输入、输出、targets（如果提供了
targets参数）以及每行指标。
- For text models, the default evaluator logs:
指标:
token_count, toxicity (需要 evaluate, torch, transformers), ari_grade_level (需要 textstat), flesch_kincaid_grade_level (需要 textstat).artifacts: 一个 JSON 文件，包含 inputs、outputs、targets（如果提供了
targets参数），以及以表格格式表示的模型每行指标。
- For retriever models, the default evaluator logs:
指标:
precision_at_k(k),recall_at_k(k)和ndcg_at_k(k)- 默认值均为retriever_k= 3.artifacts: 一个 JSON 文件，包含模型的输入、输出、目标以及按行的指标，以表格格式呈现。
对于 sklearn 模型，默认评估器还会记录模型的评估准则（例如分类器的平均准确率），该准则由 model.score 方法计算。
上面列出的指标/工件会记录到活动的 MLflow 运行中。如果不存在活动的运行，将为记录这些指标和工件创建一个新的 MLflow 运行。
此外，有关指定数据集的信息——哈希、名称（如果已指定）、路径（如果已指定）以及评估它的模型的 UUID——会被记录到
mlflow.datasets标签中。- The available
evaluator_configoptions for the default evaluator include: log_model_explainability: 一个布尔值，指定是否记录模型可解释性洞见，默认值为 True。
- log_explainer: 如果为 True，则将用于计算模型可解释性洞察的解释器记录为模型
默认值为 False。
explainability_algorithm: 一个字符串，用于指定用于模型可解释性的 SHAP Explainer 算法。支持的算法包括: ‘exact’, ‘permutation’, ‘partition’, ‘kernel’。如果未设置，
shap.Explainer会使用 “auto” 算法，auto 会根据模型选择最佳 Explainer。explainability_nsamples: 用于计算模型可解释性洞察的样本行数。默认值为2000。
explainability_kernel_link: shap kernel explainer 使用的 kernel link 函数。可用值为 “identity” 和 “logit”。默认值是 “identity”。
max_classes_for_multiclass_roc_pr: 对于多类分类任务，记录每个类别的 ROC 曲线和精确率-召回率（Precision-Recall）曲线的最大类别数。如果类别数大于配置的最大值，则不会记录这些曲线。
metric_prefix: 一个可选的前缀，用于添加到评估期间生成的每个指标和工件的名称前。
log_metrics_with_dataset_info: 一个布尔值，指定是否在评估期间将关于评估数据集的信息包含在记录到 MLflow Tracking 的每个指标名称中，默认值为 True。
pos_label: 如果指定，用于在计算二分类模型的分类指标（例如 precision、recall、f1 等）时作为正类标签。对于多类分类和回归模型，将忽略此参数。
average：在计算多类分类模型的分类指标（例如 precision、recall、f1 等）时使用的平均方法（默认：
'weighted'）。对于二元分类和回归模型，将忽略此参数。sample_weights: 在计算模型性能指标时应用于每个样本的权重。
col_mapping: 一个字典，将输入数据集或输出预测中的列名映射为在调用评估函数时使用的列名。
retriever_k: 当
model_type="retriever"时使用的参数，表示在计算内置指标precision_at_k(k)、recall_at_k(k)和ndcg_at_k(k)时要使用的排名靠前的检索文档数量。默认值为 3。对于所有其他模型类型，此参数将被忽略。
- The available
- Limitations of evaluation dataset:
对于分类任务，使用数据集标签来推断总类别数。
对于二分类任务，负标签值必须为 0、-1 或 False，正标签值必须为 1 或 True。
- Limitations of metrics/artifacts computation:
对于分类任务，某些指标和工件的计算要求模型输出类别概率。目前，对于 scikit-learn 模型，默认评估器会调用底层模型的
predict_proba方法以获取概率。对于其他模型类型，默认评估器不会计算那些需要概率输出的指标/工件。
- Limitations of default evaluator logging model explainability insights:
shap.Explainerauto算法对线性模型使用Linear解释器， 对树模型使用Tree解释器。因为 SHAP 的Linear和Tree解释器不支持多类分类， 默认评估器会在多类分类任务中回退使用Exact或Permutation解释器。目前不支持为 PySpark 模型记录可解释性洞见。
评估数据集的标签值必须为数值或布尔值，所有特征值必须为数值，并且每个特征列只能包含标量值。
- Limitations when environment restoration is enabled:
当为被评估模型启用环境还原时（即指定了非本地的
env_manager），模型会以客户端的方式加载，在一个独立的 Python 环境中调用 MLflow Model Scoring Server 进程，该环境已安装模型训练时的依赖项。因此，模型的诸如predict_proba（用于概率输出）或score（计算 sklearn 模型的评估准则）等方法将不可访问，默认评估器不会计算依赖这些方法的指标或工件。由于该模型是 MLflow Model Server 进程，SHAP 解释计算较慢。因此，当指定了非本地的
env_manager时，模型可解释性功能会被禁用，除非在evaluator_config中显式将 log_model_explainability 选项设置为True。
- Parameters
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，包含评估特征、标签，且可选地包含模型输出
当未指定 model 时，必须提供模型输出。 如果未指定
feature_names参数，除标签列和预测列外的所有列都视为特征列。否则，仅将出现在feature_names中的列名视为特征列。
- 一个 Spark DataFrame，包含评估特征和标签。若
未指定
feature_names参数，除标签列外的所有列都视为特征列。否则，仅将出现在feature_names中的列名视为特征列。Spark DataFrame 中仅使用前 10000 行作为评估数据。
- 一个
mlflow.data.dataset.Dataset实例，包含评估 特征、标签，且可选地包含模型输出。模型输出仅在 PandasDataset 中受支持。当未指定 model 时，必须提供模型输出，并且应通过 PandasDataset 的
predictions属性来指定。
- 一个
model_type –
(可选) 描述模型类型的字符串。默认评估器支持以下模型类型：
'classifier''regressor''question-answering''text-summarization''text''retriever'
如果未指定
model_type，则必须通过extra_metrics参数提供要计算的指标列表。注意
'question-answering','text-summarization','text', and'retriever'是实验性的，可能在将来版本中更改或移除。targets – 如果
data是 numpy 数组或列表，则为评估标签的 numpy 数组或列表。如果data是 DataFrame，则为data中包含评估标签的列的字符串名称。对于分类和回归模型这是必需的，但对于问答、文本摘要和文本模型是可选的。如果data是定义了 targets 的mlflow.data.dataset.Dataset，则targets可选。predictions –
可选。包含模型输出的列名。
当
model被指定并输出多列时，predictions可用于指定用于存储评估时模型输出的列名。当
model未被指定且data是一个 pandas dataframe 时，predictions可用于指定data中包含模型输出的列名。
# 评估输出多列的模型 data = pd.DataFrame({"question": ["foo"]}) def model(inputs): return pd.DataFrame({"answer": ["bar"], "source": ["baz"]}) results = evaluate( model=model, data=data, predictions="answer", # 如有需要可传入其他参数 ) # 评估静态数据集 data = pd.DataFrame({"question": ["foo"], "answer": ["bar"], "source": ["baz"]}) results = evaluate( data=data, predictions="answer", # 如有需要可传入其他参数 )
dataset_path – (可选) 数据存放路径。不得包含双引号 (
")。如果指定，该路径会被记录到mlflow.datasets标签中，以用于血缘追踪。feature_names –（可选）一个列表。如果
data参数是一个 numpy 数组或列表，feature_names是每个特征的特征名称组成的列表。如果feature_names=None，则使用格式feature_{feature_index}生成feature_names。如果data参数是 Pandas DataFrame 或 Spark DataFrame，feature_names是 DataFrame 中特征列名称的列表。如果feature_names=None，则除标签列和预测列外的所有列都被视为特征列。evaluators – 要用于模型评估的评估器名称，或评估器名称列表。如果未指定，将使用所有能够在指定模型的指定数据集上进行评估的评估器。默认评估器可以通过名称
"default"引用。要查看所有可用的评估器，请调用mlflow.models.list_evaluators()。evaluator_config – 一个提供给评估器的附加配置字典。 如果指定了多个评估器，每个配置都应作为一个嵌套字典提供，其键为评估器名称。
extra_metrics –
(Optional) A list of
EvaluationMetricobjects. These metrics are computed in addition to the default metrics associated with pre-defined model_type, and setting model_type=None will only compute the metrics specified in extra_metrics. See the mlflow.metrics module for more information about the builtin metrics and how to define extra 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 –
(Optional) A list of custom artifact functions with the following signature:
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``. """ ...
Object types that artifacts can be represented as:
A string uri representing the file path to the artifact. MLflow will infer the type of the artifact based on the file extension.
A string representation of a JSON object. This will be saved as a .json artifact.
Pandas DataFrame. This will be resolved as a CSV artifact.
Numpy array. This will be saved as a .npy artifact.
Matplotlib Figure. This will be saved as an image artifact. Note that
matplotlib.pyplot.savefigis called behind the scene with default configurations. To customize, either save the figure with the desired configurations and return its file path or define customizations through environment variables inmatplotlib.rcParams.Other objects will be attempted to be pickled with the default protocol.
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 flavor 以了解哪些键被您的特定模型支持。若未指定，则使用模型的默认模型配置（如果有）。
inference_params –（可选）一个在进行预测时要传递给模型的推理参数字典，例如
{"max_tokens": 100}。仅当model是 MLflow Deployments 的端点 URI 时使用，例如"endpoints:/my-chat"model_id – （可选）要将评估结果（例如指标和跟踪）链接到的 MLflow LoggedModel 或 Model Version 的 ID。如果 model_id 未指定但指定了 model，则将使用 model 的 ID。
_called_from_genai_evaluate – （可选）仅供内部使用。
- Returns
一个
mlflow.models.EvaluationResult实例，包含使用给定数据集评估模型的指标。
- mlflow.flush_artifact_async_logging() None[source]
刷新所有待处理的 artifact 的异步日志。
- mlflow.flush_async_logging() None[source]
刷新所有待处理的异步日志记录。
- mlflow.flush_trace_async_logging(terminate=False) None[source]
刷新所有待处理的跟踪异步日志记录。
- Parameters
terminate – 如果 True，则在刷新后关闭日志记录线程。
- mlflow.get_active_model_id() str | None[source]
获取活动模型 ID。如果没有使用
set_active_model()设置活动模型，则会使用环境变量MLFLOW_ACTIVE_MODEL_ID或旧的环境变量_MLFLOW_ACTIVE_MODEL_ID中的模型 ID 来设置默认活动模型。 如果都未设置，则返回 None。注意：该函数仅从当前线程获取活动模型 ID。- Returns
如果已设置，则为活动模型 ID，否则为 None。
- mlflow.get_active_trace_id() str | None[source]
获取当前进程中活动的跟踪 ID。
此函数是线程安全的。
示例：
import mlflow @mlflow.trace def f(): trace_id = mlflow.get_active_trace_id() print(trace_id) f()
- Returns
当前活动的跟踪 ID（如果存在），否则为 None。
- mlflow.get_artifact_uri(artifact_path: Optional[str] = None) str[source]
获取当前活动运行中指定工件的绝对 URI。
如果path未指定，将返回当前活动运行的 artifact 根 URI；对
log_artifact和log_artifacts的调用会将 artifact(s) 写入 artifact 根 URI 的子目录。如果没有活动的运行，此方法将创建一个新的活动运行。
- Parameters
artifact_path – 需要获取绝对 URI 的相对于运行的 artifact 路径。 For example, “path/to/artifact”。如果未指定，则返回当前活动运行的 artifact 根 URI。
- Returns
指向指定artifact或当前活动运行的artifact根目录的绝对 URI。例如，如果提供了artifact路径并且当前活动运行使用基于 S3 的存储，则该 URI 可能具有如下形式：
s3://。如果未提供artifact路径并且当前活动运行使用基于 S3 的存储，则该 URI 可能具有如下形式：/path/to/artifact/root/path/to/artifact s3://。/path/to/artifact/root
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) 评估[source]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
从后端存储中获取评估实体。
- Parameters
trace_id – 跟踪的 ID。
assessment_id – 要获取的评估的 ID。
- Returns
该 Assessment 对象。
- Return type
- mlflow.get_experiment(experiment_id: str) Experiment[source]
通过 experiment_id 从后端存储检索实验
- Parameters
experiment_id – 从
create_experiment返回的字符串形式的实验 ID。- Returns
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]
通过实验名称从后端存储中检索实验
- Parameters
name – 区分大小写的实验名称。
- Returns
如果存在具有指定名称的实验，则返回一个
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) 运行 | None[source]
获取给定 run id 的父运行（如果存在）。
- Parameters
run_id – 子运行的唯一标识符。
- Returns
如果父运行存在，则返回单个
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。
- Returns
注册表的 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。得到的 Run 包含一组运行元数据 – RunInfo 以及一组运行参数、标签和指标 – RunData。它还包含一组运行输入（实验性），包括关于 run 所使用的数据集的信息 – RunInputs。在为该 run 记录了多个具有相同键的指标的情况下，RunData 包含每个指标在最大 step 时最近记录的值。
- Parameters
run_id – 运行的唯一标识符。
- Returns
如果运行存在，则返回单个 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]
获取当前 tracking URI。 这可能与当前活动运行的 tracking URI 不一致，因为 tracking URI 可以通过
set_tracking_uri更新。- Returns
跟踪 URI。
import mlflow # Get the current tracking uri tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}")
Current tracking uri: file:///.../mlruns
- mlflow.is_tracking_uri_set()[source]
如果 tracking 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()
import mlflow mlflow.start_run() mlflow.end_run() run = mlflow.last_active_run()
import mlflow mlflow.start_run() run = mlflow.last_active_run() mlflow.end_run()
- Returns
活动运行（等同于
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 Tracking 加载为 pandas.DataFrame。该表从指定的 artifact_file 在指定的 run_ids 中加载。extra_columns 是不在表中但使用运行信息增强并添加到 DataFrame 的列。
- Parameters
artifact_file – 要加载的表对应的、相对于运行的 artifact 文件路径，采用 posixpath 格式（例如 “dir/file.json”）。
run_ids – 可选的 run_ids 列表，用于从中加载表格。如果未指定任何 run_ids, 表格将从当前实验中的所有运行加载。
extra_columns – 可选的额外列列表，添加到返回的 DataFrame 中。例如，如果 extra_columns=[“run_id”]，那么返回的 DataFrame 将有一个名为 run_id 的列。
- Returns
如果 artifact 存在，则返回包含已加载表的 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]
将本地文件或目录记录为当前活动运行的工件。如果没有活动运行，该方法将创建一个新的活动运行。
- Parameters
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]
将本地目录的所有内容记录为该运行的工件。如果当前没有激活的运行，此方法将创建一个新的激活运行。
- Parameters
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）作为 artifact 进行记录。序列化格式（JSON 或 YAML）会根据 artifact_file 的扩展名自动推断。如果文件扩展名不存在或不匹配 [“.json”, “.yml”, “.yaml”] 中的任何一项，则使用 JSON 格式。
- Parameters
dictionary – 要记录的字典。
artifact_file – 以 posixpath 格式表示的、相对于运行的 artifact 文件路径，字典将保存到该路径（例如 “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]
将图形记录为工件。支持以下图形对象：
- Parameters
figure – 要记录的图形。
artifact_file – 图像保存到的、以 posixpath 格式表示的运行相对 artifact 文件路径（例如 “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 中记录图像，支持两种用例：
- Time-stepped image logging:
非常适合在迭代过程中（例如模型训练阶段）跟踪变化或进展。
用法：
log_image(image, key=key, step=step, timestamp=timestamp)
- Artifact file image logging:
最适合用于静态图像的日志记录场景，其中图像被直接保存为文件工件。
用法：
log_image(image, artifact_file)
- The following image formats are supported:
-
mlflow.Image: 一个 MLflow 包装器，基于 PIL 图像，便于图像日志记录。
- Numpy array support
数据类型：
bool (对记录图像掩码很有用)
整数 [0, 255]
无符号整数 [0, 255]
float [0.0, 1.0]
警告
Out-of-range integer values will raise ValueError.
超出范围的浮点值会随 min/max 自动缩放并发出警告。
形状 (H: height, W: width):
H x W (灰度)
H x W x 1 (灰度)
H x W x 3 (假定为 RGB 通道顺序)
H x W x 4 (假定为 RGBA 通道顺序)
- Parameters
image – 要记录的 image 对象。
artifact_file – 指定以 POSIX 格式的路径，用于将图像作为 artifact 存储在相对于运行根目录的位置（例如，“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]
记录在当前运行中使用的数据集。
- Parameters
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，表示对应输入的空值。- Parameters
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_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[mlflow.data.dataset.Dataset] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
在当前运行下记录一个指标。如果没有活动的运行，此方法将创建一个新的活动运行。
- Parameters
key – 指标名称。该字符串只能包含字母数字字符、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储将支持长度最多为 250 的 keys，但有些可能支持更长的 keys。
value – 指标值。请注意，某些特殊值（例如 +/- Infinity）可能会根据存储而被替换为其他值。例如，SQLAlchemy 存储会将 +/- Infinity 替换为最大/最小浮点值。所有后端存储将支持长度最多为 5000 的值，但有些可能支持更大的值。
step – 度量的 step。未指定时默认为零。
synchronous – 实验性 如果为 True，则阻塞直到指标被成功记录。如果为 False，则以异步方式记录指标，并返回一个表示该日志记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 中读取，若未设置则默认为 False。
timestamp – 表示计算此指标的时间。默认为当前系统时间。
run_id – 如果指定，则将指标记录到指定的运行。如果未指定，则将指标记录到当前活动的运行。
model_id – 与该指标关联的模型 ID。如果未指定，则使用由
mlflow.set_active_model()设置的当前活动模型 ID。如果不存在活动模型，则使用与指定或活动运行关联的模型 ID。dataset – 与指标相关的数据集。
- Returns
当 synchronous=True 时，返回 None。 当 synchronous=False 时，返回 RunOperations，表示用于日志记录操作的 future 对象。
- 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[mlflow.data.dataset.Dataset] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
为当前运行记录多个指标。如果没有活动的运行，该方法将创建一个新的活动运行。
- Parameters
metrics – 字典，形式为 metric_name: String -> value: Float。注意，某些特殊值，例如 +/- Infinity，可能会根据存储而被替换为其他值。例如，基于 sql 的存储可能会将 +/- Infinity 替换为最大/最小 float 值。
step – 用于在指定的 Metrics 上记录的单个整数步骤。如果未指定，则每个指标都会记录在 step zero。
synchronous – 实验性 如果为 True，会阻塞直到指标成功记录。如果为 False，则以异步方式记录指标并返回一个表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 读取；如果未设置，默认为 False。
run_id – 如果指定，将指标记录到指定的运行。如果未指定，则将指标记录到当前活动的运行。
timestamp – 计算这些指标的时间。默认使用当前系统时间。
model_id – 与该指标关联的模型的 ID。如果未指定，则使用由
mlflow.set_active_model()设置的当前活动模型 ID。如果不存在活动模型，则将使用与指定的运行或活动运行关联的模型 ID。dataset – 与指标关联的数据集。
- Returns
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回一个
mlflow.utils.async_logging.run_operations.RunOperations实例，该实例代表用于日志记录操作的将来对象。
- mlflow.log_outputs(models: Optional[list[LoggedModelOutput]] = None)[source]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
将输出（例如模型）记录到活动运行。如果没有活动运行，将创建一个新运行。
- Parameters
models – 要记录为运行输出的
mlflow.entities.LoggedModelOutput实例列表。- Returns
无。
- mlflow.log_param(key: str, value: Any, synchronous: Optional[bool] = None) Any[source]
在当前运行下记录一个参数（例如模型超参数）。如果没有活动的运行，该方法将创建一个新的活动运行。
- Parameters
key – 参数名。该字符串只能包含字母数字、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储支持键的长度最多为250个字符，但有些可能支持更长的键。
value – 参数值，如果不是字符串则会被转换为字符串。所有内置后端存储支持长度最多为6000 的值，但有些可能支持更大的值。
synchronous – 实验性 如果为 True，则阻塞直到参数成功记录。如果为 False，则异步记录该参数并返回一个表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 读取，若未设置则默认为 False。
- Returns
当 synchronous=True 时，返回参数值。当 synchronous=False 时，返回一个
mlflow.utils.async_logging.run_operations.RunOperations实例，表示用于日志记录操作的 future 对象。
- 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 到当前运行。如果没有运行处于活动状态，该方法将创建一个新的活动运行。
- Parameters
params – 字典，键为 param_name: String -> 值为 value: (String，但如果不是字符串则会被转换为字符串)
synchronous – 实验性 如果为 True，则阻塞直到参数成功记录。如果为 False，则异步记录参数并返回表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 中读取，如果未设置则默认为 False。
run_id – 运行 ID。如果指定，则将参数记录到指定的运行。如果未指定，则将参数记录到当前活动的运行。
- Returns
当 synchronous=True 时，返回 None。 当 synchronous=False 时，返回一个
mlflow.utils.async_logging.run_operations.RunOperations实例，表示用于日志记录操作的 future。
- mlflow.log_table(data: Union[dict[str, typing.Any], pandas.DataFrame], artifact_file: str, run_id: str | None = None) None[source]
将表以 JSON 工件的形式记录到 MLflow Tracking。如果 artifact_file 已经存在于运行中，数据将追加到现有的 artifact_file。
- Parameters
data – 要记录的字典或 pandas.DataFrame。
artifact_file – 以 run 为基准的 artifact 文件路径，采用 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]
将文本记录为一个artifact。
- Parameters
text – String 包含要记录的文本。
artifact_file – 保存文本的相对于运行的 artifact 文件路径，采用 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: Optional[dict[str, typing.Any]] = None, attributes: Optional[dict[str, typing.Any]] = None, tags: Optional[dict[str, str]] = None, start_time_ms: Optional[int] = None, execution_time_ms: Optional[int] = None) str[source]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
创建一个具有单个根跨度的跟踪。 当您想要记录任意的（request, response）对而不使用结构化的 OpenTelemetry 跨度时，此 API 非常有用。该跟踪链接到活动实验。
- Parameters
name – 跟踪的名称（以及根 span）。默认值为 “Task”.
request – 整个 trace 的输入数据。这也会设置在 trace 的根 span 上。
response – 整个 trace 的输出数据。这也会设置在 trace 的根 span 上。
intermediate_outputs – 一个字典，包含模型或智能体在处理请求时产生的中间输出。键是输出的名称，值就是这些输出本身。值必须可被JSON序列化。
attributes – 一个字典，用于在追踪的根 span 上设置属性。
tags – 一个用于在跟踪上设置标签的字典。
start_time_ms – 跟踪的开始时间，以自 UNIX 纪元以来的毫秒为单位。 当未指定时，当前时间将被用作跟踪的开始和结束时间。
execution_time_ms – 跟踪的执行时间（自 UNIX 纪元以来的毫秒数）。
- Returns
已记录跟踪的 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) print(trace.data.intermediate_outputs)
- mlflow.login(backend: str = 'databricks', interactive: bool = True) None[source]
配置 MLflow 服务器身份验证并将 MLflow 连接到跟踪服务器。
此方法提供了一种将 MLflow 连接到其跟踪服务器的简单方式。当前仅支持 Databricks 跟踪服务器。如果未找到现有的 Databricks 配置文件，将提示用户输入凭据，凭据会被保存到 ~/.databrickscfg。
- Parameters
backend – string，跟踪服务器的后端。当前仅支持 “databricks”。
interactive – bool，控制在缺少凭证时是否请求用户输入。如果 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()。- Parameters
trace_id – 跟踪的 ID。
assessment_id – 要覆盖的评估的 ID。
value – 评估的新值。
rationale – 新评估的理由。
source – 新评估的来源。
metadata – 用于新评估的附加元数据。
- Returns
已创建的评估。
- Return type
示例
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, 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[Literal['databricks_model_serving']] = None) 模型版本[source]
在模型注册表中为由
model_uri指定的模型文件创建一个新的模型版本。请注意，该方法假定模型注册表后端的 URI 与跟踪后端的 URI 相同。
- Parameters
model_uri – 指向 MLmodel 目录的 URI。 如果您想在模型注册表中随模型记录运行 ID（推荐），请使用
runs:/URI；或者，如果要注册之前使用save_model保存的本地持久化 MLflow 模型，请传入该模型的本地文件系统路径。models:/URI 目前不受支持。name – 要在其下创建新模型版本的注册模型的名称。如果具有给定名称的注册模型不存在，则会自动创建。
await_registration_for – 等待模型版本完成创建并处于
READY状态的秒数。默认情况下，函数等待五分钟。指定 0 或 None 可跳过等待。tags – 一个由键值对组成的字典，这些键值对被转换为
mlflow.entities.model_registry.ModelVersionTag对象。env_pack –
如果指定，模型依赖将首先安装到当前 Python 环境，然后完整的环境会被打包并包含在已注册的模型工件中。当将模型部署到诸如 Databricks Model Serving 之类的服务环境时，这非常有用。
注意
实验性：此参数可能在未来版本中未经警告就更改或移除。
- Returns
由后端创建的单个
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 集群上远程运行项目。您也可以通过安装适当的第三方插件，将项目运行到其他目标。有关更多信息，请参见 Community Plugins。
有关在链式工作流中使用此方法的信息，请参见 Building Multistep Workflows。
- Raises
- Parameters
uri – 要运行的项目的 URI。一个本地文件系统路径或 Git 仓库的 URI（例如 https://github.com/mlflow/mlflow-example），指向包含 MLproject 文件的项目目录。
entry_point – 在项目中运行的入口点。如果未找到具有指定名称的入口点，则将项目文件
entry_point作为脚本运行，使用 “python” 来运行.py文件，并使用默认 shell（由环境变量$SHELL指定）来运行.sh文件。version – 对于基于 Git 的项目，可以是提交哈希或分支名。
parameters – 参数（字典），用于入口点命令。
docker_args – 用于 docker 命令的参数（字典）。
experiment_name – 要在其下启动运行的实验名称。
experiment_id – 启动运行时所属实验的 ID。
backend – 运行的执行后端：MLflow 提供对 “local”, “databricks”, 和 “kubernetes”（实验性）后端的内置支持。如果针对 Databricks 运行，将根据以下方式确定要运行的 Databricks 工作区：如果已设置形式为
databricks://profile的 Databricks 跟踪 URI（例如通过设置 MLFLOW_TRACKING_URI 环境变量），则在指定的工作区上运行。否则，在默认 Databricks CLI profile 指定的工作区上运行。 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 project APIs 在内部使用，不应指定。如果指定，则会使用 run ID 而不是创建新的运行。
run_name – 要赋予与项目执行相关联的 MLflow Run 的名称。如果
None，MLflow Run 的名称将保持未设置。env_manager –
指定用来为运行创建新环境并在该环境中安装项目依赖的环境管理器。支持以下值：
local: 使用本地环境
virtualenv: 使用 virtualenv（并使用 pyenv 管理 Python 版本）
uv: 使用 uv
conda: 使用 conda
如果未指定，MLflow 会通过检查项目目录中的文件自动确定要使用的环境管理器。例如，如果
python_env.yaml存在，则会使用 virtualenv。build_image – 是否为该项目构建新的 docker 镜像或重用现有镜像。默认：False（重用现有镜像）
docker_auth – 表示用于对 Docker 注册表进行身份验证的信息的字典。参见 docker.client.DockerClient.login 了解可用选项。
- Returns
mlflow.projects.SubmittedRun暴露已启动运行的信息（例如 run 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[实验][source]
搜索与指定搜索查询匹配的实验。
- Parameters
view_type – 枚举值之一
ACTIVE_ONLY、DELETED_ONLY或ALL，定义在mlflow.entities.ViewType。max_results – 如果传入，指定所需的最大实验数量。如果不传入，则会返回所有实验。
filter_string –
筛选查询字符串（例如，
"name = 'my_experiment'"），默认为搜索所有实验。支持以下标识符、比较运算符和逻辑运算符。- 标识符
name：实验名称creation_time：实验创建时间last_update_time：实验最后更新时间tags.：实验标签。如果tag_key包含空格，则必须用反引号包裹（例如，"tags.`extra key`"）。
- 字符串属性和标签的比较运算符
=：等于!=：不等于LIKE：区分大小写的模式匹配ILIKE：不区分大小写的模式匹配
- 数字属性的比较运算符
=：等于!=：不等于<：小于<=：小于或等于>：大于>=：大于或等于
- 逻辑运算符
AND：将两个子查询组合，如果两者都为 True 则返回 True。
order_by –
要排序的列列表。The
order_by列可以包含可选的DESC或ASC值（例如，"name DESC"）。默认排序为ASC， 所以"name"等同于"name ASC"。如果未指定，默认为["last_update_time DESC"]，这会将最近更新的实验列在前面。 以下字段受到支持：experiment_id: 实验 IDname: 实验名称creation_time: 实验创建时间last_update_time: 实验最后更新时间
- Returns
一个由
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]
搜索符合筛选条件的模型版本。
- Parameters
max_results – 如果传入，指定所需的最大模型数量。如果不传入，则会返回所有模型。
filter_string –
筛选查询字符串 (e.g.,
"name = 'a_model_name' and tag.key = 'value1'"), 默认为搜索所有模型版本。支持以下标识符、比较符和逻辑运算符。- 标识符
name：模型名称。source_path：模型版本的源路径。run_id：生成该模型版本的 mlflow 运行的 ID。tags.：模型版本标签。如果tag_key包含空格，则必须用反引号包裹（例如，"tags.`extra key`"）。
- 比较符
=：等于。!=：不等于。LIKE：区分大小写的模式匹配。ILIKE：不区分大小写的模式匹配。IN：在一个值列表中。只有run_id标识符支持IN比较符。
- 逻辑运算符
AND：连接两个子查询，只有两个子查询都为 True 时才返回 True。
order_by – 列名列表，带有 ASC|DESC 注释，用于对匹配的搜索结果进行排序。
- Returns
- 符合搜索表达式的
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]
搜索满足筛选条件的已注册模型。
- Parameters
max_results – 如果传入，指定所需的最大模型数量。如果不传入，则会返回所有模型。
filter_string –
过滤查询字符串（例如，“name = ‘a_model_name’ and tag.key = ‘value1’”），默认搜索所有已注册模型。支持以下标识符、比较运算符和逻辑运算符。
- 标识符
”name”: 已注册模型的名称。
”tags.
”: 已注册模型的标签。如果“tag_key”包含空格，必须用反引号包裹（例如，“tags.`extra key`”）。
- 比较运算符
”=”：等于。
”!=”：不等于。
”LIKE”：区分大小写的模式匹配。
”ILIKE”：不区分大小写的模式匹配。
- 逻辑运算符
”AND”：将两个子查询组合起来，只有当两者都为 True 时才返回 True。
order_by – 带有 ASC|DESC 注解的列名列表，用于对匹配的搜索结果进行排序。
- Returns
满足搜索表达式的
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]
搜索符合指定条件的运行。
- Parameters
experiment_ids – 实验 ID 的列表。搜索可以使用实验 ID 或 实验名称，但不能在同一次调用中同时使用两者。若
None或[]之外的值被提供，并且experiment_names也不是None或[]，则会导致错误。如果experiment_names是None或[]，则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或[]，则None将默认为活动实验。
- Returns
一个
mlflow.entities.Run的列表。如果 output_format 是pandas：返回一个pandas.DataFrame的运行记录，其中每个指标、参数和标签分别展开为各自的列，命名为 metrics.*、params.* 或 tags.*。对于没有某个特定指标、参数或标签的运行记录，相应列的值分别为（NumPy）Nan、None或None。- Return type
If output_format is
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.set_experiment(experiment_name: Optional[str] = None, experiment_id: Optional[str] = None) Experiment[source]
将给定的实验设置为活动实验。实验必须通过名称 experiment_name 指定，或通过 ID experiment_id 指定。实验名称和 ID 不能同时指定。
注意
如果按名称设置的实验不存在，将会使用给定的名称创建一个新的实验。实验创建后，它将被设置为活动实验。在某些平台上，例如 Databricks，实验名称必须是一个绝对路径，例如
"/Users/。/my-experiment" - Parameters
experiment_name – 要激活的实验的名称（区分大小写）。
experiment_id – 要激活的实验的 ID。如果不存在具有此 ID 的实验，则会抛出异常。
- Returns
一个表示新的活动实验的
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]
在当前实验上设置一个标签。值将被转换为字符串。
- Parameters
key – 标签名称。该字符串只能包含字母数字字符、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储将支持最长为 250 的键，但有些可能支持更长的键。
value – 标签值，但如果不是字符串则会被转换为字符串。所有后端存储都会支持长度最多为5000的值，但有些可能支持更大的值。
- mlflow.set_experiment_tags(tags: dict[str, typing.Any]) None[source]
为当前活动的实验设置标签。
- Parameters
tags – 包含标签名称及其对应值的字典。
- mlflow.set_model_version_tag(name: str, version: Optional[str] = None, key: Optional[str] = None, value: Optional[Any] = None) None[source]
为模型版本设置标签。
- Parameters
name – 已注册模型名称。
version – 注册的模型版本。
key – 要记录的标签键。key 是必需的。
value – 要记录的标签值。 value 是必需的。
- mlflow.set_registry_uri(uri: str) None[source]
设置注册表服务器 URI。此方法在你的注册表服务器与跟踪服务器不同的情况下特别有用。
- Parameters
uri – 一个空字符串，或以
file:/为前缀的本地文件路径。数据存储在提供的文件中（如果为空，则存储在./mlruns）。类似https://my-tracking-server:5000或http://my-oss-uc-server:8080的 HTTP URI。一个 Databricks 工作区，可用字符串 “databricks” 提供，或者，为使用 Databricks CLI 的 profile，使用 “databricks://”。
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]
在当前运行下设置一个标签。如果没有活动的运行，此方法将创建一个新的活动运行。
- Parameters
key – 标签名称。该字符串只能包含字母数字字符、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储将支持长度不超过 250 的 keys，但有些可能支持更长的 keys。
value – 标签的值，但如果不是字符串会被转换为字符串。所有后端存储将支持最大长度为5000的值，但有些可能支持更大的值。
synchronous – 实验性 如果为 True，则阻塞直到标签成功记录。如果为 False，则异步记录该标签并返回一个表示记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 中读取；如果未设置，默认为 False。
- Returns
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回一个
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]
为当前运行记录一批标签。如果没有活动的运行，此方法将创建一个新的活动运行。
- Parameters
tags – 字典，tag_name: String -> value: (String，但如果不是会被转为字符串)
synchronous – 实验性 如果 True，则会阻塞直到标签成功记录。如果 False，则以异步方式记录标签并返回一个表示该记录操作的 future。如果 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 中读取；如果未设置，默认为 False。
- Returns
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回一个
mlflow.utils.async_logging.run_operations.RunOperations实例，该实例代表用于日志记录操作的将来对象。
- mlflow.set_trace_tag(trace_id: str, key: str, value: str)[source]
在具有给定 trace ID 的跟踪上设置一个标签。
跟踪可以是正在进行的，也可以是已经结束并记录在后端的。下面是为正在进行的跟踪设置标签的示例。您可以替换
trace_id参数，以在已结束的跟踪上设置标签。import mlflow with mlflow.start_span(name="span") as span: mlflow.set_trace_tag(span.trace_id, "key", "value")
- Parameters
trace_id – 要设置标签的 trace 的 ID。
key – 标签的字符串键。必须最多为 250 个字符，否则在存储时会被截断。
value – 标签的字符串值。长度最多为250个字符，否则存储时会被截断。
- mlflow.set_tracking_uri(uri: str | pathlib.Path) None[source]
设置跟踪服务器 URI。 这不会影响当前活动的运行（如果存在），但会对后续运行生效。
- Parameters
uri –
一个空字符串，或一个以
file:/为前缀的本地文件路径。数据存储在提供的文件中（如果为空，则为./mlruns）。类似
https://my-tracking-server:5000的 HTTP URI。一个 Databricks 工作区，通过字符串 “databricks” 提供，或者要使用 Databricks CLI profile，使用 “databricks://
”。 一个
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.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[source]
开始一个新的 MLflow 运行，并将其设置为活动运行，在该运行下将记录指标和参数。其返回值可以在一个
with代码块中用作上下文管理器；否则，您必须调用end_run()来终止当前运行。如果你传入
run_id或者 设置了环境变量MLFLOW_RUN_ID，start_run会尝试使用指定的运行 ID 恢复运行，其它参数将被忽略。run_id优先于MLFLOW_RUN_ID。如果恢复现有运行，运行状态将设置为
RunStatus.RUNNING。MLflow 会在运行上设置各种默认标签，如在 MLflow system tags 中所定义。
- Parameters
run_id – 如果指定，获取具有指定 UUID 的运行，并在该运行下记录参数和指标。该运行的结束时间未设置，其状态被设为 running，但该运行的其他属性（
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 – 一个可选的字典，包含字符串键和值，用于在运行上设置 tags。如果正在恢复一个运行，这些 tags 会被设置到恢复后的运行上。如果正在创建一个新的运行，这些 tags 会被设置到新的运行上。
description – 一个可选的字符串，用于填充运行的描述框。 如果正在恢复一个运行，description 会被设置到恢复后的运行上。 如果正在创建一个新运行，description 会被设置到新运行上。
log_system_metrics – bool，默认为 None。如果是 True，系统指标将被记录到 MLflow，例如 CPU/GPU 利用率。如果是 None，我们将检查环境变量 MLFLOW_ENABLE_SYSTEM_METRICS_LOGGING 来确定是否记录系统指标。系统指标记录是 MLflow 2.8 中的一个实验性功能，可能会发生变化。
- Returns
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.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)[source]
使用给定的选项更新当前活动的跟踪。
- Parameters
tags – 一个用于更新跟踪记录的标签字典。标签用于可变的值，这些值可以在通过 MLflow UI 或 API 创建跟踪记录之后更新。
metadata – 一个用于更新 trace 的 metadata 字典。metadata 在 trace 被记录后无法更新。它适合记录像生成该 trace 的应用版本的 git 哈希这样的不可变值。
client_request_id – 客户提供的请求 ID，用于将该跟踪（trace）关联起来。这有助于将跟踪追溯到您应用程序或外部系统中的特定请求。如果 None，则 client_request_id 不会被更新。
request_preview – 在用户界面（UI）的跟踪列表视图中显示的请求预览。默认情况下，MLflow 会通过限制长度来简单地截断跟踪请求。此参数允许您指定自定义的预览字符串。
response_preview – 在 UI 的 Trace 列表视图中显示的响应预览。默认情况下，MLflow 会通过限制长度来简单截断 trace 响应。此参数允许您指定自定义的预览字符串。
state – The state to set on the trace. Can be a TraceState enum value or string. Only “OK” and “ERROR” are allowed. This overrides the overall trace state without affecting the status of the current span.
示例
您可以在被
@mlflow.trace装饰的函数内部使用此函数，或在 with mlflow.start_span 上下文管理器的作用域内使用。如果未找到活动 trace，该函数将抛出异常。在被 @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 默认会从环境信息中填充它们。您可以在需要时覆盖它们。有关保留元数据键的完整列表，请参阅 MLflow Tracing 文档。
mlflow.update_current_trace( metadata={ "mlflow.trace.session": "session-4f855da00427", "mlflow.trace.user": "user-id-cc156f29bcfb", "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.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)[source]
验证来自一个模型（候选模型）的评估结果与另一个模型（基线模型）的结果。如果候选结果未达到验证阈值，将抛出 ModelValidationFailedException。
注意
该 API 替代了已弃用的
mlflow.evaluate()API 中的模型验证功能。- Parameters
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, )
请参阅 the Model Validation documentation 以获取更多详细信息。
MLflow 跟踪 APIs
该 mlflow 模块提供了一组用于 MLflow Tracing 的高级 API。有关如何使用这些追踪 API 的详细指导，请参阅 Tracing Fluent APIs Guide。
- mlflow.trace(func: Optional[Callable[[...], Any]] = None, name: Optional[str] = None, span_type: str = 'UNKNOWN', attributes: Optional[dict[str, typing.Any]] = None, output_reducer: Optional[Callable[[list[typing.Any]], Any]] = None) Callable[[...], Any][source]
一个装饰器，为被装饰的函数创建一个新的 span。
When you decorate a function with this
@mlflow.trace()decorator, a span will be created for the scope of the decorated function. The span will automatically capture the input and output of the function. When it is applied to a method, it doesn’t capture the self argument. Any exception raised within the function will set the span status toERRORand detailed information such as exception message and stacktrace will be recorded to theattributesfield of the span.例如，下面的代码将生成一个名为
"my_function"的 span，捕获输入参数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 装饰器当前支持以下类型的函数：
Supported Function Types 函数类型
支持
同步
✅
异步
✅ (>= 2.16.0)
生成器
✅ (>= 2.20.2)
异步生成器
✅ (>= 2.20.2)
ClassMethod
✅ (>= 3.0.0)
StaticMethod
✅ (>= 3.0.0)
有关使用 @mlflow.trace 装饰器（包括流式/异步处理）的更多示例，请参阅 MLflow Tracing documentation。
提示
@mlflow.trace 装饰器在你想追踪自己定义的函数时很有用。但是，你也可能想追踪外部库中的函数。在这种情况下，你可以使用这个
mlflow.trace()函数来直接包裹该函数，而不是将其作为装饰器使用。这样会创建与装饰器创建的完全相同的 span，即捕获来自函数调用的信息。import math import mlflow mlflow.trace(math.factorial)(5)
- Parameters
func – 要被装饰的函数。当作为装饰器使用时，不得提供。
name – span 的名称。如果未提供，则使用函数的名称。
span_type – span 的类型。可以是字符串或
SpanType枚举值。attributes – 一个字典，用于设置 span 上的属性。
output_reducer – 一个将生成器函数的输出归约为单一值以设置为 span 输出的函数。
- mlflow.start_span(name: str = 'span', span_type: str | None = 'UNKNOWN', attributes: dict[str, typing.Any] | None = None) Generator[LiveSpan, None, None][source]
用于创建一个新的 span 并将其在上下文中设为当前 span 的上下文管理器。
This context manager automatically manages the span lifecycle and parent-child relationships. The span will be ended when the context manager exits. Any exception raised within the context manager will set the span status to
ERROR, and detailed information such as exception message and stacktrace will be recorded to theattributesfield of the span. New spans can be created within the context manager, then they will be assigned as child spans.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
当此上下文管理器在顶层作用域中使用时，即不在另一个 span 上下文内，该 span 将被视为根 span。根 span 没有父引用，且当根 span 结束时，整个 trace 将被记录。
提示
如果你想更明确地控制跟踪生命周期，可以使用mlflow.start_span_no_context() API。它提供了更底层的接口来启动 spans 并显式地控制父子关系。不过，通常建议只要该上下文管理器满足你的需求就使用它，因为它需要更少的样板代码且更不容易出错。
注意
上下文管理器默认不会在线程之间传播 span context。参见 Multi Threading 了解如何在线程之间传播 span context。
- Parameters
name – span 的名称。
span_type – span 的类型。可以是字符串，或者是一个
SpanType枚举值attributes – 一个字典，用于在 span 上设置属性。
- Returns
返回一个
mlflow.entities.Span，表示所创建的 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, experiment_id: Optional[str] = None, start_time_ns: Optional[int] = None) LiveSpan[source]
启动一个 span，而不将其附加到全局跟踪上下文。
当您想创建一个 span 而不自动将其与父 span 链接，而是手动管理父子关系时，这很有用。
通过此函数启动的 span 必须使用 span 对象的 end() 方法手动结束。
- Parameters
name – 该 span 的名称。
span_type – 该 span 的类型。可以是字符串或一个
SpanType枚举值parent_span – 要关联的父 span。如果 None，span 将被视为根 span。
inputs – 该 span 的输入数据。
attributes – 一个字典，用于在 span 上设置属性。
tags – 一个用于在跟踪上设置标签的字典。
experiment_id – 与该跟踪关联的实验 ID。如果未提供，将使用当前激活的实验。
start_time_ns – 跨度的开始时间（纳秒）。如果未提供，将使用当前时间。
- Returns
一个表示已创建的 span 的
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) Trace | None[source]
如果存在，则通过给定的请求 ID 获取跟踪。
该函数首先从内存缓冲区检索跟踪记录，如果不存在，它会从跟踪存储中获取该跟踪记录。如果在跟踪存储中找不到该跟踪记录，则返回 None。
- Parameters
trace_id – 跟踪的 ID。
silent – 如果为 True，当未找到 trace 时抑制警告信息。API 将返回 None 而不发出任何警告。默认值为 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)
- Returns
一个
mlflow.entities.Trace对象，具有给定的请求 ID。
- 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) 'pandas.DataFrame' | list[Trace][source]
返回在实验中与给定的搜索表达式列表匹配的跟踪。
注意
如果预计搜索结果数量很大，请考虑直接使用 MlflowClient.search_traces API 对结果进行分页。此函数将所有结果加载到内存中，可能不适合大规模结果集。
- Parameters
experiment_ids – 用于限定搜索范围的实验 ID 列表。如果未提供，搜索将针对当前活动实验进行。
filter_string – 一个搜索过滤字符串。
max_results – 希望返回的最大跟踪数。如果为 None，则返回匹配搜索表达式的所有跟踪。
order_by – order_by 子句的列表。
extract_fields –
指定要从 traces 提取的字段，使用以下格式
"span_name.[inputs|outputs].field_name"or"span_name.[inputs|outputs]".注意
仅当返回类型设置为 “pandas” 时才支持此参数。
例如，
"predict.outputs.result"从名为"predict"的 span 中检索输出"result"字段，而"predict.outputs"则获取整个 outputs 字典，包括键"result"和"explanation"。默认情况下，不会将任何字段提取到 DataFrame 列中。当指定多个字段时，每个字段都会作为单独的列提取。如果提供了无效的字段字符串，函数将静默返回且不会添加该字段的列。支持的字段仅限于 span 的
"inputs"和"outputs"。如果 span 名称或字段名包含点，则必须用反引号括起。例如：# span 名称包含点 extract_fields = ["`span.name`.inputs.field"] # 字段名包含点 extract_fields = ["span.inputs.`field.name`"] # span 名称和字段名都包含点 extract_fields = ["`span.name`.inputs.`field.name`"]
run_id – 用于限定搜索范围的 run id。当跟踪在一个活动运行下创建时，它会与该运行关联，你可以通过 run id 进行过滤以检索该跟踪。参见下面的示例，了解如何按 run id 过滤跟踪记录。
return_type –
返回值的类型。支持以下返回类型。如果已安装 pandas 库，默认返回类型为 “pandas”。否则，默认返回类型为 “list”。
- ”pandas”: 返回一个包含有关 traces 信息的 Pandas DataFrame
其中每一行表示单个 trace，每一列表示 trace 的一个字段，例如 trace_id、spans 等。
”list”: 返回一个
Trace对象的列表。
model_id – 如果指定，搜索与给定模型 ID 关联的跟踪。
sql_warehouse_id – 仅在 Databricks 中使用。用于在推理表中搜索跟踪记录的 SQL 仓库的 ID。
include_spans – 如果
True，则在返回的 traces 中包含 spans。否则，只返回 trace 元数据，例如 trace ID、开始时间、结束时间等，不包含任何 spans。默认值为True。
- Returns
满足搜索表达式的 Traces。可以作为一个
Trace对象列表或作为一个 Pandas DataFrame， 取决于 return_type 参数的值。
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[source]
获取全局上下文中当前活动的 span。
注意
只有当 span 使用类似 @mlflow.trace 或 with mlflow.start_span 的流式 API 创建时，此功能才有效。如果 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()
- Returns
当前活动的 span（如果存在），否则为 None。
- mlflow.get_last_active_trace_id(thread_local: bool = False) str | None[source]
如果存在，获取同一进程中最后的活动跟踪。
警告
此函数默认不是线程安全的，会返回同一进程中最后一个活动的 trace。如果想获取当前线程中最后一个活动的 trace，请将 thread_local 参数设置为 True。
- Parameters
thread_local – 如果为 True，则返回当前线程中最后活动的 trace。否则，返回同一进程中最后活动的 trace。默认值为 False。
- Returns
最后一个活动 trace 的 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)[source]
将已完成的 trace 对象添加到另一个 trace 中。
当您调用由 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)
如果你有一个特定的目标 span 用来将跟踪合并到其下，你可以传入该目标 span
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)
- Parameters
trace –
一个
Trace对象或 trace 的字典表示。该 trace 必须 已完成，即不应对其进行进一步更新。否则，此函数将引发异常。target –
要将给定 trace 合并到的目标 span。
如果提供，trace 将合并到目标 span 下。
如果未提供，trace 将合并到当前活动的 span 下。
如果未提供且没有活动的 span，将创建一个名为“Remote Trace <…>”的新 span，并将 trace 合并到该 span 下。
- mlflow.log_assessment(trace_id: str, assessment: Assessment) Assessment[source]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
将评估记录到 Trace。该评估可以是期望或反馈。
- 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)
下面的代码将人工提供的真实标签注释到 trace 上，并包含来源信息。当未提供来源时，默认来源被设置为 “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)
- The expectation value can be any JSON-serializable value. For example, you may
将完整的 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)
You can also log an error information during the feedback generation process. To do so, provide an instance of
AssessmentErrorto the error parameter, and leave the value parameter as 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[source]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
将一个期望（例如真实标签）记录到 Trace。此 API 仅接受关键字参数。
- Parameters
trace_id – 跟踪的 ID。
name – 期望评估的名称，例如 “expected_answer
value – 期望的值。它可以是任何可 JSON 序列化的值。
source – 该期望评估的来源。必须是
AssessmentSource的一个实例。如果未提供，默认为 CODE 源类型。metadata – 该期望的附加元数据。
span_id – 与期望关联的 span 的 ID，如果需要将其关联到跟踪中的特定 span。
- Returns
已创建的期望评估。
- Return type
示例
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 仅接受关键字参数。
- Parameters
trace_id – 跟踪的 ID。
name – 反馈评估的名称，例如 “faithfulness”。如果未提供，默认为 “feedback”。
value – 反馈的 value。必须是以下类型之一： - float - int - str - bool - list，由上述相同类型的值组成 - dict，键为 string，值为上述相同类型
source – 反馈评估的来源。必须是
AssessmentSource的一个实例。如果未提供，默认为 CODE 源类型。error – An error object representing any issues encountered while computing the feedback, e.g., a timeout error from an LLM judge. Accepts an exception object, or an
AssessmentErrorobject. Either this or value must be provided.rationale – 对该反馈的理由/正当性。
metadata – 用于反馈的附加元数据。
span_id – 与反馈关联的 span 的 ID，如果需要与跟踪中的特定 span 关联。
- Returns
已创建的反馈评估。
- Return type
示例
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, 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 中更新现有的 expectation（ground truth）。
- Parameters
trace_id – 跟踪的 ID。
assessment_id – 要更新的期望或反馈评估的 ID。
assessment – 更新后的 assessment.
- Returns
更新后的反馈或期望评估。
- Return type
示例：
下面的代码将现有的期望更新为新值。要更新其他字段，请提供相应的参数。
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]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
删除与跟踪相关联的评估。
- Parameters
trace_id – 跟踪的 ID。
assessment_id – 要删除的评估的 ID。
- mlflow.tracing.configure(span_processors: list[typing.Callable[[ForwardRef('LiveSpan')], NoneType]] | None = None) mlflow.tracing.config.TracingConfigContext[source]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
配置 MLflow 跟踪。可以作为函数或上下文管理器使用。
仅更新显式提供的参数，其余保持不变。
- Parameters
span_processors – 在导出之前处理 span 的函数列表。 这有助于从 span 中筛选/屏蔽特定属性以防止记录敏感数据或减少 span 的大小。 每个函数必须接受一个类型为 LiveSpan 的单个参数，并且不应返回任何值。当提供多个函数时，它们会按提供的顺序依次应用。
- Returns
- 用于临时配置更改的上下文管理器。
当作为函数使用时，配置更改会持续生效。 当作为上下文管理器使用时，退出时更改会被还原。
- Return type
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.disable()[source]
禁用跟踪。
注意
此函数将OpenTelemetry设置为使用NoOpTracerProvider，并实际禁用所有追踪操作。
示例：
import mlflow @mlflow.trace def f(): return 0 # Tracing is enabled by default f() assert len(mlflow.search_traces()) == 1 # Disable tracing mlflow.tracing.disable() f() assert len(mlflow.search_traces()) == 1
- mlflow.tracing.disable_notebook_display()[source]
禁用在笔记本输出单元格中显示 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()) == 1 # Disable tracing mlflow.tracing.disable() f() assert len(mlflow.search_traces()) == 1 # Re-enable tracing mlflow.tracing.enable() f() assert len(mlflow.search_traces()) == 2
- mlflow.tracing.enable_notebook_display()[source]
在笔记本输出单元中启用 MLflow Trace UI。该显示默认开启，当执行以下任一操作时将显示 Trace UI：
在跟踪完成时（即每当导出跟踪时）
当调用
mlflow.search_traces()链式 API 时在调用
mlflow.client.MlflowClient.get_trace()或mlflow.client.MlflowClient.search_traces()客户端 API 时
- mlflow.tracing.reset()[source]
重置指示 MLflow 跟踪器提供程序是否已初始化的标志。这可确保在下一次执行跟踪操作时重新初始化跟踪器提供程序。
- mlflow.tracing.set_destination(destination: mlflow.tracing.destination.TraceDestination)[source]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
设置一个自定义的 span 目的地，MLflow 将向其导出跟踪。
由此函数指定的目标将优先于其他配置，例如 tracking URI、OTLP environment variables。
默认情况下，指定的目标是全局生效的。要在多线程应用中为每个线程设置不同的目标，请将环境变量 ‘MLFLOW_ENABLE_THREAD_LOCAL_TRACING_DESTINATION’ 设置为 ‘true’,
要重置目标，请调用
mlflow.tracing.reset()函数。- Parameters
destination – 一个
TraceDestination对象，用于指定跟踪数据的目的地。
示例
import mlflow from mlflow.tracing.destination import MlflowExperiment # Setting the destination to an MLflow experiment with ID "123" mlflow.tracing.set_destination(MlflowExperiment(experiment_id="123")) # Reset the destination (to an active experiment as default) mlflow.tracing.reset()
- mlflow.tracing.set_span_chat_tools(span: LiveSpan, tools: list[ChatTool])[source]
在指定的 span 上设置 mlflow.chat.tools 属性。该属性在用户界面中使用，也被消费跟踪数据的下游应用程序使用，例如 MLflow evaluate。
- Parameters
span – 要向其添加属性的 LiveSpan
tools – 一个标准化聊天工具定义的列表（详情请参考spec）
示例：
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 已记录模型 APIs
mlflow 模块提供一组高级 APIs，用于与 MLflow Logged Models 交互。
- 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，其 artifacts 存储在 MLflow 之外。这对于跟踪未使用 MLflow Model 格式打包的模型、应用或生成式 AI 智能体的参数和性能数据（指标、追踪等）非常有用。
- Parameters
name – 模型的名称。如果未指定，将生成一个随机名称。
source_run_id – 模型所属运行的 ID。如果未指定且当前有一个活动运行，则将使用该活动运行的 ID。
tags – 一个字典，键和值均为字符串，用于在模型上设置 tags。
params – 一个包含字符串键和值的字典，用于在模型上设置参数。
model_type – 模型的类型。这是一个用户定义的字符串，可用于搜索和比较相关模型。例如，设置
model_type="agent"可以让你轻松搜索此模型，并在将来将其与其他类型为"agent"的模型进行比较。experiment_id – 模型所属实验的实验ID。
- Returns
一个新的
mlflow.entities.LoggedModel对象，状态为READY。
- mlflow.delete_logged_model_tag(model_id: str, key: str) None[source]
从指定的已记录模型中删除一个标签。
- Parameters
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]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
通过更新模型的状态来完成模型的最终处理。
- Parameters
model_id – 要完成的模型的 ID。
status – 要在模型上设置的最终状态。
- Returns
更新后的模型。
示例：
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 获取已记录的模型。
- Parameters
model_id – 已记录模型的 ID。
- Returns
已记录的模型。
示例：
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状态，例如通过调用特定 flavor 的log_model()方法例如mlflow.pyfunc.log_model()。- Parameters
name – 模型的名称。如果未指定，将生成一个随机名称。
source_run_id – 模型所关联运行的 ID。如果未指定且有一个运行处于活动状态，将使用该活动运行的 ID。
tags – 一个用于在模型上设置标签的字符串键和值组成的字典。
params – 一个键和值为字符串的字典，用于在模型上设置参数。
model_type – 模型的类型。
experiment_id – 模型所属实验的 ID。
- Returns
一个新的
mlflow.entities.LoggedModel对象，状态为PENDING。
- mlflow.last_logged_model() LoggedModel | None[source]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
获取当前会话中最近记录的模型。 如果没有记录任何模型，则返回 None。
- Returns
已记录的模型。
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]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
搜索符合指定搜索条件的已记录模型。
- Parameters
experiment_ids – 要搜索已记录模型的实验 ID 列表。如果未指定， 将使用活动实验。
filter_string –
一个类似 SQL 的过滤字符串，用于解析。过滤字符串语法支持：
- 实体说明：
attributes: attribute_name（如果未指定前缀，默认）
metrics: metrics.metric_name
parameters: params.param_name
tags: tags.tag_name
- 比较运算符：
对于数值实体（metrics 和数值属性）： <, <=, >, >=, =, !=
对于字符串实体（params、tags 与字符串属性）： =, !=, 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 上 accuracy > 0.9 的所有已记录的模型。任何匹配条件的数据集的指标值都会被考虑。 如果未指定数据集，则筛选器将考虑所有数据集的指标。支持以下字段：
- 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’。
- Returns
以指定的输出格式显示搜索结果。
示例：
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 设置活动模型，该模型将用于链接在模型生命周期中生成的跟踪。返回值可以在
with块中用作上下文管理器；否则，您必须调用set_active_model()来更新活动模型。- Parameters
name – 要设置为活动状态的
mlflow.entities.LoggedModel的 name。如果不存在具有该 name 的 LoggedModel，它将在当前实验下创建。如果存在多个具有该 name 的 LoggedModel，将把最新的设为活动状态。model_id – 要设置为活动的
mlflow.entities.LoggedModel的 ID。如果不存在具有该 ID 的 LoggedModel，将引发异常。
- Returns
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") assert len(traces) == 1
- mlflow.set_logged_model_tags(model_id: str, tags: dict[str, typing.Any]) None[source]
在指定的已记录模型上设置标签。
- Parameters
model_id – 模型的 ID。
tags – 在模型上设置的标签。
- Returns
无
示例：
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]
注意
实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。
将参数记录到指定的已记录模型中。
- Parameters
params – 要记录在模型上的 Params。
model_id – 模型的 ID。如果未指定，则使用当前激活的模型 ID。
- Returns
无
示例：
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)