iomgaa
8958eee11b
config/train_videomme.yaml 同时收录待入库的实验配置变更(run_id v2 / concurrency 32 / batch_size 40)。tests/integration/test_v3_contract_e2e.py 的 run_id 断言按 Task 5 显式契约同步修正(原断言依赖旧隐式实例注入)。
934 lines
35 KiB
Python
934 lines
35 KiB
Python
"""async 连续并发 gate 验证编排 — CE-Gate 局部验证的唯一独立子编排器。
|
||
|
||
多题型全部 (单元, 臂) 任务共享题槽并发（validate_skills_concurrent），
|
||
统计推进不按到达序，而按预声明的阶梯序前缀消费（_advance_prefix）：
|
||
base 臂缓存命中瞬间返回、cand 臂必新鲜跑，两臂延迟不对称，按到达序判定
|
||
会系统性偏向早到翻转；前缀消费把判定顺序钉回阶梯序，anytime-valid 无条件
|
||
成立（核心算法保真 #6，语义修订：块序贯 → 阶梯序前缀逐对序贯）。
|
||
|
||
基线与候选在同一阶梯前缀上逐单元配对，只数翻转（基线错→候选对 = W，
|
||
基线对→候选错 = L），每消费一个单元调一次 gate_decision 做四出口判定，
|
||
过线即冻结、τ 之后的 in-flight 结果整体丢弃。基线侧单元级对错走
|
||
BaselineCache 内容寻址缓存，miss 才新鲜跑；INFRA 单元不写缓存、从配对剔除。
|
||
判定逻辑全部在 core/evolution/gate，本模块只负责推理编排与证据收集。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import shutil
|
||
import tempfile
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
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, unit_correctness_view
|
||
from core.evolution import (
|
||
INFRA_STOP_REASONS,
|
||
GateParams,
|
||
GateVerdict,
|
||
RejectedEdit,
|
||
classify_quadrants,
|
||
gate_decision,
|
||
pair_block,
|
||
)
|
||
|
||
# INFRA_STOP_REASONS 复用 core.evolution.diagnose 的单一定义（M-2）：INFRA 故障
|
||
# stop_reason（推理侧基础设施错误，非模型答错）在诊断与 gate 两处必须同口径，
|
||
# 避免各自维护副本致未来漂移。
|
||
|
||
if TYPE_CHECKING:
|
||
from app.harness.inference import InferenceResult
|
||
from app.harness.log import HarnessLog
|
||
from core.types import GeneratedQuestion, QuestionUnit
|
||
|
||
|
||
# gate_decision 的 decision → ValidationOutcome.stop_reason 映射
|
||
_STOP_REASON_BY_DECISION: dict[str, str] = {
|
||
"accept_confirmed": "confirmed",
|
||
"reject_directional": "directional",
|
||
"reject_futility": "futility",
|
||
"accept_provisional": "provisional",
|
||
"reject_inertia": "inertia",
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 注入协议
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@runtime_checkable
|
||
class RunInferenceFn(Protocol):
|
||
"""注入的推理函数协议。
|
||
|
||
调用方（runner）负责绑定 llm、tool_dispatch_fn、prompt_builder、
|
||
log、concurrency、max_steps、skill_mode 等共享依赖。
|
||
validate 侧只传 questions、run_id、skills_dir 三个逐任务变化的参数。
|
||
"""
|
||
|
||
async def __call__(
|
||
self,
|
||
questions: list[GeneratedQuestion],
|
||
*,
|
||
run_id: str,
|
||
skills_dir: Path,
|
||
) -> InferenceResult: ...
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 数据类型
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class ValidationOutcome:
|
||
"""CE-Gate 局部验证结果：三态动作 + e-process 证据（单元口径）+ 逐题溯源对错。
|
||
|
||
correctness 二轨语义：W/L、准确率、四象限均按 **unit 口径** 统计
|
||
（AR pair 双向 AND 折叠为一个单元，不被 P/Q 单题计分污染）；
|
||
candidate_correctness 独立保留 **逐题** 对错（只含已观测题，早停后是阶梯前缀
|
||
子集），accept 时由 runner 按 question_id 粒度增量合并进 state.correctness。
|
||
"""
|
||
|
||
action: str # accept_confirmed | accept_provisional | reject
|
||
accepted: bool
|
||
stop_reason: str # confirmed | directional | futility | provisional | inertia
|
||
e_value: float
|
||
w: int
|
||
l: int # noqa: E741
|
||
n_used: int
|
||
delta_hat: float
|
||
delta_shrunk: 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)
|
||
stable_successes: list[str] = field(default_factory=list)
|
||
candidate_correctness: dict[str, bool] = field(default_factory=dict)
|
||
evidence_rows: list[dict] = field(default_factory=list) # gate_evidence 逐题行，runner 落库
|
||
|
||
|
||
@dataclass
|
||
class Probation:
|
||
"""一个题型的在途试用账本（每题型至多一个）。
|
||
|
||
字段:
|
||
task_type: 题型。
|
||
anchor_skills_version: 锚版本名（最近一个 CONFIRMED 的 skills 版本）——
|
||
回滚时恢复该版本中本题型 skill 文件的内容。
|
||
target_file: 该题型解析后的 skill 文件名。
|
||
correctness_snapshot: 开账时该题型 val 题的对错快照（回滚时恢复）。
|
||
opened_step: 开账时的 global_step（观测用）。
|
||
pending_edits: 试用链上全部候选 edit 的黑名单素材（回滚时整链入黑名单）。
|
||
"""
|
||
|
||
task_type: str
|
||
anchor_skills_version: str
|
||
target_file: str
|
||
correctness_snapshot: dict[str, bool]
|
||
opened_step: int
|
||
pending_edits: list[RejectedEdit] = field(default_factory=list)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 同步辅助函数
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def materialize_candidate_skill(
|
||
workspace_dir: Path,
|
||
base_skills_version: str,
|
||
target_file: str,
|
||
content: str,
|
||
) -> Path:
|
||
"""将候选 skill 正文物化为 workspace 专用临时目录下唯一命名的候选 skills 目录。
|
||
|
||
复制基线 skills 目录到 .cand_tmp/ 下的唯一命名临时目录，然后覆写 target_file。
|
||
构建失败时尽力清理已建临时目录再重抛原始异常。
|
||
|
||
参数:
|
||
workspace_dir: Workspace 根目录。基线 skills 从 workspace_dir/skills/<base>
|
||
复制，临时候选落 workspace_dir/.cand_tmp/。
|
||
base_skills_version: 基线 skills 版本名。
|
||
target_file: 被替换的 skill 文件名。
|
||
content: 候选 skill 文件全文。
|
||
|
||
返回:
|
||
新建的临时候选目录绝对路径。
|
||
|
||
契约:
|
||
构建失败（OSError）时尽力清理已建临时目录再重抛原始异常；
|
||
清理本身失败记 warning。
|
||
"""
|
||
cand_tmp_root = workspace_dir / ".cand_tmp"
|
||
cand_tmp_root.mkdir(parents=True, exist_ok=True)
|
||
cand_dir = Path(tempfile.mkdtemp(prefix=f"{base_skills_version}_cand_", dir=cand_tmp_root))
|
||
try:
|
||
base_dir = workspace_dir / "skills" / base_skills_version
|
||
shutil.copytree(base_dir, cand_dir, dirs_exist_ok=True)
|
||
(cand_dir / target_file).write_text(content, encoding="utf-8")
|
||
except OSError:
|
||
try:
|
||
shutil.rmtree(cand_dir)
|
||
except OSError as cleanup_err:
|
||
logger.warning("候选物化失败后清理临时目录也失败 {}: {}", cand_dir, cleanup_err)
|
||
raise
|
||
return cand_dir
|
||
|
||
|
||
def _load_run_rows(
|
||
log: HarnessLog,
|
||
run_id: str,
|
||
) -> dict[str, dict[str, Any]]:
|
||
"""读取单个 run 的逐题预测行并规范化轨迹字段。
|
||
|
||
从 predictions 表读取指定 run 的题目级记录，补充 _correct
|
||
与规范化后的 steps 字段。保持同步（log.query）——仅在推理完成后调用。
|
||
|
||
参数:
|
||
log: HarnessLog 共享实例（用 query 方法做只读 SELECT）。
|
||
run_id: 待读取的预测 run_id。
|
||
|
||
返回:
|
||
以 question_id 为键的行字典。每行至少包含 prediction、answer、
|
||
_correct、steps 等字段。
|
||
"""
|
||
rows = log.query(
|
||
"SELECT question_id, prediction, answer, stop_reason, steps_json "
|
||
"FROM predictions WHERE run_id=?",
|
||
(run_id,),
|
||
)
|
||
normalized: dict[str, dict[str, Any]] = {}
|
||
for row in rows:
|
||
raw_steps = row.get("steps_json")
|
||
parsed_steps: Any = raw_steps
|
||
if isinstance(raw_steps, str):
|
||
try:
|
||
parsed_steps = json.loads(raw_steps)
|
||
except json.JSONDecodeError:
|
||
parsed_steps = []
|
||
steps = parsed_steps if isinstance(parsed_steps, list) else []
|
||
normalized[row["question_id"]] = {
|
||
**row,
|
||
"_correct": row.get("prediction") == row.get("answer"),
|
||
"steps": steps,
|
||
}
|
||
return normalized
|
||
|
||
|
||
def _infra_question_ids_from_db(
|
||
log: HarnessLog,
|
||
run_id: str,
|
||
chunk: list[GeneratedQuestion],
|
||
) -> set[str]:
|
||
"""从 db 读取一个 run 中 stop_reason 属 INFRA 故障族的 question_id 集合。
|
||
|
||
参数:
|
||
log: HarnessLog 共享实例。
|
||
run_id: 推理 run_id。
|
||
chunk: 题目列表。
|
||
|
||
返回:
|
||
stop_reason ∈ {"error", "parse_error"} 的 question_id 集合。
|
||
"""
|
||
rows = _load_run_rows(log, run_id)
|
||
return {
|
||
q.question_id
|
||
for q in chunk
|
||
if rows.get(q.question_id, {}).get("stop_reason") in INFRA_STOP_REASONS
|
||
}
|
||
|
||
|
||
def _candidate_correctness_from_db(
|
||
log: HarnessLog,
|
||
run_id: str,
|
||
chunk: list[GeneratedQuestion],
|
||
) -> dict[str, bool]:
|
||
"""从 db 读取候选/基线 run 在指定题目上的逐题对错。
|
||
|
||
参数:
|
||
log: HarnessLog 共享实例。
|
||
run_id: 推理 run_id。
|
||
chunk: 题目列表。
|
||
|
||
返回:
|
||
question_id -> 是否答对的映射。缺行的题目记为 False。
|
||
"""
|
||
rows = _load_run_rows(log, run_id)
|
||
return {q.question_id: rows.get(q.question_id, {}).get("_correct", False) for q in chunk}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# INFRA 护栏
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _check_infra_guard(errors: int, infra_denom: int, gate_guard_err: float) -> None:
|
||
"""累计 INFRA 错误率护栏：分母 >=10 且超阈值时 raise。
|
||
|
||
参数:
|
||
errors: 两侧累计 error 计数。
|
||
infra_denom: 两侧累计推理题次分母。
|
||
gate_guard_err: 错误率阈值。
|
||
|
||
异常:
|
||
RuntimeError: 错误率超阈值。
|
||
"""
|
||
if infra_denom >= 10 and errors / infra_denom > gate_guard_err:
|
||
raise RuntimeError(f"gate 推理累计错误率过高 {errors / infra_denom:.0%}，中止本轮")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 终态组装
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _finalize_outcome(
|
||
verdict: GateVerdict,
|
||
w: int,
|
||
l: int, # noqa: E741
|
||
n_used: int,
|
||
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: 累计基线已观测单元对错（unit_id -> bool）。
|
||
cand_obs: 累计候选已观测单元对错（unit_id -> bool）。
|
||
candidate_per_q: 累计候选逐题对错（question_id -> bool，溯源用）。
|
||
evidence_rows: 单元级证据行。
|
||
task_type: 验证题型（日志用）。
|
||
|
||
返回:
|
||
ValidationOutcome。
|
||
"""
|
||
action = {
|
||
"accept_confirmed": "accept_confirmed",
|
||
"accept_provisional": "accept_provisional",
|
||
}.get(verdict.decision, "reject")
|
||
stop_reason = _STOP_REASON_BY_DECISION[verdict.decision]
|
||
# 只有终态单元的证据行才携带 stop_reason
|
||
evidence_rows[-1]["stop_reason"] = stop_reason
|
||
|
||
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={}/{} 单元) {}",
|
||
task_type,
|
||
baseline_acc,
|
||
candidate_acc,
|
||
w,
|
||
l,
|
||
verdict.e_value,
|
||
n_used,
|
||
n_plan,
|
||
"接受" if accepted else "回滚",
|
||
)
|
||
|
||
return ValidationOutcome(
|
||
action=action,
|
||
accepted=accepted,
|
||
stop_reason=stop_reason,
|
||
e_value=verdict.e_value,
|
||
w=w,
|
||
l=l,
|
||
n_used=n_used,
|
||
delta_hat=verdict.delta_hat,
|
||
delta_shrunk=verdict.delta_shrunk,
|
||
baseline_acc=baseline_acc,
|
||
candidate_acc=candidate_acc,
|
||
improvements=quadrants.improvements,
|
||
regressions=quadrants.regressions,
|
||
persistent_fails=quadrants.persistent_fails,
|
||
stable_successes=quadrants.stable_successes,
|
||
candidate_correctness=candidate_per_q,
|
||
evidence_rows=evidence_rows,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主编排
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 连续并发 gate:数据结构 + 前缀消费(algo #6 语义修订:块序贯 → 阶梯序前缀逐对序贯)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GateSpec:
|
||
"""单题型 gate 验证规格(runner 装配阶段产物,调度器输入)。
|
||
|
||
字段:
|
||
task_type: 题型。
|
||
target_file: 解析后生效 skill 文件名(候选物化写此文件)。
|
||
candidate_content: 候选 skill 全文。
|
||
base_skill_content: 基线侧生效 skill 全文(skill_hash 作缓存键)。
|
||
units: 阶梯序单元元组(已排除案例单元、截断 gate_n_max);元组,装配后
|
||
不可变,防 spec.units 与 run.slots 漂移。
|
||
gate_run_prefix: run_id 前缀,必须含 "_gate_"(防泄露过滤依赖)。
|
||
"""
|
||
|
||
task_type: str
|
||
target_file: str
|
||
candidate_content: str
|
||
base_skill_content: str
|
||
units: tuple[QuestionUnit, ...]
|
||
gate_run_prefix: str
|
||
|
||
|
||
@dataclass
|
||
class _UnitSlot:
|
||
"""单个阶梯单元的双臂到达状态。
|
||
|
||
base 为单元级对错(AR pair 已折叠);cand_per_q 为逐题对错(折叠交给消费时,
|
||
以复用 unit_correctness_view 并保留逐题溯源)。INFRA 标志与结果互斥。
|
||
"""
|
||
|
||
unit: QuestionUnit
|
||
base: bool | None = None
|
||
cand_per_q: dict[str, bool] | None = None
|
||
base_infra: bool = False
|
||
cand_infra: bool = False
|
||
|
||
def resolved(self) -> bool:
|
||
"""双臂均已出结果(含 INFRA 判定)。
|
||
|
||
返回:
|
||
base 臂(结果或 INFRA)与 cand 臂(结果或 INFRA)都已到达时为 True。
|
||
"""
|
||
base_done = self.base is not None or self.base_infra
|
||
cand_done = self.cand_per_q is not None or self.cand_infra
|
||
return base_done and cand_done
|
||
|
||
def excluded(self) -> bool:
|
||
"""任一臂 INFRA 即整单元剔除(不入配对)。
|
||
|
||
返回:
|
||
base_infra 或 cand_infra 任一为 True 时为 True。
|
||
"""
|
||
return self.base_infra or self.cand_infra
|
||
|
||
|
||
@dataclass
|
||
class _GateRun:
|
||
"""单题型 gate 的运行时状态(计数器 + 前缀指针 + 证据)。"""
|
||
|
||
spec: GateSpec
|
||
slots: list[_UnitSlot]
|
||
s_hash: str
|
||
prefix_ptr: int = 0
|
||
w: int = 0
|
||
l: int = 0 # noqa: E741
|
||
n_used: int = 0
|
||
n_excluded: int = 0
|
||
errors: int = 0
|
||
infra_denom: int = 0
|
||
frozen: bool = False
|
||
verdict: GateVerdict | None = None
|
||
base_obs: dict[str, bool] = field(default_factory=dict)
|
||
cand_obs: dict[str, bool] = field(default_factory=dict)
|
||
candidate_per_q: dict[str, bool] = field(default_factory=dict)
|
||
evidence_rows: list[dict] = field(default_factory=list)
|
||
|
||
@classmethod
|
||
def from_spec(cls, spec: GateSpec) -> _GateRun:
|
||
"""由规格构造初始状态(slots 与阶梯序一一对应)。
|
||
|
||
参数:
|
||
spec: 单题型 gate 规格(units 已阶梯序)。
|
||
|
||
返回:
|
||
计数器归零、slots 逐单元初始化、s_hash 已计算的 _GateRun。
|
||
"""
|
||
return cls(
|
||
spec=spec,
|
||
slots=[_UnitSlot(unit=u) for u in spec.units],
|
||
s_hash=skill_hash(spec.base_skill_content),
|
||
)
|
||
|
||
|
||
def _advance_prefix(run: _GateRun, params: GateParams) -> None:
|
||
"""沿阶梯序消费"已配齐前缀",逐单元更新 (W,L) 并判定,过线即冻结。
|
||
|
||
统计合法性关键(设计 v3 §1 / Codex C1):严禁按到达序消费——base 臂缓存命中
|
||
瞬间返回、cand 臂必新鲜跑,两臂延迟不对称,若 cand 延迟与对错相关,早到翻转
|
||
对系统性偏向 W 型 → e-值虚高假接受。前缀消费把判定顺序钉回预声明阶梯序,
|
||
anytime-valid 无条件成立;INFRA 单元视为"已解决(剔除)"不阻塞前缀。
|
||
|
||
契约:全部单元被剔除时 verdict 保持 None、frozen 保持 False,由调度编排层
|
||
(Task 3 的 validate_skills_concurrent)检测 verdict None 并 raise
|
||
RuntimeError;本函数不负责该终态。
|
||
|
||
参数:
|
||
run: 单题型 gate 运行时状态(原地更新计数器/指针/证据)。
|
||
params: e-process 判据阈值组。
|
||
|
||
返回:
|
||
无(所有效果原地写入 run;可重复调用,已消费前缀不重复消费)。
|
||
"""
|
||
while not run.frozen and run.prefix_ptr < len(run.slots):
|
||
slot = run.slots[run.prefix_ptr]
|
||
if not slot.resolved():
|
||
return
|
||
rank = run.prefix_ptr
|
||
run.prefix_ptr += 1
|
||
if slot.excluded():
|
||
run.n_excluded += 1
|
||
# 剔除使 n_remaining 缩小,必须重判(Codex plan 审 C1):否则尾部全 INFRA
|
||
# 时 verdict 停留在 "continue",绕过题尽第四出口且 _finalize_outcome
|
||
# 查 stop_reason 映射 KeyError。n_used==0(纯前导 INFRA)时无证据可判,跳过。
|
||
if run.n_used > 0:
|
||
n_remaining = (len(run.slots) - run.n_excluded) - run.n_used
|
||
run.verdict = gate_decision(run.w, run.l, run.n_used, n_remaining, params=params)
|
||
if run.verdict.decision != "continue":
|
||
run.frozen = True
|
||
continue
|
||
uid = slot.unit.unit_id
|
||
assert slot.base is not None and slot.cand_per_q is not None, (
|
||
f"slot 未配齐即被消费: unit={slot.unit.unit_id}"
|
||
)
|
||
c_units = unit_correctness_view([slot.unit], slot.cand_per_q)
|
||
pair_result = pair_block({uid: slot.base}, c_units, [uid])
|
||
run.candidate_per_q.update(slot.cand_per_q)
|
||
for u, (b, c) in pair_result.observed.items():
|
||
run.base_obs[u] = b
|
||
run.cand_obs[u] = c
|
||
run.w += pair_result.w
|
||
run.l += pair_result.l
|
||
run.n_used += 1
|
||
n_remaining = (len(run.slots) - run.n_excluded) - run.n_used
|
||
run.verdict = gate_decision(run.w, run.l, run.n_used, n_remaining, params=params)
|
||
run.evidence_rows.append(
|
||
{
|
||
"question_id": uid,
|
||
"task_type": run.spec.task_type,
|
||
"ladder_rank": rank,
|
||
"baseline_correct": slot.base,
|
||
"candidate_correct": c_units[uid],
|
||
"e_value": run.verdict.e_value,
|
||
"stop_reason": "",
|
||
}
|
||
)
|
||
if run.verdict.decision != "continue":
|
||
run.frozen = True
|
||
|
||
|
||
class _QuestionSlots:
|
||
"""按题数计数的共享并发闸:峰值在飞请求恒 ≤ width(设计 v3 §2.4)。
|
||
|
||
多槽获取(AR pair 一单元两题)经内部锁串行化,防多任务半持有交错死锁。
|
||
本类只承诺"并发上限 + 多槽获取原子性";公平性由调用方按题型 round-robin
|
||
顺序创建任务实现(实践中 asyncio 等待队列近似先来先服务,但那不是本类契约)。
|
||
"""
|
||
|
||
def __init__(self, width: int) -> None:
|
||
"""初始化题槽闸。
|
||
|
||
参数:
|
||
width: 并发宽度(全 gate 同时在飞的题数上限),必须为正。
|
||
|
||
返回:
|
||
无。
|
||
|
||
关键实现细节:
|
||
_width 供 acquire 做超宽 fail-fast;BoundedSemaphore 使多还立即
|
||
ValueError 而非静默扩容;_acquire_lock 串行化多槽获取防交错死锁。
|
||
"""
|
||
assert width > 0, f"并发宽度必须为正: {width}"
|
||
self._width = width
|
||
# BoundedSemaphore:多还立即 ValueError 而非静默扩容(Codex 质量审 3)
|
||
self._sem = asyncio.BoundedSemaphore(width)
|
||
self._acquire_lock = asyncio.Lock()
|
||
|
||
async def acquire(self, n: int) -> None:
|
||
"""原子获取 n 个题槽。
|
||
|
||
fail-fast:n > 宽度时任务持锁等待永不满足的槽位 → 自死锁
|
||
(AR pair 单元 2 题 + width=1 的病态配置,Codex plan 审 C2),直接报错。
|
||
取消安全:半持有自动回滚——逐槽获取途中被取消(或任何 BaseException)
|
||
时,已拿到的 permit 全部归还再重抛,容量不泄漏(Codex 质量审 2)。
|
||
|
||
参数:
|
||
n: 申请的题槽数(单元内题目数,single=1 / AR pair=2)。
|
||
|
||
返回:
|
||
无(成功返回即持有 n 个槽,须与 release(n) 配对)。
|
||
|
||
异常:
|
||
ValueError: n 超过并发宽度(否则自死锁)。
|
||
"""
|
||
if n > self._width:
|
||
raise ValueError(f"单次申请题槽 {n} 超过并发宽度 {self._width},将自死锁")
|
||
async with self._acquire_lock:
|
||
got = 0
|
||
try:
|
||
for _ in range(n):
|
||
await self._sem.acquire()
|
||
got += 1
|
||
except BaseException:
|
||
for _ in range(got):
|
||
self._sem.release()
|
||
raise
|
||
|
||
def release(self, n: int) -> None:
|
||
"""归还 n 个题槽。
|
||
|
||
参数:
|
||
n: 与 acquire 对应的题槽数。
|
||
|
||
返回:
|
||
无。
|
||
|
||
关键实现细节:
|
||
底层为 BoundedSemaphore——多还(release 数超过 acquire)立即
|
||
ValueError 暴露调用方配对错误,属防御性设计。
|
||
"""
|
||
for _ in range(n):
|
||
self._sem.release()
|
||
|
||
|
||
async def _run_unit_arm(
|
||
run: _GateRun,
|
||
slot_idx: int,
|
||
arm: str,
|
||
slots: _QuestionSlots,
|
||
run_inference: RunInferenceFn,
|
||
log: HarnessLog,
|
||
baseline_cache: BaselineCache,
|
||
prompts_version: str,
|
||
base_skills_dir: Path,
|
||
cand_dir: Path,
|
||
gate_params: GateParams,
|
||
gate_guard_err: float,
|
||
) -> None:
|
||
"""执行一个 (单元, 臂) 任务:缓存/推理 → 到达登记 → 前缀消费推进。
|
||
|
||
冻结检查三次:启动时(排队任务撤销点)、获得题槽后(获槽期间被冻结)、
|
||
推理返回后(τ 之后的 in-flight 结果不计入,整体丢弃)。
|
||
base 臂缓存命中不占题槽(零推理);INFRA 单元不写缓存(不永久污染基线快照)。
|
||
护栏在每次臂完成时检查(等价迁移自跨块累计,设计 v3 §2.3),超阈值 raise
|
||
中止整轮(与现行行为一致)。
|
||
|
||
参数:
|
||
run: 该题型的 gate 运行时状态。
|
||
slot_idx: 单元在阶梯中的下标。
|
||
arm: "base" 或 "cand"。
|
||
slots: 全 gate 共享题槽闸。
|
||
run_inference: 注入推理函数。
|
||
log: HarnessLog 共享实例(推理后读预测)。
|
||
baseline_cache / prompts_version: 基线缓存及键成分。
|
||
base_skills_dir / cand_dir: 两臂各自的 skills 目录。
|
||
gate_params: e-process 判据(前缀消费用)。
|
||
gate_guard_err: INFRA 错误率护栏阈值。
|
||
|
||
返回:
|
||
无(结果写入 run.slots[slot_idx] 并触发 _advance_prefix)。
|
||
|
||
异常:
|
||
RuntimeError: 累计 INFRA 错误率超护栏阈值(经 _check_infra_guard)。
|
||
"""
|
||
assert arm in ("base", "cand")
|
||
if run.frozen:
|
||
return
|
||
slot = run.slots[slot_idx]
|
||
spec = run.spec
|
||
|
||
if arm == "base":
|
||
cached = baseline_cache.get(spec.task_type, run.s_hash, prompts_version, slot.unit.unit_id)
|
||
if cached is not None:
|
||
slot.base = cached
|
||
_advance_prefix(run, gate_params)
|
||
return
|
||
|
||
questions = list(slot.unit.questions)
|
||
await slots.acquire(len(questions))
|
||
try:
|
||
if run.frozen:
|
||
return
|
||
run_id = f"{spec.gate_run_prefix}_{arm}"
|
||
skills_dir = base_skills_dir if arm == "base" else cand_dir
|
||
r = await run_inference(questions, run_id=run_id, skills_dir=skills_dir)
|
||
# 推理 await 期间该题型可能已被其他任务判定冻结:设计语义是
|
||
# "τ(冻结时刻)之后的 in-flight 结果不计入"——整体丢弃,不写
|
||
# slot/infra_denom/errors,滞后 INFRA 也不得触发护栏 raise 掀翻
|
||
# 整轮 gather(Codex 质量审 1)。
|
||
if run.frozen:
|
||
return
|
||
_register_arm_arrival(
|
||
run=run,
|
||
slot=slot,
|
||
arm=arm,
|
||
questions=questions,
|
||
inference_run_id=r.run_id,
|
||
inference_total=r.total,
|
||
log=log,
|
||
baseline_cache=baseline_cache,
|
||
prompts_version=prompts_version,
|
||
)
|
||
_check_infra_guard(run.errors, run.infra_denom, gate_guard_err)
|
||
finally:
|
||
slots.release(len(questions))
|
||
_advance_prefix(run, gate_params)
|
||
|
||
|
||
def _register_arm_arrival(
|
||
run: _GateRun,
|
||
slot: _UnitSlot,
|
||
arm: str,
|
||
questions: list[GeneratedQuestion],
|
||
inference_run_id: str,
|
||
inference_total: int,
|
||
log: HarnessLog,
|
||
baseline_cache: BaselineCache,
|
||
prompts_version: str,
|
||
) -> None:
|
||
"""把一次臂推理结果登记进 slot 与 run 计数器(INFRA 判定 + 对错折叠 + 回写缓存)。
|
||
|
||
INFRA 臂只标记不写缓存(不永久污染基线快照);正常 base 臂折叠为单元级对错并
|
||
回写 BaselineCache,正常 cand 臂保留逐题对错(折叠交给前缀消费,保留逐题溯源)。
|
||
|
||
参数:
|
||
run: 该题型的 gate 运行时状态(errors / infra_denom 原地累加)。
|
||
slot: 本单元的双臂到达状态(结果或 INFRA 标志原地写入)。
|
||
arm: "base" 或 "cand"。
|
||
questions: 本单元展开后的题目列表。
|
||
inference_run_id: 本次推理的 run_id(DB 回读键)。
|
||
inference_total: 本次推理的题次数(护栏分母增量)。
|
||
log: HarnessLog 共享实例(推理后读预测)。
|
||
baseline_cache / prompts_version: 基线缓存及键成分。
|
||
|
||
返回:
|
||
无(所有效果原地写入 run 与 slot)。
|
||
|
||
关键实现细节:
|
||
errors 按单元级去重(Codex plan 审 I3):同一单元双臂都 INFRA 只计 1 个
|
||
error,与设计 §2.3"分子=INFRA 单元数(任一臂)"及旧块实现口径一致
|
||
(旧实现 cand 不跑 base-INFRA 单元,天然无双计)。
|
||
"""
|
||
spec = run.spec
|
||
infra_qids = _infra_question_ids_from_db(log, inference_run_id, questions)
|
||
run.infra_denom += inference_total
|
||
if infra_qids:
|
||
if not slot.excluded():
|
||
run.errors += 1
|
||
if arm == "base":
|
||
slot.base_infra = True
|
||
else:
|
||
slot.cand_infra = True
|
||
return
|
||
per_q = _candidate_correctness_from_db(log, inference_run_id, questions)
|
||
if arm == "base":
|
||
folded = unit_correctness_view([slot.unit], per_q)
|
||
slot.base = folded[slot.unit.unit_id]
|
||
baseline_cache.put(
|
||
spec.task_type, run.s_hash, prompts_version, slot.unit.unit_id, slot.base
|
||
)
|
||
else:
|
||
slot.cand_per_q = per_q
|
||
|
||
|
||
def _validate_gate_specs(specs: list[GateSpec]) -> None:
|
||
"""校验各题型 gate 规格,不合法直接报错(不兜底)。
|
||
|
||
参数:
|
||
specs: 各题型 gate 规格。
|
||
|
||
异常:
|
||
ValueError: 阶梯为空,或 gate_run_prefix 缺 "_gate_"(防泄露过滤依赖
|
||
该标记识别 gate run)。
|
||
"""
|
||
for spec in specs:
|
||
if "_gate_" not in spec.gate_run_prefix:
|
||
raise ValueError(f"gate_run_prefix 必须含 '_gate_': {spec.gate_run_prefix!r}")
|
||
if not spec.units:
|
||
raise ValueError(f"task_type={spec.task_type} 阶梯为空,无法验证")
|
||
|
||
|
||
def _cleanup_candidate_dirs(cand_dirs: dict[str, Path]) -> None:
|
||
"""尽力清理全部候选临时目录,单个失败只记 warning 不中断其余清理。
|
||
|
||
参数:
|
||
cand_dirs: task_type -> 候选临时目录路径。
|
||
|
||
返回:
|
||
无。
|
||
"""
|
||
for d in cand_dirs.values():
|
||
try:
|
||
shutil.rmtree(d)
|
||
except OSError as e:
|
||
logger.warning("候选临时目录清理失败 {}: {}", d, e)
|
||
|
||
|
||
def _build_launch_order(runs: list[_GateRun]) -> list[tuple[_GateRun, int, str]]:
|
||
"""构建 (run, rank, arm) 发射队列:题型 round-robin × 题型内阶梯序。
|
||
|
||
交错顺序 = rank 0 各题型 → rank 1 各题型 → ...;同一 (题型, rank) 内
|
||
base 先 cand 后。round-robin 让各题型的阶梯头部同批起跑,配合前缀消费
|
||
使统计推进不因某题型阶梯过长而饿死其他题型。
|
||
|
||
参数:
|
||
runs: 各题型 gate 运行时状态(slots 已按阶梯序初始化)。
|
||
|
||
返回:
|
||
(run, rank, arm) 三元组列表,即任务创建顺序。
|
||
"""
|
||
order: list[tuple[_GateRun, int, str]] = []
|
||
max_rank = max((len(r.slots) for r in runs), default=0)
|
||
for rank in range(max_rank):
|
||
for r in runs:
|
||
if rank < len(r.slots):
|
||
for arm in ("base", "cand"):
|
||
order.append((r, rank, arm))
|
||
return order
|
||
|
||
|
||
async def validate_skills_concurrent(
|
||
workspace_dir: Path,
|
||
base_skills_version: str,
|
||
specs: list[GateSpec],
|
||
gate_params: GateParams,
|
||
gate_guard_err: float,
|
||
baseline_cache: BaselineCache,
|
||
prompts_version: str,
|
||
run_inference: RunInferenceFn,
|
||
log: HarnessLog,
|
||
concurrency: int,
|
||
) -> dict[str, ValidationOutcome]:
|
||
"""连续并发 gate:多题型全部臂共享题槽并发,统计按阶梯序前缀有序推进。
|
||
|
||
关键实现细节:
|
||
发射顺序 = 题型 round-robin × 题型内阶梯序(base 先 cand 后);题型过线
|
||
即冻结,其排队任务启动时自查冻结标志撤销,in-flight 结果不计入(τ 之后
|
||
样本,合法丢弃);候选目录逐个物化即登记、统一 finally 清理(中途失败不
|
||
泄漏);任一任务异常先 cancel+排水其余任务再向上传播;全部题型判定后
|
||
统一经 _finalize_outcome 组装。
|
||
|
||
参数:
|
||
workspace_dir: workspace 根目录(候选物化用)。
|
||
base_skills_version: 基线 skills 版本名。
|
||
specs: 各题型 gate 规格(units 已阶梯序 + 截断 n_max)。
|
||
gate_params: e-process 判据阈值组。
|
||
gate_guard_err: INFRA 错误率护栏阈值。
|
||
baseline_cache: 基线侧单元级对错缓存。
|
||
prompts_version: 当前 prompts 版本(缓存键成分)。
|
||
run_inference: 注入推理函数(调用方须绑定共享 HarnessLog)。
|
||
log: HarnessLog 共享实例(推理后读预测,与 run_inference 同库)。
|
||
concurrency: 题槽宽度(峰值在飞题数上限)。
|
||
|
||
返回:
|
||
{task_type: ValidationOutcome}。
|
||
|
||
异常:
|
||
RuntimeError: INFRA 护栏超阈值,或某题型全部单元被 INFRA 排除。
|
||
ValueError: spec 校验失败(空阶梯 / run_prefix 缺 "_gate_")。
|
||
"""
|
||
_validate_gate_specs(specs)
|
||
base_skills_dir = workspace_dir / "skills" / base_skills_version
|
||
runs = [_GateRun.from_spec(s) for s in specs]
|
||
cand_dirs: dict[str, Path] = {}
|
||
slots_gate = _QuestionSlots(concurrency)
|
||
try:
|
||
# 成功一个登记一个:第 N 个题型物化抛 OSError 时,已登记的前 N-1 个
|
||
# 目录仍由 finally 统一清理,不泄漏(Codex 质量审 C001)。
|
||
for r in runs:
|
||
cand_dirs[r.spec.task_type] = materialize_candidate_skill(
|
||
workspace_dir, base_skills_version, r.spec.target_file, r.spec.candidate_content
|
||
)
|
||
tasks = [
|
||
asyncio.ensure_future(
|
||
_run_unit_arm(
|
||
r,
|
||
rank,
|
||
arm,
|
||
slots_gate,
|
||
run_inference,
|
||
log,
|
||
baseline_cache,
|
||
prompts_version,
|
||
base_skills_dir,
|
||
cand_dirs[r.spec.task_type],
|
||
gate_params,
|
||
gate_guard_err,
|
||
)
|
||
)
|
||
for r, rank, arm in _build_launch_order(runs)
|
||
]
|
||
# 护栏 raise 中止整轮的语义不变(Codex 质量审 C002):首异常先取消其余
|
||
# 任务并排水(return_exceptions 吞取消回报),确保外层 finally 删除候选
|
||
# 目录时已无在飞任务访问该目录、事件循环收尾无 pending task 警告;
|
||
# _run_unit_arm 的题槽获取自带取消回滚,cancel 安全。
|
||
try:
|
||
await asyncio.gather(*tasks)
|
||
except BaseException:
|
||
for t in tasks:
|
||
t.cancel()
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
raise
|
||
finally:
|
||
_cleanup_candidate_dirs(cand_dirs)
|
||
|
||
outcomes: dict[str, ValidationOutcome] = {}
|
||
for r in runs:
|
||
if r.verdict is None:
|
||
raise RuntimeError(
|
||
f"gate[{r.spec.task_type}] 全部 unit 被判为 INFRA 排除,无法验证(检查推理基础设施)"
|
||
)
|
||
outcomes[r.spec.task_type] = _finalize_outcome(
|
||
verdict=r.verdict,
|
||
w=r.w,
|
||
l=r.l,
|
||
n_used=r.n_used,
|
||
n_plan=len(r.slots),
|
||
base_obs=r.base_obs,
|
||
cand_obs=r.cand_obs,
|
||
candidate_per_q=r.candidate_per_q,
|
||
evidence_rows=r.evidence_rows,
|
||
task_type=r.spec.task_type,
|
||
)
|
||
return outcomes
|