feat: add video-split config knobs and reproducible script

This commit is contained in:
2026-07-15 12:54:25 -04:00
parent 43d7346526
commit 6a21d80313
6 changed files with 351 additions and 0 deletions
+115
View File
@@ -0,0 +1,115 @@
"""视频级切分科研旋钮单元测试:诊断指纹 + val_wrong_min 功效护栏 + PoolConfig 新字段。
覆盖:
- diag_fingerprint 对 (prompt 版本 / 模型 / 代码版本) 三元组确定且敏感;
- split_by_video_assignment 的 val_wrong_min 门控 fail loud(验证信号不足即报错);
- val_wrong_min 默认 0 时行为与 Task 11 现有调用完全一致(不回归);
- PoolConfig 能接收视频级切分的五个新旋钮字段(纯 dataclass 装配)。
"""
from __future__ import annotations
import pytest
from app.harness.pools import InsufficientValSignal, split_by_video_assignment
from app.harness.split_selection import diag_fingerprint
from core.types import GeneratedQuestion, PoolConfig
def _q(qid: str, vid: str, tt: str = "Counting Problem") -> GeneratedQuestion:
"""构造最小可用题目(补齐 GeneratedQuestion 的必填 source_nodes/difficulty)。"""
return GeneratedQuestion(
question_id=qid,
video_id=vid,
task_type=tt,
question="",
options=("A", "B", "C", "D"),
answer="A",
source_nodes=(),
difficulty="easy",
)
def test_diag_fingerprint_deterministic_and_sensitive():
"""诊断指纹对相同三元组稳定,对任一分量变化敏感。"""
a = diag_fingerprint("p1", "deepseek-v4", "abc123")
assert a == diag_fingerprint("p1", "deepseek-v4", "abc123") # 确定性
assert a != diag_fingerprint("p2", "deepseek-v4", "abc123") # prompt 变则变
assert a != diag_fingerprint("p1", "kimi", "abc123") # model 变则变
assert a != diag_fingerprint("p1", "deepseek-v4", "def456") # 代码版本变则变
assert len(a) == 16 # sha256 截断 16 位十六进制
def test_val_wrong_min_enforced():
"""val 错题数 < val_wrong_min 时 fail loudInsufficientValSignal),不静默兜底。"""
qs = [_q("v1-1", "v1"), _q("v1-2", "v1"), _q("v1-3", "v1")]
correctness = {q.question_id: True for q in qs} # 全对 → val 无错题
with pytest.raises(InsufficientValSignal): # val 错题 < val_wrong_min
split_by_video_assignment(
qs,
{"v1": "trainval"},
correctness=correctness,
val_ratio=0.5,
seed=0,
val_wrong_min=5,
)
def test_val_wrong_min_default_zero_no_regression():
"""val_wrong_min 默认 0 时不检查错题数,保持 Task 11 现有调用契约不破。"""
qs = [_q("v1-1", "v1"), _q("v2-1", "v2")]
assignment = {"v1": "trainval", "v2": "trainval"}
correctness = {"v1-1": True, "v2-1": True} # 全对但默认不触发护栏
pools = split_by_video_assignment(
qs, assignment, correctness=correctness, val_ratio=1.0, seed=0
)
assert len(pools.validation) == 2 # 未抛异常,正常返回
def test_pool_config_accepts_video_split_knobs():
"""PoolConfig 能接收视频级切分五个新旋钮字段(默认惰性,不破坏现有构造点)。"""
cfg = PoolConfig(
task_types=None,
seed=0,
baseline_run_id="infer_adhoc",
diag_size=200,
diag_correct_ratio=0.5,
val_size=30,
val_correct_ratio=0.5,
test_size=60,
eval_min_per_class=2,
train_ratio=0.667,
test_questions_dir=None,
n_trainval=100,
floor_k={"Counting Problem": 3},
epsilon=0.1,
report_floor=27,
val_wrong_min=20,
)
assert cfg.n_trainval == 100
assert cfg.floor_k == {"Counting Problem": 3}
assert cfg.epsilon == 0.1
assert cfg.report_floor == 27
assert cfg.val_wrong_min == 20
def test_pool_config_video_split_knobs_default_inert():
"""未传视频级切分字段时默认惰性(0 / 空 dict),不破坏 GlobalPoolStrategy 现有构造。"""
cfg = PoolConfig(
task_types=None,
seed=0,
baseline_run_id="run_1",
diag_size=200,
diag_correct_ratio=0.5,
val_size=30,
val_correct_ratio=0.5,
test_size=60,
eval_min_per_class=2,
train_ratio=0.667,
test_questions_dir=None,
)
assert cfg.n_trainval == 0
assert cfg.floor_k == {}
assert cfg.epsilon == 0.0
assert cfg.report_floor == 0
assert cfg.val_wrong_min == 0