Claude Code 动态工作流：让 AI 编排 AI，周末搬完一个编译器

主 agent 分析 Bun 源码结构 → 生成移植计划
  ├── 子 agent 1: 移植 runtime/core.zig → runtime/core.rs
  ├── 子 agent 2: 移植 runtime/jsc.zig → runtime/jsc.rs
  ├── 子 agent 3: 移植 runtime/bundler.zig → runtime/bundler.rs
  ├── ...
  ├── 子 agent N: 移植第 N 个模块
  └── 所有子 agent 完成后 → 主 agent 合并、解决冲突、跑全量测试

# 创建示例项目
mkdir async-refactor-demo && cd async-refactor-demo

# 三个待改造的同步模块
cat > fetcher.py << 'EOF'
import requests

def get_user(id):
    resp = requests.get(f"https://api.example.com/users/{id}")
    return resp.json()

def get_all_users():
    resp = requests.get("https://api.example.com/users")
    return resp.json()
EOF

cat > processor.py << 'EOF'
import json

def transform_user(raw):
    return {
        "id": raw["id"],
        "display": f"{raw['first']} {raw['last']}"
    }

def batch_transform(users):
    return [transform_user(u) for u in users]
EOF

cat > writer.py << 'EOF'
import csv

def save_users(users, path="output.csv"):
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["id", "display"])
        w.writeheader()
        w.writerows(users)
    return path
EOF

# 测试文件
cat > test_sync.py << 'EOF'
from fetcher import get_user, get_all_users
from processor import transform_user, batch_transform
from writer import save_users

def test_transform():
    result = transform_user({"id": 1, "first": "A", "last": "B"})
    assert result["display"] == "A B"

def test_batch():
    users = [{"id": 1, "first": "A", "last": "B"}]
    results = batch_transform(users)
    assert len(results) == 1
EOF

# 确保 Claude Code 已安装并登录
claude --version

# 发起动态工作流任务
claude --dynamic-workflow \
  "将 fetcher.py、processor.py、writer.py 三个模块从同步改为 async。
   每个模块独立改造，并行执行。
   要求：
   1. fetcher.py 用 aiohttp 替换 requests
   2. processor.py 的 batch_transform 改为 asyncio.gather 并发处理
   3. writer.py 用 aiofiles 替换同步文件操作
   4. 所有改造后函数名加 async_ 前缀，原同步函数保留作兼容
   5. 更新 test_sync.py 为 test_async.py，用 pytest-asyncio
   6. 每个子 agent 完成后跑对应测试确认通过"

# Claude Code 会在项目目录生成工作流记录
cat .claude-workflow/plan.json

{
  "task": "async-refactor",
  "subagents": [
    {
      "id": "sa-1",
      "target": "fetcher.py",
      "status": "completed",
      "tests_passed": true
    },
    {
      "id": "sa-2",
      "target": "processor.py",
      "status": "completed",
      "tests_passed": true
    },
    {
      "id": "sa-3",
      "target": "writer.py",
      "status": "completed",
      "tests_passed": true
    }
  ],
  "merge_conflicts": 0,
  "final_test_run": "passed"
}

# 安装 async 依赖
pip install aiohttp aiofiles pytest-asyncio

# 跑测试
pytest test_async.py -v

cat .claude-workflow/orchestration.sh

#!/usr/bash
# Claude Code 动态工作流编排脚本
# 生成时间: 2024-XX-XX

# 阶段1: 并行改造（无依赖，可全并行）
claude --subagent --id sa-1 --task "改造 fetcher.py 为 async 版本" &
claude --subagent --id sa-2 --task "改造 processor.py 为 async 版本" &
claude --subagent --id sa-3 --task "改造 writer.py 为 async 版本" &
wait  # 等待所有并行子 agent 完成

# 阶段2: 合并与验证（依赖阶段1全部完成）
claude --merge-results --from sa-1,sa-2,sa-3
claude --run-tests --pattern "test_async.py"