"""视频级切分选择:signal 分层、视频聚合、贪心联合约束选择(纯函数)。 结果驱动切分管线的核心:把诊断信号投影为多样性格子,供贪心选择器最大化覆盖。 本模块起步定义 evolution_target 派生与多样性格子;后续追加 score_signal / 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", "reasoning_failure": "skill", "mixed": "system", } def evolution_target_of(error_type: str) -> str: """由 error_type 确定性派生进化目标(tool/skill/system)。 这是报告用的派生标注,非独立多样性轴(多样性主格子=task_type×error_type)。 参数: error_type: 诊断瀑布归因的错误类别(extraction/search/reasoning/mixed_failure)。 返回: 进化目标字符串 tool / skill / system。 异常: ValueError: error_type 不在已知集合内(不静默兜底)。 """ if error_type not in _EVOLUTION_TARGET: raise ValueError(f"未知 error_type: {error_type}") return _EVOLUTION_TARGET[error_type] def cell_of(task_type: str, error_type: str) -> tuple[str, str]: """构造多样性主格子 = (task_type, error_type)。 参数: task_type: 题型(12 类之一)。 error_type: 错误类别(4 类之一)。 返回: (task_type, error_type) 二元组,作为覆盖计数的格子键。 """ return (task_type, error_type) @dataclass(frozen=True) class SignalLabel: """诊断信号分层标签(DiagnosisResult 的确定性投影)。 字段: tier: 信号层级,取值 T0 / T1 / T2 / uncertain(判据见 score_signal)。 """ tier: str def score_signal(*, cause_category: str | None, infra: bool, degraded: bool) -> SignalLabel: """把诊断产物投影为信号分层 tier(不发明新分类,是确定性投影)。 分层优先级顺序固定(用早返回表达,不用魔法权重): 先判 INFRA,再判 degraded,然后 defect / lapse,最后兜底 uncertain。 各层判据来源: T0 — infra=True,即诊断 INFRA 排除(stop_reason ∈ {error, parse_error}), 基础设施失败先于一切判定,排除出可训练主体。 uncertain — degraded=True(judge 解析失败)或 cause_category 落不到 defect/lapse 上(如为 None),信号不可信,排除出 T2。 T2 — cause_category == "defect",可训练核心,进多样性覆盖与训练主体。 T1 — cause_category == "lapse",低信号(含无解题),接受但不作训练主体。 参数: cause_category: 诊断的缺陷归因("defect" / "lapse" / None)。 infra: 是否被 INFRA 护栏排除(基础设施失败)。 degraded: judge 是否解析失败导致诊断降级。 返回: SignalLabel,其 tier 字段为上述四层之一。 实现细节: 关键字参数强制传入,防止 infra / degraded 两个 bool 位置混淆。 """ if infra: return SignalLabel(tier="T0") if degraded: return SignalLabel(tier="uncertain") if cause_category == "defect": return SignalLabel(tier="T2") if cause_category == "lapse": return SignalLabel(tier="T1") return SignalLabel(tier="uncertain") @dataclass(frozen=True) class VideoRecord: """全视频画像单元(贪心选择器 Task 8 的输入单元)。 覆盖全部视频(含全对、零诊断信号的视频),既承载 test 代表性所需的难度/题型画像, 也叠加 T2 可训练缺陷的多样性格子,供选择器算覆盖与补集。 字段: video_id: 视频唯一标识。 type_set: 该视频所有题的 task_type 集合(去重,画像用)。 n_correct: 该视频答对题数。 difficulty: 难度画像桶 = 错题数 = 题数 - n_correct。 cells: 仅 tier=="T2" 信号行投影的 (task_type, error_type) 主格子并集(去重)。 wrong_by_type: 各 task_type 的 T2 计数,供选择器 floor 约束(普通 dict)。 实现细节: frozen 生成的 __hash__ 会遍历各字段;wrong_by_type 为不可哈希 dict, 故显式标注 hash=False 将其排除出哈希,避免 VideoRecord 入 set/dict 键时报错, 仍保留其参与相等性比较。 """ video_id: str type_set: frozenset[str] n_correct: int difficulty: int cells: frozenset[tuple[str, str]] wrong_by_type: dict[str, int] = field(hash=False) def build_video_records(preds: list[dict], signal_rows: list[dict]) -> list[VideoRecord]: """由全量 predictions 与诊断信号行构建全视频 VideoRecord 列表。 先按 video_id 聚合全部 predictions(覆盖全对、零信号视频),再叠加仅 tier=="T2" 的诊断信号为多样性格子与 wrong_by_type 计数。非 T2 信号行(T0/T1/uncertain) 不计入格子与计数。 参数: preds: 全量预测行,每行含 video_id / question_id / task_type / correct。 每视频含其全部题(不限于错题),correct 为布尔答对标记。 signal_rows: 诊断信号行,每行含 question_id / task_type / error_type / tier。 诊断只覆盖错题子集,正确题无对应信号行属正常,不视为错误。 返回: 全部视频的 VideoRecord 列表,按视频在 preds 中首次出现顺序排列。 无任何 T2 信号的视频其 cells 为空 frozenset、wrong_by_type 为空 dict。 实现细节: signal_rows 的 question_id 若不在 preds 中则忽略(诊断可能滞后于当前预测集, 非数据损坏),不 fail-fast;缺失必需键则按 KeyError 直接暴露(不静默兜底)。 异常: KeyError: preds 或 signal_rows 行缺少必需键(校验前置,防脏数据静默通过)。 """ # Phase 1: 按 video_id 聚合 preds(保持首次出现顺序)。 signal_by_qid = {row["question_id"]: row for row in signal_rows} aggregates: dict[str, dict] = {} for pred in preds: video_id = pred["video_id"] bucket = aggregates.setdefault( video_id, {"types": set(), "question_ids": [], "n_correct": 0} ) bucket["types"].add(pred["task_type"]) bucket["question_ids"].append(pred["question_id"]) if pred["correct"]: bucket["n_correct"] += 1 # Phase 2: 逐视频叠加 T2 信号为格子与 wrong_by_type。 records: list[VideoRecord] = [] for video_id, bucket in aggregates.items(): cells: set[tuple[str, str]] = set() wrong_by_type: dict[str, int] = {} for question_id in bucket["question_ids"]: row = signal_by_qid.get(question_id) if row is None or row["tier"] != "T2": continue task_type = row["task_type"] cells.add(cell_of(task_type, row["error_type"])) wrong_by_type[task_type] = wrong_by_type.get(task_type, 0) + 1 n_questions = len(bucket["question_ids"]) records.append( VideoRecord( video_id=video_id, type_set=frozenset(bucket["types"]), n_correct=bucket["n_correct"], difficulty=n_questions - bucket["n_correct"], cells=frozenset(cells), wrong_by_type=wrong_by_type, ) ) 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)