如果你做过两个以上 Agent 的协同项目大概会有同感模型选型、Prompt 调优这些都只是入场券真正让人熬夜到凌晨三点的往往是“另一个 Agent 到底能不能被找到、调用起来、并且把结果按时送回来”。我给这个项目取名 Agent-Reach就是想集中解决一个词——触达。这里的触达不单指网络能不能连通而是指 Agent 之间从“知道对方存在”到“完成一次可靠的调用”的整条链路。我最早做多 Agent 协作时系统里只有三四个智能体彼此用 hardcode 的 API 地址互相调用当时觉得没什么问题。等到 Agent 数量过十、能力开始重叠、调用关系变成网状之后整个人工维护成本直接失控。Agent-Reach 是我在实践中反复重构后沉淀下来的一套触达层方案解决的是能力注册、语义发现、统一调用、可靠性兜底这几件事。这篇文章把完整设计思路、代码级拆解和踩坑过程都写出来适合正在做多 Agent 系统、想摆脱人肉编排的开发者参考。1. 多 Agent 项目的真正瓶颈不是模型智商而是“触达”1.1 我在第一个多 Agent 项目里踩的坑手工胶水代码先说我最早那个系统长什么样。当时做了一个内部研究助手里面有“论文检索 Agent”“数据清洗 Agent”“报告生成 Agent”三个角色。最初实现特别朴素主程序按顺序调 Agent A拿结果拼给 Agent B再拼给 Agent C。每个 Agent 之间都是硬编码的接口参数靠手工 mapping。这套东西在链路短的时候很稳可一旦需求变化就非常痛苦。用户新增了一个“图表生成 Agent”主程序就要改一版某个 Agent 因为后端服务迁移换了地址所有调用方都要跟着改更麻烦的是当你有两个 Agent 都能“读文件”的时候路由逻辑开始堆 if-else。三个智能体还好三十个智能体的时候那个 if-else 简直像一盘意大利面。听过一句话分布式系统的难点不是节点之间的网络连通而是节点之间的“语义连通”。我在多 Agent 项目里的体会一模一样。Agent 之间不是缺网络是缺一层“触达基础设施”谁来描述能力、谁来发现能力、谁来保证调用不出岔子、谁来记录每一次协作关系。Agent-Reach 就是冲着这个基础设施去的。1.2 把“触达”拆开看注册、发现、调用、回传Agent-Reach 里一次触达我分成四段注册Register、发现Discover、调用Invoke、回传Callback。很多人会忽略注册和发现这两步直接认为 Agent 触达就是“发起 HTTP 请求”。但如果你在系统里塞过几十个 Agent就会明白最难的其实不是调而是“知道该调谁”。注册每个 Agent 上线时把自己能干什么、输入输出长什么样、调用地址是什么、有没有额外限制写成一份结构化描述交给注册中心。发现调用方不直接写死目标地址而是带着“我要什么”来问注册中心。Agent-Reach 会把请求转成语义向量和能力描述做相似度匹配。调用命中结果后由触达网关统一转发负责鉴权、超时、重试、限流、幂等。回传把结果连同调用元数据耗时、消耗、状态返回给调用方。这套四段式的好处一眼就能看出来Agent 与 Agent 之间不再互相耦合所有关系都收敛到“注册中心 触达网关”这一层。新增一个 Agent 只需注册不需要去改其他调用方。我后来把 Agent-Reach 拆成独立的服务四个模块既可以整体跑也可以单独拆出去用。1.3 为什么叫 “Reach” 而不是 “Connect” 或 “Call”取名的时候我犹豫了一阵。Connect 强调建立连接Call 强调发起调用但 Agent 之间真正稀缺的是“够得着、够得稳、够得快”——这更像 Reach 这个词的含义能不能够到够到之后靠不靠谱。举个例子一个“日程管理 Agent”和一个“天气 Agent”都在线网络也通但这不代表前者能顺利拿到后者的结果。可能前者对能力描述理解偏了可能后者只接受 JSON 结构而前者发来的是自然语言也可能后者在 10 秒内没回来前者直接当失败了。这些都是触达问题。出发点和目的地之间有无数环节可能出岔子Agent-Reach 管的就是这一整段距离。2. Agent-Reach 的总体设计把 Agent 交互看成一次“触达会话”2.1 三个核心角色能力提供方、能力消费方、触达网关Agent-Reach 在设计上把系统分成了三个角色能力提供方Provider对外暴露某个能力的 Agent。它只负责在注册中心上报“我能做什么、怎么调我”不关心谁会调用。能力消费方Consumer需要某个能力的 Agent 或主流程。它只负责提交“我想要什么”不关心目标 Agent 部署在哪、用了什么模型。触达网关Gateway中间人做意图理解、路由匹配、调用转发、可靠性保障。它是 Agent-Reach 的核心所有触达流量都经过这里。这三个角色一定要从物理上分清楚不要揉在一个进程里。我之前偷懒把网关逻辑塞进消费方进程结果每次升级都要重启所有消费方。后来把网关单独部署成无状态服务消费方只连网关地址升级成本就变得很低。Agent-Reach 的部署形态很简单一个注册中心可复用外部存储、一个触达网关、若干 Provider Agent。消费方可以是 Agent也可以是传统服务。2.2 能力注册表的数据结构让机器和人都能读懂的 Agent 说明书注册表是整个系统的地基。Agent-Reach 里每个 Provider 上线时都要提交一份能力描述字段如下字段说明示例agent_idAgent 唯一标识agent-report-generatorname能力名称报告生成description能力详细说明用于语义匹配根据结构化数据生成 Markdown 格式的周报、月报或项目复盘报告input_schema入参 JSON Schema见下方代码endpoint调用的实际地址http://provider-agent:8000/invokeauth调用所需的鉴权信息引用token-reftimeout建议超时时间30srate_limit单日/每分钟调用额度100/min关键是 description 这一栏。一开始我觉得 description 写得越自然越好结果踩了大坑后面专门讲。input_schema 我直接用 JSON Schema好处是 Agent 的 Function Calling 协议天然兼容它校验也能直接用现成库。# Agent-Reach 能力注册示例简化 CAPABILITY_SCHEMA { type: object, properties: { data: { type: array, items: {type: object}, description: 结构化数据列表形式 }, format: { type: string, enum: [markdown, html, txt], description: 输出格式 }, topic: { type: string, description: 报告主题用于生成标题 } }, required: [data, format] }注册表底层我建议先用关系型数据库或 Redis 存元数据向量索引单独挂。能力注册量不大时Postgres 加一个 pgvector 就够了量大了再拆 Milvus 或 Qdrant。不要一开始就上重组件。2.3 触达链路的数据流从意图到结果的完整生命周期一次完整的触达在 Agent-Reach 里长这样消费方调用网关的 POST /reach 接口body 里带着一段意图描述和参数。网关先把意图转成 embedding去注册中心召回 Top-K 个候选能力。对候选能力做输入校验筛掉 schema 不匹配的。网关把参数格式化成目标 Provider 期望的结构发起调用。调用过程走统一的超时、重试、幂等逻辑。Provider 返回结果后网关把结果按统一 envelope 包好回传给消费方。这个流程的每一个环节都能单独观测。我把每次触达都记一条 trace包含路由命中的能力、匹配分、耗时、状态码、token 消耗。后续排查问题和做路由优化全靠这些 trace。3. Agent-Reach 核心模块的代码级拆解3.1 定义能力描述与入参校验先把“说明书”写好先给一个 Provider 做接入的完整例子。假设我有一个“天气查询 Agent”它能力很简单输入城市名输出天气。它在 Agent-Reach 里的注册描述长这样。# agent_reach/register.py from pydantic import BaseModel, Field class CapabilityRegister(BaseModel): agent_id: str Field(..., description全局唯一Agent标识) name: str Field(..., description能力名称尽量简洁) description: str Field(..., description能力描述用于语义路由) input_schema: dict Field(..., descriptionJSON Schema格式的入参说明) endpoint: str Field(..., description调用地址) auth_ref: str | None Field(None, description鉴权引用) timeout_seconds: int Field(30, description建议超时时间)入库前我会做两个校验第一input_schema 必须是合法的 JSON Schema保证 jsonschema 库能直接 validate第二endpoint 必须是内网白名单地址防止 Agent 能力被注册成外部任意 URL——这个安全习惯很重要我一个朋友把注册接口暴露到公网结果被人注册了一堆挖矿 Agent 地址网关变成开放代理。调用请求进来之后网关先按 schema 校验参数。这一步能拦截掉大量低级错误避免把坏请求转发给 Provider 浪费 token。# agent_reach/validate.py import jsonschema def validate_input(capability: dict, payload: dict) - None: schema capability[input_schema] jsonschema.validate(instancepayload, schemaschema)3.2 语义路由用向量相似度代替 if-else路由是 Agent-Reach 最核心的模块。我最初用的是规则路由每个消费方在请求里显式声明“我要调 agent-weather”。规则路由很可靠但问题在于调用方必须提前知道每个 Agent 的能力这又回到了人肉维护耦合的老路。后来我改成语义路由消费方只说“帮我看看北京今天适不适合跑步”网关自己去匹配“天气 Agent”和“健康建议 Agent”。实现上用了 embedding 相似度。这一步的工程细节不多真正的坑在 embedding 的选择和阈值设定。# agent_reach/router.py from openai import OpenAI import numpy as np client OpenAI() def embed(text: str) - list[float]: resp client.embeddings.create( modeltext-embedding-3-small, inputtext ) return resp.data[0].embedding def semantic_route(intent: str, candidates: list[dict], top_k: int 3): intent_vec embed(intent) scored [] for cap in candidates: desc_vec np.loads(cap[desc_vector]) # 注册时预计算好 score np.dot(intent_vec, desc_vec) / ( np.linalg.norm(intent_vec) * np.linalg.norm(desc_vec) ) scored.append((score, cap)) scored.sort(keylambda x: x[0], reverseTrue) return scored[:top_k]这里强调两点。第一Provider 的 description 向量要在注册时就算好存下来不要在路由时现算否则延迟会高到没法用。第二相似度只用来“召回候选”不能直接作为最终决策。我一般会再加一道规则层如果意图里出现明确的城市名就优先偏好在描述里声明支持城市的 Agent。语义召回 规则精排比单纯相似度靠谱得多。3.3 触达网关超时、重试、幂等、限流一锅端网关的核心代码其实就是一套标准中间件逻辑但每个点都有讲究。先说超时。HTTP 调用最忌讳“无限等”我统一用 asyncio.wait_for 包住调用超时时间从注册表里读取。Provider 说自己 30 秒内能返回网关就按 30 秒守门。重试要小心。不是所有请求都适合重试——比如“创建订单”这种非幂等操作重试会导致重复下单。Agent-Reach 的做法是网关在调用链路上给每次请求生成一个全局 request_idProvider 在实现时要支持幂等键。重试只会在请求根本没送达连接异常或 Provider 明确返回可重试错误码时发生业务失败绝不盲目重试。# agent_reach/gateway.py import asyncio import httpx import uuid async def invoke_with_policy(cap, payload): request_id str(uuid.uuid4()) headers {X-Request-Id: request_id} timeout httpx.Timeout(cap[timeout_seconds]) async with httpx.AsyncClient(timeouttimeout) as client: for attempt in range(3): try: resp await client.post(cap[endpoint], jsonpayload, headersheaders) if resp.status_code 429: await asyncio.sleep(1 * attempt) continue return wrap_result(resp, request_id) except httpx.ConnectError: await asyncio.sleep(0.5 * attempt) continue return wrap_error(timeout_or_unreachable, request_id)限流我用的是令牌桶每个 Provider 有独立的配额。配额设置不是拍脑袋依据是 Provider 的历史平均耗时和 token 消耗。网关在转发前先取令牌取不到就直接返回 429不给 Provider 添乱。3.4 调用日志与性能观测每次触达都留下证据没有观测的触达层等于盲飞。Agent-Reach 里我定义了一个标准的触达事件结构每次调用都会落 JSON 日志同时推一份到 Prometheus 指标{ request_id: uuid, consumer: agent-main, intent: 北京适不适合跑步, matched_capability: weather-query, match_score: 0.87, status: success, latency_ms: 812, tokens_used: 1520, timestamp: 2025-06-01T10:00:00Z }我实际用下来日志里最有价值的字段是 match_score 和 status。match_score 长期偏低的意图说明能力描述没写好或 Agent 缺失status 是 fail 的调用要看失败模式是超时、拒绝还是内容错误。这套观测体系跑了一周后我就发现了一个规律70% 的失败集中在两个 Agent 上而且都是因为 Provider 内部依赖了一个不稳定的第三方 API。后来给它们加了本地缓存成功率直接从 78% 拉到 96%。4. 真实环境里 Agent-Reach 翻过的车希望你不用再翻4.1 能力描述写得越“文艺”路由越容易翻车第一次做语义路由时我为了让 description 更像人话把它写成了营销文案风格。比如天气 Agent 的描述是“感知天空的情绪为你解读风云变幻”。结果就是用户问“明天会不会下雨”匹配分只有 0.61经常召回不到它但用户问“今天心情怎么样”它反而被召回了。这就是典型的“描述与查询语义错位”。Agent 的能力描述应该面向“意图检索”用具体的关键词和场景而不是文学表达。后来我把这套规则定为描述里必须包含能力类型、输入限定、输出形态、典型使用场景。现在天气 Agent 的描述是“根据城市名查询实时天气与未来三天预报返回温度、降水概率、风力等级适合出行和运动规划”。召回稳定性大幅提升。4.2 把长任务跑在同步 HTTP 里超时被打成筛子有一次我接入一个“行业研究报告生成 Agent”它处理一份报告要 5 分钟。我按默认配置把超时设成了 30 秒结果这个 Agent 的上线成功率只有 12%。所有调用都在 30 秒被网关掐断。 Provider 那边其实还在干活但结果已经没人等了白白烧掉大把模型费用。后来我在 Buyer 场景应该写“在消费场景”避免奇怪直接说在长任务场景里加入异步触达模式网关收到请求后先返回一个 task_idProvider 完成后通过回调接口把结果推回来。这相当于把 HTTP 同步模型改造成任务队列模型。同时把注册表里的 timeout_seconds 按任务类型分开设置实时查询类 10 秒生成类 300 秒以上拿不准的先跑一次压测再定。现在新 Agent 接入时我会强制要求负责人回答一个问题“这个能力是实时返回还是任务型返回”这个问题能过滤掉一半的超时事故。4.3 Agent 自己判断“成功了”但业务侧是失败的这是最隐蔽的一个坑。Agent 的返回里写了“查询成功”但业务侧拿到的其实是一段模板化的兜底话术而不是真实结果。比如天气 Agent 在内部 API 挂掉时没有抛异常而是返回了一句“当前天气信息暂时无法获取请您稍后再试”。从 HTTP 层面看200、正常返回网关判定触达成功日志也是绿色的。可对用户来说这就是一次失败的触达。我的解决办法是在能力注册表里增加一个 result_validation 字段Provider 必须声明自己什么情况下算“业务失败”并允许网关配置验证函数。最简单的方式是要求返回的 JSON 里带 success 字段和 code 字段网关在包装结果前先检查这层语义。凡是业务失败即便 HTTP 200也要按失败来记录和告警。4.4 限流参数抄别人的结果成本直接翻倍Agent-Reach 刚上线时我没有认真设置限流直接抄了一个在线服务默认参数每秒 100 次。结果有一个内部测试脚本在跑批量任务几秒内把某个重型生成 Agent 的 token 消耗干到了预估成本的 10 倍。后来我总结出限流设置的正确姿势先看 Provider 的单次均耗和单次 token 均耗再定单位时间内可接受的成本上限反推出速率。公式很简单——你一天只能承受 10 万 token 的成本单个任务平均 2000 token那每天最多 50 次调用均摊到小时就是 2 次左右。再在这个值上加一个 20% 的冗余就是实际的 rate_limit。不要把限流当成防攻击工具它是成本管理工具。Agent-Reach 的限流模块现在每个 Agent 独立配额、独立警报谁超了谁提醒而不是全系统一锅限。4.5 上下文污染一个失败的 Agent 污染了下游三个 Agent还有一次事故让我印象深刻。主 Agent 调用了“摘要 Agent”摘要 Agent 因为输入内容超过上下文限制返回了一段乱码。主 Agent 没有拦截直接把乱码拼进了自己的上下文又被下游三个 Agent 当作有效信息使用。最终结果是整条链路的输出都崩了。这类问题的根子在于Agent 之间的协作太像“传话筒”拿到结果就继续往下传没有校验和净化。Agent-Reach 里我加了一个上下文净化组件网关在把 Provider 的结果返回给消费方之前会先做一次基本健康检查是不是空、是否超长、是否包含明显错误标志不合格的结果会被标记成“low_quality”并在结果里附带警告。消费方可以决定是否继续使用但不能“装作没看见”。虽然这个组件不能保证每个结果都正确但至少避免了错误在链路里无声放大。5. 往深处用Agent-Reach 能做的三件进阶事5.1 把触达能力做成团队共享的基础设施Agent-Reach 跑稳之后我把它从个人项目升成了团队基础设施。团队里新来的同学接入一个新 Agent不用再关心“调用方怎么改”只要按注册表格式提交一份能力描述再写一个符合协议的回调入口其他人立刻就能发现并调用这个能力。这本质上是在团队内部建立了一套“能力市场”。我见过很多团队有极强的模型能力却因为 Agent 之间的信息孤岛出现大量重复造轮子。Agent-Reach 解决的核心问题不是性能而是“可发现性”——我知道团队里有人已经做了一个同类 Agent才不会再花两周重做一遍。这套逻辑越早引入越好等 Agent 数量上来了再补迁移成本很高。我建议所有做多 Agent 系统的团队即使不采用 Agent-Reach 的代码也要在开工前先回答一个问题“我们的 Agent 之间要建立起怎样的一层触达标准”这个标准和代码骨架比优化某个单独 Agent 的 Prompt 要重要得多。5.2 用历史触达记录做路由自学习路由模块匹配分偏低时传统做法是人肉调整 description。我在 Agent-Reach 上做了一层半自动优化把线上真实触达日志里“意图文本 最终命中能力 成功率”回流到一个微调集里。每隔一段时间用这批数据对 embedding 的重排序模型做一次轻量更新。简单说语义相似度负责召回重排序模型负责精排。重排序模型的输入是意图、候选能力描述、历史触达统计特征输出是最终排序。这个模型不需要很大一个 1 亿参数左右的轻量模型就能带来明显提升。没有能力训练模型的团队也可以用规则替代把最近一周 match_score 低于 0.6 却调用成功的意图抽出来自动给对应能力描述追加“别名关键词”。我跑了几轮相当于在做弱监督学习效果不输模型微调。这套机制的价值在于Agent 的能力是动态演进的今天描述得再准确下周可能就过时了。路由层必须有自己的“记忆”让每次触达的成功经验沉淀回系统里。5.3 给 Agent 加一层“可观测性外挂”很多 Agent 本身是黑盒你不知道它内部经历了什么。Agent-Reach 在触达层加了不少透明的观测点但只能看到外部行为。为了补内部视角我建议 Provider 侧至少在关键步骤吐三段事件开始处理、中间里程碑、最终结果。不要求结构化哪怕是三行日志都能帮助定位 80% 的触达问题。另一个容易忽略的点是元数据透传。网关发出的请求要在 header 里带上 request_id 和 parent_request_id这样一次多级触达能串成一条完整链路。排查问题的时候顺着 trace 走一遍立刻能看到哪一级 Agent 慢了、哪一级返回了异常。我在 Agent-Reach 里把 trace 的链路深度上限设成了 10防止 Agent 互相调用形成死循环。5.4 一些实际体会如果只能从这篇文章里带走一件事我会说是不要先堆 Agent先搭好触达层。模型能力再强Agent 之间连接不到、调不通、不可观测整体系统就是一团雾。Agent-Reach 不是银弹它只是把“Agent 之间发生一次可靠协作”这件事变得可设计、可复用、可排障。我后续还会持续迭代它的异步任务模型和安全隔离能力AirByte 并没有每次改造完我都会问自己一个问题现在再接入一个全新的 Agent我需要改多少旧代码这个数字越低说明触达层越合格。