多 Agent 架构整理：四种经典模式与框架选型

                  ┌─────────────┐
                  │ Orchestrator │ ← 中央调度者
                  └──┬───┬───┬──┘
                     │   │   │
              ┌──────┘   │   └──────┐
              ▼          ▼          ▼
        ┌────────┐ ┌────────┐ ┌────────┐
        │Agent A │ │Agent B │ │Agent C │
        └────────┘ └────────┘ └────────┘

class OrchestratorAgent:
    def run(self, task: str) -> str:
        # 1. 分解任务
        subtasks = self.decompose(task)
        # ["查天气", "查机票", "查酒店"]

        # 2. 并行分发
        results = await asyncio.gather(*[
            self.dispatch(sub) for sub in subtasks
        ])

        # 3. 汇总整合
        return self.synthesize(results)

    def decompose(self, task) -> list:
        return self.llm.plan(
            f"将以下任务分解为可并行的子任务: {task}"
        )

                    ┌──────────┐
                    │  CEO Agent│
                    └────┬─────┘
                         │
              ┌──────────┼──────────┐
              ▼          ▼          ▼
        ┌────────┐ ┌────────┐ ┌────────┐
        │Manager │ │Manager │ │Manager │
        │  A     │ │  B     │ │  C     │
        └───┬────┘ └───┬────┘ └───┬────┘
            │          │          │
       ┌────┼────┐    ...        ...
       ▼    ▼    ▼
     Worker Worker Worker

class HierarchicalSystem:
    def __init__(self):
        self.ceo = CEOAgent()
        self.managers = {
            "research": ManagerAgent("research"),
            "writing": ManagerAgent("writing"),
            "review": ManagerAgent("review"),
        }

    async def run(self, task: str):
        # CEO 定策略
        strategy = self.ceo.plan(task)
        # {research: "查最新AI新闻",
        #  writing: "写一篇评论",
        #  review: "事实核查+润色"}

        # 各 Manager 并行执行自己的子任务
        results = {}
        for dept, subtask in strategy.items():
            results[dept] = await self.managers[dept].execute(subtask)

        # CEO 最终审核
        return self.ceo.review(results)

        ┌──────┐     ┌──────┐     ┌──────┐
        │Agent1│◄───►│Agent2│◄───►│Agent3│
        └──┬───┘     └──┬───┘     └──┬───┘
           │            │            │
           └────────────┼────────────┘
                        ▼
                   ┌────────┐
                   │ 环境/  │
                   │ 黑板   │
                   └────────┘

class SwarmAgent:
    def __init__(self, name: str, role: str):
        self.name = name
        self.role = role
        self.memory = []
        self.peers = []

    async def broadcast(self, message: str):
        """向所有同伴广播消息"""
        responses = []
        for peer in self.peers:
            if peer.name != self.name:
                resp = await peer.receive(self.name, message)
                responses.append(resp)
        return responses

    async def receive(self, sender: str, message: str) -> str:
        """接收同伴消息并决定是否响应"""
        if self._is_relevant(message):
            return self.act(message)
        return "not relevant"

    def _is_relevant(self, message: str) -> bool:
        return any(kw in message for kw in self.keywords)

                  ┌─────────────────┐
                  │     Blackboard   │
                  │    (共享状态)     │
                  └──┬──┬──┬──┬─────┘
                     │  │  │  │
              ┌──────┘  │  │  └──────┐
              ▼         ▼  ▼         ▼
          ┌───────┐ ┌───────┐ ┌───────┐
          │Agent 1│ │Agent 2│ │Agent 3│
          └───────┘ └───────┘ └───────┘

class Blackboard:
    def __init__(self):
        self.state = {}
        self.contributions = []

    def write(self, agent: str, key: str, value):
        self.state[key] = value
        self.contributions.append({
            "agent": agent,
            "key": key,
            "value": value,
            "timestamp": time.time()
        })

    def read(self) -> dict:
        return self.state.copy()

    def is_stable(self, threshold: int = 3) -> bool:
        """如果连续 N 轮没有新贡献，认为收敛"""
        recent = self.contributions[-threshold:]
        return len(set(c["key"] for c in recent)) == 0


class BlackboardAgent:
    def __init__(self, name: str, expertise: list[str]):
        self.name = name
        self.expertise = expertise

    async def act(self, blackboard: Blackboard):
        state = blackboard.read()
        # 只在专业领域做贡献
        for area in self.expertise:
            if area not in state or self._can_improve(state[area]):
                improvement = await self.analyze(state, area)
                blackboard.write(self.name, area, improvement)

from crewai import Agent, Task, Crew

researcher = Agent(
    role="研究员",
    goal="查找最新 AI 趋势",
    backstory="资深技术研究员",
    allow_delegation=False
)

writer = Agent(
    role="撰稿人",
    goal="根据研究结果写报告",
    backstory="科技专栏作者",
    allow_delegation=False
)

task1 = Task(description="研究2024年AI Agent的发展趋势", agent=researcher)
task2 = Task(description="基于研究结果写一篇2000字的综述", agent=writer)

crew = Crew(agents=[researcher, writer], tasks=[task1, task2])
result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent

assistant = AssistantAgent("assistant", llm_config={"model": "gpt-4"})
user_proxy = UserProxyAgent("user_proxy", code_execution_config={"work_dir": "workspace"})

# 两个 Agent 通过对话协作，Assistant 写代码，UserProxy 执行
user_proxy.initiate_chat(
    assistant,
    message="帮我写一个爬取 Hacker News 前10条新闻的脚本"
)

from langgraph.graph import StateGraph

graph = StateGraph(AgentState)

graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)
graph.add_node("reviewer", reviewer_node)

graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges(
    "reviewer",
    lambda s: "writer" if s.needs_revision else "end",
    {"writer": "writer", "end": END}
)

你的任务特征是什么？
├── 有明确的顺序步骤？
│   └── CrewAI（顺序任务链最简单）
├── 需要 Agent 之间自由协商？
│   └── AutoGen（对话驱动，灵活）
├── 需要复杂的分支和循环逻辑？
│   └── LangGraph（图结构，可视化流程）
├── 去中心化、无领导者？
│   └── 自建 Swarm（轻量，但调试难）
└── 多专业协作、共享状态？
    └── Blackboard + LangGraph（共享状态 + 图编排）