[feat] add InterleaveThinker data normalizers

This commit is contained in:
Mac Lee
2026-06-19 18:47:41 +00:00
parent a22da778a6
commit dcd82f93e8
4 changed files with 670 additions and 0 deletions
@@ -2086,3 +2086,105 @@ Conclusion:
separately, the next integration decision is whether to support all three
models in one process or prefer a split generator service for memory
isolation.
## Stage 6 Execution: Official Dataset Normalization
Status: validated, pending commit/push.
Scope for this implementation slice:
- Add upstream InterleaveThinker dataset utilities for:
- `planner_sft.json`;
- `critic_sft.json`;
- `critic_rl.jsonl`.
- Preserve upstream ShareGPT-style SFT rows while adding FastVideo-friendly
aliases:
- planner user instruction and assistant response;
- critic origin/edited image path aliases;
- critic `previous_prompt` / `rewritten_prompt` compatibility.
- Normalize critic RL rows to the reward-loop fields already used by
`InterleaveThinkerRLMethod` and `InterleaveThinkerRewardScorer`:
- `origin_prompt`;
- `previous_prompt`;
- `rewritten_prompt`;
- `origin_image_path`;
- `edited_image_path`;
- `previous_image_path`;
- `ground_truth`.
- Support `image_dir` path resolution and image-extension validation for scalar
and list image fields.
- Add clear errors for missing files, empty datasets, malformed JSON, missing
required prompt/image fields, and invalid image extensions.
Expected files:
- `fastvideo/train/models/interleave_thinker/data.py`
- `fastvideo/train/models/interleave_thinker/__init__.py`
- `tests/local_tests/test_interleave_thinker_data.py`
- this handoff file
Validation plan:
- Local syntax and `git diff --check`.
- Modal pytest for new data tests plus existing planner/critic model tests.
- Modal pre-commit on changed non-excluded files.
Implemented changes:
- Added `fastvideo/train/models/interleave_thinker/data.py` with:
- `load_interleave_dataset(...)`;
- `load_planner_sft_records(...)`;
- `load_critic_sft_records(...)`;
- `load_critic_rl_records(...)`;
- role-specific normalizers for planner SFT, critic SFT, and critic RL rows;
- `image_dir` path resolution for scalar and list image fields;
- image extension validation and optional file-existence validation;
- normalized critic RL `ground_truth` fields compatible with
`InterleaveThinkerRewardScorer`.
- Exported dataset utilities from
`fastvideo.train.models.interleave_thinker`.
- Added `tests/local_tests/test_interleave_thinker_data.py` covering:
- upstream ShareGPT planner SFT rows;
- critic SFT rows with `images` -> before/after aliases;
- critic RL JSONL rows with `evaluation` -> `ground_truth`;
- empty dataset rejection;
- missing critic RL ground-truth success rejection;
- invalid image extension rejection.
Validation completed:
- Local lightweight checks:
- `python -m py_compile` passed for
`fastvideo/train/models/interleave_thinker/data.py`,
`fastvideo/train/models/interleave_thinker/__init__.py`, and
`tests/local_tests/test_interleave_thinker_data.py`.
- `git diff --check` passed.
- First Modal dataset pytest attempt:
- App URL: `https://modal.com/apps/hao-ai-lab/main/ap-8PYqdvn1QFDZTIRBtrDIV9`
- Failed before tests because the submitted `--git-commit` value had an
incorrect full SHA for the pushed base commit.
- Modal dataset pytest rerun:
- App URL: `https://modal.com/apps/hao-ai-lab/main/ap-bOKl6Q4p0uOjGmER7lvmc5`
- Command:
`pytest tests/local_tests/test_interleave_thinker_data.py
tests/local_tests/test_interleave_thinker_critic_model.py
tests/local_tests/test_interleave_thinker_planner_model.py -q`
- Result: `17 passed, 14 warnings in 16.27s`.
- Modal pre-commit:
- App URL: `https://modal.com/apps/hao-ai-lab/main/ap-ISuDU2lwc6Pl5NYDZnnBEb`
- Command:
`pre-commit run --files
fastvideo/train/models/interleave_thinker/__init__.py
fastvideo/train/models/interleave_thinker/data.py
tests/local_tests/test_interleave_thinker_data.py`
- Result: yapf, ruff, codespell, mypy, filename check, and suggestion hooks
passed; PyMarkdown/actionlint skipped with no files to check.
Conclusion:
- FastVideo now has upstream-file-aware dataset normalization for
InterleaveThinker planner SFT, critic SFT, and critic RL data.
- These utilities do not yet replace the generic `Qwen3VLActorDataset` loader;
the next SFT/RL training integration slice should either call these utilities
from model `init_preprocessors(...)` by dataset kind or add config fields that
explicitly select the normalizer.
@@ -5,6 +5,24 @@ from fastvideo.train.models.interleave_thinker.critic import (
INTERLEAVE_CRITIC_PROMPT,
InterleaveThinkerCriticModel,
)
from fastvideo.train.models.interleave_thinker.data import (
DEFAULT_FILENAMES,
IMAGE_EXTENSIONS,
IMAGE_LIST_KEYS,
IMAGE_PATH_KEYS,
InterleaveDatasetKind,
load_critic_rl_records,
load_critic_sft_records,
load_interleave_dataset,
load_planner_sft_records,
normalize_critic_rl_record,
normalize_critic_sft_record,
normalize_ground_truth,
normalize_interleave_dataset_record,
normalize_planner_sft_record,
resolve_interleave_image_path,
validate_image_path,
)
from fastvideo.train.models.interleave_thinker.planner import (
INTERLEAVE_GUIDANCE_PLANNER_PROMPT,
INTERLEAVE_PLANNER_PROMPT,
@@ -18,9 +36,25 @@ __all__ = [
"INTERLEAVE_CRITIC_PROMPT",
"INTERLEAVE_GUIDANCE_PLANNER_PROMPT",
"INTERLEAVE_PLANNER_PROMPT",
"DEFAULT_FILENAMES",
"IMAGE_EXTENSIONS",
"IMAGE_LIST_KEYS",
"IMAGE_PATH_KEYS",
"InterleaveDatasetKind",
"InterleavePlannerOutput",
"InterleavePlannerStep",
"InterleaveThinkerCriticModel",
"InterleaveThinkerPlannerModel",
"extract_interleave_plan",
"load_critic_rl_records",
"load_critic_sft_records",
"load_interleave_dataset",
"load_planner_sft_records",
"normalize_critic_rl_record",
"normalize_critic_sft_record",
"normalize_ground_truth",
"normalize_interleave_dataset_record",
"normalize_planner_sft_record",
"resolve_interleave_image_path",
"validate_image_path",
]
@@ -0,0 +1,392 @@
# SPDX-License-Identifier: Apache-2.0
"""Dataset normalization utilities for InterleaveThinker training files."""
from __future__ import annotations
from collections.abc import Mapping, Sequence
import json
import os
from pathlib import Path
from typing import Any, Literal
InterleaveDatasetKind = Literal["planner_sft", "critic_sft", "critic_rl"]
DEFAULT_FILENAMES: dict[InterleaveDatasetKind, str] = {
"planner_sft": "planner_sft.json",
"critic_sft": "critic_sft.json",
"critic_rl": "critic_rl.jsonl",
}
IMAGE_PATH_KEYS = (
"origin_image_path",
"previous_image_path",
"edited_image_path",
"generated_image_path",
"input_image_path",
"output_image_path",
"image_path",
"target_img",
)
IMAGE_LIST_KEYS = (
"images",
"input_image_paths",
"image_paths",
)
IMAGE_EXTENSIONS = frozenset({
".bmp",
".gif",
".jpeg",
".jpg",
".png",
".tif",
".tiff",
".webp",
})
def load_interleave_dataset(
data_path: str | os.PathLike[str],
*,
kind: InterleaveDatasetKind,
image_dir: str | os.PathLike[str] = "",
validate_image_files: bool = False,
) -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
for file_path in _resolve_dataset_files(data_path, kind):
for raw in _load_json_records(file_path):
records.append(
normalize_interleave_dataset_record(
raw,
kind=kind,
image_dir=image_dir,
validate_image_files=validate_image_files,
))
if not records:
raise ValueError(f"No {kind} records found at {data_path!s}")
return records
def load_planner_sft_records(
data_path: str | os.PathLike[str],
*,
image_dir: str | os.PathLike[str] = "",
validate_image_files: bool = False,
) -> list[dict[str, Any]]:
return load_interleave_dataset(
data_path,
kind="planner_sft",
image_dir=image_dir,
validate_image_files=validate_image_files,
)
def load_critic_sft_records(
data_path: str | os.PathLike[str],
*,
image_dir: str | os.PathLike[str] = "",
validate_image_files: bool = False,
) -> list[dict[str, Any]]:
return load_interleave_dataset(
data_path,
kind="critic_sft",
image_dir=image_dir,
validate_image_files=validate_image_files,
)
def load_critic_rl_records(
data_path: str | os.PathLike[str],
*,
image_dir: str | os.PathLike[str] = "",
validate_image_files: bool = False,
) -> list[dict[str, Any]]:
return load_interleave_dataset(
data_path,
kind="critic_rl",
image_dir=image_dir,
validate_image_files=validate_image_files,
)
def normalize_interleave_dataset_record(
record: Mapping[str, Any],
*,
kind: InterleaveDatasetKind,
image_dir: str | os.PathLike[str] = "",
validate_image_files: bool = False,
) -> dict[str, Any]:
normalized = _resolve_record_image_paths(
record,
image_dir=image_dir,
validate_image_files=validate_image_files,
)
if kind == "planner_sft":
return normalize_planner_sft_record(normalized)
if kind == "critic_sft":
return normalize_critic_sft_record(normalized)
if kind == "critic_rl":
return normalize_critic_rl_record(normalized)
raise ValueError(f"Unsupported InterleaveThinker dataset kind: {kind!r}")
def normalize_planner_sft_record(record: Mapping[str, Any]) -> dict[str, Any]:
normalized = dict(record)
messages = _require_messages(normalized, where="planner_sft")
user_message = _first_message_content(messages, "user")
assistant_message = _first_message_content(messages, "assistant")
if not user_message:
raise ValueError("planner_sft record requires a user message")
if not assistant_message:
raise ValueError("planner_sft record requires an assistant message")
normalized.setdefault("instruction", user_message)
normalized.setdefault("response", assistant_message)
return normalized
def normalize_critic_sft_record(record: Mapping[str, Any]) -> dict[str, Any]:
normalized = dict(record)
if "messages" in normalized:
messages = _require_messages(normalized, where="critic_sft")
normalized.setdefault("instruction", _first_message_content(messages, "user"))
normalized.setdefault("response", _first_message_content(messages, "assistant"))
images = _string_sequence(normalized.get("images"))
if images:
if len(images) < 2:
raise ValueError("critic_sft ShareGPT record requires at least two images")
normalized.setdefault("origin_image_path", images[0])
normalized.setdefault("edited_image_path", images[1])
_require_text(normalized, "origin_image_path", "critic_sft")
_require_text(normalized, "edited_image_path", "critic_sft")
normalized.setdefault("previous_image_path", normalized["origin_image_path"])
normalized.setdefault("generated_image_path", normalized["edited_image_path"])
if "previous_prompt" not in normalized and "rewritten_prompt" in normalized:
normalized["previous_prompt"] = normalized["rewritten_prompt"]
if "rewritten_prompt" not in normalized and "previous_prompt" in normalized:
normalized["rewritten_prompt"] = normalized["previous_prompt"]
return normalized
def normalize_critic_rl_record(record: Mapping[str, Any]) -> dict[str, Any]:
normalized = dict(record)
_require_text(normalized, "origin_prompt", "critic_rl")
_require_text(normalized, "origin_image_path", "critic_rl")
_require_text(normalized, "edited_image_path", "critic_rl")
previous_prompt = _first_text(
normalized.get("previous_prompt"),
normalized.get("rewritten_prompt"),
normalized.get("refine_prompt"),
)
if not previous_prompt:
raise ValueError("critic_rl record requires previous_prompt or rewritten_prompt")
normalized["previous_prompt"] = previous_prompt
normalized.setdefault("rewritten_prompt", previous_prompt)
normalized.setdefault("previous_image_path", normalized["edited_image_path"])
normalized.setdefault("generated_image_path", normalized["edited_image_path"])
normalized["ground_truth"] = normalize_ground_truth(normalized)
return normalized
def normalize_ground_truth(record: Mapping[str, Any]) -> dict[str, Any]:
raw = record.get("ground_truth", record.get("evaluation", {}))
if isinstance(raw, str):
try:
raw = json.loads(raw)
except json.JSONDecodeError as exc:
raise ValueError(f"critic_rl ground_truth is not valid JSON: {exc}") from exc
if not isinstance(raw, Mapping):
raw = {}
success = _optional_bool(
raw.get("success", raw.get("previous_step_success", record.get("previous_step_success"))))
if success is None:
raise ValueError("critic_rl ground_truth requires boolean success or previous_step_success")
return {
"success": success,
"semantics": _optional_float(raw.get("semantics", raw.get("semantic_score")), default=0.0),
"quality": _optional_float(raw.get("quality", raw.get("quality_score")), default=0.0),
}
def resolve_interleave_image_path(
value: str,
*,
image_dir: str | os.PathLike[str] = "",
validate_image_files: bool = False,
) -> str:
expanded = os.path.expanduser(value)
if not image_dir or os.path.isabs(expanded) or looks_like_uri(expanded):
resolved = expanded
else:
resolved = str(Path(os.path.expanduser(str(image_dir))) / expanded)
validate_image_path(resolved, validate_exists=validate_image_files)
return resolved
def validate_image_path(
value: str,
*,
validate_exists: bool = False,
) -> None:
if looks_like_uri(value):
return
suffix = Path(value).suffix.lower()
if suffix not in IMAGE_EXTENSIONS:
raise ValueError(f"Unsupported image file extension for InterleaveThinker path {value!r}")
if validate_exists and not Path(value).is_file():
raise FileNotFoundError(f"InterleaveThinker image path does not exist: {value}")
def looks_like_uri(value: str) -> bool:
return "://" in value or value.startswith("data:")
def _resolve_dataset_files(
data_path: str | os.PathLike[str],
kind: InterleaveDatasetKind,
) -> list[Path]:
path = Path(os.path.expanduser(str(data_path)))
if path.is_dir():
path = path / DEFAULT_FILENAMES[kind]
if not path.exists():
raise FileNotFoundError(f"InterleaveThinker {kind} data file not found: {path}")
if path.suffix not in {".json", ".jsonl"}:
raise ValueError(f"Unsupported InterleaveThinker data file: {path}")
return [path]
def _load_json_records(file_path: Path) -> list[dict[str, Any]]:
try:
if file_path.suffix == ".jsonl":
records: list[dict[str, Any]] = []
for line_number, line in enumerate(file_path.read_text(encoding="utf-8").splitlines(), start=1):
if not line.strip():
continue
raw_line = json.loads(line)
if not isinstance(raw_line, Mapping):
raise ValueError(f"{file_path}:{line_number} must contain a JSON object")
records.append(dict(raw_line))
return records
raw = json.loads(file_path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
raise ValueError(f"Malformed JSON in InterleaveThinker data file {file_path}: {exc}") from exc
if isinstance(raw, list):
if not all(isinstance(item, Mapping) for item in raw):
raise ValueError(f"{file_path} must contain only JSON objects")
return [dict(item) for item in raw]
if isinstance(raw, Mapping):
for key in ("data", "records"):
value = raw.get(key)
if isinstance(value, list):
if not all(isinstance(item, Mapping) for item in value):
raise ValueError(f"{file_path}.{key} must contain only JSON objects")
return [dict(item) for item in value]
return [dict(raw)]
raise ValueError(f"{file_path} must contain a JSON object or list of objects")
def _resolve_record_image_paths(
record: Mapping[str, Any],
*,
image_dir: str | os.PathLike[str],
validate_image_files: bool,
) -> dict[str, Any]:
normalized = dict(record)
for key in IMAGE_PATH_KEYS:
value = normalized.get(key)
if isinstance(value, str) and value:
normalized[key] = resolve_interleave_image_path(
value,
image_dir=image_dir,
validate_image_files=validate_image_files,
)
for key in IMAGE_LIST_KEYS:
value = normalized.get(key)
if isinstance(value, Sequence) and not isinstance(value, str | bytes):
normalized[key] = [
resolve_interleave_image_path(
item,
image_dir=image_dir,
validate_image_files=validate_image_files,
) if isinstance(item, str) and item else item for item in value
]
return normalized
def _require_messages(record: Mapping[str, Any], *, where: str) -> list[dict[str, Any]]:
raw = record.get("messages")
if not isinstance(raw, Sequence) or isinstance(raw, str | bytes):
raise ValueError(f"{where} record requires ShareGPT-style messages")
messages = [dict(item) for item in raw if isinstance(item, Mapping)]
if len(messages) != len(raw):
raise ValueError(f"{where} messages must be JSON objects")
return messages
def _first_message_content(messages: Sequence[Mapping[str, Any]], role: str) -> str:
for message in messages:
if message.get("role") == role:
content = message.get("content")
if isinstance(content, str):
return content
return ""
def _require_text(record: Mapping[str, Any], key: str, where: str) -> str:
value = record.get(key)
if not isinstance(value, str) or not value:
raise ValueError(f"{where} record requires {key}")
return value
def _string_sequence(value: Any) -> list[str]:
if not isinstance(value, Sequence) or isinstance(value, str | bytes):
return []
return [item for item in value if isinstance(item, str)]
def _first_text(*values: Any) -> str:
for value in values:
if isinstance(value, str) and value:
return value
return ""
def _optional_bool(value: Any) -> bool | None:
if isinstance(value, bool):
return value
return None
def _optional_float(value: Any, *, default: float) -> float:
if value is None:
return default
try:
return float(value)
except (TypeError, ValueError):
return default
__all__ = [
"DEFAULT_FILENAMES",
"IMAGE_EXTENSIONS",
"IMAGE_LIST_KEYS",
"IMAGE_PATH_KEYS",
"InterleaveDatasetKind",
"load_critic_rl_records",
"load_critic_sft_records",
"load_interleave_dataset",
"load_planner_sft_records",
"looks_like_uri",
"normalize_critic_rl_record",
"normalize_critic_sft_record",
"normalize_ground_truth",
"normalize_interleave_dataset_record",
"normalize_planner_sft_record",
"resolve_interleave_image_path",
"validate_image_path",
]
@@ -0,0 +1,142 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import json
from pathlib import Path
import pytest
from fastvideo.train.models.interleave_thinker import (
load_critic_rl_records,
load_critic_sft_records,
load_interleave_dataset,
load_planner_sft_records,
normalize_critic_rl_record,
resolve_interleave_image_path,
)
def test_load_planner_sft_records_normalizes_sharegpt_images(tmp_path: Path) -> None:
image_dir = tmp_path / "images"
data_dir = tmp_path / "data"
data_dir.mkdir()
(data_dir / "planner_sft.json").write_text(
json.dumps([{
"messages": [
{
"role": "user",
"content": "draw a cat step by step"
},
{
"role": "assistant",
"content": "<answer>{\"execution_plan\": []}</answer>"
},
],
"images": ["planner/cat_step.png"],
}]),
encoding="utf-8",
)
records = load_planner_sft_records(data_dir, image_dir=image_dir)
assert len(records) == 1
assert records[0]["instruction"] == "draw a cat step by step"
assert records[0]["response"] == '<answer>{"execution_plan": []}</answer>'
assert records[0]["images"] == [str(image_dir / "planner/cat_step.png")]
def test_load_critic_sft_records_adds_image_pair_aliases(tmp_path: Path) -> None:
image_dir = tmp_path / "images"
data_path = tmp_path / "critic_sft.json"
data_path.write_text(
json.dumps({
"records": [{
"messages": [
{
"role": "user",
"content": "evaluate this edit"
},
{
"role": "assistant",
"content": "<answer>{\"previous_step_success\": true, \"refine_prompt\": \"keep it\"}</answer>"
},
],
"images": ["before.png", "after.webp"],
"rewritten_prompt": "make the cat orange",
}]
}),
encoding="utf-8",
)
records = load_critic_sft_records(data_path, image_dir=image_dir)
assert records[0]["origin_image_path"] == str(image_dir / "before.png")
assert records[0]["edited_image_path"] == str(image_dir / "after.webp")
assert records[0]["previous_image_path"] == str(image_dir / "before.png")
assert records[0]["generated_image_path"] == str(image_dir / "after.webp")
assert records[0]["previous_prompt"] == "make the cat orange"
assert records[0]["response"].startswith("<answer>")
def test_load_critic_rl_records_normalizes_reward_fields_and_aliases(tmp_path: Path) -> None:
image_dir = tmp_path / "images"
data_path = tmp_path / "critic_rl.jsonl"
data_path.write_text(
json.dumps({
"origin_prompt": "draw a chair",
"rewritten_prompt": "a wooden chair by a window",
"origin_image_path": "chairs/original.jpg",
"edited_image_path": "chairs/edited.png",
"evaluation": {
"success": False,
"semantic_score": "6.5",
"quality_score": 7,
},
"responses": ["<answer>{}</answer>"],
}) + "\n",
encoding="utf-8",
)
records = load_critic_rl_records(data_path, image_dir=image_dir)
assert records[0]["origin_prompt"] == "draw a chair"
assert records[0]["previous_prompt"] == "a wooden chair by a window"
assert records[0]["rewritten_prompt"] == "a wooden chair by a window"
assert records[0]["origin_image_path"] == str(image_dir / "chairs/original.jpg")
assert records[0]["edited_image_path"] == str(image_dir / "chairs/edited.png")
assert records[0]["previous_image_path"] == str(image_dir / "chairs/edited.png")
assert records[0]["ground_truth"] == {
"success": False,
"semantics": 6.5,
"quality": 7.0,
}
assert records[0]["responses"] == ["<answer>{}</answer>"]
def test_load_interleave_dataset_rejects_empty_dataset(tmp_path: Path) -> None:
data_path = tmp_path / "critic_rl.jsonl"
data_path.write_text("\n", encoding="utf-8")
with pytest.raises(ValueError, match="No critic_rl records"):
load_interleave_dataset(data_path, kind="critic_rl")
def test_normalize_critic_rl_record_requires_ground_truth_success() -> None:
with pytest.raises(ValueError, match="ground_truth requires boolean success"):
normalize_critic_rl_record({
"origin_prompt": "draw a chair",
"previous_prompt": "a wooden chair",
"origin_image_path": "before.png",
"edited_image_path": "after.png",
"ground_truth": {
"semantics": 5
},
})
def test_resolve_interleave_image_path_validates_extension(tmp_path: Path) -> None:
assert resolve_interleave_image_path("image.png", image_dir=tmp_path) == str(tmp_path / "image.png")
with pytest.raises(ValueError, match="Unsupported image file extension"):
resolve_interleave_image_path("not-an-image.txt", image_dir=tmp_path)