上个月我把家里三台设备一台工作笔记本、一台放在客厅的迷你主机、一台手机都装上了同一个 Agent 内核的实例本意是让它们各自待命、随时听我调用。结果真到用的时候发现一个很尴尬的问题A 设备上的 Agent 明明有联网检索能力我却不知道它在线B 设备上的 Agent 能执行本地文件操作可手机端根本够不着它的 API。每个 Agent 都像一座孤岛能力都在但触达不了。于是我把这个项目命名为 Agent-Reach——它不是一个新的 Agent 框架而是专门解决如何把已有的 Agent 能力暴露出去、路由过去、稳定地触达这一层问题的调度与连接层。这篇内容我会从设计动机、核心抽象、路由策略、最小闭环到生产环境翻车细节完整梳理一遍希望对正在做多端 Agent 互联、或者想把 Agent 能力服务化的朋友有帮助。1. 为什么我会去写一个叫 Agent-Reach 的调度层1.1 多端 Agent 各自为战的痛点如果你只有一个 Agent 跑在一台服务器上那确实不需要什么 Reach。但只要是个人开发者或者小团队很容易出现我开头说的那种情况内网一台高配机器跑着大模型推理实例办公电脑上跑着负责写周报的 Agent手机上的 Agent 负责语音备忘和日程提醒。拆开看每个都很轻量一旦想联动就废了。我的具体痛点是以下三条设备在线状态不可知。Agent 进程活着不代表它的服务端口可用更不代表它能执行某种能力。能力描述没有统一标准。A 设备说我能搜索B 设备说我支持 search_web同一个语义两种说法上游根本不敢自动路由。路由逻辑和业务代码耦合。每对接一个端就要在业务层写一遍该调用谁、失败怎么办接第三个端的时候代码已经像一团乱麻。这三条里任何一条都够让人烦的了合在一起基本宣告多端协作这个想法没法落地。我一开始尝试用现成的消息队列做通讯但 MQ 只管消息送达不管端的能力状态也不做路由决策送到之后依然不知道谁会处理。1.2 编排框架管的是流程不是触达我也认真考虑过直接用 LangGraph、CrewAI 这类编排框架去管理多 Agent。这些框架的优点很明显图编排、状态管理、多 Agent 协作模式都很成熟。但用了一段时间之后我意识到一个问题编排框架默认所有节点都已经暴露在同一个进程或同一套网络模型里。它们更关心这个 Agent 执行完之后下一个 Agent 是谁却不太关心这个 Agent 在另一栋楼、另一个网段、甚至开着飞行模式的手机上我到底怎么把它拽起来干活。说得直白一点编排框架管流程Reach 管触达。前者是神经网络里的决策层后者是让信号真正传达到肌肉的神经末梢。两者有交集但不等价。项目真正跑起来之后更常见的组合是用 Agent-Reach 做触达底座用编排框架做任务拆解Reach 把编排层产生的子任务分发给对应端上的 Agent 执行。1.3 Agent-Reach 的定位让能力声明与物理触达解耦所以我给 Agent-Reach 定下的设计目标就是一句话能力是一次声明触达是一个路由问题。业务调用方只描述我想完成什么能力Reach 层负责找到此刻能满足这个能力的端点完成实际请求并把结果送回。这个定位带来一个直接好处新接入一个设备端不需要改任何业务调用代码。只要新设备上的 Agent 启动时向 Reach 注册自己的能力它立刻进入可达池可以被调度。反过来某台设备下线了注册中心里它的心跳消失路由策略自然绕开它业务层完全无感。我还做了一个当时觉得有点多余、后来发现极其有用的设计把触达动作本身也建模成一个可观测资源。每一条请求从入站到出站都有 trace_id中间经历哪个端点、路由命中哪条规则、端点耗时多少全部记录下来。这个设计在后来的生产排查里救了我很多次我会在最后一章展开。2. Agent-Reach 的核心抽象从能力声明到触达路由2.1 ReachEndpoint触达端点的统一描述所有能被 Agent-Reach 路由过去的设备或进程都抽象为 ReachEndpoint。它既可以是物理设备比如手机、办公电脑也可以是逻辑进程比如一个跑在 Docker 里的服务。一个 Endpoint 最少包含三部分信息dataclass class ReachEndpoint: endpoint_id: str # 全局唯一 transport: str # 传输协议: http | websocket | mqtt | grpc address: str # 实际可达地址 capabilities: list[str] # 能力标识列表 metadata: dict # 扩展信息: 设备类型、地理位置、算力等举个例子我客厅那台迷你主机注册出来大概是这样的ReachEndpoint( endpoint_iddev-mini-01, transporthttp, addresshttp://192.168.1.42:8412, capabilities[web_search, summarize, image_describe], metadata{device_type: linux_mini, gpu: gtx_1660, cost_weight: 0.8} )关键点在于Endpoint 和任务执行是完全解耦的。Mini 主机的 Agent 不需要关心自己会被多少种类的任务调用它只需要维护自己的能力清单并在每次执行时告诉 Reach“这个能力我还支持”。有点像求职者把简历挂在招聘平台平台负责把合适的岗位推荐过来求职者不需要主动跑去每一家公司大楼前敲门。2.2 ReachableCapability能力声明的格式能力是让调用方和提供方建立契约的核心。我在设计能力声明格式时设了一个硬性条件同一个能力在任何端上必须使用同一个 Capability ID。说白了A 端叫web_searchB 端也必须叫web_search不允许出现语义相同但命名各异的混乱情况。完整的 Capability 声明里还包含输入输出 schema 和成本提示dataclass class ReachableCapability: capability_id: str input_schema: dict # JSON Schema output_schema: dict timeout_hint: float # 建议超时时间(秒) cost_per_call: float # 该端执行一次的大致成本为什么强制统一命名我在这上面踩过坑。最开始我图方便允许端上按自己的习惯注册web-search、webSearch、search_web之类五花八门的名字结果路由规则越写越多最后变成了维护一份别名映射表跟维护一份新旧字典似的非常反胃。后来我干脆写了一个启动校验器端上注册能力时Reach 会拿着全局能力字典比对未登记的 ID 直接拒绝注册并返回可用列表。这相当于给所有端立了规矩哪怕新人接入也能快速知道自己该说什么话。2.3 ReachRegistry注册中心与心跳机制ReachRegistry 是整个系统的“总路由表”负责维护所有 Endpoint 的在线状态和能力映射。我用了非常朴素但稳定的方案每个 Endpoint 启动后每 15 秒上报一次心跳超过 45 秒没有心跳就标记为离线不参与路由。这个时间窗口是可以配置的三倍心跳间隔是我当前的经验值容忍一次瞬时丢包又不会让假离线拖太久。心跳内容也尽量轻量化{ endpoint_id: dev-mini-01, capabilities: [web_search, summarize], updated_at: 1718100000, load_avg: 0.35 }这里有一个我后来才加进去的字段load_avg。它表示端点当前忙闲程度。路由策略会把负载也纳入决策避免所有任务都涌向同一台“看起来最全能”的机器。注册中心本身的实现没有追求复杂架构。由于我预期节点数量在百台以内直接用了 Python SQLite 内存字典缓存写入不落盘、读请求全走缓存。等以后节点规模到了上千再上 etcd 或者 Redis 集群也不迟。技术选型的原则我一直是能用简单方案撑住当前规模就不要提前引入复杂度。2.4 核心数据流一次触达请求的完整旅程一次业务调用的完整链路是这样的业务方发起请求帮我搜索最近关于 Agent 编排的论文。Agent-Reach 收到请求提取目标能力web_search带上请求参数和 trace_id。路由模块查询当前在线且声明了web_search的全部 Endpoint。按路由策略见下一节选出最优端点比如dev-mini-01。Reach 把请求通过 HTTP POST 发给该端点的/execute接口带上请求 token。端点上的 Agent 执行对应插件返回结果。Reach 校验结果 schema、记录耗时和状态把最终结果返回业务方。在张图换成人话就是业务方不需要知道谁在帮我搜它只需要知道搜到了结果。而 Reach 就是那个替它找人干活、并且负责交付的包工头。3. 路由策略的取舍优先级、成本感知与容错降级路由是 Agent-Reach 里最有意思的部分也是决定触达质量的核心。我前前后后写了四版路由策略每一版都对应一套真实摔过跤的场景。3.1 静态优先级路由的失灵场景第一版是最朴素的静态优先级Endpoint 注册时声明一个priorityReach 永远选优先级最高的在线端点。跑了一周就发现问题我给了办公室高性能台式机最高优先级它也确实强但一到白天它就变成高负载状态——有人在上面跑训练任务、开 IDE、跑数据库。这时候它回复速度从 0.8 秒涨到 8 秒但因为我设了最高优先级Reach 还是执拗地把任务全派给它。其他闲着的低优先级设备在旁边干瞪眼。此后的版本我再也没有用过唯一固定优先级这类策略。优先级可以存在但它只是权重因子之一不应该是唯一的决策变量。3.2 成本感知路由让 token 成本进入决策第二版引入了成本感知。每个 Endpoint 的 metadata 里有一个cost_weight表示执行一次典型任务的大致开销可以是钱、可以是算力占用、也可以是电力消耗。同时每次路由时还会结合能力匹配度给 Endpoint 打分。我常用的打分模型是三元组合能力匹配度是否精确命中请求能力、成本权重越低越好、健康度心跳新鲜度与负载均值。最终得分是这三项的加权和然后选得分最高的端点。def score_endpoint(ep, request) - float: capability_score 1.0 if request.capability in ep.capabilities else 0.0 health_score 1.0 / (1 ep.load_avg) cost_score 1.0 / (1 ep.metadata.get(cost_weight, 1.0)) return 0.4 * capability_score 0.3 * health_score 0.3 * cost_score这个公式不复杂但我建议参数按自己场景调别照抄。我这里 0.4 / 0.3 / 0.3 的含义是能力必须是硬门槛健康度和成本一个看实时、一个看长期。如果你的端之间成本差异很大就把 cost 的权重拉到 0.5 以上。3.3 容错降级请求失败后怎么接力再好的路由也会遇到端点突然崩溃。我给每次请求设置了一个失败转移链候选端点不只是选一个而是按得分排前三。如果第一个在超时时间内没响应Reach 自动把同一请求转发给第二个以此类推。这个机制我起了个名字叫 failover chain。需要注意的是并不是所有请求都适合自动 failover。比如创建文件这种操作如果第一台机器已经执行成功但响应丢了转发给第二台就会产生重复文件。所以我给 Capability 声明里加了一个字段叫idempotent标记该能力是否幂等。只有幂等能力才允许自动转移非幂等能力失败时直接向调用方抛错让人来决定要不要重试。我也因此设计出了一套全局 Request ID 和幂等 token 机制后面会说到。触达层还需要一个快速失败防线每个端点根据自己的历史响应时间报告一个建议超时值Reach 在调用时会带上这个值乘以 1.5 作为兜底超时。如果端点 500 毫秒就能干完的活调用方却摆了个 60 秒的超时那是把系统调度的可靠性寄托在运气上。3.4 与集中式 Agent 网关的对比顺手对比一下 Agent-Reach 和常见的 API 网关/AI Agent 网关。传统网关解决的是外部请求怎么进到内部系统重心在流量控制、鉴权、限流Agent-Reach 更关注内部 Agent 能力怎么分发到对应执行端。两者可以共存Reach 完全可以挂在网关后面由网关做对外鉴权Reach 做对内触达。维度API 网关Agent-Reach核心关注点入站流量治理出站能力触达路由依据URL/Header/权重能力声明/成本/健康度目标对象微服务实例多端 Agent 实例失败处理重试/熔断能力幂等判断 failover chain4. 跑通一个跨端触达 Demo从注册到执行的最小闭环这章把前面的抽象落到代码上带你把最小闭环跑起来。我用的是 Python 3.11 FastAPI不依赖任何重型中间件你本地跑通这套 Demo 大概只需要二十分钟。4.1 环境与依赖准备先建一个虚拟环境装三个依赖python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn httpx目录结构我也列一下方便你对号入座agent-reach/ ├── reach/ │ ├── __init__.py │ ├── registry.py # 注册中心 │ ├── router.py # 路由与 failover │ └── server.py # Reach 服务主入口 └── endpoint/ └── mini_agent.py # 模拟设备端 Agent4.2 实现 Reach 服务端先写最核心的注册中心。我用内存字典充当存储真实场景可以换 SQLite 或 Redis。# reach/registry.py import time class ReachRegistry: def __init__(self): self.endpoints {} self.capability_index {} def register(self, ep: dict): ep[last_seen] time.time() ep[status] online self.endpoints[ep[endpoint_id]] ep for cap in ep[capabilities]: self.capability_index.setdefault(cap, set()).add(ep[endpoint_id]) def heartbeat(self, endpoint_id: str, load_avg: float): ep self.endpoints.get(endpoint_id) if ep: ep[last_seen] time.time() ep[load_avg] load_avg def online_endpoints_for(self, capability: str) - list[dict]: now time.time() result [] for eid in self.capability_index.get(capability, []): ep self.endpoints[eid] if now - ep[last_seen] 45: result.append(ep) return result然后是路由模块加上 failover chain 逻辑。# reach/router.py from .registry import ReachRegistry class ReachRouter: def __init__(self, registry: ReachRegistry): self.registry registry def select(self, capability: str, limit: int 3) - list[dict]: candidates self.registry.online_endpoints_for(capability) scored sorted(candidates, keylambda ep: self._score(ep), reverseTrue) return scored[:limit] def _score(self, ep: dict) - float: capability_score 1.0 health_score 1.0 / (1 ep.get(load_avg, 0.0)) cost_score 1.0 / (1 ep.get(cost_weight, 1.0)) return 0.4 * capability_score 0.3 * health_score 0.3 * cost_score最后是 Reach 服务主入口处理注册、心跳、请求转发三个接口。# reach/server.py import httpx from fastapi import FastAPI from .registry import ReachRegistry from .router import ReachRouter app FastAPI() registry ReachRegistry() router ReachRouter(registry) app.post(/register) def register(ep: dict): registry.register(ep) return {status: ok} app.post(/heartbeat) def heartbeat(payload: dict): registry.heartbeat(payload[endpoint_id], payload[load_avg]) return {status: ok} app.post(/invoke) async def invoke(request: dict): capability request[capability] candidates router.select(capability) if not candidates: return {status: error, message: no reachable endpoint} for ep in candidates: try: async with httpx.AsyncClient(timeout15) as client: resp await client.post(ep[address] /execute, json{ capability: capability, payload: request.get(payload), token: request.get(token), }) if resp.status_code 200: return {status: ok, data: resp.json(), endpoint: ep[endpoint_id]} except httpx.TimeoutException: continue except httpx.ConnectError: continue return {status: error, message: all candidates failed}这里我要强调一个细节invoke接口对每个候选端点按顺序尝试看似简单但正是这个依次尝试构成了 failover chain 的最简形态。真实生产版本中我会再加入并发试探和熔断统计但最小 Demo 里让逻辑清晰比多加机制更重要。4.3 模拟设备端 Agent设备端要做的事情很简单启动时向 Reach 注册自己然后开一个/execute接口等待调用。这里我模拟一个只会做摘要的轻量 Agent。# endpoint/mini_agent.py import httpx import random import uvicorn from fastapi import FastAPI, Request app FastAPI() REACH_URL http://127.0.0.1:8000 ENDPOINT_ID dev-mini-01 MY_PORT 8412 app.post(/execute) async def execute(req: Request): body await req.json() capability body[capability] if capability ! summarize: return {key: simulate_extract, status: error, message: unsupported} text body.get(payload, {}).get(text, ) summary f摘要(前30字): {text[:30]}... return {status: ok, result: summary} if __name__ __main__: with httpx.Client() as client: client.post(f{REACH_URL}/register, json{ endpoint_id: ENDPOINT_ID, transport: http, address: fhttp://127.0.0.1:{MY_PORT}, capabilities: [summarize], cost_weight: round(random.uniform(0.3, 1.0), 2), }) uvicorn.run(app, host0.0.0.0, portMY_PORT)启动顺序是先启动 Reach 服务端再启动设备端。设备端启动时会自动注册注册完成后我们直接调用 Reach 的/invoke接口就能触发一次完整的触达。4.4 验证闭环效果我用 curl 发起一次调用curl -X POST http://127.0.0.1:8000/invoke \ -H Content-Type: application/json \ -d {capability: summarize, payload: {text: Agent-Reach 是一套面向多端 Agent 的触达调度层主要解决能力声明与物理触达之间的解耦问题。}}返回结果类似{status: ok, data: {status: ok, result: 摘要(前30字): Agent-Reach 是一套面向多端 Agent 的触达调度层主要解决能力声...}, endpoint: dev-mini-01}到这里最小闭环就跑通了。你看整个流程里调用方只知道我需要 summarize 能力它完全不关心是哪台机器、什么配置、走什么协议完成的。这套解耦逻辑就是 Reach 的立身之本。5. 生产环境下最容易翻车的五个细节Demo 跑通只是起点。真正把这套东西放到持续运行的环境里我遇到的坑基本集中在下面五个地方。这些都不是什么高深技术但每个都让我改了不少代码。5.1 触达超时与幂等边界第一个坑是超时设置。设备端可能因为正在做别的推理任务响应慢得离谱。我没设置合理超时的时候一次 invoke 能挂十分钟把业务线程池全拖死。后来我把所有出站请求的超时强制分成三档连接超时3 秒当前能力建议超时Capability 里的timeout_hint兜底超时timeout_hint * 1.5对应的代码片段timeout httpx.Timeout(connect3.0, readcapability_timeout * 1.5)幂等的问题是更隐蔽的坑。我在 3.3 节提过自动 failover但如果端点执行成功后进程就崩了响应根本没回到 ReachReach 就会把任务发给第二个端点造成重复执行。为了解决这个问题我给每个请求生成全局唯一的request_id并让端点在执行前先查一下自己的幂等表发现同一个request_id已经执行过就直接返回上次结果。幂等表的实现一点都不复杂# 端点侧执行函数 idempotency_store {} async def execute_with_idempotency(body): req_id body.get(request_id) if req_id in idempotency_store: return idempotency_store[req_id] result await do_actual_work(body) idempotency_store[req_id] result return result代价是存储占用。我的做法是每个端只保留最近 24 小时的幂等记录超过时间窗口直接删同时只对声明了idempotent: true的能力做幂等校验其他能力直接执行。5.2 注册与鉴权边界第二个坑发生在多端接入后我最初为了让 Demo 简单/register接口没有任何鉴权结果内网里一台被挖矿脚本感染的主机恶意注册了几百个假端点把路由表刷得乌烟瘴气。解决方式说穿了不值几个钱注册和心跳接口必须使用预共享密钥PSK校验开发环境用环境变量注入生产环境用密钥管理服务。新端点注册后先进入pending状态不会参与路由只有管理员手动确认或端点在规定时间内完成一次握手验证才转为online。事后我把这个机制总结为触达之前绝对不信任。你可以在任何复杂的分布式系统里加各种花哨的加密协议但最底层的边界永远是——不确认身份不分配流量。5.3 能力版本兼容第三个坑来自 Agent 升级。我的手机端 Agent 升级了某个库之后我的web_search能力返回结果从 JSON 变成了 Markdown。Reach 没有校验输出格式依然把结果原样返回给调用方结果下游所有解析逻辑全部崩了。解决办法是在 Capability 声明里带上input_schema和output_schemaReach 在返回结果前用 jsonschema 做一次轻量校验。虽然轻微增加响应延迟大约 1~2ms但换来的是接口契约稳定这个开销完全可以接受。同时每次端上 Agent 升级都应该在注册信息里递增一个version字段Reach 在路由时可以按调用方要求只选特定版本范围内的端点。5.4 日志中的敏感信息治理第四个坑比较冷门但很致命我最初为了排查问题方便把完整的请求和响应都打印到日志里。结果有一次用户让 Agent读取一下邮件里那个身份证号码——这个值穿过 Reach 的时候被完整记录在文本日志里了。这成了我的一个重大教训。我现在对 Reach 层做三层治理默认不记录请求体内容只记录 payload 的 schema 指纹比如长度、类型、必填字段数。对需要完整日志的调试环境单独用环境变量REACH_DEBUG开启并强制限制日志保留时间。对所有触达响应做字段级脱敏身份证号、手机号、邮箱等 PII 字段在日志层直接替换为***。我不建议等出事后补而是项目第一天就把这个意识焊在代码里。涉及用户数据的系统日志审计这个口子必须从前端打开。5.5 端点退化的主动探测最后一个细节是假在线问题。有些端点进程还活着心跳也正常上报但其内部依赖的某个插件服务已经崩了。比如迷你主机的 Agent 一直上报summarize能力在线但它依赖的本地推理服务已经 OOM实际一调用就报错。这种问题靠心跳发现不了必须靠真实请求探测。我加了一个主动健康检查Reach 每隔一段时间随机挑选一个在线端点发送一个非常廉价的 ping 任务比如让端点执行一个无副作用的echo操作。如果连续三次 ping 失败就把端点标记为degraded从路由候选里暂时摘除等它恢复了再放回去。这个机制本质上是一种轻量保活探针成本很低但效果立竿见影。6. 后续演进把触达做成可观测、可治理的基础设施Agent-Reach 现在的形态已经能稳定支撑我手头的多端协作场景了但我很清楚它离一个真正的基础设施还有距离。接下来我的演进方向集中在三个地方。6.1 触达链路可观测性我在 1.3 节提过每个请求都有trace_id但这个信息目前只存在日志里没有形成系统性的可观测能力。我计划把触达事件写入一个轻量事件流队列比如 Redis Stream做一个简单看板展示端点延迟分位数、错误率、路由命中分布。这样每次为什么请求这慢不再需要翻好几段日志拼时间线而是直接看链路视图。6.2 多租户与配额治理如果 Agent-Reach 要开放给团队其他人用就必须做隔离。每个调用方使用独立的 API KeyReach 根据租户维度分配资源配额每分钟多少次触达请求、允许路由到哪些端点组。我目前想到的最小实现是增加一层租户中间件在路由打分阶段强制过滤掉租户无权访问的端点并对请求进行令牌桶限流。6.3 离线任务的事件驱动触达现在的/invoke是同步请求调用方必须等端点干完活才能拿到结果。但现实里很多 Agent 任务耗时很长比如让手机端的 Agent 做一个多步骤的本地数据处理可能要跑好几分钟。同步等待既不经济也不优雅。我的计划是引入异步事件驱动模式调用方提交任务后立即拿到task_idReach 写入任务队列端点通过长轮询或 MQTT 订阅领取任务执行完成后异步回调结果。这套改造会引入消息中间件复杂度会上一个台阶所以需要谨慎设计接口语义。但触达这件事本来就是把任务可靠地送达并收口异步化之后才真正具备规模化调度的底气。把 Agent-Reach 从想法到落地这几个月我最大的感受是技术选型其实不难难的是在每一个差不多能用的瞬间忍住不凑合。心跳摘除、幂等表、主动探测这些机制没有一样是惊艳的发明但每一样都是把触达这件事从碰运气变成有保障的关键拼图。如果你也在做类似的多端 Agent 基础设施我的建议很简单——先让最小闭环跑通然后翻回来把这五个坑一个一个填上触达这件事就能站稳了。