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

question_gen 模块实现计划

For agentic workers: REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 从 TRM4 迁移出题数据结构与采样逻辑到 TRM5 Clean Architecture，预留 LLM 出题 Protocol。

Architecture: GeneratedQuestion 放 core/types.py（跨层共享），加载和采样逻辑放 app/question_gen/loader.py，QuestionGenerator Protocol 追加到 app/ports.py。

Tech Stack: Python 3.11, dataclasses, pytest, loguru

设计文档: research-wiki/designs/2026-07-07-question-gen-design.md

核心算法保真: 本计划不涉及 ARCHITECTURE.md §6 中 13 项核心算法的迁移。stratified_sample 是采样工具函数，不在保真清单内，但仍逐行比对 TRM4 实现保证行为一致。

---

Task 1: GeneratedQuestion 数据类型

Files:

• Modify: core/types.py

• Modify: tests/unit/test_core_types.py

• Step 1: 在 test_core_types.py 追加 GeneratedQuestion 测试

from core.types import GeneratedQuestion


class TestGeneratedQuestion:
    @pytest.fixture()
    def sample_question(self) -> GeneratedQuestion:
        return GeneratedQuestion(
            question_id="719-1",
            video_id="B7Hh0PY1kks",
            task_type="Action Reasoning",
            question="What are the differing motivations?",
            options=("A. Option 1", "B. Option 2", "C. Option 3", "D. Option 4"),
            answer="B",
            source_nodes=(),
            difficulty="medium",
        )

    def test_frozen_prevents_mutation(self, sample_question: GeneratedQuestion) -> None:
        with pytest.raises(AttributeError):
            sample_question.question = "篡改"

    def test_all_fields_accessible(self, sample_question: GeneratedQuestion) -> None:
        assert sample_question.question_id == "719-1"
        assert sample_question.video_id == "B7Hh0PY1kks"
        assert sample_question.task_type == "Action Reasoning"
        assert sample_question.question == "What are the differing motivations?"
        assert sample_question.options == ("A. Option 1", "B. Option 2", "C. Option 3", "D. Option 4")
        assert sample_question.answer == "B"
        assert sample_question.source_nodes == ()
        assert sample_question.difficulty == "medium"

    def test_options_is_tuple(self, sample_question: GeneratedQuestion) -> None:
        assert isinstance(sample_question.options, tuple)

    def test_source_nodes_is_tuple(self, sample_question: GeneratedQuestion) -> None:
        assert isinstance(sample_question.source_nodes, tuple)

• Step 2: 运行测试确认失败

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_core_types.py::TestGeneratedQuestion -v Expected: FAIL — ImportError: cannot import name 'GeneratedQuestion'

• Step 3: 在 core/types.py 追加 GeneratedQuestion

在 LLMResponse 类之后追加：

@dataclass(frozen=True)
class GeneratedQuestion:
    """单条生成/加载的题目。

    跨层共享类型，被 core/evolution/ 和 app/harness/、app/question_gen/ 使用。
    frozen=True 确保题目不可变。

    属性:
        question_id: 题目唯一标识。
        video_id: 所属视频标识。
        task_type: 题型（如 "Action Reasoning"）。
        question: 题目文本。
        options: 选项元组（如 ("A. ...", "B. ...", "C. ...", "D. ...")）。
        answer: 正确答案字母（如 "B"）。
        source_nodes: 来源节点 ID 元组。
        difficulty: 难度等级。
    """

    question_id: str
    video_id: str
    task_type: str
    question: str
    options: tuple[str, ...]
    answer: str
    source_nodes: tuple[str, ...]
    difficulty: str

• Step 4: 运行测试确认通过

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_core_types.py -v Expected: 全部 PASS（含原有 LLMResponse 测试 + 新增 GeneratedQuestion 测试）

• Step 5: 提交

feat(core): 追加 GeneratedQuestion frozen dataclass

---

Task 2: load_benchmark 加载函数

Files:

• Create: app/question_gen/loader.py

• Create: tests/unit/test_question_loader.py

• Step 1: 编写 load_benchmark 测试

在 tests/unit/test_question_loader.py 中创建：

"""app/question_gen/loader.py 单元测试。"""
from __future__ import annotations

import json
from pathlib import Path

import pytest

from app.question_gen.loader import load_benchmark
from core.types import GeneratedQuestion


@pytest.fixture()
def benchmark_dir(tmp_path: Path) -> Path:
    """创建包含 benchmark JSON 的临时目录。"""
    data = [
        {
            "question_id": "1-1",
            "task_type": "Action Reasoning",
            "question": "What happened?",
            "options": ["A. X", "B. Y", "C. Z", "D. W"],
            "answer": "A",
        },
        {
            "question_id": "1-2",
            "task_type": "OCR Problems",
            "question": "What text is shown?",
            "options": ["A. Hello", "B. World", "C. Foo", "D. Bar"],
            "answer": "B",
        },
    ]
    (tmp_path / "video_abc.json").write_text(json.dumps(data), encoding="utf-8")
    return tmp_path


class TestLoadBenchmark:
    def test_loads_questions_from_json(self, benchmark_dir: Path) -> None:
        questions = load_benchmark(benchmark_dir)
        assert len(questions) == 2

    def test_video_id_from_filename(self, benchmark_dir: Path) -> None:
        questions = load_benchmark(benchmark_dir)
        assert all(q.video_id == "video_abc" for q in questions)

    def test_fields_mapped_correctly(self, benchmark_dir: Path) -> None:
        questions = load_benchmark(benchmark_dir)
        q = questions[0]
        assert q.question_id == "1-1"
        assert q.task_type == "Action Reasoning"
        assert q.question == "What happened?"
        assert q.options == ("A. X", "B. Y", "C. Z", "D. W")
        assert q.answer == "A"

    def test_options_is_tuple(self, benchmark_dir: Path) -> None:
        questions = load_benchmark(benchmark_dir)
        assert isinstance(questions[0].options, tuple)

    def test_source_nodes_is_empty_tuple(self, benchmark_dir: Path) -> None:
        questions = load_benchmark(benchmark_dir)
        assert questions[0].source_nodes == ()

    def test_difficulty_defaults_to_medium_for_legacy(self, benchmark_dir: Path) -> None:
        questions = load_benchmark(benchmark_dir)
        assert questions[0].difficulty == "medium"

    def test_difficulty_from_json_when_present(self, tmp_path: Path) -> None:
        data = [
            {
                "question_id": "2-1",
                "task_type": "OCR Problems",
                "question": "Q?",
                "options": ["A. 1", "B. 2", "C. 3", "D. 4"],
                "answer": "C",
                "difficulty": "hard",
            }
        ]
        (tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8")
        questions = load_benchmark(tmp_path)
        assert questions[0].difficulty == "hard"

    def test_empty_directory_returns_empty_list(self, tmp_path: Path) -> None:
        questions = load_benchmark(tmp_path)
        assert questions == []

    def test_sorted_by_filename(self, tmp_path: Path) -> None:
        for name in ["z_video.json", "a_video.json"]:
            data = [{"question_id": f"{name}-1", "task_type": "T", "question": "Q?", "options": ["A", "B", "C", "D"], "answer": "A"}]
            (tmp_path / name).write_text(json.dumps(data), encoding="utf-8")
        questions = load_benchmark(tmp_path)
        assert questions[0].video_id == "a_video"
        assert questions[1].video_id == "z_video"

    def test_returns_generated_question_instances(self, benchmark_dir: Path) -> None:
        questions = load_benchmark(benchmark_dir)
        assert all(isinstance(q, GeneratedQuestion) for q in questions)

    def test_loads_real_benchmark(self) -> None:
        """使用真实 benchmark 数据验证加载正确性。"""
        real_dir = Path("store/questions/benchmarks/Video-MME")
        if not real_dir.exists():
            pytest.skip("真实 benchmark 数据不存在")
        questions = load_benchmark(real_dir)
        assert len(questions) > 0
        for q in questions:
            assert isinstance(q, GeneratedQuestion)
            assert len(q.options) == 4
            assert q.answer in ("A", "B", "C", "D")

    def test_malformed_json_raises(self, tmp_path: Path) -> None:
        """非法 JSON 文件应抛出 json.JSONDecodeError。"""
        (tmp_path / "bad.json").write_text("not valid json{{{", encoding="utf-8")
        with pytest.raises(json.JSONDecodeError):
            load_benchmark(tmp_path)

    def test_missing_required_field_raises(self, tmp_path: Path) -> None:
        """缺少必需字段（如 question_id）应抛出 KeyError。"""
        data = [{"task_type": "T", "question": "Q?", "options": ["A"], "answer": "A"}]
        (tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8")
        with pytest.raises(KeyError):
            load_benchmark(tmp_path)

• Step 2: 运行测试确认失败

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_question_loader.py::TestLoadBenchmark -v Expected: FAIL — ModuleNotFoundError: No module named 'app.question_gen.loader'

• Step 3: 实现 loader.py 的 load_benchmark

创建 app/question_gen/loader.py：

"""题目加载与分层采样。

从 benchmark JSON 目录加载题目，提供按对错比例的分层采样。
对应训练循环中的 DataLoader 角色。
"""

from __future__ import annotations

import json
from pathlib import Path

from core.types import GeneratedQuestion

_LEGACY_DEFAULT_DIFFICULTY = "medium"


def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]:
    """从 benchmark JSON 目录加载题目列表。

    每个 JSON 文件以文件名（不含扩展名）作为 video_id，
    文件内容为题目数组。

    参数:
        questions_dir: 包含 *.json 文件的目录路径。

    返回:
        按文件名排序加载的题目列表。
    """
    results: list[GeneratedQuestion] = []
    for path in sorted(questions_dir.glob("*.json")):
        video_id = path.stem
        with open(path, encoding="utf-8") as f:
            qa_list: list[dict] = json.load(f)
        for qa in qa_list:
            results.append(
                GeneratedQuestion(
                    question_id=qa["question_id"],
                    video_id=video_id,
                    task_type=qa["task_type"],
                    question=qa["question"],
                    options=tuple(qa["options"]),
                    answer=qa["answer"],
                    source_nodes=tuple(qa.get("source_nodes", ())),
                    difficulty=qa.get("difficulty", _LEGACY_DEFAULT_DIFFICULTY),
                )
            )
    return results

• Step 4: 运行测试确认通过

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_question_loader.py::TestLoadBenchmark -v Expected: 全部 PASS

• Step 5: 提交

feat(question_gen): load_benchmark — benchmark JSON 加载

---

Task 3: stratified_sample 分层采样

Files:

• Modify: app/question_gen/loader.py

• Modify: tests/unit/test_question_loader.py

• Step 1: 编写 stratified_sample 测试

在 tests/unit/test_question_loader.py 追加：

from app.question_gen.loader import stratified_sample


def _make_questions(n: int, task_type: str = "T") -> list[GeneratedQuestion]:
    """辅助函数：批量构造题目。"""
    return [
        GeneratedQuestion(
            question_id=f"{task_type}-{i}",
            video_id="v1",
            task_type=task_type,
            question=f"Q{i}?",
            options=("A", "B", "C", "D"),
            answer="A",
            source_nodes=(),
            difficulty="medium",
        )
        for i in range(n)
    ]


class TestStratifiedSample:
    def test_natural_distribution(self) -> None:
        """correct_ratio=None 时走自然分布随机抽样。"""
        questions = _make_questions(20)
        result = stratified_sample(
            questions=questions,
            correctness={},
            size=10,
            correct_ratio=None,
            task_types=None,
            seed=42,
            min_per_class=None,
        )
        assert len(result) == 10

    def test_natural_distribution_pool_insufficient(self) -> None:
        """自然分布时池不足应 ValueError。"""
        questions = _make_questions(5)
        with pytest.raises(ValueError, match="自然分布采样不足"):
            stratified_sample(
                questions=questions,
                correctness={},
                size=10,
                correct_ratio=None,
                task_types=None,
                seed=42,
                min_per_class=None,
            )

    def test_ratio_stratified(self) -> None:
        """按对错比例分层采样。"""
        questions = _make_questions(20)
        correctness = {f"T-{i}": i < 10 for i in range(20)}
        result = stratified_sample(
            questions=questions,
            correctness=correctness,
            size=10,
            correct_ratio=0.6,
            task_types=None,
            seed=42,
            min_per_class=None,
        )
        assert len(result) == 10
        correct_count = sum(1 for q in result if correctness.get(q.question_id, False))
        assert correct_count == 6

    def test_ratio_stratified_correct_first(self) -> None:
        """分层采样返回顺序：对题在前、错题在后。"""
        questions = _make_questions(20)
        correctness = {f"T-{i}": i < 10 for i in range(20)}
        result = stratified_sample(
            questions=questions,
            correctness=correctness,
            size=10,
            correct_ratio=0.5,
            task_types=None,
            seed=42,
            min_per_class=None,
        )
        n_correct = round(10 * 0.5)
        for q in result[:n_correct]:
            assert correctness.get(q.question_id, False) is True
        for q in result[n_correct:]:
            assert correctness.get(q.question_id, False) is False

    def test_ratio_stratified_pool_insufficient(self) -> None:
        """分层时对题或错题不足应 ValueError。"""
        questions = _make_questions(10)
        correctness = {f"T-{i}": True for i in range(10)}
        with pytest.raises(ValueError, match="分层不足"):
            stratified_sample(
                questions=questions,
                correctness=correctness,
                size=10,
                correct_ratio=0.5,
                task_types=None,
                seed=42,
                min_per_class=None,
            )

    def test_task_types_filter(self) -> None:
        """task_types 过滤只保留指定题型。"""
        q_a = _make_questions(10, task_type="TypeA")
        q_b = _make_questions(10, task_type="TypeB")
        result = stratified_sample(
            questions=q_a + q_b,
            correctness={},
            size=5,
            correct_ratio=None,
            task_types=["TypeA"],
            seed=42,
            min_per_class=None,
        )
        assert all(q.task_type == "TypeA" for q in result)

    def test_unknown_correctness_treated_as_wrong(self) -> None:
        """correctness 中不存在的 question_id 被当作错题。"""
        questions = _make_questions(20)
        correctness = {f"T-{i}": True for i in range(10)}
        result = stratified_sample(
            questions=questions,
            correctness=correctness,
            size=10,
            correct_ratio=0.5,
            task_types=None,
            seed=42,
            min_per_class=None,
        )
        n_correct = round(10 * 0.5)
        for q in result[:n_correct]:
            assert q.question_id in correctness

    def test_seed_reproducibility(self) -> None:
        """相同种子产生相同结果。"""
        questions = _make_questions(20)
        r1 = stratified_sample(questions=questions, correctness={}, size=10, correct_ratio=None, task_types=None, seed=123, min_per_class=None)
        r2 = stratified_sample(questions=questions, correctness={}, size=10, correct_ratio=None, task_types=None, seed=123, min_per_class=None)
        assert [q.question_id for q in r1] == [q.question_id for q in r2]

    def test_different_seeds_differ(self) -> None:
        """不同种子产生不同结果（概率性，但 20 选 10 几乎必然不同）。"""
        questions = _make_questions(20)
        r1 = stratified_sample(questions=questions, correctness={}, size=10, correct_ratio=None, task_types=None, seed=1, min_per_class=None)
        r2 = stratified_sample(questions=questions, correctness={}, size=10, correct_ratio=None, task_types=None, seed=2, min_per_class=None)
        assert [q.question_id for q in r1] != [q.question_id for q in r2]

    def test_min_per_class_backfill(self) -> None:
        """min_per_class 补足稀疏题型。"""
        q_a = _make_questions(10, task_type="TypeA")
        q_b = _make_questions(10, task_type="TypeB")
        all_q = q_a + q_b
        correctness = {q.question_id: True for q in q_a[:5]}
        result = stratified_sample(
            questions=all_q,
            correctness=correctness,
            size=3,
            correct_ratio=None,
            task_types=None,
            seed=42,
            min_per_class=2,
        )
        type_counts: dict[str, int] = {}
        for q in result:
            type_counts[q.task_type] = type_counts.get(q.task_type, 0) + 1
        assert type_counts.get("TypeA", 0) >= 2
        assert type_counts.get("TypeB", 0) >= 2

    def test_min_per_class_partial_backfill(self) -> None:
        """题型可补题数不足缺口时全取，不报错。"""
        q_sparse = _make_questions(1, task_type="Sparse")
        q_main = _make_questions(10, task_type="Main")
        result = stratified_sample(
            questions=q_sparse + q_main,
            correctness={},
            size=5,
            correct_ratio=None,
            task_types=None,
            seed=42,
            min_per_class=3,
        )
        sparse_in_result = [q for q in result if q.task_type == "Sparse"]
        assert len(sparse_in_result) == 1

    def test_min_per_class_no_duplicates(self) -> None:
        """补足后不产生重复 question_id。"""
        q_a = _make_questions(5, task_type="TypeA")
        q_b = _make_questions(5, task_type="TypeB")
        result = stratified_sample(
            questions=q_a + q_b,
            correctness={},
            size=3,
            correct_ratio=None,
            task_types=None,
            seed=42,
            min_per_class=2,
        )
        ids = [q.question_id for q in result]
        assert len(ids) == len(set(ids))

    def test_backfill_enumerates_all_pool_types(self) -> None:
        """补足遍历 pool 全部题型，包括主采样未命中的。"""
        q_main = _make_questions(10, task_type="Main")
        q_rare = _make_questions(3, task_type="Rare")
        result = stratified_sample(
            questions=q_main + q_rare,
            correctness={},
            size=2,
            correct_ratio=None,
            task_types=None,
            seed=0,
            min_per_class=1,
        )
        types_in_result = {q.task_type for q in result}
        assert "Rare" in types_in_result

• Step 2: 运行测试确认失败

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_question_loader.py::TestStratifiedSample -v Expected: FAIL — ImportError: cannot import name 'stratified_sample'

• Step 3: 实现 stratified_sample 及内部辅助函数

在 app/question_gen/loader.py 的导入区追加 import random（标准库，放在 import json 之后），然后在文件末尾追加以下函数：

def stratified_sample(
    questions: list[GeneratedQuestion],
    correctness: dict[str, bool],
    size: int,
    correct_ratio: float | None,
    task_types: list[str] | None,
    seed: int,
    min_per_class: int | None,
) -> list[GeneratedQuestion]:
    """按题型过滤后采样 size 道题，可选按对错比例分层并按题型保底。

    参数:
        questions: 候选题目全集。
        correctness: question_id → 基线是否答对。
        size: 采样总量。
        correct_ratio: 采样中"基线答对"题的占比；None 表示自然分布。
        task_types: 限定题型；None 表示不限。
        seed: 随机种子，保证可复现。
        min_per_class: 每个题型补足到的下限；None 表示不补足。

    返回:
        采样后的题目列表。

    异常:
        ValueError: 自然分布时池不足 size，或分层时某层题目不足。
    """
    rng = random.Random(seed)
    pool = [q for q in questions if task_types is None or q.task_type in task_types]

    if correct_ratio is None:
        if len(pool) < size:
            raise ValueError(f"自然分布采样不足: 需 {size} 道, 实有 {len(pool)} 道")
        sampled = rng.sample(pool, size)
    else:
        sampled = _ratio_stratified_sample(pool, correctness, size, correct_ratio, rng)

    if min_per_class is not None:
        sampled = _backfill_per_class(sampled, pool, min_per_class, rng)
    return sampled


def _ratio_stratified_sample(
    pool: list[GeneratedQuestion],
    correctness: dict[str, bool],
    size: int,
    correct_ratio: float,
    rng: random.Random,
) -> list[GeneratedQuestion]:
    """按对错比例分层采样：对题占 correct_ratio，其余为错题。

    参数:
        pool: 题型过滤后的候选题。
        correctness: question_id → 基线是否答对。
        size: 采样总量。
        correct_ratio: 对题占比。
        rng: 随机数发生器。

    返回:
        采样后的题目列表（对题在前、错题在后）。

    异常:
        ValueError: 对题或错题层不足。
    """
    correct = [q for q in pool if correctness.get(q.question_id, False)]
    wrong = [q for q in pool if not correctness.get(q.question_id, False)]
    n_correct = round(size * correct_ratio)
    n_wrong = size - n_correct
    if len(correct) < n_correct or len(wrong) < n_wrong:
        raise ValueError(
            f"分层不足: 需对{n_correct}/错{n_wrong}, "
            f"实有对{len(correct)}/错{len(wrong)}"
        )
    return rng.sample(correct, n_correct) + rng.sample(wrong, n_wrong)


def _backfill_per_class(
    sampled: list[GeneratedQuestion],
    pool: list[GeneratedQuestion],
    min_per_class: int,
    rng: random.Random,
) -> list[GeneratedQuestion]:
    """对候选池中出现的每个题型，将采样结果补足到 min_per_class 道。

    遍历对象是候选池 pool 里出现的全部题型（非仅 sampled 命中的），
    保证任意稀疏题型都能拿到足额样本。

    参数:
        sampled: 主采样结果（不修改，返回新列表）。
        pool: 候选题全集（补足来源 + 题型枚举来源）。
        min_per_class: 每个题型的下限。
        rng: 随机数发生器。

    返回:
        补足后的题目列表。
    """
    selected_ids = {q.question_id for q in sampled}
    result = list(sampled)
    counts: dict[str, int] = {}
    for q in sampled:
        counts[q.task_type] = counts.get(q.task_type, 0) + 1
    ordered_task_types: dict[str, None] = {}
    for q in pool:
        ordered_task_types.setdefault(q.task_type, None)
    for task_type in ordered_task_types:
        deficit = min_per_class - counts.get(task_type, 0)
        if deficit <= 0:
            continue
        candidates = [
            q
            for q in pool
            if q.task_type == task_type and q.question_id not in selected_ids
        ]
        take = rng.sample(candidates, min(deficit, len(candidates)))
        for q in take:
            selected_ids.add(q.question_id)
            result.append(q)
    return result

• Step 4: 运行测试确认通过

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_question_loader.py -v Expected: 全部 PASS（TestLoadBenchmark + TestStratifiedSample）

• Step 5: 提交

feat(question_gen): stratified_sample — 分层采样 + 题型保底

---

Task 4: QuestionGenerator Protocol 与模块公开 API

Files:

• Modify: app/ports.py

• Modify: app/question_gen/__init__.py

• Create: tests/unit/test_question_gen_api.py

• Step 1: 编写 Protocol 可导入性和 init 公开 API 测试

创建 tests/unit/test_question_gen_api.py：

"""app/ports.py QuestionGenerator Protocol 与 app/question_gen 公开 API 测试。"""
from __future__ import annotations

import importlib
from typing import runtime_checkable

from app.ports import QuestionGenerator
from core.types import GeneratedQuestion


class TestQuestionGeneratorProtocol:
    def test_importable(self) -> None:
        """QuestionGenerator 可从 app.ports 导入。"""
        assert QuestionGenerator is not None

    def test_is_runtime_checkable(self) -> None:
        """QuestionGenerator 是 runtime_checkable Protocol。"""
        assert hasattr(QuestionGenerator, "__protocol_attrs__") or hasattr(
            QuestionGenerator, "__abstractmethods__"
        )

    def test_generate_method_exists(self) -> None:
        """Protocol 定义了 generate 方法。"""
        assert hasattr(QuestionGenerator, "generate")


class TestQuestionGenPublicAPI:
    def test_load_benchmark_importable_from_package(self) -> None:
        """load_benchmark 可从 app.question_gen 直接导入。"""
        mod = importlib.import_module("app.question_gen")
        assert hasattr(mod, "load_benchmark")

    def test_stratified_sample_importable_from_package(self) -> None:
        """stratified_sample 可从 app.question_gen 直接导入。"""
        mod = importlib.import_module("app.question_gen")
        assert hasattr(mod, "stratified_sample")

    def test_all_exports(self) -> None:
        """__all__ 包含预期的公开 API。"""
        mod = importlib.import_module("app.question_gen")
        assert set(mod.__all__) == {"load_benchmark", "stratified_sample"}

• Step 2: 运行测试确认失败

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_question_gen_api.py -v Expected: FAIL — ImportError: cannot import name 'QuestionGenerator' from 'app.ports'

• Step 3: 在 app/ports.py 追加 QuestionGenerator Protocol

将已有的 if TYPE_CHECKING: 块扩展，追加 TreeIndex 和 GeneratedQuestion 导入，然后在 EmbeddingProvider 之后追加：

@runtime_checkable
class QuestionGenerator(Protocol):
    """LLM 驱动的题目生成端口（预留接口）。

    参数:
        video_id: 视频标识。
        task_type: 题型。
        tree: 视频树索引，提供锚节点上下文。
        exemplars: 风格示例题目列表。

    返回:
        生成的单条题目。
    """

    async def generate(
        self,
        video_id: str,
        task_type: str,
        tree: TreeIndex,
        *,
        exemplars: list[GeneratedQuestion],
    ) -> GeneratedQuestion: ...

合并后的 TYPE_CHECKING 块：

if TYPE_CHECKING:
    import numpy as np

    from app.tree.index import TreeIndex
    from core.types import GeneratedQuestion

• Step 4: 更新 app/question_gen/__init__.py 公开 API

"""出题模块 — benchmark 加载与分层采样。"""

from app.question_gen.loader import load_benchmark, stratified_sample

__all__ = ["load_benchmark", "stratified_sample"]

• Step 5: 运行测试确认通过

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_question_gen_api.py -v Expected: 全部 PASS

• Step 6: 运行全量测试确认无回归

Run: conda activate Video-Tree-TRM & pytest tests/ -v Expected: 全部 PASS

• Step 7: 提交

feat(question_gen): QuestionGenerator Protocol + 模块公开 API

---

Task 5: 文档同步与 lint

Files:

• Modify: research-wiki/ARCHITECTURE.md

• Modify: CLAUDE.md

• Step 1: 更新 ARCHITECTURE.md

需要修改 4 处：

1. §1 表格（第 17 行附近）：

| DataLoader | 出题 question_gen | `app/question_gen/generator.py` |

→

| DataLoader | 出题 question_gen | `app/question_gen/loader.py` |

2. §2.2 Mermaid（第 83 行附近）：

CLI --> QGEN["app/question_gen/generator.py\n新题构建"]

→

CLI --> QGEN["app/question_gen/loader.py\n新题构建"]

3. §2.3 目录树（第 132 行附近）：

│   │   ├── question_gen.py  # 数据加载、三池切分

此行描述 harness/ 内部的数据加载，但在 TRM5 中数据加载已移至 question_gen/loader.py。删除此行（harness/ 的三池切分模块在未来开发 harness 时再规划）。

4. §2.3 目录树（第 138-141 行）：

│   ├── question_gen/        # 模块3：新题构建
│   │   ├── generator.py     # 题目生成
│   │   ├── calibrator.py    # 基线校准
│   │   └── dedup.py         # 去重

→

│   ├── question_gen/        # 模块3：出题（加载 + 采样 + 未来 LLM 生成）
│   │   └── loader.py        # benchmark 加载、分层采样

• Step 2: 更新 CLAUDE.md

1. §1.5 表格（第 22 行附近）：

| `DataLoader` | 出题 question_gen | `app/question_gen/generator.py` |

→

| `DataLoader` | 出题 question_gen | `app/question_gen/loader.py` |

• Step 3: 运行 lint

Run: conda activate Video-Tree-TRM & ruff check app/ core/ --fix && ruff format app/ core/ Expected: 无错误或仅自动修复

• Step 4: 运行全量测试

Run: conda activate Video-Tree-TRM & pytest tests/ -v Expected: 全部 PASS

• Step 5: 提交

docs: 同步 question_gen 模块路径到 ARCHITECTURE.md 和 CLAUDE.md