iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: count gate INFRA guard numerator by unit not record

Browse Source
This commit is contained in:
iomgaa
2026-07-16 07:16:33 -04:00
parent ee69721ea3
commit b3ba11c7a5
2 changed files with 177 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/harness/validate.py
+26 -6
View File
@@ -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
tests/unit/test_harness_validate.py
+151 -51
View File
@@ -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,