diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 62ebaf79..02a62c8b 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -47,7 +47,9 @@ import json import math import re import unicodedata +from collections import Counter from collections.abc import Callable +from dataclasses import dataclass, field from decimal import ROUND_HALF_UP, Context, Decimal from typing import Any, Literal, TypedDict @@ -57,6 +59,7 @@ from app.data.names import FILLERS, name_key from app.data.provenance_types import DOC_PROVENANCE from app.data.tabular import UNSHAPED_NOTE from app.engine.expressions import CellError, cell_value, column_kind, locate_cell +from app.engine.guards import FALLBACK_CONTEXT from app.engine.issuance import _tolerance, extract_numbers, number_allowance from app.engine.labels import option_label @@ -3094,10 +3097,12 @@ def _sql_check_note(artifact: Any, snaps: _Snapshots) -> str | None: "引用时须如实说明，不要当作确定的结论）") -def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) -> list[str]: +def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, + shown_rows: int = _PROMPT_ROWS) -> list[str]: """一次查询在写作目录里的样子：编号、来源、列，以及能引用的行（行号 + 渲染后的值）。 - 值按 [[v:]] 的渲染规则写出来：写作者在这里看到的，就是引用之后报告里显示的字。 + 值按 [[v:]] 的渲染规则写出来：写作者在这里看到的，就是引用之后报告里显示的字。shown_rows 是列出几行： + 目录放不下时先给每次查询列前几行（_PROMPT_COMPACT_ROWS），再补到 _PROMPT_ROWS。 """ alias = entry["alias"] cols = entry.get("columns") or [] @@ -3115,7 +3120,7 @@ def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) columns = [str(c) for c in snapshot.get("columns") or []][:_PROMPT_COLS] rows = snapshot["rows"] example = None - for r, record in enumerate(rows[:_PROMPT_ROWS]): + for r, record in enumerate(rows[:shown_rows]): cells = [] for c in columns: try: @@ -3130,22 +3135,78 @@ def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) lines.append(f" r{r}：" + "，".join(cells)) if not rows: lines.append(" （0 行，没有可以引用的格）") - elif len(rows) > _PROMPT_ROWS: - lines.append(f" …共 {len(rows)} 行，这里只列出前 {_PROMPT_ROWS} 行；后面的行也能按行号引用") + elif len(rows) > shown_rows: + lines.append(f" …共 {len(rows)} 行，这里只列出前 {shown_rows} 行；后面的行也能按行号引用") if rows and columns: lines.append(f" 例：[[v:{alias}.r0.{example or columns[0]}]]") return lines -def catalog_prompt(catalog: dict[str, Any], *, budget: int = 12000, cells_allowed: bool = True, - loader: Loader | None = None) -> str: - """证据目录的文字版，放进写作者的 prompt。超过预算就截断，并照实说截断了。 +#: 写作目录的字数预算（catalog_budget）按写作模型的上下文窗口定，不再写死 12000： +#: - 占窗口的 40%。其余留给写作规则和指令（约 4 千字）、重写轮次（每轮把上一稿和违规清单追加进对话，最多 3 轮）、 +#: 推理模型的思考和报告正文（输出额度）。字数按 1 字 ≈ 1 token 算：中文一个字约 0.6–1 个 token，目录里的数字、 +#: 列名、标点每个字更少，按 1:1 算偏保守，实际占用低于 40% +#: - 拿不到窗口时按 128K（和 Agent 护栏的 FALLBACK_CONTEXT 同一个保守值），即 51200 字。原来的 12000 字只放得下 +#: 20 多次查询：2026-10 一次上百次查询的运行，十万字以上的目录被截到一万两千字以内，写作者只看到前二十来次 +#: - 下限 12000：窗口很小（32K 以下）的本地模型维持原来的篇幅；上限 160000：百万窗口的模型每写一稿、每重写一次 +#: 都要重发整份目录，花费随之上涨，而 160000 字已经放得下那次运行的查询全部展开 +CATALOG_WINDOW_SHARE = 0.4 +CATALOG_MIN_CHARS = 12_000 +CATALOG_MAX_CHARS = 160_000 +#: 目录放不下时，先给每次查询列几行 +_PROMPT_COMPACT_ROWS = 3 +#: 放不下时表和字段、知识库检索两节合起来最多占预算的几分之一（剩下的给查询） +_PROMPT_TAIL_SHARE = 4 +_PROMPT_CUT = "\n…（目录太长，后面的省略了）" - 查询结果会列出能引用的行（要读快照）；cells_allowed=False 时只列编号，说明只能当依据。 - 沙箱代码节点不列：它的产出不能引用。 + +def catalog_budget(window: int | None) -> int: + """写作目录的字数预算：写作模型上下文窗口（token）的 40%，夹在上下限之间；拿不到窗口按 FALLBACK_CONTEXT。""" + size = window if isinstance(window, int) and not isinstance(window, bool) and window > 0 else FALLBACK_CONTEXT + return max(CATALOG_MIN_CHARS, min(CATALOG_MAX_CHARS, int(size * CATALOG_WINDOW_SHARE))) + + +DEFAULT_CATALOG_BUDGET = catalog_budget(None) + +#: 目录有省略时，开头跟在省略情况后面的约定（给模型看）。和数字核对一致：没展开的查询只能当依据，数照样逐字核对 +OMITTED_RULES = ( + "没展开的查询照样在证据台账里，只是这里没有列出它的数据：\n" + "- 不要据此断言「没有查到」「不存在」「缺失」「没有数据」，只能写「本报告未覆盖」，并说明依据不足；\n" + "- 要用没展开的查询，只能在句末写 [[see:Qn]] 作依据（Qn 换成它的编号）；不要写 [[v:]]、[[table:]] 去猜它的行和列，" + "也不要凭摘要写任何数字——数字照样逐字核对，没有出处的数会被打回重写。结构化结论里写好的引用标记可以照抄。" +) +#: 结构化结论为空的来源后面跟的那句（给模型看） +EMPTY_SOURCE_PROMPT = ("结构化结论为空（字段都记为空值）。这不代表没有数据：不得据此断言没有查到、不存在或缺失，" + "只能写本报告未覆盖") + + +@dataclass +class WriterCatalog: + """给写作者的出处清单（writer_catalog）：文字，以及省略了什么。 + + - trimmed：放不下、有省略；nodes 是各来源的查询数 [{node_id, title, total, full, compact, omitted}]， + full 列出前 20 行、compact 只列前几行、omitted 只列编号和摘要；omitted 是只列摘要的查询编号 + - empty_sources：结构化结论为空的来源（Agent 的名字） + 清单有省略、或有来源的结构化结论为空时，报告节点加写作规则（ABSENCE_RULE）；有省略时组装文档另查缺失断言 """ - snaps = _Snapshots(loader) - lines = ["可引用的证据（写数字只能用下面的引用标记，系统会换成真实数值）："] + + text: str + budget: int + queries: int = 0 + trimmed: bool = False + nodes: list[dict[str, Any]] = field(default_factory=list) + omitted: list[str] = field(default_factory=list) + empty_sources: list[str] = field(default_factory=list) + + @property + def guarded(self) -> bool: + """写作者看不全证据：缺失断言没有依据。""" + return self.trimmed or bool(self.empty_sources) + + +def _metric_prompt(catalog: dict[str, Any]) -> list[str]: + """口径卡指标和运行输入在写作目录里的样子。""" + lines: list[str] = [] groups: dict[tuple[str, str, str], list[dict[str, Any]]] = {} for entry in catalog.values(): if entry.get("kind") == "metric": @@ -3172,23 +3233,249 @@ def catalog_prompt(catalog: dict[str, Any], *, budget: int = 12000, cells_allowe if inputs: lines.append("\n运行输入：") lines.extend(f"- [[{e['alias']}]] = {e.get('rendered')}" for e in inputs) + return lines + + +def _dig_path(data: Any, path: str) -> Any: + for key in path.split("."): + data = data.get(key) if isinstance(data, dict) else None + return data + + +def _field_prompt(path: str, value: Any, ev: dict[str, Any], aliases: dict[str, str], snaps: _Snapshots, + cells_allowed: bool) -> str: + """一个结构化字段在写作目录里的那一行：值（按 [[v:]] 的渲染规则）和引用它的标记。""" + status = ev.get("status") + if status == "missing": + return f"- {path}：抽取时没有找到这个值（记为空，不代表库里没有）" + if status not in ("verified", "mismatch"): + return f"- {path}：出处核对不了（记为空）" + locator = ev.get("locator") if isinstance(ev.get("locator"), dict) else {} + alias = aliases.get(str(ev.get("artifact") or "")) + if alias is None: + return f"- {path}：所依据的查询不在这份目录里，不能引用" + if not cells_allowed: + return f"- {path}：出自 {alias}，只能在句末写 [[see:{alias}]] 作依据" + if isinstance(locator.get("rows"), list) and len(locator["rows"]) == 2: + first, last = locator["rows"] + named = locator.get("columns") if isinstance(locator.get("columns"), dict) else {"": locator.get("column")} + cols = ",".join(dict.fromkeys(str(c) for c in named.values() if c)) + many = f"（共 {last - first + 1} 行，整表最多 {TABLE_MAX_ROWS} 行，要分成几张表）" \ + if last - first + 1 > TABLE_MAX_ROWS else "" + return (f"- {path}：数组，出自 {alias} 第 {first}–{last} 行，可以单独一行写 " + f"[[table:{alias} cols={cols} rows={first}-{last}]]{many}") + row, column = locator.get("row"), locator.get("column") + shown = MISSING + snapshot, why = snaps.get(alias, ev.get("artifact")) + try: + if why: + raise CellError(why) + raw, column = locate_cell(snapshot, row, column) + shown = render_cell(raw, kind=column_kind(snapshot, column)) + except (CellError, RenderError): + try: + shown = render_cell(value) + except RenderError: + pass + return f"- {path} = {shown}：[[v:{alias}.r{row}.{column}]]" + + +def _fields_prompt(sources: list[dict[str, Any]], catalog: dict[str, Any], snaps: _Snapshots, + cells_allowed: bool) -> tuple[list[str], list[str], list[str]]: + """各 Agent 经按出处核对交来的结构化字段：(写作目录里的几行, 结构化结论为空的来源, 字段引用到的查询编号)。 + + 字段的出处记的是 Agent 自己的编号（它的第几次查询），这里换成报告目录里的全局编号：按快照工件认，同一个节点的 + 优先（工件内容寻址，两次一模一样的查询共用一份快照，指哪一次都是同一格）。 + """ + mine: dict[tuple[str, str], str] = {} + anyone: dict[str, str] = {} + for e in catalog.values(): + if e.get("kind") == "query": + mine.setdefault((str(e.get("node_id") or ""), str(e.get("artifact") or "")), e["alias"]) + anyone.setdefault(str(e.get("artifact") or ""), e["alias"]) + lines: list[str] = [] + empty: list[str] = [] + cited: list[str] = [] + for source in sources: + evidence = source.get("evidence") + if not isinstance(evidence, dict) or not evidence: + continue + node_id = str(source.get("node_id") or "") + title = str(source.get("title") or node_id) + data = source.get("data") if isinstance(source.get("data"), dict) else {} + if all(_dig_path(data, path) is None for path in evidence): + empty.append(title) + lines.append(f"「{title}」：{EMPTY_SOURCE_PROMPT}") + continue + aliases = {artifact: mine.get((node_id, artifact)) or alias for artifact, alias in anyone.items()} + lines.append(f"「{title}」：") + for path, ev in evidence.items(): + ev = ev if isinstance(ev, dict) else {} + lines.append(_field_prompt(path, _dig_path(data, path), ev, aliases, snaps, cells_allowed)) + if ev.get("status") in ("verified", "mismatch") and (alias := aliases.get(str(ev.get("artifact") or ""))): + cited.append(alias) + if not lines: + return [], [], [] + return (["\n各 Agent 交来的结构化结论（系统已逐格核对到查询快照；要写这些数，用后面的引用标记）：", *lines], + empty, list(dict.fromkeys(cited))) + + +def _query_brief(entry: dict[str, Any], tables: list[str]) -> str: + """没展开的查询在目录里的一行摘要：编号、查的表（没有表结构时写工具名）、行数。""" + rows = entry.get("rows") + what = ("、".join(tables[:3]) + (f" 等 {len(tables)} 张表" if len(tables) > 3 else "")) if tables \ + else str(entry.get("tool") or entry.get("source") or "查询") + size = f"{rows} 行" if isinstance(rows, int) else "行数不明" + return f" - {entry['alias']}：{what}，{size}" + ("（结果已截断）" if entry.get("truncated") else "") + + +def _cut_lines(lines: list[str], limit: int, note: str) -> list[str]: + """从头取到放得下 limit 个字为止，后面换成一句 note。""" + out: list[str] = [] + used = len(note) + 1 + for line in lines: + if used + len(line) + 1 > limit: + return [*out, note] + out.append(line) + used += len(line) + 1 + return out + + +def writer_catalog(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) -> WriterCatalog: + """证据目录的文字版，放进写作者的 prompt，连同省略了什么（WriterCatalog）。 + + 查询结果会列出能引用的行（要读快照）；cells_allowed=False 时只列编号，说明只能当依据。 + 沙箱代码节点不列：它的产出不能引用。fields 是上游 Agent 经按出处核对交来的结构化字段 + [{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] queries = [e for e in catalog.values() if e.get("kind") == "query"] - if queries: - lines.append(f"\n{CELL_RULES}" if cells_allowed else - "\n查询结果（这次出具不允许直接引用单元格，只能写在 [[see:Q1]] 这样的依据里）：") - for e in queries: - lines.extend(_query_prompt(e, snaps, cells_allowed)) + rules = ([f"\n{CELL_RULES}" if cells_allowed else + "\n查询结果（这次出具不允许直接引用单元格，只能写在 [[see:Q1]] 这样的依据里）："] if queries else []) + tail: list[str] = [] if (index := snaps.entities(catalog)).active: - lines.extend(_entity_prompt(catalog, partial=index.partial, snaps=snaps)) + tail.extend(_entity_prompt(catalog, partial=index.partial, snaps=snaps)) retrievals = [e for e in catalog.values() if e.get("kind") == "retrieval"] if retrievals: - lines.append(f"\n{QUOTE_RULES}") + tail.append(f"\n{QUOTE_RULES}") for e in retrievals: - lines.extend(_retrieval_prompt(e, snaps)) - text = "\n".join(lines) + 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]) + 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) + + +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: + """放不下时的目录。以前从头截断：查询按编号排，前面的节点占满预算，后面的节点一次都不露面。 + + 1. 口径卡指标、运行输入、结构化字段一律列出（第一优先） + 2. 表和字段、知识库检索两节合起来最多占预算的 1/_PROMPT_TAIL_SHARE + 3. 每次查询先只占一行摘要；结构化字段引用到的查询先展开（pinned），再按产出它的节点分组、轮流展开： + 第一轮每次列前 _PROMPT_COMPACT_ROWS 行，第二轮补到 _PROMPT_ROWS 行，放不下为止 + 4. 开头写明目录有省略、各来源省略了几次，以及没展开的查询怎么用（OMITTED_RULES） + 最后整份量一次：超了就把最后展开的退回去；指标和字段本身就超（极少见）才从尾部截断。 + """ + if tail and len("\n".join(tail)) > budget // _PROMPT_TAIL_SHARE: + tail = _cut_lines(tail, budget // _PROMPT_TAIL_SHARE, "…（表和字段、知识库检索太长，后面的省略了）") + tables: dict[str, list[str]] = {} + for e in catalog.values(): + if e.get("kind") == "table": + for alias in e.get("queries") or []: + tables.setdefault(alias, []).append(str(e.get("name"))) + groups: dict[str, list[dict[str, Any]]] = {} + for e in queries: + groups.setdefault(str(e.get("node_id") or ""), []).append(e) + compact = {e["alias"]: _query_prompt(e, snaps, cells_allowed, _PROMPT_COMPACT_ROWS) for e in queries} + brief = {e["alias"]: [_query_brief(e, tables.get(e["alias"], []))] for e in queries} + shapes = {"brief": brief, "compact": compact, "full": full} + tier = {e["alias"]: "brief" for e in queries} + + def rows() -> list[dict[str, Any]]: + out = [] + for nid, entries in groups.items(): + got = Counter(tier[e["alias"]] for e in entries) + out.append({"node_id": nid, "title": titles.get(nid) or nid or "其他", "total": len(entries), + "full": got["full"], "compact": got["compact"], "omitted": got["brief"]}) + return out + + def assemble() -> str: + stats = rows() + per = "、".join(f"{r['title']} {r['total']} 次（展开 {r['full'] + r['compact']} 次，省略 {r['omitted']} 次）" + for r in stats) + n = {k: sum(r[k] for r in stats) for k in ("full", "compact", "omitted")} + 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] + for r, entries in zip(stats, groups.values()): + body.append(f"\n来自「{r['title']}」的查询（共 {r['total']} 次，展开 {r['full'] + r['compact']} 次）：") + for e in entries: + if tier[e["alias"]] != "brief": + body.extend(shapes[tier[e["alias"]]][e["alias"]]) + if r["omitted"]: + body.append(f" 没展开（{r['omitted']} 次，只列编号和摘要）：") + body.extend(brief[e["alias"]][0] for e in entries if tier[e["alias"]] == "brief") + return "\n".join([*body, *tail]) + + def size(lines: list[str]) -> int: + return sum(len(line) + 1 for line in lines) + + # 省略情况那句里的数字会变长几位：留一点余量，最后再整份量一次 + remaining = budget - len(assemble()) - 8 * (len(groups) + 3) + upgraded: list[tuple[str, str]] = [] + for alias in pinned: + if tier.get(alias) == "brief" and (delta := size(compact[alias]) - size(brief[alias])) <= remaining: + tier[alias] = "compact" + remaining -= delta + upgraded.append((alias, "brief")) + for target, source in (("compact", "brief"), ("full", "compact")): + moved = True + while moved: + moved = False + for entries in groups.values(): + for e in entries: + alias = e["alias"] + if tier[alias] != source: + continue + delta = size(shapes[target][alias]) - size(shapes[source][alias]) + if delta <= remaining: + tier[alias] = target + remaining -= delta + upgraded.append((alias, source)) + moved = True + break + text = assemble() + while len(text) > budget and upgraded: + alias, source = upgraded.pop() + tier[alias] = source + text = assemble() if len(text) > budget: - text = text[:budget].rsplit("\n", 1)[0] + "\n…（目录太长，后面的省略了）" - return text + text = text[:budget - len(_PROMPT_CUT)].rsplit("\n", 1)[0] + _PROMPT_CUT + return WriterCatalog(text=text, budget=budget, queries=len(queries), trimmed=True, nodes=rows(), + omitted=[e["alias"] for e in queries if tier[e["alias"]] == "brief"], empty_sources=empty) + + +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: + """证据目录的文字版（writer_catalog 的文字）。超过预算时分优先级、按来源均衡，并照实说省略了什么。""" + return writer_catalog(catalog, budget=budget, cells_allowed=cells_allowed, loader=loader, fields=fields, + titles=titles).text def unshaped_table(entry: dict[str, Any], fetch: Callable[[Any], Any]) -> bool: diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index 1fcd50c4..a7d7cabc 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -50,11 +50,13 @@ from app.engine.evidence import ( MARKER_RULES, SEE_AND_CODE_RULES, StreamRenderer, + WriterCatalog, build_catalog, - catalog_prompt, + catalog_budget, compose_doc, describe_violations, ledger_enabled, + writer_catalog, ) from app.engine.nodes.llm import _invoke_streaming, _model_spec, _report_call from app.engine.schema import ( @@ -70,7 +72,8 @@ from app.engine.schema import ( ) from app.engine.state import GraphState, message_text, thinking_text from app.engine.toolcalls import leaked_markup, markup_warning -from app.providers.factory import ProviderNotConfigured, get_chat_model +from app.providers import catalog as model_catalog +from app.providers.factory import ModelSpec, ProviderNotConfigured, get_chat_model, resolve_provider ON_VIOLATION = ("fail", "flag") NUMBERS = ("strict", "off") @@ -140,6 +143,57 @@ def report_catalog(state: GraphState, spec: GraphSpec, node: GraphNode) -> dict[ return catalog +def report_fields(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[dict[str, Any]]: + """上游 Agent 经按出处核对交来的结构化字段：[{node_id, title, data, evidence}]，按画布上的顺序。 + + 范围和 report_catalog 收查询的范围一致：祖先节点，写了 evidence_from 就只收那几个。只进写作目录（第一优先， + 放不下也列出），不进目录本身：出口复核按目录重建，字段的引用落到的是查询单元格，和以前一样核对。 + """ + sources = _setting(spec, node, "evidence_from", "ancestors") + if isinstance(sources, str) and sources != "ancestors": + sources = [sources] + keep = {str(s) for s in sources} if isinstance(sources, list) else None + allowed = _ancestors(spec, node.id) + outputs = state.get("nodes") or {} + found: list[dict[str, Any]] = [] + for other in spec.nodes: + result = outputs.get(other.id) + if other.id not in allowed or (keep is not None and other.id not in keep) or not isinstance(result, dict): + continue + if isinstance(result.get("data_evidence"), dict) and result["data_evidence"]: + found.append({"node_id": other.id, "title": other.title or other.id, "data": result.get("data"), + "evidence": result["data_evidence"]}) + return found + + +async def model_window(session: Any, spec: ModelSpec, model_id: str) -> int | None: + """写作模型的上下文窗口（token）：模型接入里给这个模型登记的 context 优先，其次模型目录 + （providers/catalog.py）；都没有返回 None，由 catalog_budget 取保守值。""" + try: + provider = await resolve_provider(session, spec.provider, spec.model or model_id) + except ProviderNotConfigured: + provider = None + for entry in (provider.models or []) if provider is not None else []: + if isinstance(entry, dict) and entry.get("id") == model_id: + raw = entry.get("context") + try: + value = int(raw) if not isinstance(raw, bool) and raw not in (None, "") else 0 + except (TypeError, ValueError): + value = 0 + if value > 0: + return value + return model_catalog.find_context(model_id) + + +def _trimmed_message(writer: WriterCatalog) -> str: + """写作目录有省略时的那条警告（给人看）：各来源省略了几次，写作模型看不到哪些明细。""" + per = "、".join(f"{r['title']}省略 {r['omitted']} 次" for r in writer.nodes if r["omitted"]) + shown = writer.queries - len(writer.omitted) + return (f"证据目录超出篇幅（约 {writer.budget} 字，按写作模型的上下文窗口确定）：{writer.queries} 次查询中 " + f"{shown} 次列出了数据，{len(writer.omitted)} 次只列编号和摘要（{per}）。写作模型看不到这些查询的明细，" + "报告不得据此断言没有数据") + + def report_entities(spec: GraphSpec, node: GraphNode) -> str: """报告节点的实体层：link（默认）/ off。出口复核调 verify_doc 时传 entities=(这个值 == "link")。 @@ -351,8 +405,7 @@ def _repair_request(violations: list[dict[str, Any]]) -> str: ) -def _messages(ctx: NodeContext, state: GraphState, catalog: dict[str, Any], cells: bool = True, - claims: str = "off") -> list[BaseMessage]: +def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims: str = "off") -> list[BaseMessage]: system = ctx.render_str(ctx.cfg("system", ""), state) instructions = ctx.render_str(ctx.cfg("instructions", ""), state).strip() \ or "根据下面的证据写一份简洁的报告，先总后分。" @@ -363,7 +416,7 @@ def _messages(ctx: NodeContext, state: GraphState, catalog: dict[str, Any], cell return [ SystemMessage(content="\n\n".join(p for p in (system, ROLE) if p)), HumanMessage(content=( - f"{instructions}\n\n{catalog_prompt(catalog, cells_allowed=cells)}\n\n{rules}\n\n" + f"{instructions}\n\n{writer.text}\n\n{rules}\n\n" "只输出报告正文（Markdown），不要写撰写说明。" )), ] @@ -392,6 +445,8 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: model, model_id = await get_chat_model(session, _model_spec(ctx)) except ProviderNotConfigured as e: raise NodeError(ctx.node.id, not_configured(e)) from e + # 写作目录的篇幅按这个模型的上下文窗口定（evidence.catalog_budget） + window = await model_window(session, _model_spec(ctx), model_id) spec = ctx.run.spec catalog = report_catalog(state, spec, ctx.node) @@ -408,7 +463,14 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: if warning := _no_evidence(catalog, cells): ctx.emit(EventType.LOG, level="warn", code="report_no_evidence", message=warning) - messages = _messages(ctx, state, catalog, cells, claims) + writer = writer_catalog(catalog, budget=catalog_budget(window), cells_allowed=cells, + fields=report_fields(state, spec, ctx.node), + titles={n.id: n.title or n.id for n in spec.nodes}) + if writer.trimmed: + ctx.emit(EventType.LOG, level="warn", code="report_catalog_trimmed", message=_trimmed_message(writer), + budget=writer.budget, window=window, queries=writer.queries, omitted=len(writer.omitted), + nodes=[{k: r[k] for k in ("node_id", "title", "total", "omitted")} for r in writer.nodes]) + messages = _messages(ctx, state, writer, claims) started = time.perf_counter() spent: list[dict[str, Any]] = [] nudged = False diff --git a/backend/app/engine/review.py b/backend/app/engine/review.py index 594ff9c2..98e0d36d 100644 --- a/backend/app/engine/review.py +++ b/backend/app/engine/review.py @@ -89,6 +89,8 @@ WARN_CODES: dict[str, tuple[str, str]] = { "report_repair": ("report_repair", DEGRADED), # 报告撰写节点的上游没有可引用的口径卡指标：报告里的数都没有出处 "report_no_evidence": ("report_no_evidence", DEGRADED), + # 证据目录超出篇幅，有的查询只列了编号和摘要：写作模型没看全证据，报告的覆盖面有缺口 + "report_catalog_trimmed": ("report_catalog_trimmed", DEGRADED), # 口径卡的指标集、报告文档没能落进工件库：答案照常，只是点不开证据 "evidence_store_failed": ("evidence_store_failed", DEGRADED), # agent 的结构化字段：模型报的值和查询快照对不上（已按快照取值），或者出处核对不了 diff --git a/backend/tests/fixtures/evidence/store_run.py b/backend/tests/fixtures/evidence/store_run.py new file mode 100644 index 00000000..138eb477 --- /dev/null +++ b/backend/tests/fixtures/evidence/store_run.py @@ -0,0 +1,106 @@ +"""合成的「4 个专项 Agent 并行查库 → 报告」场景：查询多、分布不均，每个 Agent 都交了按出处核对的字段。 + +形状仿一次真实运行（4 个 Agent 并行、查询次数悬殊，合计上百次），次数和内容全部虚构：门店、会员、卡券、活动。 +不连库、不调模型：台账条目、查询快照、表结构快照、Agent 的 data / data_evidence 都在内存里拼好，快照经 loader 取。 + +台账按节点跑完的先后排（state["evidence"] 是各节点结束时追加的），所以全局编号 Q1… 前 26 个都是会员专项的。 +""" +from __future__ import annotations + +from typing import Any + +from app.engine.evidence import cell_eid, make_eid + +#: (节点 id, 画布上的名字, 查询次数, 查的表) +AGENTS: list[tuple[str, str, int, str]] = [ + ("agent_member", "会员专项", 26, "member_daily"), + ("agent_store", "门店专项", 17, "store_daily"), + ("agent_coupon", "卡券核销", 79, "coupon_redemption"), + ("agent_campaign", "活动内容", 42, "campaign_view"), +] +COLUMNS = ["store_name", "biz_time", "cnt", "amount"] +TYPES = {"store_name": "text", "biz_time": "datetime", "cnt": "number", "amount": "number"} +ROWS = 30 +#: 结构化抽取失败（两次都被截断）的那个 Agent：字段全是空值 +EMPTY_AGENT = "agent_coupon" +FIELDS = ["total_cnt", "top_store", "latest_time", "top_stores", "churn_rate"] + + +def artifact_of(node_id: str, i: int) -> str: + return f"{node_id}-q{i:03d}".ljust(64, "0") + + +def snapshot(node_id: str, i: int, rows: int = ROWS) -> dict[str, Any]: + """第 i 次查询的快照：每行一个门店，时间是 ISO 写法（快照里存的就是这样）。""" + seed = sum(map(ord, node_id)) + i * 7 + return { + "columns": list(COLUMNS), + "column_types": dict(TYPES), + "rows": [[f"门店{(seed + r) % 90 + 1:02d}", f"2026-06-{r % 28 + 1:02d}T08:{(seed + r) % 60:02d}:39", + (seed * 13 + r * 17) % 900 + 10, f"{(seed * 31 + r * 7) % 9000 + 100}.50"] for r in range(rows)], + "truncated": False, + } + + +def schema_snapshot(table: str) -> dict[str, Any]: + return {"source": "shop", "tables": {table: {"columns": [{"name": c, "type": TYPES[c]} for c in COLUMNS]}}} + + +def build() -> tuple[list[dict[str, Any]], dict[str, Any], dict[str, Any], dict[str, str]]: + """(台账, 快照 {工件 id: 内容}, state["nodes"], {节点 id: 名字})。""" + ledger: list[dict[str, Any]] = [] + store: dict[str, Any] = {} + nodes: dict[str, Any] = {} + titles: dict[str, str] = {} + for node_id, title, count, table in AGENTS: + titles[node_id] = title + schema_id = f"schema-{table}".ljust(64, "0") + store[schema_id] = schema_snapshot(table) + ledger.append({"kind": "schema", "node_id": node_id, "exec": 1, "artifact": schema_id, "source": "shop", + "tables": {table: list(COLUMNS)}}) + for i in range(1, count + 1): + aid = artifact_of(node_id, i) + store[aid] = snapshot(node_id, i) + ledger.append({"kind": "query", "node_id": node_id, "exec": 1, "artifact": aid, + "via": f"tool-{aid}"[:64], "call_id": f"call_{i}", "tool": "db_query__shop", "source": "shop", + "columns": list(COLUMNS), "rows": ROWS, "truncated": False, + "schema_artifact": schema_id, "tables": [table]}) + nodes[node_id] = agent_output(node_id, store) + return ledger, store, nodes, titles + + +def agent_output(node_id: str, store: dict[str, Any]) -> dict[str, Any]: + """Agent 交出的结构化字段：本节点的 Q3 那一格、Q5 的前 5 行；EMPTY_AGENT 抽取失败，字段全空。""" + if node_id == EMPTY_AGENT: + reason = "结构化抽取没有完成：模型的输出额度在推理阶段就用完了（最后一次的额度为 32768 token），结构化结论没能写出来" + return {"text": "查到了大量核销和兑换记录。", "data": {f: None for f in FIELDS}, + "data_evidence": {f: {"status": "unresolved", "reason": reason} for f in FIELDS}} + q3, q5 = artifact_of(node_id, 3), artifact_of(node_id, 5) + first = store[q3]["rows"][0] + + def cell(column: str, index: int) -> dict[str, Any]: + return {"ref": f"Q3.r0.{column}", "call": "Q3", "artifact": q3, "via": f"tool-{q3}"[:64], + "locator": {"row": 0, "column": column}, "eid": cell_eid(q3, 0, column), "status": "verified"} + + rows_locator = {"rows": [0, 4], "column": "store_name"} + return { + "text": "结论见字段。", + "data": {"total_cnt": first[2], "top_store": first[0], "latest_time": first[1], + "top_stores": [r[0] for r in store[q5]["rows"][:5]], "churn_rate": None}, + "data_evidence": { + "total_cnt": cell("cnt", 2), "top_store": cell("store_name", 0), "latest_time": cell("biz_time", 1), + "top_stores": {"ref": "Q5.r0-4.store_name", "call": "Q5", "artifact": q5, "via": f"tool-{q5}"[:64], + "locator": rows_locator, "eid": make_eid("rows", q5, rows_locator), "status": "verified"}, + "churn_rate": {"ref": None, "status": "missing"}, + }, + } + + +def global_alias(node_id: str, local: int) -> str: + """本节点第 local 次查询在报告目录里的全局编号（台账按节点先后排）。""" + offset = 0 + for nid, _, count, _ in AGENTS: + if nid == node_id: + return f"Q{offset + local}" + offset += count + raise KeyError(node_id) diff --git a/backend/tests/test_report_catalog_budget.py b/backend/tests/test_report_catalog_budget.py new file mode 100644 index 00000000..3ea3da91 --- /dev/null +++ b/backend/tests/test_report_catalog_budget.py @@ -0,0 +1,209 @@ +"""写作者的出处清单：预算按写作模型的上下文窗口定；超出预算时分优先级、按来源均衡，省略了什么照实说。 + +2026-10 一次运行：证据台账里上百次查询（4 个 Agent 查询次数悬殊），出处清单写死 12000 字，十万字以上的目录 +被截到一万两千字以内，只有前二十来次查询进了写作者的视野，后面只有一句「目录太长，后面的省略了」。查询按编号排，前面的 +Agent 占满了预算，有的 Agent 一次都没露面；写作者于是写出「本轮没有查到任何核销结果」，可那个 Agent 查到了上百条。 + +- 预算：写作模型上下文窗口的 40%（字数按 1 字 ≈ 1 token 保守算），拿不到窗口按 128K；下限 12000、上限 160000 +- 第一优先：各 Agent 经按出处核对交来的结构化字段、口径卡指标，一律列出 +- 查询按产出它的节点分组、轮流取：先给每次查询前几行，再补到 20 行；每个节点都有代表 +- 没展开的查询列出编号和一行摘要（表、行数）；约定只能写 [[see:Qn]] 作依据，数字照样逐字核对 +- 清单开头明说「目录有省略」，各节点各省略了多少条 + +夹具全部合成（tests/fixtures/evidence/store_run.py）。 +""" +from __future__ import annotations + +import re + +import pytest + +import app.db.models # noqa: F401 - 建表（conftest 的 create_all）前要先登记模型 +from app.engine.evidence import ( + CATALOG_MAX_CHARS, + CATALOG_MIN_CHARS, + DEFAULT_CATALOG_BUDGET, + build_catalog, + catalog_budget, + catalog_prompt, + writer_catalog, +) +from tests.fixtures.evidence import store_run + +LEDGER, STORE, NODES, TITLES = store_run.build() + + +def load(artifact): + return STORE.get(artifact) + + +@pytest.fixture +def catalog(): + return build_catalog(nodes=NODES, ledger=LEDGER) + + +def sources(): + return [{"node_id": nid, "title": TITLES[nid], "data": NODES[nid]["data"], + "evidence": NODES[nid]["data_evidence"]} for nid, *_ in store_run.AGENTS] + + +def written(catalog, budget=DEFAULT_CATALOG_BUDGET, **kw): + return writer_catalog(catalog, budget=budget, loader=load, fields=sources(), titles=TITLES, **kw) + + +def group_of(text: str, title: str) -> str: + """清单里「来自「<名字>」的查询」那一段（到下一个来源或下一节为止）。""" + start = text.index(f"来自「{title}」的查询") + rest = text[start + 1:] + ends = [m.start() for m in re.finditer(r"\n(?:来自「|\n)", rest)] + return text[start:start + 1 + (ends[0] if ends else len(rest))] + + +# -------------------------------------------------------------------------- +# 预算 +# -------------------------------------------------------------------------- + + +def test_budget_follows_the_model_window(): + assert DEFAULT_CATALOG_BUDGET == catalog_budget(None) == 51_200, "拿不到窗口按 128K 的 40%" + assert catalog_budget(128_000) == 51_200 + assert catalog_budget(200_000) == 80_000 + assert catalog_budget(32_000) == 12_800 + assert catalog_budget(8_000) == CATALOG_MIN_CHARS == 12_000, "很小的窗口维持原来的篇幅" + assert catalog_budget(1_000_000) == CATALOG_MAX_CHARS == 160_000, "百万窗口也有上限：每写一稿都要重发整份目录" + assert catalog_budget(0) == catalog_budget(-5) == DEFAULT_CATALOG_BUDGET + + +def test_the_default_is_far_above_the_old_cut(): + assert DEFAULT_CATALOG_BUDGET >= 4 * 12_000 + + +# -------------------------------------------------------------------------- +# 放得下：照旧 +# -------------------------------------------------------------------------- + + +def test_a_catalog_that_fits_is_not_trimmed(): + small = build_catalog(nodes={}, ledger=[e for e in LEDGER if e.get("node_id") == "agent_store"]) + got = writer_catalog(small, loader=load) + assert not got.trimmed and not got.omitted + assert "目录有省略" not in got.text + assert got.text == catalog_prompt(small, loader=load), "没有结构化字段时和 catalog_prompt 一字不差" + assert "Q17：" in got.text and " r19：" in got.text + + +# -------------------------------------------------------------------------- +# 放不下：分优先级、按来源均衡（那次运行同样形状的场景） +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize("budget", [DEFAULT_CATALOG_BUDGET, 12_000]) +def test_every_agent_stays_in_view_within_the_budget(catalog, budget): + got = written(catalog, budget) + text = got.text + assert got.trimmed and len(text) <= budget, len(text) + for nid, title, count, table in store_run.AGENTS: + group = group_of(text, title) + assert f"共 {count} 次" in group + assert re.search(r"\n {2}r0：", group), f"「{title}」一次查询都没展开" + # 每次查询都在清单里：展开的有行，没展开的有编号和摘要 + total = sum(c for *_, c, _ in store_run.AGENTS) + for n in range(1, total + 1): + assert re.search(rf"(?:^|\n)\s*- Q{n}：", text), f"Q{n} 不见了" + + +def test_omitted_queries_are_listed_with_a_summary(catalog): + got = written(catalog) + assert got.omitted, "这个场景放不下全部展开" + for alias in got.omitted: + node = catalog[alias]["node_id"] + table = next(t for nid, _, _, t in store_run.AGENTS if nid == node) + assert re.search(rf"\n {{2}}- {alias}：{table}，30 行", got.text), alias + # 没展开的查询没有行 + first = got.omitted[0] + block = got.text[got.text.index(f"- {first}："):].split("\n", 2) + assert not block[1].lstrip().startswith("r0：") + + +def test_the_head_says_what_was_left_out_per_source(catalog): + got = written(catalog) + head = got.text[:800] + assert "目录有省略" in head + for row in got.nodes: + assert f"{row['title']} {row['total']} 次" in head and f"省略 {row['omitted']} 次" in head, row + assert sum(r["omitted"] for r in got.nodes) == len(got.omitted) + assert [r["node_id"] for r in got.nodes] == [nid for nid, *_ in store_run.AGENTS] + + +def test_how_to_use_an_omitted_query_is_spelled_out(catalog): + text = written(catalog).text + assert "[[see:Qn]]" in text and "本报告未覆盖" in text + assert "不要凭摘要写" in text and "数字照样逐字核对" in text + for phrase in ("没有查到", "不存在", "缺失", "没有数据"): + assert phrase in text + + +def test_structured_fields_come_first_and_point_at_global_cells(catalog): + got = written(catalog, 12_000) + text = got.text + fields_at = text.index("各 Agent 交来的结构化结论") + assert fields_at < text.index("来自「"), "结构化结论排在查询前面" + member_q3 = store_run.global_alias("agent_member", 3) + store_q3 = store_run.global_alias("agent_store", 3) + campaign_q5 = store_run.global_alias("agent_campaign", 5) + assert f"[[v:{member_q3}.r0.cnt]]" in text and f"[[v:{store_q3}.r0.store_name]]" in text + assert f"[[table:{campaign_q5} cols=store_name rows=0-4]]" in text + assert "churn_rate：抽取时没有找到这个值" in text + # 抽取失败的那个 Agent：结构化结论为空，不能据此断言没有数据 + block = text[text.index("「卡券核销」：", fields_at):] + assert block.startswith("「卡券核销」：结构化结论为空") and "本报告未覆盖" in block.split("\n", 1)[0] + assert got.empty_sources == ["卡券核销"] + + +def test_queries_the_fields_cite_are_expanded_first(catalog): + """结构化字段引用到的那几次查询先展开：写作者照抄字段的引用标记时，看得到那几行。""" + got = written(catalog, 12_000) + for nid in ("agent_member", "agent_store", "agent_campaign"): + for local in (3, 5): + alias = store_run.global_alias(nid, local) + assert alias not in got.omitted, alias + assert re.search(rf"\n- {alias}：.*\n {{2}}r0：", got.text), alias + + +def test_a_field_value_reads_like_the_cell_it_cites(catalog): + """字段后面写的值和引用之后报告里显示的一样（[[v:]] 的渲染规则）。""" + text = written(catalog).text + q3 = STORE[store_run.artifact_of("agent_member", 3)] + assert f"total_cnt = {q3['rows'][0][2]:,}：[[v:Q3.r0.cnt]]" in text + + +# -------------------------------------------------------------------------- +# 报告节点：窗口从哪来 +# -------------------------------------------------------------------------- + + +async def test_the_window_comes_from_the_provider_entry_then_the_catalog(): + """模型接入里给这个模型登记的 context 优先，其次模型目录；都没有返回 None（预算按保守值）。""" + from sqlalchemy import delete + + from app.db.base import SessionLocal + from app.db.models import Provider + from app.engine.nodes.report import model_window + from app.providers.factory import ModelSpec + + async with SessionLocal() as session: + row = Provider(name="fa-window-test", kind="mock", enabled=True, default_model="mock-wide", + models=[{"id": "mock-wide", "context": 64_000}, {"id": "mock-text", "context": "96000"}, + {"id": "mock-bare"}]) + session.add(row) + await session.commit() + try: + assert await model_window(session, ModelSpec(model="mock-wide"), "mock-wide") == 64_000 + assert await model_window(session, ModelSpec(model="mock-text"), "mock-text") == 96_000 + # 接入里没登记窗口：按模型目录（deepseek-chat 登记了 128K）；目录里也没有就是 None + assert await model_window(session, ModelSpec(model="deepseek-chat"), "deepseek-chat") == 128_000 + assert await model_window(session, ModelSpec(model="mock-bare"), "mock-bare") is None + assert catalog_budget(await model_window(session, ModelSpec(model="mock-wide"), "mock-wide")) == 25_600 + finally: + await session.execute(delete(Provider).where(Provider.id == row.id)) + await session.commit() diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index f82b4844..f9bab20c 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -756,6 +756,22 @@ function explainLog(code: string | undefined, message: string, d: Record typeof r?.omitted === 'number' && r.omitted > 0 && str(r?.title)) + .map((r) => `${r.title} 省略 ${formatNumber(r.omitted)}/${formatNumber(r.total)} 次`) + return { + title: omitted != null + ? `证据目录超出篇幅，${total != null ? `${formatNumber(total)} 次查询中 ` : ''}${formatNumber(omitted)} 次只列了编号和摘要` + : '证据目录超出篇幅，部分查询只列了编号和摘要', + sub: per.length ? per.join('、') : undefined, + next: '写作模型按来源均衡地看到了一部分查询的明细，被省略的查询仍在证据台账里，报告不得据此断言没有数据。' + + '需要完整明细时，请换用上下文窗口更大的写作模型，或让 Agent 在 SQL 中先汇总再交出结果', + } + } case 'agent_field_mismatch': { // 「有 1 个字段和查询快照对不上，已按快照取值：order_cnt 模型报 1240，快照是 1234」（llm.py _field_warnings） const n = message.match(/有\s*(\d+)\s*个字段/)?.[1] diff --git a/scripts/check-copy.mjs b/scripts/check-copy.mjs index 2376d6f1..e023b249 100644 --- a/scripts/check-copy.mjs +++ b/scripts/check-copy.mjs @@ -230,12 +230,13 @@ const MODEL_ONLY = { 'NODE_REFERENCE', 'ROLE', 'MARKUP_NUDGE', 'JUDGE_RULES', 'JUDGE_ROLE', 'JUDGE_RULE', 'REVIEW_SYSTEM', 'MARKER_RULES', 'GATE_SYSTEM', 'TOOL_MARKUP_NUDGE', '_UNKNOWN_ENTITY_FOR_MODEL', 'NODE_FIELD_FIX', '_FIX_TAIL', '_STREAM_PROTOCOL', '_RULES_HEAD', '_NO_LOWERING', '_ASK_HUMAN', '_UPGRADE_TASK', 'CELL_RULES', 'ENTITY_RULES', - 'EXTRACT_SYSTEM', 'GRAPH_SCHEMA', 'REVIEW_SCHEMA', '_CN_N', + 'EXTRACT_SYSTEM', 'GRAPH_SCHEMA', 'REVIEW_SCHEMA', '_CN_N', 'OMITTED_RULES', 'ABSENCE_RULE', ]), // 拼提示词的函数：生成 / 修正 / 升级请求、抽取消息、交给抽取模型的输出结构； // 助手的数据源上下文（api/copilot_context.py 的各段渲染、挑表请求，copilot.py 里挑表用的需求） funcs: new Set(['_user_message', '_repair_request', '_publish_fix_request', '_upgrade_assist_request', - '_extract_messages', '_from_rows', 'catalog_prompt', + '_extract_messages', '_from_rows', 'catalog_prompt', 'writer_catalog', '_trimmed_catalog', '_metric_prompt', + '_fields_prompt', '_field_prompt', '_query_brief', '_pick_messages', '_pick_need', '_publish_fix_need', '_upgrade_need', '_detail_block', '_selected_block', '_join_lines', '_route_line', '_source_head']), // 第二个参数是交回模型改写的指令（第一个是给人看的那句）：engine/evidence.py 的 _Reason(text, fix)、fail(text, fix) diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index 5045adb1..f11e795a 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -1300,6 +1300,26 @@ await section('报告核对 report.checked、口径卡的台账（可点击证 `${flagged?.status} ${flagged?.level}`) const bare = flatten(mod.decodeRun([E(1, 'report.checked', null, {})])).find((s) => s.code === 'report_checked') check('载荷缺字段也不崩，只说核对了报告', bare?.title === '核对报告' && !bare.artifact, bare?.title) + + // 写作目录超出篇幅（2026-10 一次运行：上百次查询只有前二十来次进了写作者的视野，界面上一个字都没说）： + // 现在按来源均衡地展开，没展开的列编号和摘要，并发一条警告。界面读结构化字段（queries、omitted、nodes） + check('后端发目录有省略的警告，带着结构化字段', reportPy.includes('code="report_catalog_trimmed"') + && reportPy.includes('queries=writer.queries, omitted=len(writer.omitted)') && reportPy.includes('"node_id", "title", "total", "omitted"'), + '改了 report.py 的字段就同步改 decode.ts 的 explainLog 和这里的夹具') + const TRIMMED = '证据目录超出篇幅（约 51200 字，按写作模型的上下文窗口确定）：164 次查询中 90 次列出了数据，74 次只列编号和摘要' + + '（卡券核销省略 52 次、活动内容省略 22 次）。写作模型看不到这些查询的明细，报告不得据此断言没有数据' + const trimmed = flatten(mod.decodeRun([E(1, 'node.started', 'w', { node_type: 'report', label: '写报告' }), + E(2, 'log', 'w', { level: 'warn', code: 'report_catalog_trimmed', message: TRIMMED, budget: 51200, window: null, queries: 164, + omitted: 74, nodes: [{ node_id: 'a', title: '会员专项', total: 26, omitted: 0 }, { node_id: 'b', title: '卡券核销', total: 79, omitted: 52 }, + { node_id: 'c', title: '活动内容', total: 42, omitted: 22 }] })])).find((s) => s.code === 'report_catalog_trimmed') + check('report_catalog_trimmed：标题说几次查询只列了编号和摘要', trimmed?.title === '证据目录超出篇幅，164 次查询中 74 次只列了编号和摘要', trimmed?.title) + check('……副标题按来源列省略了几次（没省略的来源不列）', trimmed?.sub === '卡券核销 省略 52/79 次、活动内容 省略 22/42 次', trimmed?.sub) + check('……下一步说被省略的仍在台账里、不得据此断言没有数据，以及怎么看全', !!trimmed?.next?.includes('不得据此断言没有数据') + && !!trimmed?.next?.includes('上下文窗口更大') && trimmed?.level === 'warn' && trimmed?.detail === TRIMMED, trimmed?.next) + const trimmedBare = flatten(mod.decodeRun([E(1, 'log', 'w', { level: 'warn', code: 'report_catalog_trimmed', message: '目录有省略' })])) + .find((s) => s.code === 'report_catalog_trimmed') + check('……缺字段时标题照样说人话，不出 undefined', trimmedBare?.title === '证据目录超出篇幅，部分查询只列了编号和摘要' + && trimmedBare?.sub === undefined, `${trimmedBare?.title} / ${trimmedBare?.sub}`) }) await section('工具门控 tool.gated（工具信任三档）', async () => {