diff --git a/app/harness/question_units.py b/app/harness/question_units.py index 006d87d..7586f15 100644 --- a/app/harness/question_units.py +++ b/app/harness/question_units.py @@ -111,3 +111,25 @@ def unit_correctness(unit: QuestionUnit, per_q: dict[str, bool]) -> bool: 强制上游先补齐全部单题结果再计单元正确性。 """ 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} diff --git a/app/harness/runner.py b/app/harness/runner.py index fdb1fd5..1a1c841 100644 --- a/app/harness/runner.py +++ b/app/harness/runner.py @@ -42,6 +42,7 @@ from app.harness.observation import ( write_shadow_gate, write_step_report, ) +from app.harness.question_units import build_units, unit_correctness_view from app.harness.store import advance_version from app.harness.validate import Probation, ValidationOutcome from app.harness.workspace import ( @@ -61,6 +62,7 @@ from core.evolution import ( RejectedEdit, edit_budget_at, momentum_inner, + pair_block, probation_verdict, replace_momentum, resolve_skill_file, @@ -72,11 +74,12 @@ if TYPE_CHECKING: from app.harness.pools import Pools from core.evolution.types import ( EvolutionRecord, + PairResult, SystemCasePack, ToolCasePack, ) from core.protocols import LLMProvider, TelemetryRecorder, VLMProvider - from core.types import GeneratedQuestion + from core.types import GeneratedQuestion, QuestionUnit class _InterruptError(RuntimeError): @@ -267,15 +270,57 @@ def _compute_total_steps(pools: Pools, correctness: dict[str, bool], config: Run return config.epochs * steps_per_epoch +def _probation_unit_flips( + val_units: list[QuestionUnit], + snapshot: dict[str, bool], + rows: dict[str, dict[str, Any]], + task_type: str, + eval_run_id: str, +) -> PairResult: + """按 unit 折叠锚快照与当前重跑对错,返回单元级翻转统计(W/L)。 + + 锚快照(开账时逐题对错)与当前全 val 重跑逐题对错各经 unit_correctness_view + 折叠成单元视图,再走 pair_block 计单元翻转(AR pair 双向 AND,不被 P/Q 单题 + 计分污染,核心算法保真 #5)。 + + 参数: + val_units: 该题型的 val 单元列表。 + snapshot: 开账时逐题对错快照(question_id -> bool)。 + rows: 当前全 val 重跑逐题预测行(question_id -> 规范化行)。 + task_type: 题型(错误信息用)。 + eval_run_id: 全 val 重跑 run_id(错误信息用)。 + + 返回: + PairResult(单元级 W/L 与 observed)。 + + 异常: + RuntimeError: 重跑缺某 val 题的预测行。 + """ + cur_per_q: dict[str, bool] = {} + for q in (q for u in val_units for q in u.questions): + row = rows.get(q.question_id) + if row is None: + raise RuntimeError( + f"probation 结算缺预测行: {task_type}/{q.question_id}(run={eval_run_id})" + ) + cur_per_q[q.question_id] = row["_correct"] + snap_units = unit_correctness_view(val_units, snapshot) + cur_units = unit_correctness_view(val_units, cur_per_q) + return pair_block(snap_units, cur_units, [u.unit_id for u in val_units]) + + def _outcome_to_quadrant_pairs(task_type: str, outcome: ValidationOutcome) -> list[dict]: - """把 ValidationOutcome 的四象限拍平为逐题 pair(供 quadrant_pair 表落库观测)。 + """把 ValidationOutcome 的四象限拍平为单元 pair(供 quadrant_pair 表落库观测)。 + + 四象限 id 为 **unit_id 口径**(single 即 question_id、AR pair 为 pair_id), + 与 gate e-process 同粒度;question_id 字段承载 unit_id。 参数: task_type: 该批 gate 的任务类型。 outcome: 局部验证决策结果。 返回: - 每条含 question_id/task_type/prev_correct/curr_correct/category 的 dict 列表。 + 每条含 question_id(=unit_id)/task_type/prev_correct/curr_correct/category 的 dict。 """ from app.harness.momentum import ( IMPROVED, @@ -1294,7 +1339,7 @@ class Runner: self._writeback_val_correctness(eval_r.run_id, pools, state) # Phase 4: probation 结算 - self._settle_probations(eval_r.run_id, state) + self._settle_probations(eval_r.run_id, pools, state) # Phase 5: best argmax self._maybe_promote_best( @@ -1382,15 +1427,20 @@ class Runner: # 慢更新内部方法 # ----------------------------------------------------------------------- - def _settle_probations(self, eval_run_id: str, state: _TrainState) -> None: - """epoch 末试用期一次性结算:全 val 重跑逐题结果与锚快照配对。 + def _settle_probations(self, eval_run_id: str, pools: Pools, state: _TrainState) -> None: + """epoch 末试用期一次性结算:全 val 重跑结果按 unit 与锚快照配对。 + + W/L 按 **unit 口径** 统计(AR pair 双向 AND 折叠,不被 P/Q 单题计分污染, + 核心算法保真 #5)。锚快照与当前重跑均先经 unit_correctness_view 折叠成单元 + 视图再走 pair_block 计翻转。逐题 predictions 仍逐题落库溯源。 参数: eval_run_id: 本 epoch 全 val 重跑(R)的 run_id。 + pools: 冻结三池(按 task_type 取 val 子集重建单元)。 state: 训练状态(probations 结算后清空)。 异常: - RuntimeError: 重跑缺某快照题的预测行。 + RuntimeError: 重跑缺某 val 题的预测行。 """ if not state.probations: return @@ -1410,20 +1460,12 @@ class Runner: ) for task_type in sorted(state.probations): probation = state.probations[task_type] - w = l = 0 # noqa: E741 - for qid, snap_correct in probation.correctness_snapshot.items(): - row = rows.get(qid) - if row is None: - raise RuntimeError( - f"probation 结算缺预测行: {task_type}/{qid}(run={eval_run_id})" - ) - cur = row["_correct"] - if not snap_correct and cur: - w += 1 - elif snap_correct and not cur: - l += 1 # noqa: E741 - verdict = probation_verdict(w, l, params=params) - logger.info("probation 结算[{}]: W={} L={} → {}", task_type, w, l, verdict) + val_units = build_units([q for q in pools.validation if q.task_type == task_type]) + flips = _probation_unit_flips( + val_units, probation.correctness_snapshot, rows, task_type, eval_run_id + ) + verdict = probation_verdict(flips.w, flips.l, params=params) + logger.info("probation 结算[{}]: W={} L={} → {}", task_type, flips.w, flips.l, verdict) if verdict == "rollback": self._rollback_probation(probation, state) state.probations.clear() diff --git a/app/harness/validate.py b/app/harness/validate.py index d4878b2..2ba2191 100644 --- a/app/harness/validate.py +++ b/app/harness/validate.py @@ -25,6 +25,7 @@ from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable from loguru import logger from app.harness.gate_ladder import BaselineCache, skill_hash +from app.harness.question_units import build_units, flatten_units, unit_correctness_view from core.evolution import ( GateParams, GateVerdict, @@ -37,7 +38,7 @@ from core.evolution import ( if TYPE_CHECKING: from app.harness.inference import InferenceResult from app.harness.log import HarnessLog - from core.types import GeneratedQuestion + from core.types import GeneratedQuestion, QuestionUnit # gate_decision 的 decision → ValidationOutcome.stop_reason 映射 @@ -95,10 +96,12 @@ class InferenceRunConfig: @dataclass class ValidationOutcome: - """CE-Gate 局部验证结果:三态动作 + e-process 证据 + 已观测题逐题对错。 + """CE-Gate 局部验证结果:三态动作 + e-process 证据(单元口径)+ 逐题溯源对错。 - correctness 二轨语义:candidate_correctness 只含已观测题(早停后是 - 阶梯前缀子集);accept 时由 runner 按题粒度增量合并进 state.correctness。 + correctness 二轨语义:W/L、准确率、四象限均按 **unit 口径** 统计 + (AR pair 双向 AND 折叠为一个单元,不被 P/Q 单题计分污染); + candidate_correctness 独立保留 **逐题** 对错(只含已观测题,早停后是阶梯前缀 + 子集),accept 时由 runner 按 question_id 粒度增量合并进 state.correctness。 """ action: str # accept_confirmed | accept_provisional | reject @@ -110,8 +113,8 @@ class ValidationOutcome: n_used: int delta_hat: float delta_shrunk: float - baseline_acc: float # 已观测题上的基线准确率(观测口径) - candidate_acc: float # 已观测题上的候选准确率(观测口径) + baseline_acc: float # 已观测单元上的基线准确率(unit 口径) + candidate_acc: float # 已观测单元上的候选准确率(unit 口径) improvements: list[str] = field(default_factory=list) regressions: list[str] = field(default_factory=list) persistent_fails: list[str] = field(default_factory=list) @@ -252,7 +255,7 @@ def _candidate_correctness_from_db( async def _resolve_baseline_block( - chunk: list[GeneratedQuestion], + units: list[QuestionUnit], task_type: str, s_hash: str, prompts_version: str, @@ -262,102 +265,114 @@ async def _resolve_baseline_block( log: HarnessLog, run_id: str, ) -> tuple[dict[str, bool], int, int]: - """基线侧处理一个块:缓存优先,miss 的题新鲜跑基线版本并回写缓存。 + """基线侧处理一个块:缓存优先(unit 键),miss 的单元新鲜跑基线版本并回写缓存。 + + 缓存以 unit_id 为键、存单元级对错(AR pair 双向 AND 折叠后一个布尔)。 + miss 的单元展开为逐题送推理,读回逐题预测后经 unit_correctness_view 折叠成 + 单元级对错再写缓存(核心算法保真 #5)。逐题 predictions 仍逐题落库溯源。 参数: - chunk: 当前块的题目列表。 + units: 当前块的单元列表(single 或 AR pair)。 task_type: 当前验证题型(缓存键成分)。 s_hash: 基线侧生效 skill 的内容哈希(缓存键成分)。 prompts_version: 当前 prompts 版本(缓存键成分)。 - baseline_cache: 基线侧逐题对错缓存。 + baseline_cache: 基线侧单元级对错缓存(键含 unit_id)。 base_skills_dir: 基线 skills 版本目录。 run_inference: 注入的 async 推理函数。 log: HarnessLog 共享实例(推理后读预测)。 run_id: 本块基线 run_id。 返回: - (b_map, errors_inc, denom_inc):块内 question_id -> 基线对错、 + (b_units, errors_inc, denom_inc):块内 unit_id -> 基线单元对错、 本块新增的 INFRA error 计数与推理题次分母增量(全命中时为 0, 0)。 """ - misses = [ - q - for q in chunk - if baseline_cache.get(task_type, s_hash, prompts_version, q.question_id) is None + miss_units = [ + u + for u in units + if baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id) is None ] errors_inc = 0 denom_inc = 0 - if misses: - r_b = await run_inference(misses, run_id=run_id, skills_dir=base_skills_dir) + if miss_units: + miss_questions = flatten_units(miss_units) + r_b = await run_inference(miss_questions, run_id=run_id, skills_dir=base_skills_dir) errors_inc = r_b.stop_reason_counts.get("error", 0) denom_inc = r_b.total - fresh = _candidate_correctness_from_db(log, r_b.run_id, misses) - for qid, correct in fresh.items(): - baseline_cache.put(task_type, s_hash, prompts_version, qid, correct) + fresh_per_q = _candidate_correctness_from_db(log, r_b.run_id, miss_questions) + fresh_units = unit_correctness_view(miss_units, fresh_per_q) + for uid, correct in fresh_units.items(): + baseline_cache.put(task_type, s_hash, prompts_version, uid, correct) - b_map: dict[str, bool] = {} - for q in chunk: - val = baseline_cache.get(task_type, s_hash, prompts_version, q.question_id) - assert val is not None, f"基线缓存补齐后仍有 miss: {q.question_id} run_id={run_id}" - b_map[q.question_id] = val - return b_map, errors_inc, denom_inc + b_units: dict[str, bool] = {} + for u in units: + val = baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id) + assert val is not None, f"基线缓存补齐后仍有 miss: unit={u.unit_id} run_id={run_id}" + b_units[u.unit_id] = val + return b_units, errors_inc, denom_inc async def _run_candidate_block( - chunk: list[GeneratedQuestion], + units: list[QuestionUnit], cand_dir: Path, run_inference: RunInferenceFn, log: HarnessLog, run_id: str, ) -> tuple[dict[str, bool], int, int]: - """候选侧处理一个块:全块新鲜跑候选版本并从 db 读逐题对错。 + """候选侧处理一个块:单元展开为逐题全块新鲜跑候选版本并从 db 读逐题对错。 + + 返回逐题对错映射(question_id -> bool),折叠为单元视图交由调用方完成, + 逐题结果同时用于 candidate_correctness 溯源与二轨 correctness 合并。 参数: - chunk: 当前块的题目列表。 + units: 当前块的单元列表。 cand_dir: 已物化的候选 skills 目录。 run_inference: 注入的 async 推理函数。 log: HarnessLog 共享实例(推理后读预测)。 run_id: 本块候选 run_id。 返回: - (c_map, errors_inc, denom_inc)。 + (c_per_q, errors_inc, denom_inc):块内 question_id -> 候选对错。 """ - r_c = await run_inference(chunk, run_id=run_id, skills_dir=cand_dir) - c_map = _candidate_correctness_from_db(log, r_c.run_id, chunk) - return c_map, r_c.stop_reason_counts.get("error", 0), r_c.total + questions = flatten_units(units) + r_c = await run_inference(questions, run_id=run_id, skills_dir=cand_dir) + c_per_q = _candidate_correctness_from_db(log, r_c.run_id, questions) + return c_per_q, r_c.stop_reason_counts.get("error", 0), r_c.total def _build_evidence_rows( - chunk: list[GeneratedQuestion], - b_map: dict[str, bool], - c_map: dict[str, bool], + units: list[QuestionUnit], + b_units: dict[str, bool], + c_units: dict[str, bool], task_type: str, block_idx: int, ) -> list[dict]: - """组装一个块的 gate_evidence 逐题证据行。 + """组装一个块的 gate_evidence 单元级证据行。 + 证据行按 unit 口径(question_id 字段存 unit_id、correct 存单元级对错), + 与 e-process 判定同粒度;逐题预测明细仍在 predictions 表逐题溯源。 e_value 留 None 待块判定后回填,stop_reason 留空串待终态回填。 参数: - chunk: 当前块的题目列表。 - b_map: 块内 question_id -> 基线对错。 - c_map: 块内 question_id -> 候选对错。 + units: 当前块的单元列表。 + b_units: 块内 unit_id -> 基线单元对错。 + c_units: 块内 unit_id -> 候选单元对错。 task_type: 当前验证题型。 block_idx: 当前块序号。 返回: - 逐题证据行列表。 + 单元级证据行列表。 """ return [ { - "question_id": q.question_id, + "question_id": u.unit_id, "task_type": task_type, "block_idx": block_idx, - "baseline_correct": b_map[q.question_id], - "candidate_correct": c_map[q.question_id], + "baseline_correct": b_units[u.unit_id], + "candidate_correct": c_units[u.unit_id], "e_value": None, "stop_reason": "", } - for q in chunk + for u in units ] @@ -394,20 +409,25 @@ def _finalize_outcome( n_plan: int, base_obs: dict[str, bool], cand_obs: dict[str, bool], + candidate_per_q: dict[str, bool], evidence_rows: list[dict], task_type: str, ) -> ValidationOutcome: """将块循环终态判定组装为 ValidationOutcome。 + 四象限/准确率/W/L 均按单元口径(base_obs/cand_obs 为 unit_id -> bool), + candidate_correctness 独立保留逐题溯源(供 runner 二轨合并进 state.correctness)。 + 参数: verdict: 最后一块的 gate 判定结果。 - w: 累计 W(基线错→候选对翻转)。 - l: 累计 L(基线对→候选错翻转)。 - n_used: 已消费的阶梯题数。 - n_plan: 阶梯总题数。 - base_obs: 累计基线已观测对错。 - cand_obs: 累计候选已观测对错。 - evidence_rows: 逐题证据行。 + w: 累计 W(基线错→候选对单元翻转)。 + l: 累计 L(基线对→候选错单元翻转)。 + n_used: 已消费的阶梯单元数。 + n_plan: 阶梯总单元数。 + base_obs: 累计基线已观测单元对错(unit_id -> bool)。 + cand_obs: 累计候选已观测单元对错(unit_id -> bool)。 + candidate_per_q: 累计候选逐题对错(question_id -> bool,溯源用)。 + evidence_rows: 单元级证据行。 task_type: 验证题型(日志用)。 返回: @@ -418,16 +438,16 @@ def _finalize_outcome( "accept_provisional": "accept_provisional", }.get(verdict.decision, "reject") stop_reason = _STOP_REASON_BY_DECISION[verdict.decision] - # 只有终态题的证据行才携带 stop_reason + # 只有终态单元的证据行才携带 stop_reason evidence_rows[-1]["stop_reason"] = stop_reason - quadrants = classify_quadrants({qid: (base_obs[qid], cand_obs[qid]) for qid in base_obs}) + quadrants = classify_quadrants({uid: (base_obs[uid], cand_obs[uid]) for uid in base_obs}) baseline_acc = sum(base_obs.values()) / len(base_obs) candidate_acc = sum(cand_obs.values()) / len(cand_obs) accepted = action != "reject" logger.info( - "gate 局部验证[{}]: 基线{:.1%} → 候选{:.1%} (W={} L={} E={:.2f} n={}/{}) {}", + "gate 局部验证[{}]: 基线{:.1%} → 候选{:.1%} (W={} L={} E={:.2f} n={}/{} 单元) {}", task_type, baseline_acc, candidate_acc, @@ -455,7 +475,7 @@ def _finalize_outcome( regressions=quadrants.regressions, persistent_fails=quadrants.persistent_fails, stable_successes=quadrants.stable_successes, - candidate_correctness=cand_obs, + candidate_correctness=candidate_per_q, evidence_rows=evidence_rows, ) @@ -465,13 +485,33 @@ def _finalize_outcome( # --------------------------------------------------------------------------- +def _ladder_units(ladder_items: list[GeneratedQuestion]) -> list[QuestionUnit]: + """把阶梯题序聚合为单元并保持信息阶梯顺序(按单元最早出现位置排序)。 + + build_units 会把 single 与 pair 分组重排(single 先、pair 后),破坏"难题优先" + 的阶梯序;此处按单元内题目在 ladder 中的最早下标重排,恢复原阶梯优先级, + 保证 AR pair 折叠不改变 e-process 的出题顺序(核心算法保真 #5)。非 AR 全 single + 时排序为恒等(unit_id 等于 question_id、位置即原序),与迁移前逐题行为一致。 + + 参数: + ladder_items: 阶梯出题序(可混含 single 与 AR pair 成员)。 + + 返回: + 按阶梯序排列的单元列表。 + """ + units = build_units(ladder_items) + position = {q.question_id: i for i, q in enumerate(ladder_items)} + units.sort(key=lambda u: min(position[q.question_id] for q in u.questions)) + return units + + async def _run_local_validation( workspace_dir: Path, cand_dir: Path, base_skills_version: str, task_type: str, base_skill_content: str, - plan: list[GeneratedQuestion], + units: list[QuestionUnit], gate_params: GateParams, gate_block: int, gate_guard_err: float, @@ -481,12 +521,12 @@ async def _run_local_validation( log: HarnessLog, gate_run_prefix: str, ) -> ValidationOutcome: - """块序贯循环主体:逐块基线(缓存优先)/候选配对推理,块间 e-process 判定。 + """块序贯循环主体:逐块基线(缓存优先)/候选按单元配对推理,块间 e-process 判定。 - 按 gate_block 切阶梯前缀,每块先补齐基线侧缓存 miss(新鲜跑基线版本 - 并逐题写 BaselineCache),再全块跑候选,配对累计 W/L 后调 gate_decision; - 非 continue 即早停。题尽时最后一块的判定即终态(n_remaining=0 走 - provisional/inertia 分支),无循环外补判。 + 按 gate_block 切**单元**前缀(AR pair 整锁在同一块,不跨块拆分),每块先补齐 + 基线侧缓存 miss(新鲜跑基线版本并按 unit_id 写 BaselineCache),再全块跑候选, + 折叠成单元视图后配对累计 W/L 调 gate_decision;非 continue 即早停。单元尽时 + 最后一块的判定即终态(n_remaining=0 走 provisional/inertia 分支),无循环外补判。 参数: workspace_dir: Workspace 根目录。 @@ -494,11 +534,11 @@ async def _run_local_validation( base_skills_version: 基线 skills 版本名。 task_type: 当前验证题型。 base_skill_content: 基线侧生效 skill 全文(skill_hash 作缓存键成分)。 - plan: 已截断到 gate_n_max 的阶梯出题序。 + units: 已截断到 gate_n_max 的阶梯单元序(single 或 AR pair)。 gate_params: e-process 判据阈值组。 - gate_block: 块大小。 + gate_block: 块大小(单位为**单元数**)。 gate_guard_err: 跨块累计 INFRA 错误率护栏(分母 >=10 才触发)。 - baseline_cache: 基线侧逐题对错缓存。 + baseline_cache: 基线侧单元级对错缓存(键含 unit_id)。 prompts_version: 当前 prompts 版本(缓存键成分)。 run_inference: 注入的 async 推理函数。 log: HarnessLog 共享实例。 @@ -508,8 +548,8 @@ async def _run_local_validation( ValidationOutcome。 关键实现: - INFRA 护栏跨块累计基线+候选两侧的 error 计数,分母(总推理题次)>=10 - 且错误率超 gate_guard_err 时直接 raise,避免坏批次污染判定。 + INFRA 护栏跨块累计基线+候选两侧的 error 计数,分母(总推理题次,仍逐题计) + >=10 且错误率超 gate_guard_err 时直接 raise,避免坏批次污染判定。 """ w = 0 l = 0 # noqa: E741 @@ -519,15 +559,17 @@ async def _run_local_validation( evidence_rows: list[dict] = [] base_obs: dict[str, bool] = {} cand_obs: dict[str, bool] = {} + candidate_per_q: dict[str, bool] = {} s_hash = skill_hash(base_skill_content) base_skills_dir = workspace_dir / "skills" / base_skills_version - chunks = [plan[i : i + gate_block] for i in range(0, len(plan), gate_block)] + unit_chunks = [units[i : i + gate_block] for i in range(0, len(units), gate_block)] + n_plan = len(units) verdict: GateVerdict | None = None - for block_idx, chunk in enumerate(chunks): + for block_idx, unit_chunk in enumerate(unit_chunks): # Phase 1: 基线侧(缓存优先,miss 新鲜跑)+ 候选侧(全块新鲜跑) - b_map, err_b, den_b = await _resolve_baseline_block( - chunk=chunk, + b_units, err_b, den_b = await _resolve_baseline_block( + units=unit_chunk, task_type=task_type, s_hash=s_hash, prompts_version=prompts_version, @@ -537,8 +579,8 @@ async def _run_local_validation( log=log, run_id=f"{gate_run_prefix}_b{block_idx}_base", ) - c_map, err_c, den_c = await _run_candidate_block( - chunk=chunk, + c_per_q, err_c, den_c = await _run_candidate_block( + units=unit_chunk, cand_dir=cand_dir, run_inference=run_inference, log=log, @@ -550,19 +592,21 @@ async def _run_local_validation( infra_denom += den_b + den_c _check_infra_guard(errors, infra_denom, gate_guard_err) - # Phase 3: 配对 + 证据行 + 块间判定 - qids = [q.question_id for q in chunk] - pair_result = pair_block(b_map, c_map, qids) - for qid, (b, c) in pair_result.observed.items(): - base_obs[qid] = b - cand_obs[qid] = c + # Phase 3: 折叠成单元视图 + 配对 + 证据行 + 块间判定 + c_units = unit_correctness_view(unit_chunk, c_per_q) + candidate_per_q.update(c_per_q) + unit_ids = [u.unit_id for u in unit_chunk] + pair_result = pair_block(b_units, c_units, unit_ids) + for uid, (b, c) in pair_result.observed.items(): + base_obs[uid] = b + cand_obs[uid] = c - block_rows = _build_evidence_rows(chunk, b_map, c_map, task_type, block_idx) + block_rows = _build_evidence_rows(unit_chunk, b_units, c_units, task_type, block_idx) w += pair_result.w l += pair_result.l # noqa: E741 - n_used += len(chunk) - verdict = gate_decision(w, l, n_used, len(plan) - n_used, params=gate_params) + n_used += len(unit_chunk) + verdict = gate_decision(w, l, n_used, n_plan - n_used, params=gate_params) for row in block_rows: row["e_value"] = verdict.e_value @@ -578,9 +622,10 @@ async def _run_local_validation( w=w, l=l, n_used=n_used, - n_plan=len(plan), + n_plan=n_plan, base_obs=base_obs, cand_obs=cand_obs, + candidate_per_q=candidate_per_q, evidence_rows=evidence_rows, task_type=task_type, ) @@ -618,10 +663,10 @@ async def validate_skill_local( (skill_hash(base_skill_content) 作 BaselineCache 键成分)。 ladder_items: 阶梯序题目列表(已排除本 step 案例包题)。 gate_params: e-process 判据阈值组。 - gate_block: 块大小。 - gate_n_max: 单 gate 题数上限。 + gate_block: 块大小(单位为**单元数**,AR pair 整锁不跨块拆)。 + gate_n_max: 单 gate 单元数上限(阶梯截断到此数量个单元)。 gate_guard_err: 跨块累计 INFRA 错误率护栏(分母 >=10 才触发)。 - baseline_cache: 基线侧逐题对错缓存。 + baseline_cache: 基线侧单元级对错缓存(键含 unit_id)。 prompts_version: 当前 prompts 版本(缓存键成分)。 run_inference: 注入的 async 推理函数(RunInferenceFn 协议)。 log: HarnessLog 共享实例(供 DB 回读逐题对错)。 @@ -629,7 +674,7 @@ async def validate_skill_local( (防泄露过滤靠它识别)。块 run_id = f"{prefix}_b{block_idx}_{arm}"。 返回: - ValidationOutcome。逐题证据记入 outcome.evidence_rows 随结果返回, + ValidationOutcome。单元级证据记入 outcome.evidence_rows 随结果返回, gate_evidence 落库由调用方(runner)负责。 """ if "_gate_" not in gate_run_prefix: @@ -637,7 +682,8 @@ async def validate_skill_local( if not ladder_items: raise ValueError(f"task_type={task_type} 阶梯为空,无法验证") - plan = ladder_items[:gate_n_max] + # 阶梯题序聚合为单元并按信息阶梯序截断到 gate_n_max 个单元(AR pair 整锁不拆) + units = _ladder_units(ladder_items)[:gate_n_max] cand_dir = materialize_candidate_skill( workspace_dir, base_skills_version, target_file, candidate_content ) @@ -648,7 +694,7 @@ async def validate_skill_local( base_skills_version=base_skills_version, task_type=task_type, base_skill_content=base_skill_content, - plan=plan, + units=units, gate_params=gate_params, gate_block=gate_block, gate_guard_err=gate_guard_err, diff --git a/core/evolution/validate.py b/core/evolution/validate.py index 5c140fd..0e1e0e8 100644 --- a/core/evolution/validate.py +++ b/core/evolution/validate.py @@ -1,9 +1,12 @@ """core/evolution/validate.py — 块验证纯决策函数。 -算法 #7(块顺序验证)的局部实现:pair_block 逐题比对基线与候选、 +算法 #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 @@ -12,24 +15,24 @@ from core.evolution.types import PairResult, QuadrantClassification def pair_block( baseline: dict[str, bool], candidate: dict[str, bool], - question_ids: list[str], + unit_ids: list[str], ) -> PairResult: - """逐题比对基线与候选对错,统计翻转。 + """按单元比对基线与候选对错,统计翻转。 参数: - baseline: 基线臂每题正确性映射。 - candidate: 候选臂每题正确性映射。 - question_ids: 参与比对的题目 ID 列表。 + baseline: 基线臂单元级正确性映射(unit_id → bool)。 + candidate: 候选臂单元级正确性映射(unit_id → bool)。 + unit_ids: 参与比对的单元 ID 列表(AR pair 折叠后为单一 unit_id)。 返回: PairResult,包含 w(基线错→候选对翻转数)、l(基线对→候选错翻转数) - 和 observed(每题的 (基线, 候选) 对错记录)。 + 和 observed(每单元的 (基线, 候选) 对错记录)。 """ w = l = 0 # noqa: E741 — 数学记号 W/L(win/loss),与 gate.py 一致 observed: dict[str, tuple[bool, bool]] = {} - for qid in question_ids: - b, c = baseline[qid], candidate[qid] - observed[qid] = (b, c) + 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: @@ -71,15 +74,15 @@ def classify_quadrants( def compute_accuracy( correctness: dict[str, bool], - question_ids: list[str], + unit_ids: list[str], ) -> float: - """纯算术:sum(correct) / len(ids)。 + """纯算术:sum(correct) / len(units),分母按单元数(非逐题)。 参数: - correctness: 每题正确性映射。 - question_ids: 参与计算的题目 ID 列表。 + correctness: 单元级正确性映射(unit_id → bool)。 + unit_ids: 参与计算的单元 ID 列表。 返回: - 准确率浮点数。question_ids 为空时抛出 ZeroDivisionError。 + 准确率浮点数。unit_ids 为空时抛出 ZeroDivisionError。 """ - return sum(correctness[qid] for qid in question_ids) / len(question_ids) + return sum(correctness[uid] for uid in unit_ids) / len(unit_ids) diff --git a/tests/unit/test_correctness_unit_view.py b/tests/unit/test_correctness_unit_view.py new file mode 100644 index 0000000..2b402a1 --- /dev/null +++ b/tests/unit/test_correctness_unit_view.py @@ -0,0 +1,131 @@ +"""tests/unit/test_correctness_unit_view.py — correctness 三对象口径单元测试。 + +验证 unit_correctness_view(逐题 per_q → unit 折叠:AR pair 双向 AND、非 AR single) +及 core.evolution 的 pair_block / compute_accuracy 消费 unit 口径时,混格池中 +AR pair 折叠为单元、W/L 与准确率分母不被 P/Q 单题计分污染(核心算法保真 #5)。 +""" + +from __future__ import annotations + +from app.harness.question_units import build_units, unit_correctness_view +from core.evolution.validate import classify_quadrants, compute_accuracy, pair_block +from core.types import GeneratedQuestion + + +def _single(qid: str, task_type: str = "temporal") -> GeneratedQuestion: + """构造一条非 AR single 题(unit_id 回填为 question_id)。""" + return GeneratedQuestion( + question_id=qid, + video_id=f"v_{qid}", + task_type=task_type, + question="Q?", + options=("A", "B", "C", "D"), + answer="A", + source_nodes=(), + difficulty="easy", + ) + + +def _pair(pair_id: str, task_type: str = "temporal") -> tuple[GeneratedQuestion, GeneratedQuestion]: + """构造一个 AR 孪生对(original + mirror,共享 pair_id → unit_id=pair_id)。""" + common = { + "video_id": f"v_{pair_id}", + "task_type": task_type, + "question": "Q?", + "options": ("A", "B", "C", "D"), + "answer": "A", + "source_nodes": (), + "difficulty": "easy", + "pair_id": pair_id, + "flip_axis": "before_after", + } + orig = GeneratedQuestion(question_id=f"{pair_id}_o", question_role="pair_original", **common) + mirror = GeneratedQuestion(question_id=f"{pair_id}_m", question_role="pair_mirror", **common) + return orig, mirror + + +class TestUnitCorrectnessView: + """unit_correctness_view:逐题对错折叠成 unit_id → bool。""" + + def test_mixed_pool_and_semantics(self) -> None: + """混格:single 直取、AR pair 双向 AND。""" + s0 = _single("s0") + s1 = _single("s1") + p1o, p1m = _pair("p1") + p2o, p2m = _pair("p2") + units = build_units([s0, s1, p1o, p1m, p2o, p2m]) + per_q = { + "s0": True, + "s1": False, + "p1_o": True, + "p1_m": True, + "p2_o": True, + "p2_m": False, + } + view = unit_correctness_view(units, per_q) + # single 的 unit_id 等于 question_id;pair 的 unit_id 等于 pair_id + assert view == {"s0": True, "s1": False, "p1": True, "p2": False} + + def test_missing_per_q_raises(self) -> None: + """任一题缺 per_q → KeyError(禁静默兜底)。""" + p1o, p1m = _pair("p1") + units = build_units([p1o, p1m]) + import pytest + + with pytest.raises(KeyError): + unit_correctness_view(units, {"p1_o": True}) + + +class TestPairBlockUnitFold: + """pair_block 消费 unit 口径:混格 W/L 不被 P/Q 单题计分污染。""" + + def test_pair_partial_improvement_not_counted(self) -> None: + """AR pair 基线(F,F)→候选(T,F):单元仍错,不计 W(保真 #5)。""" + p1o, p1m = _pair("p1") + s0 = _single("s0") + units = build_units([p1o, p1m, s0]) + unit_ids = [u.unit_id for u in units] + baseline_per_q = {"p1_o": False, "p1_m": False, "s0": False} + candidate_per_q = {"p1_o": True, "p1_m": False, "s0": True} + b_units = unit_correctness_view(units, baseline_per_q) + c_units = unit_correctness_view(units, candidate_per_q) + result = pair_block(b_units, c_units, unit_ids) + # 只有 s0 单元发生 F→T 翻转;pair 单元双向 AND 后仍错,不计 W + assert result.w == 1 + assert result.l == 0 + assert result.observed["p1"] == (False, False) + assert result.observed["s0"] == (False, True) + + def test_pair_full_flip_counts_once(self) -> None: + """AR pair 两题齐翻(F,F)→(T,T):单元计 1 次 W(非 2)。""" + p1o, p1m = _pair("p1") + units = build_units([p1o, p1m]) + b_units = unit_correctness_view(units, {"p1_o": False, "p1_m": False}) + c_units = unit_correctness_view(units, {"p1_o": True, "p1_m": True}) + result = pair_block(b_units, c_units, [u.unit_id for u in units]) + assert result.w == 1 + assert result.l == 0 + + +class TestComputeAccuracyUnitDenominator: + """compute_accuracy 分母按 unit 数(非逐题)。""" + + def test_denominator_is_unit_count(self) -> None: + """1 pair(错) + 1 single(对) → 1/2;逐题会误算 2/3。""" + p1o, p1m = _pair("p1") + s0 = _single("s0") + units = build_units([p1o, p1m, s0]) + unit_ids = [u.unit_id for u in units] + view = unit_correctness_view(units, {"p1_o": True, "p1_m": False, "s0": True}) + assert compute_accuracy(view, unit_ids) == 0.5 + + +class TestClassifyQuadrantsUnitKeys: + """classify_quadrants 按 unit_id 分桶(pair 单元只出现一次)。""" + + def test_pair_unit_single_bucket(self) -> None: + """pair 单元 F→F 落 persistent_fails,只记 unit_id 一次。""" + observed = {"p1": (False, False), "s0": (False, True)} + qc = classify_quadrants(observed) + assert qc.improvements == ["s0"] + assert qc.persistent_fails == ["p1"] diff --git a/tests/unit/test_gate_block_unit.py b/tests/unit/test_gate_block_unit.py new file mode 100644 index 0000000..e72f688 --- /dev/null +++ b/tests/unit/test_gate_block_unit.py @@ -0,0 +1,279 @@ +"""tests/unit/test_gate_block_unit.py — gate 块实际执行路径按 unit 跑。 + +针对 app/harness/validate.py::validate_skill_local(真实 gate 执行路径), +断言混格阶梯下 gate 块按 unit 口径运行:baseline_cache 键含 unit_id、 +n_used 按 unit 累加、pair_block 折叠 AR pair、逐题 predictions 仍溯源。 +核心算法保真 #5(信息阶梯 e-process 口径从 question_id 迁至 unit_id)。 +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest + +from app.harness.gate_ladder import BaselineCache, skill_hash +from app.harness.inference import PREDICTIONS_SCHEMA, InferenceResult +from app.harness.log import HarnessLog +from app.harness.validate import validate_skill_local +from core.evolution import GateParams +from core.types import GeneratedQuestion + +if TYPE_CHECKING: + from pathlib import Path + +_DEFAULT_GATE_PARAMS = GateParams( + e_confirm=20.0, + e_provisional=3.0, + w_net_min=2, + delta_min=0.05, + lambda_dir=-2.0, + e_rollback=10.0, +) + + +def _single(qid: str) -> GeneratedQuestion: + """构造非 AR single 题。""" + return GeneratedQuestion( + question_id=qid, + video_id=f"v_{qid}", + task_type="temporal", + question="Q?", + options=("A", "B", "C", "D"), + answer="A", + source_nodes=(), + difficulty="easy", + ) + + +def _pair(pair_id: str) -> list[GeneratedQuestion]: + """构造 AR 孪生对(original + mirror,共享 pair_id)。""" + common = { + "video_id": f"v_{pair_id}", + "task_type": "temporal", + "question": "Q?", + "options": ("A", "B", "C", "D"), + "answer": "A", + "source_nodes": (), + "difficulty": "easy", + "pair_id": pair_id, + "flip_axis": "before_after", + } + return [ + GeneratedQuestion(question_id=f"{pair_id}_o", question_role="pair_original", **common), + GeneratedQuestion(question_id=f"{pair_id}_m", question_role="pair_mirror", **common), + ] + + +def _setup_workspace(tmp_path: Path) -> Path: + """构建最小 workspace(skills/v1/temporal.md)。""" + skills_dir = tmp_path / "skills" / "v1" + skills_dir.mkdir(parents=True) + (skills_dir / "temporal.md").write_text("baseline skill content", encoding="utf-8") + return tmp_path + + +def _make_log(workspace: Path, run_id: str = "test_master") -> HarnessLog: + """创建 HarnessLog 并建 predictions 表。""" + log = HarnessLog(str(workspace / "harness.db"), run_id) + log.create_table("predictions", PREDICTIONS_SCHEMA) + return log + + +def _insert_predictions(log: HarnessLog, run_id: str, per_q: dict[str, bool]) -> None: + """逐题写 predictions(溯源仍逐题)。""" + for qid, correct in per_q.items(): + log.insert( + "predictions", + { + "run_id": run_id, + "video_id": "v0", + "question_id": qid, + "task_type": "temporal", + "prediction": "A" if correct else "Z", + "answer": "A", + "evidence": "", + "reasoning": "", + "steps_used": 1, + "prompt_tokens": 10, + "completion_tokens": 10, + "stop_reason": "completed", + "steps_json": "[]", + }, + ) + + +def _make_mock_run_inference( + log: HarnessLog, + baseline_correct: dict[str, bool], + candidate_correct: dict[str, bool], +): + """构建 mock RunInferenceFn,按 run_id 的 arm 后缀选基线/候选逐题对错。""" + call_log: list[dict[str, Any]] = [] + + async def mock_fn( + questions: list[GeneratedQuestion], + *, + run_id: str, + skills_dir: Path, + ) -> InferenceResult: + src = baseline_correct if run_id.endswith("_base") else candidate_correct + per_q = {q.question_id: src.get(q.question_id, False) for q in questions} + _insert_predictions(log, run_id, per_q) + call_log.append({"run_id": run_id, "qids": [q.question_id for q in questions]}) + total = len(questions) + return InferenceResult( + run_id=run_id, + accuracy=sum(per_q.values()) / total if total else 0.0, + total=total, + correct=sum(per_q.values()), + per_task_type={}, + steps_mean=1.0, + token_usage={"prompt_tokens": 10, "completion_tokens": 10}, + stop_reason_counts={"completed": total}, + ) + + return mock_fn, call_log + + +@pytest.mark.asyncio +async def test_gate_n_used_counts_units_not_questions(tmp_path: Path) -> None: + """混格阶梯(1 pair + 2 single)→ n_used=3 单元,非 4 题。""" + workspace = _setup_workspace(tmp_path) + log = _make_log(workspace) + cache = BaselineCache(workspace / "baseline_cache.json") + ladder = [*_pair("p1"), _single("s0"), _single("s1")] + + # 基线全错、候选全对 → 3 单元齐翻 W=3 + baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False} + candidate = {"p1_o": True, "p1_m": True, "s0": True, "s1": True} + mock_fn, call_log = _make_mock_run_inference(log, baseline, candidate) + + accept_params = GateParams( + e_confirm=15.0, + e_provisional=3.0, + w_net_min=2, + delta_min=0.05, + lambda_dir=-2.0, + e_rollback=10.0, + ) + try: + outcome = await validate_skill_local( + workspace_dir=workspace, + base_skills_version="v1", + task_type="temporal", + target_file="temporal.md", + candidate_content="improved skill", + base_skill_content="baseline skill content", + ladder_items=ladder, + gate_params=accept_params, + gate_block=10, + gate_n_max=20, + gate_guard_err=0.5, + baseline_cache=cache, + prompts_version="p1", + run_inference=mock_fn, + log=log, + gate_run_prefix="step1_gate_test", + ) + # n_used 按 unit 计(3),W 按 unit 计(3) + assert outcome.n_used == 3 + assert outcome.w == 3 + assert outcome.l == 0 + # 证据行按 unit 口径(3 行) + assert len(outcome.evidence_rows) == 3 + # baseline_cache 键含 unit_id:pair 用 pair_id、single 用 question_id + s_hash = skill_hash("baseline skill content") + assert cache.get("temporal", s_hash, "p1", "p1") is False + assert cache.get("temporal", s_hash, "p1", "s0") is False + # 逐题 question_id 不作为 baseline_cache 键(pair 成员未单独缓存) + assert cache.get("temporal", s_hash, "p1", "p1_o") is None + # 逐题 predictions 仍溯源:候选 per-q 含两 pair 成员 + assert set(outcome.candidate_correctness) >= {"p1_o", "p1_m", "s0", "s1"} + finally: + log.close() + + +@pytest.mark.asyncio +async def test_gate_pair_partial_flip_not_counted(tmp_path: Path) -> None: + """AR pair 候选仅单向翻(T,F)→单元仍错,W 不被单题污染。""" + workspace = _setup_workspace(tmp_path) + log = _make_log(workspace) + cache = BaselineCache(workspace / "baseline_cache.json") + ladder = [*_pair("p1"), _single("s0")] + + baseline = {"p1_o": False, "p1_m": False, "s0": False} + # pair 只翻一半(p1_o 对、p1_m 错)→ 单元 AND 仍错;s0 翻对 + candidate = {"p1_o": True, "p1_m": False, "s0": True} + mock_fn, _ = _make_mock_run_inference(log, baseline, candidate) + + try: + outcome = await validate_skill_local( + workspace_dir=workspace, + base_skills_version="v1", + task_type="temporal", + target_file="temporal.md", + candidate_content="improved skill", + base_skill_content="baseline skill content", + ladder_items=ladder, + gate_params=_DEFAULT_GATE_PARAMS, + gate_block=10, + gate_n_max=20, + gate_guard_err=0.5, + baseline_cache=cache, + prompts_version="p1", + run_inference=mock_fn, + log=log, + gate_run_prefix="step1_gate_test", + ) + # 只有 s0 单元翻转,pair 单元不计 W(保真 #5:不被 P/Q 单题污染) + assert outcome.w == 1 + assert outcome.l == 0 + assert outcome.n_used == 2 + # candidate_acc 分母按 unit(2 单元,1 对)→ 0.5 + assert outcome.candidate_acc == 0.5 + finally: + log.close() + + +@pytest.mark.asyncio +async def test_gate_baseline_cache_hit_by_unit(tmp_path: Path) -> None: + """基线缓存按 unit_id 预填充 → 基线侧全命中不发起推理。""" + workspace = _setup_workspace(tmp_path) + log = _make_log(workspace) + cache = BaselineCache(workspace / "baseline_cache.json") + ladder = [*_pair("p1"), _single("s0")] + + s_hash = skill_hash("baseline skill content") + # 按 unit_id 预填充(pair→pair_id,single→question_id),全错 + cache.put("temporal", s_hash, "p1", "p1", False) + cache.put("temporal", s_hash, "p1", "s0", False) + + baseline = {"p1_o": False, "p1_m": False, "s0": False} + candidate = {"p1_o": True, "p1_m": True, "s0": True} + mock_fn, call_log = _make_mock_run_inference(log, baseline, candidate) + + try: + outcome = await validate_skill_local( + workspace_dir=workspace, + base_skills_version="v1", + task_type="temporal", + target_file="temporal.md", + candidate_content="improved skill", + base_skill_content="baseline skill content", + ladder_items=ladder, + gate_params=_DEFAULT_GATE_PARAMS, + gate_block=10, + gate_n_max=20, + gate_guard_err=0.5, + baseline_cache=cache, + prompts_version="p1", + run_inference=mock_fn, + log=log, + gate_run_prefix="step1_gate_test", + ) + base_calls = [c for c in call_log if c["run_id"].endswith("_base")] + assert base_calls == [], "unit 键全命中不应发起基线推理" + assert outcome.n_used == 2 + finally: + log.close()