任务分解与编排
任务拆分、调度与验收标准
规则:任何交给 AI 执行的任务,必须先拆分为原子任务(Atomic Task),每个原子任务有明确的输入、输出和验收标准。
[Requirement] 单个原子任务的变更行数不超过 400 行(新增+修改),超过必须继续拆分。
[Requirement] 每个原子任务必须能独立验收,不依赖其他未完成任务的输出。
[Standard] 拆分遵循”自上而下、逐层分解”:从需求 → 模块 → 功能 → 子任务 → 原子任务。
[Default] 原子任务的预计执行时间不超过 4 小时,超过说明粒度过粗。
| 层级 | 范围 | 变更行数 | 执行时间 | 示例 |
|---|---|---|---|---|
| 需求 | 完整功能 | - | 数天 | 用户中心 |
| 模块 | 单个模块 | - | 1-2 天 | 用户注册模块 |
| 功能 | 单个功能点 | ≤ 1000 行 | 4-8 小时 | 注册接口 |
| 子任务 | 功能的一部分 | ≤ 400 行 | 2-4 小时 | 邮箱唯一性校验 |
| 原子任务 | 不可再分 | ≤ 100 行 | ≤ 4 小时 | 编写 EmailAlreadyExists 异常类 |
1. 识别目标:明确最终要交付什么2. 识别依赖:哪些是前置条件,哪些可以并行3. 逐层分解:大任务拆成小任务,直到原子任务4. 定义输入输出:每个原子任务的输入是什么、产出是什么5. 定义验收标准:怎样算这个任务完成了示例:
需求:实现用户注册功能
拆分结果:├─ T1: 定义 UserCreate schema(输入模型)│ ├─ T1.1: 创建 UserCreate Pydantic 模型│ └─ T1.2: 编写字段校验(邮箱格式、密码长度)├─ T2: 实现注册核心逻辑│ ├─ T2.1: 实现邮箱唯一性查询│ ├─ T2.2: 实现密码哈希│ └─ T2.3: 实现用户入库├─ T3: 接入邮件服务│ └─ T3.1: 注册成功后调用 send_welcome_email├─ T4: 编写 API 路由│ └─ T4.1: 创建 POST /users 路由└─ T5: 编写单元测试 ├─ T5.1: 正常注册测试 ├─ T5.2: 重复邮箱测试 └─ T5.3: 非法输入测试反模式:
- 把”实现用户中心”作为一个任务扔给 AI
- 单个任务包含 5 个文件、2000 行代码
- 任务没有明确的验收标准,AI 不知道什么时候算完成
规则:任务调度遵循”依赖优先、风险优先、并行优先”三原则,合理安排任务执行顺序和并行度。
[Requirement] 有依赖关系的任务,必须按依赖顺序执行,不得跳过前置任务。
[Standard] 无依赖的任务应并行执行,提高整体效率。
[Standard] 高风险任务(核心逻辑、安全相关)优先执行,尽早暴露问题。
[Suggestion] 使用 DAG(有向无环图)管理任务依赖关系。
| 原则 | 说明 | 示例 |
|---|---|---|
| 依赖优先 | 前置任务完成后才能执行后续 | 先定义 schema,再写逻辑 |
| 风险优先 | 高风险任务先做 | 先实现核心算法,再写 UI |
| 并行优先 | 无依赖的任务同时进行 | 后端接口和前端页面并行 |
T1 (schema) │ ▼ T2 (核心逻辑) ────┐ │ │ ▼ ▼ T3 (邮件) T4 (路由) │ │ └──────┬───────┘ ▼ T5 (测试)调度顺序:
- T1 先执行(所有任务的前置)
- T2 执行(T3、T4 的前置)
- T3 和 T4 并行执行
- T5 最后执行(依赖 T3、T4)
pending → in_progress → blocked → in_progress → done │ │ └──────── failed ←───────────┘| 状态 | 说明 | 可执行的操作 |
|---|---|---|
| pending | 待执行 | 启动 |
| in_progress | 执行中 | 完成/失败 |
| blocked | 被依赖阻塞 | 等待前置完成 |
| done | 已完成 | 无 |
| failed | 失败 | 重试/跳过/终止 |
反模式:
- 不看依赖关系,所有任务一起扔给 AI
- 低风险的 UI 先做完,高风险的核心逻辑留到最后
- 任务失败后不处理,继续执行后续任务
规则:每个任务必须有明确、可测试、可量化的验收标准(Acceptance Criteria),无标准的任务不得启动。
[Requirement] 验收标准必须使用 GIVEN/WHEN/THEN 格式,确保可测试。
[Requirement] 验收标准必须覆盖:正常场景、边界场景、异常场景三类。
[Standard] 验收标准必须可量化,避免”代码质量好""性能不错”等模糊描述。
[Default] 每个任务至少 3 条验收标准。
GIVEN <前置条件>WHEN <执行操作>THEN <预期结果>| 要求 | 说明 | 好的示例 | 差的示例 |
|---|---|---|---|
| 可测试 | 能通过代码或工具验证 | “返回 409” | “处理重复邮箱” |
| 可量化 | 有明确的数值 | “响应时间 < 200ms” | “响应快” |
| 全覆盖 | 正常/边界/异常 | 3 类场景都有 | 只有正常场景 |
| 无歧义 | 只有一种理解 | “密码长度 8-64 位” | “密码要够长” |
示例:
任务:实现用户注册接口
验收标准:1. GIVEN 用户输入合法邮箱和密码 WHEN 调用 POST /users THEN 返回 201,响应体包含用户 ID 和邮箱
2. GIVEN 邮箱已存在 WHEN 调用 POST /users THEN 返回 409,错误信息为"邮箱已注册"
3. GIVEN 密码长度小于 8 位 WHEN 调用 POST /users THEN 返回 422,错误信息包含"密码长度"
4. GIVEN 注册成功 WHEN 查看数据库 THEN 密码字段为 bcrypt 哈希,非明文
5. GIVEN 注册成功 WHEN 检查邮件服务 THEN 调用了 send_welcome_email 且参数为用户邮箱反模式:
- 验收标准只写”功能正常”
- 没有边界和异常场景的验收标准
- 验收标准是”代码写得好""性能不错”等无法验证的描述
规则:任务之间的依赖关系必须显式声明和管理,避免隐式依赖导致的执行混乱和死锁。
[Requirement] 每个任务必须声明其依赖项(前置任务),无依赖的任务标记为根任务。
[Requirement] 依赖关系必须是有向无环图(DAG),严禁出现循环依赖。
[Standard] 依赖分为”硬依赖”和”软依赖”:硬依赖必须完成才能执行,软依赖可并行但需注意冲突。
[Default] 跨文件修改的任务应声明文件级依赖,避免并发修改同一文件。
| 类型 | 说明 | 处理方式 | 示例 |
|---|---|---|---|
| 硬依赖 | 前置任务的输出是当前任务的输入 | 顺序执行 | 必须先有 schema 才能写逻辑 |
| 软依赖 | 可并行,但可能有冲突 | 并行 + 冲突检测 | 同时改同一文件的不同部分 |
| 隐式依赖 | 未声明但实际存在的依赖 | 必须显式化 | 都依赖同一个公共函数 |
# tasks.yaml
- id: T1
name: 定义 UserCreate schema
depends_on: []
files: [src/schemas/user.py]
- id: T2
name: 实现注册核心逻辑
depends_on: [T1]
files: [src/services/user.py]
- id: T3
name: 编写注册路由
depends_on: [T1]
files: [src/api/users.py]
- id: T4
name: 编写单元测试
depends_on: [T2, T3]
files: [tests/test_register.py]# 伪代码:检测循环依赖
def detect_cycle(tasks):
visited = set()
rec_stack = set()
def dfs(task_id):
visited.add(task_id)
rec_stack.add(task_id)
for dep in tasks[task_id].depends_on:
if dep not in visited:
if dfs(dep):
return True
elif dep in rec_stack:
return True # 发现循环依赖
rec_stack.remove(task_id)
return False
for task_id in tasks:
if task_id not in visited:
if dfs(task_id):
raise ValueError(f"检测到循环依赖: {task_id}")反模式:
- 任务不声明依赖,靠人脑记
- 出现 A 依赖 B、B 依赖 A 的循环依赖
- 两个任务同时修改同一个文件,导致合并冲突
规则:任务执行失败时必须有明确的处理策略,禁止”失败了就跳过”或”失败了继续往下跑”。
[Requirement] 任务失败后,必须立即停止依赖该任务的所有后续任务,不得继续执行。
[Requirement] 任务失败必须记录失败原因、失败步骤、相关日志,便于排查。
[Standard] 失败处理策略分为:重试、人工介入、终止三种,根据失败类型选择。
[Default] 自动重试最多 2 次,2 次仍失败则转人工介入。
| 失败类型 | 说明 | 处理策略 | 示例 |
|---|---|---|---|
| 临时性失败 | 网络超时、API 限流 | 自动重试(指数退避) | 调用邮件服务超时 |
| 逻辑性失败 | 代码错误、测试不通过 | 人工介入修复 | 生成的代码有 bug |
| 上下文失败 | 缺少信息、理解偏差 | 补充上下文后重试 | AI 没读到相关代码 |
| 约束性失败 | 违反安全/规范红线 | 终止任务,人工评估 | AI 生成了硬编码密钥 |
任务失败 │ ▼1. 记录失败信息 - 失败任务 ID - 失败步骤 - 错误信息 - 相关日志/输出 │ ▼2. 判断失败类型 ├─ 临时性 → 自动重试(最多 2 次,指数退避) ├─ 逻辑性 → 人工介入(修复后重试) ├─ 上下文 → 补充上下文后重试 └─ 约束性 → 终止任务,人工评估 │ ▼3. 阻塞后续任务 - 所有依赖此任务的任务标记为 blocked - 通知负责人 │ ▼4. 处理完成后 - 成功 → 解除阻塞,继续执行 - 终止 → 标记整个链路为 failed# 指数退避重试
import time
def retry_with_backoff(task, max_retries=2, base_delay=1):
for attempt in range(max_retries + 1):
try:
return task()
except TemporaryError as e:
if attempt == max_retries:
raise
delay = base_delay * (2 ** attempt)
time.sleep(delay)示例:
任务 T3(接入邮件服务)执行失败:
失败信息:- 任务:T3 注册成功后调用 send_welcome_email- 失败步骤:send_welcome_email 函数不存在- 错误:ModuleNotFoundError: No module named 'src.services.email'
失败类型:上下文失败(AI 不知道 email 服务还没实现)
处理:1. 阻塞 T4(依赖 T3)2. 补充上下文:说明 email 服务需要先实现3. 先执行 T0(实现 email 服务),再重试 T3反模式:
- 任务失败后直接跳过,继续执行后续任务
- 失败后不记录原因,下次同样的问题继续失败
- 无限重试,导致资源耗尽
- 约束性失败(如安全红线)也尝试重试