多Agent协作架构实战：从任务分解到论文协同写作系统搭建

1. 从单兵作战到团队协同多Agent架构到底解决了什么单Agent系统在过去两年里几乎成了大模型应用的默认形态——一个模型、一段提示词、一套工具调用能回答问题、能写代码、能查资料。但只要任务链条稍微拉长问题就暴露出来了上下文窗口被塞爆、工具调用互相干扰、一个环节出错整条链路崩掉。我自己最早做论文辅助写作的时候就吃过这个亏让一个Agent同时负责文献检索、数据分析和文字润色结果它在检索阶段就把上下文用掉了七成后面写出来的东西逻辑断层严重。多Agent协作架构的核心思路说白了就是把一个大而全的任务拆成若干职责单一的子任务交给不同的Agent分别处理再通过一套调度机制把它们串起来。这跟软件工程里的微服务拆分是一个道理——你不会把用户认证、订单处理、支付逻辑全塞进一个函数里Agent系统也一样。这套架构真正解决的问题有三个层面。第一是上下文隔离每个Agent只关心自己那部分信息检索Agent不需要知道最终论文的措辞风格写作Agent也不需要看到原始数据库里的几万行记录。第二是能力专精你可以给检索Agent配上搜索引擎工具给分析Agent配上代码执行环境给审校Agent配上质量评估标准各司其职。第三是容错与重试某个Agent输出不达标时调度器可以只让它重做而不必把整个流程推倒重来。适合关注这块内容的人其实很明确已经在用大模型做实际项目、但发现单Agent撑不住复杂流程的开发者想从“调API写Demo”进阶到“搭系统做产品”的工程师以及需要AI协同完成多步骤任务的研究人员。如果你还停留在“写个提示词让模型回答”的阶段这篇文章可能会有点超前但提前了解架构思路没有坏处。2. 协作架构的四种典型形态与选型逻辑2.1 顺序流水线最直观但最容易踩坑顺序流水线是最容易理解的协作模式——Agent A做完交给Agent BB做完交给C像工厂流水线一样。我最早搭的论文辅助系统就是这个结构检索Agent → 分析Agent → 写作Agent → 审校Agent。这种模式的好处是调试简单每个环节的输入输出都很明确出了问题直接定位到具体Agent。但它的坑也很明显上游的错误会一路传导到下游。检索Agent如果返回了一堆不相关的文献分析Agent基于这些文献得出的结论就是歪的写作Agent再把这歪结论写成流畅的文字审校Agent根本看不出问题——因为文字本身没毛病错在源头。实操心得顺序流水线一定要在每个环节加“质量门禁”。比如检索Agent返回结果后用一个轻量级的相关性评分步骤过滤掉明显不相关的条目再交给下游。这个门禁不需要多复杂哪怕只是让模型打个1-5分低于3分的直接丢弃都能大幅降低错误传导的概率。2.2 层级调度主管Agent与执行Agent的分工层级调度是我目前最推荐的架构尤其适合任务步骤超过5步的场景。它的结构是一个主管AgentOrchestrator负责理解总目标、拆解任务、分配给执行Agent并收集结果做最终整合。执行Agent只负责自己那一小块做完就汇报。这种架构的关键在于主管Agent的提示词设计。它需要具备三种能力任务分解把“写一篇综述”拆成检索、分类、对比、撰写、审校、路由决策判断当前该调用哪个Agent、结果评估判断执行Agent的输出是否合格不合格就打回重做。我实际用下来主管Agent的提示词里一定要写清楚“什么情况下算完成”“什么情况下需要重试”“什么情况下需要换一个Agent来做”。不写清楚的话它要么无限循环调用同一个Agent要么在结果明显不合格时就草草收尾。2.3 对等协商Agent之间的辩论与共识对等协商模式比较适合需要多角度验证的任务比如事实核查、方案评估、质量校准。多个Agent各自独立给出结论然后通过一轮或多轮“辩论”来达成共识。每个Agent可以看到其他Agent的输出并据此修正自己的判断。这种模式在论文质量校准场景里特别有用。我试过让三个Agent分别从“方法论严谨性”“数据支撑充分性”“逻辑连贯性”三个角度审同一篇稿子然后让它们互相看对方的意见再给第二轮判断。结果发现第一轮里某个Agent漏掉的问题往往会被另一个Agent指出来第二轮大家都会更全面。但它的代价是Token消耗成倍增加而且辩论轮数不好控制。我的经验是最多两轮再多就是浪费。第一轮独立判断第二轮看别人意见后修正两轮之后取多数共识或者由主管Agent裁决。2.4 黑板模式共享状态下的异步协作黑板模式来自经典AI领域核心思想是有一个共享的数据空间黑板所有Agent都可以往上面读写信息谁有需要谁就去取。这种模式适合任务边界不那么清晰、需要灵活协作的场景。比如一个复杂的数据分析项目数据清洗Agent把处理好的数据写到黑板上特征工程Agent看到数据就自动开始工作建模Agent看到特征就接着跑。它们之间不需要显式的调用关系通过黑板上的状态变化来触发。这种模式实现起来最复杂因为要处理并发读写、状态一致性、触发条件等问题。我一般只在任务流程高度动态、无法预先确定步骤顺序时才考虑用它。对于大多数应用场景层级调度已经够用了。架构形态适用场景实现难度Token消耗容错能力顺序流水线步骤固定、依赖明确低低弱层级调度步骤多、需要动态决策中中强对等协商需要多角度验证中高强黑板模式流程动态、边界模糊高中中3. 任务调度器的设计细节从任务分解到结果聚合3.1 任务分解的粒度控制任务分解是调度器最核心也最难做好的部分。分得太粗每个子任务还是太复杂Agent处理不好分得太细Agent数量爆炸调度开销和通信成本急剧上升。我的经验法则是每个子任务的预期输出控制在500-1500字或者一个明确的结构化结果。比如“检索相关文献”这个任务如果预期输出是“20篇文献的标题和摘要”那粒度就合适如果预期输出是“一篇完整的文献综述”那就太粗了应该继续拆成“检索”“分类”“对比分析”“撰写综述”四步。另一个关键是子任务之间的依赖关系要显式声明。哪些任务可以并行、哪些必须串行、哪些需要等待特定输入这些信息必须在任务分解阶段就确定下来。我见过不少实现是把依赖关系藏在提示词里让模型自己判断结果模型经常搞错顺序导致下游Agent拿不到需要的输入。3.2 Agent能力注册与动态路由调度器需要知道每个Agent能做什么才能正确路由任务。我通常会给每个Agent维护一份能力描述包括擅长处理的输入类型、能调用的工具列表、输出格式要求、以及一个简单的性能画像比如“在代码生成任务上表现好但在长文本理解上一般”。动态路由的逻辑可以很简单根据任务类型匹配Agent能力标签选匹配度最高的。也可以更复杂根据历史表现动态调整权重某个Agent最近在同类任务上失败率高就降低它的优先级。# Agent能力注册的简化示例 agents { retriever: { capabilities: [search, filter, summarize], input_type: query_string, output_type: document_list, tools: [web_search, vector_db], performance: {search: 0.92, filter: 0.85} }, analyzer: { capabilities: [data_analysis, statistics, visualization], input_type: structured_data, output_type: analysis_report, tools: [python_executor, chart_generator], performance: {analysis: 0.88, visualization: 0.79} } } def route_task(task_type, input_data): candidates [ (name, agent) for name, agent in agents.items() if task_type in agent[capabilities] ] if not candidates: raise ValueError(f没有Agent能处理任务类型: {task_type}) # 按历史表现排序选最优 candidates.sort( keylambda x: x[1][performance].get(task_type, 0), reverseTrue ) return candidates[0][0]3.3 结果聚合与冲突消解多个Agent的输出汇总到一起时冲突几乎不可避免。检索Agent说“这个方向有大量研究”分析Agent说“数据显示这个方向证据不足”写作Agent就不知道该听谁的。我的处理策略是分层聚合先让同层级的Agent输出结构化结果比如JSON格式包含结论、置信度、支撑证据然后由主管Agent做冲突检测。如果两个Agent的结论矛盾就看谁的证据更充分、置信度更高如果置信度接近就触发一轮额外的验证任务让第三个Agent专门去核查这个争议点。注意结果聚合时一定要保留每个Agent的原始输出和推理过程不要只保留最终结论。出了问题需要回溯时这些中间信息是唯一的线索。3.4 超时、重试与降级策略调度器必须处理Agent“卡死”或“输出不合格”的情况。我的做法是给每个任务设置三层保护第一层是超时控制单个Agent任务超过预设时间比如60秒就强制中断标记为超时。第二层是重试机制超时或输出格式错误的自动重试最多2次每次可以微调提示词比如加上“请确保输出为JSON格式”。第三层是降级方案重试仍失败的要么换一个Agent来做要么用一个简化的备用逻辑兜底比如检索失败就返回空列表让下游知道这步没完成。import asyncio async def execute_with_retry(agent, task, max_retries2, timeout60): for attempt in range(max_retries 1): try: result await asyncio.wait_for( agent.execute(task), timeouttimeout ) if validate_output(result): return result else: task add_format_hint(task, attempt) except asyncio.TimeoutError: if attempt max_retries: return fallback_result(task) continue return fallback_result(task)4. 通信机制与状态管理Agent之间怎么“说话”4.1 消息传递 vs 共享内存Agent之间的通信方式直接影响系统的复杂度和性能。消息传递是每个Agent把结果发给调度器调度器再转发给下一个Agent好处是解耦彻底坏处是调度器成为瓶颈。共享内存是所有Agent读写同一个状态存储好处是灵活坏处是并发控制麻烦。我大多数场景下用的是混合模式Agent之间通过调度器传递消息但调度器维护一个全局状态对象记录每个任务的执行状态、输入输出、依赖关系。Agent需要历史信息时从全局状态里查而不是靠消息里携带全部上下文。这样做的好处是上下文不会无限膨胀。如果每个消息都携带完整的历史到第五个Agent时消息长度可能已经上万Token了。用全局状态存储Agent按需查询消息本身只传增量信息。4.2 状态持久化与断点恢复复杂任务跑一半失败了如果要从头再来时间和Token成本都受不了。所以状态持久化是必须的。我通常用JSON文件或轻量级数据库存储每个步骤的状态包括任务ID、当前步骤、已完成步骤的输出、待执行步骤列表、全局上下文。断点恢复的逻辑就是读取状态文件找到最后一个成功完成的步骤从它的下一个步骤开始继续执行。这里有个细节要注意——已完成步骤的输出要完整保留因为下游Agent可能还需要用到。我一般会把每个步骤的输出单独存一个文件状态文件里只存文件路径和摘要。4.3 上下文窗口的分配策略多Agent系统里每个Agent的上下文窗口是有限资源怎么分配很讲究。我的策略是按需注入Agent启动时只给最必要的系统提示词和当前任务描述需要历史信息时再通过工具调用去查。而不是一上来就把所有历史塞进去。具体来说检索Agent的上下文里只需要查询词和检索工具说明分析Agent需要的是检索结果和代码执行工具说明写作Agent需要的是分析报告和写作规范。每个Agent的上下文都控制在2000Token以内这样即使并发跑多个Agent总消耗也可控。4.4 工具调用的隔离与共享不同Agent可能需要调用相同的工具比如都用搜索引擎也可能需要调用专属工具比如只有分析Agent能用代码执行器。我的做法是工具注册在调度器层面按Agent能力授权。Agent发起工具调用请求时调度器检查该Agent是否有权限有则代理执行并返回结果。这样做的好处是工具调用日志统一管理出了问题可以追溯是哪个Agent在什么时候调用了什么工具、传了什么参数、返回了什么结果。另外也方便做频率限制防止某个Agent疯狂调用搜索接口把配额用光。5. 实战搭建从零构建一个论文协同写作系统5.1 系统整体设计与Agent角色划分我拿论文协同写作这个场景来完整走一遍搭建过程。这个系统的目标是给定一个研究主题自动完成文献检索、数据分析、初稿撰写、质量审校四个阶段最终输出一篇结构完整的论文初稿。Agent角色划分如下检索Agent负责根据主题搜索相关文献返回标题、摘要、来源、年份等结构化信息。分析Agent负责对检索到的文献做分类、对比、趋势分析输出分析报告。写作Agent负责根据分析报告撰写论文初稿包括引言、相关工作、方法、实验、结论等章节。审校Agent负责检查初稿的逻辑连贯性、论证充分性、格式规范性输出修改意见。主管Agent负责整体调度、任务分解、结果聚合、质量门禁。5.2 主管Agent的提示词设计要点主管Agent的提示词是整个系统的大脑我反复调整了十几版才稳定下来。核心结构包括你是一个论文写作项目的主管调度器。你的职责是 1. 接收用户的研究主题将其分解为检索、分析、写作、审校四个阶段。 2. 每个阶段结束后评估输出质量。如果质量不达标决定是重试还是调整任务描述。 3. 所有阶段完成后整合最终结果并输出。 质量评估标准 - 检索阶段至少返回10篇相关文献覆盖近5年。 - 分析阶段必须包含分类统计、趋势判断、研究空白识别。 - 写作阶段结构完整每个章节不少于500字引用规范。 - 审校阶段必须指出至少3个具体问题并给出修改建议。 当前状态{state} 请决定下一步行动输出JSON格式{action: ..., agent: ..., task: ...}关键点是质量评估标准要具体可量化不能写“质量要好”这种模糊描述。另外状态信息要实时注入让主管Agent知道现在进行到哪一步了。5.3 检索Agent的工具配置与输出规范检索Agent需要接入搜索工具。我一般配两个一个是通用网页搜索一个是学术数据库API。提示词里要明确输出格式{ papers: [ { title: 论文标题, authors: [作者1, 作者2], year: 2024, source: 期刊/会议名称, abstract: 摘要内容, relevance_score: 0.95, key_findings: [发现1, 发现2] } ], total_found: 25, search_queries_used: [查询词1, 查询词2] }实操心得检索Agent最容易犯的错是返回一堆不相关的文献。我的解决办法是在提示词里加一条“如果摘要与主题的相关性低于0.7不要包含在结果中”并且要求它给出相关性评分。这样下游分析Agent可以再做一次过滤。5.4 分析Agent的代码执行与数据洞察分析Agent需要调用代码执行器来做统计和可视化。我给它配了一个Python执行环境提示词里要求它先写分析代码、执行、再根据结果写分析报告。# 分析Agent可能生成的代码示例 import json from collections import Counter papers json.loads(input_data)[papers] # 按年份统计 year_dist Counter(p[year] for p in papers) # 按来源统计 source_dist Counter(p[source] for p in papers) # 关键词提取 all_findings [f for p in papers for f in p[key_findings]] finding_freq Counter(all_findings).most_common(10) report { year_distribution: dict(year_dist), top_sources: source_dist.most_common(5), hot_findings: finding_freq, total_analyzed: len(papers) } print(json.dumps(report, ensure_asciiFalse))分析报告的输出要结构化包含定量结果统计数字和定性判断趋势解读、研究空白。我要求分析Agent在报告最后必须写一段“研究空白识别”明确指出当前文献没有覆盖到的方向这直接为写作Agent提供了创新点素材。5.5 写作Agent的分章节生成策略写作Agent如果一次性生成整篇论文质量很难保证。我的做法是分章节生成先写大纲确认后再逐章生成。每章生成时只注入该章需要的分析数据和写作要求。当前任务撰写“相关工作”章节。 可用素材{analysis_report} 写作要求 1. 按研究方向分类组织每类不少于2段。 2. 引用检索到的文献格式为[作者, 年份]。 3. 在每类末尾指出该方向的局限性。 4. 字数控制在800-1200字。分章节生成的好处是每个章节的上下文都很聚焦模型不容易跑偏。而且如果某一章质量不行只需要重写那一章不用整篇重来。5.6 审校Agent的质量校准清单审校Agent的提示词里我放了一份质量校准清单要求它逐项检查检查项检查标准不合格处理逻辑连贯性章节之间过渡自然无矛盾标注具体位置建议修改论证充分性每个论点有文献或数据支撑指出缺少支撑的论点引用规范性引用格式统一无遗漏列出格式错误的引用结构完整性包含所有必要章节指出缺失章节语言表达无语法错误术语准确标注问题句子审校Agent的输出是一份修改意见清单每条意见包含问题位置、问题描述、修改建议、严重程度高/中/低。主管Agent根据严重程度决定是否需要写作Agent重做。6. 性能调优与常见故障排查6.1 Token消耗的优化手段多Agent系统的Token消耗是单Agent的3-5倍优化空间很大。我常用的手段有上下文压缩Agent之间传递信息时不传原始全文传摘要加关键数据。比如检索Agent返回20篇文献的完整摘要可能有8000Token压缩成“20篇文献主要方向为X、Y、Z高相关文献5篇关键发现包括...”可能只要500Token。缓存复用相同或相似的子任务结果缓存起来下次直接取。比如同一个研究主题的检索结果24小时内不变第二次跑就直接读缓存。模型分级不是所有Agent都需要用最强的模型。检索Agent用轻量级模型做初步筛选分析Agent和写作Agent用强模型。我实测下来检索环节用7B级别的模型和用70B级别的模型最终结果差异不到5%但成本差10倍以上。6.2 Agent“死循环”的识别与中断死循环是多Agent系统最烦人的问题——主管Agent不断让同一个执行Agent重试执行Agent每次输出都差不多主管Agent每次都不满意。我遇到过最夸张的一次一个检索任务重试了17次还没通过。识别死循环的信号同一个Agent被连续调用超过3次且任务描述没有实质性变化。中断策略是设置最大重试次数我一般设3次超过后强制跳过该步骤记录失败原因继续执行后续步骤。后续步骤如果需要该步骤的输出就用降级方案兜底。6.3 输出格式不一致的强制约束Agent输出格式不一致是另一个高频问题。你要求它输出JSON它给你输出Markdown你要求它输出列表它给你输出段落。我的解决办法是三重约束第一提示词里用明确的格式模板并且加上“只输出JSON不要有任何其他文字”。第二在代码层面做格式校验不符合的直接打回重试。第三重试时把格式错误信息反馈给Agent比如“你上次的输出不是合法JSON错误在第3行请修正”。import json def validate_and_parse(output, expected_schema): try: data json.loads(output) except json.JSONDecodeError as e: return None, fJSON解析失败: {str(e)} missing_fields [ field for field in expected_schema if field not in data ] if missing_fields: return None, f缺少字段: {missing_fields} return data, None6.4 并发任务下的资源竞争当多个Agent并行执行时资源竞争问题就来了。最常见的是API调用频率限制和文件读写冲突。我的处理方式是加一个调度队列所有Agent的工具调用请求先入队由调度器统一控制并发数。对于文件读写每个Agent有自己独立的临时目录需要共享的数据通过调度器的状态存储来交换避免直接操作同一个文件。这样即使多个Agent同时跑也不会互相干扰。6.5 从日志中定位问题根源多Agent系统的日志一定要详细。我通常记录四类日志调度日志谁在什么时候调用了哪个Agent、传了什么任务、执行日志Agent的输入输出、工具调用记录、错误日志异常堆栈、超时记录、状态变更日志全局状态的每次修改。排查问题时先看调度日志找到出问题的环节再看执行日志看Agent的实际输入输出最后看错误日志确认异常类型。这三层日志配合基本能定位到90%以上的问题。提示日志里一定要记录时间戳和任务ID否则并发场景下日志混在一起根本没法看。我一般用“任务ID-步骤序号-Agent名称”作为日志前缀检索起来很方便。7. 多Agent系统的边界与我的实际体会多Agent不是银弹它有明确的适用边界。任务步骤少于3步的单Agent加好提示词就够了上多Agent反而是过度设计。任务对实时性要求极高的多Agent的调度开销可能让响应时间翻倍也不合适。团队里没有人懂Agent调试的贸然上多Agent系统出了问题连日志都看不懂。我实际用下来多Agent系统最大的价值不在于“让AI更聪明”而在于让复杂任务的执行过程变得可观测、可干预、可复现。单Agent像一个黑盒你给它输入它给你输出中间发生了什么你不知道。多Agent把黑盒拆成了若干个灰盒每个环节的输入输出都可见出了问题能定位效果不好能调优。另一个体会是主管Agent的提示词质量决定了整个系统的上限。执行Agent再强如果主管Agent任务分解得乱七八糟、质量评估标准模糊不清最终结果也好不到哪去。我在主管Agent的提示词上花的时间比其他所有Agent加起来都多。最后分享一个实用技巧先用单Agent跑通全流程再逐步拆分成多Agent。不要一上来就设计复杂的多Agent架构那样你连基线效果都不知道。先让一个Agent把任务做完记录它在哪个环节表现差然后针对性地把那个环节拆出来做成独立Agent。这样每一步拆分都有明确的收益而不是为了架构而架构。