系列目录：本文是「AI 应用开发进阶实战」系列的最终篇。前四篇我们构建了 RAG、MCP、Graph RAG 和工作流引擎，本篇将这些能力组合起来，构建一个多 Agent 协作系统。

---

一、为什么需要多 Agent？

1.1 单 Agent 的瓶颈

# 单 Agent 面对复杂任务时的表现
user_task = """
请分析这个 GitHub 仓库的代码质量，找出潜在的安全漏洞，
对比行业最佳实践，然后生成一份改进报告（含优先级排序）。
"""

# 单 Agent 做的事（串行）：
# 1. 读取所有代码文件        (30s)
# 2. 分析代码质量            (20s)
# 3. 扫描安全漏洞            (40s)
# 4. 查询行业最佳实践        (15s)
# 5. 对比分析                (20s)
# 6. 生成报告                (15s)
# 总计：~140s，context 窗口可能溢出

1.2 多 Agent 的解法

同一个任务，用 4 个专业 Agent 并行处理：
             ┌─→ CodeAnalyst: 代码质量分析 ─────┐
Coordinator ─┼─→ SecurityScanner: 安全扫描 ──────┼─→ Reporter: 汇总报告
             ├─→ BestPractice: 最佳实践对照 ─────┤
             └─→ DependencyChecker: 依赖审计 ────┘

耗时：最慢的子任务（40s）+ 汇总（15s）= ~55s（节省 60%）
且每个 Agent 的 context 干净、专注，不易出错

多 Agent 的核心优势：

维度	单 Agent	多 Agent
速度	串行	并行，瓶颈在最慢节点
质量	一个大脑全包	专业化分工，各司其职
上下文	容易溢出	各自专注，互不干扰
可靠性	单点故障	子任务失败可重试/替换
可扩展	难以水平扩展	加 Agent 即可加容量

---

二、四种协作模式

2.1 顺序模式（Pipeline）

A → B → C

适用：线性流水线，如"翻译 → 润色 → 排版"

2.2 并行模式（Map-Reduce）

       ┌→ Worker 1 ─┐
Coord  ├→ Worker 2 ─┼→ Reducer
       └→ Worker 3 ─┘

适用：独立子任务，如"分析三个竞品，汇总对比"

2.3 辩论模式（Debate）

       ┌─→ Agent A（正方）─┐
Problem├─→ Agent B（反方）─┼→ Judge → 结论
       └─→ Agent C（裁判）─┘

适用：需要多角度审视的决策，如"选择技术方案"

2.4 层级模式（Hierarchy）

           Leader
         /   |   \
    PM-A   PM-B   PM-C
    /  \   /  \   /  \
  Dev1 Dev2  ...   DevN

适用：大型项目，如"开发一个完整系统"

---

三、核心实现

3.1 任务分解器

# multi_agent_system.py
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Callable
from openai import OpenAI
import asyncio
import json
import time


@dataclass
class SubTask:
    """子任务定义"""
    id: str
    description: str
    agent_role: str           # 需要哪种 Agent："coder", "analyst", "reviewer"
    inputs: Dict[str, Any]    # 任务输入
    dependencies: List[str] = field(default_factory=list)  # 依赖的子任务ID


@dataclass  
class AgentDefinition:
    """Agent 定义"""
    name: str
    role: str
    system_prompt: str
    model: str = "gpt-4o"     # 不同角色可以用不同模型！
    tools: List[str] = field(default_factory=list)


class TaskDecomposer:
    """将复杂任务分解为子任务"""
    
    def __init__(self, api_key: str):
        self.llm = OpenAI(api_key=api_key)
    
    def decompose(self, task: str, available_agents: List[AgentDefinition]) -> List[SubTask]:
        """LLM 自动分解任务"""
        
        agent_descriptions = "\n".join([
            f"- {a.name} ({a.role}): {a.system_prompt[:100]}"
            for a in available_agents
        ])
        
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": f"""你是一个任务分解专家。将用户的复杂任务分解为多个子任务。

可用 Agent 及其专长：
{agent_descriptions}

规则：
1. 尽可能让子任务并行执行（减少依赖关系）
2. 每个子任务分配给最合适的 Agent
3. 子任务粒度适中：太大失去并行优势，太小增加通信开销
4. 输出严格 JSON 格式

输出格式：
{{
  "subtasks": [
    {{
      "id": "task_1",
      "description": "具体任务描述，包含需要的上下文信息",
      "agent_role": "agent名称",
      "dependencies": []  // 依赖的子任务 ID 列表
    }}
  ]
}}"""
            }, {
                "role": "user",
                "content": f"请分解以下任务：\n{task}"
            }],
            temperature=0.2
        )
        
        raw = response.choices[0].message.content
        if raw.startswith("```"):
            raw = raw.split("\n", 1)[1].rsplit("```", 1)[0]
        
        plan = json.loads(raw)
        
        subtasks = []
        for item in plan["subtasks"]:
            subtasks.append(SubTask(
                id=item["id"],
                description=item["description"],
                agent_role=item["agent_role"],
                dependencies=item.get("dependencies", [])
            ))
        
        print(f"Decomposed into {len(subtasks)} subtasks:")
        for st in subtasks:
            deps = f" (depends: {st.dependencies})" if st.dependencies else ""
            print(f"  [{st.id}] {st.agent_role}: {st.description[:60]}...{deps}")
        
        return subtasks

3.2 Agent 运行时

class AgentRuntime:
    """单个 Agent 的运行时"""
    
    def __init__(self, definition: AgentDefinition, api_key: str):
        self.def = definition
        self.llm = OpenAI(api_key=api_key)
        self.conversation_history: List[dict] = []
    
    async def execute(self, task: str, context: Dict[str, Any] = None) -> str:
        """执行一个子任务"""
        
        messages = [
            {"role": "system", "content": self.def.system_prompt},
        ]
        
        # 如果有上下文（来自依赖任务的输出）
        if context:
            ctx_text = "\n".join([
                f"[{task_id} 的输出]:\n{output[:500]}"
                for task_id, output in context.items()
            ])
            messages.append({
                "role": "system",
                "content": f"相关上下文（来自已完成的任务）：\n{ctx_text}"
            })
        
        messages.append({"role": "user", "content": task})
        
        # 如果需要工具调用，用 Function Calling 循环
        if self.def.tools:
            return await self._execute_with_tools(messages)
        
        response = self.llm.chat.completions.create(
            model=self.def.model,
            messages=messages,
            temperature=0.3
        )
        
        result = response.choices[0].message.content
        print(f"  [{self.def.name}] completed ({len(result)} chars)")
        return result
    
    async def _execute_with_tools(self, messages: List[dict]) -> str:
        """带工具调用的执行循环"""
        max_steps = 10
        
        for step in range(max_steps):
            response = self.llm.chat.completions.create(
                model=self.def.model,
                messages=messages,
                tools=self._get_tool_schemas(),
                tool_choice="auto"
            )
            
            msg = response.choices[0].message
            messages.append(msg)
            
            if msg.tool_calls:
                for tc in msg.tool_calls:
                    result = await self._execute_tool(tc)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": result
                    })
            else:
                return msg.content or ""
        
        return "Max steps reached without final answer"
    
    def _get_tool_schemas(self):
        # 简化版，实际从 MCP Server 获取
        return []
    
    async def _execute_tool(self, tool_call):
        # 简化版
        return f"Tool {tool_call.function.name} executed"

3.3 编排器（Orchestrator）

class MultiAgentOrchestrator:
    """多 Agent 编排器——系统的核心"""
    
    def __init__(self, api_key: str):
        self.decomposer = TaskDecomposer(api_key)
        self.agents: Dict[str, AgentRuntime] = {}
        self.results: Dict[str, str] = {}
    
    def register_agent(self, agent_def: AgentDefinition, api_key: str):
        """注册 Agent"""
        self.agents[agent_def.role] = AgentRuntime(agent_def, api_key)
    
    def register_agents(self, agents: List[AgentDefinition], api_key: str):
        """批量注册"""
        for agent_def in agents:
            self.register_agent(agent_def, api_key)
    
    async def execute(self, task: str, mode: str = "parallel") -> dict:
        """
        执行任务
        
        mode:
        - "sequential": 顺序执行
        - "parallel": 自动并行（默认）
        - "debate": 辩论模式
        """
        
        print(f"\n{'='*70}")
        print(f"Multi-Agent Orchestrator | Mode: {mode}")
        print(f"Task: {task[:100]}...")
        print(f"Agents: {len(self.agents)}")
        print(f"{'='*70}\n")
        
        if mode == "debate":
            return await self._execute_debate(task)
        
        # 1. 分解任务
        subtasks = self.decomposer.decompose(
            task, 
            [a.def for a in self.agents.values()]
        )
        
        # 2. 按依赖关系分层执行
        completed = set()
        all_results = {}
        
        while len(completed) < len(subtasks):
            # 找到所有依赖已满足的子任务
            ready = [
                st for st in subtasks 
                if st.id not in completed 
                and all(dep in completed for dep in st.dependencies)
            ]
            
            if not ready:
                raise RuntimeError("Deadlock detected: no runnable tasks")
            
            # 并行执行所有就绪任务
            print(f"\n--- Batch: {[t.id for t in ready]} ---")
            
            batch_results = await asyncio.gather(*[
                self._execute_subtask(st, all_results)
                for st in ready
            ])
            
            for st, result in zip(ready, batch_results):
                all_results[st.id] = result
                completed.add(st.id)
        
        # 3. 汇总结果
        summary = await self._summarize(task, all_results)
        
        return {
            "task": task,
            "subtasks": all_results,
            "summary": summary,
            "agent_count": len(self.agents),
            "subtask_count": len(subtasks)
        }
    
    async def _execute_subtask(
        self, 
        subtask: SubTask, 
        all_results: Dict[str, str]
    ) -> str:
        """执行单个子任务"""
        agent = self.agents.get(subtask.agent_role)
        if not agent:
            return f"Error: No agent found for role '{subtask.agent_role}'"
        
        print(f"  [{subtask.id}] → {agent.def.name}")
        
        # 收集依赖任务的输出作为上下文
        context = {
            dep_id: all_results[dep_id]
            for dep_id in subtask.dependencies
            if dep_id in all_results
        }
        
        result = await agent.execute(subtask.description, context)
        print(f"  [{subtask.id}] ✓ done ({len(result)} chars)")
        return result
    
    async def _summarize(
        self, 
        task: str, 
        results: Dict[str, str]
    ) -> str:
        """汇总所有子任务结果"""
        
        # 用最强大的模型做汇总
        summaries = "\n\n---\n\n".join([
            f"### {task_id}\n{output[:800]}"
            for task_id, output in results.items()
        ])
        
        response = OpenAI().chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": f"""你是一个报告汇总专家。整合以下子任务的结果，
形成一份连贯、完整的最终报告。

规则：
1. 逻辑连贯，不是简单拼接
2. 突出关键发现和行动建议
3. 标注信息来源（哪个子任务提供的数据）
4. 如果子任务结果有矛盾，指出并分析

原始任务：{task}

子任务结果：
{summaries}"""
            }],
            temperature=0.3
        )
        
        return response.choices[0].message.content
    
    async def _execute_debate(self, topic: str) -> dict:
        """辩论模式：多方 Agent 辩论后得出结论"""
        
        print(f"\n{' Debate Mode ':=^60}\n")
        
        # 需要至少 3 个 Agent：正方、反方、裁判
        agents_list = list(self.agents.values())
        if len(agents_list) < 3:
            raise ValueError("Debate mode requires at least 3 agents")
        
        pro_agent = agents_list[0]
        con_agent = agents_list[1]
        judge = agents_list[2]
        
        rounds = 2
        
        # 第一轮：正反方陈述
        pro_args = await pro_agent.execute(
            f"请从正面论证以下命题，提供有力论据：{topic}"
        )
        con_args = await con_agent.execute(
            f"请从反面质疑以下命题，指出潜在问题：{topic}\n\n正方观点：{pro_args[:500]}"
        )
        
        # 第二轮：互相反驳
        pro_rebuttal = await pro_agent.execute(
            f"反驳以下反对意见：\n{con_args[:500]}"
        )
        con_rebuttal = await con_agent.execute(
            f"反驳以下支持意见：\n{pro_args[:500]}"
        )
        
        # 裁判判决
        verdict = await judge.execute(f"""作为中立裁判，基于以下辩论做出判决：

命题：{topic}

正方论点：{pro_args[:300]}
反方论点：{con_args[:300]}
正方反驳：{pro_rebuttal[:300]}
反方反驳：{con_rebuttal[:300]}

请给出：
1. 判决结果（支持/反对/有条件支持）
2. 关键论据分析
3. 建议的折中方案""")
        
        return {
            "topic": topic,
            "pro_arguments": pro_args,
            "con_arguments": con_args,
            "pro_rebuttal": pro_rebuttal,
            "con_rebuttal": con_rebuttal,
            "verdict": verdict
        }

---

四、实战：代码审查多 Agent 系统

# === 定义专业 Agent ===

AGENTS = [
    AgentDefinition(
        name="SecurityScanner",
        role="security",
        system_prompt="""你是资深安全工程师。审查代码中的安全漏洞：
- SQL 注入、XSS、CSRF
- 敏感信息泄露（硬编码密钥、密码）
- 权限控制缺陷
- 依赖库的已知漏洞（CVE）
输出格式：漏洞类型 | 严重程度 | 位置 | 修复建议""",
        model="gpt-4o"
    ),
    
    AgentDefinition(
        name="PerformanceAnalyst",
        role="performance",
        system_prompt="""你是性能优化专家。分析代码性能问题：
- 时间复杂度（大O分析）
- 内存使用和泄露风险
- 数据库查询效率（N+1问题）
- 缓存策略建议
输出格式：问题类型 | 影响程度 | 位置 | 优化建议""",
        model="gpt-4o-mini"  # 性能分析用轻量模型即可
    ),
    
    AgentDefinition(
        name="CodeStyleReviewer",
        role="style",
        system_prompt="""你是代码规范审查员。检查代码风格和可维护性：
- 命名规范（变量、函数、类）
- 代码结构和模块化
- 注释和文档完整性
- SOLID 原则遵循度
输出格式：问题类型 | 位置 | 建议""",
        model="gpt-4o-mini"
    ),
    
    AgentDefinition(
        name="BestPracticeAdvisor",
        role="bestpractice",
        system_prompt="""你是架构和最佳实践顾问。从架构层面审视代码：
- 设计模式使用是否恰当
- 是否符合行业最佳实践
- 可扩展性和可测试性
- 技术选型是否合理
输出格式：建议类别 | 当前问题 | 改进方向""",
        model="gpt-4o"
    ),
    
    AgentDefinition(
        name="ReportWriter",
        role="reporter",
        system_prompt="""你是技术报告撰写专家。基于各专家的审查结果，
生成结构清晰、有优先级的综合审查报告。

报告结构：
1. 执行摘要（3-5句）
2. 严重问题（必须修复）
3. 重要问题（建议修复）
4. 优化建议（锦上添花）
5. 总体评分（1-10分，分维度）

使用 Markdown 格式。""",
        model="gpt-4o"
    )
]


# === 使用 ===

async def review_codebase(repo_url: str, api_key: str):
    """审查一个代码仓库"""
    
    orchestrator = MultiAgentOrchestrator(api_key)
    orchestrator.register_agents(AGENTS, api_key)
    
    task = f"""
    审查来自 {repo_url} 的代码。请：
    1. 先克隆仓库并读取关键文件
    2. 从安全、性能、代码风格、最佳实践四个维度审查
    3. 生成综合审查报告，按优先级排序问题
    
    注意：只审查核心业务代码，跳过测试文件和配置文件。
    """
    
    result = await orchestrator.execute(task, mode="parallel")
    
    # 保存报告
    with open("review_report.md", "w", encoding="utf-8") as f:
        f.write(result["summary"])
    
    print(f"\nReport saved to review_report.md")
    print(f"Agents used: {result['agent_count']}")
    print(f"Sub-tasks: {result['subtask_count']}")
    
    return result


# 运行
if __name__ == "__main__":
    result = asyncio.run(review_codebase(
        "https://github.com/example/project",
        api_key="sk-xxx"
    ))
    
    print("\n" + "="*60)
    print("FINAL REPORT")
    print("="*60)
    print(result["summary"][:1000])

执行流程可视化：

                           Task
                            │
                      Decomposer
                            │
         ┌──────────────────┼──────────────────┐
         │                  │                  │
    task_1:clone       task_2:list_files   task_3:detect_lang
    (security)          (performance)       (bestpractice)
         │                  │                  │
         └──────────────────┼──────────────────┘
                            │
                     analyze_results
                            │
         ┌──────────────────┼──────────────────┬──────────────────┐
         │                  │                  │                  │
    SecurityScanner  PerformanceAnalyst  CodeStyleReviewer  BestPractice
         │                  │                  │                  │
         └──────────────────┼──────────────────┴──────────────────┘
                            │
                       ReportWriter
                            │
                       Final Report

---

五、模式对比与选型

模式	适用场景	优点	缺点
Pipeline	流式处理（翻译→润色）	简单、可控	无并行，速度慢
Map-Reduce	独立分析（多维度审查）	并行快，专业分工	需等待最慢节点
Debate	决策论证（技术选型）	多角度，避免偏见	耗时长，Agent多
Hierarchy	大型项目（系统开发）	可扩展，分工细	复杂，调试困难

选型决策

任务可拆分 → No → 单 Agent
            Yes → 子任务独立 → Yes → Map-Reduce
                               No → 需要多方论证 → Yes → Debate
                                                   No → 层级复杂 → Hierarchy
                                                                  → Pipeline

---

六、生产环境注意事项

6.1 成本控制

class CostAwareOrchestrator(MultiAgentOrchestrator):
    """成本感知的编排器"""
    
    def __init__(self, api_key: str, budget: float = 1.0):
        super().__init__(api_key)
        self.budget = budget
        self.spent = 0.0
    
    async def _execute_subtask(self, subtask, all_results):
        """执行前检查预算"""
        # 估算成本
        est_cost = self._estimate_cost(subtask)
        if self.spent + est_cost > self.budget:
            print(f"  [{subtask.id}] SKIPPED: would exceed budget "
                  f"(${self.spent:.2f} + ${est_cost:.2f} > ${self.budget:.2f})")
            return "Skipped due to budget constraints"
        
        result = await super()._execute_subtask(subtask, all_results)
        
        # 实际成本记录
        actual_cost = est_cost * 0.8  # 简化
        self.spent += actual_cost
        print(f"  Cost: ${actual_cost:.4f} (total: ${self.spent:.4f})")
        
        return result
    
    def _estimate_cost(self, subtask) -> float:
        """估算子任务成本"""
        agent = self.agents.get(subtask.agent_role)
        if not agent:
            return 0
        
        # gpt-4o: $5/M input, $15/M output
        if "gpt-4o" in agent.def.model:
            return 0.02  # 估算每次调用 ~$0.02
        elif "gpt-4o-mini" in agent.def.model:
            return 0.001
        return 0.01

6.2 超时与降级

async def execute_with_timeout(orchestrator, task, timeout=300):
    """带总超时的执行"""
    try:
        return await asyncio.wait_for(
            orchestrator.execute(task),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        # 返回已完成部分的结果
        return {
            "task": task,
            "status": "partial",
            "completed": orchestrator.results,
            "error": f"Execution timed out after {timeout}s"
        }

---

七、系列总结

五篇文章，从检索到协作，构成了 AI 应用开发的完整能力栈：

┌────────────────────────────────────────────┐
│            多 Agent 协作 (第5篇)              │  ← 编排层
│    任务分解 · 通信 · 辩论 · Map-Reduce       │
├────────────────────────────────────────────┤
│          工作流引擎 (第4篇)                   │  ← 执行层
│    DAG 编排 · 动态路由 · 重试 · 可观测       │
├────────────────────────────────────────────┤
│    Graph RAG (第3篇)    │   MCP 协议 (第2篇)  │  ← 能力层
│  知识图谱增强推理       │  标准化工具调用      │
├────────────────────────────────────────────┤
│             RAG 进阶 (第1篇)                  │  ← 基础层
│   混合检索 · 重排序 · 自查询 · Agentic RAG   │
└────────────────────────────────────────────┘

各篇核心代码量

篇	主题	核心代码行数	关键依赖
1	RAG 进阶	~350 行	ChromaDB, Sentence-Transformers, OpenAI
2	MCP 协议	~400 行	asyncio, aiohttp, Neo4j
3	Graph RAG	~350 行	Neo4j, LangChain, NetworkX
4	工作流引擎	~400 行	asyncio, dataclasses
5	多 Agent	~450 行	asyncio, OpenAI

学习路径建议

入门 → 第1篇 RAG: 理解检索增强生成的核心概念
   ↓
进阶 → 第2篇 MCP: 掌握 Agent 工具调用的标准方式
   ↓
深入 → 第3篇 Graph RAG: 处理复杂实体关系的场景
   ↓
工程 → 第4篇 工作流: 构建可靠的生产级 Agent 流程
   ↓
高级 → 第5篇 多 Agent: 设计可扩展的协作系统

---

全系列 5 篇已完结。完整可运行代码已开源在 GitHub。

如果这个系列对你有帮助，欢迎点赞、收藏、转发！
有任何问题或想讨论的话题，请在评论区留言。

下一篇系列预告：LLM 推理优化——从 Prompt Engineering 到 vLLM 部署