iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(question_gen): add ActionRecognitionStrategy with 6 SubPatterns

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-14 06:49:03 -04:00
parent eaa58604b3
commit d0a8019fe1
2 changed files with 595 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/question_gen/strategy_action_recognition.py
+333
View File
@@ -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 []
tests/unit/test_strategy_action_recognition.py
+262
View File
@@ -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