From 84b52a031121a83cc82f7bfd0f3ca6720c3a294c Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 10:41:46 -0400 Subject: [PATCH] feat(pools): auto-supplement maintenance correct questions in PerCategoryPoolStrategy Co-Authored-By: Claude Opus 4.6 (1M context) --- app/harness/pools.py | 137 ++++++++++++- app/ports.py | 2 + tests/integration/test_pool_strategy.py | 254 +++++++++++++++++++++++- 3 files changed, 381 insertions(+), 12 deletions(-) diff --git a/app/harness/pools.py b/app/harness/pools.py index ac98d27..bffed72 100644 --- a/app/harness/pools.py +++ b/app/harness/pools.py @@ -116,6 +116,8 @@ class GlobalPoolStrategy: questions: list[GeneratedQuestion], correctness: dict[str, bool], config: PoolConfig, + *, + db_path: Path | None = None, ) -> Pools: """委托给现有 build_pools 函数。 @@ -534,7 +536,7 @@ def build_or_load_pools( ) correctness = {r["question_id"]: r["prediction"] == r["answer"] for r in rows} - pools = strategy.build(questions, correctness, pool_config) + pools = strategy.build(questions, correctness, pool_config, db_path=db_path) save_pools( pools, pools_path, @@ -558,6 +560,8 @@ class PerCategoryPoolStrategy: questions: list[GeneratedQuestion], correctness: dict[str, bool], config: PoolConfig, + *, + db_path: Path | None = None, ) -> Pools: """按题型分组后，每组做 correctness 分层的 train/val 划分。 @@ -565,7 +569,9 @@ class PerCategoryPoolStrategy: questions: 题目全集。 correctness: question_id -> 基线是否答对。 config: 池构建配置（使用 train_ratio, task_types, seed, - baseline_run_id, test_questions_dir）。 + baseline_run_id, test_questions_dir, batch_correct_ratio）。 + db_path: harness.db 路径，用于查询 benchmark 历史推理记录 + （maintenance 补入时需要）。 返回: 冻结的 Pools（diagnosis=train, validation=val, @@ -583,6 +589,15 @@ class PerCategoryPoolStrategy: for q in filtered: groups[q.task_type].append(q) + # Phase 2.5: 正确率检查 + maintenance 补入 + if config.batch_correct_ratio is not None: + self._check_and_supplement_maintenance( + groups, + correctness, + config, + db_path, + ) + # Phase 3: 每组分层划分 all_train: list[GeneratedQuestion] = [] all_val: list[GeneratedQuestion] = [] @@ -624,6 +639,124 @@ class PerCategoryPoolStrategy: }, ) + def _check_and_supplement_maintenance( + self, + groups: dict[str, list[GeneratedQuestion]], + correctness: dict[str, bool], + config: PoolConfig, + db_path: Path | None, + ) -> None: + """按 task_type 检查正确率，过高警告，过低则从 benchmark 补入正确题。 + + 修改 groups 和 correctness（原地更新）。 + + 参数: + groups: task_type -> 题目列表映射（原地追加补入题）。 + correctness: question_id -> 是否正确映射（原地追加补入题标记）。 + config: 含 batch_correct_ratio 和 test_questions_dir。 + db_path: harness.db 路径，用于查询 benchmark 历史推理记录。 + """ + import sqlite3 + + r = config.batch_correct_ratio + assert r is not None # 调用方已保证 + + # Phase 2.5a: 正确率检查（不依赖 test_questions_dir） + for task_type, group in groups.items(): + c = sum(1 for q in group if correctness.get(q.question_id, False)) + n = len(group) + ratio = c / n if n > 0 else 0.0 + if ratio > 1 - r: + logger.warning( + "类别 {} 正确率 {:.1%} 过高（阈值 {:.1%}），出题可能太简单", + task_type, + ratio, + 1 - r, + ) + + # Phase 2.5b: maintenance 补入（需要 test_questions_dir） + if config.test_questions_dir is None: + return + + from app.question_gen import load_benchmark + + bench_questions = load_benchmark(config.test_questions_dir) + + # 查询 DB 中 benchmark 题的历史正确性 + bench_correctness: dict[str, bool] = {} + if db_path is not None and db_path.exists(): + conn = sqlite3.connect(str(db_path)) + bench_qids = [q.question_id for q in bench_questions] + if bench_qids: + placeholders = ",".join("?" for _ in bench_qids) + rows = conn.execute( + f"SELECT question_id, prediction, answer FROM predictions " # noqa: S608 + f"WHERE question_id IN ({placeholders}) " + f"ORDER BY timestamp DESC", + bench_qids, + ).fetchall() + for qid, pred, ans in rows: + if qid not in bench_correctness: + bench_correctness[qid] = pred == ans + conn.close() + + # 按 task_type 索引 benchmark 题 + bench_by_type: dict[str, list[GeneratedQuestion]] = defaultdict(list) + for q in bench_questions: + bench_by_type[q.task_type].append(q) + + for task_type, group in groups.items(): + c = sum(1 for q in group if correctness.get(q.question_id, False)) + w = len(group) - c + n = len(group) + ratio = c / n if n > 0 else 0.0 + + if ratio >= r: + continue + + k = math.ceil((r * w - (1 - r) * c) / (1 - r)) + + existing_ids = {q.question_id for q in group} + candidates = [ + q + for q in bench_by_type.get(task_type, []) + if bench_correctness.get(q.question_id, False) and q.question_id not in existing_ids + ] + + if not candidates: + logger.warning( + "类别 {} 需补入 {} 道正确题，但 benchmark 中无可用候选", + task_type, + k, + ) + continue + + actual = min(k, len(candidates)) + for q in candidates[:actual]: + supplemented = GeneratedQuestion( + question_id=q.question_id, + video_id=q.video_id, + task_type=q.task_type, + question=q.question, + options=q.options, + answer=q.answer, + source_nodes=q.source_nodes, + difficulty=q.difficulty, + family="VME_MAINTENANCE", + skill_target=q.skill_target, + difficulty_steps=q.difficulty_steps, + ) + group.append(supplemented) + correctness[supplemented.question_id] = True + + logger.info( + "类别 {} 正确率 {:.1%} < {:.1%}，从 benchmark 补入 {} 道 maintenance 正确题", + task_type, + ratio, + r, + actual, + ) + def _split_one_category( self, questions: list[GeneratedQuestion], diff --git a/app/ports.py b/app/ports.py index 5fdb77f..3b4069d 100644 --- a/app/ports.py +++ b/app/ports.py @@ -160,6 +160,8 @@ class PoolStrategy(Protocol): questions: list[GeneratedQuestion], correctness: dict[str, bool], config: PoolConfig, + *, + db_path: Path | None = None, ) -> Pools: ... def build_incremental( diff --git a/tests/integration/test_pool_strategy.py b/tests/integration/test_pool_strategy.py index 2dca207..f4064b2 100644 --- a/tests/integration/test_pool_strategy.py +++ b/tests/integration/test_pool_strategy.py @@ -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}" + )