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。对比普通 List，Streams 支持消费组、消息 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(undefined=StrictUndefined, autoescape=False)很多触达翻车事故就是这么来的：模板里写了{{ 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(seconds=delay + 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 = f"freq: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 埋点。我想把每个环节的耗时、决策命中规则、渠道响应码都变成结构化追踪，这样排查问题时可以直接看到一条消息从事件到发送的全链路。
第二个方向是决策缓存。同一用户短期内多次触发同类事件时，决策结果是高度重复的。我打算把决策草稿存到 Redis，TTL 设为 30 分钟，减少规则评估和 LLM 调用的开销。
第三个方向是多租户隔离。现在 Agent-Reach 内部按 project_id 区分逻辑隔离，但后续如果作为公司级平台，各个业务方的配额、通道、模板需要更强隔离。计划把生产队列和 Redis key 都按租户分片，避免一个业务方的消息洪峰影响到另一个业务方。
最后说一点个人体会。Agent-Reach 落地过程中，我最大的收获不是写出了多优雅的代码，而是重新理解了"触达"这件事的本质：它不是一个技术动作，而是一个需要策略、节奏和复盘的业务决策。先把规则引擎和通道适配做好，再加 LLM 提升决策覆盖面，每一步都能验证效果，比一上来就追求"全智能"要稳妥得多。如果你也在做类似的项目，我建议从一条最简单的短信通知开始，跑通队列、幂等、频控这三件事，再往上层加决策逻辑。
