数据目录阶段 0:截断的查询结果整组拿来算的指标标为不完整、出具降档,报告整表带截断说明,恰好取满不再误报截断

This commit is contained in:
ThearchyHelios
2026-10-03 09:26:22 +08:00
parent 90ff89f200
commit 3e1f8b2591
9 changed files with 328 additions and 20 deletions
+3
View File
@@ -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,
+15 -6
View File
@@ -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,
)
+26 -1
View File
@@ -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运行输入:")
+173 -5
View File
@@ -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:
+11
View File
@@ -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,
+3 -1
View File
@@ -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,行为不变
+4
View File
@@ -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)
+62 -4
View File
@@ -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.<agent> 下多走一层 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
+31 -3
View File
@@ -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):