双盲复算复核整改:指标自己的筛选范围不再写成有效记录条件,盲算不再看到原流程的列名,差异说明收窄到这一格

写数据目录(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 条、
驳回、多列查询、分组取一行、指标筛选范围不写目录等用例;每条核心改动都做过去掉实现、测试变红的验证
This commit is contained in:
ThearchyHelios
2026-10-05 15:54:03 +08:00
parent e353706498
commit 1491b5c7f3
10 changed files with 1404 additions and 190 deletions
+2 -1
View File
@@ -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**。它自动收集上游的口径卡指标、查询结果
+378 -55
View File
@@ -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"(?<![0-9]){re.escape(word)}(?![0-9])")
if any(pattern.search(t) for t in lowered):
return word
return None
def _invalid_text(text: str) -> 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:
+38 -4
View File
@@ -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
# --------------------------------------------------------------------------
# 发起前检查绑定的工具
# --------------------------------------------------------------------------
+363 -75
View File
@@ -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"]
+16 -3
View File
@@ -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",
+24 -14
View File
@@ -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
+45 -4
View File
@@ -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
+49
View File
@@ -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
+185 -17
View File
@@ -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"),
+304 -17
View File
@@ -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
# --------------------------------------------------------------------------
# 设置:全局开关、节点覆盖、发起时的快照
# --------------------------------------------------------------------------