diff --git a/app/harness/inference.py b/app/harness/inference.py index e18e085..6092025 100644 --- a/app/harness/inference.py +++ b/app/harness/inference.py @@ -34,17 +34,17 @@ if TYPE_CHECKING: @dataclass(frozen=True) class InferenceResult: - """推理聚合结果。 + """推理聚合结果（正确率按 unit 粒度）。 属性: run_id: 运行标识。 - accuracy: 总正确率。 - total: 总题数。 - correct: 正确题数。 - per_task_type: 按题型分组的指标 {task_type: {accuracy, total, correct}}。 - steps_mean: 平均步数。 - token_usage: token 总用量 {prompt_tokens, completion_tokens}。 - stop_reason_counts: 终止原因计数 {reason: count}。 + accuracy: unit 级正确率（correct / total）。 + total: unit 总数（single 数 + pair 数，孤儿 pair 已剔除不计入）。 + correct: 正确 unit 数（single 单题正确；pair 走 original/mirror 双向 AND）。 + per_task_type: 按题型分组的 unit 级指标 {task_type: {accuracy, total, correct}}。 + steps_mean: 平均步数（record 粒度，逐题溯源）。 + token_usage: token 总用量 {prompt_tokens, completion_tokens}（record 粒度）。 + stop_reason_counts: 终止原因计数 {reason: count}（record 粒度）。 """ run_id: str @@ -210,19 +210,38 @@ def _group_by_task_type(graded: list[tuple[QuestionUnit, bool]]) -> dict[str, di return per_task_type +def _is_valid_pair(group: list[GeneratedQuestion]) -> bool: + """判定同一 pair_id 分组是否为合法孪生对（恰好 1 original + 1 mirror，无多余）。 + + 要求分组总数恰为 2 且角色齐备唯一；有额外非法 role 记录（total>2）或角色 + 缺失/重复均视为非法，交由调用方剔除，防非法记录混入 build_units。 + + 参数: + group: 归属同一 pair_id 的题目列表。 + + 返回: + 合法孪生对为 True，否则 False。 + """ + if len(group) != 2: + return False + originals = sum(1 for q in group if q.question_role == "pair_original") + mirrors = sum(1 for q in group if q.question_role == "pair_mirror") + return originals == 1 and mirrors == 1 + + def _drop_orphan_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQuestion]: """剔除收不齐 2 条 / 角色非法的孤儿 pair，告警不静默。 每条题目均会各答一次并逐题落库；能否合成 pair 单元仅取决于 questions - 是否同时含该 pair_id 的 original + mirror。收不齐者告警并整对剔除，使 - 后续 build_units 只面对合法孪生对（不触发 fail-fast），孤儿 unit 不计入 - total（对齐设计 §8 聚合入口的"告警 + 剔除"）。 + 是否同时含该 pair_id 的 original + mirror（且无多余非法记录）。非法者告警并 + 整对剔除，使后续 build_units 只面对合法孪生对（不触发 fail-fast），孤儿 unit + 不计入 total（对齐设计 §8 聚合入口的"告警 + 剔除"）。 参数: questions: 待聚合的题目列表（可混含 single 与孪生对成员）。 返回: - 可安全交给 build_units 的题目列表（single 全保留，pair 仅保留成对者）。 + 可安全交给 build_units 的题目列表（single 全保留，pair 仅保留合法成对者）。 """ by_pair: dict[str, list[GeneratedQuestion]] = defaultdict(list) singles: list[GeneratedQuestion] = [] @@ -234,16 +253,13 @@ def _drop_orphan_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQues kept_pairs: list[GeneratedQuestion] = [] for pair_id, group in by_pair.items(): - originals = [q for q in group if q.question_role == "pair_original"] - mirrors = [q for q in group if q.question_role == "pair_mirror"] - if len(originals) == 1 and len(mirrors) == 1: + if _is_valid_pair(group): kept_pairs.extend(group) else: logger.warning( - "孤儿 pair {}：收不齐 2 条（original={} mirror={}），剔除该 unit 不计入 total", + "孤儿 pair {}：非法配对（total={}），剔除该 unit 不计入 total", pair_id, - len(originals), - len(mirrors), + len(group), ) return singles + kept_pairs @@ -262,6 +278,33 @@ def _per_question_correctness(records: list[dict[str, Any]]) -> dict[str, bool]: return {r["question_id"]: r["prediction"] == r["answer"] for r in records} +def _grade_unit(unit: QuestionUnit, per_q: dict[str, bool]) -> bool: + """判定单元整体正确性，缺 prediction 时 fail-loud（带上下文）。 + + _drop_orphan_pairs 已剔除孤儿/非法配对，正常情况下 unit 内每题都应有对应 + record；若仍缺失说明聚合不变量被破坏（如 records 与 questions 不同源）。此处 + 显式抛带上下文的 ValueError（fail-loud，不 catch/不跳过/不兜底），而非放任 + unit_correctness 抛裸 KeyError 丢失定位信息。 + + 参数: + unit: 待判定单元。 + per_q: question_id → 该题是否作答正确 的映射。 + + 返回: + 单元整体是否正确（single 即单题正确；pair 走双向 AND）。 + + 异常: + ValueError: unit 内某 question_id 不在 per_q 中（聚合不变量被破坏）。 + """ + missing = [q.question_id for q in unit.questions if q.question_id not in per_q] + if missing: + raise ValueError( + f"unit {unit.unit_id} 的 question {missing} 缺 prediction" + "（_drop_orphan_pairs 后不应发生，聚合不变量被破坏）" + ) + return unit_correctness(unit, per_q) + + def _aggregate_results( records: list[dict[str, Any]], questions: list[GeneratedQuestion], @@ -287,8 +330,7 @@ def _aggregate_results( per_q = _per_question_correctness(records) units = build_units(_drop_orphan_pairs(questions)) - # unit 内任一题缺 record → unit_correctness 抛 KeyError（防静默兜底/读回校验）。 - graded = [(unit, unit_correctness(unit, per_q)) for unit in units] + graded = [(unit, _grade_unit(unit, per_q)) for unit in units] total = len(graded) correct = sum(1 for _, is_correct in graded if is_correct) diff --git a/tests/unit/test_inference_pair_aggregate.py b/tests/unit/test_inference_pair_aggregate.py index 4497afa..fb2faf1 100644 --- a/tests/unit/test_inference_pair_aggregate.py +++ b/tests/unit/test_inference_pair_aggregate.py @@ -238,6 +238,45 @@ class TestUnitLevelAggregation: assert result.correct == 1 assert any("orphan" in msg for msg in captured), "孤儿 pair 未告警（静默）" + def test_pair_with_extra_illegal_role_dropped(self) -> None: + """pair_id 下混入额外非法 role 记录（total>2）→ 整对剔除、不计入 total。""" + questions = [ + _make_question("s1", answer="B"), + _make_question("po", answer="B", pair_id="p", question_role="pair_original"), + _make_question("pm", answer="A", pair_id="p", question_role="pair_mirror"), + # 共享 pair_id 的额外非法记录（重复 original 角色） + _make_question("px", answer="C", pair_id="p", question_role="pair_original"), + ] + records = [ + _make_record("s1", prediction="B", answer="B"), + _make_record("po", prediction="B", answer="B"), + _make_record("pm", prediction="A", answer="A"), + _make_record("px", prediction="C", answer="C"), + ] + captured: list[str] = [] + sink_id = logger.add(captured.append, level="WARNING", format="{message}") + try: + result = _aggregate_results(records, questions, "run-illegal-role") + finally: + logger.remove(sink_id) + + assert result.total == 1 # 仅 single 存活，非法配对整对剔除 + assert result.correct == 1 + assert any("total=3" in msg for msg in captured), "非法配对未告警（静默）" + + def test_unit_missing_prediction_raises_descriptive_error(self) -> None: + """unit 缺 prediction → 描述性 ValueError（fail-loud，非静默、非裸 KeyError）。 + + 故意破坏聚合不变量（questions 含 s2 但 records 无 s2），验证带上下文报错。 + """ + questions = [ + _make_question("s1", answer="B"), + _make_question("s2", answer="A"), + ] + records = [_make_record("s1", prediction="B", answer="B")] # 缺 s2 的 record + with pytest.raises(ValueError, match=r"缺 prediction.*聚合不变量被破坏"): + _aggregate_results(records, questions, "run-broken-invariant") + def test_empty_records_returns_zero(self) -> None: """空 records/questions → 零值结果。""" result = _aggregate_results([], [], "run-empty")