ZenML MLOps 起步实战用三条 Pipeline 完成从特征工程到生产推理的完整闭环【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml导读本文以 ZenML 仓库中的mlops_starter示例项目为主线带你从零构建一套最小可用的 MLOps 工程一条特征工程 Pipeline 负责加载与预处理数据一条训练 Pipeline 负责训练、评估并自动晋升最优模型一条批量推理 Pipeline 负责消费生产环境的模型产出预测。读完本文你将掌握用 ZenML 组织 Pipeline/Step 代码、借助 Model Control Plane 实现模型版本管理与自动晋升、以及让推理与训练共用同一套预处理逻辑的完整实践方案。一、项目概览一个最小的 MLOps 骨架examples/mlops_starter是 ZenML 仓库中一个极简的 MLOps 示例工程其目标是帮助初学者快速理解如何把 ML 工作流真正放进生产环境。它由三个核心 Pipeline 组成特征工程 Pipelinefeature_engineering加载数据并完成清洗、切分、预处理产出训练集与测试集工件训练 Pipelinetraining消费预处理后的数据集训练模型、评估精度并决定是否晋升到production阶段批量推理 Pipelineinference用标记为production的模型对新数据本示例中为原始数据集的一个随机子集执行预测。整个工程被刻意设计得足够精简以便你能把它的结构映射到任何真实项目上。工程内代码组织如下均为仓库根目录下的相对路径模块路径职责Pipeline 定义pipelines/用pipeline装饰器串联各 StepStep 实现steps/用step装饰器实现具体逻辑运行入口run.py基于 Click 的 CLI按参数触发不同 Pipeline配置目录configs/YAML 配置环境设置、模型控制平面、Pipeline 参数工具类utils/preprocess.py自定义 sklearn Transformer沿这条路径走下来你还能顺带学会三件事如何把代码组织成 MLOps Pipeline、如何让数据/模型等工件被自动版本化、追踪与缓存、以及如何让模型从开发平滑过渡到生产。二、运行环境准备Colab 与本地两种方式2.1 零安装体验Google Colab如果不希望在本机安装任何依赖可以直接在 Colab 中打开mlops_starter/quickstart.ipynb交互式体验 ZenML无需注册或安装。该 Notebook 与本文介绍的三条 Pipeline 完全对应适合先跑通整体流程再回到本地深入。2.2 本地运行安装与初始化在本地运行需要先安装 ZenML 并克隆仓库以下命令中的仓库地址用于说明 git clone 场景# 安装 ZenML含本地 server用于 Dashboard 访问 pip install zenml[server] # 克隆 ZenML 仓库并进入示例目录 git clone https://github.com/zenml-io/zenml.git cd zenml/examples/mlops_starter项目依赖可参考 requirements.txt除zenml[server]0.50.0外还需要notebook、scikit-learn、pyarrow与pandas。安装完成后有两种运行方式方式一Jupyter Notebook 交互式探索pip install notebook jupyter notebook # 打开 quickstart.ipynb方式二Python 脚本一键执行整条链路# 安装所需的 ZenML 集成sklearn、pandas zenml integration install sklearn pandas -y # 初始化 ZenML创建本地 stack 与存储 zenml init # 启动本地 ZenServer获得 Dashboard 访问能力 zenml login --local # 运行特征工程 Pipeline python run.py --feature-pipeline # 运行训练 Pipeline python run.py --training-pipeline # 指定数据集工件版本运行训练 Pipeline使用版本化工件 python run.py --training-pipeline --train-dataset-version-name1 --test-dataset-version-name1 # 运行批量推理 Pipeline python run.py --inference-pipeline其中zenml integration install sklearn pandas -y会为本地环境安装 scikit-learn 与 pandas 相关依赖zenml init负责在当前目录初始化一个默认的 ZenML stack编排器、工件存储、元数据存储等zenml login --local启动本地服务端以便在浏览器 Dashboard 中查看 Pipeline 运行记录、工件与模型。三、Step 1加载数据并完成特征工程ML 项目的第一步几乎总是把数据导入并整理成适合训练的形状。本例使用的是 UCI 机器学习库公开的Breast Cancer Wisconsin (Diagnostic)数据集任务是二分类——预测患者是否被诊断为乳腺癌。特征工程 Pipeline 的完整定义位于 pipelines/feature_engineering.py它通过将三个 Step 的输出作为下一个 Step 的输入串成一条链路 pipeline def feature_engineering( test_size: float 0.2, drop_na: Optional[bool] None, normalize: Optional[bool] None, drop_columns: Optional[List[str]] None, target: Optional[str] target, random_state: int 17, ): raw_data data_loader(random_staterandom_state, targettarget) dataset_trn, dataset_tst data_splitter( datasetraw_data, test_sizetest_size, ) dataset_trn, dataset_tst, _ data_preprocessor( dataset_trndataset_trn, dataset_tstdataset_tst, drop_nadrop_na, normalizenormalize, drop_columnsdrop_columns, targettarget, random_staterandom_state, ) return dataset_trn, dataset_tstPipeline 的参数中test_size控制验证集比例0.0~1.0drop_na控制是否丢弃含空值的行normalize控制是否用 MinMaxScaler 归一化数值列drop_columns指定要剔除的列target指定目标列名random_state固定随机种子保证可复现。三个 Step 各自承担明确职责定义在 steps/ 目录data_loader.py调用load_breast_cancer(as_frameTrue)加载数据。值得注意的细节是它把约 5% 的样本inference_size int(len(dataset.target) * 0.05)抽成一个推理子集训练阶段会先把这个子集从主数据中剔除dataset.drop(inference_subset.index, inplaceTrue)从而保证推理 Pipeline 使用的实时数据在训练阶段从未见过模拟真实的生产数据漂移场景。当is_inferenceTrue时该 Step 返回推理子集并移除目标列。data_splitter.py调用sklearn.model_selection.train_test_split以test_size比例切分数据固定random_state42并打乱顺序返回raw_dataset_trn与raw_dataset_tst两个 DataFrame 工件。data_preprocessor.py构建一个 sklearnPipeline按需追加NADropper去空值、ColumnsDropper删列、MinMaxScaler归一化与DataFrameCaster把 numpy 结果转回带列名的 DataFrame避免启用drop_columns时出现形状/列名不匹配等自定义 Transformer。这些 Transformer 定义在 utils/preprocess.py均实现了 sklearn 标准的fit/transform接口。预处理后该 Step 还会调用log_metadata把random_state与target记录为preprocess_pipeline工件的运行元数据——这个细节是推理 Pipeline 后续能复现训练时处理逻辑的关键。执行特征工程 Pipelinepython run.py --feature-pipeline运行结束后run.py会通过Client.get_artifact_version(...)查询最新工件并打印类似如下的日志The latest feature engineering pipeline produced the following artifacts: 1. Train Dataset - Name: dataset_trn, Version Name: 1 2. Test Dataset: Name: dataset_tst, Version Name: 1这两个版本号dataset_trn:1、dataset_tst:1会在下一步训练时被引用用于消费固定版本的数据这正是 ZenML 工件版本化的典型用法。四、Step 2训练 Pipeline 与 Model Control Plane数据就绪后就可以训练模型感受任务的难度并建立基线。Breast Cancer 数据集本身足够复杂难以训练出完美模型但足以让你看清合理的基线长什么样。训练 Pipeline 选取了两个 sklearn 开箱即用的模型——SGD Classifier与Random Forest Classifier——在同一份数据上训练并对比性能。训练 Pipeline 定义在 pipelines/training.py pipeline def training( train_dataset_id: Optional[UUID] None, test_dataset_id: Optional[UUID] None, target: Optional[str] target, model_type: Optional[str] sgd, ): # 未指定数据集版本时直接内联执行特征工程获取最新数据 if train_dataset_id is None or test_dataset_id is None: dataset_trn, dataset_tst feature_engineering() else: # 否则按 ID 从工件存储中拉取指定版本 client Client() dataset_trn client.get_artifact_version(name_id_or_prefixtrain_dataset_id) dataset_tst client.get_artifact_version(name_id_or_prefixtest_dataset_id) model model_trainer(dataset_trndataset_trn, targettarget, model_typemodel_type) acc model_evaluator(modelmodel, dataset_trndataset_trn, dataset_tstdataset_tst, targettarget) model_promoter(accuracyacc)这个设计非常实用第一次训练可以不带任何 ID 直接运行此时会内联执行特征工程拿到最新数据之后想复现某次实验则传入--train-dataset-version-name/--test-dataset-version-name训练 Pipeline 会从工件存储中按 ID 取出固定版本的数据保证实验的可追溯性。运行训练显式指定版本也可以省略版本参数直接使用最新版本python run.py --training-pipeline --train-dataset-version-name 1 --test-dataset-version-name 1run.py会连续执行两次训练 Pipeline一次加载 training_sgd.yaml一次加载 training_rf.yaml从而训练出sgd与rf两个模型版本。训练链路中的三个 Stepmodel_trainer.py根据model_type实例化SGDClassifier()或RandomForestClassifier()其他值抛出ValueError在剔除目标列的训练集上fit。其输出通过ArtifactConfig(namesklearn_classifier, is_model_artifactTrue)声明为名为sklearn_classifier的模型工件——这是后续模型控制平面追踪、评估与推理读取该模型的关键命名。model_evaluator.py分别计算训练集与测试集精度model.score支持min_train_accuracy/min_test_accuracy阈值参数——精度低于阈值时输出告警日志如需快速失败也可以改成抛异常让后续 Step 跳过。评估结果通过log_metadata写入sklearn_classifier工件的train_accuracy/test_accuracy元数据供模型晋升逻辑比较。model_promoter.py条件晋升逻辑详见下一节。Model Control Plane模型控制平面是 ZenML 提供的模型中央注册表。每次训练 Pipeline 运行都会产生一个 ZenML Model Version可以在本地随时查看zenml model list运行后你会看到名为breast_cancer_classifier的模型下出现了sgd与rf两个版本。模型名、版本、许可证、描述、标签等元数据都可以在 configs/ 下的 YAML 配置中预先声明例如 training_sgd.yaml# configuration of the Model Control Plane model: name: breast_cancer_classifier version: sgd license: Apache 2.0 description: A breast cancer classifier tags: [breast_cancer, classifier]这类能力把指标写入模型元数据、把模型持久化到模型注册表等都还有很大的扩展空间感兴趣的读者可以进一步查阅 ZenML 官方文档深入了解。五、Step 3把最优模型晋升到生产模型控制平面让选择生产模型变成一行命令。你可以直接把某个模型版本的stage标记为productionzenml model version update breast_cancer_classifier rf --stage production不过为了演示手动晋升流程的清晰性README 特意指出训练 Pipeline 本身已经内置了自动化晋升逻辑。看 steps/model_promoter.py 的源码即可确认这一点其核心判断流程如下若当前模型精度低于 80%accuracy 0.8不晋升否则通过get_step_context().model取得当前模型版本用Client().get_model_version(current_model.name, stage)尝试获取当前production阶段已有的模型版本若存在生产版本则读取其sklearn_classifier工件上的test_accuracy元数据只有当新模型的测试精度高于现有生产模型时才调用current_model.set_stage(stage, forceTrue)晋升若生产阶段尚无模型捕获KeyError则直接晋升当前版本。这套评估 → 与现有生产指标对比 → 胜者晋升的流程就是训练到生产自动化晋升的最小实现。也就是说运行python run.py --training-pipeline时模型是否被推进到production完全由测试集表现自动决定手动命令只是提供了一种显式干预的手段。六、Step 4在生产中消费模型模型晋升到production后批量推理 Pipeline 就可以直接消费正确的模型版本了。这里的关键挑战在于推理阶段必须复现训练阶段的全部预处理逻辑——不仅要把模型加载进内存还要把特征工程阶段拟合好的预处理 Pipeline 一并加载保证线上数据经历与训练完全一致的变换。推理 Pipeline 定义在 pipelines/inference.py它充分利用了 ZenML 的模型上下文能力 pipeline def inference(random_state: int, target: str): # 从当前模型上下文production 版本中取出模型工件 model get_pipeline_context().model.get_artifact(sklearn_classifier) # 取出与该模型版本关联的预处理 Pipeline 工件 preprocess_pipeline get_pipeline_context().model.get_artifact(preprocess_pipeline) df_inference data_loader(random_staterandom_state, is_inferenceTrue) df_inference inference_preprocessor( dataset_infdf_inference, preprocess_pipelinepreprocess_pipeline, targettarget, ) inference_predict(modelmodel, dataset_infdf_inference)链路背后的机制非常值得注意get_pipeline_context().model返回的是当前 Pipeline 配置中所指向的模型版本。run.py会先读取 inference.yaml 中的model配置name: breast_cancer_classifier、version: production通过Client.get_model_version(...)定位到生产模型再启动 Pipeline——因此用哪个模型做推理完全由模型控制平面的production阶段决定推理代码里不需要写死任何模型 ID。由于模型工件sklearn_classifier与预处理工件preprocess_pipeline都通过ArtifactConfig与log_metadata被关联到了对应的模型版本上推理时只需要model.get_artifact(...)就能把它们一起取出实现了训练配置 → 生产推理的完整可追溯。预处理参数的自举run.py还从preprocess_pipeline工件的运行元数据中读取random_state与target这两项是特征工程阶段由log_metadata写入的作为参数传给推理 Pipeline确保抽样与列处理与训练完全一致。推理链路中的两个 Stepinference_preprocessor.py接收推理数据与训练好的preprocess_pipeline。由于推理数据没有目标列而 sklearn Pipeline 可能依赖该列的存在这里先人为补一个全 1 的target列完成transform再在变换后将其删除——一个处理 Pipeline 列依赖的小技巧。inference_predict.py把模型加载进内存后直接model.predict(dataset_inf)输出以predictions命名的 pandas Series 工件。执行推理python run.py --inference-pipelineZenML 会自动把所有相关工件包括这条推理 Pipeline 产出的预测结果关联到production模型版本上。至此从训练到推理的 MLOps 闭环就完整了数据版本化 → 模型训练与评估 → 自动晋升 → 生产版本消费 → 预测结果回流到模型记录每一步都留下了可供追溯的工件与元数据。七、配置驱动读懂四份 YAML 文件整个示例工程采用代码 配置分离的组织方式四份 YAML 配置集中定义了环境、模型控制平面与 Pipeline 参数位于 configs/配置文件作用feature_engineering.yaml特征工程 Pipelinetest_size: 0.35即 65%/35% 切分training_sgd.yamlSGD 训练模型控制平面版本sgdparameters.model_type: sgdtraining_rf.yamlRF 训练模型控制平面版本rfparameters.model_type: rfinference.yaml推理模型控制平面版本production即总是消费生产模型每份文件都包含两部分公共配置# 环境配置声明容器运行时需要的集成与额外依赖 settings: docker: required_integrations: - sklearn - pandas requirements: - pyarrow # 模型控制平面配置 model: name: breast_cancer_classifier version: production # 各文件不同sgd / rf / production license: Apache 2.0 description: A breast cancer classifier tags: [breast_cancer, classifier]settings.docker段在把 Pipeline 搬到容器/远端执行时尤其重要——它声明了镜像构建时需要预装的 ZenML 集成与 pip 依赖从而保证 Step 代码在任何执行环境中的行为一致。训练与推理文件还额外通过parameters段注入 Pipeline 参数如model_type实现同一份 Pipeline 代码、不同配置实例化。在run.py中这些配置通过pipeline.with_options(config_path...)被应用到每次运行同时 CLI 还提供了若干有用的运行选项例如--no-cache可关闭 Step 级缓存强制重跑enable_cacheFalse。值得强调的是推理 Pipeline 在run.py中被强制设置为enable_cache: False以保证每次推理都真正消费最新数据而非命中缓存。八、小结与进阶路径至此你已经完整走过了训练两个模型 → 在测试集上评估 → 用模型控制平面注册并晋升最优模型 → 消费生产模型产出批量预测的 MLOps 主链路同时也接触到了 ZenML 的几个核心抽象Pipeline/Step 装饰器、工件自动版本化与缓存、模型控制平面的阶段管理、以及通过log_metadata与get_artifact实现的训练/推理一致性保障。如果你想继续深入可以从以下几个方面拓展阅读本仓库的 Production Guide 文档对应docs/user-guide/production-guide/目录下的内容了解如何把同一条 Pipeline 迁移到云端生产环境对照 pipelines/ 与 steps/ 的源码尝试为data_preprocessor增加更多可配置的预处理 Transformer并观察工件版本与缓存行为的变化在zenml model list与zenml model version list等命令的输出基础上进一步研究模型注册表集成把生产模型同步到外部模型仓库。ZenML 的定位是从 Pipeline 到 Agent 的统一 AI 平台而mlops_starter正是理解它核心设计哲学——用可版本化、可缓存、可追溯的 Pipeline 把 ML 从开发带到生产——的最佳起点。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ZenML MLOps 起步实战用三条 Pipeline 完成从特征工程到生产推理的完整闭环【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml导读本文以 ZenML 仓库中的mlops_starter示例项目为主线带你从零构建一套最小可用的 MLOps 工程一条特征工程 Pipeline 负责加载与预处理数据一条训练 Pipeline 负责训练、评估并自动晋升最优模型一条批量推理 Pipeline 负责消费生产环境的模型产出预测。读完本文你将掌握用 ZenML 组织 Pipeline/Step 代码、借助 Model Control Plane 实现模型版本管理与自动晋升、以及让推理与训练共用同一套预处理逻辑的完整实践方案。一、项目概览一个最小的 MLOps 骨架examples/mlops_starter是 ZenML 仓库中一个极简的 MLOps 示例工程其目标是帮助初学者快速理解如何把 ML 工作流真正放进生产环境。它由三个核心 Pipeline 组成特征工程 Pipelinefeature_engineering加载数据并完成清洗、切分、预处理产出训练集与测试集工件训练 Pipelinetraining消费预处理后的数据集训练模型、评估精度并决定是否晋升到production阶段批量推理 Pipelineinference用标记为production的模型对新数据本示例中为原始数据集的一个随机子集执行预测。整个工程被刻意设计得足够精简以便你能把它的结构映射到任何真实项目上。工程内代码组织如下均为仓库根目录下的相对路径模块路径职责Pipeline 定义pipelines/用pipeline装饰器串联各 StepStep 实现steps/用step装饰器实现具体逻辑运行入口run.py基于 Click 的 CLI按参数触发不同 Pipeline配置目录configs/YAML 配置环境设置、模型控制平面、Pipeline 参数工具类utils/preprocess.py自定义 sklearn Transformer沿这条路径走下来你还能顺带学会三件事如何把代码组织成 MLOps Pipeline、如何让数据/模型等工件被自动版本化、追踪与缓存、以及如何让模型从开发平滑过渡到生产。二、运行环境准备Colab 与本地两种方式2.1 零安装体验Google Colab如果不希望在本机安装任何依赖可以直接在 Colab 中打开mlops_starter/quickstart.ipynb交互式体验 ZenML无需注册或安装。该 Notebook 与本文介绍的三条 Pipeline 完全对应适合先跑通整体流程再回到本地深入。2.2 本地运行安装与初始化在本地运行需要先安装 ZenML 并克隆仓库以下命令中的仓库地址用于说明 git clone 场景# 安装 ZenML含本地 server用于 Dashboard 访问 pip install zenml[server] # 克隆 ZenML 仓库并进入示例目录 git clone https://github.com/zenml-io/zenml.git cd zenml/examples/mlops_starter项目依赖可参考 requirements.txt除zenml[server]0.50.0外还需要notebook、scikit-learn、pyarrow与pandas。安装完成后有两种运行方式方式一Jupyter Notebook 交互式探索pip install notebook jupyter notebook # 打开 quickstart.ipynb方式二Python 脚本一键执行整条链路# 安装所需的 ZenML 集成sklearn、pandas zenml integration install sklearn pandas -y # 初始化 ZenML创建本地 stack 与存储 zenml init # 启动本地 ZenServer获得 Dashboard 访问能力 zenml login --local # 运行特征工程 Pipeline python run.py --feature-pipeline # 运行训练 Pipeline python run.py --training-pipeline # 指定数据集工件版本运行训练 Pipeline使用版本化工件 python run.py --training-pipeline --train-dataset-version-name1 --test-dataset-version-name1 # 运行批量推理 Pipeline python run.py --inference-pipeline其中zenml integration install sklearn pandas -y会为本地环境安装 scikit-learn 与 pandas 相关依赖zenml init负责在当前目录初始化一个默认的 ZenML stack编排器、工件存储、元数据存储等zenml login --local启动本地服务端以便在浏览器 Dashboard 中查看 Pipeline 运行记录、工件与模型。三、Step 1加载数据并完成特征工程ML 项目的第一步几乎总是把数据导入并整理成适合训练的形状。本例使用的是 UCI 机器学习库公开的Breast Cancer Wisconsin (Diagnostic)数据集任务是二分类——预测患者是否被诊断为乳腺癌。特征工程 Pipeline 的完整定义位于 pipelines/feature_engineering.py它通过将三个 Step 的输出作为下一个 Step 的输入串成一条链路 pipeline def feature_engineering( test_size: float 0.2, drop_na: Optional[bool] None, normalize: Optional[bool] None, drop_columns: Optional[List[str]] None, target: Optional[str] target, random_state: int 17, ): raw_data data_loader(random_staterandom_state, targettarget) dataset_trn, dataset_tst data_splitter( datasetraw_data, test_sizetest_size, ) dataset_trn, dataset_tst, _ data_preprocessor( dataset_trndataset_trn, dataset_tstdataset_tst, drop_nadrop_na, normalizenormalize, drop_columnsdrop_columns, targettarget, random_staterandom_state, ) return dataset_trn, dataset_tstPipeline 的参数中test_size控制验证集比例0.0~1.0drop_na控制是否丢弃含空值的行normalize控制是否用 MinMaxScaler 归一化数值列drop_columns指定要剔除的列target指定目标列名random_state固定随机种子保证可复现。三个 Step 各自承担明确职责定义在 steps/ 目录data_loader.py调用load_breast_cancer(as_frameTrue)加载数据。值得注意的细节是它把约 5% 的样本inference_size int(len(dataset.target) * 0.05)抽成一个推理子集训练阶段会先把这个子集从主数据中剔除dataset.drop(inference_subset.index, inplaceTrue)从而保证推理 Pipeline 使用的实时数据在训练阶段从未见过模拟真实的生产数据漂移场景。当is_inferenceTrue时该 Step 返回推理子集并移除目标列。data_splitter.py调用sklearn.model_selection.train_test_split以test_size比例切分数据固定random_state42并打乱顺序返回raw_dataset_trn与raw_dataset_tst两个 DataFrame 工件。data_preprocessor.py构建一个 sklearnPipeline按需追加NADropper去空值、ColumnsDropper删列、MinMaxScaler归一化与DataFrameCaster把 numpy 结果转回带列名的 DataFrame避免启用drop_columns时出现形状/列名不匹配等自定义 Transformer。这些 Transformer 定义在 utils/preprocess.py均实现了 sklearn 标准的fit/transform接口。预处理后该 Step 还会调用log_metadata把random_state与target记录为preprocess_pipeline工件的运行元数据——这个细节是推理 Pipeline 后续能复现训练时处理逻辑的关键。执行特征工程 Pipelinepython run.py --feature-pipeline运行结束后run.py会通过Client.get_artifact_version(...)查询最新工件并打印类似如下的日志The latest feature engineering pipeline produced the following artifacts: 1. Train Dataset - Name: dataset_trn, Version Name: 1 2. Test Dataset: Name: dataset_tst, Version Name: 1这两个版本号dataset_trn:1、dataset_tst:1会在下一步训练时被引用用于消费固定版本的数据这正是 ZenML 工件版本化的典型用法。四、Step 2训练 Pipeline 与 Model Control Plane数据就绪后就可以训练模型感受任务的难度并建立基线。Breast Cancer 数据集本身足够复杂难以训练出完美模型但足以让你看清合理的基线长什么样。训练 Pipeline 选取了两个 sklearn 开箱即用的模型——SGD Classifier与Random Forest Classifier——在同一份数据上训练并对比性能。训练 Pipeline 定义在 pipelines/training.py pipeline def training( train_dataset_id: Optional[UUID] None, test_dataset_id: Optional[UUID] None, target: Optional[str] target, model_type: Optional[str] sgd, ): # 未指定数据集版本时直接内联执行特征工程获取最新数据 if train_dataset_id is None or test_dataset_id is None: dataset_trn, dataset_tst feature_engineering() else: # 否则按 ID 从工件存储中拉取指定版本 client Client() dataset_trn client.get_artifact_version(name_id_or_prefixtrain_dataset_id) dataset_tst client.get_artifact_version(name_id_or_prefixtest_dataset_id) model model_trainer(dataset_trndataset_trn, targettarget, model_typemodel_type) acc model_evaluator(modelmodel, dataset_trndataset_trn, dataset_tstdataset_tst, targettarget) model_promoter(accuracyacc)这个设计非常实用第一次训练可以不带任何 ID 直接运行此时会内联执行特征工程拿到最新数据之后想复现某次实验则传入--train-dataset-version-name/--test-dataset-version-name训练 Pipeline 会从工件存储中按 ID 取出固定版本的数据保证实验的可追溯性。运行训练显式指定版本也可以省略版本参数直接使用最新版本python run.py --training-pipeline --train-dataset-version-name 1 --test-dataset-version-name 1run.py会连续执行两次训练 Pipeline一次加载 training_sgd.yaml一次加载 training_rf.yaml从而训练出sgd与rf两个模型版本。训练链路中的三个 Stepmodel_trainer.py根据model_type实例化SGDClassifier()或RandomForestClassifier()其他值抛出ValueError在剔除目标列的训练集上fit。其输出通过ArtifactConfig(namesklearn_classifier, is_model_artifactTrue)声明为名为sklearn_classifier的模型工件——这是后续模型控制平面追踪、评估与推理读取该模型的关键命名。model_evaluator.py分别计算训练集与测试集精度model.score支持min_train_accuracy/min_test_accuracy阈值参数——精度低于阈值时输出告警日志如需快速失败也可以改成抛异常让后续 Step 跳过。评估结果通过log_metadata写入sklearn_classifier工件的train_accuracy/test_accuracy元数据供模型晋升逻辑比较。model_promoter.py条件晋升逻辑详见下一节。Model Control Plane模型控制平面是 ZenML 提供的模型中央注册表。每次训练 Pipeline 运行都会产生一个 ZenML Model Version可以在本地随时查看zenml model list运行后你会看到名为breast_cancer_classifier的模型下出现了sgd与rf两个版本。模型名、版本、许可证、描述、标签等元数据都可以在 configs/ 下的 YAML 配置中预先声明例如 training_sgd.yaml# configuration of the Model Control Plane model: name: breast_cancer_classifier version: sgd license: Apache 2.0 description: A breast cancer classifier tags: [breast_cancer, classifier]这类能力把指标写入模型元数据、把模型持久化到模型注册表等都还有很大的扩展空间感兴趣的读者可以进一步查阅 ZenML 官方文档深入了解。五、Step 3把最优模型晋升到生产模型控制平面让选择生产模型变成一行命令。你可以直接把某个模型版本的stage标记为productionzenml model version update breast_cancer_classifier rf --stage production不过为了演示手动晋升流程的清晰性README 特意指出训练 Pipeline 本身已经内置了自动化晋升逻辑。看 steps/model_promoter.py 的源码即可确认这一点其核心判断流程如下若当前模型精度低于 80%accuracy 0.8不晋升否则通过get_step_context().model取得当前模型版本用Client().get_model_version(current_model.name, stage)尝试获取当前production阶段已有的模型版本若存在生产版本则读取其sklearn_classifier工件上的test_accuracy元数据只有当新模型的测试精度高于现有生产模型时才调用current_model.set_stage(stage, forceTrue)晋升若生产阶段尚无模型捕获KeyError则直接晋升当前版本。这套评估 → 与现有生产指标对比 → 胜者晋升的流程就是训练到生产自动化晋升的最小实现。也就是说运行python run.py --training-pipeline时模型是否被推进到production完全由测试集表现自动决定手动命令只是提供了一种显式干预的手段。六、Step 4在生产中消费模型模型晋升到production后批量推理 Pipeline 就可以直接消费正确的模型版本了。这里的关键挑战在于推理阶段必须复现训练阶段的全部预处理逻辑——不仅要把模型加载进内存还要把特征工程阶段拟合好的预处理 Pipeline 一并加载保证线上数据经历与训练完全一致的变换。推理 Pipeline 定义在 pipelines/inference.py它充分利用了 ZenML 的模型上下文能力 pipeline def inference(random_state: int, target: str): # 从当前模型上下文production 版本中取出模型工件 model get_pipeline_context().model.get_artifact(sklearn_classifier) # 取出与该模型版本关联的预处理 Pipeline 工件 preprocess_pipeline get_pipeline_context().model.get_artifact(preprocess_pipeline) df_inference data_loader(random_staterandom_state, is_inferenceTrue) df_inference inference_preprocessor( dataset_infdf_inference, preprocess_pipelinepreprocess_pipeline, targettarget, ) inference_predict(modelmodel, dataset_infdf_inference)链路背后的机制非常值得注意get_pipeline_context().model返回的是当前 Pipeline 配置中所指向的模型版本。run.py会先读取 inference.yaml 中的model配置name: breast_cancer_classifier、version: production通过Client.get_model_version(...)定位到生产模型再启动 Pipeline——因此用哪个模型做推理完全由模型控制平面的production阶段决定推理代码里不需要写死任何模型 ID。由于模型工件sklearn_classifier与预处理工件preprocess_pipeline都通过ArtifactConfig与log_metadata被关联到了对应的模型版本上推理时只需要model.get_artifact(...)就能把它们一起取出实现了训练配置 → 生产推理的完整可追溯。预处理参数的自举run.py还从preprocess_pipeline工件的运行元数据中读取random_state与target这两项是特征工程阶段由log_metadata写入的作为参数传给推理 Pipeline确保抽样与列处理与训练完全一致。推理链路中的两个 Stepinference_preprocessor.py接收推理数据与训练好的preprocess_pipeline。由于推理数据没有目标列而 sklearn Pipeline 可能依赖该列的存在这里先人为补一个全 1 的target列完成transform再在变换后将其删除——一个处理 Pipeline 列依赖的小技巧。inference_predict.py把模型加载进内存后直接model.predict(dataset_inf)输出以predictions命名的 pandas Series 工件。执行推理python run.py --inference-pipelineZenML 会自动把所有相关工件包括这条推理 Pipeline 产出的预测结果关联到production模型版本上。至此从训练到推理的 MLOps 闭环就完整了数据版本化 → 模型训练与评估 → 自动晋升 → 生产版本消费 → 预测结果回流到模型记录每一步都留下了可供追溯的工件与元数据。七、配置驱动读懂四份 YAML 文件整个示例工程采用代码 配置分离的组织方式四份 YAML 配置集中定义了环境、模型控制平面与 Pipeline 参数位于 configs/配置文件作用feature_engineering.yaml特征工程 Pipelinetest_size: 0.35即 65%/35% 切分training_sgd.yamlSGD 训练模型控制平面版本sgdparameters.model_type: sgdtraining_rf.yamlRF 训练模型控制平面版本rfparameters.model_type: rfinference.yaml推理模型控制平面版本production即总是消费生产模型每份文件都包含两部分公共配置# 环境配置声明容器运行时需要的集成与额外依赖 settings: docker: required_integrations: - sklearn - pandas requirements: - pyarrow # 模型控制平面配置 model: name: breast_cancer_classifier version: production # 各文件不同sgd / rf / production license: Apache 2.0 description: A breast cancer classifier tags: [breast_cancer, classifier]settings.docker段在把 Pipeline 搬到容器/远端执行时尤其重要——它声明了镜像构建时需要预装的 ZenML 集成与 pip 依赖从而保证 Step 代码在任何执行环境中的行为一致。训练与推理文件还额外通过parameters段注入 Pipeline 参数如model_type实现同一份 Pipeline 代码、不同配置实例化。在run.py中这些配置通过pipeline.with_options(config_path...)被应用到每次运行同时 CLI 还提供了若干有用的运行选项例如--no-cache可关闭 Step 级缓存强制重跑enable_cacheFalse。值得强调的是推理 Pipeline 在run.py中被强制设置为enable_cache: False以保证每次推理都真正消费最新数据而非命中缓存。八、小结与进阶路径至此你已经完整走过了训练两个模型 → 在测试集上评估 → 用模型控制平面注册并晋升最优模型 → 消费生产模型产出批量预测的 MLOps 主链路同时也接触到了 ZenML 的几个核心抽象Pipeline/Step 装饰器、工件自动版本化与缓存、模型控制平面的阶段管理、以及通过log_metadata与get_artifact实现的训练/推理一致性保障。如果你想继续深入可以从以下几个方面拓展阅读本仓库的 Production Guide 文档对应docs/user-guide/production-guide/目录下的内容了解如何把同一条 Pipeline 迁移到云端生产环境对照 pipelines/ 与 steps/ 的源码尝试为data_preprocessor增加更多可配置的预处理 Transformer并观察工件版本与缓存行为的变化在zenml model list与zenml model version list等命令的输出基础上进一步研究模型注册表集成把生产模型同步到外部模型仓库。ZenML 的定位是从 Pipeline 到 Agent 的统一 AI 平台而mlops_starter正是理解它核心设计哲学——用可版本化、可缓存、可追溯的 Pipeline 把 ML 从开发带到生产——的最佳起点。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考