iomgaa
4b6d1d8a50
进化引擎与 gate e-process 从 question_id 口径迁至 unit_id 口径，AR pair 双向 AND 折叠为单元、不被 P/Q 单题计分污染；逐题 predictions 仅作溯源。 - question_units: 新增 unit_correctness_view(units, per_q)->dict[unit_id,bool] 作为逐题→单元折叠的唯一入口（复用 unit_correctness）。 - core/evolution/validate: pair_block/compute_accuracy 参数改 unit_ids、 分母按单元数（键即 unit_id）。 - app/harness/validate（gate 实际执行路径）：阶梯题序聚合为单元并保持信息 阶梯序（_ladder_units），gate 块按单元切分（AR pair 整锁不跨块拆）； baseline_cache 键含 unit_id、存单元级对错；候选逐题读回后折叠成单元视图； n_used/W/L/四象限/准确率均按单元计；证据行按 unit 口径，candidate_correctness 独立保留逐题对错供 runner 二轨合并。 - runner: probation 结算按 unit 折叠计 W/L（_probation_unit_flips）；quadrant 四象限 id 承载 unit_id。 核心算法保真 #5（信息阶梯 e-process）：本次仅迁移 correctness 口径，不改冷启动 2:1 / gamma-EMA / 反泄漏算法本身（gate_ladder 迁移见 Task 8）。
136 lines
5.2 KiB
Python
136 lines
5.2 KiB
Python
"""QuestionUnit 组装/展开/校验/单元正确性——pair 契约的唯一入口。
|
||
|
||
pool 构建、批处理、推理、评测（Task 3+）均通过本模块聚合/展开孪生对，
|
||
保证 AR pair 的"两题作为整体调度"契约只在一处实现、fail-fast 暴露非法配对。
|
||
"""
|
||
|
||
from collections import defaultdict
|
||
|
||
from core.types import GeneratedQuestion, QuestionUnit
|
||
|
||
|
||
def _assemble_pair(pair_id: str, group: list[GeneratedQuestion]) -> QuestionUnit:
|
||
"""校验单个 pair 分组的数量/角色并组装为 pair 单元（fail-fast）。
|
||
|
||
参数:
|
||
pair_id: 该分组共享的孪生对标识。
|
||
group: 归属同一 pair_id 的题目列表。
|
||
|
||
返回:
|
||
kind="pair" 的 QuestionUnit。
|
||
|
||
关键实现:
|
||
- 分组必须恰好 2 条，否则视为孤儿/超员，raise ValueError。
|
||
- 显式检查 pair_original / pair_mirror 角色齐备且唯一，缺失或重复
|
||
直接 raise ValueError（防 next(...) 静默 StopIteration）。
|
||
- 合法孪生对交由 QuestionUnit.from_pair 做 video_id/task_type/flip_axis
|
||
一致性断言。
|
||
"""
|
||
if len(group) != 2:
|
||
raise ValueError(f"pair {pair_id} 数量={len(group)}≠2（孤儿或超员）")
|
||
originals = [q for q in group if q.question_role == "pair_original"]
|
||
mirrors = [q for q in group if q.question_role == "pair_mirror"]
|
||
if len(originals) != 1 or len(mirrors) != 1:
|
||
raise ValueError(
|
||
f"pair {pair_id} 角色非法：original={len(originals)} mirror={len(mirrors)}，"
|
||
"需各恰好 1 条"
|
||
)
|
||
return QuestionUnit.from_pair(originals[0], mirrors[0])
|
||
|
||
|
||
def build_units(questions: list[GeneratedQuestion]) -> list[QuestionUnit]:
|
||
"""将扁平题目列表聚合为单元列表：single 单封、pair 按 pair_id 成对聚合。
|
||
|
||
参数:
|
||
questions: 待聚合的题目列表，可混含 single 与孪生对成员。
|
||
|
||
返回:
|
||
单元列表，先 single 后 pair，顺序稳定（single 保留输入顺序，
|
||
pair 按首次出现的 pair_id 顺序）。
|
||
|
||
关键实现:
|
||
pair_id 为空 → single 单元；非空 → 归入对应 pair 桶。各 pair 桶的
|
||
数量/角色校验与组装下沉到 _assemble_pair（fail-fast），本体只做分组
|
||
与派发。
|
||
"""
|
||
by_pair: dict[str, list[GeneratedQuestion]] = defaultdict(list)
|
||
singles: list[QuestionUnit] = []
|
||
for q in questions:
|
||
if q.pair_id:
|
||
by_pair[q.pair_id].append(q)
|
||
else:
|
||
singles.append(QuestionUnit.from_single(q))
|
||
|
||
pairs = [_assemble_pair(pid, qs) for pid, qs in by_pair.items()]
|
||
return singles + pairs
|
||
|
||
|
||
def flatten_units(units: list[QuestionUnit]) -> list[GeneratedQuestion]:
|
||
"""将单元列表无损展开回扁平题目列表。
|
||
|
||
参数:
|
||
units: 单元列表。
|
||
|
||
返回:
|
||
展开后的题目列表，保持单元顺序及单元内题目顺序。
|
||
"""
|
||
return [q for u in units for q in u.questions]
|
||
|
||
|
||
def validate_units(units: list[QuestionUnit]) -> list[QuestionUnit]:
|
||
"""校验单元列表结构合法性，通过则原样返回（便于链式调用）。
|
||
|
||
参数:
|
||
units: 待校验单元列表。
|
||
|
||
返回:
|
||
校验通过的原单元列表。
|
||
|
||
关键实现:
|
||
pair 单元必须恰好含 2 题，否则 raise ValueError；single 单元无需额外
|
||
校验（构造时即为 1 题）。用于消费方在使用前做一道防御闸门。
|
||
"""
|
||
for u in units:
|
||
if u.kind == "pair" and u.size != 2:
|
||
raise ValueError(f"unit {u.unit_id} pair 不成对（size={u.size}）")
|
||
return units
|
||
|
||
|
||
def unit_correctness(unit: QuestionUnit, per_q: dict[str, bool]) -> bool:
|
||
"""计算单元级正确性：AR pair 走双向 AND，single 即单题正确性。
|
||
|
||
参数:
|
||
unit: 目标单元。
|
||
per_q: 题目 question_id → 该题是否作答正确的映射。
|
||
|
||
返回:
|
||
单元内所有题目均正确时为 True，否则 False。
|
||
|
||
关键实现:
|
||
直接以 per_q[q.question_id] 取值，缺任一题触发 KeyError（防静默兜底），
|
||
强制上游先补齐全部单题结果再计单元正确性。
|
||
"""
|
||
return all(per_q[q.question_id] for q in unit.questions)
|
||
|
||
|
||
def unit_correctness_view(units: list[QuestionUnit], per_q: dict[str, bool]) -> dict[str, bool]:
|
||
"""把逐题对错折叠成单元级视图：unit_id → 单元是否整体正确。
|
||
|
||
进化引擎（gate e-process / quadrant / probation / pair_block / compute_accuracy）
|
||
统一消费此单元视图，保证 AR pair 双向 AND、非 AR single 单题，混格池中
|
||
孪生对折叠为一个单元、不被 P/Q 单题计分污染（核心算法保真 #5）。
|
||
|
||
参数:
|
||
units: 目标单元列表（single 或 pair）。
|
||
per_q: 题目 question_id → 该题是否作答正确（唯一逐题溯源来源）。
|
||
|
||
返回:
|
||
unit_id → 单元级正确性。single 的 unit_id 等于其 question_id，
|
||
pair 的 unit_id 等于共享 pair_id。
|
||
|
||
关键实现:
|
||
逐单元复用 unit_correctness（内部以 per_q[q.question_id] 取值，缺任一题
|
||
触发 KeyError），禁静默兜底、强制上游先补齐全部单题结果。
|
||
"""
|
||
return {u.unit_id: unit_correctness(u, per_q) for u in units}
|