From f74711cd1104db516f4fa2dd57235ae67e70403b Mon Sep 17 00:00:00 2001 From: iomgaa Date: Sat, 11 Jul 2026 23:30:56 -0400 Subject: [PATCH] feat(question_gen): add v2 material sampler with family constraints Implement sample_material_v2 module that samples tree nodes with QuestionFamilySpec-aware constraint validation, providing richer MaterialContext output (subtitles, cross-L2 context, frame paths). Key components: - AnchorContext/MaterialContext frozen dataclasses - _validate_sampling_constraints: multi-level constraint checking - _collect_subtitle_sentences: subtree subtitle extraction - _collect_cross_l2_context: peer L2 event descriptions - sample_material_v2: main entry with retry-on-constraint-violation Tests: 11 unit tests covering normal sampling, used-node exclusion, constraint violation retries, cross-L2 population, and subtitle collection. Co-Authored-By: Claude Opus 4.6 (1M context) --- app/question_gen/sampler_v2.py | 549 +++++++++++++++++++++++++++++++++ tests/unit/test_sampler_v2.py | 360 +++++++++++++++++++++ 2 files changed, 909 insertions(+) create mode 100644 app/question_gen/sampler_v2.py create mode 100644 tests/unit/test_sampler_v2.py diff --git a/app/question_gen/sampler_v2.py b/app/question_gen/sampler_v2.py new file mode 100644 index 0000000..ca40393 --- /dev/null +++ b/app/question_gen/sampler_v2.py @@ -0,0 +1,549 @@ +"""v2 素材采样器 — 基于家族约束的树节点采样与上下文收集。 + +在 v1 synthesizer 的基础上引入 QuestionFamilySpec 约束验证， +为每次出题提供更丰富的素材上下文（字幕、跨 L2 上下文、帧路径）。 + +典型调用路径:: + + material = sample_material_v2( + tree=tree_index, + family_spec=REASONING_FAMILY, + task_type="Causal Reasoning", + used_node_ids=already_used, + rng=rng, + ) +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from loguru import logger + +if TYPE_CHECKING: + import random + + from app.question_gen.families import QuestionFamilySpec, SamplingConstraint + from app.tree.index import L1Node, L2Node, TreeIndex + +# --------------------------------------------------------------------------- +# 题型 → 采样层级映射 +# --------------------------------------------------------------------------- + +_TASK_TYPE_TO_LEVEL: dict[str, int] = { + # Level 3（细粒度帧级） + "Action Recognition": 3, + "Object Recognition": 3, + # Level 2（片段/事件级） + "Action Reasoning": 2, + "Action Prediction": 2, + "Action Sequence": 2, + "Object Reasoning": 2, + "Object Interaction": 2, + "Scene Understanding": 2, + "Event Reasoning": 2, + "Causal Reasoning": 2, + # Level 1（段落/场景级） + "Temporal Reasoning": 1, + "Spatial Reasoning": 1, +} + + +# --------------------------------------------------------------------------- +# 数据类型 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class AnchorContext: + """采样锚点上下文。 + + 属性: + node_id: 锚节点 ID。 + level: 锚节点所在层级（1/2/3）。 + l2_id: 锚节点所属的 L2 节点 ID（若自身为 L2 则等于 node_id； + 若为 L1 则取其第一个 L2 子节点 ID）。 + """ + + node_id: str + level: int + l2_id: str + + +@dataclass(frozen=True) +class MaterialContext: + """采样素材上下文 — 出题所需的全部素材打包。 + + 属性: + anchor: 采样锚点信息。 + source_nodes: 参与采样的节点 ID 元组。 + subtitle_sentences: 锚节点子树中收集的字幕句列表。 + frame_paths: 锚节点子树中可用的帧路径列表。 + cross_l2_texts: 跨 L2 段的上下文文本列表（仅 cross_l2_span 时填充）。 + """ + + anchor: AnchorContext + source_nodes: tuple[str, ...] + subtitle_sentences: list[str] + frame_paths: list[str] + cross_l2_texts: list[str] + + +# --------------------------------------------------------------------------- +# 内部索引辅助 +# --------------------------------------------------------------------------- + + +def _find_l2_node(tree: TreeIndex, l2_id: str) -> tuple[L2Node, L1Node] | None: + """按 ID 定位 L2 节点及其父 L1。 + + 参数: + tree: 三层树索引。 + l2_id: L2 节点 ID。 + + 返回: + (L2Node, 父L1Node) 元组；未找到返回 None。 + """ + for l1 in tree.roots: + for l2 in l1.children: + if l2.id == l2_id: + return (l2, l1) + return None + + +def _find_l1_node(tree: TreeIndex, l1_id: str) -> L1Node | None: + """按 ID 定位 L1 节点。 + + 参数: + tree: 三层树索引。 + l1_id: L1 节点 ID。 + + 返回: + L1Node；未找到返回 None。 + """ + for l1 in tree.roots: + if l1.id == l1_id: + return l1 + return None + + +# --------------------------------------------------------------------------- +# 公开辅助函数 +# --------------------------------------------------------------------------- + + +def _validate_sampling_constraints( + tree: TreeIndex, node_id: str, constraint: SamplingConstraint +) -> bool: + """校验指定节点是否满足采样约束。 + + 根据节点层级自动判断检查范围： + - L2 节点：检查其子 L3 的帧/字幕数量。 + - L1 节点：检查其下全部 L2/L3 的帧/字幕总数。 + - L3 节点：检查其所属 L2 的子树。 + + 参数: + tree: 三层树索引。 + node_id: 待检查节点 ID。 + constraint: 采样约束条件。 + + 返回: + True 表示满足所有约束，False 表示至少一项不满足。 + """ + # Phase 1: 确定目标 L2 节点列表 + target_l2_nodes: list[L2Node] = [] + parent_l1: L1Node | None = None + + # 先尝试作为 L2 + result = _find_l2_node(tree, node_id) + if result is not None: + l2_node, parent_l1 = result + target_l2_nodes = [l2_node] + else: + # 尝试作为 L1 + l1_node = _find_l1_node(tree, node_id) + if l1_node is not None: + target_l2_nodes = list(l1_node.children) + parent_l1 = l1_node + else: + # 尝试作为 L3 — 找到其所属 L2 + for l1 in tree.roots: + for l2 in l1.children: + for l3 in l2.children: + if l3.id == node_id: + target_l2_nodes = [l2] + parent_l1 = l1 + break + if target_l2_nodes: + break + if target_l2_nodes: + break + + if not target_l2_nodes: + return False + + # Phase 2: 统计 L3 节点数 + total_l3 = sum(len(l2.children) for l2 in target_l2_nodes) + if total_l3 < constraint.min_l3_nodes: + return False + + # Phase 3: 检查帧路径可用性 + if constraint.require_frames: + has_frame = any(l3.frame_path for l2 in target_l2_nodes for l3 in l2.children) + if not has_frame: + return False + + # Phase 4: 统计字幕数 + subtitle_count = 0 + for l2 in target_l2_nodes: + if l2.card.subtitle: + subtitle_count += 1 + for l3 in l2.children: + if l3.card.subtitle: + subtitle_count += 1 + if subtitle_count < constraint.min_subtitles: + return False + + # Phase 5: 检查跨 L2 可用性 + return not (constraint.cross_l2_span and (parent_l1 is None or len(parent_l1.children) < 2)) + + +def _collect_subtitle_sentences(tree: TreeIndex, node_ids: tuple[str, ...]) -> list[str]: + """从指定节点集合中收集字幕句。 + + 遍历每个 node_id 对应的子树，提取非空字幕。 + 对 L2 节点提取自身 + 子 L3 字幕；对 L1 提取下属全部。 + + 参数: + tree: 三层树索引。 + node_ids: 待收集字幕的节点 ID 元组。 + + 返回: + 非空字幕句列表（去除空白后非空的字幕）。 + """ + sentences: list[str] = [] + + for nid in node_ids: + # 尝试作为 L2 + result = _find_l2_node(tree, nid) + if result is not None: + l2_node, _ = result + if l2_node.card.subtitle: + sentences.append(l2_node.card.subtitle) + for l3 in l2_node.children: + if l3.card.subtitle: + sentences.append(l3.card.subtitle) + continue + + # 尝试作为 L1 + l1_node = _find_l1_node(tree, nid) + if l1_node is not None: + for l2 in l1_node.children: + if l2.card.subtitle: + sentences.append(l2.card.subtitle) + for l3 in l2.children: + if l3.card.subtitle: + sentences.append(l3.card.subtitle) + continue + + # 尝试作为 L3 + for l1 in tree.roots: + for l2 in l1.children: + for l3 in l2.children: + if l3.id == nid and l3.card.subtitle: + sentences.append(l3.card.subtitle) + + return sentences + + +def _collect_cross_l2_context(tree: TreeIndex, anchor_l2_id: str, max_peers: int = 3) -> list[str]: + """收集锚 L2 的同级 L2 节点描述文本（跨 L2 上下文）。 + + 找到锚 L2 所属的 L1 父节点，取该父节点下除锚 L2 之外的其他 L2 描述。 + + 参数: + tree: 三层树索引。 + anchor_l2_id: 锚 L2 节点 ID。 + max_peers: 最多返回的同级 L2 描述数量。 + + 返回: + 同级 L2 的 event_description 列表（最多 max_peers 条）。 + """ + result = _find_l2_node(tree, anchor_l2_id) + if result is None: + return [] + + _, parent_l1 = result + peers: list[str] = [] + for l2 in parent_l1.children: + if l2.id != anchor_l2_id: + peers.append(l2.card.event_description) + if len(peers) >= max_peers: + break + + return peers + + +# --------------------------------------------------------------------------- +# 层级采样策略 +# --------------------------------------------------------------------------- + + +def _sample_l3_node( + tree: TreeIndex, + used_node_ids: set[str], + rng: random.Random, +) -> tuple[str, str] | None: + """随机采样一个未使用的 L3 节点，返回 (l3_id, 所属l2_id)。 + + 参数: + tree: 三层树索引。 + used_node_ids: 已用节点 ID 集合。 + rng: 随机数生成器。 + + 返回: + (l3_id, l2_id) 元组；无候选返回 None。 + """ + candidates: list[tuple[str, str]] = [] + for l1 in tree.roots: + for l2 in l1.children: + for l3 in l2.children: + if l3.id not in used_node_ids: + candidates.append((l3.id, l2.id)) + + if not candidates: + return None + return rng.choice(candidates) + + +def _sample_l2_node( + tree: TreeIndex, + used_node_ids: set[str], + rng: random.Random, +) -> tuple[str, str] | None: + """随机采样一个未使用的 L2 节点，返回 (l2_id, l2_id)。 + + 参数: + tree: 三层树索引。 + used_node_ids: 已用节点 ID 集合。 + rng: 随机数生成器。 + + 返回: + (l2_id, l2_id) 元组；无候选返回 None。 + """ + candidates: list[str] = [] + for l1 in tree.roots: + for l2 in l1.children: + if l2.id not in used_node_ids: + candidates.append(l2.id) + + if not candidates: + return None + chosen = rng.choice(candidates) + return (chosen, chosen) + + +def _sample_l1_node( + tree: TreeIndex, + used_node_ids: set[str], + rng: random.Random, +) -> tuple[str, str] | None: + """随机采样一个未使用的 L1 节点，返回 (l1_id, 首个子l2_id)。 + + 参数: + tree: 三层树索引。 + used_node_ids: 已用节点 ID 集合。 + rng: 随机数生成器。 + + 返回: + (l1_id, first_l2_id) 元组；无候选返回 None。 + """ + candidates: list[tuple[str, str]] = [] + for l1 in tree.roots: + if l1.id not in used_node_ids and l1.children: + candidates.append((l1.id, l1.children[0].id)) + + if not candidates: + return None + return rng.choice(candidates) + + +def _collect_frame_paths(tree: TreeIndex, node_id: str) -> list[str]: + """收集节点子树下的所有可用帧路径。 + + 参数: + tree: 三层树索引。 + node_id: 目标节点 ID。 + + 返回: + 帧路径列表。 + """ + paths: list[str] = [] + + # L2 节点 + result = _find_l2_node(tree, node_id) + if result is not None: + l2_node, _ = result + for l3 in l2_node.children: + if l3.frame_path: + paths.append(l3.frame_path) + return paths + + # L1 节点 + l1_node = _find_l1_node(tree, node_id) + if l1_node is not None: + for l2 in l1_node.children: + for l3 in l2.children: + if l3.frame_path: + paths.append(l3.frame_path) + return paths + + # L3 节点 + for l1 in tree.roots: + for l2 in l1.children: + for l3 in l2.children: + if l3.id == node_id and l3.frame_path: + paths.append(l3.frame_path) + return paths + + return paths + + +def _collect_source_nodes(tree: TreeIndex, node_id: str) -> tuple[str, ...]: + """收集节点子树涉及的全部节点 ID（包含自身）。 + + 参数: + tree: 三层树索引。 + node_id: 目标节点 ID。 + + 返回: + 相关节点 ID 元组。 + """ + ids: list[str] = [node_id] + + # L2 节点：加入子 L3 + result = _find_l2_node(tree, node_id) + if result is not None: + l2_node, _ = result + for l3 in l2_node.children: + ids.append(l3.id) + return tuple(ids) + + # L1 节点：加入子 L2 + L3 + l1_node = _find_l1_node(tree, node_id) + if l1_node is not None: + for l2 in l1_node.children: + ids.append(l2.id) + for l3 in l2.children: + ids.append(l3.id) + return tuple(ids) + + # L3 节点：仅自身 + return tuple(ids) + + +# --------------------------------------------------------------------------- +# 主入口 +# --------------------------------------------------------------------------- + + +def sample_material_v2( + tree: TreeIndex, + family_spec: QuestionFamilySpec, + task_type: str, + used_node_ids: set[str], + rng: random.Random, + *, + max_attempts: int = 10, +) -> MaterialContext: + """基于家族约束从视频树中采样素材上下文。 + + 采样流程： + 1. 根据 task_type 确定采样层级 + 2. 随机选取候选节点（排除 used_node_ids） + 3. 验证 SamplingConstraint 约束 + 4. 约束不满足则重试（最多 max_attempts 次） + 5. 收集字幕、帧路径、跨 L2 上下文 + + 参数: + tree: 三层树索引。 + family_spec: 问题家族规格（含采样约束）。 + task_type: 任务类型字符串。 + used_node_ids: 本轮已用节点 ID 集合。 + rng: 可控随机数生成器。 + max_attempts: 最大尝试次数。 + + 返回: + MaterialContext 实例。 + + 异常: + RuntimeError: 耗尽 max_attempts 次尝试仍无法满足约束。 + KeyError: task_type 不在 _TASK_TYPE_TO_LEVEL 映射中。 + """ + level = _TASK_TYPE_TO_LEVEL[task_type] + constraint = family_spec.sampling + + for attempt in range(max_attempts): + # Phase 1: 按层级采样候选节点 + if level == 3: + sampled = _sample_l3_node(tree, used_node_ids, rng) + elif level == 2: + sampled = _sample_l2_node(tree, used_node_ids, rng) + else: + sampled = _sample_l1_node(tree, used_node_ids, rng) + + if sampled is None: + logger.debug( + "sample_material_v2 尝试 {}/{}: 无可用候选节点 (level={})", + attempt + 1, + max_attempts, + level, + ) + continue + + node_id, l2_id = sampled + + # Phase 2: 验证约束 + if not _validate_sampling_constraints(tree, node_id, constraint): + logger.debug( + "sample_material_v2 尝试 {}/{}: 约束违反 (node={})", + attempt + 1, + max_attempts, + node_id, + ) + continue + + # Phase 3: 构造 AnchorContext + anchor = AnchorContext(node_id=node_id, level=level, l2_id=l2_id) + + # Phase 4: 收集素材 + source_nodes = _collect_source_nodes(tree, node_id) + subtitle_sentences = _collect_subtitle_sentences(tree, (node_id,)) + frame_paths = _collect_frame_paths(tree, node_id) + + # Phase 5: 跨 L2 上下文（仅 cross_l2_span 时收集） + cross_l2_texts: list[str] = [] + if constraint.cross_l2_span: + cross_l2_texts = _collect_cross_l2_context(tree, l2_id) + + logger.debug( + "sample_material_v2 成功: node={}, level={}, attempt={}/{}", + node_id, + level, + attempt + 1, + max_attempts, + ) + + return MaterialContext( + anchor=anchor, + source_nodes=source_nodes, + subtitle_sentences=subtitle_sentences, + frame_paths=frame_paths, + cross_l2_texts=cross_l2_texts, + ) + + raise RuntimeError( + f"sample_material_v2: 耗尽 max_attempts={max_attempts} 次尝试，" + f"无法为 task_type='{task_type}' 满足家族 '{family_spec.name}' 的采样约束" + ) diff --git a/tests/unit/test_sampler_v2.py b/tests/unit/test_sampler_v2.py new file mode 100644 index 0000000..be4d60e --- /dev/null +++ b/tests/unit/test_sampler_v2.py @@ -0,0 +1,360 @@ +"""v2 素材采样器单元测试。 + +测试 sample_material_v2 及其辅助函数的核心行为： +- 正常采样返回 MaterialContext +- 已用节点排除 +- 约束违反时重试直至 RuntimeError +- 跨 L2 上下文收集 +- 字幕收集 +""" + +from __future__ import annotations + +import random + +import pytest + +from app.question_gen.families import ( + REASONING_FAMILY, + RETRIEVAL_FAMILY, + VISUAL_FAMILY, + SamplingConstraint, +) +from app.tree.index import ( + IndexMeta, + L1Card, + L1Node, + L2Card, + L2Node, + L3Card, + L3Node, + TreeIndex, +) + +# --------------------------------------------------------------------------- +# Fixture: 构建含丰富数据的真实树结构 +# --------------------------------------------------------------------------- + + +def _make_l3( + l1_idx: int, + l2_idx: int, + l3_idx: int, + *, + subtitle: str = "", + frame_path: str | None = None, +) -> L3Node: + """构建 L3 节点，带可控 subtitle/frame_path。""" + node_id = f"l1_{l1_idx}_l2_{l2_idx}_l3_{l3_idx}" + return L3Node( + id=node_id, + card=L3Card( + frame_summary=f"帧{l3_idx}描述:L1={l1_idx},L2={l2_idx}", + visible_entities=[f"实体_{l3_idx}"], + ongoing_actions=[f"动作_{l3_idx}"], + visible_text=[], + spatial_layout="居中", + visual_attributes={"lighting": "明亮"}, + subtitle=subtitle, + ), + timestamp=float(l3_idx * 2), + frame_path=frame_path, + ) + + +def _make_l2( + l1_idx: int, + l2_idx: int, + n_l3: int = 3, + *, + subtitle: str = "", + with_frames: bool = True, + with_subtitles: bool = True, +) -> L2Node: + """构建 L2 节点，可控子节点数量和属性。""" + children: list[L3Node] = [] + for i in range(n_l3): + sub = f"字幕L1={l1_idx}_L2={l2_idx}_L3={i}" if with_subtitles else "" + fp = f"frames/l1_{l1_idx}_l2_{l2_idx}_l3_{i}.jpg" if with_frames else None + children.append(_make_l3(l1_idx, l2_idx, i, subtitle=sub, frame_path=fp)) + + l2_subtitle = subtitle or (f"L2事件字幕:L1={l1_idx}_L2={l2_idx}" if with_subtitles else "") + return L2Node( + id=f"l1_{l1_idx}_l2_{l2_idx}", + card=L2Card( + event_description=f"事件:L1={l1_idx},L2={l2_idx}", + entities=[f"角色_{l2_idx}"], + actions=[f"行为_{l2_idx}"], + action_subjects=[f"主体_{l2_idx}"], + visible_text=[], + spatial_relations="左右排列", + state_changes=None, + subtitle=l2_subtitle, + ), + time_range=(l2_idx * 30.0, (l2_idx + 1) * 30.0), + children=children, + ) + + +def _make_l1(l1_idx: int, n_l2: int = 3, n_l3: int = 3) -> L1Node: + """构建 L1 节点，含多个 L2 子节点。""" + return L1Node( + id=f"l1_{l1_idx}", + card=L1Card( + scene_summary=f"场景{l1_idx}摘要", + main_setting="室内" if l1_idx % 2 == 0 else "户外", + key_entities=[f"主角_{l1_idx}"], + main_actions=[f"主行为_{l1_idx}"], + topic_keywords=[f"关键词_{l1_idx}"], + visible_text=[], + temporal_flow="从左到右", + ), + time_range=(l1_idx * 600.0, (l1_idx + 1) * 600.0), + children=[_make_l2(l1_idx, j, n_l3) for j in range(n_l2)], + ) + + +@pytest.fixture() +def real_tree() -> TreeIndex: + """构建包含 2 个 L1、每个 L1 含 3 个 L2、每个 L2 含 5 个 L3 的真实树。 + + 共 2*3*5 = 30 个 L3 节点，6 个 L2 节点，2 个 L1 节点。 + 所有节点有帧路径和字幕。满足 REASONING_FAMILY 的 min_l3_nodes=4 要求。 + """ + meta = IndexMeta(source_path="/test/video.mp4", modality="video") + roots = [_make_l1(i, n_l2=3, n_l3=5) for i in range(2)] + return TreeIndex(metadata=meta, roots=roots) + + +@pytest.fixture() +def sparse_tree() -> TreeIndex: + """构建一棵稀疏树——无帧、少字幕，用于测试约束违反。 + + 只有 1 个 L1, 1 个 L2, 1 个 L3。L3 无帧无字幕。 + """ + meta = IndexMeta(source_path="/test/sparse.mp4", modality="video") + l3 = _make_l3(0, 0, 0, subtitle="", frame_path=None) + l2 = L2Node( + id="sparse_l2_0", + card=L2Card( + event_description="稀疏事件", + entities=[], + actions=[], + action_subjects=[], + visible_text=[], + spatial_relations="", + state_changes=None, + subtitle="", + ), + time_range=(0.0, 30.0), + children=[l3], + ) + l1 = L1Node( + id="sparse_l1_0", + card=L1Card( + scene_summary="稀疏场景", + main_setting="未知", + key_entities=[], + main_actions=[], + topic_keywords=[], + visible_text=[], + temporal_flow="", + ), + time_range=(0.0, 600.0), + children=[l2], + ) + return TreeIndex(metadata=meta, roots=[l1]) + + +# --------------------------------------------------------------------------- +# 测试类 +# --------------------------------------------------------------------------- + + +class TestSampleMaterialV2: + """sample_material_v2 核心行为测试。""" + + def test_returns_material_context(self, real_tree: TreeIndex) -> None: + """正常采样返回 MaterialContext，字段类型正确。""" + from app.question_gen.sampler_v2 import MaterialContext, sample_material_v2 + + rng = random.Random(42) + result = sample_material_v2( + tree=real_tree, + family_spec=RETRIEVAL_FAMILY, + task_type="Action Reasoning", + used_node_ids=set(), + rng=rng, + ) + + assert isinstance(result, MaterialContext) + assert result.anchor.node_id # 非空 + assert result.anchor.level in (1, 2, 3) + assert len(result.source_nodes) > 0 + assert isinstance(result.subtitle_sentences, list) + assert isinstance(result.frame_paths, list) + assert isinstance(result.cross_l2_texts, list) + + def test_respects_used_nodes(self, real_tree: TreeIndex) -> None: + """已用节点被正确排除，不会重复采样。""" + from app.question_gen.sampler_v2 import sample_material_v2 + + rng = random.Random(42) + + # 把所有 L2 节点标记为已用（除了最后一个） + all_l2_ids: set[str] = set() + for l1 in real_tree.roots: + for l2 in l1.children: + all_l2_ids.add(l2.id) + + # 留下恰好一个 L2 未用 + last_l2_id = real_tree.roots[-1].children[-1].id + used = all_l2_ids - {last_l2_id} + + result = sample_material_v2( + tree=real_tree, + family_spec=RETRIEVAL_FAMILY, + task_type="Action Reasoning", + used_node_ids=used, + rng=rng, + ) + + # 锚节点应该是那个未被排除的 L2 + assert result.anchor.node_id == last_l2_id + + def test_constraint_violation_retries(self, sparse_tree: TreeIndex) -> None: + """稀疏树上，严格约束满足不了，耗尽重试后抛 RuntimeError。""" + from app.question_gen.sampler_v2 import sample_material_v2 + + rng = random.Random(42) + + # VISUAL_FAMILY 要求 require_frames=True, min_l3_nodes=3 + # sparse_tree 只有 1 个 L3 且无帧 → 约束必然违反 + with pytest.raises(RuntimeError, match="max_attempts"): + sample_material_v2( + tree=sparse_tree, + family_spec=VISUAL_FAMILY, + task_type="Object Recognition", + used_node_ids=set(), + rng=rng, + max_attempts=3, + ) + + def test_cross_l2_populated_for_reasoning(self, real_tree: TreeIndex) -> None: + """REASONING 家族要求 cross_l2_span=True，cross_l2_texts 应被填充。""" + from app.question_gen.sampler_v2 import sample_material_v2 + + rng = random.Random(42) + + result = sample_material_v2( + tree=real_tree, + family_spec=REASONING_FAMILY, + task_type="Causal Reasoning", + used_node_ids=set(), + rng=rng, + ) + + # cross_l2_span=True 时必须有跨 L2 文本 + assert len(result.cross_l2_texts) > 0 + + def test_subtitle_sentences_from_anchor(self, real_tree: TreeIndex) -> None: + """采样结果的 subtitle_sentences 来自锚节点所属子树。""" + from app.question_gen.sampler_v2 import sample_material_v2 + + rng = random.Random(42) + + result = sample_material_v2( + tree=real_tree, + family_spec=RETRIEVAL_FAMILY, + task_type="Action Reasoning", + used_node_ids=set(), + rng=rng, + ) + + # real_tree 所有节点都有字幕，所以 subtitle_sentences 非空 + assert len(result.subtitle_sentences) > 0 + # 字幕应来自锚节点所属的子树（L2 自身字幕 + 子 L3 字幕） + # fixture 中 L2 字幕格式: "L2事件字幕:L1={l1_idx}_L2={l2_idx}" + # fixture 中 L3 字幕格式: "字幕L1={l1_idx}_L2={l2_idx}_L3={l3_idx}" + # 解析锚 L2 的索引信息来验证 + anchor_l2_id = result.anchor.l2_id # 如 "l1_1_l2_2" + # 从 ID 提取 L1/L2 索引 + parts = anchor_l2_id.split("_") # ["l1", "1", "l2", "2"] + l1_idx, l2_idx = parts[1], parts[3] + # 字幕中应包含 "L1={l1_idx}_L2={l2_idx}" 格式 + pattern = f"L1={l1_idx}_L2={l2_idx}" + has_related = any(pattern in s for s in result.subtitle_sentences) + assert has_related + + +class TestValidateSamplingConstraints: + """_validate_sampling_constraints 辅助函数测试。""" + + def test_passes_relaxed_constraint(self, real_tree: TreeIndex) -> None: + """宽松约束在丰富树上应通过。""" + from app.question_gen.sampler_v2 import _validate_sampling_constraints + + relaxed = SamplingConstraint( + min_subtitles=1, + min_l3_nodes=1, + require_frames=False, + cross_l2_span=False, + ) + # 取第一个 L2 节点 + node_id = real_tree.roots[0].children[0].id + assert _validate_sampling_constraints(real_tree, node_id, relaxed) is True + + def test_fails_strict_frame_constraint(self, sparse_tree: TreeIndex) -> None: + """require_frames=True 但无帧时应返回 False。""" + from app.question_gen.sampler_v2 import _validate_sampling_constraints + + strict = SamplingConstraint( + min_subtitles=0, + min_l3_nodes=1, + require_frames=True, + cross_l2_span=False, + ) + node_id = sparse_tree.roots[0].children[0].id + assert _validate_sampling_constraints(sparse_tree, node_id, strict) is False + + +class TestCollectSubtitleSentences: + """_collect_subtitle_sentences 辅助函数测试。""" + + def test_collects_from_l2_and_l3(self, real_tree: TreeIndex) -> None: + """收集指定节点的 L2 字幕和子 L3 字幕。""" + from app.question_gen.sampler_v2 import _collect_subtitle_sentences + + l2_id = real_tree.roots[0].children[0].id + sentences = _collect_subtitle_sentences(real_tree, (l2_id,)) + # 应包含 L2 自身字幕 + 3 个 L3 子节点字幕 + assert len(sentences) >= 1 + + def test_empty_for_no_subtitles(self, sparse_tree: TreeIndex) -> None: + """无字幕节点返回空列表。""" + from app.question_gen.sampler_v2 import _collect_subtitle_sentences + + l2_id = sparse_tree.roots[0].children[0].id + sentences = _collect_subtitle_sentences(sparse_tree, (l2_id,)) + assert sentences == [] + + +class TestCollectCrossL2Context: + """_collect_cross_l2_context 辅助函数测试。""" + + def test_returns_peer_l2_descriptions(self, real_tree: TreeIndex) -> None: + """跨 L2 上下文应返回同 L1 下其他 L2 的描述。""" + from app.question_gen.sampler_v2 import _collect_cross_l2_context + + anchor_l2_id = real_tree.roots[0].children[0].id + texts = _collect_cross_l2_context(real_tree, anchor_l2_id, max_peers=3) + # L1_0 有 3 个 L2，排除 anchor 后剩 2 个 + assert len(texts) == 2 + + def test_max_peers_limits_output(self, real_tree: TreeIndex) -> None: + """max_peers 参数限制返回数量。""" + from app.question_gen.sampler_v2 import _collect_cross_l2_context + + anchor_l2_id = real_tree.roots[0].children[0].id + texts = _collect_cross_l2_context(real_tree, anchor_l2_id, max_peers=1) + assert len(texts) <= 1