1377 lines
38 KiB
Markdown
1377 lines
38 KiB
Markdown
---
|
||
description: "迭代执行与反馈：接收任务树，批量执行，动态适应"
|
||
---
|
||
|
||
# /runtime.iterate - 迭代执行与反馈循环
|
||
|
||
## 核心目标
|
||
|
||
**我们不是一次性完成者，我们是持续交付者**。
|
||
|
||
当我们通过`/runtime.plan`生成了任务树后，我们通过迭代循环：
|
||
1. **批量执行** - 一次执行一批可并行任务
|
||
2. **收集反馈** - 每个任务成功/失败/产生新发现
|
||
3. **动态适应** - 根据反馈调整计划
|
||
4. **循环直至完成** - 持续迭代直到所有任务完成
|
||
|
||
```
|
||
任务树（来自plan）
|
||
↓
|
||
[执行批次1] → 收集反馈 → 调整计划
|
||
↓
|
||
[执行批次2] → 收集反馈 → 调整计划
|
||
↓
|
||
[执行批次3] ...
|
||
↓
|
||
✅ 完成（所有任务满足DoD）
|
||
```
|
||
|
||
---
|
||
|
||
## 何时使用 /runtime.iterate
|
||
|
||
### 必须使用场景
|
||
- ✅ **已完成规划阶段** - 已通过`/runtime.plan`生成任务树
|
||
- ✅ **需要批量执行** - 任务间有依赖，需要分批次
|
||
- ✅ **需要持续反馈** - 想在执行中学习、调整计划
|
||
- ✅ **处理不确定性** - 预期会有失败、变更、新发现
|
||
- ✅ **长期项目** - 需要持续数天/数周的实施
|
||
|
||
### 使用流程
|
||
|
||
```bash
|
||
# 第一步: 生成任务树
|
||
/runtime.plan "实现用户认证系统"
|
||
↓
|
||
生成: cognition/plans/plan-2025xx.json
|
||
|
||
# 第二步: 迭代执行
|
||
/runtime.iterate --plan=plan-2025xx.json
|
||
↓ 自动执行...
|
||
|
||
Iteration 1: 执行基础设施任务（User表、Token表、JWT配置）
|
||
Iteration 2: 执行服务层任务（user.service, token.service）
|
||
Iteration 3: 执行API层任务（Register、Login API）
|
||
Iteration 4: 执行安全层任务（Password Hashing、Auth Middleware）
|
||
Iteration 5: 执行测试任务（单元测试、集成测试）
|
||
|
||
✅ 所有任务完成！
|
||
```
|
||
|
||
---
|
||
|
||
## 迭代循环详解
|
||
|
||
### 阶段1: 初始化（Iteration Setup）
|
||
|
||
#### 1.1 加载任务树
|
||
|
||
```python
|
||
def load_plan(plan_file: str) -> Plan:
|
||
"""
|
||
加载由plan生成的任务树
|
||
"""
|
||
with open(plan_file) as f:
|
||
plan_data = json.load(f)
|
||
|
||
# 验证文件格式
|
||
if "tasks" not in plan_data:
|
||
raise ValueError("无效的计划文件：缺少'tasks'字段")
|
||
|
||
if "critical_path" not in plan_data:
|
||
raise ValueError("无效的计划文件：缺少'critical_path'字段")
|
||
|
||
return Plan(
|
||
tasks=[Task.from_dict(t) for t in plan_data["tasks"]],
|
||
critical_path=plan_data["critical_path"],
|
||
total_effort=plan_data["total_effort"]
|
||
)
|
||
```
|
||
|
||
#### 1.2 初始化迭代器
|
||
|
||
```python
|
||
class IterativeExecutor:
|
||
def __init__(self, plan: Plan, strategy="breadth"):
|
||
self.plan = plan
|
||
self.iteration_count = 0
|
||
self.max_iterations = 20 # 防止无限循环
|
||
self.completed_tasks = []
|
||
self.failed_tasks = []
|
||
self.skipped_tasks = []
|
||
self.strategy = strategy
|
||
|
||
print(f"🚀 迭代执行器已初始化")
|
||
print(f" 总任务数: {len(plan.tasks)}")
|
||
print(f" 预计工时: {plan.total_effort}")
|
||
print(f" 关键路径: {' → '.join(plan.critical_path)}")
|
||
print(f" 策略: {strategy}")
|
||
```
|
||
|
||
---
|
||
|
||
### 阶段2: 迭代循环（Iteration Loop）
|
||
|
||
#### 2.1 主循环逻辑
|
||
|
||
```python
|
||
def run_iteration_loop(self) -> IterationResult:
|
||
"""
|
||
运行迭代执行循环
|
||
"""
|
||
print("\n" + "=" * 50)
|
||
print("开始迭代执行循环")
|
||
print("=" * 50)
|
||
|
||
while self.should_continue():
|
||
self.iteration_count += 1
|
||
print(f"\n📌 Iteration #{self.iteration_count}")
|
||
print(f" 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print("-" * 50)
|
||
|
||
# Step 1: 选择可执行任务
|
||
ready_tasks = self.get_ready_tasks()
|
||
print(f" 可执行任务: {len(ready_tasks)}个")
|
||
|
||
if not ready_tasks:
|
||
if self.get_remaining_tasks():
|
||
print(" ⚠️ 有未完成任务但依赖未满足")
|
||
print(f" 剩余: {len(self.get_remaining_tasks())}个")
|
||
break
|
||
else:
|
||
print(" ✅ 所有任务已完成！")
|
||
break
|
||
|
||
# Step 2: 批量执行
|
||
results = self.execute_batch(ready_tasks)
|
||
|
||
# Step 3: 收集反馈
|
||
feedback = self.collect_feedback(results)
|
||
|
||
# Step 4: 适应与调整
|
||
self.plan = self.adapt_plan(self.plan, feedback)
|
||
|
||
# Step 5: 检查完成状态
|
||
if self.is_all_completed():
|
||
print("\n" + "=" * 50)
|
||
print("✅ 所有任务完成！迭代结束")
|
||
print("=" * 50)
|
||
break
|
||
|
||
# Step 6: 休息与反思
|
||
if self.should_reflect():
|
||
self.reflect()
|
||
|
||
return self.generate_result()
|
||
```
|
||
|
||
#### 2.2 选择可执行任务
|
||
|
||
```python
|
||
def get_ready_tasks(self) -> List[Task]:
|
||
"""
|
||
选择满足以下条件的任务:
|
||
1. 未完成
|
||
2. 所有依赖已完成
|
||
3. 未被阻塞
|
||
4. 在当前策略优先级中
|
||
"""
|
||
ready = []
|
||
|
||
for task in self.plan.tasks:
|
||
# 已完成，跳过
|
||
if task.status == "completed":
|
||
continue
|
||
|
||
# 被阻塞，跳过
|
||
if task.status == "blocked":
|
||
continue
|
||
|
||
# 有失败依赖，阻塞
|
||
if any(dep in [t.id for t in self.failed_tasks] for dep in task.dependencies):
|
||
task.status = "blocked"
|
||
task.block_reason = f"依赖任务失败: {[t.name for t in self.failed_tasks]}"
|
||
self.skipped_tasks.append(task)
|
||
continue
|
||
|
||
# 依赖未完成，等待
|
||
if not all(dep in [t.id for t in self.completed_tasks] for dep in task.dependencies):
|
||
continue
|
||
|
||
# 满足所有条件，可执行
|
||
ready.append(task)
|
||
|
||
# 按策略排序
|
||
return self.sort_by_strategy(ready)
|
||
```
|
||
|
||
#### 2.3 执行策略
|
||
|
||
```python
|
||
def sort_by_strategy(self, tasks: List[Task]) -> List[Task]:
|
||
"""
|
||
根据执行策略排序任务
|
||
|
||
策略1: breadth-first (广度优先)
|
||
- 执行所有基础任务再执行上层任务
|
||
- 特点: 减少返工风险
|
||
|
||
策略2: depth-first (深度优先)
|
||
- 优先执行关键路径上的任务
|
||
- 特点: 快速验证核心链路
|
||
|
||
策略3: risk-driven (风险驱动)
|
||
- 优先执行高风险任务
|
||
- 特点: 尽早暴露问题
|
||
|
||
策略4: value-driven (价值驱动)
|
||
- 优先交付用户价值最大的任务
|
||
- 特点: 快速交付MVP
|
||
"""
|
||
|
||
if self.strategy == "breadth":
|
||
# 按层级排序（基础设施 → 服务 → API → 安全 → 测试）
|
||
layer_order = {
|
||
"基础设施": 1,
|
||
"服务层": 2,
|
||
"API层": 3,
|
||
"安全层": 4,
|
||
"测试": 5
|
||
}
|
||
return sorted(tasks, key=lambda t: layer_order.get(t.layer, 99))
|
||
|
||
elif self.strategy == "depth":
|
||
# 按是否关键路径排序
|
||
return sorted(tasks, key=lambda t: t.id in self.plan.critical_path, reverse=True)
|
||
|
||
elif self.strategy == "risk":
|
||
# 按风险等级排序
|
||
risk_order = {"high": 1, "medium": 2, "low": 3}
|
||
return sorted(tasks, key=lambda t: risk_order.get(t.risk_level, 3))
|
||
|
||
elif self.strategy == "value":
|
||
# 按价值排序（需要手动标注或从需求提取）
|
||
return sorted(tasks, key=lambda t: t.priority, reverse=True)
|
||
|
||
else:
|
||
return tasks
|
||
```
|
||
|
||
---
|
||
|
||
### 阶段3: 执行批次（Batch Execution）
|
||
|
||
#### 3.1 批量执行
|
||
|
||
```python
|
||
def execute_batch(self, tasks: List[Task]) -> List[ExecutionResult]:
|
||
"""
|
||
批量执行一批任务
|
||
"""
|
||
print(f"\n🚀 执行批次: {len(tasks)}个任务")
|
||
print("-" * 50)
|
||
|
||
results = []
|
||
|
||
for i, task in enumerate(tasks, 1):
|
||
print(f"\n [{i}/{len(tasks)}] {task.id}: {task.name}")
|
||
print(f" 预计工时: {task.effort}h")
|
||
print(f" 置信度: {task.confidence:.2f}")
|
||
|
||
try:
|
||
# 预检查
|
||
if not self.pre_check(task):
|
||
print(f" ⚠️ 预检查失败，跳过")
|
||
result = ExecutionResult(
|
||
task=task,
|
||
status="skipped",
|
||
reason="预检查失败"
|
||
)
|
||
results.append(result)
|
||
self.skipped_tasks.append(task)
|
||
continue
|
||
|
||
# 执行
|
||
print(f" ⏳ 执行中...")
|
||
execution_result = task.execute()
|
||
|
||
# 验证（Definition of Done）
|
||
validation = self.validate_task(task, execution_result)
|
||
|
||
if validation.passed:
|
||
print(f" ✅ 完成！耗时: {execution_result.duration:.1f}h")
|
||
result = ExecutionResult(
|
||
task=task,
|
||
status="success",
|
||
result=execution_result,
|
||
validation=validation
|
||
)
|
||
self.completed_tasks.append(task)
|
||
else:
|
||
print(f" ❌ 验证失败:")
|
||
for error in validation.errors:
|
||
print(f" - {error}")
|
||
|
||
result = ExecutionResult(
|
||
task=task,
|
||
status="failed",
|
||
result=execution_result,
|
||
validation=validation
|
||
)
|
||
self.failed_tasks.append(task)
|
||
|
||
results.append(result)
|
||
|
||
# 学习这次执行
|
||
self.learn_from_execution(result)
|
||
|
||
except Exception as e:
|
||
print(f" 💥 执行错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
result = ExecutionResult(
|
||
task=task,
|
||
status="error",
|
||
error=e
|
||
)
|
||
results.append(result)
|
||
self.failed_tasks.append(task)
|
||
|
||
print("\n" + "-" * 50)
|
||
print(f"批次完成: {len(tasks)}个任务")
|
||
print(f" ✅ 成功: {len([r for r in results if r.status == 'success'])}")
|
||
print(f" ❌ 失败: {len([r for r in results if r.status == 'failed'])}")
|
||
print(f" ⚠️ 跳过: {len([r for r in results if r.status == 'skipped'])}")
|
||
print(f" 💥 错误: {len([r for r in results if r.status == 'error'])}")
|
||
|
||
return results
|
||
```
|
||
|
||
#### 3.2 预检查（Pre-Check）
|
||
|
||
```python
|
||
def pre_check(self, task: Task) -> bool:
|
||
"""
|
||
执行前检查
|
||
"""
|
||
print(" 预检查:")
|
||
|
||
# 检查1: 依赖是否完成
|
||
for dep_id in task.dependencies:
|
||
dep = self.plan.get_task(dep_id)
|
||
if not dep or dep.status != "completed":
|
||
print(f" ⚠️ 依赖未完成: {dep_id}")
|
||
return False
|
||
print(f" ✅ 所有依赖已完成")
|
||
|
||
# 检查2: 必需资源是否可用
|
||
if task.required_resources:
|
||
for resource in task.required_resources:
|
||
if not self.check_resource_available(resource):
|
||
print(f" ⚠️ 资源不可用: {resource}")
|
||
return False
|
||
print(f" ✅ 资源可用")
|
||
|
||
# 检查3: 是否有已知风险
|
||
if task.risk_level == "high" and task.confidence < 0.6:
|
||
# 高风险且信心不足，建议先Spike
|
||
if not self.ask_confirmation("高风险任务，确认执行？"):
|
||
print(f" ⚠️ 用户取消执行（建议先调研）")
|
||
return False
|
||
print(f" ⚠️ 高风险但用户确认执行")
|
||
|
||
print(f" ✅ 预检查通过")
|
||
return True
|
||
```
|
||
|
||
#### 3.3 任务执行（伪代码）
|
||
|
||
```python
|
||
class Task:
|
||
def execute(self) -> ExecutionResult:
|
||
"""
|
||
执行单个任务
|
||
"""
|
||
start_time = datetime.now()
|
||
|
||
# 不同类型的任务，使用不同工具
|
||
if self.type == "database":
|
||
# 数据库任务: 执行SQL
|
||
result = bash(f"psql -f {self.sql_file}")
|
||
|
||
elif self.type == "api":
|
||
# API任务: 创建Controller + Route
|
||
# 1. 读取模板
|
||
template = read_file("templates/api-controller.template")
|
||
|
||
# 2. 填充模板
|
||
code = template.format(
|
||
controller_name=self.controller_name,
|
||
functions=self.functions
|
||
)
|
||
|
||
# 3. 写入文件
|
||
write_file(self.output_path, code)
|
||
|
||
# 4. 运行测试
|
||
bash(f"npm test {self.test_file}")
|
||
|
||
elif self.type == "service":
|
||
# 服务任务: 实现业务逻辑
|
||
# ...
|
||
|
||
duration = datetime.now() - start_time
|
||
|
||
return ExecutionResult(
|
||
task_id=self.id,
|
||
status="success",
|
||
duration=duration.total_seconds() / 3600,
|
||
artifacts=[self.output_path]
|
||
)
|
||
```
|
||
|
||
---
|
||
|
||
### 阶段4: 验证（Validation）
|
||
|
||
#### 4.1 Definition of Done
|
||
|
||
```python
|
||
def validate_task(self, task: Task, execution_result: ExecutionResult) -> ValidationResult:
|
||
"""
|
||
验证任务是否真正完成
|
||
"""
|
||
passed_checks = []
|
||
failed_checks = []
|
||
|
||
# 检查1: 代码存在且可访问
|
||
if task.output_path:
|
||
if Path(task.output_path).exists():
|
||
passed_checks.append("代码文件存在")
|
||
else:
|
||
failed_checks.append("代码文件不存在")
|
||
|
||
# 检查2: 单元测试通过
|
||
if task.requires_unit_tests:
|
||
test_result = bash(f"npm test {task.test_path}")
|
||
if test_result.exit_code == 0:
|
||
coverage = extract_coverage(test_result.output)
|
||
if coverage >= 0.8:
|
||
passed_checks.append(f"单元测试覆盖({coverage:.0%})")
|
||
else:
|
||
failed_checks.append(f"覆盖率不足: {coverage:.0%}")
|
||
else:
|
||
failed_checks.append("单元测试失败")
|
||
|
||
# 检查3: 手动测试通过
|
||
if task.acceptance_criteria:
|
||
print(" 验收标准检查:")
|
||
for criteria in task.acceptance_criteria:
|
||
if self.check_criteria(criteria):
|
||
print(f" ✅ {criteria}")
|
||
passed_checks.append(criteria)
|
||
else:
|
||
print(f" ❌ {criteria}")
|
||
failed_checks.append(criteria)
|
||
|
||
# 检查4: 无回归错误（如果配置了集成测试）
|
||
if task.requires_integration_test:
|
||
# 运行集成测试
|
||
pass
|
||
|
||
# 总结
|
||
all_passed = len(failed_checks) == 0
|
||
|
||
if all_passed:
|
||
print(" ✅ 所有验收标准通过")
|
||
else:
|
||
print(f" ❌ 未通过 {len(failed_checks)}项检查")
|
||
|
||
return ValidationResult(
|
||
passed=len(passed_checks),
|
||
failed=len(failed_checks),
|
||
all_passed=all_passed,
|
||
errors=failed_checks
|
||
)
|
||
```
|
||
|
||
#### 4.2 渐进式DoD（根据优先级）
|
||
|
||
```python
|
||
def get_definition_of_done(self, task: Task) -> List[str]:
|
||
"""
|
||
根据任务优先级返回DoD检查清单
|
||
"""
|
||
if task.priority == "P0":
|
||
# 关键任务: 必须全部满足
|
||
return [
|
||
"✓ 代码实现完成",
|
||
"✓ 单元测试覆盖率>80%",
|
||
"✓ 手动测试通过",
|
||
"✓ API文档更新",
|
||
"✓ CI/CD通过",
|
||
"✓ 代码审查通过"
|
||
]
|
||
|
||
elif task.priority == "P1":
|
||
# 重要任务: 可以稍微放宽
|
||
return [
|
||
"✓ 代码实现完成",
|
||
"✓ 单元测试覆盖率>70%",
|
||
"✓ 手动测试通过",
|
||
"✓ CI/CD通过"
|
||
# 文档可以后续补充
|
||
]
|
||
|
||
else:
|
||
# P2任务: 最小要求
|
||
return [
|
||
"✓ 代码实现完成",
|
||
"✓ 基本测试通过",
|
||
"✓ 无严重bug"
|
||
]
|
||
```
|
||
|
||
---
|
||
|
||
### 阶段5: 反馈循环与适应
|
||
|
||
#### 5.1 收集反馈
|
||
|
||
```python
|
||
def collect_feedback(self, results: List[ExecutionResult]) -> Feedback:
|
||
"""
|
||
从执行结果收集反馈
|
||
"""
|
||
feedback = Feedback(
|
||
iteration=self.iteration_count,
|
||
timestamp=datetime.now(),
|
||
results=results,
|
||
metrics={
|
||
"success_rate": len([r for r in results if r.status == "success"]) / len(results),
|
||
"avg_duration": sum(r.duration for r in results if r.duration) / len(results),
|
||
"failed_count": len([r for r in results if r.status == "failed"]),
|
||
"new_discoveries": []
|
||
}
|
||
)
|
||
|
||
# 检查是否有新发现
|
||
for result in results:
|
||
if result.discovery:
|
||
feedback.metrics["new_discoveries"].append(result.discovery)
|
||
print(f"\n✨ 新发现: {result.discovery}")
|
||
|
||
return feedback
|
||
```
|
||
|
||
#### 5.2 动态适应
|
||
|
||
```python
|
||
def adapt_plan(self, plan: Plan, feedback: Feedback) -> Plan:
|
||
"""
|
||
根据反馈动态调整计划
|
||
|
||
三种适应模式:
|
||
1. 失败处理: 重试/分解/重新设计
|
||
2. 新发现: 添加新任务
|
||
3. 性能调整: 调整后续任务估算
|
||
"""
|
||
print("\n🔄 适应调整:")
|
||
print("-" * 50)
|
||
|
||
adjusted = False
|
||
|
||
# 模式1: 处理失败
|
||
for result in feedback.results:
|
||
if result.status == "failed":
|
||
task = result.task
|
||
|
||
# 分析失败原因
|
||
failure_reason = self.analyze_failure(result)
|
||
print(f"\n 分析失败原因 ({task.id}):")
|
||
print(f" → {failure_reason}")
|
||
|
||
# 三种处理方式
|
||
|
||
# 方式1A: 临时错误 → 重试
|
||
if self.is_transient_error(failure_reason):
|
||
print(" → 临时错误，重试任务")
|
||
task.retries += 1
|
||
if task.retries < 3:
|
||
# 暂时不改计划，下次迭代重试
|
||
adjusted = True
|
||
else:
|
||
print(" → 重试3次仍失败，升级为错误")
|
||
task.status = "error"
|
||
|
||
# 方式1B: 任务过大 → 分解
|
||
elif self.is_too_complex(failure_reason):
|
||
print(" → 任务复杂度过高，分解为子任务")
|
||
sub_tasks = self.decompose_task(task)
|
||
self.plan.replace(task, sub_tasks)
|
||
adjusted = True
|
||
|
||
# 方式1C: 设计问题 → 重新设计
|
||
elif self.is_design_issue(failure_reason):
|
||
print(" → 设计问题，需要重新设计")
|
||
# 启动短暂学习
|
||
learn_result = self.learn_from_failure(task, failure_reason)
|
||
new_design = self.redesign(task, learn_result)
|
||
self.plan.replace(task, new_design)
|
||
adjusted = True
|
||
|
||
# 方式1D: 需求理解错误 → 请求澄清
|
||
else:
|
||
print(" → 需求理解可能有问题，请求用户澄清")
|
||
self.request_user_clarification(task, failure_reason)
|
||
adjusted = True
|
||
|
||
# 模式2: 处理新发现
|
||
if feedback.metrics["new_discoveries"]:
|
||
print(f"\n ✨ 发现 {len(feedback.metrics['new_discoveries'])} 个新信息")
|
||
|
||
for discovery in feedback.metrics["new_discoveries"]:
|
||
print(f" → {discovery}")
|
||
|
||
# 基于新信息生成后续任务
|
||
new_tasks = self.create_follow_up_tasks(discovery)
|
||
if new_tasks:
|
||
print(f" → 新增 {len(new_tasks)} 个任务")
|
||
self.plan.add_tasks(new_tasks)
|
||
adjusted = True
|
||
|
||
# 模式3: 性能调整（如果实际耗时与预计差异大）
|
||
if feedback.metrics["avg_duration"]:
|
||
avg_actual = feedback.metrics["avg_duration"]
|
||
avg_estimated = sum(t.effort for t in self.completed_tasks) / len(self.completed_tasks)
|
||
ratio = avg_actual / avg_estimated
|
||
|
||
if ratio > 1.5:
|
||
print(f"\n ⚠️ 实际耗时比预计高{ratio:.1f}倍")
|
||
print(" → 调整后续任务估算")
|
||
for task in self.get_remaining_tasks():
|
||
task.effort *= ratio
|
||
adjusted = True
|
||
|
||
if not adjusted:
|
||
print(" 无需调整，继续执行")
|
||
|
||
return plan
|
||
```
|
||
|
||
#### 5.3 失败分析
|
||
|
||
```python
|
||
def analyze_failure(self, result: ExecutionResult) -> FailureAnalysis:
|
||
"""
|
||
分析失败原因
|
||
|
||
失败类型:
|
||
- TYPE_UNKNOWN: 未知错误
|
||
- TYPE_TRANSIENT: 临时错误（重试可解决）
|
||
- TYPE_COMPLEXITY: 任务太复杂（需要分解）
|
||
- TYPE_DESIGN: 设计问题（需要重新设计）
|
||
- TYPE_REQUIREMENT: 需求不清（需要澄清）
|
||
- TYPE_RESOURCE: 资源不足（需要配置）
|
||
"""
|
||
if result.error:
|
||
error_msg = str(result.error).lower()
|
||
|
||
# 临时错误
|
||
if any(word in error_msg for word in [
|
||
"timeout", "connection", "network",
|
||
"EBUSY", "EAGAIN"
|
||
]):
|
||
return Failure.TYPE_TRANSIENT
|
||
|
||
# 设计问题
|
||
if any(word in error_msg for word in [
|
||
"circular dependency", "deadlock",
|
||
"stack overflow"
|
||
]):
|
||
return Failure.TYPE_DESIGN
|
||
|
||
# 资源问题
|
||
if any(word in error_msg for word in [
|
||
"out of memory", "disk full",
|
||
"quota exceeded"
|
||
]):
|
||
return Failure.TYPE_RESOURCE
|
||
|
||
# 验证错误（检查清单未通过）
|
||
if result.validation and not result.validation.all_passed:
|
||
if len(result.validation.errors) > 5:
|
||
# 错误太多，可能是需求理解问题
|
||
return Failure.TYPE_REQUIREMENT
|
||
else:
|
||
# 具体检查项失败，可能是设计问题
|
||
return Failure.TYPE_DESIGN
|
||
|
||
return Failure.TYPE_UNKNOWN
|
||
```
|
||
|
||
---
|
||
|
||
### 阶段6: 终止条件
|
||
|
||
#### 6.1 终止条件判断
|
||
|
||
```python
|
||
def should_continue(self) -> bool:
|
||
"""
|
||
判断是否继续迭代
|
||
"""
|
||
# 条件1: 达到最大迭代次数
|
||
if self.iteration_count >= self.max_iterations:
|
||
print("\n⚠️ 达到最大迭代次数，停止执行")
|
||
print(f" 已完成: {len(self.completed_tasks)}个任务")
|
||
print(f" 未完成: {len(self.get_remaining_tasks())}个任务")
|
||
return False
|
||
|
||
# 条件2: 所有任务完成
|
||
if self.is_all_completed():
|
||
print("\n✅ 所有任务完成！")
|
||
return False
|
||
|
||
# 条件3: 连续3次无进展（无法解决的阻塞）
|
||
recent_iterations = self.get_recent_iterations(3)
|
||
if all(len(r.completed_tasks) == 0 for r in recent_iterations):
|
||
print("\n⚠️ 连续3次无进展，存在无法解决的任务")
|
||
print(" 建议方案:")
|
||
print(" 1. 手动介入未完成任務")
|
||
print(" 2. 重新规划剩余任务")
|
||
print(" 3. 调整需求范围")
|
||
return False
|
||
|
||
# 条件4: 用户手动停止
|
||
if self.should_stop_requested:
|
||
print("\n⏹️ 用户手动停止执行")
|
||
return False
|
||
|
||
return True
|
||
```
|
||
|
||
#### 6.2 完成状态判断
|
||
|
||
```python
|
||
def is_all_completed(self) -> bool:
|
||
"""
|
||
检查所有任务是否已完成
|
||
"""
|
||
remaining = self.get_remaining_tasks()
|
||
|
||
if not remaining:
|
||
return True
|
||
|
||
# 检查是否有任务被永久阻塞
|
||
truly_blocked = [
|
||
task for task in remaining
|
||
if task.status == "blocked"
|
||
]
|
||
|
||
if truly_blocked:
|
||
print(f"\n⚠️ {len(truly_blocked)}个任务永久阻塞，无法完成")
|
||
return False
|
||
|
||
return False
|
||
```
|
||
|
||
---
|
||
|
||
### 阶段7: 生成结果报告
|
||
|
||
#### 7.1 执行结果统计
|
||
|
||
```python
|
||
def generate_result(self) -> IterationResult:
|
||
"""
|
||
生成执行结果报告
|
||
"""
|
||
result = IterationResult(
|
||
plan_file=self.plan.file_path,
|
||
total_iterations=self.iteration_count,
|
||
started_at=self.start_time,
|
||
ended_at=datetime.now(),
|
||
completed_tasks=self.completed_tasks,
|
||
failed_tasks=self.failed_tasks,
|
||
skipped_tasks=self.skipped_tasks,
|
||
metrics=self.calculate_metrics()
|
||
)
|
||
|
||
print("\n" + "=" * 70)
|
||
print("📊 执行结果统计")
|
||
print("=" * 70)
|
||
print(f"总迭代次数: {self.iteration_count}")
|
||
print(f"总任务数: {len(self.plan.tasks)}")
|
||
print(f"✅ 已完成: {len(self.completed_tasks)}")
|
||
print(f"❌ 失败: {len(self.failed_tasks)}")
|
||
print(f"⚠️ 跳过: {len(self.skipped_tasks)}")
|
||
print(f"⏱️ 总耗时: {result.total_duration:.1f}小时")
|
||
print(f"完成率: {result.completion_rate:.1%}")
|
||
|
||
if self.failed_tasks:
|
||
print("\n❌ 失败任务:")
|
||
for task in self.failed_tasks[:5]:
|
||
print(f" - {task.id}: {task.name}")
|
||
|
||
if result.metrics["efficiency"]:
|
||
print(f"\n🎯 效率指标:")
|
||
print(f" 估算准确率: {result.metrics['efficiency']['accuracy']:.1%}")
|
||
print(f" 平均偏差: {result.metrics['efficiency']['bias']:.1f}x")
|
||
|
||
return result
|
||
```
|
||
|
||
#### 7.2 保存到记忆
|
||
|
||
```python
|
||
def save_to_memory(self, result: IterationResult):
|
||
"""
|
||
将执行结果保存到长期记忆
|
||
"""
|
||
# 固化成功经验
|
||
if result.completion_rate == 1.0:
|
||
memory_content = f"""
|
||
## 项目成功交付 - {self.plan.project_name}
|
||
**时间**: {datetime.now().strftime('%Y-%m-%d')}
|
||
**项目**: {self.plan.project_name}
|
||
**总任务**: {len(self.plan.tasks)}
|
||
**总工时**: {result.total_duration:.1f}h
|
||
**完成率**: 100%
|
||
|
||
**关键经验**:
|
||
{"\n".join(["- " + m for m in result.metrics["learnings"]])}
|
||
|
||
**可用于未来项目**: 是
|
||
"""
|
||
self.remember(memory_content, category="project-success")
|
||
|
||
# 固化失败教训
|
||
if self.failed_tasks:
|
||
for task in self.failed_tasks:
|
||
failure_memory = f"""
|
||
## 失败任务 - {task.id}
|
||
**任务**: {task.name}
|
||
**原因**: {getattr(task, 'failure_reason', '未知')}
|
||
**教训**: {task.lesson_learned}
|
||
|
||
**防御机制**: 下次遇到类似任务，先{task.defense_action}
|
||
"""
|
||
self.remember(failure_memory, category="failure-pattern")
|
||
```
|
||
|
||
---
|
||
|
||
## 执行策略详解
|
||
|
||
### 策略1: Breadth-First (广度优先)
|
||
|
||
```
|
||
特点:
|
||
- 先完成所有基础任务
|
||
- 再执行上层任务
|
||
- 最后执行测试和优化
|
||
|
||
执行顺序:
|
||
批1: User表、Token表、JWT配置 [所有基础设施]
|
||
批2: user.service、token.service [所有服务]
|
||
批3: Register API、Login API [所有API]
|
||
批4: Password Hashing、Auth中间件 [所有安全]
|
||
批5: 单元测试、集成测试 [所有测试]
|
||
|
||
优势:
|
||
✓ 减少返工风险（基础不稳定不会上层浪费）
|
||
✓ 可以并行开发（每批任务并行）
|
||
✓ 适合团队协作（分层对接）
|
||
|
||
劣势:
|
||
⚠ 价值交付慢（用户要等所有层完成才能用）
|
||
```
|
||
|
||
### 策略2: Depth-First (深度优先)
|
||
|
||
```
|
||
特点:
|
||
- 优先完成关键路径上的任务
|
||
- 尽快验证核心链路
|
||
- 次要路径推迟
|
||
|
||
执行顺序:
|
||
批1: User表 → Token表 → Token服务 → Login API → 集成测试 [关键路径]
|
||
批2: Register API → 单元测试 [关键路径]
|
||
批3: Password重置API [非关键]
|
||
批4: 文档、代码质量优化 [非关键]
|
||
|
||
优势:
|
||
✓ 快速验证核心功能（用户可早期体验）
|
||
✓ 风险发现早（核心链路问题尽早暴露）
|
||
✓ 适合快速演示
|
||
|
||
劣势:
|
||
⚠ 需要频繁重构（后续任务可能要求前面修改）
|
||
```
|
||
|
||
### 策略3: Risk-Driven (风险驱动)
|
||
|
||
```
|
||
特点:
|
||
- 优先执行高风险任务
|
||
- 尽早暴露潜在问题
|
||
- 低风险任务推迟
|
||
|
||
执行顺序:
|
||
高风险: 密码重置流程（涉及多步骤安全）→ 先做Spike
|
||
高风险: Token刷新机制（不确定设计） → 查阅资料+原型
|
||
中风险: 认证中间件（有现成模式） → 稍后实现
|
||
低风险: 单元测试（成熟技术） → 最后补充
|
||
|
||
优势:
|
||
✓ 不确定性尽快消除（避免大坑）
|
||
✓ 适合技术选型期
|
||
✓ 风险前置管理
|
||
|
||
劣势:
|
||
⚠ 可能执行顺序不合理（不是最优价值）
|
||
```
|
||
|
||
### 策略4: Value-Driven (价值驱动)
|
||
|
||
```
|
||
特点:
|
||
- 优先交付用户价值最大的功能
|
||
- MVP -> 迭代增强
|
||
- 砍掉非核心价值
|
||
|
||
执行顺序:
|
||
MVP: User表 + Register API + Login API [可演示的核心]
|
||
迭代1: Password重置 [增强可用性]
|
||
迭代2: Rate limiting [提升质量]
|
||
迭代3: 完整测试覆盖 [质量保证]
|
||
|
||
优势:
|
||
✓ 用户价值最大化（可以早期使用）
|
||
✓ 反馈收集早（用户参与迭代）
|
||
✓ 适合创业/探索期
|
||
|
||
劣势:
|
||
⚠ 技术债务积累（前期快速可能牺牲质量）
|
||
```
|
||
|
||
---
|
||
|
||
## 命令定义
|
||
|
||
### 语法
|
||
|
||
```bash
|
||
/runtime.iterate \
|
||
--plan-file=plan-xxxx.json \
|
||
--strategy=[breadth|depth|risk|value] \
|
||
--max-iterations=20 \
|
||
--parallel=[1|2|3|...] \
|
||
--auto-adapt=true \
|
||
--reflect-interval=5
|
||
```
|
||
|
||
### 参数说明
|
||
|
||
| 参数 | 说明 | 默认值 |
|
||
|------|------|--------|
|
||
| `plan-file` | 计划文件路径（来自/runtime.plan） | 必需 |
|
||
| `strategy` | 执行策略 | `breadth` |
|
||
| `max-iterations` | 最大迭代次数（防止无限循环） | `20` |
|
||
| `parallel` | 每批并行任务数 | `3` |
|
||
| `auto-adapt` | 失败时自动适应 | `true` |
|
||
| `reflect-interval` | 每N次迭代后强制反思 | `5` |
|
||
|
||
### 使用示例
|
||
|
||
#### 示例1: 基础使用
|
||
|
||
```bash
|
||
# 第一步：规划
|
||
/runtime.plan "实现用户认证系统"
|
||
↓
|
||
生成: cognition/plans/plan-xxx.json
|
||
|
||
# 第二步：迭代执行
|
||
/runtime.iterate --plan-file=cognition/plans/plan-xxx.json
|
||
```
|
||
|
||
**输出**:
|
||
```
|
||
🚀 迭代执行器已初始化
|
||
总任务数: 17
|
||
预计工时: 17.75h
|
||
关键路径: ① → ② → ⑤ → ⑧ → ⑮ → ⑯
|
||
策略: breadth
|
||
|
||
════════════════════════════════════════════════════════════
|
||
开始迭代执行循环
|
||
════════════════════════════════════════════════════════════
|
||
|
||
📌 Iteration #1
|
||
时间: 2025-11-14 11:00:00
|
||
--------------------------------------------------
|
||
可执行任务: 3个
|
||
|
||
[1/3] ①: 创建User表
|
||
预计工时: 0.5h
|
||
置信度: 0.90
|
||
预检查:
|
||
✅ 所有依赖已完成
|
||
✅ 资源可用
|
||
⏳ 执行中...
|
||
Command: psql -f migrations/001-create-user.sql
|
||
Result: CREATE TABLE 成功
|
||
✅ 完成！耗时: 0.4h
|
||
|
||
[2/3] ②: 创建Token表
|
||
预计工时: 0.5h
|
||
置信度: 0.85
|
||
依赖: [①]
|
||
预检查:
|
||
✅ 所有依赖已完成
|
||
✅ 资源可用
|
||
⏳ 执行中...
|
||
Command: psql -f migrations/002-create-token.sql
|
||
Result: CREATE TABLE 成功
|
||
✅ 完成！耗时: 0.3h
|
||
|
||
[3/3] ③: 配置JWT
|
||
预计工时: 0.25h
|
||
置信度: 0.95
|
||
预检查:
|
||
✅ 所有依赖已完成
|
||
⏳ 执行中...
|
||
Command: node scripts/generate-jwt-keys.js
|
||
Result: 密钥生成成功
|
||
✅ 完成！耗时: 0.2h
|
||
|
||
--------------------------------------------------
|
||
批次完成: 3个任务
|
||
✅ 成功: 3
|
||
❌ 失败: 0
|
||
⚠️ 跳过: 0
|
||
💥 错误: 0
|
||
|
||
📌 Iteration #2
|
||
时间: 2025-11-14 11:30:00
|
||
--------------------------------------------------
|
||
可执行任务: 3个
|
||
|
||
[1/3] ④: user.service.js
|
||
...
|
||
|
||
════════════════════════════════════════════════════════════
|
||
📊 执行结果统计
|
||
════════════════════════════════════════════════════════════
|
||
总迭代次数: 6
|
||
总任务数: 17
|
||
✅ 已完成: 17
|
||
❌ 失败: 0
|
||
⏱️ 总耗时: 16.2小时
|
||
完成率: 100%
|
||
|
||
✅ 所有验收标准满足
|
||
✅ API文档已更新
|
||
✅ Changelog已更新
|
||
✅ CI/CD通过
|
||
|
||
项目完成！
|
||
```
|
||
|
||
#### 示例2: 使用不同策略
|
||
|
||
```bash
|
||
# 深度优先（快速验证核心）
|
||
/runtime.iterate --plan-file=plan-xxx.json --strategy=depth
|
||
|
||
# 风险驱动（优先高风险任务）
|
||
/runtime.iterate --plan-file=plan-xxx.json --strategy=risk
|
||
|
||
# 价值驱动（MVP模式）
|
||
/runtime.iterate --plan-file=plan-xxx.json --strategy=value
|
||
```
|
||
|
||
#### 示例3: 并行执行
|
||
|
||
```bash
|
||
# 一次并行执行5个任务（适合多核CPU）
|
||
/runtime.iterate --plan-file=plan-xxx.json --parallel=5
|
||
```
|
||
|
||
#### 示例4: 失败处理
|
||
|
||
```bash
|
||
# 配置失败处理策略
|
||
/runtime.iterate \
|
||
--plan-file=plan-xxx.json \
|
||
--on-failure= [retry|decompose|learn|stop]
|
||
|
||
# retry: 重试3次
|
||
# decompose: 分解为子任务
|
||
# learn: 启动学习循环
|
||
# stop: 停止等待人工
|
||
```
|
||
|
||
---
|
||
|
||
## 与 /runtime.plan 的区别
|
||
|
||
| 维度 | `/runtime.plan` | `/runtime.iterate` | 关系 |
|
||
|------|----------------|-------------------|------|
|
||
| **输入** | 需求文本 | 任务树（JSON） | plan的输出 → iterate的输入 |
|
||
| **核心** | 拆解任务 | 执行 + 反馈 | 阶段2（实施） |
|
||
| **输出** | 任务树（静态） | 执行报告（动态） | 后续:
|
||
| **函数** | 生成计划 | 执行计划 | 先后关系 |
|
||
| **循环** | 无（一次生成） | 有（多次迭代） | 迭代: iterate |
|
||
|
||
**工作流**:
|
||
```
|
||
/runtime.plan "实现功能X" → 生成任务树
|
||
↓
|
||
/runtime.iterate --plan=... → 执行任务树
|
||
↓
|
||
/runtime.reflect → 回顾整个过程
|
||
```
|
||
|
||
---
|
||
|
||
## 与 /runtime.learn 的区别
|
||
|
||
| 维度 | `/runtime.learn` | `/runtime.iterate` | 为什么分开？ |
|
||
|------|----------------|-------------------|------------|
|
||
| **范围** | 学习 + 规划 + 执行 | 仅执行 | 职责单一 |
|
||
| **自治度** | 完全自主（从问题到方案） | 半自主（需plan提供任务树） | 区分认知层次 |
|
||
| **输入** | 问题/需求 | 结构化任务 | 抽象层次不同 |
|
||
| **复杂度** | 高（需要智能决策） | 中（主要是执行控制） | 便于调试优化 |
|
||
| **典型场景** | 探索未知问题 | 执行已知计划 | 解耦关注点 |
|
||
|
||
**类比**:
|
||
- **Learn** = 资深架构师（知道如何学习、规划、实施）
|
||
- **Plan** = 项目经理（知道如何拆解任务）
|
||
- **Iterate** = 技术主管（知道如何带领团队执行）
|
||
|
||
---
|
||
|
||
## 工具与脚本
|
||
|
||
### 辅助脚本: task-executor.py
|
||
|
||
```python
|
||
#!/usr/bin/env python3
|
||
"""
|
||
任务执行器 - 执行单个任务并验证
|
||
"""
|
||
|
||
import subprocess
|
||
import time
|
||
from pathlib import Path
|
||
|
||
class TaskExecutor:
|
||
def __init__(self, workspace="."):
|
||
self.workspace = Path(workspace)
|
||
|
||
def execute(self, task):
|
||
"""
|
||
执行任务
|
||
|
||
Returns:
|
||
{
|
||
"status": "success|failed|error",
|
||
"duration": seconds,
|
||
"output": str,
|
||
"error": str (if failed),
|
||
"artifacts": [files created/modified]
|
||
}
|
||
"""
|
||
start_time = time.time()
|
||
result = {
|
||
"status": "unknown",
|
||
"duration": 0,
|
||
"output": "",
|
||
"error": None,
|
||
"artifacts": []
|
||
}
|
||
|
||
try:
|
||
# 根据任务类型选择执行方式
|
||
if task["type"] == "database":
|
||
exec_result = self._execute_sql(task["sql_file"])
|
||
|
||
elif task["type"] == "file_create":
|
||
exec_result = self._create_file(
|
||
task["file_path"],
|
||
task["content"]
|
||
)
|
||
|
||
elif task["type"] == "command":
|
||
exec_result = self._run_command(task["command"])
|
||
|
||
elif task["type"] == "test":
|
||
exec_result = self._run_tests(task["test_files"])
|
||
|
||
else:
|
||
exec_result = {
|
||
"status": "error",
|
||
"error": f"未知任务类型: {task['type']}"
|
||
}
|
||
|
||
# 记录执行结果
|
||
result.update(exec_result)
|
||
result["duration"] = time.time() - start_time
|
||
|
||
except Exception as e:
|
||
result["status"] = "error"
|
||
result["error"] = str(e)
|
||
|
||
return result
|
||
|
||
def _execute_sql(self, sql_file):
|
||
"""执行SQL文件"""
|
||
cmd = f"psql -f {sql_file}"
|
||
return self._run_command(cmd)
|
||
|
||
def _create_file(self, file_path, content):
|
||
"""创建文件"""
|
||
path = Path(file_path)
|
||
path.write_text(content)
|
||
return {
|
||
"status": "success",
|
||
"artifacts": [str(path)]
|
||
}
|
||
|
||
def _run_command(self, command):
|
||
"""运行Shell命令"""
|
||
process = subprocess.run(
|
||
command,
|
||
shell=True,
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
|
||
if process.returncode == 0:
|
||
return {
|
||
"status": "success",
|
||
"output": process.stdout
|
||
}
|
||
else:
|
||
return {
|
||
"status": "failed",
|
||
"output": process.stdout,
|
||
"error": process.stderr
|
||
}
|
||
|
||
def _run_tests(self, test_files):
|
||
"""运行测试"""
|
||
cmd = f"npm test {' '.join(test_files)}"
|
||
return self._run_command(cmd)
|
||
|
||
|
||
# 使用示例
|
||
if __name__ == "__main__":
|
||
executor = TaskExecutor(workspace=".")
|
||
|
||
# 执行任务: 创建User表
|
||
task = {
|
||
"type": "database",
|
||
"sql_file": "migrations/001-create-user.sql"
|
||
}
|
||
|
||
result = executor.execute(task)
|
||
print(json.dumps(result, indent=2))
|
||
```
|
||
|
||
---
|
||
|
||
## 最佳实践
|
||
|
||
### 实践1: 先规划，再执行
|
||
|
||
```bash
|
||
# 正确流程
|
||
✅ /runtime.plan "需求" → 生成计划
|
||
✅ /runtime.iterate --plan=xxx.json → 执行计划
|
||
✅ /runtime.reflect → 回顾
|
||
|
||
# 错误
|
||
❌ /runtime.iterate # 没有提供计划文件
|
||
```
|
||
|
||
### 实践2: 选择合适的策略
|
||
|
||
```bash
|
||
# 不同类型的项目用不同策略
|
||
✅ 新功能开发（基础重要）: --strategy=breadth
|
||
✅ PoC演示（快速验证）: --strategy=depth
|
||
✅ 技术调研（风险消除）: --strategy=risk
|
||
✅ MVP产品（价值优先）: --strategy=value
|
||
```
|
||
|
||
### 实践3: 定期检查（Reflect）
|
||
|
||
```bash
|
||
# 每5次迭代后强制检查
|
||
/runtime.iterate --plan=xxx.json --reflect-interval=5
|
||
|
||
# 或在执行后手动
|
||
/runtime.reflect
|
||
"""
|
||
上次5次迭代的模式：
|
||
- 估算准确率: 85%
|
||
- 常见失败类型: 依赖配置
|
||
- 发现的新任务: 平均每次迭代1.2个
|
||
|
||
改进建议:
|
||
- 预检查阶段增强配置验证
|
||
- 为配置任务添加专门检查清单
|
||
"""
|
||
```
|
||
|
||
### 实践4: 优雅失败与恢复
|
||
|
||
```python
|
||
# 不是暴力失败
|
||
try:
|
||
result = execute(task)
|
||
except Exception as e:
|
||
# 记录
|
||
log_failure(task, e)
|
||
|
||
# 分析
|
||
failure_type = analyze_failure(e)
|
||
|
||
# 适应（不是panic）
|
||
if failure_type == FAIL_TRANSIENT:
|
||
retry(task)
|
||
elif failure_type == FAIL_COMPLEX:
|
||
decompose(task)
|
||
elif failure_type == FAIL_DESIGN:
|
||
learn_and_redesign(task)
|
||
elif failure_type == FAIL_REQUIREMENT:
|
||
ask_clarification()
|
||
|
||
# 继续（不是所有都停止）
|
||
continue_execution()
|
||
```
|
||
|
||
---
|
||
|
||
## 宪法遵循
|
||
|
||
**遵循原则**:
|
||
- ✓ 2.2 渐进式实施: 持续交付价值
|
||
- ✓ 4.4 规划透明: 执行过程可见
|
||
- ✓ 1.3 谦逊与不确定: 失败时承认并学习
|
||
- ✓ 4.1 从经验学习: 每次迭代都更新认知
|
||
- ✓ 4.3 自我反思: 定期评估执行效果
|
||
|
||
---
|
||
|
||
**命令定义**: `.ai-runtime/commands/runtime.iterate.md`
|
||
**脚本**: `.ai-runtime/scripts/task-executor.py`
|
||
**输出**: `cognition/execution-reports/*.json`
|
||
**版本**: 1.0.0
|