Agent-Reach：用Redis Streams实现多智能体可靠消息触达

1. 项目缘起当智能体数量超过十个消息触达成了比模型推理更头疼的问题我最早做多智能体项目时以为难点全在智能体本身的逻辑上提示词怎么写、工具怎么选、模型怎么调。直到系统里跑起来十来个智能体每天要互相传递几千条任务和结果我才意识到真正让人失眠的是智能体到底能不能稳定地找到对方、把消息送过去、再拿到回执。Agent-Reach这个项目就是在这种背景下被逼出来的——它不关心你的智能体聪明不聪明它只解决一件事让任何智能体都能稳定地触达另一个智能体不管对方在哪个进程、哪台主机、是不是临时挂了。1.1 自建的SimpleQueue为什么先垮了在最早期我没有上任何消息中间件。所有智能体共用一个PostgreSQL实例里面建了一张task_queue表每个智能体启动之后起一个轮询线程每隔两秒扫一次表把分配给自己的任务抢走。这套方案在只有三五个智能体、任务量每秒几条的规模下跑得很顺。可一旦任务密度上来问题立刻暴露。最典型的现象有三个重复消费两个智能体同时扫到同一条statuspending的记录都执行了UPDATE ... WHERE id? AND statuspending在低并发下勉强还能靠行锁兜住一旦并发上去同一任务被两个智能体各执行一遍下游数据直接翻倍。状态丢了智能体A处理完任务先更新了任务状态为done再往结果表插入数据。结果插入失败了任务显示已完成但结果永远缺失。反过来也一样先插数据再更新状态要是更新失败任务就会反复执行产生脏数据。假死任务智能体B在执行一个耗时三十秒的长任务时进程崩溃了。数据库里那行running状态永远不会被改回去也没有任何机制去探测这个租约已经过期任务卡死在队列里无人认领。这些问题的本质都是我把消息触达的活交给了数据库这张表而数据库并不具备消息系统该有的投递确认、租约续期、超时重试这些语义。模型推理再准确也填补不了消息层丢包带来的随机性——整个系统的表现就是在看起来挺好和偶尔抽风之间来回横跳。1.2 Agent-Reach要解决的不是跑智能体而是智能体互相找得到当时摆在面前的选择有两个一个是直接上现成的编排框架让所有智能体都受一个中央调度器控制另一个是自研一套消息基础设施。前者的问题在于编排框架默认你有一个全知的调度器知道整个工作流的每一个步骤。但我的实际场景里智能体来自不同团队、由不同进程独立维护有的要按需扩容有的要灰度下线中央调度器反而变成了单点也成了团队协作时的审批瓶颈。所以我决定做一套专门面向智能体协作的触达层起名叫 Agent-Reach。它的核心假设是智能体是平等的、可寻址的、有生命周期的。每个智能体都有名字、属于某个逻辑分组、能声明自己处理哪些任务类型。智能体之间的通信不再走点对点HTTP而是通过一个中心辐射式总线由Agent-Reach负责寻址、投递、重试和回执。我给它划定的边界非常明确不写智能体的业务逻辑不做任务编排决策不替代你的模型调用层。它只负责三件事——寻址消息应该送给谁、投递怎么保证消息真正到达、回执发送方怎么知道结果。这套边界让Agent-Reach能适配各种不同的智能体实现也让我能够在一个代码库里专心打磨可靠性而不是跟着业务需求东改西改。2. Agent-Reach的整体设计辐射区和两级寻址怎么把找智能体变成了查表2.1 辐射区Reach Zone是把智能体组织成可寻址单元的关键Agent-Reach里有一个核心概念叫Reach Zone中文我习惯叫辐射区。它不是物理意义上的分区而是一个逻辑上的智能体分组。一个Zone可以对应一个业务域、一个团队、或者一套共享数据环境的智能体集合。为什么要引入Zone因为给消息指定接收者在分布式环境里是不现实的。你不可能在发送方写死http://collector-agent:8080/handle这样的地址智能体一扩容、一迁移地址就变了。Zone的作用是让发送方只关心消息该去哪个分组而不关心分组里现在有几个谁在执行。智能体上下线、扩容缩容对发送方完全透明。比如我把collector和validator两个智能体放进research这个Zone里那么任何向research.collect.raw_data投递的消息都会由collector这个智能体来消费。将来我把collector从一台机器迁到另一台或者同一份消息由两个并发实例来消费发送方代码都不用改。2.2 两级寻址Zone地址加Worker地址Agent-Reach的寻址是两级的格式统一为zone.worker.task_type。第一级是Zone第二级是Worker。Worker就是单个智能体实例。我把任务类型也放进了地址里这样投递消息时还能完成一次路由过滤一个Zone里可能跑着多个Worker消息根据task_type的前缀匹配只会被送到声明过自己处理该类型任务的智能体。这里有个细节Worker的地址不是网络地址而是逻辑标识。Agent-Reach内部维护一个reach registry每个Worker启动时注册自己的Zone、名字、支持的任务类型、当前负载和心跳时间戳。消息进来时路由组件只需要查这个注册表就能把消息放进目标Worker对应的消费组。整个过程不需要做服务发现也不依赖DNS解析。2.3 为什么选Redis Streams做传输层而不是直接上Kafka选传输层的时候我做了不少对比测试。很多朋友一听可靠消息投递就提议上Kafka但对我来说Kafka的引入代价太大了要维护broker集群要管topic的分区策略还要处理消费者组的rebalance。对于我的使用规模——几十个智能体、每天几十万条消息——Kafka属于杀鸡用牛刀。最终我选了Redis Streams做传输层核心原因是它能用单节点的成本提供我需要的三样东西Consumer Group天然支持多个Worker并发消费同一个Stream。Pending Entries List每条消息被消费后没有确认就会留在待处理列表里可以被其他消费者认领。流式重放历史消息可以按时间范围重新读取出问题后排查很方便。Redis Streams不是没有缺点它没有Kafka那样的分区并行度和海量堆积能力。但对于智能体协作这种场景消息量远没有日志系统那么大Redis单实例的吞吐完全够用而它带来的运维简单性和内建语义优势是实打实的。我整理过一张选型对比表放在项目文档里最关键的信息是这样的维度Redis StreamsKafkaRabbitMQPostgreSQL 表运维成本极低单节点高需管理集群中依赖业务系统消费组语义内建内建需Quorum队列自建实现消息重放支持按时间范围支持有限支持堆积能力中受内存限制极高中中但写放大失败重试完善需自己设计完善不完善3. 消息可靠投递链路至少一次、幂等消费和发件箱这样配合3.1 消息信封与幂等键变了没变都优雅处理Agent-Reach的每条消息都带一个信封信封里除了业务payload之外还有三个关键字段message_id、idempotency_key、trace_id。message_id是消息的唯一标识用来做去重。idempotency_key是业务幂等键通常由发送方根据业务语义来生成比如订单6100的补款通知在整个生命周期里应该只有一条。trace_id则贯穿一次完整调用链路帮助我在多个智能体之间串起日志。为什么幂等键这么重要因为Agent-Reach的投递语义是At-Least-Once至少一次也就是说一条消息在极端情况下可能被投递不止一次。只要发送方没有在租约时间内收到确认回执系统就会重试。重试意味着重复重复不可怕可怕的是重复没有被识别。处理重复的逻辑我放在了消费端。每个Worker在处理消息之前先把idempotency_key写入一个去重存储如果发现已经处理过就直接返回上一次的结果快照不再执行业务逻辑。这比在发送端拼命保证不重复要可靠得多——网络的不可靠性不是应用层能抹平的但幂等消费能从结果上抹平它。3.2 Leader租赁与发件箱模式避免重复消费的关键如果只是给每条消息标记一个已处理状态仍然不够。真正的难点在于一个智能体处理任务的过程中可能执行多个外部副作用比如调用模型、写数据库、调用下游接口。如果处理到一半进程挂了这条消息会被重新投递给另一个Worker实例后一个实例怎么知道前一个实例做到哪一步Agent-Reach在这里用了Leader租赁Lease机制。每条消息分配给某个Worker之后会带一个租约租约默认三十秒Worker通过周期性的心跳续期。如果租约过期其他Worker可以认领这条消息。这个机制保证了同一时刻最多只有一个Worker对该消息持有合法执行权。但光有租赁还不够还要解决状态一致的问题。我在项目里实现了一个发件箱模式Worker在处理任务时所有的业务状态更新和要对外发送的消息都写在同一个事务里。对Redis来说就是用Lua脚本保证原子性。也就是说如果Worker在任务执行到一半时崩溃了它的业务状态和消息发送记录要么一起提交要么一起回滚不会出现状态看起来成功但消息没发出去的中间态。3.3 投递时序优先级队列与死信处理除了保证消息不丢很多时候还要控制消息的处理顺序。Agent-Reach在Stream的每个入口上支持一个可选的priority字段取值范围是high、normal、low。我不打算在Stream里实现完整的优先级队列那会破坏Stream的天然顺序而是在路由层做了一个小优化普通消息直接进Stream高优先级消息同时写进一个专属的高优StreamWorker会优先消费高优Stream只有当它空了才切回普通队列。这在实际运行里很重要。比如一个用户触发了实时查询智能体同时系统在后台跑一批数据清洗任务。如果两者在同一个Stream里排队后台任务可能堵住实时查询。分开两个Stream之后实时查询总是能插队。处理不了的消息最终会进入死信队列。我对死信策略的设置是单个任务最多重试五次超出后进入reach:deadletters同时触发一个告警Webhook。死信不会自动丢弃我会定期人工查看因为大多数死信的原因不是代码bug而是下游依赖临时不可用恢复之后重新投递往往就能跑通。4. 操作实录跨主机部署四智能体集群的完整过程4.1 基础配置reach.yaml的关键字段说明Agent-Reach的部署配置文件是reach.yaml我把一个可以跑通的最小配置贴出来cluster: name: demo node_id: node-a transport: type: redis_streams endpoints: - 127.0.0.1:6379 heartbeat_interval: 5s zones: - name: research workers: - name: collector tasks: [collect.raw_data] max_concurrency: 4 lease_ttl: 30s retry_policy: max_attempts: 5 backoff: exponential配置里最容易被忽略的有两个字段。一个是lease_ttl它决定了一条消息在被其他Worker接管之前当前Worker最多可以独占多久。设得太短长任务会被误抢设得太长真挂了之后恢复太慢。我的经验是按任务P99耗时的两到三倍设置。如果任务是动态变化的就按最长的预估来。另一个是max_concurrency这个字段不是用来限制进程数的而是限制一个Worker同时处理多少条消息。一定要设置否则高吞吐下一个Worker里会瞬间堆几百个并发协程把模型的速率限制打爆也把下游服务打跨。4.2 智能体接入代码示例声明可处理的任务类型接入Agent-Reach并不需要重写智能体。它的Python客户端提供了装饰器风格的声明方式你可以把任意函数变成消息处理器。下面是我在collector智能体上的实际写法from agent_reach import Agent, task from agent_reach.context import TaskContext class CollectorAgent(Agent): def __init__(self): super().__init__(zoneresearch, namecollector) task(collect.raw_data) async def collect(self, payload: dict, ctx: TaskContext): result await crawl(payload[url]) ctx.emit(research.validated, {url: payload[url], content: result}) return {status: ok, record_id: result.id}我刻意让ctx.emit暴露在每个任务函数里因为实际业务中一个智能体处理完任务后往往还要触发下一个环节。通过ctx.emit发出去的消息会自动继承当前的trace_id和幂等键前缀这样整条链路都是可追踪的。很多第一次用的人会把后续任务写死在业务代码里直接调用另一个Agent的HTTP接口这样做不是不行只是链路追踪会断掉消息可靠性也得不到保障。4.3 从命令行发起一次辐射调用Agent-Reach带了一个CLI工具reachctl我平时在开发和排查问题的时候离不开它。比如要往researchZone里的collector发一条采集任务我会这样执行$ reachctl invoke research.collector.collect.raw_data \ --payload {url: https://example.com/docs} \ --wait 60s \ --idempotency-key doc-crawl-2025-0716-01--wait参数表示同步等待结果。Agent-Reach底层的实现是CLI发送消息之后订阅一个以message_id命名的临时响应Stream等Worker处理完写入响应或者超时返回。用它来测试智能体是否正常接入比看日志快得多。如果不加--wait就是纯异步投递发送方立刻得到消息已接受的确认。异步模式适合那些不需要立刻拿结果的环节比如批量后台处理任务。5. 压测数字与三个我反复踩的坑5.1 一套简单压测数据吞吐、延迟、重试率我在两台普通云主机上做了一个简单的压测。一台跑Redis另一台跑两个智能体Worker每个Worker开着4个并发槽位模拟一分钟内投递一万条消息的场景。结果如下指标数值总投递条数10000吞吐量平均 312 msg/sP50 延迟470msP99 延迟1.8s首次失败需要重试的比例约 2.7%最终失败进死信的比例0.06%这个数据不算惊艳但足够说明问题在智能体协作场景里单节点Redis Streams的吞吐完全不是瓶颈真正影响延迟的是Worker里任务本身的计算时长还有重试带来的后效。那2.7%的首次失败率大部分不是消息系统造成的而是下游模型接口偶尔超时。5.2 坑一把智能体状态存在内存里重启就丢任务第一次跑稳定测试的时候我发现一个诡异的现象Agent-Reach的重试机制明明生效了但有些任务在Worker重启之后再也没有被重新投递。查了半天问题出在我自己的代码上。我在CollectorAgent里用了一个内存Set来记录已经处理过的idempotency_key进程一重启这些记录全没了。消息本身在Stream里还在但因为消费者组已经确认了这条消息Agent-Reach不会再次投递它。结果就是任务状态丢在了那个重启窗口里。后来我把去重存储迁到了Redis用SETNX加过期时间来做幂等键记录问题才真正解决。Agent-Reach只保证消息不丢不保证你的业务状态能跟着进程一起复活。任何你想在智能体崩溃之后还能重建的数据都必须放进外部存储。5.3 坑二回调任务与父任务绑定在同一Zone导致级联超时在设计分工时我把一个会回调父任务结果的Worker放在了同一个Zone里。起初这看起来很合理Zone内部通信还能少走一跳。结果上线后发现一旦Zone里的某个Worker卡住堆积的任务会把整个Zone的租约资源占满连带着父任务的回调也跟着超时形成级联故障。问题出在Zone的资源隔离上。Agent-Reach的Zone是逻辑分组隔离的是寻址范围不是资源池。把请求路径上的父子任务放在同一个Zone等于让它们共享同一个消费资源池和租约窗口一个堵住全链路都堵。我的调整方案是把耗时敏感的回调Worker拆到独立的Zone和批量任务分开。批量任务再堵也不影响核心链路的触达。5.4 坑三没有限制fan-out规模一次广播把整个集群拖垮还有一个坑印象特别深。某次我做了一个全网广播的功能让一个智能体内所有已注册的Worker群发消息。开始只有几个Worker一切正常。后来Worker数量慢慢涨到了几十个又一次广播同时产生了上千条消息Redis那个节点瞬间CPU飙升几秒钟内所有Stream的消费都变慢了。Agent-Reach后来加了一个叫做fanout_limit的配置项默认每次广播最多触达五十个Worker超过的部分会返回错误。这个限制看起来粗暴但确实有效。分布式系统里的广播问题从来不是消息量本身而是你给了它一个没有上限的出口。6. 什么时候该用Agent-Reach什么时候不如自己写6.1 适用场景清单与不适合场景踩了这么多坑之后我总结出Agent-Reach真正适合的场景你的智能体数量在增加、互相之间有异步任务往来、希望每个智能体独立部署独立演进而且运行成本比较敏感。它尤其适合那些被任务丢失困扰了很久、但团队规模又不足以维护一套Kafka集群的项目。反过来如果你的智能体之间几乎都是同步调用且保证点对点直连即可那直接用HTTP加超时重试会更简单。如果你的核心诉求是复杂工作流编排需要条件分支、人工审批节点、Saga补偿这种高级语义Agent-Reach也不会帮你解决——该上编排引擎的时候还是得用编排引擎。我经常跟朋友说Agent-Reach在项目里的定位很像电话交换机。它不替你决定谁该给谁打电话也不帮你翻译通话内容但它保证你拨出去的电话能接通、占线时能排队、对方挂了能响铃提醒。很多智能体项目缺的其实不是更聪明的模型而是这样一层让人放心的通信设施。6.2 我的最终使用心得从自建PostgreSQL队列到 Agent-Reach我最大的感受是分布式多智能体系统是一个典型的木桶效应场景。模型推理的准确性、任务编排的合理性、消息触达的可靠性任何一块短板都会拉低整个系统的下限。而消息触达恰恰是最容易在做业务逻辑的时候被忽略掉的那一块。如果你现在也在为智能体之间丢任务、重复消费、状态不一致这些问题头疼我的建议是不要急着加数据库索引也不要尝试在业务代码里堆脏逻辑去补锅。先把寻址、投递、重试、幂等这四个基础能力想清楚选一个能让这些语义落到实处的消息层。Agent-Reach对我来说就是这样一个工具它帮我省掉了自己维护消息系统的大量精力让我能把注意力重新放回智能体本身该做的事上。