"Agent 能听懂话"和"Agent 的话能真的送到人手里"，中间隔着一整个工程世界。我前年参与过一个内部智能助手的落地，模型侧调得很顺，评测集上的准确率也好看，结果灰度上线第一周，用户投诉最多的问题不是"答错了"，而是"压根没收到"。回头排查，消息在触达链路上被吞了三次：一次是限流器在高峰期直接丢弃请求且没有任何补偿，一次是重试逻辑把已经成功的消息又发了一遍，还有一次是某个渠道的凭据过期后静默失败，日志里只留了一行 WARN。这三件事跟模型一点关系都没有，全是触达层的问题。
Agent-Reach 想啃的就是这一层。它把"Agent 如何把结果送达到目标渠道、如何保证送到、如何知道没送到"抽象成一套可插拔的框架，用 Channel、Adapter、Reach 三层把渠道描述、协议翻译和发送编排拆开，让换渠道不用动业务代码，让排障不用靠猜。这篇内容适合两类人看：一类是已经在做 Agent 落地、被消息丢失和重复投递折磨过的后端同学；另一类是刚接手智能体项目、还没意识到触达链路有多深的开发者。我会把抽象设计、最小可跑闭环、幂等与重试的具体做法、可观测性打点、以及一次真实故障的完整排查链路都摊开讲。
1. 先把"触达"这件事拆开，别一上来就写发送函数
1.1 从"模型能答"到"消息能到"的断层在哪里
大部分人做 Agent 的第一版代码是这样的：拿到用户输入，调模型，拿到回复，调一个 SDK 把消息发出去。本地跑没问题，因为本地只有一个用户、一条消息、一次成功。上了生产就完全是另一回事：消息要发给不同的目标，目标可能在线可能离线；发送可能超时、可能被限流、可能返回一个看起来成功但其实没送达的响应；同一个事件可能被上游重复推送两次。
这些差异不是"代码写得不够好"，而是状态空间爆炸。一条消息的旅途大致有这么几个阶段：产生（Agent 决策完成）→ 路由（决定发给谁、走哪条通道）→ 适配（翻译成目标协议）→ 传输（真正落到网络上）→ 确认（对方是否收到）→ 归档（失败怎么办）。每一步都有成功、失败、超时、重复四种可能，六步组合起来就是四千多种组合，靠一个 try-except 是兜不住的。
Agent-Reach 的价值在于把这六步显式建模。一旦建模，"重试几次""幂等键用什么""失败进哪里"这些问题就从"写代码时顺手处理"变成了"配置项里明确声明"。我个人的体感是，显式建模之后，排障时间能从半天缩短到十几分钟，因为你不用再猜消息是在哪一步没的。
1.2 触达层的四个硬指标，先定指标再选方案
在写第一行代码之前，我建议先把指标定下来。这四个指标决定了你后面所有的技术选型，缺一个都会在某个时间点狠狠反噬。
| 指标 | 含义 | 常见目标值 | 对应设计决策 |
|---|---|---|---|
| 到达率 | 真正被目标端确认接收的比例 | 核心通道 99.9% 以上 | 需要确认回执与失败补偿 |
| 幂等性 | 同一业务事件重复触达不产生重复副作用 | 重复投递率 0 | 需要业务级幂等键 |
| 可观测 | 任一消息能在 1 分钟内查到完整轨迹 | 100% 可追溯 | 需要 traceId 全链路贯穿 |
| 可回滚 | 出问题能立刻停掉某个通道 | 分钟级开关生效 | 需要运行时配置而非重启 |
这里最容易翻车的是到达率定义。很多团队把"发送 API 返回 200"当作送达，实际上这只是"对方网关收下了"，离"用户看见"还差着好几跳。我的做法是分层记录：accepted（网关接收）、delivered（通道确认）、read（可选，端侧回执），告警只看前两个，第三个只做统计不做告警，因为端侧回执的丢失率天然就高，拿它做告警会让你半夜被无效告警叫醒。
还有一个反直觉的点：别把到达率目标定成 100%。定成 100% 意味着你要为最后那 0.1% 投入不成比例的资源，而且会诱导团队用"多通道同时轰炸"这种粗暴方式提指标，结果就是用户被骚扰。我一般把核心通道定在 99.9%，长尾通道定在 99%，剩下的靠死信队列和人工兜底，这个组合的性价比最高。
2. Agent-Reach 的三段式抽象：Channel、Adapter、Reach
2.1 Channel 只描述"发给谁"，不描述"怎么发"
先说我踩过的坑。我第一版设计里，Channel 这个对象里塞了 URL、密钥、超时时间、重试次数、报文模板……结果换一个渠道，整个对象要重写。后来才想明白：Channel 应该是业务维度的概念，不是技术维度的概念。
修正后的 Channel 只回答三个问题：
- 我是谁（渠道标识，比如
ops-alert、user-notify） - 我的目标是什么（目标类型：单点、群组、广播）
- 我的优先级和配额是什么（决定被限流时谁先走）
# channels.yaml channels: - id: ops-alert target_type: group priority: P0 quota: qps: 50 daily_cap: 20000 adapter_ref: internal-im - id: user-notify target_type: single priority: P2 quota: qps: 200 daily_cap: 500000 adapter_ref: mail-gateway这样做的好处很直接：业务方只需要知道有ops-alert这个渠道，不需要知道它背后走的是内部 IM 还是邮件。哪天要让用户通知从邮件切到内部 IM，改一行adapter_ref就行，调用方零感知。这个解耦看起来简单，但它把"渠道变更"从一次跨团队联调变成了一个配置发布。
注意：优先级一定要在 Channel 层定，不要留到 Adapter 层。因为限流发生在 Adapter，但它需要知道"该丢谁"，这个信息只有业务侧的 Channel 才有。
2.2 Adapter 负责协议翻译、凭据和真正的限流
Adapter 是我认为整个框架里最值得投入的部分。它的职责边界很清晰：把统一的消息模型翻译成具体协议，并把出网的脏活全干完。
统一消息模型我一般定义成这样，字段不多但每个都有用：
from dataclasses import dataclass, field from typing import Any @dataclass class ReachMessage: biz_key: str # 业务幂等键，全局唯一 channel_id: str target: str # 目标标识，群 ID / 用户 ID / 地址 title: str body: str payload: dict[str, Any] = field(default_factory=dict) priority: str = "P2" trace_id: str = "" expire_at: int = 0 # 过期时间戳，过期直接丢弃expire_at这个字段是我强烈建议加的，而且是被最多人忽略的一个。想象一下：一条"服务 CPU 超过 90%"的告警，因为重试延迟了四十分钟才发出去，这时候人已经在处理了，这条消息只会造成干扰。带过期时间的消息在 Adapter 出网前做一次检查，过期就丢进"过期箱"而不是死信，避免污染真正的失败队列。
Adapter 内部要做四件事，按顺序：
- 凭据管理：密钥从配置中心或环境变量注入，进程内缓存但带 TTL，避免每次发送都去拉。
- 协议翻译：把
ReachMessage转成目标格式，这一步应该是纯函数，不碰 IO，方便单测。 - 限流：本地令牌桶 + 全局配额双重控制。本地桶控瞬时，全局配额控长期。
- 出网与结果归一：把各种奇怪的返回码统一成
SUCCESS、RETRYABLE、FATAL、RATE_LIMITED四种。
class InternalImAdapter: def __init__(self, token_bucket, credential_provider): self.bucket = token_bucket self.creds = credential_provider def send(self, msg: ReachMessage) -> str: if not self.bucket.acquire(timeout=0.2): return "RATE_LIMITED" wire = self._translate(msg) # 纯函数 resp = self._post(wire, self.creds.get()) return self._classify(resp.status_code) # 归一化把返回码归一化成四种，是我认为最能提升排障效率的一个设计。因为重试策略只需要针对RETRYABLE和RATE_LIMITED做，FATAL（比如参数错误、目标不存在）重试一百次也没用，直接进死信。不归一化的话，你会写出一堆针对不同错误码的 if-else，最后没人敢改。
2.3 Reach 层的编排：路由、重试、去重、熔断
Reach 是大脑，也是最容易写乱的一层。我的经验是把它当成一个小型的消息中间件来设计，而不是当成一个函数调用链。
路由部分要做的事：根据channel_id找到 Channel 配置，找到对应的 Adapter 实例，检查配额是否还有余量，检查消息是否过期。这几步全是内存操作，耗时应该控制在毫秒级。
重试部分我固定用指数退避加抖动，参数是初始 500ms、倍数 2、最大 30s、最多 4 次，然后再叠一个 ±20% 的随机抖动。抖动这一步千万不能省，我见过一次故障就是重试没有抖动，三千条消息在同一个 500ms 窗口集体重试，把刚恢复的下游又打挂了。加了抖动之后，重试流量在时间轴上自然摊开，下游压力曲线会平缓很多。
去重部分靠biz_key加一层短期缓存，一般用本地 LRU 加共享存储的双层结构。本地 LRU 挡掉绝大部分重复，共享存储（缓存或数据库唯一索引）兜住跨实例的情况。
熔断部分比较简单：按 Adapter 维度统计最近一分钟的失败率，超过 50% 且样本数大于 20 就打开熔断，半开状态放 5% 的流量试探。熔断打开期间消息不进死信，而是进"延迟队列"，等熔断恢复后重新投递，避免下游抖动导致大量消息被判定为永久失败。
3. 跑通第一条链路：从零到能用的最小闭环
3.1 目录结构和依赖准备
我不建议一上来就上分布式组件。最小闭环用进程内的队列加 SQLite 就够跑通了，验证完再替换。我实际用的目录结构是这样的：
agent-reach/ config/ channels.yaml adapters.yaml reach/ __init__.py model.py # ReachMessage 定义 router.py # 路由 retry.py # 退避与抖动 dedup.py # 幂等 breaker.py # 熔断 adapters/ base.py internal_im.py mail_gateway.py store/ delivered.sqlite tests/ replay/ fixtures.jsonl依赖很少，核心就三个：一个 HTTP 客户端、一个 YAML 解析、一个本地存储。这里的选型逻辑是：在验证阶段引入越少组件，你越容易判断问题出在哪。我见过不少团队，第一版就上了消息队列加 Redis 加配置中心，结果联调两天，最后发现是自己代码里消息体拼错了。组件越多，排查路径越长。
3.2 写一个自定义 Channel 插件：从 Adapter 基类继承
自定义渠道的接入流程我设计成三步，任何新渠道都走同样的路：
- 继承
BaseAdapter，实现send和_classify。 - 在
adapters.yaml里注册这个 adapter。 - 在
channels.yaml里加一条 Channel 指向它。
from adapters.base import BaseAdapter class MailGatewayAdapter(BaseAdapter): name = "mail-gateway" def _classify(self, code: int) -> str: if 200 <= code < 300: return "SUCCESS" if code in (408, 429) or code >= 500: return "RETRYABLE" if code == 429: return "RATE_LIMITED" return "FATAL" def send(self, msg): if not self.bucket.acquire(timeout=0.2): return "RATE_LIMITED" # 真正的发送逻辑 ...写_classify的时候有个经验：把"未知错误"默认归到FATAL还是RETRYABLE，是个需要想清楚的决策。默认RETRYABLE更安全（不会丢消息），但风险是遇到一个永久性错误时会白白重试四次、占用配额、还可能触发熔断。我的做法是默认FATAL，但凡是未知错误码都强制打一条 ERROR 日志并带上完整响应体，这样上线前几天盯一下日志，把实际出现的错误码补进分类表，一周之后基本就准了。
3.3 凭据和路由放配置里，别硬编码进代码
这一点看起来是常识，但每年还是能看到把密钥提交进仓库的事故。Agent-Reach 的配置我建议分两层：结构性配置（channels、adapters、重试参数）放配置文件进仓库；敏感性配置（密钥、token）走环境变量或配置中心，代码里只留占位符。
# adapters.yaml adapters: - name: internal-im endpoint: "https://im.internal.example/api/v2/message" credential_env: "IM_TOKEN" # 只写变量名，不写值 timeout_ms: 3000 retry: max_attempts: 4 base_ms: 500 multiplier: 2 max_ms: 30000 jitter: 0.2把重试参数也放进配置的好处是：线上出问题时你能改参数而不用发版。有一次我们的下游通道在维护窗口期间响应特别慢，超时从 3 秒变成 8 秒，我们直接把这个 adapter 的timeout_ms临时调到 10000、max_attempts降到 2，五分钟内就把重试流量压下去了。如果这些参数写死在代码里，那个晚上就得走一遍完整发布流程。
4. 生产环境真正会咬人的三个坑
4.1 重试风暴：下游刚恢复就被二度打挂
重试风暴的形成过程通常是这样的：下游因为负载高开始变慢 → 上游超时 → 上游重试 → 下游负载更高 → 更多超时 → 更多重试。这个正反馈循环可以在三分钟内把一个小问题放大成全面故障。
我在一次故障里亲眼看过这个曲线：下游 QPS 从 200 涨到 1400，其中 1100 都是重试流量。事后复盘，问题出在两个地方：一是没有抖动，二是重试没有全局预算。
全局重试预算是我后来加的，规则很简单：整个进程每秒钟允许的重试次数不超过正常发送量的 10%。超预算的重试请求不丢弃，而是塞进延迟队列等下一轮，这样既保护了下游，又不丢消息。
class RetryBudget: def __init__(self, ratio=0.1, window=60): self.ratio = ratio self.window = window def allow(self) -> bool: # 滑动窗口内 retry_count / total_count <= ratio ...还有一点：熔断和重试必须联动。单靠熔断不够，因为熔断打开之前的那几十秒，重试已经在放大流量了；单靠重试预算也不够，因为预算只能限速不能止血。两个一起上，效果才明显。
4.2 幂等键怎么设计才不会撞车也不会漏
幂等键是整个触达层里最难设计的部分，因为它要同时满足两个矛盾的要求：同一件事必须算出同一个键（防重复），不同的事必须算出不同的键（防误合并）。
我试过三种方案，最后选了第三种：
| 方案 | 组成 | 问题 |
|---|---|---|
| 时间戳 | 消息生成毫秒数 | 重试时时间变了，无法去重 |
| UUID | 随机生成 | 上游重复推送会生成两个 UUID，去重失效 |
| 业务键 | 业务实体 + 事件类型 + 版本 | 需要上游配合，但唯一可靠 |
业务键长这样：order-88213:status_changed:v3。它的来源是业务系统里的稳定标识，不由触达层生成。关键点在于幂等键必须由最上游的事件产生方生成并透传下来，如果让触达层自己算，它永远不知道该用哪个字段。
这里有个容易踩的坑：幂等键加 TTL 的时长。加太短，重复投递发生在 TTL 之后就防不住了；加太长，存储成本高，而且某些业务确实会合法地重复（比如用户手动重新发送）。我一般按业务特性定：告警类 24 小时，通知类 7 天，交易类永久（落库唯一索引）。这个时长是配置项，不是常量。
4.3 限速、优先级队列和静默期如何协同
限流器看起来简单，实际用起来最麻烦的是"多维度同时生效"。一个消息可能同时受本地 QPS 限制、全局日配额限制、目标用户的接收频率限制、以及业务定义的静默期限制。
我用的检查顺序是：静默期 → 用户频控 → 全局配额 → 本地 QPS。顺序不能乱，因为前面的检查最便宜（内存查表），后面的检查最贵（可能涉及跨实例的计数）。把便宜的放前面，能挡掉大部分无效请求。
静默期的实现有个细节：静默期内的消息是丢弃还是延迟。大部分场景应该延迟，因为"晚上十点到早上八点不发通知"的意思是"早上八点再发"，不是"这条通知不要了"。但延迟的话就要面对"消息堆积到早上八点集体释放"的问题。我的做法是把释放时间再打散，每条消息随机延迟 0 到 15 分钟，避免整点尖峰。
def apply_quiet_hours(msg, now, quiet=(22, 8)): if in_quiet(now, quiet): release_at = next_release_time(now, quiet) jitter = random.randint(0, 900) msg.schedule_at = release_at + jitter return "DEFERRED" return "PASS"提示：静默期一定要区分渠道。运维告警渠道通常不应该有静默期（半夜也得叫醒人），用户通知渠道必须严格静默。这个差异在 Channel 配置里用一个
quiet_hours字段控制，别写死在代码里。
5. 没有链路的日志等于没有日志
5.1 traceId 必须贯穿 Channel、Adapter、Reach 三层
我一开始以为只要在入口生成一个 traceId 打日志就够了，后来发现完全不够。因为一次触达会跨越异步边界：Reach 入队、后台 worker 消费、Adapter 出网、结果回调。如果 traceId 只存在于入口的请求上下文里，一旦进了队列就断了。
解决办法是把 traceId 放进消息体，而不是放在线程上下文里。ReachMessage里的trace_id字段跟着消息走，从入队到出网到归档，每一步都从消息里读出来打日志。
logger.info("adapter.send.start", extra={"trace_id": msg.trace_id, "biz_key": msg.biz_key, "channel": msg.channel_id, "attempt": attempt})这样你在日志系统里用 trace_id 一搜，就是一条完整的时间线：什么时候入队、路由到了哪个 adapter、第几次尝试、返回了什么状态、最后落在哪里。这个能力在排查"消息去哪了"这类问题时是决定性的。
5.2 三个必须打点的指标，少一个都会瞎
指标不用多，但下面三个必须有，而且必须有告警：
- 到达率分通道统计：按 channel_id 分组，分子是
SUCCESS，分母是总发送数。这个指标的下降通常是第一个信号。 - 重试率：重试次数除以总次数。超过 15% 就说明下游有问题，比例持续上升说明重试风暴可能在酝酿。
- 队列深度与最老消息年龄：这两个一起看。深度大但年龄小，说明消费快、生产更快，扩容就行；深度大且年龄大，说明卡住了，得查具体卡在哪。
我特别想说第三个指标里的"最老消息年龄"。这个数字比队列深度有用得多，因为它直接告诉你"用户最长等了多久"。队列深度一万条可能是正常波动，但最老消息年龄是 40 分钟，那就一定是出事了。
5.3 日志采样，别让可观测性把成本干爆
全量打日志在小规模时没问题，量上来之后日志成本会很吓人。我的采样策略分三类：
- 成功路径：1% 采样，或者只打指标不打日志。
- 失败路径：100% 全打，包括完整响应体。
- 状态跃迁：100% 全打，比如从
RETRYABLE到FATAL、从熔断打开到半开。
这个策略的逻辑是：成功的消息长得都一样，失败的消息各有各的失败。成功案例打一万条不会带来新信息，失败案例打一条可能就定位了根因。我们用这套策略之后，日志量降了大概七成，但排障能力基本没损失。
6. 扩容路径：从单机到多实例的三个改造点
6.1 无状态化改造：把状态从进程里挪出去
单机版的 Agent-Reach 会把幂等缓存、限流计数、熔断状态全放在进程内存里。多实例一上，这些状态各自为政，幂等失效、配额翻倍、熔断误判，问题会集中爆发。
改造的核心原则是区分"可以本地"和"必须共享"：
| 状态 | 存储位置 | 理由 |
|---|---|---|
| 幂等去重（短期） | 本地 LRU + 共享缓存 | 本地挡大部分，共享兜底 |
| 本地 QPS 限流 | 进程内存 | 每实例独立配额，天然分摊 |
| 全局日配额 | 共享计数 | 必须精确，跨实例 |
| 熔断状态 | 进程内存 | 每个实例视角独立，反而是优点 |
| 消息归档 | 共享数据库 | 需要全局查询 |
熔断状态留在本地是我做的一个反直觉选择。一开始我也觉得应该全局共享，后来发现本地熔断更稳定：因为每个实例看到的失败率略有差异，本地熔断天然形成了"部分实例先退避、其余继续"的梯度效果，不会出现全局同时熔断、同时半开造成的震荡。
6.2 长连接与连接池的取舍
如果渠道是 HTTP，用连接池就够，但要调好最大连接数和空闲回收。经验值是：最大连接数设为预期 QPS 的 1.5 倍再除以预估单连接 QPS，别直接设成几百。我曾经把一个池设到 500，结果下游的连接数被打满，反而变慢。
如果渠道是长连接（比如某些推送通道），就要面对重连和心跳。这里的坑是重连风暴：网络抖动导致大量实例同时断开、同时重连，握手流量把网关压垮。解决办法是在重连延迟里加基于实例 ID 的抖动，让重连在时间上散开。
delay = base_delay * (2 ** attempt) delay += hash(instance_id) % 1000 / 1000 * base_delay6.3 批量合并：把一百条通知压成一条摘要
当触达量变大之后，单个用户的定向通知会变得很碎：同一个用户十分钟内收到八条不同系统的通知。这时候批量合并就很有价值了，但要注意合并只适合非紧急的中低优先级内容。
我的做法是在 Reach 层加一个可选的合并窗口：同一个target在 30 秒窗口内的P2消息合并成一条摘要，P0消息不参与合并直接发。合并后的消息会标记merged_count，方便统计。这个改动让用户的日均消息量降了六成，但重要信息的到达率没有变化。
一个必须注意的点：合并后幂等键要变。合并产生的是一个新消息，不能沿用任何一条原消息的biz_key，否则去重逻辑会把后续的正常消息误判为重复。我用的规则是merge:{target}:{window_start_ts}。
7. 一次到达率骤降的完整排查过程
7.1 现象：分通道到达率从 99.7% 掉到 91%
某天下午两点十分，告警响了：user-notify渠道的到达率跌破 95% 阈值，当前 91.2%，持续五分钟。同时重试率从 3% 涨到 22%。
第一反应是下游挂了，但看了一下下游的状态页，一切正常。这就是触达层排查的典型开局：外部一切正常，问题一定在我们自己这边。
7.2 逐层排除：从出口往回倒着查
我的排查习惯是从出口往回查，因为出口最接近现象，信息最具体。具体走了五步：
- 看失败样本的错误分类分布。统计发现 87% 是
RETRYABLE，13% 是RATE_LIMITED，没有FATAL。这说明不是参数错误，是时序问题。 - 看重试时间分布。把重试记录的时间戳画出来，发现重试高度集中在每小时的整点后 10 到 30 秒。这个形状很可疑。
- 查整点在干什么。查了定时任务列表，发现有一个报表任务在整点批量生成发送请求，每分钟约 8000 条。
- 看限流器配置。
user-notify的 qps 设的是 200，也就是每秒 200 条，而整点突发的 8000 条消息要在几秒内发完，必然大量触发RATE_LIMITED。 - 看重试与限流的交互。被限流的消息会重试，重试又去抢同一个令牌桶，把桶占满，导致正常流量也被限流，形成恶性循环。
到这里根因就清楚了：报表任务的突发流量挤压了正常流量，而重试逻辑和限流器共享同一个令牌桶，放大了挤压效应。
7.3 修复：三处改动，当天上线
改动不大但都关键：
- 限流桶按流量来源隔离。正常业务流量和批量任务各用一个桶，批量任务的桶配额单独配置，避免互相挤压。
- 被限流的消息走延迟队列而不是立即重试。延迟时间设为
2^attempt秒加抖动，第一次延迟 2 秒，避免立刻回来再抢。 - 报表任务改成匀速发送。原本是一次性投递 8000 条，改成用令牌桶控制每秒 100 条，80 秒发完。
改完之后到达率半小时内回到 99.8%，重试率降到 4%。
7.4 这次故障之后我固定加的三样东西
第一样是来源隔离。只要是共享资源（限流桶、连接池、重试预算），就按流量来源打标签隔离。共享资源的好处是利用率高，坏处是一个来源能把所有人拖下水。隔离之后利用率略降，但稳定性提升明显。
第二样是限流与重试的显式联动。被限流不应该原地重试，而应该退到延迟队列，并且延迟要明显长于正常重试。这个规则我后来写进了框架的默认行为，不再是每个 adapter 自己决定。
第三样是批量任务的准入检查。任何批量任务在发送前要先查询当前通道的实时水位，如果已经在 70% 以上就先等一会儿。这个检查加在批量任务侧而不是触达层，因为触达层不应该知道业务侧的批次概念。
一点个人体会：触达层的问题几乎从来不是"某个功能没实现"，而是"几个本来合理的机制互相放大了对方的副作用"。限流是合理的，重试是合理的，突发批量也是合理的，三个凑在一起就成了故障。所以做 Agent-Reach 这类框架，最值钱的部分不是功能清单，而是那些把机制之间相互作用显式约束住的设计——共享资源要隔离，重试要有预算，延迟要带抖动，状态要能观测。这几条守住了，剩下的就是填渠道适配器，工作量不大，也不会再半夜被叫起来。
