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、Reach2.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_factorydict) 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(timeout0.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(timeout0.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, ratio0.1, window60): 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% 是RETRYABLE13% 是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 这类框架最值钱的部分不是功能清单而是那些把机制之间相互作用显式约束住的设计——共享资源要隔离重试要有预算延迟要带抖动状态要能观测。这几条守住了剩下的就是填渠道适配器工作量不大也不会再半夜被叫起来。
