mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
评测接口与题集导出导入、零售样例题集（第 2 波 W2-c）
- 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）：支付日归属、退款、取消、华南高于华东，一道「华中区」期望没有数据，一道问法含糊期望反问。测试在
按今天生成的零售库上导入，六道题都试算通过、期望和直接查库一样、陷阱按口径算。
This commit is contained in:
ThearchyHelios
+683
-3
@@ -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"]
|
||||
|
||||
@@ -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.* }} 变量（按评测的截至时间渲染成明确的日期）
|
||||
|
||||
@@ -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"]
|
||||
+5
@@ -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 = [
|
||||
|
||||
+172
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user