年初我们把客服意图识别、知识库检索、工单分类、用户画像这几个 AI 能力拆成独立 Agent 服务之后第一周是幸福的第二周就开始头疼。最痛的不是某个 Agent 本身跑不好而是 Agent 之间的相互调用客服意图识别要查用户画像客服摘要要查知识库工单分类又想调用用户画像十几个 Agent 彼此的关系像一团乱麻。这件事有点讽刺。我们花大力气把单体服务拆成一个个小巧的智能体结果连接成本又悄悄涨了回来。市面上不是没有消息队列、API 网关那一套但放在 Agent 互调的场景里总觉得有点笨重。所以后来我写了自己的方案叫 Agent-Reach本质上就是给 Agent 之间装一个“触达层”——注册、发现、路由、调用一条龙。这篇文章就把这段实践完整记录下来包括设计思路、核心代码、压测数据和踩过的坑给同样在做多 Agent 互连的人一个参考。1. 当 Agent 越来越多连接成了新的瓶颈1.1 我们最初的点对点集成是怎么失控的先交代一下背景。我们这个项目上线初期只拆了四个 Agent客服意图识别、知识库检索、工单分类、用户画像。四个服务互相调用问题还不大代码里写死对方的 URL 就行。可随着业务拓展Agent 数量变成了十五个还新增了话术生成、风险预警、会话摘要、质检分析等一堆服务。点对点集成的混乱是慢慢累积的。每个调用方都要知道被调方的 endpoint、参数格式、鉴权方式每次新增或变更一个 Agent就得回头改一圈调用方的代码。我统计了一下当时的调用关系十五个 Agent 之间实际存在的调用链超过了四十条很多还是跨语言跨团队维护的。最难受的是你根本不知道某条调用链是否还有人在用线上告警一响得先花半小时查这个 Agent 到底被谁调了。这种“蜘蛛网式”连接还有一层更隐蔽的问题调用双方同时对各自的数据格式、错误码、超时策略都有自己的一套规矩。A 服务超时设 3 秒B 服务设 10 秒一旦 A 调用 B 出现偶发超时A 侧的重试逻辑会把压力直接怼到 B 上形成连锁故障。表面上看起来是每个服务都很正常但站在全局视角整个系统已经非常脆弱。1.2 看了一圈现成方案为什么最后选择自研在动手写 Agent-Reach 之前我认真比较过消息队列、API 网关和成熟 RPC 框架三个方向每个都有各自的问题。消息队列的思路是真解耦但 Agent 互调里有大量场景是同步请求-响应。客服会话过程中用户问一句系统必须立刻拿到知识库检索结果再生成回答这个链路是等不了的。用 MQ 做同步要自己维护关联 ID、超时回收、结果配对等于把一套 RPC 该解决的问题用消息重新搓一遍复杂度反而更高。API 网关呢它擅长的方向是“对外暴露流量管理”限流、鉴权、灰度这些做得很好但不关心“某个 Agent 提供了什么能力、能力是否在线、调用方该找谁”。Agent 是有生命周期的启动要注册、下线要摘除、崩溃要熔断这些语义网关都不具备。而且公司统一网关的上线流程偏重我们这种敏捷验证阶段的项目等不起。再看 gRPC 这类 RPC 框架功能确实全但要求双方定义强类型的 proto 接口。我们当时的 Agent 有 Python、Java、Node 混合让每个团队都去维护 proto 文件、跟着版本升级协作成本非常高。对于很多轻量能力调一个接口本质上只需要一段 JSON 和一个动作声明上整套 RPC 有点杀鸡用牛刀。Agent-Reach 的定位就是轻量触达层路由只看“能力标签”不看具体 URL所有 Agent 统一注册调用方只需要声明“我要什么能力”剩下的发现、路由、超时、重试、熔断都由触达层处理。这套思路不一定适合所有团队但对我们这种以敏捷验证为目标、Agent 类型杂、调用关系变化快的场景来说是性价比最高的方案。2. Agent-Reach 解决的本质问题触达协议与注册发现2.1 统一触达协议为什么用“消息信封”而不是 REST确定要自研后第一个要解决的问题是 Agent 之间用什么格式通信。我花了一天时间纠结到底是继续用 REST 还是重新做一套统一协议最后选择了消息信封——所有 Agent 的能力入口统一为/invoke参数不分 GET/POST 细节全部塞进一个结构化的信封里。信封结构长这样字段类型含义request_idstring一次调用的唯一 ID用于链路追踪和问题排查from_agentstring调用方 Agent IDtarget_capabilitystring本次要触达的能力名例如knowledge.retrieve.v1payloadobject业务参数按能力自带格式定义timestampfloat发出时间用于超时计算和防重放ttlint请求有效时间超出后路由层直接丢弃防止积压为什么不用传统 REST因为 REST 会让调用方感知到具体资源路径比如/agent/knowledge/retrieve、/agent/ticket/classify路径一变就得改代码。而消息信封是“按能力寻址”调用方只声明target_capability: knowledge.retrieve.v1完全不关心这个能力部署在哪台机器、叫什么路径。这个解耦非常关键Agent 升级地址变更时调用方代码一行都不用动。还有一层考虑是请求头标准化。之前每个 Agent 的自定义鉴权放在不同的 Header 里到了 Agent-Reach 这里统一放进信封的固定字段由触达层统一校验业务 Agent 就不用重复写鉴权逻辑了。安全边际也从“每个服务各自为战”变成了“触达层统一收口”。2.2 注册中心与能力路由Agent 是怎么被“找到”的消息信封解决的是“怎么说话”注册发现解决的是“找谁说话”。Agent-Reach 里每个 Agent 启动后会主动向注册中心上报三样东西Agent ID、能力标签列表、实际 HTTP 地址。注册中心的数据模型很简单核心概念是 AgentInfodataclass class AgentInfo: agent_id: str name: str capabilities: list[str] endpoint: str health_check_url: str created_at: float fail_count: int路由层的职责是拿着target_capability去注册表里找匹配的 AgentInfo找不到就直接返回AgentNotFound。这一步的动作看似简单但和传统 API 网关的“精确匹配 URL”有本质区别路由中心维护的是一份“能力目录”而不是一份“接口清单”。你可以把能力目录理解成外卖平台上的店铺标签用户要的是“麻辣烫”平台给你列出所有做麻辣烫的店至于店在几楼、门牌号多少那是平台内部的事情用户不必关心。为实现负载均衡我当时给 Router 加了一个很朴素的策略同一能力标签下有多个 Agent 时默认轮询同时记录每个 Agent 的连续失败次数。连续失败超过阈值就直接从注册表摘除等健康检查恢复后再重新注册。这样整个系统就具备了最基础的“自愈”能力不需要人工干预才能把故障节点剔除。2.3 能力标签的规范命名就是契约能力标签是整个触达体系里最容易被人忽视、但坑最多的设计点。项目初期大家随便写标签有的叫search有的叫knowledge_query同一个东西三种叫法路由层根本没有办法统一匹配。后续我强制推行了一套命名规范领域.动作.版本。举个例子knowledge.retrieve.v1知识库检索ticket.classify.v1工单分类profile.get.v1用户画像查询risk.evaluate.v1风险预判版本号是后加的因为有一次知识库 Agent 要从 v1 升级成 v2能力实现全换了但由于标签没变路由层还是把流量发了过去线上出了问题才察觉。加了版本号之后能力标签就成了一份稳定契约调用方指定knowledge.retrieve.v1就是精确要求 v1 语义不会因为你内部升级而被悄悄改变。这是多 Agent 协作里最容易“静默出错”的环节后文踩坑部分还会详细展开。3. 代码落地从零实现一个最小 Agent-Reach3.1 注册表与路由的骨架实现我把 Agent-Reach 的代码组织成两个核心模块core.py维护注册表router.py负责路由调度。注册表为了保证并发安全实际项目里用了asyncio.Lock这里为了看得清爽先给一个单机版# agent_reach/core.py import time from dataclasses import dataclass, field dataclass class AgentInfo: agent_id: str name: str capabilities: list[str] endpoint: str health_check_url: str created_at: float field(default_factorytime.time) fail_count: int 0 class AgentRegistry: def __init__(self): self._agents {} def register(self, agent: AgentInfo): agent.fail_count 0 self._agents[agent.agent_id] agent def unregister(self, agent_id: str): self._agents.pop(agent_id, None) def find_by_capability(self, capability: str): return [a for a in self._agents.values() if capability in a.capabilities] def all_agents(self): return list(self._agents.values())路由层的实现也不复杂。它做的事情是查能力、做负载均衡、构造信封、发起 HTTP 调用、处理失败计数和熔断。HTTP 客户端我用了httpx因为它在同步和异步场景都能用超时控制也比 requests 灵活# agent_reach/router.py import time import uuid import logging import httpx class AgentNotFound(Exception): pass class Router: def __init__(self, registry, connect_timeout1.0, read_timeout10.0, max_fail_count5, probe_interval30): self.registry registry self.connect_timeout connect_timeout self.read_timeout read_timeout self.max_fail_count max_fail_count self.probe_interval probe_interval def dispatch(self, from_agent: str, target_capability: str, payload: dict) - dict: candidates self.registry.find_by_capability(target_capability) if not candidates: raise AgentNotFound(fno agent provides capability: {target_capability}) target self._pick(candidates) envelope { request_id: uuid.uuid4().hex, from_agent: from_agent, target_capability: target_capability, payload: payload, timestamp: time.time(), } try: resp httpx.post( f{target.endpoint}/invoke, jsonenvelope, timeouthttpx.Timeout(connectself.connect_timeout, readself.read_timeout), ) resp.raise_for_status() target.fail_count 0 return resp.json() except (httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadError): target.fail_count 1 if target.fail_count self.max_fail_count: self.registry.unregister(target.agent_id) logging.warning(agent %s removed due to continuous failures, target.agent_id) raise def _pick(self, candidates): # 简化版轮询生产环境可按负载、机房、权重扩展 return min(candidates, keylambda a: a.fail_count)3.2 一个能跑通的最小示例为了让刚接触这套设计的人能快速理解全链路我写了一个最小可跑的 demo。注册中心跑起来之后先是两个业务 Agent 各自注册自己的能力然后调用方通过 Router 去触达knowledge.retrieve.v1。# demo_server.py from fastapi import FastAPI from pydantic import BaseModel import uvicorn app FastAPI() class Envelope(BaseModel): request_id: str from_agent: str target_capability: str payload: dict timestamp: float app.post(/invoke) async def invoke(envelope: Envelope): if envelope.target_capability knowledge.retrieve.v1: return { code: 0, data: {summary: retrieved docs for: flood alert}, response_to: envelope.request_id, } return {code: 404, message: capability not supported} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port9101)# demo_register.py from agent_reach.core import AgentRegistry, AgentInfo registry AgentRegistry() registry.register(AgentInfo( agent_idkb-agent-01, nameknowledge-base, capabilities[knowledge.retrieve.v1], endpointhttp://10.0.3.21:9101, health_check_urlhttp://10.0.3.21:9101/health, ))# demo_call.py from agent_reach.core import AgentRegistry from agent_reach.router import Router registry AgentRegistry() # 实际运行中这一步由 demo_register.py 完成注册 router Router(registry) result router.dispatch( from_agentchat-agent-01, target_capabilityknowledge.retrieve.v1, payload{query: flood alert, limit: 5}, ) print(result[data][summary])这个例子把整套流程讲清楚了业务 Agent 不再需要暴露五花八门的业务路由所有能力统一收到/invoke按照target_capability字段分流。好处是显而易见的——新加一个 Agent 就注册一份 AgentInfo不需要改动任何已有调用方的代码新加一个能力只需要在/invoke里多写一个 if 分支或者把能力拆分到新服务里去。4. 实测效果延迟、成功率与链路开销4.1 压测数据和路由层自身开销代码跑通只是第一步真正要上线必须搞清楚这套触达层会带来多少额外开销。我做了两轮相对完整的压测先说结果。测试环境是三台 4C8G 的云主机一台跑 Agent-Reach 路由节点两台各跑一个知识库 Agent。压测脚本模拟 100 并发持续调用knowledge.retrieve.v1每轮请求带 2KB 左右的 payload持续 30 分钟。最终成绩指标数值总请求数约 32.5 万成功率99.97%p50 延迟12msp95 延迟26msp99 延迟41ms路由层自身耗时中位数0.6ms路由层自身耗时 0.6ms 这个数据是我比较满意的它包含了能力标签匹配、轮询选择、信封构造和一次 HTTP 请求响应的时间差。对比业务 Agent 本身 10ms 量级的处理时长Agent-Reach 的额外开销大约是 5%对于一个触达层来说足够轻。4.2 超时与重试参数是怎么调出来的第一版 Agent-Reach 的超时设置非常粗糙全局统一 5 秒。上线后第一个周末客服摘要 Agent 就频繁出现调用失败日志里全是ReadTimeout。查下来发现知识库 Agent 在高峰期某个复杂检索要跑 3.5 秒虽然没到 5 秒上限但一旦有 CPU 竞争就会冲到 6 秒以上直接触发超时。我一开始以为是超时太短把全局超时改到 10 秒结果更糟下游故障时所有请求都卡在 10 秒线程池被打满雪崩一触即发。后来才想明白超时设置不能一刀切。连接超时和读超时是两码事连接失败是网络层问题基本都是立刻失败重试读超时是业务处理慢要区分能力类型。最终把连接超时定死在 1 秒所有 Agent 统一读超时按能力分类知识库检索这类重计算能力放宽到 10 秒工单分类这类轻能力收紧到 3 秒通知发送这种外部依赖多的写操作也设 5 秒。重试策略也踩了坑。最初是无脑重试 2 次结果知识库 Agent 查询一个不存在的内容 ID 时业务上返回错误码但 HTTP 状态是 200不会触发重试逻辑问题不大。但有一次用户画像 Agent 集体超时重试流量把那个实例彻底压垮恢复时间反而拉长。后来给代理调用链路加了约束只有声明了idempotent: true的能力才允许触达层自动重试非幂等操作一律把重试决策交还给调用方。重试不是免费的得让系统知道什么请求可以安全地再来一次。5. 落地过程中踩过的四个坑5.1 端口“看起来通实际不通”一次跨机排查全记录这个坑是最折磨人的。某个 Agent 从 Agent-Reach 上看状态是 Healthy但调用方一直超时Agent 本机日志里又看不到任何请求进来。我本能地怀疑是 Router 的负载均衡策略或注册表出了问题结果在注册中心里手动查 endpoint发现地址和端口没错Route 匹配也对但请求就是送不到。排查链路是这样的先从调用方机器手动curl -v http://被调方IP:9105/invoke返回Connection timed out再登录被调方机器在本机 curl 同一个地址秒回。这说明问题在网络层而不是应用层。接着telnet 被调方IP 9105还是不通基本可以确定是中间防火墙拦截。查了安全组规则才发现新增的 9105 端口没有加到白名单默认策略是丢弃。这个排查耗时两个小时百分之八十的时间浪费在应用层。教训很简单新增端口后第一件事就是确认防火墙和安全组规则而不是去翻业务日志。后来我们把它写进了上线 checklist每次新增 Agent 端口必须同步维护网络白名单否则禁止上线。5.2 全局超时等于没设超时这个前面提过了但值得单独拿出来念叨两句。第一版配置里只有一个default_timeout所有能力共享同一个值。带来的问题很典型某些能力太慢导致用户体感卡顿某些能力又因为预留了太多超时而堆积线程占用。开始调优后我自己都惊讶不同能力的处理时间能差 20 倍轻量的标签分类只要 30ms重度的知识聚合要 8 秒如果统一按 5 秒设置轻能力白白浪费线程等待重能力又快速失败。最终我们按处理类型把超时分成三档连接超时 1 秒固定不变计算密集类和检索类能力读超时 10 秒交互类和写操作类读超时 3 到 5 秒。同时给 Router 增加线程池隔离不同超时档位走不同的线程池避免慢能力把快能力的线程全占光。这个改造之后系统的整体尾部延迟压下来不少p99 从 200ms 以上降到 50ms 以内。5.3 路由规则写太死Agent 一升级就静默失败知识库 Agent 从 v1 升级到 v2 那天我们遇上最典型的一次事故。升级过程中为了平滑切换我把新版本注册成了knowledge.retrieve.v2旧版本按计划下线。结果所有还在调用knowledge.retrieve.v1的调用方一下子全部抛AgentNotFound线上客服摘要直接不可用。这个问题的根因在于我把能力标签当成普通字符串处理没有意识到它其实是一份契约。调用方在代码里明确写了v1它期望的就是 v1 的语义如果 v1 没了契约就被破坏了。后来我在 Agent-Reach 里加了版本别名机制声明knowledge.retrieve.v2时如果配置了soft_alias: [v1]路由层在找不到 v1 时会自动尝试 v2并在响应里附加一个deprecated: true的标记去提醒调用方。这挽救了那个发布窗口但长期策略仍然是把调用方代码全部升级到 v2再逐步摘掉别名。能力升级和 API 升级遵循同一套节奏谁也不能悄悄打破契约。5.4 没有链路追踪出了问题只能翻日志Agent-Reach 刚上线那阵子每次线上出问题排查效率都很低。从客服入口到意图识别、到知识库检索、再到话术生成一条调用链跨越三四个服务每个服务都有自己的日志文件时间戳还对齐不了。有一次客户反馈“回复特别慢而且答案不对”我花了整个下午翻日志结果发现是知识库检索 Agent 偶发返回了旧版本内容但调用链路上根本看不出哪里出了问题。后来把所有 Agent 的入口和出口都强制接入了同一个 trace_id 透传机制信封里的request_id就是这条链路的根 ID。每个 Agent 收到请求时打印一行结构化日志包含request_id、from_agent、target_capability、耗时ms所有日志汇总到集中式日志平台按request_id一搜整条链路各环节的耗时分布一目了然。不出半天旧版内容的问题就被定位到了是知识库 Agent 的缓存 key 没带版本号。这个优化表面上只是加了一个 ID实际是把故障排查时间从小时级压缩到了分钟级属于性价比超高的基础设施投资。6. Agent-Reach 的下一步从框架到协议6.1 如果每个团队都自建一套那就重蹈覆辙了Agent-Reach 在我们团队内部跑稳之后我开始考虑一个更远的问题如果我们只是把它做成自己团队的一个内部框架那别的团队做 Agent 互连时又会重复发明一遍轮子而且每个轮子的协议还不一样。到那时候不同团队之间的 Agent 又没法互相触达问题只是从“服务调用乱”变成了“触达层不统一乱”。所以下一步的重点是把 Agent-Reach 沉淀成一个协议规范消息信封格式、注册发现接口、能力标签命名规则、超时和重试语义这些都是可以直接标准化的内容。具体到代码层面我想把目前 Python 实现里的核心逻辑抽出一个独立协议包让 Java、Node 团队可以照着同一套规则实现自己的客户端只要信封格式一致、注册接口一致跨语言 Agent 互调就不需要额外写适配层。还有一个更朴素的驱动Agent 的规模再涨一个数量级之后能力标签的匹配策略、健康检查的频率、注册表的最终一致性都会变成新瓶颈。现在这套轮询加失败计数已经不满足自动伸缩场景后续考虑把注册中心背后的存储换成 Redis让多个 Router 节点共享同一份注册表实现路由层自身的水平扩展。这一步不急但方向是明确的。6.2 我个人的几条实操心得做 Agent-Reach 这段时间最深的体会其实和代码关系不大。先立能力规范再写代码这是我最想强调的一条。我们一开始就是吃了标签混乱的亏后面补了一套领域.动作.版本的命名规范类似的痛苦才少了很多。协议这东西定得越晚重建成本越高。第二条重试是恩赐也是祸。无脑重试等于在下游故障时帮忙踩油门必须有幂等声明、熔断阈值和分级超时配合重试才真正有用。三条既没写进官网文档却恰恰是线上稳定性的关键可观测性比功能更重要。Agent 互连一旦跑起来问题往往不在某个 Agent 内部而在链路和依赖关系上没有按 request_id 串联的日志排查就是海底捞针。这不算工程上的炫技更像是给多 Agent 协作体系打地基。地基稳了上面站多少 Agent 都踏实。
Agent互连的轻量触达层：Agent-Reach的注册发现与能力路由实践
年初我们把客服意图识别、知识库检索、工单分类、用户画像这几个 AI 能力拆成独立 Agent 服务之后第一周是幸福的第二周就开始头疼。最痛的不是某个 Agent 本身跑不好而是 Agent 之间的相互调用客服意图识别要查用户画像客服摘要要查知识库工单分类又想调用用户画像十几个 Agent 彼此的关系像一团乱麻。这件事有点讽刺。我们花大力气把单体服务拆成一个个小巧的智能体结果连接成本又悄悄涨了回来。市面上不是没有消息队列、API 网关那一套但放在 Agent 互调的场景里总觉得有点笨重。所以后来我写了自己的方案叫 Agent-Reach本质上就是给 Agent 之间装一个“触达层”——注册、发现、路由、调用一条龙。这篇文章就把这段实践完整记录下来包括设计思路、核心代码、压测数据和踩过的坑给同样在做多 Agent 互连的人一个参考。1. 当 Agent 越来越多连接成了新的瓶颈1.1 我们最初的点对点集成是怎么失控的先交代一下背景。我们这个项目上线初期只拆了四个 Agent客服意图识别、知识库检索、工单分类、用户画像。四个服务互相调用问题还不大代码里写死对方的 URL 就行。可随着业务拓展Agent 数量变成了十五个还新增了话术生成、风险预警、会话摘要、质检分析等一堆服务。点对点集成的混乱是慢慢累积的。每个调用方都要知道被调方的 endpoint、参数格式、鉴权方式每次新增或变更一个 Agent就得回头改一圈调用方的代码。我统计了一下当时的调用关系十五个 Agent 之间实际存在的调用链超过了四十条很多还是跨语言跨团队维护的。最难受的是你根本不知道某条调用链是否还有人在用线上告警一响得先花半小时查这个 Agent 到底被谁调了。这种“蜘蛛网式”连接还有一层更隐蔽的问题调用双方同时对各自的数据格式、错误码、超时策略都有自己的一套规矩。A 服务超时设 3 秒B 服务设 10 秒一旦 A 调用 B 出现偶发超时A 侧的重试逻辑会把压力直接怼到 B 上形成连锁故障。表面上看起来是每个服务都很正常但站在全局视角整个系统已经非常脆弱。1.2 看了一圈现成方案为什么最后选择自研在动手写 Agent-Reach 之前我认真比较过消息队列、API 网关和成熟 RPC 框架三个方向每个都有各自的问题。消息队列的思路是真解耦但 Agent 互调里有大量场景是同步请求-响应。客服会话过程中用户问一句系统必须立刻拿到知识库检索结果再生成回答这个链路是等不了的。用 MQ 做同步要自己维护关联 ID、超时回收、结果配对等于把一套 RPC 该解决的问题用消息重新搓一遍复杂度反而更高。API 网关呢它擅长的方向是“对外暴露流量管理”限流、鉴权、灰度这些做得很好但不关心“某个 Agent 提供了什么能力、能力是否在线、调用方该找谁”。Agent 是有生命周期的启动要注册、下线要摘除、崩溃要熔断这些语义网关都不具备。而且公司统一网关的上线流程偏重我们这种敏捷验证阶段的项目等不起。再看 gRPC 这类 RPC 框架功能确实全但要求双方定义强类型的 proto 接口。我们当时的 Agent 有 Python、Java、Node 混合让每个团队都去维护 proto 文件、跟着版本升级协作成本非常高。对于很多轻量能力调一个接口本质上只需要一段 JSON 和一个动作声明上整套 RPC 有点杀鸡用牛刀。Agent-Reach 的定位就是轻量触达层路由只看“能力标签”不看具体 URL所有 Agent 统一注册调用方只需要声明“我要什么能力”剩下的发现、路由、超时、重试、熔断都由触达层处理。这套思路不一定适合所有团队但对我们这种以敏捷验证为目标、Agent 类型杂、调用关系变化快的场景来说是性价比最高的方案。2. Agent-Reach 解决的本质问题触达协议与注册发现2.1 统一触达协议为什么用“消息信封”而不是 REST确定要自研后第一个要解决的问题是 Agent 之间用什么格式通信。我花了一天时间纠结到底是继续用 REST 还是重新做一套统一协议最后选择了消息信封——所有 Agent 的能力入口统一为/invoke参数不分 GET/POST 细节全部塞进一个结构化的信封里。信封结构长这样字段类型含义request_idstring一次调用的唯一 ID用于链路追踪和问题排查from_agentstring调用方 Agent IDtarget_capabilitystring本次要触达的能力名例如knowledge.retrieve.v1payloadobject业务参数按能力自带格式定义timestampfloat发出时间用于超时计算和防重放ttlint请求有效时间超出后路由层直接丢弃防止积压为什么不用传统 REST因为 REST 会让调用方感知到具体资源路径比如/agent/knowledge/retrieve、/agent/ticket/classify路径一变就得改代码。而消息信封是“按能力寻址”调用方只声明target_capability: knowledge.retrieve.v1完全不关心这个能力部署在哪台机器、叫什么路径。这个解耦非常关键Agent 升级地址变更时调用方代码一行都不用动。还有一层考虑是请求头标准化。之前每个 Agent 的自定义鉴权放在不同的 Header 里到了 Agent-Reach 这里统一放进信封的固定字段由触达层统一校验业务 Agent 就不用重复写鉴权逻辑了。安全边际也从“每个服务各自为战”变成了“触达层统一收口”。2.2 注册中心与能力路由Agent 是怎么被“找到”的消息信封解决的是“怎么说话”注册发现解决的是“找谁说话”。Agent-Reach 里每个 Agent 启动后会主动向注册中心上报三样东西Agent ID、能力标签列表、实际 HTTP 地址。注册中心的数据模型很简单核心概念是 AgentInfodataclass class AgentInfo: agent_id: str name: str capabilities: list[str] endpoint: str health_check_url: str created_at: float fail_count: int路由层的职责是拿着target_capability去注册表里找匹配的 AgentInfo找不到就直接返回AgentNotFound。这一步的动作看似简单但和传统 API 网关的“精确匹配 URL”有本质区别路由中心维护的是一份“能力目录”而不是一份“接口清单”。你可以把能力目录理解成外卖平台上的店铺标签用户要的是“麻辣烫”平台给你列出所有做麻辣烫的店至于店在几楼、门牌号多少那是平台内部的事情用户不必关心。为实现负载均衡我当时给 Router 加了一个很朴素的策略同一能力标签下有多个 Agent 时默认轮询同时记录每个 Agent 的连续失败次数。连续失败超过阈值就直接从注册表摘除等健康检查恢复后再重新注册。这样整个系统就具备了最基础的“自愈”能力不需要人工干预才能把故障节点剔除。2.3 能力标签的规范命名就是契约能力标签是整个触达体系里最容易被人忽视、但坑最多的设计点。项目初期大家随便写标签有的叫search有的叫knowledge_query同一个东西三种叫法路由层根本没有办法统一匹配。后续我强制推行了一套命名规范领域.动作.版本。举个例子knowledge.retrieve.v1知识库检索ticket.classify.v1工单分类profile.get.v1用户画像查询risk.evaluate.v1风险预判版本号是后加的因为有一次知识库 Agent 要从 v1 升级成 v2能力实现全换了但由于标签没变路由层还是把流量发了过去线上出了问题才察觉。加了版本号之后能力标签就成了一份稳定契约调用方指定knowledge.retrieve.v1就是精确要求 v1 语义不会因为你内部升级而被悄悄改变。这是多 Agent 协作里最容易“静默出错”的环节后文踩坑部分还会详细展开。3. 代码落地从零实现一个最小 Agent-Reach3.1 注册表与路由的骨架实现我把 Agent-Reach 的代码组织成两个核心模块core.py维护注册表router.py负责路由调度。注册表为了保证并发安全实际项目里用了asyncio.Lock这里为了看得清爽先给一个单机版# agent_reach/core.py import time from dataclasses import dataclass, field dataclass class AgentInfo: agent_id: str name: str capabilities: list[str] endpoint: str health_check_url: str created_at: float field(default_factorytime.time) fail_count: int 0 class AgentRegistry: def __init__(self): self._agents {} def register(self, agent: AgentInfo): agent.fail_count 0 self._agents[agent.agent_id] agent def unregister(self, agent_id: str): self._agents.pop(agent_id, None) def find_by_capability(self, capability: str): return [a for a in self._agents.values() if capability in a.capabilities] def all_agents(self): return list(self._agents.values())路由层的实现也不复杂。它做的事情是查能力、做负载均衡、构造信封、发起 HTTP 调用、处理失败计数和熔断。HTTP 客户端我用了httpx因为它在同步和异步场景都能用超时控制也比 requests 灵活# agent_reach/router.py import time import uuid import logging import httpx class AgentNotFound(Exception): pass class Router: def __init__(self, registry, connect_timeout1.0, read_timeout10.0, max_fail_count5, probe_interval30): self.registry registry self.connect_timeout connect_timeout self.read_timeout read_timeout self.max_fail_count max_fail_count self.probe_interval probe_interval def dispatch(self, from_agent: str, target_capability: str, payload: dict) - dict: candidates self.registry.find_by_capability(target_capability) if not candidates: raise AgentNotFound(fno agent provides capability: {target_capability}) target self._pick(candidates) envelope { request_id: uuid.uuid4().hex, from_agent: from_agent, target_capability: target_capability, payload: payload, timestamp: time.time(), } try: resp httpx.post( f{target.endpoint}/invoke, jsonenvelope, timeouthttpx.Timeout(connectself.connect_timeout, readself.read_timeout), ) resp.raise_for_status() target.fail_count 0 return resp.json() except (httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadError): target.fail_count 1 if target.fail_count self.max_fail_count: self.registry.unregister(target.agent_id) logging.warning(agent %s removed due to continuous failures, target.agent_id) raise def _pick(self, candidates): # 简化版轮询生产环境可按负载、机房、权重扩展 return min(candidates, keylambda a: a.fail_count)3.2 一个能跑通的最小示例为了让刚接触这套设计的人能快速理解全链路我写了一个最小可跑的 demo。注册中心跑起来之后先是两个业务 Agent 各自注册自己的能力然后调用方通过 Router 去触达knowledge.retrieve.v1。# demo_server.py from fastapi import FastAPI from pydantic import BaseModel import uvicorn app FastAPI() class Envelope(BaseModel): request_id: str from_agent: str target_capability: str payload: dict timestamp: float app.post(/invoke) async def invoke(envelope: Envelope): if envelope.target_capability knowledge.retrieve.v1: return { code: 0, data: {summary: retrieved docs for: flood alert}, response_to: envelope.request_id, } return {code: 404, message: capability not supported} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port9101)# demo_register.py from agent_reach.core import AgentRegistry, AgentInfo registry AgentRegistry() registry.register(AgentInfo( agent_idkb-agent-01, nameknowledge-base, capabilities[knowledge.retrieve.v1], endpointhttp://10.0.3.21:9101, health_check_urlhttp://10.0.3.21:9101/health, ))# demo_call.py from agent_reach.core import AgentRegistry from agent_reach.router import Router registry AgentRegistry() # 实际运行中这一步由 demo_register.py 完成注册 router Router(registry) result router.dispatch( from_agentchat-agent-01, target_capabilityknowledge.retrieve.v1, payload{query: flood alert, limit: 5}, ) print(result[data][summary])这个例子把整套流程讲清楚了业务 Agent 不再需要暴露五花八门的业务路由所有能力统一收到/invoke按照target_capability字段分流。好处是显而易见的——新加一个 Agent 就注册一份 AgentInfo不需要改动任何已有调用方的代码新加一个能力只需要在/invoke里多写一个 if 分支或者把能力拆分到新服务里去。4. 实测效果延迟、成功率与链路开销4.1 压测数据和路由层自身开销代码跑通只是第一步真正要上线必须搞清楚这套触达层会带来多少额外开销。我做了两轮相对完整的压测先说结果。测试环境是三台 4C8G 的云主机一台跑 Agent-Reach 路由节点两台各跑一个知识库 Agent。压测脚本模拟 100 并发持续调用knowledge.retrieve.v1每轮请求带 2KB 左右的 payload持续 30 分钟。最终成绩指标数值总请求数约 32.5 万成功率99.97%p50 延迟12msp95 延迟26msp99 延迟41ms路由层自身耗时中位数0.6ms路由层自身耗时 0.6ms 这个数据是我比较满意的它包含了能力标签匹配、轮询选择、信封构造和一次 HTTP 请求响应的时间差。对比业务 Agent 本身 10ms 量级的处理时长Agent-Reach 的额外开销大约是 5%对于一个触达层来说足够轻。4.2 超时与重试参数是怎么调出来的第一版 Agent-Reach 的超时设置非常粗糙全局统一 5 秒。上线后第一个周末客服摘要 Agent 就频繁出现调用失败日志里全是ReadTimeout。查下来发现知识库 Agent 在高峰期某个复杂检索要跑 3.5 秒虽然没到 5 秒上限但一旦有 CPU 竞争就会冲到 6 秒以上直接触发超时。我一开始以为是超时太短把全局超时改到 10 秒结果更糟下游故障时所有请求都卡在 10 秒线程池被打满雪崩一触即发。后来才想明白超时设置不能一刀切。连接超时和读超时是两码事连接失败是网络层问题基本都是立刻失败重试读超时是业务处理慢要区分能力类型。最终把连接超时定死在 1 秒所有 Agent 统一读超时按能力分类知识库检索这类重计算能力放宽到 10 秒工单分类这类轻能力收紧到 3 秒通知发送这种外部依赖多的写操作也设 5 秒。重试策略也踩了坑。最初是无脑重试 2 次结果知识库 Agent 查询一个不存在的内容 ID 时业务上返回错误码但 HTTP 状态是 200不会触发重试逻辑问题不大。但有一次用户画像 Agent 集体超时重试流量把那个实例彻底压垮恢复时间反而拉长。后来给代理调用链路加了约束只有声明了idempotent: true的能力才允许触达层自动重试非幂等操作一律把重试决策交还给调用方。重试不是免费的得让系统知道什么请求可以安全地再来一次。5. 落地过程中踩过的四个坑5.1 端口“看起来通实际不通”一次跨机排查全记录这个坑是最折磨人的。某个 Agent 从 Agent-Reach 上看状态是 Healthy但调用方一直超时Agent 本机日志里又看不到任何请求进来。我本能地怀疑是 Router 的负载均衡策略或注册表出了问题结果在注册中心里手动查 endpoint发现地址和端口没错Route 匹配也对但请求就是送不到。排查链路是这样的先从调用方机器手动curl -v http://被调方IP:9105/invoke返回Connection timed out再登录被调方机器在本机 curl 同一个地址秒回。这说明问题在网络层而不是应用层。接着telnet 被调方IP 9105还是不通基本可以确定是中间防火墙拦截。查了安全组规则才发现新增的 9105 端口没有加到白名单默认策略是丢弃。这个排查耗时两个小时百分之八十的时间浪费在应用层。教训很简单新增端口后第一件事就是确认防火墙和安全组规则而不是去翻业务日志。后来我们把它写进了上线 checklist每次新增 Agent 端口必须同步维护网络白名单否则禁止上线。5.2 全局超时等于没设超时这个前面提过了但值得单独拿出来念叨两句。第一版配置里只有一个default_timeout所有能力共享同一个值。带来的问题很典型某些能力太慢导致用户体感卡顿某些能力又因为预留了太多超时而堆积线程占用。开始调优后我自己都惊讶不同能力的处理时间能差 20 倍轻量的标签分类只要 30ms重度的知识聚合要 8 秒如果统一按 5 秒设置轻能力白白浪费线程等待重能力又快速失败。最终我们按处理类型把超时分成三档连接超时 1 秒固定不变计算密集类和检索类能力读超时 10 秒交互类和写操作类读超时 3 到 5 秒。同时给 Router 增加线程池隔离不同超时档位走不同的线程池避免慢能力把快能力的线程全占光。这个改造之后系统的整体尾部延迟压下来不少p99 从 200ms 以上降到 50ms 以内。5.3 路由规则写太死Agent 一升级就静默失败知识库 Agent 从 v1 升级到 v2 那天我们遇上最典型的一次事故。升级过程中为了平滑切换我把新版本注册成了knowledge.retrieve.v2旧版本按计划下线。结果所有还在调用knowledge.retrieve.v1的调用方一下子全部抛AgentNotFound线上客服摘要直接不可用。这个问题的根因在于我把能力标签当成普通字符串处理没有意识到它其实是一份契约。调用方在代码里明确写了v1它期望的就是 v1 的语义如果 v1 没了契约就被破坏了。后来我在 Agent-Reach 里加了版本别名机制声明knowledge.retrieve.v2时如果配置了soft_alias: [v1]路由层在找不到 v1 时会自动尝试 v2并在响应里附加一个deprecated: true的标记去提醒调用方。这挽救了那个发布窗口但长期策略仍然是把调用方代码全部升级到 v2再逐步摘掉别名。能力升级和 API 升级遵循同一套节奏谁也不能悄悄打破契约。5.4 没有链路追踪出了问题只能翻日志Agent-Reach 刚上线那阵子每次线上出问题排查效率都很低。从客服入口到意图识别、到知识库检索、再到话术生成一条调用链跨越三四个服务每个服务都有自己的日志文件时间戳还对齐不了。有一次客户反馈“回复特别慢而且答案不对”我花了整个下午翻日志结果发现是知识库检索 Agent 偶发返回了旧版本内容但调用链路上根本看不出哪里出了问题。后来把所有 Agent 的入口和出口都强制接入了同一个 trace_id 透传机制信封里的request_id就是这条链路的根 ID。每个 Agent 收到请求时打印一行结构化日志包含request_id、from_agent、target_capability、耗时ms所有日志汇总到集中式日志平台按request_id一搜整条链路各环节的耗时分布一目了然。不出半天旧版内容的问题就被定位到了是知识库 Agent 的缓存 key 没带版本号。这个优化表面上只是加了一个 ID实际是把故障排查时间从小时级压缩到了分钟级属于性价比超高的基础设施投资。6. Agent-Reach 的下一步从框架到协议6.1 如果每个团队都自建一套那就重蹈覆辙了Agent-Reach 在我们团队内部跑稳之后我开始考虑一个更远的问题如果我们只是把它做成自己团队的一个内部框架那别的团队做 Agent 互连时又会重复发明一遍轮子而且每个轮子的协议还不一样。到那时候不同团队之间的 Agent 又没法互相触达问题只是从“服务调用乱”变成了“触达层不统一乱”。所以下一步的重点是把 Agent-Reach 沉淀成一个协议规范消息信封格式、注册发现接口、能力标签命名规则、超时和重试语义这些都是可以直接标准化的内容。具体到代码层面我想把目前 Python 实现里的核心逻辑抽出一个独立协议包让 Java、Node 团队可以照着同一套规则实现自己的客户端只要信封格式一致、注册接口一致跨语言 Agent 互调就不需要额外写适配层。还有一个更朴素的驱动Agent 的规模再涨一个数量级之后能力标签的匹配策略、健康检查的频率、注册表的最终一致性都会变成新瓶颈。现在这套轮询加失败计数已经不满足自动伸缩场景后续考虑把注册中心背后的存储换成 Redis让多个 Router 节点共享同一份注册表实现路由层自身的水平扩展。这一步不急但方向是明确的。6.2 我个人的几条实操心得做 Agent-Reach 这段时间最深的体会其实和代码关系不大。先立能力规范再写代码这是我最想强调的一条。我们一开始就是吃了标签混乱的亏后面补了一套领域.动作.版本的命名规范类似的痛苦才少了很多。协议这东西定得越晚重建成本越高。第二条重试是恩赐也是祸。无脑重试等于在下游故障时帮忙踩油门必须有幂等声明、熔断阈值和分级超时配合重试才真正有用。三条既没写进官网文档却恰恰是线上稳定性的关键可观测性比功能更重要。Agent 互连一旦跑起来问题往往不在某个 Agent 内部而在链路和依赖关系上没有按 request_id 串联的日志排查就是海底捞针。这不算工程上的炫技更像是给多 Agent 协作体系打地基。地基稳了上面站多少 Agent 都踏实。