通知器

通知器组件提供可插拔的通知系统，用于在训练过程中发送告警。当异常发生或训练事件需要关注时，通知器将消息投递到外部渠道（如钉钉 Webhook）。

基础接口

from twinkle.notifier import Notifier

class Notifier:
    def __call__(self, message: str):
        """发送通知消息"""
        ...

    def to_dict(self) -> dict:
        """序列化（用于 checkpoint 保存/恢复）"""
        ...

    @classmethod
    def from_dict(cls, data: dict) -> Notifier:
        """从序列化数据恢复"""
        ...

DingNotifier（钉钉通知）

向钉钉自定义机器人 Webhook 发送通知。

from twinkle.notifier import DingNotifier

notifier = DingNotifier(
    ding_url='https://oapi.dingtalk.com/robot/send?access_token=xxx',
    secret='SECxxxxxxx',  # 可选：签名模式
    timeout=5.0,
)

# 发送消息
notifier("### 训练完成\n\n- Steps: 1000\n- Loss: 0.25")

参数：

ding_url：完整的钉钉 Webhook URL（含 access_token）
secret：可选签名密钥（签名模式机器人）
timeout：HTTP 请求超时时间，单位秒（默认 5.0）

消息以钉钉 Markdown 格式发送。第一个标题行自动提取为聊天预览标题。

异常通知

Twinkle 提供带去重的自动异常通知：

from twinkle.notifier.base import notify_exception

# 自动发送格式化的异常信息
# 每个唯一异常只有一个 rank 发送（防止消息洪泛）
try:
    model.forward_backward(batch)
except Exception as e:
    notify_exception(notifier, context='forward_backward', exc=e, name='sft_train')

通知包含：

异常类型和消息
完整堆栈跟踪
运行时元数据（rank、PID、主机名）
去重：所有 rank 中每个唯一异常只发一条通知

自定义通知器

继承 Notifier 创建自定义通知器：

from twinkle.notifier import Notifier

class SlackNotifier(Notifier):
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    def __call__(self, message: str):
        import requests
        requests.post(self.webhook_url, json={'text': message})

    def to_dict(self):
        return {'class': 'SlackNotifier', 'webhook_url': self.webhook_url}

    @classmethod
    def _from_dict_impl(cls, data):
        return cls(webhook_url=data['webhook_url'])

通知器通过 __init_subclass__ 自动注册，因此 Notifier.from_dict() 可以按类名恢复任何子类。