From c3187167c87fa0712ead918cea70f55bebdd3a4d Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 15 Jul 2026 12:32:24 -0400 Subject: [PATCH] feat: add greedy joint-constrained split selector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 两阶段贪心视频级切分:Floor 硬约束优先满足 + 多样性 submodular 覆盖最大化, 两阶段均带 ε 守护保 test 代表性;不可行 fail loud(InfeasibleSplitError), 多样性欠额记 loguru warning 不静默。derive_reportable_types 落地长尾报告门限。 确定性:Random(seed) 预洗牌打破等增益/等槽数平局,同 config 同 videos 同解。 Co-Authored-By: Claude Opus 4.8 (1M context) --- app/harness/split_selection.py | 329 +++++++++++++++++++++++++++++ tests/unit/test_split_selection.py | 149 ++++++++++++- 2 files changed, 477 insertions(+), 1 deletion(-) diff --git a/app/harness/split_selection.py b/app/harness/split_selection.py index 3cd6ad3..9406c17 100644 --- a/app/harness/split_selection.py +++ b/app/harness/split_selection.py @@ -7,8 +7,11 @@ build_video_records / select_split。 from __future__ import annotations +import random from dataclasses import dataclass, field +from loguru import logger + _EVOLUTION_TARGET = { "extraction_failure": "tool", "search_failure": "skill", @@ -186,3 +189,329 @@ def build_video_records(preds: list[dict], signal_rows: list[dict]) -> list[Vide ) ) return records + + +@dataclass(frozen=True) +class SelectConfig: + """贪心联合约束选择器的实验配置(科研配置,随实验扫动)。 + + 字段: + n_trainval: trainval 目标视频数(多样性阶段的填充上限)。 + floor_k: 各高信号 task_type 的 T2 defect 数下限(硬约束,floor 阶段满足)。 + epsilon: test 相对全局的最大允许分布偏差(题型占比 / 难度画像两维,逐桶)。 + reportable_types: 参与 ε 题型代表性校验的 task_type 集(长尾类型不入约束)。 + seed: 预洗牌随机种子,仅用于打破等增益平局,保证同 config 同 videos 同解。 + + 实现细节: + floor_k / reportable_types 为不可哈希容器,标 hash=False 排除出自动 __hash__, + 避免 frozen dataclass 被哈希时报错(本类不作为字典键,仅承载配置)。 + """ + + n_trainval: int + floor_k: dict[str, int] = field(hash=False) + epsilon: float + reportable_types: frozenset[str] | set[str] = field(hash=False) + seed: int + + +@dataclass(frozen=True) +class SplitAssignment: + """视频级切分归属结果(交给 split_by_video_assignment 做题级切分)。 + + 字段: + trainval: 进入 trainval 的 video_id 元组(按选择顺序,确定性)。 + test: 补集视频的 video_id 元组(按 videos 原始顺序)。 + """ + + trainval: tuple[str, ...] + test: tuple[str, ...] + + +class InfeasibleSplitError(Exception): + """floor 硬约束与 ε 守护死锁、无法在不破坏 test 代表性下满足 floor 时抛出。 + + fail loud(P5):不静默兜底、不随机塞题,直接暴露不可行并报告未达标类型。 + """ + + +def derive_reportable_types(total_by_type: dict[str, int], report_floor: int) -> set[str]: + """派生可 per-type 报告的 task_type 集(长尾处理:题数 ≥ report_floor 才报告)。 + + 参数: + total_by_type: 各 task_type 的总题数(或代理承载数)。 + report_floor: 报告门限,低于此的类型并入长尾、不单独报告也不入 ε 约束。 + + 返回: + 总题数 ≥ report_floor 的 task_type 集合。 + """ + return {task_type for task_type, total in total_by_type.items() if total >= report_floor} + + +def _type_membership_fraction(records: list[VideoRecord], keys: set[str]) -> dict[str, float]: + """计算各 task_type 在给定视频集中的承载占比(含该题型的视频数 / 总视频数)。 + + 参数: + records: 视频记录子集(非空,调用方保证)。 + keys: 需计算占比的 task_type 键集。 + + 返回: + {task_type: 占比},占比 ∈ [0, 1]。 + """ + total = len(records) + return {key: sum(1 for r in records if key in r.type_set) / total for key in keys} + + +def _difficulty_fraction(records: list[VideoRecord], buckets: set[int]) -> dict[int, float]: + """计算各难度桶在给定视频集中的占比(难度 = 错题数)。 + + 参数: + records: 视频记录子集(非空,调用方保证)。 + buckets: 需计算占比的难度桶键集。 + + 返回: + {难度桶: 占比},占比 ∈ [0, 1]。 + """ + total = len(records) + return {bucket: sum(1 for r in records if r.difficulty == bucket) / total for bucket in buckets} + + +def _max_deviation(global_dist: dict, subset_dist: dict, keys: set) -> float: + """逐键取全局与子集分布的最大绝对偏差(键集为空时约定为 0.0)。 + + 参数: + global_dist: 全局分布(键 → 占比)。 + subset_dist: 子集分布(键 → 占比)。 + keys: 参与比较的键集。 + + 返回: + 逐键 |global - subset| 的最大值;keys 为空返回 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 _epsilon_ok( + test_video_records: list[VideoRecord], + videos_all: list[VideoRecord], + config: SelectConfig, +) -> bool: + """校验 test 子集相对全局在题型占比与难度画像两维的偏差是否均 ≤ epsilon。 + + test 越简单则 headline 越虚高,故 test 必须保持代表性:逐 reportable 题型、逐难度桶 + 比较 test 与全局占比,任一维超 epsilon 即判不合格。 + + 参数: + test_video_records: 候选 test 子集(trainval 补集)。 + videos_all: 全部视频(全局分布基准)。 + config: 选择配置,提供 epsilon 与 reportable_types。 + + 返回: + 两维最大偏差均 ≤ epsilon 为 True;test 为空视为不合格返回 False。 + """ + if not test_video_records: + return False + type_keys = set(config.reportable_types) + global_type = _type_membership_fraction(videos_all, type_keys) + subset_type = _type_membership_fraction(test_video_records, type_keys) + if _max_deviation(global_type, subset_type, type_keys) > config.epsilon: + return False + diff_keys = {r.difficulty for r in videos_all} + global_diff = _difficulty_fraction(videos_all, diff_keys) + subset_diff = _difficulty_fraction(test_video_records, diff_keys) + return _max_deviation(global_diff, subset_diff, diff_keys) <= config.epsilon + + +def _current_wrong_counts(selected: list[VideoRecord]) -> dict[str, int]: + """聚合已选 trainval 视频的 T2 defect 计数(供 floor 达标判定)。 + + 参数: + selected: 当前已进入 trainval 的视频记录。 + + 返回: + {task_type: T2 defect 累计数}。 + """ + counts: dict[str, int] = {} + for video in selected: + for task_type, wrong in video.wrong_by_type.items(): + counts[task_type] = counts.get(task_type, 0) + wrong + return counts + + +def _unmet_floors(selected: list[VideoRecord], floor_k: dict[str, int]) -> dict[str, int]: + """计算尚未达标的 floor 类型及其缺口(已达标类型不返回)。 + + 参数: + selected: 当前已进入 trainval 的视频记录。 + floor_k: 各高信号 task_type 的 defect 下限。 + + 返回: + {task_type: 缺口数},仅含 current < floor 的类型;全达标返回空 dict。 + """ + counts = _current_wrong_counts(selected) + return { + task_type: floor - counts.get(task_type, 0) + for task_type, floor in floor_k.items() + if counts.get(task_type, 0) < floor + } + + +def _floor_fill_count(video: VideoRecord, deficits: dict[str, int]) -> int: + """计算某视频能填补的 floor 缺口槽数(逐类型取 min(defect, 缺口) 求和)。 + + 参数: + video: 候选视频记录。 + deficits: 各未达标类型的缺口。 + + 返回: + 该视频实际可填的槽数总和(0 表示对当前缺口无贡献)。 + """ + return sum( + min(video.wrong_by_type.get(task_type, 0), deficit) + for task_type, deficit in deficits.items() + ) + + +def _marginal_gain(video: VideoRecord, current_cells: set[tuple[str, str]]) -> int: + """计算把某视频移入 trainval 的边际覆盖增益(新开的 T2 格子数)。 + + 参数: + video: 候选视频记录。 + current_cells: 当前 trainval 的 T2 格子并集。 + + 返回: + video.cells 相对 current_cells 的新增格子数(去重)。 + """ + return len(video.cells - current_cells) + + +def _prospective_test(pool: list[VideoRecord], candidate: VideoRecord) -> list[VideoRecord]: + """构造"把候选移入 trainval 后"的 test 子集 = 当前剩余池去掉候选。 + + 参数: + pool: 当前尚未进入 trainval 的视频(即当前 test 补集)。 + candidate: 拟移入 trainval 的候选视频。 + + 返回: + pool 去掉 candidate 后的视频列表。 + """ + return [r for r in pool if r.video_id != candidate.video_id] + + +def _satisfy_floors( + selected: list[VideoRecord], + pool: list[VideoRecord], + videos_all: list[VideoRecord], + config: SelectConfig, +) -> None: + """Floor 阶段:硬约束优先,逐步移入能填 floor 槽且不破 ε 的视频(就地改 selected/pool)。 + + 每轮取未达标类型的缺口,候选 = 能填 ≥1 槽 且 移入后 test 仍满足 ε 的视频;候选为空即 + 死锁抛 InfeasibleSplitError;否则选填槽最多者(等槽数按预洗牌顺序取首个,确定性)。 + + 参数: + selected: 当前 trainval(就地追加)。 + pool: 当前剩余池 = test 补集(就地移除)。 + videos_all: 全部视频(ε 全局基准)。 + config: 选择配置。 + + 异常: + InfeasibleSplitError: 存在未达标类型但无候选可在不破 ε 下填补。 + """ + while True: + deficits = _unmet_floors(selected, config.floor_k) + if not deficits: + return + candidates = [ + video + for video in pool + if _floor_fill_count(video, deficits) > 0 + and _epsilon_ok(_prospective_test(pool, video), videos_all, config) + ] + if not candidates: + raise InfeasibleSplitError( + f"floor 无法在 ε≤{config.epsilon} 下满足,未达标类型缺口: {dict(deficits)}" + ) + pick = max(candidates, key=lambda video: _floor_fill_count(video, deficits)) + selected.append(pick) + pool.remove(pick) + + +def _maximize_diversity( + selected: list[VideoRecord], + pool: list[VideoRecord], + videos_all: list[VideoRecord], + config: SelectConfig, +) -> None: + """多样性阶段:submodular 贪心,按边际覆盖增益降序填至 n_trainval(就地改 selected/pool)。 + + 每轮对剩余视频算新开格子数,按 -增益稳定排序(等增益按预洗牌顺序),取第一个移入后 test + 仍满足 ε 的视频;若无任一视频可加而不破 ε,则停并记 warning(欠额,不静默不报错)。 + + 参数: + selected: 当前 trainval(就地追加)。 + pool: 当前剩余池 = test 补集(就地移除)。 + videos_all: 全部视频(ε 全局基准)。 + config: 选择配置。 + """ + while len(selected) < config.n_trainval: + if not pool: + logger.warning( + "多样性阶段剩余池耗尽,trainval 欠额: {}/{}", len(selected), config.n_trainval + ) + return + current_cells = set().union(*(v.cells for v in selected)) if selected else set() + ranked = sorted(pool, key=lambda video: -_marginal_gain(video, current_cells)) + pick = next( + ( + video + for video in ranked + if _epsilon_ok(_prospective_test(pool, video), videos_all, config) + ), + None, + ) + if pick is None: + logger.warning( + "多样性阶段 ε 守护阻断全部候选,trainval 欠额: {}/{}", + len(selected), + config.n_trainval, + ) + return + selected.append(pick) + pool.remove(pick) + + +def select_split(videos: list[VideoRecord], *, config: SelectConfig) -> SplitAssignment: + """贪心联合约束视频级切分:floor 硬约束先满足、多样性覆盖后最大化、ε 守护 test 代表性。 + + 核心洞察:全数据集错题总数固定,越把信号塞 trainval、test 越简单、headline 越虚高, + 故 test 必须保持代表性(ε 约束),trainval 只靠 floor + 多样性覆盖富集,不从 test 偷难题。 + + 两阶段贪心(均带 ε 守护):先 Floor 阶段满足各高信号类型 defect 下限(不可行 fail loud), + 再多样性阶段按边际覆盖增益填至 n_trainval(欠额记 warning)。test = trainval 补集。 + + 确定性:入场用 random.Random(seed) 对视频列表做一次预洗牌,此后 max / 稳定排序仅取首个, + seed 只打破等增益 / 等槽数平局;同 config 同 videos → 同结果。 + + 参数: + videos: 全部视频记录(Task 7 build_video_records 产物)。 + config: 选择配置(关键字传入,含 n_trainval / floor_k / epsilon / reportable_types / seed)。 + + 返回: + SplitAssignment,trainval 按选择顺序、test 按 videos 原始顺序。 + + 异常: + InfeasibleSplitError: videos 为空,或 floor 与 ε 死锁无法满足。 + """ + if not videos: + raise InfeasibleSplitError("videos 为空,无法执行切分") + rng = random.Random(config.seed) + pool = list(videos) + rng.shuffle(pool) + selected: list[VideoRecord] = [] + _satisfy_floors(selected, pool, videos, config) + _maximize_diversity(selected, pool, videos, config) + trainval_ids = {video.video_id for video in selected} + trainval = tuple(video.video_id for video in selected) + test = tuple(video.video_id for video in videos if video.video_id not in trainval_ids) + return SplitAssignment(trainval=trainval, test=test) diff --git a/tests/unit/test_split_selection.py b/tests/unit/test_split_selection.py index 9e24245..5708bfd 100644 --- a/tests/unit/test_split_selection.py +++ b/tests/unit/test_split_selection.py @@ -1,6 +1,98 @@ +import sqlite3 +from collections import Counter +from itertools import cycle +from pathlib import Path + import pytest -from app.harness.split_selection import cell_of, evolution_target_of, score_signal +from app.harness.split_selection import ( + build_video_records, + cell_of, + evolution_target_of, + score_signal, +) + +# T2 信号构造用的错误类别循环源(4 类真实 error_type,制造多样格子)。 +_ERROR_TYPES = ["extraction_failure", "search_failure", "reasoning_failure", "mixed"] +# 二次构造 T2 信号覆盖的高信号题型(floor / diversity 需有料)。 +_SIGNAL_TYPES = {"Counting Problem", "Action Reasoning"} +_HARNESS_DB = Path(__file__).resolve().parents[2] / "workspaces" / "default" / "harness.db" + + +def _load_adhoc_predictions() -> list[dict]: + """从真实 harness.db 的 infer_adhoc run 读取 900 条预测(按 question_id 去重)。 + + 返回: + preds 列表,每行含 video_id / question_id / task_type / correct, + correct = (prediction 非空 且 prediction == answer)。 + """ + con = sqlite3.connect(str(_HARNESS_DB)) + try: + cur = con.execute( + "SELECT question_id, video_id, task_type, prediction, answer " + "FROM predictions WHERE run_id = 'infer_adhoc'" + ) + seen: set[str] = set() + preds: list[dict] = [] + for question_id, video_id, task_type, prediction, answer in cur.fetchall(): + if question_id in seen: + continue + seen.add(question_id) + preds.append( + { + "video_id": video_id, + "question_id": question_id, + "task_type": task_type, + "correct": prediction is not None and prediction == answer, + } + ) + return preds + finally: + con.close() + + +def _real_shaped_video_records() -> list: + """用真实预测结构 + 二次构造 T2 信号构建 300 个 VideoRecord。 + + 题型 / 难度画像取自真实 infer_adhoc 预测;因真实诊断尚未跑,T2 信号是二次构造: + 对高信号题型(Counting / Action Reasoning)的错题标 T2 defect,error_type 循环取 4 类, + 使 floor 约束有料、多样性格子有区分度。 + + 返回: + 全部 300 个真实结构的 VideoRecord 列表。 + """ + preds = _load_adhoc_predictions() + err = cycle(_ERROR_TYPES) + signal_rows: list[dict] = [] + for pred in preds: + if pred["task_type"] in _SIGNAL_TYPES and not pred["correct"]: + signal_rows.append( + { + "question_id": pred["question_id"], + "task_type": pred["task_type"], + "error_type": next(err), + "tier": "T2", + } + ) + return build_video_records(preds, signal_rows) + + +def _count_questions_by_type(videos: list) -> dict: + """统计各 task_type 的视频承载数(含该题型的视频数),供 derive_reportable_types。 + + VideoRecord 只留题型去重集 type_set,故这里按"含该题型的视频数"聚合, + 作为长尾报告门限的代理度量。 + + 参数: + videos: VideoRecord 列表。 + + 返回: + {task_type: 含该题型的视频数}。 + """ + counter: Counter = Counter() + for video in videos: + counter.update(video.type_set) + return dict(counter) def test_evolution_target_mapping(): @@ -61,3 +153,58 @@ def test_build_video_records_covers_all_videos_with_difficulty_and_types(): assert v1.cells == {("Counting Problem", "search_failure")} assert v1.type_set == {"Counting Problem", "Action Reasoning", "OCR Problems"} assert v1.wrong_by_type == {"Counting Problem": 1} # T2 计数供 floor + + +def test_select_split_video_disjoint_and_floor_and_deterministic(): + from app.harness.split_selection import ( + SelectConfig, + derive_reportable_types, + select_split, + ) + + videos = _real_shaped_video_records() + total_by_type = _count_questions_by_type(videos) + cfg = SelectConfig( + n_trainval=100, + floor_k={"Counting Problem": 3}, + epsilon=0.1, + reportable_types=derive_reportable_types(total_by_type, report_floor=27), + seed=7, + ) + a = select_split(videos, config=cfg) + b = select_split(videos, config=cfg) + assert set(a.trainval) & set(a.test) == set() # 视频级不相交 + assert set(a.trainval) | set(a.test) == {v.video_id for v in videos} # 补集覆盖全集 + assert a.trainval == b.trainval # 同 seed 同 config → 同解 + trainval_ids = set(a.trainval) + counting_defects = sum( + v.wrong_by_type.get("Counting Problem", 0) for v in videos if v.video_id in trainval_ids + ) + assert counting_defects >= 3 # floor 硬约束满足 + + +def test_infeasible_floor_vs_epsilon_raises(): + from app.harness.split_selection import ( + InfeasibleSplitError, + SelectConfig, + select_split, + ) + + videos = _real_shaped_video_records() + with pytest.raises(InfeasibleSplitError): # 极小 ε + 高 floor → 死锁 + select_split( + videos, + config=SelectConfig( + n_trainval=2, + floor_k={"OCR Problems": 50}, + epsilon=0.001, + reportable_types=set(), + seed=1, + ), + ) + + +def test_derive_reportable_types(): + from app.harness.split_selection import derive_reportable_types + + assert derive_reportable_types({"A": 30, "B": 10}, report_floor=27) == {"A"}