diff --git a/examples/codex_home_example/AGENTS.md b/examples/codex_home_example/AGENTS.md
index 234e20f..483372d 100644
--- a/examples/codex_home_example/AGENTS.md
+++ b/examples/codex_home_example/AGENTS.md
@@ -75,7 +75,8 @@
- 用户要求查看 Judger 过程或评测明细时，优先读取 `judger.pkl` 事件流
- `train` 优先使用 Trainer Skill
- SFT 必须使用 `train_stage=sft, train_framework=llamafactory`；GRPO 必须使用 `train_stage=grpo, train_framework=verl`，不要交叉组合
-- Verl GRPO 默认使用 Conda 环境 `verl`，输入必须是包含 `prompt`、`data_source`、`reward_model` 的训练/验证 Parquet；优先使用 `auto` 或经过用户确认的 LoopAI Reward 预设，只有预设无法覆盖任务时才使用自定义 reward Python 文件
+- Verl GRPO 默认使用 Conda 环境 `verl`；原生输入使用包含 `prompt`、`data_source`、`reward_model` 的 Parquet，Constructor 生成的 JSON/JSONL 则交给 Trainer 在 `prepare()` 内转换、切分并记录 manifest，不要要求 Constructor 输出 Verl 专用格式；优先使用 `auto` 或经过用户确认的 LoopAI Reward 预设，无法可靠识别时必须请用户指定，不得猜测 reward 或 ground truth
+- Verl 多轮默认继承上一轮已确认 YAML 的训练超参和成功导出的最佳 Hugging Face 模型，但每轮必须刷新当前数据、输出目录和 version，并重新展示完整 YAML 取得确认
- `obtain`、训练前数据获取、SFT 数据集构造、能力定向提升数据规划，优先读取 Obtainer Skill：`skills/obtainer/SKILL.md`
- 涉及 DataMixer、数据湖入湖、SFT recipe/export、按 math/code/text2sql/reasoning 域找数据时，先按 `skills/obtainer/SKILL.md` 和其中的 ObtainerCLI 流程执行，不要从 `outputs/` 里的旧 run 或旧 recipe 反推当前流程
- 执行数据搜集/下载/入湖时，starter 外层只能通过 CLI wrapper 启动 `dataset-acquisition-agent`；如果运行环境不是当前 shell 的 Python，先设置 `LOOPAI_PYTHON_EXECUTABLE=/path/to/loopai-env/bin/python`，再用 `${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm ... dataset-acquisition-agent start`，或在 start 命令上显式传 `--python-executable /path/to/loopai-env/bin/python`；然后轮询/续跑。不要使用通用 `spawn_agent` worker，不要在外层自己创建 SearchAgent task JSON、调用 `searchagent`、调用 `download manifest` 或直接入湖
diff --git a/examples/config/starter.yaml b/examples/config/starter.yaml
index 6d26cb3..77285da 100644
--- a/examples/config/starter.yaml
+++ b/examples/config/starter.yaml
@@ -168,10 +168,20 @@ default_states:
verl_algorithm: "grpo"
verl_rollout_backend: "vllm" # vllm 或 sglang
verl_model_backend: "fsdp"
- train_input_eval_dataset_path: "" # 验证集 Parquet 路径
- verl_reward_mode: "auto" # auto、preset 或 custom
+ # 可直接填原生 Verl Parquet；多轮闭环也可留空，由 Trainer 优先读取本轮 Constructor JSON/JSONL 输出。
+ verl_source_dataset_path: ""
+ verl_source_eval_dataset_path: ""
+ verl_data_adapter: "auto" # auto、native、messages、alpaca 或 qa
+ verl_validation_ratio: 0.05 # 没有验证集时确定性切分
+ verl_split_seed: 42
+ verl_reuse_previous_validation: true
+ train_input_eval_dataset_path: "" # 可选；上一轮验证 Parquet 默认复用
+ verl_reward_mode: "auto" # 明确指定 preset/custom 时优先，否则 Trainer 对生成数据给出保守建议
verl_reward_preset: "auto"
verl_reward_kwargs: {}
verl_selection_metric: "val-core/*/acc/mean@*"
verl_selection_mode: "max"
verl_max_actor_ckpt_to_keep: 10
+ verl_inherit_previous_config: true
+ verl_use_previous_best_model: true
+ verl_multi_round_enabled: true
diff --git a/loopai/common/db_tool/task.py b/loopai/common/db_tool/task.py
index 88db0f4..fd3af4e 100644
--- a/loopai/common/db_tool/task.py
+++ b/loopai/common/db_tool/task.py
@@ -155,7 +155,19 @@ def _extract_state_section(states_config: dict[str, Any], section_name: str) ->
def _coerce_update_item(raw_item: Any) -> dict[str, Any]:
if isinstance(raw_item, dict):
- return format_value(dict(raw_item))
+ # Configer accepts both schema-style wrappers (``{"value": ...}``)
+ # and direct field values. A direct mapping value must not be mistaken
+ # for the wrapper itself; that used to turn fields such as
+ # constructor.mapping_results into None. Likewise, infer the wrapper
+ # type when callers omit it so ``{"value": {...}}`` remains a mapping
+ # instead of being stringified.
+ if "value" not in raw_item and "default" not in raw_item and "type" not in raw_item:
+ return {"value": dict(raw_item), "type": "dict"}
+ item = dict(raw_item)
+ if "type" not in item:
+ value = item.get("value", item.get("default"))
+ item["type"] = wrap_attr(value)["type"]
+ return format_value(item)
return {"value": raw_item}
diff --git a/loopai/schema/states.py b/loopai/schema/states.py
index c3e66fa..c875e75 100644
--- a/loopai/schema/states.py
+++ b/loopai/schema/states.py
@@ -1358,9 +1358,60 @@ class TrainerState(BaseModel):
train_input_eval_dataset_path: str = Field(
default="",
title="验证数据集路径",
- description="Verl GRPO 使用的验证 Parquet 路径",
+ description="Verl GRPO 使用的验证 Parquet 路径；生成数据可由 Trainer 自动切分",
json_schema_extra={"ui_type": "file_path", "ui_group": "训练模型"}
)
+ verl_source_dataset_path: str = Field(
+ default="",
+ title="Verl 原始训练数据",
+ description="JSON/JSONL/Parquet 原始数据；留空时优先使用本轮 Constructor 输出",
+ json_schema_extra={"ui_type": "file_path", "ui_group": "训练模型"},
+ )
+ verl_source_dataset_origin: str = Field(
+ default="",
+ title="Verl 原始数据来源",
+ description="Trainer 记录 user/constructor/obtainer，用于下一轮区分显式覆盖与旧路径",
+ json_schema_extra={"ui_type": "text", "readOnly": True, "ui_group": "训练模型"},
+ )
+ verl_source_eval_dataset_path: str = Field(
+ default="",
+ title="Verl 原始验证数据",
+ description="可选的 JSON/JSONL/Parquet 验证数据；留空时自动切分或复用上一轮验证集",
+ json_schema_extra={"ui_type": "file_path", "ui_group": "训练模型"},
+ )
+ verl_data_adapter: str = Field(
+ default="auto",
+ title="Verl 数据适配器",
+ description="Trainer 将生成数据转换为 Verl Parquet 时使用的字段适配器",
+ json_schema_extra={"ui_type": "list", "ui_group": "训练模型",
+ "allowed_values": ["auto", "native", "messages", "alpaca", "qa"]},
+ )
+ verl_data_source: str = Field(
+ default="",
+ title="Verl data_source 覆盖值",
+ description="通常留空由 Trainer/reward 决定；仅在数据协议明确时手动填写",
+ json_schema_extra={"ui_type": "text", "ui_group": "训练模型"},
+ )
+ verl_validation_ratio: float = Field(
+ default=0.05,
+ gt=0.0,
+ lt=1.0,
+ title="Verl 验证集比例",
+ description="未提供验证数据时 Trainer 的确定性切分比例",
+ json_schema_extra={"ui_type": "number", "ui_group": "训练模型"},
+ )
+ verl_split_seed: int = Field(
+ default=42,
+ title="Verl 数据切分种子",
+ description="控制生成数据去重后的确定性训练/验证切分",
+ json_schema_extra={"ui_type": "number", "ui_group": "训练模型"},
+ )
+ verl_reuse_previous_validation: bool = Field(
+ default=True,
+ title="复用上一轮验证集",
+ description="多轮训练时保持验证集稳定；显式提供本轮验证源时以本轮为准",
+ json_schema_extra={"ui_type": "toggle_switch", "ui_group": "训练模型"},
+ )
verl_reward_function_path: str = Field(
default="",
title="Verl Reward 函数路径",
@@ -1380,6 +1431,12 @@ class TrainerState(BaseModel):
json_schema_extra={"ui_type": "list", "ui_group": "训练模型",
"allowed_values": ["auto", "preset", "custom"]},
)
+ verl_reward_origin: str = Field(
+ default="",
+ title="Verl Reward 选择来源",
+ description="记录 reward 是自动匹配还是用户指定，用于新一轮按新数据重新匹配",
+ json_schema_extra={"ui_type": "text", "readOnly": True, "ui_group": "训练模型"},
+ )
verl_reward_preset: str = Field(
default="auto",
title="Verl Reward 预设",
@@ -1416,6 +1473,24 @@ class TrainerState(BaseModel):
description="最多保留的 actor checkpoint 数量",
json_schema_extra={"ui_type": "number", "ui_group": "训练模型"}
)
+ verl_inherit_previous_config: bool = Field(
+ default=True,
+ title="继承上一轮 Verl 配置",
+ description="下一轮以此前已确认 YAML 为基线，仅刷新数据、模型、reward 和运行字段",
+ json_schema_extra={"ui_type": "toggle_switch", "ui_group": "训练模型"},
+ )
+ verl_use_previous_best_model: bool = Field(
+ default=True,
+ title="使用上一轮最佳模型",
+ description="上一轮成功导出 Hugging Face 模型后，自动作为下一轮 GRPO 初始模型",
+ json_schema_extra={"ui_type": "toggle_switch", "ui_group": "训练模型"},
+ )
+ verl_multi_round_enabled: bool = Field(
+ default=True,
+ title="启用 Verl 多轮衔接",
+ description="确保保存 checkpoint 并导出下一轮可直接加载的 Hugging Face 模型",
+ json_schema_extra={"ui_type": "toggle_switch", "ui_group": "训练模型"},
+ )
CUDA_VISIBLE_DEVICES: str = Field(
default="",
title="CUDA 可见设备",
@@ -1598,6 +1673,42 @@ class TrainerState(BaseModel):
description="每次启动 Trainer 子节点时生成的 version_id，用于 TaskRuntimeItem 和版本化输出目录",
json_schema_extra={"ui_type": "text", "ui_group": "训练模型"}
)
+ trainer_parent_version_id: str = Field(
+ default="",
+ title="上一轮 Trainer 版本 ID",
+ description="当前训练轮继承的数据、配置和模型所来自的 Trainer 版本",
+ json_schema_extra={"ui_type": "text", "readOnly": True, "ui_group": "训练模型"},
+ )
+ trainer_round_index: int = Field(
+ default=0,
+ title="Trainer 轮次",
+ description="同一 task 下从 1 开始递增的训练轮次",
+ json_schema_extra={"ui_type": "number", "readOnly": True, "ui_group": "训练模型"},
+ )
+ trainer_model_inheritance: Dict[str, Any] = Field(
+ default_factory=dict,
+ title="Trainer 模型继承结果",
+ description="是否采用上一轮最佳模型及其校验原因",
+ json_schema_extra={"ui_type": "json_viewer", "readOnly": True, "ui_group": "训练模型"},
+ )
+ verl_data_manifest_path: str = Field(
+ default="",
+ title="Verl 数据清单路径",
+ description="Trainer 数据转换、切分、去重和 reward 决策的版本化清单",
+ json_schema_extra={"ui_type": "file_path", "readOnly": True, "ui_group": "训练模型"},
+ )
+ verl_data_prepare_result: Dict[str, Any] = Field(
+ default_factory=dict,
+ title="Verl 数据准备结果",
+ description="本轮生成数据适配为 Verl Parquet 的统计结果",
+ json_schema_extra={"ui_type": "json_viewer", "readOnly": True, "ui_group": "训练模型"},
+ )
+ verl_reward_recommendation: Dict[str, Any] = Field(
+ default_factory=dict,
+ title="Verl Reward 建议",
+ description="Trainer 自动选择或确认 reward 的依据；不明确时不会猜测",
+ json_schema_extra={"ui_type": "json_viewer", "readOnly": True, "ui_group": "训练模型"},
+ )
trainer_output_dir: str = Field(
default="",
title="Trainer 本次输出目录",
diff --git a/loopai/skills/Trainer/README.md b/loopai/skills/Trainer/README.md
index 5dfb603..b058e2b 100644
--- a/loopai/skills/Trainer/README.md
+++ b/loopai/skills/Trainer/README.md
@@ -45,6 +45,38 @@ result = run_prepared(
GRPO 训练/验证 Parquet 至少需要 `prompt`、`data_source`、`reward_model` 三列。训练通过 `conda run --no-capture-output -n verl python -m verl.trainer.main_ppo ...` 启动；指标写入 Trainer 的 `metrics` 目录，checkpoint 按 `global_step_N` 识别，根据 YAML 中的验证指标选择最佳项，再把选中的 FSDP actor 合并为 Hugging Face 模型。
+## 生成数据与多轮 GRPO
+
+Constructor 仍负责生成和清洗数据；当 `train_framework=verl` 时，Trainer 在每轮 `prepare()` 中负责最后一段训练协议适配：
+
+1. 优先读取本轮 `constructor.mapping_results.output_file`，也可用 `verl_source_dataset_path` 显式覆盖。
+2. 自动识别原生 Verl、messages/ShareGPT、Alpaca 或普通 QA，转换为本轮目录下的 `prepared_data/train.parquet`。
+3. 若未提供验证数据，按 `verl_validation_ratio` 和 `verl_split_seed` 确定性切分；多轮仅在 reward 协议兼容时复用上一轮验证 Parquet。
+4. 生成 `dataset_manifest.json` 和 `rejected_rows.jsonl`，记录来源、去重、拒绝、切分、哈希和 reward 决策。
+5. 以上一轮已确认的 Verl YAML 为超参基线，只刷新本轮数据、最佳模型、reward、GPU、输出目录和 version；然后仍向用户展示整份新 YAML，等待本轮确认。
+
+生成数据的最小增量配置如下；通常不需要用户手写 Verl Parquet：
+
+```yaml
+trainer:
+ train_framework: verl
+ train_stage: grpo
+ verl_dir: /path/to/verl
+ verl_env_path: verl
+ train_input_model_name: /path/to/base-model
+ train_input_task_description: Mathematics reasoning with GRPO
+ verl_source_dataset_path: "" # 留空时使用本轮 Constructor 输出
+ verl_data_adapter: auto
+ verl_validation_ratio: 0.05
+ verl_reuse_previous_validation: true
+ verl_inherit_previous_config: true
+ verl_use_previous_best_model: true
+ verl_multi_round_enabled: true
+ verl_reward_mode: auto
+```
+
+`auto` 会针对每轮新的上游数据重新判断，只在任务/数据来源或答案标记能可靠对应已有 preset 时采用建议；无法判断时 `prepare()` 会停止并要求设置 `verl_reward_mode=preset`/`verl_reward_preset` 或 custom reward，不会猜测 ground truth 或 reward 语义。上一轮模型只有在训练成功、导出无错误且目录包含可加载的 Hugging Face 配置和权重时才会自动传给下一轮；若要改用指定模型，可在本轮调用显式传 `train_input_model_name`，或关闭 `verl_use_previous_best_model`。
+
## Verl Reward 预设
LoopAI 通过 `loopai/skills/Trainer/rewards/router.py` 提供稳定入口，预设只调用当前 Verl 环境中的实现，不复制 Verl 源码。
@@ -89,10 +121,11 @@ Verl 实时进度优先读取 `metrics/verl_metrics.jsonl` 的 `training/global_
### 1. 数据检查节点 (Data Check Node)
-**功能：** 验证数据集格式是否符合 LlamaFactory 要求
+**功能：** 验证 SFT 数据；对 Verl 则先适配生成数据，再验证训练/验证 Parquet 与 reward 协议
**输入：**
-- `train_input_dataset_path`: 数据集文件路径（支持 JSON/JSONL 格式）
+- `train_input_dataset_path`: SFT 数据或原生 Verl Parquet
+- `verl_source_dataset_path`: 可选的生成数据路径；未配置时读取 Constructor 输出
**输出：**
- 数据格式验证报告
diff --git a/loopai/skills/Trainer/nodes/config_generation_node.py b/loopai/skills/Trainer/nodes/config_generation_node.py
index 9352b97..aa23d52 100644
--- a/loopai/skills/Trainer/nodes/config_generation_node.py
+++ b/loopai/skills/Trainer/nodes/config_generation_node.py
@@ -1,11 +1,10 @@
"""
配置生成节点
-根据任务描述生成 LlamaFactory 训练配置（YAML格式）
+生成 LlamaFactory SFT 或 Verl GRPO 训练配置（YAML格式）
"""
import os
import yaml
-from pathlib import Path
from loopai.schema.states import LoopAIState
from loopai.skills.Trainer.utils.config_generator import ConfigGenerator, generate_config_explanation
from loopai.skills.Trainer.utils.verl_config_generator import (
@@ -22,7 +21,7 @@ def config_generation_node(state: LoopAIState) -> LoopAIState:
"""
配置生成节点
- 根据任务描述和数据集信息生成合理的 LlamaFactory 训练配置
+ 根据任务描述和数据集信息生成对应后端训练配置
Args:
state: LoopAIState 对象，需要包含：
@@ -155,9 +154,9 @@ def config_generation_node(state: LoopAIState) -> LoopAIState:
or './output/trainer'
)
os.makedirs(output_dir, exist_ok=True)
- config_output_path = state.get('trainer', {}).get('train_output_config_path')
- if not config_output_path or Path(str(config_output_path)).suffix.lower() not in {'.yaml', '.yml'}:
- config_output_path = os.path.join(output_dir, 'training_config.yaml')
+ # Every prepare() round owns a version-scoped YAML. Reusing a stale
+ # path would overwrite the previous round's approved configuration.
+ config_output_path = os.path.join(output_dir, 'training_config.yaml')
config = generate_verl_grpo_config(state, template_path)
config_output_path = save_verl_grpo_config(config, config_output_path)
diff --git a/loopai/skills/Trainer/nodes/data_check_node.py b/loopai/skills/Trainer/nodes/data_check_node.py
index 45ff8b5..c01c4bc 100644
--- a/loopai/skills/Trainer/nodes/data_check_node.py
+++ b/loopai/skills/Trainer/nodes/data_check_node.py
@@ -1,9 +1,10 @@
"""
数据检查节点
-验证数据集格式是否符合 LlamaFactory 要求
+验证 SFT 数据；为 Verl 准备并验证 GRPO 数据与 reward 协议
"""
import os
+from pathlib import Path
from loopai.skills.Trainer.utils.stream_events import prepare_trainer_run
from loopai.schema.states import LoopAIState
@@ -12,16 +13,29 @@
check_verl_grpo_inputs,
generate_verl_data_report,
)
+from loopai.skills.Trainer.utils.verl_dataset_builder import prepare_verl_grpo_datasets
from loopai.logger import get_logger
logger = get_logger()
+def _mapping_output_file(state: LoopAIState, section_name: str):
+ section = state.get(section_name) or {}
+ if not isinstance(section, dict):
+ raise ValueError(f"{section_name} state must be a mapping")
+ mapping = section.get('mapping_results')
+ if mapping in (None, ""):
+ return None
+ if not isinstance(mapping, dict):
+ raise ValueError(f"{section_name}.mapping_results must be a mapping")
+ return mapping.get('output_file')
+
+
def data_check_node(state: LoopAIState) -> LoopAIState:
"""
数据检查节点
- 检查数据集格式是否符合 LlamaFactory 要求
+ 检查 SFT 数据格式；Verl 模式下将生成数据适配为 Parquet 后预检查
Args:
state: LoopAIState 对象，需要包含：
@@ -38,14 +52,69 @@ def data_check_node(state: LoopAIState) -> LoopAIState:
try:
# 获取数据集路径 - 优先使用 obtainer/constructor 映射结果
- obtainer_output_file = state.get('obtainer', {}).get('mapping_results', {}).get('output_file') if state.get('obtainer', {}).get('mapping_results') else None
- constructor_output_file = state.get('constructor', {}).get('mapping_results', {}).get('output_file') if state.get('constructor', {}).get('mapping_results') else None
+ obtainer_output_file = _mapping_output_file(state, 'obtainer')
+ constructor_output_file = _mapping_output_file(state, 'constructor')
- framework = state.get('trainer', {}).get('train_framework')
+ trainer_state = state.setdefault('trainer', {})
+ framework = trainer_state.get('train_framework')
if framework == "verl":
- # Constructor/Obtainer currently produce SFT JSON. An explicitly supplied
- # RL Parquet must not be silently replaced by those artifacts.
- dataset_path = state.get('trainer', {}).get('train_input_dataset_path')
+ if trainer_state.get('_trainer_use_prepared_config'):
+ # run_prepared() must validate exactly the train/validation files in
+ # the YAML that the user approved. Never regenerate data at this gate.
+ dataset_path = trainer_state.get('train_input_dataset_path')
+ else:
+ explicit_source = trainer_state.get('_verl_source_dataset_explicit')
+ generated_source = constructor_output_file or obtainer_output_file
+ source_path = (
+ trainer_state.get('verl_source_dataset_path')
+ if explicit_source
+ else generated_source
+ or trainer_state.get('verl_source_dataset_path')
+ or trainer_state.get('train_input_dataset_path')
+ )
+ if not source_path:
+ raise ValueError(
+ "Verl 缺少数据源；请设置 verl_source_dataset_path，或先让 Constructor 生成数据"
+ )
+ trainer_state['verl_source_dataset_path'] = source_path
+ if not explicit_source and generated_source:
+ trainer_state['verl_source_dataset_origin'] = (
+ 'constructor' if constructor_output_file else 'obtainer'
+ )
+ logger.info(f"准备 Verl GRPO 数据源: {source_path}")
+ prepare_result = prepare_verl_grpo_datasets(state)
+ if generated_source and not explicit_source:
+ expected_source = Path(str(generated_source)).expanduser().resolve()
+ actual_source = prepare_result.get('source_path')
+ if not actual_source or Path(str(actual_source)).expanduser().resolve() != expected_source:
+ raise RuntimeError(
+ "Verl prepared data does not come from the current upstream output: "
+ f"expected {expected_source}, got {actual_source or 'missing'}"
+ )
+ manifest_reward = prepare_result.get('reward')
+ if not isinstance(manifest_reward, dict):
+ raise RuntimeError("Verl data manifest is missing its reward contract")
+ manifest_mode = str(manifest_reward.get('mode') or '')
+ current_mode = str(trainer_state.get('verl_reward_mode') or '')
+ if manifest_mode != current_mode:
+ raise RuntimeError(
+ f"Verl reward mode mismatch: manifest={manifest_mode}, state={current_mode}"
+ )
+ if current_mode in {'auto', 'preset'}:
+ manifest_preset = str(manifest_reward.get('preset') or '')
+ current_preset = str(trainer_state.get('verl_reward_preset') or '')
+ if manifest_preset != current_preset:
+ raise RuntimeError(
+ "Verl reward preset mismatch: "
+ f"manifest={manifest_preset}, state={current_preset}"
+ )
+ dataset_path = trainer_state.get('train_input_dataset_path')
+ logger.info(
+ "Verl 数据准备完成: train=%s, validation=%s, status=%s",
+ dataset_path,
+ trainer_state.get('train_input_eval_dataset_path'),
+ prepare_result.get('status'),
+ )
elif obtainer_output_file and os.path.exists(obtainer_output_file):
dataset_path = obtainer_output_file
logger.info(f"使用 obtainer 映射结果作为训练数据集: {dataset_path}")
@@ -102,7 +171,6 @@ def data_check_node(state: LoopAIState) -> LoopAIState:
for warning in check_result['warnings'][:3]: # 只显示前3个警告
logger.warning(f" - {warning}")
elif framework == "verl":
- trainer_state = state.setdefault('trainer', {})
if trainer_state.get('train_stage') != 'grpo':
raise ValueError("Verl 当前只支持 GRPO")
eval_path = trainer_state.get('train_input_eval_dataset_path')
diff --git a/loopai/skills/Trainer/runner.py b/loopai/skills/Trainer/runner.py
index 22b7929..cf779f5 100644
--- a/loopai/skills/Trainer/runner.py
+++ b/loopai/skills/Trainer/runner.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import copy
import hashlib
import json
import os
@@ -18,6 +19,36 @@
_TRAINER_TASK_STATE_UPDATE_FIELDS = {
+ "train_framework",
+ "train_stage",
+ "train_input_dataset_path",
+ "train_input_eval_dataset_path",
+ "train_input_model_name",
+ "train_input_task_description",
+ "train_input_config_template_path",
+ "verl_source_dataset_path",
+ "verl_source_dataset_origin",
+ "verl_source_eval_dataset_path",
+ "verl_data_adapter",
+ "verl_data_source",
+ "verl_validation_ratio",
+ "verl_split_seed",
+ "verl_reuse_previous_validation",
+ "verl_reward_mode",
+ "verl_reward_origin",
+ "verl_reward_preset",
+ "verl_reward_function_path",
+ "verl_reward_function_name",
+ "verl_reward_kwargs",
+ "verl_inherit_previous_config",
+ "verl_use_previous_best_model",
+ "verl_multi_round_enabled",
+ "trainer_parent_version_id",
+ "trainer_round_index",
+ "trainer_model_inheritance",
+ "verl_data_manifest_path",
+ "verl_data_prepare_result",
+ "verl_reward_recommendation",
"trainer_task_id",
"data_check_passed",
"data_check_result",
@@ -70,6 +101,62 @@
}
+_TRAINER_ROUND_TRANSIENT_FIELDS = {
+ "data_check_passed",
+ "data_check_result",
+ "data_check_report_path",
+ "data_check_error",
+ "config_generation_success",
+ "config_explanation_path",
+ "config_generation_error",
+ "training_success",
+ "training_execution_time",
+ "training_final_status",
+ "training_log_path",
+ "training_report_path",
+ "training_error",
+ "current_training_status",
+ "update_model_path",
+ "trainer_event_log_path",
+ "trainer_run_state_path",
+ "trainer_worker_log_path",
+ "trainer_worker_pid",
+ "trainer_state_update_error",
+ "trainer_output_dir",
+ "trainer_result",
+ "trainer_last_error",
+ "trainer_result_analysis",
+ "trainer_result_analysis_version_id",
+ "trainer_result_summary",
+ "trainer_best_checkpoint",
+ "trainer_best_metric",
+ "trainer_best_checkpoint_path",
+ "trainer_model_export_error",
+ "trainer_model_export_log_path",
+ "train_config",
+ "training_checkpoints",
+ "training_step_losses",
+ "trainer_data_check_result",
+ "trainer_data_check_passed",
+ "trainer_data_check_error",
+ "train_output_config_path",
+ "train_output_data_check_report_path",
+ "trainer_config_explanation_path",
+ "trainer_config_generation_success",
+ "trainer_config_generation_error",
+ "train_output_training_log_path",
+ "train_output_training_report_path",
+ "train_output_training_error",
+ "trainer_training_success",
+ "trainer_training_execution_time",
+ "trainer_training_final_status",
+ "trainer_current_training_status",
+ "verl_data_manifest_path",
+ "verl_data_prepare_result",
+ "verl_reward_recommendation",
+}
+
+
def _to_configer_update_item(value: Any) -> Dict[str, Any]:
if isinstance(value, bool):
type_name = "bool"
@@ -172,6 +259,208 @@ def _resolve_explicit_trainer_output_dir(kwargs: Dict[str, Any]) -> str | None:
return None
+def _previous_training_completed(trainer_state: Dict[str, Any]) -> bool:
+ if trainer_state.get("trainer_training_success") is True or trainer_state.get("training_success") is True:
+ return True
+ final_status = (
+ trainer_state.get("trainer_training_final_status")
+ or trainer_state.get("training_final_status")
+ or {}
+ )
+ return isinstance(final_status, dict) and str(final_status.get("status") or "").lower() == "completed"
+
+
+def _is_huggingface_model_dir(path_value: Any) -> bool:
+ if not path_value:
+ return False
+ path = Path(str(path_value)).expanduser().resolve()
+ if not path.is_dir() or not (path / "config.json").is_file():
+ return False
+ candidates = (
+ "model.safetensors",
+ "model.safetensors.index.json",
+ "pytorch_model.bin",
+ "pytorch_model.bin.index.json",
+ "adapter_model.safetensors",
+ "adapter_model.bin",
+ )
+ return any((path / name).is_file() for name in candidates)
+
+
+def _prepare_fresh_trainer_round(
+ state: Dict[str, Any],
+ *,
+ kwargs: Dict[str, Any],
+) -> None:
+ """Snapshot the previous successful Verl round and clear stale outputs."""
+ trainer_state = state.setdefault("trainer", {})
+ if trainer_state.get("train_framework") != "verl":
+ return
+
+ previous_version = str(trainer_state.get("trainer_version_id") or "")
+ previous_round = int(trainer_state.get("trainer_round_index") or 0)
+ previous_completed = _previous_training_completed(trainer_state)
+ previous_config = trainer_state.get("train_config")
+ if (
+ previous_completed
+ and trainer_state.get("verl_inherit_previous_config", True)
+ and isinstance(previous_config, dict)
+ and previous_config.get("framework") == "verl"
+ and previous_config.get("stage") == "grpo"
+ ):
+ inherited_config = copy.deepcopy(previous_config)
+ else:
+ inherited_config = None
+
+ previous_model = _first_non_empty(
+ trainer_state.get("update_model_path"),
+ trainer_state.get("trainer_best_checkpoint_path"),
+ )
+ explicit_model = _first_non_empty(
+ kwargs.get("train_input_model_name"),
+ kwargs.get("model_path"),
+ os.getenv("TRAIN_MODEL_PATH"),
+ )
+ model_inheritance: Dict[str, Any] = {
+ "applied": False,
+ "source_version_id": previous_version or None,
+ "source_model_path": str(previous_model) if previous_model else None,
+ }
+ inherited_model: str | None = None
+ if explicit_model:
+ model_inheritance["reason"] = "current round supplied an explicit model override"
+ elif not trainer_state.get("verl_use_previous_best_model", True):
+ model_inheritance["reason"] = "verl_use_previous_best_model is disabled"
+ elif not previous_completed:
+ model_inheritance["reason"] = "no previous successful training round"
+ elif trainer_state.get("trainer_model_export_error"):
+ model_inheritance["reason"] = "previous Verl model export failed"
+ elif _is_huggingface_model_dir(previous_model):
+ inherited_model = str(Path(str(previous_model)).expanduser().resolve())
+ model_inheritance.update({
+ "applied": True,
+ "model_path": inherited_model,
+ "reason": "previous best checkpoint is a loadable Hugging Face model",
+ })
+ else:
+ model_inheritance["reason"] = "previous best checkpoint is not a loadable Hugging Face model directory"
+
+ for field in _TRAINER_ROUND_TRANSIENT_FIELDS:
+ trainer_state.pop(field, None)
+
+ trainer_state["trainer_parent_version_id"] = previous_version if previous_completed else ""
+ trainer_state["trainer_round_index"] = max(1, previous_round + 1)
+ trainer_state["trainer_model_inheritance"] = model_inheritance
+ if inherited_config is not None:
+ trainer_state["_trainer_previous_config"] = inherited_config
+ else:
+ trainer_state.pop("_trainer_previous_config", None)
+ if inherited_model:
+ trainer_state["train_input_model_name"] = inherited_model
+ explicit_eval = _first_non_empty(
+ kwargs.get("train_input_eval_dataset_path"),
+ kwargs.get("eval_dataset_path"),
+ kwargs.get("verl_source_eval_dataset_path"),
+ os.getenv("TRAIN_EVAL_DATASET_PATH"),
+ os.getenv("VERL_SOURCE_EVAL_DATASET_PATH"),
+ trainer_state.get("verl_source_eval_dataset_path"),
+ )
+ previous_reward = previous_config.get("loopai_reward") if isinstance(previous_config, dict) else None
+ trainer_state["_trainer_eval_explicit"] = bool(explicit_eval)
+ if previous_completed and isinstance(previous_reward, dict):
+ trainer_state["_trainer_previous_reward"] = copy.deepcopy(previous_reward)
+ else:
+ trainer_state.pop("_trainer_previous_reward", None)
+ reward_contract_changed = False
+ if (
+ previous_completed
+ and isinstance(previous_reward, dict)
+ and trainer_state.get("verl_reward_origin") != "auto"
+ ):
+ previous_mode = str(previous_reward.get("mode") or "")
+ current_mode = str(trainer_state.get("verl_reward_mode") or "auto")
+ previous_reward_signature = (
+ previous_mode,
+ str(previous_reward.get("preset") or ""),
+ str(previous_reward.get("function_path") or "") if previous_mode == "custom" else "",
+ str(previous_reward.get("function_name") or "") if previous_mode == "custom" else "",
+ )
+ current_reward_signature = (
+ current_mode,
+ str(trainer_state.get("verl_reward_preset") or ""),
+ str(trainer_state.get("verl_reward_function_path") or "") if current_mode == "custom" else "",
+ str(trainer_state.get("verl_reward_function_name") or "compute_score") if current_mode == "custom" else "",
+ )
+ reward_contract_changed = previous_reward_signature != current_reward_signature
+ if (
+ not trainer_state.get("verl_reuse_previous_validation", True)
+ or reward_contract_changed
+ ) and not explicit_eval:
+ trainer_state["train_input_eval_dataset_path"] = ""
+
+
+def _first_config_path(value: Any) -> str | None:
+ if isinstance(value, list):
+ value = value[0] if value else None
+ return str(value) if value not in (None, "") else None
+
+
+def _hydrate_state_from_approved_config(
+ trainer_state: Dict[str, Any],
+ config: Dict[str, Any],
+) -> None:
+ """Make preflight validate the exact paths/reward contract in approved YAML."""
+ framework = str(config.get("framework") or trainer_state.get("train_framework") or "")
+ stage = str(config.get("stage") or trainer_state.get("train_stage") or "")
+ if framework:
+ trainer_state["train_framework"] = framework
+ if stage:
+ trainer_state["train_stage"] = stage
+ if framework != "verl":
+ return
+
+ overrides = config.get("overrides") or {}
+ environment = config.get("environment") or {}
+ result = config.get("result") or {}
+ reward = config.get("loopai_reward") or {}
+ train_path = _first_config_path(overrides.get("data.train_files"))
+ eval_path = _first_config_path(overrides.get("data.val_files"))
+ if train_path:
+ trainer_state["train_input_dataset_path"] = train_path
+ if eval_path:
+ trainer_state["train_input_eval_dataset_path"] = eval_path
+ if overrides.get("actor_rollout_ref.model.path"):
+ trainer_state["train_input_model_name"] = str(overrides["actor_rollout_ref.model.path"])
+ if overrides.get("actor_rollout_ref.rollout.name"):
+ trainer_state["verl_rollout_backend"] = str(overrides["actor_rollout_ref.rollout.name"])
+ if environment.get("verl_dir"):
+ trainer_state["verl_dir"] = str(environment["verl_dir"])
+ if environment.get("verl_env_path"):
+ trainer_state["verl_env_path"] = str(environment["verl_env_path"])
+ if environment.get("cuda_visible_devices") is not None:
+ trainer_state["CUDA_VISIBLE_DEVICES"] = str(environment["cuda_visible_devices"])
+ if reward.get("mode"):
+ trainer_state["verl_reward_mode"] = str(reward["mode"])
+ if reward.get("origin"):
+ trainer_state["verl_reward_origin"] = str(reward["origin"])
+ if reward.get("preset"):
+ trainer_state["verl_reward_preset"] = str(reward["preset"])
+ if reward.get("mode") == "custom":
+ trainer_state["verl_reward_function_path"] = str(reward.get("function_path") or "")
+ trainer_state["verl_reward_function_name"] = str(
+ reward.get("function_name") or "compute_score"
+ )
+ reward_kwargs = overrides.get("+reward.custom_reward_function.reward_kwargs")
+ if isinstance(reward_kwargs, dict):
+ reward_kwargs = copy.deepcopy(reward_kwargs)
+ reward_kwargs.pop("preset", None)
+ trainer_state["verl_reward_kwargs"] = reward_kwargs
+ if result.get("selection_metric"):
+ trainer_state["verl_selection_metric"] = str(result["selection_metric"])
+ if result.get("selection_mode"):
+ trainer_state["verl_selection_mode"] = str(result["selection_mode"])
+
+
def inspect_prepared_trainer_config(config_path: str) -> Dict[str, Any]:
"""Return the exact YAML text and digest used by the approval workflow."""
path = Path(config_path).expanduser().resolve()
@@ -381,6 +670,11 @@ def run_trainer_standalone(
resolved_state = runtime["state"]
trainer_state = resolved_state.setdefault("trainer", {})
explicit_version_id = _resolve_trainer_version_id(kwargs)
+ if prepared_config_path and explicit_version_id is None:
+ existing_version = trainer_state.get("trainer_version_id")
+ explicit_version_id = str(existing_version) if existing_version else None
+ if explicit_version_id is None and not prepared_config_path:
+ _prepare_fresh_trainer_round(resolved_state, kwargs=kwargs)
event_writer = prepare_trainer_run(
resolved_state,
version_id=explicit_version_id,
@@ -439,6 +733,7 @@ def run_trainer_standalone(
)
trainer_state["train_output_config_path"] = prepared_config["config_path"]
trainer_state["train_config"] = prepared_config["config"]
+ _hydrate_state_from_approved_config(trainer_state, prepared_config["config"])
# Carry the exact approved bytes into the trusted worker request.
# The worker materializes this snapshot instead of rereading a
# user-editable source path after the approval digest check.
@@ -475,7 +770,7 @@ def run_trainer_standalone(
graph = trainer()
graph_config = {
"configurable": {
- "thread_id": kwargs.get("graph_thread_id") or f"trainer_{runtime['thread_id']}",
+ "thread_id": kwargs.get("graph_thread_id") or f"trainer_{runtime['thread_id']}_{version_id}",
}
}
@@ -502,8 +797,33 @@ def run_trainer_standalone(
trainer_state.pop("_trainer_worker_runtime", None)
trainer_state.pop("_trainer_approved_config_yaml", None)
trainer_state.pop("_trainer_approved_config_sha256", None)
+ trainer_state.pop("_trainer_previous_config", None)
+ trainer_state.pop("_trainer_previous_reward", None)
+ trainer_state.pop("_trainer_eval_explicit", None)
+ trainer_state.pop("_verl_source_dataset_explicit", None)
+ trainer_state.pop("_verl_source_dataset_replaced", None)
if prepare_only:
+ if not trainer_state.get("trainer_config_generation_success"):
+ detail = _first_non_empty(
+ trainer_state.get("trainer_data_check_error"),
+ trainer_state.get("trainer_config_generation_error"),
+ "Trainer did not generate a configuration",
+ )
+ exc = ValueError(str(detail))
+ payload = emit_error(
+ exc,
+ code=ErrorCode.INVALID_INPUT,
+ recoverable=True,
+ stream_writer=event_writer,
+ message="Trainer config preparation failed.",
+ exit_process=emit_result,
+ print_payload=emit_result,
+ )
+ trainer_state["trainer_result"] = payload
+ trainer_state["trainer_last_error"] = payload["error"]
+ _update_trainer_task_state(runtime, trainer_state)
+ raise exc
try:
prepared_config = inspect_prepared_trainer_config(
str(trainer_state.get("train_output_config_path") or "")
@@ -526,7 +846,13 @@ def run_trainer_standalone(
preparation_data = {
"task_id": runtime["thread_id"],
"trainer_version_id": trainer_state.get("trainer_version_id"),
+ "trainer_parent_version_id": trainer_state.get("trainer_parent_version_id"),
+ "trainer_round_index": trainer_state.get("trainer_round_index"),
"trainer_output_dir": trainer_state.get("trainer_output_dir"),
+ "trainer_model_inheritance": trainer_state.get("trainer_model_inheritance"),
+ "verl_data_manifest_path": trainer_state.get("verl_data_manifest_path"),
+ "verl_data_prepare_result": trainer_state.get("verl_data_prepare_result"),
+ "verl_reward_recommendation": trainer_state.get("verl_reward_recommendation"),
"approval_required": True,
**prepared_config,
}
@@ -586,6 +912,8 @@ def run_trainer_standalone(
success_data = {
"task_id": runtime["thread_id"],
"trainer_version_id": trainer_state.get("trainer_version_id"),
+ "trainer_parent_version_id": trainer_state.get("trainer_parent_version_id"),
+ "trainer_round_index": trainer_state.get("trainer_round_index"),
"trainer_output_dir": trainer_state.get("trainer_output_dir"),
"trainer_training_task_id": trainer_state.get("trainer_training_task_id"),
"trainer_training_success": trainer_state.get("trainer_training_success"),
diff --git a/loopai/skills/Trainer/runtime_config.py b/loopai/skills/Trainer/runtime_config.py
index 0636b69..13f48c7 100644
--- a/loopai/skills/Trainer/runtime_config.py
+++ b/loopai/skills/Trainer/runtime_config.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import ast
import copy
import json
import os
@@ -50,8 +51,8 @@
},
"train_input_dataset_path": {
"required": True,
- "source": "user",
- "description": "Absolute path to the training dataset file, usually json/jsonl for SFT.",
+ "source": "user_or_upstream",
+ "description": "SFT dataset or native Verl Parquet; Verl may instead use the latest generated source.",
"example": "/path/to/data/alpaca_en_demo.json",
},
"train_input_model_name": {
@@ -100,9 +101,27 @@
"train_input_eval_dataset_path": {
"required": False,
"source": "user",
- "description": "Validation parquet required by the initial verl GRPO adapter.",
+ "description": "Optional validation data. Trainer can deterministically split generated Verl data when omitted.",
"example": "/path/to/validation.parquet",
},
+ "verl_source_dataset_path": {
+ "required": False,
+ "source": "user_or_upstream",
+ "description": "Generated JSON/JSONL/Parquet source; defaults to the latest Constructor output.",
+ "example": "/path/to/generated.jsonl",
+ },
+ "verl_source_eval_dataset_path": {
+ "required": False,
+ "source": "user",
+ "description": "Optional generated validation source before Trainer converts it to Verl Parquet.",
+ "example": "/path/to/generated-validation.jsonl",
+ },
+ "verl_data_adapter": {
+ "required": False,
+ "source": "auto",
+ "default": "auto",
+ "description": "Generated-data adapter: auto, native, messages, alpaca, or qa.",
+ },
"verl_reward_function_path": {
"required": False,
"source": "user",
@@ -121,6 +140,12 @@
"default": "auto",
"description": "Reward source: auto, preset, or custom.",
},
+ "verl_reward_origin": {
+ "required": False,
+ "source": "auto",
+ "default": "",
+ "description": "Whether the reward contract is automatically inferred or user-selected.",
+ },
"verl_reward_preset": {
"required": False,
"source": "auto",
@@ -145,6 +170,18 @@
"default": True,
"description": "Keep training, progress persistence, and finalization alive after the caller disconnects.",
},
+ "verl_inherit_previous_config": {
+ "required": False,
+ "source": "auto",
+ "default": True,
+ "description": "Use the previous approved Verl YAML as the next round's hyperparameter baseline.",
+ },
+ "verl_use_previous_best_model": {
+ "required": False,
+ "source": "auto",
+ "default": True,
+ "description": "Use the previous successful exported Hugging Face checkpoint as the next round model.",
+ },
}
@@ -174,6 +211,23 @@ def _as_int(value: Any, default: int) -> int:
return default
+def _as_float(value: Any, default: float) -> float:
+ if value is None or value == "":
+ return default
+ try:
+ return float(value)
+ except (TypeError, ValueError):
+ return default
+
+
+def _has_merge_value(value: Any) -> bool:
+ if value is None or value == "":
+ return False
+ if isinstance(value, (dict, list, tuple, set)) and not value:
+ return False
+ return True
+
+
def _as_dict(value: Any, field_name: str) -> Dict[str, Any]:
if value is None or value == "":
return {}
@@ -189,6 +243,46 @@ def _as_dict(value: Any, field_name: str) -> Dict[str, Any]:
raise ValueError(f"{field_name} must be a mapping")
+def _normalize_mapping_results(value: Any, section_name: str) -> Dict[str, Any]:
+ """Return one upstream mapping payload, including legacy serialized values.
+
+ Older Configer calls could persist a mapping as Python ``repr`` text. Read
+ those snapshots safely so an existing task can continue, while rejecting
+ malformed values instead of silently falling back to a previous dataset.
+ """
+ if value in (None, ""):
+ return {}
+ if isinstance(value, dict):
+ return copy.deepcopy(value)
+ if isinstance(value, str):
+ raw = value.strip()
+ parsed: Any
+ try:
+ parsed = json.loads(raw)
+ except json.JSONDecodeError:
+ try:
+ parsed = ast.literal_eval(raw)
+ except (SyntaxError, ValueError) as exc:
+ raise ValueError(
+ f"{section_name}.mapping_results must be a mapping"
+ ) from exc
+ if isinstance(parsed, dict):
+ return parsed
+ raise ValueError(f"{section_name}.mapping_results must be a mapping")
+
+
+def _upstream_mapping_results(state: Dict[str, Any], section_name: str) -> Dict[str, Any]:
+ section = state.get(section_name)
+ if section is None:
+ return {}
+ if not isinstance(section, dict):
+ raise ValueError(f"{section_name} state must be a mapping")
+ mapping = _normalize_mapping_results(section.get("mapping_results"), section_name)
+ if mapping:
+ section["mapping_results"] = mapping
+ return mapping
+
+
def _unwrap_config_value(value: Any) -> Any:
if isinstance(value, dict) and "value" in value:
return _unwrap_config_value(value.get("value"))
@@ -199,24 +293,61 @@ def _unwrap_config_value(value: Any) -> Any:
return value
-def _load_task_trainer_config(task_id: str, db_path: str | None) -> Dict[str, Any]:
+def _load_task_section_config(
+ section_name: str,
+ task_id: str,
+ db_path: str | None,
+) -> Dict[str, Any]:
if not task_id:
- raise ValueError("task_id is required for task-scoped Trainer config")
+ raise ValueError(f"task_id is required for task-scoped {section_name} config")
if not db_path:
raise ValueError("DB_PATH is required when TASK_ID is provided")
from loopai.skills.Configer import get_configer_task_state_config
os.environ["DB_PATH"] = str(db_path)
- result = get_configer_task_state_config(section_name="trainer", task_id=task_id)
+ result = get_configer_task_state_config(section_name=section_name, task_id=task_id)
if not result.get("ok"):
detail = (result.get("error") or {}).get("detail") or result.get("message")
- raise RuntimeError(detail or "failed to load Trainer task state config")
+ raise RuntimeError(detail or f"failed to load {section_name} task state config")
raw_config = result.get("data", {}).get("config", {})
return _unwrap_config_value(raw_config) if isinstance(raw_config, dict) else {}
+def _load_task_trainer_config(task_id: str, db_path: str | None) -> Dict[str, Any]:
+ return _load_task_section_config("trainer", task_id, db_path)
+
+
+def _merge_optional_task_section(
+ state: Dict[str, Any],
+ section_name: str,
+ task_id: str,
+ db_path: str | None,
+) -> None:
+ """Load only upstream mapping results; never pull unrelated section secrets."""
+ try:
+ persisted = _load_task_section_config(section_name, task_id, db_path)
+ except Exception:
+ return
+ caller_section = state.get(section_name)
+ caller_mapping = (
+ caller_section.get("mapping_results")
+ if isinstance(caller_section, dict)
+ else None
+ )
+ persisted_mapping = (
+ persisted.get("mapping_results")
+ if isinstance(persisted, dict)
+ else None
+ )
+ mapping_results = caller_mapping if _has_merge_value(caller_mapping) else persisted_mapping
+ if _has_merge_value(mapping_results):
+ merged = copy.deepcopy(caller_section) if isinstance(caller_section, dict) else {}
+ merged["mapping_results"] = copy.deepcopy(mapping_results)
+ state[section_name] = merged
+
+
def _load_config_state(config_path: str) -> Dict[str, Any]:
path = Path(config_path)
if not path.exists():
@@ -385,6 +516,7 @@ def resolve_trainer_runtime_config(
state = strip_retired_tracking_fields(state)
trainer = _trainer(state)
+ caller_trainer_config = copy.deepcopy(trainer)
system = _system(state)
explicit_task_id = _first_non_empty(
@@ -401,11 +533,36 @@ def resolve_trainer_runtime_config(
task_state_loaded = False
if explicit_task_id and explicit_task_id != _DEFAULT_THREAD_ID and (os.getenv("TASK_ID") or db_path):
db_trainer_config = _load_task_trainer_config(str(explicit_task_id), str(db_path) if db_path else None)
- trainer.update({
+ merged_trainer = {
key: value
for key, value in db_trainer_config.items()
if value is not None and value != "" and not is_retired_tracking_key(key)
- })
+ }
+ # Task state is the persistent baseline. Values supplied by the current
+ # caller are newer and therefore win, matching the documented
+ # kwargs > env > state > system > defaults contract.
+ caller_overrides = {
+ key: value
+ for key, value in caller_trainer_config.items()
+ if _has_merge_value(value) and not is_retired_tracking_key(key)
+ }
+ merged_trainer.update(caller_overrides)
+ state["trainer"] = merged_trainer
+ trainer = _trainer(state)
+ # A Trainer-only skill invocation still needs the current round's
+ # generated artifact. Read upstream task sections without mutating them.
+ _merge_optional_task_section(
+ state,
+ "constructor",
+ str(explicit_task_id),
+ str(db_path) if db_path else None,
+ )
+ _merge_optional_task_section(
+ state,
+ "obtainer",
+ str(explicit_task_id),
+ str(db_path) if db_path else None,
+ )
if db_path:
state["DB_PATH"] = str(db_path)
task_state_loaded = True
@@ -458,6 +615,64 @@ def resolve_trainer_runtime_config(
os.getenv("TRAIN_DATASET_PATH"),
trainer.get("train_input_dataset_path"),
)
+ constructor_mapping = _upstream_mapping_results(state, "constructor")
+ obtainer_mapping = _upstream_mapping_results(state, "obtainer")
+ generated_dataset_path = _first_non_empty(
+ constructor_mapping.get("output_file"),
+ obtainer_mapping.get("output_file"),
+ )
+ generated_dataset_origin = (
+ "constructor"
+ if constructor_mapping.get("output_file")
+ else "obtainer" if obtainer_mapping.get("output_file") else ""
+ )
+ requested_verl_source = _first_non_empty(
+ kwargs.get("verl_source_dataset_path"),
+ os.getenv("VERL_SOURCE_DATASET_PATH"),
+ )
+ persisted_verl_source = trainer.get("verl_source_dataset_path")
+ persisted_source_origin = str(trainer.get("verl_source_dataset_origin") or "").strip().lower()
+ # Only kwargs/environment values are explicit for *this* invocation. A
+ # task-state path marked ``user`` may simply be the previous round's seed
+ # dataset and must not permanently mask the current Constructor output.
+ explicit_verl_source = requested_verl_source
+ selected_verl_source = _first_non_empty(
+ explicit_verl_source,
+ generated_dataset_path,
+ persisted_verl_source,
+ (
+ trainer.get("train_input_dataset_path")
+ if str(trainer.get("train_input_dataset_path") or "").lower().endswith(
+ (".json", ".jsonl")
+ )
+ else None
+ ),
+ )
+ trainer["_verl_source_dataset_explicit"] = bool(explicit_verl_source)
+ trainer["verl_source_dataset_path"] = selected_verl_source
+ if explicit_verl_source:
+ trainer["verl_source_dataset_origin"] = "user"
+ elif generated_dataset_path:
+ trainer["verl_source_dataset_origin"] = generated_dataset_origin
+ elif persisted_verl_source:
+ trainer["verl_source_dataset_origin"] = persisted_source_origin or "user"
+ source_replaced_by_upstream = bool(
+ trainer.get("train_framework") == "verl"
+ and generated_dataset_path
+ and not explicit_verl_source
+ and str(generated_dataset_path) != str(persisted_verl_source or "")
+ )
+ trainer["_verl_source_dataset_replaced"] = source_replaced_by_upstream
+ if source_replaced_by_upstream:
+ # These fields describe Parquet derived from the previous source. The
+ # data builder will repopulate them from the new upstream artifact.
+ trainer["train_input_dataset_path"] = ""
+ trainer.pop("verl_data_manifest_path", None)
+ trainer.pop("verl_data_prepare_result", None)
+ if trainer.get("train_framework") == "verl" and not trainer.get("train_input_dataset_path"):
+ # The graph's generic required-field gate still expects this alias.
+ # data_check_node replaces it with the version-scoped prepared Parquet.
+ trainer["train_input_dataset_path"] = trainer.get("verl_source_dataset_path")
trainer["train_input_model_name"] = _first_non_empty(
kwargs.get("train_input_model_name"),
kwargs.get("model_path"),
@@ -476,6 +691,52 @@ def resolve_trainer_runtime_config(
os.getenv("TRAIN_EVAL_DATASET_PATH"),
trainer.get("train_input_eval_dataset_path"),
)
+ trainer["verl_source_eval_dataset_path"] = _first_non_empty(
+ kwargs.get("verl_source_eval_dataset_path"),
+ os.getenv("VERL_SOURCE_EVAL_DATASET_PATH"),
+ caller_trainer_config.get("verl_source_eval_dataset_path"),
+ trainer.get("verl_source_eval_dataset_path"),
+ )
+ trainer["verl_data_adapter"] = str(_first_non_empty(
+ kwargs.get("verl_data_adapter"),
+ os.getenv("VERL_DATA_ADAPTER"),
+ trainer.get("verl_data_adapter"),
+ "auto",
+ )).strip().lower()
+ if trainer["verl_data_adapter"] not in {"auto", "native", "messages", "alpaca", "qa"}:
+ raise ValueError("verl_data_adapter must be auto, native, messages, alpaca, or qa")
+ trainer["verl_data_source"] = str(_first_non_empty(
+ kwargs.get("verl_data_source"),
+ os.getenv("VERL_DATA_SOURCE"),
+ trainer.get("verl_data_source"),
+ "",
+ ) or "").strip()
+ trainer["verl_validation_ratio"] = _as_float(
+ _first_non_empty(
+ kwargs.get("verl_validation_ratio"),
+ os.getenv("VERL_VALIDATION_RATIO"),
+ trainer.get("verl_validation_ratio"),
+ ),
+ default=0.05,
+ )
+ if not 0.0 < trainer["verl_validation_ratio"] < 1.0:
+ raise ValueError("verl_validation_ratio must be between 0 and 1")
+ trainer["verl_split_seed"] = _as_int(
+ _first_non_empty(
+ kwargs.get("verl_split_seed"),
+ os.getenv("VERL_SPLIT_SEED"),
+ trainer.get("verl_split_seed"),
+ ),
+ default=42,
+ )
+ trainer["verl_reuse_previous_validation"] = _as_bool(
+ _first_non_empty(
+ kwargs.get("verl_reuse_previous_validation"),
+ os.getenv("VERL_REUSE_PREVIOUS_VALIDATION"),
+ trainer.get("verl_reuse_previous_validation"),
+ ),
+ default=True,
+ )
trainer["train_input_config_template_path"] = _first_non_empty(
kwargs.get("train_input_config_template_path"),
kwargs.get("config_template_path"),
@@ -551,24 +812,60 @@ def resolve_trainer_runtime_config(
trainer.get("verl_reward_function_name"),
"compute_score",
))
- raw_reward_mode = _first_non_empty(
+ requested_reward_mode = _first_non_empty(
kwargs.get("verl_reward_mode"),
os.getenv("VERL_REWARD_MODE"),
+ )
+ requested_reward_preset = _first_non_empty(
+ kwargs.get("verl_reward_preset"),
+ os.getenv("VERL_REWARD_PRESET"),
+ )
+ explicit_reward_override = any(
+ _has_merge_value(value)
+ for value in (
+ requested_reward_mode,
+ requested_reward_preset,
+ kwargs.get("verl_reward_function_path"),
+ os.getenv("VERL_REWARD_FUNCTION_PATH"),
+ )
+ )
+ raw_reward_mode = _first_non_empty(
+ requested_reward_mode,
trainer.get("verl_reward_mode"),
)
# A path without the new mode field is the legacy custom-reward contract.
if not raw_reward_mode:
raw_reward_mode = "custom" if trainer.get("verl_reward_function_path") else "auto"
+ persisted_reward_origin = str(trainer.get("verl_reward_origin") or "").strip().lower()
+ previous_recommendation = trainer.get("verl_reward_recommendation") or {}
+ if explicit_reward_override:
+ reward_origin = "auto" if str(raw_reward_mode).strip().lower() == "auto" else "user"
+ elif persisted_reward_origin in {"auto", "user"}:
+ reward_origin = persisted_reward_origin
+ elif (
+ isinstance(previous_recommendation, dict)
+ and previous_recommendation.get("source") == "trainer_generated_data_inference"
+ ):
+ reward_origin = "auto"
+ else:
+ reward_origin = "auto" if str(raw_reward_mode).strip().lower() == "auto" else "user"
+
+ # An automatically selected preset belongs to the previous source. Reset
+ # it before adapting a newly generated dataset so reward inference runs
+ # again. User-selected preset/custom rewards remain untouched.
+ if source_replaced_by_upstream and reward_origin == "auto" and not explicit_reward_override:
+ raw_reward_mode = "auto"
+ trainer["verl_reward_preset"] = "auto"
trainer["verl_reward_mode"] = str(raw_reward_mode).strip().lower()
if trainer["verl_reward_mode"] not in {"auto", "preset", "custom"}:
raise ValueError("verl_reward_mode must be auto, preset, or custom")
+ trainer["verl_reward_origin"] = reward_origin
if trainer["verl_reward_mode"] == "auto":
trainer["verl_reward_preset"] = "auto"
elif trainer["verl_reward_mode"] == "preset":
trainer["verl_reward_preset"] = normalize_reward_preset(_first_non_empty(
- kwargs.get("verl_reward_preset"),
- os.getenv("VERL_REWARD_PRESET"),
+ requested_reward_preset,
trainer.get("verl_reward_preset"),
"auto",
))
@@ -623,6 +920,30 @@ def resolve_trainer_runtime_config(
),
default=True,
)
+ trainer["verl_inherit_previous_config"] = _as_bool(
+ _first_non_empty(
+ kwargs.get("verl_inherit_previous_config"),
+ os.getenv("VERL_INHERIT_PREVIOUS_CONFIG"),
+ trainer.get("verl_inherit_previous_config"),
+ ),
+ default=True,
+ )
+ trainer["verl_use_previous_best_model"] = _as_bool(
+ _first_non_empty(
+ kwargs.get("verl_use_previous_best_model"),
+ os.getenv("VERL_USE_PREVIOUS_BEST_MODEL"),
+ trainer.get("verl_use_previous_best_model"),
+ ),
+ default=True,
+ )
+ trainer["verl_multi_round_enabled"] = _as_bool(
+ _first_non_empty(
+ kwargs.get("verl_multi_round_enabled"),
+ os.getenv("VERL_MULTI_ROUND_ENABLED"),
+ trainer.get("verl_multi_round_enabled"),
+ ),
+ default=True,
+ )
prefill_guide = build_trainer_prefill_guide(
state,
task_type=trainer["train_stage"],
@@ -642,13 +963,20 @@ def resolve_trainer_runtime_config(
def get_missing_trainer_fields(state: Dict[str, Any]) -> list[str]:
trainer = state.get("trainer") or {}
- missing = [field for field in _REQUIRED_TRAINER_FIELDS if not trainer.get(field)]
+ missing = [
+ field
+ for field in _REQUIRED_TRAINER_FIELDS
+ if field != "train_input_dataset_path" and not trainer.get(field)
+ ]
+ if not (
+ trainer.get("train_input_dataset_path")
+ or trainer.get("verl_source_dataset_path")
+ ):
+ missing.append("train_input_dataset_path")
if trainer.get("train_framework") == "llamafactory" and not trainer.get("llamafactory_dir"):
missing.append("llamafactory_dir")
if trainer.get("train_framework") == "verl" and not trainer.get("verl_dir"):
missing.append("verl_dir")
- if trainer.get("train_framework") == "verl" and not trainer.get("train_input_eval_dataset_path"):
- missing.append("train_input_eval_dataset_path")
if (
trainer.get("train_framework") == "verl"
and trainer.get("verl_reward_mode") == "custom"
diff --git a/loopai/skills/Trainer/utils/verl_config_generator.py b/loopai/skills/Trainer/utils/verl_config_generator.py
index c8e72a0..7cb904c 100644
--- a/loopai/skills/Trainer/utils/verl_config_generator.py
+++ b/loopai/skills/Trainer/utils/verl_config_generator.py
@@ -32,11 +32,25 @@ def generate_verl_grpo_config(state: Dict[str, Any], template_path: str) -> Dict
if not isinstance(raw, dict):
raise ValueError(f"Verl GRPO template must contain a YAML mapping: {path}")
- config = copy.deepcopy(raw)
- if config.get("framework") != "verl" or config.get("stage") != "grpo":
+ if raw.get("framework") != "verl" or raw.get("stage") != "grpo":
raise ValueError("Verl template must declare framework: verl and stage: grpo")
trainer = state.get("trainer") or {}
+ previous_config = trainer.get("_trainer_previous_config")
+ inherit_previous = bool(trainer.get("verl_inherit_previous_config", True))
+ if (
+ inherit_previous
+ and isinstance(previous_config, dict)
+ and previous_config.get("framework") == "verl"
+ and previous_config.get("stage") == "grpo"
+ ):
+ # Previous YAML has already crossed the user approval boundary. Use it
+ # as the hyperparameter baseline, then replace every round-scoped field.
+ config = copy.deepcopy(previous_config)
+ inherited = True
+ else:
+ config = copy.deepcopy(raw)
+ inherited = False
run_dir = Path(str(trainer.get("trainer_output_dir") or trainer.get("output_dir") or "./outputs")).resolve()
version_id = str(trainer.get("trainer_version_id") or run_dir.name)
train_value = trainer.get("train_input_dataset_path")
@@ -99,6 +113,7 @@ def generate_verl_grpo_config(state: Dict[str, Any], template_path: str) -> Dict
loopai_reward = config.setdefault("loopai_reward", {})
loopai_reward["mode"] = reward_mode
+ loopai_reward["origin"] = str(trainer.get("verl_reward_origin") or "user")
loopai_reward["preset"] = reward_preset or None
loopai_reward["function_path"] = resolved_reward_path
loopai_reward["function_name"] = reward_function_name
@@ -108,6 +123,30 @@ def generate_verl_grpo_config(state: Dict[str, Any], template_path: str) -> Dict
trainer.get("verl_selection_metric") or "val-core/*/acc/mean@*"
)
result["selection_mode"] = str(trainer.get("verl_selection_mode") or "max")
+ if trainer.get("verl_multi_round_enabled", False):
+ # A later round can only consume an exported Hugging Face model. Smoke
+ # templates used to disable both checkpointing and export, so make the
+ # multi-round contract explicit in the generated YAML shown to the user.
+ result["export_huggingface"] = True
+ try:
+ save_freq = int(overrides.get("trainer.save_freq", -1))
+ except (TypeError, ValueError):
+ save_freq = -1
+ if save_freq <= 0:
+ try:
+ test_freq = int(overrides.get("trainer.test_freq", 1))
+ except (TypeError, ValueError):
+ test_freq = 1
+ overrides["trainer.save_freq"] = max(1, test_freq)
+
+ config["loopai_round"] = {
+ "round_index": int(trainer.get("trainer_round_index") or 1),
+ "version_id": version_id,
+ "parent_version_id": trainer.get("trainer_parent_version_id") or None,
+ "inherited_previous_config": inherited,
+ "data_manifest_path": trainer.get("verl_data_manifest_path") or None,
+ "model_inheritance": copy.deepcopy(trainer.get("trainer_model_inheritance") or {}),
+ }
config = strip_retired_tracking_config(config)
config.setdefault("overrides", {})["trainer.logger"] = ["console", "file"]
assert_no_retired_tracking(config)
@@ -128,6 +167,7 @@ def generate_verl_config_explanation(config: Dict[str, Any]) -> str:
overrides = config.get("overrides") or {}
result = config.get("result") or {}
loopai_reward = config.get("loopai_reward") or {}
+ loopai_round = config.get("loopai_round") or {}
lines = [
"Verl GRPO configuration",
f"- entrypoint: {config.get('entrypoint')}",
@@ -142,5 +182,8 @@ def generate_verl_config_explanation(config: Dict[str, Any]) -> str:
f"- reward preset: {loopai_reward.get('preset') or 'custom'}",
f"- reward function: {overrides.get('reward.custom_reward_function.path')}",
f"- selection metric: {result.get('selection_metric')} ({result.get('selection_mode')})",
+ f"- round: {loopai_round.get('round_index')} (parent={loopai_round.get('parent_version_id')})",
+ f"- inherited previous config: {loopai_round.get('inherited_previous_config')}",
+ f"- data manifest: {loopai_round.get('data_manifest_path')}",
]
return "\n".join(lines) + "\n"
diff --git a/loopai/skills/Trainer/utils/verl_dataset_builder.py b/loopai/skills/Trainer/utils/verl_dataset_builder.py
new file mode 100644
index 0000000..fce305d
--- /dev/null
+++ b/loopai/skills/Trainer/utils/verl_dataset_builder.py
@@ -0,0 +1,1028 @@
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import math
+import random
+import re
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Sequence, Tuple
+
+from loopai.skills.Trainer.rewards import (
+ is_verl_builtin_data_source,
+ normalize_reward_preset,
+)
+
+
+_REQUIRED_COLUMNS = {"prompt", "data_source", "reward_model"}
+_SUPPORTED_SOURCE_SUFFIXES = {".json", ".jsonl", ".parquet"}
+_ADAPTERS = {"auto", "native", "messages", "alpaca", "qa"}
+_ANSWER_FIELDS = (
+ "ground_truth",
+ "answer",
+ "target",
+ "label",
+ "solution",
+ "reference_answer",
+ "reference",
+ "output",
+ "response",
+ "completion",
+)
+_RESPONSE_DERIVED_FIELDS = {"output", "response", "completion", "assistant"}
+_ROLE_ALIASES = {
+ "human": "user",
+ "user": "user",
+ "gpt": "assistant",
+ "assistant": "assistant",
+ "bot": "assistant",
+ "system": "system",
+ "tool": "tool",
+}
+_BOXED_START = re.compile(r"\\boxed\s*\{")
+_GSM8K_ANSWER = re.compile(r"####\s*([^\n]+)")
+_ANSWER_TAG = re.compile(r"\s*(.*?)\s*", re.IGNORECASE | re.DOTALL)
+_FINAL_ANSWER = re.compile(
+ r"(?:final\s+answer|answer|最终答案|答案)\s*[:：]\s*([^\n]+)",
+ re.IGNORECASE,
+)
+
+
+def _sha256_file(path: Path) -> str:
+ digest = hashlib.sha256()
+ with path.open("rb") as stream:
+ for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+ digest.update(chunk)
+ return digest.hexdigest()
+
+
+def _json_text(value: Any) -> str:
+ return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
+
+
+def _write_json(path: Path, value: Any) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ temp_path = path.with_name(f".{path.name}.tmp")
+ temp_path.write_text(
+ json.dumps(value, ensure_ascii=False, indent=2, default=str),
+ encoding="utf-8",
+ )
+ temp_path.replace(path)
+
+
+def _write_rejections(path: Path, rejected: Sequence[Dict[str, Any]]) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ temp_path = path.with_name(f".{path.name}.tmp")
+ with temp_path.open("w", encoding="utf-8") as stream:
+ for item in rejected:
+ stream.write(json.dumps(item, ensure_ascii=False, default=str) + "\n")
+ temp_path.replace(path)
+
+
+def _require_pyarrow():
+ try:
+ import pyarrow as pa
+ import pyarrow.parquet as pq
+ except ImportError as exc: # pragma: no cover - exercised in minimal deployments.
+ raise RuntimeError(
+ "Trainer needs pyarrow to convert generated JSON/JSONL data into Verl Parquet"
+ ) from exc
+ return pa, pq
+
+
+def _is_native_verl_parquet(path_value: Any) -> bool:
+ if not path_value:
+ return False
+ path = Path(str(path_value)).expanduser().resolve()
+ if not path.is_file() or path.suffix.lower() != ".parquet":
+ return False
+ try:
+ _, pq = _require_pyarrow()
+ return _REQUIRED_COLUMNS.issubset(pq.ParquetFile(path).schema_arrow.names)
+ except Exception:
+ return False
+
+
+def _source_files(path_value: str) -> List[Path]:
+ path = Path(path_value).expanduser().resolve()
+ if path.is_file():
+ if path.suffix.lower() not in _SUPPORTED_SOURCE_SUFFIXES:
+ raise ValueError(
+ f"unsupported Verl source file: {path}; expected JSON, JSONL, or Parquet"
+ )
+ return [path]
+ if path.is_dir():
+ files = sorted(
+ item
+ for item in path.rglob("*")
+ if item.is_file() and item.suffix.lower() in _SUPPORTED_SOURCE_SUFFIXES
+ )
+ if files:
+ return files
+ raise ValueError(f"no JSON, JSONL, or Parquet files found under: {path}")
+ raise FileNotFoundError(f"Verl source dataset does not exist: {path}")
+
+
+def _record_item(record: Dict[str, Any], path: Path, index: int) -> Dict[str, Any]:
+ return {
+ "record": record,
+ "source_path": str(path),
+ "source_index": index,
+ }
+
+
+def _load_source_records(path_value: str) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[str]]:
+ records: List[Dict[str, Any]] = []
+ rejected: List[Dict[str, Any]] = []
+ files = _source_files(path_value)
+ _, pq = _require_pyarrow()
+
+ for path in files:
+ suffix = path.suffix.lower()
+ if suffix == ".parquet":
+ try:
+ rows = pq.read_table(path).to_pylist()
+ except Exception as exc:
+ raise ValueError(f"unable to read source Parquet {path}: {exc}") from exc
+ for index, row in enumerate(rows):
+ if isinstance(row, dict):
+ records.append(_record_item(row, path, index))
+ else:
+ rejected.append({
+ "source_path": str(path),
+ "source_index": index,
+ "error": "record is not a mapping",
+ })
+ continue
+
+ if suffix == ".jsonl":
+ with path.open("r", encoding="utf-8") as stream:
+ for index, line in enumerate(stream):
+ if not line.strip():
+ continue
+ try:
+ row = json.loads(line)
+ except json.JSONDecodeError as exc:
+ rejected.append({
+ "source_path": str(path),
+ "source_index": index,
+ "error": f"invalid JSON: {exc}",
+ })
+ continue
+ if isinstance(row, dict):
+ records.append(_record_item(row, path, index))
+ else:
+ rejected.append({
+ "source_path": str(path),
+ "source_index": index,
+ "error": "record is not a mapping",
+ })
+ continue
+
+ try:
+ payload = json.loads(path.read_text(encoding="utf-8"))
+ except json.JSONDecodeError as exc:
+ raise ValueError(f"invalid JSON source {path}: {exc}") from exc
+ if isinstance(payload, list):
+ rows = payload
+ elif isinstance(payload, dict):
+ rows = next(
+ (
+ payload[key]
+ for key in ("data", "records", "items", "examples")
+ if isinstance(payload.get(key), list)
+ ),
+ [payload],
+ )
+ else:
+ rows = []
+ rejected.append({
+ "source_path": str(path),
+ "source_index": 0,
+ "error": "JSON root must be a mapping or list",
+ })
+ for index, row in enumerate(rows):
+ if isinstance(row, dict):
+ records.append(_record_item(row, path, index))
+ else:
+ rejected.append({
+ "source_path": str(path),
+ "source_index": index,
+ "error": "record is not a mapping",
+ })
+
+ return records, rejected, [str(path) for path in files]
+
+
+def _normalize_message(message: Any) -> Dict[str, str] | None:
+ if not isinstance(message, dict):
+ return None
+ raw_role = message.get("role", message.get("from"))
+ raw_content = message.get("content", message.get("value"))
+ role = _ROLE_ALIASES.get(str(raw_role or "").strip().lower())
+ if role is None or raw_content is None:
+ return None
+ if isinstance(raw_content, list):
+ content = "\n".join(str(item) for item in raw_content if item is not None)
+ else:
+ content = str(raw_content)
+ content = content.strip()
+ if not content:
+ return None
+ return {"role": role, "content": content}
+
+
+def _normalize_messages(value: Any) -> List[Dict[str, str]]:
+ if not isinstance(value, list):
+ return []
+ return [item for item in (_normalize_message(message) for message in value) if item]
+
+
+def _last_boxed_value(text: str) -> str | None:
+ matches = list(_BOXED_START.finditer(text))
+ for match in reversed(matches):
+ start = match.end()
+ depth = 1
+ for index in range(start, len(text)):
+ if text[index] == "{":
+ depth += 1
+ elif text[index] == "}":
+ depth -= 1
+ if depth == 0:
+ value = text[start:index].strip()
+ return value or None
+ return None
+
+
+def _concise_answer(text: str) -> str | None:
+ value = text.strip()
+ if not value or "\n" in value or len(value) > 128:
+ return None
+ if len(value.split()) > 20:
+ return None
+ return value
+
+
+def _unwrap_answer(value: Any) -> Any:
+ if isinstance(value, dict):
+ for key in ("target", "answer", "ground_truth", "value"):
+ if key in value:
+ return value[key]
+ return value
+
+
+def _normalize_ground_truth(value: Any, preset: str, *, response_derived: bool) -> Any:
+ value = _unwrap_answer(value)
+ if value is None:
+ raise ValueError("missing reference answer/ground truth")
+
+ if preset == "qa_exact_match":
+ targets = value if isinstance(value, (list, tuple, set)) else [value]
+ normalized = [str(item).strip() for item in targets if str(item).strip()]
+ if not normalized:
+ raise ValueError("QA ground truth is empty")
+ return {"target": normalized}
+
+ if isinstance(value, (list, tuple, set)):
+ values = [item for item in value if item is not None and str(item).strip()]
+ if len(values) != 1:
+ raise ValueError("generated math data requires exactly one reference answer")
+ value = values[0]
+ if isinstance(value, dict):
+ if response_derived:
+ raise ValueError("assistant response cannot be converted to a scalar ground truth")
+ return copy.deepcopy(value)
+
+ text = str(value).strip()
+ if not text:
+ raise ValueError("reference answer/ground truth is empty")
+
+ boxed = _last_boxed_value(text)
+ if boxed is not None:
+ return boxed
+ gsm_match = _GSM8K_ANSWER.search(text)
+ if gsm_match:
+ return gsm_match.group(1).strip()
+ answer_match = _ANSWER_TAG.search(text)
+ if answer_match:
+ return answer_match.group(1).strip()
+ final_matches = list(_FINAL_ANSWER.finditer(text))
+ if final_matches:
+ return final_matches[-1].group(1).strip()
+ if "" in text.lower():
+ tail = re.split(r"", text, flags=re.IGNORECASE)[-1].strip()
+ concise_tail = _concise_answer(tail)
+ if concise_tail:
+ return concise_tail
+
+ if response_derived and preset in {
+ "gsm8k_exact",
+ "math_boxed",
+ "math_dapo",
+ "prime_math",
+ "geometry",
+ }:
+ concise = _concise_answer(text)
+ if concise is None:
+ raise ValueError(
+ "assistant response contains reasoning but no reliable final-answer marker "
+ "(\\boxed{...}, ####, or ...)"
+ )
+ return concise
+ return text
+
+
+def _answer_from_record(record: Dict[str, Any]) -> Tuple[Any, str]:
+ reward_model = record.get("reward_model")
+ if isinstance(reward_model, dict) and reward_model.get("ground_truth") not in (None, ""):
+ return reward_model["ground_truth"], "reward_model.ground_truth"
+ for field in _ANSWER_FIELDS:
+ if record.get(field) not in (None, ""):
+ return record[field], field
+ raise ValueError("record has no answer/target/ground_truth/output field")
+
+
+def _detect_adapter(record: Dict[str, Any], requested: str) -> str:
+ if requested != "auto":
+ return requested
+ if (
+ isinstance(record.get("prompt"), list)
+ and record.get("data_source")
+ and isinstance(record.get("reward_model"), dict)
+ ):
+ return "native"
+ if any(
+ isinstance(record.get(field), list)
+ for field in ("messages", "conversation", "conversations")
+ ):
+ return "messages"
+ if "instruction" in record and any(field in record for field in _ANSWER_FIELDS):
+ return "alpaca"
+ if any(key in record for key in ("question", "query", "problem", "prompt")):
+ return "qa"
+ raise ValueError("unable to detect data adapter; set verl_data_adapter explicitly")
+
+
+def _prompt_suffix(preset: str) -> str:
+ if preset == "gsm8k_exact":
+ return 'End the response with "#### ".'
+ if preset == "qa_exact_match":
+ return "Put only the final answer inside ...."
+ if preset in {"math_boxed", "math_dapo", "prime_math", "geometry"}:
+ return "Put the final answer inside \\boxed{...}."
+ return ""
+
+
+def _ensure_reward_format(prompt: List[Dict[str, str]], preset: str) -> List[Dict[str, str]]:
+ suffix = _prompt_suffix(preset)
+ if not suffix:
+ return prompt
+ result = copy.deepcopy(prompt)
+ for index in range(len(result) - 1, -1, -1):
+ if result[index].get("role") != "user":
+ continue
+ content = str(result[index].get("content") or "")
+ markers = ("\\boxed", "####", "")
+ if not any(marker.lower() in content.lower() for marker in markers):
+ result[index]["content"] = content.rstrip() + "\n\n" + suffix
+ break
+ return result
+
+
+def _metadata_json(record: Dict[str, Any]) -> str:
+ metadata: Dict[str, Any] = {}
+ for field in (
+ "meta",
+ "metadata",
+ "extra_info",
+ "id",
+ "uuid",
+ "category",
+ "dataset",
+ "data_source",
+ "ability",
+ ):
+ if field in record:
+ metadata[field] = record[field]
+ return _json_text(metadata)
+
+
+def _adapt_record(
+ item: Dict[str, Any],
+ *,
+ requested_adapter: str,
+ reward_preset: str,
+ reward_mode: str,
+ configured_data_source: str,
+) -> Tuple[Dict[str, Any], str]:
+ record = item["record"]
+ adapter = _detect_adapter(record, requested_adapter)
+ answer_value: Any
+ answer_field: str
+
+ if adapter == "native":
+ prompt = _normalize_messages(record.get("prompt"))
+ if not prompt:
+ raise ValueError("native prompt must be a non-empty chat message list")
+ answer_value, answer_field = _answer_from_record(record)
+ elif adapter == "messages":
+ messages = _normalize_messages(
+ record.get("messages")
+ or record.get("conversation")
+ or record.get("conversations")
+ )
+ assistant_index = next(
+ (index for index in range(len(messages) - 1, -1, -1) if messages[index]["role"] == "assistant"),
+ -1,
+ )
+ if assistant_index <= 0:
+ raise ValueError("messages data requires a final assistant reference after a prompt")
+ prompt = messages[:assistant_index]
+ if not any(message["role"] == "user" for message in prompt):
+ raise ValueError("messages prompt has no user message")
+ try:
+ answer_value, answer_field = _answer_from_record(record)
+ except ValueError:
+ answer_value = messages[assistant_index]["content"]
+ answer_field = "assistant"
+ elif adapter == "alpaca":
+ instruction = str(record.get("instruction") or "").strip()
+ context = str(record.get("input") or "").strip()
+ if not instruction:
+ raise ValueError("Alpaca record requires a non-empty instruction")
+ content = instruction
+ if context:
+ content += "\n\nInput:\n" + context
+ prompt = [{"role": "user", "content": content}]
+ answer_value, answer_field = _answer_from_record(record)
+ else:
+ raw_prompt = next(
+ (record.get(key) for key in ("question", "query", "problem", "prompt") if record.get(key)),
+ None,
+ )
+ if isinstance(raw_prompt, list):
+ prompt = _normalize_messages(raw_prompt)
+ else:
+ prompt = [{"role": "user", "content": str(raw_prompt or "").strip()}]
+ if not prompt or not prompt[-1].get("content"):
+ raise ValueError("QA record requires a non-empty question/query/problem/prompt")
+ answer_value, answer_field = _answer_from_record(record)
+
+ ground_truth = _normalize_ground_truth(
+ answer_value,
+ reward_preset,
+ response_derived=answer_field in _RESPONSE_DERIVED_FIELDS,
+ )
+ prompt = _ensure_reward_format(prompt, reward_preset)
+ if not prompt:
+ raise ValueError("adapted prompt is empty")
+
+ original_source = str(record.get("data_source") or "").strip()
+ if configured_data_source:
+ data_source = configured_data_source
+ elif reward_mode == "auto" and original_source:
+ data_source = original_source
+ elif reward_mode == "custom":
+ data_source = original_source or "loopai/custom"
+ elif reward_preset not in {"auto", "verl_builtin", "custom"}:
+ data_source = f"loopai/{reward_preset}"
+ else:
+ data_source = original_source
+ if not data_source:
+ raise ValueError("adapted record has no data_source; choose a named reward preset")
+
+ source_record_id = str(
+ record.get("id")
+ or record.get("uuid")
+ or f"{Path(item['source_path']).name}:{item['source_index']}"
+ )
+ reward_model = {"style": "rule", "ground_truth": ground_truth}
+ if adapter == "native" and isinstance(record.get("reward_model"), dict):
+ reward_model.update({
+ key: copy.deepcopy(value)
+ for key, value in record["reward_model"].items()
+ if key != "ground_truth"
+ })
+ reward_model["ground_truth"] = ground_truth
+ result = {
+ "prompt": prompt,
+ "data_source": data_source,
+ "reward_model": reward_model,
+ "extra_info": {
+ "split": "",
+ "index": 0,
+ "source_path": item["source_path"],
+ "source_record_id": source_record_id,
+ "metadata_json": _metadata_json(record),
+ },
+ }
+ return result, adapter
+
+
+def _candidate_text(
+ state: Dict[str, Any],
+ records: Sequence[Dict[str, Any]],
+ source_paths: Iterable[str],
+) -> str:
+ trainer = state.get("trainer") or {}
+ parts: List[str] = [
+ str(trainer.get("train_input_task_description") or ""),
+ *(str(path) for path in source_paths),
+ ]
+ for item in records[:50]:
+ record = item.get("record") or {}
+ for key in ("data_source", "dataset", "source", "category", "task", "domain"):
+ if record.get(key):
+ parts.append(str(record[key]))
+ metadata = record.get("meta") or record.get("metadata")
+ if isinstance(metadata, dict):
+ parts.extend(str(value) for value in metadata.values())
+ return " ".join(parts).lower()
+
+
+def _infer_reward_preset(
+ state: Dict[str, Any],
+ records: Sequence[Dict[str, Any]],
+ source_paths: Sequence[str],
+) -> Tuple[str, str]:
+ text = _candidate_text(state, records, source_paths)
+ rules = (
+ ("gsm8k_exact", ("gsm8k",)),
+ ("geometry", ("geometry3k", "geo3k", "geometry", "几何")),
+ ("math_dapo", ("math_dapo", "dapo", "aime")),
+ ("prime_math", ("numina", "prime_math", "prime math")),
+ (
+ "qa_exact_match",
+ (
+ "searchr1",
+ "search_r1",
+ "triviaqa",
+ "hotpotqa",
+ "natural questions",
+ "exact match qa",
+ "问答",
+ ),
+ ),
+ (
+ "math_boxed",
+ ("math-500", "lighteval/math", "mathematics", "math", "algebra", "arithmetic", "数学", "代数", "算术"),
+ ),
+ )
+ for preset, keywords in rules:
+ matched = next((keyword for keyword in keywords if keyword in text), None)
+ if matched:
+ return preset, f"matched dataset/task keyword: {matched}"
+
+ sampled_outputs: List[str] = []
+ for item in records[:100]:
+ record = item.get("record") or {}
+ for field in ("output", "response", "completion", "answer", "solution"):
+ if record.get(field) is not None:
+ sampled_outputs.append(str(record[field]))
+ break
+ if any(_GSM8K_ANSWER.search(value) for value in sampled_outputs):
+ return "gsm8k_exact", "detected #### final-answer markers"
+ if any(_last_boxed_value(value) is not None for value in sampled_outputs):
+ return "math_boxed", "detected \\boxed{...} final-answer markers"
+ if any(_ANSWER_TAG.search(value) for value in sampled_outputs):
+ return "qa_exact_match", "detected ... markers"
+
+ raise ValueError(
+ "Trainer cannot safely infer a reward for generated Verl data. "
+ "Set verl_reward_mode=preset with verl_reward_preset, or use custom reward mode."
+ )
+
+
+def _all_builtin_sources(records: Sequence[Dict[str, Any]]) -> bool:
+ sources = {
+ str((item.get("record") or {}).get("data_source") or "").strip()
+ for item in records
+ }
+ return bool(sources) and "" not in sources and all(
+ is_verl_builtin_data_source(source) for source in sources
+ )
+
+
+def _resolve_reward(
+ state: Dict[str, Any],
+ records: Sequence[Dict[str, Any]],
+ source_paths: Sequence[str],
+ *,
+ source_is_native: bool,
+) -> Tuple[str, str, Dict[str, Any]]:
+ trainer = state.setdefault("trainer", {})
+ mode = str(trainer.get("verl_reward_mode") or "auto").strip().lower()
+ origin = str(trainer.get("verl_reward_origin") or "").strip().lower()
+ if origin not in {"auto", "user"}:
+ origin = "auto" if mode == "auto" else "user"
+ # ``preset`` with origin=auto is a recommendation produced for an older
+ # generated dataset. Re-enter auto mode so the current records determine
+ # the reward contract.
+ if origin == "auto" and mode == "preset":
+ mode = "auto"
+ trainer["verl_reward_mode"] = "auto"
+ trainer["verl_reward_preset"] = "auto"
+ trainer["verl_reward_origin"] = origin
+ if mode == "custom":
+ trainer["verl_reward_origin"] = "user"
+ return "custom", "custom", {
+ "applied": False,
+ "mode": "custom",
+ "reason": "user supplied a custom reward",
+ }
+ if mode == "preset":
+ preset = normalize_reward_preset(trainer.get("verl_reward_preset"))
+ if preset in {"auto", "verl_builtin"} and not source_is_native and not _all_builtin_sources(records):
+ raise ValueError(
+ "Generated data has no Verl-supported data_source for the built-in reward router; "
+ "choose a named verl_reward_preset or a custom reward."
+ )
+ trainer["verl_reward_origin"] = "user"
+ return mode, preset, {
+ "applied": False,
+ "mode": mode,
+ "preset": preset,
+ "reason": "user-selected reward preset",
+ }
+ if mode != "auto":
+ raise ValueError("verl_reward_mode must be auto, preset, or custom")
+
+ if source_is_native or _all_builtin_sources(records):
+ trainer["verl_reward_mode"] = "auto"
+ trainer["verl_reward_preset"] = "auto"
+ trainer["verl_reward_origin"] = "auto"
+ return "auto", "auto", {
+ "applied": False,
+ "mode": "auto",
+ "preset": "auto",
+ "reason": "native Verl data_source routing",
+ }
+
+ preset, reason = _infer_reward_preset(state, records, source_paths)
+ trainer["verl_reward_mode"] = "preset"
+ trainer["verl_reward_preset"] = preset
+ trainer["verl_reward_origin"] = "auto"
+ recommendation = {
+ "applied": True,
+ "mode": "preset",
+ "preset": preset,
+ "reason": reason,
+ "source": "trainer_generated_data_inference",
+ }
+ trainer["verl_reward_recommendation"] = recommendation
+ return "preset", preset, recommendation
+
+
+def _reward_contract_signature(reward: Dict[str, Any]) -> Tuple[str, str, str, str]:
+ mode = str(reward.get("mode") or "auto").strip().lower()
+ return (
+ mode,
+ str(reward.get("preset") or ""),
+ str(reward.get("function_path") or "") if mode == "custom" else "",
+ str(reward.get("function_name") or "compute_score") if mode == "custom" else "",
+ )
+
+
+def _record_fingerprint(record: Dict[str, Any]) -> str:
+ return hashlib.sha256(
+ _json_text({
+ "prompt": record.get("prompt"),
+ "data_source": record.get("data_source"),
+ "ground_truth": (record.get("reward_model") or {}).get("ground_truth"),
+ }).encode("utf-8")
+ ).hexdigest()
+
+
+def _deduplicate(records: Sequence[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], int]:
+ seen = set()
+ result: List[Dict[str, Any]] = []
+ duplicates = 0
+ for record in records:
+ key = _record_fingerprint(record)
+ if key in seen:
+ duplicates += 1
+ continue
+ seen.add(key)
+ result.append(record)
+ return result, duplicates
+
+
+def _assign_split(records: Sequence[Dict[str, Any]], split: str) -> List[Dict[str, Any]]:
+ result: List[Dict[str, Any]] = []
+ for index, record in enumerate(records):
+ item = copy.deepcopy(record)
+ item.setdefault("extra_info", {})["split"] = split
+ item["extra_info"]["index"] = index
+ result.append(item)
+ return result
+
+
+def _split_records(
+ records: Sequence[Dict[str, Any]],
+ *,
+ ratio: float,
+ seed: int,
+) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
+ if len(records) < 2:
+ raise ValueError(
+ "At least two valid generated records are required when no validation dataset is supplied"
+ )
+ if not math.isfinite(ratio) or not 0.0 < ratio < 1.0:
+ raise ValueError("verl_validation_ratio must be between 0 and 1")
+ indices = list(range(len(records)))
+ random.Random(seed).shuffle(indices)
+ validation_count = max(1, min(len(records) - 1, int(round(len(records) * ratio))))
+ validation_indices = set(indices[:validation_count])
+ train = [record for index, record in enumerate(records) if index not in validation_indices]
+ validation = [record for index, record in enumerate(records) if index in validation_indices]
+ return train, validation
+
+
+def _write_parquet(path: Path, records: Sequence[Dict[str, Any]]) -> None:
+ pa, pq = _require_pyarrow()
+ path.parent.mkdir(parents=True, exist_ok=True)
+ temp_path = path.with_name(f".{path.name}.tmp")
+ try:
+ table = pa.Table.from_pylist(list(records))
+ pq.write_table(table, temp_path)
+ temp_path.replace(path)
+ except Exception as exc:
+ try:
+ temp_path.unlink()
+ except OSError:
+ pass
+ raise ValueError(f"unable to write Verl Parquet {path}: {exc}") from exc
+
+
+def _adapt_items(
+ items: Sequence[Dict[str, Any]],
+ *,
+ requested_adapter: str,
+ reward_preset: str,
+ reward_mode: str,
+ configured_data_source: str,
+) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], Dict[str, int]]:
+ adapted: List[Dict[str, Any]] = []
+ rejected: List[Dict[str, Any]] = []
+ adapter_counts: Dict[str, int] = {}
+ for item in items:
+ try:
+ record, adapter = _adapt_record(
+ item,
+ requested_adapter=requested_adapter,
+ reward_preset=reward_preset,
+ reward_mode=reward_mode,
+ configured_data_source=configured_data_source,
+ )
+ adapted.append(record)
+ adapter_counts[adapter] = adapter_counts.get(adapter, 0) + 1
+ except Exception as exc:
+ rejected.append({
+ "source_path": item.get("source_path"),
+ "source_index": item.get("source_index"),
+ "error": str(exc),
+ })
+ return adapted, rejected, adapter_counts
+
+
+def prepare_verl_grpo_datasets(state: Dict[str, Any]) -> Dict[str, Any]:
+ """Prepare task-generated data for Verl and update Trainer's current paths.
+
+ Native train/validation Parquet remains untouched. JSON, JSONL, non-native
+ Parquet, or a native Parquet without validation is normalized into a
+ version-scoped train/validation pair plus an auditable manifest.
+ """
+ trainer = state.setdefault("trainer", {})
+ source_value = trainer.get("verl_source_dataset_path") or trainer.get(
+ "train_input_dataset_path"
+ )
+ if not source_value:
+ raise ValueError("Verl requires verl_source_dataset_path or train_input_dataset_path")
+ source_path = str(Path(str(source_value)).expanduser().resolve())
+
+ eval_value = trainer.get("verl_source_eval_dataset_path") or trainer.get(
+ "train_input_eval_dataset_path"
+ )
+ eval_path = str(Path(str(eval_value)).expanduser().resolve()) if eval_value else ""
+ source_native = _is_native_verl_parquet(source_path)
+ eval_native = _is_native_verl_parquet(eval_path) if eval_path else False
+
+ output_dir = Path(
+ str(trainer.get("trainer_output_dir") or trainer.get("output_dir") or "./outputs")
+ ).expanduser().resolve()
+ raw_validation_ratio = trainer.get("verl_validation_ratio")
+ validation_ratio = float(
+ 0.05 if raw_validation_ratio in (None, "") else raw_validation_ratio
+ )
+ raw_split_seed = trainer.get("verl_split_seed")
+ split_seed = int(42 if raw_split_seed in (None, "") else raw_split_seed)
+ prepared_dir = output_dir / "prepared_data"
+ manifest_path = prepared_dir / "dataset_manifest.json"
+ rejection_path = prepared_dir / "rejected_rows.jsonl"
+ prepared_dir.mkdir(parents=True, exist_ok=True)
+
+ if source_native and eval_native:
+ reward_mode = str(trainer.get("verl_reward_mode") or "auto").strip().lower()
+ reward_origin = str(trainer.get("verl_reward_origin") or "").strip().lower()
+ if reward_origin not in {"auto", "user"}:
+ reward_origin = "auto" if reward_mode == "auto" else "user"
+ if reward_origin == "auto":
+ reward_mode = "auto"
+ reward_preset = "auto"
+ trainer["verl_reward_mode"] = reward_mode
+ trainer["verl_reward_preset"] = reward_preset
+ recommendation = {
+ "applied": False,
+ "mode": "auto",
+ "preset": "auto",
+ "reason": "native Verl data_source routing",
+ }
+ else:
+ reward_preset = (
+ "custom"
+ if reward_mode == "custom"
+ else normalize_reward_preset(
+ "auto" if reward_mode == "auto" else trainer.get("verl_reward_preset")
+ )
+ )
+ recommendation = {
+ "applied": False,
+ "mode": reward_mode,
+ "preset": reward_preset,
+ "reason": "user-selected reward contract",
+ }
+ trainer["verl_reward_origin"] = reward_origin
+ manifest = {
+ "version": 1,
+ "status": "reused_native_parquet",
+ "source_path": source_path,
+ "train_path": source_path,
+ "validation_path": eval_path,
+ "source_paths": [source_path, eval_path],
+ "train_sha256": _sha256_file(Path(source_path)),
+ "validation_sha256": _sha256_file(Path(eval_path)),
+ "reward": {
+ "mode": reward_mode,
+ "origin": reward_origin,
+ "preset": reward_preset,
+ "recommendation": recommendation,
+ },
+ }
+ _write_json(manifest_path, manifest)
+ _write_rejections(rejection_path, [])
+ trainer["train_input_dataset_path"] = source_path
+ trainer["train_input_eval_dataset_path"] = eval_path
+ trainer["verl_data_manifest_path"] = str(manifest_path)
+ trainer["verl_data_prepare_result"] = manifest
+ trainer["verl_reward_recommendation"] = recommendation
+ return manifest
+
+ train_items, load_rejected, source_files = _load_source_records(source_path)
+ if not train_items:
+ raise ValueError(f"no readable records found in Verl source dataset: {source_path}")
+
+ reward_mode, reward_preset, recommendation = _resolve_reward(
+ state,
+ train_items,
+ source_files,
+ source_is_native=source_native,
+ )
+ previous_reward = trainer.get("_trainer_previous_reward")
+ eval_is_explicit = bool(
+ trainer.get("_trainer_eval_explicit")
+ or trainer.get("verl_source_eval_dataset_path")
+ )
+ current_reward = {
+ "mode": reward_mode,
+ "preset": reward_preset,
+ "function_path": trainer.get("verl_reward_function_path"),
+ "function_name": trainer.get("verl_reward_function_name"),
+ }
+ if (
+ eval_path
+ and not eval_is_explicit
+ and isinstance(previous_reward, dict)
+ and _reward_contract_signature(previous_reward)
+ != _reward_contract_signature(current_reward)
+ ):
+ # A validation set prepared for another preset/custom function cannot
+ # validate the current policy reliably. Split the new source instead.
+ eval_path = ""
+ eval_native = False
+ trainer["train_input_eval_dataset_path"] = ""
+ requested_adapter = str(trainer.get("verl_data_adapter") or "auto").strip().lower()
+ if requested_adapter not in _ADAPTERS:
+ raise ValueError(f"verl_data_adapter must be one of: {', '.join(sorted(_ADAPTERS))}")
+ configured_data_source = str(trainer.get("verl_data_source") or "").strip()
+
+ train_records, train_rejected, adapter_counts = _adapt_items(
+ train_items,
+ requested_adapter=requested_adapter,
+ reward_preset=reward_preset,
+ reward_mode=reward_mode,
+ configured_data_source=configured_data_source,
+ )
+ train_records, train_duplicates = _deduplicate(train_records)
+ if not train_records:
+ first_error = (train_rejected or load_rejected or [{}])[0].get("error")
+ raise ValueError(f"all generated Verl records were rejected: {first_error or 'unknown error'}")
+
+ rejected = [*load_rejected, *train_rejected]
+ reused_validation = False
+ validation_duplicates = 0
+ cross_split_duplicates = 0
+ if eval_path and eval_native:
+ validation_records: List[Dict[str, Any]] = []
+ final_eval_path = eval_path
+ final_train_records = _assign_split(train_records, "train")
+ reused_validation = True
+ elif eval_path:
+ eval_items, eval_load_rejected, eval_source_files = _load_source_records(eval_path)
+ validation_records, validation_rejected, eval_adapter_counts = _adapt_items(
+ eval_items,
+ requested_adapter=requested_adapter,
+ reward_preset=reward_preset,
+ reward_mode=reward_mode,
+ configured_data_source=configured_data_source,
+ )
+ validation_records, validation_duplicates = _deduplicate(validation_records)
+ rejected.extend(eval_load_rejected)
+ rejected.extend(validation_rejected)
+ for adapter, count in eval_adapter_counts.items():
+ adapter_counts[adapter] = adapter_counts.get(adapter, 0) + count
+ source_files.extend(eval_source_files)
+ if not validation_records:
+ raise ValueError("all generated Verl validation records were rejected")
+ validation_keys = {_record_fingerprint(record) for record in validation_records}
+ filtered_train_records = [
+ record
+ for record in train_records
+ if _record_fingerprint(record) not in validation_keys
+ ]
+ cross_split_duplicates = len(train_records) - len(filtered_train_records)
+ train_records = filtered_train_records
+ if not train_records:
+ raise ValueError("all generated training records overlap the validation dataset")
+ final_train_records = _assign_split(train_records, "train")
+ validation_records = _assign_split(validation_records, "validation")
+ final_eval_path = str(prepared_dir / "validation.parquet")
+ _write_parquet(Path(final_eval_path), validation_records)
+ else:
+ split_train, split_validation = _split_records(
+ train_records,
+ ratio=validation_ratio,
+ seed=split_seed,
+ )
+ final_train_records = _assign_split(split_train, "train")
+ validation_records = _assign_split(split_validation, "validation")
+ final_eval_path = str(prepared_dir / "validation.parquet")
+ _write_parquet(Path(final_eval_path), validation_records)
+
+ final_train_path = str(prepared_dir / "train.parquet")
+ _write_parquet(Path(final_train_path), final_train_records)
+ _write_rejections(rejection_path, rejected)
+
+ manifest = {
+ "version": 1,
+ "status": "prepared",
+ "source_path": source_path,
+ "source_paths": sorted(set(source_files)),
+ "train_path": final_train_path,
+ "validation_path": final_eval_path,
+ "reused_validation": reused_validation,
+ "input_records": len(train_items),
+ "train_records": len(final_train_records),
+ "validation_records": (
+ None if reused_validation else len(validation_records)
+ ),
+ "rejected_records": len(rejected),
+ "duplicate_records": train_duplicates + validation_duplicates + cross_split_duplicates,
+ "cross_split_duplicates": cross_split_duplicates,
+ "adapter_counts": adapter_counts,
+ "requested_adapter": requested_adapter,
+ "reward": {
+ "mode": reward_mode,
+ "origin": trainer.get("verl_reward_origin") or "user",
+ "preset": reward_preset,
+ "recommendation": recommendation,
+ },
+ "split": {
+ "ratio": validation_ratio,
+ "seed": split_seed,
+ },
+ "rejection_path": str(rejection_path),
+ "train_sha256": _sha256_file(Path(final_train_path)),
+ "validation_sha256": _sha256_file(Path(final_eval_path)),
+ }
+ _write_json(manifest_path, manifest)
+
+ trainer["train_input_dataset_path"] = final_train_path
+ trainer["train_input_eval_dataset_path"] = final_eval_path
+ trainer["verl_data_manifest_path"] = str(manifest_path)
+ trainer["verl_data_prepare_result"] = manifest
+ trainer["verl_reward_recommendation"] = recommendation
+ return manifest
+
+
+__all__ = ["prepare_verl_grpo_datasets"]
diff --git a/skills/Trainer/SKILL.md b/skills/Trainer/SKILL.md
index e180042..88f64bd 100644
--- a/skills/Trainer/SKILL.md
+++ b/skills/Trainer/SKILL.md
@@ -116,12 +116,13 @@ prepared = prepare(
### Verl GRPO State
-Use both training and validation Parquet files. When `pyarrow` is available,
-Trainer checks the schema, reads the distinct `data_source` values, and
-semantically validates up to the first 100 rows of each file. Every sampled row
-must contain a non-empty chat-message-list `prompt`, a supported `data_source`,
-and `reward_model.ground_truth`. Without `pyarrow`, Trainer checks the file and
-Parquet magic bytes, emits a warning, and defers schema validation to Verl.
+Native Verl inputs may supply both training and validation Parquet files. When
+the current round instead has Constructor-generated JSON/JSONL, leave
+`verl_source_dataset_path` empty to use `constructor.mapping_results.output_file`,
+or set it explicitly. Trainer converts native/messages/Alpaca/QA records into a
+version-scoped Parquet pair before generating YAML. `pyarrow` is required for
+conversion. Every output row contains a non-empty chat-message-list `prompt`, a
+`data_source`, and `reward_model.ground_truth`.
```python
prepared = prepare(
@@ -146,6 +147,61 @@ prepared = prepare(
)
```
+### Generated Data and Multi-Round Verl
+
+Keep the boundary between components explicit:
+
+- Constructor generates and cleans task data.
+- Trainer owns Verl-only adaptation, deterministic train/validation splitting,
+ reward contract selection, and executable GRPO config generation.
+- Do not add Verl-specific output formats to Constructor and do not change the
+ existing SFT Constructor path.
+
+On every fresh Verl `prepare()` round:
+
+1. Prefer an explicitly supplied `verl_source_dataset_path`; otherwise use the
+ current Constructor output before any persisted older source.
+2. Reuse native train/validation Parquet unchanged. Convert JSON, JSONL, or
+ non-native Parquet under
+ `{trainer_output_dir}/prepared_data/{train,validation}.parquet`.
+3. Write `dataset_manifest.json` and `rejected_rows.jsonl`. Reject records that
+ lack a reliable prompt or reference answer; never infer ground truth with an
+ LLM or copy the assistant answer into the prompt.
+4. If validation data is absent, split deterministically using
+ `verl_validation_ratio` (default `0.05`) and `verl_split_seed` (default `42`).
+ With `verl_reuse_previous_validation=true`, keep the previous validation
+ Parquet stable only while the resolved reward contract remains compatible;
+ otherwise split validation from the new source.
+5. With `verl_inherit_previous_config=true`, use the preceding successful
+ round's approved `train_config` as the hyperparameter baseline. Always
+ replace train/validation/model paths, reward fields, devices, experiment and
+ checkpoint directories, selection settings, and version metadata.
+6. With `verl_use_previous_best_model=true`, promote `update_model_path` only
+ when the preceding round completed, export did not fail, and the directory
+ contains a loadable Hugging Face config plus weights. Never pass raw FSDP
+ shards into the next round. To deliberately restart from another model,
+ pass a current-call `train_input_model_name`/`model_path` override or set
+ `verl_use_previous_best_model=false` together with the desired model path.
+7. With `verl_multi_round_enabled=true`, the prepared YAML enables Hugging Face
+ export and a positive checkpoint save cadence, including when the selected
+ smoke template originally disabled them.
+8. Show and approve the complete newly generated YAML again. Previous-round
+ approval never authorizes a new round.
+
+Minimal generated-data fields:
+
+```python
+"trainer": {
+ "train_framework": "verl",
+ "train_stage": "grpo",
+ "verl_dir": "/path/to/verl",
+ "train_input_model_name": "/path/to/base-model",
+ "train_input_task_description": "Mathematics GRPO",
+ "verl_data_adapter": "auto",
+ "verl_reward_mode": "auto",
+}
+```
+
Use exactly one reward mode:
- `auto`: route by Parquet `data_source` through Verl's built-in router. When
@@ -158,6 +214,13 @@ The preset router imports reward implementations lazily from the configured Verl
environment. Do not copy Verl reward source into LoopAI and do not silently fall
back from an unknown `data_source` or preset.
+For generated non-native data in `auto` mode, Trainer may recommend an existing
+preset only when task/dataset metadata or an explicit answer marker makes the
+mapping reliable (for example GSM8K, MATH/boxed, DAPO/AIME, Numina/PRIME,
+Geometry3K, or Search-R1-style QA). A user-specified named preset or custom
+reward always wins. If the mapping is ambiguous, stop preparation and ask the
+user to select a preset or custom reward; do not guess.
+
## Runtime Configuration
Priority:
@@ -174,6 +237,11 @@ from loopai.skills.Configer import get_configer_task_state_config
cfg = get_configer_task_state_config("trainer", task_id=TASK_ID)
```
+For Verl data handoff it also reads only `mapping_results` from the task-scoped
+`constructor` and `obtainer` sections as optional, read-only upstream state so
+a Trainer-only invocation can locate the latest output without pulling
+unrelated section configuration or credentials into the worker state.
+
`DB_PATH` must also be set for task-scoped loading. After the run completes or fails, Trainer Skill writes structured Trainer result fields back through Configer:
```python
@@ -193,6 +261,8 @@ Useful environment variables:
- `TRAIN_STAGE`
- `TRAIN_DATASET_PATH`
- `TRAIN_EVAL_DATASET_PATH`
+- `VERL_SOURCE_DATASET_PATH` and `VERL_SOURCE_EVAL_DATASET_PATH`
+- `VERL_DATA_ADAPTER`, `VERL_DATA_SOURCE`, `VERL_VALIDATION_RATIO`, and `VERL_SPLIT_SEED`
- `TRAIN_MODEL_PATH`
- `TRAIN_TASK_DESCRIPTION`
- `TRAIN_CONFIG_TEMPLATE_PATH`
@@ -202,6 +272,7 @@ Useful environment variables:
- `VERL_ENV_PATH` or `VERL_CONDA_ENV`
- `VERL_REWARD_MODE`, `VERL_REWARD_PRESET`, and `VERL_REWARD_KWARGS`
- `VERL_REWARD_FUNCTION_PATH` and `VERL_REWARD_FUNCTION_NAME` for custom reward mode
+- `VERL_INHERIT_PREVIOUS_CONFIG`, `VERL_USE_PREVIOUS_BEST_MODEL`, and `VERL_MULTI_ROUND_ENABLED`
- `TRAINER_PERSISTENT_WORKER`
- `CUDA_VISIBLE_DEVICES`
@@ -218,7 +289,7 @@ Required Trainer fields:
- `train_input_config_template_path`
- `train_input_model_name`
- for SFT: `train_framework=llamafactory`, `train_stage=sft`, and `llamafactory_dir`
-- for GRPO: `train_framework=verl`, `train_stage=grpo`, `verl_dir`, `train_input_eval_dataset_path`, and a valid reward mode
+- for GRPO: `train_framework=verl`, `train_stage=grpo`, `verl_dir`, either a native train Parquet or generated source/Constructor output, and a valid reward mode. Validation may be supplied, reused, or deterministically split.
## Versioned Runtime
@@ -286,7 +357,7 @@ Fields that usually require user or task-specific input:
- `train_input_dataset_path`
- `train_input_model_name`
- `train_input_task_description`
-- `llamafactory_dir` for SFT, or `verl_dir` and `train_input_eval_dataset_path` for GRPO
+- `llamafactory_dir` for SFT, or `verl_dir` for GRPO. A separate validation path is optional when Trainer can split generated data.
Fields that Trainer can usually prefill:
@@ -294,6 +365,9 @@ Fields that Trainer can usually prefill:
- `train_input_config_template_path`: selects the bundled SFT or GRPO YAML template
- `verl_env_path`: defaults to Conda environment `verl`
- `verl_reward_mode`: defaults to `auto`
+- `verl_data_adapter`: defaults to `auto`
+- `verl_validation_ratio` / `verl_split_seed`: default to `0.05` / `42`
+- `verl_inherit_previous_config`, `verl_use_previous_best_model`, `verl_multi_round_enabled`: default to `true`
- `trainer_persistent_worker`: defaults to `true`
- `CUDA_VISIBLE_DEVICES`: defaults to `0`
diff --git a/tests/test_trainer_verl_multiround.py b/tests/test_trainer_verl_multiround.py
new file mode 100644
index 0000000..0515381
--- /dev/null
+++ b/tests/test_trainer_verl_multiround.py
@@ -0,0 +1,730 @@
+import json
+import sqlite3
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from loopai.common.db_tool.task import update_task_state_section_config_sync
+from loopai.skills.Trainer import prepare
+from loopai.skills.Trainer.runner import (
+ _hydrate_state_from_approved_config,
+ _prepare_fresh_trainer_round,
+)
+from loopai.skills.Trainer.runtime_config import resolve_trainer_runtime_config
+from loopai.skills.Trainer.utils.verl_config_generator import generate_verl_grpo_config
+from loopai.skills.Trainer.utils.verl_dataset_builder import prepare_verl_grpo_datasets
+
+
+_TEMPLATE = (
+ Path(__file__).resolve().parents[1]
+ / "loopai"
+ / "skills"
+ / "Trainer"
+ / "templates"
+ / "verl_grpo_smoke.yaml"
+)
+
+
+def _write_jsonl(path: Path, rows: list[dict]) -> None:
+ path.write_text(
+ "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n",
+ encoding="utf-8",
+ )
+
+
+def _write_native_verl_parquet(path: Path, rows: list[dict]) -> None:
+ pq.write_table(pa.Table.from_pylist(rows), path)
+
+
+def _generated_state(tmp_path: Path, source: Path) -> dict:
+ return {
+ "task_id": "round-task",
+ "output_dir": str(tmp_path / "outputs"),
+ "trainer": {
+ "train_framework": "verl",
+ "train_stage": "grpo",
+ "trainer_output_dir": str(tmp_path / "run"),
+ "verl_source_dataset_path": str(source),
+ "train_input_task_description": "Optimize mathematical reasoning with GRPO.",
+ "verl_reward_mode": "auto",
+ "verl_reward_preset": "auto",
+ "verl_data_adapter": "auto",
+ "verl_validation_ratio": 0.34,
+ "verl_split_seed": 7,
+ },
+ }
+
+
+def test_configer_preserves_direct_and_wrapped_mapping_values(tmp_path: Path) -> None:
+ db_path = tmp_path / "state.db"
+ with sqlite3.connect(db_path) as connection:
+ connection.execute(
+ """
+ create table taskmodel (
+ id integer primary key,
+ task_id text not null,
+ name text not null,
+ config text,
+ state text
+ )
+ """
+ )
+ connection.execute(
+ "insert into taskmodel(task_id, name, config, state) values (?, ?, ?, ?)",
+ ("task", "task", "{}", '{"constructor": {}}'),
+ )
+ connection.commit()
+
+ first = {"output_file": "/tmp/round-1.jsonl", "mapped_records": 2}
+ update_task_state_section_config_sync(
+ db_path,
+ "task",
+ "constructor",
+ {"mapping_results": first},
+ )
+ with sqlite3.connect(db_path) as connection:
+ state = json.loads(connection.execute("select state from taskmodel").fetchone()[0])
+ assert state["constructor"]["mapping_results"] == first
+
+ second = {"output_file": "/tmp/round-2.jsonl", "mapped_records": 3}
+ update_task_state_section_config_sync(
+ db_path,
+ "task",
+ "constructor",
+ {"mapping_results": {"value": second}},
+ )
+ with sqlite3.connect(db_path) as connection:
+ state = json.loads(connection.execute("select state from taskmodel").fetchone()[0])
+ assert state["constructor"]["mapping_results"] == second
+
+
+def test_generated_alpaca_is_converted_to_versioned_verl_parquet(tmp_path: Path) -> None:
+ source = tmp_path / "generated.jsonl"
+ _write_jsonl(
+ source,
+ [
+ {"instruction": "2+2?", "input": "", "output": "4"},
+ {"instruction": "3+5?", "input": "", "output": "8"},
+ {"instruction": "5+7?", "input": "", "output": "12"},
+ ],
+ )
+ state = _generated_state(tmp_path, source)
+
+ result = prepare_verl_grpo_datasets(state)
+
+ trainer = state["trainer"]
+ assert trainer["verl_reward_mode"] == "preset"
+ assert trainer["verl_reward_preset"] == "math_boxed"
+ assert Path(result["train_path"]).parent == tmp_path / "run" / "prepared_data"
+ assert result["train_records"] == 2
+ assert result["validation_records"] == 1
+ assert result["rejected_records"] == 0
+ assert Path(trainer["verl_data_manifest_path"]).is_file()
+
+ train_rows = pq.read_table(result["train_path"]).to_pylist()
+ assert train_rows[0]["data_source"] == "loopai/math_boxed"
+ assert train_rows[0]["reward_model"]["ground_truth"]
+ assert "\\boxed" in train_rows[0]["prompt"][-1]["content"]
+ assert all(row["extra_info"]["split"] == "train" for row in train_rows)
+
+
+def test_generated_data_does_not_guess_ambiguous_reward(tmp_path: Path) -> None:
+ source = tmp_path / "classification.jsonl"
+ _write_jsonl(
+ source,
+ [
+ {"instruction": "Classify A", "output": "yes"},
+ {"instruction": "Classify B", "output": "no"},
+ ],
+ )
+ state = _generated_state(tmp_path, source)
+ state["trainer"]["train_input_task_description"] = "Binary classification"
+
+ with pytest.raises(ValueError, match="cannot safely infer a reward"):
+ prepare_verl_grpo_datasets(state)
+
+
+def test_prepare_surfaces_ambiguous_reward_error_without_masking_it(
+ tmp_path: Path,
+ monkeypatch,
+) -> None:
+ monkeypatch.delenv("TASK_ID", raising=False)
+ monkeypatch.delenv("DB_PATH", raising=False)
+ source = tmp_path / "classification.jsonl"
+ _write_jsonl(
+ source,
+ [
+ {"instruction": "Classify A", "output": "yes"},
+ {"instruction": "Classify B", "output": "no"},
+ ],
+ )
+ state = _generated_state(tmp_path, source)
+ state["trainer"].update({
+ "verl_dir": str(tmp_path),
+ "train_input_model_name": "/models/base",
+ "train_input_task_description": "Binary classification",
+ })
+
+ with pytest.raises(ValueError, match="cannot safely infer a reward"):
+ prepare(state=state, thread_id=f"ambiguous-{tmp_path.name}")
+
+
+def test_user_reward_preset_wins_for_generated_data(tmp_path: Path) -> None:
+ source = tmp_path / "qa.jsonl"
+ _write_jsonl(
+ source,
+ [
+ {"question": "Capital of France?", "answer": "Paris"},
+ {"question": "Capital of Japan?", "answer": "Tokyo"},
+ ],
+ )
+ state = _generated_state(tmp_path, source)
+ state["trainer"].update({
+ "train_input_task_description": "general task",
+ "verl_reward_mode": "preset",
+ "verl_reward_preset": "qa_exact_match",
+ })
+
+ result = prepare_verl_grpo_datasets(state)
+ rows = pq.read_table(result["train_path"]).to_pylist() + pq.read_table(
+ result["validation_path"]
+ ).to_pylist()
+
+ assert state["trainer"]["verl_reward_preset"] == "qa_exact_match"
+ assert rows[0]["reward_model"]["ground_truth"]["target"]
+ assert "" in rows[0]["prompt"][-1]["content"]
+
+
+def test_custom_reward_generated_data_gets_stable_default_data_source(tmp_path: Path) -> None:
+ source = tmp_path / "custom.jsonl"
+ _write_jsonl(
+ source,
+ [
+ {"instruction": "Task A", "output": "reference A"},
+ {"instruction": "Task B", "output": "reference B"},
+ ],
+ )
+ state = _generated_state(tmp_path, source)
+ state["trainer"].update({
+ "train_input_task_description": "custom task",
+ "verl_reward_mode": "custom",
+ "verl_reward_function_path": str(tmp_path / "reward.py"),
+ })
+
+ result = prepare_verl_grpo_datasets(state)
+ rows = pq.read_table(result["train_path"]).to_pylist() + pq.read_table(
+ result["validation_path"]
+ ).to_pylist()
+
+ assert {row["data_source"] for row in rows} == {"loopai/custom"}
+
+
+def test_next_round_inherits_hyperparameters_but_replaces_dynamic_fields(tmp_path: Path) -> None:
+ previous = {
+ "framework": "verl",
+ "stage": "grpo",
+ "entrypoint": "verl.trainer.main_ppo",
+ "environment": {"verl_dir": "/old/verl", "cuda_visible_devices": "0"},
+ "overrides": {
+ "data.train_files": ["/old/train.parquet"],
+ "data.val_files": ["/old/validation.parquet"],
+ "actor_rollout_ref.model.path": "/old/model",
+ "actor_rollout_ref.actor.optim.lr": 3.0e-7,
+ "actor_rollout_ref.actor.ppo_mini_batch_size": 3,
+ "actor_rollout_ref.rollout.n": 7,
+ "trainer.experiment_name": "old-version",
+ "trainer.default_local_dir": "/old/checkpoints",
+ "trainer.save_freq": 4,
+ "trainer.test_freq": 4,
+ },
+ "result": {"selection_metric": "old", "selection_mode": "min", "export_huggingface": True},
+ }
+ state = {
+ "trainer": {
+ "trainer_output_dir": str(tmp_path / "new-run"),
+ "trainer_version_id": "new-version",
+ "trainer_parent_version_id": "old-version",
+ "trainer_round_index": 2,
+ "_trainer_previous_config": previous,
+ "verl_inherit_previous_config": True,
+ "verl_multi_round_enabled": True,
+ "train_input_dataset_path": str(tmp_path / "new-train.parquet"),
+ "train_input_eval_dataset_path": str(tmp_path / "new-validation.parquet"),
+ "train_input_model_name": "/new/model/",
+ "verl_dir": "/new/verl",
+ "verl_env_path": "verl",
+ "CUDA_VISIBLE_DEVICES": "0,1",
+ "verl_reward_mode": "preset",
+ "verl_reward_preset": "math_boxed",
+ "verl_reward_kwargs": {},
+ "verl_rollout_backend": "vllm",
+ "verl_model_backend": "fsdp",
+ "verl_selection_metric": "val/reward",
+ "verl_selection_mode": "max",
+ "verl_max_actor_ckpt_to_keep": 10,
+ }
+ }
+
+ config = generate_verl_grpo_config(state, str(_TEMPLATE))
+ overrides = config["overrides"]
+
+ assert overrides["actor_rollout_ref.actor.optim.lr"] == 3.0e-7
+ assert overrides["actor_rollout_ref.actor.ppo_mini_batch_size"] == 3
+ assert overrides["actor_rollout_ref.rollout.n"] == 7
+ assert overrides["data.train_files"] == [str((tmp_path / "new-train.parquet").resolve())]
+ assert overrides["actor_rollout_ref.model.path"] == "/new/model"
+ assert overrides["trainer.experiment_name"] == "new-version"
+ assert overrides["trainer.default_local_dir"] == str((tmp_path / "new-run" / "checkpoints").resolve())
+ assert config["result"]["selection_metric"] == "val/reward"
+ assert config["loopai_round"]["inherited_previous_config"] is True
+
+
+def test_new_round_promotes_previous_exported_hf_model_and_clears_outputs(tmp_path: Path) -> None:
+ model = tmp_path / "exported-model"
+ model.mkdir()
+ (model / "config.json").write_text("{}", encoding="utf-8")
+ (model / "model.safetensors").write_bytes(b"weights")
+ previous_config = {
+ "framework": "verl",
+ "stage": "grpo",
+ "overrides": {"actor_rollout_ref.actor.optim.lr": 1.0e-6},
+ }
+ state = {
+ "trainer": {
+ "train_framework": "verl",
+ "trainer_version_id": "round-1",
+ "trainer_round_index": 1,
+ "trainer_training_success": True,
+ "trainer_training_final_status": {"status": "completed"},
+ "train_config": previous_config,
+ "update_model_path": str(model),
+ "train_input_model_name": "/models/base",
+ "verl_inherit_previous_config": True,
+ "verl_use_previous_best_model": True,
+ "verl_reuse_previous_validation": True,
+ "trainer_result": {"ok": True},
+ }
+ }
+
+ _prepare_fresh_trainer_round(state, kwargs={})
+ trainer = state["trainer"]
+
+ assert trainer["trainer_parent_version_id"] == "round-1"
+ assert trainer["trainer_round_index"] == 2
+ assert trainer["train_input_model_name"] == str(model.resolve())
+ assert trainer["trainer_model_inheritance"]["applied"] is True
+ assert trainer["_trainer_previous_config"] == previous_config
+ assert "train_config" not in trainer
+ assert "update_model_path" not in trainer
+ assert "trainer_result" not in trainer
+
+
+def test_current_round_model_override_wins_over_previous_best(tmp_path: Path) -> None:
+ exported_model = tmp_path / "exported-model"
+ exported_model.mkdir()
+ (exported_model / "config.json").write_text("{}", encoding="utf-8")
+ (exported_model / "model.safetensors").write_bytes(b"weights")
+ state = {
+ "trainer": {
+ "train_framework": "verl",
+ "trainer_version_id": "round-1",
+ "trainer_training_success": True,
+ "train_config": {
+ "framework": "verl",
+ "stage": "grpo",
+ "overrides": {"actor_rollout_ref.model.path": "/models/round-1-input"},
+ },
+ "update_model_path": str(exported_model),
+ "train_input_model_name": "/models/user-override",
+ "verl_use_previous_best_model": True,
+ }
+ }
+
+ _prepare_fresh_trainer_round(
+ state,
+ kwargs={"train_input_model_name": "/models/user-override"},
+ )
+
+ assert state["trainer"]["train_input_model_name"] == "/models/user-override"
+ assert state["trainer"]["trainer_model_inheritance"]["applied"] is False
+ assert "explicit model override" in state["trainer"]["trainer_model_inheritance"]["reason"]
+
+
+def test_runtime_prefers_current_constructor_output_over_persisted_user_seed(tmp_path: Path) -> None:
+ current = tmp_path / "current.jsonl"
+ current.write_text("{}\n", encoding="utf-8")
+ state = {
+ "task_id": "task",
+ "output_dir": str(tmp_path),
+ "constructor": {"mapping_results": {"output_file": str(current)}},
+ "trainer": {
+ "train_framework": "verl",
+ "train_stage": "grpo",
+ "verl_dir": str(tmp_path),
+ "verl_source_dataset_path": "/old/generated.jsonl",
+ "verl_source_dataset_origin": "user",
+ "verl_data_prepare_result": {"source_path": "/old/generated.jsonl"},
+ "train_input_dataset_path": "/old/prepared/train.parquet",
+ "train_input_model_name": "/models/base",
+ "train_input_task_description": "math",
+ },
+ }
+
+ runtime = resolve_trainer_runtime_config(state=state)
+
+ assert runtime["state"]["trainer"]["verl_source_dataset_path"] == str(current)
+ assert runtime["state"]["trainer"]["train_input_dataset_path"] == str(current)
+ assert runtime["state"]["trainer"]["_verl_source_dataset_explicit"] is False
+ assert runtime["state"]["trainer"]["verl_source_dataset_origin"] == "constructor"
+
+
+def test_runtime_recovers_legacy_serialized_constructor_mapping(tmp_path: Path) -> None:
+ current = tmp_path / "current.jsonl"
+ current.write_text("{}\n", encoding="utf-8")
+ state = {
+ "task_id": "task",
+ "output_dir": str(tmp_path),
+ "constructor": {
+ "mapping_results": str({"output_file": str(current), "mapped_records": 1})
+ },
+ "trainer": {
+ "train_framework": "verl",
+ "train_stage": "grpo",
+ "verl_dir": str(tmp_path),
+ "verl_source_dataset_path": "/old/train.parquet",
+ "verl_source_dataset_origin": "user",
+ "train_input_dataset_path": "/old/train.parquet",
+ "train_input_model_name": "/models/base",
+ "train_input_task_description": "math",
+ },
+ }
+
+ runtime = resolve_trainer_runtime_config(state=state)
+
+ assert runtime["state"]["constructor"]["mapping_results"]["output_file"] == str(current)
+ assert runtime["state"]["trainer"]["verl_source_dataset_path"] == str(current)
+ assert runtime["state"]["trainer"]["verl_source_dataset_origin"] == "constructor"
+
+
+def test_verl_round_source_replacement_does_not_clear_sft_runtime_input(tmp_path: Path) -> None:
+ generated = tmp_path / "generated.jsonl"
+ configured = tmp_path / "configured.jsonl"
+ generated.write_text("{}\n", encoding="utf-8")
+ configured.write_text("{}\n", encoding="utf-8")
+ state = {
+ "task_id": "sft-task",
+ "output_dir": str(tmp_path),
+ "constructor": {"mapping_results": {"output_file": str(generated)}},
+ "trainer": {
+ "train_framework": "llamafactory",
+ "train_stage": "sft",
+ "train_input_dataset_path": str(configured),
+ "train_input_model_name": "/models/base",
+ "train_input_task_description": "SFT",
+ "llamafactory_dir": str(tmp_path),
+ },
+ }
+
+ runtime = resolve_trainer_runtime_config(state=state)
+
+ assert runtime["state"]["trainer"]["train_input_dataset_path"] == str(configured)
+ assert runtime["state"]["trainer"]["_verl_source_dataset_replaced"] is False
+
+
+def test_runtime_respects_user_source_override_even_when_constructor_has_output(tmp_path: Path) -> None:
+ current = tmp_path / "current.jsonl"
+ explicit = tmp_path / "explicit.jsonl"
+ current.write_text("{}\n", encoding="utf-8")
+ explicit.write_text("{}\n", encoding="utf-8")
+ state = {
+ "task_id": "task",
+ "output_dir": str(tmp_path),
+ "constructor": {"mapping_results": {"output_file": str(current)}},
+ "trainer": {
+ "train_framework": "verl",
+ "train_stage": "grpo",
+ "verl_dir": str(tmp_path),
+ "train_input_model_name": "/models/base",
+ "train_input_task_description": "math",
+ },
+ }
+
+ runtime = resolve_trainer_runtime_config(
+ state=state,
+ verl_source_dataset_path=str(explicit),
+ )
+
+ trainer = runtime["state"]["trainer"]
+ assert trainer["verl_source_dataset_path"] == str(explicit)
+ assert trainer["_verl_source_dataset_explicit"] is True
+ assert trainer["verl_source_dataset_origin"] == "user"
+
+
+def test_task_scoped_runtime_loads_constructor_output_for_trainer_only_call(
+ tmp_path: Path,
+ monkeypatch,
+) -> None:
+ generated = tmp_path / "generated.jsonl"
+ generated.write_text("{}\n", encoding="utf-8")
+
+ def fake_section(section_name, task_id, db_path):
+ assert task_id == "task-from-db"
+ if section_name == "trainer":
+ return {
+ "train_framework": "verl",
+ "train_stage": "grpo",
+ "verl_dir": str(tmp_path),
+ "train_input_model_name": "/models/base",
+ "train_input_task_description": "math",
+ }
+ if section_name == "constructor":
+ return {"mapping_results": {"output_file": str(generated)}}
+ return {}
+
+ monkeypatch.setattr(
+ "loopai.skills.Trainer.runtime_config._load_task_section_config",
+ fake_section,
+ )
+
+ runtime = resolve_trainer_runtime_config(
+ state={"task_id": "task-from-db", "output_dir": str(tmp_path)},
+ thread_id="task-from-db",
+ db_path=str(tmp_path / "state.db"),
+ )
+
+ trainer = runtime["state"]["trainer"]
+ assert runtime["task_state_loaded"] is True
+ assert trainer["verl_source_dataset_path"] == str(generated)
+ assert trainer["verl_source_dataset_origin"] == "constructor"
+
+
+def test_approved_verl_yaml_hydrates_preflight_paths_and_reward() -> None:
+ trainer = {
+ "train_input_dataset_path": "/stale/train.parquet",
+ "verl_reward_mode": "auto",
+ }
+ config = {
+ "framework": "verl",
+ "stage": "grpo",
+ "environment": {
+ "verl_dir": "/approved/verl",
+ "verl_env_path": "verl",
+ "cuda_visible_devices": "0,1",
+ },
+ "overrides": {
+ "data.train_files": ["/approved/train.parquet"],
+ "data.val_files": ["/approved/validation.parquet"],
+ "actor_rollout_ref.model.path": "/approved/model",
+ "actor_rollout_ref.rollout.name": "sglang",
+ "+reward.custom_reward_function.reward_kwargs": {
+ "preset": "math_boxed",
+ "strict": True,
+ },
+ },
+ "loopai_reward": {"mode": "preset", "preset": "math_boxed"},
+ "result": {"selection_metric": "val/reward", "selection_mode": "max"},
+ }
+
+ _hydrate_state_from_approved_config(trainer, config)
+
+ assert trainer["train_input_dataset_path"] == "/approved/train.parquet"
+ assert trainer["train_input_eval_dataset_path"] == "/approved/validation.parquet"
+ assert trainer["train_input_model_name"] == "/approved/model"
+ assert trainer["verl_reward_mode"] == "preset"
+ assert trainer["verl_reward_preset"] == "math_boxed"
+ assert trainer["verl_reward_kwargs"] == {"strict": True}
+
+
+def test_two_prepare_rounds_convert_new_data_and_inherit_model_and_yaml(
+ tmp_path: Path,
+ monkeypatch,
+) -> None:
+ monkeypatch.delenv("TASK_ID", raising=False)
+ monkeypatch.delenv("DB_PATH", raising=False)
+ monkeypatch.setattr(
+ "loopai.skills.Trainer.utils.verl_data_checker._smoke_test_preset",
+ lambda *args, **kwargs: {"status": "skipped", "reason": "unit test"},
+ )
+ first_source = tmp_path / "round-1.jsonl"
+ second_source = tmp_path / "round-2.jsonl"
+ _write_jsonl(
+ first_source,
+ [
+ {"instruction": "1+1?", "output": "2"},
+ {"instruction": "2+3?", "output": "5"},
+ ],
+ )
+ _write_jsonl(
+ second_source,
+ [
+ {"instruction": "3+4?", "output": "7"},
+ {"instruction": "6+7?", "output": "13"},
+ ],
+ )
+ verl_dir = tmp_path / "verl"
+ verl_dir.mkdir()
+ task_id = f"two-round-{tmp_path.name}"
+ state = {
+ "task_id": task_id,
+ "output_dir": str(tmp_path / "outputs"),
+ "constructor": {"mapping_results": {"output_file": str(first_source)}},
+ "trainer": {
+ "train_framework": "verl",
+ "train_stage": "grpo",
+ "verl_dir": str(verl_dir),
+ "verl_env_path": "verl",
+ "train_input_model_name": "/models/base",
+ "train_input_task_description": "Mathematics GRPO",
+ "verl_reward_mode": "auto",
+ "CUDA_VISIBLE_DEVICES": "0,1",
+ },
+ }
+
+ first = prepare(state=state, thread_id=task_id)
+ first_trainer = first["trainer"]
+ first_approval = first_trainer["trainer_result"]["data"]
+ first_validation = first_trainer["train_input_eval_dataset_path"]
+
+ exported_model = tmp_path / "round-1-model"
+ exported_model.mkdir()
+ (exported_model / "config.json").write_text("{}", encoding="utf-8")
+ (exported_model / "model.safetensors").write_bytes(b"weights")
+ first_trainer["trainer_training_success"] = True
+ first_trainer["trainer_training_final_status"] = {"status": "completed"}
+ first_trainer["update_model_path"] = str(exported_model)
+ first["constructor"] = {"mapping_results": {"output_file": str(second_source)}}
+
+ second = prepare(state=first, thread_id=task_id)
+ second_trainer = second["trainer"]
+ second_approval = second_trainer["trainer_result"]["data"]
+ second_config = second_approval["config"]
+
+ assert second_approval["trainer_version_id"] != first_approval["trainer_version_id"]
+ assert second_trainer["trainer_parent_version_id"] == first_approval["trainer_version_id"]
+ assert second_trainer["trainer_round_index"] == 2
+ assert second_trainer["train_input_model_name"] == str(exported_model.resolve())
+ assert second_config["overrides"]["actor_rollout_ref.model.path"] == str(exported_model.resolve())
+ assert second_config["loopai_round"]["inherited_previous_config"] is True
+ assert second_config["loopai_round"]["parent_version_id"] == first_approval["trainer_version_id"]
+ assert second_trainer["verl_source_dataset_path"] == str(second_source)
+ assert Path(second_trainer["train_input_dataset_path"]).parent.parent.name == second_approval[
+ "trainer_version_id"
+ ]
+ assert second_trainer["train_input_eval_dataset_path"] == first_validation
+
+
+def test_second_round_replaces_user_seed_data_and_rematches_reward(
+ tmp_path: Path,
+ monkeypatch,
+) -> None:
+ monkeypatch.delenv("TASK_ID", raising=False)
+ monkeypatch.delenv("DB_PATH", raising=False)
+ monkeypatch.setattr(
+ "loopai.skills.Trainer.utils.verl_data_checker._smoke_test_preset",
+ lambda *args, **kwargs: {"status": "skipped", "reason": "unit test"},
+ )
+ old_train = tmp_path / "math-train.parquet"
+ old_validation = tmp_path / "math-validation.parquet"
+ _write_native_verl_parquet(
+ old_train,
+ [
+ {
+ "prompt": [{"role": "user", "content": "1+1?"}],
+ "data_source": "DigitalLearningGmbH/MATH-lighteval",
+ "reward_model": {"style": "rule", "ground_truth": "2"},
+ },
+ {
+ "prompt": [{"role": "user", "content": "2+3?"}],
+ "data_source": "DigitalLearningGmbH/MATH-lighteval",
+ "reward_model": {"style": "rule", "ground_truth": "5"},
+ },
+ ],
+ )
+ _write_native_verl_parquet(
+ old_validation,
+ [
+ {
+ "prompt": [{"role": "user", "content": "3+4?"}],
+ "data_source": "DigitalLearningGmbH/MATH-lighteval",
+ "reward_model": {"style": "rule", "ground_truth": "7"},
+ }
+ ],
+ )
+ new_source = tmp_path / "searchr1-round-2.jsonl"
+ _write_jsonl(
+ new_source,
+ [
+ {"question": "Capital of France?", "answer": "Paris"},
+ {"question": "Capital of Japan?", "answer": "Tokyo"},
+ {"question": "Capital of China?", "answer": "Beijing"},
+ ],
+ )
+
+ task_id = f"seed-to-generated-{tmp_path.name}"
+ state = {
+ "task_id": task_id,
+ "output_dir": str(tmp_path / "outputs"),
+ "trainer": {
+ "train_framework": "verl",
+ "train_stage": "grpo",
+ "verl_dir": str(tmp_path / "verl"),
+ "verl_env_path": "verl",
+ "verl_source_dataset_path": str(old_train),
+ "verl_source_dataset_origin": "user",
+ "train_input_dataset_path": str(old_train),
+ "train_input_eval_dataset_path": str(old_validation),
+ "train_input_model_name": "/models/base",
+ "train_input_task_description": "Mathematics GRPO",
+ "verl_reward_mode": "auto",
+ "CUDA_VISIBLE_DEVICES": "0,1",
+ },
+ }
+ (tmp_path / "verl").mkdir()
+
+ first = prepare(state=state, thread_id=task_id)
+ first_trainer = first["trainer"]
+ first_config = first_trainer["trainer_result"]["data"]["config"]
+ assert first_config["overrides"]["data.train_files"] == [str(old_train.resolve())]
+ assert first_config["loopai_reward"]["mode"] == "auto"
+ assert first_config["loopai_reward"]["origin"] == "auto"
+
+ exported_model = tmp_path / "round-1-model"
+ exported_model.mkdir()
+ (exported_model / "config.json").write_text("{}", encoding="utf-8")
+ (exported_model / "model.safetensors").write_bytes(b"weights")
+ first_trainer["trainer_training_success"] = True
+ first_trainer["trainer_training_final_status"] = {"status": "completed"}
+ first_trainer["update_model_path"] = str(exported_model)
+ first_trainer["train_input_task_description"] = "SearchR1 exact match QA"
+ first["constructor"] = {"mapping_results": {"output_file": str(new_source)}}
+
+ second = prepare(state=first, thread_id=task_id)
+ trainer = second["trainer"]
+ approval = trainer["trainer_result"]["data"]
+ config = approval["config"]
+ manifest = trainer["verl_data_prepare_result"]
+
+ assert trainer["trainer_round_index"] == 2
+ assert trainer["verl_source_dataset_path"] == str(new_source)
+ assert trainer["verl_source_dataset_origin"] == "constructor"
+ assert manifest["source_path"] == str(new_source.resolve())
+ assert manifest["status"] == "prepared"
+ assert manifest["reused_validation"] is False
+ assert manifest["reward"]["mode"] == "preset"
+ assert manifest["reward"]["origin"] == "auto"
+ assert manifest["reward"]["preset"] == "qa_exact_match"
+ assert config["loopai_reward"]["mode"] == "preset"
+ assert config["loopai_reward"]["origin"] == "auto"
+ assert config["loopai_reward"]["preset"] == "qa_exact_match"
+ assert config["overrides"]["data.train_files"] == [trainer["train_input_dataset_path"]]
+ assert str(old_train.resolve()) not in config["overrides"]["data.train_files"]
+ assert str(old_validation.resolve()) not in config["overrides"]["data.val_files"]
+
+ rows = pq.read_table(trainer["train_input_dataset_path"]).to_pylist()
+ rows += pq.read_table(trainer["train_input_eval_dataset_path"]).to_pylist()
+ assert len(rows) == 3
+ assert {row["data_source"] for row in rows} == {"loopai/qa_exact_match"}
+ assert all(row["reward_model"]["ground_truth"]["target"] for row in rows)
+ assert all("" in row["prompt"][-1]["content"] for row in rows)