ZenML Artifact 管理实战制品命名、版本控制、元数据与跨管道消费的完整指南【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml数据是每个机器学习工作流的核心而正确管理和版本化这些数据是保障 ML 管道可复现性与可追溯性的前提。本文基于 ZenML 官方文档《Manage artifacts》docs/book/user-guide/starter-guide/manage-artifacts.md系统讲解 ZenML 制品Artifact的自动版本化机制、自定义命名与版本号、元数据与标签、制品类型、外部数据接入以及跨管道消费等全部核心能力并结合开源仓库源码剖析关键实现细节。读完本篇你将能够编写可追溯、可复现、可跨管道复用数据的 ZenML 管道。管道制品的自动版本化制品Artifact是你管道中 Step 和 Pipeline 的输出物。在 ZenML 中所有制品在管道执行时会被自动版本化并存入 artifact store。这一机制对数据、模型、评估结果等所有类型的输出物统一生效是构建透明、高效管道开发体验的基础。为制品赋予可读名称给制品赋予自定义名称可以显著提升其可发现性与可管理性。最佳实践是使用 Python 类型系统中的Annotated对象在 Step 输出上给出精确、人类可读的名称from typing import Annotated import pandas as pd from sklearn.datasets import load_iris from zenml import pipeline, step # Using Annotated to name our dataset step def training_data_loader() - Annotated[pd.DataFrame, iris_dataset]: Load the iris dataset as pandas dataframe. iris load_iris(as_frameTrue) return iris.get(frame) pipeline def feature_engineering_pipeline(): training_data_loader() if __name__ __main__: feature_engineering_pipeline()提示未显式指定名称的 Step 输出默认遵循{pipeline_name}::{step_name}::output的命名模式。如果希望在 Dashboard 中可视化地探索数据建议为重要的输出物赋予清晰的自定义名称。这个默认命名模式可以直接在源码中得到印证。在 Step 执行器 step_runner.py 中当ArtifactConfig未提供name时会拼装如下名称# Override the artifact name if it is not a custom name. if has_custom_name: artifact_name output_name else: if step_context.pipeline_run.pipeline: pipeline_name step_context.pipeline_run.pipeline.name else: pipeline_name unlisted step_name step_context.step_run.name artifact_name f{pipeline_name}::{step_name}::{output_name}可以看到若管道本身未被记录unlistedpipeline_name会回退为字符串unlisted。命名之后的制品可以通过多种 ZenML 接口快速找到OSSCLI使用zenml artifact list列出制品CloudDashboardZenML Pro 的 Dashboard 提供了更高级的制品可视化探索能力。提示为避免视觉上的杂乱请为最需要可视化探索的关键制品赋予名称。手动指定制品版本ZenML 使用自增编号为所有创建的制品自动版本化。例如上例中创建iris_dataset制品的 Step第一次执行会生成版本1第二次执行生成版本2依此类推。若需要自定义版本号例如在正式发布等关键运行中可以使用ArtifactConfigfrom typing import Annotated import pandas as pd from zenml import step, ArtifactConfig step def training_data_loader() - ( Annotated[ pd.DataFrame, # Add ArtifactConfig to control more properties of your artifact ArtifactConfig( nameiris_dataset, versionraw_2023 ), ] ): ...下一次执行该 Step 后将生成名称为iris_dataset、版本为raw_2023的制品。这对于那些需要一眼就能区分的特别重要的管道运行比如正式 release非常有用。警告由于自定义版本号不能重复上述 Step只能成功运行一次。为了避免频繁修改代码可以考虑使用 YAML 配置参见 production-guide 中的管道配置章节来管理制品版本。执行之后可以这样查看iris_dataset及其版本raw_2023OSSCLIzenml artifact version list列出所有版本CloudDashboardDashboard 会可视化展示版本历史。从源码 artifact_config.py 可以看到ArtifactConfig的完整字段定义比文档示例中用到的更丰富class ArtifactConfig(BaseModel): name: Optional[str] None version: Optional[Union[str, int]] Field(defaultNone, union_modesmart) tags: Optional[List[str]] None run_metadata: Optional[Dict[str, MetadataType]] None artifact_type: Optional[ArtifactType] None几个值得注意的点name支持动态字符串例如name_{date}_{time}_{custom_placeholder}使用date/time以外的占位符时需要在 Step 装饰器的substitutions参数或with_options的substitutions参数中提供取值version同时支持str和intunion_modesmartrun_metadata允许直接在配置上附加元数据字典源码中还有一个校验器artifact_config.py显式废弃了旧属性is_model_artifact/is_deployment_artifact并提示改用artifact_typeArtifactType.MODEL或ArtifactType.SERVICE——这正是下一节主题的历史背景。添加元数据与标签如果想给制品和运行附加额外元数据或标签可以使用log_metadata和add_tags工具函数from zenml import step, log_metadata, add_tags # In the following step, we use the utility functions log_metadata and add_tags. # Since we are calling these functions directly from a step, both will attach # the additional information to the current run. step def annotation_approach() - str: log_metadata(metadata{metadata_key: metadata_value}) add_tags(tags[tag_name]) return string # There are other ways to attach this information to different versions of your # artifacts as well. For instance, you will see a step with a single output below. # If you modify the call to include the infer_artifact flag, these functions # will attach this information to the artifact version instead. step def annotation_approach() - str: log_metadata(metadata{metadata_key: metadata_value}, infer_artifactTrue) add_tags(tags[tag_name], infer_artifactTrue) return string关键区别在于infer_artifact标志不带infer_artifact默认直接调用时元数据/标签附加到**当前运行run**上infer_artifactTrue附加到该 Step 产出的制品版本上。若 Step 只有一个输出无需指定artifact_name若有多个输出则需显式指明目标制品名。ZenML 中与标签和元数据交互的方式有多种。更完整的标签与元数据用法可参考仓库中的 how-to 文档 下 metadata、tags 相关章节。为制品指定类型ArtifactType为制品指定类型可以让 ZenML 在 Dashboard 中差异化展示并便于按类型过滤。如果未显式指定类型ZenML 会回退到保存该制品所用的 materializer 所提供的默认制品类型这一点在源码 step_runner.py 中亦有体现跳过实体化时会保留原 materializer 的ASSOCIATED_ARTIFACT_TYPE。from typing import Annotated from zenml import ArtifactConfig, save_artifact, step from zenml.enums import ArtifactType # Assign an artifact type to a step output step def trainer() - Annotated[MyCustomModel, ArtifactConfig(artifact_typeArtifactType.MODEL)]: return MyCustomModel(...) # Assign an artifact type when manually saving artifacts model ... save_artifact(model, namemodel, artifact_typeArtifactType.MODEL)从 enums.py 可以看到ArtifactType枚举的全部取值枚举值字符串值说明ArtifactType.DATA_ANALYSISDataAnalysisArtifact数据分析类制品ArtifactType.DATADataArtifact通用数据制品ArtifactType.MODELModelArtifact模型制品ArtifactType.SCHEMASchemaArtifact已废弃ArtifactType.SERVICEServiceArtifact部署服务制品ArtifactType.STATISTICSStatisticsArtifact已废弃由DATA_ANALYSIS替代ArtifactType.BASEBaseArtifact基础类型在管道中消费外部制品多数管道以产出制品的 Step 开始但常常需要消费管道之外的数据——例如一条 Snowflake 查询产出的 DataFrame或一个需要读取的 CSV 文件。ExternalArtifact类可以用任意数据类型初始化一个 ZenML 内部制品用于向 Step 传递既非 JSON 可序列化、又非上游 Step 产出的值import numpy as np from zenml import ExternalArtifact, pipeline, step step def print_data(data: np.ndarray): print(data) pipeline def printing_pipeline(): # One can also pass data directly into the ExternalArtifact # to create a new artifact on the fly data ExternalArtifact(valuenp.array([0])) print_data(datadata) if __name__ __main__: printing_pipeline()可以结合源码 external_artifact.py 了解ExternalArtifact的完整选项value: Optional[Any] None materializer: Optional[MaterializerClassOrSource] None store_artifact_metadata: bool True store_artifact_visualizations: bool True也就是说可以为其指定自定义 materializer支持类、类名或Source或者通过store_artifact_metadataFalse/store_artifact_visualizationsFalse关闭元数据提取与可视化存储。其upload_by_value方法external_artifact.py会在上传后把value置空、改为按id引用已上传制品从而避免重复上传。提示Step 中使用ExternalArtifact会自动禁用该 Step 的缓存。消费其他管道产出的制品在下游消费上游管道或 Step 产出的制品也是常见需求。Client可以直接在管道代码中获取制品from uuid import UUID import pandas as pd from zenml import step, pipeline from zenml.client import Client step def trainer(dataset: pd.DataFrame): ... pipeline def training_pipeline(): client Client() # Fetch by ID dataset_artifact client.get_artifact_version( name_id_or_prefixUUID(3a92ae32-a764-4420-98ba-07da8f742b76) ) # Fetch by name alone - uses the latest version of this artifact dataset_artifact client.get_artifact_version(name_id_or_prefixiris_dataset) # Fetch by name and version dataset_artifact client.get_artifact_version( name_id_or_prefixiris_dataset, versionraw_2023 ) # Pass into any step trainer(datasetdataset_artifact) if __name__ __main__: training_pipeline()三种获取方式分别为按 UUID 精确获取、仅按名称获取默认最新版本、按名称加版本号获取。提示在管道代码中直接调用get_artifact_version等 Client 方法背后实际使用了 ZenML 的late materialization延迟实体化机制——制品数据直到真正在 Step 运行时被读取才从 artifact store 加载。如果想完全绕过实体化、仅下载某个制品版本关联的数据或文件可以使用.download_files方法from zenml.client import Client client Client() artifact client.get_artifact_version(name_id_or_prefixiris_dataset) artifact.download_files(path/to/save.zip)注意目标路径必须以.zip结尾制品数据以 zip 形式保存并处理该操作可能抛出的异常。管理并非由 ZenML 管道产出的制品有时制品完全产生在 ZenML 之外——典型例子是已部署模型产生的预测结果# A model is deployed, running in a FastAPI container # Lets use the ZenML client to fetch the latest model and make predictions from zenml.client import Client from zenml import save_artifact # Fetch the model from a registry or a previous pipeline model ... # Lets make a prediction prediction model.predict([[1, 1, 1, 1]]) # We now store this prediction in ZenML as an artifact # This will create a new artifact version save_artifact(prediction, nameiris_predictions)反过来任何存储在 ZenML 中的制品都可以用load_artifact加载from zenml import load_artifact # Loads the latest version load_artifact(iris_predictions)load_artifact只是以下 Client 调用的简写from zenml.client import Client client Client() client.get_artifact(iris_predictions).load()这些函数也可以在 Step 内部使用不过通常更干净的做法是把制品作为 Step 输出返回自动保存或用ExternalArtifact加载输入。无论是外部创建还是管道内创建制品一旦进入 ZenML就能享受上文描述的全部功能。从源码 artifacts/utils.py 可以看到save_artifact的完整签名比文档示例展示的参数更多def save_artifact( data: Any, name: str, version: Optional[Union[int, str]] None, artifact_type: Optional[ArtifactType] None, tags: Optional[List[str]] None, extract_metadata: bool True, include_visualizations: bool True, user_metadata: Optional[Dict[str, MetadataType]] None, materializer: Optional[MaterializerClassOrSource] None, uri: Optional[str] None, save_type: ArtifactSaveType ArtifactSaveType.MANUAL, has_custom_name: bool True, artifact_store: Optional[BaseArtifactStore] None, ) - ArtifactVersionResponse:以及 utils.py 中的load_artifactdef load_artifact( name_or_id: Union[str, UUID], version: Optional[str] None, ) - Any: artifact Client().get_artifact_version(name_or_id, version) return load_artifact_from_response(artifact)将已有数据链接为 ZenML 制品有时数据完全在 ZenML 之外产生并便捷地存放在某个存储位置——典型例子是深度学习训练过程中产生的 checkpoint 文件。这类中间数据往往很大没有必要反复搬运只要让它直接生成在 artifact store 的边界内事后**链接link**为 ZenML 制品即可。以下是一个 PyTorch Lightning 示例将 checkpoint 保存到远程位置并注册为制品import os from zenml.client import Client from zenml import register_artifact from pytorch_lightning import Trainer from pytorch_lightning.callbacks import ModelCheckpoint from uuid import uuid4 # Define where the model data should be saved # use active ArtifactStore prefix Client().active_stack.artifact_store.path # keep data separable for future runs with uuid4 folder default_root_dir os.path.join(prefix, uuid4().hex) # Define the model and fit it model ... trainer Trainer( default_root_dirdefault_root_dir, callbacks[ ModelCheckpoint( every_n_epochs1, save_top_k-1, filenamecheckpoint-{epoch:02d} ) ], ) try: trainer.fit(model) finally: # We now link those checkpoints in ZenML as an artifact # This will create a new artifact version register_artifact(default_root_dir, nameall_my_model_checkpoints)从 utils.py 中register_artifact的签名可确认其参数def register_artifact( folder_or_file_uri: str, name: str, version: Optional[Union[int, str]] None, artifact_type: Optional[ArtifactType] None, tags: Optional[List[str]] None, has_custom_name: bool True, artifact_metadata: Dict[str, MetadataType] {}, ) - ArtifactVersionResponse: Register existing data stored in the artifact store as a ZenML Artifact.其文档字符串明确说明folder_or_file_uri是 artifact store内部指向文件或文件夹的完整 URI若未指定artifact_type则默认为data类型。提示由已有数据产生的制品被加载或作为输入传递给另一个 Step 时其类型会是pathlib.Path。为制品记录元数据与制品交互最有用的方式之一就是为其关联元数据。制品元数据是一个任意的键值对字典用于帮助理解数据的本质例如把模型训练结果关联到模型制品、把表的 shape 关联到pandasDataFrame、把图片尺寸关联到 PNG 文件。对部分制品ZenML 会自动记录元数据。例如对pandas.Series和pandas.DataFrame对象ZenML 会记录其 shape 与大小from zenml.client import Client # Get an artifact version (e.g. pd.DataFrame) artifact Client().get_artifact_version(50ce903f-faa6-41f6-a95f-ff8c0ec66010) # Fetch its metadata artifact.run_metadata[storage_size].value # Size in bytes artifact.run_metadata[shape].value # Shape e.g. (500,20)在界面侧制品元数据信息可以在 OSS Dashboard 的 DAG 可视化界面中找到ZenML Pro Dashboard 则提供了专门的 artifacts 标签页及元数据可视化能力。也可以在 Step 内部直接使用log_metadata方法给制品添加元数据from typing import Tuple from typing import Annotated import numpy as np from sklearn.base import ClassifierMixin from zenml import step, log_metadata, ArtifactConfig step def model_finetuner_step( model: ClassifierMixin, dataset: Tuple[np.ndarray, np.ndarray] ) - Annotated[ ClassifierMixin, ArtifactConfig(namemy_model, tags[SVC, trained]) ]: Finetunes a given model on a given dataset. model.fit(dataset[0], dataset[1]) accuracy model.score(dataset[0], dataset[1]) log_metadata( # Metadata should be a dictionary of JSON-serializable values metadata{accuracy: float(accuracy)}, # Using infer_artifactTrue automatically attaches metadata to the # artifact produced by this step. Since this step has only one output, # we dont need to specify the artifact_name infer_artifactTrue # If the step had multiple outputs, we would need to specify which one: # artifact_namemy_model, infer_artifactTrue # A dictionary of dictionaries can also be passed to group metadata # in the dashboard # metadata {metrics: {accuracy: accuracy}} ) return model三个要点元数据字典的值必须JSON 可序列化infer_artifactTrue会把元数据附加到本 Step 产出的制品版本上单输出时自动推断多输出时须指定artifact_name嵌套字典如{metrics: {accuracy: ...}}可以在 Dashboard 中分组展示。跨运行对比元数据Pro 功能ZenML Pro Dashboard 内置Experiment Comparison工具可可视化并分析不同管道运行之间的元数据帮助你理解管道行为随时间的变化规律。该工具提供两种互补视图表格视图Table View以结构化表格对比多个运行的元数据会自动计算运行间的变化支持排序与过滤元数据值追踪随时间的变化同时对比最多20 个运行。平行坐标视图Parallel Coordinates View用平行坐标可视化识别不同元数据参数之间的关系特别适合发现不同指标之间的相关性识别跨运行的模式过滤并聚焦于特定参数范围。使用步骤在 Dashboard 中进入任意管道点击顶部导航中的 Compare 按钮选择要对比的运行通过标签页在表格视图与平行坐标视图之间切换。该工具适用于你在管道中记录的任何数值型元数据float或int。要在最大程度上利用该功能请在 Step 中记录有意义的指标。分享对比结果对比工具会把你的对比配置保留在 URL 中方便与团队成员分享特定视图直接复制并分享 URL其他人打开即可看到完全相同的对比设置与过滤器。该功能目前处于 Alpha Preview 阶段官方鼓励通过 Slack 社区反馈使用场景与需求。完整示例将全部概念组合成一个脚本下面把本节所有代码组合为一个可直接使用的完整脚本演示版本化数据加载、模型微调、指定历史版本复用、以及save_artifact/load_artifact的端到端协作from typing import Optional, Tuple from typing import Annotated import numpy as np from sklearn.base import ClassifierMixin from sklearn.datasets import load_digits from sklearn.svm import SVC from zenml import ArtifactConfig, pipeline, step, log_metadata from zenml import save_artifact, load_artifact from zenml.client import Client step def versioned_data_loader_step() - ( Annotated[ Tuple[np.ndarray, np.ndarray], ArtifactConfig( namemy_dataset, tags[digits, computer vision, classification], ), ] ): Loads the digits dataset as a tuple of flattened numpy arrays. digits load_digits() return (digits.images.reshape((len(digits.images), -1)), digits.target) step def model_finetuner_step( model: ClassifierMixin, dataset: Tuple[np.ndarray, np.ndarray] ) - Annotated[ ClassifierMixin, ArtifactConfig(namemy_model, tags[SVC, trained]), ]: Finetunes a given model on a given dataset. model.fit(dataset[0], dataset[1]) accuracy model.score(dataset[0], dataset[1]) log_metadata(metadata{accuracy: float(accuracy)}) return model pipeline def model_finetuning_pipeline( dataset_version: Optional[str] None, model_version: Optional[str] None, ): client Client() # Either load a previous version of my_dataset or create a new one if dataset_version: dataset client.get_artifact_version( name_id_or_prefixmy_dataset, versiondataset_version ) else: dataset versioned_data_loader_step() # Load the model to finetune # If no version is specified, the latest version of my_model is used model client.get_artifact_version( name_id_or_prefixmy_model, versionmodel_version ) # Finetune the model # This automatically creates a new version of my_model model_finetuner_step(modelmodel, datasetdataset) def main(): # Save an untrained model as first version of my_model untrained_model SVC(gamma0.001) save_artifact( untrained_model, namemy_model, version1, tags[SVC, untrained] ) # Create a first version of my_dataset and train the model on it model_finetuning_pipeline() # Finetune the latest model on an older version of the dataset model_finetuning_pipeline(dataset_version1) # Run inference with the latest model on an older version of the dataset latest_trained_model load_artifact(my_model) old_dataset load_artifact(my_dataset, version1) latest_trained_model.predict(old_dataset[0]) if __name__ __main__: main()脚本执行后会产生两次管道运行Run 1main()先用save_artifact保存未训练模型作为my_model的1版本随后model_finetuning_pipeline()通过versioned_data_loader_step()创建my_dataset的1版本并在其上训练模型自动创建my_model的新版本Run 2model_finetuning_pipeline(dataset_version1)用旧版本my_dataset复训最新模型——此时 DAG 中数据输入不再来自 Step而是来自外部制品引用直观展示了跨版本数据复用的能力。小结本篇以 ZenML 官方文档为主线覆盖了一个完整的制品生命周期闭环自动版本化Step 输出自动以{pipeline_name}::{step_name}::{output_name}命名源码 step_runner.py 可查证并用自增版本号区分历次运行ArtifactConfig通过name含动态占位符、version、tags、run_metadata、artifact_type五个字段精细控制制品属性artifact_config.py元数据与标签log_metadata/add_tags配合infer_artifact决定信息挂到运行还是制品版本上外部数据接入ExternalArtifactexternal_artifact.py传入任意值但会禁用 Step 缓存register_artifact把 artifact store 内已有大数据如训练 checkpoint零拷贝链接为制品跨管道消费Client.get_artifact_version按 ID/名称/名称版本获取制品并借助延迟实体化按需加载download_files可直接落盘为 zip管道外制品管理save_artifact/load_artifactutils.py让部署预测等外部产物也纳入统一的版本化追踪。配合 Pro Dashboard 的 Experiment Comparison 工具这些版本化、类型化、带元数据的制品最终可以转化为可对比、可分享的实验洞察。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考