refactor(case-verdict): extract build_case_evidence as single evidence-construction seam

用例判定证据构建收敛到 case_verdict.py 的 build_case_evidence 纯函数,
report.py 和 runs.py 各删 ~15 行重复逻辑,换一行调用。locality 回归:
证据构建改一处,全局生效。

- 新增 build_case_evidence(turns, results) -> dict[str, CaseEvidence]
- report.py:76-83 证据构建替换为一行调用
- runs.py:175-201 证据构建替换为一行调用
- 5 个新测试覆盖 build_case_evidence(纯函数,无 DB 依赖)
This commit is contained in:
sinohqb 2026-08-04 11:23:56 +08:00
parent 2b6cab6cb2
commit 2fddce8c92
4 changed files with 101 additions and 32 deletions

View File

@ -30,6 +30,35 @@ class CaseEvidence:
result_passes: tuple[bool, ...] = () result_passes: tuple[bool, ...] = ()
def build_case_evidence(turns: list, results: list) -> dict[str, CaseEvidence]:
"""Group turns + results by case_id into CaseEvidence per case.
Pure no I/O. Callers supply whatever turn/result objects expose
``.case_id`` and (for turns) ``.get_reply()`` / (for results) ``.passed``.
"""
evidence: dict[str, dict] = {}
for t in turns:
ev = evidence.get(t.case_id)
if ev is None:
ev = {"has_turns": True, "all_replied": True, "passes": []}
evidence[t.case_id] = ev
else:
ev["has_turns"] = True
if t.get_reply() is None:
ev["all_replied"] = False
for r in results:
ev = evidence.setdefault(r.case_id, {"has_turns": False, "all_replied": True, "passes": []})
ev["passes"].append(r.passed)
return {
cid: CaseEvidence(
has_turns=ev["has_turns"],
all_replied=ev["all_replied"],
result_passes=tuple(ev["passes"]),
)
for cid, ev in evidence.items()
}
def resolve_case_verdicts( def resolve_case_verdicts(
*, *,
case_outcomes: dict[str, CaseOutcomeSummary], case_outcomes: dict[str, CaseOutcomeSummary],

View File

@ -11,7 +11,7 @@ from typing import Any, Optional
from sqlmodel import Session from sqlmodel import Session
from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts from agenteval.evaluation.case_verdict import build_case_evidence, resolve_case_verdicts
from agenteval.evaluation.metrics import aggregate_runs from agenteval.evaluation.metrics import aggregate_runs
from agenteval.evaluation.report_render import render_html, render_json, render_markdown from agenteval.evaluation.report_render import render_html, render_json, render_markdown
from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary
@ -73,14 +73,7 @@ def generate_report(run_id: str, session=None) -> dict[str, Any]:
errored_case_ids = {e.get("case_id") for e in summary.case_errors} errored_case_ids = {e.get("case_id") for e in summary.case_errors}
# 权威判定:引擎经 combine_case_outcome 算一次写入 summary读路径只读不重算。 # 权威判定:引擎经 combine_case_outcome 算一次写入 summary读路径只读不重算。
# resolve_case_verdicts 统一处理「权威优先、老 run 近似回退」(唯一落点)。 # resolve_case_verdicts 统一处理「权威优先、老 run 近似回退」(唯一落点)。
evidence = { evidence = build_case_evidence(turns, results)
case_id: CaseEvidence(
has_turns=bool(item["turns"]),
all_replied=item["all_replied"],
result_passes=tuple(r["passed"] for r in item["results"]),
)
for case_id, item in case_map.items()
}
verdicts = resolve_case_verdicts( verdicts = resolve_case_verdicts(
case_outcomes=summary.case_outcomes, case_outcomes=summary.case_outcomes,
evidence=evidence, evidence=evidence,

View File

@ -6,7 +6,7 @@ from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel from pydantic import BaseModel
from sqlmodel import Session from sqlmodel import Session
from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts from agenteval.evaluation.case_verdict import build_case_evidence, resolve_case_verdicts
from agenteval.evaluation.engine import EvalEngine from agenteval.evaluation.engine import EvalEngine
from agenteval.models import EvalRun, RunStatus, RunSummary, RunTrigger from agenteval.models import EvalRun, RunStatus, RunSummary, RunTrigger
from agenteval.storage.db import get_session, iso_utc from agenteval.storage.db import get_session, iso_utc
@ -172,31 +172,11 @@ async def get_run_logs(run_id: str, session: Session = Depends(get_db)) -> dict:
# Authoritative per-case verdicts: resolve_case_verdicts prefers the engine's # Authoritative per-case verdicts: resolve_case_verdicts prefers the engine's
# stored case_outcomes and approximates only for legacy runs (single seam). # stored case_outcomes and approximates only for legacy runs (single seam).
evidence: dict[str, dict] = {}
for t in turns:
ev = evidence.get(t.case_id)
if ev is None:
ev = {"has_turns": True, "all_replied": True, "passes": []}
evidence[t.case_id] = ev
else:
ev["has_turns"] = True
if t.get_reply() is None:
ev["all_replied"] = False
for r in results:
ev = evidence.setdefault(r.case_id, {"has_turns": False, "all_replied": True, "passes": []})
ev["passes"].append(r.passed)
summary = run.summary or RunSummary() summary = run.summary or RunSummary()
errored_case_ids = {e.get("case_id") for e in summary.case_errors} errored_case_ids = {e.get("case_id") for e in summary.case_errors}
verdicts = resolve_case_verdicts( verdicts = resolve_case_verdicts(
case_outcomes=summary.case_outcomes, case_outcomes=summary.case_outcomes,
evidence={ evidence=build_case_evidence(turns, results),
cid: CaseEvidence(
has_turns=ev["has_turns"],
all_replied=ev["all_replied"],
result_passes=tuple(ev["passes"]),
)
for cid, ev in evidence.items()
},
errored_case_ids=errored_case_ids, errored_case_ids=errored_case_ids,
) )
case_verdicts = {cid: {"passed": v.passed, "connectivity": v.connectivity} for cid, v in verdicts.items()} case_verdicts = {cid: {"passed": v.passed, "connectivity": v.connectivity} for cid, v in verdicts.items()}

View File

@ -1,6 +1,6 @@
"""Unit tests for resolve_case_verdicts — the single case-verdict read seam.""" """Unit tests for resolve_case_verdicts — the single case-verdict read seam."""
from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts from agenteval.evaluation.case_verdict import CaseEvidence, build_case_evidence, resolve_case_verdicts
from agenteval.models import CaseOutcomeSummary from agenteval.models import CaseOutcomeSummary
@ -103,3 +103,70 @@ def test_mixed_authoritative_and_legacy():
) )
assert verdicts["auth"] == CaseOutcomeSummary(passed=True, connectivity=False) assert verdicts["auth"] == CaseOutcomeSummary(passed=True, connectivity=False)
assert verdicts["legacy"] == CaseOutcomeSummary(passed=True, connectivity=False) assert verdicts["legacy"] == CaseOutcomeSummary(passed=True, connectivity=False)
# ── build_case_evidence ──────────────────────────────────────────────────────
class _FakeTurn:
def __init__(self, case_id: str, has_reply: bool = True):
self.case_id = case_id
self._reply = {"msgBody": {"content": "ok"}} if has_reply else None
def get_reply(self):
return self._reply
class _FakeResult:
def __init__(self, case_id: str, passed: bool):
self.case_id = case_id
self.passed = passed
def test_build_evidence_turns_only_no_results():
"""Turns present, no results → has_turns=True, result_passes=()."""
turns = [_FakeTurn("c-1"), _FakeTurn("c-1")]
results: list = []
evidence = build_case_evidence(turns, results)
assert evidence == {
"c-1": CaseEvidence(has_turns=True, all_replied=True, result_passes=()),
}
def test_build_evidence_missing_reply_sets_all_replied_false():
"""A turn with no reply → all_replied=False for that case."""
turns = [_FakeTurn("c-1", has_reply=True), _FakeTurn("c-1", has_reply=False)]
results: list = []
evidence = build_case_evidence(turns, results)
assert evidence["c-1"].all_replied is False
assert evidence["c-1"].has_turns is True
def test_build_evidence_turns_and_results_aggregate_passes():
"""Turns + results → result_passes aggregates all rule passes for the case."""
turns = [_FakeTurn("c-1")]
results = [_FakeResult("c-1", passed=True), _FakeResult("c-1", passed=False)]
evidence = build_case_evidence(turns, results)
assert evidence["c-1"].result_passes == (True, False)
assert evidence["c-1"].has_turns is True
assert evidence["c-1"].all_replied is True
def test_build_evidence_results_without_turns():
"""Results exist but no turns → has_turns=False."""
turns: list = []
results = [_FakeResult("c-1", passed=True)]
evidence = build_case_evidence(turns, results)
assert evidence["c-1"].has_turns is False
assert evidence["c-1"].all_replied is True
assert evidence["c-1"].result_passes == (True,)
def test_build_evidence_multiple_cases():
"""Multiple cases are grouped independently."""
turns = [_FakeTurn("c-1"), _FakeTurn("c-2", has_reply=False)]
results = [_FakeResult("c-1", passed=True), _FakeResult("c-2", passed=False)]
evidence = build_case_evidence(turns, results)
assert len(evidence) == 2
assert evidence["c-1"] == CaseEvidence(has_turns=True, all_replied=True, result_passes=(True,))
assert evidence["c-2"] == CaseEvidence(has_turns=True, all_replied=False, result_passes=(False,))