diff --git a/app/question_gen/families.py b/app/question_gen/families.py new file mode 100644 index 0000000..a8c3c1a --- /dev/null +++ b/app/question_gen/families.py @@ -0,0 +1,263 @@ +"""题族规格声明 — 定义 5 大问题家族及其采样、泄漏检测、提示模板约束。 + +每个 QuestionFamilySpec 对应一种失败机制(skill_target M1–M5), +由 get_family_for_slot 在出题时按权重分配。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + import random + + +@dataclass(frozen=True) +class LeakTestProfile: + """泄漏测试配置 — 定义快捷答题捷径类型与通过阈值。 + + Attributes: + shortcut_type: 捷径类型标识。 + probe_template: store/prompts/question_gen/ 下的探测模板文件名。 + pass_threshold: 通过阈值(0–1),低于此值视为存在泄漏。 + """ + + shortcut_type: str + probe_template: str + pass_threshold: float + + +@dataclass(frozen=True) +class SamplingConstraint: + """采样约束 — 对树节点的最低要求。 + + Attributes: + min_subtitles: 最少字幕段数。 + min_l3_nodes: 最少 L3 节点数。 + require_frames: 是否要求帧图像可用。 + cross_l2_span: 是否要求跨 L2 段采样。 + """ + + min_subtitles: int + min_l3_nodes: int + require_frames: bool + cross_l2_span: bool + + +@dataclass(frozen=True) +class QuestionFamilySpec: + """问题家族规格 — 一个家族的完整声明。 + + Attributes: + name: 家族标识名(如 "RETRIEVAL")。 + skill_target: 目标失败机制编号(M1–M5)。 + sampling: 采样约束。 + legal_task_types: 该家族合法的任务类型集合(frozenset)。 + leak_profile: 泄漏测试配置。 + prompt_template: 出题 prompt 模板文件名。 + """ + + name: str + skill_target: str + sampling: SamplingConstraint + legal_task_types: frozenset[str] + leak_profile: LeakTestProfile + prompt_template: str + + +# --------------------------------------------------------------------------- +# 5 大家族实例 +# --------------------------------------------------------------------------- + +RETRIEVAL_FAMILY = QuestionFamilySpec( + name="RETRIEVAL", + skill_target="M1", + sampling=SamplingConstraint( + min_subtitles=2, + min_l3_nodes=3, + require_frames=False, + cross_l2_span=False, + ), + legal_task_types=frozenset( + [ + "Action Recognition", + "Action Reasoning", + "Action Prediction", + "Action Sequence", + "Object Recognition", + "Object Reasoning", + "Object Interaction", + "Scene Understanding", + "Event Reasoning", + "Causal Reasoning", + "Temporal Reasoning", + "Spatial Reasoning", + ] + ), + leak_profile=LeakTestProfile( + shortcut_type="temporal_proximity", + probe_template="gate_leak_retrieval.md", + pass_threshold=0.6, + ), + prompt_template="retrieval.md", +) + +REASONING_FAMILY = QuestionFamilySpec( + name="REASONING", + skill_target="M2", + sampling=SamplingConstraint( + min_subtitles=3, + min_l3_nodes=4, + require_frames=False, + cross_l2_span=True, + ), + legal_task_types=frozenset( + [ + "Action Reasoning", + "Object Reasoning", + "Event Reasoning", + "Causal Reasoning", + "Temporal Reasoning", + "Spatial Reasoning", + ] + ), + leak_profile=LeakTestProfile( + shortcut_type="frequency", + probe_template="gate_leak_reasoning.md", + pass_threshold=0.5, + ), + prompt_template="reasoning.md", +) + +ENUMERATION_FAMILY = QuestionFamilySpec( + name="ENUMERATION", + skill_target="M3", + sampling=SamplingConstraint( + min_subtitles=2, + min_l3_nodes=5, + require_frames=False, + cross_l2_span=False, + ), + legal_task_types=frozenset( + [ + "Action Sequence", + "Object Recognition", + "Object Interaction", + "Scene Understanding", + ] + ), + leak_profile=LeakTestProfile( + shortcut_type="option_length", + probe_template="gate_leak_enumeration.md", + pass_threshold=0.6, + ), + prompt_template="enumeration.md", +) + +VISUAL_FAMILY = QuestionFamilySpec( + name="VISUAL", + skill_target="M4", + sampling=SamplingConstraint( + min_subtitles=0, + min_l3_nodes=3, + require_frames=True, + cross_l2_span=False, + ), + legal_task_types=frozenset( + [ + "Object Recognition", + "Scene Understanding", + "Action Recognition", + "Spatial Reasoning", + ] + ), + leak_profile=LeakTestProfile( + shortcut_type="visual_salience", + probe_template="gate_leak_visual.md", + pass_threshold=0.5, + ), + prompt_template="visual.md", +) + +SPATIAL_FAMILY = QuestionFamilySpec( + name="SPATIAL", + skill_target="M5", + sampling=SamplingConstraint( + min_subtitles=0, + min_l3_nodes=3, + require_frames=True, + cross_l2_span=False, + ), + legal_task_types=frozenset( + [ + "Spatial Reasoning", + "Object Interaction", + "Scene Understanding", + ] + ), + leak_profile=LeakTestProfile( + shortcut_type="spatial_default", + probe_template="gate_leak_spatial.md", + pass_threshold=0.5, + ), + prompt_template="spatial.md", +) + +ALL_FAMILIES: tuple[QuestionFamilySpec, ...] = ( + RETRIEVAL_FAMILY, + REASONING_FAMILY, + ENUMERATION_FAMILY, + VISUAL_FAMILY, + SPATIAL_FAMILY, +) + +# 按 name 索引,用于 get_family_for_slot 快速查找 +_FAMILY_BY_NAME: dict[str, QuestionFamilySpec] = {f.name: f for f in ALL_FAMILIES} + + +def get_family_for_slot( + task_type: str, + family_ratios: dict[str, float], + rng: random.Random, +) -> QuestionFamilySpec: + """根据任务类型和家族权重比例,随机选择一个合法的问题家族。 + + Args: + task_type: 任务类型字符串(必须是 12 种合法类型之一)。 + family_ratios: 家族名称到权重的映射(如 {"RETRIEVAL": 0.30, ...})。 + rng: 随机数生成器实例(确保可复现)。 + + Returns: + 被选中的 QuestionFamilySpec。 + + Raises: + ValueError: task_type 不在任何家族的 legal_task_types 中。 + ValueError: 给定 task_type 下没有合法家族(所有合法族权重为 0 或不在 ratios 中)。 + """ + # 检查 task_type 是否被任一家族接受 + all_legal_types: set[str] = set() + for family in ALL_FAMILIES: + all_legal_types.update(family.legal_task_types) + if task_type not in all_legal_types: + msg = f"task_type '{task_type}' 不在任何家族的合法类型中" + raise ValueError(msg) + + # 过滤出接受该 task_type 且在 ratios 中有正权重的家族 + candidates: list[QuestionFamilySpec] = [] + weights: list[float] = [] + for family_name, weight in family_ratios.items(): + family = _FAMILY_BY_NAME.get(family_name) + if family is None: + continue + if task_type in family.legal_task_types and weight > 0: + candidates.append(family) + weights.append(weight) + + if not candidates: + msg = f"task_type '{task_type}' 下没有合法家族可选(检查 family_ratios)" + raise ValueError(msg) + + # 归一化权重 + 加权随机选择 + chosen = rng.choices(candidates, weights=weights, k=1)[0] + return chosen diff --git a/tests/unit/test_families.py b/tests/unit/test_families.py new file mode 100644 index 0000000..3f62c8c --- /dev/null +++ b/tests/unit/test_families.py @@ -0,0 +1,173 @@ +"""题族规格声明单元测试。""" + +from __future__ import annotations + +import random +from collections import Counter + +import pytest +from scipy import stats + +from app.question_gen.families import ( + ALL_FAMILIES, + RETRIEVAL_FAMILY, + get_family_for_slot, +) + +# 12 种任务类型(来自 harness config) +ALL_TASK_TYPES: frozenset[str] = frozenset( + [ + "Action Recognition", + "Action Reasoning", + "Action Prediction", + "Action Sequence", + "Object Recognition", + "Object Reasoning", + "Object Interaction", + "Scene Understanding", + "Event Reasoning", + "Causal Reasoning", + "Temporal Reasoning", + "Spatial Reasoning", + ] +) + + +class TestFamilySpec: + """验证 QuestionFamilySpec 声明的完整性与一致性。""" + + def test_all_families_cover_all_task_types(self) -> None: + """12 种任务类型在 ALL_FAMILIES 中全覆盖。""" + covered: set[str] = set() + for family in ALL_FAMILIES: + covered.update(family.legal_task_types) + assert covered == ALL_TASK_TYPES + + def test_skill_targets_unique(self) -> None: + """每个 family 的 skill_target 不重复。""" + targets = [f.skill_target for f in ALL_FAMILIES] + assert len(targets) == len(set(targets)) + + def test_all_families_tuple_length(self) -> None: + """ALL_FAMILIES 包含 5 个家族。""" + assert len(ALL_FAMILIES) == 5 + + def test_family_instances_are_frozen(self) -> None: + """Family spec 实例不可修改。""" + with pytest.raises(AttributeError): + RETRIEVAL_FAMILY.name = "hacked" # type: ignore[misc] + + def test_legal_task_types_are_frozensets(self) -> None: + """legal_task_types 是 frozenset,确保不可变。""" + for family in ALL_FAMILIES: + assert isinstance(family.legal_task_types, frozenset) + + def test_sampling_constraints_reasonable(self) -> None: + """采样约束值在合理范围。""" + for family in ALL_FAMILIES: + assert family.sampling.min_subtitles >= 0 + assert family.sampling.min_l3_nodes >= 1 + + def test_leak_profile_thresholds_valid(self) -> None: + """泄漏测试阈值在 (0, 1] 范围内。""" + for family in ALL_FAMILIES: + assert 0 < family.leak_profile.pass_threshold <= 1.0 + + def test_prompt_templates_not_empty(self) -> None: + """每个 family 的 prompt_template 非空。""" + for family in ALL_FAMILIES: + assert family.prompt_template + assert family.prompt_template.endswith(".md") + + +class TestGetFamilyForSlot: + """验证 get_family_for_slot 的核心行为。""" + + def test_respects_legal_task_types(self) -> None: + """返回的 family 必须包含给定 task_type。""" + rng = random.Random(42) + for task_type in ALL_TASK_TYPES: + for _ in range(50): + family = get_family_for_slot( + task_type=task_type, + family_ratios={ + "RETRIEVAL": 0.30, + "REASONING": 0.25, + "ENUMERATION": 0.20, + "VISUAL": 0.15, + "SPATIAL": 0.10, + }, + rng=rng, + ) + assert task_type in family.legal_task_types + + def test_deterministic_with_seed(self) -> None: + """相同 seed 产出相同结果。""" + ratios = { + "RETRIEVAL": 0.30, + "REASONING": 0.25, + "ENUMERATION": 0.20, + "VISUAL": 0.15, + "SPATIAL": 0.10, + } + results_a = [] + results_b = [] + for seed in range(10): + rng_a = random.Random(seed) + rng_b = random.Random(seed) + results_a.append(get_family_for_slot("Action Reasoning", ratios, rng_a).name) + results_b.append(get_family_for_slot("Action Reasoning", ratios, rng_b).name) + assert results_a == results_b + + def test_invalid_task_type_raises(self) -> None: + """不存在的 task_type 抛 ValueError。""" + rng = random.Random(0) + with pytest.raises(ValueError, match="task_type"): + get_family_for_slot( + task_type="Nonexistent Type", + family_ratios={"RETRIEVAL": 1.0}, + rng=rng, + ) + + def test_distribution_approximates_ratios(self) -> None: + """大样本下分布近似指定比例(chi-square p > 0.01)。 + + 使用 Action Recognition — 只有 RETRIEVAL 和 VISUAL 合法。 + """ + rng = random.Random(123) + ratios = { + "RETRIEVAL": 0.30, + "REASONING": 0.25, + "ENUMERATION": 0.20, + "VISUAL": 0.15, + "SPATIAL": 0.10, + } + n = 5000 + counter: Counter[str] = Counter() + for _ in range(n): + family = get_family_for_slot("Action Recognition", ratios, rng) + counter[family.name] += 1 + + # Action Recognition 合法族: RETRIEVAL(0.30), VISUAL(0.15) + # 归一化后: RETRIEVAL=0.30/0.45≈0.667, VISUAL=0.15/0.45≈0.333 + legal_ratios = {"RETRIEVAL": 0.30, "VISUAL": 0.15} + total_weight = sum(legal_ratios.values()) + expected_probs = {k: v / total_weight for k, v in legal_ratios.items()} + + observed = [counter.get("RETRIEVAL", 0), counter.get("VISUAL", 0)] + expected = [expected_probs["RETRIEVAL"] * n, expected_probs["VISUAL"] * n] + + chi2, p_value = stats.chisquare(observed, expected) + assert p_value > 0.01, f"分布偏离过大: chi2={chi2:.2f}, p={p_value:.4f}" + + def test_no_legal_family_raises(self) -> None: + """所有 family 权重为 0 时合法族为空,应抛出 ValueError。""" + rng = random.Random(0) + # Spatial Reasoning 合法族: RETRIEVAL, REASONING, VISUAL, SPATIAL + # 如果 ratios 中只含不合法的族名,应抛错 + with pytest.raises(ValueError, match="合法"): + get_family_for_slot( + task_type="Spatial Reasoning", + family_ratios={"ENUMERATION": 1.0}, + rng=rng, + )