去年冬天我接手了一个烂摊子。团队做了三个月的客服 Agent在演示环境里对答如流业务方点头如捣蒜上线第一周就被骂回来了。原因很朴素它只能聊天。客户问我这单为什么还没发货Agent 说得头头是道但它看不到订单系统客户说那你帮我催一下它也答应得很爽快然后什么都没发生。更糟的是同一位客户在一天内被同一个 Agent 用几乎一样的话术回复了四遍因为它压根不知道我已经回过这个人了。那段时间我把能翻的 Agent 框架都翻了一遍最后自己动手抽了一层东西出来内部代号叫Agent-Reach。Reach 取的是够得着的意思。它不负责让模型更聪明它负责让 Agent 够得着外面那个真实的、脏兮兮的世界够得着业务系统的数据够得着一条条真实的沟通渠道够得着人的确认动作也够得着自己的历史状态不至于失忆或者重复劳动。这篇东西我想写的是怎么从零把这层触达层搭起来。适合谁看如果你手上有一个跑得还行但始终落不了地的 Agent 项目或者你正准备给团队做一个能主动干活的自动化系统那这篇基本可以照着抄。做产品的、做运营的也能看里面关于频次治理和人在回路的部分跟写代码关系不大但决定了这东西上线之后会不会被人投诉。1. Agent-Reach 到底解决什么问题从会说到够得着1.1 一个几乎所有人都会撞上的尴尬场景我先描述一个场景你看看熟不熟悉。你做了一个销售线索跟进 Agent逻辑是这样的每天拉一批新线索读一下对方公司的公开信息生成一段个性化的话发出去等回复。Demo 阶段一切完美。真跑起来问题一个接一个线索数据躺在三个不同的系统里字段名都不一样发出去的渠道有邮件、有企业内部的沟通工具、有短信每家的接口形态完全不同发到第 200 条的时候账号被限流了有人回复了Agent 不知道第二天又追了一条对方直接拉黑运营想看一下昨天到底发出去了多少条你翻日志翻了半小时。这些问题的共同点是它们都不是智能问题而是连接问题。模型再强它也没法凭空生成一个可靠的发送队列没法自己算出安全的发送节奏没法记住三天前跟谁聊过。这就是 Agent-Reach 要填的那块空白。我给它下的定义很直白Agent-Reach 是一层位于模型与真实系统之间的触达中间层它把目标是谁、走哪条路、什么时候发、发完怎么跟踪、出事怎么停这五件事从业务代码里抽出来做成统一、可观测、可治理的基础设施。1.2 拆开来看Reach 有三层含义很多人第一次听到这个名字以为就是个群发工具。不是。它至少有三种够得着够得着数据Reach-in把散落在 CRM、数据库、表单、内部 API 里的目标信息统一拉进来做字段归一化和画像补全。这一层解决的是我到底在跟谁说话。够得着通道Reach-out把邮件、站内信、内部沟通工具、短信、Webhook 等各种出口抽象成同一套接口业务侧只写一次逻辑换通道不用改代码。这一层解决的是我说的话怎么送到。够得着状态Reach-back把发送结果、对方回复、退订意愿、投诉信号收回来写进状态机作为下一轮决策的输入。这一层解决的是我说过什么、还能不能再说。三层里最容易被忽略的是第三层也恰恰是决定项目生死的一层。你去看那些上线后被投诉的自动化系统八成不是话术写得差而是它不记得自己说过什么。提示如果你现在只打算做第一版宁可少做两个通道也要把第三层的状态回写做完整。缺通道只是能力弱一点缺状态回写是会出事故的。1.3 谁适合用谁别碰适合的需要多轮、跨渠道、有节奏触达的场景。比如客户成功团队的续费提醒、招聘团队的候选人跟进、开发者关系团队的开源项目 Outreach、社区运营的活动邀请、内部系统的告警升级链。共同特征是——对象是具体的人动作有先后顺序发多了会出事。不太适合的纯粹的一次性广播这种用现成的群发工具就行、对实时性要求到毫秒级的告警那是消息队列的活别硬套 Agent、以及任何没有明确同意基础的营销触达。最后这条不是技术问题是底线问题后面我还会专门讲。2. 整体架构设计我为什么没选择一把梭2.1 四层结构各管各的第一版我犯过一个错把所有逻辑写在一个大函数里从拉数据到发消息到写日志一条龙。结果改一个邮件模板得把整个链路重跑一遍。第二版拆成了四层之后两年的迭代基本没再动过骨架。接入层Ingress负责把目标数据拉进来。它不关心这些数据后面用来干嘛只做三件事抽取、字段映射、进入候选池。这一层我会强制加一个source标记方便后面回溯这条线索是从哪来的。编排层Orchestration是整个系统的大脑。它决定某个目标在当前时刻应该走哪一步、该不该等、该不该跳过。核心是一个显式状态机不是一堆 if-else。执行层Execution由一组通道适配器组成。每个适配器只管一件事把一条标准化消息发出去并把结果标准化吐回来。适配器里不允许出现任何业务判断。观测层Observability负责埋点、日志、指标、审计。这一层最容易被当成后期再说但我的经验是它必须在第一天就有哪怕只有一个最简单的计数接口。因为你排障的时候唯一能救你的就是数据。2.2 关键选型与取舍架构里几个绕不开的选择我把当时的思考过程摊开讲。决策点我的选择备选方案为什么这么选状态存储关系库 Redis 双层纯 Redis关系库存状态机的持久真相Redis 只做高频去重和限流计数重启不丢数据编排方式显式状态机让模型自由决定下一步触达动作有副作用必须可预测、可回放、可审计不能让模型即兴发挥通道抽象统一适配器接口每个通道各写一套通道会换、会加、会下线抽象一次省十次限流位置编排层集中限流各适配器自己限集中限流能看到全局配额分散限流容易出现每个通道都很克制加起来把人淹了人在回路架构内置不是补丁后期加审批按钮高风险动作首次触达、大额方案、投诉对象必须默认走确认提示如果你团队里有人主张先让模型全自动跑起来出问题再补审批请务必拦一下。触达类系统的错误是不可逆的——消息发出去就收不回来了道歉成本远高于一开始加个确认环节。2.3 为什么把人在回路放在架构正中间我见过太多团队把人工确认做成流程末尾的一个兜底按钮平时没人点出事才发现按钮坏了。正确的做法是让审批成为状态机里的一个正式状态有自己的超时时间、自己的重试策略、自己的降级路径。具体来说一条消息在SCREENED之后不是直接进队列而是先判断风险等级。低风险的比如内部通知、已有明确同意基础的例行跟进自动放行中高风险的进入AWAITING_APPROVAL状态挂起等待人工在某个时间窗内处理。超时怎么办我的选择是默认丢弃并记录而不是默认放行。默认放行等于审批环节形同虚设。这套设计的额外好处是审批环节天然产生了一批高质量标注数据人为什么拒了这条是时机不对、话术不对还是对象不对攒上几百条你再去优化 prompt 或者规则方向比拍脑袋准得多。3. 核心模块拆解与实操要点3.1 目标发现与画像归一化这一步听起来最无聊实际上最容易埋雷。我踩过的坑是不同来源对同一个人的标识不一样。CRM 里用邮箱做主键活动报名表用手机号内部系统用工号。不做归一化同一个人会被当成三个目标然后被触达三次。我的做法是引入一个身份归并层先给每个来源定义一组强标识邮箱、手机号等唯一性强的字段和一组弱标识姓名 公司、昵称等然后按优先级做匹配。强标识命中直接合并只有弱标识命中时先合并到一个疑似同一人的候选组由人工或规则二次确认。# 归一化示意强标识直接合并弱标识进待确认组 STRONG_KEYS (email_norm, phone_e164) WEAK_KEYS (name_norm, org_norm) def merge_identity(incoming: dict, index: IdentityIndex) - str: for key in STRONG_KEYS: val incoming.get(key) if val and (hit : index.find_strong(key, val)): return hit.target_id weak_hits index.find_weak(incoming, WEAK_KEYS) if len(weak_hits) 1: return weak_hits[0].target_id if len(weak_hits) 1: index.flag_for_review(incoming, weak_hits) return index.create_provisional(incoming) return index.create_provisional(incoming)注意最后两行的处理弱标识命中多个时不要赌直接标记待确认。我见过一个团队在这里用了取相似度最高的那个结果相似度算法把两个同名的人搞混了一位客户收到了另一位客户的报价单。这种事出一次信任就没了。3.2 通道适配器把五花八门拧成一个接口适配器的价值在于把不确定性关在盒子里。业务代码永远只面对一个send(msg)至于底下走的是邮件还是站内信它不需要知道。from typing import Protocol from dataclasses import dataclass dataclass class OutboundMessage: idempotency_key: str target_id: str channel: str subject: str | None body: str meta: dict dataclass class SendResult: ok: bool external_id: str | None error_code: str | None retryable: bool latency_ms: int class ChannelAdapter(Protocol): name: str def capabilities(self) - set[str]: ... def precheck(self, msg: OutboundMessage) - tuple[bool, str]: ... def send(self, msg: OutboundMessage) - SendResult: ... def health(self) - dict: ...这里有几个我认为必须有、但很多实现会省掉的方法capabilities()返回这个通道支持哪些能力比如是否支持富文本、是否支持附件、单条长度上限是多少、是否支持回执。编排层拿到这些信息之后可以在发送前做长度截断或者换通道而不是等接口报错。precheck()是发送前的本地校验比如邮箱格式、目标是否在退订名单里、消息长度是否超限。把能本地拦掉的错误拦在本地能省下大量无效请求也避免把明显的垃圾请求打到对方接口上。health()让编排层能感知通道状态。通道挂了的时候任务应该进等待队列而不是疯狂重试。通道类型常见长度限制回执支持适配难度备注邮件无硬限制但正文超 2000 字阅读率骤降部分打开/退信中需处理退信和投诉信号站内信通常 500 至 2000 字通常支持已读低最稳优先用内部沟通工具常在 2000 至 4000 字支持低注意频率限制较严短信严格 70 字/条长短信按 67 字/条拼接仅送达回执中成本敏感慎用提示短信的字数计算很多人搞错。纯 ASCII 单条 160 字符含非 ASCII 字符比如中文后单条降到 70 字符长短信按每条 67 字符拼接。中文短信稍微写长一点成本就翻倍。写模板的时候务必在适配器里做好字数预估。3.3 编排层状态机怎么写才不失控编排层的核心是一个显式状态机。我把状态定义成有限集合任何一条任务在任意时刻必须且只能处于一个状态。PENDING已入库未做合规筛查SCREENED已过筛查等待调度AWAITING_APPROVAL等待人工确认QUEUED已入发送队列SENDING发送中有超时保护SENT已提交等待回执DELIVERED确认送达REPLIED已收到回复任务挂起SUPPRESSED被抑制退订、投诉、频次超限FAILED终态失败状态之间的迁移必须集中在一个地方不能散落在各个模块。我的做法是所有迁移都经过一个transition()函数它负责校验合法性、写审计日志、触发副作用。ALLOWED { PENDING: {SCREENED, SUPPRESSED}, SCREENED: {QUEUED, AWAITING_APPROVAL, SUPPRESSED}, AWAITING_APPROVAL: {QUEUED, SUPPRESSED, FAILED}, QUEUED: {SENDING, SUPPRESSED}, SENDING: {SENT, FAILED}, SENT: {DELIVERED, REPLIED, FAILED}, DELIVERED: {REPLIED}, REPLIED: set(), SUPPRESSED: set(), FAILED: set(), } def transition(task, to_state, reason: str): if to_state not in ALLOWED[task.state]: raise IllegalTransition(f{task.state} - {to_state}) audit_log.write(task.id, task.state, to_state, reason) task.state to_state task.save()REPLIED和SUPPRESSED是终态不给任何出口。这一点很重要一旦对方回复了或者明确表达了不再接收的意愿自动化流程必须立刻停手改由人工接管。我见过有系统在对方回复别再发了之后还继续追了三条那就是设计缺陷不是模型问题。3.4 限流与频次治理参数是要算出来的这一节我写详细点因为限流参数拍脑袋的人太多了。假设业务目标是一天触达 6000 个目标工作窗口是早 9 点到晚 7 点共 10 小时。第一步算总速率。10 小时等于 36000 秒6000 除以 36000 得到 0.167 条/秒也就是每分钟 10 条。这是全局上限。第二步算单消费者速率。我一般开 8 个并发消费者那么每个消费者每分钟要处理 1.25 条换算下来约每 48 秒一条。取整到 50 秒再叠加正负 20% 的随机抖动让发送时间点不是机械的等间距。第三步加一层突发容量。纯匀速在工程上很别扭我加一个令牌桶桶容量设为单消费者每分钟配额的 3 倍也就是 3.75取 4允许短时间内小幅突发但长期速率不变。import random, time, threading class TokenBucket: def __init__(self, rate_per_sec: float, capacity: float): self.rate rate_per_sec self.capacity capacity self.tokens capacity self.ts time.monotonic() self.lock threading.Lock() def acquire(self, n: float 1.0) - bool: with self.lock: now time.monotonic() self.tokens min(self.capacity, self.tokens (now - self.ts) * self.rate) self.ts now if self.tokens n: self.tokens - n return True return False def paced_send(consumer, base_interval50.0, jitter0.2): while True: task consumer.next_task() if task is None: time.sleep(1) continue if not consumer.bucket.acquire(): time.sleep(1) continue consumer.dispatch(task) time.sleep(base_interval * (1 random.uniform(-jitter, jitter)))第四步加频次窗口约束。全局速率只是第一道闸真正决定用户体验的是单目标频次。我通常会设三层同一目标 24 小时内不超过 1 次、7 天内不超过 2 次、30 天内不超过 5 次。这三层用 Redis 的滑动窗口计数实现键设计成freq:{target_id}:{window}过期时间就是窗口长度。第五步去重。同一个 campaign、同一个步骤、同一个目标只能发一次。我用幂等键hash(campaign_id step target_id)做 Redis SETNX成功才允许入队。提示幂等键里一定要带campaign_id。我早期版本漏了这个字段结果同一个用户在两个不同活动里被误判成重复白白漏发了。幂等的作用范围要跟你的业务语义对齐不是越宽越好。3.5 工具调用的权限边界与审计Agent 能动手之后权限问题就绕不开了。我的原则是最小权限加显式声明。每个适配器和每个外部工具都要在配置里声明自己需要什么权限运行时用白名单校验。adapters: email_primary: type: smtp permissions: [send_email, read_bounce] daily_quota: 3000 requires_approval: true inapp_notice: type: http permissions: [send_inapp] daily_quota: 20000 requires_approval: false审计日志我坚持记四类信息谁触发的、对谁做的、做了什么、结果如何。日志要能回答这条消息是谁在什么时间因为什么原因发出去的这个问题。别小看这一点真出事的时候没有审计日志你连复盘都做不了。4. 从零跑通一条完整触达链路4.1 目录结构与依赖我推荐的组织方式是按职责分包而不是按技术分层不要models/、services/那种大锅饭。agent_reach/ ingress/ # 数据接入与身份归并 orchestration/ # 状态机、调度、限流、审批 adapters/ # 各通道适配器 policy/ # 合规筛查、频次策略、抑制名单 observability/ # 指标、日志、审计 config/ adapters.yaml policy.yaml依赖上没什么玄学一个 HTTP 客户端、一个 Redis 客户端、一个关系库驱动、一个模板引擎够了。我刻意不引入重型工作流引擎因为这套东西的状态机语义很清晰自己写两百行比学一个框架快得多出了问题也好查。4.2 实现一个最小可用的适配器以站内信为例一个完整适配器大概长这样。class InAppAdapter: name inapp_notice def __init__(self, client, quota: int): self.client client self.quota quota self.used 0 def capabilities(self) - set[str]: return {rich_text, read_receipt, link} def precheck(self, msg): if len(msg.body) 2000: return False, BODY_TOO_LONG if self.used self.quota: return False, QUOTA_EXCEEDED return True, def send(self, msg): t0 time.monotonic() try: resp self.client.post( /notify, json{to: msg.target_id, title: msg.subject, content: msg.body, biz_id: msg.idempotency_key}, timeout8, ) self.used 1 ok resp.status_code 200 return SendResult( okok, external_idresp.json().get(msg_id) if ok else None, error_codeNone if ok else fHTTP_{resp.status_code}, retryable(resp.status_code 500), latency_msint((time.monotonic() - t0) * 1000), ) except TimeoutError: return SendResult(False, None, TIMEOUT, True, int((time.monotonic() - t0) * 1000))几个细节值得说。超时一定要设我设的是 8 秒因为触达请求不是同步等结果的场景超时宁可重试也别把消费者堵死。retryable字段必须由适配器给出因为只有适配器知道某个错误码重试有没有意义——4xx 类错误重试一百次也是白搭5xx 和超时才值得重试。biz_id传幂等键让下游也能去重双保险。4.3 把调度串起来调度主循环不复杂关键是每一步都要写状态。def run_once(ctx): batch ctx.store.claim_batch(stateQUEUED, limit50, lease_sec120) for task in batch: if ctx.policy.is_suppressed(task.target_id): transition(task, SUPPRESSED, suppressed_list) continue allowed, reason ctx.freq.allow(task.target_id, task.campaign_id) if not allowed: ctx.store.requeue_later(task, delay_sec3600) ctx.metrics.inc(freq_blocked, tags{reason: reason}) continue if not ctx.limiter.acquire(): ctx.store.requeue_later(task, delay_sec30) continue adapter ctx.adapters[task.channel] ok, why adapter.precheck(task.to_message()) if not ok: transition(task, FAILED, fprecheck:{why}) continue transition(task, SENDING, dispatch) result adapter.send(task.to_message()) handle_result(ctx, task, result)claim_batch里用了租约lease机制任务被取走时打上 120 秒的租约消费者如果在租约期内没完成任务会被其他消费者重新领取。这是防止消费者崩溃导致任务永久卡在SENDING状态的关键。租约时长要大于单次发送的最坏耗时我一般设成超时时间的 10 倍以上。4.4 灰度与回滚怎么做触达系统上线绝对不要全量放。我的灰度节奏是这样的影子模式只生成消息、只写日志一条都不发。跑三天人工抽查生成质量。白名单 1%只对内部同事和主动报名的用户发。跑两天看回复率和投诉率。放量 10%观察限流是否触发、错误率是否稳定。放量 50%然后全量。每一步都设好回滚开关。回滚开关必须是一个独立于业务代码的配置项最好放在配置中心出问题时一键关停不需要发版。我甚至建议在监控里加一条硬规则投诉率或退订率超过阈值时自动熔断不需要人点。4.5 必须看的几个指标指标名含义健康区间参考send_success_rate发送成功率高于 98%deliver_rate送达率高于 95%reply_rate回复率因场景差异大看趋势不看绝对值unsubscribe_rate退订率低于 1%超过 2% 必须停complaint_rate投诉率高于 0.1% 立刻熔断freq_blocked_ratio被频次拦下的比例高于 30% 说明策略太松或目标池太窄p95_latency_ms发送耗时 95 分位低于 3000msfreq_blocked_ratio这个指标特别有意思。它偏高通常不是坏事反而说明频次治理在起作用。但如果它一直很高同时回复率在下降那就要反思是不是目标池太集中同一批人被反复捞出来。这时候该做的是扩充目标来源而不是放松频次限制。5. 常见问题与排查技巧实录5.1 常见问题速查表现象大概率原因排查动作处理方式同一个人收到重复消息幂等键设计漏了维度或身份归并没合并查该 target_id 的全部发送记录比对幂等键补全幂等键维度重跑归并任务卡在 SENDING 不动消费者崩溃租约未过期或未实现重领查租约字段和消费者心跳实现租约重领缩短租约时长发送成功率骤降通道侧限流或接口变更看错误码分布看 p95 延迟降速、切备用通道、联系通道方审批环节堆了几百条审批入口太深或没通知看审批平均等待时长加提醒、做批量审批、分派到人回复了但流程没停回执通道没接或状态机缺终态检查回执写入链路补回执消费REPLIED设为终态频次策略形同虚设Redis 键未设过期时间或时钟不一致抽查键的 TTL统一用服务端时间强制设 TTL5.2 我踩过的三个坑坑一用本地时间做滑动窗口。多台机器时钟有几十毫秒偏差滑动窗口就会出现边界漏发或者边界重发。解决办法是全部换成 Redis 服务端时间或者干脆用 Redis 的INCR加EXPIRE实现固定窗口牺牲一点精度换一致性。坑二把模板渲染放在适配器里。一开始觉得方便适配器拿到数据自己渲染。后来要改文案发现得改四个适配器。正确做法是模板渲染在编排层完成适配器只接收渲染好的纯文本。适配器应该尽可能笨。坑三忽略退信和投诉信号的处理。邮件退信是分类型的硬退信地址不存在应该立刻把该地址加入永久抑制名单软退信暂时性故障应该重试几次后再降级。我早期版本一视同仁地重试结果往一堆已经废弃的地址反复发信发送声誉掉得很厉害。提示硬抑制名单Hard Suppress List必须是全系统共享的一张表任何通道、任何活动都不能绕过。它是你最后一道防线绝不允许业务代码有临时跳过的口子。5.3 我自己的几条硬规矩做了两年多我给自己定了几条死规矩这里分享出来。第一条任何自动化触达都必须有明确的人工退出路径。消息里必须包含可用的退订方式而且退订动作必须实时生效不能等到第二天批量处理。用户体验的差距就在这几个小时里。第二条首次触达永远走审批。不管模型多自信第一次跟一个陌生人说话一定要有人看一眼。第二次之后可以放开。第三条宁可不发不要错发。所有拿不准的情况身份冲突、意图不明、目标状态异常一律进AWAITING_APPROVAL或者直接SUPPRESSED。少发一条的损失是零错发一条的损失可能是一个长期客户的信任。第四条监控指标要能直接反映人的感受。技术指标成功率、延迟是给工程师看的业务指标回复率、退订率、投诉率是给决策者看的两套都要有而且要放在同一块看板上。我见过只盯技术指标的团队指标全绿用户已经在骂了。如果你现在正准备动手做类似的东西我建议第一版只做一个通道、一个场景、一天不超过 100 条把状态机和抑制名单跑通。等这套骨架稳了加通道就是复制粘贴的活。反过来先铺十个通道再回头补治理基本等于重写。后续这个骨架还能往上长比如把审批环节换成基于历史决策的推荐给人一个默认选项人只需要点同意或拒绝效率能提升好几倍或者把回复内容做意图分类后自动路由到不同的处理队列。这些都是我接下来打算试的方向等跑出结果再写一篇。