前一阵我在调一个多Agent系统差点被气得想把它拆成单线程。问题不在于单个Agent的能力而在于这些Agent之间要怎么找到对方、怎么把任务递过去、怎么知道对方到底做没做完。我一度在代码里写满了一个Agent直接调用另一个AgentHTTP接口的请求改一个Agent的地址就要顺手改三个调用方。正是那段时间的反复折磨让我下决心把“Agent要让别的Agent能找到、能对话、能协同”这件事单独拎出来做了一个叫Agent-Reach的轻量连接层。先说结论Agent-Reach不是什么玄学它就是一个让AI Agent之间能够互相注册、互相发现、互相投递消息的极简协议加网关服务。它解决的痛点是“Agent孤岛化”——每个Agent单独跑得挺好但想让它俩协作就得靠开发人员硬写胶水代码。如果你也在做多Agent编排或者正在对比LangGraph这类重型框架和自研方案的边界这篇内容会比较对胃口。我会讲清楚Agent-Reach的设计思路、一个周末能做出来的最小原型以及跑起来之后真正会踩到的几个坑。1. 孤岛式Agent协作为什么这么痛苦1.1 三种“低成本”做法和它们的隐性成本市面上解决Agent协作最常见的不是自研框架而是抄近路。第一种是硬编码互相调用Agent A直接请求Agent B的接口请求里固定带上字段名。好处是直白五分钟就能跑通坏处是耦合度极高。我见过一个项目里上游Agent改了一次出参命名下游三个Agent全部静默失败最后日志里看起来像模型能力下降实际是字符串拼接错误。第二种是引入消息队列比如Kafka或RabbitMQ。消息队列适合流水线式的异步任务但对交互型Agent来说太重了你要处理分区、偏移量、重试策略还要设计一套消息格式。很多场景只是想让“数据分析Agent”把结论交给“可视化Agent”中间塞一条Kafka属于杀鸡用牛刀运维成本反而超过了协作本身。第三种是干脆全家桶式地上一个Agent编排框架比如LangGraph。框架能解决编排问题但代价是你得接受框架对Agent生命周期、状态管理、节点关系的整套假设。有些项目的诉求仅仅是有两个Agent能说上话结果被框架的图论、状态机和持久化绑得死死的。不是说框架不好而是很多人还没想清楚自己到底需要什么程度的管理能力就先把控制权交出去了。这三条路让我意识到一个共同缺陷大家把Agent之间的通讯方式当成了业务代码的一部分而不是基础设施。业务代码会频繁变动而通讯基础设施应该长期稳定。1.2 Agent-Reach想把问题解决到哪一步受之前做微服务时的服务注册与发现机制启发我开始在脑子里画一个重要的问题边界。一个Agent要和另一个Agent协作本质上要回答三个问题我是谁、别人在哪、话怎么递过去。我是谁每个Agent需要一个稳定且全网唯一的标识以及一份描述自己能力的信息。别人在哪Agent需要一个查询入口按能力找到合适的其他Agent。话怎么递过去需要一个可靠的路由机制把任务消息从发起方送到目标方并返回结果。Agent-Reach的核心定位就是把这三件事封装成一个极薄的协议和网关层。它刻意不做Agent内部的状态编排不管Agent的推理过程也不尝试定义Agent该用什么模型。你可以把Agent-Reach想象成一本通讯录加一个总机接线员Agent启动时在总机登记自己的号码和擅长领域需要协作时向总机查询接通后总机负责把消息转过去并且把对方回的话带回来。这个词——Reach强调的就是“能触达”。你的Agent再聪明它无法被别的东西找到那就没有协同的价值。2. 核心设计给Agent定义一套通用“握手”协议2.1 用Actor模型的思路理解Agent-Reach在设计Agent-Reach的通信模型时我一直绕不开Actor模型。这个在并发编程里很成熟的概念套到Agent协作上意外地协调。Actor模型的核心就几句话每个Actor拥有唯一的地址外部只能通过给这个地址发消息来跟它交互Actor内部状态是封装的没人能直接操作Actor收到消息后决定怎么响应。把这里的Actor换成AI Agent你会发现AI Agent本身天然就适合这套模型——一个Agent的内部状态、上下文窗口、正在跑的推理逻辑本就不应该被其他Agent直接读取。但Agent-Reach对传统Actor模型做了简化。我们不需要Actor框架里那么严格的消息邮箱和调度器我保留了三个元素稳定地址、消息传递、能力描述。地址告诉别人去哪找你消息传递解决怎么把任务给你能力描述解决有人要找“能做数据分析的”时怎么筛到你。这个简化很重要。如果你照搬完整的Actor框架你又要落入重武器的陷阱。Agent协作场景的常态是约几个Agent开个会不是每秒几十万条消息的流式处理。2.2 能力声明manifestAgent版的自我介绍Agent-Reach的通信协议中最核心的数据结构是manifest——Agent的能力声明。它解决的是“这个Agent能干什么”的问题。假设我有一个数据分析Agent它的manifest可能长这样{ schema_version: 1, agent: { id: agent-data-analyst-01, name: data-analyst, endpoint: http://10.0.0.22:8901/msg }, capabilities: [ { name: summarize_dataset, description: 接收CSV数据集路径返回统计数据与摘要结论, input: {type: object, fields: [csv_path, columns]}, output: {type: json, keys: [summary, stats]}, cost_hint: latency_high } ], ttl: 300 }这里有个关键的选型思路为什么不能只靠Agent名字路由因为在真实的协作场景里发起方往往不知道目标Agent叫什么但它知道自己想要什么能力。我在原先的Agent A可能只知道“我需要一个报表生成器”却不清楚那台机器上跑的Agent叫report-gen还是gen-report。有了能力声明路由就变成了一场“按能力查询”而不是“指定名字喊话”。2.3 注册、发现、转发三个原语协议里我定义了三个核心原语register、discover、route。它们分别对应“签到”“查号簿”“通话”这三个动作。register是Agent启动后向Agent-Reach网关上报自己的manifest并约定一个心跳续租周期。TTL设成300秒意味着Agent每5分钟至少要和网关“击一次掌”否则网关会认为它掉线了。这个设计是从分布式系统里借鉴的租约机制能有效避免Agent异常退出后其他人还在朝一个已经不存在的地址发消息。discover是发起方按能力描述去网关查询可用Agent。我最初的实现很简单用包含关系匹配能力名比如查询“summarize”能匹配到能力名为summarize_dataset的Agent。实际过程中发现这不够后来加了可选参数形式的模糊匹配和连通性权重。route是真正的消息投递。发起方把一个包裹结构上类似带目标地址的JSON交给网关网关找到目标Agent的回调地址endpoint把包裹原文POST过去然后把结果带回给发起方。有意思的是route这个动作在设计时要区分两种模式同步和异步。同步模式适合想要直接拿结果的场景比如“帮我算一下这个Excel的总数”调用方可以阻塞等待结果返回。异步模式适合长耗时任务网关收到消息后立即返回一个task_id之后Agent通过轮询或回调接收结果。两种模式我在原型里都做了实践中发现同步模式定义了超时上限用起来才不坑。3. 周末原型从零搭一个能跑的Agent-Reach3.1 为什么网关选FastAPI、节点用标准库关于技术选型我不想故弄玄虚。网关是整个Agent-Reach里唯一需要同时处理大量并发连接和路由逻辑的地方所以网关用FastAPI最顺手。它是异步的、自带OpenAPI文档、类型提示支持好做注册、发现、转发这三个接口非常自然。Agent节点侧则故意做轻。节点只需要能力来发HTTP请求和接收请求Python标准库里的requests和http.server就能搞定。这样做有个实际好处Agent的宿主环境往往不是开发者的机器可能是客户的服务器、一台自带环境的容器依赖越少接入门槛越低。我这里给一份当时项目的选型对比表方便你做判断组件技术选择理由网关服务FastAPI Uvicorn异步并发能力强类型安全自带API文档节点通讯标准库HTTP客户端/服务端零额外依赖Agent侧接入成本最低持久化内存字典 JSON文件备份原型期不引入数据库简化心智负担Agent注册协议JSON-RPC风格简易封装语义清晰方便扩展请求ID与错误码当时我刻意不上Redis、不上MySQL、不上消息队列三个接口加两个Agent节点两个半天就让它转了起来。3.2 网关端注册、发现和转发的核心代码网关最核心的注册和转发代码核心逻辑浓缩后大致长这样# gateway/main.py import time, uuid from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() # 用于存储Agent注册信息的字典 agents {} class AgentRegistration(BaseModel): name: str endpoint: str capabilities: list[str] ttl: int 300 app.post(/register) async def register(payload: AgentRegistration): agent_id uuid.uuid4().hex[:16] agents[agent_id] { id: agent_id, name: payload.name, endpoint: payload.endpoint, capabilities: payload.capabilities, expires_at: time.time() payload.ttl, } return {agent_id: agent_id, expires_in: payload.ttl}转发逻辑更直接根据能力名在agents字典里找到目标然后利用httpx异步发起请求import httpx class RouteRequest(BaseModel): target_agent_id: str payload: dict app.post(/route) async def route(req: RouteRequest): if req.target_agent_id not in agents: raise HTTPException(status_code404, detailagent not found) target agents[req.target_agent_id] # 检查目标Agent是否过期 if target[expires_at] time.time(): raise HTTPException(status_code410, detailagent expired) async with httpx.AsyncClient() as client: resp await client.post(target[endpoint], jsonreq.payload) return {status: delivered, reply: resp.json()}这段代码看起来土得掉渣但它确实跑通了。我当时测了两个Agent一个负责算数一个负责把结果拼成自然语言路由毫无悬念地工作。重要的是这版代码里已经埋进去了过期检查、Agent不存在和过期这两个区分开来的错误语义这个细节后面的章节里会展开讲。3.3 两个不同“血统”的Agent成功握手周末原型的最高光时刻是让两个完全不同的Agent框架下的Agent完成对话。一个基于LangChain写的数据分析AgentAgent本体住在Python进程里跑的是ReAct模式的prompt推理另一个是用来生成可视化报表的Agent用纯Flask写成没有借助任何Agent框架只暴露了一个接收JSON的HTTP端点。数据分析Agent在内部生成结论之后决定需要生成图表于是它调用了一个我封装的AgentReachClient向网关发起discover按“visualize_data”查到了报表Agent。接着构造一个route请求把数据摘要和图表配置传给报表Agent。报表Agent拿到数据后没有做任何LLM推理直接用matplotlib生成了图表然后把图片路径以JSON形式返回。从数据分析到图表输出全程没有一行代码硬编码指向报表Agent的地址。这次握手的意义在于打通了框架异构的边界。业界做Agent互联时经常陷入同一个框架内讨论问题而Agent-Reach把互操作层建立在HTTP和JSON这种最基本的协议之上所以它天然能连接任何能发HTTP请求的程序——不管它是不是一个真正的Agent框架。4. 实际跑起来后踩过的四个坑4.1 超时语义三种失败状态不能合并原型跑通之后痛苦才真正开始。第一个让我折腾半天的问题是超时语义混淆。代码如下面这样我最初判断调用结果的逻辑是try: result await route(...) except Exception as e: log.error(调用失败)这个写法害我排查了很久的灵异问题。网关抛出的404 Exception目标Agent不存在与连接超时异常被合并成了同一类日志。于是一个Agent只是临时重启另一边的日志居然显示“Agent不存在”。更严重的是一个耗时较长的任务在我没设超时的情况下无限阻塞直到调用方的HTTP客户端先崩溃才抛出一个不明确的错误。后来我把失败语义拆成了三层并且写进了协议文档第一层网络可达性失败。网关找不到目标、目标地址拒绝连接、握手超时都归为“Agent不可达”建议直接提示用户稍后重试。第二层处理超时。目标在线且收到了消息但在约定时间内没返回结果。这类失败和第一层的处理策略完全不同不能盲目重试应该询问目标Agent是否已经接手任务避免重复执行。第三层内部错误。目标返回了4xx或5xx尤其是带错误码的Response。这类信息要原样回传给调用方因为调用方Agent会把错误信息用来指导下一步行动错误信息一旦被错误归类就会污染Agent的判断。我现在所有agent的接口统一约定同步调用必须设置超时上限长任务一律切异步模式网关自己要报出明确的错误类型我用了error.reachable、error.timeout、error.internal三个错误码消息结构里永远带上type字段。4.2 消息的幂等与重启窗口别把消息弄丢第二个坑来自Agent重启。有一次我正在升级一个Agent节点从旧的容器切到新的容器切换期间网关上的manifest还没过期新的节点已经启动但还没注册。此时恰有另一个Agent向它route消息网关发现manifest还存着就继续转发结果旧容器已经端口关闭消息直接丢失。我最初的处理方案是让网关在失败后连续重试三次每次等待2、4、8秒等差退避。这样做能缓解容器切换窗口的问题但治标不治本。更彻底的方案是在消息结构中加入幂等键request_id并约定接收方在收到消息后如果发现request_id已经处理过直接返回缓存结果。{ request_id: req-2f8e61a0, sender: agent-data-analyst-01, target: agent-report-generator-02, payload: { task: visualize_data, data_ref: jobs/20240315/summary.csv, options: {chart_type: bar} } }关于消息可靠性的设计原则会因场景而异。如果只是在内部集群里传递小任务那“即发即忘”再加上失败时重试完全够用。但如果是Agent与客户系统对接或任务本身价值高、重跑成本高那我建议至少要做到at-least-once语义网关务必等待目标Agent确认收到消息之后再向调用方返回已投递状态接收方则需要按request_id做去重。这个约定虽小却能避免最让人崩溃的重复扣款或重复发邮件这类问题。4.3 循环调用A找B、B又找回A的可怕死结第三个坑比较隐蔽不容易在测试时暴露。当Agent-Reach网络中Agent数量增多后路由路径会形成环。A发现BB在执行过程中发现需要A的能力于是B又调AA再调B直到双方把上下文窗口塞满后超时拥抱在一起。我在排查一个连环报错的故障时看到调用链A - B - C - A - B差点当场怀疑是自己在梦游。两天后终于确认那不是一个并发问题而是循环调用。解决方案并不复杂但必须提前设计。我给每个route请求附加了一个trace_id和一个计时器字段hops在每次转发时递增并约定默认的最大跳数。跳数上限我默认设置成8超过就直接终止并返回一个“loop detected”错误。同时上下文里携带调用链的依赖列表目标Agent收到消息后如果发现自己已经在这个链路上就直接拒收。class RoutedMessage(BaseModel): trace_id: str hops: int 0 visited_agents: list[str] [] payload: dict这样失效的循环调用会在几跳内被掐断而不是在Agent那里反复消耗昂贵的推理时间。这个坑提醒我Agent协作网络里的“死锁检测”不是一个可选项而是一个最基本的必须项。4.4 权限收口Agent能访问的远不止它声明的能力最后一个坑是关于权限的。最开始Agent-Reach的route接口是没有任何鉴权的——任何Agent发现目标之后都能直接给目标发消息。这话听起来挺自由实际一跑就会发现等于每个Agent都可以指使任何别的Agent干活跟公司里任何人都能直接花公款一样可怕。我的场景是一个内部视觉分析Agent和外部数据查询Agent协作视觉Agent本来只需要被“数据分析Agent”调用来做图表渲染。但因为没有权限限制任何一个被攻破的Agent或者一个逻辑有Bug的上游Agent都可能拿它去生成任意类型的图片浪费计算资源还是小事如果这个Agent被诱导连上了内部网络的其他接口那问题就大了。所以后来我引入了一个很朴素的访问控制表每个Agent在注册时声明自己允许被哪些调用方调用。网关在route时先检查发起方的agent_id是否在目标Agent的允许列表中。比如视觉Agent只被report-generator调用# 视觉Agent的manifest新增字段 allow_scope: { callers: [agent-report-generator-02], capabilities: [visualize_data] }这样做还有个附带好处调用方Agent的能力模型更干净了它知道自己哪些能力是面向外部开放的哪些只是内部服务。我在生产环境里建议至少把“按Agent身份控制”做到位再进阶一点可以加入“按能力维度的细粒度权限”两个维度一结合权限问题基本能兜住。5. 从原型到落地三种部署形态与框架适配5.1 小团队一个网关进程加SQLite如果你的团队规模不大Agent数量在十个以内最省心的部署形态就是单网关加SQLite。网关本身就是无状态服务只有manifest和消息流转记录需要存储SQLite在低并发场景下绰绰有余。我当时用Docker把网关和Agent编排在一个内网子网里网关只需要暴露一个注册端口给Agent使用docker run -d \ --name agent-reach-gateway \ -p 9501:8000 \ -v ./data:/app/data \ agent-reach-gateway:1.0在这种部署形态下要特别注意给SQLite做定时备份因为Agent注册信息虽然重新启动就能恢复但路由日志和去重表丢失后可能出现重复投递的隐患。另外单一网关意味着单点风险但对于实验性项目和小团队内部工具来说这点风险完全能接受。用这套方案跑了一年多最值得自豪的是基本零运维。5.2 中规模多网关加一致性哈希Agent数量增长后单网关会遇到两个瓶颈一是manifest的并发读写二是路由流量集中在单点出故障会全瘫。我后来的优化思路不复杂——部署多个网关用一致性哈希把每个agent_id哈希到固定的网关节点上。为什么要固定哈希而不是随机找一台因为同一个Agent的注册、路由、心跳记录都落在同一台网关上网关处理起来就不需要频繁同步状态。我最初试过无状态随机路由结果Agent的注册请求落在网关A心跳却打到网关B网关两边各持一半过期信息两个节点都认为对方的数据不对。人工排查起来很耗费精力。一致性哈希其实适用于任何有状态、需要水平扩展的分布式系统。多网关之间唯一要同步的数据是所有Agent的注册记录。这个可以用etcd或Redis也可以让每个网关节点的SQLite定期互相导入导出。同步频率不用太高秒级就够了因为Agent本身有TTL心跳兜底最坏情况下丢掉一次注册信息几秒钟后Agent下个心跳周期又会补上。5.3 跟既有Agent框架共存LangChain、LangGraph、自研框架很多人会问的一个现实问题我已经在用LangChain/LangGraph了Agent-Reach还有必要吗我的回答是它不是替代品而是它们的补充。LangChain解决的是单个Agent如何用工具推理的问题LangGraph解决的是内部节点状态如何编排的问题而Agent-Reach解决的是Agent之间如何发现和通信的问题。二者在分工上是互补的不冲突。接入方式也很简单。对LangChain来说你可以把Agent-Reach的路由能力封装成一个普通的工具ToolAgent在推理过程中如果发现自己需要另一个Agent的能力就调用这个工具完成一次路由结果会作为tool observation回到推理上下文里。这样LangChain自身的Agent推理逻辑一点不用改。对LangGraph来说你可以把Agent-Reach路由定义成一个自定义节点当图内一个节点需要外部Agent能力时就走这个节点。对于自研Agent框架完全可以把Agent-Reach直接当成通信层把业务逻辑全部留在Agent内部互操作协议只有HTTP和JSON未来想换框架、换模型都会容易一点。我自己则是把Agent-Reach定位成了一个基础设施层。团队内部新接入一个Agent时只需要写一个简单的manifest声明它的能力并在启动时调用一下register接口它就自动出现在整个Agent协作网络里了不需要改任何其他Agent的代码。对一个工程团队来说这种“新Agent自动被其他人发现”的体验确实是传统硬编码方式给不了的。最后再分享一个我最近在推进的想法给Agent-Reach加上“能力可用性打分”。目前网关只做静态能力匹配下一步我打算引入Agent的历史响应耗时、成功率、平均评分这几个因子让discover返回的Agent列表默认带上排序权重。我个人在这个项目里最大的收获是Agent协作复杂的问题往往不是靠更重的框架解决的而恰恰是靠更清晰的协议边界和更朴素的基础设施。这个结论希望你做类似架构时也能体会到。