Agent-Reach：多Agent协作的语义发现与协商通信机制

开头

先说明一下Agent-Reach这个项目到底在解决什么问题。如果你最近在折腾AI Agent，大概率会碰到一个很尴尬的局面：单体的Agent跑得还算顺，但一旦想让多个Agent协作，马上就会被"通信""发现""权限"这三座大山卡死。A Agent想调用B Agent的能力，互相不知道对方在哪、能干什么、怎么调用，最后只能回到传统HTTP接口那一套，变成了"披着Agent皮的微服务"。

Agent-Reach就是在这种背景下冒出来的一个思路——它不是一个具体的框架或某个厂商的产品，而是一套让Agent之间能够互相发现、握手、协商并协同执行任务的机制集合。我这边花了大约两周时间，把概念弯路走了一遍，最终在几个真实场景里跑通了。这篇文章会把整个设计思路、核心原理、实操步骤和踩坑记录全部整理出来，重点围绕通信协议、能力注册与发现、任务协商和权限控制这几个维度展开。适合正在做多Agent系统、想要绕过单体Agent瓶颈的开发者参考，也适合刚入门、想搞明白"多Agent协作到底比单体强在哪"的读者收藏。

1. 整体设计与思路拆解

1.1 从"单体Agent"到"Agent网络"的必然转变

先聊聊为什么会有Agent-Reach这个需求。最开始我手上只有单体Agent，偶尔写点自动化脚本配一个工具调用，跑得挺顺。但是当任务粒度开始细起来——比如"根据文档站内容生成一份市场分析报告，再自动同步给团队群"这种跨数据源、跨动作域的复合任务——单体Agent的维护成本立刻失控了。

单体Agent通常把检索、推理、生成、执行全部塞在一个进程里，模型上下文一长就崩，工具一多就乱。最典型的表现是：Agent开始"幻觉调用"，它明知道自己没有某个工具，却会假装任务已经完成。这种情况下，没有任何一种prompt工程能根治，因为问题出在架构上。单体Agent的本质是"一个人干所有事"，而现实世界的业务天然是多角色协作。

所以Agent-Reach的核心假设很简单：把Agent拆成一个个具备单一职责的"能力单元"，每个单元只做好一件事，然后让这些单元通过一套轻量级的协议互相发现、互相调用。这个思路有点像微服务之于单体应用，但Agent-Reach多了两个关键差异，一个是能力描述是语义化的，不是接口文档式的；另一个是任务执行是协商式的，不是调用链式的。这两个差异，正是它与传统RPC、消息队列方案的分水岭。

1.2 为什么不用现成的消息队列或RPC框架

转这一圈的时候，团队里有人提过：直接用gRPC或者MQTT不就行了吗？为什么要自己搞一套？

我当时的回答是：因为Agent之间的通信，本质上不是"方法调用"，而是"意图协商"。用gRPC拉起一个远程调用，你得先定义好请求响应结构，这要求调用方在编码期就知道被调用方的能力。但Agent网络里，A Agent根本不可能事先知道B Agent能干什么——它是运行期通过能力描述动态发现的。这个前提决定了，传统的接口契约模式天然不适配。

MQTT这类消息总线的问题又在另一个方向上。它能解决广播和订阅，但Agent之间不是简单的事件订阅关系，而是"我有个目标，需要找到能完成子目标的Agent伙伴"这种目标驱动的匹配关系。事件总线无法表达"谁能完成某个目标"这种语义层面的问题，你最终还是要在消费者侧写一堆匹配逻辑。

因此Agent-Reach选择了"语义能力注册表+意图路由+协商握手"三条腿走路的架构，既不跟具体协议绑定，也不丢失Agent本身的语义表达力。后面会展开讲这三条腿各自怎么落地。

1.3 Agent-Reach的三大设计支柱

Agent-Reach并不是一个单点软件，它拆成三层来看比较清晰：

第一层是身份与能力描述层。每个Agent启动时，向注册中心提交一份Agent Description Document（ADD），里面包含身份标识、能力标签、输入输出Schema、权限级别、可用状态和语义描述。这相当于Agent世界的"简历"。

第二层是意图路由层。当一个Agent发起任务请求时，它不指定"我要调哪个Agent"，而是描述"我要完成什么目标"。路由层根据ADD匹配语义相似度，返回一组候选Agent列表。这一步我实测下来，纯关键词匹配会死得很惨，至少要用embedding相似度加一层用户定义规则做混合召回。

第三层是协商执行层。候选Agent不是被直接调用的，而是先经过一个"能力确认、参数对齐、执行窗口确认"的握手过程。只有双方都对约束条件达成一致，任务才真正落地执行。这个设计主要为了避免"开干之后发现角色边界没对齐"的问题。

这三层在逻辑上完全解耦，也就是说你可以替换掉任何一层，只保留其他两层。我代码里也单独打了模块边界，后面有需求的话可以只把路由层换成GraphRAG或别的算法，不折腾其他地方。

2. 核心细节解析与实操要点

2.1 能力描述文档（ADD）怎么写才不走弯路

ADD是整个体系的基石。我最初犯的错误是把它写得像一份"能力清单"，每个能力一行短句子，结果语义粒度太粗，路由层根本分不清"生成报告"和"生成周报摘要"的差别。

经过反复调试，我把ADD固定成五个必填字段。

identity：Agent的唯一标识。我推荐统一用语义化ID，比如search-engine-agent这种格式，不要用UUID字符串，因为日志排查的时候，可读性很重要。

capabilities：这是核心，每个能力对象包含name、description、inputSchema、outputSchema、level五元组。description建议写成完整的句子，例如"能够从指定知识库检索信息并返回带来源出处的摘要"，而不是"检索功能"。

constraints：包括允许并行数、最大并发数、单次执行时间上限、幂等性标记。这里有一个容易忽略的坑——如果没有声明幂等性，Agent重复执行一次请求可能产生重复工单，业务上几乎是灾难。

visibility：可被谁调用。用白名单模式，默认只允许显式列出的Agent ID调用。我之前试过黑名单模式，完全行不通，因为你无法预测哪些新加入的Agent会成为安全短板。

metadata：包含维护者、版本号、更新时间、依赖声明。依赖声明特别重要，因为我会让路由层在匹配能力时顺带校验依赖是否就绪，避免选出一个依赖未初始化的Agent。

以下是我项目里一个真实ADD的简化示例。注意字段命名是JSON风格的，便于存储和传输。

{ "identity": "knowledge-retrieval-agent", "version": "2.1.0", "capabilities": [ { "name": "semantic_search", "description": "在已接入的知识库中进行语义向量检索，返回按相关度排序的结果，每条结果包含文本片段和来源文档标识", "inputSchema": { "query": "string", "top_k": "number", "filters": "object" }, "outputSchema": { "results": "array", "total_found": "number", "latency_ms": "number" }, "level": "core" }, { "name": "document_summarize", "description": "对传入的文档内容生成结构化摘要，支持提取要点、生成结论、统计关键词", "inputSchema": { "doc_id": "string", "style": "string" }, "outputSchema": { "summary": "object", "key_points": "array", "keywords": "array" }, "level": "advanced" } ], "constraints": { "max_concurrency": 8, "max_execution_time_sec": 30, "idempotent": true, "allowed_callers": [ "report-generation-agent", "workflow-orchestrator" ] }, "metadata": { "owner": "team-knowledge", "version": "2.1.0", "dependencies": [ "vector-db-v2", "llm-service" ] } }

这个结构看着简单，但每个字段的实际产出我都花了一整天时间打磨。尤其是description的措辞，我后来总结出一个小技巧：写description时想象自己是另一个Agent的开发者，看到这段描述之后，能否准确判断"该不该把自己的任务交给这个Agent"。如果描述里出现"等等""或其他"这类含糊词，基本可以判定不合格。

2.2 能力注册与心跳保活机制

ADD写好后，Agent启动时要把自己注册到中心节点。这里需要明确一点——注册中心本身可以是一个独立进程，也可以内嵌在某个Leader Agent里，取决于你的部署规模。我初期用单节点模式，后期为了验证横向扩展能力，改成了多节点注册集群，用的是非常朴素的一致性哈希分片。

注册消息我设计成这样：

{ "action": "register", "agent_id": "knowledge-retrieval-agent", "add_version": "2.1.0", "endpoint": "grpc://10.0.0.5:9001", "ttl": 60, "timestamp": 1710000000 }

注册中心收到后，会做三件事：校验ADD的JSON Schema合法性，校验identity是否与签名匹配，写入内存哈希表并设置TTL过期时间。Agent每60秒发送一次心跳续期消息，如果超过150秒没有收到心跳，注册中心就把该Agent标记为unavailable并从路由候选池里摘除。

这里有一个实操细节很容易踩雷：Agent的endpoint是动态变化的，尤其在使用容器或动态调度平台时。我的做法是Agent每次启动后，先从环境变量读取当前调度分配的地址，拼成endpoint，再发起注册。如果直接复用配置文件里的地址，在集群环境下会有一大半的Agent注册成废地址。这个坑我花了整整半天才定位到，最后就是加了一行日志打印实际注册的endpoint才发现的。

另外建议把心跳和业务请求分开走。我最初把心跳放在Agent的主线程里，结果一旦Agent在处理长任务，心跳就会延迟，导致被误判下线。后来我单独开了goroutine做心跳发送，业务线程完全不受影响，误判率直接降为零。

2.3 意图路由：从关键词到语义匹配

路由层是决定整个系统体验最直接的部分。最早我试过三种方案，按复杂度递增分别是：关键词精确匹配、TF-IDF向量匹配、Embedding语义匹配。

关键词精确匹配最简单，但效果惨不忍睹。Agent描述里"查询"和"检索"是不同的词，关键词匹配会漏掉大量语义等价的能力。TF-IDF稍微好一些，但它只解决词汇重叠问题，解决不了"能生成日报但描述里写的是整理日报"这种同义改写情况。

最后我选用的是Embedding方案：每个能力注册时，把name加description拼成一段文本，用Sentence-BERT模型生成一个768维向量，索引到向量数据库里。路由时，把请求目标描述转成向量，用余弦相似度召回Top10，然后再用一层规则过滤做精确约束。

规则过滤部分的作用是保证硬性条件不违反。比如请求里指明"只要本地执行、不要调用外部Api的Agent"，那语义匹配得再接近也得被过滤掉。还有一个细节是：过滤规则应该尽可能少且断言性质一目了然，不要放"评分大于7"这种模型蒸馏出来的软规则，否则调试起来非常痛苦。

我的路由逻辑核心代码如下：

def route_request(request_goal, top_k=10): goal_vector = embed_model.encode(request_goal.description) candidates = vector_db.search(goal_vector, top_k=top_k) filtered = [] for candidate in candidates: matched, reason = check_constraints(candidate.agent, request_goal.constraints) if matched: filtered.append((candidate, similarity(candidate.vector, goal_vector))) else: log_route_filtered(candidate.agent.identity, reason) return sorted(filtered, key=lambda x: x[1], reverse=True)

这段代码的关键点在于check_constraints必须返回拒绝原因并打日志。我一开始只返回True/False，后来排障的时候完全无从下手——你根本不知道某个Agent为什么没有被选上。加上日志之后，排障效率提升了十倍。

2.4 协商握手：避免"发布会式"的虚假执行

所谓协商握手，简单说就是发起方给出任务请求之后，候选Agent不能直接执行，而要先进行一次双向确认。这一步被我称为"防幻觉检查"，因为很多Agent系统出问题，不是出在能力上，而是出在发起方以为对方能做到、其实做不到的情形上。

握手协议我设计为四步：

第一步，发起方发送TaskOffer，包含目标描述、输入参数、期望输出格式、截止时间、报酬或优先级字段。这里的截止时间必须精确到秒，因为我发现Agent任务经常因为"尽快"这种模糊描述导致后续排队策略完全失效。

第二步，接收方返回TaskQuote，包含接受意向、执行方案摘要、预估耗时、所需额外条件。比如检索Agent会说"该查询需要访问财务知识库的权限，当前未授权，需要确认是否已授权"。

第三步，发起方确认TaskAgreement，附带实际的输入数据或数据引用。注意这一步开始才算正式任务开始。前面两步都只是"商业谈判"。

第四步，接收方执行完成后返回TaskReceipt，包含结果数据、元数据、实际耗时、异常记录。

四步走看似增加了一次网络往返，但对于多Agent系统来说，省掉的排查成本是巨大的。我在生产环境里遇到过好几次"Agent A认为已经调用了Agent B，但Agent B其实没收到任何请求"的情况，起因全都是跳过握手、直接硬调。四步协议把这类bug消灭在根上。

3. 实操过程与核心环节实现

3.1 环境准备与基础组件列表

实践这套架构时，我用的环境是一台8核16G的开发机，操作系统是Ubuntu 22.04，Docker Compose管理依赖组件。核心组件清单如下：

• Python 3.10以上，Agent主逻辑用FastAPI框架跑HTTP接口
• Redis做注册中心和心跳存储，用有序集合记录Agent的TTL时间
• Chroma作为向量数据库存储能力Embedding
• Sentence-BERT模型使用all-MiniLM-L6-v2，维度384，足够用
• Go编写一个轻量路由网关，处理请求转发和协议转换

为什么不用Elasticsearch做向量检索？因为前期数据量也就几百个能力向量，Chroma完全能扛住，而且部署轻量很多。后期数据量过了百万级再考虑迁到专门的向量数据库不迟。

编排全部走Docker Compose，共五个容器：redis、chroma、router-gateway、agent-knowledge、agent-report。写到这发现有一个原则必须提前说明——不要把所有Agent塞进同一个Docker容器里。有人为了省资源这么干过，结果所有Agent共用一个进程内消息队列，注册中心形同虚设，Agent之间的隔离性完全丢失。

3.2 路由网关的协议转换实现

路由网关是整个系统的"翻译层"。为什么需要它？因为Agent之间可能使用不同的传输协议——有的Agent暴露gRPC接口，有的暴露HTTP。发起方不可能要求所有Agent统一协议，所以网关负责接收统一的HTTP请求，转换成目标Agent支持的协议格式。

func (g *Gateway) handleTaskOffer(w http.ResponseWriter, r *http.Request) { var offer TaskOffer if err := json.NewDecoder(r.Body).Decode(&offer); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } candidates, err := g.registry.Route(offer.Goal, offer.Constraints) if err != nil || len(candidates) == 0 { http.Error(w, "no candidates found", http.StatusNotFound) return } selected := candidates[0] endpoint, protocol := buildEndpoint(selected.Identity) switch protocol { case "grpc": err = g.grpcClient.SendOffer(offer, endpoint) case "http": err = g.httpClient.SendOffer(offer, endpoint) default: err = fmt.Errorf("unsupported protocol: %s", protocol) } // 在这里记录完整的路由决策链，方便trace logRouteDecision(offer, selected, endpoint) }

这个网关代码本身不复杂，但有一个设计取舍值得说一说：为什么不让发起方直接连接收方，而是中间插一个网关？

我实测的直接连接模式问题在于，注册中心会把每个Agent的地址暴露给所有请求方，这意味着任意Agent都能拿到全局拓扑信息，安全上完全不可控。中间插网关之后，Agent之间不直接交换地址，网关作为唯一跳板，所有调用请求都经过它做鉴权和审计。这个改动让权限控制从"occupied Spirit trust"变成了"强制强制中心化授权"，安全风险下降非常明显。

3.3 一个真实场景：市场分析报告自动生成

讲完组件，用一个真实跑通的场景把整个流程串起来。场景目标是：输入一篇产品说明文档链接，产出市场分析报告并同步给团队协作群。

整个任务拆成三个子任务：先从文档站抓取内容并清洗，然后根据清洗后的内容生成分析报告，最后把报告格式化并推送到协作群。对应三个Agent：文档采集Agent、分析Agent、推送Agent。

流程如下：

第一步，我在统一个入口提交目标："根据提供的文档链接，生成包含市场背景、竞品对比、优劣势分析、行动建议四部分的报告，并发送到指定协作群。"

路由网关收到后，把目标转成向量，到能力库中召回三个Agent候选。规则过滤阶段，推送Agent的allowed_callers校验通过，因为发起方是人工管理员账户，在白名单内。

第二步，网关向文档采集Agent发出TaskOffer。文档采集Agent返回TaskQuote，声明"该链接域名为外部站点，外部站点抓取受限，可能需要5-10秒额外延迟"。发起方确认后，它执行抓取，返回清洗后的Markdown文本。

第三步，分析Agent拿到Markdown文本，根据输入参数里的"四部分结构"要求，调用大模型生成报告。这个过程耗时18秒，在Agent声明的30秒limit以内。之后返回结构化报告，附带关键数据来源的引用链接。

第四步，推送Agent收到报告内容后，先校验报告格式是否合规（有标题、有分段、无空指针），再调用协作群API发送。发送成功返回消息ID作为回执。

整个流程用时约32秒，其中网络传输占2秒，路由召回不到100毫秒，其余时间全在LLM推理和文档抓取上。对比单体Agent跑同一个任务，单体方案的问题是上下文占用过高，一次prompt要塞下文档全文、报告模板、群接口说明、历史记录等，导致模型上下文暴掉，经常出现"分析报告只有三行废话"的质量塌方。拆成多Agent后，每个Agent一次只关心自己职责内的上下文，质量明显稳定很多。

3.4 执行场景下的参数计算与调优记录

汇报一下关键参数的实测数值，这些参数都不是拍脑袋定的，而是通过压力测试跑出来的。

并发数的设置：注册中心维护的Agent实例数量为12个，其中检索类3个、分析类4个、推送类2个、辅助工具类3个。max_concurrency我初始设10，结果在高并发测试时Redis键冲突明显增多，因为多个心路点同时写入TTL过期时间戳。实测把检索类调成8、分析类调成5之后，冲突率从3.2%降到0.4%。

任务超时时间的设置：理解类任务设30秒，搜索类任务设12秒，推送类任务设8秒。最初统一设30秒导致推送Agent失败后要白白等26秒才被路由层摘除，用户体验极差。按能力粒度分设超时时间后，失败恢复时间缩短了一个量级。

心跳间隔设60秒，从实测来看，60秒基本不会让路由层感知到明显的"僵尸Agent"。如果调成10秒，高频心跳会让Redis的写入压力翻六倍，得到的收益却不明显，因为现实中Agent崩溃后，路由层在几十秒内感知到和在一分钟内感知到，对用户体验的差别不大。

4. 常见问题与排查技巧实录

4.1 Agent注册后路由不到：九成是TTL问题

最常见的问题是Agent启动后能正常把自己注册上去，但在路由层怎么都查不到。我排查这个问题的路径是：先查心跳日志，确认Agent是否在按预期发心跳。很多情况下Agent的心跳代码被卡在业务线程里，发出去了但注册中心收不到。

再查Redis里的TTL时间戳，用命令直接看：

redis-cli ZSCORE agent_registry "knowledge-retrieval-agent"

如果返回的时间戳和当前时间相差超过150秒，说明这个Agent的TTL已经过期，注册中心把它标记为下线了。这时需要检查Agent机器与Redis主机之间的时间同步。虚拟化环境中时间漂移很常见，如果时间差大于30秒，TTL逻辑就会误判。

我的解决办法是：在Agent容器启动命令里加一行时间同步脚本，并统一使用UTC时间戳进行心跳和TTL计算，避免时区转换引起的边界误差。

4.2 语义召回结果不准：描述文本是首要嫌疑对象

路由召回不准时，大家第一反应是换Embedding模型，但大多数情况下真正的根因是能力描述写得太抽象。比如描述"能够处理内容"和"能够对文档进行语义分段、提取小节标题和段落要点"，在向量空间里的位置几乎完全不相关。

我总结了一个自测方法：把描述里的每个动词和名词圈出来，问自己一个问题——如果另一个Agent的开发者只看到这个动词名词组合，能否知道该调用它做什么任务？如果答案是可以，那这段描述就合格了。如果答案是"大概能吧"，请继续重写。

还有一个小技巧，可以在测试阶段把每个能力的description打印出来，人工观察它们之间的余弦相似度矩阵。如果发现两个本质不同的能力相似度高达0.9以上，大概率是描述措辞太相似，需要拉开语义距离。

4.3 协商握手超时：候选Agent的背压机制缺失

握手阶段也会出现超时，尤其是某个Agent瞬间收到大量TaskOffer请求时。Agent内部如果没做背压控制，会把所有请求都接进队列，结果大量请求排队等待，握手响应迟迟不出，发起方那边自然超时。

我在Agent内部加了一个容量信号量：并发处理器数达到上限后，新的TaskOffer直接返回"busy，retry later"的TaskQuote，不会进队列。发起方收到busy响应后，会重新下发路由请求，这次路由层会优先选择其他候选Agent，实现自动负载均衡。

具体信号量代码如下：

from threading import BoundedSemaphore, current_thread class AgentExecutor: def __init__(self, max_concurrency=5): self.slots = BoundedSemaphore(max_concurrency) def try_acquire(self) -> bool: return self.slots.acquire(blocking=False) def release(self): self.slots.release()

注意acquire必须用blocking=False，否则又会变成排队模式，背压失效。这个坑我踩了，最初用默认的blocking=True，结果busy响应根本没机会发出去，因为所有线程都在等待信号量。

4.4 Agent执行幂等性缺失导致重复任务

另一个隐蔽问题出在"幂等性"声明上。我在ADD里加了idempotent字段，但这个字段的价值只有在真的实现到代码里时才能发挥出来。如果某个Agent如实声明自己幂等，但代码里没有去重，那么发起方根据确认重试机制，会把同一个任务执行两遍。

我的解决办法是给每类Agent的输入增加一个requestId字段，执行前先在Redis里检查requestId是否已存在；如果存在，直接返回上次的执行结果；如果不存在，则写入执行中的状态，执行完成后再更新为完成状态。

这个模式类似数据库的写操作幂等键，虽然会多一次Redis往返，但换来的是任务执行的确定性。在Agent网络里，"不确定是否执行"比"执行失败"更可怕，因为前者会导致业务数据错乱，后者至少还能重试。

4.5 问题速查表

5. 写在最后的几点实战体会

这套Agent-Reach思路跑通之后，我最大的感受是：多Agent系统不是简单地"把单Agent拆成多个API服务"，而是在通信模式的底层把"调用"改成了"协商"。这个转变带来的好处是系统变得更有韧性——某个Agent挂掉或者忙不过来，路由层自动绕过它，任务不会失败，只会换一条路径继续跑。

我个人强烈建议，如果你现在正准备从单体Agent迁移到多Agent架构，不要先急着写业务逻辑，先把ADD的字段定义给做了。一个完善的能力描述体系，会在后面所有环节里为你省下十倍的时间。反过来，如果ADD写得很敷衍，路由、排查、扩展三个环节都会持续地折腾你。

有一个可以立刻用起来的小技巧：在注册中心的日志里，为每次路由决策打印一条结构化记录，包含请求ID、发起方、目标Agent、匹配度、过滤规则执行结果。这条记录是日后一切排障的基础，我称它为"路由黑匣子"。每次系统出现理解分歧，翻这个黑匣子基本都能在几秒钟内还原现场。

这个架构最大的扩展空间在于，后续可以把"协商协议"升级为带预算和计费的版本，让Agent之间在协商阶段就能预估成本、定义补偿条件。甚至可以在协商中加入"能力组合包"的聚合报价，比如多个Agent联合响应一个复杂请求，并且各自讲清楚分担的职责。这条路走下去，多Agent协作会越来越像一支扁平化团队，而不是一台笨重的调度机器。