为什么需要多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 skippeddataclassclass SubTask: id: str description: str agent_type: str # 指定哪种类型的Agent处理 dependencies: list[str] field(default_factorylist) # 依赖的其他子任务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), timeouttask.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 第{i1}/{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 BAgent 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并行调用LLMtoken消耗是单Agent的N倍。解决- 对简单子任务使用小模型如GPT-4o mini- 实现任务级别的token预算控制- 缓存相同任务的结果—## 总结多Agent系统的设计哲学是用架构复杂性换取任务处理能力的量变到质变。它适合处理单Agent无法完成的大规模复杂任务但不适合简单任务过度工程化。实践建议1. 先用单Agent解决问题只有在单Agent明显不够时才引入多Agent2. 从主从架构开始这是最容易理解和调试的模式3. 重视任务分解的质量——好的分解是成功的一半4. 务必实现循环检测和超时机制防止系统失控