打工人的AI外挂：牛马的Deepseek Harness插件dsh-notify-push实践笔记
先说说憋了很久的话。作为一名日常被需求、评审、临时变更反复拉扯的搬砖打工仔，我一直在琢磨怎么用AI帮自己省点力气。试过各种AI工具，也搭过个人知识库，但有个痛点是始终绕不开的：AI任务跑完了，我却不知道结果。同步执行时盯着终端发呆，异步提交后完全失联，特别是耗时较长的分析、推理任务，经常是切出去刷个页面，回头就忘了这码事。直到我折腾出这套“牛马的Deepseek Harness插件：dsh-notify-push”，才真正把AI工作流里最关键的一环——结果通知——给补上了。说白了，这就是给AI任务配一个“跑完就喊你”的消息推送插件，专门解决打工人“不知道活干完了没有”的焦虑。
这篇博文不搞虚的，把我从需求拆解、方案选型、代码实现到部署排坑的完整过程都记下来。凡是遇到同样场景的朋友，不管是想给AI加通知能力，还是单纯想把任务跑完自动推消息到手机，都能直接抄作业。
1. 需求与方案设计：为什么要给AI加一个推送通知能力
1.1 真实痛点：AI任务跑完了，我却不知道
我日常的典型工作流是这样的：写一段复杂的业务逻辑分析，丢给AI去处理；或者让AI帮我在一批日志里定位异常模式；又或者让它基于一堆需求文档生成初步方案。这些任务短则几十秒，长则十几分钟。等待的时间自然不会浪费，我可能去回消息、改简历、刷新闻。但问题来了——AI这边其实早就跑完了，我这边却还在等，或者已经忘了。
说实话，这种“结果不同步”造成的效率损失，比任务本身耗时更可怕。以前的做法是开着终端、盯着进度条，生怕错过，整个人被绑在电脑前面。后来改用异步方式提交，但又是另一个极端：完全不知道什么时候能好，时不时切回去看一眼，来回切换的上下文开销反而更大。
1.2 候选方案对比：轮询、Webhook还是主动推送
有了这个痛点，我自然开始琢磨怎么解决。首先想到的是轮询方案，写个脚本定时查询任务状态，一旦完成就触发某个动作。这方案能跑，但有两个明显问题：第一，任务状态查询接口能不能支持高频轮询不好说；第二，轮询本身也会带来持续的计算开销，属于“为了省心反而更操心”。
第二个想法是用Webhook回调。让AI任务执行端在完成时回调一个指定接口，然后由这个接口去触发通知。这个思路比轮询优雅很多，但也引入一个限制：必须有公网可达的端点，否则回调根本进不来。对于个人开发环境、居家网络或者公司内网，这几乎是个不可逾越的门槛。
第三个方案，也是我最终选择的：在AI任务执行器层面装一个插件，任务完成时直接调用推送服务的API，把结果发给消息通道。这个方案的好处在于——它不依赖公网回调、不依赖本地回环，只要任务执行器能访问外网，消息就能推出来。而且切换不同的消息通道（企业微信机器人、钉钉机器人、Telegram Bot、邮件）非常灵活，完全可以在插件层面做适配层。
1.3 架构设计：插件嵌入任务生命周期
仔细想想，这个插件的定位非常清晰：它不该是一个独立的服务，而是作为Harness（调度执行框架）里的一个插件存在，监听任务生命周期事件。简单说，就是在任务启动、任务成功、任务失败这三个关键节点上，插入自定义动作。
这就引出了插件的位置问题。如果把通知逻辑写在业务代码里，每次写新任务都重复一遍，显然太蠢。但如果做成通用事件订阅机制，让插件注册到Harness的事件总线上，那所有任务都能自动获得通知能力，且互不干扰。我最终采用了后者：Harness负责执行任务、产生事件；插件负责订阅事件、构造消息、调用推送接口。
这个设计有四个明确的好处。一是解耦，任务逻辑完全不用关心通知怎么发，模块边界清晰。二是复用，不管未来跑多少种任务，通知能力自动生效。三是可靠，推送失败不会影响任务自身执行，插件内部的异常被独立捕获并记录。四是灵活，换消息通道只需要改插件配置，不用动任何任务代码。这条设计思路，我觉得是整套方案的核心价值所在。
2. 插件功能拆解：dsh-notify-push能做什么
2.1 三类核心事件：开始、成功、失败
既然定位是“任务生命周期通知”，那自然要知道到底有哪些事件可以订阅。我在Harness里保留了三个事件类型：任务启动事件、任务成功事件、任务失败事件。每一种事件所对应的消息侧重点完全不同。
任务启动时，重点传达的是“任务已开始跑，别干等着，去忙别的吧”——消息里带着任务名称、开始时间、预计耗时（如果有）。任务成功时，重点传达的是“活干完了，看结果吧”——消息里带着完成时间、耗时、结果摘要或提取的结果文件路径。任务失败时，重点传达的是“出问题了，需要人工介入”——消息里带着失败原因、异常堆栈前几行，如果有重试逻辑，还得带上重试次数。
比较反直觉的一点是：失败通知的优先级应该高于成功通知。因为失败意味着需要立刻介入，如果消息被群聊里的垃圾信息淹没了，重试就会被拖延。所以我特意在失败通知里做了一层优化：不仅推送文字，还会在文字前面加一个“高优先级”标记（不同通道表现不同，后面细说）。
2.2 消息通道适配：企业微信、钉钉、Telegram、邮件
消息通道是整个插件里最直接的差异点。因为不同推送服务的API格式千差万别，但核心逻辑是一致的：构造一个消息体，POST到一个接口地址。我把这层抽象成了统一的Notifier接口，每种通道实现一个适配器。
目前我实际完成并跑通的适配器有四个：
- 企业微信群机器人：通过群机器人的Webhook地址，发送Markdown文本消息。企业微信的Markdown支持程度一般，但足够用，胜在“手机端弹通知无需安装额外App”，对于公司内部办公场景非常友好。
- 钉钉群机器人：也是Webhook方式，支持Markdown，还能设置自定义关键字，避免被其他群消息干扰。
- Telegram Bot：通过Bot API发送消息，需要提前创建Bot并拿到Chat ID。这通道的好处是Telegram客户端在手机上的推送体验很“原生”，几乎和短信一个级别。
- SMTP邮件：通过邮件服务器发送。这个通道虽然慢，但适合“结果很长、需要存档”的场景，比如完整日志可以通过邮件归档。
在配置上，每个适配器只需关注自己的连接参数。比如企业微信需要Webhook地址，Telegram需要Token和Chat ID，SMTP需要服务器和账号密码。这些参数被我统一放在一个配置文件里，启动时加载到内存，不会硬编码在代码中。
2.3 真正的亮点：任务结果摘要截断与格式化
一开始我以为插件写到这里就够了——发个“任务完成”的总计，Who cares？但真用起来，发现信息密度不足才是最大的痛点。只推一句“任务完成”毫无意义，手机屏幕上全是无用的状态更新，最后我只能把通知权限关掉。
于是我在消息内容上做了大量打磨。核心思路是：每条通知都必须自带“让人三秒内判断要不要打开电脑”的信息量。具体来说，成功消息除了时间，还附上任务产物的摘要。如果任务是文本类，直接截取前300个字符作为正文；如果任务是文件类，则附上文件路径和文件体积；如果任务带了结构化输出（比如JSON），则提取关键字段转成简明格式。
截断逻辑有一个细节：不能简单用Python的[:300]硬切，否则中文可能被截断成乱码。我专门写了一个按字符安全切分的函数，同时处理后缀省略号。另外，截断后的内容会统一塞进代码块，避免不同平台把特殊符号渲染得乱七八糟。
3. 完整实现流程：从零到一搭建dsh-notify-push
3.1 环境与依赖准备
这套插件我是在个人开发机（Linux环境）上完成的，Python版本用了3.10。依赖管理选择了uv，因为它的解析速度和锁文件机制比老牌工具更顺手。项目结构很简单，核心就三个模块：事件监听器、通知分发器、通道适配器。
需要安装的外部依赖不多：
httpx：同步/异步HTTP客户端，用来向Webhook和Bot接口发请求。pydantic：配置模型的声明与校验，避免配置写错时到了推送环节才炸。apscheduler：可选，如果启用了失败重试通知，用做定时任务调度。
整个项目初始化也比较简单，一条命令创建一个虚拟环境并安装依赖即可。这阶段唯一需要注意的是：插件本身必须保持轻依赖，不要引入大型框架（比如完整的Django或FastAPI），否则会让原本轻量的Harness臃肿起来。
3.2 目录结构与核心类设计
我先给出最终的目录结构：
dsh-notify-push/ ├── pyproject.toml ├── README.md ├── dsh_notify_push/ │ ├── __init__.py │ ├── plugin.py # 插件入口，注册事件回调 │ ├── config.py # pydantic配置模型 │ ├── models.py # 事件数据模型 │ ├── formatter.py # 消息模板与截断逻辑 │ ├── notifier.py # 通知分发统一入口 │ └── channels/ │ ├── __init__.py │ ├── base.py # 通道基类 │ ├── wecom.py │ ├── dingtalk.py │ ├── telegram.py │ └── smtp.pyplugin.py是Harness的接入层。Harness通过约定好的接口实例化插件，插件内部负责注册事件回调。事件回调里做两件事：第一，根据事件类型构造消息payload；第二，调用Notifier分发。
Notifier类的设计非常简单直接，它内部维护了一个通道实例列表，send方法遍历这些通道逐个发送。如果某个通道发送失败，会记录一条日志但绝不抛出异常，因为通知的失败不能反过来让任务执行器崩掉。
3.3 关键实现细节：事件回调与异步推送
这里分享一个我踩过的坑。最初我用同步方式发送HTTP请求，也就是在Harness的回调线程里直接调用httpx.post()。这在任务成功时问题不大，因为是终点；但在任务启动和失败事件里，如果推送接口响应很慢（比如企业微信Webhook偶尔延迟几秒），回调线程就会被阻塞，严重时甚至拖慢整个任务调度。
解决方案很简单：把通知发送丢到后台线程池执行。我用的是concurrent.futures.ThreadPoolExecutor，设置一个容量很小的线程池（比如2个线程），避免产生大量等待线程浪费资源。这种方式在事件触发时立即返回，实际的网络IO在线程池里慢慢消化。 回调函数本身不做重活，只负责组装数据结构。
还有一点值得注意：某些事件回调里会访问任务的上下文对象，而这个对象在任务结束后可能被释放。所以我约定在回调入口先把需要的信息全部拷贝到dict里，后续的格式化、发送全部基于这个快照，绝不引用原上下文。
3.4 消息格式化器：模板化与截断策略
formatter.py承载了前文提到的“三秒法则”。我用一个基类MessageFormatter，派生成功、失败、启动三个格式化器。每个格式化器接收事件快照，返回一个字符串。
成功消息的模板大概长这样：
任务「{task_name}」已完成 耗时：{duration} --- {result_snippet}失败消息则更激进：
任务「{task_name}」执行失败 原因：{error_summary} 重试次数：{retry_count} --- {stack_snippet}result_snippet和stack_snippet都由formatter.py安全截断函数生成。我在这个函数里做了三件小事：按字符数限制、按换行符保留完整行、超出部分追加一行“...（截断，完整结果见日志）”。
3.5 通道适配器实现：企业微信与Telegram实例
以企业微信通道为例，适配器内部逻辑相当简单：读配置拿到Webhook地址、组装JSON载荷、POST请求。有一点特别重要：企业微信Markdown消息长度有限制，超过4096字节会被截断或直接失败。因此，通道内部也会再做一次长度校验，超长时按字节安全截断（注意不是按字符，企业微信是UTF-8编码，中文占3字节）。
Telegram通道则大不一样：Bot API的接口是sendMessage，需要把token拼成URL。消息长度限制放宽到4096字符，但这平台对HTML解析严格，如果消息里带了<或&这类符号而不做转义，发送会失败。为此我做了转义处理，把文本先经过HTML escape，再交给API。
3.6 配置模型与动态加载
配置这块我用pydantic来约束数据结构。总配置长这样：
notify: channels: - type: wecom webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx" enabled: true - type: telegram bot_token: "123456:ABC-DEF" chat_id: "987654321" enabled: true events: - on_start: true - on_success: true - on_failure: true一个值得分享的细节是：events字段允许用户单独关闭某一类事件的通知。比如我平时开着成功和失败，但连续高频调试时只想收失败通知，就把成功关掉，否则消息刷屏比不看还烦。
配置加载采用了懒加载策略：插件实例化时不立即解析Webhook地址，而是首次发送前才验证配置有效性。这样可以让那些临时只想静默测试的开发者有一个宽松的入口。
3.7 完整配置示例与跑通测试
为了方便大家直接抄作业，我贴一份实际测试用的配置。注意这里只展示结构，实际的key和token请替换成你自己的。
notify: channels: - type: wecom webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key" enabled: true - type: telegram bot_token: "你的token" chat_id: "你的chat_id" enabled: true - type: smtp host: "smtp.example.com" port: 465 username: "sender@example.com" password: "你的密码" to: "receiver@example.com" use_ssl: true enabled: false events: on_start: false on_success: true on_failure: true retry_on_failure: true max_retry: 3测试方式是：先在Harness里注册插件，再创建一个极简的任务（比如sleep 2 && echo done），然后分别验证成功和失败两种路径。验证结果会同时通过企业微信和Telegram收到消息。
3.8 部署与注册到Harness
安装方式分两种场景。一种是直接用Git仓库拉下来然后uv sync，适合个人使用；另一种是把插件目录打包成一个wheel，在Harness环境里通过pip安装，适合团队共享。
Harness这边需要一个注册步骤。以配置文件的方式注册，大致思路是：
from dsh_notify_push import NotifyPlugin plugin = NotifyPlugin(config_path="notify.yaml") harness.register_plugin(plugin)真正重要的是注册顺序必须发生在任务提交之前，但不能早于Harness核心初始化。否则插件可能拿不到Harness的事件总线实例。这个坑，稍微布置一下后续的测试就能发现。
4. 常见问题与排查技巧实录
4.1 企业微信Webhook提示key错误
这里最高频的困惑其实是忘在Webhook地址里带上key参数。很多人直接复制了一个没有key的地址，请求发出去了，企业微信返回错误码提示找不到群机器人。排查方法很直接：把Webhook地址完整打印出来，确认包含key=xxx且没有多余空格。
还有个细节：企业微信的Webhook地址在群设置里生成后，复制时可能会把结尾的换行符一起复制进去，导致拼接URL时出现%0A这类附加字符。我建议在代码里对URL做一次strip()，防一手。
4.2 Telegram Bot收不到消息
这类问题通常出在chat_id上。很多人初学Bot API时拿到的是Bot的getMe信息，却没正确获取用户的chat_id。正确做法是：先往Bot发一条消息，再调用getUpdates接口查看返回里的chat.id字段。只要拿到了这个值，填入配置文件，基本就能通。
如果填对了还收不到，排查流程就按顺序来：先看Bot的隐私模式是否开启了“仅允许联系人发送”，再看网络链路能不能访问Bot API所在域名，最后看日志里是否出现了非200响应。
4.3 消息发送太慢，拖累任务调度
这个坑我在前面提到过，但具体现象值得再讲一遍：把通知发送做成同步调用后，即使任务本身只花了1秒，但任务结束回调卡在网络请求上，整个任务生命周期从外部看可能变成10秒。用户感知就是“任务怎么还没跑完”。
排查时可以给HTTP客户端加超时参数，同时看线程池情况。我的建议是：无论通知逻辑多简单，统一走异步线程池，并设置合理的超时时间（比如连接超时5秒，读取超时10秒）。这样即使推送服务暂时抖动了，也不会影响主流程。
4.4 中文内容乱码或截断异常
现象是截断后的消息在Telegram中偶尔出现半个中文的乱码字符，或者在某些平台上显示成问号。根因是对“字符数”和“字节数”的混淆。简单说，在不同编码下中文占位不一样，直接用Python的切片也可能在代理组合字符上出bug。
我的处理方案是：单字节场景直接用[:300]，多字节场景先按字符数切片，再确保尾部没有半个字符。此外，对每个切片生成后自动拼接一个省略号在末尾，避免消息看起来像戛然而止。
4.5 重复通知或事件风暴
还有一类常见问题：任务失败后自动重试，每次重试都会触发一次失败事件，于是手机一次性收到3条一样的失败通知。表面上看是推送问题，实处是事件去重缺失。
我的解决方案在插件内新增一个简易的去重缓存：以task_id + event_type为键，记录最近一次通知时间，如果在短时间内（默认60秒）重复触发同类型事件，则合并为一条“重试中”通知，并附带累计次数。这样既不会漏报，也不会轰炸手机。
5. 进阶玩法与未来扩展思路
5.1 结合多Agent协作的最终汇总通知
如果你的Harness支持多Agent并行协作，每个Agent分别是独立的子任务，直接对所有子任务都开通知，会非常折磨人。更好的做法是只订阅“总控Agent”的事件，等所有子任务都结束之后，汇总各子任务的结果，再发一条总览通知。
我在这方面做了一点扩展：新增了一个“聚合通知模式”，将来自多个任务的输出聚合为一个列表，统一格式化后发送。这样，每天早上跑完一组分析任务，我只需要看一条消息，就能大体感知整组任务的状态。
5.2 接入自定义消息模板与条件过滤
推送做得太单一，很容易变成“通知噪音”。我在计划里加了两个小东西，一是模板文件支持Jinja2语法，让用户按自己的习惯定制通知文案；二是条件过滤规则，比如“只当结果包含关键字时通知”，这可以大大减少低频紧急任务对用户的打扰。
举个例子，某次任务里我只关心“异常率超过5%”的情况，那就设置过滤条件，只有该条件成立时才推送，其他情况静默。这种能力对于常态化的监控型任务非常实用。
5.3 从被动通知走向主动汇报
最后算一个脑洞方向的扩展：不单单是任务跑完推消息，而是让插件在每天固定时间主动汇总当天所有任务的执行情况，生成一份日报推送到手机。相当于从被动通知转为主动汇报，把插件的价值从“减少焦虑”升级为“节省复盘时间”。
目前的实现思路是复用已有的格式化器和通道，加一个定时任务，扫描Harness的任务执行历史，提取统计指标并排版。这个功能我已经在一个演示Demo中跑通了，效果让人满意，后续会继续完善后放出来。
写在最后的实操体会
这套插件真正让我舒服的点，不是“技术多炫”，而是它把AI工具从一个需要看守的“定时炸弹”变成了一位干活会主动汇报的同事。以前我开一个耗时任务，总惦记着去查一眼状态；现在只安心等一条消息，干完了自然知道，干失败了也能第一时间收到原因。这种“被通知”的安全感，是肉眼可见的效率提升。
如果大家要动手做类似的东西，我唯一想强调的就是先想清楚你的通知策略：哪些事件值得通知，哪些事件不该通知，截断多少内容合适，用什么通道最顺手。这比写代码难多了，也重要多了。把策略想明白，插件本身写起来真的很快，远远不用一个通宵。
