diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index b06b96e7..84d8ba8d 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -17,8 +17,16 @@ [[i:<输入字段>]] 运行输入 [[t:<表>]] / [[c:<表>.<列>]] 实体：表 / 字段，显示名字本身（反引号里写的已知名字也会自动链接） [[q:K|<逐字引文>]] 引文：必须在那次检索命中的片段里逐字出现，显示引文本身 + [[a:A]] 分析结论的判定语：系统按模板渲染好的一整句（判定由平台的算子算出，写作者只引用） + [[a:A.<键>|万]] 分析结论的具名标量（values 里的一项），换算同 m: + [[v:A.r<行>.<列>]] 分析结论结果表里的一格；[[table:A …]] 整表，表头写结果表的列名 [[see:,…]] 依据：挂在句末，不渲染 +分析结论（docs/design/analysis-round/）在台账里是 kind analysis 的条目，写作目录编号 A1…（和 Q、K 同一套按台账先后 +编号的规则，只收报告祖先里的）。目录条目带着从工件取回、复验过哈希的整份结论（result，app/analysis/contract.catalog_result +读它）；存进报告文档的目录只留摘要，整份结论按工件取回。受管级别的正式出具里，agent 现场做的分析（条目带 call_id）和 +agent 查的格同一个规矩：出具契约没开「单元格引用」就和直接引用单元格一样解析不了；分析卡做的（带 item）总能引用。 + 实体只在这次运行冻结了表结构（schema_snapshot）时才核对：目录里没有表和字段条目的， t / c 仍按一期判为解析不了（「这种引用在后续版本支持」），反引号里的名字也不查；报告节点 写了 entities: off 的，照实说是它关掉的。表结构快照不全（库里的表太多、只存了一部分）时， @@ -54,6 +62,7 @@ from datetime import date, timedelta from decimal import ROUND_HALF_UP, Context, Decimal from typing import Any, Literal, TypedDict +from app.analysis.contract import CATALOG_PREFIX, LEDGER_KIND, MARKER_KIND, catalog_result, is_result, method_label from app.core.artifact_store import canonical_json, content_hash from app.core.artifact_store import load as load_artifact from app.data.names import FILLERS, name_key @@ -98,13 +107,21 @@ _ISO_DATETIME = re.compile(r"^(\d{4}-\d{2}-\d{2})T(\d{2}:\d{2}(?::\d{2}(?:\.\d+) class EvidenceEntry(TypedDict, total=False): """台账条目：state["evidence"] 里的一行，同时写进 node.finished.evidence。""" - kind: Literal["query", "metric_set", "retrieval", "tool", "node_output", "schema"] + kind: Literal["query", "metric_set", "retrieval", "tool", "node_output", "schema", "analysis"] node_id: str exec: int # 这个节点第几次执行（循环里会有多次），从 1 数 artifact: str # 被引用的那件工件 via: str # 外层包装的工件（query_snapshot 外面的 tool_snapshot） - call_id: str + call_id: str # agent 工具记下的那次调用（查询、分析） tool: str + #: analysis 专有（app/analysis/contract.ledger_entry）：算子、模式、算成没算成（ok / refused）、判定码、输入快照的 + #: sha256（去重、保序）；分析卡来的另记那一项的 id（item），agent 工具来的记 call_id + op: str + mode: str | None + status: str + verdict: str | None + inputs: list[str] + item: str source: str columns: list[str] rows: int @@ -126,7 +143,7 @@ class Citation(TypedDict, total=False): alias: str # 目录里的键：m:gmv、i:week、Q4、t:orders、K1 locator: dict[str, Any] eid: str - kind: str # metric / input / query / cell / table / column / quote + kind: str # metric / input / query / cell / table / column / quote / analysis role: Literal["value", "entity", "quote", "support"] status: Literal["resolved", "unresolved", "mismatch"] reason: str # unresolved 时为什么 @@ -139,7 +156,8 @@ class Citation(TypedDict, total=False): class Segment(TypedDict, total=False): id: str - kind: Literal["text", "number", "value", "entity", "quote", "structural"] + #: analysis：[[a:A3]] 渲染出的判定语（一整句，系统按模板写的）；[[a:A3.键]] 的具名标量照数字、取值算 + kind: Literal["text", "number", "value", "entity", "quote", "structural", "analysis"] text: str span: list[int] # [start, end)，在 doc.markdown 里的码点偏移 ref: str # 带前缀的原始引用：m:gmv、i:week、t:orders @@ -745,7 +763,7 @@ _MARKER_GUARD = rf"(?=[^\[\]\n]{{0,{MARKER_MAX - 4}}}\]\])" #: [[kind:body]]。body 里不许有方括号和换行——这样一个没闭合的 [[ 最多吞到行尾 MARKER_RE = re.compile( r"\[\[" + _MARKER_GUARD + r"[ \t]*(?P[A-Za-z]+)[ \t]*:(?P[^\[\]\n]*?)\]\]") -SUPPORTED_KINDS = frozenset({"m", "i", "see", "v", "table", "t", "c", "q"}) +SUPPORTED_KINDS = frozenset({"m", "i", "see", "v", "table", "t", "c", "q", MARKER_KIND}) #: 三期起没有「以后才支持」的标记了；留着这个名字给老代码 import LATER_KINDS: frozenset[str] = frozenset() #: 这次运行没冻结表结构（没查过库、升级前的运行）时，t / c 仍按一期判为解析不了，原因是这一句 @@ -759,6 +777,9 @@ PHASE_TWO_REASON = "这份报告生成于系统升级前，当时不解析这类 CODE_REASON = "沙箱代码计算出的数字不能直接引用，需先登记到口径卡" #: 受管级别的正式出具，契约里没写 "cells": true 时复核按这个原因拒掉单元格引用 CELLS_REASON = "受管级别出具需要在出具契约中开启「单元格引用」" +#: 受管级别的正式出具、契约里没开「单元格引用」时，Agent 现场做的分析（台账条目带 call_id）和它查的格同一个规矩：判定语、 +#: 具名标量、结果表一律解析不了（依据 [[see:A1]] 照样可以写）。分析卡做的（带 item）是模板定好的，总能引用 +ANALYSIS_CELLS_REASON = "受管级别出具中，Agent 现场做的分析和它查到的单元格同一个规矩：需要在出具契约中开启「单元格引用」" #: 节点字段引用：给人看的原因，和交回写作者改写时的指令（_Reason 的 fix） NODE_FIELD_REASON = "不支持引用节点字段：数字需来自查询单元格或口径卡指标" NODE_FIELD_FIX = "节点字段引用在后续版本支持：数要从查询单元格 [[v:Q1.r0.列]] 或口径卡指标 [[m:…]] 来" @@ -864,14 +885,17 @@ def build_catalog( inputs: dict[str, Any] | None = None, metrics_from: list[str] | str | None = None, allowed: set[str] | frozenset[str] | list[str] | None = None, + loader: Loader | None = None, ) -> dict[str, dict[str, Any]]: """从证据台账、节点产出、运行输入建目录。 - nodes：state["nodes"]，口径卡的指标值从这里取（和台账里的工件是同一份内容） - - ledger：state["evidence"]，决定顺序、提供工件 id；查询、检索按台账顺序编 Q1… / K1… + - ledger：state["evidence"]，决定顺序、提供工件 id；查询、检索、分析结论按台账顺序编 Q1… / K1… / A1… - metrics_from：只收这几张口径卡；不传（None、空列表、空字符串）就收全部——和 validate_graph 的理解一致，那边空列表就是「所有上游口径卡」 - allowed：只收这些节点的证据（报告节点传它的祖先），不传不限 + - loader：取分析结论工件的函数（缺省 artifact_store.load，取回时复验哈希）。A 条目照 analysis.contract.catalog_result + 的约定带着整份结论（result）；取不回、哈希不符、形状不对的不带 result，另记 unreadable（为什么），引用它时照实说 同一个指标 id 出现在两张卡里时不登记 m:，只登记 m:<卡的节点 id>.—— 按先来后到默认给一个，正是旧契约错配的那种错。 @@ -932,9 +956,13 @@ def build_catalog( # 只在有问题时有这两个键 **({"no_data": True, "no_data_reason": metric.get("no_data_reason") or ""} if metric.get("no_data") else {}), + # 跨两份同形查询算的环比、同比，口径卡执行时自动做的可比性检查（analysis.contract.MetricComparability）： + # 写作目录照实写有条件可比、不可比的原因，证据面板的指标步骤展示它。没做过检查的指标没有这个键 + **({"comparability": metric["comparability"]} if isinstance(metric.get("comparability"), dict) else {}), } - queries = retrievals = 0 + queries = retrievals = analyses = 0 + snaps: _Snapshots | None = None # 取分析结论的工件用，有 A 条目才建 schemas: list[dict[str, Any]] = [] shaped: list[tuple[dict[str, Any], str]] = [] # 带表结构快照的查询条目和它的全局编号 for entry in entries: @@ -954,6 +982,12 @@ def build_catalog( "label": f"沙箱代码 {node_id}（{option_label('evidence_role', entry.get('role') or 'compute')}）", } continue + if entry.get("kind") == LEDGER_KIND: + analyses += 1 + alias = f"{CATALOG_PREFIX}{analyses}" + snaps = snaps or _Snapshots(loader) + catalog[alias] = _analysis_entry(entry, alias, snaps) + continue if entry.get("kind") == "query": queries += 1 alias, kind = f"Q{queries}", "query" @@ -989,6 +1023,44 @@ def build_catalog( return catalog +#: A 条目从台账条目照抄的键（app/analysis/contract.ledger_entry 的形状） +_ANALYSIS_KEYS = ("node_id", "exec", "artifact", "call_id", "item", "op", "mode", "status", "verdict", "inputs") + + +def _analysis_entry(entry: dict[str, Any], alias: str, snaps: _Snapshots) -> dict[str, Any]: + """台账里的一条分析结论 → 写作目录的 A 条目：照抄台账的键，带上从工件取回、复验过哈希的整份结论（result， + analysis.contract.catalog_result 读它）。取不回、哈希不符、形状不对的不带 result，记 unreadable（以编号开头的一句，引用 + 它时照实说）。rows 是结果表的行数：依据 [[see:A1.r3]] 按它核对，存进报告文档的摘要里也留着。""" + out: dict[str, Any] = {"alias": alias, "kind": LEDGER_KIND, "eid": make_eid(LEDGER_KIND, entry.get("artifact"), {}), + "locator": {}, **{k: entry[k] for k in _ANALYSIS_KEYS if k in entry}} + result, why = snaps.analysis(alias, {"artifact": entry.get("artifact")}) + if result is None: + out["unreadable"] = why + else: + out["result"] = result + table = result.get("table") + if isinstance(table, dict) and isinstance(table.get("rows"), list): + out["rows"] = len(table["rows"]) + phrase = _clean(str((result or {}).get("phrase") or "")) + name = method_label(out.get("op"), out.get("mode")) + out["label"] = f"{name} · {phrase}" if phrase else name + return out + + +def analysis_card(entry: dict[str, Any] | None) -> bool: + """这条分析结论是不是分析卡做的（台账条目带 item、不带 call_id）：模板定好的分析总能引用（00-总方案 D10）。agent 工具 + 现场做的（带 call_id）在受管级别的正式出具里和 agent 查的格同一个规矩（cells_allowed）。""" + return isinstance(entry, dict) and bool(entry.get("item")) and not entry.get("call_id") + + +def stored_entry(entry: dict[str, Any]) -> dict[str, Any]: + """存进报告文档的目录条目：分析结论只留摘要（编号、种类、判定码、工件、输入、行数……），不带整份结论（result）——它在 + 工件里，取回时复验哈希；别的条目原样。""" + if entry.get("kind") == LEDGER_KIND and "result" in entry: + return {k: v for k, v in entry.items() if k != "result"} + return entry + + def withhold_metrics(catalog: dict[str, Any], aliases: Iterable[str], word: str) -> dict[str, Any]: """目录里值不给的指标（aliases，指标的别名）：值、渲染、标签里的值都换成 word（「已遮罩」），另记 masked。交回一份新的 目录，原来那份不动；没有要换的原样交回。 @@ -1376,6 +1448,38 @@ class _Snapshots: why = "的快照里没有表格数据" return (None, f"{alias} {why}") if why else (content, None) + def analysis(self, alias: str, entry: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None]: + """分析结论：(整份结论, 取不到的原因，以 alias 开头)。写作目录的条目带着 result（build_catalog 取回、复验过哈希的， + analysis.contract.catalog_result）就用它；存进报告文档的目录只留摘要，按工件取回、复验哈希。build_catalog 当时就没取回 + 来的（条目记着 unreadable）照它说，不再取一遍。""" + result = catalog_result(entry) + if result is not None: + return result, None + if entry.get("unreadable"): + return None, str(entry["unreadable"]) + content, why = self._fetch(entry.get("artifact"), "分析结论") + if why is None and not is_result(content): + why = "的工件不是分析结论" + return (None, f"{alias} {why}") if why else (content, None) + + def table(self, alias: str, entry: dict[str, Any]) -> tuple[Any, str | None]: + """按单元格、整表取的那张表：(表, 取不到的原因)。查询结果是查询快照本身；分析结论是它的结果表（形状同查询快照： + columns 是稳定的列 id，另带 labels 给人看的列名，整表的表头写它）。结果表另记了列类型（column_types）的照着渲染， + 没记的按值认（同老快照）。""" + if entry.get("kind") != LEDGER_KIND: + return self.get(alias, entry.get("artifact")) + result, why = self.analysis(alias, entry) + if why: + return None, why + table = result.get("table") if isinstance(result, dict) else None + if not isinstance(table, dict) or not isinstance(table.get("rows"), list): + return None, f"{alias} 没有结果表" + ("（这次分析没有做成）" if result.get("status") == "refused" else "") + labels = table.get("labels") + types = table.get("column_types") + return {"columns": [str(c) for c in table.get("columns") or []], "rows": table["rows"], + "labels": {str(k): str(v) for k, v in labels.items()} if isinstance(labels, dict) else {}, + **({"column_types": types} if isinstance(types, dict) else {})}, None + def hits(self, alias: str, artifact: Any) -> tuple[list[Any], str | None]: """检索快照里的命中：(命中列表, 取不到的原因)。""" content, why = self._fetch(artifact, "检索快照") @@ -1418,27 +1522,49 @@ def _snapshots(loader: Loader | _Snapshots | None, *, entities: bool = True, return _Snapshots(loader, entities=entities, syntax=syntax, display=display) -_CELL_REF = re.compile(r"^(?P[QK]\d+)\.r(?P\d+)\.(?P.+)$") +_CELL_REF = re.compile(rf"^(?P[QK{CATALOG_PREFIX}]\d+)\.r(?P\d+)\.(?P.+)$") _NODE_REF = re.compile(r"^N:(?P[^.\s]+)(?:\.(?P.+))?$") def parse_cell_ref(ref: str) -> dict[str, Any] | None: - """Q4.r0.amount → {alias: "Q4", row: 0, column: "amount"}；写法不对返回 None。""" + """Q4.r0.amount → {alias: "Q4", row: 0, column: "amount"}；写法不对返回 None。分析结论的结果表同一种写法（A3.r0.share）。""" m = _CELL_REF.match((ref or "").strip()) return {"alias": m.group("alias"), "row": int(m.group("row")), "column": m.group("column")} if m else None +def _is_analysis_alias(alias: Any) -> bool: + return isinstance(alias, str) and alias.startswith(CATALOG_PREFIX) and alias[len(CATALOG_PREFIX):].isdigit() + + def _query_entry(alias: str, catalog: dict[str, Any]) -> tuple[dict[str, Any] | None, str]: + """能按单元格、整表引用的条目：查询结果，或者分析结论（引用它的结果表）。""" entry = catalog.get(alias) if entry is None: + if _is_analysis_alias(alias): + return None, _Reason(f"引用的分析结论 {alias} 不存在", f"目录里没有 {alias}（能引用的分析结论见证据目录）") return None, _Reason(f"引用的查询结果 {alias} 不存在", f"目录里没有 {alias}（能引用的查询见证据目录）") - if entry.get("kind") != "query": + if entry.get("kind") not in ("query", LEDGER_KIND): return None, f"{alias} 不是查询结果，不能按单元格引用" if entry.get("code"): return None, CODE_REASON return entry, "" +def _cells_refused(alias: str, catalog: dict[str, Any], cells_allowed: bool) -> str | None: + """单元格、整表引用这一次能不能解析：受管级别的正式出具、契约没开「单元格引用」时（cells_allowed=False）解析不了， + 原因是 CELLS_REASON；分析结论的结果表同一个规矩，只是分析卡做的总能引用（analysis_card，00-总方案 D10）。""" + if cells_allowed: + return None + if _is_analysis_alias(alias): + return None if analysis_card(catalog.get(alias)) else ANALYSIS_CELLS_REASON + return CELLS_REASON + + +def _noun(entry: dict[str, Any] | None) -> str: + """按格引用的那张表在说法里叫什么：查询结果，或者分析结论的结果表。""" + return "结果表" if (entry or {}).get("kind") == LEDGER_KIND else "查询结果" + + def _resolve_cell(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) -> dict[str, Any]: ref = marker.get("ref") or "" @@ -1454,12 +1580,12 @@ def _resolve_cell(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snaps return _unresolved(marker, ref.split(".")[0], "cell", _Reason( "单元格引用格式无法识别", "单元格要写成 Q<编号>.r<行>.<列>（行号从 0 数），比如 [[v:Q1.r0.amount]]")) alias = parsed["alias"] - if not cells_allowed: - return _unresolved(marker, alias, "cell", CELLS_REASON) + if refused := _cells_refused(alias, catalog, cells_allowed): + return _unresolved(marker, alias, "cell", refused) entry, why = _query_entry(alias, catalog) if entry is None: return _unresolved(marker, alias, "cell", why) - snapshot, why = snaps.get(alias, entry.get("artifact")) + snapshot, why = snaps.table(alias, entry) if why: return _unresolved(marker, alias, "cell", why) try: @@ -1474,7 +1600,7 @@ def _resolve_cell(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snaps rendered = render_cell(shown, conv, kind=kind, plain=True) snaps.needed = max(snaps.needed, 3) except CellError as e: - return _unresolved(marker, alias, "cell", f"{alias} {e}") + return _unresolved(marker, alias, "cell", f"{alias} {str(e).replace('查询结果', _noun(entry))}") except RenderError as e: return _unresolved(marker, alias, "cell", str(e)) locator = {"row": parsed["row"], "column": column} @@ -1506,8 +1632,8 @@ def _table_plan(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapsho def fail(reason: str, fix: str | None = None) -> tuple[None, dict[str, Any]]: return None, _unresolved(marker, alias, "query", _Reason(reason, fix) if fix else reason) - if not cells_allowed: - return fail(CELLS_REASON) + if refused := _cells_refused(alias, catalog, cells_allowed): + return fail(refused) _, *rest = (marker.get("body") or "").split() or [""] odd = [p for p in rest if p.partition("=")[0] not in _TABLE_OPTIONS or "=" not in p] if odd: @@ -1516,18 +1642,20 @@ def _table_plan(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapsho entry, why = _query_entry(alias, catalog) if entry is None: return fail(why) - snapshot, why = snaps.get(alias, entry.get("artifact")) + snapshot, why = snaps.table(alias, entry) if why: return fail(why) columns = [str(c) for c in snapshot.get("columns") or []] rows = snapshot["rows"] + #: 分析结论的结果表：表头写给人看的列名（labels），格子照样按列 id 引用 + labels = snapshot.get("labels") if entry.get("kind") == LEDGER_KIND else None if "cols" in options: wanted = [c.strip() for c in options["cols"].split(",") if c.strip()] if not wanted: return fail("整表引用的 cols 没有写列名", "cols= 后面要写列名，用逗号分开") missing = [c for c in wanted if c not in columns] if missing: - return fail(f"引用的查询结果 {alias} 中没有列「{missing[0]}」", + return fail(f"引用的{_noun(entry)} {alias} 中没有列「{missing[0]}」", f"{alias} 没有列「{missing[0]}」，有：{'、'.join(columns[:12])}") if len(wanted) > TABLE_MAX_COLS: return fail(f"整表最多 {TABLE_MAX_COLS} 列，当前选了 {len(wanted)} 列", @@ -1564,7 +1692,7 @@ def _table_plan(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapsho return fail(f"列名「{_clean(c)[:30]}」无法放进表格（为空、首尾有空格、含方括号 / 竖线 / 换行，或过长）", f"列名「{_clean(c)[:30]}」写不进表格（空的、首尾有空格、带方括号竖线换行，或者太长）：" "在 SQL 里给它起个别名再查") - lines = ["| " + " | ".join(_clean(c) for c in wanted) + " |", + lines = ["| " + " | ".join(_header_text(c, labels) for c in wanted) + " |", "| " + " | ".join("---" for _ in wanted) + " |"] lines += ["| " + " | ".join(f"[[v:{alias}.r{r}.{c}]]" for c in wanted) + " |" for r in range(first, last + 1)] if snapshot.get("truncated") is True: @@ -1572,6 +1700,14 @@ def _table_plan(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapsho return "\n".join(lines), None +def _header_text(column: str, labels: dict[str, str] | None) -> str: + """整表表头的一格：查询结果写列名本身；分析结论的结果表写给人看的列名（labels，竖线换成 ¦，同单元格里的文本），没写 + 叫法的列写列 id。复核时表头核对（_header_masks）按同一个写法认。""" + if labels and labels.get(column): + return _clean(str(labels[column])).replace("|", "¦") + return _clean(column) + + def _truncated_note(first: int, last: int) -> str: """截断的查询结果生成的表格下面那行说明。表里每一格都是真实的，但读的人会把它当成全部数据。 @@ -1620,10 +1756,11 @@ def _expand_tables(text: str, catalog: dict[str, Any], snaps: _Snapshots, cells_ def resolve_marker(marker: dict[str, Any], catalog: dict[str, Any], *, cells_allowed: bool = True, loader: Loader | _Snapshots | None = None) -> dict[str, Any]: - """行内标记（m / i / v / t / c / q / table）→ Citation，带 rendered（正文里显示的字）。 + """行内标记（m / i / v / t / c / q / a / table）→ Citation，带 rendered（正文里显示的字）。 - cells_allowed=False 时单元格和整表一律解析不了，原因是 CELLS_REASON（受管出具没在契约里声明 cells）。 - loader 是取查询快照的函数，缺省 artifact_store.load。 + cells_allowed=False 时单元格和整表一律解析不了，原因是 CELLS_REASON（受管出具没在契约里声明 cells）；Agent 现场做的 + 分析结论同一个规矩（ANALYSIS_CELLS_REASON），分析卡做的照常解析。loader 是取查询快照、分析结论的函数，缺省 + artifact_store.load。 """ kind = marker["kind"] ref = marker.get("ref") or "" @@ -1635,17 +1772,20 @@ def resolve_marker(marker: dict[str, Any], catalog: dict[str, Any], *, cells_all if bad is not None: return bad # 整篇渲染时整表已经按上下文展开过了，走到这里的是复核一个片段：给出不带上下文的样子 - return {"ref": ref, "alias": ref, "locator": {}, "eid": catalog[ref]["eid"], "kind": "query", + return {"ref": ref, "alias": ref, "locator": {}, "eid": catalog[ref]["eid"], + "kind": LEDGER_KIND if catalog[ref].get("kind") == LEDGER_KIND else "query", "role": "value", "status": "resolved", "rendered": _render(table or "", catalog, snaps, cells_allowed)} if kind in ("t", "c"): return _resolve_entity(marker, catalog, _snapshots(loader)) if kind == "q": return _resolve_quote(marker, catalog, _snapshots(loader)) + if kind == MARKER_KIND: + return _resolve_analysis(marker, catalog, _snapshots(loader), cells_allowed) if kind not in ("m", "i"): return _unresolved(marker, f"{kind}:{ref}", kind, _Reason( f"引用类型「{kind}」无法识别", f"不认识的引用类型 {kind}：只支持 m（口径卡指标）、v（查询单元格）、table（整表）、" - "i（运行输入）、t / c（表 / 字段）、q（引文）、see（依据）")) + "i（运行输入）、t / c（表 / 字段）、q（引文）、a（分析结论）、see（依据）")) conv = marker.get("conv") if kind == "m": @@ -1686,6 +1826,68 @@ def resolve_marker(marker: dict[str, Any], catalog: dict[str, Any], *, cells_all return cite +#: [[a:A3]]、[[a:A3.top_share]]：编号，后面可以接一个具名标量的键 +_ANALYSIS_REF = re.compile(rf"^(?P{CATALOG_PREFIX}\d+)(?:\.(?P[^\s.|]+))?$") +_ANALYSIS_FIX = (f"分析结论的判定语写 [[{MARKER_KIND}:A1]]，具名的数写 [[{MARKER_KIND}:A1.键]]（键见证据目录，比如 " + f"[[{MARKER_KIND}:A1.top_share|pct]]），结果表的一格写 [[v:A1.r0.列]]") + + +def _resolve_analysis(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapshots, + cells_allowed: bool) -> dict[str, Any]: + """[[a:A3]] → 判定语：分析结论里系统按模板渲染好的那一整句（判定是平台的算子算的，写作者只引用，00-总方案 D5）。 + [[a:A3.top_share|pct]] → 具名标量：结论 values 里的一项（数一律十进制字符串），按单元格的规则渲染，换算同 [[m:]]。 + + 判定语是一个片段（种类 analysis），里面的数和字都是系统写的：组装、复核都按带引用的片段整段涂白，不算裸数字。受管级别的 + 正式出具、契约没开「单元格引用」时，Agent 现场做的分析解析不了（ANALYSIS_CELLS_REASON），分析卡做的照常解析。""" + ref = (marker.get("ref") or "").strip() + conv = marker.get("conv") + parsed = _ANALYSIS_REF.match(ref) + if parsed is None: + return _unresolved(marker, ref.split(".")[0] or ref, LEDGER_KIND, + _Reason("分析结论的引用格式无法识别", _ANALYSIS_FIX)) + alias, key = parsed.group("alias"), parsed.group("key") + entry = catalog.get(alias) + if entry is None: + return _unresolved(marker, alias, LEDGER_KIND, _Reason( + f"引用的分析结论 {alias} 不存在", f"目录里没有 {alias}（能引用的分析结论见证据目录）")) + if entry.get("kind") != LEDGER_KIND: + return _unresolved(marker, alias, LEDGER_KIND, f"{alias} 不是分析结论") + if not cells_allowed and not analysis_card(entry): + return _unresolved(marker, alias, LEDGER_KIND, ANALYSIS_CELLS_REASON) + result, why = snaps.analysis(alias, entry) + if result is None: + return _unresolved(marker, alias, LEDGER_KIND, why or f"{alias} 无法读取") + if key is None: + if conv: + return _unresolved(marker, alias, LEDGER_KIND, _Reason( + "判定语是一整句话，不能换算", + f"判定语 [[{MARKER_KIND}:{alias}]] 不能加换算：要换算就引用具名的数，比如 [[{MARKER_KIND}:{alias}.top_share|pct]]")) + phrase = result.get("phrase") + if not isinstance(phrase, str) or not phrase.strip(): + return _unresolved(marker, alias, LEDGER_KIND, f"{alias} 没有判定语") + # 判定语按文本格的规则写进正文：压成一行，竖线换成 ¦（同单元格里的文本，表格里不会多切出一格） + return {"ref": ref, "alias": alias, "locator": {}, "eid": make_eid(LEDGER_KIND, entry.get("artifact"), {}), + "kind": LEDGER_KIND, "role": "value", "status": "resolved", + "rendered": render_cell(phrase, kind="text")} + values = result.get("values") if isinstance(result.get("values"), dict) else {} + if key not in values: + names = "、".join(str(k) for k in list(values)[:12]) or "没有" + return _unresolved(marker, alias, LEDGER_KIND, _Reason( + f"{alias} 中没有名为「{key}」的数", f"{alias} 里没有「{key}」，能引用的数：{names}（写成 [[{MARKER_KIND}:{alias}.键]]）")) + raw = values[key] + try: + rendered = render_cell(raw, conv) + except RenderError as e: + return _unresolved(marker, alias, LEDGER_KIND, str(e)) + locator = {"key": key} + cite = {"ref": ref, "alias": alias, "locator": locator, + "eid": make_eid(LEDGER_KIND, entry.get("artifact"), locator), "kind": LEDGER_KIND, "role": "value", + "status": "resolved", "value": cell_value(raw), "rendered": rendered} + if conv: + cite["conv"] = conv + return cite + + def _resolve_entity(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapshots) -> dict[str, Any]: """[[t:orders]] / [[c:orders.amount]]：显示名字本身（照写作者写的大小写），出处是目录里的表或字段。 @@ -1855,12 +2057,15 @@ def _resolve_quote(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snap "source": {k: v for k, v in source.items() if v is not None}} -_ROW_REF = re.compile(r"^(?P[QK]\d+)(?:\.r(?P\d+))?$") +_ROW_REF = re.compile(rf"^(?P[QK{CATALOG_PREFIX}]\d+)(?:\.r(?P\d+))?$") def resolve_support(ref: str, catalog: dict[str, Any], *, loader: Loader | _Snapshots | None = None) -> dict[str, Any]: - """[[see:]] 里的一个依据：m:gmv、i:week、Q4、Q4.r0、K2、t:orders、c:orders.amount。只核对「这件证据存在」。""" + """[[see:]] 里的一个依据：m:gmv、i:week、Q4、Q4.r0、K2、A3、A3.r0、t:orders、c:orders.amount。只核对「这件证据存在」。 + 分析结论当依据和查询编号一样写（[[see:A3]]），受管级别的出具里也照样可以写；顺手写成 a:A3 的照 A3 认。""" ref = ref.strip() + if ref.startswith(f"{MARKER_KIND}:") and _is_analysis_alias(ref.partition(":")[2].strip()): + ref = ref.partition(":")[2].strip() kind, sep, body = ref.partition(":") base = {"ref": ref, "role": "support"} if sep and kind in ("t", "c"): @@ -1880,7 +2085,8 @@ def resolve_support(ref: str, catalog: dict[str, Any], *, loader: Loader | _Snap alias = ref if ref in catalog else None why = f"运行输入中没有「{body}」" elif sep: - why = _Reason(f"依据「{ref}」的写法无法识别", f"不认识的依据写法 {ref}：写 m:指标、i:输入，或者 Q1 / K1 这样的编号") + why = _Reason(f"依据「{ref}」的写法无法识别", + f"不认识的依据写法 {ref}：写 m:指标、i:输入，或者 Q1 / K1 / A1 这样的编号") elif m := _ROW_REF.match(ref): alias = m.group("alias") if m.group("alias") in catalog else None if alias and m.group("row") is not None: @@ -1910,7 +2116,7 @@ def resolve_ref(ref: str, catalog: dict[str, Any], *, cells_allowed: bool = True _UNIT_GAP = 2 _UNIT_MAX = 10 UNIT_LOOKAHEAD = _UNIT_GAP + _UNIT_MAX + 1 -_UNIT_KINDS = frozenset({"m", "i", "v"}) +_UNIT_KINDS = frozenset({"m", "i", "v", MARKER_KIND}) #: 以一个字的单位打头的更长的单位：口径卡是「人」、正文紧跟着写「人次」，是另一个单位，不当重复的去掉（_tidy） _LONGER_UNIT = re.compile( r"人次|人天|人日|人月|人年|人份|件次|台次|台套|台时|车次|架次|班次|户次|批次|个月|个季度|个工作日|个百分点" @@ -2802,6 +3008,14 @@ def _link_entities(segments: list[dict[str, Any]], catalog: dict[str, Any], return out +def _analysis_segment(marker: dict[str, Any], cite: dict[str, Any]) -> str | None: + """[[a:A3]] 的片段种类是 analysis（一整句判定语）；[[a:A3.键]] 是具名标量，照 [[v:]] 按值分数字、取值（交回 None）。""" + if marker.get("kind") != MARKER_KIND: + return None + parsed = _ANALYSIS_REF.match(str(marker.get("ref") or "").strip()) + return LEDGER_KIND if parsed is None or parsed.group("key") is None else None + + def _auto_entity(seg: dict[str, Any]) -> bool: """系统在正文里自动切出来的实体片段（自动链接的名字、反引号里的可疑名字），不是写作者写的标记。""" return seg.get("kind") == "entity" and (bool(seg.get("auto")) or not seg.get("ref")) @@ -2922,7 +3136,7 @@ def compose_doc( # 紧跟在同一张表名后面：只显示列名，引用上记 short，复核照它核对（老文档没有这一项，照全名） cite = {**cite, "rendered": short, "short": True} seg = { - "kind": _SEG_KIND.get(marker["kind"]) + "kind": _SEG_KIND.get(marker["kind"]) or _analysis_segment(marker, cite) or ("number" if marker["kind"] == "m" or (ok and is_number(cite.get("value"))) else "value"), "text": cite["rendered"], "ref": f"{marker['kind']}:{marker['body']}", @@ -2958,7 +3172,7 @@ def compose_doc( unit["segments"] = _link_entities(unit["segments"], catalog, index) all_segments = [s for block in out_blocks for unit in block["units"] for s in unit["segments"]] # 裸数字单独切成片段：前端要在那几个字底下画「无出处」 - refs = [tuple(s["span"]) for s in all_segments if s.get("ref")] + _header_masks(out_blocks, catalog) \ + refs = [tuple(s["span"]) for s in all_segments if s.get("ref")] + _header_masks(out_blocks, catalog, snaps) \ + _name_masks(markdown, catalog, snaps) syntax = [tuple(s["span"]) for s in all_segments if s["kind"] == "structural"] bare = _bare_numbers(markdown, refs, allow_numbers, syntax) @@ -2996,7 +3210,8 @@ def compose_doc( used = {c["alias"] for block in out_blocks for unit in block["units"] for c in [*(s["cite"] for s in unit["segments"] if s.get("cite")), *unit["see"]] if c.get("status") == "resolved" and c.get("alias")} - kept = {a: e for a, e in catalog.items() if e.get("kind") not in ENTITY_KINDS or a in used} + # 分析结论只留摘要（stored_entry）：整份结论在工件里，按工件取回、复验哈希 + kept = {a: stored_entry(e) for a, e in catalog.items() if e.get("kind") not in ENTITY_KINDS or a in used} # entity_syntax：这份文档按哪一版实体语法组装，复核照它走（verify_doc）。在文档顶层，参与内容哈希 # provenance：期 4 的文档标记（P4-SPEC 1.4）。带它的文档才有查询步骤的提示字段、推断来源、裁判摘录的新行； # 没有它的（期 4 之前组装的）一律按原样展示。写在顶层、参与内容哈希，事后补不上：补了哈希就变，封存链就断 @@ -3018,10 +3233,19 @@ def compose_doc( return doc -def _header_masks(blocks: list[dict[str, Any]], catalog: dict[str, Any]) -> list[tuple[int, int]]: +def _header_masks(blocks: list[dict[str, Any]], catalog: dict[str, Any], + snaps: _Snapshots | None = None) -> list[tuple[int, int]]: """表头里恰好是某次查询的列名的格：整表的表头是系统照快照列名写的，「销售额2025」这种列名 - 里的数字不是结论数字，写作者也改不了它。只放过表头、只放过和列名一字不差的格。""" + 里的数字不是结论数字，写作者也改不了它。只放过表头、只放过和列名一字不差的格。 + + 分析结论的结果表同理：整表的表头写的是结果表给人看的列名（_header_text），「基期（2026-W41）」里的数是系统写的。 + 结论按 snaps 取（目录条目带着的直接用，存进文档的摘要按工件取回）；没给 snaps 的只认查询的列名。""" names = {_clean(str(c)) for e in catalog.values() if e.get("kind") == "query" for c in e.get("columns") or []} + for alias, entry in catalog.items() if snaps is not None else (): + if isinstance(entry, dict) and entry.get("kind") == LEDGER_KIND: + table, why = snaps.table(alias, entry) + if why is None: + names.update(_header_text(str(c), table.get("labels")) for c in table.get("columns") or []) out: list[tuple[int, int]] = [] for block in blocks if names else []: if not isinstance(block, dict) or block.get("type") != "table": @@ -3049,12 +3273,13 @@ _UNIT_VALUE = re.compile( def _text_names(catalog: dict[str, Any], snaps: _Snapshots) -> set[str]: """目录里每次查询的快照中，文本格里带数字的值（「冲压5线」「KX-205 线」）：至少有一个数字、至少有一个不是数字的字， - 本身不是数、也不是带单位的数（_UNIT_VALUE）。按写作目录里显示的样子（压成一行）收。取不回的快照跳过。""" + 本身不是数、也不是带单位的数（_UNIT_VALUE）。按写作目录里显示的样子（压成一行）收。取不回的快照跳过。分析结论的结果表 + 同样收（维度值照输入快照的原样，写作目录里也列着它们）。""" names: set[str] = set() for alias, entry in catalog.items(): - if not isinstance(entry, dict) or entry.get("kind") != "query": + if not isinstance(entry, dict) or entry.get("kind") not in ("query", LEDGER_KIND): continue - snapshot, why = snaps.get(alias, entry.get("artifact")) + snapshot, why = snaps.table(alias, entry) if why or not isinstance(snapshot, dict): continue columns = [str(c) for c in snapshot.get("columns") or []] @@ -3360,7 +3585,7 @@ def verify_doc( if tiled and pos != len(markdown): violations.append(_violation("segment_mismatch", "正文末尾有一段不属于任何片段，文档可能被修改过")) - masks += _header_masks(list(doc.get("blocks") or []), catalog) + _name_masks(markdown, catalog, snaps) + masks += _header_masks(list(doc.get("blocks") or []), catalog, snaps) + _name_masks(markdown, catalog, snaps) for token in _bare_numbers(markdown, masks, allow_numbers, syntax): owner = next(((sid, uid) for s, e, sid, uid in owners if s <= token.start < e), (None, None)) violations.append(_violation( @@ -3708,11 +3933,12 @@ _PROMPT_HITS = 8 _PROMPT_HIT_CHARS = 300 -def _sql_check_note(artifact: Any, snaps: _Snapshots) -> str | None: +def _sql_check_note(artifact: Any, snaps: _Snapshots, who: str | None = None) -> str | None: """这次查询（合并查询则是它的输入）没通过 SQL 检查时，写作目录里跟在它后面的那句提醒；没问题返回 None。 口径卡的指标早就标「存疑」了，直接引用查询格子的写作者却看不到：报告照样把一对多关联后求和的数写成定论。 - 出具契约会按缺口降档（issuance.sql_check_gaps），这里先让写作者知道、引用时如实交代。 + 出具契约会按缺口降档（issuance.sql_check_gaps），这里先让写作者知道、引用时如实交代。分析结论的输入查询同样提醒 + （sql_problems.query_problems 顺着结论的输入往下找），who 是「这次分析的输入查询」。 """ from app.engine.sql_problems import query_problems @@ -3720,7 +3946,7 @@ def _sql_check_note(artifact: Any, snaps: _Snapshots) -> str | None: if not found: return None problems = list(dict.fromkeys(text for p in found for text in p.problems)) - who = "这次查询" if all(p.artifact == artifact for p in found) else "这次合并的输入查询" + who = who or ("这次查询" if all(p.artifact == artifact for p in found) else "这次合并的输入查询") return (f" （{who}未通过 SQL 检查：{'；'.join(problems[:3]) or '有错误级的问题'}。其中的数存疑：" "引用时须如实说明，不要当作确定的结论）") @@ -3776,6 +4002,141 @@ def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, return lines +#: 写作目录里「分析结论」一段开头的写法（给模型看）。判定由平台的算子算好、系统按模板渲染，写作者只引用（00-总方案 D5） +ANALYSIS_PROMPT = ( + "分析结论（系统按平台的方法算好的判定。判定语、具名的数、结果表都只能用引用标记写，系统照结论渲染、逐字核对）：\n" + "写法：判定语单独写 [[a:A1]]，系统换成整句判定；具名的数写 [[a:A1.键]]，换算同指标，比重、变化率这类比率写成 " + "[[a:A1.top_share|pct]] 显示成百分数；结果表的一格写 [[v:A1.r0.列]]（行号从 0 数），要列出多行多列，单独一行写 " + "[[table:A1 cols=列a,列b rows=0-4]]；拿它当结论的依据，句末写 [[see:A1]]。" + "「主要来自」「超出常规波动」「呈上升趋势」这类判断已经写在判定语里：引用判定语，不要自己改写判定；" + "算不了的照实写算不了的原因。下面没有的编号、键、列不要编造。" +) +#: 受管级别的出具、契约没开「单元格引用」时跟在上面那段后面（给模型看） +ANALYSIS_SEE_ONLY_PROMPT = ("这次出具不允许直接引用 Agent 现场做的分析：标着只能当依据的几条，判定语、数和结果表都不能写进" + "正文，只能在句末写 [[see:A1]] 这样的依据；分析卡做的照常引用。") +#: 一条分析结论最多列几个具名的数 +_PROMPT_VALUES = 16 + + +def _analysis_sources(entry: dict[str, Any], catalog: dict[str, Any]) -> str: + """「，来自 Q2、Q3」「，下钻自 A1」：按输入的工件对到这份目录里的编号（agent 调用时写的是它自己的编号，对不上目录）。 + 目录里没有的输入照实说有几份不在。""" + by_artifact = {str(e.get("artifact")): alias for alias, e in catalog.items() + if isinstance(e, dict) and e.get("kind") in ("query", LEDGER_KIND) and e.get("artifact")} + inputs = [str(a) for a in entry.get("inputs") or [] if a and str(a) != str(entry.get("artifact") or "")] + named = [by_artifact[a] for a in inputs if a in by_artifact] + queries = [a for a in named if not _is_analysis_alias(a)] + parents = [a for a in named if _is_analysis_alias(a)] + missing = len(inputs) - len(named) + return "".join([f"，来自 {'、'.join(queries)}" if queries else "", + f"，下钻自 {'、'.join(parents)}" if parents else "", + f"，另有 {missing} 份输入不在这份目录里" if missing else ""]) + + +def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, *, + table: bool = True) -> list[str]: + """一条分析结论在写作目录里的样子：编号、方法、来自哪几个查询，判定语（[[a:A1]] 渲染出来的那一句）、判定标准、具名的数 + （可引用的键，值按渲染规则写）、附带说明、输入查询的 SQL 检查问题、算不了的原因；table 为真时再列结果表的前 + table_top_k 段，其余写「其余 n 段」。agent 写的假设不列：它不是结论，只在证据面板显示。 + + 受管级别的正式出具、契约没开「单元格引用」时，Agent 现场做的分析只列判定语和判定标准，写明只能当依据（[[see:A1]]）， + 不列数和表：列出来的数写作者只能照抄成没有出处的数。""" + alias = entry["alias"] + head = f"- {alias}：{method_label(entry.get('op'), entry.get('mode'))}{_analysis_sources(entry, catalog)}" + result, why = snaps.analysis(alias, entry) + if result is None: + return [f"{head}（{why}，不要引用）"] + lines = [head] + if note := _sql_check_note(entry.get("artifact"), snaps, "这次分析的输入查询"): + lines.append(note) + phrase = _clean(str(result.get("phrase") or "")) + see_only = not cells_allowed and not analysis_card(entry) + if see_only: + lines.append(f" 判定语：{phrase or '没有'}（这次出具不允许直接引用 Agent 现场做的分析，只能在句末写 " + f"[[see:{alias}]] 作依据）") + else: + lines.append(f" 判定语 [[a:{alias}]] = {phrase}" if phrase else f" （没有判定语，不要写 [[a:{alias}]]）") + if result.get("status") == "refused": + lines.append(f" 算不了：{_clean(str(result.get('reason') or '没有写原因'))}。这条结论不能当「异常」「主要来自」" + "「呈上升趋势」这类判断的依据，只能照实写算不了的原因") + elif criterion := _clean(str(result.get("criterion") or "")): + lines.append(f" 判定标准：{criterion}") + caveats = [_clean(str(c.get("text") or "")) for c in result.get("caveats") or [] if isinstance(c, dict)] + if caveats := [c for c in caveats if c]: + lines.append(f" 附带说明：{'；'.join(caveats)}") + if see_only: + return lines + values = result.get("values") if isinstance(result.get("values"), dict) else {} + shown = [] + for key, raw in list(values.items())[:_PROMPT_VALUES]: + try: + shown.append(f"{key}={render_cell(raw)}") + except RenderError: + shown.append(f"{key}={MISSING}") + if shown: + more = f"，另有 {len(values) - _PROMPT_VALUES} 个" if len(values) > _PROMPT_VALUES else "" + lines.append(f" 具名的数（写 [[a:{alias}.键]]）：{'，'.join(shown)}{more}") + snapshot, why = snaps.table(alias, entry) + if why is not None: + return lines + columns = [str(c) for c in snapshot.get("columns") or []][:_PROMPT_COLS] + labels = snapshot.get("labels") or {} + rows = snapshot["rows"] + named = "、".join(f"{c}（{_clean(str(labels[c]))}）" if labels.get(c) else c for c in columns) + if not table: + lines.append(f" 结果表共 {len(rows)} 段，列 {named}（这里没列出各段；要列出就单独一行写 [[table:{alias}]]）") + return lines + settings = result.get("settings") if isinstance(result.get("settings"), dict) else {} + top = settings.get("table_top_k") + top = top if isinstance(top, int) and not isinstance(top, bool) and top > 0 else _TOP_K + lines.append(f" 结果表（共 {len(rows)} 段）：列 {named}") + plain = snaps.plain(entry.get("artifact"), snapshot) + for r in range(min(top, len(rows))): + cells = [] + for c in columns: + try: + raw, _ = locate_cell(snapshot, r, c) + cells.append(f"{c}={render_cell(snaps.shown(raw), kind=column_kind(snapshot, c), plain=c in plain)}") + except (CellError, RenderError): + cells.append(f"{c}={MISSING}") + lines.append(f" r{r}：" + "，".join(cells)) + if len(rows) > top: + lines.append(f" …其余 {len(rows) - top} 段（这里只列出前 {top} 段；后面的行也能按行号引用）") + return lines + + +#: 结论里没记 table_top_k 时写作目录列几段（同分析设置的缺省） +_TOP_K = 10 + + +@dataclass +class _AnalysisLines: + """写作目录里「分析结论」一段：每条结论两种样子——full 连同结果表的前 table_top_k 段，compact 只有判定、判定标准、 + 具名的数和说明。放得下时全按 full；放不下时先按 compact 列出，再按顺序补上结果表，排在查询的行前面（_trimmed_catalog）。""" + + order: list[str] + full: dict[str, list[str]] + compact: dict[str, list[str]] + #: 有 Agent 现场做的分析只能当依据（受管级别的出具、契约没开「单元格引用」） + see_only: bool = False + + @classmethod + def of(cls, catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) -> "_AnalysisLines": + entries = [e for e in catalog.values() if isinstance(e, dict) and e.get("kind") == LEDGER_KIND] + return cls([e["alias"] for e in entries], + {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed) for e in entries}, + {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed, table=False) for e in entries}, + see_only=not cells_allowed and any(not analysis_card(e) for e in entries)) + + def lines(self, tier: dict[str, str] | None = None) -> list[str]: + """这一段的文字：tier 里记成 compact 的按 compact，其余按 full。没有分析结论时是空的（目录和以前一字不差）。""" + if not self.order: + return [] + shapes = [self.compact[a] if (tier or {}).get(a) == "compact" else self.full[a] for a in self.order] + head = [f"\n{ANALYSIS_PROMPT}", *([ANALYSIS_SEE_ONLY_PROMPT] if self.see_only else [])] + return [*head, *(line for shape in shapes for line in shape)] + + #: 写作目录的字数预算（catalog_budget）按写作模型的上下文窗口定，不再写死 12000： #: - 占窗口的 40%。其余留给写作规则和指令（约 4 千字）、重写轮次（每轮把上一稿和违规清单追加进对话，最多 3 轮）、 #: 推理模型的思考和报告正文（输出额度）。字数按 1 字 ≈ 1 token 算：中文一个字约 0.6–1 个 token，目录里的数字、 @@ -3879,6 +4240,7 @@ def _metric_prompt(catalog: dict[str, Any]) -> list[str]: caveat += "（所依据的查询未通过 SQL 检查，结果存疑：引用时须如实说明，不要当作确定的结论）" if e.get("no_data"): caveat += NO_DATA_METRIC_NOTE + caveat += _comparability_note(e) lines.append(f"- [[{e['alias']}]] {e.get('name')} = {e.get('rendered')}{caveat}") inputs = [e for e in catalog.values() if e.get("kind") == "input"] if inputs: @@ -3887,6 +4249,21 @@ def _metric_prompt(catalog: dict[str, Any]) -> list[str]: return lines +def _comparability_note(entry: dict[str, Any]) -> str: + """指标上记的可比性（analysis.contract.MetricComparability）在写作目录里跟在值后面的那句：有条件可比、不可比照实写出 + 原因（一句一个），可比的不写。不可比的环比、同比不能当结论，引用时要照实说。""" + comparability = entry.get("comparability") + if not isinstance(comparability, dict): + return "" + reasons = "；".join(str(r) for r in comparability.get("reasons") or [] if r) \ + or str(comparability.get("phrase") or "") + if comparability.get("verdict") == "incomparable": + return f"（两期不可比：{reasons or '原因见证据面板'}。这个比较不能当结论，引用时须如实说明两期不可比）" + if comparability.get("verdict") == "conditional": + return f"（两期有条件可比：{reasons or '原因见证据面板'}。引用时须如实说明这几点）" + return "" + + def _dig_path(data: Any, path: str) -> Any: for key in path.split("."): data = data.get(key) if isinstance(data, dict) else None @@ -4011,13 +4388,15 @@ def writer_catalog(catalog: dict[str, Any], *, budget: int = DEFAULT_CATALOG_BUD [{node_id, title, data, evidence}]，titles 是节点在画布上的名字（分组标题用）。 放得下就整份给（没有结构化字段时和以前一字不差）。放不下时按优先级、按来源均衡（_trimmed_catalog）： - 口径卡指标、运行输入、结构化字段一律列出；查询按产出它的节点分组、轮流展开；没展开的列编号和摘要； - 开头照实说目录有省略、各来源省略了几次。 + 口径卡指标、运行输入、结构化字段一律列出；分析结论的判定、具名的数一律列出，结果表的各段排在查询的行前面补； + 查询按产出它的节点分组、轮流展开；没展开的列编号和摘要；开头照实说目录有省略、各来源省略了几次。 + 没有分析结论的目录和以前一字不差。 """ snaps = _Snapshots(loader) head = ["可引用的证据（写数字只能用下面的引用标记，系统会换成真实数值）："] field_lines, empty, cited = _fields_prompt(fields or [], catalog, snaps, cells_allowed) first = [*_metric_prompt(catalog), *field_lines] + analyses = _AnalysisLines.of(catalog, snaps, cells_allowed) queries = [e for e in catalog.values() if e.get("kind") == "query"] rules = ([f"\n{CELL_RULES}" if cells_allowed else "\n查询结果（这次出具不允许直接引用单元格，只能写在 [[see:Q1]] 这样的依据里）："] if queries else []) @@ -4030,26 +4409,31 @@ def writer_catalog(catalog: dict[str, Any], *, budget: int = DEFAULT_CATALOG_BUD for e in retrievals: tail.extend(_retrieval_prompt(e, snaps)) full = {e["alias"]: _query_prompt(e, snaps, cells_allowed) for e in queries} - text = "\n".join([*head, *first, *rules, *(line for e in queries for line in full[e["alias"]]), *tail]) + text = "\n".join([*head, *first, *analyses.lines(), *rules, *(line for e in queries for line in full[e["alias"]]), + *tail]) if len(text) <= budget: return WriterCatalog(text=text, budget=budget, queries=len(queries), empty_sources=empty) return _trimmed_catalog(catalog, head, first, rules, queries, full, tail, budget=budget, snaps=snaps, - cells_allowed=cells_allowed, titles=titles or {}, empty=empty, pinned=cited) + cells_allowed=cells_allowed, titles=titles or {}, empty=empty, pinned=cited, + analyses=analyses) def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], rules: list[str], queries: list[dict[str, Any]], full: dict[str, list[str]], tail: list[str], *, budget: int, snaps: _Snapshots, cells_allowed: bool, titles: dict[str, str], empty: list[str], - pinned: list[str]) -> WriterCatalog: + pinned: list[str], analyses: _AnalysisLines | None = None) -> WriterCatalog: """放不下时的目录。以前从头截断：查询按编号排，前面的节点占满预算，后面的节点一次都不露面。 - 1. 口径卡指标、运行输入、结构化字段一律列出（第一优先） + 1. 口径卡指标、运行输入、结构化字段一律列出（第一优先）；分析结论的判定、判定标准、具名的数、说明也一律列出 2. 表和字段、知识库检索两节合起来最多占预算的 1/_PROMPT_TAIL_SHARE - 3. 每次查询先只占一行摘要；结构化字段引用到的查询先展开（pinned），再按产出它的节点分组、轮流展开： + 3. 分析结论的结果表（前 table_top_k 段）先补：判定是从原始明细算出来的结论，比原始明细更值得留 + 4. 每次查询先只占一行摘要；结构化字段引用到的查询先展开（pinned），再按产出它的节点分组、轮流展开： 第一轮每次列前 _PROMPT_COMPACT_ROWS 行，第二轮补到 _PROMPT_ROWS 行，放不下为止 - 4. 开头写明目录有省略、各来源省略了几次，以及没展开的查询怎么用（OMITTED_RULES） - 最后整份量一次：超了就把最后展开的退回去；指标和字段本身就超（极少见）才从尾部截断。 + 5. 开头写明目录有省略、各来源省略了几次，以及没展开的查询怎么用（OMITTED_RULES） + 最后整份量一次：超了就把最后展开的退回去（先退查询，最后才退分析结论的结果表）；指标和字段本身就超（极少见） + 才从尾部截断。 """ + analyses = analyses or _AnalysisLines([], {}, {}) if tail and len("\n".join(tail)) > budget // _PROMPT_TAIL_SHARE: tail = _cut_lines(tail, budget // _PROMPT_TAIL_SHARE, "…（表和字段、知识库检索太长，后面的省略了）") tables: dict[str, list[str]] = {} @@ -4064,6 +4448,7 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], brief = {e["alias"]: [_query_brief(e, tables.get(e["alias"], []), snaps)] for e in queries} shapes = {"brief": brief, "compact": compact, "full": full} tier = {e["alias"]: "brief" for e in queries} + tier.update({alias: "compact" for alias in analyses.order}) def rows() -> list[dict[str, Any]]: out = [] @@ -4081,7 +4466,10 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], notice = (f"注意：证据太多，这份目录有省略（篇幅上限约 {budget} 字）。共 {len(queries)} 次查询：" f"{n['full']} 次列出前 {_PROMPT_ROWS} 行，{n['compact']} 次只列前 {_PROMPT_COMPACT_ROWS} 行，" f"{n['omitted']} 次只列编号和摘要。各来源：{per}。") - body = [*head, notice, OMITTED_RULES, *first, *rules] + if bare := sum(1 for alias in analyses.order if tier[alias] == "compact"): + notice += (f"另有 {len(analyses.order)} 条分析结论，判定和具名的数都列出了，其中 {bare} 条没列出结果表的各段" + "（要列出就单独一行写整表引用，系统照结果表生成）。") + body = [*head, notice, OMITTED_RULES, *first, *analyses.lines(tier), *rules] for r, entries in zip(stats, groups.values()): body.append(f"\n来自「{r['title']}」的查询（共 {r['total']} 次，展开 {r['full'] + r['compact']} 次）：") for e in entries: @@ -4098,6 +4486,12 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], # 省略情况那句里的数字会变长几位：留一点余量，最后再整份量一次 remaining = budget - len(assemble()) - 8 * (len(groups) + 3) upgraded: list[tuple[str, str]] = [] + for alias in analyses.order: + # 分析结论的结果表先补，排在查询的行前面；整份量下来超了，最后才退回它们 + if (delta := size(analyses.full[alias]) - size(analyses.compact[alias])) <= remaining: + tier[alias] = "full" + remaining -= delta + upgraded.append((alias, "compact")) for alias in pinned: if tier.get(alias) == "brief" and (delta := size(compact[alias]) - size(brief[alias])) <= remaining: tier[alias] = "compact" @@ -4249,6 +4643,141 @@ def describe_violations(violations: list[dict[str, Any]], *, limit: int = 20) -> return "\n".join(lines) +# -------------------------------------------------------------------------- +# 分析结论的出处：复算、代入式（证据面板、报告导出、可验证报告同一套） +# -------------------------------------------------------------------------- + +#: 复算不一致时点名哪几项不同的叫法（analysis.contract.AnalysisResult 的字段） +_RESULT_FIELDS = { + "op": "算子", "mode": "模式", "method": "方法版本", "params": "参数", "settings": "阈值", "inputs": "输入", + "probes": "现查的表范围", "status": "算成没算成", "reason_code": "算不了的原因", "reason": "算不了的原因", + "table": "结果表", "values": "具名的数", "verdict": "判定", "criterion": "判定标准", "phrase": "判定语", + "caveats": "附带说明", +} +#: 复算抛 NotImplementedError：这一版系统里还没有这种分析的算子 +RECHECK_MISSING = "系统里还没有这种分析的复算实现" + + +def analysis_input_snapshot(content: Any, artifact: str, *, ref: str | None = None, + node_id: str | None = None) -> dict[str, Any] | None: + """一份查询快照的内容 → 复算要的输入快照（analysis.contract.InputSnapshot：数据源查询工具的 to_payload 加上出处）。 + 不能当维度、度量的列（masked）按快照里记下的、查询当时的遮罩：复算按运行当时的条件重算；现在设的遮罩只管展示。 + 不是查询快照（没有 rows）的返回 None。""" + if not isinstance(content, dict) or not isinstance(content.get("rows"), list): + return None + return {"ref": str(ref or ""), "artifact": str(artifact), "node_id": str(node_id or ""), + "source": content.get("source"), "sql": str(content.get("sql") or ""), + "columns": [str(c) for c in content.get("columns") or []], + "column_types": content.get("column_types") if isinstance(content.get("column_types"), dict) else {}, + "rows": content["rows"], "truncated": bool(content.get("truncated")), + "masked": sorted(_mask_names(content.get("mask_columns")))} + + +def analysis_recheck(result: dict[str, Any], load: Callable[[str], Any], *, + load_result: Callable[[str], Any] | None = None) -> dict[str, Any]: + """把一份封存的分析结论按它记下的输入复算一遍（app.analysis.recompute：按 sha256 取回输入快照，按结论里记下的参数、 + 阈值和现查结果算，不再现查），和封存的逐字段比（contract.same，agent 写的假设不比）。 + + 交回 {status, text, method, method_now?, fields?}：status 是 same（一致）/ different（不一致）/ failed（复算不了）； + text 是给人看的一句，证据面板、报告导出、可验证报告照它写。方法升过版的照实写运行时用的是哪一版、现在是哪一版， + 复算结果同不同。复算抛异常（算子还没有、输入取不回、调用的写法现在不认）不往外抛，照实写「复算失败：原因」。""" + from app import analysis + from app.analysis.contract import NOT_RECOMPUTED, CallError, same + from app.core.errors import describe_exception + + method = str(result.get("method") or "") + try: + again = analysis.recompute(result, load, load_result=load_result) # type: ignore[arg-type] + except NotImplementedError: + return {"status": "failed", "method": method, "text": f"复算失败：{RECHECK_MISSING}"} + except CallError as e: + return {"status": "failed", "method": method, "text": f"复算失败：{e.message}"} + except Exception as e: # noqa: BLE001 - 复算出了意外：照实说复算不了，不能让证据面板、导出跟着出错 + return {"status": "failed", "method": method, "text": f"复算失败：{describe_exception(e)}"} + if not isinstance(again, dict): + return {"status": "failed", "method": method, "text": "复算失败：复算没有交回结论"} + ok = same(result, again) + fields = sorted(k for k in set(result) | set(again) if k not in NOT_RECOMPUTED and result.get(k) != again.get(k)) + out: dict[str, Any] = {"status": "same" if ok else "different", "method": method} + now = str(again.get("method") or "") + if now != method: + out["method_now"] = now + out["text"] = f"运行时用的是 {method}，现在是 {now}，复算结果{'相同' if ok else '不同'}" + elif ok: + out["text"] = "复算一致：按封存的输入重新计算，结论逐项相同" + else: + named = "、".join(dict.fromkeys(_RESULT_FIELDS.get(k, k) for k in fields)) + out["text"] = f"复算不一致：{named}和封存的不同" + if fields: + out["fields"] = fields + return out + + +def _decompose_row(result: dict[str, Any]) -> list[Any] | None: + """变化拆解的结果表里判定点名的那一段（verdict.subjects 的第一段，按维度列 params.by 对），对不上就取第一行（结果表 + 按变化的绝对值从大到小排）。""" + table = result.get("table") if isinstance(result.get("table"), dict) else {} + rows = [r for r in table.get("rows") or [] if isinstance(r, list)] + columns = [str(c) for c in table.get("columns") or []] + if not rows: + return None + subjects = ((result.get("verdict") or {}).get("subjects") or []) if isinstance(result.get("verdict"), dict) else [] + by = (result.get("params") or {}).get("by") if isinstance(result.get("params"), dict) else None + by = [str(b) for b in (by if isinstance(by, list) else [by] if by else []) if str(b) in columns] + if subjects and by and isinstance(subjects[0], list): + wanted = [str(v) for v in subjects[0]] + for row in rows: + if [str(row[columns.index(b)]) for b in by if columns.index(b) < len(row)] == wanted: + return row + return rows[0] + + +def analysis_formulas(result: dict[str, Any]) -> list[dict[str, str]]: + """出处链「计算」一步里关键数的代入式：[{label, text}]，数照单元格的规则渲染（系统写的，不是模型写的）。只给拿结论里 + 记下的数就写得出来的几条——变化拆解的总变化、变化率、主要来源的占比，比率拆解的结构效应、自身效应占比，波动基线的 + 常规波动范围和稳健 z；缺了哪个数、渲染不出来的那条不给。比如「华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ -194 = 61.9%」。""" + values = result.get("values") if isinstance(result.get("values"), dict) else {} + settings = result.get("settings") if isinstance(result.get("settings"), dict) else {} + op, mode = result.get("op"), result.get("mode") + out: list[dict[str, str]] = [] + + def add(label: str, build: Callable[[], str]) -> None: + try: + out.append({"label": label, "text": build()}) + except (KeyError, IndexError, ValueError, TypeError, RenderError): + pass # 缺一个数、渲染不出来：这条不给，不编 + + if result.get("status") != "ok": + return out + if op == "decompose" and mode == "additive": + add("总变化", lambda: f"总变化 = 本期合计 − 基期合计 = {render_cell(values['target_total'])} − " + f"{render_cell(values['base_total'])} = {render_cell(values['delta_total'])}") + add("变化率", lambda: f"变化率 = 总变化 ÷ 基期合计 = {render_cell(values['delta_total'])} ÷ " + f"{render_cell(values['base_total'])} = {render_cell(values['pct_total'], 'pct')}") + table = result.get("table") or {} + columns = [str(c) for c in table.get("columns") or []] + row = _decompose_row(result) + if row is not None and {"delta", "share"} <= set(columns): + by = [c for c in columns if c not in ("base", "target", "delta", "pct", "share", "flag")] + name = " · ".join(str(row[columns.index(c)]) for c in by if columns.index(c) < len(row)) or "这一段" + add("占比", lambda: f"{name}的占比 = {name}的变化 ÷ 总变化 = " + f"{render_cell(row[columns.index('delta')])} ÷ {render_cell(values['delta_total'])} = " + f"{render_cell(row[columns.index('share')], 'pct')}") + elif op == "decompose" and mode == "ratio": + for part, label in (("mix", "结构效应"), ("rate", "自身效应")): + add(f"{label}占比", lambda part=part, label=label: ( + f"{label}占比 = {label}合计 ÷ 整体的变化 = {render_cell(values[f'{part}_total'])} ÷ " + f"{render_cell(values['delta'])} = {render_cell(values[f'{part}_share'], 'pct')}")) + elif op == "baseline": + add("常规波动范围", lambda: ( + f"常规波动范围 = 中位数 ± {settings['k_warn']:g} 倍稳健标准差 = {render_cell(values['band_lo'])} 至 " + f"{render_cell(values['band_hi'])}（中位数 {render_cell(values['median'])}，" + f"过去 {render_cell(values['n_history'])} 期）")) + add("稳健 z", lambda: f"稳健 z = （本期 − 中位数）÷ 稳健标准差 = （{render_cell(values['x'])} − " + f"{render_cell(values['median'])}）÷ 稳健标准差 = {render_cell(values['z'])}") + return out + + # -------------------------------------------------------------------------- # 旧运行：按数值猜「可能的来源」 # diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index f3efa3cf..e21ff892 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -146,11 +146,12 @@ def report_catalog(state: GraphState, spec: GraphSpec, node: GraphNode) -> dict[ if isinstance(sources, str) and sources != "ancestors": sources = [sources] if isinstance(sources, list): - # evidence_from 只管查询、检索这类证据（连同查询当时的表结构）从哪几个节点来；指标从哪来由 metrics_from 管。 - # 写了子工作流节点的，它交回来的（子图里的查询，节点 id「子工作流节点 id/…」）一并算 + # evidence_from 只管查询、检索、分析结论这类证据（连同查询当时的表结构）从哪几个节点来；指标从哪来由 metrics_from 管。 + # 分析结论按做它的节点认（Agent，或者分析卡）。写了子工作流节点的，它交回来的（子图里的查询，节点 id + # 「子工作流节点 id/…」）一并算 keep = {str(s) for s in sources} - ledger = [e for e in ledger - if e.get("kind") not in ("query", "retrieval", "schema") or _outer(e.get("node_id")) in keep] + ledger = [e for e in ledger if e.get("kind") not in ("query", "retrieval", "schema", "analysis") + or _outer(e.get("node_id")) in keep] nodes = state.get("nodes") or {} catalog = build_catalog( nodes=nodes, diff --git a/backend/app/engine/sql_problems.py b/backend/app/engine/sql_problems.py index e67916eb..c1e60aef 100644 --- a/backend/app/engine/sql_problems.py +++ b/backend/app/engine/sql_problems.py @@ -8,7 +8,9 @@ - 写作目录里提醒写作者（engine/evidence.catalog_prompt）。 合并查询在内存 SQLite 上执行，不对照数据目录，它的快照没有自己的 checks，但记着 inputs：顺着往下找，问题记在 -出问题的那条源查询上。几个指标、几处引用出自同一条查询时，问题只算一次（problems_by_query 按快照 id 归并）。 +出问题的那条源查询上。分析结论（app/analysis 的工件）同样记着 inputs（它用到的查询快照）：报告引用了一份分析结论， +它的输入查询上的问题照样算——写作目录随 A 条目提醒，出具契约按引用的查询降档。几个指标、几处引用出自同一条查询时， +问题只算一次（problems_by_query 按快照 id 归并）。 """ from __future__ import annotations @@ -17,6 +19,7 @@ from collections.abc import Callable, Iterable from dataclasses import dataclass, field from typing import Any +from app.analysis.contract import is_result from app.core import artifact_store #: 合并查询快照的 source（engine/merge_query.MERGE_SOURCE）。这里写死同一个字面量，免得为一个常量引入执行器 @@ -57,9 +60,10 @@ def _load(artifact: str, loader: Callable[[str], Any] | None) -> Any: def query_problems(artifact: Any, *, loader: Callable[[str], Any] | None = None, _seen: set[str] | None = None, _node: str | None = None, _depth: int = 0) -> list[QueryProblem]: - """这份查询快照（或合并快照的各个输入）里有 error 级 SQL 检查的那几条查询，一条一个 QueryProblem。 + """这份查询快照（或合并快照、分析结论的各个输入）里有 error 级 SQL 检查的那几条查询，一条一个 QueryProblem。 - 读不出来的快照、没有问题的快照都不出现。自引用、成环的坏快照不会卡死。 + 读不出来的快照、没有问题的快照都不出现。自引用、成环的坏快照不会卡死。分析结论的输入没记节点（node_id 为 None）， + 调用方按证据台账补（node_of）；下钻的上一层结论（role parent）也是输入，顺着它再往下找。 """ if not isinstance(artifact, str) or not artifact or _depth > _MAX_DEPTH: return [] @@ -75,7 +79,7 @@ def query_problems(artifact: Any, *, loader: Callable[[str], Any] | None = None, errors = [c for c in checks if isinstance(c, dict) and c.get("level") == "error"] if isinstance(checks, list) else [] if errors: out.append(QueryProblem(artifact, _node, errors)) - inputs = snapshot.get("inputs") if snapshot.get("source") == MERGE_SOURCE else None + inputs = snapshot.get("inputs") if snapshot.get("source") == MERGE_SOURCE or is_result(snapshot) else None for entry in inputs if isinstance(inputs, list) else []: if isinstance(entry, dict) and isinstance(entry.get("artifact"), str): node = entry.get("node_id") if isinstance(entry.get("node_id"), str) else None diff --git a/backend/tests/test_analysis_evidence.py b/backend/tests/test_analysis_evidence.py new file mode 100644 index 00000000..0109765f --- /dev/null +++ b/backend/tests/test_analysis_evidence.py @@ -0,0 +1,533 @@ +"""分析结论进证据链（docs/design/analysis-round/00-总方案.md §3.3，第 1 波 W1-b）：引擎这一层的纯函数。 + +- 写作目录按台账编 A1、A2……（只收报告祖先里的），条目带着从工件取回、复验过哈希的整份结论（contract.catalog_result）； + 存进报告文档的目录只留摘要 +- [[a:A1]] 渲染判定语（片段种类 analysis，整段涂白、不算裸数字），[[a:A1.键|换算]] 渲染具名标量，[[v:A1.r0.列]]、 + [[table:A1 …]] 取结果表（表头写给人看的列名），[[see:A1]] 当依据 +- 受管级别的正式出具、契约没开「单元格引用」时，Agent 现场做的分析和它查的格同一个规矩，分析卡做的照常引用（D10） +- 写作目录的「分析结论」一段：判定语、判定标准、具名的数、结果表前 table_top_k 段、附带说明、算不了的原因、来自哪几个 Q， + 放不下时结果表的各段排在查询的行前面补 +- 分析结论的输入查询上的 SQL 检查问题随 A 条目出现；口径卡指标上的可比性照实写进写作目录 +- 复算（analysis_recheck）、代入式（analysis_formulas） +""" +from __future__ import annotations + +from typing import Any + +import pytest + +from app import analysis +from app.analysis.contract import CallError, ledger_entry +from app.analysis.testing import fake_result +from app.core.artifact_store import canonical_json, content_hash +from app.engine.evidence import ( + ANALYSIS_CELLS_REASON, + StreamRenderer, + analysis_formulas, + analysis_input_snapshot, + analysis_recheck, + build_catalog, + compose_doc, + iter_segments, + render_markers, + resolve_ref, + verify_doc, + writer_catalog, +) +from app.engine.sql_problems import query_problems + +#: 虚构零售库 shop：四个地区两周的销售额 +SNAPSHOT: dict[str, Any] = { + "columns": ["region", "week", "gmv"], "column_types": {"region": "text", "week": "text", "gmv": "number"}, + "rows": [["华东", "W41", 820], ["华东", "W42", 700], ["华北", "W41", 600], ["华北", "W42", 560], + ["华南", "W41", 700], ["华南", "W42", 666], ["西南", "W41", 100], ["西南", "W42", 100]], + "sql": "SELECT region, week, SUM(gmv) AS gmv FROM orders GROUP BY region, week ORDER BY region, week", + "source": "shop", "truncated": False, "row_count": 8, +} + + +class Store(dict): + """内容寻址的假工件库：put 落一份、交回 sha256；loader 是 artifact_store.load 的约定（取不到 None）。""" + + def put(self, obj: Any) -> str: + sha = content_hash(canonical_json(obj)) + self[sha] = obj + return sha + + def load(self, artifact: str) -> Any: + return self.get(artifact) + + +def setup(*, item: str | None = "drop", call_id: str | None = None, result: dict[str, Any] | None = None, + snapshot: dict[str, Any] | None = None) -> tuple[Store, list[dict[str, Any]], str, str]: + """一次查询（Q1）加一份按地区拆的分析结论（A1）：(工件库, 台账, 查询快照, 分析结论)。""" + store = Store() + q = store.put(snapshot or SNAPSHOT) + res = result or fake_result(inputs=[{"ref": "Q2", "artifact": q, "role": "data", + "columns": ["region", "week", "gmv"], "rows": "all"}]) + a = store.put(res) + ledger = [{"kind": "query", "node_id": "fetch", "exec": 1, "artifact": q, "tool": "db_query__shop", + "source": "shop", "columns": ["region", "week", "gmv"], "rows": 8, "truncated": False}, + ledger_entry(res, node_id="an", exec_no=1, artifact=a, item=item, call_id=call_id)] + return store, ledger, q, a + + +def catalog_of(**kw: Any) -> tuple[Store, dict[str, Any]]: + store, ledger, _, _ = setup(**kw) + return store, build_catalog(nodes={}, ledger=ledger, loader=store.load) + + +def cites(doc: dict[str, Any]) -> list[tuple[str, str, str | None]]: + return [(s["kind"], s["text"], s.get("issue")) for s in iter_segments(doc) if s.get("ref")] + + +# -------------------------------------------------------------------------- +# 契约里补的两样：叫法、按形状认分析结论 +# -------------------------------------------------------------------------- + + +def test_method_labels_and_result_shapes(): + from app.analysis.contract import OP_LABELS, OPS, is_result, method_label + + assert set(OP_LABELS) == set(OPS) + assert method_label("decompose", "additive") == "变化拆解（可加）" and method_label("baseline", None) == "波动基线" + assert method_label("differ", "rate") == "差异检验（比率）" and method_label("odd", "x") == "odd（x）" + assert is_result(fake_result()) and is_result(fake_result("trend", verdict="rising")) + assert not is_result(SNAPSHOT) and not is_result({"op": "decompose", "method": 1, "inputs": []}) + assert not is_result(None) and not is_result([fake_result()]) + + +# -------------------------------------------------------------------------- +# 写作目录的 A 条目 +# -------------------------------------------------------------------------- + + +def test_analyses_are_numbered_by_ledger_order_and_carry_the_whole_result(): + store, ledger, q, a = setup() + second = fake_result("baseline", verdict="within_band", inputs=[{"ref": "Q1", "artifact": q, "role": "data"}]) + b = store.put(second) + ledger.append(ledger_entry(second, node_id="trend_agent", exec_no=1, artifact=b, call_id="call-1")) + catalog = build_catalog(nodes={}, ledger=ledger, loader=store.load) + + assert [k for k, e in catalog.items() if e["kind"] == "analysis"] == ["A1", "A2"] + first = catalog["A1"] + assert first["result"] is store[a] and first["item"] == "drop" and "call_id" not in first + assert first["op"] == "decompose" and first["verdict"] == "driver" and first["inputs"] == [q] and first["rows"] == 4 + assert first["label"].startswith("变化拆解（可加） · 销售额下降") + assert catalog["A2"]["call_id"] == "call-1" and catalog["A2"]["label"].startswith("波动基线 · ") + + +def test_only_analyses_from_allowed_nodes_get_a_number(): + store, ledger, q, _ = setup() + late = fake_result(inputs=[{"ref": "Q1", "artifact": q, "role": "data"}], phrase="下游的结论") + ledger.insert(0, ledger_entry(late, node_id="late", exec_no=1, artifact=store.put(late), item="x")) + catalog = build_catalog(nodes={}, ledger=ledger, loader=store.load, allowed={"fetch", "an"}) + assert [k for k, e in catalog.items() if e["kind"] == "analysis"] == ["A1"] + assert catalog["A1"]["node_id"] == "an" + + +def test_unreadable_results_are_listed_with_the_reason_and_cannot_be_cited(): + store, ledger, _, a = setup() + store[a] = {**store[a], "phrase": "被改过的判定"} + + def tampered(artifact: str) -> Any: + if artifact == a: + raise ValueError("哈希不一致") + return store.get(artifact) + + catalog = build_catalog(nodes={}, ledger=ledger, loader=tampered) + assert "result" not in catalog["A1"] and catalog["A1"]["unreadable"] == "A1 的分析结论与哈希不一致，疑似被修改" + cite = resolve_ref("a:A1", catalog, loader=tampered) + assert cite["status"] == "unresolved" and cite["reason"] == "A1 的分析结论与哈希不一致，疑似被修改" + + gone = build_catalog(nodes={}, ledger=ledger, loader=lambda _: None) + assert gone["A1"]["unreadable"] == "A1 的分析结论已不存在，无法读取" + shaped = build_catalog(nodes={}, ledger=ledger, loader=lambda art: {"rows": []} if art == a else store.get(art)) + assert shaped["A1"]["unreadable"] == "A1 的工件不是分析结论" + + +def test_evidence_from_filters_analyses_by_the_node_that_made_them(): + from app.engine.nodes.report import report_catalog + from app.engine.schema import GraphSpec + + store, ledger, _, _ = setup() + + def node(nid: str, ntype: str, **config: Any) -> dict[str, Any]: + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + spec = GraphSpec.model_validate({ + "nodes": [node("fetch", "tool", tool="x"), node("an", "tool", tool="x"), + node("write", "report", evidence_from=["fetch"])], + "edges": [{"source": "fetch", "target": "an"}, {"source": "an", "target": "write"}]}) + write = spec.node_map()["write"] + state = {"nodes": {}, "evidence": ledger, "input": {}} + from unittest.mock import patch + + with patch("app.engine.evidence.load_artifact", store.load): + assert set(report_catalog(state, spec, write)) == {"Q1"} + write.config["evidence_from"] = ["fetch", "an"] + assert set(report_catalog(state, spec, write)) == {"Q1", "A1"} + write.config["evidence_from"] = "ancestors" + assert set(report_catalog(state, spec, write)) == {"Q1", "A1"} + + +# -------------------------------------------------------------------------- +# 引用的解析和渲染 +# -------------------------------------------------------------------------- + + +def test_the_judgment_phrase_is_one_system_rendered_segment_and_not_bare_numbers(): + store, catalog = catalog_of() + doc = compose_doc("[[a:A1]]。[[see:A1]]", catalog, loader=store.load) + [seg] = [s for s in iter_segments(doc) if s.get("ref")] + assert seg["kind"] == "analysis" and seg["state"] == "deterministic" + assert seg["text"] == "销售额下降 194 万（-8.7%），主要来自华东（-120 万，占 62%）" + assert seg["cite"]["kind"] == "analysis" and seg["cite"]["role"] == "value" and seg["cite"]["locator"] == {} + # 判定语里的 194、8.7、120、62 都是系统写的：不算裸数字 + assert doc["violations"] == [] and doc["stats"]["uncited_numbers"] == 0 + assert doc["blocks"][0]["units"][0]["cites"] == ["A1"] + assert verify_doc(doc, catalog, loader=store.load)["ok"] + + +def test_named_values_render_like_cells_with_the_metric_conversions(): + store, catalog = catalog_of() + doc = compose_doc("华东占 [[a:A1.top_share|pct]]，总变化 [[a:A1.delta_total]]，最大的是 [[a:A1.top_key]]。", + catalog, loader=store.load) + assert cites(doc) == [("number", "61.9%", None), ("number", "-194", None), ("value", "华东", None)] + share = next(s for s in iter_segments(doc) if s.get("ref") == "a:A1.top_share|pct") + assert share["cite"]["locator"] == {"key": "top_share"} and share["cite"]["conv"] == "pct" + assert share["cite"]["value"] == pytest.approx(0.6186) + assert verify_doc(doc, catalog, loader=store.load)["ok"] + + +@pytest.mark.parametrize("ref, reason", [ + ("a:A9", "引用的分析结论 A9 不存在"), + ("a:A1.nope", "A1 中没有名为「nope」的数"), + ("a:A1|pct", "判定语是一整句话，不能换算"), + ("a:A1.top_key|pct", "「华东」不是数值，不能换算为「pct」"), + ("a:Q1", "分析结论的引用格式无法识别"), + ("a:A1.top_share.x", "分析结论的引用格式无法识别"), + ("v:A1.r9.share", "A1 请求第 9 行，但结果表只有 4 行（行号从 0 开始）"), + ("v:A1.r0.nope", "A1 没有列「nope」，可用的列：region、base、target、delta、pct、share、flag"), + ("v:A9.r0.share", "引用的分析结论 A9 不存在"), +]) +def test_wrong_analysis_references_say_why(ref, reason): + store, catalog = catalog_of() + cite = resolve_ref(ref, catalog, loader=store.load) + assert cite["status"] == "unresolved" and cite["reason"] == reason + + +def test_a_q_alias_is_not_an_analysis(): + store, catalog = catalog_of() + catalog["A2"] = dict(catalog["Q1"]) + assert resolve_ref("a:A2", catalog, loader=store.load)["reason"] == "A2 不是分析结论" + + +def test_result_table_cells_and_whole_tables_use_the_labels_as_headers(): + store, catalog = catalog_of() + doc = compose_doc("华东变化 [[v:A1.r0.delta]]。\n\n[[table:A1 cols=region,delta,share rows=0-1]]", + catalog, loader=store.load) + assert "| 地区 | 变化 | 占变化的比重 |" in doc["markdown"] + assert ("number", "-120", None) in cites(doc) and ("value", "华北", None) in cites(doc) + cell = next(s for s in iter_segments(doc) if s.get("ref") == "v:A1.r0.delta") + assert cell["cite"]["kind"] == "cell" and cell["cite"]["alias"] == "A1" + assert cell["cite"]["locator"] == {"row": 0, "column": "delta"} + assert doc["violations"] == [] + assert verify_doc(doc, catalog, loader=store.load)["ok"] + + +def test_header_labels_with_digits_are_not_bare_numbers(): + result = fake_result() + result["table"]["labels"] = {**result["table"]["labels"], "base": "基期（2026-W41）", "target": "本期（2026-W42）"} + store, ledger, q, _ = setup(result={**result, "inputs": [{"ref": "Q2", "artifact": "x" * 64, "role": "data"}]}) + catalog = build_catalog(nodes={}, ledger=ledger, loader=store.load) + doc = compose_doc("[[table:A1 cols=region,base,target rows=0-1]]", catalog, loader=store.load) + assert "| 地区 | 基期（2026-W41） | 本期（2026-W42） |" in doc["markdown"] + assert doc["violations"] == [] and verify_doc(doc, catalog, loader=store.load)["ok"] + + +def test_dimension_values_with_digits_written_as_text_are_names(): + result = fake_result() + result["table"]["rows"][0][0] = "华东3区" + store, ledger, _, _ = setup(result=result, snapshot={**SNAPSHOT, "rows": [["华北", "W41", 1]]}) + catalog = build_catalog(nodes={}, ledger=ledger, loader=store.load) + doc = compose_doc("下降最多的是华东3区。[[see:A1]]", catalog, loader=store.load) + assert doc["violations"] == [] + + +def test_see_accepts_analyses_rows_and_the_a_prefix(): + store, catalog = catalog_of() + doc = compose_doc("下降主要来自华东。[[see:A1]] 第一段同理。[[see:A1.r0,a:A1]] 越界。[[see:A1.r9]]", + catalog, loader=store.load) + sees = [c for _, u in __import__("app.engine.evidence", fromlist=["iter_units"]).iter_units(doc) for c in u["see"]] + assert [(c["alias"], c["status"]) for c in sees] == [("A1", "resolved"), ("A1", "resolved"), ("A1", "resolved"), + ("A1.r9", "unresolved")] + assert sees[-1]["reason"] == "A1 只有 4 行，没有第 9 行（行号从 0 开始）" + assert sees[0]["kind"] == "analysis" + + +def test_the_doc_keeps_a_summary_and_reloads_the_result_from_the_artifact(): + store, catalog = catalog_of() + doc = compose_doc("[[a:A1]]，华东占 [[a:A1.top_share|pct]]。", catalog, loader=store.load) + stored = doc["catalog"]["A1"] + assert "result" not in stored and stored["artifact"] and stored["rows"] == 4 and stored["verdict"] == "driver" + # 文档里的摘要照样解析得了：结论按工件取回 + assert resolve_ref("a:A1", doc["catalog"], loader=store.load)["status"] == "resolved" + assert verify_doc(doc, doc["catalog"], loader=store.load)["ok"] + + +def test_a_tampered_phrase_in_the_doc_is_a_render_mismatch(): + store, catalog = catalog_of() + doc = compose_doc("[[a:A1]]。", catalog, loader=store.load) + seg = next(s for s in iter_segments(doc) if s.get("ref")) + seg["text"] = seg["cite"]["rendered"] = seg["text"].replace("华东", "华南") + doc["markdown"] = doc["markdown"].replace("华东", "华南") + codes = [v["code"] for v in verify_doc(doc, catalog, loader=store.load)["violations"]] + assert "render_mismatch" in codes + + +def test_streaming_renders_the_phrase_like_the_whole_text(): + store, catalog = catalog_of() + text = "结论：[[a:A1]]。华东占 [[a:A1.top_share|pct]]%。[[see:A1]]" + whole = render_markers(text, catalog, loader=store.load) + for size in (1, 3, 7): + renderer = StreamRenderer(catalog, loader=store.load) + out = "".join(renderer.feed(text[i:i + size]) for i in range(0, len(text), size)) + renderer.flush() + assert out == whole + # 具名标量后面重复写的单位去掉（同指标） + assert "61.9%。" in whole and "61.9%%" not in whole + + +# -------------------------------------------------------------------------- +# 受管级别的出具：Agent 现场做的分析和它查的格同一个规矩，分析卡做的总能引用（D10） +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize("ref", ["a:A1", "a:A1.top_share|pct", "v:A1.r0.share", "table:A1 cols=region,share"]) +def test_agent_analyses_follow_the_cells_rule_when_cells_are_off(ref): + store, catalog = catalog_of(item=None, call_id="call-7") + cite = resolve_ref(ref, catalog, cells_allowed=False, loader=store.load) + assert cite["status"] == "unresolved" and cite["reason"] == ANALYSIS_CELLS_REASON + assert resolve_ref(ref, catalog, cells_allowed=True, loader=store.load)["status"] == "resolved" + + +@pytest.mark.parametrize("ref", ["a:A1", "a:A1.top_share|pct", "v:A1.r0.share", "table:A1 cols=region,share"]) +def test_card_analyses_are_always_citable(ref): + store, catalog = catalog_of(item="drop") + assert resolve_ref(ref, catalog, cells_allowed=False, loader=store.load)["status"] == "resolved" + + +def test_query_cells_keep_their_reason_and_see_stays_allowed(): + store, catalog = catalog_of(item=None, call_id="call-7") + from app.engine.evidence import CELLS_REASON, resolve_support + + assert resolve_ref("v:Q1.r0.gmv", catalog, cells_allowed=False, loader=store.load)["reason"] == CELLS_REASON + assert resolve_support("A1", catalog, loader=store.load)["status"] == "resolved" + doc = compose_doc("下降主要来自华东。[[see:A1]]", catalog, cells_allowed=False, loader=store.load) + assert doc["violations"] == [] + + +# -------------------------------------------------------------------------- +# 写作目录 +# -------------------------------------------------------------------------- + + +def test_writer_catalog_lists_the_analysis_with_how_to_cite_it(): + store, catalog = catalog_of() + text = writer_catalog(catalog, loader=store.load).text + section = text[text.index("分析结论（"):text.index("查询结果（")] + assert "判定语单独写 [[a:A1]]" in section and "[[see:A1]]" in section + assert "- A1：变化拆解（可加），来自 Q1" in section + assert "判定语 [[a:A1]] = 销售额下降 194 万（-8.7%），主要来自华东（-120 万，占 62%）" in section + assert "判定标准：一段占总变化一半以上算主要来源" in section + assert "具名的数（写 [[a:A1.键]]）：base_total=2,220，target_total=2,026，delta_total=-194" in section + assert "结果表（共 4 段）：列 region（地区）、base（基期）" in section + assert "r0：region=华东，base=820，target=700，delta=-120" in section + assert "hypothesis" not in section and "地区拖累" not in section # agent 写的假设不进写作目录 + # 分析结论排在查询前面 + assert text.index("分析结论（") < text.index("- Q1：") + + +def test_writer_catalog_shows_only_the_top_segments_and_says_how_many_are_left(): + result = fake_result(settings={**fake_result()["settings"], "table_top_k": 3}) + store, catalog = catalog_of(result=result) + text = writer_catalog(catalog, loader=store.load).text + assert "r2：region=华南" in text and "r3：" not in text.split("查询结果（")[0] + assert "…其余 1 段（这里只列出前 3 段；后面的行也能按行号引用）" in text + + +def test_refused_results_say_why_and_warn_off_judgments(): + result = fake_result("baseline", status="refused", reason_code="history_short", + caveats=[{"code": "weekday_effect", "text": "日数据有星期效应"}]) + store, catalog = catalog_of(result=result) + text = writer_catalog(catalog, loader=store.load).text + assert "判定语 [[a:A1]] = 变化拆解没有做成：历史只有 5 期，至少要 8 期" in text + assert "算不了：历史只有 5 期，至少要 8 期。这条结论不能当「异常」「主要来自」「呈上升趋势」这类判断的依据" in text + assert "附带说明：日数据有星期效应" in text + assert "结果表（共" not in text and "具名的数（写" not in text + + +def test_agent_analyses_are_see_only_when_cells_are_off(): + store, catalog = catalog_of(item=None, call_id="call-7") + text = writer_catalog(catalog, cells_allowed=False, loader=store.load).text + assert "只能在句末写 [[see:A1]] 作依据" in text + assert "具名的数（写" not in text and "r0：region=华东" not in text + assert "这次出具不允许直接引用 Agent 现场做的分析：标着只能当依据的几条" in text + card_store, card = catalog_of(item="drop") + shown = writer_catalog(card, cells_allowed=False, loader=card_store.load).text + assert "具名的数（写" in shown and "标着只能当依据的几条" not in shown + + +def test_drill_downs_and_inputs_outside_the_catalog_are_named(): + store, ledger, q, a = setup() + child = fake_result(inputs=[{"ref": "Q9", "artifact": q, "role": "data"}, + {"ref": "A3", "artifact": a, "role": "parent"}, + {"ref": "Q8", "artifact": "f" * 64, "role": "data"}]) + ledger.append(ledger_entry(child, node_id="an", exec_no=2, artifact=store.put(child), item="drill")) + catalog = build_catalog(nodes={}, ledger=ledger, loader=store.load) + text = writer_catalog(catalog, loader=store.load).text + assert "- A2：变化拆解（可加），来自 Q1，下钻自 A1，另有 1 份输入不在这份目录里" in text + + +def test_unreadable_results_are_listed_as_not_citable(): + store, ledger, _, a = setup() + catalog = build_catalog(nodes={}, ledger=ledger, loader=lambda art: None if art == a else store.get(art)) + assert "- A1：变化拆解（可加），来自 Q1（A1 的分析结论已不存在，无法读取，不要引用）" in \ + writer_catalog(catalog, loader=store.load).text + + +def test_trimmed_catalogs_keep_the_judgments_and_fill_result_tables_before_query_rows(): + store, ledger, q, _ = setup() + many = [{"kind": "query", "node_id": "fetch", "exec": i + 2, "artifact": store.put({**SNAPSHOT, "sql": f"-- {i}"}), + "tool": "db_query__shop", "source": "shop", "columns": SNAPSHOT["columns"], "rows": 8} for i in range(40)] + catalog = build_catalog(nodes={}, ledger=[*ledger, *many], loader=store.load) + full = writer_catalog(catalog, loader=store.load, budget=10 ** 6) + assert not full.trimmed + trimmed = writer_catalog(catalog, loader=store.load, budget=len(full.text) // 3) + assert trimmed.trimmed and len(trimmed.text) <= trimmed.budget + assert "判定语 [[a:A1]] = " in trimmed.text and "r0：region=华东，base=820" in trimmed.text + assert trimmed.omitted # 查询有省略，分析结论的结果表没被省掉 + + # 更紧：判定、判定标准、具名的数照样列出，结果表退成一句，开头照实说 + tight = writer_catalog(catalog, loader=store.load, budget=2000) + assert len(tight.text) <= 2000 and "判定语 [[a:A1]] = " in tight.text and "具名的数（写" in tight.text + assert "r0：region=华东，base=820" not in tight.text + assert "结果表共 4 段，列 region（地区）" in tight.text and "要列出就单独一行写 [[table:A1]]" in tight.text + assert "另有 1 条分析结论，判定和具名的数都列出了，其中 1 条没列出结果表的各段" in tight.text + + +def test_metric_comparability_is_spoken_in_the_writer_catalog(): + card = {"kind": "metric_set", "caliber": "周报", "caliber_version": "v1", "artifact": "c" * 64, "metrics": [ + {"id": "wow", "name": "环比", "value": -0.087, "unit": "", "comparability": { + "artifact": "d" * 64, "verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据", + "reasons": ["本期只有 3 天数据"]}}, + {"id": "yoy", "name": "同比", "value": 0.02, "unit": "", "comparability": { + "artifact": "e" * 64, "verdict": "conditional", "phrase": "两期有条件可比：两段的周末天数不同", + "reasons": ["两段的周末天数不同：4 天对 6 天"]}}, + {"id": "gmv", "name": "销售额", "value": 2026, "unit": "", "comparability": { + "artifact": "f" * 64, "verdict": "comparable", "phrase": "两期可比", "reasons": []}}]} + catalog = build_catalog(nodes={"card": card}, ledger=[]) + assert catalog["m:wow"]["comparability"]["verdict"] == "incomparable" + text = writer_catalog(catalog).text + assert "环比 = -0.087（两期不可比：本期只有 3 天数据。这个比较不能当结论，引用时须如实说明两期不可比）" in text + assert "同比 = 0.02（两期有条件可比：两段的周末天数不同：4 天对 6 天。引用时须如实说明这几点）" in text + assert "- [[m:gmv]] 销售额 = 2,026\n" in text + "\n" + + +def test_sql_check_problems_of_the_input_queries_follow_the_analysis(): + bad = {**SNAPSHOT, "checks": [{"code": "fanout_sum", "level": "error", + "message": "一对多关联后求和，金额被重复计算。改成先汇总再关联"}]} + store, ledger, q, a = setup(snapshot=bad) + found = query_problems(a, loader=store.load) + assert [p.artifact for p in found] == [q] and found[0].problems == ["一对多关联后求和，金额被重复计算"] + catalog = build_catalog(nodes={}, ledger=ledger, loader=store.load) + text = writer_catalog(catalog, loader=store.load).text + section = text[text.index("- A1："):text.index("查询结果（")] + assert "（这次分析的输入查询未通过 SQL 检查：一对多关联后求和，金额被重复计算。其中的数存疑" in section + + +# -------------------------------------------------------------------------- +# 复算与代入式 +# -------------------------------------------------------------------------- + + +def test_recheck_says_plainly_when_the_operators_are_not_there_yet(): + store, ledger, q, a = setup() + out = analysis_recheck(store[a], lambda sha: analysis_input_snapshot(store.get(sha), sha)) + assert out == {"status": "failed", "method": "decompose.additive@1", + "text": "复算失败：系统里还没有这种分析的复算实现"} + + +def test_recheck_compares_field_by_field_and_names_the_method_versions(monkeypatch): + store, _, q, a = setup() + sealed = store[a] + seen: list[Any] = [] + + def same(result, load, *, load_result=None): + seen.append(load(q)) + return {**result, "hypothesis": "复算时没有假设"} + + monkeypatch.setattr(analysis, "recompute", same) + out = analysis_recheck(sealed, lambda sha: analysis_input_snapshot(store.get(sha), sha, ref="Q2", node_id="fetch")) + assert out == {"status": "same", "method": "decompose.additive@1", "text": "复算一致：按封存的输入重新计算，结论逐项相同"} + assert seen[0]["rows"] == SNAPSHOT["rows"] and seen[0]["masked"] == [] and seen[0]["node_id"] == "fetch" + + monkeypatch.setattr(analysis, "recompute", lambda r, load, *, load_result=None: {**r, "phrase": "别的判定"}) + assert analysis_recheck(sealed, store.get)["text"] == "复算不一致：判定语和封存的不同" + + monkeypatch.setattr(analysis, "recompute", + lambda r, load, *, load_result=None: {**r, "method": "decompose.additive@2"}) + out = analysis_recheck(sealed, store.get) + assert out["status"] == "different" and out["method_now"] == "decompose.additive@2" + assert out["text"] == "运行时用的是 decompose.additive@1，现在是 decompose.additive@2，复算结果不同" + + def refuse(r, load, *, load_result=None): + raise CallError("missing_column", "Q2 里没有列「gmv」") + + monkeypatch.setattr(analysis, "recompute", refuse) + assert analysis_recheck(sealed, store.get)["text"] == "复算失败：Q2 里没有列「gmv」" + + def broken(r, load, *, load_result=None): + raise RuntimeError("输入快照取不回来") + + monkeypatch.setattr(analysis, "recompute", broken) + assert analysis_recheck(sealed, store.get)["text"] == "复算失败：输入快照取不回来" + + +def test_input_snapshots_for_recompute_keep_the_masks_of_the_query_time(): + snap = analysis_input_snapshot({**SNAPSHOT, "mask_columns": ["Phone", "email"]}, "a" * 64, ref="Q1") + assert snap["masked"] == ["email", "phone"] and snap["artifact"] == "a" * 64 and snap["ref"] == "Q1" + assert analysis_input_snapshot({"columns": []}, "a" * 64) is None + + +def test_formulas_substitute_the_key_numbers(): + formulas = {f["label"]: f["text"] for f in analysis_formulas(fake_result())} + assert formulas == { + "总变化": "总变化 = 本期合计 − 基期合计 = 2,026 − 2,220 = -194", + "变化率": "变化率 = 总变化 ÷ 基期合计 = -194 ÷ 2,220 = -8.7%", + "占比": "华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ -194 = 61.9%", + } + baseline = fake_result("baseline", verdict="outside_band", values={ + "x": "-0.087", "median": "-0.004", "band_lo": "-0.061", "band_hi": "0.053", "z": "-2.9", "n_history": "12"}) + texts = [f["text"] for f in analysis_formulas(baseline)] + assert texts == ["常规波动范围 = 中位数 ± 2 倍稳健标准差 = -0.061 至 0.053（中位数 -0.004，过去 12 期）", + "稳健 z = （本期 − 中位数）÷ 稳健标准差 = （-0.087 − -0.004）÷ 稳健标准差 = -2.9"] + assert analysis_formulas(fake_result(status="refused", reason_code="history_short")) == [] + assert analysis_formulas(fake_result(values={})) == [] + + +# -------------------------------------------------------------------------- +# 血缘（D16）：分析卡的结果不是口径卡的原料 +# -------------------------------------------------------------------------- + + +def test_metrics_reading_analysis_results_are_not_traced_to_raw_data(): + """口径卡读 nodes.<分析卡>.results 的输入：运行时（traced）、静态（static_stops）都认作追不到原始数据，停在分析卡。 + 分析结论的输入只认台账里的查询快照，复算即血缘；写死 tool_cell、agent_field 的几处不用改。""" + from app.engine.lineage import static_stops, traced, untraced + + item = {"path": "nodes.an.results.drop.values.top_share", "value": "0.6186", "node_id": "an", "via": "analysis"} + assert not traced(item) and untraced([item]) == [item] + stops = static_stops({"an": {"type": "analysis", "config": {}}, "fetch": {"type": "tool", "config": {}}}, + "nodes.an.results.drop.values.top_share * 100") + assert [(s.path, s.node_id) for s in stops] == [("nodes.an.results.drop.values.top_share", "an")]