# 数据工程 recipe

`areal.v2.data_engineering` 提供有预算限制的前台数据任务：执行
recipe、清洗与去重、公开任务校准、筛选、保存来源记录并发布可校验的数据集。研究控制器选择目标和 recipe；领域 recipe
实现生成、清洗、评分、筛选和训练视图投影。这是面向任务数据的默认流程；任务项目可以按需组合多个作业或采用其他数据流程。

## 运行 TIR 示例

在仓库根目录、安装 AReaL 常规 Python 依赖后运行：

```bash
python3 -m areal.v2.data_engineering \
  --config examples/data_engineering/tir/data_generation.yaml \
  --output /tmp/tir-data-job \
  --export-hf /tmp/tir-training-data
```

首次运行时两个输出目录必须尚不存在；恢复数据作业使用 `--resume` 和原输出目录。该配置无需模型、GPU 或推理服务，直接生成带精确答案的数学任务；导入 AReaL
包仍需要常规 Python 依赖。stdout 最后一行返回 `dataset_id`、`dataset_path`，以及按需生成的
`training_dataset_path`。

TIR 包含八个数学题型、每个题型两种结构。训练视图包含 `task_id`、`messages` 和精确的 boxed `answer`。这是任务数据；训练
rollout、tokenization 和 RL/SFT/OPD 张量由下游处理。

## Python API 与执行协议

```python
import asyncio
from areal.v2.data_engineering import DataJobConfig, run_data_job

config = DataJobConfig(
    recipe="examples.data_engineering.tir.data_recipe:create_recipe",
    target_count=64,
    purpose="joint",
    seed=42,
    params={"candidate_count": 128},
)
snapshot = asyncio.run(run_data_job(config, "/tmp/tir-python-job"))
rows = snapshot.training_rows()
```

在已有异步控制器中使用 `await run_data_job(...)`。工厂采用可导入的 `module:callable`，执行器通过
`factory(**params)` 创建 `DataRecipe`。相关 Python 协议从 `areal.v2.data_engineering` 导出。

| 类型或方法                         | 约定                                                                                             |
| ---------------------------------- | ------------------------------------------------------------------------------------------------ |
| `PublicTask`                       | 稳定 `task_id`、领域 `input` 和公开 `metadata`；公开字段中不得放入答案                           |
| `DataSample`                       | `PublicTask` 与私有 `reference`；清洗后保留的候选须可 JSON 序列化                                |
| `generate(context)`                | 生成一批候选；context 提供 seed、批次编号、目标数量、用途和输入文件路径                          |
| `clean(sample)`                    | 返回清洗后的候选，或用 `None` 排除候选                                                           |
| `score(sample, response)`          | 使用私有参考答案，返回 `[0, 1]` 内的有限分数                                                     |
| `select(samples, scores, context)` | 返回恰好 `target_count` 个已有且不同的任务 ID；候选不足时抛出 `InsufficientDataError` 请求下一批 |
| `training_row(sample)`             | 将一个选中任务投影为一条领域训练数据                                                             |
| `source_files()`                   | 可选方法，声明影响结果的辅助代码和外部输入文件                                                   |

执行器按生成 → 清洗 → 去重与排除 → 校准 → 筛选运行。各批清洗后的候选及评分累计进入候选池。仅 `InsufficientDataError`
触发下一批；其他错误在配置的瞬态重试耗尽后令任务失败。`max_batches` 和 `max_candidates` 限制总预算，领域参数保留在 `params`
中。recipe 也可以通过 `context.inputs` 读取原始数据，执行已有数据的清洗与筛选。

输入和排除项均为文件。模块入口相对配置文件解析 `inputs`、`exclude` 中的路径；插件 `params`
中的路径由插件定义，当前校准适配器相对进程工作目录解析这些路径。

`purpose` 可为 `harness`、`model`、`joint`。它进入来源记录和 recipe context；由 recipe
决定如何影响策略。执行器不会自动归因模型或 Harness 的失败。独立 dev/test 数据以及采用哪个模型或 Harness，由上层控制器管理。

相同任务 ID 必须对应相同清洗内容；重复公开输入若有冲突的参考答案则报错。通用层按精确 JSON 输入去重，语义去重和领域规范化由 `clean()`
完成。排除文件采用包含顶层 `task_id` 的 JSONL：可传入已有数据集的 `public.jsonl`，或先前任务的
`work/batch-N-public.jsonl`，排除已选择或已校准的候选。

## 并行执行与恢复

在数据任务配置中加入：

```yaml
execution:
  max_workers: 4
  batch_size: 256
  backend: process
  max_retries: 2
  retry_delay: 1.0
```

`run_data_job` 在专用协调线程中创建独立事件循环。全局协调、来源摘要、SQLite 和 snapshot I/O
均由作业线程执行，调用方事件循环可以继续响应其他任务。校准器由工厂在作业事件循环中创建， 不应复用绑定到调用方事件循环的异步资源。SQLite
连接在协调线程中创建并由同一线程使用。

取消调用方 task 会通知作业并取消待执行的异步校准；返回取消结果前会等待 worker 退出、数据库关闭和最终状态落盘。
重复取消不会中断清理。同步插件或存储调用无法被强制打断，需等待当前调用返回，再在下一阶段或数据块之前生效。 已经原子发布的 snapshot
会保留，恢复时重新校验；这个机制不能终止永不返回的插件。

默认一个 worker、每块 256 条、`thread` 后端、不自动重试。`thread` 适合阻塞 I/O；`process` 通过 spawn 启动独立 Python
进程，并行执行 CPU 生成、清洗、评分、排名和训练视图投影。进程模式要求工厂可导入、原始候选可 pickle。 Python 脚本中的 `asyncio.run(...)`
应放在 `if __name__ == "__main__":` 中。 全局去重与最终配额合并由协调线程执行；异步校准拆成最多 `max_workers` 个并发调用，每次最多
`batch_size` 个公开任务，重复执行次数仍由校准协议控制。

```bash
python3 -m areal.v2.data_engineering \
  --config examples/data_engineering/tir/data_generation.yaml \
  --output /tmp/tir-data-job --resume
```

Python 调用使用 `await run_data_job(config, output, resume=True)`。恢复要求配置、声明的输入和排除文件、
recipe/校准源码及执行器代码保持一致；身份发生变化会在执行前拒绝恢复。成功作业恢复时返回已有的、重新校验过的 snapshot。HF 导出仍要求新目录；已经导出时省略
`--export-hf`。

`work/checkpoints.sqlite3` 原子提交每个清洗、校准、投影块。候选内容、去重索引、校准结果和排名记录保存在磁盘， worker
只预取有界数据块；snapshot 写入和校验采用流式处理。筛选接收磁盘支持的 `Sequence[DataSample]` 和评分
`Mapping`；大候选池优先迭代，反复随机索引和切片可能较慢。recipe 自行转成列表仍会占用内存， `training_rows()` 与 HF
导出也会物化数据；TIR 执行器路径使用并行排名和磁盘索引筛选；直接调用示例的 `select()` 仍会构造内存排序列表。

SQLite WAL 检查点和进程锁应放在本地文件系统。当前提供单机并行，未接入 Ray/Slurm 分布式作业。 进程死亡会释放锁，SIGKILL
后无需手动删除锁文件。`work/batch-N-{candidates,public}.jsonl` 是可从检查点重新生成的诊断导出文件。

Recipe 方法应只依赖参数和声明的输入，生成需要可确定性重放；清洗和投影 worker 分别构造自己的实例，
执行器不保存实例内部状态。旧式生成中断时重放当前批次，但跳过已提交的清洗块；分片生成会复用已完成的生成文件，仅重跑不完整的分片。 启用多个 worker
时，校准器必须支持并发调用。

仅 `OSError`（包括连接故障）和 `TimeoutError` 触发重试，采用最多 60 秒的指数退避。
校验错误和取消不重试；`Attempt.failure_reason` 表示一次已完成的校准结果，不会作为异常自动重试。
已经提交的块会复用，失败或未提交的块可能重新执行。外部服务应使用任务/重复身份保证幂等， 执行器不承诺外部副作用恰好执行一次。

## 生成、评分与全局筛选并行

已有 recipe 继续可用：没有可选方法时，生成和筛选保留原协调线程实现；评分统一按数据块交给 worker。 可选的 `ShardedDataRecipe` 和
`RankedDataRecipe` 协议描述这些能力，recipe 无需显式继承它们。 并行生成需要同时实现两个方法：

| 方法                                     | 约定                                                                                           |
| ---------------------------------------- | ---------------------------------------------------------------------------------------------- |
| `generation_shards(context, shard_size)` | 生成从零开始、连续且可确定性重放的 `GenerationShard(start, stop)`；`shard_size` 是分片大小提示 |
| `generate_shard(context, shard)`         | 生成至多 `stop-start` 条原始候选；整个计划不得超过剩余 `max_candidates` 预算                   |

生成 worker 独立构造 recipe，并流式生成、清洗和写入 `work/generation-B-S.jsonl`。 相邻 JSON
元数据记录范围、行数和内容摘要。完整且通过校验的生成文件可在恢复时复用，即使此前数据库提交失败。 协调线程按分片顺序，以 `batch_size`
大小的数据块导入并完成全局去重。较大的分片减少文件数，但重试粒度也更粗； `generate(context)` 应返回相同分片流的拼接结果。

校准响应在评分前完成校验并写入检查点。评分失败时可以直接使用已保存的响应重试或恢复，无需重复已完成的模型请求。 评分 worker 独立创建
recipe，不能依赖生成或校准阶段留下的实例状态。

并行筛选需要同时实现三个方法：

| 方法                                                    | 约定                                                                     |
| ------------------------------------------------------- | ------------------------------------------------------------------------ |
| `selection_policy(context, calibrated)`                 | 返回 `SelectionPolicy(minimum_per_group, maximum_per_bucket)`            |
| `rank(sample, scores, context)`                         | 不合格时返回 `None`，合格时返回原任务 ID 对应的 `RankedCandidate`        |
| `finalize_selection(selected, eligible_count, context)` | 返回最终任务 ID 和统计；领域质量要求未满足时抛出 `InsufficientDataError` |

`RankedCandidate` 包含有限浮点 `priority`、字符串 `tie_breaker`、`group`、`bucket` 和 JSON
`metadata`。 worker 并行判断资格并按 `(priority, tie_breaker, task_id)` 排序局部数据块，SQLite 索引保存全局顺序，
已提交的排名块在恢复时复用。候选的排名应随其输入固定，不得依赖后续批次扩展后的候选池； 需要动态全局策略的 recipe 可以继续使用原 `select()`。

协调线程先按 policy 字典顺序满足每组最低配额，再按全局排名补齐，同时检查 bucket 上限并防止重复选择。
最终合并保持串行和确定性；它是明确的贪心策略，不是任意重叠约束的求解器。 recipe 直接调用 `select_ranked_candidates`
并提供内存有序迭代器时，可以复用同一合并算法。 TIR 示例使用 16 个独立 seed 的题型配方流、题型最低配额、配方上限和最终临界题数量检查。 固定 seed 下候选不受
worker 数量影响；此生成器版本与旧版逐题结果不同，已有校准报告必须匹配实际候选和源码身份。

## 校准协议

校准工厂返回异步 `Calibrator`：

```python
async def __call__(self, tasks, context):
    # tasks: Sequence[PublicTask]；context: CalibrationContext
    # 处理公开任务，并返回实现相关的来源记录。
    return CalibrationBatch(tuple(attempts), provenance={"method": "my-calibrator-v1"})
```

`Attempt` 包含 `task_id`、从零开始的 `repeat`、可 JSON 序列化的 `response` 和可选
`failure_reason`。每个任务/重复组合必须恰好返回一次。通用协议不要求模型或 Harness 引用，非空失败原因对应 `None`
分数；已完成但答错的响应使用普通评分。校准器修改公开请求对象不会改变执行器保留的候选。

校准器接收不到 `DataSample.reference`，原始响应由执行器使用 recipe 评分。该接口是数据约定，recipe
和校准代码仍是可信执行代码。校准器自行管理依赖和版本校验，可通过 provenance 返回规则版本、来源版本或实际模型/Harness 身份。执行器逐批保存这些记录；TIR
的模型/Harness 校验由其适配器完成。

TIR 要求每题两次执行。其评分器要求最后一个 boxed 答案为整数、小数、分数字符串或数值 LaTeX 分数，不求值 `1+2`
等表达式。筛选排除执行不完整和两次均正确的题，优先选一次正确的题，再补充两次均错误的题，并检查题型覆盖和配方上限。不配置校准器时，生成任务的筛选不会宣称完成模型难度校准。

### 复用已有 TIR 评估报告

```yaml
recipe: examples.data_engineering.tir.data_recipe:create_recipe
purpose: joint
seed: 42
target_count: 512
max_batches: 8
max_candidates: 8192
params:
  candidate_count: 1024
  minimum_one_pass: 128
  maximum_per_recipe: 48
calibration:
  implementation: examples.data_engineering.tir.data_calibration:recorded_responses
  repeats: 2
  params:
    model_ref: /path/to/the/evaluated/checkpoint
    harness_ref: actual-harness-id
    reports:
      - [/path/to/batch0-repeat0.json, /path/to/batch0-repeat1.json]
    backend_model_digest: actual-model-digest
    verifier_digest: actual-verifier-digest
```

为所有可能使用的批次提供报告。报告须覆盖生成的任务 ID，并匹配模型、Harness、模型摘要和 verifier 摘要。同一批重复报告须具有不同 trial ID，以及相同
task set 和 executable 身份。执行器忽略报告中保存的分数，重新对响应评分，并保存报告与源码摘要。复用时必须匹配原生成器的 seed
和候选数量；该适配器不启动 SGLang 或重新执行评估。

## 产物与训练接入

`job.json` 保存 running/succeeded/failed/cancelled 状态。`config.json`、`progress.json` 和
`work/` 下的候选与校准记录保存执行证据。成功筛选并校验后，执行器在同一文件系统中重命名临时目录，发布完整数据集。已有输出目录需要使用
`--resume`。任务前台运行，可通过中断模块取消；恢复时沿用原输出目录和配置。

数据集目录包含：

- `public.jsonl`：选中的公开任务。
- `references.jsonl`：按任务对齐的私有参考答案。
- `train.jsonl`：recipe 生成的训练视图。
- `probes.jsonl`：原始校准响应、失败原因和重新计算的分数。
- `report.json`：生成、清洗、排除和筛选统计。
- `manifest.json`：数据集身份、文件摘要、配置、配方/后端源码摘要、输入来源摘要和实现相关的校准来源记录。

目录和数据文件仅对所有者开放。`load_snapshot(path)` 和 `snapshot.training_rows()`
校验内容摘要与行对齐。数据集按约定保持不可变，并由内容确定身份；权限设置本身不会禁止所有者写入。工厂应通过 `source_files()`
声明导入的辅助代码及配置之外的文件，执行器在发布前重新检查这些来源。

使用显式适配器接入现有训练数据加载：

```python
from areal.dataset import get_custom_dataset
from areal.v2.data_engineering import export_hf_dataset, load_snapshot

snapshot = load_snapshot("/tmp/tir-data-job/dataset")
path = export_hf_dataset(snapshot, "/tmp/another-tir-training-data")
dataset = get_custom_dataset(path=str(path), type="rl")
```

导出目录的 `areal-source.json` 记录源数据集身份。训练 Workflow 必须匹配 recipe 的字段。SFT、OPD
需要各自的投影、tokenization 或教师处理适配器；TIR 的任务视图不是 SFT 对话或已经 tokenized 的 RL 轨迹。