AI Agents：多智能体协作系统的架构设计与工程实现

import json, uuid, time
from dataclasses import dataclass, field, asdict
from collections import defaultdict, deque

@dataclass
class Message:
    sender: str
    receiver: str
    type: str                    # task / result / critique
    payload: dict
    msg_id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
    ts: float = field(default_factory=time.time)
    reply_to: str | None = None

class MessageBus:
    def __init__(self):
        self.queues = defaultdict(deque)
        self.history = []

    def send(self, msg: Message):
        self.queues[msg.receiver].append(msg)
        self.history.append(msg)

    def recv(self, agent: str) -> Message | None:
        q = self.queues[agent]
        return q.popleft() if q else None

    def trace(self, msg_id: str) -> list[Message]:
        """按消息链回溯，用于调试与审计"""
        chain, cur = [], msg_id
        while cur:
            m = next((x for x in self.history if x.msg_id == cur), None)
            if not m:
                break
            chain.append(m)
            cur = m.reply_to
        return list(reversed(chain))

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class BaseAgent:
    name = "base"
    system = "你是一个助手。"

    def __init__(self, bus: MessageBus):
        self.bus = bus

    def think(self, prompt: str, temperature=0.3) -> str:
        r = client.chat.completions.create(
            model="gpt-4o-mini", temperature=temperature,
            messages=[{"role": "system", "content": self.system},
                      {"role": "user", "content": prompt}],
        )
        return r.choices[0].message.content

    def handle(self, msg: Message):
        raise NotImplementedError

    def run_once(self) -> bool:
        msg = self.bus.recv(self.name)
        if not msg:
            return False
        self.handle(msg)
        return True

class Researcher(BaseAgent):
    name = "researcher"
    system = "你是研究员。基于给定主题，输出3-5个关键事实要点，简洁客观。"

    def handle(self, msg: Message):
        topic = msg.payload["topic"]
        facts = self.think(f"主题：{topic}\n请列出关键事实。")
        self.bus.send(Message(
            sender=self.name, receiver="writer",
            type="result", payload={"topic": topic, "facts": facts},
            reply_to=msg.msg_id))

class Writer(BaseAgent):
    name = "writer"
    system = "你是技术写作者。基于事实写一段300字说明，逻辑清晰。"

    def handle(self, msg: Message):
        facts = msg.payload["facts"]
        draft = self.think(f"事实：\n{facts}\n请撰写说明。", temperature=0.7)
        self.bus.send(Message(
            sender=self.name, receiver="reviewer",
            type="result", payload={"draft": draft},
            reply_to=msg.msg_id))

class Reviewer(BaseAgent):
    name = "reviewer"
    system = ("你是审查者。输出JSON："
              '{"score":0-5,"issues":["..."],"revised":"修订后全文"}')

    def handle(self, msg: Message):
        import json
        out = self.think(f"请审查并修订：\n{msg.payload['draft']}")
        try:
            data = json.loads(out)
        except json.JSONDecodeError:
            data = {"score": 0, "issues": ["解析失败"], "revised": out}
        self.bus.send(Message(
            sender=self.name, receiver="manager",
            type="critique", payload=data,
            reply_to=msg.msg_id))

class Orchestrator:
    def __init__(self, bus: MessageBus, agents: list[BaseAgent], max_rounds=10):
        self.bus = bus
        self.agents = {a.name: a for a in agents}
        self.max_rounds = max_rounds

    def run(self, topic: str) -> dict:
        self.bus.send(Message(
            sender="manager", receiver="researcher",
            type="task", payload={"topic": topic}))

        for _ in range(self.max_rounds):
            progressed = False
            for agent in self.agents.values():
                if agent.run_once():
                    progressed = True
            # 终止条件：reviewer 已产出结论
            done = [m for m in self.bus.history
                    if m.sender == "reviewer" and m.type == "critique"]
            if done:
                return done[-1].payload
            if not progressed:
                break
        return {"error": "未在限定轮次内完成"}

# 运行
bus = MessageBus()
orch = Orchestrator(bus, [Researcher(bus), Writer(bus), Reviewer(bus)])
result = orch.run("AI Agent 的记忆机制")
print(result.get("revised", result)[:300])