做AI Agent项目快两年我越来越确信一件事单个Agent的能力天花板远不如一群Agent协作时的爆发力。但“协作”这两个字说起来轻巧真正落地时最先撞上的墙往往不是模型能力不够而是Agent和Agent之间怎么找到对方、怎么建立信任、怎么把一次调用安全地送过去。我去年下半年动手做了一个叫Agent-Reach的轻量级方案专门解决“智能体触达”这个问题。简单说它是一套让AI Agent能够互相发现、注册、路由和通信的连接层。这篇文章就完整记录一下这个项目的来龙去脉、核心设计、动手实现过程以及我在踩坑之后整理出来的排错经验。如果你也在做多Agent系统、AI工作流编排或者想给自己的Agent加上“呼叫外部能力”的通道这篇应该能给你省不少时间。1. 项目背景与核心思路拆解1.1 为什么Agent需要“触达能力”先想一个问题你现在写的Agent本质上是什么我的答案是它是一段围绕大模型组织起来的程序内部规划、调用工具、处理上下文最终完成一个目标。单机单线程的Agent很好写你自己定义工具列表模型自己决定调哪个跑通一个闭环就行。但一旦场景复杂起来问题就来了。比如你有一个数据分析Agent、一个报表生成Agent、一个消息推送Agent三个人各管一段。数据Agent算完结果需要报表Agent去出图最后推送Agent再把结果发出去。这时候你面临三个选择把所有能力塞进一个Agent里让它一个模型上下文里塞十几个工具结果就是指令越来越混乱、上下文越来越长、出错概率指数上升。在业务代码里硬编码调用关系Agent A跑完就调用服务B。这看似稳妥但Agent一旦多了调用关系就是一张蜘蛛网维护成本极高。给Agent一套“触达其他Agent和外部服务”的标准通道让它们自己发现自己、自己协商调用。这就是Agent-Reach在做的事。我选择第三条路还有一个更现实的理由你会遇到“异构”Agent。有人用LangChain写有人用CrewAI还有人干脆自己封装了OpenAI的API。一套连接层必须协议化、语言无关否则每接入一个新Agent都要重写一遍调用逻辑那就是给自己挖坑。1.2 Agent-Reach到底解决什么项目定位从一开始就很明确Agent Connection Layer。它不是调度器不是编排引擎更不是Agent运行时。它只负责三件事。第一件事注册与发现。每个Agent上线后要把自己的身份、能力、通信地址告诉“注册中心”同时可以从注册中心订阅到其他Agent的能力列表。这叫解决“能不能找到”的问题。第二件事能力路由。当调用方Agent想要某个能力时它不需要知道目标Agent的具体地址和协议细节只要告诉Agent-Reach“我要什么能力”由连接层去做能力匹配、目标选择、协议转换。这叫解决“找谁对接”的问题。第三件事可靠通信与权限控制。两个Agent之间传消息必须有统一的格式、超时重试、鉴权校验。你不能让A直接拿HTTP去怼B的接口那既脆弱也不安全。这叫解决“敢不敢调”的问题。这三个问题其实和微服务架构里的服务注册中心非常相似。但Agent场景有几个微服务没有的特殊要求能力描述的语义粒度更细、调用参数天然是不确定的自然语言、权限控制还需要考虑“模型会不会被骗着干坏事”这一层。所以不能直接套用Eureka或Nacos要做适配。1.3 技术选型背后的取舍老实说最初我也想过直接用现成的MCPModel Context Protocol打底。MCP把工具调用标准化了服务端暴露工具客户端调用工具协议很干净。但做着发现两个问题一是MCP更适合“一个客户端对多个工具服务端”的星型结构Agent和Agent之间互相对等调用的场景它支持得比较别扭二是MCP当前对“动态发现”的支持还不够灵活Agent需要的是“给我一个能干XX的Agent”而不是“给我这个URL上的这几个工具”。所以最终架构我选择了一条混合路线底层通信借用类MCP的JSON-RPC风格上层发现机制自己做了一个轻量注册中心。注册中心存储每个Agent的能力元数据提供带语义匹配的查询接口。Agent之间实际的业务数据交换走点对点通道注册中心不转发业务消息避免成为性能瓶颈。为什么不用消息队列我也试过用Redis Stream做传输层但发现一个问题MQ适合广播和削峰不适合“按能力精确路由到某个Agent实例”。而且很多Agent部署在边缘侧、内网里没有统一的消息中间件可用。所以最终决定传输层默认HTTP但做了Transport抽象理论上可以替换成gRPC或者私有长连接。技术选型这件事最重要的原则是“不为未来过度设计”。你只需要保证今天的协议清晰、明天能平滑替换就够了。Agent-Reach的很多设计看起来“原始”恰恰是因为它要做最小可用的连接层而不是一个重型中间件。2. Agent-Reach核心架构与关键细节2.1 整体运行逻辑整个系统跑起来之后你会看到三类角色。注册中心Reach Hub是所有Agent元数据的汇聚点它维护一张“能力注册表”。服务型Agent是能力提供方启动时向Hub注册自己长期保持在线等待调用。调用型Agent是能力消费方它向Hub发起查询、获取目标连接信息然后直连目标Agent。调用一次完整链路大概是这样的调用方Agent产生一个需求“我需要把一段文本总结成5个要点”。调用方Agent把需求发给Reach Hub。Reach Hub在注册表里做语义匹配找到能力描述为“text-summarization”的Agent返回目标ID和连接地址。调用方Agent拿到目标地址发起一次标准化的Reach RPC请求。目标Agent执行完成后返回结构化结果。注意这里第2步到第3步就是Agent-Reach的核心价值。为了让语义匹配可用每个Agent在注册时必须提供一个能力描述文件Capability Profile里面写清楚这个Agent能干什么、需要什么输入、返回什么格式、有哪些调用约束。2.2 能力描述把“能干什么”说清楚这是整个项目里最容易被低估的部分。我一开始想得简单觉得给每个Agent配一个description字段就行。结果第一次联调就发现两个Agent之间经常匹配不上原因是自然语言描述太模糊。后来我参考了OpenAPI和JSON Schema的思路设计了一套三层能力描述结构。第一层是身份层Agent的唯一ID、名称、版本、负责人。这一层保证我们能知道“谁”在上线。第二层是能力语义层每个能力有一个稳定的capability_id一段简短的人类可读描述还有一组关键词用于辅助匹配。打个比方text-summarization的capability_id是固定的、机器可读的描述是“对输入文本生成精简摘要”关键词里包括“总结、摘要、summarize、key points”。这样语义匹配就有三个抓手精确ID、自然语言描述、关键词扩召。第三层是调用契约层入参的JSON Schema、出参的JSON Schema、超时时间、幂等性说明。调用方Agent拿到这份契约后可以自己生成调用参数也可以把Schema塞给大模型让模型根据Schema构造JSON成功率会明显高很多。你可能觉得每个Agent手动维护这么一份描述文件太麻烦。确实。所以后面我加了一个工具能用Agent的system prompt自动生成一份初始版能力描述再人工微调。这样从“零”到“能用”的启动成本就低了很多。2.3 注册中心数据模型与匹配逻辑Reach Hub的存储我一开始图省事用了SQLite后来发现多节点部署不方便换成了PostgreSQL。数据模型其实特别简单核心就三张表agents表存Agent实例状态capabilities表存能力定义bindings表存Agent和能力的绑定关系。但真正费脑子的是匹配逻辑。你想想这个场景调用方问“有没有能做情感分析的Agent”结果注册表里存的是“对用户评论进行情绪识别”。这俩描述字面不一样语义却一致。如果只做关键词匹配直接就漏了。我的方案是分三层匹配按优先级来先做精确ID匹配。如果调用方直接指定capability_id那不需要任何花活直接命中。再做关键词召回。把能力描述里的关键词做倒排索引用调用请求里的关键词去召回候选集。最后做向量语义排序。把调和能力描述各自转成embedding算相似度在召回结果里排序。实际用下来前两层能解决80%的问题第三层是锦上添花。只靠向量匹配也不靠谱因为向量在“能力边界”这种语义上经常体现不出来倒排索引反而更快更准。三层配合之后准确率基本能稳定在90%以上。2.4 通信协议轻量RPC与状态码设计Agent之间的数据链路我实现了一个非常薄的RPC协议。请求结构长这样{ protocol: reach-rpc, version: 1.0, request_id: req_skjd893k, target_agent: agent-summ-01, capability: text-summarization, params: { text: 很长的一段原始文本..., max_points: 5 }, metadata: { auth_token: jwt..., trace_id: trace_xz09 } }响应结构也差不多包含status、result、error。这里有一个很关键的设计决策把业务状态和通信状态分开。什么意思就是HTTP 200不代表Agent成功完成了任务。目标Agent可能接收了请求但内部执行时报错了这时RPC响应里就应该出现业务错误码而不是在传输层返回500。Agent-Reach定义了一套基础错误码CAPABILITY_NOT_FOUND、INVALID_ARGUMENT、CALLEE_ERROR、TIMEOUT、UNAUTHORIZED。这套错误码的最大价值在于调用方Agent可以把“这次调用为什么失败”直接作为上下文反馈给大模型。模型看到CALLEE_ERROR之后可以决定换个Agent再试或者降级处理。这种容错能力在多Agent协作里非常关键。让大模型能够理解错误并自行决策而不是让代码把异常吞掉或者直接抛给用户。3. 实操环节从零搭一个能跑的Agent-Reach节点3.1 环境准备与初始化这次实操我用Python做演示Python生态做Agent相关的开发最顺手JsonSchema、embedding这些库也齐。你需要准备一台机器装了Python 3.10以上一个PostgreSQL实例以及一个可用的embedding接口我用的本地开源模型或者云厂商的都行。项目目录结构我建议直接这样建分得清楚后期不会乱agent-reach/ ├── hub/ # 注册中心服务 │ ├── main.py │ ├── models.py │ └── matcher.py ├── agent_sdk/ # Agent接入用的SDK │ ├── __init__.py │ ├── client.py │ ├── server.py │ └── capability.py ├── agents/ # 示例Agent │ ├── summarizer.py │ └── caller.py └── config.yaml依赖我能省则省只装这几个fastapi、uvicorn、sqlalchemy、psycopg2-binary、httpx、pydantic加上自己做embedding需要的sentence-transformers如果你用云服务这步可以跳过。依赖少有一个隐性的好处部署时不容易出幺蛾子。很多项目最后死在“环境装不上”而不是“代码有bug”上。3.2 先写注册中心HubHub是所有连接的核心我先把最基础的Agent注册接口写出来。核心就是接收一个Agent上线时发来的注册请求把能力描述解析后存库。这个接口用FastAPI写非常直白模型预先定义好。# hub/models.py from sqlalchemy import create_engine, Column, String, JSON, Float from sqlalchemy.orm import declarative_base Base declarative_base() class AgentRecord(Base): __tablename__ agents agent_id Column(String, primary_keyTrue) name Column(String) version Column(String) endpoint Column(String) # 接收RPC回调的地址 status Column(String, defaultonline) class CapabilityRecord(Base): __tablename__ capabilities capability_id Column(String, primary_keyTrue) agent_id Column(String, indexTrue) description Column(String) keywords Column(JSON) input_schema Column(JSON) output_schema Column(JSON) embedding Column(JSON, nullableTrue) # 存向量生产环境建议独立列# hub/main.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class CapabilityPayload(BaseModel): capability_id: str description: str keywords: list[str] input_schema: dict output_schema: dict class RegisterPayload(BaseModel): agent_id: str name: str version: str endpoint: str capabilities: list[CapabilityPayload] app.post(/register) def register(payload: RegisterPayload): for cap in payload.capabilities: record CapabilityRecord( capability_idcap.capability_id, agent_idpayload.agent_id, descriptioncap.description, keywordscap.keywords, input_schemacap.input_schema, output_schemacap.output_schema, ) session.add(record) session.commit() return {ok: True}注册之后最关键的就是查询匹配接口/query。它接收调用方Agent的能力需求描述返回排名最高的候选Agent列表。我把第三层的向量排序逻辑也贴出来实际实现里这一步可以异步做避免阻塞请求。# hub/matcher.py import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity def semantic_rank(query, candidates, top_k3): texts [query] [c[description] for c in candidates] vectorizer TfidfVectorizer() tfidf vectorizer.fit_transform(texts) sims cosine_similarity(tfidf[0:1], tfidf[1:]).flatten() ranked sorted(zip(candidates, sims), keylambda x: -x[1]) return [c for c, _ in ranked[:top_k]]这里我用TF-IDF先顶一版向量召回好处是零额外依赖。如果你的场景需要真正理解语义再把embedding模型接进来替换这里的表示即可。整个匹配链路在后面的实测里响应时间基本控制在几十毫秒级别完全够用。3.3 服务型Agent接入暴露能力服务型Agent那边要做两件事启动时把自己注册到Hub以及暴露一个HTTP端口接收RPC调用。SDK里的server.py把第二件封装一下让Agent开发者只需要写一个纯函数就像写本地工具一样。# agent_sdk/server.py import uvicorn from fastapi import FastAPI, Request import inspect class ReachServer: def __init__(self, agent_id, endpoint): self.app FastAPI() self.agent_id agent_id self.endpoint endpoint self.handlers {} def register_capability(self, capability_id, handler): self.handlers[capability_id] handler property def route(self): self.app.post(/rpc) async def rpc_handler(request: Request): payload await request.json() cap payload.get(capability) if cap not in self.handlers: return {status: CAPABILITY_NOT_FOUND} try: result self.handlers[cap](**payload.get(params, {})) return {status: OK, result: result} except Exception as e: return {status: CALLEE_ERROR, error: str(e)} return rpc_handler def start(self, port): uvicorn.run(self.app, host0.0.0.0, portport)用的时候只需要这样# agents/summarizer.py from agent_sdk.server import ReachServer def summarize(text, max_points3): lines [s.strip() for s in text.split(。) if s.strip()] return {summary: lines[:max_points], total: len(lines)} server ReachServer(agent-summ-01, http://0.0.0.0:9101) server.register_capability(text-summarization, summarize) server.start(port9101)这里register_capability把业务函数和capability_id绑定本质上是做了一层“能力即函数”的映射。Agent作者完全不感知RPC细节写完函数就上线。这套抽象在前端那套叫“透明代理”在Agent世界里其实可以叫“透明能力化”。3.4 调用方Agent的查询与调用调用方Agent我会封装一个DiscoveryClient。它负责和Hub通信并且维护一个本地缓存避免每次调用都去查Hub。多Agent场景里频繁注册查询会拖慢整体并发缓存是非常必要的。# agent_sdk/client.py import httpx, time class DiscoveryClient: def __init__(self, hub_base_url): self.hub hub_base_url self.cache {} self.cache_ttl 60 # 秒 def query(self, capability_idNone, descriptionNone, keywordsNone): payload { capability_id: capability_id, description: description, keywords: keywords or [], } resp httpx.post(f{self.hub}/query, jsonpayload, timeout5) return resp.json()[candidates] def get_target(self, capability_id): # 简单缓存逻辑 cache_key capability_id if cache_key in self.cache and time.time() - self.cache[cache_key][1] self.cache_ttl: return self.cache[cache_key][0] candidates self.query(capability_idcapability_id) if not candidates: raise RuntimeError(no agent for capability: capability_id) target candidates[0][agent_id], candidates[0][endpoint] self.cache[cache_key] (target, time.time()) return target def call(self, capability_id, params, timeout15): agent_id, endpoint self.get_target(capability_id) payload { protocol: reach-rpc, version: 1.0, request_id: req_ str(time.time_ns()), target_agent: agent_id, capability: capability_id, params: params, } resp httpx.post(endpoint.rstrip(/) /rpc, jsonpayload, timeouttimeout) body resp.json() if body.get(status) ! OK: raise RuntimeError(fRPC failed: {body.get(error)}) return body[result]调用方只需要写这一行result client.call(text-summarization, {text: long_text, max_points: 5})这在已经接入了大模型的Agent里非常实用——Agent先向你自己的大模型解释任务等你识别出这是一个“摘要”需求就用一次call(text-summarization, ...)把任务分发出去。整个过程就像给Agent装了一条条可拔插的外接数据线。3.5 一次完整的联调现场我把Hub跑在8000端口两个示例Agent分别跑在9101和9102端口。启动顺序有讲究先启动Hub再启动服务型Agent最后启动调用方。原因很直观服务型Agent挂了调用方的查询就会扑空你不会希望联调刚开始就看见一片404。实际跑下来链路分四个阶段服务型Agent启动时向/register发注册请求Hub返回成功。调用方Agent启动向/query发起“text-summarization”的精确ID匹配查询Hub极低延迟返回候选列表。调用方通过call把文本参数包装成Reach RPC请求发给服务型Agent的/rpc接口。服务型Agent执行summarize函数返回结构化{summary, total}调用方解析并合并到自己的业务逻辑。这一整套在本地跑单次端到端延迟大约在10到30毫秒之间其中大头是HTTP的网络往返。这个速度在Agent协作场景里完全可以接受毕竟一次调用里大头是模型推理连接层的开销几乎可以忽略。4. 常见问题与排查技巧实录这个部分是我最想写的因为实际运行的坑和文档里的理论完全是两回事。我挑几个我在项目里切切实实遇到并且修复了的问题整理成速查表你直接抄作业就行。4.1 Agent上线了但互相发现不了这个现象最气人两边代码都跑了日志也打了就是查询不到对方。后来查出来三个原因。第一个是注册接口幂等性没做好。Agent重启后重新注册可能因为旧记录的主键冲突导致新记录没写进去。解决方案很简单注册时按(agent_id, capability_id)做upsert而不是简单insert。第二个是角色搞混了。我一度把服务型Agent也当作调用方启动时没有调/register结果它只在本地启动了服务Hub里根本没有它的影子。这个问题提醒我SDK的启动引导里要把注册这一步做成强制项没注册成功就不该继续跑。第三个原因是嵌入向量和描述向量维度不一致导致最后的语义排序报错。这个往往是换了embedding模型或者改过向量化配置引起的最好在初始化的时候做一个维度的显式校验不匹配就直接报错fail fast不要等到线上请求打进来才发现。4.2 能力匹配不准确召回了大量无关Agent匹配精度这件事在能力多起来之后非常明显。一开始只有两三个Agent怎么匹配都对到了十几个Agent发现查询“文本处理”的时候把“图像处理”也召回了因为两者描述里都出现了“处理”这个词。修这个问题的核心不是换一个更牛的模型而是规范能力描述本身。我要求所有能力描述必须包含“动词宾语场景限制”比如“对中文新闻文本生成摘要”而不是泛泛的“文本处理”。同时在匹配时我把keywords权重调高description只做软匹配。关键词是人工维护的、精准的标签描述是大模型生成的、模糊的自然语言这个优先级关系要理清楚。很多时候不是你技术不行而是你的数据入口太随意。能力描述相当于Agent世界的API文档这部分花时间打磨后面所有下游环节都会受益。4.3 RPC调用超时或消息丢失在第一版里干脆没有超时控制。结果就出现过调用方Agent傻等一个下游Agent的场景那个Agent因为第三方接口卡顿整个请求就卡了十几秒。调用方的大模型拿到超时异常后自己又编了一个“重试”策略结果反复点同一个卡住的服务雪上加霜。后面我做了两件事第一调用方必须设置超时SDK里socket层面和httpx层面都对请求设置了上限不能等一个Agent无限期执行。第二超时后要走降级链路。我在SDK里加了一个可选的failover_candidates参数当首选Agent超时调用方可以自动重试第二个候选Agent。这个策略非常实用特别是在有多个Agent提供同一种能力的时候。关于消息丢失排查下来基本上都是异步调用时忽略了HTTP状态码。RPC协议明确写了传输层返回非2xx直接按照“调用失败”处理不能当作“任务已提交”。如果目标Agent接收了任务但还没处理完就挂了那唯一可靠的方案是给关键流程加上任务持久化把RPC请求落库。不过在我的场景里这属于后期优化MVP阶段可以先不做。4.4 权限校验在联调时成了最大阻碍权限这块是我真正后悔没早点设计的地方。最初为了图快所有Agent之间的RPC不鉴权反正内网嘛想着没事。结果有一次一个测试Agent的prompt被喂了一堆恶意指令它差点去调用另一个Agent的“删库”能力。虽然是测试环境但这个教训足够深刻。我现在实现了最基础的基于JWT的调用鉴权Hub在注册时给每个Agent签发一个agent_id和secret。Agent调用其他Agent时用secret签一个JWT放进请求的metadata.auth_token里。服务型Agent在RPC入口校验JWT签名和有效期校验通过才执行。这套逻辑不复杂但能挡住绝大多数误调用和prompt注入诱导的越权操作。你甚至可以在这个基础上做细粒度权限某个Agent只允许调用特定的capability这个是后话但至少要先把基础鉴权立起来。多Agent系统的安全问题本质上和人类组织一样——谁可以指挥谁谁可以调用谁的能力一定要有明确的边界。不要等到出了事故再补。5. 落地场景与进一步扩展5.1 企业内部Agent协作网络Agent-Reach最自然的落地场景就是企业内部的多Agent系统。你可能有十几个Agent分别管数据查询、文档处理、客户工单分类、Content Generation。通过Reach Hub把这些Agent连接起来新增Agent只需要注册自己的能力老Agent不需要改代码就能发现新Agent——这就是“触达”带来的扩展性红利。我看到一个很好的真实案例把Agent-Reach接入了企业内部的工单系统。用户提一个模糊的请求“帮我查一下上个月的退款数据并且生成一页简报”入口Agent先做意图识别发现需要“数据访问”和“报告生成”两个能力于是分别触达两个下游Agent然后对结果做汇总。整个过程里用户只面对一个入口Agent内部的触达对用户完全透明。这种体验就是Agent协作最典型的商业价值。5.2 个人工作流的“Agent总线”个人开发者同样可以用这个思路。你可以不用去部署一个重型Hub哪怕只是把Agent-Reach的注册表和路由逻辑跑在本地都能让你的个人Agent集合不再是一堆脚本而是一个真正彼此协作的系统。我自己的用法是本地跑一个轻量Hub注册了“笔记整理Agent”“日程解析Agent”“代码片段生成Agent”三个能力。日常输入“整理一下今天的会议记录并把待办事项拆出来”入口Agent就依次触达笔记整理和日程解析最后返回结构化结果。个人场景对并发要求低SQLite甚至一个JSON文件都可以当Hub的存储整个系统不需要什么运维成本。5.3 后续可以扩展的方向Agent-Reach离一个“完整体”还有距离。我列几个我认为明确值得做的方向P2P模式。当前有一个中心化的Hub这在大多数场景下够用。但如果Agent数量上千Hub自己可能成为瓶颈和单点将来可以引入基于DHT或者其他去中心化协议的P2P发现机制。回调与异步长任务。现在RPC基本都是同步的但Agent世界里经常有需要跑几分钟的任务。需要扩展出提交任务、轮询状态、回调通知的模式。可观测性与追踪。多Agent调用链特别长的时候排查一个问题要横跨好几个Agent必须有开箱即用的分布式追踪。我目前是在metadata里带trace_id但Agent之间的链路数据采集和展示还需要继续完善。6. 写在最后的一点体会Agent-Reach这个项目做下来我最深刻的体会是Agent协作的瓶颈从来不是单点能力而是连接方式。模型再聪明如果它触达不到需要的数据和工具就只能在你给的有限上下文里自说自话。反过来一旦Agent和Agent之间有了标准的发现和调用通道很多原本需要写死在业务代码里的逻辑就会变成Agent自己的决策——这比“硬编码智能”灵活得多也优雅得多。还有一句实在话不要指望一开始就设计出完美的连接层一定要先让两个Agent跑通再扩展到五个、十个。跑通第一个链路的时候你会对整个系统产生真实的体感你的设计也会随之落地而不是停留在PPT里。如果你也在做Agent相关的东西并且遇到了“Agent之间怎么互相呼叫”这个头疼的问题不妨照着我这个思路用最小的成本搭一套连接层试试。一旦你尝到了“让Agent自己找到协作伙伴”的甜头你就再也不想回到一个个硬编码的老路上去了。