aue/docs/sdpes/architecture-design.md

Subagent-Driven 计划执行系统 (SDPES) - 架构设计文档

版本: v1.0.0
日期: 2026-04-24
状态: 设计完成，待实施

---

📋 系统概述

核心目标

构建一个模块化、可扩展、高可靠的基于子代理驱动的计划执行系统（Subagent-Driven Plan Execution System, SDPES），具备以下核心能力：

1. 智能任务分解 - 将复杂计划自动拆分为可独立执行的原子任务
2. 动态代理调度 - 根据任务特性分配最合适的专用子代理
3. 实时协作通信 - 支持代理间的信息共享、状态同步和结果汇总
4. 透明进度监控 - 全流程可视化追踪，异常实时告警与恢复
5. 插件化扩展 - 允许动态注册新代理类型和工具服务

适用场景

• ✅ 大型软件项目的分阶段实施
• ✅ 复杂系统的多模块并行开发
• ✅ AI辅助代码生成与审查工作流
• ✅ 自动化测试与持续集成流水线
• ✅ 多团队协作的任务编排系统

---

🏗️ 系统架构

整体架构图

┌─────────────────────────────────────────────────────────────┐
│                    SDPES 主控制器 (Master Agent)              │
│  ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐   │
│  │ 任务分解器 │ │ 任务调度器 │ │ 进度监控器 │ │ 异常处理器 │   │
│  └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ └─────┬─────┘   │
│        │              │             │             │          │
│  ┌─────▼──────────────▼─────────────▼─────────────▼─────┐  │
│  │                  事件总线 (Event Bus)                   │  │
│  └────────────────────┬─────────────────────────────────┘  │
│                       │                                    │
│  ┌────────────────────▼─────────────────────────────────┐  │
│  │               上下文管理器 (Context Manager)            │  │
│  └──────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                              │
        ┌─────────────────────┼─────────────────────┐
        ▼                     ▼                     ▼
┌───────────────┐    ┌───────────────┐    ┌───────────────┐
│ 实现者代理     │    │ 规范审查代理   │    │ 质量审查代理   │
│ (Implementer) │    │ (SpecReviewer)│    │ (CodeReviewer)│
└───────┬───────┘    └───────┬───────┘    └───────┬───────┘
        │                    │                    │
        ▼                    ▼                    ▼
┌───────────────┐    ┌───────────────┐    ┌───────────────┐
│ 工具集 (Tools) │    │ 规范检查器     │    │ 代码分析器     │
│ - 文件读写      │    │ - 需求匹配     │    │ - Lint检查    │
│ - API调用       │    │ - 边界验证     │    │ - 安全扫描    │
│ - Git操作       │    │ - 测试覆盖     │    │ - 性能评估    │
└───────────────┘    └───────────────┘    └───────────────┘

---

🎯 层级关系与职责定义

第一层：主控制器 (Master Controller)

职责： 系统总指挥，负责全局协调与决策

interface MasterController {
  // 核心能力
  loadPlan(planPath: string): Promise<ExecutionPlan>
  decomposeTasks(plan: ExecutionPlan): Task[]
  scheduleTask(task: Task): SubAgent
  monitorProgress(): ProgressReport
  
  // 决策能力
  selectAgent(taskType: TaskType): AgentType
  handleFailure(failure: FailureEvent): RecoveryAction
  optimizeSchedule(): void
  
  // 通信能力
  broadcast(event: SystemEvent): void
  collectResults(): AggregateResult
}

关键组件：

组件名称	职责	输入	输出
任务分解器	将复杂计划拆分为原子任务	ExecutionPlan	Task[]
任务调度器	分配任务给合适的子代理	Task + AgentPool	DispatchOrder
进度监控器	追踪所有任务的执行状态	StatusEvents	ProgressReport
异常处理器	捕获错误并触发恢复策略	ErrorEvent	RecoveryAction

---

第二层：专业子代理 (Specialized SubAgents)

2.1 实现者代理 (Implementer Agent)

职责： 执行具体的编码/实施任务

interface ImplementerAgent {
  id: string
  type: 'implementer'
  
  // 能力接口
  implement(task: Task, context: Context): ImplementationResult
  selfReview(implementation: ImplementationResult): ReviewFeedback
  askQuestion(question: Question): Promise<Answer>
  
  // 工具依赖
  tools: ToolSet
}

专业领域：

• 前端开发（Vue/React/Angular）
• 后端开发（Java/Python/Node.js）
• 数据库设计与迁移
• API集成与测试
• 文档编写

2.2 规范审查代理 (Spec Compliance Reviewer)

职责： 验证实现是否符合原始规范要求

interface SpecReviewerAgent {
  id: string
  type: 'spec-reviewer'
  
  // 能力接口
  reviewCompliance(
    implementation: ImplementationResult,
    spec: TaskSpecification
  ): SpecReviewReport
  
  checkRequirements(requirement: Requirement): boolean
  identifyGaps(spec: Specification, implementation: Code): Gap[]
}

审查维度：

• ✅ 功能完整性（是否实现了所有需求）
• ✅ 边界条件（是否处理了边界情况）
• ✅ 接口一致性（是否符合API规范）
• ✅ 测试覆盖（是否有对应测试用例）
• ⚠️ 过度实现（是否实现了未要求的功能）

2.3 质量审查代理 (Code Quality Reviewer)

职责： 评估代码质量和技术债务

interface CodeQualityReviewerAgent {
  id: string
  type: 'code-quality-reviewer'
  
  // 能力接口
  reviewQuality(code: CodeBase): QualityReport
  detectSmells(code: Code): CodeSmell[]
  assessSecurity(vulnerabilities: Vulnerability[]): SecurityScore
  measurePerformance(metrics: PerformanceMetrics): PerformanceGrade
}

质量维度：

• 可读性（命名规范、注释完整性）
• 可维护性（模块化程度、耦合度）
• 性能（时间复杂度、空间复杂度）
• 安全性（SQL注入、XSS等漏洞）
• 测试质量（覆盖率、断言有效性）

---

第三层：工具与服务层 (Tool & Service Layer)

interface ToolSet {
  // 文件操作
  fileSystem: FileSystemTool
  gitOperations: GitTool
  
  // 代码分析
  linter: LinterTool
  securityScanner: SecurityTool
  performanceProfiler: ProfilerTool
  
  // 外部服务
  apiClient: APIClient
  notificationService: NotificationService
}

---

🔧 核心机制详解

1. 任务分解机制 (Task Decomposition)

算法流程：

输入: ExecutionPlan (包含多个Task)
输出: AtomicTask[] (可独立执行的最小单元)

步骤:
1. 解析计划结构，提取所有Task节点
2. 分析Task之间的依赖关系（DAG）
3. 识别可并行的Task组
4. 为每个AtomicTask分配唯一ID和优先级
5. 生成上下文依赖清单
6. 输出分解结果 + 执行拓扑图

示例：

// 原始计划
const plan = {
  tasks: [
    { id: 'T1', name: '创建API模块', dependsOn: [] },
    { id: 'T2', name: '重构UI组件', dependsOn: ['T1'] },
    { id: 'T3', name: '编写测试', dependsOn: ['T1', 'T2'] },
    { id: 'T4', name: '性能优化', dependsOn: ['T3'] }
  ]
}

// 分解后（支持并行）
const executionGroups = [
  [{ task: 'T1', agents: [implementer] }],           // 第1批
  [{ task: 'T2', agents: [implementer] }],           // 第2批（依赖T1）
  [{ task: 'T3', agents: [implementer, tester] }],   // 第3批（依赖T1,T2）
  [{ task: 'T4', agents: [optimizer, reviewer] }]    // 第4批（依赖T3）
]

---

2. 任务调度算法 (Scheduling Algorithm)

策略选择矩阵：

因素	权重	说明
任务优先级	30%	紧急任务优先
代理可用性	25%	选择空闲代理
专业匹配度	20%	技能匹配度评分
上下文相关性	15%	复用已加载的上下文
资源消耗预估	10%	平衡负载

调度伪代码：

function scheduleTask(task: Task, agentPool: AgentPool): SchedulingDecision {
  const candidates = agentPool.filter(agent => 
    canHandle(agent, task.type) && isAvailable(agent)
  )
  
  if (candidates.length === 0) {
    return queueForLater(task) // 无可用代理，排队等待
  }
  
  const scored = candidates.map(agent => ({
    agent,
    score: calculateScore(agent, task)
  }))
  
  const best = scored.sort((a, b) => b.score - a.score)[0]
  
  return {
    assignedTo: best.agent,
    estimatedDuration: estimateDuration(task, best.agent),
    contextToInject: gatherRelevantContext(task)
  }
}

---

3. 通信协议 (Communication Protocol)

消息格式规范

// 基础消息格式
interface Message {
  id: string           // UUID
  timestamp: number    // Unix timestamp
  source: AgentId      // 发送方ID
  target: AgentId      // 接收方ID ('*' 表示广播)
  type: MessageType    // 消息类型
  payload: any         // 消息体
  metadata?: Record<string, any>  // 元数据
}

// 消息类型枚举
enum MessageType {
  // 任务相关
  TASK_ASSIGNED = 'task:assigned',
  TASK_STARTED = 'task:started',
  TASK_PROGRESS = 'task:progress',
  TASK_COMPLETED = 'task:completed',
  TASK_FAILED = 'task:failed',
  
  // 审查相关
  REVIEW_REQUESTED = 'review:request',
  REVIEW_RESULT = 'review:result',
  REVIEW_APPROVED = 'review:approved',
  REVIEW_REJECTED = 'review:rejected',
  
  // 协作相关
  QUESTION_ASKED = 'collab:question',
  QUESTION_ANSWERED = 'collab:answer',
  CONTEXT_SHARED = 'collab:context',
  
  // 系统事件
  SYSTEM_ERROR = 'sys:error',
  SYSTEM_WARNING = 'sys:warning',
  HEARTBEAT = 'sys:heartbeat'
}

事件总线实现

class EventBus {
  private subscribers: Map<MessageType, Subscriber[]> = new Map()
  
  subscribe(type: MessageType, handler: Subscriber): UnsubscribeFn {
    if (!this.subscribers.has(type)) {
      this.subscribers.set(type, [])
    }
    this.subscribers.get(type)!.push(handler)
    
    return () => this.unsubscribe(type, handler)
  }
  
  publish(message: Message): void {
    const handlers = this.subscribers.get(message.type) || []
    handlers.forEach(handler => handler(message))
    
    // 广播到通配符订阅者
    const wildcardHandlers = this.subscribers.get('*') || []
    wildcardHandlers.forEach(handler => handler(message))
  }
  
  async requestResponse<T>(
    message: Message,
    timeout: number = 30000
  ): Promise<T> {
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        unsubscribe()
        reject(new Error('Request timeout'))
      }, timeout)
      
      const unsubscribe = this.subscribe(
        message.type.replace(':request', ':response'),
        (response) => {
          clearTimeout(timer)
          resolve(response.payload)
        }
      )
      
      this.publish(message)
    })
  }
}

---

4. 结果汇总机制 (Result Aggregation)

汇总流程：

各子代理执行结果
       ↓
  ┌─────────────┐
  │ 结果收集器   │ ← 收集所有Task的ImplementationResult
  └──────┬──────┘
         ↓
  ┌─────────────┐
  │ 一致性校验   │ ← 检查结果间是否存在冲突
  └──────┬──────┘
         ↓
  ┌─────────────┐
  │ 合并引擎     │ ← 将多个结果合并为统一输出
  └──────┬──────┘
         ↓
  ┌─────────────┐
  │ 最终报告     │ ← 生成AggregateResult
  └─────────────┘

数据结构：

interface AggregateResult {
  executionId: string
  startTime: number
  endTime: number
  totalDuration: number
  
  tasks: {
    total: number
    completed: number
    failed: number
    skipped: number
  }
  
  results: TaskResult[]
  
  metrics: {
    successRate: number
    averageDuration: number
    qualityScore: number
  }
  
  artifacts: {
    codeChanges: FileChange[]
    testReports: TestReport[]
    documentation: Document[]
  }
  
  recommendations: string[]  // 改进建议
}

---

5. 进度监控与异常处理 (Monitoring & Exception Handling)

监控仪表板数据模型

interface ProgressDashboard {
  overview: {
    overallProgress: number      // 0-100
    currentPhase: string         // 当前阶段
    elapsedTime: number          // 已耗时间（秒）
    estimatedRemaining: number   // 预估剩余时间（秒）
  }
  
  tasks: TaskStatus[]
  
  timeline: TimelineEvent[]     // 时间线事件流
  
  alerts: Alert[]                // 活跃的告警
}

interface TaskStatus {
  taskId: string
  name: string
  status: 'pending' | 'running' | 'completed' | 'failed' | 'blocked'
  progress: number               // 0-100
  assignedAgent: string
  startTime?: number
  endTime?: number
  error?: ErrorInfo
}

异常分类与处理策略

异常类型	严重级别	自动处理策略	人工介入阈值
网络超时	Medium	重试3次，指数退避	连续失败5次
认证失败	High	刷新Token，重试1次	Token无效时
语法错误	Low	子代理自修复	无法修复时
规范偏离	Medium	要求重新实现	2次审查不通过
资源不足	High	降低并发度，排队等待	所有代理忙碌>60s
系统崩溃	Critical	回滚到上一个检查点	数据丢失风险

异常恢复流程：

async function handleException(error: ErrorEvent): Promise<RecoveryAction> {
  const severity = classifySeverity(error)
  const strategy = getRecoveryStrategy(error.type, severity)
  
  switch (strategy) {
    case 'retry':
      return retryWithBackoff(error.task, { maxRetries: 3, baseDelay: 1000 })
      
    case 'fallback':
      return executeFallbackPlan(error.task)
      
    case 'escalate':
      await notifyHumanOperator({
        error,
        context: getCurrentContext(),
        suggestedActions: generateSuggestions(error)
      })
      return waitForHumanDecision()
      
    case 'rollback':
      return rollbackToCheckpoint(error.checkpointId)
      
    case 'skip':
      markTaskAsSkipped(error.task, error.reason)
      continueWithNextTask()
  }
}

---

🔌 动态扩展机制 (Extension Mechanism)

插件注册接口

interface PluginManifest {
  name: string
  version: string
  description: string
  
  // 注册的组件
  agents?: AgentDefinition[]
  tools?: ToolDefinition[]
  schedulers?: SchedulerDefinition[]
  reviewers?: ReviewerDefinition[]
  
  // 生命周期钩子
  hooks?: {
    onTaskStart?: (task: Task) => Promise<void>
    onTaskComplete?: (result: TaskResult) => Promise<void>
    onError?: (error: Error) => Promise<void>
  }
  
  // 依赖声明
  dependencies?: string[]
}

class PluginManager {
  private plugins: Map<string, Plugin> = new Map()
  
  async register(manifest: PluginManifest): Promise<void> {
    // 验证依赖
    await this.validateDependencies(manifest)
    
    // 加载插件
    const plugin = await this.loadPlugin(manifest)
    
    // 注册组件
    manifest.agents?.forEach(agent => this.agentRegistry.register(agent))
    manifest.tools?.forEach(tool => this.toolRegistry.register(tool))
    
    // 绑定生命周期钩子
    manifest.hooks?.forEach(this.bindHook)
    
    this.plugins.set(manifest.name, plugin)
  }
  
  async unregister(pluginName: string): Promise<void> {
    const plugin = this.plugins.get(pluginName)
    if (!plugin) throw new Error(`Plugin ${pluginName} not found`)
    
    // 反注册组件
    plugin.manifest.agents?.forEach(agent => this.agentRegistry.unregister(agent.id))
    
    // 清理资源
    await plugin.cleanup()
    
    this.plugins.delete(pluginName)
  }
}

示例：自定义代理插件

// my-custom-agent-plugin.ts
const customPlugin: PluginManifest = {
  name: 'security-specialist-agent',
  version: '1.0.0',
  description: '专注于安全审计的专业代理',
  
  agents: [{
    id: 'security-auditor',
    type: 'specialized-reviewer',
    capabilities: [
      'vulnerability-scan',
      'penetration-test',
      'compliance-check'
    ],
    tools: ['owasp-zap', 'sonarqube', 'bandit']
  }],
  
  hooks: {
    onTaskComplete: async (result) => {
      // 在每个任务完成后自动运行安全扫描
      if (result.type === 'implementation') {
        await runSecurityScan(result.artifacts)
      }
    }
  },
  
  dependencies: ['core-engine-v2.0']
}

---

📊 接口规范 (Interface Specifications)

对外暴露的主接口

class SDPES {
  // 初始化
  constructor(config: SDPESConfig)
  async initialize(): Promise<void>
  
  // 计划管理
  async loadPlan(path: string): Promise<Plan>
  async validatePlan(plan: Plan): ValidationResult
  async executePlan(plan: Plan): Promise<ExecutionSession>
  
  // 执行控制
  pause(): void
  resume(): void
  cancel(): void
  
  // 状态查询
  getStatus(): ExecutionStatus
  getProgress(): ProgressDashboard
  getResults(): AggregateResult
  
  // 事件监听
  on(event: string, handler: Function): UnsubscribeFn
  
  // 扩展管理
  registerPlugin(plugin: PluginManifest): Promise<void>
  unregisterPlugin(name: string): Promise<void>
}

配置接口

interface SDPESConfig {
  // 并发控制
  maxConcurrentTasks: number        // 最大并发任务数（默认：3）
  maxRetries: number                // 最大重试次数（默认：3）
  retryDelay: number                // 重试基础延迟ms（默认：1000）
  
  // 超时设置
  taskTimeout: number               // 单个任务超时s（默认：300）
  sessionTimeout: number            // 整个会话超时s（默认：3600）
  
  // 日志配置
  logLevel: 'debug' | 'info' | 'warn' | 'error'
  logPath?: string                 // 日志文件路径
  
  // 代理配置
  defaultAgentConfig: AgentConfig
  customAgents?: AgentDefinition[]
  
  // 插件配置
  plugins?: string[]               // 要加载的插件列表
  
  // 存储配置
  storageBackend: 'memory' | 'file' | 'database'
  storageOptions?: Record<string, any>
}

---

🚀 使用示例

基础用法

import { SDPES } from '@sdpes/core'

// 创建实例
const sdpes = new SDPES({
  maxConcurrentTasks: 3,
  logLevel: 'info',
  defaultAgentConfig: {
    timeout: 300,
    retryCount: 3
  }
})

// 初始化
await sdpes.initialize()

// 加载计划
const plan = await sdpes.loadPlan('./plans/aichat-refactor.md')

// 执行
const session = await sdpes.executePlan(plan)

// 监听进度
sdpes.on('progress:update', (dashboard) => {
  console.log(`Progress: ${dashboard.overview.progress}%`)
})

sdpes.on('task:complete', (taskResult) => {
  console.log(`✅ Task "${taskResult.name}" completed`)
})

sdpes.on('error', (error) => {
  console.error(`❌ Error: ${error.message}`)
})

// 获取最终结果
const result = await sdpes.getResults()
console.log('Final Report:', result)

// 导出报告
await exportReport(result, './reports/execution-report.md')

高级用法 - 自定义代理

// 注册自定义代理
await sdpes.registerPlugin({
  name: 'my-expert-agent',
  agents: [{
    id: 'vue-specialist',
    type: 'implementer',
    expertise: ['vue3', 'composition-api', 'vite'],
    tools: ['eslint-plugin-vue', 'vue-devtools']
  }]
})

// 使用自定义代理执行特定任务
const specializedPlan = {
  tasks: [{
    id: 't1',
    name: '优化Vue组件性能',
    requiredAgent: 'vue-specialist',
    specification: {...}
  }]
}

await sdpes.executePlan(specializedPlan)

---

📈 性能与可靠性指标

目标指标

指标	目标值	说明
任务成功率	>98%	成功完成的任务占比
平均响应时间	<2s	从调度到开始执行的延迟
资源利用率	>85%	代理活跃时间占比
异常恢复率	>95%	自动恢复成功的异常占比
并发吞吐量	10 tasks/min	系统最大处理能力

可靠性保障

1. 检查点机制 - 每完成一个Task自动保存状态
2. 幂等性保证 - 相同任务多次执行结果一致
3. 事务性操作 - 关键操作的原子性保证
4. 优雅降级 - 部分失败不影响整体执行

---

🔒 安全考虑

1. 权限控制 - 子代理只能访问授权的资源
2. 沙箱隔离 - 每个子代理在独立环境中运行
3. 日志审计 - 所有操作记录完整日志链
4. 敏感信息保护 - Token、密钥等加密存储

---

📝 后续演进路线

Phase 1 (当前) - MVP版本

• ✅ 基础框架搭建
• ✅ 三种核心代理实现
• ✅ 同步串行执行模式
• ✅ 基础监控功能

Phase 2 - 增强版

• 并行任务执行
• 更丰富的代理类型
• Web UI可视化界面
• RESTful API接口

Phase 3 - 企业级

• 分布式部署支持
• 多租户隔离
• AI驱动的智能调度
• 与CI/CD系统集成

---

📚 参考文档

• Subagent-Driven Development Skill
• AIChat 流式接口重构计划
• 最佳实践指南

---

文档状态： ✅ 已完成设计评审
下一步： 开始实施核心引擎模块