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 块的末尾自动终止运行。
流畅的跟踪 API 目前不是线程安全的。跟踪 API 的任何并发调用者都必须手动实现互斥。
有关更低级别的 API，请参阅 mlflow.client 模块。
- class mlflow.ActiveModel(logged_model: LoggedModel, set_by_user: bool)[source]
mlflow.entities.LoggedModel的包装器，以支持 Python 的with语法。
- class mlflow.ActiveRun(run)[source]
mlflow.entities.Run的包装器，以支持 Python 的with语法。
- 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。
- 参数
image – 图像可以是 numpy 数组、PIL 图像或指向图像的路径。
import mlflow import numpy as np from PIL import Image # Create an image as a numpy array image = np.zeros((100, 100, 3), dtype=np.uint8) image[:, :50] = [255, 128, 0] # Create an Image object image_obj = mlflow.Image(image) # Convert the Image object to a list of pixel values pixel_values = image_obj.to_list()
- resize(size: tuple[int, int])[source]
将图像大小调整为指定大小。
- 参数
size – 要调整大小的尺寸。
- 返回
调整大小后的图像对象的副本。
- save(path: str)[source]
将图像保存到文件。
- 参数
path – 保存图像的文件路径。
- to_array()[source]
将图像转换为 numpy 数组。
- 返回
像素值的 Numpy 数组。
- to_list()[source]
将图像转换为像素值列表。
- 返回
像素值列表。
- to_pil()[source]
将图像转换为 PIL 图像。
- 返回
PIL 图像。
- exception mlflow.MlflowException(message, error_code=1, **kwargs)[source]
用于暴露面向外部操作失败信息的通用异常。与此异常相关的错误消息可能会暴露给客户端的 HTTP 响应中，用于调试目的。如果错误文本是敏感信息，请改用通用的 Exception 对象。
- get_http_status_code()[source]
- classmethod invalid_parameter_value(message, **kwargs)[source]
使用 INVALID_PARAMETER_VALUE 错误代码构造一个 MlflowException 对象。
- 参数
message – 描述发生的错误的字符串。这将被包含在异常的序列化 JSON 表示中。
kwargs – 要包含在 MlflowException 的序列化 JSON 表示中的其他键值对。
- serialize_as_json()[source]
- mlflow.active_run() ActiveRun | None[source]
获取当前活动的
Run，如果不存在则返回 None。注意
此 API 是 **线程本地** 的，仅返回当前线程中的活动运行。如果您的应用程序是多线程的，并且一个运行在不同的线程中启动，此 API 将不会检索该运行。
注意：您无法通过 `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]
启用（或禁用）并配置所有受支持的集成以进行自动日志记录。
参数将传递给所有支持它们的自动日志记录集成。
请参阅 跟踪文档 以获取支持的自动日志记录集成列表。
请注意，在任何时候设置的特定于框架的配置将优先于通过此函数设置的任何配置。例如：
import mlflow mlflow.autolog(log_models=False, exclusive=True) import sklearn
将为 sklearn 启用自动日志记录，并设置 log_models=False 和 exclusive=True，但
import mlflow mlflow.autolog(log_models=False, exclusive=True) import sklearn mlflow.sklearn.autolog(log_models=True)
将为 sklearn 启用自动日志记录，并设置 log_models=True 和 exclusive=False，后者是由于 mlflow.sklearn.autolog 中 exclusive 的默认值；其他框架自动日志函数（例如 mlflow.tensorflow.autolog）将使用 mlflow.autolog 设置的配置（在本例中为 log_models=False，exclusive=True），直到它们被用户明确调用为止。
- 参数
log_input_examples – 如果为
True，则在训练期间收集并与模型工件一起记录训练数据集的输入示例。如果为False，则不记录输入示例。注意：输入示例是 MLflow 模型属性，只有在log_models也为True时才收集。log_model_signatures – 如果为
True，则在训练期间收集并与模型工件一起记录描述模型输入和输出的ModelSignatures。如果为False，则不记录签名。注意：模型签名是 MLflow 模型属性，只有在log_models也为True时才收集。log_models – 如果为
True，则训练好的模型将作为 MLflow 模型工件进行记录。如果为False，则不记录训练好的模型。输入样本和模型签名（MLflow 模型的属性）在log_models为False时也会被省略。log_datasets – 如果为
True，则数据集信息将被记录到 MLflow Tracking。如果为False，则不记录数据集信息。log_traces – 如果为
True，则为集成收集跟踪。如果为False，则不收集任何跟踪。disable – 如果为
True，则禁用所有受支持的自动日志记录集成。如果为False，则启用所有受支持的自动日志记录集成。exclusive – 如果为
True，则自动记录的内容不会记录到用户创建的流畅运行中。如果为False，则自动记录的内容将记录到活动的流畅运行中，该运行可能是用户创建的。disable_for_unsupported_versions – 如果为
True，则禁用未经 MLflow 客户端此版本测试或不兼容的所有集成库版本的自动日志记录。silent – 如果为
True，则在自动日志记录设置和训练执行期间抑制 MLflow 的所有事件日志和警告。如果为False，则在自动日志记录设置和训练执行期间显示所有事件和警告。extra_tags – 要为自动日志记录创建的每个托管运行设置的额外标签的字典。
exclude_flavors – 要从自动日志记录中排除的风味名称列表。例如：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]
创建一个实验。
- 参数
name – 实验名称，必须是一个非空且唯一的字符串。
artifact_location – 存储运行工件的位置。如果未提供，服务器将选择合适的默认值。
tags – 一个可选的字典，包含字符串键值对，将作为标签设置在实验上。
- 返回
创建的实验的字符串 ID。
import mlflow from pathlib import Path # Create an experiment name, which must be unique and case sensitive experiment_id = mlflow.create_experiment( "Social NLP Experiments", artifact_location=Path.cwd().joinpath("mlruns").as_uri(), tags={"version": "v1", "priority": "P1"}, ) experiment = mlflow.get_experiment(experiment_id) print(f"Name: {experiment.name}") print(f"Experiment_id: {experiment.experiment_id}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Tags: {experiment.tags}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}") print(f"Creation timestamp: {experiment.creation_time}")
- mlflow.delete_experiment(experiment_id: str) None[source]
从后端存储中删除一个实验。
- 参数
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]
删除当前实验的标签。
- 参数
key – 要删除的标签的名称。
- mlflow.delete_run(run_id: str) None[source]
删除具有给定 ID 的运行。
- 参数
run_id – 要删除的运行的唯一标识符。
import mlflow with mlflow.start_run() as run: mlflow.log_param("p", 0) run_id = run.info.run_id mlflow.delete_run(run_id) lifecycle_stage = mlflow.get_run(run_id).info.lifecycle_stage print(f"run_id: {run_id}; lifecycle_stage: {lifecycle_stage}")
- mlflow.delete_tag(key: str) None[source]
从运行中删除标签。此操作不可逆。如果当前没有活动运行，此方法将创建一个新的活动运行。
- 参数
key – 标签的名称
- mlflow.delete_trace_tag(trace_id: str, key: str) None[source]
注意
参数
request_id已弃用。请使用trace_id。删除具有给定 trace_id 的跟踪上的标签。
跟踪可以是活动的，也可以是已结束并记录在后端中的。下面是一个删除活动跟踪上的标签的示例。您可以替换
trace_id参数来删除已结束跟踪上的标签。import mlflow with mlflow.start_span("my_span") as span: mlflow.set_trace_tag(span.trace_id, "key", "value") mlflow.delete_trace_tag(span.trace_id, "key")
- 参数
trace_id – 要从中删除标签的跟踪的 ID。
key – 标签的字符串键。长度最多为 250 个字符，否则在存储时将被截断。
- mlflow.disable_system_metrics_logging()[source]
全局禁用系统指标日志记录。
调用此函数将全局禁用系统指标日志记录，但用户仍可通过 mlflow.start_run(log_system_metrics=True) 为单个运行选择加入系统指标日志记录。
- mlflow.doctor(mask_envs=False)[source]
打印有用的信息，用于调试 MLflow 的问题。
- 参数
mask_envs – 如果为 True，则在输出中屏蔽 MLflow 环境变量的值（例如 “MLFLOW_ENV_VAR”: “***”），以防止泄露敏感信息。
警告
此 API 应仅用于调试目的。
输出可能包含敏感信息，例如包含密码的数据库 URI。
System information: Linux #58~20.04.1-Ubuntu SMP Thu Oct 13 13:09:46 UTC 2022 Python version: 3.8.13 MLflow version: 2.0.1 MLflow module location: /usr/local/lib/python3.8/site-packages/mlflow/__init__.py Tracking URI: sqlite:///mlflow.db Registry URI: sqlite:///mlflow.db MLflow environment variables: MLFLOW_TRACKING_URI: sqlite:///mlflow.db MLflow dependencies: Flask: 2.2.2 Jinja2: 3.0.3 alembic: 1.8.1 click: 8.1.3 cloudpickle: 2.2.0 databricks-cli: 0.17.4.dev0 docker: 6.0.0 entrypoints: 0.4 gitpython: 3.1.29 gunicorn: 20.1.0 importlib-metadata: 5.0.0 markdown: 3.4.1 matplotlib: 3.6.1 numpy: 1.23.4 packaging: 21.3 pandas: 1.5.1 protobuf: 3.19.6 pyarrow: 9.0.0 pytz: 2022.6 pyyaml: 6.0 querystring-parser: 1.2.4 requests: 2.28.1 scikit-learn: 1.1.3 scipy: 1.9.3 shap: 0.41.0 sqlalchemy: 1.4.42 sqlparse: 0.4.3
- mlflow.enable_system_metrics_logging()[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中进行评估。有关详细信息，请阅读 模型评估文档。- 默认评估器行为
默认评估器，可以通过
evaluators="default"或evaluators=None调用，支持的模型类型如下。对于每种预定义的模型类型，默认评估器都会在选定的指标集上评估您的模型并生成图表等构件。请在下方查找更多详细信息。对于
"regressor"和"classifier"模型类型，默认评估器都将使用 SHAP 生成模型摘要图和特征重要性图。- 对于回归模型，默认评估器还会另外记录
指标：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。
- 对于二元分类器，默认评估器还会另外记录
指标：true_negatives, false_positives, false_negatives, true_positives, recall, precision, f1_score, accuracy_score, example_count, log_loss, roc_auc, precision_recall_auc。
构件：提升曲线图、精确率-召回率图、ROC 图。
- 对于多类别分类器，默认评估器还会另外记录
指标：accuracy_score, example_count, f1_score_micro, f1_score_macro, log_loss
构件：一个用于“每类指标”（每类指标包括 true_negatives/false_positives/false_negatives/true_positives/recall/precision/roc_auc, precision_recall_auc）的 CSV 文件，精确率-召回率合并曲线图，ROC 合并曲线图。
- 对于问答模型，默认评估器会记录
指标：
exact_match、token_count、toxicity（需要 evaluate、torch）、flesch_kincaid_grade_level（需要 textstat）和 ari_grade_level。构件：一个 JSON 文件，其中包含模型在表格格式下的输入、输出、目标（如果提供了
targets参数）和每行指标。
- 对于文本摘要模型，默认评估器会记录
指标：
token_count、ROUGE（需要安装 evaluate、nltk 和 rouge_score）、toxicity（需要 evaluate、torch、transformers）、ari_grade_level（需要 textstat）、flesch_kincaid_grade_level（需要 textstat）。构件：一个 JSON 文件，其中包含模型在表格格式下的输入、输出、目标（如果提供了
targets参数）和每行指标。
- 对于文本模型，默认评估器会记录
指标：
token_count、toxicity（需要 evaluate、torch、transformers）、ari_grade_level（需要 textstat）、flesch_kincaid_grade_level（需要 textstat）。构件：一个 JSON 文件，其中包含模型在表格格式下的输入、输出、目标（如果提供了
targets参数）和每行指标。
- 对于检索器模型，默认评估器会记录
指标：
precision_at_k(k)、recall_at_k(k)和ndcg_at_k(k)- 这些指标的默认值为retriever_k= 3。构件：一个 JSON 文件，其中包含模型在表格格式下的输入、输出、目标和每行指标。
对于 sklearn 模型，默认评估器还会额外记录通过 model.score 方法计算出的模型评估标准（例如，分类器的平均准确率）。
上面列出的指标/构件会记录到当前活动的 MLflow 运行中。如果不存在活动的运行，则会创建一个新的 MLflow 运行来记录这些指标和构件。
此外，有关指定数据集的信息（哈希、名称（如果已指定）、路径（如果已指定）以及评估它的模型的 UUID）将被记录到
mlflow.datasets标签中。- 默认评估器可用的
evaluator_config选项包括 log_model_explainability：一个布尔值，指定是否记录模型可解释性见解，默认值为 True。
- log_explainer：如果为 True，则记录用于计算模型可解释性的解释器
见解，并将其作为模型。默认值为 False。
explainability_algorithm：一个字符串，用于指定模型可解释性的 SHAP Explainer 算法。支持的算法包括：“exact”、“permutation”、“partition”、“kernel”。如果未设置，则使用
shap.Explainer和“auto”算法，它会根据模型选择最佳的 Explainer。explainability_nsamples：用于计算模型可解释性见解的样本行数。默认值为 2000。
explainability_kernel_link：shap kernel explainer 使用的 kernel 链接函数。可用值为“identity”和“logit”。默认值为“identity”。
max_classes_for_multiclass_roc_pr：对于多类别分类任务，记录每类 ROC 曲线和精确率-召回率曲线的最大类别数。如果类别数大于配置的最大值，则不记录这些曲线。
metric_prefix：一个可选前缀，用于添加到评估期间生成的每个指标和构件的名称之前。
log_metrics_with_dataset_info：一个布尔值，指定在评估期间记录到 MLflow Tracking 的每个指标的名称中是否包含评估数据集的信息，默认值为 True。
pos_label：如果指定，则为二元分类模型计算精确率、召回率、F1 等分类指标时使用的正类标签。对于多类别分类和回归模型，将忽略此参数。
average：用于计算多类别分类模型（默认值：
'weighted'）的精确率、召回率、F1 等分类指标的平均方法。对于二元分类和回归模型，将忽略此参数。sample_weights：在计算模型性能指标时应用的每个样本的权重。
col_mapping：一个字典，用于将输入数据集或输出预测中的列名映射到调用评估函数时使用的列名。
retriever_k：当
model_type="retriever"时使用的参数，表示用于计算内置指标precision_at_k(k)、recall_at_k(k)和ndcg_at_k(k)的前 k 个检索文档的数量。默认值为 3。对于所有其他模型类型，将忽略此参数。
- 默认评估器可用的
- 评估数据集的局限性
对于分类任务，数据集标签用于推断总类别数。
对于二元分类任务，负类标签值必须是 0、-1 或 False，正类标签值必须是 1 或 True。
- 指标/构件计算的局限性
对于分类任务，某些指标和构件的计算需要模型输出类别概率。目前，对于 scikit-learn 模型，默认评估器会调用底层模型的
predict_proba方法来获取概率。对于其他模型类型，默认评估器不计算需要概率输出的指标/构件。
- 默认评估器记录模型可解释性见解的局限性
shap.Explainer的auto算法使用Linear解释器处理线性模型，使用Tree解释器处理树模型。由于 SHAP 的Linear和Tree解释器不支持多类别分类，因此默认评估器会回退到使用Exact或Permutation解释器处理多类别分类任务。目前不支持为 PySpark 模型记录模型可解释性见解。
评估数据集的标签值必须是数字或布尔值，所有特征值必须是数字，并且每个特征列只能包含标量值。
- 启用环境恢复时的局限性
当为已评估模型启用环境恢复时（即，指定了一个非本地的
env_manager），模型将作为客户端加载，该客户端会在独立的 Python 环境中调用 MLflow 模型评分服务器进程，并安装模型训练时的依赖项。因此，模型的方法如predict_proba（用于概率输出）或score（用于 sklearn 模型计算评估标准）将无法访问，并且默认评估器不会计算需要这些方法的指标或构件。由于模型是 MLflow 模型服务器进程，因此 SHAP 解释的计算速度会变慢。因此，当指定非本地
env_manager时，模型可解释性将被禁用，除非evaluator_config选项 **log_model_explainability** 被显式设置为True。
- 参数
model –
可选。如果指定，它应该是以下之一
一个 pyfunc 模型实例
指向 pyfunc 模型的 URI
指向 MLflow Deployments 端点的 URI，例如
"endpoints:/my-chat"一个可调用函数：该函数应能够接受模型输入并返回预测。它应遵循
predict方法的签名。以下是一个有效函数的示例model = mlflow.pyfunc.load_model(model_uri) def fn(model_input): return model.predict(model_input)
如果省略，则表示将使用静态数据集进行评估，而不是模型。在这种情况下，
data参数必须是包含模型输出的 Pandas DataFrame 或 mlflow PandasDataset。并且predictions参数必须是data中包含模型输出的列的名称。data –
以下之一
一个 numpy 数组或列表，包含评估特征，不包括标签。
- 一个 Pandas DataFrame，包含评估特征、标签，以及可选的模型
输出。当模型未指定时，必须提供模型输出。如果未指定
feature_names参数，则除标签列和预测列之外的所有列都被视为特征列。否则，只有feature_names中存在的列名才被视为特征列。
- 一个 Spark DataFrame，包含评估特征和标签。如果
未指定
feature_names参数，则除标签列之外的所有列都被视为特征列。否则，只有feature_names中存在的列名才被视为特征列。Spark DataFrame 中只有前 10000 行将用作评估数据。
- 一个
mlflow.data.dataset.Dataset实例，包含评估 特征、标签，以及可选的模型输出。模型输出仅支持 PandasDataset。当模型未指定时，必须提供模型输出，并且应通过 PandasDataset 的
predictions属性来指定。
- 一个
model_type –
(可选) 一个描述模型类型的字符串。默认评估器支持以下模型类型
'classifier''regressor''question-answering''text-summarization''text''retriever'
如果未指定
model_type，则必须通过extra_metrics参数提供要计算的指标列表。注意
'question-answering'、'text-summarization'、'text'和'retriever'是实验性的，可能会在未来的版本中更改或删除。targets – 如果
data是 numpy 数组或列表，则为包含评估标签的 numpy 数组或列表。如果data是 DataFrame，则为data中包含评估标签的列的字符串名称。对于分类器和回归器模型是必需的，但对于问答、文本摘要和文本模型是可选的。如果data是定义了目标的mlflow.data.dataset.Dataset，则targets是可选的。predictions –
可选。包含模型输出的列的名称。
当
model指定且输出多个列时，predictions可用于指定将用于存储模型输出以进行评估的列的名称。当
model未指定且data是 pandas DataFrame 时，predictions可用于指定data中包含模型输出的列的名称。
# Evaluate a model that outputs multiple columns data = pd.DataFrame({"question": ["foo"]}) def model(inputs): return pd.DataFrame({"answer": ["bar"], "source": ["baz"]}) results = evaluate( model=model, data=data, predictions="answer", # other arguments if needed ) # Evaluate a static dataset data = pd.DataFrame({"question": ["foo"], "answer": ["bar"], "source": ["baz"]}) results = evaluate( data=data, predictions="answer", # other arguments if needed )
dataset_path – (可选) 数据存储的路径。不得包含双引号（
"）。如果指定，路径将记录到mlflow.datasets标签中，用于谱系跟踪。feature_names – (可选) 列表。如果
data参数是 numpy 数组或列表，则feature_names是每个特征的特征名称列表。如果feature_names=None，则使用feature_{feature_index}的格式生成feature_names。如果data参数是 Pandas DataFrame 或 Spark DataFrame，则feature_names是 DataFrame 中特征列名称的列表。如果feature_names=None，则除标签列和预测列外的所有列都将被视为特征列。evaluators – 用于模型评估的评估器名称，或评估器名称列表。如果未指定，则使用所有能够根据指定数据集评估指定模型的评估器。默认评估器可以通过名称
"default"来引用。要查看所有可用的评估器，请调用mlflow.models.list_evaluators()。evaluator_config – 要提供给评估器的附加配置的字典。如果指定了多个评估器，则每个配置应作为嵌套字典提供，其键是评估器的名称。
extra_metrics –
(可选)
EvaluationMetric对象列表。这些指标将与预定义的 model_type 的默认指标一起计算，并且设置 model_type=None 将仅计算 extra_metrics 中指定的指标。有关内置指标以及如何定义额外指标的信息，请参阅 mlflow.metrics 模块。import mlflow import numpy as np def root_mean_squared_error(eval_df, _builtin_metrics): return np.sqrt((np.abs(eval_df["prediction"] - eval_df["target"]) ** 2).mean()) rmse_metric = mlflow.models.make_metric( eval_fn=root_mean_squared_error, greater_is_better=False, ) mlflow.evaluate(..., extra_metrics=[rmse_metric])
custom_artifacts –
(可选) 自定义构件函数的列表，签名如下
def custom_artifact( eval_df: Union[pandas.Dataframe, pyspark.sql.DataFrame], builtin_metrics: Dict[str, float], artifacts_dir: str, ) -> Dict[str, Any]: """ Args: eval_df: A Pandas or Spark DataFrame containing ``prediction`` and ``target`` column. The ``prediction`` column contains the predictions made by the model. The ``target`` column contains the corresponding labels to the predictions made on that row. builtin_metrics: A dictionary containing the metrics calculated by the default evaluator. The keys are the names of the metrics and the values are the scalar values of the metrics. Refer to the DefaultEvaluator behavior section for what metrics will be returned based on the type of model (i.e. classifier or regressor). artifacts_dir: A temporary directory path that can be used by the custom artifacts function to temporarily store produced artifacts. The directory will be deleted after the artifacts are logged. Returns: A dictionary that maps artifact names to artifact objects (e.g. a Matplotlib Figure) or to artifact paths within ``artifacts_dir``. """ ...
构件可以表示的对象类型
表示构件文件路径的字符串 URI。MLflow 将根据文件扩展名推断构件的类型。
JSON 对象的字符串表示。这将保存为 .json 构件。
Pandas DataFrame。这将解析为 CSV 构件。
Numpy 数组。这将保存为 .npy 构件。
Matplotlib Figure。这将保存为图像构件。请注意，
matplotlib.pyplot.savefig在后台以默认配置调用。要进行自定义，请保存具有所需配置的图形并返回其文件路径，或在matplotlib.rcParams中通过环境变量定义自定义项。其他对象将尝试使用默认协议进行 pickling。
import mlflow import matplotlib.pyplot as plt def scatter_plot(eval_df, builtin_metrics, artifacts_dir): plt.scatter(eval_df["prediction"], eval_df["target"]) plt.xlabel("Targets") plt.ylabel("Predictions") plt.title("Targets vs. Predictions") plt.savefig(os.path.join(artifacts_dir, "example.png")) plt.close() return {"pred_target_scatter": os.path.join(artifacts_dir, "example.png")} def pred_sample(eval_df, _builtin_metrics, _artifacts_dir): return {"pred_sample": pred_sample.head(10)} mlflow.evaluate(..., custom_artifacts=[scatter_plot, pred_sample])
env_manager –
指定一个环境管理器，用于在隔离的 Python 环境中加载候选
model并恢复其依赖项。默认值为local，支持以下值virtualenv: (推荐) 使用 virtualenv 恢复用于训练模型的 Python 环境。conda: 使用 Conda 恢复训练模型时使用的软件环境。local: 使用当前 Python 环境进行模型推理，这可能与训练模型时使用的环境不同，并可能导致错误或无效的预测。
model_config – 用于使用 pyfunc 加载模型的模型配置。检查模型的 pyfunc 风格以了解适用于您的特定模型的键。如果未指定，则使用模型（如果存在）的默认模型配置。
inference_params – (可选) 在进行预测时传递给模型的推理参数字典，例如
{"max_tokens": 100}。仅当model是 MLflow Deployments 端点 URI（例如"endpoints:/my-chat"）时使用。model_id – (可选) MLflow LoggedModel 或 Model Version 的 ID，评估结果（例如指标和跟踪）将链接到该 ID。如果未指定 model_id 但指定了 model，则将使用 model 中的 ID。
_called_from_genai_evaluate – (可选) 仅供内部使用。
- 返回
一个
mlflow.models.EvaluationResult实例，包含使用给定数据集评估模型的指标。
- mlflow.flush_artifact_async_logging() None[source]
刷新所有待处理的构件异步日志记录。
- mlflow.flush_async_logging() None[source]
刷新所有待处理的异步日志记录。
- mlflow.flush_trace_async_logging(terminate=False) None[source]
刷新所有待处理的跟踪异步日志记录。
- 参数
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。- 返回
活动模型 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()
- 返回
当前活动跟踪的 ID（如果存在），否则为 None。
- mlflow.get_artifact_uri(artifact_path: Optional[str] = None) str[source]
获取当前活动运行中指定构件的绝对 URI。
如果未指定 path，则将返回当前活动运行的构件根 URI；调用
log_artifact和log_artifacts会将构件写入构件根 URI 的子目录。如果不存在活动的运行，此方法将创建一个新的活动运行。
- 参数
artifact_path – 要获取绝对 URI 的运行相关构件路径。例如，“path/to/artifact”。如果未指定，则返回当前活动运行的构件根 URI。
- 返回
一个指向指定构件或当前活动运行的构件根的*绝对* URI。例如，如果提供了构件路径且当前活动运行使用基于 S3 的存储，则这可能是一个格式为
s3://<bucket_name>/path/to/artifact/root/path/to/artifact的 URI。如果未提供构件路径且当前活动运行使用基于 S3 的存储，则这可能是一个格式为s3://<bucket_name>/path/to/artifact/root的 URI。
import tempfile import mlflow features = "rooms, zipcode, median_price, school_rating, transport" with tempfile.NamedTemporaryFile("w") as tmp_file: tmp_file.write(features) tmp_file.flush() # Log the artifact in a directory "features" under the root artifact_uri/features with mlflow.start_run(): mlflow.log_artifact(tmp_file.name, artifact_path="features") # Fetch the artifact uri root directory artifact_uri = mlflow.get_artifact_uri() print(f"Artifact uri: {artifact_uri}") # Fetch a specific artifact uri artifact_uri = mlflow.get_artifact_uri(artifact_path="features/features.txt") print(f"Artifact uri: {artifact_uri}")
- mlflow.get_assessment(trace_id: str, assessment_id: str) Assessment[source]
从后端存储获取评估实体。
- 参数
trace_id – 跟踪的 ID。
assessment_id – 要获取的评估的 ID。
- 返回
Assessment 对象。
- 返回类型
- mlflow.get_experiment(experiment_id: str) Experiment[source]
从后端存储检索指定 experiment_id 的实验。
- 参数
experiment_id – 从
create_experiment返回的字符串化实验 ID。- 返回
import mlflow experiment = mlflow.get_experiment("0") print(f"Name: {experiment.name}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Tags: {experiment.tags}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}") print(f"Creation timestamp: {experiment.creation_time}")
- mlflow.get_experiment_by_name(name: str) Experiment | None[source]
从后端存储中按实验名称检索实验。
- 参数
name – 区分大小写的实验名称。
- 返回
如果存在指定名称的实验，则返回
mlflow.entities.Experiment对象，否则返回 None。
import mlflow # Case sensitive name experiment = mlflow.get_experiment_by_name("Default") print(f"Experiment_id: {experiment.experiment_id}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Tags: {experiment.tags}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}") print(f"Creation timestamp: {experiment.creation_time}")
- mlflow.get_parent_run(run_id: str) Run | None[source]
获取给定运行 ID 的父运行（如果存在）。
- 参数
run_id – 子运行的唯一标识符。
- 返回
如果父运行存在，则返回一个
mlflow.entities.Run对象。否则，返回 None。
import mlflow # Create nested runs with mlflow.start_run(): with mlflow.start_run(nested=True) as child_run: child_run_id = child_run.info.run_id parent_run = mlflow.get_parent_run(child_run_id) print(f"child_run_id: {child_run_id}") print(f"parent_run_id: {parent_run.info.run_id}")
- mlflow.get_registry_uri() str[source]
获取当前的注册表 URI。如果未指定，则默认为跟踪 URI。
- 返回
注册表 URI。
# Get the current model registry uri mr_uri = mlflow.get_registry_uri() print(f"Current model registry uri: {mr_uri}") # Get the current tracking uri tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}") # They should be the same assert mr_uri == tracking_uri
Current model registry uri: file:///.../mlruns Current tracking uri: file:///.../mlruns
- mlflow.get_run(run_id: str) Run[source]
从后端存储中获取运行。结果中的 Run 包含运行元数据的集合——RunInfo 以及运行参数、标签和指标的集合——RunData。它还包含运行输入（实验性）的集合，包括运行使用的数据集的信息——RunInputs。如果为该运行记录了多个具有相同键的指标，则 RunData 包含每个指标在最大步骤处最近记录的值。
- 参数
run_id – 运行的唯一标识符。
- 返回
如果运行存在，则返回一个 Run 对象。否则，将引发异常。
import mlflow with mlflow.start_run() as run: mlflow.log_param("p", 0) run_id = run.info.run_id print( f"run_id: {run_id}; lifecycle_stage: {mlflow.get_run(run_id).info.lifecycle_stage}" )
- mlflow.get_tracking_uri() str[source]
获取当前的跟踪 URI。这可能与当前活动运行的跟踪 URI 不符，因为跟踪 URI 可以通过
set_tracking_uri更新。- 返回
跟踪 URI。
import mlflow # Get the current tracking uri tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}")
Current tracking uri: sqlite:///mlflow.db
- mlflow.is_tracking_uri_set()[source]
如果已设置跟踪 URI，则返回 True，否则返回 False。
- mlflow.last_active_run() Run | None[source]
获取最近一次活动的运行。
示例
import mlflow from sklearn.model_selection import train_test_split from sklearn.datasets import load_diabetes from sklearn.ensemble import RandomForestRegressor mlflow.autolog() db = load_diabetes() X_train, X_test, y_train, y_test = train_test_split(db.data, db.target) # Create and train models. rf = RandomForestRegressor(n_estimators=100, max_depth=6, max_features=3) rf.fit(X_train, y_train) # Use the model to make predictions on the test dataset. predictions = rf.predict(X_test) autolog_run = mlflow.last_active_run()
- 返回
如果存在活动运行，则返回该运行（这等同于
mlflow.active_run()）。否则，返回当前 Python 进程开始的、达到终止状态（即 FINISHED、FAILED 或 KILLED）的最后一次运行。
- mlflow.load_table(artifact_file: str, run_ids: list[str] | None = None, extra_columns: list[str] | None = None) pandas.DataFrame[source]
将 MLflow 跟踪中的表加载为 pandas.DataFrame。该表从指定 run_ids 中的 artifact_file 加载。extra_columns 是表中不存在但会用运行信息进行补充并添加到 DataFrame 的列。
- 参数
artifact_file – 要加载的表的 posixpath 格式的相对运行的artifact 文件路径（例如，“dir/file.json”）。
run_ids – 可选的运行 ID 列表，用于从这些运行中加载表。如果未指定 run_ids，则表将从当前实验的所有运行中加载。
extra_columns – 可选的额外列列表，将添加到返回的 DataFrame 中。例如，如果 extra_columns=[“run_id”]，则返回的 DataFrame 将具有一个名为 run_id 的列。
- 返回
包含已加载表的 pandas.DataFrame，如果 artifact 存在，否则将引发 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]
将本地文件或目录记录为当前活动运行的 artifact。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
local_path – 要写入的文件的路径。
artifact_path – 如果提供，则为
artifact_uri中要写入的目录。run_id – 如果指定，则将 artifact 记录到指定的运行。如果未指定，则将 artifact 记录到当前活动运行。
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]
将本地目录的所有内容记录为运行的 artifact。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
local_dir – 要写入的文件所在的目录路径。
artifact_path – 如果提供，则为
artifact_uri中要写入的目录。run_id – 如果指定，则将 artifact 记录到指定的运行。如果未指定，则将 artifact 记录到当前活动运行。
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 格式。
- 参数
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]
将图形记录为 artifact。支持以下图形对象：
- 参数
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 中记录图像，支持两种用例：
- 按时间步长的图像记录
最适合跟踪迭代过程中的变化或进展（例如，在模型训练阶段）。
用法：
log_image(image, key=key, step=step, timestamp=timestamp)
- Artifact 文件图像记录
最适合静态图像记录，其中图像直接保存为文件 artifact。
用法：
log_image(image, artifact_file)
- 支持以下图像格式：
-
mlflow.Image：PIL 图像的 MLflow 包装器，方便图像记录。
- Numpy 数组支持
数据类型
bool（用于记录图像掩码很有用）
整数 [0, 255]
无符号整数 [0, 255]
浮点数 [0.0, 1.0]
警告
超出范围的整数值将引发 ValueError。
超出范围的浮点数值将根据 min/max 自动缩放并发出警告。
形状（H：高度，W：宽度）
H x W（灰度）
H x W x 1（灰度）
H x W x 3（假定为 RGB 通道顺序）
H x W x 4（假定为 RGBA 通道顺序）
- 参数
image – 要记录的图像对象。
artifact_file – 指定图像将作为 artifact 存储在运行根目录下的路径（POSIX 格式，例如，“dir/image.png”）。此参数保留用于向后兼容，不应与 key、step 或 timestamp 一起使用。
key – 用于按时间步长记录图像的图像名称。此字符串只能包含字母数字字符、下划线（_）、连字符（-）、点（.）、空格（ ）和斜杠（/）。
step – 保存图像的整数训练步（迭代）。默认为 0。
timestamp – 保存此图像的时间。默认为当前系统时间。
synchronous – *实验性* 如果为 True，则阻塞直到图像成功记录。如果为 False，则异步记录图像并返回表示日志操作的 future。
import mlflow import numpy as np image = np.random.randint(0, 256, size=(100, 100, 3), dtype=np.uint8) with mlflow.start_run(): mlflow.log_image(image, key="dogs", step=3)
import mlflow from PIL import Image image = Image.new("RGB", (100, 100)) with mlflow.start_run(): mlflow.log_image(image, key="dogs", step=3)
import mlflow from PIL import Image # If you have a preexisting saved image Image.new("RGB", (100, 100)).save("image.png") image = mlflow.Image("image.png") with mlflow.start_run() as run: mlflow.log_image(run.info.run_id, image, key="dogs", step=3)
import mlflow import numpy as np image = np.random.randint(0, 256, size=(100, 100, 3), dtype=np.uint8) with mlflow.start_run(): mlflow.log_image(image, "image.png")
- mlflow.log_input(dataset: Optional[mlflow.data.dataset.Dataset] = None, context: Optional[str] = None, tags: Optional[dict[str, str]] = None, model: Optional[LoggedModelInput] = None) None[source]
记录当前运行中使用的数据集。
- 参数
dataset – 要记录的
mlflow.data.dataset.Dataset对象。context – 使用数据集的上下文。例如：“training”、“testing”。这将作为具有键 mlflow.data.context 的输入标签进行设置。
tags – 要与数据集关联的标签。标签键 -> 标签值字典。
model – 要作为运行输入记录的
mlflow.entities.LoggedModelInput实例。
- mlflow.log_inputs(datasets: Optional[list[typing.Optional[mlflow.data.dataset.Dataset]]] = None, contexts: Optional[list[str | None]] = None, tags_list: Optional[list[dict[str, str] | None]] = None, models: Optional[list[LoggedModelInput | None]] = None) None[source]
记录当前运行中使用的一批数据集。
datasets、contexts、tags_list 的列表必须具有相同的长度。这些列表中的条目可以是
None，这表示对应输入的空值。- 参数
datasets – 要记录的
mlflow.data.dataset.Dataset对象列表。contexts – 使用数据集的上下文列表。例如：“training”、“testing”。这将作为具有键 mlflow.data.context 的输入标签进行设置。
tags_list – 要与数据集关联的标签列表。标签键 -> 标签值字典。
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[Union[mlflow.data.dataset.Dataset, Dataset]] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
在当前运行下记录一个指标。如果没有活动运行，此方法将创建一个新的活动运行。
- 参数
key – 指标名称。此字符串只能包含字母数字字符、下划线（_）、连字符（-）、点（.）、空格（ ）和斜杠（/）。所有后端存储将支持长度最多为 250 的键，但有些可能支持更长的键。
value – 指标值。请注意，某些特殊值（如 +/- Infinity）可能会被其他值替换，具体取决于存储。例如，SQLAlchemy 存储会将 +/- Infinity 替换为最大/最小值浮点数。所有后端存储将支持长度最多为 5000 的值，但有些可能支持更大的值。
step – 指标步。如果未指定，默认为零。
synchronous – *实验性* 如果为 True，则阻塞直到指标成功记录。如果为 False，则异步记录指标并返回表示日志操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 读取，如果未设置，则默认为 False。
timestamp – 计算此指标的时间。默认为当前系统时间。
run_id – 如果指定，则将指标记录到指定的运行。如果未指定，则将指标记录到当前活动运行。
model_id – 与指标关联的模型 ID。如果未指定，则使用通过
mlflow.set_active_model()设置的当前活动模型 ID。如果不存在活动模型，则将使用指定的或活动运行的模型 ID。dataset – 与指标关联的数据集。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回 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[Union[mlflow.data.dataset.Dataset, Dataset]] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
为当前运行记录多个指标。如果当前没有运行，此方法将创建一个新的运行。
- 参数
metrics – 指标字典：metric_name: String -> value: Float。请注意，某些特殊值（如 +/- Infinity）可能会被其他值替换，具体取决于存储。例如，基于 SQL 的存储可能会将 +/- Infinity 替换为最大/最小浮点值。
step – 记录指定指标的单个整数步长。如果未指定，则每个指标将记录在步长零。
synchronous – *实验性* 如果为 True，则阻塞直到指标成功记录。如果为 False，则异步记录指标并返回一个表示日志记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 读取，如果未设置，则默认为 False。
run_id – 运行 ID。如果指定，则将指标记录到指定的运行。如果未指定，则将指标记录到当前正在运行的运行。
timestamp – 计算这些指标的时间。默认为当前系统时间。
model_id – 与指标关联的模型 ID。如果未指定，则使用通过
mlflow.set_active_model()设置的当前活动模型 ID。如果不存在活动模型，则将使用指定的或活动运行的模型 ID。dataset – 与指标关联的数据集。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回一个
mlflow.utils.async_logging.run_operations.RunOperations实例，该实例代表日志记录操作的 future。
- mlflow.log_outputs(models: Optional[list[LoggedModelOutput]] = None)[source]
将模型等输出记录到当前运行。如果当前没有运行，将创建一个新的运行。
- 参数
models – 要记录为运行输出的
mlflow.entities.LoggedModelOutput实例列表。- 返回
无。
- mlflow.log_param(key: str, value: Any, synchronous: Optional[bool] = None) Any[source]
记录当前运行下的参数（例如模型超参数）。如果当前没有运行，此方法将创建一个新的运行。
- 参数
key – 参数名称。此字符串只能包含字母数字字符、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储都支持长度最多为 250 的键，但有些可能支持更大的键。
value – 参数值，但如果不是字符串，则会被转换为字符串。所有内置后端存储都支持长度最多为 6000 的值，但有些可能支持更大的值。
synchronous – *实验性* 如果为 True，则阻塞直到参数成功记录。如果为 False，则异步记录参数并返回一个表示日志记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 读取，如果未设置，则默认为 False。
- 返回
当 synchronous=True 时，返回参数值。当 synchronous=False 时，返回一个
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 – 参数字典：param_name: String -> value: (String，但如果不是字符串，则会被转换为字符串)
synchronous – *实验性* 如果为 True，则阻塞直到参数成功记录。如果为 False，则异步记录参数并返回一个表示日志记录操作的 future。如果为 None，则从环境变量 MLFLOW_ENABLE_ASYNC_LOGGING 读取，如果未设置，则默认为 False。
run_id – 运行 ID。如果指定，则将参数记录到指定的运行。如果未指定，则将参数记录到当前正在运行的运行。
- 返回
当 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]
将表记录为 MLflow Tracking 中的 JSON 伪件。如果 artifact_file 已在运行中存在，则数据将被追加到现有的 artifact_file 中。
- 参数
data – 要记录的字典或 pandas.DataFrame。
artifact_file – 要将表保存到的运行相对的伪件文件路径（以 posixpath 格式）（例如，“dir/file.json”）。
run_id – 如果指定，则将表记录到指定的运行。如果未指定，则将表记录到当前正在运行的运行。
import mlflow table_dict = { "inputs": ["What is MLflow?", "What is Databricks?"], "outputs": ["MLflow is ...", "Databricks is ..."], "toxicity": [0.0, 0.0], } with mlflow.start_run(): # Log the dictionary as a table mlflow.log_table(data=table_dict, artifact_file="qabot_eval_results.json")
import mlflow import pandas as pd table_dict = { "inputs": ["What is MLflow?", "What is Databricks?"], "outputs": ["MLflow is ...", "Databricks is ..."], "toxicity": [0.0, 0.0], } df = pd.DataFrame.from_dict(table_dict) with mlflow.start_run(): # Log the df as a table mlflow.log_table(data=df, artifact_file="qabot_eval_results.json")
- mlflow.log_text(text: str, artifact_file: str, run_id: Optional[str] = None) None[source]
将文本作为伪件记录。
- 参数
text – 包含要记录的文本的字符串。
artifact_file – 要将文本保存到的运行相对的伪件文件路径（以 posixpath 格式）（例如，“dir/file.txt”）。
run_id – 如果指定，则将 artifact 记录到指定的运行。如果未指定，则将 artifact 记录到当前活动运行。
import mlflow with mlflow.start_run(): # Log text to a file under the run's root artifact directory mlflow.log_text("text1", "file1.txt") # Log text in a subdirectory of the run's root artifact directory mlflow.log_text("text2", "dir/file2.txt") # Log HTML text mlflow.log_text("<h1>header</h1>", "index.html")
- mlflow.log_trace(name: str = 'Task', request: Optional[Any] = None, response: Optional[Any] = None, intermediate_outputs: dict[str, typing.Any] | None = None, attributes: dict[str, typing.Any] | None = None, tags: dict[str, str] | None = None, start_time_ms: int | None = None, execution_time_ms: int | None = None) str[source]
警告
mlflow.tracing.fluent.log_trace自 3.6.0 版本起已弃用。此方法将在未来的版本中删除。创建一个具有单个根 span 的 trace。此 API 在您想要记录任意（请求，响应）对而不使用结构化 OpenTelemetry span 时很有用。trace 将链接到当前实验。
- 参数
name – trace（以及根 span）的名称。默认为“Task”。
request – 整个 trace 的输入数据。此数据也设置在 trace 的根 span 上。
response – 整个 trace 的输出数据。此数据也设置在 trace 的根 span 上。
intermediate_outputs – 模型或代理在处理请求时生成的中间输出字典。键是输出的名称，值是输出本身。值必须是 JSON 可序列化的。
attributes – 要在 trace 的根 span 上设置的属性字典。
tags – 要在 trace 上设置的标签字典。
start_time_ms – trace 在自 Unix epoch 以来的毫秒数。未指定时，trace 的开始和结束时间将使用当前时间。
execution_time_ms – trace 在自 Unix epoch 以来的毫秒数。
- 返回
记录的 trace 的 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。
- 参数
backend – 字符串，跟踪服务器的后端。目前仅支持“databricks”。
interactive – 布尔值，控制在缺少凭据时是否请求用户输入。如果为 true，则在找不到凭据时会请求用户输入，否则将在找不到凭据时引发异常。
- mlflow.override_feedback(*, trace_id: str, assessment_id: str, value: float | int | str | bool | dict[str, float | int | str | bool] | list[float | int | str | bool], rationale: Optional[str] = None, source: Optional[AssessmentSource] = None, metadata: Optional[dict[str, typing.Any]] = None) Assessment[source]
覆盖现有的反馈评估，用新的评估替换。此 API 会记录一个具有 overrides 字段设置为提供的评估 ID 的新评估。原始评估将被标记为无效，但除此之外不会更改。当您想更正 LLM 裁判生成的评估，但又想保留原始评估供将来裁判微调时，此功能很有用。
如果您想就地修改评估，请使用
update_assessment()而不是。- 参数
trace_id – 跟踪的 ID。
assessment_id – 要覆盖的评估 ID。
value – 评估的新值。
rationale – 新评估的理由。
source – 新评估的来源。
metadata – 新评估的其他元数据。
- 返回
创建的评估。
- 返回类型
示例
import mlflow from mlflow.entities import AssessmentSource, AssessmentSourceType # First, log an initial LLM-generated feedback as a simulation llm_feedback = mlflow.log_feedback( trace_id="tr-1234567890abcdef", name="relevance", value=0.6, source=AssessmentSource( source_type=AssessmentSourceType.LLM_JUDGE, source_id="gpt-4" ), rationale="Response partially addresses the question", ) # Later, a human reviewer disagrees and wants to override corrected_assessment = mlflow.override_feedback( trace_id="tr-1234567890abcdef", assessment_id=llm_feedback.assessment_id, value=0.9, rationale="Response fully addresses the question with good examples", source=AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="expert_reviewer@company.com" ), metadata={ "override_reason": "LLM underestimated relevance", "review_date": "2024-01-15", "confidence": "high", }, )
- mlflow.register_model(model_uri, name, await_registration_for=300, *, tags: Optional[dict[str, typing.Any]] = None, env_pack: Optional[Union[Literal['databricks_model_serving'], mlflow.utils.env_pack.EnvPackConfig]] = None) ModelVersion[source]
为
model_uri指定的模型文件在模型注册表中创建一个新的模型版本。注意，此方法假定模型注册表后端 URI 与跟踪后端 URI 相同。
- 参数
model_uri – 指向 MLmodel 目录的 URI。如果要将运行 ID 记录到模型注册表中的模型（推荐），请使用
runs:/URI，或者如果您要注册先前使用save_model保存的本地持久化 MLflow 模型，则传递模型的本地文件系统路径。models:/URI 目前不支持。name – 要在新模型版本下创建的注册模型的名称。如果不存在具有给定名称的注册模型，它将被自动创建。
await_registration_for – 等待模型版本完成创建并处于
READY状态的秒数。默认情况下，函数等待五分钟。指定 0 或 None 可跳过等待。tags – 将转换为
mlflow.entities.model_registry.ModelVersionTag对象的键值对字典。env_pack –
可以是字符串或 EnvPackConfig。如果指定，模型依赖项将首先安装到当前 Python 环境中，然后整个环境将被打包并包含在注册的模型伪件中。如果使用字符串快捷方式“databricks_model_serving”，则会将模型依赖项安装在当前环境中。这在使用 Databricks Model Serving 等服务环境部署模型时很有用。
注意
Experimental: This parameter may change or be removed in a future release without warning. (实验性：此参数可能在未来的版本中更改或移除，恕不另行通知。)
- 返回
后端创建的单个
mlflow.entities.model_registry.ModelVersion对象。
import mlflow.sklearn from mlflow.models import infer_signature from sklearn.datasets import make_regression from sklearn.ensemble import RandomForestRegressor mlflow.set_tracking_uri("sqlite:////tmp/mlruns.db") params = {"n_estimators": 3, "random_state": 42} X, y = make_regression(n_features=4, n_informative=2, random_state=0, shuffle=False) # Log MLflow entities with mlflow.start_run() as run: rfr = RandomForestRegressor(**params).fit(X, y) signature = infer_signature(X, rfr.predict(X)) mlflow.log_params(params) mlflow.sklearn.log_model(rfr, name="sklearn-model", signature=signature) model_uri = f"runs:/{run.info.run_id}/sklearn-model" mv = mlflow.register_model(model_uri, "RandomForestRegressionModel") print(f"Name: {mv.name}") print(f"Version: {mv.version}")
- mlflow.run(uri, entry_point='main', version=None, parameters=None, docker_args=None, experiment_name=None, experiment_id=None, backend='local', backend_config=None, storage_dir=None, synchronous=True, run_id=None, run_name=None, env_manager=None, build_image=False, docker_auth=None)[源代码]
运行 MLflow 项目。项目可以是本地的，也可以存储在 Git URI 中。
MLflow 内置支持在本地或远程的 Databricks 或 Kubernetes 集群上运行项目。您还可以通过安装适当的第三方插件来运行针对其他目标的项目。有关更多信息，请参阅 社区插件。
有关在链式工作流中使用此方法的详细信息，请参阅 构建多步工作流。
- 引发
mlflow.exceptions.ExecutionException – 执行不成功。
- 参数
uri – 要运行的项目 URI。可以是本地文件系统路径，也可以是指向包含 MLproject 文件的项目目录的 Git 存储库 URI（例如 https://github.com/mlflow/mlflow-example）。
entry_point – 要在项目中运行的入口点。如果找不到具有指定名称的入口点，则将作为脚本运行项目文件
entry_point，使用“python”运行.py文件，并使用默认 shell（由环境变量$SHELL指定）运行.sh文件。version – 对于基于 Git 的项目，可以是提交哈希或分支名称。
parameters – 入口点命令的参数（字典）。
docker_args – docker 命令的参数（字典）。
experiment_name – 要在其下启动运行的实验名称。
experiment_id – 要在其下启动运行的实验 ID。
backend – 运行的执行后端：MLflow 内置支持“local”（本地）、“databricks”和“kubernetes”（实验性）后端。如果针对 Databricks 运行，将针对以下方式确定的 Databricks 工作区运行：如果已设置形式为
databricks://profile的 Databricks 跟踪 URI（例如，通过设置 MLFLOW_TRACKING_URI 环境变量），则将针对 <profile> 指定的工作区运行。否则，将针对默认 Databricks CLI 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 项目 API 内部使用，不应指定。如果指定，将使用指定的 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。
- 返回
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[Experiment][源代码]
搜索符合指定搜索查询的实验。
- 参数
view_type – 在
mlflow.entities.ViewType中定义的枚举值ACTIVE_ONLY、DELETED_ONLY或ALL之一。max_results – 如果传递，则指定所需实验的最大数量。如果未传递，则返回所有实验。
filter_string –
过滤查询字符串（例如，
"name = 'my_experiment'"），默认为搜索所有实验。支持以下标识符、比较器和逻辑运算符。- 标识符
name: 实验名称creation_time: 实验创建时间last_update_time: 实验最后更新时间tags.<tag_key>: 实验标签。如果tag_key包含空格，则必须用反引号括起来（例如，"tags.`extra key`"）。
- 字符串属性和标签的比较器
=: 等于!=: 不等于LIKE: 区分大小写的模式匹配ILIKE: 不区分大小写的模式匹配
- 数值属性的比较器
=: 等于!=: 不等于<: 小于<=: 小于或等于>: 大于>=: 大于或等于
- 逻辑运算符
AND: 组合两个子查询，当两个子查询都为 True 时返回 True。
order_by –
用于排序的列列表。
order_by列可以包含可选的DESC或ASC值（例如，"name DESC"）。默认排序为ASC，因此"name"等同于"name ASC"。如果未指定，则默认为["last_update_time DESC"]，它将最近更新的实验排在前面。支持以下字段：experiment_id: 实验 IDname: 实验名称creation_time: 实验创建时间last_update_time: 实验最后更新时间
- 返回
一个
Experiment对象列表。
import mlflow def assert_experiment_names_equal(experiments, expected_names): actual_names = [e.name for e in experiments if e.name != "Default"] assert actual_names == expected_names, (actual_names, expected_names) mlflow.set_tracking_uri("sqlite:///:memory:") # Create experiments for name, tags in [ ("a", None), ("b", None), ("ab", {"k": "v"}), ("bb", {"k": "V"}), ]: mlflow.create_experiment(name, tags=tags) # Search for experiments with name "a" experiments = mlflow.search_experiments(filter_string="name = 'a'") assert_experiment_names_equal(experiments, ["a"]) # Search for experiments with name starting with "a" experiments = mlflow.search_experiments(filter_string="name LIKE 'a%'") assert_experiment_names_equal(experiments, ["ab", "a"]) # Search for experiments with tag key "k" and value ending with "v" or "V" experiments = mlflow.search_experiments(filter_string="tags.k ILIKE '%v'") assert_experiment_names_equal(experiments, ["bb", "ab"]) # Search for experiments with name ending with "b" and tag {"k": "v"} experiments = mlflow.search_experiments(filter_string="name LIKE '%b' AND tags.k = 'v'") assert_experiment_names_equal(experiments, ["ab"]) # Sort experiments by name in ascending order experiments = mlflow.search_experiments(order_by=["name"]) assert_experiment_names_equal(experiments, ["a", "ab", "b", "bb"]) # Sort experiments by ID in descending order experiments = mlflow.search_experiments(order_by=["experiment_id DESC"]) assert_experiment_names_equal(experiments, ["bb", "ab", "b", "a"])
- mlflow.search_model_versions(max_results: Optional[int] = None, filter_string: Optional[str] = None, order_by: Optional[list[str]] = None) list[ModelVersion][源代码]
搜索满足过滤条件的模型版本。
- 参数
max_results – 如果传递，则指定所需模型的最大数量。如果未传递，则返回所有模型。
filter_string –
过滤查询字符串（例如，
"name = 'a_model_name' and tag.key = 'value1'"），默认为搜索所有模型版本。支持以下标识符、比较器和逻辑运算符。- 标识符
name: 模型名称。source_path: 模型版本源路径。run_id: 生成模型版本的 MLflow 运行 ID。tags.<tag_key>: 模型版本标签。如果tag_key包含空格，则必须用反引号括起来（例如，"tags.`extra key`"）。
- 比较器
=: 等于。!=: 不等于。LIKE: 区分大小写的模式匹配。ILIKE: 不区分大小写的模式匹配。IN: 在值列表中。只有run_id标识符支持IN比较器。
- 逻辑运算符
AND: 组合两个子查询，当两个子查询都为 True 时返回 True。
order_by – 用于对匹配的搜索结果进行排序的列名列表，带 ASC|DESC 注释。
- 返回
- 一个
mlflow.entities.model_registry.ModelVersion对象列表 满足搜索表达式。
- 一个
import mlflow from sklearn.linear_model import LogisticRegression for _ in range(2): with mlflow.start_run(): mlflow.sklearn.log_model( LogisticRegression(), name="Cordoba", registered_model_name="CordobaWeatherForecastModel", ) # Get all versions of the model filtered by name filter_string = "name = 'CordobaWeatherForecastModel'" results = mlflow.search_model_versions(filter_string=filter_string) print("-" * 80) for res in results: print(f"name={res.name}; run_id={res.run_id}; version={res.version}") # Get the version of the model filtered by run_id filter_string = "run_id = 'ae9a606a12834c04a8ef1006d0cff779'" results = mlflow.search_model_versions(filter_string=filter_string) print("-" * 80) for res in results: print(f"name={res.name}; run_id={res.run_id}; version={res.version}")
-------------------------------------------------------------------------------- name=CordobaWeatherForecastModel; run_id=ae9a606a12834c04a8ef1006d0cff779; version=2 name=CordobaWeatherForecastModel; run_id=d8f028b5fedf4faf8e458f7693dfa7ce; version=1 -------------------------------------------------------------------------------- name=CordobaWeatherForecastModel; run_id=ae9a606a12834c04a8ef1006d0cff779; version=2
- mlflow.search_registered_models(max_results: Optional[int] = None, filter_string: Optional[str] = None, order_by: Optional[list[str]] = None) list[RegisteredModel][源代码]
搜索满足过滤条件的已注册模型。
- 参数
max_results – 如果传递，则指定所需模型的最大数量。如果未传递，则返回所有模型。
filter_string –
过滤查询字符串（例如，“name = ‘a_model_name’ and tag.key = ‘value1’”），默认为搜索所有已注册模型。支持以下标识符、比较器和逻辑运算符。
- 标识符
“name”：已注册模型的名称。
“tags.<tag_key>”：已注册模型的标签。如果“tag_key”包含空格，则必须用反引号括起来（例如，“tags.`extra key`”）。
- 比较器
”=”: 等于。
”!=”: 不等于。
”LIKE”: 区分大小写的模式匹配。
”ILIKE”: 不区分大小写的模式匹配。
- 逻辑运算符
”AND”: 组合两个子查询，当两个子查询都为 True 时返回 True。
order_by – 用于对匹配的搜索结果进行排序的列名列表，带 ASC|DESC 注释。
- 返回
一个
mlflow.entities.model_registry.RegisteredModel对象列表，满足搜索表达式。
import mlflow from sklearn.linear_model import LogisticRegression with mlflow.start_run(): mlflow.sklearn.log_model( LogisticRegression(), name="Cordoba", registered_model_name="CordobaWeatherForecastModel", ) mlflow.sklearn.log_model( LogisticRegression(), name="Boston", registered_model_name="BostonWeatherForecastModel", ) # Get search results filtered by the registered model name filter_string = "name = 'CordobaWeatherForecastModel'" results = mlflow.search_registered_models(filter_string=filter_string) print("-" * 80) for res in results: for mv in res.latest_versions: print(f"name={mv.name}; run_id={mv.run_id}; version={mv.version}") # Get search results filtered by the registered model name that matches # prefix pattern filter_string = "name LIKE 'Boston%'" results = mlflow.search_registered_models(filter_string=filter_string) print("-" * 80) for res in results: for mv in res.latest_versions: print(f"name={mv.name}; run_id={mv.run_id}; version={mv.version}") # Get all registered models and order them by ascending order of the names results = mlflow.search_registered_models(order_by=["name ASC"]) print("-" * 80) for res in results: for mv in res.latest_versions: print(f"name={mv.name}; run_id={mv.run_id}; version={mv.version}")
-------------------------------------------------------------------------------- name=CordobaWeatherForecastModel; run_id=248c66a666744b4887bdeb2f9cf7f1c6; version=1 -------------------------------------------------------------------------------- name=BostonWeatherForecastModel; run_id=248c66a666744b4887bdeb2f9cf7f1c6; version=1 -------------------------------------------------------------------------------- name=BostonWeatherForecastModel; run_id=248c66a666744b4887bdeb2f9cf7f1c6; version=1 name=CordobaWeatherForecastModel; run_id=248c66a666744b4887bdeb2f9cf7f1c6; version=1
- mlflow.search_runs(experiment_ids: list[str] | None = None, filter_string: str = '', run_view_type: int = 1, max_results: int = 100000, order_by: list[str] | None = None, output_format: str = 'pandas', search_all_experiments: bool = False, experiment_names: list[str] | None = None) Union[list[Run], pandas.DataFrame][源代码]
搜索符合指定条件的 Runs。
- 参数
experiment_ids – 实验 ID 列表。搜索可以使用实验 ID 或实验名称，但不能在同一次调用中使用两者。如果
experiment_names也不是None或[]，则非None或[]的值将导致错误。如果experiment_names是None或[]，则None将默认为活动实验。filter_string – 过滤查询字符串，默认为搜索所有 run。
run_view_type –
mlflow.entities.ViewType中定义的枚举值ACTIVE_ONLY、DELETED_ONLY或ALL之一。max_results – 要放入 DataFrame 的最大 run 数量。默认值为 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将默认为活动实验。
- 返回
一个
mlflow.entities.Run列表。如果 output_format 为pandas：一个 run 的pandas.DataFrame，其中每个指标、参数和标签都将展开为其自己的列，分别命名为 metrics.*、params.* 或 tags.*。对于没有特定指标、参数或标签的 run，相应列的值分别为 (NumPy)Nan、None或None。- 返回类型
如果 output_format 为
listimport mlflow # Create an experiment and log two runs under it experiment_name = "Social NLP Experiments" experiment_id = mlflow.create_experiment(experiment_name) with mlflow.start_run(experiment_id=experiment_id): mlflow.log_metric("m", 1.55) mlflow.set_tag("s.release", "1.1.0-RC") with mlflow.start_run(experiment_id=experiment_id): mlflow.log_metric("m", 2.50) mlflow.set_tag("s.release", "1.2.0-GA") # Search for all the runs in the experiment with the given experiment ID df = mlflow.search_runs([experiment_id], order_by=["metrics.m DESC"]) print(df[["metrics.m", "tags.s.release", "run_id"]]) print("--") # Search the experiment_id using a filter_string with tag # that has a case insensitive pattern filter_string = "tags.s.release ILIKE '%rc%'" df = mlflow.search_runs([experiment_id], filter_string=filter_string) print(df[["metrics.m", "tags.s.release", "run_id"]]) print("--") # Search for all the runs in the experiment with the given experiment name df = mlflow.search_runs(experiment_names=[experiment_name], order_by=["metrics.m DESC"]) print(df[["metrics.m", "tags.s.release", "run_id"]])
metrics.m tags.s.release run_id 0 2.50 1.2.0-GA 147eed886ab44633902cc8e19b2267e2 1 1.55 1.1.0-RC 5cc7feaf532f496f885ad7750809c4d4 -- metrics.m tags.s.release run_id 0 1.55 1.1.0-RC 5cc7feaf532f496f885ad7750809c4d4 -- metrics.m tags.s.release run_id 0 2.50 1.2.0-GA 147eed886ab44633902cc8e19b2267e2 1 1.55 1.1.0-RC 5cc7feaf532f496f885ad7750809c4d4
- mlflow.set_experiment(experiment_name: Optional[str] = None, experiment_id: Optional[str] = None) Experiment[源代码]
将给定的实验设置为活动实验。该实验必须通过 experiment_name 指定名称，或通过 experiment_id 指定 ID。实验名称和 ID 不能同时指定。
注意
如果按名称设置的实验不存在，则会创建一个具有给定名称的新实验。创建实验后，它将被设置为活动实验。在某些平台（如 Databricks）上，实验名称必须是绝对路径，例如
"/Users/<username>/my-experiment"。- 参数
experiment_name – 要激活的实验的区分大小写的名称。
experiment_id – 要激活的实验的 ID。如果具有此 ID 的实验不存在，则会引发异常。
- 返回
表示新活动实验的
mlflow.entities.Experiment实例。
import mlflow # Set an experiment name, which must be unique and case-sensitive. experiment = mlflow.set_experiment("Social NLP Experiments") # Get Experiment Details print(f"Experiment_id: {experiment.experiment_id}") print(f"Artifact Location: {experiment.artifact_location}") print(f"Tags: {experiment.tags}") print(f"Lifecycle_stage: {experiment.lifecycle_stage}")
- mlflow.set_experiment_tag(key: str, value: Any) None[source]
为当前实验设置标签。值将被转换为字符串。
- 参数
key – 标签名称。此字符串只能包含字母数字字符、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储都支持长度最多为 250 的键，但有些可能支持更大的键。
value – 标签值，但如果不是字符串，则会被字符串化。所有后端存储都支持长度最多为 5000 的值，但有些可能支持更大的值。
- mlflow.set_experiment_tags(tags: dict[str, typing.Any]) None[source]
为当前活动实验设置标签。
- 参数
tags – 包含标签名称和相应值的字典。
- mlflow.set_model_version_tag(name: str, version: Optional[str] = None, key: Optional[str] = None, value: Optional[Any] = None) None[source]
为模型版本设置标签。
- 参数
name – 注册模型名称。
version – 注册模型版本。
key – 要记录的标签键。键是必需的。
value – 要记录的标签值。值是必需的。
- mlflow.set_registry_uri(uri: str) None[source]
设置注册表服务器 URI。如果您的注册表服务器与跟踪服务器不同，此方法将特别有用。
- 参数
uri – 空字符串，或以
file:/为前缀的本地文件路径。数据将存储在提供的文件（如果为空则存储在./mlruns）中。HTTP URI，如https://my-tracking-server:5000或http://my-oss-uc-server:8080。Databricks 工作区，提供为字符串“databricks”或使用 Databricks CLI 配置文件，提供为“databricks://<profileName>”。
import mflow # Set model registry uri, fetch the set uri, and compare # it with the tracking uri. They should be different mlflow.set_registry_uri("sqlite:////tmp/registry.db") mr_uri = mlflow.get_registry_uri() print(f"Current registry uri: {mr_uri}") tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}") # They should be different assert tracking_uri != mr_uri
- mlflow.set_system_metrics_node_id(node_id)[source]
设置系统指标节点 ID。
node_id 是收集指标的机器的标识符。这在多节点（分布式训练）设置中很有用。
- mlflow.set_system_metrics_samples_before_logging(samples)[source]
设置记录系统指标之前的采样数。
每次收集了 samples 个样本后，系统指标将记录到 mlflow。默认情况下 samples=1。
- mlflow.set_system_metrics_sampling_interval(interval)[source]
设置系统指标采样间隔。
每 interval 秒，将收集系统指标。默认情况下 interval=10。
- mlflow.set_tag(key: str, value: Any, synchronous: Optional[bool] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
在当前运行下设置标签。如果没有活动的运行，此方法将创建一个新的活动运行。
- 参数
key – 标签名称。此字符串只能包含字母数字字符、下划线 (_)、连字符 (-)、句点 (.)、空格 ( ) 和斜杠 (/)。所有后端存储都支持长度最多为 250 的键，但有些可能支持更大的键。
value – 标签值，但如果不是字符串，则会被字符串化。所有后端存储都支持长度最多为 5000 的值，但有些可能支持更大的值。
synchronous – 实验性 如果为 True，则阻塞直到标签成功记录。如果为 False，则异步记录标签并返回一个表示记录操作的 future。如果为 None，则读取环境变量 MLFLOW_ENABLE_ASYNC_LOGGING，如果未设置，则默认为 False。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回一个
mlflow.utils.async_logging.run_operations.RunOperations实例，该实例代表日志记录操作的 future。
- mlflow.set_tags(tags: dict[str, typing.Any], synchronous: Optional[bool] = None) mlflow.utils.async_logging.run_operations.RunOperations | None[source]
为当前运行记录一批标签。如果没有活动的运行，此方法将创建一个新的活动运行。
- 参数
tags – 标签名称：字符串 -> 值：(字符串，但如果不是字符串，则会被字符串化) 的字典
synchronous – 实验性 如果为 True，则阻塞直到标签成功记录。如果为 False，则异步记录标签并返回一个表示记录操作的 future。如果为 None，则读取环境变量 MLFLOW_ENABLE_ASYNC_LOGGING，如果未设置，则默认为 False。
- 返回
当 synchronous=True 时，返回 None。当 synchronous=False 时，返回一个
mlflow.utils.async_logging.run_operations.RunOperations实例，该实例代表日志记录操作的 future。
- mlflow.set_trace_tag(trace_id: str, key: str, value: str)[source]
注意
参数
request_id已弃用。请使用trace_id。为具有给定跟踪 ID 的跟踪设置标签。
跟踪可以是活动的跟踪，也可以是已经结束并记录在后端中的跟踪。下面是一个为活动跟踪设置标签的示例。您可以替换
trace_id参数来为已结束的跟踪设置标签。import mlflow with mlflow.start_span(name="span") as span: mlflow.set_trace_tag(span.trace_id, "key", "value")
- 参数
trace_id – 要为其设置标签的跟踪的 ID。
key – 标签的字符串键。长度最多为 250 个字符，否则在存储时将被截断。
value – 标签的字符串值。必须最多包含 250 个字符，否则在存储时将被截断。
- mlflow.set_tracking_uri(uri: str | pathlib.Path) None[source]
设置跟踪服务器 URI。这不会影响当前活动的运行（如果存在），但会影响后续的运行。
- 参数
uri –
空字符串，或以
file:/为前缀的本地文件路径。数据将存储在提供的文件（如果为空则存储在./mlruns）中。HTTP URI，如
https://my-tracking-server:5000。Databricks 工作区，提供为字符串“databricks”或使用 Databricks CLI 配置文件，提供为“databricks://<profileName>”。
pathlib.Path实例
import mlflow mlflow.set_tracking_uri("file:///tmp/my_tracking") tracking_uri = mlflow.get_tracking_uri() print(f"Current tracking uri: {tracking_uri}")
- mlflow.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 系统标签 中定义的。
- 参数
run_id – 如果指定，则获取具有指定 UUID 的运行，并在该运行下记录参数和指标。运行的结束时间将被取消设置，并且其状态将设置为运行中，但运行的其他属性（
source_version、source_type等）不会改变。experiment_id – 创建当前运行的实验 ID（仅当未指定
run_id时适用）。如果未指定experiment_id参数，将按以下顺序查找有效的实验：通过set_experiment激活的实验、MLFLOW_EXPERIMENT_NAME环境变量、MLFLOW_EXPERIMENT_ID环境变量，或跟踪服务器定义的默认实验。run_name – 新运行的名称，应为非空字符串。仅当未指定
run_id时使用。如果创建新运行且未指定run_name，将为运行生成一个随机名称。nested – 控制运行是否嵌套在父运行中。
True将创建嵌套运行。parent_run_id – 如果指定，当前运行将嵌套在具有指定 UUID 的运行下。父运行必须处于激活状态。
tags – 一个可选的字符串键值字典，用于将标签设置为运行的标签。如果正在恢复运行，这些标签将设置在恢复的运行上。如果正在创建新运行，这些标签将设置在新运行上。
description – 一个可选字符串，用于填充运行的描述框。如果正在恢复运行，描述将设置在恢复的运行上。如果正在创建新运行，描述将设置在新运行上。
log_system_metrics – bool，默认为 None。如果为 True，系统指标将记录到 MLflow，例如 cpu/gpu 利用率。如果为 None，我们将检查环境变量 MLFLOW_ENABLE_SYSTEM_METRICS_LOGGING 以确定是否记录系统指标。系统指标记录是 MLflow 2.8 中的一项实验性功能，可能会发生变化。
- 返回
充当包装运行状态的上下文管理器的
mlflow.ActiveRun对象。
import mlflow # Create nested runs experiment_id = mlflow.create_experiment("experiment1") with mlflow.start_run( run_name="PARENT_RUN", experiment_id=experiment_id, tags={"version": "v1", "priority": "P1"}, description="parent", ) as parent_run: mlflow.log_param("parent", "yes") with mlflow.start_run( run_name="CHILD_RUN", experiment_id=experiment_id, description="child", nested=True, ) as child_run: mlflow.log_param("child", "yes") print("parent run:") print(f"run_id: {parent_run.info.run_id}") print("description: {}".format(parent_run.data.tags.get("mlflow.note.content"))) print("version tag value: {}".format(parent_run.data.tags.get("version"))) print("priority tag value: {}".format(parent_run.data.tags.get("priority"))) print("--") # Search all child runs with a parent id query = f"tags.mlflow.parentRunId = '{parent_run.info.run_id}'" results = mlflow.search_runs(experiment_ids=[experiment_id], filter_string=query) print("child runs:") print(results[["run_id", "params.child", "tags.mlflow.runName"]]) # Create a nested run under the existing parent run with mlflow.start_run( run_name="NEW_CHILD_RUN", experiment_id=experiment_id, description="new child", parent_run_id=parent_run.info.run_id, ) as child_run: mlflow.log_param("new-child", "yes")
- mlflow.update_current_trace(tags: Optional[dict[str, str]] = None, metadata: Optional[dict[str, str]] = None, client_request_id: Optional[str] = None, request_preview: Optional[str] = None, response_preview: Optional[str] = None, state: Optional[Union[TraceState, str]] = None, model_id: Optional[str] = None)[source]
使用给定的选项更新当前活动跟踪。
- 参数
tags – 要用标签更新的字典。标签专为可变值设计，可以通过 MLflow UI 或 API 在创建跟踪后进行更新。
metadata – 要用元数据更新的字典。元数据在跟踪记录后无法更新。它适用于记录不可变值，例如生成跟踪的应用程序版本的 git hash。
client_request_id – 客户端提供的请求 ID，用于与跟踪关联。这有助于将跟踪链接回应用程序或外部系统中的特定请求。如果为 None，则不更新客户端请求 ID。
request_preview – 要在 UI 的跟踪列表视图中显示的请求预览。默认情况下，MLflow 会通过限制长度来粗略截断跟踪请求。此参数允许您指定自定义预览字符串。
response_preview – 要在 UI 的跟踪列表视图中显示的响应预览。默认情况下，MLflow 会通过限制长度来粗略截断跟踪响应。此参数允许您指定自定义预览字符串。
state – 要在跟踪上设置的状态。可以是 TraceState 枚举值或字符串。只允许“OK”和“ERROR”。这会覆盖整个跟踪状态，而不影响当前跨度的状态。
model_id – 要与跟踪关联的模型 ID。如果未设置，则将活动模型 ID 与跟踪关联。
示例
您可以在使用
@mlflow.trace装饰的函数内，或在 with mlflow.start_span 上下文管理器的范围内使用此函数。如果找不到活动的跟踪，此函数将引发异常。在用 @mlflow.trace 装饰的函数中使用
@mlflow.trace def my_func(x): mlflow.update_current_trace(tags={"fruit": "apple"}, client_request_id="req-12345") return x + 1
在
with mlflow.start_span上下文管理器中使用with mlflow.start_span("span"): mlflow.update_current_trace(tags={"fruit": "apple"}, client_request_id="req-12345")
更新跟踪的来源信息。这些键是保留的，MLflow 默认会从环境信息中填充它们。如果需要，您可以覆盖它们。有关保留元数据键的完整列表，请参阅 MLflow 跟踪文档。
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 中模型验证功能的替代品。- 参数
validation_thresholds – 一个字典，其中键为指标名称，值为
mlflow.models.MetricThreshold，用于模型验证。每个指标名称必须是内置指标的名称，或者是extra_metrics参数中定义的指标的名称。candidate_result – 候选模型的评估结果。由
mlflow.evaluate()API 返回。baseline_result – 基线模型的评估结果。由
mlflow.evaluate()API 返回。如果设置为 None，则候选模型结果将直接与阈值进行比较。
代码示例
import mlflow from mlflow.models import MetricThreshold thresholds = { "accuracy_score": MetricThreshold( # accuracy should be >=0.8 threshold=0.8, # accuracy should be at least 5 percent greater than baseline model accuracy min_absolute_change=0.05, # accuracy should be at least 0.05 greater than baseline model accuracy min_relative_change=0.05, greater_is_better=True, ), } # Get evaluation results for the candidate model candidate_result = mlflow.evaluate( model="<YOUR_CANDIDATE_MODEL_URI>", data=eval_dataset, targets="ground_truth", model_type="classifier", ) # Get evaluation results for the baseline model baseline_result = mlflow.evaluate( model="<YOUR_BASELINE_MODEL_URI>", data=eval_dataset, targets="ground_truth", model_type="classifier", ) # Validate the results mlflow.validate_evaluation_results( thresholds, candidate_result, baseline_result, )
有关更多详细信息，请参阅 模型验证文档。
MLflow 跟踪 API
mlflow 模块提供了一组用于 MLflow 跟踪 的高级 API。有关如何使用这些跟踪 API 的详细指南，请参阅 跟踪 Fluent API 指南。
- mlflow.trace(func: 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, trace_destination: Optional[mlflow.entities.trace_location.TraceLocationBase] = None) Callable[[...], Any][source]
一个装饰器，用于为被装饰的函数创建一个新的 span。
当您使用此
@mlflow.trace()装饰器装饰一个函数时，将在被装饰函数的范围内创建一个 span。该 span 会自动捕获函数的输入和输出。当它应用于一个方法时，它不会捕获 self 参数。函数内部引发的任何异常都会将 span 状态设置为ERROR，并且会将详细信息（如异常消息和堆栈跟踪）记录到 span 的attributes字段。例如，以下代码将生成一个名为
"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 装饰器目前支持以下类型的函数：
支持的函数类型 函数类型
支持
同步
✅
异步
✅ (>= 2.16.0)
生成器
✅ (>= 2.20.2)
异步生成器
✅ (>= 2.20.2)
类方法
✅ (>= 3.0.0)
静态方法
✅ (>= 3.0.0)
有关使用 @mlflow.trace 装饰器的更多示例，包括流式/异步处理，请参阅 MLflow 跟踪文档。
提示
当您想跟踪自己定义的函数时，@mlflow.trace 装饰器非常有用。但是，您可能也想跟踪外部库中的函数。在这种情况下，您可以使用此
mlflow.trace()函数直接包装函数，而不是将其用作装饰器。这将创建与装饰器创建的 span 完全相同的 span，即捕获函数调用中的信息。import math import mlflow mlflow.trace(math.factorial)(5)
- 参数
func – 要装饰的函数。在使用作装饰器时，不得提供此参数。
name – span 的名称。如果未提供，则使用函数的名称。
span_type – span 的类型。可以是字符串或
SpanType枚举值。attributes – 要设置在 span 上的属性字典。
output_reducer – 一个用于将生成器函数的输出归约为单个值以设置为 span 输出的函数。
trace_destination – 用于记录 trace 的目标，例如 MLflow Experiment。如果未提供，则目标将是活动的 MLflow experiment 或由
mlflow.tracing.set_destination()函数设置的目标。此参数仅应用于根 span，为非根 span 设置此参数将被忽略并发出警告。
- mlflow.start_span(name: str = 'span', span_type: str | None = 'UNKNOWN', attributes: dict[str, typing.Any] | None = None, trace_destination: mlflow.entities.trace_location.TraceLocationBase | None = None) Generator[LiveSpan, None, None][source]
用于创建新 span 并将其作为当前 span 启动的上下文管理器。
此上下文管理器可自动管理 span 的生命周期和父子关系。当上下文管理器退出时，span 将结束。在上下文管理器内引发的任何异常都会将 span 状态设置为
ERROR，并且会将详细信息（如异常消息和堆栈跟踪）记录到 span 的attributes字段。可以在上下文管理器内创建新的 span，然后它们将被指定为子 span。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 没有父引用，**整个 trace 将在根 span 结束时被记录**。
提示
如果您希望对 trace 的生命周期有更明确的控制，可以使用 mlflow.start_span_no_context() API。它提供了更底层的 API 来启动 span 并显式管理父子关系。但是，只要它满足您的需求，通常建议使用此上下文管理器，因为它需要更少的样板代码且不易出错。
注意
默认情况下，上下文管理器不会在线程之间传播 span 上下文。有关如何在线程之间传播 span 上下文的信息，请参阅 多线程。
- 参数
name – span 的名称。
span_type – span 的类型。可以是字符串或
SpanType枚举值。attributes – 要设置在 span 上的属性字典。
trace_destination – 用于记录 trace 的目标，例如 MLflow Experiment。如果未提供，则目标将是活动的 MLflow experiment 或由
mlflow.tracing.set_destination()函数设置的目标。此参数仅应用于根 span，为非根 span 设置此参数将被忽略并发出警告。
- 返回
产生一个
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() 方法手动结束。
- 参数
name – span 的名称。
span_type – span 的类型。可以是字符串或
SpanType枚举值。parent_span – 要链接的父 span。如果为 None，则 span 将被视为根 span。
inputs – span 的输入数据。
attributes – 要设置在 span 上的属性字典。
tags – 要在 trace 上设置的标签字典。
experiment_id – 要与 trace 关联的实验 ID。如果未提供，则使用当前活动的实验。
start_time_ns – span 的开始时间（以纳秒为单位）。如果未提供，则使用当前时间。
- 返回
一个
mlflow.entities.Span对象，代表创建的 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]
注意
参数
request_id已弃用。请使用trace_id。如果 trace 存在，则通过给定的请求 ID 获取 trace。
此函数首先从内存缓冲区检索 trace，如果不存在，则从跟踪存储中获取 trace。如果跟踪存储中未找到 trace，则返回 None。
- 参数
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)
- 返回
具有给定请求 ID 的
mlflow.entities.Trace对象。
- mlflow.search_traces(experiment_ids: list[str] | None = None, filter_string: str | None = None, max_results: int | None = None, order_by: list[str] | None = None, extract_fields: list[str] | None = None, run_id: str | None = None, return_type: Literal['pandas', 'list'] | None = None, model_id: str | None = None, sql_warehouse_id: str | None = None, include_spans: bool = True, locations: list[str] | None = None) 'pandas.DataFrame' | list[Trace][source]
注意
参数
experiment_ids已弃用。请使用locations。在实验中返回与给定搜索表达式列表匹配的 trace。
注意
如果预期的搜索结果数量很大，请考虑直接使用 MlflowClient.search_traces API 来分页浏览结果。此函数在内存中返回所有结果，可能不适合大型结果集。
- 参数
experiment_ids – 要限定搜索范围的实验 ID 列表。
filter_string – 搜索过滤器字符串。
max_results – 所需 trace 的最大数量。如果为 None，则返回所有匹配搜索表达式的 trace。
order_by – order_by 子句列表。
extract_fields –
自 3.6.0 版本起已弃用： 此参数已弃用，将在将来的版本中移除。
使用格式
"span_name.[inputs|outputs].field_name"或"span_name.[inputs|outputs]"指定要从 trace 中提取的字段。注意
此参数仅在 return_type 设置为 “pandas” 时支持。
例如，
"predict.outputs.result"从名为"predict"的 span 中检索输出"result"字段，而"predict.outputs"获取整个 outputs 字典，包括"result"和"explanation"键。默认情况下，不会将任何字段提取到 DataFrame 列中。当指定多个字段时，每个字段将作为自己的列提取。如果提供了无效字段字符串，函数将静默返回而不添加该字段的列。支持的字段仅限于 span 的
"inputs"和"outputs"。如果 span 名称或字段名称包含点，则必须用反引号括起来。例如：# span name contains a dot extract_fields = ["`span.name`.inputs.field"] # field name contains a dot extract_fields = ["span.inputs.`field.name`"] # span name and field name contain a dot extract_fields = ["`span.name`.inputs.`field.name`"]
run_id – 要限定搜索范围的 run ID。当 trace 在活动 run 下创建时，它将与 run 关联，您可以按 run ID 过滤以检索 trace。有关如何按 run ID 过滤 trace 的示例，请参阅以下示例。
return_type –
返回值类型。支持以下返回值类型。如果安装了 pandas 库，则默认返回值类型为 “pandas”。否则，默认返回值类型为 “list”。
- ”pandas”：返回包含 trace 信息的 Pandas DataFrame。
其中每一行代表一个 trace，每一列代表 trace 的一个字段，例如 trace_id、spans 等。
”list”：返回
Trace对象列表。
model_id – 如果指定，则搜索与给定模型 ID 关联的 trace。
sql_warehouse_id – 已弃用。请改用 MLFLOW_TRACING_SQL_WAREHOUSE_ID 环境变量。用于在推断表或 UC 表中搜索 trace 的 SQL warehouse ID。仅在 Databricks 中使用。
include_spans – 如果为
True，则在返回的 trace 中包含 span。否则，仅返回 trace 元数据，例如 trace ID、开始时间、结束时间等，不包含任何 span。默认为True。locations – 要搜索的位置列表。要搜索实验，请提供实验 ID 列表。要在 Databricks 上搜索 UC 表，请提供格式为 <catalog_name>.<schema_name> 的位置列表。如果未提供，则搜索将在当前活动实验中进行。
- 返回
满足搜索表达式的 trace。根据 return_type 参数的值，可以是
Trace对象列表或 Pandas DataFrame。
import mlflow with mlflow.start_span(name="span1") as span: span.set_inputs({"a": 1, "b": 2}) span.set_outputs({"c": 3, "d": 4}) mlflow.search_traces( extract_fields=["span1.inputs", "span1.outputs", "span1.outputs.c"], return_type="pandas", )
import mlflow with mlflow.start_span(name="non_dict_span") as span: span.set_inputs(["a", "b"]) span.set_outputs([1, 2, 3]) mlflow.search_traces( extract_fields=["non_dict_span.inputs", "non_dict_span.outputs"], )
- mlflow.get_current_active_span() LiveSpan | None[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()
- 返回
当前活动的 span（如果存在），否则为 None。
- mlflow.get_last_active_trace_id(thread_local: bool = False) str | None[source]
获取同一进程中最后一个活动的 trace（如果存在）。
警告
此函数默认不是线程安全的，它返回同一进程中最后一个活动的 trace。如果要获取当前线程中最后一个活动的 trace，请将 thread_local 参数设置为 True。
- 参数
thread_local – 如果为 True，则返回当前线程中最后一个活动的 trace。否则，返回同一进程中最后一个活动的 trace。默认为 False。
- 返回
最后一个活动的 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 自动检测的远程服务时特别有用。通过使用此函数，您可以将远程服务的 trace 合并到当前活动的本地 trace 中，这样您就可以看到包含远程服务调用内部发生情况的完整 trace。
以下示例演示了如何使用此函数将来自远程服务的 trace 合并到函数中当前的活动 trace。
@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)
如果您有要合并 trace 的特定目标 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)
- 参数
trace –
一个
Trace对象或 trace 的字典表示。trace **必须**已经完成，即不应对其进行任何进一步的更新。否则，此函数将引发异常。target –
要合并给定 trace 的目标 span。
如果提供了此参数，则 trace 将在目标 span 下合并。
如果未提供，则 trace 将在当前活动的 span 下合并。
如果未提供且没有活动 span，则将创建一个名为“Remote Trace <…>”的新 span，并将 trace 合并到其下。
- mlflow.log_assessment(trace_id: str, assessment: Assessment) Assessment[source]
将评估记录到 Trace。评估可以是期望值或反馈。
- 期望值：表示某个操作预期值的标签。
例如，聊天机器人用户问题的预期答案。
- 反馈：表示操作质量反馈的标签。
反馈可以来自不同来源，例如人工评审员、启发式评分器或 LLM 作为评判者。
以下代码使用 LLM 作为评判者提供的反馈来注释 trace。
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)
- 期望值可以是任何可 JSON 序列化的值。例如，您可以
将完整的 LLM 消息作为期望值记录。
import mlflow from mlflow.entities.assessment import Expectation expectation = Expectation( name="expected_message", # Full LLM message including expected tool calls value={ "role": "assistant", "content": "The answer is 42.", "tool_calls": [ { "id": "1234", "type": "function", "function": {"name": "add", "arguments": "40 + 2"}, } ], }, ) mlflow.log_assessment(trace_id="1234", assessment=expectation)
您还可以在反馈生成过程中记录错误信息。为此，请将
AssessmentError的实例提供给 error 参数，并将 value 参数留空。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 仅接受关键字参数。
- 参数
trace_id – 跟踪的 ID。
name – 期望评估的名称，例如，“expected_answer”
value – 期望的值。它可以是任何可 JSON 序列化的值。
source – 期望评估的来源。必须是
AssessmentSource的实例。如果未提供，则默认为 HUMAN 源类型。metadata – 期望的附加元数据。
span_id – 与期望关联的 span 的 ID，如果需要将其与 trace 中的特定 span 相关联。
- 返回
创建的期望评估。
- 返回类型
示例
import mlflow from mlflow.entities import AssessmentSource, AssessmentSourceType # Log simple expected answer expectation = mlflow.log_expectation( trace_id="tr-1234567890abcdef", name="expected_answer", value="The capital of France is Paris.", source=AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="annotator@company.com" ), metadata={"question_type": "factual", "difficulty": "easy"}, ) # Log expected classification label mlflow.log_expectation( trace_id="tr-1234567890abcdef", name="expected_category", value="positive", source=AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="data_labeler_001" ), metadata={"labeling_session": "batch_01", "confidence": 0.95}, )
- mlflow.log_feedback(*, trace_id: str, name: str = 'feedback', value: Optional[Union[float, int, str, bool, dict[str, float | int | str | bool], list[float | int | str | bool]]] = None, source: Optional[AssessmentSource] = None, error: Optional[Union[Exception, AssessmentError]] = None, rationale: Optional[str] = None, metadata: Optional[dict[str, typing.Any]] = None, span_id: Optional[str] = None) Assessment[source]
将反馈记录到 Trace。此 API 仅接受关键字参数。
- 参数
trace_id – 跟踪的 ID。
name – 反馈评估的名称，例如，“faithfulness”。如果未提供，则默认为“feedback”。
value – 反馈的值。必须是以下类型之一： - float - int - str - bool - 上述相同类型的列表 - 具有字符串键和上述相同类型的值的字典
source – 反馈评估的来源。必须是
AssessmentSource的实例。如果未提供，则默认为 CODE 源类型。error – 一个错误对象，表示在计算反馈时遇到的任何问题，例如 LLM 评估器的超时错误。接受异常对象或
AssessmentError对象。必须提供此项或 value 中的一项。rationale – 反馈的理由/依据。
metadata – 反馈的附加元数据。
span_id – 与反馈关联的 span 的 ID，如果需要将其与 trace 中的特定 span 相关联。
- 返回
创建的反馈评估。
- 返回类型
示例
import mlflow from mlflow.entities import AssessmentSource, AssessmentSourceType # Log simple feedback score feedback = mlflow.log_feedback( trace_id="tr-1234567890abcdef", name="relevance", value=0.9, source=AssessmentSource( source_type=AssessmentSourceType.LLM_JUDGE, source_id="gpt-4" ), rationale="Response directly addresses the user's question", ) # Log detailed feedback with structured data mlflow.log_feedback( trace_id="tr-1234567890abcdef", name="quality_metrics", value={"accuracy": 0.95, "completeness": 0.88, "clarity": 0.92, "overall": 0.92}, source=AssessmentSource( source_type=AssessmentSourceType.HUMAN, source_id="expert_evaluator" ), rationale="High accuracy and clarity, slightly incomplete coverage", )
- mlflow.update_assessment(trace_id: str, assessment_id: str, assessment: Assessment) Assessment[source]
更新 Trace 中现有的期望（地面真实）。
- 参数
trace_id – 跟踪的 ID。
assessment_id – 要更新的期望或反馈评估的 ID。
assessment – 更新后的评估。
- 返回
更新后的反馈或期望评估。
- 返回类型
示例
以下代码使用新值更新现有期望。要更新其他字段，请提供相应的参数。
import mlflow from mlflow.entities import Expectation, ExpectationValue # Create an expectation with value 42. response = mlflow.log_assessment( trace_id="1234", assessment=Expectation(name="expected_answer", value=42), ) assessment_id = response.assessment_id # Update the expectation with a new value 43. mlflow.update_assessment( trace_id="1234", assessment_id=assessment.assessment_id, assessment=Expectation(name="expected_answer", value=43), )
- mlflow.delete_assessment(trace_id: str, assessment_id: str)[source]
删除与 trace 关联的评估。
- 参数
trace_id – 跟踪的 ID。
assessment_id – 要删除的评估的 ID。
- mlflow.tracing.configure(span_processors: list[typing.Callable[[ForwardRef('LiveSpan')], NoneType]] | None = None) mlflow.tracing.config.TracingConfigContext[source]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
配置 MLflow 跟踪。可用作函数或上下文管理器。
仅更新显式提供的参数，不更改其他参数。
- 参数
span_processors – 在导出之前处理 spans 的函数列表。这对于过滤/屏蔽 span 的特定属性以防止敏感数据被记录或减小 span 的大小很有用。每个函数必须接受一个 LiveSpan 类型的参数，并且不应返回任何值。当提供多个函数时，它们会按提供的顺序依次应用。
- 返回
- 用于临时配置更改的上下文管理器。
当用作函数时，配置更改将保留。当用作上下文管理器时，更改将在退出时恢复。
- 返回类型
TracingConfigContext
示例
def pii_filter(span): """Example PII filter that masks sensitive data in span attributes.""" # Mask sensitive inputs if inputs := span.inputs: for key, value in inputs.items(): if "password" in key.lower() or "token" in key.lower(): span.set_inputs({**inputs, key: "[REDACTED]"}) # Mask sensitive outputs if outputs := span.outputs: if isinstance(outputs, dict): for key in outputs: if "secret" in key.lower(): outputs[key] = "[REDACTED]" span.set_outputs(outputs) # Mask sensitive attributes for attr_key in list(span.attributes.keys()): if "api_key" in attr_key.lower(): span.set_attribute(attr_key, "[REDACTED]") # Permanent configuration change mlflow.tracing.configure(span_processors=[pii_filter]) # Temporary configuration change with mlflow.tracing.configure(span_processors=[pii_filter]): # PII filtering enabled only in this block pass
- mlflow.tracing.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]
禁用在 notebook 输出单元格中显示 MLflow Trace UI。调用
mlflow.tracing.enable_notebook_display()可重新启用显示。
- mlflow.tracing.enable()[source]
启用跟踪。
示例
import mlflow @mlflow.trace def f(): return 0 # Tracing is enabled by default f() assert len(mlflow.search_traces()) == 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]
在 notebook 输出单元格中启用 MLflow Trace UI。显示默认为开启，当执行以下任何操作时，Trace UI 将显示：
在 trace 完成时（即，每次导出 trace 时）
调用
mlflow.search_traces()流式 API 时调用
mlflow.client.MlflowClient.get_trace()或mlflow.client.MlflowClient.search_traces()客户端 API 时
- mlflow.tracing.reset()[source]
重置指示 MLflow 跟踪器提供程序是否已初始化的标志。这确保在执行下一个跟踪操作时重新初始化跟踪器提供程序。
- mlflow.tracing.set_databricks_monitoring_sql_warehouse_id(sql_warehouse_id: str, experiment_id: Optional[str] = None) None[source]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
设置用于 Databricks 生产监控的 SQL warehouse ID，以跟踪记录到给定 MLflow 实验的 trace。这仅对具有 UC schema 作为 trace 位置的实验有效。
- 参数
sql_warehouse_id – 要用于监控的 SQL warehouse ID。
experiment_id – MLflow 实验 ID。如果未提供，将使用当前活动实验。
- mlflow.tracing.set_destination(destination: mlflow.entities.trace_location.TraceLocationBase, *, context_local: bool = False)[source]
设置一个自定义 span 位置，MLflow 将跟踪数据导出到该位置。
此函数指定的 destination 将优先于其他配置，例如 tracking URI、OTLP 环境变量。
- 参数
destination –
一个 trace 位置对象，用于指定 trace 数据的来源。目前，支持以下位置：
MlflowExperimentLocation：将 trace 记录到MLflow 实验。
UCSchemaLocation：将 trace 记录到Databricks Unity Catalog schema。仅在 Databricks 中可用。
context_local – 如果为 False（默认值），则 destination 是全局设置的。如果为 True，则 destination 在每个异步任务或线程之间隔离，从而在并发应用程序中提供隔离。
示例
将 trace 记录到 MLflow 实验
from mlflow.entities.trace_location import MlflowExperimentLocation mlflow.tracing.set_destination(MlflowExperimentLocation(experiment_id="123"))
注意：这与通过
MLFLOW_EXPERIMENT_ID环境变量或mlflow.set_experimentAPI 设置活动 MLflow 实验具有相同的作用，但范围更窄。将 trace 记录到 Databricks Unity Catalog
from mlflow.entities.trace_location import UCSchemaLocation mlflow.tracing.set_destination( UCSchemaLocation(catalog_name="catalog", schema_name="schema") )
在异步任务或线程之间隔离 destination
from mlflow.tracing.destination import Databricks mlflow.tracing.set_destination( MlflowExperimentLocation(experiment_id="123"), context_local=True, )
使用
context_local标志设置的 destination 仅在当前异步任务或线程内有效。这在您希望从多租户应用程序将 trace 发送到不同 destination 时特别有用。**重置 destination：**
mlflow.tracing.reset()
- mlflow.tracing.set_experiment_trace_location(location: UCSchemaLocation, experiment_id: Optional[str] = None, sql_warehouse_id: Optional[str] = None) UCSchemaLocation[source]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
配置实验 trace 的存储位置。
用于存储 trace 数据的 Unity Catalog 表将在指定的 schema 中创建。启用跟踪时，指定实验的所有 trace 都将存储在提供的 Unity Catalog schema 中。
注意
如果实验已链接到存储位置，则会引发错误。使用 mlflow.tracing.unset_experiment_trace_location 删除现有存储位置，然后设置新位置。
- 参数
location – 实验 trace 在 Unity Catalog 中的存储位置。
experiment_id – 要为其设置存储位置的 MLflow 实验 ID。如果未指定，将使用当前活动实验。
sql_warehouse_id – 用于创建视图和查询的 SQL warehouse ID。如果未指定，将使用 MLFLOW_TRACING_SQL_WAREHOUSE_ID 的值，如果未设置环境变量，则回退到默认 SQL warehouse。
- 返回
表示已配置的存储位置的 UCSchemaLocation 对象，包括 spans 和 logs 表的表名。
示例
import mlflow from mlflow.entities import UCSchemaLocation location = UCSchemaLocation(catalog_name="my_catalog", schema_name="my_schema") result = mlflow.tracing.set_experiment_trace_location( location=location, experiment_id="12345", ) print(result.full_otel_spans_table_name) # my_catalog.my_schema.otel_spans_table @mlflow.trace def add(x): return x + 1 add(1) # this writes the trace to the storage location set above
- mlflow.tracing.set_span_chat_tools(span: LiveSpan, tools: list[ChatTool])[source]
在指定的 span 上设置 mlflow.chat.tools 属性。此属性在 UI 中使用，也用于消耗 trace 数据的下游应用程序，例如 MLflow evaluate。
- 参数
span – 要添加属性的 LiveSpan。
tools – 标准化聊天工具定义的列表（有关详细信息，请参阅 规范）。
示例
import mlflow from mlflow.tracing import set_span_chat_tools tools = [ { "type": "function", "function": { "name": "add", "description": "Add two numbers", "parameters": { "type": "object", "properties": { "a": {"type": "number"}, "b": {"type": "number"}, }, "required": ["a", "b"], }, }, } ] @mlflow.trace def f(): span = mlflow.get_current_active_span() set_span_chat_tools(span, tools) return 0 f()
- mlflow.tracing.unset_experiment_trace_location(location: UCSchemaLocation, experiment_id: Optional[str] = None) None[source]
注意
实验性功能：此函数可能会在未来版本中更改或移除，恕不另行通知。
取消设置实验 trace 的存储位置。
此函数删除实验存储位置配置，包括视图和实验标签。
- 参数
location – 要取消设置的存储位置。
experiment_id – 要为其取消设置存储位置的 MLflow 实验 ID。如果未提供，将使用当前活动实验。
示例
import mlflow from mlflow.entities import UCSchemaLocation mlflow.tracing.unset_experiment_trace_location( location=UCSchemaLocation(catalog_name="my_catalog", schema_name="my_schema"), experiment_id="12345", )
MLflow 记录模型 API
mlflow 模块提供了一组高级 API 来与 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，其 artifact 存储在 MLflow 外部。这对于跟踪未通过 MLflow Model 格式打包的模型、应用程序或生成式 AI 代理的参数和性能数据（指标、trace 等）非常有用。
- 参数
name – 模型的名称。如果未指定，将生成一个随机名称。
source_run_id – 模型关联的 run 的 ID。如果未指定且存在活动 run，将使用活动 run ID。
tags – 要设置为模型标签的字符串键值对字典。
params – 要设置为模型参数的字符串键值对字典。
model_type – 模型的类型。这是一个用户定义的字符串，可用于搜索和比较相关模型。例如，设置
model_type="agent"使您能够轻松搜索此模型并将来将其与类型为"agent"的其他模型进行比较。experiment_id – 模型所属实验的实验 ID。
- 返回
一个状态为
READY的新mlflow.entities.LoggedModel对象。
- mlflow.delete_logged_model_tag(model_id: str, key: str) None[source]
从指定的已记录模型中删除标签。
- 参数
model_id – 模型的 ID。
key – 要删除的标签键。
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) mlflow.set_logged_model_tags(model_info.model_id, {"key": "value"}) model = mlflow.get_logged_model(model_info.model_id) assert model.tags["key"] == "value" mlflow.delete_logged_model_tag(model_info.model_id, "key") model = mlflow.get_logged_model(model_info.model_id) assert "key" not in model.tags
- mlflow.finalize_logged_model(model_id: str, status: Union[Literal['READY', 'FAILED'], LoggedModelStatus]) LoggedModel[source]
通过更新其状态来完成模型。
- 参数
model_id – 要完成的模型的 ID。
status – 要在模型上设置的最终状态。
- 返回
更新后的模型。
示例
import mlflow from mlflow.entities import LoggedModelStatus model = mlflow.initialize_logged_model(name="model") logged_model = mlflow.finalize_logged_model( model_id=model.model_id, status=LoggedModelStatus.READY, ) assert logged_model.status == LoggedModelStatus.READY
- mlflow.get_logged_model(model_id: str) LoggedModel[source]
按 ID 获取已记录的模型。
- 参数
model_id – 已记录模型的 ID。
- 返回
已记录的模型。
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) logged_model = mlflow.get_logged_model(model_id=model_info.model_id) assert logged_model.model_id == model_info.model_id
- mlflow.initialize_logged_model(name: Optional[str] = None, source_run_id: Optional[str] = None, tags: Optional[dict[str, str]] = None, params: Optional[dict[str, str]] = None, model_type: Optional[str] = None, experiment_id: Optional[str] = None) LoggedModel[source]
初始化一个 LoggedModel。创建一个状态为
PENDING且无构件的 LoggedModel。您必须向模型添加构件并将其最终确定为READY状态，例如通过调用特定于 flavor 的log_model()方法，如mlflow.pyfunc.log_model()。- 参数
name – 模型的名称。如果未指定，将生成一个随机名称。
source_run_id – 模型关联的 run 的 ID。如果未指定且存在活动 run，将使用活动 run ID。
tags – 要设置为模型标签的字符串键值对字典。
params – 要设置为模型参数的字符串键值对字典。
model_type – 模型的类型。
experiment_id – 模型所属实验的实验 ID。
- 返回
一个新的
mlflow.entities.LoggedModel对象，状态为PENDING。
- mlflow.last_logged_model() LoggedModel | None[source]
获取当前会话中最近一个已记录的模型。如果没有记录模型，则返回 None。
- 返回
已记录的模型。
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) last_model = mlflow.last_logged_model() assert last_model.model_id == model_info.model_id
- mlflow.search_logged_models(experiment_ids: list[str] | None = None, filter_string: str | None = None, datasets: list[dict[str, str]] | None = None, max_results: int | None = None, order_by: list[dict[str, typing.Any]] | None = None, output_format: Literal['pandas'] = 'pandas') pandas.DataFrame[source]
- mlflow.search_logged_models(experiment_ids: list[str] | None = None, filter_string: str | None = None, datasets: list[dict[str, str]] | None = None, max_results: int | None = None, order_by: list[dict[str, typing.Any]] | None = None, output_format: Literal['list'] = 'list') list[LoggedModel]
根据指定的搜索条件搜索已记录的模型。
- 参数
experiment_ids – 要搜索已记录模型的实验 ID 列表。如果未指定，则使用当前活动实验。
filter_string –
要解析的类 SQL 筛选字符串。筛选字符串语法支持
- 实体规范
属性：attribute_name（如果未指定前缀，则为默认值）
指标：metrics.metric_name
参数：params.param_name
标签：tags.tag_name
- 比较运算符
对于数字实体（指标和数字属性）：<, <=, >, >=, =, !=
对于字符串实体（参数、标签、字符串属性）：=, !=, IN, NOT IN
多个条件可以用“AND”连接
字符串值必须用单引号括起来
- 示例筛选字符串
creation_time > 100
metrics.rmse > 0.5 AND params.model_type = ‘rf’
tags.release IN (‘v1.0’, ‘v1.1’)
params.optimizer != ‘adam’ AND metrics.accuracy >= 0.9
datasets –
字典列表，用于指定要应用指标筛选器的数据集。例如，带有 metrics.accuracy > 0.9 和数据集名称“test_dataset”的筛选字符串意味着我们将返回在 test_dataset 上准确度 > 0.9 的所有已记录模型。将考虑来自满足条件的任何数据集的指标值。如果未指定数据集，则将在所有数据集的指标中进行筛选。支持以下字段
- dataset_name (str)
必需。数据集的名称。
- dataset_digest (str)
可选。数据集的摘要。
max_results – 要返回的最大已记录模型数量。
order_by –
字典列表，用于指定搜索结果的排序。支持以下字段
- field_name (str)
必需。用于排序的字段名称，例如“metrics.accuracy”。
- ascending (bool)
可选。排序是升序还是降序。
- dataset_name (str)
可选。如果
field_name指的是指标，则此字段指定与该指标关联的数据集的名称。只有与指定数据集名称关联的指标才会用于排序。仅当field_name指的是指标时，才能设置此字段。- dataset_digest (str)
可选。如果
field_name指的是指标，则此字段指定与该指标关联的数据集的摘要。只有与指定数据集名称和摘要关联的指标才会用于排序。仅当也设置了dataset_name时，才能设置此字段。
output_format – 搜索结果的输出格式。支持的值为“pandas”和“list”。
- 返回
指定输出格式的搜索结果。
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) another_model_info = mlflow.pyfunc.log_model( name="another_model", python_model=DummyModel() ) models = mlflow.search_logged_models(output_format="list") assert [m.name for m in models] == ["another_model", "model"] models = mlflow.search_logged_models( filter_string="name = 'another_model'", output_format="list" ) assert [m.name for m in models] == ["another_model"] models = mlflow.search_logged_models( order_by=[{"field_name": "creation_time", "ascending": True}], output_format="list" ) assert [m.name for m in models] == ["model", "another_model"]
- mlflow.set_active_model(*, name: Optional[str] = None, model_id: Optional[str] = None) ActiveModel[source]
设置具有指定名称或模型 ID 的活动模型，它将用于链接在模型生命周期中生成的跟踪。返回值可以用作
with块内的上下文管理器；否则，您必须调用set_active_model()来更新活动模型。- 参数
name – 要设置为活动状态的
mlflow.entities.LoggedModel的名称。如果具有该名称的 LoggedModel 不存在，则将在当前实验下创建。如果存在多个同名 LoggedModel，则将最新的一个设置为活动状态。model_id – 要设置为活动状态的
mlflow.entities.LoggedModel的 ID。如果不存在具有该 ID 的 LoggedModel，则会引发异常。
- 返回
一个
mlflow.ActiveModel对象，它充当一个上下文管理器，包装 LoggedModel 的状态。
import mlflow # Set the active model by name mlflow.set_active_model(name="my_model") # Set the active model by model ID model = mlflow.create_external_model(name="test_model") mlflow.set_active_model(model_id=model.model_id) # Use the active model in a context manager with mlflow.set_active_model(name="new_model"): print(mlflow.get_active_model_id()) # Traces are automatically linked to the active model mlflow.set_active_model(name="my_model") @mlflow.trace def predict(model_input): return model_input predict("abc") traces = mlflow.search_traces(model_id=mlflow.get_active_model_id(), return_type="list") assert len(traces) == 1
- mlflow.set_logged_model_tags(model_id: str, tags: dict[str, typing.Any]) None[source]
为指定的已记录模型设置标签。
- 参数
model_id – 模型的 ID。
tags – 要在模型上设置的标签。
- 返回
无
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) mlflow.set_logged_model_tags(model_info.model_id, {"key": "value"}) model = mlflow.get_logged_model(model_info.model_id) assert model.tags["key"] == "value"
- mlflow.log_model_params(params: dict[str, str], model_id: Optional[str] = None) None[source]
将参数记录到指定的已记录模型。
- 参数
params – 要在模型上记录的参数。
model_id – 模型的 ID。如果未指定，则使用当前活动模型 ID。
- 返回
无
示例
import mlflow class DummyModel(mlflow.pyfunc.PythonModel): def predict(self, context, model_input: list[str]) -> list[str]: return model_input model_info = mlflow.pyfunc.log_model(name="model", python_model=DummyModel()) mlflow.log_model_params(params={"param": "value"}, model_id=model_info.model_id)