diff --git a/app/harness/build_split.py b/app/harness/build_split.py new file mode 100644 index 0000000..bf36a40 --- /dev/null +++ b/app/harness/build_split.py @@ -0,0 +1,504 @@ +"""结果驱动视频级切分的顶层编排：诊断信号 → 冻结 pools.json + manifest。 + +把已实现的组件串成 capstone 管线：从 harness.db 读 canonical 基线预测、从 +DiagnosisSignalStore 读逐题诊断信号，构建全视频画像、贪心联合约束选择 trainval / +test，再以视频组为原子切出诊断 / 验证池，原子冻结 pools.json 并写溯源 manifest。 +全程带六条防御断言（P5，任一不满足即 fail-fast，绝不静默兜底）。 + +只有基线推理与诊断是上游产物；本模块纯 code-controlled，不发起任何 LLM 调用， +读预测走只读连接，不改动 harness.db。 +""" + +from __future__ import annotations + +import hashlib +import sqlite3 +from collections import Counter +from dataclasses import asdict, dataclass, field +from typing import TYPE_CHECKING + +from loguru import logger + +from app.harness.pools import save_pools, split_by_video_assignment +from app.harness.split_manifest import write_manifest +from app.harness.split_selection import ( + SelectConfig, + build_video_records, + derive_reportable_types, + select_split, +) +from app.question_gen.loader import load_benchmark + +if TYPE_CHECKING: + from pathlib import Path + + from app.harness.pools import Pools + from app.harness.split_selection import SplitAssignment, VideoRecord + from core.evolution.protocols import DiagnosisSignalStore + from core.evolution.types import DiagnosisSignalRow + from core.types import GeneratedQuestion + +# 多样性主格子 = 12 题型 × 4 错误类别 = 48 格，覆盖报告以此为分母。 +_DIVERSITY_GRID_TOTAL = 48 +_QUESTIONS_PER_VIDEO = 3 + + +@dataclass(frozen=True) +class SplitBuildConfig: + """结果驱动切分的旋钮快照（科研配置，随实验扫动）。 + + 承载贪心选择器与视频组题级切分的全部可扫参数；asdict 后直接写入 manifest 的 + config 快照，保证复现时可比对。floor_k 为不可哈希容器，标 hash=False 排除出 + 自动 __hash__，避免 frozen dataclass 被哈希时报错（本类不作字典键，仅承载配置）。 + + 字段: + n_trainval: trainval 目标视频数（多样性阶段填充上限）。 + floor_k: 各高信号 task_type 的 T2 defect 下限（select_split 硬约束）。 + epsilon: test 相对全局的最大允许分布偏差（题型 / 难度两维）。 + report_floor: per-type 报告门限，题数 ≥ 此值的 task_type 才入 ε 约束。 + select_seed: 贪心选择器预洗牌种子（打破等增益平局）。 + val_ratio: validation 占 trainval 视频组总数的比例。 + split_seed: 视频组题级切分的洗牌种子。 + """ + + n_trainval: int + floor_k: dict[str, int] = field(hash=False) + epsilon: float + report_floor: int + select_seed: int + val_ratio: float + split_seed: int + + +@dataclass(frozen=True) +class SplitBuildResult: + """build_split 的返回结果（冻结产物 + 溯源）。 + + 字段: + pools: 冻结的三池（diagnosis / validation / test）。 + manifest: 写入 split_manifest.json 的溯源字典（含 pools_sha256）。 + assignment: video_id -> "trainval" | "test" 归属字典。 + """ + + pools: Pools + manifest: dict + assignment: dict[str, str] + + def __getitem__(self, key: str) -> object: + """兼容字典式访问（result["pools"] / ["manifest"] / ["assignment"]）。 + + 参数: + key: 字段名，取值 pools / manifest / assignment。 + + 返回: + 对应字段值。 + + 异常: + KeyError: key 非上述三者之一。 + """ + if key not in {"pools", "manifest", "assignment"}: + raise KeyError(f"未知字段: {key}") + return getattr(self, key) + + +def build_split( + *, + db_path: Path, + baseline_run_id: str, + signal_store: DiagnosisSignalStore, + diag_fingerprint: str, + questions_dir: Path, + config: SplitBuildConfig, + out_path: Path, + manifest_path: Path, + generated_at: str, +) -> SplitBuildResult: + """顶层编排结果驱动视频级切分，冻结 pools.json + manifest 并跑防御断言。 + + 步骤：读 canonical 基线预测 → 读诊断信号 → 构建全视频画像 → 贪心选择 trainval / + test → 加载题库并以视频归属切三池 → 原子冻结 pools.json → 写溯源 manifest → + 六条防御断言 fail-fast 校验。 + + 参数: + db_path: harness.db 路径（只读读取 predictions，不改动）。 + baseline_run_id: 基线 run 标识（如 "infer_adhoc"）。 + signal_store: 逐题诊断信号存储端口，读 (run, fingerprint) 下全部信号行。 + diag_fingerprint: 诊断口径指纹，隔离不同诊断配置的信号。 + questions_dir: benchmark 题库目录，加载 GeneratedQuestion。 + config: 切分旋钮快照。 + out_path: 冻结 pools.json 目标路径（原子写）。 + manifest_path: 溯源 manifest 目标路径（原子写）。 + generated_at: 生成时间戳（ISO 字符串），由调用方传入以保证可复现。 + + 返回: + SplitBuildResult，含 pools / manifest / assignment，支持字典式访问。 + + 异常: + AssertionError: 六条防御断言任一不满足（fail-fast，不静默）。 + ValueError: 上游依赖校验失败（如 correctness 缺题、assignment 非法）。 + """ + # Phase 1: canonical 基线预测 + 诊断信号。 + preds = _read_canonical_predictions(db_path, baseline_run_id) + signal_rows_raw = signal_store.load(baseline_run_id, diag_fingerprint) + _assert_fingerprint_consistent(signal_rows_raw, diag_fingerprint) + signal_rows = [ + { + "question_id": row.question_id, + "task_type": row.task_type, + "error_type": row.error_type, + "tier": row.tier, + } + for row in signal_rows_raw + ] + + # Phase 2: 全视频画像 + 贪心联合约束选择。 + videos = build_video_records(preds, signal_rows) + total_by_type = Counter(pred["task_type"] for pred in preds) + reportable_types = derive_reportable_types(dict(total_by_type), config.report_floor) + assignment_obj = select_split( + videos, + config=SelectConfig( + n_trainval=config.n_trainval, + floor_k=config.floor_k, + epsilon=config.epsilon, + reportable_types=reportable_types, + seed=config.select_seed, + ), + ) + assignment = _assignment_to_dict(assignment_obj) + logger.info( + "视频级切分完成: trainval={} test={} (总 {} 视频)", + len(assignment_obj.trainval), + len(assignment_obj.test), + len(videos), + ) + + # Phase 3: 加载题库 + 视频归属切三池 + 原子冻结。 + questions = load_benchmark(questions_dir) + correctness = {pred["question_id"]: pred["correct"] for pred in preds} + pools = split_by_video_assignment( + questions, + assignment, + correctness, + config.val_ratio, + config.split_seed, + baseline_run_id=baseline_run_id, + ) + save_pools(pools, out_path) + + # Phase 4: 溯源 manifest（pools_sha256 锚定冻结内容）。 + coverage_report = _build_coverage_report( + videos, assignment_obj, signal_rows_raw, reportable_types, config + ) + manifest = write_manifest( + manifest_path, + baseline_run_id=baseline_run_id, + diag_fingerprint=diag_fingerprint, + seed=config.split_seed, + config=asdict(config), + pools_json_text=out_path.read_text(encoding="utf-8"), + coverage_report=coverage_report, + generated_at=generated_at, + ) + + # Phase 5: 防御断言 fail-fast。 + _assert_split_invariants( + pools=pools, + expected_question_ids={q.question_id for q in questions}, + signal_rows_raw=signal_rows_raw, + diag_fingerprint=diag_fingerprint, + out_path=out_path, + manifest=manifest, + ) + + return SplitBuildResult(pools=pools, manifest=manifest, assignment=assignment) + + +def _normalize_choice(choice: str | None) -> str: + """选项归一：strip → 大写 → 取首字母，None 归一为空串。 + + 参数: + choice: 原始选项文本（预测或答案），可为 None。 + + 返回: + 归一后的单字母（无内容时为空串）。 + """ + return (choice or "").strip().upper()[:1] + + +def _read_canonical_predictions(db_path: Path, baseline_run_id: str) -> list[dict]: + """从 harness.db 只读取指定 run 每题首行（ORDER BY rowid）为 canonical 预测。 + + 同一 question_id 可能有多行（重跑 / 补测），canonical 口径取 rowid 最小的首行， + 保证 distinct question 计数与对错判定确定。correct = 预测与答案归一后逐字符相等。 + + 参数: + db_path: harness.db 路径。 + baseline_run_id: 基线 run 标识。 + + 返回: + canonical 预测行列表，每行 {video_id, question_id, task_type, correct}。 + + 异常: + ValueError: 该 run 无任何预测行（fail-fast，不返回空切分）。 + + 实现细节: + 以 URI mode=ro 打开只读连接，绝不改动基线 db；按 rowid 升序遍历， + 首次见到的 question_id 即 canonical 行，后续同 qid 行跳过。 + """ + conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + try: + rows = conn.execute( + "SELECT question_id, video_id, task_type, prediction, answer " + "FROM predictions WHERE run_id = ? ORDER BY rowid", + (baseline_run_id,), + ).fetchall() + finally: + conn.close() + + canonical: dict[str, dict] = {} + for row in rows: + qid = row["question_id"] + if qid in canonical: + continue + canonical[qid] = { + "question_id": qid, + "video_id": row["video_id"], + "task_type": row["task_type"], + "correct": _normalize_choice(row["prediction"]) == _normalize_choice(row["answer"]), + } + if not canonical: + raise ValueError(f"run_id={baseline_run_id} 无任何预测行，无法切分") + return list(canonical.values()) + + +def _assignment_to_dict(assignment_obj: SplitAssignment) -> dict[str, str]: + """把 SplitAssignment 展平为 video_id -> "trainval" | "test" 归属字典。 + + 参数: + assignment_obj: 贪心选择器产出的切分归属。 + + 返回: + 全部视频的归属字典（trainval 与 test 并集，键互斥）。 + """ + assignment = dict.fromkeys(assignment_obj.trainval, "trainval") + for vid in assignment_obj.test: + assignment[vid] = "test" + return assignment + + +def _assert_fingerprint_consistent( + signal_rows_raw: list[DiagnosisSignalRow], + diag_fingerprint: str, +) -> None: + """防御④：全部诊断信号行的 diag_fingerprint 必须与传入指纹一致。 + + 参数: + signal_rows_raw: store 读回的诊断信号行。 + diag_fingerprint: 期望的诊断口径指纹。 + + 异常: + AssertionError: 存在指纹不一致的信号行（store 未正确按指纹过滤）。 + """ + mismatched = [ + row.question_id for row in signal_rows_raw if row.diag_fingerprint != diag_fingerprint + ] + if mismatched: + raise AssertionError( + f"诊断信号指纹不一致 {len(mismatched)} 行，期望 {diag_fingerprint}: {mismatched[:5]}" + ) + + +def _fraction_by_type(records: list[VideoRecord], keys: set[str]) -> dict[str, float]: + """各 task_type 在给定视频集中的承载占比（含该题型的视频数 / 总视频数）。 + + 参数: + records: 视频记录子集。 + keys: 需计算占比的 task_type 键集。 + + 返回: + {task_type: 占比}；records 为空时全部记 0.0。 + """ + total = len(records) + if total == 0: + return dict.fromkeys(keys, 0.0) + return {key: sum(1 for r in records if key in r.type_set) / total for key in keys} + + +def _fraction_by_difficulty(records: list[VideoRecord], buckets: set[int]) -> dict[int, float]: + """各难度桶（错题数）在给定视频集中的占比。 + + 参数: + records: 视频记录子集。 + buckets: 难度桶键集。 + + 返回: + {难度桶: 占比}；records 为空时全部记 0.0。 + """ + total = len(records) + if total == 0: + return dict.fromkeys(buckets, 0.0) + return {bucket: sum(1 for r in records if r.difficulty == bucket) / total for bucket in buckets} + + +def _max_dev(global_dist: dict, subset_dist: dict, keys: set) -> float: + """逐键取全局与子集分布的最大绝对偏差（键集为空约定 0.0）。""" + if not keys: + return 0.0 + return max(abs(global_dist.get(k, 0.0) - subset_dist.get(k, 0.0)) for k in keys) + + +def _build_coverage_report( + videos: list[VideoRecord], + assignment_obj: SplitAssignment, + signal_rows_raw: list[DiagnosisSignalRow], + reportable_types: set[str], + config: SplitBuildConfig, +) -> dict: + """组装 manifest 覆盖报告：48 格覆盖 / floor 达标 / test 代表性偏差 / tier 占比。 + + 参数: + videos: 全视频画像记录。 + assignment_obj: 切分归属（trainval / test）。 + signal_rows_raw: 诊断信号行（统计 tier 占比）。 + reportable_types: 参与 ε 代表性校验的题型集。 + config: 切分旋钮（floor_k / epsilon）。 + + 返回: + 覆盖报告字典，含 cells_covered / grid_total / floor_satisfied / + test_representativeness_deviation / tier_distribution。 + """ + by_id = {v.video_id: v for v in videos} + trainval = [by_id[vid] for vid in assignment_obj.trainval] + test = [by_id[vid] for vid in assignment_obj.test] + + covered_cells: set[tuple[str, str]] = set() + trainval_wrong: Counter[str] = Counter() + for video in trainval: + covered_cells |= set(video.cells) + trainval_wrong.update(video.wrong_by_type) + floor_satisfied = { + task_type: trainval_wrong.get(task_type, 0) >= floor + for task_type, floor in config.floor_k.items() + } + + type_dev = _max_dev( + _fraction_by_type(videos, reportable_types), + _fraction_by_type(test, reportable_types), + reportable_types, + ) + diff_buckets = {r.difficulty for r in videos} + diff_dev = _max_dev( + _fraction_by_difficulty(videos, diff_buckets), + _fraction_by_difficulty(test, diff_buckets), + diff_buckets, + ) + + tier_counts = Counter(row.tier for row in signal_rows_raw) + total_signals = sum(tier_counts.values()) + tier_distribution = ( + {tier: count / total_signals for tier, count in tier_counts.items()} + if total_signals + else {} + ) + + return { + "cells_covered": len(covered_cells), + "grid_total": _DIVERSITY_GRID_TOTAL, + "floor_satisfied": floor_satisfied, + "test_representativeness_deviation": { + "type_max": type_dev, + "difficulty_max": diff_dev, + "epsilon": config.epsilon, + }, + "tier_distribution": tier_distribution, + } + + +def _assert_split_invariants( + *, + pools: Pools, + expected_question_ids: set[str], + signal_rows_raw: list[DiagnosisSignalRow], + diag_fingerprint: str, + out_path: Path, + manifest: dict, +) -> None: + """六条防御断言 fail-fast：任一不满足即 AssertionError（P5，不静默不兜底）。 + + ① 三池视频集两两不相交；② 三池覆盖全部题（按 distinct question）； + ③ 每 video 恰 3 题；④ 诊断信号指纹一致；⑤ manifest pools_sha256 == sha256(冻结内容)； + ⑥ question_id 全局唯一（无重复行）。 + + 参数: + pools: 冻结的三池。 + expected_question_ids: 加载题库的 question_id 全集（覆盖基准）。 + signal_rows_raw: 诊断信号行（指纹校验）。 + diag_fingerprint: 期望诊断指纹。 + out_path: 冻结 pools.json 路径。 + manifest: 已写入的 manifest 字典。 + + 异常: + AssertionError: 任一防御断言不满足。 + """ + all_questions = pools.diagnosis + pools.validation + pools.test + _assert_pools_video_disjoint(pools) # ① + _assert_question_ids_unique(all_questions) # ⑥ + _assert_question_coverage(all_questions, expected_question_ids) # ② + _assert_three_questions_per_video(all_questions) # ③ + _assert_fingerprint_consistent(signal_rows_raw, diag_fingerprint) # ④ + _assert_pools_sha256(out_path, manifest) # ⑤ + + +def _assert_pools_video_disjoint(pools: Pools) -> None: + """防御①：diagnosis / validation / test 三池视频集两两不相交。""" + diag_v = {q.video_id for q in pools.diagnosis} + val_v = {q.video_id for q in pools.validation} + test_v = {q.video_id for q in pools.test} + if diag_v & val_v or diag_v & test_v or val_v & test_v: + raise AssertionError( + f"三池视频集非互斥: diag∩val={diag_v & val_v}, " + f"diag∩test={diag_v & test_v}, val∩test={val_v & test_v}" + ) + + +def _assert_question_ids_unique(all_questions: list[GeneratedQuestion]) -> None: + """防御⑥：三池合并后 question_id 全局唯一（无重复行）。""" + qids = [q.question_id for q in all_questions] + if len(qids) != len(set(qids)): + duplicates = [qid for qid, count in Counter(qids).items() if count > 1] + raise AssertionError(f"question_id 重复 {len(duplicates)} 个: {duplicates[:5]}") + + +def _assert_question_coverage( + all_questions: list[GeneratedQuestion], + expected_question_ids: set[str], +) -> None: + """防御②：三池覆盖题库全部题（按 distinct question，缺题 / 多题均 fail-fast）。""" + actual = {q.question_id for q in all_questions} + if actual != expected_question_ids: + missing = expected_question_ids - actual + extra = actual - expected_question_ids + raise AssertionError( + f"三池题目覆盖不完整: 缺 {len(missing)} 多 {len(extra)} (缺样例 {sorted(missing)[:5]})" + ) + + +def _assert_three_questions_per_video(all_questions: list[GeneratedQuestion]) -> None: + """防御③：每 video 恰 3 题（视频组原子切分不应劈裂视频的题）。""" + per_video = Counter(q.video_id for q in all_questions) + bad_videos = {vid: n for vid, n in per_video.items() if n != _QUESTIONS_PER_VIDEO} + if bad_videos: + raise AssertionError( + f"存在 video 题数 != {_QUESTIONS_PER_VIDEO}: {dict(list(bad_videos.items())[:5])}" + ) + + +def _assert_pools_sha256(out_path: Path, manifest: dict) -> None: + """防御⑤：manifest 的 pools_sha256 == sha256(冻结 pools.json 内容)。""" + actual_sha = hashlib.sha256(out_path.read_text(encoding="utf-8").encode("utf-8")).hexdigest() + if actual_sha != manifest["pools_sha256"]: + raise AssertionError( + f"pools_sha256 不一致: manifest={manifest['pools_sha256']} 实际={actual_sha}" + ) diff --git a/tests/integration/test_build_split_e2e.py b/tests/integration/test_build_split_e2e.py new file mode 100644 index 0000000..4307af4 --- /dev/null +++ b/tests/integration/test_build_split_e2e.py @@ -0,0 +1,155 @@ +"""build_split 端到端集成测试：真实 infer_adhoc 预测 + 构造诊断信号 → 冻结三池。 + +用 workspaces/default/harness.db 的真实 infer_adhoc 基线预测（900 题 / 300 视频， +每视频 3 题）驱动结果驱动视频级切分，构造一份 T2 诊断信号 store 喂给选择器， +验证冻结的 pools.json 满足全部防御断言（三池视频互斥、覆盖 900 题、每视频 3 题、 +内容指纹一致），确保 capstone 编排把前面所有组件正确串联。 +""" + +from __future__ import annotations + +import hashlib +import sqlite3 +from pathlib import Path + +import pytest + +from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore +from app.harness.build_split import SplitBuildConfig, build_split +from app.harness.split_selection import evolution_target_of +from core.evolution.types import DiagnosisSignalRow + +_HARNESS_DB = Path("workspaces/default/harness.db") +_BENCHMARK_DIR = Path("store/questions/benchmarks/Video-MME") +_BASELINE_RUN_ID = "infer_adhoc" +_DIAG_FINGERPRINT = "diag_test_v1" +# 四类错误类别轮转，确定性铺满多样性格子（cell = task_type × error_type）。 +_ERROR_TYPES = ("extraction_failure", "search_failure", "reasoning_failure", "mixed") + + +def _normalize(choice: str | None) -> str: + """选项归一：strip → 大写 → 取首字母（None 归一为空串）。""" + return (choice or "").strip().upper()[:1] + + +def _read_canonical_predictions() -> list[dict]: + """从真实 harness.db 读 infer_adhoc 每题首行（ORDER BY rowid）作为 canonical 预测。""" + conn = sqlite3.connect(f"file:{_HARNESS_DB}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT question_id, video_id, task_type, prediction, answer " + "FROM predictions WHERE run_id = ? ORDER BY rowid", + (_BASELINE_RUN_ID,), + ).fetchall() + conn.close() + seen: dict[str, dict] = {} + for row in rows: + qid = row["question_id"] + if qid in seen: + continue + seen[qid] = { + "question_id": qid, + "video_id": row["video_id"], + "task_type": row["task_type"], + "correct": _normalize(row["prediction"]) == _normalize(row["answer"]), + } + return list(seen.values()) + + +def _populate_signal_store(store: SqliteDiagnosisSignalStore, preds: list[dict]) -> None: + """为全部错题写入 T2 诊断信号（error_type 轮转），构造可训练缺陷多样性。""" + wrong = [p for p in preds if not p["correct"]] + for idx, pred in enumerate(wrong): + error_type = _ERROR_TYPES[idx % len(_ERROR_TYPES)] + store.upsert( + DiagnosisSignalRow( + question_id=pred["question_id"], + video_id=pred["video_id"], + baseline_run_id=_BASELINE_RUN_ID, + diag_fingerprint=_DIAG_FINGERPRINT, + task_type=pred["task_type"], + error_type=error_type, + cause_category="defect", + tier="T2", + evolution_target=evolution_target_of(error_type), + degraded=False, + infra=False, + session_id=None, + ) + ) + + +@pytest.mark.skipif( + not _HARNESS_DB.exists() or not _BENCHMARK_DIR.exists(), + reason="需要真实 workspaces/default/harness.db 与 Video-MME benchmark", +) +def test_end_to_end_freezes_valid_pools(tmp_path: Path) -> None: + """真实基线预测 + 构造 T2 信号 → 冻结 pools.json，校验六条防御断言。""" + out = tmp_path / "pools.json" + manifest_path = tmp_path / "split_manifest.json" + signal_db = tmp_path / "signals.db" + + preds = _read_canonical_predictions() + store = SqliteDiagnosisSignalStore(str(signal_db)) + _populate_signal_store(store, preds) + + config = SplitBuildConfig( + n_trainval=100, + floor_k={"Object Reasoning": 3}, + epsilon=0.1, + report_floor=30, + select_seed=7, + val_ratio=0.3, + split_seed=7, + ) + + result = build_split( + db_path=_HARNESS_DB, + baseline_run_id=_BASELINE_RUN_ID, + signal_store=store, + diag_fingerprint=_DIAG_FINGERPRINT, + questions_dir=_BENCHMARK_DIR, + config=config, + out_path=out, + manifest_path=manifest_path, + generated_at="2026-07-15T00:00:00Z", + ) + store.close() + + pools = result["pools"] + + # 防御① 三池视频集两两不相交 + diag_v = {q.video_id for q in pools.diagnosis} + val_v = {q.video_id for q in pools.validation} + test_v = {q.video_id for q in pools.test} + assert diag_v & test_v == set() + assert val_v & test_v == set() + assert diag_v & val_v == set() + + # 防御② 三池覆盖 900 题（按 distinct question） + all_q = pools.diagnosis + pools.validation + pools.test + assert len({q.question_id for q in all_q}) == 900 + assert len(all_q) == 900 + + # 防御③ 每 video 恰 3 题 + per_video: dict[str, int] = {} + for q in all_q: + per_video[q.video_id] = per_video.get(q.video_id, 0) + 1 + assert set(per_video.values()) == {3} + + # 防御⑤ manifest 的 pools_sha256 == sha256(out 文件内容) + assert ( + result["manifest"]["pools_sha256"] + == hashlib.sha256(out.read_text(encoding="utf-8").encode("utf-8")).hexdigest() + ) + + # assignment 覆盖全部 300 视频且取值合法 + assignment = result["assignment"] + assert set(assignment.values()) <= {"trainval", "test"} + assert len(assignment) == 300 + + # manifest 覆盖报告含关键指标 + coverage = result["manifest"]["coverage_report"] + assert coverage["grid_total"] == 48 + assert "tier_distribution" in coverage + assert manifest_path.exists()