先说结论：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 或 RabbitMQ，Agent-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_name="demo_workshop", registry_url="redis://localhost:6379/0", queue_url="redis://localhost:6379/1", mode="development" ) ctx.init()registry_url负责存路由注册信息，queue_url负责存消息流，分开用两个 Redis 数据库号可以避免互相挤占。开发模式给你开了一些便利，比如自动重试关闭、日志更啰嗦，等上生产再调严格策略。
3.2 写一个最简单的 Agent 节点
下面是一个最基础的订单查询 Agent。它做的事情很简单：启动时告诉注册中心“我能处理order.query”，然后不断接收路由层派发的任务，处理完再返回结果。
from agent_reach import AgentRuntime, capability agent = AgentRuntime( name="order_agent", context=ctx, concurrent=4 ) @agent.capability("order.query", timeout=10) 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", timeout=10)这个装饰器把“能力”和“处理函数”绑定，timeout告诉调度方这个函数最多跑 10 秒，超过 10 秒就按超时处理。第二，concurrent=4表示这个 Agent 最多同时处理 4 个任务，超过的请求会在队列里排队，而不是无限开线程把进程拖垮。我见过很多新手在这两个参数上不设限制，结果流量一上来直接 OOM，后面要记住。
3.3 配置路由规则
Agent 写完之后，要告诉 Agent-Reach 怎么把消息路由过去。路由规则其实就是一个匹配表：
from agent_reach import ReachRouter router = ReachRouter(context=ctx) router.declare_route( topic_pattern="order.query", required_capability="order.query", strategy="round_robin", fallback="order_fallback_agent" )这个规则的意思是：当收到一个topic为order.query的消息，并且它要求order.query能力时，就从所有注册了该能力的 Agent 里以轮询方式选一个。如果都不可用，就转给兜底 Agent。这样一旦主用处理节点崩溃，请求不会直接失败，而是进入 fallback。
实际生产环境里，路由规则最好通过配置中心下发，不要写死在代码里。因为当你有很多 Agent 时，规则改动很频繁，你不能每次改规则都发一次版本。
3.4 从客户端发起一次触达
服务端配好了，客户端只需要两行：
from agent_reach import ReachClient client = ReachClient(context=ctx) reply = await client.reach( topic="order.query", payload={"order_no": "A10086"}, timeout=8, strategy="round_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_retries | 2 | 3 | 最大重试次数，不算首次 |
retry_backoff_base | 0.5s | 0.5s | 第一次重试的等待基础时间 |
retry_backoff_factor | 2 | 2 | 每次重试翻倍，比如 0.5s、1s、2s |
retry_jitter | 0.1 | 0.1 | 加入随机抖动，防止多个任务同时重试造成惊群 |
ttl | 60s | 按业务定 | 消息最长存活时间，防止“死任务”一直占坑 |
dedup_window | 60s | 60s | 同一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( topic="system.heartbeat", payload={"ts": "2025-09-19 10:30:00"}, strategy="all" )strategy="all"会命中所有注册了对应 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_retries | 3 | 再多了容易让下游雪崩，少了对抖动场景覆盖不足 |
ttl | 30s | 每个任务最长存活 30 秒，适合绝大多数内部接口 |
dedup_window | 120s | 必须大于单次最长重试窗口，防止第 4 秒重试时重复 |
concurrent | 8 | 单个 Agent 并发处理数，先按 CPU 核心数乘 2 起步 |
queue_maxlen | 100000 | 单队列最大长度，超过直接拒绝新消息并告警 |
heartbeat_interval | 5s | Agent 心跳上报时间，决定故障发现速度 |
这里特别提醒一个细节：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(topic="order.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 只是把这件事工程化罢了。希望这篇文章能帮你减少一些“找不到人”的焦虑。
