From 645585e784ce7beb068d86cc00acec685ecd9252 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 13:28:43 -0400 Subject: [PATCH] docs: add Phase A grounded question-gen implementation plan --- ...07-14-grounded-question-gen-phaseA-plan.md | 1379 +++++++++++++++++ 1 file changed, 1379 insertions(+) create mode 100644 research-wiki/plans/2026-07-14-grounded-question-gen-phaseA-plan.md diff --git a/research-wiki/plans/2026-07-14-grounded-question-gen-phaseA-plan.md b/research-wiki/plans/2026-07-14-grounded-question-gen-phaseA-plan.md new file mode 100644 index 0000000..7f3c037 --- /dev/null +++ b/research-wiki/plans/2026-07-14-grounded-question-gen-phaseA-plan.md @@ -0,0 +1,1379 @@ +# Grounded Question-Gen Phase A Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 把 Action Recognition 干扰项从"VLM 主观写得像"下沉到机制层——候选池 + VLM 视觉打分 selector 按视觉可信度区间选出 grounded near-miss 干扰项，消灭 Easy-Options Bias（负空间干扰项→排除法秒杀）；同时修复 gate tree 错配 bug、落实 sub_pattern 持久化（Phase B 硬前提）。 + +**Architecture:** 路径隔离靠 `uses_grounded_selector` 策略属性——AR=True 走 selector 两步出题（先用现有 `generate_one_v2` 拿到"正解"，再用新模块 `distractor_selector` 生成候选池 + 视觉打分选 3 个干扰项重组四选项）；11 个非 AR 题型=False 走原路径，字节级行为不变。公共层只做纯 bug 修复、数据字段透传、门控 rubric 松绑。 + +**Tech Stack:** Python 3.11、asyncio、VLMProvider（`chat_with_images`）、sqlite3（幂等 ALTER TABLE）、json_repair、pytest。全部命令在 conda 环境 `Video-Tree-TRM` 内执行。 + +--- + +## 前置约定（所有任务通用） + +- **环境**：每条 Python/pytest/ruff 命令前缀 `conda run -n Video-Tree-TRM`。示例：`conda run -n Video-Tree-TRM pytest tests/unit/test_x.py -v`。 +- **路径隔离铁律**：除"公共纯 bug/纯数据/门控 rubric"外，任何行为变更只能发生在 AR 路径（`uses_grounded_selector=True` 分支）。每个任务末尾的回归步骤必须证明 11 个非 AR 题型行为不变。 +- **提交**：每个 Task 末尾 commit，走 `commit` skill 的消息规范（英文、imperative、`: `，**禁止任何 AI 署名**）。 +- **设计来源**：`research-wiki/designs/2026-07-14-grounded-question-gen-phaseA-design.md`。 + +--- + +## Task 1: 修复 gate tree 错配 bug（公共，纯 bug） + +`_process_one_slot` retry 换视频后，`current_tree` 已切换到新视频，但第 6 门 `run_gates` 仍传旧 `tree`，导致门控用错树验证（"无 source material"假拒绝）。 + +**Files:** +- Modify: `app/question_gen/pipeline_v2.py:471`（`run_gates(candidate=candidate, tree=tree, ...)` → `tree=current_tree`） +- Test: `tests/unit/test_pipeline_v2_tree_fix.py`（新建） + +- [ ] **Step 1: 写失败测试** + +新建 `tests/unit/test_pipeline_v2_tree_fix.py`：断言源码中 `run_gates` 调用使用 `current_tree` 而非 `tree`（AST/正则守卫测试，锁死回归）。 + +```python +"""守卫 gate tree 错配 bug：run_gates 必须用 current_tree（换视频后的当前树）。""" + +import ast +from pathlib import Path + +_PIPELINE = Path(__file__).resolve().parents[2] / "app" / "question_gen" / "pipeline_v2.py" + + +def _find_run_gates_tree_arg() -> str: + """解析 pipeline_v2.py，返回 run_gates 调用中 tree= 关键字实参的变量名。""" + tree_src = ast.parse(_PIPELINE.read_text(encoding="utf-8")) + for node in ast.walk(tree_src): + if isinstance(node, ast.Call): + func = node.func + name = getattr(func, "id", None) or getattr(func, "attr", None) + if name == "run_gates": + for kw in node.keywords: + if kw.arg == "tree": + assert isinstance(kw.value, ast.Name) + return kw.value.id + raise AssertionError("未找到 run_gates 的 tree= 关键字实参") + + +def test_run_gates_uses_current_tree(): + assert _find_run_gates_tree_arg() == "current_tree" +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_pipeline_v2_tree_fix.py -v` +Expected: FAIL（当前实参为 `tree`） + +- [ ] **Step 3: 改代码** + +`app/question_gen/pipeline_v2.py` Phase 6 的 `run_gates` 调用（约 471 行），把 `tree=tree` 改为 `tree=current_tree`： + +```python + report = await run_gates( + candidate=candidate, + tree=current_tree, + llm=llm, + leak_probe_template=strategy.leak_probe_template, + postprocess=pp, + vlm=vlm, + session_id=session_id, + ) +``` + +- [ ] **Step 4: 跑测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_pipeline_v2_tree_fix.py -v` +Expected: PASS + +- [ ] **Step 5: 提交** + +```bash +git add app/question_gen/pipeline_v2.py tests/unit/test_pipeline_v2_tree_fix.py +git commit -m "fix: run_gates must use current_tree after video resample" +``` + +--- + +## Task 2: sub_pattern 字段透传 + 持久化（公共，纯数据 / Phase B 硬前提） + +`sub_pattern` 目前只在 `_process_one_slot` 选出并写入 `store.record_item`，未随 `GeneratedQuestion` 传出，`accepted_questions.json` 也不含该字段。Phase B 按题的 `sub_pattern` 查 `supports_flip`，缺则无法工作。 + +**Files:** +- Modify: `core/types.py`（`GeneratedQuestion` 加 `sub_pattern: str | None`） +- Modify: `app/question_gen/pipeline_v2.py`（`_to_generated_question` 加 `sub_pattern` 形参；`_process_one_slot` 传入 `sub_pattern.name`） +- Modify: `tools/generate_questions.py`（`_on_accept` 与 `_append_to_json` 写 `sub_pattern`） +- Test: `tests/unit/test_generated_question_sub_pattern.py`（新建） + +- [ ] **Step 1: 写失败测试** + +新建 `tests/unit/test_generated_question_sub_pattern.py`： + +```python +"""GeneratedQuestion.sub_pattern 字段 + _to_generated_question 透传。""" + +from app.question_gen.generator_v2 import CandidateQuestion +from app.question_gen.pipeline_v2 import _to_generated_question +from core.types import GeneratedQuestion + + +def _candidate() -> CandidateQuestion: + return CandidateQuestion( + question_id="v1_Action Recognition_0001", + video_id="v1", + task_type="Action Recognition", + skill_target="M1_AR", + question="厨师最终采用了哪种烹饪方式？", + options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), + answer="A", + source_nodes=("n1", "n2"), + difficulty="hard", + ) + + +def test_generated_question_has_sub_pattern_default_none(): + q = GeneratedQuestion( + question_id="q1", video_id="v1", task_type="Action Recognition", + question="?", options=("A. x",), answer="A", + source_nodes=("n1",), difficulty="easy", + ) + assert q.sub_pattern is None + + +def test_to_generated_question_threads_sub_pattern(): + q = _to_generated_question( + _candidate(), family="ACTION_RECOGNITION", + sub_pattern="premature_evidence_anchoring", + ) + assert q.sub_pattern == "premature_evidence_anchoring" + + +def test_to_generated_question_sub_pattern_defaults_none(): + q = _to_generated_question(_candidate(), family="RETRIEVAL") + assert q.sub_pattern is None +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_generated_question_sub_pattern.py -v` +Expected: FAIL（`GeneratedQuestion` 无 `sub_pattern`；`_to_generated_question` 无该形参） + +- [ ] **Step 3: 改 `core/types.py`** + +`GeneratedQuestion` 末尾新增字段（保持 frozen dataclass，带默认值以兼容既有构造点）： + +```python + family: str | None = field(default=None) + skill_target: str | None = field(default=None) + difficulty_steps: int | None = field(default=None) + sub_pattern: str | None = field(default=None) +``` + +同步在 docstring 属性列表补一行：`sub_pattern: 出题子模式标识（AR 特化策略使用，None 表示无）。` + +- [ ] **Step 4: 改 `_to_generated_question`** + +`app/question_gen/pipeline_v2.py`，函数签名加 keyword-only 形参并透传： + +```python +def _to_generated_question( + candidate: CandidateQuestion, + *, + family: str, + options: tuple[str, ...] | None = None, + answer: str | None = None, + sub_pattern: str | None = None, +) -> GeneratedQuestion: + """... (在 docstring 参数区补 sub_pattern 说明) ...""" + return GeneratedQuestion( + question_id=candidate.question_id, + video_id=candidate.video_id, + task_type=candidate.task_type, + question=candidate.question, + options=options if options is not None else candidate.options, + answer=answer if answer is not None else candidate.answer, + source_nodes=candidate.source_nodes, + difficulty=candidate.difficulty, + family=family, + skill_target=candidate.skill_target, + difficulty_steps=None, + sub_pattern=sub_pattern, + ) +``` + +- [ ] **Step 5: 在 `_process_one_slot` 接受点传入 sub_pattern** + +`app/question_gen/pipeline_v2.py` Phase 8 接受构造处（约 529 行）： + +```python + result = _to_generated_question( + candidate, + family=strategy.strategy_name, + options=pp.options, + answer=pp.answer, + sub_pattern=sub_pattern.name if sub_pattern else None, + ) +``` + +- [ ] **Step 6: 跑单元测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_generated_question_sub_pattern.py -v` +Expected: PASS + +- [ ] **Step 7: 写 JSON 持久化测试** + +在同文件追加，验证 `_on_accept` 与 `_append_to_json` 写出 `sub_pattern` 键。因 `_on_accept` 是闭包，测试改为直接测公共写函数 `_append_to_json`（同格式）： + +```python +def test_append_to_json_writes_sub_pattern(tmp_path): + from tools.generate_questions import _append_to_json + q = GeneratedQuestion( + question_id="v1_Action Recognition_0001", video_id="v1", + task_type="Action Recognition", question="?", + options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer="A", + source_nodes=("n1",), difficulty="hard", + family="ACTION_RECOGNITION", skill_target="M1_AR", + sub_pattern="temporal_reasoning_failure", + ) + _append_to_json(tmp_path, q) + import json + data = json.loads((tmp_path / "v1.json").read_text(encoding="utf-8")) + assert data[0]["sub_pattern"] == "temporal_reasoning_failure" +``` + +- [ ] **Step 8: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_generated_question_sub_pattern.py::test_append_to_json_writes_sub_pattern -v` +Expected: FAIL（`_append_to_json` 的 entry dict 无 `sub_pattern`） + +- [ ] **Step 9: 改 `tools/generate_questions.py` 两处写 dict** + +`_append_to_json` 的 `entry` dict（约 344 行）与 `_on_accept` 内的 append dict（约 1123 行），各补一行 `"sub_pattern": q.sub_pattern,`（放在 `"skill_target"` 之后）。示例（`_append_to_json`）： + +```python + entry = { + "question_id": question.question_id, + "task_type": question.task_type, + "question": question.question, + "options": list(question.options), + "answer": question.answer, + "source_nodes": list(question.source_nodes), + "difficulty": question.difficulty, + "family": question.family, + "skill_target": question.skill_target, + "sub_pattern": question.sub_pattern, + } +``` + +`_on_accept` 内同理，在 `"skill_target": q.skill_target,` 后加 `"sub_pattern": q.sub_pattern,`。 + +- [ ] **Step 10: 跑测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_generated_question_sub_pattern.py -v` +Expected: PASS（3+1 全绿） + +- [ ] **Step 11: 提交** + +```bash +git add core/types.py app/question_gen/pipeline_v2.py tools/generate_questions.py tests/unit/test_generated_question_sub_pattern.py +git commit -m "feat: thread and persist sub_pattern into accepted questions" +``` + +--- + +## Task 3: `uses_grounded_selector` 策略开关（路径隔离核心） + +**Files:** +- Modify: `app/question_gen/strategy.py`（Protocol 加 property；`BaseTaskTypeStrategy` 默认 `False`） +- Modify: `app/question_gen/strategy_action_recognition.py`（`ActionRecognitionStrategy` 覆盖为 `True`） +- Test: `tests/unit/test_strategy_grounded_flag.py`（新建） + +- [ ] **Step 1: 写失败测试** + +```python +"""uses_grounded_selector 分流：仅 AR=True，其余 11 类=False。""" + +from app.question_gen.strategy import get_strategy + +_NON_AR = [ + "Action Reasoning", "Attribute Perception", "Counting Problem", + "Information Synopsis", "Object Recognition", "Object Reasoning", + "OCR Problems", "Spatial Perception", "Spatial Reasoning", + "Temporal Perception", "Temporal Reasoning", +] + + +def test_action_recognition_uses_grounded_selector(): + assert get_strategy("Action Recognition").uses_grounded_selector is True + + +def test_non_ar_do_not_use_grounded_selector(): + for tt in _NON_AR: + assert get_strategy(tt).uses_grounded_selector is False, tt +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_strategy_grounded_flag.py -v` +Expected: FAIL（`AttributeError: 'uses_grounded_selector'`） + +- [ ] **Step 3: 改 Protocol + Base** + +`app/question_gen/strategy.py`，在 `TaskTypeStrategy` Protocol 里（`leak_probe_template` 之后）加： + +```python + @property + def uses_grounded_selector(self) -> bool: ... +``` + +`BaseTaskTypeStrategy` 里加（返回 False）： + +```python + @property + def uses_grounded_selector(self) -> bool: + """默认不启用 grounded selector（11 类题型走原路径）。""" + return False +``` + +- [ ] **Step 4: 改 ActionRecognitionStrategy** + +`app/question_gen/strategy_action_recognition.py`，在类里加： + +```python + @property + def uses_grounded_selector(self) -> bool: + """AR 启用候选池 + VLM 视觉打分 selector。""" + return True +``` + +- [ ] **Step 5: 跑测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_strategy_grounded_flag.py -v` +Expected: PASS + +- [ ] **Step 6: 提交** + +```bash +git add app/question_gen/strategy.py app/question_gen/strategy_action_recognition.py tests/unit/test_strategy_grounded_flag.py +git commit -m "feat: add uses_grounded_selector strategy switch (AR only)" +``` + +--- + +## Task 4: `selector_scores` 观测列 + `update_selector_scores`（公共，纯数据） + +**Files:** +- Modify: `app/question_gen/run_store.py`（`_DDL_ITEMS` 加列 + 幂等 ALTER TABLE + 新方法） +- Test: `tests/unit/test_run_store_selector_scores.py`（新建） + +- [ ] **Step 1: 写失败测试** + +```python +"""selector_scores 列幂等迁移 + update_selector_scores 写入。""" + +import json +from app.question_gen.run_store import QuestionGenStore + + +def _store(tmp_path): + return QuestionGenStore(tmp_path / "q.db") + + +def test_selector_scores_column_exists(tmp_path): + store = _store(tmp_path) + cols = {r[1] for r in store._conn.execute("PRAGMA table_info(question_gen_items)")} + assert "selector_scores" in cols + store.close() + + +def test_update_selector_scores_writes_json(tmp_path): + store = _store(tmp_path) + store.record_run_start("run1", "sha", "{}") + store.record_item( + item_id="it1", run_id="run1", slot_id="s1", video_id="v1", + family="ACTION_RECOGNITION", task_type="Action Recognition", + skill_target="M1_AR", attempt=1, question_text="?", + sub_pattern="temporal_reasoning_failure", + ) + payload = {"correct_score": 0.8, "chosen": [0.7, 0.6, 0.55], + "pool_size": 24, "anneal_rounds": 0, "hard_fail": False} + store.update_selector_scores("it1", json.dumps(payload)) + row = store._conn.execute( + "SELECT selector_scores FROM question_gen_items WHERE item_id='it1'" + ).fetchone() + assert json.loads(row[0])["correct_score"] == 0.8 + store.close() + + +def test_update_selector_scores_unknown_item_raises(tmp_path): + store = _store(tmp_path) + try: + store.update_selector_scores("missing", "{}") + raise AssertionError("应抛 ValueError") + except ValueError: + pass + store.close() +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_run_store_selector_scores.py -v` +Expected: FAIL + +- [ ] **Step 3: 改 DDL + 幂等迁移** + +`app/question_gen/run_store.py`，`_DDL_ITEMS` 在 `difficulty_steps INTEGER,` 后加 `selector_scores TEXT,`（新建库直接带列）。`_init_schema` 里，仿照 `sub_pattern` 的幂等迁移追加： + +```python + cols = {r[1] for r in self._conn.execute("PRAGMA table_info(question_gen_items)")} + if "sub_pattern" not in cols: + self._conn.execute("ALTER TABLE question_gen_items ADD COLUMN sub_pattern TEXT") + self._conn.commit() + if "selector_scores" not in cols: + self._conn.execute("ALTER TABLE question_gen_items ADD COLUMN selector_scores TEXT") + self._conn.commit() +``` + +- [ ] **Step 4: 加 `update_selector_scores` 方法** + +在 `update_difficulty` 之后加： + +```python + def update_selector_scores(self, item_id: str, selector_scores_json: str) -> None: + """写入 grounded selector 打分观测（JSON 字符串）。 + + Parameters + ---------- + item_id : str + 题目唯一 ID。 + selector_scores_json : str + 观测 JSON：correct_score / chosen / pool_size / anneal_rounds / hard_fail。 + + Raises + ------ + ValueError + item_id 不存在时抛出。 + """ + cursor = self._conn.execute( + "UPDATE question_gen_items SET selector_scores=? WHERE item_id=?", + (selector_scores_json, item_id), + ) + self._conn.commit() + if cursor.rowcount == 0: + raise ValueError(f"item_id 不存在: {item_id}") +``` + +- [ ] **Step 5: 跑测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_run_store_selector_scores.py -v` +Expected: PASS + +- [ ] **Step 6: 提交** + +```bash +git add app/question_gen/run_store.py tests/unit/test_run_store_selector_scores.py +git commit -m "feat: add selector_scores observation column to question_gen_items" +``` + +--- + +## Task 5: `distractor_selector.py` — 候选池 + VLM 视觉打分 + 区间选择 + 退火 + +模块核心。纯逻辑（区间选择）单测；VLM 调用用 mock 集成测。新增两个版本化 prompt。 + +**Files:** +- Create: `app/question_gen/distractor_selector.py` +- Create: `store/prompts/question_gen/ar_distractor_pool.md` +- Create: `store/prompts/question_gen/ar_distractor_score.md` +- Test: `tests/unit/test_distractor_selector.py`（新建） + +### 5.1 纯逻辑：区间选择 + +- [ ] **Step 1: 写失败测试（区间选择）** + +```python +"""distractor_selector 区间选择纯逻辑。""" + +from app.question_gen.distractor_selector import _select_in_interval + + +def test_select_three_in_interval_by_highest_score(): + # correct=0.90, 区间 = [0.90-0.35, 0.90-0.05] = [0.55, 0.85] + cands = ["a", "b", "c", "d", "e"] + scores = [0.84, 0.70, 0.60, 0.50, 0.88] # e=0.88 太接近(>0.85)剔除, d=0.50 太低剔除 + chosen = _select_in_interval(0.90, cands, scores, delta_low=0.05, delta_high=0.35) + assert chosen == ["a", "b", "c"] # 落区间的按分数降序取 3（最难） + + +def test_select_returns_none_when_fewer_than_three(): + cands = ["a", "b"] + scores = [0.80, 0.70] + assert _select_in_interval(0.90, cands, scores, 0.05, 0.35) is None + + +def test_select_excludes_out_of_band(): + cands = ["hi", "lo", "ok1", "ok2", "ok3"] + scores = [0.89, 0.10, 0.80, 0.75, 0.70] # hi>上界, lo<下界 + chosen = _select_in_interval(0.90, cands, scores, 0.05, 0.35) + assert chosen == ["ok1", "ok2", "ok3"] +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_distractor_selector.py -v` +Expected: FAIL（模块不存在） + +- [ ] **Step 3: 建模块骨架 + 区间选择纯函数** + +新建 `app/question_gen/distractor_selector.py`： + +```python +"""Grounded 干扰项 selector — 候选池 + VLM 视觉打分 + 区间选择（仅 AR 路径）。 + +把干扰项从"VLM 主观写得像"下沉到机制层：VLM 生成 N 个候选干扰项，再对 +候选 + 正解逐一打"视觉可信度"分，按 [正解分-δ_high, 正解分-δ_low] 区间 +选 3 个 grounded near-miss，从机制上消灭 Easy-Options Bias。 + +设计: research-wiki/designs/2026-07-14-grounded-question-gen-phaseA-design.md §3 +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING + +from json_repair import repair_json +from loguru import logger + +if TYPE_CHECKING: + from app.question_gen.sampler_v2 import MaterialContext + from core.protocols import VLMProvider + +_PROMPTS_DIR = Path(__file__).resolve().parent.parent.parent / "store" / "prompts" / "question_gen" + + +@dataclass(frozen=True) +class SelectorConfig: + """selector 科研参数。 + + 属性: + candidate_pool_size: 首轮候选干扰项数 N。 + delta_low: 干扰项视觉分与正解的最小差（上界，太近=真歧义）。 + delta_high: 干扰项视觉分与正解的最大差（下界，太低=负空间）。 + max_delta_relax: δ_high 放宽次数上限（退火）。 + delta_relax_step: 每次放宽 δ_high 的增量。 + """ + + candidate_pool_size: int + delta_low: float + delta_high: float + max_delta_relax: int = 2 + delta_relax_step: float = 0.1 + + +@dataclass(frozen=True) +class SelectorOutcome: + """selector 产出。 + + 属性: + options: 重组后的四选项（A=正解），格式 ("A. ...", "B. ...", ...)。 + answer: 正解字母（恒为 "A"，后续 postprocess 洗牌）。 + observation: 打分观测 dict（落 run_store selector_scores）。 + """ + + options: tuple[str, ...] + answer: str + observation: dict + + +def _select_in_interval( + correct_score: float, + candidates: list[str], + candidate_scores: list[float], + delta_low: float, + delta_high: float, +) -> list[str] | None: + """从候选中选 3 个视觉分落 [correct-δ_high, correct-δ_low] 区间的干扰项。 + + 落区间者按分数降序取前 3（分数越高越接近正解=越难）。不足 3 个返回 None。 + + 参数: + correct_score: 正解视觉可信度分。 + candidates: 候选干扰项文本列表。 + candidate_scores: 与 candidates 对齐的视觉分列表。 + delta_low: 最小差（上界 = correct - delta_low）。 + delta_high: 最大差（下界 = correct - delta_high）。 + + 返回: + 选中的 3 个候选文本（降序）或 None（不足 3 个）。 + """ + upper = correct_score - delta_low + lower = correct_score - delta_high + eligible = [ + (c, s) + for c, s in zip(candidates, candidate_scores, strict=True) + if lower <= s <= upper + ] + if len(eligible) < 3: + return None + eligible.sort(key=lambda cs: cs[1], reverse=True) + return [c for c, _ in eligible[:3]] +``` + +- [ ] **Step 4: 跑区间选择测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_distractor_selector.py -v` +Expected: PASS（3 个区间选择测试） + +### 5.2 版本化 prompt + +- [ ] **Step 5: 建候选池 prompt** + +新建 `store/prompts/question_gen/ar_distractor_pool.md`： + +```markdown +You generate hard-negative distractor options for a video Action Recognition +multiple-choice question. + +## Given +- The question, the correct answer, subtitle context, and video frames. + +## Rules +- Produce distractors that are **grounded near-misses**: each MUST describe an + action/entity that genuinely appears in the video, differing from the correct + answer in exactly ONE dimension (timing, subject, manner, or object). +- NEVER invent events absent from the video ("negative space"). A distractor + that names something not shown is a failure. +- Each distractor must be a plausible answer to the question for someone who + only skimmed the video. +- Keep each distractor parallel in structure and length to the correct answer. + +## Output +Respond with ONLY a JSON object: +```json +{"distractors": ["...", "...", "..."]} +``` +Return exactly N distractors (N is given in the request). No option-letter +prefixes, just the raw text. +``` + +- [ ] **Step 6: 建打分 prompt** + +新建 `store/prompts/question_gen/ar_distractor_score.md`： + +```markdown +You are a strict visual grader for a video Action Recognition question. + +## Given +- The question, video frames, and a numbered list of candidate answer texts + (the first is the true answer; the rest are distractor candidates — but you + are NOT told which is which). + +## Task +For EACH candidate, judge how visually credible it is as an answer given ONLY +the frames — i.e. how strongly the frames could be read as supporting it. +Score in [0.0, 1.0]: 1.0 = frames strongly depict this; 0.0 = frames show no +trace of it (pure negative space). + +Judge visual groundedness ONLY. Do NOT reward the option for being the +"correct" answer — a good distractor is visually credible yet wrong. + +## Output +Respond with ONLY a JSON object mapping 1-based index to score, same order as +input: +```json +{"scores": [0.9, 0.7, 0.6, 0.3, 0.85]} +``` +Return exactly as many scores as candidates, in order. +``` + +### 5.3 VLM 编排 + 退火 + +- [ ] **Step 7: 写失败测试（编排，mock VLM）** + +在 `tests/unit/test_distractor_selector.py` 追加： + +```python +import pytest +from core.types import LLMResponse + + +class _FakeVLM: + """按队列返回预设响应的 mock VLM。""" + + def __init__(self, responses: list[str]): + self._responses = list(responses) + self.calls = 0 + + async def chat_with_images(self, messages, images, *, session_id=None, parent_call_id=None): + self.calls += 1 + content = self._responses.pop(0) + return LLMResponse( + content=content, thinking="", model="fake", provider="fake", + prompt_tokens=0, completion_tokens=0, latency_ms=0, + ttft_ms=None, max_inter_token_ms=None, cache_hit=False, call_id="c", + ) + + +class _Material: + subtitle_sentences = ["厨师先炒后蒸"] + frame_paths = ["/f1.jpg", "/f2.jpg"] + cross_l2_texts: list = [] + source_nodes = ("n1",) + + +@pytest.mark.asyncio +async def test_build_grounded_options_happy_path(): + from app.question_gen.distractor_selector import ( + SelectorConfig, build_grounded_options, + ) + pool = '{"distractors": ["炒", "煮", "炸", "烤"]}' + scores = '{"scores": [0.90, 0.80, 0.70, 0.60, 0.20]}' # 正解0.90; 炒0.80 煮0.70 炸0.60 落区间, 烤0.20 剔除 + vlm = _FakeVLM([pool, scores]) + cfg = SelectorConfig(candidate_pool_size=4, delta_low=0.05, delta_high=0.35) + out = await build_grounded_options( + vlm=vlm, question="厨师最终用哪种方式？", correct_text="蒸", + material=_Material(), config=cfg, session_id="s", + ) + assert out is not None + assert out.answer == "A" + assert out.options[0] == "A. 蒸" + assert {o[3:] for o in out.options[1:]} == {"炒", "煮", "炸"} + assert out.observation["hard_fail"] is False + + +@pytest.mark.asyncio +async def test_build_grounded_options_hard_fail_returns_none(): + from app.question_gen.distractor_selector import ( + SelectorConfig, build_grounded_options, + ) + # 所有候选都在负空间（分数极低），退火后仍不足 3 个 → None + pool = '{"distractors": ["x", "y", "z"]}' + scores = '{"scores": [0.90, 0.05, 0.04, 0.03]}' + pool2 = '{"distractors": ["p", "q", "r"]}' + scores2 = '{"scores": [0.90, 0.05, 0.04, 0.03, 0.05, 0.04, 0.03]}' + vlm = _FakeVLM([pool, scores, pool2, scores2, pool2, scores2]) + cfg = SelectorConfig(candidate_pool_size=3, delta_low=0.05, delta_high=0.35) + out = await build_grounded_options( + vlm=vlm, question="?", correct_text="蒸", + material=_Material(), config=cfg, session_id="s", + ) + assert out is None +``` + +- [ ] **Step 8: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_distractor_selector.py -v` +Expected: FAIL（`build_grounded_options` 未实现） + +- [ ] **Step 9: 实现 pool/score/编排** + +在 `distractor_selector.py` 追加。先补 prompt 加载与解析辅助，再 `build_grounded_options`： + +```python +def _load_prompt(name: str) -> str: + path = _PROMPTS_DIR / name + if not path.exists(): + msg = f"Prompt 模板不存在: {path}" + raise FileNotFoundError(msg) + return path.read_text(encoding="utf-8") + + +def _material_context_block(question: str, correct_text: str, material: MaterialContext) -> str: + parts = [f"## Question\n{question}", f"## Correct Answer\n{correct_text}"] + if material.subtitle_sentences: + parts.append("## Subtitles") + parts.extend(f" - {s}" for s in material.subtitle_sentences) + if getattr(material, "cross_l2_texts", None): + parts.append("## Cross-Segment Context") + parts.extend(f" - {t}" for t in material.cross_l2_texts) + return "\n".join(parts) + + +def _parse_json_object(raw: str) -> dict: + content = raw.strip() + if "```" in content: + for part in content.split("```"): + stripped = part.strip() + if stripped.startswith("json"): + stripped = stripped[4:].strip() + if stripped.startswith("{"): + content = stripped + break + data = json.loads(repair_json(content, return_objects=False)) + if not isinstance(data, dict): + msg = f"selector 响应顶层非 JSON 对象: {type(data).__name__}" + raise ValueError(msg) + return data + + +async def _generate_pool( + vlm: VLMProvider, question: str, correct_text: str, + material: MaterialContext, n: int, *, session_id: str, +) -> list[str]: + """VLM 生成 n 个候选干扰项文本。""" + system = _load_prompt("ar_distractor_pool.md") + user = _material_context_block(question, correct_text, material) + f"\n## N\nGenerate exactly {n} distractors." + messages = [{"role": "system", "content": system}, {"role": "user", "content": user}] + resp = await vlm.chat_with_images(messages, list(material.frame_paths), session_id=session_id) + data = _parse_json_object(resp.content) + raw = data.get("distractors", []) + if not isinstance(raw, list): + return [] + # 防御：去空、去重、剔除与正解字面相同者 + seen: set[str] = set() + out: list[str] = [] + for item in raw: + text = str(item).strip() + if not text or text == correct_text.strip() or text in seen: + continue + seen.add(text) + out.append(text) + return out + + +async def _score_options( + vlm: VLMProvider, question: str, options: list[str], + material: MaterialContext, *, session_id: str, +) -> list[float]: + """VLM 对 options（首个为正解）逐一打视觉可信度分 [0,1]，返回对齐分数列表。""" + system = _load_prompt("ar_distractor_score.md") + numbered = "\n".join(f"{i}. {opt}" for i, opt in enumerate(options, 1)) + user = f"## Question\n{question}\n\n## Candidates\n{numbered}" + messages = [{"role": "system", "content": system}, {"role": "user", "content": user}] + resp = await vlm.chat_with_images(messages, list(material.frame_paths), session_id=session_id) + data = _parse_json_object(resp.content) + scores_raw = data.get("scores", []) + if not isinstance(scores_raw, list) or len(scores_raw) != len(options): + msg = f"打分数量({len(scores_raw) if isinstance(scores_raw, list) else 'NA'}) != 选项数({len(options)})" + raise ValueError(msg) + return [max(0.0, min(1.0, float(s))) for s in scores_raw] + + +async def build_grounded_options( + vlm: VLMProvider, + question: str, + correct_text: str, + material: MaterialContext, + config: SelectorConfig, + *, + session_id: str, +) -> SelectorOutcome | None: + """生成候选池 → 视觉打分 → 区间选 3 干扰项 → 重组四选项。 + + 退火（凑不齐 3 个时按序）：① 扩池 N→2N 再打分；② 逐步放宽 δ_high； + ③ 仍不足返回 None（调用方走重出）。 + + 参数: + vlm: VLM 端口。 + question: 题干。 + correct_text: 正解文本（无字母前缀）。 + material: 采样素材（提供 frame_paths / subtitles）。 + config: selector 科研参数。 + session_id: 遥测会话 ID。 + + 返回: + SelectorOutcome（A=正解 + 3 grounded 干扰项）或 None（硬失败）。 + """ + candidates = await _generate_pool( + vlm, question, correct_text, material, config.candidate_pool_size, session_id=session_id + ) + # options[0] 恒为正解 + scored = await _score_options(vlm, question, [correct_text, *candidates], material, session_id=session_id) + correct_score, cand_scores = scored[0], scored[1:] + anneal_rounds = 0 + + chosen = _select_in_interval( + correct_score, candidates, cand_scores, config.delta_low, config.delta_high + ) + + # 退火 1: 扩池一轮 N→2N + if chosen is None: + anneal_rounds += 1 + more = await _generate_pool( + vlm, question, correct_text, material, config.candidate_pool_size, session_id=session_id + ) + more = [m for m in more if m not in candidates] + if more: + more_scores = await _score_options( + vlm, question, [correct_text, *more], material, session_id=session_id + ) + candidates = candidates + more + cand_scores = cand_scores + more_scores[1:] + chosen = _select_in_interval( + correct_score, candidates, cand_scores, config.delta_low, config.delta_high + ) + + # 退火 2: 放宽 δ_high（下界下移，纳入更低分候选），δ_low 不动 + relax = 0 + delta_high = config.delta_high + while chosen is None and relax < config.max_delta_relax: + relax += 1 + anneal_rounds += 1 + delta_high = delta_high + config.delta_relax_step + chosen = _select_in_interval( + correct_score, candidates, cand_scores, config.delta_low, delta_high + ) + + hard_fail = chosen is None + observation = { + "correct_score": correct_score, + "chosen": [ + cand_scores[candidates.index(c)] for c in (chosen or []) + ], + "pool_size": len(candidates), + "anneal_rounds": anneal_rounds, + "delta_high_final": delta_high, + "hard_fail": hard_fail, + } + if hard_fail: + logger.warning( + "grounded selector 硬失败: correct={:.3f}, pool={}, anneal={}", + correct_score, len(candidates), anneal_rounds, + ) + return None + + options = ( + f"A. {correct_text}", + f"B. {chosen[0]}", + f"C. {chosen[1]}", + f"D. {chosen[2]}", + ) + return SelectorOutcome(options=options, answer="A", observation=observation) +``` + +- [ ] **Step 10: 跑全模块测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_distractor_selector.py -v` +Expected: PASS（区间选择 3 + 编排 2） + +- [ ] **Step 11: 提交** + +```bash +git add app/question_gen/distractor_selector.py store/prompts/question_gen/ar_distractor_pool.md store/prompts/question_gen/ar_distractor_score.md tests/unit/test_distractor_selector.py +git commit -m "feat: add grounded distractor selector with visual scoring" +``` + +--- + +## Task 6: selector 接入 `_process_one_slot` + 配置参数（AR 路径） + +把 selector 织入 AR 出题：`generate_one_v2` 拿正解 → 提取正解文本 → `build_grounded_options` 重组四选项 → 失败则重出。配置走 `PipelineConfig` + YAML。 + +**Files:** +- Modify: `app/question_gen/pipeline_v2.py`（`PipelineConfig` 加 selector 字段；`load_pipeline_config` 读 YAML；`_process_one_slot` 织入） +- Modify: `config/question_gen_ar30.yaml`（补 selector 参数） +- Test: `tests/unit/test_pipeline_selector_wiring.py`（新建） + +- [ ] **Step 1: 加 PipelineConfig 字段 + 加载** + +`app/question_gen/pipeline_v2.py`，`PipelineConfig` 末尾加（带默认，兼容既有 YAML）： + +```python + seed: int + output_dir: Path + candidate_pool_size: int = 24 + selector_delta_low: float = 0.05 + selector_delta_high: float = 0.35 +``` + +`load_pipeline_config` 的 `return PipelineConfig(...)` 补三行（`.get` 读，缺省用设计默认）： + +```python + seed=int(section["seed"]), + output_dir=Path(section["output_dir"]), + candidate_pool_size=int(section.get("candidate_pool_size", 24)), + selector_delta_low=float(section.get("selector_delta_low", 0.05)), + selector_delta_high=float(section.get("selector_delta_high", 0.35)), +``` + +- [ ] **Step 2: 写失败测试（正解文本提取 + 织入分流）** + +新建 `tests/unit/test_pipeline_selector_wiring.py`。先测纯辅助 `_extract_correct_text`： + +```python +"""selector 织入辅助：正解文本提取 + 分流。""" + +from app.question_gen.pipeline_v2 import _extract_correct_text + + +def test_extract_correct_text_strips_prefix(): + options = ("A. 蒸", "B. 炒", "C. 煮", "D. 炸") + assert _extract_correct_text(options, "C") == "煮" + + +def test_extract_correct_text_handles_lowercase_answer(): + options = ("A. run", "B. walk", "C. jump", "D. sit") + assert _extract_correct_text(options, "b") == "walk" +``` + +- [ ] **Step 3: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_pipeline_selector_wiring.py -v` +Expected: FAIL（`_extract_correct_text` 不存在） + +- [ ] **Step 4: 加 `_extract_correct_text` 辅助** + +`app/question_gen/pipeline_v2.py`（`_to_generated_question` 附近）： + +```python +def _extract_correct_text(options: tuple[str, ...], answer: str) -> str: + """从四选项中取正解文本（去掉 "X. " 字母前缀）。 + + 参数: + options: 选项元组，格式 ("A. ...", "B. ...", ...)。 + answer: 正解字母（大小写不敏感）。 + + 返回: + 正解选项去前缀后的文本。 + """ + idx = ord(answer.strip().upper()) - ord("A") + if not 0 <= idx < len(options): + msg = f"answer '{answer}' 超出选项范围 (n={len(options)})" + raise ValueError(msg) + opt = options[idx] + prefix = f"{answer.strip().upper()}. " + return opt[len(prefix):] if opt.startswith(prefix) else opt +``` + +- [ ] **Step 5: 织入 selector 到 `_process_one_slot`** + +在 Phase 2（`generate_one_v2` 得到 `candidate`）与 Phase 3（`record_item`）之间不变；在 Phase 3 之后、Phase 4（postprocess）之前，插入 grounded 分支。用 `strategy.uses_grounded_selector` 分流；失败走 `continue`。注意：selector 成功后需用重组选项**替换 candidate 的 options/answer** 再进 postprocess。 + +```python + # Phase 3.5: grounded selector（仅 AR 路径） + if strategy.uses_grounded_selector: + from app.question_gen.distractor_selector import ( + SelectorConfig, + build_grounded_options, + ) + + correct_text = _extract_correct_text(candidate.options, candidate.answer) + selector_cfg = SelectorConfig( + candidate_pool_size=config.candidate_pool_size, + delta_low=config.selector_delta_low, + delta_high=config.selector_delta_high, + ) + try: + outcome = await build_grounded_options( + vlm=vlm, + question=candidate.question, + correct_text=correct_text, + material=material, + config=selector_cfg, + session_id=session_id, + ) + except (ValueError, FileNotFoundError) as e: + logger.warning("slot {} selector 异常 (attempt {}): {}", slot.slot_id, attempt, e) + prev_reason = f"selector_error: {e}" + continue + + if outcome is None: + prev_reason = "grounded 干扰项不足（selector 硬失败）" + store.mark_item_rejected(item_id, prev_reason) + logger.info("slot {} selector 硬失败 (attempt {})", slot.slot_id, attempt) + continue + + store.update_selector_scores(item_id, json.dumps(outcome.observation, ensure_ascii=False)) + # 用 grounded 四选项替换候选（frozen → 构造新实例） + candidate = _replace_candidate_options(candidate, outcome.options, outcome.answer) +``` + +在文件顶部 `import` 区补 `import json`（若未导入）。并加辅助： + +```python +def _replace_candidate_options( + candidate: CandidateQuestion, options: tuple[str, ...], answer: str +) -> CandidateQuestion: + """用 selector 重组的选项/答案替换候选（CandidateQuestion frozen）。""" + return CandidateQuestion( + question_id=candidate.question_id, + video_id=candidate.video_id, + task_type=candidate.task_type, + skill_target=candidate.skill_target, + question=candidate.question, + options=options, + answer=answer, + source_nodes=candidate.source_nodes, + difficulty=candidate.difficulty, + subtitle_sentences=candidate.subtitle_sentences, + frame_paths=candidate.frame_paths, + ) +``` + +- [ ] **Step 6: 写织入集成测试（mock VLM 分流）** + +在 `tests/unit/test_pipeline_selector_wiring.py` 追加一个断言：非 AR 题型不触发 selector（源码守卫——`uses_grounded_selector` 分支只在 True 时进入）。用轻量单测覆盖 `_replace_candidate_options`： + +```python +def test_replace_candidate_options(): + from app.question_gen.generator_v2 import CandidateQuestion + from app.question_gen.pipeline_v2 import _replace_candidate_options + c = CandidateQuestion( + question_id="q", video_id="v", task_type="Action Recognition", + skill_target="M1_AR", question="?", + options=("A. a", "B. b", "C. c", "D. d"), answer="A", + source_nodes=("n1",), difficulty="hard", + ) + new = _replace_candidate_options(c, ("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), "A") + assert new.options == ("A. 蒸", "B. 炒", "C. 煮", "D. 炸") + assert new.question == "?" # 其余字段不变 + assert new.source_nodes == ("n1",) +``` + +- [ ] **Step 7: 跑测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_pipeline_selector_wiring.py -v` +Expected: PASS + +- [ ] **Step 8: 补 YAML 配置** + +`config/question_gen_ar30.yaml` 的 `question_gen_v2` 区段补三行（值用设计默认）： + +```yaml + candidate_pool_size: 24 + selector_delta_low: 0.05 + selector_delta_high: 0.35 +``` + +- [ ] **Step 9: 全量出题相关单测回归** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_generate_questions.py tests/unit/test_families.py tests/unit/test_gate.py -v` +Expected: PASS（非 AR 路径不受影响） + +- [ ] **Step 10: 提交** + +```bash +git add app/question_gen/pipeline_v2.py config/question_gen_ar30.yaml tests/unit/test_pipeline_selector_wiring.py +git commit -m "feat: wire grounded selector into AR slot processing" +``` + +--- + +## Task 7: 单维反事实约束（仅 AR，SubPattern 内容） + +改 6 个 SubPattern 的 `instruction` + `distractor_rules`，硬约束"干扰项必须是视频中真实发生、仅在单一维度（时点/主体/方式/对象）与正解不同，严禁缺席事件"。纯 prompt 内容，行为由 selector 保障，此处强化生成端引导。 + +**Files:** +- Modify: `app/question_gen/strategy_action_recognition.py`（6 个 SubPattern 的 `distractor_rules`） +- Test: `tests/unit/test_ar_sub_pattern_counterfactual.py`（新建） + +- [ ] **Step 1: 写失败测试** + +```python +"""AR 6 个 SubPattern 的 distractor_rules 含单维反事实约束关键词。""" + +from app.question_gen.strategy_action_recognition import AR_SUB_PATTERNS + +_REQUIRED = ["真实", "单一维度"] # 每个 distractor_rules 都需强调 grounded + 单维 + + +def test_all_sub_patterns_enforce_single_dimension_counterfactual(): + for sp in AR_SUB_PATTERNS: + rules = sp.distractor_rules + assert "真实" in rules, sp.name # 干扰项须真实发生 + assert ("单一维度" in rules or "只在" in rules or "仅在" in rules), sp.name + + +def test_sub_pattern_count_unchanged(): + assert len(AR_SUB_PATTERNS) == 6 +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_ar_sub_pattern_counterfactual.py -v` +Expected: FAIL + +- [ ] **Step 3: 改 6 个 SubPattern 的 distractor_rules** + +`app/question_gen/strategy_action_recognition.py`，逐一替换（按设计 §3.4 的反事实维度表）。示例——`_TEMPORAL_REASONING_FAILURE`（事件顺序维）： + +```python + distractor_rules=( + "干扰项必须是视频中真实发生的事件，仅在【事件时序/顺序】这一单一维度上与正解不同——" + "即同一组真实事件的错误排列或错误的第 N 次定位。" + "严禁使用视频中未出现的缺席事件作为干扰项。" + ), +``` + +`_PREMATURE_EVIDENCE_ANCHORING`（时点维）： + +```python + distractor_rules=( + "将视频前段真实出现的局部匹配动作设为强干扰项——它真实发生，" + "仅在【时点】这一单一维度上与正解不同（前段 vs 最终结论段）。" + "其余干扰项亦须是视频中真实发生的动作，严禁缺席事件。" + ), +``` + +`_SEMANTIC_RIGIDITY`（表述维）： + +```python + distractor_rules=( + "干扰项须基于视频真实内容，仅在【表述/语义】这一单一维度上做文章：" + "保留一个复用视频原始字幕字面、但在题干限定下语义为假的选项作为陷阱，" + "其余选项描述真实动作的不同同义表述。严禁凭空编造缺席动作。" + ), +``` + +`_FINE_GRAINED_VISUAL_ACTION`（方式维）： + +```python + distractor_rules=( + "四个选项须是【同一大类动作】的不同执行方式，全部为视频中真实可见的做法，" + "仅在【执行方式】这一单一维度上不同（如顺/逆时针、扳手/螺丝刀）。" + "严禁使用明显不相关或视频中未出现的动作作为干扰项。" + ), +``` + +`_CROSS_SEGMENT_ENTITY_TRACKING`（主体维）： + +```python + distractor_rules=( + "干扰项须是视频中【另一真实实体】在相应片段真实做过的动作，" + "仅在【动作主体】这一单一维度上与正解不同；或构造只覆盖部分片段的真实子集。" + "严禁编造任何实体未做过的缺席动作。" + ), +``` + +`_EVIDENCE_GAP_CONFABULATION`（因果完整性维）： + +```python + distractor_rules=( + "正解仅陈述视频中可观测的事实或诚实承认证据不足；" + "干扰项在【因果完整性】这一单一维度上越界——补上一段视频未展示的因果链，" + "但其前提元素仍取自视频真实内容（诱导 Agent 顺势编造），而非完全凭空的缺席事件。" + ), +``` + +同时在每个 SubPattern 的 `instruction` 末尾（可选）加一句"干扰项遵循单维反事实、不得缺席"的提醒——但测试只校验 `distractor_rules`，此步以 `distractor_rules` 为准。 + +- [ ] **Step 4: 跑测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_ar_sub_pattern_counterfactual.py -v` +Expected: PASS + +- [ ] **Step 5: 回归 AR 策略既有测试** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/ -k "action or strategy or families" -v` +Expected: PASS + +- [ ] **Step 6: 提交** + +```bash +git add app/question_gen/strategy_action_recognition.py tests/unit/test_ar_sub_pattern_counterfactual.py +git commit -m "feat: enforce single-dimension counterfactual in AR distractor rules" +``` + +--- + +## Task 8: multi_true 门 rubric 松绑（公共，12 类统一） + +放行"错误选项有局部真实证据、但在题干限定（同主体/时点/方式/对象）下为假"的近似干扰项——否则 grounded 干扰项会被 multi_true 误毙。 + +**Files:** +- Modify: `store/prompts/question_gen/gate_multi_true.md` +- Test: `tests/unit/test_gate_multi_true_rubric.py`（新建） + +- [ ] **Step 1: 写失败测试** + +```python +"""multi_true rubric 已松绑为题干限定判据。""" + +from pathlib import Path + +_PROMPT = Path(__file__).resolve().parents[2] / "store" / "prompts" / "question_gen" / "gate_multi_true.md" + + +def test_rubric_uses_qualifier_constraint(): + text = _PROMPT.read_text(encoding="utf-8") + # 新 rubric 必须提到"题干限定下同时为真才 fail"（qualifier / under the question's constraints） + assert "under the question" in text.lower() or "qualifier" in text.lower() + # 必须显式放行"有局部真实证据但在限定下为假"的干扰项 + assert "partial" in text.lower() or "locally" in text.lower() +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_multi_true_rubric.py -v` +Expected: FAIL + +- [ ] **Step 3: 改 rubric** + +`store/prompts/question_gen/gate_multi_true.md` 的 `## Instructions` 段替换为： + +```markdown +## Instructions + +1. Read the source material and the question carefully. Note the question's + explicit qualifiers (subject, timing, manner, object). +2. For each option, assess whether it is **fully correct under the question's + qualifiers** — not merely whether it has some partial or local support in + the source. +3. A good hard-negative distractor MAY have partial/local evidence in the video + yet be FALSE under the question's constraints. Such an option is NOT a + second correct answer — do NOT fail the question for it. +4. Verdict "fail" ONLY IF two or more options are each fully correct under the + question's qualifiers (a genuine ambiguity). +5. Otherwise verdict "pass". +``` + +- [ ] **Step 4: 跑测试确认通过** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_multi_true_rubric.py -v` +Expected: PASS + +- [ ] **Step 5: 回归 gate 测试** + +Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate.py -v` +Expected: PASS + +- [ ] **Step 6: 提交** + +```bash +git add store/prompts/question_gen/gate_multi_true.md tests/unit/test_gate_multi_true_rubric.py +git commit -m "feat: loosen multi_true gate to qualifier-scoped correctness" +``` + +--- + +## Task 9: 全量回归 + lint + wiki 收口 + +**Files:** +- 无新代码；运行验证 + wiki 登记。 + +- [ ] **Step 1: 全量测试** + +Run: `conda run -n Video-Tree-TRM pytest tests/ -q` +Expected: 全绿（含既有 1200+ 用例，证明 11 非 AR 题型行为不变）。若有红，回到对应 Task 修复。 + +- [ ] **Step 2: lint** + +Run: `conda run -n Video-Tree-TRM ruff check app/ core/ --fix && conda run -n Video-Tree-TRM ruff format app/ core/` +Expected: 无剩余错误。 + +- [ ] **Step 3: wiki 登记 plan 实体** + +```bash +conda run -n Video-Tree-TRM python3 .claude/tools/research_wiki.py add_entity research-wiki/ --type plan --id grounded-question-gen-phaseA --title "Grounded Question-Gen Phase A" +conda run -n Video-Tree-TRM python3 .claude/tools/research_wiki.py add_edge research-wiki/ --from "plan:grounded-question-gen-phaseA" --to "design:grounded-question-gen-phaseA" --type implements --evidence "Phase A 实现计划" +conda run -n Video-Tree-TRM python3 .claude/tools/research_wiki.py rebuild_index research-wiki/ +``` + +- [ ] **Step 4: 提交** + +```bash +git add research-wiki/ +git commit -m "docs: register Phase A plan in research wiki" +``` + +--- + +## Self-Review 与保真校验 + +**核心算法保真**：本计划改动局限于出题（question_gen）的候选池/打分/门控 rubric/数据透传，**不涉及** `research-wiki/ARCHITECTURE.md §6` 的 12 项核心算法（建树 4 项 + 训练 8 项）。出题四门 gate 非核心算法清单成员。**保真校验不适用**。 + +**Spec 覆盖**：L0 tree bug→Task1；uses_grounded_selector→Task3；候选池+VLM打分+区间+退火→Task5；单维反事实→Task7；multi_true松绑→Task8；sub_pattern持久化→Task2；selector观测列→Task4；接入+配置→Task6；回归→Task9。设计 §7 三个配置参数→Task6 Step1/8。 + +**路径隔离**：Task3 建开关，Task6 用开关分流，Task5 模块只被 AR 分支调用；Task1/2/4/8 为公共纯 bug/数据/rubric。每个改行为的 Task 都含非 AR 回归步骤。