Agent-Reach 不是又一个聊天机器人框架也不是拿来即用的 Prompt 调优工具。做完这个项目之后我最大的感受是Agent 能不能真正“办事”卡点往往不在推理而在触达。你在本地跑一个单 Agent 玩 ReAct 循环跑几百轮都挺顺但一旦让几个 Agent 协作、让它们去调外部系统、让它们去等另一个 Agent 的中间结果整个系统的复杂度就噌地上来了谁负责联系谁、消息语义怎么定、超时之后是重试还是放弃、上下文怎么裁剪才不爆 token。这个触达问题的本质是给 Agent 世界补上一套可靠、可观测、可编排的“通信与投递基础设施”。Agent-Reach 就是我围绕这个目标做的一个实打实的项目它解决的问题非常适合那些正在做多 Agent 协同、做自动化任务编排、做工具调用层的开发者参考下面我把完整的设计思路和踩坑过程都摊开讲。1. 项目定位与核心设计思路1.1 为什么单 Agent 很顺手一旦协作就全是问题先复盘一下我最初碰到的场景我先写了一个负责搜集信息的 Agent它的工作是把散落在不同目录的日志文件读进来做一轮摘要。单测的时候它跑得飞快效果也不错。可当我让它把摘要结果交给另一个负责生成日报的 Agent 时问题就来了第一个 Agent 不知道第二个 Agent 的输入格式第二个 Agent 偶尔会因为等待时间太长直接超时更关键的是两个 Agent 之间的对话完全靠 Prompt 互相猜第一个 Agent 说“我已经把摘要写进 /tmp/summary.txt”第二个 Agent 却因为文件权限问题根本读不到。这其实是目前很多 Agent 项目都在硬扛的通用痛点。市面上的框架大多把重心放在“单 Agent 的推理链”上对“多 Agent 之间的触达方式”没有统一抽象。Agent 之间所谓的协作经常退化成直接在系统 Prompt 里写“你可以调用 XX 工具”然后让模型自己玩概率。这种做法不是不行而是到了生产环境就变得没法收场没有重试、没有幂等、没有超时控制、没有任何跟踪日志。所以我在做 Agent-Reach 的时候第一原则就是把触达当作一种独立的系统能力来做而不是让每个 Agent 自己去实现一遍通信逻辑。1.2 Agent-Reach 到底解决哪四类问题触达层要覆盖的问题我梳理下来大概是四类。第一类是发现一个 Agent 怎么知道另一个 Agent 存在、支持什么语义、当前是否在线。第二类是意图表达调用方不能只发一个“帮我处理一下”这种模糊消息而是要带着结构化的、对方能理解的语义标签比如“SummarizeFileRequest”或者“AlarmNotify”。第三类是上下文交换Agent 之间传递的不仅仅是消息正文还有引用关系、关联任务 ID、可选的上下文切片避免把整个对话历史都搬过去。第四类是失败编排消息发出了对方没响应怎么办对方响应了但结果是错怎么办这种失败后的补偿机制如果不提前设计后面所有 Agent 都会被卡住。Agent-Reach 把这些问题收敛成了一个中间层它不关心 Agent 内部是用 ReAct、Plan-and-Execute 还是别的什么推理方式它只关心“谁能触达谁用什么样的语义触达触达失败之后状态怎么挂起”。听起来像是给 Agent 世界重新发明了一根总线但实际实现起来并没有那么玄乎。1.3 为什么不用现成的 RPC 框架这里有个很容易被问到的点现在遍地都是 gRPC、HTTP、消息队列为什么还要自己做一个触达层我的答案是Agent 触达和普通的服务间调用有本质区别。普通 RPC 的语义是“请求-响应”大多数时候调用方明确知道要调用哪个方法、传什么参数、期待什么返回。但 Agent 触达更像是一种目标驱动的松散投递。调用方往往只知道自己想要一个什么结果但不知道系统里哪个 Agent 能处理也不知道那个 Agent 当前负载如何。而且 Agent 的响应时间波动很大可能几秒也可能几十秒传统 RPC 的固定超时和不支持异步语义在这种场景下很容易误杀任务。另外Agent 之间的消息链路通常横跨多轮、多跳需要追踪整条链路上的状态这也是普通 RPC 框架不会替你考虑的事情。所以 Agent-Reach 做的不是替换 RPC而是把 RPC 降级成底层传输方式之一。Agent 之间的触达优先走我定义的一套“可达原语”这套原语自带超时、重试、幂等、追踪语义底层可以用 HTTP、WebSocket 甚至本地函数回调来承载。这样不同的部署环境下Agent 之间依然能用同一套触达语言对话不会因为传输层换掉就得重写一遍业务代码。2. 整体架构与协议设计2.1 Hub 与 Spoke 的星型模型Agent-Reach 的整体架构采用的是经典的“Hub 与 Spoke”星型模型中心是一个叫 Reach-Hub 的消息调度服务外围是各个 Agent 实例每个 Agent 通过内置的 Reach-Spoke 适配器接入 Hub。听上去有点像消息代理但两者有个关键差别消息代理只负责转发消息而 Reach-Hub 还会维护一份“能力路由表”。每个 Agent 启动的时候会向 Hub 注册自己支持的处理语义比如“我可以处理 MarkdownToHtml”Hub 就把这个语义写到能力路由表里。后续如果有别的 Agent 触达“MarkdownToHtml”Hub 会根据路由表把消息投给对应的 Agent。我采用星型结构而不是网状直连最大原因是可观测性。网状结构里两个 Agent 直接通信出了问题你很难知道消息是在哪个环节丢的审计日志也散得到处都是。星型结构则让所有触达请求都经过一个中心节点我可以在这里统一做流量记录、延迟统计、失败转储排查问题的时候效率高得多。当然中心化会带来单点风险所以我在设计里给 Hub 加了一层状态持久化所有待投递消息都会落盘这样 Hub 重启之后能续跑而不是把在途消息全丢了。2.2 能力路由表与心跳保活机制能力路由表是 Agent-Reach 里最重要的元数据。它的结构大概这样字段说明capability能力语义如 AlarmNotify、FileSummarizespoke_id提供该能力的 Agent 实例 IDendpoint投递地址如 http://agent-a:9001/callbackweight负载权重多实例时按权重分发last_seen最近一次心跳时间statusonline / offline / draining每个 Spoke 会每隔 15 秒向 Hub 发一次心跳心跳包里不光有“我还活着”还包含当前 Agent 的负载度比如消息队列积压数。Hub 收到心跳后就更新 last_seen 和 status。一旦超过 45 秒没收到心跳Hub 会把对应 Spoke 的状态置为 offline。这里有一个我反复调整过的细节路由表不能只靠心跳续约否则 Agent 忙碌的时候可能来不及发心跳就被误判为离线。所以我在 Agent-Reach 里做了一个折中设计Spoke 在处理长任务时会主动发送一个 heartbeat/status/update 消息告诉 Hub“我这颗消息还在处理中没有死”这样 Hub 就不会着急把路由切换到别的实例上。2.3 REACH-RPC 触达协议头触达协议我起名叫 REACH-RPC每条触达消息都带一个协议头协议头的结构长这样{ proto_version: 1, msg_id: 7f3a91c2e4d84b16a2f0f1e5c6a7b8d9, msg_type: sail, source_spoke: agent-file-summary-01, target_capability: DailyReportRender, reply_to: agent-file-summary-01.callback, dedup_key: daily-report-20241215, timeout_ms: 30000, trace_id: trc_a1b2c3d4 }这个协议头解决的核心问题是语义寻址。传统 RPC 的目标地址是主机加端口REACH-RPC 的目标地址是“能力语义”source_spoke 和 target_capability 一起决定了这条消息会被谁接收。reply_to 字段则说明调用方希望从哪里接收响应它和普通回调地址不太一样它指向的不是一个端口而是另一个触达通道。协议头里的 timeout_ms 不是建议值而是强制执行值。Hub 在投递消息时会带上这个超时时间如果 Spoke 侧超过了 timeout_ms 还没有返回确认Hub 会主动标记这条消息为“疑似超时”并触发后续的重试或者失败通知。这一条在设计之初有很多争论有人觉得超时应由调用方来决定实际上在异步 Agent 场景里调用方根本没办法实时盯着所以把这个控制权放在协议头里、由 Hub 来监督是更可靠的选择。2.4 安全性设计只投递已登记的能力因为我设计这套系统的初衷是解决内部多个 Agent 协作的问题所以安全模型并不复杂但有一条原则必须守住Hub 只向已登记的能力路由投递消息不接受任何未经注册的 target_capability。如果一个调用方发来一条消息目标能力不在路由表里Hub 会直接返回一个 ReachNoRoute 的错误不会尝试猜测或者广播。这个设计避免了一个很实际的问题Agent 协作时经常会因为拼写错误或者语义不一致导致消息发到了一个根本没人监听的能力上。如果 Hub 强行广播则所有 Agent 都会收到一条莫名其妙的请求造成大量无效计算。在权限隔离方面我还给每条能力注册加了一个可见域字段比如“内部专用”“仅限数据团队”只有调用方所属域和目标能力所属域匹配时投递才会被允许。这听起来像是在做复杂权限系统但实现起来只是路由表里的一个字段比对成本很低收益却是实打实的。3. 核心实现细节与触达原语3.1 五种触达原语sail、anchor、dispatch、retreat、settleAgent-Reach 的操作模型不是简单地发消息而是提炼了五种触达原语。这套原语是整个项目的灵魂单独拎出来讲讲。sail是一次性投递语义和“发送消息”最接近。调用方指定 target_capability 和 payloadHub 负责把消息投给当前在线的一个或者多个实例。sail 不保证对方一定处理成功但保证消息被可靠投递到了至少一个匹配目标上如果所有目标都离线sail 会返回 ReachOffline。anchor是持久登记语义。一个 Spoke 如果想长期接收某个能力的数据可以 register 一个 anchor把自己钉在能力路由表里。sail 是一次性的anchor 是持续的这个区分非常重要有些能力是“调用一次就结束”有些能力是“我要常驻监听”两者在生命周期管理上完全不同。dispatch是扇出语义用于把一个触达请求同时发给多个 Agent然后等所有 Agent 的返回都齐了才继续往下走。我在做并行汇总类任务的时候经常用到比如让三个 Agent 同时分析三份文档最后汇总结果dispatch 就能一次性完成。retreat是退避语义。当触达目标失败后调用方可以选择 retreat让系统按照指数退避策略重新投递比如第一次等 2 秒第二次 4 秒第三次 8 秒最多重试 5 次后转为失败。settle是结算语义用于确认一个触达链路的最终状态。当一条多跳链路全部跑完所有子任务都返回了调用方可以发一个 settle 信号把整条链路的状态标记为 completedHub 才会把对应的追踪记录归档。3.2 任务状态机从发起到达成触达链路是有生命周期的我在 Agent-Reach 里给它定义了一个五态状态机。所有触达任务都在这五个状态之间流转任何一个触达请求你都能准确说出它当前处于哪个阶段。状态含义迁移条件IDLE任务已创建但未投递发送触达请求PROVISIONED已确认路由目标等待目标确认Spoke 回复 ACKREACHED目标已接收到消息Spoke 回复 REACHEDCONFIRMED目标完成了任务并返回结果Spoke 回复 RESULTSUNSET链路结束状态归档超时、失败或 settle这里有个细节想特别说明REACHED 和 CONFIRMED 是分开的。REACHED 表示“消息送到了对方手里”CONFIRMED 表示“对方真的做完了事”。这两个状态如果混在一起你就会经常遇到一种诡异情况消息显示发送成功但任务结果一直没出来。在 Agent-Reach 里一旦某个任务长时间停留在 REACHED 而无法进入 CONFIRMED就说明对方 Agent 接收了任务但执行卡住了这时候就该触发 retreat 或者人工介入排查。3.3 幂等控制同一颗消息不能重复生效Agent 触达和普通消息队列有个巨大差异那就是消息可能重放。网络超时、Hub 重启、Spoke 崩溃任何一个环节都可能让同一条消息被投递两次。如果这条触达请求执行的是非幂等操作比如创建订单或者发送邮件重复执行就会造成事故。我在协议头里设计了 dedup_key 字段这是幂等控制的基础。Spoke 侧在收到一条消息时会先在自己的去重表里检查 dedup_key 是否存在如果存在说明这条消息之前已经处理过了直接返回上次的结果缓存而不会重新执行一遍业务逻辑。去重表我采用的是本地文件加内存双写模式。内存表用于快速判断文件用于崩溃恢复。你可能会问为什么不直接丢给 Redis 做去重我的理由是不想引入额外的外部依赖Agent 场景里每个 Spoke 的状态应该尽量自包含否则这 Agent 一出问题又得排查 Redis复杂度又上去了。3.4 上下文裁剪把最有用的信息贴上多 Agent 协作还有一个常见的隐性坑上下文爆炸。Agent A 生成了一个任务把消息传给 Agent B 之后为了保险它把完整的对话历史也一起附上了。Agent B 处理完传给 Agent C又是把前面所有历史原封不动转发。几轮下来一条触达消息的上下文体积能膨胀到好几 MB模型输入直接爆掉。在 Agent-Reach 里我给每条触达消息加了一个 context 段但不是传递完整历史而是传递一个经过裁剪的上下文切片里面包含任务目标、关联 ID、相关数据引用、前序结论摘要。原始数据只保留引用地址比如文件路径、对象存储 key、数据库记录 ID需要详细数据时再通过引用去取。这样既保证了 Agent 有足够的背景信息做判断又不会让消息体积失控。这个裁剪动作我让调用方自己来控制框架只提供封装好的切片数据结构。实际操作中裁剪策略可以简单到“只保留最近 20 条消息的关键字段”也可以复杂到按 token 预算动态裁剪。我自己的经验是刚开始别做太复杂先保证每条消息体积不超过 8KB等有了真实数据量再迭代策略。4. 实操过程从零搭建一个可运行的触达示例4.1 核心依赖与最小运行环境Agent-Reach 实现上选型比较轻核心代码用 Python 写的依赖库只有 FastAPI、httpx 和一个轻量级的 SQLite 做持久化。运行环境只需要 Python 3.10 以上没有数据库服务也没有消息队列装完依赖就能跑起来。我推荐这种极简依赖的原因是想让触达层本身保持“低摩擦接入”。如果一个 Agent 想接入协作网络还得先部署一套 Kafka 或者 MongoDB那这个基础设施的门槛就太高了。用 SQLite 加文件系统做持久化虽然并发能力上限不如专业中间件但对于中小规模的 Agent 编排场景完全够用而且排查问题特别方便数据文件一拍就能拖下来分析。4.2 核心代码骨架起一个 Spoke下面这段代码是 Spoke 侧最精简的接入方式。我们假设要接入一个“文件摘要”能力的 Agent它负责读取日志文件生成摘要之后返回给调用方。import asyncio from agent_reach import Spoke, Capability, ReachMessage capability Capability( nameFileSummarize, version1.0, domaindata-team, ) async def handle_summarize(message: ReachMessage): file_path message.payload[file_path] lines await read_file_lines(file_path) summary extract_summary_from_lines(lines) return { status: ok, summary: summary, } async def main(): spoke Spoke( spoke_idagent-file-summary-01, capabilities[capability], handler_map{ FileSummarize: handle_summarize, }, hub_urlhttp://127.0.0.1:8765, ) await spoke.start() await asyncio.Event().wait() if __name__ __main__: asyncio.run(main())这段代码看起来很简单但实际上已经把 Spoke 的完整生命周期管理都包进去了启动时注册能力、定时发心跳、接收消息并调用 handler、返回结果。我特意把 handler_map 设计成显式映射因为 Prompts 驱动的动态函数路由看着高级实际用起来很容易让操作不可控调试的时候你根本不知道一个请求会命中哪个函数。4.3 请求端触达sail 与 dispatch 的调用方式对应的调用方代码更简单只需要创建一个 Client然后向 Hub 发起 sail 触达from agent_reach import ReachClient client ReachClient(hub_urlhttp://127.0.0.1:8765) # 发起一次性触达 result await client.sail( target_capabilityFileSummarize, payload{file_path: /var/log/app/2024-12-15.log}, timeout_ms15000, dedup_keysummary-20241215-01, ) print(result.payload)如果我有三个不同的 Agent 都在处理三种不同来源的数据我想一起做摘要再合并结果就可以用 dispatchresults await client.dispatch( requests[ { target_capability: FileSummarize, payload: {file_path: /var/log/app/nginx.log}, }, { target_capability: FileSummarize, payload: {file_path: /var/log/app/api.log}, }, { target_capability: FileSummarize, payload: {file_path: /var/log/app/worker.log}, }, ], timeout_ms30000, )dispatch 的语义是所有子触达全部完成才返回调用方不需要自己管理并发。实现上Hub 侧会为每条子触达单独做路由、超时、重试最后把三个结果按请求顺序聚合成一个列表返回。顺序一致性在这里很重要如果子任务返回结果顺序和请求顺序不一致调用方处理数据时会平白无故多出很多边界判断。4.4 一份实测运行日志触达链路到底发生了什么下面这段日志是我在一个真实测试环境里截取出来的展示了从调用方发起触达到最终 CONFIRMED 的过程[hub] accepted sail request msg_id7f3a91c2 target_capabilityFileSummarize [hub] capability lookup: FileSummarize - agent-file-summary-01 (online) [hub] routing message to spoke agent-file-summary-01 [spoke:agent-file-summary-01] received REACH_RPC request msg_id7f3a91c2 [spoke:agent-file-summary-01] dedup check passed: summary-20241215-01 [spoke:agent-file-summary-01] handler started file/var/log/app/2024-12-15.log [spoke:agent-file-summary-01] handler finished in 3.2s [spoke:agent-file-summary-01] sending RESULT msg_id7f3a91c2 [hub] received RESULT msg_id7f3a91c2, transition to CONFIRMED [hub] deliver response to caller这组日志完整展示了整个触达链路。注意 [hub] capability lookup 这一步它验证了调用方发送的 target_capability 确实存在且在线。如果真的出现了找不到路由的情况你这儿的日志会变成 [hub] no route for capability排查问题的时候第一步就是搜这种入口日志。4.5 实操中最容易忽略的三个参数第一是 timeout_ms。我见过很多人直接填一个大数字比如 60000觉得反正超时设大点更保险。但 Agent 触达讲究的是快速失败、快速反馈如果调用方等了一分钟才知道对方挂了那这个等待的体验就非常差了。建议根据任务的复杂度来估算普通文件摘要给 15 秒涉及外部 API 调用的任务给 30 秒确实需要长时间计算的才给 60 秒但这样的任务更建议改成异步回调模式。第二是 dedup_key。这个 key 的粒度要尽量稳定比如按业务日期加任务序号来生成而不是按当前时间戳否则重试时生成的新 key 和原 key 对不上幂等就失效了。第三是 Spoke 的负载上报。心跳里带负载度这个字段可能很多人忽略了但它在多实例场景下非常有用。如果不带负载度Hub 可能会把大量消息投给一个本来就很忙的实例造成消息积压而旁边空闲的实例却在闲着。负载度字段只需要一个简单的队列积压数指标就行不用做太复杂。5. 常见问题与排查技巧实录5.1 问题速查表实操中踩过的坑集中整理成一张速查表遇到症状可以直接对照着查症状可能原因排查命令与解决方向触达目标一直返回 ReachOfflineSpoke 未注册成功或心跳丢失检查 Spoke 日志中的 register 结果数一下心跳间隔是否超过 45 秒消息已发送但对方一直不处理Spoke 侧消息队列积压检查 Spoke 的负载度字段查看 handler 是否发生异常卡死重复触达导致业务数据重复dedup_key 生成不稳定检查请求是否用了相同 dedup_key确认 Spoke 侧去重表是否落盘持久化链路状态停在 REACHED 不动目标收到消息但 handler 内部崩溃查看 Spoke 的异常捕获日志确认 handler 是否有 try-finally 保证 RESULT 返回调用方响应时间过长触达链路上有多跳串行检查每跳的超时设置考虑将串行跳改为 dispatch 并行Hub 重启后消息丢失未开启状态持久化确认 Hub 的 SQLite 持久化路径配置正确检查在途消息是否落盘5.2 埋点追踪一个 trace_id 贯穿到底排查 Agent 触达问题最怕的是不知道一条消息到底走到了哪一步。Agent-Reach 里每个触达请求都会生成一个 trace_id这个 id 会跟随协议头传到每一个相关节点。我在实际排查问题时最常用的手法就是拿到一个 trace_id然后去 Hub 的日志文件里 grep 一次把这条链路的所有事件全部拉出来按时间排序一眼就能看到问题出在哪一环。说实话这套机制在开发初期帮了大忙。当时好几次线上现象看起来像是“某个 Agent 收到了消息但没干活”结果拉出日志一看消息根本没到那个 Agent而是在路由表里就因为没有匹配能力被打回了。追踪日志记录的字段包括事件类型、节点 ID、消息 ID、前后状态、耗时、错误信息。这些字段尽量标准化方便后续做统计分析。我强烈建议任何做多 Agent 系统的人参考这个方案让追踪能力从第一天就内建不要等出了问题再补否则排查成本会让你崩溃。5.3 心跳抖动与离线误判心跳机制是我在测试阶段被整得最惨的一个环节。刚开始我把心跳间隔设为 10 秒离线判定阈值设为 20 秒想着这样检测快一点。结果跑起来后发现一旦某台机器发生一次 GC 停顿或者网络出现一次 3-5 秒的抖动Spoke 就可能错过心跳窗口被 Hub 误判为 offline已经分配给它的消息会被移到其他实例造成重复处理。后来我调整策略把心跳间隔和离线判定阈值分开并且引入了租约时间的概念。现在 Spoke 在启动时向 Hub 申请一个租约默认租期 120 秒每次心跳续租 60 秒。只要 Spoke 在租期内有任意一次活跃就不会被判定为离线这个缓冲空间大大减少了对瞬时抖动的敏感度。5.4 上下文缓冲泄漏裁剪策略的后遗症还有一个问题在长时间运行时才暴露上下文裁剪虽然每次控制了单条消息的体积但如果某条链路上累积的消息数特别多每个节点都会缓存大量历史触达记录内存占用还是会上来。我在代码里给每个 Spoke 加了历史消息上限默认保留最近 200 条完成记录超出后自动淘汰最旧的记录。如果业务需要更长周期追踪可以设置 archive 目录把旧记录落盘到 SQLite 归档表而不是一直留在内存。这里有个小建议归档的粒度按 trace_id 关联归档而不是按单个消息归档。因为一次业务动作往往涉及多个消息按 trace_id 归档后事后复盘时可以把整条链路都捞出来看可读性比单条消息归档好得多。5.5 补偿任务的正确打开方式当一条触达消息经历了全部重试仍然失败时Agent-Reach 不会把它静默丢弃而是会生成一条 FailureNotice发送给调用方指定的补偿通道。我在项目里把这个补偿通道设计成可以指向另一个 Agent 的能力这样就能实现自动化的失败处理编排。举个例子如果我的“生成日报”Agent 因为数据源异常触达失败我希望系统不要直接放弃而是触达一个“数据修复”Agent让它去检查数据源。这个补偿链路的配置不需要改业务代码只需要把 FailureNotice 时执行的动作配置成这样先触达数据修复能力修复完成后再次触达生成日报能力最多两轮。这种思路比硬编码一个 if-else 可靠得多因为补偿策略本身也能被编排、被观察。我个人的体会是Agent 协作系统里“失败”不是需要消灭的异常而是一种可以预见的常态。设计触达层的时候如果把大部分精力花在“如何避免失败”上效果往往不好反过来把精力花在“失败之后如何体面地恢复”上整个系统的稳定性反而会高得多。最后再多说一句Agent-Reach 这个名字里有“Reach”我对它的理解是Agent 真正能触达什么决定了它能干成什么。推理模型再聪明如果触达层像泥潭一样那整个系统的上限也被锁死了。所以别急于做复杂的 Agent 编排先把触达层做稳把每条消息的去向和状态都搞清楚你会发现很多之前觉得玄乎的 Agent 协作问题本质上就是通信问题。这个方向后续我还会继续扩展比如把触达层的遥测数据接到可视化面板上让整条链路的状态实时可见最近已经在动手了等有结果再回来细聊。
Agent-Reach 不是又一个聊天机器人框架也不是拿来即用的 Prompt 调优工具。做完这个项目之后我最大的感受是Agent 能不能真正“办事”卡点往往不在推理而在触达。你在本地跑一个单 Agent 玩 ReAct 循环跑几百轮都挺顺但一旦让几个 Agent 协作、让它们去调外部系统、让它们去等另一个 Agent 的中间结果整个系统的复杂度就噌地上来了谁负责联系谁、消息语义怎么定、超时之后是重试还是放弃、上下文怎么裁剪才不爆 token。这个触达问题的本质是给 Agent 世界补上一套可靠、可观测、可编排的“通信与投递基础设施”。Agent-Reach 就是我围绕这个目标做的一个实打实的项目它解决的问题非常适合那些正在做多 Agent 协同、做自动化任务编排、做工具调用层的开发者参考下面我把完整的设计思路和踩坑过程都摊开讲。1. 项目定位与核心设计思路1.1 为什么单 Agent 很顺手一旦协作就全是问题先复盘一下我最初碰到的场景我先写了一个负责搜集信息的 Agent它的工作是把散落在不同目录的日志文件读进来做一轮摘要。单测的时候它跑得飞快效果也不错。可当我让它把摘要结果交给另一个负责生成日报的 Agent 时问题就来了第一个 Agent 不知道第二个 Agent 的输入格式第二个 Agent 偶尔会因为等待时间太长直接超时更关键的是两个 Agent 之间的对话完全靠 Prompt 互相猜第一个 Agent 说“我已经把摘要写进 /tmp/summary.txt”第二个 Agent 却因为文件权限问题根本读不到。这其实是目前很多 Agent 项目都在硬扛的通用痛点。市面上的框架大多把重心放在“单 Agent 的推理链”上对“多 Agent 之间的触达方式”没有统一抽象。Agent 之间所谓的协作经常退化成直接在系统 Prompt 里写“你可以调用 XX 工具”然后让模型自己玩概率。这种做法不是不行而是到了生产环境就变得没法收场没有重试、没有幂等、没有超时控制、没有任何跟踪日志。所以我在做 Agent-Reach 的时候第一原则就是把触达当作一种独立的系统能力来做而不是让每个 Agent 自己去实现一遍通信逻辑。1.2 Agent-Reach 到底解决哪四类问题触达层要覆盖的问题我梳理下来大概是四类。第一类是发现一个 Agent 怎么知道另一个 Agent 存在、支持什么语义、当前是否在线。第二类是意图表达调用方不能只发一个“帮我处理一下”这种模糊消息而是要带着结构化的、对方能理解的语义标签比如“SummarizeFileRequest”或者“AlarmNotify”。第三类是上下文交换Agent 之间传递的不仅仅是消息正文还有引用关系、关联任务 ID、可选的上下文切片避免把整个对话历史都搬过去。第四类是失败编排消息发出了对方没响应怎么办对方响应了但结果是错怎么办这种失败后的补偿机制如果不提前设计后面所有 Agent 都会被卡住。Agent-Reach 把这些问题收敛成了一个中间层它不关心 Agent 内部是用 ReAct、Plan-and-Execute 还是别的什么推理方式它只关心“谁能触达谁用什么样的语义触达触达失败之后状态怎么挂起”。听起来像是给 Agent 世界重新发明了一根总线但实际实现起来并没有那么玄乎。1.3 为什么不用现成的 RPC 框架这里有个很容易被问到的点现在遍地都是 gRPC、HTTP、消息队列为什么还要自己做一个触达层我的答案是Agent 触达和普通的服务间调用有本质区别。普通 RPC 的语义是“请求-响应”大多数时候调用方明确知道要调用哪个方法、传什么参数、期待什么返回。但 Agent 触达更像是一种目标驱动的松散投递。调用方往往只知道自己想要一个什么结果但不知道系统里哪个 Agent 能处理也不知道那个 Agent 当前负载如何。而且 Agent 的响应时间波动很大可能几秒也可能几十秒传统 RPC 的固定超时和不支持异步语义在这种场景下很容易误杀任务。另外Agent 之间的消息链路通常横跨多轮、多跳需要追踪整条链路上的状态这也是普通 RPC 框架不会替你考虑的事情。所以 Agent-Reach 做的不是替换 RPC而是把 RPC 降级成底层传输方式之一。Agent 之间的触达优先走我定义的一套“可达原语”这套原语自带超时、重试、幂等、追踪语义底层可以用 HTTP、WebSocket 甚至本地函数回调来承载。这样不同的部署环境下Agent 之间依然能用同一套触达语言对话不会因为传输层换掉就得重写一遍业务代码。2. 整体架构与协议设计2.1 Hub 与 Spoke 的星型模型Agent-Reach 的整体架构采用的是经典的“Hub 与 Spoke”星型模型中心是一个叫 Reach-Hub 的消息调度服务外围是各个 Agent 实例每个 Agent 通过内置的 Reach-Spoke 适配器接入 Hub。听上去有点像消息代理但两者有个关键差别消息代理只负责转发消息而 Reach-Hub 还会维护一份“能力路由表”。每个 Agent 启动的时候会向 Hub 注册自己支持的处理语义比如“我可以处理 MarkdownToHtml”Hub 就把这个语义写到能力路由表里。后续如果有别的 Agent 触达“MarkdownToHtml”Hub 会根据路由表把消息投给对应的 Agent。我采用星型结构而不是网状直连最大原因是可观测性。网状结构里两个 Agent 直接通信出了问题你很难知道消息是在哪个环节丢的审计日志也散得到处都是。星型结构则让所有触达请求都经过一个中心节点我可以在这里统一做流量记录、延迟统计、失败转储排查问题的时候效率高得多。当然中心化会带来单点风险所以我在设计里给 Hub 加了一层状态持久化所有待投递消息都会落盘这样 Hub 重启之后能续跑而不是把在途消息全丢了。2.2 能力路由表与心跳保活机制能力路由表是 Agent-Reach 里最重要的元数据。它的结构大概这样字段说明capability能力语义如 AlarmNotify、FileSummarizespoke_id提供该能力的 Agent 实例 IDendpoint投递地址如 http://agent-a:9001/callbackweight负载权重多实例时按权重分发last_seen最近一次心跳时间statusonline / offline / draining每个 Spoke 会每隔 15 秒向 Hub 发一次心跳心跳包里不光有“我还活着”还包含当前 Agent 的负载度比如消息队列积压数。Hub 收到心跳后就更新 last_seen 和 status。一旦超过 45 秒没收到心跳Hub 会把对应 Spoke 的状态置为 offline。这里有一个我反复调整过的细节路由表不能只靠心跳续约否则 Agent 忙碌的时候可能来不及发心跳就被误判为离线。所以我在 Agent-Reach 里做了一个折中设计Spoke 在处理长任务时会主动发送一个 heartbeat/status/update 消息告诉 Hub“我这颗消息还在处理中没有死”这样 Hub 就不会着急把路由切换到别的实例上。2.3 REACH-RPC 触达协议头触达协议我起名叫 REACH-RPC每条触达消息都带一个协议头协议头的结构长这样{ proto_version: 1, msg_id: 7f3a91c2e4d84b16a2f0f1e5c6a7b8d9, msg_type: sail, source_spoke: agent-file-summary-01, target_capability: DailyReportRender, reply_to: agent-file-summary-01.callback, dedup_key: daily-report-20241215, timeout_ms: 30000, trace_id: trc_a1b2c3d4 }这个协议头解决的核心问题是语义寻址。传统 RPC 的目标地址是主机加端口REACH-RPC 的目标地址是“能力语义”source_spoke 和 target_capability 一起决定了这条消息会被谁接收。reply_to 字段则说明调用方希望从哪里接收响应它和普通回调地址不太一样它指向的不是一个端口而是另一个触达通道。协议头里的 timeout_ms 不是建议值而是强制执行值。Hub 在投递消息时会带上这个超时时间如果 Spoke 侧超过了 timeout_ms 还没有返回确认Hub 会主动标记这条消息为“疑似超时”并触发后续的重试或者失败通知。这一条在设计之初有很多争论有人觉得超时应由调用方来决定实际上在异步 Agent 场景里调用方根本没办法实时盯着所以把这个控制权放在协议头里、由 Hub 来监督是更可靠的选择。2.4 安全性设计只投递已登记的能力因为我设计这套系统的初衷是解决内部多个 Agent 协作的问题所以安全模型并不复杂但有一条原则必须守住Hub 只向已登记的能力路由投递消息不接受任何未经注册的 target_capability。如果一个调用方发来一条消息目标能力不在路由表里Hub 会直接返回一个 ReachNoRoute 的错误不会尝试猜测或者广播。这个设计避免了一个很实际的问题Agent 协作时经常会因为拼写错误或者语义不一致导致消息发到了一个根本没人监听的能力上。如果 Hub 强行广播则所有 Agent 都会收到一条莫名其妙的请求造成大量无效计算。在权限隔离方面我还给每条能力注册加了一个可见域字段比如“内部专用”“仅限数据团队”只有调用方所属域和目标能力所属域匹配时投递才会被允许。这听起来像是在做复杂权限系统但实现起来只是路由表里的一个字段比对成本很低收益却是实打实的。3. 核心实现细节与触达原语3.1 五种触达原语sail、anchor、dispatch、retreat、settleAgent-Reach 的操作模型不是简单地发消息而是提炼了五种触达原语。这套原语是整个项目的灵魂单独拎出来讲讲。sail是一次性投递语义和“发送消息”最接近。调用方指定 target_capability 和 payloadHub 负责把消息投给当前在线的一个或者多个实例。sail 不保证对方一定处理成功但保证消息被可靠投递到了至少一个匹配目标上如果所有目标都离线sail 会返回 ReachOffline。anchor是持久登记语义。一个 Spoke 如果想长期接收某个能力的数据可以 register 一个 anchor把自己钉在能力路由表里。sail 是一次性的anchor 是持续的这个区分非常重要有些能力是“调用一次就结束”有些能力是“我要常驻监听”两者在生命周期管理上完全不同。dispatch是扇出语义用于把一个触达请求同时发给多个 Agent然后等所有 Agent 的返回都齐了才继续往下走。我在做并行汇总类任务的时候经常用到比如让三个 Agent 同时分析三份文档最后汇总结果dispatch 就能一次性完成。retreat是退避语义。当触达目标失败后调用方可以选择 retreat让系统按照指数退避策略重新投递比如第一次等 2 秒第二次 4 秒第三次 8 秒最多重试 5 次后转为失败。settle是结算语义用于确认一个触达链路的最终状态。当一条多跳链路全部跑完所有子任务都返回了调用方可以发一个 settle 信号把整条链路的状态标记为 completedHub 才会把对应的追踪记录归档。3.2 任务状态机从发起到达成触达链路是有生命周期的我在 Agent-Reach 里给它定义了一个五态状态机。所有触达任务都在这五个状态之间流转任何一个触达请求你都能准确说出它当前处于哪个阶段。状态含义迁移条件IDLE任务已创建但未投递发送触达请求PROVISIONED已确认路由目标等待目标确认Spoke 回复 ACKREACHED目标已接收到消息Spoke 回复 REACHEDCONFIRMED目标完成了任务并返回结果Spoke 回复 RESULTSUNSET链路结束状态归档超时、失败或 settle这里有个细节想特别说明REACHED 和 CONFIRMED 是分开的。REACHED 表示“消息送到了对方手里”CONFIRMED 表示“对方真的做完了事”。这两个状态如果混在一起你就会经常遇到一种诡异情况消息显示发送成功但任务结果一直没出来。在 Agent-Reach 里一旦某个任务长时间停留在 REACHED 而无法进入 CONFIRMED就说明对方 Agent 接收了任务但执行卡住了这时候就该触发 retreat 或者人工介入排查。3.3 幂等控制同一颗消息不能重复生效Agent 触达和普通消息队列有个巨大差异那就是消息可能重放。网络超时、Hub 重启、Spoke 崩溃任何一个环节都可能让同一条消息被投递两次。如果这条触达请求执行的是非幂等操作比如创建订单或者发送邮件重复执行就会造成事故。我在协议头里设计了 dedup_key 字段这是幂等控制的基础。Spoke 侧在收到一条消息时会先在自己的去重表里检查 dedup_key 是否存在如果存在说明这条消息之前已经处理过了直接返回上次的结果缓存而不会重新执行一遍业务逻辑。去重表我采用的是本地文件加内存双写模式。内存表用于快速判断文件用于崩溃恢复。你可能会问为什么不直接丢给 Redis 做去重我的理由是不想引入额外的外部依赖Agent 场景里每个 Spoke 的状态应该尽量自包含否则这 Agent 一出问题又得排查 Redis复杂度又上去了。3.4 上下文裁剪把最有用的信息贴上多 Agent 协作还有一个常见的隐性坑上下文爆炸。Agent A 生成了一个任务把消息传给 Agent B 之后为了保险它把完整的对话历史也一起附上了。Agent B 处理完传给 Agent C又是把前面所有历史原封不动转发。几轮下来一条触达消息的上下文体积能膨胀到好几 MB模型输入直接爆掉。在 Agent-Reach 里我给每条触达消息加了一个 context 段但不是传递完整历史而是传递一个经过裁剪的上下文切片里面包含任务目标、关联 ID、相关数据引用、前序结论摘要。原始数据只保留引用地址比如文件路径、对象存储 key、数据库记录 ID需要详细数据时再通过引用去取。这样既保证了 Agent 有足够的背景信息做判断又不会让消息体积失控。这个裁剪动作我让调用方自己来控制框架只提供封装好的切片数据结构。实际操作中裁剪策略可以简单到“只保留最近 20 条消息的关键字段”也可以复杂到按 token 预算动态裁剪。我自己的经验是刚开始别做太复杂先保证每条消息体积不超过 8KB等有了真实数据量再迭代策略。4. 实操过程从零搭建一个可运行的触达示例4.1 核心依赖与最小运行环境Agent-Reach 实现上选型比较轻核心代码用 Python 写的依赖库只有 FastAPI、httpx 和一个轻量级的 SQLite 做持久化。运行环境只需要 Python 3.10 以上没有数据库服务也没有消息队列装完依赖就能跑起来。我推荐这种极简依赖的原因是想让触达层本身保持“低摩擦接入”。如果一个 Agent 想接入协作网络还得先部署一套 Kafka 或者 MongoDB那这个基础设施的门槛就太高了。用 SQLite 加文件系统做持久化虽然并发能力上限不如专业中间件但对于中小规模的 Agent 编排场景完全够用而且排查问题特别方便数据文件一拍就能拖下来分析。4.2 核心代码骨架起一个 Spoke下面这段代码是 Spoke 侧最精简的接入方式。我们假设要接入一个“文件摘要”能力的 Agent它负责读取日志文件生成摘要之后返回给调用方。import asyncio from agent_reach import Spoke, Capability, ReachMessage capability Capability( nameFileSummarize, version1.0, domaindata-team, ) async def handle_summarize(message: ReachMessage): file_path message.payload[file_path] lines await read_file_lines(file_path) summary extract_summary_from_lines(lines) return { status: ok, summary: summary, } async def main(): spoke Spoke( spoke_idagent-file-summary-01, capabilities[capability], handler_map{ FileSummarize: handle_summarize, }, hub_urlhttp://127.0.0.1:8765, ) await spoke.start() await asyncio.Event().wait() if __name__ __main__: asyncio.run(main())这段代码看起来很简单但实际上已经把 Spoke 的完整生命周期管理都包进去了启动时注册能力、定时发心跳、接收消息并调用 handler、返回结果。我特意把 handler_map 设计成显式映射因为 Prompts 驱动的动态函数路由看着高级实际用起来很容易让操作不可控调试的时候你根本不知道一个请求会命中哪个函数。4.3 请求端触达sail 与 dispatch 的调用方式对应的调用方代码更简单只需要创建一个 Client然后向 Hub 发起 sail 触达from agent_reach import ReachClient client ReachClient(hub_urlhttp://127.0.0.1:8765) # 发起一次性触达 result await client.sail( target_capabilityFileSummarize, payload{file_path: /var/log/app/2024-12-15.log}, timeout_ms15000, dedup_keysummary-20241215-01, ) print(result.payload)如果我有三个不同的 Agent 都在处理三种不同来源的数据我想一起做摘要再合并结果就可以用 dispatchresults await client.dispatch( requests[ { target_capability: FileSummarize, payload: {file_path: /var/log/app/nginx.log}, }, { target_capability: FileSummarize, payload: {file_path: /var/log/app/api.log}, }, { target_capability: FileSummarize, payload: {file_path: /var/log/app/worker.log}, }, ], timeout_ms30000, )dispatch 的语义是所有子触达全部完成才返回调用方不需要自己管理并发。实现上Hub 侧会为每条子触达单独做路由、超时、重试最后把三个结果按请求顺序聚合成一个列表返回。顺序一致性在这里很重要如果子任务返回结果顺序和请求顺序不一致调用方处理数据时会平白无故多出很多边界判断。4.4 一份实测运行日志触达链路到底发生了什么下面这段日志是我在一个真实测试环境里截取出来的展示了从调用方发起触达到最终 CONFIRMED 的过程[hub] accepted sail request msg_id7f3a91c2 target_capabilityFileSummarize [hub] capability lookup: FileSummarize - agent-file-summary-01 (online) [hub] routing message to spoke agent-file-summary-01 [spoke:agent-file-summary-01] received REACH_RPC request msg_id7f3a91c2 [spoke:agent-file-summary-01] dedup check passed: summary-20241215-01 [spoke:agent-file-summary-01] handler started file/var/log/app/2024-12-15.log [spoke:agent-file-summary-01] handler finished in 3.2s [spoke:agent-file-summary-01] sending RESULT msg_id7f3a91c2 [hub] received RESULT msg_id7f3a91c2, transition to CONFIRMED [hub] deliver response to caller这组日志完整展示了整个触达链路。注意 [hub] capability lookup 这一步它验证了调用方发送的 target_capability 确实存在且在线。如果真的出现了找不到路由的情况你这儿的日志会变成 [hub] no route for capability排查问题的时候第一步就是搜这种入口日志。4.5 实操中最容易忽略的三个参数第一是 timeout_ms。我见过很多人直接填一个大数字比如 60000觉得反正超时设大点更保险。但 Agent 触达讲究的是快速失败、快速反馈如果调用方等了一分钟才知道对方挂了那这个等待的体验就非常差了。建议根据任务的复杂度来估算普通文件摘要给 15 秒涉及外部 API 调用的任务给 30 秒确实需要长时间计算的才给 60 秒但这样的任务更建议改成异步回调模式。第二是 dedup_key。这个 key 的粒度要尽量稳定比如按业务日期加任务序号来生成而不是按当前时间戳否则重试时生成的新 key 和原 key 对不上幂等就失效了。第三是 Spoke 的负载上报。心跳里带负载度这个字段可能很多人忽略了但它在多实例场景下非常有用。如果不带负载度Hub 可能会把大量消息投给一个本来就很忙的实例造成消息积压而旁边空闲的实例却在闲着。负载度字段只需要一个简单的队列积压数指标就行不用做太复杂。5. 常见问题与排查技巧实录5.1 问题速查表实操中踩过的坑集中整理成一张速查表遇到症状可以直接对照着查症状可能原因排查命令与解决方向触达目标一直返回 ReachOfflineSpoke 未注册成功或心跳丢失检查 Spoke 日志中的 register 结果数一下心跳间隔是否超过 45 秒消息已发送但对方一直不处理Spoke 侧消息队列积压检查 Spoke 的负载度字段查看 handler 是否发生异常卡死重复触达导致业务数据重复dedup_key 生成不稳定检查请求是否用了相同 dedup_key确认 Spoke 侧去重表是否落盘持久化链路状态停在 REACHED 不动目标收到消息但 handler 内部崩溃查看 Spoke 的异常捕获日志确认 handler 是否有 try-finally 保证 RESULT 返回调用方响应时间过长触达链路上有多跳串行检查每跳的超时设置考虑将串行跳改为 dispatch 并行Hub 重启后消息丢失未开启状态持久化确认 Hub 的 SQLite 持久化路径配置正确检查在途消息是否落盘5.2 埋点追踪一个 trace_id 贯穿到底排查 Agent 触达问题最怕的是不知道一条消息到底走到了哪一步。Agent-Reach 里每个触达请求都会生成一个 trace_id这个 id 会跟随协议头传到每一个相关节点。我在实际排查问题时最常用的手法就是拿到一个 trace_id然后去 Hub 的日志文件里 grep 一次把这条链路的所有事件全部拉出来按时间排序一眼就能看到问题出在哪一环。说实话这套机制在开发初期帮了大忙。当时好几次线上现象看起来像是“某个 Agent 收到了消息但没干活”结果拉出日志一看消息根本没到那个 Agent而是在路由表里就因为没有匹配能力被打回了。追踪日志记录的字段包括事件类型、节点 ID、消息 ID、前后状态、耗时、错误信息。这些字段尽量标准化方便后续做统计分析。我强烈建议任何做多 Agent 系统的人参考这个方案让追踪能力从第一天就内建不要等出了问题再补否则排查成本会让你崩溃。5.3 心跳抖动与离线误判心跳机制是我在测试阶段被整得最惨的一个环节。刚开始我把心跳间隔设为 10 秒离线判定阈值设为 20 秒想着这样检测快一点。结果跑起来后发现一旦某台机器发生一次 GC 停顿或者网络出现一次 3-5 秒的抖动Spoke 就可能错过心跳窗口被 Hub 误判为 offline已经分配给它的消息会被移到其他实例造成重复处理。后来我调整策略把心跳间隔和离线判定阈值分开并且引入了租约时间的概念。现在 Spoke 在启动时向 Hub 申请一个租约默认租期 120 秒每次心跳续租 60 秒。只要 Spoke 在租期内有任意一次活跃就不会被判定为离线这个缓冲空间大大减少了对瞬时抖动的敏感度。5.4 上下文缓冲泄漏裁剪策略的后遗症还有一个问题在长时间运行时才暴露上下文裁剪虽然每次控制了单条消息的体积但如果某条链路上累积的消息数特别多每个节点都会缓存大量历史触达记录内存占用还是会上来。我在代码里给每个 Spoke 加了历史消息上限默认保留最近 200 条完成记录超出后自动淘汰最旧的记录。如果业务需要更长周期追踪可以设置 archive 目录把旧记录落盘到 SQLite 归档表而不是一直留在内存。这里有个小建议归档的粒度按 trace_id 关联归档而不是按单个消息归档。因为一次业务动作往往涉及多个消息按 trace_id 归档后事后复盘时可以把整条链路都捞出来看可读性比单条消息归档好得多。5.5 补偿任务的正确打开方式当一条触达消息经历了全部重试仍然失败时Agent-Reach 不会把它静默丢弃而是会生成一条 FailureNotice发送给调用方指定的补偿通道。我在项目里把这个补偿通道设计成可以指向另一个 Agent 的能力这样就能实现自动化的失败处理编排。举个例子如果我的“生成日报”Agent 因为数据源异常触达失败我希望系统不要直接放弃而是触达一个“数据修复”Agent让它去检查数据源。这个补偿链路的配置不需要改业务代码只需要把 FailureNotice 时执行的动作配置成这样先触达数据修复能力修复完成后再次触达生成日报能力最多两轮。这种思路比硬编码一个 if-else 可靠得多因为补偿策略本身也能被编排、被观察。我个人的体会是Agent 协作系统里“失败”不是需要消灭的异常而是一种可以预见的常态。设计触达层的时候如果把大部分精力花在“如何避免失败”上效果往往不好反过来把精力花在“失败之后如何体面地恢复”上整个系统的稳定性反而会高得多。最后再多说一句Agent-Reach 这个名字里有“Reach”我对它的理解是Agent 真正能触达什么决定了它能干成什么。推理模型再聪明如果触达层像泥潭一样那整个系统的上限也被锁死了。所以别急于做复杂的 Agent 编排先把触达层做稳把每条消息的去向和状态都搞清楚你会发现很多之前觉得玄乎的 Agent 协作问题本质上就是通信问题。这个方向后续我还会继续扩展比如把触达层的遥测数据接到可视化面板上让整条链路的状态实时可见最近已经在动手了等有结果再回来细聊。
