多Agent路由分派六栈：从硬编码到自组织跃迁

graph TB
    subgraph "路由分派六栈演进"
        S1["栈1: 硬编码链式<br/>CrewAI Sequential<br/>延迟≈0ms"]
        S2["栈2: 管理器仲裁<br/>LangGraph Supervisor<br/>延迟200-800ms"]
        S3["栈3: LLM自适应<br/>AG2 Selector<br/>延迟300-1200ms"]
        S4["栈4: 语义匹配<br/>Embedding Router<br/>延迟50-150ms"]
        S5["栈5: 池化负载<br/>Agent Pool<br/>延迟10-50ms"]
        S6["栈6: 运行时自适应<br/>Feedback Router<br/>延迟500-2000ms"]
    end

    S1 -->|"加决策层"| S2
    S2 -->|"去中心化"| S3
    S3 -->|"去LLM依赖"| S4
    S4 -->|"加并发"| S5
    S5 -->|"加反馈闭环"| S6

    style S1 fill:#e8f5e9
    style S2 fill:#e3f2fd
    style S3 fill:#fff3e0
    style S4 fill:#f3e5f5
    style S5 fill:#fce4ec
    style S6 fill:#fff8e1

# CrewAI Sequential Process - 硬编码链式分派
from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Research Analyst",
    goal="Find and analyze information",
    backstory="Expert at deep research",
    tools=[search_tool, scrape_tool]
)

writer = Agent(
    role="Content Writer", 
    goal="Write engaging content",
    backstory="Expert writer"
)

editor = Agent(
    role="Editor",
    goal="Review and refine content", 
    backstory="Senior editor"
)

# 路由分派 = 任务列表顺序，无运行时决策
research_task = Task(description="Research {topic}", agent=researcher)
write_task = Task(description="Write article based on research", agent=writer)
edit_task = Task(description="Edit and finalize", agent=editor)

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.sequential  # 硬编码：A→B→C
)

# LangGraph Supervisor Pattern - 管理器仲裁分派
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

# 定义专业Agent
research_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[web_search, arxiv_search],
    prompt="You are a research specialist."
)

code_agent = create_react_agent(
    model="openai:gpt-4.1", 
    tools=[python_repl, file_write],
    prompt="You are a coding specialist."
)

# Supervisor = 路由决策层
supervisor = create_supervisor(
    agents=[research_agent, code_agent],
    model="openai:gpt-4.1",
    prompt="""You are a team supervisor. Route tasks:
    - Research/analysis → research_agent
    - Code/debug → code_agent
    - If unsure, ask for clarification."""
)

# 编译为可执行图
app = supervisor.compile()
result = app.invoke({"messages": [("user", "Research async Python and write example")]})

# OpenAI Agents SDK Handoff - 隐式管理器
from agents import Agent, Runner

triage_agent = Agent(
    name="Triage",
    instructions="Route to the right specialist.",
    handoffs=[  # 声明式路由：Supervisor通过handoff描述做决策
        ResearchAgent.handoff(description="For research tasks"),
        CodeAgent.handoff(description="For coding tasks"),
        WritingAgent.handoff(description="For writing tasks"),
    ]
)

# AG2 GroupChat with Selector - LLM自适应分派
import ag2

researcher = ag2.AssistantAgent("researcher", llm_config=llm_config)
coder = ag2.AssistantAgent("coder", llm_config=llm_config)
reviewer = ag2.AssistantAgent("reviewer", llm_config=llm_config)

# Selector = 路由决策引擎（非中央管理器，而是群体选择器）
groupchat = ag2.GroupChat(
    agents=[researcher, coder, reviewer],
    messages=[],
    max_round=15,
    speaker_selection_method="auto",  # LLM决定谁发言
    # 也可用 "round_robin" 或自定义函数
)

manager = ag2.GroupChatManager(groupchat=groupchat, llm_config=llm_config)
manager.initiate_chat(
    researcher,
    message="Research distributed systems and write a summary"
)

# 语义匹配路由器 - 零LLM调用的Agent分派
import numpy as np
from openai import OpenAI

client = OpenAI()

class SemanticAgentRouter:
    def __init__(self, agents: list):
        self.agents = agents
        # 预计算Agent能力描述的嵌入向量
        self.agent_embeddings = {}
        for agent in agents:
            desc = f"{agent.role}: {agent.goal}. {agent.backstory}"
            resp = client.embeddings.create(
                model="text-embedding-3-small",
                input=desc
            )
            self.agent_embeddings[agent.name] = resp.data[0].embedding

    def route(self, task: str) -> str:
        """用向量相似度匹配，而非LLM决策"""
        task_embedding = client.embeddings.create(
            model="text-embedding-3-small",
            input=task
        ).data[0].embedding

        scores = {}
        for name, emb in self.agent_embeddings.items():
            scores[name] = np.dot(task_embedding, emb) / (
                np.linalg.norm(task_embedding) * np.linalg.norm(emb)
            )

        return max(scores, key=scores.get)

    def route_top_k(self, task: str, k: int = 3) -> list:
        """返回Top-K候选，供Supervisor做最终选择"""
        # 比纯LLM路由快5-10倍，成本降95%
        ...

# Agent Pool + 负载均衡分派
import asyncio
from dataclasses import dataclass
from typing import Optional

@dataclass
class AgentInstance:
    agent_id: str
    role: str
    status: str = "idle"  # idle | busy
    current_task: Optional[str] = None

class AgentPool:
    def __init__(self, agent_factory, pool_size: int = 5):
        self.pools: dict[str, list[AgentInstance]] = {}
        # 为每种角色预创建实例池
        for role, factory in agent_factory.items():
            self.pools[role] = [
                AgentInstance(agent_id=f"{role}_{i}", role=role)
                for i in range(pool_size)
            ]

    async def dispatch(self, role: str, task: str) -> AgentInstance:
        """最小负载优先分派"""
        # 等待空闲实例
        while True:
            for instance in self.pools[role]:
                if instance.status == "idle":
                    instance.status = "busy"
                    instance.current_task = task
                    return instance
            await asyncio.sleep(0.1)  # 自旋等待

    def release(self, instance: AgentInstance):
        instance.status = "idle"
        instance.current_task = None

# 运行时自适应路由器（简化版）
from datetime import datetime
import json

class AdaptiveRouter:
    def __init__(self, agents, reflection_model):
        self.agents = agents
        self.model = reflection_model
        self.routing_history = []  # 路由决策历史
        self.success_rates = {a.name: 0.5 for a in agents}  # 先验

    async def route(self, task: str) -> str:
        # Step 1: 获取历史成功率作为先验
        recent = [h for h in self.routing_history[-50:]]

        # Step 2: LLM决策 + 历史先验加权
        prompt = f"""Route this task to the best agent.

Task: {task}

Available agents and recent success rates:
{json.dumps(self.success_rates, indent=2)}

Recent routing outcomes:
{json.dumps(recent[-5:], indent=2)}

Pick the agent most likely to succeed."""

        decision = await self.model.invoke(prompt)
        return decision.agent_name

    async def record_outcome(self, task, agent_name, success, latency):
        """记录结果，更新成功率"""
        self.routing_history.append({
            "task": task[:100],  # 截断节省存储
            "agent": agent_name,
            "success": success,
            "latency": latency,
            "timestamp": datetime.now().isoformat()
        })
        # 滑动窗口更新成功率
        recent = [h for h in self.routing_history 
                  if h["agent"] == agent_name][-20:]
        self.success_rates[agent_name] = sum(
            h["success"] for h in recent
        ) / len(recent)

graph LR
    subgraph "路由分派决策流程对比"
        direction TB

        subgraph "栈1: 硬编码"
            T1[任务] -->|固定顺序| A1[Agent A]
            A1 -->|固定顺序| A2[Agent B]
        end

        subgraph "栈2: 管理器"
            T2[任务] --> SUP[Supervisor<br/>LLM决策]
            SUP -->|分析后分派| A3[Agent A]
            SUP -->|分析后分派| A4[Agent B]
        end

        subgraph "栈4: 语义"
            T5[任务] --> EMB[Embedding<br/>相似度]
            EMB -->|Top-K| A9[Agent A]
            EMB -->|Top-K| A10[Agent B]
        end
    end

# LangGraph混合路由：Supervisor + 条件分支
from langgraph.graph import StateGraph, END

def router(state):
    """自定义路由函数 - 可插入任意决策逻辑"""
    last_message = state["messages"][-1]

    # 规则优先
    if "code" in last_message.content.lower():
        return "coder"
    elif "research" in last_message.content.lower():
        return "researcher"

    # 语义匹配兜底
    return semantic_router.route(last_message.content)

graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("coder", coder_node)
graph.add_node("researcher", researcher_node)
graph.add_conditional_edges("supervisor", router)

# CrewAI Hierarchical Process - 管理器路由
crew = Crew(
    agents=[researcher, writer, editor, coder],
    tasks=[task1, task2, task3],  # 不指定agent!
    process=Process.hierarchical,  # Manager自动分派
    manager_llm="gpt-4.1",       # Manager使用的模型
)

# AG2自定义Selector - 语义+规则混合路由
def custom_selector(last_speaker, messages):
    last_msg = messages[-1]["content"]

    # 规则1: 如果上一轮是Researcher，下一轮必须是Writer
    if last_speaker == researcher:
        return writer

    # 规则2: 用语义匹配决定
    return semantic_match(last_msg, [coder, reviewer])

groupchat = ag2.GroupChat(
    agents=[researcher, writer, coder, reviewer],
    speaker_selection_method=custom_selector
)

graph TD
    START[多Agent路由选型] --> Q1{Agent数量?}
    Q1 -->|"≤3"| SIMPLE[栈1: 硬编码链式<br/>CrewAI Sequential]
    Q1 -->|"4-10"| Q2{任务模式稳定?}
    Q1 -->|"10+"| Q3{需要高并发?}

    Q2 -->|"是"| SIMPLE
    Q2 -->|"否"| Q4{路由准确率要求?}

    Q4 -->|"≥95%"| HYBRID[栈4+2: 语义初筛<br/>+ 管理器终判]
    Q4 -->|"≥85%"| SUPERVISOR[栈2: 管理器仲裁<br/>LangGraph Supervisor]

    Q3 -->|"是"| POOL[栈5: 池化负载<br/>+ 栈2管理器]
    Q3 -->|"否"| Q5{任务是否开放?}

    Q5 -->|"是(探索型)"| ADAPTIVE[栈3: LLM自适应<br/>AG2 GroupChat]
    Q5 -->|"否(流程型)"| HYBRID

    style SIMPLE fill:#c8e6c9
    style SUPERVISOR fill:#bbdefb
    style HYBRID fill:#e1bee7
    style POOL fill:#ffcdd2
    style ADAPTIVE fill:#fff9c4