From fa33b98fef84dd528cf0f7043fd3d76035862cb5 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 15:47:00 +0800 Subject: [PATCH] =?UTF-8?q?=E5=86=99=E4=BD=9C=E7=9B=AE=E5=BD=95=E6=9C=89?= =?UTF-8?q?=E7=9C=81=E7=95=A5=E6=97=B6=EF=BC=8C=E5=88=86=E6=9E=90=E7=BB=93?= =?UTF-8?q?=E8=AE=BA=E4=B8=80=E6=AE=B5=E6=9C=80=E5=A4=9A=E5=8D=A0=E9=A2=84?= =?UTF-8?q?=E7=AE=97=E7=9A=84=E4=B8=80=E5=8D=8A=EF=BC=88=5FPROMPT=5FANALYS?= =?UTF-8?q?IS=5FSHARE=EF=BC=89=EF=BC=9A=E6=AF=8F=E6=9D=A1=E7=BB=93?= =?UTF-8?q?=E8=AE=BA=E5=9B=9B=E7=A7=8D=E6=A0=B7=E5=AD=90=E4=BB=8E=E5=85=A8?= =?UTF-8?q?=E5=88=B0=E7=AE=80=EF=BC=88ANALYSIS=5FLEVELS=EF=BC=9Afull=20?= =?UTF-8?q?=E8=BF=9E=E5=90=8C=E7=BB=93=E6=9E=9C=E8=A1=A8=E7=9A=84=E5=89=8D?= =?UTF-8?q?=E5=87=A0=E6=AE=B5=E3=80=81compact=20=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E8=A1=A8=E5=8F=AA=E5=86=99=E4=B8=80=E5=8F=A5=E3=80=81brief=20?= =?UTF-8?q?=E5=8F=AA=E7=95=99=E5=88=A4=E5=AE=9A=E8=AF=AD=E5=92=8C=E5=85=B7?= =?UTF-8?q?=E5=90=8D=E7=9A=84=E6=95=B0=E3=80=81head=20=E5=8F=AA=E5=88=97?= =?UTF-8?q?=E7=BC=96=E5=8F=B7=E9=82=A3=E4=B8=80=E8=A1=8C=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E5=85=88=E9=83=BD=E6=8C=89=20compact=20=E5=88=97=EF=BC=8C?= =?UTF-8?q?=E8=B6=85=E4=BA=86=E6=8C=89=E7=BB=93=E8=AE=BA=E7=9A=84=E5=85=88?= =?UTF-8?q?=E5=90=8E=E4=BB=8E=E5=90=8E=E5=BE=80=E5=89=8D=E9=99=8D=E6=88=90?= =?UTF-8?q?=20brief=E3=80=81=E5=86=8D=E9=99=8D=E6=88=90=20head=EF=BC=9B?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E8=A1=A8=E7=9A=84=E6=AE=B5=E7=85=A7=E6=97=A7?= =?UTF-8?q?=E5=85=88=E4=BA=8E=E6=9F=A5=E8=AF=A2=E7=9A=84=E8=A1=8C=E8=A1=A5?= =?UTF-8?q?=EF=BC=8C=E4=BD=86=E8=BF=99=E4=B8=80=E6=AE=B5=E8=A1=A5=E5=AE=8C?= =?UTF-8?q?=E4=B9=9F=E4=B8=8D=E8=B6=85=E8=BF=87=E4=B8=8A=E9=99=90=E3=80=82?= =?UTF-8?q?=E5=BC=80=E5=A4=B4=E9=82=A3=E5=8F=A5=E7=85=A7=E5=AE=9E=E5=86=99?= =?UTF-8?q?=E5=90=84=E6=A0=B7=E5=AD=90=E5=90=84=E6=9C=89=E5=87=A0=E6=9D=A1?= =?UTF-8?q?=EF=BC=8C=E4=BB=A5=E5=8F=8A=E6=B2=A1=E5=88=97=E7=9A=84=E7=85=A7?= =?UTF-8?q?=E6=A0=B7=E8=83=BD=E5=BC=95=E7=94=A8=EF=BC=88=E5=88=A4=E5=AE=9A?= =?UTF-8?q?=E8=AF=AD=E5=86=99=20[[a:=E7=BC=96=E5=8F=B7]]=E3=80=81=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E8=A1=A8=E5=86=99=E6=95=B4=E8=A1=A8=E5=BC=95=E7=94=A8?= =?UTF-8?q?=E3=80=81=E6=B2=A1=E5=88=97=E5=87=BA=E7=9A=84=E5=85=B7=E5=90=8D?= =?UTF-8?q?=E7=9A=84=E6=95=B0=E4=B8=8D=E8=A6=81=E7=BC=96=E9=80=A0=EF=BC=89?= =?UTF-8?q?=EF=BC=9B=E5=8F=AA=E6=98=AF=E7=BB=93=E6=9E=9C=E8=A1=A8=E6=B2=A1?= =?UTF-8?q?=E5=88=97=E5=90=84=E6=AE=B5=E7=9A=84=E7=85=A7=E6=97=A7=E9=82=A3?= =?UTF-8?q?=E4=B8=80=E5=8F=A5=E3=80=82=E4=BB=A5=E5=89=8D=E8=BF=99=E4=B8=80?= =?UTF-8?q?=E6=AE=B5=E6=B2=A1=E6=9C=89=E4=B8=8A=E9=99=90=EF=BC=8C40=20?= =?UTF-8?q?=E4=BB=BD=E7=BB=93=E8=AE=BA=E3=80=81=E6=AF=8F=E4=BB=BD=201000?= =?UTF-8?q?=20=E6=AE=B5=E6=97=B6=E5=9C=A8=2051.2k=20=E7=9A=84=E9=A2=84?= =?UTF-8?q?=E7=AE=97=E9=87=8C=E5=8D=A0=2050.6k=EF=BC=8C=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E5=8F=AA=E5=89=A9=E4=B8=80=E8=A1=8C=E6=91=98=E8=A6=81=EF=BC=9B?= =?UTF-8?q?=E7=8E=B0=E5=9C=A8=E5=8D=A0=2025.6k=20=E4=BB=A5=E5=86=85?= =?UTF-8?q?=EF=BC=8C=E9=82=A3=E6=AC=A1=E6=9F=A5=E8=AF=A2=E5=88=97=E6=BB=A1?= =?UTF-8?q?=2020=20=E8=A1=8C=E3=80=82=E6=B5=8B=E8=AF=95=EF=BC=9A20=20?= =?UTF-8?q?=E4=BB=BD=E7=BB=93=E8=AE=BA=E6=88=AA=E5=88=B0=20compact=20+=20b?= =?UTF-8?q?rief=E3=80=8180=20=E4=BB=BD=E6=88=AA=E5=88=B0=20brief=20+=20hea?= =?UTF-8?q?d=EF=BC=8C=E8=BF=99=E4=B8=80=E6=AE=B5=E4=B8=8D=E8=B6=85?= =?UTF-8?q?=E8=BF=87=E4=B8=80=E5=8D=8A=E3=80=81=E8=B6=8A=E5=BE=80=E5=90=8E?= =?UTF-8?q?=E8=B6=8A=E7=AE=80=E3=80=81=E7=BC=96=E5=8F=B7=E9=83=BD=E5=9C=A8?= =?UTF-8?q?=E3=80=81=E6=9F=A5=E8=AF=A2=E7=85=A7=E6=A0=B7=E5=B1=95=E5=BC=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/evidence.py | 94 +++++++++++++++++++------ backend/tests/test_analysis_evidence.py | 61 ++++++++++++++++ 2 files changed, 132 insertions(+), 23 deletions(-) diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 7c74a1b7..d3ee3206 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -4060,11 +4060,18 @@ def _analysis_sources(entry: dict[str, Any], catalog: dict[str, Any]) -> str: f"，另有 {missing} 份输入不在这份目录里" if missing else ""]) +#: 写作目录里一条分析结论的几种样子，从全到简（_analysis_prompt 的 level）：full 连同结果表的前 table_top_k 段；compact 不列 +#: 各段，结果表只写一句；brief 只留判定语和具名的数；head 只列编号、方法和来自哪几个查询 +ANALYSIS_LEVELS = ("full", "compact", "brief", "head") + + def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, *, - table: bool = True) -> list[str]: + level: str = "full") -> list[str]: """一条分析结论在写作目录里的样子：编号、方法、来自哪几个查询，判定语（[[a:A1]] 渲染出来的那一句）、判定标准、具名的数 - （可引用的键，值按渲染规则写）、附带说明、输入查询的 SQL 检查问题、算不了的原因；table 为真时再列结果表的前 - table_top_k 段，其余写「其余 n 段」。agent 写的假设不列：它不是结论，只在证据面板显示。 + （可引用的键，值按渲染规则写）、附带说明、输入查询的 SQL 检查问题、算不了的原因；level 为 full 时再列结果表的前 + table_top_k 段，其余写「其余 n 段」。目录放不下时往简里降（ANALYSIS_LEVELS）：compact 结果表只写一句；brief 只留 + 判定语和具名的数（判定标准、附带说明、SQL 检查的提醒、算不了的原因那句不列，判定语里本来就写着）；head 只列编号那一行。 + agent 写的假设不列：它不是结论，只在证据面板显示。 受管级别的正式出具、契约没开「单元格引用」时，Agent 现场做的分析只列判定语和判定标准，写明只能当依据（[[see:A1]]）， 不列数和表：列出来的数写作者只能照抄成没有出处的数。""" @@ -4073,8 +4080,11 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna result, why = snaps.analysis(alias, entry) if result is None: return [f"{head}（{why}，不要引用）"] + if level == "head": + return [head] + brief = level == "brief" lines = [head] - if note := _sql_check_note(entry.get("artifact"), snaps, "这次分析的输入查询"): + if not brief and (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) @@ -4083,13 +4093,15 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna f"[[see:{alias}]] 作依据）") else: lines.append(f" 判定语 [[a:{alias}]] = {phrase}" if phrase else f" （没有判定语，不要写 [[a:{alias}]]）") - if result.get("status") == "refused": + if brief: + pass + elif 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]: + if not brief and (caveats := [c for c in caveats if c]): lines.append(f" 附带说明：{'；'.join(caveats)}") if see_only: return lines @@ -4103,6 +4115,8 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna if shown: more = f"，另有 {len(values) - _PROMPT_VALUES} 个" if len(values) > _PROMPT_VALUES else "" lines.append(f" 具名的数（写 [[a:{alias}.键]]）：{'，'.join(shown)}{more}") + if brief: + return lines snapshot, why = snaps.table(alias, entry) if why is not None: return lines @@ -4110,7 +4124,7 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna 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: + if level != "full": lines.append(f" 结果表共 {len(rows)} 段，列 {named}（这里没列出各段；要列出就单独一行写 [[table:{alias}]]）") return lines settings = result.get("settings") if isinstance(result.get("settings"), dict) else {} @@ -4138,12 +4152,13 @@ _TOP_K = 10 @dataclass class _AnalysisLines: - """写作目录里「分析结论」一段：每条结论两种样子——full 连同结果表的前 table_top_k 段，compact 只有判定、判定标准、 - 具名的数和说明。放得下时全按 full；放不下时先按 compact 列出，再按顺序补上结果表，排在查询的行前面（_trimmed_catalog）。""" + """写作目录里「分析结论」一段：每条结论四种样子（ANALYSIS_LEVELS，从全到简）。放得下时全按 full；放不下时这一段最多占 + 预算的 1/_PROMPT_ANALYSIS_SHARE：先按 compact 列出，还超就从后往前降成 brief、再降成 head，然后按顺序补上结果表，排在 + 查询的行前面（_trimmed_catalog）。""" order: list[str] - full: dict[str, list[str]] - compact: dict[str, list[str]] + #: 样子 → {编号: 那几行} + levels: dict[str, dict[str, list[str]]] #: 有 Agent 现场做的分析只能当依据（受管级别的出具、契约没开「单元格引用」） see_only: bool = False @@ -4151,15 +4166,19 @@ class _AnalysisLines: 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}, + {level: {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed, level=level) + for e in entries} for level in ANALYSIS_LEVELS}, see_only=not cells_allowed and any(not analysis_card(e) for e in entries)) + def shape(self, alias: str, level: str) -> list[str]: + return self.levels[level][alias] + def lines(self, tier: dict[str, str] | None = None) -> list[str]: - """这一段的文字：tier 里记成 compact 的按 compact，其余按 full。没有分析结论时是空的（目录和以前一字不差）。""" + """这一段的文字：每条按 tier 里记的样子，没记的按 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] + shapes = [self.shape(a, level if (level := (tier or {}).get(a)) in ANALYSIS_LEVELS else "full") + 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)] @@ -4179,6 +4198,8 @@ CATALOG_MAX_CHARS = 160_000 _PROMPT_COMPACT_ROWS = 3 #: 放不下时表和字段、知识库检索两节合起来最多占预算的几分之一（剩下的给查询） _PROMPT_TAIL_SHARE = 4 +#: 放不下时分析结论一段最多占预算的几分之一：几十份结论、每份上千段时，不能把查询挤得只剩一行摘要 +_PROMPT_ANALYSIS_SHARE = 2 _PROMPT_CUT = "\n…（目录太长，后面的省略了）" @@ -4454,16 +4475,17 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], pinned: list[str], analyses: _AnalysisLines | None = None) -> WriterCatalog: """放不下时的目录。以前从头截断：查询按编号排，前面的节点占满预算，后面的节点一次都不露面。 - 1. 口径卡指标、运行输入、结构化字段一律列出（第一优先）；分析结论的判定、判定标准、具名的数、说明也一律列出 + 1. 口径卡指标、运行输入、结构化字段一律列出（第一优先）；分析结论的判定、判定标准、具名的数、说明也列出，只是这一段 + 最多占预算的 1/_PROMPT_ANALYSIS_SHARE：超了就按结论的先后从后往前降，先只留判定语和具名的数，再只列编号 2. 表和字段、知识库检索两节合起来最多占预算的 1/_PROMPT_TAIL_SHARE - 3. 分析结论的结果表（前 table_top_k 段）先补：判定是从原始明细算出来的结论，比原始明细更值得留 + 3. 分析结论的结果表（前 table_top_k 段）先补（不超过那个上限）：判定是从原始明细算出来的结论，比原始明细更值得留 4. 每次查询先只占一行摘要；结构化字段引用到的查询先展开（pinned），再按产出它的节点分组、轮流展开： 第一轮每次列前 _PROMPT_COMPACT_ROWS 行，第二轮补到 _PROMPT_ROWS 行，放不下为止 5. 开头写明目录有省略、各来源省略了几次，以及没展开的查询怎么用（OMITTED_RULES） 最后整份量一次：超了就把最后展开的退回去（先退查询，最后才退分析结论的结果表）；指标和字段本身就超（极少见） 才从尾部截断。 """ - analyses = analyses or _AnalysisLines([], {}, {}) + 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]] = {} @@ -4496,9 +4518,7 @@ 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}。") - if bare := sum(1 for alias in analyses.order if tier[alias] == "compact"): - notice += (f"另有 {len(analyses.order)} 条分析结论，判定和具名的数都列出了，其中 {bare} 条没列出结果表的各段" - "（要列出就单独一行写整表引用，系统照结果表生成）。") + notice += _analysis_notice(Counter(tier[alias] for alias in analyses.order), len(analyses.order)) 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']} 次）：") @@ -4513,14 +4533,24 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], def size(lines: list[str]) -> int: return sum(len(line) + 1 for line in lines) + # 分析结论一段最多占预算的 1/_PROMPT_ANALYSIS_SHARE：超了按结论的先后从后往前降，先只留判定语和具名的数，再只列编号 + cap = budget // _PROMPT_ANALYSIS_SHARE + for level in ("brief", "head"): + for alias in reversed(analyses.order): + if size(analyses.lines(tier)) <= cap: + break + tier[alias] = level # 省略情况那句里的数字会变长几位：留一点余量，最后再整份量一次 remaining = budget - len(assemble()) - 8 * (len(groups) + 3) upgraded: list[tuple[str, str]] = [] + listed = size(analyses.lines(tier)) for alias in analyses.order: - # 分析结论的结果表先补，排在查询的行前面；整份量下来超了，最后才退回它们 - if (delta := size(analyses.full[alias]) - size(analyses.compact[alias])) <= remaining: + # 分析结论的结果表先补，排在查询的行前面（这一段照样不超过上限）；整份量下来超了，最后才退回它们 + delta = size(analyses.shape(alias, "full")) - size(analyses.shape(alias, "compact")) + if tier[alias] == "compact" and delta <= remaining and listed + delta <= cap: tier[alias] = "full" remaining -= delta + listed += delta upgraded.append((alias, "compact")) for alias in pinned: if tier.get(alias) == "brief" and (delta := size(compact[alias]) - size(brief[alias])) <= remaining: @@ -4554,6 +4584,24 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], omitted=[e["alias"] for e in queries if tier[e["alias"]] == "brief"], empty_sources=empty) +#: 目录有省略时，开头那句里分析结论各种样子的说法（给模型看） +_ANALYSIS_LEVEL_WORDS = (("full", "列出了结果表的前几段"), ("compact", "没列出结果表的各段"), + ("brief", "只列了判定语和具名的数（判定标准、附带说明没列）"), ("head", "只列了编号")) + + +def _analysis_notice(levels: Counter, total: int) -> str: + """目录有省略时开头那句里分析结论的情况：都列全了的不说；只是结果表没列各段的照旧一句；还降到了只留判定语和具名的数、 + 只列编号的，照实写各有几条，以及没列的照样能引用。""" + if not total or levels["full"] == total: + return "" + if not levels["brief"] and not levels["head"]: + return (f"另有 {total} 条分析结论，判定和具名的数都列出了，其中 {levels['compact']} 条没列出结果表的各段" + "（要列出就单独一行写整表引用，系统照结果表生成）。") + parts = "，".join(f"{levels[level]} 条{word}" for level, word in _ANALYSIS_LEVEL_WORDS if levels[level]) + return (f"另有 {total} 条分析结论，这一段最多占篇幅的一半：{parts}。没列出的照样能引用：判定语单独写 [[a:编号]]，" + "系统照结论渲染；要列出结果表就单独一行写整表引用，系统照结果表生成；没列出的具名的数不要编造。") + + def catalog_prompt(catalog: dict[str, Any], *, budget: int = DEFAULT_CATALOG_BUDGET, cells_allowed: bool = True, loader: Loader | None = None, fields: list[dict[str, Any]] | None = None, titles: dict[str, str] | None = None) -> str: diff --git a/backend/tests/test_analysis_evidence.py b/backend/tests/test_analysis_evidence.py index 4bc1449f..b24c0bf4 100644 --- a/backend/tests/test_analysis_evidence.py +++ b/backend/tests/test_analysis_evidence.py @@ -14,6 +14,7 @@ from __future__ import annotations import asyncio +import re from concurrent.futures import ThreadPoolExecutor from datetime import date from typing import Any @@ -26,6 +27,7 @@ from app.analysis.contract import CallError, effective_settings, 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_LEVELS, ANALYSIS_CELLS_REASON, StreamRenderer, analysis_formulas, @@ -473,6 +475,65 @@ def test_trimmed_catalogs_keep_the_judgments_and_fill_result_tables_before_query assert "另有 1 条分析结论，判定和具名的数都列出了，其中 1 条没列出结果表的各段" in tight.text +def _analysis_part(text: str) -> str: + """写作目录里「分析结论」一段（从那段开头的说明到查询结果那段之前）。""" + start = text.index("\n分析结论（") + end = text.find("\n查询结果（", start) + return text[start:end if end >= 0 else len(text)] + + +def _many_analyses(n: int, queries: int = 30) -> tuple[Store, dict[str, Any]]: + """n 份分析结论（A1…An，内容各不相同）、1 + queries 次查询的写作目录。""" + store, ledger, q, _ = setup() + for i in range(n - 1): + res = fake_result(inputs=[{"ref": "Q2", "artifact": q, "role": "data", "columns": ["region", "week", "gmv"], + "rows": "all"}], hypothesis=f"假设 {i}") + ledger.append(ledger_entry(res, node_id="an", exec_no=1, artifact=store.put(res), item=f"i{i}")) + 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(queries)] + return store, build_catalog(nodes={}, ledger=[*ledger, *many], loader=store.load) + + +@pytest.mark.parametrize("n, first, last, words", [ + # 先截结果表的段，还超就从后往前只留判定语和具名的数 + (20, "compact", "brief", ("条没列出结果表的各段", "条只列了判定语和具名的数")), + # 都只留判定语和具名的数还超：再从后往前只列编号 + (80, "brief", "head", ("条只列了判定语和具名的数", "条只列了编号")), +]) +def test_the_analysis_part_of_a_trimmed_catalog_takes_at_most_half_of_the_budget(n, first, last, words): + """几十份结论时，分析结论一段最多占预算的一半：按结论的先后从后往前降，先截结果表的段、再只留判定语和具名的数、再只列 + 编号，开头照实写各有几条、没列的照样能引用；剩下的篇幅给查询展开。以前这一段没有上限，40 份结论、每份上千段时在 51.2k + 的预算里占了 50.6k，查询只剩一行摘要。""" + store, catalog = _many_analyses(n) + budget = 12_000 + writer = writer_catalog(catalog, loader=store.load, budget=budget) + assert writer.trimmed and len(writer.text) <= budget + part = _analysis_part(writer.text) + assert len(part) <= budget // 2 + blocks = [b.strip("\n") for b in re.split(r"\n(?=- A\d+：)", part)[1:]] + assert len(blocks) == n # 每条结论都还在，编号连着 + + def level(block: str) -> str: + if "\n" not in block: + return "head" + if "判定标准：" not in block: + return "brief" + return "full" if "\n r0：" in block else "compact" + + assert level(blocks[0]) == first and level(blocks[-1]) == last + assert blocks[-1].startswith(f"- A{n}：变化拆解（可加），来自 Q1") + for block in blocks: + if level(block) == "brief": + assert "判定语 [[a:" in block and "具名的数（写 [[a:" in block and "附带说明" not in block + order = [ANALYSIS_LEVELS.index(level(b)) for b in blocks] + assert order == sorted(order) # 越往后越简 + notice = next(line for line in writer.text.split("\n") if line.startswith("注意：")) + assert f"另有 {n} 条分析结论，这一段最多占篇幅的一半：" in notice and all(w in notice for w in words), notice + assert "没列出的照样能引用：判定语单独写 [[a:编号]]" in notice + assert writer.queries == 31 and writer.compact > 0 # 查询照样有展开的 + + 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": {