diff --git a/core/types.py b/core/types.py index 48f1ec0..fd32aad 100644 --- a/core/types.py +++ b/core/types.py @@ -49,6 +49,11 @@ class GeneratedQuestion: skill_target: 目标技能标识(v2 出题管线使用,None 表示未指定)。 difficulty_steps: 推理步数估计(v2 出题管线使用,None 表示未指定)。 sub_pattern: 出题子模式标识(AR 特化策略使用,None 表示无)。 + unit_id: 所属题目单元标识;缺省时 __post_init__ 回填为 pair_id 或 + question_id,保证 single 题的 unit_id 等于自身 question_id。 + pair_id: 孪生对标识;同一对的 original/mirror 共享该值,None 表示非配对题。 + question_role: 在单元内的角色("single" | "pair_original" | "pair_mirror")。 + flip_axis: 孪生对的翻转轴(如 "before_after"),None 表示无翻转。 """ question_id: str @@ -63,6 +68,78 @@ class GeneratedQuestion: skill_target: str | None = field(default=None) difficulty_steps: int | None = field(default=None) sub_pattern: str | None = field(default=None) + unit_id: str = "" + pair_id: str | None = field(default=None) + question_role: str = "single" + flip_axis: str | None = field(default=None) + + def __post_init__(self) -> None: + """回填 unit_id:缺省时取 pair_id(配对题)或 question_id(single 题)。 + + frozen dataclass 无法直接赋值,故通过 object.__setattr__ 绕过不可变约束。 + """ + if not self.unit_id: + object.__setattr__(self, "unit_id", self.pair_id or self.question_id) + + +@dataclass(frozen=True) +class QuestionUnit: + """题目单元:贯穿评测/训练 harness 的最小不可分调度契约实体。 + + single 题为 1 题单元,AR pair 孪生对为 2 题单元(original + mirror), + 两条题目必须作为整体被批处理/推理/评测,保证配对指标(collapse 等)可算。 + frozen=True 确保单元不可变。 + + 属性: + kind: 单元类型("single" | "pair")。 + unit_id: 单元唯一标识;single 取题目 unit_id,pair 取共享 pair_id。 + task_type: 单元题型;pair 内两题题型必须一致。 + questions: 单元内题目元组(single 为 1 条,pair 为 2 条)。 + unit_hash: P/Q payload 合成 hash,用于断点续跑失效检测(T11 消费)。 + """ + + kind: str + unit_id: str + task_type: str + questions: tuple[GeneratedQuestion, ...] + unit_hash: str = "" + + @property + def size(self) -> int: + """单元内题目数量(single=1,pair=2)。""" + return len(self.questions) + + @classmethod + def from_single(cls, q: GeneratedQuestion) -> QuestionUnit: + """由单条题目构造 single 单元。 + + 参数: + q: 待封装的题目。 + + 返回: + kind="single" 的单元,unit_id 取 q.unit_id。 + """ + return cls("single", q.unit_id, q.task_type, (q,)) + + @classmethod + def from_pair(cls, original: GeneratedQuestion, mirror: GeneratedQuestion) -> QuestionUnit: + """由孪生对(original + mirror)构造 pair 单元。 + + 参数: + original: 原始题(question_role="pair_original")。 + mirror: 镜像题(question_role="pair_mirror")。 + + 返回: + kind="pair" 的单元,unit_id 取共享 pair_id。 + + 关键实现: + 断言两题共享非空 pair_id、且 video_id/task_type/flip_axis 一致, + 确保只有合法孪生对才能聚合成对,非法配对直接报错而非静默兜底。 + """ + assert original.pair_id and original.pair_id == mirror.pair_id + assert original.video_id == mirror.video_id and original.task_type == mirror.task_type + assert original.flip_axis == mirror.flip_axis + return cls("pair", original.pair_id, original.task_type, (original, mirror)) @dataclass(frozen=True) diff --git a/tests/unit/test_question_unit.py b/tests/unit/test_question_unit.py new file mode 100644 index 0000000..65a5e53 --- /dev/null +++ b/tests/unit/test_question_unit.py @@ -0,0 +1,51 @@ +"""QuestionUnit 领域实体与 GeneratedQuestion 字段扩展的单元测试。 + +覆盖: + - 非 AR single 题的默认值向后兼容(新增字段不改变现有构造点行为)。 + - AR pair 孪生对经 QuestionUnit.from_pair 聚合后携带两条题目。 +""" + +from core.types import GeneratedQuestion, QuestionUnit + + +def _q(qid, role="single", pair_id=None, flip_axis=None): + """构造一条 GeneratedQuestion 测试样本。""" + return GeneratedQuestion( + question_id=qid, + video_id="v", + task_type="Action Recognition", + question="?", + options=("A. a", "B. b", "C. c", "D. d"), + answer="A", + source_nodes=("n1",), + difficulty="hard", + unit_id=pair_id or qid, + pair_id=pair_id, + question_role=role, + flip_axis=flip_axis, + ) + + +def test_single_defaults_backward_compatible(): + """不传新字段时,single 题默认值应与旧行为一致。""" + q = GeneratedQuestion( + question_id="q", + video_id="v", + task_type="X", + question="?", + options=("A. a", "B. b", "C. c", "D. d"), + answer="A", + source_nodes=("n1",), + difficulty="hard", + ) + assert q.question_role == "single" + assert q.pair_id is None and q.flip_axis is None and q.unit_id == "q" + + +def test_pair_unit_carries_two_questions(): + """孪生对经 from_pair 聚合后 kind=pair、size=2、unit_id=pair_id。""" + p = _q("q_o", "pair_original", "pid", "before_after") + m = _q("q_m", "pair_mirror", "pid", "before_after") + u = QuestionUnit.from_pair(p, m) + assert u.kind == "pair" and u.size == 2 and u.unit_id == "pid" + assert {qq.question_id for qq in u.questions} == {"q_o", "q_m"}