前阵子我这边同时在跑三个Agent一个负责收集线索和整理销售情报一个负责根据素材自动生成推广内容还有一个做客户回访和建联。单看每个Agent都挺能干但真正串起来的时候我发现事情开始变得不可控——它们之间互相喊话靠的是轮询数据库、互相监听临时文件、甚至有人直接让我手工复制产物再塞进下一个Agent的输入框。一时间我分不清自己是在做系统集成还是在当人肉管道。后来我花了两周时间写了一个轻量级中间层把它叫Agent-Reach。这个东西解决的核心问题就一句话让Agent之间、Agent与外部渠道之间有一个稳定、可控、可观测的触达方式。今天这篇文章就把它的设计思路、核心实现和踩坑记录完整捋一遍给正在搭多Agent系统的朋友做一个参考。先说清楚Agent-Reach适合谁。如果你只是单机调一个Agent不需要这玩意儿如果我上面说的三个Agent之间的协作场景你还没有遇到那也暂时用不上。但如果你已经开始让不同Agent互相传递结果、想让Agent自动发飞书通知、想给Agent挂一个Webhook入口、想让多个Agent共享同一套消息路由能力那Agent-Reach的模型对你应该是有用的。1. 为什么在Agent协作里最难的不是模型能力而是触达1.1 我遇到的实际场景三个Agent各说各话我先交代一下具体背景。线索Agent是Python写的每小时从公开渠道抓一批潜在客户信息它需要一个地方把结果吐出来。内容Agent是Node写的它能接收一段素材自动生成小红书文案但它不关心素材从哪来。回访Agent更麻烦它需要知道内容Agent生成的结果再决定给哪些客户发消息同时还要把回访记录回传给线索Agent闭环。最原始的方案是什么线索Agent把结果写进MySQL一张表内容Agent定时扫表回访Agent再从内容表里读。这个方案的问题不是跑不通而是只要有一个Agent挂了、或者表结构变了、或者有人手动改了数据整条链路就静默失败。排查的时候要从日志一层层翻非常痛苦。我意识到真正缺的不是Agent的推理能力而是它们之间的一条标准化通路。这个痛点在多Agent系统里其实有个名字叫Agent-to-Agent通信问题。单个Agent内部有ReAct循环、有工具调用很成熟但多个Agent之间怎么串联、怎么定协议、怎么确认送达基本上没有统一答案。市面上有消息队列、有HTTP Webhook、有RPC框架但很少有专门以Agent为第一公民设计的触达中间件。1.2 现有方案梳理都不是不行但都差一点意思我mentally过了一遍可以复用的技术方案列个表感受一下方案能做Agent触达吗痛点HTTP Webhook能一对一没有路由、鉴权、重试消费者地址变了就要改代码Redis Pub/Sub部分能适合广播不适合定向路由消息丢失无感知消息队列RabbitMQ/Kafka能重要维护集群Agent需引入客户端SDK偏数据管道数据库轮询能但笨定时差、负载高、状态难追踪MCP模型上下文协议不是干这个的MCP解决的是模型调用外部工具的协议不是Agent之间互相投递消息对比完之后你会发现我需要的是一个介于Webhook和MQ之间的东西它要有标准化的消息格式要有路由能力要能接不同渠道又要轻到单个进程就能跑。这就是Agent-Reach的直接动机。1.3 Agent-Reach的定义一个专注于触达的轻量中间层Agent-Reach的核心抽象可以概括为三条Agent到Agent任意Agent可以向任意Agent发送结构化消息无需关心对方的真实地址。Agent到渠道Agent可以通过连接器把消息推到飞书、企业微信、邮件、Webhook等外部系统。渠道到Agent外部系统可以通过统一入口回调某个Agent实现双向触达。这样它就同时兼了消息路由和渠道适配两件事。你可能会问这跟一个带Webhook的聊天机器人框架有什么区别ChatBot框架通常预设了人作为交互对象而Agent-Reach把交互对象泛化成Agent人只是其中一种渠道。这个视角差异在后面做设计时会体现得很明显。2. Agent-Reach核心设计一个消息信封决定的架构2.1 信封Envelope所有消息的统一外壳做Agent-Reach的时候我第一个决定就是绝对不能让Agent之间直接丢裸JSON过来。没有统一外壳的消息就像没贴面单的快递路由系统根本没法处理。所以我定义了一个信封结构用Python的dataclass表达出来就是这样的from __future__ import annotations import time import uuid from dataclasses import dataclass, field from typing import Any, Dict, Optional dataclass class Envelope: msg_id: str field(default_factorylambda: uuid.uuid4().hex) trace_id: str field(default_factorylambda: uuid.uuid4().hex) sender: str # 来源Agent ID recipient: str # 目标Agent ID或路由Key channel: str agent # 目标渠道类型默认agent payload: Dict[str, Any] field(default_factorydict) event_type: str message # 业务事件类型 ttl: int 60 # 消息存活时间秒 timestamp: float field(default_factorytime.time) retry_count: int 0 ack_required: bool True说说每个字段为什么存在。msg_id是幂等去重的依据trace_id是一条消息从发出到最终投递的全局追踪ID排查问题全靠它sender和recipient是路由的起点和终点channel决定这条消息是继续发给Agent还是从某个渠道发出去ttl防止消息积压在队列里成为僵尸消息retry_count配合ack_required实现可靠投递。这里有一个容易被忽略的设计点event_type。Agent-Reach不把消息定义成传统RPC里的方法调用而是定义为事件。调用是你请求我、我返回结果事件是你通知我、处理结果由我自主决定。Agent之间应该更偏事件驱动因为对方Agent什么时候处理完、怎么处理调用方其实不需要同步等待。把语义定位成事件整个系统的解耦程度会高很多。2.2 为什么先用Envelope路由而不是直接连MQTT可能有人会想MQTT不也是发布订阅吗Agent订阅一组Topic不也能实现触达理论上可以但MQTT的消息体对应用层是无感知的它只保证字节级传输。Agent-Reach信封的核心价值在于把路由决策、鉴权、去重、追踪这些能力放到了消息协议层面而不只是发出去就完事。用生活类比来说MQTT像一条公路保证车能开过去但车上装的是什么、该在哪个路口下来、下了车之后有没有人接它不管。Agent-Reach更像一个有分拣中心、有配送员、有签收确认的快递系统它的职责不只是送达还包括送到对的地方、确认对方收到、丢了会补发。多Agent协作场景里确认送达和路由准确的重要性往往比传输速度更关键。2.3 路由表与连接器一个轻量交通枢纽信封定义好之后Agent-Reach的进程本体就好办多了。它内置一个路由表维护每条路由规则routes: - from: lead_agent to: content_agent event_type: lead.ready - from: content_agent to: reach_agent event_type: content.generated - from: reach_agent to: channel/slack event_type: notification.send路由规则分为两类Agent到Agent的定向路由和Agent到渠道的外发路由。路由匹配顺序是精确匹配优先然后事件类型匹配最后兜底丢进死信队列。同时我还做了一个很轻的注册机制。Agent启动的时候如果配置了agent.register就到Agent-Reach这里登记自己的ID和健康状况。注册表直接放在内存里加一个30秒一次的TTL过期不做持久化。因为Agent-Reach本身不是注册中心它只需要知道当前有哪些Agent处于可触达状态就够了。这个做法牺牲了一点强一致性但换来了极高的吞吐和极低的运维复杂度。3. Agent-Reach的代码落地几个关键模块我这样写3.1 消息路由核心一个函数搞定分发Agent-Reach的消息循环非常朴素从入口队列里拿一条消息解析信封查路由表决定下一跳。我把核心路由逻辑收敛到了一个处理函数里async def process_envelope(env: Envelope, router: Router) - None: # 1. 幂等检查 if await dedup.is_duplicate(env.msg_id): logger.info(fduplicated message ignored: {env.msg_id}) return # 2. TTL检查 if time.time() - env.timestamp env.ttl: await dead_letter.push(env, reasonttl_expired) return # 3. 路由匹配 targets router.resolve(env) if not targets: await dead_letter.push(env, reasonno_route) return # 4. 按目标分发 for target in targets: try: connector connector_registry.get(target.connector_type) await connector.dispatch(env, target) env.ack_required and await ack_queue.put(env) except Exception as exc: logger.exception(dispatch failed %s, env.msg_id) await retry_queue.put(env)这个函数看起来简单但每一步都有讲究。幂等检查必须在路由之前否则渠道高并发的重复消息会把后面的环节冲垮TTL检查放在路由前能避免把过期的消息发给消费者connector.dispatch失败之后统一丢给重试队列由重试模块负责退避与封顶。实际运行的时候process_envelope会被包装成一个Worker消费入口队列的并发数可以通过配置调整。我在单机上用8个Worker压过每秒大概能处理2万条轻量消息完全够Agent场景使用。3.2 连接器Connector每个渠道一个插件连接器是我认为Agent-Reach里最体现插拔价值的部分。每个外部系统都对应一个Connector实现核心接口就四个方法class Connector(ABC): abstractmethod async def connect(self) - None: ... abstractmethod async def dispatch(self, env: Envelope, target: Target) - bool: ... abstractmethod async def healthcheck(self) - bool: ... abstractmethod async def close(self) - None: ...以飞书机器人这个渠道为例connect阶段构建一个带签名的HTTP客户端dispatch阶段把信封里的payload转成飞书消息卡片async def dispatch(self, env, target): resp await self.http_client.post( fhttps://open.feishu.cn/open-apis/im/v1/messages, params{receive_id_type: open_id}, json{ receive_id: target.address, msg_type: text, content: json.dumps({text: env.payload[text]}) }, headersself.headers ) return resp.status_code 200关键点在于dispatch方法的返回值。我让Connector在成功投递到对方API时返回True而在网络错误、API超时这类不确定情况下抛异常或者返回False。Agent-Reach只根据这个返回值决定要不要进重试队列至于业务上消息有没有真的被用户看到那是由业务层自己负责的。把技术投递和业务送达分开能避免很多过度承诺的麻烦。3.3 重试队列与退避策略别用蛮力要有节奏消息投递失败的时候直接丢弃肯定不行无限重试也肯定不行。Agent-Reach里我采用了一套指数退避加抖动的重试策略async def retry_worker(env: Envelope) - None: for attempt in range(MAX_RETRY): try: await connector_registry.dispatch(env, env.target) return except Exception: wait_time BASE_BACKOFF * (2 ** attempt) random.uniform(0, 0.2) await asyncio.sleep(wait_time) await dead_letter.push(env, reasonmax_retry_exceeded)参数我设的是BASE_BACKOFF1s、MAX_RETRY5。这样第一次重试等1秒第二次2秒第三次4秒第四次8秒第五次16秒加上最大0.2秒的随机抖动20秒左右就会停止。抖动不是玄学是为了避免多个Agent同时失败后在同一时刻发起重试形成请求风暴。这个话题在分布式系统里叫惊群你如果自己实现过定时任务应该会有体感。4. 接入真实业务从两个Agent到一个完整闭环4.1 我的第一个Demo自动线索到内容的流转光看代码不好建立体感我分享一下第一个真正跑通的Demo链路。我当时技术选型用Python是因为团队已有模块都在这套技术栈里Agent-Reach被做成一个独立服务各Agent通过HTTP长连接与它通信。Demo里两端是这样的线索Agent把抓取完成的线索列表封装成lead.ready事件调用Agent-Reach的/v1/messages接口推入。内容Agent通过WebSocket长连接订阅到lead.ready事件拿到线索摘要后自动生成文案再以content.generated事件回传。Agent-Reach在这中间做的事情只有一个把消息从A路由到B。但如果少了这层内容Agent就要自己维护线索Agent的地址一旦线索Agent换了端口、加了鉴权、改了消息格式两边都要跟着改。有了Agent-Reach这种变更被收敛到配置层Agent代码完全不用动。4.2 项目配置文件一个YAML看清全部拓扑我习惯把所有Agent和路由的声明收敛到一份配置文件里。看这个示例agents: - id: lead_agent entry: { type: http, bind: :9100 } - id: content_agent entry: { type: websocket, bind: :9200 } connectors: - name: feishu_notifier type: feishu config: app_id: cli_xxx app_secret_env: FEISHU_SECRET default_ack_timeout: 10 dead_letter: { type: kafka, topic: agent-reach.dlq }配置最多的是路由表和连接器配置Agent本身的逻辑一点不用写进Agent-Reach。这个做法让我后来在加第三个Agent的时候只花了十分钟加一条注册信息加一条路由规则完事。这也是我坚持把Agent-Reach做成一个独立服务而不是一个SDK的原因——SDK侵入每个Agent进程独立服务则让触达层变成一个可以独立升级、独立监控的基础设施。4.3 全链路追踪用trace_id串联一切消息一旦开始在多Agent之间流转调试难度指数级上升。Agent-Reach最初版本没有任何追踪能力我Debug的时候特别痛苦。后来加了全链路日志每次消息进入系统或离开系统都打印一行结构日志字段统一[agent-reach] dispatch msg_id4f1a3c9d trace_id9b8e7a6f senderlead_agent recipientcontent_agent event_typelead.ready statusok cost23ms有了这份日志我可以在排查问题的时候做一件很简单的事grep trace_id拿全部链路。哪一段延迟高、哪一段漏发、哪一段重试了一目了然。给Agent系统做观测性我认为第一优先级就是搭好这条trace链路否则后面Agent数量一多出问题基本只能靠猜。5. 踩坑记录我在Agent-Reach上栽过的几个跟头5.1 消息顺序问题不保证全局有序但保证因果有序第一个坑是关于消息顺序的。刚开始我天真地认为只要消息是一条一条发的接收方就会一条一条收到。实际跑了一阵子发现HTTP异步请求导致的乱序是常态。比如内容Agent发送content.ready事件中间走了一次重试后面的事件反而先到了。Agent-Reach最后采用的方案是不承诺全局有序只承诺同一个sender recipient event_type里如果消息带seq字段则按序投递。实现方式也不复杂接收方每个窗口缓存一个expected_seq乱序的消息先入缓冲区等待超时了再触发重新拉取。这个妥协是必须的。如果你真的需要所有消息严格有序那是消息队列的领域不是触达中间件该干的事。5.2 幂等性渠道侧必须靠业务消息做去重第二个坑在渠道侧暴露得很明显。某个连接器在投递飞书消息时偶尔会碰到网络超时——消息其实已经发送成功但响应没回来。Agent-Reach判断投递失败会自动重试结果就是用户在飞书上收到了两条一模一样的通知。这类重复投递问题在分布式系统里叫at-least-once语义几乎可以肯定会发生。修复方案是在所有渠道侧的消息卡片里带上msg_id。飞书这种IM平台天然支持一条消息的msg_id去重Webhook渠道则是在消息体的扩展字段里放trace_id让接收方自己决定要不要处理。记住一句话去重逻辑一定要放在业务侧不要指望Agent-Reach的队列机制帮你完全规避重复。5.3 权限模型触达能力不能随便开放再多说一个权限坑。Agent-Reach初版只有简单的路由功能任何Agent都可以给任何Agent发消息这在内部环境没毛病。但当我通过Agent-Reach向企业微信、短信、邮件这些真实触达渠道发消息时这个设计就危险了。一个被攻破的Agent理论上可以拿着Agent-Reach的令牌给你全量客户发垃圾短信。最后我加了两层控制。第一层是每个Agent独占一个API TokenAgent-Reach根据Token解析sender身份第二层是路由授权白名单规定某个Agent只能投递到哪些Agent和哪些渠道。这两层加完至少拦截了90%以上的误触达风险。如果你要把Agent-Reach推到生产环境权限这一块千万不要省。6. Agent-Reach的边界它不是全能中间件讲了这么多能做什么聊聊它不做什么。Agent-Reach不适合当工作流编排引擎用一个复杂业务流程需要条件分支、并行聚合、人工审批那是工作流系统的事你非要用消息路由硬凑会非常别扭。它也不适合做高吞吐日志管道几GB级别的日志、需要批量刷盘的数据流还是用更专业的数据管道更合适。它也没有完整的事务语义Agent之间的消息投递是尽力可靠不是分布式事务。我的判断标准是如果你的核心诉求是把一条结构化的消息从一个Agent送到另一个Agent/渠道并且你能看到它到了没有Agent-Reach就适合如果你的核心诉求是让一个流程严格地执行三百个步骤并在中途随时回滚请用它旁边的专用流程引擎。最后分享一点实际体会。Agent-Reach真正让我的Agent协作从演示Demo变成可依赖的工程的关键不是它有多聪明而是它把触达变得标准化、可见、可控。先把信封结构定义下来再一个连接器一个连接器地补不要一开始就贪大求全。多Agent系统里每个Agent都在变聪明但让它们之间能稳定地喊到彼此这件事本身就值得做扎实。如果你也在搭类似的东西建议从最小闭环开始两个Agent通过中间层互通消息加一条路由规则加一条链路追踪跑通了再扩展。这个路径我可以负责任地说非常顺。