1466 lines
57 KiB
Markdown
1466 lines
57 KiB
Markdown
# 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、`<type>: <desc>`，**禁止任何 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: 写共享序列化测试**
|
||
|
||
`_append_to_json` 与 `_on_accept` 是两条独立写路径（后者才写 `accepted_questions.json`），各自维护一份 entry dict——易漏改一处而测试不红。抽共享函数 `_question_to_entry(q) -> dict` 供两处复用，直接测它保证两条路径都含 `sub_pattern`：
|
||
|
||
```python
|
||
def test_question_to_entry_includes_sub_pattern():
|
||
from tools.generate_questions import _question_to_entry
|
||
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",
|
||
)
|
||
entry = _question_to_entry(q)
|
||
assert entry["sub_pattern"] == "temporal_reasoning_failure"
|
||
assert entry["question_id"] == "v1_Action Recognition_0001"
|
||
assert entry["options"] == ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"]
|
||
|
||
|
||
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 -k "entry or append" -v`
|
||
Expected: FAIL（`_question_to_entry` 不存在）
|
||
|
||
- [ ] **Step 9: 抽共享 `_question_to_entry` + 两处复用**
|
||
|
||
`tools/generate_questions.py`，在 `_append_to_json` 之前加共享函数：
|
||
|
||
```python
|
||
def _question_to_entry(question: GeneratedQuestion) -> dict:
|
||
"""将题目序列化为 JSON entry（_append_to_json 与 _on_accept 共用）。"""
|
||
return {
|
||
"question_id": question.question_id,
|
||
"video_id": question.video_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,
|
||
}
|
||
```
|
||
|
||
`_append_to_json` 内 `entry = {...}` 整体替换为 `entry = _question_to_entry(question)`。`_on_accept` 内 `existing.append({...})` 整体替换为 `existing.append(_question_to_entry(q))`。
|
||
|
||
> 注：`_append_to_json` 原 entry 不含 `video_id` 键（按 video 分文件），改用共享函数后会多出 `video_id` 键——无害（下游按需取键），且与 `accepted_questions.json` 格式统一。若下游有严格 schema 校验，保留两函数但都调用 `_question_to_entry` 后 `entry.pop("video_id", None)`；实现时确认下游读取无强约束即可直接统一。
|
||
|
||
- [ ] **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 + 新方法）
|
||
- Modify: `research-wiki/schemas/question-gen-items.md`（登记 `selector_scores` 列 + JSON 结构）
|
||
- 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: 登记 schema 文档**
|
||
|
||
`research-wiki/schemas/question-gen-items.md` 的列清单补一行 `selector_scores TEXT`，并说明其 JSON 结构：`{correct_score: float, chosen: float[], pool_size: int, anneal_rounds: int, delta_high_final: float, hard_fail: bool}`。若文档用表格，追加一行；保持与既有 `sub_pattern` 条目同风格。
|
||
|
||
- [ ] **Step 7: 提交**
|
||
|
||
```bash
|
||
git add app/question_gen/run_store.py research-wiki/schemas/question-gen-items.md 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 产出。observation 始终存在（含 hard-fail），供 run_store 落库。
|
||
|
||
属性:
|
||
observation: 打分观测 dict（correct_score/chosen/pool_size/anneal_rounds/hard_fail）。
|
||
options: 重组四选项（A=正解），hard-fail 时为 None。
|
||
answer: 正解字母（恒 "A"），hard-fail 时为 None。
|
||
"""
|
||
|
||
observation: dict
|
||
options: tuple[str, ...] | None = None
|
||
answer: str | None = None
|
||
|
||
@property
|
||
def hard_fail(self) -> bool:
|
||
"""是否硬失败（凑不齐 3 个 grounded 干扰项）。"""
|
||
return self.options is None
|
||
|
||
|
||
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.hard_fail is False
|
||
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_keeps_observation():
|
||
from app.question_gen.distractor_selector import (
|
||
SelectorConfig, build_grounded_options,
|
||
)
|
||
# 所有候选都在负空间（分数极低），退火后仍不足 3 个 → hard_fail。
|
||
# VLM 只被调 2 次（首轮 pool+score）+ 1 次退火 pool + 1 次退火 score = 4 次；
|
||
# δ_high 放宽轮次是纯重选，不调 VLM。退火 pool 打分含正解，共 4 个分数。
|
||
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]}'
|
||
vlm = _FakeVLM([pool, scores, 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.hard_fail is True
|
||
assert out.options is None
|
||
assert out.observation["hard_fail"] is True
|
||
assert out.observation["pool_size"] == 6 # 首轮 3 + 退火追加 3
|
||
```
|
||
|
||
- [ ] **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:
|
||
"""生成候选池 → 视觉打分 → 区间选 3 干扰项 → 重组四选项。
|
||
|
||
退火（凑不齐 3 个时按序）：① 追加 N 个候选使池达 2N 再打分；② 逐步放宽
|
||
δ_high（纯重选，不再调 VLM）；③ 仍不足则 hard_fail（调用方走重出）。
|
||
|
||
参数:
|
||
vlm: VLM 端口。
|
||
question: 题干。
|
||
correct_text: 正解文本（无字母前缀）。
|
||
material: 采样素材（提供 frame_paths / subtitles）。
|
||
config: selector 科研参数。
|
||
session_id: 遥测会话 ID。
|
||
|
||
返回:
|
||
SelectorOutcome。成功时 options=A 正解+3 grounded 干扰项；hard_fail
|
||
时 options=None，但 observation 始终存在供落库。
|
||
"""
|
||
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,
|
||
)
|
||
# observation 仍返回，供 pipeline 落 selector_scores（设计 §3.3 退化观测）
|
||
return SelectorOutcome(observation=observation)
|
||
|
||
options = (
|
||
f"A. {correct_text}",
|
||
f"B. {chosen[0]}",
|
||
f"C. {chosen[1]}",
|
||
f"D. {chosen[2]}",
|
||
)
|
||
return SelectorOutcome(observation=observation, options=options, answer="A")
|
||
```
|
||
|
||
- [ ] **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)),
|
||
```
|
||
|
||
**同步修 CLI seed override**：`tools/generate_questions.py:900` 的 `--seed` 覆盖手工重建 `PipelineConfig`，只复制旧字段会把 selector 三参重置为默认。补三行：
|
||
|
||
```python
|
||
config = PipelineConfig(
|
||
per_type=config.per_type,
|
||
retry_limit=config.retry_limit,
|
||
heavy_sample_rate=config.heavy_sample_rate,
|
||
dedup_threshold=config.dedup_threshold,
|
||
concurrency=config.concurrency,
|
||
seed=args.seed,
|
||
output_dir=config.output_dir,
|
||
candidate_pool_size=config.candidate_pool_size,
|
||
selector_delta_low=config.selector_delta_low,
|
||
selector_delta_high=config.selector_delta_high,
|
||
)
|
||
```
|
||
|
||
> 更稳健的等价写法是 `dataclasses.replace(config, seed=args.seed)`；若采用请在文件顶部 `import dataclasses` 或 `from dataclasses import replace`。二选一即可，实现时保持一致。
|
||
|
||
- [ ] **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
|
||
|
||
# observation 始终落库（含 hard-fail），供 EOB 退化观测与调参
|
||
store.update_selector_scores(
|
||
item_id, json.dumps(outcome.observation, ensure_ascii=False)
|
||
)
|
||
|
||
if outcome.hard_fail:
|
||
prev_reason = "grounded 干扰项不足（selector 硬失败）"
|
||
store.mark_item_rejected(item_id, prev_reason)
|
||
logger.info("slot {} selector 硬失败 (attempt {})", slot.slot_id, attempt)
|
||
continue
|
||
|
||
# 用 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: 更新 AR 集成测试 MockVLM（selector 启用后必须能应答 pool/score）**
|
||
|
||
`tests/integration/test_pipeline_v2.py` 的 `MockVLM.chat_with_images` 现只区分门控（含 "verdict"）与生成。selector 启用后 AR 路径会额外发 pool 请求（system prompt 含 "distractor"）和 score 请求（含 "grader" / "scores"）。若不识别，pool 会解析成候选 JSON → `_generate_pool` 得空列表 → hard-fail → AR slot 全拒，破坏既有断言。改 `chat_with_images` 顶部按 system prompt 关键词分流：
|
||
|
||
```python
|
||
async def chat_with_images(self, messages, images, *, session_id=None, parent_call_id=None):
|
||
prompt_text = str(messages)
|
||
system_text = messages[0].get("content", "") if messages else ""
|
||
if "verdict" in prompt_text.lower():
|
||
return _make_llm_response(self._gate_response)
|
||
if "distractor" in system_text.lower() and "grader" not in system_text.lower():
|
||
# 候选池请求：返回 4 个 grounded 干扰项
|
||
return _make_llm_response('{"distractors": ["蒸", "煮", "炸", "烤"]}')
|
||
if "grader" in system_text.lower():
|
||
# 打分请求：正解高分、3 个落区间、1 个负空间
|
||
return _make_llm_response('{"scores": [0.90, 0.80, 0.70, 0.60, 0.20]}')
|
||
idx = min(self._gen_count, len(self._responses) - 1)
|
||
self._gen_count += 1
|
||
return _make_llm_response(self._responses[idx])
|
||
```
|
||
|
||
> 打分响应长度需匹配"正解 + 候选数"。若某测试自定义候选池大小，须相应调整该 mock（打分列表长度 = pool 返回的干扰项数 + 1）。默认候选 JSON 4 个 → 打分 5 个，与上面一致。
|
||
|
||
- [ ] **Step 10: 全量出题相关单测 + AR 集成回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM pytest tests/integration/test_pipeline_v2.py tests/unit/test_generate_questions.py tests/unit/test_families.py tests/unit/test_gates.py -v`
|
||
Expected: PASS（AR 集成经 selector 仍通过；非 AR 路径不受影响）
|
||
|
||
- [ ] **Step 11: 提交**
|
||
|
||
```bash
|
||
git add app/question_gen/pipeline_v2.py config/question_gen_ar30.yaml tools/generate_questions.py tests/unit/test_pipeline_selector_wiring.py tests/integration/test_pipeline_v2.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 误毙。
|
||
|
||
> **这是本计划唯一一处授权的公共路径行为变更**（用户在 brainstorming 明确答复"全局松绑"，见设计 §3.5）。它作用于全部 12 题型的 multi_true 门。**无回归破坏风险**：`_gate_multi_true`（`app/question_gen/gates.py:363`）加载 prompt 后调 `llm.chat`，`tests/unit/test_gates.py` 用 mock LLM 返回固定 verdict、不校验 prompt 内容，故 rubric 文案变更不会使既有非 AR 门控测试变红。
|
||
|
||
**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_gates.py -v`
|
||
Expected: PASS（mock LLM，rubric 文案变更不影响判定断言）
|
||
|
||
- [ ] **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 回归步骤。
|