很多人第一次看到Agent-Reach这个名字脑子里冒出来的问题大概跟我当时一样Agent 我懂Reach 是什么意思是触达是覆盖还是让 Agent 之间能够到彼此我当时正在做一个内部的多智能体项目三个团队各自维护着三个完全不同技术栈的 Agent——订单查询、物流跟踪、售后工单。前期没人把这玩意儿当系统工程做谁要调用哪个能力直接问人要接口文档然后 HTTP 怼过去就完事了。结果第一个月还好第二个月开始频繁出问题某团队顺手改了个字段名调用方直接崩物流 Agent 高峰期响应超时没人重试用户那边体验就是客服突然不说话了更离谱的是有一次 A 团队的 Agent 想调用 B 团队的 Agent结果发现对方根本没有对外接口只能让运维手动开防火墙端口然后传了一个临时 token。那一周我都在思考一个问题如果 Agent 的数量从三个变成三十个、三百个呢我们需要的根本不是再加一个接口网关而是一层能理解能力语义的连接层让任何一个 Agent 都能知道谁有这个能力我该怎么触达它它没空的时候怎么办。这其实就是 Agent-Reach 这个项目的起点。这篇博文我就把 Agent-Reach 从需求分析、架构设计到落地实现、运维排坑的完整过程梳理一遍。如果你也在做多 Agent 协作平台、打算把公司内部的 AI 助手串起来或者单纯对智能体如何互相找到对方并可靠协作感兴趣这篇文章应该能给你一个可以直接参考的路线图。1. 为什么需要Agent-Reach多Agent协作里的触达问题先聊聊项目背景。当时我们手上的三个 Agent 其实都不复杂业务逻辑都是接收自然语言或结构化参数调用一些内部系统返回结果。但把它们组合成一个完整业务流程的时候问题出现了——不是某一个 Agent 的问题而是它们之间怎么说话的问题。1.1 一次失败的多Agent对接给我的教训我印象最深的一次事故是这样的售前咨询机器人接入了订单查询 Agent 和物流查询 Agent理想状态下用户问我的货到哪了机器人先调用订单 Agent 确认订单号再调用物流 Agent 查询轨迹。我当时图省事让机器人直接硬编码调两个 HTTP 接口。上线前两天一切正常第三天突然报错——物流 Agent 那边把接口从/api/v1/track升级到了/api/v2/track老接口直接下掉了。结果凡是涉及物流轨迹的问题全部返回系统错误。这件事本身是物流团队没按规范做兼容但我在复盘的时候问了自己一个更根本的问题就算这次他们不改了下一次别人改了呢如果我不能把调用方依赖死接口转变成调用方依赖能力本身那任何一次 Agent 的升级都可能引发崩溃。除了接口变更有风险还有两个隐蔽问题。第一是发现机制缺失调用方必须事先知道对方的 URL 和鉴权方式这是一个完全静态的、靠人传话的协作模式。第二是故障处理缺失调用超时了怎么办重试会不会导致重复下单谁来记录这次调用成功还是失败当时这些问题的答案全部是看日志。1.2 Agent-Reach的核心定位所以 Agent-Reach 在我心里逐渐清晰起来它不是一个简单 API 网关也不是纯消息队列而是一个面向 Agent 能力的触达与编排层。它需要解决三件事第一注册与发现。每个 Agent 在启动时向 Reach 层注册自己声明我拥有什么能力我的入口在哪我能接受什么格式的请求。其他 Agent 不需要提前知道你的存在它只告诉 Reach 层我要什么能力Reach 层负责帮它找到合适的提供方。第二协议与路由。所有 Agent 之间的消息统一封装成标准信封Reach 层根据能力标识甚至自然语言描述把消息路由到正确的 Agent屏蔽底层传输协议差异——你用的是 HTTP 还是 WebSocket 还是 Kafka对调用方一律透明。第三可靠性与可观测性。超时、重试、幂等、结果回执、全链路追踪这些是分布式系统的基础设施问题不应该让每个 Agent 自己再实现一遍。Reach 层统一接管让业务 Agent 只需要关心收到请求、处理、返回结果。用生活里的话来说Reach 层就像小区物业的总机。以前各商户各自装电话客户找人得打十几个号码碰上线路故障就彻底失联。有了总机每个人只管报名字总机负责转接、排队、记录通话是否完成。Agent-Reach 就是给智能体同事配了这么一台总机。2. Agent-Reach的架构拆解注册、信封与路由想清楚了为什么做接下来的问题是怎么做。我看了不少现成方案最后设计了一套以三个核心模型为支柱的架构。这套东西不需要特别高级的组件但每个模型背后的设计考量都得讲清楚。2.1 三层模型能力注册、消息信封、路由引擎先看能力注册模型。每个 Agent 在注册时需要提交的不是一个接口地址而是一段能力描述能力标识capability机器可读的短标识比如order.query、logistics.track对应一组操作。能力描述description自然语言描述比如查询订单状态输入订单号返回当前状态和预计送达时间用于语义匹配。输入输出Schema定义请求参数和返回结果的结构调用方可以据此生成参数也可以校验响应。可达渠道endpoints该 Agent 实际接收消息的地址和协议可能是 HTTP 回调地址、消息队列主题甚至是一个 WebSocket 通道。实例元数据权重、健康状态、版本号等用于负载均衡和灰度。注册完成后Agent 的能力信息会进入注册中心后续路由引擎从这里查询。然后是消息信封模型。我们设计了一个统一的信封结构所有经过 Reach 层的消息都长这样{ message_id: uuid-uuid-uuid, trace_id: trace-uuid, producer: crm-assistant, consumer: logistics-agent, capability: logistics.track, payload: { order_id: SO20240101 }, meta: { timeout_ms: 5000, max_retries: 2, reply_to: sync://crm-assistant } }这个信封解决的是协议标准化问题。每个 Agent 内部可以有自己的数据结构但只要进出 Reach 层一律按这个格式处理。message_id是全局唯一标识用来做幂等和链路关联capability是路由依据reply_to决定了调用是同步等待还是异步回调。最后是路由引擎。这是整个 Reach 层最有技术含量的部件。它要完成两件事确定消息该发给哪个 Agent以及确定该发给那个 Agent 的哪个实例。路由优先按能力标识精确匹配如果消息里没有携带明确的能力标识或者精确匹配命中不了就退到语义匹配。所谓语义匹配就是把消息内容和 Agent 注册时的自然语言能力描述做向量相似度打分分数超过阈值才允许路由。2.2 为什么不直接用服务网格或消息队列当时有人问过我我们不是已经有服务网格了吗为什么还要自己做一层这个问题很有代表性。服务网格解决的问题是网络层的通信可靠性——服务发现、负载均衡、mTLS、重试这些它都擅长。但服务网格理解不了帮我查一下 618 那个订单现在到哪了这句话和物流 Agent 的能力描述查询物流轨迹是同一件事。服务网格的路由依据是应用名和 URL 路径不是能力语义。消息队列也有类似的问题。Kafka 和 RabbitMQ 确实能解决异步解耦但它们不关心消息内容是不是一个 Agent 能理解的请求也没有内建的请求-响应关联机制——你发一个请求到队列处理完之后怎么回到调用方虽然可以手动实现回调但那等于把编排逻辑散落在各个 Agent 里又回到了一盘散沙的局面。Agent-Reach 的价值恰恰在于它理解 Agent 之间的对话是能力请求。它既负责网络层面的可靠性也负责语义层面的门当户对。说得直白一点服务网格和消息队列是地基和管线Agent-Reach 是前台接待——地基管线当然重要但最终帮客户找到正确办事窗口的是前台。2.3 路由策略细节再说说路由策略的细节。我们的路由引擎采用二级路由先匹配能力标识后匹配实例元数据。第一级能力标识匹配。调用方如果明确指定capability: logistics.trackReach 层直接去注册中心查所有注册了这个能力的 Agent。如果有多个 Agent 注册了同一个能力比如两个团队都做了物流查询就进入第二级。第二级实例选择。选择策略包括轮询、随机、最少在线数、最近最少调用等。初期我们用的轮询后来改为最少在线数 健康检查加权。加权的原因是两个 Agent 虽然能力相同但一个在一台 4 核 8G 的机器上一个在 8 核 16G 的机器上前者每秒能处理 50 个请求后者能处理 200 个按 1:4 的权重分配流量才合理。语义匹配兜底逻辑当capability字段缺失或者精确匹配返回空时路由引擎会把消息payload里的文本内容做向量化然后和所有 Agent 的描述向量做相似度计算。这里有两个关键参数候选集大小我们取 Top 5和相似度阈值默认 0.72低于阈值直接返回未找到可处理该请求的 Agent。阈值不能设太低否则会把查一下订单物流信息路由到订单 Agent 而不是物流 Agent形成错误的编排链。3. 从零复刻一个最小Agent-Reach核心代码与选型思路理论架构说完很多人最关心的还是这东西到底怎么搭起来。我拿 Python 加 FastAPI 写了一版最小实现全部代码大约 600 行可以跑通注册—路由—同步调用—异步回调全套流程。我把关键模块和选型理由分享出来。3.1 技术选型为什么是FastAPI Redis SQLite先解释一下选型。FastAPI是因为它原生支持async/await在高并发下不容易被 IO 阻塞非常适合做转发层。而且它有自动生成 OpenAPI 文档的能力对调试路由规则很有帮助。Redis在方案里承担三个角色注册表的缓存加速路由查询、消息暂存队列异步模式下保存待发送消息、分布式锁防止多个 Reach 实例并发处理同一条消息导致重复投递。SQLite用作注册中心和调用记录的持久化存储。项目初期数据量很小SQLite 完全够用不用单独运维一个 MySQL 实例。等 Agent 数量上来了可以平滑迁移到 PostgreSQL。3.2 注册与发现模块实现先看最核心的注册接口。每个 Agent 启动时会调用/agent/register把自己登记在册。我简化了代码核心逻辑是保存能力描述和更新健康状态。from datetime import datetime from typing import Dict, List import sqlite3 import json class Registry: 注册中心管理 Agent 能力信息与实例状态 def __init__(self, db_path: str): self.db_path db_path self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS agent_capabilities ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, name TEXT NOT NULL, capability TEXT NOT NULL, description TEXT NOT NULL, input_schema TEXT NOT NULL, output_schema TEXT NOT NULL, endpoint TEXT NOT NULL, protocol TEXT NOT NULL DEFAULT http, version TEXT NOT NULL DEFAULT 1.0, weight INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL, UNIQUE(agent_id, capability) ) ) conn.execute( CREATE TABLE IF NOT EXISTS agent_health ( agent_id TEXT PRIMARY KEY, healthy INTEGER NOT NULL DEFAULT 1, last_heartbeat TEXT NOT NULL ) ) async def register(self, agent_info: Dict) - str: with sqlite3.connect(self.db_path) as conn: conn.execute( INSERT OR REPLACE INTO agent_capabilities ..., (agent_info[agent_id], agent_info[name], agent_info[capability], agent_info[description], json.dumps(agent_info[input_schema]), json.dumps(agent_info[output_schema]), agent_info[endpoint], agent_info.get(protocol, http), agent_info.get(version, 1.0), agent_info.get(weight, 1), datetime.utcnow().isoformat()) ) conn.execute( INSERT OR REPLACE INTO agent_health VALUES (?, 1, ?), (agent_info[agent_id], datetime.utcnow().isoformat()) ) return agent_info[agent_id]这里有个容易被忽略的细节UNIQUE(agent_id, capability)。同一个 Agent 可以注册多个能力但不能重复注册相同能力。如果有更新用INSERT OR REPLACE保证新描述覆盖旧描述这就是 Agent 升级时的基本保障——新版本启动后注册一条新记录旧记录不会残留。这里还涉及一个经验注册接口必须同时写入agent_health表并且要求 Agent 每隔一段时间上报心跳。否则你只能知道 Agent 注册过却不知道它现在还活着没有。3.3 消息信封与路由引擎实现消息信封的实现是一个 Pydantic 模型这里不展开贴完整代码只展示路由引擎的核心逻辑——这也是整个项目最值得看的部分。import numpy as np from sentence_transformers import SentenceTransformer class Router: def __init__(self, registry: Registry, embed_model: SentenceTransformer): self.registry registry self.embed_model embed_model async def route(self, message: Dict) - Dict: # 1. 先尝试精确能力标识匹配 if message.get(capability): candidates self.registry.query_by_capability(message[capability]) if candidates: return self._select_instance(candidates) # 2. 精确匹配失败走语义匹配 embed_text self._extract_text(message[payload]) query_vec self.embed_model.encode(embed_text) all_agents self.registry.query_all_healthy() scored [] for agent in all_agents: agent_vec self.embed_model.encode(agent[description]) score float(np.dot(query_vec, agent_vec) / (np.linalg.norm(query_vec) * np.linalg.norm(agent_vec))) if score 0.72: scored.append((score, agent)) scored.sort(keylambda x: -x[0]) if not scored: raise ValueError(no_agent_found: 没有找到可以处理该请求的Agent) return self._select_instance([a for _, a in scored[:5]])这段代码反映了三个设计决定。第一精确匹配永远优先语义匹配只是兜底这能避免很多误路由。第二语义匹配的候选集只看健康的 Agent避免把一个请求路由到一个已经挂掉的实例上。第三路由结果不是直接返回一个 Agent 地址而是经过_select_instance做权重选择。调用方永远不需要知道具体是哪台机器在处理这对上层是透明的。3.4 同步调用与异步回调的实现差异最小实现里我同时支持了两种调用模式。它们的代码路径差别主要体现在reply_to字段的处理上。同步模式下reply_to设置为sync://...。Reach 层转发请求时会用一个内存字典保存message_id - asyncio.Future的映射。当被调用的 Agent 返回结果时Reach 层根据message_id找到对应的 Future把结果 set 进去外层协程拿到结果后包装成响应返回给调用方。异步模式下reply_to设置为一个回调 URL比如https://reach.example.com/callback/crm-assistant。Reach 层接收到 Agent 的结果后不直接返回同步响应而是把结果保存到 Redis 的callback:message_id键里然后向回调 URL 发送一个 POST 请求。调用方如果想轮询结果也可以直接用它当初拿到的message_id去 Reach 层查询结果。同步模式适合交互型场景——用户正在等待客服机器人给出答案等不了十秒钟异步模式适合流水线型场景——一个 Agent 在后台批处理任务处理完了告诉你我完成了。一个成熟的 Reach 层必须两种都支持否则会被业务场景卡死。4. 部署后我踩过的四个真实坑架构看起来不错代码也跑通了但真正部署上线之后问题才开始冒出来。我把印象最深的四个坑详细记录下来每个都有完整的排查思路和最终解法希望对你有帮助。4.1 重试导致订单重复下单传输可靠≠业务幂等上线第一天就出了个大事故。当时我们把一个支付 Agent 接入了 Reach 层给测试环境导数据用脚本并发模拟 100 个支付请求。结果跑完一看有一笔支付记录出现了两次。一开始我怀疑是 Reach 层重复投递了消息。查 Reach 层的日志发现确实重试了两次——原因是支付 Agent 在处理第一次请求时响应超时了因为测试环境数据库锁性能差。Reach 层在超时后按配置重试了一次支付 Agent 第二次收到消息又执行了一次扣款。问题出在哪出在我把重试设计成无条件的。重试能保证消息至少被送达一次但 Agent 侧如果没有做幂等处理重试就会造成业务上的重复操作。这不是 Reach 层的错但也绝不能说是 Agent 的问题——真实世界里你根本没法强制每个 Agent 都自带幂等逻辑。排查链路是这样的先确认重复执行的 message_id 是否一致。查日志发现两次扣款消息的 message_id 一样说明是同一逻辑消息的重试。确认超时发生在哪一段。日志显示 Reach 层发出请求后支付 Agent 处理了 4 秒Reach 层设置的超时时间是 3 秒于是判定超时。确认支付 Agent 侧是否做了幂等校验。没有——它只校验了业务参数没有校验 message_id 是否已经处理过。解决方案分两层。Reach 层做了改进设置idempotency_retry模式在重试前先查询投递记录如果发现同一条 message_id 已经有成功回执就禁止重试直接返回上次的结果。支付 Agent 侧也做了改进在业务表里增加message_id唯一索引插入时如果发现重复就直接返回原结果而不是再执行扣款。最重要的教训传输层的至少一次语义和应用层的恰好一次语义不能混为一谈。Reach 层能做的是保证投递不丢但不要重复扣钱这件事必须由 Agent 自己保证。4.2 语义路由的误入歧途第二个坑是语义匹配的过度自信。当时我们正式接入了物流 Agent 和订单 Agent然后我拿了一批历史问答消息去测试路由正确率发现有一条消息被路由错了。那条消息是查一下订单号 SO20240202 的物流信息我预期的目标能力是logistics.track。但路由引擎把它路由到了order.query。看相似度打分才发现订单号这个词和订单 Agent 描述里的输入订单号得分极高超过了阈值而物流 Agent 虽然也匹配到不少词但总分略低。这个问题的根源在于语义匹配模型对领域术语的敏感度不够。SentenceTransformer的通用 embedding 模型并不理解订单号出现在物流上下文中只是一个凭证不代表这个请求就在问订单状态。解决思路不是换一个更复杂的模型而是调整路由策略——我上面写代码的时候就强调了精确匹配优先。上线部署时我们进一步明确当一条消息既包含业务上下文又包含能力关键词时先用一个轻量规则抽取器识别带logistics.*前缀的能力词命中就直接走精确匹配不进入语义匹配。另外我们在语义匹配的阈值判断后面加了一步冲突仲裁如果 Top 1 候选和 Top 2 候选的分数差值小于 0.05说明模型自己也犹豫了这时候主动拒绝路由并返回请求不明确请提供更多信息避免强行打包给一个错误的 Agent。4.3 Agent假死健康检查不能只看进程是否活着第三个坑非常隐蔽。某天客服机器人突然大面积报错我登录物流 Agent 的机器一看进程还活着CPU 占用率也正常但所有请求都卡住。检查健康检查接口返回 200。问题在于健康检查接口只是一个 trivial 函数判断结构就是进程没退出就返回 200。但那个 Agent 真正依赖的数据库已经连不上了线程池也几乎耗尽所有请求在数据库连接等待上排队——健康检查却对这一切毫无感知。排查方式是选择一个请求高峰期手动调了一次 Agent 的真实业务接口发现耗时达到 30 秒远超正常值。那一刻才意识到健康检查的口径完全错了。解决方案是把健康检查升级为业务探测。Reach 层对每个 Agent 健康检查时不只发ping而是模拟一个最小业务请求——比如对物流 Agent 发一个查询一个约定好的测试订单号请求。如果这个请求在 2 秒内返回正确结果才标记为健康。如果只是进程活着但业务已经无法服务就立刻标记为不健康路由引擎立刻把它从候选集里踢掉。这个改造在初期会带来一些额外负载但收益远大于成本——它彻底杜绝了假活Agent 占用流量的问题。顺带我还把健康检查数据做了落库统计每个 Agent 从标记不健康到恢复的时长用来观察 Agent 依赖服务的稳定性。4.4 回调风暴高并发回调把Reach打挂了第四个坑是在某次大促压测时出现的。当时我们接了一个批量 AI 分析的 Agent它一次会处理 1000 个任务处理完后会并发回调 Reach 层的/callback接口。1000 个回调请求同时进来Reach 层的 SQLite 连接直接打满大量请求排队等待数据库锁最终导致了 30 秒的雪崩。问题本质是回调接口的写入路径太脆弱它要往数据库里写一条结果记录然后更新消息状态。高并发下 SQLite 的写锁只能串行执行入口没有做任何限流或削峰。排查链路看 Reach 层访问日志发现/callback路由的 P99 延迟从 50ms 飙升到 30s并且出现大量database is locked错误。看 Redis 指标消息队列堆积正常问题集中在回调处理的写路径。压测复现发现只要超过 200 并发写SQLite 就锁死。解决办法分三步走。第一步回调接口接收请求后立即返回 200只把原始结果写入 Redis 的callback_raw列表然后异步消费这个列表批量落库——用 Redis 做缓冲削掉尖峰。第二步落库消费端加上批量写逻辑每攒够 100 条或等待 200ms 才批量执行一次 INSERT。第三步给回调接口加了基于令牌桶的限流超过阈值的请求直接返回 429Agent 收到 429 后有退避重试逻辑。改造之后再压测即便 5000 个回调并发进来Reach 层的响应延迟也稳定在 100ms 以内。5. 从项目到产品Agent-Reach的扩展路径最小版本跑通之后我陆续把 Agent-Reach 从内部工具往更完整的方向推进。这期间有几个扩展点我认为是所有对 Agent 协作平台有兴趣的人都会遇到的值得单独拿出来聊。5.1 Agent能力版本与Schema兼容性治理第一个扩展是能力描述和消息格式的版本管理。最初版本里一个 Agent 更新了输出 Schema所有调用方的代码只要字段名对不上就崩。这其实跟微服务里的接口版本问题一样但在 Agent 场景下更隐蔽——因为调用方可能不是人写的代码而是另一个 Agent 在运行时通过语义匹配找到的。我引入了一个简单的兼容性校验体系。每个能力注册信息里带schema_versionReach 层维护每个能力最新 schema 与旧 schema 的映射关系。当请求携带的 payload 校验失败时Reach 层不是直接报错而是尝试按兼容规则做字段转换比如旧字段order_no映射到新字段order_id。如果找不到兼容规则才返回明确的 schema 校验失败错误。同时对每个 Agent 的上游调用方做影响面分析。因为所有消息都经过 Reach 层Reach 层天然知道谁在调用谁。当某个 Agent 要下线某个能力时可以先查询哪些调用方依赖它然后在下线前主动通知所有调用方而不是等上线后逐个炸过去。5.2 流式响应与半双工通信第二个扩展是流式响应。大模型 Agent 运行时经常需要一边生成一边输出而不是等全部生成完再一次性返回。一开始我们用的是简单的 HTTP 长轮询但效果不理想。后来在 Reach 层增加了stream模式的支持消息信封里reply_to可以标记为stream://Reach 层把该条消息的路由结果与一个 WebSocket/SSE 通道绑定。实现思路不复杂当请求进入 Reach 层时如果识别到stream模式就同时建立一个通道 ID也就是 message_id所有被调用 Agent 产出的增量结果都作为流式事件写入 Redis 的 channelReach 层再把 channel 里的内容转发到调用方建立的 SSE 连接上。这里的关键点是流式消息的序列问题——一个 Agent 可能在处理中输出多段内容必须确保message_idsequence_no的组合是唯一的否则前端拿到乱序的流式结果会导致生成文本错乱。5.3 可观测性从调用链到成本账本第三个扩展是深度的可观测性。项目运行久了之后我发现路由成功和用户满意是两回事。于是我在消息信封的基础上增加了一套指标采集每个trace_id贯穿整个调用链记录每次路由的耗时、目标 Agent、重试次数、token 消耗、返回码。这些指标最终汇入一个时间序列数据库用来形成三个维度的视图第一个维度是健康视图每个 Agent 请求量、成功率、P99 延迟这些是日常运维最依赖的数据。第二个维度是链路视图一次用户提问到底触发了多少个 Agent 的协作哪个环节最慢哪个环节经常失败——这种跨 Agent 的链路信息不通过 Reach 层根本拿不到。第三个维度是成本视图大模型 Agent 的 token 消耗最终都会体现为账单Reach 层按能力标识和调用方维度汇总 token 消耗方便业务团队做成本分摊和优化。从我个人的使用体验来说可观测性建设最容易被低估但它其实是 Agent-Reach 长期运营最值钱的部分。没有它你只知道自己有多少 Agent永远不知道自己调度得有多差。我做 Agent-Reach 这段时间最大的体会是多 Agent 协作的瓶颈从来不是某个 Agent 的智能程度而是连接的质量。当你把触达层做扎实了上层的 Agent 才能真的协作起来。如果你正在构建自己的 Agent 网络希望这篇梳理能帮你少走一些我走过的弯路。