# 代码任务（算法题） — Agentic RL 最佳实践

> **任务类型**: 代码生成（竞赛编程）  
> **基座模型**: Qwen3.5-9B  
> **训练任务 ID**: `ft-202606151516-e36e`  
> **训练任务名**: `cangyu_qwen3.5-9b_code_gen_ablation_taco_bs256_2ep_lr7e7_kl01`  
> **Rollout 处理器**: `CodeRolloutProcessor`（LangChain StructuredTool + LangGraph）  
> **Reward 处理器**: `CodeRewardProcessor`（format 0.1 + accuracy/pass@1 0.9）

---

## 1. 数据来源与准备

### 1.1 训练集：TACO 随机采样子集

训练使用 **TACO 随机采样子集**（`data/ablation_taco.jsonl`，4300 样本，~783 MB），由 [`scripts/sample_ablation_subsets.py`](scripts/sample_ablation_subsets.py) 从 HuggingFace `agentica-org/DeepCoder-Preview-Dataset` 的 TACO 子集采样生成，并保留历史采样流程的随机状态以确保样本与实验记录一致。数据集文件较大，demo 的 `data/` 目录通常为空；运行 `submit_job.py` 时会先检查脚本同级 `data/` 下是否已有该文件，若不存在则自动执行生成脚本，存在则跳过。

### 1.2 验证集：LiveCodeBench v6

使用 [LiveCodeBench](https://huggingface.co/datasets/livecodebench/code_generation_lite) v6 的 **2501-2505 子集**作为验证集（`data/livecodebench_v6_test.jsonl`，182 样本），这是一个高质量的实时竞赛编程评测基准。验证集由 [`scripts/download_livecodebench.py`](scripts/download_livecodebench.py) 下载并转换，脚本会显式保留 `contest_date` 落在 `[2025-01-01, 2025-06-01)` 的题目，输出文件名与提交脚本中的 `validation_file` 保持一致。

数据准备方式：

- **自动准备**：直接运行 `python submit_job.py`，提交脚本会检查 `data/ablation_taco.jsonl` 和 `data/livecodebench_v6_test.jsonl`；缺哪个就自动运行对应脚本，已有文件会跳过。全新 `data/` 目录下最终只会生成这两个与实验记录一致的数据文件。
- **手动预下载**：如需在提交前显式准备数据，可执行：
  ```bash
  python scripts/sample_ablation_subsets.py  # 仅生成训练集 data/ablation_taco.jsonl
  python scripts/download_livecodebench.py   # 生成验证集 data/livecodebench_v6_test.jsonl（LiveCodeBench v6 2501-2505）
  ```

### 1.3 数据预处理

预处理脚本：[`scripts/sample_ablation_subsets.py`](scripts/sample_ablation_subsets.py)

主要处理步骤：
1. 从 HuggingFace TACO 子集随机采样 4300 条，保留历史采样随机状态并输出 `data/ablation_taco.jsonl`
2. 解析原始 test_cases（JSON 字符串形式的 inputs/outputs）
3. 从全量 test cases 中选取 public_test_cases（最短 + 分层采样子集，供 rollout 工具使用）
4. 过滤含 `<image>` token 的多模态样本

### 1.4 JSONL 数据 Schema

#### 训练集（TACO）— stdin 模式

```json
{
  "messages": [
    {"content": "You are given an array of n integers...", "role": "user"}
  ],
  "rollout_extra": {
    "test_cases": "{\"inputs\": [\"3\\n1 2 3\", ...], \"outputs\": [\"6\", ...]}",
    "public_test_cases": "[{\"input\": \"3\\n1 2 3\", \"output\": \"6\", \"testtype\": \"stdin\"}]",
    "dataset": "deepcoder",
    "data_source": "codeforces",
    "difficulty": "1200"
  }
}
```

#### 验证集（LiveCodeBench）— functional 模式

```json
{
  "messages": [
    {"content": "Given an array nums, return the maximum...", "role": "user"}
  ],
  "rollout_extra": {
    "public_test_cases": "[{\"input\": \"nums = [1,2,3]\", \"output\": \"6\"}]",
    "private_test_cases": "...(base64+zlib+pickle 编码)",
    "starter_code": "class Solution:\n    def maxSum(self, nums: List[int]) -> int:",
    "metadata": "{\"func_name\": \"maxSum\"}"
  }
}
```

| 字段 | 说明 |
|------|------|
| `test_cases` | 全量测试用例（reward 评分用） |
| `public_test_cases` | 公开测试用例（rollout 工具调用用，仅样例） |
| `starter_code` | 有此字段时使用 functional 模式 |
| `metadata.func_name` | 函数名，functional 模式下用于构造测试调用 |
| `difficulty` | 难度标签（如 Codeforces rating） |

---

## 2. 训练参数配置

> 配置文件：[`submit_job.py`](submit_job.py)

### 2.1 核心超参数

> 源码位置：[`submit_job.py` L143-160](submit_job.py#L143-L160)

```python
hyper_parameters={
    "n_epochs": 1,
    "learning_rate": 7e-7,
    "kl_loss_coef": 0.01,
    "max_prompt_length": 12288,
    "max_length": 20480,
    "sequence_parallel_size": 2,
    "ppo_mini_batch_size": 64,
    "batch_size": 256,
    "eval_steps": 2,
    "nnodes_training": 2,
    "nnodes_rollout": 1,
    "save_strategy": "steps",
    "save_steps": save_steps,
    "save_total_limit": 3,
    "lr_scheduler_type": "constant",
    "lr_warmup_steps": 3,
    "inference_gpu_memory_utilization": 0.75,
}
```

### 2.2 与数学任务的关键差异

| 参数 | MATH/U-MATH | Code | 差异原因 |
|------|-------------|------|---------|
| `learning_rate` | 1e-6 | **7e-7** | 代码生成策略更敏感，需要更小的学习率 |
| `kl_loss_coef` | 默认 (0.001) | **0.01** | 代码任务策略变化更敏感，增大 KL 正则约束偏移幅度 |
| `max_prompt_length` | 8192 | **12288** | 代码题多轮工具调用使 context 更长 |
| `max_length` | 16384 | **20480** | 12288 prompt + 8192 response |
| `sequence_parallel_size` | 默认 (1) | **2** | 将序列切分到 2 张卡上，突破单卡 16k 序列长度限制 |
| `batch_size` | 128 | **256** | 更大 batch 降低梯度方差 |
| `ppo_mini_batch_size` | 默认 (16) | **64** | 增大 mini batch 降低训练更新频率，减少 rollout 策略过期 |
| `lr_scheduler_type` | cosine | **constant** | 恒定学习率，配合短训练步数 |
| `lr_warmup_steps` | 无 | **3** | warmup 阶段稳定训练初期 |
| Rollout `max_turns` | 10 | **5** | 限制工具交互轮数，控制 context 增长 |

### 2.3 训练步数计算

```
total_samples = 4300
steps_per_epoch = 4300 // 256 = 16
total_steps = 16 × 2 = 32
```

---

## 3. 工具设计配置与跟踪

### 3.1 架构总览

> 源码位置：[`functions/rollout/code_rollout.py`](functions/rollout/code_rollout.py)（`CodeRolloutProcessor` [L159](functions/rollout/code_rollout.py#L159)）

与数学任务的关键架构差异：**代码任务不使用 MCP Server**，而是通过 LangChain `StructuredTool` 在每次 rollout 时动态创建工具。

```mermaid
flowchart LR
    subgraph RolloutProcessor["CodeRolloutProcessor"]
        LLM["策略模型<br/>Qwen3.5-9B"]
        Agent["LangGraph<br/>Agent 节点"]
        Tools["LangGraph<br/>Tools 节点"]
    end

    subgraph ToolClosure["Per-Problem Tool"]
        EC["execute_code<br/>StructuredTool"]
        ET["extract_code()"]
        RT["run_public_tests()"]
    end

    Agent -->|"tool_calls"| Tools
    Tools --> EC
    EC --> ET
    ET --> RT
    RT -->|"pass/fail 结果"| Tools
    Tools --> Agent
    Agent -->|"最终代码"| Output["RolloutOutput"]
```

### 3.2 Public / Private Test Case 分割

> 采样策略实现：[`scripts/sample_ablation_subsets.py` L36-72](scripts/sample_ablation_subsets.py#L36-L72)（`_split_public()`）
> Rollout 侧解析：[`functions/rollout/code_rollout.py` L105-118](functions/rollout/code_rollout.py#L105-L118)（`_resolve_public_tests` / `_resolve_fn_name`）
> Reward 侧归一化：[`functions/reward/code_testing.py` L219-267](functions/reward/code_testing.py#L219-L267)（`normalise_test_cases()`）

代码任务的工具设计中最关键的一点是：**rollout 工具只提供部分测试用例（public），reward 评分使用全部测试用例（public + private）**。这模拟了真实竞赛场景——选手只能看到样例测试，提交后由裁判运行完整测试。

```mermaid
flowchart LR
    subgraph 数据预处理["sample_ablation_subsets.py"]
        All["全部 test cases<br/>(几十~上百个)"]
        Split["shortest + stratified<br/>采样策略"]
        All --> Split
        Split --> Public["public_test_cases<br/>3~6 个样例"]
        Split --> Private["剩余 = private<br/>(不暴露给模型)"]
    end

    subgraph Rollout["Rollout (模型可见)"]
        Tool["execute_code 工具"]
        Public -.->|"绑定"| Tool
    end

    subgraph Reward["Reward (模型不可见)"]
        Judge["run_tests()"]
        All -.->|"normalise_test_cases()"| Judge
    end
```

#### 数据中的字段分工

每条训练样本的 `rollout_extra` 包含两组测试用例：

```json
{
  "test_cases": "{\"inputs\": [...], \"outputs\": [...]}",
  "public_test_cases": "[{\"input\": ..., \"output\": ..., \"testtype\": \"stdin\"}]"
}
```

| 字段 | 用途 | 使用者 |
|------|------|--------|
| `test_cases` | **全部**测试用例（用于评分） | Reward Processor (`normalise_test_cases()`) |
| `public_test_cases` | **部分**测试用例（用于调试） | Rollout 工具 (`_resolve_public_tests()`) |

#### Public Test Case 采样策略

[`scripts/sample_ablation_subsets.py`](scripts/sample_ablation_subsets.py) 中的 `_split_public()` 实现了一个 **shortest + stratified** 采样策略，在给模型足够调试信息的同时控制 token 消耗：

```
全部 test cases
  └── 按大小排序
       └── 取短半段（前 50%）
            ├── Step 1: 取最短的 3 个（NUM_SHORTEST=3）
            ├── Step 2: 从剩余短半段中，过滤掉超过 6000 字符的，
            │           均匀分层取 3 个（NUM_STRATIFIED=3）
            └── Step 3: 累计字符数不超过 18000（TOTAL_CHAR_BUDGET）
```

| 参数 | 值 | 说明 |
|------|-----|------|
| `NUM_SHORTEST` | 3 | 优先选最短的 case，保证模型至少能快速验证基本逻辑 |
| `NUM_STRATIFIED` | 3 | 从短半段中分层采样，增加覆盖度（不全选最简单的） |
| `PER_CASE_CHAR_BUDGET` | 6000 字符 | 单个 case 上限 ≈ 2000 tokens |
| `TOTAL_CHAR_BUDGET` | 18000 字符 | 全部 public cases 累计上限 ≈ 6000 tokens |

**设计动机**：
- **为什么不给全部 test cases？** 很多题目有几十甚至上百个测试，全部放入 rollout context 会超出 `max_length`；且模型若能看到全部预期输出，相当于泄露答案，无法训练出泛化能力
- **为什么不只给 1 个？** 太少的样例无法帮助模型有效调试，RL 中模型需要通过工具反馈来迭代改进
- **为什么用 shortest + stratified？** 最短的 case 运行快、占 token 少，保证基本功能验证；分层取样增加多样性，覆盖更多边界情况
- **为什么只从短半段取？** 长 test case 的 expected output 可能几百 KB，即使经过截断也会消耗大量 context

#### Rollout vs Reward 的数据流对比

| 阶段 | 读取字段 | 函数 | test cases 范围 | 目的 |
|------|---------|------|----------------|------|
| **Rollout** | `public_test_cases` | `_resolve_public_tests()` | 3~6 个样例 | 给模型提供调试反馈 |
| **Reward** | `test_cases` + `public_test_cases` + `private_test_cases` | `normalise_test_cases()` | 全部 | 公正评分 |

> **LiveCodeBench 数据源**的差异：LiveCodeBench 格式将 private tests 编码为 `base64 + zlib + pickle`，
> `normalise_test_cases()` 会自动解码并合并。TACO 格式直接用 `test_cases` 字段存储全部测试。

### 3.3 为什么不用 MCP Server？

数学任务中 `python_interpreter` 是一个通用计算工具，所有题目共享同一个 MCP Server 实例即可。代码任务则不同——每道题有独立的 public test cases，工具需要知道"当前题目的测试用例"才能运行。

两种可选方案：

| 方案 | 实现 | 适用场景 |
|------|------|---------|
| MCP Server + 状态注入 | 每次 rollout 前通过 API 设置当前题目的 test cases | 工具逻辑复杂、需要进程隔离 |
| **StructuredTool + 闭包** | 每次 rollout 新建一个绑定了 test cases 的工具实例 | 工具逻辑简单、无需进程隔离 ✅ |

本项目选择了方案 2，因为代码测试逻辑本身很轻量（`subprocess.run`），不需要常驻进程，闭包模式更简洁。

### 3.4 工具：execute_code

> 工具构建：[`functions/rollout/code_rollout.py` L121-157](functions/rollout/code_rollout.py#L121-L157)（`_build_execute_code_tool`）
> 工具绑定：[`functions/rollout/code_rollout.py` L264-274](functions/rollout/code_rollout.py#L264-L274)（`_build_tools`）
> 底层测试：[`functions/reward/code_testing.py`](functions/reward/code_testing.py)

#### 3.4.1 Per-Problem 动态绑定

核心设计模式：通过 Python 闭包将当前题目的 test cases 绑定到工具函数中。

```python
def _build_execute_code_tool(public_tests, fn_name):
    """每次 rollout 调用，创建一个绑定了当前题目 test cases 的工具。"""

    def _execute_code(code: str) -> str:
        extracted = extract_code(code)            # 提取 ```python``` 代码块
        actual_code = extracted if extracted else code
        return run_public_tests(actual_code, public_tests, fn_name=fn_name)

    return StructuredTool.from_function(
        func=_execute_code,
        name="execute_code",
        description="Run your Python solution against sample test cases...",
    )
```

在 `CodeRolloutProcessor._build_tools()` 中，从 `rollout_extra` 提取当前题目信息并创建工具：

```python
def _build_tools(self, rollout_extra):
    public_tests = _resolve_public_tests(rollout_extra)  # JSON 解析 public_test_cases
    fn_name = _resolve_fn_name(rollout_extra)             # 提取 func_name（functional 模式）
    tool = _build_execute_code_tool(public_tests, fn_name)
    return [tool]
```

#### 3.4.2 与 python_interpreter 的对比

| 对比维度 | python_interpreter (数学) | execute_code (代码) |
|---------|---------------------------|---------------------|
| 注册方式 | MCP Server (全局共享) | StructuredTool (per-problem 闭包) |
| 生命周期 | 进程级单例 | 每次 rollout 新建 |
| 执行内容 | 任意 Python 计算 | 对代码方案运行公开测试用例 |
| 输入 | `code: str`（任意脚本） | `code: str`（完整解题方案） |
| 输出格式 | stdout + stderr（原始文本） | 结构化 PASS/FAIL + diff 诊断 |
| 测试绑定 | 无 | 绑定当前题目的 public_test_cases |
| 执行模式 | 单一 | stdin / functional 两种 |

#### 3.4.3 代码提取：`extract_code()`

> 源码位置：[`functions/reward/code_testing.py` L35-69](functions/reward/code_testing.py#L35-L69)

模型通常在 ` ```python ` 代码块中输出代码，但也可能直接输出裸代码。提取逻辑优先取**最后一个** ` ```python ` 块（因为前面的可能是中间尝试）：

```python
def extract_code(response: str) -> Optional[str]:
    if "```python" in response:
        code = response.split("```python")[-1].split("```")[0].strip()
        return code or None
    parts = response.split("```")
    if len(parts) >= 3:
        code = parts[-2].strip()       # 回退：取最后一个任意语言代码块
        return code or None
    return None                         # 无代码块，原文透传给 run_public_tests
```

#### 3.4.4 测试执行：两种模式

> 源码位置：[`functions/reward/code_testing.py` L93](functions/reward/code_testing.py#L93)（`_run_stdin_test_detailed`），[L127](functions/reward/code_testing.py#L127)（`_run_functional_test_detailed`）

根据题目类型（由 `rollout_extra` 中是否有 `starter_code` / `func_name` 决定），工具自动选择执行模式：

**stdin 模式**（Codeforces / AtCoder 风格）：
```python
subprocess.run([sys.executable, "-c", code],
               input=test_input,       # 通过 stdin 注入测试输入
               capture_output=True, timeout=10)
# 比较 stdout 与 expected_output
```

**functional 模式**（LeetCode 风格）：
```python
# 自动生成 wrapper，实例化 Solution() 并调用目标函数
wrapper = f"""
import json, sys
from typing import *
{code}
_args = [{args_json}]
_sol = Solution()
_result = getattr(_sol, "{fn_name}")(*_args)
print(json.dumps(_result))
"""
subprocess.run([sys.executable, "-c", wrapper], ...)
# 比较 JSON 序列化结果
```

**输出比较的容忍度**（`_compare_output`）：
- 忽略行尾空格和空行
- 数值比较容忍 `1e-6` 误差（浮点数题目）

#### 3.4.5 输出格式化：智能截断

> 源码位置：[`functions/reward/code_testing.py` L361](functions/reward/code_testing.py#L361)（`format_test_results`），[L270](functions/reward/code_testing.py#L270)（`run_public_tests`）

工具返回给模型的测试结果经过截断控制，确保不会超出 rollout context 容量：

| 参数 | 值 | 说明 |
|------|-----|------|
| `_INPUT_DISPLAY_MAX` | 300 字符 | 测试输入最大展示长度 |
| `_OUTPUT_HEAD` | 300 字符 | FAIL case 的 actual/expected 展示头部 |
| `_OUTPUT_TAIL` | 100 字符 | FAIL case 的 actual/expected 展示尾部 |
| `_PASS_OUTPUT_HEAD` | 200 字符 | PASS case 展示（实际只输出一行状态） |

**设计考量**：
- **PASS 的 case 极简展示**：只输出 `"Test 1/5: PASS"`，不展示 input/output，节省 context
- **FAIL 的 case 诊断导向**：展示截断后的 input + diff summary + 截断后的 actual/expected
- **Diff summary**：当 actual 和 expected 长度不同时展示 `length mismatch`；长度相同时定位首个不同字符的 index，帮助模型快速定位 bug
- **整体 token 预算**：约 6 个 test case 的工具输出控制在 ~3KB ≈ 750 tokens，在 `max_length=20480` 下即使多轮 rollout 也不会溢出

示例输出：
```
Test 1/5: PASS
Test 2/5: PASS
--- Test 3/5: FAIL ---
Input:
5
1 2 3 4 5
Diff summary: length mismatch (actual=2 chars, expected=3 chars)
Your output:
14
Expected output:
15

Result: 2/5 tests passed.
```

> **设计原则**：对 actual/expected 输出设置硬截断上限，彻底解耦"测试复杂度"与"模型看到的 token 量"，
> 避免某些题目的超长 expected output 占满 model context。

#### 3.4.6 安全限制

| 限制 | 值 | 说明 |
|------|-----|------|
| 单 case 超时 | 10s | `subprocess.run(timeout=10)` |
| 可见 test cases | 仅 public | hidden tests 只在 reward 阶段使用 |
| 最大 case 数 | 20 | `_MAX_TEST_CASES = 20` |
| 执行方式 | `subprocess.run` | 独立进程，崩溃不影响主进程 |
| stderr 截断 | 500 字符 | 错误信息只保留末尾 500 字符 |

### 3.5 两种 System Prompt

> 源码位置：[`functions/rollout/code_rollout.py` L60-101](functions/rollout/code_rollout.py#L60-L101)（`SYSTEM_PROMPT_STDIN` / `SYSTEM_PROMPT_FUNCTIONAL`），[L259-262](functions/rollout/code_rollout.py#L259-L262)（`_get_system_prompt`）

根据题目是否有 `starter_code` 自动选择：

#### stdin 模式（无 starter_code）

```
You are an expert competitive programmer. You have an `execute_code` tool
that runs your Python solution against sample test cases and reports
pass/fail, actual vs expected output, and any errors.

Workflow:
1. Analyze the problem and identify edge cases.
2. Write an initial solution.
3. Call `execute_code` to test it — do NOT skip this step.
4. If any test fails, read the diff carefully, fix, and retest.
5. Once all sample tests pass, output your final solution in a
```python``` code block.

Rules:
- Read from stdin, print to stdout.
- Sample tests are a subset; handle all edge cases.
- You MUST call `execute_code` at least once before finalizing.
- To submit your final answer, write the complete solution in a
```python``` code block in your response.
```

#### functional 模式（有 starter_code）

```
You are an expert competitive programmer. You have an `execute_code` tool
that runs your Python solution against sample test cases and reports
pass/fail, actual vs expected output, and any errors.

Workflow:
1. Analyze the problem and identify edge cases.
2. Complete the required function/method inside a `Solution` class.
3. Call `execute_code` to test it — do NOT skip this step.
4. If any test fails, read the diff carefully, fix, and retest.
5. Once all sample tests pass, output your final solution in a
```python``` code block.

Rules:
- Include necessary imports (e.g. `from typing import *`).
- Sample tests are a subset; handle all edge cases.
- You MUST call `execute_code` at least once before finalizing.
- To submit your final answer, write the complete solution in a
```python``` code block in your response.
```

**System Prompt 设计要点**：
- 明确 5 步工作流：分析 → 编写 → 测试 → 修复 → 提交
- 强调 "MUST call execute_code at least once"——与 reward 中的格式奖励对齐
- 明确区分 stdin 和 functional 两种输入输出模式
- 提醒 "sample tests are a subset"——引导模型考虑边界情况

### 3.6 可观测性

> 源码位置：[`functions/rollout/code_rollout.py` L45-46](functions/rollout/code_rollout.py#L45-L46)（import），[L239](functions/rollout/code_rollout.py#L239)（`trace_client`），[L291](functions/rollout/code_rollout.py#L291)（`trace_tool`），[L366](functions/rollout/code_rollout.py#L366)（`@observe_processor`）

| 机制 | 说明 |
|------|------|
| `trace_client(llm)` | 自动追踪每次 LLM 调用 |
| `trace_tool(tools)` | 自动追踪 execute_code 每次调用 |
| `@observe_processor` | 追踪整个 rollout process() |
| `ENABLE_TRAJECTORY` | 记录完整多轮对话轨迹 |

### 3.7 Rollout Metrics 自定义追踪

> 源码位置：[`functions/rollout/code_rollout.py` L195-215](functions/rollout/code_rollout.py#L195-L215)（`_execute_tools` 计时），[L320-336](functions/rollout/code_rollout.py#L320-L336)（`rollout_metrics` 汇总）

与数学任务相同，代码任务也通过 `rollout_metrics` 上报工具性能指标。实现模式一致：

```python
start_time = time.time()
tool_call_durations = []

# _execute_tools 中记录每次 execute_code 调用耗时
t0 = time.time()
result = await tool_node.ainvoke(state, config)
tool_call_durations.append(time.time() - t0)

# rollout 结束后汇总
rollout_metrics={
    "latency": time.time() - start_time,
    "tool_call_count": len(tool_call_durations),
    "tool_call_total_time": sum(durations),
    "tool_call_avg_time": avg(durations),
}
```

**自定义扩展**：只需在 `rollout_metrics` 字典中添加新的 key-value，框架会自动聚合为 `trace _ rollout_metrics _ {key} _ avg/sum` 等指标。例如代码任务可以额外追踪 `"public_pass_rate"` 或 `"code_length"` 等业务指标。

| 指标 | 说明 | 训练面板对应 |
|------|------|------------|
| `latency` | 单次 rollout 端到端耗时（秒） | `trace _ rollout_metrics _ latency _ avg` |
| `tool_call_count` | execute_code 调用次数 | `trace _ rollout_metrics _ tool_call_count _ avg` |
| `tool_call_total_time` | 所有代码测试的累计耗时 | `trace _ rollout_metrics _ tool_call_total_time _ avg` |
| `tool_call_avg_time` | 单次代码测试平均耗时 | `trace _ rollout_metrics _ tool_call_avg_time _ avg` |

---

## 4. 奖励设计配置与跟踪

> 实现文件：[`functions/reward/code_reward.py`](functions/reward/code_reward.py)（`CodeRewardProcessor` [L31](functions/reward/code_reward.py#L31)）
> 底层测试：[`functions/reward/code_testing.py`](functions/reward/code_testing.py)（`run_tests` [L175](functions/reward/code_testing.py#L175)，`extract_code` [L35](functions/reward/code_testing.py#L35)）

### 4.1 奖励公式

$$
R_{\text{total}} = 0.1 \times R_{\text{format}} + 0.9 \times R_{\text{accuracy}}
$$

**注意**：代码任务的 format 权重（0.1）比数学任务（0.2）更低，accuracy 权重（0.9）更高。这是因为代码正确性是核心目标，格式要求（只需包含 ` ```python``` ` 代码块）相对简单。

### 4.2 子奖励详解

#### Format 子奖励 (权重 0.1)

- **检测方式**：调用 `extract_code(content)` 检测是否包含 ` ```python``` ` 代码块
- **评分**：
  - 提取到代码块 → **+1.0**
  - 未找到 → **-1.0**

#### Accuracy 子奖励 (权重 0.9)

- **评分方式**：**按比例评分**（proportional score），而非 0/1 二值
  - `score = passed_tests / total_tests`
  - 例如通过 7/10 个测试用例 → accuracy = 0.7
- **测试范围**：使用全量 test_cases（包含 hidden tests），而非 rollout 时的 public tests
- **额外指标**：
  - `pass@1`：严格全对率（passed == total → 1.0，否则 0.0）
  - `pass_rate`：等同于 proportional score

**代码提取策略**：反向扫描所有 assistant 消息，找到最后一个包含 ` ```python``` ` 代码块的消息，提取其中的代码。这是因为在多轮交互中，最终代码可能不在最后一条消息中。

### 4.3 代码测试执行

> 源码位置：[`functions/reward/code_testing.py` L175-217](functions/reward/code_testing.py#L175-L217)（`run_tests`），[L219-267](functions/reward/code_testing.py#L219-L267)（`normalise_test_cases`）

底层由 [`functions/reward/code_testing.py`](functions/reward/code_testing.py) 提供：

```mermaid
flowchart TB
    Code["模型生成代码"] --> Extract["extract_code()<br/>提取 python 代码块"]
    Extract -->|"提取成功"| Normalize["normalise_test_cases()<br/>统一 test case 格式"]
    Extract -->|"提取失败"| Zero["accuracy = 0.0"]
    Normalize --> RunTests["run_tests()<br/>逐个执行 test case"]
    RunTests --> Score["score = passed / total"]
    Score --> Metrics["reward_metrics:<br/>pass_rate, pass@1,<br/>dataset/xxx, difficulty/xxx"]
```

支持两种执行模式：
- **stdin 模式**：通过 subprocess 执行，stdin 输入 → stdout 输出
- **functional 模式**：构造 `Solution.method()` 调用，比较返回值

每个 test case 有 10 秒超时限制。

### 4.4 Reward Metrics 追踪

| 指标 | 说明 |
|------|------|
| `format` | 格式分（-1 或 1） |
| `accuracy` | 比例正确率（0.0 ~ 1.0） |
| `pass_rate` | 等同于 accuracy |
| `pass@1` | 严格全对率（0 或 1） |
| `total` | 加权总分 |
| `dataset/{name}` | 按数据源分类的 pass_rate |
| `difficulty/{level}` | 按难度分类的 pass_rate |

### 4.5 Rollout 工具 vs Reward 评估的差异

| 维度 | Rollout 工具 (execute_code) | Reward 评估 |
|------|---------------------------|-------------|
| 测试用例 | 仅 public (sample) tests | **全量** test_cases (含 hidden) |
| 目的 | 帮助模型调试代码 | 评估最终提交的正确性 |
| 反馈格式 | 结构化 PASS/FAIL 文本 | 数值 score (0.0 ~ 1.0) |
| 超时设置 | 每 case 10s | 每 case 10s |

这个设计确保了模型在 rollout 时只能看到 sample tests（模拟真实竞赛场景），而 reward 使用全量测试做更全面的评估。

---

## 5. 部署推理验证

部署验证需先将训练产出的模型在平台部署为推理服务，获取模型 ID 后即可在 LiveCodeBench v6 测试集上进行 pass@1 评估。具体验证方法可参考 U-MATH Demo 中的推理验证流程。

---

## 6. 最终效果

> **参考实验**：任务 ID `ft-202606151516-e36e`（该实验实际使用 2 epochs，但 Epoch 2 验证集 reward 出现明显震荡，因此推荐配置为 `n_epochs=1`）
> 以下展示 **Epoch 1（steps 1-16）** 的效果。

### 6.1 关键指标汇总

| 指标 | 初始值 (step 1) | Epoch 1 结束 (step 16) | 变化趋势 |
|------|-----------------|----------------------|---------|
| 训练 reward (total) | 0.150 | 0.821 | ↑ 持续上升 |
| 训练 accuracy (pass_rate) | 0.226 | 0.824 | ↑ 持续上升 |
| 训练 pass@1 | 0.222 | 0.719 | ↑ 持续上升 |
| 训练 format score | -0.539 | 0.790 | ↑ 从负分到正分 |
| 验证 reward mean@1 | -0.027 (step 0) | 0.568 (step 16) | ↑ 峰值 0.610 (step 12) |
| execute_code 调用次数 | 0.75 | 2.48 | ↑ +230%，工具使用频率显著增加 |
| LLM 推理次数 | 1.74 | 3.48 | ↑ 多轮交互能力增强 |
| KL loss | 0.00034 | 0.00651 | ↑ 增长平缓，策略未出现大幅偏移 |
| Entropy | 0.425 | 0.405 | ↓ 输出分布更集中 |

### 6.2 训练曲线

#### 训练 Reward

![训练 Reward 曲线](charts/train_reward.png)

训练 reward（total）和 accuracy 在 epoch 1 内从 ~0.15 持续上升至 ~0.85。Format score 从初始 -0.54（多数输出不含 ` ```python``` ` 代码块）快速上升至 ~0.79，说明模型学会了按格式要求输出代码。三条指标曲线走势一致，训练过程稳定。

#### 验证 Reward

![验证 Reward 曲线](charts/val_reward.png)

验证集 reward 从初始 -0.03 上升至 0.61（step 12 峰值），step 16 时为 0.57。验证集的绝对值低于训练集（0.57 vs 0.82），符合预期——验证集是 LiveCodeBench 的不同题目，且 greedy 解码（temperature=0）降低了多样性。

### 6.3 工具使用趋势

![工具使用趋势](charts/tool_calls.png)

execute_code 调用次数从 0.75 增至 2.48（+230%），LLM 推理次数从 1.74 增至 3.48，模型学会了通过多轮测试和修复来迭代改进代码。

### 6.4 训练稳定性

![Learning Rate & KL Loss](charts/lr_kl.png)

Learning rate 在 warmup 3 步后稳定在 7e-7（constant scheduler）。KL loss 增长平缓（0.00034 → 0.00651），表明 `kl_loss_coef=0.01` 有效约束了策略偏移。

#### Rollout Latency

![Rollout Latency](charts/latency.png)

单次 rollout 延迟在 108-150 秒之间波动，随训练进行整体趋于下降——模型输出更精炼，减少了冗余对话。

### 6.5 典型案例

以下案例来自验证集（LiveCodeBench v6），展示同一道题在 step 6（训练初期）与 step 16（epoch 1 结束）的完整轨迹差异。

#### 案例 A：算法优化提升（Sample `f63f150e`，reward 0.91 → 1.00）

> **改进维度**：算法效率——暴力 O(N × max_val) → 筛法 + √n 因子分解
>
> **题目**：给定长度为 N 的序列 A 和正整数 K，对每个 A[i]，从 A 中选出包含 A[i] 的 K 个元素，使 GCD 最大。

<details>
<summary><b>▶ Step 6 轨迹（reward = 0.91, tools = 1, turns = 2）— 暴力解法，hidden tests 部分超时 ⚠️</b></summary>

```
┌─ [SYSTEM] ────────────────────────────────────────────────────────────────────
│ You are an expert competitive programmer. You have an `execute_code` tool
│ that runs your Python solution against sample test cases and reports
│ pass/fail, actual vs expected output, and any errors.
│
│ Workflow:
│ 1. Analyze the problem and identify edge cases.
│ 2. Write an initial solution.
│ 3. Call `execute_code` to test it — do NOT skip this step.
│ 4. If any test fails, read the diff carefully, fix, and retest.
│ 5. Once all sample tests pass, output your final solution in a ```python```
│    code block.
│
│ Rules:
│ - Read from stdin, print to stdout.
│ - Sample tests are a subset; handle all edge cases.
│ - You MUST call `execute_code` at least once before finalizing.
│ - To submit your final answer, write the complete solution in a ```python```
│   code block in your response.
└───────────────────────────────────────────────────────────────────────────────

┌─ [USER] ──────────────────────────────────────────────────────────────────────
│ You are given a sequence A = (A_1, A_2, ..., A_N) of length N and a positive
│ integer K (at most N).
│ For each i = 1, 2, ..., N, solve the following problem:
│ - When you choose K elements from A that include A_i, find the maximum
│   possible GCD (greatest common divisor) of those chosen elements.
│
│ Input: N K / A_1 A_2 ... A_N
│ Output: N lines, the j-th line is the answer for i=j.
│ Constraints: 1 ≤ K ≤ N ≤ 1.2×10^6, 1 ≤ A_i ≤ 10^6
│
│ Sample Input 1:  5 2 / 3 4 6 7 12
│ Sample Output 1: 3 4 6 1 6
│ Sample Input 2:  3 3 / 6 10 15
│ Sample Output 2: 1 1 1
│ Sample Input 3:  10 3 / 414003 854320 485570 52740 833292 625990 909680
│                  885153 435420 221663
│ Sample Output 3: 59 590 590 879 879 590 20 879 590 59
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 🔧 execute_code (第 1 次) ─────────────────────────────────────
│ import sys
│ from collections import defaultdict
│
│ def solve():
│     input_data = sys.stdin.read().split()
│     idx = 0
│     N = int(input_data[idx]); idx += 1
│     K = int(input_data[idx]); idx += 1
│     A = [int(input_data[idx + i]) for i in range(N)]; idx += N
│
│     max_val = max(A)
│
│     # 暴力：对每个元素遍历其所有因子 → O(N × max_val)
│     cnt = [0] * (max_val + 1)
│     for x in A:
│         for g in range(1, x + 1):
│             if x % g == 0:
│                 cnt[g] += 1
│
│     valid_g = [cnt[g] >= K for g in range(max_val + 1)]
│
│     # 预计算所有数的因子列表 → O(max_val × log(max_val))
│     divisors = [[] for _ in range(max_val + 1)]
│     for g in range(1, max_val + 1):
│         for multiple in range(g, max_val + 1, g):
│             divisors[multiple].append(g)
│
│     results = []
│     for x in A:
│         best_g = 1
│         for g in divisors[x]:
│             if valid_g[g]:
│                 best_g = max(best_g, g)
│         results.append(best_g)
│
│     for r in results:
│         print(r)
│
│ solve()
└───────────────────────────────────────────────────────────────────────────────

┌─ [TOOL] → (执行结果) ─────────────────────────────────────────────────────────
│ Test 1/3: PASS
│ Test 2/3: PASS
│ Test 3/3: PASS
│
│ Result: 3/3 tests passed.
│ All tests passed. If you are confident in your solution, output the final
│ code in a ```python``` code block.
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 📝 最终提交 ──────────────────────────────────────────────────
│ All tests passed. Here's the final solution:
│
│ ```python
│ import sys
│ from collections import defaultdict
│
│ def solve():
│     input_data = sys.stdin.read().split()
│     idx = 0
│     N = int(input_data[idx]); idx += 1
│     K = int(input_data[idx]); idx += 1
│     A = [int(input_data[idx + i]) for i in range(N)]; idx += N
│     max_val = max(A)
│     cnt = [0] * (max_val + 1)
│     for x in A:
│         for g in range(1, x + 1):        ← O(max_val) per element
│             if x % g == 0:
│                 cnt[g] += 1
│     valid_g = [cnt[g] >= K for g in range(max_val + 1)]
│     divisors = [[] for _ in range(max_val + 1)]
│     for g in range(1, max_val + 1):
│         for multiple in range(g, max_val + 1, g):
│             divisors[multiple].append(g)
│     results = []
│     for x in A:
│         best_g = 1
│         for g in divisors[x]:
│             if valid_g[g]:
│                 best_g = max(best_g, g)
│         results.append(best_g)
│     for r in results:
│         print(r)
│ solve()
│ ```
└───────────────────────────────────────────────────────────────────────────────

  ⚠️ public tests 全部通过，但暴力因子枚举 O(N × max_val) 在 hidden tests
  大数据量下超时，导致 reward = 0.91 < 1.0。
```

</details>

<details open>
<summary><b>▶ Step 16 轨迹（reward = 1.00, tools = 1, turns = 2）— 筛法 + √n 分解 ✅</b></summary>

```
┌─ [SYSTEM] / [USER] ── 同上 ──────────────────────────────────────────────────
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 🔧 execute_code (第 1 次) ─────────────────────────────────────
│ import sys
│
│ def solve():
│     input_data = sys.stdin.read().split()
│     if not input_data:
│         return
│     N = int(input_data[0])
│     K = int(input_data[1])
│     A = list(map(int, input_data[2:2+N]))
│
│     max_val = max(A)
│
│     # 筛法：先统计频率
│     cnt = [0] * (max_val + 1)
│     for x in A:
│         cnt[x] += 1
│
│     # 调和级数方式统计可整除计数 → O(max_val × log(max_val))
│     divisible_count = [0] * (max_val + 1)
│     for g in range(1, max_val + 1):
│         for multiple in range(g, max_val + 1, g):
│             divisible_count[g] += cnt[multiple]
│
│     # 对每个 x，只枚举 √x 个因子
│     result = []
│     for x in A:
│         best_g = 1
│         for g in range(1, int(x**0.5) + 1):
│             if x % g == 0:
│                 if divisible_count[g] >= K:
│                     best_g = max(best_g, g)
│                 if g * g != x and divisible_count[x // g] >= K:
│                     best_g = max(best_g, x // g)
│         result.append(best_g)
│
│     for r in result:
│         print(r)
│
│ if __name__ == "__main__":
│     solve()
└───────────────────────────────────────────────────────────────────────────────

┌─ [TOOL] → (执行结果) ─────────────────────────────────────────────────────────
│ Test 1/3: PASS
│ Test 2/3: PASS
│ Test 3/3: PASS
│
│ Result: 3/3 tests passed.
│ All tests passed. If you are confident in your solution, output the final
│ code in a ```python``` code block.
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 📝 最终提交 ──────────────────────────────────────────────────
│ The solution passed all test cases. Here's the final code:
│
│ ```python
│ import sys
│
│ def solve():
│     input_data = sys.stdin.read().split()
│     if not input_data:
│         return
│     N = int(input_data[0])
│     K = int(input_data[1])
│     A = list(map(int, input_data[2:2+N]))
│     max_val = max(A)
│     cnt = [0] * (max_val + 1)
│     for x in A:
│         cnt[x] += 1
│     divisible_count = [0] * (max_val + 1)
│     for g in range(1, max_val + 1):
│         for multiple in range(g, max_val + 1, g):
│             divisible_count[g] += cnt[multiple]
│     result = []
│     for x in A:
│         best_g = 1
│         for g in range(1, int(x**0.5) + 1):
│             if x % g == 0:
│                 if divisible_count[g] >= K:
│                     best_g = max(best_g, g)
│                 if g * g != x and divisible_count[x // g] >= K:
│                     best_g = max(best_g, x // g)
│         result.append(best_g)
│     for r in result:
│         print(r)
│ if __name__ == "__main__":
│     solve()
│ ```
└───────────────────────────────────────────────────────────────────────────────
```

</details>

> **关键对比**：两次都一轮通过 public tests，但 Step 6 使用暴力因子枚举 O(N × max_val)，hidden tests 大数据下超时；Step 16 改用"频率数组 + 筛法 + √n 分解"，将复杂度降至 O(max_val × log(max_val) + N × √max_val)，通过全部测试。

---

#### 案例 B：工具使用效率提升（Sample `7ec8266d`，reward 1.00 → 1.00，tools 2 → 1）

> **改进维度**：工具使用规范——学会不在提交代码中混入多余 print/assert
>
> **题目**：将数组中偶数替换为 0、奇数替换为 1，然后排序返回。

<details>
<summary><b>▶ Step 6 轨迹（reward = 1.00, tools = 2, turns = 3）— 多余输出导致首次失败 ⚠️</b></summary>

```
┌─ [SYSTEM] ────────────────────────────────────────────────────────────────────
│ You are an expert competitive programmer. You have an `execute_code` tool
│ that runs your Python solution against sample test cases and reports
│ pass/fail, actual vs expected output, and any errors.
│
│ Workflow:
│ 1. Analyze the problem and identify edge cases.
│ 2. Complete the required function/method inside a `Solution` class.
│ 3. Call `execute_code` to test it — do NOT skip this step.
│ 4. If any test fails, read the diff carefully, fix, and retest.
│ 5. Once all sample tests pass, output your final solution in a ```python```
│    code block.
│
│ Rules:
│ - Include necessary imports (e.g. `from typing import *`).
│ - Sample tests are a subset; handle all edge cases.
│ - You MUST call `execute_code` at least once before finalizing.
│ - To submit your final answer, write the complete solution in a ```python```
│   code block in your response.
└───────────────────────────────────────────────────────────────────────────────

┌─ [USER] ──────────────────────────────────────────────────────────────────────
│ You are given an integer array nums. Transform nums by performing the
│ following operations in the exact order specified:
│ - Replace each even number with 0.
│ - Replace each odd number with 1.
│ - Sort the modified array in non-decreasing order.
│ Return the resulting array after performing these operations.
│
│ Example 1: Input: nums = [4,3,2,1] → Output: [0,0,1,1]
│ Example 2: Input: nums = [1,5,1,4,2] → Output: [0,0,1,1,1]
│ Constraints: 1 <= nums.length <= 100, 1 <= nums[i] <= 1000
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 🔧 execute_code (第 1 次) ─────────────────────────────────────
│ from typing import List
│
│ class Solution:
│     def transformArray(self, nums: List[int]) -> List[int]:
│         modified = [0 if num % 2 == 0 else 1 for num in nums]
│         modified.sort()
│         return modified
│
│ # ↓↓↓ 多余的测试代码 ↓↓↓
│ solution = Solution()
│
│ nums1 = [4, 3, 2, 1]
│ result1 = solution.transformArray(nums1)
│ print(f"Example 1: {result1}")                ← 多余输出
│ assert result1 == [0, 0, 1, 1]               ← 干扰判题
│
│ nums2 = [1, 5, 1, 4, 2]
│ result2 = solution.transformArray(nums2)
│ print(f"Example 2: {result2}")
│ assert result2 == [0, 0, 1, 1, 1]
│
│ nums3 = [2]
│ result3 = solution.transformArray(nums3)
│ print(f"Single even: {result3}")
│ assert result3 == [0]
│
│ nums4 = [3]
│ result4 = solution.transformArray(nums4)
│ print(f"Single odd: {result4}")
│ assert result4 == [1]
│
│ nums5 = [2, 4, 6]
│ result5 = solution.transformArray(nums5)
│ print(f"All evens: {result5}")
│ assert result5 == [0, 0, 0]
│
│ nums6 = [1, 3, 5]
│ result6 = solution.transformArray(nums6)
│ print(f"All odds: {result6}")
│ assert result6 == [1, 1, 1]
│
│ print("All tests passed!")
└───────────────────────────────────────────────────────────────────────────────

┌─ [TOOL] → (执行结果) ─────────────────────────────────────────────────────────
│ --- Test 1/2: FAIL ---
│ Input: [4, 3, 2, 1]
│ Error: Output is not valid JSON
│ Diff summary: length mismatch (actual=155 chars, expected=12 chars)
│ Your output:
│ Example 1: [0, 0, 1, 1]
│ Example 2: [0, 0, 1, 1, 1]
│ Single even: [0]
│ Single odd: [1]
│ All evens: [0, 0, 0]
│ All odds: [1, 1, 1]
│ All tests passed!
│ [0, 0, 1, 1]
│ Expected output: [0, 0, 1, 1]
│
│ --- Test 2/2: FAIL ---
│ Input: [1, 5, 1, 4, 2]
│ Error: Output is not valid JSON
│ Diff summary: length mismatch (actual=158 chars, expected=15 chars)
│
│ Result: 0/2 tests passed.
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 🔧 execute_code (第 2 次) ─────────────────────────────────────
│ from typing import List
│
│ class Solution:
│     def transformArray(self, nums: List[int]) -> List[int]:
│         modified = [0 if num % 2 == 0 else 1 for num in nums]
│         modified.sort()
│         return modified
└───────────────────────────────────────────────────────────────────────────────

┌─ [TOOL] → (执行结果) ─────────────────────────────────────────────────────────
│ Test 1/2: PASS
│ Test 2/2: PASS
│
│ Result: 2/2 tests passed.
│ All tests passed. If you are confident in your solution, output the final
│ code in a ```python``` code block.
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 📝 最终提交 ──────────────────────────────────────────────────
│ ```python
│ from typing import List
│
│ class Solution:
│     def transformArray(self, nums: List[int]) -> List[int]:
│         modified = [0 if num % 2 == 0 else 1 for num in nums]
│         modified.sort()
│         return modified
│ ```
└───────────────────────────────────────────────────────────────────────────────
```

</details>

<details open>
<summary><b>▶ Step 16 轨迹（reward = 1.00, tools = 1, turns = 2）— 一次通过 ✅</b></summary>

```
┌─ [SYSTEM] / [USER] ── 同上 ──────────────────────────────────────────────────
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 🔧 execute_code (第 1 次) ─────────────────────────────────────
│ from typing import List
│
│ class Solution:
│     def transformArray(self, nums: List[int]) -> List[int]:
│         modified = [0 if num % 2 == 0 else 1 for num in nums]
│         modified.sort()
│         return modified
└───────────────────────────────────────────────────────────────────────────────

┌─ [TOOL] → (执行结果) ─────────────────────────────────────────────────────────
│ Test 1/2: PASS
│ Test 2/2: PASS
│
│ Result: 2/2 tests passed.
│ All tests passed. If you are confident in your solution, output the final
│ code in a ```python``` code block.
└───────────────────────────────────────────────────────────────────────────────

┌─ [ASSISTANT] → 📝 最终提交 ──────────────────────────────────────────────────
│ The solution passed all test cases. Here's the final code:
│
│ ```python
│ from typing import List
│
│ class Solution:
│     def transformArray(self, nums: List[int]) -> List[int]:
│         modified = [0 if num % 2 == 0 else 1 for num in nums]
│         modified.sort()
│         return modified
│ ```
└───────────────────────────────────────────────────────────────────────────────
```

</details>

> **关键对比**：Step 6 在代码中混入了 print/assert 测试语句，导致输出污染被判 0/2 FAIL，不得不重新提交纯净代码（2 次工具调用）。Step 16 直接提交纯函数代码，一次通过（1 次工具调用）。训练使模型学会了 `execute_code` 的使用规范。

---

#### 案例总结

| 案例 | Step 6 问题 | Step 16 改进 | 改进维度 |
|------|-----------|-----------|---------|
| A. GCD 最大化 | 暴力 O(N×max_val)，hidden tests 超时 | 筛法 + √n 分解，全部通过 | **算法效率** |
| B. 数组变换 | 混入 print/assert 导致首次失败 | 纯净代码一次通过 | **工具使用规范** |

---

## 附录 A：三个 Demo 配置对比

| 维度 | MATH | U-MATH | Code |
|------|------|--------|------|
| **数据规模** | ~3500 (lv4-5) | 700 | 4300 |
| **数据来源** | Hendrycks MATH | U-MATH | TACO (随机采样) |
| **model** | qwen3.5-9b | qwen3.5-9b | qwen3.5-9b |
| **n_epochs** | 1 | 3 | 1 |
| **learning_rate** | 1e-6 | 1e-6 | 7e-7 |
| **lr_scheduler_type** | linear | cosine | constant |
| **kl_loss_coef** | 默认 | 默认 | 0.01 |
| **batch_size** | 128 | 128 | 256 |
| **max_prompt_length** | 8192 | 8192 | 12288 |
| **max_length** | 16384 | 16384 | 20480 |
| **Rollout 工具** | python_interpreter (MCP) | 同左 | execute_code (StructuredTool) |
| **Reward format 权重** | 0.2 | 0.2 | 0.1 |
| **Reward accuracy 权重** | 0.8 | 0.8 | 0.9 |
| **Accuracy 验证方式** | SymPy 精确匹配 | LLM judge | 代码执行 pass rate |

## 附录 B：目录结构与文件放置（重要）

> 本 demo 的 `extra_demos/code/` 目录已整理为**可直接运行的完整 demo**。目录内已包含：
> - `functions/` 包（含 `__init__.py`），与 `submit_job.py` 中的 `classpath` 对应；
> - `requirements.txt`、`start.sh` 等运行所需文件；
> - `data/` 与 `scripts/`，提交脚本会检查同级 `data/` 并按需运行 `scripts/` 下的数据脚本生成训练/验证数据（也可按 B.2 手动预下载）。

提交任务时，SDK 会根据 `FunctionComponentModel` 中的 `zipdir` 把 `extra_demos/code/` 整个打包上传，远程再按 `classpath` 加载模块。因此可直接在 `extra_demos/code/` 目录下运行。

### B.1 文件放置映射表

| demo 内文件路径 | 说明 |
|--------------|------------------------------------|
| `extra_demos/code/submit_job.py` | 任务提交入口 |
| `extra_demos/code/functions/rollout/code_rollout.py` | rollout 处理器 |
| `extra_demos/code/functions/reward/code_reward.py` | reward 处理器 |
| `extra_demos/code/functions/reward/code_testing.py` | 代码测试工具 |
| `extra_demos/code/scripts/download_livecodebench.py` | 验证数据下载脚本 |
| `extra_demos/code/scripts/sample_ablation_subsets.py` | 训练数据生成脚本 |

> 关键约束：rollout / reward 代码**必须**放进 `functions/` 包内（与 classpath 对应），且 `functions/__init__.py` 必须存在。
> 数据文件需用 B.2 的脚本生成：`submit_job.py` 用的是 `./data/ablation_taco.jsonl`（训练）和 `./data/livecodebench_v6_test.jsonl`（验证）。

### B.2 数据准备（可选）

`submit_job.py` 会自动检查同级 `data/` 并按需运行 `scripts/` 下的脚本生成数据。如需手动预下载，在 `extra_demos/code/` 下执行：

```bash
mkdir -p data
python scripts/sample_ablation_subsets.py  # 生成 data/ablation_taco.jsonl（训练集）
python scripts/download_livecodebench.py   # 生成 data/livecodebench_v6_test.jsonl（验证集）
```

### B.3 demo 目录已具备的运行环境文件

| 文件 | 用途 |
|------|------|
| `../../dashscope-1.25.23-py3-none-any.whl` | SDK 本地包（也可改用 PyPI 版本） |
| `./requirements.txt` | 函数组件依赖（langchain、mcp 等） |
| `./start.sh` | 本地启动函数组件服务示例 |

> Code 的 accuracy 由**代码执行 pass rate** 判定，不调用 LLM judge，因此**不需要** `llm_api_config.json`。

`extra_demos/code/` 目录结构如下：

```
extra_demos/code/
├── submit_job.py
├── start.sh
├── requirements.txt
├── functions/
│   ├── __init__.py
│   ├── rollout/
│   │   └── code_rollout.py
│   └── reward/
│       ├── code_reward.py
│       └── code_testing.py
├── data/
│   ├── ablation_taco.jsonl          # 由 sample_ablation_subsets.py 生成
│   └── livecodebench_v6_test.jsonl  # 由 download_livecodebench.py 生成
└── scripts/
    ├── sample_ablation_subsets.py
    └── download_livecodebench.py
```

随后即可按 **附录 C** 的清单执行 `python submit_job.py`。

---

## 附录 C：快速上手清单

> 前提：在 `extra_demos/code/` 目录下执行以下命令。

1. **安装依赖**：`pip install ../../dashscope-1.25.23-py3-none-any.whl && pip install -r requirements.txt`
2. **设置 API Key**：`export DASHSCOPE_API_KEY=sk-xxx`
3. **准备数据（可选）**：`submit_job.py` 会在提交前自动检查 `data/ablation_taco.jsonl` 和 `data/livecodebench_v6_test.jsonl`，缺失时自动运行对应脚本；如需提前准备，可手动执行：
   ```bash
   python scripts/sample_ablation_subsets.py  # 仅下载并采样生成 data/ablation_taco.jsonl
   python scripts/download_livecodebench.py   # 下载并生成 LiveCodeBench v6 2501-2505 验证集
   ```
4. **提交训练**：`python submit_job.py`
5. **查看指标**：在训练监控面板查看 reward/pass@1 曲线、工具使用情况
6. **部署验证**：将训练模型部署为推理服务，获取模型 ID 后在 LiveCodeBench 测试集上评估 pass@1
