From a597a9f90165feb542593af0de273e346339f754 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 9 Jul 2026 05:18:25 -0400 Subject: [PATCH] =?UTF-8?q?feat(question=5Fgen):=20sample=5Fanchor=20?= =?UTF-8?q?=E2=80=94=20=E6=8C=89=E9=A2=98=E5=9E=8B=E5=B1=82=E7=BA=A7?= =?UTF-8?q?=E9=87=87=E6=A0=B7=E9=94=9A=E8=8A=82=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 含 6 种层级分支：L3 单帧、L2 多帧、Temporal Perception 特例、 L1 全量/采样 L2、L1-L2 混合。时间排序 + used_node_ids 排除。 Co-Authored-By: Claude Opus 4.6 (1M context) --- app/question_gen/synthesizer.py | 399 ++++++++++++++++++++++++++++++++ tests/unit/test_synthesizer.py | 100 +++++++- 2 files changed, 497 insertions(+), 2 deletions(-) diff --git a/app/question_gen/synthesizer.py b/app/question_gen/synthesizer.py index 9aedb29..d5ee161 100644 --- a/app/question_gen/synthesizer.py +++ b/app/question_gen/synthesizer.py @@ -7,6 +7,12 @@ from __future__ import annotations from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + import random + + from app.tree.index import L1Node, L2Node, L3Node, TreeIndex @dataclass(frozen=True) @@ -66,3 +72,396 @@ TASK_TYPE_LEVEL_MAP: dict[str, TaskTypeSpec] = { "Information Synopsis": TaskTypeSpec("L1", True, "per-L2", ("scene_summary",)), "Object Reasoning": TaskTypeSpec("L1-L2", True, "per-L2", ("event_description",)), } + + +# --------------------------------------------------------------------------- +# 内部辅助函数 +# --------------------------------------------------------------------------- + + +def _serialize_l3_card(l3: L3Node, context_fields: tuple[str, ...]) -> str: + """将 L3 节点 card 按 context_fields 序列化为可读文本。 + + 参数: + l3: L3 节点。 + context_fields: 需提取的字段名元组。 + + 返回: + 多行 "field: value" 格式的文本。 + """ + parts: list[str] = [] + for fld in context_fields: + val = getattr(l3.card, fld, None) + if val is not None: + parts.append(f"{fld}: {val}") + return "\n".join(parts) + + +def _l2_time_range_str(l2: L2Node) -> str: + """将 L2 的 time_range 格式化为可读字符串。 + + 参数: + l2: L2 节点。 + + 返回: + "time_range: (start, end)" 格式，或 "time_range: unknown"。 + """ + if l2.time_range is not None: + return f"time_range: ({l2.time_range[0]:.2f}, {l2.time_range[1]:.2f})" + return "time_range: unknown" + + +def _representative_frame(l2: L2Node) -> str | None: + """取 L2 的代表帧路径——第一个有 frame_path 的 L3 子节点。 + + 参数: + l2: L2 节点。 + + 返回: + 帧路径字符串，或 None（无可用帧时）。 + """ + for l3 in l2.children: + if l3.frame_path: + return l3.frame_path + return None + + +def _sort_l2_by_time(l2_nodes: list[L2Node]) -> list[L2Node]: + """按 time_range 升序排列 L2 节点（None 排末尾）。 + + 参数: + l2_nodes: 待排序的 L2 节点列表。 + + 返回: + 排序后的新列表（不修改原列表）。 + """ + return sorted( + l2_nodes, + key=lambda n: n.time_range[0] if n.time_range is not None else float("inf"), + ) + + +# --------------------------------------------------------------------------- +# 各层级采样策略 +# --------------------------------------------------------------------------- + + +def _sample_l3( + tree: TreeIndex, + task_type: str, + spec: TaskTypeSpec, + used_node_ids: set[str], + rng: random.Random, +) -> AnchorContext: + """L3 层级锚节点采样。 + + 收集全部 L3 节点，排除已用节点，随机选取一个。 + + 参数: + tree: 三层树索引。 + task_type: 题型名称。 + spec: 题型规格。 + used_node_ids: 已用节点 ID 集合。 + rng: 随机数生成器。 + + 返回: + AnchorContext 实例。 + + 异常: + ValueError: 候选 L3 节点不足。 + """ + # Phase 1: 收集所有 L3 候选 + candidates: list[tuple[L3Node, L2Node]] = [] + for root in tree.roots: + for l2 in root.children: + for l3 in l2.children: + if l3.id not in used_node_ids: + candidates.append((l3, l2)) + + if not candidates: + raise ValueError(f"锚节点不足: {task_type} 无可用 L3 节点") + + # Phase 2: 随机选取 + chosen_l3, parent_l2 = rng.choice(candidates) + + # Phase 3: 构造上下文 + card_text = _serialize_l3_card(chosen_l3, spec.context_fields) + frame_paths = [chosen_l3.frame_path] if chosen_l3.frame_path else [] + subtitle = chosen_l3.subtitle or "" + + # Phase 4: 干扰项——同 L2 下其他 L3 的 frame_summary + distractor_texts = [l3.card.frame_summary for l3 in parent_l2.children if l3.id != chosen_l3.id] + + return AnchorContext( + node_id=chosen_l3.id, + card_text=card_text, + frame_paths=frame_paths, + subtitle=subtitle, + distractor_texts=distractor_texts, + ) + + +def _sample_l2( + tree: TreeIndex, + task_type: str, + spec: TaskTypeSpec, + used_node_ids: set[str], + rng: random.Random, +) -> AnchorContext: + """L2 层级锚节点采样（含 Temporal Perception 特殊处理）。 + + 普通 L2 题型：随机选 1 个 L2，取 2-3 个子 L3 帧。 + Temporal Perception：0-1 帧，card_text 必含 time_range。 + + 参数: + tree: 三层树索引。 + task_type: 题型名称。 + spec: 题型规格。 + used_node_ids: 已用节点 ID 集合。 + rng: 随机数生成器。 + + 返回: + AnchorContext 实例。 + + 异常: + ValueError: 候选 L2 节点不足。 + """ + # Phase 1: 收集所有 L2 候选 + all_l2: list[tuple[L2Node, L1Node]] = [] + for root in tree.roots: + for l2 in root.children: + if l2.id not in used_node_ids: + all_l2.append((l2, root)) + + if not all_l2: + raise ValueError(f"锚节点不足: {task_type} 无可用 L2 节点") + + # Phase 2: 随机选取 + chosen_l2, parent_l1 = rng.choice(all_l2) + + is_temporal_perception = task_type == "Temporal Perception" + + # Phase 3: 帧路径 + if is_temporal_perception: + # 0-1 帧：有子节点则取 1 帧，否则 0 帧 + frame_paths: list[str] = [] + if chosen_l2.children: + first_frame = chosen_l2.children[0].frame_path + if first_frame: + frame_paths = [first_frame] + else: + # 普通 L2：随机采样 2-3 个 L3 帧 + children_with_frames = [l3 for l3 in chosen_l2.children if l3.frame_path] + n_frames = min(rng.randint(2, 3), len(children_with_frames)) + sampled = rng.sample(children_with_frames, n_frames) if n_frames > 0 else [] + frame_paths = [l3.frame_path for l3 in sampled if l3.frame_path] + + # Phase 4: card_text + card_text = f"event_description: {chosen_l2.card.event_description}" + if is_temporal_perception: + card_text += f"\n{_l2_time_range_str(chosen_l2)}" + + # Phase 5: 字幕（L2 无自身字幕，取首个子 L3 字幕） + subtitle = "" + if chosen_l2.children and chosen_l2.children[0].subtitle: + subtitle = chosen_l2.children[0].subtitle + + # Phase 6: 干扰项——同 L1 下其他 L2 的 event_description + distractor_texts = [ + l2.card.event_description for l2 in parent_l1.children if l2.id != chosen_l2.id + ] + + return AnchorContext( + node_id=chosen_l2.id, + card_text=card_text, + frame_paths=frame_paths, + subtitle=subtitle, + distractor_texts=distractor_texts, + ) + + +def _sample_l1( + tree: TreeIndex, + task_type: str, + spec: TaskTypeSpec, + used_node_ids: set[str], + rng: random.Random, +) -> AnchorContext: + """L1 层级锚节点采样（Temporal Reasoning / Information Synopsis）。 + + Information Synopsis：使用目标 L1 下全部 L2 子节点。 + Temporal Reasoning：使用 >=3 个 L2 子节点（不足 3 个则全部使用）。 + L2 按 time_range 升序排列，每个 L2 取一帧代表。 + + 参数: + tree: 三层树索引。 + task_type: 题型名称。 + spec: 题型规格。 + used_node_ids: 已用节点 ID 集合。 + rng: 随机数生成器。 + + 返回: + AnchorContext 实例。 + + 异常: + ValueError: 候选 L1 节点不足。 + """ + # Phase 1: 收集可用 L1 + candidates = [r for r in tree.roots if r.id not in used_node_ids] + if not candidates: + raise ValueError(f"锚节点不足: {task_type} 无可用 L1 节点") + + # Phase 2: 随机选取 + chosen_l1 = rng.choice(candidates) + + # Phase 3: 选定 L2 子集 + if task_type == "Information Synopsis": + # 必须使用全部 L2 + selected_l2 = list(chosen_l1.children) + else: + # Temporal Reasoning：>=3 个 L2（不足则全部） + if len(chosen_l1.children) <= 3: + selected_l2 = list(chosen_l1.children) + else: + selected_l2 = rng.sample(chosen_l1.children, rng.randint(3, len(chosen_l1.children))) + + # Phase 4: 按 time_range 升序排列 + selected_l2 = _sort_l2_by_time(selected_l2) + + # Phase 5: card_text（场景摘要） + card_text = f"scene_summary: {chosen_l1.card.scene_summary}" + + # Phase 6: 帧路径——每个 L2 取一帧代表 + frame_paths: list[str] = [] + for l2 in selected_l2: + rep = _representative_frame(l2) + if rep: + frame_paths.append(rep) + + # Phase 7: 字幕（L1 无字幕） + subtitle = "" + + # Phase 8: 干扰项——其他 L1 的 scene_summary + distractor_texts = [r.card.scene_summary for r in tree.roots if r.id != chosen_l1.id] + + return AnchorContext( + node_id=chosen_l1.id, + card_text=card_text, + frame_paths=frame_paths, + subtitle=subtitle, + distractor_texts=distractor_texts, + ) + + +def _sample_l1_l2( + tree: TreeIndex, + task_type: str, + spec: TaskTypeSpec, + used_node_ids: set[str], + rng: random.Random, +) -> AnchorContext: + """L1-L2 跨层级锚节点采样（Object Reasoning）。 + + 从全部 L2 中随机选 2-3 个，按 time_range 排序， + card_text 为各 L2 的 event_description 拼接。 + + 参数: + tree: 三层树索引。 + task_type: 题型名称。 + spec: 题型规格。 + used_node_ids: 已用节点 ID 集合。 + rng: 随机数生成器。 + + 返回: + AnchorContext 实例。 + + 异常: + ValueError: 候选 L2 节点不足。 + """ + # Phase 1: 收集全部 L2 + all_l2: list[L2Node] = [] + for root in tree.roots: + for l2 in root.children: + if l2.id not in used_node_ids: + all_l2.append(l2) + + if not all_l2: + raise ValueError(f"锚节点不足: {task_type} 无可用 L2 节点") + + # Phase 2: 随机选 2-3 个 + n_pick = min(rng.randint(2, 3), len(all_l2)) + selected = rng.sample(all_l2, n_pick) + + # Phase 3: 按 time_range 升序排列 + selected = _sort_l2_by_time(selected) + + # Phase 4: card_text = 各 L2 event_description 拼接 + card_text = "\n".join(f"event_description: {l2.card.event_description}" for l2 in selected) + + # Phase 5: 帧路径——每个 L2 取一帧代表 + frame_paths: list[str] = [] + for l2 in selected: + rep = _representative_frame(l2) + if rep: + frame_paths.append(rep) + + # Phase 6: 字幕 + subtitle = "" + + # Phase 7: 干扰项——未被选中的 L2 的 event_description + selected_ids = {l2.id for l2 in selected} + distractor_texts = [l2.card.event_description for l2 in all_l2 if l2.id not in selected_ids] + + # 使用第一个被选中节点的 ID 作为锚节点 ID + anchor_id = selected[0].id + + return AnchorContext( + node_id=anchor_id, + card_text=card_text, + frame_paths=frame_paths, + subtitle=subtitle, + distractor_texts=distractor_texts, + ) + + +# --------------------------------------------------------------------------- +# 公开接口 +# --------------------------------------------------------------------------- + + +def sample_anchor( + tree: TreeIndex, + task_type: str, + used_node_ids: set[str], + rng: random.Random, +) -> AnchorContext: + """根据题型从视频树中采样锚节点及上下文素材。 + + 依据 TASK_TYPE_LEVEL_MAP 中的层级规格，分发到对应的层级采样策略。 + 每种层级有不同的帧选取、card 序列化和干扰项收集逻辑。 + + 参数: + tree: 三层树索引。 + task_type: 12 种 Video-MME 题型之一。 + used_node_ids: 本轮已用节点 ID 集合（避免重复采样）。 + rng: 可控随机数生成器（保证可复现）。 + + 返回: + AnchorContext 实例，包含锚节点 ID、card 文本、帧路径、字幕和干扰项。 + + 异常: + KeyError: task_type 不在 TASK_TYPE_LEVEL_MAP 中。 + ValueError: 候选节点不足（全部被 used_node_ids 排除）。 + """ + spec = TASK_TYPE_LEVEL_MAP[task_type] + + if spec.level == "L3": + return _sample_l3(tree, task_type, spec, used_node_ids, rng) + elif spec.level == "L2": + return _sample_l2(tree, task_type, spec, used_node_ids, rng) + elif spec.level == "L1": + return _sample_l1(tree, task_type, spec, used_node_ids, rng) + elif spec.level == "L1-L2": + return _sample_l1_l2(tree, task_type, spec, used_node_ids, rng) + else: + raise ValueError(f"未知层级: {spec.level}") diff --git a/tests/unit/test_synthesizer.py b/tests/unit/test_synthesizer.py index 579b2e6..2437111 100644 --- a/tests/unit/test_synthesizer.py +++ b/tests/unit/test_synthesizer.py @@ -1,10 +1,20 @@ -"""synthesizer 模块单元测试 — AnchorContext + 题型映射常量。""" +"""synthesizer 模块单元测试 — AnchorContext + 题型映射常量 + sample_anchor。""" from __future__ import annotations import dataclasses +import random +from pathlib import Path -from app.question_gen.synthesizer import TASK_TYPE_LEVEL_MAP, AnchorContext, TaskTypeSpec +import pytest + +from app.question_gen.synthesizer import ( + TASK_TYPE_LEVEL_MAP, + AnchorContext, + TaskTypeSpec, + sample_anchor, +) +from app.tree.index import TreeIndex ALL_12_TYPES = [ "Object Recognition", @@ -132,3 +142,89 @@ class TestTaskTypeSpec: assert isinstance(spec.context_fields, tuple), ( f"{task_type} 的 context_fields 不是 tuple" ) + + +# --------------------------------------------------------------------------- +# sample_anchor 测试 +# --------------------------------------------------------------------------- + + +def _load_test_tree() -> tuple[TreeIndex, str]: + """加载真实测试树（store/videos/ 下第一棵）。""" + videos_dir = Path("store/videos") + first_vid = sorted(videos_dir.iterdir())[0] + tree = TreeIndex.load_json(str(first_vid / "tree.json")) + return tree, first_vid.name + + +class TestSampleAnchor: + """sample_anchor 锚节点采样测试（基于真实树数据）。""" + + def test_l3_type_returns_single_frame(self) -> None: + """L3 题型（Object Recognition）应返回恰好 1 帧。""" + tree, _vid = _load_test_tree() + ctx = sample_anchor(tree, "Object Recognition", set(), random.Random(42)) + assert len(ctx.frame_paths) == 1 + assert ctx.node_id.startswith("L") or "_L3_" in ctx.node_id + assert len(ctx.distractor_texts) > 0 + + def test_l2_type_returns_multiple_frames(self) -> None: + """L2 题型（Action Reasoning）应返回 2-3 帧。""" + tree, _vid = _load_test_tree() + ctx = sample_anchor(tree, "Action Reasoning", set(), random.Random(42)) + assert 2 <= len(ctx.frame_paths) <= 3 + + def test_temporal_perception_zero_or_one_frame(self) -> None: + """Temporal Perception 特殊处理：0-1 帧，且 card_text 包含 time_range。""" + tree, _vid = _load_test_tree() + ctx = sample_anchor(tree, "Temporal Perception", set(), random.Random(42)) + assert len(ctx.frame_paths) <= 1 + assert "time_range" in ctx.card_text.lower() or "time" in ctx.card_text.lower() + + def test_information_synopsis_uses_all_l2(self) -> None: + """Information Synopsis 必须使用目标 L1 下所有 L2 子节点。""" + tree, _vid = _load_test_tree() + ctx = sample_anchor(tree, "Information Synopsis", set(), random.Random(42)) + # 至少应有帧（每个 L2 取一帧代表） + total_l2 = sum(len(r.children) for r in tree.roots) + assert len(ctx.frame_paths) >= min(total_l2, 1) + + def test_l1_type_l2_nodes_in_time_order(self) -> None: + """L1 题型（Temporal Reasoning）的 card_text 应有实质内容。""" + tree, _vid = _load_test_tree() + ctx = sample_anchor(tree, "Temporal Reasoning", set(), random.Random(42)) + assert len(ctx.frame_paths) >= 1 + assert len(ctx.card_text) > 20 + + def test_used_node_ids_excluded(self) -> None: + """used_node_ids 中的节点不应被再次选中。""" + tree, _vid = _load_test_tree() + rng = random.Random(42) + ctx1 = sample_anchor(tree, "Object Recognition", set(), rng) + ctx2 = sample_anchor(tree, "Object Recognition", {ctx1.node_id}, random.Random(43)) + assert ctx2.node_id != ctx1.node_id + + def test_insufficient_nodes_raises(self) -> None: + """所有候选节点均被排除时应抛出 ValueError。""" + tree, _vid = _load_test_tree() + all_l3_ids: set[str] = set() + for root in tree.roots: + for l2 in root.children: + for l3 in l2.children: + all_l3_ids.add(l3.id) + with pytest.raises(ValueError, match="锚节点不足"): + sample_anchor(tree, "Object Recognition", all_l3_ids, random.Random(42)) + + def test_object_reasoning_l1_l2_type(self) -> None: + """Object Reasoning (L1-L2) 应选 2-3 个 L2 并按时间排序。""" + tree, _vid = _load_test_tree() + ctx = sample_anchor(tree, "Object Reasoning", set(), random.Random(42)) + assert 1 <= len(ctx.frame_paths) <= 3 + assert len(ctx.card_text) > 10 + + def test_spatial_reasoning_context_fields(self) -> None: + """Spatial Reasoning 的 card_text 应包含 spatial_layout 内容。""" + tree, _vid = _load_test_tree() + ctx = sample_anchor(tree, "Spatial Reasoning", set(), random.Random(42)) + # context_fields 包含 spatial_layout，card_text 应含有该字段内容 + assert len(ctx.card_text) > 20