From 3e1f8b25914524f162c53a41476d4cee11258b05 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sat, 3 Oct 2026 09:26:22 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=200=EF=BC=9A=E6=88=AA=E6=96=AD=E7=9A=84=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E7=BB=93=E6=9E=9C=E6=95=B4=E7=BB=84=E6=8B=BF=E6=9D=A5?= =?UTF-8?q?=E7=AE=97=E7=9A=84=E6=8C=87=E6=A0=87=E6=A0=87=E4=B8=BA=E4=B8=8D?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E3=80=81=E5=87=BA=E5=85=B7=E9=99=8D=E6=A1=A3?= =?UTF-8?q?=EF=BC=8C=E6=8A=A5=E5=91=8A=E6=95=B4=E8=A1=A8=E5=B8=A6=E6=88=AA?= =?UTF-8?q?=E6=96=AD=E8=AF=B4=E6=98=8E=EF=BC=8C=E6=81=B0=E5=A5=BD=E5=8F=96?= =?UTF-8?q?=E6=BB=A1=E4=B8=8D=E5=86=8D=E8=AF=AF=E6=8A=A5=E6=88=AA=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/evidence.py | 3 + backend/app/data/engine.py | 21 ++- backend/app/engine/evidence.py | 27 +++- backend/app/engine/expressions.py | 178 +++++++++++++++++++++++- backend/app/engine/issuance.py | 11 ++ backend/app/engine/nodes/io.py | 4 +- backend/app/engine/nodes/llm.py | 4 + backend/app/engine/nodes/metrics.py | 66 ++++++++- backend/tests/test_truncated_results.py | 34 ++++- 9 files changed, 328 insertions(+), 20 deletions(-) diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 3e2fefec..7b3ba4c7 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -822,6 +822,9 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s "version": (card or {}).get("caliber_version", entry.get("version")), "expression": (metric or {}).get("expression"), "substituted": (metric or {}).get("substituted"), "recompute_ok": (metric or {}).get("recompute_ok"), "status": source.get("status"), + # 拿截断的查询结果整组算出来的：面板标「结果不完整」并写明原因。完整的指标没有这两个键 + **({"incomplete": True, "incomplete_reason": source.get("incomplete_reason") or ""} + if source.get("incomplete") else {}), "node_id": entry.get("node_id"), "artifact": artifact or None, "eid": cite.get("eid"), "eid_ok": cite.get("eid") == entry.get("eid") == make_eid("metric", artifact or None, {"metric": metric_id}), "hash_ok": hash_ok and metric is not None, diff --git a/backend/app/data/engine.py b/backend/app/data/engine.py index f8109720..28439565 100644 --- a/backend/app/data/engine.py +++ b/backend/app/data/engine.py @@ -71,7 +71,10 @@ _MASK_SPLIT = re.compile(r"[,，、;；\n]+") class QueryResult: columns: list[str] rows: list[list[Any]] + #: 取回的行数。截断时只是取回的那一截，不是完整结果的行数 row_count: int + #: 撞了行数或字节上限、而且后面确实还有行：rows 只是完整结果的前 row_count 行。按行号取的单格是真实的， + #: 整组拿去计数、求和算出来的数不完整——口径卡据此标 incomplete，出具据此降档。恰好取满、后面没有了的不算 truncated: bool elapsed_ms: int sql: str @@ -888,17 +891,20 @@ async def run_query( columns = list(cursor.keys()) rows: list[list[Any]] = [] seen = _TypeTally(len(columns)) - truncated = False + truncated = full = False size = 0 async for row in cursor: + if full: + # 到了上限之后还有下一行：确实没取完。以前一到上限就标截断，恰好 1000 行的完整结果 + # 也算截断——截断的结果整组拿去算的指标要降档出具，误报就是把完整的结果降了档 + truncated = True + break # 逐行累加，边收边判上限：一次性 fetchall 一个亿级表就晚了 seen.add(row) values = [_jsonable(v) for v in row] rows.append(values) size += len(json.dumps(values, ensure_ascii=False, default=str)) - if len(rows) >= limits.max_rows or size >= limits.max_bytes: - truncated = True - break + full = len(rows) >= limits.max_rows or size >= limits.max_bytes return columns, rows, truncated, seen.types(columns) columns, rows, truncated, types = await _bounded( @@ -976,8 +982,9 @@ async def _run_write( async def _write(conn: AsyncConnection) -> tuple[list[str], list[list[Any]]]: result = await conn.execute(text(statement)) if result.returns_rows: # RETURNING、PG 的数据修改 CTE + # 多取一行：取到了才说明确实截断了（和查询的口径一致，恰好取满不算截断） return (list(result.keys()), - [[_jsonable(v) for v in row] for row in result.fetchmany(limits.max_rows)]) + [[_jsonable(v) for v in row] for row in result.fetchmany(limits.max_rows + 1)]) return ["affected_rows"], [[result.rowcount]] # engine.begin()：正常退出提交，出错（包括数据库按时限停下）回滚 @@ -985,11 +992,13 @@ async def _run_write( engine.begin, kind, limits, f"写操作超过 {_seconds(limits.timeout_seconds)} 秒被中断，已回滚。", _write, ) + truncated = len(rows) > limits.max_rows + rows = rows[:limits.max_rows] return QueryResult( columns=columns, rows=rows, row_count=len(rows), - truncated=len(rows) >= limits.max_rows, + truncated=truncated, elapsed_ms=int((time.perf_counter() - started) * 1000), sql=statement, ) diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 52865556..198bb816 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -898,6 +898,10 @@ def build_catalog( "value": metric.get("value"), "status": metric.get("status") or ("ok" if metric.get("value") is not None else "missing_input"), "rendered": rendered, "label": f"{metric.get('name') or mid} = {rendered}", + # 拿截断的查询结果整组算出来的：写作目录里提醒写作者，证据面板标「结果不完整」。 + # 只在不完整时有这两个键，完整的指标条目和以前一字不差 + **({"incomplete": True, "incomplete_reason": metric.get("incomplete_reason") or ""} + if metric.get("incomplete") else {}), } queries = retrievals = 0 @@ -1473,9 +1477,24 @@ def _table_plan(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapsho lines = ["| " + " | ".join(_clean(c) for c in wanted) + " |", "| " + " | ".join("---" for _ in wanted) + " |"] lines += ["| " + " | ".join(f"[[v:{alias}.r{r}.{c}]]" for c in wanted) + " |" for r in range(first, last + 1)] + if snapshot.get("truncated") is True: + lines.append(_truncated_note(first, last)) return "\n".join(lines), None +def _truncated_note(first: int, last: int) -> str: + """截断的查询结果生成的表格下面那行说明。表里每一格都是真实的，但读的人会把它当成全部数据。 + + 写成紧跟表格的引用块：切块（_scan_blocks）和前端 Markdown.tsx 都在不以竖线开头的行结束表格，引用块 + 只认以 > 开头的行，不会把下一段吞进来——普通段落接在表格后面，下一行正文就成了同一段。 + 不写取回了多少行：正文里没有出处的数字是裸数字（叙述层没有算术权限），行数写进来，每份带截断表格的 + 报告都会违规。「前 N 行」只在从第一行显示起时写，N 不超过整表上限 20，按序号放行（_structural_number）； + 行数在证据面板的查询步骤里能看到。 + """ + shown = f"表中只显示了前 {last + 1} 行；" if first == 0 else "" + return f"> 注：{shown}查询结果已截断，取回的数据并不完整。" + + def _expand_tables(text: str, catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, prev: str = "") -> str: """把能展开的整表标记换成 Markdown 表（每格一个 [[v:]]）。展开不了的原样留着，后面按解析不了处理。 @@ -3033,6 +3052,9 @@ def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) snapshot, why = snaps.get(alias, entry.get("artifact")) if why: return [*lines, f" （{why}，不要引用）"] + if snapshot.get("truncated") is True: + lines.append(" （查询结果已截断：只取回了前面一部分行，库里还有更多。单格可以引用；" + "不要把取回的行数当成总数，也不要据此写合计、全部之类的结论）") if not cells_allowed: return lines columns = [str(c) for c in snapshot.get("columns") or []][:_PROMPT_COLS] @@ -3084,7 +3106,10 @@ def catalog_prompt(catalog: dict[str, Any], *, budget: int = 12000, cells_allowe # 有值但按口径卡的格式显示不出来（太小、太大、不是有限的数）：引用了也是解析不了 lines.append(f"- [[{e['alias']}]] {e.get('name')}：按口径卡的格式显示不出来，不要引用") else: - lines.append(f"- [[{e['alias']}]] {e.get('name')} = {e.get('rendered')}") + # 不完整的指标照样能引用（单格、口径都对），但写作者得知道它不是全量：不写行数，免得照抄成裸数字 + caveat = "（基于被截断的查询结果计算，结果不完整：引用时须如实说明，不要写成全部数据）" \ + if e.get("incomplete") else "" + lines.append(f"- [[{e['alias']}]] {e.get('name')} = {e.get('rendered')}{caveat}") inputs = [e for e in catalog.values() if e.get("kind") == "input"] if inputs: lines.append("\n运行输入：") diff --git a/backend/app/engine/expressions.py b/backend/app/engine/expressions.py index 09ac4e15..cab96e95 100644 --- a/backend/app/engine/expressions.py +++ b/backend/app/engine/expressions.py @@ -285,6 +285,143 @@ def table_cell(table: Any, row: Any, column: Any) -> Any: return cell_value(raw) +# -------------------------------------------------------------------------- +# 截断的结果：按行号取单格照常，当成整组用时记一笔 +# +# 查询撞了行数或字节上限时，数据层只交回前面一截，标 truncated: true（data/engine.py 的 run_query）。 +# 这一截的前 N 行就是完整结果的前 N 行，所以 cell(x, 0, 'gmv')、rows[0] 取到的那一格是真实的； +# 可 len(rows) 只是取回的行数，sum / max / sorted / 成员判断 / 倒数第几行，说的都是这一截而不是 +# 完整结果。口径卡拿这种数出具，等于把「前 1000 行的人次」当成「入园人次」。 +# +# 不按正则猜表达式（写法千变万化，get(x, 'rows')、x['rows']、x.rows 都是同一件事），而是在求值时跟踪： +# 从截断结果里取 rows 的那一刻包一层 TruncatedRows，它照常当列表用，只在被整组用到时往 TruncatedUse +# 里记一笔。数据源工具交回的是 JSON 文本，表达式里只有 cell() 读得了它；经数据整形解析成对象之后 +# 才有 .rows 可取，所以这里认的是「带 truncated: true 和 rows 列表的对象」这种形状。 +# -------------------------------------------------------------------------- + + +class TruncatedUse: + """一次求值里，哪些截断的结果被当成了整组：sources 是 {键: {path, rows, artifact?}}。 + + - path：取到这组行的取值链（vars.visits；agent 的数组字段是字段本身 vars.kpi.amounts） + - rows：取回了几行——完整结果至少比这多 + - artifact：查询快照的工件 id，结果里带着的才有 + 同一份结果按工件 id（没有就按 path）只记一次：len(x.rows) + x.row_count 只用了一份结果。 + + edge(path) 给调用方认定的截断整组：口径卡用它标出 agent 经 cite_fields 交来、引用的行段到了截断结果 + 末行的数组字段。返回 {rows, artifact?} 或 None。求值器只对取到列表的取值链问它。 + """ + + __slots__ = ("sources", "edge") + + def __init__(self, edge: Any = None) -> None: + self.sources: dict[str, dict[str, Any]] = {} + self.edge = edge + + def record(self, path: str, rows: int, artifact: Any = None) -> None: + key = artifact if isinstance(artifact, str) and artifact else path + self.sources.setdefault(key, {"path": path, "rows": rows, + **({"artifact": artifact} if isinstance(artifact, str) and artifact else {})}) + + +class TruncatedRows: + """截断结果里的一组行。只活在一次求值里：eval_expression 交出结果前一律换回普通 list。 + + 整组用到的写法都要经过这里的某个特殊方法：len()、迭代（sum / min / max / sorted / any / all）、 + in、比较、拼接、转成文本、倒数第几行、取回的行之后的行号。按行号取取回范围之内的一行不记； + 真假判断也不记——截断的结果至少有一行，「有数据」这件事是真的。 + + 不继承 list：list 的 C 实现里有绕过子类方法的快路径（json 编码、[] + x 的 sq_concat 都直接读底层数组）， + 记漏一处就是把截断的数当成完整的出具。属性一律带下划线，表达式碰不到（私有属性在静态检查里就拒了）。 + """ + + __slots__ = ("_rows", "_use", "_path", "_count", "_artifact") + __hash__ = None # type: ignore[assignment] + + def __init__(self, rows: list[Any], use: TruncatedUse, path: str, count: int, artifact: Any = None) -> None: + self._rows, self._use, self._path, self._count, self._artifact = rows, use, path, count, artifact + + def _whole(self) -> list[Any]: + self._use.record(self._path, self._count, self._artifact) + return self._rows + + def __len__(self) -> int: + return len(self._whole()) + + def __iter__(self) -> Any: + return iter(self._whole()) + + def __reversed__(self) -> Any: + return reversed(self._whole()) + + def __contains__(self, item: Any) -> bool: + return item in self._whole() + + def __getitem__(self, index: Any) -> Any: + # 倒数第几行：完整结果的最后一行不是取回的最后一行；取回范围之外的行号：完整结果里有这一行， + # 只是没取回来（照旧抛 IndexError，求值器按取不到处理，原因记在这里） + if isinstance(index, int) and not isinstance(index, bool) and not 0 <= index < len(self._rows): + self._whole() + return self._rows[index] + + def __bool__(self) -> bool: + return bool(self._rows) + + def __eq__(self, other: Any) -> bool: + return self._whole() == _plain_rows(other) + + def __ne__(self, other: Any) -> bool: + return self._whole() != _plain_rows(other) + + def __lt__(self, other: Any) -> bool: + return self._whole() < _plain_rows(other) + + def __le__(self, other: Any) -> bool: + return self._whole() <= _plain_rows(other) + + def __gt__(self, other: Any) -> bool: + return self._whole() > _plain_rows(other) + + def __ge__(self, other: Any) -> bool: + return self._whole() >= _plain_rows(other) + + def __add__(self, other: Any) -> Any: + return self._whole() + _plain_rows(other) + + def __radd__(self, other: Any) -> Any: + return _plain_rows(other) + self._whole() + + def __mul__(self, times: Any) -> Any: + return self._whole() * times + + __rmul__ = __mul__ + + def __repr__(self) -> str: + return repr(self._whole()) + + +def _plain_rows(value: Any) -> Any: + return value._whole() if isinstance(value, TruncatedRows) else value + + +def _truncated_table(value: Any) -> bool: + """查询结果的形状、而且被截断了：数据源工具交回的 JSON 经数据整形解析成对象之后就是这样。""" + return isinstance(value, dict) and value.get("truncated") is True and isinstance(value.get("rows"), list) + + +def _settle(value: Any) -> Any: + """交出结果前把 TruncatedRows 换回普通 list（整组交出去也是整组用到）。没有包装的原样返回、不复制。""" + if isinstance(value, TruncatedRows): + return list(value._whole()) + if type(value) is list: + items = [_settle(v) for v in value] + return items if any(a is not b for a, b in zip(items, value)) else value + if type(value) is dict: + pairs = {k: _settle(v) for k, v in value.items()} + return pairs if any(pairs[k] is not v for k, v in value.items()) else value + return value + + _SAFE_FUNCS: dict[str, Any] = { "len": len, "str": str, @@ -551,17 +688,39 @@ def substitute(tree: ast.AST, values: dict[str, Any]) -> str: return ast.unparse(ast.fix_missing_locations(replaced)) -def eval_expression(expr: str, ctx: dict[str, Any]) -> Any: +def eval_expression(expr: str, ctx: dict[str, Any], *, track: TruncatedUse | None = None) -> Any: """求值一个受限的 Python 表达式。 只放行字面量、比较、布尔/算术运算、下标、属性访问和一小撮白名单函数。 没有 import、没有属性魔法、没有函数定义，所以用户在画布上写条件是安全的。 + + track：跟踪截断的结果有没有被当成整组用，记进 track.sources（口径卡用，见 TruncatedUse）。 + 不给就不跟踪：分支条件、数据整形、复算的行为和以前完全一样。 """ expr = (expr or "").strip() if not expr: return None tree, _, _ = parse_expression(expr) + def _tracked(base_src: Any, full_src: Any, base: Any, key: Any, value: Any) -> Any: + """取到的值要不要包一层：截断结果的 rows 包成 TruncatedRows，row_count 直接记一笔 + （它就是取回的行数，读它就是在用整组）；调用方经 edge 认定的截断整组也包起来。 + base_src / full_src 是取值链的源码（截断结果本身 / 取到的这个值），要用时才拼。""" + if track is None or isinstance(value, TruncatedRows): + return value + if isinstance(key, str) and key in ("rows", "row_count") and _truncated_table(base): + rows, artifact = base["rows"], base.get("artifact") + if key == "row_count": + track.record(base_src(), len(rows), artifact) + return value + return TruncatedRows(rows, track, base_src(), len(rows), artifact) + if isinstance(value, list) and track.edge is not None: + path = full_src() + info = track.edge(path) + if info: + return TruncatedRows(value, track, path, int(info.get("rows") or len(value)), info.get("artifact")) + return value + def _eval(node: ast.AST) -> Any: if isinstance(node, ast.Expression): return _eval(node.body) @@ -582,7 +741,8 @@ def eval_expression(expr: str, ctx: dict[str, Any]) -> Any: if isinstance(node, ast.Attribute): base = _eval(node.value) if isinstance(base, dict): - return base.get(node.attr) + return _tracked(lambda: ast.unparse(node.value), lambda: ast.unparse(node), base, node.attr, + base.get(node.attr)) if node.attr.startswith("_"): raise ExpressionError("不允许访问私有属性") return getattr(base, node.attr, None) @@ -590,9 +750,10 @@ def eval_expression(expr: str, ctx: dict[str, Any]) -> Any: base = _eval(node.value) key = _eval(node.slice) try: - return base[key] + value = base[key] except (KeyError, IndexError, TypeError): return None + return _tracked(lambda: ast.unparse(node.value), lambda: ast.unparse(node), base, key, value) if isinstance(node, ast.BinOp): op = _BIN_OPS.get(type(node.op)) if not op: @@ -635,7 +796,13 @@ def eval_expression(expr: str, ctx: dict[str, Any]) -> Any: raise ExpressionError("只能调用允许的函数") args = [_eval(a) for a in node.args] kwargs = {kw.arg: _eval(kw.value) for kw in node.keywords if kw.arg} - return _SAFE_FUNCS[node.func.id](*args, **kwargs) + result = _SAFE_FUNCS[node.func.id](*args, **kwargs) + if node.func.id == "get" and len(args) >= 2 and isinstance(args[1], str): + # get(x, 'rows') 和 x.rows 是同一件事，按 x['rows'] 的写法交给 edge + source = node.args[0] + result = _tracked(lambda: ast.unparse(source), lambda: f"{ast.unparse(source)}[{args[1]!r}]", + args[0], args[1], result) + return result if isinstance(node, (ast.List, ast.Tuple)): return [_eval(e) for e in node.elts] if isinstance(node, ast.Dict): @@ -646,7 +813,8 @@ def eval_expression(expr: str, ctx: dict[str, Any]) -> Any: } raise ExpressionError(_unsupported(type(node).__name__)) - return _eval(tree) + result = _eval(tree) + return _settle(result) if track is not None else result def eval_condition(expr: str, ctx: dict[str, Any]) -> bool: diff --git a/backend/app/engine/issuance.py b/backend/app/engine/issuance.py index d795eb3c..c7aad257 100644 --- a/backend/app/engine/issuance.py +++ b/backend/app/engine/issuance.py @@ -451,6 +451,17 @@ def decide_tier( return "formal" +def incomplete_gaps(metrics: list[dict[str, Any]]) -> list[str]: + """契约收来的指标里结果不完整的（口径卡标了 incomplete：拿截断的查询结果整组算出来的），每个一句缺口。 + + 交给 decide_tier 的 gaps：完整出具的前提是每个数都算全了。「前 1000 行的人次」不是「入园人次」， + 数字照样回指得上口径卡，只看回指的话它会被盖上完整出具的章。原因用口径卡记下的那句，读的人 + 看得懂是哪个指标、为什么不完整。 + """ + return [f"指标「{m.get('name') or m.get('id')}」{m.get('incomplete_reason') or '结果不完整'}" + for m in metrics if isinstance(m, dict) and m.get("incomplete")] + + def _claims_effect( policy: str | dict[str, Any] | None, unsupported: list[Any] | int | None, diff --git a/backend/app/engine/nodes/io.py b/backend/app/engine/nodes/io.py index 0134fea3..6ed8c676 100644 --- a/backend/app/engine/nodes/io.py +++ b/backend/app/engine/nodes/io.py @@ -147,7 +147,7 @@ def _apply_contract( ) -> dict[str, Any]: from datetime import datetime, timezone - from app.engine.issuance import decide_tier, trace_numbers + from app.engine.issuance import decide_tier, incomplete_gaps, trace_numbers nodes = state.get("nodes") or {} # 写了 report_from 就是引用模式：按报告文档逐段复核，不再按数值回指叙述 @@ -186,6 +186,8 @@ def _apply_contract( unresolved.append(source) if unresolved: gaps.append(f"指标来源未解析：{'、'.join(unresolved[:5])}") + # 拿截断的查询结果整组算出来的指标：数回指得上，但算的只是取回的那一截，不能完整出具 + gaps.extend(incomplete_gaps(metrics)) # 引用模式下，缺输入记成空值的指标（口径卡 on_missing=null）就是缺：报告里引用不了它， # 也不能因为 id 在清单里就算「齐了」。旧模式照旧只看 id，行为不变 diff --git a/backend/app/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index f4ef715b..42852f23 100644 --- a/backend/app/engine/nodes/llm.py +++ b/backend/app/engine/nodes/llm.py @@ -1163,6 +1163,10 @@ def _verify_rows(schema: dict[str, Any], told: Any, source: Any, by_alias: dict[ **({"columns": names} if fields is not None else {"column": names[""]})} cited = {"ref": f"{call}.r{first}-{last}" + ("" if fields is not None else f".{names['']}"), **base, "locator": locator, "eid": make_eid("rows", q["artifact"], locator)} + if content.get("truncated") is True and last == len(content.get("rows") or []) - 1: + # 行段到了截断快照的末行：库里还有没取回的行，可能也属于这个字段。口径卡对它整组求和、计数时 + # 据此标「不完整」（metrics._truncated_field）。停在末行之前的不标：引用的每一行都在快照里 + cited["truncated"] = True same = isinstance(told, list) and len(told) == len(truth) and all( (isinstance(m, dict) and all(same_value(m.get(f), t[f]) for f in t)) if isinstance(t, dict) else same_value(m, t) diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index b926b0e7..d3826b12 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -22,6 +22,7 @@ from app.engine.evidence import ( from app.engine.expressions import ( CellError, ExpressionError, + TruncatedUse, cell_parts, cell_value, eval_expression, @@ -45,6 +46,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: 每个指标还记下它是怎么来的：用到了哪些输入、各取自哪个节点、代入后的算式、 拿代入式复算是否一致。整张卡落成 metric_set 工件，id 进证据台账——报告里的 数字点开，一路能追到这里。 + + 表达式把截断的查询结果当成整组用了（len、sum、max……，见 expressions.TruncatedUse）的指标， + 值照算，但标 incomplete 并写明原因：它算的只是取回的那一截。出具契约据此降档（issuance.incomplete_gaps）。 """ definitions = ctx.cfg("metrics", []) or [] caliber_raw, caliber_version = ctx.cfg("caliber", "") or ctx.node.title, str(ctx.cfg("caliber_version", "") or "v1") @@ -87,8 +91,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: missing = [path for path in paths if values[path] is None] value: Any = None + use = TruncatedUse(edge=lambda path: _truncated_field(path, state, ctx)) try: - value = eval_expression(expr, tctx) + value = eval_expression(expr, tctx, track=use) except ExpressionError as e: errors.append(f"指标「{metric_id}」：表达式有误（{e}）") continue @@ -117,6 +122,11 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: if isinstance(value, float) and decimals is not None: value = round(value, decimals) substituted = substitute(tree, values) + incomplete: dict[str, Any] = {} + if use.sources: + sources = list(use.sources.values()) + incomplete = {"incomplete": True, "incomplete_reason": _incomplete_reason(sources), + "truncated_sources": sources} metrics.append( { "id": metric_id, @@ -133,6 +143,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 值是空的就是缺输入，不论 on_missing 是哪一种：fail 模式下直接取到空值的 # 指标（原式就是 vars.x.y 而 y 不存在）以前也是照常产出一个 None "status": "ok" if value is not None else "missing_input", + # 只在不完整时出现：完整的指标和以前一字不差，老运行复核时目录也对得上 + **incomplete, } ) @@ -142,10 +154,16 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # text 是给叙述节点 prompt 用的清单——它是叙述层唯一的数字来源 lines = [ - f"- {m['id']}（{m['name']}）= {m['value']}{m['unit']}" if m["value"] is not None - else f"- {m['id']}（{m['name']}）= {MISSING}（缺少输入，无值）" + (f"- {m['id']}（{m['name']}）= {m['value']}{m['unit']}" if m["value"] is not None + else f"- {m['id']}（{m['name']}）= {MISSING}（缺少输入，无值）") + # 不写行数：叙述节点照抄进正文就是一个没有出处的数 + + ("（不完整：基于被截断的查询结果计算）" if m.get("incomplete") else "") for m in metrics ] + for m in metrics: + if m.get("incomplete"): + ctx.emit(EventType.LOG, level="warn", code="metric_incomplete", metric=m["id"], + message=f"指标「{m['name']}」{m['incomplete_reason']}") card = { "kind": "metric_set", "caliber": caliber, @@ -305,6 +323,46 @@ def _value_of(path: str, tctx: dict[str, Any]) -> Any: return None +def _incomplete_reason(sources: list[dict[str, Any]]) -> str: + """「基于被截断的查询结果计算（只取回了前 1000 行），结果不完整」：出具声明、运行日志、证据面板都用这句。""" + fetched = [f"前 {s['rows']} 行" for s in sources] + if len(fetched) == 1: + return f"基于被截断的查询结果计算（只取回了{fetched[0]}），结果不完整" + return f"基于 {len(fetched)} 份被截断的查询结果计算（分别只取回了{'、'.join(fetched)}），结果不完整" + + +def _truncated_field(path: str, state: GraphState, ctx: NodeContext) -> dict[str, Any] | None: + """agent 经 cite_fields 交来、被截断切开的数组字段：{rows, artifact}。别的取值链返回 None。 + + 数组字段的出处是快照里的一段行（data_evidence 的 locator.rows = [a, b]）。cite_fields 核对时，这段行 + 到了截断快照的末行就记 truncated（llm._verify_rows）：快照后面还有没取回的行，可能也属于这个字段， + 对它求和、计数算的就只是取回的那一截。停在末行之前的不记——引用的每一行都在快照里，截断拿不走其中 + 任何一行。标量字段是一格，不是聚合，也不记。取回的行数就是末行的行号加一。 + + 路径怎么对到字段和 _input_of 一样：vars.X 找最后写它的节点，nodes. 下多走一层 data。 + 求值器只对取到列表的取值链问这里，一个指标问不了几次。 + """ + head = _REF_HEAD.match(path) + if not head or head.group(1) == "input": + return None + root, name, rest = head.groups() + field = rest.lstrip(".") + node_id = name if root == "nodes" else _writer_of(name, state, ctx) + producer = ctx.run.spec.node_map().get(node_id) if node_id and field else None + if producer is None or str(producer.type) != "agent": + return None + output = (state.get("nodes") or {}).get(producer.id) + evidence = output.get("data_evidence") if isinstance(output, dict) else None + key = _field_key(field) + if root == "nodes": + key = key[len("data."):] if key.startswith("data.") else "" + cited = evidence.get(key) if isinstance(evidence, dict) and key else None + rows = (cited.get("locator") or {}).get("rows") if isinstance(cited, dict) and cited.get("truncated") else None + if not (isinstance(rows, list) and len(rows) == 2 and isinstance(rows[1], int)): + return None + return {"rows": rows[1] + 1, "artifact": cited.get("artifact")} + + def _missing_reason(missing: list[str]) -> str: if not missing: return "有输入为空值，无法计算（请检查上游是否产出了这些字段）" @@ -404,7 +462,7 @@ def _agent_field(state: GraphState, node_id: str, root: str, field: str) -> dict if not isinstance(cited, dict): return _agent_element(evidence, key) out: dict[str, Any] = {"via": "agent_field", "field": key, "status": cited.get("status") or "unresolved"} - for k in ("ref", "artifact", "locator", "eid", "model_value", "reason"): + for k in ("ref", "artifact", "locator", "eid", "model_value", "reason", "truncated"): if k in cited: out[k] = cited[k] return out diff --git a/backend/tests/test_truncated_results.py b/backend/tests/test_truncated_results.py index c6693248..2dc1ac2e 100644 --- a/backend/tests/test_truncated_results.py +++ b/backend/tests/test_truncated_results.py @@ -20,6 +20,7 @@ import sqlite3 from types import SimpleNamespace import pytest +from httpx import ASGITransport, AsyncClient from langchain_core.messages import AIMessage, ToolMessage from sqlalchemy import select @@ -28,10 +29,11 @@ from app.data.engine import engines, run_query from app.data.guard import QueryLimits from app.db.base import SessionLocal from app.db.models import DataSource, Run, RunEvent -from app.engine.evidence import StreamRenderer, build_catalog, compose_doc, verify_doc +from app.engine.evidence import StreamRenderer, build_catalog, compose_doc, iter_segments, verify_doc from app.engine.expressions import TruncatedUse, eval_expression from app.engine.issuance import decide_tier, incomplete_gaps from app.engine.runner import run_manager +from app.main import app # -------------------------------------------------------------------------- # 求值器：整组用到截断结果时记一笔，单格不记 @@ -373,10 +375,18 @@ def daily(limit: int | None) -> dict: class Writer: + """报告写作者：每次都照 text 写；记下看到的提示。""" + def __init__(self, monkeypatch, text: str = DAILY): from app.providers import mock_model - monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda model, messages: AIMessage(content=text)) + self.prompts: list[str] = [] + + def decide(model, messages): + self.prompts.append("\n".join(str(m.content) for m in messages)) + return AIMessage(content=text) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) async def wait(run_id: str) -> Run: @@ -403,7 +413,7 @@ async def output_of(run_id: str, node_id: str) -> dict: async def test_counting_a_truncated_result_degrades_the_issuance(park, monkeypatch): - Writer(monkeypatch) + writer = Writer(monkeypatch) run = await run_manager.start(graph=daily(limit=2), input_payload={}) row = await wait(run.id) assert row.status == "succeeded", row.error @@ -432,6 +442,24 @@ async def test_counting_a_truncated_result_degrades_the_issuance(park, monkeypat assert "首张门票 60。" in report [checked] = [e.data for e in await events(row.id, "report.checked")] assert checked["ok"] is True, checked["violations"] + # 写作目录提醒写作者：指标不完整、查询已截断（不写行数，免得照抄成裸数字） + prompt = writer.prompts[0] + assert "入园人次 = 2人次（基于被截断的查询结果计算，结果不完整" in prompt + assert "查询结果已截断：只取回了前面一部分行" in prompt + + # 证据面板：点开报告里的这个指标，指标步骤带着「不完整」和原因 + doc = artifact_store.load(row.output["_evidence"]["doc_artifact"]) + assert doc["catalog"]["m:visits"]["incomplete"] is True + assert "incomplete" not in doc["catalog"]["m:first"] + seg = next(s for s in iter_segments(doc) if s["text"] == "2人次") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json() + metric = body["chain"][0] + assert metric["step"] == "metric" and metric["incomplete"] is True + assert metric["incomplete_reason"] == reason + first = next(s for s in iter_segments(doc) if s.get("ref") == "m:first") + body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{first['id']}")).json() + assert "incomplete" not in body["chain"][0] async def test_the_same_card_on_a_complete_result_issues_formally(park, monkeypatch):