先说结论Agent-Reach 不是我突发奇想造出来的玩具。它是我在多个 AI Agent 协作调度这件事上折腾了大半年后沉淀出来的一个轻量通信层核心就两个词触达和路由。所谓触达就是当你有一个任务需要某个 Agent 去处理时你不用管它具体在哪台机器、叫什么、当前是不是忙你只负责把消息交出去剩下的事情由 Agent-Reach 保证它一定会被合适的 Agent 接住。如果你正在做客服机器人、智能工单、多步数据报表这些需要多个 Agent 配合的项目我强烈建议你花十分钟看看这篇文章它能帮你少踩很多调度上的坑。1. Agent-Reach 是做什么的我为什么需要写它1.1 一次“找不到人”的调度事故我印象最深的一次事故发生在做智能客服的那段时间。当时系统里同时跑着三个 Agent意图识别 Agent、订单查询 Agent 和退款审批 Agent。正常情况下意图识别 Agent 判断完“用户想退款”之后会去调用退款审批 Agent 完成后续流程。看起来没问题但上线后我发现只要退款审批 Agent 因为某个依赖服务抖动而超过 5 秒没有应答意图识别 Agent 就开始原地等待消息队列里积压了一大批未处理的消息。更要命的是我后来加了一个重试逻辑结果一个退款请求被重复推给审批 Agent 三次虽然审批这一步做了幂等没出大错但下游的通知 Agent 却给用户发了三条一模一样的通知。事后复盘时我意识到问题根源并不在单个 Agent 的模型能力而在于我们一直把精力放在“单个 Agent 是否聪明”上完全忽略了它们之间的通信协议。当系统里只有两个 Agent 时你用函数调用就能解决但 Agent 数量一旦多起来每个人都要记住别人叫什么、在哪个进程、什么接口调用关系就变成一张乱麻。这种强耦合的代码表面上跑得通实际上任何一个节点抖动都会传导到整条链路最终表现就是“消息发出去却不知道谁在处理也不知道处理到哪一步了”。Agent-Reach 就是在这次事故之后开始写的。我最初想的很简单能不能让 A 不用关心 B 在哪只告诉系统“我要做什么”系统自己找到合适的 B并且负责把这件事做成。后来这个思路慢慢变成了一个独立的通信框架也就是你现在看到的 Agent-Reach。1.2 设计目标与适用范围Agent-Reach 的设计目标可以归纳成三句话按语义路由发消息的人只需要声明“这件事是什么类型”比如order.query、refund.apply而不需要写死“发给 order_agent”。保证触达消息发出去之后系统负责超时重试、熔断、死信处理让发件方不用接收“B 没开机”这种错误。全程可控消息从入队到被消费、再返回结果每一步都有唯一的轨迹 ID随时能查。适用的场景非常明确。如果你在做多 Agent 协作比如客服机器人需要先识别用户情绪再查订单再决定要不要给优惠券或者智能运维告警系统里告警 Agent 要分发给不同的处理 Agent又或者数据分析流水线里要按步骤调度不同的分析 Agent那 Agent-Reach 是合适的。它解决的是“Agent 之间如何可靠地互相找到对方并传递任务”的问题而不是“模型怎么生成智能回答”的问题。但 Agent-Reach 不适合做超大数据量的流式传输。如果你每秒要推上百万条日志或者做的是实时交易流水汇总那是 Kafka 和 Flink 的活不要把 Agent-Reach 当成流处理引擎用。它定位在任务级触达不是数据管道。2. Agent-Reach 的通信模型从点对点调用升级为触达路由2.1 点对点调用为什么容易崩先想象一个传统写法。你在代码里直接拿到order_agent agents[order_agent]然后调用order_agent.process(request)。这就像你天天给同一个快递员打电话让他帮你取快递。刚开始没什么问题但有一天快递员休假了你傻眼了你不知道还有谁能取又过了一段时间快递公司重排了工号你存的电话全变成了空号。多 Agent 系统也是一样。一旦每个 Agent 都持有其他 Agent 的引用你们之间的通信协议就变成一张蜘蛛网。新增一个 Agent 要改一堆调用方代码下线一个 Agent 还要担心有没有别的地方在调用它。这种“点对点”模式的最大问题不是性能而是脆弱它把“位置”和“能力”混为一谈。Agent-Reach 的做法是引入“触达路由”。发件方把消息投递给一个路由中心路由中心根据消息的 topic、能力标签、当前在线 Agent 列表来决定最终送给谁。发件方手里没有收件方的地址只有一张“需求描述”。这就像寄快递你在快递单上写收件地址由快递公司的分拣中心决定哪个网点派送。只要没人收快递会退回地址模糊分拣中心会尝试匹配。Agent-Reach 干的就是分拣中心的活。2.2 消息信封、能力注册与路由键为了实现这个模型Agent-Reach 提出了三个核心概念消息信封、能力注册、路由键。消息信封是通信的载体。每个进入系统的事件都会被封装成标准结构我直接贴一个最简单的信封示例{ message_id: msg_8f3a9c01, trace_id: tr_20250919A001, topic: order.query, content: { order_no: A10086 }, ttl: 60, required_capability: order.query }注意message_id用于去重trace_id用于全链路追踪topic是广播式的主题描述content是真正的业务负载ttl是消息存活时间。这些字段看起来简单但每一个都在后续的可靠性设计里扮演关键角色。能力注册解决“谁能处理这件事”的问题。每个 Agent 节点启动后会向注册中心上报自己的能力标签比如我负责order.query另一个 Agent 负责order.update还有一个负责logistics.track。注册中心只保存能力与节点的映射不保存节点之间的调用关系。这就是一个电话号码簿但它登记的是“我能做什么”而不是“我是谁”。路由键则是用来匹配的规则。它通常是一个带通配符的字符串比如order.*能匹配所有和订单相关的事件*.query能匹配所有查询类事件。路由层把信封里的topic和required_capability拿去做匹配最终得出一个候选 Agent 列表再按策略选一个。2.3 为什么没有直接用 RabbitMQ 或 Celery很多朋友会问你说的这套东西RabbitMQ 和 Celery 不是早就做过了吗确实消息队列可以保证消息不丢Celery 也可以做任务分发。但我对比之后发现它们解决的是“消息如何可靠传输”的问题而不是“Agent 之间如何语义匹配”的问题。方案擅长的事缺什么RabbitMQ / 主流 MQ高吞吐、持久化、削峰填谷交换机与队列的绑定关系偏运维化需要预先写死收件队列不懂“能力”和“情绪表达”Celery异步后台任务、定时任务Worker 名称和任务名称写死在代码里调度偏命令式不适合动态能力发现Agent-Reach语义路由、触达控制、可观测不负责海量消息存储底层还是要依赖 Redis Streams所以我的选择是Agent-Reach 做“上层协议”底层队列用 Redis Streams 做可靠存储。这样你不用为了用 Agent-Reach 就把你现有的消息队列全拆掉。如果公司已经有 Kafka 或 RabbitMQAgent-Reach 可以作为一个路由服务接在它们前面负责把“语义请求”翻译成“该投递给谁”。3. 五分钟跑通一个 Demo从零搭建 Agent-Reach 调度环境3.1 安装与基础配置先说快速上手的体验。Agent-Reach 是一个 Python 框架你只需要pip install agent-reach就能拿到核心包。为了不让 Demo 依赖太多外部基础设施开发模式可以用内存版 Registry 和内存版队列但如果想模拟生产环境我建议连一个本地 Redis消息的可靠性和持久化会真实很多。pip install agent-reach redis-server --port 6379然后初始化接入点配置你的 Redis 地址以及应用名称from agent_reach import ReachContext ctx ReachContext( app_namedemo_workshop, registry_urlredis://localhost:6379/0, queue_urlredis://localhost:6379/1, modedevelopment ) ctx.init()registry_url负责存路由注册信息queue_url负责存消息流分开用两个 Redis 数据库号可以避免互相挤占。开发模式给你开了一些便利比如自动重试关闭、日志更啰嗦等上生产再调严格策略。3.2 写一个最简单的 Agent 节点下面是一个最基础的订单查询 Agent。它做的事情很简单启动时告诉注册中心“我能处理order.query”然后不断接收路由层派发的任务处理完再返回结果。from agent_reach import AgentRuntime, capability agent AgentRuntime( nameorder_agent, contextctx, concurrent4 ) agent.capability(order.query, timeout10) async def query_order(params): order_no params[order_no] # 这里可以是数据库查询也可以是外部API调用 return {order_no: order_no, status: delivered, last_event: 2025-09-19 10:30:00} agent.start()关键点有两个。第一agent.capability(order.query, timeout10)这个装饰器把“能力”和“处理函数”绑定timeout告诉调度方这个函数最多跑 10 秒超过 10 秒就按超时处理。第二concurrent4表示这个 Agent 最多同时处理 4 个任务超过的请求会在队列里排队而不是无限开线程把进程拖垮。我见过很多新手在这两个参数上不设限制结果流量一上来直接 OOM后面要记住。3.3 配置路由规则Agent 写完之后要告诉 Agent-Reach 怎么把消息路由过去。路由规则其实就是一个匹配表from agent_reach import ReachRouter router ReachRouter(contextctx) router.declare_route( topic_patternorder.query, required_capabilityorder.query, strategyround_robin, fallbackorder_fallback_agent )这个规则的意思是当收到一个topic为order.query的消息并且它要求order.query能力时就从所有注册了该能力的 Agent 里以轮询方式选一个。如果都不可用就转给兜底 Agent。这样一旦主用处理节点崩溃请求不会直接失败而是进入 fallback。实际生产环境里路由规则最好通过配置中心下发不要写死在代码里。因为当你有很多 Agent 时规则改动很频繁你不能每次改规则都发一次版本。3.4 从客户端发起一次触达服务端配好了客户端只需要两行from agent_reach import ReachClient client ReachClient(contextctx) reply await client.reach( topicorder.query, payload{order_no: A10086}, timeout8, strategyround_robin ) print(reply.result)reach本质上不是“调用接口”而是“把消息交给路由中心并等待符合要求的 Agent 应答”。如果 8 秒内没有 Agent 返回成功客户端会收到一个超时异常。你可以捕获这个异常做自己的业务兜底比如记录日志、通知人工处理。值得高兴的是路由中心内部已经帮你处理了大部分重试所以客户端很少要写复杂的重试代码。4. 核心机制拆解路由、重试、熔断与广播4.1 路由匹配是怎么工作的路由匹配是 Agent-Reach 的心脏。很多人以为它就是简单比较字符串其实里面有一个明确的优先级顺序我列一下匹配方式优先级说明精确 topic高order.query只匹配order.query最直接性能最高通配符 topic中order.*匹配所有order开头的话题适合业务分组能力标签匹配低消息没有指定 topic只要求“谁能处理”按能力动态发现fallback 规则兜底前面都没命中走配置的备用 Agent再不行进死信这里有一个细节如果一条消息同时命中精确规则和通配符规则精确规则优先。因为通配符是给“粗粒度场景”用的一旦业务上明确了具体类型就不该再拿模糊规则去猜。我踩过的一个坑是当时既配了order.*又配了order.query结果订单查询消息总是被通配符规则抢走因为两条规则都匹配但通配符规则注册得早路由表是按注册顺序遍历的。后来我加了优先级字段把精确规则排到最前面才解决。路由命中之后就要决定“发给哪个 Agent”。round_robin是默认策略还有random、min_load和prefer_local。min_load会查询每个 Agent 当前的并发任务数挑最小的投递适合下游处理能力不均衡的场景。prefer_local则优先把任务留在本机减少跨网络传输适合单体多进程部署。4.2 重试、超时和熔断怎么保证“一定到达”“触达”两个字听着简单落地很难。最难的地方在于你不能因为害怕重复就完全不重试也不能因为追求必达就无限重试把下游打爆。Agent-Reach 给每个触达任务提供了几个核心参数参数默认值建议值作用max_retries23最大重试次数不算首次retry_backoff_base0.5s0.5s第一次重试的等待基础时间retry_backoff_factor22每次重试翻倍比如 0.5s、1s、2sretry_jitter0.10.1加入随机抖动防止多个任务同时重试造成惊群ttl60s按业务定消息最长存活时间防止“死任务”一直占坑dedup_window60s60s同一message_id在窗口内只消费一次避免重复处理我强烈建议每次触达都带一个唯一的message_id。因为 Agent-Reach 的投递语义是“至少一次”不是“恰好一次”。也就是说网络抖动、Agent 响应超时、进程重启都可能导致一条消息被投递两次。如果消费端不做幂等重试就会变成事故。比如你做了一个“发送验证码”的 Agent触发重试时用户可能收到两条验证码体验很差。所以在消费逻辑里第一步应该是检查message_id是不是已经处理过处理过就直接返回上次结果。重试也不是无限重试。当连续失败达到阈值比如 3 次时Agent-Reach 会触发熔断直接把后续发往该能力或该 Agent 的请求切换给 fallback或者在规定时间内不再选择这个故障节点。熔断的意义是保护下游而不是拯救单条消息。你让一个已经超负荷的 Agent 再多收 10 条重试任务只会让它雪上加霜不如让它缓缓。4.3 广播与定向两种触达模式前面讲的主要是“定向触达”也就是一条任务只找一个 Agent 处理。但实际场景里还有一类需求是企业微信群发式的好消息比如“所有 Agent 给我同步自己的状态”“订单状态变化要通知所有关注方”。这时单点投递就不够用了需要广播。Agent-Reach 用broadcast方法处理results await client.broadcast( topicsystem.heartbeat, payload{ts: 2025-09-19 10:30:00}, strategyall )strategyall会命中所有注册了对应 topic 的 Agent并把每个 Agent 的返回结果汇总成一个列表。如果没有 Agent 注册广播消息不会报错只会返回空列表。这里有一个设计哲学广播消息本身就是“尽力通知”你并不知道也不用关心谁在听但如果需要确认每条都触达成功就要用broadcast加上结果校验。你可以把“定向”和“广播”结合使用我经常这么搭先用广播让业务 Agent 刷新本地缓存再用定向让主处理 Agent 执行任务最后用广播通知所有下游结果。Agent-Reach 手写起来也不复杂几行代码就能把链路组装起来。5. 实操中的坑和排查经验从“能用”到“好用”5.1 我踩过的三个坑写这个框架的过程中我把自己埋进了一堆莫名其妙的坑里。分享三个最典型的希望大家不要重复踩。第一个坑是“消息显示已投递但 Agent 没处理”。这个问题的根源是 ACK 时机不对。我一开始在路由中心把消息放入队列后就立刻标记为已投递但那时 Agent 还什么都没做。后来我把投递状态拆成了“已入队”“已派发”“已确认”“已完成”四个阶段只有在 Agent 执行完业务函数并返回结果时才把消息标记为真正完成。排查时你用reach_ctl task info --trace_id tr_xxx一看就知道卡在哪一步。第二个坑是“重试风暴”。有一次某个下游数据库连接池被打满导致大量 Agent 响应超时。由于每条消息都开启了重试而且重试间隔没加抖动几乎在同一瞬间几十条重试任务同时涌向同一个故障 Agent直接把它的连接池打到了 100%。从那以后我给所有路由规则默认开启了熔断并加上了 0.1 秒抖动。重试的目的是“给系统一个恢复的机会”不是“多试几次证明它真的不行”。第三个坑是“Agent 重启后任务神秘消失”。原因是消费者在处理消息时奔溃Redis Stream 里虽然还能看到消息但已经超时进不了消费者组。后来我专门为这种情况配置了死信队列dead_letter.topic所有超过重试次数或者 TTL 的消息统一进死信再定时跑一个重放脚本扫死信。这样至少不会让问题静默消失日志里能看到“这 10 条消息处理失败了”。5.2 生产环境参数调试记录我把我常用的一套生产参数贴出来起点参考这个再根据业务压测调整参数我的配置说明max_retries3再多了容易让下游雪崩少了对抖动场景覆盖不足ttl30s每个任务最长存活 30 秒适合绝大多数内部接口dedup_window120s必须大于单次最长重试窗口防止第 4 秒重试时重复concurrent8单个 Agent 并发处理数先按 CPU 核心数乘 2 起步queue_maxlen100000单队列最大长度超过直接拒绝新消息并告警heartbeat_interval5sAgent 心跳上报时间决定故障发现速度这里特别提醒一个细节ttl一定要比你的最大业务耗时 最大重试耗时都大。如果你的单次业务处理要 20 秒重试 3 次各间隔 0.5s、1s、2s那ttl至少得给 30 秒左右。否则消息还没处理完TTL 就到期了路由中心会把任务重新派发导致同一个 Agent 在处理同一个消息时产生并发冲突。5.3 可观测性设计写 Agent 协作框架最容易忽略的就是可观测性。Agent 一多任务流转就很复杂如果没有全链路追踪出了问题根本不知道从哪查起。Agent-Reach 里的每个任务都要求带trace_id并且这个 ID 会透传到所有日志、Redis 消息体和外部调用的 headers 里。我建议在业务代码里统一加一个日志过滤器把trace_id、topic、agent_name、status、duration_ms都打出来。这样你就可以在日志系统里直接按trace_id串起一整条链路。我还给每个 Agent 暴露了小端点GET /healthz # Agent 是否活 GET /metrics # 待处理任务数、成功率、平均耗时、重试次数/metrics输出的是纯文本指标可以直接接到 Prometheus。我最常盯的指标是route_unmatched_total也就是“找不到 Agent 处理的消息数”。这个指标一上涨说明你的业务语义和 Agent 能力注册之间出现了偏差要么是有新业务没注册要么是注册的 Agent 全挂了。处理优先级比单纯看 CPU 使用率重要得多。6. 后面怎么玩把 Agent-Reach 用得更顺手6.1 和常见的 Agent 编排框架结合Agent-Reach 并不排斥现有的编排框架。比如你可以把 Agent-Reach 的触达能力放在“工具调用”层让 LLM Agent 在需要查订单时不是直接调用 Python 函数而是调用client.reach(topicorder.query, ...)。这样 LLM 不需要知道订单在那台机器上只需要描述意图剩下的由 Agent-Reach 处理。我试过这种方式之后最直接的变化是新增 Agent 时不需要再改 LLM 的 few-shot 样例里的工具列表因为工具参数只剩一行“topic payload”Agent 注册好能力系统自动就把路由打通了。这里要注意的是LLM 生成的 topic 和 payload 不像手工代码那么可靠。我建议在 LLM 调用 Agent-Reach 之前加一个校验层把 topic 白名单和 payload 的必填字段提前确认好防止模型幻觉生成一个不存在的路由键把消息打进死信。6.2 加一层权限与规则版本管理如果你和我一样是多团队协作路由规则会变成公共资产。谁都能改route就意味着谁都能把消息导到自己的 Agent 里这很危险。我后来给 Agent-Reach 加了一个简单的规则版本管理每条路由规则都有version、owner、status三个字段默认状态是draft只有审批通过后才变成active。规则变更记录会写进审计日志万一出问题可以快速回滚到上一个版本。这一层改动很小但价值极大。它本质上把“技术路由决策”变成了“团队协作决策”避免了一个人改了规则、所有人跟着不稳定。6.3 最后想多说一句Agent-Reach 最让我感慨的并不是代码量而是“触达”这两个字被认真对待之后整个系统给人的安全感完全不一样了。以前我每天担心某个 Agent 会不会挂了现在我把这份担心交给了协议层。每次有人跟我说他写了一套多 Agent 系统我都会先问一句话如果某个 Agent 不回应了你的消息会怎样如果你答不上来那可能你需要的不是一个更智能的模型而是一个更可靠的触达层。我个人在实际操作中体会最深的是框架选型和参数调优都是表面功夫真正的分水岭在于你是否真的想清楚了消息的生命周期。想清楚了Agent-Reach 只是把这件事工程化罢了。希望这篇文章能帮你减少一些“找不到人”的焦虑。 
长河君 · 2026/10/28 11:56:20 · 阅读 1.2k Agent-Reach：为多Agent协作打造的轻量级触达与路由通信层
先说结论Agent-Reach 不是我突发奇想造出来的玩具。它是我在多个 AI Agent 协作调度这件事上折腾了大半年后沉淀出来的一个轻量通信层核心就两个词触达和路由。所谓触达就是当你有一个任务需要某个 Agent 去处理时你不用管它具体在哪台机器、叫什么、当前是不是忙你只负责把消息交出去剩下的事情由 Agent-Reach 保证它一定会被合适的 Agent 接住。如果你正在做客服机器人、智能工单、多步数据报表这些需要多个 Agent 配合的项目我强烈建议你花十分钟看看这篇文章它能帮你少踩很多调度上的坑。1. Agent-Reach 是做什么的我为什么需要写它1.1 一次“找不到人”的调度事故我印象最深的一次事故发生在做智能客服的那段时间。当时系统里同时跑着三个 Agent意图识别 Agent、订单查询 Agent 和退款审批 Agent。正常情况下意图识别 Agent 判断完“用户想退款”之后会去调用退款审批 Agent 完成后续流程。看起来没问题但上线后我发现只要退款审批 Agent 因为某个依赖服务抖动而超过 5 秒没有应答意图识别 Agent 就开始原地等待消息队列里积压了一大批未处理的消息。更要命的是我后来加了一个重试逻辑结果一个退款请求被重复推给审批 Agent 三次虽然审批这一步做了幂等没出大错但下游的通知 Agent 却给用户发了三条一模一样的通知。事后复盘时我意识到问题根源并不在单个 Agent 的模型能力而在于我们一直把精力放在“单个 Agent 是否聪明”上完全忽略了它们之间的通信协议。当系统里只有两个 Agent 时你用函数调用就能解决但 Agent 数量一旦多起来每个人都要记住别人叫什么、在哪个进程、什么接口调用关系就变成一张乱麻。这种强耦合的代码表面上跑得通实际上任何一个节点抖动都会传导到整条链路最终表现就是“消息发出去却不知道谁在处理也不知道处理到哪一步了”。Agent-Reach 就是在这次事故之后开始写的。我最初想的很简单能不能让 A 不用关心 B 在哪只告诉系统“我要做什么”系统自己找到合适的 B并且负责把这件事做成。后来这个思路慢慢变成了一个独立的通信框架也就是你现在看到的 Agent-Reach。1.2 设计目标与适用范围Agent-Reach 的设计目标可以归纳成三句话按语义路由发消息的人只需要声明“这件事是什么类型”比如order.query、refund.apply而不需要写死“发给 order_agent”。保证触达消息发出去之后系统负责超时重试、熔断、死信处理让发件方不用接收“B 没开机”这种错误。全程可控消息从入队到被消费、再返回结果每一步都有唯一的轨迹 ID随时能查。适用的场景非常明确。如果你在做多 Agent 协作比如客服机器人需要先识别用户情绪再查订单再决定要不要给优惠券或者智能运维告警系统里告警 Agent 要分发给不同的处理 Agent又或者数据分析流水线里要按步骤调度不同的分析 Agent那 Agent-Reach 是合适的。它解决的是“Agent 之间如何可靠地互相找到对方并传递任务”的问题而不是“模型怎么生成智能回答”的问题。但 Agent-Reach 不适合做超大数据量的流式传输。如果你每秒要推上百万条日志或者做的是实时交易流水汇总那是 Kafka 和 Flink 的活不要把 Agent-Reach 当成流处理引擎用。它定位在任务级触达不是数据管道。2. Agent-Reach 的通信模型从点对点调用升级为触达路由2.1 点对点调用为什么容易崩先想象一个传统写法。你在代码里直接拿到order_agent agents[order_agent]然后调用order_agent.process(request)。这就像你天天给同一个快递员打电话让他帮你取快递。刚开始没什么问题但有一天快递员休假了你傻眼了你不知道还有谁能取又过了一段时间快递公司重排了工号你存的电话全变成了空号。多 Agent 系统也是一样。一旦每个 Agent 都持有其他 Agent 的引用你们之间的通信协议就变成一张蜘蛛网。新增一个 Agent 要改一堆调用方代码下线一个 Agent 还要担心有没有别的地方在调用它。这种“点对点”模式的最大问题不是性能而是脆弱它把“位置”和“能力”混为一谈。Agent-Reach 的做法是引入“触达路由”。发件方把消息投递给一个路由中心路由中心根据消息的 topic、能力标签、当前在线 Agent 列表来决定最终送给谁。发件方手里没有收件方的地址只有一张“需求描述”。这就像寄快递你在快递单上写收件地址由快递公司的分拣中心决定哪个网点派送。只要没人收快递会退回地址模糊分拣中心会尝试匹配。Agent-Reach 干的就是分拣中心的活。2.2 消息信封、能力注册与路由键为了实现这个模型Agent-Reach 提出了三个核心概念消息信封、能力注册、路由键。消息信封是通信的载体。每个进入系统的事件都会被封装成标准结构我直接贴一个最简单的信封示例{ message_id: msg_8f3a9c01, trace_id: tr_20250919A001, topic: order.query, content: { order_no: A10086 }, ttl: 60, required_capability: order.query }注意message_id用于去重trace_id用于全链路追踪topic是广播式的主题描述content是真正的业务负载ttl是消息存活时间。这些字段看起来简单但每一个都在后续的可靠性设计里扮演关键角色。能力注册解决“谁能处理这件事”的问题。每个 Agent 节点启动后会向注册中心上报自己的能力标签比如我负责order.query另一个 Agent 负责order.update还有一个负责logistics.track。注册中心只保存能力与节点的映射不保存节点之间的调用关系。这就是一个电话号码簿但它登记的是“我能做什么”而不是“我是谁”。路由键则是用来匹配的规则。它通常是一个带通配符的字符串比如order.*能匹配所有和订单相关的事件*.query能匹配所有查询类事件。路由层把信封里的topic和required_capability拿去做匹配最终得出一个候选 Agent 列表再按策略选一个。2.3 为什么没有直接用 RabbitMQ 或 Celery很多朋友会问你说的这套东西RabbitMQ 和 Celery 不是早就做过了吗确实消息队列可以保证消息不丢Celery 也可以做任务分发。但我对比之后发现它们解决的是“消息如何可靠传输”的问题而不是“Agent 之间如何语义匹配”的问题。方案擅长的事缺什么RabbitMQ / 主流 MQ高吞吐、持久化、削峰填谷交换机与队列的绑定关系偏运维化需要预先写死收件队列不懂“能力”和“情绪表达”Celery异步后台任务、定时任务Worker 名称和任务名称写死在代码里调度偏命令式不适合动态能力发现Agent-Reach语义路由、触达控制、可观测不负责海量消息存储底层还是要依赖 Redis Streams所以我的选择是Agent-Reach 做“上层协议”底层队列用 Redis Streams 做可靠存储。这样你不用为了用 Agent-Reach 就把你现有的消息队列全拆掉。如果公司已经有 Kafka 或 RabbitMQAgent-Reach 可以作为一个路由服务接在它们前面负责把“语义请求”翻译成“该投递给谁”。3. 五分钟跑通一个 Demo从零搭建 Agent-Reach 调度环境3.1 安装与基础配置先说快速上手的体验。Agent-Reach 是一个 Python 框架你只需要pip install agent-reach就能拿到核心包。为了不让 Demo 依赖太多外部基础设施开发模式可以用内存版 Registry 和内存版队列但如果想模拟生产环境我建议连一个本地 Redis消息的可靠性和持久化会真实很多。pip install agent-reach redis-server --port 6379然后初始化接入点配置你的 Redis 地址以及应用名称from agent_reach import ReachContext ctx ReachContext( app_namedemo_workshop, registry_urlredis://localhost:6379/0, queue_urlredis://localhost:6379/1, modedevelopment ) ctx.init()registry_url负责存路由注册信息queue_url负责存消息流分开用两个 Redis 数据库号可以避免互相挤占。开发模式给你开了一些便利比如自动重试关闭、日志更啰嗦等上生产再调严格策略。3.2 写一个最简单的 Agent 节点下面是一个最基础的订单查询 Agent。它做的事情很简单启动时告诉注册中心“我能处理order.query”然后不断接收路由层派发的任务处理完再返回结果。from agent_reach import AgentRuntime, capability agent AgentRuntime( nameorder_agent, contextctx, concurrent4 ) agent.capability(order.query, timeout10) async def query_order(params): order_no params[order_no] # 这里可以是数据库查询也可以是外部API调用 return {order_no: order_no, status: delivered, last_event: 2025-09-19 10:30:00} agent.start()关键点有两个。第一agent.capability(order.query, timeout10)这个装饰器把“能力”和“处理函数”绑定timeout告诉调度方这个函数最多跑 10 秒超过 10 秒就按超时处理。第二concurrent4表示这个 Agent 最多同时处理 4 个任务超过的请求会在队列里排队而不是无限开线程把进程拖垮。我见过很多新手在这两个参数上不设限制结果流量一上来直接 OOM后面要记住。3.3 配置路由规则Agent 写完之后要告诉 Agent-Reach 怎么把消息路由过去。路由规则其实就是一个匹配表from agent_reach import ReachRouter router ReachRouter(contextctx) router.declare_route( topic_patternorder.query, required_capabilityorder.query, strategyround_robin, fallbackorder_fallback_agent )这个规则的意思是当收到一个topic为order.query的消息并且它要求order.query能力时就从所有注册了该能力的 Agent 里以轮询方式选一个。如果都不可用就转给兜底 Agent。这样一旦主用处理节点崩溃请求不会直接失败而是进入 fallback。实际生产环境里路由规则最好通过配置中心下发不要写死在代码里。因为当你有很多 Agent 时规则改动很频繁你不能每次改规则都发一次版本。3.4 从客户端发起一次触达服务端配好了客户端只需要两行from agent_reach import ReachClient client ReachClient(contextctx) reply await client.reach( topicorder.query, payload{order_no: A10086}, timeout8, strategyround_robin ) print(reply.result)reach本质上不是“调用接口”而是“把消息交给路由中心并等待符合要求的 Agent 应答”。如果 8 秒内没有 Agent 返回成功客户端会收到一个超时异常。你可以捕获这个异常做自己的业务兜底比如记录日志、通知人工处理。值得高兴的是路由中心内部已经帮你处理了大部分重试所以客户端很少要写复杂的重试代码。4. 核心机制拆解路由、重试、熔断与广播4.1 路由匹配是怎么工作的路由匹配是 Agent-Reach 的心脏。很多人以为它就是简单比较字符串其实里面有一个明确的优先级顺序我列一下匹配方式优先级说明精确 topic高order.query只匹配order.query最直接性能最高通配符 topic中order.*匹配所有order开头的话题适合业务分组能力标签匹配低消息没有指定 topic只要求“谁能处理”按能力动态发现fallback 规则兜底前面都没命中走配置的备用 Agent再不行进死信这里有一个细节如果一条消息同时命中精确规则和通配符规则精确规则优先。因为通配符是给“粗粒度场景”用的一旦业务上明确了具体类型就不该再拿模糊规则去猜。我踩过的一个坑是当时既配了order.*又配了order.query结果订单查询消息总是被通配符规则抢走因为两条规则都匹配但通配符规则注册得早路由表是按注册顺序遍历的。后来我加了优先级字段把精确规则排到最前面才解决。路由命中之后就要决定“发给哪个 Agent”。round_robin是默认策略还有random、min_load和prefer_local。min_load会查询每个 Agent 当前的并发任务数挑最小的投递适合下游处理能力不均衡的场景。prefer_local则优先把任务留在本机减少跨网络传输适合单体多进程部署。4.2 重试、超时和熔断怎么保证“一定到达”“触达”两个字听着简单落地很难。最难的地方在于你不能因为害怕重复就完全不重试也不能因为追求必达就无限重试把下游打爆。Agent-Reach 给每个触达任务提供了几个核心参数参数默认值建议值作用max_retries23最大重试次数不算首次retry_backoff_base0.5s0.5s第一次重试的等待基础时间retry_backoff_factor22每次重试翻倍比如 0.5s、1s、2sretry_jitter0.10.1加入随机抖动防止多个任务同时重试造成惊群ttl60s按业务定消息最长存活时间防止“死任务”一直占坑dedup_window60s60s同一message_id在窗口内只消费一次避免重复处理我强烈建议每次触达都带一个唯一的message_id。因为 Agent-Reach 的投递语义是“至少一次”不是“恰好一次”。也就是说网络抖动、Agent 响应超时、进程重启都可能导致一条消息被投递两次。如果消费端不做幂等重试就会变成事故。比如你做了一个“发送验证码”的 Agent触发重试时用户可能收到两条验证码体验很差。所以在消费逻辑里第一步应该是检查message_id是不是已经处理过处理过就直接返回上次结果。重试也不是无限重试。当连续失败达到阈值比如 3 次时Agent-Reach 会触发熔断直接把后续发往该能力或该 Agent 的请求切换给 fallback或者在规定时间内不再选择这个故障节点。熔断的意义是保护下游而不是拯救单条消息。你让一个已经超负荷的 Agent 再多收 10 条重试任务只会让它雪上加霜不如让它缓缓。4.3 广播与定向两种触达模式前面讲的主要是“定向触达”也就是一条任务只找一个 Agent 处理。但实际场景里还有一类需求是企业微信群发式的好消息比如“所有 Agent 给我同步自己的状态”“订单状态变化要通知所有关注方”。这时单点投递就不够用了需要广播。Agent-Reach 用broadcast方法处理results await client.broadcast( topicsystem.heartbeat, payload{ts: 2025-09-19 10:30:00}, strategyall )strategyall会命中所有注册了对应 topic 的 Agent并把每个 Agent 的返回结果汇总成一个列表。如果没有 Agent 注册广播消息不会报错只会返回空列表。这里有一个设计哲学广播消息本身就是“尽力通知”你并不知道也不用关心谁在听但如果需要确认每条都触达成功就要用broadcast加上结果校验。你可以把“定向”和“广播”结合使用我经常这么搭先用广播让业务 Agent 刷新本地缓存再用定向让主处理 Agent 执行任务最后用广播通知所有下游结果。Agent-Reach 手写起来也不复杂几行代码就能把链路组装起来。5. 实操中的坑和排查经验从“能用”到“好用”5.1 我踩过的三个坑写这个框架的过程中我把自己埋进了一堆莫名其妙的坑里。分享三个最典型的希望大家不要重复踩。第一个坑是“消息显示已投递但 Agent 没处理”。这个问题的根源是 ACK 时机不对。我一开始在路由中心把消息放入队列后就立刻标记为已投递但那时 Agent 还什么都没做。后来我把投递状态拆成了“已入队”“已派发”“已确认”“已完成”四个阶段只有在 Agent 执行完业务函数并返回结果时才把消息标记为真正完成。排查时你用reach_ctl task info --trace_id tr_xxx一看就知道卡在哪一步。第二个坑是“重试风暴”。有一次某个下游数据库连接池被打满导致大量 Agent 响应超时。由于每条消息都开启了重试而且重试间隔没加抖动几乎在同一瞬间几十条重试任务同时涌向同一个故障 Agent直接把它的连接池打到了 100%。从那以后我给所有路由规则默认开启了熔断并加上了 0.1 秒抖动。重试的目的是“给系统一个恢复的机会”不是“多试几次证明它真的不行”。第三个坑是“Agent 重启后任务神秘消失”。原因是消费者在处理消息时奔溃Redis Stream 里虽然还能看到消息但已经超时进不了消费者组。后来我专门为这种情况配置了死信队列dead_letter.topic所有超过重试次数或者 TTL 的消息统一进死信再定时跑一个重放脚本扫死信。这样至少不会让问题静默消失日志里能看到“这 10 条消息处理失败了”。5.2 生产环境参数调试记录我把我常用的一套生产参数贴出来起点参考这个再根据业务压测调整参数我的配置说明max_retries3再多了容易让下游雪崩少了对抖动场景覆盖不足ttl30s每个任务最长存活 30 秒适合绝大多数内部接口dedup_window120s必须大于单次最长重试窗口防止第 4 秒重试时重复concurrent8单个 Agent 并发处理数先按 CPU 核心数乘 2 起步queue_maxlen100000单队列最大长度超过直接拒绝新消息并告警heartbeat_interval5sAgent 心跳上报时间决定故障发现速度这里特别提醒一个细节ttl一定要比你的最大业务耗时 最大重试耗时都大。如果你的单次业务处理要 20 秒重试 3 次各间隔 0.5s、1s、2s那ttl至少得给 30 秒左右。否则消息还没处理完TTL 就到期了路由中心会把任务重新派发导致同一个 Agent 在处理同一个消息时产生并发冲突。5.3 可观测性设计写 Agent 协作框架最容易忽略的就是可观测性。Agent 一多任务流转就很复杂如果没有全链路追踪出了问题根本不知道从哪查起。Agent-Reach 里的每个任务都要求带trace_id并且这个 ID 会透传到所有日志、Redis 消息体和外部调用的 headers 里。我建议在业务代码里统一加一个日志过滤器把trace_id、topic、agent_name、status、duration_ms都打出来。这样你就可以在日志系统里直接按trace_id串起一整条链路。我还给每个 Agent 暴露了小端点GET /healthz # Agent 是否活 GET /metrics # 待处理任务数、成功率、平均耗时、重试次数/metrics输出的是纯文本指标可以直接接到 Prometheus。我最常盯的指标是route_unmatched_total也就是“找不到 Agent 处理的消息数”。这个指标一上涨说明你的业务语义和 Agent 能力注册之间出现了偏差要么是有新业务没注册要么是注册的 Agent 全挂了。处理优先级比单纯看 CPU 使用率重要得多。6. 后面怎么玩把 Agent-Reach 用得更顺手6.1 和常见的 Agent 编排框架结合Agent-Reach 并不排斥现有的编排框架。比如你可以把 Agent-Reach 的触达能力放在“工具调用”层让 LLM Agent 在需要查订单时不是直接调用 Python 函数而是调用client.reach(topicorder.query, ...)。这样 LLM 不需要知道订单在那台机器上只需要描述意图剩下的由 Agent-Reach 处理。我试过这种方式之后最直接的变化是新增 Agent 时不需要再改 LLM 的 few-shot 样例里的工具列表因为工具参数只剩一行“topic payload”Agent 注册好能力系统自动就把路由打通了。这里要注意的是LLM 生成的 topic 和 payload 不像手工代码那么可靠。我建议在 LLM 调用 Agent-Reach 之前加一个校验层把 topic 白名单和 payload 的必填字段提前确认好防止模型幻觉生成一个不存在的路由键把消息打进死信。6.2 加一层权限与规则版本管理如果你和我一样是多团队协作路由规则会变成公共资产。谁都能改route就意味着谁都能把消息导到自己的 Agent 里这很危险。我后来给 Agent-Reach 加了一个简单的规则版本管理每条路由规则都有version、owner、status三个字段默认状态是draft只有审批通过后才变成active。规则变更记录会写进审计日志万一出问题可以快速回滚到上一个版本。这一层改动很小但价值极大。它本质上把“技术路由决策”变成了“团队协作决策”避免了一个人改了规则、所有人跟着不稳定。6.3 最后想多说一句Agent-Reach 最让我感慨的并不是代码量而是“触达”这两个字被认真对待之后整个系统给人的安全感完全不一样了。以前我每天担心某个 Agent 会不会挂了现在我把这份担心交给了协议层。每次有人跟我说他写了一套多 Agent 系统我都会先问一句话如果某个 Agent 不回应了你的消息会怎样如果你答不上来那可能你需要的不是一个更智能的模型而是一个更可靠的触达层。我个人在实际操作中体会最深的是框架选型和参数调优都是表面功夫真正的分水岭在于你是否真的想清楚了消息的生命周期。想清楚了Agent-Reach 只是把这件事工程化罢了。希望这篇文章能帮你减少一些“找不到人”的焦虑。