双盲复算节点:按输入盲算、代回原式比对,不一致交人裁决,只差一个方面时写进数据目录

占位的直通桩换成真实现(engine/nodes/dual_check.py):
- 按输入复算:口径卡上输入来自查询快照的指标,每个这样的输入另起一个看不到原 SQL 的只读 agent(verify_agent),只给它指标名、
  口径说明、这个输入在计算式里是哪一项(「输入1」,不露取数节点和 cell() 的行列)、分组结果那一行的维度列、数据源的数据目录;
  它交一个格子引用,由代码从它自己的快照取值。盲值用原式代入盲算的输入算(engine/recompute.py,取整同口径卡),按相对误差比,
  原值为 0 时按绝对误差。两个指标用到同一格只盲算一次。合并查询、整段结果、遮罩列、文本输入不能单独复算,记「无法复算」、不花钱。
- 一致不挂起,产出 agreed;不一致时逐个对不上的输入用 explain_sql_diff 说清差在哪(规范化后一致却值不同的写「差在数据」),
  完整对照落成 card_artifact 工件(避开审批 payload 的截断),按约定的 payload 用 ask() 挂起。审核人选 a 无缺口,选 b、都不对
  写 issuance_gaps,on_reject 为 fail 时节点失败;明确驳回按人工驳回失败;没选的按「都不对」。触顶、核查模型不可用、交不出数的
  指标记「无法复算」并写缺口,核查模型本身用不了时后面的盲算不再一个个去试。
- 设置读发起时的快照,没快照的读一次记进 checkpoint;读设置、盲算连同比对、差异说明连同工件、写目录各在一个 @task 里,审批恢复
  重放时不再调用模型、不再计费、不再写,事件也只发一次。日志按约定的 code 和结构化字段发(dual_check_skipped / same_model /
  agreed / disagreed / resolved / catalog / limit)。盲算的查询只发 tool.end(role: verify),台账只记一条 kind: dual_check,
  报告的写作目录不含盲算的查询;用量计入节点。

选择写进数据目录(engine/dual_check_facts.py):只在审核人选 a 或 b、差异只有一条具体条目、又对得上目录事实时写——时间列不同记
业务日期,一边多一个只涉及一张表的筛选条件记有效记录条件,聚合只差去不去重记去重规则,都按被采纳那一边的写法。照对话提案的
规矩核对,记为人工填写、已确认,备注写明运行、节点、指标;已是同样的已确认说法不再写,确认过的不同说法不覆盖、照实交回原因。

测试:虚构零售库埋了退款行、下单日对支付日两个陷阱,盲算模型是 MockChatModel 剧本,覆盖一致、不一致挂起与 payload / 工件形状、
三种选择与 on_reject、写目录与幂等、不覆盖、全局关闭与节点覆盖、发起时快照、重放不再执行盲算、写作目录不含盲算查询、用量、
同模型提示、触顶、不能单独复算的输入。占位节点的旧测试只留反方审稿。
This commit is contained in:
ThearchyHelios
2026-10-05 14:23:43 +08:00
parent ff23c6a8ad
commit 77f3119cf9
7 changed files with 2361 additions and 11 deletions
+308
View File
@@ -0,0 +1,308 @@
"""双盲复算的选择写进数据目录:审核人在不一致的指标上选了原流程或盲算,而两边只差一个方面、又正好对得上数据目录
里的一项事实时,把被采纳那一边的写法记成人工确认过的目录项。
对得上的事实只有三种(data/catalog.py 的表级项):
- 两边按不同的时间列筛时间 → 业务日期(business_date):被采纳的一边按哪一列筛,业务日期就记哪一列
- 一边多一个只涉及底层表列的筛选条件 → 有效记录条件(valid_filter):被采纳的一边有这个条件才记,记的就是这个条件
- 聚合只差去不去重 → 去重规则(dedup):按被采纳那一边去不去重写一句
**为什么要这么严。** 目录是给助手、db_schema 工具和 SQL 检查看的数据事实,人工确认过的项还会让 SQL 检查报错误级。
只差一个方面时,人选的那一边就是对这一处的判断;差了好几处、或者差异说明里有兜底(未能归纳的差异),人选的是
「整体上哪边对」,拆不出是哪一处对,写进去就是替人编了一条事实。两条 SQL 规范化后一致却算出不同的值(两次取数之间
数据变了、结果被截断),差在数据不在写法,更没有可记的事实。
**怎么写。** 照对话提案的规矩核对(plan_patch:列在不在表结构里、取值格式),记为人工填写、已确认,备注写明来自哪次
运行、哪个节点、哪个指标。目录里这一项已经是人工确认过的另一种说法时不覆盖,照实交回原因;已经是同样的说法就什么都
不写(再写一遍也不升版本)。整个写入由节点放在 ask() 之后的一个 @task 里:审批恢复重放、接着运行时不会再写一遍。
catalog.save_patch 不收备注,这里用它的同一套原料(read_entry、apply_patch、write_entry 的乐观锁)写,只多把备注放进项里。
"""
from __future__ import annotations
import logging
from collections.abc import Mapping
from dataclasses import dataclass
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
@dataclass(frozen=True)
class Fact:
"""要写进目录的一项:哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明。"""
source: str
table: str
item: str
value: Any
metric: str
choice: str
def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None) -> tuple[Fact | None, str | None]:
"""一个候选(审核人在某个指标上选了 a 或 b)→ (要写的目录项, None) 或 (None, 为什么不写)。
candidate:{metric 指标名, choice "a" | "b", source 数据源名, dialect, sql: {a, b}, diff: {equivalent, fallback,
items: [FacetDiff.as_dict()]}}。只看差异说明,不重新比 SQL:审核人看到的就是这份说明。
"""
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.get("dialect"))
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}」不在数据源的表结构里,不写入数据目录"
return Fact(source=str(candidate.get("source") or ""), table=resolved, item=key, value=value, metric=name,
choice=str(side)), None
def _business_date(item: Mapping[str, Any], side: str) -> tuple[tuple[str, str, Any] | None, str | None]:
detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {}
mine = [c for c in detail.get(side) or [] if isinstance(c, Mapping) and c.get("table") and c.get("column")]
if not mine:
return None, f"被采纳的{SIDE_LABEL[side]}没有按时间筛选,没有可记的业务日期"
if len(mine) > 1:
return None, f"被采纳的{SIDE_LABEL[side]}按好几列筛时间,无法确定业务日期按哪一列"
return (str(mine[0]["table"]), "business_date", {"column": str(mine[0]["column"])}), None
def _valid_filter(item: Mapping[str, Any], side: str, dialect: Any) -> tuple[tuple[str, str, Any] | None, str | None]:
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, "差异说明里没有被采纳一边的筛选条件"
detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {}
tables = {str(c.get("table")) for c in detail.get("columns") or [] if isinstance(c, Mapping) and c.get("table")}
if len(tables) != 1:
return None, "筛选条件不是只涉及一张表的列"
(table,) = tables
return (table, "valid_filter", _unqualified(condition.strip(), dialect)), None
def _unqualified(condition: str, dialect: Any) -> str:
"""条件里的列去掉表名:有效记录条件记在这张表自己名下,写成 status <> 'refunded' 而不是 orders.status <> …。
解析不了就原样。"""
import sqlglot
from sqlglot import exp
from app.data.sqlcheck import dialect_of
read = dialect_of(dialect) if isinstance(dialect, str) else None
try:
tree = sqlglot.parse_one(condition, read=read)
except Exception: # noqa: BLE001 - 写法特殊、解析不了:差异说明里给人看的就是这一句,原样记
return condition
if not any(col.table for col in tree.find_all(exp.Column)):
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 _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]
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]
verb = _VERB.get(func, func)
text = f"对 {column} {verb}时去重" if adopted else f"对 {column} {verb}时不去重"
return (str(table), "dedup", text), None
def _aggregate_pair(a: Any, b: Any, dialect: Any) -> tuple[str, tuple[bool, bool]] | None:
"""两边的聚合写法 → (函数名, (A 去重吗, B 去重吗));不是「同一个函数、同一个参数、只差 DISTINCT」返回 None。"""
import sqlglot
from sqlglot import exp
from app.data.sqlcheck import dialect_of
read = dialect_of(dialect) if isinstance(dialect, str) else None
def shape(text: Any) -> tuple[str, str, bool] | None:
if not isinstance(text, str) or not text.strip():
return None
try:
node = sqlglot.parse_one(text, read=read)
except Exception: # noqa: BLE001
return None
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:
"""目录项上的备注:来自哪次运行、哪个节点、哪个指标、采纳了哪一边。"""
return (f"双盲复算(运行 {run_id},节点「{node_title}」):审核人在指标「{fact.metric}」上"
f"采用了{SIDE_LABEL.get(fact.choice, fact.choice)}的写法")
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)
async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, node_title: str,
actor: str | None) -> dict[str, list[dict[str, Any]]]:
"""把审核人的选择写进数据目录:{written: [{source, table, item}], skipped: [{reason}]}。永远不抛:数据源没了、
目录写不进去,都落在 skipped 里照实说。"""
written: list[dict[str, Any]] = []
skipped: list[dict[str, Any]] = []
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
fact, why = derive(candidate, row.schema_cache)
if fact is None:
skipped.append({"reason": why})
continue
done, why = await _write(session, row, fact, note=note_of(fact, run_id=run_id, node_title=node_title),
actor=actor)
except Exception: # noqa: BLE001 - 目录写不进去不能毁掉审批的结果:缺口照样按选择记
logger.exception("双盲复算的选择写入数据目录失败(数据源 %s)", name)
skipped.append({"reason": f"指标「{candidate.get('metric')}」的选择写入数据目录时出错,未写入"})
continue
if done:
written.append({"source": fact.source, "table": fact.table, "item": fact.item})
else:
skipped.append({"reason": why})
return {"written": written, "skipped": skipped}
async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: str | None) -> tuple[bool, str | None]:
"""写一项。(写了没有, 没写的原因)。"""
label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None
for _ in range(_RETRIES):
entry = await catalog.read_entry(session, row.id, fact.table)
notes = entry.notes if entry else {}
version = entry.version if entry else 0
live = notes.get(fact.item) if isinstance(notes.get(fact.item), dict) else None
live = live if live is not None and live.get("status") != "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"]
changes = [{"path": fact.item, "value": value, "reason": note}]
plan = catalog.plan_patch(notes, changes, table=fact.table, tables=tables)
if plan.problems:
return False, f"「{fact.table}」的{label}未写入数据目录:{plan.problems[0]}"
change = plan.changes[0]
if change.state == "same":
return False, (f"「{fact.table}」的{label}已是人工确认过的「{_value_text(fact.item, value)}」,"
"无需重复写入")
if change.before_status == "confirmed":
return False, (f"「{fact.table}」的{label}已有人工确认过的说法「{_value_text(fact.item, change.before)}」,"
f"与这次指标「{fact.metric}」采纳的「{_value_text(fact.item, value)}」不同,未覆盖。"
"如需更改,请到数据目录中修改")
at = catalog.now_iso()
try:
merged = catalog.apply_patch(notes, changes, table=fact.table, tables=tables, at=at)
except catalog.CatalogInvalid as e:
return False, f"「{fact.table}」的{label}未写入数据目录:{e}"
# 不论原来是谁提的:这一项现在出自人的选择,记为人工填写、已确认,备注写明出处
merged[fact.item] = catalog.make_item(change.after, "human", "confirmed", note=note, at=at)
try:
await catalog.write_entry(session, row.id, fact.table, merged, if_version=version, actor=actor)
except catalog.CatalogConflict:
await session.rollback()
continue
except catalog.CatalogInvalid as e:
return False, f"「{fact.table}」的{label}未写入数据目录:{e}"
return True, None
return False, f"「{fact.table}」的数据目录正被修改,{label}未能写入,请稍后在数据目录中确认"
__all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "record_choices"]
+976 -9
View File
@@ -1,21 +1,988 @@
"""双盲复算节点:另起一路、不看原结果独立算一遍,再和原结果逐项比对。
"""双盲复算节点:正式出具前,口径卡上输入来自查询的指标,另起一个看不到原 SQL 的只读 agent 独立算一遍再比对。
目前只是占位:图里放了这个节点也能运行,执行器原样跳过、记一条日志,不改任何状态。
真正的复算和比对由后续工作包补上;比不上的项通过节点产出里的 issuance_gaps 交给成果节点的
出具契约降档(见 nodes/io.py 的 _upstream_issuance_gaps)。
**按输入复算。** 指标由口径卡的原式算出,原式本身确定、可复算(metrics 的 recompute_ok);会错的是喂给它的那几个数:
查哪张表、按哪天归属、排没排除退款……所以盲算的单位是「一个来自查询快照的输入」。盲算 agent(engine/verify_agent.py)
只看得到指标名、口径说明(节点配置的 definitions)、这个输入在计算式里是哪一项、数据源的数据目录,看不到原 SQL 和
原结果;它自己写只读 SQL,最后交一个格子引用,由代码从它自己的查询快照里取值。盲值是拿原式代入盲算的输入算出来的
(engine/recompute.py,取整同口径卡),和原值按相对误差比,原值为 0 时按绝对误差。两个指标用到同一格的只盲算一次。
**一致就放行,不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪(data/sql_semantics.explain_sql_diff),
完整对照落成一件工件(审批 payload 经 runner._safe 会截断,长 SQL 放不下),用 replay.ask 挂起。审核人逐个指标选采用
原流程(a)、采用盲算(b)或两边都不对(neither):a 没有缺口;b、neither 写进产出的 issuance_gaps,成果节点的出具
契约据此降档(on_reject 为 fail 时节点直接失败)。两边只差一个方面、又对得上数据目录事实的,选择写进数据目录
(engine/dual_check_facts.py)。盲算触顶、出错、交不出数,或者输入本身没法单独复算的指标按「无法复算」记缺口。
**重放。** 审批恢复时节点从头重放。读设置(发起时没快照的才读)、盲算连同比对、差异说明连同工件、写数据目录各在一个
@task 里,重放时直接取 checkpoint 里的结果:不再调用模型、不再计费、不再写目录,事件也都在 task 里发、不会再发一遍;
task 外面的日志经 replay.once 发。task 的先后只取决于这次运行固定下来的东西(图、运行选项、上游的产出)。设置读发起
时的快照(launch_run 写进 run.options["verify"]):审批停下期间改了设置,恢复后照旧按发起时的。
**不进写作目录。** 盲算的查询发 tool.end(role: verify),随运行封存、证据面板能下钻;台账只记一条 kind: dual_check,
绝不写 kind: query——报告的写作目录按台账里的 query 编号,盲算的查询混进去,写作者就会引用它们。
"""
from __future__ import annotations
import ast
import asyncio
import copy
import math
import re
from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass
from typing import Any
from langgraph.func import task
from sqlalchemy import select
from app.core import artifact_store
from app.core.errors import describe_exception
from app.core.events import EventType
from app.engine.context import NodeContext
from app.db.base import SessionLocal
from app.db.models import DataSource
from app.engine import recompute
from app.engine.approval import read_decision
from app.engine.context import NodeContext, NodeError
from app.engine.evidence import build_catalog, is_number, next_exec
from app.engine.expressions import ExpressionError, _cell_call, _chain_root, cell_value, column_kind, numeric_text, \
parse_expression
from app.engine.replay import _in_graph, ask, once
from app.engine.schema import DUAL_CHECK_FIELD_LABEL, DUAL_CHECK_ON_REJECT, NodeType, dual_check_card
from app.engine.state import GraphState
#: 占位期间的跳过原因:节点产出和日志里都写这一句
PENDING_REASON = "双盲复算功能开发中,本节点暂未生效,已跳过"
#: 节点产出的状态:跳过、无需裁决、人已裁决
SKIPPED, AGREED, RESOLVED = "skipped", "agreed", "resolved"
#: 单个指标的结果:一致、不一致、无法复算
SAME, DIFFERENT, UNCHECKED = "agreed", "disagreed", "unchecked"
#: 审核人的三种选择:采用原流程、采用盲算、两边都不对
CHOICES = ("a", "b", "neither")
#: 盲算触顶的原因里交给界面认的几种(dual_check_limit 日志的 limit);max_steps 只由核查本身的 verify_limit 日志说
LIMIT_CODES = ("max_queries", "timeout_s", "max_cost_usd", "daily_max_usd")
#: 输入的出处:口径卡的 cell() 取的格、Agent 经 cite_fields 交来的字段,都落在一份查询快照上
QUERY_VIAS = ("tool_cell", "agent_field")
#: 差异说明里两边的叫法,先原流程后盲算:dual_check_facts 按 a / b 认这个顺序
SIDES = ("原流程", "盲算")
#: 审批卡摘要里每个指标最多列几条差异
SUMMARY_DIFFS = 3
#: 一次 SQL 比较最多等多久:解析是纯计算,放在线程里,超时就给兜底说明
DIFF_TIMEOUT_S = 15.0
#: 交给盲算 agent 的数据目录最多这么多字(表清单、各表的说明分开算),再多就请它自己用 db_schema 查
_INDEX_CHARS = 3_000
_NOTES_CHARS = 6_000
#: 交给盲算 agent 的同一行其他列最多几列、每个值最多几个字
_DIMS = 6
_DIM_CHARS = 40
#: 卡上给容器输入记的大概(metrics._brief):「[12 项]」「{3 个字段}」,写进计算式时不当成字面量
_BRIEF = re.compile(r"^(?:\[\d+ 项\]|\{\d+ 个字段\})$")
SKIP_DISABLED = "双盲复算已在设置中关闭,本节点直接放行"
SKIP_NODE_OFF = "本节点已关闭双盲复算,直接放行"
SKIP_NO_INPUTS = "口径卡上没有输入来自查询的指标,无需双盲复算,直接放行"
#: 两条 SQL 规范化后一致、算出的值却不同:差在数据(两次取数之间数据变了、结果被截断),不在写法
SAME_SQL = "两条 SQL 规范化后一致,差异可能来自两次取数之间数据有变化或取数结果被截断"
SAME_SQL_ASPECT = "data"
DIFF_TIMEOUT = "两条 SQL 比较超时,未能归纳成具体差异,请对照原文"
DIFF_FAILED = "两条 SQL 比较时出错,未能归纳成具体差异,请对照原文"
NO_SQL = "有一边没有可比较的 SQL,请对照原文"
#: 交给成果节点的缺口(不带节点名:出具契约统一加「双盲复算「节点名」:」)
GAP_BLIND = "指标「{name}」双盲复算不一致,审核人选择了盲算结果"
GAP_NEITHER = "指标「{name}」双盲复算不一致,审核人判断两边都不对"
GAP_UNCHECKED = "指标「{name}」无法复算:{reason}"
BLIND_SYSTEM = (
"你在做双盲复算:另一条流程已经算出了口径卡上的一个指标,你要在看不到它的 SQL 和结果的情况下,按业务定义独立算出"
"计算式里的一个数,交给人和原流程比对。\n"
"- 业务含义只按下面给的指标名称、口径说明和数据目录理解。数据目录里人工确认过的说明(业务日期按哪一列、有效记录"
"条件、去重规则)要遵守,标「推断,未确认」的谨慎使用\n"
"- 时间范围、要排除哪些行、要不要去重,都按口径说明来;口径说明没写到的,按数据目录和最常见的业务理解,并在理由里"
"写明你的取舍\n"
"- 尽量用一条 SQL 直接算出要交的那个数(聚合放在 SQL 里做),最后交出它所在的格子"
)
BLIND_TASK_PROMPT = (
"指标:{name}{unit}\n"
"口径说明:{definition}\n"
"计算式:{formula}\n"
"你要算的是计算式里的 {label}{whole}:原流程里它是一次查询结果中「{column}」列的值{dims}。只交这一个数。\n"
"数据源:{source}\n\n"
"{catalog}"
)
BLIND_UNIT_PROMPT = "(单位:{unit})"
BLIND_NO_DEFINITION_PROMPT = "(未提供,按指标名称和数据目录理解)"
BLIND_WHOLE_PROMPT = "(这个指标就是它本身)"
BLIND_DIMS_PROMPT = ",所在行的其他列:{dims}"
BLIND_INDEX_PROMPT = "数据源里的表(表名|中文名|粒度):\n{index}"
BLIND_NOTES_PROMPT = "数据目录(标「推断,未确认」的项是推断,使用前请核实):\n{notes}"
BLIND_NO_CATALOG_PROMPT = "这个数据源还没有数据目录,请用 db_schema 工具查看表结构。"
BLIND_MORE_PROMPT = "\n…(其余的请用 db_schema 工具查看)"
#: 盲算交回的结构:一个数(交给模型时换成格子引用,由代码取值),verify_agent 另加理由
BLIND_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {"value": {"type": "number", "description": "要你算的那个数"}},
"required": ["value"],
}
@dataclass(frozen=True)
class _Options:
"""这次复算按什么来:节点配置优先,没写的取发起时的核查设置快照。"""
tolerance: float
on_reject: str
metrics: list[str] | None
definitions: dict[str, str]
max_queries: int | None
timeout_s: float | None
daily_max_usd: float | None
async def run_dual_check(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
ctx.emit(EventType.LOG, level="info", code="node_pending", message=PENDING_REASON)
return {"nodes": {ctx.node.id: {"skipped": True, "reason": PENDING_REASON}}}
settings = await _settings(ctx)
part = settings.get("dual_check") if isinstance(settings.get("dual_check"), dict) else {}
switch = ctx.config.get("enabled")
if switch is False or (switch is not True and not part.get("enabled", True)):
await once(ctx, EventType.LOG, level="info", code="dual_check_skipped", reason="disabled",
message=SKIP_NODE_OFF if switch is False else SKIP_DISABLED)
return _skipped(ctx, "disabled")
card_id, card = _card(state, ctx)
options = _options(ctx, part, settings)
targets = _targets(card, options, ctx) if card is not None else []
if not targets:
await once(ctx, EventType.LOG, level="info", code="dual_check_skipped", reason="no_inputs",
message=SKIP_NO_INPUTS)
return _skipped(ctx, "no_inputs")
checked = await _step("dual_check_blind", lambda: _check(state, ctx, targets, options, settings))
rows = checked["rows"]
disagreed = [r for r in rows if r["status"] == DIFFERENT]
if not disagreed:
return _finish(state, ctx, card_id, card, checked, status=AGREED, gaps=_unchecked_gaps(rows))
review = await _step("dual_check_review", lambda: _review(ctx, checked))
response = await ask(ctx, review["payload"], **review["payload"])
# 谁裁决的跟着事件走,和人工审批节点一样
await once(ctx, EventType.HUMAN_RESOLVED, response=response, actor=ctx.actor())
choices, note = _choices(ctx, response, disagreed)
gaps = _gaps(rows, choices)
await once(ctx, EventType.LOG, level="info", code="dual_check_resolved", choices=choices, gaps=gaps,
message=_resolved_text(disagreed, choices))
candidates = [{**review["facts"][r["metric_id"]], "choice": choices[r["metric_id"]]}
for r in disagreed if choices[r["metric_id"]] in ("a", "b") and r["metric_id"] in review["facts"]]
catalog = await _step("dual_check_catalog", lambda: _catalog(ctx, candidates)) if candidates else None
rejected = [r for r in disagreed if choices[r["metric_id"]] != "a"]
if rejected and options.on_reject == "fail":
# 目录照样写了:审核人的选择是对数据的判断,和这次运行停不停无关
names = "、".join(f"「{r['name']}」" for r in rejected[:5])
raise NodeError(ctx.node.id, f"双盲复算不一致,审核人没有采用原流程的结果(指标{names})。"
f"「{DUAL_CHECK_FIELD_LABEL['on_reject']}」设为「判为失败」,运行已停止;"
"请按审核意见修正取数或口径后重新运行")
return _finish(state, ctx, card_id, card, checked, status=RESOLVED, gaps=gaps, choices=choices, note=note,
artifact=review["card_artifact"], catalog=catalog)
# --------------------------------------------------------------------------
# 设置、口径卡、要复算的指标
# --------------------------------------------------------------------------
async def _step(name: str, work: Callable[[], Awaitable[Any]]) -> Any:
"""在一个 @task 里做 work:审批恢复、接着运行时节点从头重放,结果直接取 checkpoint 里的,不再做一遍;work 里发的
事件也就只发一次。不在图里执行(单独调用执行器)时直接做。"""
if not _in_graph():
return await work()
@task(name=name)
async def step() -> Any:
return await work()
return await step()
async def _settings(ctx: NodeContext) -> dict[str, Any]:
"""这次运行的核查设置:发起时的快照(run.options["verify"])。发起时没快照的(不经 launch_run 发起、子工作流里的
节点)读当前设置,读一次记进 checkpoint:重放时不再读,审批停下期间改了设置也不会让重放走上另一条路。"""
from app.api.settings import VERIFY_GROUP, verify_settings, verify_values
from app.engine.runner import RUN_OPTIONS_KEY
options = ctx.run.extra.get(RUN_OPTIONS_KEY)
snapshot = options.get(VERIFY_GROUP) if isinstance(options, dict) else None
if isinstance(snapshot, dict):
return verify_values(snapshot)
async def read() -> dict[str, Any]:
async with SessionLocal() as session:
return await verify_settings(session)
return await _step("dual_check_settings", read)
def _card(state: GraphState, ctx: NodeContext) -> tuple[str, dict[str, Any] | None]:
"""要复算的口径卡:(节点 id, 它这次的产出)。配置不对(发起前的校验已经拦下,这里兜底)报节点错误;卡没执行到
(在没走到的分支上)产出为 None。"""
card_id, problem = dual_check_card(ctx.run.spec, ctx.node)
if problem or not card_id:
raise NodeError(ctx.node.id, problem or f"「{DUAL_CHECK_FIELD_LABEL['card']}」需要选择一个口径卡节点")
node = ctx.run.spec.node_map().get(card_id)
if node is None or node.type != NodeType.METRICS:
raise NodeError(ctx.node.id, f"「{DUAL_CHECK_FIELD_LABEL['card']}」选择的「{card_id}」不是口径卡节点,"
"请重新选择")
output = (state.get("nodes") or {}).get(card_id)
return card_id, output if isinstance(output, dict) and output.get("kind") == "metric_set" else None
def _options(ctx: NodeContext, part: Mapping[str, Any], settings: Mapping[str, Any]) -> _Options:
cfg = ctx.config
tolerance = cfg.get("tolerance")
if isinstance(tolerance, bool) or not isinstance(tolerance, (int, float)) or not math.isfinite(tolerance) \
or tolerance < 0:
tolerance = part.get("tolerance")
if isinstance(tolerance, bool) or not isinstance(tolerance, (int, float)) or tolerance < 0:
tolerance = 0.0
wanted = cfg.get("metrics")
definitions = cfg.get("definitions") if isinstance(cfg.get("definitions"), dict) else {}
return _Options(
tolerance=float(tolerance),
on_reject=cfg.get("on_reject") if cfg.get("on_reject") in DUAL_CHECK_ON_REJECT else "degrade",
metrics=[str(m) for m in wanted if isinstance(m, str) and m.strip()] if isinstance(wanted, list) else None,
definitions={str(k): str(v) for k, v in definitions.items() if isinstance(v, str) and v.strip()},
max_queries=part.get("max_queries"), timeout_s=part.get("timeout_s"),
daily_max_usd=settings.get("daily_max_usd"),
)
def _from_query(item: Any) -> bool:
return isinstance(item, dict) and item.get("via") in QUERY_VIAS and isinstance(item.get("artifact"), str) \
and bool(item["artifact"]) and isinstance(item.get("path"), str)
def _number(value: Any) -> bool:
if not is_number(value):
return False
try:
return math.isfinite(value)
except OverflowError:
return False
def _targets(card: dict[str, Any], options: _Options, ctx: NodeContext) -> list[dict[str, Any]]:
"""要复算的指标,按配置的先后(没配就按卡上的先后):原值是数、至少一个输入来自查询快照。配置里写了卡上没有的
指标 ID 报节点错误(口径卡引用自别的工作流时,画图时查不出来)。"""
metrics = [m for m in card.get("metrics") or [] if isinstance(m, dict) and m.get("id") is not None]
by_id = {str(m["id"]): m for m in metrics}
if options.metrics:
if unknown := [m for m in options.metrics if m not in by_id]:
raise NodeError(ctx.node.id, f"「{DUAL_CHECK_FIELD_LABEL['metrics']}」中的「{'、'.join(unknown[:5])}」"
f"不是口径卡上的指标 ID,可选的指标 ID:{'、'.join(list(by_id)[:12])}")
chosen = [by_id[m] for m in dict.fromkeys(options.metrics)]
else:
chosen = metrics
return [m for m in chosen if _number(m.get("value")) and any(_from_query(i) for i in m.get("inputs") or [])]
def _skipped(ctx: NodeContext, reason: str) -> dict[str, Any]:
return {"nodes": {ctx.node.id: {"status": SKIPPED, "reason": reason}}}
# --------------------------------------------------------------------------
# 盲算与比对(一个 @task)
# --------------------------------------------------------------------------
class _Sources:
"""一次复算里要读的查询快照和数据源:同一份只读一次。"""
def __init__(self) -> None:
self.snapshots: dict[str, tuple[Any, str | None]] = {}
self.rows: dict[str, DataSource | None] = {}
def snapshot(self, artifact: str) -> tuple[dict[str, Any] | None, str | None]:
if artifact not in self.snapshots:
try:
content = artifact_store.load(artifact)
self.snapshots[artifact] = (content, None if content is not None else "查询快照无法读取")
except ValueError:
self.snapshots[artifact] = (None, "查询快照与哈希不一致,疑似被修改")
content, why = self.snapshots[artifact]
if why is None and not (isinstance(content, dict) and isinstance(content.get("rows"), list)):
return None, "查询快照无法读取"
return content, why
async def source(self, name: str) -> DataSource | None:
if name not in self.rows:
async with SessionLocal() as session:
self.rows[name] = (await session.execute(
select(DataSource).where(DataSource.name == name))).scalar_one_or_none()
return self.rows[name]
async def hidden(self, artifact: str) -> frozenset[str]:
"""这份快照要遮的列(小写):查询当时记下的,加上数据源现在设的。宁可多遮。"""
from app.data.engine import masked_columns
snap, _ = self.snapshot(artifact)
recorded = (snap or {}).get("mask_columns")
out = {c.lower() for c in recorded if isinstance(c, str)} if isinstance(recorded, list) else set()
name = (snap or {}).get("source")
if isinstance(name, str) and name:
row = await self.source(name)
out |= {c.lower() for c in masked_columns(row.options if row is not None else None)}
return frozenset(out)
def _loader(state: GraphState) -> tuple[Callable[[str], Any], dict[str, str]]:
"""重算要的取工件函数和节点产出:nodes.* 的容器输入从这次运行的状态里取(口径卡执行时上游的产出就是这些)。"""
prefix = "state:"
outputs = {str(nid): f"{prefix}{nid}" for nid in state.get("nodes") or {}}
def load(key: str) -> Any:
if key.startswith(prefix):
return (state.get("nodes") or {}).get(key[len(prefix):])
return artifact_store.load(key)
return load, outputs
async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, Any]], options: _Options,
settings: Mapping[str, Any]) -> dict[str, Any]:
"""逐个指标认出能复算的输入、按输入盲算、代回原式比对。返回 {rows, results, model, same_model, usage}。
rows 每个指标一项:{metric_id, name, unit, decimals, status, reason?, original: {value, rendered}, blind: {value,
rendered}, delta_pct, inputs: [{path, key, value, sql, source, alias, artifact, row, column, node_id}], differing};
results 是每个盲算(按格子去重)的结果,见 _slim。
"""
sources = _Sources()
load, outputs = _loader(state)
aliases = _original_aliases(state)
rows: list[dict[str, Any]] = []
jobs: dict[str, dict[str, Any]] = {}
for metric in targets:
row, metric_jobs = await _plan_metric(state, metric, options, sources, aliases, load, outputs)
rows.append(row)
for key, job in metric_jobs.items():
# 两个指标用到同一格只盲算一次,按先出现的那个指标说清要算什么;先出现的没写口径说明、后一个写了,用后一个
if key not in jobs or (not jobs[key]["definition"] and job["definition"]):
jobs[key] = job
results, model, same, usage = await _blind(ctx, jobs, options, settings, _producer_models(state, rows))
for row, metric in zip(rows, targets):
if row["status"] != UNCHECKED:
_compare(row, metric, results, options, load, outputs, row.pop("_masks"))
else:
row.pop("_masks", None)
agreed = sum(1 for r in rows if r["status"] == SAME)
if agreed and not any(r["status"] == DIFFERENT for r in rows):
ctx.emit(EventType.LOG, level="info", code="dual_check_agreed", metrics=agreed,
message=f"双盲复算一致:{agreed} 个指标两边的结果相同,已放行")
return {"rows": rows, "results": results, "model": model, "same_model": same, "usage": usage}
def _original_aliases(state: GraphState) -> dict[str, str]:
"""原流程的查询快照在这次运行的证据台账里的编号(Q1、Q2…,和报告取全部上游查询时的编号一致)。"""
ledger = [e for e in state.get("evidence") or [] if isinstance(e, dict)]
catalog = build_catalog(nodes={}, ledger=ledger)
return {str(entry.get("artifact")): alias for alias, entry in catalog.items()
if entry.get("kind") == "query" and entry.get("artifact")}
async def _plan_metric(state: GraphState, metric: dict[str, Any], options: _Options, sources: _Sources,
aliases: dict[str, str], load: Callable[[str], Any], outputs: dict[str, str]
) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]:
"""一个指标:认出来自查询的输入能不能单独复算,能的话给每个输入建一个盲算任务。"""
row: dict[str, Any] = {
"metric_id": str(metric["id"]), "name": str(metric.get("name") or metric["id"]),
"unit": str(metric.get("unit") or ""), "decimals": metric.get("decimals"),
"status": "pending", "original": {"value": metric.get("value"), "rendered": metric.get("rendered")},
"blind": {"value": None, "rendered": None}, "delta_pct": None, "inputs": [], "differing": [],
}
tops = _tops(metric)
masks: dict[str, frozenset[str]] = {}
problem: str | None = None
for item in metric.get("inputs") or []:
if not _from_query(item):
continue
info, why = await _query_input(item, sources, aliases, tops)
masks[item["artifact"]] = await sources.hidden(item["artifact"])
if why:
problem = problem or why
continue
row["inputs"].append(info)
row["_masks"] = masks
if problem is None:
# 什么都不改先重算一遍:和卡上记的对不上(输入还原不了、用到了遮罩列),换了盲算的输入也算不准
check = recompute.recompute(metric, recompute.Changes(), recompute.Sources(
load=load, outputs=outputs, masked=masks, queries=tuple(dict.fromkeys(masks))))
if check["status"] != "ok" or check["changed"]:
problem = f"无法按原表达式重算这个指标({check.get('reason') or '重算结果与口径卡不一致'})"
if problem:
row.update(status=UNCHECKED, reason=problem)
return row, {}
labels = {inp["path"]: f"输入{i}" for i, inp in enumerate(row["inputs"], 1)}
formula = _formula(metric, labels)
jobs: dict[str, dict[str, Any]] = {}
for inp in row["inputs"]:
snap, _ = sources.snapshot(inp["artifact"])
jobs[inp["key"]] = {
"key": inp["key"], "metric_id": row["metric_id"], "metric": row["name"], "unit": row["unit"],
"definition": options.definitions.get(row["metric_id"]) or "", "formula": formula,
"label": labels[inp["path"]], "column": inp["column"], "source": inp["source"],
"dims": _dims(snap or {}, inp["row"], inp["column"], masks.get(inp["artifact"]) or frozenset()),
}
return row, jobs
async def _query_input(item: dict[str, Any], sources: _Sources, aliases: dict[str, str], tops: list[str]
) -> tuple[dict[str, Any], str | None]:
"""一个来自查询快照的输入:(盲算要用的信息, None) 或 (_, 为什么不能单独复算)。"""
from app.engine.merge_query import MERGE_SOURCE
artifact = item["artifact"]
snap, why = sources.snapshot(artifact)
if why:
return {}, f"输入 {item['path']} 的{why}"
source = snap.get("source")
if source == MERGE_SOURCE:
return {}, f"输入 {item['path']} 来自合并查询(跨数据源合并的结果),盲算只能在单个数据源里查询"
if not isinstance(source, str) or not source:
return {}, f"输入 {item['path']} 的查询快照没有记下数据源"
locator = item.get("locator") if isinstance(item.get("locator"), dict) else {}
row, column = locator.get("row"), locator.get("column")
if isinstance(row, bool) or not isinstance(row, int) or row < 0 or not isinstance(column, str):
return {}, f"输入 {item['path']} 取的是一整段查询结果,盲算只能复核单个格子"
if item["path"] not in tops:
return {}, f"输入 {item['path']} 决定取哪一格,不能单独复算"
value = cell_value(item.get("value"))
if not _number(value):
return {}, f"输入 {item['path']} 不是数值"
if column.lower() in await sources.hidden(artifact):
return {}, f"输入 {item['path']} 取自设置了遮罩的列,不能用于复算"
found = await sources.source(source)
if found is None or not found.enabled:
return {}, f"数据源「{source}」不存在或已停用"
return {"path": item["path"], "key": f"{artifact}|{row}|{column}", "value": value,
"sql": str(snap.get("sql") or ""), "source": source, "alias": aliases.get(artifact),
"artifact": artifact, "row": row, "column": column, "node_id": item.get("node_id"),
"via": item["via"]}, None
def _tops(metric: dict[str, Any]) -> list[str]:
"""表达式最外层的引用:只有它们的值直接参与计算,嵌在下标、cell() 行列里的决定取哪一格。"""
try:
tree, _, _ = parse_expression(str(metric.get("expression") or ""))
except ExpressionError:
return []
tops, _ = recompute._refs(tree)
return tops
class _Swap(ast.NodeTransformer):
"""把最外层的引用换成给人(给盲算 agent)看的写法:取值链、整个 cell() 调用,认法同 recompute._Bind。"""
def __init__(self, shown: Callable[[str], ast.AST | None]) -> None:
self.shown = shown
def visit_Call(self, node: ast.Call) -> ast.AST:
if _cell_call(node) and (swapped := self.shown(ast.unparse(node))) is not None:
return ast.copy_location(swapped, node)
return self.generic_visit(node)
def _chain(self, node: ast.AST) -> ast.AST:
if _chain_root(node) is None:
return self.generic_visit(node)
swapped = self.shown(ast.unparse(node))
return ast.copy_location(swapped, node) if swapped is not None else node
visit_Attribute = visit_Subscript = visit_Name = _chain
def _formula(metric: dict[str, Any], labels: dict[str, str]) -> str:
"""交给盲算 agent 的计算式:来自查询的输入写成「输入1」「输入2」,别的输入写当时代进去的值(容器写「其他输入」)。
不露出取数节点的 id、cell() 的行列:那是原流程的取法,不是业务定义。"""
expr = str(metric.get("expression") or "")
try:
tree, _, _ = parse_expression(expr)
except ExpressionError:
return "、".join(labels.values())
values = {i.get("path"): i.get("value") for i in metric.get("inputs") or [] if isinstance(i, dict)}
def shown(path: str) -> ast.AST | None:
if path in labels:
return ast.Name(id=labels[path], ctx=ast.Load())
value = values.get(path)
if value is None or isinstance(value, (bool, int, float)) or (
isinstance(value, str) and len(value) <= _DIM_CHARS and not _BRIEF.match(value)):
return ast.Constant(value=value)
return ast.Name(id="其他输入", ctx=ast.Load())
swapped = _Swap(shown).visit(copy.deepcopy(tree))
return ast.unparse(ast.fix_missing_locations(swapped))
def _dims(snapshot: dict[str, Any], row: int, column: str, hidden: frozenset[str]) -> list[str]:
"""原流程那一格所在行的其他列里不是数的那些(地区、日期这类维度):结果分了组时,盲算 agent 要知道算的是哪一组。
数不给(同一行别的指标值会带偏它),遮罩列不给。"""
rows = snapshot.get("rows") or []
if not 0 <= row < len(rows):
return []
record = rows[row]
columns = [str(c) for c in snapshot.get("columns") or []]
pairs = list(record.items()) if isinstance(record, dict) else list(zip(columns, record))
out: list[str] = []
for name, value in pairs:
name = str(name)
if name == column or name.lower() in hidden or value is None:
continue
kind = column_kind(snapshot, name)
if kind == "number" or (kind is None and (is_number(value) or numeric_text(value) is not None)):
continue
text = str(value).replace("\n", " ")
out.append(f"{name} 为「{text[:_DIM_CHARS]}」")
if len(out) >= _DIMS:
break
return out
async def _catalog_text(sources: _Sources, name: str) -> str:
"""交给盲算 agent 的数据目录:表清单(表名、中文名、粒度)和有目录的表的业务说明(业务日期、有效记录条件、去重、
字段含义、码值、关联关系)。太长的截断,其余请它用 db_schema 查。"""
from app.data import catalog
row = await sources.source(name)
if row is None:
return BLIND_NO_CATALOG_PROMPT
async with SessionLocal() as session:
entries = await catalog.read_catalog(session, row.id)
cache = row.schema_cache if isinstance(row.schema_cache, dict) else {}
tables = cache.get("tables") or {}
notes = {table: entry.notes for table, entry in entries.items()}
parts: list[str] = []
if index := catalog.render_table_index(cache, notes):
parts.append(BLIND_INDEX_PROMPT.format(index=_clip(index, _INDEX_CHARS)))
blocks = []
for table, body in notes.items():
block = catalog.render_table_notes(body, meta=tables.get(table), system_notes=catalog.system_notes_source(row))
# 每张表的说明去掉标题行,整段只在前面说一次「推断,未确认」的意思
lines = block.split("\n", 1)[1] if "\n" in block else ""
if lines:
blocks.append(f"表 {table}:\n{lines}")
if blocks:
parts.append(BLIND_NOTES_PROMPT.format(notes=_clip("\n".join(blocks), _NOTES_CHARS)))
return "\n\n".join(parts) or BLIND_NO_CATALOG_PROMPT
def _clip(text: str, limit: int) -> str:
return text if len(text) <= limit else text[:limit].rsplit("\n", 1)[0] + BLIND_MORE_PROMPT
def _task_text(job: dict[str, Any], catalog: str) -> str:
return BLIND_TASK_PROMPT.format(
name=job["metric"], unit=BLIND_UNIT_PROMPT.format(unit=job["unit"]) if job["unit"] else "",
definition=job["definition"] or BLIND_NO_DEFINITION_PROMPT, formula=job["formula"], label=job["label"],
whole=BLIND_WHOLE_PROMPT if job["formula"].strip() == job["label"] else "", column=job["column"],
dims=BLIND_DIMS_PROMPT.format(dims="、".join(job["dims"])) if job["dims"] else "",
source=job["source"], catalog=catalog)
async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Options, settings: Mapping[str, Any],
producers: set[str]) -> tuple[dict[str, Any], str | None, bool, dict[str, Any]]:
"""按先后逐个盲算:(各格子的结果, 盲算模型, 是否和产出原数据的模型相同, 用量)。producers 是产出原数据的模型。"""
from app.api.settings import verify_model_spec
from app.engine.verify_agent import run_verify_agent
from app.tools.datasource import run_versions
usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "cost_usd": 0.0, "calls": 0}
if not jobs:
return {}, None, False, usage
async with SessionLocal() as session:
spec = await verify_model_spec(session, ctx.config, settings=dict(settings))
sources = _Sources()
catalogs: dict[str, str] = {}
results: dict[str, Any] = {}
model: str | None = None
same = False
broken: dict[str, Any] | None = None
for key, job in jobs.items():
if broken is not None:
# 核查模型本身用不了(没配好、接入出错):后面的盲算一样用不了,照这一次的原因记,不再一个个去试
results[key] = broken
continue
if job["source"] not in catalogs:
catalogs[job["source"]] = await _catalog_text(sources, job["source"])
outcome = await run_verify_agent(
spec=spec, system=BLIND_SYSTEM, task=_task_text(job, catalogs[job["source"]]), sources=[job["source"]],
output_schema=BLIND_SCHEMA, max_queries=options.max_queries, timeout_s=options.timeout_s,
daily_max_usd=options.daily_max_usd, emit=ctx.emit, run_id=ctx.run.run_id, node_id=ctx.node.id,
data_versions=run_versions(ctx.run))
results[key] = _slim(outcome)
if outcome.error and not outcome.model:
broken = results[key]
for k in usage:
usage[k] += outcome.usage.get(k) or 0
if outcome.model and model is None:
model = outcome.model
if model in producers:
same = True
ctx.emit(EventType.LOG, level="warn", code="dual_check_same_model", model=model,
message=f"盲算模型与产出原数据的模型相同(「{model}」),难以发现原流程自身的错误。"
"请在「设置 → 偏好设置 → 核查」中选择另一个核查模型,"
f"或在节点的「{DUAL_CHECK_FIELD_LABEL['model']}」中指定")
if outcome.limit in LIMIT_CODES:
ctx.emit(EventType.LOG, level="warn", code="dual_check_limit", limit=outcome.limit, metric=job["metric_id"],
message=f"指标「{job['metric']}」的盲算{_limit_text(outcome.limit, options)},按无法复算处理")
usage["cost_usd"] = round(usage["cost_usd"], 6)
return results, model, same, usage
def _producer_models(state: GraphState, rows: list[dict[str, Any]]) -> set[str]:
"""产出原数据的模型:要复算的输入是 Agent 经 cite_fields 交来的,那个 Agent 用的模型(节点产出里的 model)。调用
工具节点写死的 SQL 没有模型参与,不算。"""
outputs = state.get("nodes") or {}
out: set[str] = set()
for row in rows:
for inp in row["inputs"]:
found = outputs.get(inp.get("node_id")) if inp.get("via") == "agent_field" else None
if isinstance(found, dict) and isinstance(found.get("model"), str) and found["model"]:
out.add(found["model"])
return out
def _slim(outcome: Any) -> dict[str, Any]:
"""一次盲算的结果里要留进 checkpoint 的部分。"""
cell = next((c for c in outcome.cells if c.get("path") == "value"), None) or {}
query = next((q for q in outcome.queries if q.get("alias") == cell.get("alias")), None) or {}
value = cell.get("value") if cell.get("status") == "resolved" else None
reason = None
if outcome.error:
reason = outcome.error
elif cell.get("status") == "missing":
reason = "盲算没有查到这个数"
elif cell.get("status") == "unresolved":
reason = f"盲算交回的格子引用不可用({cell.get('reason') or '没有交这个数'})"
elif value is None:
reason = "盲算取到的那一格是空值"
elif _cell_masked(cell):
value, reason = None, "盲算取的是设置了遮罩的列,不采用"
return {
"value": value, "reason": reason, "limit": outcome.limit, "model": outcome.model,
"alias": cell.get("alias"), "ref": cell.get("ref"), "artifact": cell.get("artifact"),
"row": cell.get("row"), "column": cell.get("column"), "eid": cell.get("eid"),
"sql": query.get("sql"), "source": query.get("source"),
"rationale": (outcome.result or {}).get("rationale") or "",
"queries": [{k: q.get(k) for k in ("alias", "artifact", "sql", "row_count", "truncated", "source")}
for q in outcome.queries],
"notes": list(outcome.notes),
}
def _cell_masked(cell: dict[str, Any]) -> bool:
"""盲算交的那一格在遮罩列上(查询当时数据源设的遮罩记在它自己的快照里):值会进审批卡和对照工件,不能用。"""
if not cell.get("artifact"):
return False
try:
snap = artifact_store.load(str(cell["artifact"]))
except (ValueError, OSError):
return True
recorded = snap.get("mask_columns") if isinstance(snap, dict) else None
column = str(cell.get("column") or "").lower()
return isinstance(recorded, list) and any(isinstance(c, str) and c.lower() == column for c in recorded)
def _limit_text(limit: str | None, options: _Options) -> str:
"""盲算为什么停下,接在「盲算」后面读:「盲算已达每项最多查询次数(6 次)」。"""
if limit == "max_queries" and options.max_queries is not None:
return f"已达每项最多查询次数({options.max_queries} 次)"
if limit == "timeout_s" and options.timeout_s is not None:
return f"已达每项时长上限({options.timeout_s:g} 秒)"
if limit == "daily_max_usd" and options.daily_max_usd is not None:
return f"已达核查每日金额上限(${float(options.daily_max_usd):g})"
if limit in LIMIT_CODES:
return "已达核查的上限"
return "轮数已用完"
def _agree(original: Any, blind: Any, tolerance: float) -> bool:
"""按相对误差比(原值为 0 时按绝对误差)。"""
if original == blind:
return True
if not (_number(original) and _number(blind)):
return False
if original == 0:
return abs(blind) <= tolerance
return abs(blind - original) <= tolerance * abs(original)
def _delta_pct(original: Any, blind: Any) -> float | None:
"""相差比例,口径同运行对比:(盲算值 − 原流程值) ÷ |原流程值| × 100,两位小数;原流程值为 0 时为 None。"""
if not (_number(original) and _number(blind)) or original == 0:
return None
return round((blind - original) / abs(original) * 100, 2)
def _compare(row: dict[str, Any], metric: dict[str, Any], results: dict[str, Any], options: _Options,
load: Callable[[str], Any], outputs: dict[str, str], masks: dict[str, frozenset[str]]) -> None:
"""盲算的输入代回原式,和原值比。row 就地改成一致 / 不一致 / 无法复算。"""
values: dict[str, Any] = {}
for inp in row["inputs"]:
found = results.get(inp["key"])
if found is None:
row.update(status=UNCHECKED, reason="盲算没有执行")
return
if found.get("limit"):
row.update(status=UNCHECKED, reason="盲算" + _limit_text(found["limit"], options))
return
if found.get("value") is None:
row.update(status=UNCHECKED, reason=found.get("reason") or "盲算没有交回可用的数值")
return
values[inp["path"]] = found["value"]
outcome = recompute.recompute(metric, recompute.Changes(inputs=values), recompute.Sources(
load=load, outputs=outputs, masked=masks, queries=tuple(dict.fromkeys(masks))))
if outcome["status"] != "ok":
row.update(status=UNCHECKED, reason=f"用盲算的输入按原表达式算不出结果({outcome.get('reason') or '无法计算'})")
return
original = row["original"]["value"]
blind = outcome["new"]
row["blind"] = {"value": blind, "rendered": outcome.get("rendered_new")}
row["delta_pct"] = _delta_pct(original, blind)
if _agree(original, blind, options.tolerance):
row["status"] = SAME
return
row["status"] = DIFFERENT
# 对不上的输入:超出误差的;都没超出(误差经算式放大了)就取值不同的,再不行就全部
differing = [i["key"] for i in row["inputs"] if not _agree(i["value"], values[i["path"]], options.tolerance)]
differing = differing or [i["key"] for i in row["inputs"] if i["value"] != values[i["path"]]]
row["differing"] = differing or [i["key"] for i in row["inputs"]]
def _unchecked_gaps(rows: list[dict[str, Any]]) -> list[str]:
return [GAP_UNCHECKED.format(name=r["name"], reason=r.get("reason") or "原因不明")
for r in rows if r["status"] == UNCHECKED]
# --------------------------------------------------------------------------
# 不一致:差异说明、对照工件、审批(一个 @task)
# --------------------------------------------------------------------------
async def _review(ctx: NodeContext, checked: dict[str, Any]) -> dict[str, Any]:
"""不一致的指标逐个说清差在哪,完整对照落成工件,组好审批的 payload。返回 {payload, facts, card_artifact}。
facts 是每个不一致指标写数据目录要用的东西(dual_check_facts.derive 的候选,少一个 choice):只有一个输入对不上时是
那一处完整的差异条目(带两边的写法和涉及的表、列),几个输入都对不上时只说「不止一处」。
"""
rows = [r for r in checked["rows"] if r["status"] == DIFFERENT]
results = checked["results"]
sources = _Sources()
infos: dict[str, dict[str, Any]] = {}
metrics: list[dict[str, Any]] = []
summary: list[dict[str, Any]] = []
facts: dict[str, dict[str, Any]] = {}
for row in rows:
per_input: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]] = []
for inp in row["inputs"]:
if inp["key"] not in row["differing"]:
continue
if inp["source"] not in infos:
infos[inp["source"]] = await _source_info(sources, inp["source"])
blind = results.get(inp["key"]) or {}
diff = await _explain(inp["sql"], blind.get("sql"), infos[inp["source"]])
per_input.append((inp, blind, diff))
merged = _merge(per_input)
metrics.append({
"metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"], "decimals": row["decimals"],
"original": {**row["original"], "inputs": [
{"path": i["path"], "value": i["value"], "sql": i["sql"], "source": i["source"], "alias": i["alias"]}
for i in row["inputs"]]},
"blind": {**row["blind"], "inputs": [
{"path": i["path"], "value": (results.get(i["key"]) or {}).get("value"),
"sql": (results.get(i["key"]) or {}).get("sql"), "source": (results.get(i["key"]) or {}).get("source"),
"alias": (results.get(i["key"]) or {}).get("alias"),
"rationale": (results.get(i["key"]) or {}).get("rationale") or ""}
for i in row["inputs"]]},
"diff": {"equivalent": merged["equivalent"],
"items": [{"aspect": i.get("aspect"), "text": i.get("text")} for i in merged["items"]],
"fallback": merged["fallback"], "text_diff": merged["text_diff"]},
})
summary.append({
"metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"],
"original_rendered": row["original"]["rendered"], "blind_rendered": row["blind"]["rendered"],
"delta_pct": row["delta_pct"], "diff": [str(i.get("text")) for i in merged["items"][:SUMMARY_DIFFS]],
"fallback": not any(i.get("aspect") not in ("other", "parse") for i in merged["items"]),
})
first = per_input[0][0] if per_input else (row["inputs"][0] if row["inputs"] else {})
single = per_input[0] if len(per_input) == 1 else None
facts[row["metric_id"]] = {
"metric_id": row["metric_id"], "metric": row["name"], "source": first.get("source"),
"dialect": (infos.get(str(first.get("source"))) or {}).get("kind"),
"sql": {"a": single[0]["sql"], "b": single[1].get("sql")} if single else {},
"diff": {"equivalent": merged["equivalent"], "fallback": merged["fallback"] or single is None,
"items": merged["items"]},
}
content = {"metrics": metrics, "model": checked.get("model"), "same_model": bool(checked.get("same_model"))}
try:
artifact = await artifact_store.put_json(content, kind="dual_check", run_id=ctx.run.run_id,
node_id=ctx.node.id, meta={"metrics": len(metrics)})
except Exception as e: # noqa: BLE001 - 对照存不下,审核人就看不到完整的两条 SQL:不能让人凭摘要裁决
raise NodeError(ctx.node.id, f"双盲复算的对照结果保存失败({describe_exception(e)}),请稍后接着运行") from e
payload = {
"kind": "dual_check", "node_id": ctx.node.id, "mode": "input",
"title": f"双盲复算:{len(rows)} 个指标两边结果不一致,请选择采用哪一边的结果",
"card_artifact": artifact, "summary": summary,
}
names = "、".join(f"「{r['name']}」" for r in rows[:5]) + (f"等 {len(rows)} 个" if len(rows) > 5 else "")
ctx.emit(EventType.LOG, level="warn", code="dual_check_disagreed",
metrics=[{"id": r["metric_id"], "name": r["name"]} for r in rows], card_artifact=artifact,
message=f"双盲复算不一致:指标{names}两边的结果对不上,已暂停等待审核")
return {"payload": payload, "facts": facts, "card_artifact": artifact}
async def _source_info(sources: _Sources, name: str) -> dict[str, Any]:
"""比较两条 SQL 要用的:方言、表结构、列的中文名(数据目录里的)。数据源没了就都不给,照样比原文。"""
from app.data import catalog
from app.data.sql_semantics import catalog_labels
row = await sources.source(name)
if row is None:
return {"kind": None, "schema": None, "names": {}}
async with SessionLocal() as session:
entries = await catalog.read_catalog(session, row.id)
return {"kind": row.kind, "schema": row.schema_cache if isinstance(row.schema_cache, dict) else None,
"names": catalog_labels({table: entry.notes for table, entry in entries.items()})}
async def _explain(original: Any, blind: Any, info: dict[str, Any]) -> dict[str, Any]:
"""两条 SQL 的差异说明(SqlDiff.as_dict 的形状,条目带两边的写法和 detail)。规范化后一致的换成「差在数据」的
一句:值不同,写法却一样,具体差异一条都说不出。"""
from app.data.sql_semantics import explain_sql_diff
if not isinstance(original, str) or not original.strip() or not isinstance(blind, str) or not blind.strip():
return {"equivalent": False, "items": [{"aspect": "other", "a": None, "b": None, "text": NO_SQL}],
"fallback": True, "text_diff": []}
try:
diff = await asyncio.wait_for(asyncio.to_thread(
explain_sql_diff, original, blind, info.get("kind"), schema=info.get("schema"), names=info.get("names"),
labels=SIDES), timeout=DIFF_TIMEOUT_S)
except Exception as e: # noqa: BLE001 - 比较本身出了意外(超时、规范化的缺陷):给兜底说明,审核人对照原文
text = DIFF_TIMEOUT if isinstance(e, asyncio.TimeoutError) else DIFF_FAILED
return {"equivalent": False, "items": [{"aspect": "other", "a": None, "b": None, "text": text}],
"fallback": True, "text_diff": [f"- {original.strip()}", f"+ {blind.strip()}"]}
out = diff.as_dict()
if out["equivalent"]:
return {"equivalent": True, "items": [{"aspect": SAME_SQL_ASPECT, "a": None, "b": None, "text": SAME_SQL}],
"fallback": False, "text_diff": []}
return out
def _merge(per_input: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]]) -> dict[str, Any]:
"""一个指标几个对不上的输入的差异合成一份:一个输入时就是它的;几个时每条前面写明是哪一列的输入。"""
if not per_input:
return {"equivalent": False, "items": [{"aspect": "other", "a": None, "b": None, "text": NO_SQL}],
"fallback": True, "text_diff": []}
if len(per_input) == 1:
diff = per_input[0][2]
return {"equivalent": bool(diff["equivalent"]), "items": list(diff["items"]),
"fallback": bool(diff["fallback"]), "text_diff": list(diff.get("text_diff") or [])}
items: list[dict[str, Any]] = []
lines: list[str] = []
for inp, _, diff in per_input:
items += [{**i, "text": f"输入「{inp['column']}」:{i.get('text')}"} for i in diff["items"]]
lines += list(diff.get("text_diff") or [])
return {"equivalent": all(d["equivalent"] for _, _, d in per_input), "items": items,
"fallback": any(d["fallback"] for _, _, d in per_input), "text_diff": lines[:40]}
# --------------------------------------------------------------------------
# 恢复之后:审核人的选择、缺口、数据目录
# --------------------------------------------------------------------------
def _choices(ctx: NodeContext, response: Any, rows: list[dict[str, Any]]) -> tuple[dict[str, str], str]:
"""审核人的回复 → ({指标 id: a | b | neither}, 备注)。回复的 value 是 {choices, note};某个指标没选、选了认不出的
按「两边都不对」:不替人认可任何一边。明确驳回(approved 为否)的按人工驳回让节点失败,和人工审批节点的填写模式
一样:这一步没有驳回的出口。"""
decision = read_decision(response)
if isinstance(response, dict) and decision.rejected:
raise NodeError(ctx.node.id, f"人工驳回:{decision.note or '未说明原因'}")
value = decision.value if isinstance(decision.value, dict) else response if isinstance(response, dict) else {}
raw = value.get("choices") if isinstance(value.get("choices"), dict) else {}
choices = {r["metric_id"]: raw[r["metric_id"]] if raw.get(r["metric_id"]) in CHOICES else "neither" for r in rows}
note = value.get("note") if isinstance(value.get("note"), str) and value.get("note").strip() else decision.note
return choices, str(note or "").strip()
def _gaps(rows: list[dict[str, Any]], choices: dict[str, str]) -> list[str]:
out: list[str] = []
for row in rows:
if row["status"] == UNCHECKED:
out.append(GAP_UNCHECKED.format(name=row["name"], reason=row.get("reason") or "原因不明"))
elif row["status"] == DIFFERENT and choices.get(row["metric_id"]) == "b":
out.append(GAP_BLIND.format(name=row["name"]))
elif row["status"] == DIFFERENT and choices.get(row["metric_id"]) != "a":
out.append(GAP_NEITHER.format(name=row["name"]))
return out
_CHOICE_TEXT = {"a": "采用原流程", "b": "采用盲算", "neither": "两边都不对"}
def _resolved_text(rows: list[dict[str, Any]], choices: dict[str, str]) -> str:
parts = ";".join(f"「{r['name']}」{_CHOICE_TEXT[choices[r['metric_id']]]}" for r in rows[:8])
return f"审核人已裁决 {len(rows)} 个不一致的指标:{parts}" + ("等" if len(rows) > 8 else "")
async def _catalog(ctx: NodeContext, candidates: list[dict[str, Any]]) -> dict[str, Any]:
from app.data.catalog import TABLE_FIELD_LABEL
from app.engine.dual_check_facts import record_choices
result = await record_choices(candidates, run_id=ctx.run.run_id, node_title=ctx.node.title, actor=ctx.actor())
written, skipped = result["written"], result["skipped"]
parts = []
if written:
parts.append("已写入数据目录:" + "、".join(
f"「{w['table']}」的{TABLE_FIELD_LABEL.get(w['item'], w['item'])}" for w in written))
if skipped:
parts.append("未写入数据目录:" + ";".join(s["reason"] for s in skipped[:3])
+ (f"等 {len(skipped)} 条" if len(skipped) > 3 else ""))
ctx.emit(EventType.LOG, level="info", code="dual_check_catalog", written=written, skipped=skipped,
message="。".join(parts))
return result
# --------------------------------------------------------------------------
# 产出
# --------------------------------------------------------------------------
def _finish(state: GraphState, ctx: NodeContext, card_id: str, card: dict[str, Any] | None, checked: dict[str, Any],
*, status: str, gaps: list[str], choices: dict[str, str] | None = None, note: str = "",
artifact: str | None = None, catalog: dict[str, Any] | None = None) -> dict[str, Any]:
rows = checked["rows"]
picked = choices or {}
metrics = [{
"metric_id": r["metric_id"], "name": r["name"], "unit": r["unit"], "status": r["status"],
"original_rendered": r["original"]["rendered"], "blind_rendered": r["blind"]["rendered"],
"delta_pct": r["delta_pct"],
**({"reason": r["reason"]} if r.get("reason") else {}),
**({"choice": picked[r["metric_id"]]} if r["metric_id"] in picked else {}),
} for r in rows]
result: dict[str, Any] = {"status": status, "metrics": metrics, "model": checked.get("model"),
"same_model": bool(checked.get("same_model"))}
if artifact:
result["card_artifact"] = artifact
if choices is not None:
result.update(choices=choices, note=note)
if catalog is not None:
result["catalog"] = catalog
if gaps:
result["issuance_gaps"] = gaps
entry: dict[str, Any] = {
"kind": "dual_check", "node_id": ctx.node.id, "exec": next_exec(state, ctx.node.id), "card": card_id,
"status": status,
"metrics": [{"id": m["metric_id"], "status": m["status"], **({"choice": m["choice"]} if "choice" in m else {})}
for m in metrics],
**({"metric_set": card["artifact"]} if card and isinstance(card.get("artifact"), str) else {}),
**({"artifact": artifact} if artifact else {}),
}
updates: dict[str, Any] = {"nodes": {ctx.node.id: result}, "evidence": [entry]}
usage = checked.get("usage") or {}
if usage.get("calls"):
updates["usage"] = dict(usage)
return updates
__all__ = ["AGREED", "CHOICES", "RESOLVED", "SKIPPED", "run_dual_check"]
+145
View File
@@ -0,0 +1,145 @@
"""双盲复算测试用的虚构零售库和盲算模型的剧本。表名、数字、地名都是编的。
订单表埋了两个陷阱:退款行(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
"""
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"),
(2, "华东", 102, 200.0, "paid", "2026-09-02", "2026-09-02"),
(3, "华南", 101, 50.0, "refunded", "2026-09-03", "2026-09-03"),
(4, "华南", 103, 80.0, "paid", "2026-09-07", "2026-09-08"),
(5, "华北", 102, 30.0, "paid", "2026-09-05", "2026-09-05"),
]
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);")
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, …]、列}。盲算循环里依次查这几条 SQL,收尾时交最后一条查询的
第 0 行那一列(row 可改)。没对上指标名的盲算交 null。不是盲算的调用(报告撰写)返回 writer 的正文并记下收到的消息。
calls 记下每次盲算调用(step / final),tasks 记下盲算收到的任务原文。
"""
def __init__(self, monkeypatch, source: str, plans: dict, *, writer: str = "本周销售额见口径卡。", row: int = 0):
from app.providers import mock_model
self.source = source
self.plans = plans
self.writer = writer
self.row = row
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 not blind.tasks or blind.tasks[-1] != task_text:
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):
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 name, plan in self.plans.items():
if f"指标:{name}" 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")
+160
View File
@@ -0,0 +1,160 @@
"""单独调用双盲复算的执行器(不在图里执行:@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)
+198
View File
@@ -0,0 +1,198 @@
"""双盲复算的选择写进数据目录(engine/dual_check_facts.py)。
- 只差一个方面、又对得上目录事实时才写:时间列 → 业务日期,一边多一个只涉及一张表的筛选条件 → 有效记录条件,
聚合只差去不去重 → 去重规则;按被采纳那一边的写法记
- 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法:不写,照实说原因
- 写的是人工填写、已确认,备注写明运行、节点、指标;同样的说法已确认过就不再写(不升版本),确认过的不同说法不覆盖,
推断的不同说法照改
差异条目一律由 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, record_choices
from tests.fixtures.dual_check_shop import (
BOTH_WRONG,
BUYERS,
BUYERS_DISTINCT,
BY_ORDER_DATE,
RIGHT,
WITH_REFUNDS,
add_source,
build,
)
from tests.fixtures.sources import drop_source
SCHEMA = {"tables": {"orders": {"name": "orders", "columns": [
{"name": "id", "type": "INTEGER"}, {"name": "region", "type": "TEXT"}, {"name": "customer_id", "type": "INTEGER"},
{"name": "amount", "type": "REAL"}, {"name": "status", "type": "TEXT"}, {"name": "order_date", "type": "TEXT"},
{"name": "paid_date", "type": "TEXT"}], "primary_key": ["id"]}}}
def candidate(original: str, blind: str, choice: str, *, metric: str = "销售额", source: str = "shop") -> dict:
diff = explain_sql_diff(original, blind, "sqlite", schema=SCHEMA, labels=("原流程", "盲算")).as_dict()
return {"metric": metric, "choice": choice, "source": source, "dialect": "sqlite",
"sql": {"a": original, "b": blind}, "diff": diff}
# --------------------------------------------------------------------------
# 差异 → 目录事实
# --------------------------------------------------------------------------
@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_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'")
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
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'"
@pytest.mark.parametrize("choice, text", [("b", "对 customer_id 计数时去重"), ("a", "对 customer_id 计数时不去重")])
def test_counting_with_or_without_distinct_becomes_the_dedup_rule(choice, text):
fact, why = derive(candidate(BUYERS, BUYERS_DISTINCT, choice, metric="付款客户数"), SCHEMA)
assert why is None and (fact.table, fact.item, fact.value) == ("orders", "dedup", text)
def test_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 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_different_item_is_replaced(shop):
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"] == [{"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_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_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"]
+571
View File
@@ -0,0 +1,571 @@
"""双盲复算节点(engine/nodes/dual_check.py):真跑一遍图。虚构的 SQLite 订单库埋了退款行、下单日和支付日两个陷阱
(tests/fixtures/dual_check_shop.py),盲算模型是 MockChatModel 剧本。
- 一致:不挂起,产出 agreed,日志 dual_check_agreed;盲算的查询发 tool.end(role: verify),台账只有一条 dual_check,
用量计入运行;盲算 agent 的任务里没有原 SQL
- 不一致:挂起,审批 payload 和对照工件按约定;a / b / neither 三种选择的产出和缺口,on_reject 为 fail 时失败
- 只差一个方面时写数据目录(业务日期、有效记录条件),差了不止一处不写,人工确认过的不同说法不覆盖
- 全局关闭直通、节点覆盖;设置按发起时的快照;重放时盲算不再执行、事件不重复;报告的写作目录不含盲算的查询
- 触顶、分组结果的维度、两个指标用到同一格只盲算一次
"""
from __future__ import annotations
import asyncio
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import delete, select
from app.core import artifact_store
from app.data import catalog
from app.db.base import SessionLocal
from app.db.models import Approval, Run, RunEvent, Setting
from app.engine import spend
from app.engine.launch import launch_run
from app.engine.runner import run_manager
from app.main import app
from tests.fixtures.dual_check_shop import (
BOTH_WRONG,
BY_ORDER_DATE,
BY_REGION,
DEFINITION,
RIGHT,
WEEK,
WITH_REFUNDS,
Blind,
add_source,
build,
)
from tests.fixtures.sources import drop_source
#: 和 RIGHT 等价、写法不同(别名、列名都换了),值也是 330.0:报告的写作目录里若混进了盲算的查询,认得出来
BLIND_TOTAL = f"SELECT SUM(o.amount) AS blind_total FROM orders o WHERE o.status <> 'refunded' AND o.paid_date {WEEK}"
GMV = {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands",
"expression": "cell(nodes.fetch, 0, 'gmv')"}
@pytest.fixture(autouse=True)
async def engine_up():
await run_manager.setup()
yield
await run_manager.shutdown()
@pytest.fixture(autouse=True)
async def clean_settings():
async def clear() -> None:
async with SessionLocal() as session:
await session.execute(delete(Setting).where(Setting.key.in_(["verify", spend.SPEND_KEY])))
await session.commit()
await clear()
yield
await clear()
@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)
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
def node(nid, ntype, label=None, **config):
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}}
def graph(source: str, sql: str, *, dual: dict | None = None, metrics: list | None = None, report: bool = False):
"""取数 → 口径卡 → 双盲复算 →(报告)→ 成果。成果的叙述里没有数字:不加双盲复算时是完整出具。"""
nodes = [
node("start", "input"),
node("fetch", "tool", tool=f"db_query__{source}", args={"sql": sql}),
node("card", "metrics", caliber="销售口径", caliber_version="v1", metrics=metrics or [GMV]),
node("dual", "dual_check", "复核销售额", **{"definitions": {"gmv": DEFINITION}, **(dual or {})}),
]
if report:
nodes.append(node("write", "report", instructions="写一句周报", on_violation="flag"))
nodes.append(node("out", "output", fields=[{"name": "结论", "value": "本周销售额见口径卡"}],
contract={"metrics_from": "card", "narrative": "本周销售额见口径卡"}))
return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
async def wait(run_id: str, statuses=("succeeded", "failed", "interrupted")) -> Run:
for _ in range(600):
await asyncio.sleep(0.05)
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None):
await run_manager.wait_idle(run_id)
return row
raise AssertionError(f"运行 {run_id} 停在 {row.status}:{row.error}")
async def start(spec: dict, **kwargs) -> Run:
run = await run_manager.start(graph=spec, input_payload={}, **kwargs)
return await wait(run.id)
async def events(run_id: str) -> list[RunEvent]:
async with SessionLocal() as session:
return list((await session.execute(
select(RunEvent).where(RunEvent.run_id == run_id).order_by(RunEvent.seq))).scalars())
def logs(evs: list[RunEvent], code: str) -> list[dict]:
return [e.data for e in evs if e.type == "log" and e.data.get("code") == code]
async def pending(run_id: str) -> Approval:
async with SessionLocal() as session:
return (await session.execute(select(Approval).where(
Approval.run_id == run_id, Approval.status == "pending"))).scalar_one()
async def decide(client, run_id: str, choices: dict, note: str = "") -> Run:
approval = await pending(run_id)
res = await client.post(f"/api/approvals/{approval.id}/decide",
json={"approved": True, "value": {"choices": choices, "note": note}})
assert res.status_code == 200, res.text
return await wait(run_id, ("succeeded", "failed"))
def dual_output(run: Run, evs: list[RunEvent]) -> dict:
finished = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"]
assert finished, "双盲复算节点没有跑完"
return artifact_store.load(finished[-1].data["artifact"])
async def notes_of(source_id: str, table: str = "orders"):
async with SessionLocal() as session:
return await catalog.read_entry(session, source_id, table)
# --------------------------------------------------------------------------
# 一致:不挂起
# --------------------------------------------------------------------------
async def test_agreeing_check_passes_without_asking(monkeypatch, shop):
name, _ = shop
blind = Blind(monkeypatch, name, {"销售额": (BLIND_TOTAL, "blind_total")})
run = await start(graph(name, RIGHT))
assert run.status == "succeeded", run.error
evs = await events(run.id)
out = dual_output(run, evs)
assert out["status"] == "agreed" and "issuance_gaps" not in out
[metric] = out["metrics"]
assert metric["metric_id"] == "gmv" and metric["status"] == "agreed" and metric["delta_pct"] == 0.0
assert metric["original_rendered"] == metric["blind_rendered"]
assert [d["metrics"] for d in logs(evs, "dual_check_agreed")] == [1]
assert not logs(evs, "dual_check_disagreed") and not [e for e in evs if e.type == "human.requested"]
# 盲算的查询发 tool.end(role: verify),带查询快照,随运行封存
ends = [e.data for e in evs if e.type == "tool.end" and e.node_id == "dual"]
assert ends and all(d.get("role") == "verify" for d in ends)
assert any(d.get("query_artifact") for d in ends)
# 台账只记一条 dual_check,绝不写 query:报告的写作目录按台账里的 query 编号
[finished] = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"]
kinds = [entry["kind"] for entry in finished.data.get("evidence") or []]
assert kinds == ["dual_check"]
entry = finished.data["evidence"][0]
assert entry["card"] == "card" and entry["status"] == "agreed"
assert entry["metrics"] == [{"id": "gmv", "status": "agreed"}]
# 用量计入运行:图里只有盲算调用了模型
assert blind.calls and run.usage["calls"] == len(blind.calls)
assert run.output["_issuance"]["tier"] == "formal" and run.output["_issuance"]["gaps"] == []
# 盲算 agent 只看得到指标名、口径说明、计算式里是哪一项:看不到原 SQL、取数节点
[task] = blind.tasks
assert "指标:销售额" in task and DEFINITION in task and "输入1" in task
assert "status <> 'refunded'" not in task and "nodes.fetch" not in task and "SUM(" not in task
async def test_grouped_rows_tell_the_blind_agent_which_group(monkeypatch, shop):
"""结果分了组:盲算 agent 要知道算的是哪一组(同一行的维度列),但看不到同一行别的数。"""
name, _ = shop
metric = {**GMV, "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
blind = Blind(monkeypatch, name, {"华东销售额": (BY_REGION, "gmv")})
run = await start(graph(name, BY_REGION, metrics=[metric], dual={"definitions": {}}))
assert run.status == "succeeded", run.error
[task] = blind.tasks
assert "region 为「华东」" in task and "(未提供" in task
out = dual_output(run, await events(run.id))
assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed"
async def test_two_metrics_using_one_cell_are_checked_once(monkeypatch, shop):
name, _ = shop
share = {"id": "gmv_k", "name": "销售额(千元)", "unit": "千元", "decimals": 2, "format": "thousands",
"expression": "round(cell(nodes.fetch, 0, 'gmv') / 1000, 2)"}
blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, RIGHT, metrics=[GMV, share]))
assert run.status == "succeeded", run.error
assert blind.runs == 1, "同一格盲算了不止一次"
out = dual_output(run, await events(run.id))
assert [m["status"] for m in out["metrics"]] == ["agreed", "agreed"]
async def test_hitting_a_limit_counts_as_unchecked(monkeypatch, shop):
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, RIGHT), options={"verify": {"dual_check": {"max_queries": 1}}})
assert run.status == "succeeded", run.error
evs = await events(run.id)
assert [(d["limit"], d["metric"]) for d in logs(evs, "dual_check_limit")] == [("max_queries", "gmv")]
out = dual_output(run, evs)
assert out["status"] == "agreed" and out["metrics"][0]["status"] == "unchecked"
assert out["issuance_gaps"] == ["指标「销售额」无法复算:盲算已达每项最多查询次数(1 次)"]
assert not [e for e in evs if e.type == "human.requested"]
assert run.output["_issuance"]["tier"] == "degraded"
assert run.output["_issuance"]["gaps"] == ["双盲复算「复核销售额」:指标「销售额」无法复算:盲算已达每项最多查询次数(1 次)"]
async def test_an_unusable_verify_model_is_tried_once(monkeypatch, shop):
"""核查模型没配好:每个指标都记无法复算、原因照实,后面的盲算不再一个个去试。"""
from app.engine import verify_agent
from app.providers.factory import ProviderNotConfigured
name, _ = shop
tried: list[str] = []
async def broken(session, spec):
tried.append(spec.model or "")
raise ProviderNotConfigured("核查用的模型接入已停用", "请到「设置 → 模型接入」中启用")
monkeypatch.setattr(verify_agent, "get_chat_model", broken)
east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
north = {**GMV, "id": "north", "name": "华北销售额", "expression": "cell(nodes.fetch, 1, 'gmv')"}
run = await start(graph(name, BY_REGION, metrics=[east, north], dual={"definitions": {}}))
assert run.status == "succeeded", run.error
out = dual_output(run, await events(run.id))
assert [m["status"] for m in out["metrics"]] == ["unchecked", "unchecked"]
assert all("核查模型不可用" in m["reason"] and "已停用" in m["reason"] for m in out["metrics"])
assert len(out["issuance_gaps"]) == 2 and len(tried) == 1
# --------------------------------------------------------------------------
# 不一致:挂起,按约定的 payload 和工件
# --------------------------------------------------------------------------
async def test_disagreement_pauses_with_the_agreed_payload_and_card(monkeypatch, shop):
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
assert run.status == "interrupted", run.error
approval = await pending(run.id)
assert approval.mode == "input" and approval.node_id == "dual"
payload = approval.payload
assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary"}
assert payload["kind"] == "dual_check" and payload["node_id"] == "dual" and payload["mode"] == "input"
[item] = payload["summary"]
assert set(item) == {"metric_id", "name", "unit", "original_rendered", "blind_rendered", "delta_pct", "diff",
"fallback"}
assert (item["metric_id"], item["name"], item["unit"]) == ("gmv", "销售额", "元")
assert item["original_rendered"].startswith("310") and item["blind_rendered"].startswith("330")
assert item["delta_pct"] == round((330 - 310) / 310 * 100, 2)
assert item["fallback"] is False and len(item["diff"]) == 1
assert "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] and "原流程" in item["diff"][0]
card = artifact_store.load(payload["card_artifact"])
assert set(card) == {"metrics", "model", "same_model"} and card["same_model"] is False
[metric] = card["metrics"]
assert set(metric) == {"metric_id", "name", "unit", "decimals", "original", "blind", "diff"}
assert metric["decimals"] == 1
[orig] = metric["original"]["inputs"]
assert set(orig) == {"path", "value", "sql", "source", "alias"}
assert (orig["sql"], orig["value"], orig["source"], orig["alias"]) == (BY_ORDER_DATE, 310.0, name, "Q1")
[mine] = metric["blind"]["inputs"]
assert set(mine) == {"path", "value", "sql", "source", "alias", "rationale"}
assert (mine["sql"], mine["value"], mine["source"], mine["alias"]) == (RIGHT, 330.0, name, "Q1")
assert mine["path"] == orig["path"] == "cell(nodes.fetch, 0, 'gmv')" and "330" in mine["rationale"]
assert metric["original"]["value"] == 310.0 and metric["blind"]["value"] == 330.0
assert set(metric["diff"]) == {"equivalent", "items", "fallback", "text_diff"}
assert metric["diff"]["equivalent"] is False and metric["diff"]["fallback"] is False
assert [set(i) for i in metric["diff"]["items"]] == [{"aspect", "text"}]
assert metric["diff"]["items"][0]["aspect"] == "time" and metric["diff"]["text_diff"]
evs = await events(run.id)
[said] = logs(evs, "dual_check_disagreed")
assert said["metrics"] == [{"id": "gmv", "name": "销售额"}] and said["card_artifact"] == payload["card_artifact"]
assert len([e for e in evs if e.type == "human.requested"]) == 1
async def test_choosing_the_original_leaves_no_gap(monkeypatch, shop, client):
name, source_id = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
done = await decide(client, run.id, {"gmv": "a"}, note="周报一直按下单日统计")
assert done.status == "succeeded", done.error
evs = await events(run.id)
out = dual_output(done, evs)
assert out["status"] == "resolved" and out["choices"] == {"gmv": "a"} and "issuance_gaps" not in out
assert out["note"] == "周报一直按下单日统计" and out["metrics"][0]["choice"] == "a"
assert [d["gaps"] for d in logs(evs, "dual_check_resolved")] == [[]]
assert logs(evs, "dual_check_resolved")[0]["choices"] == {"gmv": "a"}
assert done.output["_issuance"]["tier"] == "formal"
assert len([e for e in evs if e.type == "human.resolved"]) == 1
# 只差时间列一处:采纳原流程,业务日期记原流程那一列
entry = await notes_of(source_id)
item = entry.notes["business_date"]
assert item["value"] == {"column": "order_date"} and (item["source"], item["status"]) == ("human", "confirmed")
assert run.id in item["note"] and "复核销售额" in item["note"] and "销售额" in item["note"]
[wrote] = logs(evs, "dual_check_catalog")
assert wrote["written"] == [{"source": name, "table": "orders", "item": "business_date"}] and wrote["skipped"] == []
async def test_choosing_the_blind_result_degrades_and_writes_the_catalog(monkeypatch, shop, client):
name, source_id = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
done = await decide(client, run.id, {"gmv": "b"})
assert done.status == "succeeded", done.error
out = dual_output(done, await events(run.id))
assert out["issuance_gaps"] == ["指标「销售额」双盲复算不一致,审核人选择了盲算结果"]
issuance = done.output["_issuance"]
assert issuance["tier"] == "degraded"
assert issuance["gaps"] == ["双盲复算「复核销售额」:指标「销售额」双盲复算不一致,审核人选择了盲算结果"]
assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"}
async def test_neither_degrades_and_writes_nothing(monkeypatch, shop, client):
name, source_id = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
done = await decide(client, run.id, {"gmv": "neither"}, note="两边都没排除测试单")
assert done.status == "succeeded", done.error
evs = await events(run.id)
out = dual_output(done, evs)
assert out["issuance_gaps"] == ["指标「销售额」双盲复算不一致,审核人判断两边都不对"]
assert done.output["_issuance"]["tier"] == "degraded"
assert not logs(evs, "dual_check_catalog") and await notes_of(source_id) is None
async def test_a_missing_choice_counts_as_neither(monkeypatch, shop, client):
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
done = await decide(client, run.id, {"gmv": "maybe"})
out = dual_output(done, await events(run.id))
assert out["choices"] == {"gmv": "neither"}
assert out["issuance_gaps"] == ["指标「销售额」双盲复算不一致,审核人判断两边都不对"]
async def test_on_reject_fail_stops_the_run(monkeypatch, shop, client):
name, source_id = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE, dual={"on_reject": "fail"}))
done = await decide(client, run.id, {"gmv": "b"})
assert done.status == "failed" and "审核人没有采用原流程的结果" in (done.error or "") and "销售额" in done.error
evs = await events(run.id)
assert [d["gaps"] for d in logs(evs, "dual_check_resolved")] == [
["指标「销售额」双盲复算不一致,审核人选择了盲算结果"]]
# 审核人的选择照样写进了目录:那是对数据的判断,和运行停不停无关
entry = await notes_of(source_id)
assert entry.notes["business_date"]["value"] == {"column": "paid_date"} and entry.version == 1
async def test_on_reject_fail_lets_the_original_through(monkeypatch, shop, client):
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE, dual={"on_reject": "fail"}))
done = await decide(client, run.id, {"gmv": "a"})
assert done.status == "succeeded", done.error
async def test_the_same_choice_twice_writes_the_catalog_once(monkeypatch, shop, client):
"""同样的不一致、同样的选择再来一次:目录里已经是这个说法,不再写,也不升版本。"""
name, source_id = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
for _ in range(2):
run = await start(graph(name, BY_ORDER_DATE))
await decide(client, run.id, {"gmv": "b"})
[wrote] = logs(await events(run.id), "dual_check_catalog")
assert wrote["written"] == [] and "无需重复写入" in wrote["skipped"][0]["reason"]
assert (await notes_of(source_id)).version == 1
# --------------------------------------------------------------------------
# 写数据目录:只差一个方面才写,确认过的不覆盖
# --------------------------------------------------------------------------
async def test_a_refund_filter_becomes_the_valid_row_filter(monkeypatch, shop, client):
name, source_id = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, WITH_REFUNDS))
payload = (await pending(run.id)).payload
assert "status = 'refunded'" in payload["summary"][0]["diff"][0]
await decide(client, run.id, {"gmv": "b"})
item = (await notes_of(source_id)).notes["valid_filter"]
assert item["value"] == "status <> 'refunded'" and item["status"] == "confirmed" and item["source"] == "human"
async def test_the_adopted_side_without_the_filter_writes_nothing(monkeypatch, shop, client):
"""多出条件的是原流程,审核人却采纳了没有这个条件的盲算:目录里没有对应的写法,照实说。"""
name, source_id = shop
Blind(monkeypatch, name, {"销售额": (WITH_REFUNDS, "gmv")})
run = await start(graph(name, RIGHT))
done = await decide(client, run.id, {"gmv": "b"})
[wrote] = logs(await events(done.id), "dual_check_catalog")
assert wrote["written"] == [] and "被采纳的盲算没有这个条件" in wrote["skipped"][0]["reason"]
assert await notes_of(source_id) is None
async def test_two_differences_are_not_written(monkeypatch, shop, client):
name, source_id = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BOTH_WRONG))
assert len((await pending(run.id)).payload["summary"][0]["diff"]) == 2
done = await decide(client, run.id, {"gmv": "b"})
[wrote] = logs(await events(done.id), "dual_check_catalog")
assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"]
assert await notes_of(source_id) is None
async def test_a_confirmed_catalog_entry_is_not_overwritten(monkeypatch, shop, client):
name, source_id = shop
async with SessionLocal() as session:
await catalog.write_entry(session, source_id, "orders", {
"business_date": catalog.make_item({"column": "order_date"}, "human", "confirmed")}, if_version=0,
actor="数据组")
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
done = await decide(client, run.id, {"gmv": "b"})
[wrote] = logs(await events(done.id), "dual_check_catalog")
assert wrote["written"] == []
assert "已有人工确认过的说法「order_date」" in wrote["skipped"][0]["reason"] and "paid_date" in wrote["skipped"][0]["reason"]
entry = await notes_of(source_id)
assert entry.version == 1 and entry.notes["business_date"]["value"] == {"column": "order_date"}
# --------------------------------------------------------------------------
# 设置:全局开关、节点覆盖、发起时的快照
# --------------------------------------------------------------------------
async def save_verify(values: dict) -> None:
async with SessionLocal() as session:
row = await session.get(Setting, "verify")
if row is None:
session.add(Setting(key="verify", value=values))
else:
row.value = values
await session.commit()
async def launch(spec: dict) -> Run:
async with SessionLocal() as session:
run = await launch_run(session, graph=spec, actor="测试")
return await wait(run.id)
async def test_globally_disabled_passes_through(monkeypatch, shop):
name, _ = shop
blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
await save_verify({"dual_check": {"enabled": False}})
run = await launch(graph(name, BY_ORDER_DATE))
assert run.status == "succeeded", run.error
assert run.options["verify"]["dual_check"]["enabled"] is False
evs = await events(run.id)
assert dual_output(run, evs) == {"status": "skipped", "reason": "disabled"}
assert [d["reason"] for d in logs(evs, "dual_check_skipped")] == ["disabled"]
assert blind.calls == [] and run.output["_issuance"]["tier"] == "formal"
async def test_the_node_can_override_the_global_switch(monkeypatch, shop):
name, _ = shop
blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
await save_verify({"dual_check": {"enabled": False}})
run = await launch(graph(name, RIGHT, dual={"enabled": True}))
assert run.status == "succeeded" and dual_output(run, await events(run.id))["status"] == "agreed"
assert blind.runs == 1
await save_verify({"dual_check": {"enabled": True}})
off = await launch(graph(name, RIGHT, dual={"enabled": False}))
assert dual_output(off, await events(off.id)) == {"status": "skipped", "reason": "disabled"}
assert blind.runs == 1
async def test_no_query_inputs_pass_through(monkeypatch, shop):
name, _ = shop
blind = Blind(monkeypatch, name, {})
constant = {"id": "target", "name": "目标", "unit": "元", "expression": "300"}
run = await start(graph(name, RIGHT, metrics=[constant], dual={"definitions": {}}))
evs = await events(run.id)
assert dual_output(run, evs) == {"status": "skipped", "reason": "no_inputs"}
assert [d["reason"] for d in logs(evs, "dual_check_skipped")] == ["no_inputs"] and blind.calls == []
assert run.output["_issuance"]["tier"] == "formal" and run.output["_issuance"]["gaps"] == []
async def test_settings_changed_while_waiting_do_not_apply(monkeypatch, shop, client):
"""发起时是开着的:停在审批上的时候把双盲复算关了,恢复后照旧按发起时的设置裁决,不会走成直接放行。"""
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await launch(graph(name, BY_ORDER_DATE))
assert run.status == "interrupted" and run.options["verify"]["dual_check"]["enabled"] is True
await save_verify({"dual_check": {"enabled": False}})
done = await decide(client, run.id, {"gmv": "b"})
assert done.status == "succeeded", done.error
evs = await events(run.id)
out = dual_output(done, evs)
assert out["status"] == "resolved" and out["issuance_gaps"]
assert not logs(evs, "dual_check_skipped")
started = [e.data for e in evs if e.type == "run.started"]
assert len(started) == 2 and all(s["options"]["verify"]["dual_check"]["enabled"] is True for s in started)
async def test_the_node_reads_the_snapshot_not_the_current_settings(monkeypatch, shop):
name, _ = shop
blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE), options={"verify": {"dual_check": {"enabled": False}}})
assert dual_output(run, await events(run.id)) == {"status": "skipped", "reason": "disabled"}
assert blind.calls == []
# --------------------------------------------------------------------------
# 重放、写作目录
# --------------------------------------------------------------------------
async def test_resuming_does_not_run_the_blind_agent_again(monkeypatch, shop, client):
"""照 test_replay_side_effects:恢复后节点从头重放,盲算 agent 不再执行,事件不重复。"""
name, _ = shop
blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
calls_before = list(blind.calls)
before = await events(run.id)
verify_starts = [e for e in before if e.type == "tool.start" and e.node_id == "dual"]
llm_ends = [e for e in before if e.type == "llm.end" and e.node_id == "dual"]
assert verify_starts and llm_ends
done = await decide(client, run.id, {"gmv": "a"})
assert done.status == "succeeded", done.error
assert blind.calls == calls_before, f"恢复之后盲算又调用了模型:{blind.calls[len(calls_before):]}"
evs = await events(run.id)
assert [e.seq for e in evs if e.type == "tool.start" and e.node_id == "dual"] == [e.seq for e in verify_starts]
assert [e.seq for e in evs if e.type == "llm.end" and e.node_id == "dual"] == [e.seq for e in llm_ends]
for code in ("dual_check_disagreed", "dual_check_resolved", "dual_check_catalog"):
assert len(logs(evs, code)) == 1, code
assert len([e for e in evs if e.type == "human.requested"]) == 1
assert len([e for e in evs if e.type == "human.resolved"]) == 1
assert len([e for e in evs if e.type == "node.finished" and e.node_id == "dual"]) == 1
async def test_the_report_catalog_leaves_out_the_blind_queries(monkeypatch, shop):
name, _ = shop
blind = Blind(monkeypatch, name, {"销售额": (BLIND_TOTAL, "blind_total")}, writer="本周销售额 [[m:gmv]]。")
run = await start(graph(name, RIGHT, report=True))
assert run.status == "succeeded", run.error
[prompt] = blind.writer_prompts
assert "blind_total" not in prompt and "Q2" not in prompt, "盲算的查询进了报告的写作目录"
assert "Q1" in prompt and "m:gmv" in prompt
evs = await events(run.id)
blind_queries = [e.data["query_artifact"] for e in evs
if e.type == "tool.end" and e.node_id == "dual" and e.data.get("query_artifact")]
ledger = [entry for e in evs if e.type == "node.finished" for entry in e.data.get("evidence") or []]
assert blind_queries and not any(entry.get("artifact") in blind_queries for entry in ledger)
+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)