1. Agent跑完没人告诉你这类通知需求到底要解决什么问题如果你已经开始把AI Agent用在真实任务里——不管是让LangGraph跑一个多轮推理的爬虫还是用Claude写代码、调脚本、批量处理数据——你大概率也遇到过这种场景任务丢进去之后它自己跑起来了但你可能是在开会、在吃饭、甚至已经躺下睡觉了。等你想起来切回终端看一眼发现任务其实在十几分钟前就结束了全程没人告诉你。我自己最崩溃的一次是在半夜跑一个批量处理任务Agent需要逐个读取上百个文件并生成结构化摘要按当时的进度估算得跑一个多小时。我把命令一挂就去睡了。第二天早上醒来一看终端里登录会话早就断了进程也在凌晨就挂掉了——因为我没处理网络断连和会话超时的问题。但关键问题不在进程崩溃而在于我没有任何渠道及时知道它跑完了或者跑失败了。那这个Agent做得再智能对于需要等待、然后拿结果的人来说体验依然是断裂的。这就是AI Agent跑完任务怎么通知你这个看起来很不起眼的问题实际上的价值所在。它不只是省点心的问题而是把Agent从一个运行在黑盒里的程序变成了真正能被人调度的生产工具。我当时就想得很明白我要的不是什么花哨的看板、复杂的监控系统我要的只是一个足够可靠、延迟低、并且我手机里一定会第一时间看到的东西——微信推送服务。微信推送的好处不需要我多讲手机必装、推送到达率高、消息可回溯、不需要额外装App。于是我就花了一个晚上写了这个推送服务并且在接下来的几个星期里把Agent项目的通知链路全部接了上去。这篇文章就把我这次的完整思路、代码实现、以及踩过的坑全部写出来。适合的人群有两类一是已经在跑Agent、但还在用人肉盯梢方式等结果的人二是打算自己搭一套通知服务、但不知道从哪下手的人。不管你是用LangChain、LangGraph、AutoGPT还是自研Agent这套思路都能直接用。2. 通知渠道选型微信推送为什么打赢了邮件和钉钉先说结论选微信推送不是因为它技术最先进而是因为它是用户门槛和实现成本之间平衡得最好的一条路。我刚开始也没直接选微信先花了一点时间把所有候选方案列了一遍逐个验证最后才定下来。2.1 常见的Agent通知方案对比我给Agent接通知的时候周围小伙伴用的大概是这几种路子通知渠道优点缺点适合场景邮件SMTP协议标准、代码成熟、可以发附件延迟不稳定、容易被丢进垃圾箱、手机不一定会立刻弹通知长时间批处理、需要附件的正式报告钉钉/企微群机器人到达快、免费、配置简单、支持Markdown需要企业组织、个人使用者需要先建个群团队协作场景短信到达率高、绝对可靠每条都要钱、签名审核麻烦、有频率限制高优先级告警微信服务号/企业微信/小程序订阅消息手机必装、触达率高、免费需要有对应的开发配置或第三方封装个人Agent和轻量通知自建App Push可定制性最强要开发App、运维成本高几乎没有需求看了一圈就能发现对个人开发者来说最务实的其实就是两个方向一是用钉钉或企业微信群机器人二是走微信生态的推送服务。我当时两个方向都试了最后留下了企业微信群机器人这条路作为主力——因为它既能发到微信客户端收到通知又是官方提供、代码实现极其简单的方案。2.2 关于个人微信能不能直接推的边界问题这里必须提醒一下很多人第一反应是我直接用个人微信给自己发消息不就好了但个人微信官方并没有开放API给第三方直接调用。市面上那些个人微信机器人大多基于非官方协议本质上属于逆向和外挂轻则随时失效重则账号受限。这条路我坚决不碰原因很简单——Agent通知是生产链路里最后一段它如果挂了前面做得再好都白搭我不能把最后一段建立在随时可能被封的沙子上。所以我说的微信推送服务实际指的是两条官方合规路径企业微信群机器人Webhook注册一个企业微信个人就能注册建一个群添加群机器人得到一个Webhook地址。向这个地址POST一段JSON消息就会出现在企业微信里你在微信里也能收到提醒。免费、不需要审核、当天就能搞定。第三方推送平台Server酱、PushPlus、WxPusher等这类平台的模式是——用微信扫码绑定你的微信身份平台给你一个SendKey或Token你的代码把消息POST到平台API平台再通过自己的企业微信或微信模板消息推给你。优点是连企业微信都不用自己配缺点是信任链在第三方手里且部分平台有每日条数限制。我个人的取舍是主力用企业微信群机器人备用一个第三方平台做降级容错。这样既有充分的控制权又不会出现单点故障。下面代码部分我会把两条路径都写出来。2.3 最终架构的形态整个通知服务我没打算引入什么重量级框架最后定下来的架构非常简单AI Agent执行任务 ↓ (任务开始/结束/失败时调用) 通知客户端模块HTTP请求 ↓ (POST JSON) 企业微信群机器人Webhook / 第三方推送平台 ↓ (微信客户端内弹出消息) 你本人这个架构的好处是所有组件都是标准HTTP接口不依赖内网互通不依赖Agent运行在哪里——本地笔记本、云服务器、Docker容器里都一样能推。而且它天然跨语言今天你用Python写Agent明天换成TypeScript通知这一段逻辑完全不受影响。3. 推送代码落地一个函数跑通通知主链路聊完选型和设计直接上代码。这里我不搞那种几十层的抽象架构先给一个能立刻跑起来的最小实现再逐步加功能。3.1 企业微信群机器人的最简实现先创建一个群机器人在企业微信客户端里找到工作台-添加成员-添加机器人选一个群加入拿到一个形如https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxxxxxx-xxxxxxxx-xxxxxxxx的Webhook地址配置就完成了。然后是一个极简的Python函数import requests import json import time WEBHOOK_URL https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的key def send_wechat_notify(content: str, msg_type: str text, webhook_url: str WEBHOOK_URL, max_retries: int 3) - bool: 向企业微信群机器人发送消息 :param content: 消息文本内容 :param msg_type: text 或 markdown :param webhook_url: 机器人webhook地址可覆盖 :param max_retries: 失败重试次数 :return: 是否发送成功 payload { msgtype: msg_type, msg_type: {content: content} if msg_type text else {content: content} } headers {Content-Type: application/json} for attempt in range(1, max_retries 1): try: resp requests.post(webhook_url, jsonpayload, headersheaders, timeout10) result resp.json() # 企业微信返回码: 0 表示成功 if result.get(errcode) 0: return True # 频率限制特殊处理 if result.get(errcode) 45009: # 每60秒最多发20条, 这里睡一会儿再重试 time.sleep(60) continue print(f[notify] 发送失败: {result}) return False except requests.exceptions.RequestException as e: print(f[notify] 网络异常(第{attempt}次): {e}) if attempt max_retries: time.sleep(2 * attempt) else: return False return False这个函数干了几件容易被忽略的事设置了超时时间10秒。如果没有超时当网络异常时requests.post会一直挂着导致Agent主流程卡在通知这一步。通知是辅助链路绝不能让它的故障拖累主任务。处理了频率限制。企业微信群机器人的限制是每分钟最多20条超了会返回errcode为45009如果代码里不处理某些批量场景下会莫名其妙丢通知。做了简单的重试。网络抖动在现在这种环境里太常见了第一次请求失败就放弃太重了。我按指数退避做三次重试实测下来绝大多数临时故障都能自动恢复。3.2 消息内容格式的设计要点很早的时候我踩过一个坑消息里什么状态都写最后堆出来一篇又臭又长的报告自己根本没心思看第二遍。后来我把每类消息的格式做了标准化现在形成了这几个模板任务开始通知{ msgtype: text, text: { content: [Agent] 任务已启动\n项目: 数据清洗\n任务ID: 20250101_001\n开始时间: 2025-01-01 14:30:22\n预计耗时: 30分钟 } }任务成功通知{ msgtype: markdown, markdown: { content: **[Agent] 任务已完成** \n\n 项目: 数据清洗\n 任务ID: 20250101_001\n 耗时: 28分15秒\n 输出: 处理文件128个成功率 100%\n\n[日志详情](http://你的流水线地址/tasks/20250101_001) } }任务失败通知{ msgtype: markdown, markdown: { content: **[Agent] 任务执行失败** \n\n 项目: 数据清洗\n 任务ID: 20250101_001\n 失败阶段: 文件解析\n 失败原因: 第57个文件格式非法\n 耗时: 12分03秒 } }有几个细节我想特别说明一下Markdown模式只对企业微信的机器人消息生效而且支持的是它的特定子集标题、加粗、引用、链接不是完整Github Flavored Markdown。图片、表格这类元素在不同客户端表现不稳定我一般只用加粗、引用和链接。每条消息都要有项目和任务ID。当多个Agent并行跑的时候没有这两个字段你根本分不清是谁发的消息。任务ID需要能在你的任务系统里反查到日志这是排查问题的基础。失败通知里必须写明失败阶段和失败原因。很多人只发任务失败四个字收到消息后还得去翻日志那推送的及时性就白费了。我的习惯是——失败原因尽量让代码自己捕获并拼进去这样90%的情况你不需要打开电脑就知道该从哪里下手。3.3 用第三方平台做备用通道企业微信这条路本身已经相当稳了但为了做到双保险我还加了一路备用通道——用Server酱的API。这类平台的调用更简单本质上就是一个GET请求发送完还能在手机上看到推送历史记录。核心代码import requests SERVER_CHAN_KEY 你的SendKey def send_serverchan(title: str, content: str) - bool: 备用通知通道: 通过Server酱推送到微信 url fhttps://sctapi.ftqq.com/{SERVER_CHAN_KEY}.send payload { title: title[:32], # 标题限制32字符 desp: content, } try: resp requests.post(url, datapayload, timeout10) result resp.json() return result.get(code) 0 except Exception as e: print(f[serverchan] 通知失败: {e}) return False在正式的发送入口里我的发送顺序是先走企业微信失败再走Server酱。封装起来就是def send_agent_notify(title: str, content: str) - None: ok send_wechat_notify(content) if not ok: send_serverchan(title, content)注意Server酱的标题有32字符限制直接把完整内容怼进去会把接口打挂。所以我在封装层做了截断处理——长字段一律放正文标题只放项目名和状态。4. AI Agent接入推送的三种姿势服务端写好了接下来就是怎么让它和Agent接上。我在实际开发中总结出三种方式分别对应不同场景你们可以根据自己Agent的架构选。4.1 姿势一包装器方式最简单几乎万能如果你的Agent是拿某个函数或某条命令启动的比如result run_agent(task给这些数据生成摘要)那你不用改Agent内部代码只用包装器包一层import traceback from datetime import datetime def notify_agent_run(func): 装饰器: 自动给Agent任务加开始/结束/异常通知 def wrapper(task, *args, **kwargs): start_time datetime.now() send_agent_notify( f[{task.get(project, Agent)}] 任务已启动, f任务ID: {task.get(task_id, N/A)}\n开始时间: {start_time:%Y-%m-%d %H:%M:%S} ) try: result func(task, *args, **kwargs) elapsed (datetime.now() - start_time).total_seconds() send_agent_notify( f[{task.get(project, Agent)}] 任务已完成, f任务ID: {task.get(task_id, N/A)}\n耗时: {elapsed/60:.1f}分钟\n结果: {str(result)[:500]} ) return result except Exception as e: elapsed (datetime.now() - start_time).total_seconds() error_info traceback.format_exc() send_agent_notify( f[{task.get(project, Agent)}] 任务失败, f任务ID: {task.get(task_id, N/A)}\n耗时: {elapsed/60:.1f}分钟\n异常: {e}\n堆栈尾部: {error_info[-500:]} ) raise return wrapper # 使用 notify_agent_run def run_agent(task): # Agent原有逻辑 pass这种包装器的方式有两点好处改成装饰器后只需要在原来的函数定义上加一行notify_agent_run完全不需要动Agent内部的推理逻辑。失败时发完通知再把异常抛出去保证Agent原有的错误处理链路不受影响外面该重试的重试、该告警的告警。4.2 姿势二框架回调方式适合LangChain等如果你用的是LangChain或LangGraph这类有回调机制的框架更优雅的做法是直接实现它的callback接口。比如LangChain的CallbackHandlerfrom langchain_core.callbacks import BaseCallbackHandler class WeChatNotifyHandler(BaseCallbackHandler): Agent关键节点回调通知 def __init__(self, task_id: str, project: str): self.task_id task_id self.project project self.start_time None def on_chain_start(self, serialized, inputs, **kwargs): if self.start_time is None: self.start_time datetime.now() send_agent_notify( f[{self.project}] Agent开始执行, f任务ID: {self.task_id}\n时间: {self.start_time:%Y-%m-%d %H:%M:%S} ) def on_agent_finish(self, finish, **kwargs): elapsed (datetime.now() - self.start_time).total_seconds() output_text finish.return_values.get(output, ) if isinstance(finish.return_values, dict) else str(finish.return_values) send_agent_notify( f[{self.project}] Agent执行完成, f任务ID: {self.task_id}\n耗时: {elapsed/60:.1f}分钟\n输出摘要: {str(output_text)[:300]} ) def on_chain_error(self, error, **kwargs): elapsed (datetime.now() - self.start_time).total_seconds() send_agent_notify( f[{self.project}] Agent执行失败, f任务ID: {self.task_id}\n耗时: {elapsed/60:.1f}分钟\n错误: {str(error)[:300]} )然后把这个handler挂到你的Agent上handler WeChatNotifyHandler(task_idtask_001, project数据摘要) agent_executor create_agent_executor(...) agent_executor.run(处理这批文件, callbacks[handler])这种方式最优雅的地方在于它能同时捕获LangChain内部工具调用的完成事件。比如Agent中途调用了某个搜索工具或代码执行器你可以在on_tool_end里追加通知真正做到Agent每一步干了什么你都心里有数。但也要小心不要在回调里做太多同步操作——通知发送控制在几十毫秒内没问题如果逻辑复杂了建议改成异步任务避免拖慢Agent的推理主流程。4.3 姿势三HTTP回调方式适合Agent跑在远程服务器很多生产级Agent并不是和你在同一台设备上跑的你本地开发、代码推到服务器上、服务器上的Agent异步执行任务。这时包装器和回调都不太适用了——因为代码执行完你和服务器之间的连接可能已经断开了。我的做法是在通知服务里加一个极小的HTTP入口让Agent执行完自己往这个入口打一个请求from fastapi import FastAPI, Request from pydantic import BaseModel app FastAPI() class NotifyPayload(BaseModel): project: str task_id: str status: str # success / failed / started message: str elapsed_seconds: float 0 app.post(/agent/notify) async def agent_notify(payload: NotifyPayload): 远程Agent回调入口 if payload.status success: title f[{payload.project}] 任务成功 content f任务ID: {payload.task_id}\n耗时: {payload.elapsed_seconds/60:.1f}分钟\n{payload.message} elif payload.status failed: title f[{payload.project}] 任务失败 content f任务ID: {payload.task_id}\n耗时: {payload.elapsed_seconds/60:.1f}分钟\n错误: {payload.message} else: title f[{payload.project}] 任务启动 content f任务ID: {payload.task_id}\n{payload.message} send_agent_notify(title, content) return {code: 0}Agent那边只需要在结束时发一次请求import requests requests.post(http://你的服务器:8000/agent/notify, json{ project: 数据清洗, task_id: 20250101_001, status: success, message: 处理文件128个成功率100%, elapsed_seconds: 1695 })这个HTTP回调方案有个额外的安全要求是必须做的不要把这个端点裸放在公网上。我的做法是加一个简单的Token鉴权Agent请求时带上请求头Authorization: Bearer 你定的tokenFastAPI这边做一个依赖注入校验否则任何人都能往你的微信里塞垃圾消息。5. 上线实测踩过的坑和对应的修复手段前面都是理想状态下的设计。真正把这套通知服务接到Agent上跑了一个多月之后我才知道发通知这件事本身充满了让程序员头疼的细节。这里写几个最典型的坑和对应的解决经验。5.1 坑一长消息触发企业微信的需要阅读全文企业微信机器人发长文本时微信端会折叠成长条卡片需要点击阅读原文才能看到完整内容。这个问题在手机上看尤其明显——我一开始把所有日志都塞进通知里结果手机上看到的是半截消息还得点进企业微信App才能展开这就完全违背了快速扫一眼的初衷。解决方案是做内容裁剪设置500字符上限超出部分截断并附上详见日志链接。我现在所有通知都遵循这个规则手机上一次扫完不再需要二次点击。5.2 坑二重试导致重复通知我最开始的重试逻辑是只要网络超时就重试结果遇到一次网络抖动同一条消息发了三遍微信里连续弹了三次通知。后来我把重试范围严格限制在网络异常和返回码不是0这两种情况并且加入了一个去重ID机制import hashlib _sent_set set() def send_wechat_notify(content: str, msg_type: str text, webhook_url: str WEBHOOK_URL, max_retries: int 3, dedup_key: str ) - bool: # 生成去重ID同一任务同一阶段只发一次 if dedup_key: hash_key hashlib.md5(dedup_key.encode()).hexdigest() if hash_key in _sent_set: return True _sent_set.add(hash_key) # ... 原有发送逻辑注意这个去重集合要设一个上限否则跑久了内存越来越大。我用的是collections.deque(maxlen10000)加集合的方式来控制。5.3 坑三Agent内部异常被回调静默吞掉这是最危险的一个坑。LangGraph这类框架里有on_chain_error回调我最初以为只要Agent内部抛异常这个回调就一定会触发。后来发现并非如此——有些工具类的异常被框架吞掉后转成了特殊返回值不会走到on_chain_error。也就是说Agent看起来是正常完成了实际上产出的结果是错误的。这导致我第一次上线时收到任务成功通知但客户那边反馈结果不对。排查了半天才发现Agent在处理中抛错被一个工具节点拦截后转成了{error: ...}输出然后整体流程走了on_agent_finish。所以现在我的逻辑是在on_agent_finish里也要检查输出内容是否包含明显的错误标记如果识别到错误关键词就会主动改成失败通知。def on_agent_finish(self, finish, **kwargs): output_text if isinstance(finish.return_values, dict): output_text str(finish.return_values.get(output, )) else: output_text str(finish.return_values) if error in output_text.lower() or exception in output_text.lower(): # 转成失败通知 send_agent_notify(...) return # 正常成功逻辑这个经验说明了一个通用道理通知服务不是只报流程状态还要关心内容健康度。流程正常结束不代表结果正确。5.4 坑四通知本身比Agent还慢我最初在Agent的同步链路里直接调用通知函数实际测下来有时候一次通知要花两三秒尤其是企业微信那边偶尔会慢。对于Agent这种需要频繁调工具的任务流每步都等两三秒通知累积起来很影响性能。后来我把通知改成了异步发送模式。如果用FastAPI直接asyncio.create_task(send_agent_notify(...))如果用普通Python可以起一个消息队列线程通知函数只负责把消息扔进队列真正发送的工作由后台消费者来做。这样通知的耗时对Agent主流程的干扰基本可以忽略。import threading import queue _notify_queue queue.Queue() def async_send_agent_notify(title, content): 异步通知: 不阻塞Agent主流程 _notify_queue.put((title, content)) def notify_worker(): while True: title, content _notify_queue.get() try: send_agent_notify(content) except Exception as e: print(f[notify] 后台通知失败: {e}) finally: _notify_queue.task_done() # 启动一个后台线程 threading.Thread(targetnotify_worker, daemonTrue).start()当然要控制队列长度队列积压了说明发送通道出问题了这时候可以考虑直接把降级消息打日志。5.5 完整排查链路一次通知丢失的定位过程最后分享一次真实的排障过程给大家一个排查思路参考。现象是我有一批Agent任务大概跑到第37个左右就不再发完成通知了。我没在第一时间改任何代码而是按下面顺序一步步定位先看Agent是否真的执行完查Agent日志确认任务跑到第37个时就已经异常退出不是通知的问题。再看失败原因里有没有通知失败的记录发现异常堆栈里带了requests.exceptions.ConnectionError说明Agent是网络出境失败不是通知服务本身的代码问题。如果Agent没挂但通知丢了那就要去企业微信后台查看机器人消息发送记录看是不是每分钟20条限制了或者是Webhook地址有没有被多人改掉。如果发送有返回但客户端没收到去查看手机端企业微信的消息免打扰设置群机器人被静音了通知照样发得出去但你收不到。这条链路走一遍基本能把通知丢失问题收敛到Agent侧、网关侧、还是客户端设置上。我这台服务器是自己搭的排查到第二步就发现是任务本身断网了通知服务延后触发根本没机会执行——所以以后我在启动Agent前还会加一个网络连通性自检先发一条Agent即将启动的测试通知如果这条都发不出去就直接在启动阶段拦截不要等到跑挂了再被动发现。6. 还能怎么扩展从单Agent通知到任务中心这套微信推送服务跑通以后我很快发现它不只是给单个Agent跑完通知一声用的——稍微抽象一层它完全可以变成一个通用的通知中枢把Agent相关的所有事件都纳入管理。我目前已经接进去的扩展玩法包括定时任务完成提醒每个月1号的月报Agent、每周五的周报自动生成Agent跑完都会推送确认消息。以前总担心定时任务到底有没有跑现在到点看手机就知道。多Agent协作时的状态广播当主Agent把任务派发给多个子Agent时主完成、子挂掉、重试成功这些关键节点都推。群里的消息会形成一个事件流整个多Agent的协作过程变成可追踪的。Agent配额消耗提醒我用的模型API不是无限量的每个Agent跑完会把本轮消耗的Token数带上累计到某个阈值时主动通知。这个对控制成本非常有用。与自定义监控脚本联动不只是Agent通知服务器磁盘告警、数据管线异常这些我都顺手接进来了。一个Webhook变量搞定所有有异常需要人介入的场景。另外如果Agent已经是部署在Kubernetes或Docker里的服务可以考虑把健康检查、重启事件也接进来。这些事看着小但在生产环境里它们决定了Agent是一个能跑的东西还是一个真正靠谱的系统。关于后续要不要再做一个任务中心页面我的想法是如果在自己的服务器上跑可以给每个Agent任务生成一条链接点进来看历史日志和运行指标也就是把通知从单向触达升级成双向查阅。但现在我的实际使用中通知本身已经覆盖了95%以上的场景更高端的东西等真有需求再动手。从最开始跑完没人告诉我的痛点到最终跑通一个轻量、可靠、有兜底的微信推送服务整个过程其实没有用到任何特别高深的技术。核心思路很简单先在官方渠道里选最省事的通路再把消息格式做规范化最后通过在关键节点埋入可复用的通知调用让Agent自己把状态报给你。这条链路现在稳定跑了几个月每天几十条Agent状态消息准时到手机我基本已经不追着终端看了。如果你也正在被Agent跑起来之后只能干等这件事折磨照着我这个方案抄一遍一个晚上足够跑起来。
AI Agent跑完任务如何自动通知？微信推送方案全解析
1. Agent跑完没人告诉你这类通知需求到底要解决什么问题如果你已经开始把AI Agent用在真实任务里——不管是让LangGraph跑一个多轮推理的爬虫还是用Claude写代码、调脚本、批量处理数据——你大概率也遇到过这种场景任务丢进去之后它自己跑起来了但你可能是在开会、在吃饭、甚至已经躺下睡觉了。等你想起来切回终端看一眼发现任务其实在十几分钟前就结束了全程没人告诉你。我自己最崩溃的一次是在半夜跑一个批量处理任务Agent需要逐个读取上百个文件并生成结构化摘要按当时的进度估算得跑一个多小时。我把命令一挂就去睡了。第二天早上醒来一看终端里登录会话早就断了进程也在凌晨就挂掉了——因为我没处理网络断连和会话超时的问题。但关键问题不在进程崩溃而在于我没有任何渠道及时知道它跑完了或者跑失败了。那这个Agent做得再智能对于需要等待、然后拿结果的人来说体验依然是断裂的。这就是AI Agent跑完任务怎么通知你这个看起来很不起眼的问题实际上的价值所在。它不只是省点心的问题而是把Agent从一个运行在黑盒里的程序变成了真正能被人调度的生产工具。我当时就想得很明白我要的不是什么花哨的看板、复杂的监控系统我要的只是一个足够可靠、延迟低、并且我手机里一定会第一时间看到的东西——微信推送服务。微信推送的好处不需要我多讲手机必装、推送到达率高、消息可回溯、不需要额外装App。于是我就花了一个晚上写了这个推送服务并且在接下来的几个星期里把Agent项目的通知链路全部接了上去。这篇文章就把我这次的完整思路、代码实现、以及踩过的坑全部写出来。适合的人群有两类一是已经在跑Agent、但还在用人肉盯梢方式等结果的人二是打算自己搭一套通知服务、但不知道从哪下手的人。不管你是用LangChain、LangGraph、AutoGPT还是自研Agent这套思路都能直接用。2. 通知渠道选型微信推送为什么打赢了邮件和钉钉先说结论选微信推送不是因为它技术最先进而是因为它是用户门槛和实现成本之间平衡得最好的一条路。我刚开始也没直接选微信先花了一点时间把所有候选方案列了一遍逐个验证最后才定下来。2.1 常见的Agent通知方案对比我给Agent接通知的时候周围小伙伴用的大概是这几种路子通知渠道优点缺点适合场景邮件SMTP协议标准、代码成熟、可以发附件延迟不稳定、容易被丢进垃圾箱、手机不一定会立刻弹通知长时间批处理、需要附件的正式报告钉钉/企微群机器人到达快、免费、配置简单、支持Markdown需要企业组织、个人使用者需要先建个群团队协作场景短信到达率高、绝对可靠每条都要钱、签名审核麻烦、有频率限制高优先级告警微信服务号/企业微信/小程序订阅消息手机必装、触达率高、免费需要有对应的开发配置或第三方封装个人Agent和轻量通知自建App Push可定制性最强要开发App、运维成本高几乎没有需求看了一圈就能发现对个人开发者来说最务实的其实就是两个方向一是用钉钉或企业微信群机器人二是走微信生态的推送服务。我当时两个方向都试了最后留下了企业微信群机器人这条路作为主力——因为它既能发到微信客户端收到通知又是官方提供、代码实现极其简单的方案。2.2 关于个人微信能不能直接推的边界问题这里必须提醒一下很多人第一反应是我直接用个人微信给自己发消息不就好了但个人微信官方并没有开放API给第三方直接调用。市面上那些个人微信机器人大多基于非官方协议本质上属于逆向和外挂轻则随时失效重则账号受限。这条路我坚决不碰原因很简单——Agent通知是生产链路里最后一段它如果挂了前面做得再好都白搭我不能把最后一段建立在随时可能被封的沙子上。所以我说的微信推送服务实际指的是两条官方合规路径企业微信群机器人Webhook注册一个企业微信个人就能注册建一个群添加群机器人得到一个Webhook地址。向这个地址POST一段JSON消息就会出现在企业微信里你在微信里也能收到提醒。免费、不需要审核、当天就能搞定。第三方推送平台Server酱、PushPlus、WxPusher等这类平台的模式是——用微信扫码绑定你的微信身份平台给你一个SendKey或Token你的代码把消息POST到平台API平台再通过自己的企业微信或微信模板消息推给你。优点是连企业微信都不用自己配缺点是信任链在第三方手里且部分平台有每日条数限制。我个人的取舍是主力用企业微信群机器人备用一个第三方平台做降级容错。这样既有充分的控制权又不会出现单点故障。下面代码部分我会把两条路径都写出来。2.3 最终架构的形态整个通知服务我没打算引入什么重量级框架最后定下来的架构非常简单AI Agent执行任务 ↓ (任务开始/结束/失败时调用) 通知客户端模块HTTP请求 ↓ (POST JSON) 企业微信群机器人Webhook / 第三方推送平台 ↓ (微信客户端内弹出消息) 你本人这个架构的好处是所有组件都是标准HTTP接口不依赖内网互通不依赖Agent运行在哪里——本地笔记本、云服务器、Docker容器里都一样能推。而且它天然跨语言今天你用Python写Agent明天换成TypeScript通知这一段逻辑完全不受影响。3. 推送代码落地一个函数跑通通知主链路聊完选型和设计直接上代码。这里我不搞那种几十层的抽象架构先给一个能立刻跑起来的最小实现再逐步加功能。3.1 企业微信群机器人的最简实现先创建一个群机器人在企业微信客户端里找到工作台-添加成员-添加机器人选一个群加入拿到一个形如https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxxxxxx-xxxxxxxx-xxxxxxxx的Webhook地址配置就完成了。然后是一个极简的Python函数import requests import json import time WEBHOOK_URL https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的key def send_wechat_notify(content: str, msg_type: str text, webhook_url: str WEBHOOK_URL, max_retries: int 3) - bool: 向企业微信群机器人发送消息 :param content: 消息文本内容 :param msg_type: text 或 markdown :param webhook_url: 机器人webhook地址可覆盖 :param max_retries: 失败重试次数 :return: 是否发送成功 payload { msgtype: msg_type, msg_type: {content: content} if msg_type text else {content: content} } headers {Content-Type: application/json} for attempt in range(1, max_retries 1): try: resp requests.post(webhook_url, jsonpayload, headersheaders, timeout10) result resp.json() # 企业微信返回码: 0 表示成功 if result.get(errcode) 0: return True # 频率限制特殊处理 if result.get(errcode) 45009: # 每60秒最多发20条, 这里睡一会儿再重试 time.sleep(60) continue print(f[notify] 发送失败: {result}) return False except requests.exceptions.RequestException as e: print(f[notify] 网络异常(第{attempt}次): {e}) if attempt max_retries: time.sleep(2 * attempt) else: return False return False这个函数干了几件容易被忽略的事设置了超时时间10秒。如果没有超时当网络异常时requests.post会一直挂着导致Agent主流程卡在通知这一步。通知是辅助链路绝不能让它的故障拖累主任务。处理了频率限制。企业微信群机器人的限制是每分钟最多20条超了会返回errcode为45009如果代码里不处理某些批量场景下会莫名其妙丢通知。做了简单的重试。网络抖动在现在这种环境里太常见了第一次请求失败就放弃太重了。我按指数退避做三次重试实测下来绝大多数临时故障都能自动恢复。3.2 消息内容格式的设计要点很早的时候我踩过一个坑消息里什么状态都写最后堆出来一篇又臭又长的报告自己根本没心思看第二遍。后来我把每类消息的格式做了标准化现在形成了这几个模板任务开始通知{ msgtype: text, text: { content: [Agent] 任务已启动\n项目: 数据清洗\n任务ID: 20250101_001\n开始时间: 2025-01-01 14:30:22\n预计耗时: 30分钟 } }任务成功通知{ msgtype: markdown, markdown: { content: **[Agent] 任务已完成** \n\n 项目: 数据清洗\n 任务ID: 20250101_001\n 耗时: 28分15秒\n 输出: 处理文件128个成功率 100%\n\n[日志详情](http://你的流水线地址/tasks/20250101_001) } }任务失败通知{ msgtype: markdown, markdown: { content: **[Agent] 任务执行失败** \n\n 项目: 数据清洗\n 任务ID: 20250101_001\n 失败阶段: 文件解析\n 失败原因: 第57个文件格式非法\n 耗时: 12分03秒 } }有几个细节我想特别说明一下Markdown模式只对企业微信的机器人消息生效而且支持的是它的特定子集标题、加粗、引用、链接不是完整Github Flavored Markdown。图片、表格这类元素在不同客户端表现不稳定我一般只用加粗、引用和链接。每条消息都要有项目和任务ID。当多个Agent并行跑的时候没有这两个字段你根本分不清是谁发的消息。任务ID需要能在你的任务系统里反查到日志这是排查问题的基础。失败通知里必须写明失败阶段和失败原因。很多人只发任务失败四个字收到消息后还得去翻日志那推送的及时性就白费了。我的习惯是——失败原因尽量让代码自己捕获并拼进去这样90%的情况你不需要打开电脑就知道该从哪里下手。3.3 用第三方平台做备用通道企业微信这条路本身已经相当稳了但为了做到双保险我还加了一路备用通道——用Server酱的API。这类平台的调用更简单本质上就是一个GET请求发送完还能在手机上看到推送历史记录。核心代码import requests SERVER_CHAN_KEY 你的SendKey def send_serverchan(title: str, content: str) - bool: 备用通知通道: 通过Server酱推送到微信 url fhttps://sctapi.ftqq.com/{SERVER_CHAN_KEY}.send payload { title: title[:32], # 标题限制32字符 desp: content, } try: resp requests.post(url, datapayload, timeout10) result resp.json() return result.get(code) 0 except Exception as e: print(f[serverchan] 通知失败: {e}) return False在正式的发送入口里我的发送顺序是先走企业微信失败再走Server酱。封装起来就是def send_agent_notify(title: str, content: str) - None: ok send_wechat_notify(content) if not ok: send_serverchan(title, content)注意Server酱的标题有32字符限制直接把完整内容怼进去会把接口打挂。所以我在封装层做了截断处理——长字段一律放正文标题只放项目名和状态。4. AI Agent接入推送的三种姿势服务端写好了接下来就是怎么让它和Agent接上。我在实际开发中总结出三种方式分别对应不同场景你们可以根据自己Agent的架构选。4.1 姿势一包装器方式最简单几乎万能如果你的Agent是拿某个函数或某条命令启动的比如result run_agent(task给这些数据生成摘要)那你不用改Agent内部代码只用包装器包一层import traceback from datetime import datetime def notify_agent_run(func): 装饰器: 自动给Agent任务加开始/结束/异常通知 def wrapper(task, *args, **kwargs): start_time datetime.now() send_agent_notify( f[{task.get(project, Agent)}] 任务已启动, f任务ID: {task.get(task_id, N/A)}\n开始时间: {start_time:%Y-%m-%d %H:%M:%S} ) try: result func(task, *args, **kwargs) elapsed (datetime.now() - start_time).total_seconds() send_agent_notify( f[{task.get(project, Agent)}] 任务已完成, f任务ID: {task.get(task_id, N/A)}\n耗时: {elapsed/60:.1f}分钟\n结果: {str(result)[:500]} ) return result except Exception as e: elapsed (datetime.now() - start_time).total_seconds() error_info traceback.format_exc() send_agent_notify( f[{task.get(project, Agent)}] 任务失败, f任务ID: {task.get(task_id, N/A)}\n耗时: {elapsed/60:.1f}分钟\n异常: {e}\n堆栈尾部: {error_info[-500:]} ) raise return wrapper # 使用 notify_agent_run def run_agent(task): # Agent原有逻辑 pass这种包装器的方式有两点好处改成装饰器后只需要在原来的函数定义上加一行notify_agent_run完全不需要动Agent内部的推理逻辑。失败时发完通知再把异常抛出去保证Agent原有的错误处理链路不受影响外面该重试的重试、该告警的告警。4.2 姿势二框架回调方式适合LangChain等如果你用的是LangChain或LangGraph这类有回调机制的框架更优雅的做法是直接实现它的callback接口。比如LangChain的CallbackHandlerfrom langchain_core.callbacks import BaseCallbackHandler class WeChatNotifyHandler(BaseCallbackHandler): Agent关键节点回调通知 def __init__(self, task_id: str, project: str): self.task_id task_id self.project project self.start_time None def on_chain_start(self, serialized, inputs, **kwargs): if self.start_time is None: self.start_time datetime.now() send_agent_notify( f[{self.project}] Agent开始执行, f任务ID: {self.task_id}\n时间: {self.start_time:%Y-%m-%d %H:%M:%S} ) def on_agent_finish(self, finish, **kwargs): elapsed (datetime.now() - self.start_time).total_seconds() output_text finish.return_values.get(output, ) if isinstance(finish.return_values, dict) else str(finish.return_values) send_agent_notify( f[{self.project}] Agent执行完成, f任务ID: {self.task_id}\n耗时: {elapsed/60:.1f}分钟\n输出摘要: {str(output_text)[:300]} ) def on_chain_error(self, error, **kwargs): elapsed (datetime.now() - self.start_time).total_seconds() send_agent_notify( f[{self.project}] Agent执行失败, f任务ID: {self.task_id}\n耗时: {elapsed/60:.1f}分钟\n错误: {str(error)[:300]} )然后把这个handler挂到你的Agent上handler WeChatNotifyHandler(task_idtask_001, project数据摘要) agent_executor create_agent_executor(...) agent_executor.run(处理这批文件, callbacks[handler])这种方式最优雅的地方在于它能同时捕获LangChain内部工具调用的完成事件。比如Agent中途调用了某个搜索工具或代码执行器你可以在on_tool_end里追加通知真正做到Agent每一步干了什么你都心里有数。但也要小心不要在回调里做太多同步操作——通知发送控制在几十毫秒内没问题如果逻辑复杂了建议改成异步任务避免拖慢Agent的推理主流程。4.3 姿势三HTTP回调方式适合Agent跑在远程服务器很多生产级Agent并不是和你在同一台设备上跑的你本地开发、代码推到服务器上、服务器上的Agent异步执行任务。这时包装器和回调都不太适用了——因为代码执行完你和服务器之间的连接可能已经断开了。我的做法是在通知服务里加一个极小的HTTP入口让Agent执行完自己往这个入口打一个请求from fastapi import FastAPI, Request from pydantic import BaseModel app FastAPI() class NotifyPayload(BaseModel): project: str task_id: str status: str # success / failed / started message: str elapsed_seconds: float 0 app.post(/agent/notify) async def agent_notify(payload: NotifyPayload): 远程Agent回调入口 if payload.status success: title f[{payload.project}] 任务成功 content f任务ID: {payload.task_id}\n耗时: {payload.elapsed_seconds/60:.1f}分钟\n{payload.message} elif payload.status failed: title f[{payload.project}] 任务失败 content f任务ID: {payload.task_id}\n耗时: {payload.elapsed_seconds/60:.1f}分钟\n错误: {payload.message} else: title f[{payload.project}] 任务启动 content f任务ID: {payload.task_id}\n{payload.message} send_agent_notify(title, content) return {code: 0}Agent那边只需要在结束时发一次请求import requests requests.post(http://你的服务器:8000/agent/notify, json{ project: 数据清洗, task_id: 20250101_001, status: success, message: 处理文件128个成功率100%, elapsed_seconds: 1695 })这个HTTP回调方案有个额外的安全要求是必须做的不要把这个端点裸放在公网上。我的做法是加一个简单的Token鉴权Agent请求时带上请求头Authorization: Bearer 你定的tokenFastAPI这边做一个依赖注入校验否则任何人都能往你的微信里塞垃圾消息。5. 上线实测踩过的坑和对应的修复手段前面都是理想状态下的设计。真正把这套通知服务接到Agent上跑了一个多月之后我才知道发通知这件事本身充满了让程序员头疼的细节。这里写几个最典型的坑和对应的解决经验。5.1 坑一长消息触发企业微信的需要阅读全文企业微信机器人发长文本时微信端会折叠成长条卡片需要点击阅读原文才能看到完整内容。这个问题在手机上看尤其明显——我一开始把所有日志都塞进通知里结果手机上看到的是半截消息还得点进企业微信App才能展开这就完全违背了快速扫一眼的初衷。解决方案是做内容裁剪设置500字符上限超出部分截断并附上详见日志链接。我现在所有通知都遵循这个规则手机上一次扫完不再需要二次点击。5.2 坑二重试导致重复通知我最开始的重试逻辑是只要网络超时就重试结果遇到一次网络抖动同一条消息发了三遍微信里连续弹了三次通知。后来我把重试范围严格限制在网络异常和返回码不是0这两种情况并且加入了一个去重ID机制import hashlib _sent_set set() def send_wechat_notify(content: str, msg_type: str text, webhook_url: str WEBHOOK_URL, max_retries: int 3, dedup_key: str ) - bool: # 生成去重ID同一任务同一阶段只发一次 if dedup_key: hash_key hashlib.md5(dedup_key.encode()).hexdigest() if hash_key in _sent_set: return True _sent_set.add(hash_key) # ... 原有发送逻辑注意这个去重集合要设一个上限否则跑久了内存越来越大。我用的是collections.deque(maxlen10000)加集合的方式来控制。5.3 坑三Agent内部异常被回调静默吞掉这是最危险的一个坑。LangGraph这类框架里有on_chain_error回调我最初以为只要Agent内部抛异常这个回调就一定会触发。后来发现并非如此——有些工具类的异常被框架吞掉后转成了特殊返回值不会走到on_chain_error。也就是说Agent看起来是正常完成了实际上产出的结果是错误的。这导致我第一次上线时收到任务成功通知但客户那边反馈结果不对。排查了半天才发现Agent在处理中抛错被一个工具节点拦截后转成了{error: ...}输出然后整体流程走了on_agent_finish。所以现在我的逻辑是在on_agent_finish里也要检查输出内容是否包含明显的错误标记如果识别到错误关键词就会主动改成失败通知。def on_agent_finish(self, finish, **kwargs): output_text if isinstance(finish.return_values, dict): output_text str(finish.return_values.get(output, )) else: output_text str(finish.return_values) if error in output_text.lower() or exception in output_text.lower(): # 转成失败通知 send_agent_notify(...) return # 正常成功逻辑这个经验说明了一个通用道理通知服务不是只报流程状态还要关心内容健康度。流程正常结束不代表结果正确。5.4 坑四通知本身比Agent还慢我最初在Agent的同步链路里直接调用通知函数实际测下来有时候一次通知要花两三秒尤其是企业微信那边偶尔会慢。对于Agent这种需要频繁调工具的任务流每步都等两三秒通知累积起来很影响性能。后来我把通知改成了异步发送模式。如果用FastAPI直接asyncio.create_task(send_agent_notify(...))如果用普通Python可以起一个消息队列线程通知函数只负责把消息扔进队列真正发送的工作由后台消费者来做。这样通知的耗时对Agent主流程的干扰基本可以忽略。import threading import queue _notify_queue queue.Queue() def async_send_agent_notify(title, content): 异步通知: 不阻塞Agent主流程 _notify_queue.put((title, content)) def notify_worker(): while True: title, content _notify_queue.get() try: send_agent_notify(content) except Exception as e: print(f[notify] 后台通知失败: {e}) finally: _notify_queue.task_done() # 启动一个后台线程 threading.Thread(targetnotify_worker, daemonTrue).start()当然要控制队列长度队列积压了说明发送通道出问题了这时候可以考虑直接把降级消息打日志。5.5 完整排查链路一次通知丢失的定位过程最后分享一次真实的排障过程给大家一个排查思路参考。现象是我有一批Agent任务大概跑到第37个左右就不再发完成通知了。我没在第一时间改任何代码而是按下面顺序一步步定位先看Agent是否真的执行完查Agent日志确认任务跑到第37个时就已经异常退出不是通知的问题。再看失败原因里有没有通知失败的记录发现异常堆栈里带了requests.exceptions.ConnectionError说明Agent是网络出境失败不是通知服务本身的代码问题。如果Agent没挂但通知丢了那就要去企业微信后台查看机器人消息发送记录看是不是每分钟20条限制了或者是Webhook地址有没有被多人改掉。如果发送有返回但客户端没收到去查看手机端企业微信的消息免打扰设置群机器人被静音了通知照样发得出去但你收不到。这条链路走一遍基本能把通知丢失问题收敛到Agent侧、网关侧、还是客户端设置上。我这台服务器是自己搭的排查到第二步就发现是任务本身断网了通知服务延后触发根本没机会执行——所以以后我在启动Agent前还会加一个网络连通性自检先发一条Agent即将启动的测试通知如果这条都发不出去就直接在启动阶段拦截不要等到跑挂了再被动发现。6. 还能怎么扩展从单Agent通知到任务中心这套微信推送服务跑通以后我很快发现它不只是给单个Agent跑完通知一声用的——稍微抽象一层它完全可以变成一个通用的通知中枢把Agent相关的所有事件都纳入管理。我目前已经接进去的扩展玩法包括定时任务完成提醒每个月1号的月报Agent、每周五的周报自动生成Agent跑完都会推送确认消息。以前总担心定时任务到底有没有跑现在到点看手机就知道。多Agent协作时的状态广播当主Agent把任务派发给多个子Agent时主完成、子挂掉、重试成功这些关键节点都推。群里的消息会形成一个事件流整个多Agent的协作过程变成可追踪的。Agent配额消耗提醒我用的模型API不是无限量的每个Agent跑完会把本轮消耗的Token数带上累计到某个阈值时主动通知。这个对控制成本非常有用。与自定义监控脚本联动不只是Agent通知服务器磁盘告警、数据管线异常这些我都顺手接进来了。一个Webhook变量搞定所有有异常需要人介入的场景。另外如果Agent已经是部署在Kubernetes或Docker里的服务可以考虑把健康检查、重启事件也接进来。这些事看着小但在生产环境里它们决定了Agent是一个能跑的东西还是一个真正靠谱的系统。关于后续要不要再做一个任务中心页面我的想法是如果在自己的服务器上跑可以给每个Agent任务生成一条链接点进来看历史日志和运行指标也就是把通知从单向触达升级成双向查阅。但现在我的实际使用中通知本身已经覆盖了95%以上的场景更高端的东西等真有需求再动手。从最开始跑完没人告诉我的痛点到最终跑通一个轻量、可靠、有兜底的微信推送服务整个过程其实没有用到任何特别高深的技术。核心思路很简单先在官方渠道里选最省事的通路再把消息格式做规范化最后通过在关键节点埋入可复用的通知调用让Agent自己把状态报给你。这条链路现在稳定跑了几个月每天几十条Agent状态消息准时到手机我基本已经不追着终端看了。如果你也正在被Agent跑起来之后只能干等这件事折磨照着我这个方案抄一遍一个晚上足够跑起来。