Agent 编排模式（二）：从 OpenAI Symphony 到多智能体协作范式

// Agent 编排核心接口定义
interface AgentRegistry {
  register(agentId: string, capabilities: string[]): void;
  findAgent(task: string): string | null;
  getCapabilities(agentId: string): string[];
}

interface Orchestrator {
  decomposeTask(task: string): SubTask[];
  schedule(subTask: SubTask): AgentAssignment;
  execute(assignment: AgentAssignment): Promise<Result>;
  aggregate(results: Result[]): FinalResult;
}

interface SubTask {
  id: string;
  description: string;
  requiredCapabilities: string[];
  dependencies: string[];
  timeoutMs: number;
}

interface AgentAssignment {
  subTaskId: string;
  agentId: string;
  input: Record<string, unknown>;
  retryCount: number;
}

// 混合编排器实现示例
class HybridOrchestrator {
  private groups: Map<string, AgentGroup> = new Map();
  private orchestrator: CentralOrchestrator;

  constructor(agents: Agent[]) {
    this.orchestrator = new CentralOrchestrator(agents);
    // 编排器负责顶层任务分解
    const groups = this.orchestrator.decomposeAndGroup(task);
    for (const g of groups) {
      this.groups.set(g.id, new AgentGroup(g.agents));
    }
  }

  async execute(task: Task): Promise<FinalResult> {
    // 第一步：编排器分解任务
    const subTasks = this.orchestrator.decompose(task);

    // 第二步：分配到各组
    const groupResults = await Promise.all(
      subTasks.map(async (st) => {
        const group = this.groups.get(st.groupId)!;
        // 组内 Agent 自主协商执行顺序（去中心化）
        return group.executeNegotiated(st);
      })
    );

    // 第三步：编排器聚合结果（中心化）
    return this.orchestrator.aggregate(groupResults);
  }
}

{
  "score_id": "score-2026-04-30-001",
  "version": "1.0",
  "movements": [
    {
      "movement_id": "research",
      "type": "parallel",
      "tasks": [
        {
          "task_id": "web-search",
          "agent": "search-agent",
          "input": { "query": "OpenAI Symphony 规范" },
          "timeout_ms": 30000
        },
        {
          "task_id": "doc-analysis",
          "agent": "analysis-agent",
          "input": { "source": "symphony-spec.md" },
          "timeout_ms": 60000
        }
      ]
    },
    {
      "movement_id": "synthesize",
      "type": "sequential",
      "depends_on": ["research"],
      "tasks": [
        {
          "task_id": "write-report",
          "agent": "writer-agent",
          "input": {
            "references": ["web-search.result", "doc-analysis.result"]
          },
          "timeout_ms": 120000
        }
      ]
    }
  ]
}

// 声明式状态机：基于配置的状态转换
interface StateMachineConfig {
  states: Record<string, Record<string, string>>;
  initialState: string;
}

class DeclarativeStateMachine {
  private config: StateMachineConfig;
  private currentState: string;

  constructor(config: StateMachineConfig) {
    this.config = config;
    this.currentState = config.initialState;
  }

  transition(event: string): string | null {
    const transitions = this.config.states[this.currentState];
    if (!transitions || !transitions[event]) {
      return null; // 无效转换
    }
    const nextState = transitions[event];
    this.currentState = nextState;
    return nextState;
  }

  getState(): string {
    return this.currentState;
  }
}

// 使用示例：Agent 编排状态机
const agentFSM = new DeclarativeStateMachine({
  states: {
    pending:   { dispatch: 'running' },
    running:   { complete: 'completed', fail: 'failed', timeout: 'retrying' },
    retrying:  { retry: 'running', escalate: 'failed' },
    completed: {},
    failed:    { restart: 'pending' },
  },
  initialState: 'pending',
});

import { StateGraph, END } from "@langchain/langgraph";
import { Annotation } from "@langchain/langgraph";

// 定义编排状态
const AgentState = Annotation.Root({
  codeChanges: Annotation<string[]>({
    reducer: (x, y) => x.concat(y),
    default: () => [],
  }),
  lintResults: Annotation<string[]>({
    reducer: (x, y) => x.concat(y),
    default: () => [],
  }),
  securityResults: Annotation<string[]>({
    reducer: (x, y) => x.concat(y),
    default: () => [],
  }),
  reviewReport: Annotation<string>,
  currentStep: Annotation<string>,
});

// Lint 检查节点
async function lintCheck(state: typeof AgentState.State) {
  const results: string[] = [];
  for (const file of state.codeChanges) {
    const lintOutput = await runLinter(file);
    if (lintOutput.violations.length > 0) {
      results.push(`[${file}] ${lintOutput.violations.length} 个规范问题`);
    }
  }
  return { lintResults: results, currentStep: "lint_done" };
}

// 安全扫描节点
async function securityScan(state: typeof AgentState.State) {
  const results: string[] = [];
  for (const file of state.codeChanges) {
    const vulns = await scanVulnerabilities(file);
    if (vulns.length > 0) {
      results.push(`[${file}] ${vulns.length} 个安全漏洞`);
    }
  }
  return { securityResults: results, currentStep: "security_done" };
}

// 生成审查报告节点
async function generateReport(state: typeof AgentState.State) {
  const report = [
    "# 代码审查报告",
    ...state.lintResults.map(r => `- ${r}`),
    ...state.securityResults.map(r => `- ${r}`),
    state.securityResults.length > 0
      ? "❌ 存在安全漏洞，建议修复后重新提交"
      : "✅ 代码质量良好，可以合并",
  ].join("\n");
  return { reviewReport: report, currentStep: "completed" };
}

// 构建编排图
const workflow = new StateGraph(AgentState)
  .addNode("lint", lintCheck)
  .addNode("security", securityScan)
  .addNode("review", generateReport)
  .addEdge("__start__", "lint")
  .addEdge("lint", "security")
  .addEdge("security", "review")
  .addEdge("review", END);

const app = workflow.compile();

# CrewAI 角色驱动编排示例
from crewai import Agent, Task, Crew, Process

# 定义角色
researcher = Agent(
    role="高级技术研究员",
    goal="深入调研 Agent 编排框架的技术细节",
    backstory="你是一位有 10 年经验的系统架构师，擅长分布式系统设计。",
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role="技术写作者",
    goal="将调研结果转化为清晰的技术文档",
    backstory="你是一位资深技术作家，善于将复杂概念用简单语言解释。",
    verbose=True,
    allow_delegation=False,
)

# 定义任务
research_task = Task(
    description="对比 LangGraph、CrewAI、AutoGen 三个框架",
    expected_output="一份包含编程模型、状态管理、容错能力的对比报告",
    agent=researcher,
)

write_task = Task(
    description="基于调研报告编写技术博客文章",
    expected_output="一篇 3000 字以上的深度技术博客",
    agent=writer,
)

# 创建 Crew 并执行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # 顺序执行
)

result = crew.kickoff()

// 熔断器模式实现
class CircuitBreaker {
  private state: 'closed' | 'open' | 'half-open' = 'closed';
  private failureCount = 0;
  private lastFailureTime = 0;

  constructor(
    private threshold: number,
    private resetTimeout: number
  ) {}

  async execute<T>(fn: () => Promise<T>): Promise<T> {
    if (this.state === 'open') {
      if (Date.now() - this.lastFailureTime > this.resetTimeout) {
        this.state = 'half-open';
      } else {
        throw new Error('熔断器打开，请求被拒绝');
      }
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  private onSuccess() {
    this.failureCount = 0;
    this.state = 'closed';
  }

  private onFailure() {
    this.failureCount++;
    this.lastFailureTime = Date.now();
    if (this.failureCount >= this.threshold) {
      this.state = 'open';
    }
  }
}

// 在编排系统中使用熔断器
const agentCircuitBreaker = new CircuitBreaker(
  5,      // 连续 5 次失败后打开熔断器
  30000   // 30 秒后进入半开状态
);

// 最佳实践：为 Agent 定义清晰的契约
interface SearchAgentInput {
  query: string;
  maxResults: number;
  language: 'zh' | 'en';
}

interface SearchAgentOutput {
  results: Array<{
    title: string;
    url: string;
    snippet: string;
    relevanceScore: number;
  }>;
  totalCount: number;
}

interface SearchAgentError {
  code: 'TIMEOUT' | 'RATE_LIMIT' | 'INVALID_QUERY' | 'NETWORK_ERROR';
  message: string;
  retryable: boolean;
}

// 契约驱动的 Agent 调用
async function callSearchAgent(
  input: SearchAgentInput,
  timeoutMs: number = 30000
): Promise<SearchAgentOutput> {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), timeoutMs);

  try {
    const response = await fetch('http://search-agent/api/search', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(input),
      signal: controller.signal,
    });

    if (!response.ok) {
      const error: SearchAgentError = await response.json();
      throw new Error(`[${error.code}] ${error.message}`);
    }

    return response.json();
  } finally {
    clearTimeout(timeout);
  }
}