feat(question_gen): AnchorContext + 12 题型-层级映射常量

- AnchorContext frozen dataclass: 锚节点生成上下文(node_id, card_text, frame_paths, subtitle, distractor_texts)
- TaskTypeSpec frozen dataclass: 题型生成规格(level, needs_frames, frame_count, context_fields)
- TASK_TYPE_LEVEL_MAP: 12 种 Video-MME 题型 → 树层级 + 生成规格映射
- 11 项单元测试覆盖:映射完整性、值类型、层级合法性、frozen 不变性

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-09 05:10:58 -04:00
parent eb15ab315e
commit 9eb9b86954
2 changed files with 202 additions and 0 deletions
+68
View File
@@ -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",)),
}
+134
View File
@@ -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"
)