多智能体编排模式实战对比：Supervisor / Swarm / Hierarchical

from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langchain_openai import ChatOpenAI

# 定义共享状态
class AgentState(TypedDict):
    messages: list
    next_agent: str
    research_output: str
    draft_output: str
    final_output: str

# 初始化模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Supervisor 节点：决定下一步路由
def supervisor_node(state: AgentState) -> Command:
    system_prompt = """你是一个 Supervisor，负责协调以下 Agent：
    - researcher：负责信息搜索和数据收集
    - writer：负责撰写内容
    - critic：负责审查和改进
    
    根据当前进度决定下一步应该交给哪个 Agent，
    或者返回 FINISH 表示任务完成。"""
    
    response = llm.invoke([
        {"role": "system", "content": system_prompt},
        *state["messages"]
    ])
    
    # 解析路由决策
    next_agent = parse_routing_decision(response.content)
    
    if next_agent == "FINISH":
        return Command(goto=END, update={"final_output": state["draft_output"]})
    
    return Command(goto=next_agent, update={"next_agent": next_agent})

# Researcher Agent 节点
def researcher_node(state: AgentState) -> Command:
    response = llm.invoke([
        {"role": "system", "content": "你是一个研究专家，负责搜集和整理信息。"},
        {"role": "user", "content": state["messages"][-1]["content"]}
    ])
    return Command(
        goto="supervisor",
        update={"research_output": response.content,
                "messages": state["messages"] + [{"role": "assistant", "content": response.content}]}
    )

# Writer Agent 节点
def writer_node(state: AgentState) -> Command:
    response = llm.invoke([
        {"role": "system", "content": "你是一个写作专家，根据研究资料撰写高质量内容。"},
        {"role": "user", "content": f"基于以下研究撰写文章：\n{state['research_output']}"}
    ])
    return Command(
        goto="supervisor",
        update={"draft_output": response.content,
                "messages": state["messages"] + [{"role": "assistant", "content": response.content}]}
    )

# 构建 Graph
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("critic", critic_node)

workflow.add_edge(START, "supervisor")

app = workflow.compile()

# 执行
result = app.invoke({
    "messages": [{"role": "user", "content": "写一篇关于 AI Agent 编排模式的技术分析"}],
    "next_agent": "",
    "research_output": "",
    "draft_output": "",
    "final_output": ""
})

import { StateGraph, START, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { BaseMessage, HumanMessage } from "@langchain/core/messages";

// 定义状态类型
interface AgentState {
  messages: BaseMessage[];
  nextAgent: string;
  researchOutput: string;
  draftOutput: string;
}

const llm = new ChatOpenAI({ model: "gpt-4o", temperature: 0 });

// Supervisor 节点
async function supervisorNode(state: AgentState) {
  const response = await llm.invoke([
    { role: "system", content: "Route to: researcher | writer | critic | FINISH" },
    ...state.messages,
  ]);

  const nextAgent = parseRoute(response.content as string);
  return { nextAgent };
}

// 条件路由
function routeFromSupervisor(state: AgentState): string {
  if (state.nextAgent === "FINISH") return END;
  return state.nextAgent;
}

// 构建 Graph
const workflow = new StateGraph<AgentState>({
  channels: {
    messages: { default: () => [] },
    nextAgent: { default: () => "" },
    researchOutput: { default: () => "" },
    draftOutput: { default: () => "" },
  },
});

workflow.addNode("supervisor", supervisorNode);
workflow.addNode("researcher", researcherNode);
workflow.addNode("writer", writerNode);

workflow.addEdge(START, "supervisor");
workflow.addConditionalEdges("supervisor", routeFromSupervisor);

const app = workflow.compile();

from swarm import Swarm, Agent

client = Swarm()

# 定义 handoff 函数
def transfer_to_tech_support():
    """将用户转接给技术支持 Agent"""
    return tech_agent

def transfer_to_order_agent():
    """将用户转接给订单处理 Agent"""
    return order_agent

def escalate_to_human():
    """升级到人工客服"""
    return "ESCALATE: 已转接人工客服，工单号 #" + generate_ticket_id()

# Triage Agent - 入口分流
triage_agent = Agent(
    name="Triage Agent",
    instructions="""你是客服入口分流员。根据用户问题类型：
    - 技术问题（安装、配置、Bug）-> 转接技术支持
    - 订单问题（退款、物流、支付）-> 转接订单处理
    - 无法分类 -> 升级人工""",
    functions=[transfer_to_tech_support, transfer_to_order_agent, escalate_to_human],
)

# Tech Support Agent
tech_agent = Agent(
    name="Tech Support",
    instructions="""你是技术支持专家。解决用户的技术问题。
    如果涉及退款或订单操作，调用 transfer_to_order_agent。
    如果问题超出能力范围，调用 escalate_to_human。""",
    functions=[transfer_to_order_agent, escalate_to_human],
    model="gpt-4o",
)

# Order Agent
order_agent = Agent(
    name="Order Agent",
    instructions="""你是订单处理专员。处理退款、物流查询、支付问题。
    如果涉及技术排查，调用 transfer_to_tech_support。""",
    functions=[transfer_to_tech_support, escalate_to_human,
              process_refund, check_order_status],
)

# 执行对话
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "我的订单显示已发货但三天没收到，而且 App 一直闪退"}],
)

print(response.messages[-1]["content"])
# Agent 会自动在 triage -> order -> tech 之间流转

interface SwarmAgent {
  name: string;
  instructions: string;
  functions: AgentFunction[];
  model?: string;
}

interface AgentFunction {
  name: string;
  description: string;
  handler: (args: any) => SwarmAgent | string;
}

class SwarmOrchestrator {
  private currentAgent: SwarmAgent;
  private conversationHistory: Message[] = [];
  private maxHandoffs = 10; // 防止无限循环

  constructor(entryAgent: SwarmAgent) {
    this.currentAgent = entryAgent;
  }

  async run(userMessage: string): Promise<string> {
    this.conversationHistory.push({ role: "user", content: userMessage });
    let handoffCount = 0;

    while (handoffCount < this.maxHandoffs) {
      const response = await this.callLLM(this.currentAgent);

      // 检查是否有 handoff
      if (response.functionCall) {
        const fn = this.currentAgent.functions.find(
          f => f.name === response.functionCall!.name
        );
        const result = fn!.handler(response.functionCall!.arguments);

        if (typeof result === "object" && "name" in result) {
          // handoff 到新 Agent
          console.log(`[Handoff] ${this.currentAgent.name} -> ${result.name}`);
          this.currentAgent = result;
          handoffCount++;
          continue;
        }
        // 普通函数调用结果
        this.conversationHistory.push({ role: "function", content: result });
        continue;
      }

      // 无 handoff，返回最终回答
      return response.content;
    }

    throw new Error("Exceeded maximum handoff limit");
  }
}

from crewai import Agent, Task, Crew, Process

# 顶层 Manager（自动由 CrewAI hierarchical 模式管理）
manager_llm = "gpt-4o"

# 中层 Team Lead Agents
research_lead = Agent(
    role="Research Team Lead",
    goal="协调研究团队，收集并验证所有相关数据",
    backstory="你是一位资深研究负责人，擅长分配搜索任务和交叉验证信息。",
    llm="gpt-4o",
    allow_delegation=True,  # 可以向下委派
)

engineering_lead = Agent(
    role="Engineering Team Lead",
    goal="协调工程团队，确保代码质量和架构合理性",
    backstory="你是一位技术总监，负责协调前后端开发和运维部署。",
    llm="gpt-4o",
    allow_delegation=True,
)

# 底层 Worker Agents
web_researcher = Agent(
    role="Web Researcher",
    goal="搜索并提取网络上的最新技术文献",
    backstory="你擅长网络信息检索，能快速定位高质量信息源。",
    llm="gpt-4o-mini",  # Worker 用小模型降低成本
    allow_delegation=False,
)

data_analyst = Agent(
    role="Data Analyst",
    goal="分析数据并生成可视化洞察",
    backstory="你是数据分析专家，擅长统计分析和趋势识别。",
    llm="gpt-4o-mini",
    allow_delegation=False,
)

backend_dev = Agent(
    role="Backend Developer",
    goal="实现后端 API 和业务逻辑",
    backstory="你是资深后端工程师，精通 Python 和分布式系统。",
    llm="gpt-4o-mini",
    allow_delegation=False,
)

# 定义任务
research_task = Task(
    description="调研当前多智能体编排模式的最新进展，包括论文、开源项目和企业实践",
    expected_output="结构化的调研报告，包含至少 10 个关键发现",
    agent=research_lead,
)

implementation_task = Task(
    description="基于调研结果，设计并实现一个多模式编排引擎的原型",
    expected_output="可运行的原型代码和架构文档",
    agent=engineering_lead,
    context=[research_task],  # 依赖研究任务的输出
)

# 构建层级 Crew
crew = Crew(
    agents=[research_lead, engineering_lead, web_researcher, 
            data_analyst, backend_dev],
    tasks=[research_task, implementation_task],
    process=Process.hierarchical,  # 关键：启用层级模式
    manager_llm=manager_llm,
    verbose=True,
)

# 执行
result = crew.kickoff()
print(result)

import { AutoGenGroupChat, Agent, UserProxy } from "autogen";

// 定义层级结构
const projectManager = new Agent({
  name: "ProjectManager",
  systemMessage: `你是项目经理。将复杂任务分解为子任务，
    分配给对应的 Team Lead。监控进度并处理跨团队依赖。`,
  model: "gpt-4o",
});

const researchLead = new Agent({
  name: "ResearchLead",
  systemMessage: `你是研究负责人。接收 PM 的研究任务，
    将其拆解为具体检索任务分配给 Worker，汇总结果后向上汇报。`,
  model: "gpt-4o",
});

const webResearcher = new Agent({
  name: "WebResearcher",
  systemMessage: "你是网络调研员。执行具体的搜索和信息提取任务。",
  model: "gpt-4o-mini",
});

// 多级 GroupChat 配置
const researchTeam = new AutoGenGroupChat({
  agents: [researchLead, webResearcher],
  maxRound: 5,
  speakerSelectionMethod: "round_robin",
});

const topLevelChat = new AutoGenGroupChat({
  agents: [projectManager, researchLead],
  maxRound: 10,
  speakerSelectionMethod: "auto",
  nestedChats: {
    ResearchLead: researchTeam, // 嵌套子团队
  },
});

await topLevelChat.initiate(
  "设计一个支持三种编排模式热切换的 Agent 框架"
);

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def call_agent_with_timeout(agent_fn, state, timeout=30):
    """带超时和重试的 Agent 调用"""
    try:
        result = await asyncio.wait_for(agent_fn(state), timeout=timeout)
        return result
    except asyncio.TimeoutError:
        # 记录超时日志并尝试降级
        logger.warning(f"Agent {agent_fn.__name__} timed out after {timeout}s")
        raise
    except Exception as e:
        logger.error(f"Agent {agent_fn.__name__} failed: {e}")
        raise

from opentelemetry import trace
from opentelemetry.trace import StatusCode

tracer = trace.get_tracer("multi-agent-orchestrator")

def traced_agent_call(agent_name: str):
    """为每次 Agent 调用创建 Span"""
    def decorator(fn):
        async def wrapper(state):
            with tracer.start_as_current_span(f"agent.{agent_name}") as span:
                span.set_attribute("agent.name", agent_name)
                span.set_attribute("agent.input_tokens", count_tokens(state))
                try:
                    result = await fn(state)
                    span.set_attribute("agent.output_tokens", count_tokens(result))
                    span.set_status(StatusCode.OK)
                    return result
                except Exception as e:
                    span.set_status(StatusCode.ERROR, str(e))
                    span.record_exception(e)
                    raise
        return wrapper
    return decorator