diff --git a/app/question_gen/synthesizer.py b/app/question_gen/synthesizer.py new file mode 100644 index 0000000..9aedb29 --- /dev/null +++ b/app/question_gen/synthesizer.py @@ -0,0 +1,68 @@ +"""赛题合成核心逻辑 — 节点采样、prompt 构造、去重。 + +纯函数为主，异步编排仅 generate_one。 +通过 DI 接收 VLMProvider / EmbeddingProvider，不 import adapters/。 +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class AnchorContext: + """锚节点上下文——生成单道题所需的全部素材。 + + 属性: + node_id: 锚节点 ID。 + card_text: 锚节点 card 序列化文本。 + frame_paths: 帧图片路径列表。 + subtitle: 对应字幕（可空）。 + distractor_texts: 同视频其他节点摘要（供 VLM 生成干扰项）。 + """ + + node_id: str + card_text: str + frame_paths: list[str] + subtitle: str + distractor_texts: list[str] + + +@dataclass(frozen=True) +class TaskTypeSpec: + """题型的生成规格。 + + 属性: + level: 锚定层级（"L3" / "L2" / "L1" / "L1-L2"）。 + needs_frames: 是否必须提供帧图。 + frame_count: 帧数范围描述（如 "1", "2-3", "0-1"）。 + context_fields: 需要提取的 card 字段元组。 + """ + + level: str + needs_frames: bool + frame_count: str + context_fields: tuple[str, ...] + + +# --------------------------------------------------------------------------- +# 12 种 Video-MME 题型 → 树层级 + 生成规格映射 +# --------------------------------------------------------------------------- + +TASK_TYPE_LEVEL_MAP: dict[str, TaskTypeSpec] = { + # --- L3 单帧题型 --- + "Object Recognition": TaskTypeSpec("L3", True, "1", ("frame_summary",)), + "Attribute Perception": TaskTypeSpec("L3", True, "1", ("frame_summary",)), + "OCR Problems": TaskTypeSpec("L3", True, "1", ("frame_summary",)), + "Spatial Reasoning": TaskTypeSpec("L3", True, "1", ("frame_summary", "spatial_layout")), + "Spatial Perception": TaskTypeSpec("L3", True, "1", ("frame_summary",)), + # --- L2 多帧 / 事件级题型 --- + "Action Recognition": TaskTypeSpec("L2", True, "2-3", ("event_description",)), + "Action Reasoning": TaskTypeSpec("L2", True, "2-3", ("event_description",)), + "Counting Problem": TaskTypeSpec("L2", True, "2-3", ("event_description",)), + "Temporal Perception": TaskTypeSpec("L2", False, "0-1", ("event_description", "time_range")), + # --- L1 / 跨层级题型 --- + "Temporal Reasoning": TaskTypeSpec("L1", True, "per-L2", ("scene_summary",)), + "Information Synopsis": TaskTypeSpec("L1", True, "per-L2", ("scene_summary",)), + "Object Reasoning": TaskTypeSpec("L1-L2", True, "per-L2", ("event_description",)), +} diff --git a/tests/unit/test_synthesizer.py b/tests/unit/test_synthesizer.py new file mode 100644 index 0000000..579b2e6 --- /dev/null +++ b/tests/unit/test_synthesizer.py @@ -0,0 +1,134 @@ +"""synthesizer 模块单元测试 — AnchorContext + 题型映射常量。""" + +from __future__ import annotations + +import dataclasses + +from app.question_gen.synthesizer import TASK_TYPE_LEVEL_MAP, AnchorContext, TaskTypeSpec + +ALL_12_TYPES = [ + "Object Recognition", + "Attribute Perception", + "OCR Problems", + "Spatial Reasoning", + "Spatial Perception", + "Action Recognition", + "Action Reasoning", + "Counting Problem", + "Temporal Perception", + "Temporal Reasoning", + "Information Synopsis", + "Object Reasoning", +] + + +class TestTaskTypeLevelMap: + """TASK_TYPE_LEVEL_MAP 覆盖性与结构测试。""" + + def test_covers_all_12_types(self) -> None: + """映射表必须覆盖全部 12 种 Video-MME 题型。""" + assert set(TASK_TYPE_LEVEL_MAP.keys()) == set(ALL_12_TYPES) + + def test_no_extra_types(self) -> None: + """映射表不得包含 12 种标准题型之外的条目。""" + assert len(TASK_TYPE_LEVEL_MAP) == 12 + + def test_all_values_are_task_type_spec(self) -> None: + """每个映射值必须是 TaskTypeSpec 实例。""" + for task_type, spec in TASK_TYPE_LEVEL_MAP.items(): + assert isinstance(spec, TaskTypeSpec), f"{task_type} 映射值类型错误: {type(spec)}" + + def test_level_values_valid(self) -> None: + """每个 spec 的 level 必须是合法层级标识。""" + valid_levels = {"L1", "L2", "L3", "L1-L2"} + for task_type, spec in TASK_TYPE_LEVEL_MAP.items(): + assert spec.level in valid_levels, ( + f"{task_type} 层级 '{spec.level}' 不在 {valid_levels}" + ) + + def test_context_fields_non_empty(self) -> None: + """每个 spec 的 context_fields 至少有一个字段。""" + for task_type, spec in TASK_TYPE_LEVEL_MAP.items(): + assert len(spec.context_fields) >= 1, f"{task_type} 的 context_fields 为空" + + +class TestAnchorContext: + """AnchorContext 数据类测试。""" + + def test_frozen(self) -> None: + """AnchorContext 是不可变的。""" + ctx = AnchorContext( + node_id="L3_001", + card_text="A person walks into a room", + frame_paths=["/data/frames/001.jpg"], + subtitle="Hello there", + distractor_texts=["A car drives by"], + ) + assert ctx.node_id == "L3_001" + assert ctx.card_text == "A person walks into a room" + assert ctx.frame_paths == ["/data/frames/001.jpg"] + assert ctx.subtitle == "Hello there" + assert ctx.distractor_texts == ["A car drives by"] + + def test_mutation_raises(self) -> None: + """frozen dataclass 拒绝赋值修改。""" + ctx = AnchorContext( + node_id="L3_001", + card_text="test", + frame_paths=["a.jpg"], + subtitle="", + distractor_texts=["other node"], + ) + try: + ctx.node_id = "L3_002" # type: ignore[misc] + raise AssertionError("应抛出 FrozenInstanceError") + except dataclasses.FrozenInstanceError: + pass + + def test_empty_subtitle_allowed(self) -> None: + """subtitle 可以为空字符串。""" + ctx = AnchorContext( + node_id="L2_010", + card_text="scene card", + frame_paths=[], + subtitle="", + distractor_texts=[], + ) + assert ctx.subtitle == "" + + def test_multiple_frame_paths(self) -> None: + """frame_paths 可包含多个路径。""" + paths = ["/data/f1.jpg", "/data/f2.jpg", "/data/f3.jpg"] + ctx = AnchorContext( + node_id="L2_005", + card_text="multi-frame event", + frame_paths=paths, + subtitle="Dialogue line", + distractor_texts=["other1", "other2"], + ) + assert len(ctx.frame_paths) == 3 + + +class TestTaskTypeSpec: + """TaskTypeSpec 数据类测试。""" + + def test_frozen(self) -> None: + """TaskTypeSpec 是不可变的。""" + spec = TaskTypeSpec( + level="L3", + needs_frames=True, + frame_count="1", + context_fields=("frame_summary",), + ) + try: + spec.level = "L2" # type: ignore[misc] + raise AssertionError("应抛出 FrozenInstanceError") + except dataclasses.FrozenInstanceError: + pass + + def test_context_fields_is_tuple(self) -> None: + """context_fields 应为 tuple（不可变）。""" + for task_type, spec in TASK_TYPE_LEVEL_MAP.items(): + assert isinstance(spec.context_fields, tuple), ( + f"{task_type} 的 context_fields 不是 tuple" + )