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 显式契约同步修正(原断言依赖旧隐式实例注入)。
311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""tests/unit/test_gate_unit_scope.py — gate 真实执行路径按 unit 口径跑。
|
||
|
||
迁移自块序贯版 test_gate_block_unit.py（载体 validate_skill_local，Task 6 删除）：
|
||
针对 app/harness/validate.py::validate_skills_concurrent（连续并发 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 GateSpec, _ladder_units, validate_skills_concurrent
|
||
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
|
||
|
||
|
||
def _mk_spec(ladder: list[GeneratedQuestion]) -> GateSpec:
|
||
"""由混格阶梯题序构造单题型 GateSpec（units 经 _ladder_units 聚合）。"""
|
||
return GateSpec(
|
||
task_type="temporal",
|
||
target_file="temporal.md",
|
||
candidate_content="improved skill",
|
||
base_skill_content="baseline skill content",
|
||
units=tuple(_ladder_units(ladder)),
|
||
gate_run_prefix="step1_gate_test",
|
||
)
|
||
|
||
|
||
async def _run_gate(workspace: Path, spec: GateSpec, mock_fn, log: HarnessLog, cache, params):
|
||
"""跑单 spec 的 validate_skills_concurrent 并返回该题型的 outcome。"""
|
||
outcomes = await validate_skills_concurrent(
|
||
workspace_dir=workspace,
|
||
base_skills_version="v1",
|
||
specs=[spec],
|
||
gate_params=params,
|
||
gate_guard_err=0.5,
|
||
baseline_cache=cache,
|
||
prompts_version="p1",
|
||
run_inference=mock_fn,
|
||
log=log,
|
||
concurrency=8,
|
||
)
|
||
return outcomes[spec.task_type]
|
||
|
||
|
||
class TestLadderUnits:
|
||
"""_ladder_units：阶梯题序聚合为单元并保持信息阶梯序。"""
|
||
|
||
def test_pure_non_ar_is_identity(self) -> None:
|
||
"""纯非 AR 输入 → 单元序恒等（unit 序==原题序、unit_id==question_id）。"""
|
||
ladder = [_single("s0"), _single("s1"), _single("s2")]
|
||
units = _ladder_units(ladder)
|
||
assert [u.kind for u in units] == ["single", "single", "single"]
|
||
# unit_id 逐一等于原 question_id，且顺序与输入完全一致
|
||
assert [u.unit_id for u in units] == ["s0", "s1", "s2"]
|
||
assert [q.question_id for u in units for q in u.questions] == ["s0", "s1", "s2"]
|
||
|
||
def test_mixed_preserves_ladder_order(self) -> None:
|
||
"""混格交错输入 → 按单元最早出现下标重排，pair 整锁、不被 single-first 污染。"""
|
||
p1o, p1m = _pair("p1")
|
||
# 交错布置：pair 两成员分居 idx 1、3；single 分居 idx 0、2、4
|
||
ladder = [_single("s0"), p1o, _single("s1"), p1m, _single("s2")]
|
||
units = _ladder_units(ladder)
|
||
# 最早出现下标：s0=0, p1=min(1,3)=1, s1=2, s2=4 → 阶梯序 [s0,p1,s1,s2]
|
||
assert [u.unit_id for u in units] == ["s0", "p1", "s1", "s2"]
|
||
# pair 整锁为一个单元（含两成员），不被拆
|
||
p1_unit = next(u for u in units if u.unit_id == "p1")
|
||
assert p1_unit.kind == "pair"
|
||
assert {q.question_id for q in p1_unit.questions} == {"p1_o", "p1_m"}
|
||
|
||
def test_mixed_differs_from_build_units_default(self) -> None:
|
||
"""混格重排必须纠正 build_units 的 single-first 顺序（否则阶梯序被污染）。"""
|
||
from app.harness.question_units import build_units
|
||
|
||
p1o, p1m = _pair("p1")
|
||
ladder = [_single("s0"), p1o, _single("s1"), p1m, _single("s2")]
|
||
default_order = [u.unit_id for u in build_units(ladder)]
|
||
ladder_order = [u.unit_id for u in _ladder_units(ladder)]
|
||
# build_units 把 pair 排到 single 之后；_ladder_units 恢复信息阶梯序
|
||
assert default_order == ["s0", "s1", "s2", "p1"]
|
||
assert ladder_order == ["s0", "p1", "s1", "s2"]
|
||
assert ladder_order != default_order
|
||
|
||
|
||
@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, _ = _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 _run_gate(workspace, _mk_spec(ladder), mock_fn, log, cache, accept_params)
|
||
# n_used 按 unit 计（3），W 按 unit 计（3）
|
||
assert outcome.n_used == 3
|
||
assert outcome.w == 3
|
||
assert outcome.l == 0
|
||
# 证据行按 unit 口径（3 行），ladder_rank 沿阶梯序连续
|
||
assert len(outcome.evidence_rows) == 3
|
||
assert [r["ladder_rank"] for r in outcome.evidence_rows] == [0, 1, 2]
|
||
# 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 不被单题污染（迁移自块序贯版）。
|
||
|
||
前缀逐单元判定下 2 单元小阶梯会在首单元 futility 早停，观测不到 pair 语义；
|
||
补 2 个 single 拉长阶梯：4 单元中 3 个 single 翻转 → W=3（pair 不计入），
|
||
candidate_acc = 3/4。
|
||
"""
|
||
workspace = _setup_workspace(tmp_path)
|
||
log = _make_log(workspace)
|
||
cache = BaselineCache(workspace / "baseline_cache.json")
|
||
ladder = [*_pair("p1"), _single("s0"), _single("s1"), _single("s2")]
|
||
|
||
baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False, "s2": False}
|
||
# pair 只翻一半（p1_o 对、p1_m 错）→ 单元 AND 仍错；singles 全翻对
|
||
candidate = {"p1_o": True, "p1_m": False, "s0": True, "s1": True, "s2": True}
|
||
mock_fn, _ = _make_mock_run_inference(log, baseline, candidate)
|
||
|
||
try:
|
||
outcome = await _run_gate(
|
||
workspace, _mk_spec(ladder), mock_fn, log, cache, _DEFAULT_GATE_PARAMS
|
||
)
|
||
# 只有 single 单元翻转，pair 单元不计 W（保真 #5：不被 P/Q 单题污染）
|
||
assert outcome.w == 3
|
||
assert outcome.l == 0
|
||
assert outcome.n_used == 4
|
||
# candidate_acc 分母按 unit（4 单元，1 对）→ 3/4
|
||
assert outcome.candidate_acc == 0.75
|
||
finally:
|
||
log.close()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_gate_baseline_cache_hit_by_unit(tmp_path: Path) -> None:
|
||
"""基线缓存按 unit_id 预填充 → 基线侧全命中不发起推理（迁移自块序贯版）。
|
||
|
||
阶梯补长到 4 单元避免首单元 futility 早停，覆盖 pair 与 single 两种 unit 键。
|
||
"""
|
||
workspace = _setup_workspace(tmp_path)
|
||
log = _make_log(workspace)
|
||
cache = BaselineCache(workspace / "baseline_cache.json")
|
||
ladder = [*_pair("p1"), _single("s0"), _single("s1"), _single("s2")]
|
||
|
||
s_hash = skill_hash("baseline skill content")
|
||
# 按 unit_id 预填充（pair→pair_id，single→question_id），全错
|
||
for unit_id in ("p1", "s0", "s1", "s2"):
|
||
cache.put("temporal", s_hash, "p1", unit_id, False)
|
||
|
||
baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False, "s2": False}
|
||
candidate = {"p1_o": True, "p1_m": True, "s0": True, "s1": True, "s2": True}
|
||
mock_fn, call_log = _make_mock_run_inference(log, baseline, candidate)
|
||
|
||
try:
|
||
outcome = await _run_gate(
|
||
workspace, _mk_spec(ladder), mock_fn, log, cache, _DEFAULT_GATE_PARAMS
|
||
)
|
||
base_calls = [c for c in call_log if c["run_id"].endswith("_base")]
|
||
assert base_calls == [], "unit 键全命中不应发起基线推理"
|
||
assert outcome.n_used == 4
|
||
finally:
|
||
log.close()
|