拓冰网站定制 
1. 一次单Agent翻车让我开始认真研究多Agent协作1.1 我遇到的问题复杂任务在单模型会话里越滚越乱这件事说起来有点丢人但确实是我转做AI应用开发以来印象最深的一次失败。当时接了一个内部需求让大模型自动完成一份竞品分析报告要求从资料收集、数据提取、维度拆解、观点生成到最终排版一条龙跑完。我第一反应是写一个超长Prompt把步骤、角色、输出格式全部塞进去然后一次性调大模型。试了几轮结果惨不忍睹。前几轮回答还算正常但任务一长模型就开始“失忆”——前面说要分析定价策略后面总结的时候又变成了渠道策略中间偶尔自己造几个不存在的竞品数据我还得人工去核。更难受的是一旦中途某一个环节的结果不理想整个任务就废了必须从头再跑一遍。而且我根本定位不到是哪一步出的问题。整个对话上下文越滚越长到最后一次请求可能要传几万token的聊天记录成本和延迟都上来了答案反而更差。后来我读了LangGraph、AutoGen、CrewAI这些框架的源码和相关实践材料才意识到问题的根源我用一个Agent包办了所有事本质上是在让单个模型会话承担完整的业务流水线。这个思路在任务路径短、交互少的场景下没问题但一旦任务复杂起来就会同时踩中上下文污染、角色冲突、错误扩散三个坑。1.2 上下文污染、职责混乱、错误扩散三个坑一起踩先说上下文污染。单Agent模式下所有历史信息都存在同一个上下文中。竞品分析这种任务资料收集阶段的噪声数据会一路传递到观点生成阶段早期的小错误会被后期的推理当作“事实”引用。模型本身又没有一个“忘记机制”你只能靠截断上下文来控制但截断之后早期关键信息也没了顾此失彼。第二个是职责混乱。同一个模型开头让它当一个调研员中间让它当一个数据分析师最后让它当一个报告写手。模型确实能做到一定的角色切换但每次切换都是一次隐性的重新校准。而且在这种长流程里它经常分不清“我现在应该专注于哪个目标”于是输出里就开始出现身份漂移——上一段还在做总结下一段又开始假装自己是调研员补了一堆原始资料。第三个是错误扩散。单Agent流程是线性的一步错后面每一步都建立在这个错误之上。更麻烦的是模型不会自动纠偏它只会顺着已经生成的思路继续往下编。你想事后问责结果发现所有输出都混在一个会话里根本分不清是哪一步、由哪个决策引入的错误。那一次之后我做的第一件事就是把“一个超长Prompt跑到底”的方案彻底丢掉开始研究怎么把任务拆给多个模型会话去协作。也就是今天要聊的大模型多Agent系统。1.3 多Agent不是把Prompt拆开而是把系统拆开很多人刚接触多Agent容易有一个误解多Agent就是把一个长Prompt拆成好几个短Prompt每个负责一段再串起来。这确实能解决一部分上下文长度问题但真正的多Agent协作核心不在于“切分”而在于重新设计系统的分工与交互方式。你真正要做的是把一个完整的任务拆成多个可以被独立验证、独立重试、独立扩展的子任务然后交给不同的Agent去执行再由一个调度机制把结果组合起来。每个Agent拥有自己的上下文自己的工具集自己的输出格式甚至自己的兜底策略。这就像开一家餐厅。单Agent模式是一个全能厨师从头做到尾从买菜到配菜到炒菜到摆盘全包客人一多必乱。多Agent模式是后厨分工采购员只管备料砧板师傅只管切配炉头师傅只管炒传菜员只管上桌。每道工序有明确的责任边界出了错可以精确定位到人、到步骤整个系统稳定性一下子就有了质的提升。换句话说多Agent最核心的能力不是“让多个模型一起跑”而是“让多个模型在一条清晰的生产线上各司其职”。协作架构是这条线的骨架任务调度是它的脉搏。2. 协作架构的本质先搞清楚“怎么分活”再决定“怎么干活”2.1 四种主流协作拓扑星型、管道、网状、分层在动手写代码之前你得先建立一张架构地图。我看了大量资料和开源项目之后把多Agent的协作架构归纳成四种主流拓扑理解这四种后面选型就有谱了。星型拓扑也叫中心化编排所有的Agent都围绕一个中心节点工作。中心节点负责任务拆解、分配、收集结果其他Agent只和中心通信。这是最常用的一种实现简单控制力强适合大多数业务场景。缺点呢中心节点容易成为瓶颈如果编排逻辑设计得不好所有压力都堆在它身上。管道拓扑是让Agent排成一串每个Agent只和上下游通信A的输出作为B的输入B的输出作为C的输入。这种结构非常适合内容加工类任务比如“清洗数据 - 提取特征 - 生成报告”。它的好处是链路清晰每个环节可以独立替换升级坏处是整体延迟等于所有环节延迟之和而且中间任何一个环节挂了整条线都得停。网状拓扑里每个Agent都能和其他Agent自由通信没有中心节点。这种架构最灵活但也最容易失控——没有中心约束Agent之间来回传递消息很容易出现“谁都在说话但没人对结果负责”的局面。我只在探索性、低风险的小规模实验里用过网状结构生产环境基本不建议。分层拓扑是大型系统最常见的做法顶层有一个“管理者”Agent中层有几个“小组长”底层是一批执行Agent。每一级只需要关心自己的上下级不需要了解全局。这种架构扩展性最好适合团队规模大、任务类型杂的系统。我画了一张对比表方便你直观理解拓扑类型优势劣势适用场景星型实现简单编排可控中心节点压力大中小规模任务、流程可控的系统管道链路清晰易于优化单点延迟叠加单点故障影响全局有严格先后顺序的内容流水线网状灵活性最高通信混乱结果不可控低风险探索性实验不建议生产使用分层扩展性好权责清晰设计复杂度高层级间通信成本大大型系统、多团队任务协同2.2 中心编排Orchestrator-Workers为什么是大多数场景的默认解我在生产项目里用得最多的还是星型拓扑的变体一般叫Orchestrator-Workers模式也就是中心编排模式。原因很朴素它能用可控的复杂度换来最高的系统可控性。在这个模式里Orchestrator是关键角色。它的职责不是自己干活而是“派活”把用户的目标解析成子任务把子任务分配给对应的Worker收集Worker的返回结果然后决定是继续下一个任务、重试失败任务、还是把结果汇总返回给用户。Worker则是专注的执行者每个Worker只负责一种技能。比如在企业知识库问答系统里我会拆出检索Worker、摘要Worker、推理Worker、格式整理Worker。每个Worker的Prompt非常聚焦上下文也很短基本上不会出现角色漂移。这套结构的最大收益在于故障隔离如果某个Worker的表现不行你只需要替换它一个其他组件完全不受影响。这一点在实际维护中价值极大。我之前维护过一个客服工单分类系统刚开始把分类和优先级判断塞给同一个Agent做结果发现分类做得好但优先级判断很离谱。改成两个Worker之后分类Worker我加了一条规则就修好了优先级Worker的做法则是换了一种思路去调整互相不干扰。还有一个经常被忽略的好处便于并发。Orchestrator收到任务后如果发现多个子任务相互独立就可以并行发给多个Worker整体耗时能压缩不少。这一点在后面的任务调度部分还会详细说。2.3 选型对照什么时候用Pipeline什么时候用Plan-Execute说完了星型再聊两个容易被混为一谈的概念Pipeline模式和Plan-Execute模式。Pipeline模式就是管道拓扑的工程化落地。它的特点是任务路径在任务开始前就已经固定了。比如“上传文档 - 文档解析 - 信息抽取 - 存档”这是一条固定的流水线每个节点做什么是一开始就确定好的。它适合流程相对固化、业务逻辑不会频繁变化的场景优点是稳定、可预测、调优容易。Plan-Execute模式则复杂一些。系统先由一个Planner Agent根据用户目标生成执行计划这个计划不是写死的而是模型根据具体情况动态生成的。然后由执行器按计划逐步执行每步执行完后系统会检查结果再决定下一步。我个人的经验是如果你的业务场景“流程固定、步骤有限”直接用Pipeline简单可靠别硬上Plan-Execute。但如果你面对的是开放式任务比如“帮我把这个GitHub仓库研究透输出一份技术选型报告”这种任务没有固定路径Plan-Execute才是合适的方案。很多人在第一阶段做项目时特别喜欢Plan-Execute因为看起来“高级”。但从成本角度说Plan-Execute需要更多的模型调用、更复杂的错误处理单位任务的token消耗至少是Pipeline的两到三倍。没必要为不需要灵活性的场景付出这个代价。2.4 协作模式与任务类型的匹配关系我把常见任务类型和推荐模式做了个匹配这套逻辑我几乎每个项目都会复用。任务类型典型例子推荐模式原因固定流程型工单分发、文档解析、定时报表Pipeline流程确定管道最稳定开放探索型竞品调研、代码库分析、方案设计Plan-Execute路径未知需要动态规划并行计算型多语言翻译、多文件分析Orchestrator-Workers可并发分发整体耗时最短大型复杂项目新产品研发、跨模块协同分层管理任务多、角色杂、需逐层分解需要逐步推理验证数学证明、代码调试ReAct循环需要边执行边反思纠错这套匹配逻辑的核心出发点只有一个不要让架构的复杂度超过任务本身需要的复杂度。能用Pipeline解决的事就不要上Plan-Execute。能用一个Agent搞定的事就绝对不要强行拆成三个。3. 任务调度从Plan到Tick的核心链路3.1 任务分解怎么把一个大目标拆成可验证的子任务任务调度是整个多Agent系统里最容易被低估的一环。很多系统的瓶颈不在模型能力而在调度设计。任务调度的第一步是任务分解。如果这一步做得不好后面所有环节都会跟着乱。我自己有一个分解原则叫“子任务可验证性原则”拆出来的每一个子任务必须有一个明确的、不依赖模型主观判断的完成标准。打个比方你让一个Agent“研究一下新能源汽车市场”这个任务就没法验证。但如果你拆成“收集2024年全球新能源汽车销量Top10品牌的数据输出为表格字段包含品牌、销量、同比变化”那这个任务就是可验证的——你可以拿输出对照原始数据源马上判断它对不对。在实际操作中我会让Planner Agent先输出一个JSON结构里面包含任务ID、任务描述、依赖关系、预期输出格式、验证方式。这个JSON就是调度器的输入。注意Planner Agent的任务不是“生成一个完美的计划”而是“生成一个可执行的计划”。我会在Planner Agent的Prompt里要求它只生成能够被后续步骤明确验证的任务宁可细一点、碎一点也不要出一个笼统的“分析一下”。3.2 调度策略串行、并行、动态切换任务分解完之后就到了调度策略的选择。这里我讲三种最基本的串行执行、并行执行、动态切换。串行执行是最简单的策略前一个任务完成拿到结果再启动下一个任务。它适合有强依赖关系的任务链比如“先获取用户信息再根据用户信息推荐方案”。串行执行的成本好控制不容易出错缺点是慢。并行执行适合多个互不依赖的子任务同时进行。在Orchestrator-Workers架构里我经常把多个独立的Worker调用用并发方式发起。比如要给十个文档分别做摘要我会用一个并发池同时开五个Worker去处理。这里有一个经验值并发数不要超过4到6。这跟模型服务本身的限流有关更关键的是你需要在并发提升的收益和系统出错的风险之间找平衡。并发太高一旦某个Worker触发重试系统逻辑复杂度会指数级上升。动态切换调度是进阶玩法核心是让调度器根据每步的执行结果实时调整后续计划。这需要一套“观察-决策-执行”循环每个任务执行完都必须回传状态信息调度器根据状态信息更新剩余任务列表。我常用的一种实现方式是这样的任务节点不直接指定“下一步是谁”而是指定“下一步的条件”。比如任务A执行完成后如果A的返回结果包含“error”字段就走重试分支否则走正常分支。这个思路有点像状态机把执行路径变成一张可动态导航的图。3.3 上下文隔离与传递不要让每个Agent都读全量对话这是我认为多Agent系统里最重要、也最容易做错的一个环节。单Agent时代的思路是“上下文越全越好”多Agent时代恰恰相反关键原则是每个Agent只拿到它完成当前任务所需的最小上下文。我举个具体的例子。我的一个Agent在执行“检索资料”任务它需要的关键信息只有用户原话中的问题描述、检索到的候选文档标题和摘要。它完全不需要知道另一个Agent正在生成的总结内容也不需要知道这个任务之前的十轮对话历史。最小上下文原则有两个直接收益第一显着降低token消耗省成本第二减少上下文噪声提高模型输出的准确率。模型在信息过载时的表现远不如它面对精炼信息时的表现。具体怎么传呢我建议定义一个统一的任务消息结构包含几个标准字段task_id、user_goal、input_data、context_pointer、output_format。其中context_pointer指向一个共享存储里的文件路径而不是直接嵌入全量上下文。这样如果某个子任务需要更多历史信息Agent可以通过指定路径去读取而不需要把这些信息无脑塞进Prompt。3.4 状态机与超时控制真正容易出问题的地方调度器本质上是一个状态机。每个任务有未开始、执行中、已完成、失败、已超时、已重试这样几个状态。把状态转换明确写清楚代码才不会乱。我最常踩的坑是超时控制。模型调用是不可预测的特别是大模型在高负载时可能拖很久才返回。如果没有超时机制一个Worker卡住了整个调度器都会卡住。我的做法是给每个Worker任务设置两层超时第一层是模型API的请求超时通常设置在30到60秒第二层是整体任务超时比如3分钟。如果第二层超时了调度器要主动标记该任务为失败并根据策略决定是重试还是跳过。这里有一个非常关键的小细节超时的任务其对应的“副作用”要一并处理。如果Worker调用了数据库写入、调用了外部API下单超时并不代表没有执行只是在响应超时。调度器必须区分“真的没执行”和“执行了但没响应”否则重复重试会导致重复操作。我在生产环境里专门维护了一张“副作用登记表”每个Worker在执行任何有副作用的操作前先登记一个操作ID超时重试前先根据操作ID去查一下这个副作用是否已经发生。这个设计帮我挡住了好几次可能导致重复扣费的事故。4. 通信与状态同步Agent之间到底怎么“聊天”4.1 消息协议设计字段最小化原则Agent之间通信很多人第一反应是让它们“自然语言对话”——A把一段话发给BB理解了再回复。这种方式在小规模实验里没问题但规模一大就完了因为自然语言存在歧义B完全可能误解A的意思。我的建议是Agent之间的通信尽量使用结构化消息字段最小化。一个标准的Agent间消息我一般包含这几个字段字段作用示例msg_id消息唯一IDmsg_001msg_type消息类型task_result / task_request / errorfrom_agent发送方标识retriever_agentto_agent接收方标识orchestratortask_id关联任务IDtask_002status状态标记success / failed / partialpayload_size载荷大小提示1240_tokensdata_ref数据引用指向共享存储file://workspace/task_002_result.jsonerror_detail错误详情可选”no results found”这套事务消息的好处是机器可解析、可以监控、可以审计。它把Agent之间的模糊地带压缩到最小。4.2 共享工作区还是消息总线关于状态同步行业内有两套路线共享工作区和消息总线。共享工作区的思路是所有Agent共享一个存储空间本地目录、对象存储、数据库每个Agent把结果写入指定的路径需要数据的Agent按路径去取。这就像团队共用一个网盘每个人把自己做完的部分放进去别人要用时自己拿。消息总线的思路是Agent不直接对接而是通过一个消息队列/事件流传递消息。A把消息发到队列B订阅队列消费。这个方案解耦性更好也更容易做异步处理。我个人的实践是中小型项目用共享工作区就够了简单直接。维护成本也低。大型项目再上消息总线比如Kafka或者Redis Stream。有一个界线判断标准如果你发现多个Agent经常需要同时读写同一个文件、产生竞态冲突那就该引入消息总线了共享文件方案撑不住这种并发场景。4.3 Tool调用与副作用控制等价于Agent的“手”Agent之间的通信只解决“信息流转”问题真正让Agent做事的是工具调用。检索工具、数据库查询工具、API调用工具、代码执行工具这些才是Agent影响真实世界的“手”。工具调用的关键设计原则是工具必须注册不能自创。也就是说Agent能调用哪些工具必须在系统初始化时写死由调度器统一管理。这从根本上杜绝了模型凭空编造工具名的情况。另一个原则是工具的输入输出必须有严格的Schema。我对每个工具都会定义工具名、描述、输入参数Schema、输出格式Schema、错误码列表。所有工具调用都经由调度器转发不直接开放给Agent自由调用。这样设计是为了能做权限控制、日志审计和副作用管理。这里特别提醒一点工具调用返回的结果不要直接、无过滤地塞回Agent的上下文。最好经过一层“精简器”把无关紧要的字段去掉。因为工具返回的结果往往很冗余比如数据库查询会返回一堆字段真正用到的可能就两三个。让Agent读一堆无关字段等于主动给它制造噪声。4.4 可观测性没有Trace就没法调优多Agent系统调试的难度远高于单Agent系统。单Agent你还能通过对话日志来复盘多Agent任何一个结果的产生背后可能是一连串的任务拆分、通信、工具调用。没有一套可观测性体系出了问题你会非常抓狂。我目前在用的方案是每个任务的完整生命周期都记录一条trace链路记录内容包括Agent间的所有消息时间戳、来源、目标、关键字段每次模型调用的token数、延迟、返回状态每次工具调用的输入、输出摘要、耗时调度器的每次决策记录为什么选择了下一步、基于什么条件有了这条链路当一个复杂的产出结果出现问题时你可以顺着链路过一遍很快就定位到是哪个环节出了问题。我还习惯给trace记录加上一个重要字段该Agent所在的版本号。模型是有迭代的记录不清晰还真的会分不清这个问题到底是旧版本模型导致的还是新版本引入的。5. 完整实例搭建一个三Agent协同的数据洞察系统5.1 任务目标与整体设计前面讲了这么多理论还是得上手跑一遍。我用一个稍微真实一点的例子来演示建立一个三Agent协同系统完成“分析一份销售数据输出洞察报告”的任务。我把整个系统拆成四个角色Orchestrator总控、Analyzer数据分析执行者、SearchAgent外部资料检索者、Reporter报告生成者。流程是这样的用户上传一份销售CSVOrchestrator收到后先让Analyzer分析数据的基础统计量比如月度销售额、区域分布、Top产品同时并行让SearchAgent去检索与行业相关的背景趋势。两者结果都回来后Orchestrator再把这些内容汇总交给Reporter生成一份结构化洞察报告。这套设计的价值在于数据分析和资料检索互不依赖可以并行报告生成依赖前两者的结果所以放最后。每个Agent的上下文都很干净Analyzer永远不需要读检索资料Reporter也不需要看原始数据它只拿已经整理好的结论。5.2 代码骨架Orchestrator Analyzer SearchAgent Reporter我用Python写一个可运行的最小骨架方便你理解调度逻辑。先定义消息结构和工具层。import json import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from typing import Any, Optional dataclass class AgentMessage: msg_id: str msg_type: str # task_request / task_result / error from_agent: str to_agent: str task_id: str status: str pending data_ref: Optional[str] None error_detail: str payload: dict field(default_factorydict) class ToolRegistry: 工具注册表Agent只能使用注册过的工具。 def __init__(self): self._tools {} def register(self, name: str, schema: dict, fn: callable): self._tools[name] {schema: schema, fn: fn} def call(self, name: str, **kwargs): if name not in self._tools: raise ValueError(funknown tool: {name}) return self._tools[name][fn](**kwargs) class BaseAgent: 所有Agent的基础类。子类重写execute来处理任务。 def __init__(self, name: str, tools: ToolRegistry): self.name name self.tools tools def execute(self, msg: AgentMessage) - AgentMessage: raise NotImplementedError三个Worker Agent的实现核心是每个Agent只专注处理自己那部分class AnalyzerAgent(BaseAgent): 数据分析Agent只做一件事对CSV做统计计算。 def execute(self, msg: AgentMessage) - AgentMessage: # 实际项目中你会调用大模型做数据理解与代码生成 # 这里用简化的统计逻辑演示。 import csv from io import StringIO file_path msg.data_ref rows [] with open(file_path, r, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: rows.append(row) total_sales sum(float(r[sales]) for r in rows) monthly_sales {} for r in rows: month r[month] monthly_sales[month] monthly_sales.get(month, 0) float(r[sales]) result {total_sales: total_sales, monthly_sales: monthly_sales} return AgentMessage( msg_idfmsg_analyzer_{int(time.time())}, msg_typetask_result, from_agentself.name, to_agentorchestrator, task_idmsg.task_id, statussuccess, payloadresult, ) class SearchAgent(BaseAgent): 检索Agent只负责拉取外部背景资料。 def execute(self, msg: AgentMessage) - AgentMessage: # 实际这里是检索API调用, 示例写死一个结论 result {market_trend: 行业整体处于温和增长期华东区域增速领先} return AgentMessage( msg_idfmsg_search_{int(time.time())}, msg_typetask_result, from_agentself.name, to_agentorchestrator, task_idmsg.task_id, statussuccess, payloadresult, ) class ReporterAgent(BaseAgent): 报告生成Agent接收分析结果与检索结果输出最终报告。 def execute(self, msg: AgentMessage) - AgentMessage: analyzer_result msg.payload.get(analyzer_result, {}) search_result msg.payload.get(search_result, {}) report f # 销售洞察报告 整体销售额约 {analyzer_result[total_sales]:.2f} 万元。 月度趋势: {analyzer_result[monthly_sales]}。 外部背景: {search_result[market_trend]}。 return AgentMessage( msg_idfmsg_reporter_{int(time.time())}, msg_typetask_result, from_agentself.name, to_agentorchestrator, task_idmsg.task_id, statussuccess, payload{report: report}, )Orchestrator是核心调度器。它接收用户请求先并行调度Analyzer和SearchAgent等两侧都就绪后再调度Reporterclass Orchestrator: def __init__(self, agents: dict): self.agents agents def run(self, user_goal: str, file_path: str): # Step 1: 并行调度两个无依赖的Worker with ThreadPoolExecutor(max_workers2) as pool: analyzer_future pool.submit( self.agents[analyzer].execute, AgentMessage( msg_idreq_analyzer, msg_typetask_request, from_agentorchestrator, to_agentanalyzer, task_idtask_analyze, data_reffile_path, ), ) search_future pool.submit( self.agents[searcher].execute, AgentMessage( msg_idreq_searcher, msg_typetask_request, from_agentorchestrator, to_agentsearcher, task_idtask_search, payload{query: user_goal}, ), ) analyzer_result analyzer_future.result() search_result search_future.result() # Step 2: 汇总结果交给Reporter reproter_result self.agents[reporter].execute( AgentMessage( msg_idreq_reporter, msg_typetask_request, from_agentorchestrator, to_agentreporter, task_idtask_report, payload{ analyzer_result: analyzer_result.payload, search_result: search_result.payload, }, ) ) return reproter_result.payload[report]这段骨架里最值得关注的是调度器只管理“任务之间的依赖关系”不干预每个Agent的内部逻辑。Analyzer和SearchAgent完全独立它们的失败互不影响——如果SearchAgent挂了Orchestrator可以把Reporter的输入里的外部背景置为“暂无”系统仍然能产出报告。5.3 运行结果与关键决策点我用一份模拟数据跑了一遍Orchestrator输出的报告大致长这样# 销售洞察报告 整体销售额约 386.50 万元。 月度趋势: {2024-01: 82.3, 2024-02: 95.1, 2024-03: 112.2, 2024-04: 96.9}。 外部背景: 行业整体处于温和增长期华东区域增速领先。生成这个结果的过程中有三个关键决策点值得展开第一个我为什么让Analyzer用代码计算而不是直接让模型读CSV因为统计计算这种任务规则化方法比模型推理可靠得多。模型可以对数字做解释、做归纳但不应该用它做直接的数字求和。多Agent系统的一个核心原则就是能让代码做的就不要让模型做。第二个Analyzer和SearchAgent并行是不是一定好在这个例子里两个任务都很快并行带来的收益不明显。但如果SearchAgent是一个耗时3秒的API调用而Analyzer只需要0.1秒并行就很明显了。实际判断标准是预估两个子任务是否都在100毫秒以上且有独立的调度空间才考虑并行。第三个Reporter收到的输入是“已经结构化好的中间结果”不是原始数据。这是我的一个坚持上游Agent必须把结果整理成结构化数据而不是把一堆原始文件路径丢给下游。这样Reporter的Prompt就非常简单只需要做“把输入数据翻译成报告文字”这一件事模型的表现会稳定很多。5.4 如何扩展到更多Agent这个三Agent骨架虽然简单但扩展路径是清晰的。想加一个“数据清洗Agent”那就放在Analyzer上游用户上传CSV后先由清洗Agent处理缺失值和异常格式输出干净的CSV再交给Analyzer。想加一个“图表生成Agent”就放在Reporter下游Reporter完成报告文字后图表Agent根据Analyzer的统计数据绘制图表。想加一个“质量校验Agent”则可以在Reporter之后加一个审核节点对报告的文字进行事实核查、格式校验。扩展时有一个最重要的注意点每次加Agent都要先问自己这个新Agent的存在是否真的能降低系统整体的复杂度如果只是把本来一个模型调用就能做的事拆出来加上一个Agent只会让通信成本变高、延迟变长得不偿失。多Agent的价值边界是做单Agent做不了或做不好的事而不是把能做的事换个方式再做一遍。6. 生产环境里必须处理的五个坑6.1 循环依赖和Agent死循环多Agent系统最常见的稳定性问题是死循环。两个Agent互相等待对方的结果或者一个Agent反复重启同一个失败任务调度器就像卡在了原地。我在生产环境里经验是用“任务深度计数器”来兜底。每个任务有一个深度值每经过一层调度就加一超过阈值通常是6到8就强制终止并走人工兜底流程。另外一个实用技巧是在Agent的回复中增加一个“fatal_error”标记当Agent连续两次对同一个任务返回错误时就不要再自动重试了直接标记为人工需要介入的状态。6.2 上下文预算爆炸多Agent系统的token消耗比单Agent更容易失控因为每个Agent都在独立消耗上下文。你可能觉得每个Agent的Prompt很短算下来没多少。但一旦调度器把一些冗余信息层层传递给下游Agent上下文量就会像滚雪球一样涨。我的解决办法是在发送前对Payload做一次“瘦身”。用JSONPath把结果里真正需要的字段筛出来其余全部丢弃。同时给每个Agent的输入设置一个最大token阈值超过这个阈值就拒绝执行并提示上游收敛信息。没有这套约束等账单出来的时候再后悔就晚了。6.3 幻觉从子Agent放大到整体单Agent的幻觉你已经很熟悉了而多Agent的幻觉有一个放大效应如果某个Worker在信息不确定时“自信地”生成了一个错误结论上游筛选和下游生成都会把这个错误当作既定事实最后生成的报告会显得非常有说服力因为错误已经经过多层的“精心编织”。应对方案是在Agent的Prompt里明确加入“如果信息缺失请在结果中标记uncertain而不是猜测”。这看起来很简单但真的很有效。同时在系统层面我还习惯在任务流中插入一个验证节点对关键结论做二次校验比如用另一条路径检索同样的信息对比结果是否一致。这种双路径验证的代价不低所以我只在结论影响比较大的任务上启用。6.4 失败重试与补偿重试机制是多Agent系统里最不能想当然的环节。我刚才提到过副作用登记表具体实现上我在工具层统一封装了一个事务包装器每次调用外部副作用API的时候都生成一个request_id存到数据库里。重试时先查这个request_id是否已经存在成功记录如果存在就直接返回旧结果不再发起新的调用。这个设计一开始很麻烦但确实能够防住几类高频事故模型调API超时但服务端其实执行了、网络抖动导致请求发出去没收到返回、Agent陷入重试循环时系统重复发单。如果你做一个涉及支付、下单、扣减库存的多Agent系统这个机制几乎是必选项。6.5 评测多Agent系统怎么评估最后聊一个大多数人没认真想过的问题多Agent系统的评测怎么做。单Agent系统的评测我们一般准备一组标准问答对算准确率就行。多Agent系统不一样它的产出链路长中间环节多一个单纯准确率指标根本反映不出问题。我目前的评测思路是分层评测单元层每个Agent单独评测。给固定的合成输入跑固定的任务看输出是否符合预期。链路层把多个Agent串起来用一组完整业务场景看最终结果的质量以及链路各节点是否正常流转。容错层故意制造异常——让某个Worker返回超时、让工具调用报错、让输入数据缺失——看系统能否正常降级还是直接崩溃。每一层跑完以后我会把整个链路日志里耗时最高、token消耗最多的节点拉出来单看。多Agent系统的优化从来不是笼统地“调Prompt”而是精确到节点的“换参数”“换工具”“换模型”。我个人现在对多Agent系统评测的底线是无论效果多炫都必须保证“可复现”。也就是说同一份输入系统在相同条件下跑三次结果不能差得太远。如果连同一份数据三次生成的结果都完全不同那这个系统在我这里过不了生产的大关。回到开头那个失败的竞品分析项目我在做完这一整套改造之后又把同样的需求跑了一遍。这次系统分了五个步骤、用了三个Agent协作中间环节各自可控出问题能定位到具体节点。跑出来的结果虽然不能说完美但至少让我知道每一步为什么这么做、哪里还能改。这种“可控感”才是多Agent系统最吸引我的地方。如果你正准备在自己的项目里上多Agent我建议你先别急着追新框架。找一个小而真实的业务场景按我上面说的先把任务拆清楚再把调度逻辑写明白跑通最小闭环之后再去考虑更复杂的架构。先跑起来再优化稳扎稳打比一次性堆满技术要靠谱得多。 大模型多Agent协作架构与任务调度实战指南
1. 一次单Agent翻车让我开始认真研究多Agent协作1.1 我遇到的问题复杂任务在单模型会话里越滚越乱这件事说起来有点丢人但确实是我转做AI应用开发以来印象最深的一次失败。当时接了一个内部需求让大模型自动完成一份竞品分析报告要求从资料收集、数据提取、维度拆解、观点生成到最终排版一条龙跑完。我第一反应是写一个超长Prompt把步骤、角色、输出格式全部塞进去然后一次性调大模型。试了几轮结果惨不忍睹。前几轮回答还算正常但任务一长模型就开始“失忆”——前面说要分析定价策略后面总结的时候又变成了渠道策略中间偶尔自己造几个不存在的竞品数据我还得人工去核。更难受的是一旦中途某一个环节的结果不理想整个任务就废了必须从头再跑一遍。而且我根本定位不到是哪一步出的问题。整个对话上下文越滚越长到最后一次请求可能要传几万token的聊天记录成本和延迟都上来了答案反而更差。后来我读了LangGraph、AutoGen、CrewAI这些框架的源码和相关实践材料才意识到问题的根源我用一个Agent包办了所有事本质上是在让单个模型会话承担完整的业务流水线。这个思路在任务路径短、交互少的场景下没问题但一旦任务复杂起来就会同时踩中上下文污染、角色冲突、错误扩散三个坑。1.2 上下文污染、职责混乱、错误扩散三个坑一起踩先说上下文污染。单Agent模式下所有历史信息都存在同一个上下文中。竞品分析这种任务资料收集阶段的噪声数据会一路传递到观点生成阶段早期的小错误会被后期的推理当作“事实”引用。模型本身又没有一个“忘记机制”你只能靠截断上下文来控制但截断之后早期关键信息也没了顾此失彼。第二个是职责混乱。同一个模型开头让它当一个调研员中间让它当一个数据分析师最后让它当一个报告写手。模型确实能做到一定的角色切换但每次切换都是一次隐性的重新校准。而且在这种长流程里它经常分不清“我现在应该专注于哪个目标”于是输出里就开始出现身份漂移——上一段还在做总结下一段又开始假装自己是调研员补了一堆原始资料。第三个是错误扩散。单Agent流程是线性的一步错后面每一步都建立在这个错误之上。更麻烦的是模型不会自动纠偏它只会顺着已经生成的思路继续往下编。你想事后问责结果发现所有输出都混在一个会话里根本分不清是哪一步、由哪个决策引入的错误。那一次之后我做的第一件事就是把“一个超长Prompt跑到底”的方案彻底丢掉开始研究怎么把任务拆给多个模型会话去协作。也就是今天要聊的大模型多Agent系统。1.3 多Agent不是把Prompt拆开而是把系统拆开很多人刚接触多Agent容易有一个误解多Agent就是把一个长Prompt拆成好几个短Prompt每个负责一段再串起来。这确实能解决一部分上下文长度问题但真正的多Agent协作核心不在于“切分”而在于重新设计系统的分工与交互方式。你真正要做的是把一个完整的任务拆成多个可以被独立验证、独立重试、独立扩展的子任务然后交给不同的Agent去执行再由一个调度机制把结果组合起来。每个Agent拥有自己的上下文自己的工具集自己的输出格式甚至自己的兜底策略。这就像开一家餐厅。单Agent模式是一个全能厨师从头做到尾从买菜到配菜到炒菜到摆盘全包客人一多必乱。多Agent模式是后厨分工采购员只管备料砧板师傅只管切配炉头师傅只管炒传菜员只管上桌。每道工序有明确的责任边界出了错可以精确定位到人、到步骤整个系统稳定性一下子就有了质的提升。换句话说多Agent最核心的能力不是“让多个模型一起跑”而是“让多个模型在一条清晰的生产线上各司其职”。协作架构是这条线的骨架任务调度是它的脉搏。2. 协作架构的本质先搞清楚“怎么分活”再决定“怎么干活”2.1 四种主流协作拓扑星型、管道、网状、分层在动手写代码之前你得先建立一张架构地图。我看了大量资料和开源项目之后把多Agent的协作架构归纳成四种主流拓扑理解这四种后面选型就有谱了。星型拓扑也叫中心化编排所有的Agent都围绕一个中心节点工作。中心节点负责任务拆解、分配、收集结果其他Agent只和中心通信。这是最常用的一种实现简单控制力强适合大多数业务场景。缺点呢中心节点容易成为瓶颈如果编排逻辑设计得不好所有压力都堆在它身上。管道拓扑是让Agent排成一串每个Agent只和上下游通信A的输出作为B的输入B的输出作为C的输入。这种结构非常适合内容加工类任务比如“清洗数据 - 提取特征 - 生成报告”。它的好处是链路清晰每个环节可以独立替换升级坏处是整体延迟等于所有环节延迟之和而且中间任何一个环节挂了整条线都得停。网状拓扑里每个Agent都能和其他Agent自由通信没有中心节点。这种架构最灵活但也最容易失控——没有中心约束Agent之间来回传递消息很容易出现“谁都在说话但没人对结果负责”的局面。我只在探索性、低风险的小规模实验里用过网状结构生产环境基本不建议。分层拓扑是大型系统最常见的做法顶层有一个“管理者”Agent中层有几个“小组长”底层是一批执行Agent。每一级只需要关心自己的上下级不需要了解全局。这种架构扩展性最好适合团队规模大、任务类型杂的系统。我画了一张对比表方便你直观理解拓扑类型优势劣势适用场景星型实现简单编排可控中心节点压力大中小规模任务、流程可控的系统管道链路清晰易于优化单点延迟叠加单点故障影响全局有严格先后顺序的内容流水线网状灵活性最高通信混乱结果不可控低风险探索性实验不建议生产使用分层扩展性好权责清晰设计复杂度高层级间通信成本大大型系统、多团队任务协同2.2 中心编排Orchestrator-Workers为什么是大多数场景的默认解我在生产项目里用得最多的还是星型拓扑的变体一般叫Orchestrator-Workers模式也就是中心编排模式。原因很朴素它能用可控的复杂度换来最高的系统可控性。在这个模式里Orchestrator是关键角色。它的职责不是自己干活而是“派活”把用户的目标解析成子任务把子任务分配给对应的Worker收集Worker的返回结果然后决定是继续下一个任务、重试失败任务、还是把结果汇总返回给用户。Worker则是专注的执行者每个Worker只负责一种技能。比如在企业知识库问答系统里我会拆出检索Worker、摘要Worker、推理Worker、格式整理Worker。每个Worker的Prompt非常聚焦上下文也很短基本上不会出现角色漂移。这套结构的最大收益在于故障隔离如果某个Worker的表现不行你只需要替换它一个其他组件完全不受影响。这一点在实际维护中价值极大。我之前维护过一个客服工单分类系统刚开始把分类和优先级判断塞给同一个Agent做结果发现分类做得好但优先级判断很离谱。改成两个Worker之后分类Worker我加了一条规则就修好了优先级Worker的做法则是换了一种思路去调整互相不干扰。还有一个经常被忽略的好处便于并发。Orchestrator收到任务后如果发现多个子任务相互独立就可以并行发给多个Worker整体耗时能压缩不少。这一点在后面的任务调度部分还会详细说。2.3 选型对照什么时候用Pipeline什么时候用Plan-Execute说完了星型再聊两个容易被混为一谈的概念Pipeline模式和Plan-Execute模式。Pipeline模式就是管道拓扑的工程化落地。它的特点是任务路径在任务开始前就已经固定了。比如“上传文档 - 文档解析 - 信息抽取 - 存档”这是一条固定的流水线每个节点做什么是一开始就确定好的。它适合流程相对固化、业务逻辑不会频繁变化的场景优点是稳定、可预测、调优容易。Plan-Execute模式则复杂一些。系统先由一个Planner Agent根据用户目标生成执行计划这个计划不是写死的而是模型根据具体情况动态生成的。然后由执行器按计划逐步执行每步执行完后系统会检查结果再决定下一步。我个人的经验是如果你的业务场景“流程固定、步骤有限”直接用Pipeline简单可靠别硬上Plan-Execute。但如果你面对的是开放式任务比如“帮我把这个GitHub仓库研究透输出一份技术选型报告”这种任务没有固定路径Plan-Execute才是合适的方案。很多人在第一阶段做项目时特别喜欢Plan-Execute因为看起来“高级”。但从成本角度说Plan-Execute需要更多的模型调用、更复杂的错误处理单位任务的token消耗至少是Pipeline的两到三倍。没必要为不需要灵活性的场景付出这个代价。2.4 协作模式与任务类型的匹配关系我把常见任务类型和推荐模式做了个匹配这套逻辑我几乎每个项目都会复用。任务类型典型例子推荐模式原因固定流程型工单分发、文档解析、定时报表Pipeline流程确定管道最稳定开放探索型竞品调研、代码库分析、方案设计Plan-Execute路径未知需要动态规划并行计算型多语言翻译、多文件分析Orchestrator-Workers可并发分发整体耗时最短大型复杂项目新产品研发、跨模块协同分层管理任务多、角色杂、需逐层分解需要逐步推理验证数学证明、代码调试ReAct循环需要边执行边反思纠错这套匹配逻辑的核心出发点只有一个不要让架构的复杂度超过任务本身需要的复杂度。能用Pipeline解决的事就不要上Plan-Execute。能用一个Agent搞定的事就绝对不要强行拆成三个。3. 任务调度从Plan到Tick的核心链路3.1 任务分解怎么把一个大目标拆成可验证的子任务任务调度是整个多Agent系统里最容易被低估的一环。很多系统的瓶颈不在模型能力而在调度设计。任务调度的第一步是任务分解。如果这一步做得不好后面所有环节都会跟着乱。我自己有一个分解原则叫“子任务可验证性原则”拆出来的每一个子任务必须有一个明确的、不依赖模型主观判断的完成标准。打个比方你让一个Agent“研究一下新能源汽车市场”这个任务就没法验证。但如果你拆成“收集2024年全球新能源汽车销量Top10品牌的数据输出为表格字段包含品牌、销量、同比变化”那这个任务就是可验证的——你可以拿输出对照原始数据源马上判断它对不对。在实际操作中我会让Planner Agent先输出一个JSON结构里面包含任务ID、任务描述、依赖关系、预期输出格式、验证方式。这个JSON就是调度器的输入。注意Planner Agent的任务不是“生成一个完美的计划”而是“生成一个可执行的计划”。我会在Planner Agent的Prompt里要求它只生成能够被后续步骤明确验证的任务宁可细一点、碎一点也不要出一个笼统的“分析一下”。3.2 调度策略串行、并行、动态切换任务分解完之后就到了调度策略的选择。这里我讲三种最基本的串行执行、并行执行、动态切换。串行执行是最简单的策略前一个任务完成拿到结果再启动下一个任务。它适合有强依赖关系的任务链比如“先获取用户信息再根据用户信息推荐方案”。串行执行的成本好控制不容易出错缺点是慢。并行执行适合多个互不依赖的子任务同时进行。在Orchestrator-Workers架构里我经常把多个独立的Worker调用用并发方式发起。比如要给十个文档分别做摘要我会用一个并发池同时开五个Worker去处理。这里有一个经验值并发数不要超过4到6。这跟模型服务本身的限流有关更关键的是你需要在并发提升的收益和系统出错的风险之间找平衡。并发太高一旦某个Worker触发重试系统逻辑复杂度会指数级上升。动态切换调度是进阶玩法核心是让调度器根据每步的执行结果实时调整后续计划。这需要一套“观察-决策-执行”循环每个任务执行完都必须回传状态信息调度器根据状态信息更新剩余任务列表。我常用的一种实现方式是这样的任务节点不直接指定“下一步是谁”而是指定“下一步的条件”。比如任务A执行完成后如果A的返回结果包含“error”字段就走重试分支否则走正常分支。这个思路有点像状态机把执行路径变成一张可动态导航的图。3.3 上下文隔离与传递不要让每个Agent都读全量对话这是我认为多Agent系统里最重要、也最容易做错的一个环节。单Agent时代的思路是“上下文越全越好”多Agent时代恰恰相反关键原则是每个Agent只拿到它完成当前任务所需的最小上下文。我举个具体的例子。我的一个Agent在执行“检索资料”任务它需要的关键信息只有用户原话中的问题描述、检索到的候选文档标题和摘要。它完全不需要知道另一个Agent正在生成的总结内容也不需要知道这个任务之前的十轮对话历史。最小上下文原则有两个直接收益第一显着降低token消耗省成本第二减少上下文噪声提高模型输出的准确率。模型在信息过载时的表现远不如它面对精炼信息时的表现。具体怎么传呢我建议定义一个统一的任务消息结构包含几个标准字段task_id、user_goal、input_data、context_pointer、output_format。其中context_pointer指向一个共享存储里的文件路径而不是直接嵌入全量上下文。这样如果某个子任务需要更多历史信息Agent可以通过指定路径去读取而不需要把这些信息无脑塞进Prompt。3.4 状态机与超时控制真正容易出问题的地方调度器本质上是一个状态机。每个任务有未开始、执行中、已完成、失败、已超时、已重试这样几个状态。把状态转换明确写清楚代码才不会乱。我最常踩的坑是超时控制。模型调用是不可预测的特别是大模型在高负载时可能拖很久才返回。如果没有超时机制一个Worker卡住了整个调度器都会卡住。我的做法是给每个Worker任务设置两层超时第一层是模型API的请求超时通常设置在30到60秒第二层是整体任务超时比如3分钟。如果第二层超时了调度器要主动标记该任务为失败并根据策略决定是重试还是跳过。这里有一个非常关键的小细节超时的任务其对应的“副作用”要一并处理。如果Worker调用了数据库写入、调用了外部API下单超时并不代表没有执行只是在响应超时。调度器必须区分“真的没执行”和“执行了但没响应”否则重复重试会导致重复操作。我在生产环境里专门维护了一张“副作用登记表”每个Worker在执行任何有副作用的操作前先登记一个操作ID超时重试前先根据操作ID去查一下这个副作用是否已经发生。这个设计帮我挡住了好几次可能导致重复扣费的事故。4. 通信与状态同步Agent之间到底怎么“聊天”4.1 消息协议设计字段最小化原则Agent之间通信很多人第一反应是让它们“自然语言对话”——A把一段话发给BB理解了再回复。这种方式在小规模实验里没问题但规模一大就完了因为自然语言存在歧义B完全可能误解A的意思。我的建议是Agent之间的通信尽量使用结构化消息字段最小化。一个标准的Agent间消息我一般包含这几个字段字段作用示例msg_id消息唯一IDmsg_001msg_type消息类型task_result / task_request / errorfrom_agent发送方标识retriever_agentto_agent接收方标识orchestratortask_id关联任务IDtask_002status状态标记success / failed / partialpayload_size载荷大小提示1240_tokensdata_ref数据引用指向共享存储file://workspace/task_002_result.jsonerror_detail错误详情可选”no results found”这套事务消息的好处是机器可解析、可以监控、可以审计。它把Agent之间的模糊地带压缩到最小。4.2 共享工作区还是消息总线关于状态同步行业内有两套路线共享工作区和消息总线。共享工作区的思路是所有Agent共享一个存储空间本地目录、对象存储、数据库每个Agent把结果写入指定的路径需要数据的Agent按路径去取。这就像团队共用一个网盘每个人把自己做完的部分放进去别人要用时自己拿。消息总线的思路是Agent不直接对接而是通过一个消息队列/事件流传递消息。A把消息发到队列B订阅队列消费。这个方案解耦性更好也更容易做异步处理。我个人的实践是中小型项目用共享工作区就够了简单直接。维护成本也低。大型项目再上消息总线比如Kafka或者Redis Stream。有一个界线判断标准如果你发现多个Agent经常需要同时读写同一个文件、产生竞态冲突那就该引入消息总线了共享文件方案撑不住这种并发场景。4.3 Tool调用与副作用控制等价于Agent的“手”Agent之间的通信只解决“信息流转”问题真正让Agent做事的是工具调用。检索工具、数据库查询工具、API调用工具、代码执行工具这些才是Agent影响真实世界的“手”。工具调用的关键设计原则是工具必须注册不能自创。也就是说Agent能调用哪些工具必须在系统初始化时写死由调度器统一管理。这从根本上杜绝了模型凭空编造工具名的情况。另一个原则是工具的输入输出必须有严格的Schema。我对每个工具都会定义工具名、描述、输入参数Schema、输出格式Schema、错误码列表。所有工具调用都经由调度器转发不直接开放给Agent自由调用。这样设计是为了能做权限控制、日志审计和副作用管理。这里特别提醒一点工具调用返回的结果不要直接、无过滤地塞回Agent的上下文。最好经过一层“精简器”把无关紧要的字段去掉。因为工具返回的结果往往很冗余比如数据库查询会返回一堆字段真正用到的可能就两三个。让Agent读一堆无关字段等于主动给它制造噪声。4.4 可观测性没有Trace就没法调优多Agent系统调试的难度远高于单Agent系统。单Agent你还能通过对话日志来复盘多Agent任何一个结果的产生背后可能是一连串的任务拆分、通信、工具调用。没有一套可观测性体系出了问题你会非常抓狂。我目前在用的方案是每个任务的完整生命周期都记录一条trace链路记录内容包括Agent间的所有消息时间戳、来源、目标、关键字段每次模型调用的token数、延迟、返回状态每次工具调用的输入、输出摘要、耗时调度器的每次决策记录为什么选择了下一步、基于什么条件有了这条链路当一个复杂的产出结果出现问题时你可以顺着链路过一遍很快就定位到是哪个环节出了问题。我还习惯给trace记录加上一个重要字段该Agent所在的版本号。模型是有迭代的记录不清晰还真的会分不清这个问题到底是旧版本模型导致的还是新版本引入的。5. 完整实例搭建一个三Agent协同的数据洞察系统5.1 任务目标与整体设计前面讲了这么多理论还是得上手跑一遍。我用一个稍微真实一点的例子来演示建立一个三Agent协同系统完成“分析一份销售数据输出洞察报告”的任务。我把整个系统拆成四个角色Orchestrator总控、Analyzer数据分析执行者、SearchAgent外部资料检索者、Reporter报告生成者。流程是这样的用户上传一份销售CSVOrchestrator收到后先让Analyzer分析数据的基础统计量比如月度销售额、区域分布、Top产品同时并行让SearchAgent去检索与行业相关的背景趋势。两者结果都回来后Orchestrator再把这些内容汇总交给Reporter生成一份结构化洞察报告。这套设计的价值在于数据分析和资料检索互不依赖可以并行报告生成依赖前两者的结果所以放最后。每个Agent的上下文都很干净Analyzer永远不需要读检索资料Reporter也不需要看原始数据它只拿已经整理好的结论。5.2 代码骨架Orchestrator Analyzer SearchAgent Reporter我用Python写一个可运行的最小骨架方便你理解调度逻辑。先定义消息结构和工具层。import json import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from typing import Any, Optional dataclass class AgentMessage: msg_id: str msg_type: str # task_request / task_result / error from_agent: str to_agent: str task_id: str status: str pending data_ref: Optional[str] None error_detail: str payload: dict field(default_factorydict) class ToolRegistry: 工具注册表Agent只能使用注册过的工具。 def __init__(self): self._tools {} def register(self, name: str, schema: dict, fn: callable): self._tools[name] {schema: schema, fn: fn} def call(self, name: str, **kwargs): if name not in self._tools: raise ValueError(funknown tool: {name}) return self._tools[name][fn](**kwargs) class BaseAgent: 所有Agent的基础类。子类重写execute来处理任务。 def __init__(self, name: str, tools: ToolRegistry): self.name name self.tools tools def execute(self, msg: AgentMessage) - AgentMessage: raise NotImplementedError三个Worker Agent的实现核心是每个Agent只专注处理自己那部分class AnalyzerAgent(BaseAgent): 数据分析Agent只做一件事对CSV做统计计算。 def execute(self, msg: AgentMessage) - AgentMessage: # 实际项目中你会调用大模型做数据理解与代码生成 # 这里用简化的统计逻辑演示。 import csv from io import StringIO file_path msg.data_ref rows [] with open(file_path, r, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: rows.append(row) total_sales sum(float(r[sales]) for r in rows) monthly_sales {} for r in rows: month r[month] monthly_sales[month] monthly_sales.get(month, 0) float(r[sales]) result {total_sales: total_sales, monthly_sales: monthly_sales} return AgentMessage( msg_idfmsg_analyzer_{int(time.time())}, msg_typetask_result, from_agentself.name, to_agentorchestrator, task_idmsg.task_id, statussuccess, payloadresult, ) class SearchAgent(BaseAgent): 检索Agent只负责拉取外部背景资料。 def execute(self, msg: AgentMessage) - AgentMessage: # 实际这里是检索API调用, 示例写死一个结论 result {market_trend: 行业整体处于温和增长期华东区域增速领先} return AgentMessage( msg_idfmsg_search_{int(time.time())}, msg_typetask_result, from_agentself.name, to_agentorchestrator, task_idmsg.task_id, statussuccess, payloadresult, ) class ReporterAgent(BaseAgent): 报告生成Agent接收分析结果与检索结果输出最终报告。 def execute(self, msg: AgentMessage) - AgentMessage: analyzer_result msg.payload.get(analyzer_result, {}) search_result msg.payload.get(search_result, {}) report f # 销售洞察报告 整体销售额约 {analyzer_result[total_sales]:.2f} 万元。 月度趋势: {analyzer_result[monthly_sales]}。 外部背景: {search_result[market_trend]}。 return AgentMessage( msg_idfmsg_reporter_{int(time.time())}, msg_typetask_result, from_agentself.name, to_agentorchestrator, task_idmsg.task_id, statussuccess, payload{report: report}, )Orchestrator是核心调度器。它接收用户请求先并行调度Analyzer和SearchAgent等两侧都就绪后再调度Reporterclass Orchestrator: def __init__(self, agents: dict): self.agents agents def run(self, user_goal: str, file_path: str): # Step 1: 并行调度两个无依赖的Worker with ThreadPoolExecutor(max_workers2) as pool: analyzer_future pool.submit( self.agents[analyzer].execute, AgentMessage( msg_idreq_analyzer, msg_typetask_request, from_agentorchestrator, to_agentanalyzer, task_idtask_analyze, data_reffile_path, ), ) search_future pool.submit( self.agents[searcher].execute, AgentMessage( msg_idreq_searcher, msg_typetask_request, from_agentorchestrator, to_agentsearcher, task_idtask_search, payload{query: user_goal}, ), ) analyzer_result analyzer_future.result() search_result search_future.result() # Step 2: 汇总结果交给Reporter reproter_result self.agents[reporter].execute( AgentMessage( msg_idreq_reporter, msg_typetask_request, from_agentorchestrator, to_agentreporter, task_idtask_report, payload{ analyzer_result: analyzer_result.payload, search_result: search_result.payload, }, ) ) return reproter_result.payload[report]这段骨架里最值得关注的是调度器只管理“任务之间的依赖关系”不干预每个Agent的内部逻辑。Analyzer和SearchAgent完全独立它们的失败互不影响——如果SearchAgent挂了Orchestrator可以把Reporter的输入里的外部背景置为“暂无”系统仍然能产出报告。5.3 运行结果与关键决策点我用一份模拟数据跑了一遍Orchestrator输出的报告大致长这样# 销售洞察报告 整体销售额约 386.50 万元。 月度趋势: {2024-01: 82.3, 2024-02: 95.1, 2024-03: 112.2, 2024-04: 96.9}。 外部背景: 行业整体处于温和增长期华东区域增速领先。生成这个结果的过程中有三个关键决策点值得展开第一个我为什么让Analyzer用代码计算而不是直接让模型读CSV因为统计计算这种任务规则化方法比模型推理可靠得多。模型可以对数字做解释、做归纳但不应该用它做直接的数字求和。多Agent系统的一个核心原则就是能让代码做的就不要让模型做。第二个Analyzer和SearchAgent并行是不是一定好在这个例子里两个任务都很快并行带来的收益不明显。但如果SearchAgent是一个耗时3秒的API调用而Analyzer只需要0.1秒并行就很明显了。实际判断标准是预估两个子任务是否都在100毫秒以上且有独立的调度空间才考虑并行。第三个Reporter收到的输入是“已经结构化好的中间结果”不是原始数据。这是我的一个坚持上游Agent必须把结果整理成结构化数据而不是把一堆原始文件路径丢给下游。这样Reporter的Prompt就非常简单只需要做“把输入数据翻译成报告文字”这一件事模型的表现会稳定很多。5.4 如何扩展到更多Agent这个三Agent骨架虽然简单但扩展路径是清晰的。想加一个“数据清洗Agent”那就放在Analyzer上游用户上传CSV后先由清洗Agent处理缺失值和异常格式输出干净的CSV再交给Analyzer。想加一个“图表生成Agent”就放在Reporter下游Reporter完成报告文字后图表Agent根据Analyzer的统计数据绘制图表。想加一个“质量校验Agent”则可以在Reporter之后加一个审核节点对报告的文字进行事实核查、格式校验。扩展时有一个最重要的注意点每次加Agent都要先问自己这个新Agent的存在是否真的能降低系统整体的复杂度如果只是把本来一个模型调用就能做的事拆出来加上一个Agent只会让通信成本变高、延迟变长得不偿失。多Agent的价值边界是做单Agent做不了或做不好的事而不是把能做的事换个方式再做一遍。6. 生产环境里必须处理的五个坑6.1 循环依赖和Agent死循环多Agent系统最常见的稳定性问题是死循环。两个Agent互相等待对方的结果或者一个Agent反复重启同一个失败任务调度器就像卡在了原地。我在生产环境里经验是用“任务深度计数器”来兜底。每个任务有一个深度值每经过一层调度就加一超过阈值通常是6到8就强制终止并走人工兜底流程。另外一个实用技巧是在Agent的回复中增加一个“fatal_error”标记当Agent连续两次对同一个任务返回错误时就不要再自动重试了直接标记为人工需要介入的状态。6.2 上下文预算爆炸多Agent系统的token消耗比单Agent更容易失控因为每个Agent都在独立消耗上下文。你可能觉得每个Agent的Prompt很短算下来没多少。但一旦调度器把一些冗余信息层层传递给下游Agent上下文量就会像滚雪球一样涨。我的解决办法是在发送前对Payload做一次“瘦身”。用JSONPath把结果里真正需要的字段筛出来其余全部丢弃。同时给每个Agent的输入设置一个最大token阈值超过这个阈值就拒绝执行并提示上游收敛信息。没有这套约束等账单出来的时候再后悔就晚了。6.3 幻觉从子Agent放大到整体单Agent的幻觉你已经很熟悉了而多Agent的幻觉有一个放大效应如果某个Worker在信息不确定时“自信地”生成了一个错误结论上游筛选和下游生成都会把这个错误当作既定事实最后生成的报告会显得非常有说服力因为错误已经经过多层的“精心编织”。应对方案是在Agent的Prompt里明确加入“如果信息缺失请在结果中标记uncertain而不是猜测”。这看起来很简单但真的很有效。同时在系统层面我还习惯在任务流中插入一个验证节点对关键结论做二次校验比如用另一条路径检索同样的信息对比结果是否一致。这种双路径验证的代价不低所以我只在结论影响比较大的任务上启用。6.4 失败重试与补偿重试机制是多Agent系统里最不能想当然的环节。我刚才提到过副作用登记表具体实现上我在工具层统一封装了一个事务包装器每次调用外部副作用API的时候都生成一个request_id存到数据库里。重试时先查这个request_id是否已经存在成功记录如果存在就直接返回旧结果不再发起新的调用。这个设计一开始很麻烦但确实能够防住几类高频事故模型调API超时但服务端其实执行了、网络抖动导致请求发出去没收到返回、Agent陷入重试循环时系统重复发单。如果你做一个涉及支付、下单、扣减库存的多Agent系统这个机制几乎是必选项。6.5 评测多Agent系统怎么评估最后聊一个大多数人没认真想过的问题多Agent系统的评测怎么做。单Agent系统的评测我们一般准备一组标准问答对算准确率就行。多Agent系统不一样它的产出链路长中间环节多一个单纯准确率指标根本反映不出问题。我目前的评测思路是分层评测单元层每个Agent单独评测。给固定的合成输入跑固定的任务看输出是否符合预期。链路层把多个Agent串起来用一组完整业务场景看最终结果的质量以及链路各节点是否正常流转。容错层故意制造异常——让某个Worker返回超时、让工具调用报错、让输入数据缺失——看系统能否正常降级还是直接崩溃。每一层跑完以后我会把整个链路日志里耗时最高、token消耗最多的节点拉出来单看。多Agent系统的优化从来不是笼统地“调Prompt”而是精确到节点的“换参数”“换工具”“换模型”。我个人现在对多Agent系统评测的底线是无论效果多炫都必须保证“可复现”。也就是说同一份输入系统在相同条件下跑三次结果不能差得太远。如果连同一份数据三次生成的结果都完全不同那这个系统在我这里过不了生产的大关。回到开头那个失败的竞品分析项目我在做完这一整套改造之后又把同样的需求跑了一遍。这次系统分了五个步骤、用了三个Agent协作中间环节各自可控出问题能定位到具体节点。跑出来的结果虽然不能说完美但至少让我知道每一步为什么这么做、哪里还能改。这种“可控感”才是多Agent系统最吸引我的地方。如果你正准备在自己的项目里上多Agent我建议你先别急着追新框架。找一个小而真实的业务场景按我上面说的先把任务拆清楚再把调度逻辑写明白跑通最小闭环之后再去考虑更复杂的架构。先跑起来再优化稳扎稳打比一次性堆满技术要靠谱得多。