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(name="sql_query", description="执行SQL查询") def sql_query(self, sql: str) -> str: return self.db.run(sql) @tool(name="web_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 应用最怕的不是模型不够聪明，而是行为和失败路径不可控。
后续我计划做两件事：一是给调度器加一个简单的“经验学习”机制，把历史上失败过的任务方案记录下来，下次遇到类似任务时直接照抄成功方案作为参考；二是把各个触手的状态面板做成可视化界面，这样任务卡住时不用翻日志就能一眼看出堵在哪个环节。第一个版本我已经开始写了，其实就是在记忆池里多存一张“任务模式-成功路径”映射表，实现上没有想象中复杂。等这两个功能跑稳了，我再开一篇讲讲里面的细节。
