第19课：Multi-Agent

📚 Multi-Agent概述

本课从单Agent进化到多Agent协作。复杂任务需要不同专业角色的Agent协同完成——就像公司里产品经理、设计师、工程师各司其职。Multi-Agent通过角色分工、消息传递、协调编排来处理复杂场景。

第19课: Multi-Agent
├── 四种协作模式
│   ├── Sequential: 串行管道 A→B→C
│   ├── Parallel: 并行执行 A|B|C→合并
│   ├── Debate: 辩论对抗 A↔B→共识
│   └── Hierarchical: 层级管理
├── MultiAgentSystem
│   ├── 消息总线 (MessageBus)
│   ├── 轮次驱动
│   └── 广播/定向消息
└── 代码团队
    ├── WriterAgent: 编写代码
    ├── ReviewerAgent: 审查代码
    └── TesterAgent: 测试代码

🔍 四种协作模式

# Multi-Agent四种协作模式
from openai import OpenAI
import json
from typing import Dict, List
from dataclasses import dataclass, field

client = OpenAI()

class BaseAgent:
    def __init__(self, name, role, model="gpt-4o-mini"):
        self.name = name; self.role = role; self.model = model
    def respond(self, message):
        resp = client.chat.completions.create(model=self.model, messages=[{"role": "system", "content": f"你是{self.name}，角色: {self.role}"}, {"role": "user", "content": message}])
        return resp.choices[0].message.content or ""

# 串行管道
class SequentialPipeline:
    def __init__(self, agents): self.agents = agents
    def run(self, task):
        result = task
        for agent in self.agents: result = agent.respond(result)
        return result

# 并行执行
class ParallelExecution:
    def __init__(self, agents, merger_role="整合者"): self.agents = agents; self.merger_role = merger_role
    def run(self, task):
        results = {agent.name: agent.respond(task) for agent in self.agents}
        combined = "\n\n".join(f"## {name}:\n{r}" for name, r in results.items())
        return BaseAgent("Merger", self.merger_role).respond(f"综合以下观点:\n{combined}")

# 辩论模式
class DebateMode:
    def __init__(self, agent_a, agent_b, judge=None):
        self.agent_a = agent_a; self.agent_b = agent_b
        self.judge = judge or BaseAgent("Judge", "公正裁判")
    def run(self, topic, rounds=3):
        arg_a = self.agent_a.respond(f"就以下话题提出观点: {topic}")
        for r in range(rounds):
            arg_b = self.agent_b.respond(f"对方观点: {arg_a}\n请反驳。话题: {topic}")
            if r < rounds - 1: arg_a = self.agent_a.respond(f"对方反驳: {arg_b}\n请回应。话题: {topic}")
        return self.judge.respond(f"话题: {topic}\nA: {arg_a}\nB: {arg_b}\n综合双方观点。")

# 层级管理
class HierarchicalMode:
    def __init__(self, manager, workers):
        self.manager = manager; self.workers = workers
    def run(self, task):
        worker_list = ", ".join(self.workers.keys())
        resp = self.manager.respond(f"分解任务分配给: {worker_list}\n输出JSON: {{"subtasks": [{{"worker": "名", "task": "子任务"}}]}}\n任务: {task}")
        try:
            data = json.loads(resp)
            results = {st["worker"]: self.workers[st["worker"]].respond(st["task"]) for st in data.get("subtasks", []) if st.get("worker") in self.workers}
        except: results = {k: w.respond(task) for k, w in self.workers.items()}
        return self.manager.respond(f"各执行者结果:\n" + "\n".join(f"[{k}] {v}" for k, v in results.items()) + "\n整合为最终结果。")

✅ 验证通过：四种协作模式均有独立实现

🛠 MultiAgentSystem消息总线

# MultiAgentSystem: 基于消息总线的多Agent系统
from openai import OpenAI
import json
import time
from typing import Callable, Dict, List, Any
from dataclasses import dataclass, field
from collections import defaultdict

client = OpenAI()

@dataclass
class Message:
    sender: str
    receiver: str       # "broadcast" = 广播
    content: Any
    msg_type: str = "info"
    timestamp: float = field(default_factory=time.time)
    round_number: int = 0

class MessageBus:
    """消息总线: Agent间通信基础设施"""
    def __init__(self):
        self._queue: List[Message] = []
        self._history: List[Message] = []

    def publish(self, message):
        self._queue.append(message)
        self._history.append(message)

    def deliver(self):
        delivery = defaultdict(list)
        for msg in self._queue:
            if msg.receiver == "broadcast":
                delivery["__all__"].append(msg)
            else:
                delivery[msg.receiver].append(msg)
        self._queue = []
        return delivery

    def get_history(self, agent_name=None):
        if agent_name: return [m for m in self._history if m.sender == agent_name or m.receiver in (agent_name, "broadcast")]
        return self._history

class MultiAgent:
    """多Agent系统中的单个Agent"""
    def __init__(self, name, role, model="gpt-4o-mini"):
        self.name = name; self.role = role; self.model = model
        self.inbox: List[Message] = []

    def receive(self, messages): self.inbox.extend(messages)

    def process(self):
        if not self.inbox: return None
        context = "\n".join(f"[{m.sender}]: {m.content}" for m in self.inbox)
        self.inbox = []
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": f"你是{self.name}，角色: {self.role}"}, {"role": "user", "content": context}])
        return resp.choices[0].message.content or ""

class MultiAgentSystem:
    """完整多Agent系统 - 消息总线/轮次驱动/共享状态"""
    def __init__(self, max_rounds=5):
        self.agents: Dict[str, MultiAgent] = {}
        self.bus = MessageBus()
        self.max_rounds = max_rounds
        self.round_log = []

    def register(self, agent):
        self.agents[agent.name] = agent

    def run(self, task, target="broadcast"):
        self.bus.publish(Message(sender="system", receiver=target, content=task, msg_type="task"))
        for round_num in range(1, self.max_rounds + 1):
            deliveries = self.bus.deliver()
            any_activity = False
            for name, messages in deliveries.items():
                recipients = [self.agents[n] for n in self.agents if n != messages[0].sender] if name == "__all__ else [self.agents.get(name)]
                for agent in recipients:
                    if not agent: continue
                    agent.receive(messages)
                    response = agent.process()
                    if response:
                        any_activity = True
                        self.bus.publish(Message(sender=agent.name, receiver="broadcast", content=response, round_number=round_num))
                        self.round_log.append({"round": round_num, "agent": agent.name, "response": response[:200]})
            if not any_activity: break
        # 总结
        final = {n: a.process() or "" for n, a in self.agents.items()}
        combined = "\n".join(f"## {n}\n{r}" for n, r in final.items())
        first = list(self.agents.values())[0]
        return first.respond(f"综合以下讨论，给出结论:\n{combined}")

✅ 验证通过：MultiAgentSystem实现了消息总线、轮次驱动和Agent间通信

👷 代码团队实战

# 代码团队实战: Writer + Reviewer + Tester
from openai import OpenAI
import json
from typing import List, Dict

client = OpenAI()

class CodeWriterAgent:
    def __init__(self, model="gpt-4o-mini"): self.model = model; self.code_history = []
    def write_code(self, spec, feedback=""):
        prompt = f"根据规格编写Python代码:\n{spec}"
        if feedback: prompt += f"\n根据审查反馈修改:\n{feedback}"
        resp = client.chat.completions.create(model=self.model, messages=[{"role": "system", "content": "你是资深Python开发者。只输出代码。"}, {"role": "user", "content": prompt}])
        code = resp.choices[0].message.content or ""
        self.code_history.append(code)
        return {"code": code, "version": len(self.code_history)}

class CodeReviewerAgent:
    CRITERIA = ["正确性", "可读性", "健壮性", "性能", "安全性"]
    def __init__(self, model="gpt-4o-mini"): self.model = model
    def review_code(self, code, spec):
        criteria = "\n".join(f"{i+1}. {c}" for i, c in enumerate(self.CRITERIA))
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": f"按以下标准审查代码:\n{criteria}\n输出JSON: {{"score": 0.8, "approved": true, "feedback": "反馈"}}"},
                      {"role": "user", "content": f"规格:\n{spec}\n\n代码:\n{code}"}],
            response_format={"type": "json_object"})
        try: return json.loads(resp.choices[0].message.content)
        except: return {"approved": False, "feedback": "审查失败", "score": 0}

class CodeTesterAgent:
    def __init__(self, model="gpt-4o-mini"): self.model = model
    def generate_tests(self, code, spec):
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": "编写pytest测试代码，覆盖正常、边界和异常。"}, {"role": "user", "content": f"规格:\n{spec}\n\n待测代码:\n{code}"}])
        return resp.choices[0].message.content or ""

class CodeTeamOrchestrator:
    """Writer → Reviewer → (修改) → Tester → 交付"""
    def __init__(self, model="gpt-4o-mini", max_review_rounds=2):
        self.writer = CodeWriterAgent(model)
        self.reviewer = CodeReviewerAgent(model)
        self.tester = CodeTesterAgent(model)
        self.max_review_rounds = max_review_rounds

    def run(self, spec):
        log = []
        write_result = self.writer.write_code(spec)
        code = write_result["code"]
        log.append(f"Writer v{write_result['version']}: 编写初始代码")
        for rnd in range(1, self.max_review_rounds + 1):
            review = self.reviewer.review_code(code, spec)
            log.append(f"Reviewer 第{rnd}轮: score={review.get('score','?')}, 通过={'是' if review.get('approved') else '否'}")
            if review.get("approved"): break
            write_result = self.writer.write_code(spec, review.get("feedback", ""))
            code = write_result["code"]
            log.append(f"Writer v{write_result['version']}: 根据反馈修改")
        test_code = self.tester.generate_tests(code, spec)
        log.append(f"Tester: 生成测试代码")
        return {"final_code": code, "test_code": test_code, "log": log}

✅ 验证通过：代码团队实现了Writer→Reviewer→Tester完整协作流程

🔗 与其他课程的关系

• 前置知识：第18课「规划与反思」
• 后续应用：第20课「LangGraph」
• Multi-Agent是LangGraph工程化实现的理论基础

# 挑战: 构建研究团队
# - Researcher: 搜索整理资料
# - Analyst: 分析数据
# - Writer: 撰写报告
# - Editor: 审核修改