1. 为什么做 Agent-Reach我从单一大模型调度一切的教训说起先交代一下背景。我手上同时跑着一组垂直 Agent一个负责抓取并清洗数据一个负责做摘要改写一个负责生成代码还有一个负责 Code Review 和测试用例补全。早期它们的协作方式是典型的中心化指挥——所有任务先进一个主 Agent由这个大模型自己判断该把活分给谁。听起来很智能但真实跑起来问题非常集中主 Agent 的上下文窗口被塞得越来越满它越来越懒经常把明明该给数据 Agent 的活自己顺手干了更离谱的是有一次它判断错了归属把一段 SQL 生成任务丢给了摘要 Agent结果产出了一篇看起来很有道理的假 SQL。我开始意识到一个问题多 Agent 系统里任务能不能准确、高效地到达正确的 Agent这件事本身就应该是一个独立的工程问题而不应该交给模型现场猜。于是有了 Agent-Reach 这个项目。它的定位很简单在多 Agent 之上加一层轻量的路由调度层负责三件事——知道每个 Agent 能干什么、判断当前任务该给谁、以及处理给出去之后没干成的各种意外。这个项目适合谁看呢如果你手上有两个以上职责不同的 Agent并且已经为任务分错人、Agent 互相踩踏、任务静默失败这些事头疼过那这篇文章里的设计思路和踩坑记录应该能直接帮到你。下面我按为什么这样设计 → 怎么落地 → 踩了什么坑 → 实测结果的顺序完整讲一遍。2. Agent-Reach 的核心设计注册中心 打分路由 降级链2.1 先注册、后干活能力画像表是路由的地基我做 Agent-Reach 时定的第一条规矩是任何 Agent 想接入系统必须先注册自己的能力画像否则路由层一律不认它。这个画像不是一句话简介而是一张结构化表格我通常叫它 capability profile包含下面这些字段字段说明示例agent_nameAgent 唯一标识data_fetchercapability_tags能力标签建议控制在 5~10 个[数据抓取, 清洗, csv, api]input_schema它接受的输入结构{url: string, pagination: int}output_schema它返回的输出结构{records: array, sample: object}priority同一任务多候选时的优先级3timeout_seconds单次执行超时上限60health_check健康检查接口或命令ping为什么必须这么做因为路由的前提是比较而比较的前提是有统一的度量方式。如果只靠自然语言描述主模型去判断哪个 Agent 更适合时仍然会抖——今天选 A明天可能选 B没有任何一致性。结构化画像就能让打分逻辑变成确定的代码而不是看模型心情。另外注册表必须支持运行时刷新。我见过不少团队把 Agent 能力写死在配置文件里一改就要发版重启。Agent-Reach 里我把注册表做成了可以热更新的新 Agent 上线、旧 Agent 下架、能力标签调整通过管理接口直接改路由层下一跳就能感知。这一点在后来的灰度发布中帮了大忙。2.2 任务不靠猜靠打分意图提取与匹配排序注册有了接下来是核心怎么决定一个任务该路由给谁。我的做法分两步——先提取意图再计算匹配分。意图提取这一步我并不是每次都用大模型而是先走一遍轻量规则 关键词词典的快速判断命中率高到一定程度就直接出结果。例如任务文本里出现了抓取爬取采集下载这类词意图基本可以锁定为 data_fetch只有快速判断置信度低于阈值时才调用一次大模型做补充提取。这样做的原因只有一个——延迟和成本。路由层本身不应该成为全系统的性能瓶颈每单都额外调一次大模型延迟直接翻倍小流量场景无所谓流量一上来就顶不住。匹配分我用一个简单但有效的加权公式def match_score(task_intent, capability_profile, task_payload): tag_score 0.0 for tag in capability_profile[capability_tags]: if tag in task_intent.keywords or tag in task_payload: tag_score 1.0 schema_score check_schema_compatibility( capability_profile[input_schema], task_payload ) # 0.0 ~ 1.0 health_score capability_profile.get(health_score, 0.9) return ( tag_score * 0.5 schema_score * 0.3 health_score * 0.2 )打分之后路由层拿到一个降序排列的候选列表。注意我从来不做只取第一名这种绝对策略而是取前三名进入下一轮判断。原因很实际第一名可能恰好处于过载状态也可能它的历史成功率最近下滑得厉害。这些信息不在匹配分里但在下一层的决策里很有价值。2.3 降级链把没人接单变成一种常态处理任务路由出去之后无非两种结果Agent 接单并成功、或者接单失败。真正容易出问题的是系统对失败的处理方式。Agent-Reach 里每条任务都有一个降级链fallback chain最少两环最多五环。第一环是匹配分最高的 Agent如果它超时、报错或者返回结果校验不通过路由层自动把任务投给第二环第二环再不行就第三环。每一环之间我强制设置了最小间隔默认 2 秒这是为了给上一个 Agent 留出清理资源的时间避免同一时间多个 Agent 同时处理同一份不干净的任务数据。降级链的最后一环一定是兜底策略三种可选重试、放队列延迟再试、转人工。我一般默认把转人工作为最终兜底因为自动化的链条上如果全都失败了说明大概率是任务本身有问题比如输入数据格式变了这时候继续让机器试只会浪费 token不如把原任务和失败原因完整打出来丢给人工看一眼。3. 落地实现消息协议、路由配置与关键代码3.1 消息协议五个核心字段多一个都嫌多Agent-Reach 的任务消息格式我反复精简过最后稳定成五个字段{ task_id: task_810937, intent: data_fetch, payload: { url: https://example.com/list, pagination: 50 }, priority: 2, idempotency_key: fetch_810937_20241112 }task_id用于全链路追踪intent是路由层判断后的意图标签payload是实际任务内容priority决定队列里的调度权重idempotency_key是幂等键后面踩坑部分我会专门讲它。很多人会在协议里加一堆扩展字段预留字段我的建议是不要加。协议字段越少Agent 侧的解析成本越低出错面越小。真需要扩展信息的时候塞进 payload 里就行没必要动协议本身。3.2 路由核心代码不到两百行但每一行都踩过坑路由层的核心逻辑我不喜欢搞复杂就是取候选 → 打分排序 → 按降级链投递 - 校验结果 - 记录观测数据这个循环。直接看代码import time, uuid from collections import defaultdict class AgentReachRouter: def __init__(self, registry, queue): self.registry registry # Agent 注册表 self.queue queue # 任务队列 def route(self, task): candidates self.registry.find_candidates(task[intent]) scored sorted( candidates, keylambda c: match_score(task[intent], c, task[payload]), reverseTrue ) chain scored[:3] self.registry.dedicated_fallback() for idx, agent in enumerate(chain): if not agent.healthy(): continue try: result self.dispatch_with_timeout(agent, task) if self.validate_result(result, agent): self.record(agent, task, success, result) return result self.record(agent, task, invalid, result) except TimeoutError: self.record(agent, task, timeout, None) except Exception as exc: self.record(agent, task, error, str(exc)) if idx len(chain) - 1: time.sleep(2) # 降级间隔给上一个 Agent 留清理时间 return self.registry.escalate_to_human(task)我特别想强调两点。第一record()这个方法绝对不能省它把每次投递的成功、超时、无效、异常全部落库。没有这些数据你后面根本没法回答哪个 Agent 最近变弱了哪个时间段失败率最高这类问题。第二validate_result()必须按照 Agent 注册时声明的output_schema来校验不能只看没报错就算成功。我见过太多 Agent 返回了 200但返回体的字段结构完全不对下游程序直接炸。3.3 配置文件把变化的部分和稳定的部分分开Agent-Reach 的配置我觉得最有价值的一个设计是把 Agent 自身参数和路由策略参数拆成两份配置。Agent 自身的参数超时、重试次数、并发上限、健康检查间隔跟着 Agent 走放 agent 配置区agents: data_fetcher: timeout_seconds: 60 max_concurrency: 4 health_check_interval: 30 priority: 3 code_generator: timeout_seconds: 120 max_concurrency: 2 health_check_interval: 60 priority: 2路由策略参数降级链长度、降级间隔、打分权重、兜底方式放路由配置区routing: fallback_chain_length: 3 fallback_interval_seconds: 2 score_weights: tag: 0.5 schema: 0.3 health: 0.2 final_fallback: human为什么要分开因为两个维度的变更频率完全不一样。Agent 参数是某个 Agent 出问题时局部调整路由策略是整体调优。混在一个文件里每次改路由权重都要把十几个 Agent 的配置一起审一遍改动影响面被人为放大了。分开之后改一个 Agent 不需要碰路由改路由不需要碰 Agent干净很多。4. 踩坑实录三个差点让我放弃 Agent-Reach 的问题4.1 并发投递导致的乱序任务到了但结果不是最新的第一次压测的时候我盯着日志发现一个非常诡异的现象同一个源数据的抓取任务明明按时间顺序先后投递了 task_001、task_002、task_003但最终入库的结果却是 task_003 的旧数据覆盖了 task_002 的新数据。排查下来问题出在并发投递上。我的队列用了多消费者模式三个任务被不同 worker 同时取走但执行快的反而是后投递的 task_003。最终写入时没有任何版本控制先写完的 task_002 反被后写完的 task_001 覆盖了。这个坑的教训是在分布式任务系统里任务的投递顺序不等于完成顺序数据落地必须以版本号或时间戳为准。我的解决方案是在 payload 里增加version字段同时数据落库时做条件更新只有传入的task_id关联的序列号比库里的新才允许写入。这个改动很小但彻底杜绝了脏覆盖。4.2 上下文污染上一单的残渣全流进了下一单里第二个坑更隐蔽。我的摘要 Agent 内部维护了一个上下文缓存本意是给同一源站的多篇内容做对比总结时用。结果某天线上出现了一批牛头不对马嘴的摘要——第一篇讲的是汽车行业财报第二篇摘要的开头居然还残留着车企毛利率的引用。原因很简单上下文缓存没有按 task_id 隔离。后一个任务进来的时候前一个任务的系统提示和参考材料还留在 buffer 里被当成了当前任务的上下文一并送入模型。这类问题比乱序更危险因为它不报错只会在输出里悄悄混入不该有的内容人眼不仔细看很难发现。修法是执行会话的严格隔离每个任务独立创建上下文容器任务结束立刻销毁绝不复用。同时我在路由层的validate_result()里加了一道抽查逻辑对比输出文本里是否存在不属于当前 payload 的高频实体词有就标为 invalid。这个防呆设计后来真的拦下来两次上游脏数据事件。4.3 幂等性缺失一次重试用户被扣了两笔钱第三个坑是最严重的发生在接了一个计费类业务之后。某个付款通知任务第一次投递给核心 Agent 时超时了我设置的降级链触发了重试。结果其实第一次调用并没有真正失败只是在网络上多绕了一圈服务端已经处理成功了。重试的第二次调用又成功处理了一次——用户被扣了两次钱。这个事故让我把幂等性提到了最高优先级。Agent-Reach 的每个任务现在必须携带idempotency_key所有可能产生外部副作用的 Agent 在执行前都要先检查这个键对应的状态如果已经是processed就直接返回上一次的结果不再执行。状态机只有四个状态pending→running→done/faileddone状态不可回退。def execute_with_idempotency(agent, task): key task[idempotency_key] state agent.state_store.get(key) if state and state[status] done: return state[result], from_cache agent.state_store.set(key, {status: running}) result agent.execute(task[payload]) agent.state_store.set(key, {status: done, result: result}) return result, executed这个改动之后我们的重试逻辑才真正敢放开。以前重试是心惊胆战的现在是业务上可以接受的常态。5. 实测数据与非功能指标Agent-Reach 的稳定性验证5.1 压测方案与关键观测指标Agent-Reach 跑通之后我做了一轮为期两周的稳定性压测重点盯四个指标任务分派成功率任务成功落到某个 Agent 的比例预期 100%、路由层额外延迟不加路由层 vs 加路由层的端到端差异、降级触发率多少任务动用了第二环及以后、无效结果检出率validate 环节拦截下来的脏输出占比。实测数据大概是这样的指标不加路由层加 Agent-Reach任务分派成功率87%主要失败在分错 Agent99.6%端到端平均延迟4.2s4.5s路由毛刺约 0.3s降级触发率无统计7.8%大部分是超时触发无效结果检出率无统计3.1%全部被拦截路由层额外延迟控制在 0.3 秒左右主要花在意图提取的规则匹配上。这个代价换来的收益非常明显分派成功率从 87% 提到 99.6%而那 0.4% 的失败也都有明确的 error 记录和人工兜底跟踪不再有静默的活丢了。5.2 为什么这套设计比让模型自己选 Agent更稳我把 Agent-Reach 和新老两种方案做过对比方案 A 是主 Agent 自由决定分给谁方案 B 是固定规则写死 1 对 1 映射方案 C 是本文的注册 打分路由。方案 A 的问题是输出不稳定同样的任务今天给 A 明天给 B且主 Agent 上下文一大就频繁误判方案 B 的问题是太脆任何新增 Agent 都要改规则运维成本随 Agent 数量线性上升方案 C 的注册表天然支持横向扩展新 Agent 上线就是一次注册操作路由逻辑本身不用动。这套设计的本质是用结构化的能力描述去替代模型的主观判断只把语义理解这一步留在模型侧而且还有规则快速通道兜底其余全部变成确定性逻辑。稳定性的收益就是这么来的。6. 如果重新做一次我会保留和改掉什么Agent-Reach 用了大半年整体上它解决了最初任务总到不了对的 Agent这个核心痛点。但如果让我重新做一遍有三件事我会从一开始就调整。第一意图提取我会直接上一个小型分类模型而不是先规则后大模型的凑合方案。规则通道虽然快但维护关键词词典本身就是一种隐形成本每周都有新词要补分类微调模型的维护反而更省心。第二健康检查不应该只做成能响应就行的 ping应该加入输出质量回归测试——每天用一组固定样例任务跑一遍所有 Agent输出和基线差异超过阈值的直接标黄。这个想法是我后来才补上的早期有一些 Agent 能力悄悄退化了两周都没被发现因为 ping 一直都是通的。第三我会更早地把观测数据接入可视化面板。Agent-Reach 从一开始就埋了观测点但很长一段时间我只查日志不看趋势图。直到某天发现某个 Agent 的成功率曲线是断崖式下跌又自动恢复的形状才意识到这种模式其实是它的依赖接口在做上游切换。这种事靠人盯日志是盯不出来的必须靠趋势图。最后分享一个小技巧任务消息里的intent字段一定要人工可以覆盖覆盖调试时可以直接手动指定意图观察路由结果。这个看似不起眼的后门在排查问题的时候能省一大半时间。Agent-Reach 这个项目本身不是为了追求复杂而是为了把多 Agent 协作里最烦人的那部分不确定性用工程手段压到最低——做完之后你会发现Agent 们终于开始各司其职了。