1. 多Agent协作到底在解决什么问题
单Agent跑任务，跑到一定复杂度就会撞墙。我最早做文档分析流水线的时候，一个Agent既要读文件、又要抽实体、还要生成摘要、最后还得做质量校验，提示词写到三千字还是压不住——它会在某个环节“忘记”前面的约束，或者把抽取阶段的任务和校验阶段的任务搅在一起。这不是模型不够强，而是单点上下文里塞了太多互相冲突的目标。
多Agent协作的核心思路很朴素：把一个大任务拆成若干职责单一的子任务，每个子任务交给一个独立的Agent，Agent之间通过消息传递来协同。这跟软件工程里拆微服务是一个道理——不是技术炫技，而是为了可控、可调试、可替换。
这套东西能做什么？举几个我实际跑过的场景：
- 科研论文辅助流水线：一个Agent负责研究定位与文献梳理，一个负责数据分析与图表生成，一个负责初稿撰写，最后一个专门做质量校准（查逻辑漏洞、查数据引用一致性）。四个Agent串起来，比单个Agent硬写质量高出一大截。
- 代码开发协同：规划Agent拆需求，编码Agent写实现，审查Agent做静态检查和边界测试，三个角色互相制衡。
- 复杂信息抽取：抽取Agent负责召回，验证Agent负责过滤幻觉，汇总Agent负责结构化输出。
适合谁来参考？如果你已经能跑通单Agent的调用，想往复杂任务编排走，这篇就是给你写的。如果你还没接触过Agent，建议先把单Agent的提示词工程和工具调用搞明白再回来。
注意：多Agent不是银弹。任务本身如果三步就能搞定，硬拆成三个Agent只会增加延迟和调试成本。我踩过的最大坑就是“为了多Agent而多Agent”。
2. 协作架构的四种主流形态与选型逻辑
2.1 顺序流水线架构：最稳但最死板
顺序架构就是Agent A输出给Agent B，B输出给C，像工厂流水线。它的优点是链路清晰、每一步可单独测试、失败点容易定位。缺点是没有反馈回路，前一步错了后面全错，而且整体延迟是各步之和。
我做过一个合同信息抽取的项目，用的就是顺序架构：OCR清洗Agent → 条款切分Agent → 要素抽取Agent → 格式校验Agent。为什么选顺序而不是别的？因为合同处理天然是线性的，条款切分必须在抽取之前完成，没有并行的必要。实测下来，四步串行总耗时约12秒，其中抽取Agent占了7秒，是瓶颈。
选型判断标准很简单：如果子任务之间存在严格的输入输出依赖，且不需要回头修正，就用顺序架构。
2.2 层级调度架构：一个主管带一队执行者
层级架构里有一个调度Agent（Orchestrator），它不干具体活，只负责拆解任务、分配任务、收集结果、决定下一步。下面挂若干执行Agent（Worker），每个Worker只擅长一件事。
这种架构的好处是调度逻辑和执行逻辑分离。调度Agent的提示词只需要关注“怎么拆、怎么派、怎么判断完成”，执行Agent的提示词只需要关注“怎么把这一件事做好”。两者互不干扰，调试的时候可以分别优化。
我见过一个比较典型的实现：调度Agent维护一个任务队列，每个任务带状态（pending / running / done / failed）。它根据任务类型路由到对应的Worker，Worker完成后回调更新状态，调度Agent检查是否所有任务都done，是则汇总输出。
实操心得：调度Agent的提示词里一定要明确“什么情况下任务算完成”。我早期没写清楚，调度Agent会在Worker已经给出完整答案后还反复追问，白白烧token。
2.3 对等协商架构：Agent之间互相辩论
对等架构没有主管，Agent之间平级通信，可以互相提问、质疑、补充。最典型的玩法是辩论式校验：两个Agent对同一份数据分别给出结论，如果结论不一致，就进入辩论轮次，各自陈述理由，直到达成一致或达到最大轮次。
这种架构在质量校准场景特别有用。比如论文写作流水线里，撰写Agent写完一段，校准Agent会逐条质疑：“这里的数据引用和第三段矛盾”“这个结论缺少对照组支撑”。撰写Agent收到质疑后修正，再交回校准Agent复核。
代价是通信开销大、轮次不可控。我建议一定要设最大辩论轮次（通常2-3轮足够），否则两个Agent可能无限扯皮。另外，辩论的收敛条件要写死，比如“连续两轮无新增质疑则终止”。
2.4 混合架构：实际项目里最常见
纯顺序、纯层级、纯对等都不多见，真实项目往往是混合的。比如外层是层级调度，某个Worker内部又跑一个顺序流水线，关键校验环节再插入一个对等辩论。
我的经验是：先用层级架构搭骨架，把调度和执行分开；然后在需要质量把关的节点插入对等校验；最后把某些高频调用的Worker内部做成顺序流水线以降低调度开销。这样既保证了整体可控，又兼顾了局部灵活性。
| 架构类型 | 适用场景 | 优点 | 缺点 | 我的推荐指数 |
|---|---|---|---|---|
| 顺序流水线 | 线性依赖任务 | 链路清晰、易调试 | 无反馈、延迟叠加 | 四星 |
| 层级调度 | 任务可并行拆分 | 职责分离、可扩展 | 调度Agent易成瓶颈 | 五星 |
| 对等协商 | 质量校验、争议裁决 | 结论更可靠 | 开销大、收敛难 | 三星 |
| 混合架构 | 复杂真实项目 | 兼顾可控与灵活 | 设计复杂度高 | 五星 |
3. 任务调度的核心机制拆解
3.1 任务拆解：从一句话需求到可执行子任务
调度Agent拿到的往往是一句模糊的需求，比如“帮我分析这份销售数据并给出建议”。它需要把这句话拆成可执行的子任务。拆解的质量直接决定后续所有环节的成败。
我的做法是给调度Agent一套拆解模板，强制它按固定维度输出：
{ "task_id": "t001", "goal": "分析销售数据并给出建议", "subtasks": [ {"id": "s1", "type": "data_load", "desc": "读取并清洗销售数据", "depends_on": []}, {"id": "s2", "type": "analysis", "desc": "按区域和品类做趋势分析", "depends_on": ["s1"]}, {"id": "s3", "type": "insight", "desc": "基于分析结果生成业务建议", "depends_on": ["s2"]}, {"id": "s4", "type": "verify", "desc": "校验建议是否有数据支撑", "depends_on": ["s3"]} ] }关键点是depends_on字段。有了依赖关系，调度器就知道哪些任务可以并行、哪些必须等待。s2和s3有依赖必须串行，但如果再加一个“生成可视化图表”的子任务，它只依赖s1，就可以和s2并行跑。
注意事项：拆解粒度不要太细。我试过把“读取数据”拆成“打开文件”“解析表头”“逐行读取”三个子任务，结果调度开销比执行开销还大。一般一个子任务对应一个Agent的一次完整调用比较合适。
3.2 依赖管理与并行执行
依赖管理说白了就是拓扑排序。把所有子任务按depends_on画成有向无环图，然后逐层执行：第一层是没有依赖的任务，可以全部并行；第二层依赖第一层的任务，等第一层全部完成后并行执行；以此类推。
我用Python实现过一个简易调度器，核心逻辑大概是这样：
import asyncio from collections import defaultdict async def run_dag(subtasks, agent_pool): # 构建依赖图和入度表 graph = defaultdict(list) indegree = {t["id"]: 0 for t in subtasks} task_map = {t["id"]: t for t in subtasks} for t in subtasks: for dep in t["depends_on"]: graph[dep].append(t["id"]) indegree[t["id"]] += 1 # 逐层执行 ready = [tid for tid, deg in indegree.items() if deg == 0] results = {} while ready: # 当前层全部并行 layer_results = await asyncio.gather(*[ agent_pool.execute(task_map[tid], results) for tid in ready ]) for tid, res in zip(ready, layer_results): results[tid] = res # 更新入度，找出下一层 next_ready = [] for tid in ready: for nxt in graph[tid]: indegree[nxt] -= 1 if indegree[nxt] == 0: next_ready.append(nxt) ready = next_ready return results这段代码的核心是按层并行。同一层的任务互不依赖，可以同时发起调用，总延迟等于最慢那个任务的时间，而不是所有任务之和。实测在一个8子任务的项目里，并行执行比串行快了将近3倍。
3.3 状态传递：Agent之间怎么“交接工作”
Agent之间传递的不只是文本，还包括结构化状态。我一般把状态分成三类：
- 原始输入：用户需求、原始数据，全程只读。
- 中间产物：各Agent的输出，按task_id索引，后续Agent按需读取。
- 全局上下文：所有Agent共享的背景信息，比如“这是一份2024年Q3的销售数据”“输出格式要求是Markdown表格”。
传递方式有两种：全量传递和按需传递。全量传递是把所有中间产物都塞给下一个Agent，简单但浪费上下文窗口；按需传递是只给下一个Agent它depends_on的那些任务的输出，精准但需要调度器做路由。
我推荐按需传递。上下文窗口是稀缺资源，尤其是处理长文档的时候，把无关的中间产物塞进去只会稀释模型的注意力。调度器在派发任务时，根据depends_on字段把上游输出拼进提示词就行。
实操心得：中间产物最好做一次摘要压缩再传递。比如上游Agent输出了一千字的分析，传给下游时压缩成两百字的关键结论加数据引用，既保留了信息又省了token。我一般让上游Agent在输出末尾附一个“给下游的摘要”字段。
3.4 失败重试与降级策略
Agent调用失败是常态——超时、格式错误、幻觉导致输出不可解析，都会发生。调度器必须有一套失败处理机制。
我的策略是三级处理：
- 自动重试：格式错误这类问题，把错误信息附在提示词里重试一次，成功率能到70%以上。
- 降级执行：重试仍失败，换一个更简单的提示词或更小的模型兜底。比如抽取任务失败，降级成“只输出你能确定的部分，不确定的标注为unknown”。
- 人工介入：关键任务连续失败，挂起并通知人工处理，不要让错误往下游传播。
重试次数我一般设2次，超过就降级。因为Agent调用有成本，无限重试既烧钱又拖时间。
4. 从零搭建一个多Agent协同任务的完整实操
4.1 环境准备与基础框架选型
先说环境。我用的技术栈是Python 3.10 + asyncio + 一个大模型API。框架层面，早期我用过几个开源的多Agent框架，后来发现自己写调度器反而更可控，因为框架的抽象层往往和实际需求对不上，改起来比自己写还费劲。
如果你不想从零写，几个方向可以参考：AgentScope这类框架提供了Agent定义、消息传递、流程编排的基础能力；LangGraph适合把Agent流程画成图来管理状态；AutoGen在对话式多Agent场景比较成熟。选哪个取决于你的任务形态——流程固定的用LangGraph，对话协商多的用AutoGen。
我的建议是：先用框架跑通一个最小demo，理解多Agent的通信模式，然后根据项目需求决定是继续用框架还是自己写调度层。
基础依赖就几个：
pip install openai asyncio aiohttp pydanticpydantic用来做输出格式校验，这个后面会讲为什么重要。
4.2 定义Agent角色与提示词模板
每个Agent本质上就是一段系统提示词 + 一个模型调用 + 一套输出格式约束。我一般用一个类来封装：
class Agent: def __init__(self, name, system_prompt, output_schema=None): self.name = name self.system_prompt = system_prompt self.output_schema = output_schema async def execute(self, task_desc, context): messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": f"任务：{task_desc}\n\n上下文：{context}"} ] response = await call_llm(messages) if self.output_schema: return validate_and_parse(response, self.output_schema) return response系统提示词我遵循一个模板：角色定义 + 职责边界 + 输出格式 + 禁止事项。举个例子，一个“数据校验Agent”的提示词：
你是一个数据校验专家。你的唯一职责是检查上游分析结论是否有数据支撑。 你不负责生成新结论，只负责指出问题。 输出格式：JSON数组，每个元素包含 {"issue": "问题描述", "severity": "high/medium/low", "suggestion": "修正建议"} 禁止事项：不要输出与校验无关的内容；不要臆造数据。注意事项：职责边界一定要写死。我踩过的坑是校验Agent开始“顺手”帮上游改结论，结果两个Agent的输出混在一起，根本分不清谁对谁错。
4.3 调度器实现：任务分发与结果回收
调度器是整个系统的中枢。它的核心职责是：接收任务、拆解、按依赖关系派发、收集结果、判断完成。
我实现过一个简化版调度器，核心流程如下：
class Orchestrator: def __init__(self, agents): self.agents = agents # {agent_name: Agent} async def run(self, user_request): # 第一步：拆解任务 plan = await self.decompose(user_request) # 第二步：按DAG执行 results = await self.execute_dag(plan) # 第三步：汇总输出 final = await self.aggregate(results) return final async def decompose(self, request): prompt = f"将以下需求拆解为子任务，输出JSON格式：{request}" return await call_llm_with_schema(prompt, PLAN_SCHEMA) async def execute_dag(self, plan): # 拓扑排序 + 并行执行，逻辑同3.2节 ... async def aggregate(self, results): prompt = f"汇总以下子任务结果，生成最终输出：{results}" return await call_llm(prompt)这里有个关键设计：拆解和汇总用的是同一个调度Agent，但提示词不同。拆解时它关注“怎么分”，汇总时它关注“怎么合”。分开写提示词比用一个通用提示词效果好得多。
4.4 输出格式校验与结构化解析
多Agent系统里，格式错误是最常见的失败原因。上游Agent输出了一段自然语言，下游Agent解析不了，整条链路就断了。
我的做法是强制所有Agent输出JSON，并且用pydantic做校验：
from pydantic import BaseModel, ValidationError class AnalysisResult(BaseModel): summary: str key_findings: list[str] data_references: list[str] confidence: float def validate_and_parse(raw_output, schema): try: # 提取JSON部分（模型有时会在JSON前后加解释文字） json_str = extract_json(raw_output) return schema.model_validate_json(json_str) except ValidationError as e: # 校验失败，返回错误信息供重试 return {"error": str(e), "raw": raw_output}校验失败时，调度器会把错误信息附在重试提示词里，让Agent重新输出。实测这个机制能把格式错误率从30%降到5%以下。
实操心得：提示词里一定要给输出示例。光说“输出JSON”不够，模型不知道字段名和结构。给一个完整的示例，格式正确率会大幅提升。
4.5 完整跑通一个论文辅助流水线
我用这套框架搭过一个论文辅助流水线，四个Agent：
- 定位Agent：输入研究主题，输出研究问题、目标期刊、创新点方向。
- 分析Agent：输入研究问题，输出数据分析方案和预期图表。
- 撰写Agent：输入前两步结果，输出论文初稿。
- 校准Agent：输入初稿，输出问题清单和修正建议。
调度流程是：定位 → 分析 → 撰写 → 校准 → 如果校准有问题则回到撰写修正，最多循环2轮。
实测下来，一篇8000字的论文初稿，四个Agent跑完约需6-8分钟，token消耗约15万。质量上，校准Agent平均能找出8-12个问题，其中约60%是真实存在的逻辑或引用问题。这个比例比单Agent自检高不少，因为校准Agent没有“自己写的东西舍不得改”的偏见。
5. 常见问题与排查技巧实录
5.1 Agent之间“踢皮球”怎么办
现象：调度Agent把任务派给Worker A，A说“这不是我的职责”，派给Worker B，B也说“不该我管”，任务卡死。
根因：职责边界模糊。调度Agent的拆解粒度和Worker的职责定义对不上。
解决：在调度Agent的提示词里明确每个Worker的能力清单，拆解时只生成能力清单内的任务类型。同时给Worker加一个兜底规则：“如果任务不属于你的职责，输出{“status”: “reject”, “reason”: “...”}，不要自行处理”。
5.2 上下文爆炸：token消耗失控
现象：跑一个任务烧了几十万token，成本远超预期。
根因：全量传递中间产物，每个Agent都拿到所有上游输出，上下文越滚越大。
解决：按需传递 + 摘要压缩。调度器只把depends_on的直接上游输出传给下游，并且要求上游在输出末尾附一个200字以内的摘要供传递用。我实测这个改动能把token消耗降低60%以上。
5.3 死循环：两个Agent无限辩论
现象：撰写Agent和校准Agent来回修改，永远达不成一致。
根因：没有设置终止条件。
解决：三重保险——最大轮次限制（硬性2-3轮）、收敛判断（连续两轮无新增问题则终止）、超时熔断（单任务超过N秒强制结束并输出当前最优结果）。
5.4 输出格式反复出错
现象：某个Agent总是输出格式不对，重试多次仍失败。
根因：提示词里的格式说明不够具体，或者模型能力不足以稳定输出复杂结构。
解决：给完整示例 + 用pydantic校验 + 降级策略（格式实在出不来就退化成纯文本输出，由下游做容错解析）。
| 问题类型 | 典型现象 | 根因 | 解决手段 |
|---|---|---|---|
| 踢皮球 | 任务无人认领 | 职责边界模糊 | 能力清单 + reject机制 |
| 上下文爆炸 | token消耗失控 | 全量传递 | 按需传递 + 摘要压缩 |
| 死循环 | 无限辩论 | 无终止条件 | 轮次限制 + 收敛判断 + 超时熔断 |
| 格式错误 | 解析失败 | 提示词不具体 | 示例 + 校验 + 降级 |
5.5 独家避坑技巧汇总
几个我踩过坑之后总结的硬经验：
- 调度Agent用强模型，Worker可以用弱模型。调度需要理解复杂依赖关系，Worker只需要执行单一任务，用便宜模型完全够。
- 每个Agent的输出都加一个confidence字段。下游可以根据置信度决定是否采信，低置信度的结果触发人工复核。
- 日志要记全。每次Agent调用的输入、输出、耗时、token消耗都记下来，出问题的时候能快速定位是哪个环节。
- 先串行跑通再改并行。并行虽然快，但调试难度大。我一般先用串行验证逻辑正确，再改成并行优化速度。
- 给每个Agent设超时。单个Agent调用超过60秒直接判失败，不要让整个流水线卡在一个Agent上。
这套多Agent协作的架子搭起来之后，后面加新Agent、改调度逻辑都很方便。我现在的做法是把Agent定义和调度逻辑分开管理，Agent像插件一样注册进调度器，需要什么能力就挂什么Agent。这个内容后续还可以往Agent能力自动发现、动态路由、基于历史表现的Agent选择这些方向扩展，等我把动态路由跑稳定了再单独写一篇。
