ZenML Metadata 完整指南为 Pipeline、Step、Artifact 与 Model 记录和检索上下文元数据【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenmlMetadata 是 ZenML 中为 ML 工作流提供关键上下文的能力它允许你在 step、pipeline run、artifact 和 model 上附加额外的结构化信息从而增强可追踪性帮助更好地理解、比较和复现实验。本文以 ZenML 官方文档为主线结合本仓库源码src/zenml/metadata/metadata_types.py、src/zenml/utils/metadata_utils.py、src/zenml/models/v2/misc/param_groups.py等深入讲解log_metadata、bulk_log_metadata的完整用法、特殊元数据类型、Dashboard 可视化组织以及通过StepContext/PipelineContext/ ZenML Client 检索元数据的方法读完后你将能够为自己的实验搭建一套可复现、可比较、可审计的元数据体系。Metadata 是什么可以附加到哪里Metadata 是你希望与 ML 工作流组件关联的任何附加上下文信息。在 ZenML 中你可以把元数据附加到四类实体上Steps步骤记录评估指标、执行细节或配置信息例如accuracy、precision、模型超参数Pipeline Runs流水线运行追踪整体运行特征例如环境变量、Git 提交信息、训练总耗时Artifacts产物记录数据特征、来源信息或处理细节例如行数、列名、存储大小、分布偏移Models模型记录评估结果、超参数或部署信息例如端点地址、模型版本、deployment_info。ZenML 通过一个简单的接口即可完成元数据的写入与读取并在 Dashboard 中可视化展示便于快速分析。从源码结构看ZenML 内部把元数据统一建模为「Run Metadata」并支持pipeline_run、step_run、artifact_version、model_version四类资源另外还预留了schedule、wait_condition两类扩展资源见 enums.py 中的MetadataResourceTypes。Logging Metadatalog_metadata基础用法写入元数据的主要方式是log_metadata函数它允许你把 JSON 可序列化的键值对附加到各类实体上。支持的数据类型基础类型str、int、float、bool集合类型list、dict、set、tupleset和tuple在存储时会被自动转换为list特殊 ZenML 类型Uri、Path、DType、StorageSize详见下文「特殊元数据类型」from zenml import log_metadata # 基础元数据写入 log_metadata( metadata{accuracy: 0.95, precision: 0.92}, # 附加参数用于指定把元数据写到哪个实体上 )log_metadata非常灵活根据传入的参数不同可以定位到不同的目标实体。从源码实现看metadata_utils.py其参数组合遵循一个明确的优先级分支step_id→step_name run_id_name_or_prefix→run_id_name_or_prefix→model_name model_version→model_version_id→infer_model→artifact_name artifact_version→artifact_version_id→infer_artifact→ 全部参数为None裸调用自动定位到当前 step。如果这些组合都不匹配会抛出ValueError并提示所有合法的调用方式。附加 Metadata 到 Steps要给 step 记录元数据可以在 step 内部调用log_metadata自动关联当前 step也可以显式指定一个 stepfrom zenml import step, log_metadata # 方法一在 step 内部自动关联当前 step step def train_model_step(data): model train_model(data) accuracy evaluate_model(model, data) # 直接在 step 内记录指标 log_metadata( metadata{evaluation_metrics: {accuracy: accuracy}} ) return model # 方法二step 执行后指定具体的 step log_metadata( metadata{post_analysis: {feature_importance: [0.2, 0.5, 0.3]}}, step_nametrain_model_step, run_id_name_or_prefixmy_run_id ) # 方法三使用 step_id log_metadata( metadata{post_analysis: {feature_importance: [0.2, 0.5, 0.3]}}, step_idstep_uuid )从源码看当通过step_name run_id_name_or_prefix定位时ZenML 会先调用client.get_pipeline_run()解析 run再调用client.list_run_steps()按名字在运行中查找 step 并取其 ID见 metadata_utils.py。附加 Metadata 到 Pipeline Runs可以为整条 pipeline run 记录元数据既可以在某个 step 执行期间写入也可以在 run 结束后手动写入from zenml import get_step_context, pipeline, step, log_metadata # 方法一在 step 内写入当前 run step def log_run_info_step(): context get_step_context() # 获取一些运行时信息 git_commit get_git_hash() environment get_env_info() # 写入当前 pipeline run log_metadata( metadata{ git_info: {commit: git_commit}, environment: environment }, run_id_name_or_prefixcontext.pipeline_run.id, ) # 方法二手动指定一个 run log_metadata( metadata{post_run_analysis: {total_training_time: 350}}, run_id_name_or_prefixmy_run_id )需要注意的是当在 step 内部向 pipeline run 写入元数据时元数据的 key 会带上step_name::metadata_key前缀这样多个 step 可以使用相同的 metadata key 而互不冲突。附加 Metadata 到 ArtifactsArtifact 是 pipeline step 产出的数据对象。为它们记录元数据可以提供关于数据本身的更多上下文from zenml import step, log_metadata from zenml.metadata.metadata_types import StorageSize # 方法一在 step 内为输出 artifact 记录step 只有单个输出 step def process_data_step(raw_data): processed_data transform(raw_data) # 为输出 artifact 记录元数据 log_metadata( metadata{ data_stats: { row_count: len(processed_data), columns: list(processed_data.columns), storage_size: StorageSize(processed_data.memory_usage().sum()) } }, infer_artifactTrue # 自动定位到输出 artifact ) return processed_data # 方法二step 有多个输出时按名字定位 step def split_data_step(data): train, test split_data(data) # 为指定的输出记录元数据 log_metadata( metadata{split_info: {train_size: len(train)}}, artifact_nameoutput_0, # 指定具体输出的名字 infer_artifactTrue ) return train, test # 方法三显式按 artifact 名称和版本定位 log_metadata( metadata{validation_results: {distribution_shift: 0.03}}, artifact_nameprocessed_data, artifact_version20230615 ) # 方法四按 artifact version ID 定位 log_metadata( metadata{validation_results: {distribution_shift: 0.03}}, artifact_version_idartifact_uuid )关于infer_artifact有一个重要的行为细节见 metadata_utils.py它必须在一个有输出的 step 内部调用否则抛出ValueError当 step 有多个输出而你又没有通过artifact_name指定时同样会抛出ValueError提示「There is more than one output…」。infer_artifact最终调用的是step_context.add_output_metadata()。附加 Metadata 到 ModelsModel 在 ZenML 中是一个更高层次的概念可以封装多个 artifact 和 step。为 model 记录元数据有助于追踪性能和其他重要信息from zenml import step, log_metadata # 方法一在产出模型的 step 内记录 step def train_model_step(data): model train_model(data) metrics evaluate_model(model, data) # 把元数据写入 model log_metadata( metadata{ evaluation_metrics: metrics, hyperparameters: model.get_params() }, infer_modelTrue # 自动定位到与当前 step 关联的 model ) return model # 方法二显式按 model 名称和版本定位 log_metadata( metadata{deployment_info: {endpoint: api.example.com/model}}, model_namefraud_detector, model_version1.0.0 ) # 方法三按 model version ID 定位 log_metadata( metadata{deployment_info: {endpoint: api.example.com/model}}, model_version_idmodel_version_uuid )源码中对infer_model的要求是必须在装饰器里配置了model的 step 内部调用并且 step context 中必须存在 model version否则会抛出ValueError见 metadata_utils.py。Bulk Metadata Logging一次写入多个实体log_metadata不支持在单次调用中为多个实体写入同一份元数据。要实现这一点可以使用bulk_log_metadata函数from zenml.models import ( ArtifactVersionIdentifier, ModelVersionIdentifier, PipelineRunIdentifier, StepRunIdentifier, ) from zenml import bulk_log_metadata bulk_log_metadata( metadata{python_version: 3.11, environment: macosx}, pipeline_runs[ PipelineRunIdentifier(idrun_id), PipelineRunIdentifier(namerun name) ], step_runs[ StepRunIdentifier(idstep_run_id), StepRunIdentifier(namestep_name, runPipelineRunIdentifier(idrun_id)) ], artifact_versions[ ArtifactVersionIdentifier(idartifact_version_id), ArtifactVersionIdentifier(nameartifact_name, versionartifact_version) ], model_versions[ ModelVersionIdentifier(idmodel_version_id), ModelVersionIdentifier(namemodel_name, versionmodel_version) ] )注意bulk_log_metadata与log_metadata的签名略有不同。你可以使用 Identifier 类对象来指定任意一组能唯一标识实体的参数VersionedIdentifiers带版本标识符ArtifactVersionIdentifier与ModelVersionIdentifier可以指定id也可以指定nameversion组合PipelineRunIdentifier可以指定id、name或prefix前缀匹配StepRunIdentifier可以指定id或name pipeline run identifier 组合。这些 Identifier 类在源码中有严格的参数校验见 param_groups.py例如VersionedIdentifier不允许同时提供id和namename与version必须成对出现PipelineRunIdentifier只允许id/name/prefix三选一StepRunIdentifier用name定位时必须同时给出run。与log_metadata类似如果在 step 内部调用bulk_log_metadata同样可以使用 infer 选项自动为 step 关联的 model version 或 artifact 写入元数据from zenml import bulk_log_metadata, step step() def get_train_test_datasets(): train_dataset, test_dataset get_datasets() bulk_log_metadata( metadata{python_version: 3.11, environment: macosx}, infer_modelsTrue, infer_artifactsTrue ) return train_dataset, test_dataset请记住使用infer_artifacts选项时bulk_log_metadata会把元数据写入该 step 的所有输出 artifact源码中遍历step_context._outputs.keys()逐个调用add_output_metadata见 metadata_utils.py。有时你可能想同时使用 infer 选项和显式的 identifier 引用。例如想为某个 step 的输出写入元数据同时也要写入它的输入。bulk_log_metadata支持一次调用中两种方式混用from zenml import bulk_log_metadata, get_step_context, step from zenml.models import ArtifactVersionIdentifier def calculate_metrics(model, test_dataset): ... def summarize_metrics(metrics_report): ... step def model_evaluation(test_dataset, model): metrics_report calculate_metrics(model, test_dataset) slim_metrics_version summarize_metrics(metrics_report) bulk_log_metadata( metadataslim_metrics_version, infer_artifactsTrue, # 为输出写入元数据 artifact_versions[ ArtifactVersionIdentifier(idget_step_context().inputs[model].id) ] # 为 model 输入也写入元数据 ) return metrics_report性能优化提示log_metadata与bulk_log_metadata内部都会利用 name、version 等参数来解析实体真正的 ID。例如当你提供 artifact 的 name 和 version 时函数会额外执行一次数据库查找来解析出 artifact version ID见 metadata_utils.py 中逐个 Identifier 的client.get_*解析逻辑。为了提升性能尽量直接使用实体的 ID而不是 name、version 或其他标识符。直接使用 Client如果log_metadata或bulk_log_metadata对你的用例来说限制太多可以直接使用 ZenML Client 来为资源创建 run metadatafrom zenml.client import Client from zenml.enums import MetadataResourceTypes from zenml.models import RunMetadataResource client Client() client.create_run_metadata( metadata{python: 3.11}, resources[ RunMetadataResource(idstep_run_id, typeMetadataResourceTypes.STEP_RUN), RunMetadataResource(idrun_id, typeMetadataResourceTypes.PIPELINE_RUN), RunMetadataResource(idartifact_version_id, typeMetadataResourceTypes.ARTIFACT_VERSION), RunMetadataResource(idmodel_version_id, typeMetadataResourceTypes.MODEL_VERSION) ] )RunMetadataResource是一个轻量的 pydantic 模型只包含id和type两个字段并重写了__eq__/__hash__因此bulk_log_metadata内部使用set来去重资源见 run_metadata.py。MetadataResourceTypes枚举定义了所有可附加元数据的资源类型见 enums.py。特殊元数据类型ZenML 提供了几种特殊的元数据类型用于以标准化方式表示常见的元数据from zenml import log_metadata from zenml.metadata.metadata_types import StorageSize, DType, Uri, Path log_metadata( metadata{ dataset_source: Uri(gs://my-bucket/datasets/source.csv), # 外部 URI preprocessing_script: Path(/scripts/preprocess.py), # 文件路径 column_types: { age: DType(int), # 数据类型 income: DType(float), score: DType(int) }, processed_data_size: StorageSize(2500000) # 以字节为单位的存储大小 }, infer_artifactTrue )这些特殊类型在源码 metadata_types.py 中定义Uri/Path/DType都是str的子类分别用来标注「URI」「文件路径」「数据类型」语义StorageSize是int的子类表示以字节为单位的存储大小完整的MetadataType联合类型为str | int | float | bool | dict | list | set | tuple | Uri | Path | DType | StorageSize并通过MetadataTypeEnum枚举与数据库中的字符串类型一一映射见 metadata_types.py。这些特殊类型保证元数据以一致、可解释的方式记录并在 ZenML Dashboard 中得到特殊渲染。此外validate_metadata函数metadata_types.py会在写入前对元数据做校验key 过长、值序列化后超过TEXT_FIELD_MAX_LENGTH、或类型不受支持的值都会被跳过并给出警告日志。在 Dashboard 中组织元数据为了改善 ZenML Dashboard 的可视化效果你可以通过「字典套字典」的方式把元数据分组为逻辑区块from zenml import log_metadata from zenml.metadata.metadata_types import StorageSize log_metadata( metadata{ model_metrics: { # Dashboard 中的第一张卡片 accuracy: 0.95, precision: 0.92, recall: 0.90 }, data_details: { # Dashboard 中的第二张卡片 dataset_size: StorageSize(1500000), feature_columns: [age, income, score] } }, artifact_namemy_artifact, artifact_versionversion, )在 ZenML Dashboard 中model_metrics和data_details会分别显示为独立的卡片每张卡片内包含各自的键值对从而更便于导航和解读元数据。可视化与比较元数据ZenML Pro 功能元数据写入之后可以使用 ZenML 的Experiment Comparison实验对比工具跨 run 分析和比较指标。注意元数据对比工具是ZenML Pro 专属功能可参考 getting-started/zenml-pro 了解 Pro 版本能力。对比视图Experiment Comparison 工具提供两种互补的视图来分析 pipeline 元数据1. 表格视图Table View跨 run 比较元数据并带有自动变化追踪。2. 平行坐标图Parallel Coordinates Plot可视化不同指标之间的关系。该工具支持同时对比最多20 个 pipeline run并支持你在 pipeline 中记录的任何数值型元数据float或int。检索Fetch元数据以编程方式检索元数据元数据写入后可以使用 ZenML Client 检索from zenml.client import Client client Client() # 从 step 获取元数据 step client.get_pipeline_run(pipeline_run_id).steps[step_name] step_metadata step.run_metadata[metadata_key] # 从 run 获取元数据 run client.get_pipeline_run(pipeline_run_id) run_metadata run.run_metadata[metadata_key] # 从 artifact 获取元数据 artifact client.get_artifact_version(artifact_name, version) artifact_metadata artifact.run_metadata[metadata_key] # 从 model 获取元数据 model client.get_model_version(model_name, version) model_metadata model.run_metadata[metadata_key]提示当使用特定 key 获取元数据时返回的值始终是该 key 最新的条目底层RunMetadataEntry按创建时间created排序见 run_metadata.py。在 Step 内通过 StepContext 访问上下文StepContext对象是 step 执行期间访问当前pipeline/step run 的句柄。你可以用它读取 run/step 信息、检查上游输入的元数据以及操作 step 输出的 URI、materializer、run metadata 和 tags。它只在以下场景可用在step装饰的函数内部执行期间而非组合期间在on_failure/on_success等 step hooks 内部参见 Hooks 文档在 step 触发的save/loadmaterializer 内部在其他位置调用get_step_context()会抛出RuntimeError。获取上下文的方式是get_step_context()from zenml import step, get_step_context step def trainer(param: int 1): ctx get_step_context() print(run:, ctx.pipeline_run.name, ctx.pipeline_run.id) print(step:, ctx.step_run.name, ctx.step_run.id) print(params:, ctx.step_run.config.parameters)它暴露了以下属性ctx.pipeline→ 本次 run 对应的PipelineResponse便捷属性如果 run 没有对应的 pipeline 对象可能抛出异常ctx.pipeline_run→PipelineRunResponse包含 id、name、status、时间戳等ctx.step_run→StepRunResponse包含 name、parameters通过ctx.step_run.config.parameters访问、statusctx.model→ 配置好的Model从 step 或 pipeline 解析而来未配置则抛出异常ctx.inputs→{input_name: StepRunInputResponse}可以用...[x].run_metadata读取上游元数据ctx.step_name→ 便捷的 step 名字符串。操作 step 输出对于单输出 step 可以省略output_name。对于多输出 step必须传入output_name未命名的输出会被命名为output_1、output_2以此类推。这些方法定义在 step_context.pyget_output_artifact_uri(output_nameNone) - str– 输出 artifact 所在位置写入侧文件等get_output_materializer(output_nameNone, *, custom_materializer_classNone, data_typeNone) - BaseMaterializer– 获取一个已初始化的 materializer传入data_type可从Union[...]materializer 中选择或传custom_materializer_class覆盖add_output_metadata(metadata, output_nameNone)/get_output_metadata(output_nameNone)– 为输出设置/读取 run metadata。返回注解中通过ArtifactConfig(..., run_metadata...)提供的值会与运行时值合并add_output_tags(tags, output_nameNone)/get_output_tags(output_nameNone)/remove_output_tags(tags, output_nameNone)– 管理产出的 artifact version 的 tags。通过ArtifactConfig(..., tags...)配置的 tags 会与运行时 tags 取并集重复项在最终 artifact 中会被去重。最小示例from typing import Annotated, Tuple from zenml import step, get_step_context, log_metadata from zenml.artifacts.artifact_config import ArtifactConfig step def produce(name: str) - Tuple[ Annotated[ str, ArtifactConfig( namecustom_name, run_metadata{config_metadata: bar}, tags[config_tags], ), ], str, ]: ctx get_step_context() # 把元数据和 tags 附加到指定或默认输出 ctx.add_output_metadata({m: 1}, output_namename) ctx.add_output_tags([t1, t1], output_namename) # 重复也没关系 return a, b通过inputs读取上游元数据from zenml import step, get_step_context, log_metadata step def upstream() - int: log_metadata({quality: ok}, infer_artifactTrue) return 42 step def downstream(x: int) - None: md get_step_context().inputs[x].run_metadata assert md[quality] ok这个模式非常适合在 downstream step 中做数据质量校验、分布检查等「上游产出、下游消费」的元数据流转。Hooks 与 Materializers进阶用法from zenml import step, get_step_context from zenml.materializers.base_materializer import BaseMaterializer def on_failure(exc: BaseException): c get_step_context() print(Failed step:, c.step_run.name, -, type(exc).__name__) class ExampleMaterializer(BaseMaterializer): def save(self, data): # step 触发物化时上下文可用 data.meta get_step_context().pipeline.name super().save(data) step(on_failureon_failure) def my_step(): raise ValueError(boom)常见错误在非运行中的 step 外调用get_step_context()会抛出RuntimeError输出相关 helper 会抛出StepContextError触发条件包括step 没有输出、多输出 step 上省略了output_name、引用了未知的output_name。在 Pipeline 组合期间访问上下文在 pipeline 组合期间可以使用PipelineContext访问 pipeline 配置from zenml import pipeline, get_pipeline_context pipeline( extra{ model_configs: [ (sklearn.tree, DecisionTreeClassifier), (sklearn.ensemble, RandomForestClassifier), ] } ) def my_pipeline(): # 获取 pipeline 上下文 context get_pipeline_context() # 访问配置 model_configs context.extra[model_configs] # 利用配置动态创建 step for i, (model_package, model_class) in enumerate(model_configs): train_model( model_packagemodel_package, model_classmodel_class, idftrain_model_{i} )这个模式非常适合基于配置驱动地动态生成多个同类 step例如批量训练不同模型并借助extra字典传递任意结构化配置。Best Practices最佳实践要充分发挥 ZenML 元数据能力建议遵循以下原则使用一致的 key为组织定义标准的元数据 key保证一致性分组相关元数据用嵌套字典在 Dashboard 中创建逻辑分组善用特殊类型使用 ZenML 的特殊元数据类型Uri、Path、DType、StorageSize实现标准化表示记录相关信息只记录有助于复现、理解和决策的元数据考虑自动化为标准指标和信息设置自动元数据写入例如在公共基类 step 或 hooks 中统一记录 Git 版本、Python 版本、环境信息与 tags 结合将元数据与 tags 配合使用构建完整的组织体系元数据侧重结构化键值tags 侧重灵活的标签分类。Conclusion结语ZenML 的元数据能力为 ML 工作流提供了强大的上下文增强手段。通过为 steps、runs、artifacts 和 models 追踪附加细节你可以更深入地洞察实验、做出更明智的决策并确保 ML pipeline 的可复现性。从log_metadata的实体定位逻辑见 metadata_utils.py到bulk_log_metadata的多实体批量写入与 Identifier 校验见 param_groups.py再到StepContext的输出元数据操作见 step_context.py这套机制在源码层面环环相扣。建议在真实 pipeline 中从「step 内记录评估指标 artifact 记录数据统计」开始实践逐步扩展到 run 级环境信息与 model 级部署信息最终形成一套可比较、可审计的完整元数据体系。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ZenML Metadata 完整指南为 Pipeline、Step、Artifact 与 Model 记录和检索上下文元数据【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenmlMetadata 是 ZenML 中为 ML 工作流提供关键上下文的能力它允许你在 step、pipeline run、artifact 和 model 上附加额外的结构化信息从而增强可追踪性帮助更好地理解、比较和复现实验。本文以 ZenML 官方文档为主线结合本仓库源码src/zenml/metadata/metadata_types.py、src/zenml/utils/metadata_utils.py、src/zenml/models/v2/misc/param_groups.py等深入讲解log_metadata、bulk_log_metadata的完整用法、特殊元数据类型、Dashboard 可视化组织以及通过StepContext/PipelineContext/ ZenML Client 检索元数据的方法读完后你将能够为自己的实验搭建一套可复现、可比较、可审计的元数据体系。Metadata 是什么可以附加到哪里Metadata 是你希望与 ML 工作流组件关联的任何附加上下文信息。在 ZenML 中你可以把元数据附加到四类实体上Steps步骤记录评估指标、执行细节或配置信息例如accuracy、precision、模型超参数Pipeline Runs流水线运行追踪整体运行特征例如环境变量、Git 提交信息、训练总耗时Artifacts产物记录数据特征、来源信息或处理细节例如行数、列名、存储大小、分布偏移Models模型记录评估结果、超参数或部署信息例如端点地址、模型版本、deployment_info。ZenML 通过一个简单的接口即可完成元数据的写入与读取并在 Dashboard 中可视化展示便于快速分析。从源码结构看ZenML 内部把元数据统一建模为「Run Metadata」并支持pipeline_run、step_run、artifact_version、model_version四类资源另外还预留了schedule、wait_condition两类扩展资源见 enums.py 中的MetadataResourceTypes。Logging Metadatalog_metadata基础用法写入元数据的主要方式是log_metadata函数它允许你把 JSON 可序列化的键值对附加到各类实体上。支持的数据类型基础类型str、int、float、bool集合类型list、dict、set、tupleset和tuple在存储时会被自动转换为list特殊 ZenML 类型Uri、Path、DType、StorageSize详见下文「特殊元数据类型」from zenml import log_metadata # 基础元数据写入 log_metadata( metadata{accuracy: 0.95, precision: 0.92}, # 附加参数用于指定把元数据写到哪个实体上 )log_metadata非常灵活根据传入的参数不同可以定位到不同的目标实体。从源码实现看metadata_utils.py其参数组合遵循一个明确的优先级分支step_id→step_name run_id_name_or_prefix→run_id_name_or_prefix→model_name model_version→model_version_id→infer_model→artifact_name artifact_version→artifact_version_id→infer_artifact→ 全部参数为None裸调用自动定位到当前 step。如果这些组合都不匹配会抛出ValueError并提示所有合法的调用方式。附加 Metadata 到 Steps要给 step 记录元数据可以在 step 内部调用log_metadata自动关联当前 step也可以显式指定一个 stepfrom zenml import step, log_metadata # 方法一在 step 内部自动关联当前 step step def train_model_step(data): model train_model(data) accuracy evaluate_model(model, data) # 直接在 step 内记录指标 log_metadata( metadata{evaluation_metrics: {accuracy: accuracy}} ) return model # 方法二step 执行后指定具体的 step log_metadata( metadata{post_analysis: {feature_importance: [0.2, 0.5, 0.3]}}, step_nametrain_model_step, run_id_name_or_prefixmy_run_id ) # 方法三使用 step_id log_metadata( metadata{post_analysis: {feature_importance: [0.2, 0.5, 0.3]}}, step_idstep_uuid )从源码看当通过step_name run_id_name_or_prefix定位时ZenML 会先调用client.get_pipeline_run()解析 run再调用client.list_run_steps()按名字在运行中查找 step 并取其 ID见 metadata_utils.py。附加 Metadata 到 Pipeline Runs可以为整条 pipeline run 记录元数据既可以在某个 step 执行期间写入也可以在 run 结束后手动写入from zenml import get_step_context, pipeline, step, log_metadata # 方法一在 step 内写入当前 run step def log_run_info_step(): context get_step_context() # 获取一些运行时信息 git_commit get_git_hash() environment get_env_info() # 写入当前 pipeline run log_metadata( metadata{ git_info: {commit: git_commit}, environment: environment }, run_id_name_or_prefixcontext.pipeline_run.id, ) # 方法二手动指定一个 run log_metadata( metadata{post_run_analysis: {total_training_time: 350}}, run_id_name_or_prefixmy_run_id )需要注意的是当在 step 内部向 pipeline run 写入元数据时元数据的 key 会带上step_name::metadata_key前缀这样多个 step 可以使用相同的 metadata key 而互不冲突。附加 Metadata 到 ArtifactsArtifact 是 pipeline step 产出的数据对象。为它们记录元数据可以提供关于数据本身的更多上下文from zenml import step, log_metadata from zenml.metadata.metadata_types import StorageSize # 方法一在 step 内为输出 artifact 记录step 只有单个输出 step def process_data_step(raw_data): processed_data transform(raw_data) # 为输出 artifact 记录元数据 log_metadata( metadata{ data_stats: { row_count: len(processed_data), columns: list(processed_data.columns), storage_size: StorageSize(processed_data.memory_usage().sum()) } }, infer_artifactTrue # 自动定位到输出 artifact ) return processed_data # 方法二step 有多个输出时按名字定位 step def split_data_step(data): train, test split_data(data) # 为指定的输出记录元数据 log_metadata( metadata{split_info: {train_size: len(train)}}, artifact_nameoutput_0, # 指定具体输出的名字 infer_artifactTrue ) return train, test # 方法三显式按 artifact 名称和版本定位 log_metadata( metadata{validation_results: {distribution_shift: 0.03}}, artifact_nameprocessed_data, artifact_version20230615 ) # 方法四按 artifact version ID 定位 log_metadata( metadata{validation_results: {distribution_shift: 0.03}}, artifact_version_idartifact_uuid )关于infer_artifact有一个重要的行为细节见 metadata_utils.py它必须在一个有输出的 step 内部调用否则抛出ValueError当 step 有多个输出而你又没有通过artifact_name指定时同样会抛出ValueError提示「There is more than one output…」。infer_artifact最终调用的是step_context.add_output_metadata()。附加 Metadata 到 ModelsModel 在 ZenML 中是一个更高层次的概念可以封装多个 artifact 和 step。为 model 记录元数据有助于追踪性能和其他重要信息from zenml import step, log_metadata # 方法一在产出模型的 step 内记录 step def train_model_step(data): model train_model(data) metrics evaluate_model(model, data) # 把元数据写入 model log_metadata( metadata{ evaluation_metrics: metrics, hyperparameters: model.get_params() }, infer_modelTrue # 自动定位到与当前 step 关联的 model ) return model # 方法二显式按 model 名称和版本定位 log_metadata( metadata{deployment_info: {endpoint: api.example.com/model}}, model_namefraud_detector, model_version1.0.0 ) # 方法三按 model version ID 定位 log_metadata( metadata{deployment_info: {endpoint: api.example.com/model}}, model_version_idmodel_version_uuid )源码中对infer_model的要求是必须在装饰器里配置了model的 step 内部调用并且 step context 中必须存在 model version否则会抛出ValueError见 metadata_utils.py。Bulk Metadata Logging一次写入多个实体log_metadata不支持在单次调用中为多个实体写入同一份元数据。要实现这一点可以使用bulk_log_metadata函数from zenml.models import ( ArtifactVersionIdentifier, ModelVersionIdentifier, PipelineRunIdentifier, StepRunIdentifier, ) from zenml import bulk_log_metadata bulk_log_metadata( metadata{python_version: 3.11, environment: macosx}, pipeline_runs[ PipelineRunIdentifier(idrun_id), PipelineRunIdentifier(namerun name) ], step_runs[ StepRunIdentifier(idstep_run_id), StepRunIdentifier(namestep_name, runPipelineRunIdentifier(idrun_id)) ], artifact_versions[ ArtifactVersionIdentifier(idartifact_version_id), ArtifactVersionIdentifier(nameartifact_name, versionartifact_version) ], model_versions[ ModelVersionIdentifier(idmodel_version_id), ModelVersionIdentifier(namemodel_name, versionmodel_version) ] )注意bulk_log_metadata与log_metadata的签名略有不同。你可以使用 Identifier 类对象来指定任意一组能唯一标识实体的参数VersionedIdentifiers带版本标识符ArtifactVersionIdentifier与ModelVersionIdentifier可以指定id也可以指定nameversion组合PipelineRunIdentifier可以指定id、name或prefix前缀匹配StepRunIdentifier可以指定id或name pipeline run identifier 组合。这些 Identifier 类在源码中有严格的参数校验见 param_groups.py例如VersionedIdentifier不允许同时提供id和namename与version必须成对出现PipelineRunIdentifier只允许id/name/prefix三选一StepRunIdentifier用name定位时必须同时给出run。与log_metadata类似如果在 step 内部调用bulk_log_metadata同样可以使用 infer 选项自动为 step 关联的 model version 或 artifact 写入元数据from zenml import bulk_log_metadata, step step() def get_train_test_datasets(): train_dataset, test_dataset get_datasets() bulk_log_metadata( metadata{python_version: 3.11, environment: macosx}, infer_modelsTrue, infer_artifactsTrue ) return train_dataset, test_dataset请记住使用infer_artifacts选项时bulk_log_metadata会把元数据写入该 step 的所有输出 artifact源码中遍历step_context._outputs.keys()逐个调用add_output_metadata见 metadata_utils.py。有时你可能想同时使用 infer 选项和显式的 identifier 引用。例如想为某个 step 的输出写入元数据同时也要写入它的输入。bulk_log_metadata支持一次调用中两种方式混用from zenml import bulk_log_metadata, get_step_context, step from zenml.models import ArtifactVersionIdentifier def calculate_metrics(model, test_dataset): ... def summarize_metrics(metrics_report): ... step def model_evaluation(test_dataset, model): metrics_report calculate_metrics(model, test_dataset) slim_metrics_version summarize_metrics(metrics_report) bulk_log_metadata( metadataslim_metrics_version, infer_artifactsTrue, # 为输出写入元数据 artifact_versions[ ArtifactVersionIdentifier(idget_step_context().inputs[model].id) ] # 为 model 输入也写入元数据 ) return metrics_report性能优化提示log_metadata与bulk_log_metadata内部都会利用 name、version 等参数来解析实体真正的 ID。例如当你提供 artifact 的 name 和 version 时函数会额外执行一次数据库查找来解析出 artifact version ID见 metadata_utils.py 中逐个 Identifier 的client.get_*解析逻辑。为了提升性能尽量直接使用实体的 ID而不是 name、version 或其他标识符。直接使用 Client如果log_metadata或bulk_log_metadata对你的用例来说限制太多可以直接使用 ZenML Client 来为资源创建 run metadatafrom zenml.client import Client from zenml.enums import MetadataResourceTypes from zenml.models import RunMetadataResource client Client() client.create_run_metadata( metadata{python: 3.11}, resources[ RunMetadataResource(idstep_run_id, typeMetadataResourceTypes.STEP_RUN), RunMetadataResource(idrun_id, typeMetadataResourceTypes.PIPELINE_RUN), RunMetadataResource(idartifact_version_id, typeMetadataResourceTypes.ARTIFACT_VERSION), RunMetadataResource(idmodel_version_id, typeMetadataResourceTypes.MODEL_VERSION) ] )RunMetadataResource是一个轻量的 pydantic 模型只包含id和type两个字段并重写了__eq__/__hash__因此bulk_log_metadata内部使用set来去重资源见 run_metadata.py。MetadataResourceTypes枚举定义了所有可附加元数据的资源类型见 enums.py。特殊元数据类型ZenML 提供了几种特殊的元数据类型用于以标准化方式表示常见的元数据from zenml import log_metadata from zenml.metadata.metadata_types import StorageSize, DType, Uri, Path log_metadata( metadata{ dataset_source: Uri(gs://my-bucket/datasets/source.csv), # 外部 URI preprocessing_script: Path(/scripts/preprocess.py), # 文件路径 column_types: { age: DType(int), # 数据类型 income: DType(float), score: DType(int) }, processed_data_size: StorageSize(2500000) # 以字节为单位的存储大小 }, infer_artifactTrue )这些特殊类型在源码 metadata_types.py 中定义Uri/Path/DType都是str的子类分别用来标注「URI」「文件路径」「数据类型」语义StorageSize是int的子类表示以字节为单位的存储大小完整的MetadataType联合类型为str | int | float | bool | dict | list | set | tuple | Uri | Path | DType | StorageSize并通过MetadataTypeEnum枚举与数据库中的字符串类型一一映射见 metadata_types.py。这些特殊类型保证元数据以一致、可解释的方式记录并在 ZenML Dashboard 中得到特殊渲染。此外validate_metadata函数metadata_types.py会在写入前对元数据做校验key 过长、值序列化后超过TEXT_FIELD_MAX_LENGTH、或类型不受支持的值都会被跳过并给出警告日志。在 Dashboard 中组织元数据为了改善 ZenML Dashboard 的可视化效果你可以通过「字典套字典」的方式把元数据分组为逻辑区块from zenml import log_metadata from zenml.metadata.metadata_types import StorageSize log_metadata( metadata{ model_metrics: { # Dashboard 中的第一张卡片 accuracy: 0.95, precision: 0.92, recall: 0.90 }, data_details: { # Dashboard 中的第二张卡片 dataset_size: StorageSize(1500000), feature_columns: [age, income, score] } }, artifact_namemy_artifact, artifact_versionversion, )在 ZenML Dashboard 中model_metrics和data_details会分别显示为独立的卡片每张卡片内包含各自的键值对从而更便于导航和解读元数据。可视化与比较元数据ZenML Pro 功能元数据写入之后可以使用 ZenML 的Experiment Comparison实验对比工具跨 run 分析和比较指标。注意元数据对比工具是ZenML Pro 专属功能可参考 getting-started/zenml-pro 了解 Pro 版本能力。对比视图Experiment Comparison 工具提供两种互补的视图来分析 pipeline 元数据1. 表格视图Table View跨 run 比较元数据并带有自动变化追踪。2. 平行坐标图Parallel Coordinates Plot可视化不同指标之间的关系。该工具支持同时对比最多20 个 pipeline run并支持你在 pipeline 中记录的任何数值型元数据float或int。检索Fetch元数据以编程方式检索元数据元数据写入后可以使用 ZenML Client 检索from zenml.client import Client client Client() # 从 step 获取元数据 step client.get_pipeline_run(pipeline_run_id).steps[step_name] step_metadata step.run_metadata[metadata_key] # 从 run 获取元数据 run client.get_pipeline_run(pipeline_run_id) run_metadata run.run_metadata[metadata_key] # 从 artifact 获取元数据 artifact client.get_artifact_version(artifact_name, version) artifact_metadata artifact.run_metadata[metadata_key] # 从 model 获取元数据 model client.get_model_version(model_name, version) model_metadata model.run_metadata[metadata_key]提示当使用特定 key 获取元数据时返回的值始终是该 key 最新的条目底层RunMetadataEntry按创建时间created排序见 run_metadata.py。在 Step 内通过 StepContext 访问上下文StepContext对象是 step 执行期间访问当前pipeline/step run 的句柄。你可以用它读取 run/step 信息、检查上游输入的元数据以及操作 step 输出的 URI、materializer、run metadata 和 tags。它只在以下场景可用在step装饰的函数内部执行期间而非组合期间在on_failure/on_success等 step hooks 内部参见 Hooks 文档在 step 触发的save/loadmaterializer 内部在其他位置调用get_step_context()会抛出RuntimeError。获取上下文的方式是get_step_context()from zenml import step, get_step_context step def trainer(param: int 1): ctx get_step_context() print(run:, ctx.pipeline_run.name, ctx.pipeline_run.id) print(step:, ctx.step_run.name, ctx.step_run.id) print(params:, ctx.step_run.config.parameters)它暴露了以下属性ctx.pipeline→ 本次 run 对应的PipelineResponse便捷属性如果 run 没有对应的 pipeline 对象可能抛出异常ctx.pipeline_run→PipelineRunResponse包含 id、name、status、时间戳等ctx.step_run→StepRunResponse包含 name、parameters通过ctx.step_run.config.parameters访问、statusctx.model→ 配置好的Model从 step 或 pipeline 解析而来未配置则抛出异常ctx.inputs→{input_name: StepRunInputResponse}可以用...[x].run_metadata读取上游元数据ctx.step_name→ 便捷的 step 名字符串。操作 step 输出对于单输出 step 可以省略output_name。对于多输出 step必须传入output_name未命名的输出会被命名为output_1、output_2以此类推。这些方法定义在 step_context.pyget_output_artifact_uri(output_nameNone) - str– 输出 artifact 所在位置写入侧文件等get_output_materializer(output_nameNone, *, custom_materializer_classNone, data_typeNone) - BaseMaterializer– 获取一个已初始化的 materializer传入data_type可从Union[...]materializer 中选择或传custom_materializer_class覆盖add_output_metadata(metadata, output_nameNone)/get_output_metadata(output_nameNone)– 为输出设置/读取 run metadata。返回注解中通过ArtifactConfig(..., run_metadata...)提供的值会与运行时值合并add_output_tags(tags, output_nameNone)/get_output_tags(output_nameNone)/remove_output_tags(tags, output_nameNone)– 管理产出的 artifact version 的 tags。通过ArtifactConfig(..., tags...)配置的 tags 会与运行时 tags 取并集重复项在最终 artifact 中会被去重。最小示例from typing import Annotated, Tuple from zenml import step, get_step_context, log_metadata from zenml.artifacts.artifact_config import ArtifactConfig step def produce(name: str) - Tuple[ Annotated[ str, ArtifactConfig( namecustom_name, run_metadata{config_metadata: bar}, tags[config_tags], ), ], str, ]: ctx get_step_context() # 把元数据和 tags 附加到指定或默认输出 ctx.add_output_metadata({m: 1}, output_namename) ctx.add_output_tags([t1, t1], output_namename) # 重复也没关系 return a, b通过inputs读取上游元数据from zenml import step, get_step_context, log_metadata step def upstream() - int: log_metadata({quality: ok}, infer_artifactTrue) return 42 step def downstream(x: int) - None: md get_step_context().inputs[x].run_metadata assert md[quality] ok这个模式非常适合在 downstream step 中做数据质量校验、分布检查等「上游产出、下游消费」的元数据流转。Hooks 与 Materializers进阶用法from zenml import step, get_step_context from zenml.materializers.base_materializer import BaseMaterializer def on_failure(exc: BaseException): c get_step_context() print(Failed step:, c.step_run.name, -, type(exc).__name__) class ExampleMaterializer(BaseMaterializer): def save(self, data): # step 触发物化时上下文可用 data.meta get_step_context().pipeline.name super().save(data) step(on_failureon_failure) def my_step(): raise ValueError(boom)常见错误在非运行中的 step 外调用get_step_context()会抛出RuntimeError输出相关 helper 会抛出StepContextError触发条件包括step 没有输出、多输出 step 上省略了output_name、引用了未知的output_name。在 Pipeline 组合期间访问上下文在 pipeline 组合期间可以使用PipelineContext访问 pipeline 配置from zenml import pipeline, get_pipeline_context pipeline( extra{ model_configs: [ (sklearn.tree, DecisionTreeClassifier), (sklearn.ensemble, RandomForestClassifier), ] } ) def my_pipeline(): # 获取 pipeline 上下文 context get_pipeline_context() # 访问配置 model_configs context.extra[model_configs] # 利用配置动态创建 step for i, (model_package, model_class) in enumerate(model_configs): train_model( model_packagemodel_package, model_classmodel_class, idftrain_model_{i} )这个模式非常适合基于配置驱动地动态生成多个同类 step例如批量训练不同模型并借助extra字典传递任意结构化配置。Best Practices最佳实践要充分发挥 ZenML 元数据能力建议遵循以下原则使用一致的 key为组织定义标准的元数据 key保证一致性分组相关元数据用嵌套字典在 Dashboard 中创建逻辑分组善用特殊类型使用 ZenML 的特殊元数据类型Uri、Path、DType、StorageSize实现标准化表示记录相关信息只记录有助于复现、理解和决策的元数据考虑自动化为标准指标和信息设置自动元数据写入例如在公共基类 step 或 hooks 中统一记录 Git 版本、Python 版本、环境信息与 tags 结合将元数据与 tags 配合使用构建完整的组织体系元数据侧重结构化键值tags 侧重灵活的标签分类。Conclusion结语ZenML 的元数据能力为 ML 工作流提供了强大的上下文增强手段。通过为 steps、runs、artifacts 和 models 追踪附加细节你可以更深入地洞察实验、做出更明智的决策并确保 ML pipeline 的可复现性。从log_metadata的实体定位逻辑见 metadata_utils.py到bulk_log_metadata的多实体批量写入与 Identifier 校验见 param_groups.py再到StepContext的输出元数据操作见 step_context.py这套机制在源码层面环环相扣。建议在真实 pipeline 中从「step 内记录评估指标 artifact 记录数据统计」开始实践逐步扩展到 run 级环境信息与 model 级部署信息最终形成一套可比较、可审计的完整元数据体系。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考