From 811ffa648b97e1d5203a8ae7a879bc93fe47d3be Mon Sep 17 00:00:00 2001 From: iomgaa Date: Sat, 11 Jul 2026 23:05:38 -0400 Subject: [PATCH] feat(types): extend GeneratedQuestion with skill_target & difficulty_steps - Add skill_target (str | None) and difficulty_steps (int | None) fields to GeneratedQuestion dataclass with field(default=None) - Update loader.py to pass new fields from JSON (backward-compatible) - Update pools.py _q_to_dict/_dict_to_q for serialization compat - Add question_gen_v2 config section to default.yaml - Add comprehensive test coverage (7 tests) Co-Authored-By: Claude Opus 4.6 (1M context) --- app/harness/pools.py | 4 + app/question_gen/loader.py | 2 + config/default.yaml | 22 +++++ core/types.py | 6 +- tests/unit/test_types_v2.py | 173 ++++++++++++++++++++++++++++++++++++ 5 files changed, 206 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_types_v2.py diff --git a/app/harness/pools.py b/app/harness/pools.py index 809f01b..66e5769 100644 --- a/app/harness/pools.py +++ b/app/harness/pools.py @@ -139,6 +139,8 @@ def _q_to_dict(q: GeneratedQuestion) -> dict: "answer": q.answer, "source_nodes": list(q.source_nodes), "difficulty": q.difficulty, + "skill_target": q.skill_target, + "difficulty_steps": q.difficulty_steps, } @@ -160,6 +162,8 @@ def _dict_to_q(d: dict) -> GeneratedQuestion: answer=d["answer"], source_nodes=tuple(d.get("source_nodes", ())), difficulty=d.get("difficulty", "medium"), + skill_target=d.get("skill_target"), + difficulty_steps=d.get("difficulty_steps"), ) diff --git a/app/question_gen/loader.py b/app/question_gen/loader.py index d7f11b8..23845e7 100644 --- a/app/question_gen/loader.py +++ b/app/question_gen/loader.py @@ -46,6 +46,8 @@ def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]: answer=qa["answer"], source_nodes=tuple(qa.get("source_nodes", ())), difficulty=qa.get("difficulty", _LEGACY_DEFAULT_DIFFICULTY), + skill_target=qa.get("skill_target"), + difficulty_steps=qa.get("difficulty_steps"), ) ) return results diff --git a/config/default.yaml b/config/default.yaml index ff4449a..4e4b3e5 100644 --- a/config/default.yaml +++ b/config/default.yaml @@ -69,3 +69,25 @@ harness: eval_min_per_class: 2 early_stop_patience: 8 use_slow_momentum: true + +# ── 出题管线 v2 ── +question_gen_v2: + family_ratios: + retrieval: 0.30 + reasoning: 0.25 + enumeration: 0.20 + visual: 0.15 + spatial: 0.10 + gate: + blind_answer_model: "gpt-4.1-mini" + leak_test_model: "gpt-4.1-mini" + key_verify_model: "gpt-4.1-mini" + multi_true_model: "gpt-4.1-mini" + dedup_threshold: 0.85 + retry_limit: 3 + heavy_sample_rate: 0.15 + heavy_agent_model: "gpt-4.1-mini" + output_dir: "store/questions/generated-v2" + per_type: 20 # 12 类 x 20 = 240 题（设计 §3 硬约束） + concurrency: 4 + seed: 42 diff --git a/core/types.py b/core/types.py index c8838c7..54908b3 100644 --- a/core/types.py +++ b/core/types.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field @dataclass(frozen=True) @@ -41,6 +41,8 @@ class GeneratedQuestion: answer: 正确答案字母（如 "B"）。 source_nodes: 来源节点 ID 元组。 difficulty: 难度等级。 + skill_target: 目标技能标识（v2 出题管线使用，None 表示未指定）。 + difficulty_steps: 推理步数估计（v2 出题管线使用，None 表示未指定）。 """ question_id: str @@ -51,3 +53,5 @@ class GeneratedQuestion: answer: str source_nodes: tuple[str, ...] difficulty: str + skill_target: str | None = field(default=None) + difficulty_steps: int | None = field(default=None) diff --git a/tests/unit/test_types_v2.py b/tests/unit/test_types_v2.py new file mode 100644 index 0000000..b74f290 --- /dev/null +++ b/tests/unit/test_types_v2.py @@ -0,0 +1,173 @@ +"""GeneratedQuestion v2 类型扩展测试。 + +验证 skill_target 和 difficulty_steps 新字段的默认值、显式赋值、 +frozen 不可变性，以及 loader 的前后向兼容性。 +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +from core.types import GeneratedQuestion + + +class TestGeneratedQuestionV2: + """新增字段的单元测试。""" + + def test_new_fields_default_none(self) -> None: + """无显式赋值时，skill_target 和 difficulty_steps 默认为 None。""" + q = GeneratedQuestion( + question_id="q1", + video_id="v1", + task_type="TN", + question="What happens?", + options=("A. yes", "B. no", "C. maybe", "D. none"), + answer="A", + source_nodes=("n1",), + difficulty="medium", + ) + assert q.skill_target is None + assert q.difficulty_steps is None + + def test_new_fields_explicit(self) -> None: + """显式赋值时字段正确存储。""" + q = GeneratedQuestion( + question_id="q2", + video_id="v2", + task_type="AR", + question="Why did this occur?", + options=("A. cause1", "B. cause2", "C. cause3", "D. cause4"), + answer="B", + source_nodes=("n2", "n3"), + difficulty="hard", + skill_target="M1", + difficulty_steps=5, + ) + assert q.skill_target == "M1" + assert q.difficulty_steps == 5 + + def test_frozen_immutable(self) -> None: + """frozen=True 保证新字段不可修改。""" + q = GeneratedQuestion( + question_id="q3", + video_id="v3", + task_type="EN", + question="How many items?", + options=("A. 1", "B. 2", "C. 3", "D. 4"), + answer="C", + source_nodes=("n4",), + difficulty="easy", + skill_target="M2", + difficulty_steps=3, + ) + with pytest.raises(AttributeError): + q.skill_target = "M3" # type: ignore[misc] + with pytest.raises(AttributeError): + q.difficulty_steps = 7 # type: ignore[misc] + + +class TestLoaderV2Compat: + """loader.py 前后向兼容性测试。""" + + def test_load_old_format_json(self, tmp_path: Path) -> None: + """旧格式 JSON（无新字段）加载后新字段为 None。""" + from app.question_gen.loader import load_benchmark + + data = [ + { + "question_id": "old_q1", + "task_type": "TN", + "question": "What is shown?", + "options": ["A. cat", "B. dog", "C. bird", "D. fish"], + "answer": "A", + "source_nodes": ["seg_01"], + "difficulty": "easy", + } + ] + json_file = tmp_path / "video_old.json" + json_file.write_text(json.dumps(data), encoding="utf-8") + + questions = load_benchmark(tmp_path) + assert len(questions) == 1 + q = questions[0] + assert q.skill_target is None + assert q.difficulty_steps is None + assert q.video_id == "video_old" + + def test_load_new_format_json(self, tmp_path: Path) -> None: + """新格式 JSON（含新字段）正确加载。""" + from app.question_gen.loader import load_benchmark + + data = [ + { + "question_id": "new_q1", + "task_type": "AR", + "question": "Why did X happen?", + "options": ["A. r1", "B. r2", "C. r3", "D. r4"], + "answer": "B", + "source_nodes": ["seg_02", "seg_03"], + "difficulty": "hard", + "skill_target": "M3", + "difficulty_steps": 7, + } + ] + json_file = tmp_path / "video_new.json" + json_file.write_text(json.dumps(data), encoding="utf-8") + + questions = load_benchmark(tmp_path) + assert len(questions) == 1 + q = questions[0] + assert q.skill_target == "M3" + assert q.difficulty_steps == 7 + + +class TestPoolsSerializationV2: + """pools.py _q_to_dict / _dict_to_q 新字段兼容性测试。""" + + def test_roundtrip_with_new_fields(self) -> None: + """新字段经过序列化反序列化后保持一致。""" + from app.harness.pools import _dict_to_q, _q_to_dict + + q = GeneratedQuestion( + question_id="rt_q1", + video_id="v_rt", + task_type="VS", + question="Where is the object?", + options=("A. left", "B. right", "C. top", "D. bottom"), + answer="A", + source_nodes=("node_1",), + difficulty="medium", + skill_target="M5", + difficulty_steps=10, + ) + d = _q_to_dict(q) + assert d["skill_target"] == "M5" + assert d["difficulty_steps"] == 10 + + restored = _dict_to_q(d) + assert restored.skill_target == "M5" + assert restored.difficulty_steps == 10 + + def test_roundtrip_without_new_fields(self) -> None: + """旧字典（无新字段）反序列化时新字段默认为 None。""" + from app.harness.pools import _dict_to_q + + d = { + "question_id": "legacy_q1", + "video_id": "v_legacy", + "task_type": "TN", + "question": "What?", + "options": ["A", "B", "C", "D"], + "answer": "A", + "source_nodes": ["n1"], + "difficulty": "medium", + } + q = _dict_to_q(d) + assert q.skill_target is None + assert q.difficulty_steps is None