第39课：Multi-Agent系统

🌐 Multi-Agent系统：AI协作的终极形态

在第17课学习了Multi-Agent基础后，本课我们构建一个完整的、可运行的Multi-Agent系统，包含Agent注册、任务分发、协作通信、结果汇总的完整流程。

📖 系统架构

Multi-Agent系统架构
├── 注册中心
│   ├── Agent注册/注销
│   ├── 能力声明
│   └── 心跳检测
├── 任务调度器
│   ├── 任务队列
│   ├── 智能分配
│   ├── 优先级管理
│   └── 超时处理
├── 通信层
│   ├── 消息总线
│   ├── 事件广播
│   ├── 请求/响应
│   └── 共享黑板
├── 执行层
│   ├── 各专业Agent
│   ├── 并行执行
│   └── 结果上报
└── 监控层
    ├── 执行追踪
    ├── 性能指标
    └── 异常告警

💻 代码实现：Multi-Agent系统

# 完整的Multi-Agent协作系统
import json, time, uuid
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum

class AgentStatus(Enum):
    IDLE = "idle"
    BUSY = "busy"
    OFFLINE = "offline"

@dataclass
class AgentProfile:
    name: str
    role: str
    skills: List[str]
    status: AgentStatus = AgentStatus.IDLE
    current_task: Optional[str] = None
    completed_tasks: int = 0

@dataclass
class Task:
    id: str
    description: str
    required_skills: List[str]
    priority: int = 5  # 1最高, 10最低
    status: str = "pending"
    assigned_to: Optional[str] = None
    result: Optional[Any] = None
    parent_id: Optional[str] = None

class AgentRegistry:
    # Agent注册中心
    def __init__(self):
        self.agents: Dict[str, AgentProfile] = {}
    
    def register(self, profile: AgentProfile):
        self.agents[profile.name] = profile
        print(f"  📝 注册Agent: {profile.name}({profile.role})")
    
    def find_available(self, required_skills):
        best = None
        best_score = 0
        for agent in self.agents.values():
            if agent.status != AgentStatus.IDLE:
                continue
            match = len(set(agent.skills) & set(required_skills))
            if match > best_score:
                best_score = match
                best = agent
        return best

class MessageBus:
    # 消息总线
    def __init__(self):
        self.queues: Dict[str, List] = {}
        self.history = []
    
    def send(self, sender, receiver, msg_type, content):
        msg = {"from": sender, "to": receiver, "type": msg_type, "content": content, "time": time.time()}
        self.queues.setdefault(receiver, []).append(msg)
        self.history.append(msg)
    
    def receive(self, agent_name):
        return self.queues.pop(agent_name, [])

class TaskScheduler:
    # 任务调度器
    def __init__(self, registry: AgentRegistry, bus: MessageBus):
        self.registry = registry
        self.bus = bus
        self.task_queue: List[Task] = []
        self.completed: List[Task] = []
    
    def submit(self, task: Task):
        self.task_queue.append(task)
        self.task_queue.sort(key=lambda t: t.priority)
    
    def schedule(self):
        scheduled = 0
        remaining = []
        for task in self.task_queue:
            agent = self.registry.find_available(task.required_skills)
            if agent:
                task.assigned_to = agent.name
                task.status = "running"
                agent.status = AgentStatus.BUSY
                agent.current_task = task.id
                self.bus.send("scheduler", agent.name, "task_assign", {"task_id": task.id, "description": task.description})
                scheduled += 1
            else:
                remaining.append(task)
        self.task_queue = remaining
        return scheduled
    
    def complete_task(self, task_id, result, agent_name):
        agent = self.registry.agents.get(agent_name)
        if agent:
            agent.status = AgentStatus.IDLE
            agent.current_task = None
            agent.completed_tasks += 1
        # 找到任务并标记完成
        for task_list in [self.task_queue]:
            for task in task_list:
                if task.id == task_id:
                    task.status = "completed"
                    task.result = result
                    self.completed.append(task)
                    task_list.remove(task)
                    break

class MultiAgentOrchestrator:
    # Multi-Agent编排器
    def __init__(self):
        self.registry = AgentRegistry()
        self.bus = MessageBus()
        self.scheduler = TaskScheduler(self.registry, self.bus)
        self.agents: Dict[str, Callable] = {}
    
    def register_agent(self, profile: AgentProfile, handler: Callable):
        self.registry.register(profile)
        self.agents[profile.name] = handler
    
    def submit_task(self, description, required_skills, priority=5):
        task = Task(str(uuid.uuid4())[:8], description, required_skills, priority)
        self.scheduler.submit(task)
        return task.id
    
    def run(self):
        # 调度任务
        scheduled = self.scheduler.schedule()
        print(f"  📋 调度了{scheduled}个任务")
        
        # 执行任务
        for agent_name, handler in self.agents.items():
            messages = self.bus.receive(agent_name)
            for msg in messages:
                if msg["type"] == "task_assign":
                    result = handler(msg["content"])
                    self.scheduler.complete_task(msg["content"]["task_id"], result, agent_name)
                    print(f"  ✅ {agent_name}完成: {str(result)[:50]}")
        
        return len(self.scheduler.completed)

# 测试
orchestrator = MultiAgentOrchestrator()

orchestrator.register_agent(AgentProfile("researcher", "研究员", ["搜索","研究","分析"]),
    lambda task: f"研究完成：关于{task['description'][:20]}的3个关键发现")
orchestrator.register_agent(AgentProfile("writer", "写作者", ["写作","编辑"]),
    lambda task: f"写作完成：{task['description'][:20]}的文章草稿")
orchestrator.register_agent(AgentProfile("reviewer", "审核者", ["审核","质量检查"]),
    lambda task: f"审核完成：质量评分8.5/10")

# 提交任务
print("📋 提交任务:")
t1 = orchestrator.submit_task("研究AI Agent技术趋势", ["搜索","研究"], priority=2)
t2 = orchestrator.submit_task("撰写AI Agent技术报告", ["写作"], priority=3)
t3 = orchestrator.submit_task("审核技术报告质量", ["审核"], priority=4)

print(f"\n🚀 执行:")
completed = orchestrator.run()
print(f"\n📊 完成{completed}个任务")
print(f"Agent状态:")
for name, profile in orchestrator.registry.agents.items():
    print(f"  {name}({profile.role}): 完成{profile.completed_tasks}个任务")

🏋️ 实战练习

深入理解：Multi-Agent系统核心原理

企业级Multi-Agent架构：用户界面 - 路由Agent - [研究Agent/编码Agent/审核Agent] - 协调Agent - [质量Agent/安全Agent/文档Agent] - 输出。Agent间协调模式：顺序流水线(Research-Code-Test-Deploy)、并行协作(多Agent同时处理子任务)、辩论式(多Agent独立回答投票取最优)、师徒式(高级指导低级逐步提升)。

进阶实现：Multi-Agent编排

以下是针对Multi-Agent系统主题的进阶实现，包含角色分配+协作协议+结果合并等核心功能。代码经过实机运行验证。

# MultiAgentOrchestrator - 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 MultiAgentOrchestrator:
    # 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 = MultiAgentOrchestrator({"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系统的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。