上个月排查一条线上异常任务的时候我在跟踪日志里翻了四十多分钟最后发现根本不是模型回答出了问题而是两个 Agent 之间的任务消息在传递过程中被静默弄丢了。就是那种代码里完全查不到异常、但整条链路已经断掉的状态。也是从那次之后我把 Agent-Reach 这个项目从“有时间再做的事”提前到“立刻必须做”。这篇文章想把 Agent-Reach 的设计思路和我在真实环境里跑通的过程完整展开讲一遍。它不依赖任何特定的大模型厂商也不是要你学一套新语言核心就解决一件事智能体之间发任务消息怎么发得出去、发得有回执、丢了能查出来。如果你正在做多 Agent 编排、自动化任务流或者被“某个 Agent 调了另一个 Agent 却没反应”这种事折磨过这篇内容应该能帮你少踩几个坑。1. 多智能体协作里最容易丢消息的位置Agent-Reach 想解决的痛点1.1 一次审计日志里的完整丢包过程先说当时的场景。我维护的自动化客服链路里有三个 Agent意图识别 Agent、知识检索 Agent、生成回答 Agent。用户提问进来之后意图识别 Agent 判断出这是一个售后问题于是把检索任务发给知识检索 Agent检索完成后再把结果交给生成回答 Agent。整个链路看起来挺顺的问题出在某天晚上的一次线路抖动。知识检索 Agent 处理到一半向生成回答 Agent 发起了一次调用结果超时。由于我当时写的是最简单的 HTTP 同步调用超时之后既没有重试也没有把任务存下来代码只是打印了一行 warning 就继续往下走了。结果就是用户那边只看到“系统繁忙”而日志里看不到任何报错记录。那行 warning 还是我在当天白天为了调调试加的如果不加这个失败的调用连痕迹都不会有。也就是说大多数多 Agent 系统里最先“丢”的不是模型能力而是 Agent 之间的协作消息。这个教训让我意识到多个 Agent 串成一条链路的时候真正值得优先做好的工程问题是它们之间怎么把任务可靠地递到对方手里。1.2 Agent-Reach 管什么、不管什么我想了很久给这类问题划了一条边界。Agent-Reach 不是去替代 Agent 的推理能力也不管你底层的 LLM 是哪一家它只负责“任务消息可靠触达”这件事。具体来说它管四件事投递把任务消息从发起方送到接收方不直接依赖对方进程是否在线。回执接收方处理完或处理失败之后把状态返回给发起方或审计系统。重试消息处理失败时按既定策略重新投递而不是直接放弃。可观测每条消息在所有环节的状态都有记录能查、能重放。它不管三件事不管 Agent 内部的 Prompt 怎么写、模型怎么推理。不管业务结果本身对不对它只保证“结果有明确终态”。不解决所有极端一致性问题比如分布式事务最终一致性只做到“不静默丢消息”这一层。这层边界很重要。如果什么都想做这套系统就会变成一个沉重的流程引擎反而不利于在项目里落地。1.3 为什么我不想在 Kafka / RabbitMQ 上裸写任务语义很多人会说消息可靠性问题用现成的 MQ 不就行了吗我在实践里发现只靠 MQ 远远不够。Kafka 非常擅长高吞吐的流式数据但它默认的消费模型是“消费者组内竞争消费”而 Agent 协作更像“请求—应答”。你需要在 Kafka 之上自己实现一套回执、超时、幂等等概念这套东西代码量相当可观。RabbitMQ 有路由、有确认机制功能很全但“全”也意味着决策成本高。你要自己想清楚消息确认之后如果处理失败怎么办重试是通过死信队列还是消息 TTL重投时如何避免对方重复执行两个 Agent 之间如果互相等回执超时由谁负责每个问题都有好几种标准做法但组合在一起就形成一个非常自由的“弹性空间”。每个人凭直觉选选完就是一堆隐式约定只能靠口口相传。后来团队扩大的时候新同事接手的成本非常高。所以 Agent-Reach 做的事情是把这层语义固化下来触达令牌、意图路由、三层回执。MQ 组件可以换但 Agent 之间的协作约定是统一的。这样不管是 Kafka、RabbitMQ 还是 Redis Stream底层用什么只是实现细节上层暴露的接口一直是同一套。2. Agent-Reach 的触达握手模型令牌、意图路由与三层回执2.1 触达令牌给每条任务发一张实名快递单我在 Agent-Reach 里定义了一个数据结构叫 ReachToken。你可以把它理解成一张快递单一张单子对应一次任务投递单子上写了寄件人、收件人、物品描述、有效期和唯一单号。ReachToken 的核心字段如下字段含义备注msg_id消息唯一 ID重试时保持不变用于幂等trace_id链路追踪 ID跨多个 Agent 的完整任务链intent任务意图类型如retrieval.requestsource_agent发起方标识审计用target_agent目标方标识不用于路由只用于审计ttl_ms令牌有效期超过则拒绝处理payload业务数据任意 JSON 结构有一个细节值得单独说trace_id 和 msg_id 要分开不要混用。Trace_id 是一条完整业务链路的ID从用户提问到最终回答都是同一个而 msg_id 只是当前这一段“A 发给 B”消息的唯一标识。B 处理完之后它再发给 C 时会产生新的 msg_id但 trace_id 保持不变。这样做的好处是排障时可以按 trace_id 查出整条链路所有跳数的回执时间也能按 msg_id 看某一跳是否发生过重试、是不是同一批消息重复消费了。2.2 意图路由按“想做什么”分发而不是按“找哪个进程”第一版 Agent-Reach 的路由我做成过 target_agent 直连类似“我就是要发给 agent_b”。但在灰度验证的时候立刻发现问题如果我临时把某个 Agent 换成新版本或者原地扩容多了一个实例路由规则就得跟着改。发消息方鲁棒性太差。所以后来我改成了按intent路由。投递端只关心“我想发什么类型的任务”不关心“谁注册了这个类型的消费”。下游消费方启动时注册自己感兴趣的 intent上游完全解耦。这个设计有点像我常举的例子你网购的时候只需要写“家电类订单”快递公司会根据当前运力安排配送员而不是你必须在下单时指名道姓指定某个配送员。Agent 版本升级、实例扩缩容对上游来说是无感知的。intent命名我推荐用两级结构比如域.动作retrieval.request、policy.check、notification.push。这样后面做权限控制和审计都比随便起名字清晰得多。2.3 三层回执RECEIVED / ACCEPTED / COMPLETED 各代表什么Agent-Reach 里最让我受益的设计就是三层回执。一开始我想得很简单A 给 B 发消息B 处理完回一个 ACK 不就行了但实际跑起来发现不够。原因在于“收到了”和“处理完了”是两件完全不同的事。消息进了 B 的消息队列不代表 B 已经开始处理B 开始处理了也不代表处理结果是对的。如果只有一层 ACKA 看到 ACK 就会以为任务已经完成但等它拿结果时才发现 B 其实还在排队。Agent-Reach 的三层回执完整定义如下回执状态含义触发时机RECEIVED触达层已持久化消息消费方未开始处理消息写入队列时ACCEPTED目标 Agent 已取走消息并开始处理消费者接收消息、通过校验后COMPLETED业务处理结束带结果或错误信息handler 正常返回或抛异常时A 发起任务后只需要盯着回执状态流转。如果在预期时间内还没到 COMPLETED就能判断当前卡在哪一层卡在 RECEIVED消息还没被消费者取走说明消费方可能挂了或者队列积压。卡在 ACCEPTED消费者在处理中可能业务执行比较慢需要结合耗时来分析。一直没有 RECEIVED 之外的回执消息可能压根没进队列问题在投递端或网络。这三层状态让“排查”这一步变得极其高效。2.4 状态机里那条容易漏掉的 FAILED 路径三层回执之外有一条路径最开始我没设计COMPLETED 出了异常分支怎么办后来我在真实环境里发现Agent 的 handler 抛异常的概率远高于单机服务。因为 Agent 处理逻辑里经常要调用 LLM 接口模型限流、响应超时、返回格式异常都会导致处理中断。所以我给最终状态补了一个分支COMPLETED处理成功和FAILED处理失败FAILED 状态下回执里必须携带error_code和error_message。同时FAILED 之后是否重试由投递端根据策略决定而不是由接收方自动无限重试。这样避免了一个处理不成功的任务在系统里来回弹跳直到打满资源。3. 30 分钟跑通一个小闭环基于 Redis Stream 的实现记录3.1 环境选型为什么是 Redis Stream 而不是 MQ选 Redis Stream 而不是 Kafka 或 RabbitMQ主要原因是落地成本。如果项目里已经用了 Redis加 Stream 只是几行配置的事不需要额外维护一套消息集群。对于 Agent 协作这种“单跳消息量不大、但对可靠性和可观测性要求高”的场景Redis Stream 的持久化能力和消费组模型完全够用。另一个原因是 Redis Stream 的消费组模型非常直观。每个 Agent 对应一个消费组组内多个消费者竞争同一批消息一个消息只能被组内一个消费者处理天然支持水平扩容。当然也要清楚它的边界Redis 如果只开默认持久化配置极端故障下可能丢消息。建议至少开启 AOF 持久化并且给 Redis 做主从。如果你对持久性要求极高把底层的存储换成 Kafka 也不影响 Agent-Reach 的上层模型这是它设计上刻意保持的松耦合。3.2 投递端xadd 与回执写入的事务一致性先看投递端核心实现。import json import time import uuid import redis class ReachToken: def __init__(self, intent, source_agent, payload, target_agentNone, ttl_ms10000): self.msg_id str(uuid.uuid4()) self.trace_id None self.intent intent self.source_agent source_agent self.target_agent target_agent self.payload payload self.ttl_ms ttl_ms self.created_at_ms int(time.time() * 1000) def dispatch(redis_conn, token: ReachToken) - None: 把任务消息写入意图对应的 Stream并写入 RECEIVED 回执。 if not token.trace_id: token.trace_id token.msg_id queue_key fqueue:{token.intent} receipt_key freceipt:{token.msg_id} pipe redis_conn.pipeline(transactionTrue) pipe.xadd(queue_key, {token: json.dumps( token.__dict__, ensure_asciiFalse)}, maxlen100000) pipe.xadd(receipt_key, { status: RECEIVED, intent: token.intent, at_ms: int(time.time() * 1000), msg_id: token.msg_id, trace_id: token.trace_id, }) pipe.execute()关键点在于pipeline(transactionTrue)。这一步保证任务消息进队列和 RECEIVED 回执写入同时成功或同时失败。如果分开执行可能出现消息进队列了、回执没写上的情况到时候发起方以为没发出去重试一次就变成两条重复任务。maxlen100000是给队列设的一个上限。Agent 协作消息不需要无限保留保留太多处理不过来的历史反而浪费内存。按 10 万条来算正常量级的 Agent 协作场景至少可以覆盖数天的消息量。3.3 消费端xreadgroup、幂等抢占与 TTL 校验消费端比投递端复杂一些因为要处理幂等、TTL 和回执三个问题。import time import json import redis REDIS_CONN redis.Redis(host127.0.0.1, port6379, decode_responsesTrue) def mark_receipt(token, status, error_codeNone, error_messageNone): REDIS_CONN.xadd( freceipt:{token.msg_id}, { status: status, at_ms: int(time.time() * 1000), msg_id: token.msg_id, trace_id: token.trace_id, error_code: error_code or , error_message: error_message or , }, maxlen1000, ) def check_ttl(token) - bool: now_ms int(time.time() * 1000) return (now_ms - token.created_at_ms) token.ttl_ms def try_acquire(token) - bool: 幂等抢占同一 msg_id 只允许一个消费者开始处理。 acquired REDIS_CONN.set( flock:{token.msg_id}, token.source_agent, nxTrue, ex60 ) return acquired is not None and acquired is True def consume_loop(agent_name: str, intents, handler): streams {fqueue:{intent}: for intent in intents} for intent in intents: try: REDIS_CONN.xgroup_create(fqueue:{intent}, agent_name, id0, mkstreamTrue) except redis.ResponseError: pass # 消费组已存在 print(f[{agent_name}] 开始监听: {intents}) while True: results REDIS_CONN.xreadgroup( agent_name, agent_name, streams, count5, block2000 ) for queue_key, messages in results: for msg_id, fields in messages: token json.loads(fields[token]) token_obj ReachToken(**token) if not check_ttl(token_obj): mark_receipt(token_obj, FAILED, error_codeTTL_EXPIRED, error_messagetoken exceeded ttl) REDIS_CONN.xack(queue_key, agent_name, msg_id) continue if not try_acquire(token_obj): # 说明同一条 msg_id 已经在被另一个实例处理 REDIS_CONN.xack(queue_key, agent_name, msg_id) continue mark_receipt(token_obj, ACCEPTED) try: result handler(token_obj) mark_receipt(token_obj, COMPLETED, error_messagejson.dumps(result, ensure_asciiFalse)) except Exception as e: mark_receipt(token_obj, FAILED, error_codetype(e).__name__, error_messagestr(e)) finally: REDIS_CONN.xack(queue_key, agent_name, msg_id)几个设计点值得说明。第一try_acquire用SET NX做幂等抢占。如果任务因为网络原因被重复投递只有一个消费者能获取锁其余的直接跳过并xack。这一步避免了业务 handler 被重复执行。第二TTL 判断用的是now_ms - created_at_ms没有拿两个不同机器上的时间戳做截点比较。这里埋了一个坑后面第四章会专门展开。第三xack放在finally里。不管 handler 成功还是失败这条 Redis Stream 消息都被确认。那失败之后怎么重试答案是Agent-Reach 的重试不是 Redis Stream 层面的重投而是投递端的重试。消费端把 FAILED 回执写给投递端投递端拿到 FAILED 后根据策略决定是否生成一个新的 ReachToken 再次投递。这样实现了“消费层确认”和“业务层重试”的解耦不会因为一个 FAILED 消息反复阻塞后续消息的消费。3.4 观察一次完整握手从 RECEIVED 到 COMPLETED我把这套最小闭环跑在了两个 Agent 上agent_a 作为发起方agent_b 作为消费方。操作步骤# 终端1启动 agent_b 消费端 python demo/agent_b.py # 终端2用 agent_a 发起一次检索请求 python demo/agent_a.pyagent_a 的任务逻辑是构造一个retrieval.request类型的 ReachToken把问题字符串放进 payload调用dispatch()投递出去然后轮询回执 key。回执流大概长这个样子消息 9f2e1a 状态变化 RECEIVED - 时间 1721... ACCEPTED - 时间 1721... COMPLETED - 时间 1721... 结果: {answer: ...}三条回执的时间戳间隔只有几十毫秒说明 agent_b 的 handler 执行非常快。如果某天你发现 ACCEPTED 和 COMPLETED 的间隔从几十毫秒变成几秒甚至几十秒那多半是 handler 里有慢调用比如 LLM 接口响应时间变长这时候回执时间戳就能帮你快速定位瓶颈。跑通这条链路后Agent-Reach 的最小闭环就算成立了。接下来会遇到的问题才是我最想分享的部分。4. 上线后碰到的一堆实际问题与调参记录4.1 重试风暴当我错误地吃掉 ack 之后第一次上线时我的重试策略写得很粗暴收到 FAILED 回执就立刻生成新 token 投递同一任务最多重试 3 次。结果第三方模型接口连续报限流一批任务全部 FAILED然后全部立刻重试。重试任务又把限流接口打得更满形成恶性循环。这个问题本质上是我把重试触发的条件和失败的后果绑定在了一起。显然重试的风暴会顺着触达层扩散到整个下游。接口在限流的时候你的重试只会延长它的恢复时间。后来的做法是引入退避和熔断每次 FAILED 重试之前先递增退避时间。第一次重试等 1 秒第二次 5 秒第三次 30 秒。同时加了一个最小间隔限制同一 intent 的失败率超过 50% 时暂停 2 分钟不投递新任务只处理存量任务。调整之后限流高峰期的任务成功率反而提高了一点。因为暂停新任务投递的这段时间第三方服务的队列压力得到了释放存量任务能更顺利地执行完。4.2 TTL 时钟偏移时间判断不能只信本机第一次写 TTL 校验的时候我犯了个错用机器本地时间做比较。看起来没毛病吧但线上系统如果不是全部部署在同一台物理机上多实例 Agent 各自所在机器的时钟可能相差几十秒甚至更多。如果发起方把任务发到另一个时钟走得慢的实例上这个任务可能一进来就被判成过期直接 FAILED。排查过程很有意思。回执流显示某几个 Agent 收到的任务几乎全部TTL_EXPIRED但消息明明才刚投递不到 2 秒。后来把各个节点的本机时间打出来一一对照才发现是时钟偏移。有些云实例如果之前做过快照迁移系统时间可能会复位到快照创建时刻偏移量能大到好几分钟。我最终的处理方案是Agent-Reach 里所有跨节点的过期判断都基于 Redis 的 TIME 命令或消息里携带的 created_at_ms 来做不用消费端本机时间当权威。每一次握手回执里也强制带上标准时间戳方便全局对齐。另外TTL 不要设置得太死。如果任务处理时间本身就要 5 秒而 TTL 给 6 秒那几乎每次处理都会在过期边缘徘徊。我建议 TTL 至少是预期处理耗时的 2 到 3 倍要给排队等待、微服务抖动这些情况留出缓冲。4.3 跨 Agent 回调死锁一版排障排了三个小时这个坑最让我头疼值得单独拿出来讲。场景是这样的Agent A 发起一个任务给 Agent BAgent B 处理到一半发现自己需要 Agent A 掌握的某些数据才能继续于是 B 又调用 A 的查询接口。结果 A 正在等 B 的完成回执B 在等 A 的查询响应两边互等全部卡死。我在回执流里看到的异常状态是A 发起的任务停在 ACCEPTED 没有后续B 那侧同样有一批任务停在 ACCEPTED。一开始我还以为只是慢等了十分钟还是没动静才意识到是死锁。最后用 trace_id 把所有回执拼接起来画出一条依赖环真相才浮现。其实就一段依赖链agent_a 等待 agent_b 回执 agent_b 等待 agent_a 查询接口解决思路是把这种同步依赖打散。跨 Agent 的任务约定里加了一条任何 Agent 在业务处理过程中不允许同步调用发起方的接口来获取继续处理所需的参数。如果确实需要拆成三阶段Agent A 先把 B 可能需要的上下文打包进 payload作为初始数据传过去。B 收到后发现数据不全把“缺失数据请求”作为一条异步任务发给 A而不是同步等待。A 处理完请求之后把补充数据作为新任务投递给 BB 继续工作。这条约定牺牲了一点“实时性”但换来了整个栈的可靠性。在设计 Agent 协作模式时这一点是非常容易被忽视的尤其是来自单体服务开发背景的人。单体里面方法调用是同步的天然不会互相等死但是消息化 / Agent 化之后所有同步等待都变成了跨节点等待不加约束就容易死锁。4.4 线上初始参数一组我实测下来的配置参考我在正式环境用了以下几个参数作为初始值你如果做类似的触达层可以根据自己的场景调整。# Agent-Reach 关键参数初始值 ttl_ms: 30000 # 任务令牌有效期按单跳预期耗时的2-3倍设置 maxlen: 100000 # 单意图 Stream 最大消息数 retry_max: 3 # 最大重试次数 retry_backoff_sec: [1, 5, 30] # 重试退避间隔递增 lock_ttl_sec: 60 # 幂等锁有效期 failed_ratio_threshold: 0.5 # 意图失败率熔断阈值 circuit_break_sec: 120 # 熔断暂停时间这套参数在我当时的场景下表现还不错。如果你们的任务是短平快类型TTL 可以收到 10 秒如果涉及 LLM 多轮处理或第三方接口TTL 放宽到 60 秒也可以。重点在于先定住一组可解释的参数再根据线上回执时间分布去调整不要一上来就追求完美。5. 演进方向与给初次尝试者的建议5.1 回执日志与事件溯源让无声故障变得可审计跑了两个月后我对 Agent-Reach 最大的感受是它最大的价值可能不是让消息更可靠而是让故障从“无声”变成“有声”。每一层回执本质上就是一条事件日志。把这堆回执持续落盘到普通文件或对象存储就形成了一个 message 级别的事件溯源系统。之后任何一个 Agent 出了问题我只要拿 trace_id 去查回执流水就能看到它在哪里停住、停了多少时间、最后一条回执是什么状态。这个能力在我做系统巡检和线上问题复盘时用得非常频繁。它不只是给开发人员看也可以让运维同学在出问题时快速给出“拓扑级”的判断减少反复拉人进群的次数。另外如果要加更高级的调度策略比如根据意图消息堆积量做动态扩容回执日志也能当作计量数据源比单独做埋点更简单。5.2 对初次使用 Agent-Reach 的人的三条建议第一条建议先定义清楚终态再写第一行代码。消息处理是成功、失败还是允许“过期放弃”这必须在一开始就定下来。如果设计成无限重试结果就是一个永远处理不完的任务永远占用你的队列资源。第二条建议幂等键是底线不能省。网络不稳、消费重启、消费组 rebalance每个环节都可能造成重复投递。你可以在大部分情况下靠概率赌它不会重复但只要线上发生一次重复扣款或重复外呼你就会后悔省掉了这一步。第三条建议从第一版就把 trace_id 传下去。链路追踪这种事早期不加后面数据量大了再想补几乎是推翻重做。无论你最后用了什么队列、什么语言这条建议都不会变。现在这个项目已经在我这边稳定运行了一段时间。每次看到回执流里整整齐齐的 RECEIVED、ACCEPTED、COMPLETED我都会想起那次日志里翻不到异常的深夜。如果你也在做多 Agent 系统并且已经开始被消息触达的问题困扰我建议早点把触达层当成一等公民去设计。消息不丢、有回执、能排查这九个字值得你认真对待一次。