1. “Agent-Reach”到底解决的是哪一刀大概两年前我在做多智能体协作项目时被一个怪异场景逼到抓狂每个 Agent 单独调起来都正常一旦组合在一起就频频出现互相找不到、任务重复执行、状态不同步的问题。翻遍日志才发现几乎所有 Agent 都维护了一份“别的 Agent 地址”的写死清单A 改一个端口B 就得跟着改配置改完还得重新部署。那段时间我最大的感受是Agent 本身再聪明Agent 之间的连接方式仍然是石器时代。后来我在团队内部沉淀了一套轻量级服务名字就叫“Agent-Reach”。它的定位非常单纯给一群 AI Agent 准备一个集中的注册、路由和可观测入口让 Agent 之间不再靠人肉配置互相连接而是通过“能力声明”被发现、被调度。打个比方你不需要记住同事的手机号只需要知道“找谁可以处理订单”前台会帮你转接。Agent-Reach 就是 Agent 世界的那个前台。如果你是正在做 Agent 编排、多智能体工作流或者准备把 Agent 接入现有业务系统的人这篇文章里的实现思路可以直接搬到你项目里。它不要求你引入多重的中间件核心就是一张注册表、一张路由表、几个异步任务门槛其实比想象中低很多。先说痛点来源。多 Agent 系统发展到一定规模必然撞上这几类问题第一连接关系写死。每个 Agent 的调用方直接持有 IP 和端口配置散落在各处。只要环境一变整个拓扑就跟着崩。第二能力边界模糊。你想让一个“摘要 Agent”把会议记录转给“搜索 Agent”得先搞清楚搜索 Agent 接受什么输入、返回什么格式这层全靠人工对齐沟通成本高得离谱。第三健康状态不可见。哪个 Agent 已经挂掉、哪个在抖动、哪个在高负载系统里没有任何一方能给出可靠答案失败重试全靠猜。Agent-Reach 做的事是把这三种混乱统一集中化但注意不是把所有逻辑都塞进中心而是把“注册、发现、路由、状态”这四件事收拢起来业务逻辑仍然留在各个 Agent 本地。这样既有中心的秩序又不至于让中心变成性能单点。很多人会问市面已有编排框架为什么还自己写一个我的回答是不少框架把 Agent 的通信、组织、甚至运行方式都捆得很重你一旦选型后续扩展就受制于人。Agent-Reach 刻意保持轻量它不决定 Agent 内部怎么规划 prompt、怎么调模型只解决“Agent 链条之间怎么触达”。如果你只想快速验证一个多 Agent 协作方案又不想引入重框架这条路径的性价比非常明显。另外Agent-Reach 专门为大模型场景做了几处简化。传统微服务里的服务发现偏重协议透传、负载均衡策略而在 Agent 场景里真正重要的往往是“能力语义”而不是地址。请求里写“找到昨天用户的反馈”路由层应该根据“搜索能力”这个语义去匹配 Agent而不是根据某个固定服务名。所以我把“服务名”升级为“能力标签”让路由更贴近自然语言表达。2. 架构设计与模块拆解注册表、路由表、状态机在动手写代码前我先把 Agent-Reach 拆成五个核心模块注册模块、能力目录、路由引擎、心跳与状态机、监控 API。这五个模块责任独立组合起来就是一条完整链路。模块职责与关键接口模块核心职责关键接口/数据结构Registry保存 Agent 基础信息、能力列表、endpointPOST /v1/agents/registerDELETE /v1/agents/{id}Heartbeat维护心跳、探测 Agent 在线状态POST /v1/agents/heartbeat异步后台检测Capability Index维护“能力标签 - Agent 列表”映射agent.capabilities 字段Router按任务描述匹配可用 AgentPOST /v1/routeMonitor查询注册总量、健康度、最近调用链GET /v1/statusGET /v1/agentsAgent 实体我建议最少保留这些字段agent_id、name、capabilities数组、endpoint、status、heartbeat_at、registered_at。其中 capabilities 是路由判断的依据重要程度排第一设计时务必想清楚你的能力标签体系。标签太细会频繁匹配失败标签太粗又会让路由失去区分度我后面会细讲这个平衡。状态机我做成三态healthy正常、degraded连续几次心跳丢失但仍可接管、offline超过阈值不再进入路由。实现里最好把状态迁移逻辑做成独立函数方便挂回调钩子。例如状态变为 offline 时自动触发一个通知 webhook把容错决策留给上层编排器。Router 是灵魂。它接收一个任务请求请求里包含语义化描述比如{task: 搜索过去7天的订单}。Router 先按关键词粗筛出候选 Agent再用可配置的评分函数打分排序。粗筛阶段我用的是“能力标签同义词表”同义词表既可以从配置文件读也可以由运营人员实时维护。候选为空则返回 404并带一句“没有 Agent 具备该能力”的提示方便调用方快速定位是标签问题还是 Agent 未注册问题。配置文件我建议保持极简用 YAML 或 JSON 都行。下面这份是我 MVP 阶段的配置注册模式、心跳超时、离线判定次数、是否启用 LLM 兜底路由。server: host: 0.0.0.0 port: 8000 registry: storage: sqlite sqlite_path: ./agent.db heartbeat: interval_seconds: 10 offline_after_misses: 3 router: use_llm_fallback: false llm_provider: openai auth: enabled: true api_tokens: - dev-token-123配置文件里有个点值得专门提醒offline_after_misses不要设成 1。如果设成 1一次网络抖动就会让 Agent 被误判下线Router 随后把所有任务都踢开它反而制造更大的不稳定。长连接和短轮询混用的场景下我最终用 3也就是连续三次心跳超时才开始降级这个参数是血泪教训换来的。数据存储我用 SQLite 打底生产环境可平滑切到 Redis。SQLite 的好处是单文件、零运维开发和单机部署都非常舒服。注意 SQLite 不适合极高并发写入但 Agent 注册和心跳的并发量级通常不高在接入 10 万级 Agent 之前完全扛得住。真要切 Redis只需把 Registry 存储接口替换成 Redis Hash不影响上层 Router。架构层核心流程是Agent 启动后向 Registry 注册把 endpoint、能力数组、元数据同步上来Registry 写库后返回 agent_id 和心跳 tokenAgent 再按固定频率心跳。业务侧发起任务时Router 根据任务描述做能力匹配选中最优 Agent把任务以 HTTP POST 转发过去并跟踪任务状态。有人问为什么不用消息队列做任务传递我并非排斥 MQ而是觉得 MQ 会带来额外部署成本且 Agent 的响应通常需要同步拿结果HTTP 回调最直观。如果未来任务量真的大到一定程度可以在 Agent-Reach 和 Agent 之间加任务队列但路由决策仍然留在 Agent-Reach 内部不用跟着换。3. 核心代码实现FastAPI 版 Agent-Reach 落地步骤这一节直接上可运行代码。我选 FastAPI因为它原生支持异步接口开发效率高文档自动生成Agent 团队里的成员能快速看懂。依赖只有 fastapi、uvicorn、sqlite3 标准库和可选的 pydantic。先定义数据模型。Agent 注册的请求体我限定必填字段避免脏数据入库。from datetime import datetime from typing import Optional, List from pydantic import BaseModel, Field class AgentRegisterRequest(BaseModel): name: str Field(..., min_length1, max_length100) endpoint: str Field(..., descriptionAgent 的可访问地址) capabilities: List[str] Field(..., min_length1) meta: Optional[dict] None class HeartbeatRequest(BaseModel): agent_id: str token: str这里把 token 单独拿出来是为了防伪装注册。MVP 阶段先做简单校验生产环境要换成签名的动态 token后面坑的部分细说。Registry 存储层封装一个类挂着三个最核心的方法register、heartbeat、find_candidates。import sqlite3 import uuid import time import json class AgentRegistry: def __init__(self, db_path: str ./agent.db): self.db_path db_path self.init_db() def _conn(self): conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def init_db(self): with self._conn() as conn: conn.execute( CREATE TABLE IF NOT EXISTS agents ( agent_id TEXT PRIMARY KEY, name TEXT, endpoint TEXT, status TEXT, capabilities TEXT, heartbeat_at REAL, registered_at REAL ) )注册方法生成 agent_id并把 capabilities 序列化成 JSON 字符串存储。SQLite 没有原生数组类型实战里这样最简单读出来用json.loads还原即可。def register(self, req: AgentRegisterRequest): agent_id uuid.uuid4().hex[:12] now time.time() with self._conn() as conn: conn.execute( INSERT INTO agents (agent_id, name, endpoint, status, capabilities, heartbeat_at, registered_at) VALUES (?, ?, ?, ?, ?, ?, ?) , (agent_id, req.name, req.endpoint, healthy, json.dumps(req.capabilities), now, now) ) return agent_id心跳更新就是一条 UPDATE关键点是同时更新 heartbeat_at 和 status。如果入库前 Agent 是 degraded收到心跳就先恢复 healthy比等后台任务复查快得多。这里有个实现细节值得说明我用了last_heartbeat_seconds_ago这样的辅助判断函数。因为心跳和状态检查是两套时间线状态检查是后台任务每分钟扫一次心跳却可能每 10 秒来一次直接基于 heartbeat_at 差值判断比在内存里维护字典更可靠重启服务不丢状态。路由初版用同义词粗筛加分数排序def find_candidates(self, task_desc: str, top_k: int 3): with self._conn() as conn: rows conn.execute( SELECT * FROM agents WHERE status healthy ).fetchall() scored [] for row in rows: caps json.loads(row[capabilities]) score 0 for cap in caps: for kw in KEYWORD_MAP.get(cap, []): if kw in task_desc: score 1 if score 0: scored.append((score, row)) scored.sort(keylambda x: x[0], reverseTrue) return [r[1] for r in scored[:top_k]]KEYWORD_MAP 放单独配置文件本质是一个能力到近义关键词的映射。比如“搜索”能力映射到[搜索, find, 检索, 查询]。这层非常廉价却能把大规模候选集快速收敛到个位数之后即使让 LLM 做精细排序压力也小很多。FastAPI 接口层很薄注册、心跳、路由各一个文件。路由接口接收 task 描述后调用 find_candidates有候选就把任务转发给得分最高的 Agent再等 Agent 响应。转发时统一用 httpx.AsyncClient 发异步请求避免阻塞事件循环。转发任务时我会在请求头或 body 里附带一个 request_id这是全链路追踪的抓手排查超时、重复执行都靠它。兼容 OpenTelemetry traceparent 头的做法线下我再展开初期只需要保证每条链路有唯一 ID。健康检查后台任务我用最简单的 asyncio 循环每 60 秒扫一次所有 Agent把 heartbeat_at 超过阈值但还在 healthy 的标记为 degraded超过两倍阈值标记为 offline。不要用 threadingFastAPI 事件循环和 threading 混用容易踩共享状态的坑直接 asyncio.create_task 最省心。到这里Agent-Reach 主体 300 行以内跑起来。它不是算法项目而是工程骨架。骨架扎实以后真正花精力的是业务对接和参数调优。4. 本地联调从注册到任务转发的一次完整演练光有代码不够我给你看一条完整调用链。下面模拟双 Agent 场景搜索 Agent 负责检索本地文件摘要 Agent 把长文本压缩成要点。业务方要完成“搜索最近三天的日志并总结异常”。先启动 Agent-Reach然后把两个 Agent 拉起来。两个 Agent 第一件事是注册curl -X POST http://localhost:8000/v1/agents/register \ -H Content-Type: application/json \ -H Authorization: Bearer dev-token-123 \ -d { name: search-agent, endpoint: http://localhost:9001/process, capabilities: [搜索, 检索, log_lookup] }搜索 Agent 注册后会拿到类似d45f9a2c1b0e的 agent_id。摘要 Agent 用同样方式注册能力标签是[摘要, 总结, summary]。此时服务端能看到两条健康记录。注册完心跳进程每 10 秒刷一次。查状态可以用curl -H Authorization: Bearer dev-token-123 http://localhost:8000/v1/agents返回里有每个 Agent 的 status、heartbeat_at、capabilities。这里有个容易被忽视的细节心跳请求返回 200 只代表 Agent 的注册进程活着不代表业务健康。想真正验证业务健康Agent 必须在心跳里附带轻量自检结果比如最近一次任务是否成功。我后来把这块做成了可选的health_payload字段业务自检复杂的 Agent 可以自行实现。业务侧发起任务curl -X POST http://localhost:8000/v1/route \ -H Content-Type: application/json \ -H Authorization: Bearer dev-token-123 \ -d { task: 搜索最近三天的日志, payload: {log_dir: /var/log/app, days: 3}, expect_callback: false }Router 先做关键词粗筛任务描述里出现“搜索”命中搜索 Agent 的能力映射搜索 Agent score 为 1摘要 Agent score 为 0路由结果只有一个候选。Agent-Reach 直接把 payload 转发到搜索 Agent 的/process接口。搜索 Agent 在/process里完成文件检索结果返回给 Agent-ReachAgent-Reach 原样返回调用方。这一步我刻意没有加“链式自动触发”因为链式调用容易放大错误。MVP 阶段建议由调用方决定下一步业务稳定后再通过配置开启next_agent自动流转。接下来是链式场景。调用方拿到搜索 Agent 结果后发起第二个任务curl -X POST http://localhost:8000/v1/route \ -H Content-Type: application/json \ -H Authorization: Bearer dev-token-123 \ -d { task: 总结异常日志, payload: {logs: [...]} }这次“总结”命中摘要 Agent路由把它转发给摘要 Agent。你会发现链路控制权始终在调用方手里Agent-Reach 负责的关键点是“正确触达”而不是替业务编排完整流程。联调阶段我强烈建议在路由接口返回里带上matched_agents数组。这样即使路由成功也能知道 Router 考虑了哪些候选、打分如何。生产关掉这个开关开发联调时它能省大量沟通时间。这套链路跑通之后你会对 Agent-Reach 形成直观认识注册表管“谁在线、谁能干什么”路由管“任务该给谁”健康检查管“谁掉线要踢出候选”。三者配合就是一个可靠的 Agent 协作底座。关于 mock 联调我还有一个百试百灵的小技巧在注册接口里加一个mocktrue参数允许 Agent 先用本地模拟进程完成注册。真实 Agent 没开发完时先用一个返回固定数据的 mock 进程把链路打通等真实 Agent 就绪直接替换进程即可架构零改动。我在多个项目里靠这个技巧把联调周期压缩了一半。5. 生产落地踩坑记录五个坑我花了一周才填平代码能跑通只是开始。下面这些坑按出现频率从高到低排都是我真实踩过的。坑一心跳参数没调好健康 Agent 被频繁误判离线。初版心跳间隔 5 秒offline_after_misses 设 2。一次线上发布所有 Agent 集体抖动Router 瞬间踢开所有候选业务直接挂掉。后来改成心跳间隔 10 秒、offline_after_misses 为 3同时给心跳接口加幂等性同一秒收到多次心跳只取最新一次避免数据库写放大。坑二任务执行时间远大于 HTTP 超时。Agent 调大模型动辄几十秒调用方设置 5 秒超时必然失败。这个问题靠同步等待永远无解。我最终引入 task_status 模型路由接口收到任务立即返回 request_idAgent 完成后回调 Agent-Reach 更新状态调用方通过GET /v1/tasks/{request_id}轮询结果。长任务和短任务都统一暴露成同一个状态查询协议。坑三注册接口没有鉴权任何人都能塞假 Agent 进注册中心。一旦混入 endpoint 指向恶意地址的 AgentRouter 就会把任务转发过去这是很严重的信息安全问题。MVP 阶段先做固定 API Token 校验注册和心跳请求必须携带共享 token。后续可以走 mTLS 思路为每个 Agent 下发独立密钥并在 token 里绑定 agent_id。这里不要偷懒哪怕先做静态 token也能拦住大部分误操作。坑四Agent 数量少时按分数选最强策略导致单点过载。早期候选集只有两三个 AgentRouter 每次都选 score 最高的另一个始终闲置。后来加入加权随机score 相同的候选按权重轮询score 高但正处理长任务的 Agent 会被临时降权。权重信息放在心跳 payload 里包含当前排队任务数和平均处理耗时。排序公式初版是final_score base_score * (1 / (1 busy_factor))实测能显著改善资源偏斜。坑五LLM 兜底路由看似智能延迟和成本都不可忽视。我试过让 Router 每个任务都调一次大模型判断能力归属结果路由本身比 Agent 执行还慢模型偶发误判还会导致任务投错。最后我把 LLM 兜底开关默认关闭只在关键词粗筛候选为空时才触发。触发频率低、效果可控成本也压得住。除坑以外还有一条边界经验非常值钱不要把业务状态也塞进 Agent-Reach。早期我想让 Agent 把中间结果都上报中心为的是全局可视。结果中间结果越来越多数据结构越改越重中心逐渐变成大泥球。后来我划了一条清晰边界——Agent-Reach 只管连接元数据注册信息、健康状态、任务调度状态业务数据由 Agent 自己存、自己提供查询接口。中心展示进度时只展示状态字段不展示 payload 内容。这条边界让稳定性直接上了一个台阶。日志规范我建议第一天就统一。Agent-Reach 输出的日志如果包含 request_id、agent_id、task_desc、耗时、结果码五个字段后续排查线上问题能缩短一半时间。我在路由转发外层包了一个log_request函数统一打印这五个字段。动作很简单收益远超预期。最后再分享一个心得多 Agent 系统上线初期不要追求路由策略的完美。先把注册、心跳、健康检查这几个地基打牢再慢慢加能力匹配权重、任务队列和监控告警。地基不稳的时候任何花哨路由都是空中楼阁。Agent-Reach 的价值不在于多聪明而在于让 Agent 之间的连接有秩序、可观测、可演进。它是我在多 Agent 路上踩了无数坑之后沉淀下来的骨架希望你用的时候能少踩几个。