From 7f143134161e3054cfb390d2c861e348e3209b60 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 14:05:22 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=84=E6=B5=8B=E9=A2=98=E7=9B=AE=E7=9A=84?= =?UTF-8?q?=E5=87=A0=E5=A4=84=E6=94=B6=E7=B4=A7=EF=BC=9A=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E9=A2=98=E5=8F=AA=E8=83=BD=E5=9B=BA=E5=AE=9A=E5=B7=B2=E5=8F=91?= =?UTF-8?q?=E5=B8=83=E6=9C=AA=E9=80=80=E5=BD=B9=E7=9A=84=E7=89=88=E6=9C=AC?= =?UTF-8?q?=E3=80=81=E5=A4=84=E7=BD=AE=E5=8D=87=E7=89=88=E5=90=8E=E6=8C=89?= =?UTF-8?q?=E6=96=B0=E7=89=88=E6=9C=AC=E5=86=8D=E6=A0=B8=E4=B8=80=E9=81=8D?= =?UTF-8?q?=E3=80=81=E8=AF=95=E7=AE=97=E5=89=8D=E7=BB=93=E6=9D=9F=E8=AF=BB?= =?UTF-8?q?=E4=BA=8B=E5=8A=A1=E3=80=81=E5=BD=B1=E5=AD=90=E8=B5=B7=E8=8D=89?= =?UTF-8?q?=E5=88=86=E5=BC=80=E8=AE=B0=E5=8F=AA=E5=86=99=E5=87=BA=E4=B8=80?= =?UTF-8?q?=E9=83=A8=E5=88=86=E7=9A=84=E8=A1=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - cases.resolve_case：口径题钉的版本没发布过或已退役的，保存时照实拒收（422，field=gold.version）；以前要等试算跑到 升版判断才说得出。 - 处置升版（recompute / incomparable）改钉新版本之前，按新版本再过一遍 resolve_case（入口字段、指标），说得清是哪一项不行。 - 接口里的试算要等一次运行跑完（最多两分钟）：先结束这个请求的读事务再等，不一直开着它（导入时同样）。 - 影子起草：只写出一部分的表另记（tables_partial），快照照出、状态 partial；一张都没起草出来才算失败、不出快照。以前 只写出一部分的表算成没起草出来，几张表都只写出一部分时整次记失败、写好的部分也丢了。 - 题集里推起草哪些表时，期望 SQL 按题集的时区渲染。 --- backend/app/api/evals.py | 27 ++++++++++++++------- backend/app/eval/cases.py | 4 ++++ backend/app/eval/gold.py | 2 +- backend/app/eval/snapshots.py | 21 ++++++++++------ backend/tests/test_eval_snapshots.py | 36 +++++++++++++++++++++++++++- backend/tests/test_evals_api.py | 15 ++++++++++++ 6 files changed, 87 insertions(+), 18 deletions(-) diff --git a/backend/app/api/evals.py b/backend/app/api/evals.py index c3f5fdc8..2e3acc95 100644 --- a/backend/app/api/evals.py +++ b/backend/app/api/evals.py @@ -156,9 +156,11 @@ def _context(suite: EvalSuite, actor: str | None, **extra: Any) -> gold_rules.Co 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: - """按此刻试算一次期望。""" +async def _trial(session: AsyncSession, suite: EvalSuite, fields: dict[str, Any], actor: str | None, *, + case_id: str | None = None, seq: int | None = None) -> gold_rules.GoldOutcome: + """按此刻试算一次期望。试算要等一次运行跑完（最多 TRIAL_TIMEOUT_S 秒）：先结束这个请求的读事务，不一直开着它等 + （会话的 expire_on_commit 是关的，读好的对象照用）。""" + await session.commit() return await gold_rules.compute_gold({**fields, "id": case_id, "seq": seq}, _context(suite, actor)) @@ -305,7 +307,7 @@ async def create_case(suite_id: str, request: Request, session: AsyncSession = D 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) + outcome = await _trial(session, 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)), @@ -318,7 +320,7 @@ async def trial_unsaved(suite_id: str, request: Request, session: AsyncSession = """只试算、不保存（界面上写 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()} + return {"trial": (await _trial(session, suite, fields, actor_of(x_actor))).to_dict()} @router.patch("/suites/{suite_id}/cases/{case_id}") @@ -335,7 +337,7 @@ async def update_case(suite_id: str, case_id: str, request: Request, session: As 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) + outcome = await _trial(session, 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) @@ -366,7 +368,7 @@ async def trial_case(suite_id: str, case_id: str, session: AsyncSession = Depend 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) + outcome = await _trial(session, 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() @@ -448,7 +450,7 @@ async def create_from_run(suite_id: str, request: Request, session: AsyncSession "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) + outcome = await _trial(session, 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()} @@ -503,7 +505,13 @@ async def dispose_upgrade(suite_id: str, case_id: str, request: Request, session 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, + if policy != "dual": + # 改钉到新版本：按新版本再核一遍（入口字段、指标），说得清是哪一项不行；dual 的新版由试算一起算 + try: + await resolve_case(session, fields, tz=suite.timezone) + except CaseInvalid as e: + raise _invalid(e) from e + outcome = await _trial(session, 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": @@ -589,6 +597,7 @@ async def import_suite(request: Request, session: AsyncSession = Depends(get_ses _context(suite, actor, imported=True)) passed: list[tuple[transfer.PlannedCase, gold_rules.GoldOutcome]] = [] + await session.commit() # 试算要等运行跑完：不一直开着读事务等（同 _trial） for item, outcome in zip(ready, await asyncio.gather(*(trial(c) for c in ready))): if outcome.ok: passed.append((item, outcome)) diff --git a/backend/app/eval/cases.py b/backend/app/eval/cases.py index 75e315cd..c505efd7 100644 --- a/backend/app/eval/cases.py +++ b/backend/app/eval/cases.py @@ -470,6 +470,10 @@ async def resolve_case(session: AsyncSession, fields: Mapping[str, Any], *, tz: pinned = await derive.load_pinned(session, gold["workflow_id"], gold["version"], gold["node_id"]) except derive.NotDerivable as e: raise CaseInvalid(f"这张口径卡不能当口径题：{e.message}", "gold", CALIBER_NOT_REUSABLE) from e + if pinned.retired or not pinned.published: + raise CaseInvalid(f"「{pinned.workflow_name}」v{pinned.version} " + f"{'已退役' if pinned.retired else '没有发布过'}：口径题只能固定已发布且未退役的版本", + "gold.version", CALIBER_NOT_REUSABLE) refs.pinned = pinned ids_ = {m["id"] for m in pinned.metrics} for i, exp in enumerate(fields.get("expectations") or []): diff --git a/backend/app/eval/gold.py b/backend/app/eval/gold.py index 28bb86e5..afd1bd91 100644 --- a/backend/app/eval/gold.py +++ b/backend/app/eval/gold.py @@ -179,8 +179,8 @@ def _origin(case: Mapping[str, Any], ctx: Context, **more: Any) -> dict[str, Any # ========================================================================== -#: 探针运行带的选项：查询一律只读、结果在工具边界就按遮罩遮好（同使用者的问数据运行） def _probe_options() -> dict[str, Any]: + """探针运行带的选项：查询一律只读、结果在工具边界就按遮罩遮好（同使用者的问数据运行）。""" from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION return {MASK_RESULTS_OPTION: True, READ_ONLY_OPTION: True} diff --git a/backend/app/eval/snapshots.py b/backend/app/eval/snapshots.py index 48fd82f1..183a754b 100644 --- a/backend/app/eval/snapshots.py +++ b/backend/app/eval/snapshots.py @@ -159,12 +159,15 @@ async def suite_tables(session: AsyncSession, suite_id: str) -> dict[str, list[s """题集用得到的表：{源 id: [SQL 里写的表名]}。从期望 SQL（SQL 题、从运行认定）和口径题的取数 SQL 推；工作流题、 只有反问的题推不出。名字还没对上表结构（plan_shadow 再对）。""" from app.db.models import DataSource - from app.db.models_eval import EvalCase + from app.db.models_eval import EvalCase, EvalSuite from app.eval.cases import render, variable_values from app.ledger import derive from app.tools.datasource import QUERY_PREFIX - values = variable_values(utcnow(), "Asia/Shanghai") + suite = await session.get(EvalSuite, suite_id) + if suite is None: + raise SnapshotInvalid("题集不存在，可能已被删除", field="suite_id") + values = variable_values(utcnow(), suite.timezone) out: dict[str, list[str]] = {} rows = (await session.execute(select(EvalCase).where(EvalCase.suite_id == suite_id, EvalCase.archived.is_(False)) .order_by(EvalCase.seq))).scalars().all() @@ -359,7 +362,9 @@ class ShadowDraft(runtime.JobRun): self.copy: dict[str, dict[str, dict[str, Any]]] = {} self.usage = {"calls": 0, "input_tokens": 0, "output_tokens": 0, "timeouts": 0, "failed_calls": 0} self.done = 0 + #: 没起草出来的表（一项都没写出来、表结构里没有）和只写出一部分的表：{「源名.表名」: 原因} self.failed: dict[str, str] = {} + self.partial: dict[str, str] = {} self.notice: str | None = None def lane_of(self, job: Any) -> str | None: @@ -451,8 +456,8 @@ class ShadowDraft(runtime.JobRun): self.done += 1 if outcome is not None and outcome.status == "failed": self.failed[f"{source.name}.{table}"] = outcome.error or catalog.MODEL_EMPTY_REASON - elif outcome is not None and outcome.status == "partial" and outcome.error: - self.failed[f"{source.name}.{table}"] = outcome.error + elif outcome is not None and outcome.status == "partial": + self.partial[f"{source.name}.{table}"] = outcome.error or "只写出了一部分" await self._write(self._save_progress) return hook @@ -468,7 +473,8 @@ class ShadowDraft(runtime.JobRun): "unpriced": bool(self.usage["calls"]) and not priced} def _progress(self, current: dict[str, Any]) -> dict[str, Any]: - return {**current, "tables_done": self.done, "tables_failed": len(self.failed), "failures": dict(self.failed), + return {**current, "tables_done": self.done, "tables_failed": len(self.failed), + "tables_partial": len(self.partial), "failures": {**self.partial, **self.failed}, "notice": self.notice} async def _save_progress(self, session: AsyncSession) -> None: @@ -495,8 +501,9 @@ class ShadowDraft(runtime.JobRun): "draft": {**dict(snap.draft or {}), "model_id": self.model_id or None}, "tables": planned} content = _content({sid: self.copy.get(sid, {}) for sid in ids}, taken=taken) total = sum(len(v or []) for v in planned.values()) - ok = total - len(self.failed) - status = "succeeded" if not self.failed else "partial" if ok > 0 else "failed" + # 一张都没起草出来才算失败（不出快照）；有没写出来、只写出一部分的表是 partial，快照照出、原因写在进度里 + wrote = total - len(self.failed) + status = "failed" if wrote <= 0 else "partial" if self.failed or self.partial else "succeeded" artifact = await put_json(content, kind=SNAPSHOT_KIND, meta={"origin": "shadow", "sources": ids}) \ if status != "failed" else None diff --git a/backend/tests/test_eval_snapshots.py b/backend/tests/test_eval_snapshots.py index 00291c3d..a6702b06 100644 --- a/backend/tests/test_eval_snapshots.py +++ b/backend/tests/test_eval_snapshots.py @@ -19,7 +19,7 @@ from app.data import catalog, catalog_draft, catalog_profile, catalog_view from app.data.catalog import make_item from app.data.catalog_view import CatalogView from app.db.base import SessionLocal -from app.db.models import CatalogNote +from app.db.models import CatalogNote, DataSource from app.db.models_eval import EvalCatalogSnapshot, EvalSuite from app.eval import snapshots from tests.fixtures import dual_check_shop as shop @@ -231,3 +231,37 @@ async def test_a_mock_provider_is_not_a_drafting_model(): row = (await session.execute(select(Provider).where(Provider.name == "演示接入_快照"))).scalar_one() await session.delete(row) await session.commit() + + +async def test_a_partly_failed_shadow_draft_still_snapshots_and_says_which_table(tmp_path, monkeypatch): + import sqlite3 + + from app.data import introspect + + path = tmp_path / "two.db" + with sqlite3.connect(path) as db: + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, amount REAL);" + "CREATE TABLE stores (id INTEGER PRIMARY KEY, name TEXT);") + async with SessionLocal() as session: + row = DataSource(name=f"two_{id(path) % 100000}", kind="sqlite", database=str(path)) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + sid = row.id + try: + # stores 那一段一直出错：起草引擎拆小重试之后仍没写出来 + monkeypatch.setattr(snapshots.ShadowDraft, "model_override", + FakeModel({"orders": {"label": "订单"}, "stores": {"label": "门店"}}, fail={"stores"})) + monkeypatch.setattr(catalog, "MODEL_RETRIES", 0) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) # 每张表一段：一段出错只影响那一张 + async with SessionLocal() as session: + plan = await snapshots.plan_shadow(session, {"sources": [{"id": sid, "tables": ["orders", "stores"]}]}) + snap = await snapshots.start_shadow(session, plan, actor=None) + done = await wait_done(snap.id) + assert done.status == "partial" and done.artifact_id, (done.status, done.error) + assert list(done.progress["failures"]) == [f"{row.name}.stores"] and done.progress["tables_failed"] == 1 + notes = await read_view(done.artifact_id, sid) + assert notes["orders"]["label"]["value"] == "订单" and "stores" not in notes + finally: + await drop_source(sid) diff --git a/backend/tests/test_evals_api.py b/backend/tests/test_evals_api.py index 9fc4e1c1..c4efa1f5 100644 --- a/backend/tests/test_evals_api.py +++ b/backend/tests/test_evals_api.py @@ -327,6 +327,21 @@ async def test_caliber_cases_list_and_dispose_upgrades(client, source): assert res.status_code == 409 and res.json()["code"] == "eval_dispose_nothing_pending" +async def test_caliber_cases_pin_only_published_versions(client, source): + async with SessionLocal() as session: + wf = Workflow(name=f"草稿口径_{uuid.uuid4().hex[:6]}", graph=_weekly(source["name"]), status="draft", version=1) + session.add(wf) + await session.flush() + session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=_weekly(source["name"]))) + await session.commit() + wf_id = wf.id + suite = await new_suite(client) + res = await client.post(f"/api/evals/suites/{suite['id']}/cases", json={ + "question": "q", "gold": {"kind": "caliber", "workflow_id": wf_id, "version": 1, "node_id": "card"}, + "expectations": [{"kind": "scalar", "metric": "gmv"}]}) + assert res.status_code == 422 and res.json()["field"] == "gold.version" and "没有发布过" in res.json()["detail"] + + # -------------------------------------------------------------------------- # 导出导入 # --------------------------------------------------------------------------