552 lines
22 KiB
Python
552 lines
22 KiB
Python
"""视频级切分选择：signal 分层、视频聚合、贪心联合约束选择（纯函数）。
|
||
|
||
结果驱动切分管线的核心：把诊断信号投影为多样性格子，供贪心选择器最大化覆盖。
|
||
本模块起步定义 evolution_target 派生与多样性格子；后续追加 score_signal /
|
||
build_video_records / select_split。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import random
|
||
from dataclasses import dataclass, field
|
||
|
||
from loguru import logger
|
||
|
||
|
||
def diag_fingerprint(prompt_version: str, model: str, code_version: str) -> str:
|
||
"""由 (诊断 prompt 版本, 模型名, 代码 git 短 SHA) 合成诊断口径指纹。
|
||
|
||
诊断信号以 (question_id, baseline_run_id, diag_fingerprint) 为主键持久化，
|
||
指纹隔离不同诊断配置的信号——换 prompt 版本 / 换模型 / 换代码实现都会得到
|
||
新指纹，从而 `--force` 用新指纹重跑诊断时**不覆盖旧记录**（旧指纹行仍在），
|
||
保证不同口径的诊断结果可并存、可回溯、可比对。
|
||
|
||
参数:
|
||
prompt_version: 诊断 prompt 的版本标识（如 prompts/diagnose_*.md 的版本）。
|
||
model: 执行诊断的模型名（如 "deepseek-v4"）。
|
||
code_version: 诊断代码的版本（约定为 git 短 SHA）。
|
||
|
||
返回:
|
||
16 位十六进制指纹（sha256 截断），对三分量任一变化敏感、对相同三元组确定。
|
||
|
||
实现细节:
|
||
三分量用 "|" 分隔后 sha256，取前 16 位；分隔符防止 ("ab","c") 与 ("a","bc")
|
||
碰撞成同一指纹。纯函数，相同输入永远同输出，可安全用于主键。
|
||
"""
|
||
return hashlib.sha256("|".join([prompt_version, model, code_version]).encode()).hexdigest()[:16]
|
||
|
||
|
||
_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；否则选填槽最多者（等槽数按预洗牌顺序取首个，确定性）。
|
||
|
||
n_trainval 是硬预算：floor 需求超出预算（尚有缺口却已达 n_trainval）也判不可行 fail loud，
|
||
保证返回的 trainval 永不超过 n_trainval（不因硬约束悄悄超额、挤占 test）。
|
||
|
||
参数:
|
||
selected: 当前 trainval（就地追加）。
|
||
pool: 当前剩余池 = test 补集（就地移除）。
|
||
videos_all: 全部视频（ε 全局基准）。
|
||
config: 选择配置。
|
||
|
||
异常:
|
||
InfeasibleSplitError: 存在未达标类型但无候选可在不破 ε 下填补，
|
||
或 floor 需求超过 n_trainval 预算。
|
||
"""
|
||
while True:
|
||
deficits = _unmet_floors(selected, config.floor_k)
|
||
if not deficits:
|
||
return
|
||
if len(selected) >= config.n_trainval:
|
||
raise InfeasibleSplitError(
|
||
f"floor 需求超过 n_trainval={config.n_trainval} 预算，剩余缺口: {dict(deficits)}"
|
||
)
|
||
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 与 ε 死锁无法满足、
|
||
或 floor 需求超过 n_trainval 预算。
|
||
"""
|
||
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)
|