1. 从“会聊天的模型”到“能落地的触达”Agent-Reach要解决的真实问题去年年中我接手了一个挺典型的智能体项目模型层已经跑通了Agent能根据用户意图生成完整的响应方案也能调用内部工具查数据、算价格但到了真正把结果“送到”用户面前那一步整个系统就像被掐住了脖子。邮件服务商拒信、短信通道限流、IM机器人消息发出去用户收不到、回调地址偶尔失联……你说模型能力不行吧它明明能把事情想得很明白你说系统架构有问题吧每个环节单拎出来都有日志、有监控。可合在一起就是不稳定。我一直觉得2024年之后做Agent相关的系统真正拉开差距的地方早就不是“哪个模型聪明”而是“Agent的最后一公里”——触达。你让Agent写一封完美的客户挽回邮件它确实能写但这封邮件能不能在正确的时间、通过用户偏好的渠道、以不触发频控的方式、稳定地进入用户收件箱并且收到回执后还能让Agent自己判断这轮沟通是否有效这才是工程问题。Agent-Reach这个名字说到底就是把“智能体Agent”和“触达Reach”焊在一起让Agent的产出真正抵达它该抵达的地方。我一开始也天真地以为接几个渠道API、套一层发送封装就算完事了。真正动手之后才发现触达这件事涉及到一整套工程体系渠道适配、策略编排、状态回执、频控熔断、幂等去重、用户偏好管理。任何一个环节偷懒线上就会用事故回报你。这篇文章不聊模型微调、不聊提示词工程只聊Agent在触达侧的系统设计。如果你也在做智能体产品、自动化运营系统或者想把内部的AI能力接出去服务真实用户这篇内容应该能帮你少走不少弯路。整个触达链路如果拆开看实际上一共有六个环节Agent生成触达内容正文、标题、摘要策略层决定“用什么渠道、什么时间、什么频率”触达路由层根据策略选择可用的通道实例适配器把统一消息格式翻译成渠道方要求的格式渠道方投递后返回状态回执成功、失败、已读、退订等回执数据回流供策略引擎调优。Agent-Reach的核心就是把这六个环节串成一个可观测、可干预的闭环。任何一个环节都不难真正难的是把它们“串”起来的那一刻各种边界问题全冒出来了。2. Agent-Reach的核心架构Agent层、策略层、通道层三层解耦2.1 三层各管各的事别让它们互相串门我在设计Agent-Reach的架构时第一原则就是隔离。整个系统分成三层最上面是Agent层中间是策略层最下面是通道层。三层之间通过明确的数据结构通信谁都不直接依赖谁。Agent层只负责“生成什么”。它接收任务上下文产出触达内容的正文、标题、摘要、附件索引以及一个初步的意图标签比如“催收提醒”“活动邀约”“异常告警”。Agent层完全不关心这些内容走什么渠道。策略层只负责“怎么送”。它读取触达请求结合用户偏好、频控状态、渠道健康度、时段规则最终输出一个路由决策。通道层只负责“送出去”。它维护各个渠道的连接、限流参数、请求签名、回执解析。通道层不关心内容好不好只关心格式对不对、能不能送达。我用一张表说明各层的关注点层核心输入核心输出直接依赖Agent层任务上下文、用户画像摘要触达内容包、意图标签模型服务、知识库策略层触达请求、用户偏好、频控状态路由决策、渠道优先级策略配置中心、状态存储通道层统一触达消息渠道侧请求、回执数据各渠道API、通道配置之所以坚持三层解耦是因为我见过太多把渠道API直接塞进Agent代码里的做法。开始很爽第二个渠道接入的时候也还行等渠道一多Agent的主逻辑里全是if-else分支每个渠道的鉴权方式、重试机制、回执格式全堆在一起改一个渠道的状态解析都可能影响Agent的核心判断。这就像你把电话线、网线、电线全捆在一起走可能勉强能用但任何一路出问题排查起来都是灾难。2.2 核心数据结构就是三张“单据”三层之间靠三个核心数据结构通信我习惯叫它们三张“单据”。第一张是触达请求单TouchRequestAgent层生成策略层消费第二张是路由决策单RouteDecision策略层生成通道层消费第三张是回执单DeliveryReceipt通道层生成回流到策略层。dataclass class TouchRequest: request_id: str # 全局唯一幂等键的基础 agent_id: str # 哪个Agent发起的 user_id: str # 触达对象 content_pack: dict # 正文、标题、摘要、附件索引 intent_label: str # 意图标签比如 promotion、alert urgency_level: int # 1-5紧急程度影响策略 dataclass class RouteDecision: request_id: str channel: str # 最终选择的渠道标识 template_id: str # 命中哪套消息模板 estimated_cost: float # 预估成本用于预算控制 retry_policy: dict # 重试参数 fallback_order: list # 降级顺序 dataclass class DeliveryReceipt: request_id: str channel: str status: str # delivered / failed / blocked / read provider_message_id: str # 渠道侧的消息ID raw_payload: dict # 渠道返回的原始内容 timestamp: int这三张单据的设计有个好处每一层只消费自己需要的字段。比如策略层看到urgency_level等于5就知道要优先走最快通道、跳过非紧急时段的限流规则Agent层看到DeliveryReceipt里status是blocked就知道这轮触达被用户或系统拦截了下次要换措辞、换渠道、换时间。2.3 路由逻辑别写得像玄学要能解释路由是策略层最核心的模块。它不能是“感觉这个渠道该不行了随便换一个”而必须是可解释的、可审计的决策过程。我通常把路由做成一个可组合的策略链先过滤掉不可用的通道比如健康度低、已经被熔断、或者用户设置了屏蔽再对剩余通道打分最后按分数选路。分数计算我一般用这五个因子$$score w_1 \cdot speed w_2 \cdot reliability w_3 \cdot engagement - w_4 \cdot cost - w_5 \cdot frequency_penalty$$speed是历史平均送达耗时reliability是最近24小时成功率engagement是用户对该渠道的历史交互率cost是单条消息成本frequency_penalty是该渠道在这个时间段被使用的频次压力。权重可以做成动态的但这个版本先把权重固定用一段时间再调。打分制比硬编码路由规则灵活得多。比如短信成本涨了我只需要调w_4不需要改判断逻辑。再比如某个渠道最近老是超时reliability自然下降系统会自动减少它的命中概率——这是我认为策略层最实用的一个设计。3. 通道接入的工程细节用统一适配器收编邮件、短信、IM和Webhook3.1 通道数量一多适配器就得有“标准姿势”做完三层拆解之后最花时间的其实是通道层。我一开始就定了规矩所有渠道必须实现同一个适配器接口谁不遵守谁就别想上线。这个接口不能太薄否则每个渠道的差异还是裸露在策略层面前也不能太厚太厚的统一语义会抹掉渠道特性。BaseChannelAdapter的标准接口如下from abc import ABC, abstractmethod class BaseChannelAdapter(ABC): channel_name: str base abstractmethod def configure(self, config: dict): 加载渠道配置鉴权信息、端点、默认限额 abstractmethod def serialize(self, message: dict) - dict: 把统一消息体翻译成渠道方要求的格式 abstractmethod def send(self, payload: dict) - dict: 真正调用渠道API返回渠道原始响应 abstractmethod def parse_receipt(self, raw: dict) - DeliveryReceipt: 解析渠道回调/回执归一化成标准回执单 abstractmethod def health_check(self) - dict: 健康检查返回延迟、当前配额余量、错误状态拿邮件、短信、IM、Webhook四种典型渠道来说各自的实现侧重点很不一样我整理了一个对比表渠道类型鉴权方式限流特征回执复杂度常见失败邮件SMTP/APIAPI Key/OAuth按日配额为主有退信、打开回执异步延迟高退信、被识别为垃圾邮件短信签名模板ID按秒/QPS限流状态回执通常延迟30秒以上签名不匹配、模板变量超长IM机器人Token签名校验按分钟频控消息可达但无法保证已读主动消息被限制、被用户拉黑WebhookHMAC签名目标侧说不准需自己定义ack机制目标服务不可达、超时适配器接口里最容易踩坑的是parse_receipt。每个渠道的回执字段千奇百怪邮件的回执可能是异步邮件通知短信回执是一个状态码加时间戳IM的回执跟消息事件混在一起。如果不做归一化策略层就得写一大堆if-else来判断状态这等于把通道层的差异又泄漏到了上层。3.2 一个Webhook适配器的完整实现示例Webhook是最常用的渠道之一比如企业内部通知、对接第三方系统实现起来也最能看明白适配器的设计套路。我直接给出一版可用代码import hashlib import hmac import time import requests from adapters.base import BaseChannelAdapter class WebhookAdapter(BaseChannelAdapter): channel_name webhook def configure(self, config: dict): self.endpoint config[endpoint] self.secret config[secret].encode(utf-8) self.timeout config.get(timeout, 10) def serialize(self, message: dict) - dict: # 统一消息体 - 目标系统要求的格式 return { event: message.get(event_type, agent_touch), payload: { title: message.get(title, ), content: message.get(body, ), message_id: message[request_id], agent_id: message.get(agent_id), timestamp: int(time.time()), }, } def send(self, payload: dict) - dict: body payload body_bytes self._json_bytes(body) signature hmac.new( self.secret, body_bytes, hashlib.sha256 ).hexdigest() headers { Content-Type: application/json, X-Reach-Signature: fsha256{signature}, X-Reach-Timestamp: str(int(time.time())), } resp requests.post( self.endpoint, jsonbody, headersheaders, timeoutself.timeout ) return {status_code: resp.status_code, body: resp.text} def parse_receipt(self, raw: dict) - DeliveryReceipt: # 目标服务通过回调同步返回受理结果 return DeliveryReceipt( request_idraw[request_id], channelself.channel_name, statusdelivered if raw.get(accepted) else failed, provider_message_idraw.get(provider_msg_id, ), raw_payloadraw, timestampint(time.time()), ) def health_check(self) - dict: latency self._measure_latency() return { ok: latency self.timeout, latency_ms: latency, quota_remaining: None, } staticmethod def _json_bytes(obj: dict) - bytes: import json return json.dumps(obj, separators(,, :)).encode(utf-8) def _measure_latency(self) - float: start time.time() try: requests.get(self.endpoint, timeoutself.timeout) except requests.RequestException: return float(self.timeout * 1000) return (time.time() - start) * 1000这里有个细节我得提醒一下HMAC签名的timestamp一定要带上并且目标服务端要校验时间戳偏移量。不然你的签名请求被捕获重放目标服务根本区分不出来。我们内部统一要求时间戳偏差超过5分钟的请求直接拒收。3.3 通道健康度与自动熔断别让慢通道拖死全链路适配器只解决了“会用这个渠道”解决不了“这个渠道还能不能用”。所以我给每个通道实例配了一个健康状态机状态有三种健康healthy、降级degraded、熔断open。熔断的判断逻辑不复杂我用的是滑动窗口统计窗口长度5分钟每次发送后记录结果成功/失败/超时失败率超过40%且样本量大于20状态切到degraded连续3个窗口处于degraded切到openopen状态下不再路由任何新消息只保留一个“半开探测”通道每30秒放一条测试消息探活。熔断最大的价值不是“不发消息”而是“快速失败”。如果你每条消息都要等到渠道超时比如20秒才知道失败一旦某个渠道出问题所有触达请求都会被拖住积压的消息越来越多。熔断机制让策略层在路由决策阶段就直接绕开不可用的通道整个系统的吞吐几乎不受影响。4. 触达策略编排频控、退避、降级和人工兜底的配置实践通道层稳定之后真正的业务考验在策略层。我见过不少系统通道接得挺好结果Agent太“勤奋”一天给同一个用户发十几条通知用户直接拉黑。触达策略说白了就是给Agent的“触达冲动”上一套规矩。4.1 频率控制不能依赖Agent的自觉频控的第一原则是全局限流不能单看单个Agent。同一个用户既接收跟进的触达也接收营销的触达还接收系统告警的触达。如果每个Agent各管各的频控累计起来照样爆炸。所以在Agent-Reach里频控是按user_id为维度做的规则配置在策略层frequency_limits: - user_scope: all rule_id: global_daily max_touches: 5 period: 24h except_intents: [alert] - user_scope: all rule_id: same_intent_cooldown max_touches: 1 period: 6h match: intent_label - user_scope: premium rule_id: vip_bump max_touches: 10 period: 24h我特别解释一下except_intents。规则里把alert类消息排除在全局限频之外是因为告警不能等。但为了避免Agent滥用“告警”这个标签策略层会校验intent_label是不是真的由告警触发链路生成而不是Agent自己贴上去的。这个校验逻辑在策略层入口统一做Agent无权直接指定一个消息“免限流”。4.2 退避重试参数不是拍脑袋定的触达失败之后要不要重试很多人第一反应是“重试呗失败了就再来一次”。但无脑重试会把一次瞬时故障放大成渠道雪崩。我使用的重试策略是“指数退避抖动”Exponential Backoff with Jitter核心公式$$delay_n \min(base \cdot 2^{n-1}, max_delay) random_jitter$$推荐一组初始参数初始间隔30秒最大间隔15分钟重试次数上限5次抖动范围0%~25%。举个例子第一次失败后延迟30秒重试第二次失败后延迟60秒第三次120秒第四次240秒第五次480秒加上抖动最长也就15分钟左右。超过5次进入人工兜底队列不再自动重试。我见过有人把初始间隔设成1秒、重试20次的配置小流量测试看不出来一到线上高峰一个渠道抖动重试风暴直接把自己服务打挂。重试不是越多越好重试的本质是“等系统自己恢复”而不是“逼系统恢复”。4.3 降级链路从最优渠道平滑切到可用渠道降级和重试是两回事。重试是同一个渠道再来一次降级是换渠道。Agent-Reach的策略引擎支持配置降级顺序比如route_plan: primary: app_push fallbacks: - email - sms condition: app_push_health: healthy意思是优先走App推送如果App推送渠道不健康熔断或降级自动落到邮件邮件也不行了再走短信。短信一般单价最高、打扰感最强作为最后手段。降级逻辑里最容易犯的错是“无感降级导致语义不匹配”。比如用户本来在等一个App推送的验证码结果你降级成邮件用户没看到还反复催。所以在Agent-Reach里降级不是简单换通道而是换通道的同时调整内容模板——降级到邮件时邮件标题里会明确标注“验证码”正文一开头就重复关键信息。4.4 人工兜底队列让机器知道“我搞不定了”自动触达再怎么设计总会有搞不定的情况。我在策略层里放了人工兜底队列满足任一条件就进队列同一request_id连续失败超过5次用户长时间未响应且urgency_level为1最高优先级触达内容命中敏感词规则需要人工审核后再发。兜底队列不追求自动处理追求的是“有迹可循”。每一条进队列的记录都会附带完整的触达历史包括哪一步失败、失败原因、已经重试了几次。运营人员只需要打开工单看一眼决定是人工补发还是放弃不需要再去日志里翻上下文。5. 生产中踩过的坑回执丢失、幂等、渠道限流的完整排查链路理论说了一堆我想重点讲三个真实踩过的坑。这三个问题有个共同特点不在高并发下跑几天根本暴露不出来一旦暴露就是线上事故。5.1 回执丢失导致重复触达一次“幽灵重发”事故事故现象某个用户在三分钟内收到了两条一模一样的营销短信。问题排查链路是这样的第一步我先去看了Agent层的日志确认Agent只生成了一次触达内容request_id是唯一的。这说明问题不在上层。第二步看策略层的路由日志发现RouteDecision也只产生了一条记录channel是sms没有重试记录。第三步看短信通道适配器的发送日志发现requests.post调了一次返回了HTTP 200响应里带着provider_message_id。第四步问题来了。短息服务商的回执推送是异步的回调接口把回执推到我们的Webhook端点。那天凌晨我们的回调服务发了一次版本重启期间丢失了部分回执通知。策略层没等到回执以为消息“发送状态未知”触发了超时重发逻辑于是又发了一遍。根因很清楚我没有对“重发”这件事做幂等控制。消息实际上是成功送达的只是回执丢了。所以Agent-Reach把所有触达请求都强制要求带上request_id在通道层落一张“已发送流水表”。每次重发之前先查流水表如果同一个request_id已经有成功的发送记录直接返回“已送达”绝不二次调用渠道API。5.2 幂等键设计不能只用“用户内容”当唯一键幂等键的选型也踩过小坑。第一版我用的是user_id content_hash后来发现内容一长hash碰撞概率虽然低但同一个用户两次合法触达如果文案恰好相同比如每月固定的账单提醒就会被误判成重复导致第二次提醒被吞掉。后来统一改成request_id作为幂等键request_id由Agent层在生成内容时用UUID生成全局唯一。流水表结构大概是字段说明request_id全局唯一触达请求ID主键user_id用户标识channel实际发送渠道provider_message_id渠道侧消息IDstatusdelivered / failed / unknownfirst_sent_at首次发送时间last_attempt_at最近一次尝试时间attempt_count尝试次数status为delivered的记录会被当作“终态”缓存任何重试逻辑看到这个状态都会直接短路。提示幂等表要保留足够长的历史不能只留当天。有些渠道的回执延迟特别离谱短信回执延迟几小时都有如果当天清理跨天可能重复。5.3 渠道QPS限流本地速率限制要永远比服务商严格另一个印象深刻的问题是我们的邮件通道。邮件服务商给我们的配额是每分钟600封刚开始日发送量小根本没摸到过限制。后来双十一活动Agent批量生成个性化邮件发送量突增结果几秒钟内把600封的窗口打满服务商直接返回429后续请求全部被拒。当时我以为是服务商的问题想着“等它限流恢复就可以了”但实际上服务商是按“滑动窗口”算的我们突刺一次窗口期内的所有请求都会被打回。第二次我学乖了在适配器里加了本地速率限制用令牌桶算法桶容量是服务商配额的80%填充速率也是80%。import time class TokenBucket: def __init__(self, capacity, fill_rate): self.capacity capacity self.fill_rate fill_rate self.tokens capacity self.updated_at time.monotonic() def acquire(self, tokens1) - bool: self._refill() if self.tokens tokens: return False self.tokens - tokens return True def _refill(self): now time.monotonic() elapsed now - self.updated_at self.tokens min(self.capacity, self.tokens elapsed * self.fill_rate) self.updated_at now这个设计让我把重试率降了90%。之所以要留20%余量是因为服务商的限流窗口和我们的本地时钟不可能完全同步留余量是给系统偏差一个缓冲。同样的思路也适用于IM机器人、短信通道。5.4 一次线上事故的完整排查链路复盘我可以把排查过程做成一个通用方法后面遇到触达问题可以直接照着查先确认“到底有没有发送”。看流水表如果没有任何发送记录问题大概率在策略层之前——Agent没产出或者路由判断拦截了。如果发送了但用户没收到看渠道侧回执状态。delivered说明渠道已受理不代表用户已看到要结合已读回执继续判断。如果渠道侧根本没受理看适配器日志里渠道返回的错误码。429就是限流、4xx大概率是参数格式问题、5xx可能是渠道方故障。如果渠道侧返回成功但用户没收到重点查“回执解析”和“消息被拦截”两个方向。回执解析丢失会导致系统误判为失败而重发消息被拦截则要看内容是否命中反垃圾策略。这套排查链路我贴在团队群里每次有人报“消息发不出去”先自己按这个顺序查一遍再升级。有效率提升得非常明显。6. 从单向触达走向双向闭环让Agent在触达中真正积累经验Agent-Reach做到这个阶段往下的核心不只是把消息发出去而是要变成一个数据闭环的起点。每一封邮件、每一条短信、每一次IM推送送的到底有没有效用户是看了没看、点了没点、回了没回这些信息如果不回收Agent就永远在“盲发”。6.1 回执数据回流把“已读、点击、回复”变成可学习的信号目前Agent-Reach针对每条触达回收的信号分三个层级基础信号delivered、failed、blocked等状态回执交互信号邮件打开、链接点击、短信回复、IM会话发起业务信号用户是否完成目标动作比如下单、充值、签署一般由业务系统通过API写入。这三个层级的信号会统一写入触达事件流再回流到策略层。策略层每24小时会算一次渠道质量分不是只看送达率而是看“送达后有没有带来交互”。有些渠道送达率很高但几乎没人打开还不如降级到送达率低一点但打开率高的渠道。6.2 基于反馈自动调整触达参数手动调优迟早被业务速度甩开我最后悔的是前期把太多策略参数写死成了配置。用户活跃时段、渠道偏好、消息措辞风格这些其实都可以从历史触达数据里学出来。目前Agent-Reach支持三个维度的自动调优第一用户活跃时段学习。我统计每个用户历史上“产生交互”的触达时间点把用户分成早型、午型、晚型三类。策略层路由时优先选择用户活跃时段窗口而不是统一上午十点发送。第二渠道偏好学习。同一个用户可能邮件秒读、短信不看。系统根据交互信号为每个用户生成渠道偏好得分路由时把偏好写进打分权重。第三内容风格学习。这个相对粗糙一些目前只看主题行长度、是否带数字、是否带emoji这类表层特征与打开率的关系给Agent生成内容时反馈一份“风格建议”。这个方向的边界也很好把握Agent可以在策略层给定范围内调整触达方式但不允许自行解禁频控、不允许修改降级顺序、不允许突破用户屏蔽。规则层的权限永远大于Agent的实时决策。用一句话说Agent可以随便发挥但安全绳始终攥在平台手里。我做Agent-Reach最深的体会是把Agent的能力真正送到用户面前这中间没有一步是可以靠“感觉”撑过去的。每一层都需要明确的接口、严谨的状态管理、可解释的路由决策以及放在最后的兜底机制。踩过那几次坑之后我现在设计任何触达相关的新功能都会先问自己一句如果这个环节挂了我的系统怎么知道它挂了挂了之后又怎么保证不把事情变得更糟如果这两个问题的答不出来就说明设计还没到位。