企业官网建设流程全解析

ZenML Secrets 完整实战指南：集中式密钥管理与 LLM/Agent 凭证安全实践

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

ZenML 的 Secrets（密钥）机制是 AI 平台安全体系中连接管道、Stack 组件与外部服务的核心环节：它以「名称 + 键值对」的形式把数据库口令、模型注册中心访问令牌以及 LLM API Key、第三方 Agent 工具凭证统一存放到安全的密钥存储中，并通过{{<SECRET_NAME>.<SECRET_KEY>}}引用语法与 Python SDK 让管道和 Agent 无需硬编码即可安全取用。读完本文，你将掌握使用 CLI 与 Python SDK 创建、查看、更新、删除密钥的完整操作，理解公开/私有密钥的访问控制与同名查找优先级，并能把密钥安全地注入 Stack 组件属性和 Step 中，为传统 ML 工作流与 AI Agent 应用建立可落地的凭证管理方案。

什么是 ZenML Secret

ZenML 官方文档对 Secret 的定义非常明确：Secret 是一组「键值对」（key-value pairs）的集合，安全地存储在 ZenML 的 secrets store（密钥存储）中，并且每个 Secret 都有一个名称（name），用于在管道（pipeline）和 Stack 中获取或引用它。

# 一个 Secret 的逻辑结构 { "name": "openai_secret", "values": { "api_key": "sk-proj-...", "organization_id": "org-..." } }

Secret 同时服务两类典型场景：

• 传统 ML 工作流：数据库凭据、模型注册中心（如 MLflow）访问认证等敏感信息；
• AI Agent 开发：LLM API Key、第三方服务（搜索 API、天气 API 等）凭证。

从源码实现看，Client 层通过create_secret方法把名称、键值对和可见性封装为SecretRequest模型提交给zen_store（见 client.py），存储层则由部署的 secrets store 后端（如本地 SQLite、云端托管服务等）负责落盘与加密。需要特别说明的是：集中式密钥管理依赖 ZenML Server 的 secrets store 能力，如果目标 ZenML 部署不支持或显式禁用了该功能，调用会抛出NotImplementedError（"centralized secrets management is not supported or explicitly disabled..."），这是使用本机制的前提条件。

如何创建 Secret

创建 Secret 有 CLI 与 Python SDK 两条路径，两者能力对齐，可任意混用。

方式一：CLI 创建

zenml secret create是创建密钥的入口命令，支持三种传参方式：

1. 直接以--<KEY>=<VALUE>传参

zenml secret create <SECRET_NAME> \ --<KEY_1>=<VALUE_1> \ --<KEY_2>=<VALUE_2>

2. 使用--values传入 JSON/YAML 键值对

zenml secret create <SECRET_NAME> \ --values='{"key1":"value2","key2":"value2"}'

3. 实战示例：为 LLM 与多 Agent 系统创建凭证

# 为 OpenAI 创建 API 密钥 Secret zenml secret create openai_secret \ --api_key=sk-proj-... \ --organization_id=org-... # 为 Anthropic 创建 API 密钥 Secret zenml secret create anthropic_secret \ --api_key=sk-ant-api03-... # 为多 Agent 系统创建一组工具凭证 zenml secret create agent_tools_secret \ --google_search_api_key=AIza... \ --weather_api_key=abc123 \ --database_url=postgresql://user:pass@host/db # 创建私有 Secret（仅创建者本人可访问） zenml secret create my_private_secret --private \ --api_key=secret-value

4. 交互式创建

值比较多或不想留在 shell 历史中时，可用--interactive/-i参数进入交互会话，ZenML 会逐个询问密钥名与值：

zenml secret create <SECRET_NAME> -i

5. 从文件读取大值或含特殊字符的值

对于过大或含特殊字符、不便作为命令行参数直接传递的值，可以使用@语法让 ZenML 从文件读取：

zenml secret create <SECRET_NAME> \ --key=@path/to/file.txt \ ... # 或者把 JSON/YAML 键值对写入文件，用 --values=@path 传入 zenml secret create <SECRET_NAME> \ --values=@path/to/file.txt

从 CLI 源码看（cli/secret.py），create命令还隐含了若干重要细节：

• --private/-p是is_flag=True的布尔开关，用于创建私有 Secret；
• 交互模式下不允许再同时传入--key=value参数，二者互斥；
• 密钥名不能是"name"，键也不能命名为"name"，否则直接报错；
• 创建前会对每个键调用validate_keys校验，并调用pretty_print_secret(..., hide_secret=True)隐藏值地打印将要注册的 Secret；
• 如果同名 Secret 已存在，会抛出EntityExistsError并提示创建失败。

方式二：Python SDK 创建

ZenML Client 提供了程序化创建接口，适合在初始化脚本、Notebook 或自动化流程中使用：

from zenml.client import Client client = Client() # 基础用法 client.create_secret( name="my_secret", values={ "username": "admin", "password": "abc123" } ) # 示例：程序化创建 LLM API 密钥 client.create_secret( name="openai_secret", values={ "api_key": "sk-proj-...", "organization_id": "org-..." } ) # 创建私有 Secret（仅创建者本人可访问） client.create_secret( name="my_private_secret", values={"api_key": "secret-value"}, private=True, )

对应 SDK 签名（client.py）：

def create_secret( self, name: str, values: Dict[str, str], private: bool = False, ) -> SecretResponse:

private默认为False（公开）；置为True即创建仅创建者可访问的私有 Secret。

其他管理操作

CLI 与 Client 还提供完整的生命周期管理命令：

其中list_secrets支持丰富的过滤与排序条件；get_secret可按名称、ID 或名称/ID 前缀获取（详见下文「同名 Secret 的获取顺序」）。

Size limits（大小限制）

一个 ZenML Secret 本质上就是「名称 + 一组键值对」，其大小通常按所有键和值的 UTF-8 总字节数计算。不同 secrets store 后端对该上限的约束不同。如果超出限制，官方给出的三种处理策略：

1. 新增一个命名 Secret，把键分散到多个 Secret 中；
2. 拆分键，将大值按逻辑切分到不同键/不同 Secret；
3. 大文件不要存入 ZenML，只把文件路径或引用存放在某个值里。

这种设计把「存储敏感的小凭证」与「存储大文件」的职责分离，让密钥库始终轻量、可审计。

私有与公开 Secret（Private and public secrets）

可见性规则

ZenML 的 Secret 分为两种可见性：

• 私有 Secret（private）：仅创建者本人可查看、使用和管理，无论其他用户的角色或权限如何，都无法访问；
• 公开 Secret（public，默认）：其他用户依据你的RBAC（基于角色的访问控制）配置决定能否访问。在 ZenML Pro 上，公开 Secret 的访问由角色权限设置管控。

关键提示：private属性的优先级高于 RBAC。即使 RBAC 本来允许某用户访问，只要 Secret 是私有的，它仍然只对创建者可见。

创建私有 Secret

默认创建的 Secret 是公开的（private=False）。要创建私有 Secret：

# CLI：使用 --private 或短参数 -p zenml secret create <SECRET_NAME> --private \ --<KEY_1>=<VALUE_1> \ --<KEY_2>=<VALUE_2> # 短参数形式 zenml secret create <SECRET_NAME> -p \ --<KEY_1>=<VALUE_1>

# Python SDK from zenml.client import Client client = Client() client.create_secret( name="my_private_secret", values={"api_key": "..."}, private=True, # 使该 Secret 变为私有 )

注意：目前设置私有状态仅支持 CLI 与 Python SDK 两种途径，Dashboard 界面尚不支持创建或修改私有 Secret。

同名 Secret 的获取顺序

由于私有与公开 Secret 存在于相互独立的命名空间，你可以同时拥有一个私有 Secret 和一个公开 Secret 且名称相同。当不带可见性条件按名称获取时：

1. ZenML先搜索私有 Secret；
2. 再搜索公开 Secret；
3. 返回第一个匹配项。

从源码（client.py）可以看到，未指定private时的搜索顺序被显式定义为search_private_statuses = [False, True]，即先查私有再查公开，与文档描述完全一致。

要显式获取指定可见性的同名 Secret：

# 显式获取私有 Secret zenml secret get my_secret --private=true # 显式获取公开 Secret zenml secret get my_secret --private=false

from zenml.client import Client client = Client() # 显式获取私有 Secret private_secret = client.get_secret("my_secret", private=True) # 显式获取公开 Secret public_secret = client.get_secret("my_secret", private=False)

更新 Secret 的可见性

创建后也可以随时调整可见性：

# 把公开 Secret 改为私有 zenml secret update my_secret --private=true # 把私有 Secret 改为公开 zenml secret update my_secret --private=false

from zenml.client import Client client = Client() client.update_secret("my_secret", update_private=True) # 改为私有

在 Stack 组件属性与设置中引用 Secret

Stack 中的部分组件（如 experiment tracker、orchestrator、model deployer 等）需要配置口令或令牌才能连接底层基础设施。**Secret 引用（secret references）**让你不必把敏感值直接写进配置，而是通过引用语法把值与 Secret 解耦。

引用语法为：

{{<SECRET_NAME>.<SECRET_KEY>}}

例如，为 MLflow experiment tracker 注册带认证信息的组件：

# 先创建名为 mlflow_secret 的 Secret，包含连接 MLflow tracking server # 所需的 username 和 password # 使用集中式密钥管理 zenml secret create mlflow_secret \ --username=admin \ --password=abc123 # 然后在 experiment tracker 组件中引用 username 和 password zenml experiment-tracker register mlflow \ --flavor=mlflow \ --tracking_username={{mlflow_secret.username}} \ --tracking_password={{mlflow_secret.password}} \ ...

从源码层面看，这个机制由 secret_utils.py 实现：_secret_reference_expression = re.compile(r"\{\{\s*\S+?\.\S+\s*\}\}")负责识别形如{{name.key}}的引用字符串，is_secret_reference判断某个属性值是否为 Secret 引用，parse_secret_reference则把引用解析为SecretReference(name, key)结构；stack_component.py 中的required_secrets属性会遍历组件配置，收集所有被引用的 Secret，作为后续校验的依据。

管道运行前的 Secret 预校验

当 Stack 中使用 Secret 引用时，ZenML 会在运行管道前验证 Stack 组件中引用的所有 Secret 和键都存在。这样做的目的是「快速失败」——避免管道运行一段时间后才因缺失 Secret 而失败，白白消耗算力与时间。

默认情况下，该校验需要读取每一个 Secret，以确认 Secret 本身和指定的键值对都真实存在。这带来两个副作用：一是校验可能耗时较长；二是如果运行环境没有读取 Secret 的权限，校验本身就会失败。

为此 ZenML 提供环境变量ZENML_SECRET_VALIDATION_LEVEL来控制校验强度（常量定义见 constants.py，枚举定义见 enums.py）：

实现位于 stack.py 的_validate_secrets：它读取环境变量（缺省为SECRET_AND_KEY_EXISTS），遍历required_secrets，逐一通过client.get_secret(secret_ref.name)拉取 Secret；在校验级别为SECRET_AND_KEY_EXISTS时还会访问secret.values[secret_ref.key]确认键存在。无法解析的引用会累积成错误信息，最终抛出StackValidationError（或在非致命模式下仅打 warning）。错误提示中还会建议你运行zenml stack register-secrets <STACK_NAME>补齐缺失的 Secret——这正是下面要讲的辅助命令。

交互式补齐 Stack 缺失的 Secret

如果你的 Stack 里存在带 Secret 引用的组件，运行管道前必须保证所有被引用的 Secret 都已注册。为此 ZenML 提供了：

zenml stack register-secrets [<STACK_NAME>]

该命令会解析 Stack 的required_secrets（实现在 cli/stack.py），对每个缺失或需要更新的 Secret 以交互方式询问键值；若传入--skip_existing，已存在值的键会被跳过，避免重复输入。不传<STACK_NAME>时默认处理当前激活的 Stack。

在 Step 中获取 Secret 值

使用集中式密钥管理时，可以直接在 Step 内部通过 ZenML Client API 访问 Secret，从而在调用外部 API 时不再把访问密钥硬编码进代码。这对 LLM Agent 类工作流尤其关键——API Key 只存于密钥库，Step 代码可以安全地提交到版本库。

from zenml import step from zenml.client import Client import openai @step def secret_loader() -> None: """Load the example secret from the server.""" # 从 ZenML 获取 Secret secret = Client().get_secret(<SECRET_NAME>) # secret.secret_values 是一个包含该 Secret 所有键值对的字典 authenticate_to_some_api( username=secret.secret_values["username"], password=secret.secret_values["password"], ) ... @step def run_llm_agent(prompt: str, query: str) -> str: """使用安全存储的 API 密钥执行 LLM Agent。""" # 从 ZenML Secrets 获取 LLM API 凭据 openai_secret = Client().get_secret("openai_secret") # 使用凭据初始化 OpenAI 客户端 from openai import OpenAI client = OpenAI( api_key=openai_secret.secret_values["api_key"], organization=openai_secret.secret_values["organization_id"] ) # 执行 Agent response = client.chat.completions.create( model="gpt-4", messages=[ {"role": "system", "content": prompt}, {"role": "user", "content": query} ] ) return response.choices[0].message.content

要点说明：

• Client().get_secret(...)返回的SecretResponse对象通过secret_values属性暴露键值字典，可直接按secret_values["key"]取值；
• 取回的密钥只存在于运行时的进程内存中，不会写入代码或日志，配合前面的可见性控制，可以做到「代码无密、凭据可审计」；
• 若部署环境未启用集中式密钥管理，get_secret同样会抛出NotImplementedError，需要先确认 ZenML Server 的 secrets store 已正确配置。

小结

ZenML Secrets 提供了从「创建 → 引用 → 校验 → 消费」的完整凭证管理闭环：

1. 创建：CLI（zenml secret create，支持--values、-i交互、@file文件读取）与 Python SDK（client.create_secret）双通道；
2. 可见性：默认公开受 RBAC 管控，--private创建仅创建者可访问的私有 Secret，私有优先级高于 RBAC，且二者命名空间独立、可按需显式获取；
3. 引用：Stack 组件属性用{{SECRET_NAME.SECRET_KEY}}语法解耦敏感配置，配合zenml stack register-secrets交互补齐；
4. 校验：ZENML_SECRET_VALIDATION_LEVEL（NONE/SECRET_EXISTS/SECRET_AND_KEY_EXISTS）精确控制管道运行前的预检强度，实现快速失败；
5. 消费：Step 内通过Client().get_secret(...)动态取用，支撑传统 ML 管道的数据库/模型仓库认证，也支撑 LLM 与多 Agent 系统的 API 密钥管理。

相关实现与文档入口：核心 SDK 实现见 client.py，CLI 命令见 cli/secret.py，Stack 预校验见 stack.py，引用解析工具见 utils/secret_utils.py，本文完整出处为 docs/book/how-to/secrets/secrets.md。

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