ZenML Alerter 组件指南在管道中对接 Slack/Discord 实现自动化告警与人工审批【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenmlZenML 的Alerter告警器组件让你可以直接在管道pipeline内部向 Slack、Discord 等聊天服务发送消息既能实现失败时的即时通知与监控上报也能通过ask()步骤构建人在回路human-in-the-loop的人工审批流程。读完本文你将掌握 Alerter 的核心概念、Slack/Discord 两种内置 flavor 的完整注册与使用流程、post()/ask()步骤的底层工作原理以及如何基于BaseAlerter抽象为任意聊天服务开发自定义 alerter。什么是 Alerter在 组件指南Alerters 中Alerter 被定义为允许你从管道内部向聊天服务如 Slack、Discord、Mattermost 等发送消息的 ZenML 栈组件。它的典型价值体现在三个场景失败即时通知管道中某一步骤抛出异常、模型性能退化或数据漂移发生时立刻在团队频道收到消息监控与上报将每次运行的进度、指标、结论自动汇总到指定频道形成流水线运行的可视化日志构建 Human-in-the-loop ML在模型部署、关键决策等节点插入人工确认环节由人类在聊天工具中一键批准或拒绝。Alerter 是 ZenML 标准栈组件类型之一StackComponentType.ALERTER其抽象基类定义在 src/zenml/alerter/base_alerter.py核心只有两个方法post(message, params)把字符串消息发送到聊天服务成功返回True失败返回Falseask(question, params)先发送问题消息然后等待用户回复批准/拒绝关键词只有成功发送且被批准才返回True否则返回False。class BaseAlerter(StackComponent, ABC): def post(self, message: str, paramsNone) - bool: return True def ask(self, question: str, paramsNone) - bool: return True围绕这两个方法ZenML 还配套定义了BaseAlerterConfig组件配置基类、BaseAlerterStepParameters步骤参数基类与BaseAlerterFlavorflavor 基类其type固定为ALERTER构成了完整的三件套Flavor 负责注册与描述Config 负责校验参数Implementation 负责真正收发消息。可用的 Alerter Flavors目前 ZenML 官方提供两个内置 alerter 集成外加一个自定义扩展入口AlerterFlavorIntegration说明Slackslackslack与 Slack 频道交互Discorddiscorddiscord与 Discord 频道交互自定义实现custom—扩展 alerter 抽象提供自己的实现在终端中查看当前环境可用的全部 alerter flavorzenml alerter flavor list运行后你应能看到slack、discord两个内置 flavor若注册过自定义 flavor如my_alerter它也会一并列出。快速上手注册 Alerter 并接入 Stack注册 Alerter 组件每个 alerter 集成都带有开箱即用的标准步骤但前提是先把 alerter 作为栈组件注册到 ZenMLzenml alerter register ALERTER_NAME ...以 Slack 为例注册命令通常携带--flavor、token 与频道 ID 等参数完整参数见下文 Slack/Discord 小节。把 Alerter 加入 Stack注册完成后通过-al即--alerter把它附加到栈上zenml stack register ... -al ALERTER_NAME一个栈可以挂多个 alerterZenML 将第一个附加的 alerter 视为默认 alerter单数形式的访问器Client().active_stack.alerter解析到的就是那个默认实例。这意味着你可以同时注册 Slack 与 Discord并在不同步骤中分别指定使用哪一个。在管道中使用注册并挂载到栈之后直接导入对应集成提供的标准步骤即可在管道里使用from zenml.integrations.slack.steps.slack_alerter_post_step import slack_alerter_post_step from zenml.integrations.slack.steps.slack_alerter_ask_step import slack_alerter_ask_step内置集成一Slack Alerter完整说明见 Slack Alerter 文档。准备一个 Slack App使用SlackAlerter前你需要在 Slack 工作区中准备一个目标频道然后创建一个带 bot 的 Slack App。在OAuth Permissions选项卡的Scopes下为 bot 授予以下权限chat:write发送消息channels:read读取频道信息channels:history读取频道历史——ask()轮询用户回复依赖它要使用ask()功能还必须把 App 邀请进目标频道可以直接在频道里输入/invite或通过频道设置添加。注意 App 在创建工作区后可能需要一段时间才会出现在可邀请列表中。安装集成并注册组件先安装 ZenML 的slack集成zenml integration install slack -y然后创建 secret 并注册 alerterzenml secret create slack_token --oauth_tokenSLACK_TOKEN zenml alerter register slack_alerter \ --flavorslack \ --slack_token{{slack_token.oauth_token}} \ --slack_channel_idSLACK_CHANNEL_ID参数说明SLACK_CHANNEL_ID在频道详情中查看以C....开头SLACK_TOKENSlack App 设置中OAuth Permissions下 bot 的 token。推荐用{{secret_name.key}}语法在组件属性中引用 ZenML secret而不是把 token 明文写在命令里——这是官方文档明确推荐的密钥管理最佳实践从源码可见 slack_alerter_flavor.py 中slack_token字段被标记为SecretField()注册后 token 会自动以密文形式安全存储。最后把 alerter 加入栈zenml stack register ... -al slack_alerter --setSlack 的可配置项SlackAlerterConfig继承自SlackAlerterSettings支持以下配置配置项类型默认值说明slack_tokenstrSecretField必填Slack bot 的 tokenslack_channel_idOptional[str]None默认通信频道 IDtimeoutint300ask()等待用户回复的超时秒数默认 5 分钟此外SlackAlerterParameters提供了步骤级运行时参数slack_channel_id临时切换频道、approve_msg_options、disapprove_msg_options、payload管道/步骤/栈元信息、include_format_blocks默认True自动附带运行上下文、blocks自定义 Slack Block Kit 消息块。组件注册后ZenML 会通过auth_test()与conversations_info()校验 token 与频道是否有效见 slack_alerter_flavor.py。内置集成二Discord Alerter完整说明见 Discord Alerter 文档。Discord 集成提供两个标准步骤discord_alerter_post_step接收字符串消息发布到 Discord 频道返回是否成功discord_alerter_ask_step发布消息后等待用户反馈仅当用户在 Discord 中明确批准例如回复approve才返回True。典型用途post_step用于失败/指标即时告警ask_step用于部署新模型前的人工确认。准备 Discord Bot安装discord集成zenml integration install discord -y创建一个带 bot 的 Discord App 并邀请进服务器频道。需要注意如果复制 token 时找不到复制按钮可先点击 reset token 重置以获取新 token邀请 bot 时至少授予以下权限Read Messages / View Channels读取消息/查看频道Send Messages发送消息Send Messages in Threads在线程中发送消息频道 ID 的获取方式右键点击文本频道 →Copy Channel ID若没有该选项到User Settings Advanced打开Developer Mode开发者模式。注册 Discord Alerterzenml alerter register discord_alerter \ --flavordiscord \ --discord_tokenDISCORD_TOKEN \ --default_discord_channel_idDISCORD_CHANNEL_ID同样推荐用 secret 托管 tokenzenml secret create discord_secret --discord_tokenDISCORD_TOKEN zenml alerter register discord_alerter \ --flavordiscord \ --discord_token{{discord_secret.discord_token}} \ --default_discord_channel_idDISCORD_CHANNEL_ID加入栈zenml stack register ... -al discord_alerterDiscordAlerterConfig的字段为discord_tokenSecretField与default_discord_channel_id默认频道 ID见 discord_alerter_flavor.py。Discord 的使用模式由于 post/ask 步骤都期望一个字符串消息作为输入且该输入通常来自另一个步骤的输出实践中一般要定义一个 formatter 步骤把要传达的数据格式化成消息from zenml import step, pipeline from zenml.integrations.discord.steps.discord_alerter_ask_step import discord_alerter_ask_step step def my_formatter_step(artifact_to_be_communicated) - str: return fHere is my artifact {artifact_to_be_communicated}! step def process_approval_response(artifact, approved: bool) - None: if approved: print(fUser approved! Processing {artifact}) else: print(User disapproved. Skipping operation.) pipeline def my_pipeline(...): artifact_to_be_communicated ... message my_formatter_step(artifact_to_be_communicated) approved discord_alerter_ask_step(message) process_approval_response(artifact_to_be_communicated, approved)Ask 步骤构建 Human-in-the-Loop 工作流所有 alerter 都提供ask()方法及对应的 ask 步骤它们是人工审批流的关键构件常用于将模型部署到生产环境前获取批准确认关键管道决策在自动化流程中插入人工干预点。Ask 步骤的工作方式ask步骤如discord_alerter_ask_step、slack_alerter_ask_step遵循三步流程向聊天服务发布你的问题消息等待用户回复——回复内容匹配特定的批准/拒绝关键词返回布尔值——批准返回True拒绝或超时返回False。一个完整的部署审批管道示例如下from zenml import step, pipeline from zenml.integrations.slack.steps.slack_alerter_ask_step import slack_alerter_ask_step step def train_model(): # Training logic here - this is a placeholder function return trained_model_object step def deploy_model(model, approved: bool) - None: if approved: # Deploy the model to production print(Deploying model to production...) else: print(Deployment cancelled by user) pipeline def deployment_pipeline(): trained_model train_model() # Ask for human approval before deployment approved slack_alerter_ask_step(Deploy model to production?) deploy_model(trained_model, approved)默认响应关键词默认情况下alerter 识别以下回复关键词批准approve、LGTM、ok、yes拒绝decline、disapprove、no、reject自定义响应关键词通过 alerter 参数类可以完全自定义批准/拒绝关键词例如使用团队内部的部署黑话或 emojifrom zenml.integrations.slack.steps.slack_alerter_ask_step import slack_alerter_ask_step from zenml.integrations.slack.alerters.slack_alerter import SlackAlerterParameters # Use custom approval/disapproval keywords params SlackAlerterParameters( approve_msg_options[deploy, ship it, ✅], disapprove_msg_options[stop, cancel, ❌] ) approved slack_alerter_ask_step( Deploy model to production?, paramsparams )Discord 侧使用等价的DiscordAlerterParameters见 discord.md参数名与用法一致。从源码看只要params.approve_msg_options/disapprove_msg_options不为None就会覆盖DEFAULT_APPROVE_MSG_OPTIONS/DEFAULT_DISAPPROVE_MSG_OPTIONS定义于 slack_alerter.py 与 discord_alerter.py。关键注意事项返回类型ask 步骤返回布尔值管道逻辑必须正确处理这个结果关键词大小写默认关键词大小写敏感Discord 要求严格区分大小写回复LGTM不能写成lgtmSlack 除外——Slack 实现会把回复与关键词统一转为小写后再匹配因此LGTM、lgtm、Lgtm均可生效超时超时时间内未收到有效回复时步骤返回FalseSlack 默认超时 300 秒可通过配置修改权限确保 bot 拥有在目标频道读取消息的权限Slack 需要channels:historyDiscord 需要 Read Messages。源码级原理剖析步骤如何找到 alerter以 slack_alerter_ask_step.py 为例步骤内部通过Client().active_stack.alerter取当前活跃栈的默认 alerter并做类型检查若它不是SlackAlerter立即抛出RuntimeError。同时当params.include_format_blocksTrue默认时步骤会自动通过get_step_context()与活跃栈名称构造SlackAlerterPayload含pipeline_name、step_name、stack_name把运行上下文注入消息step def slack_alerter_ask_step(message, paramsNone) - bool: alerter Client().active_stack.alerter if not isinstance(alerter, SlackAlerter): raise RuntimeError(... requires an alerter component of flavor slack ...) if params and params.include_format_blocks: params.payload SlackAlerterPayload( pipeline_namecontext.pipeline.name, step_namecontext.step_run.name, stack_nameactive_stack.name, ) return alerter.ask(message, params)post_step的实现同构只是调用alerter.post(...)。Slack 的轮询式 ask 实现SlackAlerter.ask() 的实现要点用chat_postMessage发送问题记录消息时间戳ts进入轮询循环以conversations_history(channel, oldesttimestamp)拉取该时间戳之后的新消息逐条把文本strip().lower()后与同样转为小写的批准/拒绝关键词比对每轮轮询间隔time.sleep(1)直到超过timeout默认 300 秒命中批准词返回True命中拒绝词或超时返回False任何SlackApiError或异常都会被捕获并记录日志后返回False——这正是文档中出错时ask()默认返回False的底层原因。消息渲染方面_create_blocks支持自定义blocks未提供自定义块但提供了payload时会自动生成包含 Pipeline/Step/Stack/Message 字段的 Block Kit 富文本消息。Discord 的 wait_for 式 ask 实现DiscordAlerter.ask() 采用不同机制通过discord.Client建立连接on_ready回调中向频道发送消息再用client.wait_for(message, checkcheck)阻塞等待用户回复check函数对消息内容做**精确区分大小写**匹配——这也是文档强调 Discord 关键词大小写敏感的原因。发送失败、未找到频道、异常或未收到匹配消息时approved保持False。Channel ID 的解析优先级两个实现的_get_channel_id()都遵循步骤参数优先于组件配置的解析顺序先看params中的频道 ID再看运行时 settings 中的频道 IDSlack 支持通过step(settings{alerter: {slack_channel_id: ...}})在步骤级覆盖最后回落到组件注册时的slack_channel_id/default_discord_channel_id全部未设置则抛出ValueError。开发自定义 Alerter详细步骤见 Develop a Custom Alerter。由于基类抽象极其精简为 Slack、Mattermost、Telegram 等任何聊天服务扩展 alerter 只需四步第 1 步实现BaseAlerter子类完成post()与ask()。实现ask()时应遵循等待包含批准关键词的回复、等待包含拒绝关键词的回复、只有收到明确批准才返回True、拒绝/超时/任何错误都返回False并尽量通过参数支持关键词可配置import logging from typing import Optional from zenml.alerter import BaseAlerter, BaseAlerterStepParameters class MyAlerter(BaseAlerter): My alerter class. def post(self, message: str, params: Optional[BaseAlerterStepParameters]) - bool: try: # Implement your chat service posting logic here logging.info(fPosting message: {message}) return True except Exception as e: logging.error(fFailed to post message: {e}) return False def ask(self, question: str, params: Optional[BaseAlerterStepParameters]) - bool: try: if not self.post(question, params): return False approve_options [approve, yes, ok, LGTM] disapprove_options [reject, no, cancel, stop] if params and hasattr(params, approve_msg_options): approve_options params.approve_msg_options if params and hasattr(params, disapprove_msg_options): disapprove_options params.disapprove_msg_options response self._wait_for_user_response() if response.lower() in [opt.lower() for opt in approve_options]: return True return False except Exception as e: print(fFailed to get approval: {e}) return False第 2 步定义配置对象可选from zenml.alerter.base_alerter import BaseAlerterConfig class MyAlerterConfig(BaseAlerterConfig): my_param: str第 3 步定义步骤参数类可选用于支持可配置关键词等from typing import List, Optional from zenml.alerter.base_alerter import BaseAlerterStepParameters class MyAlerterParameters(BaseAlerterStepParameters): approve_msg_options: Optional[List[str]] None disapprove_msg_options: Optional[List[str]] None custom_channel: Optional[str] None第 4 步编写 Flavor 类把实现与配置绑定from typing import Type, TYPE_CHECKING from zenml.alerter import BaseAlerterFlavor if TYPE_CHECKING: from zenml.stack import StackComponent, StackComponentConfig class MyAlerterFlavor(BaseAlerterFlavor): property def name(self) - str: return my_alerter property def config_class(self) - Type[StackComponentConfig]: from my_alerter_config import MyAlerterConfig return MyAlerterConfig property def implementation_class(self) - Type[StackComponent]: from my_alerter import MyAlerter return MyAlerter然后通过 CLI 注册 flavor必须使用点号路径指向 flavor 类zenml alerter flavor register flavors.my_flavor.MyAlerterFlavor注册后即可用zenml alerter flavor list确认新 flavor 已出现。理解这套抽象在各阶段扮演的角色很重要MyAlerterFlavor在 CLI 注册 flavor 时被导入MyAlerterConfig在注册/更新my_alerter组件时被导入并用于校验参数Config 是 pydantic 对象可自定义 validatorsMyAlerter只在组件真正运行时才参与。这种配置与实现分离的设计让你在本地未安装实现依赖的情况下也能注册 flavor 与组件。实践建议密钥一律走 ZenML Secret用{{secret_name.key}}引用 token避免敏感信息进入命令历史或配置文件频道 ID 优先在组件/步骤配置中固定运行时如确需切换频道Slack 可用step(settings{alerter: {slack_channel_id: ...}})覆盖但使用ask()前务必确认 App/bot 已受邀进入该频道把审批结果接入下游分支approved布尔值应像普通步骤输出一样参与管道编排未批准时优雅跳过部署等关键动作理解超时与失败语义ask 步骤超时或出错一律返回False因此审批流天然默认拒绝安全可靠团队频道保持关键词统一如果团队使用自定义批准词如 emoji通过SlackAlerterParameters/DiscordAlerterParameters显式配置并注意 Slack 大小写不敏感、Discord 大小写敏感的实现差异。通过 Alerter 组件ZenML 管道与团队日常沟通工具之间形成了一条双向通道机器自动上报运行状态人类在熟悉的地方完成关键决策——这正是 MLOps 中可观测性与可控性的完美结合。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考