★ FEATURED ARTICLE 我去年下半年密集做了一堆 Agent 项目从简单的工作流自动化到多智能体协作系统都有。demo 阶段确实很爽模型把工具一调、把流程一走感觉很多东西都打通了。但真放到生产环境里跑上一阵子我发现自己一直在跟一个看不见摸不着的问题较劲Agent 明明执行了任务但结果没有到达该到的地方Agent 明明注册了工具但关键时刻总是够不到正确的数据多个 Agent 一起干活却经常互相找不到对方在说什么。这个问题我想了很久最后总结成两个字触达。Agent 的能力再强如果它的决策、输出、状态、调用的工具、协作的消息不能可靠地“抵达”目标位置那这个 Agent 系统就是一堆华丽的空转代码。后来我把这套解决“智能体触达问题”的方法论和代码沉淀成了一个内部项目代号就叫 Agent-Reach。这篇博文就是我对 Agent-Reach 这个项目的一次完整复盘。我会把三层触达模型拆开讲透把我在真实项目里落地时的代码贴出来也会把踩过的坑和排查思路一并交代清楚。无论你是在做单 Agent 工具调用还是在搞多智能体协同只要你的系统里存在“Agent 干了活但别人不知道、够不到、连不上”的情况这篇东西应该都能给你一些直接能用的思路。1. 先搞清楚 Agent-Reach 到底在解决什么问题1.1 为什么 Agent 项目总是 demo 很好、生产拉胯先说个我自己的观察。很多团队做 Agent 项目最容易陷入的状态是跑通第一个端到端流程之后大家都很兴奋觉得核心逻辑已经没问题了剩下就是修修补补。但真实情况完全不是这样。模型调用工具的时候工具的 API 地址是从配置中心读的网络偶尔抖一下调用就失败了。Agent 自己不知道失败还会照着幻觉往下编。任务执行完之后结果写在日志里但用户没有收到任何通知以为系统卡住了。还有更隐蔽的两个 Agent 协作A 把任务结果放在自己的内存对象里B 完全不知道去哪里拿于是 B 又按自己的理解重新执行了一遍任务。这些问题的共性是什么不是模型推理能力不够也不是业务流程设计错而是 Agent 的结果、状态、能力和消息在整个系统里没有一条可靠的通路。我把这条通路叫做 Reach——可达性。没有 ReachAgent 就是一个孤岛再聪明也是孤岛。1.2 我给 Reach 下的定义三层触达模型在 Agent-Reach 项目里我把触达拆成了三层每一层解决一类问题。第一层是消息触达。Agent 产生的输出、通知、告警、结果摘要能不能可靠地送到该接收的人或系统手里。这一层最简单但最容易在细节上翻车。第二层是能力触达。Agent 要完成一个任务需要调用工具、读取知识库、访问上下文记忆。这些资源和能力是不是在 Agent“够得着”的范围内。很多 Agent 不是不会用工具而是根本不知道工具有哪些、参数应该怎么填、哪些数据有权限读。第三层是协同触达。在多 Agent 系统里一个 Agent 怎么发现另一个 Agent 存在怎么把任务委派出去怎么把结果传回来。这是我最晚开始做、也最庆幸做了的一层因为没有协同触达多 Agent 就是多个单体 Agent 硬凑在一起。这三层不是割裂的实际运行时它们相互依赖。比如一个 Agent 要发送消息消息内容可能需要另一个 Agent 产出的数据这就同时涉及消息触达和协同触达。设计的时候分层想落地的时候按层去梳理问题会清楚很多。1.3 这篇文章给谁看、能拿到什么我写这篇东西主要是给三类人参考。第一类是正在做 Agent 应用开发的朋友。你大概率已经跑通了一个原型接下来要把原型变成真正的产品触达能力就是你绕不开的坎。第二类是做后端平台、中间件、基础设施的工程师。你可能不直接写 Agent 的业务逻辑但 Agent 要接入现有系统靠的就是你负责的那些通路。我讲的可靠投递、事件总线、注册发现这些概念你一点都不会陌生。第三类是技术负责人或架构师。你更关心的是一个 Agent 项目能不能稳定地跑在生产环境里能不能拆成可维护的模块。我的三层触达模型可以直接拿去做系统设计时的检查清单。如果你读了之后发现“这些我都知道”那我会很高兴因为这说明你已经在正确的路上了。如果你读完发现自己踩过其中几个坑但没总结出来那这篇文章就更有价值了。2. 第一层触达消息与状态的可达性2.1 先想清楚谁需要知道 Agent 做了什么我见过不少 Agent 项目日志写得非常详细但用户根本不知道任务进行到哪一步了。问就是“它自己会跑不需要管”。这种想法在产品原型阶段没问题但一旦 Agent 承担的是真实业务动作不通告就等于没有执行。做消息触达之前先列一个清单谁是这个 Agent 执行结果的利益相关方。可能是一个等待审批结果的运营人员可能是需要把数据落库的下游系统也可能是另一个等待输入才能继续的 Agent。把这些角色列出来之后再去设计触达的路径而不是先选技术方案再想发给谁。我自己习惯用一个简单的表格来梳理消息类型接收方时效性要求丢失容忍度需要反馈吗任务完成通知用户分钟级不能丢需要异常告警值班群秒级不能丢需要过程日志摘要存储系统小时级可容忍不需要中间结果下游 Agent秒级不能丢需要梳理完之后你会发现不同消息的触达策略是完全不一样的。告警消息需要多渠道备份过程日志只需要异步写入而 Agent 之间的中间结果则需要有去重和确认机制。一视同仁地处理所有消息是很多触达问题的根源。2.2 触达通道怎么选没有银弹通道选型是个特别容易被轻视的环节。很多人觉得“发个消息有什么难的”可真到生产环境就发现每条通道都有自己的脾气。Webhook 是最灵活的基本任何系统都能接但 Webhook 的问题在于回调地址经常没人维护而且目标服务挂了你就得自己处理重试。IM 机器人比如飞书、钉钉、企微触达效果好用户及时性好但都有频率限制和审批限制不适合大批量消息。邮件触达的正式性好适合报告类内容但延迟高而且容易被当成垃圾邮件。短信是最强触达但成本高内容审核严格只适合最高优先级的告警。我的建议是不要选单一通道而是按消息的丢失容忍度做分级。关键消息走双通道比如同时发 IM 和邮件普通消息单通道批量消息走存储。这套做法的核心不是技术难度而是“不要把所有鸡蛋放进一个篮子”的运维常识。有个细节容易被忽略通道的幂等性。同一个消息因为网络重试被发送了两次接收方如果不知道去重用户就会收到两条一模一样的内容。这个问题我在 2.3 里会详细讲。2.3 可靠投递的细节重试、幂等、去重消息触达的可靠性本质上就是三个关键词重试、幂等、去重。先说重试。网络调用没有 100% 成功这一说所以发送失败必须重试。但重试不能无脑重试要有退避策略。我之前一开始用固定间隔重试结果下游服务一直没恢复的时候重试风暴把对方的网关打崩了。后来改成指数退避加随机抖动大幅降低了重试对下游的冲击。再说幂等。所谓幂等就是同一条消息无论被发送多少次接收方处理的效果和只处理一次是一样的。实现起来就是在消息里带上唯一 id接收方根据 id 判断这条消息是不是已经处理过了。没有幂等设计一次微小的网络重试就可能造成业务上的重复扣款、重复建单这类事故。最后说去重。去重其实是幂等在发送端的体现。发送方维护一个最近消息 id 的集合如果短时间内要发同一条消息直接丢弃。这个逻辑放在发送端可以减少无谓的网络调用但真正可靠的去重一定要放在接收端做因为只有接收端才有能力判断“这条消息我到底收没收到过”。2.4 别忘了反馈回路消息触达如果把“把消息发给用户”当成终点那就少做了一半。真正的触达应该是双向的Agent 把消息发给用户用户能把反馈传回给 Agent。这个反馈可以是简单的按钮确认也可以是一条自然语言回复。我在 Agent-Reach 里给所有关键消息都绑定了一个回执机制。消息体里带一个 reply_token用户点击确认或者回复一句话系统就把这个 token 和用户输入一起传回给 Agent。这样 Agent 收到的不再是一条程序化的事件而是带着上下文和用户意图的反馈。这一层设计在初期看起来好像没那么必要但一旦 Agent 要开始承担需要人工确认的动作比如审批、下决心执行高风险操作你会发现反馈回路是必须的。没有反馈Agent 只能靠猜来确认用户意图这就回到了最原始的不可控状态。3. 第二层触达能力与上下文的可达性3.1 工具注册让 Agent 知道“自己有什么可用”如果说消息触达解决的是“外部世界怎么接收 Agent 的动作”那么能力触达解决的就是“Agent 怎么够到外部世界的能力”。这两件事正好是反方向的。很多 Agent 在能力方面出问题不是因为模型不会调工具而是因为工具的暴露方式对模型不友好。模型不是人它不理解“你调用那个订单系统的接口就行”这种话。它需要的是清晰、完整、没有歧义的工具描述。所以工具注册这件事看起来简单实际上决定了一个 Agent 的能力上限。我在 Agent-Reach 里为每个工具维护了一份完整的描述信息包括工具名称、一句话简介、参数 schema、使用场景和示例。这份描述会被组装进发给模型的上下文中让模型在决策的时候有足够的信息依据。这里有个容易犯的错工具描述贪多求全。描述太长会挤占模型的有效上下文参数 schema 太复杂会让模型不知所措。我的经验是一个 Agent 单轮可选的工具尽量控制在 15 个以内每个工具的描述不要超过 200 字参数只保留必填项和最关键的选填项。3.2 上下文触达RAG、记忆与长上下文的取舍Agent 第二次碰到同一类问题时它需要能想起来上次是怎么解决的这就涉及上下文触达。上下文触达可以简单分成两块一块是静态的知识库另一块是动态的记忆。静态知识库现在大家普遍用 RAG 来做把文档切块、向量化、存进向量数据库查询时按相似度取回。但我要提醒一点RAG 的召回质量决定了 Agent 回答的下限。向量检索只解决“有没有相关内容”不解决“这些内容对不对”。所以我在实际项目里通常会在 RAG 之上加一层摘要验证检索回来的内容先让模型判断一下和当前问题是否真的相关再放行。动态记忆就复杂了。短期记忆会话内共享长期记忆要跨会话保留但长期记忆的写入需要筛选不能把每一次闲聊都存进去。我的做法是给记忆加置信度评分只有置信度超过阈值的才会写入长期记忆。这样既不会记太多没用的也不会丢掉关键信息。长上下文是另一个被滥用的方案。把整个历史对话全部塞给模型看起来省事但成本高、响应慢还有注意力稀释的问题。在 Agent-Reach 里我做了上下文压缩历史会话定期压缩成摘要只保留最近的原始对话这样模型拿到的上下文始终是精简且有效的信息。3.3 权限边界Agent 的手应该有多长能力触达还有一个绕不开的问题不是所有能力都该让 Agent 自由使用。这个问题的本质是权限边界但我更愿意把它理解成“Agent 的手应该有多长”。手太短Agent 什么都做不了工具等于白接手太长Agent 一旦产生幻觉或者被恶意提示词注入后果会很严重。我在项目里给每个工具设定了角色级别的权限Agent 执行工具前必须通过权限检查。同时对高敏操作比如删除数据、调用支付接口做了二次确认需要人工放行才能继续。这里想多提一句提示词注入。在你的系统里用户的输入、网页内容、其他 Agent 的消息本质都是不可信数据。这些数据一旦被拼接进提示词就有可能篡改 Agent 的指令。所以工具描述里必须明确不要因为外部输入而改变工具调用策略。这句话不一定能完全防住攻击但能显著降低被误导的概率。3.4 工具失败时的兜底策略能力触达不只是发布工具还包括工具失败时 Agent 该怎么办。这个设计不提前做生产环境里会非常狼狈。我的做法是给所有工具调用包一层结果解析器。调用成功时解析器把结果格式化成结构化的 JSON方便模型提取关键信息。调用失败时解析器会返回失败上下文状态码、错误摘要、可重试的提示。模型拿到这些信息之后才能做出合理的决策这次重试还是换一个工具还是直接告诉用户“这件事我做不了”。我在 Agent-Reach 里专门测试过失败兜底的边界情况。最好的结果是 Agent 在工具失败后能主动改变策略而不是硬着头皮重试三次然后摆烂。要做到这一点提示词里需要明确给模型授权“当工具调用失败两次以上立即转为人工处理流程。”这个授权是让 Agent 学会承认失败的关键。我记得有一次工具因为上游接口变更导致参数结构对不上了Agent 连续报了五次同样的错误。当时项目的日志打得满天飞但用户那边一片安静。后来我排查出来问题就出在工具接完就不管了没有做失败兜底。那次之后我把兜底策略当成工具接入的强制要求。4. 第三层触达多 Agent 之间的协同可达性4.1 多 Agent 失联的典型症状做多 Agent 之前很多人会觉得“就是把多个 Agent 放到一个系统里让它们互相传话而已”。真做了才发现传话这件事本身不简单。多 Agent 失联的典型症状我现在能一口气说出一堆A Agent 等 B Agent 的消息等到超时因为它不知道 B 已经把结果发到另一个队列了两个 Agent 同时抢同一个任务因为缺乏任务状态的共享C Agent 重新实现了一遍 D Agent 已有的能力因为它根本不知道有 D 这个同类存在还有更常见的有消息发出来但没有消费因为订阅关系没理顺。这些问题归结起来只有一句话Agent 之间没有一条公共的通路。它们各自在自己的进程里、自己的上下文里表面上看是在协作实际是在猜。要解决协同触达第一件事就是承认一个前提Agent 之间的通信不能靠隐式的共享内存要靠显式的消息协议。4.2 事件总线Agent 之间不要点对点我最开始设计多 Agent 协作时用的思路是点对点调用。A 直接调用 B 的接口B 直接调用 C 的接口。跑通了但扩展性特别差而且一改业务逻辑就牵一发动全身。后来我把协作方式改成了事件驱动所有 Agent 的产出都发布到一个事件总线谁需要这个事件就自己去订阅。这样设计的好处是A 不需要知道哪些 Agent 对它的产出感兴趣B 也不需要知道数据是从哪个 Agent 那边来的。发布方只管发布订阅方只管订阅两边的耦合降到最低。实现事件总线可以用 Redis Stream、Kafka也可以用轻量的 MQ。重点是定义好事件的格式。我在 Agent-Reach 里给事件设计了一套最小公共字段event_id、event_type、producer、timestamp、payload。这个格式可以帮助下游快速判断要不要响应和接收再决定如何处理。我见过不少团队用队列把事件传递做通之后就以为万事大吉了。但实际上事件驱动最大的坑是事件语义的漂移同样的 event_type发布方改了一次 payload 结构订阅方没跟上直接解析失败。这种问题通常在半夜爆发处理的代价远高于一开始就约定好 schema 的成本。4.3 RegistryAgent 之间怎么互相发现事件总线解决的是消息怎么流动但还有一个前置问题Agent 彼此怎么知道对方存在。这就需要服务注册与发现了。在 Agent-Reach 里每个 Agent 启动后都会向一个中心化的 Registry 注册自己的信息包括 Agent 名称、能力描述、当前状态。Registry 会定期做心跳检查和状态更新发现某个 Agent 长时间没心跳就标记为不可用。这样当其他 Agent 需要协同时可以先查询 Registry找到“谁有这个能力且当前可用”。你可能觉得这听起来像微服务架构里的服务发现没错本质上就是同一套东西。多 Agent 系统就是一个特殊的分布式系统分布式系统踩过的坑它一个都不会少。所以不要觉得 Agent 是新技术就不去借鉴服务发现的成熟方案。Registry 的更新要有一定的延时容忍。Agent 刚注册完下游可能等几秒才能看到这是正常的。但如果 Agent 经常闪崩重启Registry 的公告板就会频繁变化下游的请求就会不断打到一个刚死掉的 Agent 上。我的做法是给 Registry 增加一个“冷却期”Agent 挂掉之后Registry 不立即把它从可用列表移除而是先标记异常等它恢复。这听起来和“快速失败”的理念有点矛盾但在协同触达这个场景里短暂容忍异常比频繁重新路由更稳定。因为 Agent 重启本身可能只需要几秒而路由状态在系统里传播需要的时间往往更长。4.4 协作模式怎么选请求、发布订阅与委派有协同就有模式的选择这直接决定你的多 Agent 系统长什么样。请求-响应模式最简单适合 A 需要 B 立即给结果的场景。B 的处理是同步的A 等 B 的响应。在这种模式下超时设置一定要合理不要等一个慢 Agent 等得太久。发布-订阅模式适合广播的通知类消息。一个 Agent 发布事件多个 Agent 订阅并各自响应。这个模式的好处是扩展性极好但要注意别让事件风暴发生一个事件触发十个 Agent 响应十个 Agent 又各自发布新事件最终把整个系统的消息量打爆。任务委派模式适合把大任务拆解成小任务分发给多个 Agent 执行的场景。委派时要带上完整的任务上下文和验收标准否则接收方只能靠猜来完成。我见过很多委派失败的案例最后发现是因为委派方觉得这个任务很简单一句话没说清楚接收方理解错方向就开始埋头干。我做 Agent-Reach 时的总体感觉是不要迷信单一模式。一个健康的 Agent 协同系统可能同时跑着请求、订阅和委派三种流量。关键不是选哪个而是每种流量要有清晰的边界不要混在一个通道里互相影响。5. 落地实战给一个真实 Agent 系统补上触达能力5.1 系统背景与改造前后对比这块我拿一个真实的项目来演示。这是一个内部工单处理系统原来的架构是用户提交工单一个主 Agent 接收并解析然后调用工具查库存、查历史记录最后把处理结果写入工单系统。看起来通了但用户反馈“工单提交了之后半天没消息不知道到底处理了没有”。这就是典型的消息触达缺失。后来我引入了 Agent-Reach 的思路对系统做了三层改造消息层面工单每进入一个新状态就发一条事件到通知服务由通知服务按优先级分发到 IM 和邮件能力层面把所有工具统一接入工具注册表加上权限检查和参数校验失败时自动告警协同层面把原来单主 Agent 改成“主 Agent 处理 Agent 通知 Agent”的结构三者之间靠事件总线协作。维度改造前改造后结果通知只有日志事件驱动多渠道分发工具调用散落在代码里统一注册、鉴权、兜底多 Agent 通信点对点调用事件总线 Registry人工确认不支持回执反馈这套改造大概花了一周时间核心代码不算多但设计上的梳理占了大头。5.2 消息触达模块的代码实现消息触达的核心是一个带重试和去重的发送调度器。这里我给出一个简化版本直接可以抄作业import asyncio import hashlib import json import logging import random logger logging.getLogger(__name__) class MessageDispatcher: def __init__(self, max_retry3, base_delay1.0): self._channels {} self.max_retry max_retry self.base_delay base_delay self._seen set() def register_channel(self, name, sender): self._channels[name] sender def _msg_id(self, channel, message): raw f{channel}:{json.dumps(message, sort_keysTrue)} return hashlib.sha1(raw.encode()).hexdigest()[:16] async def send(self, channel, message): msg_id self._msg_id(channel, message) if msg_id in self._seen: logger.warning(duplicated message dropped: %s, msg_id) return False self._seen.add(msg_id) sender self._channels.get(channel) if not sender: raise ValueError(fchannel {channel} not registered) for attempt in range(1, self.max_retry 1): try: await sender(message) return True except Exception as exc: delay self.base_delay * (2 ** (attempt - 1)) random.uniform(0, 0.5) logger.warning(send to %s failed: %s, retry in %ss, channel, exc, delay) await asyncio.sleep(delay) logger.error(send to %s failed after %s attempts, channel, self.max_retry) return False这个实现里有几个细节说一下。_msg_id是幂等键基于通道名加消息内容计算目的是让同一条消息重试时不重复触发。_seen集合实现发送端去重。random.uniform的抖动是为了防止多个 Agent 同时重试打爆下游。指数退避从 1 秒开始翻倍尝试 3 次后放弃并记录错误。5.3 能力触达模块的代码实现能力触达的核心是一个工具注册表。每个工具接入之前都必须注册注册的内容会进入 Agent 的工具上下文。下面是一个带权限检查的简化版本TOOL_REGISTRY {} class ToolNotFoundError(Exception): pass class PermissionDeniedError(Exception): pass def register_tool(name, description, schema, required_rolesNone): def decorator(func): TOOL_REGISTRY[name] { name: name, description: description, schema: schema, func: func, required_roles: required_roles or [], } return func return decorator async def dispatch_tool(name, arguments, rolesNone): tool TOOL_REGISTRY.get(name) if not tool: raise ToolNotFoundError(name) allowed set(roles or []) required set(tool[required_roles]) if required and not required.issubset(allowed): raise PermissionDeniedError(name) return await tool[func](**arguments)使用时的写法很直观。注册一个查询库存的工具register_tool( namequery_inventory, description查询商品库存参数 sku 为必填项, schema{ type: object, properties: { sku: {type: string, description: 商品编码}, warehouse: {type: string, description: 仓库编号可不填} }, required: [sku] }, required_roles[support] ) async def query_inventory(sku, warehouseNone): # 真实逻辑调用库存系统 API return {sku: sku, stock: 32, warehouse: warehouse or default}这个实现的好处是工具描述和函数逻辑完全解耦。后续要调整工具说明不需要改业务代码只要改注册函数上的元信息即可。权限检查发生在工具真正执行之前调用失败时抛出具体的异常类型方便上层做兜底。5.4 协同触达的事件总线实现多 Agent 协同我在生产里用的是 Redis Stream因为它和现有的 Redis 集群能复用运维成本低。一个极简的版本可以基于 asyncio 的队列实现但真实生产环境我建议还是用可靠的中间件。这里给出接口层的设计思路和关键代码class EventBus: def __init__(self, client): self._client client self._subscriptions {} async def publish(self, stream, event): record { event_id: event[event_id], event_type: event[event_type], producer: event[producer], timestamp: event[timestamp], payload: json.dumps(event[payload]) } await self._client.xadd(stream, record) def subscribe(self, stream, group, consumer): 消费组订阅同一事件只被组内一个消费者处理 # 具体实现依赖 Redis Stream 的 xgroup/xreadgroup 命令 pass事件总线设计里最关键的部分不是 publish 和 subscribe 本身而是事件 schema 的约定。我强烈建议在实际开工前把所有事件的 event_type 和 payload 结构用一份文档或 Protobuf schema 固定下来后续变更要走兼容升级而不是直接改。否则你会被事件结构不一致的 bug 反复折磨。5.5 实测效果与参数选择改造完成后我做了为期两周的观察重点记录了两个指标消息送达成功率和工单处理时效。消息送达这块在重试和去重机制上线后推送类消息的送达率从原来的 97% 提升到了 99.9% 左右绝大部分失败的场景出现在 IM 接口的限流频控上。之前没有去重时用户偶尔会收到重复的工单状态通知后来加上去重后这类问题彻底消失。工单处理时效方面原来用户平均要等 40 秒才能看到处理结果改造后能在 10 秒内收到事件通知。收益主要来自两个地方一是事件驱动让通知 Agent 可以并行启动不需要等主 Agent 的整个流程跑完二是工具调用失败时兜底策略能立即通知处理人而不用等超时再人工发现。参数选择上我也踩了一些坑。重试次数定成 3时间基数定成 1 秒是在“快速失败”和“给下游留恢复时间”之间找的平衡点。如果你下游服务恢复速度比较慢可以适当把 base_delay 调大到 2 秒但总超时时间不要超过 20 秒否则链路等待成本太高。6. 问题排查实录触达失败到底怎么查6.1 高频问题速查表途中摘了一些我在项目里遇到过多的问题整理成速查表你在排查时可以直接对照症状可能原因排查入口Agent 说已执行用户没收到消息触达链路断了通道未注册、回调地址失效查消息发送日志看是否有 failed 记录同一条通知用户收到多次缺少幂等键或去重逻辑查消息 id 是否重复生成工具调用报参数错误schema 定义和真实函数签名不一致比对注册时的 schema 与实际参数Agent 一直调同一个失败工具兜底策略未生效模型未被告知允许放弃检查工具结果解析器是否返回了失败上下文多个 Agent 重复执行同一任务缺少任务状态的共享事件被多个消费者重复处理检查事件消费组的 idempotent 设置A 发的消息 B 收不到事件类型名不一致或订阅关系未建立查看事件总线的 consumer 列表Agent 感知到工具但权限不足权限角色未配置完整检查 required_roles 和调用方 roles这张表不是拿来背的而是在遇到问题时先快速定位是哪一层触达出了问题再对照表里的排查入口去看日志。6.2 一次消息丢失的完整排查过程有一次线上告警用户在工单群里喊“处理结果没收到通知”。我第一反应是消息发送失败于是去查通知服务的日志。结果发现日志里根本没有这条消息的发送记录说明消息根本没进入发送链路。往源头定位发现工单系统更新状态的逻辑抛了一个异常状态更新失败后续的事件发布代码根本没走到。再往下挖发现异常来自 Redis Stream 的一个字段类型不匹配。旧版本的事件里 payload 是一个字符串新版本的代码把 payload 写成了对象导致序列化时报错。这个案例很典型消息丢失的背后不一定是发送端的问题很可能是链路源头的数据结构变更。这里特别强调一下兼容性测试的重要性因为触发它的问题往往要在数据形态变化时才会暴露。修改方式也很简单publish 之前在代码里做一次 schema 校验不合法的事件及时拦截并告警不让脏数据进入总线。另外事件总线里的消息我建议都保留原始 JSON 版本号升级 schema 时按版本号做兼容迁移不能直接覆盖。6.3 三个让我印象深刻的坑最后一个部分分享三个我踩得比较深的坑都是在文档里很难找到的。第一个坑是发送端去重集合无限膨胀。我最初用set()保存消息 id刚开始内存很小很 nice但系统跑了一个月后set 里积累了上千万个 id内存直线上升。后来我换成了带 TTL 的缓存结构比如 Redis 或带过期时间的 LRU只保留最近 5 分钟的消息 id因为重试导致的重复场景一定会在这 5 分钟内发生。第二个坑是 Agent 的工具数量太多导致模型选错工具。工具一多描述一长模型在决策时会有明显的注意力漂移经常把订单系统的工具用到库存系统上。后来我把工具做了分组路由先根据用户请求粗选一个工具组比如“库存相关”“订单相关”再把组内工具的描述传给模型。这个改动上线后选错率降了一半还多。第三个坑是反馈回执的死循环。我在回执里加了一个“如果用户情绪激烈就直接转人工”的逻辑结果有用户连续回复了几条催促消息每个词都触发了情绪识别阈值系统就一直在“转人工”和“让 Agent 回复”之间反复切换。后来加了一个频率限制同一个工单在 5 分钟内只能转人工一次。这个修复让我意识到反馈回路设计里必须有节流机制不然反馈本身就会成为新的噪音源。我个人在实际操作里最深的体会是Agent 系统的难点从来不在模型本身而在于它周围那一圈笨重的工程细节——消息能不能到、工具能不能调、对方能不能收到、收到后能不能确认。这些东西单个拿出来都不难但组合在一起、跑在真实流量里就会变成一场持续不断的排障战。Agent-Reach 这个项目做到最后我最大的收获不是代码而是这套排查的框架和心态面对 Agent 问题永远先分层先搞清楚是消息、能力还是协同的问题再动手而不是上来就重发一遍、重试一遍。这样你才不会在混乱的日志和数据里越陷越深。
阅读完成 · 觉得有帮助？
