yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 00:11:59 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
AgentLab/backend/tests/test_eval_score.py
T

358 lines
22 KiB
Python
Raw Permalink Blame History

"""评分（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_a_wrong_answer_shows_the_paired_cell():
"""判错时「答案里的数」给评分时配对上的那一格（按列名、指标、维度配），不是答案引用的第一个值；配不上照实写配不上。"""
case = _case({**EXP, "column": "refund", "label": "退款金额"})
gold = _gold({"e1": _scalar(1234.5, column="refund")})
def cell(value, column, **dims):
return {"value": value, "kind": "cell", "artifact": "a1", "row": 0, "column": column, "dims": dims}
question = cell("上周退款金额是多少", "问题")
both = score_case(case, gold, Answer(run_id="r", values=[question, cell(36, "单数"), cell(1200.0, "refund")]),
path="ran")
assert both.verdict == "fail" and both.scores[0]["got"] == 1200.0, both.scores
assert both.scores[0]["cell"]["column"] == "refund"
labelled = score_case(case, gold, Answer(run_id="r", values=[cell(36, "单数"), cell(1200.0, "退款金额")]), path="ran")
assert labelled.scores[0]["got"] == 1200.0, "期望的说明（label）和答案的列名一样也算配上"
lone = score_case(case, gold, Answer(run_id="r", values=[question, cell(36, "单数")]), path="ran")
[s] = lone.scores
assert s["outcome"] == "wrong" and "got" not in s, "单数不是退款金额：配不上就不给「答案里的数」"
assert "配不上" in s["reason"] and "单数 36" in s["reason"] and "上周退款" not in s["reason"], s["reason"]
# 按维度配：华东那一行只有一个数时就是它；同一行有几个数、列名又都配不上时不猜
by_dims = _case({**EXP, "column": "refund", "where": {"region": "华东"}})
gold = _gold({"e1": _scalar(1234.5, column="refund", where={"region": "华东"}, dims={"region": "华东"})})
one = score_case(by_dims, gold, Answer(run_id="r", values=[cell(1200.0, "total", region="华东")]), path="ran")
assert one.scores[0]["got"] == 1200.0, one.scores
two = score_case(by_dims, gold, Answer(run_id="r", values=[cell(36, "单数", region="华东"),
cell(1200.0, "total", region="华东")]), path="ran")
assert "got" not in two.scores[0] and "配不上" in two.scores[0]["reason"], two.scores
text_only = score_case(case, _gold({"e1": _scalar(1234.5, column="refund")}), Answer(run_id="r", values=[question]),
path="ran")
assert "got" not in text_only.scores[0] and text_only.scores[0]["reason"] == "答案里没有对得上的数"
def test_none_counts_only_numbers_as_giving_data():
"""期望「没有数据」：只有引用了数值格、或叙述里写了数，才算答案给了数。引用的是文字格（地区名）、空值格（NULL）的，
答案明说没有数据就按没有数据判对。"""
none = _case({"id": "n", "kind": "none", "required": True})
gold = _gold({"n": {"kind": "none", "rows": 0}})
empty = [{"artifact": "q", "row_count": 0, "sql": "SELECT 1"}]
text_cell = {"value": "华中", "kind": "cell", "artifact": "q", "row": 0, "column": "region", "dims": {}}
named = score_case(none, gold, Answer(run_id="r", text="华中区没有数据", values=[text_cell], queries=empty), path="ran")
assert named.verdict == "pass" and named.scores[0]["outcome"] == "traced", named.scores
null_cell = {"value": None, "kind": "cell", "artifact": "q", "row": 0, "column": "gmv", "dims": {}}
null = score_case(none, gold, Answer(run_id="r", text="华中区的销售额为空，没有数据", values=[null_cell],
queries=empty), path="ran")
assert null.verdict == "pass", null.scores
number = score_case(none, gold, Answer(run_id="r", text="华中区 12 元", values=[_cell(12.0)], queries=empty),
path="ran")
assert number.verdict == "fail" and number.scores[0]["got"] == 12.0, number.scores
said = score_case(none, gold, Answer(run_id="r", text="华中区 12 元", values=[text_cell], bare=_bare("12 元"),
queries=empty), path="ran")
assert said.verdict == "fail" and said.scores[0]["raw"] == "12", said.scores
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_the_per_case_sql_match_compares_the_query_behind_the_answer(engine_on, monkeypatch, tmp_path):
"""逐题 E1 比的是产出答案的那条查询：答案引用了就比被引用的；没引用、这次运行又查了几条时不硬比第一条（往往是探索
查询），照实写没法对上是哪条；没引用、只查了一条时它就是唯一可能的那条，按它比、写明依据。"""
from app.db.base import SessionLocal
from app.db.models import DataSource
from app.engine.runner import run_manager
from app.eval import score
from app.ledger import match as ledger_match
from sqlalchemy import select
from tests import ledger_samples as ls
await ls.fresh_source(tmp_path / "shop.db")
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": "x"}, "expectations": []}
gold = {"status": "ok", "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'")}, "sets": []}}
# 先探了一下有哪些地区，再查销售额；答案没引用查询（数写在叙述里）
graph = ls.chain(ls.node("start", "input"),
ls.node("peek", "tool", tool=ls.tool(), args={"sql": "SELECT DISTINCT region FROM orders"}),
ls.node("q", "tool", tool=ls.tool(), args={"sql": ls.WITH_REFUNDS}),
ls.node("write", "report", instructions="写答案"),
ls.node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}]))
ls.script(monkeypatch, "本周销售额约 200 元。")
started = await run_manager.start(graph=graph, input_payload={}, run_class="exploratory", origin_kind="canvas")
await ls.wait(started.id)
await ledger_match.drain()
answer = await score.read_answer(started.id)
assert answer.cited == [] and len(answer.queries) == 2, answer.queries
match = await score.sql_match(case, gold, answer)
assert match["status"] == "skipped" and match["cited"] is False and "没有引用查询" in match["reason"], match
assert match.get("query") is None and match.get("aspects") in (None, []), "没法对上是哪条，就不给差在哪"
# 只查了一条、答案没引用：唯一的那条就是产出答案的，按它比，依据写明
run = await ls.explore(monkeypatch, ls.EQUIVALENT, text="本周销售额 170.5 元。")
only = await score.read_answer(run.id)
assert only.cited == [] and len(only.queries) == 1
match = await score.sql_match(case, gold, only)
assert match["status"] == "proven" and match["basis"] == "only_query" and not match["cited"], match
async def test_the_per_case_sql_match_projects_to_the_cited_columns(engine_on, monkeypatch, tmp_path):
"""逐题 E1 和账本同一个口径：答案的 SQL 多查了一列订单数、报告只引用销售额那一列时，按引用的列投影后和期望 SQL
证明等价；多出来的那列不算差在哪个方面。"""
from app.db.base import SessionLocal
from app.db.models import DataSource
from app.eval import score
from sqlalchemy import select
from tests import ledger_samples as ls
await ls.fresh_source(tmp_path / "shop.db")
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": "x"}, "expectations": []}
gold = {"status": "ok", "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'")}, "sets": []}}
wider = ("select count(*) as 单数, sum(o.amount) as 销售额 from orders o where o.paid_at >= '2026-09-01' "
"and o.status = 'paid' and o.paid_at < '2026-09-08'")
run = await ls.explore(monkeypatch, wider, text="本周销售额 [[v:Q1.r0.销售额]] 元。")
answer = await score.read_answer(run.id)
assert answer.cited, answer
match = await score.sql_match(case, gold, answer)
assert match["status"] == "proven" and match["cited"] is True, match
assert "projection" not in (match.get("aspects") or []), match
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, score
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"
rescored = await session.get(EvalRun, er_id)
assert rescored.summary["accuracy"]["passed"] == 1 and rescored.summary["rescored_at"]
assert rescored.fingerprint["scorer_version"] == score.SCORER_VERSION
await es.drop_suites()
Reference in New Issue View Git Blame Copy Permalink