1. Agent-Reach 到底想解决什么问题第一次看到 Agent-Reach 这个名字我脑子里蹦出来的不是又一个智能体框架而是三个字——够不着。我们做智能体的团队大概都有过这种体验模型在对话框里说得头头是道真要它去把一件事落实到外部系统上立刻掉链子。查个订单状态返回超时发个通知重复发了三遍调用一个内部接口因为没有白名单被网关拦掉任务卡在半路没人知道。Agent-Reach 就是冲着这一段来的它不管模型怎么想只管模型想完之后怎么把动作稳稳当当地送到该去的地方并且拿回一个可信的结果。说人话它是一层触达能力层。定位在智能体推理循环和外部世界之间负责通道抽象、任务编排、重试补偿、幂等去重、权限校验和全链路审计。适合谁来参考如果你正在做企业内部智能助手、自动化运维助手、客服工单机器人、数据巡检机器人这类东西只要你遇到模型能力没问题但执行不可靠的困惑这套思路基本可以直接搬。如果你只是做纯对话问答不落地任何外部动作那它对你价值不大别硬上。我做这个项目的起因很朴素一个内部工单助手上线第一周就出了两次事故一次是同一张工单被派给了三个组一次是夜间批处理任务全部超时但没人收到告警。复盘下来发现问题都不在模型全在触达链路上。于是我把触达这块单独抽出来当成一个项目做也就是 Agent-Reach。1.1 为什么触达比推理更难做推理难在质量触达难在确定性。模型输出一句话好一点差一点用户能忍但一条通知发重了、一个写操作执行了两次、一笔状态更新丢了这是事故。两者的工程约束完全不是一个量级。我总结触达链路有四个绕不开的麻烦。第一个是外部系统不可控对方接口可能超时、可能限流、可能返回一个语义模糊的 200。第二个是网络语义天然不精确请求发出去了回包没回来你根本不知道对方执行了没有这是分布式系统里的老问题。第三个是智能体行为不确定同一个意图模型这次调一个工具下次可能拆成两步调两个工具。第四个是多通道差异巨大消息推送、邮件、内部 RPC、Webhook 回调它们的超时标准、幂等支持程度、错误码语义全都不一样。把这四件事同时放在智能体主循环里处理代码会迅速烂掉。Agent-Reach 的价值就是把这些脏活收拢到一层让主循环只管我要做这个动作剩下的交给它。1.2 三条设计红线项目一开始我就定了三条不能破的线后面所有取舍都围绕它们。第一条任何触达必须幂等。不管上游重试多少次、网络抖动多少次业务侧最多只应该感知到一次执行。做不到幂等的通道宁可只读不写或者强制人工确认。第二条任何触达必须有回执。没有回执的调用等于没调用。哪怕是异步的也必须有一个可查询的状态记录否则任务就成了黑洞。第三条任何触达必须可追溯。谁发起的、什么时候发起的、参数是什么、重试了几次、最终结果如何全部落库。这条在事故复盘时救过我好几次。这三条听起来像常识但真做起来每一条都会逼你改架构。比如第一条会逼你设计幂等键的生成规则第二条会逼你把同步调用改造成提交任务 轮询/回调的两段式第三条会逼你在每个环节埋点而不只是打日志。1.3 选型取舍为什么不直接上重型编排框架一开始我也考虑过直接用现成的工作流编排引擎把每个触达动作做成一个节点靠引擎的重试和状态机兜底。试了两周放弃了原因有两个。一是粒度不匹配。编排引擎擅长的是长周期、少分支的流程而智能体触达的特点是高频、短周期、分支极度发散。一个早上可能跑几千次触达每次就一两秒用重型引擎跑调度开销比业务本身还大。二是参数动态性太强。智能体调工具的参数是运行时生成的不是提前编排好的硬塞进静态流程图里会变成一堆动态字段可读性归零。最后我的方案是自研一层轻量触达层只在需要跨天、跨系统、带人工审批的长流程时才把任务交给外部编排引擎。分层不是炫技是为了让每层只干自己擅长的事。2. 核心细节拆解与实操要点架构定了之后真正决定成败的是细节。这一章我挑四个最关键的讲通道抽象、状态机、幂等键、权限审计。这四个东西设计错了后面怎么补都是补丁摞补丁。2.1 通道抽象一个接口装下所有外部世界通道适配器的接口我改过三版最后稳定下来的核心方法只有四个class Channel: name: str def validate(self, action) - None: 参数与权限的静态校验不产生副作用 def submit(self, action, idem_key) - str: 提交动作返回外部任务号必须幂等 def poll(self, external_id) - Status: 查询状态返回 PENDING / SUCCESS / FAILED / UNKNOWN def cancel(self, external_id) - bool: 尽力取消允许不支持时返回 False关键点是submit只提交不等待。这是我吃过大亏之后改的。最早我把submit写成同步阻塞通道 A 快、通道 B 慢一个慢通道就把整个智能体循环拖住。改成两段式之后主循环提交完立刻拿一个任务号走人状态由后台的轮询器统一收割。poll返回UNKNOWN这个状态很重要很多人会漏。它表示我查不到但也不能断定失败。比如对方系统返回 5xx或者查询超时。UNKNOWN不能当成失败去重试否则就是重复执行也不能当成成功会漏掉真正的失败。我的处理是UNKNOWN进入一个独立的延迟队列按指数退避反复查询超过设定次数后升级为人工介入。注意poll千万不要设计成查不到就返回 FAILED。这是我见过最多人踩的坑一次含糊的失败判定会在下游制造几十次重复写。至于cancel我的态度是尽力而为。大部分第三方系统并不提供可靠的回滚能力所以真正的策略不是失败了回滚而是提交前尽量确认清楚。回滚是奢侈品前置校验才是刚需。2.2 任务、动作、回执三段式状态机的设计Agent-Reach 内部只有三个核心对象我把它们的关系理得很死不允许交叉写入。**任务Task**是业务语义单位一次用户意图对应一个任务比如给这张工单派单。任务是长生命周期的可能包含多个动作。**动作Action**是技术执行单位一个任务可能拆成三四个动作比如校验工单状态查询组负载写入派单结果发送通知。动作是短生命周期的有明确的成功/失败。**回执Receipt**是动作执行的外部凭证包含外部任务号、返回码、原始响应片段、耗时。状态流转也很简单任务待执行 → 动作进行中 → 动作完成 → 任务完成。任务只有在所有必需动作成功后才会收尾任一必需动作硬失败任务进入失败态并触发补偿或告警。这里有个实操细节值得说动作之间要有依赖声明。写入派单结果必须依赖校验工单状态成功但发送通知不依赖写入派单结果能不能成功。我用一个简单的depends_on列表表达执行器只调度依赖已满足的动作。这样部分失败时能跑的动作照跑不会因为一个分支失败就整体卡住。2.3 幂等键一行字符串决定系统可靠不可靠整个项目里我认为最重要的一个设计就是幂等键。它是一串字符串作为动作的唯一身份标识外部系统如果支持幂等头就直接透传不支持就在本地做去重表。生成规则我用的是拼接 哈希四个组成部分缺一不可任务标识业务实体的唯一 ID比如工单号动作类型比如dispatch、notify关键参数摘要把影响结果的参数排序后序列化再取哈希逻辑时间窗比如按天或按批次防止跨周期误去重拼起来长这样task:WO20240517-8821/action:dispatch/params:a3f9.../window:2024-05-17为什么第四个时间窗必须有因为有些场景是周期性重复的比如每天给这个组发一次日报提醒如果只按实体和动作去重第二天就发不出去了。加上日期窗天然按天隔离。为什么关键参数摘要而不是全部参数因为有些参数不影响结果比如 trace_id、重试次数、日志级别。如果把它们也算进去一次重试就会生成新的幂等键去重直接失效。我的做法是维护一个参与幂等计算的字段白名单显式声明不靠猜。注意幂等键一旦生成就不能变。我在代码里给它加了不可变约束任何地方想改它都会抛异常。因为一个变来变去的幂等键等于没有幂等。本地去重表用数据库唯一索引实现别用内存缓存进程重启就失效了。表结构就是幂等键做唯一索引加上状态和首次写入时间。插入冲突就说明是重复请求直接返回已有记录不执行。2.4 权限、白名单与审计让智能体别乱伸手智能体最大的风险不是能力不够而是能力太够。它可能调用你根本没打算让它调的工具可能传一个越权的参数。所以在触达层做权限控制比在提示词里写请不要做危险操作靠谱一万倍。我的做法是三层校验。第一层是工具级白名单哪些工具允许被智能体调用配置化声明没登记的通道直接拒绝。第二层是参数级约束每个通道声明自己的参数规则比如目标组必须是当前用户所属组单次派单数量不超过 20。这些规则写成声明式配置在validate阶段执行越界直接拒绝并记录。第三层是数据级隔离不同租户、不同业务线的数据互不可见在查询和写入时都带上隔离字段而不是靠应用层过滤。审计日志我要求记录五个必备字段调用者身份哪个智能体会话、幂等键、完整参数、执行结果、耗时。日志本身不做业务逻辑只追加不修改。这里有个小技巧参数落库前要做一次脱敏处理手机号、身份证、地址这类字段统一打码否则审计日志本身会变成合规风险点。3. 手把手落地从零搭一个能跑的最小版本理论讲完了说点能直接抄的。这一章我按搭积木的顺序从目录结构到代码骨架再到参数计算尽量给到可直接复现的细节。你做的时候不用完全照搬抓住结构就行。3.1 最小骨架与依赖选择目录结构我建议这样分边界清楚后面扩展不痛苦agent_reach/ channels/ 各通道适配器 core/ task.py 任务定义 action.py 动作与依赖 idem.py 幂等键生成与去重 executor.py 执行器与调度 retry.py 重试策略 guards/ permission.py audit.py obs/ metrics.py tracing.py依赖上我刻意保持极简一个异步 HTTP 客户端、一个数据库驱动、一个序列化库就这些。理由很直接触达层的核心诉求是稳依赖越少出问题的面越小升级时被第三方库拖着走的概率也越低。数据库我选了支持唯一索引和事务的关系型库。别用纯 K-V 存去重表因为你需要按状态、按时间去检索失败任务K-V 会很痛苦。异步框架用原生 asyncio 就够了不用引入额外的事件循环。3.2 一个能跑的通道适配器下面这个适配器实现了完整的四方法接口同时支持透传幂等头可以直接当模板改import hashlib, httpx class WebhookChannel: name webhook def __init__(self, base_url, token, supports_idem_headerTrue): self.base_url base_url self.token token self.supports_idem_header supports_idem_header self.client httpx.AsyncClient(timeouthttpx.Timeout(3.0, read8.0)) def validate(self, action): if not action.params.get(url): raise ValueError(missing url) if len(action.params.get(body, )) 64 * 1024: raise ValueError(payload too large) async def submit(self, action, idem_key): headers {Authorization: fBearer {self.token}} if self.supports_idem_header: headers[X-Idempotency-Key] idem_key resp await self.client.post( action.params[url], jsonaction.params[body], headersheaders ) if resp.status_code 500: raise TransientError(resp.text) if resp.status_code 400: raise PermanentError(resp.text) return resp.json().get(task_id, idem_key) async def poll(self, external_id): try: r await self.client.get(f{self.base_url}/tasks/{external_id}, headers{Authorization: fBearer {self.token}}) except Exception: return Status.UNKNOWN if r.status_code 404: return Status.UNKNOWN state r.json().get(state) return {done: Status.SUCCESS, failed: Status.FAILED}.get(state, Status.PENDING)两个细节要展开讲。一是超时拆开设置连接超时给 3 秒读取超时给 8 秒。连接超时短是因为连不上基本就是网络问题等再久也没用读取超时长是因为对方可能在处理给点耐心。很多人统一设 30 秒结果是慢调用把连接池占满整个系统雪崩。二是异常分类。这里我把异常明确分成TransientError可重试和PermanentError不可重试。这个区分太重要了重试一个永久性错误比如参数格式不对除了浪费配额没有任何意义。5xx 和超时归为可重试4xx 归为不可重试这是最基本的判断具体到每个通道还要按对方文档细化。3.3 重试退避与熔断参数怎么算出来重试策略我用的是指数退避加随机抖动公式delay min(base * 2^n, cap) * (1 random(-jitter, jitter))。参数取值我给一套实践过的base取 0.5 秒cap取 30 秒最大重试 4 次jitter取 0.3。算下来延迟序列大概是 0.5、1、2、4 秒加上抖动后总耗时在 10 秒左右。为什么是四次因为 0.5×2^4 8 秒已经接近大多数同步接口的用户可接受上限再往上延任务就该转异步了。jitter为什么必须有防惊群。如果 1000 个任务同时失败全部按同样的节奏重试对方系统会迎来四波整齐的流量高峰本来能恢复的也被打挂了。加 30% 抖动后请求被打散对方压力平缓很多。熔断我用的是简单的滑动窗口计数10 秒窗口内失败率超过 50% 且样本数大于 20就打开熔断持续 30 秒。为什么要有样本数下限因为 3 个请求失败 2 个也是 66%但那只是正常波动直接熔断会误伤。熔断打开期间请求直接快速失败不再真实调用给下游喘息时间。30 秒后进入半开状态放 5 个探针请求成功就恢复。注意重试要区分幂等动作和非幂等动作。非幂等动作用户层面根本不该重试只能靠前置查询确认状态。上线前一定把每个动作的幂等性标注清楚这是硬性要求。3.4 接进智能体主循环只暴露一个方法这层对智能体来说应该极度简单。我只暴露一个方法输入是动作描述输出是任务号async def reach(action_spec: dict) - str: action Action.from_spec(action_spec) channel registry.get(action.channel) channel.validate(action) idem_key build_idem_key(action) if (existing : await store.find(idem_key)): return existing.task_id await guard.check(action) await store.reserve(idem_key, action) try: external_id await channel.submit(action, idem_key) await store.update(idem_key, external_id, Status.PENDING) except TransientError: await scheduler.enqueue_retry(idem_key) except PermanentError as e: await store.fail(idem_key, str(e)) await alerts.raise_task(action.task_id, str(e)) await audit.log(action, idem_key) return action.task_id重点是store.reserve这一步它靠数据库唯一索引抢锁抢到了才执行抢不到说明有人在做同一件事直接返回。这个先占坑再执行的顺序不能反反过来就失去去重意义。给智能体的工具描述也要写得克制。我见过把十几个通道全塞给模型的结果它经常选错。正确做法是按业务场景裁剪一个场景只暴露三到五个工具工具名和描述里明确写清这个工具会产生外部副作用让模型知道轻重。3.5 可观测性三个指标定位八成问题触达层出问题时我基本只看三个指标覆盖了绝大多数场景。第一个是各状态的计数分布待执行、进行中、成功、失败、未知。这个分布一旦有异常比如未知持续上涨基本就是某个通道在抖动。第二个是端到端耗时分布我关注 P50、P95、P99 三个分位。P99 突然拉高但 P50 没变通常是少数慢请求或者某个通道开始限流。第三个是重试率按通道分维度看。某个通道重试率超过 5%就该去查对方文档或者联系对接人了。链路追踪上我给每个任务生成一个 trace 标识从智能体会话一路透传到外部调用。这样出问题时从用户反馈直接能定位到具体哪一次网络请求。日志我坚持结构化输出不写拼接字符串方便事后检索聚合。4. 常见问题与排查技巧实录这一章是纯经验都是我在生产环境里被真实咬过的。每条包括现象、原因、处理你可以当成一个速查清单用。4.1 重复触达最常见的那个坑现象是用户收到两条一模一样的通知或者一张单被派了两次。原因排下来无非四个。第一个是重试没带幂等键。请求超时了代码自动重试但重试时生成了一个新的幂等键下游认为是两次不同的请求。解决办法是幂等键在动作创建时就固定重试只复用不重建。第二个是去重表用了缓存。进程重启、缓存过期去重信息就没了。解决办法是缓存只做加速真正的判定必须落库用唯一索引。第三个是外部系统不支持幂等头。有些老系统就是没有这个能力。这种情况下只能在本地加状态查询提交前先查一次是否已存在虽然不能百分百避免但能挡掉绝大多数。第四个是并发窗口。两个请求几乎同时到达都查了一次不存在然后都执行了。解决办法是把查询和占坑做成一个原子操作靠唯一索引冲突来判定。4.2 任务卡在进行中不动这个问题的排查思路我固定成四步。先看这个通道的 poll 是否正常返回如果一直返回 PENDING说明对方状态没更新或者我们查错了任务号。再看外部任务号是否正确落库我曾经遇到过一次序列化问题任务号写进去多了个引号导致永远查不到。第三步看轮询器本身是不是活着有没有被某个异常循环卡死。第四步才是看对方系统。这个顺序很重要先从自己能控的部分查起比一上来就怀疑对方高效得多。处理上我加了一个停滞超时机制任何动作在 PENDING 状态超过设定时间一般按通道配置短的 2 分钟长的 30 分钟自动升级为 UNKNOWN 并触发人工核对。不要让它无限期挂着那才是真正的黑洞。4.3 限流与配额打满现象是某段时间大量请求返回 429 或者类似错误码。核心原因是我们只管发不管节奏。我的处理是给每个通道加令牌桶限流速率按对方文档给的配额打个七折配置留出余量。为什么要打七折因为对方统计口径可能和你的不一样它算的是 QPS你算的是每分钟调用数边界情况容易超。留 30% 余量能挡住大部分突发。另外限流错误要和普通错误区别对待它应该有独立的退避策略退避时间更长并且优先降级为异步处理而不是继续在高频重试上耗着。4.4 常见问题速查表现象高概率原因处理动作重复通知/重复派单幂等键重建、去重靠缓存、并发窗口固定幂等键、改唯一索引、原子占坑任务长期 PENDINGpoll 返回错任务号、轮询器卡死校验落库字段、加停滞超时升级大量 429无本地限流、突发流量令牌桶限速、降级异步失败率突增但对方正常参数被智能体生成得越界加强参数级校验、收窄工具白名单审计日志缺失埋点位置在异常分支之后把审计提到 finally 中执行5xx 后大量重复写把 UNKNOWN 当 FAILED 重试拆分 UNKNOWN独立延迟队列慢请求拖垮整体未拆连接/读取超时连接 3s、读取 8s 分开配置夜间批处理无声失败缺少失败告警按任务类型配置告警阈值这张表我压在团队 wiki 首页新人进来看一遍就能少踩一半坑。特别是最后一条无声失败是最可怕的它不是出错是没人知道出错了所以告警配置的优先级应该和功能开发一样高。5. 场景延展与容量估算把核心跑通之后我陆续把这套东西用在了几个不同场景上顺便做了些容量估算这块经验也挺值得分享。5.1 不同场景该怎么组合通道工单类场景核心动作是查询、校验、写入、通知。我的组合是查询和校验走内部 RPC要求同步、低延迟写入走内部 API 但配幂等头通知走消息通道且允许异步。这个组合的关键是把写和通知解耦写成功就必须通知到位但通知失败不应该让整个任务回滚。巡检类场景特点是批量、周期、可容忍延迟。这时候我会把触达层切成两档轻量检查走同步耗时的深度检查丢进队列异步跑用任务号串起来。巡检最怕的是任务堆积所以我给它单独设了一个并发上限避免把其他场景的资源挤掉。对外通知类场景比如给外部合作方推送数据对时效和准确性要求都高。这类我会额外加一层发送前二次校验把关键字段再核一遍同时把回执完整落库因为一旦出错对外的影响面比内部大得多。5.2 容量与成本怎么估估算我一般从三个数出发日均任务量、单任务平均动作数、单动作平均耗时。举个例子日均一万个任务平均每个任务三个动作单动作平均一点五秒那一天的触达总量是三万次序号化处理的有效负载约占 12.5 小时的机器时间。再乘上并发系数。因为大部分任务是突发性的我会按峰值 QPS 日均次数 / 有效小时 / 3600 × 峰值倍数峰值倍数一般取 5 到 8。这样算出来的并发量才是配置连接池和限流阈值的依据按平均值配一定会被打爆。存储上去重表是主要增长点。一条记录大概两百字节一天三万次就是六兆一年两个多 G完全可控。但如果高频场景上到百万级就得考虑分区或者定期归档了只保留近三个月的去重记录再往前的靠幂等键本身的时间窗来规避。注意去重记录不能删太早。保留期要覆盖最长的重试链路 最长的人工处理时长我一般设三个月起步宁可多占点存储也别在这个地方省。5.3 我个人踩过的几个坑第一个坑是过早抽象。一开始我设计了七种通道类型、完整的插件机制、热加载能力结果前三个月只用到两种通道抽象层反而成了每次改动的阻碍。后来我一狠心砍掉一半抽象代码量少了三分之一改动速度反而快了。结论是触达层这种基础设施抽象要跟着真实需求长别提前建。第二个坑是把校验放在模型那一侧。我一度把参数合法性交给提示词去约束结果模型十次里有两次不听话。后来把校验全部下沉到通道的validate里模型怎么折腾都没事因为它再能说也得过这道关。凡是能用代码强制的就别指望提示词。第三个坑是告警阈值拍脑袋。最早告警一响就发结果一天收到几百条大家自动忽略了。后来改成按通道和历史基线做动态阈值只有偏离基线三倍标准差才告警噪音少了九成真正的问题也终于有人看了。第四个坑是没有区分业务失败和技术失败。早期所有失败都走同一条告警业务侧收到大量技术告警技术侧收到大量业务告警双方都烦。后来在动作定义里明确标注失败类型技术失败给运维业务失败给业务方各看各的效率一下就上来了。最后分享一个我认为最值钱的小习惯每周花十分钟把上周所有进入 UNKNOWN 状态的动作捞出来逐条看一遍。这个动作看起来很小但它能帮你在问题变成事故之前就发现通道的隐患。我做过大概二十周其中至少有五次提前发现了对方接口的语义变更避免了大面积故障。触达这件事本质上不是把功能做出来而是把不确定性一点点关进笼子里笼子越结实上层的智能体才越敢放手动。
Agent-Reach：智能体触达层的幂等、回执与审计设计
1. Agent-Reach 到底想解决什么问题第一次看到 Agent-Reach 这个名字我脑子里蹦出来的不是又一个智能体框架而是三个字——够不着。我们做智能体的团队大概都有过这种体验模型在对话框里说得头头是道真要它去把一件事落实到外部系统上立刻掉链子。查个订单状态返回超时发个通知重复发了三遍调用一个内部接口因为没有白名单被网关拦掉任务卡在半路没人知道。Agent-Reach 就是冲着这一段来的它不管模型怎么想只管模型想完之后怎么把动作稳稳当当地送到该去的地方并且拿回一个可信的结果。说人话它是一层触达能力层。定位在智能体推理循环和外部世界之间负责通道抽象、任务编排、重试补偿、幂等去重、权限校验和全链路审计。适合谁来参考如果你正在做企业内部智能助手、自动化运维助手、客服工单机器人、数据巡检机器人这类东西只要你遇到模型能力没问题但执行不可靠的困惑这套思路基本可以直接搬。如果你只是做纯对话问答不落地任何外部动作那它对你价值不大别硬上。我做这个项目的起因很朴素一个内部工单助手上线第一周就出了两次事故一次是同一张工单被派给了三个组一次是夜间批处理任务全部超时但没人收到告警。复盘下来发现问题都不在模型全在触达链路上。于是我把触达这块单独抽出来当成一个项目做也就是 Agent-Reach。1.1 为什么触达比推理更难做推理难在质量触达难在确定性。模型输出一句话好一点差一点用户能忍但一条通知发重了、一个写操作执行了两次、一笔状态更新丢了这是事故。两者的工程约束完全不是一个量级。我总结触达链路有四个绕不开的麻烦。第一个是外部系统不可控对方接口可能超时、可能限流、可能返回一个语义模糊的 200。第二个是网络语义天然不精确请求发出去了回包没回来你根本不知道对方执行了没有这是分布式系统里的老问题。第三个是智能体行为不确定同一个意图模型这次调一个工具下次可能拆成两步调两个工具。第四个是多通道差异巨大消息推送、邮件、内部 RPC、Webhook 回调它们的超时标准、幂等支持程度、错误码语义全都不一样。把这四件事同时放在智能体主循环里处理代码会迅速烂掉。Agent-Reach 的价值就是把这些脏活收拢到一层让主循环只管我要做这个动作剩下的交给它。1.2 三条设计红线项目一开始我就定了三条不能破的线后面所有取舍都围绕它们。第一条任何触达必须幂等。不管上游重试多少次、网络抖动多少次业务侧最多只应该感知到一次执行。做不到幂等的通道宁可只读不写或者强制人工确认。第二条任何触达必须有回执。没有回执的调用等于没调用。哪怕是异步的也必须有一个可查询的状态记录否则任务就成了黑洞。第三条任何触达必须可追溯。谁发起的、什么时候发起的、参数是什么、重试了几次、最终结果如何全部落库。这条在事故复盘时救过我好几次。这三条听起来像常识但真做起来每一条都会逼你改架构。比如第一条会逼你设计幂等键的生成规则第二条会逼你把同步调用改造成提交任务 轮询/回调的两段式第三条会逼你在每个环节埋点而不只是打日志。1.3 选型取舍为什么不直接上重型编排框架一开始我也考虑过直接用现成的工作流编排引擎把每个触达动作做成一个节点靠引擎的重试和状态机兜底。试了两周放弃了原因有两个。一是粒度不匹配。编排引擎擅长的是长周期、少分支的流程而智能体触达的特点是高频、短周期、分支极度发散。一个早上可能跑几千次触达每次就一两秒用重型引擎跑调度开销比业务本身还大。二是参数动态性太强。智能体调工具的参数是运行时生成的不是提前编排好的硬塞进静态流程图里会变成一堆动态字段可读性归零。最后我的方案是自研一层轻量触达层只在需要跨天、跨系统、带人工审批的长流程时才把任务交给外部编排引擎。分层不是炫技是为了让每层只干自己擅长的事。2. 核心细节拆解与实操要点架构定了之后真正决定成败的是细节。这一章我挑四个最关键的讲通道抽象、状态机、幂等键、权限审计。这四个东西设计错了后面怎么补都是补丁摞补丁。2.1 通道抽象一个接口装下所有外部世界通道适配器的接口我改过三版最后稳定下来的核心方法只有四个class Channel: name: str def validate(self, action) - None: 参数与权限的静态校验不产生副作用 def submit(self, action, idem_key) - str: 提交动作返回外部任务号必须幂等 def poll(self, external_id) - Status: 查询状态返回 PENDING / SUCCESS / FAILED / UNKNOWN def cancel(self, external_id) - bool: 尽力取消允许不支持时返回 False关键点是submit只提交不等待。这是我吃过大亏之后改的。最早我把submit写成同步阻塞通道 A 快、通道 B 慢一个慢通道就把整个智能体循环拖住。改成两段式之后主循环提交完立刻拿一个任务号走人状态由后台的轮询器统一收割。poll返回UNKNOWN这个状态很重要很多人会漏。它表示我查不到但也不能断定失败。比如对方系统返回 5xx或者查询超时。UNKNOWN不能当成失败去重试否则就是重复执行也不能当成成功会漏掉真正的失败。我的处理是UNKNOWN进入一个独立的延迟队列按指数退避反复查询超过设定次数后升级为人工介入。注意poll千万不要设计成查不到就返回 FAILED。这是我见过最多人踩的坑一次含糊的失败判定会在下游制造几十次重复写。至于cancel我的态度是尽力而为。大部分第三方系统并不提供可靠的回滚能力所以真正的策略不是失败了回滚而是提交前尽量确认清楚。回滚是奢侈品前置校验才是刚需。2.2 任务、动作、回执三段式状态机的设计Agent-Reach 内部只有三个核心对象我把它们的关系理得很死不允许交叉写入。**任务Task**是业务语义单位一次用户意图对应一个任务比如给这张工单派单。任务是长生命周期的可能包含多个动作。**动作Action**是技术执行单位一个任务可能拆成三四个动作比如校验工单状态查询组负载写入派单结果发送通知。动作是短生命周期的有明确的成功/失败。**回执Receipt**是动作执行的外部凭证包含外部任务号、返回码、原始响应片段、耗时。状态流转也很简单任务待执行 → 动作进行中 → 动作完成 → 任务完成。任务只有在所有必需动作成功后才会收尾任一必需动作硬失败任务进入失败态并触发补偿或告警。这里有个实操细节值得说动作之间要有依赖声明。写入派单结果必须依赖校验工单状态成功但发送通知不依赖写入派单结果能不能成功。我用一个简单的depends_on列表表达执行器只调度依赖已满足的动作。这样部分失败时能跑的动作照跑不会因为一个分支失败就整体卡住。2.3 幂等键一行字符串决定系统可靠不可靠整个项目里我认为最重要的一个设计就是幂等键。它是一串字符串作为动作的唯一身份标识外部系统如果支持幂等头就直接透传不支持就在本地做去重表。生成规则我用的是拼接 哈希四个组成部分缺一不可任务标识业务实体的唯一 ID比如工单号动作类型比如dispatch、notify关键参数摘要把影响结果的参数排序后序列化再取哈希逻辑时间窗比如按天或按批次防止跨周期误去重拼起来长这样task:WO20240517-8821/action:dispatch/params:a3f9.../window:2024-05-17为什么第四个时间窗必须有因为有些场景是周期性重复的比如每天给这个组发一次日报提醒如果只按实体和动作去重第二天就发不出去了。加上日期窗天然按天隔离。为什么关键参数摘要而不是全部参数因为有些参数不影响结果比如 trace_id、重试次数、日志级别。如果把它们也算进去一次重试就会生成新的幂等键去重直接失效。我的做法是维护一个参与幂等计算的字段白名单显式声明不靠猜。注意幂等键一旦生成就不能变。我在代码里给它加了不可变约束任何地方想改它都会抛异常。因为一个变来变去的幂等键等于没有幂等。本地去重表用数据库唯一索引实现别用内存缓存进程重启就失效了。表结构就是幂等键做唯一索引加上状态和首次写入时间。插入冲突就说明是重复请求直接返回已有记录不执行。2.4 权限、白名单与审计让智能体别乱伸手智能体最大的风险不是能力不够而是能力太够。它可能调用你根本没打算让它调的工具可能传一个越权的参数。所以在触达层做权限控制比在提示词里写请不要做危险操作靠谱一万倍。我的做法是三层校验。第一层是工具级白名单哪些工具允许被智能体调用配置化声明没登记的通道直接拒绝。第二层是参数级约束每个通道声明自己的参数规则比如目标组必须是当前用户所属组单次派单数量不超过 20。这些规则写成声明式配置在validate阶段执行越界直接拒绝并记录。第三层是数据级隔离不同租户、不同业务线的数据互不可见在查询和写入时都带上隔离字段而不是靠应用层过滤。审计日志我要求记录五个必备字段调用者身份哪个智能体会话、幂等键、完整参数、执行结果、耗时。日志本身不做业务逻辑只追加不修改。这里有个小技巧参数落库前要做一次脱敏处理手机号、身份证、地址这类字段统一打码否则审计日志本身会变成合规风险点。3. 手把手落地从零搭一个能跑的最小版本理论讲完了说点能直接抄的。这一章我按搭积木的顺序从目录结构到代码骨架再到参数计算尽量给到可直接复现的细节。你做的时候不用完全照搬抓住结构就行。3.1 最小骨架与依赖选择目录结构我建议这样分边界清楚后面扩展不痛苦agent_reach/ channels/ 各通道适配器 core/ task.py 任务定义 action.py 动作与依赖 idem.py 幂等键生成与去重 executor.py 执行器与调度 retry.py 重试策略 guards/ permission.py audit.py obs/ metrics.py tracing.py依赖上我刻意保持极简一个异步 HTTP 客户端、一个数据库驱动、一个序列化库就这些。理由很直接触达层的核心诉求是稳依赖越少出问题的面越小升级时被第三方库拖着走的概率也越低。数据库我选了支持唯一索引和事务的关系型库。别用纯 K-V 存去重表因为你需要按状态、按时间去检索失败任务K-V 会很痛苦。异步框架用原生 asyncio 就够了不用引入额外的事件循环。3.2 一个能跑的通道适配器下面这个适配器实现了完整的四方法接口同时支持透传幂等头可以直接当模板改import hashlib, httpx class WebhookChannel: name webhook def __init__(self, base_url, token, supports_idem_headerTrue): self.base_url base_url self.token token self.supports_idem_header supports_idem_header self.client httpx.AsyncClient(timeouthttpx.Timeout(3.0, read8.0)) def validate(self, action): if not action.params.get(url): raise ValueError(missing url) if len(action.params.get(body, )) 64 * 1024: raise ValueError(payload too large) async def submit(self, action, idem_key): headers {Authorization: fBearer {self.token}} if self.supports_idem_header: headers[X-Idempotency-Key] idem_key resp await self.client.post( action.params[url], jsonaction.params[body], headersheaders ) if resp.status_code 500: raise TransientError(resp.text) if resp.status_code 400: raise PermanentError(resp.text) return resp.json().get(task_id, idem_key) async def poll(self, external_id): try: r await self.client.get(f{self.base_url}/tasks/{external_id}, headers{Authorization: fBearer {self.token}}) except Exception: return Status.UNKNOWN if r.status_code 404: return Status.UNKNOWN state r.json().get(state) return {done: Status.SUCCESS, failed: Status.FAILED}.get(state, Status.PENDING)两个细节要展开讲。一是超时拆开设置连接超时给 3 秒读取超时给 8 秒。连接超时短是因为连不上基本就是网络问题等再久也没用读取超时长是因为对方可能在处理给点耐心。很多人统一设 30 秒结果是慢调用把连接池占满整个系统雪崩。二是异常分类。这里我把异常明确分成TransientError可重试和PermanentError不可重试。这个区分太重要了重试一个永久性错误比如参数格式不对除了浪费配额没有任何意义。5xx 和超时归为可重试4xx 归为不可重试这是最基本的判断具体到每个通道还要按对方文档细化。3.3 重试退避与熔断参数怎么算出来重试策略我用的是指数退避加随机抖动公式delay min(base * 2^n, cap) * (1 random(-jitter, jitter))。参数取值我给一套实践过的base取 0.5 秒cap取 30 秒最大重试 4 次jitter取 0.3。算下来延迟序列大概是 0.5、1、2、4 秒加上抖动后总耗时在 10 秒左右。为什么是四次因为 0.5×2^4 8 秒已经接近大多数同步接口的用户可接受上限再往上延任务就该转异步了。jitter为什么必须有防惊群。如果 1000 个任务同时失败全部按同样的节奏重试对方系统会迎来四波整齐的流量高峰本来能恢复的也被打挂了。加 30% 抖动后请求被打散对方压力平缓很多。熔断我用的是简单的滑动窗口计数10 秒窗口内失败率超过 50% 且样本数大于 20就打开熔断持续 30 秒。为什么要有样本数下限因为 3 个请求失败 2 个也是 66%但那只是正常波动直接熔断会误伤。熔断打开期间请求直接快速失败不再真实调用给下游喘息时间。30 秒后进入半开状态放 5 个探针请求成功就恢复。注意重试要区分幂等动作和非幂等动作。非幂等动作用户层面根本不该重试只能靠前置查询确认状态。上线前一定把每个动作的幂等性标注清楚这是硬性要求。3.4 接进智能体主循环只暴露一个方法这层对智能体来说应该极度简单。我只暴露一个方法输入是动作描述输出是任务号async def reach(action_spec: dict) - str: action Action.from_spec(action_spec) channel registry.get(action.channel) channel.validate(action) idem_key build_idem_key(action) if (existing : await store.find(idem_key)): return existing.task_id await guard.check(action) await store.reserve(idem_key, action) try: external_id await channel.submit(action, idem_key) await store.update(idem_key, external_id, Status.PENDING) except TransientError: await scheduler.enqueue_retry(idem_key) except PermanentError as e: await store.fail(idem_key, str(e)) await alerts.raise_task(action.task_id, str(e)) await audit.log(action, idem_key) return action.task_id重点是store.reserve这一步它靠数据库唯一索引抢锁抢到了才执行抢不到说明有人在做同一件事直接返回。这个先占坑再执行的顺序不能反反过来就失去去重意义。给智能体的工具描述也要写得克制。我见过把十几个通道全塞给模型的结果它经常选错。正确做法是按业务场景裁剪一个场景只暴露三到五个工具工具名和描述里明确写清这个工具会产生外部副作用让模型知道轻重。3.5 可观测性三个指标定位八成问题触达层出问题时我基本只看三个指标覆盖了绝大多数场景。第一个是各状态的计数分布待执行、进行中、成功、失败、未知。这个分布一旦有异常比如未知持续上涨基本就是某个通道在抖动。第二个是端到端耗时分布我关注 P50、P95、P99 三个分位。P99 突然拉高但 P50 没变通常是少数慢请求或者某个通道开始限流。第三个是重试率按通道分维度看。某个通道重试率超过 5%就该去查对方文档或者联系对接人了。链路追踪上我给每个任务生成一个 trace 标识从智能体会话一路透传到外部调用。这样出问题时从用户反馈直接能定位到具体哪一次网络请求。日志我坚持结构化输出不写拼接字符串方便事后检索聚合。4. 常见问题与排查技巧实录这一章是纯经验都是我在生产环境里被真实咬过的。每条包括现象、原因、处理你可以当成一个速查清单用。4.1 重复触达最常见的那个坑现象是用户收到两条一模一样的通知或者一张单被派了两次。原因排下来无非四个。第一个是重试没带幂等键。请求超时了代码自动重试但重试时生成了一个新的幂等键下游认为是两次不同的请求。解决办法是幂等键在动作创建时就固定重试只复用不重建。第二个是去重表用了缓存。进程重启、缓存过期去重信息就没了。解决办法是缓存只做加速真正的判定必须落库用唯一索引。第三个是外部系统不支持幂等头。有些老系统就是没有这个能力。这种情况下只能在本地加状态查询提交前先查一次是否已存在虽然不能百分百避免但能挡掉绝大多数。第四个是并发窗口。两个请求几乎同时到达都查了一次不存在然后都执行了。解决办法是把查询和占坑做成一个原子操作靠唯一索引冲突来判定。4.2 任务卡在进行中不动这个问题的排查思路我固定成四步。先看这个通道的 poll 是否正常返回如果一直返回 PENDING说明对方状态没更新或者我们查错了任务号。再看外部任务号是否正确落库我曾经遇到过一次序列化问题任务号写进去多了个引号导致永远查不到。第三步看轮询器本身是不是活着有没有被某个异常循环卡死。第四步才是看对方系统。这个顺序很重要先从自己能控的部分查起比一上来就怀疑对方高效得多。处理上我加了一个停滞超时机制任何动作在 PENDING 状态超过设定时间一般按通道配置短的 2 分钟长的 30 分钟自动升级为 UNKNOWN 并触发人工核对。不要让它无限期挂着那才是真正的黑洞。4.3 限流与配额打满现象是某段时间大量请求返回 429 或者类似错误码。核心原因是我们只管发不管节奏。我的处理是给每个通道加令牌桶限流速率按对方文档给的配额打个七折配置留出余量。为什么要打七折因为对方统计口径可能和你的不一样它算的是 QPS你算的是每分钟调用数边界情况容易超。留 30% 余量能挡住大部分突发。另外限流错误要和普通错误区别对待它应该有独立的退避策略退避时间更长并且优先降级为异步处理而不是继续在高频重试上耗着。4.4 常见问题速查表现象高概率原因处理动作重复通知/重复派单幂等键重建、去重靠缓存、并发窗口固定幂等键、改唯一索引、原子占坑任务长期 PENDINGpoll 返回错任务号、轮询器卡死校验落库字段、加停滞超时升级大量 429无本地限流、突发流量令牌桶限速、降级异步失败率突增但对方正常参数被智能体生成得越界加强参数级校验、收窄工具白名单审计日志缺失埋点位置在异常分支之后把审计提到 finally 中执行5xx 后大量重复写把 UNKNOWN 当 FAILED 重试拆分 UNKNOWN独立延迟队列慢请求拖垮整体未拆连接/读取超时连接 3s、读取 8s 分开配置夜间批处理无声失败缺少失败告警按任务类型配置告警阈值这张表我压在团队 wiki 首页新人进来看一遍就能少踩一半坑。特别是最后一条无声失败是最可怕的它不是出错是没人知道出错了所以告警配置的优先级应该和功能开发一样高。5. 场景延展与容量估算把核心跑通之后我陆续把这套东西用在了几个不同场景上顺便做了些容量估算这块经验也挺值得分享。5.1 不同场景该怎么组合通道工单类场景核心动作是查询、校验、写入、通知。我的组合是查询和校验走内部 RPC要求同步、低延迟写入走内部 API 但配幂等头通知走消息通道且允许异步。这个组合的关键是把写和通知解耦写成功就必须通知到位但通知失败不应该让整个任务回滚。巡检类场景特点是批量、周期、可容忍延迟。这时候我会把触达层切成两档轻量检查走同步耗时的深度检查丢进队列异步跑用任务号串起来。巡检最怕的是任务堆积所以我给它单独设了一个并发上限避免把其他场景的资源挤掉。对外通知类场景比如给外部合作方推送数据对时效和准确性要求都高。这类我会额外加一层发送前二次校验把关键字段再核一遍同时把回执完整落库因为一旦出错对外的影响面比内部大得多。5.2 容量与成本怎么估估算我一般从三个数出发日均任务量、单任务平均动作数、单动作平均耗时。举个例子日均一万个任务平均每个任务三个动作单动作平均一点五秒那一天的触达总量是三万次序号化处理的有效负载约占 12.5 小时的机器时间。再乘上并发系数。因为大部分任务是突发性的我会按峰值 QPS 日均次数 / 有效小时 / 3600 × 峰值倍数峰值倍数一般取 5 到 8。这样算出来的并发量才是配置连接池和限流阈值的依据按平均值配一定会被打爆。存储上去重表是主要增长点。一条记录大概两百字节一天三万次就是六兆一年两个多 G完全可控。但如果高频场景上到百万级就得考虑分区或者定期归档了只保留近三个月的去重记录再往前的靠幂等键本身的时间窗来规避。注意去重记录不能删太早。保留期要覆盖最长的重试链路 最长的人工处理时长我一般设三个月起步宁可多占点存储也别在这个地方省。5.3 我个人踩过的几个坑第一个坑是过早抽象。一开始我设计了七种通道类型、完整的插件机制、热加载能力结果前三个月只用到两种通道抽象层反而成了每次改动的阻碍。后来我一狠心砍掉一半抽象代码量少了三分之一改动速度反而快了。结论是触达层这种基础设施抽象要跟着真实需求长别提前建。第二个坑是把校验放在模型那一侧。我一度把参数合法性交给提示词去约束结果模型十次里有两次不听话。后来把校验全部下沉到通道的validate里模型怎么折腾都没事因为它再能说也得过这道关。凡是能用代码强制的就别指望提示词。第三个坑是告警阈值拍脑袋。最早告警一响就发结果一天收到几百条大家自动忽略了。后来改成按通道和历史基线做动态阈值只有偏离基线三倍标准差才告警噪音少了九成真正的问题也终于有人看了。第四个坑是没有区分业务失败和技术失败。早期所有失败都走同一条告警业务侧收到大量技术告警技术侧收到大量业务告警双方都烦。后来在动作定义里明确标注失败类型技术失败给运维业务失败给业务方各看各的效率一下就上来了。最后分享一个我认为最值钱的小习惯每周花十分钟把上周所有进入 UNKNOWN 状态的动作捞出来逐条看一遍。这个动作看起来很小但它能帮你在问题变成事故之前就发现通道的隐患。我做过大概二十周其中至少有五次提前发现了对方接口的语义变更避免了大面积故障。触达这件事本质上不是把功能做出来而是把不确定性一点点关进笼子里笼子越结实上层的智能体才越敢放手动。
