mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:31:58 +00:00
运行对比复核修正（第 1 轮）：输入的出处追到合并查询的上游、Agent 引用的列和行段，没出处的数、数据没变数却变了的不再报成数据变化
出处判定（每个指标的分类靠它）： - 合并查询：读合并结果的指标按别名一路追到喂它的每一份上游快照（嵌套的合并也追），上游 SQL 结构变了是 sql_changed， 不拆、不归因；合并节点多接、少接了输入也算结构变化 - 图里写死的字面量改了（'paid' 改成 'refunded'、系数、LIMIT）：执行的 SQL 只差字面量，但图里的 SQL 改过，算模板变化； 图里一字未改（定时运行代入的日期）才是参数。图里的 SQL 只改了空白、大小写的不算改过 - Agent 按出处交来的字段：还要引用同一列、同一行（段），两边都引用整份结果的不算换了；换了列、换了行段的说不清 （incomparable），不再把没引用的区域算成「数据没了」去归因 - 模型、人直接交出的值（Agent 没按出处交、模型调用、人工审批……）：值变了说不清是数据还是模型本身，判 incomparable； 它自己查的库结构变了、配置变了照常标出 - 代码等确定性节点：上游的查询结果两次一模一样、交出的值却变了的，判 incomparable，不再说「变化来自上游数据」； 上游只有运行输入变了的算参数 - 图级变化落到节点上：连线有增删的终点节点、默认设置改了且没有自行设置这几项的模型类节点，都算改了配置； 「模板和取数的 SQL 都没有变化」只在整张图都没变时说 - 运行输入是列表、对象的，按还原出来的完整值比，长度不变、内容变了的是参数变化 - 同一节点按 SQL 配对时先配字面量也相同的：Agent 换了先后查东区、南区，两条各自配上自己，不再配成两条「只有参数不同」 - 一边记不出输入是哪个节点写的，说清是哪一边没记录（不再写出「None」） 口径与数值： - 口径变了的指标，喂它的取数也变了、说不清的照样写进说明（只在一边才有的输入是改表达式带来的，不重复说）； 分类次序写进 compare() 的说明 - 千分位和不分组互换只影响显示，写进说明、不算口径变化；比率与否不同才算 - 升级前的口径卡没记小数位、显示格式：按那次运行图里的指标定义核对，定义也没有就不比，不再误报「小数位由 未设置 改为 1」 - 百分比：基准极小、这次很大时十进制量化放不下，退回浮点算，不再整个对比报 500 - 没设小数位的指标拆分：三个数按同一精度量化（最多 15 位有效数字），转成浮点交出去后两效应之和仍逐位等于总变化 - 维度归因标出用的是哪个输入、两边各自哪份快照（path、base_artifact、target_artifact） 接口：已经结束、只是没有封存清单的运行，409 说「没有封存清单，不能用于对比」，不再说「运行结束后才能对比」。 测试：补上复核指出的必测项（只一边对得上不给占比、只截断基准一侧、换了数据源、按旧口径小数位取整、写它的节点多查一次、 口径和 SQL 一起变不归因、小数位和比率变化、日期类型和常量列不当维度、合计行浮点误差、只改名称、结束过又接着运行的不当 基准），以及上面每一处修正的用例。32 处变异逐条打红验证，全部被测试抓到。
This commit is contained in:
ThearchyHelios
@@ -7,8 +7,8 @@ GET /api/runs/{run_id}/compare/base：默认拿哪一次来比，{base_run_id, r
|
||||
|
||||
GET /api/runs/{run_id}/compare?base={base_id}：{target, base, summary, template, metrics, queries, notes}，每个字段
|
||||
的形状见 engine/compare.py。两次运行都必须已结束、封存且封存核对一致，封存判定照搬证据接口（api/evidence._sealed）：
|
||||
没封存的运行事件还会变，封存核对不一致的事件被改过，拿它们比出来的结论都不作数，回 409。不是同一个工作流、或者
|
||||
是同一次运行，回 422。
|
||||
没封存的运行事件还会变，封存核对不一致的事件被改过，拿它们比出来的结论都不作数，回 409（还没结束的和已经结束却没有
|
||||
封存清单的分开说：后者再等也等不来封存）。不是同一个工作流、或者是同一次运行，回 422。
|
||||
|
||||
只读：两次运行都不追加事件、不落工件、不改运行记录；结果只交回给发请求的人。只认各自封存范围内的工件（证据接口的
|
||||
白名单，加上封存的节点产出和指标集里写着的完整值，同推演），遮罩列（查询当时记下的加数据源现在设的）的值不出现在
|
||||
@@ -48,6 +48,8 @@ SAME_RUN = "不能和同一次运行自己对比，请选择另一次运行作
|
||||
UNSAVED = "{who}不属于任何已保存的工作流，无法确认两次运行是同一个工作流，不能对比"
|
||||
OTHER_WORKFLOW = "两次运行不属于同一个工作流（「{target}」与「{base}」），不能对比"
|
||||
UNSEALED_TEXT = "{who}尚未结束并封存：对比只用封存的运行记录，运行结束后才能对比"
|
||||
#: 已经结束、只是没有封存清单（封存上线前结束的、结束时没封上的）：再等也等不来封存，不能说「运行结束后才能对比」
|
||||
NO_MANIFEST_TEXT = "{who}没有封存清单，不能用于对比"
|
||||
SEAL_BROKEN_TEXT = "{who}的封存核对不一致：封存后有事件被修改、删除或插入，不能用于对比。可在记录页点「核对」查看核对结果"
|
||||
NO_BASE_TEXT = ("没有找到可以对比的上一次运行：需要同一工作流、同一运行类别（定时任务发起的优先找同一定时任务）、"
|
||||
"更早结束并已封存的运行")
|
||||
@@ -120,8 +122,10 @@ async def compare_runs(run_id: str, base_id: str) -> dict[str, Any]:
|
||||
sealed = []
|
||||
for run, who in ((target, TARGET), (base, BASE)):
|
||||
found = await _sealed(run.id)
|
||||
if not found.sealed or found.status not in _TERMINAL:
|
||||
if found.status not in _TERMINAL:
|
||||
raise CodedHTTPException(409, UNSEALED_TEXT.format(who=who), UNSEALED)
|
||||
if not found.sealed:
|
||||
raise CodedHTTPException(409, NO_MANIFEST_TEXT.format(who=who), UNSEALED)
|
||||
if not found.trusted:
|
||||
raise CodedHTTPException(409, SEAL_BROKEN_TEXT.format(who=who), SEAL_BROKEN)
|
||||
sealed.append(found)
|
||||
|
||||
+481
-107
@@ -7,22 +7,31 @@
|
||||
怎么对：
|
||||
|
||||
- 查询：两次运行各自封存范围内交回过的查询快照，按「节点 + 该节点里第几次查询 + 工具」配对，配不上的再按规范化
|
||||
SQL（data/sql_semantics.normalize_sql）配；同一节点、同一工具里规范化 SQL 相同的先配上（Agent 每次查询的先后
|
||||
可能不同，见 _pair）。结构：规范文本和字面量都相同是 same；规范文本相同、只有字面量不同是 params_only（定时运行
|
||||
每次的日期不同，属于参数变化不是模板变化）；否则 changed（数据源换了也算）。核查类的查询（双盲复算、反方审稿发的
|
||||
role: verify）不进报告，不参与对比。
|
||||
- 口径卡：同一个口径卡节点按指标 id 配对；循环里执行了几次的取最后一次。表达式、口径版本、小数位、单位、显示格式
|
||||
有一样不同就是口径变了：小数位改变取整，单位、比率与否改变数值的含义，都不只是换个写法。
|
||||
- 每个指标一个分类，按这个次序判：一边有一边没有（added / removed）；口径变了（caliber_and_data，可能连同数据）；
|
||||
喂它的查询结构变了、上游节点改了配置（sql_changed，标出、不拆）；输入对不上具体的查询，上游又有变化，说不清变化
|
||||
从哪来（incomparable）；只有参数变了（params_only，含运行输入）；都没变时数值相等是 unchanged，不等是 data_only。
|
||||
「喂它的查询」按输入的出处认：cell() 那一格、Agent 按出处交来的字段落在哪份快照上就是哪条查询（两边各自的快照
|
||||
直接比 SQL）；不落在快照上的，看写它的节点自己查过的库，节点不查库（代码、数据整形……）就看它的配置和画布上的
|
||||
全部上游。
|
||||
SQL（data/sql_semantics.normalize_sql）配；同一节点、同一工具里规范化 SQL 相同的先配上，字面量也相同的又在前
|
||||
（Agent 每次查询的先后可能不同，见 _pair）。结构：规范文本和字面量都相同是 same；规范文本相同、只有字面量不同是
|
||||
params_only（定时运行每次的日期不同，属于参数变化不是模板变化）；否则 changed（数据源换了也算）。核查类的查询
|
||||
（双盲复算、反方审稿发的 role: verify）不进报告，不参与对比。
|
||||
- 口径卡：同一个口径卡节点按指标 id 配对；循环里执行了几次的取最后一次。表达式、口径版本、小数位、单位、比率与否
|
||||
有一样不同就是口径变了：小数位改变取整，单位、比率与否改变数值的含义，都不只是换个写法。千分位和不分组互换只影响
|
||||
显示，写进说明、不算口径变化。升级前的卡没记小数位、显示格式的，按那次运行图里的指标定义核对，定义也没有就不比。
|
||||
- 每个指标一个分类，按这个次序判：一边有一边没有（added / removed）；口径变了（caliber_and_data，可能连同数据；
|
||||
喂它的取数也有变化的照样写进说明）；喂它的查询结构变了、上游节点改了配置（sql_changed，标出、不拆）；输入对不上、
|
||||
说不清变化从哪来（incomparable）；只有参数变了（params_only，含运行输入）；都没变时数值相等是 unchanged，
|
||||
不等是 data_only。
|
||||
- 「喂它的查询」按输入的出处认：
|
||||
- cell() 那一格、Agent 按出处交来的字段落在哪份快照上就是哪条查询，两边各自的快照直接比 SQL；快照是合并查询的
|
||||
结果时，再按别名一路追到喂它的每一份上游快照。只差字面量时还要看图里的 SQL：图里的 SQL 改过（改了写死的筛选值、
|
||||
系数）是模板变化，图里一字未改的才是运行时代入的参数
|
||||
- Agent 按出处交来的字段还要引用同一列、同一段行（两边都引用整份结果的不算换了）：取哪一列、哪几行是 Agent 每次
|
||||
运行时定的，换了就不是同一个数
|
||||
- 不落在快照上的：模型、人交出的值（Agent 没按出处交、模型调用、人工审批……）没有出处，值变了说不清是数据还是它
|
||||
本身的差异；代码、数据整形这类节点看它自己查过的库，不查库的看它的配置和画布上的全部上游，上游的查询结果两次
|
||||
一模一样、值却变了的同样说不清
|
||||
- 图级的变化落到节点上：连线有增删的，终点节点算改了；默认设置改了的，没有自行设置这几项的模型类节点算改了
|
||||
- 严格拆分（只对口径变了的）：旧表达式的每个叶子路径都能在这次的口径卡输入里取得值时，
|
||||
口径效应 = f_new(x_new) − f_old(x_new)，数据效应 = f_old(x_new) − f_old(x_old)。f_old(x_new) 用推演重算
|
||||
（engine/recompute.py）的还原和求值算，取整按旧口径卡的小数位；三个数按十进制精确相减，两效应之和逐位等于
|
||||
总变化。取数的 SQL 也变了、用到了遮罩列、单位不同的不拆，说明为什么。
|
||||
总变化（转成浮点后也逐位相加，见 _split_exact）。取数的 SQL 也变了、用到了遮罩列、单位不同的不拆，说明为什么。
|
||||
- 维度归因：指标的输入落在查询快照的某一格（或 Agent 数组字段的一段行）、两边快照都没截断时，取快照里不是数值的
|
||||
列作维度（遮罩的、日期类型的不算；两边都只有一个值的、两边没有一个值相同的——日期、期次这类每期都不同的列——
|
||||
也不拿来配对），按维度键配对两边的行，算每个键的 base / target / delta，按 |delta| 取前 10。只有度量列之和
|
||||
@@ -44,9 +53,10 @@ from app.data.sql_semantics import NormalizedSql, diff_facets, facets, normalize
|
||||
from app.engine import recompute
|
||||
from app.engine.evidence import DEFAULT_FORMAT, RenderError, is_number, render_number
|
||||
from app.engine.expressions import ExpressionError, cell_value, leaf_refs, parse_expression
|
||||
from app.engine.labels import field_label
|
||||
from app.engine.labels import TYPE_LABEL, field_label
|
||||
from app.engine.merge_query import MERGE_SOURCE
|
||||
from app.engine.nodes.metrics import _REF_HEAD
|
||||
from app.engine.nodes.metrics import _REF_HEAD, _display_of
|
||||
from app.engine.toolcalls import QUERY_PREFIX
|
||||
|
||||
UNCHANGED = "unchanged"
|
||||
DATA_ONLY = "data_only"
|
||||
@@ -74,6 +84,18 @@ _MERGE_DIALECT = "sqlite"
|
||||
#: 日期类的列不当维度：每期都不同，按它配对两边的行一个都配不上
|
||||
_TEMPORAL = frozenset({"date", "datetime", "time"})
|
||||
_FORMAT_LABEL = {"thousands": "千分位", "plain": "不分组", "percent_of_ratio": "比率按百分数显示"}
|
||||
_RATIO = "percent_of_ratio"
|
||||
#: 图里写 SQL 的配置项：调用工具节点的参数、合并查询的合并 SQL
|
||||
_SQL_KEY = {"tool": "args.sql", "merge": "sql"}
|
||||
#: 交出的值由模型或人给出的节点（类型、口径卡记的来历）：同样的输入、同样的配置，交出的值也可能不同
|
||||
_GENERATED = frozenset({"agent", "llm", "supervisor", "human", "memory", "retrieve", "subgraph"})
|
||||
_GENERATED_VIA = frozenset({"agent", "agent_field", "llm", "team", "human", "memory", "retrieval", "subgraph"})
|
||||
#: 图级默认设置只有模型类节点在读（节点没配 provider、model 时回退到它，GraphSpec.defaults）
|
||||
_MODEL_TYPES = frozenset({"llm", "agent", "supervisor"})
|
||||
#: 改了也不影响交出的值的默认设置：审批策略只管调用工具前问不问人
|
||||
_QUIET_DEFAULTS = frozenset({"approval"})
|
||||
#: 浮点能原样写回的十进制有效数字位数（DBL_DIG）
|
||||
_FLOAT_DIGITS = 15
|
||||
|
||||
NO_CARDS = "两次运行都没有口径卡，没有可以对比的指标"
|
||||
EDGES_CHANGED = "节点之间的连线有变化"
|
||||
@@ -82,6 +104,8 @@ MASKED_NOTE = "部分列在数据源中设置了遮罩，对比中不显示这
|
||||
NO_LINEAGE = "口径卡没有记录输入（生成于系统升级前），无法判断变化来源"
|
||||
DATA_FROM_QUERIES = "取数的 SQL 没有变化，查询结果有变化"
|
||||
DATA_FROM_UPSTREAM = "模板和取数的 SQL 都没有变化，变化来自上游数据"
|
||||
#: 图有别的改动、都不在它的上游：不能说「模板没有变化」
|
||||
DATA_FROM_UPSTREAM_SCOPED = "它上游的节点和取数的 SQL 都没有变化，变化来自上游数据"
|
||||
SQL_NOT_SPLIT = "取数的 SQL 或上游配置有变化，变化无法在口径与数据之间拆分"
|
||||
NOT_NUMBERS = "有一边的值不是数值，无法拆分"
|
||||
SPLIT_MASKED = "用到了设置了遮罩的列，不做拆分"
|
||||
@@ -136,9 +160,13 @@ def compare(base: Side, target: Side) -> dict[str, Any]:
|
||||
delta_pct（相对基准的百分比，两位小数；基准为 0 时为 None）, classification, reasons: [说明], decomposition:
|
||||
{caliber_effect, data_effect, caliber_rendered, data_rendered} 或 None, decomposition_note（为什么没拆）,
|
||||
inputs: [{path, base, target, masked?, base_artifact?, target_artifact?}], attribution}
|
||||
- attribution：{columns, measure, rows: [{key: [维度值…], base, target, delta, share}], total_delta, note} 或
|
||||
None。归不了因的（快照截断、度量列遮罩、快照读不出来、维度列每期都不同）rows 为空、total_delta 为 None，原因
|
||||
写在 note；一边没有的维度值那一边写 None
|
||||
- classification 按这个次序取第一个对得上的：added / removed → caliber_and_data → sql_changed → incomparable
|
||||
→ params_only → unchanged / data_only。口径变了的指标喂它的取数也变了时，分类仍是 caliber_and_data，取数的
|
||||
变化接在口径的说明后面写进 reasons
|
||||
- attribution：{path, base_artifact, target_artifact（归因用的是哪个输入、两边各自哪份快照）, columns, measure,
|
||||
rows: [{key: [维度值…], base, target, delta, share}], total_delta, note} 或 None。归不了因的（快照截断、
|
||||
度量列遮罩、快照读不出来、维度列每期都不同）rows 为空、total_delta 为 None，原因写在 note；一边没有的维度值
|
||||
那一边写 None
|
||||
- queries：每对查询一项 {node_id, node_title, ordinal, structure, base_sql, target_sql, literals: [{base,
|
||||
target}]（只列不同的字面量）, base_rows, target_rows, base_truncated, target_truncated, base_artifact,
|
||||
target_artifact, differences: [结构上差在哪], problem?（快照读不出来、被改过）}
|
||||
@@ -168,6 +196,8 @@ class _Query:
|
||||
#: 快照读不出来的原因：missing / tampered
|
||||
problem: str | None = None
|
||||
source: str | None = None
|
||||
#: 合并查询的各个输入 [{alias, node_id, artifact, …}]：快照里记的，快照读不出来时取 merge.end 的
|
||||
inputs: list[dict[str, Any]] | None = None
|
||||
|
||||
|
||||
def _snapshot(load: Callable[[str], Any], artifact: str) -> tuple[dict[str, Any] | None, str | None]:
|
||||
@@ -189,16 +219,20 @@ def _queries(side: Side) -> list[_Query]:
|
||||
seen: set[tuple[str, str]] = set()
|
||||
asked: dict[tuple[str, Any], str] = {} # (节点, call_id) → tool.start 里的 SQL
|
||||
merged: dict[str, str] = {} # 合并节点 → merge.end 里的 SQL
|
||||
merged_inputs: dict[str, Any] = {} # 合并节点 → merge.end 里的输入
|
||||
count: dict[str, int] = {}
|
||||
|
||||
def add(node_id: str, artifact: str, tool: Any, call_id: Any, sql: Any, source: Any) -> None:
|
||||
count[node_id] = count.get(node_id, 0) + 1
|
||||
snap, problem = _snapshot(side.load, artifact)
|
||||
own = snap.get("sql") if snap is not None and isinstance(snap.get("sql"), str) else None
|
||||
source = (snap or {}).get("source") or source
|
||||
inputs = (snap or {}).get("inputs") if snap is not None else merged_inputs.get(node_id)
|
||||
out.append(_Query(artifact=artifact, node_id=node_id, tool=str(tool) if tool else None,
|
||||
ordinal=count[node_id], call_id=str(call_id) if call_id else None,
|
||||
sql=own or (sql if isinstance(sql, str) and sql.strip() else None), snap=snap,
|
||||
problem=problem, source=(snap or {}).get("source") or source))
|
||||
problem=problem, source=source,
|
||||
inputs=inputs if source == MERGE_SOURCE and isinstance(inputs, list) else None))
|
||||
|
||||
for _, etype, node_id, data in side.events:
|
||||
if not node_id or not isinstance(data, dict) or data.get("role") == "verify":
|
||||
@@ -212,8 +246,10 @@ def _queries(side: Side) -> list[_Query]:
|
||||
seen.add((node_id, artifact))
|
||||
add(node_id, artifact, data.get("tool"), data.get("call_id"), asked.get((node_id, data.get("call_id"))),
|
||||
None)
|
||||
elif etype == EventType.MERGE_END and isinstance(data.get("sql"), str):
|
||||
merged[node_id] = data["sql"]
|
||||
elif etype == EventType.MERGE_END:
|
||||
if isinstance(data.get("sql"), str):
|
||||
merged[node_id] = data["sql"]
|
||||
merged_inputs[node_id] = data.get("inputs")
|
||||
elif etype == EventType.NODE_FINISHED:
|
||||
for entry in data.get("evidence") or []:
|
||||
if not (isinstance(entry, dict) and entry.get("kind") == "query" and entry.get("artifact")):
|
||||
@@ -232,12 +268,13 @@ def _queries(side: Side) -> list[_Query]:
|
||||
|
||||
|
||||
def _pair(base: list[_Query], target: list[_Query],
|
||||
text: Callable[[_Query], str | None]) -> list[tuple[_Query | None, _Query | None]]:
|
||||
norm: Callable[[_Query], NormalizedSql | None]) -> list[tuple[_Query | None, _Query | None]]:
|
||||
"""配对，一轮比一轮宽：
|
||||
1. 同一节点、同一工具、规范化 SQL 相同（只有字面量不同也算）：Agent 每次查询的先后可能不同，按「第几次」硬配
|
||||
会把换了先后的两条都配成「结构变了」；循环里同一个模板查几次的，按先后一一配上，和按第几次配一样
|
||||
2. 剩下的按「节点 + 第几次 + 工具」配：同一个位置上的查询换了写法，就是这条查询的结构变了
|
||||
3. 再剩下的跨节点按规范化 SQL 配：节点换了 ID，查的还是同一个模板
|
||||
1. 同一节点、同一工具、规范化 SQL 和字面量都相同：Agent 每次查询的先后可能不同，按「第几次」硬配会把换了先后的
|
||||
两条都配成「结构变了」；先配一字不差的，换了先后的东区、南区两条才各自配上自己，不会配成两条「参数不同」
|
||||
2. 同一节点、同一工具、规范化 SQL 相同（只有字面量不同）：循环里同一个模板查几次的，按先后一一配上
|
||||
3. 剩下的按「节点 + 第几次 + 工具」配：同一个位置上的查询换了写法，就是这条查询的结构变了
|
||||
4. 再剩下的跨节点按规范化 SQL 配（同样先配字面量也相同的）：节点换了 ID，查的还是同一个模板
|
||||
结果按这次运行的先后排，基准运行多出来的接在后面。"""
|
||||
matched: dict[int, int] = {}
|
||||
taken: set[int] = set()
|
||||
@@ -251,12 +288,20 @@ def _pair(base: list[_Query], target: list[_Query],
|
||||
matched[j] = i
|
||||
taken.add(i)
|
||||
|
||||
def same_sql(b: _Query, q: _Query) -> bool:
|
||||
key = text(q)
|
||||
return key is not None and text(b) == key
|
||||
def same_sql(b: _Query, q: _Query, *, literals: bool = False) -> bool:
|
||||
nq = norm(q)
|
||||
if nq is None:
|
||||
return False
|
||||
nb = norm(b)
|
||||
return nb is not None and nb.text == nq.text and (not literals or nb.literals == nq.literals)
|
||||
|
||||
link(lambda b, q: b.node_id == q.node_id and b.tool == q.tool and same_sql(b, q))
|
||||
def here(b: _Query, q: _Query) -> bool:
|
||||
return b.node_id == q.node_id and b.tool == q.tool
|
||||
|
||||
link(lambda b, q: here(b, q) and same_sql(b, q, literals=True))
|
||||
link(lambda b, q: here(b, q) and same_sql(b, q))
|
||||
link(lambda b, q: (b.node_id, b.ordinal, b.tool) == (q.node_id, q.ordinal, q.tool))
|
||||
link(lambda b, q: same_sql(b, q, literals=True))
|
||||
link(same_sql)
|
||||
pairs: list[tuple[_Query | None, _Query | None]] = [
|
||||
(base[matched[j]] if j in matched else None, q) for j, q in enumerate(target)]
|
||||
@@ -382,6 +427,44 @@ def _plain(d: Decimal, *, whole: bool) -> int | float | None:
|
||||
return value if math.isfinite(value) else None
|
||||
|
||||
|
||||
def _percent(base: Decimal, target: Decimal) -> float | None:
|
||||
"""相对基准的百分比，两位小数；基准为 0 没有百分比。基准极小（浮点误差那么小的数）、这次很大时，百分比的整数
|
||||
部分比十进制的默认精度（28 位）还长，量化到两位小数会抛 InvalidOperation：退回浮点算，浮点也放不下的给 None。"""
|
||||
if base == 0:
|
||||
return None
|
||||
ratio = (target - base) / abs(base) * 100
|
||||
try:
|
||||
return _plain(ratio.quantize(Decimal("0.01")), whole=False)
|
||||
except ArithmeticError:
|
||||
value = float(ratio)
|
||||
return round(value, 2) if math.isfinite(value) else None
|
||||
|
||||
|
||||
def _round_trips(d: Decimal) -> bool:
|
||||
"""这个十进制数换成浮点、再按最短写法写回来还是它自己（JSON 里交出去的就是浮点）。"""
|
||||
value = float(d)
|
||||
return math.isfinite(value) and Decimal(repr(value)) == d
|
||||
|
||||
|
||||
def _split_exact(delta: Decimal, caliber: Decimal) -> tuple[Decimal, Decimal, Decimal]:
|
||||
"""(总变化, 口径效应, 数据效应)，三个数转成浮点交出去以后仍逐位相加。
|
||||
|
||||
十进制里数据效应 = 总变化 − 口径效应是精确的，可转成浮点时三个数各舍入各的：没设小数位的指标（0.1 + 0.2 =
|
||||
0.30000000000000004 这种）交出去的两效应之和会和总变化差在最后一位。这时三个数按同一个精度量化（最多 15 位
|
||||
有效数字，浮点能原样写回的位数）：总变化和口径效应各自量化，数据效应取两者之差。设了小数位的值本来就写得回，
|
||||
原样交回。"""
|
||||
data = delta - caliber
|
||||
if all(_round_trips(x) for x in (delta, caliber, data)):
|
||||
return delta, caliber, data
|
||||
top = max(abs(delta), abs(caliber), abs(data))
|
||||
for digits in range(_FLOAT_DIGITS, _FLOAT_DIGITS - 3, -1):
|
||||
exp = Decimal(1).scaleb(top.adjusted() + 1 - digits)
|
||||
d, c = delta.quantize(exp), caliber.quantize(exp)
|
||||
if all(_round_trips(x) for x in (d, c, d - c)):
|
||||
return d, c, d - c
|
||||
return delta, caliber, data
|
||||
|
||||
|
||||
def _places(*metrics: dict[str, Any]) -> int | None:
|
||||
"""几个指标里最多的小数位；有一个没设的就不定小数位（按值本来的样子写）。"""
|
||||
found = [m.get("decimals") for m in metrics]
|
||||
@@ -483,6 +566,38 @@ def _key_label(key: str, node_type: Any) -> str:
|
||||
return "节点类型" if key == "type" else field_label(key, node_type)
|
||||
|
||||
|
||||
def _unset(value: Any) -> bool:
|
||||
"""节点上没写这一项：运行时回退到图级默认设置（NodeContext.cfg 同一个规则）。"""
|
||||
return value in (None, "")
|
||||
|
||||
|
||||
def _graph_sql(node: dict[str, Any] | None) -> tuple[str, str | None] | None:
|
||||
"""节点在图里写的 SQL 和它的数据源：调用工具节点是参数里的 SQL（数据源从工具名来），合并查询是合并 SQL。
|
||||
别的节点（Agent 自己写 SQL）返回 None。"""
|
||||
if node is None:
|
||||
return None
|
||||
config = _config(node)
|
||||
if node.get("type") == "tool":
|
||||
args = config.get("args")
|
||||
sql = args.get("sql") if isinstance(args, dict) else None
|
||||
tool = str(config.get("tool") or "")
|
||||
source = tool[len(QUERY_PREFIX):] if tool.startswith(QUERY_PREFIX) else None
|
||||
return (sql, source) if isinstance(sql, str) and sql.strip() else None
|
||||
if node.get("type") == "merge":
|
||||
sql = config.get("sql")
|
||||
return (sql, MERGE_SOURCE) if isinstance(sql, str) and sql.strip() else None
|
||||
return None
|
||||
|
||||
|
||||
def _definition(graph: dict[str, Any], card_id: str, metric_id: str) -> dict[str, Any] | None:
|
||||
"""图里口径卡节点上写的指标定义；指标定义来自别的工作流（口径卡来源）的不算，那里写的才是当时用的。"""
|
||||
card = _nodes(graph).get(card_id)
|
||||
if card is None or card.get("type") != "metrics" or _config(card).get("caliber_from") not in (None, "", {}):
|
||||
return None
|
||||
return next((d for d in _config(card).get("metrics") or []
|
||||
if isinstance(d, dict) and str(d.get("id")) == metric_id), None)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 输入的出处有没有变
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -523,11 +638,12 @@ class _Compare:
|
||||
self._norm: dict[tuple[str, str | None], NormalizedSql] = {}
|
||||
self._shapes: dict[tuple[int, int], _Shape] = {}
|
||||
self._ancestors: dict[str, set[str]] = {}
|
||||
self.base_nodes, self.target_nodes = _nodes(base.graph), _nodes(target.graph)
|
||||
self.base_queries, self.target_queries = _queries(base), _queries(target)
|
||||
self.pairs = _pair(self.base_queries, self.target_queries, self._text)
|
||||
self.pairs = _pair(self.base_queries, self.target_queries, self._norm_of)
|
||||
self.template, self.changed = self._template()
|
||||
self.titles = {**{k: self._title_of(n) for k, n in _nodes(base.graph).items()},
|
||||
**{k: self._title_of(n) for k, n in _nodes(target.graph).items()}}
|
||||
self.titles = {**{k: self._title_of(n) for k, n in self.base_nodes.items()},
|
||||
**{k: self._title_of(n) for k, n in self.target_nodes.items()}}
|
||||
self.parents: dict[str, set[str]] = {}
|
||||
for source, target_id, _ in _edges(base.graph) | _edges(target.graph):
|
||||
self.parents.setdefault(target_id, set()).add(source)
|
||||
@@ -576,8 +692,15 @@ class _Compare:
|
||||
self._norm[key] = normalize_sql(sql, dialect)
|
||||
return self._norm[key]
|
||||
|
||||
def _text(self, q: _Query) -> str | None:
|
||||
return self._normalize(q.sql, self._dialect(q.source)).text if q.sql else None
|
||||
def _norm_of(self, q: _Query) -> NormalizedSql | None:
|
||||
return self._normalize(q.sql, self._dialect(q.source)) if q.sql else None
|
||||
|
||||
def _same_graph_sql(self, a: tuple[str, str | None] | None, b: tuple[str, str | None] | None) -> bool:
|
||||
"""图里的两段 SQL 规范化以后一模一样（只差空白、大小写、注释）。"""
|
||||
if a is None or b is None:
|
||||
return False
|
||||
na, nb = self._normalize(a[0], self._dialect(a[1])), self._normalize(b[0], self._dialect(b[1]))
|
||||
return (na.text, na.literals) == (nb.text, nb.literals)
|
||||
|
||||
def shape(self, b: _Query | None, t: _Query | None) -> _Shape:
|
||||
if b is None:
|
||||
@@ -656,10 +779,71 @@ class _Compare:
|
||||
return _Change("params", f"{name}的 SQL 只有参数不同（{shown}{more}）")
|
||||
return _Change("same")
|
||||
|
||||
def _snapshot_change(self, b: _Query | None, t: _Query | None, seen: frozenset[str] = frozenset()) -> _Change:
|
||||
"""一对查询的取数变没变：SQL 和数据源（_query_change）；只差字面量时再看图里的 SQL 改没改过；合并查询再
|
||||
一路追到喂它的每一份上游快照。"""
|
||||
change = self._query_change(b, t)
|
||||
if b is None or t is None:
|
||||
return change
|
||||
found = [change]
|
||||
if change.kind == "params" and (edited := self._edited_sql(b, t)) is not None:
|
||||
found.append(edited)
|
||||
if b.source == MERGE_SOURCE and t.source == MERGE_SOURCE:
|
||||
found += self._merge_change(b, t, seen | {b.artifact, t.artifact})
|
||||
return _worst(found)
|
||||
|
||||
def _edited_sql(self, b: _Query, t: _Query) -> _Change | None:
|
||||
"""执行的 SQL 只差字面量时，图里写的 SQL 改没改过。定时运行代进去的日期不同，图里一字未改：是参数。图里写死的
|
||||
筛选值、系数改了（'paid' 改成 'refunded'、0.9 改成 0.8）：执行的 SQL 同样只差字面量，却是改了模板。规范化
|
||||
以后一模一样的（只改了空白、大小写）不算改过。Agent 自己写的 SQL 图里没有，返回 None。"""
|
||||
was, now = _graph_sql(self.base_nodes.get(b.node_id)), _graph_sql(self.target_nodes.get(t.node_id))
|
||||
if was is None or now is None or self._same_graph_sql(was, now):
|
||||
return None
|
||||
nb, nt = self._normalize(was[0], self._dialect(was[1])), self._normalize(now[0], self._dialect(now[1]))
|
||||
shown = [f"{x} → {y}" for x, y in zip(nb.literals, nt.literals) if x != y] if nb.text == nt.text else []
|
||||
detail = "（" + "，".join(shown[:2]) + (f" 等 {len(shown)} 处" if len(shown) > 2 else "") + "）" if shown else ""
|
||||
return _Change("sql", f"工作流里「{self.title(t.node_id)}」的 SQL 改过{detail}，不是运行时代入的参数")
|
||||
|
||||
def _merge_change(self, b: _Query, t: _Query, seen: frozenset[str]) -> list[_Change]:
|
||||
"""合并查询的每个输入（按别名配对两边）各自的取数变没变。合并 SQL 一字未改，喂它的取数 SQL 加了筛选，合并
|
||||
结果说的也已经不是同一回事。输入是另一个合并查询的结果时接着往上追。"""
|
||||
name = f"合并查询「{self.title(t.node_id)}」"
|
||||
before, after = _merge_inputs(b), _merge_inputs(t)
|
||||
if before is None or after is None:
|
||||
return [_Change("unknown", f"{name}没有记录它的输入，无法比较上游的取数")]
|
||||
out = []
|
||||
for alias in dict.fromkeys([*after, *before]):
|
||||
was, now = before.get(alias), after.get(alias)
|
||||
if was is None or now is None:
|
||||
out.append(_Change("sql", f"{name}的输入有变化：这次{'多了' if was is None else '少了'}「{alias}」"))
|
||||
continue
|
||||
if was in seen or now in seen:
|
||||
continue # 合并查询的输入只能是上游的快照，成不了环；防一手
|
||||
qb = next((q for q in self.base_queries if q.artifact == was), None)
|
||||
qt = next((q for q in self.target_queries if q.artifact == now), None)
|
||||
if qb is None or qt is None:
|
||||
out.append(_Change("unknown", f"{name}的输入「{alias}」落在的查询快照不在封存范围内，无法比较上游的取数"))
|
||||
continue
|
||||
change = self._snapshot_change(qb, qt, seen)
|
||||
out.append(_Change(change.kind, f"{name}的输入「{alias}」：{change.reason}") if change.reason else change)
|
||||
return out
|
||||
|
||||
def _node_change(self, node_id: str) -> _Change | None:
|
||||
"""一个节点自己查过的库在两次运行之间变没变；它两边都没查过库返回 None。"""
|
||||
own = [(b, t) for b, t in self.pairs if (b and b.node_id == node_id) or (t and t.node_id == node_id)]
|
||||
return _worst(self._query_change(b, t) for b, t in own) if own else None
|
||||
return _worst(self._snapshot_change(b, t) for b, t in own) if own else None
|
||||
|
||||
def _data_moved(self, scope: set[str]) -> bool:
|
||||
"""这几个节点查到的数据两次有没有不同：有一对查询的行（或列）不同就算有；快照读不出来、只在一边有的核对不了，
|
||||
也算有（不能说它们一样）。一条查询都没有的算没有。"""
|
||||
for b, t in self.pairs:
|
||||
if (t or b).node_id not in scope:
|
||||
continue
|
||||
if b is None or t is None or b.snap is None or t.snap is None:
|
||||
return True
|
||||
if b.snap.get("rows") != t.snap.get("rows") or b.snap.get("columns") != t.snap.get("columns"):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _upstream(self, node_id: str) -> set[str]:
|
||||
"""画布上 node_id 的全部上游（两次运行的图合起来，宁可多算）。"""
|
||||
@@ -674,56 +858,137 @@ class _Compare:
|
||||
self._ancestors[node_id] = seen
|
||||
return self._ancestors[node_id]
|
||||
|
||||
def provenance(self, path: str, b: Any, t: Any) -> _Change:
|
||||
"""一个输入的出处在两次运行之间变没变（b、t 是推演重算还原出来的输入）。"""
|
||||
def _type_of(self, node_id: str) -> str | None:
|
||||
node = self.target_nodes.get(node_id) or self.base_nodes.get(node_id)
|
||||
return str(node.get("type")) if node is not None else None
|
||||
|
||||
def _moved(self, node_id: str) -> str:
|
||||
"""一个影响取值的节点变化（self.changed）写成一句话。"""
|
||||
name = f"上游节点「{self.title(node_id)}」"
|
||||
return {"added": f"{name}是这次新增的", "removed": f"{name}这次被删除了", "edges": f"{name}的连线有变化",
|
||||
"defaults": f"工作流默认设置有变化，{name}没有单独设置这些项，随之改变"
|
||||
}.get(self.changed.get(node_id, ""), f"{name}的配置有变化")
|
||||
|
||||
def provenance(self, path: str, b: Any, t: Any, *, moved: bool = True) -> _Change:
|
||||
"""一个输入的出处在两次运行之间变没变（b、t 是推演重算还原出来的输入）。moved 是这个指标的值变没变：输入是
|
||||
容器、完整值又还原不出来，核对不了它变没变时，按它推断。"""
|
||||
if b is None or t is None:
|
||||
return _Change("unknown", f"输入「{path}」只在一边的口径卡里有记录")
|
||||
if b.snapshot or t.snapshot:
|
||||
# 落在查询快照上：两边各自的那份快照直接比 SQL，不管全局配对把它们配给了谁
|
||||
if not (b.snapshot and t.snapshot):
|
||||
return _Change("unknown", f"输入「{path}」只有一边对得上查询快照，无法比较取数")
|
||||
qb, qt = self._side_query(self.base_queries, b), self._side_query(self.target_queries, t)
|
||||
if qb is None or qt is None:
|
||||
return _Change("unknown", f"输入「{path}」落在的查询快照不在封存范围内的查询里，无法比较取数")
|
||||
change = self._query_change(qb, qt)
|
||||
agent = t.item.get("node_id")
|
||||
if t.item.get("via") == "agent_field" and agent and self.changed.get(str(agent)) \
|
||||
and change.kind in ("same", "params"):
|
||||
# Agent 按出处交来的字段：SQL 没变，Agent 自己改了配置（提示词、工具……）也可能换了取哪一格。
|
||||
# 调用工具节点的格子由口径卡的 cell() 定，节点配置变了要么是 SQL 变了（上面已经比过），要么不影响取值
|
||||
return _Change("template", f"Agent「{self.title(agent)}」的配置有变化，它按出处交来的字段可能换了取法")
|
||||
return change
|
||||
return self._linked(path, b, t)
|
||||
nb, nt = b.item.get("node_id"), t.item.get("node_id")
|
||||
if nb != nt:
|
||||
if not nb or not nt:
|
||||
side = "基准运行" if not nb else "这次运行"
|
||||
return _Change("unknown", f"{side}没有记录输入「{path}」由哪个节点写入，无法比较来源")
|
||||
return _Change("template", f"输入「{path}」的来源节点由「{self.title(nb)}」换成了「{self.title(nt)}」")
|
||||
same = _same_input(b, t)
|
||||
if _from_input(path, b.item):
|
||||
if recompute._same(b.item.get("value"), t.item.get("value")):
|
||||
return _Change("same")
|
||||
return _Change("params", f"运行输入「{path}」由「{_shown(b.item.get('value'))}」"
|
||||
f"改为「{_shown(t.item.get('value'))}」")
|
||||
if nb and (own := self._node_change(str(nb))) is not None:
|
||||
# 写它的节点自己查过库（Agent 没按出处交、调用工具节点的结果整份用）：它查的那几条就是喂它的查询
|
||||
if self.changed.get(str(nb)) and own.kind in ("same", "params"):
|
||||
return _Change("template", f"上游节点「{self.title(nb)}」的配置有变化")
|
||||
return self._run_input(path, b, t, same, moved=moved)
|
||||
if same is None:
|
||||
same = not moved
|
||||
writer = str(nb) if nb else None
|
||||
if writer and (self._type_of(writer) in _GENERATED
|
||||
or {b.item.get("via"), t.item.get("via")} & _GENERATED_VIA):
|
||||
return self._generated(path, writer, same)
|
||||
own = self._node_change(writer) if writer else None
|
||||
if writer and own is not None:
|
||||
# 写它的节点自己查过库（调用工具节点的结果整份用、合并查询……）：它查的那几条就是喂它的查询
|
||||
if writer in self.changed and own.kind in ("same", "params"):
|
||||
return _Change("template", self._moved(writer))
|
||||
if own.kind == "same" and not same and not self._data_moved({writer}):
|
||||
return _Change("unknown", f"输入「{path}」来自「{self.title(writer)}」，它查到的数据两次相同，"
|
||||
"交出的值却变了，无法判断变化来源")
|
||||
return own
|
||||
if nb and self.changed.get(str(nb)):
|
||||
return _Change("template", f"上游节点「{self.title(nb)}」的配置有变化")
|
||||
# 写它的节点不查库（代码、数据整形、模型调用……）：看画布上它的全部上游；写它的节点不知道就看整张图
|
||||
scope = self._upstream(str(nb)) if nb else set(self.titles)
|
||||
moved = [n for n in scope if self.changed.get(n)]
|
||||
changes = [self._query_change(b2, t2) for b2, t2 in self.pairs if (t2 or b2).node_id in scope]
|
||||
worst = _worst(changes)
|
||||
where = f"来自「{self.title(nb)}」" if nb else "不知道由哪个节点写入"
|
||||
if moved or worst.kind in ("sql", "template", "unknown"):
|
||||
if writer and writer in self.changed:
|
||||
return _Change("template", self._moved(writer))
|
||||
# 写它的节点不查库（代码、数据整形……）：看画布上它的全部上游；写它的节点不知道就看整张图
|
||||
scope = self._upstream(writer) if writer else set(self.titles)
|
||||
moved_nodes = [n for n in scope if n in self.changed]
|
||||
worst = _worst(self._query_change(b2, t2) for b2, t2 in self.pairs if (t2 or b2).node_id in scope)
|
||||
where = f"来自「{self.title(writer)}」" if writer else "不知道由哪个节点写入"
|
||||
if moved_nodes or worst.kind in ("sql", "template", "unknown"):
|
||||
return _Change("unknown", f"输入「{path}」{where}，对不上具体的查询；它的上游有变化，无法判断变化来源")
|
||||
if worst.kind == "same" and not same and not self._data_moved(scope):
|
||||
# 上游查到的数据两次一模一样，它交的值却变了：不是数据变的。上游只有运行输入变了的，算参数
|
||||
if (self.base.inputs or {}) != (self.target.inputs or {}) \
|
||||
and any(self._type_of(n) == "input" for n in scope):
|
||||
return _Change("params", f"输入「{path}」{where}，它的上游只有运行输入有变化")
|
||||
return _Change("unknown", f"输入「{path}」{where}，它上游的查询结果两次相同，交出的值却变了，"
|
||||
"无法判断变化来源")
|
||||
return worst
|
||||
|
||||
def _linked(self, path: str, b: Any, t: Any) -> _Change:
|
||||
"""落在查询快照上的输入：两边各自的那份快照直接比取数，不管全局配对把它们配给了谁。Agent 按出处交来的字段还要
|
||||
引用同一列、同一段行。"""
|
||||
if not (b.snapshot and t.snapshot):
|
||||
return _Change("unknown", f"输入「{path}」只有一边对得上查询快照，无法比较取数")
|
||||
qb, qt = self._side_query(self.base_queries, b), self._side_query(self.target_queries, t)
|
||||
if qb is None or qt is None:
|
||||
return _Change("unknown", f"输入「{path}」落在的查询快照不在封存范围内的查询里，无法比较取数")
|
||||
change = self._snapshot_change(qb, qt)
|
||||
if "agent_field" not in (b.item.get("via"), t.item.get("via")):
|
||||
# cell() 取哪一格由口径卡的表达式定（行列用到的变量是另外的输入，各比各的）
|
||||
return change
|
||||
agent = t.item.get("node_id") or b.item.get("node_id")
|
||||
if (cited := self._citation(agent, b, t, qb, qt)) is not None:
|
||||
change = _worst([change, cited])
|
||||
if agent and str(agent) in self.changed and change.kind in ("same", "params"):
|
||||
# SQL 没变、引用的也是同一处，Agent 自己的配置（提示词、工具、模型……）变了：字段的意思可能已经不同
|
||||
return _Change("template", f"{self._moved(str(agent))}，它按出处交来的字段可能换了取法")
|
||||
return change
|
||||
|
||||
def _citation(self, agent: Any, b: Any, t: Any, qb: _Query, qt: _Query) -> _Change | None:
|
||||
"""Agent 按出处交来的字段两次引用的是不是同一处：同一列（对象数组是同一组列）、同一行或同一段行。行段不同、但
|
||||
两边引用的都是整份结果（这次多了几行）的算同一处。不是同一处返回说明。"""
|
||||
name = f"Agent「{self.title(agent)}」" if agent else "Agent"
|
||||
lb, lt = _locator(b), _locator(t)
|
||||
cb, ct = lb.get("column") or lb.get("columns"), lt.get("column") or lt.get("columns")
|
||||
if cb != ct:
|
||||
return _Change("unknown", f"{name}引用的列由{_columns_text(cb)}改为{_columns_text(ct)}，取的不是同一个数")
|
||||
if (b.row, b.rows) == (t.row, t.rows) or (b.row is None and t.row is None and _whole(b, qb) and _whole(t, qt)):
|
||||
return None
|
||||
return _Change("unknown", f"{name}引用的行由{_rows_text(b)}改为{_rows_text(t)}，取的不是同一组数")
|
||||
|
||||
def _generated(self, path: str, writer: str, same: bool) -> _Change:
|
||||
"""模型、人交出的值（Agent 没按出处交、模型调用、人工审批……）：没有出处，值变了说不清是数据变了还是它本身的
|
||||
差异——同样的数据、同样的配置，模型两次交出的数也可能不同。它自己查的库结构变了、它的配置变了照常标出。"""
|
||||
own = self._node_change(writer)
|
||||
if own is not None and own.kind == "sql":
|
||||
return own
|
||||
if writer in self.changed:
|
||||
return _Change("template", self._moved(writer))
|
||||
if own is not None and own.kind == "unknown":
|
||||
return own
|
||||
if same:
|
||||
return own or _Change("same")
|
||||
kind = TYPE_LABEL.get(self._type_of(writer) or "", "节点")
|
||||
return _Change("unknown", f"输入「{path}」由节点「{self.title(writer)}」（{kind}）直接给出，没有对应的查询快照，"
|
||||
"无法判断变化是否来自数据")
|
||||
|
||||
def _run_input(self, path: str, b: Any, t: Any, same: bool | None, *, moved: bool) -> _Change:
|
||||
"""运行输入（input.*、入口节点写进 vars 的）：变了就是参数变化。列表、对象按还原出来的完整值比，卡上只记了
|
||||
「[2 项]」这样的大概，长度不变、内容变了的不能当没变。"""
|
||||
if same is None:
|
||||
if moved:
|
||||
return _Change("unknown", f"运行输入「{path}」是列表或对象，当时没有保存完整值，无法核对是否有变化")
|
||||
return _Change("same")
|
||||
if same:
|
||||
return _Change("same")
|
||||
if b.shape == "container" or t.shape == "container":
|
||||
return _Change("params", f"运行输入「{path}」的内容有变化")
|
||||
return _Change("params", f"运行输入「{path}」由「{_shown(b.item.get('value'))}」"
|
||||
f"改为「{_shown(t.item.get('value'))}」")
|
||||
|
||||
# ---------------------------------------------------------------- 模板
|
||||
|
||||
def _template(self) -> tuple[dict[str, Any], dict[str, str]]:
|
||||
"""两次运行的图快照比较：新增、删除的节点，配置有变化的节点和配置项名（不展示值）。changed 是影响取值的
|
||||
变化（新增、删除、类型或配置变了；只改名称的不算），判断输入的出处时用。"""
|
||||
before, after = _nodes(self.base.graph), _nodes(self.target.graph)
|
||||
"""两次运行的图快照比较：新增、删除的节点，配置有变化的节点和配置项名（不展示值）。
|
||||
|
||||
changed 是影响取值的变化，判断输入的出处时用：节点 → added / removed / config / edges / defaults。只改名称的
|
||||
不算；图里的 SQL 只改了空白、大小写的不算（规范化以后一模一样）。图级的变化也落到节点上：连线有增删的，终点
|
||||
节点交出的值可能变了（接进来的东西、执行的条件变了）；默认设置改了的，没有自行设置这几项的模型类节点跟着变。"""
|
||||
before, after = self.base_nodes, self.target_nodes
|
||||
nodes: list[dict[str, Any]] = []
|
||||
changed: dict[str, str] = {}
|
||||
|
||||
@@ -737,17 +1002,32 @@ class _Compare:
|
||||
nodes.append(entry(node_id, node, "added", []))
|
||||
changed[node_id] = "added"
|
||||
continue
|
||||
keys, renamed_only = _config_keys(was, node)
|
||||
keys, _ = _config_keys(was, node)
|
||||
if keys:
|
||||
nodes.append(entry(node_id, node, "config", keys))
|
||||
if not renamed_only:
|
||||
sql_key = _SQL_KEY.get(str(node.get("type"))) if was.get("type") == node.get("type") else None
|
||||
if any(k != "label" and not (k == sql_key and self._same_graph_sql(_graph_sql(was), _graph_sql(node)))
|
||||
for k in keys):
|
||||
changed[node_id] = "config"
|
||||
for node_id, node in before.items():
|
||||
if node_id not in after:
|
||||
nodes.append(entry(node_id, node, "removed", []))
|
||||
changed[node_id] = "removed"
|
||||
same_edges = _edges(self.base.graph) == _edges(self.target.graph)
|
||||
same_defaults = (self.base.graph.get("defaults") or {}) == (self.target.graph.get("defaults") or {})
|
||||
edges_before, edges_after = _edges(self.base.graph), _edges(self.target.graph)
|
||||
for _, target_id, _ in edges_before ^ edges_after:
|
||||
if target_id in before and target_id in after:
|
||||
changed.setdefault(target_id, "edges")
|
||||
defaults_before = self.base.graph.get("defaults") or {}
|
||||
defaults_after = self.target.graph.get("defaults") or {}
|
||||
differing = [k for k in dict.fromkeys([*defaults_before, *defaults_after])
|
||||
if defaults_before.get(k) != defaults_after.get(k) and k not in _QUIET_DEFAULTS]
|
||||
for node_id, node in after.items():
|
||||
was = before.get(node_id)
|
||||
if differing and was is not None and str(node.get("type")) in _MODEL_TYPES and any(
|
||||
_unset(_config(node).get(k)) and _unset(_config(was).get(k)) for k in differing):
|
||||
changed.setdefault(node_id, "defaults")
|
||||
same_edges = edges_before == edges_after
|
||||
same_defaults = defaults_before == defaults_after
|
||||
if not same_edges:
|
||||
self.notes.append(EDGES_CHANGED)
|
||||
if not same_defaults:
|
||||
@@ -828,33 +1108,46 @@ class _Compare:
|
||||
out.update(classification=cls, reasons=[f"{side}的口径卡「{self.title(card_id)}」里没有这个指标"])
|
||||
return out
|
||||
assert bm is not None and tm is not None and b is not None and t is not None
|
||||
caliber = _caliber_reasons(bm, tm, b.content or {}, t.content or {})
|
||||
defs = (_definition(self.base.graph, card_id, mid), _definition(self.target.graph, card_id, mid))
|
||||
caliber = _caliber_reasons(bm, tm, b.content or {}, t.content or {}, defs)
|
||||
if shown := _display_change(bm, tm, defs):
|
||||
self.notes.append(f"口径卡「{self.title(card_id)}」的指标「{out['name']}」{shown}，只影响显示，不算口径变化")
|
||||
self._delta(out, bm, tm)
|
||||
lineage = isinstance(bm.get("inputs"), list) and isinstance(tm.get("inputs"), list)
|
||||
moved = not recompute._same(bm.get("value"), tm.get("value"))
|
||||
changes: dict[str, _Change] = {}
|
||||
if lineage and bp is not None and tp is not None:
|
||||
changes = {p: self.provenance(p, bp.inputs.get(p), tp.inputs.get(p), moved=moved)
|
||||
for p in dict.fromkeys([*tp.inputs, *bp.inputs])}
|
||||
if caliber:
|
||||
out.update(classification=CALIBER_AND_DATA, reasons=caliber)
|
||||
out["decomposition"], out["decomposition_note"] = self._decompose(bm, tm, b, t, bp, tp)
|
||||
# 口径优先；喂它的取数也变了、说不清的，接在口径的说明后面照样写出来，不因为口径变了就不提。只在一边才有
|
||||
# 的输入是改表达式带来的，口径的说明已经说了
|
||||
both = set(bp.inputs) & set(tp.inputs) if bp is not None and tp is not None else set()
|
||||
also = [c.reason for p, c in changes.items() if p in both and c.kind in ("sql", "template", "unknown")
|
||||
and c.reason]
|
||||
out.update(classification=CALIBER_AND_DATA, reasons=list(dict.fromkeys([*caliber, *also])))
|
||||
self._decompose(out, bm, tm, b, t, bp, tp)
|
||||
elif not lineage:
|
||||
quiet = self.template["same_graph"] and all(self.shape(x, y).structure == "same" for x, y in self.pairs)
|
||||
same = recompute._same(bm.get("value"), tm.get("value"))
|
||||
out.update(classification=UNCHANGED if quiet and same else INCOMPARABLE,
|
||||
reasons=[] if quiet and same else [NO_LINEAGE])
|
||||
out.update(classification=UNCHANGED if quiet and not moved else INCOMPARABLE,
|
||||
reasons=[] if quiet and not moved else [NO_LINEAGE])
|
||||
else:
|
||||
paths = list(dict.fromkeys([*tp.inputs, *bp.inputs]))
|
||||
changes = [self.provenance(p, bp.inputs.get(p), tp.inputs.get(p)) for p in paths]
|
||||
worst = _worst(changes)
|
||||
reasons = list(dict.fromkeys(c.reason for c in changes if c.kind == worst.kind and c.reason))
|
||||
worst = _worst(changes.values())
|
||||
reasons = list(dict.fromkeys(c.reason for c in changes.values() if c.kind == worst.kind and c.reason))
|
||||
if worst.kind in ("sql", "template"):
|
||||
out.update(classification=SQL_CHANGED, reasons=reasons, decomposition_note=SQL_NOT_SPLIT)
|
||||
elif worst.kind == "unknown":
|
||||
out.update(classification=INCOMPARABLE, reasons=reasons)
|
||||
elif worst.kind == "params":
|
||||
out.update(classification=PARAMS_ONLY, reasons=reasons)
|
||||
elif recompute._same(bm.get("value"), tm.get("value")):
|
||||
elif not moved:
|
||||
out.update(classification=UNCHANGED)
|
||||
elif all(inp.snapshot for inp in tp.inputs.values()) and tp.inputs:
|
||||
out.update(classification=DATA_ONLY, reasons=[DATA_FROM_QUERIES])
|
||||
else:
|
||||
linked = all(inp.snapshot for inp in tp.inputs.values()) and bool(tp.inputs)
|
||||
out.update(classification=DATA_ONLY, reasons=[DATA_FROM_QUERIES if linked else DATA_FROM_UPSTREAM])
|
||||
# 「模板没有变化」只在整张图都没变时说；图有别的改动、都不在它的上游的，只说它的上游没变
|
||||
out.update(classification=DATA_ONLY, reasons=[
|
||||
DATA_FROM_UPSTREAM if self.template["same_graph"] else DATA_FROM_UPSTREAM_SCOPED])
|
||||
for side, metric in (("基准运行", bm), ("这次运行", tm)):
|
||||
if metric.get("status") == "missing_input":
|
||||
out["reasons"].append(f"{side}这个指标缺少输入，没有值")
|
||||
@@ -872,11 +1165,21 @@ class _Compare:
|
||||
whole = all(isinstance(m.get("value"), int) for m in (bm, tm))
|
||||
delta = _plain(t - b, whole=whole)
|
||||
meta = {**tm, "decimals": _places(bm, tm)}
|
||||
out.update(delta=delta, delta_rendered=_signed(meta, delta) if delta != 0 else None,
|
||||
delta_pct=_plain(round((t - b) / abs(b) * 100, 2), whole=False) if b != 0 else None)
|
||||
out.update(delta=delta, delta_rendered=_signed(meta, delta) if delta != 0 else None, delta_pct=_percent(b, t))
|
||||
|
||||
def _decompose(self, bm: dict[str, Any], tm: dict[str, Any], b: _Card, t: _Card, bp: Any,
|
||||
tp: Any) -> tuple[dict[str, Any] | None, str | None]:
|
||||
def _decompose(self, out: dict[str, Any], bm: dict[str, Any], tm: dict[str, Any], b: _Card, t: _Card, bp: Any,
|
||||
tp: Any) -> None:
|
||||
"""严格拆分，写进 out 的 decomposition（拆不了的原因写进 decomposition_note）。为了两效应之和在 JSON 里也逐位
|
||||
等于总变化，总变化可能按拆分的精度重写一遍（_split_exact）。"""
|
||||
split, note = self._split(bm, tm, b, t, bp, tp)
|
||||
out["decomposition"], out["decomposition_note"] = split, note
|
||||
if split is not None and "delta" in split:
|
||||
delta = split.pop("delta")
|
||||
meta = {**tm, "decimals": _places(bm, tm)}
|
||||
out.update(delta=delta, delta_rendered=_signed(meta, delta) if delta != 0 else None)
|
||||
|
||||
def _split(self, bm: dict[str, Any], tm: dict[str, Any], b: _Card, t: _Card, bp: Any,
|
||||
tp: Any) -> tuple[dict[str, Any] | None, str | None]:
|
||||
"""严格拆分：(拆分, None) 或 (None, 为什么没法拆)。"""
|
||||
old, new = _exact(bm.get("value")), _exact(tm.get("value"))
|
||||
if old is None or new is None:
|
||||
@@ -899,8 +1202,10 @@ class _Compare:
|
||||
return None, SPLIT_MASKED
|
||||
for path in leaves:
|
||||
change = self.provenance(path, bp.inputs.get(path), t.inputs[path])
|
||||
if change.kind in ("sql", "template", "unknown"):
|
||||
if change.kind in ("sql", "template"):
|
||||
return None, f"旧口径用到的输入取数也有变化，无法严格拆分：{change.reason}"
|
||||
if change.kind == "unknown":
|
||||
return None, f"无法确认旧口径用到的输入只是数据有变化，无法严格拆分：{change.reason}"
|
||||
# f_old(x_new)：旧表达式配这次的输入，按推演重算的规则还原、求值，取整按旧口径卡的小数位
|
||||
synthetic = {"id": bm.get("id"), "expression": bm.get("expression"), "decimals": bm.get("decimals"),
|
||||
"inputs": [inp.item for inp in used], "substituted": None, "value": None}
|
||||
@@ -918,10 +1223,14 @@ class _Compare:
|
||||
if middle is None:
|
||||
return None, "按旧口径在这次的数据上算出的不是有限的数值，无法严格拆分"
|
||||
whole = all(isinstance(v, int) and not isinstance(v, bool) for v in (bm.get("value"), tm.get("value"), value))
|
||||
caliber, data = _plain(new - middle, whole=whole), _plain(middle - old, whole=whole)
|
||||
total, caliber_d, data_d = _split_exact(new - old, new - middle)
|
||||
caliber, data = _plain(caliber_d, whole=whole), _plain(data_d, whole=whole)
|
||||
meta = {**tm, "decimals": _places(bm, tm)}
|
||||
return {"caliber_effect": caliber, "data_effect": data, "caliber_rendered": _signed(meta, caliber),
|
||||
"data_rendered": _signed(meta, data)}, None
|
||||
split = {"caliber_effect": caliber, "data_effect": data, "caliber_rendered": _signed(meta, caliber),
|
||||
"data_rendered": _signed(meta, data)}
|
||||
if total != new - old:
|
||||
split["delta"] = _plain(total, whole=whole)
|
||||
return split, None
|
||||
|
||||
def _inputs(self, bp: Any, tp: Any) -> list[dict[str, Any]]:
|
||||
"""两边的输入按路径对齐：值（遮罩列、核对不了遮没遮的不给），各自落在的查询快照。"""
|
||||
@@ -961,13 +1270,15 @@ class _Compare:
|
||||
return next((r for r in found if r["rows"]), found[0] if found else None)
|
||||
|
||||
def _attribute(self, b: Any, t: Any) -> dict[str, Any] | None:
|
||||
"""一个输入的维度归因：{columns, measure, rows, total_delta, note}；归不了因的照实写在 note 里（rows 为空）。
|
||||
快照里没有可以当维度的列（只有一行汇总这种）返回 None：没什么可归的。"""
|
||||
"""一个输入的维度归因：{path, base_artifact, target_artifact, columns, measure, rows, total_delta, note}；
|
||||
归不了因的照实写在 note 里（rows 为空）。快照里没有可以当维度的列（只有一行汇总这种）返回 None：没什么可归的。"""
|
||||
measure = _locator(b).get("column")
|
||||
if not isinstance(measure, str) or _locator(t).get("column") != measure:
|
||||
return None
|
||||
sb, st = self._snap(self.base_queries, b), self._snap(self.target_queries, t)
|
||||
empty = {"columns": [], "measure": measure, "rows": [], "total_delta": None}
|
||||
# 归因用的是哪个输入、两边各自哪份快照：一个指标有几个落在快照上的输入时分得清
|
||||
trace = {"path": t.path, "base_artifact": b.snapshot, "target_artifact": t.snapshot}
|
||||
empty = {**trace, "columns": [], "measure": measure, "rows": [], "total_delta": None}
|
||||
if sb is None or st is None:
|
||||
return {**empty, "note": UNREADABLE_SNAPSHOT}
|
||||
for side, snap in (("基准运行", sb), ("这次运行", st)):
|
||||
@@ -1016,7 +1327,7 @@ class _Compare:
|
||||
rows.sort(key=lambda r: -r[0]) # 稳定排序：变化一样大的按先出现的
|
||||
if len(rows) > TOP_KEYS:
|
||||
notes.append(MORE_KEYS.format(count=len(rows), top=TOP_KEYS))
|
||||
return {"columns": dims, "measure": measure, "rows": [r for _, r in rows[:TOP_KEYS]],
|
||||
return {**trace, "columns": dims, "measure": measure, "rows": [r for _, r in rows[:TOP_KEYS]],
|
||||
"total_delta": _plain(total, whole=False), "note": "；".join(notes) or None}
|
||||
|
||||
def _snap(self, queries: list[_Query], inp: Any) -> dict[str, Any] | None:
|
||||
@@ -1043,26 +1354,50 @@ def _rendered(metric: dict[str, Any] | None) -> str | None:
|
||||
return rendered if isinstance(rendered, str) else recompute._render(metric, metric.get("value"))
|
||||
|
||||
|
||||
def _caliber_reasons(bm: dict[str, Any], tm: dict[str, Any], bc: dict[str, Any], tc: dict[str, Any]) -> list[str]:
|
||||
"""口径变了的地方，一处一句；没变返回空列表。"""
|
||||
def _setting(metric: dict[str, Any], definition: dict[str, Any] | None, key: str) -> tuple[bool, Any]:
|
||||
"""指标的小数位（decimals）或显示格式（format）：(知不知道, 值)。卡上记了的按卡上的；升级前的卡没记这两项（当时
|
||||
按定义里的小数位取过整），按那次运行图里的指标定义（和当时一样规整）；定义也找不到的不知道，不拿来比。"""
|
||||
if key in metric:
|
||||
value = metric[key]
|
||||
return True, (value or DEFAULT_FORMAT) if key == "format" else value
|
||||
if definition is None:
|
||||
return False, None
|
||||
display, problem = _display_of(definition)
|
||||
return (False, None) if problem else (True, display[key])
|
||||
|
||||
|
||||
def _caliber_reasons(bm: dict[str, Any], tm: dict[str, Any], bc: dict[str, Any], tc: dict[str, Any],
|
||||
defs: tuple[dict[str, Any] | None, dict[str, Any] | None] = (None, None)) -> list[str]:
|
||||
"""口径变了的地方，一处一句；没变返回空列表。defs 是两次运行图里这个指标的定义（老卡没记小数位、格式时用）。"""
|
||||
out = []
|
||||
if _expression(bm) != _expression(tm):
|
||||
out.append(f"表达式由「{_shown(bm.get('expression'))}」改为「{_shown(tm.get('expression'))}」")
|
||||
before, after = str(bc.get("caliber_version") or ""), str(tc.get("caliber_version") or "")
|
||||
if before != after:
|
||||
out.append(f"口径版本由 {before or '未设置'} 改为 {after or '未设置'}")
|
||||
if bm.get("decimals") != tm.get("decimals"):
|
||||
(kb, db), (kt, dt) = _setting(bm, defs[0], "decimals"), _setting(tm, defs[1], "decimals")
|
||||
if kb and kt and db != dt:
|
||||
def places(d: Any) -> str:
|
||||
return "未设置" if d is None else str(d)
|
||||
out.append(f"小数位由 {places(bm.get('decimals'))} 改为 {places(tm.get('decimals'))}")
|
||||
out.append(f"小数位由 {places(db)} 改为 {places(dt)}")
|
||||
if (bm.get("unit") or "") != (tm.get("unit") or ""):
|
||||
out.append(f"单位由「{bm.get('unit') or '无'}」改为「{tm.get('unit') or '无'}」")
|
||||
fb, ft = bm.get("format") or DEFAULT_FORMAT, tm.get("format") or DEFAULT_FORMAT
|
||||
if fb != ft:
|
||||
(kb, fb), (kt, ft) = _setting(bm, defs[0], "format"), _setting(tm, defs[1], "format")
|
||||
if kb and kt and (fb == _RATIO) != (ft == _RATIO):
|
||||
# 比率（0.087 显示成 8.7%）和普通数值的含义不同；千分位、不分组只是写法，见 _display_change
|
||||
out.append(f"显示格式由「{_FORMAT_LABEL.get(fb, fb)}」改为「{_FORMAT_LABEL.get(ft, ft)}」")
|
||||
return out
|
||||
|
||||
|
||||
def _display_change(bm: dict[str, Any], tm: dict[str, Any],
|
||||
defs: tuple[dict[str, Any] | None, dict[str, Any] | None]) -> str | None:
|
||||
"""只影响显示的格式变化（千分位和不分组互换）：一句说明；没有返回 None。"""
|
||||
(kb, fb), (kt, ft) = _setting(bm, defs[0], "format"), _setting(tm, defs[1], "format")
|
||||
if not (kb and kt) or fb == ft or _RATIO in (fb, ft):
|
||||
return None
|
||||
return f"显示格式由「{_FORMAT_LABEL.get(fb, fb)}」改为「{_FORMAT_LABEL.get(ft, ft)}」"
|
||||
|
||||
|
||||
def _input_value(inp: Any) -> Any:
|
||||
"""输入在对比里给人看的值：遮罩列、落在读不出来的快照上的不给；容器写卡上记的大概。"""
|
||||
if inp is None or inp.masked or (inp.unreadable and inp.shape == "scalar"):
|
||||
@@ -1077,6 +1412,45 @@ def _locator(inp: Any) -> dict[str, Any]:
|
||||
return locator if isinstance(locator, dict) else {}
|
||||
|
||||
|
||||
def _same_input(b: Any, t: Any) -> bool | None:
|
||||
"""一个输入两边的值相不相同。容器按还原出来的完整值比（卡上只记了「[2 项]」这样的大概，长度相同、内容不同的
|
||||
不能当一样）；完整值还原不出来时，大概不同的就是不同，大概相同的说不清，返回 None。"""
|
||||
if b.shape == "container" or t.shape == "container":
|
||||
if b.shape == t.shape == "container" and b.problem is None and t.problem is None:
|
||||
return recompute._same(b.value, t.value)
|
||||
return None if recompute._same(b.item.get("value"), t.item.get("value")) else False
|
||||
return recompute._same(b.item.get("value"), t.item.get("value"))
|
||||
|
||||
|
||||
def _whole(inp: Any, q: _Query) -> bool:
|
||||
"""这个输入引用的是整份查询结果：数组字段的行段从第一行到最后一行，或者容器本身就是整份结果。"""
|
||||
if inp.table:
|
||||
return True
|
||||
return inp.rows is not None and q.snap is not None and inp.rows == (0, len(q.snap["rows"]) - 1)
|
||||
|
||||
|
||||
def _rows_text(inp: Any) -> str:
|
||||
if inp.row is not None:
|
||||
return f"第 {inp.row + 1} 行"
|
||||
if inp.rows is not None:
|
||||
return f"第 {inp.rows[0] + 1} 至 {inp.rows[1] + 1} 行"
|
||||
return "整份结果"
|
||||
|
||||
|
||||
def _columns_text(columns: Any) -> str:
|
||||
if isinstance(columns, dict):
|
||||
return "「" + "、".join(str(c) for c in columns.values()) + "」"
|
||||
return f"「{columns}」" if columns else "（未记录）"
|
||||
|
||||
|
||||
def _merge_inputs(q: _Query) -> dict[str, str] | None:
|
||||
"""合并查询的输入：别名 → 上游快照。没记录的返回 None。"""
|
||||
if q.inputs is None:
|
||||
return None
|
||||
return {str(i["alias"]): str(i["artifact"]) for i in q.inputs
|
||||
if isinstance(i, dict) and i.get("alias") and i.get("artifact")}
|
||||
|
||||
|
||||
def _records(snap: dict[str, Any], inp: Any) -> list[tuple[int, dict[str, Any]]] | None:
|
||||
"""归因用到的行：(行号, {列: 值})。数组字段只用它那一段行，单格用整份结果。"""
|
||||
columns = [str(c) for c in snap.get("columns") or []]
|
||||
|
||||
+49
-13
@@ -12,6 +12,7 @@ from app.core.artifact_store import canonical_json, content_hash
|
||||
from app.engine import compare, recompute
|
||||
from app.engine.evidence import render_number
|
||||
from app.engine.expressions import parse_expression, substitute
|
||||
from app.engine.merge_query import MERGE_SOURCE
|
||||
|
||||
TOOL = "db_query__shop"
|
||||
WEEK36 = ("2026-08-31", "2026-09-06")
|
||||
@@ -110,15 +111,15 @@ class Fake:
|
||||
return artifact
|
||||
|
||||
def agent(self, node_id, sql, columns=COLUMNS, rows=BASE_ROWS, *, types=TYPES, truncated=False, masks=None,
|
||||
verify=()):
|
||||
"""Agent 节点自己查一次库，按出处交回数组字段 amounts（快照 gmv 列的全部行）。verify 是同一节点里核查类的
|
||||
查询（role: verify，不进台账）。"""
|
||||
verify=(), column="gmv", span=None):
|
||||
"""Agent 节点自己查一次库，按出处交回数组字段 amounts（默认是快照 gmv 列的全部行；column、span 改引用的列和
|
||||
行段）。verify 是同一节点里核查类的查询（role: verify，不进台账）。"""
|
||||
artifact = self.snapshot(sql, columns, rows, truncated=truncated, types=types, masks=masks)
|
||||
last = len(rows) - 1
|
||||
gmv = columns.index("gmv")
|
||||
data = {"amounts": [r[gmv] for r in rows]}
|
||||
evidence = {"amounts": {"ref": f"Q1.r0-{last}.gmv", "call": "Q1", "artifact": artifact,
|
||||
"locator": {"rows": [0, last], "column": "gmv"}, "status": "verified"}}
|
||||
first, last = span or (0, len(rows) - 1)
|
||||
at = columns.index(column)
|
||||
data = {"amounts": [r[at] for r in rows[first:last + 1]]}
|
||||
evidence = {"amounts": {"ref": f"Q1.r{first}-{last}.{column}", "call": "Q1", "artifact": artifact,
|
||||
"locator": {"rows": [first, last], "column": column}, "status": "verified"}}
|
||||
output = self.put({"text": "按区域列出了金额", "data": data, "data_evidence": evidence})
|
||||
self.add("node.started", node_id, {"node_type": "agent"})
|
||||
self.add("tool.start", node_id, {"tool": TOOL, "args": {"sql": sql}, "call_id": "c1"})
|
||||
@@ -130,6 +131,40 @@ class Fake:
|
||||
self.entry(node_id, artifact, columns, rows, call_id="c1", truncated=truncated)]})
|
||||
return artifact
|
||||
|
||||
def plain_agent(self, node_id, sqls, data, columns=COLUMNS, rows=BASE_ROWS, *, types=TYPES):
|
||||
"""没开按出处核对的 Agent：查几次库，交回的 data 没有出处（data_evidence），口径卡只知道是它写的。"""
|
||||
self.add("node.started", node_id, {"node_type": "agent"})
|
||||
entries = []
|
||||
for i, sql in enumerate(sqls):
|
||||
artifact = self.snapshot(sql, columns, rows, types=types)
|
||||
self.add("tool.start", node_id, {"tool": TOOL, "args": {"sql": sql}, "call_id": f"c{i}"})
|
||||
self.add("tool.end", node_id, {"tool": TOOL, "call_id": f"c{i}", "duration_ms": 5,
|
||||
"query_artifact": artifact})
|
||||
entries.append(self.entry(node_id, artifact, columns, rows, call_id=f"c{i}"))
|
||||
output = self.put({"text": "按区域分析了金额", "data": data})
|
||||
self.add("node.finished", node_id, {"duration_ms": 9, "attempt": 1, "artifact": output, "evidence": entries})
|
||||
|
||||
def merge(self, node_id, sql, inputs, columns, rows, *, types=None):
|
||||
"""合并查询：几份上游快照按别名落进内存库，再执行一条合并 SQL。快照里记着各个输入（别名、节点、快照），
|
||||
事件照合并节点的形状写（merge.end 加 node.finished 台账里的 query 条目，没有 tool.end）。
|
||||
inputs 是 [(别名, 上游节点, 上游快照)]。"""
|
||||
snap = {"columns": columns, "rows": rows, "row_count": len(rows), "truncated": False, "sql": sql,
|
||||
"source": MERGE_SOURCE, "sources": [self.source],
|
||||
"inputs": [{"alias": alias, "node_id": upstream, "artifact": artifact,
|
||||
"rows": len(self.store[artifact]["rows"]), "columns": self.store[artifact]["columns"],
|
||||
"source": self.store[artifact].get("source")} for alias, upstream, artifact in inputs],
|
||||
"warnings": [], **({"column_types": types} if types else {})}
|
||||
artifact = self.put(snap)
|
||||
self.queries.append(artifact)
|
||||
output = self.put({**snap, "artifact": artifact})
|
||||
view = [{k: v for k, v in item.items() if k != "columns"} for item in snap["inputs"]]
|
||||
self.add("node.started", node_id, {"node_type": "merge"})
|
||||
self.add("merge.end", node_id, {"inputs": view, "sql": sql, "rows": len(rows), "query_artifact": artifact})
|
||||
self.add("node.finished", node_id, {"duration_ms": 2, "attempt": 1, "artifact": output, "evidence": [
|
||||
{"kind": "query", "node_id": node_id, "exec": 1, "artifact": artifact, "source": MERGE_SOURCE,
|
||||
"columns": columns, "rows": len(rows), "truncated": False}]})
|
||||
return artifact
|
||||
|
||||
def code(self, node_id, value, var):
|
||||
"""沙箱代码节点算出一个数写进变量：它不查库，对不上具体的查询。"""
|
||||
output = self.put({var: value})
|
||||
@@ -169,11 +204,12 @@ def cell_in(row, column, value, artifact, node_id="fetch"):
|
||||
"eid": f"ev:cell:{artifact[:8]}:{row}:{column}", "status": "verified"}
|
||||
|
||||
|
||||
def amounts_in(artifact, rows):
|
||||
last = len(rows) - 1
|
||||
return {"path": "vars.kpi.amounts", "value": f"[{len(rows)} 项]", "node_id": "bot", "via": "agent_field",
|
||||
"field": "amounts", "ref": f"Q1.r0-{last}.gmv", "artifact": artifact,
|
||||
"locator": {"rows": [0, last], "column": "gmv"}, "eid": f"ev:rows:{artifact[:8]}", "status": "verified"}
|
||||
def amounts_in(artifact, rows, *, column="gmv", span=None):
|
||||
first, last = span or (0, len(rows) - 1)
|
||||
return {"path": "vars.kpi.amounts", "value": f"[{last - first + 1} 项]", "node_id": "bot", "via": "agent_field",
|
||||
"field": "amounts", "ref": f"Q1.r{first}-{last}.{column}", "artifact": artifact,
|
||||
"locator": {"rows": [first, last], "column": column}, "eid": f"ev:rows:{artifact[:8]}:{first}:{column}",
|
||||
"status": "verified"}
|
||||
|
||||
|
||||
def scalar_in(path, value, node_id, via):
|
||||
|
||||
@@ -149,12 +149,14 @@ async def test_unsealed_runs_are_refused(client):
|
||||
seal=False)
|
||||
body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}", 409)
|
||||
assert body["code"] == "compare_unsealed" and body["detail"].startswith("这次运行")
|
||||
# 已经结束、只是没有封存清单的：说没有封存清单，不说「运行结束后才能对比」
|
||||
assert "没有封存清单" in body["detail"] and "尚未结束" not in body["detail"]
|
||||
body = await get(client, f"/api/runs/{base_id}/compare?base={target_id}", 409)
|
||||
assert body["code"] == "compare_unsealed" and body["detail"].startswith("作为基准的运行")
|
||||
# 封存过、又被接着运行（还在运行）的：封存范围之后还会变，同样不比
|
||||
running = await persist(week("r3", NEXT_ROWS, source=SOURCE), workflow, status="running")
|
||||
body = await get(client, f"/api/runs/{running}/compare?base={base_id}", 409)
|
||||
assert body["code"] == "compare_unsealed"
|
||||
assert body["code"] == "compare_unsealed" and "尚未结束" in body["detail"]
|
||||
|
||||
|
||||
async def test_a_broken_seal_is_refused(client):
|
||||
@@ -233,12 +235,15 @@ async def test_masked_columns_never_leave_the_server(client):
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def bare(workflow_id, finished, *, status="succeeded", run_class="formal", schedule_id=None, sealed=True) -> str:
|
||||
"""只为挑基准用的运行记录：挑基准只看运行记录上的几列。"""
|
||||
async def bare(workflow_id, finished, *, status="succeeded", run_class="formal", schedule_id=None, sealed=True,
|
||||
resumed=False) -> str:
|
||||
"""只为挑基准用的运行记录：挑基准只看运行记录上的几列。resumed：结束、封存过又被接着运行（状态已回到运行中，
|
||||
结束时间和封存还没清掉），只能靠状态把它挡在外面。"""
|
||||
done = status in ("succeeded", "failed", "cancelled") or resumed
|
||||
async with SessionLocal() as session:
|
||||
run = Run(workflow_id=workflow_id, workflow_name="区域周报", status=status, run_class=run_class,
|
||||
schedule_id=schedule_id, started_at=finished - timedelta(minutes=2),
|
||||
finished_at=finished if status in ("succeeded", "failed", "cancelled") else None,
|
||||
finished_at=finished if done else None,
|
||||
manifest_hash="ab" * 32 if sealed else None, manifest_seq=1 if sealed else None)
|
||||
session.add(run)
|
||||
await session.commit()
|
||||
@@ -271,6 +276,7 @@ async def test_the_default_base_skips_what_cannot_be_compared(client):
|
||||
good = await bare(workflow, T0)
|
||||
await bare(workflow, T0 + day, sealed=False) # 没封存
|
||||
await bare(workflow, T0 + 2 * day, status="running") # 还在运行
|
||||
await bare(workflow, T0 + 2 * day, status="running", resumed=True) # 结束过、又被接着运行
|
||||
await bare(workflow, T0 + 3 * day, run_class="exploratory") # 别的运行类别
|
||||
await bare(await new_workflow("另一个周报"), T0 + 4 * day) # 别的工作流
|
||||
current = await bare(workflow, T0 + 5 * day)
|
||||
|
||||
@@ -9,9 +9,12 @@ SQL，Agent 节点 bot 自己再查一次、把各区域的金额作为数组字
|
||||
为 1；对不上时占比为空并说明；截断的快照不归因，照实写
|
||||
- 遮罩列的值绝不出现在结果里；用到遮罩列的指标不拆分
|
||||
- 模板对比：新增、删除的节点，配置有变化的节点和配置项名
|
||||
- 输入的出处：合并查询追到它的上游查询；Agent 按出处交的字段还要引用同一列、同一段行；没按出处交的数、上游数据
|
||||
没变数却变了的，说不清变化从哪来（incomparable）；图里写死的字面量改了、默认设置和连线变了都算模板变化
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
from decimal import Decimal
|
||||
|
||||
@@ -19,13 +22,16 @@ from app.engine import compare
|
||||
from app.engine.merge_query import MERGE_SOURCE
|
||||
from tests.fixtures.compare_runs import (
|
||||
BASE_ROWS,
|
||||
COLUMNS,
|
||||
EAST,
|
||||
NEXT_ROWS,
|
||||
REFUND,
|
||||
TOOL,
|
||||
WEEK37,
|
||||
Fake,
|
||||
amounts_in,
|
||||
cell_in,
|
||||
chain,
|
||||
default_metrics,
|
||||
metric,
|
||||
node,
|
||||
@@ -167,14 +173,14 @@ def test_the_split_adds_up_exactly_with_awkward_decimals():
|
||||
return make
|
||||
|
||||
base_rows = [["东区", 1000.1, 0.3], *BASE_ROWS[1:]]
|
||||
next_rows = [["东区", 1000.7, 0.2], *NEXT_ROWS[1:]]
|
||||
next_rows = [["东区", 1000.75, 0.2], *NEXT_ROWS[1:]]
|
||||
base = week("r1", base_rows, metrics=build(f"{EAST} * 0.1 - {REFUND}", 2))
|
||||
target = week("r2", next_rows, metrics=build(f"{EAST} * 0.3 + {REFUND}", 3), version="v2")
|
||||
ratio = by_id(run_compare(base, target))["ratio"]
|
||||
split = ratio["decomposition"]
|
||||
assert split is not None, ratio["decomposition_note"]
|
||||
# f_old(x_new) 按旧口径的小数位取整：round(1000.7 * 0.1 - 0.2, 2) = 99.87
|
||||
assert exact(split["data_effect"]) == exact(99.87) - exact(ratio["base"])
|
||||
# f_old(x_new) 按旧口径的小数位取整：1000.75 * 0.1 - 0.2 = 99.875，取 2 位是 99.88（按新口径的 3 位是 99.875）
|
||||
assert exact(split["data_effect"]) == exact(99.88) - exact(ratio["base"])
|
||||
assert exact(split["caliber_effect"]) + exact(split["data_effect"]) == exact(ratio["delta"])
|
||||
assert exact(ratio["delta"]) == exact(ratio["target"]) - exact(ratio["base"])
|
||||
|
||||
@@ -204,6 +210,20 @@ def test_no_split_when_the_sql_changed_too():
|
||||
net = by_id(run_compare(base, target))["net"]
|
||||
assert net["classification"] == "caliber_and_data" and net["decomposition"] is None
|
||||
assert "SQL" in net["decomposition_note"]
|
||||
# 两条不同的查询的行不拿来按维度配对
|
||||
assert net["attribution"] is None
|
||||
|
||||
|
||||
def test_a_caliber_change_still_names_the_sql_change():
|
||||
"""口径变了、SQL 结构也变了、旧口径的叶子这次又没记录：分类按口径，SQL 的变化照样写进说明。"""
|
||||
base = week("r1", metrics=net_metrics(f"{EAST} - {REFUND}"), version="v1")
|
||||
paid = region_sql(extra=" AND status = 'paid'")
|
||||
target = week("r2", NEXT_ROWS, sql=paid, metrics=net_metrics(EAST), version="v2")
|
||||
net = by_id(run_compare(base, target))["net"]
|
||||
assert net["classification"] == "caliber_and_data" and net["decomposition"] is None
|
||||
assert any("区域取数" in r and "SQL" in r for r in net["reasons"]), net["reasons"]
|
||||
# 退款那一格只在旧口径里用到：改表达式带来的，不再另说一遍「只在一边有记录」
|
||||
assert not any("只在一边" in r for r in net["reasons"]), net["reasons"]
|
||||
|
||||
|
||||
def test_a_unit_change_is_not_subtracted():
|
||||
@@ -217,15 +237,102 @@ def test_a_unit_change_is_not_subtracted():
|
||||
assert "单位" in east["decomposition_note"]
|
||||
|
||||
|
||||
def test_display_settings_that_change_the_value_are_caliber_changes():
|
||||
def build(**kw):
|
||||
def make(fetch, bot, rows, bot_rows):
|
||||
return [metric("east", EAST, rows[0][1], [cell_in(0, "gmv", rows[0][1], fetch)], **kw)]
|
||||
return make
|
||||
|
||||
def east(before, after):
|
||||
return by_id(run_compare(week("r1", metrics=build(**before)), week("r2", NEXT_ROWS, metrics=build(**after))))
|
||||
|
||||
# 小数位改变取整
|
||||
found = east({"decimals": 1}, {"decimals": 0})
|
||||
assert found["east"]["classification"] == "caliber_and_data" and any("小数位" in r for r in found["east"]["reasons"])
|
||||
# 比率与否改变数值的含义
|
||||
found = east({"unit": "", "fmt": "plain"}, {"unit": "", "fmt": "percent_of_ratio"})
|
||||
assert found["east"]["classification"] == "caliber_and_data" and found["east"]["delta"] is None
|
||||
# 千分位和不分组只影响显示：分类照数据算，变化写进说明
|
||||
result = run_compare(week("r1", metrics=build(fmt="thousands")), week("r2", NEXT_ROWS, metrics=build(fmt="plain")))
|
||||
assert by_id(result)["east"]["classification"] == "data_only"
|
||||
assert any("显示格式" in n for n in result["notes"])
|
||||
|
||||
|
||||
def test_an_old_card_without_display_settings_is_not_a_caliber_change():
|
||||
"""升级前封存的运行：卡上的指标没记小数位、显示格式和输入（当时按定义里的小数位取过整）。"""
|
||||
def old(run_id, value, graph=None):
|
||||
run = Fake(run_id, graph=graph)
|
||||
run.tool("fetch", region_sql())
|
||||
run.agent("bot", region_sql())
|
||||
run.card([{"id": "east", "name": "东区金额", "unit": "元", "value": value, "expression": EAST}])
|
||||
return run
|
||||
|
||||
def new(run_id, value):
|
||||
run = Fake(run_id)
|
||||
fetch = run.tool("fetch", region_sql())
|
||||
run.agent("bot", region_sql())
|
||||
run.card([metric("east", EAST, value, [cell_in(0, "gmv", value, fetch)], name="东区金额")])
|
||||
return run
|
||||
|
||||
east = by_id(run_compare(old("r1", 1200.5), new("r2", 1200.5)))["east"]
|
||||
assert (east["classification"], east["reasons"]) == ("unchanged", [])
|
||||
east = by_id(run_compare(old("r1", 1200.5), new("r2", 1500.5)))["east"]
|
||||
assert east["classification"] == "incomparable" and east["reasons"] == [compare.NO_LINEAGE]
|
||||
# 那次运行的图里写着定义：按定义里的小数位核对
|
||||
graph = weekly_graph()
|
||||
graph["nodes"][3]["data"]["config"]["metrics"] = [{"id": "east", "expression": EAST, "decimals": 2}]
|
||||
east = by_id(run_compare(old("r1", 1200.5, graph), new("r2", 1200.5)))["east"]
|
||||
assert east["classification"] == "caliber_and_data" and any("小数位由 2 改为 1" in r for r in east["reasons"])
|
||||
|
||||
|
||||
def test_the_split_adds_up_in_json_without_fixed_decimals():
|
||||
"""没设小数位的指标：三个数各自转成浮点也要逐位相加（接口按 JSON 交回的就是浮点）。"""
|
||||
def build(expression):
|
||||
def make(fetch, bot, rows, bot_rows):
|
||||
gmv, refund = rows[0][1], rows[0][2]
|
||||
inputs = [cell_in(0, "gmv", gmv, fetch), cell_in(0, "refund", refund, fetch)]
|
||||
value = eval(expression.replace(EAST, repr(gmv)).replace(REFUND, repr(refund)))
|
||||
return [metric("m", expression, value, inputs, decimals=None, unit="")]
|
||||
return make
|
||||
|
||||
base_rows = [["东区", 0.1, 0.2], *BASE_ROWS[1:]]
|
||||
next_rows = [["东区", 0.7, 0.1], *NEXT_ROWS[1:]]
|
||||
base = week("r1", base_rows, metrics=build(f"{EAST} + {REFUND}")) # 0.30000000000000004
|
||||
target = week("r2", next_rows, metrics=build(f"{EAST} * 3 + {REFUND}"), version="v2") # 2.1999999999999997
|
||||
m = by_id(run_compare(base, target))["m"]
|
||||
split = m["decomposition"]
|
||||
assert split is not None, m["decomposition_note"]
|
||||
assert exact(split["caliber_effect"]) + exact(split["data_effect"]) == exact(m["delta"])
|
||||
assert abs(m["delta"] - (m["target"] - m["base"])) < 1e-12
|
||||
|
||||
|
||||
def test_a_tiny_base_does_not_break_the_percentage():
|
||||
"""基准是浮点误差那么小的数、这次很大：百分比写得出来（或者为空），对比不能整个失败。"""
|
||||
def build(value):
|
||||
def make(fetch, bot, rows, bot_rows):
|
||||
m = metric("diff", EAST, 1.0, [cell_in(0, "gmv", rows[0][1], fetch)], decimals=None)
|
||||
m["value"], m["rendered"] = value, "—"
|
||||
return [m]
|
||||
return make
|
||||
diff = by_id(run_compare(week("r1", metrics=build(0.30000000000000004 - 0.3)),
|
||||
week("r2", NEXT_ROWS, metrics=build(1e8))))["diff"]
|
||||
assert diff["delta"] == 1e8 - (0.30000000000000004 - 0.3)
|
||||
assert diff["delta_pct"] is not None and diff["delta_pct"] > 1e25
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 维度归因
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_attribution_pairs_keys_and_shares_add_up_to_one():
|
||||
total = by_id(run_compare(week("r1"), week("r2", NEXT_ROWS)))["total"]
|
||||
base, target = week("r1"), week("r2", NEXT_ROWS)
|
||||
total = by_id(run_compare(base, target))["total"]
|
||||
attribution = total["attribution"]
|
||||
assert attribution["columns"] == ["region"] and attribution["measure"] == "gmv"
|
||||
# 归因用的是哪个输入、两边各自哪份快照
|
||||
assert (attribution["path"], attribution["base_artifact"], attribution["target_artifact"]) == (
|
||||
"vars.kpi.amounts", base.queries[1], target.queries[1])
|
||||
assert attribution["total_delta"] == 200.0
|
||||
rows = attribution["rows"]
|
||||
assert [(r["key"], r["base"], r["target"], r["delta"]) for r in rows] == [
|
||||
@@ -265,6 +372,49 @@ def test_no_share_when_the_measure_does_not_add_up_to_the_input():
|
||||
assert all(r["share"] is None for r in attribution["rows"]) and attribution["note"]
|
||||
|
||||
|
||||
def test_no_share_when_only_one_side_adds_up():
|
||||
"""基准运行里东区就是全部（各区域之和等于它），这次南区也有了：只有一边对得上，不给占比。"""
|
||||
base_rows = [["东区", 100.0, 0.0], ["南区", 0.0, 0.0], ["西区", 0.0, 0.0]]
|
||||
next_rows = [["东区", 150.0, 0.0], ["南区", 20.0, 0.0], ["西区", 0.0, 0.0]]
|
||||
east = by_id(run_compare(week("r1", base_rows), week("r2", next_rows)))["east"]
|
||||
attribution = east["attribution"]
|
||||
assert east["classification"] == "data_only" and len(attribution["rows"]) == 3
|
||||
assert all(r["share"] is None for r in attribution["rows"]) and compare.NO_SHARE in attribution["note"]
|
||||
|
||||
|
||||
def test_a_total_row_summed_in_floating_point_still_adds_up():
|
||||
"""合计行是库里按浮点加出来的（0.1 + 0.2 + 0.3 = 0.6000000000000001）：差在最后几位也算相等，照给占比。"""
|
||||
base_rows = [["东区", 0.1, 0.0], ["南区", 0.2, 0.0], ["西区", 0.3, 0.0], ["合计", 0.1 + 0.2 + 0.3, 0.0]]
|
||||
next_rows = [["东区", 0.4, 0.0], ["南区", 0.2, 0.0], ["西区", 0.3, 0.0], ["合计", 0.4 + 0.2 + 0.3, 0.0]]
|
||||
|
||||
def build(fetch, bot, rows, bot_rows):
|
||||
return [metric("all", "cell(nodes.fetch, 3, 'gmv')", rows[3][1], [cell_in(3, "gmv", rows[3][1], fetch)],
|
||||
decimals=None)]
|
||||
found = by_id(run_compare(week("r1", base_rows, metrics=build), week("r2", next_rows, metrics=build)))["all"]
|
||||
rows = found["attribution"]["rows"]
|
||||
assert [r["key"] for r in rows] == [["东区"], ["南区"], ["西区"]]
|
||||
assert abs(sum(r["share"] for r in rows) - 1) < 1e-9
|
||||
|
||||
|
||||
def test_date_typed_and_constant_columns_are_not_dimensions():
|
||||
"""日期类型的列（两边有相同的日子也不算）和两边都只有一个值的列（币种）不当维度。"""
|
||||
columns = ["day", "region", "currency", "gmv"]
|
||||
types = {"day": "date", "region": "text", "currency": "text", "gmv": "number"}
|
||||
|
||||
def make(run_id, rows):
|
||||
run = Fake(run_id)
|
||||
sql = "SELECT day, region, currency, SUM(amount) AS gmv FROM orders GROUP BY day, region, currency"
|
||||
bot = run.agent("bot", sql, columns, rows, types=types)
|
||||
run.card([metric("total", "sum(vars.kpi.amounts)", sum(r[3] for r in rows), [amounts_in(bot, rows)])])
|
||||
return run
|
||||
|
||||
base_rows = [["2026-09-01", "东区", "CNY", 10.0], ["2026-09-02", "南区", "CNY", 20.0]]
|
||||
next_rows = [["2026-09-02", "东区", "CNY", 15.0], ["2026-09-03", "南区", "CNY", 18.0]]
|
||||
attribution = by_id(run_compare(make("r1", base_rows), make("r2", next_rows)))["total"]["attribution"]
|
||||
assert attribution["columns"] == ["region"]
|
||||
assert [(r["key"], r["delta"]) for r in attribution["rows"]] == [(["东区"], 5.0), (["南区"], -2.0)]
|
||||
|
||||
|
||||
def test_a_total_row_is_left_out_of_its_own_breakdown():
|
||||
"""合计行和各区域在同一份结果里：指标取合计那一格，按其余各行归因，占比之和为 1。"""
|
||||
base_rows = [*BASE_ROWS, ["合计", 2600.5, 150.0]]
|
||||
@@ -290,6 +440,16 @@ def test_a_truncated_snapshot_is_not_attributed():
|
||||
assert result_queries(run_compare(base, target), "bot")["target_truncated"] is True
|
||||
|
||||
|
||||
def test_a_truncated_base_snapshot_is_not_attributed_either():
|
||||
base = Fake("r1")
|
||||
fetch = base.tool("fetch", region_sql())
|
||||
bot = base.agent("bot", region_sql(), truncated=True)
|
||||
base.card(default_metrics(fetch, bot, BASE_ROWS, BASE_ROWS))
|
||||
total = by_id(run_compare(base, week("r2", NEXT_ROWS)))["total"]
|
||||
assert total["classification"] == "data_only"
|
||||
assert total["attribution"]["rows"] == [] and "基准运行" in total["attribution"]["note"]
|
||||
|
||||
|
||||
def result_queries(result, node_id):
|
||||
return next(q for q in result["queries"] if q["node_id"] == node_id)
|
||||
|
||||
@@ -412,10 +572,49 @@ def test_an_input_from_a_code_node_follows_its_upstream():
|
||||
# 上游什么都没变：变化来自数据
|
||||
calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12)))["calc"]
|
||||
assert calc["classification"] == "data_only" and calc["delta"] == 4
|
||||
assert calc["reasons"] == [compare.DATA_FROM_UPSTREAM]
|
||||
# 上游的 SQL 结构变了：说不清它变没变，不能报成数据变化
|
||||
calc = by_id(run_compare(make("r1", BASE_ROWS, 10),
|
||||
make("r2", NEXT_ROWS, 12, sql=region_sql(extra=" AND status = 'paid'"))))["calc"]
|
||||
assert calc["classification"] == "incomparable" and any("调整" in r for r in calc["reasons"])
|
||||
# 上游查到的数据两次一模一样，它交的数却变了：变化不来自数据
|
||||
calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", BASE_ROWS, 12)))["calc"]
|
||||
assert calc["classification"] == "incomparable" and any("调整" in r for r in calc["reasons"])
|
||||
assert compare.DATA_FROM_UPSTREAM not in calc["reasons"]
|
||||
|
||||
|
||||
def test_a_value_from_a_querying_node_that_moved_without_its_data_is_incomparable():
|
||||
"""调用工具节点整份交出的结果（不是 cell() 取的格）：它查到的行两次一样，交出的值（这里是查询耗时）却变了，
|
||||
变化不来自数据；查到的行也变了的，照常是数据变化。"""
|
||||
def make(run_id, elapsed, rows=BASE_ROWS):
|
||||
run = Fake(run_id)
|
||||
run.tool("fetch", region_sql(), rows=rows)
|
||||
run.agent("bot", region_sql(), rows=rows)
|
||||
run.card([metric("cost", "vars.res.elapsed_ms", elapsed,
|
||||
[scalar_in("vars.res.elapsed_ms", elapsed, "fetch", "tool")], unit="毫秒", decimals=None)])
|
||||
return run
|
||||
cost = by_id(run_compare(make("r1", 3), make("r2", 5)))["cost"]
|
||||
assert cost["classification"] == "incomparable" and any("区域取数" in r for r in cost["reasons"])
|
||||
cost = by_id(run_compare(make("r1", 3), make("r2", 5, NEXT_ROWS)))["cost"]
|
||||
assert cost["classification"] == "data_only"
|
||||
|
||||
|
||||
def test_data_from_upstream_is_only_claimed_when_the_graph_is_the_same():
|
||||
"""改了它下游的节点（成果字段）：它的上游没变，变化照样来自上游数据，但不能说「模板没有变化」。"""
|
||||
def make(run_id, rows, value, graph):
|
||||
run = Fake(run_id, graph=graph)
|
||||
run.tool("fetch", region_sql(), rows=rows)
|
||||
run.agent("bot", region_sql(), rows=rows)
|
||||
run.code("calc", value, "adj")
|
||||
run.card(code_metrics(value)(None, None, rows, rows))
|
||||
return run
|
||||
|
||||
base_graph = weekly_graph(between=(node("calc", "code", "调整", code="print(1)", assign_to="adj"),))
|
||||
target_graph = copy.deepcopy(base_graph)
|
||||
target_graph["nodes"][-1]["data"]["config"] = {"fields": [{"name": "周报", "value": "{{ nodes.write.text }}"}]}
|
||||
calc = by_id(run_compare(make("r1", BASE_ROWS, 10, base_graph), make("r2", NEXT_ROWS, 12, target_graph)))["calc"]
|
||||
assert calc["classification"] == "data_only"
|
||||
assert calc["reasons"] == [compare.DATA_FROM_UPSTREAM_SCOPED]
|
||||
|
||||
|
||||
def test_the_code_node_itself_changed_is_a_template_change():
|
||||
@@ -430,6 +629,133 @@ def test_the_code_node_itself_changed_is_a_template_change():
|
||||
assert calc["classification"] == "sql_changed" and any("调整" in r for r in calc["reasons"])
|
||||
|
||||
|
||||
def test_rewired_edges_are_a_template_change():
|
||||
"""调整节点改由取数节点接进来（各节点的配置都没动）：它交的数变了算模板变化，不能说变化来自上游数据。"""
|
||||
calc_node = node("calc", "code", "调整", code="result = 1", assign_to="adj")
|
||||
base_graph = weekly_graph(between=(calc_node,))
|
||||
target_graph = copy.deepcopy(base_graph)
|
||||
target_graph["edges"] = [e for e in target_graph["edges"] if (e["source"], e["target"]) != ("bot", "calc")]
|
||||
target_graph["edges"].append({"source": "fetch", "target": "calc"})
|
||||
|
||||
def make(run_id, graph, value):
|
||||
run = Fake(run_id, graph=graph)
|
||||
run.tool("fetch", region_sql())
|
||||
run.agent("bot", region_sql())
|
||||
run.code("calc", value, "adj")
|
||||
run.card(code_metrics(value)(None, None, BASE_ROWS, BASE_ROWS))
|
||||
return run
|
||||
result = run_compare(make("r1", base_graph, 10), make("r2", target_graph, 12))
|
||||
calc = by_id(result)["calc"]
|
||||
assert calc["classification"] == "sql_changed" and any("连线" in r for r in calc["reasons"]), calc["reasons"]
|
||||
assert compare.EDGES_CHANGED in result["notes"]
|
||||
|
||||
|
||||
def uncited(run_id, share, *, graph=None, sqls=None):
|
||||
"""Agent 没开按出处核对，口径卡用了它交的占比 vars.kpi.share；东区金额照旧取取数节点的那一格。"""
|
||||
run = Fake(run_id, graph=graph)
|
||||
fetch = run.tool("fetch", region_sql())
|
||||
run.plain_agent("bot", sqls or [region_sql()], {"kpi": {"share": share}})
|
||||
run.card([metric("east", EAST, BASE_ROWS[0][1], [cell_in(0, "gmv", BASE_ROWS[0][1], fetch)], name="东区金额"),
|
||||
metric("share", "vars.kpi.share * 100", share * 100, [scalar_in("vars.kpi.share", share, "bot", "agent")],
|
||||
unit="%", fmt="plain")])
|
||||
return run
|
||||
|
||||
|
||||
def test_a_number_the_agent_gave_without_citing_is_incomparable():
|
||||
"""Agent 没按出处交的数：两次查到的数据一样，数却变了（模型本身的差异），不能说是数据变化。"""
|
||||
result = run_compare(uncited("r1", 0.46), uncited("r2", 0.52))
|
||||
share = by_id(result)["share"]
|
||||
assert share["classification"] == "incomparable" and share["attribution"] is None
|
||||
assert any("区域分析" in r and "查询快照" in r for r in share["reasons"]), share["reasons"]
|
||||
assert by_id(result)["east"]["classification"] == "unchanged"
|
||||
# 交的数一样：照常没变
|
||||
assert by_id(run_compare(uncited("r1", 0.46), uncited("r2", 0.46)))["share"]["classification"] == "unchanged"
|
||||
|
||||
|
||||
def test_an_extra_query_by_the_writer_is_a_sql_change():
|
||||
extra = "SELECT COUNT(*) AS n FROM refunds"
|
||||
share = by_id(run_compare(uncited("r1", 0.46), uncited("r2", 0.52, sqls=[region_sql(), extra])))["share"]
|
||||
assert share["classification"] == "sql_changed" and any("多了一次查询" in r for r in share["reasons"])
|
||||
|
||||
|
||||
def test_a_changed_default_model_is_a_template_change_for_agents_that_follow_it():
|
||||
base_graph = weekly_graph()
|
||||
base_graph["defaults"] = {"provider": "anthropic", "model": "model-a"}
|
||||
target_graph = copy.deepcopy(base_graph)
|
||||
target_graph["defaults"]["model"] = "model-b"
|
||||
result = run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=target_graph))
|
||||
share = by_id(result)["share"]
|
||||
assert share["classification"] == "sql_changed" and any("默认设置" in r for r in share["reasons"]), share["reasons"]
|
||||
assert result["template"]["same_graph"] is False and compare.DEFAULTS_CHANGED in result["notes"]
|
||||
# 取数节点不用模型：东区金额不受影响
|
||||
assert by_id(result)["east"]["classification"] == "unchanged"
|
||||
# Agent 自己指定了模型：默认设置变了它不跟着变
|
||||
for graph in (base_graph, target_graph):
|
||||
graph["nodes"][2]["data"]["config"]["model"] = "model-x"
|
||||
share = by_id(run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=target_graph)))["share"]
|
||||
assert share["classification"] == "incomparable"
|
||||
|
||||
|
||||
def cited(run_id, rows=BASE_ROWS, *, column="gmv", span=None):
|
||||
"""Agent 按出处交回数组字段：引用快照的哪一列、哪一段行由调用方定。"""
|
||||
run = Fake(run_id)
|
||||
run.tool("fetch", region_sql(), rows=rows)
|
||||
bot = run.agent("bot", region_sql(), rows=rows, column=column, span=span)
|
||||
first, last = span or (0, len(rows) - 1)
|
||||
values = [r[COLUMNS.index(column)] for r in rows[first:last + 1]]
|
||||
run.card([metric("total", "sum(vars.kpi.amounts)", sum(values),
|
||||
[amounts_in(bot, rows, column=column, span=span)])])
|
||||
return run
|
||||
|
||||
|
||||
def test_an_agent_citing_another_column_is_not_a_data_change():
|
||||
"""SQL 一字不差、数据一模一样，Agent 这次引用的是退款列：取的不是同一个数。"""
|
||||
total = by_id(run_compare(cited("r1"), cited("r2", column="refund")))["total"]
|
||||
assert total["classification"] == "incomparable" and total["attribution"] is None
|
||||
assert any("gmv" in r and "refund" in r for r in total["reasons"]), total["reasons"]
|
||||
|
||||
|
||||
def test_an_agent_citing_other_rows_is_not_a_data_change():
|
||||
"""Agent 这次只引用了后两行：东区的数还在快照里，不能把它算成「东区的数据没了」。"""
|
||||
total = by_id(run_compare(cited("r1"), cited("r2", span=(1, 2))))["total"]
|
||||
assert total["classification"] == "incomparable" and total["attribution"] is None
|
||||
assert any("行" in r for r in total["reasons"]), total["reasons"]
|
||||
# 两边引用的都是整份结果（这次多了一个区域）：照常是数据变化
|
||||
total = by_id(run_compare(cited("r1"), cited("r2", [*NEXT_ROWS, ["北区", 90.0, 0.0]])))["total"]
|
||||
assert total["classification"] == "data_only" and total["attribution"]["rows"]
|
||||
|
||||
|
||||
def test_an_unknown_writer_on_one_side_is_named_plainly():
|
||||
"""一边记不出 vars.adj 是哪个节点写的：说清是哪一边没记录，不写「None」。"""
|
||||
def make(run_id, writer, value):
|
||||
run = Fake(run_id, graph=weekly_graph(between=(node("calc", "code", "调整", code="x", assign_to="adj"),)))
|
||||
run.tool("fetch", region_sql())
|
||||
run.agent("bot", region_sql())
|
||||
run.code("calc", value, "adj")
|
||||
item = scalar_in("vars.adj", value, writer, "code" if writer else "state")
|
||||
run.card([metric("calc", "vars.adj * 2", value * 2, [item])])
|
||||
return run
|
||||
calc = by_id(run_compare(make("r1", None, 10), make("r2", "calc", 12)))["calc"]
|
||||
assert calc["classification"] == "incomparable"
|
||||
assert not any("None" in r for r in calc["reasons"]) and any("基准运行" in r for r in calc["reasons"])
|
||||
|
||||
|
||||
def test_a_list_run_input_is_compared_by_its_content():
|
||||
"""运行输入是列表：卡上只记了「[2 项]」，按入口节点的产出还原出完整值再比。"""
|
||||
def make(run_id, targets):
|
||||
run = Fake(run_id, inputs={"targets": targets})
|
||||
run.add("node.started", "start", {"node_type": "input"})
|
||||
run.add("node.finished", "start", {"artifact": run.put({"targets": targets})})
|
||||
item = {"path": "input.targets", "value": f"[{len(targets)} 项]", "node_id": "start", "via": "input",
|
||||
"field": "targets", "status": "ok"}
|
||||
run.card([metric("goal", "sum(input.targets)", float(sum(targets)), [item])])
|
||||
return run
|
||||
goal = by_id(run_compare(make("r1", [100.0, 200.0]), make("r2", [100.0, 500.0])))["goal"]
|
||||
assert goal["classification"] == "params_only" and any("targets" in r for r in goal["reasons"]), goal["reasons"]
|
||||
goal = by_id(run_compare(make("r1", [100.0, 200.0]), make("r2", [100.0, 200.0])))["goal"]
|
||||
assert goal["classification"] == "unchanged"
|
||||
|
||||
|
||||
def test_a_run_input_change_is_a_parameter_change():
|
||||
def make(run_id, goal):
|
||||
run = Fake(run_id, inputs={"goal": goal})
|
||||
@@ -534,6 +860,107 @@ def test_an_agent_asking_in_another_order_is_paired_by_sql():
|
||||
assert pairs == [("params_only", by_day[:20]), ("params_only", by_region[:20])]
|
||||
|
||||
|
||||
def test_swapped_queries_with_the_same_literals_pair_up_exactly():
|
||||
"""Agent 两次都查了东区和南区，只是先后互换：先配字面量也相同的，两条都没变；下游的数照常是数据变化。"""
|
||||
def region(name):
|
||||
return f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{name}'"
|
||||
|
||||
def make(run_id, regions, value, gmv):
|
||||
run = Fake(run_id, graph=weekly_graph(between=(node("calc", "code", "调整", code="x", assign_to="adj"),)))
|
||||
run.plain_agent("bot", [region(r) for r in regions], {"text": "ok"}, ["gmv"], [[gmv]], types={"gmv": "number"})
|
||||
run.code("calc", value, "adj")
|
||||
run.card(code_metrics(value)(None, None, None, None))
|
||||
return run
|
||||
|
||||
result = run_compare(make("r1", ["东区", "南区"], 10, 1.0), make("r2", ["南区", "东区"], 12, 2.0))
|
||||
assert [(q["structure"], q["literals"]) for q in result["queries"]] == [("same", []), ("same", [])]
|
||||
assert by_id(result)["calc"]["classification"] == "data_only"
|
||||
|
||||
|
||||
def test_a_switched_data_source_is_a_structure_change():
|
||||
"""SQL 一字不差、换了数据源：查的不是同一份数据。"""
|
||||
result = run_compare(week("r1"), week("r2", NEXT_ROWS, source="shop_backup"))
|
||||
fetch = result_queries(result, "fetch")
|
||||
assert fetch["structure"] == "changed" and any("shop_backup" in d for d in fetch["differences"])
|
||||
assert by_id(result)["east"]["classification"] == "sql_changed"
|
||||
|
||||
|
||||
def test_a_literal_edited_in_the_workflow_is_a_template_change():
|
||||
"""图里写死的筛选值改了（'paid' 改成 'refunded'）：执行的 SQL 只差字面量，但这是改模板，不是运行时代入的参数。"""
|
||||
paid, refunded = region_sql(extra=" AND status = 'paid'"), region_sql(extra=" AND status = 'refunded'")
|
||||
result = run_compare(week("r1", sql=paid, graph=weekly_graph(paid)),
|
||||
week("r2", NEXT_ROWS, sql=refunded, graph=weekly_graph(refunded)))
|
||||
east = by_id(result)["east"]
|
||||
assert east["classification"] == "sql_changed", east["reasons"]
|
||||
assert any("区域取数" in r and "'paid'" in r for r in east["reasons"])
|
||||
assert east["decomposition"] is None and east["attribution"] is None
|
||||
# 查询表照实写：执行的 SQL 确实只差这一处字面量
|
||||
assert result_queries(result, "fetch")["structure"] == "params_only"
|
||||
|
||||
|
||||
def test_reformatting_the_workflow_sql_is_not_a_change():
|
||||
loose = region_sql().replace(" FROM ", "\n FROM ").lower()
|
||||
result = run_compare(week("r1"), week("r2", NEXT_ROWS, sql=loose, graph=weekly_graph(loose)))
|
||||
assert {m["classification"] for m in result["metrics"]} == {"data_only"}
|
||||
|
||||
|
||||
def test_renaming_a_node_is_not_a_change():
|
||||
graph = weekly_graph()
|
||||
graph["nodes"][2]["data"]["label"] = "区域分析（新）"
|
||||
result = run_compare(week("r1"), week("r2", NEXT_ROWS, graph=graph))
|
||||
assert {m["classification"] for m in result["metrics"]} == {"data_only"}
|
||||
[renamed] = result["template"]["nodes"]
|
||||
assert (renamed["node_id"], renamed["keys"]) == ("bot", ["label"])
|
||||
|
||||
|
||||
MERGE_SQL = "SELECT region, gmv FROM a ORDER BY region"
|
||||
OUTER_SQL = "SELECT region, gmv FROM m ORDER BY region"
|
||||
MERGED_TYPES = {"region": "text", "gmv": "number"}
|
||||
|
||||
|
||||
def merged(run_id, sql, rows, *, template=None, nested=False, twice=False):
|
||||
"""取数节点查一次，合并节点把它当输入 a 再执行一条合并 SQL（nested：再套一层合并；twice：这次多接一个输入 b），
|
||||
口径卡取合并结果的一格。"""
|
||||
nodes = [node("start", "input", "开始"), node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": template or sql}),
|
||||
node("join", "merge", "合并", sql=MERGE_SQL, inputs={"a": "fetch", **({"b": "fetch"} if twice else {})})]
|
||||
if nested:
|
||||
nodes.append(node("outer", "merge", "再合并", sql=OUTER_SQL, inputs={"m": "join"}))
|
||||
nodes.append(node("card", "metrics", "口径卡", caliber="周报口径", caliber_version="v1"))
|
||||
run = Fake(run_id, graph=chain(*nodes))
|
||||
fetch = run.tool("fetch", sql, rows=rows)
|
||||
table = [[r[0], r[1]] for r in rows]
|
||||
inputs = [("a", "fetch", fetch)] + ([("b", "fetch", fetch)] if twice else [])
|
||||
top, owner = run.merge("join", MERGE_SQL, inputs, ["region", "gmv"], table, types=MERGED_TYPES), "join"
|
||||
if nested:
|
||||
top, owner = run.merge("outer", OUTER_SQL, [("m", "join", top)], ["region", "gmv"], table,
|
||||
types=MERGED_TYPES), "outer"
|
||||
run.card([metric("east", f"cell(nodes.{owner}, 0, 'gmv')", table[0][1],
|
||||
[cell_in(0, "gmv", table[0][1], top, node_id=owner)], name="东区金额")])
|
||||
return run
|
||||
|
||||
|
||||
def test_a_merged_cell_follows_the_queries_feeding_the_merge():
|
||||
"""合并 SQL 一字没改，喂它的取数 SQL 加了筛选：读合并结果的指标是 SQL 变了，不拆、不归因（嵌套的合并也追到底）。"""
|
||||
paid = region_sql(extra=" AND status = 'paid'")
|
||||
for nested in (False, True):
|
||||
east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS, nested=nested),
|
||||
merged("r2", paid, NEXT_ROWS, nested=nested)))["east"]
|
||||
assert east["classification"] == "sql_changed", (nested, east["reasons"])
|
||||
assert east["decomposition"] is None and east["attribution"] is None
|
||||
assert any("区域取数" in r and "SQL" in r for r in east["reasons"]), east["reasons"]
|
||||
# 上游只是代进去的日期不同：参数变化
|
||||
template = region_sql(("{{ input.start }}", "{{ input.end }}"))
|
||||
east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS, template=template),
|
||||
merged("r2", region_sql(WEEK37), NEXT_ROWS, template=template)))["east"]
|
||||
assert east["classification"] == "params_only", east["reasons"]
|
||||
# 合并 SQL 和上游都没变：数据变化
|
||||
east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS), merged("r2", region_sql(), NEXT_ROWS)))["east"]
|
||||
assert east["classification"] == "data_only", east["reasons"]
|
||||
# 合并节点这次多接了一个输入
|
||||
east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS), merged("r2", region_sql(), NEXT_ROWS, twice=True)))
|
||||
assert east["east"]["classification"] == "sql_changed" and any("「b」" in r for r in east["east"]["reasons"])
|
||||
|
||||
|
||||
def test_an_agent_reconfigured_with_the_same_sql_is_a_template_change():
|
||||
"""Agent 改了提示词、SQL 一字不差：它按出处交来的字段可能换了取法，不能报成数据变化。"""
|
||||
graph = weekly_graph()
|
||||
|
||||
Reference in New Issue
Block a user