From 1491b5c7f389f31933e29efba80a18ecab8e054b Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 15:54:03 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=8C=E7=9B=B2=E5=A4=8D=E7=AE=97=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=E6=95=B4=E6=94=B9=EF=BC=9A=E6=8C=87=E6=A0=87=E8=87=AA?= =?UTF-8?q?=E5=B7=B1=E7=9A=84=E7=AD=9B=E9=80=89=E8=8C=83=E5=9B=B4=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E5=86=99=E6=88=90=E6=9C=89=E6=95=88=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E6=9D=A1=E4=BB=B6=EF=BC=8C=E7=9B=B2=E7=AE=97=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E7=9C=8B=E5=88=B0=E5=8E=9F=E6=B5=81=E7=A8=8B=E7=9A=84=E5=88=97?= =?UTF-8?q?=E5=90=8D=EF=BC=8C=E5=B7=AE=E5=BC=82=E8=AF=B4=E6=98=8E=E6=94=B6?= =?UTF-8?q?=E7=AA=84=E5=88=B0=E8=BF=99=E4=B8=80=E6=A0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 写数据目录（dual_check_facts.py） - 有效记录条件只收「排除无效记录」的条件：被排除的取值是退款、作废、测试这类（或码值表里写着这类意思）， 标记列只认留下有效记录的一边（is_test = 0、is_valid = 1、deleted_at IS NULL）；取值出现在指标名称、口径说明、 所在分组、运行输入里的，用到分组列的，一律不写，原因写明「可能是指标自己的筛选范围」。以前「华东销售额」的 region = '华东' 会被记成订单表人工确认过的有效记录条件，之后所有查订单表的工作流（含下一次盲算）都只看华东 - 目录里已有的有效记录条件被采纳的一边也按它筛了的，和新条件合起来写（已确认的不改，照实说）；没按它筛的不动。 以前整项替换，两边都用着的 is_test = 0 会被静默丢掉。推断的去重规则说法不同的也不替换 - 业务日期只在两边都按时间筛、按的列不同时才写；一边根本没按时间筛的不写 - 采纳了不去重的一边不写去重规则：那是指标自己的算法，写成「不去重」会把别的指标带偏 - 同一次裁决里几个指标对同一项给出矛盾的说法，整组都不写 节点（nodes/dual_check.py） - 盲算任务里不再给原流程结果的列名和同一行的维度列名：分组的只给取值（日期说成「2026-09-02 这一天」）， 原 SQL 写死的文字列连取值也不给；只有几项输入都来自查询又没写口径说明、列名看不出取舍时才给列名。 以前「order_date 为 2026-09-02」等于告诉盲算按下单日归属，下单日和支付日的陷阱正好漏掉 - 差异说明先把两条 SQL 收窄到这一格（只留算出这一列的那一项，分组的按这一行的分组值改成等值条件）再比， 多列查询、分组取一行时不再混进「选出的列」「分组」这类噪声，只差一个方面的也写得进目录 - 产出状态：全部一致才是 agreed，一个都没核成是 unchecked，核成一部分是 partial；产出和台账加 counts； 无法复算的指标发 dual_check_unchecked {metrics: [{id, name, reason}]} - 共用一格的几个指标触顶时每个指标各发一条 dual_check_limit - 查询次数刚好用完、交回的格子是真查出来的照常比对（时长、金额触顶仍按无法复算） - 审核人没选或选了认不出的照实记「审核人没有对指标作出选择」，不再替人记成「两边都不对」 - 盲算先单独做第一个，其余最多 3 个同时进行，避免串行撞上运行单段的时长上限 配置 - 画布校验：双盲复算配了失败重试报错（重试会把盲算整轮重做、再问审核人一次）；enabled 用 is 判断，0、1 不再放过 - 发起时的核查设置快照：子工作流（按运行时的取法读）里有双盲复算、反方审稿节点时也快照 - 助手节点说明补一句不要给双盲复算配 retries 测试：补上遮罩两道防线、规范化后一致但值不同、两个输入都对不上、只复算配置的指标、原值为 0、摘要最多 3 条、 驳回、多列查询、分组取一行、指标筛选范围不写目录等用例；每条核心改动都做过去掉实现、测试变红的验证 --- backend/app/api/copilot.py | 3 +- backend/app/engine/dual_check_facts.py | 433 ++++++++++++++++++--- backend/app/engine/launch.py | 42 ++- backend/app/engine/nodes/dual_check.py | 438 ++++++++++++++++++---- backend/app/engine/schema.py | 19 +- backend/tests/fixtures/dual_check_shop.py | 38 +- backend/tests/test_dual_check_config.py | 49 ++- backend/tests/test_dual_check_executor.py | 49 +++ backend/tests/test_dual_check_facts.py | 202 +++++++++- backend/tests/test_dual_check_node.py | 321 +++++++++++++++- 10 files changed, 1404 insertions(+), 190 deletions(-) diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index fdb042d6..32994055 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -224,7 +224,8 @@ NODE_REFERENCE = """\ 全部指标）；definitions 是 {指标 id: 口径说明}，写清业务定义（按哪个日期归属、排除哪些行、是否去重），盲算 agent 只看得到它和数据目录，所以要按业务写，不要抄原 SQL；provider / model 指定盲算模型（**默认不写**，取设置里的核查 模型；建议和取数的 agent 不同）；tolerance 是允许的相对误差（不写取设置）；on_reject：degrade（默认，计入缺口、 - 出具降档）/ fail（审核人没采用原流程的结果就判为失败）；enabled 不写跟随设置里的开关，true / false 强制开关 + 出具降档）/ fail（审核人没采用原流程的结果就判为失败）；enabled 不写跟随设置里的开关，true / false 强制开关。 + 不要给它配 retries：它的失败是确定的，重试只会把盲算再做一遍、再请审核人选一次（画布校验会报错） - report：报告撰写（带引用）。config: {instructions, system, metrics_from:[口径卡id], numbers, on_violation, max_repairs, claims, entities, assign_to} **凡是给人看的、带数字的报告、分析、结论，都用 report，不要用 llm**。它自动收集上游的口径卡指标、查询结果 diff --git a/backend/app/engine/dual_check_facts.py b/backend/app/engine/dual_check_facts.py index 5cbd3f34..c1ad8ae1 100644 --- a/backend/app/engine/dual_check_facts.py +++ b/backend/app/engine/dual_check_facts.py @@ -3,24 +3,37 @@ 对得上的事实只有三种（data/catalog.py 的表级项）： -- 两边按不同的时间列筛时间 → 业务日期（business_date）：被采纳的一边按哪一列筛，业务日期就记哪一列 -- 一边多一个只涉及底层表列的筛选条件 → 有效记录条件（valid_filter）：被采纳的一边有这个条件才记，记的就是这个条件 -- 聚合只差去不去重 → 去重规则（dedup）：按被采纳那一边去不去重写一句 +- 两边按不同的时间列筛时间 → 业务日期（business_date）：被采纳的一边按哪一列筛，业务日期就记哪一列。只有一边按时间 + 筛的不算：审核人判断的是要不要限定时间，没在两列之间选过 +- 一边多一个排除无效记录的条件 → 有效记录条件（valid_filter）：被采纳的一边有这个条件才记，记的就是这个条件 +- 聚合只差去不去重、被采纳的一边去重 → 去重规则（dedup）。采纳了不去重的一边不记：那是这个指标自己的算法（数行）， + 写成表的「不去重」会把别的指标带偏 -**为什么要这么严。** 目录是给助手、db_schema 工具和 SQL 检查看的数据事实，人工确认过的项还会让 SQL 检查报错误级。 -只差一个方面时，人选的那一边就是对这一处的判断；差了好几处、或者差异说明里有兜底（未能归纳的差异），人选的是 -「整体上哪边对」，拆不出是哪一处对，写进去就是替人编了一条事实。两条 SQL 规范化后一致却算出不同的值（两次取数之间 -数据变了、结果被截断），差在数据不在写法，更没有可记的事实。 +**为什么要这么严。** 目录是给助手、db_schema 工具、SQL 检查和下一次盲算看的数据事实，人工确认过的项还会让 SQL 检查 +报错误级、以后的推断也不再覆盖它，写错了就一直错下去。只差一个方面时，人选的那一边就是对这一处的判断；差了好几处、 +或者差异说明里有兜底（未能归纳的差异），人选的是「整体上哪边对」，拆不出是哪一处对，写进去就是替人编了一条事实。 +两条 SQL 规范化后一致却算出不同的值（两次取数之间数据变了、结果被截断），差在数据不在写法，更没有可记的事实。 + +**有效记录条件和指标自己的筛选范围分不开。** 「华东销售额」的 region = '华东'、「线上订单数」的 channel = 'online' +在写法上和 status <> 'refunded' 一样是一个筛选条件，审核人选原流程只说明这个指标该这么筛，不说明整张表的有效记录 +都得这么筛。写成表的有效记录条件，所有查这张表的工作流（包括下一次盲算）都会被带偏。所以条件要同时满足三条才写： +条件里的取值没有出现在指标名称、口径说明、这一格所在的分组、本次运行的输入里，用到的列也不是分组的列；每一段都是在 +排除无效记录（排除的取值是退款、作废、测试这类，或者码值表里写着这类意思；标记列只认留下有效记录的那一边： +is_test = 0、is_valid = 1、deleted_at IS NULL）；只涉及一张表。拿不准就不写，原因照实交回。 **怎么写。** 照对话提案的规矩核对（plan_patch：列在不在表结构里、取值格式），记为人工填写、已确认，备注写明来自哪次 运行、哪个节点、哪个指标。目录里这一项已经是人工确认过的另一种说法时不覆盖，照实交回原因；已经是同样的说法就什么都 -不写（再写一遍也不升版本）。整个写入由节点放在 ask() 之后的一个 @task 里：审批恢复重放、接着运行时不会再写一遍。 +不写（再写一遍也不升版本）。有效记录条件是一整句条件：目录里已有的条件被采纳的一边也按它筛了的，和新条件合起来写 +（已确认的不改，照实说）；被采纳的一边没按它筛的，两者的关系拿不准，不动。推断的去重规则说的未必是同一件事，也不 +替换。同一次裁决里几个指标对同一项给出不同说法的，一个都不写。整个写入由节点放在 ask() 之后的一个 @task 里：审批 +恢复重放、接着运行时不会再写一遍。 catalog.save_patch 不收备注，这里用它的同一套原料（read_entry、apply_patch、write_entry 的乐观锁）写，只多把备注放进项里。 """ from __future__ import annotations import logging +import re from collections.abc import Mapping from dataclasses import dataclass from typing import Any @@ -48,10 +61,25 @@ _VERB = {"count": "计数", "sum": "求和", "avg": "求平均"} #: 写入时版本对不上（别人刚改过这张表）最多重试几次 _RETRIES = 2 +#: 无效记录的取值：被排除的取值（或码值表里它的意思）里有这些词，才算在排除无效记录 +_INVALID_WORDS = frozenset({ + "refund", "refunded", "refunding", "cancel", "canceled", "cancelled", "void", "voided", "delete", "deleted", + "invalid", "test", "testing", "fraud", "spam", "dummy", "fake", "sandbox", "fail", "failed", "duplicate", + "duplicated", "obsolete", +}) +_INVALID_CJK = ("退款", "作废", "取消", "测试", "无效", "删除", "撤销", "失败", "重复", "虚假", "欺诈", "废弃") +#: 标记列：列名里有这些词的，标记的是无效记录（is_test、is_deleted、del_flag、deleted_at……），留下的是没标记的 +_MARKS_INVALID = _INVALID_WORDS | {"del", "removed", "hidden", "archived", "internal", "disabled", "inactive"} +#: 标记列：列名里有这些词的，标记的是有效记录（is_valid、enabled、is_active……），留下的是标记了的 +_MARKS_VALID = frozenset({"valid", "enabled", "active", "effective"}) +#: 取值和指标名称、口径说明比对时，数字至少这么长才比（0、1 这种标记值到处都是） +_SCOPE_DIGITS = 2 + @dataclass(frozen=True) class Fact: - """要写进目录的一项：哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明。""" + """要写进目录的一项：哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明；sql、dialect 是被采纳 + 那一边的写法和方言，有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。""" source: str table: str @@ -59,13 +87,19 @@ class Fact: value: Any metric: str choice: str + sql: str = "" + dialect: str | None = None -def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None) -> tuple[Fact | None, str | None]: +def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None, + notes: Mapping[str, Mapping[str, Any]] | None = None) -> tuple[Fact | None, str | None]: """一个候选（审核人在某个指标上选了 a 或 b）→ (要写的目录项, None) 或 (None, 为什么不写)。 - candidate：{metric 指标名, choice "a" | "b", source 数据源名, dialect, sql: {a, b}, diff: {equivalent, fallback, - items: [FacetDiff.as_dict()]}}。只看差异说明，不重新比 SQL：审核人看到的就是这份说明。 + candidate：{metric 指标名, choice "a" | "b", source 数据源名, dialect, sql: {a, b}（比较的两条 SQL）, diff: + {equivalent, fallback, items: [FacetDiff.as_dict()]}, scope: {texts, columns}}。scope 是这个指标自己的范围： + texts 是口径说明、所在分组的取值、本次运行的输入（指标名称总算在内），columns 是分组的列；有效记录条件用它排除 + 指标自己的筛选范围。notes 是这个数据源的目录（{表名: notes}），认码值的意思用。只看差异说明，不重新比 SQL：审核人 + 看到的就是这份说明。 """ name = str(candidate.get("metric") or "") side = candidate.get("choice") @@ -83,7 +117,7 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None) if aspect == "time": fact, why = _business_date(item, side) elif aspect == "filter": - fact, why = _valid_filter(item, side, candidate.get("dialect")) + fact, why = _valid_filter(item, side, candidate, schema_cache, notes or {}) elif aspect == "aggregate": fact, why = _dedup(item, side, candidate, schema_cache) else: @@ -95,21 +129,36 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None) resolved = catalog.resolve_table_name(schema_cache, table) if resolved is None: return None, f"指标「{name}」涉及的表「{table}」不在数据源的表结构里，不写入数据目录" + sql = candidate.get("sql") if isinstance(candidate.get("sql"), Mapping) else {} + dialect = candidate.get("dialect") return Fact(source=str(candidate.get("source") or ""), table=resolved, item=key, value=value, metric=name, - choice=str(side)), None + choice=str(side), sql=str(sql.get(side) or ""), + dialect=dialect if isinstance(dialect, str) else None), None def _business_date(item: Mapping[str, Any], side: str) -> tuple[tuple[str, str, Any] | None, str | None]: + other = "b" if side == "a" else "a" detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {} - mine = [c for c in detail.get(side) or [] if isinstance(c, Mapping) and c.get("table") and c.get("column")] + + def columns(key: str) -> list[Mapping[str, Any]]: + return [c for c in detail.get(key) or [] if isinstance(c, Mapping) and c.get("table") and c.get("column")] + + mine, theirs = columns(side), columns(other) if not mine: return None, f"被采纳的{SIDE_LABEL[side]}没有按时间筛选，没有可记的业务日期" - if len(mine) > 1: - return None, f"被采纳的{SIDE_LABEL[side]}按好几列筛时间，无法确定业务日期按哪一列" + if not theirs: + return None, (f"只有被采纳的{SIDE_LABEL[side]}按时间筛选，{SIDE_LABEL[other]}没有：审核人判断的是要不要限定" + "时间，不是按哪一列归属，没有可记的业务日期") + if len(mine) > 1 or len(theirs) > 1: + return None, "有一边按好几列筛时间，无法确定业务日期按哪一列" + if (mine[0]["table"], mine[0]["column"]) == (theirs[0]["table"], theirs[0]["column"]): + return None, "两边按同一列筛时间，没有可记的业务日期" return (str(mine[0]["table"]), "business_date", {"column": str(mine[0]["column"])}), None -def _valid_filter(item: Mapping[str, Any], side: str, dialect: Any) -> tuple[tuple[str, str, Any] | None, str | None]: +def _valid_filter(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any], + schema_cache: Mapping[str, Any] | None, + notes: Mapping[str, Mapping[str, Any]]) -> tuple[tuple[str, str, Any] | None, str | None]: other = "b" if side == "a" else "a" condition, rival = item.get(side), item.get(other) if rival is not None: @@ -118,28 +167,209 @@ def _valid_filter(item: Mapping[str, Any], side: str, dialect: Any) -> tuple[tup return None, "两边各有不同的筛选条件，不是一边多出一个条件" if not isinstance(condition, str) or not condition.strip(): return None, "差异说明里没有被采纳一边的筛选条件" + condition = condition.strip() detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {} - tables = {str(c.get("table")) for c in detail.get("columns") or [] if isinstance(c, Mapping) and c.get("table")} + refs = [c for c in detail.get("columns") or [] if isinstance(c, Mapping) and c.get("table") and c.get("column")] + tables = {str(c["table"]) for c in refs} if len(tables) != 1: return None, "筛选条件不是只涉及一张表的列" (table,) = tables - return (table, "valid_filter", _unqualified(condition.strip(), dialect)), None + read = _read(candidate.get("dialect")) + tree = _parse(condition, read) + if tree is None: + return None, f"多出的条件「{condition}」无法解析，无法确定是不是有效记录条件" + scope = candidate.get("scope") if isinstance(candidate.get("scope"), Mapping) else {} + grouped = {str(c).lower() for c in scope.get("columns") or [] if isinstance(c, str)} + if hit := sorted({str(c["column"]) for c in refs if str(c["column"]).lower() in grouped}): + return None, (f"多出的条件「{condition}」用到的「{'、'.join(hit)}」是原流程分组的列，可能是指标自己的筛选范围，" + "不是整张表的有效记录条件") + texts = [str(candidate.get("metric") or ""), *(str(t) for t in scope.get("texts") or [] if isinstance(t, str))] + if (word := _scope_hit(tree, texts)) is not None: + return None, (f"多出的条件「{condition}」里的「{word}」出现在指标名称、口径说明、所在分组或本次运行的输入里，" + "可能是指标自己的筛选范围，不是整张表的有效记录条件") + resolved = catalog.resolve_table_name(schema_cache, table) or table + if not _drops_invalid(tree, notes.get(resolved) or {}): + return None, (f"多出的条件「{condition}」不是在排除退款、作废、测试这类无效记录，可能是指标自己的筛选范围" + "（例如只看某个地区、某个渠道），不是整张表的有效记录条件") + return (table, "valid_filter", _unqualified(condition, candidate.get("dialect"))), None + + +# ---- 有效记录条件：每一段都在排除无效记录 + +def _read(dialect: Any) -> str | None: + from app.data.sqlcheck import dialect_of + + return dialect_of(dialect) if isinstance(dialect, str) else None + + +def _parse(condition: str, read: str | None) -> Any: + import sqlglot + + try: + return sqlglot.parse_one(condition, read=read) + except Exception: # noqa: BLE001 - 写法特殊、解析不了：调用方当作认不出 + return None + + +def _unparen(node: Any) -> Any: + from sqlglot import exp + + while isinstance(node, exp.Paren): + node = node.this + return node + + +def _and_parts(node: Any) -> list[Any]: + """条件按 AND 拆成几段（括号拆开）。""" + from sqlglot import exp + + node = _unparen(node) + if isinstance(node, exp.And): + return _and_parts(node.this) + _and_parts(node.expression) + return [node] + + +def _literal(node: Any) -> tuple[bool, Any]: + """常量的值：(是不是常量, 值)。字符串原样，数字转成 int / float，TRUE / FALSE 是 bool。""" + from sqlglot import exp + + node = _unparen(node) + if isinstance(node, exp.Boolean): + return True, bool(node.this) + negative = isinstance(node, exp.Neg) + if negative: + node = _unparen(node.this) + if not isinstance(node, exp.Literal): + return False, None + if node.is_string: + return (False, None) if negative else (True, str(node.this)) + try: + number = float(node.this) if any(ch in str(node.this) for ch in ".eE") else int(node.this) + except ValueError: + return False, None + return True, -number if negative else number + + +def _scope_hit(tree: Any, texts: list[str]) -> str | None: + """条件里的取值出现在指标自己的范围里（指标名称、口径说明、分组、运行输入）：返回那个取值，没有返回 None。""" + from sqlglot import exp + + lowered = [t.lower() for t in texts if t.strip()] + for node in tree.find_all(exp.Literal): + word = str(node.this).strip() + if not word: + continue + if node.is_string: + if any(word.lower() in t for t in lowered): + return word + elif len(word) >= _SCOPE_DIGITS: + pattern = re.compile(rf"(? bool: + lowered = text.lower() + return any(w in _INVALID_WORDS for w in re.split(r"[^a-z0-9]+", lowered) if w) or \ + any(mark in text for mark in _INVALID_CJK) + + +def _invalid_value(value: Any, column: str, notes: Mapping[str, Any]) -> bool: + """被排除的取值是不是无效记录：取值本身（'refunded'、'已作废'），或者码值表里它的意思（9 → 作废）。""" + if isinstance(value, bool): + return False + if isinstance(value, str) and _invalid_text(value): + return True + columns = notes.get("columns") if isinstance(notes.get("columns"), Mapping) else {} + entry = next((v for k, v in columns.items() if str(k).lower() == column.lower()), None) + codes = entry.get("codes") if isinstance(entry, Mapping) else None + if not isinstance(codes, Mapping) or codes.get("status") == "rejected" \ + or not isinstance(codes.get("value"), Mapping): + return False + key = str(int(value)) if isinstance(value, float) and value.is_integer() else str(value) + meaning = codes["value"].get(key) + return isinstance(meaning, str) and _invalid_text(meaning) + + +def _polarity(column: str) -> str | None: + """标记列标记的是什么：invalid（is_test、deleted_at）、valid（is_valid、enabled），不是标记列为 None。""" + words = catalog.name_words(column) + if any(w in _MARKS_INVALID for w in words): + return "invalid" + if any(w in _MARKS_VALID for w in words): + return "valid" + return None + + +def _flag_value(value: Any) -> bool | None: + """标记值：0 / 1 / TRUE / FALSE → bool，别的为 None。""" + if isinstance(value, bool): + return value + if isinstance(value, (int, float)) and value in (0, 1): + return bool(value) + return None + + +def _drops_invalid(tree: Any, notes: Mapping[str, Any]) -> bool: + """条件的每一段都在排除无效记录。""" + return all(_drops_invalid_part(part, notes) for part in _and_parts(tree)) + + +def _drops_invalid_part(node: Any, notes: Mapping[str, Any]) -> bool: + from sqlglot import exp + + node = _unparen(node) + negated = False + if isinstance(node, exp.Not): + negated, node = True, _unparen(node.this) + if isinstance(node, exp.NEQ): + negated, node = not negated, exp.EQ(this=node.this.copy(), expression=node.expression.copy()) + if isinstance(node, exp.EQ): + left, right = _unparen(node.this), _unparen(node.expression) + column, other = (left, right) if isinstance(left, exp.Column) else (right, left) + ok, value = _literal(other) + return isinstance(column, exp.Column) and ok and _keeps_valid(column.name, [value], negated, notes) + if isinstance(node, exp.In) and isinstance(_unparen(node.this), exp.Column) and not node.args.get("query"): + values = [_literal(e) for e in node.expressions] + return bool(values) and all(ok for ok, _ in values) and \ + _keeps_valid(_unparen(node.this).name, [v for _, v in values], negated, notes) + if isinstance(node, exp.Is) and isinstance(node.expression, exp.Null) \ + and isinstance(_unparen(node.this), exp.Column): + # 列 IS NULL 留下的是没标记的行：标记无效记录的列（deleted_at IS NULL）才是有效记录；IS NOT NULL 反过来 + polarity = _polarity(_unparen(node.this).name) + return polarity == ("valid" if negated else "invalid") + if isinstance(node, exp.Column): + # 布尔标记列直接当条件写：is_valid、NOT is_test + return _polarity(node.name) == ("invalid" if negated else "valid") + if isinstance(node, (exp.Like, exp.ILike)) and negated: + ok, pattern = _literal(node.expression) + return ok and isinstance(pattern, str) and _invalid_text(pattern) + return False + + +def _keeps_valid(column: str, values: list[Any], negated: bool, notes: Mapping[str, Any]) -> bool: + """列 = 取值（negated 时是 <>、NOT IN）留下的是不是有效记录。""" + polarity = _polarity(column) + flags = [_flag_value(v) for v in values] + if polarity is not None and all(f is not None for f in flags): + if len(set(flags)) != 1: + return False + # is_test = 1 留下的是标记了的行，is_test <> 1 留下的是没标记的；标记有效记录的列反过来 + keeps_marked = flags[0] != negated + return keeps_marked if polarity == "valid" else not keeps_marked + # 排除几个取值：被排除的都得是无效记录。只保留几个取值（status = 'paid'）就是指标自己的范围 + return negated and all(_invalid_value(v, column, notes) for v in values) def _unqualified(condition: str, dialect: Any) -> str: """条件里的列去掉表名：有效记录条件记在这张表自己名下，写成 status <> 'refunded' 而不是 orders.status <> …。 解析不了就原样。""" - import sqlglot from sqlglot import exp - from app.data.sqlcheck import dialect_of - - read = dialect_of(dialect) if isinstance(dialect, str) else None - try: - tree = sqlglot.parse_one(condition, read=read) - except Exception: # noqa: BLE001 - 写法特殊、解析不了：差异说明里给人看的就是这一句，原样记 - return condition - if not any(col.table for col in tree.find_all(exp.Column)): + read = _read(dialect) + tree = _parse(condition, read) + if tree is None or not any(col.table for col in tree.find_all(exp.Column)): return condition for col in tree.find_all(exp.Column): col.set("table", None) @@ -149,9 +379,57 @@ def _unqualified(condition: str, dialect: Any) -> str: return condition +def _normalized(node: Any, read: str | None) -> str: + """一段条件的比较用写法：列去掉表名、列名小写，按方言重新写一遍（<> 和 !=、NOT IN 的两种写法认成一样）。""" + from sqlglot import exp + + copy = node.copy() + for col in list(copy.find_all(exp.Column)): + col.set("table", None) + col.set("this", exp.to_identifier(col.name.lower())) + return copy.sql(dialect=read) + + +def _conditions_of(sql: str, read: str | None) -> set[str] | None: + """一条 SQL 里所有 WHERE 的每一段（比较用写法）；解析不了返回 None。""" + import sqlglot + from sqlglot import exp + + try: + tree = sqlglot.parse_one(sql, read=read) + except Exception: # noqa: BLE001 + return None + return {_normalized(part, read) for where in tree.find_all(exp.Where) for part in _and_parts(where.this)} + + +def _combined_filter(existing: str, fact: Fact) -> tuple[str | None, str | None]: + """目录里已有的有效记录条件和这次的新条件合起来：(合起来的整句条件, None) 或 (None, 为什么不动)。 + + 有效记录条件是一整句（copilot 提案也要求「原有条件也要带上」）。只写新条件会把两边都按着筛的原有条件静默替换掉； + 原有条件被采纳的一边没按它筛，审核人的选择说明不了它对不对，也不能替人删掉。""" + label = catalog.TABLE_FIELD_LABEL["valid_filter"] + read = _read(fact.dialect) + have_tree, new_tree = _parse(existing, read), _parse(str(fact.value), read) + if have_tree is None or new_tree is None: + return None, f"「{fact.table}」已有的{label}「{existing}」无法解析，与这次的条件合不起来，未改动" + have = [_normalized(p, read) for p in _and_parts(have_tree)] + used = _conditions_of(fact.sql, read) or set() + if not set(have) <= used: + return None, (f"「{fact.table}」已有的{label}「{existing}」，被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" + "没有按它筛选，两者的关系无法确定，未改动。如需更改，请到数据目录中修改") + if {_normalized(p, read) for p in _and_parts(new_tree)} <= set(have): + return existing, None + from sqlglot import exp + + # 原有条件最外层是 OR 的，接 AND 之前要加括号，不然新条件只并到 OR 的后一半上 + head = f"({existing})" if isinstance(have_tree, exp.Or) else existing + return f"{head} AND {fact.value}", None + + def _dedup(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None) -> tuple[tuple[str, str, Any] | None, str | None]: - """聚合只差去不去重（同一个函数、同一个参数，一边写了 DISTINCT）。表从被采纳那一边的 SQL 里认：参数列属于哪张实表。""" + """聚合只差去不去重（同一个函数、同一个参数，一边写了 DISTINCT），被采纳的一边去重。表从被采纳那一边的 SQL 里认： + 参数列属于哪张实表。""" from app.data.sql_semantics import facets pair = _aggregate_pair(item.get("a"), item.get("b"), candidate.get("dialect")) @@ -159,6 +437,9 @@ def _dedup(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any], return None, "两边的聚合不只是去不去重的差别" func, distinct = pair adopted = distinct[0] if side == "a" else distinct[1] + if not adopted: + return None, (f"被采纳的{SIDE_LABEL[side]}没有去重：不去重是这个指标自己的算法，" + "不是表的去重规则") sql = (candidate.get("sql") or {}).get(side) if isinstance(candidate.get("sql"), Mapping) else None found = facets(str(sql or ""), candidate.get("dialect"), schema=schema_cache) hits = [a for a in found.aggregates if a.func == func and a.distinct == adopted and not a.scope] @@ -172,27 +453,19 @@ def _dedup(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any], if len(names) != 1: return None, "无法确定去重的列属于哪一张表" table = names[0] - verb = _VERB.get(func, func) - text = f"对 {column} {verb}时去重" if adopted else f"对 {column} {verb}时不去重" - return (str(table), "dedup", text), None + return (str(table), "dedup", f"对 {column} {_VERB.get(func, func)}时去重"), None def _aggregate_pair(a: Any, b: Any, dialect: Any) -> tuple[str, tuple[bool, bool]] | None: """两边的聚合写法 → (函数名, (A 去重吗, B 去重吗))；不是「同一个函数、同一个参数、只差 DISTINCT」返回 None。""" - import sqlglot from sqlglot import exp - from app.data.sqlcheck import dialect_of - - read = dialect_of(dialect) if isinstance(dialect, str) else None + read = _read(dialect) def shape(text: Any) -> tuple[str, str, bool] | None: if not isinstance(text, str) or not text.strip(): return None - try: - node = sqlglot.parse_one(text, read=read) - except Exception: # noqa: BLE001 - return None + node = _parse(text, read) if not isinstance(node, exp.AggFunc): return None arg = node.this @@ -236,12 +509,23 @@ def _same_fact(item: str, a: Any, b: Any) -> bool: return isinstance(a, str) and isinstance(b, str) and " ".join(a.split()) == " ".join(b.split()) +def _contradiction(facts: list[Fact]) -> str: + """同一次裁决里对同一项给出的不同说法，说清是谁、各说了什么。""" + first = facts[0] + label = catalog.TABLE_FIELD_LABEL.get(first.item, first.item) + said = "；".join(f"指标「{f.metric}」采纳的是「{_value_text(f.item, f.value)}」" for f in facts[:4]) + return (f"同一次裁决里对「{first.table}」的{label}给出的选择互相矛盾（{said}），无法确定哪一个是数据事实，" + "都未写入数据目录") + + async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, node_title: str, actor: str | None) -> dict[str, list[dict[str, Any]]]: """把审核人的选择写进数据目录：{written: [{source, table, item}], skipped: [{reason}]}。永远不抛：数据源没了、 - 目录写不进去，都落在 skipped 里照实说。""" - written: list[dict[str, Any]] = [] + 目录写不进去，都落在 skipped 里照实说。 + + 先把每个候选认成目录项，再按（数据源、表、项）归拢：同一项上说法互相矛盾的整组不写，剩下的逐个写。""" skipped: list[dict[str, Any]] = [] + facts: list[Fact] = [] for candidate in candidates: name = str(candidate.get("source") or "") try: @@ -250,26 +534,50 @@ async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, no if row is None: skipped.append({"reason": f"数据源「{name}」已不存在，指标「{candidate.get('metric')}」的选择未写入数据目录"}) continue - fact, why = derive(candidate, row.schema_cache) - if fact is None: - skipped.append({"reason": why}) - continue - done, why = await _write(session, row, fact, note=note_of(fact, run_id=run_id, node_title=node_title), - actor=actor) - except Exception: # noqa: BLE001 - 目录写不进去不能毁掉审批的结果：缺口照样按选择记 - logger.exception("双盲复算的选择写入数据目录失败（数据源 %s）", name) - skipped.append({"reason": f"指标「{candidate.get('metric')}」的选择写入数据目录时出错，未写入"}) + notes = {table: entry.notes for table, entry in (await catalog.read_catalog(session, row.id)).items()} + fact, why = derive(candidate, row.schema_cache, notes) + except Exception: # noqa: BLE001 - 目录读不出来不能毁掉审批的结果：缺口照样按选择记 + logger.exception("双盲复算的选择核对数据目录失败（数据源 %s）", name) + skipped.append({"reason": f"指标「{candidate.get('metric')}」的选择核对数据目录时出错，未写入"}) continue - if done: - written.append({"source": fact.source, "table": fact.table, "item": fact.item}) - else: + if fact is None: skipped.append({"reason": why}) + else: + facts.append(fact) + groups: dict[tuple[str, str, str], list[Fact]] = {} + for fact in facts: + groups.setdefault((fact.source, fact.table, fact.item), []).append(fact) + written: list[dict[str, Any]] = [] + for members in groups.values(): + if any(not _same_fact(f.item, f.value, members[0].value) for f in members[1:]): + skipped.append({"reason": _contradiction(members)}) + continue + for fact in members: + done, why = await _record(fact, run_id=run_id, node_title=node_title, actor=actor) + if done: + written.append({"source": fact.source, "table": fact.table, "item": fact.item}) + else: + skipped.append({"reason": why}) return {"written": written, "skipped": skipped} +async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None) -> tuple[bool, str | None]: + try: + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == fact.source))).scalar_one_or_none() + if row is None: + return False, f"数据源「{fact.source}」已不存在，指标「{fact.metric}」的选择未写入数据目录" + return await _write(session, row, fact, note=note_of(fact, run_id=run_id, node_title=node_title), + actor=actor) + except Exception: # noqa: BLE001 - 目录写不进去不能毁掉审批的结果：缺口照样按选择记 + logger.exception("双盲复算的选择写入数据目录失败（数据源 %s）", fact.source) + return False, f"指标「{fact.metric}」的选择写入数据目录时出错，未写入" + + async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: str | None) -> tuple[bool, str | None]: """写一项。(写了没有, 没写的原因)。""" label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item) + side = SIDE_LABEL.get(fact.choice, fact.choice) tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None for _ in range(_RETRIES): entry = await catalog.read_entry(session, row.id, fact.table) @@ -287,6 +595,21 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: and live["value"].get("column") == value.get("column"): # 业务日期按哪一列是这次选出来的事实；目录里同一列另写了规则、时区的，原样留着 value = live["value"] + elif fact.item == "valid_filter" and isinstance(live, dict) and isinstance(live.get("value"), str) \ + and live["value"].strip() and not _same_fact(fact.item, live["value"], value): + combined, why = _combined_filter(live["value"].strip(), fact) + if combined is None: + return False, why + if live.get("status") == "confirmed" and not _same_fact(fact.item, combined, live["value"]): + return False, (f"「{fact.table}」的{label}已有人工确认过的说法「{live['value'].strip()}」，被采纳的" + f"{side}另外多一个条件「{fact.value}」，未改动。如需补上，请到数据目录中修改") + value = combined + elif fact.item == "dedup" and isinstance(live, dict) and live.get("status") != "confirmed" \ + and not _same_fact(fact.item, live.get("value"), value): + # 推断的去重规则说的未必是同一件事（按哪一列去重、哪种汇总要去重），替换掉就把它丢了 + return False, (f"「{fact.table}」的{label}已有推断的说法「{_value_text(fact.item, live.get('value'))}」，" + f"与这次指标「{fact.metric}」采纳的「{value}」不同，无法确定两者的关系，未覆盖。" + "如需更改，请到数据目录中修改") changes = [{"path": fact.item, "value": value, "reason": note}] plan = catalog.plan_patch(notes, changes, table=fact.table, tables=tables) if plan.problems: diff --git a/backend/app/engine/launch.py b/backend/app/engine/launch.py index 0f05c1c0..dddfd94b 100644 --- a/backend/app/engine/launch.py +++ b/backend/app/engine/launch.py @@ -186,12 +186,13 @@ _VERIFY_NODES = (NodeType.DUAL_CHECK, NodeType.CHALLENGE) async def _with_verify_snapshot(session: AsyncSession, graph: dict[str, Any], options: dict[str, Any] | None) -> dict[str, Any] | None: - """图里有双盲复算或反方审稿节点时，把核查设置（api/settings.py 的 verify 一组）快照进 options["verify"]。 + """图里（含子工作流）有双盲复算或反方审稿节点时，把核查设置（api/settings.py 的 verify 一组）快照进 + options["verify"]。 这两种节点在运行中途读设置（开关、上限、核查模型）。审批停下期间有人改了设置，恢复后的那一段读到的就和前面 不一样：本该复算的被关掉了、重放时走了另一条路，正式运行也说不清当时按什么规则核的。所以在发起这一刻定下来， - 节点只读运行选项里的这一份（同 _approval_default 的道理）。没有这两种节点的图 options 原样不动，run.started - 和以前逐字相同；调用方已经给了 options["verify"] 的照它的。 + 节点只读运行选项里的这一份（同 _approval_default 的道理；子工作流的节点经运行上下文拿到的是同一份选项）。没有 + 这两种节点的图 options 原样不动，run.started 和以前逐字相同；调用方已经给了 options["verify"] 的照它的。 """ from app.api.settings import VERIFY_GROUP, verify_settings @@ -201,11 +202,44 @@ async def _with_verify_snapshot(session: AsyncSession, graph: dict[str, Any], spec = GraphSpec.model_validate(graph) except Exception: # noqa: BLE001 - 图本身不成形交给 run_manager.start 报 return options - if not any(node.type in _VERIFY_NODES for node in spec.nodes): + if not await _has_verify_nodes(session, spec): return options return {**(options or {}), VERIFY_GROUP: await verify_settings(session)} +async def _has_verify_nodes(session: AsyncSession, spec: GraphSpec) -> bool: + """图里有没有双盲复算、反方审稿节点，子工作流里的也算。 + + 子工作流照运行时的取法读（同 runner._bound_sources）：固定了版本的读那一版，没固定的读工作流现在的图。读不到、 + 图不成形的跳过——那个节点运行时自己会报错。同一个工作流（同一版）只读一次，互相引用也不会转圈。 + """ + seen: set[tuple[str, str]] = set() + queue: list[GraphSpec] = [spec] + while queue: + graph = queue.pop(0) + if any(node.type in _VERIFY_NODES for node in graph.nodes): + return True + for node in graph.nodes: + if node.type != NodeType.SUBGRAPH: + continue + wid, pinned = (node_setting(graph, node, k) for k in ("workflow_id", "workflow_version")) + key = (str(wid or ""), str(pinned or "")) + if not wid or key in seen: + continue + seen.add(key) + try: + # 只取图：Workflow 的 versions 是 selectin，整行读会把所有历史版本一起读出来 + query = (select(WorkflowVersion.graph).where( + WorkflowVersion.workflow_id == wid, WorkflowVersion.version == int(pinned)) + if pinned else select(Workflow.graph).where(Workflow.id == wid)) + sub = (await session.execute(query)).scalar_one_or_none() + if sub: + queue.append(GraphSpec.model_validate(sub)) + except Exception: # noqa: BLE001 - 读不到、不成形：子工作流节点运行时会说清楚 + continue + return False + + # -------------------------------------------------------------------------- # 发起前检查绑定的工具 # -------------------------------------------------------------------------- diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 596b211a..25e2a736 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -2,16 +2,27 @@ **按输入复算。** 指标由口径卡的原式算出，原式本身确定、可复算（metrics 的 recompute_ok）；会错的是喂给它的那几个数： 查哪张表、按哪天归属、排没排除退款……所以盲算的单位是「一个来自查询快照的输入」。盲算 agent（engine/verify_agent.py） -只看得到指标名、口径说明（节点配置的 definitions）、这个输入在计算式里是哪一项（结果分了组的另给那一行的维度）、本次 -运行的输入参数和数据源的数据目录，看不到原 SQL 和原结果；它自己写只读 SQL，最后交一个格子引用，由代码从它自己的查询 -快照里取值。盲值是拿原式代入盲算的输入算出来的（engine/recompute.py，取整同口径卡），和原值按相对误差比，原值为 0 时 -按绝对误差。两个指标用到同一格的只盲算一次。 +只看得到指标名、口径说明（节点配置的 definitions）、这个输入在计算式里是哪一项、本次运行的输入参数和数据源的数据目录， +看不到原 SQL 和原结果；它自己写只读 SQL，最后交一个格子引用，由代码从它自己的查询快照里取值。盲值是拿原式代入盲算的 +输入算出来的（engine/recompute.py，取整同口径卡），和原值按相对误差比，原值为 0 时按绝对误差。两个指标用到同一格的 +只盲算一次。几个盲算同时进行（BLIND_CONCURRENCY 个），先单独试第一个：核查模型用不了时不必个个去试。 -**一致就放行，不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪（data/sql_semantics.explain_sql_diff）， -完整对照落成一件工件（审批 payload 经 runner._safe 会截断，长 SQL 放不下），用 replay.ask 挂起。审核人逐个指标选采用 -原流程（a）、采用盲算（b）或两边都不对（neither）：a 没有缺口；b、neither 写进产出的 issuance_gaps，成果节点的出具 -契约据此降档（on_reject 为 fail 时节点直接失败）。两边只差一个方面、又对得上数据目录事实的，选择写进数据目录 -（engine/dual_check_facts.py）。盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口。 +**原流程的取法一点都不透。** 原流程结果的列名（gmv_by_order_date、下单金额）、同一行的维度列名（order_date）都是原流程 +的取舍，告诉盲算就等于告诉它按哪一列归属，它照着算出同一个数，下单日和支付日这类陷阱就漏过去了。所以结果分了组时只给 +这一行的取值（「华东」、2026-09-02 这一天），按哪一列分组请它按口径说明判断；原 SQL 里写死的文字列（'下单日' AS basis） +连取值也不给；遮罩列不给。计算式只有一项来自查询、或者写了口径说明时不给列名；几项都来自查询又没写口径说明的，列名 +里看不出日期、口径的取舍时才给，好让它分得清哪一项是哪一项。 + +**一致就放行，不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪（data/sql_semantics.explain_sql_diff）： +两边先收窄到这一格（只留算出这一列的那一项；分组的按这一行的分组值改成等值条件），一条查询算好几列、分组后取一行时 +才不会混进「选出的列」「分组」这类和这一格无关的差别。完整对照落成一件工件（审批 payload 经 runner._safe 会截断， +长 SQL 放不下），用 replay.ask 挂起。审核人逐个指标选采用原流程（a）、采用盲算（b）或两边都不对（neither）：a 没有 +缺口；b、neither、没选的写进产出的 issuance_gaps，成果节点的出具契约据此降档（on_reject 为 fail 时节点直接失败）。 +两边只差一个方面、又对得上数据目录事实的，选择写进数据目录（engine/dual_check_facts.py，指标自己的筛选范围不写）。 +盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口，发 dual_check_unchecked。 + +**产出的状态。** 全部一致是 agreed；没有不一致、但有无法复算的：一个都没核成是 unchecked，核成了一部分是 partial； +有不一致、人裁决过的是 resolved。产出和台账都带 counts（一致、不一致、无法复算各几个）。 **重放。** 审批恢复时节点从头重放。读设置（发起时没快照的才读）、盲算连同比对、差异说明连同工件、写数据目录各在一个 @task 里，重放时直接取 checkpoint 里的结果：不再调用模型、不再计费、不再写目录，事件也都在 task 里发、不会再发一遍； @@ -50,8 +61,8 @@ from app.engine.replay import _in_graph, ask, once from app.engine.schema import DUAL_CHECK_FIELD_LABEL, DUAL_CHECK_ON_REJECT, NodeType, dual_check_card from app.engine.state import GraphState -#: 节点产出的状态：跳过、无需裁决、人已裁决 -SKIPPED, AGREED, RESOLVED = "skipped", "agreed", "resolved" +#: 节点产出的状态：跳过、全部一致、核成了一部分（其余无法复算）、一个都没核成、人已裁决 +SKIPPED, AGREED, PARTIAL, NONE_CHECKED, RESOLVED = "skipped", "agreed", "partial", "unchecked", "resolved" #: 单个指标的结果：一致、不一致、无法复算 SAME, DIFFERENT, UNCHECKED = "agreed", "disagreed", "unchecked" #: 审核人的三种选择：采用原流程、采用盲算、两边都不对 @@ -73,6 +84,15 @@ _NOTES_CHARS = 6_000 _DIMS = 6 _DIM_CHARS = 40 _INPUTS = 12 +#: 几个盲算同时进行：串行时几个慢的输入就能把运行这一段的时长上限（max_run_seconds）耗光 +BLIND_CONCURRENCY = 3 +#: 交给盲算的取值像日期、月份时换个说法：「2026-09-02 这一天」，不说是哪一列 +_DAY = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_MONTH = re.compile(r"^\d{4}-\d{2}$") +#: 原流程结果的列名里出现这些，就带着原流程的取舍（按哪天归属、排没排除、去没去重），不交给盲算 +_LOADED_NAME = ("date", "day", "time", "dt", "_at", "month", "week", "year", "paid", "pay", "order", "refund", + "cancel", "distinct", "dedup", "unique", "net", "gross", "valid", "test", "日", "月", "周", "年", + "时间", "下单", "支付", "付款", "退款", "去重", "剔除", "排除", "不含", "含", "有效", "净", "毛") #: 卡上给容器输入记的大概（metrics._brief）：「[12 项]」「{3 个字段}」，写进计算式时不当成字面量 _BRIEF = re.compile(r"^(?:\[\d+ 项\]|\{\d+ 个字段\})$") @@ -88,6 +108,7 @@ NO_SQL = "有一边没有可比较的 SQL，请对照原文" #: 交给成果节点的缺口（不带节点名：出具契约统一加「双盲复算「节点名」：」） GAP_BLIND = "指标「{name}」双盲复算不一致，审核人选择了盲算结果" GAP_NEITHER = "指标「{name}」双盲复算不一致，审核人判断两边都不对" +GAP_UNANSWERED = "指标「{name}」双盲复算不一致，审核人没有作出选择" GAP_UNCHECKED = "指标「{name}」无法复算：{reason}" BLIND_SYSTEM = ( @@ -103,7 +124,7 @@ BLIND_TASK_PROMPT = ( "指标：{name}{unit}\n" "口径说明：{definition}\n" "计算式：{formula}\n" - "你要算的是计算式里的 {label}{whole}：原流程里它是一次查询结果中「{column}」列的值{dims}。只交这一个数。\n" + "你要算的是计算式里的 {label}{whole}{hint}。{dims}只交这一个数。\n" "数据源：{source}\n" "{inputs}\n" "{catalog}" @@ -111,7 +132,10 @@ BLIND_TASK_PROMPT = ( BLIND_UNIT_PROMPT = "（单位：{unit}）" BLIND_NO_DEFINITION_PROMPT = "（未提供，按指标名称和数据目录理解）" BLIND_WHOLE_PROMPT = "（这个指标就是它本身）" -BLIND_DIMS_PROMPT = "，所在行的其他列：{dims}" +BLIND_HINT_PROMPT = "（原流程的结果里这一项叫「{column}」）" +BLIND_DIMS_PROMPT = "原流程的结果分了组，这个数属于对应 {dims} 的那一组；按哪一列分组、怎么归属，请按口径说明和数据目录判断。" +BLIND_DAY_PROMPT = "{value} 这一天" +BLIND_MONTH_PROMPT = "{value} 这个月" BLIND_INPUTS_PROMPT = "本次运行的输入（统计期之类的参数在这里）：{inputs}\n" BLIND_INDEX_PROMPT = "数据源里的表（表名｜中文名｜粒度）：\n{index}" BLIND_NOTES_PROMPT = "数据目录（标「推断，未确认」的项是推断，使用前请核实）：\n{notes}" @@ -159,9 +183,12 @@ async def run_dual_check(state: GraphState, ctx: NodeContext) -> dict[str, Any]: rows = checked["rows"] disagreed = [r for r in rows if r["status"] == DIFFERENT] if not disagreed: - return _finish(state, ctx, card_id, card, checked, status=AGREED, gaps=_unchecked_gaps(rows)) + agreed = sum(1 for r in rows if r["status"] == SAME) + # agreed 只说「全部一致」：一个都没核成、只核成一部分的另记，时间线和证据面板才不会把降档的出具说成一致 + status = AGREED if agreed == len(rows) else PARTIAL if agreed else NONE_CHECKED + return _finish(state, ctx, card_id, card, checked, status=status, gaps=_unchecked_gaps(rows)) - review = await _step("dual_check_review", lambda: _review(ctx, checked)) + review = await _step("dual_check_review", lambda: _review(state, ctx, checked, options)) response = await ask(ctx, review["payload"], **review["payload"]) # 谁裁决的跟着事件走，和人工审批节点一样 await once(ctx, EventType.HUMAN_RESOLVED, response=response, actor=ctx.actor()) @@ -170,9 +197,9 @@ async def run_dual_check(state: GraphState, ctx: NodeContext) -> dict[str, Any]: await once(ctx, EventType.LOG, level="info", code="dual_check_resolved", choices=choices, gaps=gaps, message=_resolved_text(disagreed, choices)) candidates = [{**review["facts"][r["metric_id"]], "choice": choices[r["metric_id"]]} - for r in disagreed if choices[r["metric_id"]] in ("a", "b") and r["metric_id"] in review["facts"]] + for r in disagreed if choices.get(r["metric_id"]) in ("a", "b") and r["metric_id"] in review["facts"]] catalog = await _step("dual_check_catalog", lambda: _catalog(ctx, candidates)) if candidates else None - rejected = [r for r in disagreed if choices[r["metric_id"]] != "a"] + rejected = [r for r in disagreed if choices.get(r["metric_id"]) != "a"] if rejected and options.on_reject == "fail": # 目录照样写了：审核人的选择是对数据的判断，和这次运行停不停无关 names = "、".join(f"「{r['name']}」" for r in rejected[:5]) @@ -349,8 +376,8 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An """逐个指标认出能复算的输入、按输入盲算、代回原式比对。返回 {rows, results, model, same_model, usage}。 rows 每个指标一项：{metric_id, name, unit, decimals, status, reason?, original: {value, rendered}, blind: {value, - rendered}, delta_pct, inputs: [{path, key, value, sql, source, alias, artifact, row, column, node_id}], differing}； - results 是每个盲算（按格子去重）的结果，见 _slim。 + rendered}, delta_pct, inputs: [{path, key, value, sql, source, alias, artifact, row, column, node_id, via}], + differing}；results 是每个盲算（按格子去重）的结果，见 _slim。 """ sources = _Sources() load, outputs = _loader(state) @@ -358,15 +385,18 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An inputs = _run_inputs(state) rows: list[dict[str, Any]] = [] jobs: dict[str, dict[str, Any]] = {} + sharers: dict[str, list[dict[str, str]]] = {} for metric in targets: row, metric_jobs = await _plan_metric(state, metric, options, sources, aliases, load, outputs) - for job in metric_jobs.values(): - job["inputs"] = inputs rows.append(row) for key, job in metric_jobs.items(): + job["inputs"] = inputs + sharers.setdefault(key, []).append({"id": row["metric_id"], "name": row["name"]}) # 两个指标用到同一格只盲算一次，按先出现的那个指标说清要算什么；先出现的没写口径说明、后一个写了，用后一个 if key not in jobs or (not jobs[key]["definition"] and job["definition"]): jobs[key] = job + for key, job in jobs.items(): + job["metrics"] = sharers[key] results, model, same, usage = await _blind(ctx, jobs, options, settings, _producer_models(state, rows)) for row, metric in zip(rows, targets): if row["status"] != UNCHECKED: @@ -377,6 +407,14 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An if agreed and not any(r["status"] == DIFFERENT for r in rows): ctx.emit(EventType.LOG, level="info", code="dual_check_agreed", metrics=agreed, message=f"双盲复算一致：{agreed} 个指标两边的结果相同，已放行") + if unchecked := [r for r in rows if r["status"] == UNCHECKED]: + # 结构上没法单独复算的（合并查询、遮罩列）以前一条 dual_check_* 日志都没有，时间线上看不出为什么没核成 + parts = "；".join(f"「{r['name']}」（{r.get('reason') or '原因不明'}）" for r in unchecked[:5]) + ctx.emit(EventType.LOG, level="warn", code="dual_check_unchecked", + metrics=[{"id": r["metric_id"], "name": r["name"], "reason": r.get("reason") or "原因不明"} + for r in unchecked], + message=f"双盲复算未能核对 {len(unchecked)} 个指标：{parts}" + + ("等" if len(unchecked) > 5 else "") + "，按无法复算计入缺口") return {"rows": rows, "results": results, "model": model, "same_model": same, "usage": usage} @@ -435,14 +473,17 @@ async def _plan_metric(state: GraphState, metric: dict[str, Any], options: _Opti return row, {} labels = {inp["path"]: f"输入{i}" for i, inp in enumerate(row["inputs"], 1)} formula = _formula(metric, labels) + definition = options.definitions.get(row["metric_id"]) or "" jobs: dict[str, dict[str, Any]] = {} for inp in row["inputs"]: snap, _ = sources.snapshot(inp["artifact"]) + hidden = (masks.get(inp["artifact"]) or frozenset()) | await _constant_columns(snap or {}, sources) + # 列名是原流程的取舍：只有一项来自查询、或者写了口径说明时用不着；几项又没口径说明的，看不出取舍才给 + hint = inp["column"] if len(row["inputs"]) > 1 and not definition and _neutral(inp["column"]) else None jobs[inp["key"]] = { "key": inp["key"], "metric_id": row["metric_id"], "metric": row["name"], "unit": row["unit"], - "definition": options.definitions.get(row["metric_id"]) or "", "formula": formula, - "label": labels[inp["path"]], "column": inp["column"], "source": inp["source"], - "dims": _dims(snap or {}, inp["row"], inp["column"], masks.get(inp["artifact"]) or frozenset()), + "definition": definition, "formula": formula, "label": labels[inp["path"]], "hint": hint, + "source": inp["source"], "dims": _dims(snap or {}, inp["row"], inp["column"], hidden), } return row, jobs @@ -534,30 +575,80 @@ def _formula(metric: dict[str, Any], labels: dict[str, str]) -> str: return ast.unparse(ast.fix_missing_locations(swapped)) -def _dims(snapshot: dict[str, Any], row: int, column: str, hidden: frozenset[str]) -> list[str]: - """原流程那一格所在行的其他列里不是数的那些（地区、日期这类维度）：结果分了组时，盲算 agent 要知道算的是哪一组。 - 数不给（同一行别的指标值会带偏它），遮罩列不给。""" +def _row_pairs(snapshot: Mapping[str, Any], row: int, column: str) -> list[tuple[str, Any]]: + """原流程那一格所在行的其他列里不是数的那些：(列名, 取值)。数不算（同一行别的指标值）。""" rows = snapshot.get("rows") or [] - if not 0 <= row < len(rows): + if not isinstance(row, int) or not 0 <= row < len(rows): return [] record = rows[row] columns = [str(c) for c in snapshot.get("columns") or []] pairs = list(record.items()) if isinstance(record, dict) else list(zip(columns, record)) - out: list[str] = [] + out: list[tuple[str, Any]] = [] for name, value in pairs: name = str(name) - if name == column or name.lower() in hidden or value is None: + if name == column or value is None: continue kind = column_kind(snapshot, name) if kind == "number" or (kind is None and (is_number(value) or numeric_text(value) is not None)): continue - text = str(value).replace("\n", " ") - out.append(f"{name} 为「{text[:_DIM_CHARS]}」") + out.append((name, value)) + return out + + +def _dims(snapshot: Mapping[str, Any], row: int, column: str, hidden: frozenset[str]) -> list[str]: + """结果分了组时，交给盲算 agent 的这一行的分组取值（地区、日期这类维度）：它要知道算的是哪一组。只给取值、不给 + 列名（原流程按哪一列分的组正是要它自己判断的），遮罩列、原 SQL 里写死的文字列（hidden，小写）不给。""" + out: list[str] = [] + for name, value in _row_pairs(snapshot, row, column): + if name.lower() in hidden: + continue + out.append(str(value).replace("\n", " ")[:_DIM_CHARS]) if len(out) >= _DIMS: break return out +def _dim_text(value: str) -> str: + """一个分组取值给盲算看的说法：日期、月份说「2026-09-02 这一天」，别的加「」。""" + if _DAY.match(value): + return BLIND_DAY_PROMPT.format(value=value) + if _MONTH.match(value): + return BLIND_MONTH_PROMPT.format(value=value) + return f"「{value}」" + + +def _neutral(column: str) -> bool: + """原流程结果的列名看不出取舍（按哪天归属、排没排除、去没去重）：可以交给盲算区分计算式里的几项。""" + lowered = column.lower() + return bool(column.strip()) and not any(word in lowered for word in _LOADED_NAME) + + +async def _constant_columns(snapshot: Mapping[str, Any], sources: _Sources) -> frozenset[str]: + """原 SQL 最外层写死的文字列（'下单日' AS basis）的列名（小写）：那是原流程自己的标注，不是分组。解析不了为空。""" + import sqlglot + from sqlglot import exp + + from app.data.sqlcheck import dialect_of + + sql = snapshot.get("sql") + name = snapshot.get("source") + if not isinstance(sql, str) or not sql.strip(): + return frozenset() + found = await sources.source(name) if isinstance(name, str) and name else None + read = dialect_of(found.kind) if found is not None else None + try: + tree = sqlglot.parse_one(sql, read=read) + except Exception: # noqa: BLE001 - 解析不了就当没有：交给盲算的只是取值，最坏多给一个 + return frozenset() + out: set[str] = set() + for select in [tree] if isinstance(tree, exp.Select) else list(tree.find_all(exp.Select))[:1]: + for item in select.expressions: + inner = item.this if isinstance(item, exp.Alias) else item + if isinstance(inner, (exp.Literal, exp.Boolean, exp.Null)): + out.add(_output_name(item, read).lower()) + return frozenset(out) + + async def _catalog_text(sources: _Sources, name: str) -> str: """交给盲算 agent 的数据目录：表清单（表名、中文名、粒度）和有目录的表的业务说明（业务日期、有效记录条件、去重、 字段含义、码值、关联关系）。太长的截断，其余请它用 db_schema 查。""" @@ -591,18 +682,25 @@ def _clip(text: str, limit: int) -> str: def _task_text(job: dict[str, Any], catalog: str) -> str: + dims = "、".join(_dim_text(v) for v in job["dims"]) return BLIND_TASK_PROMPT.format( name=job["metric"], unit=BLIND_UNIT_PROMPT.format(unit=job["unit"]) if job["unit"] else "", definition=job["definition"] or BLIND_NO_DEFINITION_PROMPT, formula=job["formula"], label=job["label"], - whole=BLIND_WHOLE_PROMPT if job["formula"].strip() == job["label"] else "", column=job["column"], - dims=BLIND_DIMS_PROMPT.format(dims="、".join(job["dims"])) if job["dims"] else "", + whole=BLIND_WHOLE_PROMPT if job["formula"].strip() == job["label"] else "", + hint=BLIND_HINT_PROMPT.format(column=job["hint"]) if job.get("hint") else "", + dims=BLIND_DIMS_PROMPT.format(dims=dims) if dims else "", source=job["source"], catalog=catalog, inputs=BLIND_INPUTS_PROMPT.format(inputs="；".join(job["inputs"])) if job.get("inputs") else "") async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Options, settings: Mapping[str, Any], producers: set[str]) -> tuple[dict[str, Any], str | None, bool, dict[str, Any]]: - """按先后逐个盲算：(各格子的结果, 盲算模型, 是否和产出原数据的模型相同, 用量)。producers 是产出原数据的模型。""" + """盲算：(各格子的结果, 盲算模型, 是否和产出原数据的模型相同, 用量)。producers 是产出原数据的模型。 + + 先单独做第一个：核查模型本身用不了（没配好、接入出错）时，后面的一样用不了，照这一次的原因记，不再一个个去试。 + 其余的同时进行，最多 BLIND_CONCURRENCY 个：串行时几个慢的输入（每个最多 timeout_s 加收尾的宽限）就能把运行这一段 + 的时长上限耗光，整段盲算没有 checkpoint，接着运行时全部重做、再花一次钱。结果按格子写回，日志按任务的先后发， + 和完成的先后无关。""" from app.api.settings import verify_model_spec from app.engine.verify_agent import run_verify_agent from app.tools.datasource import run_versions @@ -614,38 +712,56 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op spec = await verify_model_spec(session, ctx.config, settings=dict(settings)) sources = _Sources() catalogs: dict[str, str] = {} - results: dict[str, Any] = {} - model: str | None = None - same = False - broken: dict[str, Any] | None = None - for key, job in jobs.items(): - if broken is not None: - # 核查模型本身用不了（没配好、接入出错）：后面的盲算一样用不了，照这一次的原因记，不再一个个去试 - results[key] = broken - continue + for job in jobs.values(): if job["source"] not in catalogs: catalogs[job["source"]] = await _catalog_text(sources, job["source"]) - outcome = await run_verify_agent( + outcomes: dict[str, Any] = {} + + async def one(key: str, job: dict[str, Any]) -> None: + outcomes[key] = await run_verify_agent( spec=spec, system=BLIND_SYSTEM, task=_task_text(job, catalogs[job["source"]]), sources=[job["source"]], output_schema=BLIND_SCHEMA, max_queries=options.max_queries, timeout_s=options.timeout_s, daily_max_usd=options.daily_max_usd, emit=ctx.emit, run_id=ctx.run.run_id, node_id=ctx.node.id, data_versions=run_versions(ctx.run)) + + keys = list(jobs) + await one(keys[0], jobs[keys[0]]) + first = outcomes[keys[0]] + if len(keys) > 1 and not (first.error and not first.model): + gate = asyncio.Semaphore(BLIND_CONCURRENCY) + + async def gated(key: str) -> None: + async with gate: + await one(key, jobs[key]) + + await asyncio.gather(*(gated(key) for key in keys[1:])) + results: dict[str, Any] = {} + model: str | None = None + same = False + for key in keys: + outcome = outcomes.get(key) + if outcome is None: + results[key] = dict(results[keys[0]]) + continue results[key] = _slim(outcome) - if outcome.error and not outcome.model: - broken = results[key] for k in usage: usage[k] += outcome.usage.get(k) or 0 if outcome.model and model is None: model = outcome.model - if model in producers: - same = True - ctx.emit(EventType.LOG, level="warn", code="dual_check_same_model", model=model, - message=f"盲算模型与产出原数据的模型相同（「{model}」），难以发现原流程自身的错误。" - "请在「设置 → 偏好设置 → 核查」中选择另一个核查模型，" - f"或在节点的「{DUAL_CHECK_FIELD_LABEL['model']}」中指定") - if outcome.limit in LIMIT_CODES: - ctx.emit(EventType.LOG, level="warn", code="dual_check_limit", limit=outcome.limit, metric=job["metric_id"], - message=f"指标「{job['metric']}」的盲算{_limit_text(outcome.limit, options)}，按无法复算处理") + if model is not None and model in producers: + same = True + ctx.emit(EventType.LOG, level="warn", code="dual_check_same_model", model=model, + message=f"盲算模型与产出原数据的模型相同（「{model}」），难以发现原流程自身的错误。" + "请在「设置 → 偏好设置 → 核查」中选择另一个核查模型，" + f"或在节点的「{DUAL_CHECK_FIELD_LABEL['model']}」中指定") + for key in keys: + found = results[key] + if found.get("limit") in LIMIT_CODES and not _settled(found): + # 几个指标共用这一格：每个指标都按无法复算记，各发一条，时间线上不会只说先出现的那个 + for metric in jobs[key]["metrics"]: + ctx.emit(EventType.LOG, level="warn", code="dual_check_limit", limit=found["limit"], + metric=metric["id"], + message=f"指标「{metric['name']}」的盲算{_limit_text(found['limit'], options)}，按无法复算处理") usage["cost_usd"] = round(usage["cost_usd"], 6) return results, model, same, usage @@ -704,6 +820,13 @@ def _cell_masked(cell: dict[str, Any]) -> bool: return isinstance(recorded, list) and any(isinstance(c, str) and c.lower() == column for c in recorded) +def _settled(found: Mapping[str, Any]) -> bool: + """触了查询次数的上限，但交回的那一格是真查出来的：照常比对。verify_agent 查满 max_queries 次就记触顶（哪怕最后 + 一次正是答案），按无法复算处理的话，「每项最多查询次数」设成 1 时盲算永远核不成。时长、金额触顶的是半路被掐断， + 照旧按无法复算。""" + return found.get("limit") == "max_queries" and found.get("value") is not None + + def _limit_text(limit: str | None, options: _Options) -> str: """盲算为什么停下，接在「盲算」后面读：「盲算已达每项最多查询次数（6 次）」。""" if limit == "max_queries" and options.max_queries is not None: @@ -744,7 +867,7 @@ def _compare(row: dict[str, Any], metric: dict[str, Any], results: dict[str, Any if found is None: row.update(status=UNCHECKED, reason="盲算没有执行") return - if found.get("limit"): + if found.get("limit") and not _settled(found): row.update(status=UNCHECKED, reason="盲算" + _limit_text(found["limit"], options)) return if found.get("value") is None: @@ -780,11 +903,14 @@ def _unchecked_gaps(rows: list[dict[str, Any]]) -> list[str]: # -------------------------------------------------------------------------- -async def _review(ctx: NodeContext, checked: dict[str, Any]) -> dict[str, Any]: +async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], options: _Options) -> dict[str, Any]: """不一致的指标逐个说清差在哪，完整对照落成工件，组好审批的 payload。返回 {payload, facts, card_artifact}。 + 两条 SQL 先收窄到这一格再比（_narrowed）：原流程一条查询算好几列、分组后取一行时，整条拿去比会混进「选出的列」 + 「分组」这类和这一格无关的差别，审批卡上只列前几条，审核人看到的多半是噪声，只差一个方面的也就写不进目录。 facts 是每个不一致指标写数据目录要用的东西（dual_check_facts.derive 的候选，少一个 choice）：只有一个输入对不上时是 - 那一处完整的差异条目（带两边的写法和涉及的表、列），几个输入都对不上时只说「不止一处」。 + 那一处完整的差异条目（带两边比较的写法和涉及的表、列），几个输入都对不上时只说「不止一处」；scope 是这个指标自己的 + 范围（口径说明、所在分组、运行输入、分组的列），有效记录条件据此排除指标自己的筛选范围。 """ rows = [r for r in checked["rows"] if r["status"] == DIFFERENT] results = checked["results"] @@ -793,16 +919,31 @@ async def _review(ctx: NodeContext, checked: dict[str, Any]) -> dict[str, Any]: metrics: list[dict[str, Any]] = [] summary: list[dict[str, Any]] = [] facts: dict[str, dict[str, Any]] = {} + inputs = _run_inputs(state) for row in rows: per_input: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]] = [] + compared: list[tuple[str, str]] = [] + scope_texts = [options.definitions.get(row["metric_id"]) or "", *inputs] + scope_columns: set[str] = set() for inp in row["inputs"]: - if inp["key"] not in row["differing"]: - continue if inp["source"] not in infos: infos[inp["source"]] = await _source_info(sources, inp["source"]) + kind = infos[inp["source"]].get("kind") blind = results.get(inp["key"]) or {} - diff = await _explain(inp["sql"], blind.get("sql"), infos[inp["source"]]) + original_snap, _ = sources.snapshot(inp["artifact"]) + pairs = _row_pairs(original_snap or {}, inp["row"], inp["column"]) + scope_texts += [str(v) for _, v in pairs] + scope_columns |= {name.lower() for name, _ in pairs} | _group_columns(inp["sql"], kind) \ + | _group_columns(blind.get("sql"), kind) + if inp["key"] not in row["differing"]: + continue + blind_snap, _ = sources.snapshot(str(blind["artifact"])) if blind.get("artifact") else (None, None) + a_sql = _narrowed(inp["sql"], inp["column"], _record(original_snap, inp["row"]), kind) or inp["sql"] + b_sql = _narrowed(blind.get("sql"), blind.get("column"), _record(blind_snap, blind.get("row")), kind) \ + or blind.get("sql") + diff = await _explain(a_sql, b_sql, infos[inp["source"]]) per_input.append((inp, blind, diff)) + compared.append((a_sql, str(b_sql or ""))) merged = _merge(per_input) metrics.append({ "metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"], "decimals": row["decimals"], @@ -826,13 +967,14 @@ async def _review(ctx: NodeContext, checked: dict[str, Any]) -> dict[str, Any]: "fallback": not any(i.get("aspect") not in ("other", "parse") for i in merged["items"]), }) first = per_input[0][0] if per_input else (row["inputs"][0] if row["inputs"] else {}) - single = per_input[0] if len(per_input) == 1 else None + single = compared[0] if len(per_input) == 1 else None facts[row["metric_id"]] = { "metric_id": row["metric_id"], "metric": row["name"], "source": first.get("source"), "dialect": (infos.get(str(first.get("source"))) or {}).get("kind"), - "sql": {"a": single[0]["sql"], "b": single[1].get("sql")} if single else {}, + "sql": {"a": single[0], "b": single[1]} if single else {}, "diff": {"equivalent": merged["equivalent"], "fallback": merged["fallback"] or single is None, "items": merged["items"]}, + "scope": {"texts": [t for t in dict.fromkeys(scope_texts) if t], "columns": sorted(scope_columns)}, } content = {"metrics": metrics, "model": checked.get("model"), "same_model": bool(checked.get("same_model"))} try: @@ -889,6 +1031,147 @@ async def _explain(original: Any, blind: Any, info: dict[str, Any]) -> dict[str, return out +def _record(snapshot: Mapping[str, Any] | None, row: Any) -> dict[str, Any] | None: + """快照里第 row 行，写成 {列名: 值}；没有这一行为 None。""" + rows = (snapshot or {}).get("rows") or [] + if isinstance(row, bool) or not isinstance(row, int) or not 0 <= row < len(rows): + return None + record = rows[row] + if isinstance(record, dict): + return {str(k): v for k, v in record.items()} + columns = [str(c) for c in (snapshot or {}).get("columns") or []] + return dict(zip(columns, record)) if isinstance(record, (list, tuple)) else None + + +def _output_name(item: Any, read: str | None) -> str: + """最外层 SELECT 一项的结果列名：有别名用别名，裸列用列名，别的是驱动按原文起的名。""" + from sqlglot import exp + + if isinstance(item, exp.Alias): + return item.alias + if isinstance(item, exp.Column): + return item.name + return item.sql(dialect=read) + + +def _same_column(a: str, b: str) -> bool: + return "".join(a.split()).lower() == "".join(b.split()).lower() + + +def _group_projection(key: Any, projections: list[Any]) -> int | None: + """分组键是最外层 SELECT 的第几项：GROUP BY 1、GROUP BY 原样的表达式、GROUP BY 别名都认，认不出为 None。""" + from sqlglot import exp + + if isinstance(key, exp.Literal) and not key.is_string and str(key.this).isdigit(): + index = int(key.this) - 1 + return index if 0 <= index < len(projections) else None + for i, item in enumerate(projections): + if (item.this if isinstance(item, exp.Alias) else item) == key: + return i + if isinstance(key, exp.Column) and not key.table: + hits = [i for i, item in enumerate(projections) if isinstance(item, exp.Alias) + and item.alias.lower() == key.name.lower()] + return hits[0] if len(hits) == 1 else None + return None + + +def _sql_literal(value: Any) -> Any: + """快照里的取值写回 SQL 常量；写不回（容器、非有限数）为 None。""" + from sqlglot import exp + + if isinstance(value, bool): + return exp.Boolean(this=value) + if isinstance(value, (int, float)): + return exp.Literal.number(value) if math.isfinite(value) else None + if isinstance(value, str): + return exp.Literal.string(value) + return None + + +def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str | None) -> str | None: + """把一条查询收窄到这一格：最外层 SELECT 只留算出这一列的那一项；分组的去掉分组，按这一行的分组值补上等值条件 + （华东那一行 → WHERE … AND region = '华东'），排序、取行数跟着去掉。 + + 差异说明拿收窄后的两条比，才只说这一格是怎么算出来的。认不准的（SELECT *、DISTINCT、窗口函数、集合运算、分组键 + 对不上结果列、这一行没有分组值）返回 None，调用方照整条比。""" + import sqlglot + from sqlglot import exp + + from app.data.sqlcheck import dialect_of + + if not isinstance(sql, str) or not sql.strip() or not isinstance(column, str) or not column.strip(): + return None + read = dialect_of(kind) + try: + tree = sqlglot.parse_one(sql, read=read) + except Exception: # noqa: BLE001 - 解析不了：照整条比，比较本身会说清 + return None + if not isinstance(tree, exp.Select) or tree.args.get("distinct"): + return None + projections = list(tree.expressions) + if not projections or any(isinstance(p, exp.Star) or (isinstance(p, exp.Column) and isinstance(p.this, exp.Star)) + or p.find(exp.Window) for p in projections): + return None + hits = [i for i, p in enumerate(projections) if _same_column(_output_name(p, read), column)] + if len(hits) != 1: + return None + target = projections[hits[0]] + group = tree.args.get("group") + if group is not None: + if record is None or any(group.args.get(k) for k in ("rollup", "cube", "grouping_sets", "totals")): + return None + values = {k.lower(): v for k, v in record.items()} + conditions = [] + for key in group.expressions: + index = _group_projection(key, projections) + name = _output_name(projections[index], read).lower() if index is not None else None + if name is None or name not in values: + return None + expr = projections[index].this if isinstance(projections[index], exp.Alias) else projections[index] + if values[name] is None: + conditions.append(exp.Is(this=expr.copy(), expression=exp.Null())) + continue + literal = _sql_literal(values[name]) + if literal is None: + return None + conditions.append(exp.EQ(this=expr.copy(), expression=literal)) + for arg in ("group", "order", "limit", "offset"): + tree.set(arg, None) + tree = tree.where(*conditions, copy=False) + elif tree.args.get("limit") is None and tree.args.get("offset") is None: + # 不取前几行时，排序不影响算出来的数 + tree.set("order", None) + tree.set("expressions", [target]) + try: + return tree.sql(dialect=read) + except Exception: # noqa: BLE001 + return None + + +def _group_columns(sql: Any, kind: str | None) -> set[str]: + """一条查询分组用到的列（小写）：有效记录条件不收分组列上的条件（那是指标自己的筛选范围）。""" + import sqlglot + from sqlglot import exp + + from app.data.sqlcheck import dialect_of + + if not isinstance(sql, str) or not sql.strip(): + return set() + read = dialect_of(kind) + try: + tree = sqlglot.parse_one(sql, read=read) + except Exception: # noqa: BLE001 + return set() + out: set[str] = set() + for group in tree.find_all(exp.Group): + select = group.parent if isinstance(group.parent, exp.Select) else None + for key in group.expressions: + index = _group_projection(key, list(select.expressions)) if select is not None else None + node = select.expressions[index] if index is not None else key + out |= {col.name.lower() for col in node.find_all(exp.Column) if col.name} + return out + + def _merge(per_input: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]]) -> dict[str, Any]: """一个指标几个对不上的输入的差异合成一份：一个输入时就是它的；几个时每条前面写明是哪一列的输入。""" if not per_input: @@ -914,14 +1197,15 @@ def _merge(per_input: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]] def _choices(ctx: NodeContext, response: Any, rows: list[dict[str, Any]]) -> tuple[dict[str, str], str]: """审核人的回复 → ({指标 id: a | b | neither}, 备注)。回复的 value 是 {choices, note}；某个指标没选、选了认不出的 - 按「两边都不对」：不替人认可任何一边。明确驳回（approved 为否）的按人工驳回让节点失败，和人工审批节点的填写模式 - 一样：这一步没有驳回的出口。""" + 不进 choices：照实记成「没有作出选择」，不替人说「两边都不对」，也不替人认可任何一边（缺口照样记、出具照样降档）。 + 通用的输入卡送来的是一句话，所有指标都按没选记。明确驳回（approved 为否）的按人工驳回让节点失败，和人工审批节点的 + 填写模式一样：这一步没有驳回的出口。""" decision = read_decision(response) if isinstance(response, dict) and decision.rejected: raise NodeError(ctx.node.id, f"人工驳回：{decision.note or '未说明原因'}") value = decision.value if isinstance(decision.value, dict) else response if isinstance(response, dict) else {} raw = value.get("choices") if isinstance(value.get("choices"), dict) else {} - choices = {r["metric_id"]: raw[r["metric_id"]] if raw.get(r["metric_id"]) in CHOICES else "neither" for r in rows} + choices = {r["metric_id"]: raw[r["metric_id"]] for r in rows if raw.get(r["metric_id"]) in CHOICES} note = value.get("note") if isinstance(value.get("note"), str) and value.get("note").strip() else decision.note return choices, str(note or "").strip() @@ -929,12 +1213,12 @@ def _choices(ctx: NodeContext, response: Any, rows: list[dict[str, Any]]) -> tup def _gaps(rows: list[dict[str, Any]], choices: dict[str, str]) -> list[str]: out: list[str] = [] for row in rows: + choice = choices.get(row["metric_id"]) if row["status"] == UNCHECKED: out.append(GAP_UNCHECKED.format(name=row["name"], reason=row.get("reason") or "原因不明")) - elif row["status"] == DIFFERENT and choices.get(row["metric_id"]) == "b": - out.append(GAP_BLIND.format(name=row["name"])) - elif row["status"] == DIFFERENT and choices.get(row["metric_id"]) != "a": - out.append(GAP_NEITHER.format(name=row["name"])) + elif row["status"] == DIFFERENT and choice != "a": + text = GAP_BLIND if choice == "b" else GAP_NEITHER if choice == "neither" else GAP_UNANSWERED + out.append(text.format(name=row["name"])) return out @@ -942,7 +1226,8 @@ _CHOICE_TEXT = {"a": "采用原流程", "b": "采用盲算", "neither": "两边 def _resolved_text(rows: list[dict[str, Any]], choices: dict[str, str]) -> str: - parts = "；".join(f"「{r['name']}」{_CHOICE_TEXT[choices[r['metric_id']]]}" for r in rows[:8]) + parts = "；".join(f"「{r['name']}」{_CHOICE_TEXT.get(choices.get(r['metric_id']), '没有作出选择')}" + for r in rows[:8]) return f"审核人已裁决 {len(rows)} 个不一致的指标：{parts}" + ("等" if len(rows) > 8 else "") @@ -974,6 +1259,9 @@ def _finish(state: GraphState, ctx: NodeContext, card_id: str, card: dict[str, A artifact: str | None = None, catalog: dict[str, Any] | None = None) -> dict[str, Any]: rows = checked["rows"] picked = choices or {} + counts = {"agreed": sum(1 for r in rows if r["status"] == SAME), + "disagreed": sum(1 for r in rows if r["status"] == DIFFERENT), + "unchecked": sum(1 for r in rows if r["status"] == UNCHECKED)} metrics = [{ "metric_id": r["metric_id"], "name": r["name"], "unit": r["unit"], "status": r["status"], "original_rendered": r["original"]["rendered"], "blind_rendered": r["blind"]["rendered"], @@ -981,7 +1269,7 @@ def _finish(state: GraphState, ctx: NodeContext, card_id: str, card: dict[str, A **({"reason": r["reason"]} if r.get("reason") else {}), **({"choice": picked[r["metric_id"]]} if r["metric_id"] in picked else {}), } for r in rows] - result: dict[str, Any] = {"status": status, "metrics": metrics, "model": checked.get("model"), + result: dict[str, Any] = {"status": status, "counts": counts, "metrics": metrics, "model": checked.get("model"), "same_model": bool(checked.get("same_model"))} if artifact: result["card_artifact"] = artifact @@ -993,7 +1281,7 @@ def _finish(state: GraphState, ctx: NodeContext, card_id: str, card: dict[str, A result["issuance_gaps"] = gaps entry: dict[str, Any] = { "kind": "dual_check", "node_id": ctx.node.id, "exec": next_exec(state, ctx.node.id), "card": card_id, - "status": status, + "status": status, "counts": counts, "metrics": [{"id": m["metric_id"], "status": m["status"], **({"choice": m["choice"]} if "choice" in m else {})} for m in metrics], **({"metric_set": card["artifact"]} if card and isinstance(card.get("artifact"), str) else {}), @@ -1006,4 +1294,4 @@ def _finish(state: GraphState, ctx: NodeContext, card_id: str, card: dict[str, A return updates -__all__ = ["AGREED", "CHOICES", "RESOLVED", "SKIPPED", "run_dual_check"] +__all__ = ["AGREED", "CHOICES", "NONE_CHECKED", "PARTIAL", "RESOLVED", "SKIPPED", "run_dual_check"] diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index 306f5a2a..caf383cb 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -455,16 +455,29 @@ def dual_check_card(spec: GraphSpec, node: GraphNode) -> tuple[str | None, str | def _check_dual_check(node: GraphNode, spec: GraphSpec, result: ValidationResult) -> None: """双盲复算的配置：口径卡必须是它上游的口径卡；复算的指标、口径说明的键必须是那张卡上的指标 ID；开关、 - 误差、处置的取值。指标定义从别的工作流引用来的口径卡（caliber_from），画图时不知道有哪些指标，不查指标 ID。""" + 误差、处置的取值；不能配失败重试。指标定义从别的工作流引用来的口径卡（caliber_from），画图时不知道有哪些指标， + 不查指标 ID。""" cfg = node.config label = DUAL_CHECK_FIELD_LABEL def add(message: str, field: str, code: str) -> None: result.add(message, node_id=node.id, field=field, code=code) - if cfg.get("enabled") not in (None, True, False): - add(f"「{label['enabled']}」只能选跟随设置、开启或关闭，当前为「{cfg.get('enabled')}」", "enabled", + switch = cfg.get("enabled") + # 用 is 比：0 == False、1 == True，用 in 比会放过去，执行时却按「跟随设置」走（执行器认的是 True / False 本身） + if not (switch is None or switch is True or switch is False): + add(f"「{label['enabled']}」只能选跟随设置、开启或关闭，当前为「{switch}」", "enabled", "dual_check.option_invalid") + try: + retries = int(cfg.get("retries", 0) or 0) # 和 compiler 取重试次数的写法一样 + except (TypeError, ValueError): + retries = 0 + if retries > 0: + # 失败都是确定的（审核人没采用原流程、口径卡配错）：重试时 @task 的调用序号往后走、对不上缓存，盲算整轮 + # 再做一遍、再花一次钱，还要审核人再选一次 + add(f"「双盲复算」节点不支持「{field_label('retries')}」：它的失败是确定的（审核人没有采用原流程的结果、" + "口径卡配置有误），重试只会把盲算再做一遍并再次请审核人选择。请把失败重试次数设为 0", "retries", + "dual_check.retries") if cfg.get("on_reject") not in (None, "", *DUAL_CHECK_ON_REJECT): allowed = "或".join(f"「{v}」" for v in DUAL_CHECK_ON_REJECT_LABEL.values()) add(f"「{label['on_reject']}」只能选{allowed}，当前为「{cfg.get('on_reject')}」", "on_reject", diff --git a/backend/tests/fixtures/dual_check_shop.py b/backend/tests/fixtures/dual_check_shop.py index 215b09ac..7ac6bff7 100644 --- a/backend/tests/fixtures/dual_check_shop.py +++ b/backend/tests/fixtures/dual_check_shop.py @@ -7,6 +7,8 @@ - 按支付日、不排除退款：1、2、3、5 号单，380.0 - 按下单日、不排除退款：2、3、4、5 号单，360.0 - 付款客户数（按支付日、排除退款）：按行计数 3，去重计数 2 + +另有渠道（channel）和测试单标记（is_test，都是 0）两列：写数据目录的测试拿它们凑「指标自己的筛选范围」和标记列。 """ from __future__ import annotations @@ -23,11 +25,11 @@ from app.db.base import SessionLocal from app.db.models import DataSource ROWS = [ - (1, "华东", 101, 100.0, "paid", "2026-08-31", "2026-09-01"), - (2, "华东", 102, 200.0, "paid", "2026-09-02", "2026-09-02"), - (3, "华南", 101, 50.0, "refunded", "2026-09-03", "2026-09-03"), - (4, "华南", 103, 80.0, "paid", "2026-09-07", "2026-09-08"), - (5, "华北", 102, 30.0, "paid", "2026-09-05", "2026-09-05"), + (1, "华东", 101, 100.0, "paid", "2026-08-31", "2026-09-01", "online", 0), + (2, "华东", 102, 200.0, "paid", "2026-09-02", "2026-09-02", "offline", 0), + (3, "华南", 101, 50.0, "refunded", "2026-09-03", "2026-09-03", "online", 0), + (4, "华南", 103, 80.0, "paid", "2026-09-07", "2026-09-08", "online", 0), + (5, "华北", 102, 30.0, "paid", "2026-09-05", "2026-09-05", "offline", 0), ] WEEK = "BETWEEN '2026-09-01' AND '2026-09-07'" #: 口径说明要的写法：按支付日、排除退款 @@ -52,8 +54,8 @@ def build(path: Path) -> Path: db = sqlite3.connect(path) db.executescript( "CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, customer_id INTEGER, amount REAL, status TEXT, " - "order_date TEXT, paid_date TEXT);") - db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?, ?)", ROWS) + "order_date TEXT, paid_date TEXT, channel TEXT, is_test INTEGER);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ROWS) db.commit() db.close() return path @@ -80,18 +82,22 @@ def _text(messages) -> str: class Blind: """盲算模型的剧本，也管图里别的模型调用。 - plans：{任务里出现的指标名: (SQL, 要交的列) 或 [SQL, …]、列}。盲算循环里依次查这几条 SQL，收尾时交最后一条查询的 - 第 0 行那一列（row 可改）。没对上指标名的盲算交 null。不是盲算的调用（报告撰写）返回 writer 的正文并记下收到的消息。 - calls 记下每次盲算调用（step / final），tasks 记下盲算收到的任务原文。 + plans：{任务里出现的指标名: (SQL, 要交的列) 或 [SQL, …]、列}。键写成「指标名|输入2」时只认计算式里那一项的盲算 + （一个指标有几个输入来自查询时用）。盲算循环里依次查这几条 SQL，收尾时交最后一条查询的第 0 行那一列（row 可改）。 + 没对上指标名的盲算交 null。不是盲算的调用（报告撰写）返回 writer 的正文并记下收到的消息。 + calls 记下每次盲算调用（step / final），tasks 记下盲算收到的任务原文（几个盲算同时进行时也是每个任务一条）。 + on_query：盲算第一次查库之前调用一次（比如往库里插一行，造出「两次取数之间数据变了」）。 """ - def __init__(self, monkeypatch, source: str, plans: dict, *, writer: str = "本周销售额见口径卡。", row: int = 0): + def __init__(self, monkeypatch, source: str, plans: dict, *, writer: str = "本周销售额见口径卡。", row: int = 0, + on_query=None): from app.providers import mock_model self.source = source self.plans = plans self.writer = writer self.row = row + self.on_query = on_query self.calls: list[str] = [] self.tasks: list[str] = [] self.writer_prompts: list[str] = [] @@ -114,11 +120,14 @@ class Blind: blind.calls.append("step") # 盲算的任务是第一条用户消息；同一次盲算的几轮收到的是同一条，只记一次 task_text = next((str(m.content) for m in messages if m.type == "human"), "") - if not blind.tasks or blind.tasks[-1] != task_text: + if task_text not in blind.tasks: blind.tasks.append(task_text) done = sum(isinstance(m, ToolMessage) for m in messages) sqls = blind.sqls(plan) if plan is not None else [] if done < len(sqls): + if blind.on_query is not None: + hook, blind.on_query = blind.on_query, None + hook() call = {"name": f"db_query__{blind.source}", "args": {"sql": sqls[done]}, "id": f"blind-{uuid.uuid4().hex[:6]}"} return AIMessage(content="查一下。", tool_calls=[call]) @@ -129,8 +138,9 @@ class Blind: monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) def plan_of(self, text: str): - for name, plan in self.plans.items(): - if f"指标：{name}" in text: + for key, plan in self.plans.items(): + name, _, label = key.partition("|") + if f"指标：{name}" in text and (not label or f"计算式里的 {label}" in text): return plan return None diff --git a/backend/tests/test_dual_check_config.py b/backend/tests/test_dual_check_config.py index c88696bb..2da4c559 100644 --- a/backend/tests/test_dual_check_config.py +++ b/backend/tests/test_dual_check_config.py @@ -1,9 +1,9 @@ """双盲复算的配置、发起时的设置快照、助手的节点说明。 - 画布校验（engine/schema.py）：口径卡必须是上游的口径卡（没写时上游只能有一张），复算的指标、口径说明的键必须是 - 那张卡上的指标 ID，开关、误差、处置的取值；报错写检查器上的中文叫法 -- 发起（engine/launch.py）：图里有双盲复算或反方审稿节点时把核查设置快照进 options["verify"]，没有时 options 不变、 - run.started 和以前逐字相同；调用方给了的照它的 + 那张卡上的指标 ID，开关（只认 true / false，0、1 不算）、误差、处置的取值，不能配失败重试；报错写检查器上的中文叫法 +- 发起（engine/launch.py）：图里（含子工作流）有双盲复算或反方审稿节点时把核查设置快照进 options["verify"]，没有时 + options 不变、run.started 和以前逐字相同；调用方给了的照它的 - 助手的节点说明（api/copilot.py 的 NODE_REFERENCE）有 dual_check 一条：什么时候加、各配置项 表名、数字都是虚构的。 @@ -16,7 +16,7 @@ import pytest from sqlalchemy import delete, select from app.db.base import SessionLocal -from app.db.models import Run, RunEvent, Setting +from app.db.models import Run, RunEvent, Setting, Workflow, WorkflowVersion from app.engine.runner import run_manager from app.engine.schema import GraphSpec, validate_graph @@ -97,6 +97,20 @@ def test_option_values_are_checked(): assert "「允许的相对误差」需要填写不小于 0 的数" in found["tolerance"] +@pytest.mark.parametrize("value", [0, 1, 0.0]) +def test_zero_and_one_are_not_a_switch(value): + """0 == False、1 == True：用 in 比会放过去，执行时却按「跟随设置」走（执行器认的是 True / False 本身）。""" + [(where, text)] = problems(spec(CARD, dual={"enabled": value})) + assert where == "enabled" and "「启用双盲复算」" in text + + +def test_retries_are_refused(): + """双盲复算失败都是确定的（审核人没采用原流程、口径卡不对）：重试只会把盲算再做一遍、再问审核人一次。""" + [(where, text)] = problems(spec(CARD, dual={"retries": 1})) + assert where == "retries" and "「失败重试次数」" in text and "双盲复算" in text + assert problems(spec(CARD, dual={"retries": 0})) == [] + + # -------------------------------------------------------------------------- # 发起：设置快照 # -------------------------------------------------------------------------- @@ -164,6 +178,33 @@ async def test_a_challenge_node_also_snapshots_the_settings(engine): assert set(out) == {"verify"} and out["verify"]["challenge"]["enabled"] is True +async def test_a_verify_node_inside_a_sub_workflow_also_snapshots_the_settings(engine): + """子工作流里的双盲复算同样在运行中途读设置：按运行时的取法找子工作流（固定了版本的读那一版），任何一层有就快照。""" + from app.engine.launch import _with_verify_snapshot + + inner_v1 = {"nodes": [node("start", "input"), node("out", "output", fields=[])], "edges": []} + inner_v2 = {"nodes": [node("start", "input"), node("review", "dual_check")], "edges": []} + async with SessionLocal() as session: + wf = Workflow(name="子流程-复核", graph=inner_v1, version=2) + session.add(wf) + await session.flush() + session.add_all([WorkflowVersion(workflow_id=wf.id, version=1, graph=inner_v1), + WorkflowVersion(workflow_id=wf.id, version=2, graph=inner_v2)]) + await session.commit() + wid = wf.id + + def outer(**config) -> dict: + return {"nodes": [node("start", "input"), node("sub", "subgraph", workflow_id=wid, **config)], "edges": []} + + async with SessionLocal() as session: + pinned = await _with_verify_snapshot(session, outer(workflow_version=2), None) + old_pin = await _with_verify_snapshot(session, outer(workflow_version=1), None) + current = await _with_verify_snapshot(session, outer(), None) + assert set(pinned) == {"verify"} and pinned["verify"]["dual_check"]["enabled"] is True + # 固定在没有核查节点的那一版、或者读工作流现在的图（也没有）：options 原样不动 + assert old_pin is None and current is None + + async def test_without_verify_nodes_options_and_run_started_are_unchanged(engine): from app.engine.launch import launch_run diff --git a/backend/tests/test_dual_check_executor.py b/backend/tests/test_dual_check_executor.py index 7a1f3736..be7029d5 100644 --- a/backend/tests/test_dual_check_executor.py +++ b/backend/tests/test_dual_check_executor.py @@ -158,3 +158,52 @@ async def test_unknown_metric_ids_fail_the_node(shop): ctx = context(graph) with pytest.raises(NodeError, match="「复算的指标」中的「refund」不是口径卡上的指标 ID"): await run_dual_check({"nodes": {"card": card_of(metric(art))}, "evidence": [], "trail": []}, ctx) + + +async def test_constant_columns_of_the_original_query_are_not_shown(monkeypatch, shop): + """原 SQL 里写死的文字列（'下单日' AS basis）是原流程的取舍，不是分组：不交给盲算。""" + sql = "SELECT '下单日' AS basis, '华东' AS region, SUM(amount) AS gmv FROM orders WHERE region = '华东'" + art = await snapshot(shop, columns=("basis", "region", "gmv"), rows=(("下单日", "华东", 330.0),), sql=sql) + blind = Blind(monkeypatch, shop, {"销售额": (RIGHT, "gmv")}) + ctx = context(AGENT_GRAPH) + item = metric(art, path="cell(nodes.fetch, 0, 'gmv')", via="tool_cell", node_id="fetch", + locator={"row": 0, "column": "gmv"}) + state = {"nodes": {"agent": {"model": "other"}, "card": card_of(item)}, "evidence": [], "trail": []} + await run_dual_check(state, ctx) + [task] = blind.tasks + assert "下单日" not in task and "basis" not in task and "华东" not in task + + +async def test_blind_checks_run_side_by_side_after_the_first(monkeypatch, shop): + """几个盲算同时进行（有上限），先单独试第一个：核查模型用不了时不必个个去试。结果按格子对回各自的指标。""" + import asyncio + + from app.engine import verify_agent + from app.engine.nodes import dual_check + + art = await snapshot(shop, columns=("gmv",), rows=tuple((float(100 + i),) for i in range(5))) + running = {"now": 0, "peak": 0, "order": []} + + async def fake(**kwargs): + running["now"] += 1 + running["peak"] = max(running["peak"], running["now"]) + running["order"].append(kwargs["task"]) + await asyncio.sleep(0.05) + running["now"] -= 1 + return verify_agent.VerifyOutcome(result=None, cells=[], queries=[], rationale=None, + usage={"calls": 1, "cost_usd": 0.01}, error="没有交结果", model="mock-fast") + + monkeypatch.setattr(verify_agent, "run_verify_agent", fake) + metrics = [] + for i in range(5): + item = metric(art, locator={"row": i, "column": "gmv"}, value=float(100 + i)) + metrics.append({**item, "id": f"m{i}", "name": f"第 {i} 组销售额", "value": float(100 + i), + "rendered": f"{100 + i}.0元", "substituted": f"{100 + i}.0"}) + ctx = context(AGENT_GRAPH) + state = {"nodes": {"agent": {"model": "other"}, "card": card_of(*metrics)}, "evidence": [], "trail": []} + updates = await run_dual_check(state, ctx) + out = updates["nodes"]["dual"] + assert [m["status"] for m in out["metrics"]] == ["unchecked"] * 5 + assert 1 < running["peak"] <= dual_check.BLIND_CONCURRENCY + assert "第 0 组销售额" in running["order"][0] + assert updates["usage"]["calls"] == 5 and updates["usage"]["cost_usd"] == 0.05 diff --git a/backend/tests/test_dual_check_facts.py b/backend/tests/test_dual_check_facts.py index 6ecd32a2..9f263441 100644 --- a/backend/tests/test_dual_check_facts.py +++ b/backend/tests/test_dual_check_facts.py @@ -1,10 +1,13 @@ """双盲复算的选择写进数据目录（engine/dual_check_facts.py）。 -- 只差一个方面、又对得上目录事实时才写：时间列 → 业务日期，一边多一个只涉及一张表的筛选条件 → 有效记录条件， - 聚合只差去不去重 → 去重规则；按被采纳那一边的写法记 -- 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法：不写，照实说原因 -- 写的是人工填写、已确认，备注写明运行、节点、指标；同样的说法已确认过就不再写（不升版本），确认过的不同说法不覆盖， - 推断的不同说法照改 +- 只差一个方面、又对得上目录事实时才写：两边按不同的时间列筛 → 业务日期，一边多一个排除无效记录的条件 → 有效记录 + 条件，聚合只差去不去重、被采纳的一边去重 → 去重规则；按被采纳那一边的写法记 +- 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法、只有一边按时间筛：不写， + 照实说原因 +- 指标自己的筛选范围（只看华东、只看线上）不是有效记录条件：取值出现在指标名称、口径说明、所在分组、运行输入里的， + 条件用到分组的列的，不是在排除退款、作废、测试这类无效记录的，都不写 +- 写的是人工填写、已确认，备注写明运行、节点、指标；同样的说法已确认过就不再写（不升版本），确认过的不同说法不覆盖； + 目录里已有的有效记录条件被采纳的一边也用着的，合起来写，没用着的不动；同一次裁决里互相矛盾的选择都不写 差异条目一律由 sql_semantics.explain_sql_diff 真算出来，和节点交给审核人看的是同一份。表名、数字都是虚构的。 """ @@ -22,22 +25,43 @@ from tests.fixtures.dual_check_shop import ( BUYERS_DISTINCT, BY_ORDER_DATE, RIGHT, + WEEK, WITH_REFUNDS, add_source, build, ) from tests.fixtures.sources import drop_source -SCHEMA = {"tables": {"orders": {"name": "orders", "columns": [ - {"name": "id", "type": "INTEGER"}, {"name": "region", "type": "TEXT"}, {"name": "customer_id", "type": "INTEGER"}, - {"name": "amount", "type": "REAL"}, {"name": "status", "type": "TEXT"}, {"name": "order_date", "type": "TEXT"}, - {"name": "paid_date", "type": "TEXT"}], "primary_key": ["id"]}}} +SCHEMA = {"tables": { + "orders": {"name": "orders", "columns": [ + {"name": "id", "type": "INTEGER"}, {"name": "region", "type": "TEXT"}, + {"name": "customer_id", "type": "INTEGER"}, {"name": "amount", "type": "REAL"}, + {"name": "status", "type": "TEXT"}, {"name": "order_date", "type": "TEXT"}, + {"name": "paid_date", "type": "TEXT"}, {"name": "channel", "type": "TEXT"}, + {"name": "is_test", "type": "INTEGER"}], "primary_key": ["id"]}, + "customers": {"name": "customers", "columns": [ + {"name": "id", "type": "INTEGER"}, {"name": "level", "type": "TEXT"}, {"name": "status", "type": "TEXT"}], + "primary_key": ["id"]}, +}} +#: 不带时间条件、只排除退款：和 RIGHT 只差「按不按时间筛」 +NO_TIME = "SELECT SUM(amount) AS gmv FROM orders WHERE status <> 'refunded'" +COUNT_WEEK = f"SELECT COUNT(*) AS n FROM orders WHERE paid_date {WEEK}" +SUM_WEEK = f"SELECT SUM(amount) AS gmv FROM orders WHERE paid_date {WEEK}" -def candidate(original: str, blind: str, choice: str, *, metric: str = "销售额", source: str = "shop") -> dict: +def candidate(original: str, blind: str, choice: str, *, metric: str = "销售额", source: str = "shop", + scope: dict | None = None) -> dict: diff = explain_sql_diff(original, blind, "sqlite", schema=SCHEMA, labels=("原流程", "盲算")).as_dict() - return {"metric": metric, "choice": choice, "source": source, "dialect": "sqlite", - "sql": {"a": original, "b": blind}, "diff": diff} + out = {"metric": metric, "choice": choice, "source": source, "dialect": "sqlite", + "sql": {"a": original, "b": blind}, "diff": diff} + if scope is not None: + out["scope"] = scope + return out + + +def extra(condition: str, base: str = SUM_WEEK) -> str: + """base 多一个条件。""" + return base.replace(" WHERE ", f" WHERE {condition} AND ", 1) # -------------------------------------------------------------------------- @@ -52,12 +76,24 @@ def test_a_different_time_column_becomes_the_business_date(choice, column): assert (fact.table, fact.item, fact.value, fact.choice) == ("orders", "business_date", {"column": column}, choice) +def test_a_time_filter_on_one_side_only_is_not_a_business_date(): + """一边按支付日筛、另一边根本没按时间筛：审核人判断的是要不要限定时间，没在两列之间选过。""" + picked = candidate(RIGHT, NO_TIME, "a") + assert [i["aspect"] for i in picked["diff"]["items"]] == ["time"] + fact, why = derive(picked, SCHEMA) + assert fact is None and "只有被采纳的原流程按时间筛选" in why and "不是按哪一列归属" in why + fact, why = derive(candidate(RIGHT, NO_TIME, "b"), SCHEMA) + assert fact is None and "被采纳的盲算没有按时间筛选" in why + + def test_an_extra_filter_on_the_adopted_side_becomes_the_valid_filter(): fact, why = derive(candidate(WITH_REFUNDS, RIGHT, "b"), SCHEMA) assert why is None and (fact.item, fact.value) == ("valid_filter", "status <> 'refunded'") # 原流程多出这个条件、审核人采纳原流程：同样记这个条件 fact, why = derive(candidate(RIGHT, WITH_REFUNDS, "a"), SCHEMA) assert why is None and (fact.item, fact.value) == ("valid_filter", "status <> 'refunded'") + # 被采纳那一边的写法跟着记下来，有效记录条件合并时要看它还用了哪些条件 + assert fact.sql == RIGHT and fact.dialect == "sqlite" def test_the_adopted_side_without_the_extra_filter_has_nothing_to_record(): @@ -65,6 +101,63 @@ def test_the_adopted_side_without_the_extra_filter_has_nothing_to_record(): assert fact is None and "被采纳的盲算没有这个条件" in why and "销售额" in why +@pytest.mark.parametrize("original, blind, choice, metric", [ + # 原流程按指标名只看华东，盲算漏了：审核人自然选原流程 + (extra("region = '华东'", RIGHT), RIGHT, "a", "华东销售额"), + # 盲算按指标名只数线上的单 + (COUNT_WEEK, extra("channel = 'online'", COUNT_WEEK), "b", "线上订单数"), + # 只保留一部分取值的写法，状态列上也一样 + (COUNT_WEEK, extra("status IN ('paid')", COUNT_WEEK), "b", "已支付订单数"), + (SUM_WEEK, extra("(region = '华东' OR region = '华南')"), "b", "南方销售额"), + # 排除的写法，排除的却不是无效记录 + (COUNT_WEEK, extra("channel <> 'offline'", COUNT_WEEK), "b", "线上订单数"), + (SUM_WEEK, extra("region <> '华东'"), "b", "非华东销售额"), + (SUM_WEEK, extra("status <> 'paid'"), "b", "未支付订单金额"), + (SUM_WEEK, extra("customer_id IS NOT NULL"), "b", "会员销售额"), +]) +def test_a_metric_scope_filter_is_not_a_valid_filter(original, blind, choice, metric): + picked = candidate(original, blind, choice, metric=metric) + assert [i["aspect"] for i in picked["diff"]["items"]] == ["filter"] + fact, why = derive(picked, SCHEMA) + assert fact is None and "可能是指标自己的筛选范围" in why and metric in why + + +def test_the_metric_scope_is_read_from_the_names_groups_and_inputs(): + """排除的写法、排除的也像无效记录，但取值出现在口径说明、所在分组、运行输入里，或者用的是分组的列：拿不准，不写。""" + blind = extra("channel <> 'test'") + assert derive(candidate(SUM_WEEK, blind, "b"), SCHEMA)[0] is not None + for scope in ({"texts": ["只看正式渠道：不含 test 渠道的订单"]}, {"texts": ["channel = test"]}, + {"columns": ["channel"]}): + fact, why = derive(candidate(SUM_WEEK, blind, "b", scope=scope), SCHEMA) + assert fact is None and "可能是指标自己的筛选范围" in why, scope + + +@pytest.mark.parametrize("condition, written", [ + ("is_test = 0", "is_test = 0"), + ("is_test <> 1", "is_test <> 1"), + ("status NOT IN ('refunded', 'void')", "NOT status IN ('refunded', 'void')"), + ("status <> '已作废'", "status <> '已作废'"), +]) +def test_conditions_that_drop_invalid_rows_are_valid_filters(condition, written): + fact, why = derive(candidate(SUM_WEEK, extra(condition), "b"), SCHEMA) + assert why is None and (fact.item, fact.value) == ("valid_filter", written) + + +@pytest.mark.parametrize("condition", ["is_test = 1", "is_test <> 0", "status <> 9"]) +def test_conditions_that_keep_invalid_rows_or_unknown_codes_are_not(condition): + fact, why = derive(candidate(SUM_WEEK, extra(condition), "b"), SCHEMA) + assert fact is None and "可能是指标自己的筛选范围" in why + + +def test_a_code_the_catalog_calls_void_counts_as_an_invalid_row(): + notes = {"orders": {"columns": {"status": {"codes": catalog.make_item({"9": "作废", "1": "已支付"}, "human")}}}} + fact, why = derive(candidate(SUM_WEEK, extra("status <> 9"), "b"), SCHEMA, notes) + assert why is None and fact.value == "status <> 9" + # 码值表里写的是正常的状态：排除它就是指标自己的范围 + fact, why = derive(candidate(SUM_WEEK, extra("status <> 1"), "b"), SCHEMA, notes) + assert fact is None and "可能是指标自己的筛选范围" in why + + def test_table_names_are_dropped_from_the_recorded_filter(): qualified = RIGHT.replace("FROM orders WHERE status", "FROM orders o WHERE o.status") \ .replace("paid_date", "o.paid_date") @@ -72,10 +165,24 @@ def test_table_names_are_dropped_from_the_recorded_filter(): assert why is None and fact.value == "status <> 'refunded'" -@pytest.mark.parametrize("choice, text", [("b", "对 customer_id 计数时去重"), ("a", "对 customer_id 计数时不去重")]) -def test_counting_with_or_without_distinct_becomes_the_dedup_rule(choice, text): - fact, why = derive(candidate(BUYERS, BUYERS_DISTINCT, choice, metric="付款客户数"), SCHEMA) - assert why is None and (fact.table, fact.item, fact.value) == ("orders", "dedup", text) +def test_a_condition_across_two_tables_is_not_recorded(): + joined = (f"SELECT SUM(o.amount) AS gmv FROM orders o JOIN customers c ON o.customer_id = c.id " + f"WHERE o.paid_date {WEEK}") + picked = candidate(joined, f"{joined} AND o.status <> c.status", "b") + assert [i["aspect"] for i in picked["diff"]["items"]] == ["filter"] + fact, why = derive(picked, SCHEMA) + assert fact is None and "不是只涉及一张表的列" in why + + +def test_counting_distinct_becomes_the_dedup_rule(): + fact, why = derive(candidate(BUYERS, BUYERS_DISTINCT, "b", metric="付款客户数"), SCHEMA) + assert why is None and (fact.table, fact.item, fact.value) == ("orders", "dedup", "对 customer_id 计数时去重") + + +def test_counting_without_distinct_is_not_a_dedup_rule(): + """采纳了不去重的一边：那是这个指标自己的算法（数行），写成表的「不去重」会把别的指标带偏。""" + fact, why = derive(candidate(BUYERS, BUYERS_DISTINCT, "a", metric="付款笔数"), SCHEMA) + assert fact is None and "被采纳的原流程没有去重" in why def test_more_than_one_difference_is_not_recorded(): @@ -165,9 +272,48 @@ async def test_a_confirmed_different_item_is_kept(shop): assert entry.version == 1 and entry.notes["business_date"]["value"]["column"] == "order_date" -async def test_an_inferred_different_item_is_replaced(shop): +async def test_an_inferred_condition_both_sides_use_is_extended(shop): + """目录里推断的「is_test = 0」两边都按它筛了，盲算另外排除了退款：合起来写，原有的条件不丢。""" + name, source_id = shop + await seed(source_id, {"valid_filter": catalog.make_item("is_test = 0", "llm")}) + base = f"SELECT SUM(amount) AS gmv FROM orders WHERE is_test = 0 AND paid_date {WEEK}" + out = await record_choices([candidate(base, extra("status <> 'refunded'", base), "b", source=name)], + run_id="run-1", node_title="复核", actor=None) + assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out + item = (await entry_of(source_id)).notes["valid_filter"] + assert item["value"] == "is_test = 0 AND status <> 'refunded'" + assert (item["source"], item["status"]) == ("human", "confirmed") + + +async def test_a_confirmed_condition_is_not_extended(shop): + name, source_id = shop + await seed(source_id, {"valid_filter": catalog.make_item("is_test = 0", "human", "confirmed")}) + base = f"SELECT SUM(amount) AS gmv FROM orders WHERE is_test = 0 AND paid_date {WEEK}" + out = await record_choices([candidate(base, extra("status <> 'refunded'", base), "b", source=name)], + run_id="run-1", node_title="复核", actor=None) + assert out["written"] == [] + reason = out["skipped"][0]["reason"] + assert "已有人工确认过的说法「is_test = 0」" in reason and "status <> 'refunded'" in reason and "未改动" in reason + entry = await entry_of(source_id) + assert entry.version == 1 and entry.notes["valid_filter"]["value"] == "is_test = 0" + + +async def test_an_inferred_condition_the_adopted_side_does_not_use_is_kept(shop): + """目录里推断的「status = 'paid'」两边都没用：审核人的选择说明不了它对不对，不静默替换。""" name, source_id = shop await seed(source_id, {"valid_filter": catalog.make_item("status = 'paid'", "llm")}) + out = await record_choices([candidate(WITH_REFUNDS, RIGHT, "b", source=name)], run_id="run-1", + node_title="复核", actor=None) + assert out["written"] == [] + reason = out["skipped"][0]["reason"] + assert "「status = 'paid'」" in reason and "被采纳的盲算没有按它筛选" in reason and "未改动" in reason + entry = await entry_of(source_id) + assert entry.version == 1 and entry.notes["valid_filter"]["value"] == "status = 'paid'" + + +async def test_an_inferred_same_condition_is_confirmed(shop): + name, source_id = shop + await seed(source_id, {"valid_filter": catalog.make_item("status <> 'refunded'", "llm")}) out = await record_choices([candidate(WITH_REFUNDS, RIGHT, "b", source=name)], run_id="run-1", node_title="复核", actor=None) assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}] @@ -175,6 +321,15 @@ async def test_an_inferred_different_item_is_replaced(shop): assert item["value"] == "status <> 'refunded'" and (item["source"], item["status"]) == ("human", "confirmed") +async def test_an_inferred_different_dedup_rule_is_kept(shop): + name, source_id = shop + await seed(source_id, {"dedup": catalog.make_item("同一订单可能有多行，按 id 去重", "llm")}) + out = await record_choices([candidate(BUYERS, BUYERS_DISTINCT, "b", metric="付款客户数", source=name)], + run_id="run-1", node_title="复核", actor=None) + assert out["written"] == [] and "按 id 去重" in out["skipped"][0]["reason"] + assert (await entry_of(source_id)).notes["dedup"]["value"] == "同一订单可能有多行，按 id 去重" + + async def test_the_same_column_inferred_is_confirmed_and_keeps_its_rule(shop): """目录里推断的业务日期正是这一列（还写了规则、时区）：确认它，规则、时区原样留着。""" name, source_id = shop @@ -203,6 +358,19 @@ async def test_a_rejected_item_is_not_brought_back_but_a_rejected_other_value_do assert (await entry_of(source_id)).notes["business_date"]["value"] == {"column": "order_date"} +async def test_contradictory_choices_in_one_decision_write_nothing(shop): + """同一次裁决里，一个指标采纳了按下单日、另一个采纳了按支付日：写哪个都是替人拿主意，都不写。""" + name, source_id = shop + out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "a", metric="销售额", source=name), + candidate(BY_ORDER_DATE, RIGHT, "b", metric="复核销售额", source=name)], + run_id="run-1", node_title="复核", actor=None) + assert out["written"] == [] + [skipped] = out["skipped"] + assert "「销售额」" in skipped["reason"] and "「复核销售额」" in skipped["reason"] + assert "order_date" in skipped["reason"] and "paid_date" in skipped["reason"] and "互相矛盾" in skipped["reason"] + assert await entry_of(source_id) is None + + async def test_a_missing_source_or_an_unrecordable_choice_is_reported(shop): name, _ = shop out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source="no_such_source"), diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 09392a85..c68837cf 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -2,15 +2,21 @@ （tests/fixtures/dual_check_shop.py），盲算模型是 MockChatModel 剧本。 - 一致：不挂起，产出 agreed，日志 dual_check_agreed；盲算的查询发 tool.end（role: verify），台账只有一条 dual_check， - 用量计入运行；盲算 agent 的任务里没有原 SQL -- 不一致：挂起，审批 payload 和对照工件按约定；a / b / neither 三种选择的产出和缺口，on_reject 为 fail 时失败 -- 只差一个方面时写数据目录（业务日期、有效记录条件），差了不止一处不写，人工确认过的不同说法不覆盖 + 用量计入运行；盲算 agent 的任务里没有原 SQL，也没有原流程结果的列名 +- 一个都没核成是 unchecked、只核成一部分是 partial，日志 dual_check_unchecked 说清哪几个、为什么 +- 不一致：挂起，审批 payload 和对照工件按约定；a / b / neither 三种选择的产出和缺口，没选的照实记，驳回、on_reject + 为 fail 时失败 +- 差异说明只比这一格所在的那一列（一条查询算好几列、分组后取一行的，收窄了再比） +- 只差一个方面时写数据目录（业务日期、有效记录条件），差了不止一处不写，人工确认过的不同说法不覆盖，指标自己的 + 筛选范围不写，两条 SQL 规范化后一致的不写 - 全局关闭直通、节点覆盖；设置按发起时的快照；重放时盲算不再执行、事件不重复；报告的写作目录不含盲算的查询 -- 触顶、分组结果的维度、两个指标用到同一格只盲算一次 +- 触顶、分组结果的维度（只给取值、遮罩列不给）、两个指标用到同一格只盲算一次、遮罩列上的盲算结果不采用 """ from __future__ import annotations import asyncio +import json +import sqlite3 import pytest from httpx import ASGITransport, AsyncClient @@ -26,6 +32,7 @@ from app.engine.runner import run_manager from app.main import app from tests.fixtures.dual_check_shop import ( BOTH_WRONG, + BUYERS, BY_ORDER_DATE, BY_REGION, DEFINITION, @@ -70,6 +77,14 @@ async def shop(tmp_path): await drop_source(source_id) +@pytest.fixture +async def masked_shop(tmp_path): + """遮了 region、secret_gmv 两列的同一个库。""" + name, source_id = await add_source(build(tmp_path / "masked.db"), options={"mask_columns": "region, secret_gmv"}) + yield name, source_id + await drop_source(source_id) + + @pytest.fixture async def client(): async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: @@ -159,6 +174,7 @@ async def test_agreeing_check_passes_without_asking(monkeypatch, shop): evs = await events(run.id) out = dual_output(run, evs) assert out["status"] == "agreed" and "issuance_gaps" not in out + assert out["counts"] == {"agreed": 1, "disagreed": 0, "unchecked": 0} [metric] = out["metrics"] assert metric["metric_id"] == "gmv" and metric["status"] == "agreed" and metric["delta_pct"] == 0.0 assert metric["original_rendered"] == metric["blind_rendered"] @@ -174,18 +190,21 @@ async def test_agreeing_check_passes_without_asking(monkeypatch, shop): assert kinds == ["dual_check"] entry = finished.data["evidence"][0] assert entry["card"] == "card" and entry["status"] == "agreed" - assert entry["metrics"] == [{"id": "gmv", "status": "agreed"}] + assert entry["metrics"] == [{"id": "gmv", "status": "agreed"}] and entry["counts"] == out["counts"] + assert not logs(evs, "dual_check_unchecked") # 用量计入运行：图里只有盲算调用了模型 assert blind.calls and run.usage["calls"] == len(blind.calls) assert run.output["_issuance"]["tier"] == "formal" and run.output["_issuance"]["gaps"] == [] - # 盲算 agent 只看得到指标名、口径说明、计算式里是哪一项：看不到原 SQL、取数节点 + # 盲算 agent 只看得到指标名、口径说明、计算式里是哪一项：看不到原 SQL、取数节点、原流程结果的列名 [task] = blind.tasks assert "指标：销售额" in task and DEFINITION in task and "输入1" in task assert "status <> 'refunded'" not in task and "nodes.fetch" not in task and "SUM(" not in task + assert "gmv" not in task async def test_grouped_rows_tell_the_blind_agent_which_group(monkeypatch, shop): - """结果分了组：盲算 agent 要知道算的是哪一组（同一行的维度列），但看不到同一行别的数；统计期这类参数在运行输入里。""" + """结果分了组：盲算 agent 要知道算的是哪一组（同一行维度列的取值），但看不到同一行别的数、原流程是按哪一列分的组； + 统计期这类参数在运行输入里。""" name, _ = shop metric = {**GMV, "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} blind = Blind(monkeypatch, name, {"华东销售额": (BY_REGION, "gmv")}) @@ -194,12 +213,27 @@ async def test_grouped_rows_tell_the_blind_agent_which_group(monkeypatch, shop): run = await start(spec, {"week_start": "2026-09-01", "week_end": "2026-09-07"}) assert run.status == "succeeded", run.error [task] = blind.tasks - assert "region 为「华东」" in task and "（未提供" in task and "300" not in task + assert "「华东」" in task and "（未提供" in task and "300" not in task + assert "region" not in task and "gmv" not in task assert "week_start = 2026-09-01" in task and "week_end = 2026-09-07" in task out = dual_output(run, await events(run.id)) assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed" +async def test_the_blind_agent_is_not_told_which_date_column_the_original_grouped_by(monkeypatch, shop): + """原流程按下单日分组、口径卡取其中一天：盲算只知道是哪一天，按哪一列归属要它自己按口径说明判断。告诉它 + 「order_date 为 2026-09-02」，它照着算出同一个数，下单日和支付日的陷阱就漏过去了。""" + name, _ = shop + by_day = (f"SELECT order_date, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK} " + "GROUP BY order_date ORDER BY order_date") + blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, by_day)) + assert run.status in ("succeeded", "interrupted"), run.error + [task] = blind.tasks + assert "2026-09-02 这一天" in task + assert "order_date" not in task and "gmv" not in task + + async def test_two_metrics_using_one_cell_are_checked_once(monkeypatch, shop): name, _ = shop share = {"id": "gmv_k", "name": "销售额（千元）", "unit": "千元", "decimals": 2, "format": "thousands", @@ -213,20 +247,72 @@ async def test_two_metrics_using_one_cell_are_checked_once(monkeypatch, shop): async def test_hitting_a_limit_counts_as_unchecked(monkeypatch, shop): + """查询次数用完了还没查到要交的那一格：按无法复算计入缺口；一个指标都没核成，产出是 unchecked，不是 agreed。""" name, _ = shop - Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + Blind(monkeypatch, name, {"销售额": ([BLIND_TOTAL, RIGHT], "gmv")}) run = await start(graph(name, RIGHT), options={"verify": {"dual_check": {"max_queries": 1}}}) assert run.status == "succeeded", run.error evs = await events(run.id) assert [(d["limit"], d["metric"]) for d in logs(evs, "dual_check_limit")] == [("max_queries", "gmv")] out = dual_output(run, evs) - assert out["status"] == "agreed" and out["metrics"][0]["status"] == "unchecked" + assert out["status"] == "unchecked" and out["metrics"][0]["status"] == "unchecked" + assert out["counts"] == {"agreed": 0, "disagreed": 0, "unchecked": 1} assert out["issuance_gaps"] == ["指标「销售额」无法复算：盲算已达每项最多查询次数（1 次）"] + [said] = logs(evs, "dual_check_unchecked") + assert said["metrics"] == [{"id": "gmv", "name": "销售额", "reason": "盲算已达每项最多查询次数（1 次）"}] + assert not logs(evs, "dual_check_agreed") assert not [e for e in evs if e.type == "human.requested"] + [finished] = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"] + assert finished.data["evidence"][0]["status"] == "unchecked" assert run.output["_issuance"]["tier"] == "degraded" assert run.output["_issuance"]["gaps"] == ["双盲复算「复核销售额」：指标「销售额」无法复算：盲算已达每项最多查询次数（1 次）"] +async def test_an_answer_on_the_last_allowed_query_still_counts(monkeypatch, shop): + """查询次数刚好用完，交回的那一格是真查出来的：照常比对，不按触顶算无法复算。""" + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, RIGHT), options={"verify": {"dual_check": {"max_queries": 1}}}) + assert run.status == "succeeded", run.error + evs = await events(run.id) + out = dual_output(run, evs) + assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed" and "issuance_gaps" not in out + assert not logs(evs, "dual_check_limit") and not logs(evs, "dual_check_unchecked") + assert run.output["_issuance"]["tier"] == "formal" + + +async def test_every_metric_sharing_a_cell_reports_the_limit(monkeypatch, shop): + name, _ = shop + share = {"id": "gmv_k", "name": "销售额（千元）", "unit": "千元", "decimals": 2, "format": "thousands", + "expression": "round(cell(nodes.fetch, 0, 'gmv') / 1000, 2)"} + blind = Blind(monkeypatch, name, {"销售额": ([BLIND_TOTAL, RIGHT], "gmv")}) + run = await start(graph(name, RIGHT, metrics=[GMV, share]), options={"verify": {"dual_check": {"max_queries": 1}}}) + assert run.status == "succeeded", run.error + assert blind.runs == 1 + evs = await events(run.id) + assert [(d["limit"], d["metric"]) for d in logs(evs, "dual_check_limit")] == [ + ("max_queries", "gmv"), ("max_queries", "gmv_k")] + [said] = logs(evs, "dual_check_unchecked") + assert [m["id"] for m in said["metrics"]] == ["gmv", "gmv_k"] + + +async def test_some_unchecked_metrics_make_a_partial_result(monkeypatch, shop): + name, _ = shop + east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + north = {**GMV, "id": "north", "name": "华北销售额", "expression": "cell(nodes.fetch, 1, 'gmv')"} + Blind(monkeypatch, name, {"华东销售额": (BY_REGION, "gmv")}) + run = await start(graph(name, BY_REGION, metrics=[east, north], dual={"definitions": {}})) + assert run.status == "succeeded", run.error + evs = await events(run.id) + out = dual_output(run, evs) + assert out["status"] == "partial" and [m["status"] for m in out["metrics"]] == ["agreed", "unchecked"] + assert out["counts"] == {"agreed": 1, "disagreed": 0, "unchecked": 1} + assert [d["metrics"] for d in logs(evs, "dual_check_agreed")] == [1] + [said] = logs(evs, "dual_check_unchecked") + assert [(m["id"], m["name"]) for m in said["metrics"]] == [("north", "华北销售额")] and said["metrics"][0]["reason"] + assert out["issuance_gaps"] == [f"指标「华北销售额」无法复算：{said['metrics'][0]['reason']}"] + + async def test_an_unusable_verify_model_is_tried_once(monkeypatch, shop): """核查模型没配好：每个指标都记无法复算、原因照实，后面的盲算不再一个个去试。""" from app.engine import verify_agent @@ -244,10 +330,13 @@ async def test_an_unusable_verify_model_is_tried_once(monkeypatch, shop): north = {**GMV, "id": "north", "name": "华北销售额", "expression": "cell(nodes.fetch, 1, 'gmv')"} run = await start(graph(name, BY_REGION, metrics=[east, north], dual={"definitions": {}})) assert run.status == "succeeded", run.error - out = dual_output(run, await events(run.id)) - assert [m["status"] for m in out["metrics"]] == ["unchecked", "unchecked"] + evs = await events(run.id) + out = dual_output(run, evs) + assert out["status"] == "unchecked" and [m["status"] for m in out["metrics"]] == ["unchecked", "unchecked"] assert all("核查模型不可用" in m["reason"] and "已停用" in m["reason"] for m in out["metrics"]) assert len(out["issuance_gaps"]) == 2 and len(tried) == 1 + [said] = logs(evs, "dual_check_unchecked") + assert [m["id"] for m in said["metrics"]] == ["east", "north"] and "核查模型不可用" in said["metrics"][0]["reason"] # -------------------------------------------------------------------------- @@ -348,14 +437,33 @@ async def test_neither_degrades_and_writes_nothing(monkeypatch, shop, client): assert not logs(evs, "dual_check_catalog") and await notes_of(source_id) is None -async def test_a_missing_choice_counts_as_neither(monkeypatch, shop, client): - name, _ = shop +async def test_a_missing_choice_is_recorded_as_no_choice(monkeypatch, shop, client): + """没选、选了认不出的：照实记「没有作出选择」，不替审核人说「两边都不对」；没采用原流程，照样降档。""" + name, source_id = shop Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) run = await start(graph(name, BY_ORDER_DATE)) done = await decide(client, run.id, {"gmv": "maybe"}) - out = dual_output(done, await events(run.id)) - assert out["choices"] == {"gmv": "neither"} - assert out["issuance_gaps"] == ["指标「销售额」双盲复算不一致，审核人判断两边都不对"] + assert done.status == "succeeded", done.error + evs = await events(run.id) + out = dual_output(done, evs) + assert out["choices"] == {} and "choice" not in out["metrics"][0] + assert out["issuance_gaps"] == ["指标「销售额」双盲复算不一致，审核人没有作出选择"] + [resolved] = logs(evs, "dual_check_resolved") + assert resolved["choices"] == {} and resolved["gaps"] == out["issuance_gaps"] + assert "没有作出选择" in resolved["message"] and "两边都不对" not in resolved["message"] + assert done.output["_issuance"]["tier"] == "degraded" + assert not logs(evs, "dual_check_catalog") and await notes_of(source_id) is None + + +async def test_rejecting_the_approval_fails_the_node(monkeypatch, shop, client): + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + approval = await pending(run.id) + res = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": False, "note": "口径要重新对"}) + assert res.status_code == 200, res.text + done = await wait(run.id, ("succeeded", "failed")) + assert done.status == "failed" and "人工驳回" in (done.error or "") and "口径要重新对" in done.error async def test_on_reject_fail_stops_the_run(monkeypatch, shop, client): @@ -446,6 +554,185 @@ async def test_a_confirmed_catalog_entry_is_not_overwritten(monkeypatch, shop, c assert entry.version == 1 and entry.notes["business_date"]["value"] == {"column": "order_date"} +async def test_a_metric_scope_filter_is_not_written_as_a_valid_filter(monkeypatch, shop, client): + """「华东销售额」原流程按指标名只看华东，盲算漏了地区：审核人选原流程，但 region = '华东' 是这个指标自己的筛选 + 范围，写成表的有效记录条件，以后所有查订单表的工作流（包括下一次盲算）都会只看华东。""" + name, source_id = shop + east = {**GMV, "id": "east", "name": "华东销售额"} + Blind(monkeypatch, name, {"华东销售额": (RIGHT, "gmv")}) + original = RIGHT.replace("WHERE ", "WHERE region = '华东' AND ") + run = await start(graph(name, original, metrics=[east], dual={"definitions": {"east": "华东地区已支付订单的金额"}})) + [item] = (await pending(run.id)).payload["summary"] + assert len(item["diff"]) == 1 and "region = '华东'" in item["diff"][0] + done = await decide(client, run.id, {"east": "a"}) + assert done.status == "succeeded", done.error + [wrote] = logs(await events(run.id), "dual_check_catalog") + assert wrote["written"] == [] and "可能是指标自己的筛选范围" in wrote["skipped"][0]["reason"] + assert await notes_of(source_id) is None + + +async def test_equivalent_sql_with_different_values_points_at_the_data(monkeypatch, shop, client, tmp_path): + """原流程查完之后库里多了一单：两条 SQL 规范化后一致、值却不同，差异说明写「差在数据」，不写数据目录。""" + from app.engine.nodes.dual_check import SAME_SQL + + name, source_id = shop + + def late_order() -> None: + db = sqlite3.connect(tmp_path / "shop.db") + db.execute("INSERT INTO orders VALUES (6, '华东', 104, 70.0, 'paid', '2026-09-04', '2026-09-04', 'online', 0)") + db.commit() + db.close() + + Blind(monkeypatch, name, {"销售额": (BLIND_TOTAL, "blind_total")}, on_query=late_order) + run = await start(graph(name, RIGHT)) + assert run.status == "interrupted", run.error + payload = (await pending(run.id)).payload + [item] = payload["summary"] + assert item["diff"] == [SAME_SQL] and item["fallback"] is False + assert item["original_rendered"].startswith("330") and item["blind_rendered"].startswith("400") + [metric] = artifact_store.load(payload["card_artifact"])["metrics"] + assert metric["diff"]["equivalent"] is True and metric["diff"]["items"] == [{"aspect": "data", "text": SAME_SQL}] + done = await decide(client, run.id, {"gmv": "b"}) + [wrote] = logs(await events(done.id), "dual_check_catalog") + assert wrote["written"] == [] and "规范化后一致" in wrote["skipped"][0]["reason"] + assert await notes_of(source_id) is None + + +async def test_a_multi_column_query_is_compared_on_its_own_column(monkeypatch, shop, client): + """原流程一条查询算了好几列，口径卡取其中一列：差异只比这一列是怎么算的，不混进「选出的列」这类差别。""" + name, source_id = shop + wide = f"SELECT SUM(amount) AS gmv, COUNT(*) AS n FROM orders WHERE status <> 'refunded' AND order_date {WEEK}" + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, wide)) + payload = (await pending(run.id)).payload + [item] = payload["summary"] + assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] + [metric] = artifact_store.load(payload["card_artifact"])["metrics"] + assert [i["aspect"] for i in metric["diff"]["items"]] == ["time"] + # 对照里的原 SQL 还是原流程整条查询：给人看的是原文 + assert metric["original"]["inputs"][0]["sql"] == wide + await decide(client, run.id, {"gmv": "b"}) + assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"} + + +async def test_a_grouped_query_is_compared_on_the_row_it_was_taken_from(monkeypatch, shop, client): + """原流程按地区分组、取华东那一行：收窄成「华东这一组」再比，差异里不出现分组、地区条件这类和这一格无关的差别。""" + name, source_id = shop + by_region = (f"SELECT region, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK} " + "GROUP BY region ORDER BY region") + east_paid = RIGHT.replace("WHERE ", "WHERE region = '华东' AND ") + east = {**GMV, "id": "east", "name": "华东销售额"} + Blind(monkeypatch, name, {"华东销售额": (east_paid, "gmv")}) + run = await start(graph(name, by_region, metrics=[east], dual={"definitions": {}})) + [item] = (await pending(run.id)).payload["summary"] + assert item["original_rendered"].startswith("200") and item["blind_rendered"].startswith("300") + assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] + await decide(client, run.id, {"east": "b"}) + assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"} + + +async def test_two_differing_inputs_do_not_write_the_catalog(monkeypatch, shop, client): + """一个指标的两个输入都对不上：审核人选的是整个指标哪边对，拆不出是哪一处，不写目录。""" + name, source_id = shop + share = (f"SELECT SUM(CASE WHEN region = '华东' THEN amount ELSE 0 END) AS east, SUM(amount) AS total FROM orders " + f"WHERE status <> 'refunded' AND order_date {WEEK}") + east_paid = (f"SELECT SUM(CASE WHEN region = '华东' THEN amount ELSE 0 END) AS east FROM orders " + f"WHERE status <> 'refunded' AND paid_date {WEEK}") + ratio = {"id": "east_share", "name": "华东占比", "unit": "", "decimals": 3, "format": "plain", + "expression": "cell(nodes.fetch, 0, 'east') / cell(nodes.fetch, 0, 'total')"} + blind = Blind(monkeypatch, name, {"华东占比|输入1": (east_paid, "east"), "华东占比|输入2": (RIGHT, "gmv")}) + run = await start(graph(name, share, metrics=[ratio], dual={"definitions": {}})) + assert blind.runs == 2 + [item] = (await pending(run.id)).payload["summary"] + assert len(item["diff"]) == 2 and all("输入「" in d for d in item["diff"]) + done = await decide(client, run.id, {"east_share": "b"}) + [wrote] = logs(await events(done.id), "dual_check_catalog") + assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"] + assert await notes_of(source_id) is None + + +async def test_only_the_configured_metrics_are_checked(monkeypatch, shop): + name, _ = shop + wide = f"SELECT SUM(amount) AS gmv, COUNT(*) AS n FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}" + count = {"id": "orders", "name": "订单数", "unit": "单", "decimals": 0, "expression": "cell(nodes.fetch, 0, 'n')"} + blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv"), "订单数": (BUYERS, "buyers")}) + run = await start(graph(name, wide, metrics=[GMV, count], dual={"metrics": ["gmv"]})) + assert run.status == "succeeded", run.error + out = dual_output(run, await events(run.id)) + assert [m["metric_id"] for m in out["metrics"]] == ["gmv"] and out["status"] == "agreed" + assert blind.runs == 1 and all("指标：订单数" not in t for t in blind.tasks) + + +async def test_a_zero_original_compares_by_absolute_error_and_has_no_delta_pct(monkeypatch, shop): + name, _ = shop + voids = {"id": "voids", "name": "作废单数", "unit": "单", "decimals": 0, "expression": "cell(nodes.fetch, 0, 'n')"} + none_void = f"SELECT COUNT(*) AS n FROM orders WHERE status = 'void' AND paid_date {WEEK}" + refunded = f"SELECT COUNT(*) AS n FROM orders WHERE status = 'refunded' AND paid_date {WEEK}" + Blind(monkeypatch, name, {"作废单数": (refunded, "n")}) + run = await start(graph(name, none_void, metrics=[voids], dual={"definitions": {}})) + [item] = (await pending(run.id)).payload["summary"] + assert item["original_rendered"].startswith("0") and item["blind_rendered"].startswith("1") + assert item["delta_pct"] is None + + +def test_zero_originals_use_absolute_error(): + from app.engine.nodes.dual_check import _agree, _delta_pct + + assert _agree(0, 1e-7, 1e-6) and not _agree(0, 0.5, 1e-6) + assert _agree(100.0, 100.00005, 1e-6) and not _agree(100.0, 100.01, 1e-6) + assert _delta_pct(0, 5) is None and _delta_pct(310.0, 330.0) == 6.45 and _delta_pct(-10, -5) == 50.0 + + +async def test_the_summary_lists_at_most_three_differences(monkeypatch, shop): + name, _ = shop + blind_sql = RIGHT + " AND region <> '华北' AND customer_id <> 999" + Blind(monkeypatch, name, {"销售额": (blind_sql, "gmv")}) + run = await start(graph(name, BOTH_WRONG)) + payload = (await pending(run.id)).payload + [item] = payload["summary"] + [metric] = artifact_store.load(payload["card_artifact"])["metrics"] + assert len(metric["diff"]["items"]) == 4 and len(item["diff"]) == 3 + assert item["diff"] == [i["text"] for i in metric["diff"]["items"][:3]] + + +# -------------------------------------------------------------------------- +# 遮罩 +# -------------------------------------------------------------------------- + + +async def test_a_blind_cell_on_a_masked_column_is_not_used(monkeypatch, masked_shop, client): + """盲算交的那一格在遮罩列上：值不采用（不进审批卡、对照工件、产出），这个指标按无法复算记。""" + name, _ = masked_shop + east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + north = {**GMV, "id": "north", "name": "华北销售额", "expression": "cell(nodes.fetch, 1, 'gmv')"} + east_by_order = BY_ORDER_DATE.replace("WHERE ", "WHERE region = '华东' AND ") + north_masked = (f"SELECT SUM(amount) + 0.25 AS secret_gmv FROM orders WHERE status <> 'refunded' " + f"AND paid_date {WEEK} AND region = '华北'") + Blind(monkeypatch, name, {"华东销售额": (east_by_order, "gmv"), "华北销售额": (north_masked, "secret_gmv")}) + run = await start(graph(name, BY_REGION, metrics=[east, north], dual={"definitions": {}})) + assert run.status == "interrupted", run.error + payload = (await pending(run.id)).payload + card = artifact_store.load(payload["card_artifact"]) + assert [m["metric_id"] for m in payload["summary"]] == ["east"] + assert [m["metric_id"] for m in card["metrics"]] == ["east"] + assert "30.25" not in json.dumps(payload, ensure_ascii=False) + assert "30.25" not in json.dumps(card, ensure_ascii=False) + done = await decide(client, run.id, {"east": "a"}) + out = dual_output(done, await events(run.id)) + [_, row] = out["metrics"] + assert row["status"] == "unchecked" and "遮罩" in row["reason"] and row["blind_rendered"] is None + assert "30.25" not in json.dumps(out, ensure_ascii=False) + + +async def test_masked_group_columns_are_not_shown_to_the_blind_agent(monkeypatch, masked_shop): + name, _ = masked_shop + first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + blind = Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")}) + await start(graph(name, BY_REGION, metrics=[first], dual={"definitions": {}})) + [task] = blind.tasks + assert "华东" not in task and "region" not in task + + # -------------------------------------------------------------------------- # 设置：全局开关、节点覆盖、发起时的快照 # --------------------------------------------------------------------------