多Agent触达治理与协作编排：从声明式能力卡到动态路由实践

1. 项目概述

1.1 从名字说起：Agent-Reach到底解决什么问题

先聊个实际场景。我之前在做一套多智能体协作系统的时候，最头疼的事情不是某个Agent的提示词写得不好，也不是模型本身的推理能力不够，而是Agent与Agent之间、Agent与外部工具之间的“触达”常常失败。明明每个单点能力都验证过了，一旦组合起来，就会出现A调不动B、B找不到该用的工具、工具返回了结果但C解析不了这些鬼问题。

当时团队内部复盘，发现我们把大量精力花在了单个Agent的“智能”上面，却几乎没花精力在Agent的“连接能力”上。Agent-Reach这个名字，就是这个背景下冒出来的——它不是什么大厂开源框架，而是我们自己沉淀的一套多Agent触达与协作编排方案。Reach这个词，在计算机网络里叫“可达性”，在业务里叫“触达率”，放在Agent场景里，恰好就是那个被大多数人忽视的命门：你的Agent到底能稳定、准确、高效地触达多少资源、多少其他Agent、多少外部服务。

这篇内容适合正在做Agent落地、尤其是做多Agent协作或者复杂任务编排的开发者。如果你刚接触Agent，能帮你建立一张完整的Agent协作地图；如果你已经在做相关工作，下面这些机制拆解、参数配置和踩坑记录，应该能直接拿来用。

1.2 一个案例说清楚核心价值

举个我实际遇到的例子。运营部门想做一套“竞品动态自动追踪”流程：一个监测Agent每小时扫描指定竞品官网和公开渠道，把变更内容丢给分析Agent，分析Agent判断是否值得跟进，再让写手Agent生成一份简报推送到企业微信。

听起来不复杂对吧？真跑起来就全暴露了。监测Agent偶尔爬不到页面没报告，分析Agent的输入格式一变化就误判，写手Agent等不到上游结果直接超时，企业微信的机器人Webhook偶发限流，任何一个环节断了，整条链路静默失败，没人知道。

我一个人盯着三个Agent和五个外部API，像在同时维护五条易碎的水管。后来把这些经验沉淀成Agent-Reach这套方案，核心就一句话：把“触达”当作一等公民来设计。每个Agent能触达谁、通过什么协议触达、用什么格式触达、触达失败怎么降级，这些全部显式声明、统一治理、随时可观测。做完之后，同样的链路上线后基本不用再管它。

2. 核心机制拆解：让每个Agent“够得着”该够的东西

2.1 触达治理模型：声明式的能力边界

我之前见过太多团队做多Agent系统，最后都会做成“每个Agent的System Prompt里塞一大段别的Agent能干什么”，长此以往提示词会膨胀到几千字，且根本维护不了。Agent-Reach的第一层设计，是把这种隐式的、散落在提示词里的协作信息，全部抽出来变成显式的触达声明。

具体就是每个Agent在注册的时候，要交一份类似“能力卡”的声明，里面包含三大块：能触达的外部对象清单（工具、API、另一个Agent）、每个对象的输入输出协议（字段名、类型、样例）、触达的约束条件（超时上限、频率限制、失败时的降级策略）。

这份声明不是写给人看的文档，而是存在统一注册中心里的结构化配置。某个Agent想调另一个Agent，不直接传HTTP地址，而是传“目标Agent ID + 意图”，由注册中心根据当前版本的路由表决定怎么转发。这样做的好处，一是解耦了Agent之间的物理依赖，二是让系统具备动态调整能力，三是任何一次触达动作都能被记录和审计，不是黑盒。

在实际落地中，我们把能力卡做成YAML配置。举个例子，分析Agent的能力卡里有一项是“接收监测Agent的变更事件”，输入协议定义一个JSON结构，包含URL、变更类型、变更摘要、原始抓取时间四个字段；约束条件写的是“超时10秒，失败后重试2次，仍失败则进入待处理队列”；降级策略是“如果监测Agent暂时不可用，允许接收来自队列补偿器的批量事件”。

这套显式声明看起来多写了几个配置文件，但它把原先藏在提示词里的逻辑全部变成了可测试、可版本管理、可灰度发布的东西，这才是能支撑规模化协作的地基。

2.2 路由策略：不只是发个消息那么简单

确定了每个Agent的触达声明之后，下一层就要处理“消息去哪儿”的问题。早期我们只用最原始的方式：给每个Agent一个固定的下游地址表，A永远发给B。这个方案在Agent少于五个的时候还行，一旦数量上来，每次新增一个Agent都要去改N个上游的配置，太痛苦了。

现在的方案是引入意图路由层。发送方不用关心具体找谁，只需要说“我要做一次竞品分析”或者“我有一份文档需要摘要”，路由层根据触达声明和能力卡，结合当前的可用状态、负载情况、历史成功率，动态选择一个合适的接收方。这里面的核心逻辑，是一个基于多维度权重的打分器。

我分享一下我们的路由打分设计。每个候选接收方会从三个维度打分：语义匹配分，用向量化后的意图描述和Agent能力描述算余弦相似度，占比40%；可用性分，根据最近五分钟的成功率、平均响应时延、当前排队任务数综合评估，占比40%；偏好分，根据发送方的历史成功合作Agent做正向加权，占比20%。

这个权重比例是我们在实际业务里调出来的。最早语义匹配占比是60%，结果常常把请求路由给“听起来很擅长”但实际负载已经很重的Agent，导致整体时延变大；后来把可用性的权重拉高到和语义并重，效果立刻好了很多。另外要注意，路由决策本身必须快，我们把这层逻辑做成了独立服务，单次路由决策的P99耗时控制在5毫秒内，不能因为路由本身拖慢了整条链路。

一个好的路由层，还应该有全网可见的可观测面板。我比较推荐的状态指标包括：每个Agent的实时成功率、平均时延、排队深度、近一小时触达次数分布、超时和失败的明细列表。有了这些数据，你才能判断一个Agent“够得着”到底是够得着还是只是偶尔够得着。

3. 实操构建：从零搭一套Agent-Reach的最小骨架

3.1 技术选型：不追新，只求稳

我看见很多朋友一上来就要上K8s、要上服务网格、要上消息中间件全家桶，结果第一周连本地开发环境都跑不起来。Agent-Reach这套体系，技术选型的原则很简单：尽量少引入组件，每个组件都要能回答“它解决什么不可替代的问题”。

• 注册与路由服务：用Python FastAPI实现，轻量、生态好、写起来快。核心作用就是存能力卡、算路由、做健康检查。
• 消息传递：小规模用Redis Stream即可，支持发布订阅和持久化，足够支撑几百个Agent的协作场景。等规模真的上来了，再平滑替换成Kafka之类的重组件不迟。
• Agent执行器：每个Agent本体就是一个常驻进程，订阅自己相关的消息队列，处理完把结果发布到下一个主题。这样单个Agent挂了，不影响全局链路，队列里的消息不会丢。
• 状态存储：用一个PostgreSQL存协作会话状态和审计日志。PostgreSQL在这里的主要作用是方便复杂查询，例如“某个会话经过的所有Agent路径”。

这套组合的好处是，任何一台开发机上只要有Docker，一条命令就能把所有依赖拉起来，本地调试体验和线上环境完全一致。不要在你还不清楚系统瓶颈在哪的时候，就引入一堆分布式基础设施。

3.2 核心代码实现：注册、路由、协作的基本逻辑

下面我会用一段精简但完整的Python代码演示三个核心环节：注册能力卡、路由打分、消息投递。这段代码不是玩具，而是我们最初版本的真实原型，你可以直接复制改改就能用。

# agent_reach_core.py # Agent-Reach 最小核心：注册中心 + 意图路由 + 消息投递 from dataclasses import dataclass, field from typing import Dict, List, Optional import time import hashlib import json import numpy as np from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI() @dataclass class AgentCapability: """Agent 的能力卡声明""" agent_id: str name: str description: str input_protocol: Dict[str, str] # 字段名 -> 类型说明 output_protocol: Dict[str, str] supported_intents: List[str] timeout_seconds: int = 10 retry_count: int = 2 fallback_action: str = "enqueue" # 降级策略：enqueue / drop / notify @dataclass class AgentStatus: """Agent 的实时运行状态""" agent_id: str success_count: int = 0 fail_count: int = 0 total_latency_ms: int = 0 queue_depth: int = 0 last_heartbeat: float = 0.0 @property def success_rate(self) -> float: total = self.success_count + self.fail_count if total == 0: return 0.0 return self.success_count / total @property def avg_latency_ms(self) -> float: total = self.success_count + self.fail_count if total == 0: return 0.0 return self.total_latency_ms / total class ReachRegistry: """注册中心：存储所有 Agent 能力卡和运行时状态""" def __init__(self): self._capabilities: Dict[str, AgentCapability] = {} self._statuses: Dict[str, AgentStatus] = {} self._embedding_cache: Dict[str, List[float]] = {} def register(self, cap: AgentCapability) -> None: self._capabilities[cap.agent_id] = cap self._statuses[cap.agent_id] = AgentStatus(agent_id=cap.agent_id) # 注册时清掉缓存，防止能力描述变更后向量还是旧的 self._embedding_cache.pop(cap.agent_id, None) def get_embedding(self, text: str) -> List[float]: """实际系统里调用 embedding 模型接口，这里用哈希模拟稳定向量""" if text in self._embedding_cache: return self._embedding_cache[text] raw = hashlib.md5(text.encode()).hexdigest() vec = np.array([int(raw[i:i+2], 16) / 255.0 for i in range(0, 32, 2)]) self._embedding_cache[text] = vec.tolist() return vec.tolist() def cosine_sim(self, a: List[float], b: List[float]) -> float: a = np.array(a) b = np.array(b) return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))) class IntentRouter: """意图路由器：根据语义、健康度、偏好三个维度做打分""" def __init__(self, registry: ReachRegistry): self.registry = registry self.weight_semantic = 0.4 self.weight_health = 0.4 self.weight_preference = 0.2 self.preference: Dict[str, Dict[str, float]] = {} # 发送方 -> {接收方: 历史好感度} def route(self, intent: str, sender_id: str) -> Optional[str]: """ 根据意图找到最合适的接收 Agent。 返回 None 表示没有可用的 Agent，需要触发降级策略。 """ candidates = [] intent_vec = self.registry.get_embedding(intent) for agent_id, cap in self.registry._capabilities.items(): if agent_id == sender_id: continue # 语义分：用意图向量和 Agent 能力描述算相似度 cap_vec = self.registry.get_embedding(cap.description) semantic_score = self.registry.cosine_sim(intent_vec, cap_vec) # 健康分：结合成功率、时延、队列深度，满分为 1 status = self.registry._statuses[agent_id] health_score = 0.0 # 成功率低于 50% 的 Agent 直接一票否决 if status.success_rate > 0.0 and status.success_rate < 0.5: continue health_score = ( status.success_rate * 0.5 + min(1.0, 500.0 / max(status.avg_latency_ms, 1)) * 0.3 + max(0.0, 1.0 - status.queue_depth / 20) * 0.2 ) # 偏好分：历史合作越顺手的 Agent 越优先 pref_score = self.preference.get(sender_id, {}).get(agent_id, 0.0) # 发送方声明里的 supported_intents 不匹配时直接剔除 if intent in cap.supported_intents or not cap.supported_intents: final_score = ( self.weight_semantic * semantic_score + self.weight_health * health_score + self.weight_preference * pref_score ) candidates.append((final_score, agent_id)) if not candidates: return None candidates.sort(key=lambda x: x[0], reverse=True) return candidates[0][1] def record_success(self, sender: str, receiver: str, latency_ms: int) -> None: """路由成功后记录偏好，避免同一发送方在相似意图下频繁换接收方""" self.registry._statuses[receiver].success_count += 1 self.registry._statuses[receiver].total_latency_ms += latency_ms self.preference.setdefault(sender, {}).setdefault(receiver, 0.0) self.preference[sender][receiver] = min(1.0, self.preference[sender][receiver] + 0.1) def record_failure(self, sender: str, receiver: str) -> None: self.registry._statuses[receiver].fail_count += 1 self.preference.setdefault(sender, {}).setdefault(receiver, 0.0) self.preference[sender][receiver] = max(0.0, self.preference[sender][receiver] - 0.2) # ------- HTTP API 部分 ------- class CapabilityPayload(BaseModel): agent_id: str name: str description: str supported_intents: List[str] = [] timeout_seconds: int = 10 retry_count: int = 2 class ReachMessage(BaseModel): sender_id: str intent: str payload: Dict request_id: str = "" registry = ReachRegistry() router = IntentRouter(registry) @app.post("/register") def register_agent(cap: CapabilityPayload): """注册一个新的 Agent 能力卡""" new_cap = AgentCapability( agent_id=cap.agent_id, name=cap.name, description=cap.description, input_protocol={}, output_protocol={}, supported_intents=cap.supported_intents, timeout_seconds=cap.timeout_seconds, retry_count=cap.retry_count, ) registry.register(new_cap) return {"status": "registered", "agent_id": cap.agent_id} @app.post("/route") def route_message(msg: ReachMessage): """ 接收一条消息，路由到目标 Agent。 目标 Agent 不存在时，直接抛出降级事件，由业务侧决定处理。 """ start = time.time() target = router.route(msg.intent, msg.sender_id) if target is None: return { "status": "fallback_needed", "reason": "no_reachable_agent", "request_id": msg.request_id, "fallback_action": "enqueue", } latency_ms = int((time.time() - start) * 1000) router.record_success(msg.sender_id, target, latency_ms) # 实际消息投递：往 Redis Stream 的目标队列里 push # stream_key = f"agent:{target}:inbox" return { "status": "routed", "target_agent": target, "request_id": msg.request_id, "route_latency_ms": latency_ms, }

这段代码的核心思路有三个。第一，注册和状态是分离的，能力卡是静态配置，状态是动态指标，路由时两张表一起看。第二，健康分里面带“一票否决”，成功率低于50%的直接不选，不要浪费时间纠结，这在SNOW——这是我自己总结的优先级规则——里排第一。第三，偏好分只做增量更新，避免频繁切换接收方导致上下文断裂。

实际部署的时候，FastAPI跑在容器里，Redis Stream作为消息层，每个Agent独立进程订阅自己队列。路由服务本身无状态，可以水平扩展，压力全在Redis和PostgreSQL上。

3.3 参数调优记录：权重和超时的实测数据

参数配置这块，我把自己踩过的坑梳理成了几条有用的经验，每条都附带着实测数据。

语义权重和健康权重的平衡。最早我设定语义40%、健康40%、偏好20%并不是拍脑袋拍出来的，而是拿真实业务跑了一周做的对比。语义权重提到60%时，整体路由准确率看上去高了，但端到端时延P90从3.2秒涨到5.8秒，原因就是总把请求路由给同一个“听起来最合适”的Agent，导致它长期处于满载。把健康权重提到40%以后，请求在多个Agent之间分散，P90稳定在2.1秒左右，整体吞吐反而提升了一倍。

超时设置不要一刀切。不同类型Agent的合理超时差异极大。内部计算型Agent模型推理一次2-3秒，超时设5秒足够；调用外部API的Agent，考虑到网络波动和第三方响应慢，超时至少15秒；处理长文档解析的Agent，可能30秒以上。统一设10秒的结果是，内部计算型Agent经常白白等待，外部调用型Agent又频繁超时。

降级策略优先用“重新入队”。我们现在默认所有失败消息先进待处理队列，而不是直接丢弃或者通知人工。原因是大多数超时都是瞬时故障，等几秒后重新投递给同一个Agent，成功率很高。只有当重试次数超过上限后，才进入人工通知通道，保证重要业务不丢消息。

4. 踩坑实录与排查技巧

4.1 那些文档里不会写的坑

4.1.1 上下文膨胀导致Agent“忘了自己是谁”

这是一个特别隐蔽的问题。多Agent链路里，消息在传递过程中经常会把上游的中间结果原样拼到下游，比如监测Agent把抓到的50KB网页全文直接丢给分析Agent，分析Agent要先把这堆内容塞到自己的上下文里，再做摘要；摘要结果又连同原文一起传给写手Agent，一来二去，每个Agent的上下文窗口都在被无意义的历史消息占据。

这就造成了模型越来越容易忽略System Prompt里的指令，因为上下文太长了，注意力分散，微妙的指令全被淹没。我们的解法是引入了持久化的上下文压缩规则：每条消息在投递之前，都会先经过一层专门的压缩器，把大段原文压缩成结构化摘要，再附上原文的存储引用地址，需要精确引用时可以访问地址取回。这个改动让整体任务成功率从72%提到了88%，我觉得“上下文工程”这个词本身就该跟“提示词工程”并列，不能只写在提示词里。

4.1.2 循环调用让系统进入死循环

多Agent系统里，如果A调B、B发现缺数据又调C、C发现自己缺数据又调A，而且没有任何人设置跳数限制，消息就会被复制、转发、再复制。我们在一次压测里遇到过五千多个僵尸消息，全是在A、B、C三个Agent之间互相踢皮球。

后来加了一条硬性规则：每个请求都会附带一个跳数计数字段，每经过一个Agent就+1，超过五跳直接丢弃并发送告警。这条规则不放在Agent内部逻辑里，而是放在消息队列消费者端，一旦拦截到超跳消息，立刻打印整个调用链路径。这是排查循环依赖的最直观手段。

4.1.3 路由的“黑天鹅”误判

路由层用向量相似度打分，在绝大多数情况下是准确的，但偶尔会出现意图描述和Agent能力描述在语义上相似、实际上完全不对口的误判。有一次运营部门发消息说需要“生成一份周报”，结果这个词和“生成一张报表”的Agent描述相似度极高，把消息路由到了报表Agent，出来的产物完全不是周报的样子。

我们现在给路由加了一个兜底机制：每条消息在路由决策之后，接收方Agent会先做一个快速的意图校验，判断这条消息是否真的是自己该处理的。不匹配就立即返回一个“意图不匹配”的拒绝响应，让发送方重新路由到其他候选。这个机制把误判的恢复时间从小时级别缩短到了秒级别。

4.2 稳定运行的日常检查清单

做了一段时间之后，我总结出一份每天盯盘的检查清单，分享给你参考。

成功率趋势：监控每个Agent近一小时的成功率曲线，重点关注两类异常：一是断崖式下跌，通常是外部依赖挂了；二是缓慢下降，通常是上下文膨胀或者模型prompt悄悄漂移导致的。P95响应时延每一个Agent的P95时延如果持续走高，多半是队列堆积，这时候需要看队列深度而不是单纯加超时。

路由分布热力图：观察发送方Agent和接收方Agent的流量矩阵，发现有些Agent几乎收不到消息，说明能力卡描述和实际能力脱节了，需要重新梳理。反过来，发现某个Agent收到了大量不相关的意图请求，说明路由的语义权重需要调低。队列堆积量：Redis Stream里的每条队列积压量如果持续大于0，说明下游处理速度跟不上，这时候光加快一个Agent不够，要看是不是整条链路有瓶颈。

告警与重试记录：每天抽看一次降级策略触发记录。这类事件应该是异常的偶发情况，如果数量可观，说明路由或者超时配置需要重新调整。

这套体系跑稳之后，我最大的体会是：多Agent系统不是把Agent堆在一起就完事了，真正决定上限的是Agent能稳定触达多少东西，以及触达失败的恢复速度有多快。Agent-Reach这个名字，在我这里已经从项目代号变成了方法论，后续不管迁到什么模型底座上，这套触达治理的思路还能继续用。希望这篇内容，能帮你少走几段弯路。