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）。
89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
"""core/evolution/validate.py — 块验证纯决策函数。
|
||
|
||
算法 #7（块顺序验证）的局部实现：pair_block 按 unit 比对基线与候选、
|
||
classify_quadrants 四象限分类、compute_accuracy 纯算术准确率。
|
||
|
||
三个函数均为纯函数，无副作用、无外部依赖。输入的对错映射均为 **unit 口径**
|
||
（unit_id → 单元级正确性，AR pair 已在上游经 unit_correctness_view 双向 AND
|
||
折叠），保证 e-process W/L 与准确率分母按单元计、不被 P/Q 单题计分污染
|
||
（核心算法保真 #5：信息阶梯口径从 question_id 迁至 unit_id）。
|
||
"""
|
||
|
||
from core.evolution.types import PairResult, QuadrantClassification
|
||
|
||
|
||
def pair_block(
|
||
baseline: dict[str, bool],
|
||
candidate: dict[str, bool],
|
||
unit_ids: list[str],
|
||
) -> PairResult:
|
||
"""按单元比对基线与候选对错，统计翻转。
|
||
|
||
参数:
|
||
baseline: 基线臂单元级正确性映射（unit_id → bool）。
|
||
candidate: 候选臂单元级正确性映射（unit_id → bool）。
|
||
unit_ids: 参与比对的单元 ID 列表（AR pair 折叠后为单一 unit_id）。
|
||
|
||
返回:
|
||
PairResult，包含 w（基线错→候选对翻转数）、l（基线对→候选错翻转数）
|
||
和 observed（每单元的 (基线, 候选) 对错记录）。
|
||
"""
|
||
w = l = 0 # noqa: E741 — 数学记号 W/L（win/loss），与 gate.py 一致
|
||
observed: dict[str, tuple[bool, bool]] = {}
|
||
for uid in unit_ids:
|
||
b, c = baseline[uid], candidate[uid]
|
||
observed[uid] = (b, c)
|
||
if not b and c:
|
||
w += 1
|
||
elif b and not c:
|
||
l += 1 # noqa: E741
|
||
return PairResult(w=w, l=l, observed=observed)
|
||
|
||
|
||
def classify_quadrants(
|
||
observed: dict[str, tuple[bool, bool]],
|
||
) -> QuadrantClassification:
|
||
"""按 (baseline, candidate) 四组分类，各组内 sorted。
|
||
|
||
参数:
|
||
observed: 每题的 (基线是否正确, 候选是否正确) 记录。
|
||
|
||
返回:
|
||
QuadrantClassification，四个象限各含排序后的题目 ID 列表。
|
||
"""
|
||
improvements: list[str] = []
|
||
regressions: list[str] = []
|
||
persistent_fails: list[str] = []
|
||
stable_successes: list[str] = []
|
||
for qid, (prev, curr) in observed.items():
|
||
if not prev and curr:
|
||
improvements.append(qid)
|
||
elif prev and not curr:
|
||
regressions.append(qid)
|
||
elif not prev and not curr:
|
||
persistent_fails.append(qid)
|
||
else:
|
||
stable_successes.append(qid)
|
||
return QuadrantClassification(
|
||
improvements=sorted(improvements),
|
||
regressions=sorted(regressions),
|
||
persistent_fails=sorted(persistent_fails),
|
||
stable_successes=sorted(stable_successes),
|
||
)
|
||
|
||
|
||
def compute_accuracy(
|
||
correctness: dict[str, bool],
|
||
unit_ids: list[str],
|
||
) -> float:
|
||
"""纯算术：sum(correct) / len(units)，分母按单元数（非逐题）。
|
||
|
||
参数:
|
||
correctness: 单元级正确性映射（unit_id → bool）。
|
||
unit_ids: 参与计算的单元 ID 列表。
|
||
|
||
返回:
|
||
准确率浮点数。unit_ids 为空时抛出 ZeroDivisionError。
|
||
"""
|
||
return sum(correctness[uid] for uid in unit_ids) / len(unit_ids)
|