From dea8a7d3f6848a4a9bc6a33236c81a677e45c1d0 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 04:41:25 -0400 Subject: [PATCH] =?UTF-8?q?feat(question=5Fgen):=20load=5Fbenchmark=20?= =?UTF-8?q?=E2=80=94=20benchmark=20JSON=20=E5=8A=A0=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从 JSON 目录 glob *.json 加载题目，stem 作 video_id。 legacy schema 无 difficulty 字段时赋 _LEGACY_DEFAULT_DIFFICULTY 常量。 options/source_nodes 转 tuple 配合 frozen dataclass。 Co-Authored-By: Claude Opus 4.6 (1M context) --- app/question_gen/loader.py | 50 +++++++++++ tests/unit/test_question_loader.py | 129 +++++++++++++++++++++++++++++ 2 files changed, 179 insertions(+) create mode 100644 app/question_gen/loader.py create mode 100644 tests/unit/test_question_loader.py diff --git a/app/question_gen/loader.py b/app/question_gen/loader.py new file mode 100644 index 0000000..6febe36 --- /dev/null +++ b/app/question_gen/loader.py @@ -0,0 +1,50 @@ +"""题目加载与分层采样。 + +从 benchmark JSON 目录加载题目，提供按对错比例的分层采样。 +对应训练循环中的 DataLoader 角色。 +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +from core.types import GeneratedQuestion + +if TYPE_CHECKING: + from pathlib import Path + +_LEGACY_DEFAULT_DIFFICULTY = "medium" + + +def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]: + """从 benchmark JSON 目录加载题目列表。 + + 每个 JSON 文件以文件名（不含扩展名）作为 video_id， + 文件内容为题目数组。 + + 参数: + questions_dir: 包含 *.json 文件的目录路径。 + + 返回: + 按文件名排序加载的题目列表。 + """ + results: list[GeneratedQuestion] = [] + for path in sorted(questions_dir.glob("*.json")): + video_id = path.stem + with open(path, encoding="utf-8") as f: + qa_list: list[dict] = json.load(f) + for qa in qa_list: + results.append( + GeneratedQuestion( + question_id=qa["question_id"], + video_id=video_id, + task_type=qa["task_type"], + question=qa["question"], + options=tuple(qa["options"]), + answer=qa["answer"], + source_nodes=tuple(qa.get("source_nodes", ())), + difficulty=qa.get("difficulty", _LEGACY_DEFAULT_DIFFICULTY), + ) + ) + return results diff --git a/tests/unit/test_question_loader.py b/tests/unit/test_question_loader.py new file mode 100644 index 0000000..7df8963 --- /dev/null +++ b/tests/unit/test_question_loader.py @@ -0,0 +1,129 @@ +"""app/question_gen/loader.py 单元测试。""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from app.question_gen.loader import load_benchmark +from core.types import GeneratedQuestion + + +@pytest.fixture() +def benchmark_dir(tmp_path: Path) -> Path: + """创建包含 benchmark JSON 的临时目录。""" + data = [ + { + "question_id": "1-1", + "task_type": "Action Reasoning", + "question": "What happened?", + "options": ["A. X", "B. Y", "C. Z", "D. W"], + "answer": "A", + }, + { + "question_id": "1-2", + "task_type": "OCR Problems", + "question": "What text is shown?", + "options": ["A. Hello", "B. World", "C. Foo", "D. Bar"], + "answer": "B", + }, + ] + (tmp_path / "video_abc.json").write_text(json.dumps(data), encoding="utf-8") + return tmp_path + + +class TestLoadBenchmark: + def test_loads_questions_from_json(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert len(questions) == 2 + + def test_video_id_from_filename(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert all(q.video_id == "video_abc" for q in questions) + + def test_fields_mapped_correctly(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + q = questions[0] + assert q.question_id == "1-1" + assert q.task_type == "Action Reasoning" + assert q.question == "What happened?" + assert q.options == ("A. X", "B. Y", "C. Z", "D. W") + assert q.answer == "A" + + def test_options_is_tuple(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert isinstance(questions[0].options, tuple) + + def test_source_nodes_is_empty_tuple(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert questions[0].source_nodes == () + + def test_difficulty_defaults_to_medium_for_legacy(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert questions[0].difficulty == "medium" + + def test_difficulty_from_json_when_present(self, tmp_path: Path) -> None: + data = [ + { + "question_id": "2-1", + "task_type": "OCR Problems", + "question": "Q?", + "options": ["A. 1", "B. 2", "C. 3", "D. 4"], + "answer": "C", + "difficulty": "hard", + } + ] + (tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8") + questions = load_benchmark(tmp_path) + assert questions[0].difficulty == "hard" + + def test_empty_directory_returns_empty_list(self, tmp_path: Path) -> None: + questions = load_benchmark(tmp_path) + assert questions == [] + + def test_sorted_by_filename(self, tmp_path: Path) -> None: + for name in ["z_video.json", "a_video.json"]: + data = [ + { + "question_id": f"{name}-1", + "task_type": "T", + "question": "Q?", + "options": ["A", "B", "C", "D"], + "answer": "A", + } + ] + (tmp_path / name).write_text(json.dumps(data), encoding="utf-8") + questions = load_benchmark(tmp_path) + assert questions[0].video_id == "a_video" + assert questions[1].video_id == "z_video" + + def test_returns_generated_question_instances(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert all(isinstance(q, GeneratedQuestion) for q in questions) + + def test_loads_real_benchmark(self) -> None: + """使用真实 benchmark 数据验证加载正确性。""" + real_dir = Path("store/questions/benchmarks/Video-MME") + if not real_dir.exists(): + pytest.skip("真实 benchmark 数据不存在") + questions = load_benchmark(real_dir) + assert len(questions) > 0 + for q in questions: + assert isinstance(q, GeneratedQuestion) + assert len(q.options) == 4 + assert q.answer in ("A", "B", "C", "D") + + def test_malformed_json_raises(self, tmp_path: Path) -> None: + """非法 JSON 文件应抛出 json.JSONDecodeError。""" + (tmp_path / "bad.json").write_text("not valid json{{{", encoding="utf-8") + with pytest.raises(json.JSONDecodeError): + load_benchmark(tmp_path) + + def test_missing_required_field_raises(self, tmp_path: Path) -> None: + """缺少必需字段（如 question_id）应抛出 KeyError。""" + data = [{"task_type": "T", "question": "Q?", "options": ["A"], "answer": "A"}] + (tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8") + with pytest.raises(KeyError): + load_benchmark(tmp_path)