合入期 3 双盲复算(后端):口径卡上来自查询的输入另起看不到原 SQL 的只读 agent 独立复算,代回原式按误差比对;不一致时挂起交审核人三选一,选择只在差一个方面、且是排除无效记录这类数据事实时写进数据目录;发起时快照核查设置

This commit is contained in:
ThearchyHelios
2026-10-05 18:08:03 +08:00
11 changed files with 4549 additions and 12 deletions
+12
View File
@@ -214,6 +214,18 @@ NODE_REFERENCE = """\
写 2 只剩「2%」,0.004 这样的小比率还会显示不出来、报告引用不了。on_missing 默认 fail(缺输入整张卡失败);写 null 则缺输入的指标记为空值,交给出具契约判档
取 tool 节点查库的结果用 cell(nodes.取数节点id, 行, '列名')(行号从 0 数,列按列名):cell() 直接认查询工具
交回的结果,出处精确到快照里的那一格,不用再接 transform 解析。取 agent 的字段写 vars.<agent 的 assign_to>.字段
- dual_check:双盲复算。正式出具前,口径卡上输入来自查询的指标另起一个看不到原 SQL 的只读 agent 独立写 SQL 再算一遍:
一致就放行;不一致时停下来,审批卡并排列出两边的 SQL、差异说明和两个值,由人逐个指标选采用原流程、采用盲算还是
两边都不对。选盲算或都不对的计入缺口、出具降档;两边只差一个方面(业务日期按哪一列、有效记录条件、去重)时,选择
会写进数据目录。**要正式出具的模板、口径卡上的指标来自查询(tool 节点查库,或 agent 查库并开了 cite_fields)时加**,
接在口径卡之后、report 之前(取数 → 口径卡 → dual_check → report → output);只回答一个问题的问数据、指标都是
常量或运行输入的不必加。config: {card, metrics, definitions, provider, model, tolerance, on_reject, enabled}:
card 是要复算的口径卡节点 id(上游只有一张口径卡时不写);metrics 是要复算的指标 id 列表(不写就是输入来自查询的
全部指标);definitions 是 {指标 id: 口径说明},写清业务定义(按哪个日期归属、排除哪些行、是否去重),盲算 agent
只看得到它和数据目录,所以要按业务写,不要抄原 SQL;provider / model 指定盲算模型(**默认不写**,取设置里的核查
模型;建议和取数的 agent 不同);tolerance 是允许的相对误差(不写取设置);on_reject:degrade(默认,计入缺口、
出具降档)/ fail(审核人没采用原流程的结果就判为失败);enabled 不写跟随设置里的开关,true / false 强制开关。
不要给它配 retries:它的失败是确定的,重试只会把盲算再做一遍、再请审核人选一次(画布校验会报错)
- report:报告撰写(带引用)。config: {instructions, system, metrics_from:[口径卡id], numbers, on_violation, max_repairs,
claims, entities, assign_to}
**凡是给人看的、带数字的报告、分析、结论,都用 report,不要用 llm**。它自动收集上游的口径卡指标、查询结果
+825
View File
@@ -0,0 +1,825 @@
"""双盲复算的选择写进数据目录:审核人在不一致的指标上选了原流程或盲算,而两边只差一个方面、又正好对得上数据目录
里的一项事实时,把被采纳那一边的写法记成人工确认过的目录项。
对得上的事实只有三种(data/catalog.py 的表级项):
- 两边按不同的时间列筛时间 → 业务日期(business_date):被采纳的一边按哪一列筛,业务日期就记哪一列。只有一边按时间
筛的不算:审核人判断的是要不要限定时间,没在两列之间选过
- 一边多一个排除无效记录的条件 → 有效记录条件(valid_filter):被采纳的一边有这个条件才记,记的就是这个条件
- 聚合只差去不去重、被采纳的一边去重 → 去重规则(dedup)。采纳了不去重的一边不记:那是这个指标自己的算法(数行),
写成表的「不去重」会把别的指标带偏
**为什么要这么严。** 目录是给助手、db_schema 工具、SQL 检查和下一次盲算看的数据事实,人工确认过的项还会让 SQL 检查
报错误级、以后的推断也不再覆盖它,写错了就一直错下去。只差一个方面时,人选的那一边就是对这一处的判断;差了好几处、
或者差异说明里有兜底(未能归纳的差异),人选的是「整体上哪边对」,拆不出是哪一处对,写进去就是替人编了一条事实。
两条 SQL 规范化后一致却算出不同的值(两次取数之间数据变了、结果被截断),差在数据不在写法,更没有可记的事实。
**有效记录条件和指标自己的筛选范围分不开。** 「华东销售额」的 region = '华东'、「线上订单数」的 channel = 'online'
在写法上和 status <> 'refunded' 一样是一个筛选条件,审核人选原流程只说明这个指标该这么筛,不说明整张表的有效记录
都得这么筛。写成表的有效记录条件,所有查这张表的工作流(包括下一次盲算)都会被带偏。所以条件要同时满足三条才写:
条件里的取值没有出现在指标名称、口径说明、这一格所在的分组、本次运行的输入里,用到的列也不是分组的列;每一段都是在
排除无效记录(排除的取值是退款、作废、测试这类,或者码值表里写着这类意思;标记列只认留下有效记录的那一边:
is_test = 0、is_valid = 1、deleted_at IS NULL);只涉及一张表。拿不准就不写,原因照实交回。
**怎么写。** 照对话提案的规矩核对(plan_patch:列在不在表结构里、取值格式),记为人工填写、已确认,备注写明来自哪次
运行、哪个节点、哪个指标。目录里这一项已经是人工确认过的另一种说法时不覆盖,照实交回原因;已经是同样的说法就什么都
不写(再写一遍也不升版本);这一项被人工驳回过、要写的又带回了驳回的说法(有效记录条件按段比:驳回的那句每一段都在
要写的条件里),两次人工判断对不上,不写。有效记录条件是一整句条件:目录里已有的条件被采纳的一边(几个指标并成一项
时是每一个)也按它筛了的,和新条件合起来写,已有的段只写一次(已确认的不改,照实说多出了哪几段);有一个没按它筛
的,两者的关系拿不准,不动。推断的去重规则说的未必是同一件事,也不替换。整个写入由节点放在 ask() 之后的一个 @task
里:审批恢复重放、接着运行时不会再写一遍。
**同一次裁决里几个指标落到同一项上。** 说法一样的写一次,备注里列出每个指标。有效记录条件是能用 AND 叠加的排除条件
(这里只收排除无效记录的),几个指标各自采纳的条件都有人的选择撑着,合成一句写——「排除退款」和「排除测试单」并不
矛盾。业务日期一张表只有一个,去重规则说的未必是同一件事,说法不同就一个都不写。合起来之前先按每个指标自己的说法
过一遍驳回检查:自己采纳的说法曾被人工驳回的指标拿出来不写、照实说原因,其余的照写;合起来以后又包含了驳回的整句的,
也不写。
**预告。** 审批卡上要让审核人看得出选了之后会往目录里写什么(preview):按发起审批时的目录,选原流程、选盲算各会
写什么、不写的为什么。和真写是同一套判断(derive 认事实、_plan 对目录、_revives 认驳回),所以只在两处可能和最后写
的不一样:审批停下期间目录被人改了;同一次裁决里几个指标落到同一项上(上一段)——写进去的是合起来的说法,预告说会
写的也可能因为说法互相矛盾、和已有条件的关系拿不准、合起来又成了驳回过的说法而不写。反过来不会:预告说不写的,
不会因为和别的指标并成一项被写进去。
catalog.save_patch 不收备注,这里用它的同一套原料(read_entry、apply_patch、write_entry 的乐观锁)写,只多把备注放进项里。
"""
from __future__ import annotations
import logging
import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, replace
from typing import Any
from sqlalchemy import select
from app.data import catalog
from app.db.base import SessionLocal
from app.db.models import DataSource
logger = logging.getLogger(__name__)
#: 能写的三项(表级)
FACT_ITEMS = ("business_date", "valid_filter", "dedup")
#: 审核人采纳的一边在差异说明里的位置:a 是原流程、b 是盲算(节点调用 explain_sql_diff 时按这个顺序给两边)
SIDE_LABEL = {"a": "原流程", "b": "盲算"}
#: 差异条目的方面(sql_semantics.FacetDiff.aspect)在说明里的叫法
ASPECT_LABEL = {
"time": "时间列", "time_range": "时间范围", "filter": "筛选条件", "aggregate": "聚合方式", "expression": "选出的列",
"join": "关联", "tables": "查询的表", "set_op": "几支查询的合并方式", "group_by": "分组", "having": "分组后的筛选",
"limit": "取哪几行", "order": "排序", "parse": "能否解析", "other": "未能归纳的地方",
}
#: 聚合函数的说法(去重规则那一句用)
_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 只用来写备注和说明;sql、dialect 是被采纳
那一边的写法和方言,有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。同一次裁决里几个指标落到同一项上
时并成一项:members 是并进来的每个指标各自的那一项(各带自己的说法,审批卡上各自的预告就是按它算的),value 是
合起来之后的说法;单个指标的 members 为空。"""
source: str
table: str
item: str
value: Any
metric: str
choice: str
sql: str = ""
dialect: str | None = None
members: tuple[Fact, ...] = ()
def _members(fact: Fact) -> tuple[Fact, ...]:
"""这一项背后的每个指标的选择:并成一项的是并进来的每一个(各带自己的说法),单个指标的是它自己。"""
return fact.members or (fact,)
def _names(members: Sequence[Fact]) -> str:
"""说明里的指标叫法:「销售额」,几个指标时「销售额」、「订单金额」。"""
names = list(dict.fromkeys(m.metric for m in members))
return "、".join(f"「{n}」" for n in names[:4]) + (f"等 {len(names)} 个" if len(names) > 4 else "")
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}(比较的两条 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")
diff = candidate.get("diff") if isinstance(candidate.get("diff"), Mapping) else {}
if side not in SIDE_LABEL:
return None, f"指标「{name}」没有采用任何一边的结果,不写入数据目录"
if diff.get("equivalent"):
return None, f"指标「{name}」两边的 SQL 规范化后一致,差异来自数据而不是写法,不写入数据目录"
items = [i for i in diff.get("items") or [] if isinstance(i, Mapping)]
if diff.get("fallback") or len(items) != 1:
return None, (f"指标「{name}」两边的 SQL 不止差一个方面,或有未能归纳的差异,"
"无法确定审核人认可的是哪一处,不写入数据目录")
item = items[0]
aspect = str(item.get("aspect") or "")
if aspect == "time":
fact, why = _business_date(item, side)
elif aspect == "filter":
fact, why = _valid_filter(item, side, candidate, schema_cache, notes or {})
elif aspect == "aggregate":
fact, why = _dedup(item, side, candidate, schema_cache)
else:
return None, (f"指标「{name}」两边差在{ASPECT_LABEL.get(aspect, '其他地方')},"
"对不上数据目录里的业务日期、有效记录条件或去重规则,不写入数据目录")
if why is not None:
return None, f"指标「{name}」{why},不写入数据目录"
table, key, value = fact
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), 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 {}
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 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, 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:
if condition is None:
return None, f"多出筛选条件的是{SIDE_LABEL[other]},被采纳的{SIDE_LABEL[side]}没有这个条件"
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 {}
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
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 <> …。
解析不了就原样。"""
from sqlglot import exp
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)
try:
return tree.sql(dialect=read)
except Exception: # noqa: BLE001
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 _keys(condition: Any, read: str | None) -> set[str] | None:
"""一句条件按 AND 拆开后每一段的比较用写法;不是文字、解析不了返回 None。"""
tree = _parse(condition, read) if isinstance(condition, str) and condition.strip() else None
return {_normalized(part, read) for part in _and_parts(tree)} if tree is not None else None
def _and_join(base: str | None, values: list[str], dialect: str | None, *, keep_base: bool = True) -> str | None:
"""几句条件用 AND 接成一句:已经有的段(base 里的、前面接过的,按 _normalized 比)不再接。一句里每一段都是新的,
原样接上(保留原来的写法);只有几段是新的,只接那几段(按方言重新写)——整句接上会把已有的段再写一遍。
keep_base 为假时 base 只用来认哪些段已经有了、不写进结果(说「另外多出哪些条件」用),什么都没多出时为空串。
不止一段时,最外层是 OR 的加括号,不然后面接的条件只并到 OR 的后一半上。有一句解析不了返回 None。"""
from sqlglot import exp
read = _read(dialect)
pieces: list[tuple[str, bool]] = [] # (写法, 最外层是不是 OR)
seen: set[str] = set()
for i, text in enumerate([*([base] if base else []), *values]):
tree = _parse(str(text), read)
if tree is None:
return None
parts = _and_parts(tree)
fresh = []
for part in parts:
key = _normalized(part, read)
if key not in seen:
seen.add(key)
fresh.append(part)
if not fresh or (base and i == 0 and not keep_base):
continue
if len(fresh) == len(parts):
pieces.append((str(text).strip(), isinstance(tree, exp.Or)))
else:
pieces += [(part.sql(dialect=read), isinstance(part, exp.Or)) for part in fresh]
if len(pieces) == 1:
return pieces[0][0]
return " AND ".join(f"({text})" if is_or else text for text, is_or in pieces)
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 = _keys(existing, read)
combined = _and_join(existing, [str(m.value) for m in _members(fact)], fact.dialect) \
if have is not None else None
if have is None or combined is None:
return None, f"「{fact.table}」已有的{label}「{existing}」无法解析,与这次的条件合不起来,未改动"
for member in _members(fact):
if not have <= (_conditions_of(member.sql, read) or set()):
return None, (f"「{fact.table}」已有的{label}「{existing}」,指标「{member.metric}」被采纳的"
f"{SIDE_LABEL.get(member.choice, member.choice)}没有按它筛选,两者的关系无法确定,未改动。"
"如需更改,请到数据目录中修改")
return combined, 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 里认:
参数列属于哪张实表。"""
from app.data.sql_semantics import facets
pair = _aggregate_pair(item.get("a"), item.get("b"), candidate.get("dialect"))
if pair is None:
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]
columns = {ref for a in hits for ref in a.arg.columns}
if len(columns) != 1:
return None, "无法确定去重的是哪一张表的哪一列"
((table, column),) = columns
if table is None:
# 参数列没认出属于哪张表(没给表结构、列没写表名):整条语句只查了一张表时就是它
names = list(dict.fromkeys(t.text for t in found.tables))
if len(names) != 1:
return None, "无法确定去重的列属于哪一张表"
table = names[0]
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。"""
from sqlglot import exp
read = _read(dialect)
def shape(text: Any) -> tuple[str, str, bool] | None:
if not isinstance(text, str) or not text.strip():
return None
node = _parse(text, read)
if not isinstance(node, exp.AggFunc):
return None
arg = node.this
distinct = isinstance(arg, exp.Distinct)
if distinct:
inner = arg.expressions
if len(inner) != 1:
return None
arg = inner[0]
if arg is None:
return None
return node.key, arg.sql(dialect=read), distinct
left, right = shape(a), shape(b)
if left is None or right is None or left[:2] != right[:2] or left[2] == right[2]:
return None
return left[0], (left[2], right[2])
# --------------------------------------------------------------------------
# 写入
# --------------------------------------------------------------------------
def note_of(fact: Fact, *, run_id: str, node_title: str) -> str:
"""目录项上的备注:来自哪次运行、哪个节点、哪些指标、各采纳了哪一边。"""
parts = []
for side, label in SIDE_LABEL.items():
names = list(dict.fromkeys(m.metric for m in _members(fact) if m.choice == side))
if names:
parts.append(f"在指标{'、'.join(f'「{n}」' for n in names)}上采用了{label}的写法")
return f"双盲复算(运行 {run_id},节点「{node_title}」):审核人" + ",".join(parts)
def _value_text(item: str, value: Any) -> str:
if item == "business_date" and isinstance(value, Mapping):
return str(value.get("column") or "")
return str(value)
def _same_fact(item: str, a: Any, b: Any) -> bool:
"""目录里的一项和要写的是不是同一个说法:业务日期只看哪一列(规则、时区是补充),文字压掉多余的空白比。"""
if item == "business_date":
return isinstance(a, Mapping) and isinstance(b, Mapping) and a.get("column") == b.get("column")
return isinstance(a, str) and isinstance(b, str) and " ".join(a.split()) == " ".join(b.split())
def _differing(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}),无法确定哪一个是数据事实,"
"都未写入数据目录")
def _merged(members: list[Fact]) -> tuple[Fact | None, str | None]:
"""同一次裁决里落到同一项上的几个说法合成一项:(合起来的那一项, None) 或 (None, 为什么都不写)。只有一个时就是它。
说法都一样的写一次。有效记录条件用 AND 合成一句(derive 只收排除无效记录的条件,几句叠起来不会互相打架)。
业务日期一张表只有一个、去重规则说的未必是同一件事,说法不同就都不写。每个指标各自的那一项原样留在 members 里:
驳回检查(_screen)、和目录里已有条件的合并(_combined_filter)都要按各自的说法算。"""
first = members[0]
if len(members) == 1:
return first, None
if first.item == "valid_filter":
value = _and_join(None, [str(m.value) for m in members], first.dialect)
return (replace(first, value=value, members=tuple(members)), None) if value is not None \
else (None, _differing(members))
if all(_same_fact(first.item, m.value, first.value) for m in members[1:]):
return replace(first, members=tuple(members)), None
return None, _differing(members)
def _rejected(notes: Mapping[str, Any], item: str) -> tuple[bool, Any]:
"""目录里这一项是不是被人工驳回的:(是不是, 驳回的说法)。"""
stored = notes.get(item)
if isinstance(stored, Mapping) and stored.get("status") == "rejected":
return True, stored.get("value")
return False, None
def _revives(fact: Fact, rejected: Any) -> bool:
"""要写的说法是不是又带回了曾被人工驳回的说法。业务日期、去重规则:同一个说法。有效记录条件:驳回的那句的每一段
都在要写的条件里(按 _normalized 比,!= 和 <>、列名大小写、先后不同都认得出;有一边解析不了按原文比)——驳回的
那句只是要写的一部分也算,写进去就等于替人把驳回过的条件又确认了一遍。"""
if fact.item != "valid_filter":
return _same_fact(fact.item, rejected, fact.value)
read = _read(fact.dialect)
mine, theirs = _keys(fact.value, read), _keys(rejected, read)
if mine is None or theirs is None:
return _same_fact(fact.item, rejected, fact.value)
return bool(theirs) and theirs <= mine
def _rejected_text(fact: Fact, rejected: Any, members: Sequence[Fact]) -> str:
"""驳回检查拦下时的说法:驳回的是哪一句(照目录里记的原文),哪几个指标的选择又带回了它。审批卡上的预告和真写
拦下时用的是同一句话。"""
label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
return (f"「{fact.table}」的{label}「{_value_text(fact.item, rejected)}」曾被人工驳回,与这次指标{_names(members)}"
"的选择相反,未写入。如需更改,请到数据目录中修改")
def _screen(notes: Mapping[str, Any], fact: Fact) -> tuple[Fact | None, list[str]]:
"""几个指标并成的一项,先拿掉自己采纳的说法曾被人工驳回的指标:(剩下的重新并成的一项, 拿掉的为什么不写),一个
都不剩时前者为 None。单个指标的、这一项没被驳回过的原样返回,由 _plan 照常判断。
审批卡上每个指标的预告是按它自己的说法单独算的(preview → _plan),在那里被驳回检查拦下、卡上写着「不写」的,
并进来以后也得拦下。只拿合起来的整句去比,「status <> 'refunded'」和别的条件 AND 在一起就不再等于驳回的那句,
会被写成人工确认,和卡上说的正相反。拿掉的指标照实说原因(和它的预告同一句话),其余的照写,也和各自的预告一致。"""
rejected, said = _rejected(notes, fact.item)
if not rejected or not fact.members:
return fact, []
out: list[Fact] = []
kept: list[Fact] = []
for member in fact.members:
(out if _revives(member, said) else kept).append(member)
if not out:
return fact, []
reasons = [_rejected_text(fact, said, out)]
if not kept:
return None, reasons
narrowed, why = _merged(kept)
return narrowed, reasons if why is None else [*reasons, why]
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 里照实说。
先把每个候选认成目录项,再按(数据源、表、项)归拢成一项(_merged:说法一样的、有效记录条件合起来写,说法
不同的整组不写),逐项写(_write:按读到的目录先拿掉各自带回驳回说法的指标,其余的照写)。"""
skipped: list[dict[str, Any]] = []
facts: list[Fact] = []
for candidate in candidates:
name = str(candidate.get("source") or "")
try:
async with SessionLocal() as session:
row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none()
if row is None:
skipped.append({"reason": f"数据源「{name}」已不存在,指标「{candidate.get('metric')}」的选择未写入数据目录"})
continue
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 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():
fact, why = _merged(members)
if fact is None:
skipped.append({"reason": why})
continue
done, reasons = 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})
skipped += [{"reason": r} for r in reasons]
return {"written": written, "skipped": skipped}
async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None) -> tuple[bool, list[str]]:
"""写一项:(写了没有, 没写的原因)。几个指标并成的一项可能写了、同时有几个指标被拿出来没写,原因照样交回。"""
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}」已不存在,指标{_names(_members(fact))}的选择未写入数据目录"]
return await _write(session, row, fact, run_id=run_id, node_title=node_title, actor=actor)
except Exception: # noqa: BLE001 - 目录写不进去不能毁掉审批的结果:缺口照样按选择记
logger.exception("双盲复算的选择写入数据目录失败(数据源 %s)", fact.source)
return False, [f"指标{_names(_members(fact))}的选择写入数据目录时出错,未写入"]
def _plan(notes: Mapping[str, Any], fact: Fact, tables: Mapping[str, Any] | None) -> tuple[Any, str | None]:
"""按目录里这张表现在的说法,算出这一项要写成什么:(要写的取值, None) 或 (None, 为什么不写)。不读库、不写库:
真写(_write)和审批卡上的预告(preview)用的是这同一套判断。几个指标并成的一项,先由 _screen 拿掉各自带回驳回
说法的指标再交到这里。"""
label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
names = _names(_members(fact))
stored = notes.get(fact.item) if isinstance(notes.get(fact.item), dict) else None
rejected, said = _rejected(notes, fact.item)
if rejected and _revives(fact, said):
# 同一个说法以前被人明确驳回过:两次人工判断对不上,不替人拿主意。并成的一项各自的说法都不是驳回的那句、
# 合起来却又包含了它,也在这里拦下
return None, _rejected_text(fact, said, _members(fact))
live = stored if stored is not None and not rejected else None
value = fact.value
if fact.item == "business_date" and live is not None and isinstance(live.get("value"), dict) \
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 None, why
if live.get("status") == "confirmed" and not _same_fact(fact.item, combined, live["value"]):
# 只说已确认的说法里没有的那几段:已有的段算成「多出」,审核人会以为连它也被改了
extra = _and_join(live["value"].strip(), [str(m.value) for m in _members(fact)], fact.dialect,
keep_base=False)
adopted = f"被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" if not fact.members \
else "这次几个指标被采纳的写法"
return None, (f"「{fact.table}」的{label}已有人工确认过的说法「{live['value'].strip()}」,{adopted}"
f"另外多出条件「{extra or 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 None, (f"「{fact.table}」的{label}已有推断的说法「{_value_text(fact.item, live.get('value'))}」,"
f"与这次指标{names}采纳的「{value}」不同,无法确定两者的关系,未覆盖。"
"如需更改,请到数据目录中修改")
plan = catalog.plan_patch(notes, [{"path": fact.item, "value": value}], table=fact.table, tables=tables)
if plan.problems:
return None, f"「{fact.table}」的{label}未写入数据目录:{plan.problems[0]}"
change = plan.changes[0]
if change.state == "same":
return None, f"「{fact.table}」的{label}已是人工确认过的「{_value_text(fact.item, value)}」,无需重复写入"
if change.before_status == "confirmed":
return None, (f"「{fact.table}」的{label}已有人工确认过的说法「{_value_text(fact.item, change.before)}」,"
f"与这次指标{names}采纳的「{_value_text(fact.item, value)}」不同,未覆盖。"
"如需更改,请到数据目录中修改")
return change.after, None
def preview(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None,
notes: Mapping[str, Mapping[str, Any]] | None = None) -> dict[str, dict[str, Any]]:
"""审批卡上的预告:审核人在这个指标上选原流程(a)、选盲算(b),按目前的数据目录各会往目录里写什么。
candidate 同 derive(choice 不用给);notes 是这个数据源的目录({表名: notes})。返回 {a: …, b: …},每边是
{write: true, source, table, item, value}(value 是真要写进去的说法:和目录里已有的有效记录条件合起来的,是合起来
之后的整句)或 {write: false, reason}。和恢复后真写是同一套判断,只在模块说明里说的两处可能不一样。永远不抛。"""
notes = notes or {}
tables = schema_cache.get("tables") if isinstance(schema_cache, Mapping) else None
out: dict[str, dict[str, Any]] = {}
for side in SIDE_LABEL:
try:
fact, why = derive({**candidate, "choice": side}, schema_cache, notes)
value = None
if fact is not None:
value, why = _plan(notes.get(fact.table) or {}, fact, tables)
except Exception: # noqa: BLE001 - 预告算不出来不能拦住审批:照实说,以写入后的结果为准
logger.exception("双盲复算预告数据目录的写入失败(指标 %s)", candidate.get("metric"))
fact, why = None, f"指标「{candidate.get('metric')}」的选择核对数据目录时出错,无法预告是否写入"
out[side] = {"write": True, "source": fact.source, "table": fact.table, "item": fact.item, "value": value} \
if fact is not None and why is None else {"write": False, "reason": why}
return out
async def _write(session: Any, row: DataSource, fact: Fact, *, run_id: str, node_title: str,
actor: str | None) -> tuple[bool, list[str]]:
"""写一项:(写了没有, 没写的原因)。几个指标并成的一项先按读到的目录拿掉各自带回驳回说法的指标(_screen),写什么、
写不写再由 _plan 定;备注只列真写进去的指标。版本对不上(别人刚改过)重读再来。"""
label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None
reasons: list[str] = []
for _ in range(_RETRIES):
entry = await catalog.read_entry(session, row.id, fact.table)
notes = entry.notes if entry else {}
version = entry.version if entry else 0
kept, reasons = _screen(notes, fact)
if kept is None:
return False, reasons
value, why = _plan(notes, kept, tables)
if why is not None:
return False, [*reasons, why]
note = note_of(kept, run_id=run_id, node_title=node_title)
at = catalog.now_iso()
try:
merged = catalog.apply_patch(notes, [{"path": kept.item, "value": value, "reason": note}],
table=kept.table, tables=tables, at=at)
except catalog.CatalogInvalid as e:
return False, [*reasons, f"「{kept.table}」的{label}未写入数据目录:{e}"]
# 不论原来是谁提的:这一项现在出自人的选择,记为人工填写、已确认,备注写明出处
merged[kept.item] = catalog.make_item(value, "human", "confirmed", note=note, at=at)
try:
await catalog.write_entry(session, row.id, kept.table, merged, if_version=version, actor=actor)
except catalog.CatalogConflict:
await session.rollback()
continue
except catalog.CatalogInvalid as e:
return False, [*reasons, f"「{kept.table}」的{label}未写入数据目录:{e}"]
return True, reasons
return False, [*reasons, f"「{fact.table}」的数据目录正被修改,{label}未能写入,请稍后在数据目录中确认"]
__all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "preview", "record_choices"]
+63 -1
View File
@@ -60,7 +60,8 @@ async def launch_run(
运行也没有开始执行。
参数和 POST /api/runs 的请求体一一对应(input 叫 input_payload);actor 是记在运行上的发起人。
schedule_id 是发起它的定时任务,options 是整次运行沿用的选项快照,都原样交给 run_manager.start。
schedule_id 是发起它的定时任务,options 是整次运行沿用的选项快照,都原样交给 run_manager.start;图里有双盲复算、
反方审稿节点时 options 另加核查设置的快照(_with_verify_snapshot)。
"""
# 没挂在工作流上(画布没保存就发起、问数据当场搭的)。老数据里是「临时图」,前端两种都认
name = "未保存的工作流"
@@ -135,6 +136,7 @@ async def launch_run(
if not graph:
raise LaunchRefused(400, "没有要运行的内容:请求中既没有工作流 ID,也没有工作流内容")
await _refuse_missing_tools(session, graph)
options = await _with_verify_snapshot(session, graph, options)
from app.api.settings import resolve_run_scope
@@ -178,6 +180,66 @@ async def launch_run(
return run
#: 运行中途要读核查设置的节点:双盲复算、反方审稿
_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"]。
这两种节点在运行中途读设置(开关、上限、核查模型)。审批停下期间有人改了设置,恢复后的那一段读到的就和前面
不一样:本该复算的被关掉了、重放时走了另一条路,正式运行也说不清当时按什么规则核的。所以在发起这一刻定下来,
节点只读运行选项里的这一份(同 _approval_default 的道理;子工作流的节点经运行上下文拿到的是同一份选项)。没有
这两种节点的图 options 原样不动,run.started 和以前逐字相同;调用方已经给了 options["verify"] 的照它的。
"""
from app.api.settings import VERIFY_GROUP, verify_settings
if isinstance(options, dict) and VERIFY_GROUP in options:
return options
try:
spec = GraphSpec.model_validate(graph)
except Exception: # noqa: BLE001 - 图本身不成形交给 run_manager.start 报
return options
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
# --------------------------------------------------------------------------
# 发起前检查绑定的工具
# --------------------------------------------------------------------------
File diff suppressed because it is too large Load Diff
+106
View File
@@ -425,6 +425,110 @@ def _check_report_sources(node: GraphNode, spec: GraphSpec, result: ValidationRe
node_id=node.id, field="metrics_from", code=code)
#: 双盲复算节点各配置项在检查器上的叫法(前端 nodeDefs 的 dual_check 字段)。按 labels.py 的规矩应放在那里,
#: 那个文件不归双盲复算的工作包,先放在校验旁边,节点执行时报错也从这里取
DUAL_CHECK_FIELD_LABEL = {
"card": "口径卡", "metrics": "复算的指标", "definitions": "口径说明", "provider": "盲算模型 · 接入",
"model": "盲算模型", "tolerance": "允许的相对误差", "on_reject": "未采用原流程时", "enabled": "启用双盲复算",
}
#: 审核人没有采用原流程的结果(选了盲算,或者判断两边都不对)时:degrade 计入缺口、出具降档(默认)/ fail 判为失败
DUAL_CHECK_ON_REJECT = ("degrade", "fail")
DUAL_CHECK_ON_REJECT_LABEL = {"degrade": "计入缺口、出具降档", "fail": "判为失败"}
def dual_check_card(spec: GraphSpec, node: GraphNode) -> tuple[str | None, str | None]:
"""双盲复算要复算的口径卡:(节点 id, None) 或 (None, 为什么认不出)。写了「口径卡」(card)就是它,否则取上游唯一的
那张口径卡。校验和节点执行用的是这一个认法;写了的那个节点存不存在、是不是口径卡、在不在上游由调用方查。"""
card = node.config.get("card")
if card not in (None, ""):
return (card, None) if isinstance(card, str) else (None, f"「{DUAL_CHECK_FIELD_LABEL['card']}」需要选择一个口径卡节点")
upstream = _ancestors(spec, node.id)
cards = [n for n in spec.nodes if n.id in upstream and n.type == NodeType.METRICS]
if not cards:
return None, "「双盲复算」的上游没有口径卡:它复算的是口径卡上的指标,请把它接在口径卡之后"
if len(cards) > 1:
names = "、".join(f"「{n.title}」" for n in cards[:5])
return None, (f"「双盲复算」的上游有 {len(cards)} 张口径卡({names}),"
f"请在「{DUAL_CHECK_FIELD_LABEL['card']}」中选择要复算的那一张")
return cards[0].id, None
def _check_dual_check(node: GraphNode, spec: GraphSpec, result: ValidationResult) -> None:
"""双盲复算的配置:口径卡必须是它上游的口径卡;复算的指标、口径说明的键必须是那张卡上的指标 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)
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",
"dual_check.option_invalid")
tolerance = cfg.get("tolerance")
if tolerance not in (None, "") and (isinstance(tolerance, bool) or not isinstance(tolerance, (int, float))
or not math.isfinite(tolerance) or tolerance < 0):
add(f"「{label['tolerance']}」需要填写不小于 0 的数(例如 0.000001 表示百万分之一),留空表示按设置,"
f"当前为「{tolerance}」", "tolerance", "dual_check.option_invalid")
card_id, problem = dual_check_card(spec, node)
if problem:
add(problem, "card", "dual_check.card_invalid")
return
target = spec.node_map().get(card_id or "")
if target is None:
add(f"「{label['card']}」选择了「{card_id}」,但工作流中不存在这个节点", "card", "dual_check.card_invalid")
return
if target.type != NodeType.METRICS:
add(f"「{label['card']}」选择了「{target.title}」,它是「{type_label(target.type)}」节点,不是口径卡;"
"双盲复算只能复算口径卡上的指标", "card", "dual_check.card_invalid")
return
if target.id not in _ancestors(spec, node.id):
add(f"「{label['card']}」选择的口径卡「{target.title}」不在双盲复算的上游:复算时这张卡还没有计算。"
"请把口径卡连到双盲复算之前", "card", "dual_check.card_invalid")
return
if target.config.get("caliber_from") not in (None, "", {}):
known = None
else:
known = [str(d.get("id")) for d in target.config.get("metrics") or [] if isinstance(d, dict) and d.get("id")]
hint = (",可选的指标 ID:" + "、".join(known[:12])) if known else ""
metrics = cfg.get("metrics")
if metrics not in (None, "", []):
if not isinstance(metrics, list) or not all(isinstance(m, str) and m.strip() for m in metrics):
add(f"「{label['metrics']}」需要填写口径卡上的指标 ID 列表,留空表示复算输入来自查询的全部指标", "metrics",
"dual_check.metrics_invalid")
elif known is not None and (unknown := [m for m in metrics if m not in known]):
add(f"「{label['metrics']}」中的「{'、'.join(unknown[:5])}」不是口径卡「{target.title}」上的指标 ID{hint}",
"metrics", "dual_check.metrics_invalid")
definitions = cfg.get("definitions")
if definitions not in (None, "", {}):
if not isinstance(definitions, dict) or not all(isinstance(k, str) and isinstance(v, str)
for k, v in definitions.items()):
add(f"「{label['definitions']}」需要写成「指标 ID → 口径说明」的对照,口径说明写成文字", "definitions",
"dual_check.definitions_invalid")
elif known is not None and (unknown := [k for k in definitions if k not in known]):
add(f"「{label['definitions']}」中的「{'、'.join(unknown[:5])}」不是口径卡「{target.title}」上的指标 ID{hint}",
"definitions", "dual_check.definitions_invalid")
def _produces_query(node: GraphNode) -> bool:
"""这个节点的产出是一份存成快照的查询结果:调用数据库查询工具的「调用工具」节点,或另一个合并查询。"""
if node.type == NodeType.MERGE:
@@ -674,6 +778,8 @@ def validate_graph(spec: GraphSpec) -> ValidationResult:
if not d.get("id") or not d.get("expression"):
result.add(f"指标定义缺少 ID 或表达式:{d.get('id') or '(空)'}",
node_id=node.id, field=f"metrics[{i}]")
elif node.type == NodeType.DUAL_CHECK:
_check_dual_check(node, spec, result)
elif node.type == NodeType.REPORT:
_check_report_sources(node, spec, result)
_check_report_claims(node, spec, result)
+155
View File
@@ -0,0 +1,155 @@
"""双盲复算测试用的虚构零售库和盲算模型的剧本。表名、数字、地名都是编的。
订单表埋了两个陷阱:退款行(status = 'refunded'),以及下单日和支付日不在同一周的单。2026-09-01 至 2026-09-07 这一周:
- 按支付日、排除退款(口径说明要的):1、2、5 号单,330.0
- 按下单日、排除退款:2、4、5 号单,310.0
- 按支付日、不排除退款:1、2、3、5 号单,380.0
- 按下单日、不排除退款:2、3、4、5 号单,360.0
- 付款客户数(按支付日、排除退款):按行计数 3,去重计数 2
另有渠道(channel)和测试单标记(is_test,都是 0)两列:写数据目录的测试拿它们凑「指标自己的筛选范围」和标记列。
"""
from __future__ import annotations
import json
import re
import sqlite3
import uuid
from pathlib import Path
from langchain_core.messages import AIMessage, ToolMessage
from app.data import introspect
from app.db.base import SessionLocal
from app.db.models import DataSource
ROWS = [
(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'"
#: 口径说明要的写法:按支付日、排除退款
RIGHT = f"SELECT SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}"
#: 只错在时间列:按下单日
BY_ORDER_DATE = f"SELECT SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK}"
#: 只错在没排除退款
WITH_REFUNDS = f"SELECT SUM(amount) AS gmv FROM orders WHERE paid_date {WEEK}"
#: 两处都错
BOTH_WRONG = f"SELECT SUM(amount) AS gmv FROM orders WHERE order_date {WEEK}"
#: 付款客户数:按行计数 / 去重计数
BUYERS = f"SELECT COUNT(customer_id) AS buyers FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}"
BUYERS_DISTINCT = (f"SELECT COUNT(DISTINCT customer_id) AS buyers FROM orders "
f"WHERE status <> 'refunded' AND paid_date {WEEK}")
#: 按地区分组:华东一行是 300.0(1、2 号单)
BY_REGION = (f"SELECT region, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK} "
"GROUP BY region ORDER BY region")
DEFINITION = "已支付订单的金额合计:按支付日期(paid_date)归属到当周,不含退款"
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, channel TEXT, is_test INTEGER);")
db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ROWS)
db.commit()
db.close()
return path
async def add_source(path: Path, *, options: dict | None = None) -> tuple[str, str]:
"""登记成只读的手工源,探查好表结构。返回 (数据源名, id)。"""
name = f"dc_shop_{uuid.uuid4().hex[:8]}"
async with SessionLocal() as session:
row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True,
options=dict(options or {}))
session.add(row)
await session.flush()
row.schema_cache = await introspect.introspect(row)
await session.commit()
return name, row.id
def _text(messages) -> str:
return "\n".join(m.content if isinstance(m.content, str) else json.dumps(m.content, ensure_ascii=False)
for m in messages)
class Blind:
"""盲算模型的剧本,也管图里别的模型调用。
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,
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] = []
blind = self
def decide(model, messages):
text = _text(messages)
plan = blind.plan_of(text)
if model.response_format:
blind.calls.append("final")
done = len(re.findall(r"【证据 Q\d+】", text))
if plan is None or not done:
return AIMessage(content=json.dumps({"value": None, "rationale": "没有查到。"}, ensure_ascii=False))
_, column = plan
alias = f"Q{len(blind.sqls(plan))}"
body = {"value": {"alias": alias, "row": blind.row, "column": column},
"rationale": f"按口径说明统计,结果是 [[v:{alias}.r{blind.row}.{column}]]。"}
return AIMessage(content=json.dumps(body, ensure_ascii=False))
if model.tools:
blind.calls.append("step")
# 盲算的任务是第一条用户消息;同一次盲算的几轮收到的是同一条,只记一次
task_text = next((str(m.content) for m in messages if m.type == "human"), "")
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])
return AIMessage(content="查完了。")
blind.writer_prompts.append(text)
return AIMessage(content=blind.writer)
monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
def plan_of(self, text: str):
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
@staticmethod
def sqls(plan) -> list[str]:
sql = plan[0]
return list(sql) if isinstance(sql, (list, tuple)) else [sql]
@property
def runs(self) -> int:
"""盲算 agent 一共执行了几次(每次收尾交一次结果)。"""
return self.calls.count("final")
+243
View File
@@ -0,0 +1,243 @@
"""双盲复算的配置、发起时的设置快照、助手的节点说明。
- 画布校验(engine/schema.py):口径卡必须是上游的口径卡(没写时上游只能有一张),复算的指标、口径说明的键必须是
那张卡上的指标 ID,开关(只认 true / false,0、1 不算)、误差、处置的取值,不能配失败重试;报错写检查器上的中文叫法
- 发起(engine/launch.py):图里(含子工作流)有双盲复算或反方审稿节点时把核查设置快照进 options["verify"],没有时
options 不变、run.started 和以前逐字相同;调用方给了的照它的
- 助手的节点说明(api/copilot.py 的 NODE_REFERENCE)有 dual_check 一条:什么时候加、各配置项
表名、数字都是虚构的。
"""
from __future__ import annotations
import asyncio
import pytest
from sqlalchemy import delete, select
from app.db.base import SessionLocal
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
def node(nid, ntype, label=None, **config):
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}}
CARD = node("card", "metrics", "销售口径卡", caliber="销售口径", metrics=[
{"id": "gmv", "name": "销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"},
{"id": "orders", "name": "订单数", "expression": "cell(nodes.fetch, 0, 'n')"}])
def spec(*middle: dict, dual: dict | None = None, edges: list | None = None) -> dict:
nodes = [node("start", "input"), node("fetch", "tool", tool="db_query__shop", args={"sql": "SELECT 1"}), *middle,
node("dual", "dual_check", "复核", **(dual or {})), node("out", "output", fields=[])]
chain = edges or [(a["id"], b["id"]) for a, b in zip(nodes, nodes[1:])]
return {"nodes": nodes, "edges": [{"source": s, "target": t} for s, t in chain]}
def problems(graph: dict) -> list[tuple[str | None, str]]:
report = validate_graph(GraphSpec.model_validate(graph))
return [(i.field, i.message) for i in report.issues if i.level == "error" and i.node_id == "dual"]
# --------------------------------------------------------------------------
# 画布校验
# --------------------------------------------------------------------------
def test_a_valid_config_has_no_problems():
assert problems(spec(CARD, dual={"card": "card", "metrics": ["gmv"], "definitions": {"gmv": "按支付日期,不含退款"},
"tolerance": 0.001, "on_reject": "fail", "enabled": True})) == []
# 不写口径卡:上游只有一张,就是它
assert problems(spec(CARD)) == []
def test_the_card_must_be_an_upstream_metrics_card():
[(where, text)] = problems(spec(CARD, dual={"card": "fetch"}))
assert where == "card" and "「口径卡」选择了「fetch」" in text and "不是口径卡" in text
[(where, text)] = problems(spec(CARD, dual={"card": "ghost"}))
assert where == "card" and "不存在这个节点" in text
# 口径卡接在双盲复算后面:复算时它还没算
nodes = [node("start", "input"), node("dual", "dual_check", "复核", card="card"), CARD]
graph = {"nodes": nodes, "edges": [{"source": "start", "target": "dual"}, {"source": "dual", "target": "card"}]}
[(where, text)] = problems(graph)
assert where == "card" and "不在双盲复算的上游" in text
def test_without_a_card_there_must_be_exactly_one_upstream():
[(where, text)] = problems(spec())
assert where == "card" and "上游没有口径卡" in text
second = node("card2", "metrics", "利润口径卡", metrics=[{"id": "profit", "expression": "1"}])
[(where, text)] = problems(spec(CARD, second))
assert where == "card" and "2 张口径卡" in text and "「销售口径卡」" in text and "「利润口径卡」" in text
assert problems(spec(CARD, second, dual={"card": "card2"})) == []
def test_metrics_and_definitions_must_be_on_the_card():
found = dict(problems(spec(CARD, dual={"metrics": ["gmv", "refund"],
"definitions": {"gmv": "按支付日", "cost": "成本"}})))
assert "「复算的指标」中的「refund」不是口径卡「销售口径卡」上的指标 ID" in found["metrics"]
assert "可选的指标 ID:gmv、orders" in found["metrics"]
assert "「口径说明」中的「cost」不是口径卡「销售口径卡」上的指标 ID" in found["definitions"]
[(where, text)] = problems(spec(CARD, dual={"definitions": ["gmv"]}))
assert where == "definitions" and "指标 ID → 口径说明" in text
def test_a_card_pinned_elsewhere_is_not_checked_for_metric_ids():
pinned = node("card", "metrics", "引用的口径卡", caliber_from={"workflow_id": "w", "workflow_version": 1, "node_id": "c"})
assert problems(spec(pinned, dual={"definitions": {"anything": "说明"}})) == []
def test_option_values_are_checked():
found = dict(problems(spec(CARD, dual={"enabled": "yes", "on_reject": "ignore", "tolerance": -1})))
assert "「启用双盲复算」" in found["enabled"]
assert "「未采用原流程时」只能选「计入缺口、出具降档」或「判为失败」" in found["on_reject"]
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})) == []
# --------------------------------------------------------------------------
# 发起:设置快照
# --------------------------------------------------------------------------
@pytest.fixture
async def engine():
await run_manager.setup()
async with SessionLocal() as session:
await session.execute(delete(Setting).where(Setting.key == "verify"))
await session.commit()
yield
async with SessionLocal() as session:
await session.execute(delete(Setting).where(Setting.key == "verify"))
await session.commit()
await run_manager.shutdown()
async def settle(run_id: str) -> Run:
for _ in range(400):
await asyncio.sleep(0.03)
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
await run_manager.wait_idle(run_id)
return row
raise AssertionError(f"运行 {run_id} 停在 {row.status}")
async def started_of(run_id: str) -> dict:
async with SessionLocal() as session:
return (await session.execute(select(RunEvent).where(
RunEvent.run_id == run_id, RunEvent.type == "run.started"))).scalars().first().data
PLAIN = {"nodes": [node("start", "input"), node("out", "output", fields=[{"name": "r", "value": "完成"}])],
"edges": [{"source": "start", "target": "out"}]}
GATED = {"nodes": [node("start", "input"), node("card", "metrics", metrics=[{"id": "x", "expression": "1"}]),
node("dual", "dual_check"), node("out", "output", fields=[])],
"edges": [{"source": "start", "target": "card"}, {"source": "card", "target": "dual"},
{"source": "dual", "target": "out"}]}
async def test_a_graph_with_a_verify_node_snapshots_the_settings(engine):
from app.api.settings import verify_settings
from app.engine.launch import launch_run
async with SessionLocal() as session:
session.add(Setting(key="verify", value={"dual_check": {"tolerance": 0.01}, "daily_max_usd": 3}))
await session.commit()
expected = await verify_settings(session)
run = await launch_run(session, graph=GATED, actor="测试")
assert expected["dual_check"]["tolerance"] == 0.01 and run.options == {"verify": expected}
row = await settle(run.id)
assert row.status == "succeeded", row.error
assert (await started_of(run.id))["options"] == {"verify": expected}
async def test_a_challenge_node_also_snapshots_the_settings(engine):
from app.engine.launch import _with_verify_snapshot
graph = {"nodes": [node("start", "input"), node("review", "challenge")], "edges": []}
async with SessionLocal() as session:
out = await _with_verify_snapshot(session, graph, None)
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
async with SessionLocal() as session:
run = await launch_run(session, graph=PLAIN, actor="测试")
assert run.options == {}
await settle(run.id)
assert "options" not in await started_of(run.id)
async def test_options_given_by_the_caller_are_kept(engine):
from app.engine.launch import launch_run
mine = {"verify": {"dual_check": {"enabled": False}}, "note": "例行"}
async with SessionLocal() as session:
run = await launch_run(session, graph=GATED, actor="测试", options=mine)
assert run.options == mine
await settle(run.id)
# --------------------------------------------------------------------------
# 助手的节点说明
# --------------------------------------------------------------------------
def test_the_copilot_reference_explains_when_and_how_to_add_it():
from app.api.copilot import NODE_REFERENCE
assert "- dual_check:" in NODE_REFERENCE
block = NODE_REFERENCE.split("- dual_check:", 1)[1].split("\n- ", 1)[0]
for key in ("card", "metrics", "definitions", "provider", "model", "tolerance", "on_reject", "enabled"):
assert key in block, key
assert "正式出具" in block and "口径卡之后" in block and "degrade" in block and "fail" in block
# 排在口径卡和报告之间:搭图的先后就是这样
at = NODE_REFERENCE.index
assert at("- metrics:") < at("- dual_check:") < at("- report:")
+209
View File
@@ -0,0 +1,209 @@
"""单独调用双盲复算的执行器(不在图里执行:@task 直接执行,once 直接发),测图里不好凑的几种情形。
- 盲算模型和产出原数据的 Agent 是同一个模型时提醒;调用工具节点写死的 SQL 没有模型参与,不提醒
- 合并查询、整段结果、遮罩列、文本这类输入不能单独复算:记缺口,一分钱不花
- 口径卡没执行到的直接放行;配置里写了卡上没有的指标 ID 报节点错误
查询快照是现造的工件,表名、数字都是虚构的。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
import pytest
from sqlalchemy import delete
from app.core import artifact_store
from app.db.base import SessionLocal
from app.db.models import Setting
from app.engine.context import NodeContext, NodeError, RunContext
from app.engine.evidence import cell_eid
from app.engine.nodes.dual_check import run_dual_check
from app.engine.runner import RUN_OPTIONS_KEY
from app.engine.schema import GraphSpec
from app.tools.datasource import RUN_VERSIONS_KEY
from tests.fixtures.dual_check_shop import RIGHT, Blind, add_source, build
from tests.fixtures.sources import drop_source
def node(nid, ntype, label=None, **config):
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}}
@dataclass
class Recorder(NodeContext):
"""记下节点发的事件。不在图里执行:@task 直接执行,once 直接发。"""
events: list[tuple[str, dict[str, Any]]] = field(default_factory=list)
def emit(self, event_type, **data):
self.events.append((str(event_type), data))
def logs(self, code: str) -> list[dict[str, Any]]:
return [d for t, d in self.events if t == "log" and d.get("code") == code]
@pytest.fixture
async def shop(tmp_path):
async with SessionLocal() as session:
await session.execute(delete(Setting).where(Setting.key == "verify"))
await session.commit()
name, source_id = await add_source(build(tmp_path / "shop.db"), options={"mask_columns": "secret_gmv"})
yield name
await drop_source(source_id)
async def snapshot(source: str, *, columns=("gmv",), rows=((330.0,),), sql=RIGHT, **extra) -> str:
content = {"columns": list(columns), "rows": [list(r) for r in rows], "row_count": len(rows), "truncated": False,
"sql": sql, "source": source, **extra}
return await artifact_store.put_json(content, kind="query_snapshot", run_id="unit", node_id="agent")
def context(graph: dict, node_id: str = "dual") -> Recorder:
parsed = GraphSpec.model_validate(graph)
run = RunContext(run_id="unit-run", thread_id="unit-run", spec=parsed,
extra={RUN_OPTIONS_KEY: {"verify": {}}, RUN_VERSIONS_KEY: None})
return Recorder(node=parsed.node_map()[node_id], run=run)
AGENT_GRAPH = {"nodes": [node("start", "input"), node("agent", "agent", prompt="查销售额"),
node("card", "metrics", metrics=[{"id": "gmv", "expression": "vars.kpi.gmv"}]),
node("dual", "dual_check", "复核"), node("out", "output", fields=[])],
"edges": [{"source": a, "target": b} for a, b in
(("start", "agent"), ("agent", "card"), ("card", "dual"), ("dual", "out"))]}
def card_of(*metrics: dict) -> dict:
return {"kind": "metric_set", "caliber": "销售口径", "caliber_version": "v1", "metrics": list(metrics),
"artifact": "card-artifact"}
def metric(artifact: str, *, path: str = "vars.kpi.gmv", locator: dict | None = None, value: Any = 330.0,
via: str = "agent_field", node_id: str = "agent", expression: str | None = None) -> dict:
where = locator or {"row": 0, "column": "gmv"}
return {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands", "value": 330.0,
"rendered": "330.0元", "expression": expression or path, "substituted": "330.0", "status": "ok",
"inputs": [{"path": path, "value": value, "node_id": node_id, "via": via, "field": "gmv",
"artifact": artifact, "locator": where, "status": "verified",
**({"eid": cell_eid(artifact, where["row"], where["column"])} if "row" in where else {})}]}
async def test_the_same_model_as_the_agent_that_fetched_the_data_is_flagged(monkeypatch, shop):
art = await snapshot(shop)
Blind(monkeypatch, shop, {"销售额": (RIGHT, "gmv")})
ctx = context(AGENT_GRAPH)
state = {"nodes": {"agent": {"model": "mock-fast", "data": {"gmv": 330.0}}, "card": card_of(metric(art))},
"evidence": [], "trail": []}
out = (await run_dual_check(state, ctx))["nodes"]["dual"]
assert out["status"] == "agreed" and out["same_model"] is True and out["model"] == "mock-fast"
[warned] = ctx.logs("dual_check_same_model")
assert warned["model"] == "mock-fast" and "盲算模型与产出原数据的模型相同" in warned["message"]
assert "设置 → 偏好设置 → 核查" in warned["message"]
async def test_data_fetched_by_fixed_sql_raises_no_same_model_warning(monkeypatch, shop):
art = await snapshot(shop)
Blind(monkeypatch, shop, {"销售额": (RIGHT, "gmv")})
ctx = context(AGENT_GRAPH)
state = {"nodes": {"agent": {"model": "mock-fast"},
"card": card_of(metric(art, path="cell(nodes.fetch, 0, 'gmv')", via="tool_cell",
node_id="fetch"))},
"evidence": [], "trail": []}
out = (await run_dual_check(state, ctx))["nodes"]["dual"]
assert out["status"] == "agreed" and out["same_model"] is False and not ctx.logs("dual_check_same_model")
@pytest.mark.parametrize("case, reason", [
("merge", "来自合并查询"),
("rows", "取的是一整段查询结果"),
("masked", "设置了遮罩的列"),
("text", "不是数值"),
])
async def test_inputs_that_cannot_be_checked_alone_become_gaps(monkeypatch, shop, case, reason):
blind = Blind(monkeypatch, shop, {"销售额": (RIGHT, "gmv")})
if case == "merge":
art = await snapshot("合并查询", sources=[shop])
item = metric(art)
elif case == "rows":
art = await snapshot(shop, rows=((100.0,), (230.0,)))
item = metric(art, path="vars.kpi.weeks", expression="sum(vars.kpi.weeks)",
locator={"rows": [0, 1], "column": "gmv"}, value="[2 项]")
elif case == "masked":
art = await snapshot(shop, columns=("secret_gmv",))
item = metric(art, locator={"row": 0, "column": "secret_gmv"})
else:
art = await snapshot(shop, columns=("gmv",), rows=(("三百三",),))
item = metric(art, value="三百三") # 指标值是数,输入却是文本(比如拿它做了判断)
ctx = context(AGENT_GRAPH)
state = {"nodes": {"agent": {"model": "other"}, "card": card_of(item)}, "evidence": [], "trail": []}
out = (await run_dual_check(state, ctx))["nodes"]["dual"]
[row] = out["metrics"]
assert row["status"] == "unchecked" and reason in row["reason"]
assert out["issuance_gaps"] == [f"指标「销售额」无法复算:{row['reason']}"]
assert blind.calls == [], "不能单独复算的输入也花了钱"
async def test_a_card_that_did_not_run_passes_through(shop):
ctx = context(AGENT_GRAPH)
out = await run_dual_check({"nodes": {}, "evidence": [], "trail": []}, ctx)
assert out == {"nodes": {"dual": {"status": "skipped", "reason": "no_inputs"}}}
assert [d["reason"] for d in ctx.logs("dual_check_skipped")] == ["no_inputs"]
async def test_unknown_metric_ids_fail_the_node(shop):
art = await snapshot(shop)
graph = {**AGENT_GRAPH, "nodes": [*AGENT_GRAPH["nodes"][:3], node("dual", "dual_check", "复核", metrics=["refund"]),
AGENT_GRAPH["nodes"][4]]}
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
+581
View File
@@ -0,0 +1,581 @@
"""双盲复算的选择写进数据目录(engine/dual_check_facts.py)。
- 只差一个方面、又对得上目录事实时才写:两边按不同的时间列筛 → 业务日期,一边多一个排除无效记录的条件 → 有效记录
条件,聚合只差去不去重、被采纳的一边去重 → 去重规则;按被采纳那一边的写法记
- 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法、只有一边按时间筛:不写,
照实说原因
- 指标自己的筛选范围(只看华东、只看线上)不是有效记录条件:取值出现在指标名称、口径说明、所在分组、运行输入里的,
条件用到分组的列的,不是在排除退款、作废、测试这类无效记录的,都不写
- 写的是人工填写、已确认,备注写明运行、节点、指标;同样的说法已确认过就不再写(不升版本),确认过的不同说法不覆盖;
目录里已有的有效记录条件被采纳的一边也用着的,合起来写(已有的段只写一次),没用着的不动;驳回过的说法不写回
(!= 和 <>、列名大小写、先后不同也认得出)
- 同一次裁决里几个指标落到同一项上:说法一样的写一次;有效记录条件(都在排除无效记录)用 AND 合成一句写;业务日期
说法不同的都不写;自己的说法驳回过的指标拿出来不写、其余照写(和各自的预告一致),合起来又成了驳回的整句也不写
- 审批卡上的预告(preview):选 a、选 b 各会写什么、不写的为什么,按目录里已有的说法算,和真写同一套判断
差异条目一律由 sql_semantics.explain_sql_diff 真算出来,和节点交给审核人看的是同一份。表名、数字都是虚构的。
"""
from __future__ import annotations
import pytest
from app.data import catalog
from app.data.sql_semantics import explain_sql_diff
from app.db.base import SessionLocal
from app.engine.dual_check_facts import derive, preview, record_choices
from tests.fixtures.dual_check_shop import (
BOTH_WRONG,
BUYERS,
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"}, {"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",
scope: dict | None = None) -> dict:
diff = explain_sql_diff(original, blind, "sqlite", schema=SCHEMA, labels=("原流程", "盲算")).as_dict()
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)
# --------------------------------------------------------------------------
# 差异 → 目录事实
# --------------------------------------------------------------------------
@pytest.mark.parametrize("choice, column", [("b", "paid_date"), ("a", "order_date")])
def test_a_different_time_column_becomes_the_business_date(choice, column):
fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, choice), SCHEMA)
assert why is None
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():
fact, why = derive(candidate(RIGHT, WITH_REFUNDS, "b"), SCHEMA)
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")
fact, why = derive(candidate(WITH_REFUNDS, qualified, "b"), SCHEMA)
assert why is None and fact.value == "status <> 'refunded'"
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():
fact, why = derive(candidate(BOTH_WRONG, RIGHT, "b"), SCHEMA)
assert fact is None and "不止差一个方面" in why
def test_equivalent_sql_or_fallback_or_other_aspects_are_not_recorded():
same = candidate(RIGHT, RIGHT.replace("SUM(amount) AS gmv", "SUM(amount) AS total"), "b")
assert same["diff"]["equivalent"] is True
assert "规范化后一致" in derive(same, SCHEMA)[1]
fallback = candidate(BY_ORDER_DATE, RIGHT, "b")
fallback["diff"]["fallback"] = True
assert "未能归纳" in derive(fallback, SCHEMA)[1]
joined = candidate(RIGHT, RIGHT.replace("SUM(amount)", "MAX(amount)"), "b")
assert [i["aspect"] for i in joined["diff"]["items"]] == ["aggregate"]
assert "不只是去不去重" in derive(joined, SCHEMA)[1]
ranged = candidate(RIGHT, RIGHT.replace("'2026-09-07'", "'2026-09-06'"), "b")
assert [i["aspect"] for i in ranged["diff"]["items"]] == ["time_range"]
assert "时间范围" in derive(ranged, SCHEMA)[1]
assert "没有采用任何一边" in derive({**candidate(BY_ORDER_DATE, RIGHT, "b"), "choice": "neither"}, SCHEMA)[1]
def test_a_table_missing_from_the_schema_is_not_recorded():
fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, "b"), {"tables": {"refunds": {"columns": []}}})
assert fact is None and "不在数据源的表结构里" in why
# --------------------------------------------------------------------------
# 写进目录
# --------------------------------------------------------------------------
@pytest.fixture
async def shop(tmp_path):
name, source_id = await add_source(build(tmp_path / "shop.db"))
yield name, source_id
await drop_source(source_id)
async def entry_of(source_id: str):
async with SessionLocal() as session:
return await catalog.read_entry(session, source_id, "orders")
async def seed(source_id: str, notes: dict) -> None:
async with SessionLocal() as session:
await catalog.write_entry(session, source_id, "orders", notes, if_version=0, actor="数据组")
async def seed_again(source_id: str, notes: dict) -> None:
"""把目录这张表整份改回 notes(在现有版本上写)。"""
async with SessionLocal() as session:
entry = await catalog.read_entry(session, source_id, "orders")
await catalog.write_entry(session, source_id, "orders", notes, if_version=entry.version, actor="数据组")
async def test_a_choice_is_written_as_a_confirmed_human_item_with_its_origin(shop):
name, source_id = shop
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-42",
node_title="复核销售额", actor="张三")
assert out == {"written": [{"source": name, "table": "orders", "item": "business_date"}], "skipped": []}
entry = await entry_of(source_id)
item = entry.notes["business_date"]
assert item["value"] == {"column": "paid_date"} and (item["source"], item["status"]) == ("human", "confirmed")
assert "run-42" in item["note"] and "「复核销售额」" in item["note"] and "「销售额」" in item["note"]
assert "盲算" in item["note"] and entry.version == 1 and entry.updated_by == "张三"
async def test_writing_the_same_choice_again_changes_nothing(shop):
name, source_id = shop
choice = [candidate(BY_ORDER_DATE, RIGHT, "b", source=name)]
await record_choices(choice, run_id="run-1", node_title="复核", actor=None)
again = await record_choices(choice, run_id="run-2", node_title="复核", actor=None)
assert again["written"] == [] and "无需重复写入" in again["skipped"][0]["reason"]
entry = await entry_of(source_id)
assert entry.version == 1 and "run-1" in entry.notes["business_date"]["note"]
async def test_a_confirmed_different_item_is_kept(shop):
name, source_id = shop
await seed(source_id, {"business_date": catalog.make_item({"column": "order_date", "rule": "按下单日"}, "human",
"confirmed")})
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-1",
node_title="复核", actor=None)
assert out["written"] == []
reason = out["skipped"][0]["reason"]
assert "已有人工确认过的说法「order_date」" in reason and "paid_date" in reason and "未覆盖" in reason
entry = await entry_of(source_id)
assert entry.version == 1 and entry.notes["business_date"]["value"]["column"] == "order_date"
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"}]
item = (await entry_of(source_id)).notes["valid_filter"]
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
await seed(source_id, {"business_date": catalog.make_item(
{"column": "paid_date", "rule": "按付款成功时间", "timezone": "Asia/Shanghai"}, "profile")})
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-1",
node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "business_date"}]
item = (await entry_of(source_id)).notes["business_date"]
assert item["value"] == {"column": "paid_date", "rule": "按付款成功时间", "timezone": "Asia/Shanghai"}
assert (item["source"], item["status"]) == ("human", "confirmed")
async def test_a_rejected_item_is_not_brought_back_but_a_rejected_other_value_does_not_block(shop):
"""同一个说法以前被人驳回过:两次人工判断对不上,不写;驳回的是别的说法:照写。"""
name, source_id = shop
await seed(source_id, {"business_date": catalog.make_item({"column": "paid_date"}, "llm", "rejected")})
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-1",
node_title="复核", actor=None)
assert out["written"] == [] and "曾被人工驳回" in out["skipped"][0]["reason"]
assert (await entry_of(source_id)).version == 1
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "a", source=name)], run_id="run-2",
node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "business_date"}]
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"]
assert "给出了不同的说法" in skipped["reason"] and "无法确定哪一个是数据事实" in skipped["reason"]
assert await entry_of(source_id) is None
async def test_complementary_valid_filters_in_one_decision_are_combined(shop):
"""同一次裁决里,一个指标采纳了排除退款、另一个采纳了排除测试单:两条都在排除无效记录,各有人的选择撑着,
不互相矛盾,用 AND 合成一句写。"""
name, source_id = shop
out = await record_choices([
candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name),
candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name),
# 第三个指标采纳的是和第一个一样的条件:合并时不重复
candidate(COUNT_WEEK, extra("status <> 'refunded'", COUNT_WEEK), "b", metric="订单数", source=name),
], run_id="run-1", node_title="复核", actor=None)
assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter"}], "skipped": []}
entry = await entry_of(source_id)
item = entry.notes["valid_filter"]
assert item["value"] == "status <> 'refunded' AND is_test = 0" and entry.version == 1
assert (item["source"], item["status"]) == ("human", "confirmed")
assert all(f"「{m}」" in item["note"] for m in ("销售额", "订单金额", "订单数")) and "run-1" in item["note"]
async def test_combined_filters_extend_an_inferred_condition_every_adopted_side_uses(shop):
"""目录里推断的「channel <> 'test'」:几个被采纳的一边都按它筛了,合起来写;有一个没按它筛的,关系拿不准,不动。"""
name, source_id = shop
await seed(source_id, {"valid_filter": catalog.make_item("channel <> 'test'", "llm")})
base = extra("channel <> 'test'")
both = [candidate(base, extra("status <> 'refunded'", base), "b", metric="销售额", source=name),
candidate(base, extra("is_test = 0", base), "b", metric="订单金额", source=name)]
out = await record_choices(both, run_id="run-1", node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out
assert (await entry_of(source_id)).notes["valid_filter"]["value"] == \
"channel <> 'test' AND status <> 'refunded' AND is_test = 0"
await seed_again(source_id, {"valid_filter": catalog.make_item("channel <> 'test'", "llm")})
one_without = [both[0], candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name)]
out = await record_choices(one_without, run_id="run-2", node_title="复核", actor=None)
assert out["written"] == []
reason = out["skipped"][0]["reason"]
assert "「channel <> 'test'」" in reason and "指标「订单金额」被采纳的盲算没有按它筛选" in reason and "未改动" in reason
assert (await entry_of(source_id)).notes["valid_filter"]["value"] == "channel <> 'test'"
async def test_a_rejected_condition_is_not_written_back_through_a_combined_decision(shop):
"""目录里「status <> 'refunded'」被人工驳回过。同一次裁决里「销售额」「订单数」采纳了排除退款、「订单金额」采纳了
排除测试单:审批卡上预告前两个不写(曾被人工驳回)、后一个写「is_test = 0」,真写也照这样。驳回过的条件不能借着
合并又写成人工确认,拿出来的指标照实说原因。"""
name, source_id = shop
rejected = {"valid_filter": catalog.make_item("status <> 'refunded'", "llm", "rejected")}
await seed(source_id, rejected)
refunds = candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name)
tests = candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name)
counted = candidate(COUNT_WEEK, extra("status <> 'refunded'", COUNT_WEEK), "b", metric="订单数", source=name)
notes = {"orders": (await entry_of(source_id)).notes}
assert preview(refunds, SCHEMA, notes)["b"]["write"] is False
assert preview(tests, SCHEMA, notes)["b"]["value"] == "is_test = 0"
out = await record_choices([refunds, tests, counted], run_id="run-1", node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out
[skipped] = out["skipped"]
assert "「status <> 'refunded'」曾被人工驳回" in skipped["reason"] and "「销售额」、「订单数」" in skipped["reason"]
assert "订单金额" not in skipped["reason"]
item = (await entry_of(source_id)).notes["valid_filter"]
assert item["value"] == "is_test = 0" and (item["source"], item["status"]) == ("human", "confirmed")
assert "「订单金额」" in item["note"] and "销售额" not in item["note"] and "订单数" not in item["note"]
# 并成一项的每个指标采纳的都是驳回过的说法:一个都不写,一条原因说清是哪几个指标
await seed_again(source_id, rejected)
out = await record_choices([refunds, counted], run_id="run-2", node_title="复核", actor=None)
assert out["written"] == [] and len(out["skipped"]) == 1, out
assert "曾被人工驳回" in out["skipped"][0]["reason"] and "「销售额」、「订单数」" in out["skipped"][0]["reason"]
assert (await entry_of(source_id)).notes["valid_filter"]["status"] == "rejected"
async def test_a_rejected_statement_is_not_rebuilt_from_several_metrics(shop):
"""驳回的是「is_test = 0 AND status != 'refunded'」整句:单看每个指标都不是这句(预告各写各的),合起来却正是这句
(!= 和 <>、先后不同也是同一个说法)。两次人工判断对不上,都不写;再并进一个别的条件,驳回的整句仍在里面,照样不写。"""
name, source_id = shop
rejected = catalog.make_item("is_test = 0 AND status != 'refunded'", "llm", "rejected")
await seed(source_id, {"valid_filter": rejected})
refunds = candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name)
tests = candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name)
channel = candidate(COUNT_WEEK, extra("channel <> 'test'", COUNT_WEEK), "b", metric="订单数", source=name)
notes = {"orders": (await entry_of(source_id)).notes}
assert all(preview(c, SCHEMA, notes)["b"]["write"] is True for c in (refunds, tests, channel))
for picked, names in (([refunds, tests], "「销售额」、「订单金额」"),
([refunds, tests, channel], "「销售额」、「订单金额」、「订单数」")):
out = await record_choices(picked, run_id="run-1", node_title="复核", actor=None)
assert out["written"] == [], out
[skipped] = out["skipped"]
assert "曾被人工驳回" in skipped["reason"] and names in skipped["reason"]
entry = await entry_of(source_id)
assert entry.version == 1 and entry.notes["valid_filter"]["status"] == "rejected"
@pytest.mark.parametrize("rejected", ["status != 'refunded'", "STATUS <> 'refunded'"])
async def test_a_rejected_condition_is_recognised_however_it_was_written(shop, rejected):
"""驳回时写的是 != 、列名大写:和这次要写的「status <> 'refunded'」是同一个说法,预告和真写都不写。"""
name, source_id = shop
await seed(source_id, {"valid_filter": catalog.make_item(rejected, "llm", "rejected")})
picked = candidate(WITH_REFUNDS, RIGHT, "b", source=name)
said = preview(picked, SCHEMA, {"orders": (await entry_of(source_id)).notes})
assert said["b"]["write"] is False and f"「{rejected}」曾被人工驳回" in said["b"]["reason"]
out = await record_choices([picked], run_id="run-1", node_title="复核", actor=None)
assert out == {"written": [], "skipped": [{"reason": said["b"]["reason"]}]}
assert (await entry_of(source_id)).version == 1
async def test_combined_filters_do_not_repeat_a_condition_already_in_the_catalog(shop):
"""目录里推断的「status <> 'refunded'」:「销售额」采纳了带着它的盲算,「订单金额」两边都按它筛、盲算另外排除了
测试单。合起来写时已有的条件只出现一次(正是「订单金额」的预告);已有的条件是人工确认过的,原因里只说另外多出
的那一段。"""
name, source_id = shop
await seed(source_id, {"valid_filter": catalog.make_item("status <> 'refunded'", "llm")})
base = extra("status <> 'refunded'")
picked = [candidate(SUM_WEEK, base, "b", metric="销售额", source=name),
candidate(base, extra("is_test = 0", base), "b", metric="订单金额", source=name)]
said = preview(picked[1], SCHEMA, {"orders": (await entry_of(source_id)).notes})
out = await record_choices(picked, run_id="run-1", node_title="复核", actor=None)
assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter"}], "skipped": []}
item = (await entry_of(source_id)).notes["valid_filter"]
assert item["value"] == said["b"]["value"] == "status <> 'refunded' AND is_test = 0"
await seed_again(source_id, {"valid_filter": catalog.make_item("status <> 'refunded'", "human", "confirmed")})
out = await record_choices(picked, run_id="run-2", node_title="复核", actor=None)
assert out["written"] == []
reason = out["skipped"][0]["reason"]
assert "已有人工确认过的说法「status <> 'refunded'」" in reason and "另外多出条件「is_test = 0」" in reason
assert (await entry_of(source_id)).notes["valid_filter"]["value"] == "status <> 'refunded'"
def test_conditions_are_joined_without_repeating_a_part():
"""几句条件接成一句:已经有的段不再接(整句接上会把已有的段再写一遍);只说多出的时不带原有的;最外层是 OR 的
只在接了别的条件时加括号。"""
from app.engine.dual_check_facts import _and_join
assert _and_join("status <> 'refunded'", ["status <> 'refunded' AND is_test = 0"], "sqlite") == \
"status <> 'refunded' AND is_test = 0"
assert _and_join("status <> 'refunded'", ["STATUS != 'refunded'", "is_test = 0", "is_test = 0"], "sqlite") == \
"status <> 'refunded' AND is_test = 0"
assert _and_join("status <> 'refunded'", ["status <> 'refunded' AND is_test = 0"], "sqlite",
keep_base=False) == "is_test = 0"
assert _and_join("status = 'void' OR is_test = 1", ["(status = 'void' OR is_test = 1)"], "sqlite") == \
"status = 'void' OR is_test = 1"
assert _and_join("status = 'void' OR is_test = 1", ["is_test = 0"], "sqlite") == \
"(status = 'void' OR is_test = 1) AND is_test = 0"
async def test_the_same_choice_from_two_metrics_is_written_once(shop):
name, source_id = shop
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", metric="销售额", source=name),
candidate(BY_ORDER_DATE, RIGHT, "b", metric="订单金额", source=name)],
run_id="run-1", node_title="复核", actor=None)
assert out == {"written": [{"source": name, "table": "orders", "item": "business_date"}], "skipped": []}
entry = await entry_of(source_id)
assert entry.version == 1 and "「销售额」" in entry.notes["business_date"]["note"]
assert "「订单金额」" in entry.notes["business_date"]["note"]
# --------------------------------------------------------------------------
# 审批卡上的预告:选哪一边各会写什么(和真写同一套判断)
# --------------------------------------------------------------------------
def test_the_preview_says_what_each_choice_would_write():
picked = candidate(BY_ORDER_DATE, RIGHT, "a")
assert preview(picked, SCHEMA, {}) == {
"a": {"write": True, "source": "shop", "table": "orders", "item": "business_date",
"value": {"column": "order_date"}},
"b": {"write": True, "source": "shop", "table": "orders", "item": "business_date",
"value": {"column": "paid_date"}},
}
# 只有被采纳的一边有条件时才写:选另一边照实说为什么不写
said = preview(candidate(WITH_REFUNDS, RIGHT, "a"), SCHEMA, {})
assert said["a"]["write"] is False and "被采纳的原流程没有这个条件" in said["a"]["reason"]
assert said["b"] == {"write": True, "source": "shop", "table": "orders", "item": "valid_filter",
"value": "status <> 'refunded'"}
def test_the_preview_reads_the_catalog_as_it_is():
"""目录里已有的说法照写入时的规矩算:确认过的不覆盖、同样的说法不重写、推断的有效记录条件合起来写。"""
confirmed = {"orders": {"business_date": catalog.make_item({"column": "order_date"}, "human", "confirmed")}}
said = preview(candidate(BY_ORDER_DATE, RIGHT, "a"), SCHEMA, confirmed)
assert said["a"]["write"] is False and "无需重复写入" in said["a"]["reason"]
assert said["b"]["write"] is False and "已有人工确认过的说法「order_date」" in said["b"]["reason"]
inferred = {"orders": {"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}"
said = preview(candidate(base, extra("status <> 'refunded'", base), "a"), SCHEMA, inferred)
assert said["b"]["write"] is True and said["b"]["value"] == "is_test = 0 AND status <> 'refunded'"
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"),
candidate(BOTH_WRONG, RIGHT, "b", source=name)], run_id="run-1",
node_title="复核", actor=None)
assert out["written"] == []
assert "数据源「no_such_source」已不存在" in out["skipped"][0]["reason"]
assert "不止差一个方面" in out["skipped"][1]["reason"]
File diff suppressed because it is too large Load Diff
+3 -2
View File
@@ -86,8 +86,9 @@ async def test_upstream_gap_degrades_issuance_and_names_the_node():
async def test_stub_gate_nodes_pass_through_without_touching_issuance():
"""双盲复算、反方审稿目前是占位:放在图里照样能运行,出具不受影响"""
for kind in ("dual_check", "challenge"):
"""反方审稿目前是占位:放在图里照样能运行,出具不受影响。双盲复算已经实现,没有可复算的指标时直接放行、
不碰出具,见 test_dual_check_node.py"""
for kind in ("challenge",):
row = await finish(graph(node("gate", kind)))
issuance = row.output["_issuance"]
assert issuance["tier"] == "formal", (kind, issuance)