Agent-Reach：多智能体系统服务发现与可靠调用框架实践-育师

这年头聊到多智能体系统（Multi-Agent System），大家第一反应都是“大模型+提示词编排”，但真正落地过的人都懂：最难的不是让单个 Agent 变聪明，而是让一群 Agent 之间能够互相找到、互相调用、互相兜底。Agent-Reach 这个项目，说白了就是冲着“智能体协作的通信与触达问题”去的。它不是一个炫技的模型项目，而是一套给 Agent 用的地址簿、路由器和服务框架，解决的是“A 要调用 B 的能力时，怎么找到 B、怎么把请求递过去、B 挂了怎么办”这一整条链路的问题。如果你正在做多智能体应用、想要统一管理各种异构 Agent，或者只是想把以前一地鸡毛的脚本服务梳理成能互相发现的体系，这篇文章应该能给你一条可以照着搭的完整路子。

我在这里把 Agent-Reach 的整个架构、核心代码、踩过的问题、还有后续的扩展方向完整盘一遍，全部来自实际跑过的项目，不是 PPT 架构。

1. 项目出发点：为什么需要一套 Agent 之间的“触达层”

1.1 多 Agent 协作最常见的三类痛点

先说场景。假设你手上有三个 Agent：一个是专门做文档解析和内容提取的 DocAgent，一个负责对接外部天气数据并且能给出穿衣建议的 LifeAgent，还有一个管用户消息的 ChatAgent。单看每个 Agent 都很容易做，但一旦要让 ChatAgent 在用户问“今天适合穿什么”时去调用 LifeAgent 拿建议，问题就来了。

第一，接口不统一。DocAgent 可能暴露的是 HTTP JSON 接口，LifeAgent 用的是 WebSocket 消息，ChatAgent 又是函数调用风格。每个 Agent 各自定义参数格式，互相之间没法直接对话。

第二，找不到对方。Agent-A 并不知道 Agent-B 的地址、端口、当前是否存活、支持哪些能力。写死配置吧，Agent 扩容、迁移、缩容之后就全对不上了。

第三，调用方要处理一堆“脏活”。比如重试、超时、熔断、并发的任务去重。如果每个模块都自己实现一遍，代码会重复到让人崩溃，而且实现水平参差不齐，线上净是些玄学故障。

Agent-Reach 这个名字就源于这个处境：Reach，触达。它做的是一层薄的“触达层”，让 Agent 之间像打电话一样，通过一个总机找到对方、拨号、通话，而不是各自拉一条专线。

1.2 项目定位与设计原则

Agent-Reach 不是业务逻辑框架，它不关心你的 Agent 内部是怎么实现的，只关心四件事：注册、发现、路由、可靠调用。

注册：每个 Agent 启动时，把自身的能力清单和地址上报到中心目录。
发现：调用方按“能力名”而不是“具体 IP”来查找目标。
路由：同一个能力如果被多个 Agent 提供，按策略选一个合适的。
可靠调用：超时、重试、熔断机制统一在框架层做掉。

整个项目围绕一条核心原则建设：无中心依赖的业务逻辑，有中心依赖的目录服务。意思是，Agent 之间的数据面交互尽量走点对点，但目录和路由这种控制面信息集中管理。这样的好处是，Agent 崩溃不会影响目录，而目录短时不可用也不会立刻让整个系统瘫掉——Agent 本地会缓存一份地址表。

注意：这里说的“中心”容灾是基础功能，但我实际跑下来发现，真正让系统稳定性发生质变的并不是中心高可用，而是“调用方侧的重试退避 + 目录缓存”这两件事。后面有专门一节讲。

2. 整体架构与核心设计思路

2.1 三层结构：连接层、路由层、策略层

Agent-Reach 的架构拆成三层，每一层只做一件事，避免逻辑纠缠。

连接层负责维持 Agent 与 Agent 之间的实际数据传输。底层我直接用了 HTTP/2 的流式调用，为什么没有选 WebSocket 或者自定义 TCP 协议？原因是 Agent 之间的调用模式比较杂：有短小快速的 RPC（比如查天气）、有流式返回的长任务（比如文档解析后一段一段返回结果）、也有文件级别的传输。HTTP/2 天然支持多路复用和流式响应，而且在公网、内网环境下穿透性和防火墙友好度都比裸 TCP 好很多。连接层维护了一个连接池，复用同一个目标 Agent 的多个连接，减少频繁建连开销。

路由层是最核心的部分。调用方发起请求时，路由层先查本地的“能力路由表”，看目标能力注册在哪个 Agent 上、当前是否健康、路由策略是什么。如果本地没有缓存或者缓存过期，就回源到中心目录拉取一次最新表。路由层还处理一种特殊情况——能力由多个 Agent 同时提供服务，这时会按策略选目标，比如最简单的轮询，或者按 Agent 上报的负载值加权。

策略层管的是跨请求的策略规则，包括超时时间、重试次数、熔断阈值、限流等。这些策略独立于业务代码，通过配置文件或者注册时的元数据来设定。我最初的版本把策略写死在框架里，后来发现不同能力差异太大。比如“查天气”可以只重试 1 次，但“解析一份大文档”重试成本就很高，而且大概率不是偶发故障，重试反而会加剧压力。所以策略层一定要支持按能力配置。

2.2 为什么用“能力名”而不是“Agent 名”做路由

这算是 Agent-Reach 设计里我认为最关键的一个决策。很多类似系统是直接按 Agent ID 路由，比如“调用 ChatAgent”，听起来很直观，但实际维护你会遇到很头疼的问题。

当你有 5 个 Agent 都提供“文本摘要”能力时，按 Agent ID 调用意味着调用方必须知道具体是谁。一旦这 5 个 Agent 里有一个下线、一个新的启动，调用方就要改配置。而按“能力名”路由，调用方只关心“我要摘要”，不关心谁来做。目录层会动态更新，哪个 Agent 挂了就从路由表里摘掉，新启动的自动顶上。这本质上是把“服务发现”和“业务语义”解耦了。

举个例子，同一个生命周期管理 Agent 它既提供summarize.text能力，也提供convert.format能力。调用方按能力名寻址时，完全感知不到它承载了多少种能力，也不需要关心“这是同一个进程”。

2.3 目录服务的数据模型

目录服务存储的数据结构其实不复杂，核心是这样一个注册表：

{ "agent_id": "life_agent_001", "host": "192.168.1.20", "port": 8807, "capabilities": [ { "name": "weather.query", "version": "1.2.0", "params_schema": { "city": "string", "date": "string" }, "strategy": { "timeout_ms": 3000, "retry": 1 } }, { "name": "clothing.suggest", "version": "1.0.0", "params_schema": { "temperature": "float", "condition": "string" }, "strategy": { "timeout_ms": 2000, "retry": 0 } } ], "status": "online", "last_heartbeat": 1720000000, "load": 0.3 }

这个 JSON 结构里有个容易被忽略的细节：params_schema。别看它只是几个字段描述，它其实承载了一个很实用的功能——路由层在转发请求前先做一次入参校验，不合法就直接拒绝，根本不会把脏数据递到目标 Agent。这能拦截掉很多低级错误。

另一个设计点：每个能力都带独立的超时和重试策略。因为同一个 Agent 的不同能力复杂度差异可能很明显，统一策略会牺牲效率或者导致超时误判。

3. 核心实现：最小可用版本怎么搭

3.1 技术选型与依赖

Agent-Reach 的核心实现我用了 Python 3.10，配合 FastAPI 做 HTTP 服务、SQLite 存目录信息、asyncio 做并发调度。选 Python 是因为生态快、好改；FastAPI 是个人习惯，异步支持好；SQLite 能单文件跑，适合中小规模，后面如果要上亿级注册量再换 PostgreSQL 也容易。

# requirements.txt fastapi==0.110.0 uvicorn[standard]==0.29.0 httpx==0.27.0 aiosqlite==0.20.0 pydantic==2.6.0

3.2 注册与心跳模块

每个 Agent 在启动时调用目录服务的注册接口，把自己的能力信息上报上去。注册接口长这样：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Dict, Any import time import aiosqlite app = FastAPI() class CapabilityRegister(BaseModel): name: str version: str = "1.0.0" params_schema: Dict[str, Any] = {} strategy: Dict[str, Any] = {} class AgentRegister(BaseModel): agent_id: str host: str port: int capabilities: List[CapabilityRegister] @app.post("/register") async def register_agent(agent: AgentRegister): now = int(time.time()) agent_key = f"{agent.agent_id}:{agent.host}:{agent.port}" async with aiosqlite.connect("agent_directory.db") as db: for cap in agent.capabilities: await db.execute( """ INSERT INTO capabilities (agent_key, cap_name, cap_version, params_schema, strategy, status, last_heartbeat) VALUES (?, ?, ?, ?, ?, 'online', ?) """, (agent_key, cap.name, cap.version, str(cap.params_schema), str(cap.strategy), now) ) await db.commit() return {"ok": True, "agent_key": agent_key}

注册后每 10 秒发一次心跳，目录服务收到心跳就更新last_heartbeat。如果超过 30 秒没收到心跳，服务端会把对应的 Agent 标记为offline，并从路由表里剔除。

这里有个细节值得说：注册时如果没有带版本号，我默认给了 1.0.0，并且同一个 Agent 重复注册时应该先删除旧记录再插入新记录，不然表里会积累一堆过期条目。生产环境下，同一个 Agent 重启后 agent_id 不变但端口可能变了，这时候如果旧记录不清理，就会出现“路由到了旧地址但连不上”的情况。

3.3 服务发现与路由表缓存

调用方侧维护了一个本地缓存的路由表，结构是“能力名 -> Agent 列表”。每次调用前先查缓存，缓存失效才去目录服务拉一次全量表。

import httpx import time class LocalRouteTable: def __init__(self, directory_url: str, ttl: int = 30): self.directory_url = directory_url self.ttl = ttl # 秒 self._table: Dict[str, list] = {} self._last_updated = 0 self._lock = asyncio.Lock() async def _refresh(self): async with self._lock: if time.time() - self._last_updated < self.ttl: return async with httpx.AsyncClient() as client: resp = await client.get(f"{self.directory_url}/routes") routes = resp.json() self._table = routes self._last_updated = time.time() async def get_agents(self, capability_name: str): await self._refresh() return self._table.get(capability_name, [])

TTL 设 30 秒是一个平衡值。太短会导致目录服务被频繁请求，太长会导致新增 Agent 不能及时被感知。如果你在容器环境里 Agent 启停非常频繁，建议把 TTL 调到 10 秒，同时把目录服务的查询接口做成支持?updated_since=timestamp的增量拉取，减少不必要的数据传输。

3.4 路由与调用：带重试和熔断的请求转发

拿到目标 Agent 列表后，路由层选择一个目标并转发请求。核心函数如下：

def select_target(agent_list: list, strategy: str = "round_robin"): if not agent_list: raise NoAvailableAgentException() if strategy == "random": return random.choice(agent_list) # round_robin 用全局计数器轮转 counter = 0 with counter_lock: index = counter % len(agent_list) counter += 1 return agent_list[index] async def call_agent(target_agent, payload, cap_name, strategy_cfg): url = f"http://{target_agent['host']}:{target_agent['port']}/invoke/{cap_name}" timeout_ms = strategy_cfg.get("timeout_ms", 5000) retry = strategy_cfg.get("retry", 1) last_exc = None for attempt in range(retry + 1): try: async with httpx.AsyncClient(timeout=timeout_ms / 1000) as client: resp = await client.post(url, json=payload) resp.raise_for_status() return resp.json() except Exception as e: last_exc = e # 退避：等 200 * attempt 毫秒 await asyncio.sleep(0.2 * attempt) # 打熔断标记，后续一段时间不再路由到该 Agent circuit_breaker.mark_failure(target_agent) raise last_exc

这个实现里有三个容易被忽略的点：

重试不是盲目的。重试只针对连接超时、5xx 这类可能因为临时抖动产生的问题，而参数错误（4xx）直接抛错，不重试。代码里没写出来，但线上逻辑必须判断resp.status_code再决定是抛错还是继续重试。
熔断器按照“Agent 实例”粒度而不是“能力”粒度。同一个 Agent 提供多个能力，它网络异常时大概率所有能力都不行，所以按实例熔断比按能力熔断更能快速切断故障。
退避时间不能太长。Agent 高频调用场景下，如果退避 1 秒会导致整个调用链路的尾部延迟飙升。200ms 起步、最多等 1 秒是经验值。

4. 实操踩坑记录与排查技巧实录

这节全是实际跑 Agent-Reach 遇到的问题和我的排查思路。

4.1 问题一：心跳超时导致误杀正常 Agent

项目刚上线时，内网还算稳定，但我发现偶尔有 Agent 被目录服务错误标记为 offline。排查日志发现，目录服务在某个时刻连续收不到某 Agent 的心跳，触发了 30 秒超时。但那个 Agent 本身 CPU 占用很高，一直在跑一个长计算任务，导致心跳发送线程被抢占，发不出来。

这个问题的根子在于：心跳是单独线程发的，但 Python 的 GIL 在密集计算时会让心跳线程严重延迟。解决办法有两种：一是把心跳发送放到独立的进程里，二是降低误杀阈值的影响。我最终选了“三层判定”策略：

心跳超时 30 秒，标记为suspect（可疑），不立刻剔除
如果可疑状态又持续 30 秒且主动探测（发一个轻量 ping 到 Agent）也失败，才标记offline
只要主动探测成功，马上恢复online

虽然多了一段“观察期”，但换来的是再也不会因为 Agent 忙而误杀。代价就是故障转移时间变长了一点，但 Agent 系统的稳定性比故障转移速度更重要。如果你做的是实时性极强的调用，可以把这个观察期缩短到 10 秒。

4.2 问题二：循环调用导致消息风暴

多 Agent 协作场景非常容易出现循环调用：Agent-A 调用 Agent-B，Agent-B 调用 Agent-C，Agent-C 的逻辑里又调回了 Agent-A。最直接的后果就是消息量指数级上升，把所有 Agent 都打挂。

我加了一层非常轻量的防护：每次调用在消息头里带一个trace_id和max_depth字段。每次转发时depth + 1，如果depth > 5就直接拒绝并返回错误说明“检测到可能循环调用”。同时记录下整个调用链的路径，方便事后排查。

def check_depth(message_header): depth = message_header.get("depth", 0) if depth > 5: raise MaxDepthExceededException(message_header.get("trace_id")) return depth + 1

这个深度阈值 5 是拍脑袋加实测的值，如果你的 Agent 链路比较深，可以调到 8 或 10，但必须配合超时策略——因为深度越深，单次请求的端到端时间就越不可控。

4.3 问题三：Agent 重启后端口漂移导致路由到失效地址

容器环境下 Agent 每次重启端口都会变。Agent-Reach 的注册逻辑里，Agent 重启后会用新端口重新注册，但问题出在窗口期——老端口还没从路由表里消失，新注册已经写进来了，客户端可能在这个间隙里路由到老地址。

我用了“版本号 + 注册时间戳”双重判断来规避这个窗口：每个 Agent 在注册时带上启动时间戳，目录服务对比同一个agent_id的新旧记录，只保留时间戳更晚的那条。同时客户端在调用失败时，会把失败地址上报给目录服务，目录服务如果确认该地址已经没有对应注册记录，就主动清理。

4.4 问题四：高并发下目录服务 SQLite 锁竞争

SQLite 在并发写入时会出现database is locked。Agent 数量一多，注册和心跳的写入频率很高，这个问题马上就暴露了。

解决方案是把 SQLite 设为 WAL 模式，并且把心跳更新改成“批量 UPSERT”。心跳接口不再每 10 秒写一次每条记录，而是攒了一批心跳到内存里，每 30 秒统一做一次批量更新。实测下来写入压力降了一个数量级，锁竞争基本消失。

PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL;

这里多说一句，SQLite 做目录服务撑到几千个 Agent、几万个能力注册还是够用的。再往上你就该换 PostgreSQL 了，但因为数据模型简单，迁移成本不高。

4.5 常见问题速查表

现象	可能原因	排查手段	解决办法
Agent 间歇性掉线	心跳占不到 CPU	看心跳线程日志延迟	独立进程心跳+三层判定
调用全部超时	路由表过期，目标地址已失效	检查目录服务日志/调用失败上报	客户端失败上报+服务端主动清理
消息风暴	Agent 之间循环调用	看 trace_id 调用链	深度限制 + 环路检测
SQLite 写入锁死	并发写太频繁	查看 error 日志	WAL 模式 + 批量心跳
新 Agent 上线后调用仍失败	客户端路由表 TTL 未过期	检查客户端缓存刷新时间	调低 TTL 或支持主动推送

5. 扩展方向：Agent-Reach 还能怎么用

5.1 接入大模型，做动态能力编排

基础版的 Agent-Reach 是“调用方显式指定要哪个能力”，但接入了大模型之后，可以让 LLM 充当“意图路由器”。用户说“今天上海适合穿什么”，ChatAgent 收到这句问话后，没有直接去查天气，而是先让 LLM 做意图拆解，得到两个子任务：查询上海天气、根据天气生成穿衣建议。然后 ChatAgent 通过 Agent-Reach 分别调用weather.query和clothing.suggest这两个能力，再把结果拼装成最终回复。

这个模式下，Agent-Reach 的价值从“服务发现层”升级成了“能力执行层”。LLM 不需要关心底层有多少 Agent 在支撑，只需要把它们当成可调用的工具函数。而 Agent-Reach 提供的超时、重试、熔断恰好弥补了 LLM 调用外部工具时最头疼的稳定性问题。

5.2 接入可观测性：全链路追踪

Agent 调用链多了以后，出问题时的排查难度指数上升。我在 Agent-Reach 的消息头里加上了trace_id，但这还不够，必须把 trace 信息主动上报到日志系统，才能在跨 Agent 调用时看清楚整条链路。

具体做法：每个 Agent 在收到请求时，生成一个 span 记录（包括开始时间、结束时间、目标能力、返回状态），把 span 统一发到日志采集端。前端用现有的链路追踪工具渲染出调用瀑布图。这样做以后，定位“某个请求为什么慢”从“看 N 个 Agent 日志拼凑”变成了“直接看瀑布图”。

5.3 更丰富的路由策略：按负载、按地域、按成本

目前内置的策略只有轮询和随机，但实际场景里已经有同事提需求：内部有两个 Agent 都提供文本翻译能力，一个用了云端付费 API，另一个是本地开源模型，成本差很多。他希望非实时场景优先走便宜的本地模型，只有本地排队严重时才走付费 API。

这个需求其实只改路由层的一个插件函数。Agent-Reach 把路由策略设计成了可扩展接口：

class RouteStrategy: def select(self, agent_list: list, context: dict) -> dict: raise NotImplementedError() class CostAwareStrategy(RouteStrategy): def select(self, agent_list, context): return min(agent_list, key=lambda a: a.get("cost_per_call", 0))

你有任何自定义的择路逻辑（最低延迟、最低负载、指定区域优先），都可以通过实现这个接口挂进去。