189 lines
7.5 KiB
Python
189 lines
7.5 KiB
Python
"""视频级切分选择：signal 分层、视频聚合、贪心联合约束选择（纯函数）。
|
||
|
||
结果驱动切分管线的核心：把诊断信号投影为多样性格子，供贪心选择器最大化覆盖。
|
||
本模块起步定义 evolution_target 派生与多样性格子；后续追加 score_signal /
|
||
build_video_records / select_split。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
|
||
_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
|