第17课：Multi-Agent协作

👥 Multi-Agent协作：1+1>2

单个Agent的能力是有限的，但多个Agent协作可以处理更复杂的任务。就像人类社会一样，分工合作往往比单打独斗更高效。本课我们学习Multi-Agent系统的设计与实现。

📖 Multi-Agent架构模式

Multi-Agent架构
├── 中心化 (Centralized)
│   └── 主Agent → 分配任务给子Agent → 汇总结果
├── 去中心化 (Decentralized)
│   └── Agent间直接通信 → 协商决策
├── 层级化 (Hierarchical)
│   └── 管理Agent → 中层Agent → 执行Agent
├── 对话式 (Conversational)
│   └── Agent间多轮对话 → 达成共识
└── 黑板式 (Blackboard)
    └── 共享黑板 → Agent读写 → 协同求解

📊 Multi-Agent框架对比

💻 代码实现：Multi-Agent系统

# Multi-Agent协作系统
import json, time
from typing import Dict, List, Any, Callable, Optional
from dataclasses import dataclass, field

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    content: Any
    msg_type: str = "info"  # info, task, result, broadcast

class Agent:
    # 单个Agent
    def __init__(self, name, role, expertise):
        self.name = name
        self.role = role
        self.expertise = expertise
        self.inbox: List[AgentMessage] = []
        self.knowledge: Dict = {}
    
    def receive(self, message: AgentMessage):
        self.inbox.append(message)
    
    def process(self) -> Optional[AgentMessage]:
        if not self.inbox:
            return None
        msg = self.inbox.pop(0)
        # 模拟处理
        result = f"{self.name}({self.role})处理了: {str(msg.content)[:50]}"
        self.knowledge[msg.sender] = msg.content
        return AgentMessage(self.name, msg.sender, result, "result")

class MultiAgentSystem:
    # Multi-Agent系统 - 中心化架构
    def __init__(self):
        self.agents: Dict[str, Agent] = {}
        self.coordinator = Agent("coordinator", "协调者", "任务分配与结果汇总")
        self.message_log = []
    
    def add_agent(self, name, role, expertise):
        self.agents[name] = Agent(name, role, expertise)
    
    def assign_task(self, task_description):
        # 主协调者分配任务
        print(f"📋 协调者收到任务: {task_description}")
        
        # 1. 分析任务，分配给合适的Agent
        assignments = self._analyze_and_assign(task_description)
        
        # 2. 执行任务
        results = {}
        for agent_name, subtask in assignments.items():
            agent = self.agents[agent_name]
            msg = AgentMessage("coordinator", agent_name, subtask, "task")
            agent.receive(msg)
            response = agent.process()
            if response:
                results[agent_name] = response.content
                self.message_log.append(response)
        
        # 3. 汇总结果
        final = self._aggregate(results)
        return final
    
    def _analyze_and_assign(self, task):
        # 简化的任务分配逻辑
        assignments = {}
        for name, agent in self.agents.items():
            if any(kw in task for kw in agent.expertise.split()):
                assignments[name] = f"请处理: {task}"
        if not assignments:
            # 分配给所有Agent
            for name in self.agents:
                assignments[name] = f"协助处理: {task}"
        return assignments
    
    def _aggregate(self, results):
        return {"status": "completed", "agent_results": results, "summary": f"共{len(results)}个Agent协作完成"}

# 测试
system = MultiAgentSystem()
system.add_agent("researcher", "研究员", "搜索 研究 调查")
system.add_agent("writer", "写作者", "写作 撰写 文章")
system.add_agent("reviewer", "审核者", "审核 检查 质量")

result = system.assign_task("搜索AI Agent的最新研究并写一篇文章")
print(f"\n🎯 结果: {result['summary']}")
for agent, output in result['agent_results'].items():
    print(f"  {agent}: {output}")

🏋️ 实战练习

深入理解：Multi-Agent协作核心原理

Multi-Agent协作的四种范式：中央调度(星形，AutoGen，任务明确)、层级管理(树形，CrewAI，复杂项目)、民主协商(网状，ChatDev，创意任务)、黑板系统(共享状态，MetaGPT，知识密集)。通信方式包括：直接消息(AutoGen风格)、共享黑板(MetaGPT风格)、事件驱动(发布/订阅)、管道传递(LangGraph风格)。

进阶实现：Multi-Agent框架

以下是针对Multi-Agent协作主题的进阶实现，包含角色定义+消息路由+任务分配+结果聚合等核心功能。代码经过实机运行验证。

# MultiAgentSystem - Multi-Agent协作进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class MultiAgentSystem:
    # Multi-Agent协作进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = MultiAgentSystem({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

Multi-Agent协作最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：Multi-Agent协作的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。