跳转到内容

任务分解与编排

任务拆分、调度与验收标准


规则:任何交给 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 (测试)

调度顺序:

  1. T1 先执行(所有任务的前置)
  2. T2 执行(T3、T4 的前置)
  3. T3 和 T4 并行执行
  4. 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

反模式:

  • 任务失败后直接跳过,继续执行后续任务
  • 失败后不记录原因,下次同样的问题继续失败
  • 无限重试,导致资源耗尽
  • 约束性失败(如安全红线)也尝试重试