1. 多Agent协作到底在解决什么问题单Agent跑不通复杂任务这是我做了大半年Agent开发后最深的体会。你让一个Agent去完成“分析一份财报、生成摘要、翻译成英文、再做成PPT大纲”这种链路它大概率会在第三步开始胡言乱语或者干脆忘记第一步拿到的数据。原因不复杂——单Agent的上下文窗口是有限的工具调用能力是有限的而且它没有“第二双眼睛”帮它检查错误。多Agent协作的核心思路就一句话把一个大任务拆成多个子任务每个子任务交给专门的Agent再通过一套调度机制把它们串起来。这跟人类团队干活是一个道理——你不会让一个人同时做财务分析、翻译和设计而是分给不同角色最后有人负责汇总。这套东西能做什么举几个我实际接触过的场景。第一个是研究报告生成一个Agent负责检索资料一个负责数据分析一个负责撰写最后一个负责质量校准。第二个是代码审查流水线一个Agent写代码一个Agent跑测试一个Agent做安全审查还有一个负责合并建议。第三个是客服工单处理分类Agent先判断工单类型然后路由给对应的处理Agent复杂工单再升级给人工。适合谁来参考如果你已经用过大模型API写过简单的Prompt想从“单次对话”升级到“多步骤自动化”那这篇内容就是给你准备的。如果你还没碰过Agent建议先把工具调用Function Calling和ReAct模式搞清楚再来看协作架构不然容易一头雾水。注意多Agent不是银弹。任务越复杂Agent数量越多通信开销和出错概率也会上升。我见过有人上来就搞七八个Agent结果调试了两周还没跑通一条完整链路。建议从两个Agent开始跑通了再加。2. 协作架构的几种主流模式与选型逻辑2.1 顺序流水线最简单也最常用顺序流水线就是Agent A做完交给Agent BB做完交给C。这种模式的好处是逻辑清晰、调试容易每个Agent的输入输出都是确定的。坏处是没有反馈回路如果A的输出质量差B和C只能跟着错。我一般用这种模式处理“步骤固定、依赖明确”的任务。比如前面说的报告生成检索→分析→撰写→校对每一步的输入就是上一步的输出不需要来回沟通。实现上你可以用一个简单的列表来定义流水线pipeline [ {name: retriever, prompt: 根据主题检索相关资料并整理成结构化数据}, {name: analyzer, prompt: 对结构化数据进行统计分析提取关键指标}, {name: writer, prompt: 根据分析结果撰写报告初稿}, {name: reviewer, prompt: 检查报告的逻辑性和数据准确性输出修改建议} ] context {topic: 某行业季度趋势} for step in pipeline: result call_llm(step[prompt], context) context[step[name] _output] result这段代码的关键在于context字典——它像一个共享白板每个Agent往上面写自己的输出后面的Agent从上面读需要的信息。实际用的时候你需要在每个Prompt里明确告诉Agent“你可以从context里拿到哪些字段”。2.2 层级调度主管Agent分配任务层级调度模式里有一个“主管Agent”它负责理解用户需求、拆解任务、分配给下面的“工人Agent”最后汇总结果。这种模式适合任务类型不固定、需要动态决策的场景。举个例子用户说“帮我分析一下这份销售数据看看哪个区域表现最好再预测下个季度的趋势”。主管Agent需要判断这涉及数据分析和趋势预测两个子任务然后分别调用对应的Agent。主管Agent的Prompt设计是关键。我通常这样写你是一个任务调度主管。你的职责是 1. 理解用户需求判断需要哪些子任务 2. 将子任务分配给对应的工人Agent 3. 收集所有工人的输出整合成最终答案 可用的工人Agent - data_analyst: 擅长数据统计、对比分析 - trend_predictor: 擅长基于历史数据做趋势预测 - report_writer: 擅长将分析结果整理成可读报告 输出格式要求先列出任务拆解再逐个调用工人Agent最后汇总。这种模式的风险在于主管Agent可能拆错任务。我踩过的坑是主管把“预测趋势”拆成了“计算平均值”因为它没理解“预测”和“统计”的区别。解决办法是在主管的Prompt里加几个Few-shot示例明确什么任务该分给谁。2.3 辩论与投票多个Agent互相校验这种模式让多个Agent对同一个问题给出答案然后通过投票或辩论选出最优解。适合需要高准确性、容错率低的场景比如事实核查、代码安全审查。我做过一个实验让三个Agent分别判断一段代码是否有SQL注入风险然后取多数意见。结果发现单个Agent的准确率是78%三个Agent投票后提升到了91%。代价是Token消耗翻了三倍响应时间也变长了。辩论模式更复杂一些Agent A提出观点Agent B反驳Agent C做裁判。这种模式我一般只在争议性大、需要多角度分析的任务里用比如政策解读、竞品分析。2.4 选型对照表模式适用场景优点缺点我的推荐指数顺序流水线步骤固定、依赖明确调试容易、成本低无反馈、错误累积五颗星层级调度任务类型不固定灵活、可扩展主管可能拆错任务四颗星辩论投票高准确性要求容错率高成本高、速度慢三颗星混合模式复杂生产环境兼顾灵活与稳定实现复杂度高四颗星选型的时候我一般问自己三个问题任务步骤是否固定需不需要动态决策错误容忍度有多高三个问题的答案基本就能确定用哪种模式。3. 任务调度的核心机制与实现细节3.1 任务拆解从自然语言到可执行单元任务拆解是多Agent协作里最容易被低估的环节。很多人以为“让大模型拆一下就行了”实际上拆解的质量直接决定了后续所有环节的成败。我用的拆解策略是两层拆解第一层是“意图识别”判断用户到底想要什么第二层是“步骤生成”把意图转化成具体的执行步骤。意图识别可以用一个轻量级的Prompt分析用户输入判断属于以下哪类意图 - 信息检索用户想要查找某些事实或数据 - 分析计算用户想要对数据进行统计或推理 - 内容生成用户想要创作文章、报告、代码等 - 多步复合用户需求包含以上多种类型 用户输入{user_input} 输出格式{intent: 意图类型, confidence: 0.95}步骤生成则要结合可用Agent的能力列表根据意图和可用Agent生成执行步骤 可用Agent - search_agent: 联网检索 - calc_agent: 数据计算 - write_agent: 内容撰写 意图多步复合 用户需求分析某公司财报并写摘要 输出 [ {step: 1, agent: search_agent, task: 检索该公司最新财报数据}, {step: 2, agent: calc_agent, task: 计算营收增长率、利润率等关键指标}, {step: 3, agent: write_agent, task: 根据指标撰写200字摘要} ]实操心得步骤生成的时候一定要让大模型输出结构化的JSON不要输出自然语言。自然语言的步骤描述在后续解析时容易出歧义JSON虽然看起来死板但稳定得多。3.2 状态管理让每个Agent知道“现在到哪了”多Agent协作最头疼的问题之一是状态同步。Agent A完成了任务Agent B怎么知道Agent B执行到一半失败了怎么回滚我的做法是维护一个共享状态对象所有Agent都读写这个对象。状态对象里至少包含这几个字段shared_state { task_id: 唯一标识, current_step: 2, total_steps: 4, step_results: { step_1: {status: completed, output: ...}, step_2: {status: running, output: None} }, errors: [], metadata: {created_at: ..., updated_at: ...} }每个Agent在执行前先读current_step知道自己该做什么执行后更新step_results和current_step。如果出错就往errors里追加一条记录调度器根据错误类型决定是重试还是跳过。这种设计的好处是可追溯。出了问题你看一眼step_results就知道哪个环节卡住了不用去翻日志。3.3 通信协议Agent之间怎么说话Agent之间的通信有两种方式共享内存和消息传递。共享内存就是前面说的共享状态对象适合同一进程内的Agent协作。消息传递则是Agent之间发消息适合分布式部署的场景。我大多数项目用的是共享内存因为实现简单、延迟低。但如果Agent数量超过五个或者需要跨机器部署就会换成消息队列。常用的消息格式是JSON{ from: analyzer_agent, to: writer_agent, type: task_result, payload: { analysis: ..., confidence: 0.87 }, timestamp: 2024-01-15T10:30:00Z }消息传递的坑在于消息丢失和重复。我遇到过Agent B没收到Agent A的消息导致整个流水线卡死。解决办法是加确认机制接收方收到消息后回一个ACK发送方在超时后重发。3.4 超时与重试别让一个Agent拖垮整个系统大模型API的响应时间不稳定有时候几秒有时候几十秒。如果不设超时一个慢请求可能把整个流水线堵死。我的配置是单个Agent调用超时30秒重试2次重试间隔5秒。超过重试次数就标记为失败调度器决定是跳过还是终止。import time def call_agent_with_retry(agent, input_data, max_retries2, timeout30): for attempt in range(max_retries 1): try: result agent.run(input_data, timeouttimeout) return {status: success, output: result} except TimeoutError: if attempt max_retries: time.sleep(5) continue return {status: timeout, output: None} except Exception as e: return {status: error, output: str(e)}注意重试不是万能的。如果Agent是因为输入数据有问题而失败重试只会浪费Token。我一般会在重试前判断错误类型——网络超时可以重试数据格式错误直接跳过。4. 完整实操搭建一个四Agent协同的研究报告生成系统4.1 系统架构与Agent角色定义这个系统包含四个Agent检索Agent、分析Agent、撰写Agent、校对Agent。整体走顺序流水线模式但在校对环节加了一个反馈回路——如果校对不通过打回给撰写Agent重写。每个Agent的角色定义如下Agent名称职责输入输出retriever检索并整理资料研究主题结构化资料列表analyzer数据分析与指标提取结构化资料分析结果JSONwriter撰写报告初稿分析结果报告文本reviewer质量检查与反馈报告文本通过/不通过修改建议4.2 检索Agent的实现细节检索Agent的核心是查询生成和结果过滤。用户给一个主题Agent需要生成多个检索查询然后从返回结果里筛选出相关的。RETRIEVER_PROMPT 你是一个研究资料检索专家。根据用户提供的主题生成3-5个检索查询 每个查询从不同角度覆盖主题。 主题{topic} 输出格式 { queries: [查询1, 查询2, 查询3], focus_areas: [重点领域1, 重点领域2] } def retriever_agent(topic): # 第一步生成查询 queries call_llm(RETRIEVER_PROMPT.format(topictopic)) # 第二步执行检索这里用模拟的检索函数 raw_results [] for q in queries[queries]: results search_engine.search(q, top_k5) raw_results.extend(results) # 第三步过滤和去重 filtered filter_by_relevance(raw_results, topic, threshold0.7) deduplicated deduplicate_by_url(filtered) return { topic: topic, sources: deduplicated[:10], focus_areas: queries[focus_areas] }这里的关键参数是threshold0.7——相关性低于0.7的结果直接丢掉。这个阈值是我试了好几次才定下来的太低会引入噪音太高会漏掉有用信息。4.3 分析Agent的数据处理流程分析Agent拿到检索结果后需要提取关键数据点、计算指标、识别趋势。我一般让它输出结构化的JSON方便后续Agent解析。ANALYZER_PROMPT 你是一个数据分析专家。根据以下资料提取关键数据点并进行分析。 资料 {sources} 分析要求 1. 提取至少5个关键数据点每个数据点包含数值、来源、时间 2. 计算同比/环比变化率如果有历史数据 3. 识别至少2个趋势或模式 4. 标注数据可信度高/中/低 输出格式 { key_metrics: [ {name: 指标名, value: 数值, source: 来源, confidence: 高} ], trends: [ {description: 趋势描述, evidence: 支撑数据} ], data_quality: 整体数据质量评估 } 实操心得分析Agent最容易犯的错误是“编数据”。如果资料里没有某个指标它可能会根据常识推测一个数值。解决办法是在Prompt里明确写“如果资料中没有相关数据标注为‘数据缺失’不要推测”。4.4 撰写Agent与校对Agent的反馈回路撰写Agent根据分析结果写报告校对Agent检查逻辑性、数据准确性和可读性。如果校对不通过撰写Agent根据反馈修改最多循环三次。def writer_reviewer_loop(analysis_result, max_iterations3): draft writer_agent(analysis_result) for i in range(max_iterations): review reviewer_agent(draft) if review[status] approved: return {report: draft, iterations: i 1} # 根据反馈修改 draft writer_agent(analysis_result, feedbackreview[feedback]) return {report: draft, iterations: max_iterations, warning: 达到最大迭代次数}校对Agent的Prompt里我会明确列出检查项检查以下报告逐项判断是否通过 1. 数据准确性报告中的数据是否与分析结果一致 2. 逻辑连贯性段落之间是否有清晰的逻辑关系 3. 可读性语言是否通顺有没有明显的语病 4. 完整性是否覆盖了所有关键分析点 输出格式 { status: approved/rejected, issues: [问题1, 问题2], feedback: 具体的修改建议 }4.5 调度器的完整代码框架把上面四个Agent串起来调度器的核心逻辑如下class MultiAgentOrchestrator: def __init__(self): self.state { task_id: generate_id(), current_step: 0, results: {}, errors: [] } def run(self, topic): try: # Step 1: 检索 self.state[current_step] 1 retrieval retriever_agent(topic) self.state[results][retrieval] retrieval # Step 2: 分析 self.state[current_step] 2 analysis analyzer_agent(retrieval[sources]) self.state[results][analysis] analysis # Step 3: 撰写校对循环 self.state[current_step] 3 final writer_reviewer_loop(analysis) self.state[results][final] final return {status: success, data: final} except Exception as e: self.state[errors].append(str(e)) return {status: failed, state: self.state}这个框架跑下来一个完整的研究报告生成大概需要45-90秒消耗8000-15000个Token具体取决于报告长度和校对迭代次数。5. 踩坑记录与常见问题排查5.1 Agent之间“踢皮球”怎么办我遇到过最诡异的问题是检索Agent返回了空结果分析Agent说“没有数据无法分析”撰写Agent说“没有分析结果无法撰写”最后用户收到一个空报告。排查后发现检索Agent的查询生成有问题——它生成的查询太具体搜索引擎返回了零结果。解决办法是加一个兜底逻辑如果检索结果少于3条自动放宽查询条件重新检索。if len(filtered_results) 3: # 放宽条件去掉限定词用更宽泛的查询 broad_query extract_core_keywords(topic) raw_results search_engine.search(broad_query, top_k10) filtered_results filter_by_relevance(raw_results, topic, threshold0.5)5.2 Token消耗失控的三种情况多Agent系统的Token消耗是单Agent的3-5倍如果不加控制成本会很难看。我总结的三种失控情况第一种是上下文无限增长。每个Agent都把完整的历史记录传给下一个Agent导致Prompt越来越长。解决办法是只传必要字段不要传完整对话历史。第二种是校对循环不收敛。撰写Agent和校对Agent来回改改了十几次还没通过。解决办法是设最大迭代次数超过就强制输出并标记警告。第三种是重复检索。多个Agent都需要同一份数据各自去检索了一遍。解决办法是加缓存层相同查询直接返回缓存结果。问题类型表现解决方案预计节省上下文膨胀Prompt长度持续增长只传必要字段40-60%循环不收敛校对迭代超过5次设最大迭代次数20-30%重复检索相同查询多次执行加缓存层15-25%5.3 输出格式不稳定的处理技巧大模型输出JSON的时候偶尔会多一个逗号、少一个引号导致解析失败。我试过三种解决办法第一种是用JSON Schema约束输出。现在很多大模型API支持response_format参数直接指定JSON Schema输出格式会稳定很多。第二种是加解析容错。用正则表达式提取JSON部分然后用json.loads的strictFalse模式解析。第三种是让模型自己修复。解析失败时把错误信息和原始输出一起发给模型让它重新输出正确的JSON。def safe_json_parse(text): try: return json.loads(text) except json.JSONDecodeError: # 尝试提取JSON部分 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group(), strictFalse) except: pass # 让模型修复 fixed call_llm(f以下JSON格式有误请修复{text}) return json.loads(fixed)5.4 常见问题速查表现象可能原因排查步骤解决方法流水线卡死某个Agent超时未返回检查各Agent的响应时间加超时和重试机制输出质量差Prompt不够具体检查Prompt是否包含明确要求加Few-shot示例数据不一致多个Agent数据源不同检查是否共用同一份数据统一数据源成本过高Token消耗失控统计各Agent的Token用量加缓存、精简Prompt循环不结束校对标准太模糊检查校对Prompt的通过条件明确通过标准最后分享一个小技巧调试多Agent系统的时候我会给每个Agent的输出加一个debug字段记录它的输入摘要、输出摘要和耗时。这样出问题的时候一眼就能看出是哪个环节的锅不用去翻完整的日志。这套多Agent协作架构我前后迭代了四个版本从最初的两个Agent顺序流水线到现在支持动态调度的混合模式。最大的体会是不要追求一步到位。先把最简单的流水线跑通再逐步加Agent、加反馈回路、加容错机制。每加一个东西都要问自己“它解决了什么具体问题”如果答不上来就不加。