最近在折腾智能体 Fleet说实话一开始我是想偷懒的——一个能干的智能体搞定所有事不比维护一帮萝卜头省心结果实际跑下来发现单体智能体在真实业务场景里根本扛不住。上下文窗口有限、工具权限混在一起、一个大任务跑着跑着就状态错乱。把智能体从单个升级成 Fleet一队协同工作的智能体集群成了我近期最值得的一次重构。这篇内容直接把我的设计思路、踩坑记录、可复现的最小实现都摊开讲适合正在做多智能体应用、或者被单体 Agent 逼疯的开发者参考。1. 为什么要把智能体从单体改成Fleet1.1 单体Agent的真正瓶颈在哪很多人一开始做智能体应用都是一个 Agent 包打天下给它一个目标让大模型自己规划、自己调工具、自己给结果。Demo 阶段很爽但一旦任务变复杂问题就集中爆发。首先是上下文危机。我试过让单个智能体处理一个包含多步骤的数据分析任务模型要同时记住用户原始需求、中间查询结果、代码执行输出、历史修正记录。上下文窗口再大也架不住这样消耗而且越到后面模型越容易忘记前面的关键约束输出的质量肉眼可见地下降。这就像让一个人既当前台接待、又当财务核算、还要写代码记性再好也有崩溃的时候。其次是故障放大效应。单体 Agent 里只要有一个环节出错——比如某个工具调用格式返回异常、某段代码执行超时——整个任务就失败重来前面所有步骤全部作废。没有隔离、没有重试粒度、没有局部恢复手段。我踩过一次很深的坑一个跑了十几分钟的长任务在最后一步输出解析环节报错整个流程打回原点气得我直接失眠。最后是扩展性很差。想提升能力只能换更大的模型换一次模型要重新适配所有提示词和工具调用逻辑。想让不同环节用不同规格的模型做成本优化单体结构根本做不到。1.2 Fleet带来的四个实际收益Fleet 的核心思想很简单把一个复杂的智能体拆成多个职责单一的工作单元由一个编排器统一协调。这样每个单元只需要做好一件事模型压力小上下文干净失败也能局部处理。收益一专业化分工。让一个模型只做任务拆解另一个模型只做代码生成还有一个只做结果审查。每个单元的提示词可以打磨得非常具体输出质量比一个万能 Agent 稳定得多。这就像做菜的流程洗菜、切配、掌勺、装盘分开每个环节的人手艺都越练越精。收益二并行执行能力。没有依赖关系的子任务可以同时跑。我实测过原来串行要跑 8 分钟的任务拆成三个并行单元后压到 3 分钟。模型 API 调用是现在成本的大头并行虽然不能省成本但能大幅压缩等待时间。收益三故障隔离与重试精细化。某个工作单元挂了编排器只需要重新调度这个单元其他正常单元的产出可以保留。配合超时检查和重试策略整体成功率能提升一大截。收益四成本与模型选型解耦。拆开之后简单的子任务用便宜的小模型跑只有核心推理环节才动用顶级模型。我从全用强模型改成混合规格之后单次复杂任务的成本降了大概四成。1.3 这套思路适合谁参考如果你手上正好遇到这几类情况Fleet 这套思路就值得认真看看任务流程超过五个步骤且每个步骤都需要不同能力单个 Agent 频繁出现做到一半忘了目标多个工具之间需要隔离权限团队里有人维护了一个复杂 Agent 但不敢加新功能。反过来如果你的任务只需要一问一答、调用一个工具或者就是简单的内容改写完全没必要上 Fleet。多一个智能体就多一份消息传递和状态同步的开销杀鸡用牛刀只会让系统更难维护。我的建议是先把单体做到实在做不下去再考虑拆分不要为了架构而架构。2. 架构设计与模式选型2.1 先定拓扑中心化编排还是网状协作智能体 Fleet 的拓扑方案网上讨论最多的是两种中心化编排Orchestrator-Worker和网状协作Peer-to-Peer。中心化编排是最稳妥的起步方案一个编排器负责拆任务、派发、回收结果、处理异常所有工作单元都听它指挥。好处是流程清晰、状态好管理、出错好追踪坏处是编排器可能成为瓶颈而且编排器本身要做很多逻辑判断。网状协作是让智能体之间直接对话各自决定要不要给别的单元派活。听起来很酷实际用起来简直是一场灾难。我试过让几个智能体自由讨论来解决一个复杂问题结果它们在一个细节上礼貌地争论了十几轮既没有收敛也没有产出任何可用结果。大模型生成的对话你根本没法控制方向除非你给非常强的限制但那又回到了变相编排。我的结论很直接第一版 Fleet 一律用中心化编排。把编排器当成一个项目经理工作单元就是执行者所有信息都汇聚到项目经理这里做决策。等系统稳定了再考虑在局部引入一点灵活的协作模式。2.2 通信机制怎么选工作单元之间的通信方式决定了整个系统的可靠性和复杂度。我试过三种方案感受完全不同。第一种是直接 HTTP 回调编排器把任务发给某个工作单元的服务服务处理完再回调编排器。实现最简单但一旦某个单元崩溃消息就丢了没有持久化重试也很麻烦。第二种是消息队列编排器把任务写成消息投递到队列工作单元消费消息处理完把结果写到另一个队列。这样消息有持久化消费者崩了消息也不会丢而且天然支持并发消费。我当时用了一个轻量的基于流式存储的消息队列来做配合消费组机制一套系统非常稳。第三种是共享数据库轮询通过任务表的状态字段流转来同步。适合团队协作场景因为数据库是所有人共同的基础设施但轮询会有延迟而且频繁写状态容易把数据库打成热点。我的选型建议是如果你一个人维护、任务量不大直接点用内存队列加数据库记录最终结果就够了如果要做成多人维护的服务果断上持久化消息队列。通信协议用 JSON 就够了关键是定义一个稳定的消息结构比如包含 task_id、goal、context、expected_output、status 这些字段。2.3 分工策略静态角色还是动态调度工作单元怎么分工我拆成两个方向静态角色和动态调度。静态角色就是你提前定义好有哪些岗位比如规划员、编码员、审查员、执行员每个岗位对应一个固定的模型和一套固定的提示词。编排器拿到目标后按固定流程把任务分配给这些岗位。这种方式简单、可预测、容易调试适合流程相对固定的业务。动态调度是让编排器每次根据任务的实际情况现场决定要拆成几个子任务、每个子任务用什么能力完成。这要求编排器本身有很强的规划能力通常要调用一个更聪明的模型来做动态策划。好处是灵活坏处是结果不可控同一个目标每次拆出来的任务可能完全不一样。我第一版用的是半动态策略流程骨架固定规划 - 生成 - 审查 - 执行但每个环节内部的子任务拆分由编排器动态决定。这样既有稳定的宏观框架又保留了微观层面的灵活性。等你跑熟了再逐步把更多决策交给动态调度。3. 核心细节任务拆解、状态同步与工具治理3.1 任务拆解是Fleet的命根子Fleet 能不能稳定工作七成取决于任务拆解质量。子任务拆得太大单个工作单元压力大容易出错拆得太小消息往返开销大效率反而低。我总结了一个可验证子任务原则每个子任务下发时必须带上明确的输入上下文和预期的输出标准。比如不能只告诉编码单元写一个数据清洗逻辑而要告诉它输入是某张原始表格请输出清洗后的表格字段包含 A、B、C并且空值率要低于 5%。预期输出标准越清晰后续的自动验收就越简单。任务拆解还要注意依赖关系。我第一版吃过亏让两个工作单元并行处理结果其中一个单元的输入依赖另一个单元的输出程序跑完也没等它最后拿到了空数据。后来我在编排器里给每个子任务增加了一个 depend_on 字段调度时先处理没有依赖的任务再处理有依赖的任务问题立刻解决。另一个心得是控制并行度。不要贪多一开始并行 3 个单元就够了。并行太多不仅模型 API 的限流容易触发而且编排器处理回调的逻辑会变得非常复杂出现问题也难以排查。3.2 共享状态与记忆同步Fleet 最大的坑之一是状态不同步。多个工作单元处理同一个目标如果不共享中间状态每个单元都在自己的世界里操作结果根本拼不起来。我采用的方案是双写重要中间产物统一写到文件系统或数据库里消息里只传递引用地址同时把关键决策记录到共享记忆库中。比如规划单元拆完任务把任务清单写成一个结构化文件并把摘要存入记忆库执行单元拿到的是文件路径而不是一截被截断的文本。这一步非常关键——大模型消息传递一旦超过一定长度信息损耗就会明显增加传引用比传内容可靠得多。共享记忆库我用的是向量数据库。每个子任务完成时把产出描述、关键结论、遇到的异常原因都向量化存进去。后续单元如果需要参考历史信息直接做语义检索。这套设计让 Fleet 有了团队记忆不再是各干各的、互相不认识。关于状态命名我有一个惨痛教训一定要定义全局统一的状态枚举比如 pending、running、waiting_dependency、succeeded、failed、retrying。不要让不同单元自己发明状态词。我见过某个单元输出 done、另一个单元输出 complete编排器判断状态时直接懵掉。3.3 工具权限与沙箱隔离Fleet 的能力来自工具调用但工具权限不隔离出事就是大事。单体 Agent 里模型通常能直接访问所有工具一旦提示词注入或者模型输出异常工具就可能被滥用。Fleet 拆开之后每个单元的职责单一工具权限也应当跟着收窄。我给每个工作单元配了一份工具白名单。比如编码单元只能调用代码执行沙箱和文件读写审查单元只能调用文件读取执行单元才能调用外部服务的接口。白名单在编排器下发任务时一并传入工作单元收到消息后先从配置中心加载自己的工具集不在白名单内的工具直接拒绝调用。沙箱隔离一定要做。我的做法是把代码执行放到独立的容器沙箱里限制网络访问、限制文件系统挂载、限制执行时间。不要觉得麻烦某个工作单元如果因为提示词注入开始执行恶意命令没有沙箱的话整个主机都暴露了。实测下来加上超时控制和资源限制之后系统安全性提升了一个量级。4. 实操从零搭一套最小可用的智能体Fleet4.1 最小架构与组件清单我不会一上来就摆一个大而全的架构先给你一套能在一台机器上跑通的最小实现。组件清单如下一个编排器进程负责接收目标、调规划模型拆解任务、投递任务、回收结果、处理重试。三个工作单元规划单元单设一个专门细化子任务、编码执行单元、审查验收单元。先跑一条线后面再横向加。一个消息队列用来在编排器和工作单元之间传递任务与结果。一个共享存储任务产物和状态记录我用的是一个简单的文件目录加一个 JSON 状态文件先不上数据库。一个共享记忆库向量数据库存历史任务摘要。这套架构代理了你未来扩展的骨架工作单元是无状态的可以随时横向多开几个编排器是唯一有状态的核心先单点跑着问题不大。4.2 编排器核心逻辑实现编排器是这个系统的大脑它的核心逻辑其实就三件事拆任务、派任务、收结果。下面这个伪代码是整个编排器最核心的循环class FleetOrchestrator: def __init__(self, queue, store): self.queue queue self.store store self.max_retries 3 def run(self, goal): # 第一步用规划模型拆解目标 plan self.planner.plan(goal) # 第二步按依赖关系分批入队 ready_tasks self.resolve_dependencies(plan) for task in ready_tasks: self.queue.push(task) # 第三步循环回收结果直到全部完成 while not self.is_all_done(plan): result self.queue.wait_result() if result.status failed: self.handle_retry(result) else: self.store.save(result) def handle_retry(self, result): if result.retry_count self.max_retries: self.mark_failed(result) else: self.queue.push(self.build_retry_task(result))每一步逻辑都不复杂但要注意几个实现细节。resolve_dependencies 是关键。任务之间如果有依赖必须先让所有无依赖任务先跑等依赖完成后再放行后续任务。我第一版偷懒直接按列表顺序入队结果两个任务互相等队列卡死整个 Fleet 陷入死锁。后来改成按依赖层级分批入队再配合超时检查才算稳定下来。wait_result 不要用简单的死循环轮询。我在实现中用了消息队列的阻塞消费设置了 30 秒超时超时后做一次全量状态体检发现卡死任务就走重试逻辑。这样可以避免一个任务卡住导致整条链路都僵住。4.3 工作单元的实现要点工作单元本质上就是一个提示词 模型调用 工具集合的封装。它收到消息后做四件事加载上下文、调用模型生成动作、执行动作、返回结构化结果。我用一个统一的 Worker 基类来写所有单元class BaseWorker: def __init__(self, model, tools, queue): self.model model self.tools tools self.queue queue def handle_message(self, msg): context self.load_context(msg.context_ref) action self.model.decide( system_promptself.PROMPT, user_contentmsg.goal, contextcontext, toolsself.tools ) output self.tools.execute(action) result { task_id: msg.task_id, status: succeeded if output.valid else failed, output_ref: self.save_output(output), summary: output.summary } self.queue.report(result)每个子类只需要重写 PROMPT 和 tool 白名单。比如编码执行单元的 PROMPT 要求必须给出代码和测试用例审查单元的 PROMPT 要求只能检查逻辑漏洞不得修改代码。这个模式让扩展新单元非常快很多时候只是多写一段提示词的事。模型输出格式一定要做解析容错。我在项目中吃过亏模型偶尔会在 JSON 输出前后追加解释文字导致解析失败。后来我在每个工作单元里加了提取首个合法 JSON 块的逻辑并且把解析失败也纳入重试条件——不要一失败就丢给人工换一次生成往往就好了。4.4 联调验证与参数调优整套系统搭好后先用一个最简单的小目标联调比如把某目录下的日志文件按级别统计数量并生成报告。这能串起规划、执行、审查、结果回收的完整链路。联调通过后再压测这时候要关注几个关键参数。第一个参数是超时时间。子任务类型不同执行时间差异很大。我给不同类型的任务设置了不同的超时生成类任务 60 秒代码执行类任务 120 秒外部接口调用类任务 30 秒。超时时间设太短容易误杀长任务设太长则整体效率受影响。第二个参数是重试次数。我默认重试 2 次重试 3 次以上性价比就很低了。每次重试时要带着上一次失败的原因摘要重新生成让工作单元知道上次错在哪。盲目原样重发只会得到同样的失败结果。第三个参数是并行度上限。建议从 2 开始逐个往上加同时观察模型 API 的响应时间和错误率。如果错误率开始飙升说明并行度已经超过实际承载能力需要往回调。5. 常见问题与排查技巧实录5.1 高频问题速查表我把实际运行中遇到的典型问题整理成了一个速查表排查的时候对着这张表能省出大量时间。问题现象根本原因解决方案子任务一直处于 pending 状态依赖任务没有完成或队列消费异常检查依赖关系配置在编排器中输出全量任务依赖图两个任务互相等待整个 Fleet 卡死任务依赖出现环在规划环节做环检测发现成环立即重新规划单元返回结果格式解析失败模型在 JSON 前后附加了文本增加提取首个合法 JSON 块解析逻辑纳入重试条件多个单元拿到的是过期状态共享状态没有加版本号所有状态写入带上版本号读取时校验收敛工具调用被莫名拒绝工作单元工具白名单未更新把白名单配置放到统一配置中心不要在单元内硬编码重试导致外部接口重复调用重试时没有携带幂等键所有外部调用携带 task_id 作为幂等键接口做去重排查问题的基本思路是先看消息队列里消息的状态和堆积情况再看编排器日志里任务的流转记录最后看工作单元的输出。大部分问题都能在这三步里定位不要一上来就怀疑模型能力。5.2 我踩过的三个坑第一个坑是共享记忆库的数据噪声。最初我把工作单元的中间思考过程全都存进向量库结果检索时返回的大多是无关信息反而污染了后续决策。后来我改成只存结论、不存过程每个子任务完成后由编排器调用一个小模型把产出摘要压缩成两三句话再入库。检索质量明显提升。第二个坑是模型选择一刀切。我刚开始所有工作单元都用同一个顶级模型成本高且没有必要。后来我把工作单元按难度分级规划单元和审查单元用强模型执行单元和格式处理单元用便宜的小模型。同样的任务质量几乎没下降成本按之前的账本统计降了四成左右。第三个坑是日志不统一。Fleet 的日志散落在编排器、工作单元、消息队列三个地方排查问题时要不停切换窗口效率极低。后来我统一了日志格式每条日志都带 task_id、unit_name、timestamp、level全部汇到一个集中日志里按 task_id 一搜就能拉出整个任务的生命周期。这个改动看着不起眼却让后面的排障效率提高了一倍不止。另外补充一个排查技巧当某个子任务连续失败多次时不要一遍遍自动重试先把失败的原始消息体完整 dump 下来人工看一次。很多问题其实是规划阶段给出的预期输出标准本身就不合理工作单元被一个错误的 KPI 逼着反复试错。这种问题靠重试解决不了只能回过去改规划逻辑。回到开头那个问题Fleet 到底值不值得搭我的体会是如果你的业务里已经出现了一个 Agent 实在撑不住的信号那就别再硬扛了。Fleet 不是炫技而是一种更接近真实团队协作的工作方式。它带来的代价是消息流转和状态管理的复杂度但只要你把任务拆解和状态同步这两个基本功做扎实这套复杂度是完全可控的。我现在已经在这套架构上继续叠加了更细粒度的路由策略和更自动化的任务验收逻辑每一步跑稳了再往下走整体推进反而比当初单打独斗时快得多。