mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
分析结论进证据链（引擎）：台账 kind analysis 的条目在写作目录里按台账先后编 A1、A2……（只收报告祖先里的，报告节点的「证据来自」按做它的节点过滤），条目照 catalog_result 的约定带着从工件取回、复验过哈希的整份结论，取不回、哈希不符的记明原因；存进报告文档的目录只留摘要，整份结论按工件取回。引用：[[a:A3]] 渲染判定语（一个片段，种类 analysis，整段涂白、不算裸数字），[[a:A3.键|换算]] 渲染具名标量（换算同指标），[[v:A3.r0.列]]、[[table:A3 …]] 取结果表（表头写结果表的列名，表头里的数不算裸数字），[[see:A3]]、[[see:A3.r0]] 当依据；写错的编号、键、越界的行、判定语加换算都照实说为什么。受管级别的正式出具、契约没开「单元格引用」时，Agent 现场做的分析和它查的格同一个规矩，分析卡做的照常引用。写作目录加「分析结论」一段（判定语、判定标准、具名的数、结果表前 table_top_k 段与其余几段、附带说明、算不了的原因、来自哪几个 Q、下钻自哪个 A、输入查询的 SQL 检查问题，外加引用写法一句），放不下时判定和具名的数一律列出、结果表排在查询的行前面补、最后才退；口径卡指标上的可比性（有条件可比、不可比）照实写出原因。SQL 问题顺着分析结论的输入往下找。另有复算（analysis_recheck：算子没落地、调用写错、出意外都照实写「复算失败：原因」，方法升版写明前后两版）、复算用的输入快照、关键数的代入式，证据面板、导出、可验证报告共用
This commit is contained in:
ThearchyHelios
+579
-50
@@ -17,8 +17,16 @@
|
||||
[[i:<输入字段>]] 运行输入
|
||||
[[t:<表>]] / [[c:<表>.<列>]] 实体：表 / 字段，显示名字本身（反引号里写的已知名字也会自动链接）
|
||||
[[q:K<n>|<逐字引文>]] 引文：必须在那次检索命中的片段里逐字出现，显示引文本身
|
||||
[[a:A<n>]] 分析结论的判定语：系统按模板渲染好的一整句（判定由平台的算子算出，写作者只引用）
|
||||
[[a:A<n>.<键>|万]] 分析结论的具名标量（values 里的一项），换算同 m:
|
||||
[[v:A<n>.r<行>.<列>]] 分析结论结果表里的一格；[[table:A<n> …]] 整表，表头写结果表的列名
|
||||
[[see:<ref>,<ref>…]] 依据：挂在句末，不渲染
|
||||
|
||||
分析结论（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<kind>[A-Za-z]+)[ \t]*:(?P<body>[^\[\]\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:<id>，只登记 m:<卡的节点 id>.<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<alias>[QK]\d+)\.r(?P<row>\d+)\.(?P<column>.+)$")
|
||||
_CELL_REF = re.compile(rf"^(?P<alias>[QK{CATALOG_PREFIX}]\d+)\.r(?P<row>\d+)\.(?P<column>.+)$")
|
||||
_NODE_REF = re.compile(r"^N:(?P<node>[^.\s]+)(?:\.(?P<field>.+))?$")
|
||||
|
||||
|
||||
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<alias>{CATALOG_PREFIX}\d+)(?:\.(?P<key>[^\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<alias>[QK]\d+)(?:\.r(?P<row>\d+))?$")
|
||||
_ROW_REF = re.compile(rf"^(?P<alias>[QK{CATALOG_PREFIX}]\d+)(?:\.r(?P<row>\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
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 旧运行：按数值猜「可能的来源」
|
||||
#
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")]
|
||||
Reference in New Issue
Block a user