diff --git a/app/question_gen/synthesizer.py b/app/question_gen/synthesizer.py index d5ee161..fb9ceaf 100644 --- a/app/question_gen/synthesizer.py +++ b/app/question_gen/synthesizer.py @@ -12,7 +12,7 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: import random - from app.tree.index import L1Node, L2Node, L3Node, TreeIndex + from app.tree.index import L2Node, L3Node, TreeIndex @dataclass(frozen=True) @@ -170,27 +170,33 @@ def _sample_l3( 异常: ValueError: 候选 L3 节点不足。 """ - # Phase 1: 收集所有 L3 候选 + # Phase 1: 收集所有 L3 候选(必须有 frame_path) 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: + if l3.id not in used_node_ids and l3.frame_path: candidates.append((l3, l2)) if not candidates: - raise ValueError(f"锚节点不足: {task_type} 无可用 L3 节点") + raise ValueError(f"锚节点不足: {task_type} 无可用 L3 节点(需具备 frame_path)") # Phase 2: 随机选取 chosen_l3, parent_l2 = rng.choice(candidates) - # Phase 3: 构造上下文 + # Phase 3: 构造上下文(frame_path 已在候选过滤中保证非 None) card_text = _serialize_l3_card(chosen_l3, spec.context_fields) - frame_paths = [chosen_l3.frame_path] if chosen_l3.frame_path else [] + frame_paths = [chosen_l3.frame_path] # type: ignore[list-item] 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] + # Phase 4: 干扰项——整棵树中其他 L3 的 frame_summary + distractor_texts = [ + l3.card.frame_summary + for root in tree.roots + for l2 in root.children + for l3 in l2.children + if l3.id != chosen_l3.id + ] return AnchorContext( node_id=chosen_l3.id, @@ -227,17 +233,17 @@ def _sample_l2( ValueError: 候选 L2 节点不足。 """ # Phase 1: 收集所有 L2 候选 - all_l2: list[tuple[L2Node, L1Node]] = [] + 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, root)) + all_l2.append(l2) if not all_l2: raise ValueError(f"锚节点不足: {task_type} 无可用 L2 节点") # Phase 2: 随机选取 - chosen_l2, parent_l1 = rng.choice(all_l2) + chosen_l2 = rng.choice(all_l2) is_temporal_perception = task_type == "Temporal Perception" @@ -252,8 +258,13 @@ def _sample_l2( else: # 普通 L2:随机采样 2-3 个 L3 帧 children_with_frames = [l3 for l3 in chosen_l2.children if l3.frame_path] + if len(children_with_frames) < 2: + raise ValueError( + f"锚节点不足: {task_type} 需要 >=2 个子帧," + f"但 {chosen_l2.id} 仅有 {len(children_with_frames)} 个可用帧" + ) n_frames = min(rng.randint(2, 3), len(children_with_frames)) - sampled = rng.sample(children_with_frames, n_frames) if n_frames > 0 else [] + sampled = rng.sample(children_with_frames, n_frames) frame_paths = [l3.frame_path for l3 in sampled if l3.frame_path] # Phase 4: card_text @@ -266,9 +277,12 @@ def _sample_l2( if chosen_l2.children and chosen_l2.children[0].subtitle: subtitle = chosen_l2.children[0].subtitle - # Phase 6: 干扰项——同 L1 下其他 L2 的 event_description + # Phase 6: 干扰项——整棵树中其他 L2 的 event_description distractor_texts = [ - l2.card.event_description for l2 in parent_l1.children if l2.id != chosen_l2.id + l2.card.event_description + for root in tree.roots + for l2 in root.children + if l2.id != chosen_l2.id ] return AnchorContext( @@ -290,7 +304,7 @@ def _sample_l1( """L1 层级锚节点采样(Temporal Reasoning / Information Synopsis)。 Information Synopsis:使用目标 L1 下全部 L2 子节点。 - Temporal Reasoning:使用 >=3 个 L2 子节点(不足 3 个则全部使用)。 + Temporal Reasoning:严格要求 >=3 个 L2 子节点,不足则抛 ValueError。 L2 按 time_range 升序排列,每个 L2 取一帧代表。 参数: @@ -304,7 +318,7 @@ def _sample_l1( AnchorContext 实例。 异常: - ValueError: 候选 L1 节点不足。 + ValueError: 候选 L1 节点不足,或 Temporal Reasoning 的 L2 子节点 <3。 """ # Phase 1: 收集可用 L1 candidates = [r for r in tree.roots if r.id not in used_node_ids] @@ -319,8 +333,13 @@ def _sample_l1( # 必须使用全部 L2 selected_l2 = list(chosen_l1.children) else: - # Temporal Reasoning:>=3 个 L2(不足则全部) - if len(chosen_l1.children) <= 3: + # Temporal Reasoning:严格要求 >=3 个 L2 + if len(chosen_l1.children) < 3: + raise ValueError( + f"锚节点不足: {task_type} 需要 >=3 个 L2 子节点," + f"但 {chosen_l1.id} 仅有 {len(chosen_l1.children)} 个" + ) + 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))) @@ -385,8 +404,10 @@ def _sample_l1_l2( if l2.id not in used_node_ids: all_l2.append(l2) - if not all_l2: - raise ValueError(f"锚节点不足: {task_type} 无可用 L2 节点") + if len(all_l2) < 2: + raise ValueError( + f"锚节点不足: {task_type} 需要 >=2 个 L2 节点,但仅有 {len(all_l2)} 个可用" + ) # Phase 2: 随机选 2-3 个 n_pick = min(rng.randint(2, 3), len(all_l2)) diff --git a/tests/unit/test_synthesizer.py b/tests/unit/test_synthesizer.py index 2437111..eaf1d3b 100644 --- a/tests/unit/test_synthesizer.py +++ b/tests/unit/test_synthesizer.py @@ -185,15 +185,15 @@ class TestSampleAnchor: """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) + # 找到被选中的 L1,验证 frame_paths 数量 == 该 L1 下全部 L2 数量 + chosen_l1 = next(r for r in tree.roots if r.id == ctx.node_id) + assert len(ctx.frame_paths) == len(chosen_l1.children) def test_l1_type_l2_nodes_in_time_order(self) -> None: - """L1 题型(Temporal Reasoning)的 card_text 应有实质内容。""" + """Temporal Reasoning 应返回 >=3 帧且 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.frame_paths) >= 3 assert len(ctx.card_text) > 20 def test_used_node_ids_excluded(self) -> None: @@ -219,12 +219,13 @@ class TestSampleAnchor: """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 2 <= 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 内容。""" + """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 应含有该字段内容 + # context_fields 包含 spatial_layout,card_text 必须出现该字段名 + assert "spatial_layout" in ctx.card_text assert len(ctx.card_text) > 20