From 271bc6e78213b8e67af640d23c40beb84cee3d61 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sat, 17 Oct 2026 13:07:19 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=A3=E5=BE=84=E6=A0=87=E7=AD=BE=E9=80=90?= =?UTF-8?q?=E6=A0=BC=E5=88=A4=EF=BC=9A=E5=BC=95=E7=94=A8=E4=BA=86=E5=A5=BD?= =?UTF-8?q?=E5=87=A0=E5=88=97=E7=9A=84=E6=9F=A5=E8=AF=A2=EF=BC=8C=E6=AF=8F?= =?UTF-8?q?=E4=B8=AA=E6=95=B0=E6=8C=89=E5=AE=83=E8=87=AA=E5=B7=B1=E9=82=A3?= =?UTF-8?q?=E4=B8=80=E6=A0=BC=E6=A0=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 真实模型下问「上周的销售额是多少」，助手写 SUM(amount) AS 销售额, COUNT(*) AS 订单数, MIN / MAX(paid_date) AS 区间首日 / 末日，报告四列都引用了，口径「周销售」只取 SUM(amount)。按「全部引用列」投影后还是 4 列，对不上口径的 1 列，整条比、 四个数都标相关口径，说明整段兜底。 - ledger/match._cells / _cell（algo 4）：整条（含按全部引用列投影）没证明、又引用了这条查询两列以上的，被引用的每一列单独 投影到那一列，和每份口径的每个指标读的那一列（card_reads；口径版本没有口径卡的按取数的每一列）逐个比，仍只走 match_template / explain_sql_diff，上次的投影规则照用：投影不了（分组列、DISTINCT……）、过了时限、不可比的格照用整条 查询的那一个候选。结果记在 queries[].cells {列名: {status, candidates}}，另记 one_row（sql_semantics.one_row：不分组的 聚合，一定只出一行）。整条这一层的结论、候选一字不改；老工件没有 cells，照旧按整条读。 - ledger/labels：单元格按它那一格的结论标（_view）；指标用了同一条查询的好几格时，只有那条查询一定只出一行才把逐格的 结论合起来用（多行的结果里每一列各自对得上，不等于每一行的几列一起对得上）。整题照拍板 L1 不变：每个数都证明来自 同一份口径才等价；一格证明了、别的格只是相关的整题不等价，原因写「有的数和相关口径不一致，不来自口径」（以前一律说 「这次的推理」）。出具入口照 ledger/issue 现有判法只在整题等价时给，没有放宽。 - 逐格证明了的口径各处都算上（labels.candidates_of）：工件 meta 的 calibers（召回按旧版本反查）、ledger/uses 的使用记录、 召回时找原答案对应的那一格、E4；E1 另数各种结论各几格（cells），E2 的兜底比例按标签上给人看的那一份说明数。 - GET /runs/{id}/ledger 的 queries 带上逐格的 cells（遮罩同候选）；证据片段的 ledger 键、轮次摘要的计数本来就按每个数的 标签，数据变了、形状不变。前端只补类型；check-ledger 加一段：同一条查询一格来自口径（正文挂小标记）、一格相关（差在哪 先说多查了几列），整题不等价不给出具。 - 评测逐题比 SQL（eval/score.sql_match）取和期望配上的那一格（score.matched_cells：评分时配上的格），没给时取逐格证明了 的那一格；结果带 column。runner、rescore 传它。 测试：tests/test_ledger_cells.py（同一条查询一列等价、一列不等价各自标对；两份口径只证明了一份标那一份；口径多个指标按列 配对；指标跨两格只在一行的查询上用逐格结论；分组列、DISTINCT 逐格也不投影；照真实例子造的虚构库真跑一遍：销售额标等价 已证明，订单数、首末日标相关，整题不等价、不给出具、不自动出具，证据片段的 ledger 键一致；评测取配上的那一格）。去掉逐格 再跑，10 条失败；去掉「一行才合起来用」、去掉评测取格，各 1 条失败。 --- backend/app/data/sql_semantics.py | 17 +- backend/app/eval/rescore.py | 2 +- backend/app/eval/runner.py | 2 +- backend/app/eval/score.py | 49 ++++- backend/app/ledger/labels.py | 60 +++++- backend/app/ledger/match.py | 110 +++++++++- backend/app/ledger/recall.py | 10 +- backend/app/ledger/stats.py | 18 +- backend/app/ledger/store.py | 9 +- backend/app/ledger/uses.py | 5 +- backend/app/ledger/view.py | 32 +-- backend/tests/test_ledger_cells.py | 265 ++++++++++++++++++++++++ backend/tests/test_ledger_compose.py | 2 +- backend/tests/test_ledger_projection.py | 7 +- frontend/src/typesLedger.ts | 5 + scripts/check-ledger.mjs | 26 ++- scripts/ledger-fixtures.mjs | 18 ++ 17 files changed, 575 insertions(+), 62 deletions(-) create mode 100644 backend/tests/test_ledger_cells.py diff --git a/backend/app/data/sql_semantics.py b/backend/app/data/sql_semantics.py index ecf808e2..465b706a 100644 --- a/backend/app/data/sql_semantics.py +++ b/backend/app/data/sql_semantics.py @@ -9147,6 +9147,19 @@ def project_template(template_sql: str, keep: Sequence[int], dialect: str | None return Projection(None, keep, refused=_PROJ_TEXT) +def one_row(sql: str, dialect: str | None = None) -> bool: + """这条查询一定只出一行（或者一行都不出）：一段 SELECT、没有 GROUP BY、本身是聚合（不分组的聚合，HAVING 至多筛掉那 + 一行）。逐格比出来的结论要合起来用（一个指标用了同一条查询的好几格）时，只在这种查询上成立：多行的结果里，每一列 + 各自对得上不等于每一行的几列一起对得上。认不出的为假。""" + from sqlglot import exp + + read = _read(dialect) + root, _ = _parse(_mysql_dashes(sql) if read == "mysql" and isinstance(sql, str) else sql, read) + if not isinstance(root, exp.Select) or root.args.get("group") is not None or root.args.get("qualify") is not None: + return False + return _aggregating(root) + + def output_names(sql: str, dialect: str | None = None, *, template: bool = False) -> list[str] | None: """最外层选择列表各项的输出名（按位置；同 _result_columns，不展开 *：有 * 的为 None）。template 为真时 sql 是口径 模板，按哨兵渲染之后认。解析不了、不是一段 SELECT 的为 None。""" @@ -9329,5 +9342,5 @@ def sql_shape(sql: str, dialect: str | None = None) -> SqlShape | None: __all__ = ["BINDING_LEVELS", "MAX_SQL_CHARS", "PLACEHOLDER", "Aggregate", "ColumnRef", "Computed", "FacetDiff", "Facets", "JoinFacet", "LimitFacet", "NormalizedSql", "Part", "Projection", "SqlDiff", "SqlShape", "TemplateBinding", "TemplateMatch", "TemplateSlot", "TimePredicate", "bind_template", "catalog_labels", "diff_facets", - "explain_sql_diff", "facets", "match_template", "normalize_sql", "output_names", "project_select", - "project_template", "render_bound", "sql_shape"] + "explain_sql_diff", "facets", "match_template", "normalize_sql", "one_row", "output_names", + "project_select", "project_template", "render_bound", "sql_shape"] diff --git a/backend/app/eval/rescore.py b/backend/app/eval/rescore.py index 355609da..67422777 100644 --- a/backend/app/eval/rescore.py +++ b/backend/app/eval/rescore.py @@ -53,7 +53,7 @@ async def rescore(eval_run_id: str) -> int: answer = await score.read_answer(row.run_id if row.path == "ran" else None, reply=turn.get("reply")) graded = score.score_case(case, row.gold, answer, path=row.path, path_reason=turn.get("reason"), judgement=(row.judge or {}).get("text")) - match = await score.sql_match(case, row.gold, answer) + match = await score.sql_match(case, row.gold, answer, cells=score.matched_cells(graded)) if graded.verdict != row.verdict: changed += 1 updates[row.id] = {"verdict": graded.verdict, "scores": graded.scores, "sql_match": match, diff --git a/backend/app/eval/runner.py b/backend/app/eval/runner.py index 83c21730..dd340d9f 100644 --- a/backend/app/eval/runner.py +++ b/backend/app/eval/runner.py @@ -576,7 +576,7 @@ class EvalExecution(runtime.JobRun): judge_cost += float(judge["text"].get("cost_usd") or 0) graded = score.score_case(case, gold_dict, answer, path=turn.path, path_reason=turn.reason, judgement=judge.get("text")) - match = await score.sql_match(case, gold_dict, answer) + match = await score.sql_match(case, gold_dict, answer, cells=score.matched_cells(graded)) writer = await writer_usage(turn.runs) assistant = add_usage(turn.usage.get("assistant"), turn.usage.get("review")) usage = {"assistant": assistant, "writer": writer, "review": turn.usage.get("review"), diff --git a/backend/app/eval/score.py b/backend/app/eval/score.py index 16fd52ff..792f3d1f 100644 --- a/backend/app/eval/score.py +++ b/backend/app/eval/score.py @@ -613,10 +613,14 @@ async def gold_calibers(case: Mapping[str, Any], gold: Mapping[str, Any] | None) async def sql_match(case: Mapping[str, Any], gold: Mapping[str, Any] | None, answer: Answer, *, - timeout: float = 8.0) -> dict[str, Any] | None: - """逐题的 E1：{status: proven|related|none|skipped, query, aspects, diff, reason, cited, basis}。没有作答运行、没有期望 - 取数的为 None。比的是哪条查询见 _answering：basis 是 cited（答案引用的）或 only_query（答案没引用、运行只查了这一条）。 - 比对在线程里跑（纯计算，和账本一样限时）。""" + timeout: float = 8.0, cells: Sequence[Any] = ()) -> dict[str, Any] | None: + """逐题的 E1：{status: proven|related|none|skipped, query, column?, aspects, diff, reason, cited, basis}。没有作答运行、 + 没有期望取数的为 None。比的是哪条查询见 _answering：basis 是 cited（答案引用的）或 only_query（答案没引用、运行只查了 + 这一条）。比对在线程里跑（纯计算，和账本一样限时）。 + + 和账本同一个口径逐格判（ledger/match._cells）：引用了好几列、整条比不上的查询，每一列各比一次。cells 是评分时和期望 + 配上的那几格（各项期望的 cell：{artifact, column}）：取那一格的结论（column 写是哪一列）；没给、配上的格没逐格比过时， + 逐格证明了的取那一格，再不然照旧按整条查询。""" import asyncio import time @@ -653,8 +657,9 @@ async def sql_match(case: Mapping[str, Any], gold: Mapping[str, Any] | None, ans pool, basis = _answering(rows, answer) if isinstance(pool, dict): return pool + column, cell = _cell_of(pool, cells) proven = next((r for r in pool if r.get("status") == "proven"), None) - chosen = proven or next((r for r in pool if r.get("status") == "related"), None) or \ + chosen = cell or proven or next((r for r in pool if r.get("status") == "related"), None) or \ next((r for r in pool if r.get("status") == "skipped"), None) or pool[0] candidate = next((c for c in chosen.get("candidates") or [] if c.get("status") == chosen.get("status")), None) or \ next(iter(chosen.get("candidates") or []), None) or {} @@ -662,9 +667,34 @@ async def sql_match(case: Mapping[str, Any], gold: Mapping[str, Any] | None, ans # projection 那一条只是说「多查了几列，不影响这个数」，不是差在哪个方面（账本的 E1 差异类型也不数它） aspects = sorted({i.get("aspect") for i in diff.get("items") or [] if i.get("aspect") not in (None, "", "projection")}) sql = next((q.get("sql") for q in answer.queries if q.get("artifact") == chosen.get("artifact")), None) - return {"status": chosen.get("status"), "query": chosen.get("artifact"), "sql": sql, "aspects": aspects, - "diff": diff or None, "via": candidate.get("via"), "binding": candidate.get("binding"), - "reason": chosen.get("reason") or candidate.get("reason"), "cited": basis == "cited", "basis": basis} + return {"status": chosen.get("status"), "query": chosen.get("artifact"), **({"column": column} if column else {}), + "sql": sql, "aspects": aspects, "diff": diff or None, "via": candidate.get("via"), + "binding": candidate.get("binding"), "reason": chosen.get("reason") or candidate.get("reason"), + "cited": basis == "cited", "basis": basis} + + +def matched_cells(graded: "CaseScore") -> list[dict[str, Any]]: + """评分时各项期望和答案里配上的那一格（答对的、量纲不同的、答错时配对上的）：逐题 E1 取它们的逐格结论。""" + return [dict(item["cell"]) for item in graded.scores + if isinstance(item, Mapping) and isinstance(item.get("cell"), dict)] + + +def _cell_of(pool: list[dict[str, Any]], cells: Sequence[Any]) -> tuple[str | None, dict[str, Any] | None]: + """逐格比过的格里挑一格：评分时和期望配上的那几格先（其中证明了的先），再不然逐格证明了的那一格。交回 (列名, 那一格的 + 结论，带上查询快照 artifact)；没有可挑的为 (None, None)。""" + hinted = [(str(c.get("artifact")), str(c.get("column"))) for c in cells + if isinstance(c, Mapping) and c.get("artifact") and c.get("column") is not None] + found: list[tuple[str, dict[str, Any]]] = [] + for row in pool: + for name, entry in (row.get("cells") or {}).items() if isinstance(row.get("cells"), dict) else []: + if isinstance(entry, dict): + found.append((name, {**entry, "artifact": row.get("artifact"), "_hinted": (row.get("artifact"), name) + in hinted})) + ranked = sorted(found, key=lambda f: (not f[1]["_hinted"], f[1].get("status") != "proven")) + for name, entry in ranked: + if entry["_hinted"] or entry.get("status") == "proven": + return name, {k: v for k, v in entry.items() if k != "_hinted"} + return None, None def _answering(rows: list[dict[str, Any]], answer: Answer) -> tuple[list[dict[str, Any]] | dict[str, Any], str]: @@ -692,4 +722,5 @@ def _answering(rows: list[dict[str, Any]], answer: Answer) -> tuple[list[dict[st __all__ = ["Answer", "CaseScore", "DEFAULT_TOLERANCE", "OUTCOMES", "OUTCOME_LABEL", "PATH_LABEL", "SCORER_VERSION", - "TEXT_LABEL", "TEXT_OUTCOMES", "VERDICT_LABEL", "gold_calibers", "read_answer", "score_case", "sql_match"] + "TEXT_LABEL", "TEXT_OUTCOMES", "VERDICT_LABEL", "gold_calibers", "matched_cells", "read_answer", + "score_case", "sql_match"] diff --git a/backend/app/ledger/labels.py b/backend/app/ledger/labels.py index 14be0c3b..e876f6c2 100644 --- a/backend/app/ledger/labels.py +++ b/backend/app/ledger/labels.py @@ -8,10 +8,14 @@ - unknown：没比对、比对没做完（超时、解析失败）、快照读不出来。 怎么认： -- 单元格（[[v:]]、[[table:]] 里的格）：按证据目录找到查询快照，取那条查询的比对状态。等价时按列映射写出口径那边的列名。 +- 单元格（[[v:]]、[[table:]] 里的格）：按证据目录找到查询快照，取那条查询的比对状态；这条查询逐格比过（algo 4：整条按全部 + 引用列投影没证明时，被引用的每一列单独投影、和口径的每个指标列逐个比，ledger/match._cells）的，按这一格所在的那一列的 + 结论——同一条查询里一列等价、一列不等价各标各的。等价时按列映射写出口径那边的列名。 - 指标（[[m:]]）：指标集工件里这个指标的每个输入都要来自查询快照，而且这些快照都被证明等价于同一份口径的同一版、绑定一致； 指标式还要和那一版里某张口径卡上的某个指标相同——两边的式子把每个输入换成「口径的哪个取数节点、第几行、哪一列」之后 一字不差（caliber_from 转抄来的式子本来就是那一张卡的，同样按这个比）。输入有不来自查询的（运行输入、整形、代码）不算来自口径。 + 逐格比过的输入按它那一格的结论；同一条查询的好几格合起来用时，只在这条查询一定只出一行（one_row）时才按逐格的结论， + 别的按整条查询：多行的结果里每一列各自对得上，不等于每一行的几列一起对得上。 整题（whole_of）：主报告里所有带标签的数都来自同一口径的同一版、绑定一致，就记 equivalent，并写出还缺的入口字段（口径入口 节点声明了、这次的绑定里没有的；required 且没有缺省值的另列）。 @@ -92,11 +96,38 @@ def _caliber_column(cand: Mapping[str, Any], column: Any) -> str | None: return None +def cells_of(query: Mapping[str, Any]) -> list[Mapping[str, Any]]: + """这条查询逐格比出来的几格（algo 4 的 cells，见 ledger/match._cells）；老工件、没逐格比的为空。""" + cells = query.get("cells") + return [c for c in cells.values() if isinstance(c, Mapping)] if isinstance(cells, Mapping) else [] + + +def candidates_of(query: Mapping[str, Any], *, shown: bool = False) -> list[Mapping[str, Any]]: + """一条查询的候选：整条比的，加上逐格比出来的。召回反查、使用次数、工件 meta 都要算上逐格证明了的那几份口径。shown 为真 + 时只要标签上给人看的那一份：逐格比过的只要逐格的（整条的那一份不在标签上），别的照旧是整条的（统计兜底说明的比例用）。""" + cells = [c for cell in cells_of(query) for c in cell.get("candidates") or [] if isinstance(c, Mapping)] + whole = [c for c in query.get("candidates") or [] if isinstance(c, Mapping)] + if shown: + return cells or whole + return whole + cells + + +def _view(query: Mapping[str, Any], column: Any, *, alone: bool = True) -> Mapping[str, Any]: + """这一格按什么判：这条查询逐格比过这一列（cells 里有）的按那一格的结论，别的按整条查询的。alone 为假（一个指标用了 + 这条查询的好几格）时，只有这条查询一定只出一行（one_row）才按逐格的（见模块说明）。老工件没有 cells，照旧按整条。""" + cells = query.get("cells") + cell = cells.get(column) if isinstance(cells, Mapping) and isinstance(column, str) else None + if isinstance(cell, Mapping) and (alone or query.get("one_row") is True): + return cell + return query + + def cell_label(artifact: str, column: Any, queries: Mapping[str, Mapping[str, Any]]) -> dict[str, Any]: - """单元格的标签：它所在的那条查询的比对状态。""" - query = queries.get(artifact) - if query is None: + """单元格的标签：它所在的那条查询的比对状态（逐格比过的按这一格，_view）。""" + found = queries.get(artifact) + if found is None: return _unknown(NO_RECORD) + query = _view(found, column) status = query.get("status") if status in ("proven", "composed"): cand = _proven(query)[0] if _proven(query) else {} @@ -247,9 +278,18 @@ def metric_label(card: Mapping[str, Any] | None, metric_id: Any, queries: Mappin from_queries = [i for i in inputs if i.get("artifact") and i.get("via") in _SNAPSHOT_VIAS] if not from_queries: return {"kind": "reasoning"} - found = [queries.get(str(i["artifact"])) for i in from_queries] - if any(q is None for q in found): + rows = [queries.get(str(i["artifact"])) for i in from_queries] + if any(q is None for q in rows): return _unknown(NO_RECORD) + # 每个输入按它那一格判（_view）：同一条查询用了好几格的，那条查询一定只出一行才按逐格的结论 + used: dict[str, set[Any]] = {} + for item in from_queries: + where = item.get("locator") if isinstance(item.get("locator"), dict) else {} + used.setdefault(str(item["artifact"]), set()).add(where.get("column")) + found: list[Mapping[str, Any] | None] = [ + _view(q, (i.get("locator") if isinstance(i.get("locator"), dict) else {}).get("column"), + alone=len(used[str(i["artifact"])]) == 1) if q is not None else None + for q, i in zip(rows, from_queries)] if any(q.get("status") == "skipped" for q in found): # type: ignore[union-attr] return _unknown(next(str(q.get("reason") or UNFINISHED) for q in found if q.get("status") == "skipped")) # type: ignore[union-attr] @@ -353,8 +393,9 @@ def whole_of(labels: Mapping[str, Mapping[str, Any]], calibers: Mapping[str, Map return {"equivalent": False, "reason": "报告里没有可标口径的数", "counts": counts} keys = {label.get("caliber") for label in labels.values() if label.get("kind") == "caliber"} if counts["caliber"] != len(labels): - reason = ("有的数没有比对" if counts["unknown"] else "有的数是这次的推理" if counts["caliber"] - else "没有数来自口径") + # 逐格判之后常见「一部分数来自口径、另几个只是相关」：照实说是不一致，不说成这次的推理 + reason = ("有的数没有比对" if counts["unknown"] else "没有数来自口径" if not counts["caliber"] + else "有的数和相关口径不一致，不来自口径" if counts["related"] else "有的数是这次的推理") return {"equivalent": False, "reason": reason, "counts": counts} if len(keys) != 1: return {"equivalent": False, "reason": "数来自不同的口径", "counts": counts} @@ -431,5 +472,6 @@ def related_view(item: Mapping[str, Any], *, hide: Callable[[Any], bool], shown: __all__ = ["CERTIFIED_NOTE", "HOWS", "KINDS", "LEVEL_TEXT", "POST_SEAL_NOTE", "RETRO_NOTE", "binding_view", "caliber_key", - "card_expressions", "card_reads", "cell_label", "diff_view", "label_doc", "metric_label", + "candidates_of", "card_expressions", "card_reads", "cell_label", "cells_of", "diff_view", "label_doc", + "metric_label", "referenced_columns", "related_view", "view_label", "whole_of"] diff --git a/backend/app/ledger/match.py b/backend/app/ledger/match.py index 2a128b40..f5d72135 100644 --- a/backend/app/ledger/match.py +++ b/backend/app/ledger/match.py @@ -25,6 +25,14 @@ MAX_CANDIDATES 份（按工作流、版本、节点排好的先后）。 答案多选了列的在最前面写一句「这次多查了 N 列，不影响这个数」）。没有投影可试的（报告没引用这条查询、两边都不投影、 投影不了）照旧是整条比的结果，逐对一字不差。 +**逐格判**（_cells，algo 4）。报告常把一条查询的好几列都引用了（销售额、订单数、区间首日、末日），口径只取其中一列：按全部 +引用列投影还是好几列，对不上口径，整条比只能标相关。整条（含按全部引用列投影）没证明、又引用了这条查询两列以上的，被引用 +的每一列单独投影到那一列（_cell），和每份口径的每个指标读的那一列逐个比，照样只走 match_template——上面的投影规则一条 +不少（多一个分组列、DISTINCT 的照旧不投影，那一格照用整条的结论）。结果记在 queries[].cells {列名: {status, candidates}}， +每个数按它那一格标（labels._view）；同一个指标用了这条查询的好几格时，只有这条查询一定只出一行（one_row）才把逐格的结论 +合起来用。整题照拍板 L1：每个数都证明来自同一份口径才算等价，一格证明了、别的格只是相关的，整题不等价、不自动出具。召回 +反查、使用次数、工件 meta 把逐格证明了的口径也算上（labels.candidates_of）。 + **助手选用的（composed）。** 问数据的助手 use_caliber 选用口径时，取数节点是服务端照那一版拷进来的（ledger/compose.py）， 口径卡的 caliber_from 带着改名映射 {原 id: 新 id}。这几个节点交回的查询照样要证明：证明了的那一份记 how=composed，查询 状态记 composed——数确实来自口径，但「这份口径答的就是这个问题」是助手凭问法判断的，不能说成「等价已证明」（拍板 L1： @@ -447,6 +455,89 @@ def _projected(caliber: Caliber, fetch: Fetch, query: Query, wanted: tuple[int, return cand +def _cell(caliber: Caliber, fetch: Fetch, query: Query, index: int, deadline: float | None) -> dict[str, Any] | None: + """逐格比（algo 4，见模块说明）：答案只留这一格所在的那一列（project_select），口径只留它的指标读到的某一列 + （口径卡上读这个取数节点的每一列各试一次；口径版本里没有口径卡的，取数选出来的每一列各试一次），投影后照样交 + match_template 判。证明了的候选（带 projection，列映射就是这一列对口径那一列）；都没证明时第一种比法的那次比较 + （差在哪按投影之后说，答案多选了列的先写一句）；投影不了、没有可比的为 None（这一格照旧用这条查询的结论）。""" + from app.data.sql_semantics import FacetDiff, match_template, output_names, project_select, project_template + + width = len(query.columns or []) + ours = project_select(query.sql, (index,), query.dialect, width=width) + names = output_names(fetch.sql, query.dialect, template=True) + if ours.sql is None or not names: + return None + if caliber.cards: + columns = sorted({i for group in _caliber_sets(caliber, fetch, names) for i in group}) + else: + columns = list(range(len(names))) + theirs = [project_template(fetch.sql, (i,), query.dialect) for i in columns] if len(names) > 1 \ + else [project_template(fetch.sql, (0,), query.dialect)] if columns == [0] else [] + first: tuple[Any, Any] | None = None + for c in theirs[:MAX_PROJECTIONS]: + if c.sql is None or (deadline is not None and first is not None and time.monotonic() > deadline): + continue + m = match_template(c.sql, list(caliber.fields), ours.sql, query.dialect, schema=query.schema, + names=query.names, columns=[str((query.columns or [])[index])], deadline=deadline) + if m.equivalent: + return {**_candidate(caliber, fetch, m), "projection": _projection_view(query, ours, c)} + if first is None: + first = (c, m) + if first is None: + return None + c, m = first + cand = {**_candidate(caliber, fetch, m), "projection": _projection_view(query, ours, c)} + if ours.dropped and isinstance(cand.get("diff"), dict): + note = FacetDiff("projection", None, None, EXTRA_COLUMNS.format(n=ours.dropped)).as_dict() + cand["diff"] = {**cand["diff"], "items": [note, *cand["diff"]["items"]]} + return cand + + +def _verdict(candidates: list[dict[str, Any]]) -> tuple[str, str | None]: + """一组候选的结论（查询、逐格同一个判法）：(status, 没比完时的原因)。""" + statuses = {c["status"] for c in candidates} + status = "proven" if "proven" in statuses else "skipped" if "skipped" in statuses else "related" + if status == "proven" and any(c.get("how") == "composed" for c in candidates if c["status"] == "proven"): + # 证明了的是助手选用的那一份：数来自口径，但不是这次独立写出来的（见模块说明）。同时和别的口径等价的也按选用的算 + status = "composed" + reason = next((c.get("reason") for c in candidates if c["status"] == "skipped"), None) if status == "skipped" \ + else None + return status, reason + + +def _cells(query: Query, wanted: tuple[int, ...], pool: list[tuple[Caliber, Fetch]], whole: list[dict[str, Any]], + chosen: tuple[str, str] | None, deadline: float | None) -> dict[str, dict[str, Any]]: + """这条查询被引用的每一列各比一次（_cell）：{列名: {status, candidates, reason?}}。每份口径取数一个候选：逐格比出来的， + 比不了（投影不了、过了时限、不可比、整条比就没比完）的照用这条查询的那一个候选。同名的几列认不出是哪一列，不逐格比。""" + columns = [str(c) for c in query.columns or []] + out: dict[str, dict[str, Any]] = {} + for index in wanted: + name = columns[index] + if columns.count(name) != 1: + continue + candidates: list[dict[str, Any]] = [] + for (caliber, fetch), base in zip(pool, whole): + cand = None + if base.get("status") == "related" and base.get("comparable", True) and \ + (deadline is None or time.monotonic() <= deadline): + try: + cand = _cell(caliber, fetch, query, index, deadline) + except Exception: # noqa: BLE001 - 逐格比是附加的一步：出错这一格照用这条查询的结论 + logger.exception("逐格比对出错（运行查询 %s，口径 %s）", query.artifact[:12], caliber.key) + if cand is None: + candidates.append(dict(base)) + continue + if chosen is not None and (caliber.key, fetch.node_id) == chosen: + if cand["status"] == "proven": + cand["how"] = "composed" + else: + cand.setdefault("reason", COMPOSED_UNPROVEN) + candidates.append(cand) + status, reason = _verdict(candidates) + out[name] = {"status": status, "candidates": candidates, **({"reason": reason} if reason else {})} + return out + + def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: float | None = None, composed: dict[str, tuple[str, str]] | None = None, referenced: Mapping[str, Collection[str]] | None = None) -> list[dict[str, Any]]: @@ -455,7 +546,7 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: 等价）/ none（没有相关的口径）/ skipped（没比完：超时、解析失败、快照读不出）。composed 是 composed_nodes 的结果， 不给就没有助手选用的。referenced 是报告引用了哪条查询的哪几列（labels.referenced_columns：{查询快照: {列名}}）： 给了的，整条比不等价时按引用的列投影再比（_projected）；不给就只整条比。纯计算，从不抛异常。评测逐题比 SQL 也调它。""" - from app.data.sql_semantics import match_template, normalize_sql + from app.data.sql_semantics import match_template, normalize_sql, one_row out: list[dict[str, Any]] = [] for query in queries: @@ -500,16 +591,17 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: else: cand.setdefault("reason", COMPOSED_UNPROVEN) candidates.append(cand) - statuses = {c["status"] for c in candidates} - status = "proven" if "proven" in statuses else "skipped" if "skipped" in statuses else "related" - if status == "proven" and any(c.get("how") == "composed" for c in candidates): - # 证明了的是助手选用的那一份：数来自口径，但不是这次独立写出来的（见模块说明）。同时和别的口径等价的也按选用的算 - status = "composed" + status, reason = _verdict(candidates) if len(pool) > MAX_CANDIDATES: row["note"] = MORE_CANDIDATES.format(n=len(pool) - MAX_CANDIDATES, max=MAX_CANDIDATES) - if status == "skipped": - row["reason"] = next(c.get("reason") for c in candidates if c["status"] == "skipped") - out.append({**row, "status": status, "candidates": candidates}) + if reason: + row["reason"] = reason + row.update(status=status, candidates=candidates) + if wanted is not None and len(wanted) >= 2 and status == "related": + # 整条（按全部引用列投影）没证明：被引用的几列逐格各比一次，每个数按它自己那一格的结论标（labels.cell_label） + row["cells"] = _cells(query, wanted, pool[:MAX_CANDIDATES], candidates, chosen, deadline) + row["one_row"] = one_row(query.sql, query.dialect) + out.append(row) return out diff --git a/backend/app/ledger/recall.py b/backend/app/ledger/recall.py index 938a81b4..62bf2dbe 100644 --- a/backend/app/ledger/recall.py +++ b/backend/app/ledger/recall.py @@ -871,12 +871,14 @@ async def original_values(use: Mapping[str, Any], meta: Sequence[Mapping[str, An from app.engine.evidence import render_number from app.engine.expressions import cell_parts from app.ledger import store + from app.ledger.labels import candidates_of recorded = await store.latest(run_id) content = recorded.content if recorded is not None else None query = next((q for q in (content or {}).get("queries") or [] if q.get("artifact") == use.get("query")), None) - cand = next((c for c in (query or {}).get("candidates") or [] if c.get("status") == "proven" - and c.get("node_id") == use.get("node_id")), None) + # 整条证明了的、逐格证明了的都算：口径的哪一列对这次的哪一列，按各自的列映射找（逐格的只映射那一列） + proven = [c for c in candidates_of(query or {}) + if c.get("status") == "proven" and c.get("node_id") == use.get("node_id")] try: snap = load(str(use.get("query") or "")) if use.get("query") else None except ValueError: @@ -893,8 +895,8 @@ async def original_values(use: Mapping[str, Any], meta: Sequence[Mapping[str, An column = ast.literal_eval(parts[2]) except (ValueError, SyntaxError): column = None - ours = next((m.get("ours") for m in (cand or {}).get("columns") or [] if m.get("caliber") == column), None) \ - if isinstance(column, str) else None + ours = next((m.get("ours") for c in proven for m in c.get("columns") or [] if m.get("caliber") == column), + None) if isinstance(column, str) else None if ours in columns and isinstance(rows, list) and rows and isinstance(rows[0], (list, tuple)): value = rows[0][columns.index(ours)] try: diff --git a/backend/app/ledger/stats.py b/backend/app/ledger/stats.py index 4c4ead14..2f405e1e 100644 --- a/backend/app/ledger/stats.py +++ b/backend/app/ledger/stats.py @@ -9,9 +9,10 @@ - E1 等价命中：比对过的查询里 proven / composed / related / none / skipped 各几条（skipped 单独列，不混进 none），有候选口径 的查询里各自的比率；证明等价的绑定是在哪一级配上的，其中几份是按引用的列投影之后才证明的（proven_projected，ledger/match - 的 _projected）；没证明的原因（不可比、参数绑不上、配上了但不等价）；不等价的差异类型分布（time / filter / aggregate / - join / other / parse，每份口径的差异每类只计一次；投影之后那句「这次多查了 N 列」不是差异，不计）；逐个数的标签各几个、 - 整题等价几次。 + 的 _projected）；逐格比过的格（algo 4，ledger/match._cells）各种结论各几格（cells）；没证明的原因（不可比、参数绑不上、 + 配上了但不等价）；不等价的差异类型分布（time / filter / aggregate / join / other / parse，每份口径的差异每类只计一次； + 投影之后那句「这次多查了 N 列」不是差异，不计）；逐个数的标签各几个、整题等价几次。 + E2 的兜底比例按标签上给人看的那一份说明数：逐格比过的按逐格的（labels.candidates_of 的 shown）。 - E2 对账：对账运行（origin=reconcile，第 3 波起由 ledger/reconcile 发起）发起、完成、失败各几次，对过账的答案几个，其中参数 经人核对过的、对照的指标由人选的各几次；相关口径的差异说明里兜底说明的占比；对账结果上「说明有用 / 看不懂」的反馈 （第 4 波，ledger/feedback：对账运行上的封存后事件 ledger.feedback，每人只算最后一次）各几个、有用的比率。 @@ -103,6 +104,7 @@ def e1_of(contents: list[dict[str, Any]]) -> dict[str, Any]: with_candidates = Counter({s: 0 for s in STATUSES}) via: Counter[str] = Counter() projected = 0 + cells = Counter({s: 0 for s in STATUSES}) unproven: Counter[str] = Counter() diffs = Counter({g: 0 for g in DIFF_GROUPS}) labels = Counter({k: 0 for k in ledger_labels.KINDS}) @@ -113,6 +115,8 @@ def e1_of(contents: list[dict[str, Any]]) -> dict[str, Any]: queries += 1 s = str(query.get("status") or "") status[s] += 1 + for cell in ledger_labels.cells_of(query): + cells[str(cell.get("status") or "")] += 1 if query.get("candidates"): with_candidates[s] += 1 for cand in query.get("candidates") or []: @@ -142,7 +146,8 @@ def e1_of(contents: list[dict[str, Any]]) -> dict[str, Any]: "runs": len(contents), "queries": queries, "status": dict(status), "with_candidates": candidates, "rates": {s: _ratio(with_candidates[s], candidates) for s in STATUSES if s != "none"}, - "proven_via": dict(via), "proven_projected": projected, "unproven_reasons": dict(unproven.most_common(20)), + "cells": dict(cells), "proven_via": dict(via), "proven_projected": projected, + "unproven_reasons": dict(unproven.most_common(20)), "diff_aspects": dict(diffs), "labels": dict(labels), "whole_equivalent": whole, } @@ -169,7 +174,8 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic explained = fallback = 0 for content in contents: for query in content.get("queries") or []: - for cand in query.get("candidates") or []: + # 标签上给人看的那一份：逐格比过的按逐格的说明数（ledger_labels.candidates_of 的 shown） + for cand in ledger_labels.candidates_of(query, shown=True): if cand.get("status") == "related" and isinstance(cand.get("diff"), dict): explained += 1 fallback += bool(cand["diff"].get("fallback")) @@ -285,7 +291,7 @@ async def _e4(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic proven: Counter[str] = Counter() for content in contents: - keys = {c.get("caliber") for q in content.get("queries") or [] for c in q.get("candidates") or [] + keys = {c.get("caliber") for q in content.get("queries") or [] for c in ledger_labels.candidates_of(q) if c.get("status") == "proven" and c.get("caliber")} for key in keys: proven[str(key)] += 1 diff --git a/backend/app/ledger/store.py b/backend/app/ledger/store.py index abe1717c..e4fa92ce 100644 --- a/backend/app/ledger/store.py +++ b/backend/app/ledger/store.py @@ -32,7 +32,9 @@ FORMAT = "agentlab.caliber_match/1" #: 2（第 3 波）：认出助手选用的口径（查询 composed、标签 how=composed），指标的标签记它取自哪几条查询（queries，认证队列 #: 认「被报告引用的查询」用） #: 3：整条比不等价时按报告引用的列投影再比（ledger/match 的 _projected），候选记 projection；没投影的照旧 -ALGO = "3" +#: 4：按全部引用列投影仍没证明的查询，被引用的每一列逐格各比一次（queries[].cells、one_row），每个数按它那一格标；老工件 +#: 没有 cells，照旧按整条查询读 +ALGO = "4" #: 触发方式（见模块说明）。certified（第 3 波）：从问数据固化出的口径发布之后，对来源运行补比的（ledger/certify） TRIGGERS = ("sealed", "backfill", "on_demand", "certified") @@ -73,10 +75,13 @@ def counts_of(queries: list[dict[str, Any]]) -> dict[str, int]: def meta_of(content: dict[str, Any]) -> dict[str, Any]: """工件引用的 meta：{calibers: [等价已证明的「工作流@版本」], related: [相关的], counts}。""" + from app.ledger.labels import candidates_of + proven: list[str] = [] related: list[str] = [] for query in content.get("queries") or []: - for cand in query.get("candidates") or []: + # 逐格证明了的口径也算（召回按旧版本反查、统计按种类数都只读这一份 meta） + for cand in candidates_of(query): key = str(cand.get("caliber") or "") if not key: continue diff --git a/backend/app/ledger/uses.py b/backend/app/ledger/uses.py index 1bfdd825..b473e110 100644 --- a/backend/app/ledger/uses.py +++ b/backend/app/ledger/uses.py @@ -110,7 +110,7 @@ async def _answers(session: AsyncSession, workflow_id: str, below: int | None, o """问数回答：比对工件 meta 里等价已证明的那几份口径里有 W@V 的探索运行（见模块说明）。""" from app.engine.origin import counted_runs from app.ledger import store - from app.ledger.labels import caliber_key + from app.ledger.labels import caliber_key, candidates_of prefix = f"{workflow_id}@" rows = (await session.execute( @@ -134,7 +134,8 @@ async def _answers(session: AsyncSession, workflow_id: str, below: int | None, o continue seen: set[tuple[int, str]] = set() for query in content.get("queries") or []: - for cand in query.get("candidates") or []: + # 逐格证明了的也是用了这份口径（ledger/labels.candidates_of） + for cand in candidates_of(query): if cand.get("status") != "proven" or cand.get("workflow_id") != workflow_id: continue version = cand.get("version") diff --git a/backend/app/ledger/view.py b/backend/app/ledger/view.py index 6ea951d5..2c72c336 100644 --- a/backend/app/ledger/view.py +++ b/backend/app/ledger/view.py @@ -78,21 +78,27 @@ def _whole_view(whole: dict[str, Any], hider: _Hide, names: dict[str, str]) -> d return out +def _candidate_view(cand: dict[str, Any], hider: _Hide, names: dict[str, str]) -> dict[str, Any]: + item = ledger_labels.related_view(cand, hide=hider.hide, shown=hider.shown, names=names) + item["status"] = cand.get("status") + if cand.get("how"): + item["how"] = cand["how"] + if cand.get("columns") is not None: + item["columns"] = cand["columns"] + if isinstance(cand.get("projection"), dict): + # 按引用的列投影之后比的（ledger/match 的 _projected、_cell）：两边各留下了哪几列、去掉了几列（列名，不带值） + item["projection"] = cand["projection"] + return item + + def _query_view(query: dict[str, Any], hider: _Hide, names: dict[str, str]) -> dict[str, Any]: out = {k: query[k] for k in ("artifact", "node_id", "source", "tables", "status", "reason", "note") if k in query} - cands = [] - for cand in query.get("candidates") or []: - item = ledger_labels.related_view(cand, hide=hider.hide, shown=hider.shown, names=names) - item["status"] = cand.get("status") - if cand.get("how"): - item["how"] = cand["how"] - if cand.get("columns") is not None: - item["columns"] = cand["columns"] - if isinstance(cand.get("projection"), dict): - # 按引用的列投影之后比的（ledger/match 的 _projected）：两边各留下了哪几列、去掉了几列（列名，不带值） - item["projection"] = cand["projection"] - cands.append(item) - out["candidates"] = cands + out["candidates"] = [_candidate_view(c, hider, names) for c in query.get("candidates") or []] + if isinstance(query.get("cells"), dict): + # 逐格比过的（algo 4）：每一列的结论和候选，标签按它标；遮罩同上 + out["cells"] = {column: {**{k: cell[k] for k in ("status", "reason") if k in cell}, + "candidates": [_candidate_view(c, hider, names) for c in cell.get("candidates") or []]} + for column, cell in query["cells"].items() if isinstance(cell, dict)} return out diff --git a/backend/tests/test_ledger_cells.py b/backend/tests/test_ledger_cells.py new file mode 100644 index 00000000..fd34279d --- /dev/null +++ b/backend/tests/test_ledger_cells.py @@ -0,0 +1,265 @@ +"""口径标签逐格判（app/ledger/match._cells、labels._view，algo 4）。 + +真实模型下问「上周的销售额是多少」，助手写 SUM(amount) AS 销售额, COUNT(*) AS 订单数, MIN / MAX(paid_date) AS 区间首日 / 末日， +报告四列都引用了；口径「周销售」只取 SUM(amount)。以前按「全部引用列」投影后还是 4 列，对不上口径的 1 列，整条比、四个数都 +标相关口径，说明整段兜底。现在每个被引用的格单独投影到那一列，和口径每个指标读的那一列逐个比（仍只走 match_template / +explain_sql_diff，上次的投影规则照用）： +- 销售额标「等价已证明」，订单数、首末日照实标相关口径；整题不等价（订单数不来自口径），不自动出具、不给出具入口； +- 同一个数在两份口径里只证明了一份时标那一份；口径有多个指标时按列配对；一个指标用了同一条一行查询的两格，两格各自 + 证明了就算来自口径，多行的查询不拿逐格的结论合起来用； +- 不错给信任：多一个分组列、DISTINCT 的查询逐格也不投影； +- 评测逐题比 SQL 取和期望配上的那一格。 +数据是虚构的。 +""" +from __future__ import annotations + +import sqlite3 +import uuid +from types import SimpleNamespace + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.engine.runner import run_manager +from app.ledger import labels as ledger_labels +from app.ledger import match as ledger_match +from app.ledger import store +from tests import ledger_samples as S + +#: 口径「周销售」的取数（照真实例子：按支付日期、两头都含） +WEEKLY = ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' " + "AND paid_date >= '{{ input.start }}' AND paid_date <= '{{ input.end }}'") +WINDOW = "status = 'paid' AND paid_date >= '2026-10-05' AND paid_date <= '2026-10-11'" +#: 助手写的（照真实例子） +ANSWER = ("SELECT SUM(amount) AS 销售额, COUNT(*) AS 订单数, MIN(paid_date) AS 区间首日, MAX(paid_date) AS 区间末日 " + f"FROM orders WHERE {WINDOW}") +COLUMNS = ["销售额", "订单数", "区间首日", "区间末日"] +ROWS = [(1, "paid", 120.0, "2026-10-04"), (2, "paid", 80.0, "2026-10-05"), (3, "paid", 45.5, "2026-10-08"), + (4, "refunded", 30.0, "2026-10-09"), (5, "paid", 64.5, "2026-10-11"), (6, "open", 10.0, "2026-10-10"), + (7, "paid", 99.0, "2026-10-12")] + + +async def week_source(tmp_path) -> None: + path = tmp_path / "week.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, status TEXT, amount REAL, paid_date TEXT);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", ROWS) + db.commit() + db.close() + S.CURRENT["source"] = f"week_{uuid.uuid4().hex[:8]}" + await S.set_source(path) + + +def caliber(sql: str = WEEKLY, metrics: list[dict] | None = None, workflow_id: str = "wf") -> ledger_match.Caliber: + graph = S.caliber_graph(sql) + if metrics is not None: + graph["nodes"][2]["data"]["config"]["metrics"] = metrics + found = ledger_match.caliber_of(workflow_id, SimpleNamespace(graph=graph, version=1, level="governed", + graph_hash="h")) + assert found is not None + return found + + +def compare(sql: str, columns: list[str], cited: set[str], calibers: list[ledger_match.Caliber] | None = None) -> dict: + query = ledger_match.Query("q1", "q", S.CURRENT["source"], sql, columns, frozenset({"orders"}), "sqlite") + [row] = ledger_match.compare_queries([query], calibers or [caliber()], referenced={"q1": cited}) + return row + + +def label(row: dict, column: str) -> dict: + return ledger_labels.cell_label("q1", column, {"q1": row}) + + +def texts(cand: dict) -> list[str]: + return [i["text"] for i in (cand.get("diff") or {}).get("items") or []] + + +# -------------------------------------------------------------------------- +# 逐格判：同一条查询里一列等价、一列不等价各自标对 +# -------------------------------------------------------------------------- + + +def test_each_cited_column_is_judged_on_its_own(): + row = compare(ANSWER, COLUMNS, set(COLUMNS)) + assert row["status"] == "related", "按全部引用列投影还是 4 列，整条照旧不等价" + assert row["one_row"] is True and set(row["cells"]) == set(COLUMNS) + sales = label(row, "销售额") + assert sales["kind"] == "caliber" and sales["how"] == "independent" and sales["column"] == "gmv" + assert sales["binding"] == {"start": "2026-10-05", "end": "2026-10-11"} + assert [label(row, c)["kind"] for c in COLUMNS[1:]] == ["related", "related", "related"] + [orders] = row["cells"]["订单数"]["candidates"] + assert texts(orders) == ["这次多查了 3 列，不影响这个数", "这次数的是行数，口径对 amount 求和"] + assert orders["diff"]["fallback"] is False + # 整题：订单数不来自口径，不等价 + whole = ledger_labels.whole_of({c: label(row, c) for c in COLUMNS}, {}) + assert whole["equivalent"] is False and whole["counts"]["caliber"] == 1 + assert whole["reason"] == "有的数和相关口径不一致，不来自口径" + # 只引用销售额一格（不逐格）：整条按引用的列投影就证明了，照旧 + single = compare(ANSWER, COLUMNS, {"销售额"}) + assert single["status"] == "proven" and "cells" not in single + + +def test_a_number_proven_against_only_one_of_two_calibers_is_labelled_with_that_one(): + paid = caliber(workflow_id="wf-paid") + everything = caliber(WEEKLY.replace("status = 'paid' AND ", ""), workflow_id="wf-all") + row = compare(ANSWER, COLUMNS, set(COLUMNS), [paid, everything]) + sales = label(row, "销售额") + assert sales["kind"] == "caliber" and sales["caliber"] == "wf-paid@1" + cands = {c["caliber"]: c for c in row["cells"]["销售额"]["candidates"]} + assert cands["wf-paid@1"]["status"] == "proven" and cands["wf-all@1"]["status"] == "related" + assert any("status = 'paid'" in t for t in texts(cands["wf-all@1"])) + # 反过来：含退款的写法只对得上另一份 + refunds = ANSWER.replace("status = 'paid' AND ", "") + other = label(compare(refunds, COLUMNS, set(COLUMNS), [paid, everything]), "销售额") + assert other["kind"] == "caliber" and other["caliber"] == "wf-all@1" + + +def test_a_caliber_with_several_metrics_is_paired_column_by_column(): + two = caliber(WEEKLY.replace("SUM(amount) AS gmv", "SUM(amount) AS gmv, COUNT(*) AS n"), + [{"id": "gmv", "name": "销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}, + {"id": "n", "name": "订单数", "expression": "cell(nodes.fetch, 0, 'n')"}]) + sql = f"SELECT COUNT(*) AS 单数, SUM(amount) AS 销售额, MAX(paid_date) AS 末日 FROM orders WHERE {WINDOW}" + row = compare(sql, ["单数", "销售额", "末日"], {"单数", "销售额", "末日"}, [two]) + assert row["status"] == "related" + assert (label(row, "单数")["kind"], label(row, "单数")["column"]) == ("caliber", "n") + assert (label(row, "销售额")["kind"], label(row, "销售额")["column"]) == ("caliber", "gmv") + assert label(row, "末日")["kind"] == "related" + + +def test_a_metric_over_two_cells_uses_the_cell_proofs_only_on_a_one_row_query(): + two = caliber(WEEKLY.replace("SUM(amount) AS gmv", "SUM(amount) AS gmv, COUNT(*) AS n"), + [{"id": "aov", "name": "客单价", "expression": "cell(nodes.fetch, 0, 'gmv') / cell(nodes.fetch, 0, 'n')"}]) + sql = f"SELECT SUM(amount) AS 销售额, COUNT(*) AS 订单数, MAX(paid_date) AS 末日 FROM orders WHERE {WINDOW}" + row = compare(sql, ["销售额", "订单数", "末日"], {"销售额", "订单数", "末日"}, [two]) + inputs = [{"path": f"cell(nodes.q, 0, '{c}')", "artifact": "q1", "via": "tool_cell", "locator": {"row": 0, "column": c}} + for c in ("销售额", "订单数")] + card = {"metrics": [{"id": "aov", "expression": "cell(nodes.q, 0, '销售额') / cell(nodes.q, 0, '订单数')", + "inputs": inputs}]} + by_key = {two.key: {"expressions": list(two.expressions)}} + got = ledger_labels.metric_label(card, "aov", {"q1": row}, by_key) + assert got["kind"] == "caliber" and got["metric"] == "aov", got + # 同一条查询要是多行的，每一列各自对得上不等于每一行的两列一起对得上：不拿逐格的结论合起来用 + multi = {**row, "one_row": False} + assert ledger_labels.metric_label(card, "aov", {"q1": multi}, by_key)["kind"] == "related" + + +# -------------------------------------------------------------------------- +# 不错给信任：逐格投影照样守上次的投影规则 +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize("sql", [ + f"SELECT paid_date, SUM(amount) AS 销售额, COUNT(*) AS 订单数 FROM orders WHERE {WINDOW} GROUP BY paid_date", + f"SELECT DISTINCT amount AS 销售额, paid_date AS 订单数, id FROM orders WHERE {WINDOW}", +], ids=["extra-group-by", "distinct"]) +def test_cells_are_not_projected_where_the_projection_rules_say_no(sql): + columns = ["paid_date", "销售额", "订单数"] if "GROUP BY" in sql else ["销售额", "订单数", "id"] + row = compare(sql, columns, {"销售额", "订单数"}) + assert row["status"] == "related" + for name in ("销售额", "订单数"): + cell = row["cells"][name] + assert cell["status"] == "related" and all("projection" not in c for c in cell["candidates"]), name + assert cell["candidates"] == row["candidates"], "投影不了的格照用整条查询的结论" + assert label(row, name)["kind"] == "related" + + +# -------------------------------------------------------------------------- +# 真跑一遍真实例子：标签、整题、出具入口、证据片段的 ledger 键 +# -------------------------------------------------------------------------- + + +@pytest.fixture +async def engine_up(tmp_path): + await run_manager.setup() + await week_source(tmp_path) + yield + await ledger_match.drain() + await run_manager.shutdown() + + +async def by_column(run_id: str, labels: dict) -> dict[str, dict]: + """标签按它引用的列排开（报告文档里每个片段引用的是哪一列）。""" + from app.api.evidence import _reports, _sealed + from app.engine.evidence import iter_units + + [report] = _reports(await _sealed(run_id)) + out = {} + for _, unit in iter_units(report.doc): + for seg in unit.get("segments") or []: + if seg.get("id") in labels: + out[seg["cite"]["locator"]["column"]] = {**labels[seg["id"]], "_segment": seg["id"]} + return out + + +async def test_last_weeks_sales_labels_only_the_sales_number_as_from_the_caliber(engine_up, monkeypatch): + graph = S.caliber_graph(WEEKLY) + wf = await S.publish(graph=graph) + row = await S.explore(monkeypatch, ANSWER, origin="chat", text=( + "上周销售额 [[v:Q1.r0.销售额]] 元，共 [[v:Q1.r0.订单数]] 单（[[v:Q1.r0.区间首日]] 至 [[v:Q1.r0.区间末日]]）。")) + content = (await store.latest(row.id)).content + assert content["algo"] == store.ALGO == "4" + [query] = content["queries"] + assert query["status"] == "related" and set(query["cells"]) == set(COLUMNS) + [labels] = [r["labels"] for r in content["reports"].values()] + cols = await by_column(row.id, labels) + assert cols["销售额"]["kind"] == "caliber" and cols["销售额"]["caliber"] == f"{wf}@1" + assert cols["销售额"]["binding"] == {"start": "2026-10-05", "end": "2026-10-11"} + assert {c: cols[c]["kind"] for c in COLUMNS[1:]} == {c: "related" for c in COLUMNS[1:]} + whole = content["whole"] + assert whole["equivalent"] is False and whole["counts"] == {"caliber": 1, "related": 3, "reasoning": 0, + "unknown": 0} + # 工件 meta：逐格证明了的口径也记上（召回按旧版本反查只读它） + from sqlalchemy import select + + from app.db.base import SessionLocal + from app.db.models import Artifact + + async with SessionLocal() as session: + meta = (await session.execute(select(Artifact.meta).where( + Artifact.id == (await store.latest(row.id)).artifact, Artifact.run_id == row.id))).scalar_one() + assert meta["calibers"] == [f"{wf}@1"] + + from app.main import app + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + body = (await client.get(f"/api/runs/{row.id}/ledger")).json() + assert not [a for a in body["actions"] if a["kind"] == "issue"], "整题不等价：不给按口径出具的入口（拍板 L1）" + assert body["issued"] == [], "不自动出具" + [q] = body["queries"] + assert q["cells"]["销售额"]["status"] == "proven" and q["cells"]["订单数"]["status"] == "related" + shown = (await by_column(row.id, body["reports"]["write"]["labels"])) + assert shown["销售额"]["kind"] == "caliber" and shown["销售额"]["workflow_name"] == "周销售" + seg = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cols['销售额']['_segment']}")).json() + assert seg["ledger"]["label"]["kind"] == "caliber" + seg = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cols['订单数']['_segment']}")).json() + assert seg["ledger"]["label"]["kind"] == "related" + where = seg["ledger"]["label"]["related"][0]["diff"]["items"] + assert [i["text"] for i in where] == ["这次多查了 3 列，不影响这个数", "这次数的是行数，口径对 amount 求和"] + + +# -------------------------------------------------------------------------- +# 评测逐题比 SQL：取和期望配上的那一格 +# -------------------------------------------------------------------------- + + +async def test_the_per_case_sql_match_takes_the_cell_that_matched_the_expectation(engine_up, monkeypatch): + from sqlalchemy import select + + from app.db.base import SessionLocal + from app.db.models import DataSource + from app.eval import score + + run = await S.explore(monkeypatch, ANSWER, text="上周销售额 [[v:Q1.r0.销售额]] 元，共 [[v:Q1.r0.订单数]] 单。") + answer = await score.read_answer(run.id) + async with SessionLocal() as session: + sid = (await session.execute(select(DataSource.id).where(DataSource.name == S.CURRENT["source"]))).scalar() + fixed = "SELECT SUM(amount) AS gmv FROM orders WHERE " + WINDOW + case = {"gold": {"kind": "sql", "datasource_id": sid, "sql": "x"}, "expectations": []} + gold = {"status": "ok", "before": {"rendered": {"sql": fixed}, "sets": []}} + # 没给配上的格：逐格证明了的那一格 + match = await score.sql_match(case, gold, answer) + assert match["status"] == "proven" and match["column"] == "销售额", match + # 期望配上的是订单数那一格：取那一格的结论，差在哪按那一格说 + query = answer.cited[0] + match = await score.sql_match(case, gold, answer, cells=[{"artifact": query, "column": "订单数"}]) + assert match["status"] == "related" and match["column"] == "订单数" and match["aspects"] == ["aggregate"], match diff --git a/backend/tests/test_ledger_compose.py b/backend/tests/test_ledger_compose.py index 6d50941e..9171b521 100644 --- a/backend/tests/test_ledger_compose.py +++ b/backend/tests/test_ledger_compose.py @@ -253,7 +253,7 @@ async def test_an_answer_built_with_use_caliber_is_labelled_composed(client, mon await ledger_match.drain() recorded = await store.latest(row.id) content = recorded.content - assert content["algo"] == store.ALGO == "3" + assert content["algo"] == store.ALGO == "4" [query] = content["queries"] assert query["status"] == "composed" and query["node_id"] == "c1_fetch" [cand] = query["candidates"] diff --git a/backend/tests/test_ledger_projection.py b/backend/tests/test_ledger_projection.py index 869dff52..9fa3a257 100644 --- a/backend/tests/test_ledger_projection.py +++ b/backend/tests/test_ledger_projection.py @@ -130,7 +130,10 @@ def test_without_a_projection_every_pair_is_exactly_the_old_result(): (f"SELECT SUM(amount) AS gmv, COUNT(*) AS n FROM orders WHERE {WINDOW}", ["gmv"], {"gmv"}), # 列数对不上 ] for sql, columns, cited in cases: - assert compare(sql, columns, cited) == compare(sql, columns, None), sql + row = compare(sql, columns, cited) + # 逐格比（algo 4）是另加的 cells、one_row 两个键：整条查询这一层的结论和候选照旧一字不差 + assert {k: v for k, v in row.items() if k not in ("cells", "one_row")} == compare(sql, columns, None), sql + assert ("cells" in row) == (len(cited) >= 2 and row["status"] == "related"), sql # -------------------------------------------------------------------------- @@ -185,7 +188,7 @@ async def test_last_weeks_sales_with_extra_columns_is_whole_equivalent(engine_up row = await S.explore(monkeypatch, sql, text="上周销售额 [[v:Q1.r0.销售额]]，共 [[v:Q1.r0.单数]] 单。") recorded = await store.latest(row.id) content = recorded.content - assert content["algo"] == "3" + assert content["algo"] == store.ALGO [q] = content["queries"] assert q["status"] == "related", "销售额、单数两列都被引用，口径只有销售额：投影之后列数对不上，照旧不等价" diff --git a/frontend/src/typesLedger.ts b/frontend/src/typesLedger.ts index ae5e868a..bde041d7 100644 --- a/frontend/src/typesLedger.ts +++ b/frontend/src/typesLedger.ts @@ -82,6 +82,11 @@ export interface LedgerQuery { reason?: string note?: string candidates: (LedgerRelated & { status?: string; how?: string })[] + /** + * 逐格比过的（服务端 ledger/match._cells，algo 4）：被引用的每一列各自的结论和候选。整条（按全部引用列投影）没证明、 + * 报告引用了这条查询好几列时才有；每个数的标签已经按它那一格标好，界面照标签画，不拿它重算 + */ + cells?: Record } /** 整题判断（主报告）：所有带标签的数都来自同一口径的同一版、参数一致时 equivalent */ diff --git a/scripts/check-ledger.mjs b/scripts/check-ledger.mjs index e72fa721..f4c1c54e 100644 --- a/scripts/check-ledger.mjs +++ b/scripts/check-ledger.mjs @@ -24,7 +24,8 @@ import { readFileSync } from 'node:fs' import { chromium } from '../frontend/node_modules/playwright-core/index.mjs' import { - ISSUE, QUEUE, RECONCILE, RECONCILE_VALUES, WHOLE, WHOLE_LABELS, copiedMetricDetail, ledgerOf, metricDetail, reconciledOf, + CELL_LABELS, CELL_WHOLE, ISSUE, QUEUE, RECONCILE, RECONCILE_VALUES, WHOLE, WHOLE_LABELS, copiedMetricDetail, ledgerOf, + metricDetail, reconciledOf, } from './ledger-fixtures.mjs' const WEB = process.env.AGENTLAB_WEB ?? 'http://localhost:5273' @@ -180,6 +181,29 @@ await section('「口径」一步的三种情况：来自口径、相关口径 check('没有运行时报错', errors.length === 0, errors.join(' | ')) }) +// ---------------------------------------------------------------- 逐格判 + +await section('同一条查询的几个数逐格标：一格来自口径、一格只是相关，整题不等价不给出具', async () => { + const { page, errors } = await open('/ui-harness.html?evidence=1', { + reply: evidenceReply(() => ledgerOf(RUN, { labels: CELL_LABELS, whole: CELL_WHOLE, actions: [] })) }) + await page.waitForSelector('#evidence-wide [data-ev-caliber-of="s6"]', { timeout: 10000 }) + check('正文的小标记按数挂：来自口径的那一格挂上，相关的那一格不挂', await page.locator('#evidence-wide [data-ev-caliber-of="s6"]').count() === 1 + && await page.locator('#evidence-wide [data-ev-caliber-of="s10"]').count() === 0) + await openSeg(page, 's6') + await page.locator('[data-evidence-panel] [data-ev-ledger="caliber"]').waitFor({ timeout: 5000 }) + const caliber = await panelPart(page).innerText() + check('销售额那一格：来自口径、等价已证明、按口径的哪一列、参数', caliber.includes('来自口径「周报口径」v2（受管）') + && caliber.includes('等价已证明') && caliber.includes('gmv') && caliber.includes('start = 2026-10-05'), caliber) + check('……整题不等价：不给按口径出具', await page.locator('[data-evidence-panel] [data-ledger-issue]').count() === 0) + await openSeg(page, 's10') + await page.locator('[data-evidence-panel] [data-ev-ledger="related"]').waitFor({ timeout: 5000 }) + const related = await panelPart(page).innerText() + check('同一条查询的订单数那一格：相关口径，差在哪先说多查了几列、再说计数对求和', related.includes('相关口径「周报口径」v2') + && related.includes('这次多查了 3 列，不影响这个数') && related.includes('这次数的是行数，口径对 amount 求和') + && !related.includes('未能归纳'), related) + check('没有运行时报错', errors.length === 0, errors.join(' | ')) +}) + // ---------------------------------------------------------------- 对账 await section('口径对账：核对参数 → 发起 → 选对照的指标 → 三个数各自点得开', async () => { diff --git a/scripts/ledger-fixtures.mjs b/scripts/ledger-fixtures.mjs index ef3d7bf9..5b171032 100644 --- a/scripts/ledger-fixtures.mjs +++ b/scripts/ledger-fixtures.mjs @@ -94,6 +94,24 @@ export const WHOLE = { equivalent: true, caliber: WEEKLY.caliber, workflow_id: W level_text: '受管', workflow_name: '周报口径', how: 'independent', binding: [{ field: 'week', value: '2026-W37' }], missing: [], missing_required: [], counts: { caliber: 3, related: 0, reasoning: 0, unknown: 0 } } +/** + * 逐格判（服务端 ledger/match._cells，algo 4）的真实例子：同一条查询（q-gmv）报告引用了好几列，s6 销售额那一格证明了 + * 等价于周报口径，s10 订单数那一格只是相关（差在哪先说多查了几列、再说计数对求和），整题不等价、没有出具入口 + */ +export const CELL_DIFF = { items: [ + { aspect: 'projection', a: null, b: null, text: '这次多查了 3 列，不影响这个数' }, + { aspect: 'aggregate', a: 'COUNT(*)', b: 'SUM(amount)', text: '这次数的是行数，口径对 amount 求和' }], + fallback: false, text_diff: [] } +export const CELL_LABELS = { + s6: caliberLabel(WEEKLY, 'independent', { binding: [{ field: 'start', value: '2026-10-05' }, { field: 'end', value: '2026-10-11' }], + query: 'q-gmv', node_id: 'fetch', column: 'gmv' }), + s10: { kind: 'related', query: 'q-gmv', related: [{ caliber: WEEKLY.caliber, workflow_id: WEEKLY.workflow_id, version: 2, + level: 'governed', level_text: '受管', workflow_name: '周报口径', + node_id: 'fetch', comparable: true, diff: CELL_DIFF }] }, +} +export const CELL_WHOLE = { equivalent: false, reason: '有的数和相关口径不一致，不来自口径', + counts: { caliber: 1, related: 1, reasoning: 0, unknown: 0 } } + /** 对过的一次账（单行：一组对照）。status 可以换 */ export const reconciledOf = (status = 'succeeded') => ({ run_id: 'run-rec-1', status, query: 'q-orders', caliber: ORDERS.caliber, workflow_id: ORDERS.workflow_id, version: 1,