From e29adf80d0b9cf05112342a4a33caa44c3bcc824 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 16:49:47 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=84=E6=B5=8B=E6=89=A7=E8=A1=8C=E7=9A=84?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=EF=BC=9A=E6=89=A7=E8=A1=8C=E3=80=81=E8=AF=84?= =?UTF-8?q?=E5=88=86=E3=80=81=E7=BB=9F=E8=AE=A1=E4=B8=8E=E5=AF=B9=E6=AF=94?= =?UTF-8?q?=E3=80=81=E6=B8=85=E7=90=86=E3=80=81=E5=AE=9A=E6=97=B6=E4=B8=8E?= =?UTF-8?q?=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：照细案 01 的测试清单把这一波的行为钉住，顺带修了跑测试时撞出来的两处。 - test_eval_runner：后台排队、并发不超过 max_parallel 也不超过全局并发的四分之一、同一题集同时只能一次、停止后只重做 没做完的（评完的不动）、重启标中断不续跑、单次金额上限触顶不再开新题、评测每日账单独记（不占裁判、核查的额度）、 连续 3 题助手调用失败止损、每题超时取消正在跑的运行、停在审批上的作答取消掉（评测不代批）、接口的估算与进度、设置里 eval 一组和 /settings/eval/spend - test_eval_score：五种结论、写出精度和百分数、维度配对与弱匹配、多行 / 名单 / 没有数据 / 反问、路径和期望先于数字、 dual 两版、允许的相对误差和双盲复算是同一个函数；读封存的答案、逐题 E1 用账本的比对函数；重评只读封存、不调模型 - test_eval_stats：Wilson、精确 McNemar、P50/P90 和手算对照；总表的各项；转移与四种归因；指纹差异；配对表；选基准 - test_eval_cleanup：保留规则；forget 只删一条引用都不剩的文件；清理后总表还在、结果和底下的运行删掉、共用的工件留着 - test_eval_schedule：定时接口收题集（拒工作流、输入模板、Webhook）、到点建一次评测并写 eval_run_id、重叠跳过、错过的按 宽限补；有退步推「评测有退步」、没评完推「评测失败」，消息里没有期望值和答案里的数 - test_eval_e2e 补：开评拍目录快照、记表结构前后；打开裁判时结论裁判走按需裁判那条路、花费记在评测账上，文字要点单列 - 修：期望反问的题在建图、运行失败（含超时）时把原因写全；改非工作流目标的定时任务时提交后不再读过期对象 --- backend/app/api/schedules.py | 3 +- backend/app/eval/score.py | 3 + backend/tests/test_eval_cleanup.py | 143 ++++++++++++ backend/tests/test_eval_e2e.py | 44 ++++ backend/tests/test_eval_runner.py | 333 ++++++++++++++++++++++++++++ backend/tests/test_eval_schedule.py | 227 +++++++++++++++++++ backend/tests/test_eval_score.py | 233 +++++++++++++++++++ backend/tests/test_eval_stats.py | 162 ++++++++++++++ 8 files changed, 1147 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_eval_cleanup.py create mode 100644 backend/tests/test_eval_runner.py create mode 100644 backend/tests/test_eval_schedule.py create mode 100644 backend/tests/test_eval_score.py create mode 100644 backend/tests/test_eval_stats.py diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py index fca3934f..ad115aa1 100644 --- a/backend/app/api/schedules.py +++ b/backend/app/api/schedules.py @@ -694,9 +694,10 @@ async def _update_target_schedule(session: AsyncSession, row: Schedule, body: di if result.rowcount != 1: await session.rollback() raise HTTPException(409, "这个定时任务刚被别人修改过，请刷新后再保存") + schedule_id = row.id await session.commit() session.expire_all() - return await _schedule_out(session, await _get(session, row.id)) + return await _schedule_out(session, await _get(session, schedule_id)) async def retime(session: AsyncSession, row: Schedule, body: dict[str, Any]) -> Schedule: diff --git a/backend/app/eval/score.py b/backend/app/eval/score.py index 6e4d8fcd..be673a75 100644 --- a/backend/app/eval/score.py +++ b/backend/app/eval/score.py @@ -510,6 +510,9 @@ def score_case(case: Mapping[str, Any], gold: Mapping[str, Any] | None, answer: verdict, traced, reason = _verdict(scores) if ask_only and path != "asked": verdict, traced, reason = "fail", False, "期望反问（问法含糊），助手却没有反问" + if path in ("build_failed", "run_failed"): + # 没走到能不能反问那一步（建图、运行失败、超时）：照实写是哪一步、为什么 + reason = WRONG_PATH[path] + (f"：{path_reason}" if path_reason else "") by_version = [] for extra in sets[1:]: more = _expectation_scores(expectations, (extra or {}).get("values") or {}, answer, path, judgement, cache) diff --git a/backend/tests/test_eval_cleanup.py b/backend/tests/test_eval_cleanup.py new file mode 100644 index 00000000..ec7b8e7c --- /dev/null +++ b/backend/tests/test_eval_cleanup.py @@ -0,0 +1,143 @@ +"""评测的清理与工件回收（app/eval/cleanup.py、core/artifact_store.forget；细案 01 做法 16）。 + +- 保留规则：最近 N 次、钉住的基线、A/B 的另几臂，还没结束的不动； +- 清掉的：逐题结果、明细工件、底下的运行（作答、算期望的）和它们的工件引用行；总表永久保留，cleaned_at 写明已清理； +- 工件文件一条引用都不剩才删：被别的运行引用着的同一份内容照留； +- 清理过的评测不能重试、不能重评。 +题集、问法都是虚构的。 +""" +from __future__ import annotations + +from datetime import timedelta + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import func, select + +from app.core import artifact_store +from app.db.base import SessionLocal, utcnow +from app.db.models import Artifact, Run +from app.db.models_eval import EvalResult, EvalRun +from app.engine.runner import run_manager +from app.eval import answer as answer_mod +from app.eval import cleanup, runner +from app.main import app +from tests import eval_samples as es +from tests.fixtures.sources import drop_source + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await runner.manager.shutdown() + await es.drop_suites() + await run_manager.shutdown() + + +def test_the_keep_rule(): + now = utcnow() + + def run(i: int, **extra) -> EvalRun: + return EvalRun(id=f"r{i}", suite_id="s", status=extra.pop("status", "succeeded"), + created_at=now - timedelta(minutes=i), **extra) + + runs = [run(1), run(2, pair_id="r2"), run(3, pair_id="r2"), run(4, pinned=True), run(5), run(6, status="running")] + kept = cleanup.keep_set(runs, 2) + assert kept == {"r1", "r2", "r3", "r4", "r6"}, "最近 2 次 + A/B 另一臂 + 钉住的 + 还没结束的" + + +async def test_forget_only_removes_files_nobody_refers_to(): + lonely = await artifact_store.put_json({"评测": "只有一处引用", "n": 1}, kind="test_forget") + shared = await artifact_store.put_json({"评测": "两处引用", "n": 2}, kind="test_forget") + async with SessionLocal() as session: + session.add(Artifact(id=shared, run_id="x" * 32, kind="test_forget")) + await session.execute(Artifact.__table__.delete().where(Artifact.id.in_([lonely, shared]), + Artifact.run_id == "")) + await session.commit() + removed = await artifact_store.forget([lonely, shared, "not-a-hash", lonely]) + assert removed == [lonely] + assert artifact_store.load(lonely) is None and artifact_store.load(shared)["n"] == 2 + async with SessionLocal() as session: + await session.execute(Artifact.__table__.delete().where(Artifact.id == shared)) + await session.commit() + assert await artifact_store.forget([shared]) == [shared] + + +async def test_cleaning_keeps_the_summary_and_shared_files(monkeypatch, tmp_path): + async def drive(self): + self.turn.path, self.turn.reply = "reply", "不查了" + return self.turn + + monkeypatch.setattr(answer_mod.Driver, "drive", drive) + name, sid = await es.make_source(tmp_path) + try: + suite_id = await es.make_suite(sid) + made: list[str] = [] + for i in range(3): + # 中间那一次换一下数：它的期望查询结果只有它自己有（别的两次查到的是同一份，文件共用） + _amount(tmp_path, 101.0 if i == 1 else 100.0) + async with SessionLocal() as session: + from app.db.models_eval import EvalSuite + from app.eval.variants import normalize_variant + + suite = await session.get(EvalSuite, suite_id) + [run] = await runner.start(session, suite, [normalize_variant({})], actor="评测测试") + await es.wait_eval(run.id) + made.append(run.id) + first, middle, last = made + async with SessionLocal() as session: + (await session.get(EvalRun, first)).pinned = True + gold_runs = (await session.execute(select(EvalResult.gold_run_ids).where( + EvalResult.eval_run_id == middle))).scalar_one() + owned = list((await session.execute(select(Artifact.id).where(Artifact.run_id.in_(gold_runs)))).scalars()) + details = (await session.execute(select(EvalResult.details_artifact).where( + EvalResult.eval_run_id == middle))).scalar_one() + # 同一份内容另有一次运行（不是这次评测的）也引用着：文件要留 + shared = owned[0] + session.add(Artifact(id=shared, run_id="y" * 32, kind="query_snapshot")) + await session.commit() + summary_before = (await _run(middle)).summary + out = await cleanup.clean_suite(suite_id, keep=1) + assert out["runs"] == 1 and out["results"] == 1 and out["deleted_runs"] == 2 and out["files"] >= 1 + cleaned = await _run(middle) + assert cleaned.cleaned_at is not None and cleaned.summary == summary_before, "总表永久保留" + assert (await _run(first)).cleaned_at is None and (await _run(last)).cleaned_at is None + async with SessionLocal() as session: + assert await session.scalar(select(func.count()).select_from(EvalResult).where( + EvalResult.eval_run_id == middle)) == 0 + assert await session.scalar(select(func.count()).select_from(Run).where(Run.id.in_(gold_runs))) == 0 + assert await session.scalar(select(func.count()).select_from(Artifact).where( + Artifact.run_id.in_(gold_runs))) == 0 + assert artifact_store.load(shared) is not None, "别的运行还引用着的文件不删" + async with SessionLocal() as session: + still = set((await session.execute(select(Artifact.id).where(Artifact.id.in_(owned)))).scalars()) + # 同样的内容被另外两次评测的运行引用着的（同一条期望 SQL 查到同一份结果）也留着；一条引用都不剩的才删 + for artifact in owned: + assert (artifact_store.load(artifact) is not None) == (artifact in still), artifact + assert any(a not in still for a in owned), "这次评测独有的工件文件删掉了" + assert artifact_store.load(details) is None, "明细工件跟着删" + with pytest.raises(runner.EvalRefused): + await runner.retry(middle) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + refused = await client.post(f"/api/evals/evaluations/{middle}/rescore") + assert refused.status_code == 409 and refused.json()["code"] == "eval_details_cleaned" + compared = (await client.get(f"/api/evals/evaluations/{last}/compare", + params={"base": middle})).json() + assert compared["cleaned"] is True and compared["transitions"] is None + again = (await client.post(f"/api/evals/suites/{suite_id}/cleanup")).json() + assert again["runs"] == 0, "已经清过的不再清" + finally: + await drop_source(sid) + + +def _amount(tmp_path, value: float) -> None: + import sqlite3 + + with sqlite3.connect(tmp_path / "shop.db") as db: + db.execute("UPDATE orders SET amount = ? WHERE id = 1", (value,)) + + +async def _run(run_id: str) -> EvalRun: + async with SessionLocal() as session: + return await session.get(EvalRun, run_id) diff --git a/backend/tests/test_eval_e2e.py b/backend/tests/test_eval_e2e.py index 3a017db2..3b72ac3d 100644 --- a/backend/tests/test_eval_e2e.py +++ b/backend/tests/test_eval_e2e.py @@ -87,6 +87,13 @@ async def test_catalog_contribution_pair_end_to_end(client, monkeypatch, source) assert summary["accuracy"]["passed"] == 1 and summary["traced_accuracy"]["passed"] == 1 assert summary["accuracy"]["ci"] is not None and summary["e1"]["passed"] == 1 + # 开评时拍了一份当前目录（不占数据源的队），上传表格版本和表结构哈希前后各记一次 + async with SessionLocal() as session: + ev = await session.get(EvalRun, live["id"]) + assert ev.catalog_snapshot and ev.fingerprint["catalog"].startswith("live:") + assert ev.fingerprint["sources_before"] == ev.fingerprint["sources_after"] + assert ev.fingerprint["sources_changed"] == [] and ev.fingerprint["prompt"] + # 评测运行：来源 eval、不进使用统计、名字写题集和题号 async with SessionLocal() as session: run = await session.get(Run, right.run_id) @@ -120,3 +127,40 @@ async def test_comparing_with_the_previous_run_finds_the_regression(client, monk assert item["transition"] == "regressed" and item["cause"] == "method", item assert "catalog" in {d["key"] for d in compared["fingerprint_diff"]}, compared["fingerprint_diff"] assert shop.RIGHT != shop.BY_ORDER_DATE + + +async def test_judge_on_charges_the_eval_ledger_and_text_points_stay_apart(client, monkeypatch, source): + from langchain_core.messages import AIMessage + + from app.engine import judge + from app.engine.spend import eval_spend + from app.eval.judging import POINTS_SYSTEM_PROMPT + from app.providers import mock_model + + es.use_copilot(monkeypatch, es.Copilot(source["name"])) + + def decide(self, messages): + text = "\n".join(str(m.content) for m in messages) + if POINTS_SYSTEM_PROMPT[:20] in text: + return AIMessage(content='[{"point": 1, "verdict": "covered", "reason": "说了按支付日"}]') + return AIMessage(content=es.REPORT) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) + case = es.paid_case(source["id"]) + case["expectations"].append({"kind": "text", "points": ["说明按支付日统计"]}) + suite_id = await es.make_suite(source["id"], cases=[case]) + judge_before = await judge.judge_spend.daily() + ledger_before = (await eval_spend.daily())["calls"] + [ev] = (await client.post(f"/api/evals/suites/{suite_id}/evaluations", + json={"variant": {"judge": True}})).json()["evaluations"] + await es.wait_eval(ev["id"]) + [item] = await _results(ev["id"]) + assert item.verdict == "pass", item.error + text = next(s for s in item.scores if s["kind"] == "text") + assert text["outcome"] == "covered" and text["required"] is False, "文字要点单列" + assert item.judge["claims"]["calls"] >= 1, "走了按需裁判那条路" + assert set(item.judge["claims"]["counts"]) == {"supported", "partial", "contradicted", "insufficient", "unjudged"} + assert await judge.judge_spend.daily() == judge_before, "裁判的花费记在评测那本账上" + assert (await eval_spend.daily())["calls"] > ledger_before + got = (await client.get(f"/api/evals/evaluations/{ev['id']}")).json() + assert got["summary"]["judge"]["on"] is True and got["summary"]["text"] == {"covered": 1} diff --git a/backend/tests/test_eval_runner.py b/backend/tests/test_eval_runner.py new file mode 100644 index 00000000..06a2c233 --- /dev/null +++ b/backend/tests/test_eval_runner.py @@ -0,0 +1,333 @@ +"""评测执行（app/eval/runner.py；细案 01 做法 9，评测-D9）。 + +作答这一步多半换成桩（Driver.drive）：这里守的是后台任务本身——排队、并发上限（不超过 max_parallel，也不超过全局运行 +并发的四分之一）、停止、只重试没做完的、重启标中断不续跑、单次和每日金额上限（评测单独一本账，不占裁判、核查的额度）、 +连续几题助手调用失败止损、每题超时会取消正在跑的运行；停在审批上的作答走真的运行（评测不代批，取消掉）。 +题目多半是「期望反问」的题：没有期望来源，不用跑期望探针。问法、题集名都是虚构的。 +""" +from __future__ import annotations + +import asyncio + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.db.base import SessionLocal, utcnow +from app.db.models import Run, Setting +from app.db.models_eval import EvalResult, EvalRun +from app.engine.runner import run_manager +from app.eval import answer as answer_mod +from app.eval import runner +from app.main import app +from tests import eval_samples as es +from tests.fixtures.sources import drop_source + + +async def _clear_ledger() -> None: + async with SessionLocal() as session: + row = await session.get(Setting, "eval_spend") + if row is not None: + await session.delete(row) + await session.commit() + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + await _clear_ledger() + yield + await _clear_ledger() + await runner.manager.shutdown() + await es.drop_suites() + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def ask_cases(n: int) -> list[dict]: + return [{"question": f"第 {i} 个含糊的问题", "expectations": [{"kind": "ask"}]} for i in range(1, n + 1)] + + +def stub(monkeypatch, *, delay: float = 0.0, path: str = "asked", cost: float = 0.0, copilot_failed: bool = False, + current_run: str | None = None, live: list | None = None): + """作答换成桩：等 delay 秒，交回 path；cost 记在助手的用量上。live 记下同时在答的有几题。""" + peak = {"now": 0, "max": 0} + + async def drive(self): + peak["now"] += 1 + peak["max"] = max(peak["max"], peak["now"]) + try: + self.turn.current_run = current_run + await self._phase("build") + await asyncio.sleep(delay) + self.turn.path = path + self.turn.copilot_failed = copilot_failed + if copilot_failed: + self.turn.reason = "助手本轮未完成：接入返回了 502" + self.turn.usage["assistant"] = {"by_model": {}, "input_tokens": 0, "output_tokens": 0, "calls": 1, + "cost_usd": cost, "unpriced_calls": 0} + return self.turn + finally: + peak["now"] -= 1 + + monkeypatch.setattr(answer_mod.Driver, "drive", drive) + return peak + + +async def start(suite_id: str, variant: dict | None = None, **kw) -> list[EvalRun]: + from app.db.models_eval import EvalSuite + from app.eval.variants import normalize_variant + + async with SessionLocal() as session: + suite = await session.get(EvalSuite, suite_id) + return await runner.start(session, suite, [normalize_variant(variant or {})], actor="评测测试", **kw) + + +async def results(run_id: str) -> list[EvalResult]: + async with SessionLocal() as session: + return list((await session.execute(select(EvalResult).where(EvalResult.eval_run_id == run_id) + .order_by(EvalResult.seq))).scalars()) + + +async def get(run_id: str) -> EvalRun: + async with SessionLocal() as session: + return await session.get(EvalRun, run_id) + + +async def test_items_run_in_the_background_within_the_parallel_cap(monkeypatch): + peak = stub(monkeypatch, delay=0.15) + suite_id = await es.make_suite("none", cases=ask_cases(5), budget={"max_parallel": 3}) + [run] = await start(suite_id) + assert run.status == "queued" and run.counts["total"] == 5 + await es.wait_eval(run.id) + row = await get(run.id) + assert row.status == "succeeded" and row.counts["done"] == 5 and row.counts["passed"] == 5 + assert peak["max"] == 3, "同时最多 max_parallel 题" + assert all(r.status == "ok" and r.verdict == "pass" and r.current is None for r in await results(run.id)) + assert row.summary["accuracy"]["passed"] == 5 and row.summary["verdicts"]["pass"] == 5 + + +async def test_the_cap_never_exceeds_a_quarter_of_the_global_concurrency(monkeypatch): + from app.core.config import settings + + monkeypatch.setattr(settings, "max_concurrent_runs", 4) + peak = stub(monkeypatch, delay=0.1) + suite_id = await es.make_suite("none", cases=ask_cases(3), budget={"max_parallel": 3}) + [run] = await start(suite_id) + await es.wait_eval(run.id) + assert peak["max"] == 1, "全局并发 4 的四分之一是 1" + + +async def test_a_second_evaluation_of_the_same_suite_waits(monkeypatch): + stub(monkeypatch, delay=0.3) + suite_id = await es.make_suite("none", cases=ask_cases(1)) + [run] = await start(suite_id) + with pytest.raises(runner.EvalRefused) as refused: + await start(suite_id) + assert refused.value.code == runner.EVAL_BUSY and refused.value.status == 409 + await es.wait_eval(run.id) + + +async def test_stop_then_retry_only_the_unfinished(monkeypatch): + stub(monkeypatch, delay=0.05) + suite_id = await es.make_suite("none", cases=ask_cases(4), budget={"max_parallel": 1}) + [run] = await start(suite_id) + # 等第一题评完再停 + for _ in range(100): + if (await get(run.id)).counts.get("done", 0) >= 1: + break + await asyncio.sleep(0.02) + stub(monkeypatch, delay=5.0) + await asyncio.sleep(0.1) + assert await runner.stop(run.id) + row = await get(run.id) + assert row.status == "cancelled" and row.error == runner.STOPPED + before = {r.id: (r.status, r.finished_at) for r in await results(run.id)} + done = [rid for rid, (status, _) in before.items() if status == "ok"] + assert done and any(status == "cancelled" for status, _ in before.values()) + assert not await runner.stop(run.id), "已经结束的停不了" + + stub(monkeypatch, delay=0.01) + await runner.retry(run.id) + await es.wait_eval(run.id) + after = {r.id: (r.status, r.finished_at) for r in await results(run.id)} + assert all(status == "ok" for status, _ in after.values()) + assert all(after[rid] == before[rid] for rid in done), "评完了的不重做" + assert (await get(run.id)).status == "succeeded" + with pytest.raises(runner.EvalRefused) as nothing: + await runner.retry(run.id) + assert nothing.value.code == runner.NOTHING_TO_RETRY + + +async def test_a_restart_marks_unfinished_evaluations_interrupted_and_does_not_resume(): + suite_id = await es.make_suite("none", cases=ask_cases(1)) + async with SessionLocal() as session: + row = EvalRun(suite_id=suite_id, status="running", started_at=utcnow()) + session.add(row) + await session.flush() + session.add(EvalResult(eval_run_id=row.id, status="running")) + await session.commit() + run_id = row.id + assert await runner.startup() >= 1 + row = await get(run_id) + assert row.status == "interrupted" and row.error == runner.INTERRUPTED + [item] = await results(run_id) + assert item.status == "cancelled" + assert not runner.manager.running(run_id), "不自动续跑" + + +async def test_the_per_run_cap_stops_new_items(monkeypatch): + stub(monkeypatch, cost=1.0) + suite_id = await es.make_suite("none", cases=ask_cases(3), budget={"max_parallel": 1, "per_run_max_usd": 1.5}) + [run] = await start(suite_id) + await es.wait_eval(run.id) + row = await get(run.id) + items = await results(run.id) + assert [r.status for r in items] == ["ok", "ok", "skipped"] + assert row.status == "partial" and row.stop_reason == runner.LIMIT_RUN.format(limit=1.5) + assert "金额上限" in items[2].error and row.summary["cost_usd"] == 2.0 + + +async def test_the_daily_cap_is_its_own_ledger(monkeypatch): + from app.engine import judge + from app.engine.spend import _today, eval_spend, verify_spend + + stub(monkeypatch, cost=0.4) + async with SessionLocal() as session: + for key in ("eval_spend", "judge_spend", "verify_spend"): + row = await session.get(Setting, key) + if row is not None: + await session.delete(row) + session.add(Setting(key="eval_spend", value={"date": _today(), "usd": 4.9, "calls": 3, "unpriced_calls": 0})) + await session.commit() + judge_before, verify_before = await judge.judge_spend.daily(), await verify_spend.daily() + suite_id = await es.make_suite("none", cases=ask_cases(3), budget={"max_parallel": 1}) + [first] = await start(suite_id) + await es.wait_eval(first.id) + # 第一次没有可估的每题花费：按实际累计，越过每日上限之后不再开新题 + spent = await eval_spend.daily() + assert spent["usd"] > 5.0 - 1e-9 and spent["calls"] == 3 + 1, spent + row = await get(first.id) + assert [r.status for r in await results(first.id)] == ["ok", "skipped", "skipped"] + assert row.stop_reason == runner.LIMIT_DAILY.format(limit=5.0) + assert await judge.judge_spend.daily() == judge_before and await verify_spend.daily() == verify_before, \ + "评测不占裁判、核查的额度" + + +async def test_consecutive_assistant_failures_stop_the_evaluation(monkeypatch): + stub(monkeypatch, path="build_failed", copilot_failed=True) + suite_id = await es.make_suite("none", cases=ask_cases(5), budget={"max_parallel": 1}) + [run] = await start(suite_id) + await es.wait_eval(run.id) + row = await get(run.id) + statuses = [r.status for r in await results(run.id)] + assert statuses[:3] == ["ok", "ok", "ok"] and statuses[3:] == ["skipped", "skipped"] + assert row.stop_reason.startswith("连续 3 题助手调用失败") and "502" in row.stop_reason + + +async def test_a_case_over_its_time_limit_stops_its_run(monkeypatch): + stopped: list[str | None] = [] + + async def fake_stop(run_id, actor=None): + stopped.append(run_id) + + monkeypatch.setattr(answer_mod, "stop_run", fake_stop) + stub(monkeypatch, delay=5.0, current_run="run-in-flight") + original = runner.EvalExecution._prepare + + async def prepare(self): + units = await original(self) + self.budget["per_case_timeout_s"] = 0.2 + return units + + monkeypatch.setattr(runner.EvalExecution, "_prepare", prepare) + suite_id = await es.make_suite("none", cases=ask_cases(1)) + [run] = await start(suite_id) + await es.wait_eval(run.id) + [item] = await results(run.id) + assert stopped == ["run-in-flight"], "超时取消正在跑的那次运行" + assert item.status == "ok" and item.path == "build_failed" and "每题时限" in item.error + + +async def test_a_run_stopped_at_an_approval_is_cancelled_not_approved(client, monkeypatch, tmp_path): + from app.eval.answer import APPROVAL_REASON + + class Approving(es.Copilot): + async def astream(self, messages): + if "上线前自查" in messages[1][1]: + yield es._line({"op": "done"}) + return + for op in [{"op": "add_node", "node": {"id": "start", "type": "input", "label": "输入", "config": {}}}, + {"op": "add_node", "node": {"id": "ok", "type": "human", "label": "确认", + "config": {"mode": "approve", "title": "请确认"}}}, + {"op": "add_edge", "edge": {"source": "start", "target": "ok"}}, + {"op": "add_node", "node": {"id": "out", "type": "output", "label": "答案", + "config": {"fields": [{"name": "答案", "value": "好"}]}}}, + {"op": "add_edge", "edge": {"source": "ok", "target": "out", "sourceHandle": "approved"}}, + {"op": "done", "explanation": "确认一下", "run": True}]: + yield es._line(op) + + name, sid = await es.make_source(tmp_path) + try: + es.use_copilot(monkeypatch, Approving(name)) + suite_id = await es.make_suite(sid, cases=[{"question": "要人确认的问题", "expectations": [{"kind": "ask"}]}]) + [run] = await start(suite_id) + await es.wait_eval(run.id) + [item] = await results(run.id) + assert item.path == "run_failed" and "评测不代批" in item.error + async with SessionLocal() as session: + answered = await session.get(Run, item.run_id) + assert answered.status == "cancelled", "评测不代批：取消掉" + details = (await client.get(f"/api/evals/evaluations/{run.id}/results/{item.id}")).json() + assert details["details"]["turn"]["reason"] == APPROVAL_REASON + finally: + await drop_source(sid) + + +async def test_the_api_starts_estimates_and_reports_progress(client, monkeypatch): + stub(monkeypatch, delay=0.05) + suite_id = await es.make_suite("none", cases=ask_cases(2)) + estimate = (await client.post(f"/api/evals/suites/{suite_id}/evaluations/estimate", json={})).json() + assert estimate["cases"] == 2 and estimate["estimate"]["total_usd"] is None and estimate["parallel"] == 2 + assert estimate["budget"]["daily_max_usd"] == 5.0 and estimate["arms"][0]["models"]["assistant"] == "mock-fast" + started = await client.post(f"/api/evals/suites/{suite_id}/evaluations", json={"variant": {"repeats": 2}}) + assert started.status_code == 202, started.text + [ev] = started.json()["evaluations"] + assert ev["counts"]["total"] == 4 and ev["variant"]["repeats"] == 2 + await es.wait_eval(ev["id"]) + got = (await client.get(f"/api/evals/evaluations/{ev['id']}")).json() + assert got["status"] == "succeeded" and got["current"] == [] and got["summary"]["pass_rates"] + listed = (await client.get(f"/api/evals/suites/{suite_id}/evaluations")).json()["evaluations"] + assert [e["id"] for e in listed] == [ev["id"]] + rows = (await client.get(f"/api/evals/evaluations/{ev['id']}/results")).json()["results"] + assert len(rows) == 4 and rows[0]["question"] == "第 1 个含糊的问题" + second = (await client.post(f"/api/evals/suites/{suite_id}/evaluations/estimate", json={})).json() + assert second["estimate"]["basis"] == ev["id"], "第二次按上一次的每题平均花费估" + bad = await client.post(f"/api/evals/suites/{suite_id}/evaluations", json={"variant": {"catalog": "nope"}}) + assert bad.status_code == 422 and bad.json()["code"] == "eval_variant_invalid" + + +async def test_eval_settings_group_and_spend_endpoint(client): + got = (await client.get("/api/settings")).json()["eval"] + assert got == {"max_parallel": 2, "per_case_timeout_s": 900, "per_run_max_usd": 2.0, "daily_max_usd": 5.0, + "keep_runs": 20, "judge": False} + bad = await client.put("/api/settings", json={"values": {"eval": {"keep_runs": 0}}}) + assert bad.status_code == 422 and "保留" in bad.json()["detail"] + bad = await client.put("/api/settings", json={"values": {"eval": {"per_run_max_usd": 9.0}}}) + assert bad.status_code == 422, "单次上限不能高于每日上限" + ok = await client.put("/api/settings", json={"values": {"eval": {"daily_max_usd": 3.0, "judge": True}, + "eval_spend": {"usd": 0}}}) + assert ok.status_code == 200 and ok.json()["eval"]["daily_max_usd"] == 3.0 and "eval_spend" not in ok.json() + spend = (await client.get("/api/settings/eval/spend")).json() + assert set(spend) == {"date", "usd", "calls", "unpriced_calls", "daily_max_usd"} and spend["daily_max_usd"] == 3.0 + # 题集只能把每日上限往下收 + from app.eval.suites import effective_budget + + assert effective_budget({"daily_max_usd": 9.0}, {"daily_max_usd": 3.0})["daily_max_usd"] == 3.0 + assert effective_budget({"daily_max_usd": 1.0}, {"daily_max_usd": 3.0})["daily_max_usd"] == 1.0 + await client.put("/api/settings", json={"values": {"eval": {}}}) diff --git a/backend/tests/test_eval_schedule.py b/backend/tests/test_eval_schedule.py new file mode 100644 index 00000000..cab8b309 --- /dev/null +++ b/backend/tests/test_eval_schedule.py @@ -0,0 +1,227 @@ +"""评测题集的定时与通知（app/eval/scheduled.py、notify/service.notify_eval；细案 01 做法 15，总方案 X4、X6）。 + +- 定时接口按 target_kind 收评测题集：核对目标交给 app/eval/scheduled.check_target（题集在、没归档、有启用的题）；非工作流 + 目标不收工作流、输入模板、Webhook（Webhook 的事件说的是那一次运行，非工作流目标收不到，照实拒）；接口返回目标的种类、id、 + 名字；工作流的定时行为一字不变（test_schedules_api.py）； +- 到点建一次评测：定时跑 1 次，截至时间是计划触发时间，ScheduleFire.eval_run_id 指向它；上一次还没结束就 skipped_overlap； + 错过的按宽限补（同一套调度循环）； +- 有一道题从通过变成不通过推「评测有退步」，没评完推「评测失败」：只推订阅了的全局渠道，消息里只有题集名、通过数、退步题数、 + 链接，不放期望值和答案里的数。题集名、渠道都是虚构的。 +""" +from __future__ import annotations + +import asyncio +import json +import uuid +from datetime import datetime, timedelta, timezone + +import httpx +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete, select + +from app.core.crypto import encrypt +from app.db.base import SessionLocal, utcnow +from app.db.models import NotifyChannel, Schedule, ScheduleFire +from app.db.models_eval import EvalResult, EvalRun +from app.engine.runner import run_manager +from app.eval import answer as answer_mod +from app.eval import runner +from app.main import app +from app.notify import core +from app.notify import service as notify +from app.scheduler import loop, service +from tests import eval_samples as es + +UTC = timezone.utc +URL = "https://hooks.example.com/agentlab/eval-7c2a" + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + async with SessionLocal() as session: + await session.execute(delete(Schedule)) + await session.commit() + yield + await runner.manager.shutdown() + await notify.drain() + async with SessionLocal() as session: + await session.execute(delete(Schedule)) + await session.execute(delete(NotifyChannel).where(NotifyChannel.name.like("评测通知-%"))) + await session.commit() + await es.drop_suites() + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture +def received(monkeypatch): + got: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + got.append(request) + return httpx.Response(200, json={"ok": True}) + + monkeypatch.setattr(core, "_transport", httpx.MockTransport(handler)) + monkeypatch.setattr(core, "RETRY_DELAYS", (0, 0, 0)) + return got + + +def stub(monkeypatch, *, delay: float = 0.0, path: str = "asked"): + async def drive(self): + await asyncio.sleep(delay) + self.turn.path = path + return self.turn + + monkeypatch.setattr(answer_mod.Driver, "drive", drive) + + +ASK = [{"question": "含糊的问题", "expectations": [{"kind": "ask"}]}] + + +async def _schedule(client, suite_id: str, **extra) -> dict: + made = await client.post("/api/schedules", json={"name": "每周回归", "target_kind": "eval_suite", + "target_id": suite_id, "cron": "0 9 * * 1", + "timezone": "Asia/Shanghai", **extra}) + assert made.status_code == 201, made.text + return made.json() + + +async def test_the_schedule_api_takes_an_eval_suite(client): + suite_id = await es.make_suite("none", name="虚构回归题集", cases=ASK) + body = await _schedule(client, suite_id) + assert (body["target_kind"], body["target_id"], body["target_name"]) == ("eval_suite", suite_id, "虚构回归题集") + assert body["target_label"] == "评测题集" and body["workflow_id"] == "" and body["webhook"] is None + assert body["input_template"] == {} and body["target_exists"] is True + listed = (await client.get("/api/schedules")).json()["items"] + assert [s["target_name"] for s in listed] == ["虚构回归题集"] + for bad, field in (({"target_id": "nope"}, "target_id"), ({"target_id": None}, "target_id"), + ({"webhook": {"url": URL}}, "webhook"), ({"input_template": {"a": "b"}}, "input_template"), + ({"workflow_id": "wf"}, "workflow_id"), ({"target_kind": "nope"}, "target_kind")): + refused = await client.post("/api/schedules", json={"name": "x", "target_kind": "eval_suite", + "target_id": suite_id, "cron": "0 9 * * 1", + "timezone": "Asia/Shanghai", **bad}) + assert refused.status_code == 422 and refused.json()["field"] == field, (bad, refused.text) + # 改名、改时间、停用都照常；目标种类建好不能改 + patched = await client.patch(f"/api/schedules/{body['id']}", json={"name": "每天回归", "cron": "0 9 * * *", + "enabled": False, "webhook": None, + "version": body["version"]}) + assert patched.status_code == 200, patched.text + assert patched.json()["name"] == "每天回归" and patched.json()["next_run_at"] is None + moved = await client.patch(f"/api/schedules/{body['id']}", json={"target_kind": "workflow", + "version": patched.json()["version"]}) + assert moved.status_code == 422 and moved.json()["field"] == "target_kind" + # 题集归档了：照实拒 + archived = await es.make_suite("none", name="归档的题集", cases=ASK) + async with SessionLocal() as session: + from app.db.models_eval import EvalSuite + + (await session.get(EvalSuite, archived)).archived = True + await session.commit() + refused = await client.post("/api/schedules", json={"name": "x", "target_kind": "eval_suite", "target_id": archived, + "cron": "0 9 * * 1", "timezone": "Asia/Shanghai"}) + assert refused.status_code == 422 and "已归档" in refused.json()["detail"] + + +async def test_firing_starts_one_evaluation_and_skips_on_overlap(client, monkeypatch): + stub(monkeypatch, delay=0.5) + suite_id = await es.make_suite("none", cases=ASK, default_variant={"repeats": 3}) + body = await _schedule(client, suite_id) + now = utcnow().replace(microsecond=0) + fire = await service.fire_once(body["id"], now, "manual", actor="王芳") + assert fire.status == "started" and fire.eval_run_id and fire.run_id is None + async with SessionLocal() as session: + run = await session.get(EvalRun, fire.eval_run_id) + assert run.trigger == "schedule" and run.schedule_id == body["id"] and run.repeats == 1, "定时只跑 1 次" + assert run.as_of.replace(tzinfo=UTC) == now and run.created_by == "王芳" + again = await service.fire_once(body["id"], now + timedelta(seconds=1), "manual") + assert again.status == "skipped_overlap" and f"#{run.id[:6]}" in again.reason + same = await service.fire_once(body["id"], now, "manual") + assert same.id == fire.id, "同一次触发只发起一次" + await es.wait_eval(run.id) + history = (await client.get(f"/api/schedules/{body['id']}/fires")).json()["items"] + assert {f["eval_run_id"] for f in history if f["status"] == "started"} == {run.id} + + +async def test_a_missed_fire_within_grace_is_made_up(client, monkeypatch): + stub(monkeypatch) + suite_id = await es.make_suite("none", cases=ASK) + body = await _schedule(client, suite_id) + due = utcnow().replace(microsecond=0) - timedelta(minutes=20) + async with SessionLocal() as session: + await session.execute(Schedule.__table__.update().where(Schedule.id == body["id"]).values(next_run_at=due)) + await session.commit() + fires = await loop.fire_due(due + timedelta(minutes=20)) + [made_up] = [f for f in fires if f.status == "started"] + assert made_up.kind == "misfire" and made_up.eval_run_id + await es.wait_eval(made_up.eval_run_id) + + +async def _channel(events: list[str]) -> str: + async with SessionLocal() as session: + row = NotifyChannel(name=f"评测通知-{uuid.uuid4().hex[:6]}", scope="global", format="generic", url=encrypt(URL), + events=events) + session.add(row) + await session.commit() + return row.id + + +async def test_a_regression_is_pushed_without_numbers(client, monkeypatch, received, tmp_path): + name, sid = await es.make_source(tmp_path) + es.use_copilot(monkeypatch, es.Copilot(name)) + es.use_report(monkeypatch) + suite_id = await es.make_suite(sid, name="虚构周回归") + body = await _schedule(client, suite_id) + await _channel(["eval.regressed", "eval.failed"]) + first = await service.fire_once(body["id"], utcnow().replace(microsecond=0) - timedelta(minutes=5), "manual") + await es.wait_eval(first.eval_run_id) + assert received == [], "第一次没有可比的、也评完了：不推" + await es.set_notes(sid, {"label": es.NOTES["label"]}) + second = await service.fire_once(body["id"], utcnow().replace(microsecond=0), "manual") + await es.wait_eval(second.eval_run_id) + [request] = received + message = json.loads(request.content) + assert message["event"] == "eval.regressed" and message["title"] == "评测「虚构周回归」有退步" + text = message["text"] + assert "通过：0 / 1 题" in text and "退步：1 题" in text + for secret in ("330", "310", "gmv", "orders", "paid_date"): + assert secret not in json.dumps(message, ensure_ascii=False), f"通知里不该有 {secret}" + from app.db.models import NotifyDelivery + + async with SessionLocal() as session: + [delivery] = (await session.execute(select(NotifyDelivery).where( + NotifyDelivery.subject_id == second.eval_run_id))).scalars().all() + assert delivery.subject_kind == "eval_run" and delivery.status == "ok" + from tests.fixtures.sources import drop_source + + await drop_source(sid) + + +async def test_an_unfinished_scheduled_evaluation_pushes_failed(client, monkeypatch, received): + async def drive(self): + self.turn.path, self.turn.copilot_failed, self.turn.reason = "build_failed", True, "助手本轮未完成" + return self.turn + + monkeypatch.setattr(answer_mod.Driver, "drive", drive) + suite_id = await es.make_suite("none", cases=ASK * 4, name="虚构失败题集", budget={"max_parallel": 1}) + body = await _schedule(client, suite_id) + await _channel(["eval.failed"]) + fire = await service.fire_once(body["id"], utcnow().replace(microsecond=0), "manual") + await es.wait_eval(fire.eval_run_id) + async with SessionLocal() as session: + run = await session.get(EvalRun, fire.eval_run_id) + skipped = (await session.execute(select(EvalResult).where( + EvalResult.eval_run_id == run.id, EvalResult.status == "skipped"))).scalars().all() + assert run.status == "partial" and run.stop_reason.startswith("连续 3 题") and len(skipped) == 1 + [request] = received + message = json.loads(request.content) + assert message["event"] == "eval.failed" and message["title"] == "评测「虚构失败题集」没有评完" + assert "通过：0 / 3 题" in message["text"] + # 链接去那条定时任务的触发历史（设了对外访问地址才有链接） + assert all("/schedules/" in link["url"] for link in message["links"]) diff --git a/backend/tests/test_eval_score.py b/backend/tests/test_eval_score.py new file mode 100644 index 00000000..645f92b4 --- /dev/null +++ b/backend/tests/test_eval_score.py @@ -0,0 +1,233 @@ +"""评分（app/eval/score.py；细案 01 做法 10，评测-D4）。纯计算的部分直接喂构造好的答案；读封存、逐题 E1、重评走一次真的 +运行（tests/ledger_samples 的店铺库）。数字、问法、地区都是虚构的。 + +- scalar：对且可追溯（引用的格对上）、对但没引用（只有正文裸写的数对上，含叙述里自己算的比率）、错、量纲不同（差 10 的 + 整数次幂）、答案引用遮罩格的不能评分；维度配对（华东的那一格才算）、小整数又没维度的标弱匹配； +- table：按键配对、要求顺序时核对顺序；set：名单齐不齐、顺序；none：没给数 + 有查询返回 0 行；ask：反问才对；text：单列不计； +- 题目结论：路径不对直接不通过；期望没算出来不能评分 / 升版待处置跳过；只有文字要点的不能评分； +- dual：新旧两版各给一份结论； +- 允许的相对误差和双盲复算是同一个函数（engine/tolerance）； +- 重评只读封存、不调模型。 +""" +from __future__ import annotations + +import pytest + +import app.db.models # noqa: F401 - conftest 按已导入的模型建表：只跑这一个文件时也要有表 +from app.engine import tolerance +from app.eval.score import Answer, score_case + + +def _case(*expectations: dict, gold_kind: str | None = "sql") -> dict: + return {"kind": "ask", "question": "虚构问题", "expectations": list(expectations), + "gold": {"kind": gold_kind, "datasource_id": "s", "sql": "SELECT 1"} if gold_kind else None} + + +def _gold(values: dict, *more: dict) -> dict: + sets = [{"run_id": "g", "snapshot": "x", "version": None, "values": values}] + sets += [{"run_id": "g2", "snapshot": "y", "version": m.pop("_version", 2), "values": m} for m in more] + return {"status": "ok", "reason": None, "code": None, "before": {"status": "ok", "sets": sets}, + "after": {"status": "ok", "sets": sets}} + + +def _scalar(value, **extra) -> dict: + return {"kind": "scalar", "value": value, "column": "gmv", "row": 0, "dims": {}, **extra} + + +def _cell(value, **dims) -> dict: + return {"value": value, "kind": "cell", "artifact": "a1", "row": 0, "column": "gmv", "dims": dims} + + +def _bare(text: str) -> list[dict]: + from app.eval.score import _numbers + + return _numbers(text) + + +EXP = {"id": "e1", "kind": "scalar", "column": "gmv", "required": True} + + +def test_scalar_outcomes(): + case = _case(EXP) + gold = _gold({"e1": _scalar(330.0)}) + traced = score_case(case, gold, Answer(run_id="r", values=[_cell(330.0)]), path="ran") + assert traced.verdict == "pass" and traced.traced and traced.scores[0]["outcome"] == "traced" + untraced = score_case(case, gold, Answer(run_id="r", text="销售额 330 元", bare=_bare("销售额 330 元")), path="ran") + assert untraced.verdict == "pass" and not untraced.traced and untraced.scores[0]["outcome"] == "untraced" + wrong = score_case(case, gold, Answer(run_id="r", values=[_cell(310.0)]), path="ran") + assert wrong.verdict == "fail" and wrong.scores[0]["outcome"] == "wrong" and wrong.scores[0]["got"] == 310.0 + # 差 100 倍：量纲不同，单列，也不算对 + scale = score_case(case, gold, Answer(run_id="r", values=[_cell(3.3)]), path="ran") + assert scale.verdict == "fail" and scale.scores[0]["outcome"] == "scale" + masked = score_case(case, gold, Answer(run_id="r", values=[{**_cell(330.0), "masked": True}]), path="ran") + assert masked.verdict == "ungradable" and "遮罩" in masked.scores[0]["reason"] + + +def test_bare_numbers_follow_their_written_precision_and_percent(): + case = _case(EXP) + # 3.3 万写到千位：33,000 ± 500 都算对 + gold = _gold({"e1": _scalar(33280.0)}) + assert score_case(case, gold, Answer(run_id="r", bare=_bare("约 3.3 万")), path="ran").verdict == "pass" + # 比率写成百分数 + ratio = _gold({"e1": _scalar(0.125)}) + got = score_case(case, ratio, Answer(run_id="r", bare=_bare("占比 12.5%")), path="ran") + assert got.verdict == "pass" and got.scores[0]["outcome"] == "untraced", "叙述里写的比率只算对但没引用" + + +def test_dimensions_pair_and_small_integers_are_weak(): + case = _case({**EXP, "where": {"region": "华东"}}) + gold = _gold({"e1": _scalar(300.0, where={"region": "华东"}, dims={"region": "华东"})}) + other = score_case(case, gold, Answer(run_id="r", values=[_cell(300.0, region="华南")]), path="ran") + assert other.verdict == "fail", "华南那一格碰巧也是 300，不算" + same = score_case(case, gold, Answer(run_id="r", values=[_cell(300.0, 地区="华东")]), path="ran") + assert same.verdict == "pass" and same.scores[0]["key"] == {"region": "华东"} + # 按条件筛过、只剩一个数（没有维度列）不算对不上 + assert score_case(case, gold, Answer(run_id="r", values=[_cell(300.0)]), path="ran").verdict == "pass" + weak = score_case(_case(EXP), _gold({"e1": _scalar(3)}), Answer(run_id="r", values=[_cell(3)]), path="ran") + assert weak.verdict == "pass" and weak.scores[0]["weak"] is True + + +def test_tables_sets_none_and_ask(): + table = {"id": "t", "kind": "table", "keys": ["region"], "ordered": True, "required": True} + gold = _gold({"t": {"kind": "table", "keys": ["region"], "columns": ["region", "gmv"], "ordered": True, + "rows": [{"region": "华东", "gmv": 300.0}, {"region": "华北", "gmv": 30.0}]}}) + cells = [_cell(300.0, region="华东"), _cell(30.0, region="华北")] + ok = score_case(_case(table), gold, Answer(run_id="r", text="华东 300，华北 30", values=cells), path="ran") + assert ok.verdict == "pass" and ok.traced + flipped = score_case(_case(table), gold, Answer(run_id="r", text="华北 30，华东 300", values=cells), path="ran") + assert flipped.verdict == "fail" and "先后" in flipped.scores[0]["reason"] + + names = {"id": "s", "kind": "set", "column": "region", "required": True} + gold = _gold({"s": {"kind": "set", "column": "region", "values": ["华东", "华北"], "ordered": False}}) + assert score_case(_case(names), gold, Answer(run_id="r", text="有华东、华北两个"), path="ran").verdict == "pass" + assert score_case(_case(names), gold, Answer(run_id="r", text="只有华东"), path="ran").verdict == "fail" + + none = {"id": "n", "kind": "none", "required": True} + gold = _gold({"n": {"kind": "none", "rows": 0}}) + empty = [{"artifact": "q", "row_count": 0, "sql": "SELECT 1"}] + traced = score_case(_case(none), gold, Answer(run_id="r", text="华中区没有数据", queries=empty), path="ran") + assert traced.verdict == "pass" and traced.traced + with_number = score_case(_case(none), gold, Answer(run_id="r", text="华中区 12 元", bare=_bare("华中区 12 元"), + queries=empty), path="ran") + assert with_number.verdict == "fail" + + ask = _case({"id": "a", "kind": "ask", "required": True}, gold_kind=None) + assert score_case(ask, {"status": "ok"}, Answer(run_id=None), path="asked").verdict == "pass" + assert score_case(ask, {"status": "ok"}, Answer(run_id="r"), path="ran").verdict == "fail" + + +def test_the_path_and_the_gold_decide_before_numbers(): + case = _case(EXP) + gold = _gold({"e1": _scalar(330.0)}) + replied = score_case(case, gold, Answer(run_id=None, text="330", bare=_bare("330")), path="reply") + assert replied.verdict == "fail" and "该查不查" in replied.reason, "直接回答的数是没查过的" + skipped = score_case(case, {"status": "skipped", "reason": "口径已升版到 v2，待处置"}, Answer(run_id=None), path=None) + assert skipped.verdict == "skipped" and "待处置" in skipped.reason + drifted = score_case(case, {"status": "drifted", "reason": "作答期间数据变了"}, Answer(run_id="r"), path="ran") + assert drifted.verdict == "ungradable", "数据变了不判，不算答错" + rejected = score_case(case, gold, Answer(run_id="r"), path="rejected") + assert rejected.verdict == "ungradable", "驳回是人的决定，不算失败" + text_only = _case({"id": "p", "kind": "text", "points": ["说明口径"], "required": False}, gold_kind=None) + judged = {"expectations": {"p": {"points": [{"point": 1, "verdict": "covered", "reason": ""}]}}} + scored = score_case(text_only, {"status": "ok"}, Answer(run_id="r", text="按支付日"), path="ran", judgement=judged) + assert scored.verdict == "ungradable" and scored.scores[0]["outcome"] == "covered", "文字要点单列、不计准确率" + unjudged = score_case(text_only, {"status": "ok"}, Answer(run_id="r"), path="ran") + assert unjudged.scores[0]["outcome"] == "unjudged" + + +def test_dual_reports_both_versions(): + case = _case(EXP, gold_kind="caliber") + case["gold"] = {"kind": "caliber", "workflow_id": "w", "version": 1, "node_id": "card", "inputs": {}} + case["expectations"] = [{"id": "e1", "kind": "scalar", "metric": "gmv", "required": True}] + gold = _gold({"e1": _scalar(330.0)}, {"e1": _scalar(380.0), "_version": 2}) + got = score_case(case, gold, Answer(run_id="r", values=[_cell(330.0)]), path="ran") + assert got.verdict == "pass" and got.by_version[0]["version"] == 2 and got.by_version[0]["verdict"] == "fail" + + +def test_the_tolerance_is_the_dual_check_function(): + from app.engine.nodes import dual_check + + assert dual_check._agree is tolerance.agree and dual_check._delta_pct is tolerance.delta_pct + assert tolerance.agree(100.0, 100.001, 1e-6) is False and tolerance.agree(100.0, 100.001, 1e-3) + assert tolerance.agree(0, 0.0000001, 1e-6) and tolerance.delta_pct(200, 180) == -10.0 + assert tolerance.delta_pct(0, 1) is None and tolerance.agree("华东", "华东", 0) + + +# -------------------------------------------------------------------------- +# 读封存的答案、逐题 E1、重评（走一次真的运行） +# -------------------------------------------------------------------------- + + +@pytest.fixture +async def engine_on(): + from app.engine.runner import run_manager + + await run_manager.setup() + yield + await run_manager.shutdown() + + +async def test_reading_the_sealed_answer_and_the_per_case_sql_match(engine_on, monkeypatch, tmp_path): + from app.eval import score + from tests import ledger_samples as ls + + await ls.fresh_source(tmp_path / "shop.db") + run = await ls.explore(monkeypatch, ls.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]，同比增长 12%。") + answer = await score.read_answer(run.id) + assert answer.problem is None and answer.report == "write" + [value] = answer.values + assert value["value"] == 170.5 and value["kind"] == "cell" and value["column"] == "销售额" + assert [b["raw"] for b in answer.bare] == ["12%"] and answer.stats["numbers_cited"] == 1 + assert len(answer.queries) == 1 and answer.cited == [answer.queries[0]["artifact"]] + + from app.db.base import SessionLocal + from app.db.models import DataSource + from sqlalchemy import select + + async with SessionLocal() as session: + sid = (await session.execute(select(DataSource.id).where(DataSource.name == ls.CURRENT["source"]))).scalar() + case = {"gold": {"kind": "sql", "datasource_id": sid, "sql": ls.WITH_REFUNDS}, "expectations": []} + gold = {"status": "ok", "before": {"rendered": {"sql": ls.WITH_REFUNDS}, "sets": []}} + match = await score.sql_match(case, gold, answer) + assert match["status"] == "related" and match["cited"] and "filter" in match["aspects"], match + gold["before"]["rendered"]["sql"] = ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' " + "AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'") + assert (await score.sql_match(case, gold, answer))["status"] == "proven" + + +async def test_rescoring_reads_the_seal_and_never_calls_a_model(engine_on, monkeypatch, tmp_path): + from app.db.base import SessionLocal + from app.db.models_eval import EvalResult, EvalRun + from app.eval import rescore + from app.providers import mock_model + from tests import eval_samples as es + from tests import ledger_samples as ls + + await ls.fresh_source(tmp_path / "shop.db") + run = await ls.explore(monkeypatch, ls.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。") + suite_id = await es.make_suite("none", cases=[{"question": "本周销售额？", "expectations": [ + {"kind": "scalar", "column": "gmv"}], "gold": {"kind": "sql", "datasource_id": "none", "sql": "SELECT 1"}}]) + + def boom(*_a, **_k): + raise AssertionError("重评不该调模型") + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", boom) + async with SessionLocal() as session: + from app.db.models_eval import EvalCase + from sqlalchemy import select + + case = (await session.execute(select(EvalCase).where(EvalCase.suite_id == suite_id))).scalar_one() + er = EvalRun(suite_id=suite_id, status="succeeded") + session.add(er) + await session.flush() + session.add(EvalResult(eval_run_id=er.id, case_id=case.id, case_hash=case.case_hash, status="ok", path="ran", + run_id=run.id, verdict="fail", scores=[], + gold=_gold({"e1": _scalar(170.5)}))) + await session.commit() + er_id = er.id + assert await rescore.rescore(er_id) == 1 + async with SessionLocal() as session: + row = (await session.execute(select(EvalResult).where(EvalResult.eval_run_id == er_id))).scalar_one() + assert row.verdict == "pass" and row.scores[0]["outcome"] == "traced" + assert (await session.get(EvalRun, er_id)).summary["accuracy"]["passed"] == 1 + await es.drop_suites() diff --git a/backend/tests/test_eval_stats.py b/backend/tests/test_eval_stats.py new file mode 100644 index 00000000..9f2f290a --- /dev/null +++ b/backend/tests/test_eval_stats.py @@ -0,0 +1,162 @@ +"""评测的统计与对比（app/eval/stats.py、app/eval/compare.py；细案 01 做法 11、12，评测-D10）。纯计算，和手算对照。 + +- Wilson 95% 区间、精确 McNemar 双侧 p 值、P50 / P90； +- 总表：准确率和可追溯准确率分开报（分母只算能判的）、作答路径、出具档位、复核、look、按角色的用量、引用率、逐题 E1、 + 重复 k 次的题给通过率； +- 和上一次比：逐题转移（退步 / 进步 / 不变 / 不可比）和归因（数据变、做法变、只参数变、路径变），列出指纹里哪几项变了； +- A/B 配对表：都对 / 都错 / 只 A 对 / 只 B 对，不能判的不进表； +- 选基准：点名的 > 同一变体里钉住的 > 同一变体、比这次先发起、已经结束的最近一次。 +""" +from __future__ import annotations + +from datetime import timedelta + +import pytest + +import app.db.models # noqa: F401 - conftest 按已导入的模型建表 +from app.eval import compare, stats + + +def test_wilson_and_mcnemar_match_hand_computation(): + assert stats.wilson(8, 10) == (0.4902, 0.9433) + assert stats.wilson(0, 0) is None + low, high = stats.wilson(0, 5) + assert low == 0.0 and 0.43 < high < 0.44 + assert stats.mcnemar_exact(1, 0) == 1.0 + assert stats.mcnemar_exact(6, 0) == 0.03125 + assert stats.mcnemar_exact(7, 2) == 0.179688 + assert stats.mcnemar_exact(0, 0) == 1.0 + assert stats.percentile([5, 1, 3, 2, 4], 50) == 3 and stats.percentile([5, 1, 3, 2, 4], 90) == 5 + assert stats.percentile([], 50) is None + + +def _row(case: str, verdict: str | None, *, path: str = "ran", status: str = "ok", repeat: int = 0, **extra) -> dict: + scores = [{"id": "e1", "kind": "scalar", "required": True, + "outcome": {"pass": "traced", "fail": "wrong"}.get(verdict or "", "ungradable")}] + return {"case_id": case, "case_seq": int(case[-1]) if case[-1].isdigit() else 1, "case_hash": f"h-{case}", + "verdict": verdict, "path": path, "status": status, "repeat": repeat, "scores": scores, + "traced": verdict == "pass", **extra} + + +def test_summary_counts_only_gradable_items_and_splits_roles(): + usage = {"assistant": {"by_model": {"m": {"calls": 3, "cost_usd": 0.03}}, "calls": 3, "cost_usd": 0.03, + "input_tokens": 10, "output_tokens": 5, "unpriced_calls": 0}, + "writer": {"by_model": {}, "calls": 1, "cost_usd": 0.01, "input_tokens": 1, "output_tokens": 1, + "unpriced_calls": 1}, + "judge": {"cost_usd": 0.0, "calls": 0}} + rows = [ + _row("c1", "pass", tier="formal", usage=usage, looks={"model": 3, "system": 0, "limit_hit": True}, + timings={"build_ms": 1000, "total_ms": 4000, "review_ms": 30_000}, review={"verdict": "annotated", + "severity": "degraded"}, + answer_stats={"numbers": 2, "numbers_cited": 1, "uncited_numbers": 1}, attempts=1, + sql_match={"status": "proven", "aspects": []}), + _row("c2", "fail", usage=usage, timings={"build_ms": 3000, "total_ms": 6000}, + sql_match={"status": "related", "aspects": ["time"]}), + _row("c3", "ungradable"), + _row("c4", None, path=None, status="skipped"), + ] + rows[0]["scores"][0]["outcome"] = "traced" + summary = stats.summarize(rows) + assert summary["accuracy"] == {"passed": 1, "n": 2, "rate": 0.5, "ci": stats.wilson(1, 2)} + assert summary["traced_accuracy"]["passed"] == 1 + assert summary["verdicts"] == {"pass": 1, "fail": 1, "ungradable": 1, "skipped": 0} + assert summary["paths"] == {"ran": 3} and summary["tiers"]["formal"] == 1 and summary["tiers"]["explore"] == 2 + assert summary["review"]["retries"] == 1 and summary["review"]["slow"] == 1 + assert summary["review"]["severity"] == {"degraded": 1} + assert summary["looks"]["model"] == 3 and summary["looks"]["limit_hit"] == round(1 / 3, 4) + assert summary["timings"]["build_ms"] == {"p50": 1000, "p90": 3000, "n": 2} + assert summary["usage"]["assistant"]["calls"] == 6 and summary["usage"]["writer"]["unpriced_calls"] == 2 + assert summary["cost_usd"] == 0.08 and summary["unpriced_calls"] == 2 + assert summary["usage"]["gold"] == {"cost_usd": 0.0, "calls": 0}, "期望探针不调模型" + assert summary["citations"]["rate"] == 0.5 + assert summary["e1"]["passed"] == 1 and summary["e1"]["n"] == 2 and summary["e1"]["aspects"] == {"time": 1} + + +def test_repeats_give_a_pass_rate_per_case(): + rows = [_row("c1", "pass", repeat=0), _row("c1", "fail", repeat=1), _row("c1", "pass", repeat=2)] + summary = stats.summarize(rows) + assert summary["pass_rates"]["c1"]["passed"] == 2 and summary["pass_rates"]["c1"]["n"] == 3 + assert compare.case_outcome(rows)["verdict"] == "pass", "多数决" + + +def test_transitions_and_their_causes(): + sql_a = [{"text": "SELECT SUM(x) FROM t WHERE d = ?", "literals": ["'2026-09-01'"]}] + sql_b = [{"text": "SELECT SUM(x) FROM t WHERE d = ?", "literals": ["'2026-08-01'"]}] + sql_c = [{"text": "SELECT SUM(x) FROM t WHERE e = ?", "literals": ["'2026-09-01'"]}] + base = [_row("c1", "pass", gold_key="g1", sql_shapes=sql_a), _row("c2", "pass", gold_key="g2", sql_shapes=sql_a), + _row("c3", "fail", gold_key="g3", sql_shapes=sql_a), _row("c4", "pass", gold_key="g4", path="ran"), + _row("c5", "pass", gold_key="g5"), _row("c6", "fail", gold_key="g6")] + target = [_row("c1", "fail", gold_key="g1-new", sql_shapes=sql_a), _row("c2", "fail", gold_key="g2", sql_shapes=sql_c), + _row("c3", "pass", gold_key="g3", sql_shapes=sql_b), _row("c4", "fail", gold_key="g4", path="reply"), + {**_row("c5", "fail", gold_key="g5"), "case_hash": "changed"}, _row("c6", "ungradable", gold_key="g6"), + _row("c7", "pass")] + moved = compare.transitions(base, target) + by = {i["case_id"]: i for i in moved["items"]} + assert (by["c1"]["transition"], by["c1"]["cause"]) == ("regressed", "data") + assert (by["c2"]["transition"], by["c2"]["cause"]) == ("regressed", "method") + assert (by["c3"]["transition"], by["c3"]["cause"]) == ("improved", "params") + assert (by["c4"]["transition"], by["c4"]["cause"]) == ("regressed", "path") + assert by["c5"]["transition"] == "incomparable" and "题目改过" in by["c5"]["reason"] + assert by["c6"]["transition"] == "incomparable" and by["c7"]["transition"] == "incomparable" + assert moved["counts"] == {"regressed": 3, "improved": 1, "same": 0, "incomparable": 3} + assert compare.sql_change(sql_a, sql_a) == "same" + + +def test_fingerprint_diff_lists_what_changed(): + a = {"assistant_model": "m1", "writer_model": None, "catalog": "live:aaa", "prompt": "p1", "schemas": {"s": "x"}, + "cases": "c", "scorer_version": 1} + b = {**a, "assistant_model": "m2", "schemas": {"s": "y"}} + diff = compare.fingerprint_diff(a, b) + assert [d["key"] for d in diff] == ["assistant_model", "schemas"] + assert diff[0] == {"key": "assistant_model", "label": "助手模型", "before": "m1", "after": "m2"} + assert "before" not in diff[1], "表结构只说变没变" + + +def test_the_pair_table(): + a = [_row("c1", "pass"), _row("c2", "pass"), _row("c3", "fail"), _row("c4", "ungradable")] + b = [_row("c1", "pass"), _row("c2", "fail"), _row("c3", "fail"), _row("c4", "pass")] + table = compare.pair_table([("当前目录", a), ("关目录", b)]) + [pair] = table["pairs"] + assert (pair["both"], pair["neither"], pair["only_a"], pair["only_b"], pair["excluded"]) == (1, 1, 1, 0, 1) + assert pair["p_value"] == 1.0 and pair["note"] == compare.NO_DIFFERENCE + assert table["arms"][0]["accuracy"]["rate"] == round(2 / 3, 4) + + +async def test_picking_the_baseline(): + from app.db.base import SessionLocal, utcnow + from app.db.models_eval import EvalRun, EvalSuite + + now = utcnow() + async with SessionLocal() as session: + suite = EvalSuite(name="虚构题集-基准") + session.add(suite) + await session.flush() + + def run(minutes: int, key: str, status: str = "succeeded", **extra) -> EvalRun: + row = EvalRun(suite_id=suite.id, status=status, fingerprint={"variant_key": key}, + created_at=now - timedelta(minutes=minutes), **extra) + session.add(row) + return row + + old = run(50, "k1") + pinned = run(40, "k1", pinned=True) + latest = run(30, "k1") + run(20, "k2") + run(15, "k1", status="failed") + target = run(10, "k1") + later = run(5, "k1") + await session.flush() + assert (await compare.pick_baseline(session, target)).id == pinned.id, "同一变体里钉住的优先" + pinned.pinned = False + await session.flush() + assert (await compare.pick_baseline(session, target)).id == latest.id, "比这次先发起、已结束的最近一次" + assert (await compare.pick_baseline(session, target, explicit=old.id)).id == old.id + assert (await compare.pick_baseline(session, target, explicit=later.id)).id == later.id + await session.delete(suite) + await session.commit() + + +@pytest.mark.parametrize("n, k", [(1, 0), (3, 3)]) +def test_rates_carry_their_sample_size(n, k): + r = stats.rate(k, n) + assert r["n"] == n and r["passed"] == k and r["ci"] == stats.wilson(k, n)