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_idcap.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:i2], 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(keylambda x: x[0], reverseTrue) 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_idcap.agent_id, namecap.name, descriptioncap.description, input_protocol{}, output_protocol{}, supported_intentscap.supported_intents, timeout_secondscap.timeout_seconds, retry_countcap.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 fagent:{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这个名字在我这里已经从项目代号变成了方法论后续不管迁到什么模型底座上这套触达治理的思路还能继续用。希望这篇内容能帮你少走几段弯路。