Agent-Reach：多智能体通信与路由编排基础设施实践

如果你同时在跑多个智能体Agent应用大概率遇到过这个场景A 智能体刚把意图解析完需要找 B 智能体拿数据而 B 智能体可能部署在另一台机器上甚至用的是完全不同的框架。你接了一圈 HTTP 接口写了若干回调地址最后发现链路乱成一团排查一个问题要在日志系统里跳转五次。我做的这套方案内部代号就叫 Agent-Reach定位很直接——解决多智能体之间“怎么触达、怎么协作、怎么不出错”的问题。它不是某个 AI 模型也不是一个业务系统而是一层独立的通信与编排基础设施负责把分散的智能体连接成一张可观测、可控制的协作网络。这篇文章我会把 Agent-Reach 的设计思路、核心协议、路由策略、完整落地代码、以及我在线上踩过的坑一并分享出来。适合正在做 Multi-Agent 系统、智能体平台、RPA 调度或类似“多节点多角色协作”场景的工程师参考。你不需要提前具备什么特殊知识只要写过 Python、见过消息队列就能把最小可用版本搭起来。1. 整体设计与思路拆解1.1 Agent-Reach 究竟解决了什么痛点先说痛点。多智能体系统看起来很美真正落地时最先遇到的不是模型效果问题而是“沟通成本”。智能体之间要互相调用可能要经过这些步骤知道对方存在、知道对方的能力、按约定格式传参、传完要确认对方是否收到、处理完要能拿到结果、整个链路还要能追踪。如果这些全靠开发者在业务代码里临时拼凑就会出现三类问题调用关系硬编码。Agent A 写死调用 Agent B 的 URLAgent B 换了地址就得改代码。协议五花八门。有人用 JSON 串行传输有人用 XML有人直接把 Markdown 结果塞进请求体导致对端解析逻辑越来越脆弱。可靠性无人负责。消息发出去没有回执超时了不知道是对方没收到还是处理太慢更谈不上重试、幂等、降级。Agent-Reach 的做法是在所有智能体之上增加一层独立通信平面。每个 Agent 启动时向中心注册自己的身份、能力、地址和负载状态调用方不再需要关心目标在哪儿、用的是什么协议只需要告诉 Agent-Reach“我要完成什么意图”由框架负责寻找合格执行体、投递请求、跟踪结果。一句话概括Agent-Reach 把智能体之间的“点到点乱连”改成了“星型协作网络”把临时协商改成了标准化流程。1.2 四层架构与关键取舍我设计 Agent-Reach 时没有发明新协议而是复用了成熟组件围绕四个层次搭建分别是接入层、路由层、协议层、可靠性层。接入层负责屏蔽异构框架差异。不管是 Python 写的 Agent、Node.js 写的服务还是通过命令行的脚本 Agent都能通过 Agent SDK 注册进来。它保证的是“人人可接入”。路由层负责决定消息找谁。支持三种路由模式直接寻址指定 Agent ID 投递、能力寻址按技能标签路由、意图寻址解析意图后动态选择执行链。线上使用最频繁的是能力寻址因为它能支持 Agent 水平扩展和故障切换。协议层统一了消息格式。所有跨 Agent 通信都走同一种信封结构内容再千差万别信封永远是同一套字段。这样日志采集、链路追踪、权限校验才有稳定的抓手。可靠性层承担重试、超时、幂等、熔断。智能体毕竟不是数据库它可能因为模型响应慢、外部 API 限流而反馈迟滞所以可靠性不能只靠 TCP 那种简单确认。设计上有个刻意的取舍没有把 Agent-Reach 做成一个“大而全”的分布式中间件而是强调薄、轻、透明。它不做消息持久化之外的复杂存储也不接管 Agent 内部的业务流程这样既降低了接入成本也避免成为性能瓶颈。2. 核心协议、路由与编排2.1 消息信封所有协作的公共语法跨 Agent 通信的第一要务是让通信双方对“一条消息长什么样”达成一致。Agent-Reach 采用了一个类似消息信封的 JSON 结构我把它简化为六个核心字段。{ version: 1.0, message_id: uuid-字符串-全局唯一, trace_id: 链路追踪用-ID, sender: agent-a, target: agent-b, context: { msg_type: request, timeout_ms: 30000, qos_level: 1 }, payload: { intent: fetch_user_order, params: { user_id: U12345 } } }字段不多但每个都有明确作用。message_id 是幂等判断的基础重试前后必须是同一个 ID接收方才好去重。trace_id 贯穿整条调用链排查多跳问题时靠它把日志串起来。sender 和 target 用于路由匹配。context.qos_level 是服务质量档位0 表示尽力投递1 表示至少一次2 表示精确一次。值得多提一句 payload。payload 里尽量不要传大文本比如“把这一段网页内容总结一下”更适合的方式是传资源引用地址让接收方自己去取。因为智能体的执行时间和模型推理时间都不可控如果消息体里塞了大量数据内存、队列、序列化开销会一起爆掉。我在早期设计里踩过这个坑后面会细说。2.2 路由策略从“人肉指定”到“意图分发”路由是整个 Agent-Reach 最核心的部分。我实现了三层路由策略从简单到复杂分别为直连路由最简单消息里显式写好 target路由层直接匹配注册表里的 Agent ID找到地址就投递。适合确定性的内部调用比如“OCR Agent 必须调用图片压缩 Agent”。能力路由是生产环境中用得最多的场景。Agent 注册时会声明自己的 skill 列表比如 skill:[order_query,refund_apply]。请求方无需知道谁有这能力只需要在 target 字段填一个虚拟能力名由路由层从注册中心拉取具备该能力的在线节点按负载选择一个投递。这样做的好处是当你新增一个性能翻倍的 Agent 处理订单查询时不需要改动任何上游调用方只要注册进来即可分流压力。意图路由则是更上游的方案。调用方只描述要完成的目标比如“帮我查一下最近一笔订单的物流状态”意图路由会先把这句话解析成结构化意图拆解为查订单节点、查物流节点再按依赖关系编排成执行图依次投递给对应 Agent。这个路由模式最灵活但对底层自然语言理解的依赖强我一般建议在核心链路上先用手工编排意图路由作为锦上添花。2.3 任务编排状态机取代回调地狱跨 Agent 调用往往不是单跳而是多跳。比如用户问“我的退款什么时候到账”可能需要订单 Agent 查出原订单支付 Agent 查退款单风控 Agent 校验状态最后汇总给对话 Agent 生成回复。这种多跳逻辑如果用回调函数写代码会非常难以维护。Agent-Reach 用任务编排状态机来管理。每种任务模板定义了几个基本状态PENDING、RUNNING、WAITING_DEPENDENCY、SUCCEEDED、FAILED、TIMEOUT。任务启动时先生成实例然后由编排引擎按依赖关系逐步推进。这一步最大的收益在于“失败可视化”。以前链路断在某处你只能来回翻日志才知道哪里断了。用状态机后每个节点独立展示状态谁卡住了、谁失败了、重试了几次都一清二楚。排查问题的思路也从“猜”变成了“看”。3. 从零搭建 Agent-Reach 最小可用系统3.1 技术选型与依赖清单搭建这套系统我不建议一开始就上大而全的微服务全家桶。先保持最小依赖跑通主线之后再按需替换组件。我这里的技术栈是 Python 为主依赖如下Python 3.10FastAPI提供 Agent 注册、路由转发、状态查询的 REST APIRedis保存注册表与路由缓存也用来做分布式锁RabbitMQ承担异步消息投递队列SQLite / PostgreSQL保存任务实例与执行轨迹消息队列为什么选 RabbitMQ 而不是 Kafka因为智能体交互更看重灵活路由和确认机制RabbitMQ 的 Direct 交换机、Topic 交换机、手动 ack、死信队列都很贴合 Agent-Reach 的需求。Kafka 更适合同一个大规模数据流被多个消费者读的场景比如日志收集而不是点对点的任务分发。注册中心用 Redis 而不是 ZooKeeper因为 AgentReach 的注册信息是弱一致的节点崩溃后几秒内被摘除完全可以接受Redis 的简单性和低运维成本更适合这个场景。3.2 核心实现注册与路由首先是 Agent 注册的数据结构。我用一个 hash 存储 Agent 的基本信息key 是 agent_id字段包括地址、能力列表、负载状态、最后心跳时间。# register.py import redis import json import time r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) AGENT_HASH_KEY agent_reach:registry def register_agent(agent_id, address, skills, metadataNone): record { agent_id: agent_id, address: address, skills: json.dumps(skills), updated_at: time.time(), status: online, metadata: json.dumps(metadata or {}), } r.hset(AGENT_HASH_KEY, agent_id, json.dumps(record)) # 单独存一份 skill - agent_id 的索引加快能力路由查找 for skill in skills: r.sadd(fagent_reach:index:{skill}, agent_id) return record然后是能力路由查找。处理请求时先查 skill 索引拿 Agent ID 列表再根据负载因子选一个。这里负载信息来自心跳报文中的 pending_tasks 数量。# router.py def route_by_skill(skill): agent_ids r.smembers(fagent_reach:index:{skill}) if not agent_ids: return None best None best_load float(inf) for agent_id in agent_ids: raw r.hget(AGENT_HASH_KEY, agent_id) if not raw: continue record json.loads(raw) if record.get(status) ! online: continue load int(record[metadata].get(pending_tasks, 0)) if load best_load: best_load load best record return best这段代码在生产环境会加缓存和熔断逻辑但核心思想很朴素——不要固定打到一个 Agent谁空闲就找谁。3.3 幂等消费与消息确认消息投递之后接收方必须支持幂等。HTTP 请求天然可能被重试如果 Agent 在处理消息时崩溃重试消息就会导致业务重复执行。解决思路是消费前查重。RabbitMQ 消费者里每条消息带上 message_id在 Redis 里设置一个执行锁。如果锁存在说明这条消息正在被处理或已经处理完直接 ack 并丢弃否则先设置短过期锁再执行业务逻辑。# consumer.py import pika import redis import uuid r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def process_message(ch, method, properties, body): msg json.loads(body) lock_key fagent_reach:dedup:{msg[message_id]} if r.set(lock_key, 1, nxTrue, ex300): try: handle_business(msg) ch.basic_ack(delivery_tagmethod.delivery_tag) except Exception as e: log_exception(e) # 重试逻辑省略按策略决定是重新入队还是进死信 ch.basic_ack(delivery_tagmethod.delivery_tag) else: # 重复消息直接确认 ch.basic_ack(delivery_tagmethod.delivery_tag)消息队列的自动 ack 默认是“一收就确认”如果 Agent 处理到一半崩溃消息就丢了。所以务必使用手动 ack业务代码跑完再确认。这一点看起来细节却是线上事故最常出现的地方。3.4 可观测性链路追踪与轨迹回放Agent-Reach 里每个请求都有一个 trace_id所有日志打印都会携带这个字段。我自己实现了一个极简的日志过滤器把 trace_id 注入到结构化日志里。# logger.py import logging import threading class TraceContext(threading.local): def __init__(self): self.trace_id - trace TraceContext() class TraceFilter(logging.Filter): def filter(self, record): record.trace_id trace.trace_id return True后面接上 Loki 或 Jaeger就可以实现从“用户问题进入”到“最终回答返回”的全程追踪。没有这个能力多 Agent 协作排查问题的成本会高到你怀疑人生。我见过太多团队在上线第二周就嚷嚷要重构就是因为链路追踪欠了技术债。4. 常见问题与排查技巧实录4.1 消息丢、重复和延迟怎么分清跨节点系统网络抖动是常态。消息落地到 RabbitMQ 后最怕的不是队列变成而是“投递成功但消费失败”。我遇到最多的报错是 Agent 收到了消息但业务处理中抛异常导致手动 ack 没执行消息又回到队列头部。如果异常持续出现就形成了死循环。排查口令是先看 ack 时机再看异常捕获范围最后看消息去重机制。具体来说确认收到的消息多久才 ack处理完成前根本不该 ack。检查 try/except 是否把异常吞掉了而没有走重试或死信。确认 Redis 去重锁的过期时间不能短于业务最大执行时间否则同一消息会被重复消费。消息延迟则是另一个维度。出现延迟不要急着怀疑 RabbitMQ先查消费者侧。Agent 如果还在向大模型请求输出队列消息就会积压。这一类延迟的本质原因是消费速度和上游模型推理时间不匹配缓解手段是给不同优先级任务配置独立队列不能所有消息混在同一个队列里。4.2 路由黑洞服务明明注册了请求就是找不到路由黑洞是 Agent-Reach 最经典的问题。现象是调能力路由返回 None可注册中心的 Agent 明明在线。原因基本就三类一是心跳过期。我把 Agent 的心跳超时时间设为 15 秒注册信息上有 updated_at但代码里可能忘了更新心跳线程。排查方式是看 Redis 里记录的时间是否新鲜。二是 skill 索引和注册表数据不一致。新增 skill 时只更新了 agent hash忘了同步 sk.index 集合路由自然找不到。三是负载信息里的 pending_tasks 涨到阈值被自己的熔断逻辑摘除了。这种情况日志里会有熔断记录排查时先看日志关键字“circuit_break”就能命中。路由黑洞的共性规律是大多数时候不是网络断了而是“元数据不一致”。所以我在 Agent-Reach 里设计了一个修复脚本每天定时扫描注册表把有 agent 记录但没有 skill 索引的节点重建索引。这类“对账逻辑”看起来不性感却是保证系统稳定性的重要防线。4.3 性能瓶颈序列化和大消息体是隐形杀手早期版本我放过两个重要的性能坑。第一是 JSON 序列化开销每个跨 Agent 消息都做一次 json.dumps / loads节点一多CPU 时间全耗在来回编码上。之后我对大消息体改用 MessagePack小消息保持 JSON实测 CPU 占用下降约 18%。第二是队列里堆了大量“大 payload”。有个业务把整份 PDF 内容塞进消息体导致 RabbitMQ 单条消息达到几十 MB队列内存直接告警。后来的规则是消息体尽量传引用地址文件走对象存储Agent 拿到 URI 后自行下载。这条约定写进了团队开发规范之后相关的内存告警基本绝迹。如果流量继续上涨可以考虑把 RabbitMQ 换成 gRPC 做同步短连接或者引入流式处理框架。但默认情况下保持消息小、路径短、尽量异步已经能支撑中等规模的智能体集群。4.4 一套实用排查口诀我在团队里总结了一套排查口诀“先看追踪链再查路由段慢则看队列错则查 ack”。每次线上报警按这个顺序走大多数问题能在十分钟内定位。补充几个实用的操作细节Redis 里用 keys agent_reach:* 快速捞全部状态键实测在生产环境几百个键性能完全扛得住。RabbitMQ 管理后台主要看 ready 数和 unacked 数ready 持续上涨说明消费跟不上unacked 持续上涨说明消费者被业务逻辑卡住了。日志里必须打 message_id 和 trace_id否则重试消息和原始消息你很难对应上。为关键链路配置告警规则比如“某能力路由 5 分钟命中率为 0”就要立即告警别等服务方找上门才意识到路由挂了。我在实际操作中最深的体会是Agent-Reach 这类基础设施真正难得的部分不是“把消息发出去”而是把“发出去之后发生了什么”这件事讲清楚。设计一套协作协议时先不要追求花哨的调度策略先把注册、路由、幂等、追踪这四根柱子立稳后面的协作逻辑才有踏实的地基。这个方案后续还能扩展出不少东西比如把意图路由接入更强大的语义模型、增加多集群联邦调度、或者把路由策略做成可动态编排的 DSL。不过在当前阶段先把最小可用系统跑稳把线上问题挨个处理干净就是最有价值的投入了。