iomgaa
8fef7ced42
C-1: persist_infra_t0_rows 补 INFRA/空预测错题的 T0 信号行（不进诊断故须单独落库），run_pipeline 加 Phase 0，dry-run 用假数据走通。 C-2: CLI 加 --generated-at，真实运行默认盖真实 UTC now，可显式固定以字节级复现 manifest。 I-1: coverage_report 增 evolution_target_distribution（T2 信号按 tool/skill/system 计数）。 I-2: 删除 PoolConfig 死字段 n_trainval/floor_k/epsilon/report_floor/val_wrong_min（grep 确认无消费者，视频级切分用独立 VideoSplitConfig/SplitBuildConfig/SelectConfig）。 I-3: 抽共享 load_canonical_predictions(db_path, run_id)，CLI 与 build_split 共用；消除 canonical 取行 + correct 判定重复。 M-1: build_split docstring 注明 val_wrong_min-agnostic 契约（McNemar 护栏由 CLI 冻结后执行，Task 11 契约）。
553 lines
21 KiB
Python
553 lines
21 KiB
Python
"""结果驱动视频级切分的顶层编排：诊断信号 → 冻结 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 校验。
|
||
|
||
契约（Task 11，非疏漏）：build_split 有意保持 val_wrong_min-agnostic——内部调
|
||
split_by_video_assignment 时不传 val_wrong_min（默认 0，不校验 validation 错题
|
||
数）。McNemar 功效护栏是切分**冻结后**的独立校验，由 CLI 的 check_mcnemar_power
|
||
在 build_split 返回后执行；切分构造本身不因功效阈失败，二者关注点分离。
|
||
|
||
参数:
|
||
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 = load_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 load_canonical_predictions(db_path: Path, baseline_run_id: str) -> list[dict]:
|
||
"""从 harness.db 只读取指定 run 每题首行（ORDER BY rowid）为 canonical 预测。
|
||
|
||
共享口径 helper：CLI（可诊断错题筛选 + INFRA T0 补录）与 build_split（切分）
|
||
共用同一"每 qid 取 rowid 最小首行 + 归一化 correct 判定"口径，消除两处重复实现。
|
||
同一 question_id 可能有多行（重跑 / 补测），canonical 口径取 rowid 最小的首行，
|
||
保证 distinct question 计数与对错判定确定。correct = 预测与答案各自归一
|
||
（strip → 大写 → 取首字母）后逐字符相等。
|
||
|
||
口径边界：旧 build_or_load_pools 的 legacy 池构建路径（app/harness/pools.py）是
|
||
另一条独立既有链路，不共用本 helper，两者刻意不统一（本次不动 legacy 路径）。
|
||
|
||
参数:
|
||
db_path: harness.db 路径（URI mode=ro 只读打开，绝不改动基线 db）。
|
||
baseline_run_id: 基线 run 标识。
|
||
|
||
返回:
|
||
canonical 预测行列表，每行含 question_id / video_id / task_type /
|
||
prediction / answer / stop_reason / correct（bool）。按 rowid 升序去重，
|
||
每 qid 保留首行。
|
||
|
||
异常:
|
||
ValueError: 该 run 无任何预测行（fail-fast，不返回空切分）。
|
||
|
||
实现细节:
|
||
按 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, stop_reason "
|
||
"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"],
|
||
"prediction": row["prediction"],
|
||
"answer": row["answer"],
|
||
"stop_reason": row["stop_reason"],
|
||
"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 /
|
||
evolution_target_distribution。
|
||
|
||
实现细节:
|
||
evolution_target_distribution 只统计 T2 信号（可训练缺陷），按
|
||
tool / skill / system 计数，报告"哪层参数组拿到梯度"；T0/T1/uncertain 行
|
||
evolution_target 恒为 None，不入该分布。
|
||
"""
|
||
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_distribution, evolution_target_distribution = _signal_distributions(signal_rows_raw)
|
||
|
||
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,
|
||
"evolution_target_distribution": evolution_target_distribution,
|
||
}
|
||
|
||
|
||
def _signal_distributions(
|
||
signal_rows_raw: list[DiagnosisSignalRow],
|
||
) -> tuple[dict[str, float], dict[str, int]]:
|
||
"""由诊断信号行算 tier 占比分布与 T2 进化目标计数分布。
|
||
|
||
参数:
|
||
signal_rows_raw: 诊断信号行。
|
||
|
||
返回:
|
||
(tier_distribution, evolution_target_distribution) 二元组：
|
||
- tier_distribution: {tier: 占比}，无信号时为空 dict；
|
||
- evolution_target_distribution: 仅统计 T2（可训练缺陷）信号，按
|
||
tool / skill / system 计数，报告哪层参数组拿到梯度；T0/T1/uncertain 行
|
||
evolution_target 恒为 None，不入该分布。
|
||
"""
|
||
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 {}
|
||
)
|
||
evolution_target_distribution = dict(
|
||
Counter(
|
||
row.evolution_target
|
||
for row in signal_rows_raw
|
||
if row.tier == "T2" and row.evolution_target is not None
|
||
)
|
||
)
|
||
return tier_distribution, evolution_target_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}"
|
||
)
|