我刚接触智能体开发时一直没想明白一个问题每个 Agent智能体都跑得好好的但怎么让它们互相找到、互相喊话我的环境里有一台本地工作站、一台跑批的云服务器、还有一块树莓派上面各自挂着不同的 Agent有做 PDF 文本抽取的、有做图像分类的、有负责定时发通知的。平时调试时我要分别记住 IP、端口、认证方式用不同客户端去连一旦某个机器重启或者 IP 变了整个人就会陷入“翻配置”状态。---后来我动手做了个周末项目Agent-Reach把它定位成一个轻量级的智能体触达中间件。它做的事情非常简单所有 Agent 启动后先向 Agent-Reach 注册拿到一个唯一的 ID 和通道之后我只需要向 Agent-Reach 发一条标准消息它就能把消息精确投递给对应的 Agent 并拉回结果。整个过程对客户端几乎没有侵入也不需要写一堆胶水代码。这篇文章把项目拆解、核心机制、部署过程和踩坑经验完整记录下来适合跟我一样在折腾多 Agent 协同、家庭自动化或者想在 NAS 上统一管理智能体的朋友参考。1. 拆解 Agent-Reach 的设计思路1.1 分布式 Agent 协同里的“最后一公里”现在做多 Agent 系统大家最不缺的是单个模型的推理能力缺的是“触达”能力。这里触达包含三层意思一是网络层面能不能连上二是语义层面能不能对上三是运维层面能不能管住。很多团队直接用 WebSocket 点对点写死通信短时间跑着没问题但一旦 Agent 数量从两个变成十个你就会发现有一半精力花在解决“谁掉线了”“消息没收到”“回调地址写错”这类破事上。Agent-Reach 想解决的正是这最后一公里。它把自己藏在所有 Agent 的背后成为一个虚拟总线不处理业务逻辑只负责注册、路由、投递和回执。本质上和家里的集线器类似——你不用记住每个插座在哪个房间只要知道总开关在哪按个按钮就能把电送到该去的地方。1.2 为什么不是 MQTT也不是纯 HTTP做这个项目之前我认真比较过两种常见方案。MQTT 本身非常适合低功耗设备带有 Broker 和 QoS 机制但它的 topic 模型比较线性表达“给某个能力为ocr的 Agent 发任务”这类请求不够自然而且需要额外部署一个 MQTT Broker纯 HTTP 最直观可以 RESTful 一把梭但要求 Agent 必须暴露一个公网可访问的端口在家庭网络、公司 NAT 后面基本走不通还得自己做轮询和超时管理。Agent-Reach 采用“注册中心 双向长连接”的混合架构所有 Agent 作为客户端主动连接到 Hub不要求 Agent 开放入站端口Hub 维护一份动态注册表记录 Agent 的 ID、能力标签、状态调用方不管是人还是另一个 Agent只和 Hub 通信由 Hub 负责路由。特性MQTT纯 HTTPAgent-Reach设备入站端口不需要需要不需要语义路由弱需自行实现原生支持结果回执需自己叠加需自己叠加原生支持部署复杂度需要 Broker低中离线缓存支持 QoS无支持暂存表格能看出取舍Agent-Reach 不是技术上最炫的但在“自托管多智能体触达”这个具体问题上是平衡得比较好的。这套设计思路是反复迭代后的选择初版试过纯 HTTP 轮询后来发现连接数量和实时性撑不住才改成现在的长连接。1.3 设计目标十分钟接入一个 Agent我给自己定的硬性指标有三个。第一客户端要轻。一个 Agent 只需要装一个很小的 SDKPython 版本核心代码不到 500 行不依赖重型框架。用 Go、Node 甚至 shell script 也能通过 HTTP API 接入保证语言无关。第二要能离线缓存。Agent 断网重连后错过的任务如果直接丢掉那么很多自动化流程就会静默失败。所以 Hub 端需要一个简单的持久化队列存住一段时间内无人认领的任务。第三状态可观测。我不能接受“消息发出去了但不知道 Agent 到底收到没有”的状态。所以每条任务必须有生命周期已投递、已接收、已执行、已回执、已超时。这些目标决定了后续每一个细节的实现方式。代码写起来很克制没有刻意引入微服务就是一个 Python 异步进程加 SQLite 存储一个 Docker 镜像搞定。2. 核心机制与关键模块实现2.1 Agent 注册中心门牌号与电话簿注册中心是 Agent-Reach 的心脏。每个 Agent 上线时必须上报三个东西身份信息全局唯一 ID格式为agent-{namespace}-{name}-{random}例如agent-main-ocr-3f9a元数据能力标签、版本号、健康检查路径、回调 base URL运行时状态当前是否忙碌、最大并发数、最近心跳时间。Hub 把这些信息写入一个内存 dict 的同时异步刷进 SQLite。读取路径全部走内存保证毫秒级查询写入路径异步落盘避免 I/O 阻塞关键链路。心跳机制是注册中心最容易被忽略又最关键的细节。每个 Agent 默认每 30 秒向 Hub 发送一个PING报文Hub 返回PONG并更新“最近心跳时间”。如果超过 90 秒即 3 个心跳周期没有收到任何报文Hub 会把这个 Agent 标为quasi-offline意思是“可能还在但我不确定”不立即摘除再过 60 秒仍然没心跳就彻底标记为offline。为什么用两段式状态因为我踩过坑树莓派的 Wi-Fi 偶发抖动一个 2 秒的瞬断就会让 Agent 被误下线然后一堆任务发给离线节点全超时。两段式状态能避免在边缘场景下做出错误决策。心跳报文本身是空的小 JSON但我会在报文里附上 agent 当前的内存占用和队列深度这样 Hub 可以做个简单的负载收集为后续“按能力选择最优 Agent”提供依据。2.2 消息路由按能力调用的三种模式注册表搞定后路由就很简单了。Agent-Reach 支持三种投递模式点对点Direct调用方明确指定目标agent_idHub 直接检查该 Agent 是否在线在线就投递。这个模式用在明确知道要调用哪个实例的场景。按能力调度Capability调用方不关心具体哪个 Agent 执行只需要声明“我要能 OCR 的”。Hub 会在在线 Agent 里过滤出带ocr标签的列表再按负载和最近响应时间排序选最优的一个投递。这是我最常用的模式因为可以把 Agent 扩容、缩容完全透明化。扇出广播Fanout调用方指定标签Hub 把消息并行投递给所有带该标签的 Agent。这个用在“通知所有设备”“刷新所有缓存”这类场景。消息体采用类 JSON-RPC 格式{ id: task-01HQZ6..., method: agent.reach.task, params: { request_id: req_123456, target: { type: capability, value: ocr }, payload: { input: documents/contract.pdf }, timeout_ms: 30000 } }每个请求都带一个全局唯一的request_id这是后面做去重和追踪的基础。Hub 收到消息后先落库再路由确保崩溃重启后还能知道“这条消息到底发出去没有”。2.3 安全与身份不让陌生 Agent 混进来让 Agent 接入一个中间件最怕的是任何人伪造身份往里灌消息。Agent-Reach 的安全设计分三层接入层Agent 和 Hub 之间的 WebSocket 连接强制走 TLSHub 侧配置自签名证书或 Lets Encrypt 证书。认证层每个 Agent 在创建时分配一个 token启动后用这个 token 换成短期会话票据后续所有消息头都带这个票据Hub 端校验签名。授权层注册元数据里有一个acl字段声明“这个 Agent 可以接受哪些 namespace 的任务”。比如ocrAgent 只能接收来自mainnamespace 的任务避免被其他乱七八糟的任务源打爆。实现上认证用 HMAC-SHA256 对当前时间戳加 agent_id 签名Hub 端用同一把密钥验签防止重放。这套方案比 OAuth2.0 轻很多对自托管工具已经够用。如果以后要暴露到外网可以再加一层反向代理的白名单锁来源 IP。3. 实操过程从零搭起 Agent-Reach3.1 启动 Hub注册中心我用 Docker Compose 把 Hub 跑起来配置文件写得很简练。version: 3.8 services: agent-reach-hub: image: agentreach/hub:0.9.0 container_name: agent-reach-hub restart: unless-stopped ports: - 8080:8080 # HTTP API - 8081:8081 # WebSocket 接入端口 environment: REACH_DATA_DIR: /data REACH_AUTH_SECRET: change-me-to-a-long-random-string REACH_DEFAULT_LEASE_TTL: 90s REACH_TASK_TTL: 3600s volumes: - ./data:/data healthcheck: test: [CMD, curl, -f, http://localhost:8080/healthz] interval: 10s timeout: 3s retries: 3启动后先检查健康接口curl http://localhost:8080/healthz如果返回{status:ok}说明 Hub 起来了。接着要有第一个账号才能创建 Agent。Agent-Reach 提供一个简易的引导 API首次启动可以通过环境变量里配置的REACH_BOOTSTRAP_TOKEN调用curl -X POST http://localhost:8080/v1/agents \ -H Authorization: Bearer some-bootstrap-token \ -H Content-Type: application/json \ -d { name: ocr-agent-01, namespace: main, tags: [ocr, image], callback_base_url: http://192.168.1.20:9001, lease_ttl: 90 }返回值里带agent_id和agent_token第一次显示后不再重复展示所以建议当场存到密码管理器。3.2 用 Python SDK 接入一个模拟 Agent为了让读者更直观地看到接入过程我用一个模拟 OCR 的 Python 脚本做演示。核心代码就几块先初始化客户端然后注册自己之后进入消息监听循环。import time from agent_reach_sdk import AgentClient def handle_task(message: dict) - dict: # 模拟 OCR 过程实际在这里调用你的推理库 payload message[payload] time.sleep(1) return { status: ok, output: ffake_ocr_result_for_{payload[input]}, latency_ms: 1020, } def main(): client AgentClient( hub_urlwss://reach.example.com:8081, agent_idagent-main-ocr-3f9a, tokenyour-agent-token, tags[ocr, image], healthcheck_interval30, ) client.register() client.subscribe(handlerhandle_task) client.start_heartbeat() print(Agent is online and waiting for tasks...) while True: time.sleep(1)SDK 内部做了这些事与 Hub 建立 WebSocket 连接发送注册信息等待REGISTER_ACK启动后台心跳线程定时发送PING监听服务端下发的EXEC消息调用上层业务函数把业务函数结果封装成RESULT消息回传给 Hub。实际跑起来后你会看到类似下面的日志[INFO] Connected to hub at wss://reach.example.com:8081 [INFO] Register ack received, agent_idagent-main-ocr-3f9a [INFO] Health reporter started (interval30s) [INFO] Waiting for tasks... [INFO] Received task req_123456, dispatching to handler [INFO] Task req_123456 completed in 1020ms, result sent.3.3 通过 HTTP API 触达并返回结果Agent 在线后从任意一台能访问 Hub 的机器发任务。比如调用带ocr能力的 Agentcurl -X POST http://localhost:8080/v1/tasks \ -H Authorization: Bearer caller-token \ -H Content-Type: application/json \ -d { request_id: req_123456, target: {type: capability, value: ocr}, payload: {input: documents/contract.pdf}, timeout_ms: 30000 }Agent-Reach 是异步模式这个接口会立刻返回一条任务状态记录不会同步等待 Agent 执行完。如果你需要同步结果有两种选择轮询任务状态接口GET /v1/tasks/req_123456直到状态变成succeeded用 WebSocket 订阅终端Hub 会主动把结果推给调用方。我一般测试时用轮询生产里用订阅。轮询代码很短while True: resp requests.get(f{hub}/v1/tasks/{request_id}, headersheaders) data resp.json() if data[status] in (succeeded, failed, timeout): print(data) break time.sleep(0.5)最终返回结果里有 Agent 回传的output字段{ status: succeeded, result: { status: ok, output: fake_ocr_result_for_documents/contract.pdf, latency_ms: 1020 }, executed_by: agent-main-ocr-3f9a }整个过程看下来调用方根本不需要知道 Agent 的 IP、端口甚至不需要知道它运行在哪个系统上。唯一打交道的就是 Hub。3.4 给现有 Agent 加个健康检查 API为了让 Hub 能主动感知业务存活状态我在每个 Agent 内置了一个/healthzHTTP 接口。SDK 启动后会自动监听 0.0.0.0 的一个随机端口并把这个端口上报给 Hub。Hub 在心跳报文之外还可以每隔 60 秒主动探测一次这个接口返回值里带{load: 0.8, queue_depth: 3}这样的负载数据。这个设计让我可以在注册中心里看到每个 Agent 的真实负载做能力路由时不只是随机挑而是挑负载最低的。代码如下from agent_reach_sdk import HealthServer health HealthServer(port0) # port0 表示随机端口 health.set_handler(lambda: { load: current_load(), queue_depth: queue.qsize() }) health.start()不过要提醒一点健康检查的 HTTP 端口不能和 Hub 的消息通道端口冲突。如果 Agent 运行在容器里记得把容器端口映射出来或者让 Hub 通过 Docker 网络直连。4. 实操中的坑与排查实录4.1 心跳偶发丢失导致 Agent 被误下线这个坑我在设计时已经提过解决思路但实操中仍然会反复踩到。现象是Agent 明明在正常运行Hub 却隔三岔五地把状态改成quasi-offline然后路由时跳过它导致部分任务堆积。排查方法查看 Hub 日志里有没有heartbeat missed报警用tcpdump抓包确认 Agent 是否真的把PING报文发出去了检查 Agent 与 Hub 之间是否有 NAT 超时比如家用路由器默认的 UDP 会话超时可能在 30 秒左右恰好和心跳间隔接近。我这里最终把心跳间隔从 30 秒调到 20 秒同时把“误判阈值”从 3 次加大到 5 次。代价是误判收敛时间变长但对稳定性要求高的场景宁可延迟判定也不冤枉好人。4.2 重复投递导致业务被执行两次WebSocket 连接断开重连后如果之前有一条消息已经发给 Agent但 Agent 还没来得及回执Hub 会认为投递失败于是重试一次。这样一来Agent 端就会把同一个request_id执行两遍。对于只读任务无所谓但对于“发邮件”“转账”这类操作就是事故。解决办法有两个层面Hub 端重试前检查 Agent 是否已恢复连接并且检查是否有未完成的回执Agent 端SDK 内置一个按request_id去重的 set缓存最近 10000 条已执行任务的 ID重复消息直接返回缓存结果不执行业务逻辑。去重代码就几行_seen set() def deduplicate(request_id): if request_id in _seen: return False _seen.add(request_id) return True我建议无论如何都要在业务入口做一次幂等处理再信任 Hub 的重试机制。分布式环境下“at least once”是常态设计业务时要默认消息可能重复。4.3 回调地址写错导致结果石沉大海Agent 注册时有一个callback_base_url字段SDK 在返回结果时会把结果 POST 到这个地址。我一度以为这里填的是 Agent 自己的地址结果填成了http://localhost:9001/result。Agent 跑在树莓派上没问题但 Hub 跑在另一台机器上它去请求localhost只会打到 Hub 自己结果自然找不到。正确理解是callback_base_url是 Hub 回调 Agent 时使用的地址因此必须填写 Hub 能访问到的 Agent 地址。如果 Hub 和 Agent 在同一个 Docker 网络中要填http://agent-container-name:9001不能填 localhost。如果 Agent 在另一个局域网需要配置反向代理或路由映射。我后来把回调逻辑改成了走 WebSocket 通道回传优先用通道不用 HTTP避免这整类地址配置问题。HTTP 回调只作为降级方案。4.4 常见问题速查表症状可能原因处理方式Hub 显示 Agent 在线但任务一直 pendingAgent 的 handler 处理太慢或阻塞查看 Agent 进程 CPU 与队列深度必要时调大timeout_ms任务 succeeded 但 result 为空Agent 的 handler 没有返回 dict检查 handler 是否return NoneSDK 会当空结果回传多个 Agent 同时在线但总是同一个执行负载评估逻辑过于简单在注册元数据里配置 priority 权重重连后收不到新任务会话票据过期检查 token 有效期SDK 应在重连时重新换取票据时区导致心跳时间计算错误Hub 与 Agent 时区不一致全部统一使用 UTC不读取本地时间SQLite 锁导致任务写入超时并发写入冲突关闭 WAL 模式或改成 Postgres 存储这张表是我在实际跑了两周后整理出来的每一个问题都真实遇到过。尤其是时区的那一个看似小事但真的会让心跳误判排查了很久。5. 经验补充与后续扩展5.1 从“触达”到“编排”让 Agent 互相调度Agent-Reach 本身只做触达但触达一旦稳定多 Agent 协作就变成简单的代码逻辑。比如我做一个“合同审阅”流程用户上传 PDF 到主服务主服务通过 Agent-Reach 调用ocrAgent提取文本OCR 结果文本通过 Agent-Reach 发给llm-reviewerAgent让大模型生成摘要摘要再广播给notify组里的所有 Agent邮件、钉钉、Slack。整个过程因为每条消息都带request_id所以串联起来非常清晰。我甚至可以查看任务拓扑图某个摘要到底依赖哪次 OCR在哪一步花费了多少时间。这比直接把几个 Agent 用 Python 函数链起来好维护得多至少你不会在某个 Agent 升级后惊慌失措。实现编排层时不需要改 Agent-Reach只需要一个很低成本的 worker订阅任务完成事件然后构造新任务再发起调用。这个逻辑可以放在 Hub 之外用任何语言写都行。5.2 运维视角监控与告警要趁早做Agent-Reach 暴露了/metrics端点基于 Prometheus 格式。我设计了五个关键指标reach_agent_online_count在线 Agent 数量reach_task_created_total创建任务总数reach_task_succeeded_total成功任务总数reach_task_failed_total失败任务总数reach_task_duration_seconds任务执行耗时直方图。用 Grafana 建了一个看板实时看有没有 Agent 掉线以及任务延迟分布。实践经验是一旦发现某个 Agent 平均耗时翻倍基本就是它的业务逻辑出了问题而不是网络。这时候去翻 Agent 的日志能精准定位。另外我设置了最简单的告警规则任何reach_task_failed_total在 5 分钟内增长超过 10 次就触发钉钉通知。这让我在故障发生时不会被动。5.3 后续扩展方向接入大模型和 Webhook下一步我还想把 Agent-Reach 接到 MCP 生态里让大模型直接通过工具调用方式触达任何 Agent。目前实现了一个很薄的 MCP 服务把大模型发出的工具请求转成Agent-Reach任务再把结果塞回给大模型。这样模型不再是只能“聊天”而是真的能指挥所有智能体干活。另外计划支持 Webhook 触发GitHub push 事件直接生成一个Agent-Reach广播让所有监听代码变更的 Agent 自动刷新缓存或跑测试。做成这件事以后Agent-Reach 就不只是一个人工调度的工具而是变成一套自动化事件分发底座。我自己实际跑下来最爽的一点是以前要在五台机器上来回 ssh 敲命令现在只需要在一个 Webhook 或一句 curl 里带上targetcustom就能精准触达。最后再分享一个小技巧如果你也给每台机器跑着多个 Agent记得给 Agent 命名时带上具体位置和用途比如agent-nas-ocr-01而不是agent-01否则排查日志的时候你能感受到什么叫“同名混乱”。这个项目的价值不在于代码多复杂而在于它让“Agent 之间互相找到并正确传话”这件事变得不值得再花精力去考虑。