多Agent编排三范式：DAG·事件·Actor六栈实测

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_agent: str
    task_queue: list

def router(state: AgentState) -> str:
    """根据任务类型决定下一个Agent"""
    task = state["task_queue"][0] if state["task_queue"] else None
    if task and task["type"] == "research":
        return "research_agent"
    elif task and task["type"] == "code":
        return "code_agent"
    return END

graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_fn)
graph.add_node("research_agent", research_fn)
graph.add_node("code_agent", code_fn)
graph.add_conditional_edges("supervisor", router)
graph.add_edge("research_agent", "supervisor")
graph.add_edge("code_agent", "supervisor")
graph.set_entry_point("supervisor")

app = graph.compile(checkpointer=SqliteSaver(conn))

// Temporal多Agent工作流示例
import { proxyActivities, workflowInfo } from '@temporalio/workflow';

const { researchAgent, codeAgent, reviewAgent } = proxyActivities({
  startToCloseTimeout: '5 minutes',
  retry: { maximumAttempts: 3, backoffCoefficient: 2 },
});

export async function multiAgentWorkflow(task: string): Promise<string> {
  const spec = await researchAgent(task);

  // 并行Spawn多个编码Agent
  const codeResults = await Promise.all([
    codeAgent(spec.moduleA),
    codeAgent(spec.moduleB),
    codeAgent(spec.moduleC),
  ]);

  // 审查Agent串行检查
  for (const result of codeResults) {
    const review = await reviewAgent(result);
    if (!review.approved) {
      // 退回重做，Workflow自动记录完整历史
      await codeAgent(review.feedback);
    }
  }

  return 'completed';
}

import { Agent, Workflow, Step } from '@mastra/core';

const researcher = new Agent({
  name: 'researcher',
  instructions: 'You are a research specialist...',
  model: openai('gpt-4o'),
});

const writer = new Agent({
  name: 'writer', 
  instructions: 'You are a content writer...',
  model: openai('gpt-4o'),
});

const pipeline = new Workflow({
  name: 'content-pipeline',
  triggerSchema: z.object({ topic: z.string() }),
})
  .step(new Step({ id: 'research', execute: async (ctx) => researcher.run(ctx) }))
  .step(new Step({ id: 'write', execute: async (ctx) => writer.run(ctx) }))
  .after('research')
  .then('write');

const { runId, result } = await pipeline.run({ topic: 'multi-agent systems' });

# conductor.yaml - 多Agent代码审查流水线
name: code-review-pipeline
description: "Multi-agent code review with static analysis"
agents:
  - id: linter
    model: gpt-4o-mini
    prompt: "Review code style and formatting issues"
    tools: [eslint, prettier]
  - id: security
    model: gpt-4o
    prompt: "Identify security vulnerabilities"  
    tools: [semgrep, bandit]
  - id: architect
    model: claude-sonnet-4-6
    prompt: "Evaluate architectural decisions"
    tools: [read_file, search_code]
flow:
  - parallel:
      - agent: linter
        input: "${github.diff}"
      - agent: security
        input: "${github.diff}"
  - sequential:
      - agent: architect
        input: 
          diff: "${github.diff}"
          linter_result: "${linter.output}"
          security_result: "${security.output}"
  - output:
      summary: "${architect.output}"
      approved: "${architect.output.approved}"

from strands_agents import Agent, Graph

researcher = Agent(name="researcher", 
                   model="nova-pro",
                   tools=[web_search, read_url])
coder = Agent(name="coder",
              model="nova-pro", 
              tools=[write_file, execute_code])
reviewer = Agent(name="reviewer",
                 model="nova-pro",
                 tools=[read_file, diff])

graph = Graph()
graph.add_node("research", researcher)
graph.add_node("code", coder)
graph.add_node("review", reviewer)
graph.add_edge("research", "code")
graph.add_edge("code", "review")
graph.add_condition("review", 
    lambda r: "code" if not r.approved else END)

result = graph.run({"task": "Build a REST API for todos"})

# AOrchestra核心逻辑（简化）
from aorchestra import Orchestrator, SubAgentSpec

orchestrator = Orchestrator(
    model="gpt-4o",
    max_sub_agents=10,
    sub_agent_lifetime="task",  # 任务完成后自动销毁
)

# 指挥家分析任务后动态创建子Agent
result = orchestrator.run(
    task="Analyze this repository and generate comprehensive documentation",
    # 不预定义Agent——orchestrator自行决定需要哪些专家
    available_tools=[file_read, search, write, git],
)
# 内部过程：
# 1. orchestrator分析任务，决定需要：
#    - repo_analyzer (指令: 分析代码结构)
#    - api_doc_writer (指令: 编写API文档)  
#    - example_generator (指令: 生成使用示例)
# 2. 动态创建3个子Agent，分配工具子集
# 3. 监控执行，按需创建额外Agent
# 4. 子Agent完成后自动回收资源

graph TB
    subgraph "DAG范式"
        LG[LangGraph<br/>StateGraph]
        CD[Conductor<br/>YAML DAG]
        SG[Strands Graph<br/>AWS Native]
    end

    subgraph "Event-Driven范式"
        TP[Temporal<br/>Durable Workflow]
        MA[Mastra<br/>TS Workflow]
    end

    subgraph "动态编队范式"
        AO[AOrchestra<br/>On-Demand Spawn]
    end

    subgraph "选型决策"
        Q1{确定性优先?}
        Q2{长时任务?}
        Q3{TS技术栈?}
        Q4{云锁定可接受?}
        Q5{任务结构不确定?}
    end

    Q1 -->|是| CD
    Q1 -->|否| Q2
    Q2 -->|是| TP
    Q2 -->|否| Q3
    Q3 -->|是| MA
    Q3 -->|否| Q4
    Q4 -->|是| SG
    Q4 -->|否| Q5
    Q5 -->|是| AO
    Q5 -->|否| LG

    style LG fill:#4CAF50,color:#fff
    style TP fill:#2196F3,color:#fff
    style MA fill:#FF9800,color:#fff
    style CD fill:#9C27B0,color:#fff
    style SG fill:#FF5722,color:#fff
    style AO fill:#607D8B,color:#fff

sequenceDiagram
    participant User
    participant Planner as DAG Planner<br/>(LangGraph)
    participant Bus as Event Bus<br/>(Temporal)
    participant A1 as Agent 1<br/>Research
    participant A2 as Agent 2<br/>Analysis
    participant A3 as Agent 3<br/>Synthesis

    User->>Planner: 提交任务
    Planner->>Planner: 编译DAG拓扑
    Planner->>Bus: 发布 task.assigned 事件
    Bus->>A1: research.start
    A1->>Bus: research.complete(数据)
    Bus->>A2: analysis.start(数据)
    Note over A2: 运行时发现需要<br/>额外的数据清洗Agent
    A2->>Bus: spawn.request(数据清洗)
    Bus-->>A2: 新Agent创建确认
    A2->>Bus: analysis.complete(结果)
    Bus->>A3: synthesis.start(结果)
    A3->>Bus: synthesis.complete(报告)
    Bus->>Planner: workflow.completed
    Planner->>User: 返回最终结果