mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 01:51:57 +00:00
数据目录阶段 2B：运行时接上 SQL 检查——查询结果和快照带 checks、证据接口的查询步骤带上，契约指标的来源查询有 error 时出具降档
This commit is contained in:
ThearchyHelios
@@ -825,6 +825,9 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
|
||||
# 拿截断的查询结果整组算出来的：面板标「结果不完整」并写明原因。完整的指标没有这两个键
|
||||
**({"incomplete": True, "incomplete_reason": source.get("incomplete_reason") or ""}
|
||||
if source.get("incomplete") else {}),
|
||||
# 来源查询没通过 SQL 检查：面板标出来并写明原因（问题和改法在查询步骤的 checks 里）。没问题的没有这两个键
|
||||
**({"sql_check_failed": True, "sql_check_reason": source.get("sql_check_reason") or ""}
|
||||
if source.get("sql_check_failed") else {}),
|
||||
"node_id": entry.get("node_id"), "artifact": artifact or None, "eid": cite.get("eid"),
|
||||
"eid_ok": cite.get("eid") == entry.get("eid") == make_eid("metric", artifact or None, {"metric": metric_id}),
|
||||
"hash_ok": hash_ok and metric is not None,
|
||||
@@ -904,6 +907,8 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str
|
||||
hint（期 4，P4-SPEC 2.8.2）：调用方是带文档标记的单元格片段时为真。这时快照若是上传表格的（有 data_version），
|
||||
步骤多一个 `provenance: true`，界面据此才去请求推断来源接口；其余情况一个键都不加，形状和期 4 之前完全相同
|
||||
（期 4 之前的文档、手工源、指标链都不多发请求）。不看封存状态：未封存的运行照常推断，响应里照实标明。
|
||||
|
||||
checks：快照里记下的 SQL 检查结果（数据目录阶段 2B，见 _query_checks），只在查出问题时有这个键。
|
||||
"""
|
||||
catalog = doc.get("catalog") or {}
|
||||
alias = _query_alias(catalog, artifact)
|
||||
@@ -948,7 +953,8 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str
|
||||
masked=[columns[i] for i in sorted(masked)],
|
||||
# 查询时从驱动的原始值记下的列类型（老快照没有）：文本列里的 "2026" 按文本显示，不当数
|
||||
column_types=snap.get("column_types") if isinstance(snap.get("column_types"), dict) else {},
|
||||
**({"provenance": True} if hint and snap.get("data_version") else {}))
|
||||
**({"provenance": True} if hint and snap.get("data_version") else {}),
|
||||
**({"checks": checks} if (checks := _query_checks(snap)) else {}))
|
||||
if len(index) > MAX_WINDOW_ROWS:
|
||||
index, step["window_truncated"] = index[:MAX_WINDOW_ROWS], True
|
||||
step["row_index"] = index
|
||||
@@ -958,6 +964,19 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str
|
||||
return step
|
||||
|
||||
|
||||
#: 查询步骤里每条 SQL 检查结果交给界面的字段（data/sqlcheck.SqlCheck.as_dict 去掉 for_model）
|
||||
_CHECK_FIELDS = ("code", "level", "message", "table", "column", "relation_id", "sql_excerpt")
|
||||
|
||||
|
||||
def _query_checks(snap: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"""查询快照里记下的 SQL 检查结果（数据源查询工具执行后对照数据目录查的）。给模型的那句改法不上界面。
|
||||
快照里没有（没查出问题、或者是升级前的快照）返回空列表，步骤里就不加这个键。"""
|
||||
checks = snap.get("checks")
|
||||
if not isinstance(checks, list):
|
||||
return []
|
||||
return [{k: c[k] for k in _CHECK_FIELDS if k in c} for c in checks if isinstance(c, dict) and c.get("code")]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# GET /api/runs/{run_id}/evidence/segments/{segment_id}/provenance（期 4：推断的来源）
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
@@ -902,6 +902,9 @@ def build_catalog(
|
||||
# 只在不完整时有这两个键，完整的指标条目和以前一字不差
|
||||
**({"incomplete": True, "incomplete_reason": metric.get("incomplete_reason") or ""}
|
||||
if metric.get("incomplete") else {}),
|
||||
# 来源查询没通过 SQL 检查（data/sqlcheck.py 的 error 级问题）：同样只在有问题时有这两个键
|
||||
**({"sql_check_failed": True, "sql_check_reason": metric.get("sql_check_reason") or ""}
|
||||
if metric.get("sql_check_failed") else {}),
|
||||
}
|
||||
|
||||
queries = retrievals = 0
|
||||
@@ -3109,6 +3112,9 @@ def catalog_prompt(catalog: dict[str, Any], *, budget: int = 12000, cells_allowe
|
||||
# 不完整的指标照样能引用（单格、口径都对），但写作者得知道它不是全量：不写行数，免得照抄成裸数字
|
||||
caveat = "（基于被截断的查询结果计算，结果不完整：引用时须如实说明，不要写成全部数据）" \
|
||||
if e.get("incomplete") else ""
|
||||
# 取数的 SQL 没通过检查（重复计算、存量跨期加总）：数是算错的可能很大，写作者引用时得交代
|
||||
if e.get("sql_check_failed"):
|
||||
caveat += "（所依据的查询未通过 SQL 检查，结果存疑：引用时须如实说明，不要当作确定的结论）"
|
||||
lines.append(f"- [[{e['alias']}]] {e.get('name')} = {e.get('rendered')}{caveat}")
|
||||
inputs = [e for e in catalog.values() if e.get("kind") == "input"]
|
||||
if inputs:
|
||||
|
||||
@@ -462,6 +462,17 @@ def incomplete_gaps(metrics: list[dict[str, Any]]) -> list[str]:
|
||||
for m in metrics if isinstance(m, dict) and m.get("incomplete")]
|
||||
|
||||
|
||||
def sql_check_gaps(metrics: list[dict[str, Any]]) -> list[str]:
|
||||
"""契约收来的指标里来源查询没通过 SQL 检查的（口径卡标了 sql_check_failed：取数的那条 SQL 对照数据目录查出
|
||||
error 级问题，比如一对多关联之后对「一」那一侧求和），每个一句缺口。
|
||||
|
||||
和 incomplete_gaps 同一个道理：数字照样回指得上口径卡，口径卡照样复算一致，但算它的那条查询本身就把数算错了
|
||||
（重复计算、存量跨期加总），盖完整出具的章等于替一个错数作保。原因用口径卡记下的那句，写明是哪条检查。
|
||||
"""
|
||||
return [f"指标「{m.get('name') or m.get('id')}」{m.get('sql_check_reason') or '所依据的查询未通过 SQL 检查'}"
|
||||
for m in metrics if isinstance(m, dict) and m.get("sql_check_failed")]
|
||||
|
||||
|
||||
def _claims_effect(
|
||||
policy: str | dict[str, Any] | None,
|
||||
unsupported: list[Any] | int | None,
|
||||
|
||||
@@ -147,7 +147,7 @@ def _apply_contract(
|
||||
) -> dict[str, Any]:
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app.engine.issuance import decide_tier, incomplete_gaps, trace_numbers
|
||||
from app.engine.issuance import decide_tier, incomplete_gaps, sql_check_gaps, trace_numbers
|
||||
|
||||
nodes = state.get("nodes") or {}
|
||||
# 写了 report_from 就是引用模式：按报告文档逐段复核，不再按数值回指叙述
|
||||
@@ -188,6 +188,8 @@ def _apply_contract(
|
||||
gaps.append(f"指标来源未解析：{'、'.join(unresolved[:5])}")
|
||||
# 拿截断的查询结果整组算出来的指标：数回指得上，但算的只是取回的那一截，不能完整出具
|
||||
gaps.extend(incomplete_gaps(metrics))
|
||||
# 来源查询没通过 SQL 检查（error 级：重复计算、存量跨期加总）的指标：口径卡复算得一致，算的却是错数
|
||||
gaps.extend(sql_check_gaps(metrics))
|
||||
|
||||
# 引用模式下，缺输入记成空值的指标（口径卡 on_missing=null）就是缺：报告里引用不了它，
|
||||
# 也不能因为 id 在清单里就算「齐了」。旧模式照旧只看 id，行为不变
|
||||
|
||||
@@ -49,6 +49,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
|
||||
|
||||
表达式把截断的查询结果当成整组用了（len、sum、max……，见 expressions.TruncatedUse）的指标，
|
||||
值照算，但标 incomplete 并写明原因：它算的只是取回的那一截。出具契约据此降档（issuance.incomplete_gaps）。
|
||||
|
||||
取数的那条 SQL 对照数据目录查出 error 级问题的（查询快照的 checks，见 data/sqlcheck.py：一对多关联后重复计算、
|
||||
存量跨期加总），同样值照算，标 sql_check_failed 并写明原因，出具契约据此降档（issuance.sql_check_gaps）。
|
||||
"""
|
||||
definitions = ctx.cfg("metrics", []) or []
|
||||
caliber_raw, caliber_version = ctx.cfg("caliber", "") or ctx.node.title, str(ctx.cfg("caliber_version", "") or "v1")
|
||||
@@ -71,6 +74,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
|
||||
|
||||
metrics: list[dict[str, Any]] = []
|
||||
errors: list[str] = []
|
||||
#: 查询快照工件 id → 它的 error 级检查结果。一张卡里多个指标常常取自同一次查询，快照只读一次
|
||||
checked: dict[str, list[dict[str, Any]]] = {}
|
||||
for definition in definitions:
|
||||
metric_id = str(definition.get("id") or "").strip()
|
||||
expr = str(definition.get("expression") or "").strip()
|
||||
@@ -127,6 +132,12 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
|
||||
sources = list(use.sources.values())
|
||||
incomplete = {"incomplete": True, "incomplete_reason": _incomplete_reason(sources),
|
||||
"truncated_sources": sources}
|
||||
inputs = [_input_of(path, values[path], state, ctx, tctx) for path in paths]
|
||||
sql_check: dict[str, Any] = {}
|
||||
if failed := _failed_queries(inputs, list(use.sources.values()), paths, tctx, checked):
|
||||
sql_check = {"sql_check_failed": True, "sql_check_reason": _sql_check_reason(failed),
|
||||
"sql_check_sources": [{"artifact": artifact, "codes": list(dict.fromkeys(
|
||||
str(c.get("code")) for c in found))} for artifact, found in failed]}
|
||||
metrics.append(
|
||||
{
|
||||
"id": metric_id,
|
||||
@@ -137,7 +148,7 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
|
||||
"decimals": decimals,
|
||||
"format": display["format"],
|
||||
"rendered": _rendered(value, definition, display),
|
||||
"inputs": [_input_of(path, values[path], state, ctx, tctx) for path in paths],
|
||||
"inputs": inputs,
|
||||
"substituted": substituted,
|
||||
"recompute_ok": _recompute(substituted, tctx, value, decimals),
|
||||
# 值是空的就是缺输入，不论 on_missing 是哪一种：fail 模式下直接取到空值的
|
||||
@@ -145,6 +156,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
|
||||
"status": "ok" if value is not None else "missing_input",
|
||||
# 只在不完整时出现：完整的指标和以前一字不差，老运行复核时目录也对得上
|
||||
**incomplete,
|
||||
# 只在来源查询没通过 SQL 检查时出现，同上
|
||||
**sql_check,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -158,12 +171,16 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
|
||||
else f"- {m['id']}（{m['name']}）= {MISSING}（缺少输入，无值）")
|
||||
# 不写行数：叙述节点照抄进正文就是一个没有出处的数
|
||||
+ ("（不完整：基于被截断的查询结果计算）" if m.get("incomplete") else "")
|
||||
+ ("（存疑：所依据的查询未通过 SQL 检查）" if m.get("sql_check_failed") else "")
|
||||
for m in metrics
|
||||
]
|
||||
for m in metrics:
|
||||
if m.get("incomplete"):
|
||||
ctx.emit(EventType.LOG, level="warn", code="metric_incomplete", metric=m["id"],
|
||||
message=f"指标「{m['name']}」{m['incomplete_reason']}")
|
||||
if m.get("sql_check_failed"):
|
||||
ctx.emit(EventType.LOG, level="warn", code="metric_sql_check", metric=m["id"],
|
||||
message=f"指标「{m['name']}」{m['sql_check_reason']}")
|
||||
card = {
|
||||
"kind": "metric_set",
|
||||
"caliber": caliber,
|
||||
@@ -331,6 +348,72 @@ def _incomplete_reason(sources: list[dict[str, Any]]) -> str:
|
||||
return f"基于 {len(fetched)} 份被截断的查询结果计算（分别只取回了{'、'.join(fetched)}），结果不完整"
|
||||
|
||||
|
||||
#: 取值链的最后一段：.字段 或 [下标]
|
||||
_LAST_STEP = re.compile(r"(?:\.[^.\[\]]+|\[[^\[\]]*\])$")
|
||||
|
||||
|
||||
def _failed_queries(inputs: list[dict[str, Any]], truncated: list[dict[str, Any]], paths: list[str],
|
||||
tctx: dict[str, Any], checked: dict[str, list[dict[str, Any]]],
|
||||
) -> list[tuple[str, list[dict[str, Any]]]]:
|
||||
"""这个指标用到的查询里，SQL 检查有 error 的：[(查询快照工件 id, error 级检查结果)]。
|
||||
|
||||
用到了哪些查询，三处来历合起来认：输入的出处（cell() 取的格、Agent 经 cite_fields 交来的字段都带着快照 id）、
|
||||
截断跟踪记下的整组、取值链上的查询结果本身（数据整形解析过的 vars.res.rows[0][0]，往上找到 vars.res 那份
|
||||
带 artifact 的结果）。检查结果从快照里读，不信内存里那份：快照按哈希寻址，就是查询当时存下的那份。
|
||||
读不出来的快照不算（另有别的检查报快照的问题）。Agent 没开 cite_fields 时说不清数出自哪次查询，不算。
|
||||
"""
|
||||
artifacts = [item.get("artifact") for item in inputs] + [s.get("artifact") for s in truncated]
|
||||
artifacts += [_result_artifact(path, tctx) for path in paths if cell_parts(path) is None]
|
||||
out = []
|
||||
for artifact in dict.fromkeys(a for a in artifacts if isinstance(a, str) and a):
|
||||
if artifact not in checked:
|
||||
checked[artifact] = _error_checks(artifact)
|
||||
if checked[artifact]:
|
||||
out.append((artifact, checked[artifact]))
|
||||
return out
|
||||
|
||||
|
||||
def _result_artifact(path: str, tctx: dict[str, Any]) -> str | None:
|
||||
"""顺着取值链往上找（vars.res.rows[0][0] → vars.res.rows[0] → vars.res.rows → vars.res），第一个查询结果的快照 id。
|
||||
数据源工具交回的是 JSON 文本（调用工具节点的输出），解析过的是对象（数据整形之后），两种都认。"""
|
||||
current = path
|
||||
for _ in range(8):
|
||||
value = _value_of(current, tctx)
|
||||
if isinstance(value, str) and value.lstrip().startswith("{") and '"artifact"' in value:
|
||||
try:
|
||||
value = json.loads(value)
|
||||
except ValueError:
|
||||
value = None
|
||||
if isinstance(value, dict) and isinstance(value.get("artifact"), str) and isinstance(value.get("rows"), list):
|
||||
return value["artifact"]
|
||||
shorter = _LAST_STEP.sub("", current)
|
||||
if not shorter or shorter == current:
|
||||
return None
|
||||
current = shorter
|
||||
return None
|
||||
|
||||
|
||||
def _error_checks(artifact: str) -> list[dict[str, Any]]:
|
||||
try:
|
||||
snapshot = artifact_store.load(artifact)
|
||||
except Exception: # noqa: BLE001 - 快照读不出来、被改过：这里不报，证据层另有说法
|
||||
return []
|
||||
checks = snapshot.get("checks") if isinstance(snapshot, dict) else None
|
||||
return [c for c in checks if isinstance(c, dict) and c.get("level") == "error"] if isinstance(checks, list) else []
|
||||
|
||||
|
||||
def _sql_check_reason(failed: list[tuple[str, list[dict[str, Any]]]]) -> str:
|
||||
"""「所依据的查询未通过 SQL 检查（「订单」关联「订单明细」是一对多，对「订单」的「订单金额」求和会重复计算），
|
||||
结果不可靠」：出具声明、运行日志、证据面板都用这句。只摘问题那半句，改法留在查询步骤里看。"""
|
||||
problems = list(dict.fromkeys(str(c.get("message") or "").split("。")[0] for _, found in failed for c in found))
|
||||
problems = [p for p in problems if p]
|
||||
if not problems:
|
||||
return "所依据的查询未通过 SQL 检查，结果不可靠"
|
||||
if len(problems) == 1:
|
||||
return f"所依据的查询未通过 SQL 检查（{problems[0]}），结果不可靠"
|
||||
return f"所依据的查询有 {len(problems)} 处未通过 SQL 检查（{problems[0]}等），结果不可靠"
|
||||
|
||||
|
||||
def _truncated_field(path: str, state: GraphState, ctx: NodeContext) -> dict[str, Any] | None:
|
||||
"""agent 经 cite_fields 交来、被截断切开的数组字段：{rows, artifact}。别的取值链返回 None。
|
||||
|
||||
|
||||
@@ -33,6 +33,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from app.core.events import EventType
|
||||
from app.data import catalog as data_catalog
|
||||
from app.data import introspect, table_versions
|
||||
from app.data.sqlcheck import SqlChecker
|
||||
from app.data.engine import SnapshotTampered, masked_columns, query_timeout, run_query
|
||||
from app.data.guard import QueryLimits, SqlRejected, is_write
|
||||
from app.data.tabular import UNSHAPED_NOTE
|
||||
@@ -53,6 +54,10 @@ DATA_UNAVAILABLE = "查询失败（数据不可用）："
|
||||
#: 运行中途补固定一个源时发的那条日志事件的 code（_pin_late）
|
||||
LATE_PIN_CODE = "data_version_pinned"
|
||||
|
||||
#: 每次查询之后基于数据目录的 SQL 检查最多等这么久。正常一条 SQL 几毫秒；等不到就放弃这一次检查，
|
||||
#: 查询结果照常交回——检查是附加的，不能拖慢查询
|
||||
SQL_CHECK_TIMEOUT_S = 2.0
|
||||
|
||||
|
||||
class _QueryArgs(BaseModel):
|
||||
sql: str = Field(description="要执行的 SQL，一次一条")
|
||||
@@ -445,6 +450,15 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
|
||||
# 数据源在 options.query_timeout_s 里配了就用它，否则取缺省。数据库按它停下语句，
|
||||
# 引擎按它告诉界面上限是多少（metadata["timeout_s"]）
|
||||
seconds = query_timeout(source)
|
||||
# 基于数据目录的 SQL 检查器：和冻结进表结构快照的是同一版目录。第一次查询时才建（关系图也是用到才建）
|
||||
checker: list[SqlChecker] = []
|
||||
|
||||
def _checker() -> SqlChecker:
|
||||
if not checker:
|
||||
checker.append(SqlChecker(kind=source.kind, schema_cache=source.schema_cache,
|
||||
notes={name: entry.notes for name, entry in (catalog or {}).items()},
|
||||
source_name=source.name))
|
||||
return checker[0]
|
||||
|
||||
async def _run(sql: str, limit: int | None = None) -> str:
|
||||
if gone := _vanished(source, fixed):
|
||||
@@ -470,6 +484,11 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
|
||||
|
||||
payload = result.to_payload()
|
||||
payload["source"] = source.name
|
||||
# 对照数据目录检查这条 SQL（一对多关联后重复计算、存量跨期求和……）。结果交给模型（Agent 据此改写），
|
||||
# 也随查询快照存下：证据面板看得到，口径卡据此给指标标「存疑」、出具降档。没查出问题不加这个键，
|
||||
# 快照和以前一字不差
|
||||
if checks := await _sql_checks(_checker(), sql):
|
||||
payload["checks"] = checks
|
||||
# 表结构快照：这次查询时数据源的结构冻结下来，报告里写的表名、字段名按它核对。事后数据源
|
||||
# 重新探查、改了结构，已经跑完的运行核对的还是当时那一份。上传源冻结的是绑定快照里那份
|
||||
if schema := await _store_schema(source, ctx, catalog=catalog):
|
||||
@@ -511,6 +530,20 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
|
||||
)
|
||||
|
||||
|
||||
async def _sql_checks(checker: SqlChecker, sql: str) -> list[dict[str, Any]]:
|
||||
"""一条查询的 SQL 检查结果（as_dict 的形状）。在线程里跑、限时 SQL_CHECK_TIMEOUT_S：解析是纯计算，放在事件循环里
|
||||
会挡住同一进程里别的运行。出错、超时都返回空列表——检查不能让一次成功的查询变成失败。"""
|
||||
if not checker.enabled:
|
||||
return []
|
||||
try:
|
||||
return await asyncio.wait_for(asyncio.to_thread(checker.check_dicts, sql), timeout=SQL_CHECK_TIMEOUT_S)
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning("SQL 检查超过 %.1f 秒，本次跳过（数据源 %s）", SQL_CHECK_TIMEOUT_S, checker.source_name)
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("SQL 检查出错，本次跳过（数据源 %s）", checker.source_name)
|
||||
return []
|
||||
|
||||
|
||||
async def _store_schema(source: Any, ctx: ToolContext, *,
|
||||
catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> str | None:
|
||||
"""把数据源此刻的 schema_cache 存成 schema_snapshot 工件，返回工件 id。
|
||||
|
||||
@@ -0,0 +1,288 @@
|
||||
"""运行时的 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
|
||||
|
||||
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]
|
||||
|
||||
|
||||
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()
|
||||
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):
|
||||
source_id, name = await _source(scenic_db)
|
||||
yield name
|
||||
await engines.invalidate(source_id)
|
||||
|
||||
|
||||
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():
|
||||
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}", f"指标「订单金额（整形后）」{REASON}"]
|
||||
|
||||
issuance = row.output["_issuance"]
|
||||
assert issuance["tier"] == "degraded", issuance
|
||||
assert f"指标「订单金额」{REASON}" in issuance["gaps"]
|
||||
# 写作目录提醒写作者：这个数存疑，引用时要说明
|
||||
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"]
|
||||
Reference in New Issue
Block a user