多Agent编排三范式：DAG·事件驱动·Actor模型生产实测

graph TB
    subgraph "DAG/图状态机范式"
        A1[用户请求] --> B1[Planner Node]
        B1 --> C1[Router Edge]
        C1 --> D1[Agent A Node]
        C1 --> E1[Agent B Node]
        D1 --> F1[Merge Node]
        E1 --> F1
        F1 --> G1[Output]
    end

    subgraph "事件驱动范式"
        A2[Event Bus] --> B2[Event: task.created]
        B2 --> C2[Agent A 订阅]
        B2 --> D2[Agent B 订阅]
        C2 --> E2[Event: result.ready]
        D2 --> E2
        E2 --> F2[Aggregator]
    end

    subgraph "Actor范式"
        A3[Supervisor Actor] --> B3[Agent Actor A]
        A3 --> C3[Agent Actor B]
        B3 -->|消息| C3
        C3 -->|消息| B3
        B3 --> D3[结果汇聚]
        C3 --> D3
    end

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_task: str
    results: dict

def researcher(state: AgentState) -> dict:
    """研究Agent节点"""
    query = state["current_task"]
    findings = search_and_analyze(query)
    return {"messages": [f"Research: {findings}"], "results": {"research": findings}}

def coder(state: AgentState) -> dict:
    """编码Agent节点"""
    spec = state["results"].get("research", "")
    code = generate_code(spec)
    return {"messages": [f"Code: {code}"], "results": {"code": code}}

def reviewer(state: AgentState) -> dict:
    """审查Agent节点"""
    code = state["results"].get("code", "")
    review = review_code(code)
    return {"messages": [f"Review: {review}"]}

def should_review(state: AgentState) -> str:
    """条件路由边"""
    return "reviewer" if state["results"].get("code") else END

# 构建图
graph = StateGraph(AgentState)
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.add_node("reviewer", reviewer)
graph.add_edge("researcher", "coder")
graph.add_conditional_edges("coder", should_review)
graph.set_entry_point("researcher")

# 编译并执行
app = graph.compile(checkpointer=SqliteSaver("agent.db"))
result = app.invoke({"current_task": "Build REST API", "messages": [], "results": {}})

import boto3
import json
from strands import Agent, tool

# 事件总线客户端
eventbridge = boto3.client('events')

@tool
def publish_event(event_type: str, payload: dict):
    """发布事件到EventBridge"""
    eventbridge.put_events(
        Entries=[{
            'Source': 'multi-agent-system',
            'DetailType': event_type,
            'Detail': json.dumps(payload)
        }]
    )

# 仲裁Agent - 监听所有事件，决定路由
arbiter = Agent(
    name="Arbiter",
    system_prompt="""你是多Agent系统的仲裁者。
    收到任务后分析需求，决定分配给哪个专家Agent。
    通过发布事件来触发下游Agent。""",
    tools=[publish_event]
)

# 专家Agent - 各自订阅特定事件
researcher = Agent(
    name="Researcher",
    system_prompt="你是研究专家，收到research.request事件后执行调研。",
    tools=[publish_event]
)

coder = Agent(
    name="Coder", 
    system_prompt="你是编码专家，收到code.request事件后生成代码。",
    tools=[publish_event]
)

# 事件驱动流程：
# 1. 用户请求 → task.created事件
# 2. Arbiter监听 → 发布research.request
# 3. Researcher监听 → 完成 → 发布research.completed  
# 4. Arbiter监听 → 发布code.request
# 5. Coder监听 → 完成 → 发布code.completed

# Event Sourcing for Agent State
from dataclasses import dataclass
from datetime import datetime
from typing import List

@dataclass
class AgentEvent:
    agent_id: str
    event_type: str
    payload: dict
    timestamp: datetime
    sequence: int

class AgentEventStore:
    def __init__(self):
        self.streams = {}  # agent_id -> list of events

    def append(self, event: AgentEvent):
        if event.agent_id not in self.streams:
            self.streams[event.agent_id] = []
        self.streams[event.agent_id].append(event)

    def get_events(self, agent_id: str, from_seq: int = 0) -> List[AgentEvent]:
        return self.streams.get(agent_id, [])[from_seq:]

    def rebuild_state(self, agent_id: str) -> dict:
        """从事件流重建Agent当前状态"""
        state = {"tasks_completed": 0, "last_action": None}
        for event in self.get_events(agent_id):
            if event.event_type == "task.completed":
                state["tasks_completed"] += 1
                state["last_action"] = event.payload
        return state

# CQRS写端
store = AgentEventStore()
store.append(AgentEvent("researcher-1", "task.started", {"task": "analyze"}, datetime.now(), 0))
store.append(AgentEvent("researcher-1", "task.completed", {"result": "done"}, datetime.now(), 1))

# CQRS读端 - 构建物化视图
state = store.rebuild_state("researcher-1")

from autogen_core import AgentRuntime, MessageContext, RoutedAgent, message_handler
from autogen_core import SingleThreadedAgentRuntime

# 定义消息类型
class TaskMessage:
    def __init__(self, task: str, priority: int = 0):
        self.task = task
        self.priority = priority

class ResultMessage:
    def __init__(self, result: str, source: str):
        self.result = result
        self.source = source

# 定义Agent Actor
class ResearcherAgent(RoutedAgent):
    def __init__(self, name: str):
        super().__init__("Research tasks")
        self.name = name
        self._processed = 0  # Actor私有状态

    @message_handler
    async def handle_task(self, message: TaskMessage, ctx: MessageContext) -> ResultMessage:
        self._processed += 1  # 无需加锁，Actor串行处理邮箱消息
        result = await self._do_research(message.task)
        return ResultMessage(result=result, source=self.name)

    async def _do_research(self, task: str) -> str:
        # 实际研究逻辑
        return f"Research result for: {task}"

# 构建Actor运行时
runtime = SingleThreadedAgentRuntime()

# 注册Agent Actor
researcher1 = ResearcherAgent("researcher-alpha")
researcher2 = ResearcherAgent("researcher-beta")
runtime.register_agent(researcher1)
runtime.register_agent(researcher2)

# 发送消息
await runtime.send_message(TaskMessage("Analyze market trends"), researcher1.id)
await runtime.send_message(TaskMessage("Survey competitors"), researcher2.id)

quadrantChart
    title 多Agent编排范式定位矩阵
    x-axis "低确定性" --> "高确定性"
    y-axis "低吞吐" --> "高吞吐"
    quadrant-1 "确定性+吞吐"
    quadrant-2 "吞吐优先"
    quadrant-3 "弹性优先"
    quadrant-4 "确定性优先"
    "DAG/图状态机": [0.85, 0.40]
    "事件驱动/响应式": [0.35, 0.55]
    "Actor模型": [0.25, 0.90]
    "混合模式（趋势）": [0.60, 0.75]

┌─────────────────────────────────────────────────┐
│              编排控制层（DAG/图状态机）            │
│  LangGraph定义宏观工作流，确定Agent协作的整体骨架   │
└────────────────┬────────────────────────────────┘
                 │ 发出事件
┌────────────────▼────────────────────────────────┐
│              事件传输层（Event Bus）               │
│  EventBridge/Kafka 解耦Agent，提供弹性和背压      │
└────────────────┬────────────────────────────────┘
                 │ 消息分发
┌────────────────▼────────────────────────────────┐
│              执行运行层（Actor Pool）              │
│  AutoGen Core管理Agent实例池，处理高并发任务      │
└─────────────────────────────────────────────────┘

from langgraph.graph import StateGraph, END
from strands import Agent
import json

# Strands Agent池（Actor层）
agent_pool = {
    "researcher": Agent(name="Researcher", system_prompt="你是研究专家"),
    "coder": Agent(name="Coder", system_prompt="你是编码专家"),
    "tester": Agent(name="Tester", system_prompt="你是测试专家"),
}

# LangGraph编排层
class WorkflowState(dict):
    pass

def dispatch_to_agent(state: dict, agent_role: str) -> dict:
    """将任务分发到Agent Actor池"""
    agent = agent_pool[agent_role]
    result = agent(state["current_task"])
    state["results"][agent_role] = result
    state["messages"].append(f"{agent_role}: {result}")
    # 发布事件（事件层）
    publish_event(f"{agent_role}.completed", {"result": result})
    return state

def research_node(state):
    return dispatch_to_agent(state, "researcher")

def code_node(state):
    return dispatch_to_agent(state, "coder")

def test_node(state):
    return dispatch_to_agent(state, "tester")

# 构建混合图
graph = StateGraph(WorkflowState)
graph.add_node("research", research_node)
graph.add_node("code", code_node)
graph.add_node("test", test_node)
graph.add_edge("research", "code")
graph.add_edge("code", "test")
graph.add_edge("test", END)
graph.set_entry_point("research")

app = graph.compile()

graph LR
    subgraph "Agent通信层 - A2A"
        A1[Agent A] -->|A2A Task| A2[Agent B]
        A2 -->|A2A Message| A3[Agent C]
    end

    subgraph "工具调用层 - MCP"
        A1 -->|MCP Tool Call| M1[(数据库)]
        A2 -->|MCP Tool Call| M2[(API)]
        A3 -->|MCP Tool Call| M3[(文件系统)]
    end

graph TD
    START[多Agent系统选型] --> Q1{Agent协作流程是否确定?}
    Q1 -->|是，流程固定| Q2{是否需要审计追踪?}
    Q1 -->|否，动态变化| Q3{并发量级?}

    Q2 -->|是| DAG[DAG/图状态机<br/>LangGraph]
    Q2 -->|否| Q4{是否需要高性能?}

    Q4 -->|是| ACTOR[Actor模型<br/>AutoGen Core]
    Q4 -->|否| DAG

    Q3 -->|<100 Agent| EVENT[事件驱动<br/>AWS Strands]
    Q3 -->|≥100 Agent| HYBRID[混合模式<br/>DAG + Event + Actor]

    Q3 --> Q5{是否跨组织协作?}
    Q5 -->|是| A2A[A2A协议层<br/>Google ADK]
    Q5 -->|否| EVENT

    DAG --> FINAL[✅ 选定]
    EVENT --> FINAL
    ACTOR --> FINAL
    HYBRID --> FINAL
    A2A --> FINAL