AI Agent 编排模式与架构：从 ReAct 到多智能体协同的完整技术体系

from enum import Enum
from typing import Optional
from dataclasses import dataclass
import time

class TaskState(Enum):
    PENDING = "pending"
    RUNNING = "running"
    REVIEWING = "reviewing"
    TESTING = "testing"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class TaskResult:
    success: bool
    output: str
    metadata: dict = None

class WorkerAgent:
    """Worker Agent 基类，定义统一的执行接口"""
    
    def __init__(self, name: str, role: str, timeout: int = 60):
        self.name = name
        self.role = role
        self.timeout = timeout
        self.task_count = 0
    
    def execute(self, task: str, context: dict) -> TaskResult:
        """执行子任务，子类实现具体逻辑"""
        start_time = time.time()
        try:
            result = self._do_execute(task, context)
            self.task_count += 1
            return result
        except TimeoutError:
            return TaskResult(
                success=False,
                output=f"Worker {self.name} 超时 ({self.timeout}s)"
            )
        except Exception as e:
            return TaskResult(
                success=False,
                output=f"Worker {self.name} 执行失败: {str(e)}"
            )
    
    def _do_execute(self, task: str, context: dict) -> TaskResult:
        raise NotImplementedError

class AnalystAgent(WorkerAgent):
    def __init__(self):
        super().__init__("analyst", "需求分析", timeout=30)
    
    def _do_execute(self, task: str, context: dict) -> TaskResult:
        # 实际实现中调用 LLM API 进行需求分析
        return TaskResult(
            success=True,
            output=f"需求分析完成: {task}",
            metadata={"requirements": ["功能需求", "非功能需求"]}
        )

class SupervisorAgent:
    """Supervisor 编排器"""
    
    def __init__(self, workers: list[WorkerAgent], max_retries: int = 3):
        self.workers = {w.role: w for w in workers}
        self.max_retries = max_retries
        self.task_log = []
    
    def orchestrate(self, task: str) -> TaskResult:
        """编排完整的任务执行流程"""
        state = TaskState.PENDING
        context = {}
        
        # Phase 1: 需求分析
        state = TaskState.RUNNING
        result = self._execute_with_retry(
            "需求分析", self.workers.get("需求分析"), task, context
        )
        if not result.success:
            return TaskResult(success=False, output="需求分析失败")
        context["requirements"] = result.output
        
        # Phase 2: 代码生成
        result = self._execute_with_retry(
            "代码生成", self.workers.get("代码生成"), task, context
        )
        if not result.success:
            return TaskResult(success=False, output="代码生成失败")
        context["code"] = result.output
        state = TaskState.REVIEWING
        
        # Phase 3: 代码审查
        result = self._execute_with_retry(
            "代码审查", self.workers.get("代码审查"), task, context
        )
        if not result.success:
            return TaskResult(success=False, output="代码审查失败")
        context["review"] = result.output
        state = TaskState.TESTING
        
        # Phase 4: 测试生成
        result = self._execute_with_retry(
            "测试生成", self.workers.get("测试生成"), task, context)
        if not result.success:
            return TaskResult(success=False, output="测试生成失败")
        
        state = TaskState.COMPLETED
        return TaskResult(
            success=True,
            output="所有阶段完成",
            metadata=context
        )
    
    def _execute_with_retry(self, phase: str, worker, task, context):
        for attempt in range(self.max_retries):
            result = worker.execute(task, context)
            if result.success:
                self.task_log.append({"phase": phase, "status": "ok"})
                return result
            self.task_log.append({
                "phase": phase, 
                "attempt": attempt + 1,
                "status": "retry"
            })
        return TaskResult(success=False, output=f"{phase} 重试 {self.max_retries} 次后失败")

interface TaskState {
  id: string;
  currentPhase: string;
  status: 'pending' | 'running' | 'completed' | 'failed';
  results: Map<string, any>;
  startTime: number;
  retryCount: number;
}

interface WorkerDef {
  role: string;
  execute: (task: string, ctx: Record<string, any>) => Promise<{
    success: boolean;
    output: string;
    metadata?: Record<string, any>;
  }>;
  timeoutMs?: number;
}

class SupervisorEngine {
  private workers: Map<string, WorkerDef>;
  private activeTasks: Map<string, TaskState>;
  private maxRetries: number;

  constructor(
    workers: WorkerDef[],
    options: { maxRetries?: number } = {}
  ) {
    this.workers = new Map();
    workers.forEach(w => this.workers.set(w.role, w));
    this.activeTasks = new Map();
    this.maxRetries = options.maxRetries ?? 3;
  }

  async orchestrate(
    taskId: string,
    phases: Array<{ name: string; role: string; input?: string }>
  ): Promise<TaskState> {
    const state: TaskState = {
      id: taskId,
      currentPhase: '',
      status: 'running',
      results: new Map(),
      startTime: Date.now(),
      retryCount: 0,
    };
    this.activeTasks.set(taskId, state);

    for (const phase of phases) {
      state.currentPhase = phase.name;
      const worker = this.workers.get(phase.role);
      
      if (!worker) {
        state.status = 'failed';
        return state;
      }

      const result = await this.executeWithRetry(
        worker,
        phase.input || '',
        Object.fromEntries(state.results),
        state
      );

      if (!result.success) {
        state.status = 'failed';
        return state;
      }

      state.results.set(phase.name, result.output);
    }

    state.status = 'completed';
    return state;
  }

  private async executeWithRetry(
    worker: WorkerDef,
    task: string,
    ctx: Record<string, any>,
    state: TaskState
  ) {
    for (let attempt = 0; attempt < this.maxRetries; attempt++) {
      const timeout = worker.timeoutMs ?? 60000;
      const result = await Promise.race([
        worker.execute(task, ctx),
        new Promise((_, reject) =>
          setTimeout(() => reject(new Error('timeout')), timeout)
        ).catch(() => ({ success: false, output: 'timeout' }))
      ]) as any;

      if (result.success) return result;
      state.retryCount++;
    }
    return { success: false, output: 'max retries exceeded' };
  }
}

interface AgentMessage {
  id: string;
  taskId: string;
  fromAgent: string;
  toAgent: string;
  type: 'task_assign' | 'task_result' | 'error' | 'heartbeat';
  payload: Record<string, any>;
  timestamp: number;
  retryCount: number;
  maxRetries: number;
}

class AgentMessageBus {
  private queues: Map<string, AgentMessage[]> = new Map();
  private subscribers: Map<string, Set<(msg: AgentMessage) => void>> = new Map();
  private messageLog: AgentMessage[] = [];

  // 发布消息到指定 Agent 的队列
  publish(message: AgentMessage): void {
    message.id = message.id || crypto.randomUUID();
    message.timestamp = Date.now();
    
    const queue = this.queues.get(message.toAgent) || [];
    queue.push(message);
    this.queues.set(message.toAgent, queue);
    
    // 通知订阅者
    const subs = this.subscribers.get(message.toAgent);
    if (subs) {
      subs.forEach(cb => cb(message));
    }
    
    // 记录消息日志
    this.messageLog.push(message);
  }

  // Agent 订阅自己的消息队列
  subscribe(agentId: string, handler: (msg: AgentMessage) => void): void {
    const subs = this.subscribers.get(agentId) || new Set();
    subs.add(handler);
    this.subscribers.set(agentId, subs);
  }

  // 获取指定 Agent 的待处理消息
  getPendingMessages(agentId: string): AgentMessage[] {
    return this.queues.get(agentId) || [];
  }

  // 标记消息处理成功（从队列中移除）
  markProcessed(agentId: string, messageId: string): void {
    const queue = this.queues.get(agentId) || [];
    this.queues.set(
      agentId,
      queue.filter(m => m.id !== messageId)
    );
  }

  // 重试失败的消息
  retryFailed(agentId: string): void {
    const queue = this.queues.get(agentId) || [];
    for (const msg of queue) {
      if (msg.retryCount < msg.maxRetries) {
        msg.retryCount++;
        this.publish(msg);
      }
    }
  }

  // 查询指定任务的所有消息
  getTaskMessages(taskId: string): AgentMessage[] {
    return this.messageLog.filter(m => m.taskId === taskId);
  }
}

from dataclasses import dataclass, field
from typing import Any
import json
import time

@dataclass
class AgentStateSnapshot:
    """Agent 状态快照，用于回滚和审计"""
    agent_id: str
    task_id: str
    state: dict
    timestamp: float
    checksum: str = ""
    
    def compute_checksum(self) -> str:
        import hashlib
        data = json.dumps(self.state, sort_keys=True)
        self.checksum = hashlib.md5(data.encode()).hexdigest()
        return self.checksum

class StateTracker:
    """Agent 状态追踪器，支持快照、回滚和审计"""
    
    def __init__(self, max_snapshots: int = 100):
        self.snapshots: dict[str, list[AgentStateSnapshot]] = {}
        self.max_snapshots = max_snapshots
    
    def save_snapshot(self, agent_id: str, task_id: str, state: dict) -> AgentStateSnapshot:
        """保存状态快照"""
        snapshot = AgentStateSnapshot(
            agent_id=agent_id,
            task_id=task_id,
            state=state.copy(),
            timestamp=time.time()
        )
        snapshot.compute_checksum()
        
        key = f"{agent_id}:{task_id}"
        if key not in self.snapshots:
            self.snapshots[key] = []
        
        self.snapshots[key].append(snapshot)
        
        # 清理旧快照
        if len(self.snapshots[key]) > self.max_snapshots:
            self.snapshots[key] = self.snapshots[key][-self.max_snapshots:]
        
        return snapshot
    
    def rollback(self, agent_id: str, task_id: str, steps: int = 1) -> dict | None:
        """回滚到指定步数之前的状态"""
        key = f"{agent_id}:{task_id}"
        if key not in self.snapshots:
            return None
        
        snapshots = self.snapshots[key]
        if len(snapshots) <= steps:
            return None
        
        target = snapshots[-(steps + 1)]
        return target.state.copy()
    
    def get_audit_trail(self, agent_id: str, task_id: str) -> list[dict]:
        """获取完整的审计轨迹"""
        key = f"{agent_id}:{task_id}"
        if key not in self.snapshots:
            return []
        
        return [
            {
                "timestamp": s.timestamp,
                "checksum": s.checksum,
                "state_keys": list(s.state.keys())
            }
            for s in self.snapshots[key]
        ]