AI Agent 编排系统设计模式：从单 Agent 到项目级自主编排的完整指南

// Agent 消息格式定义
interface AgentMessage {
  id: string;                    // 消息唯一标识
  sourceAgent: string;           // 发送方 Agent ID
  targetAgent: string;           // 接收方 Agent ID
  timestamp: number;             // 发送时间戳
  priority: 'low' | 'normal' | 'high' | 'critical';
  type: 'task' | 'result' | 'status' | 'error' | 'heartbeat';
  body: TaskBody | ResultBody | StatusBody | ErrorBody;
  metadata: {
    traceId: string;             // 分布式追踪 ID
    retryCount: number;          // 重试次数
    timeoutMs: number;           // 超时时间
    correlationId?: string;      // 关联消息 ID
  };
}

interface TaskBody {
  description: string;           // 任务描述
  input: Record<string, any>;    // 输入数据
  expectedOutput: string;        // 预期输出格式
  constraints: string[];         // 约束条件
  contextRefs: string[];         // 上下文引用（文件路径、URL 等）
}

// 基于 EventEmitter 的 Agent 通信框架
import { EventEmitter } from 'events';

class AgentOrchestrator extends EventEmitter {
  private agents: Map<string, Agent> = new Map();
  private messageQueue: AgentMessage[] = [];
  private traceLog: AgentMessage[] = [];

  registerAgent(agent: Agent): void {
    this.agents.set(agent.id, agent);
    agent.on('message', (msg: AgentMessage) => {
      this.handleMessage(msg);
    });
  }

  private handleMessage(msg: AgentMessage): void {
    this.traceLog.push(msg);
    const target = this.agents.get(msg.targetAgent);
    if (!target) {
      console.error(`Agent not found: ${msg.targetAgent}`);
      return;
    }
    if (msg.priority === 'critical') {
      this.messageQueue.unshift(msg);  // 高优先级插队
    } else {
      this.messageQueue.push(msg);
    }
    this.processQueue();
  }

  private processQueue(): void {
    while (this.messageQueue.length > 0) {
      const msg = this.messageQueue.shift()!;
      const target = this.agents.get(msg.targetAgent);
      target?.processMessage(msg);
    }
  }
}

import asyncio
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from enum import Enum
import time

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Task:
    id: str
    agent_id: str
    description: str
    input_data: Dict[str, Any]
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[Any] = None
    error: Optional[str] = None
    start_time: Optional[float] = None
    end_time: Optional[float] = None

class TaskScheduler:
    """异步任务调度器：负责任务分配、执行监控和结果收集"""
    
    def __init__(self, max_concurrent: int = 5):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.tasks: Dict[str, Task] = {}
        self.results: List[Task] = []
        self._running = True
    
    async def submit_task(self, task: Task) -> None:
        """提交任务到调度队列"""
        self.tasks[task.id] = task
        task.status = TaskStatus.PENDING
        asyncio.create_task(self._execute_task(task))
    
    async def _execute_task(self, task: Task) -> None:
        """执行单个任务（带并发控制）"""
        async with self.semaphore:
            task.status = TaskStatus.RUNNING
            task.start_time = time.time()
            try:
                # 这里调用实际的 Agent 执行逻辑
                result = await self._run_agent(task.agent_id, task.input_data)
                task.result = result
                task.status = TaskStatus.COMPLETED
            except Exception as e:
                task.error = str(e)
                task.status = TaskStatus.FAILED
            finally:
                task.end_time = time.time()
                self.results.append(task)
    
    async def _run_agent(self, agent_id: str, input_data: Dict) -> Any:
        """调用 Agent 执行（需根据实际实现替换）"""
        await asyncio.sleep(1)  # 模拟执行延迟
        return {"status": "success", "data": f"Result from {agent_id}"}
    
    async def get_all_results(self) -> List[Task]:
        """等待所有任务完成后返回结果"""
        while any(t.status in (TaskStatus.PENDING, TaskStatus.RUNNING) 
                  for t in self.tasks.values()):
            await asyncio.sleep(0.1)
        return self.results

// 代码审查编排系统 - 完整实现
import { EventEmitter } from 'events';

interface CodeChange {
  filePath: string;
  diff: string;
  language: string;
}

interface ReviewFinding {
  filePath: string;
  severity: 'critical' | 'high' | 'medium' | 'low' | 'info';
  category: 'security' | 'style' | 'performance' | 'bug' | 'maintainability';
  description: string;
  suggestion: string;
  lineNumbers?: number[];
}

interface ReviewResult {
  agentId: string;
  findings: ReviewFinding[];
  summary: string;
}

// Worker Agent 基类
abstract class ReviewAgent extends EventEmitter {
  abstract readonly id: string;
  abstract readonly role: string;

  async review(changes: CodeChange[]): Promise<ReviewResult> {
    const findings: ReviewFinding[] = [];
    for (const change of changes) {
      const result = await this.analyzeChange(change);
      findings.push(...result);
    }
    return { agentId: this.id, findings, summary: this.generateSummary(findings) };
  }

  protected abstract analyzeChange(change: CodeChange): Promise<ReviewFinding[]>;
  protected abstract generateSummary(findings: ReviewFinding[]): string;
}

// 安全检查 Agent
class SecurityAgent extends ReviewAgent {
  readonly id = 'security-reviewer';
  readonly role = '安全检查专家';

  protected async analyzeChange(change: CodeChange): Promise<ReviewFinding[]> {
    const findings: ReviewFinding[] = [];
    // 检查 SQL 注入
    if (change.diff.includes('SELECT') && change.diff.includes('+')) {
      findings.push({
        filePath: change.filePath,
        severity: 'critical',
        category: 'security',
        description: '可能存在 SQL 注入漏洞：使用字符串拼接构造 SQL 查询',
        suggestion: '使用参数化查询或 ORM 的查询构建器'
      });
    }
    // 检查敏感信息泄露
    const sensitivePatterns = [/password['"]?s*[:=]/i, /api[_-]?key['"]?s*[:=]/i, /secret['"]?s*[:=]/i];
    for (const pattern of sensitivePatterns) {
      if (pattern.test(change.diff)) {
        findings.push({
          filePath: change.filePath,
          severity: 'high',
          category: 'security',
          description: '代码中可能包含敏感信息',
          suggestion: '使用环境变量或密钥管理服务存储敏感信息'
        });
      }
    }
    return findings;
  }

  protected generateSummary(findings: ReviewFinding[]): string {
    const critical = findings.filter(f => f.severity === 'critical').length;
    const high = findings.filter(f => f.severity === 'high').length;
    return `安全检查完成：发现 ${critical} 个严重问题和 ${high} 个高风险问题`;
  }
}

// 主管 Agent：整合所有审查结果
class SupervisorAgent {
  private agents: ReviewAgent[] = [];

  addAgent(agent: ReviewAgent): void {
    this.agents.push(agent);
  }

  async reviewAll(changes: CodeChange[]): Promise<{
    report: ReviewResult[];
    consolidatedFindings: ReviewFinding[];
    summary: string;
  }> {
    // 并行执行所有 Agent 的审查
    const results = await Promise.all(
      this.agents.map(agent => agent.review(changes))
    );

    // 整合结果
    const allFindings = results.flatMap(r => r.findings);
    const deduplicated = this.deduplicateFindings(allFindings);
    const sorted = this.sortBySeverity(deduplicated);

    return {
      report: results,
      consolidatedFindings: sorted,
      summary: this.generateReportSummary(sorted)
    };
  }

  private deduplicateFindings(findings: ReviewFinding[]): ReviewFinding[] {
    const seen = new Set<string>();
    return findings.filter(f => {
      const key = `${f.filePath}:${f.category}:${f.description}`;
      if (seen.has(key)) return false;
      seen.add(key);
      return true;
    });
  }

  private sortBySeverity(findings: ReviewFinding[]): ReviewFinding[] {
    const order = { critical: 0, high: 1, medium: 2, low: 3, info: 4 };
    return findings.sort((a, b) => order[a.severity] - order[b.severity]);
  }

  private generateReportSummary(findings: ReviewFinding[]): string {
    const bySeverity: Record<string, number> = {};
    for (const f of findings) {
      bySeverity[f.severity] = (bySeverity[f.severity] || 0) + 1;
    }
    return `审查完成：${findings.length} 个问题 (${Object.entries(bySeverity).map(([s, c]) => `${c} ${s}`).join(', ')})`;
  }
}

"""
基于 Celery 的分布式 Agent 编排系统
适用于需要水平扩展的生产环境
"""
from celery import Celery, chain, group
from celery.result import AsyncResult
from dataclasses import dataclass
from typing import List, Dict, Any
import json

# Celery 应用配置
app = Celery(
    'agent_orchestrator',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)

app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Shanghai',
    enable_utc=True,
    task_acks_late=True,  # 确保任务执行完成后再确认
    worker_prefetch_multiplier=1,  # 每个 worker 一次只取一个任务
)

@dataclass
class OrchestrationTask:
    task_id: str
    agent_type: str
    input_data: Dict[str, Any]
    priority: int = 0

# Celery 任务定义
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def execute_agent_task(self, task_data: Dict[str, Any]) -> Dict[str, Any]:
    """执行单个 Agent 任务"""
    agent_type = task_data['agent_type']
    input_data = task_data['input_data']
    
    try:
        # 根据 agent_type 路由到具体的 Agent 实现
        if agent_type == 'security_analyzer':
            result = analyze_security(input_data)
        elif agent_type == 'style_checker':
            result = check_style(input_data)
        elif agent_type == 'performance_analyzer':
            result = analyze_performance(input_data)
        else:
            raise ValueError(f"Unknown agent type: {agent_type}")
        
        return {
            'status': 'success',
            'agent_type': agent_type,
            'result': result
        }
    except Exception as exc:
        # 自动重试
        raise self.retry(exc=exc)

def analyze_security(input_data: Dict) -> Dict:
    """安全检查 Agent 实现"""
    # 实际的安全分析逻辑
    return {"findings": [], "score": 0.95}

def check_style(input_data: Dict) -> Dict:
    """风格检查 Agent 实现"""
    return {"violations": [], "score": 0.90}

def analyze_performance(input_data: Dict) -> Dict:
    """性能分析 Agent 实现"""
    return {"bottlenecks": [], "score": 0.85}

@app.task
def consolidate_results(results: List[Dict[str, Any]]) -> Dict[str, Any]:
    """整合所有 Agent 的结果"""
    consolidated = {
        'total_tasks': len(results),
        'success_count': sum(1 for r in results if r['status'] == 'success'),
        'failure_count': sum(1 for r in results if r['status'] == 'failure'),
        'findings': []
    }
    for result in results:
        if result['status'] == 'success':
            consolidated['findings'].extend(
                result['result'].get('findings', [])
            )
    return consolidated

def run_orchestration(tasks: List[OrchestrationTask]) -> AsyncResult:
    """运行编排流程：并行执行 + 结果整合"""
    # 创建并行任务组
    parallel_tasks = group([
        execute_agent_task.s({
            'agent_type': t.agent_type,
            'input_data': t.input_data
        })
        for t in tasks
    ])
    
    # 链式连接整合任务
    workflow = chain(parallel_tasks, consolidate_results.s())
    
    # 异步执行
    return workflow.apply_async()