多Agent协作系统设计2026：从任务分解到结果聚合的工程实践

为什么需要多Agent协作

单个Agent在处理复杂任务时面临天然的局限：1.上下文窗口有限：一个需要分析10万行代码库的任务，单Agent无法在一次对话中完成2.并行能力缺失：需要同时进行多个独立子任务时，单Agent只能串行处理3.专业化不足：通用Agent在特定领域（数学推理、代码生成、数据分析）的表现往往不如专门调优的专业Agent4.可靠性瓶颈：单点失败会导致整个任务失败，缺乏容错机制多Agent协作系统通过任务分解、并行执行和结果聚合，解决上述问题——代价是更高的系统复杂度和协调成本。本文重点讨论多Agent系统中最关键的工程挑战及其解决方案。—## 核心架构模式### 模式一：主从架构（Orchestrator-Worker）最常用的多Agent架构：一个Orchestrator Agent负责规划和协调，多个Worker Agent负责具体执行：用户任务 ↓Orchestrator（任务分解、规划、结果聚合） ├─── Worker A（专注代码生成） ├─── Worker B（专注数据分析） └─── Worker C（专注文档检索）### 模式二：流水线架构（Pipeline）任务在Agent之间顺序流转，每个Agent处理并传递给下一个：输入 → Agent1（数据清洗）→ Agent2（分析）→ Agent3（报告生成）→ 输出### 模式三：辩证架构（Debate）多个Agent对同一问题给出不同视角，最后综合：问题 ─┬─ Agent1（支持视角）──┐ ├─ Agent2（反对视角）──┼─ Synthesizer → 最终答案 └─ Agent3（中立分析）──┘—## 任务分解：把大问题变成小问题pythonfrom dataclasses import dataclass, fieldfrom enum import Enumfrom typing import Anyclass TaskStatus(Enum): PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" SKIPPED = "skipped"@dataclassclass SubTask: id: str description: str agent_type: str # 指定哪种类型的Agent处理 dependencies: list[str] = field(default_factory=list) # 依赖的其他子任务ID status: TaskStatus = TaskStatus.PENDING result: Any = None error: str = None timeout: int = 120 # 秒class TaskDecomposer: """使用LLM将复杂任务分解为子任务有向无环图（DAG）""" def __init__(self, llm_client): self.llm = llm_client async def decompose(self, task_description: str) -> list[SubTask]: prompt = f"""将以下任务分解为可以并行或串行执行的子任务：任务：{task_description}可用的Agent类型：- code_agent: 代码生成、调试和优化- search_agent: 网络搜索和信息检索- analysis_agent: 数据分析和图表生成- write_agent: 文档写作和内容创作输出JSON格式：[ {{ "id": "task_001", "description": "具体任务描述", "agent_type": "code_agent", "dependencies": [] // 空表示可以立即开始，非空表示需要等待依赖完成 }}, ...]原则：1. 尽可能并行化（无依赖关系的任务可以同时执行）2. 每个子任务应该足够原子（单个Agent一次对话可以完成）3. 子任务数量不超过10个""" response = await self.llm.generate(prompt) import json task_defs = json.loads(response) return [SubTask(**td) for td in task_defs] def build_execution_plan(self, subtasks: list[SubTask]) -> list[list[SubTask]]: """将子任务组织成可以并行执行的批次""" completed = set() batches = [] remaining = list(subtasks) while remaining: # 找出所有依赖已满足的任务 ready = [ t for t in remaining if all(dep in completed for dep in t.dependencies) ] if not ready: # 存在循环依赖 raise ValueError(f"循环依赖检测：{[t.id for t in remaining]}") batches.append(ready) completed.update(t.id for t in ready) remaining = [t for t in remaining if t not in ready] return batches—## 并行执行引擎pythonimport asynciofrom typing import Callableclass ParallelExecutionEngine: """并行执行多个Agent任务""" def __init__( self, agent_factory: Callable, max_concurrent: int = 5 ): self.agent_factory = agent_factory self.semaphore = asyncio.Semaphore(max_concurrent) async def execute_batch( self, tasks: list[SubTask], context: dict # 共享上下文（之前完成任务的结果） ) -> dict[str, SubTask]: """并行执行一批无依赖关系的任务""" async def execute_single(task: SubTask) -> SubTask: async with self.semaphore: task.status = TaskStatus.RUNNING try: agent = self.agent_factory(task.agent_type) # 注入依赖任务的结果 enriched_context = { **context, "dependency_results": { dep_id: context.get(dep_id) for dep_id in task.dependencies } } result = await asyncio.wait_for( agent.execute(task.description, enriched_context), timeout=task.timeout ) task.result = result task.status = TaskStatus.COMPLETED print(f" ✅ [{task.id}] 完成") except asyncio.TimeoutError: task.status = TaskStatus.FAILED task.error = f"超时 (>{task.timeout}s)" print(f" ❌ [{task.id}] 超时") except Exception as e: task.status = TaskStatus.FAILED task.error = str(e) print(f" ❌ [{task.id}] 失败: {e}") return task # 并发执行所有任务 print(f"\n并行执行 {len(tasks)} 个任务...") results = await asyncio.gather(*[execute_single(t) for t in tasks]) return {t.id: t for t in results} async def execute_dag(self, subtasks: list[SubTask]) -> dict: """按DAG顺序执行所有子任务""" decomposer = TaskDecomposer(None) batches = decomposer.build_execution_plan(subtasks) all_results = {} for i, batch in enumerate(batches): print(f"\n=== 第{i+1}/{len(batches)}批，共{len(batch)}个任务 ===") batch_results = await self.execute_batch(batch, all_results) all_results.update(batch_results) # 检查是否有关键任务失败 failed = [t for t in batch if t.status == TaskStatus.FAILED] if failed: print(f"警告：{len(failed)}个任务失败，继续执行后续任务...") return all_results—## 结果聚合：把碎片化结果合并成完整答案pythonclass ResultAggregator: """聚合多Agent的执行结果""" def __init__(self, llm_client): self.llm = llm_client async def aggregate( self, original_task: str, subtask_results: dict[str, SubTask] ) -> str: """综合所有子任务结果，生成最终答案""" # 构建结果摘要 results_summary = [] for task_id, task in subtask_results.items(): if task.status == TaskStatus.COMPLETED: results_summary.append( f"子任务[{task_id}]（{task.description[:50]}）：\n{task.result}" ) else: results_summary.append( f"子任务[{task_id}]（{task.description[:50]}）：⚠️ 执行失败 - {task.error}" ) aggregation_prompt = f"""原始任务：{original_task}以下是各子任务的执行结果：{chr(10).join(results_summary)}请综合以上所有结果，给出对原始任务的完整、连贯的回答。要求：1. 整合各部分信息，消除重复2. 明确指出哪些部分因子任务失败而信息不完整3. 按逻辑顺序组织，而不是按子任务顺序堆砌""" return await self.llm.generate(aggregation_prompt) def create_execution_report(self, subtask_results: dict[str, SubTask]) -> dict: """生成执行报告，供调试和监控使用""" total = len(subtask_results) completed = sum(1 for t in subtask_results.values() if t.status == TaskStatus.COMPLETED) failed = sum(1 for t in subtask_results.values() if t.status == TaskStatus.FAILED) return { "total_tasks": total, "completed": completed, "failed": failed, "success_rate": f"{completed/total:.1%}", "failures": [ {"id": t.id, "description": t.description, "error": t.error} for t in subtask_results.values() if t.status == TaskStatus.FAILED ] }—## 常见陷阱与解决方案### 陷阱一：Agent之间的信息孤岛问题：Worker Agent之间无法共享中间结果，Orchestrator成为通信瓶颈。解决：建立共享的任务上下文（Task Context），允许Worker读取其他Worker的结果。### 陷阱二：循环调用（Agent Loop）问题：Agent A调用Agent B，Agent B又调用回Agent A，陷入无限循环。解决：实现调用链追踪，检测循环并主动终止：pythonclass CallChainTracker: def __init__(self, max_depth: int = 10): self.call_stack = [] self.max_depth = max_depth def enter(self, agent_id: str): if agent_id in self.call_stack: raise ValueError(f"检测到循环调用：{' -> '.join(self.call_stack)} -> {agent_id}") if len(self.call_stack) >= self.max_depth: raise ValueError(f"调用深度超过限制（{self.max_depth}层）") self.call_stack.append(agent_id) def exit(self, agent_id: str): self.call_stack.remove(agent_id)### 陷阱三：成本失控问题：多Agent并行调用LLM，token消耗是单Agent的N倍。解决：- 对简单子任务使用小模型（如GPT-4o mini）- 实现任务级别的token预算控制- 缓存相同任务的结果—## 总结多Agent系统的设计哲学是：用架构复杂性换取任务处理能力的量变到质变。它适合处理单Agent无法完成的大规模复杂任务，但不适合简单任务（过度工程化）。实践建议：1. 先用单Agent解决问题，只有在单Agent明显不够时才引入多Agent2. 从主从架构开始，这是最容易理解和调试的模式3. 重视任务分解的质量——好的分解是成功的一半4. 务必实现循环检测和超时机制，防止系统失控