yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 02:31:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

评测题目的几处收紧：口径题只能固定已发布未退役的版本、处置升版后按新版本再核一遍、试算前结束读事务、影子起草分开记只写出一部分的表

Browse Source 
- cases.resolve_case：口径题钉的版本没发布过或已退役的，保存时照实拒收（422，field=gold.version）；以前要等试算跑到
  升版判断才说得出。
- 处置升版（recompute / incomparable）改钉新版本之前，按新版本再过一遍 resolve_case（入口字段、指标），说得清是哪一项不行。
- 接口里的试算要等一次运行跑完（最多两分钟）：先结束这个请求的读事务再等，不一直开着它（导入时同样）。
- 影子起草：只写出一部分的表另记（tables_partial），快照照出、状态 partial；一张都没起草出来才算失败、不出快照。以前
  只写出一部分的表算成没起草出来，几张表都只写出一部分时整次记失败、写好的部分也丢了。
- 题集里推起草哪些表时，期望 SQL 按题集的时区渲染。
This commit is contained in:
ThearchyHelios
2026-10-16 14:05:22 +08:00
parent a7e91b16a1
commit 7f14313416
6 changed files with 87 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/evals.py
+18 -9
View File
@@ -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))
backend/app/eval/cases.py
+4
View File
@@ -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 []):
backend/app/eval/gold.py
+1 -1
View File
@@ -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}
backend/app/eval/snapshots.py
+14 -7
View File
@@ -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
backend/tests/test_eval_snapshots.py
+35 -1
View File
@@ -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)
backend/tests/test_evals_api.py
+15
View File
@@ -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"]
# --------------------------------------------------------------------------
# 导出导入
# --------------------------------------------------------------------------