From b3ba11c7a53f0fa6aa344d665a3d2e9600402ad5 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 16 Jul 2026 07:16:33 -0400 Subject: [PATCH] fix: count gate INFRA guard numerator by unit not record --- app/harness/validate.py | 32 ++++- tests/unit/test_harness_validate.py | 202 +++++++++++++++++++++------- 2 files changed, 177 insertions(+), 57 deletions(-) diff --git a/app/harness/validate.py b/app/harness/validate.py index 304d124..c55e9be 100644 --- a/app/harness/validate.py +++ b/app/harness/validate.py @@ -259,6 +259,23 @@ def _infra_question_ids_from_db( } +def _count_infra_units(units: list[QuestionUnit], infra_qids: set[str]) -> int: + """统计含 INFRA record 的 unit 数（一个 unit 任一题 INFRA 即计 1）。 + + 使护栏分子与分母（r.total，unit 粒度）同口径：AR pair 一 unit 含两 record， + 逐 record 计数会放大分子致 gate_guard_err 误触发，破坏 unit 粒度一致性 + （核心算法保真 #5/#6）。 + + 参数: + units: 当前块的单元列表（single 或 AR pair）。 + infra_qids: 本 run 中 stop_reason 属 INFRA 故障族的 question_id 集合。 + + 返回: + 含至少一题 INFRA 的 unit 数。 + """ + return sum(1 for u in units if any(q.question_id in infra_qids for q in u.questions)) + + def _candidate_correctness_from_db( log: HarnessLog, run_id: str, @@ -332,11 +349,12 @@ async def _resolve_baseline_block( 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) - # 护栏错误计数与 INFRA 判定口径一致：error + parse_error 都计入， - # 使 parse_error 风暴同样能触发 gate_guard_err 熔断（不被绕过）。 - errors_inc = sum(r_b.stop_reason_counts.get(reason, 0) for reason in _INFRA_STOP_REASONS) - denom_inc = r_b.total infra_qids = _infra_question_ids_from_db(log, r_b.run_id, miss_questions) + # 护栏分子与分母（r.total，unit 粒度）同口径：含 INFRA record 的 unit 计 1， + # 避免 AR pair（一 unit 两 record）逐 record 计数放大分子致误触发；仍涵盖 + # error + parse_error（_infra_question_ids_from_db 口径），parse_error 风暴不被绕过。 + errors_inc = _count_infra_units(miss_units, infra_qids) + denom_inc = r_b.total fresh_per_q = _candidate_correctness_from_db(log, r_b.run_id, miss_questions) fresh_units = unit_correctness_view(miss_units, fresh_per_q) # 只回写非 INFRA 单元；INFRA 单元不入缓存（不永久污染基线快照） @@ -382,8 +400,10 @@ async def _run_candidate_block( 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) - # 护栏错误计数与 INFRA 判定口径一致：error + parse_error 都计入。 - errors_inc = sum(r_c.stop_reason_counts.get(reason, 0) for reason in _INFRA_STOP_REASONS) + infra_qids = _infra_question_ids_from_db(log, r_c.run_id, questions) + # 护栏分子与分母（r.total，unit 粒度）同口径：含 INFRA record 的 unit 计 1 + # （见 _count_infra_units），涵盖 error + parse_error。 + errors_inc = _count_infra_units(units, infra_qids) return c_per_q, errors_inc, r_c.total diff --git a/tests/unit/test_harness_validate.py b/tests/unit/test_harness_validate.py index f93b641..7680d67 100644 --- a/tests/unit/test_harness_validate.py +++ b/tests/unit/test_harness_validate.py @@ -112,7 +112,6 @@ def _make_mock_run_inference( log: HarnessLog, baseline_correctness: dict[str, bool], candidate_correctness: dict[str, bool], - error_count: int = 0, ): """构建 mock RunInferenceFn。 @@ -136,9 +135,6 @@ def _make_mock_run_inference( correct = sum(per_q.values()) total = len(questions) - stop_counts: dict[str, int] = {"completed": total - error_count} - if error_count > 0: - stop_counts["error"] = error_count return InferenceResult( run_id=run_id, accuracy=correct / total if total else 0.0, @@ -147,7 +143,57 @@ def _make_mock_run_inference( per_task_type={}, steps_mean=1.0, token_usage={"prompt_tokens": 10, "completion_tokens": 10}, - stop_reason_counts=stop_counts, + stop_reason_counts={"completed": total}, + ) + + return mock_fn, call_log + + +def _make_all_infra_mock(log: HarnessLog, stop_reason: str): + """构建基线全 INFRA 的 mock：每 record 写指定 INFRA stop_reason（error/parse_error）。 + + 与真实推理一致——per-record DB stop_reason 与汇总 stop_reason_counts 同源；护栏 + 分子按 unit 从 DB 读（_infra_question_ids_from_db），故须真实落 DB。total 返回 + unit 粒度（single 时 == 题数），使护栏分子/分母同粒度。 + """ + call_log: list[dict[str, Any]] = [] + + async def mock_fn( + questions: list[GeneratedQuestion], + *, + run_id: str, + skills_dir: Path, + ) -> InferenceResult: + call_log.append({"run_id": run_id, "n": len(questions)}) + for q in questions: + log.insert( + "predictions", + { + "run_id": run_id, + "video_id": "v0", + "question_id": q.question_id, + "task_type": "temporal", + "prediction": "", + "answer": "A", + "evidence": "", + "reasoning": "", + "steps_used": 1, + "prompt_tokens": 10, + "completion_tokens": 10, + "stop_reason": stop_reason, + "steps_json": "[]", + }, + ) + total = len(questions) # 全 single → unit 数 == 题数 + return InferenceResult( + run_id=run_id, + accuracy=0.0, + total=total, + correct=0, + per_task_type={}, + steps_mean=1.0, + token_usage={"prompt_tokens": 10, "completion_tokens": 10}, + stop_reason_counts={stop_reason: total}, ) return mock_fn, call_log @@ -421,17 +467,14 @@ async def test_gate_prefix_must_contain_gate(tmp_path: Path) -> None: @pytest.mark.asyncio async def test_infra_guard_threshold(tmp_path: Path) -> None: - """推理错误率超阈值时抛 RuntimeError。""" + """推理错误率超阈值时抛 RuntimeError（护栏分子/分母 unit 同粒度）。""" workspace = _setup_workspace(tmp_path) log = _make_log(workspace) - # 需要 >=10 题次才触发 INFRA 护栏 - questions = _make_questions(6) + # 需要 >=10 unit 分母才触发护栏：12 个 single，基线全 INFRA error。 + # 首块全 INFRA → valid_chunk 空 → errors=12/denom=12=1.0>0.5 触发护栏。 + questions = _make_questions(12) cache = BaselineCache(workspace / "baseline_cache.json") - - baseline_correct = {f"q{i}": False for i in range(6)} - candidate_correct = {f"q{i}": False for i in range(6)} - # 每次 run_inference 报 error_count=5，两侧各 5 → 10/12 > 0.5 - mock_fn, _ = _make_mock_run_inference(log, baseline_correct, candidate_correct, error_count=5) + mock_fn, _ = _make_all_infra_mock(log, "error") try: with pytest.raises(RuntimeError, match="错误率过高"): @@ -444,7 +487,7 @@ async def test_infra_guard_threshold(tmp_path: Path) -> None: base_skill_content="baseline skill content", ladder_items=questions, gate_params=_DEFAULT_GATE_PARAMS, - gate_block=6, + gate_block=12, gate_n_max=20, gate_guard_err=0.5, baseline_cache=cache, @@ -575,6 +618,95 @@ async def test_baseline_infra_error_not_cached(tmp_path: Path) -> None: log.close() +@pytest.mark.asyncio +async def test_infra_guard_counts_units_not_records(tmp_path: Path) -> None: + """护栏分子按 unit 计：AR pair 两 record 全 INFRA 只计 1 个 INFRA unit（而非 2）。 + + 回归 I-3：分子此前用 stop_reason_counts 逐 record 计数，分母 denom_inc=r.total + 是 unit 粒度；AR pair（一 unit 两 record）致分子被放大、误触发 gate_guard_err。 + 分子改为"含 INFRA record 的 unit 数"后与分母同粒度（核心算法保真 #5/#6）。 + """ + from app.harness.gate_ladder import skill_hash + from app.harness.question_units import build_units + from app.harness.validate import _resolve_baseline_block + + workspace = _setup_workspace(tmp_path) + log = _make_log(workspace) + # 一个 AR pair（两成员共享 pair_id）→ build_units 折叠为 1 个 pair unit + common = { + "video_id": "vp", + "task_type": "temporal", + "question": "Q?", + "options": ("A", "B", "C", "D"), + "answer": "A", + "source_nodes": (), + "difficulty": "easy", + "pair_id": "p1", + "flip_axis": "before_after", + } + pair = [ + GeneratedQuestion(question_id="p1_o", question_role="pair_original", **common), + GeneratedQuestion(question_id="p1_m", question_role="pair_mirror", **common), + ] + units = build_units(pair) + assert len(units) == 1 # 前置：pair 折叠为 1 个 unit + cache = BaselineCache(workspace / "baseline_cache.json") + s_hash = skill_hash("baseline skill content") + + async def mock_fn(qs, *, run_id, skills_dir): + # 两 record 皆 INFRA error + for q in qs: + log.insert( + "predictions", + { + "run_id": run_id, + "video_id": "vp", + "question_id": q.question_id, + "task_type": "temporal", + "prediction": "", + "answer": "A", + "evidence": "", + "reasoning": "", + "steps_used": 1, + "prompt_tokens": 10, + "completion_tokens": 10, + "stop_reason": "error", + "steps_json": "[]", + }, + ) + # total 为 unit 粒度（1 个 pair unit）；stop_reason_counts 为 record 粒度（2） + return InferenceResult( + run_id=run_id, + accuracy=0.0, + total=1, + correct=0, + per_task_type={}, + steps_mean=1.0, + token_usage={"prompt_tokens": 20, "completion_tokens": 20}, + stop_reason_counts={"error": 2}, + ) + + try: + _b_units, valid_units, errors_inc, denom_inc = await _resolve_baseline_block( + units=units, + task_type="temporal", + s_hash=s_hash, + prompts_version="p1", + baseline_cache=cache, + base_skills_dir=workspace / "skills" / "v1", + run_inference=mock_fn, + log=log, + run_id="step1_gate_b0_base", + ) + # 分子按 unit 计：1 个 INFRA unit（不是 2 条 record）；分母同粒度 = r.total = 1 + assert errors_inc == 1 + assert denom_inc == 1 + # 整对 INFRA → 从有效单元剔除 + assert valid_units == [] + finally: + log.close() + + @pytest.mark.asyncio async def test_all_infra_ladder_raises_clear_error(tmp_path: Path) -> None: """整个阶梯所有 unit 都被判为 INFRA 排除 → 明确 RuntimeError（非误导性空阶梯断言）。""" @@ -651,43 +783,11 @@ async def test_parse_error_counts_toward_guard(tmp_path: Path) -> None: """stop_reason=parse_error 也计入护栏错误率（与 INFRA 判定口径一致）→ 超阈值熔断。""" workspace = _setup_workspace(tmp_path) log = _make_log(workspace) - questions = _make_questions(6) + # 12 个 single，基线全 parse_error（per-record 落 DB，护栏按 unit 从 DB 读）。 + # 首块全 INFRA → errors=12/denom=12=1.0>0.5 → parse_error 亦触发护栏。 + questions = _make_questions(12) cache = BaselineCache(workspace / "baseline_cache.json") - - async def mock_fn(qs, *, run_id, skills_dir): - # 逐题 stop_reason 保持 completed（不触发 per-unit INFRA 排除）， - # 但汇总 stop_reason_counts 报大量 parse_error（应计入护栏）。 - for q in qs: - log.insert( - "predictions", - { - "run_id": run_id, - "video_id": "v0", - "question_id": q.question_id, - "task_type": "temporal", - "prediction": "Z", - "answer": "A", - "evidence": "", - "reasoning": "", - "steps_used": 1, - "prompt_tokens": 10, - "completion_tokens": 10, - "stop_reason": "completed", - "steps_json": "[]", - }, - ) - total = len(qs) - # 两臂各 5 个 parse_error → 累计 10/12 > 0.5 触发护栏 - return InferenceResult( - run_id=run_id, - accuracy=0.0, - total=total, - correct=0, - per_task_type={}, - steps_mean=1.0, - token_usage={"prompt_tokens": 10, "completion_tokens": 10}, - stop_reason_counts={"completed": 1, "parse_error": 5}, - ) + mock_fn, _ = _make_all_infra_mock(log, "parse_error") try: with pytest.raises(RuntimeError, match="错误率过高"): @@ -700,7 +800,7 @@ async def test_parse_error_counts_toward_guard(tmp_path: Path) -> None: base_skill_content="baseline skill content", ladder_items=questions, gate_params=_DEFAULT_GATE_PARAMS, - gate_block=6, + gate_block=12, gate_n_max=20, gate_guard_err=0.5, baseline_cache=cache,