看到“Agent-Reach”这个标题先别急着往“又一个人工智能框架”上靠。它其实是我近期整理的一个内部项目名字听着挺宏大核心就一件事让各种 Agent包括 AI 智能体、自动化脚本、运营机器人在需要触达用户或下游系统时不丢消息、不重复发、还能在通道挂掉时自动换路。简单说它是个触达编排层。这类项目在团队里常常被忽略但真到线上出问题的时候最让人头疼的往往不是业务代码而是“用户该收到消息却没收到”“一次活动通知被重复发了三遍”“短信网关超时后面所有任务全部卡死”。Agent-Reach 想解决的就是这一类问题把不可靠的触达动作变成一个可重试、可观测、可审计、可降级的过程。如果你正在做消息推送平台、自动化运营工具、AI Agent 的工具调用链或者只是想把散落在业务代码里的通知逻辑收拢起来这篇内容应该能给你一个可以直接参考的落地思路。我会把设计思路、核心模型、关键代码和排查方法全部拆开讲不绕弯子。1. 项目从哪来Agent-Reach 到底在解决什么事1.1 名字里的两个词Agent 和 Reach“Agent”在这里不单指 AI Agent而是泛指一切“需要主动发起动作的执行体”。运营系统里的定时任务、审批流里的提醒服务、客服机器人里的对话回复、AI Agent 在接到指令后调用的外部工具本质上都是 Agent 的一种形态。“Reach”也不只是“发出去”它强调的是“到达”。一个消息从产生到真正被用户看到中间隔着队列、适配器、第三方通道、重试、回执确认好几个环节。大多数系统的问题出在“我认为发出去了”和“用户确实收到了”之间的差距上。所以 Agent-Reach 这个名字的意思是把触达当成一条完整的链路来管而不是只做一个 send 函数。它负责的任务不是“怎么构造消息内容”而是“怎么可靠地把消息送到目标手上”。1.2 它解决的三类典型痛我在实际项目中见过大量类似场景痛点基本可以归成三类。第一类是“发了等于没发”。比如短信网关偶发超时业务代码只 catch 了异常但没有重试机制又比如回调接口返回 200但业务代码里面处理失败用户始终没收到提醒。这类问题最难查因为日志显示“成功”用户却说“没收到”。第二类是“没发等于重复发”。任务在队列里被消费者拉取后进程 crash消息重新入队业务侧没有做幂等用户就会收到两条一样的短信。看似小概率但在大促、活动通知这类场景里小概率乘以百万级用户量就会变成线上事故。第三类是“全堵死”。触达动作散落在各业务模块没有统一队列没有背压控制。某个通道一慢调用方线程被占用最后整个服务跟着卡死。我见过一次邮件服务响应慢直接把订单服务拖崩的事故原因就是订单完成通知里同步发送了邮件。Agent-Reach 把这些场景集中抽象成“触达任务”由统一调度层来管理通道、重试、状态和幂等业务方只需要描述“我要触达谁、用什么模板、希望优先走哪个通道”。1.3 适合谁来用如果你所在的团队已经有专门的消息平台或者只做站内信这种单通道场景Agent-Reach 这种编排层会显得有点重。但如果你的系统需要同时接入站内信、邮件、短信、webhook并且业务方经常提出“不同用户走不同通道”“失败后按顺序降级”“某些任务必须保证最终送达”那这套思路就非常合适。同样适合的还有正在做 AI Agent 应用的开发者。AI Agent 在调用外部工具、给用户发送异步结果、触发下游工作流的时候本质上也是一次触达行为。把触达逻辑从 Agent 代码里抽出来交给 Agent-Reach 这样的编排层能让 Agent 本身更轻也更容易排查“为什么工具调用后没有效果”这类问题。2. 整体设计拆解为什么把触达做成“编排层”2.1 别把发送逻辑散落在业务代码里很多项目的触达逻辑是散落的。订单模块发订单通知用户模块发验证码运营模块发活动推送每个模块自己写一套调用第三方接口的逻辑自己处理重试。这种做法在业务量小的时候没问题但一旦通道种类变多、消息量变大就会暴露几个问题。业务代码会越来越脏。一个下单服务里既要管库存、扣款、发票还要管短信模板、通道切换、失败重试任何一行通道配置修改都会影响业务发布。其次是通道故障难以隔离。短信服务商出问题理论上应该只影响短信这个通道但因为代码散落可能导致所有包含短信发送逻辑的业务模块同时报错。Agent-Reach 的做法是把触达从业务里抽出来做成独立的编排层。业务只负责提交一个“意图”比如“通知用户 A 订单支付成功”至于走短信还是走邮件、什么时候发、失败重试几次、重试间隔多久全部由编排层决定。你可以把它理解成外卖平台。用户下单后平台负责调度骑手、规划路线、处理异常而不是让每个餐厅自己招骑手送餐。这样餐厅只需要做好餐配送的事交给统一平台。2.2 三级触达模型策略、执行、通道Agent-Reach 内部把触达拆成三层策略层、执行层、通道层。策略层只负责描述“触达意图”。包括目标用户、消息模板、要尝试的通道顺序、最大重试次数、过期时间、幂等键。策略层不关心短信接口长什么样不关心邮件模板怎么渲染只负责表达需求。执行层负责把策略变成执行计划。它会去消费队列里的触达任务根据通道顺序逐个尝试记录每次尝试的返回结果更新任务状态决定下一步是继续换通道、重试还是进入死信队列。执行层是整个系统的中枢。通道层是最具体的部分。每个通道对应一个适配器比如 SmsAdapter、EmailAdapter、WebhookAdapter。适配器做的事情非常简单接收一个触达任务和该通道的配置调用底层服务返回成功或失败。通道层不关心业务语义也不关心重试策略。这三层的好处是职责清晰互相之间只通过任务对象通信。以后要新增一个通道只需要写一个适配器注册进去策略层和执行层一行代码都不用改。要调整重试策略也只影响执行层。2.3 技术选型与部署形态我最早的版本没有引入任何新组件只是在一个业务服务里加了一张表、一个定时任务每五秒扫描一次需要发送的任务。后来消息量大了定时扫描的实时性不够才引入了队列。最终的技术选型很朴素PostgreSQL 存状态和审计日志Redis 做队列Python 写 Worker。Redis Stream 负责承载待处理任务PostgreSQL 是唯一的事实来源。为什么选 Redis Stream 而不是直接用 Redis List因为 Stream 天然支持消费组、消息确认、消息历史对“至少一次”的消费语义支持得更好。这里有一个很关键的设计原则不把 Redis 当成最终存储。所有触达任务先写入 PostgreSQL状态也是以 PostgreSQL 为准Redis 里只是“待处理游标”的缓存。就算 Redis 里的数据全部丢失重启后仍然可以从扫描表中恢复出所有未完成任务。这也是 Agent-Reach 能承受部分组件故障的底气。组件我这里用的替代方案选型理由状态存储PostgreSQLMySQL、DynamoDB需要事务、唯一索引、审计查询队列Redis StreamRabbitMQ、Kafka轻量、运维简单、团队熟悉WorkerPython asyncioGo、Node.js迭代快、适配器开发方便任务格式JSONProtobuf调试方便兼容性好如果团队已经有一套现成的 RabbitMQ 或者 Kafka完全可以把 Redis 换成它们多出的收益主要是消息持久化和回溯能力。核心模型不变变的只是队列的接入方式。3. 核心细节任务模型、状态机与通道降级3.1 统一的任务描述长什么样在 Agent-Reach 里所有触达意图都使用同一套 JSON 结构描述。下面这个例子是一个“订单支付成功”通知任务。{ task_id: c82a4f7e-3b2a-4f8c-9d1a-8f0a76b3e2d1, dedup_key: user_10001:order_8823:pay_success, recipient: { user_id: 10001, channels: { in_app: { conversation_id: 10001 }, email: u_10001example.com, sms: 8613800000000 } }, template_id: order_pay_success, channels_priority: [in_app, email, sms], strategy: { max_attempts: 3, timeout_seconds: 10, retry_backoff: exponential, schedule_start: 2025-06-01T10:00:00Z, expire_at: 2025-06-01T23:00:00Z } }这里最值得关注的是dedup_key和channels_priority。dedup_key是整个幂等机制的核心。它由“接收方标识 业务标识 业务动作”组成而不是简单用 task_id因为 task_id 每次生成都不同无法屏蔽业务侧的重复提交。比如用户下单成功后服务重启触发了一次补偿逻辑重新提交了同一个订单的通知如果没有 dedup_key用户就会收到两条一样的消息。channels_priority表达的是降级顺序。站内信成本最低优先尝试如果用户长期不登录站内信送达率不高可以降级到邮件邮件也不是实时通道紧急消息再降级到短信。这个顺序是策略层的选择执行层只需要按顺序执行。3.2 状态机与审计触达任务不是“发完就完”很多团队在实现发送模块的时候只在数据库里存了“发送状态”要么是成功要么是失败。但真实触达过程远比这两种状态要复杂。Agent-Reach 的每个任务都有一套明确的状态机。状态含义可流转到CREATED任务已创建尚未进入队列PENDING / FAILEDPENDING已进入队列等待 Worker 消费SENDING / FAILEDSENDING正在向某个通道投递SUCCEEDED / RETRYING / FAILEDRETRYING上一通道失败等待重试或换通道SENDING / FAILED / DEADSUCCEEDED已有通道确认触达成功终态DEAD所有策略都用完最终失败终态可人工重放为什么状态要设计得这么细因为只有状态细才能回答“这个任务到底卡在哪”。线上排查用户没收到消息的时候第一件事就是查任务状态。如果任务停在 SENDING 超过两分钟说明 Worker 可能一直在等待下游响应或者进程已经崩溃但队列里的消息没有被重新消费。如果任务停在 RETRYING说明通道在持续报错需要去检查第三方通道状态。如果任务进入 DEAD说明配置的策略已经耗尽需要人工介入。每一条状态变更都会写入审计日志。这个日志在形式上和业务日志很像但它只记录触达任务自己的生命周期谁在什么时间把它从什么状态变到什么状态当时的通道返回是什么。没有这份审计事后复盘“为什么多发了消息”会非常痛苦因为你只能靠猜。3.3 通道优先级与降级选择触达通道没有绝对的好坏只有适不适合当前场景。Agent-Reach 的策略层需要让业务方显式声明优先级而不是在代码里写死。三种常见通道的差异可以简单对照。通道成本触达率实时性典型场景站内信低中秒级日活用户、运营通知邮件低中分钟级报表、账单、订阅内容短信高高秒级验证码、告警、紧急通知这里有一个容易踩坑的地方通道优先级不能只看成本还要看业务紧急程度。有一次我在日志里发现一批重要告警居然全部通过站内信发送原因是业务方按照“成本从低到高”的固定模板配置了策略。结果告警用户在当天并没有打开 App导致告警延迟了几个小时才通过邮件触达。后来我在策略层加了一个“紧急程度”字段当紧急程度为高时强制第一个通道走短信再退化到邮件和站内信。所以 Agent-Reach 不尝试替业务方做价值判断它只提供一个机制让业务方显式声明通道优先级并且允许按场景配置多套策略模板。3.4 幂等、去重与多通道互斥多通道降级带来的最大副作用之一是同一个任务可能被多次投递。比如站内信发送时网络超时执行层判断失败降级到邮件但实际情况是站内信已经写入成功只是响应超时用户就会同时收到站内信和邮件。要解决这个问题需要同时做两件事。第一件事是锁。每个任务在被 Worker 消费的时候都会尝试获取一个分布式锁锁的 key 是dedup_key。获取失败说明已经有其他 Worker 在处理同一任务直接跳过。这个锁在任务进入终态之前不会被释放所以第二个通道开始发送之前一定会检查任务状态。第二件事是幂等键要设计得足够精确。我见过很多团队使用“用户ID 消息类型”作为鉴权幂等键这明显不够。用户可能同一订单收到多条通知比如支付成功、发货、到货它们的消息类型不同但如果只按用户类型去重正确的多条通知会被误杀。正确的方式是用“业务凭证”。比如订单号、审批单号、交易流水号。Agent-Reach 的dedup_key规范就是接收方标识 业务凭证 动作名称。这样既允许同一个用户的不同业务各收到一条也能保证同一个业务同一种动作不会重复触达。还有一层很容易忽略的保障任务创建时的事务一致性。如果业务代码先写数据库再往 Redis 推队列Redis 推失败的时候数据库已经提交了这时候任务就悄无声息地丢了。Agent-Reach 采用 outbox 模式业务方在同一个数据库事务里写业务数据和 outbox 表之后再有一个独立 Task Relay 把 outbox 里的记录转发到 Redis。这样即使 Redis 短暂不可用任务也只是停留在 outbox 表里不会丢失等 Redis 恢复后继续转发。4. 实操过程从零跑通一个最小可用版本4.1 目录与依赖Agent-Reach 的最小可用版本不复杂我把项目结构控制在五个目录以内。agent_reach/ ├── adapters/ │ ├── base.py │ ├── webhook.py │ └── sms.py ├── core/ │ ├── models.py │ ├── queue.py │ └── worker.py ├── schema/ │ └── task.json └── scripts/ ├── produce_task.py └── scan_retry.py依赖只有三个核心库asyncpg或者任何 PostgreSQL 驱动、redis-py、httpx。如果你的 Worker 是单机部署甚至不需要引入额外的任务调度框架直接写一个常驻进程循环消费即可。数据库侧需要三张核心表tasks 表保存任务当前状态task_audit 表保存状态变更日志outbox 表保存待转发到队列的任务。这三张表都不需要复杂设计唯一要注意的是 tasks 表必须给dedup_key建唯一索引。4.2 队列消费与通道适配器代码Worker 的核心逻辑是一个循环从 Redis 取任务加锁执行通道发送更新状态写审计日志。下面的代码是通道适配器的基础接口。class BaseAdapter: name base def __init__(self, config: dict): self.config config async def send(self, task: dict, channel_spec: dict) - dict: raise NotImplementedError async def close(self): return None每个通道只需要继承 BaseAdapter 并实现 send 方法。下面是一个简单的 WebhookAdapter 示例。import httpx class ChannelError(Exception): pass class WebhookAdapter(BaseAdapter): name webhook def __init__(self, config: dict): super().__init__(config) self.timeout config.get(timeout, 10) self.client httpx.AsyncClient(timeoutself.timeout) async def send(self, task: dict, channel_spec: dict) - dict: payload { template_id: task[template_id], user_id: task[recipient][user_id], } resp await self.client.post( channel_spec[target_url], jsonpayload, headers{X-Reach-Task: task[task_id]}, ) if resp.status_code 400: raise ChannelError( f{self.name} return {resp.status_code}: {resp.text[:200]} ) return { channel: self.name, status: accepted, remote_id: resp.headers.get(X-Event-Id), }Worker 的核心消费逻辑也保持得很薄。关键在于队列消费使用brpoplpush先把任务从 pending 列表移到 processing 列表处理完成后再删除。这样如果 Worker 进程崩溃任务还停留在 processing 列表里可以从那里恢复。async def run_worker(queue, lock, adapter_map): while True: raw await queue.brpoplpush(reach:pending, reach:processing, timeout5) if not raw: continue task json.loads(raw) if not await lock.acquire(task[dedup_key], timeout30): continue try: for channel_name in task[channels_priority]: if await is_task_finished(task[task_id]): break adapter adapter_map[channel_name] channel_spec task[recipient][channels].get(channel_name) if not channel_spec: continue try: result await adapter.send(task, channel_spec) await mark_success(task[task_id], channel_name, result) break except ChannelError: await mark_failed(task[task_id], channel_name) continue else: await move_to_dead_letter(task[task_id]) finally: await queue.lrem(reach:processing, 1, raw) await lock.release(task[dedup_key])这里的is_task_finished很重要它会在每次切换通道前检查任务是不是已经被某个通道标记成功防止出现“站内信其实已经收到客户又收到邮件”的情况。4.3 关键参数与效果检查运行起来之后有几个参数需要根据实际场景调整。我整理了一份参考值注意这只是初始值正式上线前要做压测。参数建议值说明Worker 并发数CPU 核心数 * 2并发过高会打爆下游通道每次请求超时5 到 10 秒超时不是越大越好最大重试次数3超过 3 次应进入死信人工处理重试基础间隔1 秒指数退避避免重试风暴重试最大间隔60 秒防止长任务长期占用状态队列积压告警阈值1000超过说明消费能力不足我一般会用三个指标判断系统是否健康。第一个是task_created_total和task_succeeded_total两者长期接近说明成功率正常。第二个是queue_depth如果持续增长说明 Worker 消费速度跟不上生产速度需要扩容或者优化下游通道。第三个是通道调用耗时用 P99 来看如果某个通道 P99 超过重试超时时间大量的任务会进入无意义的重试循环。启动完最小版本之后建议立刻模拟三类故障手动停掉一个 Worker 进程、让某个通道返回 500、把 Redis 停掉一分钟再恢复。观察任务状态变化看看是不是最终能回到预期状态。5. 常见问题与排查技巧实录5.1 我踩过的三个坑第一个坑是没有区分“通道受理成功”和“最终触达成功”。短信场景尤其明显。短信网关只要成功接收提交请求就会返回一个状态码但这个状态码不代表用户手机真的收到了。早期 Agent-Reach 把网关的“受理成功”当成“成功”直接把任务置为 SUCCEEDED。后来发现部分短信因为手机号停机、网关路由问题被退回用户根本没收到但系统里显示成功。修复方式是让每个通道适配器返回两级状态。一级是“accepted”表示渠道已受理二级是“delivered”表示渠道通过回执确认已到达用户。对于支持回执的通道比如短信、邮件Worker 会等待异步回执再置为最终成功对于不支持回执的通道则配置一个冷静时间在状态上标记“已受理等待确认”。第二个坑是重试退避没有加抖动。早期我把指数退避写成一种固定的函数第一次重试等 1 秒第二次等 2 秒第三次等 4 秒。看起来没问题但高并发场景下会形成“重试风暴”一大批失败任务在同一时刻进入下一次重试下游通道被打得更惨然后又集体失败形成恶性循环。修复方式很简单在退避时间上乘以一个随机因子让每次重试的精确时间点错开。import random def retry_delay(attempt, base_seconds1, factor2, max_delay60): delay base_seconds * (factor ** (attempt - 1)) delay min(delay, max_delay) return delay * random.uniform(0.8, 1.2)第三个坑是任务创建时的数据一致性问题。最初我把“写任务”和“推队列”放在同一个函数里先写数据库再推 Redis。Redis 偶尔抖动任务就会丢。后来我意识到这个问题不能只靠重试解决因为如果推 Redis 失败后重新执行整个函数数据库里可能已经存在这条 task_id再次插入会报唯一键冲突。最终用了 outbox 模式业务方只负责写 outbox 表Task Relay 负责把 outbox 里的记录转发给 Redis。这样业务方的操作变成一次单库事务不会再出现跨系统一致性问题。5.2 排查触达问题的通用思路遇到触达问题不要一上来就看代码。我的排查顺序永远是先看任务状态再看通道耗时最后才看日志。任务状态能直接告诉我们问题处在哪一段。如果大量任务堆积在 PENDING说明队列消费出了问题。如果集中在 SENDING说明通道响应慢。如果集中在 RETRYING说明通道在持续失败。如果 DEAD 数量多说明策略配置不合理或者第三方通道确实长时间不可用。现象可能原因检查方式队列积压Worker 消费能力不足查看并发数和下游通道 P99 耗时用户收到重复消息幂等键设计不合理检查 dedup_key 是否包含业务凭证显示成功但用户没收到把受理回执当成送达回执检查适配器返回的状态字段任务丢失先写队列后写数据库产生不一致改造为 outbox 模式重试风暴退避没有抖动检查 Retry 时间分布是否过于集中每条线上故障都有一个“最早出现的信号”可能是一条通道超时日志也可能是某个指标突然掉了。重点是平时就把这些信号收集起来而不是等出了事故再去翻日志。我在 Worker 里会同步输出一个结构化日志包含 task_id、当前通道、尝试次数、耗时、返回码这样回溯问题时一行日志就能定位到完整链路。还有一个很实用的小技巧所有进入 DEAD 状态的任务必须保留原始任务体和最后一次通道异常信息。有些死信任务其实只是下游业务返回了一个短暂异常人工重放后就能成功。如果没有原始任务体重放时要重新构造上下文非常容易出错。5.3 离线补偿扫描兜底最后一公里即便有队列、有重试、有幂等线上还是会出现极端情况Worker 进程被杀、Redis 数据由于误操作被清空、某个消息从队列里消费后还没来得及写数据库就跟数据库断了连接。这些情况单靠队列本身无法覆盖所以 Agent-Reach 保留了一个离线补偿扫描任务。它会定期扫描 tasks 表把所有“看起来还卡在中间状态”的任务重新拉回处理队列。UPDATE agent_reach.tasks SET status RETRYING, updated_at now() WHERE status SENDING AND updated_at now() - interval 2 minutes AND attempt_count max_attempts;这条 SQL 的含义是如果任务处于发送中状态超过两分钟说明 Worker 很可能已经失去心跳就把它重置到 RETRYING让队列重新接管。要注意一个隐患如果有多台 Worker 同时执行补偿扫描同一个任务可能被同时捞出来。所以这个扫描任务必须配合分布式锁确保同一时间只有一个执行者。另外补偿扫描不能让 DEAD 状态的任务自动复活否则会把所有失败任务反复重试到天荒地老。DEAD 状态必须保留给人工决策。离线补偿是兜底方案不是主路径。它能让你在极端情况下依然保证“任务不会默默消失”但真正决定触达质量的还是主流程里的幂等、超时和降级策略是否设计得清楚。兜底做得再好也不能弥补主路线的混乱。最后再分享一个我个人的体会Agent-Reach 这种编排层项目最大的风险不是技术复杂度而是从一开始就想覆盖所有通道、所有场景。先用一个通道把状态机跑通再往上加通道优先级和重试策略每一步演进都要有明确的线上数据支撑。先把简单的路走通再让复杂的路可控这才是触达编排系统真正应该有的进化节奏。
看到“Agent-Reach”这个标题先别急着往“又一个人工智能框架”上靠。它其实是我近期整理的一个内部项目名字听着挺宏大核心就一件事让各种 Agent包括 AI 智能体、自动化脚本、运营机器人在需要触达用户或下游系统时不丢消息、不重复发、还能在通道挂掉时自动换路。简单说它是个触达编排层。这类项目在团队里常常被忽略但真到线上出问题的时候最让人头疼的往往不是业务代码而是“用户该收到消息却没收到”“一次活动通知被重复发了三遍”“短信网关超时后面所有任务全部卡死”。Agent-Reach 想解决的就是这一类问题把不可靠的触达动作变成一个可重试、可观测、可审计、可降级的过程。如果你正在做消息推送平台、自动化运营工具、AI Agent 的工具调用链或者只是想把散落在业务代码里的通知逻辑收拢起来这篇内容应该能给你一个可以直接参考的落地思路。我会把设计思路、核心模型、关键代码和排查方法全部拆开讲不绕弯子。1. 项目从哪来Agent-Reach 到底在解决什么事1.1 名字里的两个词Agent 和 Reach“Agent”在这里不单指 AI Agent而是泛指一切“需要主动发起动作的执行体”。运营系统里的定时任务、审批流里的提醒服务、客服机器人里的对话回复、AI Agent 在接到指令后调用的外部工具本质上都是 Agent 的一种形态。“Reach”也不只是“发出去”它强调的是“到达”。一个消息从产生到真正被用户看到中间隔着队列、适配器、第三方通道、重试、回执确认好几个环节。大多数系统的问题出在“我认为发出去了”和“用户确实收到了”之间的差距上。所以 Agent-Reach 这个名字的意思是把触达当成一条完整的链路来管而不是只做一个 send 函数。它负责的任务不是“怎么构造消息内容”而是“怎么可靠地把消息送到目标手上”。1.2 它解决的三类典型痛我在实际项目中见过大量类似场景痛点基本可以归成三类。第一类是“发了等于没发”。比如短信网关偶发超时业务代码只 catch 了异常但没有重试机制又比如回调接口返回 200但业务代码里面处理失败用户始终没收到提醒。这类问题最难查因为日志显示“成功”用户却说“没收到”。第二类是“没发等于重复发”。任务在队列里被消费者拉取后进程 crash消息重新入队业务侧没有做幂等用户就会收到两条一样的短信。看似小概率但在大促、活动通知这类场景里小概率乘以百万级用户量就会变成线上事故。第三类是“全堵死”。触达动作散落在各业务模块没有统一队列没有背压控制。某个通道一慢调用方线程被占用最后整个服务跟着卡死。我见过一次邮件服务响应慢直接把订单服务拖崩的事故原因就是订单完成通知里同步发送了邮件。Agent-Reach 把这些场景集中抽象成“触达任务”由统一调度层来管理通道、重试、状态和幂等业务方只需要描述“我要触达谁、用什么模板、希望优先走哪个通道”。1.3 适合谁来用如果你所在的团队已经有专门的消息平台或者只做站内信这种单通道场景Agent-Reach 这种编排层会显得有点重。但如果你的系统需要同时接入站内信、邮件、短信、webhook并且业务方经常提出“不同用户走不同通道”“失败后按顺序降级”“某些任务必须保证最终送达”那这套思路就非常合适。同样适合的还有正在做 AI Agent 应用的开发者。AI Agent 在调用外部工具、给用户发送异步结果、触发下游工作流的时候本质上也是一次触达行为。把触达逻辑从 Agent 代码里抽出来交给 Agent-Reach 这样的编排层能让 Agent 本身更轻也更容易排查“为什么工具调用后没有效果”这类问题。2. 整体设计拆解为什么把触达做成“编排层”2.1 别把发送逻辑散落在业务代码里很多项目的触达逻辑是散落的。订单模块发订单通知用户模块发验证码运营模块发活动推送每个模块自己写一套调用第三方接口的逻辑自己处理重试。这种做法在业务量小的时候没问题但一旦通道种类变多、消息量变大就会暴露几个问题。业务代码会越来越脏。一个下单服务里既要管库存、扣款、发票还要管短信模板、通道切换、失败重试任何一行通道配置修改都会影响业务发布。其次是通道故障难以隔离。短信服务商出问题理论上应该只影响短信这个通道但因为代码散落可能导致所有包含短信发送逻辑的业务模块同时报错。Agent-Reach 的做法是把触达从业务里抽出来做成独立的编排层。业务只负责提交一个“意图”比如“通知用户 A 订单支付成功”至于走短信还是走邮件、什么时候发、失败重试几次、重试间隔多久全部由编排层决定。你可以把它理解成外卖平台。用户下单后平台负责调度骑手、规划路线、处理异常而不是让每个餐厅自己招骑手送餐。这样餐厅只需要做好餐配送的事交给统一平台。2.2 三级触达模型策略、执行、通道Agent-Reach 内部把触达拆成三层策略层、执行层、通道层。策略层只负责描述“触达意图”。包括目标用户、消息模板、要尝试的通道顺序、最大重试次数、过期时间、幂等键。策略层不关心短信接口长什么样不关心邮件模板怎么渲染只负责表达需求。执行层负责把策略变成执行计划。它会去消费队列里的触达任务根据通道顺序逐个尝试记录每次尝试的返回结果更新任务状态决定下一步是继续换通道、重试还是进入死信队列。执行层是整个系统的中枢。通道层是最具体的部分。每个通道对应一个适配器比如 SmsAdapter、EmailAdapter、WebhookAdapter。适配器做的事情非常简单接收一个触达任务和该通道的配置调用底层服务返回成功或失败。通道层不关心业务语义也不关心重试策略。这三层的好处是职责清晰互相之间只通过任务对象通信。以后要新增一个通道只需要写一个适配器注册进去策略层和执行层一行代码都不用改。要调整重试策略也只影响执行层。2.3 技术选型与部署形态我最早的版本没有引入任何新组件只是在一个业务服务里加了一张表、一个定时任务每五秒扫描一次需要发送的任务。后来消息量大了定时扫描的实时性不够才引入了队列。最终的技术选型很朴素PostgreSQL 存状态和审计日志Redis 做队列Python 写 Worker。Redis Stream 负责承载待处理任务PostgreSQL 是唯一的事实来源。为什么选 Redis Stream 而不是直接用 Redis List因为 Stream 天然支持消费组、消息确认、消息历史对“至少一次”的消费语义支持得更好。这里有一个很关键的设计原则不把 Redis 当成最终存储。所有触达任务先写入 PostgreSQL状态也是以 PostgreSQL 为准Redis 里只是“待处理游标”的缓存。就算 Redis 里的数据全部丢失重启后仍然可以从扫描表中恢复出所有未完成任务。这也是 Agent-Reach 能承受部分组件故障的底气。组件我这里用的替代方案选型理由状态存储PostgreSQLMySQL、DynamoDB需要事务、唯一索引、审计查询队列Redis StreamRabbitMQ、Kafka轻量、运维简单、团队熟悉WorkerPython asyncioGo、Node.js迭代快、适配器开发方便任务格式JSONProtobuf调试方便兼容性好如果团队已经有一套现成的 RabbitMQ 或者 Kafka完全可以把 Redis 换成它们多出的收益主要是消息持久化和回溯能力。核心模型不变变的只是队列的接入方式。3. 核心细节任务模型、状态机与通道降级3.1 统一的任务描述长什么样在 Agent-Reach 里所有触达意图都使用同一套 JSON 结构描述。下面这个例子是一个“订单支付成功”通知任务。{ task_id: c82a4f7e-3b2a-4f8c-9d1a-8f0a76b3e2d1, dedup_key: user_10001:order_8823:pay_success, recipient: { user_id: 10001, channels: { in_app: { conversation_id: 10001 }, email: u_10001example.com, sms: 8613800000000 } }, template_id: order_pay_success, channels_priority: [in_app, email, sms], strategy: { max_attempts: 3, timeout_seconds: 10, retry_backoff: exponential, schedule_start: 2025-06-01T10:00:00Z, expire_at: 2025-06-01T23:00:00Z } }这里最值得关注的是dedup_key和channels_priority。dedup_key是整个幂等机制的核心。它由“接收方标识 业务标识 业务动作”组成而不是简单用 task_id因为 task_id 每次生成都不同无法屏蔽业务侧的重复提交。比如用户下单成功后服务重启触发了一次补偿逻辑重新提交了同一个订单的通知如果没有 dedup_key用户就会收到两条一样的消息。channels_priority表达的是降级顺序。站内信成本最低优先尝试如果用户长期不登录站内信送达率不高可以降级到邮件邮件也不是实时通道紧急消息再降级到短信。这个顺序是策略层的选择执行层只需要按顺序执行。3.2 状态机与审计触达任务不是“发完就完”很多团队在实现发送模块的时候只在数据库里存了“发送状态”要么是成功要么是失败。但真实触达过程远比这两种状态要复杂。Agent-Reach 的每个任务都有一套明确的状态机。状态含义可流转到CREATED任务已创建尚未进入队列PENDING / FAILEDPENDING已进入队列等待 Worker 消费SENDING / FAILEDSENDING正在向某个通道投递SUCCEEDED / RETRYING / FAILEDRETRYING上一通道失败等待重试或换通道SENDING / FAILED / DEADSUCCEEDED已有通道确认触达成功终态DEAD所有策略都用完最终失败终态可人工重放为什么状态要设计得这么细因为只有状态细才能回答“这个任务到底卡在哪”。线上排查用户没收到消息的时候第一件事就是查任务状态。如果任务停在 SENDING 超过两分钟说明 Worker 可能一直在等待下游响应或者进程已经崩溃但队列里的消息没有被重新消费。如果任务停在 RETRYING说明通道在持续报错需要去检查第三方通道状态。如果任务进入 DEAD说明配置的策略已经耗尽需要人工介入。每一条状态变更都会写入审计日志。这个日志在形式上和业务日志很像但它只记录触达任务自己的生命周期谁在什么时间把它从什么状态变到什么状态当时的通道返回是什么。没有这份审计事后复盘“为什么多发了消息”会非常痛苦因为你只能靠猜。3.3 通道优先级与降级选择触达通道没有绝对的好坏只有适不适合当前场景。Agent-Reach 的策略层需要让业务方显式声明优先级而不是在代码里写死。三种常见通道的差异可以简单对照。通道成本触达率实时性典型场景站内信低中秒级日活用户、运营通知邮件低中分钟级报表、账单、订阅内容短信高高秒级验证码、告警、紧急通知这里有一个容易踩坑的地方通道优先级不能只看成本还要看业务紧急程度。有一次我在日志里发现一批重要告警居然全部通过站内信发送原因是业务方按照“成本从低到高”的固定模板配置了策略。结果告警用户在当天并没有打开 App导致告警延迟了几个小时才通过邮件触达。后来我在策略层加了一个“紧急程度”字段当紧急程度为高时强制第一个通道走短信再退化到邮件和站内信。所以 Agent-Reach 不尝试替业务方做价值判断它只提供一个机制让业务方显式声明通道优先级并且允许按场景配置多套策略模板。3.4 幂等、去重与多通道互斥多通道降级带来的最大副作用之一是同一个任务可能被多次投递。比如站内信发送时网络超时执行层判断失败降级到邮件但实际情况是站内信已经写入成功只是响应超时用户就会同时收到站内信和邮件。要解决这个问题需要同时做两件事。第一件事是锁。每个任务在被 Worker 消费的时候都会尝试获取一个分布式锁锁的 key 是dedup_key。获取失败说明已经有其他 Worker 在处理同一任务直接跳过。这个锁在任务进入终态之前不会被释放所以第二个通道开始发送之前一定会检查任务状态。第二件事是幂等键要设计得足够精确。我见过很多团队使用“用户ID 消息类型”作为鉴权幂等键这明显不够。用户可能同一订单收到多条通知比如支付成功、发货、到货它们的消息类型不同但如果只按用户类型去重正确的多条通知会被误杀。正确的方式是用“业务凭证”。比如订单号、审批单号、交易流水号。Agent-Reach 的dedup_key规范就是接收方标识 业务凭证 动作名称。这样既允许同一个用户的不同业务各收到一条也能保证同一个业务同一种动作不会重复触达。还有一层很容易忽略的保障任务创建时的事务一致性。如果业务代码先写数据库再往 Redis 推队列Redis 推失败的时候数据库已经提交了这时候任务就悄无声息地丢了。Agent-Reach 采用 outbox 模式业务方在同一个数据库事务里写业务数据和 outbox 表之后再有一个独立 Task Relay 把 outbox 里的记录转发到 Redis。这样即使 Redis 短暂不可用任务也只是停留在 outbox 表里不会丢失等 Redis 恢复后继续转发。4. 实操过程从零跑通一个最小可用版本4.1 目录与依赖Agent-Reach 的最小可用版本不复杂我把项目结构控制在五个目录以内。agent_reach/ ├── adapters/ │ ├── base.py │ ├── webhook.py │ └── sms.py ├── core/ │ ├── models.py │ ├── queue.py │ └── worker.py ├── schema/ │ └── task.json └── scripts/ ├── produce_task.py └── scan_retry.py依赖只有三个核心库asyncpg或者任何 PostgreSQL 驱动、redis-py、httpx。如果你的 Worker 是单机部署甚至不需要引入额外的任务调度框架直接写一个常驻进程循环消费即可。数据库侧需要三张核心表tasks 表保存任务当前状态task_audit 表保存状态变更日志outbox 表保存待转发到队列的任务。这三张表都不需要复杂设计唯一要注意的是 tasks 表必须给dedup_key建唯一索引。4.2 队列消费与通道适配器代码Worker 的核心逻辑是一个循环从 Redis 取任务加锁执行通道发送更新状态写审计日志。下面的代码是通道适配器的基础接口。class BaseAdapter: name base def __init__(self, config: dict): self.config config async def send(self, task: dict, channel_spec: dict) - dict: raise NotImplementedError async def close(self): return None每个通道只需要继承 BaseAdapter 并实现 send 方法。下面是一个简单的 WebhookAdapter 示例。import httpx class ChannelError(Exception): pass class WebhookAdapter(BaseAdapter): name webhook def __init__(self, config: dict): super().__init__(config) self.timeout config.get(timeout, 10) self.client httpx.AsyncClient(timeoutself.timeout) async def send(self, task: dict, channel_spec: dict) - dict: payload { template_id: task[template_id], user_id: task[recipient][user_id], } resp await self.client.post( channel_spec[target_url], jsonpayload, headers{X-Reach-Task: task[task_id]}, ) if resp.status_code 400: raise ChannelError( f{self.name} return {resp.status_code}: {resp.text[:200]} ) return { channel: self.name, status: accepted, remote_id: resp.headers.get(X-Event-Id), }Worker 的核心消费逻辑也保持得很薄。关键在于队列消费使用brpoplpush先把任务从 pending 列表移到 processing 列表处理完成后再删除。这样如果 Worker 进程崩溃任务还停留在 processing 列表里可以从那里恢复。async def run_worker(queue, lock, adapter_map): while True: raw await queue.brpoplpush(reach:pending, reach:processing, timeout5) if not raw: continue task json.loads(raw) if not await lock.acquire(task[dedup_key], timeout30): continue try: for channel_name in task[channels_priority]: if await is_task_finished(task[task_id]): break adapter adapter_map[channel_name] channel_spec task[recipient][channels].get(channel_name) if not channel_spec: continue try: result await adapter.send(task, channel_spec) await mark_success(task[task_id], channel_name, result) break except ChannelError: await mark_failed(task[task_id], channel_name) continue else: await move_to_dead_letter(task[task_id]) finally: await queue.lrem(reach:processing, 1, raw) await lock.release(task[dedup_key])这里的is_task_finished很重要它会在每次切换通道前检查任务是不是已经被某个通道标记成功防止出现“站内信其实已经收到客户又收到邮件”的情况。4.3 关键参数与效果检查运行起来之后有几个参数需要根据实际场景调整。我整理了一份参考值注意这只是初始值正式上线前要做压测。参数建议值说明Worker 并发数CPU 核心数 * 2并发过高会打爆下游通道每次请求超时5 到 10 秒超时不是越大越好最大重试次数3超过 3 次应进入死信人工处理重试基础间隔1 秒指数退避避免重试风暴重试最大间隔60 秒防止长任务长期占用状态队列积压告警阈值1000超过说明消费能力不足我一般会用三个指标判断系统是否健康。第一个是task_created_total和task_succeeded_total两者长期接近说明成功率正常。第二个是queue_depth如果持续增长说明 Worker 消费速度跟不上生产速度需要扩容或者优化下游通道。第三个是通道调用耗时用 P99 来看如果某个通道 P99 超过重试超时时间大量的任务会进入无意义的重试循环。启动完最小版本之后建议立刻模拟三类故障手动停掉一个 Worker 进程、让某个通道返回 500、把 Redis 停掉一分钟再恢复。观察任务状态变化看看是不是最终能回到预期状态。5. 常见问题与排查技巧实录5.1 我踩过的三个坑第一个坑是没有区分“通道受理成功”和“最终触达成功”。短信场景尤其明显。短信网关只要成功接收提交请求就会返回一个状态码但这个状态码不代表用户手机真的收到了。早期 Agent-Reach 把网关的“受理成功”当成“成功”直接把任务置为 SUCCEEDED。后来发现部分短信因为手机号停机、网关路由问题被退回用户根本没收到但系统里显示成功。修复方式是让每个通道适配器返回两级状态。一级是“accepted”表示渠道已受理二级是“delivered”表示渠道通过回执确认已到达用户。对于支持回执的通道比如短信、邮件Worker 会等待异步回执再置为最终成功对于不支持回执的通道则配置一个冷静时间在状态上标记“已受理等待确认”。第二个坑是重试退避没有加抖动。早期我把指数退避写成一种固定的函数第一次重试等 1 秒第二次等 2 秒第三次等 4 秒。看起来没问题但高并发场景下会形成“重试风暴”一大批失败任务在同一时刻进入下一次重试下游通道被打得更惨然后又集体失败形成恶性循环。修复方式很简单在退避时间上乘以一个随机因子让每次重试的精确时间点错开。import random def retry_delay(attempt, base_seconds1, factor2, max_delay60): delay base_seconds * (factor ** (attempt - 1)) delay min(delay, max_delay) return delay * random.uniform(0.8, 1.2)第三个坑是任务创建时的数据一致性问题。最初我把“写任务”和“推队列”放在同一个函数里先写数据库再推 Redis。Redis 偶尔抖动任务就会丢。后来我意识到这个问题不能只靠重试解决因为如果推 Redis 失败后重新执行整个函数数据库里可能已经存在这条 task_id再次插入会报唯一键冲突。最终用了 outbox 模式业务方只负责写 outbox 表Task Relay 负责把 outbox 里的记录转发给 Redis。这样业务方的操作变成一次单库事务不会再出现跨系统一致性问题。5.2 排查触达问题的通用思路遇到触达问题不要一上来就看代码。我的排查顺序永远是先看任务状态再看通道耗时最后才看日志。任务状态能直接告诉我们问题处在哪一段。如果大量任务堆积在 PENDING说明队列消费出了问题。如果集中在 SENDING说明通道响应慢。如果集中在 RETRYING说明通道在持续失败。如果 DEAD 数量多说明策略配置不合理或者第三方通道确实长时间不可用。现象可能原因检查方式队列积压Worker 消费能力不足查看并发数和下游通道 P99 耗时用户收到重复消息幂等键设计不合理检查 dedup_key 是否包含业务凭证显示成功但用户没收到把受理回执当成送达回执检查适配器返回的状态字段任务丢失先写队列后写数据库产生不一致改造为 outbox 模式重试风暴退避没有抖动检查 Retry 时间分布是否过于集中每条线上故障都有一个“最早出现的信号”可能是一条通道超时日志也可能是某个指标突然掉了。重点是平时就把这些信号收集起来而不是等出了事故再去翻日志。我在 Worker 里会同步输出一个结构化日志包含 task_id、当前通道、尝试次数、耗时、返回码这样回溯问题时一行日志就能定位到完整链路。还有一个很实用的小技巧所有进入 DEAD 状态的任务必须保留原始任务体和最后一次通道异常信息。有些死信任务其实只是下游业务返回了一个短暂异常人工重放后就能成功。如果没有原始任务体重放时要重新构造上下文非常容易出错。5.3 离线补偿扫描兜底最后一公里即便有队列、有重试、有幂等线上还是会出现极端情况Worker 进程被杀、Redis 数据由于误操作被清空、某个消息从队列里消费后还没来得及写数据库就跟数据库断了连接。这些情况单靠队列本身无法覆盖所以 Agent-Reach 保留了一个离线补偿扫描任务。它会定期扫描 tasks 表把所有“看起来还卡在中间状态”的任务重新拉回处理队列。UPDATE agent_reach.tasks SET status RETRYING, updated_at now() WHERE status SENDING AND updated_at now() - interval 2 minutes AND attempt_count max_attempts;这条 SQL 的含义是如果任务处于发送中状态超过两分钟说明 Worker 很可能已经失去心跳就把它重置到 RETRYING让队列重新接管。要注意一个隐患如果有多台 Worker 同时执行补偿扫描同一个任务可能被同时捞出来。所以这个扫描任务必须配合分布式锁确保同一时间只有一个执行者。另外补偿扫描不能让 DEAD 状态的任务自动复活否则会把所有失败任务反复重试到天荒地老。DEAD 状态必须保留给人工决策。离线补偿是兜底方案不是主路径。它能让你在极端情况下依然保证“任务不会默默消失”但真正决定触达质量的还是主流程里的幂等、超时和降级策略是否设计得清楚。兜底做得再好也不能弥补主路线的混乱。最后再分享一个我个人的体会Agent-Reach 这种编排层项目最大的风险不是技术复杂度而是从一开始就想覆盖所有通道、所有场景。先用一个通道把状态机跑通再往上加通道优先级和重试策略每一步演进都要有明确的线上数据支撑。先把简单的路走通再让复杂的路可控这才是触达编排系统真正应该有的进化节奏。
