mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 06:11:58 +00:00
ThearchyHelios
268 lines
14 KiB
Python
268 lines
14 KiB
Python
"""目录快照与影子起草（app/eval/snapshots.py；细案 01 做法 6）。
|
|
|
|
- 当前目录的快照：按 data/catalog_view.py 的读法读得出、和拍的那一刻一样；之后线上目录改了，快照不变（内容寻址、不可变）。
|
|
- 影子起草：同一个起草引擎、只起草指定（或题集用得到）的表，写进内存里的副本、出快照；线上目录一项不动（版本不变、
|
|
不调 write_entry）；不查业务库（开了剖析也不去数有效行过滤的排除比例）；不占数据源的队；可以先去掉模型起草、还没人
|
|
审过的项；停止的不出快照；服务重启后没结束的标中断。开始前给估算。
|
|
|
|
库、表、目录内容都是虚构的。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.core import artifact_store, job_runtime
|
|
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, DataSource
|
|
from app.db.models_eval import EvalCatalogSnapshot, EvalSuite
|
|
from app.eval import snapshots
|
|
from tests.fixtures import dual_check_shop as shop
|
|
from tests.fixtures.catalog.fake_model import FakeModel
|
|
from tests.fixtures.sources import drop_source
|
|
|
|
LIVE_NOTES = {
|
|
"label": make_item("门店订单", "human"), # 人工确认
|
|
"grain": make_item("每笔订单一行", "llm"), # 模型起草、没人审过
|
|
"columns": {"amount": {"label": make_item("实收金额", "llm", "confirmed")}}, # 模型起草、人工确认过
|
|
}
|
|
REPLY = {"orders": {"label": "影子订单", "grain": "影子：每单一行", "kind": "fact",
|
|
"valid_filter": {"condition": "status <> 'refunded'", "reason": "状态列有退款"},
|
|
"columns": [{"name": "status", "label": "订单状态", "meaning": "paid 已支付，refunded 已退款"}]}}
|
|
|
|
|
|
@pytest.fixture
|
|
async def source(tmp_path):
|
|
name, sid = await shop.add_source(shop.build(tmp_path / "shop.db"),
|
|
options={catalog_profile.PROFILE_OPTION: {"enabled": True}})
|
|
async with SessionLocal() as session:
|
|
await catalog.write_entry(session, sid, "orders", LIVE_NOTES, if_version=0, actor="审阅人")
|
|
yield {"name": name, "id": sid}
|
|
await drop_source(sid)
|
|
|
|
|
|
async def live_rows(source_id: str) -> list[tuple[str, int, dict]]:
|
|
async with SessionLocal() as session:
|
|
rows = (await session.execute(select(CatalogNote).where(CatalogNote.source_id == source_id))).scalars().all()
|
|
return sorted((r.table_name, r.version, r.notes) for r in rows)
|
|
|
|
|
|
async def read_view(snapshot_artifact: str, source_id: str) -> dict:
|
|
view = CatalogView(mode="snapshot", snapshot=snapshot_artifact)
|
|
async with SessionLocal() as session:
|
|
found = await catalog_view.read_many(session, view, [source_id])
|
|
return {t: e.notes for t, e in found.get(source_id, {}).items()}
|
|
|
|
|
|
async def wait_done(snap_id: str) -> EvalCatalogSnapshot:
|
|
await snapshots.manager.wait(snap_id, timeout=30)
|
|
async with SessionLocal() as session:
|
|
return await session.get(EvalCatalogSnapshot, snap_id)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 当前目录
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
async def test_a_live_snapshot_reads_back_through_the_catalog_view_and_never_changes(source):
|
|
async with SessionLocal() as session:
|
|
snap = await snapshots.take_live(session, [source["id"]], actor="评测人")
|
|
assert snap.status == "succeeded" and snap.origin == "live" and snap.created_by == "评测人"
|
|
assert snap.sources == [{"id": source["id"], "name": source["name"]}]
|
|
content = artifact_store.load(snap.artifact_id)
|
|
assert content["taken"]["origin"] == "live" and content["taken"]["by"] == "评测人"
|
|
assert artifact_store.content_hash(artifact_store.canonical_json(content)) == snap.artifact_id
|
|
assert (await read_view(snap.artifact_id, source["id"]))["orders"] == LIVE_NOTES
|
|
# 之后线上目录改了：快照还是拍的那一份
|
|
async with SessionLocal() as session:
|
|
await catalog.write_entry(session, source["id"], "orders", {"label": make_item("改过的", "human")},
|
|
if_version=1, actor="审阅人")
|
|
assert (await read_view(snap.artifact_id, source["id"]))["orders"]["label"]["value"] == "门店订单"
|
|
# 只用已核实的项：读的时候按视图筛，快照本身照原样
|
|
reviewed = CatalogView(mode="snapshot", statuses="reviewed", snapshot=snap.artifact_id)
|
|
async with SessionLocal() as session:
|
|
notes = (await catalog_view.read_many(session, reviewed, [source["id"]]))[source["id"]]["orders"].notes
|
|
assert "grain" not in notes and notes["label"]["value"] == "门店订单"
|
|
|
|
|
|
async def test_a_live_snapshot_needs_existing_sources():
|
|
async with SessionLocal() as session:
|
|
with pytest.raises(snapshots.SnapshotInvalid, match="不存在"):
|
|
await snapshots.take_live(session, ["nope"], actor=None)
|
|
with pytest.raises(snapshots.SnapshotInvalid, match="请选择"):
|
|
await snapshots.take_live(session, [], actor=None)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 影子起草
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
async def test_shadow_drafting_writes_only_the_copy(source, monkeypatch):
|
|
model = FakeModel(REPLY)
|
|
monkeypatch.setattr(snapshots.ShadowDraft, "model_override", model)
|
|
|
|
async def no_business_queries(*_a, **_k):
|
|
raise AssertionError("影子起草不该查业务库")
|
|
|
|
monkeypatch.setattr(catalog_profile, "check_valid_filter", no_business_queries)
|
|
before = await live_rows(source["id"])
|
|
async with SessionLocal() as session:
|
|
plan = await snapshots.plan_shadow(session, {"sources": [{"id": source["id"], "tables": ["orders"]}],
|
|
"guidance": "订单表按支付日统计"}, model_id="mock-fast")
|
|
snap = await snapshots.start_shadow(session, plan, actor="评测人")
|
|
assert snap.status in ("queued", "running")
|
|
# 不占数据源的队：同一个源照样能马上起草、剖析
|
|
await asyncio.sleep(0)
|
|
assert await job_runtime.lanes.holders(source["id"]) == []
|
|
done = await wait_done(snap.id)
|
|
assert done.status == "succeeded", (done.error, done.progress)
|
|
# 线上目录一项不动
|
|
assert await live_rows(source["id"]) == before
|
|
notes = (await read_view(done.artifact_id, source["id"]))["orders"]
|
|
assert notes["label"]["value"] == "门店订单" # 人工确认的不被起草覆盖
|
|
assert notes["grain"]["value"] == "影子：每单一行" # 没人审过的模型项换成这次起草的
|
|
assert notes["columns"]["status"]["label"]["value"] == "订单状态"
|
|
# 开了剖析也不去业务库数排除比例：照收，写明没有核对
|
|
assert catalog_draft.SHADOW_NOT_CHECKED in notes["valid_filter"]["note"]
|
|
# 起草要求交给了模型；只起草了点名的表
|
|
assert len(model.calls) == 1 and "订单表按支付日统计" in model.calls[0]
|
|
content = artifact_store.load(done.artifact_id)
|
|
assert content["taken"]["origin"] == "shadow" and content["taken"]["tables"] == {source["id"]: ["orders"]}
|
|
assert done.usage["calls"] == 1 and done.progress["tables_done"] == 1
|
|
assert done.estimate["tables"] == 1 and done.estimate["calls"] == 1
|
|
|
|
|
|
async def test_shadow_drafting_can_start_without_unreviewed_model_items(source, monkeypatch):
|
|
monkeypatch.setattr(snapshots.ShadowDraft, "model_override", FakeModel({"orders": {"kind": "fact"}}))
|
|
async with SessionLocal() as session:
|
|
plan = await snapshots.plan_shadow(session, {"sources": [{"id": source["id"], "tables": ["orders"]}],
|
|
"strip_llm": True})
|
|
snap = await snapshots.start_shadow(session, plan, actor=None)
|
|
done = await wait_done(snap.id)
|
|
notes = (await read_view(done.artifact_id, source["id"]))["orders"]
|
|
assert "grain" not in notes # 没人审过的模型项去掉了
|
|
assert notes["columns"]["amount"]["label"]["value"] == "实收金额" # 人工确认过的模型项留着
|
|
assert notes["label"]["value"] == "门店订单" and notes["kind"]["value"] == "fact"
|
|
|
|
|
|
async def test_tables_come_from_the_suite_when_not_listed(source):
|
|
from app.db.models_eval import EvalCase
|
|
|
|
async with SessionLocal() as session:
|
|
suite = EvalSuite(name="虚构题集")
|
|
session.add(suite)
|
|
await session.flush()
|
|
session.add(EvalCase(suite_id=suite.id, seq=1, question="q", expectations=[], gold={
|
|
"kind": "sql", "datasource_id": source["id"],
|
|
"sql": "SELECT SUM(amount) FROM orders WHERE paid_date = '{{ schedule.yesterday }}'"}))
|
|
await session.commit()
|
|
found = await snapshots.suite_tables(session, suite.id)
|
|
assert found == {source["id"]: ["orders"]}
|
|
plan = await snapshots.plan_shadow(session, {"sources": [{"id": source["id"]}], "suite_id": suite.id})
|
|
assert plan.sources[0]["tables"] == ["orders"]
|
|
with pytest.raises(snapshots.SnapshotInvalid, match="没有指定要起草的表"):
|
|
await snapshots.plan_shadow(session, {"sources": [{"id": source["id"]}]})
|
|
with pytest.raises(snapshots.SnapshotInvalid, match="表结构里没有"):
|
|
await snapshots.plan_shadow(session, {"sources": [{"id": source["id"], "tables": ["ghost"]}]})
|
|
|
|
|
|
def test_the_estimate_is_said_to_be_an_estimate():
|
|
from types import SimpleNamespace
|
|
|
|
cols = lambda n: [{"name": f"c{i}", "type": "TEXT"} for i in range(n)] # noqa: E731
|
|
src = SimpleNamespace(schema_cache={"tables": {"a": {"columns": cols(10)}, "b": {"columns": cols(10)},
|
|
"wide": {"columns": cols(100)}}})
|
|
est = snapshots.estimate([(src, ["a", "b", "wide"])], "no-such-model")
|
|
# 窄表拼一段；宽表是表级一段 + 列分段（100 列按 40 列一段是 3 段）
|
|
assert est["calls"] == 1 + (1 + 3) and est["tables"] == 3 and est["columns"] == 120
|
|
assert est["unpriced"] is True and est["cost_usd"] is None and "不含思考" in est["note"]
|
|
assert est["output_tokens"] == 3 * snapshots.EST_OUTPUT_PER_TABLE + 120 * snapshots.EST_OUTPUT_PER_COLUMN
|
|
|
|
|
|
async def test_a_stopped_shadow_draft_leaves_no_snapshot(source, monkeypatch):
|
|
monkeypatch.setattr(snapshots.ShadowDraft, "model_override", FakeModel(REPLY, hang=5))
|
|
async with SessionLocal() as session:
|
|
plan = await snapshots.plan_shadow(session, {"sources": [{"id": source["id"], "tables": ["orders"]}]})
|
|
snap = await snapshots.start_shadow(session, plan, actor=None)
|
|
for _ in range(100):
|
|
await asyncio.sleep(0.02)
|
|
async with SessionLocal() as session:
|
|
if (await session.get(EvalCatalogSnapshot, snap.id)).status == "running":
|
|
break
|
|
assert await snapshots.stop(snap.id)
|
|
done = await wait_done(snap.id)
|
|
assert done.status == "cancelled" and done.artifact_id is None and done.error == snapshots.STOPPED
|
|
assert not await snapshots.stop(snap.id)
|
|
|
|
|
|
async def test_unfinished_shadow_drafts_are_marked_interrupted_after_a_restart(source):
|
|
async with SessionLocal() as session:
|
|
row = EvalCatalogSnapshot(origin="shadow", status="running", sources=[{"id": source["id"], "name": "x"}])
|
|
session.add(row)
|
|
await session.commit()
|
|
await snapshots.reconcile_stale(session, [row])
|
|
assert row.status == "interrupted" and row.error == snapshots.INTERRUPTED
|
|
other = EvalCatalogSnapshot(origin="shadow", status="queued", sources=[])
|
|
session.add(other)
|
|
await session.commit()
|
|
other_id = other.id
|
|
assert await snapshots.startup() >= 1
|
|
async with SessionLocal() as session:
|
|
assert (await session.get(EvalCatalogSnapshot, other_id)).status == "interrupted"
|
|
|
|
|
|
async def test_a_mock_provider_is_not_a_drafting_model():
|
|
from app.db.models import Provider
|
|
|
|
async with SessionLocal() as session:
|
|
session.add(Provider(name="演示接入_快照", kind="mock", default_model="mock-fast", enabled=True))
|
|
await session.commit()
|
|
try:
|
|
with pytest.raises(snapshots.SnapshotInvalid, match="演示用的模型接入"):
|
|
await snapshots.resolve_model(session, "演示接入_快照", None)
|
|
finally:
|
|
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)
|