尧图网站建设 
ZenML 历史 Pipeline 运行检查指南从 Pipeline 到 Artifact 的完整获取链路【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml本文基于 ZenML 教程 fetching-pipelines 展开讲解如何检查一条已经完成或正在进行的 Pipeline 运行及其产出物先用Client或 CLI 列出并获取 Pipeline再逐层深入 Run、Step 与 Artifact最终把训练好的模型或数据集直接load()回代码中使用。读完后你可以独立完成找回昨天的模型、复现某次运行的配置、对比历史运行指标这类日常 MLOps 操作并且能从源码层面理解每个属性背后的实现。前置条件开始之前请确保已安装并配置好 ZenML至少有一条 Pipeline 被成功执行过一次本教程的所有操作都面向历史运行对 ZenML 的 Pipeline 与 Step 基本概念有初步了解。理解对象层级Pipeline → Run → Step → ArtifactZenML 中历史产物的层级结构是一条清晰的 1 对 N 链一条 Pipeline 可以被执行多次产生多个Run每个 Run 包含若干Step每个 Step 又有若干输入/输出Artifact。后文将沿这条层级自上而下逐层演示获取方法。第一步获取 Pipeline通过Client.get_pipeline()获取单个 PipelinePipeline 至少运行过一次后即可通过Client拿到它的响应对象from zenml.client import Client pipeline_model Client().get_pipeline(first_pipeline)从源码 client.py 可以看到get_pipeline()的签名是get_pipeline(name_id_or_prefix, projectNone, hydrateTrue)内部走_get_entity_by_id_or_name_or_prefix也就是说你可以传名称、完整 ID 或名称前缀三种形式定位 Pipeline且默认hydrateTrue会在响应中携带元数据字段。发现并列出所有 Pipeline不确定要获取哪条 Pipeline 时可以在 ZenML Dashboard 中查看也可以用 Python 或 CLI 程序化列出PythonClient.list_pipelines()from zenml.client import Client pipelines Client().list_pipelines() # 展示每条 Pipeline 的基本信息 for pipeline_model in pipelines: print(fPipeline: {pipeline_model.name}) print(- * 40)CLIzenml pipeline list深入看 client.py 中的实现list_pipelines()的完整参数如下默认值即源码默认值参数默认值说明sort_bycreated排序字段page/size1 / 默认页大小分页控制nameNone按名称过滤支持StringFilterOption如startswith:created/updatedNone按创建/更新时间过滤latest_run_statusNone按该 Pipeline最近一次运行的状态过滤latest_run_userNone按最近一次运行的用户过滤tagsNone按标签过滤hydrateFalse是否在水合响应中包含元数据字段其中latest_run_status很实用例如Client().list_pipelines(latest_run_statusfailed)可以直接找出所有最近一次跑挂了的 Pipeline。第二步访问 Pipeline 运行Runs每条 Pipeline 可以被执行多次产生多个Run。获取某条 Pipeline 的所有运行通过 Pipeline 对象的runs属性获取runs pipeline_model.runs结果是一个按时间从新到旧排序的 Run 列表。从源码 pipeline.py 可知runs属性实际调用get_runs()而get_runs()会代理到Client().list_pipeline_runs(pipeline_idself.id, **kwargs)并返回该页的.items。这意味着pipeline_model.runs默认只取第一页20 条需要精细过滤或翻页时使用pipeline_model.get_runs()并透传参数例如# 按状态过滤 分页 runs pipeline_model.get_runs(statuscompleted, size50, page1)获取最近一次运行last_run/last_successful_runlast_run pipeline_model.last_run # 等价于 pipeline_model.runs[0] # 打印该运行的一些基本信息 print(fRun ID: {last_run.id}) print(fStatus: {last_run.status}) print(fCreated at: {last_run.created})实现细节见 pipeline.pylast_run本质是get_runs(size1)的第一条而last_successful_run是get_runs(statusExecutionStatus.COMPLETED, size1)即只统计成功完成的运行——如果最近几次都失败了用它可以直接拿到最后一次能用的运行。两者在找不到记录时都会抛出RuntimeError。此外还有num_runs属性源码可以直接返回该 Pipeline 的运行总数。执行 Pipeline 时直接拿到 Run 对象调用一条 pipeline 函数执行完会返回这次全新运行的响应对象run training_pipeline()注意方法调用瞬间返回的run是数据库中那一刻的快照——此时运行可能还在初始化各 step 尚未执行。要拿到最新状态需要从 Client 重新获取一份刷新后的对象from zenml.client import Client # 获取刷新后的运行状态 Client().get_pipeline_run(runs[0].id)通过Client.get_pipeline_run()直接获取某个 Run如果你已经知道确切的 Run例如从 Dashboard 上看到的可以跳过 Pipeline 直接取from zenml.client import Client pipeline_run Client().get_pipeline_run(first_pipeline-2023_06_20-16_20_13_274466)与 Pipeline 类似Run 也支持按ID、名称或名称前缀查询源码 中allow_name_prefix_match默认为True还有include_full_metadata参数可控制是否包含全部 step 元数据。发现 Run 则可以用Client.list_pipeline_runs()签名见 client.py支持按pipeline_name、pipeline_id、stack_id、created等维度过滤或 CLIzenml pipeline runs list第三步检查 Run 的关键信息每个 Run 都携带了大量可用于复现实验的信息。下面是最常用的几项完整的字段列表可参考PipelineRunResponse的模型定义位于 pipeline_run.py。状态 status一次 Pipeline 运行共有五种可能状态initialized已初始化、failed失败、completed完成、running运行中、cached已缓存。run runs[0] status run.status配置 configrun.config是一个包含该 Pipeline 与本次运行全部配置的对象其中也包括 pipeline 级设置如requirements、docker等pipeline_config run.config pipeline_settings run.config.settings # 示例检查是否配置了 Docker 设置 docker_settings pipeline_settings.get(docker, {}) print(fDocker settings: {docker_settings})保留这份配置意味着你随时可以回答上次那个运行到底用了什么参数、什么依赖。组件专属元数据 run_metadata根据你使用的 Stack 组件Run 上可能还挂有组件专属元数据例如某个远程 Orchestrator 的 UI 地址。通过run.run_metadata访问run_metadata run.run_metadata # 示例获取 Orchestrator UI 地址对部分远程 orchestrator 有效 if orchestrator_url in run_metadata: orchestrator_url run_metadata[orchestrator_url].value print(fOrchestrator UI URL: {orchestrator_url})run_metadata是键值字典每个键对应一个带.value属性的元数据项这也是下文 Step、Artifact 层元数据的统一访问范式。第四步深入 Step给定一次 Pipeline 运行可以继续下钻到具体 Step。从 pipeline_run.py 可知run.steps返回一个Dict[str, StepRunResponse]——键是 step 的 invocation ID# 获取该次运行的所有 step steps run.steps # 通过 invocation ID 获取某个具体 step step run.steps[first_step] # 打印每个 step 的信息 for step_name, step_info in steps.items(): print(fStep name: {step_name}) print(fStatus: {step_info.status}) print(fStarted at: {step_info.start_time}) print(fCompleted at: {step_info.end_time}) print(- * 40)关于 invocation ID如果你在 Pipeline 里每个 step 都只调用了一次invocation ID 就等于 step 函数名如果一个 step 被多次调用如不同参数复用同一 stepZenML 会自动生成不同的 invocation ID 加以区分复杂 Pipeline 需要留意这一点。Step 的参数、设置与元数据与 Run 类似step对象暴露了多种实用信息模型定义见 step_run.pystep.config.parameters该 step 运行时使用的参数step.config.settingsstep 级设置step.run_metadata组件专属元数据如 experiment tracker 或 model deployer 的 URL。# 获取一个具体 step step run.steps[trainer_step] # 访问 step 参数 parameters step.config.parameters print(fStep parameters: {parameters}) # 访问 step 设置 settings step.config.settings print(fStep settings: {settings}) # 访问 step 元数据 step_metadata step.run_metadata print(fStep metadata: {step_metadata})用 VS Code 扩展检查运行状态如果你使用 ZenML 的 VS Code 扩展打开侧边栏点击 ZenML 图标即可查看 Pipeline 运行点击任意一条运行可看到其状态与部分元数据还可以在同一侧边栏视图中删除运行——这比纯代码方式更适合日常扫一眼最近跑得怎么样的场景。第五步访问 Artifact每个 Step 都有输入/输出 Artifact通过outputs/inputs属性检查。访问输出与输入 Artifact# step 的输出可以按名字索引 output step.outputs[output_name] # 如果只有一个输出可以用 .output 属性作为快捷方式 output step.output # 用 .load() 把 artifact 加载进内存 my_pytorch_model output.load() # 打印 artifact 信息 print(fArtifact ID: {output.id}) print(fArtifact type: {output.type}) print(fArtifact version: {output.version})同样地输入 artifact 可以用inputs/input属性获取# 访问某个具体输入 artifact input_data step.inputs[input_name] # 如果只有一个输入使用快捷方式 input_data step.input # 加载输入数据 data input_data.load()从源码 step_run.py 可以看到两个值得注意的实现细节outputs属性返回的是Dict[str, List[ArtifactVersionResponse]]——每个输出名对应一个版本列表。因此在多输出或不确定顺序的场景下稳妥写法是取列表首元素step.outputs[trained_model][0].load()文末完整示例即采用这种写法单数快捷属性output/input在输入/输出数量为零或多个时会抛出ValueErroroutput 属性实现所以只在确定单输出的 step 上使用。每个 Step 的输出对应某个具体的 artifact版本load()的底层实现在 artifact_version.py它负责把存储中的数据物化materialize回 Python 对象并支持disable_cacheTrue跳过缓存。如果你不确定自己 step 的输出名可以通过在step函数返回值上使用Annotated类型来显式命名输出也可以先运行下文常见问题中的排查代码列出所有可用名称。通过 Client 直接获取 Artifact如果只想直接拿某个 artifact 或其版本from zenml.client import Client # 获取 artifact artifact Client().get_artifact(iris_dataset) artifact.versions # 包含该 artifact 的所有版本 output artifact.versions[2022] # 取版本名为 2022 的版本 # 直接获取 artifact 版本 # 按版本名 output Client().get_artifact_version(iris_dataset, 2022) # 按 UUID output Client().get_artifact_version(f429f94c-fb15-43b5-961d-dbea287507c5) loaded_artifact output.load()对应实现分别位于 client.py 的 get_artifact 与 get_artifact_version同样遵循名称 / ID / 前缀的查询范式。Artifact 元数据 metadata凡是通过 ZenML 保存的输出 artifact都会自动附带类型专属元数据。例如 NumPy 数组会保存存储大小、shape、dtype以及若干统计属性。通过输出的run_metadata属性访问output_metadata output.run_metadata storage_size_in_bytes output_metadata[storage_size].value # 对于 numpy 数组访问 shape 和 dtype if shape in output_metadata: shape output_metadata[shape].value print(fArray shape: {shape}) if dtype in output_metadata: dtype output_metadata[dtype].value print(fData type: {dtype})可视化 visualizationsZenML 会为许多常见数据类型自动生成可视化在 Jupyter 中调用visualize()即可展示实现见 artifact_version.pyoutput.visualize()如果你不在 Jupyter 环境中可以运行zenml login --local后在 Dashboard 中打开对应 Pipeline Run 的 DAG点击具体 artifact 查看可视化效果。第六步在运行中的 Pipeline 内部获取历史信息前面都聚焦于运行完成后取回对象但同样的逻辑同样可以在step 内部使用。这在 Pipeline 持续运行、需要根据历史运行做决策时非常有用。例如在 step 中获取同一条 Pipeline 的上一次运行from zenml import get_step_context from zenml.client import Client step def my_step(): # 获取当前 pipeline 运行的名称 current_run_name get_step_context().pipeline_run.name # 获取当前 pipeline 运行 current_run Client().get_pipeline_run(current_run_name) # 获取同一条 pipeline 的上一次运行索引 0 是当前运行 previous_run current_run.pipeline.runs[1] # 基于上一次运行的数据做处理 # 例如对比当前运行与上一次的指标 if evaluator in previous_run.steps: prev_metrics previous_run.steps[evaluator].output.load() print(fPrevious run metrics: {prev_metrics})示例中通过StepContextget_step_context()获取当前运行的附加信息它封装了当前 step 所处的运行上下文是运行期自省的核心入口。完整可运行示例训练后回捞模型把以上全部内容串起来下面是一个完整示例定义一个用 Iris 数据集训练 sklearn SVC 的 Pipeline运行之后通过三种方式拿到 Run 对象再从svc_trainerstep 中load()出训练好的模型和准确率。from typing import Tuple, Annotated import pandas as pd from sklearn.datasets import load_iris from sklearn.model_selection import train_test_split from sklearn.base import ClassifierMixin from sklearn.svm import SVC from zenml import pipeline, step from zenml.client import Client step def training_data_loader() - Tuple[ Annotated[pd.DataFrame, X_train], Annotated[pd.DataFrame, X_test], Annotated[pd.Series, y_train], Annotated[pd.Series, y_test], ]: 加载 iris 数据集为 Pandas DataFrame / Series 元组。 iris load_iris(as_frameTrue) X_train, X_test, y_train, y_test train_test_split( iris.data, iris.target, test_size0.2, shuffleTrue, random_state42 ) return X_train, X_test, y_train, y_test step def svc_trainer( X_train: pd.DataFrame, y_train: pd.Series, gamma: float 0.001, ) - Tuple[ Annotated[ClassifierMixin, trained_model], Annotated[float, training_acc], ]: 训练一个 sklearn SVC 分类器并记录到 MLflow。 model SVC(gammagamma) model.fit(X_train.to_numpy(), y_train.to_numpy()) train_acc model.score(X_train.to_numpy(), y_train.to_numpy()) print(fTrain accuracy: {train_acc}) return model, train_acc pipeline def training_pipeline(gamma: float 0.002): X_train, X_test, y_train, y_test training_data_loader() svc_trainer(gammagamma, X_trainX_train, y_trainy_train) if __name__ __main__: # 如果尚未运行过先执行一次 pipeline training_pipeline(gamma0.005) # 方法 1直接执行 pipeline拿到返回的 run 对象 last_run training_pipeline() print(fLast run ID: {last_run.id}) # 方法 2直接通过 pipeline 的 model 对象取最近一次运行 last_run training_pipeline.model.last_run print(fLast run ID via model: {last_run.id}) # 方法 3执行结束后通过 Client 获取 pipeline Client().get_pipeline(training_pipeline) last_run pipeline.last_run print(fLast run ID via client: {last_run.id}) # 现在可以取出模型 trainer_step last_run.steps[svc_trainer] model trainer_step.outputs[trained_model][0].load() accuracy trainer_step.outputs[training_acc][0].load() print(fModel type: {type(model).__name__}) print(fModel parameters: {model.get_params()}) print(fTraining accuracy: {accuracy}) # 随后可用该模型做推理 # new_data ... # predictions model.predict(new_data)注意training_pipeline.model.last_run这一用法被pipeline装饰的函数对象上带有.model属性即 Pipeline 的响应对象无需Client也能直达最近一次运行是方法 3的简化形式。常见问题排查Run Not Found 错误如果提示找不到 Run先确认 Run ID 格式是否正确。Run 名称通常遵循pipeline_name-YYYY_MM_DD-HH_MM_SS_XXXXXX的格式可以先列出最近运行确认准确 ID# 确认使用了正确的 run ID 格式 # Run ID 通常形如pipeline_name-YYYY_MM_DD-HH_MM_SS_XXXXXX # 列出最近的运行以找到正确的 ID recent_runs Client().list_pipeline_runs(size5) for run in recent_runs: print(fID: {run.id}, Created: {run.created})由于get_pipeline_run()支持名称前缀匹配把前缀逐步加长直到唯一也是定位运行的实用技巧。找不到正确的输出 Artifact 名不确定 step 的输出名时直接列出该 step 的全部输出键# 列出某个 step 的所有输出 step run.steps[step_name] print(fAvailable outputs: {list(step.outputs.keys())})下一步掌握了历史运行的检查与取回能力后你可以继续构建依赖上一次运行结果的 Pipeline如增量训练、漂移监控在不同实验配置之间生成对比报告加载训练好的模型用于评估或部署跨多次运行提取并分析指标结合 user-guide/tutorial 目录见 tutorial 目录下的超参数调优、数据集管理、大数据处理等教程进一步扩展工作流。小结与源码索引本文的核心链路——Client().get_pipeline()→pipeline.last_run→run.steps[name]→step.outputs[name][0].load()——在仓库中的实现入口如下便于深入阅读能力源码位置list_pipelines/get_pipelinesrc/zenml/client.pyget_pipeline_run/list_pipeline_runssrc/zenml/client.pyget_artifact/get_artifact_versionsrc/zenml/client.pyruns/last_run/last_successful_runsrc/zenml/models/v2/core/pipeline.pyPipelineRunResponse.stepssrc/zenml/models/v2/core/pipeline_run.pyStepRunResponse.inputs/outputs/input/outputsrc/zenml/models/v2/core/step_run.pyArtifactVersionResponse.load/visualizesrc/zenml/models/v2/core/artifact_version.py【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ZenML 历史 Pipeline 运行检查指南：从 Pipeline 到 Artifact 的完整获取链路
ZenML 历史 Pipeline 运行检查指南从 Pipeline 到 Artifact 的完整获取链路【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml本文基于 ZenML 教程 fetching-pipelines 展开讲解如何检查一条已经完成或正在进行的 Pipeline 运行及其产出物先用Client或 CLI 列出并获取 Pipeline再逐层深入 Run、Step 与 Artifact最终把训练好的模型或数据集直接load()回代码中使用。读完后你可以独立完成找回昨天的模型、复现某次运行的配置、对比历史运行指标这类日常 MLOps 操作并且能从源码层面理解每个属性背后的实现。前置条件开始之前请确保已安装并配置好 ZenML至少有一条 Pipeline 被成功执行过一次本教程的所有操作都面向历史运行对 ZenML 的 Pipeline 与 Step 基本概念有初步了解。理解对象层级Pipeline → Run → Step → ArtifactZenML 中历史产物的层级结构是一条清晰的 1 对 N 链一条 Pipeline 可以被执行多次产生多个Run每个 Run 包含若干Step每个 Step 又有若干输入/输出Artifact。后文将沿这条层级自上而下逐层演示获取方法。第一步获取 Pipeline通过Client.get_pipeline()获取单个 PipelinePipeline 至少运行过一次后即可通过Client拿到它的响应对象from zenml.client import Client pipeline_model Client().get_pipeline(first_pipeline)从源码 client.py 可以看到get_pipeline()的签名是get_pipeline(name_id_or_prefix, projectNone, hydrateTrue)内部走_get_entity_by_id_or_name_or_prefix也就是说你可以传名称、完整 ID 或名称前缀三种形式定位 Pipeline且默认hydrateTrue会在响应中携带元数据字段。发现并列出所有 Pipeline不确定要获取哪条 Pipeline 时可以在 ZenML Dashboard 中查看也可以用 Python 或 CLI 程序化列出PythonClient.list_pipelines()from zenml.client import Client pipelines Client().list_pipelines() # 展示每条 Pipeline 的基本信息 for pipeline_model in pipelines: print(fPipeline: {pipeline_model.name}) print(- * 40)CLIzenml pipeline list深入看 client.py 中的实现list_pipelines()的完整参数如下默认值即源码默认值参数默认值说明sort_bycreated排序字段page/size1 / 默认页大小分页控制nameNone按名称过滤支持StringFilterOption如startswith:created/updatedNone按创建/更新时间过滤latest_run_statusNone按该 Pipeline最近一次运行的状态过滤latest_run_userNone按最近一次运行的用户过滤tagsNone按标签过滤hydrateFalse是否在水合响应中包含元数据字段其中latest_run_status很实用例如Client().list_pipelines(latest_run_statusfailed)可以直接找出所有最近一次跑挂了的 Pipeline。第二步访问 Pipeline 运行Runs每条 Pipeline 可以被执行多次产生多个Run。获取某条 Pipeline 的所有运行通过 Pipeline 对象的runs属性获取runs pipeline_model.runs结果是一个按时间从新到旧排序的 Run 列表。从源码 pipeline.py 可知runs属性实际调用get_runs()而get_runs()会代理到Client().list_pipeline_runs(pipeline_idself.id, **kwargs)并返回该页的.items。这意味着pipeline_model.runs默认只取第一页20 条需要精细过滤或翻页时使用pipeline_model.get_runs()并透传参数例如# 按状态过滤 分页 runs pipeline_model.get_runs(statuscompleted, size50, page1)获取最近一次运行last_run/last_successful_runlast_run pipeline_model.last_run # 等价于 pipeline_model.runs[0] # 打印该运行的一些基本信息 print(fRun ID: {last_run.id}) print(fStatus: {last_run.status}) print(fCreated at: {last_run.created})实现细节见 pipeline.pylast_run本质是get_runs(size1)的第一条而last_successful_run是get_runs(statusExecutionStatus.COMPLETED, size1)即只统计成功完成的运行——如果最近几次都失败了用它可以直接拿到最后一次能用的运行。两者在找不到记录时都会抛出RuntimeError。此外还有num_runs属性源码可以直接返回该 Pipeline 的运行总数。执行 Pipeline 时直接拿到 Run 对象调用一条 pipeline 函数执行完会返回这次全新运行的响应对象run training_pipeline()注意方法调用瞬间返回的run是数据库中那一刻的快照——此时运行可能还在初始化各 step 尚未执行。要拿到最新状态需要从 Client 重新获取一份刷新后的对象from zenml.client import Client # 获取刷新后的运行状态 Client().get_pipeline_run(runs[0].id)通过Client.get_pipeline_run()直接获取某个 Run如果你已经知道确切的 Run例如从 Dashboard 上看到的可以跳过 Pipeline 直接取from zenml.client import Client pipeline_run Client().get_pipeline_run(first_pipeline-2023_06_20-16_20_13_274466)与 Pipeline 类似Run 也支持按ID、名称或名称前缀查询源码 中allow_name_prefix_match默认为True还有include_full_metadata参数可控制是否包含全部 step 元数据。发现 Run 则可以用Client.list_pipeline_runs()签名见 client.py支持按pipeline_name、pipeline_id、stack_id、created等维度过滤或 CLIzenml pipeline runs list第三步检查 Run 的关键信息每个 Run 都携带了大量可用于复现实验的信息。下面是最常用的几项完整的字段列表可参考PipelineRunResponse的模型定义位于 pipeline_run.py。状态 status一次 Pipeline 运行共有五种可能状态initialized已初始化、failed失败、completed完成、running运行中、cached已缓存。run runs[0] status run.status配置 configrun.config是一个包含该 Pipeline 与本次运行全部配置的对象其中也包括 pipeline 级设置如requirements、docker等pipeline_config run.config pipeline_settings run.config.settings # 示例检查是否配置了 Docker 设置 docker_settings pipeline_settings.get(docker, {}) print(fDocker settings: {docker_settings})保留这份配置意味着你随时可以回答上次那个运行到底用了什么参数、什么依赖。组件专属元数据 run_metadata根据你使用的 Stack 组件Run 上可能还挂有组件专属元数据例如某个远程 Orchestrator 的 UI 地址。通过run.run_metadata访问run_metadata run.run_metadata # 示例获取 Orchestrator UI 地址对部分远程 orchestrator 有效 if orchestrator_url in run_metadata: orchestrator_url run_metadata[orchestrator_url].value print(fOrchestrator UI URL: {orchestrator_url})run_metadata是键值字典每个键对应一个带.value属性的元数据项这也是下文 Step、Artifact 层元数据的统一访问范式。第四步深入 Step给定一次 Pipeline 运行可以继续下钻到具体 Step。从 pipeline_run.py 可知run.steps返回一个Dict[str, StepRunResponse]——键是 step 的 invocation ID# 获取该次运行的所有 step steps run.steps # 通过 invocation ID 获取某个具体 step step run.steps[first_step] # 打印每个 step 的信息 for step_name, step_info in steps.items(): print(fStep name: {step_name}) print(fStatus: {step_info.status}) print(fStarted at: {step_info.start_time}) print(fCompleted at: {step_info.end_time}) print(- * 40)关于 invocation ID如果你在 Pipeline 里每个 step 都只调用了一次invocation ID 就等于 step 函数名如果一个 step 被多次调用如不同参数复用同一 stepZenML 会自动生成不同的 invocation ID 加以区分复杂 Pipeline 需要留意这一点。Step 的参数、设置与元数据与 Run 类似step对象暴露了多种实用信息模型定义见 step_run.pystep.config.parameters该 step 运行时使用的参数step.config.settingsstep 级设置step.run_metadata组件专属元数据如 experiment tracker 或 model deployer 的 URL。# 获取一个具体 step step run.steps[trainer_step] # 访问 step 参数 parameters step.config.parameters print(fStep parameters: {parameters}) # 访问 step 设置 settings step.config.settings print(fStep settings: {settings}) # 访问 step 元数据 step_metadata step.run_metadata print(fStep metadata: {step_metadata})用 VS Code 扩展检查运行状态如果你使用 ZenML 的 VS Code 扩展打开侧边栏点击 ZenML 图标即可查看 Pipeline 运行点击任意一条运行可看到其状态与部分元数据还可以在同一侧边栏视图中删除运行——这比纯代码方式更适合日常扫一眼最近跑得怎么样的场景。第五步访问 Artifact每个 Step 都有输入/输出 Artifact通过outputs/inputs属性检查。访问输出与输入 Artifact# step 的输出可以按名字索引 output step.outputs[output_name] # 如果只有一个输出可以用 .output 属性作为快捷方式 output step.output # 用 .load() 把 artifact 加载进内存 my_pytorch_model output.load() # 打印 artifact 信息 print(fArtifact ID: {output.id}) print(fArtifact type: {output.type}) print(fArtifact version: {output.version})同样地输入 artifact 可以用inputs/input属性获取# 访问某个具体输入 artifact input_data step.inputs[input_name] # 如果只有一个输入使用快捷方式 input_data step.input # 加载输入数据 data input_data.load()从源码 step_run.py 可以看到两个值得注意的实现细节outputs属性返回的是Dict[str, List[ArtifactVersionResponse]]——每个输出名对应一个版本列表。因此在多输出或不确定顺序的场景下稳妥写法是取列表首元素step.outputs[trained_model][0].load()文末完整示例即采用这种写法单数快捷属性output/input在输入/输出数量为零或多个时会抛出ValueErroroutput 属性实现所以只在确定单输出的 step 上使用。每个 Step 的输出对应某个具体的 artifact版本load()的底层实现在 artifact_version.py它负责把存储中的数据物化materialize回 Python 对象并支持disable_cacheTrue跳过缓存。如果你不确定自己 step 的输出名可以通过在step函数返回值上使用Annotated类型来显式命名输出也可以先运行下文常见问题中的排查代码列出所有可用名称。通过 Client 直接获取 Artifact如果只想直接拿某个 artifact 或其版本from zenml.client import Client # 获取 artifact artifact Client().get_artifact(iris_dataset) artifact.versions # 包含该 artifact 的所有版本 output artifact.versions[2022] # 取版本名为 2022 的版本 # 直接获取 artifact 版本 # 按版本名 output Client().get_artifact_version(iris_dataset, 2022) # 按 UUID output Client().get_artifact_version(f429f94c-fb15-43b5-961d-dbea287507c5) loaded_artifact output.load()对应实现分别位于 client.py 的 get_artifact 与 get_artifact_version同样遵循名称 / ID / 前缀的查询范式。Artifact 元数据 metadata凡是通过 ZenML 保存的输出 artifact都会自动附带类型专属元数据。例如 NumPy 数组会保存存储大小、shape、dtype以及若干统计属性。通过输出的run_metadata属性访问output_metadata output.run_metadata storage_size_in_bytes output_metadata[storage_size].value # 对于 numpy 数组访问 shape 和 dtype if shape in output_metadata: shape output_metadata[shape].value print(fArray shape: {shape}) if dtype in output_metadata: dtype output_metadata[dtype].value print(fData type: {dtype})可视化 visualizationsZenML 会为许多常见数据类型自动生成可视化在 Jupyter 中调用visualize()即可展示实现见 artifact_version.pyoutput.visualize()如果你不在 Jupyter 环境中可以运行zenml login --local后在 Dashboard 中打开对应 Pipeline Run 的 DAG点击具体 artifact 查看可视化效果。第六步在运行中的 Pipeline 内部获取历史信息前面都聚焦于运行完成后取回对象但同样的逻辑同样可以在step 内部使用。这在 Pipeline 持续运行、需要根据历史运行做决策时非常有用。例如在 step 中获取同一条 Pipeline 的上一次运行from zenml import get_step_context from zenml.client import Client step def my_step(): # 获取当前 pipeline 运行的名称 current_run_name get_step_context().pipeline_run.name # 获取当前 pipeline 运行 current_run Client().get_pipeline_run(current_run_name) # 获取同一条 pipeline 的上一次运行索引 0 是当前运行 previous_run current_run.pipeline.runs[1] # 基于上一次运行的数据做处理 # 例如对比当前运行与上一次的指标 if evaluator in previous_run.steps: prev_metrics previous_run.steps[evaluator].output.load() print(fPrevious run metrics: {prev_metrics})示例中通过StepContextget_step_context()获取当前运行的附加信息它封装了当前 step 所处的运行上下文是运行期自省的核心入口。完整可运行示例训练后回捞模型把以上全部内容串起来下面是一个完整示例定义一个用 Iris 数据集训练 sklearn SVC 的 Pipeline运行之后通过三种方式拿到 Run 对象再从svc_trainerstep 中load()出训练好的模型和准确率。from typing import Tuple, Annotated import pandas as pd from sklearn.datasets import load_iris from sklearn.model_selection import train_test_split from sklearn.base import ClassifierMixin from sklearn.svm import SVC from zenml import pipeline, step from zenml.client import Client step def training_data_loader() - Tuple[ Annotated[pd.DataFrame, X_train], Annotated[pd.DataFrame, X_test], Annotated[pd.Series, y_train], Annotated[pd.Series, y_test], ]: 加载 iris 数据集为 Pandas DataFrame / Series 元组。 iris load_iris(as_frameTrue) X_train, X_test, y_train, y_test train_test_split( iris.data, iris.target, test_size0.2, shuffleTrue, random_state42 ) return X_train, X_test, y_train, y_test step def svc_trainer( X_train: pd.DataFrame, y_train: pd.Series, gamma: float 0.001, ) - Tuple[ Annotated[ClassifierMixin, trained_model], Annotated[float, training_acc], ]: 训练一个 sklearn SVC 分类器并记录到 MLflow。 model SVC(gammagamma) model.fit(X_train.to_numpy(), y_train.to_numpy()) train_acc model.score(X_train.to_numpy(), y_train.to_numpy()) print(fTrain accuracy: {train_acc}) return model, train_acc pipeline def training_pipeline(gamma: float 0.002): X_train, X_test, y_train, y_test training_data_loader() svc_trainer(gammagamma, X_trainX_train, y_trainy_train) if __name__ __main__: # 如果尚未运行过先执行一次 pipeline training_pipeline(gamma0.005) # 方法 1直接执行 pipeline拿到返回的 run 对象 last_run training_pipeline() print(fLast run ID: {last_run.id}) # 方法 2直接通过 pipeline 的 model 对象取最近一次运行 last_run training_pipeline.model.last_run print(fLast run ID via model: {last_run.id}) # 方法 3执行结束后通过 Client 获取 pipeline Client().get_pipeline(training_pipeline) last_run pipeline.last_run print(fLast run ID via client: {last_run.id}) # 现在可以取出模型 trainer_step last_run.steps[svc_trainer] model trainer_step.outputs[trained_model][0].load() accuracy trainer_step.outputs[training_acc][0].load() print(fModel type: {type(model).__name__}) print(fModel parameters: {model.get_params()}) print(fTraining accuracy: {accuracy}) # 随后可用该模型做推理 # new_data ... # predictions model.predict(new_data)注意training_pipeline.model.last_run这一用法被pipeline装饰的函数对象上带有.model属性即 Pipeline 的响应对象无需Client也能直达最近一次运行是方法 3的简化形式。常见问题排查Run Not Found 错误如果提示找不到 Run先确认 Run ID 格式是否正确。Run 名称通常遵循pipeline_name-YYYY_MM_DD-HH_MM_SS_XXXXXX的格式可以先列出最近运行确认准确 ID# 确认使用了正确的 run ID 格式 # Run ID 通常形如pipeline_name-YYYY_MM_DD-HH_MM_SS_XXXXXX # 列出最近的运行以找到正确的 ID recent_runs Client().list_pipeline_runs(size5) for run in recent_runs: print(fID: {run.id}, Created: {run.created})由于get_pipeline_run()支持名称前缀匹配把前缀逐步加长直到唯一也是定位运行的实用技巧。找不到正确的输出 Artifact 名不确定 step 的输出名时直接列出该 step 的全部输出键# 列出某个 step 的所有输出 step run.steps[step_name] print(fAvailable outputs: {list(step.outputs.keys())})下一步掌握了历史运行的检查与取回能力后你可以继续构建依赖上一次运行结果的 Pipeline如增量训练、漂移监控在不同实验配置之间生成对比报告加载训练好的模型用于评估或部署跨多次运行提取并分析指标结合 user-guide/tutorial 目录见 tutorial 目录下的超参数调优、数据集管理、大数据处理等教程进一步扩展工作流。小结与源码索引本文的核心链路——Client().get_pipeline()→pipeline.last_run→run.steps[name]→step.outputs[name][0].load()——在仓库中的实现入口如下便于深入阅读能力源码位置list_pipelines/get_pipelinesrc/zenml/client.pyget_pipeline_run/list_pipeline_runssrc/zenml/client.pyget_artifact/get_artifact_versionsrc/zenml/client.pyruns/last_run/last_successful_runsrc/zenml/models/v2/core/pipeline.pyPipelineRunResponse.stepssrc/zenml/models/v2/core/pipeline_run.pyStepRunResponse.inputs/outputs/input/outputsrc/zenml/models/v2/core/step_run.pyArtifactVersionResponse.load/visualizesrc/zenml/models/v2/core/artifact_version.py【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考