前一阵我在调一个多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 能力声明(manifest):Agent版的自我介绍
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-Reach
3.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_code=404, detail="agent not found") target = agents[req.target_agent_id] # 检查目标Agent是否过期 if target["expires_at"] < time.time(): raise HTTPException(status_code=410, detail="agent expired") async with httpx.AsyncClient() as client: resp = await client.post(target["endpoint"], json=req.payload) return {"status": "delivered", "reply": resp.json()}这段代码看起来土得掉渣,但它确实跑通了。我当时测了两个Agent,一个负责算数,一个负责把结果拼成自然语言,路由毫无悬念地工作。重要的是,这版代码里已经埋进去了过期检查、Agent不存在和过期这两个区分开来的错误语义,这个细节后面的章节里会展开讲。
3.3 两个不同“血统”的Agent成功握手
周末原型的最高光时刻,是让两个完全不同的Agent框架下的Agent完成对话。一个基于LangChain写的数据分析Agent,Agent本体住在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发现B,B在执行过程中发现需要A的能力,于是B又调A,A再调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的路由能力封装成一个普通的工具(Tool),Agent在推理过程中如果发现自己需要另一个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协作复杂的问题,往往不是靠更重的框架解决的,而恰恰是靠更清晰的协议边界和更朴素的基础设施。这个结论,希望你做类似架构时也能体会到。