From a7e91b16a11b53817434fd11adbc8655312dfc8e Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 13:58:36 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=84=E6=B5=8B=E6=8E=A5=E5=8F=A3=E4=B8=8E?= =?UTF-8?q?=E9=A2=98=E9=9B=86=E5=AF=BC=E5=87=BA=E5=AF=BC=E5=85=A5=E3=80=81?= =?UTF-8?q?=E9=9B=B6=E5=94=AE=E6=A0=B7=E4=BE=8B=E9=A2=98=E9=9B=86=EF=BC=88?= =?UTF-8?q?=E7=AC=AC=202=20=E6=B3=A2=20W2-c=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - api/evals.py（骨架已登记、只给搭建者）：题集、题目的增删改查（改要带读到的 version，对不上 409；有评测记录的题集不能删， 只能归档）。保存题目时试算一次（细案 01 做法 8）：没过只读守卫、用到遮罩列、取不到值、运行失败、超时、口径升版待处置的， 照实回 422（带机器码和 field）、不保存；只改标签、启用的不重新试算。另有只试算不保存、试算已有的题。 - 从运行认定：GET /runs/{id}/queries 列出那次运行的查询（只给 SQL、列名、行数，不给格），POST …/cases/from-run 拷出一条 查询的 SQL 建题，记确认人、时间、出自哪次运行；客户端不能自己写 from_run，要改 SQL 就改成 SQL 题。 - 口径：GET /calibers 和蓝图助手同一份口径清单（能不能复用、为什么）；GET/POST …/upgrade 看、处置口径升版， recompute / dual / incomparable 三选一、没有缺省；不可比的旧题归档、按新版本另建新题。 - 目录快照：当前目录（当场写好）、影子起草的计划和估算、后台影子起草（202）、查看、停止。 - app/eval/transfer.py：题集导出成 JSON（数据源、工作流按名字；不含期望值、答案、运行，从运行认定只带确认人和时间），导入 按名字对上、逐题过同一套校验和同一道试算关，对不上、没过的跳过并写原因，一道都导不进来就不建题集。 - tests/fixtures/eval/retail_suite.json：零售样例题集，配 tests/fixtures/acceptance/retail.py（数据源名 retail_demo， retail.SOURCE_NAME）：支付日归属、退款、取消、华南高于华东，一道「华中区」期望没有数据，一道问法含糊期望反问。测试在 按今天生成的零售库上导入，六道题都试算通过、期望和直接查库一样、陷阱按口径算。 --- backend/app/api/evals.py | 686 +++++++++++++++++- backend/app/db/models_eval.py | 2 +- backend/app/eval/transfer.py | 255 +++++++ backend/tests/fixtures/acceptance/retail.py | 5 + backend/tests/fixtures/eval/retail_suite.json | 172 +++++ backend/tests/test_eval_cases.py | 130 ++++ backend/tests/test_evals_api.py | 503 +++++++++++++ 7 files changed, 1749 insertions(+), 4 deletions(-) create mode 100644 backend/app/eval/transfer.py create mode 100644 backend/tests/fixtures/eval/retail_suite.json create mode 100644 backend/tests/test_eval_cases.py create mode 100644 backend/tests/test_evals_api.py diff --git a/backend/app/api/evals.py b/backend/app/api/evals.py index 21238230..c3f5fdc8 100644 --- a/backend/app/api/evals.py +++ b/backend/app/api/evals.py @@ -1,9 +1,689 @@ -"""评测接口（docs/design/next-round/01-效果度量与回归评测.md）：题集、题目、期望答案试算、评测、结果、对比。 +"""评测接口（docs/design/next-round/01-效果度量与回归评测.md）：题集、题目、期望答案试算、目录快照；第 3 波加评测、结果、对比。 -整组只给搭建者及以上（main.py 挂 require_role("builder")）。第 2 波 W2-c 起填。 +整组只给搭建者及以上（main.py 挂 require_role("builder")）：期望值和查询原文只给搭建者。这一波没有界面（第 4 波）， +接口的形状就是给界面的契约。请求体自己解析、自己校验，错误一律回中文 detail + 机器码（422 {detail, code, field}， +同 api/coded.py），表单据 field 把报错落到那一格。 + +- 题集：GET/POST /suites，GET/PATCH/DELETE /suites/{id}（改要带读到的 version，对不上 409；有评测记录的不能删，只能归档）； + GET /suites/{id}/export、POST /suites/import/preview、POST /suites/import（app/eval/transfer.py）。 +- 题目：POST /suites/{id}/cases、PATCH/DELETE /suites/{id}/cases/{cid}（改要带 version）。保存时试算一次（app/eval/ + gold.py）：没过只读守卫、用到遮罩列、取不到值、运行失败、超时、口径升版待处置的，照实回 422、不保存。 + POST /suites/{id}/cases/trial 只试算不保存（界面上「试算」按钮），POST /suites/{id}/cases/{cid}/trial 试算已有的题并记下。 +- 从运行认定：GET /runs/{run_id}/queries 列出那次运行里的查询，POST /suites/{id}/cases/from-run 拷出一条查询的 SQL 建题 + （记确认人、时间、出自哪次运行），之后每次评测按这条 SQL 重算。 +- 口径：GET /calibers 列出能选的口径（和蓝图助手同一份清单：每个工作流最新的已发布未退役版本里的源头口径卡，能不能复用、 + 为什么不能）；GET/POST /suites/{id}/cases/{cid}/upgrade 看、处置口径升版（recompute / dual / incomparable 三选一）。 +- 目录快照：GET/POST /catalog-snapshots（当前目录）、POST /catalog-snapshots/shadow/estimate（影子起草的计划和估算）、 + POST /catalog-snapshots/shadow（后台起草，202）、GET /catalog-snapshots/{id}、POST /catalog-snapshots/{id}/stop。 """ from __future__ import annotations -from fastapi import APIRouter +import asyncio +from typing import Any + +from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response +from sqlalchemy import func, select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.coded import CodedHTTPException +from app.api.deps import actor_of +from app.db.base import get_session, new_id, utcnow +from app.db.models_eval import EvalCase, EvalCatalogSnapshot, EvalRun, EvalSuite +from app.eval import cases as case_rules +from app.eval import gold as gold_rules +from app.eval import snapshots, transfer +from app.eval.cases import CaseInvalid, case_fields, case_hash, normalize_case, resolve_case +from app.eval.suites import effective_budget, normalize_suite router = APIRouter(prefix="/api/evals", tags=["evals"]) + +#: 机器码 +VERSION_CONFLICT = "eval_version_conflict" +SUITE_NOT_FOUND = "eval_suite_not_found" +CASE_NOT_FOUND = "eval_case_not_found" +SUITE_HAS_RUNS = "eval_suite_has_runs" +RUN_NOT_FOUND = "eval_run_query_not_found" +SNAPSHOT_NOT_FOUND = "eval_snapshot_not_found" +IMPORT_INVALID = "eval_import_invalid" +#: 导入时同时试算几道题 +IMPORT_TRIALS = 4 + + +def _invalid(e: CaseInvalid) -> CodedHTTPException: + return CodedHTTPException(422, e.message, e.code, field=e.field) + + +async def _body(request: Request) -> dict[str, Any]: + try: + data = await request.json() + except ValueError: + data = None + if not isinstance(data, dict): + raise CodedHTTPException(422, "请求内容需要是 JSON 对象", case_rules.CASE_INVALID) + return data + + +def _version(body: dict[str, Any], current: int, what: str) -> int: + version = body.get("version") + if isinstance(version, bool) or not isinstance(version, int): + raise CodedHTTPException(422, "保存时需要带上读到的版本号 version，请刷新后重试", case_rules.CASE_INVALID, + field="version") + if version != current: + raise CodedHTTPException(409, f"这个{what}已被修改（当前是第 {current} 版，提交的是第 {version} 版），" + "请刷新后再保存", VERSION_CONFLICT) + return version + + +def _iso(moment: Any) -> str | None: + return moment.isoformat() if moment is not None else None + + +# ========================================================================== +# 形状 +# ========================================================================== + + +def _suite_out(row: EvalSuite, *, counts: dict[str, int] | None = None) -> dict[str, Any]: + return {"id": row.id, "name": row.name, "description": row.description, "timezone": row.timezone, + "default_variant": dict(row.default_variant or {}), "budget": dict(row.budget or {}), + "effective_budget": effective_budget(row.budget), "archived": bool(row.archived), "version": row.version, + "created_by": row.created_by, "created_at": _iso(row.created_at), "updated_at": _iso(row.updated_at), + **({"counts": counts} if counts is not None else {})} + + +async def _source_names(session: AsyncSession, rows: list[EvalCase]) -> dict[str, str]: + from app.db.models import DataSource + + ids = {s for r in rows for s in (r.datasource_ids or [])} | { + (r.gold or {}).get("datasource_id") for r in rows if (r.gold or {}).get("datasource_id")} + if not ids: + return {} + return dict((await session.execute(select(DataSource.id, DataSource.name).where(DataSource.id.in_(ids)))).all()) + + +async def _upgrade_of(session: AsyncSession, row: EvalCase) -> dict[str, Any] | None: + """口径题钉的那一版现在怎么样（gold.upgrade_state）；读不出那一版的照实写原因。别的题为 None。""" + from app.ledger import derive + + gold = row.gold or {} + if gold.get("kind") != "caliber" or row.archived: + return None + try: + pinned = await derive.load_pinned(session, gold["workflow_id"], gold["version"], gold["node_id"]) + except derive.NotDerivable as e: + return {"pinned": gold.get("version"), "pending": True, "message": e.message, "code": e.code} + return (await gold_rules.upgrade_state(session, gold, pinned)).to_dict() + + +def _case_out(row: EvalCase, names: dict[str, str], upgrade: dict[str, Any] | None = None) -> dict[str, Any]: + """gold 原样给（改题时整份带回来就行）；它查的数据源的名字另放在 gold_source。""" + gold_source = (row.gold or {}).get("datasource_id") + return {**case_fields(row), + "gold_source": {"id": gold_source, "name": names.get(gold_source)} if gold_source else None, + "datasources": [{"id": s, "name": names.get(s)} for s in row.datasource_ids or []], + "archived": bool(row.archived), "superseded_by": row.superseded_by, "version": row.version, + "case_hash": row.case_hash, "trial": row.trial, "upgrade": upgrade, "created_by": row.created_by, + "created_at": _iso(row.created_at), "updated_at": _iso(row.updated_at)} + + +async def _cases_out(session: AsyncSession, rows: list[EvalCase]) -> list[dict[str, Any]]: + names = await _source_names(session, rows) + return [_case_out(r, names, await _upgrade_of(session, r)) for r in rows] + + +# ========================================================================== +# 读写的小件 +# ========================================================================== + + +async def _suite(session: AsyncSession, suite_id: str) -> EvalSuite: + row = await session.get(EvalSuite, suite_id) + if row is None: + raise CodedHTTPException(404, "题集不存在，可能已被删除", SUITE_NOT_FOUND) + return row + + +async def _case(session: AsyncSession, suite_id: str, case_id: str) -> EvalCase: + row = await session.get(EvalCase, case_id) + if row is None or row.suite_id != suite_id: + raise CodedHTTPException(404, "题目不存在，可能已被删除", CASE_NOT_FOUND) + return row + + +def _context(suite: EvalSuite, actor: str | None, **extra: Any) -> gold_rules.Context: + return gold_rules.Context(as_of=utcnow(), tz=suite.timezone, actor=actor, suite_id=suite.id, suite_name=suite.name, + role="gold", extra={"trial": True, **extra}) + + +async def _trial(suite: EvalSuite, fields: dict[str, Any], actor: str | None, *, case_id: str | None = None, + seq: int | None = None) -> gold_rules.GoldOutcome: + """按此刻试算一次期望。""" + return await gold_rules.compute_gold({**fields, "id": case_id, "seq": seq}, _context(suite, actor)) + + +def _trial_record(outcome: gold_rules.GoldOutcome, actor: str | None) -> dict[str, Any]: + return {**outcome.summary(), "at": utcnow().isoformat(), "by": actor} + + +def _gate(outcome: gold_rules.GoldOutcome) -> None: + """保存题目的那一道关：试算没过就照实回 422、不保存。""" + if outcome.ok: + return + reason = outcome.reason or "试算没有通过" + if outcome.status == "skipped": + reason += "：新建、修改的口径题请固定最新的已发布版本，或在这道题上选择升版的处置" + raise CodedHTTPException(422, f"试算没有通过，题目没有保存：{reason}", outcome.code or gold_rules.GOLD_FAILED, + field="gold") + + +async def _next_seq(session: AsyncSession, suite_id: str) -> int: + top = await session.scalar(select(func.max(EvalCase.seq)).where(EvalCase.suite_id == suite_id)) + return int(top or 0) + 1 + + +async def _insert_case(session: AsyncSession, suite: EvalSuite, fields: dict[str, Any], actor: str | None, + trial: dict[str, Any] | None) -> EvalCase: + """建一道题（提交）。题号撞了（两个请求同时建）重取一次。""" + for _attempt in range(3): + row = EvalCase(suite_id=suite.id, seq=await _next_seq(session, suite.id), case_hash=case_hash(fields), + trial=trial, created_by=actor, **fields) + session.add(row) + try: + await session.commit() + except IntegrityError: + await session.rollback() + continue + await session.refresh(row) + return row + raise HTTPException(409, "题集刚被别人加了题，请重试") + + +# ========================================================================== +# 题集 +# ========================================================================== + + +@router.get("/suites") +async def list_suites(archived: bool = False, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + query = select(EvalSuite).order_by(EvalSuite.updated_at.desc()) + if not archived: + query = query.where(EvalSuite.archived.is_(False)) + rows = list((await session.execute(query)).scalars()) + counts: dict[str, dict[str, int]] = {r.id: {"cases": 0, "enabled": 0} for r in rows} + if rows: + for suite_id, enabled, n in (await session.execute( + select(EvalCase.suite_id, EvalCase.enabled, func.count()).where( + EvalCase.suite_id.in_(list(counts)), EvalCase.archived.is_(False)) + .group_by(EvalCase.suite_id, EvalCase.enabled))).all(): + counts[suite_id]["cases"] += n + if enabled: + counts[suite_id]["enabled"] += n + return {"suites": [_suite_out(r, counts=counts[r.id]) for r in rows]} + + +@router.post("/suites", status_code=201) +async def create_suite(request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + try: + values = normalize_suite(await _body(request)) + except CaseInvalid as e: + raise _invalid(e) from e + row = EvalSuite(**values, created_by=actor_of(x_actor)) + session.add(row) + await session.commit() + await session.refresh(row) + return _suite_out(row, counts={"cases": 0, "enabled": 0}) + + +@router.get("/suites/{suite_id}") +async def get_suite(suite_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """题集和它的题目（归档的题也在，archived 标着）。口径题带上升版状态：pending_upgrades 是待处置的题数（界面顶部提示）。""" + suite = await _suite(session, suite_id) + rows = list((await session.execute(select(EvalCase).where(EvalCase.suite_id == suite_id) + .order_by(EvalCase.seq))).scalars()) + out = await _cases_out(session, rows) + active = [c for c in out if not c["archived"]] + return {**_suite_out(suite, counts={"cases": len(active), "enabled": sum(1 for c in active if c["enabled"])}), + "cases": out, "pending_upgrades": sum(1 for c in active if (c["upgrade"] or {}).get("pending"))} + + +@router.patch("/suites/{suite_id}") +async def update_suite(suite_id: str, request: Request, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + suite = await _suite(session, suite_id) + body = await _body(request) + version = _version(body, suite.version, "题集") + current = {k: getattr(suite, k) for k in ("name", "description", "timezone", "default_variant", "budget", + "archived")} + try: + values = normalize_suite(body, current=current) + except CaseInvalid as e: + raise _invalid(e) from e + result = await session.execute(update(EvalSuite).where(EvalSuite.id == suite_id, EvalSuite.version == version) + .values(**values, version=version + 1, updated_at=utcnow()) + .execution_options(synchronize_session=False)) + if result.rowcount != 1: + await session.rollback() + raise CodedHTTPException(409, "这个题集刚被别人修改过，请刷新后再保存", VERSION_CONFLICT) + await session.commit() + await session.refresh(suite) + return await get_suite(suite_id, session) + + +@router.delete("/suites/{suite_id}", status_code=204) +async def delete_suite(suite_id: str, session: AsyncSession = Depends(get_session)) -> Response: + """删题集和它的题目。有评测记录的不能删：每次评测的总表永久保留，请改为归档。""" + suite = await _suite(session, suite_id) + runs = await session.scalar(select(func.count()).select_from(EvalRun).where(EvalRun.suite_id == suite_id)) + if runs: + raise CodedHTTPException(409, f"这个题集有 {runs} 次评测记录，不能删除（评测的总表要留着）。请改为归档", + SUITE_HAS_RUNS) + await session.delete(suite) + await session.commit() + return Response(status_code=204) + + +# ========================================================================== +# 题目 +# ========================================================================== + + +async def _validated(session: AsyncSession, suite: EvalSuite, body: dict[str, Any], *, + current: dict[str, Any] | None = None, trusted_gold: bool = False) -> dict[str, Any]: + try: + fields = normalize_case(body, current=current, trusted_gold=trusted_gold) + await resolve_case(session, fields, tz=suite.timezone) + except CaseInvalid as e: + raise _invalid(e) from e + return fields + + +@router.post("/suites/{suite_id}/cases", status_code=201) +async def create_case(suite_id: str, request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """建一道题：校验、试算一次（没过不保存），交回题目和这次试算的结果（含期望值）。""" + suite = await _suite(session, suite_id) + actor = actor_of(x_actor) + fields = await _validated(session, suite, await _body(request)) + outcome = await _trial(suite, fields, actor) + _gate(outcome) + row = await _insert_case(session, suite, fields, actor, _trial_record(outcome, actor)) + return {"case": _case_out(row, await _source_names(session, [row]), await _upgrade_of(session, row)), + "trial": outcome.to_dict()} + + +@router.post("/suites/{suite_id}/cases/trial") +async def trial_unsaved(suite_id: str, request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """只试算、不保存（界面上写 SQL、点选格或列时的「试算」）：交回试算结果，没过的照实写原因（不回 422）。""" + suite = await _suite(session, suite_id) + fields = await _validated(session, suite, await _body(request)) + return {"trial": (await _trial(suite, fields, actor_of(x_actor))).to_dict()} + + +@router.patch("/suites/{suite_id}/cases/{case_id}") +async def update_case(suite_id: str, case_id: str, request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """改一道题（带上读到的 version）。题目内容（指纹）变了要重新试算、没过不保存；只改标签、启用的不试算。""" + suite = await _suite(session, suite_id) + row = await _case(session, suite_id, case_id) + body = await _body(request) + version = _version(body, row.version, "题目") + actor = actor_of(x_actor) + fields = await _validated(session, suite, body, current=case_fields(row)) + new_hash = case_hash(fields) + outcome = None + values: dict[str, Any] = {**fields, "case_hash": new_hash} + if new_hash != row.case_hash: + outcome = await _trial(suite, fields, actor, case_id=row.id, seq=row.seq) + _gate(outcome) + values["trial"] = _trial_record(outcome, actor) + result = await session.execute(update(EvalCase).where(EvalCase.id == case_id, EvalCase.version == version) + .values(**values, version=version + 1, updated_at=utcnow()) + .execution_options(synchronize_session=False)) + if result.rowcount != 1: + await session.rollback() + raise CodedHTTPException(409, "这道题刚被别人修改过，请刷新后再保存", VERSION_CONFLICT) + await session.commit() + await session.refresh(row) + return {"case": _case_out(row, await _source_names(session, [row]), await _upgrade_of(session, row)), + "trial": outcome.to_dict() if outcome is not None else None} + + +@router.delete("/suites/{suite_id}/cases/{case_id}", status_code=204) +async def delete_case(suite_id: str, case_id: str, session: AsyncSession = Depends(get_session)) -> Response: + """删一道题。已有的评测结果照留（按题目 id 和指纹记着，不随题目删）。""" + row = await _case(session, suite_id, case_id) + await session.delete(row) + await session.commit() + return Response(status_code=204) + + +@router.post("/suites/{suite_id}/cases/{case_id}/trial") +async def trial_case(suite_id: str, case_id: str, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """按此刻试算一道已有的题，记进题目的 trial（不带期望值），交回试算结果（含期望值）。不改题目的 version。""" + suite = await _suite(session, suite_id) + row = await _case(session, suite_id, case_id) + actor = actor_of(x_actor) + outcome = await _trial(suite, case_fields(row), actor, case_id=row.id, seq=row.seq) + await session.execute(update(EvalCase).where(EvalCase.id == case_id).values(trial=_trial_record(outcome, actor)) + .execution_options(synchronize_session=False)) + await session.commit() + return {"trial": outcome.to_dict()} + + +# ========================================================================== +# 从运行认定 +# ========================================================================== + + +async def _run_queries(session: AsyncSession, run_id: str) -> list[dict[str, Any]]: + from app.core import artifact_store + from app.data.catalog import QUERY_SNAPSHOT_KIND + from app.db.models import Artifact + + rows = (await session.execute(select(Artifact).where(Artifact.run_id == run_id, Artifact.kind == QUERY_SNAPSHOT_KIND) + .order_by(Artifact.created_at))).scalars().all() + out = [] + for ref in rows: + try: + snap = artifact_store.load(ref.id) or {} + except ValueError: + continue + out.append({"artifact": ref.id, "node_id": ref.node_id, "source": snap.get("source"), + "source_id": (ref.meta or {}).get("source_id"), "sql": snap.get("sql"), + "columns": list(snap.get("columns") or []), "row_count": snap.get("row_count"), + "truncated": bool(snap.get("truncated")), + "masked": bool(snap.get("masked_at_query") or snap.get("masked_cells") or snap.get("masked_whole"))}) + return out + + +async def _readable_run(session: AsyncSession, run_id: str) -> Any: + from app.api.deps import can_read_run, current_principal + from app.db.models import Run + + run = await session.get(Run, run_id) + if run is None or not await can_read_run(current_principal(), run): + raise CodedHTTPException(404, "运行不存在，可能已被删除", RUN_NOT_FOUND) + return run + + +@router.get("/runs/{run_id}/queries") +async def run_queries(run_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """一次运行里执行过的查询（查询快照）：选一条认定成期望。只给 SQL、列名、行数，不给结果的格。""" + await _readable_run(session, run_id) + return {"queries": await _run_queries(session, run_id)} + + +@router.post("/suites/{suite_id}/cases/from-run", status_code=201) +async def create_from_run(suite_id: str, request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """从一次运行里认定一条查询：{run_id, query（查询快照的工件 id）, expectations, question?, tags?, datasource_ids?}。 + 拷出那条查询的 SQL 建 from_run 题，记确认人（就是这次请求的人）、时间、出自哪次运行；之后每次评测按这条 SQL 重算。 + 问法没给的取那次运行的问题。同样要试算通过才保存。""" + from app.db.models import DataSource + + suite = await _suite(session, suite_id) + body = await _body(request) + actor = actor_of(x_actor) + run_id, query = body.get("run_id"), body.get("query") + if not isinstance(run_id, str) or not isinstance(query, str): + raise CodedHTTPException(422, "请指定运行（run_id）和其中的一条查询（query）", case_rules.CASE_INVALID, + field="query") + run = await _readable_run(session, run_id) + picked = next((q for q in await _run_queries(session, run_id) if q["artifact"] == query), None) + if picked is None or not picked.get("sql"): + raise CodedHTTPException(404, "这次运行里没有这条查询", RUN_NOT_FOUND) + source_id = picked.get("source_id") + if not source_id and picked.get("source"): + source_id = await session.scalar(select(DataSource.id).where(DataSource.name == picked["source"])) + if not source_id or await session.get(DataSource, source_id) is None: + raise CodedHTTPException(422, f"这条查询查的数据源「{picked.get('source') or '?'}」已不在，没法按它重算", + case_rules.CASE_INVALID, field="query") + question = body.get("question") + if question in (None, "") and isinstance((run.input or {}).get("question"), str): + question = run.input["question"] + gold = {"kind": "from_run", "datasource_id": source_id, "sql": picked["sql"], "run_id": run_id, "query": query, + "node_id": picked.get("node_id"), "confirmed_by": actor, "confirmed_at": utcnow().isoformat()} + case_body = {k: body[k] for k in ("expectations", "tags", "datasource_ids", "enabled") if k in body} + fields = await _validated(session, suite, {**case_body, "question": question, "gold": gold}, trusted_gold=True) + outcome = await _trial(suite, fields, actor) + _gate(outcome) + row = await _insert_case(session, suite, fields, actor, _trial_record(outcome, actor)) + return {"case": _case_out(row, await _source_names(session, [row])), "trial": outcome.to_dict()} + + +# ========================================================================== +# 口径 +# ========================================================================== + + +@router.get("/calibers") +async def list_calibers(session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """能选的口径：每个工作流最新的已发布未退役版本里的源头口径卡的每个指标（和蓝图助手同一份，look/calibers.CaliberKind）。 + 不能复用的照列，写明原因（选了也存不了）。""" + from app.look.calibers import CaliberKind + + kind = await CaliberKind.load(session) + return {"calibers": [{"workflow_id": i.workflow_id, "workflow_name": i.workflow_name, "version": i.version, + "level": i.level, "node_id": i.node_id, "card": i.card, "caliber": i.caliber, + "metric_id": i.metric_id, "name": i.name, "unit": i.unit, "reusable": i.reason is None, + "reason": i.reason} for i in kind.items]} + + +@router.get("/suites/{suite_id}/cases/{case_id}/upgrade") +async def case_upgrade(suite_id: str, case_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + row = await _case(session, suite_id, case_id) + return {"upgrade": await _upgrade_of(session, row), "choices": gold_rules.UPGRADE_LABEL} + + +@router.post("/suites/{suite_id}/cases/{case_id}/upgrade") +async def dispose_upgrade(suite_id: str, case_id: str, request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """处置口径升版：{policy: recompute|dual|incomparable, version}，没有缺省。 + + - recompute：改钉新版本（题目指纹变，和上一次比时标不可比）； + - dual：新旧两版各算一份期望（旧的那一份照旧可比）； + - incomparable：旧题归档（不再评），按新版本另建一道新题、重开基线。 + 处置后的题同样要试算通过才生效。""" + suite = await _suite(session, suite_id) + row = await _case(session, suite_id, case_id) + body = await _body(request) + version = _version(body, row.version, "题目") + actor = actor_of(x_actor) + policy = body.get("policy") + current = case_fields(row) + try: + planned, _latest = await gold_rules.plan_dispose(session, current, policy if isinstance(policy, str) else "") + except gold_rules.DisposeRefused as e: + raise CodedHTTPException(e.status, e.message, e.code, field="policy") from e + if isinstance(planned.get("upgrade"), dict): + planned["upgrade"]["by"] = actor + fields = {**current, "gold": planned} + for key in ("id", "suite_id", "seq"): + fields.pop(key, None) + outcome = await _trial(suite, fields, actor, case_id=row.id if policy != "incomparable" else None, + seq=row.seq if policy != "incomparable" else None) + _gate(outcome) + if policy == "incomparable": + new = await _insert_case(session, suite, fields, actor, _trial_record(outcome, actor)) + result = await session.execute(update(EvalCase).where(EvalCase.id == case_id, EvalCase.version == version) + .values(archived=True, enabled=False, superseded_by=new.id, + version=version + 1, updated_at=utcnow()) + .execution_options(synchronize_session=False)) + if result.rowcount != 1: + await session.delete(new) + await session.commit() + raise CodedHTTPException(409, "这道题刚被别人修改过，请刷新后再处置", VERSION_CONFLICT) + await session.commit() + await session.refresh(row) + names = await _source_names(session, [row, new]) + return {"case": _case_out(row, names), "new_case": _case_out(new, names, await _upgrade_of(session, new)), + "trial": outcome.to_dict()} + result = await session.execute(update(EvalCase).where(EvalCase.id == case_id, EvalCase.version == version) + .values(gold=planned, case_hash=case_hash(fields), trial=_trial_record(outcome, actor), + version=version + 1, updated_at=utcnow()) + .execution_options(synchronize_session=False)) + if result.rowcount != 1: + await session.rollback() + raise CodedHTTPException(409, "这道题刚被别人修改过，请刷新后再处置", VERSION_CONFLICT) + await session.commit() + await session.refresh(row) + return {"case": _case_out(row, await _source_names(session, [row]), await _upgrade_of(session, row)), + "new_case": None, "trial": outcome.to_dict()} + + +# ========================================================================== +# 导出导入 +# ========================================================================== + + +@router.get("/suites/{suite_id}/export") +async def export_suite(suite_id: str, session: AsyncSession = Depends(get_session)) -> Response: + """题集导出成 JSON 文件（按数据源名、工作流名对上；不含期望值、答案、运行）。""" + import json + from urllib.parse import quote + + suite = await _suite(session, suite_id) + rows = list((await session.execute(select(EvalCase).where(EvalCase.suite_id == suite_id))).scalars()) + content = await transfer.export_suite(session, suite, rows) + filename = transfer.attachment_name(suite.name) + return Response(content=json.dumps(content, ensure_ascii=False, indent=2), media_type="application/json", + headers={"Content-Disposition": f"attachment; filename=\"eval-suite-{suite.id}.json\"; " + f"filename*=UTF-8''{quote(filename)}"}) + + +async def _plan(session: AsyncSession, request: Request) -> tuple[transfer.ImportPlan, dict[str, Any]]: + body = await _body(request) + try: + return await transfer.plan_import(session, body.get("content")), body + except transfer.TransferInvalid as e: + raise CodedHTTPException(422, str(e), IMPORT_INVALID, field="content") from e + + +@router.post("/suites/import/preview") +async def import_preview(request: Request, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """{content: 题集文件}：哪些题对得上、哪些跳过和原因。不写库、不查业务库（试算在导入时做）。""" + plan, _ = await _plan(session, request) + return plan.to_dict() + + +@router.post("/suites/import", status_code=201) +async def import_suite(request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """{content: 题集文件, name?}：建一个新题集，对得上的题逐道试算（和保存题目同一道关），过了的导进来，没过的跳过、写原因。 + 一道都导不进来就不建题集（422，带每道题的原因）。""" + plan, body = await _plan(session, request) + actor = actor_of(x_actor) + if isinstance(body.get("name"), str) and body["name"].strip(): + plan.suite["name"] = body["name"].strip()[:200] + # 先定好 id：试算的运行记着是哪个题集的（一道都导不进来时题集不建，那几次试算的运行照样留着、点得开） + suite = EvalSuite(id=new_id(), **plan.suite, created_by=actor) + ready = [c for c in plan.cases if c.fields is not None] + gate = asyncio.Semaphore(IMPORT_TRIALS) + + async def trial(item: transfer.PlannedCase) -> gold_rules.GoldOutcome: + async with gate: + return await gold_rules.compute_gold({**item.fields, "id": None, "seq": item.seq}, + _context(suite, actor, imported=True)) + + passed: list[tuple[transfer.PlannedCase, gold_rules.GoldOutcome]] = [] + for item, outcome in zip(ready, await asyncio.gather(*(trial(c) for c in ready))): + if outcome.ok: + passed.append((item, outcome)) + else: + item.fields, item.reason, item.code = None, f"试算没有通过：{outcome.reason}", outcome.code + if not passed: + raise CodedHTTPException(422, "题集里没有一道题能导进来：" + "；".join( + f"第 {c.seq or '?'} 题：{c.reason}" for c in plan.cases[:10]), IMPORT_INVALID, field="content") + session.add(suite) + await session.flush() + passed.sort(key=lambda pair: (pair[0].seq is None, pair[0].seq or 0)) + for seq, (item, outcome) in enumerate(passed, start=1): + assert item.fields is not None + session.add(EvalCase(suite_id=suite.id, seq=seq, case_hash=case_hash(item.fields), + trial=_trial_record(outcome, actor), created_by=actor, **item.fields)) + await session.commit() + detail = await get_suite(suite.id, session) + return {"suite": detail, "imported": len(passed), "skipped": [c.to_dict() for c in plan.cases if c.fields is None]} + + +# ========================================================================== +# 目录快照 +# ========================================================================== + + +def _snapshot_error(e: snapshots.SnapshotInvalid) -> CodedHTTPException: + return CodedHTTPException(e.status, e.message, e.code, field=e.field) + + +@router.get("/catalog-snapshots") +async def list_snapshots(session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + rows = list((await session.execute(select(EvalCatalogSnapshot).order_by(EvalCatalogSnapshot.created_at.desc()) + .limit(200))).scalars()) + await snapshots.reconcile_stale(session, rows) + return {"snapshots": [snapshots.snapshot_out(r) for r in rows]} + + +@router.post("/catalog-snapshots", status_code=201) +async def take_snapshot(request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """{source_ids}：拍这几个数据源此刻的目录（当场写好、不可变）。""" + body = await _body(request) + try: + row = await snapshots.take_live(session, body.get("source_ids") or [], actor=actor_of(x_actor)) + except snapshots.SnapshotInvalid as e: + raise _snapshot_error(e) from e + return snapshots.snapshot_out(row) + + +async def _shadow_plan(session: AsyncSession, body: dict[str, Any]) -> snapshots.ShadowPlan: + try: + _model, model_id = await snapshots.resolve_model(session, body.get("provider"), body.get("model")) + return await snapshots.plan_shadow(session, body, model_id=model_id) + except snapshots.SnapshotInvalid as e: + raise _snapshot_error(e) from e + + +@router.post("/catalog-snapshots/shadow/estimate") +async def shadow_estimate(request: Request, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """影子起草的计划和估算（起草哪些表、调几次模型、多少 token、多少钱）：开始之前给人看。不调模型。""" + return (await _shadow_plan(session, await _body(request))).to_dict() + + +@router.post("/catalog-snapshots/shadow", status_code=202) +async def start_shadow(request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """开始影子起草（后台）：{sources: [{id, tables?}], suite_id?, provider?, model?, guidance?, note?, strip_llm?}。 + 交回登记的那一行（queued）；进度、用量、结果按 GET /catalog-snapshots/{id} 读。""" + plan = await _shadow_plan(session, await _body(request)) + row = await snapshots.start_shadow(session, plan, actor=actor_of(x_actor)) + return snapshots.snapshot_out(row) + + +async def _snapshot(session: AsyncSession, snapshot_id: str) -> EvalCatalogSnapshot: + row = await session.get(EvalCatalogSnapshot, snapshot_id) + if row is None: + raise CodedHTTPException(404, "目录快照不存在", SNAPSHOT_NOT_FOUND) + return row + + +@router.get("/catalog-snapshots/{snapshot_id}") +async def get_snapshot(snapshot_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + row = await _snapshot(session, snapshot_id) + await snapshots.reconcile_stale(session, [row]) + return snapshots.snapshot_out(row) + + +@router.post("/catalog-snapshots/{snapshot_id}/stop") +async def stop_snapshot(snapshot_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """停止一次影子起草：不出快照。已经结束的回 409。""" + await _snapshot(session, snapshot_id) + if not await snapshots.stop(snapshot_id): + raise CodedHTTPException(409, "这次影子起草已经结束了", snapshots.SNAPSHOT_FINISHED) + session.expire_all() + return snapshots.snapshot_out(await _snapshot(session, snapshot_id)) + + +__all__ = ["router"] diff --git a/backend/app/db/models_eval.py b/backend/app/db/models_eval.py index 14924e2a..9d8e4f63 100644 --- a/backend/app/db/models_eval.py +++ b/backend/app/db/models_eval.py @@ -82,7 +82,7 @@ class EvalCase(Base, TimestampMixin): id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) suite_id: Mapped[str] = mapped_column(ForeignKey("eval_suites.id", ondelete="CASCADE")) - # 题号：题集里从 1 起，删了的题号不复用（运行名「评测：题集名·第 n 题」按它写） + # 题号：题集里从 1 起、按建的先后（运行名「评测：题集名·第 n 题」按它写）。和上一次比按题目 id 和指纹配对，不按题号 seq: Mapped[int] = mapped_column(Integer) kind: Mapped[str] = mapped_column(String(20), default="ask") # 问法。可以带 {{ schedule.* }} 变量（按评测的截至时间渲染成明确的日期） diff --git a/backend/app/eval/transfer.py b/backend/app/eval/transfer.py new file mode 100644 index 00000000..16942225 --- /dev/null +++ b/backend/app/eval/transfer.py @@ -0,0 +1,255 @@ +"""题集的导出导入（细案 01 做法 19）：本机整理好的题集带到部署实例上；仓库里的「零售样例题集」也按这个格式写 +（tests/fixtures/eval/retail_suite.json）。 + +**为什么要有它。** 真实业务的题集只在部署实例上建、上跑（评测-D15：真实数据不出部署实例、不进公开仓库）；样例题集、 +本机写好的题目要能带过去。硬编码在测试里的题集部署实例上用不了。 + +**文件**（export_suite）：{format: "agentlab-eval-suite", version: 1, exported_at, suite: {name, description, timezone, +budget}, cases: [{seq, kind, question, datasources: [数据源名], workflow: {name, version}, input_template, expectations, +gold, tags, enabled}]}。期望来源里的数据源、工作流写名字（id 只在这台机器上有效）：sql 题 {kind, datasource, sql}， +口径题 {kind, workflow, version, node_id, inputs}，从运行认定 {kind, datasource, sql, confirmed_by, confirmed_at}。 +**不含期望值、答案、运行**：期望每次评测都重算，不冻结数值；从运行认定的那次运行、升版的处置记录只在这台机器上有意义， +不导出。默认变体也不导出（里面是这台机器上的模型、目录快照）。归档的题不导出。 + +**导入**（plan_import 预览和导入共用这一份规矩）： +- 数据源、工作流按名字对上（一字不差）；对不上的题跳过、写原因。 +- 每道题过和新建题目同一套校验（cases.normalize_case + resolve_case），导入时再试算一次（和保存题目时同一道关：只读守卫、 + 遮罩拒收、取得到值），没过的跳过、写原因（接口层做，见 api/evals.import_suite）。一道都导不进来就不建题集。 +- 文件格式不对（不是对象、格式标记或版本不对）抛 TransferInvalid，接口回 422。 +""" +from __future__ import annotations + +import copy +import re +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.eval.cases import CaseInvalid, normalize_case, resolve_case +from app.eval.suites import normalize_suite + +FORMAT = "agentlab-eval-suite" +VERSION = 1 +FILE_SUFFIX = "-题集.json" +#: 一个文件最多几道题 +MAX_CASES = 500 +_BAD_FILENAME = re.compile(r'[\\/:*?"<>|\x00-\x1f\x7f]') + +#: 跳过的原因 +SOURCE_UNMATCHED = "数据源「{name}」在这里没有（按名字对，一字不差）" +WORKFLOW_UNMATCHED = "工作流「{name}」在这里没有（按名字对，一字不差）" + + +class TransferInvalid(ValueError): + """文件格式不对。message 给人看（接口回 422）。""" + + +def attachment_name(suite_name: str) -> str: + name = _BAD_FILENAME.sub("_", suite_name or "").strip(" .")[:80].strip(" .") + return f"{name or '题集'}{FILE_SUFFIX}" + + +# ========================================================================== +# 导出 +# ========================================================================== + + +async def _names(session: AsyncSession, model: Any, ids: set[str]) -> dict[str, str]: + if not ids: + return {} + rows = await session.execute(select(model.id, model.name).where(model.id.in_(sorted(ids)))) + return {i: n for i, n in rows.all()} + + +def _gold_out(gold: Mapping[str, Any] | None, sources: Mapping[str, str], workflows: Mapping[str, str] + ) -> dict[str, Any] | None: + if not gold: + return None + kind = gold.get("kind") + if kind == "caliber": + return {"kind": "caliber", "workflow": workflows.get(gold.get("workflow_id"), gold.get("workflow_id")), + "version": gold.get("version"), "node_id": gold.get("node_id"), + "inputs": copy.deepcopy(gold.get("inputs") or {})} + out = {"kind": kind, "datasource": sources.get(gold.get("datasource_id"), gold.get("datasource_id")), + "sql": gold.get("sql")} + if kind == "from_run": + out |= {k: gold[k] for k in ("confirmed_by", "confirmed_at") if gold.get(k)} + return out + + +async def export_suite(session: AsyncSession, suite: Any, cases: list[Any]) -> dict[str, Any]: + from app.db.base import utcnow + from app.db.models import DataSource, Workflow + + kept = [c for c in cases if not c.archived] + source_ids = {sid for c in kept for sid in c.datasource_ids or []} | { + (c.gold or {}).get("datasource_id") for c in kept if (c.gold or {}).get("datasource_id")} + workflow_ids = {c.workflow_id for c in kept if c.workflow_id} | { + (c.gold or {}).get("workflow_id") for c in kept if (c.gold or {}).get("workflow_id")} + sources = await _names(session, DataSource, {s for s in source_ids if s}) + workflows = await _names(session, Workflow, {w for w in workflow_ids if w}) + out_cases = [] + for c in sorted(kept, key=lambda c: c.seq): + item: dict[str, Any] = {"seq": c.seq, "kind": c.kind, "question": c.question, + "datasources": [sources.get(s, s) for s in c.datasource_ids or []], + "expectations": copy.deepcopy(list(c.expectations or [])), + "gold": _gold_out(c.gold, sources, workflows), "tags": list(c.tags or []), + "enabled": bool(c.enabled)} + if c.kind == "workflow": + item["workflow"] = {"name": workflows.get(c.workflow_id, c.workflow_id), "version": c.workflow_version} + item["input_template"] = dict(c.input_template or {}) + out_cases.append(item) + return {"format": FORMAT, "version": VERSION, "exported_at": utcnow().isoformat(), + "suite": {"name": suite.name, "description": suite.description or "", "timezone": suite.timezone, + "budget": dict(suite.budget or {})}, + "cases": out_cases} + + +# ========================================================================== +# 导入 +# ========================================================================== + + +@dataclass +class PlannedCase: + """文件里的一道题：对上了的（fields 是 normalize_case 的结果，可以直接建）或者跳过的（reason）。""" + + seq: int | None + question: str + fields: dict[str, Any] | None = None + reason: str | None = None + code: str | None = None + + def to_dict(self) -> dict[str, Any]: + return {"seq": self.seq, "question": self.question, "ok": self.fields is not None, "reason": self.reason, + "code": self.code} + + +@dataclass +class ImportPlan: + suite: dict[str, Any] + cases: list[PlannedCase] = field(default_factory=list) + #: 文件里提到的数据源、工作流名 → 这里的 id（对不上为 None） + sources: dict[str, str | None] = field(default_factory=dict) + workflows: dict[str, str | None] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return {"suite": dict(self.suite), "cases": [c.to_dict() for c in self.cases], + "sources": dict(self.sources), "workflows": dict(self.workflows), + "importable": sum(1 for c in self.cases if c.fields is not None), + "skipped": sum(1 for c in self.cases if c.fields is None)} + + +def read_file(content: Any) -> tuple[dict[str, Any], list[Any]]: + if not isinstance(content, Mapping): + raise TransferInvalid("题集文件需要是一个 JSON 对象") + if content.get("format") != FORMAT: + raise TransferInvalid(f"不是题集文件（format 应为 {FORMAT}）") + if content.get("version") != VERSION: + raise TransferInvalid(f"题集文件的版本是 {content.get('version')}，这里只认 {VERSION}") + suite = content.get("suite") + cases = content.get("cases") + if not isinstance(suite, Mapping) or not isinstance(cases, list): + raise TransferInvalid("题集文件里需要有 suite（对象）和 cases（列表）") + if len(cases) > MAX_CASES: + raise TransferInvalid(f"一个题集文件最多 {MAX_CASES} 道题") + return dict(suite), cases + + +async def _lookup(session: AsyncSession, model: Any, names: set[str]) -> dict[str, str | None]: + found = dict((await session.execute(select(model.name, model.id).where(model.name.in_(sorted(names))))).all()) \ + if names else {} + return {n: found.get(n) for n in sorted(names)} + + +async def plan_import(session: AsyncSession, content: Any) -> ImportPlan: + """按这台机器上的数据源、工作流对上文件里的每道题，逐题过 normalize_case + resolve_case。不写库、不查业务库。""" + from app.db.models import DataSource, Workflow + + raw_suite, raw_cases = read_file(content) + try: + suite = normalize_suite({k: raw_suite.get(k) for k in ("name", "description", "timezone", "budget") + if raw_suite.get(k) is not None}) + except CaseInvalid as e: + raise TransferInvalid(f"题集：{e.message}") from e + source_names = set() + workflow_names = set() + for c in raw_cases: + if not isinstance(c, Mapping): + continue + source_names |= {s for s in c.get("datasources") or [] if isinstance(s, str)} + gold = c.get("gold") if isinstance(c.get("gold"), Mapping) else {} + if isinstance(gold.get("datasource"), str): + source_names.add(gold["datasource"]) + if isinstance(gold.get("workflow"), str): + workflow_names.add(gold["workflow"]) + if isinstance(c.get("workflow"), Mapping) and isinstance(c["workflow"].get("name"), str): + workflow_names.add(c["workflow"]["name"]) + plan = ImportPlan(suite=suite, sources=await _lookup(session, DataSource, source_names), + workflows=await _lookup(session, Workflow, workflow_names)) + for i, c in enumerate(raw_cases): + plan.cases.append(await _plan_case(session, c, i, plan, suite["timezone"])) + return plan + + +async def _plan_case(session: AsyncSession, raw: Any, i: int, plan: ImportPlan, tz: str) -> PlannedCase: + if not isinstance(raw, Mapping): + return PlannedCase(seq=None, question="", reason=f"第 {i + 1} 道题不是一个对象") + seq = raw.get("seq") if isinstance(raw.get("seq"), int) and not isinstance(raw.get("seq"), bool) else None + question = str(raw.get("question") or "") + planned = PlannedCase(seq=seq, question=question) + + def source_id(name: Any) -> str | None: + return plan.sources.get(name) if isinstance(name, str) else None + + def workflow_id(name: Any) -> str | None: + return plan.workflows.get(name) if isinstance(name, str) else None + + body: dict[str, Any] = {k: copy.deepcopy(raw[k]) for k in ("kind", "question", "expectations", "tags", "enabled", + "input_template") if k in raw} + ids = [] + for name in raw.get("datasources") or []: + if (sid := source_id(name)) is None: + planned.reason = SOURCE_UNMATCHED.format(name=name) + return planned + ids.append(sid) + body["datasource_ids"] = ids + if isinstance(raw.get("workflow"), Mapping): + if (wid := workflow_id(raw["workflow"].get("name"))) is None: + planned.reason = WORKFLOW_UNMATCHED.format(name=raw["workflow"].get("name")) + return planned + body["workflow_id"], body["workflow_version"] = wid, raw["workflow"].get("version") + gold = raw.get("gold") + trusted = False + if isinstance(gold, Mapping): + gold = dict(gold) + if "datasource" in gold: + name = gold.pop("datasource") + if (sid := source_id(name)) is None: + planned.reason = SOURCE_UNMATCHED.format(name=name) + return planned + gold["datasource_id"] = sid + if "workflow" in gold: + name = gold.pop("workflow") + if (wid := workflow_id(name)) is None: + planned.reason = WORKFLOW_UNMATCHED.format(name=name) + return planned + gold["workflow_id"] = wid + # 从运行认定的题带着认定人和时间过来（运行本身不导出）；它只能由服务端建，这里就是服务端 + trusted = gold.get("kind") == "from_run" + body["gold"] = gold + try: + fields = normalize_case(body, trusted_gold=trusted) + await resolve_case(session, fields, tz=tz) + except CaseInvalid as e: + planned.reason, planned.code = e.message, e.code + return planned + planned.fields = fields + return planned + + +__all__ = ["FORMAT", "FILE_SUFFIX", "ImportPlan", "MAX_CASES", "PlannedCase", "TransferInvalid", "VERSION", + "attachment_name", "export_suite", "plan_import", "read_file"] diff --git a/backend/tests/fixtures/acceptance/retail.py b/backend/tests/fixtures/acceptance/retail.py index 47124870..b1ca0f63 100644 --- a/backend/tests/fixtures/acceptance/retail.py +++ b/backend/tests/fixtures/acceptance/retail.py @@ -13,6 +13,9 @@ customers.phone 是手机号，验收时把它设成遮罩列，检验质询、证据面板、可验证报告不露原值。append_day 给定时任务 验收用：两次触发之间往「昨天」追加几单，制造「只数据变」。 + +评测的零售样例题集（tests/fixtures/eval/retail_suite.json）按数据源名 SOURCE_NAME 对上这个库：生成之后登记成这个名字的 +只读数据源，再导入题集（POST /api/evals/suites/import），题目的期望都按这个库算。 """ from __future__ import annotations @@ -24,6 +27,8 @@ from pathlib import Path DAYS = 60 SEED = 20261006 +#: 零售样例题集里写的数据源名 +SOURCE_NAME = "retail_demo" #: (门店 id, 店名, 地区, 城市)：华南门店多、客单价高，华东次之 STORES = [ diff --git a/backend/tests/fixtures/eval/retail_suite.json b/backend/tests/fixtures/eval/retail_suite.json new file mode 100644 index 00000000..cf38b1f4 --- /dev/null +++ b/backend/tests/fixtures/eval/retail_suite.json @@ -0,0 +1,172 @@ +{ + "format": "agentlab-eval-suite", + "version": 1, + "exported_at": "2026-10-16T00:00:00+00:00", + "suite": { + "name": "零售样例题集", + "description": "配 tests/fixtures/acceptance/retail.py 的虚构零售库（数据源名 retail_demo）：支付日归属、退款、取消、华南高于华东几个口径陷阱，一道期望答「没有数据」的题，一道问法含糊、期望反问的题。用于演示、验收和真实模型回归；店名、金额都是虚构的", + "timezone": "Asia/Shanghai", + "budget": {} + }, + "cases": [ + { + "seq": 1, + "kind": "ask", + "question": "上周（{{ schedule.last_week_start }} 至 {{ schedule.last_week_end }}）的销售额是多少？", + "datasources": [ + "retail_demo" + ], + "tags": [ + "口径陷阱", + "支付日归属", + "相对日期" + ], + "gold": { + "kind": "sql", + "datasource": "retail_demo", + "sql": "SELECT ROUND(SUM(amount), 2) AS 销售额 FROM orders WHERE status = 'paid' AND paid_date BETWEEN '{{ schedule.last_week_start }}' AND '{{ schedule.last_week_end }}'" + }, + "expectations": [ + { + "id": "sales", + "kind": "scalar", + "column": "销售额", + "label": "上周销售额（按支付日归属、不含退款和取消）" + } + ], + "enabled": true + }, + { + "seq": 2, + "kind": "ask", + "question": "上周（{{ schedule.last_week_start }} 至 {{ schedule.last_week_end }}）有多少单退款？退款金额是多少？", + "datasources": [ + "retail_demo" + ], + "tags": [ + "口径陷阱", + "退款" + ], + "gold": { + "kind": "sql", + "datasource": "retail_demo", + "sql": "SELECT COUNT(*) AS 退款单数, ROUND(SUM(amount), 2) AS 退款金额 FROM orders WHERE status = 'refunded' AND paid_date BETWEEN '{{ schedule.last_week_start }}' AND '{{ schedule.last_week_end }}'" + }, + "expectations": [ + { + "id": "refund_orders", + "kind": "scalar", + "column": "退款单数", + "label": "上周退款单数" + }, + { + "id": "refund_amount", + "kind": "scalar", + "column": "退款金额", + "label": "上周退款金额" + } + ], + "enabled": true + }, + { + "seq": 3, + "kind": "ask", + "question": "上周（{{ schedule.last_week_start }} 至 {{ schedule.last_week_end }}）下单后取消的订单有多少单？", + "datasources": [ + "retail_demo" + ], + "tags": [ + "口径陷阱", + "取消" + ], + "gold": { + "kind": "sql", + "datasource": "retail_demo", + "sql": "SELECT COUNT(*) AS 取消单数 FROM orders WHERE status = 'cancelled' AND order_date BETWEEN '{{ schedule.last_week_start }}' AND '{{ schedule.last_week_end }}'" + }, + "expectations": [ + { + "id": "cancelled", + "kind": "scalar", + "column": "取消单数", + "label": "上周取消的订单数（按下单日）" + } + ], + "enabled": true + }, + { + "seq": 4, + "kind": "ask", + "question": "上周（{{ schedule.last_week_start }} 至 {{ schedule.last_week_end }}）华南和华东的销售额各是多少？哪个地区更高？", + "datasources": [ + "retail_demo" + ], + "tags": [ + "口径陷阱", + "地区对比" + ], + "gold": { + "kind": "sql", + "datasource": "retail_demo", + "sql": "SELECT s.region AS 地区, ROUND(SUM(o.amount), 2) AS 销售额 FROM orders o JOIN stores s ON s.id = o.store_id WHERE o.status = 'paid' AND o.paid_date BETWEEN '{{ schedule.last_week_start }}' AND '{{ schedule.last_week_end }}' AND s.region IN ('华南', '华东') GROUP BY s.region ORDER BY 销售额 DESC" + }, + "expectations": [ + { + "id": "by_region", + "kind": "table", + "keys": [ + "地区" + ], + "ordered": true, + "label": "华南、华东上周的销售额，按销售额从高到低" + } + ], + "enabled": true + }, + { + "seq": 5, + "kind": "ask", + "question": "上周（{{ schedule.last_week_start }} 至 {{ schedule.last_week_end }}）华中区的销售额是多少？", + "datasources": [ + "retail_demo" + ], + "tags": [ + "拒答", + "没有数据" + ], + "gold": { + "kind": "sql", + "datasource": "retail_demo", + "sql": "SELECT s.region AS 地区, ROUND(SUM(o.amount), 2) AS 销售额 FROM orders o JOIN stores s ON s.id = o.store_id WHERE s.region = '华中' AND o.status = 'paid' AND o.paid_date BETWEEN '{{ schedule.last_week_start }}' AND '{{ schedule.last_week_end }}' GROUP BY s.region" + }, + "expectations": [ + { + "id": "no_data", + "kind": "none", + "label": "没有华中区的门店，应答「没有数据」，不给数" + } + ], + "enabled": true + }, + { + "seq": 6, + "kind": "ask", + "question": "最近卖得怎么样？", + "datasources": [ + "retail_demo" + ], + "tags": [ + "问法含糊" + ], + "gold": null, + "expectations": [ + { + "id": "clarify", + "kind": "ask", + "label": "问法含糊（哪段时间、哪个指标都没说），应先反问" + } + ], + "enabled": true + } + ] +} diff --git a/backend/tests/test_eval_cases.py b/backend/tests/test_eval_cases.py new file mode 100644 index 00000000..0fd7fa9e --- /dev/null +++ b/backend/tests/test_eval_cases.py @@ -0,0 +1,130 @@ +"""题目的形状与校验（app/eval/cases.py、suites.py）：数值期望不许人填（评测-D4）、文字要点不计入准确率、只做单轮 +（评测-D12）、变量只认定时任务那份、期望和期望来源配得上、题目指纹只看内容。纯函数，不碰库。""" +from __future__ import annotations + +from datetime import datetime +from zoneinfo import ZoneInfo + +import pytest + +from app.eval import cases, suites +from app.eval.cases import CaseInvalid, normalize_case + +SQL = {"kind": "sql", "datasource_id": "ds1", "sql": "SELECT SUM(amount) AS gmv FROM orders"} + + +def invalid(body: dict, **kw) -> CaseInvalid: + with pytest.raises(CaseInvalid) as caught: + normalize_case(body, **kw) + return caught.value + + +@pytest.mark.parametrize("key", cases.HUMAN_VALUE_KEYS) +def test_numbers_cannot_be_filled_in_by_hand(key): + e = invalid({"question": "q", "gold": SQL, "expectations": [{"kind": "scalar", "column": "gmv", key: 330}]}) + assert "数值期望不许人填" in e.message and e.field == f"expectations.0.{key}" + + +def test_text_points_never_count_toward_accuracy(): + fields = normalize_case({"question": "为什么涨了？", "expectations": [ + {"kind": "text", "points": [" 华南门店多 ", "客单价高"], "required": True}]}) + [exp] = fields["expectations"] + assert exp == {"id": "e1", "kind": "text", "required": False, "points": ["华南门店多", "客单价高"]} + assert fields["gold"] is None + + +def test_only_single_turn_cases_this_round(): + assert invalid({"question": "q", "history": [{"q": "上一轮"}], "expectations": [{"kind": "ask"}]}).field == "history" + assert normalize_case({"question": "q", "history": [], "expectations": [{"kind": "ask"}]})["history"] is None + + +@pytest.mark.parametrize("body, field", [ + ({"question": "q", "expectations": [{"kind": "scalar", "column": "gmv"}]}, "gold"), + ({"question": "q", "gold": SQL, "expectations": [{"kind": "ask"}]}, "gold"), + ({"question": "q", "gold": SQL, "expectations": [{"kind": "ask"}, {"kind": "none"}]}, "expectations"), + ({"question": "q", "gold": SQL, "expectations": [{"kind": "scalar", "metric": "gmv"}]}, "expectations.0.metric"), + ({"question": "q", "gold": SQL, "expectations": [{"kind": "scalar"}]}, "expectations.0.column"), + ({"question": "q", "gold": SQL, "expectations": [{"kind": "scalar", "column": "g", "row": 1, "where": {"a": 1}}]}, + "expectations.0.row"), + ({"question": "q", "gold": SQL, "expectations": [{"kind": "table"}]}, "expectations.0.keys"), + ({"question": "q", "gold": SQL, "expectations": [{"kind": "scalar", "column": "g", "colum": "x"}]}, + "expectations.0.colum"), + ({"question": "q", "gold": SQL, "expectations": [{"kind": "scalar", "column": "g", "tolerance": 2}]}, + "expectations.0.tolerance"), + ({"question": "q", "gold": SQL, "expectations": [{"id": "a", "kind": "none"}, {"id": "a", "kind": "none"}]}, + "expectations.1.id"), + ({"question": "", "expectations": [{"kind": "ask"}]}, "question"), + ({"question": "q", "expectations": []}, "expectations"), + ({"question": "q", "workflow_id": "w1", "expectations": [{"kind": "ask"}]}, "workflow_id"), + ({"kind": "workflow", "workflow_id": "w1", "workflow_version": 1, "expectations": [{"kind": "ask"}]}, + "expectations"), + ({"question": "上周 {{ schedule.lastweek }}", "expectations": [{"kind": "ask"}]}, "question"), + ({"question": "q", "gold": {**SQL, "sql": "SELECT {{ schedule.today | shout }}"}, + "expectations": [{"kind": "none"}]}, "gold.sql"), + ({"question": "q", "gold": {**SQL, "sql": "SELECT {{ schedule.today "}, "expectations": [{"kind": "none"}]}, + "gold.sql"), + ({"question": "q", "gold": {"kind": "caliber", "workflow_id": "w", "version": 1, "node_id": "c", + "inputs": {"start": "{{ vars.x }}"}}, "expectations": [{"kind": "scalar", + "metric": "m"}]}, + "gold.inputs"), + ({"question": "q", "gold": {"kind": "caliber", "workflow_id": "w", "version": 1, "node_id": "c"}, + "expectations": [{"kind": "table", "keys": ["a"]}]}, "expectations.0.kind"), + ({"question": "q", "gold": {"kind": "magic"}, "expectations": [{"kind": "none"}]}, "gold.kind"), +]) +def test_expectations_and_gold_must_fit_together(body, field): + assert invalid(body).field == field + + +def test_from_run_gold_is_only_built_by_the_server(): + gold = {"kind": "from_run", "datasource_id": "ds1", "sql": "SELECT 1 AS n", "run_id": "r1", "query": "a1", + "confirmed_by": "甲", "confirmed_at": "2026-10-01T00:00:00+00:00"} + assert invalid({"question": "q", "gold": gold, "expectations": [{"kind": "scalar", "column": "n"}]}).field \ + == "gold.kind" + fields = normalize_case({"question": "q", "gold": gold, "expectations": [{"kind": "scalar", "column": "n"}]}, + trusted_gold=True) + # 原样带回来的可以过（改标签时整份 gold 跟着回来） + again = normalize_case({"gold": gold, "tags": ["x"]}, current=fields) + assert again["gold"] == fields["gold"] and again["tags"] == ["x"] + + +def test_the_case_hash_follows_content_not_records(): + base = normalize_case({"question": "q", "gold": SQL, "expectations": [{"kind": "scalar", "column": "gmv"}], + "tags": ["a"]}) + assert cases.case_hash({**base, "tags": ["b"], "enabled": False}) == cases.case_hash(base) + assert cases.case_hash({**base, "question": "q2"}) != cases.case_hash(base) + caliber = {"kind": "caliber", "workflow_id": "w", "version": 1, "node_id": "c", "inputs": {}} + pinned = {**base, "gold": caliber} + # 升版处置的记录不进指纹（dual 时旧的那一份照旧可比），改钉版本才进 + assert cases.case_hash({**pinned, "gold": {**caliber, "upgrade": {"policy": "dual", "to_version": 2}}}) == \ + cases.case_hash(pinned) + assert cases.case_hash({**pinned, "gold": {**caliber, "version": 2}}) != cases.case_hash(pinned) + + +def test_the_disposal_record_is_kept_only_by_the_server(): + caliber = {"kind": "caliber", "workflow_id": "w", "version": 1, "node_id": "c", "inputs": {}} + stored = {**caliber, "upgrade": {"policy": "dual", "from_version": 1, "to_version": 2}} + body = {"question": "q", "expectations": [{"kind": "scalar", "metric": "m"}]} + kept = normalize_case({**body, "gold": {**caliber, "upgrade": {"policy": "recompute", "to_version": 9}}}, + current={"gold": stored}) + assert kept["gold"]["upgrade"] == stored["upgrade"], "客户端带来的不算，照库里的留着" + moved = normalize_case({**body, "gold": {**caliber, "node_id": "other"}}, current={"gold": stored}) + assert "upgrade" not in moved["gold"], "换了卡就没有处置" + fresh = normalize_case({**body, "gold": {**caliber, "upgrade": {"policy": "dual"}}}) + assert "upgrade" not in fresh["gold"] + + +def test_variables_render_at_the_as_of_moment(): + as_of = datetime(2027, 1, 1, 8, 0, tzinfo=ZoneInfo("Asia/Shanghai")) + values = cases.variable_values(as_of, "Asia/Shanghai") + assert values["last_week"] == "2026-W52" and values["yesterday"] == "2026-12-31" + assert cases.render("'{{ schedule.last_week_start }}'..'{{ schedule.last_week_end }}'", values) == \ + "'2026-12-21'..'2026-12-27'" + + +def test_suite_budget_defaults_and_ranges(): + assert suites.effective_budget({}) == {"max_parallel": 2, "per_case_timeout_s": 900, "per_run_max_usd": 2.0, + "daily_max_usd": 5.0} + assert suites.normalize_budget({"max_parallel": 2, "per_run_max_usd": 1}) == {"per_run_max_usd": 1.0} + with pytest.raises(CaseInvalid) as caught: + suites.normalize_budget({"max_parallel": 1.5}) + assert caught.value.field == "budget.max_parallel" and caught.value.code == suites.SUITE_INVALID diff --git a/backend/tests/test_evals_api.py b/backend/tests/test_evals_api.py new file mode 100644 index 00000000..9fc4e1c1 --- /dev/null +++ b/backend/tests/test_evals_api.py @@ -0,0 +1,503 @@ +"""评测接口（app/api/evals.py）：题集和题目的增删改查（乐观锁）、保存时试算、从运行认定、口径升版处置、导出导入、 +目录快照；零售样例题集（tests/fixtures/eval/retail_suite.json）在虚构零售库上试算通过。 + +库、表、数字、人名都是虚构的。 +""" +from __future__ import annotations + +import json +import sqlite3 +import uuid +from datetime import datetime, timedelta +from pathlib import Path +from urllib.parse import quote +from zoneinfo import ZoneInfo + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete + +from app.data import introspect +from app.db.base import SessionLocal +from app.db.models import DataSource, Run, Workflow, WorkflowVersion +from app.db.models_eval import EvalRun, EvalSuite +from app.engine.runner import run_manager +from app.eval import snapshots +from app.main import app +from tests.fixtures import dual_check_shop as shop +from tests.fixtures.acceptance import retail +from tests.fixtures.catalog.fake_model import FakeModel +from tests.fixtures.sources import drop_source + +SUITE_FILE = Path(__file__).parent / "fixtures" / "eval" / "retail_suite.json" +LAST_WEEK = "BETWEEN '{{ schedule.last_week_start }}' AND '{{ schedule.last_week_end }}'" +#: dual_check_shop 的订单都在 2026-09：「上周」按截至时间算，这里用写死日期的 SQL +SEPT = "BETWEEN '2026-09-01' AND '2026-09-07'" +ACTOR = {"X-Actor": quote("评测员甲")} + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + 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 +async def source(tmp_path): + name, sid = await shop.add_source(shop.build(tmp_path / "shop.db")) + yield {"name": name, "id": sid} + await drop_source(sid) + + +async def new_suite(client, **body) -> dict: + res = await client.post("/api/evals/suites", json={"name": f"虚构题集_{uuid.uuid4().hex[:6]}", **body}, headers=ACTOR) + assert res.status_code == 201, res.text + return res.json() + + +def sql_case(source_id: str, sql: str = f"SELECT SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' " + f"AND paid_date {SEPT}", **kw) -> dict: + return {"question": "9 月第一周按支付日、不含退款的销售额？", "datasource_ids": [source_id], + "gold": {"kind": "sql", "datasource_id": source_id, "sql": sql}, + "expectations": [{"kind": "scalar", "column": "gmv"}], **kw} + + +# -------------------------------------------------------------------------- +# 题集 +# -------------------------------------------------------------------------- + + +async def test_suites_crud_with_optimistic_lock(client): + made = await new_suite(client, description="说明", budget={"max_parallel": 3}) + assert made["created_by"] == "评测员甲" and made["version"] == 1 and made["timezone"] == "Asia/Shanghai" + assert made["budget"] == {"max_parallel": 3} + assert made["effective_budget"] == {"max_parallel": 3, "per_case_timeout_s": 900, "per_run_max_usd": 2.0, + "daily_max_usd": 5.0} + listed = (await client.get("/api/evals/suites")).json()["suites"] + assert any(s["id"] == made["id"] and s["counts"] == {"cases": 0, "enabled": 0} for s in listed) + + res = await client.patch(f"/api/evals/suites/{made['id']}", json={"name": "改了名"}) + assert res.status_code == 422 and res.json()["field"] == "version" + res = await client.patch(f"/api/evals/suites/{made['id']}", json={"name": "改了名", "version": 1}) + assert res.status_code == 200 and res.json()["name"] == "改了名" and res.json()["version"] == 2 + assert res.json()["budget"] == {"max_parallel": 3}, "没带的项照旧" + res = await client.patch(f"/api/evals/suites/{made['id']}", json={"name": "再改", "version": 1}) + assert res.status_code == 409 and res.json()["code"] == "eval_version_conflict" + + for body, field in (({"name": ""}, "name"), ({"name": "x", "timezone": "火星"}, "timezone"), + ({"name": "x", "budget": {"per_case_timeout_s": 5}}, "budget.per_case_timeout_s"), + ({"name": "x", "budget": {"per_run_max_usd": 50, "daily_max_usd": 10}}, "budget.per_run_max_usd"), + ({"name": "x", "budget": {"nope": 1}}, "budget.nope")): + res = await client.post("/api/evals/suites", json=body) + assert res.status_code == 422 and res.json()["field"] == field, (body, res.text) + + # 有评测记录的不能删，只能归档 + async with SessionLocal() as session: + session.add(EvalRun(suite_id=made["id"], status="succeeded")) + await session.commit() + res = await client.delete(f"/api/evals/suites/{made['id']}") + assert res.status_code == 409 and res.json()["code"] == "eval_suite_has_runs" + async with SessionLocal() as session: + await session.execute(delete(EvalRun).where(EvalRun.suite_id == made["id"])) + await session.commit() + assert (await client.delete(f"/api/evals/suites/{made['id']}")).status_code == 204 + assert (await client.get(f"/api/evals/suites/{made['id']}")).status_code == 404 + + +# -------------------------------------------------------------------------- +# 题目：保存时试算 +# -------------------------------------------------------------------------- + + +async def test_a_case_is_saved_only_after_a_trial_passes(client, source): + suite = await new_suite(client) + url = f"/api/evals/suites/{suite['id']}/cases" + res = await client.post(url, json=sql_case(source["id"]), headers=ACTOR) + assert res.status_code == 201, res.text + body = res.json() + case, trial = body["case"], body["trial"] + assert case["seq"] == 1 and case["version"] == 1 and case["created_by"] == "评测员甲" and len(case["case_hash"]) == 64 + assert trial["status"] == "ok" and trial["sets"][0]["values"]["e1"]["value"] == 330.0 + # 题目上只记试算的结论和运行，不存期望值（期望每次重算） + assert case["trial"]["status"] == "ok" and case["trial"]["by"] == "评测员甲" and "sets" not in case["trial"] + assert case["gold_source"] == {"id": source["id"], "name": source["name"]} + async with SessionLocal() as session: + run = await session.get(Run, trial["run_ids"][0]) + assert run.origin_kind == "eval" and run.origin["trial"] is True and run.origin["suite_id"] == suite["id"] + + # 只试算、不保存 + res = await client.post(f"{url}/trial", json=sql_case(source["id"], sql="SELECT nope FROM orders")) + assert res.status_code == 200 and res.json()["trial"]["status"] == "failed" + + for bad, code, field in ( + (sql_case(source["id"], expectations=[{"kind": "scalar", "column": "gmv", "value": 330}]), + "eval_case_invalid", "expectations.0.value"), # 评测-D4：数值期望不许人填 + (sql_case(source["id"], history=[{"question": "上一轮"}]), "eval_case_invalid", "history"), # 评测-D12 + (sql_case(source["id"], sql="DELETE FROM orders"), "eval_gold_guard", "gold.sql"), + (sql_case(source["id"], sql="SELECT SUM(amount) AS gmv FROM orders WHERE {{ schedule.nope }}"), + "eval_case_invalid", "gold.sql"), + (sql_case(source["id"], sql="SELECT region FROM orders WHERE region = '华中'"), "eval_gold_no_value", "gold"), + ({"question": "q", "expectations": [{"kind": "scalar", "column": "gmv"}]}, "eval_case_invalid", "gold"), + ({"question": "最近卖得怎么样", "expectations": [{"kind": "ask"}, {"kind": "none"}], + "gold": {"kind": "sql", "datasource_id": source["id"], "sql": "SELECT 1"}}, "eval_case_invalid", "expectations"), + (sql_case("nope"), "eval_case_invalid", "gold.datasource_id"), + ): + res = await client.post(url, json=bad) + assert res.status_code == 422, (bad, res.text) + assert (res.json()["code"], res.json().get("field")) == (code, field), res.json() + detail = (await client.get(f"/api/evals/suites/{suite['id']}")).json() + assert [c["seq"] for c in detail["cases"]] == [1], "没过的一道都没存" + + +async def test_masked_columns_are_refused_at_save(client, tmp_path): + name, sid = await shop.add_source(shop.build(tmp_path / "masked.db"), options={"mask_columns": ["region"]}) + try: + suite = await new_suite(client) + body = sql_case(sid, sql="SELECT region AS 地区, SUM(amount) AS gmv FROM orders GROUP BY region", + expectations=[{"kind": "table", "keys": ["地区"]}]) + res = await client.post(f"/api/evals/suites/{suite['id']}/cases", json=body) + assert res.status_code == 422 and res.json()["code"] == "eval_gold_masked" and "遮罩" in res.json()["detail"] + finally: + await drop_source(sid) + + +async def test_editing_a_case_retries_only_when_its_content_changes(client, source): + suite = await new_suite(client) + made = (await client.post(f"/api/evals/suites/{suite['id']}/cases", json=sql_case(source["id"]))).json()["case"] + url = f"/api/evals/suites/{suite['id']}/cases/{made['id']}" + res = await client.patch(url, json={"tags": ["口径陷阱"], "version": 1}) + assert res.status_code == 200 and res.json()["trial"] is None, "只改标签不试算" + case = res.json()["case"] + assert case["tags"] == ["口径陷阱"] and case["version"] == 2 and case["case_hash"] == made["case_hash"] + res = await client.patch(url, json={"gold": {"kind": "sql", "datasource_id": source["id"], + "sql": f"SELECT SUM(amount) AS gmv FROM orders WHERE paid_date {SEPT}"}, + "version": 2}) + assert res.status_code == 200 and res.json()["trial"]["sets"][0]["values"]["e1"]["value"] == 380.0 + assert res.json()["case"]["case_hash"] != made["case_hash"], "题目改过，指纹跟着变" + res = await client.patch(url, json={"gold": {"kind": "sql", "datasource_id": source["id"], + "sql": "DELETE FROM orders"}, "version": 3}) + assert res.status_code == 422 and res.json()["code"] == "eval_gold_guard" + res = await client.patch(url, json={"tags": [], "version": 2}) + assert res.status_code == 409 + # 已有的题按此刻再试算一次：记进 trial、不改 version + res = await client.post(f"{url}/trial") + assert res.status_code == 200 and res.json()["trial"]["status"] == "ok" + assert (await client.get(f"/api/evals/suites/{suite['id']}")).json()["cases"][0]["version"] == 3 + assert (await client.delete(url)).status_code == 204 + + +# -------------------------------------------------------------------------- +# 从运行认定 +# -------------------------------------------------------------------------- + + +async def test_a_query_from_a_run_becomes_a_case(client, source): + from app.engine.launch import launch_run + + sql = f"SELECT SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {SEPT}" + graph = {"nodes": [{"id": "start", "type": "input", "data": {"config": {}}}, + {"id": "q", "type": "tool", "data": {"config": {"tool": f"db_query__{source['name']}", + "args": {"sql": sql}}}}, + {"id": "out", "type": "output", "data": {"config": {}}}], + "edges": [{"source": "start", "target": "q"}, {"source": "q", "target": "out"}]} + async with SessionLocal() as session: + run = await launch_run(session, graph=graph, input_payload={"question": "9 月第一周卖了多少？"}, actor="提问人", + origin_kind="chat") + await run_manager.wait_idle(run.id, timeout=30) + queries = (await client.get(f"/api/evals/runs/{run.id}/queries")).json()["queries"] + assert len(queries) == 1 and queries[0]["sql"] == sql and queries[0]["source_id"] == source["id"] + assert "rows" not in queries[0], "只给 SQL、列名、行数，不给结果的格" + suite = await new_suite(client) + res = await client.post(f"/api/evals/suites/{suite['id']}/cases/from-run", headers=ACTOR, + json={"run_id": run.id, "query": queries[0]["artifact"], + "expectations": [{"kind": "scalar", "column": "gmv"}]}) + assert res.status_code == 201, res.text + case = res.json()["case"] + assert case["question"] == "9 月第一周卖了多少？", "问法没给就取那次运行的问题" + gold = case["gold"] + assert gold["kind"] == "from_run" and gold["sql"] == sql and gold["run_id"] == run.id + assert gold["confirmed_by"] == "评测员甲" and gold["confirmed_at"] and gold["query"] == queries[0]["artifact"] + assert res.json()["trial"]["sets"][0]["values"]["e1"]["value"] == 330.0 + # 之后按这条 SQL 重算，不冻结数值；要改 SQL 就改成 SQL 题，不能在「从运行认定」上改 + res = await client.patch(f"/api/evals/suites/{suite['id']}/cases/{case['id']}", + json={"gold": {**gold, "sql": "SELECT 1 AS gmv"}, "version": 1}) + assert res.status_code == 422 and res.json()["field"] == "gold.kind" + res = await client.patch(f"/api/evals/suites/{suite['id']}/cases/{case['id']}", + json={"gold": gold, "tags": ["从运行"], "version": 1}) + assert res.status_code == 200 and res.json()["case"]["gold"]["confirmed_by"] == "评测员甲" + res = await client.post(f"/api/evals/suites/{suite['id']}/cases/from-run", + json={"run_id": run.id, "query": "nope", "expectations": [{"kind": "none"}]}) + assert res.status_code == 404 + # 客户端不能自己写「从运行认定」 + res = await client.post(f"/api/evals/suites/{suite['id']}/cases", json={**sql_case(source["id"]), "gold": gold}) + assert res.status_code == 422 and res.json()["field"] == "gold.kind" + + +# -------------------------------------------------------------------------- +# 口径题：清单与升版处置 +# -------------------------------------------------------------------------- + + +def _weekly(source_name: str, *, refunds: bool = False) -> dict: + status = "" if refunds else "status <> 'refunded' AND " + + def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "data": {"label": nid, "config": config}} + + nodes = [node("start", "input", fields=[{"name": "start", "default": "2026-09-01"}, + {"name": "end", "default": "2026-09-07"}]), + node("fetch", "tool", tool=f"db_query__{source_name}", args={"sql": ( + f"SELECT SUM(amount) AS gmv FROM orders WHERE {status}" + "paid_date BETWEEN '{{ input.start }}' AND '{{ input.end }}'")}), + node("card", "metrics", caliber="周销售", metrics=[{"id": "gmv", "name": "销售额", + "expression": "cell(nodes.fetch, 0, 'gmv')"}]), + node("out", "output")] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def test_caliber_cases_list_and_dispose_upgrades(client, source): + async with SessionLocal() as session: + wf = Workflow(name=f"周销售_{uuid.uuid4().hex[:6]}", graph=_weekly(source["name"]), status="published", + published_version=1, version=1) + session.add(wf) + await session.flush() + session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=_weekly(source["name"]), level="published")) + await session.commit() + wf_id = wf.id + calibers = (await client.get("/api/evals/calibers")).json()["calibers"] + mine = [c for c in calibers if c["workflow_id"] == wf_id] + assert mine == [{"workflow_id": wf_id, "workflow_name": mine[0]["workflow_name"], "version": 1, + "level": "published", "node_id": "card", "card": "card", "caliber": "周销售 @ v1", + "metric_id": "gmv", "name": "销售额", "unit": "", "reusable": True, "reason": None}] + suite = await new_suite(client) + body = {"question": "上周的销售额？", "gold": {"kind": "caliber", "workflow_id": wf_id, "version": 1, + "node_id": "card", "inputs": {}}, + "expectations": [{"kind": "scalar", "metric": "gmv"}]} + res = await client.post(f"/api/evals/suites/{suite['id']}/cases", json=body) + assert res.status_code == 201, res.text + case = res.json()["case"] + assert case["upgrade"]["pending"] is False and res.json()["trial"]["sets"][0]["values"]["e1"]["value"] == 330.0 + bad = {**body, "expectations": [{"kind": "scalar", "metric": "nope"}]} + assert (await client.post(f"/api/evals/suites/{suite['id']}/cases", json=bad)).json()["field"] == \ + "expectations.0.metric" + + async with SessionLocal() as session: + wf = await session.get(Workflow, wf_id) + wf.published_version = wf.version = 2 + session.add(WorkflowVersion(workflow_id=wf_id, version=2, graph=_weekly(source["name"], refunds=True), + level="published")) + await session.commit() + detail = (await client.get(f"/api/evals/suites/{suite['id']}")).json() + assert detail["pending_upgrades"] == 1 and detail["cases"][0]["upgrade"]["message"] == "口径已升版到 v2，待处置" + url = f"/api/evals/suites/{suite['id']}/cases/{case['id']}/upgrade" + info = (await client.get(url)).json() + assert set(info["choices"]) == {"recompute", "dual", "incomparable"} + res = await client.post(url, json={"version": 1}) + assert res.status_code == 422 and res.json()["code"] == "eval_dispose_invalid", "没有缺省" + # 不可比：旧题归档，按新版本另建新题 + res = await client.post(url, json={"policy": "incomparable", "version": 1}, headers=ACTOR) + assert res.status_code == 200, res.text + old, new = res.json()["case"], res.json()["new_case"] + assert old["archived"] and not old["enabled"] and old["superseded_by"] == new["id"] + assert new["gold"]["version"] == 2 and new["seq"] == 2 and new["upgrade"]["pending"] is False + assert res.json()["trial"]["sets"][0]["values"]["e1"]["value"] == 380.0 + detail = (await client.get(f"/api/evals/suites/{suite['id']}")).json() + assert detail["pending_upgrades"] == 0 and detail["counts"]["cases"] == 1 + # 归档的题不再待处置；在新题上再来一版就改钉 + async with SessionLocal() as session: + wf = await session.get(Workflow, wf_id) + wf.published_version = wf.version = 3 + session.add(WorkflowVersion(workflow_id=wf_id, version=3, graph=_weekly(source["name"]), level="published")) + await session.commit() + res = await client.post(f"/api/evals/suites/{suite['id']}/cases/{new['id']}/upgrade", + json={"policy": "recompute", "version": 1}, headers=ACTOR) + assert res.status_code == 200, res.text + moved = res.json()["case"] + assert moved["gold"]["version"] == 3 and moved["gold"]["upgrade"]["by"] == "评测员甲" + assert moved["case_hash"] != new["case_hash"] and moved["version"] == 2 + res = await client.post(f"/api/evals/suites/{suite['id']}/cases/{new['id']}/upgrade", + json={"policy": "dual", "version": 2}) + assert res.status_code == 409 and res.json()["code"] == "eval_dispose_nothing_pending" + + +# -------------------------------------------------------------------------- +# 导出导入 +# -------------------------------------------------------------------------- + + +async def test_export_then_import_matches_sources_by_name(client, source): + suite = await new_suite(client, description="导出用") + await client.post(f"/api/evals/suites/{suite['id']}/cases", json=sql_case(source["id"], tags=["口径陷阱"])) + await client.post(f"/api/evals/suites/{suite['id']}/cases", + json={"question": "最近怎么样？", "expectations": [{"kind": "ask"}]}) + res = await client.get(f"/api/evals/suites/{suite['id']}/export") + assert res.status_code == 200 and "attachment" in res.headers["content-disposition"] + content = res.json() + text = json.dumps(content, ensure_ascii=False) + assert content["format"] == "agentlab-eval-suite" and len(content["cases"]) == 2 + # 按名字、不含期望值、答案、运行 + first = content["cases"][0] + assert first["datasources"] == [source["name"]] and first["gold"]["datasource"] == source["name"] + assert "datasource_id" not in first["gold"] and source["id"] not in text + assert "330" not in text and "run_id" not in text and "trial" not in text + + preview = (await client.post("/api/evals/suites/import/preview", json={"content": content})).json() + assert preview["importable"] == 2 and preview["sources"] == {source["name"]: source["id"]} + res = await client.post("/api/evals/suites/import", json={"content": content, "name": "导进来的"}, headers=ACTOR) + assert res.status_code == 201, res.text + imported = res.json()["suite"] + assert imported["name"] == "导进来的" and res.json()["imported"] == 2 and res.json()["skipped"] == [] + assert [c["question"] for c in imported["cases"]] == [c["question"] for c in content["cases"]] + assert imported["cases"][0]["gold"]["datasource_id"] == source["id"] + assert imported["cases"][0]["trial"]["status"] == "ok" + + # 对不上的数据源：跳过、写原因；一道都导不进来就不建题集 + other = json.loads(json.dumps(content).replace(source["name"], "ghost_source")) + preview = (await client.post("/api/evals/suites/import/preview", json={"content": other})).json() + assert preview["skipped"] == 1 and "ghost_source" in preview["cases"][0]["reason"] + only_bad = {**other, "cases": other["cases"][:1]} + res = await client.post("/api/evals/suites/import", json={"content": only_bad}) + assert res.status_code == 422 and "ghost_source" in res.json()["detail"] + res = await client.post("/api/evals/suites/import/preview", json={"content": {"format": "x"}}) + assert res.status_code == 422 and res.json()["code"] == "eval_import_invalid" + + +async def test_the_retail_sample_suite_trials_on_the_retail_database(client, tmp_path): + """仓库里附的零售样例题集：在虚构零售库上导入，每道题都试算通过，期望和直接查库一样，陷阱按口径算。""" + today = datetime.now(ZoneInfo("Asia/Shanghai")).date() + path = retail.build(tmp_path / "retail.db", today) + await drop_source(name=retail.SOURCE_NAME) + async with SessionLocal() as session: + row = DataSource(name=retail.SOURCE_NAME, kind="sqlite", database=str(path), readonly=True, + options={"mask_columns": ["phone"]}) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + sid = row.id + try: + content = json.loads(SUITE_FILE.read_text(encoding="utf-8")) + res = await client.post("/api/evals/suites/import", json={"content": content}, headers=ACTOR) + assert res.status_code == 201, res.text + assert res.json()["imported"] == 6 and res.json()["skipped"] == [] + suite = res.json()["suite"] + cases = {c["seq"]: c for c in suite["cases"]} + assert all(c["trial"]["status"] == "ok" for c in cases.values()) + + monday = today - timedelta(days=today.weekday()) + start, end = (monday - timedelta(days=7)).isoformat(), (monday - timedelta(days=1)).isoformat() + + def one(sql: str): + with sqlite3.connect(path) as db: + return db.execute(sql).fetchone() + + async def trial(seq: int) -> dict: + out = (await client.post(f"/api/evals/suites/{suite['id']}/cases/{cases[seq]['id']}/trial")).json()["trial"] + assert out["status"] == "ok", out + assert f"'{start}' AND '{end}'" in (out["rendered"].get("sql") or f"'{start}' AND '{end}'") + return out["sets"][0]["values"] if out["sets"] else {} + + paid = one(f"SELECT ROUND(SUM(amount), 2) FROM orders WHERE status = 'paid' AND paid_date BETWEEN '{start}' " + f"AND '{end}'")[0] + by_order_date = one(f"SELECT ROUND(SUM(amount), 2) FROM orders WHERE status = 'paid' AND order_date BETWEEN " + f"'{start}' AND '{end}'")[0] + with_refunds = one(f"SELECT ROUND(SUM(amount), 2) FROM orders WHERE status <> 'cancelled' AND paid_date " + f"BETWEEN '{start}' AND '{end}'")[0] + values = await trial(1) + assert values["sales"]["value"] == paid + # 陷阱确实埋着：按下单日、含退款算出来的都不一样 + assert paid not in (by_order_date, with_refunds) + refunds = await trial(2) + assert (refunds["refund_orders"]["value"], refunds["refund_amount"]["value"]) == one( + f"SELECT COUNT(*), ROUND(SUM(amount), 2) FROM orders WHERE status = 'refunded' AND paid_date BETWEEN " + f"'{start}' AND '{end}'") + assert refunds["refund_orders"]["value"] > 0 + cancelled = await trial(3) + assert cancelled["cancelled"]["value"] == one( + f"SELECT COUNT(*) FROM orders WHERE status = 'cancelled' AND order_date BETWEEN '{start}' AND '{end}'")[0] + regions = (await trial(4))["by_region"] + assert [r["地区"] for r in regions["rows"]] == ["华南", "华东"] and regions["ordered"] is True + assert regions["rows"][0]["销售额"] > regions["rows"][1]["销售额"] + assert (await trial(5)) == {"no_data": {"kind": "none", "rows": 0}} + ask = (await client.post(f"/api/evals/suites/{suite['id']}/cases/{cases[6]['id']}/trial")).json()["trial"] + assert ask["status"] == "ok" and ask["run_ids"] == [] and cases[6]["expectations"][0]["kind"] == "ask" + assert cases[1]["tags"] == ["口径陷阱", "支付日归属", "相对日期"] + finally: + async with SessionLocal() as session: + await session.execute(delete(EvalSuite).where(EvalSuite.name == "零售样例题集")) + await session.commit() + await drop_source(sid) + + +# -------------------------------------------------------------------------- +# 目录快照 +# -------------------------------------------------------------------------- + + +async def test_catalog_snapshots_live_and_shadow(client, source, monkeypatch): + res = await client.post("/api/evals/catalog-snapshots", json={"source_ids": [source["id"]]}, headers=ACTOR) + assert res.status_code == 201, res.text + live = res.json() + assert live["status"] == "succeeded" and live["view"] == {"mode": "snapshot", "statuses": "all", + "snapshot": live["artifact_id"]} + assert (await client.post("/api/evals/catalog-snapshots", json={"source_ids": []})).status_code == 422 + + # 测试环境没有真实接入：起草模型用不了，照实回 409，不建任务 + res = await client.post("/api/evals/catalog-snapshots/shadow/estimate", + json={"sources": [{"id": source["id"], "tables": ["orders"]}]}) + assert res.status_code == 409 and res.json()["code"] == snapshots.SNAPSHOT_MODEL_UNAVAILABLE + + async def fake_model(session, provider, model): + return FakeModel({"orders": {"label": "影子订单"}}), "fake-draft" + + monkeypatch.setattr(snapshots, "resolve_model", fake_model) + monkeypatch.setattr(snapshots.ShadowDraft, "model_override", FakeModel({"orders": {"label": "影子订单"}})) + est = (await client.post("/api/evals/catalog-snapshots/shadow/estimate", + json={"sources": [{"id": source["id"], "tables": ["orders"]}]})).json() + assert est["sources"][0]["tables"] == ["orders"] and est["estimate"]["calls"] == 1 + assert est["estimate"]["unpriced"] is True and est["estimate"]["model"] == "fake-draft" + res = await client.post("/api/evals/catalog-snapshots/shadow", headers=ACTOR, + json={"sources": [{"id": source["id"], "tables": ["orders"]}], "guidance": "按支付日"}) + assert res.status_code == 202 and res.json()["status"] in ("queued", "running") + snap_id = res.json()["id"] + await snapshots.manager.wait(snap_id, timeout=30) + done = (await client.get(f"/api/evals/catalog-snapshots/{snap_id}")).json() + assert done["status"] == "succeeded" and done["artifact_id"] and done["created_by"] == "评测员甲" + assert done["draft"]["guidance"] == "按支付日" + listed = (await client.get("/api/evals/catalog-snapshots")).json()["snapshots"] + assert {live["id"], snap_id} <= {s["id"] for s in listed} + res = await client.post(f"/api/evals/catalog-snapshots/{snap_id}/stop") + assert res.status_code == 409 and res.json()["code"] == snapshots.SNAPSHOT_FINISHED + assert (await client.get("/api/evals/catalog-snapshots/nope")).status_code == 404 + + +async def test_evals_are_for_builders_only(): + """期望值和查询原文只给搭建者及以上：使用者 403，搭建者能用。""" + from tests.fakes import auth_env + + try: + admin = await auth_env.turn_on_local() + await admin.aclose() + await auth_env.make_user("eval_user", "user") + await auth_env.make_user("eval_builder", "builder") + user = await auth_env.login("eval_user") + res = await user.get("/api/evals/suites") + assert res.status_code == 403 and res.json()["code"] == "role_required" + assert (await user.post("/api/evals/suites", json={"name": "x"})).status_code == 403 + await user.aclose() + builder = await auth_env.login("eval_builder") + res = await builder.post("/api/evals/suites", json={"name": "搭建者的题集"}) + assert res.status_code == 201 and res.json()["created_by"] == "eval_builder" + await builder.aclose() + finally: + await auth_env.reset() + async with SessionLocal() as session: + await session.execute(delete(EvalSuite).where(EvalSuite.name == "搭建者的题集")) + await session.commit()