From 36569e1fa458a01453f0a679270ed4fe3de8de1e Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 12:28:38 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=90=E8=A1=8C=E5=AF=B9=E6=AF=94=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=E4=BF=AE=E6=AD=A3=EF=BC=88=E7=AC=AC=203=20=E8=BD=AE?= =?UTF-8?q?=EF=BC=89=EF=BC=9A=E8=A1=8C=E5=8F=B7=E7=9B=B8=E5=90=8C=E7=9A=84?= =?UTF-8?q?=E5=BC=95=E7=94=A8=E8=BF=98=E8=A6=81=E5=BC=95=E7=94=A8=E5=90=8C?= =?UTF-8?q?=E4=B8=80=E6=89=B9=E7=BB=B4=E5=BA=A6=E5=80=BC=EF=BC=8C=E8=AE=B0?= =?UTF-8?q?=E4=B8=8D=E5=87=BA=E5=86=99=E5=85=A5=E8=80=85=E6=97=B6=E5=8F=AA?= =?UTF-8?q?=E7=9C=8B=E5=8F=A3=E5=BE=84=E5=8D=A1=E4=B9=8B=E5=89=8D=E7=9A=84?= =?UTF-8?q?=E8=8A=82=E7=82=B9=EF=BC=8C=E9=BB=98=E8=AE=A4=E8=AE=BE=E7=BD=AE?= =?UTF-8?q?=E7=9A=84=E7=99=BB=E8=AE=B0=E6=8C=89=E8=8A=82=E7=82=B9=E7=B1=BB?= =?UTF-8?q?=E5=9E=8B=E9=80=90=E7=B1=BB=E6=A0=B8=E5=AF=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Agent 两次引用的行号相同、都不是整份结果时，一边引用了的维度值（区域……）在另一边的快照里却没被引用（按区域排序多了北区、按金额取前 3 名名次变了），判为说不清，说明写出是哪个维度值、在第几行；不再把还在快照里的区域算成「数据没了」、给出远超 100% 的占比。另一边快照里真的没有了的照常是数据变化。数组字段里的单个元素同样核对。维度按两边整份结果认，遮罩列不拿来认，分不出维度的仍按行号认 - 两边都记不出写入者时，范围从整张图改为口径卡之前可能执行过的节点（它的上游加并行支路）：口径卡自己、只在它下游的写周报写不了它读到的值，不再被点名成来源；并行支路上的节点不说成「上游」 - 「整份结果」统一按引用了哪几行判断，容器一支不再单列 - 默认设置登记表的核对测试改为按节点类型：从各执行器出发顺着用到的函数（含就地 import、回调、类）找 cfg 读法，在字面量元组上循环读的键名认得出，认不出的变量键名直接失败；核对节点模块里每个读配置的函数都走到了，登记了不读的也算错。据此补登记成果节点出具契约读的报告节点三项（evidence_from、metrics_from、entities） - 补测试守住上一轮的几条规则：Agent 交的不同而代码节点的数没变是没变，只有这次运行的快照读不出来时说这次，默认设置同时改了结论句裁判时 Agent 只写模型，调不调模型由默认设置决定，报告节点在代码节点上游，改了几个节点时点名画布上靠前的 --- backend/app/engine/compare.py | 179 +++++++++++---- backend/tests/test_compare_engine.py | 311 ++++++++++++++++++++++++--- 2 files changed, 417 insertions(+), 73 deletions(-) diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py index b727210a..81b7e701 100644 --- a/backend/app/engine/compare.py +++ b/backend/app/engine/compare.py @@ -23,10 +23,12 @@ 结果时，再按别名一路追到喂它的每一份上游快照。只差字面量时还要看图里的 SQL：图里的 SQL 改过（改了写死的筛选值、 系数）是模板变化，图里一字未改的才是运行时代入的参数 - Agent 按出处交来的字段还要引用同一列、同一段行（两边都引用整份结果的不算换了；行段相同、一边是整份结果一边 - 不是的算换了）：取哪一列、哪几行是 Agent 每次运行时定的，换了就不是同一个数 + 不是的算换了；行号相同、都不是整份结果的，一边引用了的维度值在另一边的快照里、却没被引用的也算换了——取前几名 + 时名次变了、按区域排序时多了一个区域）：取哪一列、哪几行是 Agent 每次运行时定的，换了就不是同一个数 - 不落在快照上的：模型、人交出的值（Agent 没按出处交、模型调用、人工审批……）没有出处，值变了说不清是数据还是它 - 本身的差异；代码、数据整形这类节点看它自己查过的库，不查库的看它的配置和画布上的全部上游。上游的查询结果两次 - 一模一样、值却变了的同样说不清；上游有模型、人交出值的节点、两次交出的不同的也说不清 + 本身的差异；代码、数据整形这类节点看它自己查过的库，不查库的看它的配置和画布上的全部上游；记不出写入者的看 + 口径卡之前可能执行过的节点（口径卡自己、只在它下游的不算）。上游的查询结果两次一模一样、值却变了的同样说不清； + 上游有模型、人交出值的节点、两次交出的不同的也说不清 - 图级的变化落到节点上：连线有增删的，终点节点算改了；默认设置改了的，运行时会读这几项（_CFG_KEYS）、又没有自行 设置的节点算改了 - 严格拆分（只对口径变了的）：旧表达式的每个叶子路径都能在这次的口径卡输入里取得值时， @@ -97,10 +99,12 @@ _MODEL_KEYS = frozenset({"provider", "model", "temperature", "max_tokens", "thin #: 各类节点运行时按 NodeContext.cfg 读的配置项：节点上没写这一项就回退到图级默认设置（GraphSpec.defaults），默认 #: 设置改了这一项，没自行设置它的这类节点交出的值可能跟着变；不读它的不受影响（结论句裁判只有报告节点读）。读了也 #: 不影响交出的值的见 _QUIET_DEFAULTS。合并查询、双盲复算、反方审稿不经 cfg 读配置。执行器新读了一项要在这里登记， -#: tests/test_compare_engine.py 按源码核对 +#: tests/test_compare_engine.py 按各类节点的执行器（连同它用到的函数）的源码逐类核对 _CFG_KEYS: dict[str, frozenset[str]] = { "input": frozenset({"fields"}), - "output": frozenset({"fields", "contract"}), + # 出具契约按报告节点同一套参数重建证据目录：后三项读的是报告节点上的，报告节点没写同样回退到默认设置。成果节点 + # 自己从不写这三项，默认设置改了就算它跟着变（报告节点自己写了的也算，宁可多算） + "output": frozenset({"fields", "contract", "evidence_from", "metrics_from", "entities"}), "transform": frozenset({"mode", "expression", "template", "assign_to"}), "llm": _MODEL_KEYS | {"system", "skills", "use_history", "prompt", "output_schema", "emit_message", "assign_to"}, "agent": _MODEL_KEYS | {"system", "skills", "use_history", "prompt", "output_schema", "cite_fields", "tools", @@ -691,6 +695,20 @@ def _worst(changes: Iterable[_Change]) -> _Change: return max(found, key=lambda c: _WEIGHT[c.kind]) if found else _Change("same") +def _reach(node_id: str, links: Mapping[str, set[str]], memo: dict[str, set[str]]) -> set[str]: + """顺着连线（links：节点 → 上一层或下一层的节点）能走到的全部节点，不含它自己（成环的除外）。""" + if node_id not in memo: + seen: set[str] = set() + todo = [node_id] + while todo: + for nxt in links.get(todo.pop(), ()): + if nxt not in seen: + seen.add(nxt) + todo.append(nxt) + memo[node_id] = seen + return memo[node_id] + + def _from_input(path: str, item: dict[str, Any]) -> bool: head = _REF_HEAD.match(path) return item.get("via") == "input" or (head is not None and head.group(1) == "input") @@ -709,6 +727,7 @@ class _Compare: self._norm: dict[tuple[str, str | None], NormalizedSql] = {} self._shapes: dict[tuple[int, int], _Shape] = {} self._ancestors: dict[str, set[str]] = {} + self._descendants: 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._norm_of) @@ -719,8 +738,10 @@ class _Compare: self.order = {n: i for i, n in enumerate(dict.fromkeys([*self.target_nodes, *self.base_nodes]))} self.base_outputs, self.target_outputs = _outputs(base), _outputs(target) self.parents: dict[str, set[str]] = {} + self.children: dict[str, set[str]] = {} for source, target_id, _ in _edges(base.graph) | _edges(target.graph): self.parents.setdefault(target_id, set()).add(source) + self.children.setdefault(source, set()).add(target_id) # ---------------------------------------------------------------- 结果 @@ -921,24 +942,26 @@ class _Compare: def _upstream(self, node_id: str) -> set[str]: """画布上 node_id 的全部上游（两次运行的图合起来，宁可多算）。""" - if node_id not in self._ancestors: - seen: set[str] = set() - todo = [node_id] - while todo: - for parent in self.parents.get(todo.pop(), ()): - if parent not in seen: - seen.add(parent) - todo.append(parent) - self._ancestors[node_id] = seen - return self._ancestors[node_id] + return _reach(node_id, self.parents, self._ancestors) + + def _downstream(self, node_id: str) -> set[str]: + """画布上 node_id 的全部下游（两次运行的图合起来）。""" + return _reach(node_id, self.children, self._descendants) + + def _before(self, card: str) -> set[str]: + """口径卡执行之前可能执行过的节点：画布上它的上游，加上和它不在一条线上的（并行的支路可能先写了变量）。它自己、 + 只在它下游的节点在它之后才执行，写不了它读到的值（循环回边连回来的算上游）。""" + after = self._downstream(card) - self._upstream(card) + return {n for n in self.titles if n != card and n not in after} 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)}」" + def _moved(self, node_id: str, *, upstream: bool = True) -> str: + """一个影响取值的节点变化（self.changed）写成一句话。upstream 是它在不在画布上的上游（并行支路上的节点不在， + 不说成「上游」）。""" + name = f"{'上游节点' if upstream else '节点'}「{self.title(node_id)}」" change = self.changed.get(node_id, "") if change == "defaults": keys = self.inherited.get(node_id, []) @@ -959,9 +982,9 @@ class _Compare: before, after = self.base_outputs.get(node_id, ()), self.target_outputs.get(node_id, ()) return before != after or None in before or None in after - def provenance(self, path: str, b: Any, t: Any, *, moved: bool = True) -> _Change: - """一个输入的出处在两次运行之间变没变（b、t 是推演重算还原出来的输入）。moved 是这个指标的值变没变：输入是 - 容器、完整值又还原不出来，核对不了它变没变时，按它推断。""" + def provenance(self, path: str, b: Any, t: Any, *, card: str, moved: bool = True) -> _Change: + """一个输入的出处在两次运行之间变没变（b、t 是推演重算还原出来的输入，card 是读它的口径卡节点）。moved 是这个 + 指标的值变没变：输入是容器、完整值又还原不出来，核对不了它变没变时，按它推断。""" if b is None or t is None: return _Change("unknown", f"输入「{path}」只在一边的口径卡里有记录") if b.snapshot or t.snapshot: @@ -992,21 +1015,30 @@ class _Compare: return own if writer and writer in self.changed: return _Change("template", self._moved(writer)) - # 写它的节点不查库（代码、数据整形……）：看画布上它的全部上游；写它的节点不知道就看整张图 - scope = self._upstream(writer) if writer else set(self.titles) + # 写它的节点不查库（代码、数据整形……）：看画布上它的全部上游。写它的节点不知道时，看口径卡之前可能执行过的 + # 节点（_before）：口径卡自己、只在它下游的节点（写周报……）写不了它读到的值，点名它们就把下游的变化说成了 + # 它的来源。above 是点名时算「上游」的节点 + if writer: + scope = above = self._upstream(writer) + else: + scope, above = self._before(card), self._upstream(card) moved_nodes = self._ordered(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: - return _Change("unknown", f"输入「{path}」{where}，{self._moved(moved_nodes[0])}，无法判断变化来源") + first = moved_nodes[0] + return _Change("unknown", f"输入「{path}」{where}，{self._moved(first, upstream=first in above)}，" + "无法判断变化来源") if worst.kind in ("sql", "template", "unknown"): return _Change("unknown", f"输入「{path}」{where}，对不上具体的查询；它的上游有变化，无法判断变化来源") if not same and (spoken := self._ordered(n for n in scope if self._regenerated(n))): # 上游有模型、人交出的值，两次不同：它交的数变了，可能是数据变了，也可能是模型本身的差异（它可能用的 # 正是 Agent 没按出处交的数）。上游的查询结果变没变都一样说不清 - kind = TYPE_LABEL.get(self._type_of(spoken[0]) or "", "节点") - return _Change("unknown", f"输入「{path}」{where}，它的上游节点「{self.title(spoken[0])}」（{kind}）两次" - "交出的结果不同，无法判断变化是否来自数据") + first = spoken[0] + who = ("它的上游节点" if writer else "上游节点") if first in above else "节点" + kind = TYPE_LABEL.get(self._type_of(first) or "", "节点") + return _Change("unknown", f"输入「{path}」{where}，{who}「{self.title(first)}」（{kind}）两次交出的结果不同，" + "无法判断变化是否来自数据") if worst.kind == "same" and not same and not self._data_moved(scope): # 上游查到的数据两次一模一样，它交的值却变了：不是数据变的。上游只有运行输入变了的，算参数 if (self.base.inputs or {}) != (self.target.inputs or {}) \ @@ -1039,7 +1071,8 @@ class _Compare: def _citation(self, agent: Any, b: Any, t: Any, qb: _Query, qt: _Query) -> _Change | None: """Agent 按出处交来的字段两次引用的是不是同一处：同一列（对象数组是同一组列）、同一行或同一段行。两边引用的都是 整份结果的算同一处（这次多了几行也是）；行段相同、一边是整份结果一边不是的不算——没引用的那几行还在这次的快照 - 里，按维度配对会把它们算成「数据没了」。不是同一处、核对不了是不是同一处的，返回说明。""" + 里，按维度配对会把它们算成「数据没了」。行号相同、都不是整份结果的，还要引用同一批维度值（_left_out）。不是 + 同一处、核对不了是不是同一处的，返回说明。""" 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") @@ -1050,11 +1083,13 @@ class _Compare: if b.row is not None or t.row is not None: # 单格（数组字段里的一个元素、按出处交的一格）：取哪一行由口径卡的下标和 Agent 交的行段一起定，同一行才是 # 同一处 - return None if (b.row, b.rows) == (t.row, t.rows) else changed + return self._left_out(name, what, b, t, qb, qt, cb) if (b.row, b.rows) == (t.row, t.rows) else changed wb, wt = _whole(b, qb), _whole(t, qt) same_span = b.rows == t.rows - if (wb and wt) or (same_span and wb is False and wt is False): + if wb and wt: return None + if same_span and wb is False and wt is False: + return self._left_out(name, what, b, t, qb, qt, cb) if (wb is None or wt is None) and (same_span or (wb is not False and wt is not False)): # 快照读不出来，核对不了是不是整份结果：行段相同的可能一边整份一边不是，行段不同的也可能两边都是整份 side = "基准运行" if wb is None else "这次运行" @@ -1067,6 +1102,45 @@ class _Compare: return _Change("unknown", f"{name}在基准运行中只引用了{_rows_text(b)}（结果共 {len(qb.snap['rows'])} 行），" "这次引用了整份结果，取的不是同一组数") + def _left_out(self, name: str, what: str, b: Any, t: Any, qb: _Query, qt: _Query, column: Any) -> _Change | None: + """行号相同、都不是整份结果的引用，引用的是不是同一批维度值（区域……）。一边引用了的，另一边没引用、却还在那边的 + 快照里，取的就不是同一组数：按区域排序的结果多了北区，西区挪出了引用的行段；按金额取前 3 名，北区挤了进来—— + 按维度配对会把还在的西区算成「数据没了」。另一边的快照里也没有的，是数据真的没了，照常比。维度按两边整份结果 + 认（和归因同一个规则，_dimensions）；分不出维度（没有不是数值的列、维度列设了遮罩）、快照读不出来、行的形状 + 不对的，只能按行号认，返回 None。""" + if qb.snap is None or qt.snap is None: + return None + rb, rt = _records(qb.snap, None), _records(qt.snap, None) + if rb is None or rt is None: + return None + dims, _ = _dimensions(qb.snap, qt.snap, rb, rt, column if isinstance(column, str) else "", self._hidden(b, t)) + if not dims: + return None + + def split(records: list[tuple[int, dict[str, Any]]], inp: Any) -> tuple[dict[Any, int], dict[Any, int]]: + """{维度键: 第一次出现的行号}：引用的行里的，引用的行外面的。""" + first, last = _cited(inp, len(records)) + inside: dict[Any, int] = {} + outside: dict[Any, int] = {} + for i, row in records: + (inside if first <= i <= last else outside).setdefault(tuple(_hashable(row.get(d)) for d in dims), i) + return inside, outside + + (in_b, out_b), (in_t, out_t) = split(rb, b), split(rt, t) + for key in in_b: + if key not in in_t and key in out_t: + return _Change("unknown", f"{name}两次都引用{_rows_text(t)}，{_key_text(key)}这次仍在查询结果第 " + f"{out_t[key] + 1} 行，不在引用的行里，取的不是{what}") + for key in in_t: + if key not in in_b and key in out_b: + return _Change("unknown", f"{name}两次都引用{_rows_text(b)}，{_key_text(key)}在基准运行中位于查询结果第 " + f"{out_b[key] + 1} 行，不在引用的行里，取的不是{what}") + return None + + def _hidden(self, b: Any, t: Any) -> frozenset[str]: + """两边输入落在的快照上要遮的列（小写）。""" + return (self.base.masked.get(b.snapshot) or frozenset()) | (self.target.masked.get(t.snapshot) or frozenset()) + def _generated(self, path: str, writer: str, same: bool) -> _Change: """模型、人交出的值（Agent 没按出处交、模型调用、人工审批……）：没有出处，值变了说不清是数据变了还是它本身的 差异——同样的数据、同样的配置，模型两次交出的数也可能不同。它自己查的库结构变了、它的配置变了照常标出。""" @@ -1241,7 +1315,7 @@ class _Compare: 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) + changes = {p: self.provenance(p, bp.inputs.get(p), tp.inputs.get(p), card=card_id, moved=moved) for p in dict.fromkeys([*tp.inputs, *bp.inputs])} if caliber: # 口径优先；喂它的取数也变了、说不清的，接在口径的说明后面照样写出来，不因为口径变了就不提。只在一边才有 @@ -1276,7 +1350,7 @@ class _Compare: if metric.get("status") == "missing_input": out["reasons"].append(f"{side}这个指标缺少输入，没有值") if out["classification"] in _ATTRIBUTED and bp is not None and tp is not None: - out["attribution"] = self._attribution(bp, tp) + out["attribution"] = self._attribution(bp, tp, card_id) if any(inp.masked for plan in (bp, tp) if plan is not None for inp in plan.inputs.values()): self.masked_seen = True return out @@ -1325,7 +1399,7 @@ class _Compare: if any(inp.masked for inp in used) or any(inp.masked for plan in plans for inp in plan.inputs.values()): return None, SPLIT_MASKED for path in leaves: - change = self.provenance(path, bp.inputs.get(path), t.inputs[path]) + change = self.provenance(path, bp.inputs.get(path), t.inputs[path], card=t.node_id) if change.kind in ("sql", "template"): return None, f"旧口径用到的输入取数也有变化，无法严格拆分：{change.reason}" if change.kind == "unknown": @@ -1375,7 +1449,7 @@ class _Compare: # ---------------------------------------------------------------- 维度归因 - def _attribution(self, bp: Any, tp: Any) -> dict[str, Any] | None: + def _attribution(self, bp: Any, tp: Any, card: str) -> dict[str, Any] | None: """挑一个落在查询快照上、两边都变了的输入做维度归因：先找给得出占比的，没有就取第一个列得出行的，再没有 就取第一个说明（截断、度量列遮罩……）。一个都挑不出来返回 None。""" found: list[dict[str, Any]] = [] @@ -1383,7 +1457,7 @@ class _Compare: b = bp.inputs.get(path) if b is None or not (b.snapshot and t.snapshot) or recompute._same(b.value, t.value): continue - if self.provenance(path, b, t).kind not in ("same", "params"): + if self.provenance(path, b, t, card=card).kind not in ("same", "params"): continue result = self._attribute(b, t) if result is None: @@ -1409,12 +1483,12 @@ class _Compare: if snap.get("truncated"): # 截断的结果只是前几行：各键的和不是完整的和，按它归因会把没取回的行的变化算到别的键头上 return {**empty, "note": TRUNCATED.format(side=side, rows=len(snap["rows"]))} - hidden = (self.base.masked.get(b.snapshot) or frozenset()) | (self.target.masked.get(t.snapshot) or frozenset()) + hidden = self._hidden(b, t) if any(str(c).lower() in hidden for c in [*(sb.get("columns") or []), *(st.get("columns") or [])]): self.masked_seen = True if measure.lower() in hidden: return {**empty, "note": MASKED_MEASURE.format(column=measure)} - rb, rt = _records(sb, b), _records(st, t) + rb, rt = _records(sb, b.rows), _records(st, t.rows) if rb is None or rt is None: return None values = (_input_total(b), _input_total(t)) @@ -1546,14 +1620,28 @@ def _same_input(b: Any, t: Any) -> bool | None: return recompute._same(b.item.get("value"), t.item.get("value")) +def _cited(inp: Any, count: int) -> tuple[int, int]: + """输入引用了快照的哪几行（第一行, 最后一行），count 是快照共几行：单格是那一行，数组字段是它那一段行，容器本身 + 是一份查询结果的是全部的行。""" + if inp.row is not None: + return inp.row, inp.row + if inp.rows is not None: + return inp.rows + return 0, count - 1 + + def _whole(inp: Any, q: _Query) -> bool | None: - """这个输入引用的是不是整份查询结果：容器本身就是整份结果，或者数组字段的行段从第一行到最后一行。快照读不出来、 - 不知道结果共几行的返回 None。""" - if inp.table: - return True + """这个输入引用的是不是整份查询结果：引用的行从第一行到最后一行（容器本身就是整份结果）。快照读不出来、不知道 + 结果共几行的返回 None。""" if q.snap is None: return None - return inp.rows is not None and inp.rows == (0, len(q.snap["rows"]) - 1) + count = len(q.snap["rows"]) + return _cited(inp, count) == (0, count - 1) + + +def _key_text(key: tuple[Any, ...]) -> str: + """维度值写进说明：几列的用「 / 」连起来，和对比页上归因行的写法一样。""" + return "「" + " / ".join(_shown(v) for v in key) + "」" def _rows_text(inp: Any) -> str: @@ -1578,11 +1666,12 @@ def _merge_inputs(q: _Query) -> dict[str, str] | None: 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: - """归因用到的行：(行号, {列: 值})。数组字段只用它那一段行，单格用整份结果。""" +def _records(snap: dict[str, Any], span: tuple[int, int] | None) -> list[tuple[int, dict[str, Any]]] | None: + """快照里的行：(行号, {列: 值})。span 给了只取那一段行（归因时数组字段只用它那一段），None 是整份结果（单格按 + 整份结果归因）。行的形状不对返回 None。""" columns = [str(c) for c in snap.get("columns") or []] rows = snap.get("rows") or [] - first, last = (inp.rows if inp.rows is not None else (0, len(rows) - 1)) + first, last = span if span is not None else (0, len(rows) - 1) out = [] for i in range(first, min(last, len(rows) - 1) + 1): row = rows[i] diff --git a/backend/tests/test_compare_engine.py b/backend/tests/test_compare_engine.py index 374ad4ae..0417a722 100644 --- a/backend/tests/test_compare_engine.py +++ b/backend/tests/test_compare_engine.py @@ -10,20 +10,27 @@ SQL，Agent 节点 bot 自己再查一次、把各区域的金额作为数组字 - 遮罩列的值绝不出现在结果里；用到遮罩列的指标不拆分 - 模板对比：新增、删除的节点，配置有变化的节点和配置项名 - 输入的出处：合并查询追到它的上游查询；Agent 按出处交的字段还要引用同一列、同一段行（行段相同、一边是整份结果 - 一边不是的不算）；没按出处交的数、上游数据没变数却变了的、上游的 Agent 两次交出的结果不同的，说不清变化从哪来 - （incomparable）；图里写死的字面量改了、连线变了、读到的默认设置变了都算模板变化 + 一边不是的不算；都不是整份结果的，一边引用了的区域另一边还在快照里却没引用的也不算）；没按出处交的数、上游数据 + 没变数却变了的、上游的 Agent 两次交出的结果不同的，说不清变化从哪来（incomparable），记不出写入者的只看口径卡 + 之前可能执行过的节点；图里写死的字面量改了、连线变了、读到的默认设置变了都算模板变化 """ from __future__ import annotations +import ast import copy +import importlib import inspect import json -import re +import pkgutil +import textwrap from decimal import Decimal +from types import SimpleNamespace +import app.engine.nodes as nodes_package from app.engine import compare from app.engine.compiler import RUNNERS from app.engine.merge_query import MERGE_SOURCE +from app.engine.schema import NodeType from tests.fixtures.compare_runs import ( BASE_ROWS, COLUMNS, @@ -610,6 +617,9 @@ def test_a_code_node_after_an_agent_that_answered_differently_is_incomparable(): assert calc["classification"] == "incomparable" and calc["attribution"] is None assert calc["reasons"] == ["输入「vars.adj」来自「调整」，它的上游节点「区域分析」（Agent）两次交出的结果不同，" "无法判断变化是否来自数据"] + # Agent 交的不同，代码节点交的数两次一样：照常没变，没有可说不清的变化 + calc = by_id(run_compare(make("r1", BASE_ROWS, 0.46, value=1.0), make("r2", NEXT_ROWS, 0.52, value=1.0)))["calc"] + assert (calc["classification"], calc["reasons"]) == ("unchanged", []) # Agent 的产出两次都没落成工件（写工件失败）：一样不一样核对不了，同样不报成数据变化 def unstored(run): @@ -622,20 +632,89 @@ def test_a_code_node_after_an_agent_that_answered_differently_is_incomparable(): def test_a_reconfigured_node_upstream_of_a_code_node_is_named(): - """代码节点上游的数据整形节点改了表达式（SQL 一字没改）：代码节点交的数变了说不清从哪来，说明里点出改的是哪个节点。""" - def make(run_id, expression, rows, value): - shape = node("shape", "transform", "整理", mode="expression", expression=expression, assign_to="kept") - run = Fake(run_id, graph=weekly_graph(after_fetch=(shape, calc_node()))) + """代码节点上游的数据整形节点改了表达式（SQL 一字没改）：代码节点交的数变了说不清从哪来，说明里点出改的是哪个节点； + 改了几个的，点名画布上最靠前的那个（节点 ID 的先后和画布上的先后不同）。""" + def make(run_id, expressions, rows, value): + shapes = [node(nid, "transform", label, mode="expression", expression=expression, assign_to=f"kept_{nid}") + for (nid, label), expression in zip((("shape_m", "整理"), ("shape_z", "再整理"), ("shape_a", "汇总")), + expressions)] + run = Fake(run_id, graph=weekly_graph(after_fetch=(*shapes, calc_node()))) 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 - calc = by_id(run_compare(make("r1", "nodes.fetch.rows", BASE_ROWS, 10), - make("r2", "nodes.fetch.rows[:2]", NEXT_ROWS, 12)))["calc"] + same = ["nodes.fetch.rows"] * 3 + calc = by_id(run_compare(make("r1", same, BASE_ROWS, 10), + make("r2", ["nodes.fetch.rows[:2]", *same[1:]], NEXT_ROWS, 12)))["calc"] assert calc["classification"] == "incomparable" assert calc["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「整理」的配置有变化，无法判断变化来源"] + calc = by_id(run_compare(make("r1", same, BASE_ROWS, 10), + make("r2", ["nodes.fetch.rows[:2]"] * 3, NEXT_ROWS, 12)))["calc"] + assert calc["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「整理」的配置有变化，无法判断变化来源"] + + +def test_a_report_upstream_of_a_code_node_that_wrote_differently_is_incomparable(): + """代码节点上游有报告撰写节点（文字是模型写的），两次写得不同：代码节点交的数变了，可能来自数据，也可能来自模型 + 本身的差异；两次写得一字不差的（产出工件一样）照常是数据变化。""" + def make(run_id, rows, text, value): + run = Fake(run_id, graph=weekly_graph(after_fetch=(node("draft", "report", "初稿"), calc_node()))) + run.tool("fetch", region_sql(), rows=rows) + run.add("node.started", "draft", {"node_type": "report"}) + run.add("node.finished", "draft", {"artifact": run.put({"text": text})}) + run.agent("bot", region_sql(), rows=rows) + run.code("calc", value, "adj") + run.card(code_metrics(value)(None, None, rows, rows)) + return run + + calc = by_id(run_compare(make("r1", BASE_ROWS, "上周初稿", 10), make("r2", NEXT_ROWS, "本周初稿", 12)))["calc"] + assert calc["classification"] == "incomparable" + assert calc["reasons"] == ["输入「vars.adj」来自「调整」，它的上游节点「初稿」（报告撰写）两次交出的结果不同，" + "无法判断变化是否来自数据"] + calc = by_id(run_compare(make("r1", BASE_ROWS, "初稿", 10), make("r2", NEXT_ROWS, "初稿", 12)))["calc"] + assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM]) + + +def test_an_unknown_writer_is_looked_for_only_among_nodes_before_the_card(): + """两边都记不出 vars.adj 是哪个节点写的：只看口径卡之前可能执行过的节点。口径卡自己改了配置、它下游的写周报两次写得 + 不同，都写不了它读到的值，不能点名成它的上游；它上游的节点改了配置照常说不清，并行支路上的节点也可能先写了变量， + 同样说不清，只是不说成「上游」。""" + def make(run_id, rows, value, *, other="1", report="上周", reshape="nodes.fetch.rows", aside="nodes.fetch.rows"): + graph = chain(node("start", "input", "开始"), + node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": region_sql()}), + node("shape", "transform", "整理", mode="expression", expression=reshape, assign_to="kept"), + node("calc", "code", "调整", code="x", assign_to="adj"), + node("card", "metrics", "周报口径卡", caliber="周报口径", caliber_version="v1", + metrics=[{"id": "other", "expression": other}]), + node("write", "report", "写周报"), node("out", "output", "成果")) + # 并行支路：开始 → 旁路整理 → 成果，和口径卡不在一条线上 + graph["nodes"].append(node("aside", "transform", "旁路整理", mode="expression", expression=aside, + assign_to="extra")) + graph["edges"] += [{"source": "start", "target": "aside"}, {"source": "aside", "target": "out"}] + run = Fake(run_id, graph=graph) + run.tool("fetch", region_sql(), rows=rows) + run.code("calc", value, "adj") + run.card([metric("calc", "vars.adj * 2", value * 2, [scalar_in("vars.adj", value, None, "state")])]) + run.add("node.started", "write", {"node_type": "report"}) + run.add("node.finished", "write", {"artifact": run.put({"text": report})}) + return run + + # 口径卡自己改了另一个指标的定义，写周报两次写得不同：都在它之后执行，变化照常来自上游数据 + calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, other="2", report="本周")))["calc"] + assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM_SCOPED]) + calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, report="本周")))["calc"] + assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM]) + # 口径卡上游的整理节点改了表达式 + calc = by_id(run_compare(make("r1", BASE_ROWS, 10), + make("r2", NEXT_ROWS, 12, reshape="nodes.fetch.rows[:2]")))["calc"] + assert calc["classification"] == "incomparable" + assert calc["reasons"] == ["输入「vars.adj」不知道由哪个节点写入，上游节点「整理」的配置有变化，无法判断变化来源"] + # 并行支路上的旁路整理改了表达式 + calc = by_id(run_compare(make("r1", BASE_ROWS, 10), + make("r2", NEXT_ROWS, 12, aside="nodes.fetch.rows[:2]")))["calc"] + assert calc["classification"] == "incomparable" + assert calc["reasons"] == ["输入「vars.adj」不知道由哪个节点写入，节点「旁路整理」的配置有变化，无法判断变化来源"] def test_a_run_input_upstream_of_a_code_node_is_a_parameter_change(): @@ -798,6 +877,11 @@ def test_a_changed_default_model_is_a_template_change_for_agents_that_follow_it( many["defaults"] = {"provider": "openai", "model": "model-b", "temperature": 0.5, "max_tokens": 2000} share = by_id(run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=many)))["share"] assert share["reasons"] == ["上游节点「区域分析」跟随的工作流默认设置（模型接入、模型、温度等 4 项）有变化"] + # 同时改了 Agent 不读的结论句裁判：只写它跟着变的模型 + judged = copy.deepcopy(target_graph) + judged["defaults"]["judge"] = {"max_cost_usd": 1.0} + share = by_id(run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=judged)))["share"] + assert share["reasons"] == ["上游节点「区域分析」跟随的工作流默认设置（模型）有变化"] # Agent 自己指定了模型：默认设置变了它不跟着变 for graph in (base_graph, target_graph): graph["nodes"][2]["data"]["config"]["model"] = "model-x" @@ -823,17 +907,19 @@ def test_defaults_that_agents_do_not_read_leave_them_alone(): def test_a_model_default_reaches_every_node_that_calls_the_model(): """默认模型换了，跟着换的不只 Agent：让模型分类的条件分支、让模型修复的结构校验（默认开着）、用模型重排的知识检索 也是。它们在代码节点上游时，代码节点交的数变了不能说来自上游数据；不调模型的照常。""" - def calc(extra): - def make(run_id, model, rows, value): + def calc(extra, before=None, after=None): + def make(run_id, defaults, rows, value): graph = weekly_graph(after_fetch=(extra, calc_node())) - graph["defaults"] = {"provider": "anthropic", "model": model} + graph["defaults"] = defaults 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 - return by_id(run_compare(make("r1", "model-a", BASE_ROWS, 10), make("r2", "model-b", NEXT_ROWS, 12)))["calc"] + before = before or {"provider": "anthropic", "model": "model-a"} + after = after or {"provider": "anthropic", "model": "model-b"} + return by_id(run_compare(make("r1", before, BASE_ROWS, 10), make("r2", after, NEXT_ROWS, 12)))["calc"] cases = [{"key": "peak", "condition": "true"}] for extra in (node("route", "branch", "分流", mode="llm", instruction="判断是否旺季", cases=cases), @@ -849,29 +935,121 @@ def test_a_model_default_reaches_every_node_that_calls_the_model(): found = calc(extra) assert found["classification"] == "data_only", (extra["type"], found["reasons"]) + # 调不调模型也可能由默认设置定：条件分支没写判断方式、默认设置写了让模型判断的，跟着默认模型变；结构校验的默认 + # 设置关了让模型修复的，不跟着变 + asks = node("route", "branch", "分流", instruction="判断是否旺季", cases=cases) + found = calc(asks, {"mode": "llm", "model": "model-a"}, {"mode": "llm", "model": "model-b"}) + assert found["classification"] == "incomparable", found["reasons"] + assert found["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「分流」跟随的工作流默认设置（模型）有变化，" + "无法判断变化来源"] + found = calc(node("route", "validate", "校验", schema={"type": "object"}), + {"repair_with_llm": False, "model": "model-a"}, {"repair_with_llm": False, "model": "model-b"}) + assert found["classification"] == "data_only", found["reasons"] + # 默认的判断方式这次改成了按表达式：基准运行那次让模型判断过，跟着变的是判断方式和模型两项 + found = calc(asks, {"mode": "llm", "model": "model-a"}, {"mode": "expression", "model": "model-b"}) + assert found["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「分流」跟随的工作流默认设置（判断方式、模型）" + "有变化，无法判断变化来源"] + + +#: 只顺着节点模块里的函数往下找：别处的函数不经 NodeContext.cfg 读节点配置 +_NODE_MODULES = "app.engine.nodes." + + +def _defined_here(obj): + """obj 是定义在节点模块里的函数或类（装饰过的取原函数），否则 None。""" + obj = inspect.unwrap(obj) if inspect.isfunction(obj) else getattr(obj, "func", obj) + if (inspect.isfunction(obj) or inspect.isclass(obj)) and obj.__module__.startswith(_NODE_MODULES): + return obj + return None + + +def _settings_read(runner): + """一类节点的执行器运行时按 NodeContext.cfg 读的配置项（节点上没写就回退到默认设置）：执行器自己的函数体，加上它 + 用到的、定义在节点模块里的函数和类，一层层往下找（函数里就地 import 的也算）。返回 (配置项, 认不出键名的读法, + 走到的函数)。键名是变量的，只认在字面量元组上循环出来的（Agent 的预算三项）；别的变量键名认不出是哪几项， + 原样交回，测试据此失败。""" + read, unknown, seen = set(), [], set() + todo = [runner] + while todo: + fn = _defined_here(todo.pop()) + if fn is None or (fn.__module__, fn.__qualname__) in seen: + continue + seen.add((fn.__module__, fn.__qualname__)) + names = vars(importlib.import_module(fn.__module__)) # 它所在模块里的名字（类没有 __globals__） + parts = list(ast.walk(ast.parse(textwrap.dedent(inspect.getsource(fn))))) + imported = {alias.asname or alias.name: (part.module, alias.name) for part in parts + if isinstance(part, ast.ImportFrom) and part.module for alias in part.names} + looped = {} # 循环变量 → 它在字面量元组上取到的键名：{k: ctx.cfg(k) for k in ("max_steps", …)} + for part in parts: + for gen in getattr(part, "generators", None) or ([part] if isinstance(part, ast.For) else []): + if isinstance(gen.target, ast.Name) and isinstance(gen.iter, (ast.Tuple, ast.List)) and all( + isinstance(e, ast.Constant) and isinstance(e.value, str) for e in gen.iter.elts): + looped[gen.target.id] = {e.value for e in gen.iter.elts} + for part in parts: + called = part.func if isinstance(part, ast.Call) else None + if isinstance(called, ast.Attribute) and called.attr == "cfg": + key = part.args[0] if part.args else None + if isinstance(key, ast.Constant) and isinstance(key.value, str): + read.add(key.value) + elif isinstance(key, ast.Name) and key.id in looped: + read |= looped[key.id] + else: + unknown.append(f"{fn.__qualname__}: {ast.unparse(part)}") + elif isinstance(called, ast.Attribute) and called.attr == "approval_mode": + read.add("approval") # NodeContext.approval_mode 同样先读节点、再回退到默认设置 + elif isinstance(called, ast.Name) and called.id == "_setting": + # 报告模块的 _setting(spec, 节点, 键) 和 NodeContext.cfg 同一个规则 + key = part.args[2] if len(part.args) > 2 else None + if isinstance(key, ast.Constant) and isinstance(key.value, str): + read.add(key.value) + else: + unknown.append(f"{fn.__qualname__}: {ast.unparse(part)}") + elif isinstance(part, ast.Name) and isinstance(part.ctx, ast.Load): + # 调用、当回调传出去的函数都算用到了 + target = names.get(part.id) + if target is None and part.id in imported: + module, name = imported[part.id] + target = getattr(importlib.import_module(module), name, None) + todo.append(target) + elif isinstance(part, ast.Attribute) and isinstance(part.value, ast.Name): + module = names.get(part.value.id) + if inspect.ismodule(module): + todo.append(getattr(module, part.attr, None)) + return read, unknown, seen + def test_every_setting_a_node_reads_is_known_to_the_comparison(): """节点运行时按 NodeContext.cfg 读配置，节点上没写就回退到图级默认设置。对比按「哪类节点读哪几项」把默认设置的 - 变化落到节点上（compare._CFG_KEYS）：执行器新读了一项、这里没登记，默认设置改了这一项就会被漏掉。""" - by_module: dict = {} + 变化落到节点上（compare._CFG_KEYS）：执行器新读了一项、这里没登记，默认设置改了这一项就会被漏掉；登记了不读的， + 默认设置改了它就会被误标。按节点类型比，不按模块比：一个模块管几类节点（Agent 和模型调用、调用工具和代码……）， + 其中一类读了、另一类没读的，不能拿另一类的登记充数。""" + reached = set() for kind, runner in RUNNERS.items(): - by_module.setdefault(inspect.getmodule(runner), set()).add(str(kind)) - for module, kinds in by_module.items(): - source = inspect.getsource(module) - read = set(re.findall(r'ctx\.cfg\(\s*"(\w+)"', source)) - read |= set(re.findall(r'_setting\(spec, node, "(\w+)"', source)) - if "ctx.approval_mode()" in source: - read.add("approval") # NodeContext.approval_mode 同样先读节点、再回退到默认设置 - known = set().union( - *(compare._CFG_KEYS.get(k, frozenset()) | compare._CFG_KEYS_IF.get(k, frozenset()) for k in kinds)) - assert read <= known, (module.__name__, sorted(read - known)) + read, unknown, seen = _settings_read(runner) + assert not unknown, (str(kind), unknown) + known = compare._CFG_KEYS.get(str(kind), frozenset()) | compare._CFG_KEYS_IF.get(str(kind), frozenset()) + assert read == known, (str(kind), "没登记", sorted(read - known), "登记了不读", sorted(known - read)) + reached |= seen + # 键名不是字面量的读法也认得出来 + assert {"max_steps", "budget_tokens", "budget_usd"} <= _settings_read(RUNNERS[NodeType.AGENT])[0] + # 节点模块里每个按 cfg 读配置的函数都走到了：走不到的说明上面漏了某种用法，查出来的配置项就不全 + holders = set() + for info in pkgutil.iter_modules(nodes_package.__path__): + module = importlib.import_module(_NODE_MODULES + info.name) + for top in ast.parse(inspect.getsource(module)).body: + if isinstance(top, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and any( + isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) + and n.func.attr in ("cfg", "approval_mode") for n in ast.walk(top)): + holders.add((module.__name__, top.name)) + assert holders and holders <= reached, sorted(holders - reached) -def cited(run_id, rows=BASE_ROWS, *, column="gmv", span=None): +def cited(run_id, rows=BASE_ROWS, *, column="gmv", span=None, sql=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) + sql = sql or region_sql() + run.tool("fetch", sql, rows=rows) + bot = run.agent("bot", 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), @@ -917,6 +1095,65 @@ def test_the_same_rows_are_not_the_same_citation_when_only_one_side_is_the_whole assert total["classification"] == "data_only" and total["attribution"]["rows"] +def test_the_same_rows_are_not_the_same_citation_when_a_cited_region_moved_out_of_them(): + """两边都只引用第 1 至 3 行、都不是整份结果：基准运行引用了西区，这次北区排进了前 3 行，西区挪到第 4 行，还在这次 + 的快照里，只是没被引用。取的不是同一组数，不能把西区算成「数据没了」（占比会远超 100%）；反过来（这次引用的区域 + 在基准运行的快照里、没被引用）也一样。西区这次真的没有了的，照常是数据变化。""" + with_middle = [["东区", 1200.5, 100.0], ["南区", 800.0, 50.0], ["西区", 600.0, 0.0], ["中区", 30.0, 0.0]] + total = by_id(run_compare(cited("r1", with_middle, span=(0, 2)), cited("r2", WITH_NORTH, span=(0, 2))))["total"] + assert total["classification"] == "incomparable" and total["attribution"] is None + assert total["reasons"] == ["Agent「区域分析」两次都引用第 1 至 3 行，「西区」这次仍在查询结果第 4 行，不在引用的" + "行里，取的不是同一组数"] + total = by_id(run_compare(cited("r1", WITH_NORTH, span=(0, 2)), cited("r2", with_middle, span=(0, 2))))["total"] + assert total["classification"] == "incomparable" and total["attribution"] is None + assert total["reasons"] == ["Agent「区域分析」两次都引用第 1 至 3 行，「西区」在基准运行中位于查询结果第 4 行，不在" + "引用的行里，取的不是同一组数"] + + # 按金额从大到小取前 3 名：北区这次挤进了前 3，西区落到第 4 行 + top = region_sql().replace("ORDER BY region", "ORDER BY gmv DESC") + before = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["西区", 600.0, 0.0], ["中区", 30.0, 0.0]] + after = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["北区", 700.0, 0.0], ["西区", 600.0, 0.0]] + total = by_id(run_compare(cited("r1", before, span=(0, 2), sql=top), + cited("r2", after, span=(0, 2), sql=top)))["total"] + assert total["classification"] == "incomparable" and total["attribution"] is None + assert "「西区」这次仍在查询结果第 4 行" in total["reasons"][0] + # 对照：西区这次真的没有了（不在这次的快照里），照常是数据变化，按区域归因，占比之和为 1 + gone = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["北区", 700.0, 0.0], ["中区", 30.0, 0.0]] + total = by_id(run_compare(cited("r1", before, span=(0, 2), sql=top), + cited("r2", gone, span=(0, 2), sql=top)))["total"] + assert (total["classification"], total["reasons"]) == ("data_only", [compare.DATA_FROM_QUERIES]) + rows = total["attribution"]["rows"] + assert [(r["key"], r["base"], r["target"], r["share"]) for r in rows] == [ + (["北区"], None, 700.0, 7.0), (["西区"], 600.0, None, -6.0), (["东区"], 1200.0, 1200.0, 0.0), + (["南区"], 800.0, 800.0, 0.0)] + assert sum(exact(r["share"]) for r in rows) == 1 + + # 区域列设了遮罩：不拿它认是哪一区，只能按行号认，遮住的区域名不出现在结果里 + def masked(run): + run.masked = {a: frozenset({"region"}) for a in run.queries} + return run + result = run_compare(masked(cited("r1", with_middle, span=(0, 2))), masked(cited("r2", WITH_NORTH, span=(0, 2)))) + assert by_id(result)["total"]["classification"] == "data_only" + text = json.dumps(result, ensure_ascii=False) + assert all(region not in text for region in ("东区", "南区", "西区", "北区", "中区")) + + +def test_an_input_cites_the_whole_result_only_when_it_covers_every_row(): + """怎么认整份结果（compare._whole）：容器本身就是一份查询结果的（没有行号、行段）、数组字段的行段从第一行到最后 + 一行的是；单格、只有一段行的不是；快照读不出来的不知道。Agent 的字段在真实运行里总带着行号或行段，容器这一支 + 只是兜底，造不出真实的运行来走它，在这里直接核对。""" + query = compare._Query(artifact="q1", node_id="bot", tool=TOOL, ordinal=1, snap={"rows": [["东区"], ["南区"], ["西区"]]}) + lost = compare._Query(artifact="q1", node_id="bot", tool=TOOL, ordinal=1) + + def cites(row=None, rows=None): + return SimpleNamespace(row=row, rows=rows) + assert compare._whole(cites(), query) is True + assert compare._whole(cites(rows=(0, 2)), query) is True + assert compare._whole(cites(rows=(0, 1)), query) is False + assert compare._whole(cites(row=0), query) is False + assert compare._whole(cites(rows=(0, 2)), lost) is None + + def test_cited_rows_on_an_unreadable_snapshot_are_not_taken_as_the_same(): """基准运行 Agent 那次查询的快照文件不在了：两边都引用第 1 至 3 行，可基准那次是不是整份结果核对不了，不报成 数据变化。""" @@ -929,6 +1166,18 @@ def test_cited_rows_on_an_unreadable_snapshot_are_not_taken_as_the_same(): total = by_id(run_compare(base, cited("r2", WITH_NORTH)))["total"] assert total["classification"] == "incomparable" and "无法核对" in total["reasons"][0] + # 反过来只有这次运行的快照读不出来：说的是这次运行 + def unreadable(run): + del run.store[run.queries[1]] + return run + total = by_id(run_compare(cited("r1"), unreadable(cited("r2", WITH_NORTH, span=(0, 2)))))["total"] + assert total["classification"] == "incomparable" + assert total["reasons"] == ["这次运行的查询快照无法读取，无法核对 Agent「区域分析」引用的是不是整份结果"] + # 基准运行只引用了一部分、这次的行段又不同：这次是不是整份结果都不是同一处，照实说行段换了 + total = by_id(run_compare(cited("r1", WITH_NORTH, span=(0, 2)), unreadable(cited("r2", span=(1, 2)))))["total"] + assert total["classification"] == "incomparable" + assert total["reasons"] == ["Agent「区域分析」引用的行由第 1 至 3 行改为第 2 至 3 行，取的不是同一组数"] + def test_an_element_of_a_cited_array_follows_its_row(): """口径卡取 Agent 数组字段里的第一个元素：Agent 交的行段换了起点，第一个元素落到了别的行，不是同一个数；起点没变 @@ -949,6 +1198,12 @@ def test_an_element_of_a_cited_array_follows_its_row(): assert first["reasons"] == ["Agent「区域分析」引用的行由第 1 行改为第 2 行，取的不是同一个数"] first = by_id(run_compare(make("r1", BASE_ROWS, (0, 2)), make("r2", NEXT_ROWS, (0, 2))))["first"] assert (first["classification"], first["reasons"]) == ("data_only", [compare.DATA_FROM_QUERIES]) + # 起点没变，可这次北区排在了第一行：基准运行取的是东区，东区这次还在第 2 行，取的不是同一个数 + first = by_id(run_compare(make("r1", BASE_ROWS, (0, 2)), + make("r2", [["北区", 90.0, 0.0], *NEXT_ROWS], (0, 3))))["first"] + assert first["classification"] == "incomparable" + assert first["reasons"] == ["Agent「区域分析」两次都引用第 1 行，「东区」这次仍在查询结果第 2 行，不在引用的行里，" + "取的不是同一个数"] def test_no_split_when_an_old_leaf_now_cites_another_column():