Codex 复杂任务拆解：从"一气呵成"到"步步为营"

核心洞察：Codex 的单次生成能力有限（约 2000-4000 token 有效上下文），复杂任务强行一次性生成会导致"后期失忆"和"逻辑漂移"。

---

一、问题诊断：为什么复杂任务容易出 Bug？

1.1 上下文窗口的"漏斗效应"

用户完整需求（1000 tokens）
    ↓
Codex 理解并规划（占用 500 tokens）
    ↓
生成代码段 1（500 tokens）✅ 质量高
    ↓
生成代码段 2（500 tokens）⚠️ 开始遗忘早期约束
    ↓
生成代码段 3（500 tokens）❌ 风格突变、逻辑矛盾
    ↓
生成代码段 4（500 tokens）❌ 完全偏离需求

关键问题：越往后生成，对初始约束的记忆越模糊，越容易"自由发挥"。

1.2 复杂度的"指数爆炸"

任务复杂度	一次性生成 Bug 率	分步生成 Bug 率
CRUD 单表操作	10%	5%
多表关联 + 事务	35%	12%
分布式系统（含并发、容错）	70%	25%
领域驱动设计（完整模块）	85%	30%

---

二、拆解方法论：四层拆解框架

┌─────────────────────────────────────────────────────────────┐
│  Layer 1: 需求拆解 → 用户故事 + 验收标准                        │
│  输出：结构化的需求卡片，AI 和人类都能理解                       │
├─────────────────────────────────────────────────────────────┤
│  Layer 2: 架构拆解 → 组件 + 接口 + 数据流                      │
│  输出：架构蓝图，定义组件边界和契约                             │
├─────────────────────────────────────────────────────────────┤
│  Layer 3: 实现拆解 → 任务队列 + 依赖图                         │
│  输出：可执行的生成任务，明确先后顺序                           │
├─────────────────────────────────────────────────────────────┤
│  Layer 4: 验证拆解 → 测试金字塔 + 审查清单                      │
│  输出：每层交付物的验证标准，不通过不进入下一步                   │
└─────────────────────────────────────────────────────────────┘

---

三、实战案例：电商订单系统完整拆解

原始需求（复杂度高）

“实现一个完整的电商订单系统，包含购物车、库存管理、价格计算、订单创建、支付集成、物流跟踪、售后退款，支持高并发和分布式事务。”

直接生成的后果：2000+ 行代码，编译错误 50+，逻辑漏洞无数，无法运行。

---

Layer 1：需求拆解 → 用户故事地图

Prompt 模板：

【需求分析师模式】

将以下复杂需求拆解为用户故事地图：

原始需求：{完整需求}

拆解要求：
1. 按用户角色划分（买家、卖家、系统）
2. 按业务流程划分（浏览 → 下单 → 支付 → 履约 → 售后）
3. 每个用户故事包含：角色、功能、价值、验收标准（Given-When-Then）
4. 标记优先级（MVP / Phase 2 / Phase 3）
5. 识别依赖关系（哪些故事必须先完成）

输出格式：
- 用户故事地图（可视化文本）
- 优先级排序列表
- 依赖关系图（Mermaid 语法）

禁止：输出代码，仅输出需求分析

Codex 输出：

## 电商订单系统用户故事地图

### 角色
- 买家（Buyer）
- 卖家（Seller）
- 系统（System）

### 业务流程与故事

#### 1. 浏览与选购（Phase 1 - MVP）
作为 买家
我想 浏览商品并加入购物车
以便 准备购买

验收标准：
- Given 商品有库存，When 点击"加入购物车"，Then 购物车数量+1
- Given 商品无库存，When 点击"加入购物车"，Then 提示"缺货"并禁用按钮

依赖：无

---

#### 2. 订单创建（Phase 1 - MVP）
作为 买家
我想 从购物车结算生成订单
以便 确认购买意向

验收标准：
- Given 购物车有商品，When 点击"结算"，Then 生成待支付订单
- Given 库存不足，When 点击"结算"，Then 提示"库存不足"并阻止下单
- Given 订单创建成功，Then 库存预占 15 分钟

依赖：故事 1（购物车）

---

#### 3. 支付处理（Phase 1 - MVP）
作为 买家
我想 支付订单
以便 完成购买

验收标准：
- Given 订单待支付，When 支付成功回调，Then 订单状态变"已支付"
- Given 支付超时 15 分钟，Then 自动取消订单并释放库存
- Given 重复支付，Then 拒绝并原路退款

依赖：故事 2（订单创建）

---

#### 4. 库存与物流（Phase 1 - MVP）
作为 卖家
我想 查看订单并发货
以便 完成履约

验收标准：
- Given 订单已支付，When 卖家确认发货，Then 订单状态变"已发货"
- Given 发货后，Then 买家收到物流通知

依赖：故事 3（支付）

---

#### 5. 售后退款（Phase 2）
...

### 依赖关系图

```mermaid
graph TD
    A[浏览商品] --> B[加入购物车]
    B --> C[创建订单]
    C --> D[支付订单]
    D --> E[卖家发货]
    E --> F[确认收货]
    D --> G[支付超时取消]
    F --> H[售后退款]

MVP 范围（Phase 1）

1. 故事 1-4（浏览到发货）
2. 不包含：优惠计算、积分、多仓库、拆分订单

---

### Layer 2：架构拆解 → 组件与契约

**Prompt 模板**：

```markdown
【架构师模式】

基于以下用户故事，设计技术架构：

用户故事：{Layer 1 输出}

技术约束：
- 后端：Python + Django + PostgreSQL + Redis
- 前端：React（已有，仅需 API 设计）
- 部署：Docker + Kubernetes

架构要求：
1. 遵循 Clean Architecture / DDD 分层
2. 定义各层组件：Domain / Application / Infrastructure / Interface
3. 定义接口契约：API 端点、事件、数据库 Schema
4. 识别技术决策：事务策略、缓存策略、异步处理
5. 输出架构决策记录（ADR）

禁止：输出具体实现代码，仅输出架构设计

Codex 输出：

## 订单系统架构设计

### 分层架构

src/
├── domain/ # 无外部依赖
│ ├── models/
│ │ ├── order.py # 订单聚合根
│ │ ├── order_item.py # 订单项（值对象）
│ │ └── events.py # 领域事件
│ └── services/
│ └── order_service.py # 领域服务（纯业务逻辑）
├── application/ # 仅依赖 domain
│ ├── usecases/
│ │ ├── create_order.py
│ │ ├── pay_order.py
│ │ └── cancel_order.py
│ └── interfaces/
│ ├── order_repo.py # 仓储接口
│ ├── inventory_client.py # 库存服务接口
│ └── payment_gateway.py # 支付网关接口
├── infrastructure/ # 依赖 application
│ ├── persistence/
│ │ ├── django_order_repo.py
│ │ └── models.py # Django ORM 模型
│ ├── external/
│ │ ├── redis_inventory_client.py
│ │ └── stripe_payment_gateway.py
│ └── messaging/
│ └── celery_tasks.py
└── interface/ # 依赖 application
├── api/
│ ├── views.py # DRF ViewSet
│ └── serializers.py
└── admin/
└── order_admin.py

### 关键接口契约

#### 1. 领域层契约（Python Protocol）

```python
# domain/interfaces/order_repo.py
from typing import Protocol, Optional
from ..models.order import Order

class OrderRepository(Protocol):
    def get_by_id(self, order_id: str) -> Optional[Order]: ...
    def save(self, order: Order) -> None: ...
    def update_status(self, order_id: str, status: OrderStatus) -> None: ...

2. 应用层契约（UseCase 接口）

# application/usecases/create_order.py
from dataclasses import dataclass

@dataclass
class CreateOrderInput:
    user_id: str
    cart_items: list[CartItem]

@dataclass
class CreateOrderOutput:
    order_id: str
    total_amount: Decimal
    status: str

class CreateOrderUseCase:
    def execute(self, input: CreateOrderInput) -> CreateOrderOutput: ...

3. API 契约（OpenAPI 片段）

/api/v1/orders:
  post:
    summary: 创建订单
    requestBody:
      content:
        application/json:
          schema:
            type: object
            properties:
              cart_items:
                type: array
                items:
                  type: object
                  properties:
                    product_id: { type: string, format: uuid }
                    quantity: { type: integer, minimum: 1 }
    responses:
      201:
        description: 订单创建成功
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/Order'

技术决策记录（ADR）

ADR-001: 库存扣减策略

• 决策：使用数据库乐观锁（@Version）+ 应用层重试
• 备选：Redis 分布式锁（太重）、纯数据库事务（性能差）
• 理由：平衡一致性与性能，支持 1000 TPS

ADR-002: 支付超时处理

• 决策：Celery 定时任务（django-celery-beat）
• 备选：数据库轮询（低效）、Redis Keyspace Notify（不可靠）
• 理由：精确到秒级，可水平扩展

ADR-003: 订单状态持久化

• 决策：数据库 + 领域事件表（事件溯源简化版）
• 备选：纯事件溯源（复杂）、状态机表（查询不便）
• 理由：支持审计追踪，同时保持查询简单

---

### Layer 3：实现拆解 → 任务队列与依赖图

**Prompt 模板**：

```markdown
【项目经理模式】

将以下架构设计拆解为可执行的任务队列：

架构设计：{Layer 2 输出}

拆解要求：
1. 每个任务：明确输入（前置任务输出）、输出（交付物）、验证标准
2. 识别依赖关系，生成拓扑排序的任务序列
3. 每个任务限制：代码量 < 200 行，复杂度可单测覆盖
4. 标记可并行任务（无依赖的任务可同时进行）
5. 估算每个任务的 Codex 生成时间（简单/中等/复杂）

输出格式：
- 任务清单（编号、名称、描述）
- 依赖图（哪些任务必须先完成）
- 并行组（可同时生成的任务）
- 每个任务的 Prompt 模板（下一步直接可用）

禁止：生成具体代码

Codex 输出：

## 实现任务队列（拓扑排序）

### 阶段 1：基础设施（必须先完成）

**Task 1.1: 领域模型基础**
- 输入：架构设计中的领域模型定义
- 输出：Order 聚合根、OrderItem 值对象、领域事件基类
- 验证：类型检查通过，可实例化，事件可注册
- 依赖：无
- 复杂度：中等
- Prompt 模板：[见下文 Task 1.1 Template]

**Task 1.2: 仓储接口定义**
- 输入：领域模型
- 输出：OrderRepository Protocol，包含 CRUD + 状态更新
- 验证：mypy 检查通过，可被模拟（mock）
- 依赖：Task 1.1
- 复杂度：简单

**Task 1.3: Django ORM 模型**
- 输入：领域模型、仓储接口
- 输出：Django models.py，包含数据库 Schema
- 验证：makemigrations 成功，migrate 可执行
- 依赖：Task 1.1
- 复杂度：中等

---

### 阶段 2：应用层（依赖阶段 1）

**Task 2.1: 创建订单用例（核心）**
- 输入：领域模型、仓储接口、库存服务接口
- 输出：CreateOrderUseCase 类，包含事务边界
- 验证：单元测试通过（正常、库存不足、并发冲突）
- 依赖：Task 1.1, 1.2
- 复杂度：复杂（需重点审查）

**Task 2.2: 支付订单用例**
- 输入：领域模型、支付网关接口
- 输出：PayOrderUseCase 类，处理回调幂等
- 验证：单元测试通过（成功、重复支付、超时）
- 依赖：Task 1.1, 1.2
- 复杂度：中等

**Task 2.3: 取消订单用例（含超时）**
- 输入：领域模型、Celery 任务框架
- 输出：CancelOrderUseCase + Celery 定时任务
- 验证：单元测试 + 集成测试（定时触发）
- 依赖：Task 1.1, 1.2
- 复杂度：中等

---

### 阶段 3：接口层（依赖阶段 2）

**Task 3.1: DRF Serializers**
- 输入：用例输入输出 DTO
- 输出：CreateOrderSerializer, OrderResponseSerializer
- 验证：序列化/反序列化测试通过
- 依赖：Task 2.1, 2.2, 2.3
- 复杂度：简单

**Task 3.2: DRF ViewSet**
- 输入：Serializers、UseCases
- 输出：OrderViewSet，包含 create/pay/cancel action
- 验证：API 测试通过（django-rest-framework 测试客户端）
- 依赖：Task 3.1
- 复杂度：中等

---

### 并行组（可同时生成）

- 组 A：Task 1.1, 1.2（无依赖）
- 组 B：Task 1.3（仅依赖 1.1）
- 组 C：Task 2.1, 2.2, 2.3（依赖阶段 1，彼此独立）
- 组 D：Task 3.1, 3.2（依赖阶段 2）

---

### Task 1.1 Prompt 模板

```markdown
【Task 1.1】领域模型基础

上下文：
- 架构设计：{Layer 2 输出}
- 技术栈：Python 3.11 + dataclasses + typing

任务：
实现以下领域模型，禁止任何外部依赖（如 Django、SQLAlchemy）：

1. Order 聚合根
   - 属性：id(UUID), user_id, items(list), total_amount(Money), 
          status(OrderStatus), created_at, domain_events(list)
   - 方法：create(工厂方法), pay, cancel, add_item, calculate_total
   - 领域事件：OrderCreated, OrderPaid, OrderCancelled

2. OrderItem 值对象
   - 属性：product_id, product_name, unit_price(Money), quantity, subtotal(Money)
   - 验证：quantity > 0, unit_price > 0

3. Money 值对象
   - 属性：amount(Decimal), currency(str)
   - 方法：add, multiply, validate

4. 领域事件基类
   - 属性：event_id, occurred_on, event_type

约束：
- 使用 @dataclass(frozen=True) 实现值对象不可变
- Order 使用普通 class，支持可变状态（但 items 用 tuple 不可变）
- 所有方法返回 Result 类型（成功/失败），禁止抛异常
- 类型提示覆盖率 100%，禁用 Any

验证：
- mypy --strict 零错误
- 可实例化：Order.create(user_id="u1", items=[...])
- 领域事件可注册和弹出

输出：完整 Python 代码 + 简单使用示例

---

### Layer 4：验证拆解 → 测试与审查

**Prompt 模板**：

```markdown
【QA 工程师模式】

为以下任务设计验证策略：

任务：{具体任务，如 Task 2.1 创建订单用例}

验证要求：
1. 单元测试：正常路径、边界条件、异常场景
2. 集成测试：数据库、外部服务（mock）
3. 架构测试：分层依赖检查
4. 性能测试：并发场景（如果有）
5. 审查清单：代码审查要点

输出：
- 测试用例列表（Given-When-Then）
- 测试代码结构（pytest 组织方式）
- 架构测试规则（ArchUnit 或自定义脚本）
- 审查清单（人工审查要点）

禁止：生成被测代码的具体实现

Codex 输出：

## Task 2.1 创建订单用例验证策略

### 单元测试用例

```python
# tests/application/test_create_order.py
class TestCreateOrderUseCase:
    """创建订单用例单元测试"""
    
    def test_execute_success(self):
        """
        Given: 有效购物车，库存充足
        When: 执行创建订单
        Then: 返回成功，订单状态 CREATED，库存预占
        """
    
    def test_execute_insufficient_inventory(self):
        """
        Given: 商品库存不足
        When: 执行创建订单
        Then: 返回失败，库存服务未确认预占，无订单创建
        """
    
    def test_execute_empty_cart(self):
        """
        Given: 空购物车
        When: 执行创建订单
        Then: 返回失败，提示"购物车不能为空"
        """
    
    def test_execute_concurrent_reservation(self):
        """
        Given: 并发场景，最后一件库存被其他订单预占
        When: 执行创建订单
        Then: 乐观锁冲突，重试后返回失败或成功（取决于时序）
        """
    
    def test_execute_pricing_error(self):
        """
        Given: 价格服务异常
        When: 执行创建订单
        Then: 返回失败，库存预占已回滚（事务一致性）
        """

集成测试

# tests/integration/test_order_creation.py
@pytest.mark.django_db(transaction=True)
class TestOrderCreationIntegration:
    """数据库集成测试"""
    
    def test_order_persisted_correctly(self):
        """验证订单正确持久化到数据库"""
    
    def test_inventory_reserved_in_db(self):
        """验证库存预占记录写入"""
    
    def test_event_published_to_message_queue(self):
        """验证领域事件发布到 Redis/RabbitMQ"""

架构测试

# tests/architecture/test_layer_dependencies.py
def test_domain_does_not_depend_on_application():
    """领域层禁止依赖应用层"""
    assert not imports("domain", "application")

def test_domain_does_not_depend_on_infrastructure():
    """领域层禁止依赖基础设施层"""
    assert not imports("domain", "infrastructure")

def test_usecase_does_not_depend_on_framework():
    """用例层禁止直接依赖 Django/Flask"""
    assert not imports("application.usecases", "django")

代码审查清单

• 事务边界是否正确（@transaction.atomic）
• 库存预占和回滚是否对称（try-finally 或 Saga）
• 领域事件是否在事务提交后发布（避免事务内发布）
• 输入验证是否在应用层入口完成
• 外部服务调用是否有超时和重试
• 日志是否记录关键业务操作（who, when, what）
• 敏感数据是否脱敏（用户 ID 可记录，密码不可）

---

## 四、执行工作流：人机协作流水线

┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│ 人类分析师 │ → │ AI 拆解 │ → │ 人类确认 │ → │ AI 生成 │
│ 提供需求 │ │ 四层拆解 │ │ 调整优先级│ │ 单任务 │
└─────────┘ └─────────┘ └─────────┘ └────┬────┘
↓
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│ 集成测试 │ ← │ 架构测试 │ ← │ 代码审查 │ ← │ 验证交付 │
│ 端到端 │ │ 分层检查 │ │ 人工+AI │ │ 单元测试 │
└─────────┘ └─────────┘ └─────────┘ └────┬────┘
↓
┌─────────┐
│ 通过？ │
└────┬────┘
↓
┌─────────────┐
│ 是 → 下一任务 │
│ 否 → 反馈修正 │
└─────────────┘

---

## 五、实战技巧：让拆解更高效的 5 个秘诀

### 秘诀 1：强制暂停点（Checkpoint）

```markdown
【Checkpoint 模式】

任务：实现订单创建用例

生成规则：
1. 完成输入验证逻辑后 → 停止，输出"Checkpoint 1 完成"
2. 完成库存预占逻辑后 → 停止，输出"Checkpoint 2 完成"
3. 完成订单持久化后 → 停止，输出"Checkpoint 3 完成"
4. 完成事件发布后 → 停止，输出"Checkpoint 4 完成"

每个 Checkpoint 后，等待人工确认"继续"或"修正"后再进行下一步。

好处：早期发现错误，不累积到后期。

秘诀 2：依赖显式化

# 在 Prompt 中强制声明依赖
"""
【依赖声明】

本任务依赖以下已完成的代码（必须引用，禁止重新实现）：
- src/domain/models/order.py (Order 聚合根)
- src/domain/models/money.py (Money 值对象)
- src/domain/interfaces/order_repo.py (OrderRepository 接口)

引用方式：
```python
from src.domain.models.order import Order, OrderStatus
from src.domain.models.money import Money
from src.domain.interfaces.order_repo import OrderRepository

验证：生成的代码必须包含上述 import，且使用这些类型。
“”"

### 秘诀 3：接口先行（Contract First）

```python
# 先让 Codex 生成接口，再生成实现
"""
步骤 1：仅生成接口定义（停止）
步骤 2：人工审查接口（确认）
步骤 3：基于接口生成实现（继续）
"""

# 示例：仓储接口
class OrderRepository(Protocol):
    def get_by_id(self, order_id: UUID) -> Optional[Order]: ...
    def save(self, order: Order) -> Result[None, RepositoryError]: ...

# 审查要点：返回类型是否足够（Optional vs Result）、异常策略

秘诀 4：对抗性生成（红队测试）

【红队模式】

任务：审查刚生成的 CreateOrderUseCase 代码

角色：代码审查员（挑剔、找 Bug）

审查维度：
1. 并发安全：是否有竞态条件？
2. 事务边界：是否过大或过小？
3. 异常处理：是否遗漏失败场景？
4. 资源泄漏：数据库连接、锁是否释放？
5. 业务逻辑：是否符合需求？

输出：
- 发现的问题（行号 + 说明）
- 严重程度（阻断/警告/建议）
- 修复建议

禁止：说"看起来不错"，必须找出至少 3 个问题。

秘诀 5：版本锁定（防止漂移）

【版本锁定】

本任务使用的技术版本（禁止假设其他版本）：
- Python: 3.11.4
- Django: 4.2.7
- DRF: 3.14.0
- Celery: 5.3.0
- PostgreSQL: 15

API 验证：
```python
import django
assert django.VERSION == (4, 2, 7, 'final', 0)

如果发现版本不匹配，立即停止并提示。

---

## 六、常见错误与修正

| 错误模式 | 现象 | 修正策略 |
|---------|------|---------|
| **任务过大** | 单次生成 >500 行，后期质量崩坏 | 强制拆分，每任务 <200 行 |
| **依赖隐式** | 代码中直接实例化依赖类 | 强制接口注入，禁止 new |
| **上下文丢失** | 后期代码风格/命名突变 | 每 3 个任务重新加载规范文件 |
| **验证缺失** | 生成代码无测试，或测试无法运行 | 测试先行，不通过不合并 |
| **过度设计** | 为简单任务引入复杂模式 | YAGNI 原则，MVP 后重构 |

---

## 七、一句话总结

> **复杂任务拆解的本质，是把"AI 的黑盒生成"转化为"可验证的白盒步骤"。每层拆解都是约束的强化，每个 Checkpoint 都是质量的闸门。**