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) <noreply@anthropic.com>
This commit is contained in:
@@ -139,6 +139,8 @@ def _q_to_dict(q: GeneratedQuestion) -> dict:
|
|||||||
"answer": q.answer,
|
"answer": q.answer,
|
||||||
"source_nodes": list(q.source_nodes),
|
"source_nodes": list(q.source_nodes),
|
||||||
"difficulty": q.difficulty,
|
"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"],
|
answer=d["answer"],
|
||||||
source_nodes=tuple(d.get("source_nodes", ())),
|
source_nodes=tuple(d.get("source_nodes", ())),
|
||||||
difficulty=d.get("difficulty", "medium"),
|
difficulty=d.get("difficulty", "medium"),
|
||||||
|
skill_target=d.get("skill_target"),
|
||||||
|
difficulty_steps=d.get("difficulty_steps"),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -46,6 +46,8 @@ def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]:
|
|||||||
answer=qa["answer"],
|
answer=qa["answer"],
|
||||||
source_nodes=tuple(qa.get("source_nodes", ())),
|
source_nodes=tuple(qa.get("source_nodes", ())),
|
||||||
difficulty=qa.get("difficulty", _LEGACY_DEFAULT_DIFFICULTY),
|
difficulty=qa.get("difficulty", _LEGACY_DEFAULT_DIFFICULTY),
|
||||||
|
skill_target=qa.get("skill_target"),
|
||||||
|
difficulty_steps=qa.get("difficulty_steps"),
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
return results
|
return results
|
||||||
|
|||||||
@@ -69,3 +69,25 @@ harness:
|
|||||||
eval_min_per_class: 2
|
eval_min_per_class: 2
|
||||||
early_stop_patience: 8
|
early_stop_patience: 8
|
||||||
use_slow_momentum: true
|
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
|
||||||
|
|||||||
+5
-1
@@ -2,7 +2,7 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass, field
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
@@ -41,6 +41,8 @@ class GeneratedQuestion:
|
|||||||
answer: 正确答案字母(如 "B")。
|
answer: 正确答案字母(如 "B")。
|
||||||
source_nodes: 来源节点 ID 元组。
|
source_nodes: 来源节点 ID 元组。
|
||||||
difficulty: 难度等级。
|
difficulty: 难度等级。
|
||||||
|
skill_target: 目标技能标识(v2 出题管线使用,None 表示未指定)。
|
||||||
|
difficulty_steps: 推理步数估计(v2 出题管线使用,None 表示未指定)。
|
||||||
"""
|
"""
|
||||||
|
|
||||||
question_id: str
|
question_id: str
|
||||||
@@ -51,3 +53,5 @@ class GeneratedQuestion:
|
|||||||
answer: str
|
answer: str
|
||||||
source_nodes: tuple[str, ...]
|
source_nodes: tuple[str, ...]
|
||||||
difficulty: str
|
difficulty: str
|
||||||
|
skill_target: str | None = field(default=None)
|
||||||
|
difficulty_steps: int | None = field(default=None)
|
||||||
|
|||||||
@@ -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
|
||||||
Reference in New Issue
Block a user