搭建多智能体协作层时我为什么选择了 Agent-Reach路由、注册与调度的完整落地记录现在做 AI 应用基本绕不开“智能体”这三个字。但很多团队在从单 Agent 走向多 Agent 协作时会突然发现一个很尴尬的现状每个 Agent 都像一座孤岛能力很强却互相够不着。你要让一个搜索 Agent 把结果交给一个分析 Agent中间要么靠人肉搬运 JSON要么把代码写死成面条式调用改一个接口恨不得重启全链路。Agent-Reach 就是为了解决这个问题出现的。你可以把它理解成 Agent 之间的通信总线或者说“电话交换机”——所有智能体都注册上来通过它来发现彼此、传递消息、编排任务。它解决的痛点是让多个 Agent 能像微服务一样被统一注册、调度和触达。适合正在做复杂 Agent 应用、多角色协作系统或者是想把现有单 Agent 工作流升级成可扩展架构的开发者参考。全文不涉及任何外部网络依赖全部基于本地环境的工程实践。1. 项目背景为什么多 Agent 协作需要一张“通信网”1.1 “Agent 孤岛”是怎么出现的先还原一个非常典型的开发场景。你可能已经有一个能查天气的 Agent还有一个能做数学计算的 Agent甚至有一个能调用内部 API 查询库存的 Agent。单独跑每个都很好。但一旦要求它们协同完成一个复合任务比如“查一下天气再根据天气推算明天出行需要的交通时间”问题就来了第一个 Agent 的结果怎么传给第二个 Agent如果未来再加入第三个、第四个呢Agent 数量一多调用关系就像一堆乱麻A 调 BB 调 CC 又要调 A代码里全是硬编码的 URL 和 API Key。我在做这个项目之前团队内部就已经被这种“手牵手”式的调用折磨过。于是干脆把问题抽象了一下其实多 Agent 协作需要的是一个中间层它干三件事——让 Agent 能“报名字”注册自己让调用方“按名字找人”以及把一次复杂的任务编排成一条可追踪的链路。Agent-Reach 就是按这个思路设计的。1.2 Agent-Reach 在设计上到底做了什么Agent-Reach 并不是一个 MRP 或者业务流程引擎那种重家伙它的核心设计哲学是“轻路由、强协议”。也就是说它不关心你的 Agent 内部用什么模型、什么框架只要求你实现一个统一的接入协议。任何一个 Agent只要按照协议声明自己的能力、接收消息的地址、以及返回结果的格式就可以被全局发现和调用。这个思路其实借用了后端领域非常成熟的注册中心模式类似于你手机通讯录里的联系人——你不需要记得每个人的电话号码只需要知道名字就能拨通。项目启动后整体实现大概分成了四块注册中心、消息路由、任务编排和日志追踪。前两个解决“找到人”的问题后两个解决“把事情办完”的问题。后面的章节我会逐一拆开来讲我具体怎么实现的每一步踩过什么坑也会详细说。1.3 这套方案的适用边界先泼一盆冷水Agent-Reach 并不是所有场景下的万能药。如果你的 Agent 只有一两个代码里 if-else 直接写死互相调用完全够用。但如果你要面对的是一群数量在五个以上、各自独立演进、甚至未来还要由不同团队维护的 Agent那中间层就是刚需了。适用场景概括起来有三类多个 Agent 由不同团队开发需要统一接入和调用入口。同一个任务要跨多个 Agent 编排执行且需要完整的链路追踪。Agent 有可能增加、下线或迁移地址不能容忍调用方跟着频繁改动。我实际的体验是引入 Agent-Reach 之后新增一个 Agent 的成本从“改一堆调用方”降到了“注册一个名字”这个体验是非常直观的。2. Agent-Reach 整体架构与核心设计拆解2.1 注册中心让每个 Agent 都能“报个到”注册中心是整个 Agent-Reach 的基石。我用了一张表来存储所有 Agent 的元信息字段包括Agent 的唯一标识agent_id、能力描述capabilities、访问地址endpoint、认证令牌api_key、健康状态和最后心跳时间。每个 Agent 启动时调用/register接口把自己塞进这张表然后每隔一段时间发一个心跳包。这里有一个设计细节值得展开我特意把“能力描述”做成了半结构化的标签列表而不是一段自然语言。例如“weather.query”“calculator.advanced”这样路由的时候做匹配效率高也方便未来做能力矩阵的聚合展示。注册中心本身我用 Python FastAPI 写的存储层用一个轻量的 SQLite生产环境可以平替成 MySQL 或者 Redis。先贴出核心的数据模型定义# registry/models.py from pydantic import BaseModel from typing import List class AgentInfo(BaseModel): agent_id: str name: str capabilities: List[str] endpoint: str api_key: str status: str active这个类是理解 Agent-Reach 的第一个入口。agent_id 是全局唯一标识endpoint 是其他 Agent 调它的地址capabilities 面向路由匹配。我建议 agent_id 的命名尽量带上部门或者业务前缀比如ops.weather.v1排查问题的时候一眼能看出是谁。2.2 消息路由让请求精确找到对应的 Agent有了注册表下一步就是路由。路由模块存在的意义是当一个外部请求进来它自带一个意图比如“查一下上海的天气然后算一下明后天温差”路由模块必须负责把这个意图翻译成一个具体的 Agent 调用链。在 Agent-Reach 里我实现了一个“基于能力匹配 上下文传递”的路由器。请求进来时先按 capabilities 命中第一个 Agent拿到第一个结果之后再结合既定流程传给下一个 Agent。整个过程是配置驱动的流程定义长这样{ workflow: weather_and_analysis, steps: [ {agent: ops.weather.v1, input_mapping: {city: $request.city}}, {agent: analysis.calculator.v1, input_mapping: {data: $steps[0].output}} ] }这个 JSON 是一个“最小可用”的任务编排描述。每个步骤声明用哪个 Agent以及输入数据从哪来。$steps[0].output这种写法类似于模板引擎里的引用变量让数据在 Agent 之间流动时可以明确溯源。我就是靠这个方式把原先硬编码的调用链变成了可视化、可修改的配置。2.3 任务编排从单次调用到跨 Agent 的流程任务编排是 Agent-Reach 里稍微复杂的一块。它不是一个 UI 拖拉拽的编排器而是一个执行引擎。核心接口只做三件事接收一个 workflow 定义按步骤串行或并行地调用 Agent最后汇总结果。实现上我并没有去引入像 Celery 那样的任务队列——目前这个量级还用不上。Agent-Reach 采用的是轻量异步执行基于 asyncio 的gather做并行调用配合一个内存中的任务状态表来追踪执行进度。如果某一个步骤失败默认策略是快速失败返回错误步骤号和上下文你也可以配置“失败后继续”的策略用于不关键的可降级步骤。这里我贴一段执行引擎的简化代码# core/executor.py import asyncio from typing import Dict, Any async def run_workflow(workflow: Dict[str, Any]): steps workflow[steps] results {} for idx, step in enumerate(steps): agent_id step[agent] input_data resolve_input(step.get(input_mapping, {}), results) result await call_agent(agent_id, input_data) results[idx] result if not result.get(success) and step.get(critical, True): return {workflow: workflow[id], failed_step: idx, results: results} return {workflow: workflow[id], status: done, results: results}这段代码虽然简单却是整个 Agent-Reach 最核心的执行骨架。我刻意没有引入外部任务框架就是想让项目保持“拿来即懂”的体量。你要是把这段改成用消息队列驱动也只是替换掉call_agent这一层的实现逻辑而已。3. 实操从零实现一个 Agent-Reach 实例3.1 准备基础环境与依赖动手之前先把环境理清楚。我这边是本机 Ubuntu 22.04Python 3.10 版本。你完全可以在 macOS 甚至 Windows 上跑因为依赖很少。核心就三个库FastAPI 用于提供注册和路由接口uvicorn 作为服务器httpx 做 Agent 之间的 HTTP 调用。如果有要求再加 pydantic 做数据校验不过 FastAPI 本身已经携带了。安装命令就一行pip install fastapi uvicorn httpx建议提前建一个虚拟环境免得污染系统 Python。我自己管理项目依赖有个习惯无论项目多小都单独建 venv。这个习惯在 Agent-Reach 的多 Agent 联调阶段救了我很多次——有个 Agent 依赖的是旧版 pydantic如果全都装到同一个环境里那基本是没法跑的。3.2 动手实现一个最小注册中心注册中心的实现说穿了就是两张路由表。第一张是/register接收 Agent 上报的信息并保存第二张是/heartbeat接收心跳。为了让后面的 Agent 能够发现彼此我额外加了一个/lookup接口按 agent_id 返回地址信息。关键代码在这里# registry/server.py from fastapi import FastAPI, HTTPException import uvicorn app FastAPI() agents {} app.post(/register) async def register(info: AgentInfo): agents[info.agent_id] info return {status: ok, agent_id: info.agent_id} app.post(/heartbeat) async def heartbeat(agent_id: str): if agent_id not in agents: raise HTTPException(status_code404, detailagent not found) agents[agent_id].status active return {status: ok} app.get(/lookup/{agent_id}) async def lookup(agent_id: str): agent agents.get(agent_id) if not agent: raise HTTPException(status_code404, detailagent not found) return agent if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)这段代码看着简单但它是整个 Agent-Reach 的基础。注册中心一旦活不起来后面所有 Agent 互相“打电话”就全没戏。我建议你写完这段代码后先启动服务然后手动用 curl 试一下/register和/lookup确认基本链路通了再往下走。3.3 如何让一个 Agent 接入 Agent-Reach先拿一个自带的“天气 Agent”来做示例。它内部逻辑很简单接收城市名返回一份模拟天气数据。重要的是怎么接入 Agent-Reach 的协议。每个 Agent 需要做三件事启动时向注册中心报到暴露一个 HTTP 接口供其他 Agent 调用周期性地发送心跳。我给天气 Agent 写了一个独立的 FastAPI 实例监听在 8001 端口它的注册代码如下# agents/weather_agent.py import httpx, uvicorn from fastapi import FastAPI app FastAPI() app.post(/invoke) async def invoke(req: dict): city req.get(city, ) # 这里是模拟的天气数据 return {success: True, city: city, temperature: 26} if __name__ __main__: with httpx.Client() as client: client.post(http://localhost:8000/register, json{ agent_id: ops.weather.v1, name: Weather Agent, capabilities: [weather.query], endpoint: http://localhost:8001/invoke, }) uvicorn.run(app, host0.0.0.0, port8001)这里有个小但关键的细节Agent 启动时注册用的是 httpx 同步客户端在 uvicorn.run 之前完成注册。这样做的好处是Agent 在开始对外提供服务之前就已经在注册中心报到过了不会被调用方短暂地“发现不到”。如果你把注册代码放到服务启动之后再去做就会有一段窗口期别人的请求过来会发现查无此 Agent。3.4 让两个 Agent 通过 Agent-Reach 互相触达一个 Agent 不算协作我们再加一个计算 Agent负责处理温差。它注册到 Agent-Reach 时capabilities 写的是calculator.advanced地址在 8002 端口。现在问题来了天气 Agent 算出了今天的温度怎么让计算 Agent 拿到这个温度再去算温差在 Agent-Reach 的架构里这两个 Agent 不需要知道对方的存在。外部请求只需要发给路由模块路由模块按照 workflow 配置先调 8001 拿到温度再把温度当作参数调 8002。整个过程相当于一次“投递”外部请求是信封workflow 是快递单路由模块是快递员。我在本地实测时使用的调用结果是天气 Agent 返回温度 26 度计算 Agent 自动把它减去一个基准值返回差值。整个链路在日志里非常清楚先显示dispatch - ops.weather.v1然后显示dispatch - analysis.calculator.v1哪个环节出了问题一眼就能定位。这也是 Agent-Reach 相比硬编码调用“分布式链路清晰”的优势。4. 常见问题与排查技巧实录4.1 Agent 注册成功但调用时 404这是个人第一次接触 Agent-Reach 时会踩得最多的坑。现象是注册接口返回成功但调用时路由模块报 404。排查后发现80% 的情况是 Agent 的 endpoint 写错了地址——尤其常见于把127.0.0.1和localhost混用。注册时写的是localhost但 Agent 服务实际监听在某个局域网 IP 上导致路由模块无法访问。解决思路很简单统一约定 endpoint 的写法并且拿到注册中心之后立刻用curl命令做一遍连通性验证。不要同时混用容器网络和宿主机网络如果 Agent 跑在 Docker 里endpoint 就要填容器网络的地址而不是容器的内部地址。这个经验在第 3 章那段注册代码里体现不出来但实际部署时 90% 的问题都出在这里。4.2 Agent 之间出现循环调用多 Agent 协作中最隐蔽的问题就是循环调用。A 请求 BB 在处理过程中又去请求 A如果没有任何保护机制两个 Agent 就会互相发消息直到资源耗尽。Agent-Reach 里我加了一个调用深度计数每次进入路由模块时给请求加一个头部标记比如x-agent-reach-depth: 1每往下一层深度加一超过设定的最大深度我这里设的是 5就直接拒绝然后返回“调用链路过深”的错误。这个检查和防重用的思路是一样的。因为 Agent 编排的调用链在多数情况下是有限的 DAG绝对不该出现深度无限增长的情况。如果你在实战中发现了深度超限的报错首先要做的不是调大阈值而是去检查 workflow 定义是不是出现了环形依赖。4.3 认证和可控性如何取舍Agent-Reach 目前内置了简单的 api_key 校验。每个 Agent 在注册时可以带上一个 token路由模块在调用它的时候会在 HTTP 头部带上这个 tokenAgent 收到请求后校验一下。团队里做联调的时候这个机制能有效地拦住很多误调用。但注意这只是基础认证真正上生产之前建议在网关层做更完善的鉴权比如引入 OAuth2 或者 mTLS。我把这个取舍写在这里是希望你清楚Agent-Reach 的设计重点是消息路由和编排而不是安全边界。安全是一个完整体系不要把一个小中间件当成全功能安全网关。如果你部署的环境对安全要求很高建议结合公司的统一网关、审计日志和网络策略一起用。4.4 性能瓶颈与优化实录用 Agent-Reach 做联调的过程中我压过一次简单的并发测试。场景是 100 个并发请求同时调用一条包含两个 Agent 的 workflow。结果发现注册中心虽然是异步的但 SQLite 的写入锁开始成为瓶颈表现为偶尔的超时、日志里有 database is locked 的报错。我当时做了两个优化。第一注册中心的写操作从同步改为异步并加内存缓冲心跳更新先在内存里做定期批量刷到 SQLite第二日常查询走内存副本避免高频读操作打到底层数据库。优化之后压测结果提升明显数据库锁报错完全消失。这是所有带持久化状态的中间件都会遇到的问题你一旦在 Agent-Reach 上做生产部署内存缓冲几乎是必选项。4.5 调试一个小技巧结构化日志Agent-Reach 的日志我统一走 JSON 格式每条日志带上请求 ID、Agent ID、步骤号。这个方法在排查问题时效率极高——因为跨 Agent 的调用链如果只是普通的文本日志你根本没法把一串关联的操作串起来。有了请求 ID 之后你可以用一条命令过滤整条链路的日志grep req-20240601-abc123 agent_reach.log这个习惯我从写第一行代码就保持了实际排障时帮了很大的忙。很多项目问题难查不是代码逻辑复杂而是日志根本看不懂因果链条。5. 工具选型解析与后续扩展思考5.1 为什么用 FastAPI 而不是 Spring 或 Node.js很多做后端的朋友看到我用 Python FastAPI 会问为什么不用 Spring Cloud 那一套。其实道理很简单Agent-Reach 的核心用户是 AI 应用开发者团队里最熟悉的语言大概率是 Python而且 Agent 生态里大部分组件LangChain、LlamaIndex 等都对 Python 友好选 FastAPI 可以把整个技术栈的认知成本降到最低。FastAPI 的异步性能也足够支撑中等规模的 Agent 调度。一次简单的消息路由耗时基本在几十毫秒级别除非你到了每天千万级请求的体量否则 FastAPI 完全够用。技术选型的一个经验是选择团队维护成本最低的方案而不是性能最强的方案。5.2 从 Agent-Reach 可以扩展出的三个方向Agent-Reach 目前是一个“可用”的中间件但距离真正的生产级编排平台还有距离。我自己梳理了三个可行的扩展方向这里列出来仅供参考。第一个方向是增加“人工审核”节点。某些流程涉及关键决策时不应该完全交给 Agent 自动执行。你可以在 workflow 里增加一个human_approval: true的步骤Agent-Reach 执行到这个步骤时挂起通过 webhook 通知人去审批审批通过后自动继续。这在自动化办公、自动化运维场景里非常实用。第二个方向是引入“多版本灰度”。每个 Agent 可能都有新老版本在交接期你想让一部分流量走新版本一部分继续走旧版本。现在 Agent-Reach 的注册表是按 agent_id 唯一识别的我建议未来可以改成支持agent_id加version的双字段并在路由层做按比例分发。第三个方向是提供“能力地图”的可视化。现在只能在代码里看注册表数据不直观。如果做一个简单的管理面板把所有 Agent 的关系、调用频次、错误率展示出来对线上运维会有很大价值。我自己下一步就打算给 Agent-Reach 加这个功能。5.3 Agent-Reach 在团队协作中的真实感受最后聊点软性的东西。这个项目做完后我最大的感受是多 Agent 项目真正的复杂度不在单个 Agent 的智商而在它们之间怎么配合。Agent-Reach 不是那种能让你“哇”一下的炫酷技术但它能持续帮你省下那些无休止的接口沟通、联调和排障时间。当一个新的 Agent 加入到系统里团队只需要让它报个到不需要去戳其他组的代码这种体验是可以直接转化为交付速度的。如果你也在做 Agent 相关的应用我的建议是尽早把路由、注册、编排这层抽象建起来。不用做得像我的这个项目那么完整哪怕只是先把你两个 Agent 之间的调用改成“通过一个名字找到对方”后面扩展新 Agent 时你就会发现收益远超预期。