Supervisor（监督者模式）

核心：单 “老板” Agent 负责任务分解、分配与进度决策，多个 “员工” Agent 各司其职，完成后均向监督者反馈。
代码：基于 LangGraph 搭建 StateGraph，定义研究员、写作者、审稿人专业 Agent，监督者通过条件判断分配下一步任务。
优：集中控制、流程清晰、易调试、天然支持任务优先级；劣：监督者是单点瓶颈、LLM 调用成本高、不支持 Agent 直接对话。
适用：研究报告生成、代码审查等复杂多步骤流程任务。

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent

# 定义专业Agent
researcher = create_react_agent(
    model, tools=[web_search, arxiv_search],
    prompt="你是一个资深研究员，专注于收集和分析技术资料。"
)

writer = create_react_agent(
    model, tools=[write_document, format_markdown],
    prompt="你是一个技术写作专家，擅长将复杂概念转化为清晰文章。"
)

reviewer = create_react_agent(
    model, tools=[check_facts, grammar_check],
    prompt="你是一个严格的审稿人，确保内容的准确性和质量。"
)

# Supervisor逻辑
def supervisor(state):
    """监督者决定下一步由谁执行"""
    messages = state["messages"]

    response = model.invoke([
        SystemMessage(content="""你是团队leader。根据当前进度决定：
        - 需要更多研究 → 'researcher'
        - 可以开始写作 → 'writer'  
        - 需要审核 → 'reviewer'
        - 任务完成 → 'FINISH'"""),
        *messages
    ])

    return {"next": response.content}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)

graph.add_conditional_edges("supervisor", 
    lambda state: state["next"],
    {"researcher": "researcher", "writer": "writer", 
     "reviewer": "reviewer", "FINISH": END})

# 所有Agent完成后都回到Supervisor
for agent in ["researcher", "writer", "reviewer"]:
    graph.add_edge(agent, "supervisor")

graph.set_entry_point("supervisor")
app = graph.compile()

Handoff（交接模式）

核心：Agent 间直接交接控制权，无中间协调层，是 OpenAI Agents SDK 核心设计模式。
代码：基于 OpenAI Agents 定义分诊、技术、销售 Agent，通过handoffs配置规则实现任务直接交接。
优：去中心化、通信延迟低、模拟人类团队协作；
劣：交接链易形成环路、调试困难、需 Agent 明确交接时机。
适用：客服、用户对话等需要动态任务流转的场景。

from openai import agents

# 定义Agent及其交接规则
triage_agent = agents.Agent(
    name="分诊Agent",
    instructions="根据用户需求，交接给合适的专业Agent。",
    handoffs=[
        agents.Handoff(target="tech_agent", 
                      description="技术问题"),
        agents.Handoff(target="sales_agent",
                      description="销售咨询"),
        agents.Handoff(target="support_agent",
                      description="售后支持"),
    ]
)

tech_agent = agents.Agent(
    name="技术Agent",
    instructions="你是技术专家。解决技术问题，必要时交回分诊。",
    tools=[search_docs, run_code],
    handoffs=[
        agents.Handoff(target="triage_agent",
                      description="非技术问题，交回分诊")
    ]
)

sales_agent = agents.Agent(
    name="销售Agent",
    instructions="你是销售顾问。处理定价、方案咨询。",
    tools=[get_pricing, create_quote],
    handoffs=[
        agents.Handoff(target="triage_agent",
                      description="非销售问题，交回分诊"),
        agents.Handoff(target="tech_agent",
                      description="需要技术确认")
    ]
)

# 运行
result = agents.run(triage_agent, "我想了解你们的API性能指标")

Router（路由器模式）

核心：轻量级路由层基于输入特征分发任务，规则匹配优先、LLM 兜底，路由层不参与后续处理。
代码：自定义 AgentRouter 类，通过关键词规则匹配分发至代码、数据、运维 Agent，规则失效时调用 LLM 分类。
优：路由延迟极低（规则匹配 < 1ms）、可预测性强、易监控计费；劣：规则维护成本高、不支持多 Agent 协作、一次仅选一个 Agent。
适用：简单的任务分类分发场景。

# 高效路由器实现——不用LLM做路由决策
import re
from typing import Dict, Callable

class AgentRouter:
    def __init__(self):
        self.routes: list[tuple[Callable, str]] = []
        self.agents: Dict[str, Agent] = {}

    def register(self, agent_name: str, agent: Agent, 
                 matcher: Callable[[str], bool]):
        self.agents[agent_name] = agent
        self.routes.append((matcher, agent_name))

    def route(self, query: str) -> str:
        # 规则优先（快速、可预测、零成本）
        for matcher, agent_name in self.routes:
            if matcher(query):
                return agent_name
        # fallback到LLM路由
        return self._llm_route(query)

    def _llm_route(self, query: str) -> str:
        """仅在规则匹配失败时使用LLM"""
        response = model.invoke(
            f"将以下查询分类到: {list(self.agents.keys())}\n查询: {query}"
        )
        return response.content.strip()

# 注册Agent和路由规则
router = AgentRouter()
router.register("code_agent", code_agent, 
    lambda q: any(kw in q for kw in ["代码", "bug", "编程", "函数"]))
router.register("data_agent", data_agent,
    lambda q: any(kw in q for kw in ["数据", "SQL", "查询", "报表"]))
router.register("infra_agent", infra_agent,
    lambda q: any(kw in q for kw in ["部署", "服务器", "K8s", "Docker"]))

Swarm（群体智能模式）

核心：无中心控制，多个 Agent 通过共享黑板（Shared Blackboard） 读写状态、并行工作，可互相参考结果补充任务。
代码：基于 asyncio 实现异步并行，通过 SharedBlackboard 类实现 Agent 间状态共享，学术、新闻、GitHub 等 Agent 并行搜索并补充研究。
优：真正并行、效率最高、去中心化无单点故障、Agent 间可互相启发；劣：协调复杂、易重复工作、结果合成难度大。
适用：研究、创意生成等需要多维度并行探索的场景。

import asyncio
from dataclasses import dataclass, field

@dataclass
class SharedBlackboard:
    """共享黑板——所有Agent都可以读写"""
    findings: list = field(default_factory=list)
    tasks: list = field(default_factory=list)
    status: dict = field(default_factory=dict)
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    async def post(self, agent_name: str, content: dict):
        async with self.lock:
            self.findings.append({
                "agent": agent_name,
                "content": content,
                "timestamp": time.time()
            })

    async def get_recent(self, n: int = 10) -> list:
        return self.findings[-n:]

# Swarm中的每个Agent独立运行
async def research_swarm(topic: str, num_agents: int = 5):
    blackboard = SharedBlackboard()

    agents = [
        ("学术Agent", search_arxiv),
        ("新闻Agent", search_news),
        ("GitHub Agent", search_github),
        ("专利Agent", search_patents),
        ("社交Agent", search_twitter),
    ]

    async def agent_worker(name, search_fn):
        # 独立搜索
        results = await search_fn(topic)
        await blackboard.post(name, {"raw": results})

        # 查看其他Agent的发现，补充研究
        others = await blackboard.get_recent()
        gaps = identify_gaps(results, others)
        if gaps:
            supplementary = await search_fn(gaps)
            await blackboard.post(name, {"supplementary": supplementary})

    # 所有Agent并行执行
    await asyncio.gather(*[
        agent_worker(name, fn) for name, fn in agents
    ])

    # 汇总Agent合成最终结果
    return await synthesize(blackboard.findings)

模式对比与选型策略