diff --git a/app/harness/gate_ladder.py b/app/harness/gate_ladder.py index aad589d..c47bbd2 100644 --- a/app/harness/gate_ladder.py +++ b/app/harness/gate_ladder.py @@ -1,13 +1,18 @@ -"""CE-Gate 信息量阶梯与基线缓存。 +"""CE-Gate 信息量阶梯与基线缓存（unit 粒度，核心算法保真 #5）。 -阶梯（每题型一条）：gate 的出题顺序表。冷启动（FRESH）用种子基线对错 -两档粗排（错题高优先 2:1 交错 + 全错题 probe_quota 探针插尾）； -epoch >=1 用非 gate run 观测做 gamma-EMA 更新 p_hat，按信息量 p_hat(1-p_hat) 降序、 -剔 p_hat 不在 [p_low, p_high]。防泄露铁律：gate 内 rollout 永不回流 p_hat -（调用方以 run_id 含 "_gate_" 过滤观测源）。 +阶梯（每题型一条）：gate 的出题顺序表，键为 **unit_id**（single 题 unit_id +等于 question_id，AR pair 折叠为一个单元、unit_id 等于共享 pair_id）。冷启动 +（FRESH）用种子基线的**单元级**对错两档粗排（错 unit 高优先 2:1 交错 + 全错 +unit 的 probe_quota 探针插尾）；epoch >=1 用非 gate run 观测**折叠成单元观测**后做 +gamma-EMA 更新 p_hat，按信息量 p_hat(1-p_hat) 降序、剔 p_hat 不在 [p_low, p_high]。 +单元错 = 该单元任一成员错（AR pair 双向 AND）。防泄露铁律：gate 内 rollout 永不 +回流 p_hat（调用方以 run_id 含 "_gate_" 过滤观测源），本迁移不改此过滤。 -BaselineCache：基线侧逐题对错缓存，键 = (task_type, skill_hash, -prompts_version, qid) 内容寻址、无显式失效。JSON 持久化到 workspace， +持久化门控：gate_pools.json 带 schema_version（当前 = 2，unit 键）。旧版无 +schema_version（v1、qid 键）加载时**直接报错**，拒绝静默混用 qid/unit 键。 + +BaselineCache：基线侧单元级对错缓存，键 = (task_type, skill_hash, +prompts_version, unit_id) 内容寻址、无显式失效。JSON 持久化到 workspace， 供 resume 后合法复用已冻结阶梯上的新鲜 draw。 """ @@ -22,10 +27,16 @@ from typing import TYPE_CHECKING from loguru import logger +from app.harness.question_units import build_units + if TYPE_CHECKING: from pathlib import Path - from core.types import GeneratedQuestion + from core.types import GeneratedQuestion, QuestionUnit + +# gate_pools.json 结构版本。v1（隐式、无此字段）为逐题 qid 键的存量格式； +# v2 起改为 unit_id 键。load 时严格校验，不匹配即报错（不静默迁移/混用）。 +SCHEMA_VERSION = 2 def skill_hash(content: str) -> str: @@ -42,50 +53,66 @@ def skill_hash(content: str) -> str: @dataclass class LadderEntry: - """阶梯单元：题目与其估计答对率。 + """阶梯单元：题目单元与其估计答对率。 字段: - question_id: 题目唯一标识。 + unit_id: 单元唯一标识（single 等于 question_id，AR pair 等于共享 pair_id）。 p_hat: 估计答对率。冷启动为 Beta(1,1) 平滑的单次观测后验均值 （错=1/3、对=2/3），此后经 gamma-EMA 更新。 """ - question_id: str + unit_id: str p_hat: float +def _unit_correct(unit: QuestionUnit, correctness: dict[str, bool]) -> bool: + """单元级正确性：AR pair 双向 AND，single 即单题；单元错 = 任一成员错。 + + 参数: + unit: 目标单元。 + correctness: question_id -> 是否答对（缺项按未答对处理，与迁移前 + correctness.get(qid, False) 的默认语义一致，不改判定）。 + + 返回: + 单元内所有成员均答对时 True，否则 False。 + """ + return all(correctness.get(q.question_id, False) for q in unit.questions) + + def build_cold_entries( - questions: list[GeneratedQuestion], + units: list[QuestionUnit], correctness: dict[str, bool], probe_quota: float, seed: int, ) -> list[LadderEntry]: - """冷启动排序：错题高优先 2:1 交错 + 全错题 probe_quota 探针插尾。 + """冷启动排序（unit 粒度）：错 unit 高优先 2:1 交错 + 全错 unit 探针插尾。 参数: - questions: 该题型的全部候选题（已排除 test 池）。 - correctness: question_id -> 种子基线是否答对（900 题全量对错）。 - probe_quota: 从错题中随机抽出插到梯尾的探针比例（防"解锁新能力"盲区）。 + units: 该题型的全部候选单元（已排除 test 池；AR pair 已折叠成单元）。 + correctness: question_id -> 种子基线是否答对（900 题全量逐题对错）。 + 单元级对错由 _unit_correct 折叠（任一成员错 → 单元错）。 + probe_quota: 从错 unit 中随机抽出插到梯尾的探针比例（防"解锁新能力"盲区）。 seed: 洗牌种子，保证确定性重建。 返回: - 排序后的 LadderEntry 列表（p_hat 用 Beta(1,1) 平滑：错=1/3、对=2/3， - 与 warm 阶段 gamma-EMA / 信息量排序自然衔接）。 + 排序后的 LadderEntry 列表（键=unit_id；p_hat 用 Beta(1,1) 平滑：错=1/3、 + 对=2/3，与 warm 阶段 gamma-EMA / 信息量排序自然衔接）。 关键实现细节: - 错题、对题各自固定种子洗牌 -> 抽探针 -> 剩余按 错错对 2:1 交错 - （一方耗尽后顺排另一方）-> 探针追加尾部。 + 与逐题版**同公式、同比例、同顺序**，仅把调度粒度从题换成单元：错 unit、 + 对 unit 各自固定种子洗牌 -> 按 probe_quota 从错 unit 抽探针 -> 剩余按 + 错错对 2:1 交错（一方耗尽后顺排另一方）-> 探针追加尾部。 """ rng = random.Random(seed) - wrong = [q for q in questions if not correctness.get(q.question_id, False)] - right = [q for q in questions if correctness.get(q.question_id, False)] + wrong = [u for u in units if not _unit_correct(u, correctness)] + right = [u for u in units if _unit_correct(u, correctness)] rng.shuffle(wrong) rng.shuffle(right) n_probe = int(len(wrong) * probe_quota) probes, wrong_main = wrong[:n_probe], wrong[n_probe:] - interleaved: list[GeneratedQuestion] = [] + interleaved: list[QuestionUnit] = [] wi, ri = 0, 0 while wi < len(wrong_main) or ri < len(right): for _ in range(2): @@ -97,10 +124,10 @@ def build_cold_entries( ri += 1 interleaved.extend(probes) - def _p0(q: GeneratedQuestion) -> float: - return 2 / 3 if correctness.get(q.question_id, False) else 1 / 3 + def _p0(u: QuestionUnit) -> float: + return 2 / 3 if _unit_correct(u, correctness) else 1 / 3 - return [LadderEntry(q.question_id, _p0(q)) for q in interleaved] + return [LadderEntry(u.unit_id, _p0(u)) for u in interleaved] def order_ladder(entries: list[LadderEntry], p_low: float, p_high: float) -> list[LadderEntry]: @@ -135,23 +162,24 @@ class GatePools: def ladder_for( self, task_type: str, - exclude_qids: set[str], + exclude_units: set[str], p_low: float, p_high: float, cold: bool, ) -> list[str]: - """取该题型的 gate 出题序（qid 列表），排除本 step 进化案例包题。 + """取该题型的 gate 出题序（unit_id 列表），排除本 step 进化案例包所在单元。 参数: task_type: 目标题型。 - exclude_qids: 本 step 案例包（failure/success cases）的题目 id， - 防止在"刚学的那道题"上自测。 + exclude_units: 本 step 案例包（failure/success cases）所在单元的 + unit_id，防止在"刚学的那道题"上自测。按 **unit** 排除：命中单元 + 整体剔除，避免只排 AR pair 半个成员而向 gate 池灌入半个 pair。 p_low / p_high: warm 阶段的 p_hat 保留区间。 cold: True 表示尚无 epoch 级观测（epoch 1），用冷启动存储序； False 走 order_ladder 信息量排序。 返回: - 排除后的有序 question_id 列表。 + 排除后的有序 unit_id 列表。 异常: ValueError: 该题型无阶梯（冷启动构建缺失），或该题型阶梯为空。 @@ -162,33 +190,54 @@ class GatePools: if not pool: raise ValueError(f"task_type={task_type} 阶梯为空，无可出题目") ordered = pool if cold else order_ladder(pool, p_low, p_high) - return [e.question_id for e in ordered if e.question_id not in exclude_qids] + return [e.unit_id for e in ordered if e.unit_id not in exclude_units] - def update_probs(self, observations: dict[str, bool], gamma: float) -> None: - """gamma-EMA 更新 p_hat：p_hat <- gamma * p_hat + (1-gamma) * obs。只更新有新观测的题。 + def update_probs( + self, + per_q_observations: dict[str, bool], + units_by_id: dict[str, QuestionUnit], + gamma: float, + ) -> None: + """gamma-EMA 更新 p_hat：先把逐题观测折叠成单元观测，再按 unit_id 匹配更新。 + + p_hat <- gamma * p_hat + (1-gamma) * unit_obs。只更新"整个单元都被观测到" + 的单元；单元观测 = 成员逐题对错的 AND（任一成员错 → 单元错）。折叠是必需的： + AR pair 的 unit_id 是 pair_id，若直接按 unit_id 去逐题观测里匹配将永不命中、 + 导致 gamma-EMA 停摆（核心算法保真 #5）。 参数: - observations: question_id -> 本 epoch 非 gate run 的最新对错。 + per_q_observations: question_id -> 本 epoch 非 gate run 的最新逐题对错。 调用方必须已按 run_id 过滤掉 gate 内 rollout（防泄露铁律）。 + units_by_id: unit_id -> QuestionUnit，用于把逐题观测折叠成单元观测。 gamma: EMA 衰减系数。 + + 关键实现细节: + 单元只有在其**全部**成员都出现在 per_q_observations 时才更新；半观测 + （AR pair 只见一半）跳过，避免用不完整证据污染 p_hat。 """ for entries in self.entries.values(): for e in entries: - if e.question_id in observations: - obs = 1.0 if observations[e.question_id] else 0.0 - e.p_hat = gamma * e.p_hat + (1 - gamma) * obs + unit = units_by_id.get(e.unit_id) + if unit is None: + continue + if not all(q.question_id in per_q_observations for q in unit.questions): + continue + unit_correct = all(per_q_observations[q.question_id] for q in unit.questions) + obs = 1.0 if unit_correct else 0.0 + e.p_hat = gamma * e.p_hat + (1 - gamma) * obs def save(self, path: Path) -> None: - """原子写 gate_pools.json（.tmp 再 replace）。 + """原子写 gate_pools.json（.tmp 再 replace），落 schema_version + unit_id 键。 参数: path: 目标 JSON 路径。 """ payload = { + "schema_version": SCHEMA_VERSION, "seed": self.seed, "fingerprint": self.fingerprint, "entries": { - t: [{"question_id": e.question_id, "p_hat": e.p_hat} for e in es] + t: [{"unit_id": e.unit_id, "p_hat": e.p_hat} for e in es] for t, es in self.entries.items() }, } @@ -198,18 +247,28 @@ class GatePools: @classmethod def load(cls, path: Path) -> GatePools: - """从 gate_pools.json 恢复。 + """从 gate_pools.json 恢复；schema_version 不匹配直接报错（不静默混用）。 参数: path: gate_pools.json 路径。 返回: 复活的 GatePools。 + + 异常: + RuntimeError: 缺 schema_version（存量 v1、qid 键）或版本不等于 + SCHEMA_VERSION——拒绝把 qid 键当 unit 键静默复用，须 FRESH 重建。 """ d = json.loads(path.read_text(encoding="utf-8")) + version = d.get("schema_version") + if version != SCHEMA_VERSION: + raise RuntimeError( + f"gate_pools.json schema_version={version!r} 与当前 {SCHEMA_VERSION} 不符" + f"（存量 qid 键池不可当 unit 键复用），请删除后 FRESH 重建: {path}" + ) return cls( entries={ - t: [LadderEntry(x["question_id"], x["p_hat"]) for x in es] + t: [LadderEntry(x["unit_id"], x["p_hat"]) for x in es] for t, es in d["entries"].items() }, seed=d["seed"], @@ -264,22 +323,46 @@ def build_or_load_gate_pools( entries: dict[str, list[LadderEntry]] = {} for t in task_types: - pool = [q for q in questions if q.task_type == t and q.question_id not in test_qids] - if not pool: - raise ValueError(f"task_type={t} 无非 test 题，无法建阶梯") - entries[t] = build_cold_entries(pool, baseline_correctness, probe_quota, seed) - logger.info("gate 阶梯[{}]: {} 题（冷启动）", t, len(entries[t])) + units = _task_units_excluding_test(questions, t, test_qids) + if not units: + raise ValueError(f"task_type={t} 无非 test 单元，无法建阶梯") + entries[t] = build_cold_entries(units, baseline_correctness, probe_quota, seed) + logger.info("gate 阶梯[{}]: {} 单元（冷启动）", t, len(entries[t])) pools = GatePools(entries=entries, seed=seed, fingerprint=fingerprint) pools.save(path) return pools -class BaselineCache: - """基线侧逐题对错缓存（内容寻址，JSON 持久化）。 +def _task_units_excluding_test( + questions: list[GeneratedQuestion], task_type: str, test_qids: set[str] +) -> list[QuestionUnit]: + """取某题型的非 test 候选单元：先按 unit 折叠，再整体排除含 test 成员的单元。 - 键 = (task_type, skill_hash, prompts_version, qid)：任何影响该题型 + 先折叠后排除保证 AR pair 不被拆半（否则半个 pair 交给下游会触发 build_units 的 + 孤儿 fail-fast）；single 单元等价于逐题排除（核心算法保真 #5）。 + + 参数: + questions: benchmark 全量题。 + task_type: 目标题型。 + test_qids: held-out test 池题目 id。 + + 返回: + 该题型下不含任何 test 成员的候选单元列表。 + """ + pool = [q for q in questions if q.task_type == task_type] + return [ + u for u in build_units(pool) if all(q.question_id not in test_qids for q in u.questions) + ] + + +class BaselineCache: + """基线侧单元级对错缓存（内容寻址，JSON 持久化）。 + + 键 = (task_type, skill_hash, prompts_version, unit_id)：任何影响该题型 有效 skill 的变化（含共享 default-strategy.md 被他类 accept 改写） - 都使 skill_hash 变化、缓存自然 miss；prompts 版本变化同理。 + 都使 skill_hash 变化、缓存自然 miss；prompts 版本变化同理。unit_id 维度 + 使 single 题以自身 question_id、AR pair 以共享 pair_id 寻址，缓存单元级 + 对错（pair 双向 AND 折叠后一个布尔）。 """ def __init__(self, path: Path) -> None: @@ -294,32 +377,32 @@ class BaselineCache: self._store = json.loads(path.read_text(encoding="utf-8")) @staticmethod - def _key(task_type: str, s_hash: str, prompts_version: str, qid: str) -> str: - """拼缓存键（四维内容寻址）。""" - return f"{task_type}|{s_hash}|{prompts_version}|{qid}" + def _key(task_type: str, s_hash: str, prompts_version: str, unit_id: str) -> str: + """拼缓存键（四维内容寻址，第四维为 unit_id）。""" + return f"{task_type}|{s_hash}|{prompts_version}|{unit_id}" - def get(self, task_type: str, s_hash: str, prompts_version: str, qid: str) -> bool | None: + def get(self, task_type: str, s_hash: str, prompts_version: str, unit_id: str) -> bool | None: """读缓存；未命中返回 None。 参数: task_type: 题型。 s_hash: 基线侧生效 skill 文件的内容哈希。 prompts_version: 当前 prompts 版本。 - qid: 题目 id。 + unit_id: 单元 id（single=question_id，AR pair=pair_id）。 返回: - 缓存的对错；未命中 None。 + 缓存的单元级对错；未命中 None。 """ - return self._store.get(self._key(task_type, s_hash, prompts_version, qid)) + return self._store.get(self._key(task_type, s_hash, prompts_version, unit_id)) def put( - self, task_type: str, s_hash: str, prompts_version: str, qid: str, correct: bool + self, task_type: str, s_hash: str, prompts_version: str, unit_id: str, correct: bool ) -> None: """写缓存并落盘（原子写，gate 频度低、全量重写成本可忽略）。 参数: - task_type / s_hash / prompts_version / qid: 缓存键四维。 - correct: 基线侧该题对错。 + task_type / s_hash / prompts_version / unit_id: 缓存键四维。 + correct: 基线侧该单元对错（AR pair 双向 AND 折叠后一个布尔）。 关键实现细节: 先盘后存：新条目先原子落盘（tmp 写 + os.replace）成功后才更新 @@ -327,7 +410,7 @@ class BaselineCache: """ updated = { **self._store, - self._key(task_type, s_hash, prompts_version, qid): correct, + self._key(task_type, s_hash, prompts_version, unit_id): correct, } tmp = self._path.with_suffix(".json.tmp") tmp.write_text(json.dumps(updated, ensure_ascii=False), encoding="utf-8") diff --git a/app/harness/runner.py b/app/harness/runner.py index 1a1c841..80c6110 100644 --- a/app/harness/runner.py +++ b/app/harness/runner.py @@ -854,6 +854,11 @@ class Runner: self._gate_questions_by_id: dict[str, GeneratedQuestion] = { q.question_id: q for q in questions } + # unit 索引：ladder_for 返回 unit_id、update_probs 折叠逐题观测均需按 unit_id + # 反查成员题（核心算法保真 #5：gate 阶梯 unit 化）。不进 checkpoint、每次启动重建。 + self._gate_units_by_id: dict[str, QuestionUnit] = { + u.unit_id: u for u in build_units(questions) + } with HarnessLog(str(self._paths.db_path), pools.baseline_run_id) as log: rows = log.query( "SELECT question_id, prediction, answer FROM predictions WHERE run_id=?", @@ -1108,21 +1113,29 @@ class Runner: from app.harness.log import HarnessLog from app.harness.validate import validate_skill_local - exclude_qids = {c.question_id for c in pack.failure_cases + pack.success_cases} - ladder_qids = state.gate_pools.ladder_for( + # 案例包按 unit 排除：把每个 case 的 question_id 映射到其所属 unit_id， + # 命中单元整体排除，防止只排 AR pair 半个成员而给 gate 池灌半个 pair + # （下游 _ladder_units 会 fail-fast）。核心算法保真 #5。 + exclude_units = { + self._gate_questions_by_id[c.question_id].unit_id + for c in pack.failure_cases + pack.success_cases + if c.question_id in self._gate_questions_by_id + } + ladder_unit_ids = state.gate_pools.ladder_for( task_type, - exclude_qids, + exclude_units, p_low=self._config.gate_p_low, p_high=self._config.gate_p_high, cold=not state.gate_epoch_observed, ) - missing = [qid for qid in ladder_qids if qid not in self._gate_questions_by_id] + missing = [uid for uid in ladder_unit_ids if uid not in self._gate_units_by_id] if missing: raise ValueError( - f"gate 阶梯[{task_type}] 含 benchmark 中不存在的题: " + f"gate 阶梯[{task_type}] 含 benchmark 中不存在的单元: " f"{missing[:5]}（gate_pools.json 与题库失配）" ) - ladder_items = [self._gate_questions_by_id[qid] for qid in ladder_qids] + # 单元展开为逐题（unit 内成员顺序保持），下游 validate 再按阶梯序聚合回单元。 + ladder_items = [q for uid in ladder_unit_ids for q in self._gate_units_by_id[uid].questions] base_skill_content = (self._paths.skills_dir / record.target_file).read_text( encoding="utf-8" ) @@ -1799,7 +1812,10 @@ class Runner: obs.update({r["question_id"]: r["prediction"] == r["answer"] for r in slow_rows}) for extra_rows in extra_rows_lists: obs.update({r["question_id"]: r["prediction"] == r["answer"] for r in extra_rows}) - state.gate_pools.update_probs(obs, gamma=self._config.gate_gamma_decay) + # 逐题观测折叠成单元观测后按 unit_id 匹配更新（防按 qid 匹配 pair 失效致 EMA 停摆）。 + state.gate_pools.update_probs( + obs, self._gate_units_by_id, gamma=self._config.gate_gamma_decay + ) state.gate_pools.save(self._config.workspace_dir / "gate_pools.json") state.gate_epoch_observed = True diff --git a/tests/unit/test_gate_ladder_unit_migration.py b/tests/unit/test_gate_ladder_unit_migration.py new file mode 100644 index 0000000..af7b6aa --- /dev/null +++ b/tests/unit/test_gate_ladder_unit_migration.py @@ -0,0 +1,295 @@ +"""gate_ladder.py 单元化迁移测试（核心算法保真 #5）。 + +验证 gate 信息量阶梯从"逐题"迁移到"unit"粒度后，冷启动 2:1 错优先交错、 +gamma-EMA、Beta 先验、schema_version 门控等语义"只换键、不改公式/比例/顺序"： + +(a) LadderEntry 按 unit_id 键（AR pair 折叠为一个阶梯单元）； +(b) 冷启动"错优先 2:1"以 unit 为单位，unit 错 = P 或 Q 任一错； +(c) update_probs 观测先折叠成 unit 再匹配（防按 qid 匹配失效致 EMA 停摆）； +(d) GatePools.save/load 带 schema_version，存量无版本 json 明确报错（不静默混用）； +(e) BaselineCache 键含 unit_id（pair 的 unit_id 与成员 qid 不同）。 +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +from app.harness.gate_ladder import ( + SCHEMA_VERSION, + BaselineCache, + GatePools, + LadderEntry, + build_cold_entries, +) +from app.harness.question_units import build_units +from core.types import GeneratedQuestion + +if TYPE_CHECKING: + from pathlib import Path + + +# ── 构造工具 ────────────────────────────────────────────────────────── + + +def _single(qid: str, task_type: str = "AR") -> GeneratedQuestion: + """构造 single 题（unit_id 回填为 question_id）。""" + return GeneratedQuestion( + question_id=qid, + video_id="v1", + task_type=task_type, + question="dummy", + options=("A", "B", "C", "D"), + answer="A", + source_nodes=("n1",), + difficulty="easy", + ) + + +def _pair(pair_id: str, task_type: str = "AR") -> list[GeneratedQuestion]: + """构造一条 AR 孪生对（original + mirror），共享 pair_id 即 unit_id。""" + base = { + "video_id": "v1", + "task_type": task_type, + "question": "dummy", + "options": ("A", "B", "C", "D"), + "answer": "A", + "source_nodes": ("n1",), + "difficulty": "easy", + "pair_id": pair_id, + "flip_axis": "before_after", + } + return [ + GeneratedQuestion(question_id=f"{pair_id}_o", question_role="pair_original", **base), + GeneratedQuestion(question_id=f"{pair_id}_m", question_role="pair_mirror", **base), + ] + + +# ── (a) LadderEntry 按 unit_id ───────────────────────────────────────── + + +class TestLadderEntryKeyedByUnit: + """LadderEntry 以 unit_id 为键；AR pair 折叠为一个阶梯单元。""" + + def test_entry_has_unit_id(self) -> None: + """LadderEntry 暴露 unit_id 字段。""" + e = LadderEntry("u1", 0.5) + assert e.unit_id == "u1" + + def test_pair_collapses_to_single_entry(self) -> None: + """一条孪生对（2 题）在阶梯中只产生 1 个 unit 条目（键=pair_id）。""" + units = build_units(_pair("pr1")) + correctness = {"pr1_o": True, "pr1_m": True} + entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=1) + assert len(entries) == 1 + assert entries[0].unit_id == "pr1" + + +# ── (b) 冷启动 2:1 错优先 + unit 错 = P 或 Q 任一错 ─────────────────────── + + +class TestColdStartUnit: + """冷启动排序在 unit 粒度保持 2:1 交错与 Beta 先验；unit 错 = 任一成员错。""" + + def test_unit_wrong_if_any_member_wrong(self) -> None: + """pair 中任一成员错 → 该 unit 判错（p_hat=1/3）；全对才判对（2/3）。""" + units = build_units(_pair("wrong") + _pair("right")) + # wrong: original 对、mirror 错 → unit 错；right: 两题均对 → unit 对 + correctness = { + "wrong_o": True, + "wrong_m": False, + "right_o": True, + "right_m": True, + } + entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=3) + p_by_unit = {e.unit_id: e.p_hat for e in entries} + assert p_by_unit["wrong"] == pytest.approx(1 / 3) + assert p_by_unit["right"] == pytest.approx(2 / 3) + + def test_two_to_one_interleave_over_units(self) -> None: + """6 错 unit + 3 对 unit（含 pair），probe_quota=0 → 交错序 W W R W W R W W R。 + + 交错是 unit 粒度：pair 折叠成一个 unit 参与交错，序列长度为 9（unit 数）， + 而非 18（题数），证明 2:1 比例语义只换键不改。 + """ + wrong_units_q: list[GeneratedQuestion] = [] + for i in range(6): + wrong_units_q += _pair(f"w{i}") # 6 个 pair unit + right_units_q = [_single(f"r{i}") for i in range(3)] # 3 个 single unit + units = build_units(wrong_units_q + right_units_q) + + correctness: dict[str, bool] = {} + for i in range(6): + correctness[f"w{i}_o"] = True + correctness[f"w{i}_m"] = False # 任一错 → unit 错 + for i in range(3): + correctness[f"r{i}"] = True + + entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=42) + assert len(entries) == 9 + # unit 错 = p_hat≈1/3；unit 对 = p_hat≈2/3 + pattern = ["W" if e.p_hat < 0.5 else "R" for e in entries] + assert pattern == ["W", "W", "R", "W", "W", "R", "W", "W", "R"] + + def test_probe_at_tail_unit(self) -> None: + """probe_quota>0 时从错 unit 抽探针追加梯尾（按 unit 抽，非按题）。""" + wrong_q: list[GeneratedQuestion] = [] + for i in range(10): + wrong_q += _pair(f"w{i}") + right_q = [_single(f"r{i}") for i in range(2)] + units = build_units(wrong_q + right_q) + correctness = {} + for i in range(10): + correctness[f"w{i}_o"] = False + correctness[f"w{i}_m"] = False + for i in range(2): + correctness[f"r{i}"] = True + + entries = build_cold_entries(units, correctness, probe_quota=0.3, seed=7) + # 10 错 unit * 0.3 = 3 个探针 unit 在尾部，均为错 unit + tail = entries[-3:] + for e in tail: + assert e.p_hat < 0.5 + + +# ── (c) update_probs 折叠成 unit 再匹配 ────────────────────────────────── + + +class TestUpdateProbsFold: + """gamma-EMA 更新前先把逐题观测折叠成 unit 观测（AR pair 双向 AND）。""" + + def _pools_and_units(self) -> tuple[GatePools, dict]: + """构造含一个 pair unit + 一个 single unit 的池与 unit 索引。""" + units = build_units(_pair("pr1") + [_single("s1")]) + units_by_id = {u.unit_id: u for u in units} + entries = {"AR": [LadderEntry("pr1", 0.5), LadderEntry("s1", 0.5)]} + return GatePools(entries=entries, seed=0, fingerprint="x"), units_by_id + + def test_pair_updates_after_fold(self) -> None: + """pair 两成员均观测 → 折叠成 unit 观测 → EMA 更新（不停摆）。""" + pools, units_by_id = self._pools_and_units() + # pair 两题均对 → unit 对(1.0)；single 错(0.0) + per_q = {"pr1_o": True, "pr1_m": True, "s1": False} + pools.update_probs(per_q, units_by_id, gamma=0.8) + p = {e.unit_id: e.p_hat for e in pools.entries["AR"]} + assert p["pr1"] == pytest.approx(0.8 * 0.5 + 0.2 * 1.0) # 0.6 + assert p["s1"] == pytest.approx(0.8 * 0.5 + 0.2 * 0.0) # 0.4 + + def test_pair_wrong_if_any_member_wrong(self) -> None: + """pair 任一成员错 → unit 观测为错(0.0)，EMA 向下。""" + pools, units_by_id = self._pools_and_units() + per_q = {"pr1_o": True, "pr1_m": False, "s1": True} + pools.update_probs(per_q, units_by_id, gamma=0.8) + p = {e.unit_id: e.p_hat for e in pools.entries["AR"]} + assert p["pr1"] == pytest.approx(0.8 * 0.5 + 0.2 * 0.0) # 0.4 + + def test_partial_pair_observation_skips_update(self) -> None: + """pair 只观测到半个成员 → 无法折叠 → 该 unit p_hat 不变（不半 pair 污染）。""" + pools, units_by_id = self._pools_and_units() + per_q = {"pr1_o": True} # 缺 mirror + pools.update_probs(per_q, units_by_id, gamma=0.8) + p = {e.unit_id: e.p_hat for e in pools.entries["AR"]} + assert p["pr1"] == pytest.approx(0.5) # 未更新 + + def test_qid_keyed_observation_does_not_match_pair(self) -> None: + """若观测里错用 pair_id 之外的裸 qid 键、且未提供 unit 折叠，pair 不应被误更新。 + + 证明"必须折叠"：单 single unit 用其自身 qid 可更新，pair 需 unit 折叠。 + """ + pools, units_by_id = self._pools_and_units() + # 只给 single 的观测，pair 两成员均无观测 + per_q = {"s1": True} + pools.update_probs(per_q, units_by_id, gamma=0.8) + p = {e.unit_id: e.p_hat for e in pools.entries["AR"]} + assert p["pr1"] == pytest.approx(0.5) # pair 未观测 → 不变 + assert p["s1"] == pytest.approx(0.8 * 0.5 + 0.2 * 1.0) # single 更新 + + +# ── (d) schema_version 门控 ───────────────────────────────────────────── + + +class TestSchemaVersion: + """GatePools.save/load 带 schema_version；存量无版本 json 明确报错。""" + + def test_save_writes_schema_version_and_unit_id(self, tmp_path: Path) -> None: + """save 落盘含 schema_version 且 entries 用 unit_id 键。""" + entries = {"AR": [LadderEntry("pr1", 0.33), LadderEntry("s1", 0.67)]} + pools = GatePools(entries=entries, seed=1, fingerprint="fp") + path = tmp_path / "gate_pools.json" + pools.save(path) + + raw = json.loads(path.read_text(encoding="utf-8")) + assert raw["schema_version"] == SCHEMA_VERSION + assert raw["entries"]["AR"][0]["unit_id"] == "pr1" + assert "question_id" not in raw["entries"]["AR"][0] + + def test_save_load_roundtrip(self, tmp_path: Path) -> None: + """save → load 往返保真（unit_id + p_hat + seed + fingerprint）。""" + entries = {"AR": [LadderEntry("pr1", 0.4)]} + pools = GatePools(entries=entries, seed=9, fingerprint="fp2") + path = tmp_path / "gate_pools.json" + pools.save(path) + loaded = GatePools.load(path) + assert loaded.seed == 9 + assert loaded.fingerprint == "fp2" + assert loaded.entries["AR"][0].unit_id == "pr1" + assert loaded.entries["AR"][0].p_hat == pytest.approx(0.4) + + def test_load_legacy_without_schema_version_raises(self, tmp_path: Path) -> None: + """存量无 schema_version（旧 qid 键）→ 明确报错，不静默混用。""" + legacy = { + "seed": 1, + "fingerprint": "fp", + "entries": {"AR": [{"question_id": "q1", "p_hat": 0.5}]}, + } + path = tmp_path / "gate_pools.json" + path.write_text(json.dumps(legacy), encoding="utf-8") + with pytest.raises(RuntimeError, match="schema_version"): + GatePools.load(path) + + def test_load_wrong_schema_version_raises(self, tmp_path: Path) -> None: + """schema_version 不匹配 → 明确报错。""" + bad = { + "schema_version": SCHEMA_VERSION + 99, + "seed": 1, + "fingerprint": "fp", + "entries": {"AR": [{"unit_id": "pr1", "p_hat": 0.5}]}, + } + path = tmp_path / "gate_pools.json" + path.write_text(json.dumps(bad), encoding="utf-8") + with pytest.raises(RuntimeError, match="schema_version"): + GatePools.load(path) + + +# ── (e) BaselineCache 键含 unit_id ───────────────────────────────────── + + +class TestBaselineCacheUnitKey: + """BaselineCache 以 unit_id 为第四维；pair 的 unit_id 与成员 qid 区分。""" + + def test_unit_id_key_distinct_from_member_qid(self, tmp_path: Path) -> None: + """pair unit_id（=pair_id）与其成员 qid 是不同缓存键。""" + cache = BaselineCache(tmp_path / "baseline_cache.json") + cache.put("AR", "h1", "v1", "pr1", True) # unit_id=pair_id + assert cache.get("AR", "h1", "v1", "pr1") is True + # 成员 qid 不是同一键 → miss + assert cache.get("AR", "h1", "v1", "pr1_o") is None + assert cache.get("AR", "h1", "v1", "pr1_m") is None + + +# ── ladder_for 排除按 unit ───────────────────────────────────────────── + + +class TestLadderForExcludeUnit: + """ladder_for 返回 unit_id 序，按 unit 排除（防半 pair 灌入）。""" + + def test_exclude_units_filters_whole_unit(self) -> None: + """exclude_units 命中的 unit 被整体排除，返回 unit_id 列表。""" + entries = {"AR": [LadderEntry("pr1", 0.5), LadderEntry("s1", 0.4), LadderEntry("s2", 0.6)]} + pools = GatePools(entries=entries, seed=0, fingerprint="x") + result = pools.ladder_for("AR", exclude_units={"pr1"}, p_low=0.0, p_high=1.0, cold=True) + assert "pr1" not in result + assert "s1" in result + assert "s2" in result diff --git a/tests/unit/test_harness_gate_ladder.py b/tests/unit/test_harness_gate_ladder.py index 18edfb5..d5a8b07 100644 --- a/tests/unit/test_harness_gate_ladder.py +++ b/tests/unit/test_harness_gate_ladder.py @@ -20,6 +20,7 @@ from app.harness.gate_ladder import ( order_ladder, skill_hash, ) +from app.harness.question_units import build_units from core.types import GeneratedQuestion if TYPE_CHECKING: @@ -43,6 +44,11 @@ def _make_q(qid: str, task_type: str = "AR") -> GeneratedQuestion: ) +def _units(questions: list[GeneratedQuestion]) -> list: + """把题目列表折叠为单元列表（single 题 unit_id 等于 question_id）。""" + return build_units(questions) + + # ── 冷启动 ──────────────────────────────────────────────────────────── @@ -50,52 +56,52 @@ class TestColdStart: """冷启动排序：2:1 交错 + 探针插尾 + Beta(1,1) 平滑。""" def test_cold_start_interleaving(self) -> None: - """错题:对题 = 2:1 交错顺序。 + """错 unit:对 unit = 2:1 交错顺序。 6 错 3 对（probe_quota=0 无探针）→ 交错序应为 W W R W W R W W R。 """ wrong_ids = [f"w{i}" for i in range(6)] right_ids = [f"r{i}" for i in range(3)] - questions = [_make_q(qid) for qid in wrong_ids + right_ids] + units = _units([_make_q(qid) for qid in wrong_ids + right_ids]) correctness = dict.fromkeys(wrong_ids, False) correctness.update(dict.fromkeys(right_ids, True)) - entries = build_cold_entries(questions, correctness, probe_quota=0.0, seed=42) + entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=42) assert len(entries) == 9 # 验证 2:1 交错模式（seed 固定后 shuffle 结果确定） - pattern = ["W" if not correctness[e.question_id] else "R" for e in entries] + pattern = ["W" if not correctness[e.unit_id] else "R" for e in entries] # 前 9 个交错应为 W W R W W R W W R assert pattern == ["W", "W", "R", "W", "W", "R", "W", "W", "R"] def test_cold_start_p_hat_beta(self) -> None: """p_hat 遵循 Beta(1,1) 平滑：错=1/3，对=2/3。""" - questions = [_make_q("q1"), _make_q("q2")] + units = _units([_make_q("q1"), _make_q("q2")]) correctness = {"q1": False, "q2": True} - entries = build_cold_entries(questions, correctness, probe_quota=0.0, seed=0) + entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=0) - p_map = {e.question_id: e.p_hat for e in entries} + p_map = {e.unit_id: e.p_hat for e in entries} assert p_map["q1"] == pytest.approx(1 / 3) assert p_map["q2"] == pytest.approx(2 / 3) def test_cold_start_probe_at_tail(self) -> None: - """probe_quota > 0 时探针题追加在尾部。""" + """probe_quota > 0 时探针 unit 追加在尾部。""" wrong_ids = [f"w{i}" for i in range(10)] right_ids = [f"r{i}" for i in range(2)] - questions = [_make_q(qid) for qid in wrong_ids + right_ids] + units = _units([_make_q(qid) for qid in wrong_ids + right_ids]) correctness = dict.fromkeys(wrong_ids, False) correctness.update(dict.fromkeys(right_ids, True)) - entries = build_cold_entries(questions, correctness, probe_quota=0.3, seed=7) + entries = build_cold_entries(units, correctness, probe_quota=0.3, seed=7) # 10 错 * 0.3 = 3 个探针在尾部 n_probe = int(10 * 0.3) assert n_probe == 3 - # 尾部 3 个都应为错题 + # 尾部 3 个都应为错 unit tail = entries[-n_probe:] for e in tail: - assert not correctness[e.question_id] + assert not correctness[e.unit_id] # ── warm 排序 ────────────────────────────────────────────────────────── @@ -113,9 +119,9 @@ class TestWarmOrdering: LadderEntry("d", 0.3), ] ordered = order_ladder(entries, p_low=0.0, p_high=1.0) - assert ordered[0].question_id == "b" # 0.5*(1-0.5)=0.25 最高 + assert ordered[0].unit_id == "b" # 0.5*(1-0.5)=0.25 最高 # d: 0.3*0.7=0.21, a: 0.1*0.9=0.09, c: 0.9*0.1=0.09 - assert ordered[1].question_id == "d" + assert ordered[1].unit_id == "d" def test_warm_filter_bounds(self) -> None: """p_hat 不在 [p_low, p_high] 区间的题被剔除。""" @@ -125,7 +131,7 @@ class TestWarmOrdering: LadderEntry("high", 0.95), ] ordered = order_ladder(entries, p_low=0.1, p_high=0.9) - ids = [e.question_id for e in ordered] + ids = [e.unit_id for e in ordered] assert "mid" in ids assert "low" not in ids assert "high" not in ids @@ -155,9 +161,9 @@ class TestGatePoolsPersistence: assert loaded.seed == 42 assert loaded.fingerprint == "abc123" assert len(loaded.entries["AR"]) == 2 - assert loaded.entries["AR"][0].question_id == "q1" + assert loaded.entries["AR"][0].unit_id == "q1" assert loaded.entries["AR"][0].p_hat == pytest.approx(0.33) - assert loaded.entries["CR"][0].question_id == "q3" + assert loaded.entries["CR"][0].unit_id == "q3" def test_gate_pools_fingerprint_mismatch(self, tmp_path: Path) -> None: """指纹不一致 -> RuntimeError（不静默重建）。""" @@ -196,8 +202,8 @@ class TestGatePoolsPersistence: class TestLadderFor: """ladder_for 取题序与排除逻辑。""" - def test_ladder_for_excludes_qids(self) -> None: - """exclude_qids 中的题被排除。""" + def test_ladder_for_excludes_units(self) -> None: + """exclude_units 中的单元被排除。""" entries = { "AR": [ LadderEntry("q1", 0.5), @@ -206,7 +212,7 @@ class TestLadderFor: ], } pools = GatePools(entries=entries, seed=0, fingerprint="x") - result = pools.ladder_for("AR", exclude_qids={"q2"}, p_low=0.0, p_high=1.0, cold=True) + result = pools.ladder_for("AR", exclude_units={"q2"}, p_low=0.0, p_high=1.0, cold=True) assert "q2" not in result assert "q1" in result assert "q3" in result @@ -236,28 +242,30 @@ class TestLadderFor: class TestGammaEMA: - """gamma-EMA 更新 p_hat。""" + """gamma-EMA 更新 p_hat（single 单元：unit_id 等于 question_id）。""" def test_gamma_ema_update(self) -> None: """p_hat <- gamma * p_hat + (1-gamma) * obs。""" entries = {"AR": [LadderEntry("q1", 0.5)]} pools = GatePools(entries=entries, seed=0, fingerprint="x") + units_by_id = {u.unit_id: u for u in _units([_make_q("q1")])} # 观测为正确(1.0), gamma=0.8 - pools.update_probs({"q1": True}, gamma=0.8) + pools.update_probs({"q1": True}, units_by_id, gamma=0.8) expected = 0.8 * 0.5 + 0.2 * 1.0 # 0.6 assert pools.entries["AR"][0].p_hat == pytest.approx(expected) # 再次观测为错误(0.0), gamma=0.8 - pools.update_probs({"q1": False}, gamma=0.8) + pools.update_probs({"q1": False}, units_by_id, gamma=0.8) expected2 = 0.8 * expected + 0.2 * 0.0 # 0.48 assert pools.entries["AR"][0].p_hat == pytest.approx(expected2) def test_update_probs_no_observation_unchanged(self) -> None: - """无观测的题 p_hat 不变。""" + """无观测的单元 p_hat 不变。""" entries = {"AR": [LadderEntry("q1", 0.5), LadderEntry("q2", 0.3)]} pools = GatePools(entries=entries, seed=0, fingerprint="x") - pools.update_probs({"q1": True}, gamma=0.9) + units_by_id = {u.unit_id: u for u in _units([_make_q("q1"), _make_q("q2")])} + pools.update_probs({"q1": True}, units_by_id, gamma=0.9) assert pools.entries["AR"][1].p_hat == pytest.approx(0.3) @@ -290,7 +298,8 @@ class TestLeakPrevention: # 只有普通 run 的观测进入 update_probs assert filtered == {"q1": True} - pools.update_probs(filtered, gamma=0.8) + units_by_id = {u.unit_id: u for u in _units([_make_q("q1")])} + pools.update_probs(filtered, units_by_id, gamma=0.8) expected = 0.8 * 0.5 + 0.2 * 1.0 assert pools.entries["AR"][0].p_hat == pytest.approx(expected)