★ FEATURED ARTICLE “智能体 Fleet”这个词我第一次从某团队搭建行情监控系统时听到。当时他们要处理十几路数据源还要做实时盯盘、自动提醒、盘后汇总单靠一个Agent根本扛不住——不是能力不够是编排复杂度上去了之后单体会失控。最后我们落地的形态不是一个大模型而是一组小型智能体组成的集群各有分工、互相校验像舰队一样协同工作。我把它命名为Fleet这套思路至今仍是我做Agent工程的默认起点。Fleet能解决一个很实际的问题当你需要让AI在真实业务里持续干活而不是聊一句答一句时怎么让任务拆解、并行执行、结果汇总、异常重试这些环节可控、可靠、可观测。它不同于LangChain里那种线性链式调用也不同于AutoGPT那种完全让模型自由发挥的“放手模式”而是介于两者之间——由少量规则锚定边界把灵活性留给模型把确定性留给调度层。这篇内容适合两类人一类是做Agent落地时被“任务总是跑飞”“结果不可复现”困扰的工程师另一类是刚入门多Agent编排、想找个可以直接抄作业架构的开发者。1. 为什么是“舰队”而不是“单体”智能体集群的设计思路拆解1.1 单体Agent的三重困境先说说我们当时走过的弯路。第一版方案很简单用一个大模型Agent把所有任务都塞给它——拉取行情、分析数据、生成摘要、发送告警全在一个循环里完成。上线第一天就有问题单个Agent的上下文窗口有限任务一多前面的指令就会被“挤”出记忆模型开始忘事。比如让它拉完数据再分析等分析的时候它已经把拉数据的格式要求忘了输出就变成了一堆格式错乱的JSON。这还不是最要命的。单体Agent的执行路径是暗盒你只知道它开始执行了、结束了中间调了哪些工具、为什么跳过某个步骤、某个分支为什么走了A没走B完全没有记录。出问题的时候你对着日志干瞪眼因为它“思考”的过程只存在于模型的概率空间里没法回放也没法定向修复。这就是第一重困境不可观测。第二重是故障放大。单体Agent只要有一个步骤抛异常例如某个数据源超时整个链路的后续动作全部阻塞。而大模型的输出还有概率性同一个Prompt跑五次第二次可能就换了个工具调用方式于是同样的输入产生不同的行为。你在测试环境里调得好好的上了生产一个时间字段格式变了整套流程就稀碎。第三重是并行能力缺失。行情数据有七八路来源如果逐条串行处理一轮下来黄花菜都凉了。单体Agent架构里并行需要你自己写协程、自己管理并发、自己处理结果合并这些代码逻辑一多很快就失控。你会发现大模型Agent本身不是一个可靠的调度器——它擅长的是理解与生成而不是精确的任务编排。1.2 舰队模型的核心思想分散执行、统一指挥“舰队”的思路很简单不要试图让一个智能体干所有事而是拆成多个专职智能体让它们并行工作由一个轻量级的调度中心负责分派、监控、回收结果。每艘“舰”只干一件事且只干得精——拉数据的只拉数据写摘要的只写摘要发告警的只发告警。这个设计有一个非常重要的底层逻辑把“会出错的智能部分”限制在最小范围内把“不能出错的编排部分”用确定性代码实现。模型负责理解指令、撰写内容、选择策略这些是它的强项调度中心负责分配任务、检查超时、重试失败、聚合结果这些是纯逻辑用代码写死绝不让模型插手。这样一来模型再怎么“发挥”影响范围也只是在自己的那艘舰上——一艘舰跑偏了不会影响整个舰队。我理想中的Fleet有一个三层结构调度层指挥官接收原始任务拆分子任务维护任务队列分发给对应智能体回收结果汇总输出。执行层舰船各专职智能体并行运行每个都有独立上下文、独立工具集、独立Prompt模板只处理自己领域内的子任务。护栏层护航编队超时熔断、重试策略、结果格式校验、权限边界检查。这一层通常由代码实现不调用模型。最后落地时调度层用了消息队列这里我用的是一个轻量级的Redis Stream执行层用独立进程/线程池跑护栏层嵌入在调度循环里。整个Fleet更像一个“插槽式”架构新增一个智能体只需实现统一的接口然后向调度中心注册不需要改动其他任何部分。这就是为什么项目叫“Fleet”——它是一个可以持续纳编、扩展的舰队而不只是几个Agent的拼凑。2. 核心智能体的行为定义与“系统黑话”设计2.1 先定消息契约再写智能体多智能体系统里第一件要做的事不是写Prompt而是定义消息格式。Fleet里每个智能体之间不直接对话所有通信都通过调度中心的“任务单”流转。任务单是一个JSON对象包含任务ID、类型、输入数据、状态、回调地址、TTL等字段。这个设计借鉴了微服务里的消息契约概念——智能体之间不存在隐式依赖只认字段。我定义的消息头长这样{ task_id: c8f9e2a1-7b3d-4f6e-9a2c-3d5e7f8a9b1c, task_type: fetch_market_data, source: fleet-scheduler, target: agent-market-feeder, payload: { symbol: BTC-USDT, time_range: 2025-01-01~2025-01-07, data_format: candlestick_1h }, deadline_ms: 15000, retry_count: 0, max_retries: 3, callback: fleet://aggregator/main }看到这个结构你应该能理解我的思路payload是各舰船关心的业务数据deadline_ms是超时控制retry_count和max_retries是重试状态callback是结果回收的地址。每个智能体拿到任务单之后先校验字段是否完整再执行自己的专职逻辑最后把结果写回一个统一的result通道。这个“校验—执行—回写”三步走让每个智能体变得非常单纯——它不需要知道整个舰队在干什么只需要知道自己的任务是什么、结果交到哪里。这里有一个关键心得任务单里千万不要放模糊的自然语言指令。早期版本里我直接在payload里写“请分析最近七天的行情走势并给出建议”结果不同智能体对“最近七天”的理解不一致——有的按自然日有的按交易日有的按截稿时间往前推168小时。后来我把所有业务参数都结构化提取出来模型只负责在结构化字段之上做推理不再自己“理解”任务边界。这个改动直接把系统出错率砍掉了一半。2.2 智能体接口每个舰船都遵循同一个协议每个专职智能体在Fleet里都是同一个接口的实现这样可以保证调度中心统一管理。接口的核心是四个方法init()加载配置与Prompt模板、execute(task_json)执行任务并返回结构化结果、health_check()上报自身运行状态、shutdown()优雅退出。所有智能体跑在同一套基座上彼此不感知对方的存在。举个例子行情拉取智能体的execute逻辑大概是接任务单、解析参数、请求外部数据源、清洗字段、格式化为统一schema、写结果通道。整个流程不涉及任何“智能”——它就是一个带网络请求能力的脚本。但它的价值在于当数据源变动、字段改名时只需要改这一艘舰船调度层和其他智能体完全无感。在真实项目中我就遇到过行情源把金额字段从“price”改成“last_price”的事当时只改了这一个智能体的映射逻辑其他十几个智能体零改动系统照跑。这就是拆分的红利。而那些需要模型能力的智能体比如摘要生成、异常识别、策略建议它们同样接的是结构化输入只负责“理解与生成”。模型被隔离在专职位置不会被调度逻辑干扰也不会产生游离于系统之外的副作用。我经常说多Agent系统里最好的Prompt就是不用Prompt——把一切能用代码确定的逻辑都从模型职责里剥离出去模型只做它不可替代的那一部分。3. 实操Fleet调度器的实现与智能体注册机制3.1 调度器的基础形态队列加任务表调度器是整个Fleet的心脏。一个合格的调度器至少要完成四件事接收任务、拆分任务、派发任务、回收结果。这里我不引入重量级框架只用Python标准库加Redis Stream就能搭出一个可用的版本。选择Redis Stream而不是直接用RabbitMQ是因为轻——一个列表结构就能承载消息且天然支持消费者组的负载均衡、消息确认、Pending列表。调度循环的主逻辑如下import redis import json import uuid import time from concurrent.futures import ThreadPoolExecutor class FleetScheduler: def __init__(self, redis_client, registry): self.redis redis_client self.registry registry # {task_type: agent_id} self.executor ThreadPoolExecutor(max_workers16) self.stream fleet:task_stream self.result_stream fleet:result_stream def submit_task(self, task_type, payload, deadline_ms15000): task { task_id: str(uuid.uuid4()), task_type: task_type, payload: payload, status: pending, created_at: time.time(), deadline_ms: deadline_ms, retry_count: 0, max_retries: 3, } self.redis.xadd(self.stream, task) return task[task_id] def dispatch_loop(self): while True: # 从Stream读取新任务block1表示最长阻塞1秒 entries self.redis.xread({self.stream: }, count10, block1000) for stream_name, messages in entries: for msg_id, msg in messages: task msg agent_id self.registry.get(task[task_type]) if not agent_id: self.handle_unknown_task(task) continue self.executor.submit(self.execute_on_agent, agent_id, task) def execute_on_agent(self, agent_id, task): task[status] running try: result self.registry.execute(agent_id, task) task[status] success task[result] result self.redis.xadd(self.result_stream, task) except Exception as e: task[status] failed task[error] str(e) if task[retry_count] task[max_retries]: task[retry_count] 1 time.sleep(2 ** task[retry_count]) # 指数退避 self.redis.xadd(self.stream, task) # 重新入队 else: self.redis.xadd(self.result_stream, task) # 记录失败这段代码大约百来行但把Fleet最关键的机制都涵盖了任务入队、消费者读取、派发到对应智能体、结果回写、失败重试、指数退避。我没有用复杂的分布式事务因为消息在Redis Stream里天然持久化任务状态通过status字段标记即可。上限是丢失消息会被重复消费——但这在Agent场景里是可接受的我会在下一节详细说幂等设计。3.2 智能体注册协议让扩展像插U盘一样简单真正让Fleet“可生长”的是它的注册机制。某种意义上看调度器不直接依赖任何具体智能体实现只依赖一个注册表这张注册表把task_type映射到具体的执行函数或远程服务地址。添加一个新能力本质上就是向注册表里新增一条映射记录。我给注册表设计的接口是一个SimpleRegistry类支持本地函数执行和远程HTTP调用两种模式。本地模式用于同进程内的轻量智能体远程模式用于需要独立部署、独立扩缩容的重型服务。注册一个智能体的代码大概长这样class SimpleRegistry: def __init__(self): self._handlers {} def register(self, task_type, handler, modelocal, endpointNone): self._handlers[task_type] { mode: mode, handler: handler, endpoint: endpoint } def execute(self, task_type, task): entry self._handlers[task_type] if entry[mode] local: return entry[handler](task) elif entry[mode] http: resp requests.post(entry[endpoint], jsontask, timeout10) resp.raise_for_status() return resp.json() else: raise ValueError(funknown mode: {entry[mode]}) # 使用示例 registry SimpleRegistry() registry.register(fetch_market_data, fetch_market_data_agent, modelocal) registry.register(generate_summary, summary_agent_execute, modehttp, endpointhttp://localhost:9002/execute)调用registry.register()就像把一艘新舰船编入舰队——传入一个task_type、一个处理函数或远程端点调度中心立刻就能把该类型的任务派发给它。新加入的智能体不需要知道系统里其他舰船的存在也不需要修改调度器代码。我在实际项目里加新数据源时最快的一次只花了一个小时写一个拉取函数注册一行跑测试完事。这里要特别提醒注册表必须是程序启动时构建的静态配置不要做成运行时动态修改。早期版本我为了“灵活”支持在运行时通过API动态注册任务类型结果某个上游服务异常时把自己的异常数据当成任务类型注册了进来导致调度器路由错乱。后来改成只能用配置文件或启动参数注册运行期间注册表是只读的再也没出过这类事故。3.3 任务拆分从“一句话需求”到“可派发任务”调度器要处理的原始输入往往不是结构化任务而是一段自然语言需求比如“盯一下最近三天的大盘行情如果出现明显异动就写个简报发给我”。这一步需要一道“理解工序”——由一个拆解智能体负责把自然语言转成标准化的子任务列表。拆解逻辑我建议写成“意图识别加参数抽取”的组合。意图识别决定是否要找行情数据、要不要分析、要不要生成报告、要不要告警参数抽取则把“最近三天”“大盘”“异动”等要素转为结构化字段。这部分可以用大模型完成但Prompt里必须给出严格的JSON schema约束并且要求模型只输出JSON片段不输出任何解释性文字。我在生产环境里的Prompt模板大致如下SUMMARY_PROMPT 你是一个任务拆解器。把用户输入的意图解析为结构化任务列表。 只输出JSON不要任何解释文字。JSON格式如下 { tasks: [ {task_type: fetch_market_data, payload: {symbol: ..., time_range: ...}}, {task_type: detect_anomaly, payload: {metric: ..., threshold: ...}} ] } 用户需求{user_request} 请确保task_type必须在以下类型中{supported_types} 这一步是整个流水线里唯一允许“模糊理解”的环节原因很简单用户需求天然是模糊的必须有一个环节把它固化成机器可执行的计划。一旦它输出了结构化任务列表后续所有环节都不再依赖大模型做“计划”而是依赖确定性的调度逻辑做“执行”。这能防止模型在后续环节里自由发挥——拆解阶段的自由是整个系统允许的全部自由。4. 监控、重试与健康检查让Fleet在无人值守时也能跑稳4.1 幂等设计重复执行不产生脏数据当我在前一节说“消息可能被重复消费”时你可能会担心。这个问题的标准解法叫幂等设计——同一个任务无论执行多少次结果都必须一致。对Fleet来说最重要的是在任务单里带一个唯一的task_id并在执行端记录已处理过的task_id列表。具体做法每个智能体在执行前先查询本地或Redis里的task_id记录用SETNX命令如果任务ID已存在直接返回已缓存的结果不重复执行业务逻辑。这就像快递员送包裹——同一个运单号如果系统里已经标记“已签收”第二次配送就直接返回成功不会多送一次。针对行情数据这种“可再生”的数据幂等尤其重要否则重复拉取不仅浪费资源还可能在后续统计里产生重复记录。实现上我在调度器的execute_on_agent里加了一段逻辑# 幂等检查在Redis里用task_id做唯一键 dedup_key ffleet:task_result:{task[task_id]} done redis_client.get(dedup_key) if done: # 已经执行过直接回写结果 task[status] success task[result] json.loads(done) redis_client.xadd(self.result_stream, task) return # 用SETNX抢占任务执行权防止同一任务被多个worker同时执行 acquired redis_client.set(dedup_key, running, nxTrue, ex3600) if not acquired: return这个改动虽然只有几行却解决了分布式环境下的两个典型问题重复执行和重复写入。在多worker部署时同一任务可能被多个消费者同时读到没有这层保护系统就有概率产生重复的告警、重复的汇总报告。4.2 超时熔断与指数退避别让慢舰拖垮全舰队真实场景里外部数据源偶尔会慢。如果某个智能体阻塞在HTTP请求上迟迟不返回调度器的线程就会被占着不放新任务排不上队整个舰队都跟着减速。所以我给execute_on_agent里的本地函数执行加了超时控制用ThreadPoolExecutor的future对象配合as_completed实现from concurrent.futures import as_completed def execute_with_timeout(self, agent_id, task, timeout15): future self.executor.submit(self.registry.execute, agent_id, task) try: result future.result(timeouttimeout) return result except TimeoutError: future.cancel() raise TimeoutError(fagent {agent_id} timed out)超时时间内没完成的智能体结果直接丢弃任务标记为失败进入重试流程。重试不是盲目重复而是用指数退避第一次失败后等2秒第二次等4秒第三次等8秒。这样做的目的是给临时故障一个恢复窗口——如果外部数据源只是瞬时抖动退避之后大概率就恢复了如果确实挂了连续快速重试只会雪上加霜。健康检查也很重要。我在注册表里让每个智能体实现一个health_check方法调度器每30秒轮询一次判断各舰船是否活着。响应时间超过阈值的智能体标记为degraded这时候调度器会把发给它的任务暂时路由到备用实现如果配置了或者直接排队等待。这个机制在远程HTTP智能体上体现得尤其明显——某端点是Java服务打包上线后第一次调用要加载依赖耗时可能5秒如果不做预热第一次任务必然超时。后来我在智能体启动时加了一个warmup接口提前加载依赖问题就消失了。4.3 日志聚合与全链路追踪定位问题要到“舰”当系统里有十几个智能体并行跑的时候出了错第一反应是“到底哪艘舰出的问题”。我的做法是所有子任务的执行日志都统一收集到中央日志系统每一行日志都带上task_id和agent_id两个标签。这样不论是查错误还是复盘流程都可以按task_id把所有相关日志拉成一条链路。在设计初期我吃过亏——每个智能体自己打印日志结果不同进程的日志混在一起完全分不清顺序。后来我在消息格式化里强制要求任务流入、开始执行、调用外部服务、返回结果、结果回写、异常抛出这六个节点必须各打点一次且带上同一个task_id。这样粘起来一条任务的完整生命周期就一目了然。异常排查的效率提升了不止一倍——不用再靠猜直接看链路断在哪里就知道是哪艘舰、哪个环节出了问题。5. 常见问题与排查技巧实录5.1 问题速查表从症状到根因这一节分享几个在真实部署中遇到的问题整理成一张速查表方便你在自己的Fleet里对照排查。现象可能原因排查思路修复建议任务只被一台worker消费其他worker闲置Redis Stream消费者组没配置group使用了普通xread确认消费代码是用xreadgroup还是xread统计各worker消费数量切换到xreadgroup启用消费者组模式开启消息确认机制同一任务被执行多次结果重复缺少幂等控制或worker崩溃前已处理但未确认查Redis里的task_id记录前缀看是否有重复处理检查ack确认时机在executor入口加SETNX幂等锁先在业务完成后确认消息而不是读取即确认某个智能体频繁超时影响其他任务线程数配置过小或该智能体的外部依赖响应慢看监控面板的线程占用定位是排队还是外部慢为慢智能体单独加大超时阈值或者把它分离到独立部署避免挤占公共线程池重试一直失败系统消息积压重试策略没跟上具体场景外部依赖确实已经不可恢复看失败任务的error字段是否一致检查是真实故障还是Promop配置错误增加死信队列重试多次后自动隔离该任务类型人工介入排查模型输出不符合JSON schema无法解析Prompt约束不够严格或者模型温度设置过高复现失败样本观察模型返回的原始内容在智能体侧加强JSON解析兜底逻辑降低温度增加一次JSON校验与修复步骤调度中心重启后消息丢失任务还在处理中时进程被杀未ack的消息被Redis清掉检查Redis持久化配置确认是否在杀掉进程前执行了shutdown钩子给调度器增加优雅退出逻辑Redis开启AOF持久化消费者组用Pending列表做恢复5.2 一个印象深刻的夜晚模型把任务ID写成了“不存在的键”那一次排查让我长记性。某天夜里系统突然开始疯狂重试“fetch_market_data”任务日志里全是“agent执行失败字段symbol不存在”。我盯着日志看了半天发现错误不是外部数据源返回异常而是payload里根本没有symbol键取而代之的是一个叫“不存在的键”的字符串。后来定位到拆解智能体的输出——模型想把“用户说不用管币种”这句话当成参数写进了payloadjson解析还成功了。这个问题的根源不在模型而在拆解阶段的schema约束不够严。修复方案有三层第一层拆解Prompt加入“未提取到参数时使用null不要自行编写”的硬规则第二层在拆解智能体的输出后增加一个schema校验环节发现未知字段直接丢弃第三层在任务入队前做一次必填字段检查symbol缺失的任务直接打回拆解层重新生成。这三层之后同类问题再也没出现。我把它写进项目文档里当作一个经典案例——永远不要相信模型会严格遵守格式你要在代码里校验模型的输出而不是在Prompt里祈祷。5.3 监控面板该盯哪三个数搭建Fleet的监控时我不建议一开始就上全套指标。先盯三个数任务积压量、失败率、平均执行时长。任务积压量反映调度层的负载能力如果这个数一直往上走说明消费者消费速度跟不上生产速度就要扩容worker或者优化智能体执行效率。失败率是整体健康度的晴雨表超过阈值就该触发告警。平均执行时长则帮助你感知智能体是否变慢——同一个任务类型平均耗时比基线多了一倍通常意味着外部依赖出了异常或者Prompt加了没必要的长输出。这三个数我全部用Redis里的计数器加定时扫描实现没有引入额外的监控系统。每五分钟把三个指标写一次日志推送一次摘要。等团队规模大了再考虑上完整的可观测性平台。对小规模Fleet而言花哨的监控面板远不如这三个数字管用。6. 写在最后的工具选型与后续扩展建议整个Fleet的底座选型我推荐Python 3.10加Redis 7再加一个轻量级的消息队列协议。如果团队已经维护了RabbitMQ或Kafka完全可以直接切换核心架构不变。Python的优势是生态全、写AI相关代码顺手缺点是线程模型受限如果任务量到了每秒几百单以上可以考虑把调度中心用Go重写智能体层仍然保留Python服务。就我看到的情况绝大多数业务场景的瓶颈不在调度器本身而在智能体执行的外部调用——网络等待、数据源响应、模型推理这些才是消耗时间的大头架构选型反而不需要过度纠结。关于后续扩展我最推荐的方向是给Fleet增加一个“观察者智能体”——它不参与业务执行只看其他智能体的运行日志和结果质量发现异常模式时主动向调度中心提出调整建议。比如它发现“summary生成智能体”连续三次输出的报告里日期格式错了就可以自动触发一次Prompt版本切换甚至发起回滚。这本质上是用Agent去管理Agent也是我下一阶段最想落地的事情。个人体会是构建Fleet最难的部分从来不是写代码而是克制——克制每个智能体“多做一点”的冲动克制调度器“多用模型”的冲动克制系统“把所有环节都智能化”的冲动。多Agent系统的稳定性恰恰来自对智能的边界保持清醒。每次你想让某个环节再“聪明”一点的时候先问问自己这一步能不能用确定性代码解决如果能就别让模型进场。守住这条线Fleet会越跑越稳越扩越顺。
阅读完成 · 觉得有帮助？
