diff --git a/backend/app/api/compare.py b/backend/app/api/compare.py index 5b648c45..6d5a7b2b 100644 --- a/backend/app/api/compare.py +++ b/backend/app/api/compare.py @@ -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) diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py index 6fe89179..c899b533 100644 --- a/backend/app/engine/compare.py +++ b/backend/app/engine/compare.py @@ -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 []] diff --git a/backend/tests/fixtures/compare_runs.py b/backend/tests/fixtures/compare_runs.py index 1d8bb6e2..a961018e 100644 --- a/backend/tests/fixtures/compare_runs.py +++ b/backend/tests/fixtures/compare_runs.py @@ -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): diff --git a/backend/tests/test_compare_api.py b/backend/tests/test_compare_api.py index cd2cf0d9..eda3efdf 100644 --- a/backend/tests/test_compare_api.py +++ b/backend/tests/test_compare_api.py @@ -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) diff --git a/backend/tests/test_compare_engine.py b/backend/tests/test_compare_engine.py index 1d2d9f3f..328b0044 100644 --- a/backend/tests/test_compare_engine.py +++ b/backend/tests/test_compare_engine.py @@ -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()