iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(pools): auto-supplement maintenance correct questions in PerCategoryPoolStrategy

Browse Source 
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-14 10:41:46 -04:00
parent 72befa2bd4
commit 84b52a0311
3 changed files with 381 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_pool_strategy.py
+244 -10
View File
@@ -6,11 +6,11 @@
from __future__ import annotations
import json
import sqlite3
from collections import Counter
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pathlib import Path
from loguru import logger
from app.harness.pools import (
PerCategoryPoolStrategy,
@@ -19,6 +19,9 @@ from app.harness.pools import (
)
from core.types import GeneratedQuestion, PoolConfig
if TYPE_CHECKING:
from pathlib import Path
def _make_question(qid: str, task_type: str) -> GeneratedQuestion:
"""构造测试用 GeneratedQuestion。"""
@@ -40,17 +43,17 @@ class TestPerCategoryE2E:
def test_full_flow(self, tmp_path: Path) -> None:
"""完整流程：12 类各 30 题 → 策略构建 → 冻结 → 加载 → 三池校验。"""
task_types = [
"Action Prediction",
"Action Reasoning",
"Action Recognition",
"Action Sequence",
"Causal Reasoning",
"Event Reasoning",
"Object Interaction",
"Object Reasoning",
"Action Reasoning",
"Attribute Perception",
"Counting Problem",
"Information Synopsis",
"Object Recognition",
"Scene Understanding",
"Object Reasoning",
"OCR Problems",
"Spatial Perception",
"Spatial Reasoning",
"Temporal Perception",
"Temporal Reasoning",
]
questions = []
@@ -150,3 +153,234 @@ class TestPerCategoryE2E:
loaded = load_pools(pools_path)
assert len(loaded.diagnosis) == 20
assert len(loaded.validation) == 10
# ── helpers ──
def _make_benchmark_dir(tmp_path: Path, task_type: str, n: int) -> Path:
"""在 tmp_path 下创建含 n 道题的 benchmark JSON 目录。"""
bench_dir = tmp_path / "benchmark"
bench_dir.mkdir(exist_ok=True)
questions = []
for i in range(n):
questions.append(
{
"question_id": f"bench_{task_type}_{i:03d}",
"video_id": "v_bench",
"task_type": task_type,
"question": f"Benchmark Q{i}?",
"options": ["A. a", "B. b", "C. c", "D. d"],
"answer": "A",
"source_nodes": ["n1"],
"difficulty": "medium",
}
)
(bench_dir / "bench.json").write_text(
json.dumps(questions, ensure_ascii=False),
encoding="utf-8",
)
return bench_dir
def _make_db_with_correctness(
tmp_path: Path,
qids_correct: list[str],
qids_wrong: list[str],
) -> Path:
"""创建含 predictions 表的 SQLite DB，模拟 benchmark 历史推理记录。"""
db_path = tmp_path / "harness.db"
conn = sqlite3.connect(str(db_path))
conn.execute(
"CREATE TABLE predictions ("
" question_id TEXT, prediction TEXT, answer TEXT, timestamp TEXT"
")"
)
for qid in qids_correct:
conn.execute(
"INSERT INTO predictions VALUES (?, ?, ?, ?)",
(qid, "A", "A", "2026-01-01T00:00:00"),
)
for qid in qids_wrong:
conn.execute(
"INSERT INTO predictions VALUES (?, ?, ?, ?)",
(qid, "B", "A", "2026-01-01T00:00:00"),
)
conn.commit()
conn.close()
return db_path
class TestMaintenanceSupplementation:
"""maintenance 自动补入集成测试。"""
def test_supplements_when_correct_ratio_too_low(self, tmp_path: Path) -> None:
"""正确率过低时从 benchmark 补入正确题。
30 题（5 correct, 25 wrong），benchmark 40 题（30 correct），
batch_correct_ratio=0.5 → 需补入 k=ceil((0.5*25 - 0.5*5)/0.5)=20，
总量变为 50。
"""
tt = "Object Recognition"
questions = [_make_question(f"q_{i:03d}", tt) for i in range(30)]
correctness: dict[str, bool] = {}
for i, q in enumerate(questions):
correctness[q.question_id] = i < 5 # 前 5 正确
bench_dir = _make_benchmark_dir(tmp_path, tt, 40)
# benchmark 前 30 题在 DB 中为正确
bench_qids_correct = [f"bench_{tt}_{i:03d}" for i in range(30)]
bench_qids_wrong = [f"bench_{tt}_{i:03d}" for i in range(30, 40)]
db_path = _make_db_with_correctness(tmp_path, bench_qids_correct, bench_qids_wrong)
config = PoolConfig(
task_types=(tt,),
seed=42,
baseline_run_id="bl",
diag_size=0,
diag_correct_ratio=0.0,
val_size=0,
val_correct_ratio=0.0,
test_size=0,
eval_min_per_class=0,
train_ratio=0.7,
test_questions_dir=bench_dir,
batch_correct_ratio=0.5,
)
strategy = PerCategoryPoolStrategy()
pools = strategy.build(questions, correctness, config, db_path=db_path)
total = len(pools.diagnosis) + len(pools.validation)
assert total == 50, f"补入后总量应为 50，实际 {total}"
# 验证补入题的 family 标记
all_qs = pools.diagnosis + pools.validation
maint = [q for q in all_qs if q.family == "VME_MAINTENANCE"]
assert len(maint) == 20
def test_no_supplement_when_ratio_satisfied(self, tmp_path: Path) -> None:
"""正确率达标时无需补入。
30 题（20 correct），batch_correct_ratio=0.5，无 test_questions_dir。
"""
tt = "Object Recognition"
questions = [_make_question(f"q_{i:03d}", tt) for i in range(30)]
correctness = {q.question_id: (i < 20) for i, q in enumerate(questions)}
config = PoolConfig(
task_types=(tt,),
seed=42,
baseline_run_id="bl",
diag_size=0,
diag_correct_ratio=0.0,
val_size=0,
val_correct_ratio=0.0,
test_size=0,
eval_min_per_class=0,
train_ratio=0.7,
test_questions_dir=None,
batch_correct_ratio=0.5,
)
strategy = PerCategoryPoolStrategy()
pools = strategy.build(questions, correctness, config)
total = len(pools.diagnosis) + len(pools.validation)
assert total == 30, f"无补入时总量应为 30，实际 {total}"
def test_no_supplement_when_ratio_not_configured(self) -> None:
"""batch_correct_ratio=None 时不触发补入逻辑。"""
tt = "Object Recognition"
questions = [_make_question(f"q_{i:03d}", tt) for i in range(30)]
correctness = {q.question_id: (i < 5) for i, q in enumerate(questions)}
config = PoolConfig(
task_types=(tt,),
seed=42,
baseline_run_id="bl",
diag_size=0,
diag_correct_ratio=0.0,
val_size=0,
val_correct_ratio=0.0,
test_size=0,
eval_min_per_class=0,
train_ratio=0.7,
test_questions_dir=None,
batch_correct_ratio=None,
)
strategy = PerCategoryPoolStrategy()
pools = strategy.build(questions, correctness, config)
total = len(pools.diagnosis) + len(pools.validation)
assert total == 30
def test_caps_at_available_candidates(self, tmp_path: Path) -> None:
"""benchmark 候选不足时以实际可用数量为上限。
30 题全部 wrong，benchmark 仅 10 题 correct → 补入 10 题，总量 40。
"""
tt = "Object Recognition"
questions = [_make_question(f"q_{i:03d}", tt) for i in range(30)]
correctness = {q.question_id: False for q in questions}
bench_dir = _make_benchmark_dir(tmp_path, tt, 15)
bench_qids_correct = [f"bench_{tt}_{i:03d}" for i in range(10)]
bench_qids_wrong = [f"bench_{tt}_{i:03d}" for i in range(10, 15)]
db_path = _make_db_with_correctness(tmp_path, bench_qids_correct, bench_qids_wrong)
config = PoolConfig(
task_types=(tt,),
seed=42,
baseline_run_id="bl",
diag_size=0,
diag_correct_ratio=0.0,
val_size=0,
val_correct_ratio=0.0,
test_size=0,
eval_min_per_class=0,
train_ratio=0.7,
test_questions_dir=bench_dir,
batch_correct_ratio=0.5,
)
strategy = PerCategoryPoolStrategy()
pools = strategy.build(questions, correctness, config, db_path=db_path)
total = len(pools.diagnosis) + len(pools.validation)
assert total == 40, f"候选不足时总量应为 40，实际 {total}"
def test_high_correct_ratio_warning(self) -> None:
"""正确率过高时发出警告。
30 题（28 correct），batch_correct_ratio=0.5，test_questions_dir=None。
阈值 1-0.5=0.5，实际 28/30=93.3% > 50% → 触发警告。
"""
tt = "Object Recognition"
questions = [_make_question(f"q_{i:03d}", tt) for i in range(30)]
correctness = {q.question_id: (i < 28) for i, q in enumerate(questions)}
config = PoolConfig(
task_types=(tt,),
seed=42,
baseline_run_id="bl",
diag_size=0,
diag_correct_ratio=0.0,
val_size=0,
val_correct_ratio=0.0,
test_size=0,
eval_min_per_class=0,
train_ratio=0.7,
test_questions_dir=None,
batch_correct_ratio=0.5,
)
strategy = PerCategoryPoolStrategy()
captured: list[str] = []
sink_id = logger.add(lambda msg: captured.append(str(msg)), level="WARNING")
try:
strategy.build(questions, correctness, config)
finally:
logger.remove(sink_id)
assert any("出题可能太简单" in m for m in captured), (
f"未捕获到正确率过高警告，captured: {captured}"
)