Agent-Reach：让AI Agent像微服务一样互相发现与协作

如果你手里已经有三五个 AI Agent 在跑，你会发现一个尴尬的情况：每个 Agent 都挺聪明，但彼此之间完全不认识。A 做舆情分析，B 做报表生成，C 负责自动回复——它们各自守着各自的 API Key，各跑各的进程，遇到对方擅长的事就只能干瞪眼，或者靠你在中间手动搬数据。Agent-Reach 这个名字，翻译成大白话就是"让智能体互相够得着"。它解决的问题特别朴素：在 Agent 越来越多、任务越来越复杂的今天，让它们像微服务一样能注册、能被发现、能互相调用，而不是靠人肉胶水做集成。在这篇里，我会从设计思路、核心协议、配套代码到踩坑记录，把一个可落地的 Agent 协作框架原型完整拆给你看，适合已经在折腾 Agent 应用、又不想被某家厂商绑定在私有协议里的开发者和架构师。

1. 先把 Agent-Reach 要解决的事想清楚

1.1 碎片化 Agent 的"互相不通"难题

很多人做了一个 Agent 之后，第二个 Agent 很快就来了，第三个也不远了。等到你有三个以上 Agent，立刻会撞上一堵墙：每个 Agent 都是一个"信息孤岛"，它们之间的通信基本靠你和 Excel 文件。你今天从舆情 Agent 那边拿结果，手动喂给报表 Agent，明天报表 Agent 的输出又要夹带进客服 Agent 的上下文。这套流程短期能跑，但稍微一变化就崩——比如你换了数据源，或者某个 Agent 的服务地址变了，牵一发动全身。

Agent-Reach 的思路不是再做一个"超级 Agent"来管所有 Agent，而是给每个 Agent 配一套标准的"自我介绍"和通信协议，让它们通过一个轻量级的协调层互相发现和请求。这个概念其实特别像微服务架构里的注册中心，但语义层级高得多。微服务注册的是接口和实例，Agent-Reach 注册的是能力和上下文边界。它不关心你的 Agent 是用 LangChain 还是直接裸调 LLM，也不关心你跑在 Python 还是 Node.js，只要按照协议说话，就能在同一个协作网络里干活。

回头看，这套设计有一个很关键的判断：不要把 Agent 之间的协作做成"人工编排的流程图"。流程图在场景固定的时候很好用，但现实是 Agent 的任务经常是拆分的——主 Agent 拿到目标后，需要动态判断该找谁、问什么、怎么合并结果。所以 Agent-Reach 把重点放在"能力发现"和"动态路由"上，而不是"预制工作流"。

1.2 为什么是 Reach，而不是又一个编排平台

市面上已经有不少 Agent 编排方案，有的主打复杂工作流，有的主打多角色对话，有的干脆给你一个可视化拖拽面板。它们的通病是：一旦你用了它，你的 Agent 就和这个框架绑死了。你今天评估没问题，半年后想换框架，迁移成本让你怀疑人生。Agent-Reach 从一开始就把 choice 做小——它只定义一种"最小协作契约"，具体业务逻辑全部留在各个 Agent 内部。

我觉得这种取舍是对的。编排平台解决的是"趋势可控"的问题，而 Reach 解决的是"相互可见、可以对话"的问题。如果一套系统的定位是"协作底座"，它就不该教你业务怎么做，而该像 HTTP 协议一样，保证你在上面跑的任意应用都能互通。HTTP 不管你是电商还是博客，只管请求和响应的格式；Reach 不管你的 Agent 是做什么的，只管"你能做什么、我要怎么调用你、结果怎么安全还回来"这三件事。

还有一层现实考虑：你真的希望所有 Agent 都跑在一个进程里，共享一个复杂的运行时吗？凡是上过生产的都知道，一个 Agent 因为上游 API 超时把整个 Brother 都拖下水的情况太常见了。Reach 这种"各自独立、松散耦合"的方式，天然适合跨进程、跨机器甚至跨团队部署。你团队里另一个人写的 Agent，只要它实现了协议，你不用改一行代码就能调起来。

1.3 项目落地后的三层价值

我做了个小范围落地之后，对收益感受特别直观。第一层是技术价值：Agent 之间的调用从"人肉配置"变成了"运行时发现"，新增一个 Agent 的成本从一天缩短到半小时，重点就剩写能力描述和联调。第二层是组织价值：不同小组可以各自维护自己的 Agent，交付物不再是"一堆脚本"，而是"一个有协议地址的服务"，边界清晰，责任也好划。第三层是演进价值：因为 Agent 之间靠能力匹配而不是写死的接口耦合，整体的拓扑结构可以随时调，哪个 Agent 效率低就换哪个，不用动调用方。

这个框架给我的整体感觉是：它特别适合在 Agent 数量从"玩具级"往"生产级"过度的那个阶段介入。1 个 Agent 不需要它，50 个 Agent 你需要的可能是另一套更重量级的东西。但在 3 到 20 个 Agent 这个甜蜜区间，Agent-Reach 是复杂度收益比最合理的一种方案。

2. 核心细节拆解：从协议到数据结构的取舍

2.1 一套"自我描述"的 Agent Profile

Agent-Reach 的基石，是每个 Agent 启动时向协调中心注册一份"Agent Profile"。你可以把它理解成一张名片的加强版。名片上不仅有名字和地址，还有三个关键字段：agent_id、agent_type 和 capabilities。agent_id 是全网唯一标识，一般直接用 UUID 或者"组织名.服务名"的倒置域名风格，避免冲突。agent_type 描述这 Agent 属于哪一类，比如"content_generator""data_analyzer"或"watcher"，方便做粗粒度的路由过滤。

capabilities 是整张名片里最核心的部分。它不是一句"擅长做分析"这种人类语言，而是一组结构化的能力声明。我踩过坑之后强烈建议，capabilities 至少要包含三件事：能力名称（capability_id）、输入描述（input_schema）和输出描述（output_schema）。input_schema 和 output_schema 直接用 JSON Schema 描述，这样调用方在发起请求前就能校验参数，不用等到运行时才发现"哦，原来你要的是这种格式"。

这份 Profile 还应该带一个 health_check 的地址。协调中心会周期性去 GET 这个地址，只要返回 2xx，就认为 Agent 活着，继续参与路由；连续几次超时，就直接把它从可用列表里摘掉。一个容易忽略的细节是 health_check 的返回体里应该带上当前 Agent 的负载水位，比如 "load": 0.8，调度时可以把流量优先给空闲的 Agent。这个字段早期不做没关系，但协议设计时留好位置，后面扩展不用改版本。

2.2 Capability Schema 怎么写才不坑

我见过很多人写能力声明，直接简单写一句"能生成报告"。这不行，等于没写。别人拿到这份声明，根本不知道要传什么参数、会拿回什么结果。真正能落到代码里的 Capability Schema，应该做到"调用方只看 schema，就能把请求构造出来"。

例如，一个"周报生成器"的能力声明，不要写"I can generate weekly reports"，而要给出一个类似这样的 JSON Schema：required 里明确列出 start_date、end_date、team_id、summary_style 这几个必填字段，summary_style 用 enum 限定死取值范围是"concise""detailed""funny"，team_id 用 pattern 限定格式。输出 schema 则描述返回体里可能带 report_id、content、recommendation_links 这些字段，以及各自类型。这样做的好处有两个：一是调用方可以在静态阶段做参数校验，二是协调中心可以根据 schema 做字段级的路由匹配——比如某个任务指定了 team_id，就能自动过滤掉不支持该字段的 Agent。

这里要说一个容易踩的坑：Schema 的粒度不要太细，也别太粗。太细的话，比如连"用户名必须 3-20 个字符，首字母大写"这种都写进去，会让能力声明很难维护，上游一改需求就崩。太粗的话，比如"data": "any"，那又回到了没有声明。我在实践中掌握的尺度是：把业务上必须强对齐的字段写清楚，其余自由字段放一个"extensions" JSONObject 接住。这样既保证了核心契约稳定，又给未来演进留了空间。

2.3 消息路由与任务协商的原子性

Agent-Reach 的调用链跟 REST API 不太一样。REST 是客户端直连服务端，而 Reach 在中间加了"协商"这一步。发起方（Consumer）先向协调中心发一个"请求意图"，说清楚自己想做"报表分析"，附上所需的输入 schema 和偏好。协调中心拿着这个意图，去匹配所有 Agent Profile，返回一批符合条件的候选列表，每个候选还附带当前健康状态和负载信息。

Consumer 拿到候选列表后，有两种选择：直接选第一个可用的，或者按某种策略打分选最优。我推荐在早期阶段直接选"健康且负载最低"的那个，这是最省事的策略，但它有个隐患——如果多个 Agent 能力完全一样，负载均衡是不错，但无法体现 Agent 的个体特性。所以我在设计里加了一个可选的"投标"机制：Consumer 可以 broadcast 一个 task_offer，候选 Agent 根据自身上下文判断接受还是拒绝，并回传一个 confidence 分数。Consumer 汇总之后，选出置信度最高的执行，这个动作叫"任务协商"。

任务协商这种模式，代价是延迟变高了。如果 Consumer 和 Agent 在同一内网，一次协商的额外耗时通常在 30ms 左右，还能接受；一旦跨越公网，额外延迟可能到几百毫秒，就必须做超时和降级。我的建议是：同步短任务用"直接路由"，异步长任务用"协商投标"，把选择权交给场景，不要一刀切。很多框架死在"什么都想动态协商"，最后变成了一个没人愿意等的中间层。

2.4 工程上容易被忽略的三个禁忌

第一，不要在 Agent 之间直接传大报文。Agent 的上下文本就宝贵，一个动辄几 MB 的中间结果传来传去，既费时间又费 token。正确姿势是让产出方先把结果写到对象存储或者本地文件系统，通话里只传存储路径和一个文件的 SHA256 摘要。这样还有一个额外好处：对方可以校验文件完整性，排查"传输过程中数据被改坏"的问题。

第二，不要把 Agent ID 当作安全边界。Reach 协议设计得像 HTTP，但网络环境并不一定可信。所有 Agent 之间的通信至少要做鉴权，简单点就是共用 Token 放进 Header，复杂点可以上 mTLS。我先用共享 Token，后面要放开给跨团队调用时，再补一层每个 Agent 自己的密钥签名，防止协调中心被中间人劫持后冒充任意 Agent。

第三，不要在调试时开着"全链路同步调用"跑长任务。我一开始图省事，让所有调用都是同步阻塞，结果一个 Agent 跑去调 LLM 生成长文，整个调用链就挂起，超时像传染病一样扩散。后来把所有耗时可能超过 5 秒的调用全部改造成异步任务模式：Consumer 提交任务拿到 task_id，Agent 完成后回调一个 webhook，或者 Consumer 轮询 task 状态。改造之后系统稳定性明显上一个台阶。

3. 实操实录：搭一个双 Agent 协作的 Reach 实例

3.1 最小依赖环境与目录结构

我不喜欢一个演示就要装一堆重型依赖。这里我用 Python 3.10，核心就是 FastAPI 加 jsonschema 两个库。协调中心本身也是一个 FastAPI 服务，Agent 侧是普通 Python 进程，通过 HTTP 注册上去。不想用异步框架的人也可以换成 Flask，但 FastAPI 的异步性能和自动文档是真的省事，测试阶段直接用 FastAPI 的话，省掉很多 URL 拼错的麻烦。

目录结构我按"协调中心、两个 Agent、共享库"四个包来组织。共享库（shared.py）里放 Agent Profile 的定义、能力声明的数据类，以及一个简单的 Schema 校验函数。协调中心（registry.py）只做三件事：接收注册、处理发现请求、转发调用。两个 Agent 分别是 data_fetcher（负责拉取静态数据并整理成 CSV）和 report_builder（负责把 CSV 转成最终报告）。这个例子虽然业务简单，但已经覆盖了 Agent-Reach 的核心链路：注册、发现、能力匹配、异步任务执行、结果回传。

agent-reach-demo/ ├── shared_profile.py # Profile、Capability 定义 ├── registry.py # 协调中心（发现+转发） ├── agent_fetcher.py # Agent A：拉数据 ├── agent_builder.py # Agent B：写报告 └── consumer_demo.py # 调用方示例

3.2 注册中心与 Agent 注册实现

先看 shared_profile.py。我偷了点懒，把 Agent Profile 直接定义成 Pydantic 模型，字段名和线上协议保持一致。Capability 里的 input_schema 和 output_schema 就用 dict 存 JSON Schema，不强制建类型，保持灵活性。

# shared_profile.py from uuid import uuid4 from typing import Dict, Any from pydantic import BaseModel, Field class Capability(BaseModel): capability_id: str name: str description: str = "" input_schema: Dict[str, Any] output_schema: Dict[str, Any] class AgentProfile(BaseModel): agent_id: str = Field(default_factory=lambda: str(uuid4())) agent_type: str display_name: str endpoint: str # Agent 自己的服务地址 health_check_url: str # 健康检查地址 capabilities: list[Capability] version: str = "0.1.0" load: float = 0.0 # 当前负载水位，Agent 周期性上报

注册中心 registry.py 的逻辑不长，关键是保存所有 AgentProfile，并提供三个 API：POST /agents 注册、GET /agents?capability_id=xxx 查询候选、POST /dispatch 做任务转发。为了演示效果，我把数据存在字典里，生产环境替换成 Redis 或者直接上数据库都行，原理是一样的。

# registry.py from fastapi import FastAPI, HTTPException, Depends from pathlib import Path import hashlib from shared_profile import AgentProfile, Capability app = FastAPI(title="Agent-Reach Registry") REGISTRY: dict[str, AgentProfile] = {} @app.post("/agents") def register(profile: AgentProfile): REGISTRY[profile.agent_id] = profile return {"status": "ok", "agent_id": profile.agent_id} @app.get("/agents") def discover(capability_id: str | None = None): if not capability_id: return list(REGISTRY.values()) candidates = [ p for p in REGISTRY.values() if any(c.capability_id == capability_id for c in p.capabilities) ] return candidates

你可能会问：为什么 dispatch 不在 registry 里实现？我的习惯是 registry 只做注册与发现，转发逻辑单独拆出来，因为转发会涉及超时、重试、鉴权、文件哈希校验，跟注册中心的职责混在一起会让它变得极其臃肿。在正式项目里，dispatch 是一个独立服务或 Consumer SDK 里的功能，不是一个 API 端点。

3.3 Agent 侧注册与能力暴露

Agent 侧的代码风格比较统一：启动时先起一个 FastAPI 应用（提供自己的服务接口），然后向 registry 发起注册请求。以 data_fetcher 为例，它会暴露一个 /fetch_data 接口，输入参数是 keyword 和 date_range，返回的是一个 CSV 文件的下载链接和文件哈希。

# agent_fetcher.py import httpx, uvicorn, json from pathlib import Path from fastapi import FastAPI from shared_profile import AgentProfile, Capability app = FastAPI() MY_PROFILE = AgentProfile( agent_type="data_provider", display_name="data_fetcher", endpoint="http://127.0.0.1:9101", health_check_url="http://127.0.0.1:9101/health", capabilities=[ Capability( capability_id="fetch_csv", name="fetch_data", input_schema={ "type": "object", "required": ["keyword", "date_range"], "properties": { "keyword": {"type": "string"}, "date_range": {"type": "string"} } }, output_schema={ "type": "object", "required": ["file_url", "file_sha256"], "properties": { "file_url": {"type": "string"}, "file_sha256": {"type": "string"} } } ) ] ) @app.get("/health") def health(): return {"status": "up", "load": 0.2} @app.post("/fetch_data") def fetch_data(payload: dict): # 业务逻辑省略：假设这里生成一个 data.csv file = Path("output/data.csv") sha = hashlib.sha256(file.read_bytes()).hexdigest() return { "file_url": "http://127.0.0.1:9101/files/data.csv", "file_sha256": sha } @app.on_event("startup") def register_to_registry(): httpx.post("http://127.0.0.1:9000/agents", json=MY_PROFILE.model_dump())

设备上的端口分配要注意：registry 跑在 9000，data_fetcher 跑在 9101，report_builder 跑在 9102。Agent 注册的同时，最好把 health_check 的轮询周期放到 5 秒，这样 registry 对 Agent 的存活感知延迟能控制在 5 秒内，故障转移不太迟钝。

3.4 跑通一次完整的跨 Agent 协作

现在看 consumer_demo.py。它扮演的是最上层的业务方：想生成一份"某关键词的报告"，但它自己既不拉数据也不写报告。它在 registry 上先发现两个候选 Agent，然后依次调用。先调 data_fetcher 的 fetch_csv，拿到文件 URL 和 SHA256，然后把这个信息作为输入参数传给 report_builder 的 build_report。

# consumer_demo.py import httpx, json REGISTRY_URL = "http://127.0.0.1:9000" FETCHER_URL = "http://127.0.0.1:9101" BUILDER_URL = "http://127.0.0.1:9102" # 1. 发现能力 fetchers = httpx.get(f"{REGISTRY_URL}/agents", params={"capability_id": "fetch_csv"}).json() assert fetchers, "no fetcher found" # 2. 调用 data_fetcher resp = httpx.post(FETCHER_URL + "/fetch_data", json={ "keyword": "agent协作", "date_range": "2024-01-01~2024-12-31" }).json() print("fetch result:", resp) # 3. 发现 report_builder builders = httpx.get(f"{REGISTRY_URL}/agents", params={"capability_id": "build_report"}).json() assert builders, "no builder found" # 4. 调用 report_builder report = httpx.post(BUILDER_URL + "/build_report", json={ "file_url": resp["file_url"], "file_sha256": resp["file_sha256"], "title": "Agent 协作年度分析" }).json() print("report result:", report)

这个流程看起来简单，但跑起来容易出问题的是第三步——你的 report_builder 可能已经注册了，但它的 input_schema 里要求传 sha256，而你在代码里没传，它直接就报 422。这就是前面说的 Schema 的作用：它能让错误提前到"构建请求"阶段，而不是等到业务执行完才发现参数错了。所以我在真正跑 demo 前后，建议先调一下 registry 的 GET /agents 接口，把每个 Agent 的 capabilities 完整打印出来，对着 schema 一条条核对。

3.5 实测中的参数调优记录

这一节算是我个人比较想分享的干货。第一次跑通时，我用的全是同步调用，整个链路耗时大约 2.4 秒，其中 report_builder 生成报告占了 2 秒，因为底层调了 LLM。这个延迟对 demo 来说还好，但对生产不可接受。我做了两处调整：一是把 report_builder 改造成异步任务，提交后立刻返回 task_id；二是把 registry 的心跳超时从默认 10 秒改成了 3 秒，这样 Agent 挂了能在 3 秒内摘除，避免调用方拿到失效地址。

另外要留意的一个参数是 HTTP 客户端连接池。早期我没限制并发连接数，consumer 同时触发 30 个任务时，协作链路里 Agent 的 FastAPI 直接撑了满脸的 503。后来我在所有 httpx.Client 里显式设置了 limits=httpx.Limits(max_connections=20, max_keepalive_connections=10)，然后把 FastAPI 的 worker 数从 1 调到 4，测试环境瞬间稳多了。Agent 协作毕竟是网络密集型，这类参数调整往往比优化 Agent 本身更立竿见影。

4. 常见问题与排查技巧实录

4.1 问题速查表

这张表是我在联调阶段被坑得最狠的几个点的汇总。每一条背后都对应着我一次"调了半小时才发现是这种蠢问题"的教训，建议你直接抄到排查手册里。

4.2 现场复盘：一次注册信息不一致的崩溃

说个真实案例。有一次我新起了一个 report_builder 版本，为了测试新特性，把 capability_id 从 build_report 改成了 build_report_v2，但消费方代码没改，还是按 build_report 去发现。结果表现是：调用方报"no builder found"，我被这个错误提示迷惑了很久，以为是注册中心的网络问题。后来我把 registry 的全量清单拉出来一看，才发现旧版 Agent 已经因为心跳超时被摘除了，而新版 Agent 的能力 ID 变了，导致注册表里其实一个可用对象都没有。

这种问题的根源在于我有一份"隐藏的契约"没有显式化——代码里写死了 capability_id，却没有把它当作 API 版本管理来看待。后来我在所有 Agent 配置里加了一个 capability_version 字段，一旦能力的行为有破坏性变更，就升级语义化版本号，同时保留旧版 Agent 跑一段时间做过渡。这是让我少掉很多头发的关键改进。

4.3 快速验证 Agent-Reach 链路是否健康的三个命令

排查时我会按"从底层到上层"的顺序打三枪。第一枪打 Agent 自身：curl 它的 health 接口，确认它真的活着。第二枪打注册中心：curl registry 的 /agents 接口，确认注册表里的数据齐全，字段没有拼写错误。第三枪打一次最小化调用：直接用 curl 给 Agent 的接口发一个最简 payload，排除调用方代码的 bug。

curl -s http://127.0.0.1:9101/health curl -s http://127.0.0.1:9000/agents curl -s -X POST http://127.0.0.1:9101/fetch_data \ -H 'Content-Type: application/json' \ -d '{"keyword":"test","date_range":"2024-01-01~2024-01-31"}'

只要这三枪都过，链路基本没问题。如果哪一枪有问题，就把那一段单独拎出来细查，不要跑到上层代码里去大海捞针。这套排查顺序看起来朴素，但它能帮你把"网络问题、注册问题、数据问题"三种故障快速隔离，效率比一上来就看日志高得多。

5. 个人经验与后续扩展方向

5.1 我在实际落地中反复强调的三件事

第一件事：Agent 的"能力声明"要像对待 API 文档一样认真对待。我见过太多协作系统死在文档跟不上实现这一步。你 Code Review 时应该把 Capability Schema 的变更当成接口变更来审，不然很容易出现"注册表说的是一套，实际接收的是另一套"的灾难。第二件事：永远假设对方 Agent 会挂。你在代码里做好超时、重试、降级，是对整体链路最大的善意。第三件更偏管理向：给每个 Agent 指定一个 owner。现在 Agent 多了之后，我最怕的是某个 Agent 没人管，它挂了整个链路安静地坏掉半天没人发现。owner 机制配合上 registry 的预警通知，能大幅降低这类"安静故障"给团队带来的伤害。

5.2 Agent-Reach 后续可以怎么扩展

我目前实现的是最小可用版本，如果要让它往生产级走，有几个方向我建议值得投入。第一个是给协调中心加"策略组"能力，支持按 Agent 的语义标签（比如"付费""免费""内网""外网"）做路由隔离。第二个是完善任务协商的回退策略——如果候选 Agent 全部拒绝，当前是直接报错，未来可以做二次广播或降级到低置信度 Agent，避免任务完全失败。第三个是增加可观测性，把每一次协作请求的耗时、调用链、成败结果都记录上报，这样在 Agent 数量过 10 个、调用关系变复杂后，你还能用一张拓扑图快速定位性能瓶颈。

我个人最看好的是"语义增量"方向——让 Agent 在注册时除了 schema，还能挂上自己的 Prompt 模板和使用须知，调用方拿到的不只是参数接口，还有一些必要的业务语义。比如一个"财务 Agent"可以在 Profile 里说明"本 Agent 只处理对公账目，不走个人报销"，这样上层路由时就能避免把它推荐给不同业务线的需求方，减少无谓的调用。Agent 之间协作的终极形态，应该是像人和人协作那样，除了知道对方"能做什么"，还能理解对方"习惯怎么做"和"边界在哪"。