# 赛题生成工具实现计划 > **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 实现基于视频树的题目合成工具（generate + calibrate），含推理依赖 factory 提取。 **Architecture:** `app/question_gen/synthesizer.py` 承载核心业务逻辑（节点采样、prompt 构造、去重），`app/harness/factory.py` 提取推理依赖组装（TreeEnvironment + SearchToolDispatcher + PromptManager），`tools/generate_questions.py` 作为 CLI 壳编排并发和 I/O。 **Tech Stack:** Python 3.11, asyncio, GovernedVLMClient, EmbeddingProvider, scipy.stats.fisher_exact, loguru **核心算法保真校验：** 本计划不涉及核心算法迁移（13 项均已在先前 PR 完成），保真校验不适用。 --- ## 文件结构总览 | 动作 | 文件 | 职责 | |------|------|------| | 修改 | `app/harness/config.py:24` | 前置修复 `_VIDEO_MME_TASK_TYPE_COUNT` 11→12 | | 新建 | `app/question_gen/synthesizer.py` | 出题核心逻辑 | | 新建 | `app/harness/factory.py` | 推理依赖组装 | | 新建 | `tools/generate_questions.py` | CLI 壳 | | 新建 | `tests/unit/test_synthesizer.py` | synthesizer 单测 | | 新建 | `tests/unit/test_factory.py` | factory 单测 | | 新建 | `tests/unit/test_generate_questions.py` | CLI 集成测试 | | 修改 | `app/question_gen/__init__.py` | 追加 synthesizer re-export | --- ### Task 0: 前置修复 _VIDEO_MME_TASK_TYPE_COUNT **Files:** - Modify: `app/harness/config.py:24` - Test: `tests/unit/test_harness_config.py` - [ ] **Step 1: 写失败测试** 在 `tests/unit/test_harness_config.py` 中追加： ```python def test_video_mme_task_type_count_is_12(): """Video-MME 实际有 12 种题型，常量必须与之一致。""" from app.harness.config import _VIDEO_MME_TASK_TYPE_COUNT assert _VIDEO_MME_TASK_TYPE_COUNT == 12 ``` - [ ] **Step 2: 运行测试确认失败** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_harness_config.py::test_video_mme_task_type_count_is_12 -v ``` 预期：FAIL，`assert 11 == 12` - [ ] **Step 3: 修改常量** `app/harness/config.py:24`：`_VIDEO_MME_TASK_TYPE_COUNT = 11` → `_VIDEO_MME_TASK_TYPE_COUNT = 12` - [ ] **Step 4: 运行测试确认通过** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_harness_config.py -v ``` 预期：全部 PASS - [ ] **Step 5: 提交** ```bash git add app/harness/config.py tests/unit/test_harness_config.py git commit -m "fix(config): _VIDEO_MME_TASK_TYPE_COUNT 11→12，Video-MME 实际有 12 种题型" ``` --- ### Task 1: synthesizer.py — AnchorContext + 题型映射常量 **Files:** - Create: `app/question_gen/synthesizer.py` - Create: `tests/unit/test_synthesizer.py` - [ ] **Step 1: 写失败测试 — 题型映射完整性** ```python # tests/unit/test_synthesizer.py from app.question_gen.synthesizer import TASK_TYPE_LEVEL_MAP, AnchorContext ALL_12_TYPES = [ "Object Recognition", "Attribute Perception", "OCR Problems", "Spatial Reasoning", "Spatial Perception", "Action Recognition", "Action Reasoning", "Counting Problem", "Temporal Perception", "Temporal Reasoning", "Information Synopsis", "Object Reasoning", ] def test_task_type_level_map_covers_all_12_types(): """映射表必须覆盖全部 12 种 Video-MME 题型。""" assert set(TASK_TYPE_LEVEL_MAP.keys()) == set(ALL_12_TYPES) def test_anchor_context_frozen(): """AnchorContext 是不可变的。""" ctx = AnchorContext( node_id="L3_001", card_text="test", frame_paths=["a.jpg"], subtitle="", distractor_texts=["other node"], ) assert ctx.node_id == "L3_001" ``` - [ ] **Step 2: 运行测试确认失败** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v ``` 预期：ImportError - [ ] **Step 3: 实现 AnchorContext + TASK_TYPE_LEVEL_MAP** 创建 `app/question_gen/synthesizer.py`： ```python """赛题合成核心逻辑 — 节点采样、prompt 构造、去重。 纯函数为主，异步编排仅 generate_one。 通过 DI 接收 VLMProvider / EmbeddingProvider，不 import adapters/。 """ from __future__ import annotations from dataclasses import dataclass from typing import Any @dataclass(frozen=True) class AnchorContext: """锚节点上下文——生成单道题所需的全部素材。 属性: node_id: 锚节点 ID。 card_text: 锚节点 card 序列化文本。 frame_paths: 帧图片路径列表。 subtitle: 对应字幕（可空）。 distractor_texts: 同视频其他节点摘要（供 VLM 生成干扰项）。 """ node_id: str card_text: str frame_paths: list[str] subtitle: str distractor_texts: list[str] @dataclass(frozen=True) class TaskTypeSpec: """题型的生成规格。 属性: level: 锚定层级（"L3" / "L2" / "L1" / "L1-L2"）。 needs_frames: 是否必须提供帧图。 frame_count: 帧数范围描述（如 "1", "2-3", "0-1"）。 context_fields: 需要提取的 card 字段元组。 """ level: str needs_frames: bool frame_count: str context_fields: tuple[str, ...] TASK_TYPE_LEVEL_MAP: dict[str, TaskTypeSpec] = { "Object Recognition": TaskTypeSpec("L3", True, "1", ("frame_summary",)), "Attribute Perception": TaskTypeSpec("L3", True, "1", ("frame_summary",)), "OCR Problems": TaskTypeSpec("L3", True, "1", ("frame_summary",)), "Spatial Reasoning": TaskTypeSpec("L3", True, "1", ("frame_summary", "spatial_layout")), "Spatial Perception": TaskTypeSpec("L3", True, "1", ("frame_summary",)), "Action Recognition": TaskTypeSpec("L2", True, "2-3", ("event_description",)), "Action Reasoning": TaskTypeSpec("L2", True, "2-3", ("event_description",)), "Counting Problem": TaskTypeSpec("L2", True, "2-3", ("event_description",)), "Temporal Perception": TaskTypeSpec("L2", False, "0-1", ("event_description", "time_range")), "Temporal Reasoning": TaskTypeSpec("L1", True, "per-L2", ("scene_summary",)), "Information Synopsis": TaskTypeSpec("L1", True, "per-L2", ("scene_summary",)), "Object Reasoning": TaskTypeSpec("L1-L2", True, "per-L2", ("event_description",)), } ``` - [ ] **Step 4: 运行测试确认通过** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v ``` 预期：PASS - [ ] **Step 5: 提交** ```bash git add app/question_gen/synthesizer.py tests/unit/test_synthesizer.py git commit -m "feat(question_gen): AnchorContext + 12 题型-层级映射常量" ``` --- ### Task 2: synthesizer.py — sample_anchor **Files:** - Modify: `app/question_gen/synthesizer.py` - Modify: `tests/unit/test_synthesizer.py` - [ ] **Step 1: 写失败测试** ```python import json import random from pathlib import Path from app.tree.index import TreeIndex from app.question_gen.synthesizer import sample_anchor def _load_test_tree() -> tuple[TreeIndex, str]: """加载真实测试树（store/videos/ 下第一棵）。""" videos_dir = Path("store/videos") first_vid = sorted(videos_dir.iterdir())[0] tree = TreeIndex.load_json(str(first_vid / "tree.json")) return tree, first_vid.name class TestSampleAnchor: def test_l3_type_returns_single_frame(self): tree, vid = _load_test_tree() ctx = sample_anchor(tree, "Object Recognition", set(), random.Random(42)) assert len(ctx.frame_paths) == 1 assert ctx.node_id.startswith("L") assert len(ctx.distractor_texts) > 0 def test_l2_type_returns_multiple_frames(self): tree, vid = _load_test_tree() ctx = sample_anchor(tree, "Action Reasoning", set(), random.Random(42)) assert 2 <= len(ctx.frame_paths) <= 3 def test_temporal_perception_zero_or_one_frame(self): """Temporal Perception 帧数 0-1，且 card_text 含 time_range。""" tree, vid = _load_test_tree() ctx = sample_anchor(tree, "Temporal Perception", set(), random.Random(42)) assert len(ctx.frame_paths) <= 1 assert "time_range" in ctx.card_text.lower() or "time" in ctx.card_text.lower() def test_information_synopsis_uses_all_l2(self): """Information Synopsis 必须包含全部 L2 card（非采样子集）。""" tree, vid = _load_test_tree() ctx = sample_anchor(tree, "Information Synopsis", set(), random.Random(42)) total_l2 = sum(len(r.children) for r in tree.roots) # card_text 中应包含全部 L2 的事件描述 assert len(ctx.frame_paths) >= min(total_l2, 1) def test_l1_type_l2_nodes_in_time_order(self): """L1 题型的 L2 子节点应按时间顺序组织。""" tree, vid = _load_test_tree() ctx = sample_anchor(tree, "Temporal Reasoning", set(), random.Random(42)) assert len(ctx.frame_paths) >= 1 assert len(ctx.card_text) > 20 def test_used_node_ids_excluded(self): tree, vid = _load_test_tree() rng = random.Random(42) ctx1 = sample_anchor(tree, "Object Recognition", set(), rng) ctx2 = sample_anchor(tree, "Object Recognition", {ctx1.node_id}, random.Random(43)) assert ctx2.node_id != ctx1.node_id def test_insufficient_nodes_raises(self): tree, vid = _load_test_tree() all_l3_ids = set() for root in tree.roots: for l2 in root.children: for l3 in l2.children: all_l3_ids.add(l3.id) import pytest with pytest.raises(ValueError, match="锚节点不足"): sample_anchor(tree, "Object Recognition", all_l3_ids, random.Random(42)) ``` - [ ] **Step 2: 运行测试确认失败** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestSampleAnchor -v ``` 预期：ImportError（sample_anchor 不存在） - [ ] **Step 3: 实现 sample_anchor** 在 `app/question_gen/synthesizer.py` 中追加 `sample_anchor` 函数，核心逻辑： 1. 根据 `TASK_TYPE_LEVEL_MAP[task_type].level` 确定采样层级 2. L3 题型：从所有 L3 节点中随机选一个（排除 used_node_ids），取单帧 + card 3. L2 题型（Action Recognition / Action Reasoning / Counting Problem）：随机选一个 L2 节点，均匀采样 2-3 子帧，card 取 event_description 4. **Temporal Perception 特例**：随机选一个 L2 节点，取 0-1 帧（有子帧取 1 帧，无则 0），card_text 必须包含 event_description + time_range 5. L1 题型：取根节点 card（scene_summary）。**Information Synopsis 使用全部 L2 card，Temporal Reasoning 选 ≥3 个**。L2 子节点按 time_range 升序排列，每个 L2 取 1 张代表帧 6. L1-L2 题型：随机选 2-3 个 L2 节点（按 time_range 排序），每个取 1 张代表帧 7. distractor_texts：收集同树中**其他**同层级节点的摘要文本 8. 候选不足时 `raise ValueError("锚节点不足: ...")` 详细实现需参考 `app/tree/index.py` 中 L1Node/L2Node/L3Node 的字段结构（L3Card.frame_summary, L2Card.event_description, L1Card.scene_summary）和 frame_path 位置。 - [ ] **Step 4: 运行测试确认通过** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestSampleAnchor -v ``` - [ ] **Step 5: 提交** ```bash git add app/question_gen/synthesizer.py tests/unit/test_synthesizer.py git commit -m "feat(question_gen): sample_anchor — 按题型层级采样锚节点" ``` --- ### Task 3: synthesizer.py — build_generation_prompt + parse_vlm_response **Files:** - Modify: `app/question_gen/synthesizer.py` - Modify: `tests/unit/test_synthesizer.py` - [ ] **Step 1: 写失败测试** ```python from core.types import GeneratedQuestion from app.question_gen.synthesizer import ( build_generation_prompt, parse_vlm_response, AnchorContext, ) class TestBuildGenerationPrompt: def test_messages_structure(self): anchor = AnchorContext( node_id="L3_001", card_text="A person typing on a laptop", frame_paths=["store/videos/test/frames/L1_000_L2_000_L3_000.jpg"], subtitle="Hello world", distractor_texts=["Another person walking in park"], ) exemplars = [ GeneratedQuestion( question_id="ex-1", video_id="v1", task_type="Object Recognition", question="What object?", options=("A. Cat", "B. Dog", "C. Bird", "D. Fish"), answer="A", source_nodes=(), difficulty="medium", ), ] messages, image_paths = build_generation_prompt( "Object Recognition", anchor, exemplars, ) assert messages[0]["role"] == "system" assert "Object Recognition" in messages[0]["content"] assert any("What object?" in str(m) for m in messages) assert image_paths == anchor.frame_paths def test_distractor_in_user_message(self): anchor = AnchorContext( node_id="L2_003", card_text="Event card text", frame_paths=["a.jpg", "b.jpg"], subtitle="", distractor_texts=["Distractor node summary"], ) messages, _ = build_generation_prompt("Action Reasoning", anchor, []) user_msg = [m for m in messages if m["role"] == "user"][0] assert "Distractor node summary" in user_msg["content"] class TestParseVlmResponse: def test_valid_json(self): raw = '{"question": "What?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "A"}' result = parse_vlm_response(raw, "vid1", "Object Recognition", 1) assert result["question"] == "What?" assert result["answer"] == "A" assert len(result["options"]) == 4 def test_invalid_json_raises(self): import pytest with pytest.raises(ValueError, match="VLM 返回"): parse_vlm_response("not json", "vid1", "Object Recognition", 1) def test_missing_fields_raises(self): import pytest raw = '{"question": "What?"}' with pytest.raises(ValueError): parse_vlm_response(raw, "vid1", "Object Recognition", 1) def test_options_must_be_four(self): import pytest raw = '{"question": "Q?", "options": ["A. X", "B. Y"], "answer": "A"}' with pytest.raises(ValueError, match="4"): parse_vlm_response(raw, "vid1", "Object Recognition", 1) def test_answer_must_be_abcd(self): import pytest raw = '{"question": "Q?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "E"}' with pytest.raises(ValueError, match="A.*D"): parse_vlm_response(raw, "vid1", "Object Recognition", 1) ``` - [ ] **Step 2: 运行测试确认失败** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestBuildGenerationPrompt -v conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestParseVlmResponse -v ``` 预期：ImportError - [ ] **Step 3: 实现 build_generation_prompt + parse_vlm_response** `build_generation_prompt(task_type, anchor, exemplars) -> (messages, image_paths)`： - system message：角色设定 + 题型 + exemplar 示例 + 约束（基于节点内容、干扰项来自其他节点） - user message：锚节点 card_text + subtitle + distractor_texts - image_paths：直接取 anchor.frame_paths `parse_vlm_response(raw, video_id, task_type, seq) -> dict`： - 尝试 `json.loads(raw)`，失败时尝试从 markdown code block 提取 JSON - 校验必需字段 question / options / answer 存在 - 返回 `{"question_id": f"gen-{video_id}-{seq:03d}", "question": ..., "options": [...], "answer": ...}` - 缺字段或解析失败 → `raise ValueError("VLM 返回...")` - [ ] **Step 4: 运行测试确认通过** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v ``` - [ ] **Step 5: 提交** ```bash git add app/question_gen/synthesizer.py tests/unit/test_synthesizer.py git commit -m "feat(question_gen): build_generation_prompt + parse_vlm_response" ``` --- ### Task 4: synthesizer.py — is_duplicate + generate_one **Files:** - Modify: `app/question_gen/synthesizer.py` - Modify: `tests/unit/test_synthesizer.py` - [ ] **Step 1: 写失败测试** ```python import numpy as np from unittest.mock import AsyncMock, MagicMock from app.question_gen.synthesizer import is_duplicate, generate_one class TestIsDuplicate: @staticmethod def _fake_embed(texts): """确定性 + L2 归一化的 fake embedding。""" if isinstance(texts, str): texts = [texts] vecs = [] for t in texts: rs = np.random.RandomState(hash(t) % 2**31) v = rs.randn(4).astype(np.float32) v /= np.linalg.norm(v) vecs.append(v) return np.array(vecs, dtype=np.float32) def test_empty_pool_never_duplicate(self): pool = np.zeros((0, 4), dtype=np.float32) assert is_duplicate("anything", pool, self._fake_embed, 0.85) is False def test_identical_text_is_duplicate(self): text = "What is happening in the video?" emb = self._fake_embed(text) pool = emb.copy() assert is_duplicate(text, pool, self._fake_embed, 0.85) is True def test_different_text_not_duplicate(self): pool_texts = ["aaa", "bbb", "ccc", "ddd", "eee"] pool = self._fake_embed(pool_texts) assert is_duplicate("completely unique text xyz", pool, self._fake_embed, 0.99) is False class TestGenerateOne: @staticmethod async def test_success_path(): """mock VLM 返回合法 JSON，应成功生成。""" vlm = AsyncMock() vlm.chat_with_images.return_value = MagicMock( content='{"question":"Q?","options":["A. 1","B. 2","C. 3","D. 4"],"answer":"A"}' ) embed_fn = lambda t: np.zeros((1, 4) if isinstance(t, str) else (len(t), 4), dtype=np.float32) tree, vid = _load_test_tree() result = await generate_one( vlm=vlm, embed_fn=embed_fn, tree=tree, video_id=vid, task_type="Object Recognition", seq=1, exemplars=[], pool_embeddings=np.zeros((0, 4), dtype=np.float32), used_node_ids=set(), max_retries=3, similarity_threshold=0.85, rng=random.Random(42), session_id="test", ) assert result is not None assert result.question_id == f"gen-{vid}-001" assert result.task_type == "Object Recognition" @staticmethod async def test_all_retries_exhausted_returns_none(): """VLM 始终返回无效 JSON，耗尽重试后返回 None。""" vlm = AsyncMock() vlm.chat_with_images.return_value = MagicMock(content="invalid") embed_fn = lambda t: np.zeros((1, 4), dtype=np.float32) tree, vid = _load_test_tree() result = await generate_one( vlm=vlm, embed_fn=embed_fn, tree=tree, video_id=vid, task_type="Object Recognition", seq=1, exemplars=[], pool_embeddings=np.zeros((0, 4), dtype=np.float32), used_node_ids=set(), max_retries=2, similarity_threshold=0.85, rng=random.Random(42), session_id="test", ) assert result is None ``` - [ ] **Step 2: 运行测试确认失败** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestIsDuplicate -v conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestGenerateOne -v ``` - [ ] **Step 3: 实现 is_duplicate + generate_one** `is_duplicate(question_text, pool_embeddings, embed_fn, threshold) -> bool`： - `embed_fn(question_text)` → `[1, D]`，squeeze 为 `[D]` - `pool_embeddings @ query` 余弦相似度（pool 和 query 都已 L2 归一化） - `max(similarities) >= threshold` → True `generate_one(vlm, embed_fn, tree, video_id, task_type, seq, *, ...)` → `GeneratedQuestion | None`： - 循环最多 `max_retries` 次： 1. `sample_anchor(tree, task_type, used_node_ids, rng)` → anchor 2. `build_generation_prompt(task_type, anchor, exemplars)` → messages, images 3. `await vlm.chat_with_images(messages, images, session_id=session_id)` → response 4. `parse_vlm_response(response.content, video_id, task_type, seq)` → parsed_dict（含四选一 schema 校验） 5. 用 anchor.node_id 补齐 `source_nodes`，`difficulty="medium"` 6. 构造并返回 `GeneratedQuestion`（**不在此处做去重**——去重在调用方的单线程汇总点原子执行） - 全部重试失败（parse 异常）→ return None **并发去重安全**：`generate_one` 只负责生成候选题。调用方（tools/ CLI）在收到候选后，在单线程汇总点（async for + await）原子执行：① `is_duplicate` 检查当前题型的 embedding 池 → ② 通过则添加 embedding + 写 JSON + 更新 progress → ③ 不通过则丢弃并重试。embedding 池按 `dict[str, np.ndarray]`（key=task_type）维护，确保只在同题型内去重。 - [ ] **Step 4: 运行测试确认通过** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v ``` - [ ] **Step 5: 提交** ```bash git add app/question_gen/synthesizer.py tests/unit/test_synthesizer.py git commit -m "feat(question_gen): is_duplicate + generate_one — 去重与单题生成编排" ``` --- ### Task 5: factory.py — build_inference_deps **Files:** - Create: `app/harness/factory.py` - Create: `tests/unit/test_factory.py` - [ ] **Step 1: 写失败测试** ```python # tests/unit/test_factory.py import random from pathlib import Path from unittest.mock import MagicMock, AsyncMock import numpy as np import pytest from app.harness.factory import build_inference_deps, InferenceDeps class TestBuildInferenceDeps: def test_returns_inference_deps(self, tmp_path): """用 fake adapters 验证返回类型和字段非 None。""" # 准备一棵最小树 import json vid_dir = tmp_path / "videos" / "test_vid" vid_dir.mkdir(parents=True) frames_dir = vid_dir / "frames" frames_dir.mkdir() minimal_tree = { "metadata": {"source_path": "test", "modality": "video"}, "roots": [{ "id": "L1_000", "card": { "scene_summary": "s", "main_setting": "s", "key_entities": [], "main_actions": [], "topic_keywords": [], "visible_text": [], "temporal_flow": "s", }, "time_range": [0, 10], "children": [], }], } (vid_dir / "tree.json").write_text(json.dumps(minimal_tree)) # 准备 prompts prompts_dir = tmp_path / "prompts" prompts_dir.mkdir() (prompts_dir / "system.md").write_text("You are a search agent.") fake_llm = AsyncMock() fake_vlm = AsyncMock() fake_embed = MagicMock() fake_embed.dim = 4 fake_embed.embed = lambda t: np.zeros((1, 4), dtype=np.float32) deps = build_inference_deps( store_dir=tmp_path, video_id="test_vid", prompts_dir=prompts_dir, skills_dir=None, skill_mode="none", embed_provider=fake_embed, llm=fake_llm, vlm=fake_vlm, ocr=None, verify_vision=False, anchor=False, assemble_mode="ids", ) assert isinstance(deps, InferenceDeps) assert deps.llm is fake_llm assert callable(deps.tool_dispatch_fn) assert callable(deps.prompt_builder) # 验证 prompt_builder 真正可调用（wiring 正确） from core.types import GeneratedQuestion fake_q = GeneratedQuestion( question_id="q1", video_id="test_vid", task_type="Object Recognition", question="What?", options=("A. X", "B. Y", "C. Z", "D. W"), answer="A", source_nodes=(), difficulty="medium", ) system, user = deps.prompt_builder(fake_q) assert isinstance(system, str) and len(system) > 0 assert isinstance(user, str) and "What?" in user def test_missing_tree_raises(self, tmp_path): prompts_dir = tmp_path / "prompts" prompts_dir.mkdir() (prompts_dir / "system.md").write_text("x") vid_dir = tmp_path / "videos" / "nonexist" vid_dir.mkdir(parents=True) with pytest.raises(FileNotFoundError): build_inference_deps( store_dir=tmp_path, video_id="nonexist", prompts_dir=prompts_dir, skills_dir=None, skill_mode="none", embed_provider=MagicMock(), llm=AsyncMock(), vlm=AsyncMock(), ocr=None, verify_vision=False, anchor=False, assemble_mode="ids", ) ``` - [ ] **Step 2: 运行测试确认失败** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_factory.py -v ``` 预期：ImportError - [ ] **Step 3: 实现 factory.py** 创建 `app/harness/factory.py`： ```python """推理依赖组装 — 给定 store + config 构建可工作的推理依赖集。 factory 只做组装，不持有状态。adapter 实例由调用方创建并传入。 消费者：tools/generate_questions.py（校准）、未来 main.py、Runner。 """ from __future__ import annotations from dataclasses import dataclass from functools import partial from pathlib import Path from typing import TYPE_CHECKING, Any from app.search.prompt import PromptManager from app.search.skills import SkillRegistry, discover_skills from app.search.tools import SearchToolDispatcher from app.tree.environment import TreeEnvironment from app.tree.index import TreeIndex if TYPE_CHECKING: from collections.abc import Callable from app.ports import EmbeddingProvider, OCRProvider from core.protocols import LLMProvider, VLMProvider from core.types import GeneratedQuestion @dataclass(frozen=True) class InferenceDeps: """跑一次推理所需的全套依赖（不含 HarnessLog）。 属性: llm: LLM 端口实例。 tool_dispatch_fn: SearchToolDispatcher.dispatch 的绑定方法。 prompt_builder: (GeneratedQuestion) -> (system_prompt, user_prompt)。 """ llm: LLMProvider tool_dispatch_fn: Callable[..., Any] prompt_builder: Callable[[GeneratedQuestion], tuple[str, str]] def build_inference_deps( *, store_dir: Path, video_id: str, prompts_dir: Path, skills_dir: Path | None, skill_mode: str, embed_provider: EmbeddingProvider, llm: LLMProvider, vlm: VLMProvider, ocr: OCRProvider | None, verify_vision: bool, anchor: bool, assemble_mode: str, ) -> InferenceDeps: """组装单个视频的推理依赖。 参数: store_dir: store 根目录（含 videos/{video_id}/tree.json）。 video_id: 视频标识。 prompts_dir: prompt 文件目录（含 system.md）。 skills_dir: skill 文件目录（None 不启用）。 skill_mode: "auto" / "manual" / "none"。 embed_provider: 文本嵌入端口。 llm: LLM 端口。 vlm: VLM 端口。 ocr: OCR 端口（None 不启用）。 verify_vision: observe_frame 是否执行验证轮。 anchor: view_node 是否启用行号锚模式。 assemble_mode: 锚模式装配形态。 返回: InferenceDeps 实例。 异常: FileNotFoundError: tree.json 不存在。 """ # Phase 1: 加载树 tree_path = store_dir / "videos" / video_id / "tree.json" if not tree_path.exists(): raise FileNotFoundError(f"树文件不存在: {tree_path}") tree = TreeIndex.load_json(str(tree_path)) frames_dir = store_dir / "videos" / video_id / "frames" env = TreeEnvironment(tree, frames_dir if frames_dir.exists() else None) # Phase 2: Skills skills: SkillRegistry | None = None always_skills_text = "" task_skill_map: dict[str, str] = {} catalog_text = "" if skills_dir and skills_dir.exists(): always_skills_text, task_skill_map, catalog_text, skills = discover_skills(skills_dir) # Phase 3: SearchToolDispatcher dispatcher = SearchToolDispatcher( env=env, tool_llm=llm, vlm=vlm, ocr=ocr, prompts_dir=prompts_dir, skills=skills, embed_fn=embed_provider.embed, verify_vision=verify_vision, anchor=anchor, assemble_mode=assemble_mode, ) # Phase 4: PromptManager → prompt_builder 偏函数 pm = PromptManager(prompts_dir) l1_ids = [r.id for r in tree.roots] def _prompt_builder( qa: GeneratedQuestion, _pm: PromptManager = pm, _skill_mode: str = skill_mode, _always: str = always_skills_text, _tsm: dict = task_skill_map, _cat: str = catalog_text, _l1_ids: list = l1_ids, ) -> tuple[str, str]: system = _pm.build_inference_prompt( _skill_mode, qa.task_type, _always, _tsm, _cat, ) user = _pm.format_user_prompt( qa.question, list(qa.options), _l1_ids, qa.task_type, ) return system, user return InferenceDeps( llm=llm, tool_dispatch_fn=dispatcher.dispatch, prompt_builder=_prompt_builder, ) ``` - [ ] **Step 4: 运行测试确认通过** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_factory.py -v ``` - [ ] **Step 5: 提交** ```bash git add app/harness/factory.py tests/unit/test_factory.py git commit -m "feat(harness): factory.py — 推理依赖组装，可复用于 calibrate + main.py" ``` --- ### Task 6: tools/generate_questions.py — generate 子命令 **Files:** - Create: `tools/generate_questions.py` - Modify: `tests/unit/test_generate_questions.py`（新建） - [ ] **Step 1: 写失败测试** ```python # tests/unit/test_generate_questions.py import json import sys from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import numpy as np import pytest class TestGenerateSmoke: def test_generate_writes_json(self, tmp_path): """mock VLM + 1 棵真实树 + per_type=1，验证 JSON 输出格式。""" import shutil from unittest.mock import AsyncMock, MagicMock, patch # 复制一棵真实树到 tmp src = Path("store/videos") / sorted(Path("store/videos").iterdir())[0].name dst = tmp_path / "videos" / src.name shutil.copytree(src, dst) # 准备 benchmark（至少 1 道题做 exemplar） bench_dir = tmp_path / "questions" / "benchmarks" bench_dir.mkdir(parents=True) bench_file = bench_dir / f"{src.name}.json" bench_file.write_text(json.dumps([{ "question_id": "ex-1", "task_type": "Object Recognition", "question": "What?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "A", }])) output_dir = tmp_path / "output" output_dir.mkdir() # import CLI 模块的内部函数 sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from tools.generate_questions import _load_or_init_progress, _select_exemplars # 验证 progress 初始化 progress = _load_or_init_progress(output_dir) assert progress["completed"] == {} # 验证 exemplar 选择 from app.question_gen.loader import load_benchmark bench_qs = load_benchmark(bench_dir) exemplars = _select_exemplars(bench_qs, "Object Recognition", 3, random.Random(42)) assert len(exemplars) >= 1 assert all(e.task_type == "Object Recognition" for e in exemplars) class TestProgressResume: def test_skips_completed_and_rebuilds_pool(self, tmp_path): """progress.json 中已完成的题应被跳过，embedding 池应从已有 JSON 恢复。""" output_dir = tmp_path / "output" output_dir.mkdir() # 写一个已完成的 JSON (output_dir / "test_vid.json").write_text(json.dumps([{ "question_id": "gen-test_vid-001", "task_type": "Object Recognition", "question": "Existing question?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "A", "source_nodes": ["L3_001"], "difficulty": "medium", }])) progress = { "completed": {"Object Recognition": ["gen-test_vid-001"]}, "output_dir": str(output_dir), } (output_dir / "progress.json").write_text(json.dumps(progress)) from tools.generate_questions import _load_or_init_progress loaded = _load_or_init_progress(output_dir) assert "gen-test_vid-001" in loaded["completed"]["Object Recognition"] ``` - [ ] **Step 2: 运行测试确认失败** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py -v ``` - [ ] **Step 3: 实现 tools/generate_questions.py — generate 子命令** 创建 `tools/generate_questions.py`，核心结构： ```python #!/usr/bin/env python3 """赛题生成工具：generate + calibrate。 用法: conda activate Video-Tree-TRM python tools/generate_questions.py generate --store-dir store ... python tools/generate_questions.py calibrate --generated-dir ... --benchmark-dir ... app/core/adapters 不 import 此脚本。 """ from __future__ import annotations import argparse import asyncio import json import sys from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROJECT_ROOT)) from dotenv import load_dotenv from loguru import logger load_dotenv(PROJECT_ROOT / ".env") # generate 子命令： # 1. 加载 video_id 列表 + benchmark exemplars + 初始化 embedding 池 # 2. 断点续跑：读 progress.json + 恢复已生成题 embedding + used_node_ids # 3. 实例化 GovernedVLMClient + EmbeddingProvider（从 .env 读配置） # 4. 对 12 题型 × per_type，asyncio.Semaphore 并发调 generate_one # 5. 单线程汇总：检查去重 → 加入 pool → 写 JSON → 更新 progress # 6. 全部完成删除 progress.json ``` 实现要点： - `_load_or_init_progress(output_dir)` / `_save_progress(output_dir, progress)` 断点续跑 - `_rebuild_embedding_pool(output_dir, embed_fn, benchmark_questions)` 续跑时恢复 embedding - `_build_vlm_client()` / `_build_embed_provider()` 从 .env 实例化 adapters - `_select_exemplars(benchmark, task_type, n, rng)` 跨视频采样 few-shot - `async def _run_generate(args)` 主流程 - 并发模型：Semaphore 限流 VLM 调用，但去重+写入在主协程中顺序执行 - [ ] **Step 4: 运行测试 + lint** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py -v conda activate Video-Tree-TRM & ruff check tools/generate_questions.py --fix ``` - [ ] **Step 5: 提交** ```bash git add tools/generate_questions.py tests/unit/test_generate_questions.py git commit -m "feat(tools): generate_questions.py generate 子命令 — VLM 出题 + 去重 + 断点续跑" ``` --- ### Task 7: tools/generate_questions.py — calibrate 子命令 **Files:** - Modify: `tools/generate_questions.py` - Modify: `tests/unit/test_generate_questions.py` - [ ] **Step 1: 写失败测试** ```python from scipy.stats import fisher_exact class TestCalibrateJudgment: def test_pass_when_delta_small(self): """差异小于 tolerance → PASS。""" from tools.generate_questions import _judge_task_type verdict = _judge_task_type( bench_correct=60, bench_total=100, gen_correct=12, gen_total=20, tolerance=0.10, alpha=0.05, ) assert verdict == "PASS" def test_fail_when_delta_large_and_significant(self): """差异大且统计显著 → FAIL。""" from tools.generate_questions import _judge_task_type verdict = _judge_task_type( bench_correct=144, bench_total=240, gen_correct=6, gen_total=20, tolerance=0.10, alpha=0.05, ) assert verdict == "FAIL" def test_warn_when_delta_large_but_not_significant(self): """差异大但样本不足（p > alpha）→ WARN。""" from tools.generate_questions import _judge_task_type verdict = _judge_task_type( bench_correct=2, bench_total=3, gen_correct=8, gen_total=20, tolerance=0.10, alpha=0.05, ) assert verdict == "WARN" class TestCalibrateIntegration: def test_baseline_params_must_be_paired(self): """--baseline-db 和 --baseline-run-id 必须成对出现。""" from tools.generate_questions import _validate_calibrate_args import pytest with pytest.raises(ValueError, match="成对"): _validate_calibrate_args(baseline_db="some.db", baseline_run_id=None) def test_has_fail_returns_exit_code_1(self): """存在 FAIL 判定时，_calibrate_exit_code 返回 1。""" from tools.generate_questions import _calibrate_exit_code verdicts = {"Object Recognition": "PASS", "Action Reasoning": "FAIL"} assert _calibrate_exit_code(verdicts) == 1 def test_all_pass_or_warn_returns_exit_code_0(self): from tools.generate_questions import _calibrate_exit_code verdicts = {"Object Recognition": "PASS", "Spatial Perception": "WARN"} assert _calibrate_exit_code(verdicts) == 0 ``` - [ ] **Step 2: 运行测试确认失败** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py::TestCalibrateJudgment -v ``` - [ ] **Step 3: 实现 calibrate 子命令** 在 `tools/generate_questions.py` 中追加： `_judge_task_type(bench_correct, bench_total, gen_correct, gen_total, tolerance, alpha) -> str`： - `delta = abs(gen_correct/gen_total - bench_correct/bench_total)` - `delta <= tolerance` → "PASS" - Fisher exact test p-value：`table = [[bench_correct, bench_total-bench_correct], [gen_correct, gen_total-gen_correct]]` - `p < alpha and delta > tolerance` → "FAIL" - else → "WARN" `async def _run_calibrate(args)` 主流程： 1. `load_benchmark` 加载两组题 2. benchmark 基线：有 `--baseline-db` 则从 DB 读，否则按 video_id 分组 → `build_inference_deps` → `run_inference` 3. 生成题同理按 video_id 分组 → 分组推理 4. 汇总 per_task_type accuracy → `_judge_task_type` 逐题型判定 5. 输出对比表 6. 有 FAIL → `sys.exit(1)` - [ ] **Step 4: 运行测试确认通过** ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py -v ``` - [ ] **Step 5: 提交** ```bash git add tools/generate_questions.py tests/unit/test_generate_questions.py git commit -m "feat(tools): generate_questions.py calibrate 子命令 — Fisher exact test 校准" ``` --- ### Task 8: __init__.py 更新 + lint + 全量测试 **Files:** - Modify: `app/question_gen/__init__.py` - [ ] **Step 1: 写失败测试** ```python # 在 tests/unit/test_synthesizer.py 中追加 def test_public_api_importable(): """synthesizer 的公共 API 必须可从 app.question_gen 直接 import。""" from app.question_gen import generate_one, AnchorContext, TASK_TYPE_LEVEL_MAP, sample_anchor assert callable(generate_one) assert callable(sample_anchor) ``` 运行确认失败（当前 __init__.py 不 export 这些）： ```bash conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::test_public_api_importable -v ``` - [ ] **Step 2: 更新 __init__.py re-export** ```python """出题模块 — benchmark 加载、分层采样与赛题合成。""" from app.question_gen.loader import load_benchmark, stratified_sample from app.question_gen.synthesizer import ( TASK_TYPE_LEVEL_MAP, AnchorContext, generate_one, sample_anchor, ) __all__ = [ "load_benchmark", "stratified_sample", "TASK_TYPE_LEVEL_MAP", "AnchorContext", "generate_one", "sample_anchor", ] ``` - [ ] **Step 2: 全量 lint** ```bash conda activate Video-Tree-TRM & ruff check app/question_gen/ app/harness/factory.py tools/generate_questions.py --fix conda activate Video-Tree-TRM & ruff format app/question_gen/ app/harness/factory.py tools/generate_questions.py ``` - [ ] **Step 3: 全量测试** ```bash conda activate Video-Tree-TRM & pytest tests/ -v --tb=short ``` 预期：全部 PASS - [ ] **Step 4: 提交** ```bash git add app/question_gen/__init__.py git commit -m "refactor(question_gen): __init__.py 追加 synthesizer re-export" ``` --- ## Self-Review 核对 **范围说明**：设计 §4.4 要求 `Runner._make_tool_dispatch_fn` / `_make_prompt_builder` 委托 factory，本计划不包含该改造——Runner 改造随 `main.py` 一起实施更合理。factory.py 已就绪可复用。 | 设计文档章节 | 对应 Task | |-------------|-----------| | §2 模块结构 | Task 1-5 (synthesizer) + Task 5 (factory) + Task 6-7 (CLI) | | §3.1 题型映射 | Task 1 | | §3.2 AnchorContext | Task 1 | | §3.3 函数签名 | Task 2 (sample_anchor) + Task 3 (prompt/parse) + Task 4 (dedup/generate) | | §3.4 exemplar 选择 | Task 6 (_select_exemplars) | | §3.6 去重 + 并发安全 | Task 4 (is_duplicate) + Task 6 (单线程汇总) | | §4 factory.py | Task 5 | | §5 CLI 设计 | Task 6 (generate) + Task 7 (calibrate) | | §6 Fisher 校准 | Task 7 (_judge_task_type) | | §7 断点续跑 | Task 6 (progress + embedding 恢复) | | §8 输出格式 | Task 6 (JSON 写入) | | §9 前置修复 | Task 0 | | §10 测试策略 | Task 1-7 各含测试 |