合入 FA:结构化抽取截断重试、写作清单按模型窗口定预算并按来源均衡、缺失断言交回重写、日期时间按可读格式显示

This commit is contained in:
ThearchyHelios
2026-10-04 11:43:21 +08:00
16 changed files with 1857 additions and 76 deletions
+422 -37
View File
@@ -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
@@ -69,6 +72,15 @@ DOC_SCHEMA = "agentlab.report/1"
#: 和记下的状态对不上,会被当成文档被改过(state_mismatch)。
ENTITY_SYNTAX = 2
ENTITY_SYNTAXES = frozenset({1, 2})
#: 单元格显示规则的版本(文档顶层 cell_display):
#: - 1:照快照里的原样显示,ISO 日期时间带 T(2026-06-30T08:21:39)。没有这个字段的老文档按它复核
#: - 2:ISO 日期时间显示成「2026-06-30 08:21:39」(小数秒、时区照留)。只改显示:快照里的值、引用记下的 value、
#: eid、证据下钻都不变
#: 新组装的文档里真有日期时间被改写时才记 cell_display: 2(参与内容哈希),没有的文档和以前一字不差。复核必须按
#: 文档生成时的那一版:老文档存的是带 T 的字,拿新规则重新渲染对不上,会被当成文档被改过(render_mismatch)
CELL_DISPLAY = 2
CELL_DISPLAYS = frozenset({1, 2})
_ISO_DATETIME = re.compile(r"^(\d{4}-\d{2}-\d{2})T(\d{2}:\d{2}(?::\d{2}(?:\.\d+)?)?(?:Z|[+-]\d{2}:?\d{2})?)$")
# --------------------------------------------------------------------------
@@ -1270,7 +1282,8 @@ class _Snapshots:
compose / verify / 流式渲染里:出口复核是另一次调用,照样重新取、重新验。
"""
def __init__(self, loader: Loader | None = None, *, entities: bool = True, syntax: int | None = None) -> None:
def __init__(self, loader: Loader | None = None, *, entities: bool = True, syntax: int | None = None,
display: int | None = None) -> None:
self.loader = loader or load_artifact
self.memo: dict[str, tuple[Any, str | None]] = {}
self._index: tuple[dict[str, Any], _EntityIndex] | None = None
@@ -1278,6 +1291,17 @@ class _Snapshots:
self.entities_on = entities
#: 这一次解析按哪一版实体语法:组装、流式渲染用当前版,复核按文档记下的版本(verify_doc)
self.syntax = ENTITY_SYNTAX if syntax is None else syntax
#: 单元格按哪一版显示规则渲染(CELL_DISPLAY):组装、流式渲染用当前版,复核按文档记下的版本
self.display = CELL_DISPLAY if display is None else display
#: 这一次解析里有没有哪一格因为显示规则改了写法:组装时据此决定文档记不记 cell_display
self.reformatted = False
def shown(self, raw: Any) -> Any:
"""按显示规则要显示的原值:2 版起 ISO 日期时间的 T 换成空格。没改就是原值本身。"""
if self.display >= 2 and isinstance(raw, str) and (m := _ISO_DATETIME.match(raw.strip())):
self.reformatted = True
return f"{m.group(1)} {m.group(2)}"
return raw
def entities(self, catalog: dict[str, Any]) -> _EntityIndex:
if self._index is None or self._index[0] is not catalog:
@@ -1286,11 +1310,12 @@ class _Snapshots:
self._index = (catalog, _EntityIndex(shown, syntax=self.syntax))
return self._index[1]
def with_syntax(self, syntax: int) -> "_Snapshots":
"""换一版实体语法的同一次解析。取过的快照共用:快照内容和语法版本无关,不必再取、再验一遍。"""
if syntax == self.syntax:
def with_syntax(self, syntax: int, display: int | None = None) -> "_Snapshots":
"""换一版实体语法(和单元格显示规则)的同一次解析。取过的快照共用:快照内容和版本无关,不必再取、再验一遍。"""
display = self.display if display is None else display
if syntax == self.syntax and display == self.display:
return self
twin = _Snapshots(self.loader, entities=self.entities_on, syntax=syntax)
twin = _Snapshots(self.loader, entities=self.entities_on, syntax=syntax, display=display)
twin.memo = self.memo
return twin
@@ -1336,11 +1361,11 @@ class _Snapshots:
def _snapshots(loader: Loader | _Snapshots | None, *, entities: bool = True,
syntax: int | None = None) -> _Snapshots:
"""loader 已经是 _Snapshots 时沿用它(entities 不看);给了 syntax 就换成那一版。"""
syntax: int | None = None, display: int | None = None) -> _Snapshots:
"""loader 已经是 _Snapshots 时沿用它(entities 不看);给了 syntax、display 就换成那一版。"""
if isinstance(loader, _Snapshots):
return loader if syntax is None else loader.with_syntax(syntax)
return _Snapshots(loader, entities=entities, syntax=syntax)
return loader if syntax is None and display is None else loader.with_syntax(syntax or loader.syntax, display)
return _Snapshots(loader, entities=entities, syntax=syntax, display=display)
_CELL_REF = re.compile(r"^(?P<alias>[QK]\d+)\.r(?P<row>\d+)\.(?P<column>.+)$")
@@ -1390,7 +1415,7 @@ def _resolve_cell(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snaps
try:
raw, column = locate_cell(snapshot, parsed["row"], parsed["column"])
kind = column_kind(snapshot, column)
rendered = render_cell(raw, conv, kind=kind)
rendered = render_cell(snaps.shown(raw), conv, kind=kind)
except CellError as e:
return _unresolved(marker, alias, "cell", f"{alias} {e}")
except RenderError as e:
@@ -2471,11 +2496,13 @@ def compose_doc(
cells_allowed: bool = True,
loader: Loader | None = None,
entities: bool = True,
absence: bool = False,
) -> dict[str, Any]:
"""模型写的原文(带标记)→ 报告文档:渲染后的 markdown、块、句、片段、统计、违规。
entities=False(报告节点 entities: off):表名、字段名整层不管——不链接、不核对反引号,
[[t:]] / [[c:]] 解析不了,原因是 ENTITIES_OFF_REASON。
absence=True(写作目录有省略):另查缺失断言(absence_claims),和句中的 [[see:]] 一样只在组装时查。
stats / violations 由 verify_doc 算出,和出口契约复核用的是同一个函数。
整表标记先展开成每格一个 [[v:]] 的 Markdown 表,之后和模型自己写的表格走同一条路;
@@ -2625,14 +2652,18 @@ def compose_doc(
# entity_syntax:这份文档按哪一版实体语法组装,复核照它走(verify_doc)。在文档顶层,参与内容哈希
# provenance:期 4 的文档标记(P4-SPEC 1.4)。带它的文档才有查询步骤的提示字段、推断来源、裁判摘录的新行;
# 没有它的(期 4 之前组装的)一律按原样展示。写在顶层、参与内容哈希,事后补不上:补了哈希就变,封存链就断
# cell_display:有单元格因为显示规则改了写法(ISO 日期时间)才记,复核照它渲染;没有的文档和以前一字不差
doc: dict[str, Any] = {
"schema": DOC_SCHEMA, "entity_syntax": snaps.syntax, "provenance": DOC_PROVENANCE,
**({"cell_display": snaps.display} if snaps.reformatted and snaps.display != 1 else {}),
"run_id": run_id, "node_id": node_id,
"markdown": markdown, "source": source, "catalog": kept, "blocks": out_blocks,
}
checked = verify_doc(doc, catalog, allow_numbers=allow_numbers, cells_allowed=cells_allowed, loader=snaps)
# 句中的 [[see:]] 只在组装时查(交回写作者重写):出口复核看的是数字和出处,句子通不通不影响出具
doc["stats"], doc["violations"] = checked["stats"], checked["violations"] + inline_see(source)
# 句中的 [[see:]]、目录有省略时的缺失断言只在组装时查(交回写作者重写):出口复核看的是数字和出处,
# 句子通不通、结论有没有越过证据不影响出具
doc["stats"], doc["violations"] = checked["stats"], \
checked["violations"] + inline_see(source) + (absence_claims(doc) if absence else [])
return doc
@@ -2778,7 +2809,8 @@ def verify_doc(
有知识库检索时才有(它们才可能不是 0):升级前的运行统计的键和以前一模一样。
"""
entity_syntax = doc_entity_syntax(doc)
snaps = _snapshots(loader, entities=entities, syntax=entity_syntax or ENTITY_SYNTAX)
display = doc_cell_display(doc)
snaps = _snapshots(loader, entities=entities, syntax=entity_syntax or ENTITY_SYNTAX, display=display or CELL_DISPLAY)
index = snaps.entities(catalog)
violations: list[dict[str, Any]] = []
uncited: list[dict[str, Any]] = []
@@ -2796,6 +2828,8 @@ def verify_doc(
if entity_syntax is None:
violations.append(_violation("bad_schema",
f"报告文档记录的表名、字段名规则版本「{doc.get('entity_syntax')}」无法识别"))
if display is None:
violations.append(_violation("bad_schema", f"报告文档记录的单元格显示规则版本「{doc.get('cell_display')}」无法识别"))
markdown = str(doc.get("markdown") or "")
masks: list[tuple[int, int]] = [] # 引用渲染出来的字
@@ -2942,6 +2976,13 @@ def verify_doc(
return {"ok": not violations, "violations": violations, "stats": stats, "uncited": uncited}
def doc_cell_display(doc: Any) -> int | None:
"""文档按哪一版单元格显示规则组装:没有 cell_display 字段的是 1 版(老文档、没有日期时间被改写的文档),
记了认不出的值返回 None。"""
raw = doc.get("cell_display", 1) if isinstance(doc, dict) else 1
return raw if type(raw) is int and raw in CELL_DISPLAYS else None
def doc_entity_syntax(doc: Any) -> int | None:
"""文档按哪一版实体语法组装:没有 entity_syntax 字段的是老文档(1 版),记了认不出的值返回 None。"""
raw = doc.get("entity_syntax", 1) if isinstance(doc, dict) else 1
@@ -3032,6 +3073,58 @@ def inline_see(source: str) -> list[dict[str, Any]]:
return [_violation("inline_see", INLINE_SEE_MESSAGE, text=m.group(0)) for m in _INLINE_SEE.finditer(source or "")]
#: 写作目录有省略、或者有来源的结构化结论为空时,拼在写作规则后面(给模型看)
ABSENCE_RULE = (
"这次你看不到全部证据(证据目录有省略,或者有来源的结构化结论为空):不得断言「没有查到」「未查到」「不存在」"
"「缺失」「没有数据」「无数据」这类结论。某个方向的证据没有展开或为空时,只能写「本报告未覆盖……」,并说明依据不足"
"(比如「目录有省略,相关查询未展开」),句末可以挂 [[see:Qn]] 指向相关查询。"
)
#: 缺失断言:断言「没有数据」的几种说法。只认拿得准的,不泛化:「不存在明显差异」「缺少口径说明」这类说的是看到的
#: 数据里的情况,不算;「不存在」「为空」「缺失」只认跟在数据、记录、结果后面(或者前面)的
_ABSENCE = re.compile(
r"(?:没有|没|未|未能|没能)(?:查询|查)到|查不到|查无"
r"|(?:没有|暂无|并无)(?:任何)?(?:相关的?)?(?:数据|记录|结果)"
r"|无(?:任何)?(?:相关的?)?(?:数据|记录)(?![库源集表权字项列])"
r"|不存在(?:任何)?(?:相关的?)?(?:数据|记录|结果)|(?:数据|记录|结果)(?:均|都|全部)?(?:不存在|为空|缺失|空缺)"
)
#: 前面是这些词的不算:「不代表没有数据」「并非查不到」说的恰恰是不能下这个结论
_ABSENCE_NEGATED = re.compile(r"(?:不代表|并不代表|不等于|不意味着|并非|不是|不能说明|不能断言|不得断言|不应断言)[^。!?\n]{0,6}$")
ABSENCE_MESSAGE = _Reason(
"证据目录有省略时写了缺失断言:写作模型没有看到全部证据,「没有查到」「不存在」「缺失」这类结论依据不足",
"证据目录有省略(没展开的查询只列了编号和摘要),这一句却断言了没有数据。看不到的证据不能当成不存在:"
"改成「本报告未覆盖……」,并说明依据不足(目录有省略、相关查询未展开),不要写「没有查到」「不存在」「缺失」「没有数据」",
)
def absence_claims(doc: dict[str, Any]) -> list[dict[str, Any]]:
"""正文里的缺失断言(_ABSENCE),每处一条违规 absence_claim。只在写作目录有省略时由报告节点要求查(compose_doc 的
absence):目录全给了的时候,「本周没有退款记录」可能就是看了数据说的。
在渲染后的正文里找(标记已经换成了字),代码块不查。带着 span、所在片段和句子:证据面板能定位、能显示原句。
"""
markdown = str(doc.get("markdown") or "")
found: list[dict[str, Any]] = []
for block in doc.get("blocks") or []:
if block.get("type") == "code":
continue
for unit in block.get("units") or []:
span = unit.get("span")
if not (isinstance(span, list) and len(span) == 2):
continue
sentence = markdown[span[0]:span[1]]
for m in _ABSENCE.finditer(sentence):
if _ABSENCE_NEGATED.search(sentence[:m.start()]):
continue
at = span[0] + m.start()
seg = next((g for g in unit.get("segments") or [] if g.get("kind") != "structural"
and isinstance(g.get("span"), list) and g["span"][0] <= at < g["span"][1]), None)
found.append(_violation("absence_claim", ABSENCE_MESSAGE, text=m.group(0),
span=[at, span[0] + m.end()], context=sentence.strip()[:120],
segment=seg.get("id") if seg else None, unit=unit.get("id")))
return found
#: 引用查询结果的写法。只跟着目录里的查询出现(catalog_prompt):写作规则本身保持一期那份,升级前
#: 发起的运行跑到报告节点时提示一字不差;只有口径卡的目录也不该教写作者去引用不存在的 Q1
CELL_RULES = (
@@ -3094,10 +3187,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,13 +3210,13 @@ 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:
raw, _ = locate_cell(snapshot, r, c)
kind = column_kind(snapshot, c)
shown, numeric = render_cell(raw, kind=kind), is_number(cell_value(raw, kind))
shown, numeric = render_cell(snaps.shown(raw), kind=kind), is_number(cell_value(raw, kind))
except (CellError, RenderError):
shown, numeric = MISSING, False
cells.append(f"{c}={shown}")
@@ -3130,22 +3225,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 +3323,257 @@ 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(snaps.shown(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], snaps: _Snapshots) -> str:
"""没展开的查询在目录里的一行摘要:编号、查的表、行数。
表先看目录里的表条目(带表结构快照的查询才有),没有就从查询快照里存的 SQL 认(sql_tables),
再没有写工具名。
"""
if not tables:
snapshot, why = snaps.get(entry["alias"], entry.get("artifact"))
if not why and isinstance(snapshot.get("sql"), str):
tables = sql_tables(snapshot["sql"])
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"], []), snaps)] 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:
+127 -22
View File
@@ -541,25 +541,66 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
async def extract_step(payload: list[BaseMessage], schema: dict[str, Any]) -> dict[str, Any]:
# cite_fields 的结构化抽取:进 checkpoint,节点重放时不再调用、不再计费。失败也在这里收住、
# 作为结果返回——agent 的结论还在,字段记空值、发警告,而不是把整个节点赔进去。
# thinking 关掉:结构化输出靠强制工具调用,和思考模式不兼容,抽取也用不着思考
spec = model_spec.model_copy(update={
"thinking": "off", "max_tokens": max(int(ctx.cfg("max_tokens") or 0), 4096)})
started = time.perf_counter()
ctx.emit(EventType.LLM_START, model=model_id, structured=True, purpose="cite_fields",
message_count=len(payload))
try:
async with SessionLocal() as session:
extractor, _ = await get_chat_model(session, spec)
got = await extractor.with_structured_output(schema, include_raw=True).ainvoke(payload)
except Exception as e: # noqa: BLE001
# thinking 关掉:结构化输出靠强制工具调用,和思考模式不兼容,抽取也用不着思考。但经 OpenAI 兼容
# 接口接入的推理模型关不掉思考,推理照样算在额度里:额度见 EXTRACT_MAX_TOKENS,被截断时加大重试一次
budget = max(int(ctx.cfg("max_tokens") or 0), EXTRACT_MAX_TOKENS)
spent = {"input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0, "calls": 0}
rejected: int | None = None # 服务方嫌太大、拒收的额度:之后重试不再超过它
retried = False
def count(usage: dict[str, Any]) -> None:
for key in ("input_tokens", "output_tokens", "calls"):
spent[key] += int(usage.get(key) or 0)
spent["cost_usd"] = round(spent["cost_usd"] + float(usage.get("cost_usd") or 0), 6)
while True:
spec = model_spec.model_copy(update={"thinking": "off", "max_tokens": budget})
started = time.perf_counter()
ctx.emit(EventType.LLM_START, model=model_id, structured=True, purpose="cite_fields",
message_count=len(payload), max_tokens=budget)
failure: Exception | None = None
raw: BaseMessage | None = None
parsed: Any = None
try:
async with SessionLocal() as session:
extractor, _ = await get_chat_model(session, spec)
got = await extractor.with_structured_output(schema, include_raw=True).ainvoke(payload)
raw, parsed = _split_structured(got)
usage = _usage_of(raw if raw is not None else AIMessage(content=""), model_id)
except Exception as e: # noqa: BLE001
failure, usage = e, _error_usage(e, model_id)
count(usage)
cut = truncated_error(failure) if failure is not None else (parsed is None and truncated_reply(raw))
elapsed = int((time.perf_counter() - started) * 1000)
if failure is not None and not cut and _budget_rejected(failure) \
and (lower := next((b for b in _EXTRACT_FALLBACKS if b < budget), None)) is not None:
# 服务方的输出上限比默认额度小(DeepSeek chat 系只收 8K):退一档再来,别因为额度提高反而抽不成
ctx.emit(EventType.LLM_END, agent=ctx.node.title, model=model_id, purpose="cite_fields",
duration_ms=elapsed, **usage,
error=f"模型接入不接受 {budget} token 的输出额度,已改用 {lower} token 重试")
rejected, budget = budget, lower
continue
bigger = min(budget * 2, EXTRACT_RETRY_CAP)
if cut and not retried and bigger > budget and (rejected is None or bigger < rejected):
ctx.emit(EventType.LLM_END, agent=ctx.node.title, model=model_id, purpose="cite_fields",
duration_ms=elapsed, **usage,
error=f"输出被截断:{budget} token 的输出额度用完了,结构化结论没有写完,"
f"已把额度加大到 {bigger} token 重试一次")
retried, budget = True, bigger
continue
if cut:
ctx.emit(EventType.LLM_END, agent=ctx.node.title, model=model_id, purpose="cite_fields",
duration_ms=elapsed, **usage,
error=f"输出被截断:{budget} token 的输出额度用完了,结构化结论没有写完")
return {"error": TRUNCATED_EXTRACT.format(budget=budget), "truncated": True,
"max_tokens": budget, **spent}
if failure is not None:
ctx.emit(EventType.LLM_END, agent=ctx.node.title, model=model_id, purpose="cite_fields",
duration_ms=elapsed, **usage, error=describe_exception(failure))
return {"error": f"结构化抽取失败:{explain_model_error(failure, model_id)}", **spent}
ctx.emit(EventType.LLM_END, agent=ctx.node.title, model=model_id, purpose="cite_fields",
duration_ms=int((time.perf_counter() - started) * 1000), error=describe_exception(e))
return {"error": f"结构化抽取失败:{explain_model_error(e, model_id)}"}
raw, parsed = _split_structured(got)
usage = _usage_of(raw if raw is not None else AIMessage(content=""), model_id)
ctx.emit(EventType.LLM_END, agent=ctx.node.title, model=model_id, purpose="cite_fields",
duration_ms=int((time.perf_counter() - started) * 1000), **usage)
return {"parsed": parsed, **usage}
duration_ms=elapsed, **usage)
return {"parsed": parsed, **spent}
final_text = ""
#: 模型把工具调用写成了文字:纠正过一次了没有
@@ -778,8 +819,10 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
total_usage[key] += int(extracted.get(key) or 0)
total_usage["cost_usd"] = round(total_usage["cost_usd"] + float(extracted.get("cost_usd") or 0), 6)
cited = verify_cited_fields(extracted, fields, queries)
for code, message, names in _field_warnings(*cited, extracted.get("error")):
await once(ctx, EventType.LOG, level="warn", code=code, message=message, fields=names)
truncated = extracted.get("truncated") is True
for code, message, names in _field_warnings(*cited, extracted.get("error"), truncated=truncated):
extra = {"max_tokens": extracted.get("max_tokens")} if code == "agent_field_truncated" else {}
await once(ctx, EventType.LOG, level="warn", code=code, message=message, fields=names, **extra)
total_usage["total_tokens"] = total_usage["input_tokens"] + total_usage["output_tokens"]
result = {
@@ -941,6 +984,60 @@ def cited_schema(schema: dict[str, Any]) -> dict[str, Any]:
return {"title": "cited_fields", **rewrite(schema)}
#: cite_fields 结构化抽取的输出额度。thinking: off 只对 Anthropic 有用:经 OpenAI 兼容接口接入的推理模型
#: (DeepSeek 系等)关不掉思考,推理也算在这个额度里。以前是 4096:2026-10 一次运行 4 个 Agent 的抽取全部在
#: 推理阶段就用完了额度(completion_tokens=4096),一个字段都没写出来。节点配了更大的「最大输出 token」就用节点的
EXTRACT_MAX_TOKENS = 16384
#: 被截断时额度翻倍重试一次,最多到这里
EXTRACT_RETRY_CAP = 32768
#: 服务方嫌额度太大、拒收时依次退到这几档(DeepSeek chat 系只收 8K)。拒收发生在请求刚发出时,不产生用量
_EXTRACT_FALLBACKS = (8192, 4096)
#: 服务方嫌额度太大时报错里会出现的字样(和 api/copilot.py 改图时退回小额度是同一个判断)
_BUDGET_WORDS = ("max_tokens", "max_completion_tokens", "max_output_tokens", "maximum context length")
#: 截断类报错的原话:openai 的 LengthFinishReasonError(结构化输出解析时 finish_reason=length)
_LENGTH_WORDS = ("length limit was reached", "finish_reason=length", "finish_reason='length'")
#: 两次都被截断时的原因(给人看)。重试、去设置页测试模型都不解决额度问题,不能照通用的模型报错那样说
TRUNCATED_EXTRACT = ("结构化抽取没有完成:模型的输出额度在推理阶段就用完了(最后一次的额度为 {budget} token),"
"结构化结论没能写出来")
#: agent_field_truncated 的下一步(接在「N 个字段都记为空值」后面)
TRUNCATED_NEXT = ("可以为该 Agent 换用非推理模型(抽取沿用 Agent 的模型),或者调大该 Agent 的"
"「最大输出 token」后重新运行")
def truncated_error(error: BaseException | None) -> bool:
"""这个异常是不是在说「输出被截断」:openai 的 LengthFinishReasonError,或者 langchain、网关转述的同一件事。
服务方嫌额度太大的 400(报错里也有 max_tokens)不算:那是请求被拒,模型一个字都没写。
"""
if error is None:
return False
if type(error).__name__ == "LengthFinishReasonError":
return True
text = str(error).lower()
return any(word in text for word in _LENGTH_WORDS)
def truncated_reply(raw: Any) -> bool:
"""这条回复是不是因为额度用完停下的:OpenAI 系是 finish_reason=length,Anthropic 是 stop_reason=max_tokens。"""
meta = getattr(raw, "response_metadata", None) or {}
reason = meta.get("finish_reason") or meta.get("stop_reason")
return reason in ("length", "max_tokens")
def _budget_rejected(error: BaseException) -> bool:
status = getattr(error, "status_code", None) or getattr(getattr(error, "response", None), "status_code", None)
return status in (400, 422) and any(word in str(error).lower() for word in _BUDGET_WORDS)
def _error_usage(error: BaseException, model_id: str) -> dict[str, Any]:
"""失败的那次调用花掉的用量。截断时 openai 的异常带着 completion.usage:额度是真花出去了,要记账。"""
usage = getattr(getattr(error, "completion", None), "usage", None)
inp = int(getattr(usage, "prompt_tokens", 0) or 0)
out = int(getattr(usage, "completion_tokens", 0) or 0)
return {"input_tokens": inp, "output_tokens": out, "total_tokens": inp + out,
"cost_usd": round(catalog.estimate_cost(model_id, inp, out), 6), "calls": int(bool(inp or out))}
EXTRACT_SYSTEM = (
"你是数据抽取员:把智能体查到的数据按给定结构填好,每个字段都写出处。\n"
"- value 是字段的值;from 写它在哪次查询的哪一格:call 是【证据 Qn】里的编号,row 是行号(从 0 数),"
@@ -1184,9 +1281,12 @@ def _shown(value: Any) -> str:
def _field_warnings(data: dict[str, Any], evidence: dict[str, Any],
failed: str | None) -> list[tuple[str, str, list[str]]]:
failed: str | None, *, truncated: bool = False) -> list[tuple[str, str, list[str]]]:
"""核对结果里要让人看见的:(code, message, 字段)。mismatch 和 unresolved 各一条;missing 是模型
照实说没查到,值已经是空的,不另发警告。"""
照实说没查到,值已经是空的,不另发警告。
抽取两次都被截断(truncated)时 code 是 agent_field_truncated:原因是输出额度在推理阶段就用完了,
下一步是换非推理模型或调大最大输出 token,不是「稍后重试」——和别的抽取失败分开说。"""
out: list[tuple[str, str, list[str]]] = []
mismatched = [(p, e) for p, e in evidence.items() if e.get("status") == "mismatch"]
if mismatched:
@@ -1196,6 +1296,11 @@ def _field_warnings(data: dict[str, Any], evidence: dict[str, Any],
f"有 {len(mismatched)} 个字段与查询快照不一致,已按快照取值:{detail}", [p for p, _ in mismatched]))
unresolved = [(p, e) for p, e in evidence.items() if e.get("status") == "unresolved"]
if unresolved:
if failed and truncated:
out.append(("agent_field_truncated",
f"{failed}。{len(unresolved)} 个字段都记为空值(未以 0 代替)。{TRUNCATED_NEXT}",
[p for p, _ in unresolved]))
return out
if failed:
message = f"{failed}。{len(unresolved)} 个字段都记为空值(未以 0 代替)"
else:
+86 -10
View File
@@ -44,17 +44,20 @@ from app.engine import judge as judging
from app.engine.context import NodeContext, NodeError
from app.engine.labels import field_label, option_label
from app.engine.evidence import (
ABSENCE_RULE,
CELLS_REASON,
CLAIMS_RULE,
ENTITY_KINDS,
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 +73,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")
@@ -81,8 +85,10 @@ CLAIMS = CLAIMS_VALUES
#: 只标注、不让写作者重写也不判失败的违规:可疑实体(受管的正式出具按缺口降档)、表结构快照不全时
#: 核对不了的名字(出口只标注)
SOFT = frozenset({"unknown_entity", "unverified_entity"})
#: 交回写作者重写、重写不好也不判失败的违规:句中的 [[see:]](渲染后留下断句,数字和出处都没问题)
REWRITE_ONLY = frozenset({"inline_see"})
#: 交回写作者重写一次、重写不好也不判失败的违规:句中的 [[see:]](渲染后留下断句,数字和出处都没问题);
#: 写作目录有省略时的缺失断言(「没有查到」「数据缺失」:写作者没看全证据,结论依据不足,但数字和出处都核对过)。
#: 只剩这几类时不按 max_repairs 反复重写:重写一次没改掉,再写也多半一样,留在核对结果里标注
REWRITE_ONLY = frozenset({"inline_see", "absence_claim"})
#: 重写一次就是多一次整篇的模型调用,写不对的模型多给几次也大多写不对
MAX_REPAIRS = 3
@@ -140,6 +146,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,19 +408,21 @@ 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 "根据下面的证据写一份简洁的报告,先总后分。"
rules = f"{MARKER_RULES}\n{SEE_AND_CODE_RULES}"
if writer.guarded:
# 写作者看不全证据(目录有省略、有来源的结构化结论为空):「没有查到」这类结论没有依据
rules = f"{rules}\n{ABSENCE_RULE}"
if claims in ("require_citation", CLAIMS_JUDGE):
# 裁判只看写作者挂的依据:要判,就更得每句都挂上
rules = f"{rules}\n{CLAIMS_RULE}" + (f"\n{judging.JUDGE_RULE}" if claims == CLAIMS_JUDGE else "")
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 +451,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 +469,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
@@ -438,12 +506,13 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
def compose(raw: str) -> dict[str, Any]:
return compose_doc(raw, catalog, node_id=ctx.node.id, run_id=ctx.run.run_id, allow_numbers=allow,
cells_allowed=cells, entities=entities)
cells_allowed=cells, entities=entities, absence=writer.trimmed)
response, raw = await draft(messages)
doc = compose(raw)
repairs = 0
while (blocking := _blocking(doc["violations"], numbers)) and repairs < max_repairs:
while (blocking := _blocking(doc["violations"], numbers)) and repairs < max_repairs \
and not (repairs and not _fatal(blocking)):
repairs += 1
ctx.emit(EventType.LOG, level="warn", code="report_repair",
message=f"报告有 {len(blocking)} 处未通过核对({_named(blocking)}),"
@@ -466,6 +535,13 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
draft=draft, compose=compose, numbers=numbers, blocking=blocking)
spent.extend(usages)
if absent := [v for v in doc["violations"] if v.get("code") == "absence_claim"]:
# 重写之后还在(或者没让重写):报告照常产出,这几处标注出来,时间线上单独说清为什么依据不足
texts = list(dict.fromkeys(str(v.get("text") or "") for v in absent))
tried = f"已要求模型重写 {repairs} 次仍未改掉" if repairs else "未要求重写"
ctx.emit(EventType.LOG, level="warn", code="report_absence_claim", count=len(absent), texts=texts,
message=f"报告在证据目录有省略时写了 {len(absent)} 处缺失断言({_named(absent)}),{tried}:"
"被省略的查询仍在证据台账里,这类结论依据不足,应写作本报告未覆盖。报告照常产出,这几处已标注")
doc_artifact = await _store(doc, ctx)
ctx.emit(EventType.REPORT_CHECKED, doc_artifact=doc_artifact, ok=not doc["violations"],
stats=doc["stats"], violations=doc["violations"][:20], repairs=repairs,
+6
View File
@@ -89,12 +89,18 @@ 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),
# 目录有省略时写了「没有查到」这类缺失断言、重写后仍在:报告照常产出并标注,这几句结论依据不足
"report_absence_claim": ("report_absence_claim", DEGRADED),
# 口径卡的指标集、报告文档没能落进工件库:答案照常,只是点不开证据
"evidence_store_failed": ("evidence_store_failed", DEGRADED),
# agent 的结构化字段:模型报的值和查询快照对不上(已按快照取值),或者出处核对不了
# (记成空值,不兜底成 0)。答案能用,缺口要说清是哪一种
"agent_field_mismatch": ("agent_field_mismatch", DEGRADED),
"agent_field_unverified": ("agent_field_unverified", DEGRADED),
# 结构化抽取两次都被截断(输出额度在推理阶段就用完了):字段都记空值,答案能用,缺口要说清是额度问题
"agent_field_truncated": ("agent_field_truncated", DEGRADED),
# 结论句裁判(claims: judge):触顶、没跑成的句子记为未裁判,出具不能判完整——是缺口,要说明
"judge_limit": ("judge_limit", DEGRADED),
"judge_failed": ("judge_failed", DEGRADED),
+106
View File
@@ -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)
+208
View File
@@ -0,0 +1,208 @@
"""cite_fields 的结构化抽取:输出被截断时加大额度重试,仍截断就照实说是额度在推理阶段用完了。
2026-10 一次运行里 4 个开了 cite_fields 的 Agent 抽取全部失败:抽取用 thinking: off、额度 4096,而经 OpenAI 兼容接口
接入的推理模型关不掉思考,4096 在推理阶段就用完了,openai 抛 LengthFinishReasonError(「Could not parse response
content as the length limit was reached」)。日志却写「请稍后重试;如持续失败,请到设置测试该模型」——重试、测试都没用。
- 额度默认至少 16384,节点配了更大的用节点的
- 认出「输出被截断」:openai 的 length limit 类错误、finish_reason=length、Anthropic 的 stop_reason=max_tokens;
被截断时额度翻倍(最多 32768)重试一次
- 仍然截断:日志 code agent_field_truncated,说清是额度在推理阶段用完了,给出能做的下一步;字段记空值,不以 0 代替
- 服务方嫌额度太大(400,报错里有 max_tokens 一类字样):退回小一档的额度重来,不能因为额度提高反而抽不成
- 非截断类的失败照旧(agent_field_unverified)
夹具全部合成:门店、会员这类虚构的表。
"""
from __future__ import annotations
import json
import pytest
from langchain_core.messages import AIMessage
from openai import LengthFinishReasonError
from openai.types import CompletionUsage
from openai.types.chat import ChatCompletion
from app.engine.nodes import llm as llm_mod
from app.engine.runner import run_manager
from tests.test_agent_cited_fields import ( # noqa: F401 - 夹具按名字引入
EXTRACTED,
FINAL,
SCHEMA,
SQLS,
TOOL,
Script,
engine_up,
events,
graph,
logs,
output_of,
shop,
wait,
)
def length_error(prompt: int = 12000, completion: int = 4096) -> LengthFinishReasonError:
"""openai 的结构化输出在 finish_reason=length 时抛的那个错(和真实运行里的原话一样)。"""
usage = CompletionUsage(completion_tokens=completion, prompt_tokens=prompt, total_tokens=prompt + completion)
return LengthFinishReasonError(completion=ChatCompletion.model_construct(
id="x", choices=[], created=0, model="m", object="chat.completion", usage=usage))
class BudgetRejected(Exception):
"""服务方嫌输出额度太大时的 400(DeepSeek chat 系只收 8K)。"""
status_code = 400
class Extraction:
"""记下每次抽取调用的额度(thinking 关掉的那次 get_chat_model),并按额度决定这次怎么回。
outcome(budget) 返回 None 表示照常回 EXTRACTED,抛异常就是这次调用失败。
"""
def __init__(self, monkeypatch, outcome):
from app.providers import mock_model
self.budgets: list[int | None] = []
real = llm_mod.get_chat_model
async def spy(session, spec):
if spec.thinking == "off":
self.budgets.append(spec.max_tokens)
return await real(session, spec)
monkeypatch.setattr(llm_mod, "get_chat_model", spy)
base = Script(monkeypatch)
loop = mock_model.MockChatModel._decide
def decide(model, messages):
if model.response_format:
outcome(self.budgets[-1])
return AIMessage(content=json.dumps(EXTRACTED, ensure_ascii=False))
return loop(model, messages)
monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
self.script = base
# --------------------------------------------------------------------------
# 认出「输出被截断」
# --------------------------------------------------------------------------
def test_length_errors_are_recognised():
assert llm_mod.truncated_error(length_error())
assert llm_mod.truncated_error(RuntimeError("Could not parse response content as the length limit was reached"))
assert not llm_mod.truncated_error(RuntimeError("网关超时"))
assert not llm_mod.truncated_error(BudgetRejected("max_tokens is too large: 16384"))
def test_cut_off_replies_are_recognised_for_both_families():
assert llm_mod.truncated_reply(AIMessage(content="", response_metadata={"finish_reason": "length"}))
assert llm_mod.truncated_reply(AIMessage(content="", response_metadata={"stop_reason": "max_tokens"}))
assert not llm_mod.truncated_reply(AIMessage(content="", response_metadata={"finish_reason": "stop"}))
assert not llm_mod.truncated_reply(None)
# --------------------------------------------------------------------------
# 额度与重试
# --------------------------------------------------------------------------
async def test_the_default_budget_leaves_room_for_reasoning(monkeypatch):
spy = Extraction(monkeypatch, lambda budget: None)
row = await wait((await run_manager.start(graph=graph(card=False), input_payload={})).id)
assert row.status == "succeeded", row.error
assert spy.budgets == [16384], "推理模型的思考也算在额度里:默认至少 16384"
async def test_a_larger_node_budget_is_used(monkeypatch):
spy = Extraction(monkeypatch, lambda budget: None)
row = await wait((await run_manager.start(graph=graph(card=False, max_tokens=24000), input_payload={})).id)
assert row.status == "succeeded", row.error
assert spy.budgets == [24000]
async def test_a_cut_off_extraction_is_retried_with_a_larger_budget(monkeypatch):
def outcome(budget):
if budget < 32768:
raise length_error(completion=budget)
spy = Extraction(monkeypatch, outcome)
row = await wait((await run_manager.start(graph=graph(card=False), input_payload={})).id)
assert row.status == "succeeded", row.error
assert spy.budgets == [16384, 32768], "截断后翻倍重试一次"
bot = await output_of(row.id, "bot")
assert bot["data"]["gmv"] == 600.5 and bot["data_evidence"]["gmv"]["status"] == "verified"
evts = await events(row.id)
assert not logs(evts, "agent_field_truncated")
# EXTRACTED 里 region 指向本节点没有的 Q9:只有它核对不了,别的字段都照常核对
assert [w["fields"] for w in logs(evts, "agent_field_unverified")] == [["region"]]
# 两次调用各记一行:第一次写明为什么重来
ends = [e.data for e in evts if e.type == "llm.end" and e.data.get("purpose") == "cite_fields"]
assert len(ends) == 2 and "32768" in ends[0]["error"] and not ends[1].get("error")
# 被截断那次的用量照样记账(花出去的 token 是真的)
assert ends[0]["output_tokens"] == 16384
async def test_cut_off_twice_is_reported_as_truncation_with_fields_empty(monkeypatch):
def outcome(budget):
raise length_error(completion=budget)
spy = Extraction(monkeypatch, outcome)
row = await wait((await run_manager.start(graph=graph(card=False), input_payload={})).id)
assert row.status == "succeeded", row.error
assert spy.budgets == [16384, 32768], "只重试一次"
bot = await output_of(row.id, "bot")
assert bot["data"] == {k: None for k in SCHEMA["properties"]}, "记空值,不以 0 代替"
assert all(e["status"] == "unresolved" for e in bot["data_evidence"].values())
evts = await events(row.id)
assert not logs(evts, "agent_field_unverified")
[warn] = logs(evts, "agent_field_truncated")
assert warn["level"] == "warn" and warn["fields"] == list(SCHEMA["properties"])
message = warn["message"]
assert "推理阶段就用完了" in message and "结构化结论没能写出来" in message
assert "32768" in message and "非推理模型" in message and "最大输出 token" in message
assert "请稍后重试" not in message and "测试该模型" not in message, "重试、测试模型都不解决额度问题"
assert "未以 0 代替" in message
assert warn.get("max_tokens") == 32768
async def test_a_node_budget_at_the_cap_is_not_retried(monkeypatch):
def outcome(budget):
raise length_error(completion=budget)
spy = Extraction(monkeypatch, outcome)
row = await wait((await run_manager.start(graph=graph(card=False, max_tokens=40000), input_payload={})).id)
assert row.status == "succeeded", row.error
assert spy.budgets == [40000], "额度已经超过重试上限,再调一次也一样"
[warn] = logs(await events(row.id), "agent_field_truncated")
assert "40000" in warn["message"]
async def test_a_budget_the_service_rejects_falls_back(monkeypatch):
def outcome(budget):
if budget > 8192:
raise BudgetRejected(f"Invalid max_tokens value, the valid range of max_tokens is [1, 8192]: {budget}")
spy = Extraction(monkeypatch, outcome)
row = await wait((await run_manager.start(graph=graph(card=False), input_payload={})).id)
assert row.status == "succeeded", row.error
assert spy.budgets == [16384, 8192]
bot = await output_of(row.id, "bot")
assert bot["data_evidence"]["gmv"]["status"] == "verified"
async def test_other_failures_are_reported_as_before(monkeypatch):
def outcome(budget):
raise RuntimeError("网关超时")
spy = Extraction(monkeypatch, outcome)
row = await wait((await run_manager.start(graph=graph(card=False), input_payload={})).id)
assert row.status == "succeeded", row.error
assert spy.budgets == [16384], "不是截断就不重试"
evts = await events(row.id)
assert not logs(evts, "agent_field_truncated")
[warn] = logs(evts, "agent_field_unverified")
assert "网关超时" in warn["message"] and "未以 0 代替" in warn["message"]
@@ -0,0 +1,82 @@
"""缺失断言(absence_claim):写作目录有省略时,「没有查到」「不存在」「缺失」「没有数据」这类结论依据不足。
2026-10 一次运行,写作者只看到上百次查询里的前二十来次,写出「某类数据缺失:本轮没有查到任何结果」这样的断言,
可那个 Agent 实际查到了上百条记录。
- 写作规则:清单有省略、或有来源的结构化结论为空时,不得断言没有数据,只能写「本报告未覆盖」并说明依据不足
- 组装文档:清单有省略时按一组明确的说法认出缺失断言,记 absence_claim,交回重写一次;重写后仍有的只标注、不判失败
(report.REWRITE_ONLY,和句中的 [[see:]] 同一种处理)
- 只认拿得准的说法:「不存在明显差异」「本报告未覆盖」「不代表没有数据」都不算
"""
from __future__ import annotations
import pytest
from app.engine.evidence import ABSENCE_RULE, build_catalog, compose_doc
from app.engine.nodes.report import REWRITE_ONLY
@pytest.fixture
def catalog():
card = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v1",
"metrics": [{"id": "gmv", "name": "销售额", "value": 45678.5, "unit": "元", "format": "thousands"}]}
return build_catalog(nodes={"card": card}, ledger=[{"kind": "metric_set", "node_id": "card", "artifact": "c" * 64}])
def claims(doc):
return [v for v in doc["violations"] if v["code"] == "absence_claim"]
def test_the_rule_tells_the_writer_what_to_write_instead():
for phrase in ("没有查到", "不存在", "缺失", "没有数据", "本报告未覆盖", "依据不足"):
assert phrase in ABSENCE_RULE
assert "absence_claim" in REWRITE_ONLY and "inline_see" in REWRITE_ONLY
@pytest.mark.parametrize("text, found", [
("卡券数据缺失:本轮没有查到核销、兑换、退券的任何结果。", ["数据缺失", "没有查到"]),
("卡券核销没有任何记录。", ["没有任何记录"]),
("本周门店客流数据为空。", ["数据为空"]),
("活动内容无数据。", ["无数据"]),
("未查询到会员流失的情况。", ["未查询到"]),
("库里不存在相关记录。", ["不存在相关记录"]),
("兑换记录均不存在。", ["记录均不存在"]),
("暂无数据,查不到核销明细。", ["暂无数据", "查不到"]),
("| 方向 | 结论 |\n| --- | --- |\n| 卡券 | 没有查到 |", ["没有查到"]),
])
def test_absence_claims_are_recognised(catalog, text, found):
doc = compose_doc(text, catalog, absence=True)
got = claims(doc)
assert [v["text"] for v in got] == found, doc["violations"]
for v in got:
assert doc["markdown"][v["span"][0]:v["span"][1]] == v["text"], "span 指着正文里的那几个字"
assert v.get("unit") and v.get("context")
assert "依据不足" in v["message"] and "本报告未覆盖" in v["for_model"]
@pytest.mark.parametrize("text", [
"本报告未覆盖卡券核销的明细:证据目录有省略,相关查询未展开,依据不足。",
"两家门店的客单价不存在明显差异。",
"目录里没展开的查询不代表没有数据。",
"这并非没有查到,而是没有展开。",
"本周无数据库变更,无数据源切换。",
"本周销售额 [[m:gmv]],缺少口径说明的部分已注明。",
"```sql\nSELECT * FROM t -- 没有查到\n```",
])
def test_other_wording_is_left_alone(catalog, text):
assert not claims(compose_doc(text, catalog, absence=True))
def test_only_checked_when_the_catalog_was_trimmed(catalog):
text = "卡券核销数据缺失:没有查到任何记录。"
assert not claims(compose_doc(text, catalog))
assert claims(compose_doc(text, catalog, absence=True))
def test_the_claim_can_be_located_in_the_document(catalog):
doc = compose_doc("## 卡券\n\n本周销售额 [[m:gmv]]。核销数据缺失。", catalog, absence=True)
[v] = claims(doc)
segments = {s["id"]: s for b in doc["blocks"] for u in b["units"] for s in u["segments"]}
seg = segments[v["segment"]]
assert seg["kind"] == "text" and seg["span"][0] <= v["span"][0] < seg["span"][1]
assert v["context"] == "核销数据缺失。"
+209
View File
@@ -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()
@@ -0,0 +1,118 @@
"""报告里单元格的 ISO 日期时间显示成「2026-06-30 08:21:39」,不再是「2026-06-30T08:21:39」。
只改显示:快照里的值、引用记下的 value、eid、证据下钻都不变。复核要按文档生成时的规则:老文档里存的是带 T 的字,
拿新规则重新渲染会对不上,被当成文档被改过(render_mismatch)——所以新文档在顶层记 cell_display: 2(只在真有日期
时间被改写时记,没有的文档和以前一字不差),没有这个字段的老文档按 1 版(原样)复核。期 4 的金样
(test_provenance_unchanged)照旧逐字不变。
夹具合成:门店、卡券。
"""
from __future__ import annotations
import pytest
from app.core.artifact_store import canonical_json
from app.engine import evidence as ev
from app.engine.evidence import (
CELL_DISPLAY,
StreamRenderer,
build_catalog,
cell_eid,
compose_doc,
doc_cell_display,
iter_segments,
verify_doc,
writer_catalog,
)
ART = "d" * 64
SNAP = {"columns": ["store_name", "redeemed_at", "biz_date", "amount", "note"],
"column_types": {"store_name": "text", "redeemed_at": "datetime", "biz_date": "date", "amount": "number",
"note": "text"},
"rows": [["门店01", "2026-06-30T08:21:39", "2026-06-30", 120, "T 开头的备注"],
["门店02", "2026-06-29T23:05:00.250000+08:00", "2026-06-29", 80, "2026-06-29T10:00:00Z"]],
"truncated": False}
STORE = {ART: SNAP}
def load(artifact):
return STORE.get(artifact)
@pytest.fixture
def catalog():
ledger = [{"kind": "query", "node_id": "fetch", "exec": 1, "artifact": ART, "tool": "db_query__shop",
"columns": SNAP["columns"], "rows": 2, "truncated": False}]
return build_catalog(nodes={}, ledger=ledger)
def cells(doc):
return [s for s in iter_segments(doc) if s.get("ref", "").startswith("v:")]
def test_iso_datetimes_read_with_a_space(catalog):
doc = compose_doc("最近一笔核销在 [[v:Q1.r0.redeemed_at]],门店 [[v:Q1.r0.store_name]]。[[see:Q1]]", catalog, loader=load)
[when, store] = cells(doc)
assert when["text"] == "2026-06-30 08:21:39" and "2026-06-30 08:21:39" in doc["markdown"]
# 只改显示:引用记下的值、证据标识照旧
assert when["cite"]["value"] == "2026-06-30T08:21:39"
assert when["cite"]["eid"] == cell_eid(ART, 0, "redeemed_at")
assert store["text"] == "门店01"
assert doc["cell_display"] == CELL_DISPLAY == 2 and doc_cell_display(doc) == 2
assert '"cell_display":2' in canonical_json(doc), "在文档顶层,参与内容哈希"
assert verify_doc(doc, catalog, loader=load)["ok"]
def test_fractions_and_offsets_are_kept_and_dates_are_untouched(catalog):
doc = compose_doc("时间 [[v:Q1.r1.redeemed_at]],日期 [[v:Q1.r1.biz_date]],备注 [[v:Q1.r1.note]]。", catalog,
loader=load)
assert [s["text"] for s in cells(doc)] == ["2026-06-29 23:05:00.250000+08:00", "2026-06-29", "2026-06-29 10:00:00Z"]
assert verify_doc(doc, catalog, loader=load)["ok"]
def test_tables_read_the_same_way(catalog):
doc = compose_doc("[[table:Q1 cols=store_name,redeemed_at]]", catalog, loader=load)
assert "| 门店01 | 2026-06-30 08:21:39 |" in doc["markdown"]
assert "T08:21:39" not in doc["markdown"]
assert verify_doc(doc, catalog, loader=load)["ok"]
def test_documents_without_datetimes_are_unchanged(catalog):
doc = compose_doc("门店 [[v:Q1.r0.store_name]],金额 [[v:Q1.r0.amount]]。", catalog, loader=load)
assert "cell_display" not in doc, "没有被改写的日期时间:文档和以前一字不差"
assert doc_cell_display(doc) == 1
def test_old_documents_still_verify_as_they_were(catalog, monkeypatch):
"""改动前组装的文档:存的是带 T 的字、没有 cell_display。按 1 版复核,不报 render_mismatch。"""
monkeypatch.setattr(ev, "CELL_DISPLAY", 1)
old = compose_doc("最近一笔核销在 [[v:Q1.r0.redeemed_at]]。", catalog, loader=load)
monkeypatch.undo()
assert "cell_display" not in old and "2026-06-30T08:21:39" in old["markdown"]
checked = verify_doc(old, catalog, loader=load)
assert checked["ok"], checked["violations"]
def test_a_document_edited_back_is_caught(catalog):
doc = compose_doc("最近一笔核销在 [[v:Q1.r0.redeemed_at]]。", catalog, loader=load)
[seg] = cells(doc)
start = seg["span"][0]
doc["markdown"] = doc["markdown"].replace("2026-06-30 08:21:39", "2026-06-30T08:21:39")
seg["text"] = "2026-06-30T08:21:39"
seg["span"] = [start, start + len(seg["text"])]
codes = {v["code"] for v in verify_doc(doc, catalog, loader=load)["violations"]}
assert "render_mismatch" in codes
def test_an_unknown_display_version_is_reported(catalog):
doc = compose_doc("最近一笔核销在 [[v:Q1.r0.redeemed_at]]。", catalog, loader=load)
doc["cell_display"] = 9
assert "bad_schema" in {v["code"] for v in verify_doc(doc, catalog, loader=load)["violations"]}
def test_the_writer_and_the_stream_see_what_the_report_will_show(catalog):
text = writer_catalog(catalog, loader=load).text
assert "redeemed_at=2026-06-30 08:21:39" in text and "T08:21:39" not in text
stream = StreamRenderer(catalog, loader=load)
shown = stream.feed("最近一笔在 [[v:Q1.r0.redeemed_at]]。") + stream.flush()
assert "2026-06-30 08:21:39" in shown
@@ -0,0 +1,316 @@
"""回归:「4 个专项 Agent 并行查库 → 报告」整条跑一遍,复现 2026-10 那次运行的形状。
那次运行状态是 succeeded,报告却不可靠:4 个 Agent 的结构化抽取全部被截断(推理模型的额度在推理阶段就用完了);
写作者的出处清单写死 12000 字,上百次查询只有前二十来次露了面;写作者于是在清单被截断的情况下写了「数据缺失:
本轮没有查到任何核销结果」,可那个 Agent 查到了上百条。
这里用合成的门店、会员、卡券、活动库(SQLite)真跑:4 个 Agent 分别查 26、17、79、42 次(虚构的次数,一次并行发完),都配了
output_schema 和按出处核对字段;写作模型、抽取模型都是脚本。验证:
- 抽取被截断后加大额度重试成功;两次都截断的记 agent_field_truncated,字段为空值
- 出处清单在预算内,每个 Agent 都有代表,被省略的查询有编号和摘要,开头说明各来源省略了几次
- 清单有省略时的缺失断言被认出来(absence_claim)、交回重写;重写后仍有的只标注、不判失败
"""
from __future__ import annotations
import asyncio
import json
import re
import sqlite3
import pytest
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from openai import LengthFinishReasonError
from openai.types import CompletionUsage
from openai.types.chat import ChatCompletion
from sqlalchemy import select
from app.core import artifact_store
from app.data.engine import engines
from app.db.base import SessionLocal
from app.db.models import DataSource, Run, RunEvent
from app.engine.evidence import ABSENCE_RULE, DEFAULT_CATALOG_BUDGET, MARKER_RULES
from app.engine.nodes import llm as llm_mod
from app.engine.nodes import report as report_mod
from app.engine.runner import run_manager
from tests.fixtures.sources import drop_source
SOURCE = "fa_store"
TOOL = f"db_query__{SOURCE}"
#: (节点 id, 画布上的名字, 记号, 查询次数, 表, 列)
AGENTS = [
("agent_member", "会员专项", "member", 26, "member_daily", ["store_name", "biz_date", "new_members", "active_members"]),
("agent_store", "门店专项", "store", 17, "store_daily", ["store_name", "biz_date", "visits", "sales"]),
("agent_coupon", "卡券核销", "coupon", 79, "coupon_redemption", ["store_name", "redeemed_at", "coupon_type", "amount"]),
("agent_campaign", "活动内容", "campaign", 42, "campaign_view", ["campaign", "biz_date", "views", "clicks"]),
]
SCHEMA = {"type": "object", "properties": {
"first_store": {"type": "string", "description": "第一次查询第一行的门店或活动"},
"first_value": {"type": "number", "description": "第一次查询第一行的第三列"},
"churn_rate": {"type": "number", "description": "流失率(要计算,填 null)"},
}}
ROWS_PER_QUERY = 5
ABSENT = ("## 概览\n\n各专项都已完成查询。[[see:Q1]]\n\n"
"## 卡券核销\n\n卡券数据缺失:本轮没有查到核销、兑换的任何结果。")
COVERED = ("## 概览\n\n各专项都已完成查询。[[see:Q1]]\n\n"
"## 卡券核销\n\n本报告未覆盖卡券核销的明细:证据目录有省略,相关查询未展开,依据不足。[[see:Q1]]")
def length_error(completion: int) -> LengthFinishReasonError:
usage = CompletionUsage(completion_tokens=completion, prompt_tokens=14000, total_tokens=14000 + completion)
return LengthFinishReasonError(completion=ChatCompletion.model_construct(
id="x", choices=[], created=0, model="m", object="chat.completion", usage=usage))
@pytest.fixture(autouse=True)
async def engine_up():
await run_manager.setup()
yield
await run_manager.shutdown()
@pytest.fixture
async def store_db(tmp_path):
"""合成库:每张表 120 行,第 i 次查询取 OFFSET i 的 5 行,次次结果不同。"""
path = tmp_path / "store.db"
db = sqlite3.connect(path)
for _, _, _, _, table, cols in AGENTS:
db.execute(f"CREATE TABLE {table} ({cols[0]} TEXT, {cols[1]} TEXT, {cols[2]} {'TEXT' if table == 'coupon_redemption' else 'INTEGER'}, {cols[3]} REAL)")
rows = []
for r in range(120):
when = f"2026-06-{r % 28 + 1:02d}" + (f"T08:{r % 60:02d}:39" if table == "coupon_redemption" else "")
third = f"券型{r % 4 + 1}" if table == "coupon_redemption" else (r * 7) % 300 + 5
rows.append((f"门店{r % 30 + 1:02d}", when, third, round((r * 13) % 900 + 0.5, 1)))
db.executemany(f"INSERT INTO {table} VALUES (?, ?, ?, ?)", rows)
db.commit()
db.close()
async with SessionLocal() as session:
row = DataSource(name=SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True)
session.add(row)
await session.commit()
source_id = row.id
await engines.invalidate(source_id)
yield path
await drop_source(source_id)
def sql_of(table: str, cols: list[str], i: int) -> str:
return f"SELECT {', '.join(cols)} FROM {table} ORDER BY rowid LIMIT {ROWS_PER_QUERY} OFFSET {i}"
def node(nid, ntype, label, **config):
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label, "config": config}}
def graph(counts: dict[str, int] | None = None, **report):
nodes = [node("start", "input", "开始")]
for nid, title, key, count, table, cols in AGENTS:
nodes.append(node(nid, "agent", title, prompt=f"【{key}】按{title}的口径查数,每次查一段", tools=[TOOL],
approval="never", parallel_tools=True, output_schema=SCHEMA, cite_fields=True,
assign_to=f"{key}_findings"))
nodes.append(node("write", "report", "专项报告", instructions="汇总各专项的发现", **report))
nodes.append(node("out", "output", "成果", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]))
edges = [{"source": "start", "target": a[0]} for a in AGENTS] + [{"source": a[0], "target": "write"} for a in AGENTS]
edges.append({"source": "write", "target": "out"})
return {"nodes": nodes, "edges": edges}
class World:
"""脚本化的模型:Agent 一步并行发完自己的查询再给结论;抽取按 outcome 决定;写作者依次回 drafts。"""
def __init__(self, monkeypatch, db_path, *, counts: dict[str, int], drafts: list[str], outcome):
from app.providers import mock_model
self.counts = counts
self.drafts = drafts
self.writer: list[list] = []
self.extract_calls: dict[str, list[int | None]] = {a[2]: [] for a in AGENTS}
self.budgets: list[int | None] = []
db = sqlite3.connect(db_path)
self.first_rows = {key: db.execute(sql_of(table, cols, 1)).fetchone() for _, _, key, _, table, cols in AGENTS}
db.close()
real = llm_mod.get_chat_model
async def spy(session, spec):
if spec.thinking == "off":
self.budgets.append(spec.max_tokens)
return await real(session, spec)
monkeypatch.setattr(llm_mod, "get_chat_model", spy)
world = self
def text(messages) -> str:
return "\n".join(m.content if isinstance(m.content, str) else json.dumps(m.content, ensure_ascii=False)
for m in messages)
def who(messages) -> str:
found = re.search(r"【(member|store|coupon|campaign)】", text(messages))
assert found, "认不出是哪个 Agent"
return found.group(1)
def decide(model, messages):
if model.response_format:
key = who(messages)
world.extract_calls[key].append(world.budgets[-1])
outcome(key, len(world.extract_calls[key]), world.budgets[-1])
return AIMessage(content=json.dumps(world.extracted(key), ensure_ascii=False))
if any(isinstance(m, SystemMessage) and "报告撰写人" in str(m.content) for m in messages):
world.writer.append(list(messages))
return AIMessage(content=world.drafts[min(len(world.writer), len(world.drafts)) - 1])
key = who(messages)
if model.tools and not any(isinstance(m, ToolMessage) for m in messages):
_, _, _, _, table, cols = next(a for a in AGENTS if a[2] == key)
return AIMessage(content="分段查询。", tool_calls=[
{"name": TOOL, "args": {"sql": sql_of(table, cols, i)}, "id": f"{key}_{i}"}
for i in range(1, world.counts[key] + 1)])
return AIMessage(content=f"【{key}】查完了,结论见各次查询。")
monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
def extracted(self, key: str) -> dict:
first = self.first_rows[key]
_, _, _, _, _, cols = next(a for a in AGENTS if a[2] == key)
return {"first_store": {"value": first[0], "from": {"call": "Q1", "row": 0, "column": cols[0]}},
"first_value": {"value": first[3], "from": {"call": "Q1", "row": 0, "column": cols[3]}},
"churn_rate": {"value": None, "from": None}}
async def finish(graph_json) -> Run:
run = await run_manager.start(graph=graph_json, input_payload={})
for _ in range(1200):
await asyncio.sleep(0.05)
async with SessionLocal() as session:
row = await session.get(Run, run.id)
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
return row
raise AssertionError(f"run {run.id} 没跑完:{row.status}")
async def events(run_id: str, etype: str | None = None) -> list[RunEvent]:
async with SessionLocal() as session:
q = select(RunEvent).where(RunEvent.run_id == run_id)
if etype:
q = q.where(RunEvent.type == etype)
return list((await session.execute(q.order_by(RunEvent.seq))).scalars())
def logs(evts, code):
return [e.data for e in evts if e.type == "log" and e.data.get("code") == code]
async def output_of(run_id: str, node_id: str) -> dict:
[finished] = [e for e in await events(run_id, "node.finished") if e.node_id == node_id]
return artifact_store.load(finished.data["artifact"])
def catalog_text(messages) -> str:
"""写作者第一次收到的提示里出处清单那一段(从「可引用的证据」到写作规则之前)。"""
prompt = next(m.content for m in messages if isinstance(m, HumanMessage))
return prompt[prompt.index("可引用的证据"):prompt.index(MARKER_RULES)].rstrip()
# --------------------------------------------------------------------------
# 和那次运行同样的形状:上百次查询、分布不均
# --------------------------------------------------------------------------
async def test_the_reported_run_shape_end_to_end(monkeypatch, store_db):
counts = {key: count for _, _, key, count, _, _ in AGENTS}
def outcome(key, attempt, budget):
if key == "coupon":
raise length_error(budget) # 两次都截断
if key == "member" and attempt == 1:
raise length_error(budget) # 截断一次,加大额度后成功
world = World(monkeypatch, store_db, counts=counts, drafts=[ABSENT, COVERED], outcome=outcome)
row = await finish(graph())
assert row.status == "succeeded", row.error
evts = await events(row.id)
# 1. 抽取:会员专项截断后重试成功;卡券核销两次都截断,字段为空值
assert world.extract_calls["member"] == [16384, 32768]
assert world.extract_calls["coupon"] == [16384, 32768]
assert world.extract_calls["store"] == world.extract_calls["campaign"] == [16384]
member = await output_of(row.id, "agent_member")
assert member["data"]["first_store"] == world.first_rows["member"][0]
assert member["data_evidence"]["first_store"]["status"] == "verified"
coupon = await output_of(row.id, "agent_coupon")
assert coupon["data"] == {"first_store": None, "first_value": None, "churn_rate": None}
[cut] = logs(evts, "agent_field_truncated")
assert cut["fields"] == ["first_store", "first_value", "churn_rate"] and "推理阶段就用完了" in cut["message"]
assert not logs(evts, "agent_field_unverified")
# 2. 出处清单:预算内,每个 Agent 都有代表,被省略的有编号和摘要,开头说明各来源省略了几次
first = world.writer[0]
text = catalog_text(first)
assert len(text) <= DEFAULT_CATALOG_BUDGET, len(text)
head = text[:900]
assert "目录有省略" in head
for _, title, _, count, table, _ in AGENTS:
assert f"{title} {count} 次" in head
group = text[text.index(f"来自「{title}」的查询"):]
group = group[:group.index("\n\n", 1)] if "\n\n" in group[1:] else group
assert f"共 {count} 次" in group and re.search(r"\n {2}r0:", group), f"「{title}」没有展开的查询"
briefs = re.findall(r"\n {2}- (Q\d+):(\w+),5 行", text)
assert briefs, "没有列出被省略的查询"
assert {t for _, t in briefs} <= {a[4] for a in AGENTS}
assert all(f"Q{n}:" in text for n in range(1, 165)), "每次查询都要在清单里(展开的或摘要)"
# 结构化结论排在查询前面;抽取失败的卡券核销写明结构化结论为空、不代表没有数据
assert text.index("各 Agent 交来的结构化结论") < text.index("来自「")
assert "「卡券核销」:结构化结论为空" in text
# 写作规则里有缺失断言那一条
assert ABSENCE_RULE in next(m.content for m in first if isinstance(m, HumanMessage))
[trimmed] = logs(evts, "report_catalog_trimmed")
assert trimmed["queries"] == 164 and trimmed["budget"] == DEFAULT_CATALOG_BUDGET
assert [n["title"] for n in sorted(trimmed["nodes"], key=lambda n: n["title"])] == \
sorted(a[1] for a in AGENTS)
assert trimmed["omitted"] == sum(n["omitted"] for n in trimmed["nodes"]) > 0
# 3. 缺失断言:认出来、交回重写一次;改成「本报告未覆盖」后通过
assert len(world.writer) == 2
request = str(world.writer[1][-1].content)
assert "本报告未覆盖" in request and "没有查到" in request
[repair] = logs(evts, "report_repair")
assert "数据缺失" in repair["message"] or "没有查到" in repair["message"]
[checked] = [e.data for e in evts if e.type == "report.checked"]
assert not [v for v in checked["violations"] if v["code"] == "absence_claim"]
assert not logs(evts, "report_absence_claim")
# --------------------------------------------------------------------------
# 重写后仍然写缺失断言:只标注、不判失败
# --------------------------------------------------------------------------
SMALL = {"member": 6, "store": 4, "coupon": 12, "campaign": 8}
async def test_an_absence_claim_that_survives_the_rewrite_is_flagged_not_failed(monkeypatch, store_db):
monkeypatch.setattr(report_mod, "catalog_budget", lambda window: 4000)
world = World(monkeypatch, store_db, counts=SMALL, drafts=[ABSENT], outcome=lambda *a: None)
row = await finish(graph(on_violation="fail", max_repairs=3))
assert row.status == "succeeded", row.error
assert len(world.writer) == 2, "缺失断言只交回重写一次,不按最多重写次数反复重写"
evts = await events(row.id)
[checked] = [e.data for e in evts if e.type == "report.checked"]
assert checked["failed"] is False
claims = [v for v in checked["violations"] if v["code"] == "absence_claim"]
assert {v["text"] for v in claims} == {"数据缺失", "没有查到"}
assert all(v.get("segment") and v.get("context") for v in claims), "带着片段和原句:证据面板能定位"
[flag] = logs(evts, "report_absence_claim")
assert flag["level"] == "warn" and flag["count"] == 2 and flag["texts"] == ["数据缺失", "没有查到"]
assert "本报告未覆盖" in flag["message"] or "依据不足" in flag["message"]
async def test_without_omissions_an_absence_statement_is_left_alone(monkeypatch, store_db):
"""目录全给了:「没有查到」可能就是看了数据说的,不查。"""
counts = {"member": 1, "store": 1, "coupon": 1, "campaign": 1}
world = World(monkeypatch, store_db, counts=counts, drafts=[ABSENT], outcome=lambda *a: None)
row = await finish(graph())
assert row.status == "succeeded", row.error
assert len(world.writer) == 1
evts = await events(row.id)
assert not logs(evts, "report_catalog_trimmed") and not logs(evts, "report_absence_claim")
prompt = next(m.content for m in world.writer[0] if isinstance(m, HumanMessage))
assert ABSENCE_RULE not in prompt and "目录有省略" not in prompt
+13
View File
@@ -230,6 +230,19 @@ export function evidenceTally(cited: number, total: number, other = 0): string {
return parts.join(' · ')
}
/**
* 缺失断言:证据目录有省略时,报告写了「没有查到」「不存在」「缺失」这类结论(违规 absence_claim,重写后仍有时
* 报告节点另发日志 report_absence_claim)。证据面板的违规清单和运行时间线共用这一套说法
*/
export const ABSENCE_CLAIM_TEXT = {
label: '缺失断言',
hint: '写作模型没有看到全部证据:被省略的查询仍在证据台账里,看不到不等于不存在,这类结论应写作「本报告未覆盖」。'
+ '报告照常产出,这一处只标注、不判失败',
title: (n?: number) => `报告写了${n ? ` ${formatNumber(n)} 处` : ''}缺失断言,已标注(证据目录有省略)`,
next: '被省略的查询仍在证据台账里,可在证据面板中查看。请在报告撰写节点的指令中要求未展开的方向写作「本报告未覆盖」,'
+ '或换用上下文窗口更大的写作模型后重新运行',
}
/** 证据面板和证据条的文案 */
export const EVIDENCE_TEXT = {
allCited: (n: number) => `${formatNumber(n)} 个数字都有出处`,
+9 -3
View File
@@ -18,7 +18,7 @@ import {
} from '../lib/evidence'
import { humanizeError } from '../lib/errors'
import { formatDateTime, formatNumber, NONE, shortId } from '../lib/format'
import { EVIDENCE_TEXT, JUDGE_TEXT, MERGE_TEXT, NODE_TYPE_LABEL, RUN_SQL_CHECK_TEXT, SQL_CHECK_TEXT, nodeTypeLabel } from '../lib/terms'
import { ABSENCE_CLAIM_TEXT, EVIDENCE_TEXT, JUDGE_TEXT, MERGE_TEXT, NODE_TYPE_LABEL, RUN_SQL_CHECK_TEXT, SQL_CHECK_TEXT, nodeTypeLabel } from '../lib/terms'
import { useCatalog } from '../store/catalog'
import { askKey, segmentKey, useEvidence, useExplore, useVerdicts } from '../store/evidence'
import type {
@@ -1552,6 +1552,7 @@ const CODE_LABEL: Record<string, string> = {
unresolved_ref: '引用无法解析',
unknown_entity: '疑似不存在的名称',
unverified_entity: '无法核实的名称',
absence_claim: ABSENCE_CLAIM_TEXT.label,
}
/** 违规清单:报告撰写节点核对出来的全部问题,画得出线的能定位,画不出的说清在哪 */
@@ -1563,7 +1564,9 @@ function ViolationList({ doc, onLocate }: { doc: EvidenceDocData; onLocate: (seg
<p className="text-2xs text-faint">{EVIDENCE_TEXT.violationsHint}</p>
<ul className="space-y-2" data-ev-violations="">
{list.map((v, i) => {
const seg = locatable(doc, v)
// 缺失断言说的是一整句话,正文里普通文字不是可聚焦的片段:不给「定位」,原句写在下面
const sentence = v.code === 'absence_claim'
const seg = sentence ? null : locatable(doc, v)
return (
<li key={i} className="rounded border px-2 py-1.5" data-ev-violation={v.code}
data-ev-locatable={seg ? 'yes' : 'no'}>
@@ -1576,8 +1579,11 @@ function ViolationList({ doc, onLocate }: { doc: EvidenceDocData; onLocate: (seg
)}
</div>
{v.message && <p className="mt-1 text-dim">{v.message}</p>}
{v.code === 'absence_claim' && (
<p className="mt-0.5 text-2xs text-dim" data-ev-absence-hint="">{ABSENCE_CLAIM_TEXT.hint}</p>
)}
{v.context && <p className="mono mt-0.5 text-2xs text-faint [overflow-wrap:anywhere]">「{v.context.trim()}」</p>}
{!seg && (
{!seg && !sentence && (
<p className="mt-0.5 text-2xs" style={{ color: 'var(--st-waiting)' }}>
{v.segment ? EVIDENCE_TEXT.structural : EVIDENCE_TEXT.noSegment}
</p>
+43 -1
View File
@@ -2,7 +2,7 @@ import type { CatalogDriftTable, RunEvent, SqlCheckItem, TeamMember, TeamRound,
import type { NodeState, RunPhase } from './trace'
import { formatDuration, formatNumber } from '../lib/format'
import {
CATALOG_DRIFT_TEXT, CATALOG_PATCH_TEXT, COPILOT_CONTEXT_TEXT, JUDGE_TEXT, MERGE_TEXT, RUN_SQL_CHECK_TEXT, SCHEMA_PARTIAL_TEXT, SQL_CHECK_TEXT, TYPE_LABEL,
ABSENCE_CLAIM_TEXT, CATALOG_DRIFT_TEXT, CATALOG_PATCH_TEXT, COPILOT_CONTEXT_TEXT, JUDGE_TEXT, MERGE_TEXT, RUN_SQL_CHECK_TEXT, SCHEMA_PARTIAL_TEXT, SQL_CHECK_TEXT, TYPE_LABEL,
claimTally,
evidenceTally, issuanceLabel, nodeTypeLabel,
} from '../lib/terms'
@@ -759,6 +759,32 @@ function explainLog(code: string | undefined, message: string, d: Record<string,
sub: message.match(/(?:没|未)通过核对((.+?))/)?.[1],
}
}
case 'report_absence_claim': {
// 证据目录有省略时报告写了缺失断言,重写后仍在(report.py):读结构化字段 count、texts,不从原话里认
const n = typeof d.count === 'number' && d.count > 0 ? d.count : undefined
const texts = strs(d.texts)
return {
title: ABSENCE_CLAIM_TEXT.title(n),
sub: texts.length ? texts.map((t) => `「${t}」`).join('、') : undefined,
next: ABSENCE_CLAIM_TEXT.next,
}
}
case 'report_catalog_trimmed': {
// 写作目录超出篇幅(report.py _trimmed_message)。读结构化字段(queries、omitted、nodes),不从原话里认
const total = typeof d.queries === 'number' ? d.queries : undefined
const omitted = typeof d.omitted === 'number' ? d.omitted : undefined
const rows: any[] = Array.isArray(d.nodes) ? d.nodes : []
const per = rows.filter((r) => 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]
@@ -787,6 +813,22 @@ function explainLog(code: string | undefined, message: string, d: Record<string,
next: '模型给出的出处在查询结果中不存在。请让模型先查询到这些字段再提交,查询不到的字段记为空值',
}
}
case 'agent_field_truncated': {
// 结构化抽取两次都被截断(llm.py TRUNCATED_EXTRACT):「结构化抽取没有完成:模型的输出额度在推理阶段就用完了
// (最后一次的额度为 32768 token),结构化结论没能写出来。5 个字段都记为空值(未以 0 代替)。可以……」。
// 字段数、额度读结构化字段(fields、max_tokens),不从原话里认;不说「稍后重试」——重试解决不了额度问题
const n = strs(d.fields).length || Number(message.match(/(\d+)\s*个字段都?记为空值/)?.[1] ?? 0)
const budget = typeof d.max_tokens === 'number' && d.max_tokens > 0 ? d.max_tokens : undefined
return {
title: `模型的输出额度在推理阶段用完,${n ? `${n} 个` : ''}字段记为空值(未以 0 代替)`,
sub: `按出处抽取字段时输出被截断${budget ? `(已加大额度重试,最后一次为 ${formatNumber(budget)} token)` : ''},`
+ '结构化结论没能写出来',
next: '这些字段以空值传给下游,口径卡按缺少输入处理,报告不得据此断言没有数据。请为该 Agent 换用非推理模型'
+ '(抽取沿用 Agent 的模型),或调大它的「最大输出 token」后重新运行',
fix: 'canvas',
fixField: 'max_tokens',
}
}
case 'metric_incomplete': {
// 结构化字段(metrics.py):metric_name、reason(「基于被截断的查询结果计算(只取回了前 1000 行),结果不完整」)。
// 老运行的日志没有这两个字段:只给不点名的标题,原话留在展开区,不从中文里解析
+3 -2
View File
@@ -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)
+81
View File
@@ -1322,6 +1322,46 @@ 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}`)
// 目录有省略时报告写了缺失断言(「没有查到」「数据缺失」),交回重写一次仍在:只标注、不判失败,时间线单独一行说清
check('后端发缺失断言的警告,带着 count、texts', reportPy.includes('code="report_absence_claim"')
&& reportPy.includes('count=len(absent), texts=texts'), '改了 report.py 的字段就同步改 decode.ts 的 explainLog 和这里的夹具')
const ABSENT = '报告在证据目录有省略时写了 2 处缺失断言(「数据缺失」、「没有查到」),已要求模型重写 1 次仍未改掉:'
+ '被省略的查询仍在证据台账里,这类结论依据不足,应写作本报告未覆盖。报告照常产出,这几处已标注'
const absentRun = flatten(mod.decodeRun([E(1, 'node.started', 'w', { node_type: 'report', label: '写报告' }),
E(2, 'log', 'w', { level: 'warn', code: 'report_absence_claim', message: ABSENT, count: 2, texts: ['数据缺失', '没有查到'] }),
E(3, 'report.checked', 'w', { doc_artifact: 'abc', repairs: 1, ok: false, on_violation: 'fail', failed: false,
stats: { numbers: 0, numbers_cited: 0, uncited_numbers: 0, unresolved: 0 },
violations: [{ code: 'absence_claim', text: '数据缺失', message: '证据目录有省略时写了缺失断言:写作模型没有看到全部证据,「没有查到」「不存在」「缺失」这类结论依据不足' }] }),
]))
const ab = absentRun.find((s) => s.code === 'report_absence_claim')
check('report_absence_claim:标题说几处缺失断言、已标注、为什么(目录有省略)', ab?.title === '报告写了 2 处缺失断言,已标注(证据目录有省略)', ab?.title)
check('……副标题点名那几处说法', ab?.sub === '「数据缺失」、「没有查到」', ab?.sub)
check('……下一步说被省略的仍在台账里、应写「本报告未覆盖」;是提醒(warn),原话留在展开区',
!!ab?.next?.includes('本报告未覆盖') && !!ab?.next?.includes('证据台账') && ab?.level === 'warn' && ab?.detail === ABSENT, ab?.next)
const rcAbsent = absentRun.find((s) => s.code === 'report_checked')
check('……核对那一行照常完成(不判失败),违规清单里写着这条的说明', rcAbsent?.status === 'done'
&& !!rcAbsent?.detail?.includes('缺失断言') && !!rcAbsent?.detail?.includes('依据不足'), `${rcAbsent?.status} ${rcAbsent?.detail}`)
})
await section('工具门控 tool.gated(工具信任三档)', async () => {
@@ -1503,6 +1543,47 @@ await section('agent 字段按出处核对(可点击证据第二期):抽
check('「抽取结果格式有误(不是对象)」:同样说「失败,N 个字段记为空值」,原因放副标题,不说「出处不存在」',
notObject?.title === '按出处抽取字段失败,2 个字段记为空值(未以 0 代替)' && notObject?.sub === '抽取结果格式有误(不是对象)'
&& !notObject?.next?.includes('出处在查询结果中不存在'), `${notObject?.title} / ${notObject?.sub} / ${notObject?.next}`)
// 两次抽取都被截断(2026-10 一次运行:推理模型关不掉思考,4096 的额度在推理阶段就用完了):单独的 code,
// 说清是额度问题、给出能做的下一步,不说「稍后重试」「去设置测试该模型」(llm.py TRUNCATED_EXTRACT、TRUNCATED_NEXT)
const TRUNCATED = '结构化抽取没有完成:模型的输出额度在推理阶段就用完了(最后一次的额度为 32768 token),结构化结论没能写出来。'
+ '5 个字段都记为空值(未以 0 代替)。可以为该 Agent 换用非推理模型(抽取沿用 Agent 的模型),或者调大该 Agent 的「最大输出 token」后重新运行'
check('后端发截断警告的说法还是夹具里这一种', llmPy.includes('"agent_field_truncated"')
&& llmPy.includes('结构化抽取没有完成:模型的输出额度在推理阶段就用完了(最后一次的额度为 {budget} token),"')
&& llmPy.includes('"结构化结论没能写出来"') && llmPy.includes('可以为该 Agent 换用非推理模型(抽取沿用 Agent 的模型),或者调大该 Agent 的"')
&& llmPy.includes('已把额度加大到 {bigger} token 重试一次'), '改了 llm.py 的说法就同步改 decode.ts 的 explainLog 和这里的夹具')
const FIELDS5 = ['members', 'stores', 'coupons', 'redeemed', 'rate']
const cut = flatten(mod.decodeRun([
E(1, 'node.started', 'fetch', { node_type: 'agent', label: '会员专项' }),
E(2, 'llm.start', 'fetch', { model: 'm1', structured: true, purpose: 'cite_fields', message_count: 5, max_tokens: 16384 }),
E(3, 'llm.end', 'fetch', { agent: '会员专项', model: 'm1', purpose: 'cite_fields', duration_ms: 40000, output_tokens: 16384,
error: '输出被截断:16384 token 的输出额度用完了,结构化结论没有写完,已把额度加大到 32768 token 重试一次' }),
E(4, 'llm.start', 'fetch', { model: 'm1', structured: true, purpose: 'cite_fields', message_count: 5, max_tokens: 32768 }),
E(5, 'llm.end', 'fetch', { agent: '会员专项', model: 'm1', purpose: 'cite_fields', duration_ms: 80000, output_tokens: 32768,
error: '输出被截断:32768 token 的输出额度用完了,结构化结论没有写完' }),
E(6, 'log', 'fetch', { level: 'warn', code: 'agent_field_truncated', message: TRUNCATED, fields: FIELDS5, max_tokens: 32768 }),
E(7, 'node.finished', 'fetch', { duration_ms: 125000 }),
]))
const tr = cut.find((s) => s.code === 'agent_field_truncated')
check('agent_field_truncated:标题说额度在推理阶段用完、几个字段记为空值', tr?.title === '模型的输出额度在推理阶段用完,5 个字段记为空值(未以 0 代替)', tr?.title)
check('……副标题说抽取时输出被截断、重试到的额度(读 max_tokens 字段)',
tr?.sub === '按出处抽取字段时输出被截断(已加大额度重试,最后一次为 32,768 token),结构化结论没能写出来', tr?.sub)
check('……下一步:换非推理模型或调大最大输出 token;不说稍后重试,不叫人去测试模型',
!!tr?.next?.includes('非推理模型') && !!tr?.next?.includes('「最大输出 token」') && !tr?.next?.includes('稍后重试')
&& !tr?.next?.includes('测试') && !!tr?.next?.includes('不得据此断言没有数据'), tr?.next)
check('……入口打开该 Agent 的设置、落到最大输出 token;是提醒(warn),原话留在展开区',
tr?.fix === 'canvas' && tr?.fixField === 'max_tokens' && tr?.level === 'warn' && tr?.detail === TRUNCATED,
`${tr?.fix} ${tr?.fixField} ${tr?.level}`)
const tries = cut.filter((s) => s.kind === 'llm')
check('……两次抽取各成一行,第一次的展开区写明已加大额度重试', tries.length === 2
&& !!tries[0].detail?.includes('已把额度加大到 32768 token 重试一次') && !!tries[1].detail?.includes('结构化结论没有写完'),
tries.map((s) => s.detail).join(' | '))
check('……节点本身照样成功', cut.find((s) => s.kind === 'node')?.status === 'done')
const bareCut = flatten(mod.decodeRun([E(1, 'node.started', 'x', { node_type: 'agent' }),
E(2, 'log', 'x', { level: 'warn', code: 'agent_field_truncated', message: '结构化抽取没有完成。3 个字段都记为空值(未以 0 代替)' })]))
.find((s) => s.code === 'agent_field_truncated')
check('……缺 fields、max_tokens 字段时字段数从原话里认,不出 undefined',
bareCut?.title === '模型的输出额度在推理阶段用完,3 个字段记为空值(未以 0 代替)'
&& bareCut?.sub === '按出处抽取字段时输出被截断,结构化结论没能写出来', `${bareCut?.title} / ${bareCut?.sub}`)
const node = mod.decodeRun([E(1, 'node.started', 'fetch', { node_type: 'agent' }), E(2, 'node.finished', 'fetch', { duration_ms: 1 })])
check('对照:普通 agent 节点不多出任何行', flatten(node).filter((s) => s.kind !== 'node').length === 0)
const odd = flatten(mod.decodeRun([E(1, 'node.started', 'x', { node_type: 'agent' }),
+28 -1
View File
@@ -1094,7 +1094,34 @@ await section('sqlcheck', 'SQL 检查(数据目录阶段 4B):查询步骤
await bad.ctx.close()
})
await section('q-narrow', '查询步骤在 360px:栏内展开、底部抽屉都不横向滚动;减少动效', async () => {
await section('absence', '缺失断言:证据目录有省略时写了「没有查到」,违规清单写明是什么、为什么依据不足,带着原句', async () => {
// 违规的形状:backend/app/engine/evidence.py 的 absence_claims(code、message、text、span、segment、unit、context)
const MESSAGE = '证据目录有省略时写了缺失断言:写作模型没有看到全部证据,「没有查到」「不存在」「缺失」这类结论依据不足'
const r = await probeQ((id, body) => body, {
patchQuery: (f) => {
const units = f.doc.blocks.flatMap((b) => b.units ?? [])
const unit = units.find((u) => (u.segments ?? []).some((s) => s.kind === 'text' && s.text.trim().length > 2))
const seg = unit.segments.find((s) => s.kind === 'text' && s.text.trim().length > 2)
f.doc.violations = [...(f.doc.violations ?? []), { code: 'absence_claim', message: MESSAGE, text: '没有查到',
span: [seg.span[0], seg.span[0] + 1], segment: seg.id, unit: unit.id, context: '核销数据缺失:本轮没有查到任何结果。' }]
return f
},
})
await openQ(r.page, '1,288')
await r.page.locator('[data-evidence-panel] [data-ev-open-violations]').click()
await r.page.waitForSelector('[data-ev-violations]')
const item = r.page.locator('[data-ev-violation="absence_claim"]')
const text = await item.innerText().catch(() => '')
check('违规清单里有这一条,标「缺失断言」,不是笼统的「文档核对未通过」', text.includes('缺失断言') && !text.includes('文档核对未通过'), text)
check('……写明为什么依据不足、该怎么写(文案在 terms.ts)', text.includes(MESSAGE)
&& (await item.locator('[data-ev-absence-hint]').innerText().catch(() => '')).includes('本报告未覆盖'), text.replace(/\s+/g, ' '))
check('……带着原句;说的是一整句,不给「定位」,也不说「正文里没有对应文本」', text.includes('本轮没有查到任何结果')
&& await item.getByRole('button', { name: '定位' }).count() === 0 && !text.includes('没有对应文本'), text.replace(/\s+/g, ' '))
check('没有运行时报错', r.errors.length === 0, r.errors.join(' | '))
await r.ctx.close()
})
await section('q-narrow','查询步骤在 360px:栏内展开、底部抽屉都不横向滚动;减少动效', async () => {
const overflow = () => page.evaluate((sel) => {
const el = document.querySelector(sel)
return { sw: el.scrollWidth, cw: el.clientWidth }