企业官网建设流程全解析

年初我们把客服意图识别、知识库检索、工单分类、用户画像这几个 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 细节，全部塞进一个结构化的信封里。

信封结构长这样：

为什么不用传统 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 地址。

注册中心的数据模型很简单，核心概念是 AgentInfo：

@dataclass 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-Reach

3.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_factory=time.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_timeout=1.0, read_timeout=10.0, max_fail_count=5, probe_interval=30): 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(f"no 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", json=envelope, timeout=httpx.Timeout(connect=self.connect_timeout, read=self.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, key=lambda 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, host="0.0.0.0", port=9101)

# demo_register.py from agent_reach.core import AgentRegistry, AgentInfo registry = AgentRegistry() registry.register(AgentInfo( agent_id="kb-agent-01", name="knowledge-base", capabilities=["knowledge.retrieve.v1"], endpoint="http://10.0.3.21:9101", health_check_url="http://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_agent="chat-agent-01", target_capability="knowledge.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 分钟。最终成绩：

路由层自身耗时 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 都踏实。