diff --git a/app/harness/pools.py b/app/harness/pools.py index 4e9b5b4..58859a1 100644 --- a/app/harness/pools.py +++ b/app/harness/pools.py @@ -214,16 +214,16 @@ def _sample_excluding( ) -> list[GeneratedQuestion]: """排除已选 unit 后，以 unit 为原子按 cfg 分层采样，返回展开后的逐题列表。 - 每个单元以其首题作为分层采样的代表参与 stratified_sample，correct_ratio / - size 因此按 unit 计数（pair 计 1 个 unit）；命中的单元整体展开，孪生对两题 - 永远同进同出。single-only 输入下 unit 与 question 一一对应、顺序不变，采样 - 结果与逐题采样完全一致。 + 候选单元展开为逐题列表后透传给 stratified_sample，后者内部重新 build_units + 做单元原子采样：correct_ratio / size 按 unit 计数（pair 计 1 个 unit），单元级 + 正确性由 stratified_sample 内部对成员取 AND，命中的孪生对两题永远同进同出。 + single-only 输入下 unit 与 question 一一对应、顺序不变，采样结果与逐题采样一致。 参数: units: 单元全集（single 单封、pair 成对聚合）。 exclude_unit_ids: 已被其他池选走的 unit_id，从候选中剔除以保证三池互斥。 - correctness: question_id -> 基线是否答对；单元级正确性取成员的 AND - （缺失按 False，与 stratified_sample 的宽松口径一致）。 + correctness: question_id -> 基线是否答对；单元级正确性由 stratified_sample + 对成员取 AND（缺失按 False，宽松口径）。 cfg: 透传给 stratified_sample 的采样配置 （size/correct_ratio/task_types[/seed/min_per_class]）。 @@ -231,12 +231,7 @@ def _sample_excluding( 采样命中单元展开后的题目列表。 """ candidates = [u for u in units if u.unit_id not in exclude_unit_ids] - rep_to_unit = {u.questions[0].question_id: u for u in candidates} - reps = [u.questions[0] for u in candidates] - unit_correct = {u.questions[0].question_id: _unit_correct(u, correctness) for u in candidates} - sampled_reps = stratified_sample(reps, unit_correct, **cfg) - sampled_units = [rep_to_unit[rep.question_id] for rep in sampled_reps] - return flatten_units(sampled_units) + return stratified_sample(flatten_units(candidates), correctness, **cfg) def _q_to_dict(q: GeneratedQuestion) -> dict: diff --git a/app/question_gen/loader.py b/app/question_gen/loader.py index 878bfef..bc95eb1 100644 --- a/app/question_gen/loader.py +++ b/app/question_gen/loader.py @@ -15,6 +15,8 @@ from core.types import GeneratedQuestion if TYPE_CHECKING: from pathlib import Path + from core.types import QuestionUnit + _LEGACY_DEFAULT_DIFFICULTY = "medium" @@ -26,6 +28,11 @@ def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]: video_id），v2 生成题把多视频题目合并在单个 JSON 中（每条记录自带 ``video_id``），两种格式均兼容。 + pair 契约字段（``pair_id`` / ``question_role`` / ``flip_axis`` / ``unit_id``） + 用 ``.get`` 读取：旧 benchmark 无这些键时退化为 single（``question_role`` + 默认 "single"，``unit_id`` 留空由 __post_init__ 回填为 question_id）， + 保证历史题库可无缝加载。 + 参数: questions_dir: 包含 *.json 文件的目录路径。 @@ -52,6 +59,12 @@ def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]: skill_target=qa.get("skill_target"), difficulty_steps=qa.get("difficulty_steps"), sub_pattern=qa.get("sub_pattern"), + # pair 契约字段：旧 benchmark 无这些键时按 single 默认兜底， + # unit_id 留空交由 GeneratedQuestion.__post_init__ 回填。 + pair_id=qa.get("pair_id"), + question_role=qa.get("question_role", "single"), + flip_axis=qa.get("flip_axis"), + unit_id=qa.get("unit_id", ""), ) ) return results @@ -66,62 +79,88 @@ def stratified_sample( seed: int, min_per_class: int | None, ) -> list[GeneratedQuestion]: - """按题型过滤后采样 size 道题，可选按对错比例分层并按题型保底。 + """按题型过滤后采样 size 个单元，可选按对错比例分层并按题型保底。 参数: - questions: 候选题目全集。 - correctness: question_id -> 基线是否答对。 - size: 采样总量。 - correct_ratio: 采样中"基线答对"题的占比；None 表示自然分布。 + questions: 候选题目全集（single 与孪生对成员可混含）。 + correctness: question_id -> 基线是否答对（单元级正确性取成员 AND）。 + size: 采样单元总量（single 计 1、pair 计 1）。 + correct_ratio: 采样中"基线答对"单元的占比；None 表示自然分布。 task_types: 限定题型；None 表示不限。 seed: 随机种子，保证可复现。 - min_per_class: 每个题型补足到的下限；None 表示不补足。 + min_per_class: 每个题型补足到的单元下限；None 表示不补足。 返回: - 采样后的题目列表。 + 采样后的题目列表（pair 单元展开为原始的两道题）。 异常: - ValueError: 自然分布时池不足 size，或分层时某层题目不足。 + ValueError: 自然分布时单元池不足 size，或分层时某层单元不足。 + + 关键实现: + 以 **QuestionUnit 为采样原子**（single 计 1、pair 计 1），size / + correct_ratio / min_per_class 均按 unit 计数，孪生对两题永不被劈开。 + 采样完成后 flatten_units 展开回逐题列表。纯 single 输入时 build_units + 与题目一一对应、顺序不变，rng 消耗与旧逐题实现完全一致（字节级回归）。 + + build_units / flatten_units 采用函数内延迟导入：loader 属 question_gen， + question_units 属 harness，而 harness 包初始化会反向 import question_gen， + 模块级导入将触发循环依赖（沿用 adversarial_filter 的既有做法）。 """ + from app.harness.question_units import build_units, flatten_units + rng = random.Random(seed) - pool = [q for q in questions if task_types is None or q.task_type in task_types] + units = build_units(questions) + pool = [u for u in units if task_types is None or u.task_type in task_types] if correct_ratio is None: if len(pool) < size: - raise ValueError(f"自然分布采样不足: 需 {size} 道, 实有 {len(pool)} 道") + raise ValueError(f"自然分布采样不足: 需 {size} 个单元, 实有 {len(pool)} 个") sampled = rng.sample(pool, size) else: sampled = _ratio_stratified_sample(pool, correctness, size, correct_ratio, rng) if min_per_class is not None: sampled = _backfill_per_class(sampled, pool, min_per_class, rng) - return sampled + return flatten_units(sampled) + + +def _unit_correct(unit: QuestionUnit, correctness: dict[str, bool]) -> bool: + """单元级正确性：成员全部答对才算对（缺失按 False，宽松口径）。 + + 参数: + unit: 目标单元（single 1 题，pair 2 题）。 + correctness: question_id -> 基线是否答对。 + + 返回: + pair 走双向 AND、single 即单题正确性；任一成员缺失或答错即 False。 + """ + return all(correctness.get(q.question_id, False) for q in unit.questions) def _ratio_stratified_sample( - pool: list[GeneratedQuestion], + pool: list[QuestionUnit], correctness: dict[str, bool], size: int, correct_ratio: float, rng: random.Random, -) -> list[GeneratedQuestion]: - """按对错比例分层采样：对题占 correct_ratio，其余为错题。 +) -> list[QuestionUnit]: + """按对错比例分层采样：对单元占 correct_ratio，其余为错单元。 参数: - pool: 题型过滤后的候选题。 + pool: 题型过滤后的候选单元。 correctness: question_id -> 基线是否答对。 - size: 采样总量。 - correct_ratio: 对题占比。 + size: 采样单元总量。 + correct_ratio: 对单元占比。 rng: 随机数发生器。 返回: - 采样后的题目列表（对题在前、错题在后）。 + 采样后的单元列表（对单元在前、错单元在后）。 异常: - ValueError: 对题或错题层不足。 + ValueError: 对单元或错单元层不足。 """ - correct = [q for q in pool if correctness.get(q.question_id, False)] - wrong = [q for q in pool if not correctness.get(q.question_id, False)] + correct = [u for u in pool if _unit_correct(u, correctness)] + wrong = [u for u in pool if not _unit_correct(u, correctness)] n_correct = round(size * correct_ratio) n_wrong = size - n_correct if len(correct) < n_correct or len(wrong) < n_wrong: @@ -132,42 +171,40 @@ def _ratio_stratified_sample( def _backfill_per_class( - sampled: list[GeneratedQuestion], - pool: list[GeneratedQuestion], + sampled: list[QuestionUnit], + pool: list[QuestionUnit], min_per_class: int, rng: random.Random, -) -> list[GeneratedQuestion]: - """对候选池中出现的每个题型，将采样结果补足到 min_per_class 道。 +) -> list[QuestionUnit]: + """对候选池中出现的每个题型，将采样单元补足到 min_per_class 个。 遍历对象是候选池 pool 里出现的全部题型（非仅 sampled 命中的）， - 保证任意稀疏题型都能拿到足额样本。 + 保证任意稀疏题型都能拿到足额样本。补足以 unit 为原子，孪生对整进整出。 参数: - sampled: 主采样结果（不修改，返回新列表）。 - pool: 候选题全集（补足来源 + 题型枚举来源）。 - min_per_class: 每个题型的下限。 + sampled: 主采样结果单元（不修改，返回新列表）。 + pool: 候选单元全集（补足来源 + 题型枚举来源）。 + min_per_class: 每个题型的单元下限。 rng: 随机数发生器。 返回: - 补足后的题目列表。 + 补足后的单元列表。 """ - selected_ids = {q.question_id for q in sampled} + selected_ids = {u.unit_id for u in sampled} result = list(sampled) counts: dict[str, int] = {} - for q in sampled: - counts[q.task_type] = counts.get(q.task_type, 0) + 1 + for u in sampled: + counts[u.task_type] = counts.get(u.task_type, 0) + 1 ordered_task_types: dict[str, None] = {} - for q in pool: - ordered_task_types.setdefault(q.task_type, None) + for u in pool: + ordered_task_types.setdefault(u.task_type, None) for task_type in ordered_task_types: deficit = min_per_class - counts.get(task_type, 0) if deficit <= 0: continue - candidates = [ - q for q in pool if q.task_type == task_type and q.question_id not in selected_ids - ] + candidates = [u for u in pool if u.task_type == task_type and u.unit_id not in selected_ids] take = rng.sample(candidates, min(deficit, len(candidates))) - for q in take: - selected_ids.add(q.question_id) - result.append(q) + for u in take: + selected_ids.add(u.unit_id) + result.append(u) return result diff --git a/tests/unit/test_loader_unit_sampling.py b/tests/unit/test_loader_unit_sampling.py new file mode 100644 index 0000000..c8dc738 --- /dev/null +++ b/tests/unit/test_loader_unit_sampling.py @@ -0,0 +1,297 @@ +"""loader.stratified_sample 按 unit 采样 + load_benchmark 读回 pair 字段的单元测试。 + +覆盖 question-gen v3 Phase 1 Task 4 的四项契约： + (a) pair 采样同进同出——同一 pair_id 两题要么都被选、要么都不被选； + (b) size / correct_ratio 按 **unit 计数**（pair 计 1 个 unit，unit 正确性走双向 AND）； + (c) min_per_class 补足路径不拆 pair； + (d) load_benchmark 反序列化读回 pair_id/question_role/flip_axis/unit_id， + 且旧 JSON 缺这些字段时按 single 默认兜底、不崩。 +另加一条纯 single 输入的字节级回归守护：单题场景采样序列必须与旧逐题行为一致。 +""" + +from __future__ import annotations + +import json +import random +from pathlib import Path + +import pytest + +from app.harness.question_units import build_units +from app.question_gen.loader import load_benchmark, stratified_sample +from core.types import GeneratedQuestion + + +def _single(qid: str, task_type: str = "Single") -> GeneratedQuestion: + """构造一道 single 题（无 pair 归属）。""" + return GeneratedQuestion( + question_id=qid, + video_id="v", + task_type=task_type, + question="?", + options=("A", "B", "C", "D"), + answer="A", + source_nodes=(), + difficulty="medium", + ) + + +def _pair(pid: str, task_type: str = "AR") -> list[GeneratedQuestion]: + """构造一个合法孪生对（original + mirror），共享 pair_id / unit_id / flip_axis。""" + base = dict( + video_id="v", + task_type=task_type, + question="?", + options=("A", "B", "C", "D"), + answer="A", + source_nodes=(), + difficulty="hard", + pair_id=pid, + unit_id=pid, + flip_axis="before_after", + ) + return [ + GeneratedQuestion(question_id=f"{pid}_o", question_role="pair_original", **base), + GeneratedQuestion(question_id=f"{pid}_m", question_role="pair_mirror", **base), + ] + + +def _pair_roles_by_id(questions: list[GeneratedQuestion]) -> dict[str, set[str]]: + """将采样结果按 pair_id 聚合出现的角色集合（仅统计配对题）。""" + loc: dict[str, set[str]] = {} + for q in questions: + if q.pair_id: + loc.setdefault(q.pair_id, set()).add(q.question_role) + return loc + + +class TestPairAtomicSampling: + def test_pair_never_split_in_natural_sample(self) -> None: + """自然分布采样：命中的 pair 必两题齐全，size 按 unit 计数。""" + qs = [q for i in range(8) for q in _pair(f"p{i}")] + result = stratified_sample( + questions=qs, + correctness={}, + size=4, + correct_ratio=None, + task_types=None, + seed=7, + min_per_class=None, + ) + loc = _pair_roles_by_id(result) + assert len(loc) == 4, "size=4 应命中 4 个 pair 单元" + for pid, roles in loc.items(): + assert roles == {"pair_original", "pair_mirror"}, f"pair {pid} 被拆: {roles}" + assert len(result) == 8, "4 个 pair 单元展开应为 8 道题" + + def test_size_counts_units_not_questions(self) -> None: + """混合 single + pair 时 size 仍按 unit 计数（pair 计 1）。""" + qs = [_single(f"s{i}") for i in range(3)] + qs += [q for i in range(3) for q in _pair(f"p{i}")] + result = stratified_sample( + questions=qs, + correctness={}, + size=6, # 全部 6 个单元（3 single + 3 pair） + correct_ratio=None, + task_types=None, + seed=1, + min_per_class=None, + ) + singles = [q for q in result if not q.pair_id] + loc = _pair_roles_by_id(result) + assert len(singles) == 3 + assert len(loc) == 3 + for roles in loc.values(): + assert roles == {"pair_original", "pair_mirror"} + assert len(result) == 3 + 3 * 2 + + def test_ratio_counts_units(self) -> None: + """correct_ratio 按 unit 计数：对/错单元按比例各取整数个。""" + correct_pairs = [_pair(f"c{i}") for i in range(4)] + wrong_pairs = [_pair(f"w{i}") for i in range(4)] + qs = [q for p in correct_pairs + wrong_pairs for q in p] + correctness: dict[str, bool] = {} + for p in correct_pairs: + for q in p: + correctness[q.question_id] = True + result = stratified_sample( + questions=qs, + correctness=correctness, + size=4, + correct_ratio=0.5, + task_types=None, + seed=3, + min_per_class=None, + ) + loc = _pair_roles_by_id(result) + assert len(loc) == 4, "4 个 unit" + for roles in loc.values(): + assert roles == {"pair_original", "pair_mirror"} + correct_units = sum( + 1 for pid in loc if all(correctness.get(f"{pid}_{s}", False) for s in ("o", "m")) + ) + assert correct_units == 2, "correct_ratio=0.5 * 4 unit = 2 个对单元" + + def test_unit_correctness_uses_and_not_any(self) -> None: + """unit 正确性走双向 AND：混合对（P 对 Q 错）不得计入对单元层。 + + 构造 2 个全对 pair + 1 个混合 pair（original 对、mirror 错）。请求 3 个对单元， + 若按 AND，对池仅 2 个 → 分层不足报错；若错误地按 any-member，对池 3 个 → 不报错。 + 以是否抛错来判别语义，规避随机命中导致的假阳性。 + """ + good = [_pair("g0"), _pair("g1")] + mixed = _pair("mx") + qs = [q for p in good for q in p] + mixed + correctness: dict[str, bool] = {} + for p in good: + for q in p: + correctness[q.question_id] = True + correctness[mixed[0].question_id] = True + correctness[mixed[1].question_id] = False + with pytest.raises(ValueError, match="分层不足"): + stratified_sample( + questions=qs, + correctness=correctness, + size=3, + correct_ratio=1.0, + task_types=None, + seed=1, + min_per_class=None, + ) + + +class TestBackfillPairAtomic: + def test_backfill_keeps_pairs_atomic(self) -> None: + """min_per_class 补足稀疏 pair 题型时，补入的 pair 两题齐全不拆。""" + ar = [q for i in range(3) for q in _pair(f"a{i}", task_type="AR")] + main = [_single(f"m{i}", task_type="Main") for i in range(10)] + result = stratified_sample( + questions=ar + main, + correctness={}, + size=2, + correct_ratio=None, + task_types=None, + seed=5, + min_per_class=2, + ) + loc = _pair_roles_by_id(result) + assert len(loc) >= 2, "AR 至少被补足到 2 个 pair 单元" + for pid, roles in loc.items(): + assert roles == {"pair_original", "pair_mirror"}, f"补足拆散了 pair {pid}: {roles}" + + def test_backfill_no_duplicate_units(self) -> None: + """补足不得重复选入同一 pair 单元。""" + ar = [q for i in range(4) for q in _pair(f"a{i}", task_type="AR")] + result = stratified_sample( + questions=ar, + correctness={}, + size=1, + correct_ratio=None, + task_types=None, + seed=9, + min_per_class=3, + ) + ids = [q.question_id for q in result] + assert len(ids) == len(set(ids)), "存在重复题目" + + +class TestLoadBenchmarkPairFields: + def test_reads_pair_fields(self, tmp_path: Path) -> None: + """新格式 JSON 的 pair_id/question_role/flip_axis/unit_id 被正确读回。""" + data = [ + { + "question_id": "pp_o", + "video_id": "v", + "task_type": "AR", + "question": "?", + "options": ["A", "B", "C", "D"], + "answer": "A", + "pair_id": "pp", + "question_role": "pair_original", + "flip_axis": "before_after", + "unit_id": "pp", + }, + { + "question_id": "pp_m", + "video_id": "v", + "task_type": "AR", + "question": "?", + "options": ["A", "B", "C", "D"], + "answer": "A", + "pair_id": "pp", + "question_role": "pair_mirror", + "flip_axis": "before_after", + "unit_id": "pp", + }, + ] + (tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8") + qs = load_benchmark(tmp_path) + q0 = next(q for q in qs if q.question_id == "pp_o") + assert q0.pair_id == "pp" + assert q0.question_role == "pair_original" + assert q0.flip_axis == "before_after" + assert q0.unit_id == "pp" + # 读回后应能被 build_units 重新聚合成 1 个 pair 单元 + units = build_units(qs) + assert len(units) == 1 + assert units[0].kind == "pair" + + def test_legacy_json_defaults_to_single(self, tmp_path: Path) -> None: + """旧 JSON 缺 pair 字段时按 single 兜底、unit_id 回填为 question_id，不崩。""" + data = [ + { + "question_id": "x1", + "video_id": "v", + "task_type": "T", + "question": "?", + "options": ["A", "B", "C", "D"], + "answer": "A", + } + ] + (tmp_path / "legacy.json").write_text(json.dumps(data), encoding="utf-8") + qs = load_benchmark(tmp_path) + assert qs[0].pair_id is None + assert qs[0].question_role == "single" + assert qs[0].flip_axis is None + assert qs[0].unit_id == "x1" + units = build_units(qs) + assert len(units) == 1 + assert units[0].kind == "single" + + +class TestPureSingleByteIdentical: + def test_natural_matches_legacy_rng(self) -> None: + """纯 single 输入的自然分布采样，与旧逐题 rng.sample 序列字节级一致。""" + qs = [_single(f"s{i}") for i in range(20)] + seed = 123 + result = stratified_sample( + questions=qs, + correctness={}, + size=8, + correct_ratio=None, + task_types=None, + seed=seed, + min_per_class=None, + ) + reference = random.Random(seed).sample(qs, 8) + assert [q.question_id for q in result] == [q.question_id for q in reference] + + def test_ratio_matches_legacy_rng(self) -> None: + """纯 single 输入的比例分层采样，与旧逐题实现的抽样序列一致。""" + qs = [_single(f"s{i}") for i in range(20)] + correctness = {f"s{i}": i < 10 for i in range(20)} + seed = 77 + result = stratified_sample( + questions=qs, + correctness=correctness, + size=10, + correct_ratio=0.6, + task_types=None, + seed=seed, + min_per_class=None, + ) + rng = random.Random(seed) + correct = [q for q in qs if correctness.get(q.question_id, False)] + wrong = [q for q in qs if not correctness.get(q.question_id, False)] + reference = rng.sample(correct, 6) + rng.sample(wrong, 4) + assert [q.question_id for q in result] == [q.question_id for q in reference]