跑深度任务的时候最怕什么不是 Agent 答错题而是它跑完了没人知道。我自己把 AI Agent 接进日常自动化流程后经常遇到这种情况任务丢给 Agent自己转头去开会、写文档等想起来回去看终端它两小时前就跑完了结果那条链路中间有个参数错了整个任务等于白跑。我身边用 Agent 干活的朋友也普遍有这个困扰。邮件通知淹没在垃圾邮件里。日志告警你一天真的会看几次日志系统反而是微信不管是电脑上还是手机上几乎都能秒触达消息放一会儿不点开也不会丢。所以我花了点时间写了一个独立的微信推送服务专门承接 Agent 任务完成、失败、卡死这类状态的通知。这篇文章就把这套服务从需求拆解、架构设计到代码实现的完整过程分享出来给同样被“任务跑完没人通知”困扰的人一个可直接抄作业的方案。1. 为什么非要一个独立推送服务而不是直接在 Agent 里写几行推送代码先说个很现实的问题Agent 代码里直接调一下微信推送接口不就完事了吗为了一句话的事单独做一个服务是不是小题大做我自己也这么想过但在实际动手之前先把需求列了一遍发现直接在业务逻辑里写推送代码至少有三个绕不开的坑。1.1 直接写在 Agent 里的三个坑第一个坑是业务代码被污染。Agent 本身要做的事情是理解任务、拆解步骤、调用工具、汇总结果推送通知属于旁路逻辑。如果每个 Agent 里都混入一段“拼 Markdown、调 webhook、处理失败重试”的代码整个 Agent 的逻辑会越来越臃肿。你维护的是业务不是通知模块。第二个坑是多 Agent 场景下你根本写不过来。我现在本地跑的就不止一个 Agent有做数据处理管线的有定时抓取信息的还有调外部 API 做审核的。每个 Agent 都要通知如果每个都复制一段推送代码后面想统一改消息模板、想加一个备用推送通道就得所有 Agent 全改一遍。这种重复劳动属于典型的坏味道。第三个坑是定位问题依赖业务可观测性。推送如果失败了是 Agent 代码的问题还是推送接口的问题直接在业务里写出了问题排查链路会很长。独立出一个服务后Agent 只需要 “POST 一条消息” 出去剩下的事交给推送服务链路非常清晰。1.2 通知方式选型为什么最后还是落在微信确定要独立做推送后接下来是通道选型。我把自己能想到的方案全部列了一遍各有各的问题最后才定下微信。通知方式触达速度手机上能否及时收到维护成本适合场景终端输出盯梢实时否人必须在电脑前无交互式调试Web 面板/轮询一般要自己打开页面看中有运维面板的团队邮件快但容易被忽略会有提醒但常被折叠低正式报告、周报IM 机器人企业微信/钉钉/Slack秒级是低团队协作、单人通知短信快是高按条收费付费告警、紧急情况邮件在触达率上其实没那么差但现在的邮箱收件箱基本是个信息垃圾场一条 Agent 通知进去几分钟就被淹没了。短信效果好但按条付费对高频任务来说成本不划算。最终我选择了微信生态里的通知方式核心原因是微信的到达率和阅读率在我个人环境中最高无论电脑还是手机钉在聊天列表里的通知几乎不会被漏掉。而且微信通知不需要额外装客户端对日常办公的人来说微信本来就是常驻软件。2. 推送服务的整体设计先想清楚再做别一上来就写接口通道定下来之后我没有立刻写代码而是先把整个服务要承担什么职责理了一遍。这个步骤很关键很多半路烂尾的推送服务都是因为一开始没想清楚它到底是一个“发送工具”还是一个“通知中枢”。我最后把职责定成接收 Agent 的状态事件格式化成可读消息通过微信通道推送并保证消息尽量不丢、不重、不乱。围绕这个职责服务内部拆成了四层。2.1 服务内部的四个逻辑层接入层或者说 API 层它的职责是给所有 Agent 一个统一的入口。我的做法是一个 HTTP 接口Agent 不用关心后面走的是什么通道只负责把结构化的事件数据 POST 过来。这一层要做得足够简单协议越简单Agent 接入成本就越低。消息缓冲层这是我特意加的。如果 Agent 在短时间内跑完十几个子任务一下子来了十几个推送请求微信通道又有频率限制直接同步推送容易把通道打爆。这一层用一个内存队列缓冲消息接口收到请求之后立刻返回 200后台再慢慢消费队列去推送。模板与格式化层。Agent 传过来的是 JSON 结构体直接推给微信会非常丑。这一层负责把结构化数据渲染成适合在微信聊天窗口里阅读的 Markdown 文本成功通知、失败通知、卡死告警各有各的模板以后改模板只改这一层。通道层也是最底层的适配层。它封装了具体的微信推送实现可以是企业微信群机器人 webhook也可以是 Server酱这类第三方服务。通道层把“推送”这个动作抽象成一个统一函数上层根本不关心消息到底从哪个通道发出去的。2.2 关键设计取舍同步还是异步重试还是放弃在思考架构时最纠结几个点我把结论和理由一并说清楚。接口到底是同步推送还是异步推送我选了异步。原因是同步推送把推送的耗时直接算进了 Agent 主流程如果微信接口刚好卡了 5 秒Agent 就得无谓等多 5 秒。异步模式下接口只做一件事把消息丢进队列立即返回成功。Agent 端没有感知推送服务的压力也小。消息要不要重试当然要但必须有限度。通道调用失败最常见的原因是网络抖动、微信频率限制、偶发的超时。我做了简单的失败重试前 3 次间隔 1 秒、5 秒、15 秒超过 3 次还是失败就把这条消息落到本地日志里等待人工处理。为什么不无限重试因为有些失败不是重试能解决的比如 webhook 地址配置错了再多次重试都是徒劳。要不要做消息去重要。Agent 任务触发通知时最怕的是 Agent 本身也做重试或者推送服务消费队列时因为超时导致重复消费。我在服务里维护了一个最近处理过的 task_id 集合重复的 task_id 直接丢弃保证同一条任务通知不会在微信里刷屏。注意消息队列用内存实现有一个前提就是部署形态必须是单进程。如果以后要把推送服务做成多实例高可用内存队列必须换成 Redis。我目前是自己用单进程足够所以没有为了一个通知服务特意引入 Redis 这个重量级依赖。2.3 部署形态与控制面这个服务做出来是要常驻运行的所以我按“轻量、低维护”的标准来部署。一个 Python FastAPI 服务运行在一台小机器或者家里的 NAS 上不需要数据库数据持久化只需要一个 SQLite 文件记录推送历史甚至连历史记录都可以不要。只要保证 Agent 能访问到服务的地址即可。其实通知服务本质上是个强调稳定性的边缘系统它自己不应该有太复杂的依赖。依赖越少运维成本越低反而是最大的稳定性。3. 核心实现从零搭一个可用的微信推送服务设计讲完了进入实操环节。前面说的四层结构落到底层其实就是几件事搭一个 HTTP 服务接收 Agent 消息、写消息格式化模板、封装微信推送通道。这一节我给出能直接跑的代码和关键配置。3.1 微信推送通道的几种接法目前普通人最容易接触到的微信推送方式主要有三种我分别说下优劣。企业微信群机器人 webhook 是最简单直接的方式。在任意一个企业微信群里添加一个群机器人就能拿到一个 webhook 地址向这个地址 POST 一段 JSON消息就会出现在群里。好处是不需要企业认证个人也能建群添加机器人支持 Markdown 文本消息格式比较自由。Server酱方糖这类第三方服务也不复杂。它本质上借助微信公众号的模板消息能力你关注一下它的服务号拿到一个 SendKey调用它的 HTTP API 就能把消息推到你的微信里。消息会以服务号模板消息的形式出现好处是个人微信直接收不需要企业微信客户端坏处是模板消息在聊天列表里的存在感不如群聊消息。微信公众号模板消息属于更正统的做法需要你自己注册一个公众号申请模板消息权限然后调公众号接口推送。缺点很明显门槛高、审核周期长、模板消息调用有较大的频率限制个人开发者不值得折腾。通道方案是否要求企业身份消息形式个人可接入难度我的推荐程度企业微信群机器人不需要认证个人可建群群内消息低高Server酱不需要微信服务号模板消息低中适合个人直接收消息公众号模板消息需要公众号模板消息高低我自己主力用的是企业微信群机器人 webhook原因很实在它能推 Markdown能渲染成结构化内容一个专门建来接收 Agent 通知的群里消息不会打扰到别人而且 webhook 方式完全免费、没有模板审批流程。Server酱我作为备用通道保留万一企业微信 webhook 偶尔被风控或者网络不通可以自动切到 Server酱。3.2 接收端 API 实现服务端核心代码我用 FastAPI 写完整流程是接收 POST /notify 请求把消息体放入全局消息队列后台线程负责消费队列并调用微信通道发送。这一步把接收和发送彻底解耦。import json import queue import threading import time import os from fastapi import FastAPI from pydantic import BaseModel app FastAPI() msg_queue queue.Queue() class NotifyBody(BaseModel): event: str # 事件类型task_finished / task_failed / task_timeout / heartbeat agent: str # Agent 名称 task_id: str # 任务唯一 ID用于去重 status: str # 任务状态success / failed / timeout message: str # 结果摘要 duration: int 0 # 任务耗时单位秒 extra: dict {} # 扩展字段可放日志链接等 def worker(): while True: item msg_queue.get() if item is None: break try: text render_template(item) send_qyweixin(text) except Exception as e: # 重试失败后写入日志等待人工处理 log_failed_push(item, str(e)) finally: msg_queue.task_done() app.on_event(startup) def start_worker(): t threading.Thread(targetworker, daemonTrue) t.start() app.post(/notify) def notify(body: NotifyBody): # 去重逻辑task_id 如果近期处理过直接丢弃 if deduplicate(body.task_id): return {code: duplicated, task_id: body.task_id} msg_queue.put(body) return {code: ok, task_id: body.task_id}这个接口少了很多花哨的东西但每个字段都有明确用途。event 和 status 决定了走哪个模板task_id 是去重和追踪的核心duration 用来在通知里直观展示任务耗时extra 里我一般放一个任务详情页或者日志文件的链接。3.3 消息格式化让通知一眼就懂消息格式化我单独讲是因为这一层直接决定了你每天看到的通知好不好用。如果只推送一句“任务跑了 23 分钟”跟没推差不多如果推送一整段冗长的 JSON你也懒得看。我最常用的四种模板任务成功、任务失败、任务超时、Agent 心跳。成功和失败是主线超时是给长时间静默任务用的心跳是低频巡检用来确认 Agent 还活着。def render_template(body: NotifyBody) - str: if body.event task_failed: title 任务失败 elif body.event task_timeout: title 任务超时 elif body.event heartbeat: title 心跳巡检 else: title 任务完成 duration_text if body.duration: minutes, seconds divmod(body.duration, 60) duration_text f- 耗时{minutes} 分 {seconds} 秒\n text f## AI Agent 通知{title} - 任务 ID{body.task_id} - Agent{body.agent} - 状态{body.status} {duration_text} {body.message} if body.extra.get(url): text f[查看详情]({body.extra[url]})\n return text这个模板不长但在群聊里一眼扫过去任务 ID、哪个 Agent、什么状态、跑了多久、结果摘要全都有了。这比一堆日志片段要直观得多。3.4 微信通道封装企业微信机器人与备用通道通道层我做了两个实现一个是企业微信群机器人一个是 Server酱。企业微信的 webhook 推送支持 Markdown我用了 Markdown 格式。import httpx QYWX_WEBHOOK_URL os.getenv(QYWX_WEBHOOK_URL, ) def send_qyweixin(markdown_text: str): payload { msgtype: markdown, markdown: { content: markdown_text } } resp httpx.post(QYWX_WEBHOOK_URL, jsonpayload, timeout10) resp.raise_for_status() SERVERCHAN_SENDKEY os.getenv(SERVERCHAN_SENDKEY, ) def send_serverchan(title: str, desp: str): url fhttps://sctapi.ftqq.com/{SERVERCHAN_SENDKEY}.send resp httpx.post(url, data{title: title, desp: desp}, timeout10) resp.raise_for_status()主通道失败时切备用通道的逻辑放在 worker 里主通道抛异常后先打日志再尝试备用通道。这里注意备用通道的标题和正文要重新组装一下因为 Server酱不直接支持 Markdown但 desp 字段支持简单排版。4. 与 AI Agent 的集成方式四种姿势看你 Agent 是怎么跑的推送服务本身接口很简单难点在于怎么让 Agent 在任务结束时“自然地”触发这个通知。根据 Agent 的形态不同我总结了四种接入方式各有各的适用场景。4.1 HTTP 回调最适合自己写的 Agent如果你是自己用代码在跑 Agent不管是 LangChain、自研的工作流还是简单的脚本 Agent最干净的方式就是任务结束时直接调一下推送服务的 HTTP 接口。我用 Python 封装了一个工具函数Agent 收尾时调用它业务代码里只多了一行推送细节全部隔离在工具函数里。import httpx import os NOTIFY_URL os.getenv(NOTIFY_URL, http://localhost:8000/notify) def notify_done( agent: str, task_id: str, status: str, message: str, duration: int 0, extra: dict | None None, ): body { event: task_finished if status success else task_failed, agent: agent, task_id: task_id, status: status, message: message, duration: duration, extra: extra or {}, } try: httpx.post(NOTIFY_URL, jsonbody, timeout3) except Exception: # 通知失败不影响主业务 pass把 timeout 设置成 3 秒这个细节很重要通知服务如果挂了接口超时最多拖住主业务 3 秒不会让 Agent 本身卡住。异常处理直接 pass是因为推送通知不能反过来绑架业务主流程。4.2 装饰器埋点适合封装成类的 Agent如果 Agent 是面向对象的写法可以用装饰器在任务方法结束时自动通知避免每个方法里手动加一行。装饰器的好处是侵入性更低只要在方法上打一个标记成功和失败都会自动进入通知逻辑。def notify_decorator(agent_name: str): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): task_id ft-{int(time.time())} start time.time() try: result func(*args, **kwargs) duration int(time.time() - start) notify_done(agent_name, task_id, success, str(result), duration) return result except Exception as e: duration int(time.time() - start) notify_done(agent_name, task_id, failed, str(e), duration) raise return wrapper return decorator notify_decorator(data-pipeline-agent) def run_agent_task(): # Agent 核心逻辑 pass这个方式适合那种长期运行、内部状态比较复杂但不方便直接改主流程的 Agent 类实现。4.3 日志监听适合外部黑盒 Agent有些 Agent 不是你写的或者不太方便改它的代码比如你跑的是一个开源项目你不想为了加通知去改动别人的源码。这种情况下可以接受侧边方案让 Agent 正常往 stdout 或日志文件写内容然后写一个轻量日志监听脚本抓取关键关键字比如 “Agent finished”“ERROR”“Timeout” 之类触发推送。日志监听的实现不复杂本质就是一个 tail 模式脚本逐行读日志正则匹配到关键字就构造一条推送消息。但要注意日志监听是脆弱的方式依赖日志格式稳定。Agent 版本一升级日志格式一变匹配规则可能全部失效。4.4 定时任务场景没有回调就主动查结果有些 Agent 不是“跑完能回调”而是作为定时任务被动运行。比如每天早上 9 点跑一次销售数据汇总跑完结果写进数据库。这种情况推送服务的角色要反过来不是等 Agent 通知而是到了时间点主动去查结果有结果就推没结果就推一条“任务未执行”。我用 APScheduler 写了一个小的轮询任务每隔一段时间查一下 Agent 的结果表发现新的成功记录就往推送服务发一条通知。这里核心参数是轮询间隔间隔太短会增加数据库压力太长通知不够及时。我一般设置 1 分钟轮询一次对绝大多数结果类任务足够。5. 实操验证从部署到收到第一条微信通知的完整路径代码写完了是骡子是马拉出来遛遛。这一节我走一遍完整流程从启动推送服务到 Agent 触发通知再到微信群里收到消息整个过程大概 5 分钟。5.1 启动推送服务先把代码保存成notify_server.py安装依赖后启动pip install fastapi uvicorn httpx pydantic export QYWX_WEBHOOK_URLhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的机器人key uvicorn notify_server:app --host 0.0.0.0 --port 8000看到Uvicorn running on http://0.0.0.0:8000说明服务已经起来了。注意 QYWX_WEBHOOK_URL 是敏感凭证不要写进代码仓库用环境变量注入。5.2 用 curl 模拟 Agent 发一条测试通知服务起来后先手动发一条消息验证链路。这一步在做任何 Agent 集成之前都要做可以帮你把“Agent 代码问题”和“推送链路问题”彻底隔开。curl -X POST http://localhost:8000/notify \ -H Content-Type: application/json \ -d { event: task_finished, agent: manual-test, task_id: t-20250118-001, status: success, message: 这是一条由 curl 触发的测试通知, duration: 125, extra: {url: http://example.com/logs/123} }如果正常接口会立即返回{code: ok, task_id: t-20250118-001}这时候企业微信群里应该已经出现一条 Markdown 格式的任务完成通知了。这一步能通整条推送链路就是通的。5.3 在 Agent 主流程里接入并验证推送链路没问题后把第 4 节的notify_done函数放进 Agent 工程里在 Agent 执行结束的位置调用。跑一次真实任务观察两个点一是 Agent 主流程没有因为推送而变慢二是微信群里消息内容里 task_id、耗时、摘要都显示正确。我自己第一次接入的时候还验证了异常场景故意让 Agent 抛了一个异常确认失败通知能收到然后又模拟了推送服务挂掉的情况确认 Agent 主流程不会因为推送失败而出问题。这三个场景都跑通之后整套方案才算真正落地。6. 踩坑实录微信推送链路常见的 6 个问题与排查方法上线这几个月我不是没翻过车。下面这些坑都是我在实际操作中遇到的整理成了速查表能帮你绕开大部分常见问题。问题现象可能原因排查与解决接口返回 ok 但群里没消息webhook 地址配错或群机器人被移除先 curl 直接调 webhook看返回是否提示“invalid webhook”消息内容里中文显示乱码HTTP 请求头缺少 UTF-8 编码或 JSON 中用了不兼容字符确认请求头Content-Type: application/json; charsetutf-8消息发多了被提示“超过频率限制”企业微信机器人有每分钟条数限制在缓冲层做限速队列消费时加间隔最多每分钟 20 条以内手动测试 webhook 时报“keywords not in content”创建机器人时设置了推送关键词但消息里没有包含在消息模板里固定加一个关键词或在机器人设置里改为不限制关键词同一条任务被重复通知很多次Agent 自身有重试或推送服务重复消费用 task_id 做幂等处理过的 task_id 直接丢弃推送服务明明在跑Agent 调接口却超时双方网络不通或推送服务所在机器防火墙挡了端口从 Agent 所在机器 curl 一下推送服务地址判断网络连通性6.1 两条最重要的排查经验第一是“企业微信机器人必须在创建时记住关键词这个配置”。我版本迭代的时候重建过机器人忘了加关键词结果推送的 Markdown 里没有任何一个命中关键词的词消息全部被微信服务端静默丢弃。当时查了很久最后一条一条重新读 webhook 文档才发现问题。所以这里特别提醒一句如果你在创建企业微信机器人时设置了自定义关键词那么你推送内容的纯文本部分必须包含其中一个关键词。第二是“推送失败不能只靠看日志”。我把推送服务的日志打到文件里但风险是文件轮转很久之后你根本记不清某条消息是失败还是根本没触发。后来我在服务里加了一个简单的可用性检查接口/healthz每隔几分钟从外部调一次同时把最近一次推送时间和状态暴露在一个/status接口里团队里有人怀疑“推送是不是挂了”直接看一下 status 接口就知道。6.2 给服务做一层稳定性加固跑了一段时间后我逐渐意识到通知服务的核心指标是“不掉链子”所以在基础版本上做了几层加固。第一层是缓冲队列的持久化降级。内存队列最怕进程重启重启时队列里还没消费的消息会全部丢失。我的方案是消费失败的消息落盘到一个 JSON Lines 文件重启服务时先加载未发送成功的消息重新入队。第二层是备用通道自动切换企业微信 webhook 连续失败两次后自动切换到 Server酱通道。第三层是给服务加了简单的系统监控进程还活着队列积压长度在合理范围最近一次推送成功时间不超过 24 小时。任何一项异常用通道自愈的方式再推一条告警给自己。7. 最后再聊点实际感受这个推送服务写完之后最大的体感变化不是“我终于能收到通知了”而是我的 Agent 使用方式开始变了。以前只有短任务才敢放手让它跑现在跑几小时的大任务也不用盯着屏幕微信一响看到“任务完成”就能继续下一步看到“任务失败”也能第一时间回去处理Agent 真正从“玩具”变成了可以托付工作的工具。还有一点是通知服务别做太大。很多人一听要做推送服务就想上 Redis、上消息队列、上 Grafana完全没必要。它就是个“小而关键”的边缘系统能撑住几十个 Agent 的日常通知需求就够了。把复杂度控制在“一个 Python 进程 一个队列 一个 webhook”的级别维护成本极低反而能稳定跑很久。最后分享一个小经验通知服务尽量在所有 Agent 项目里统一使用不要这个 Agent 用企业微信那个 Agent 用 Server酱各搞各的。统一到一个服务之后排查问题、改模板、加通道都只要改一个地方这也是我最初决定独立做这个服务的核心动机。