docs(datasets): document Daily-Omni and Video-R1-260k prep for Qwen3-Omni RL

datasets/daily_omni_av/README.md

@@ -0,0 +1,140 @@
+# Daily-Omni Audio/Video MCQA RL Dataset
+
+Audio-visual multiple-choice QA prompts for RL training of Qwen3-Omni Thinker with
+`use_audio_in_video: true`. Each sample is one video (whose embedded audio track carries
+part of the answer) plus a 4-way A/B/C/D question.
+
+Used by:
+
+- `examples/ar/qwen3_omni_audio_video_gspo_lora_vllm_omni_1x8.yaml`
+- `examples/ar/qwen3_omni_audio_video_gspo_lora_vllm_omni_1x4.yaml`
+
+## Source
+
+- **Daily-Omni**: [liarliar/Daily-Omni](https://huggingface.co/datasets/liarliar/Daily-Omni)
+  - Code: [Lliar-liar/Daily-Omni](https://github.com/Lliar-liar/Daily-Omni)
+  - Paper: [arXiv:2505.17862](https://arxiv.org/abs/2505.17862) — *Daily-Omni: Towards Audio-Visual Reasoning with Temporal Alignment across Modalities*
+  - License: **CC BY-NC-SA 4.0** (non-commercial, share-alike)
+- 684 YouTube videos (375 × 30 s, 309 × 60 s) and 1,197 four-way multiple-choice QA pairs.
+- Each MP4 contains H.264 video and AAC audio; the release also includes a standalone `.wav`.
+
+> **Daily-Omni is an evaluation benchmark with no official train/val split.** Hugging Face's
+> `train` label is only the default name for the single `qa.json`. Training on a subset makes
+> the resulting `eval/acc` incomparable with the published benchmark; treat this recipe as an
+> audio-in-video RL smoke test.
+
+## Download
+
+```bash
+hf download liarliar/Daily-Omni --repo-type dataset --local-dir /path/to/Daily-Omni
+
+# Videos.tar is an uncompressed tar — do NOT pass -z.
+tar -xf /path/to/Daily-Omni/Videos.tar -C /path/to/Daily-Omni
+```
+
+The download is ~3.9 GB; budget ~8 GB while the tar and extracted tree coexist.
+After extraction:
+
+```
+/path/to/Daily-Omni/
+├── qa.json
+└── Videos/<video_id>/<video_id>_video.mp4
+                     /<video_id>_audio.wav
+```
+
+`qa.json` rows use capitalized keys:
+
+```json
+{
+  "Question": "What visual elements were displayed immediately after ...?",
+  "Choice": ["A. ...", "B. ...", "C. ...", "D. ..."],
+  "Answer": "B",
+  "video_id": "Ec_lQgZ9wlg",
+  "Type": "Event Sequence",
+  "video_duration": "30s"
+}
+```
+
+## Cook
+
+```bash
+python datasets/daily_omni_av/convert_daily_omni_dataset_format_to_unirl.py \
+  --qa-json /path/to/Daily-Omni/qa.json \
+  --out-dir datasets/daily_omni_av
+```
+
+`--videos-root` defaults to `Videos/` next to `qa.json`; point it elsewhere if you unpacked
+the tar somewhere else. `--val-ratio` (default `0.1`) and `--seed` (default `42`) control the
+holdout. Rows whose MP4 is missing on disk are dropped and counted — pass
+`--keep-missing` to emit them anyway, which is useful for a dry run before the tar finishes
+extracting. An unparseable ground truth aborts with the offending `qa.json[index]`.
+
+Two details matter:
+
+- **It splits by `video_id`, not by row.** Several QA pairs share one clip; a row-level split
+  leaks the same video into train and val and inflates `eval/acc`.
+- **It appends an explicit answer-format instruction.** Both recipes use
+  `require_answer_phrase: true`, so a bare `B` scores 0; the reply must contain an
+  `answer is B` phrase or an `<answer>B</answer>` tag.
+
+## Format
+
+Each output line:
+
+```json
+{
+  "prompt": "Question text\nA. ...\nB. ...\nC. ...\nD. ...\nWatch the video and listen to its audio, then answer the multiple-choice question.\nReason step by step, then end your reply with the exact phrase: The answer is [X]",
+  "prompt_id": "daily_omni_av:train:000042:Ec_lQgZ9wlg",
+  "media_refs": [{"modality": "video", "role": "prompt", "uri": "/abs/path/Ec_lQgZ9wlg_video.mp4"}],
+  "metadata": {"answer": "B", "video_id": "Ec_lQgZ9wlg", "qa_type": "Event Sequence"}
+}
+```
+
+## Why cook it this way
+
+- **One video ref, no audio ref.** `MultimodalRLDataSource` does not support audio media refs.
+  Qwen3-Omni instead reads the MP4's AAC track because the recipes set
+  `use_audio_in_video: true`; the standalone `_audio.wav` is not referenced.
+- **`role: "prompt"`, not `"condition"`.** `(video, condition)` decodes the clip into a frame
+  tensor for diffusion V2V. `(video, prompt)` passes the URI through to the Qwen3-Omni
+  conversation builder, which is what an AR prompt video needs. A batch may not mix the two.
+- **Absolute URIs.** The converter resolves every clip against `--videos-root` and writes the
+  absolute path. This avoids dependence on the training process's working directory, but the
+  output is machine-local; rerun the converter if the video tree moves.
+- **`metadata.answer` is a single uppercase letter.** `MCExactMatchRewardScorer` reads
+  `metadata["answer"]` and nothing else, and the converter unwraps `[B]` → `B` and validates
+  membership in `{A,B,C,D}` up front. This matters because the scorer **fails silently**: a
+  missing or malformed `answer` returns 0.0 rather than raising, which looks identical to a
+  model that never gets anything right.
+- **Unique `prompt_id`.** It becomes the root `sample_id` (`prompt:{id}:sample:0`) and is what
+  GSPO groups siblings by; duplicates inside one batch raise `ValueError`. The
+  `{split}:{index}:{video_id}` form stays unique even when several questions share a clip.
+
+## Usage
+
+```yaml
+data_source:
+  _target_: unirl.data.data_source.MultimodalRLDataSource
+  args:
+    run:
+      data_path: datasets/daily_omni_av/train.jsonl
+      eval_data_path: datasets/daily_omni_av/val.jsonl
+      seed: 42
+    algorithm:
+      prompts_per_rollout: ${batch_size}
+```
+
+```bash
+QWEN3_OMNI_PATH=/path/to/Qwen3-Omni-30B-A3B-Instruct \
+DATA_PATH=datasets/daily_omni_av/train.jsonl \
+EVAL_DATA_PATH=datasets/daily_omni_av/val.jsonl \
+ENTRY=train_ar bash examples/run_experiment_single_node.sh \
+  ar/qwen3_omni_audio_video_gspo_lora_vllm_omni_1x8
+```
+
+## Notes
+
+- `video_fps: 2.0` × `video_max_frames: 64` caps a clip at 32 s of sampled frames, so 60 s
+  videos are subsampled. Frames plus question must fit `max_prompt_length: 16384`; lower
+  `video_max_pixels` before lowering `video_max_frames` if you overflow.
+- Non-commercial license: do not ship models trained on this without checking CC BY-NC-SA 4.0.

datasets/daily_omni_av/convert_daily_omni_dataset_format_to_unirl.py

@@ -1,13 +1,25 @@
 #!/usr/bin/env python3
 """Convert Daily-Omni audio/video MCQA data to UniRL JSONL.
 
-The output contains only a video media reference. Qwen3-Omni extracts the
+Reads the official Daily-Omni release — ``qa.json`` plus the ``Videos/`` tree unpacked from
+``Videos.tar`` (https://huggingface.co/datasets/liarliar/Daily-Omni) — and writes the jsonl
+that ``unirl.data.data_source.MultimodalRLDataSource`` expects::
+
+    {"prompt": <question + choices + answer-format instruction>,
+     "prompt_id": "daily_omni_av:<split>:<index>:<video_id>",
+     "media_refs": [{"modality": "video", "role": "prompt", "uri": "<abs>/<id>_video.mp4"}],
+     "metadata": {"answer": "<A|B|C|D>", "video_id": <str>, "qa_type": <str>}}
+
+Only a video media ref is emitted: the loader has no audio role, and Qwen3-Omni extracts the
 embedded audio track from that same file when ``use_audio_in_video=true``.
 
+Daily-Omni is published as a benchmark with no official split, so the split is carved here by
+``video_id`` — several questions share one clip, and splitting per row would leak the same
+video into both sides.
+
 Example:
     python datasets/daily_omni_av/convert_daily_omni_dataset_format_to_unirl.py \
-      --train-input /path/to/daily_omni_av_train.jsonl \
-      --val-input /path/to/daily_omni_av_val.jsonl \
+      --qa-json /path/to/Daily-Omni/qa.json \
       --out-dir datasets/daily_omni_av
 """
 
@@ -16,111 +28,123 @@
 import argparse
 import json
 import os
-from typing import Any, Dict, Iterable
-
-
-def _user_text(row: Dict[str, Any]) -> str:
-    messages = row.get("prompt") or []
-    user = next((message for message in reversed(messages) if message.get("role") == "user"), None)
-    if user is None:
-        raise ValueError("row has no user message")
-    content = user.get("content", "")
-    if isinstance(content, str):
-        return content.strip()
-    for part in content:
-        if isinstance(part, dict) and part.get("type") == "text":
-            return str(part.get("text", "")).strip()
-    raise ValueError("user message has no text content")
-
-
-def _video_uri(row: Dict[str, Any]) -> str:
-    videos = row.get("videos") or []
-    if videos:
-        first = videos[0]
-        uri = first.get("video") if isinstance(first, dict) else first
-        if uri:
-            return os.path.abspath(os.path.expanduser(str(uri)))
-
-    messages = row.get("prompt") or []
-    for message in reversed(messages):
-        content = message.get("content", [])
-        if not isinstance(content, list):
-            continue
-        for part in content:
-            if isinstance(part, dict) and part.get("type") == "video" and part.get("video"):
-                return os.path.abspath(os.path.expanduser(str(part["video"])))
-    raise ValueError("row has no video path")
-
-
-def _answer(row: Dict[str, Any]) -> str:
-    answer = str((row.get("reward_model") or {}).get("ground_truth", "")).strip().upper()
+import random
+from typing import Any, Dict, List
+
+# The recipes score with MCExactMatchSpec(require_answer_phrase=True), so append the expected
+# format instead of relying on the model to infer it from the answer choices.
+ANSWER_INSTRUCTION = (
+    "Watch the video and listen to its audio, then answer the multiple-choice question.\n"
+    "Reason step by step, then end your reply with the exact phrase: The answer is [X]"
+)
+
+
+def _answer(raw: Any) -> str:
+    answer = str(raw).strip().upper()
     if len(answer) == 3 and answer[0] == "[" and answer[-1] == "]":
         answer = answer[1]
     if answer not in {"A", "B", "C", "D"}:
         raise ValueError(f"invalid ground-truth answer: {answer!r}")
     return answer
 
 
-def _iter_records(path: str, split: str, keep_missing: bool) -> Iterable[Dict[str, Any]]:
-    with open(path, encoding="utf-8") as source:
-        output_index = 0
-        for source_index, line in enumerate(source):
-            if not line.strip():
+def _prompt(row: Dict[str, Any]) -> str:
+    question = str(row["Question"]).strip()
+    choices = [str(choice).strip() for choice in row["Choice"]]
+    if not question or not choices:
+        raise ValueError("row has no question or no choices")
+    return "\n".join([question, *choices, ANSWER_INSTRUCTION])
+
+
+def _group_by_video(qa_json: str, videos_root: str, keep_missing: bool) -> Dict[str, List[Dict[str, Any]]]:
+    with open(qa_json, encoding="utf-8") as source:
+        rows = json.load(source)
+
+    grouped: Dict[str, List[Dict[str, Any]]] = {}
+    missing = 0
+    for index, row in enumerate(rows):
+        try:
+            video_id = str(row["video_id"])
+            video = os.path.abspath(os.path.join(videos_root, video_id, f"{video_id}_video.mp4"))
+            if not keep_missing and not os.path.isfile(video):
+                missing += 1
                 continue
-            row = json.loads(line)
-            try:
-                video = _video_uri(row)
-                if not keep_missing and not os.path.isfile(video):
-                    continue
-                extra = row.get("extra_info") or {}
-                video_id = str(extra.get("video_id") or source_index)
-                yield {
-                    "prompt": _user_text(row),
-                    "prompt_id": f"daily_omni_av:{split}:{output_index:06d}:{video_id}",
+            grouped.setdefault(video_id, []).append(
+                {
+                    "prompt": _prompt(row),
                     "media_refs": [{"modality": "video", "role": "prompt", "uri": video}],
                     "metadata": {
-                        "answer": _answer(row),
+                        "answer": _answer(row["Answer"]),
                         "video_id": video_id,
-                        "qa_type": extra.get("qa_type"),
+                        "qa_type": row.get("Type"),
                     },
                 }
-                output_index += 1
-            except (KeyError, TypeError, ValueError) as exc:
-                raise ValueError(f"{path}:{source_index + 1}: {exc}") from exc
+            )
+        except (KeyError, TypeError, ValueError) as exc:
+            raise ValueError(f"{qa_json}[{index}]: {exc}") from exc
+
+    if missing:
+        print(f"[stats] skipped {missing} rows whose mp4 is not on disk (pass --keep-missing to emit them)")
+    return grouped
+
+
+def _split(grouped: Dict[str, List[Dict[str, Any]]], val_ratio: float, seed: int) -> Dict[str, List[Dict[str, Any]]]:
+    video_ids = sorted(grouped)
+    random.Random(seed).shuffle(video_ids)
+
+    n_val = 0
+    if val_ratio > 0 and len(video_ids) > 1:
+        n_val = max(1, min(int(len(video_ids) * val_ratio), len(video_ids) - 1))
+    val_ids = set(video_ids[:n_val])
+
+    splits: Dict[str, List[Dict[str, Any]]] = {"train": [], "val": []}
+    for video_id in video_ids:
+        name = "val" if video_id in val_ids else "train"
+        for payload in grouped[video_id]:
+            splits[name].append(
+                {
+                    "prompt": payload["prompt"],
+                    "prompt_id": f"daily_omni_av:{name}:{len(splits[name]):06d}:{video_id}",
+                    "media_refs": payload["media_refs"],
+                    "metadata": payload["metadata"],
+                }
+            )
+    return splits
 
 
-def _write_split(input_path: str, output_path: str, split: str, keep_missing: bool) -> int:
-    count = 0
+def _write_split(records: List[Dict[str, Any]], output_path: str) -> None:
     with open(output_path, "w", encoding="utf-8") as output:
-        for record in _iter_records(input_path, split, keep_missing):
+        for record in records:
             output.write(json.dumps(record, ensure_ascii=False) + "\n")
-            count += 1
-    print(f"[write] {output_path}: {count} rows")
-    return count
+    print(f"[write] {output_path}: {len(records)} rows")
 
 
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
-    parser.add_argument("--train-input", required=True, help="Daily-Omni training JSONL")
-    parser.add_argument("--val-input", help="Daily-Omni validation JSONL")
+    parser.add_argument("--qa-json", required=True, help="official Daily-Omni qa.json")
+    parser.add_argument("--videos-root", help="unpacked Videos/ tree (default: Videos/ next to qa.json)")
     parser.add_argument("--out-dir", default="datasets/daily_omni_av")
+    parser.add_argument("--val-ratio", type=float, default=0.1, help="fraction of videos held out for val.jsonl")
+    parser.add_argument("--seed", type=int, default=42, help="shuffle seed (deterministic split)")
     parser.add_argument("--keep-missing", action="store_true", help="keep rows whose video is unavailable locally")
     args = parser.parse_args()
 
-    os.makedirs(args.out_dir, exist_ok=True)
-    _write_split(
-        args.train_input,
-        os.path.join(args.out_dir, "train.jsonl"),
-        "train",
-        args.keep_missing,
+    qa_json = os.path.abspath(os.path.expanduser(args.qa_json))
+    videos_root = (
+        os.path.abspath(os.path.expanduser(args.videos_root))
+        if args.videos_root
+        else os.path.join(os.path.dirname(qa_json), "Videos")
     )
-    if args.val_input:
-        _write_split(
-            args.val_input,
-            os.path.join(args.out_dir, "val.jsonl"),
-            "val",
-            args.keep_missing,
-        )
+
+    grouped = _group_by_video(qa_json, videos_root, args.keep_missing)
+    if not grouped:
+        raise SystemExit(f"No usable rows. Is {videos_root} unpacked? (tar -xf Videos.tar)")
+
+    splits = _split(grouped, args.val_ratio, args.seed)
+    os.makedirs(args.out_dir, exist_ok=True)
+    _write_split(splits["train"], os.path.join(args.out_dir, "train.jsonl"))
+    if splits["val"]:
+        _write_split(splits["val"], os.path.join(args.out_dir, "val.jsonl"))
 
 
 if __name__ == "__main__":