mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 20:01:57 +00:00
ThearchyHelios
303 lines
14 KiB
Python
303 lines
14 KiB
Python
"""运行时的 SQL 检查：数据源查询工具每次执行成功后对照数据目录检查一遍。
|
||
|
||
- 结果写进查询结果（模型看得到，Agent 能据此改写 SQL）和查询快照的 checks 字段；没查出问题时不加这个键，
|
||
快照和以前一字不差。检查出错、超时都不影响查询本身。
|
||
- 证据接口的查询步骤带上 checks（不带给模型的那句 for_model）。
|
||
- 契约指标的来源查询有 error 级问题：口径卡把这个指标标 sql_check_failed 并写明原因，出具按 gaps 降档；
|
||
写作目录提醒写作者，证据面板的指标步骤标出来。和阶段 0 的「结果不完整」走同一条路。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import time
|
||
import uuid
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
from langchain_core.messages import AIMessage
|
||
from sqlalchemy import select
|
||
|
||
from app.core import artifact_store
|
||
from app.data import catalog
|
||
from app.data.engine import engines
|
||
from app.data.introspect import introspect
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import DataSource, Run, RunEvent
|
||
from app.engine.evidence import iter_segments
|
||
from app.engine.issuance import decide_tier, sql_check_gaps
|
||
from app.engine.runner import run_manager
|
||
from app.main import app
|
||
from app.tools.datasource import build_datasource_tools
|
||
from app.tools.registry import ToolContext
|
||
from tests.fixtures.catalog import scenic_notes
|
||
from tests.fixtures.sources import drop_source
|
||
|
||
FANOUT_SQL = ("SELECT SUM(o.total_amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id "
|
||
"WHERE o.status = 1")
|
||
CLEAN_SQL = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1"
|
||
REASON = ("所依据的查询未通过 SQL 检查（「订单」关联「订单明细」是一对多，对「订单」的「订单金额」求和会重复计算），"
|
||
"结果不可靠")
|
||
|
||
_CACHE: dict[str, dict] = {}
|
||
|
||
|
||
async def _schema(path: str) -> dict:
|
||
if path not in _CACHE:
|
||
probe = SimpleNamespace(id=f"probe-{uuid.uuid4().hex[:6]}", name="scenic", kind="sqlite", database=path,
|
||
host=None, port=None, username=None, password=None, options={}, readonly=True,
|
||
description="", schema_cache={}, origin="manual")
|
||
try:
|
||
_CACHE[path] = await introspect(probe)
|
||
finally:
|
||
await engines.invalidate(probe.id)
|
||
return _CACHE[path]
|
||
|
||
|
||
_CREATED: list[str] = []
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def _drop_created_sources():
|
||
yield
|
||
while _CREATED:
|
||
await drop_source(_CREATED.pop())
|
||
|
||
|
||
async def _source(path: str, *, with_catalog: bool = True) -> tuple[str, str]:
|
||
name = f"scenic_{uuid.uuid4().hex[:8]}"
|
||
async with SessionLocal() as session:
|
||
row = DataSource(name=name, kind="sqlite", database=path, readonly=True, enabled=True)
|
||
row.schema_cache = await _schema(path)
|
||
session.add(row)
|
||
await session.commit()
|
||
_CREATED.append(row.id)
|
||
if with_catalog:
|
||
for table, notes in scenic_notes.notes().items():
|
||
await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏")
|
||
return row.id, name
|
||
|
||
|
||
@pytest.fixture
|
||
async def scenic_source(scenic_db):
|
||
_, name = await _source(scenic_db)
|
||
yield name
|
||
|
||
|
||
async def _query_tool(name: str):
|
||
ctx = ToolContext(run_id=uuid.uuid4().hex, node_id="n1", data_versions=None)
|
||
async with SessionLocal() as session:
|
||
[tool] = await build_datasource_tools([f"db_query__{name}"], ctx, session)
|
||
return tool
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 数据源查询工具
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_query_result_and_snapshot_carry_checks(scenic_source):
|
||
tool = await _query_tool(scenic_source)
|
||
result = json.loads(await tool.coroutine(sql=FANOUT_SQL))
|
||
assert result["rows"] and result["artifact"]
|
||
[check] = result["checks"]
|
||
assert check["code"] == "fanout_sum" and check["level"] == "error" and check["table"] == "orders"
|
||
assert check["message"].startswith("「订单」关联「订单明细」是一对多") and "order_items" in check["for_model"]
|
||
assert artifact_store.load(result["artifact"])["checks"] == result["checks"]
|
||
|
||
clean = json.loads(await tool.coroutine(sql=CLEAN_SQL))
|
||
assert "checks" not in clean and "checks" not in artifact_store.load(clean["artifact"])
|
||
|
||
|
||
async def test_sources_without_catalog_are_not_checked(scenic_db):
|
||
source_id, name = await _source(scenic_db, with_catalog=False)
|
||
try:
|
||
result = json.loads(await (await _query_tool(name)).coroutine(sql=FANOUT_SQL))
|
||
assert result["rows"] and "checks" not in result
|
||
finally:
|
||
await engines.invalidate(source_id)
|
||
|
||
|
||
async def test_a_failing_or_slow_check_never_breaks_the_query(scenic_source, monkeypatch):
|
||
from app.data import sqlcheck
|
||
from app.tools import datasource
|
||
|
||
tool = await _query_tool(scenic_source)
|
||
|
||
def boom(self, sql):
|
||
raise RuntimeError("规则缺陷")
|
||
|
||
monkeypatch.setattr(sqlcheck.SqlChecker, "check", boom)
|
||
result = json.loads(await tool.coroutine(sql=FANOUT_SQL))
|
||
assert result["rows"] and "checks" not in result
|
||
|
||
def slow(self, sql):
|
||
time.sleep(1.0)
|
||
return []
|
||
|
||
monkeypatch.setattr(sqlcheck.SqlChecker, "check", slow)
|
||
monkeypatch.setattr(datasource, "SQL_CHECK_TIMEOUT_S", 0.05)
|
||
started = time.monotonic()
|
||
result = json.loads(await tool.coroutine(sql=FANOUT_SQL))
|
||
assert result["rows"] and "checks" not in result
|
||
assert time.monotonic() - started < 0.9, "检查超时就放弃，不等它跑完"
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 出具：契约指标的来源查询有 error 级问题时降档
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def test_sql_check_failures_become_readable_gaps():
|
||
# 没记来源快照（或快照读不出来）时退回口径卡那句；按查询归并的写法见 test_report_sqlcheck_cells
|
||
metrics = [{"id": "gmv", "name": "订单金额", "value": 1, "sql_check_failed": True, "sql_check_reason": REASON},
|
||
{"id": "n", "name": "订单数", "value": 2}]
|
||
assert sql_check_gaps(metrics) == [f"指标「订单金额」{REASON}"]
|
||
assert sql_check_gaps(metrics[1:]) == []
|
||
tier = decide_tier(missing_required=[], missing_expected=[], unmatched=[], strict=False,
|
||
gaps=sql_check_gaps(metrics))
|
||
assert tier == "degraded"
|
||
|
||
|
||
@pytest.fixture
|
||
async def engine_up():
|
||
await run_manager.setup()
|
||
yield
|
||
await run_manager.shutdown()
|
||
|
||
|
||
def node(nid, ntype, **config):
|
||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||
|
||
|
||
def chain(nodes: list[dict]) -> dict:
|
||
return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
|
||
|
||
|
||
GMV = {"id": "gmv", "name": "订单金额", "unit": "元", "format": "plain", "expression": "cell(nodes.fetch, 0, 'gmv')"}
|
||
#: 同一份结果经数据整形解析后按下标取：来历记成 transform，要顺着取值链找到那份查询结果
|
||
GMV2 = {"id": "gmv2", "name": "订单金额（整形后）", "unit": "元", "format": "plain",
|
||
"expression": "vars.res.rows[0][0]"}
|
||
WEEKLY = "本周订单金额 [[m:gmv]]。"
|
||
|
||
|
||
def weekly(source: str, sql: str) -> dict:
|
||
return chain([
|
||
node("start", "input"),
|
||
node("fetch", "tool", tool=f"db_query__{source}", args={"sql": sql}),
|
||
node("parse", "transform", mode="json", template="{{ nodes.fetch }}", assign_to="res"),
|
||
node("card", "metrics", caliber="订单口径", caliber_version="v1", metrics=[GMV, GMV2]),
|
||
node("write", "report", instructions="写订单周报"),
|
||
node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
|
||
contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"]}),
|
||
])
|
||
|
||
|
||
class Writer:
|
||
"""报告写作者：每次都照 text 写；记下看到的提示。"""
|
||
|
||
def __init__(self, monkeypatch, text: str = WEEKLY):
|
||
from app.providers import mock_model
|
||
|
||
self.prompts: list[str] = []
|
||
|
||
def decide(model, messages):
|
||
self.prompts.append("\n".join(str(m.content) for m in messages))
|
||
return AIMessage(content=text)
|
||
|
||
monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
|
||
|
||
|
||
async def wait(run_id: str) -> Run:
|
||
for _ in range(400):
|
||
await asyncio.sleep(0.05)
|
||
async with SessionLocal() as session:
|
||
row = await session.get(Run, run_id)
|
||
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
|
||
return row
|
||
raise AssertionError(f"run {run_id} 没有结束：{row.status}")
|
||
|
||
|
||
async def events(run_id: str, etype: str, node_id: str | None = None) -> list[RunEvent]:
|
||
async with SessionLocal() as session:
|
||
q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype)
|
||
if node_id:
|
||
q = q.where(RunEvent.node_id == node_id)
|
||
return list((await session.execute(q.order_by(RunEvent.seq))).scalars())
|
||
|
||
|
||
async def output_of(run_id: str, node_id: str) -> dict:
|
||
[finished] = await events(run_id, "node.finished", node_id)
|
||
return artifact_store.load(finished.data["artifact"])
|
||
|
||
|
||
async def test_a_fanout_behind_a_contract_metric_degrades_the_issuance(scenic_source, engine_up, monkeypatch):
|
||
writer = Writer(monkeypatch)
|
||
run = await run_manager.start(graph=weekly(scenic_source, FANOUT_SQL), input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
|
||
card = await output_of(row.id, "card")
|
||
metrics = {m["id"]: m for m in card["metrics"]}
|
||
[query_end] = [e.data for e in await events(row.id, "tool.end", "fetch")]
|
||
for mid in ("gmv", "gmv2"):
|
||
assert metrics[mid]["sql_check_failed"] is True and metrics[mid]["sql_check_reason"] == REASON
|
||
assert metrics[mid]["sql_check_sources"] == [{"artifact": query_end["query_artifact"],
|
||
"codes": ["fanout_sum"]}]
|
||
assert "（存疑：所依据的查询未通过 SQL 检查）" in card["text"]
|
||
warns = [e.data for e in await events(row.id, "log", "card") if e.data.get("code") == "metric_sql_check"]
|
||
# 同一条查询算出的两个指标：时间线上一行，降档原因里一句（按查询归并，列出受影响的指标）
|
||
assert [w["message"] for w in warns] == [f"指标「订单金额」「订单金额（整形后）」{REASON}"]
|
||
|
||
issuance = row.output["_issuance"]
|
||
assert issuance["tier"] == "degraded", issuance
|
||
assert [g for g in issuance["gaps"] if "SQL 检查" in g] == [
|
||
f"查询「fetch」（Q1）{REASON.removeprefix('所依据的查询')}；受影响的引用：指标「订单金额」、指标「订单金额（整形后）」"]
|
||
# 写作目录提醒写作者：这个数存疑，引用时要说明
|
||
assert "所依据的查询未通过 SQL 检查，结果存疑" in writer.prompts[0]
|
||
|
||
# 证据面板：指标步骤标出来，查询步骤带着检查结果（不带给模型的那句）
|
||
doc = artifact_store.load(row.output["_evidence"]["doc_artifact"])
|
||
assert doc["catalog"]["m:gmv"]["sql_check_failed"] is True
|
||
seg = next(s for s in iter_segments(doc) if s.get("ref") == "m:gmv")
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json()
|
||
metric = body["chain"][0]
|
||
assert metric["step"] == "metric" and metric["sql_check_failed"] is True
|
||
assert metric["sql_check_reason"] == REASON
|
||
[query] = [s for s in body["chain"] if s["step"] == "query"]
|
||
[check] = query["checks"]
|
||
assert check["code"] == "fanout_sum" and check["level"] == "error" and check["table"] == "orders"
|
||
assert check["message"].startswith("「订单」关联「订单明细」是一对多") and "for_model" not in check
|
||
|
||
|
||
async def test_the_same_card_on_a_clean_query_issues_formally(scenic_source, engine_up, monkeypatch):
|
||
Writer(monkeypatch)
|
||
run = await run_manager.start(graph=weekly(scenic_source, CLEAN_SQL), input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
card = await output_of(row.id, "card")
|
||
assert all("sql_check_failed" not in m for m in card["metrics"])
|
||
issuance = row.output["_issuance"]
|
||
assert issuance["tier"] == "formal" and issuance["gaps"] == [], issuance
|
||
doc = artifact_store.load(row.output["_evidence"]["doc_artifact"])
|
||
seg = next(s for s in iter_segments(doc) if s.get("ref") == "m:gmv")
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json()
|
||
assert "sql_check_failed" not in body["chain"][0]
|
||
assert all("checks" not in s for s in body["chain"] if s["step"] == "query")
|
||
|
||
|
||
async def test_warnings_alone_do_not_degrade(scenic_source, engine_up, monkeypatch):
|
||
"""只有 warning（这里是没按订单状态筛）：查询结果和快照里照样带着，但不影响出具档位。"""
|
||
Writer(monkeypatch)
|
||
sql = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id"
|
||
run = await run_manager.start(graph=weekly(scenic_source, sql), input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
[query_end] = [e.data for e in await events(row.id, "tool.end", "fetch")]
|
||
snapshot = artifact_store.load(query_end["query_artifact"])
|
||
assert [c["code"] for c in snapshot["checks"]] == ["missing_valid_filter"]
|
||
assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]
|