From d0a8019fe17c38e2500bf4337a7ea41d2ffba421 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 06:49:03 -0400 Subject: [PATCH] feat(question_gen): add ActionRecognitionStrategy with 6 SubPatterns Self-contained strategy targeting 6 Agent failure modes in Action Recognition: premature_evidence_anchoring, temporal_reasoning_failure, semantic_rigidity, fine_grained_visual_action, cross_segment_entity_tracking, evidence_gap_confabulation. - L2 default sampling (upgrade from L3) with 3 patterns overriding to L1 - Weighted random SubPattern selection (0.20/0.20/0.15/0.15/0.15/0.15) - Each SubPattern includes instruction, examples, distractor rules - Satisfies TaskTypeStrategy Protocol without extending BaseTaskTypeStrategy - 32 unit tests covering all properties, definitions, and selection behavior Co-Authored-By: Claude Opus 4.6 (1M context) --- .../strategy_action_recognition.py | 333 ++++++++++++++++++ .../unit/test_strategy_action_recognition.py | 262 ++++++++++++++ 2 files changed, 595 insertions(+) create mode 100644 app/question_gen/strategy_action_recognition.py create mode 100644 tests/unit/test_strategy_action_recognition.py diff --git a/app/question_gen/strategy_action_recognition.py b/app/question_gen/strategy_action_recognition.py new file mode 100644 index 0000000..f681238 --- /dev/null +++ b/app/question_gen/strategy_action_recognition.py @@ -0,0 +1,333 @@ +"""Action Recognition 特化出题策略。 + +靶向 Agent 在动作识别类题目上的 6 种典型失败机制，通过加权随机 +SubPattern 选择为 VLM 出题提供聚焦指令。 + +与 BaseTaskTypeStrategy 完全自包含，不依赖 QuestionFamilySpec。 +采样从 L3 提升至 L2（跨段推理需要更广视野），3 个 SubPattern +进一步覆盖至 L1 以测试全局追踪能力。 +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from app.question_gen.families import SamplingConstraint +from app.question_gen.strategy import SubPattern + +if TYPE_CHECKING: + import random + + +# --------------------------------------------------------------------------- +# 6 个 SubPattern 定义 +# --------------------------------------------------------------------------- + +_PREMATURE_EVIDENCE_ANCHORING = SubPattern( + name="premature_evidence_anchoring", + weight=0.20, + sampling_level_override=1, + constraint_override=None, + instruction=( + "设计一道需要考察视频中完整证据链的动作识别题。" + "正确答案的关键证据出现在视频后半段或跨多个片段，" + "但视频前段包含一个看似合理的局部匹配——" + "Agent 若仅凭首条匹配停止搜索就会出错。" + ), + positive_examples=[ + { + "question": "视频中厨师最终采用了哪种烹饪方式？", + "answer": "B. 蒸", + "why": "厨师先演示了炒（前段），但最终菜品使用蒸制（后段），" + "锚定首段证据的 Agent 会错选'炒'。", + }, + ], + negative_examples=[ + { + "question": "视频开头厨师在做什么？", + "why": "答案仅需前段信息，不会触发过早锚定失败。", + }, + ], + distractor_rules=( + "将视频前段出现的局部匹配动作设为强干扰项。" + "其余干扰项使用语义相近但未出现的动作，" + "确保仅看前段片段的 Agent 有高概率选错。" + ), +) + +_TEMPORAL_REASONING_FAILURE = SubPattern( + name="temporal_reasoning_failure", + weight=0.20, + sampling_level_override=1, + constraint_override=None, + instruction=( + "设计一道要求正确排序或定位第 N 次出现的动作识别题。" + "题目需要 Agent 追踪事件的时间先后顺序，" + "或准确识别某动作在视频中第几次出现。" + "打乱时序或错误计数即会答错。" + ), + positive_examples=[ + { + "question": "运动员第三次尝试起跳前做了什么准备动作？", + "answer": "C. 深蹲热身", + "why": "需要准确定位'第三次'起跳而非其他次，" + "时序推理失败的 Agent 会混淆不同次尝试的准备动作。", + }, + ], + negative_examples=[ + { + "question": "运动员在视频中做了什么？", + "why": "不涉及时序排序或计数，Agent 无需追踪顺序。", + }, + ], + distractor_rules=( + "干扰项使用其他次出现时的关联动作（如第 1 次或第 2 次的准备动作）。" + "确保各选项在视频中确实出现过，只是对应的时间点不同。" + ), +) + +_SEMANTIC_RIGIDITY = SubPattern( + name="semantic_rigidity", + weight=0.15, + sampling_level_override=None, + constraint_override=None, + instruction=( + "设计一道动作识别题，正确选项使用与视频原始描述不同的同义表达。" + "Agent 需理解语义等价而非依赖字面匹配——" + "例如视频字幕说'奔跑'，正确选项写作'快速移动'。" + ), + positive_examples=[ + { + "question": "工人对墙面进行了什么操作？", + "answer": "A. 涂覆保护层", + "why": "视频中字幕描述为'刷漆'，正确答案改写为'涂覆保护层'，" + "依赖字面匹配的 Agent 会因找不到完全一致的表述而错选。", + }, + ], + negative_examples=[ + { + "question": "工人在刷漆吗？", + "why": "选项直接复用视频原文，不考察语义理解。", + }, + ], + distractor_rules=( + "干扰项使用与正确答案表面相似但语义不同的表达。" + "保留一个与视频原始字幕字面接近但含义偏移的选项作为陷阱。" + ), +) + +_FINE_GRAINED_VISUAL_ACTION = SubPattern( + name="fine_grained_visual_action", + weight=0.15, + sampling_level_override=None, + constraint_override=None, + instruction=( + "设计一道需要区分细粒度动作方式的识别题。" + "题目聚焦于 HOW（怎么做）而非 WHAT（做什么），" + "如区分'搅拌'与'翻炒'、'拧'与'拉'等视觉上相似但方式不同的动作。" + ), + positive_examples=[ + { + "question": "维修人员是如何拆卸螺丝的？", + "answer": "D. 用扳手逆时针旋转", + "why": "需要区分拆卸的具体方式（扳手 vs 螺丝刀、顺时针 vs 逆时针），" + "粗粒度识别只能判断'在拆螺丝'，无法区分方式。", + }, + ], + negative_examples=[ + { + "question": "维修人员在做什么？", + "why": "只需粗粒度动作识别（'拆螺丝'），不考察具体方式。", + }, + ], + distractor_rules=( + "干扰项使用同一大类动作的不同执行方式（同一动词的不同修饰）。" + "确保所有选项描述的都是合理的执行方式，仅细节不同。" + "避免使用明显不相关的动作作为干扰。" + ), +) + +_CROSS_SEGMENT_ENTITY_TRACKING = SubPattern( + name="cross_segment_entity_tracking", + weight=0.15, + sampling_level_override=1, + constraint_override=None, + instruction=( + "设计一道需要跨视频段落追踪同一实体动作的识别题。" + "目标实体在不同片段中外观、称呼或上下文发生变化，" + "Agent 需要将多段信息合并才能正确回答关于该实体的动作问题。" + ), + positive_examples=[ + { + "question": "穿红色外套的人在视频中总共完成了哪些动作？", + "answer": "B. 先讲解、后示范、最后总结", + "why": "该人物在前段穿外套讲解，中段脱外套示范，后段重新穿上总结，" + "无法跨段追踪的 Agent 会遗漏某段动作。", + }, + { + "question": "主持人在节目不同环节中分别做了什么？", + "answer": "A. 开场介绍、采访嘉宾、总结点评", + "why": "主持人在不同场景切换中持续出现，需要跨段聚合。", + }, + ], + negative_examples=[ + { + "question": "视频第一个片段中的人在做什么？", + "why": "仅需单段信息，不考察跨段追踪。", + }, + ], + distractor_rules=( + "干扰项遗漏某些片段的动作或混入其他实体的动作。构造一个只包含部分片段信息的选项作为强干扰。" + ), +) + +_EVIDENCE_GAP_CONFABULATION = SubPattern( + name="evidence_gap_confabulation", + weight=0.15, + sampling_level_override=None, + constraint_override=None, + instruction=( + "设计一道动作识别题，视频中存在证据空缺（如遮挡、跳切、画外音）。" + "正确答案承认信息不足或基于间接证据推断，" + "而非凭空编造因果链。Agent 若虚构缺失证据即会出错。" + ), + positive_examples=[ + { + "question": "画面切走后，演讲者下一步做了什么？", + "answer": "C. 无法从视频中直接确定", + "why": "画面跳切导致该动作无直接视觉证据，" + "倾向于虚构的 Agent 会编造一个看似合理的动作。", + }, + ], + negative_examples=[ + { + "question": "画面中演讲者正在做什么？", + "why": "动作在画面中可见，不存在证据空缺。", + }, + ], + distractor_rules=( + "干扰项使用看似合理但缺乏视频证据支持的因果推断。" + "至少一个干扰项应构造完整的虚假因果链以诱导 Agent 选择。" + ), +) + +AR_SUB_PATTERNS: tuple[SubPattern, ...] = ( + _PREMATURE_EVIDENCE_ANCHORING, + _TEMPORAL_REASONING_FAILURE, + _SEMANTIC_RIGIDITY, + _FINE_GRAINED_VISUAL_ACTION, + _CROSS_SEGMENT_ENTITY_TRACKING, + _EVIDENCE_GAP_CONFABULATION, +) + +# 预计算：名称列表 + 权重列表（避免每次 select 重新构建） +_AR_PATTERN_NAMES: list[str] = [sp.name for sp in AR_SUB_PATTERNS] +_AR_PATTERN_WEIGHTS: list[float] = [sp.weight for sp in AR_SUB_PATTERNS] +_AR_PATTERN_BY_NAME: dict[str, SubPattern] = {sp.name: sp for sp in AR_SUB_PATTERNS} + + +# --------------------------------------------------------------------------- +# ActionRecognitionStrategy +# --------------------------------------------------------------------------- + +_SAMPLING_CONSTRAINT = SamplingConstraint( + min_subtitles=3, + min_l3_nodes=5, + require_frames=True, + cross_l2_span=True, +) + + +class ActionRecognitionStrategy: + """Action Recognition 特化出题策略。 + + 自包含实现，不依赖 BaseTaskTypeStrategy 或 QuestionFamilySpec。 + 靶向 6 种典型失败机制，通过加权随机 SubPattern 引导 VLM 出题。 + + 属性: + task_type: 题型名 — "Action Recognition"。 + strategy_name: 策略标识 — "ACTION_RECOGNITION"。 + skill_target: 目标失败机制 — "M1_AR"。 + sampling_level: 采样层级 — 2（L2，从 L3 升级以获得跨段视野）。 + sampling_constraint: 采样约束（min_subtitles=3, min_l3_nodes=5, require_frames, cross_l2_span）。 + prompt_template: 出题模板 — "action_recognition.md"。 + leak_probe_template: 泄漏探测模板 — "gate_leak_retrieval.md"。 + """ + + @property + def task_type(self) -> str: + """返回题型名。""" + return "Action Recognition" + + @property + def strategy_name(self) -> str: + """返回策略标识名。""" + return "ACTION_RECOGNITION" + + @property + def skill_target(self) -> str: + """返回目标失败机制编号。""" + return "M1_AR" + + @property + def sampling_level(self) -> int: + """返回采样层级（L2）。""" + return 2 + + @property + def sampling_constraint(self) -> SamplingConstraint: + """返回采样约束。""" + return _SAMPLING_CONSTRAINT + + @property + def prompt_template(self) -> str: + """返回出题 prompt 模板文件名。""" + return "action_recognition.md" + + @property + def leak_probe_template(self) -> str: + """返回泄漏探测模板文件名。""" + return "gate_leak_retrieval.md" + + def select_sub_pattern(self, rng: random.Random) -> SubPattern: + """按权重随机选择一个 SubPattern。 + + 参数: + rng: 随机数生成器（确保可复现）。 + + 返回: + 选中的 SubPattern 实例（永不为 None）。 + """ + chosen_name = rng.choices( + _AR_PATTERN_NAMES, + weights=_AR_PATTERN_WEIGHTS, + k=1, + )[0] + return _AR_PATTERN_BY_NAME[chosen_name] + + def build_prompt_context(self, material: Any, sub_pattern: SubPattern | None) -> dict: + """构建 prompt 上下文字典。 + + 参数: + material: 采样素材（当前未使用，留给管线扩展）。 + sub_pattern: 选中的子模式。 + + 返回: + 包含 family_name, prompt_template, sub_pattern 的字典。 + """ + return { + "family_name": self.strategy_name, + "prompt_template": self.prompt_template, + "sub_pattern": sub_pattern.name if sub_pattern is not None else None, + } + + def extra_gates(self, candidate: Any) -> list: + """返回额外门控列表（当前为空）。 + + 参数: + candidate: 候选题目。 + + 返回: + 空列表。 + """ + return [] diff --git a/tests/unit/test_strategy_action_recognition.py b/tests/unit/test_strategy_action_recognition.py new file mode 100644 index 0000000..423adf9 --- /dev/null +++ b/tests/unit/test_strategy_action_recognition.py @@ -0,0 +1,262 @@ +"""ActionRecognitionStrategy 单元测试。 + +验证策略属性、SubPattern 定义、加权随机选择行为。 +""" + +from __future__ import annotations + +import random +from collections import Counter + +import pytest + +from app.question_gen.families import SamplingConstraint +from app.question_gen.strategy import SubPattern, TaskTypeStrategy +from app.question_gen.strategy_action_recognition import ( + AR_SUB_PATTERNS, + ActionRecognitionStrategy, +) + +# --------------------------------------------------------------------------- +# SubPattern 分组 +# --------------------------------------------------------------------------- + +_L1_PATTERN_NAMES = frozenset( + { + "premature_evidence_anchoring", + "temporal_reasoning_failure", + "cross_segment_entity_tracking", + } +) +_L2_PATTERN_NAMES = frozenset( + { + "semantic_rigidity", + "fine_grained_visual_action", + "evidence_gap_confabulation", + } +) +_ALL_PATTERN_NAMES = _L1_PATTERN_NAMES | _L2_PATTERN_NAMES + + +class TestStrategyProperties: + """策略静态属性与 Protocol 符合性。""" + + @pytest.fixture() + def strategy(self) -> ActionRecognitionStrategy: + """创建策略实例。""" + return ActionRecognitionStrategy() + + def test_implements_protocol(self, strategy: ActionRecognitionStrategy) -> None: + """ActionRecognitionStrategy 满足 TaskTypeStrategy Protocol。""" + assert isinstance(strategy, TaskTypeStrategy) + + def test_task_type(self, strategy: ActionRecognitionStrategy) -> None: + """task_type 为 'Action Recognition'。""" + assert strategy.task_type == "Action Recognition" + + def test_strategy_name(self, strategy: ActionRecognitionStrategy) -> None: + """strategy_name 为 'ACTION_RECOGNITION'。""" + assert strategy.strategy_name == "ACTION_RECOGNITION" + + def test_skill_target(self, strategy: ActionRecognitionStrategy) -> None: + """skill_target 为 'M1_AR'。""" + assert strategy.skill_target == "M1_AR" + + def test_sampling_level(self, strategy: ActionRecognitionStrategy) -> None: + """sampling_level 为 2（L2）。""" + assert strategy.sampling_level == 2 + + def test_sampling_constraint(self, strategy: ActionRecognitionStrategy) -> None: + """采样约束匹配设计值。""" + expected = SamplingConstraint( + min_subtitles=3, + min_l3_nodes=5, + require_frames=True, + cross_l2_span=True, + ) + assert strategy.sampling_constraint == expected + + def test_prompt_template(self, strategy: ActionRecognitionStrategy) -> None: + """prompt_template 为 'action_recognition.md'。""" + assert strategy.prompt_template == "action_recognition.md" + + def test_leak_probe_template(self, strategy: ActionRecognitionStrategy) -> None: + """leak_probe_template 为 'gate_leak_retrieval.md'。""" + assert strategy.leak_probe_template == "gate_leak_retrieval.md" + + def test_extra_gates_empty(self, strategy: ActionRecognitionStrategy) -> None: + """extra_gates 返回空列表。""" + assert strategy.extra_gates(None) == [] + + +class TestSubPatternDefinitions: + """SubPattern 定义的完整性与正确性。""" + + def test_ar_sub_patterns_count(self) -> None: + """AR_SUB_PATTERNS 包含 6 个子模式。""" + assert len(AR_SUB_PATTERNS) == 6 + + def test_ar_sub_patterns_is_tuple(self) -> None: + """AR_SUB_PATTERNS 为 tuple 类型。""" + assert isinstance(AR_SUB_PATTERNS, tuple) + + def test_all_are_sub_pattern_instances(self) -> None: + """所有元素均为 SubPattern 实例。""" + for sp in AR_SUB_PATTERNS: + assert isinstance(sp, SubPattern) + + def test_all_names_present(self) -> None: + """6 个子模式名称完整覆盖。""" + names = {sp.name for sp in AR_SUB_PATTERNS} + assert names == _ALL_PATTERN_NAMES + + def test_l1_patterns_have_level_override_1(self) -> None: + """L1 子模式的 sampling_level_override 为 1。""" + for sp in AR_SUB_PATTERNS: + if sp.name in _L1_PATTERN_NAMES: + assert sp.sampling_level_override == 1, ( + f"{sp.name} 应为 L1 (override=1)，实际 {sp.sampling_level_override}" + ) + + def test_l2_patterns_have_no_level_override(self) -> None: + """L2 子模式的 sampling_level_override 为 None。""" + for sp in AR_SUB_PATTERNS: + if sp.name in _L2_PATTERN_NAMES: + assert sp.sampling_level_override is None, ( + f"{sp.name} 应为 L2 (override=None)，实际 {sp.sampling_level_override}" + ) + + def test_weights_sum_to_one(self) -> None: + """所有子模式权重之和为 1.0。""" + total = sum(sp.weight for sp in AR_SUB_PATTERNS) + assert abs(total - 1.0) < 1e-9 + + def test_individual_weights(self) -> None: + """各子模式权重匹配设计值。""" + weight_map = {sp.name: sp.weight for sp in AR_SUB_PATTERNS} + assert abs(weight_map["premature_evidence_anchoring"] - 0.20) < 1e-9 + assert abs(weight_map["temporal_reasoning_failure"] - 0.20) < 1e-9 + assert abs(weight_map["semantic_rigidity"] - 0.15) < 1e-9 + assert abs(weight_map["fine_grained_visual_action"] - 0.15) < 1e-9 + assert abs(weight_map["cross_segment_entity_tracking"] - 0.15) < 1e-9 + assert abs(weight_map["evidence_gap_confabulation"] - 0.15) < 1e-9 + + def test_all_have_nonempty_instruction(self) -> None: + """每个子模式有非空 instruction。""" + for sp in AR_SUB_PATTERNS: + assert sp.instruction.strip(), f"{sp.name} instruction 为空" + + def test_all_have_nonempty_distractor_rules(self) -> None: + """每个子模式有非空 distractor_rules。""" + for sp in AR_SUB_PATTERNS: + assert sp.distractor_rules.strip(), f"{sp.name} distractor_rules 为空" + + def test_all_have_positive_examples(self) -> None: + """每个子模式有至少 1 个正面示例。""" + for sp in AR_SUB_PATTERNS: + assert len(sp.positive_examples) >= 1, f"{sp.name} 缺少 positive_examples" + + def test_all_have_negative_examples(self) -> None: + """每个子模式有至少 1 个反面示例。""" + for sp in AR_SUB_PATTERNS: + assert len(sp.negative_examples) >= 1, f"{sp.name} 缺少 negative_examples" + + def test_positive_examples_have_required_keys(self) -> None: + """正面示例包含 question, answer, why 字段。""" + required = {"question", "answer", "why"} + for sp in AR_SUB_PATTERNS: + for ex in sp.positive_examples: + missing = required - set(ex.keys()) + assert not missing, f"{sp.name} 正面示例缺少字段: {missing}" + + def test_negative_examples_have_required_keys(self) -> None: + """反面示例包含 question, why 字段。""" + required = {"question", "why"} + for sp in AR_SUB_PATTERNS: + for ex in sp.negative_examples: + missing = required - set(ex.keys()) + assert not missing, f"{sp.name} 反面示例缺少字段: {missing}" + + def test_all_constraint_overrides_none(self) -> None: + """所有子模式的 constraint_override 为 None。""" + for sp in AR_SUB_PATTERNS: + assert sp.constraint_override is None, f"{sp.name} constraint_override 应为 None" + + +class TestSelectSubPattern: + """select_sub_pattern 加权随机选择行为。""" + + @pytest.fixture() + def strategy(self) -> ActionRecognitionStrategy: + """创建策略实例。""" + return ActionRecognitionStrategy() + + def test_returns_sub_pattern(self, strategy: ActionRecognitionStrategy) -> None: + """select_sub_pattern 返回 SubPattern（不是 None）。""" + rng = random.Random(42) + result = strategy.select_sub_pattern(rng) + assert isinstance(result, SubPattern) + + def test_deterministic_with_same_seed(self, strategy: ActionRecognitionStrategy) -> None: + """相同种子产生相同结果。""" + results_a = [strategy.select_sub_pattern(random.Random(99)).name for _ in range(20)] + results_b = [strategy.select_sub_pattern(random.Random(99)).name for _ in range(20)] + assert results_a == results_b + + def test_covers_all_patterns(self, strategy: ActionRecognitionStrategy) -> None: + """足够多次采样覆盖全部 6 个子模式。""" + rng = random.Random(12345) + seen = {strategy.select_sub_pattern(rng).name for _ in range(500)} + assert seen == _ALL_PATTERN_NAMES + + def test_distribution_roughly_matches_weights( + self, strategy: ActionRecognitionStrategy + ) -> None: + """采样分布大致匹配权重（容差 +-0.05）。""" + rng = random.Random(42) + n = 5000 + counter: Counter[str] = Counter() + for _ in range(n): + counter[strategy.select_sub_pattern(rng).name] += 1 + + weight_map = {sp.name: sp.weight for sp in AR_SUB_PATTERNS} + for name, expected_w in weight_map.items(): + observed_ratio = counter[name] / n + assert abs(observed_ratio - expected_w) < 0.05, ( + f"{name}: expected ~{expected_w:.2f}, got {observed_ratio:.3f}" + ) + + +class TestBuildPromptContext: + """build_prompt_context 输出验证。""" + + @pytest.fixture() + def strategy(self) -> ActionRecognitionStrategy: + """创建策略实例。""" + return ActionRecognitionStrategy() + + def test_returns_dict_with_required_keys(self, strategy: ActionRecognitionStrategy) -> None: + """返回包含 family_name, prompt_template, sub_pattern 的字典。""" + sp = AR_SUB_PATTERNS[0] + ctx = strategy.build_prompt_context(material=None, sub_pattern=sp) + assert "family_name" in ctx + assert "prompt_template" in ctx + assert "sub_pattern" in ctx + + def test_family_name_value(self, strategy: ActionRecognitionStrategy) -> None: + """family_name 为 ACTION_RECOGNITION。""" + sp = AR_SUB_PATTERNS[0] + ctx = strategy.build_prompt_context(material=None, sub_pattern=sp) + assert ctx["family_name"] == "ACTION_RECOGNITION" + + def test_prompt_template_value(self, strategy: ActionRecognitionStrategy) -> None: + """prompt_template 为 action_recognition.md。""" + sp = AR_SUB_PATTERNS[0] + ctx = strategy.build_prompt_context(material=None, sub_pattern=sp) + assert ctx["prompt_template"] == "action_recognition.md" + + def test_sub_pattern_name(self, strategy: ActionRecognitionStrategy) -> None: + """sub_pattern 字段等于子模式名称。""" + for sp in AR_SUB_PATTERNS: + ctx = strategy.build_prompt_context(material=None, sub_pattern=sp) + assert ctx["sub_pattern"] == sp.name