Agent-Reach 这个项目最初其实是被一套又笨又重的 CRM 触达配置逼出来的。之前团队做用户运营每次活动通知都要改一遍后台配置定时任务、渠道调用、失败重试全揉在一起改一个参数就得等发版。后来我抽了半个月时间把这块重写成了一个独立组件也就是 Agent-Reach——一个把该不该触达、选哪条通道、说什么内容从业务代码里抽出来的智能触达编排框架。它不接大模型也能干活接了大模型就能变成真正的 Agent。适合做用户召回、告警通知、事务消息的团队参考。这不是一个发短信/发邮件的工具库而是一套完整的触达链路治理方案。你在做类似事情的时候会发现真正难的不是调通某个渠道的 API而是怎么让决策和执行解耦、怎么保证高并发下消息不丢不重、怎么让运营人员自己改规则而不需要研发介入。Agent-Reach 就是冲着这些痛点去的下面我把整个设计思路和落地过程拆开讲。1. 拆解 Agent-Reach 的出发点触达链路为什么值得单独做一个框架1.1 传统触达系统最让人难受的三个问题我先说一个很常见的场景用户下单失败系统需要给用户发一条短信提示。大多数团队的做法是在支付回调里直接调用短信 SDK失败就重试两次日志打到业务日志里。表面看没啥问题但当你有一百个类似的触发场景时事情就开始失控。第一个问题是业务代码与触达逻辑紧耦合。每个业务里都塞着一小段发短信/发邮件/推企微的代码参数散落在各处运营同学想改一句话都得来找研发。第二个问题是渠道 API 不一致。有些渠道要求签名、有些要求模板 ID、有些要传回调 URL对接三个渠道之后适配代码就变成一堆 if-else再换渠道基本等于重写。第三个问题是可观测性差。消息发出去了没有统一追踪用户投诉没收到时你只能去日志里捞捞不捞得到全看运气。Agent-Reach 最早就是想解决这三个问题。我把什么时候触达通过什么触达触达什么内容这三件事全部标准化做成配置和策略业务方只负责抛事件剩下的交给 Agent-Reach 去决策和执行。1.2 用 Agent 概念重新定义触达决策边界很多人听到 Agent 就想到大模型但 Agent-Reach 里的 Agent 是一个可插拔的决策器大模型只是它的一种实现。核心思路是把原来散落在代码里的if 条件 - 调用 SDK抽成一个独立的决策节点。具体来说一条触达请求会经过下面这条链路事件进入系统归一化为统一的事件结构带上用户 ID、事件类型、业务参数。决策器根据规则或 LLM 判断是不是要触达用哪个渠道用哪个模板编排层把决策结果生成 Message丢进持久化队列。消费端拿到 Message 后通过对应的通道适配器发送并记录状态。这个链路里决策和执行是严格分开的。决策器可以被替换通道也可以被替换互不影响。这也是为什么叫 Agent-Reach——让智能体去决定如何触达而不是人肉判断。2. Agent-Reach 整体架构与核心数据结构2.1 分层决策、编排、通道、存储四层Agent-Reach 的代码我一开始就按四层来拆避免后续堆成浆糊层次职责核心组件决策层判断是否触达、选择渠道和模板RuleEngine / LLMDecision编排层生成消息、管理状态机、执行重试Dispatcher / StateMachine通道层对接各类外部触达渠道BaseChannel 及各实现存储层持久化消息、队列、幂等表Postgres / Redis Streams每层之间只通过标准的数据结构通信。决策层不关心消息怎么发出去通道层也不关心为什么发这条消息。这样做的直接好处是我可以同时接多个渠道也可以在同一个渠道上叠加不同的决策策略而不会互相污染。2.2 Message 状态机与队列模型消息是 Agent-Reach 的核心实体我给它定义了这几个状态pending事件已进入等待决策。queued决策完成已入队等待发送。sending正在调用外部渠道。sent渠道确认接收。failed多次重试后仍失败。canceled决策被撤销或过期。rate_limited被频控拦截等待延时重试。对应数据结构简化版本是这样的dataclass class Message: id: str # 业务消息ID幂等键 trace_id: str # 链路追踪ID project_id: str event_type: str # 例如 order_failed / user_silence recipient: str # 收件人标识 channel: str # email / webhook / dingtalk template_id: str params: dict status: str attempts: int next_retry_at: datetime | None created_at: datetime sent_at: datetime | None last_error: str | None队列模型我用的是 Redis Streams。对比普通 ListStreams 支持消费组、消息 ACK 和 Pending 列表这三点对触达场景太重要了。消费者 A 发完短信还没写状态进程挂了消息不会丢等它恢复后 Pending 里的消息还能取回来重新做。# 生产端 XADD messages * event_type order_failed recipient user_123 channel email template order_failed_v1 params {...}消费端用 XREADGROUP 配合 XACK保证消息至少被处理一次。至于最多一次还是至少一次对触达场景来说我选择至少一次加幂等而不是最多一次加丢消息。3. 通道适配层与模板引擎实现一次编排处处触达3.1 统一 Channel 接口兼容 邮件/Webhook/IM通道层最重要的设计就是那个极简的接口。所有渠道实现都围绕一个方法展开class BaseChannel(ABC): channel_type: str abstractmethod def send(self, message: Message) - SendResult: 发送消息必须返回 SendResult abstractmethod def validate_message(self, message: Message) - None: 发送前校验避免无效调用SendResult 里至少包含三个字段success、channel_message_id、raw_response。channel_message_id是渠道返回的确认 ID比如邮件服务商的 message_id、钉钉的 msg_id这个字段对幂等非常关键后面会细说。邮件通道实现大概是这样的逻辑读取消息里的 recipient、template 渲染结果通过 SMTP 或邮件 API 发送。Webhook 通道则是把渲染后的内容 POST 到配置的 URL 上。IM 通道会额外注意 Markdown 格式兼容性问题钉钉和飞书对 Markdown 的支持细节并不一致所以我在适配器里做了二次清理。每个通道通过register_channel注册到通道工厂外部依赖倒置CHANNELS: dict[str, BaseChannel] {} def register_channel(channel: BaseChannel): CHANNELS[channel.channel_type] channel def get_channel(channel_type: str) - BaseChannel: return CHANNELS[channel_type]这样做以后新对接一个渠道只需要新增一个实现类注册进去其他代码一行不动。我后来接两个渠道时总计花了两天一半时间是在等对方的 API 文档审核。3.2 模板渲染的严格模式与上下文校验模板引擎我直接用了 Jinja2但和大多数人的用法不同——我开了StrictUndefined。这意味着模板里任何一个变量不存在渲染都会直接抛异常而不是默默输出空字符串。from jinja2 import Environment, StrictUndefined env Environment(undefinedStrictUndefined, autoescapeFalse)很多触达翻车事故就是这么来的模板里写了{{ user.name }}结果用户数据里没有 name 字段渲染出来是一块空白用户收到您好 您的账单已出这种消息体验很差。开了严格模式后这类错误在发送前就会被拦截宁可失败也不发错误内容。除此之外我在每个模板旁边维护了一份参数声明有点像 JSON Schema 简化版template: order_failed_v1 subject: 您的支付未成功请重新支付 params: order_id: required user_name: optional发送前编排层会拿这份声明去校验 Message.params缺了必填字段直接打回。加上渲染严格模式基本可以保证模板写错就发不出去把问题暴露在测试环境而不是线上用户面前。4. 决策引擎规则驱动让位LLM 补位4.1 规则引擎的优先级与条件评估决策层我先实现的是规则引擎因为大部分触达场景实际上是可穷举的。运营同学维护一张规则表每个规则包含条件、渠道、模板、优先级。Agent-Reach 的规则我用 YAML 描述比 JSON 可读性好很多rules: - id: churn_risk_recall name: 沉默14天用户召回 priority: 10 condition: user.last_active_days 14 and user.order_count 2 channel: email template: recall_07 - id: order_failed_notify name: 支付失败提醒 priority: 5 condition: event.type order_failed and order.retry_count 3 channel: sms template: order_failed_v1这里condition不是用 eval 硬解析的而是通过一个轻量表达式求值器类似simpleeval执行。限制只允许访问event、user、order这类预定义上下文对象不允许导入模块、不允许访问私有属性防止规则脚本成为安全漏洞。优先级的意义在于冲突消解。一个用户可能同时命中多条规则比如信用卡还款提醒和沉默用户召回但同一时间段内只触发最重要的一条所以我设计为按 priority 排序只执行第一条命中的规则。如果你希望多条规则都执行可以在规则上标记allow_parallel: true。4.2 引入 LLM 后的兜底与人工确认规则引擎能搞定 90% 的固定逻辑但有些触达决策确实需要更灵活的推理。比如用户刚投诉过这时候还按常规无脑推营销消息就等着流失了。我在 Agent-Reach 里预留了一个 LLMDecision 实现专门处理规则引擎无法明确判定的场景。我的做法不是让 LLM 直接输出发还是不发而是让它输出一个带置信度的建议{ should_send: true, channel: email, template: recall_08, reason: 该用户30天内有购买记录但近14天未活跃适合发起温和召回, confidence: 0.82 }confidence低于 0.7 时消息会进入pending_review状态由运营或客服在后台确认后才真正发送。这个机制很关键——LLM 只负责降低决策成本不直接承担骚扰用户的后果。另外我在 prompt 里会注入用户最近的触达记录和投诉记录避免明知对方刚投诉还继续发的低级错误。从我观察到的落地效果来说规则引擎的准确率高但覆盖有限LLM 能覆盖长尾场景但必须配有审核闸门。如果你想在生产环境直接用 LLM 做全自动触达请一定做好人审和限流。5. 上线后踩过的坑重试风暴、幂等和高频屏蔽5.1 重试风暴与退避策略这条坑我印象太深了。第一次灰度时某个渠道在晚高峰出现间歇性超时我当时的重试策略是失败后立刻重试最多重试 5 次间隔固定 30 秒。结果渠道抖动持续了 20 分钟Agent-Reach 的队列里堆了十几万条待重试消息Redis 内存告警直接触发而被重试的渠道压力更大了形成恶性循环。后来我把重试改成了指数退避加抖动第一次重试 1 分钟第二次 2 分钟第三次 4 分钟最大间隔 15 分钟同时加一个随机 0~5 秒的抖动。关键改动是记录每条消息的 next_retry_at消费者只处理当前时间大于该字段的消息避免重启后一秒内把堆积消息全部扫出去。retry_delays [60, 120, 240, 480, 900] delay retry_delays[min(message.attempts - 1, len(retry_delays) - 1)] message.next_retry_at datetime.now() timedelta(secondsdelay random.randint(0, 5))重试次数超过上限后消息进入failed状态并写入死信表由定时任务去做人工排查。同时我在监控大盘上加了重试占比和渠道错误率两个指标超过阈值就告警而不是等用户投诉。5.2 幂等控制的两种锁至少一次消息投递模型下幂等是必须做的。我一开始只在消息表上做了一个唯一索引(biz_id, event_type)用来防止同一个事件重复生成消息。这个防住了上游重复事件但没有防住另一个问题渠道已经发出消息但响应丢失导致重试多次发送。第二次坑就出现在这里。某用户收到三条订单支付失败短信。原因很简单邮件服务商已经接收邮件但回调响应超时Agent-Reach 以为失败了重试发送。要解决这个必须引入渠道回执 ID 去重。我在发送记录表里增加了一个channel_request_id字段逻辑是发送前生成一个唯一的request_id传给渠道如果渠道支持。发送后如果响应正常记录渠道返回的message_id。重试前先查发送记录如果某条消息最近一次尝试的channel_message_id已存在并且那次尝试的返回码是已受理或已发送则不再真正调用渠道直接把消息标记为sent。简单说就是要同时做两层幂等消息生成幂等靠biz_id唯一索引发送幂等靠渠道回执。只做一层都存在漏洞。5.3 用户频控与全局限流除了系统稳定性还有用户体验层面的限流。我记得灰度时规则配置错了某个用户同时命中三条活动触达规则一晚上收到 4 条不同渠道的营销消息第二天就取消关注了。后来我加了用户维度的频控在 Redis 里用INCR EXPIRE实现一个简单的令牌桶# 用户维度每天最多 2 条营销类消息 key ffreq:user:{user_id}:marketing:{date_today} count redis.incr(key) if count 1: redis.expire(key, 86400) if count 2: message.status rate_limited message.next_retry_at tomorrow_morning注意业务类型要区分。事务消息比如支付失败、账户异常不受营销频控限制否则会造成用户收不到关键通知。通道维度也要限流尤其是短信这种按条数计费、单通道 QPS 也有限的渠道。我在配置里给每个通道设了最大 QPS超过后消息进入rate_limited状态用固定速率慢慢消化而不是立刻丢弃。6. Agent-Reach 的实测结果与下一步计划6.1 两周灰度数据对比我在一个用户量约 20 万的业务上跑了两个星期灰度对比如下指标改造前老系统改造后Agent-Reach触达消息处理延迟平均 30 秒 重试不可控平均 1.8 秒不含渠道延迟重复触达率3.4%0.2%模板渲染错误每次上线靠人肉检查测试环境自动拦截新增渠道接入耗时3 天1 天纯适配层开发用户投诉收不到每周 20第一周 3 例第二周 0 例这个效果并不意外。核心不是代码写得多漂亮而是把原来混乱的调用关系收敛成了清晰的状态机和可配置的决策规则。尤其是幂等和频控的引入直接把重复率压下来了这比任何性能优化都更直观地反映在用户体验上。6.2 我给自己留的三个扩展方向第一个方向是可观测性增强。目前 trace_id 已经贯穿全程但还缺完整的 OpenTelemetry 埋点。我想把每个环节的耗时、决策命中规则、渠道响应码都变成结构化追踪这样排查问题时可以直接看到一条消息从事件到发送的全链路。第二个方向是决策缓存。同一用户短期内多次触发同类事件时决策结果是高度重复的。我打算把决策草稿存到 RedisTTL 设为 30 分钟减少规则评估和 LLM 调用的开销。第三个方向是多租户隔离。现在 Agent-Reach 内部按 project_id 区分逻辑隔离但后续如果作为公司级平台各个业务方的配额、通道、模板需要更强隔离。计划把生产队列和 Redis key 都按租户分片避免一个业务方的消息洪峰影响到另一个业务方。最后说一点个人体会。Agent-Reach 落地过程中我最大的收获不是写出了多优雅的代码而是重新理解了触达这件事的本质它不是一个技术动作而是一个需要策略、节奏和复盘的业务决策。先把规则引擎和通道适配做好再加 LLM 提升决策覆盖面每一步都能验证效果比一上来就追求全智能要稳妥得多。如果你也在做类似的项目我建议从一条最简单的短信通知开始跑通队列、幂等、频控这三件事再往上层加决策逻辑。
