Video-Tree-TRM5/research-wiki/plans/2026-07-09-question-gen-synth.md

赛题生成工具实现计划

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 中追加：

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: 运行测试确认失败

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: 运行测试确认通过

conda activate Video-Tree-TRM & pytest tests/unit/test_harness_config.py -v

预期：全部 PASS

• Step 5: 提交

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: 写失败测试 — 题型映射完整性

# 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: 运行测试确认失败

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：

"""赛题合成核心逻辑 — 节点采样、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: 运行测试确认通过

conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v

预期：PASS

• Step 5: 提交

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: 写失败测试

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: 运行测试确认失败

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: 运行测试确认通过

conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestSampleAnchor -v

• Step 5: 提交

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: 写失败测试

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: 运行测试确认失败

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: 运行测试确认通过

conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v

• Step 5: 提交

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: 写失败测试

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: 运行测试确认失败

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: 运行测试确认通过

conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v

• Step 5: 提交

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: 写失败测试

# 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: 运行测试确认失败

conda activate Video-Tree-TRM & pytest tests/unit/test_factory.py -v

预期：ImportError

• Step 3: 实现 factory.py

创建 app/harness/factory.py：

"""推理依赖组装 — 给定 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: 运行测试确认通过

conda activate Video-Tree-TRM & pytest tests/unit/test_factory.py -v

• Step 5: 提交

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: 写失败测试

# 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: 运行测试确认失败

conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py -v

• Step 3: 实现 tools/generate_questions.py — generate 子命令

创建 tools/generate_questions.py，核心结构：

#!/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

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: 提交

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: 写失败测试

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: 运行测试确认失败

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: 运行测试确认通过

conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py -v

• Step 5: 提交

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: 写失败测试

# 在 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 这些）：

conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::test_public_api_importable -v

• Step 2: 更新 __init__.py re-export

"""出题模块 — 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

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: 全量测试

conda activate Video-Tree-TRM & pytest tests/ -v --tb=short

预期：全部 PASS

• Step 4: 提交

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 各含测试