做 Agent 项目的人越来越多但大多数团队其实都卡在同一个地方单个 Agent 能做的小任务很顺一旦任务变复杂需要多个 Agent 协作时整个链路就乱成一团。我前阵子自己动手搭了一个叫 Agent-Reach 的项目核心目标很直接——让不同能力的 Agent 能互相找到彼此、把任务分发下去、再把结果收回来。这篇文章就是我自己趟完坑之后的完整复盘从设计思路到实操踩坑都会聊适合正在做 Agent 编排、多智能体协作或者客服分流系统的朋友参考。Agent-Reach 这个名字里有两个关键词一个是Agent指的是具备某种专业能力的智能体一个是Reach强调的是触达和覆盖。合在一起它的定位就是一个轻量级的智能体协作网络重点解决三件事谁能接这个任务、怎么把任务给到它、它的结果怎么被其他 Agent 认账。听起来不复杂但真正跑起来之后需要处理的问题远比想象中多。1. 整体设计思路Agent 协作本质上是分工加共识在动手写第一行代码之前我先花了大量时间思考一个问题Agent 协作和平时我们写微服务到底有什么区别后来想明白了——微服务的接口是提前定义的A 服务调 B 服务参数和返回值都写死但 Agent 协作的输入输出是自然语言你根本没法提前穷举所有任务类型。所以 Agent-Reach 的核心设计必须围绕一个词动态。1.1 选择 Hub-Spoke 而不是 P2P 的原因最朴素的想法是让每个 Agent 都能直接调其他 Agent这就是 P2P 拓扑。我一开始也试过两个 Agent 互相调还行到第五个 Agent 的时候调用关系变成了网状每个人都维护一份谁能干什么活的名单更新一个 Agent 的能力描述其他 Agent 全得跟着改根本没有可维护性。所以 Agent-Reach 改成了 Hub-Spoke中心辐射结构中间一个 Router旁边挂着一堆 Worker Agent。所有任务先统一到 Router由 Router 决定分给谁Worker 之间不直接通信也互相不知道对方的存在。这样做有三个好处新增一个 Agent 只需要向 Router 注册一次能力描述其他 Agent 无感知。路由决策集中在一处方便做权限控制、负载均衡和审计。出现问题可以单独把某个 Worker 摘掉不影响整个网络。代价也很明显Router 变成了单点一旦挂了整个协作网络就瘫了。我的解决方式是 Router 纯内存态 无状态设计搭配一个简单的健康检查重启只需要几毫秒任务通过持久化队列做了补偿所以实际跑下来问题不大。1.2 唯一的全局共享产物任务协议Task ProtocolHub-Spoke 解决了找谁干活的问题但还没解决活干完之后结果算谁的。多个 Agent 协作最怕的是各自为政、结果格式对不上。A Agent 给的是 MarkdownB Agent 期望的是 JSON中间还得专门写一个转换器。我参考了行业里比较成熟的 MCPModel Context Protocol思路但没直接上agents商用框架因为它的学习成本和运行体量对我来说太重了。Agent-Reach 自定义了一套轻量级任务协议核心就五个字段{ task_id: uuid-v4, type: analysis | generation | search | action, input: { 原始数据或引用 }, output_schema: { 期望的返回结构 }, callback: http://router/collect }关键不是这五个字段本身而是约定所有 Agent 必须返回 JSON且必须带task_id回传给 Router。没有这个约定你就没法把分散在不同 Agent 手里的结果拼回一个完整答案。这算是 Agent-Reach 最重要的隐性设计——它不约束你怎么说但约束你回哪去。2. 核心细节解析Router 的路由策略是成败关键Router 是 Agent-Reach 的大脑它的工作流程拆开看是三步意图识别 → Agent 匹配 → 结果路由。前两步决定了活派得准不准最后一步决定了活交得顺不顺。2.1 意图识别不是直接问 LLM 分给谁我最初的做法很天真直接把用户的问题和所有 Agent 的说明书拼成一段话让 LLM 自己判断。效果特别不稳定Agent 一多LLM 就开始幻觉式分配明明只有 3 个候选它总能编出第 4 个。后来我改成了两段式第一步先做意图分类也就是把用户的问题归到几个大类型里比如数据分析、文案生成、信息检索、流程触发第二步再由一个小的匹配器根据类型去查 Agent 注册表。这样一来LLM 要做的选择题从从 20 个 Agent 里选一个变成从 5 个类型里选一个准确率明显提升而且分类错误比选错 Agent 更容易被后续校验发现。这里有个经验LLM 做加法容易做减法难。让它在开放集合里挑答案不如收敛几条路让它选。2.2 Agent 匹配能力描述不是给人看的Router 内部维护了一张 Agent 注册表每增加一个 Worker就需要提交一段能力描述。这段描述不是写给人看的而是写给匹配算法看的所以冗长的自然语言反而有害。我踩过的坑是第一个版本的 Agent 说明书写得特别丰满什么我是一个资深的数据分析师拥有十年数据分析经验熟悉 Python Pandas 和各种统计方法……结果匹配的时候LLM 经常被前几行吸引忽略了真正管用的关键词。后来规定能力描述必须缩成一句话格式是[能力标签][一句话说明]。比如数据分析输入表格数据输出统计指标和可视化建议。标签用于硬匹配过滤一句话用于 LLM 排序。用标签先粗筛再用 LLM 精排这样既准又省 Token。2.3 触达协议Worker 之间通过 Router 传数据之前说 Worker 之间不直接通信那任务需要多轮协作怎么办比如查一下最近一个季度的销售额然后对比去年同期的增长率。这其实不是一个 Agent 能搞定的需要检索 Agent 先拿数据再给分析 Agent 用。Agent-Reach 的处理方式是任务链Task Chain。Router 收到任务后不是只派给一个人而是拆成多步每步的输出作为下一步的输入由 Router 在中间做流转。每个 Worker 都只拿到自己需要的部分不需要感知整条链。# 伪代码展示 Router 如何编排任务链 chain [ {agent: data_fetcher, input: 2024_Q1 销售明细}, {agent: analyst, input: 上一步的原始数据计算同比增长率}, ] for step in chain: result dispatch(step) # 发给对应 Worker step[input] result[output] # 流转这是整个系统里我最满意的一块设计。它等于把所有 Agent 变成了一个流水线上的工位每个人只认自己面前的零件和图纸组装是 Router 的事。3. 实操过程把 Agent-Reach 的最小闭环跑起来不废话直接说怎么从零开始落地。我用的是 Python FastAPI OpenAI-compatible 接口本地接的 Qwen整个系统只需要三个部分组成Router 服务、Worker 服务、一份注册表配置。加起来核心代码不到 400 行但足够跑通一条完整任务链。3.1 环境准备与依赖pip install fastapi uvicorn httpx openaiWorker 不一定要用 Python我用 Node.js 写过一个版本只要它能调用 HTTP 就行。关键点是Agent-Reach 不限制语言限制的只是协议。任何一个语言只要能发起 HTTP 请求、接收 JSON、处理 JSON 并回传都可以接入。3.2 Router 实现一个核心函数的演进Router 最核心的函数就是route(task)。它的输入是任务字典输出是聚合后的结果。我贴一个高度简化但可运行的版本import asyncio import aiohttp async def route(task, agent_registry): # 第1步意图分类 task_type await classify_intent(task[input]) # 简单一点用关键词LLM兜底 # 第2步匹配候选 Agent candidates [a for a in agent_registry if task_type in a[tags]] # 第3步LLM 精排 chosen await rank_agents(candidates, task[input]) # 第4步分发并等待 async with aiohttp.ClientSession() as session: async with session.post(chosen[url], jsontask) as resp: result await resp.json() # 第5步校验结果确保 task_id 一致 if result.get(task_id) ! task[task_id]: raise ValueError(Agent 返回了错误的 task_id) return result这里有三个细节值得展开第一个是classify_intent。我最后没用纯 LLM而是做了一个混合方案先用一组正则和关键词做快速分类命中直接返回没命中再用 LLM 兜底。原因很简单很多任务长得很像分析一下这份销售数据和把这份销售数据做成图表都包含销售数据但前者是分析后者是生成。规则能搞定大部分高频场景LLM 只处理模糊场景成本和延迟都能降下来。第二个是rank_agents。当候选有多个的时候我会把任务输入和候选 Agent 的一句话描述拼在一起让 LLM 打分要求只返回一个序号和一句话原因。这步不能省的原因是同一个数据分析 Agent 下可能挂了销售分析和用户行为分析两个子 Agent标签一样但专长不同硬匹配会随机选。第三个是task_id校验。这个很多人会忽略但它是防串味的关键。没有它一旦某个 Worker 延迟响应Router 把上一个任务的结果返回给了下一个任务排查起来想哭。3.3 结果聚合把多条链的返回值拼成最终答案任务链跑完之后每个步骤的输出都收集在一个列表里。聚合层做的事情很简单按链的顺序把每一步的输出拼接成一个最终的 JSON 结果同时附上完整的链路日志。def aggregate(chain_results): final { task_id: chain_results[0][task_id], status: success, chain_summary: [] } for step in chain_results: final[chain_summary].append({ agent: step[agent], output_preview: str(step[output])[:200] }) # 默认取最后一步的输出作为主结果 final[final_output] chain_results[-1][output] return final聚合逻辑我建议保持在尽可能傻的水平。不要在这里做智能拼接、不要试图用 LLM 把各步结果再润色一遍除非你有极强的结果一致性要求。理由很简单聚合层一旦开始聪明你很难判断最终结果的偏差是哪一步造成的。先保证透明可追溯再考虑优化输出质量。4. 常见问题与排查技巧实录Agent-Reach 从第一版到跑稳大概两周时间一半的时间都花在排查下面这几个问题上。我把它们按出现频率排了个序大家大概率也会遇到。4.1 路由递归失控这是我遇到的最吓人的一个问题Router 把一个任务分给 Agent AA 发现任务需要 B 的输入于是 A 调 Router 发了一个新任务给 BB 又发现需要 CC 又调回 A……最后所有 Agent 都在互相等Router 的队列里堆了几百个任务。排查思路很简单给任务链加最大深度限制。我在协议里加了一个max_depth字段Router 每次派发前检查当前深度超过 3 层直接返回错误。另外给每个任务链生成了chain_id同一个chain_id的任务如果重复到达 Router直接丢弃。我在生产系统里把这个叫做任务染色——每一轮任务都带一个唯一颜色撞色就说明出圈了。4.2 上下文丢失与状态隔离问题多 Agent 协作最常见的记性差问题根源通常不在 Agent 本身而在 Router 没有做好状态管理。比如第一步的检索结果应该在第二步还能引用但如果 Worker 是独立的 HTTP 服务它根本不记得自己上一个任务干了什么。我的对策是Router 不传全文上下文传带引用的数据快照。每一次路由都携带一个payload内容是上一步输出的全部 JSON。虽然笨但可以保证每一步都是无状态的Worker 每一次任务都从零开始不存在记错的问题。代价是 Token 开销变大解决方法是只传递当前步骤真正需要的字段而不是把整条链的中间结果都带上。4.3 成本控制与 Token 优化Agent-Reach 跑起来之后最直观的痛点是调用量叠加。一个用户问题可能要先做意图分类几百 Token再调一次精排又几百 Token再跑 3 步任务链每步上千 Token一个简单的任务轻松烧掉几千 Token。这个成本在真实场景里是没有办法靠砸钱解决问题的得从架构上想办法。实操中比较管用的四招分类尽量用规则我在前面提到的关键词快速分类在实测里能拦下 60% 的请求完全不用调 LLM。精排只在有歧义时触发如果候选 Agent 只有一个直接跳过精排步骤。小任务用小模型意图分类、关键词匹配这种低复杂度任务用最新的轻量模型足够了没必要让大模型出场。缓存任务链如果同样的任务链连续出现且输入相似直接把上一步结果缓存住避免重复执行昂贵的 Agent 调用。4.4 权限与安全边界Agent 能触达的数据和操作权限是整个系统里最容易出事的环节但也是很多项目最早妥协掉的。Agent-Reach 的初始版本就吃过亏一个 Agent 可以读取任意用户数据另一个 Agent 能触发任意操作。在测试环境里没事一上线就被安全同事盯上了。后来我加了一个相对实用的设计——权限标记。每个 Agent 注册的时候绑定一组权限标签比如read_user_data、send_email、modify_order。Router 在路由之前检查用户请求的权限范围如果请求根本没触发能修改订单的 Agent 权限就直接拒绝。虽然增加了不少配置工作但至少在出了问题的时候你可以迅速定位是哪一步越权。4.5 超时与重试策略多 Agent 协作里单点超时是最烦人的。某个 Worker 是个外部模型接口慢的时候几十秒才返回后端链路会超时前端用户早就离开页面了。我的策略是分级设置超时轻量 Agent如检索类8 秒重量级 Agent如生成长文30 秒超过时间直接标记失败并尝试重试一次。重试不是重新发同一个请求而是带上stale_ok标记让 Agent 返回已缓存的结果——避免因为重试导致重复执行。这个方法不能解决所有超时问题但至少能让链路由全部挂掉变成部分可用。5. 实测效果数据说话不吹不黑Agent-Reach 在我自己的一个小型知识库问答 数据分析场景里跑了一周用 10 个 Worker Agent处理了 1650 条真实请求。我记录了几个关键数据可以给大家做个参考这些数据也直接影响了我后续的优化方向。指标数值说明路由准确率91.6%正确识别意图并分派给正确 Agent全链路成功率85.2%完整跑完所有步骤且无错误平均端到端耗时9.8 秒从用户提交到拿到最终结果核心链路超时率4.3%单步任务超过30秒单任务平均 Token 消耗4126包含分类、精排、任务链91.6% 的路由准确率不算特别高但考虑到里面还有不少 LLM 判断带来的误差我认为可以接受。整个链路跑下来最有价值的反馈其实不是准确率而是谁能用、谁不能用。我重点复盘了失败案例中的几类也给出了对应的解决方案大家可以直接抄作业。失败场景比较典型的是这几种列成速查表方便对照问题现象可能的根因我的解决手法任务被 Router 分配到错误的 Agent意图分类把类型搞错了把这个任务样本加入分类训练集并在关键词规则里补充正则某一步 Agent 总是超时模型接口响应慢或输入过长给该 Agent 单独配置更高的超时阈值同时压缩输入字段两个 Agent 返回结果风格差异太大各自 prompt 风格不同导致输出乱在协议里统一增加format_style字段要求按模板输出任务链第二次执行时结果不如第一次模型输出本身有随机性对纯检索类步骤开启幂等缓存同输入直接复用上次结果最终结果里有一些字段没填某一步 Agent 返回了不完整 JSON增加输出校验缺失字段会在聚合层直接标红并重试该步某个 Agent 突然返回 401 错误接口 Token 过期或权限被回收写一个定时脚本每次调用前检查 token 有效期过期自动刷新我最想多啰嗦一句的是输出不完全 JSON这个坑。你可能会想JSON 格式不对不是基础问题吗但真实模型真的会时不时给你返回一段 Markdown 或一个额外的说明后缀。我在聚合层加了一个extract_json函数优先用正则提取{...}部分失败再用 LLM 修正。这个处理逻辑看起来很土但非常扛打救了我很多次。6. 个人体会和后续扩展方向Agent-Reach 做到现在我个人最深的体会是做多 Agent 协作不要一开始就追求视觉上的智能感先把基础的路由、协议、幂等、超时这些工程问题解决掉Agent 们才能真正干起活来。很多项目翻车翻的往往不是模型能力不够而是 Agent 之间缺乏一个可靠的调度中枢。这中间我用到的很多经验来自行业里公开讨论过的通用模式——像任务链、意图分类、结果校验这些思路合理延伸后用在 Agent-Reach 上整体是行得通的。如果有朋友正打算做类似的 Agent 编排系统我的建议是先画清楚三个图——任务流转图、Agent 能力矩阵图、权限边界图然后才动工写 Router。后续我打算给 Agent-Reach 扩展两块内容一块是可视化任务链追踪目前链路日志只能靠 JSON 翻不太直观另一块是增量式路由学习把每次人工纠错的样本收集起来定期微调分类模型让路由准确率从 91% 往 95% 以上走。这块做好了Agent-Reach 对外的价值就不仅仅是一个内部玩具而是真正可以对外输出能力的一套方案了。