iomgaa
dbc9d38cd7
6 个 Task: telemetry 防御加固 → call_id 根因修复 → detector L2/L1 扩展 → progress 管理 → 并发编排+CLI → lint+全量测试
895 lines
30 KiB
Markdown
895 lines
30 KiB
Markdown
# 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 测试**
|
||
|
||
```python
|
||
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` 类之后追加：
|
||
|
||
```python
|
||
@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` 中创建：
|
||
|
||
```python
|
||
"""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`：
|
||
|
||
```python
|
||
"""题目加载与分层采样。
|
||
|
||
从 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` 追加：
|
||
|
||
```python
|
||
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` 之后），然后在文件末尾追加以下函数：
|
||
|
||
```python
|
||
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`：
|
||
|
||
```python
|
||
"""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` 之后追加：
|
||
|
||
```python
|
||
@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` 块：
|
||
|
||
```python
|
||
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**
|
||
|
||
```python
|
||
"""出题模块 — 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
|
||
```
|