企业官网建设流程全解析

ZenML 用户指南：从 MLOps 基础到生产级 LLM Agent 的完整学习路线

【免费下载链接】zenmlZenML 🙏: One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml

本文是对 ZenML 开源仓库 docs/book/user-guide 用户指南目录体系的系统性解读。该目录覆盖从「创建第一个 ML Pipeline」到「生产化部署」「LLMOps 与 Agent 平台 Kitaru」的完整进阶路径，是理解 ZenML 如何将机器学习流水线从开发环境平滑迁移到云端生产环境的权威入口。

ZenML 的用户指南（User Guide）是整个开源项目src/zenml之外最重要的学习资产：它不像 API 文档那样零散罗列接口，而是按照 MLOps 工程师的真实成长曲线，组织成「Starter Guide → Production Guide → LLMOps Guide → Tutorials & Best Practices → Projects & Examples」五层递进的学习路线。无论你是刚接触 MLOps 的数据科学家，还是要在企业内部落地 AI 平台的工程师，这套指南都能告诉你：如何在 ZenML 中定义 Step 与 Pipeline、如何利用缓存加速迭代、如何把本地流水线迁移到云端 Stack、如何用 ZenML 构建 RAG / LLM 微调流水线，以及如何与 ZenML 的姊妹项目 Kitaru 一起构建生产级 AI Agent。

本文将以该目录为骨架，结合仓库内真实源码与示例，带你走完这条从入门到实战的完整路线。

一、指南总览：五层递进的学习体系

ZenML 用户指南的整体结构记录在 docs/book/user-guide/toc.md 中，主要包含以下五个板块：

其中 LLMOps Guide 还专门引出了 ZenML 的姊妹项目Kitaru——一个用于生产级 AI Agent 的平台：把真实运行录制为 session，在真实代码上重放，把「失败」变成「回归检查」。ZenML 负责 ML Pipeline，Kitaru 负责 Agent，二者职责清晰互补。

二、Starter Guide：从零搭建第一个 ML Pipeline

2.1 环境准备与安装

在开始之前，需要先完成 ZenML 的安装与初始化（见 docs/book/getting-started/installation.md 的相关说明）：

pip install "zenml[server]" zenml login --local # 会在本地启动 Dashboard

强烈建议在新项目根目录执行zenml init，它告诉 ZenML 在远程运行流水线时应该包含哪些文件。

2.2 最小的 Step + Pipeline 示例

ZenML 的核心抽象只有两个装饰器：@step把普通函数变成流水线中的可执行单元，@pipeline把这些单元按数据依赖串联成端到端流水线。来自 docs/book/user-guide/starter-guide/create-an-ml-pipeline.md 的最小示例：

from zenml import pipeline, step @step def load_data() -> dict: """模拟加载训练数据和标签""" training_data = [[1, 2], [3, 4], [5, 6]] labels = [0, 1, 0] return {'features': training_data, 'labels': labels} @step def train_model(data: dict) -> None: """模拟训练过程，展示如何使用输入数据""" total_features = sum(map(sum, data['features'])) total_labels = sum(data['labels']) print(f"Trained model using {len(data['features'])} data points. " f"Feature sum is {total_features}, label sum is {total_labels}") @pipeline def simple_ml_pipeline(): """定义流水线，把步骤连接起来""" dataset = load_data() train_model(dataset) if __name__ == "__main__": run = simple_ml_pipeline() # 可用 run 对象查看 steps、outputs 等

把上述代码保存为run.py并执行python run.py，终端会输出：

Initiating a new run for the pipeline: simple_ml_pipeline. Using stack: default orchestrator: default artifact_store: default Step load_data has started. Step load_data has finished in 0.385s. Step train_model has started. Trained model using 3 data points. Feature sum is 21, label sum is 1 Step train_model has finished in 0.265s. Run simple_ml_pipeline-2023_11_23-10_51_59_657489 has finished in 1.612s.

• @step：把函数转换为可在 Pipeline 中使用的 Step；
• @pipeline：把函数定义为 Pipeline，函数内部调用 Step 并通过返回值串联数据流。

最佳实践：务必把流水线执行嵌套在if __name__ == "__main__"中，避免在其他地方导入该模块时误触发执行。

2.3 在 Dashboard 中查看运行结果

运行zenml login --local会打开浏览器，Dashboard 默认地址为http://127.0.0.1:8237/，默认用户名default（无需密码）。你可以看到：

• DAG 视图：理解 Pipeline 结构与步骤依赖；
• Timeline 视图：分析执行性能，适合步骤较多的流水线做性能优化。

如果关闭了浏览器标签页，随时可以用zenml show重新打开。

2.4 多输出 Step 与参数化训练

真实 ML 流水线通常需要「一个 Step 输出多个数据」以及「可配置超参数」。使用Tuple类型注解定义多输出，用Annotated为输出命名（参见 manage-artifacts.md 中的自定义输出名）：

from typing import Annotated, Tuple 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 @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 数据集为多个命名输出""" iris = load_iris(as_frame=True) X_train, X_test, y_train, y_test = train_test_split( iris.data, iris.target, test_size=0.2, shuffle=True, random_state=42 ) 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 分类器""" model = SVC(gamma=gamma) model.fit(X_train.to_numpy(), y_train.to_numpy()) train_acc = model.score(X_train.to_numpy(), y_train.to_numpy()) print(f"Train 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(gamma=gamma, X_train=X_train, y_train=y_train) if __name__ == "__main__": training_pipeline(gamma=0.0015)

要点：

• 前置依赖：pip install matplotlib和zenml integration install sklearn -y。集成安装本质是执行pip install sklearn；若出问题可用zenml integration requirements sklearn查看兼容的版本清单后手动安装。
• 日志建议使用 Python 标准logging模块，ZenML 会把 root logger 的输出写入 artifact store，最终显示在 Dashboard 中。
• 单独执行一个 Step 也可以：在 Pipeline 外直接调用svc_trainer(X_train=..., y_train=...)即可。

2.5 用 YAML 文件配置 Pipeline

生产环境中希望「不改代码只改配置」，ZenML 支持从 YAML 文件加载参数：

training_pipeline = training_pipeline.with_options( config_path='/local/path/to/config.yaml' ) training_pipeline()

配置文件示例：

parameters: gamma: 0.01

注意：YAML 中的参数优先级高于代码中传入的参数。如果不确定格式，可以用training_pipeline.write_run_configuration_template(path=...)生成模板。

三、Step 缓存：加速迭代的核心机制

ML 流水线开发是高度迭代的，ZenML 通过步骤缓存让重复执行跳过未变化的 Step。在 docs/book/user-guide/starter-guide/cache-previous-executions.md 中有完整的机制说明。

3.1 缓存的工作原理

再次运行同一 Pipeline 时，你会看到日志：

Step training_data_loader has started. Using cached version of training_data_loader. Step svc_trainer has started. Train accuracy: 0.3416666666666667

ZenML 自动跟踪并版本化所有输入、输出与参数。只要 Step 的输入、参数或代码没有变化，就不会在后续同一 Pipeline 运行中重新执行，而是复用上次存储在 artifact store 中的输出。

客户端缓存优化：不带 schedule 运行时，ZenML 可在客户端机器上直接计算缓存步骤，远程 orchestrator 无需重复执行，节省时间与成本；若希望 orchestrator 总是动态计算缓存步骤，可设置环境变量ZENML_PREVENT_CLIENT_SIDE_CACHING=True。

重要警告：缓存不会自动检测文件系统或外部 API 的变化。依赖外部输入、文件系统变化的 Step 应手动关闭缓存：

@step(enable_cache=False) def load_data_from_external_system(...) -> ...: # 该步骤每次都执行

3.2 缓存控制的三层优先级

Pipeline Settings --> Step Settings --> 代码/输入/参数变化

• Pipeline 级：@pipeline(enable_cache=False)关闭整条流水线缓存（Step 可显式enable_cache=True覆盖）；
• 运行时动态配置：first_pipeline.with_options(enable_cache=False)覆盖所有装饰器设置；
• Step 级：@step(enable_cache=False)或import_data_from_api.with_options(enable_cache=False)只影响单个步骤。

3.3 用 CachePolicy 精细调优缓存

CachePolicy类（位于src/zenml/config，可通过from zenml.config import CachePolicy导入）可以精确控制缓存键的生成因子：

from zenml.config import CachePolicy custom_cache_policy = CachePolicy(include_step_code=False) @step(cache_policy=custom_cache_policy) def my_step(): ... # 或 my_step = my_step.with_options(cache_policy=custom_cache_policy)

缓存键的组成因子：

source_dependencies与cache_func既可以直接传对象，也可以传 source 字符串（如"run.my_helper_function"），后者在 YAML 配置文件中同样适用。

缓存过期：默认任何执行成功的 Step 都是后续运行的缓存候选，可通过CachePolicy(expires_after=60*60*24)设置 24 小时过期；也可以手动过期某个 step run：

from zenml import Client from datetime import datetime, timezone now = datetime.now(timezone.utc) Client().update_step_run(<STEP-RUN-ID>, cache_expires_at=now)

四、Production Guide：从本地到云端的生产化

如果说 Starter Guide 解决「跑起来」，Production Guide 解决「跑得稳、跑得远」。它在 docs/book/user-guide/production-guide/README.md 中系统讲解了 8 个生产化主题，建议读者先准备 Python 环境与virtualenv，并选择 AWS / GCP / Azure 之一安装好对应的 CLI。

4.1 生产化主题一览

Stack 是生产化的核心概念：它把编排器（orchestrator）、artifact store、容器 registry、step operator 等组件组合为一个可复用的部署环境。同一个 Pipeline 可以在本地 Stack 上开发，再切换到云端 Stack 生产运行——这正是 src/zenml/stack 中 Stack 抽象的设计初衷。云端部署相关配置还可以参考 helm/values.yaml（Helm Chart 方式部署 ZenML Server）以及 docker 目录下的各类 Dockerfile。

五、LLMOps Guide：RAG、评估与模型微调

LLMOps Guide（docs/book/user-guide/llmops-guide/README.md）面向想用 ZenML 构建 LLM 应用的工程人员，整个指南贯穿一个具体场景：为 ZenML 构建一个能回答 ZenML 常见问题的问答系统，从简单 RAG 起步，逐步演进到微调 embedding、引入 reranking、甚至微调 LLM 本身。

5.1 五大 LLMOps 主题

每个主题都有「极简实现 → 原理讲解 → 生产实践」的递进结构，例如：

• RAG in 85 lines：用 85 行代码跑通完整 RAG（rag-85-loc.md）；
• Evaluation in 65 LOC：用 65 行代码完成评估（evaluation-in-65-loc.md）；
• Finetuning in 100 LOC：100 行代码完成 LLM 微调（finetuning-100-loc.md）。

5.2 LLMOps 与仓库示例的对应关系

指南中的概念在仓库examples/目录中有大量可直接运行的实现可以对照学习：

这些示例与 LLMOps Guide 的章节一一呼应，读者可以在读完某个章节后直接进入对应示例目录运行验证。

六、Tutorials 与 Best Practices：进阶实践与团队协作

6.1 Tutorials 深度专题

docs/book/user-guide/tutorial 目录下的专题文档解决具体工程难题：

6.2 Best Practices 团队协作规范

docs/book/user-guide/best-practices 提供了 12 篇团队级最佳实践，包括：

• 项目与团队：set-up-your-repository.md（项目仓库规范）、shared-components-for-teams.md（团队共享组件）、access-management.md（访问管理）；
• 环境与工程：configure-python-environments.md（Python 环境）、debug-and-solve-issues.md（问题排查）、quick-wins.md（5 分钟快速见效）；
• 基础设施：iac.md（Terraform 基础设施即代码）、project-templates.md（ML 平台模板）、choose-orchestration-environment.md（编排器选型）、mcp-chat-with-server.md（通过 MCP 与 Server 对话）、vscode-extension.md（VS Code 扩展）。

七、Projects 与 Examples：端到端实战参考

7.1 仓库内 Examples 速查

用户指南将 Examples 定义为「针对特定 ML 工作流挑战的聚焦代码片段与模板」。仓库examples/目录提供了可直接运行的参考实现，以下按用途分类：

以最简的 examples/quickstart 为例，它的入口run.py只有几行：

from pipelines.simple_pipeline import simple_pipeline def main() -> None: print("🚀 Running ZenML quickstart pipeline...") _ = simple_pipeline() if __name__ == "__main__": main()

而 examples/quickstart/pipelines/simple_pipeline.py 展示了@pipeline装饰器的高级用法：通过settings={"deployment": DeploymentSettings(...)}配置部署模式，同时支持python run.py批量执行与zenml pipeline deploy部署模式两种运行方式——这正是用户指南中「Pipeline 可配置」概念的真实落地。

7.2 官方 Projects 生态

用户指南还列出了一些基于 ZenML 构建的完整项目（如 FloraCast 时序预测、Retail Forecast 零售预测、OncoClear 乳腺癌分类、ZenML Support Agent 等），它们代表了 ZenML 在真实业务场景中的最佳实践模板，可作为自己项目的起点。需要说明的是，这些项目大多托管在外部站点，仓库内可验证的实现以examples/目录为准。

八、学习路线建议

结合以上内容，这里给出一个可执行的 ZenML 学习路线：

1. 第 1 周 · 入门：安装 ZenML（pip install "zenml[server]"+zenml login --local），阅读 create-an-ml-pipeline.md，跑通最小 Pipeline 并运行 examples/quickstart；
2. 第 2 周 · 核心机制：掌握缓存（cache-previous-executions.md）、Artifact 管理（manage-artifacts.md）与模型跟踪（track-ml-models.md）；
3. 第 3-4 周 · 生产化：按 production-guide 顺序部署 ZenML Server、配置云端 Stack 与远程存储，跑通 examples/e2e；
4. 第 5 周起 · LLMOps：从 rag-85-loc.md 起步，逐层深入评估、reranking、embedding 微调与 LLM 微调，对照 examples/llm_finetuning 与 examples/agent_framework_integrations 实践；
5. 长期 · 团队落地：参考 best-practices 建立团队规范，用 iac.md 实现基础设施代码化。

总结

ZenML 用户指南的价值在于它不只是一份 API 参考，而是一条可执行的成长路线：从@step/@pipeline两个装饰器开始，到缓存、Stack、云端编排等生产化机制，再到 RAG、评估、微调等 LLMOps 专题，最终落到仓库内examples/与最佳实践文档构成的实战参考。无论你的目标是「构建生产级 ML Pipeline」还是「构建生产级 AI Agent」，都可以从 docs/book/user-guide/toc.md 这份目录出发，按图索骥、逐层深入。

【免费下载链接】zenmlZenML 🙏: One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml