Agent-Reach 这名字是我自己随手起的意思是“把 Agent 的触达范围做得足够大”。之所以起这个名字是因为我前几个月被单个 Agent 的边界问题折磨得够呛要么上下文塞满后开始胡说八道要么工具一多就互相干扰要么任务一复杂整个执行链条直接中断。后来我改了思路不再逼着一个 Agent 干所有事而是把能力拆开、把触角伸出去让它带上一群专职 Agent、一套工具注册表和一套兜底策略去做事。这套东西我内部代号就叫 Agent-Reach。这篇文章就把 Agent-Reach 从设计到落地的全过程写出来。它解决的核心问题是单 Agent 能力有限怎么通过多 Agent 协作、动态路由和工具化调度把系统整体处理复杂任务的上限撑高。适合正在做 Agent 应用、做 RAG、做工具调用聚合的开发者参考尤其适合那些已经写完一个 Agent demo、但不知道怎么把它变稳定的人。下面讲的内容没有特别高深的理论全是实打实能直接抄的架构思路、核心代码和踩坑记录。1. 设计初衷单 Agent 的触达瓶颈与 Agent-Reach 的定位1.1 单 Agent 为什么撑不住复杂任务最早我做的版本非常朴素一个大模型 Agent一把工具函数用户问什么我就让 Agent 去调什么。一开始跑 demo 很兴奋感觉自己已经掌握了 AI 应用的灵魂。但真正放到业务里去跑问题一个接一个冒出来。第一个问题是上下文窗口被工具调用记录快速消耗。你在一次任务里让它连续查三个数据源、做两次汇总计算中间还要处理报错Agent 的上下文中就会堆积大量工具输入输出。那些输出往往是 JSON 原文、日志片段、接口返回体又长又乱。上下文一长模型就开始“选择性失明”明明工具已经返回了正确结果它下一步却用起了自己编出来的数字。第二个问题是单点故障非常密集。Agent 的每一次工具调用都可能失败接口超时、参数格式不对、目标地址变了、权限过期。单 Agent 的逻辑是“出错了我就换个姿势重试”但几个错误叠加起来它往往会陷入重试死循环甚至在同一个错误上来回横跳。有一次我日志里看到同一个查询被反复执行了 11 次每次都返回同样的错误Agent 还在坚持说“正在重试请稍候”。这种体验放到生产环境是完全不可接受的。第三个问题是没法并行。一个 Agent 本质上是一个串行执行器想让它做十件事它就得按顺序一件一件来中间任何一环卡住后面全卡住。业务方要的是“同时把 10 家供应商的价格拉回来汇总”单 Agent 只能一家一家逛最后性能瓶颈不在模型推理速度上而在 Agent 的工作模式上。1.2 Agent-Reach 想解决什么Agent-Reach 的核心思路就是把一个全能的 Agent 拆成一组互相配合的专职 Agent每个 Agent 只负责一类触达。有的 Agent 负责检索资料有的 Agent 负责参数校验有的 Agent 负责写最终报告还有一个调度器 Agent 负责判断“这件事该派给谁”。我管这些专职 Agent 叫“触手”。每个触手都有一个明确的能力边界和工具列表触手之间不直接对话而是统一通过调度器来协调。这样做最大的好处是上下文不用再塞在一个模型里而是分散到各个触手各自的小上下文里。调度器只需要知道“任务拆到哪一步”“谁成功了谁失败了”“最终结果汇总在哪”具体的执行细节由触手自己承担。另外我还做了一个很关键的设计兜底策略绝对前置。调度器的指令集里明确写着任何触手连续重试两次没解决就立刻把任务标记为“异常”把不完整结果连同异常日志一起返回不允许无限重试。这个策略在真实环境里意义重大它保证了系统的响应时间是可控的不会因为某个接口抽风就拖死整个业务流程。Agent-Reach 的目标不是让 Agent 变得更聪明而是让 Agent 系统变得更可控、更可扩展、更不容易被单个节点拖垮。2. 核心架构把触角延伸到能覆盖一切复杂流程2.1 注册中心与 Agent 生命周期我的第一个决定是每个 Agent 都不是写死在大流程里的而是注册进来的。系统维护了一个Agent 注册中心里面保存着每个触手的能力描述、所拥有的工具列表、模型的配置参数以及当前的状态空闲中、执行中、异常。调度器在做任务路由的时候不是去硬编码“任务 A 走 Agent B”而是根据任务类型去注册中心查看哪个 Agent 的能力描述最匹配。这个设计的直接好处是加一个能力不需要改动现有流程。比如我突然想支持“定时任务监控”只需要写一个新的监控 Agent注册进去然后在调度器的任务分类规则里加一条映射关系就行。老流程完全不用碰。Agent 生命周期我用一个简单的状态机管理idle空闲待命running正在处理任务blocked等待其他 Agent 的输出或等待工具返回retrying失败了但还在重试阈值内dead超过重试阈值或被人为熔断调度器只跟非idle状态的 Agent 交流任务进度如果一个 Agent 长期停留在retrying调度器会直接把它标记为dead并把它的未完成任务转派给备用 Agent。2.2 任务拆解与路由规则Agent-Reach 里最重要的模块是任务拆解器。它接收用户的原始请求先把请求拆成若干个最小可执行单元然后逐个判断这些单元应该交给哪个触手。拆解逻辑我不建议直接用模型自由发挥而是要配置一个结构化的拆分模板。我以前踩过一个坑让大模型自由拆分任务结果它拆出来的子任务互相包含、逻辑重复调度器完全没法执行。后面我改成了这样拆解器的大模型固定输出一个 JSON里面包含任务名称、任务目标、依赖关系、预期输出格式每个字段都做格式约束不符合直接让模型重生成一次。路由规则也不靠纯语义匹配而是结合了关键词匹配和能力描述匹配的两层评分。比如任务里出现“查询”“搜索”“检索”这些动词检索 Agent 的基础得分就会被拉高出现“计算”“汇总”“对比”分析类 Agent 得分更高。如果两个 Agent 的评分非常接近调度器默认选择当前负载更小的那个避免每次都把所有任务堆到同一个触手上。2.3 上下文管理与任务记忆多 Agent 系统最容易踩的坑是上下文被切得七零八落。每个 Agent 都只看到了自己的小片段做决策时缺少全局信息导致上下游结果对不上。Agent-Reach 的做法是设计一个“任务记忆池”。每个任务从创建开始就会生成一个任务 ID所有 Agent 在执行过程中产生的重要中间结果都会写回这个记忆池。调度器在向触手派发新任务时除了说“去执行什么”还会附带一段从记忆池里捞出来的前置摘要。这个摘要不需要包含所有细节只需要包含执行下一步所必需的最小信息集合。举个例子用户要一份“华东区所有门店的库存积压报告”调度器先派“门店检索 Agent”去拉门店列表得到结果后由“存量分析 Agent”去计算积压。派给分析 Agent 的时候调度器会把门店列表的 ID 集合和名称映射表塞进附带上文而不是把原始大 JSON 全部塞进去。这样既保证了下游 Agent 有足够信息干活又避免了上下文膨胀。2.4 失败回退与兜底策略Agent-Reach 里我强制给每个触手配了三个等级的回退方案一级回退同一个工具重试一次用于应对瞬时网络抖动。二级回退换一个工具实现相同能力用于应对“主工具格式变了”或“主工具挂了”的情况。三级回退放弃执行把任务标记为异常返回“能力未覆盖”的说明。比较核心的一点是每一个触手都要在注册时就声明自己的二级回退工具。没有二级回退的 Agent 不允许注册上线。这么硬性要求是因为我见过太多次“一个工具失效整套流程瘫掉”的场面。有了二级工具兜底很多肉眼可见的故障在调度器层面就能消化掉。之前有一次某个数据平台接口调整了返回结构主工具挂了备用工具自动顶上业务侧完全无感知。3. 从零到可用Agent-Reach 的搭建与核心实现3.1 环境准备与基础依赖Agent-Reach 是一个纯 Python 项目核心依赖就四个openai负责调用大模型接口。如果你用的不是 OpenAI 兼容接口其他 SDK 也成但接口最好统一成一套。pydantic做结构化输出的校验。项目里所有 Agent 传入传出的 JSON 都用它做格式约束。fastapi跑调度器的 HTTP 服务。Agent 之间通过网络通信解耦比较干净。redis存放任务记忆池中的中间状态。用 Redis 的原因很简单要支持多实例部署不能把状态存在单机内存里。我用的是 Python 3.11逻辑上 3.10 都能跑。Redis 我是本地起了一个默认端口没有做持久化因为任务记忆池里的数据是瞬时的任务结束就该清理。3.2 调度器核心代码先说调度器整个 Agent-Reach 的心脏。它接收一个任务请求拆完子任务后把任务派发出去。这里我给出一个简化但完整可运行的版本class Scheduler: def __init__(self, registry, memory_pool): self.registry registry self.memory_pool memory_pool async def submit(self, task: dict): # 生成任务 ID task_id uuid.uuid4().hex self.memory_pool.init(task_id) # 任务拆解拆成多个子任务 subtasks await self.decompose(task[query]) results {} for st in subtasks: # 根据路由规则选 Agent agent_name self.route(st) agent self.registry.get(agent_name) if not agent: results[st[name]] { status: failed, reason: no capable agent } continue # 组装上下文附上记忆池里的前置摘要 context self.assemble_context(st, task_id) # 执行并记录结果 res await agent.execute(st, context) self.memory_pool.write(task_id, st[name], res) results[st[name]] res return self.post_process(results)这里面的decompose、route、assemble_context都是需要你按自己业务去填的实现。我给一个拆解函数的思路调大模型让它输出 JSON 格式的子任务列表然后用 Pydantic 校验校验不过就重新生成一次。两次都不合格就降级成“整单派发给一个全能 Agent”保证系统不会因为拆解失败而卡死。调度器代码看起来简单但真正的功夫全在“状态跟踪”上。生产版本里我给每个子任务都维护了一个超时计时器超过 60 秒没返回就自动标记失败并转派。3.3 工具注册与 Agent 间通信工具注册我采用了最简单可靠的“装饰器注册”方式。每个触手在初始化时都会将自己可用的工具扫描进自己的技能表from agent_reach import tool, Agent class SearchAgent(Agent): name search_agent description 负责各类资料检索与数据查询 tool(namesql_query, description执行SQL查询) def sql_query(self, sql: str) - str: return self.db.run(sql) tool(nameweb_search, description网页搜索) def web_search(self, keyword: str, top_k: int 5) - str: return self.searcher.search(keyword, top_k) async def execute(self, subtask, context): # Agent 自己决定怎么用这些工具 ...工具注册的核心价值在于能力可视化。调度器在路由的时候不需要理解每个工具具体能干什么只要读取每个 Agent 的描述和工具列表就能判断是否匹配当前任务。我把这些信息称为“能力元数据”。每次任务派出去之前调度器都会带着这些元数据去跟大模型做一次简短的“职介”让它确认手里的 Agent 确实适合这个任务。Agent 和 Agent 之间不直接通信所以没有复杂的消息协议。但只要存在跨 Agent 协作就必须解决“上游输出格式不稳定”的问题。我的方案是在上游 Agent 的定义里明确指定它所有输出都是一份“标准结果对象”里面包含三项——data业务数据、meta本次执行的元信息、warnings异常警告。下游 Agent 统一按这个格式解析上游结果不兼容就丢给调度器做一次转换。3.4 关键参数怎么调我实测下来影响 Agent-Reach 稳定性的参数主要有三个模型温度参数。调度器的拆解模型和路由模型温度一律设置成 0否则任务拆解会出现随机漂移同一个请求这次拆成三步、下次拆成五步调度逻辑根本没法稳定。执行层的 Agent 可以把温度设到 0.2 到 0.3保留一点灵活性但不要高于 0.5。上下文最大长度。我给每个 Agent 的上下文上限设置为模型窗口的 60% 左右。上下文余量是用来让模型处理工具返回值的如果一上来就把窗口占满模型后面基本没法正常工作。我通常在代码里做一次预检当上下文 token 接近阈值时先压缩中间结果压缩了还超就直接丢弃最老的工具调用记录。重试次数。这个建议别超过 2 次。我一开始设成 3 次结果发现大多数失败在第 3 次仍然是同样的失败只会白白增加延迟并消耗更多 token。设成 2 次后整个系统的平均失败恢复时间反而更短了因为多出来的时间交给了更靠谱的三级兜底。4. 实战中踩过的坑常见问题与排查实录4.1 死循环与无限递归最早跑 Agent-Reach 原型的时候我碰到最恐怖的问题就是死循环。场景是某个触手在执行任务时发现自己缺了某个前置数据于是向调度器发了一个“补充请求”。我当时为了偷懒让调度器直接把补充请求当作新任务再拆解一遍结果这个任务又回到了同一个触手手里触手发现自己还是缺数据又发补充请求整个系统就这么卡死了。后来我加的规矩是一个任务 ID 最多只能被调度器处理三次。第四次出现时直接拒绝返回“任务层级过深已终止”。我还把所有“补充请求”都打上了“DEBUG”标签每次触发日志都会高亮显示方便我观察到异常链路。你要做多 Agent 系统建议从一开始就引入类似的深度计数器别像我一样等线上事故才补。4.2 上下文污染上下文污染是多 Agent 系统里最隐蔽的问题。表现形式是A 触手执行完一个任务之后它的上下文里残留了上次任务的信息导致它处理新任务时带了“上一题的思维定式”。有一次我在测试中让一个检索 Agent 先搜“华东区门店面积”再搜“华南区库存积压”结果第二次它给出的报告里竟然还带上了华东区的门店名。排查了半天才发现是触手执行完任务后把历史消息原封不动保留了下来下一次任务时模型把两条历史消息串在一起解读了。解决办法是在每次任务开始之前对触手的上下文做一次“重置”。重置不是清空全部历史而是只保留系统提示词和任务描述类消息把上一次任务的工具调用记录全部清掉。如果跨任务确实需要保留一些长期经验就塞进系统提示词的“长期记忆”区域并且固定放在消息列表的最前面。4.3 Agent 之间互相阻塞多 Agent 系统还有一个很典型的死锁场景A 触手在等待 B 触手的结果而 B 触手也在等待 A 触手的结果。这在纯串行架构里不会出现但一旦引入了并行处理就很容易踩中。我的架构没有引入真正意义上的自由并行而是采用了一个叫“依赖拓扑”的策略拆解器在输出子任务的 JSON 里强制要求每个子任务声明depends_on字段。调度器只对depends_on为空的子任务进行并行派发其他任务严格按依赖顺序执行。这个策略让系统失去了部分并行效率但换来了非常强的可控性。对多数真实业务场景来说这个取舍是值得的。4.4 压测与稳定性检查我最后分享一个非常管用的压测工具组合用locust做并发测试用structlog给每个任务打上全局 Trace ID。每个 Agent 执行任务时日志里都会带上任务 ID、Agent 名和当前状态这样我从调度器日志里就能完整还原一条任务链的每个环节。压测的重点不是看成功率而是看失败模式的分布。如果失败全部集中在同一类工具调用上说明这个工具本身有问题如果失败出现在各个 Agent 的交互阶段说明消息协议或者上下文组装有 bug。有一次我压测时发现所有失败请求的耗时都集中在 65 秒左右排查后才发现是有个网络请求没有设置超时。后来我把所有 HTTP 调用的默认超时都设成了 20 秒错误立刻少了一半以上。这种问题不压测根本发现不了。5. 我的使用场景与后续想做的事Agent-Reach 目前的版本已经稳定跑了三个多星期。我主要拿它做这么几类事情一是做批量数据报告的自动汇总原来要业务同事手动查五个平台再拼报告现在派给 Agent 组自动跑二是做内部知识库的检索增强一个 Agent 负责跳转检索多个库另一个 Agent 负责把检索结果整合成带引用的回答三是做日常运维告警的初步分类和处置建议负责告警的 Agent 会先去查指标数据再回来生成一段分析说明。我自己的体会是Agent-Reach 最值钱的地方不是“支持多少个 Agent”或“调用起来多方便”而是把不确定性处理到了系统的边界上。单个环节失败不会拖垮整个任务上下文不会因为工具调用疯狂增长路由逻辑清晰到任何一个新接手的人都能读懂。做 Agent 应用最怕的不是模型不够聪明而是行为和失败路径不可控。后续我计划做两件事一是给调度器加一个简单的“经验学习”机制把历史上失败过的任务方案记录下来下次遇到类似任务时直接照抄成功方案作为参考二是把各个触手的状态面板做成可视化界面这样任务卡住时不用翻日志就能一眼看出堵在哪个环节。第一个版本我已经开始写了其实就是在记忆池里多存一张“任务模式-成功路径”映射表实现上没有想象中复杂。等这两个功能跑稳了我再开一篇讲讲里面的细节。