研究提案

HarnessForge v2 — Claude Code CLI 实现版

用 Claude Code CLI + Python 编排脚本实现 6-Agent 代码任务工厂，可直接在终端执行

HarnessForge Claude Code Agent SDK 数据合成 编码任务 RL训练

HarnessForge v2 — Claude Code CLI 实现版

前置依赖：claude CLI（Claude Code）已安装、ANTHROPIC_API_KEY 已设置、Docker 已安装

与规格书版的区别：规格书版是抽象的 Agent 协议设计，本文是具体的 claude -p 命令和 Python 编排脚本，复制粘贴就能跑。

为什么 Claude Code CLI 是最佳实现方式

需求Claude Code CLI 能力
Agent 在真实目录中读写代码✅ 原生支持 Read/Write/Edit/Bash
沙箱执行（不影响宿主）✅ 指定 --cwd 到 Docker volume
多轮对话（PM 感知 Dev 状态）✅ --resume session_id 继续对话
结构化输出（JSON task 卡片）✅ --output-format json --json-schema
工具权限控制（Dev 不能上网）✅ --allowedTools 白名单
编排多 Agent 协作Python 脚本调用 claude -p，解析 JSON 传递

不需要写框架。每个 Agent 就是一次 claude -p 调用，Python 脚本做编排和状态传递。

整体架构

run.py（Python 编排器）
  │
  ├─ Step 1: Repo Selector   →  claude -p "..." → repos.json
  ├─ Step 2: Task Designer    →  claude -p "..." → blueprints/{repo}.json
  ├─ Step 3: PM               →  claude -p "..." → tasks/{task_id}.json
  ├─ Step 4-5: Dev + Test 循环 →  claude -p "..." --resume → trajectories/{task_id}.jsonl
  │      ↑                ↓
  │      PM 观察 Dev 输出，决定是否释放 hint
  │
  └─ Step 6: Reviewer         →  claude -p "..." → verified/{task_id}.json

关键点：Dev 和 PM 之间的交互通过 Python 编排器中转，不是两个 Claude 实例直接对话。编排器解析 Dev 的 JSON 输出，用规则判断是否需要提示，如果需要就把 hint 注入下一轮 Dev 的 prompt。

目录结构

harnessforge/
├── run.py                    # 主编排脚本
├── config.yaml               # 配置（repo 列表、模型选择、预算）
├── prompts/                  # 各 Agent 的 system prompt
│   ├── repo_selector.md
│   ├── task_designer.md
│   ├── pm.md
│   ├── dev.md
│   └── reviewer.md
├── schemas/                  # JSON Schema（约束输出格式）
│   ├── repos.schema.json
│   ├── blueprint.schema.json
│   ├── task.schema.json
│   └── verdict.schema.json
├── workdir/                  # 运行时数据
│   ├── repos.json
│   ├── blueprints/
│   ├── tasks/
│   ├── trajectories/
│   └── verified/
└── docker/                   # Dev 沙箱 Dockerfile 模板
    └── Dockerfile.template

Step 1: Repo Selector

一次性执行，输出 repo 列表。

claude -p \
  --bare \
  --append-system-prompt-file prompts/repo_selector.md \
  --output-format json \
  --json-schema "$(cat schemas/repos.schema.json)" \
  --allowedTools "Bash(curl *),Bash(gh *),Read" \
  "从 GitHub 上挑选 30 个适合生成编码训练任务的仓库。
   要求：
   - 有 CI + 测试覆盖 >70%
   - 近 6 月 >100 commits
   - 10K-500K 行代码
   - 有 release tag（至少 3 个版本）
   - 依赖管理完善
   - 覆盖 Python(12) / TypeScript(10) / Go(5) / Rust(3)
   输出 JSON 格式的 repo 列表。" \
  | jq '.structured_output' > workdir/repos.json

prompts/repo_selector.md：

你是一个代码仓库挑选专家。你的任务是从 GitHub 上找到适合生成 AI 编码训练数据的高质量仓库。

评估标准：
1. 测试基础设施完善（CI/CD 配置、测试框架、覆盖率）
2. 有清晰的版本历史（release tags、语义化版本）
3. 代码规模适中（太小没挑战、太大环境不可控）
4. PR 和 Issue 质量高（描述规范、有 label）
5. 社区活跃（近期有维护）

你可以用 `gh` CLI 查询 GitHub API，用 `curl` 获取仓库信息。

schemas/repos.schema.json：

{
  "type": "object",
  "properties": {
    "repos": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "full_name": { "type": "string" },
          "language": { "type": "string" },
          "stars": { "type": "number" },
          "latest_release": { "type": "string" },
          "release_count": { "type": "number" },
          "test_framework": { "type": "string" },
          "loc": { "type": "number" },
          "reason": { "type": "string" }
        },
        "required": ["full_name", "language", "latest_release", "reason"]
      }
    }
  },
  "required": ["repos"]
}

Step 2: Task Designer

每个 repo 执行一次。在 Docker 中分析 repo，设计 feature 任务。

# 先 clone repo 并切到合适的 base version
git clone --depth=100 https://github.com/$REPO workdir/repos/$REPO_SLUG
cd workdir/repos/$REPO_SLUG
git checkout $BASE_TAG

# 让 Claude 分析 repo 并设计任务
claude -p \
  --bare \
  --cwd "workdir/repos/$REPO_SLUG" \
  --append-system-prompt-file prompts/task_designer.md \
  --output-format json \
  --json-schema "$(cat schemas/blueprint.schema.json)" \
  --allowedTools "Read,Bash(git *),Bash(find *),Bash(wc *),Bash(head *),Bash(cat *)" \
  "你在 $REPO 的 $BASE_TAG 版本中。
   1. 分析 release notes 和近期 PR，理解 repo 演进方向
   2. 设计 3-5 个有挑战的 feature 任务
   3. 每个 feature 必须包含：特性描述、验证方式（测试命令+期望行为）、预估难度
   4. 如果某个 feature 对应后续 release 中的真实变更，记录 ground_truth_tag
   
   repo: $REPO
   base_tag: $BASE_TAG
   next_tags: $NEXT_TAG_1, $NEXT_TAG_2" \
  | jq '.structured_output' > workdir/blueprints/$REPO_SLUG.json

prompts/task_designer.md：

你是一个资深软件架构师，负责分析开源仓库并设计有挑战性的开发任务。

你的目标是设计出能训练 AI 编码 Agent 的高质量任务。好的任务应该：
1. 需要理解多个模块的交互关系
2. 不能只改一两行就完成（至少涉及 3+ 文件）
3. 有明确的验证方式（测试命令能判定成功/失败）
4. 难度适中（一个优秀工程师需要 2-8 小时完成）

分析 repo 时：
- 用 `git log --oneline -50` 看近期提交方向
- 用 `git tag -l` 看版本历史
- 读 CHANGELOG.md 或 release notes 了解版本特性
- 浏览核心目录结构理解架构

设计任务时：
- 基于真实的演进方向（不要凭空编造需求）
- 如果 next_tag 中包含了某个 feature 的实现，记录为 ground_truth
- 验证方式优先用 repo 现有的测试框架

Step 3: PM（需求重写 + 提示准备）

每个 task 执行一次。把技术性的 blueprint 改写为用户视角的需求。

claude -p \
  --bare \
  --append-system-prompt-file prompts/pm.md \
  --output-format json \
  --json-schema "$(cat schemas/task.schema.json)" \
  --allowedTools "Read" \
  "将以下技术 blueprint 转换为产品需求文档。

   Blueprint:
   $(cat workdir/blueprints/$REPO_SLUG.json | jq ".features[$I]")
   
   Repo: $REPO
   
   要求：
   1. 用产品经理的语言描述需求（不暴露实现细节）
   2. 包含验收标准（用户可感知的行为）
   3. 准备 3 级渐进提示（L1 方向级 / L2 模块级 / L3 代码级）
   4. L1 在 Dev 连续 3 次失败后释放
   5. L2 在 L1 释放 5 轮后仍无进展时释放
   6. L3 仅作为最后手段" \
  | jq '.structured_output' > workdir/tasks/$TASK_ID.json

Step 4-5: Dev + Test + PM 交互循环

核心循环。Python 编排器控制 Dev 执行、Test 验证、PM 提示释放。

这是最关键的部分。不能用单次 claude -p 实现——需要多轮交互，且编排器在每轮之间做决策。

run_dev.py（核心编排逻辑）

#!/usr/bin/env python3
"""HarnessForge Dev 执行循环 — 基于 Claude Code CLI"""

import json
import subprocess
import os
import sys
from dataclasses import dataclass, field
from typing import Optional

MAX_TURNS = 50
HINT_COOLDOWN = 8          # 前 8 轮不介入
L1_TRIGGER_FAILS = 3       # 连续 3 次相同失败 → L1
L2_TRIGGER_TURNS = 5       # L1 释放 5 轮后仍无进展 → L2
L3_TRIGGER_TURNS = 5       # L2 释放 5 轮后仍无进展 → L3

@dataclass
class DevState:
    turn: int = 0
    test_results: list = field(default_factory=list)  # 每轮测试结果
    hint_level: int = 0     # 当前已释放的最高提示级别 (0=无, 1=L1, 2=L2, 3=L3)
    hint_released_at: int = 0  # 最后一次释放 hint 的 turn
    trajectory: list = field(default_factory=list)

    @property
    def consecutive_same_failures(self) -> int:
        """连续出现相同错误信息的次数"""
        if len(self.test_results) < 2:
            return 0
        count = 1
        for i in range(len(self.test_results) - 2, -1, -1):
            if (self.test_results[i].get("status") == "fail" and
                self.test_results[i+1].get("status") == "fail" and
                self.test_results[i].get("error_summary") == self.test_results[i+1].get("error_summary")):
                count += 1
            else:
                break
        return count

    @property
    def turns_since_hint(self) -> int:
        if self.hint_released_at == 0:
            return 999
        return self.turn - self.hint_released_at

    @property
    def has_progress(self) -> bool:
        """最近 5 轮是否有进展（通过的测试数增加）"""
        recent = self.test_results[-5:]
        if len(recent) < 2:
            return True
        passes = [r.get("pass_count", 0) for r in recent if r.get("status") == "fail"]
        if len(passes) < 2:
            return True
        return passes[-1] > passes[0]

    def should_release_hint(self) -> Optional[int]:
        """决定是否释放 hint，返回 hint level 或 None"""
        if self.turn < HINT_COOLDOWN:
            return None
        if self.hint_level == 0:
            if self.consecutive_same_failures >= L1_TRIGGER_FAILS:
                return 1
        elif self.hint_level == 1:
            if self.turns_since_hint >= L2_TRIGGER_TURNS and not self.has_progress:
                return 2
        elif self.hint_level == 2:
            if self.turns_since_hint >= L3_TRIGGER_TURNS and not self.has_progress:
                return 3
        return None


def run_claude(prompt: str, cwd: str, session_id: str = None,
               system_prompt_file: str = None,
               allowed_tools: str = "Read,Edit,Write,Bash") -> dict:
    """调用 claude CLI 一次，返回 JSON 结果"""
    cmd = [
        "claude", "-p", prompt,
        "--bare",
        "--output-format", "json",
        "--cwd", cwd,
        "--allowedTools", allowed_tools,
        "--permission-mode", "acceptEdits",
    ]
    if session_id:
        cmd.extend(["--resume", session_id])
    if system_prompt_file:
        cmd.extend(["--append-system-prompt-file", system_prompt_file])

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    if result.returncode != 0:
        return {"error": result.stderr, "result": ""}
    return json.loads(result.stdout)


def run_test(cwd: str, test_cmd: str) -> dict:
    """在 Docker 中运行测试，返回结构化结果"""
    result = subprocess.run(
        ["docker", "exec", "harnessforge-sandbox", "bash", "-c",
         f"cd /workspace && {test_cmd}"],
        capture_output=True, text=True, timeout=120
    )
    # 解析测试输出
    output = result.stdout + result.stderr
    pass_count = output.count("PASSED") or output.count("passed")
    fail_count = output.count("FAILED") or output.count("failed")

    return {
        "status": "pass" if result.returncode == 0 else "fail",
        "pass_count": pass_count,
        "fail_count": fail_count,
        "output": output[-2000:],  # 截断保留最后 2000 字符
        "error_summary": output.split("\n")[-3] if result.returncode != 0 else ""
    }


def execute_task(task_file: str, repo_dir: str):
    """执行一个完整的 Dev 任务"""
    task = json.load(open(task_file))
    task_id = task["task_id"]
    test_cmd = task["test_cmd"]
    hints = task.get("hints", {})

    state = DevState()
    session_id = None

    # ── 初始 prompt：给 Dev 任务描述 ──
    initial_prompt = f"""你是一个高级软件工程师。请完成以下开发任务：

## 任务描述
{task['description']}

## 验收标准
{chr(10).join('- ' + c for c in task['acceptance_criteria'])}

## 测试命令
完成后运行：{test_cmd}

## 规则
- 先浏览项目结构和相关代码，理解架构
- 修改代码后运行测试验证
- 如果测试失败，分析错误并调整方案
- 所有测试通过后，输出 "TASK_COMPLETE"
"""

    print(f"[HarnessForge] 开始任务 {task_id}")
    print(f"[HarnessForge] 测试命令: {test_cmd}")

    for turn in range(MAX_TURNS):
        state.turn = turn
        prompt = initial_prompt if turn == 0 else None

        # ── PM 决策：是否释放 hint ──
        hint_level = state.should_release_hint()
        if hint_level and hint_level > state.hint_level:
            hint_key = f"L{hint_level}"
            hint_content = hints.get(hint_key, [])
            if hint_content:
                hint_text = "\n".join(f"- {h}" for h in hint_content)
                prompt = f"""[来自技术负责人的反馈]
你的方向可能需要调整。以下是一些额外信息：

{hint_text}

请根据这些提示重新考虑你的方案，然后继续开发。
完成后运行测试：{test_cmd}"""
                state.hint_level = hint_level
                state.hint_released_at = turn
                print(f"[PM] Turn {turn}: 释放 L{hint_level} 提示")

        # ── Dev 执行一轮 ──
        if prompt is None:
            # 中间轮次：让 Dev 继续
            prompt = "继续你的开发工作。如果之前测试失败了，分析错误原因并修复。完成后运行测试。"

        result = run_claude(
            prompt=prompt,
            cwd=repo_dir,
            session_id=session_id,
            system_prompt_file="prompts/dev.md" if turn == 0 else None,
        )

        session_id = result.get("session_id", session_id)
        dev_output = result.get("result", "")

        # ── 记录轨迹 ──
        state.trajectory.append({
            "turn": turn,
            "hint_level": state.hint_level if hint_level else None,
            "dev_output_preview": dev_output[:500],
            "session_id": session_id,
        })

        # ── 检查是否完成 ──
        if "TASK_COMPLETE" in dev_output:
            test_result = run_test(repo_dir, test_cmd)
            state.test_results.append(test_result)
            state.trajectory[-1]["test_result"] = test_result

            if test_result["status"] == "pass":
                print(f"[HarnessForge] ✅ 任务完成！Turn {turn}, Hints: L{state.hint_level}")
                break
            else:
                print(f"[HarnessForge] ❌ 测试未通过 (pass:{test_result['pass_count']} fail:{test_result['fail_count']})")
        else:
            # Dev 没说完成，但可能跑了测试
            test_result = run_test(repo_dir, test_cmd)
            state.test_results.append(test_result)
            state.trajectory[-1]["test_result"] = test_result

    # ── 保存轨迹 ──
    trajectory_file = f"workdir/trajectories/{task_id}.jsonl"
    os.makedirs(os.path.dirname(trajectory_file), exist_ok=True)
    with open(trajectory_file, "w") as f:
        for entry in state.trajectory:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    # ── 保存执行摘要 ──
    summary = {
        "task_id": task_id,
        "total_turns": state.turn + 1,
        "final_status": state.test_results[-1]["status"] if state.test_results else "unknown",
        "max_hint_level": state.hint_level,
        "hint_released_at_turn": state.hint_released_at,
        "trajectory_file": trajectory_file,
    }
    summary_file = f"workdir/trajectories/{task_id}_summary.json"
    with open(summary_file, "w") as f:
        json.dump(summary, f, indent=2, ensure_ascii=False)

    print(f"[HarnessForge] 轨迹保存到 {trajectory_file}")
    return summary


if __name__ == "__main__":
    task_file = sys.argv[1]
    repo_dir = sys.argv[2]
    execute_task(task_file, repo_dir)

使用方式

# 执行单个任务
python run_dev.py workdir/tasks/django-async-queryset.json workdir/repos/django

# 批量执行所有任务
for task in workdir/tasks/*.json; do
  repo=$(jq -r '.repo_dir' "$task")
  python run_dev.py "$task" "$repo"
done

Step 6: Reviewer

claude -p \
  --bare \
  --append-system-prompt-file prompts/reviewer.md \
  --output-format json \
  --json-schema "$(cat schemas/verdict.schema.json)" \
  --allowedTools "Read" \
  "审核以下任务执行结果，判断是否有训练价值。

   任务描述：
   $(cat workdir/tasks/$TASK_ID.json | jq -r '.description')

   执行摘要：
   $(cat workdir/trajectories/${TASK_ID}_summary.json)

   轨迹（前 20 轮）：
   $(head -20 workdir/trajectories/$TASK_ID.jsonl)

   判断标准：
   - difficulty 1-5（5=需要 L2+ 提示才突破）
   - 过程是否有逻辑（不是随机瞎试）
   - 提示后是否有效调整方向
   - 是否值得作为 RL 训练数据

   输出 verdict（保留/丢弃）和最终 task 卡片。" \
  | jq '.structured_output' > workdir/verified/$TASK_ID.json

完整执行脚本 run.py

#!/usr/bin/env python3
"""HarnessForge v2 — 完整编排脚本"""

import json
import subprocess
import os
import glob

def run_step(name, cmd):
    print(f"\n{'='*60}")
    print(f"[Step] {name}")
    print(f"{'='*60}")
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"[ERROR] {result.stderr[:500]}")
    return result

def main():
    os.makedirs("workdir/blueprints", exist_ok=True)
    os.makedirs("workdir/tasks", exist_ok=True)
    os.makedirs("workdir/trajectories", exist_ok=True)
    os.makedirs("workdir/verified", exist_ok=True)

    # ── Step 1: Repo Selector ──
    if not os.path.exists("workdir/repos.json"):
        run_step("Repo Selector", """
            claude -p --bare \
              --append-system-prompt-file prompts/repo_selector.md \
              --output-format json \
              --json-schema "$(cat schemas/repos.schema.json)" \
              --allowedTools "Bash(curl *),Bash(gh *),Read" \
              "挑选 30 个适合生成编码训练任务的 GitHub 仓库..." \
              | jq '.structured_output' > workdir/repos.json
        """)

    repos = json.load(open("workdir/repos.json"))["repos"]
    print(f"[Info] {len(repos)} 个 repo 已选定")

    # ── Step 2-6: 每个 repo 循环 ──
    for repo_info in repos:
        repo = repo_info["full_name"]
        slug = repo.replace("/", "_")
        repo_dir = f"workdir/repos/{slug}"

        # Clone
        if not os.path.exists(repo_dir):
            os.system(f"git clone --depth=200 https://github.com/{repo} {repo_dir}")

        # Step 2: Task Designer
        blueprint_file = f"workdir/blueprints/{slug}.json"
        if not os.path.exists(blueprint_file):
            run_step(f"Task Designer: {repo}", f"""
                claude -p --bare \
                  --cwd {repo_dir} \
                  --append-system-prompt-file prompts/task_designer.md \
                  --output-format json \
                  --json-schema "$(cat schemas/blueprint.schema.json)" \
                  --allowedTools "Read,Bash(git *),Bash(find *),Bash(wc *),Bash(head *),Bash(cat *)" \
                  "分析此仓库并设计 3-5 个有挑战的 feature 任务。repo: {repo}" \
                  | jq '.structured_output' > {blueprint_file}
            """)

        blueprint = json.load(open(blueprint_file))

        # Step 3: PM 为每个 feature 生成 task
        for i, feature in enumerate(blueprint.get("features", [])):
            task_id = f"{slug}_feature_{i}"
            task_file = f"workdir/tasks/{task_id}.json"
            if not os.path.exists(task_file):
                feature_json = json.dumps(feature, ensure_ascii=False)
                run_step(f"PM: {task_id}", f"""
                    claude -p --bare \
                      --append-system-prompt-file prompts/pm.md \
                      --output-format json \
                      --json-schema "$(cat schemas/task.schema.json)" \
                      --allowedTools "Read" \
                      '将以下技术 blueprint 转换为产品需求 + 渐进提示：{feature_json}' \
                      | jq '.structured_output' > {task_file}
                """)

            # Step 4-5: Dev + Test 循环
            traj_file = f"workdir/trajectories/{task_id}.jsonl"
            if not os.path.exists(traj_file):
                run_step(f"Dev: {task_id}",
                    f"python run_dev.py {task_file} {repo_dir}")

            # Step 6: Reviewer
            verdict_file = f"workdir/verified/{task_id}.json"
            if not os.path.exists(verdict_file):
                run_step(f"Reviewer: {task_id}", f"""
                    claude -p --bare \
                      --append-system-prompt-file prompts/reviewer.md \
                      --output-format json \
                      --json-schema "$(cat schemas/verdict.schema.json)" \
                      --allowedTools "Read" \
                      "审核任务 {task_id} 的执行结果..." \
                      | jq '.structured_output' > {verdict_file}
                """)

    # ── 最终统计 ──
    verdicts = glob.glob("workdir/verified/*.json")
    kept = sum(1 for v in verdicts
               if json.load(open(v)).get("verdict") == "keep")
    print(f"\n{'='*60}")
    print(f"[Done] {len(verdicts)} 个任务审核完成，{kept} 个保留")
    print(f"{'='*60}")


if __name__ == "__main__":
    main()

快速开始（3 步）

# 1. 初始化目录和 prompt 文件
mkdir -p harnessforge/{prompts,schemas,workdir,docker}
cd harnessforge

# 2. 创建 prompt 和 schema 文件（从本文档复制）
# ... 或者让 Claude Code 自动创建：
claude -p "根据 /research/harnessforge-v2-claude-code-impl 的内容，
           在当前目录创建所有需要的 prompt 和 schema 文件"

# 3. 运行
python run.py

与 Agent SDK Python 版的对比

如果需要更精细的控制（比如实时观察 Dev 的每一步 tool call），可以用 Agent SDK 替代 CLI：

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def run_dev_with_sdk(task, repo_dir):
    options = ClaudeAgentOptions(
        system_prompt=open("prompts/dev.md").read(),
        permission_mode="acceptEdits",
        cwd=repo_dir,
        allowed_tools=["Read", "Edit", "Write", "Bash"],
        disallowed_tools=["WebSearch", "WebFetch"],
        max_turns=50,
    )

    async with ClaudeSDKClient(options=options) as client:
        # 发送初始任务
        await client.query(f"完成以下开发任务：\n{task['description']}")

        # 逐消息接收，实时监控
        async for message in client.receive_messages():
            # 可以实时解析 Dev 的 tool call
            # 比如检测到 Dev 运行了测试 → 立即解析结果 → 决定是否释放 hint
            if is_test_execution(message):
                test_result = parse_test_result(message)
                if should_release_hint(test_result, state):
                    await client.interrupt()
                    # 排空
                    async for _ in client.receive_response():
                        pass
                    # 注入 hint
                    await client.query(f"[技术负责人反馈] {hint_content}")

CLI vs SDK 的选择：

维度CLI (claude -p)SDK (ClaudeSDKClient)
实现复杂度低（shell 脚本即可）中（需要 async Python）
PM 响应速度每轮结束后判断实时（可在 Dev tool call 中途介入）
轨迹粒度轮级（每轮一个 JSON）消息级（每个 tool call）
推荐场景PoC 验证、小规模大规模生产

建议：先用 CLI 版跑通 PoC（10 个任务），验证 pipeline 可行后再迁移到 SDK 版提升精度和效率。

与规格书版的关系

规格书版（harnessforge-v2-executable-spec）CLI 实现版（本文）
定义了”做什么”定义了”怎么用 Claude Code 做”
Agent 协议和数据格式具体的命令和 Python 代码
适合发给团队理解方案适合直接执行
PM 感知逻辑用伪代码描述PM 感知逻辑用可运行的 Python 实现