# 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)

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

```typescript
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)

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

```typescript
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)

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

```typescript
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)

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

```typescript
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)

```typescript
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. 输出分解结果 + 执行拓扑图
```

**示例：**

```javascript
// 原始计划
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% | 平衡负载 |

**调度伪代码：**

```typescript
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)

#### 消息格式规范

```typescript
// 基础消息格式
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'
}
```

#### 事件总线实现

```typescript
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
  └─────────────┘
```

**数据结构：**

```typescript
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)

#### 监控仪表板数据模型

```typescript
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 | 回滚到上一个检查点 | 数据丢失风险 |

**异常恢复流程：**

```typescript
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)

### 插件注册接口

```typescript
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)
  }
}
```

### 示例：自定义代理插件

```typescript
// 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)

### 对外暴露的主接口

```typescript
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>
}
```

### 配置接口

```typescript
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>
}
```

---

## 🚀 使用示例

### 基础用法

```typescript
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')
```

### 高级用法 - 自定义代理

```typescript
// 注册自定义代理
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](./skills/subagent-driven-development.md)
- [AIChat 流式接口重构计划](./plans/2026-04-24-aichat-sse-refactor.md)
- [最佳实践指南](./docs/best-practices.md)

---

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