diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 540c082a..545d545f 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -229,7 +229,8 @@ NODE_REFERENCE = """\ **凡是给人看的、带数字的报告、分析、结论，都用 report，不要用 llm**。它自动收集上游的口径卡指标、查询结果 （tool 节点和 agent 查过的库、merge 的合并结果，按查询先后编成 Q1、Q2…）、查库当时的表结构、知识库检索（按先后编成 K1、K2…） 和运行输入，模型只能写引用标记，数值由系统从证据里取出来渲染；模型自己写的数字会被打回重写： - - [[m:指标id]]：口径卡指标。比率、增幅、占比、差值这类派生计算只能先在口径卡里登记再引用 + - [[m:指标id]]：口径卡指标。比率、增幅、占比、差值这类派生计算只能先在口径卡里登记再引用。渲染出来自带单位（口径卡里的 + unit，「8条」），instructions 里不要写成「覆盖 [[m:line_count]] 条产线」这样后面再跟一个单位，写「覆盖 [[m:line_count]] 产线」 - [[v:Q1.r0.列名]]：第 1 次查询结果里第 0 行那一列的原值（行号从 0 数） - [[table:Q1 cols=列a,列b rows=0-4]]：把查询结果的几行几列原样生成表格，每一格都点得开出处 - [[i:输入字段]]：运行输入 @@ -289,6 +290,10 @@ NODE_REFERENCE = """\ 4. 不要用 transform 解析 agent / llm 的文字：口径卡读 cell(nodes.查库节点id, 行, '列名') 或 vars..字段 5. 数据在两个库、要按行对齐时：两个 tool 节点各自聚合到相同的键和粒度 → merge 按键合并 → 口径卡算派生指标 → report。 同一个库写成一条 SQL；只组合几个单值用口径卡的 cell()；不要用 code 节点合并（出处会断，喂给口径卡时核对不了） +6. 问题里的「今天、昨天、上周、本月、去年」这类相对日期一律按运行当天算：SQL 用数据源方言里取当天的写法（CURRENT_DATE、 + CURDATE()、date('now')、SYSDATE），agent 先调 current_time 拿到当天日期再写条件（agent 的 prompt 里写明这一点、tools 绑上 + current_time）。不要用表里最大的日期（MAX(日期)）去推「上周」「本月」：数据没更新到那段时间时，推出来的是另一段，报告还会 + 把它当成上周来写。想让报告说清数据最新到哪天，另取一列 MAX(日期) AS 数据最新日期 模型字段（llm / agent / supervisor / report 的 config.model）： - **不要填**。留空表示跟随当前供应商的默认模型，这几乎总是对的。 @@ -928,9 +933,11 @@ async def generate( async for _op in gen.draft(human): pass if applier.replied is None and gen.asked is None and all(o.get("op") == "plan" for o in applier.seen): - raise CopilotIncomplete("模型没有给出任何修改，也没有说明原因") + raise gen.empty() except Exception as e: # noqa: BLE001 reason, _ = explain_error(e) + if isinstance(e, CopilotIncomplete) and e.detail: + reason = f"{reason}；{e.detail}" raise HTTPException( 502, f"模型未返回可用的工作流（{reason}）。请换一种描述重试；如反复出现，" "请在输入框的「助手使用的模型」中换用指令遵循能力更强的模型", @@ -1765,11 +1772,13 @@ async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None): if gen.asked is None and all(o.get("op") == "plan" for o in applier.seen): # 只写了思考（deepseek 这类推理模型偶发）、或者格式跑偏：带机读码，问数据页和评测按 chat/turn 的规矩自动 # 重来一次（error.next = retry），还不行再照实报 - raise CopilotIncomplete("模型没有给出任何修改，也没有说明原因", - hint="画布未修改。可以重试，或把需求描述得更具体", code=turn.EMPTY_GENERATION) + # 错误详情里记诊断（结束原因、正文和思考的长度、跳过的行和原因、工具调用……，不记原文），重来时交给助手的 + # 原因换成一句纠正（Generation.empty） + raise gen.empty(hint="画布未修改。可以重试，或把需求描述得更具体") except CopilotIncomplete as e: failed = {"op": "error", "message": f"助手本轮未完成：{e}", "hint": e.hint, "detail": e.detail, - "reason": str(e), **({"code": e.code} if e.code else {})} + "reason": e.correction or str(e), **({"code": e.code} if e.code else {}), + **({"diagnostics": e.diagnostics} if e.diagnostics else {})} yield {**failed, "next": turn.next_step(failed)} return except Exception as e: # noqa: BLE001 @@ -2451,8 +2460,7 @@ async def app_stream(payload: AppStreamIn, session: AsyncSession = Depends(get_s return if gen.asked is None and gen.done is None and not applier.effective: # 流正常结束，却没有 done、也没有一个操作落地：不能按「无需修改」交付 - raise CopilotIncomplete("模型没有给出任何修改，也没有说明原因", - hint="应用草稿未修改。可以重试，或把要求描述得更具体") + raise gen.empty(hint="应用草稿未修改。可以重试，或把要求描述得更具体") async for op in _with_heartbeat(gen.repair(), phase="repairing"): yield _sse(op) remaining = await checker.issues() @@ -2804,5 +2812,7 @@ async def review_of_run(session: AsyncSession, run: Any, *, question: str, provi chat_model, question=question, answer=answer, signals=signals, # 报告撰写节点产出的答案逐段对应着证据文档：只能加说明，不能改写 locked=bool(output and output.get("_evidence")), + # 已经处理好的（首稿没过、重写后已全部通过……）：不是缺口，照实告诉复核模型 + settled=rv.notes(rows), ) return done(result) diff --git a/backend/app/api/copilot_blueprint.py b/backend/app/api/copilot_blueprint.py index 515f0ba2..af3a52f5 100644 --- a/backend/app/api/copilot_blueprint.py +++ b/backend/app/api/copilot_blueprint.py @@ -321,8 +321,7 @@ async def blueprint_stream(payload: BlueprintStreamIn, session: AsyncSession = D meta={"looks": looks.records}) return if gen.asked is None and gen.done is None and not applier.effective: - raise copilot.CopilotIncomplete("模型没有给出任何修改，也没有说明原因", - hint="蓝图未修改。可以重试，或把需求描述得更具体") + raise gen.empty(hint="蓝图未修改。可以重试，或把需求描述得更具体") async for op in copilot._with_heartbeat(gen.repair(), phase="repairing", started=started): yield _sse(op) remaining = await checker.issues() if gen.asked is None else [] diff --git a/backend/app/api/copilot_gen.py b/backend/app/api/copilot_gen.py index 465ff222..dcc03d68 100644 --- a/backend/app/api/copilot_gen.py +++ b/backend/app/api/copilot_gen.py @@ -24,9 +24,11 @@ from __future__ import annotations import copy import json +import re import time from collections.abc import AsyncIterator from contextlib import aclosing +from dataclasses import dataclass, field from typing import Any from app.engine.schema import NodeType @@ -53,30 +55,120 @@ DRAIN_CHARS = 400 # ========================================================================== -def _parse_op_line(line: str) -> dict[str, Any] | None: - """解析一行操作。围栏行、空行、解析失败、缺 op 的一律返回 None—— - 模型偶尔犯规不该毁掉整条流，跳过就是。""" +#: 操作行前后的围栏：```json {"op":…} 这样和围栏写在同一行的，去掉围栏再读 +_FENCE_HEAD = re.compile(r"^`{3,}[A-Za-z0-9_-]*\s*") +_FENCE_TAIL = re.compile(r"\s*`{3,}$") +#: 一个操作分成多行写（缩进排版的 JSON）时最多攒多少行、多少字：攒不成一个完整的 JSON 就当跳过 +_PENDING_LINES = 200 +_PENDING_CHARS = 20000 +#: 操作的名字（各入口的都在里面）：原生工具调用叫这些名字的，参数就是操作的字段，照操作读（_tool_ops） +OP_NAMES = frozenset({ + "plan", "add_node", "update_node", "remove_node", "add_edge", "remove_edge", "done", "reply", "look", "ask", + "catalog_patch", "use_caliber", "question", + "app_add", "app_update", "app_remove", "app_move", "app_page", "app_set", + "bp_ask", "bp_set", "bp_add", "bp_remove", "bp_template", "fill_joint", +}) + + +def _op_text(line: str) -> str: + """一行里要按操作读的那段：去掉首尾空白和同一行上的围栏。只有围栏的行（```、```json）交回空串。""" text = line.strip() - if not text or text.startswith("```"): - return None + if text.startswith("```"): + text = _FENCE_HEAD.sub("", text, count=1) + return _FENCE_TAIL.sub("", text).strip() + + +def _classify(text: str) -> tuple[dict[str, Any] | None, str | None]: + """(操作, None) 或 (None, 为什么跳过)。空的交回 (None, None)：不算跳过。""" + if not text: + return None, None try: obj = json.loads(text) except json.JSONDecodeError: - return None - if not isinstance(obj, dict) or "op" not in obj: - return None - return obj + return None, "不是完整的 JSON" + if not isinstance(obj, dict): + return None, "不是 JSON 对象" + if "op" not in obj: + return None, "缺 op" + return obj, None + + +def _parse_op_line(line: str) -> dict[str, Any] | None: + """解析一行操作。围栏行、空行、解析失败、缺 op 的一律返回 None—— + 模型偶尔犯规不该毁掉整条流，跳过就是（跳过的照实记进 StreamDiagnosis）。和围栏写在同一行的操作去掉围栏再读。""" + return _classify(_op_text(line))[0] + + +@dataclass +class StreamDiagnosis: + """一轮生成里模型交回了什么：只记长度、个数和原因，不记原文。一个操作都没认出来（copilot_empty）时写进错误详情，看得出 + 是只写了思考、格式跑偏、还是用了工具调用；自动重来时据此给一句纠正（correction）。 + + finish 是最后一次调用的结束原因（finish_reason / stop_reason）；text_chars、thinking_chars 是正文（不算空白）和思考的 + 字数；dropped 是跳过的行按原因数；tool_calls 是原生工具调用的 + 名字（按操作读了的也记）；markup 是正文里写成文字的工具调用标记（、DeepSeek 的 DSML……）出现了几次调用。""" + + calls: int = 0 + finish: str = "" + text_chars: int = 0 + thinking_chars: int = 0 + ops: int = 0 + tool_calls: list[str] = field(default_factory=list) + dropped: dict[str, int] = field(default_factory=dict) + markup: int = 0 + + def drop(self, reason: str) -> None: + self.dropped[reason] = self.dropped.get(reason, 0) + 1 + + def as_dict(self) -> dict[str, Any]: + return {"calls": self.calls, "finish": self.finish or None, "text_chars": self.text_chars, + "thinking_chars": self.thinking_chars, "ops": self.ops, "tool_calls": list(self.tool_calls), + "dropped": dict(self.dropped), "markup": self.markup} + + def summary(self) -> str: + """给人看的一句：「结束原因 stop；正文 0 字，思考 11 字；认出 0 行操作；……」。""" + parts = [f"结束原因 {self.finish or '未报'}", f"正文 {self.text_chars} 字，思考 {self.thinking_chars} 字", + f"认出 {self.ops} 行操作"] + if self.tool_calls: + parts.append(f"原生工具调用 {len(self.tool_calls)} 个（{'、'.join(self.tool_calls[:5])}）") + if self.dropped: + parts.append(f"跳过 {sum(self.dropped.values())} 行：" + "、".join(f"{n} 行{why}" for why, n in self.dropped.items())) + if self.markup: + parts.append("正文里有写成文字的工具调用标记") + if self.calls > 1: + parts.append(f"共调用模型 {self.calls} 次") + return "；".join(parts) + + def correction(self) -> str: + """自动重来时交给助手的那句纠正（chat/turn.empty_retry_instruction 接在问题后面）。""" + if self.tool_calls: + return (f"用了工具调用接口（{'、'.join(self.tool_calls[:3])}），这里不走工具调用；每个操作写成正文里单独一行的 " + "JSON，要看表就写一行 {\"op\":\"look\",…}") + if self.markup: + return "把操作写成了工具调用的标记，这里不走工具调用；每个操作写成单独一行的 JSON，要看表就写一行 {\"op\":\"look\",…}" + if not self.text_chars and self.thinking_chars: + return "只写了思考、没有给出任何操作；这次请直接输出操作（每行一个 JSON），要看表就先输出一行 look" + if not self.text_chars: + return "什么都没有输出；这次请直接输出操作（每行一个 JSON），要看表就先输出一行 look" + if self.dropped: + return ("输出里没有一行能认出的操作；每个操作单独一行、写成一行完整的 JSON（带 op），不要夹说明文字，" + "要看表就先输出一行 look") + return "没有给出任何操作；请先 look 看表，或者直接给出操作" class CopilotIncomplete(Exception): """模型这一轮没说完：输出被截断，或者一个操作都没给。画布不能按「无需修改」交付。code 是机读码：一个操作都没给的是 - chat/turn.EMPTY_GENERATION（问数据一轮按它自动重来一次），截断的不给。""" + chat/turn.EMPTY_GENERATION（问数据一轮按它自动重来一次），截断的不给。correction 是重来时交给助手的那句纠正（error 的 + reason），diagnostics 是机读的诊断（StreamDiagnosis.as_dict），都只在一个操作都没给时有。""" - def __init__(self, reason: str, *, hint: str = "", detail: str = "", code: str = "") -> None: + def __init__(self, reason: str, *, hint: str = "", detail: str = "", code: str = "", correction: str = "", + diagnostics: dict[str, Any] | None = None) -> None: super().__init__(reason) self.hint = hint or "画布未修改。可以重试，或把需求拆成几步分别提出" self.detail = detail self.code = code + self.correction = correction + self.diagnostics = diagnostics async def _drain(stream: Any, already: int) -> None: @@ -92,7 +184,29 @@ async def _drain(stream: Any, already: int) -> None: return -async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None = None): +def _tool_ops(calls: dict[int, dict[str, str]], diag: StreamDiagnosis) -> list[dict[str, Any]]: + """原生工具调用（流里的 tool_call_chunks，按 index 拼好的 {name, args}）→ 操作：名字是操作名的，参数就是操作的字段。 + 名字不认识、参数解析不了的照实记进诊断——以前整个工具调用一声不响就丢了。""" + out: list[dict[str, Any]] = [] + for index in sorted(calls): + name, args = calls[index].get("name") or "", calls[index].get("args") or "" + diag.tool_calls.append(name or "（没有名字）") + if name not in OP_NAMES: + diag.drop("工具调用的名字不认识") + continue + try: + fields = json.loads(args) if args.strip() else {} + except json.JSONDecodeError: + fields = None + if not isinstance(fields, dict): + diag.drop("工具调用的参数解析不了") + continue + out.append({**{k: v for k, v in fields.items() if k != "op"}, "op": name}) + return out + + +async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None = None, + diag: StreamDiagnosis | None = None): """流式读模型输出，按行切出操作。done、look、ask 之后即停——后面就算有闲话也不要了；look 之后停下也就是 「读到这一行就停止本次输出」。停下之后不马上关流：交出那一行、调用方接着要下一条时先把流读完（_drain，内容丢掉，最多 DRAIN_CHARS 字），流末尾那一段用量才收得到——以前读到这一行就关流，输出 token 一个都拿不到，问数据这一轮的花费多半是下限 @@ -101,35 +215,99 @@ async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None 除了操作，还把模型的思考原样转出去（thinking_text：Claude 的 thinking 块、OpenAI 兼容系的 reasoning_content）， 不把非 JSON 行当成思考。lines 给了的话，认出来的操作行原样记进去：接着写时作为 assistant 消息交回模型。 + 格式跑偏的几种写法照样认（以前一声不响就丢了）：一个操作分成多行写的 JSON（以「{」开头的一行读不成，就接着往后攒， + 攒到读得成为止；中途来了一行自己就读得成的，攒着的那段算跳过）、和围栏写在同一行的操作、原生工具调用里名字是操作名 + 的（_tool_ops，流读完再按先后交出）。diag 给了的话，交回了什么（长度、个数、跳过的原因）记进去，不记原文。 + 没等到 done 就因为额度用完停下的，抛 CopilotIncomplete：截断之前的操作照常转出去，由调用方决定整轮作废。 """ from app.engine.state import message_text as _text from app.engine.state import thinking_text + from app.engine.toolcalls import leaked_markup + diag = diag if diag is not None else StreamDiagnosis() + diag.calls += 1 buffer = "" cut = "" + pending: list[str] = [] + calls: dict[int, dict[str, str]] = {} + head = "" # 正文的开头一段：只用来认写成文字的工具调用标记，不进诊断 + + def read(line: str) -> dict[str, Any] | None: + """一行 → 操作（认出来了）或 None；多行的 JSON 攒在 pending 里。""" + text = _op_text(line) + if pending: + op, _ = _classify(text) + if op is not None: + diag.drop("不是完整的 JSON") # 自己就是一个操作的一行来了：攒着的那段读不成了 + pending.clear() + return _single(op, None) + pending.append(text) + joined = "\n".join(pending) + op, why = _classify(joined) + if op is not None: + pending.clear() + return _single(op, None) + if len(pending) > _PENDING_LINES or len(joined) > _PENDING_CHARS: + diag.drop("不是完整的 JSON") + pending.clear() + return None + op, why = _classify(text) + if why == "不是完整的 JSON" and text.startswith("{"): + pending.append(text) # 可能是分成多行写的一个操作：接着往后攒 + return None + return _single(op, why) + + def _single(op: dict[str, Any] | None, why: str | None) -> dict[str, Any] | None: + if why: + diag.drop(why) + if op is not None: + diag.ops += 1 + return op + + def record(op: dict[str, Any]) -> None: + if lines is not None: + lines.append(json.dumps(op, ensure_ascii=False)) + async with aclosing(model.astream(messages)) as stream: async for chunk in stream: reasoning = thinking_text(chunk) if reasoning: + diag.thinking_chars += len(reasoning) yield {"op": "thinking", "delta": reasoning} + for part in getattr(chunk, "tool_call_chunks", None) or []: + slot = calls.setdefault(int(part.get("index") or 0), {"name": "", "args": ""}) + slot["name"] = slot["name"] or str(part.get("name") or "") + slot["args"] += str(part.get("args") or "") + meta = getattr(chunk, "response_metadata", None) or {} + if meta.get("finish_reason") or meta.get("stop_reason"): + diag.finish = str(meta.get("finish_reason") or meta.get("stop_reason")) - buffer += _text(chunk) + text = _text(chunk) + diag.text_chars += sum(1 for c in text if not c.isspace()) + if len(head) < 4000: + head += text[:4000 - len(head)] + buffer += text while "\n" in buffer: line, buffer = buffer.split("\n", 1) - op = _parse_op_line(line) + op = read(line) if op: - if lines is not None: - lines.append(line.strip()) + record(op) yield op if op.get("op") in _STOP_OPS: await _drain(stream, len(buffer)) return cut = _cut_off(chunk) or cut - op = _parse_op_line(buffer) - if op: - if lines is not None: - lines.append(buffer.strip()) + if leaked_markup(head): + diag.markup += 1 + tail = [op for op in [read(buffer)] if op] + if pending: + diag.drop("不是完整的 JSON") + pending.clear() + tools = _tool_ops(calls, diag) + diag.ops += len(tools) + for op in tail + tools: + record(op) yield op if op.get("op") in _STOP_OPS: return @@ -709,6 +887,15 @@ class Generation: self.repair_error: Exception | None = None #: 这一轮调了几次主模型（含接着写、修正），验收时统计 self.calls = 0 + #: 起草阶段模型交回了什么（只记长度、个数和原因）：一个操作都没认出来时写进错误详情 + self.diagnosis = StreamDiagnosis() + + def empty(self, message: str = "模型没有给出任何修改，也没有说明原因", *, hint: str = "") -> CopilotIncomplete: + """一个操作都没给（copilot_empty）的报错：带诊断（detail、diagnostics）和重来时交给助手的纠正（correction）。""" + from app.chat.turn import EMPTY_GENERATION + + return CopilotIncomplete(message, hint=hint, code=EMPTY_GENERATION, detail=self.diagnosis.summary(), + correction=self.diagnosis.correction(), diagnostics=self.diagnosis.as_dict()) async def draft(self, request: str) -> AsyncIterator[dict[str, Any]]: async for op in self._converse(request, LookBudget(self.look_limit), repair=False): @@ -752,7 +939,8 @@ class Generation: lines: list[str] = [] look: dict[str, Any] | None = None self.calls += 1 - async with aclosing(_iter_ops(self.model, messages, lines=lines)) as ops: + async with aclosing(_iter_ops(self.model, messages, lines=lines, + diag=None if repair else self.diagnosis)) as ops: async for op in ops: kind = op.get("op") if kind == "thinking": diff --git a/backend/app/bundle/verifier.js b/backend/app/bundle/verifier.js index ffca5f15..c760352b 100644 --- a/backend/app/bundle/verifier.js +++ b/backend/app/bundle/verifier.js @@ -615,11 +615,14 @@ /** * render_cell(shown(raw))：快照里的一格。kind 是查询时记下的列类型，display 是报告文档记下的单元格显示规则版本 - * （2 版起 ISO 日期时间的 T 换成空格）。DECIMAL 落成的数字文本按数渲染、不经 float；文字压成一行，竖线换成 ¦ + * （2 版起 ISO 日期时间的 T 换成空格；3 版起标识列的数不分千位）。plain 是引用上记下的「这一格按标识列显示」： + * 哪几列是标识列服务端按 SQL 和冻结的数据目录判（cell_format.plain_columns），本页解析不了 SQL，照引用上记的渲染—— + * 它只决定分不分千位，值本身照样逐位比对。DECIMAL 落成的数字文本按数渲染、不经 float；文字压成一行，竖线换成 ¦ */ - function renderCell(raw, token, conv, kind, display) { - if (display !== 1 && display !== 2) throw new Unshown(T.shownDisplay(display)) + function renderCell(raw, token, conv, kind, display, plain = false) { + if (display !== 1 && display !== 2 && display !== 3) throw new Unshown(T.shownDisplay(display)) if (!PY_NUMERIC || !ISO_DATETIME) throw new Unshown(T.shownOld) + const fmt = display >= 3 && plain === true ? 'plain' : 'thousands' let value = raw if (display >= 2 && typeof raw === 'string') { const m = ISO_DATETIME.exec(pyStrip(raw)) @@ -630,12 +633,12 @@ // Python 把全角、阿拉伯文这些数字也当数（int('1٢٣') 是 123），本页只照 ASCII 算 if (/[^\x00-\x7f]/.test(text)) throw new Unshown(T.shownDigits) if (!/^-?\d+$/.test(text) || text.replace('-', '').length <= INT_DIGITS) { - return renderNumeric(decimalOf(text), '', null, 'thousands', conv) + return renderNumeric(decimalOf(text), '', null, fmt, conv) } } // 对象、列表：服务端按 Python 的 json.dumps 写（键的先后、浮点的写法），本页没有移植 if (isObj(value) || Array.isArray(value)) throw new Unshown(T.shownShape) - return renderNumber(value, token, { conv }).replace(/\|/g, '¦') + return renderNumber(value, token, { conv, fmt }).replace(/\|/g, '¦') } /** @@ -1469,7 +1472,7 @@ if (!sameValue(cite.value, info.truth)) return { ...info, result: FAIL, reason: T.numCellDiffers } const display = Object.prototype.hasOwnProperty.call(doc, 'cell_display') ? doc.cell_display : 1 return checkShown(info, seg, () => renderCell(cell.raw, info.truthToken, convOf(cite), - columnKind(art.json, cell.column), display)) + columnKind(art.json, cell.column), display, own(cite, 'plain') === true)) } /** 引用上记的换算（|万、|pct 这类），没写是 null */ diff --git a/backend/app/chat/drive.py b/backend/app/chat/drive.py index 95b4fe82..d406b27f 100644 --- a/backend/app/chat/drive.py +++ b/backend/app/chat/drive.py @@ -163,7 +163,8 @@ class TurnDriver: async def on_op(self, op: dict[str, Any]) -> None: """生成过程的每一条操作（含思考、心跳），收下之前先交给这里。""" - async def on_retry(self, reason: str, *, kind: str, base_graph: dict[str, Any] | None) -> None: + async def on_retry(self, reason: str, *, kind: str, base_graph: dict[str, Any] | None, + detail: str | None = None) -> None: """要自动重来一次了（kind：empty 生成过程一个操作都没给、blocked 没过检查、review 复核判不可用）。""" def prepare_graph(self, graph: dict[str, Any]) -> dict[str, Any]: @@ -338,11 +339,14 @@ class TurnDriver: # 生成过程一个操作都没给（只写了思考）：按重来一次的规矩带着原因重来（chat/turn.next_step） step = terminal.get("next") or policy.next_step(terminal) if step == policy.NEXT_RETRY and t.attempts + 1 < policy.MAX_ATTEMPTS: + # 原因是生成过程给的那句纠正（只写了思考、格式跑偏、用了工具调用……），附在问题后面重来， + # 不原样重发；诊断（error 的 detail）随重来的事件记下 reason = policy.error_reason(terminal) t.attempts += 1 t.retries.append(reason) - instruction = policy.retry_instruction(self.question, reason, policy.has_nodes(base_graph)) - await self.on_retry(reason, kind="empty", base_graph=base_graph) + instruction = policy.empty_retry_instruction(self.question, reason) + await self.on_retry(reason, kind="empty", base_graph=base_graph, + detail=str(terminal.get("detail") or "") or None) continue t.path, t.reason, t.copilot_failed = "build_failed", str(terminal.get("message") or "助手本轮未完成"), True return t diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py index 438f6fba..140ae93f 100644 --- a/backend/app/chat/job.py +++ b/backend/app/chat/job.py @@ -191,13 +191,14 @@ class ChatDriver(TurnDriver): finally: await self._flush_thinking() - async def on_retry(self, reason: str, *, kind: str, base_graph: dict[str, Any] | None) -> None: + async def on_retry(self, reason: str, *, kind: str, base_graph: dict[str, Any] | None, + detail: str | None = None) -> None: from app.chat import turn as policy # 换下来的那次运行交出去并进 earlier；这一次从「还没有运行」起，收尾时不再把它当这一次的运行记一遍 run, self.last_run = (self.last_run if kind == "review" else None), None await self.execution.retrying(reason, kind=kind, base=policy.has_nodes(base_graph), attempt=self.turn.attempts, - run=run) + run=run, detail=detail) async def run_input(self, session: AsyncSession, graph: dict[str, Any]) -> dict[str, Any]: """入口填问题和这一轮之前的几轮问答（conversations.run_history，同以前页面上的 historyOf）：重试、重新运行较早的 @@ -364,13 +365,16 @@ class ChatTurnExecution(runtime.JobRun): meta={"runId": run.id, "runClass": run.run_class, "runStatus": run.status}) await self.emit("run", {"run_id": run.id, "status": run.status, "run_class": run.run_class}) - async def retrying(self, reason: str, *, kind: str, base: bool, attempt: int, run: Any = None) -> None: - """自动重来一次：换下来的那次运行的用量并进 earlier；这一行回到「还没有运行」。""" + async def retrying(self, reason: str, *, kind: str, base: bool, attempt: int, run: Any = None, + detail: str | None = None) -> None: + """自动重来一次：换下来的那次运行的用量并进 earlier；这一行回到「还没有运行」。detail：生成过程一个操作都没给时的 + 诊断（结束原因、正文和思考的长度、跳过的行……，不含原文），随 turn.retry 事件记下。""" ops: list[tuple[Any, ...]] = [("set", "run", record.usage_total(run.usage) or {})] if run is not None else [] self.run_id = None await self._job(attempt=attempt, run_id=None) usage = await self.save(run_id=None, meta={"runId": None, "runStatus": None}, usage=[*ops, ("retire",)]) - await self.emit("turn.retry", {"kind": kind, "reason": reason, "base": base, "attempt": attempt, "usage": usage}) + await self.emit("turn.retry", {"kind": kind, "reason": reason, "base": base, "attempt": attempt, "usage": usage, + **({"detail": detail} if detail else {})}) # ---------------- 执行 ---------------- diff --git a/backend/app/chat/turn.py b/backend/app/chat/turn.py index ac361b52..f6796d85 100644 --- a/backend/app/chat/turn.py +++ b/backend/app/chat/turn.py @@ -51,6 +51,12 @@ RETRY_WITH_BASE = ("这个工作流上一次运行没有成功：{reason}。\n\n "改完继续回答原来的问题：{question}") #: 没有图可交回（上一次连图都没建出来）时：原问题后面缀上原因，从零再建 RETRY_FROM_SCRATCH = "{question}\n\n（上一次没有成功：{reason}。请针对这个原因调整，重新给出完整的工作流。）" +#: 生成过程一个操作都没给（EMPTY_GENERATION）时自动重来交给助手的话：原问题后面缀上那句纠正（error 的 reason，见 +#: copilot_gen.StreamDiagnosis.correction：只写了思考、格式跑偏、用了工具调用各有一句）。不原样重发——原样重发多半还是 +#: 只写思考（部署上撞过两次都是空的） +RETRY_EMPTY = "{question}\n\n（上一次没有给出可用的操作：{reason}。）" +#: 没有纠正可用（老服务端只给了「模型没有给出任何修改」）时补的那句 +EMPTY_FALLBACK = "这次请直接按输出格式给出操作（每行一个 JSON），要看表就先输出一行 look" #: 用户点「重试本轮」时交给助手的话（以前在问数据页 retryTurn 里拼，一轮搬到服务端之后在这里）：有图时说的是「没有给出可用的 #: 结果」——上一次未必是运行失败（停止了、被驳回、结论不可用都算），原因是那一次的归档说法（chat/record.attempt_reason）； #: 没有图的同 RETRY_FROM_SCRATCH @@ -149,6 +155,14 @@ def retry_instruction(question: str, reason: str, has_base: bool) -> str: return template.format(question=question, reason=reason) +def empty_retry_instruction(question: str, reason: str) -> str: + """生成过程一个操作都没给、自动重来时交给助手的话：问题后面附那句纠正（reason）；纠正里没说怎么做的补 EMPTY_FALLBACK。""" + reason = reason.strip().rstrip("。") + if "look" not in reason: + reason = f"{reason}；{EMPTY_FALLBACK}" if reason else EMPTY_FALLBACK + return RETRY_EMPTY.format(question=question, reason=reason) + + def user_retry_instruction(question: str, reason: str, has_base: bool) -> str: """用户点「重试本轮」时交给助手的话。has_base：有上一次建出的图可以交回去改。""" template = RETRY_USER_WITH_BASE if has_base else RETRY_FROM_SCRATCH @@ -160,7 +174,8 @@ def has_nodes(graph: Any) -> bool: return isinstance(graph, Mapping) and bool(graph.get("nodes")) -__all__ = ["EMPTY_GENERATION", "MAX_ATTEMPTS", "NEXT_ASKED", "NEXT_FAIL", "NEXT_READY", "NEXT_REPLY", "NEXT_RETRY", - "NEXT_RUN", "NEXT_STEPS", "QUESTION_FIELD", "RETRY_ERRORS", "RETRY_FROM_SCRATCH", "RETRY_USER_WITH_BASE", - "RETRY_WITH_BASE", "blocked_reason", "error_reason", "fill_input", "has_nodes", "next_step", "retry_after_review", +__all__ = ["EMPTY_FALLBACK", "EMPTY_GENERATION", "MAX_ATTEMPTS", "NEXT_ASKED", "NEXT_FAIL", "NEXT_READY", + "NEXT_REPLY", "NEXT_RETRY", "NEXT_RUN", "NEXT_STEPS", "QUESTION_FIELD", "RETRY_EMPTY", "RETRY_ERRORS", + "RETRY_FROM_SCRATCH", "RETRY_USER_WITH_BASE", "RETRY_WITH_BASE", "blocked_reason", + "empty_retry_instruction", "error_reason", "fill_input", "has_nodes", "next_step", "retry_after_review", "retry_instruction", "review_reason", "user_retry_instruction"] diff --git a/backend/app/engine/cell_format.py b/backend/app/engine/cell_format.py new file mode 100644 index 00000000..04dc22fc --- /dev/null +++ b/backend/app/engine/cell_format.py @@ -0,0 +1,183 @@ +"""查询结果里哪几列是标识列：报告里显示这几列的数不分千位（evidence.render_cell 的 plain，单元格显示规则 3 版起）。 + +部署上撞过：产线编码这类数值型的编码列（编码 1203）在报告的表格里写成「1,203」，读的人以为是个量。千分位只该给量用。 +按序看三样信号，第一样说得出的算数；都说不出的照旧分千位： + +1. 查询当时冻结的数据目录（查询快照的 schema_artifact 指向的表结构快照里的 catalog）给源列标的度量类型：identifier + （标识）、status（状态）、attribute（属性）不分千位；flow（流量）、stock（存量）、ratio（比率）是量，分千位； +2. 结果列取自的源列名像编码、编号：CODE、ID、NO 整个词或者在结尾（STATION_CODE、order_id、userNo），带「编码」「编号」 + 「代码」的； +3. 结果列是分组键（GROUP BY 里有它，写列、写别名、写序号都算），不是聚合。 + +SUM、COUNT、AVG 这类聚合出来的列是量，分千位，不看后面几样；MIN、MAX 取的是那一列本身的值，按那一列判。SQL 解析不了、 +结果列在 SQL 里找不到（SELECT *、UNION）的，只按结果列名本身和目录里唯一那张表判。 + +只看查询快照（SQL、列名）和它指向的表结构快照（都按内容寻址、不会变）：同一份快照什么时候判都是同一个结果，组装、 +流式渲染、出口复核一致。离线核对页（bundle/verifier.js）没法解析 SQL，照引用上记下的 plain 渲染（见 evidence._resolve_cell）。 +""" +from __future__ import annotations + +import re +from collections.abc import Callable, Mapping +from typing import Any + +from app.data.names import name_key + +#: 目录里不是量的度量类型：这些列的数是编号、码值、属性，不分千位 +PLAIN_MEASURES = frozenset({"identifier", "status", "attribute"}) +#: 目录里是量的度量类型 +COUNTED_MEASURES = frozenset({"flow", "stock", "ratio"}) +#: 列名像编码、编号：CODE / ID / NO 整个词、或者以 _CODE、_ID、_NO 结尾（不分大小写）；驼峰的 stationCode、orderId、userNo； +#: 带「编码」「编号」「代码」的 +_CODE_NAME = re.compile(r"(?i:(?:^|_)(?:code|id|no)$)|[a-z0-9](?:Code|Id|ID|No|NO)$|编码|编号|代码") +#: 解析 SQL 按这几种读法依次试（快照里没记是哪种库）：先通用读法，读不通再换，取第一种读得通的——同一条 SQL 结果固定 +_READINGS = (None, "mysql", "postgres", "oracle", "sqlite") + + +def code_like(name: Any) -> bool: + """列名像编码、编号（_CODE_NAME）。""" + return isinstance(name, str) and bool(_CODE_NAME.search(name.strip())) + + +def plain_columns(snapshot: Mapping[str, Any], schema: Callable[[str], Any] | None = None) -> frozenset[str]: + """查询快照里要按标识列显示（数不分千位）的结果列名（照快照 columns 的写法）。 + + schema 取表结构快照的内容（参数是 schema_artifact，取不到返回 None），用来读冻结的数据目录；不给就不看目录。 + 不抛异常：分析出了意外就当判断不了（照旧分千位）。""" + columns = [str(c) for c in snapshot.get("columns") or []] + if not columns: + return frozenset() + try: + catalog = _frozen_catalog(snapshot, schema) + select = _outer_select(snapshot.get("sql")) + return frozenset(c for c in columns if _plain(c, select, catalog)) + except Exception: # noqa: BLE001 - 判断不了就照旧分千位，不能让报告渲染崩掉 + return frozenset() + + +def _frozen_catalog(snapshot: Mapping[str, Any], schema: Callable[[str], Any] | None) -> dict[str, dict[str, str]]: + """冻结的数据目录里各列的度量类型：{表名的 name_key: {列名的 name_key: 度量类型}}。表名带库名的，最后一段也记一份。""" + artifact = snapshot.get("schema_artifact") + content = schema(artifact) if schema is not None and isinstance(artifact, str) and artifact else None + frozen = content.get("catalog") if isinstance(content, dict) else None + out: dict[str, dict[str, str]] = {} + for table, entry in (frozen or {}).items() if isinstance(frozen, dict) else (): + notes = entry.get("notes") if isinstance(entry, dict) else None + cols = notes.get("columns") if isinstance(notes, dict) else None + measures = {name_key(str(c)): str(item["measure"]["value"]) for c, item in (cols or {}).items() + if isinstance(item, dict) and isinstance(item.get("measure"), dict) + and isinstance(item["measure"].get("value"), str)} + if measures: + for key in {name_key(str(table)), name_key(str(table).rsplit(".", 1)[-1])}: + out.setdefault(key, {}).update(measures) + return out + + +def _outer_select(sql: Any) -> Any: + """最外层的 SELECT（sqlglot 的节点）；解析不了、最外层不是一个 SELECT（UNION 这类）的为 None。""" + from sqlglot import exp + + from app.data.sql_semantics import _parse + + for read in _READINGS: + root, _ = _parse(sql, read) + if root is not None: + return root if isinstance(root, exp.Select) else None + return None + + +def _plain(column: str, select: Any, catalog: dict[str, dict[str, str]]) -> bool: + from sqlglot import exp + + item = _projection(select, column) if select is not None else None + if item is None: + # SELECT *、UNION、解析不了：只看结果列名本身，目录按唯一那张表找 + tables = _tables(select) if select is not None else {} + measure = _measure(catalog, next(iter(tables.values())) if len(tables) == 1 else None, column) + return measure in PLAIN_MEASURES if measure else code_like(column) + value = _unwrap(item.unalias()) + if isinstance(value, exp.AggFunc) and not isinstance(value, (exp.Min, exp.Max, exp.AnyValue)): + return False # SUM / COUNT / AVG……：量 + if isinstance(value, (exp.Min, exp.Max, exp.AnyValue)): + value = _unwrap(value.this) + if isinstance(value, exp.Column): + tables = _tables(select) + table = tables.get(name_key(value.table)) if value.table else \ + (next(iter(tables.values())) if len(tables) == 1 else None) + measure = _measure(catalog, table, value.name) + if measure in PLAIN_MEASURES: + return True + if measure in COUNTED_MEASURES: + return False + if code_like(value.name): + return True + return _grouped(select, item) + + +def _unwrap(node: Any) -> Any: + from sqlglot import exp + + while isinstance(node, (exp.Paren, exp.Cast, exp.TryCast)): + node = node.this + return node + + +def _projection(select: Any, column: str) -> Any: + """结果列 column 对应的那一项选取（按别名或列名，先一字不差、再按 name_key）。找不到、或者对上不止一项的为 None。""" + items = [p for p in select.expressions if p.alias_or_name] + for same in (lambda a: a == column, lambda a: name_key(a) == name_key(column)): + hits = [p for p in items if same(p.alias_or_name)] + if hits: + return hits[0] if len(hits) == 1 else None + return None + + +def _tables(select: Any) -> dict[str, str]: + """这一层 FROM / JOIN 里的实表：{别名或表名的 name_key: 表名（带库名的照写）}。子查询、CTE 不算实表。""" + from sqlglot import exp + + with_ = select.args.get("with_") or select.args.get("with") + ctes = {name_key(c.alias_or_name) for c in with_.expressions} if with_ is not None else set() + sources = [(select.args.get("from_") or select.args.get("from"))] + sources += [j for j in select.args.get("joins") or []] + out: dict[str, str] = {} + for source in sources: + table = getattr(source, "this", None) + if not isinstance(table, exp.Table) or name_key(table.name) in ctes: + continue + name = ".".join(p for p in (table.db, table.name) if p) + out[name_key(table.alias_or_name)] = name + return out + + +def _measure(catalog: dict[str, dict[str, str]], table: str | None, column: str) -> str | None: + if not table or not catalog: + return None + for key in (name_key(table), name_key(table.rsplit(".", 1)[-1])): + if key in catalog: + return catalog[key].get(name_key(column)) + return None + + +def _grouped(select: Any, item: Any) -> bool: + """这一项选取是分组键：GROUP BY 里写了同样的表达式、它的别名，或者它的序号（GROUP BY 1）。""" + from sqlglot import exp + + group = select.args.get("group") + if group is None: + return False + value = _unwrap(item.unalias()) + position = next((k for k, p in enumerate(select.expressions, start=1) if p is item), None) + alias = name_key(item.alias) if item.alias else None + for key in group.expressions: + key = _unwrap(key) + if isinstance(key, exp.Literal) and key.is_int and int(key.this) == position: + return True + if alias and isinstance(key, exp.Column) and not key.table and name_key(key.name) == alias: + return True + if key == value: + return True + return False + + +__all__ = ["COUNTED_MEASURES", "PLAIN_MEASURES", "code_like", "plain_columns"] diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 02f3c2da..66dcea28 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -50,6 +50,7 @@ import unicodedata from collections import Counter from collections.abc import Callable, Iterable from dataclasses import dataclass, field +from datetime import date, timedelta from decimal import ROUND_HALF_UP, Context, Decimal from typing import Any, Literal, TypedDict @@ -60,7 +61,7 @@ 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.issuance import _CN_COUNT_UNIT, _tolerance, extract_numbers, number_allowance from app.engine.labels import option_label DOC_SCHEMA = "agentlab.report/1" @@ -76,10 +77,16 @@ ENTITY_SYNTAXES = frozenset({1, 2}) #: - 1：照快照里的原样显示，ISO 日期时间带 T（2026-06-30T08:21:39）。没有这个字段的老文档按它复核 #: - 2：ISO 日期时间显示成「2026-06-30 08:21:39」（小数秒、时区照留）。只改显示：快照里的值、引用记下的 value、 #: eid、证据下钻都不变 -#: 新组装的文档里真有日期时间被改写时才记 cell_display: 2（参与内容哈希），没有的文档和以前一字不差。复核必须按 +#: - 3：整表标记 rows= 的结束行越界、起始行在范围内时，截到最后一行（以前整张表算解析不了）。截完照样展开成一格格的 +#: 单元格引用、按格复核，所以这一条只影响组装；复核 3 版以前的文档时，没展开的整表标记照老规矩判，不然老文档里那句 +#: 「没有第 N 行」重新解析得了，会被当成文档被改过（render_mismatch）。另外标识列（编码、编号、分组键，见 +#: cell_format.plain_columns）的数不分千位：编码 1203 不写成「1,203」。真有格子因此改了写法的文档记 cell_display: 3， +#: 引用上记 plain: true（离线核对页解析不了 SQL，照它渲染） +#: 新组装的文档里真有格子因为显示规则改了写法时才记 cell_display（参与内容哈希），记的是渲染得出这些格子的最低一版 +#: （只有日期时间被改写的记 2：标识列没有因此改写法的，3 版的整表截行不改格子的写法），没有的文档和以前一字不差。复核必须按 #: 文档生成时的那一版：老文档存的是带 T 的字，拿新规则重新渲染对不上，会被当成文档被改过（render_mismatch） -CELL_DISPLAY = 2 -CELL_DISPLAYS = frozenset({1, 2}) +CELL_DISPLAY = 3 +CELL_DISPLAYS = frozenset({1, 2, 3}) _ISO_DATETIME = re.compile(r"^(\d{4}-\d{2}-\d{2})T(\d{2}:\d{2}(?::\d{2}(?:\.\d+)?)?(?:Z|[+-]\d{2}:?\d{2})?)$") @@ -699,10 +706,11 @@ def render_metric(metric: dict[str, Any], conv: str | None = None) -> str: ) -def render_cell(value: Any, conv: str | None = None, *, kind: str | None = None) -> str: +def render_cell(value: Any, conv: str | None = None, *, kind: str | None = None, plain: bool = False) -> str: """查询快照里的一格渲染成字。 - - 数按千分位，小数位照值本来的样子（最多 10 位、去尾零）；DECIMAL 落盘后的数字文本 + - 数按千分位，小数位照值本来的样子（最多 10 位、去尾零）；plain 时不分组（标识列：编码 1203 不写成「1,203」， + 哪几列是标识列见 cell_format.plain_columns）。DECIMAL 落盘后的数字文本 （"45678.50"，小数位为 0 的 "45678"，规则见 expressions.cell_value）按数算，用 Decimal 渲染，不经 float 丢精度 - kind 是这一列查询时记下的类型：text 列按文本，VARCHAR 的 "2026" 不渲染成「2,026」； @@ -710,16 +718,17 @@ def render_cell(value: Any, conv: str | None = None, *, kind: str | None = None) - NULL 显示「—」：这是快照里的事实（那一格就是空的），不是缺值，更不是 0 - 文本压成一行，竖线换成 ¦：它常常落在表格里，原样的 | 会多切出一格 """ + fmt = "plain" if plain else DEFAULT_FORMAT if isinstance(value, str) and cell_value(value, kind) is not value: try: - return _render_numeric(Decimal(value.strip()), unit="", decimals=None, fmt=DEFAULT_FORMAT, conv=conv) + return _render_numeric(Decimal(value.strip()), unit="", decimals=None, fmt=fmt, conv=conv) except RenderError: raise except (ArithmeticError, ValueError) as e: raise RenderError(f"「{value}」无法按规则显示") from e if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False, default=str) - return render_number(value, fmt=DEFAULT_FORMAT, conv=conv).replace("|", "¦") + return render_number(value, fmt=fmt, conv=conv).replace("|", "¦") # -------------------------------------------------------------------------- @@ -1313,13 +1322,27 @@ class _Snapshots: 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 + #: 这一次解析里的单元格要按哪一版显示规则才渲染得出来（有格子因为某一版的规则改了写法，就至少是那一版）： + #: 组装时据此决定文档记不记、记哪一版 cell_display + self.needed = 1 + #: 每份查询快照里的标识列（cell_format.plain_columns），按快照的工件 id 记 + self.plains: dict[str, frozenset[str]] = {} + + def plain(self, artifact: Any, snapshot: Any) -> frozenset[str]: + """这份查询快照里按标识列显示（数不分千位）的结果列。3 版以前的显示规则没有这一条，一律分千位。""" + if self.display < 3 or not isinstance(snapshot, dict): + return frozenset() + key = str(artifact or "") + if key not in self.plains: + from app.engine.cell_format import plain_columns + + self.plains[key] = plain_columns(snapshot, lambda a: self._fetch(a, "表结构快照")[0]) + return self.plains[key] 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 + self.needed = max(self.needed, 2) return f"{m.group(1)} {m.group(2)}" return raw @@ -1337,6 +1360,7 @@ class _Snapshots: return self twin = _Snapshots(self.loader, entities=self.entities_on, syntax=syntax, display=display) twin.memo = self.memo + twin.plains = self.plains return twin def get(self, alias: str, artifact: Any) -> tuple[Any, str | None]: @@ -1435,7 +1459,14 @@ 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(snaps.shown(raw), conv, kind=kind) + shown = snaps.shown(raw) + rendered = render_cell(shown, conv, kind=kind) + plain = column in snaps.plain(entry.get("artifact"), snapshot) \ + and render_cell(shown, conv, kind=kind, plain=True) != rendered + if plain: + # 标识列（3 版起）：数不分千位。真改了写法才记——文档的 cell_display 记成 3，引用上记 plain，离线核对页照它渲染 + rendered = render_cell(shown, conv, kind=kind, plain=True) + snaps.needed = max(snaps.needed, 3) except CellError as e: return _unresolved(marker, alias, "cell", f"{alias} {e}") except RenderError as e: @@ -1446,6 +1477,8 @@ def _resolve_cell(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snaps "rendered": rendered} if conv: cite["conv"] = conv + if plain: + cite["plain"] = True return cite @@ -1508,12 +1541,15 @@ def _table_plan(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Snapsho if last < first: return fail(f"整表引用的行号范围格式有误，当前为「{options['rows']}」", f"rows 要写成 0-4 这样从小到大的行号范围（从 0 数），写的是 {options['rows']}") + if first >= len(rows) or (last >= len(rows) and snaps.display < 3): + # 3 版以前结束行越界也算解析不了：复核老文档里没展开的整表标记照老规矩判（见 CELL_DISPLAY） + return fail(f"{alias} 只有 {len(rows)} 行，没有第 {first if first >= len(rows) else last} 行（行号从 0 开始）") + # 结束行写过了头（写作者以为有 5 行、其实只有 3 行）：起始行在范围内，就截到最后一行，不加注——表格照实只有这几行 + last = min(last, len(rows) - 1) if last - first + 1 > TABLE_MAX_ROWS: return fail(f"整表最多 {TABLE_MAX_ROWS} 行，当前选了 {last - first + 1} 行", f"整表最多 {TABLE_MAX_ROWS} 行，rows={options['rows']} 是 {last - first + 1} 行：" "挑出要的行，或者分成几张表") - if last >= len(rows): - return fail(f"{alias} 只有 {len(rows)} 行，没有第 {last} 行（行号从 0 开始）") else: first, last = 0, min(TABLE_DEFAULT_ROWS, len(rows)) - 1 for c in wanted: @@ -1683,6 +1719,41 @@ def _resolve_entity(marker: dict[str, Any], catalog: dict[str, Any], snaps: _Sna "kind": ev_kind, "role": "entity", "status": "resolved", "rendered": ref} +#: 字段标记前面这几种连接：「产量表的」「产量表中的」「产量表里的」，或者只隔空白（_short_column） +_SHORT_JOINERS = ("中的", "里的", "的", "") +#: 流式渲染时为 _short_column 留着的已渲染文字有多长：表名再长也够看 +SHORT_LOOKBACK = 160 + + +def _short_column(before: str, ref: str) -> str | None: + """[[c:表.列]] 前面紧挨着（隔空白、「的」「中的」「里的」）刚写过同一张表的名字时，正文里只显示列名：交回列名，不用缩短的 + 交回 None。部署上撞过「取自 某表 的 某表.某列」。 + + before 是标记前面已经渲染出来的正文（[[t:表]] 渲染出来就是表名本身，写成文字的表名也算）。表名按 name_key 比（只对 + ASCII 不分大小写），带库名的写法（db.表.列）全名、末段都认；英文名前面还连着字母、数字、下划线的不算（myorders 不是 + orders）。粗体、斜体收尾的星号、下划线跳过。""" + table, dot, column = ref.strip().rpartition(".") + if not dot or not table or not column: + return None + tail = before.rstrip(_BLANK) + for joiner in _SHORT_JOINERS: + if joiner and not tail.endswith(joiner): + continue + head = tail[:len(tail) - len(joiner)].rstrip(_BLANK) if joiner else tail + if not joiner and head == before: + continue # 什么都没隔：「产量表产量表.工厂」不是这种写法，不动 + head = head.rstrip("*_") + for name in dict.fromkeys((table, table.rsplit(".", 1)[-1])): + start = len(head) - len(name) + if start < 0 or name_key(head[start:]) != name_key(name): + continue + if start > 0 and name[0].isascii() and (head[start - 1].isascii() and (head[start - 1].isalnum() + or head[start - 1] == "_")): + continue + return column + return None + + # -------------------------------------------------------------------------- # 引文：[[q:K1|原话]] 要在检索快照的命中片段里逐字出现 # -------------------------------------------------------------------------- @@ -1833,6 +1904,10 @@ _UNIT_GAP = 2 _UNIT_MAX = 10 UNIT_LOOKAHEAD = _UNIT_GAP + _UNIT_MAX + 1 _UNIT_KINDS = frozenset({"m", "i", "v"}) +#: 以一个字的单位打头的更长的单位：口径卡是「人」、正文紧跟着写「人次」，是另一个单位，不当重复的去掉（_tidy） +_LONGER_UNIT = re.compile( + r"人次|人天|人日|人月|人年|人份|件次|台次|台套|台时|车次|架次|班次|户次|批次|个月|个季度|个工作日|个百分点" + r"|小时|分钟|吨公里|公里|公斤|千克|千米|千瓦|万元|亿元|美元|港元|欧元|日元|元/") #: 渲染结果末尾的单位：最后一个数字后面的字（12,345人次 → 人次，4.57万元 → 万元，12.5% → %） _UNIT_TAIL = re.compile(r"\d([^\d\s][^\d]*)$") _BLANK = " \t\u3000" @@ -1888,8 +1963,9 @@ def _tidy(text: str, catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: """渲染前收拾两处排版，数和引用都不动： - 数字标记后面重复写的单位去掉。[[m:visits]] 渲染出来已经带单位（12,345人次），写作者常常 - 还跟着写一遍（[[m:visits]] 人次），读者看到的是「12,345人次 人次」。单字的单位后面紧跟着字时 - 不动：写作者写的可能是另一个单位（口径卡是「人」，正文写「人次」），硬删会把不一致藏起来； + 还跟着写一遍（[[m:visits]] 人次），读者看到的是「12,345人次 人次」。单字的单位后面紧跟着字、和它 + 连起来是一个更长的单位时（_LONGER_UNIT）不动：写作者写的是另一个单位（口径卡是「人」，正文写「人次」）， + 硬删会把不一致藏起来；连起来不是单位的（「覆盖 [[m:lines]] 条产线」，口径卡的单位是「条」）照样去掉； - [[see:…]] 渲染成空，它周围留下的空白、括号、标点去掉（_voids、_see_cuts）：「最高月份 [[see:Q2]]。」会剩下 「最高月份 。」，「结果一致（[[see:Q3]]，原报告…）」会剩下「结果一致（，原报告…）」。依据（连同去掉的字）两边 挨上的字会读成一串的，补一个空格（_glue）：「4,919.5[[see:Q1]]212 笔」会读成「4,919.5212 笔」。 @@ -1906,7 +1982,8 @@ def _tidy(text: str, catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: hit = re.compile(f"[{_BLANK}]{{0,{_UNIT_GAP}}}" + re.escape(unit)).match(text, marker["end"]) if hit: after = text[hit.end():hit.end() + 1] - if len(unit) > 1 or not after.isalnum(): + longer = _LONGER_UNIT.match(text, hit.end() - len(unit)) + if len(unit) > 1 or not after.isalnum() or not (longer and len(longer.group(0)) > len(unit)): cuts.append((marker["end"], hit.end())) see: list[tuple[int, int]] = [] spaces: set[int] = set() @@ -2112,14 +2189,19 @@ def _wordish(ch: str) -> bool: return ch.isdecimal() or (ch.isascii() and ch.isalnum()) -def _render(text: str, catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, prev: str = "") -> str: +def _render(text: str, catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, prev: str = "", + before: str = "") -> str: + """before：这段文字前面已经渲染出来的正文（流式渲染分块时用），字段标记要看前面刚写过的表名（_short_column）。""" text = _tidy(_expand_tables(text, catalog, snaps, cells_allowed, prev), catalog, snaps, cells_allowed, prev) out: list[str] = [] last = 0 for marker in parse_markers(text): out.append(text[last:marker["start"]]) if marker["kind"] != "see": - out.append(resolve_marker(marker, catalog, cells_allowed=cells_allowed, loader=snaps)["rendered"]) + cite = resolve_marker(marker, catalog, cells_allowed=cells_allowed, loader=snaps) + short = _short_column(before + "".join(out), cite["ref"]) \ + if marker["kind"] == "c" and cite["status"] == "resolved" else None + out.append(short or cite["rendered"]) last = marker["end"] out.append(text[last:]) return "".join(out) @@ -2205,6 +2287,8 @@ class StreamRenderer: self.snaps = _Snapshots(loader) self.pending = "" self.prev = "" + #: 已经放出去的正文的末尾一截：字段标记要看前面刚写过的表名（_short_column） + self.shown = "" def _has_unit(self, marker: dict[str, Any]) -> bool: return bool(_rendered_unit(marker, self.catalog, self.snaps, self.cells_allowed)) @@ -2212,8 +2296,9 @@ class StreamRenderer: def _release(self, ready: str) -> str: if not ready: return "" - out = _render(ready, self.catalog, self.snaps, self.cells_allowed, self.prev) + out = _render(ready, self.catalog, self.snaps, self.cells_allowed, self.prev, self.shown) self.prev = ready[-1] + self.shown = (self.shown + out)[-SHORT_LOOKBACK:] return out def feed(self, delta: str) -> str: @@ -2826,6 +2911,9 @@ def compose_doc( continue cite = _stored(resolve_marker(marker, catalog, cells_allowed=cells_allowed, loader=snaps)) ok = cite["status"] == "resolved" + if ok and marker["kind"] == "c" and (short := _short_column("".join(parts), cite["ref"])): + # 紧跟在同一张表名后面：只显示列名，引用上记 short，复核照它核对（老文档没有这一项，照全名） + cite = {**cite, "rendered": short, "short": True} seg = { "kind": _SEG_KIND.get(marker["kind"]) or ("number" if marker["kind"] == "m" or (ok and is_number(cite.get("value"))) @@ -2863,7 +2951,8 @@ def compose_doc( unit["segments"] = _link_entities(unit["segments"], catalog, index) all_segments = [s for block in out_blocks for unit in block["units"] for s in unit["segments"]] # 裸数字单独切成片段：前端要在那几个字底下画「无出处」 - refs = [tuple(s["span"]) for s in all_segments if s.get("ref")] + _header_masks(out_blocks, catalog) + refs = [tuple(s["span"]) for s in all_segments if s.get("ref")] + _header_masks(out_blocks, catalog) \ + + _name_masks(markdown, catalog, snaps) syntax = [tuple(s["span"]) for s in all_segments if s["kind"] == "structural"] bare = _bare_numbers(markdown, refs, allow_numbers, syntax) for block in out_blocks: @@ -2904,10 +2993,11 @@ def compose_doc( # entity_syntax：这份文档按哪一版实体语法组装，复核照它走（verify_doc）。在文档顶层，参与内容哈希 # provenance：期 4 的文档标记（P4-SPEC 1.4）。带它的文档才有查询步骤的提示字段、推断来源、裁判摘录的新行； # 没有它的（期 4 之前组装的）一律按原样展示。写在顶层、参与内容哈希，事后补不上：补了哈希就变，封存链就断 - # cell_display：有单元格因为显示规则改了写法（ISO 日期时间）才记，复核照它渲染；没有的文档和以前一字不差 + # 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 {}), + **({"cell_display": snaps.needed} if snaps.needed > 1 else {}), "run_id": run_id, "node_id": node_id, "markdown": markdown, "source": source, "catalog": kept, "blocks": out_blocks, } @@ -2943,6 +3033,56 @@ def _header_masks(blocks: list[dict[str, Any]], catalog: dict[str, Any]) -> list return out +#: 文本格的值本身是数、或者是带单位的数（「120台」「3.5万元」「45%」「约 30 天」「205L」）：不算名字（_name_masks），照抄进 +#: 正文的那个数照样要出处。单位认出具校验数量词那一套（issuance._CN_COUNT_UNIT），再加百分号、万亿千百和 1–3 个拉丁字母 +_UNIT_VALUE = re.compile( + r"^(?:约|共|近|超|逾|合计|总计|≈|~)?\s*[+-]?(?:\d[\d,]*(?:\.\d+)?|\.\d+)\s*" + r"(?:[%％‰]|[万亿千百]?(?:" + _CN_COUNT_UNIT.pattern + r"|[A-Za-z]{1,3}))?\s*$") + + +def _text_names(catalog: dict[str, Any], snaps: _Snapshots) -> set[str]: + """目录里每次查询的快照中，文本格里带数字的值（「冲压5线」「KX-205 线」）：至少有一个数字、至少有一个不是数字的字， + 本身不是数、也不是带单位的数（_UNIT_VALUE）。按写作目录里显示的样子（压成一行）收。取不回的快照跳过。""" + names: set[str] = set() + for alias, entry in catalog.items(): + if not isinstance(entry, dict) or entry.get("kind") != "query": + continue + snapshot, why = snaps.get(alias, entry.get("artifact")) + if why or not isinstance(snapshot, dict): + continue + columns = [str(c) for c in snapshot.get("columns") or []] + kinds = [column_kind(snapshot, c) for c in columns] + for row in snapshot.get("rows") or []: + cells = row if isinstance(row, list) else [row.get(c) for c in columns] if isinstance(row, dict) else [] + for raw, kind in zip(cells, kinds): + if not isinstance(raw, str) or cell_value(raw, kind) is not raw: + continue + text = _clean(raw) + if any(c.isdigit() for c in text) and not text.isdigit() and not _UNIT_VALUE.match(text): + names.add(text) + return names + + +def _name_masks(markdown: str, catalog: dict[str, Any], snaps: _Snapshots) -> list[tuple[int, int]]: + """正文里一字不差地写出了查询结果中文本格的值（_text_names）的那几段：名字里的数字不是结论数字，写作者照着查询结果 + 写名字，不该因为名字里有个 5 被打回重写。只认快照里真有的值、整段一字不差，前后紧挨着数字的不算（「冲压15线」不是 + 「冲压5线」）；同一个数单独写出来照样要出处。组装和复核用同一份目录，放过的范围一致。""" + out: list[tuple[int, int]] = [] + runs = set(re.findall(r"\d+", markdown)) + for name in _text_names(catalog, snaps) if runs else (): + # 名字一字不差地出现、前后不紧挨着数字，名字里的每一串数字在正文里就是完整的一串：先按它挡掉大多数名字 + if not all(r in runs for r in re.findall(r"\d+", name)): + continue + at = markdown.find(name) + while at >= 0: + end = at + len(name) + before, after = markdown[at - 1:at], markdown[end:end + 1] + if not (before.isdigit() or before == "." or after.isdigit()): + out.append((at, end)) + at = markdown.find(name, at + 1) + return out + + def _split_bare(segments: list[dict[str, Any]], bare: list[Any]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] for seg in segments: @@ -3127,6 +3267,10 @@ def verify_doc( cite = _phase_two_cite(seg, ref, text) \ or resolve_ref(ref, catalog, cells_allowed=cells_allowed, loader=snaps) ok = cite["status"] == "resolved" + if ok and ref.startswith("c:") and (seg.get("cite") or {}).get("short") is True \ + and "." in str(cite.get("rendered") or ""): + # 组装时紧跟在同一张表名后面、只显示了列名（_short_column）：照列名核对 + cite = {**cite, "rendered": str(cite["rendered"]).rsplit(".", 1)[-1]} # 反引号里的实体：片段的字带着两个反引号 shown = f"`{cite['rendered']}`" if seg.get("code") and seg.get("kind") == "entity" else cite["rendered"] matches = shown == text @@ -3209,7 +3353,7 @@ def verify_doc( if tiled and pos != len(markdown): violations.append(_violation("segment_mismatch", "正文末尾有一段不属于任何片段，文档可能被修改过")) - masks += _header_masks(list(doc.get("blocks") or []), catalog) + masks += _header_masks(list(doc.get("blocks") or []), catalog) + _name_masks(markdown, catalog, snaps) for token in _bare_numbers(markdown, masks, allow_numbers, syntax): owner = next(((sid, uid) for s, e, sid, uid in owners if s <= token.start < e), (None, None)) violations.append(_violation( @@ -3306,6 +3450,29 @@ SEE_AND_CODE_RULES = ( "（「状态为 9」「类型 1」）：码值是没有出处的数字，系统会当成裸数字打回重写。" ) +#: [[m:]]、[[v:]] 渲染出来自带单位（部署上撞过「覆盖 8条 条产线」：口径卡的单位是「条」，写作者照着助手的指令又写了一个）。 +#: 渲染时紧跟着的同一个单位会去掉（_tidy），这里先说给写作者听 +UNIT_RULE = ("数字标记渲染出来自带单位（[[m:…]] 是口径卡里的单位，比如「8条」「12,345人次」），后面不要再写一遍单位：" + "写「覆盖 [[m:line_count]] 产线」，不写「覆盖 [[m:line_count]] 条产线」。") + +#: 相对日期（「今天、昨天、上周、本月、去年」）一律按运行当天算（period_rule）。部署上撞过：数据只更新到两周前，查库时拿表里 +#: 最大的日期往前推七天当「上周」，写作者照着写成「上周产量……」。写作者不知道今天是哪天，就分不出数据有没有覆盖问的那段， +#: 所以把运行当天和上周的起止日期直接告诉它：没覆盖时先照实说、写明数据最新到哪天；最近一段的数可以给，但写具体起止日期 +PERIOD_RULE = ( + "相对日期：今天是 {today}（{weekday}），上周是 {week_start} 至 {week_end}。问题里的「今天、昨天、上周、本月、去年」" + "这类说法一律按今天算，不按数据里最新的日期推。查到的数据没覆盖问的那段时间时，先照实说没覆盖、数据最新到哪天" + "（写具体日期）；可以接着给出数据里最近一段的数，但写明它的起止日期，不要把它叫作「上周」「本月」这类相对的说法。" +) +_WEEKDAYS = ("周一", "周二", "周三", "周四", "周五", "周六", "周日") + + +def period_rule(today: date) -> str: + """写作规则里讲相对日期的那一条（PERIOD_RULE），按运行当天 today 填好：上周是上周一到上周日。""" + monday = today - timedelta(days=today.weekday() + 7) + return PERIOD_RULE.format(today=today.isoformat(), weekday=_WEEKDAYS[today.weekday()], + week_start=monday.isoformat(), week_end=(monday + timedelta(days=6)).isoformat()) + + #: 句中的 [[see:]]：前面是介词、后面接着句子的写法（「以 [[see:Q1]] 为依据」「根据 [[see:Q1]]，」「详见 [[see:Q1]]」）。 #: 只认这几种拿得准的：「以」前面是「所、可、加……」的是另一个词（所以、可以），不算；「依据 [[see:…]]。」可能是 #: 「这是结论的依据」，不算。认出来就交回写作者重写（report.REWRITE_ONLY：重写不好也不判失败） @@ -3496,7 +3663,8 @@ ENTITY_RULES = ( "表和字段（系统会核对）：提到表名、字段名时写 [[t:表名]]、[[c:表名.字段名]]；英文的名字也可以放进反引号" "（`orders`），中文的表名、字段名一律用 [[t:]] / [[c:]]——反引号里的中文名系统不核对；" "只写下面列出的、或者这次查过的数据源里真实存在的名字——本次运行的表结构快照、查询用到的表、" - "查询结果列里都没有的名字，会被标成「可能是编造的名字」。\n" + "查询结果列里都没有的名字，会被标成「可能是编造的名字」。" + "紧跟在同一张表名后面的字段（「[[t:orders]] 的 [[c:orders.amount]]」）只显示字段名，不用为了不重复把字段改写成别的说法。\n" "说明一个数是怎么算出来的（按哪个字段去重、怎么汇总、筛了什么条件）时，句末挂 [[see:Q1]] 指向算出它的那次" "查询（下面每张表后面写了哪几次查询用到它）：只挂 [[t:]]、[[c:]] 指不出是哪一次查询算出的这个数。" ) @@ -3561,6 +3729,7 @@ def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, return lines columns = [str(c) for c in snapshot.get("columns") or []][:_PROMPT_COLS] rows = snapshot["rows"] + plain = snaps.plain(entry.get("artifact"), snapshot) example = None for r, record in enumerate(rows[:shown_rows]): cells = [] @@ -3568,7 +3737,8 @@ def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, try: raw, _ = locate_cell(snapshot, r, c) kind = column_kind(snapshot, c) - shown, numeric = render_cell(snaps.shown(raw), kind=kind), is_number(cell_value(raw, kind)) + shown = render_cell(snaps.shown(raw), kind=kind, plain=c in plain) + numeric = is_number(cell_value(raw, kind)) except (CellError, RenderError): shown, numeric = MISSING, False cells.append(f"{c}={shown}") @@ -3723,7 +3893,8 @@ def _field_prompt(path: str, value: Any, ev: dict[str, Any], aliases: dict[str, if why: raise CellError(why) raw, column = locate_cell(snapshot, row, column) - shown = render_cell(snaps.shown(raw), kind=column_kind(snapshot, column)) + shown = render_cell(snaps.shown(raw), kind=column_kind(snapshot, column), + plain=column in snaps.plain(ev.get("artifact"), snapshot)) except (CellError, RenderError): try: shown = render_cell(value) diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index ca4fff39..00ab5795 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -37,7 +37,8 @@ import asyncio import re import time from collections import Counter -from collections.abc import Iterable +from collections.abc import Iterable, Mapping +from datetime import date, datetime from typing import Any from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -56,6 +57,7 @@ from app.engine.evidence import ( ENTITY_KINDS, MARKER_RULES, SEE_AND_CODE_RULES, + UNIT_RULE, StreamRenderer, WriterCatalog, build_catalog, @@ -63,6 +65,7 @@ from app.engine.evidence import ( compose_doc, describe_violations, ledger_enabled, + period_rule, withhold_metrics, writer_catalog, ) @@ -234,13 +237,18 @@ def failed_queries(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[ 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 - for call in result.get("tool_calls") or []: - if not isinstance(call, dict) or call.get("repeat") or call.get("skipped") or call.get("denied"): - continue + calls = [c for c in result.get("tool_calls") or [] + if isinstance(c, dict) and not (c.get("repeat") or c.get("skipped") or c.get("denied")) + and is_platform_query(str(c.get("tool") or ""))] + for i, call in enumerate(calls): name, said = str(call.get("tool") or ""), str(call.get("result") or "").strip() - if not is_platform_query(name) or not (call.get("ok") is False or said.startswith("查询失败")): + if not _query_failed(call): continue args = call.get("args") if isinstance(call.get("args"), dict) else {} + if _superseded(name, str(args.get("sql") or ""), calls[i + 1:]): + # 写错了一条、下一条马上改对了：这次失败已被替代，不算「没有查到」（不然写作模型会写「还有一次取数失败」， + # 复核据此判有缺口） + continue reason = said.splitlines()[0] if said else "没有交回结果" found.append({"node_id": other.id, "title": other.title or other.id, "tool": name, "sql": " ".join(str(args.get("sql") or "").split())[:200], @@ -248,6 +256,37 @@ def failed_queries(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[ return found +def _query_failed(call: Mapping[str, Any]) -> bool: + said = str(call.get("result") or "").strip() + return call.get("ok") is False or said.startswith("查询失败") + + +def _tables(sql: str) -> set[str] | None: + """这条 SQL 读的表（sql_semantics.facets 的 tables，按比较用的写法）；解析不了的为 None。""" + from app.data.sql_semantics import facets + + f = facets(sql) + if f.parse_error or not f.tables: + return None + return {str(p.key) for p in f.tables} + + +def _superseded(tool: str, sql: str, later: list[dict[str, Any]]) -> bool: + """同一个节点里这次失败之后，还有一条成功的平台查询替代了它：同一个数据源（同一个查询工具）、读的表覆盖了失败那条读的 + 表；失败那条解析不了（取不出表）的，同一个数据源之后有成功的查询就算。""" + wanted = _tables(sql) + for call in later: + if str(call.get("tool") or "") != tool or _query_failed(call): + continue + if wanted is None: + return True + args = call.get("args") if isinstance(call.get("args"), dict) else {} + got = _tables(str(args.get("sql") or "")) + if got is not None and wanted <= got: + return True + return False + + def _failed_section(failed: list[dict[str, Any]], *, any_query: bool) -> str: """写作提示里「没有查到的数据」一节，和照实说没查到的要求。一次成功的查询都没有时，不许写单元格引用。""" lines = [f"- 「{f['title']}」{f['tool']}" + (f"：{f['sql']}" if f["sql"] else "") + f"（{f['reason']}）" @@ -598,12 +637,17 @@ def _repair_request(violations: list[dict[str, Any]], failed_rule: str | None = ) +def run_today() -> date: + """写作规则里的「今天」（period_rule）：服务端所在时区的日期，和查库时 CURRENT_DATE 一类按运行当天现算的写法同一天。""" + return datetime.now().astimezone().date() + + def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims: str = "off", failed: str | None = None) -> 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}" + rules = f"{MARKER_RULES}\n{SEE_AND_CODE_RULES}\n{UNIT_RULE}\n{period_rule(run_today())}" if writer.guarded: # 写作者看不全证据（目录有省略、有来源的结构化结论为空）：「没有查到」这类结论没有依据 rules = f"{rules}\n{ABSENCE_RULE}" @@ -791,7 +835,7 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: 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", + ctx.emit(EventType.LOG, level="warn", code="report_repair", count=len(blocking), attempt=repairs, message=f"报告有 {len(blocking)} 处未通过核对（{_named(blocking)}），" f"已要求模型重写（第 {repairs} 次）") messages = [*messages, response, HumanMessage(content=_repair_request(blocking, failed_rule))] @@ -979,7 +1023,7 @@ async def _judge_inline(ctx: NodeContext, doc: dict[str, Any], catalog: dict[str if judge_cfg["rewrite_once"] and bad: texts = judging.unit_texts(doc) sentences = [(texts.get(uid, ""), _rewrite_reason(outcome["verdicts"][uid])) for uid in bad] - ctx.emit(EventType.LOG, level="warn", code="report_rewrite", + ctx.emit(EventType.LOG, level="warn", code="report_rewrite", count=len(bad), message=f"裁判认为 {len(bad)} 句结论证据不支持（{_named([{'text': t} for t, _ in sentences])}），" "已退回写作模型仅修改这几句") again, raw = await draft([*messages, HumanMessage(content=_rewrite_request(sentences))]) diff --git a/backend/app/engine/review.py b/backend/app/engine/review.py index 26b10939..308d06db 100644 --- a/backend/app/engine/review.py +++ b/backend/app/engine/review.py @@ -206,6 +206,8 @@ def scan(events: Iterable[Any], output: Mapping[str, Any] | None = None) -> list elif etype == EventType.TOOL_END and tool and tool in failed_at: recovered.add(tool) + #: 每个报告节点最后一次 report.checked：「改过一次」的信号按最终结果定档（_settled） + finals = _final_checks(rows) #: 报过的节点失败 (节点 id, 原话)：子工作流节点跟着子图里那一步失败时认得出是同一件事 failures: list[tuple[str, str]] = [] for e in rows: @@ -257,6 +259,10 @@ def scan(events: Iterable[Any], output: Mapping[str, Any] | None = None) -> list message = str(data.get("message") or "").strip() code = str(data.get("code") or "") kind, severity = WARN_CODES.get(code, ("warn", DEGRADED)) + if code in _SETTLED_BY_REPORT and _settled(code, finals.get(str(node_id))): + # 首稿没过、改过一次之后最终通过了（同一个报告节点最后一次 report.checked）：不是缺口，不进异常清单。 + # 复核模型被别的信号叫起来时，在 notes() 里照实说「重写后已全部通过」 + continue reason = data.get("reason") if message: add(kind, message[:300], severity, reason if isinstance(reason, str) and reason else None) @@ -282,6 +288,68 @@ def _step_name(node_id: str, titles: Mapping[str, str]) -> str: return " › ".join(titles.get("/".join(parts[:i + 1])) or parts[i] for i in range(len(parts))) +#: 报告节点「改过一次」的信号：按同一个节点最后一次 report.checked 的结果定档（最终通过的不是缺口） +_SETTLED_BY_REPORT = ("report_repair", "report_rewrite") +#: 证据不支持的几种判定（engine/judge 的 CONTRADICTED 加证据不足）：最终还剩这些，改写就没有改好 +_UNSUPPORTED = ("contradicted", "unsupported", "insufficient") + + +def _final_checks(rows: list[Any]) -> dict[str, dict[str, Any]]: + """每个报告节点最后一次 report.checked 的内容（节点 id → data）。""" + out: dict[str, dict[str, Any]] = {} + for e in rows: + if str(_event_field(e, "type", "")) == EventType.REPORT_CHECKED: + data = _event_field(e, "data") or {} + out[str(_event_field(e, "node_id") or "")] = data if isinstance(data, dict) else {} + return out + + +def _settled(code: str, final: Mapping[str, Any] | None) -> bool: + """「改过一次」之后最终是不是好了：report_repair 看最后一次核对有没有违规（ok、violations 为空）；report_rewrite + 看最后的裁判摘要里还有没有证据不支持的句子。找不到最终结果（老运行、节点失败没写出 report.checked）的当没好。""" + if not final: + return False + if code == "report_repair": + return final.get("ok") is True and not final.get("violations") + counts = ((final.get("judge") or {}).get("counts") or {}) if isinstance(final.get("judge"), dict) else {} + return bool(counts) and not any(int(counts.get(k) or 0) for k in _UNSUPPORTED) + + +def notes(events: Iterable[Any]) -> list[str]: + """不是异常、但复核模型被别的信号叫起来时该知道的事（review_prompt 的「已经处理好的」）：报告首稿没过核对、重写后 + 已全部通过；裁判认为证据不支持的句子改写后都有了依据。只说处理好了的，没好的照旧是信号（scan）。""" + rows = list(events) + finals = _final_checks(rows) + titles: dict[str, str] = {} + first: dict[tuple[str, str], dict[str, Any]] = {} + times: dict[tuple[str, str], int] = {} + for e in rows: + etype = str(_event_field(e, "type", "")) + data = _event_field(e, "data") or {} + node_id = str(_event_field(e, "node_id") or "") + if etype == EventType.NODE_STARTED and data.get("label"): + titles.setdefault(node_id, str(data["label"])) + if etype == EventType.LOG and data.get("code") in _SETTLED_BY_REPORT: + key = (node_id, str(data["code"])) + first.setdefault(key, data) + times[key] = times.get(key, 0) + 1 + out: list[str] = [] + for (node_id, code), data in first.items(): + if not _settled(code, finals.get(node_id)): + continue + name = f"报告「{titles.get(node_id) or node_id}」" + count = data.get("count") + if code == "report_repair": + head = f"首稿有 {count} 处没过核对" if isinstance(count, int) and count > 0 else "首稿没过核对" + again = f"重写 {times[(node_id, code)]} 次后" if times[(node_id, code)] > 1 else "重写后" + out.append(f"{name}{head}，{again}已全部通过") + else: + head = f"有 {count} 句结论裁判认为证据不支持" if isinstance(count, int) and count > 0 \ + else "有几句结论裁判认为证据不支持" + out.append(f"{name}{head}，改写后都有了依据") + return out + + def worst(signals: Iterable[Signal]) -> str: """一组信号里最重的那档。没有信号时返回空串。""" ranked = sorted(signals, key=lambda s: _SEVERITY_RANK.get(s.severity, 1), reverse=True) @@ -347,19 +415,23 @@ LOCKED_NOTE = ( ) -def review_prompt(question: str, answer: str, signals: list[Signal], *, locked: bool = False) -> str: +def review_prompt(question: str, answer: str, signals: list[Signal], *, locked: bool = False, + settled: Iterable[str] = ()) -> str: """拼给复核模型的请求。 只给信号清单，不给原始事件流：几百条事件塞进 prompt 既贵又没用，而信号 - 已经是规则层提炼过的结论。 + 已经是规则层提炼过的结论。settled 是已经处理好的（notes()：首稿没过、重写后已全部通过……）： + 照实告诉复核模型，免得它把「改过一次」写成「仍未全部通过、可能不准」。 """ lines = "\n".join( f"- [{'严重' if s.severity == BROKEN else '降级'}] {s.detail}" for s in signals ) + done = "\n".join(f"- {n}" for n in settled) return ( f"用户的问题：\n{question or '（未记录）'}\n\n" f"系统给出的原答案：\n{answer or '（空）'}\n\n" f"检测到的异常：\n{lines}" + + (f"\n\n已经处理好的（不是异常，不要写成缺口或「可能不准」）：\n{done}" if done else "") + (LOCKED_NOTE if locked else "") ) @@ -392,6 +464,7 @@ class ReviewResult: async def review( model: Any, *, question: str, answer: str, signals: list[Signal], locked: bool = False, + settled: Iterable[str] = (), ) -> ReviewResult: """叫一次模型复核。调用方负责确保 signals 非空——空信号根本不该走到这里。 @@ -407,7 +480,8 @@ async def review( s.detail for s in signals ) - messages = [("system", REVIEW_SYSTEM), ("human", review_prompt(question, answer, signals, locked=locked))] + messages = [("system", REVIEW_SYSTEM), + ("human", review_prompt(question, answer, signals, locked=locked, settled=settled))] try: raw = await model.with_structured_output(REVIEW_SCHEMA).ainvoke(messages) if not isinstance(raw, dict): diff --git a/backend/tests/test_bundle_shown.py b/backend/tests/test_bundle_shown.py index af720153..a29efb6f 100644 --- a/backend/tests/test_bundle_shown.py +++ b/backend/tests/test_bundle_shown.py @@ -60,7 +60,7 @@ function render(c) { try { const value = v.parseJson(c.json) const token = v.tokenAt(v.numberTokens(c.json), []) - if (c.cell) return { shown: v.renderCell(value, token, c.conv, c.kind, c.display) } + if (c.cell) return { shown: v.renderCell(value, token, c.conv, c.kind, c.display, c.plain) } const decimals = v.parseJson(c.decimals) return { shown: v.renderNumber(value, token, { unit: c.unit, decimals, decimalsToken: v.tokenAt(v.numberTokens(c.decimals), []), fmt: c.format, conv: c.conv }) } @@ -137,9 +137,10 @@ def py_number(value: Any, unit: str, decimals: Any, fmt: Any, conv: Any) -> str return None -def py_cell(raw: Any, conv: Any, kind: Any, display: int) -> str | None: +def py_cell(raw: Any, conv: Any, kind: Any, display: int, plain: bool = False) -> str | None: try: - return render_cell(_Snapshots(display=display).shown(raw), conv, kind=kind) + # 3 版起标识列的数不分千位：服务端按 SQL 判（cell_format），引用上记 plain，页面照它渲染 + return render_cell(_Snapshots(display=display).shown(raw), conv, kind=kind, plain=plain and display >= 3) except RenderError: return None @@ -190,9 +191,9 @@ def cell_cases() -> list[dict[str, Any]]: text = json.dumps(raw, ensure_ascii=False) for kind in CELL_KINDS: for conv in CELL_CONVS: - for display in (1, 2): + for display, plain in ((1, False), (2, False), (3, False), (3, True), (2, True)): cases.append({"cell": True, "json": text, "kind": kind, "conv": conv, "display": display, - "py": py_cell(json.loads(text), conv, kind, display)}) + "plain": plain, "py": py_cell(json.loads(text), conv, kind, display, plain)}) return cases @@ -331,7 +332,8 @@ def test_every_format_in_the_monthly_sample_passes_as_exported(): "m:conv_rate": "2.35%", "m:growth": "-8.2%", "m:orders": "12840单", "m:avg": "5.81元", "m:avg|.3": "5.811元", "m:avg|int": "6元", "m:share|pct": "34.6%", "m:export": "15,230.4美元", "v:Q1.r0.region": "华南", "v:Q1.r0.gmv": "45,678.5", "v:Q1.r0.gmv|万": "4.57万", "v:Q1.r1.region": "华东", - "v:Q1.r1.gmv": "-1,200.75", "v:Q1.r0.order_no": "202,609,071,234,567,890", + # 订单号是标识列（列名以 _no 结尾）：3 版显示规则起不分千位，超过 2^53 也一位不差 + "v:Q1.r1.gmv": "-1,200.75", "v:Q1.r0.order_no": "202609071234567890", "v:Q1.r0.paid_at": "2026-09-07 08:21:39", } @@ -353,8 +355,8 @@ def test_every_format_in_the_monthly_sample_passes_as_exported(): ("v:Q1.r0.gmv", "54,678.5", "45678.5", "45,678.5"), # DECIMAL 落成的数字文本 ("v:Q1.r1.gmv", "1,200.75", "-1200.75", "-1,200.75"), # 负数单元格 ("v:Q1.r0.gmv|万", "5.57万", "45678.5", "4.57万"), # 单元格换算成万 - ("v:Q1.r0.order_no", "202,609,071,234,567,891", "202609071234567890", - "202,609,071,234,567,890"), # 超过 2^53 的整数 + # 超过 2^53 的整数；标识列不分组，应显示的字和引用的值一样，原因里不另写 + ("v:Q1.r0.order_no", "202609071234567891", "202609071234567890", None), ("v:Q1.r0.paid_at", "2026-09-08 08:21:39", "2026-09-07T08:21:39", "2026-09-07 08:21:39"), # 日期时间 ]) def test_changing_only_the_shown_text_fails_and_names_both(ref, text, cited, expected): @@ -391,13 +393,27 @@ def test_a_format_the_page_does_not_know_cannot_pass(ref, cite, text): assert hit["result"] == "na" and hit["reason"].startswith("无法核对显示文字："), hit +def test_identifier_cells_read_without_separators_and_the_page_follows(): + """标识列（订单号这类编号，列名以 _no 结尾）的数不分千位：文档记 cell_display: 3、引用上记 plain，页面照它渲染、 + 原样通过；把 plain 去掉的伪造页，页面按千分位渲染，和正文显示的字对不上，不通过。""" + from tests.bundle_samples import formats_bundle + + bundle = formats_bundle() + art, doc = doc_of(bundle) + seg = segment(doc, "v:Q1.r0.order_no") + assert doc["cell_display"] == 3 and seg["text"] == "202609071234567890" and seg["cite"]["plain"] is True + assert item(verdicts(bundle), "v:Q1.r0.order_no")["result"] == "pass" + seg["cite"].pop("plain") + assert item(verdicts(reseal(bundle, art, doc)), "v:Q1.r0.order_no")["result"] == "fail" + + def test_an_unknown_cell_display_version_cannot_pass(): """文档记下的单元格显示规则版本认不出（服务端复核记 bad_schema）：单元格的显示文字判无法核对；口径卡的照常核对。""" from tests.bundle_samples import formats_bundle bundle = formats_bundle() art, doc = doc_of(bundle) - doc["cell_display"] = 3 + doc["cell_display"] = 4 got = verdicts(reseal(bundle, art, doc)) cells = [n for n in got["numbers"] if n["ref"].startswith("v:")] assert cells and all(n["result"] == "na" and n["reason"].startswith("无法核对显示文字：") for n in cells) @@ -521,7 +537,7 @@ def test_numeric_text_cells_show_their_own_digits_in_the_panel(): table["rows"][0][2] = "202609071234567891" got = verdicts(formats_bundle(table=table)) order = item(got, "v:Q1.r0.order_no") - assert order["result"] == "pass" and order["text"] == "202,609,071,234,567,891", order + assert order["result"] == "pass" and order["text"] == "202609071234567891", order assert order["truthText"] == "202609071234567891", order assert item(got, "v:Q1.r0.gmv")["truthText"] == "45678.50" # 不是数字文本的格子不另写原文：照旧按值写 diff --git a/backend/tests/test_chat_turn_policy.py b/backend/tests/test_chat_turn_policy.py index 609c019b..3fceb663 100644 --- a/backend/tests/test_chat_turn_policy.py +++ b/backend/tests/test_chat_turn_policy.py @@ -194,7 +194,10 @@ async def test_answer_ops_says_to_retry_an_empty_generation(client, monkeypatch) events.append(json.loads(line[6:])) [err] = [e for e in events if e["op"] == "error"] assert err["code"] == turn.EMPTY_GENERATION and err["next"] == "retry", err - assert err["reason"] == "模型没有给出任何修改，也没有说明原因" and err["message"].startswith("助手本轮未完成") + # 原因是重来时交给助手的那句纠正（这次连思考都没有），诊断写在 detail（不含原文） + assert err["reason"].startswith("什么都没有输出") and "look" in err["reason"], err + assert err["message"] == "助手本轮未完成：模型没有给出任何修改，也没有说明原因" + assert "正文 0 字，思考 0 字" in err["detail"] and err["diagnostics"]["ops"] == 0 async def test_the_eval_driver_retries_an_empty_generation_once(monkeypatch): @@ -226,8 +229,10 @@ async def test_the_eval_driver_retries_an_empty_generation_once(monkeypatch): suite_id="s", suite_name="虚构题集", eval_run_id="e", case_id=None, case_seq=1, actor=None)) t = await driver.drive() assert t.path == path and t.attempts == 1 and model.calls == 2, (t.path, t.attempts, model.calls, t.reason) - assert t.retries == ["模型没有给出任何修改，也没有说明原因"], t.retries - assert seen[1] == turn.retry_instruction("上周华中区销售额", t.retries[0], False), seen + assert len(t.retries) == 1 and t.retries[0].startswith("什么都没有输出"), t.retries + # 重来时问题后面附那句纠正，不原样重发 + assert seen[1] == turn.empty_retry_instruction("上周华中区销售额", t.retries[0]) != seen[0], seen + assert "什么都没有输出" in seen[1] if path == "build_failed": assert t.copilot_failed and t.reason.startswith("助手本轮未完成"), t.reason assert t.usage["assistant"]["calls"] == 2, "两次生成都是这一题花的" diff --git a/backend/tests/test_chat_turns.py b/backend/tests/test_chat_turns.py index 77104fe7..21c22209 100644 --- a/backend/tests/test_chat_turns.py +++ b/backend/tests/test_chat_turns.py @@ -866,7 +866,9 @@ async def test_an_empty_generation_is_retried_once_on_the_server(client, monkeyp await _done(turn_id) events = await _events(turn_id) [retry] = [e["data"] for e in events if e["type"] == "turn.retry"] - assert retry["kind"] == "empty" and retry["reason"] == "模型没有给出任何修改，也没有说明原因" + # 原因是重来时交给助手的那句纠正（这次连思考都没有），诊断（不含原文）随事件记下 + assert retry["kind"] == "empty" and retry["reason"].startswith("什么都没有输出"), retry + assert "正文 0 字" in retry["detail"], retry row = await _row(turn_id) assert row.status == "done" and row.answer == "上周没有华中区的数据" and row.meta["noQuery"] is True assert row.meta["usage"]["copilot"]["calls"] == 2, "两次建图都是这一轮花的" diff --git a/backend/tests/test_copilot_answer_ops.py b/backend/tests/test_copilot_answer_ops.py index e65d8d30..9a3203d1 100644 --- a/backend/tests/test_copilot_answer_ops.py +++ b/backend/tests/test_copilot_answer_ops.py @@ -151,7 +151,14 @@ async def test_the_sse_output_is_unchanged_but_for_a_trailing_usage(client, monk assert [(op.pop("next"), op.pop("code", None)) for op in errors] == \ ([ERROR_NEXT[name]] if name in ERROR_NEXT else []) for op in errors: - assert op["message"] == f"助手本轮未完成：{op.pop('reason')}", op + reason = op.pop("reason") + if name == "no_ops": + # 一个操作都没给：原因是重来时交给助手的那句纠正，诊断（不含原文）写进 detail、diagnostics，别的照旧 + assert "每个操作单独一行" in reason and op.pop("diagnostics")["dropped"] == {"不是 JSON 对象": 1}, op + assert op["detail"].startswith("结束原因"), op + op["detail"] = "" + else: + assert op["message"] == f"助手本轮未完成：{reason}", op assert got[:-1] == expected diff --git a/backend/tests/test_copilot_empty_diagnosis.py b/backend/tests/test_copilot_empty_diagnosis.py new file mode 100644 index 00000000..6d054135 --- /dev/null +++ b/backend/tests/test_copilot_empty_diagnosis.py @@ -0,0 +1,122 @@ +"""助手一个操作都没给（copilot_empty）时看得出模型交回了什么，自动重来那一次带一句纠正；格式跑偏的操作不再被静默丢掉。 + +部署上撞过：问数据建图两次都是空的——模型在思考里说「先看一下表」，却一个操作都没给，自动重来一次还是这样，错误详情为空， +看不出模型实际返回了什么。现在： + +- 诊断（只记长度、个数和原因，不记原文）：结束原因、正文和思考各多少字、认出几行操作、原生工具调用几个（叫什么）、跳过了 + 哪些行（不是完整的 JSON、缺 op……）、正文里有没有写成文字的工具调用标记。写进 error 的 detail（给人看的一句）和 + diagnostics（机读）； +- 重来时交给助手的原因换成一句纠正（只写了思考 → 直接输出操作、要看表先写一行 look；格式跑偏 → 每个操作单独一行完整的 + JSON；用了工具调用 → 这里不走工具调用），不原样重发； +- 原来会被静默丢掉的几种写法认出来：一个操作分成多行写的 JSON、和围栏写在同一行的操作、原生工具调用里名字是操作名的 + （look、add_node……，参数就是操作的字段）。名字不认识、参数解析不了的照实记进诊断。 + +需求、表名都是虚构的。 +""" +from __future__ import annotations + +import json + +import pytest +from langchain_core.messages import AIMessageChunk + +from app.api.copilot_gen import StreamDiagnosis, _iter_ops +from app.chat import turn as policy +from tests.test_copilot_answer_ops import _sse, _use, client # noqa: F401 + + +class _Model: + def __init__(self, chunks: list[AIMessageChunk]) -> None: + self.chunks = chunks + + async def astream(self, messages): + for chunk in self.chunks: + yield chunk + + +def _think(text: str) -> AIMessageChunk: + return AIMessageChunk(content="", additional_kwargs={"reasoning_content": text}) + + +async def _ops(chunks: list[AIMessageChunk]) -> tuple[list[dict], StreamDiagnosis]: + diag = StreamDiagnosis() + ops = [op async for op in _iter_ops(_Model(chunks), [], diag=diag)] + return [op for op in ops if op["op"] != "thinking"], diag + + +async def test_only_thinking_is_diagnosed(): + ops, diag = await _ops([_think("先看一下表结构再决定。"), + AIMessageChunk(content="", response_metadata={"finish_reason": "stop"})]) + assert ops == [] + assert diag.text_chars == 0 and diag.thinking_chars == len("先看一下表结构再决定。") and diag.ops == 0 + assert diag.finish == "stop" + assert "只写了思考" in diag.correction() and "look" in diag.correction() + summary = diag.summary() + assert "结束原因 stop" in summary and "正文 0 字" in summary and "思考 11 字" in summary + assert "先看一下表" not in summary and "先看一下表" not in json.dumps(diag.as_dict(), ensure_ascii=False) + + +async def test_skipped_lines_are_counted_with_reasons(): + # 最后一行以「{」开头、读不成：可能是分成多行写的操作，往后攒，流读完还是读不成的算跳过 + text = '这是说明文字\n[1, 2]\n{"note": "缺 op"}\n{"op": "look", "need": \n' + ops, diag = await _ops([AIMessageChunk(content=text)]) + assert ops == [] + assert diag.dropped == {"不是完整的 JSON": 2, "不是 JSON 对象": 1, "缺 op": 1}, diag.dropped + assert "跳过 4 行：2 行不是完整的 JSON、1 行不是 JSON 对象、1 行缺 op" in diag.summary() + assert "每个操作单独一行" in diag.correction() + + +async def test_a_pretty_printed_op_is_read_not_dropped(): + text = '{"op": "plan", "summary": "先看表"}\n{\n "op": "look",\n "need": "产量表的字段"\n}\n' + ops, diag = await _ops([AIMessageChunk(content=text)]) + assert [op["op"] for op in ops] == ["plan", "look"] and ops[1]["need"] == "产量表的字段" + assert diag.dropped == {} and diag.ops == 2 + + +async def test_an_op_on_the_fence_line_is_read(): + text = '```json {"op": "plan", "summary": "先看表"}\n{"op": "look", "need": "产量表"}```\n' + ops, _ = await _ops([AIMessageChunk(content=text)]) + assert [op["op"] for op in ops] == ["plan", "look"] + + +async def test_a_native_tool_call_named_after_an_op_becomes_that_op(): + chunks = [AIMessageChunk(content="", tool_call_chunks=[ + {"name": "look", "args": '{"need": "产量', "id": "c1", "index": 0}]), + AIMessageChunk(content="", tool_call_chunks=[{"name": None, "args": '表的字段"}', "id": None, "index": 0}]), + AIMessageChunk(content="", tool_call_chunks=[{"name": "run_sql", "args": '{"sql": "SELECT 1"}', "id": "c2", + "index": 1}]), + AIMessageChunk(content="", tool_call_chunks=[{"name": "add_node", "args": '{"node": ', "id": "c3", + "index": 2}])] + ops, diag = await _ops(chunks) + assert ops == [{"op": "look", "need": "产量表的字段"}] + assert diag.tool_calls == ["look", "run_sql", "add_node"] + assert diag.dropped == {"工具调用的名字不认识": 1, "工具调用的参数解析不了": 1}, diag.dropped + + +async def test_tool_call_markup_in_text_is_diagnosed(): + ops, diag = await _ops([AIMessageChunk(content='{"name": "look", "arguments": {}}\n')]) + assert ops == [] and diag.markup == 1 + assert "不走工具调用" in diag.correction() + + +async def test_the_empty_error_carries_the_diagnosis_and_a_correction(client, monkeypatch): # noqa: F811 + from tests.test_copilot_answer_ops import BODY + + _use(monkeypatch, _Model([_think("先看一下表。"), AIMessageChunk(content="", response_metadata={ + "finish_reason": "stop"})])) + got = await _sse(client, BODY) + [error] = [op for op in got if op["op"] == "error"] + assert error["code"] == "copilot_empty" and error["next"] == "retry" + assert error["message"] == "助手本轮未完成：模型没有给出任何修改，也没有说明原因" + assert "正文 0 字" in error["detail"] and "思考 6 字" in error["detail"] and "结束原因 stop" in error["detail"] + assert error["diagnostics"]["text_chars"] == 0 and error["diagnostics"]["thinking_chars"] == 6 + assert "只写了思考" in error["reason"] + # 重来时交给助手的话：问题后面附这句纠正，不是原样重发 + again = policy.empty_retry_instruction("上周各条线的产量", policy.error_reason(error)) + assert again.startswith("上周各条线的产量") and "只写了思考" in again and again != "上周各条线的产量" + + +@pytest.mark.parametrize("reason", ["", "模型没有给出任何修改，也没有说明原因"]) +def test_without_a_correction_the_retry_still_asks_for_ops(reason): + again = policy.empty_retry_instruction("产量是多少", reason) + assert "look" in again and "产量是多少" in again diff --git a/backend/tests/test_copilot_truncation.py b/backend/tests/test_copilot_truncation.py index b3f0733c..9bc29a0d 100644 --- a/backend/tests/test_copilot_truncation.py +++ b/backend/tests/test_copilot_truncation.py @@ -163,15 +163,27 @@ async def test_a_cut_in_the_middle_of_the_ops_is_not_delivered_as_done(client, m async def test_output_that_parses_to_nothing_is_an_error(client, monkeypatch): - """格式跑偏（整段放进 markdown 围栏、JSON 分了行），每一行都被跳过：一个操作都没有，也没有 done。""" - pretty = "```json\n" + json.dumps({"op": "add_node", "node": {"id": "yoy", "type": "llm"}}, indent=2) + "\n```\n" - model = _Scripted([AIMessageChunk(content=pretty), + """格式跑偏（只写了说明文字、没有一行是操作），每一行都被跳过：一个操作都没有，也没有 done。诊断写明跳过了几行。""" + model = _Scripted([AIMessageChunk(content="先看一下表结构，\n再决定怎么搭。\n"), AIMessageChunk(content="", response_metadata={"finish_reason": "stop"})]) events = await _events(client, monkeypatch, model) kinds = _kinds(events) assert "final" not in kinds [err] = [e for e in events if e["op"] == "error"] assert err["message"].startswith("助手本轮未完成：") and "画布未修改" in err["hint"] + assert "跳过 2 行：2 行不是完整的 JSON" in err["detail"] and "先看一下表" not in err["detail"], err + + +async def test_a_pretty_printed_op_in_a_fence_is_read(client, monkeypatch): + """整段放进 markdown 围栏、JSON 分了行：以前每一行都被跳过、报一个操作都没有；现在攒成一个操作照常落地。""" + pretty = "```json\n" + json.dumps({"op": "add_node", "node": {"id": "yoy", "type": "llm", "label": "同比", + "config": {"prompt": "算同比"}}}, + indent=2, ensure_ascii=False) + "\n```\n" + model = _Scripted([AIMessageChunk(content=pretty), + AIMessageChunk(content="", response_metadata={"finish_reason": "stop"})]) + events = await _events(client, monkeypatch, model) + assert "error" not in _kinds(events), events + assert any(e["op"] == "add_node" and e["node"]["id"] == "yoy" for e in events) async def test_a_model_that_says_done_without_changes_is_still_unchanged(client, monkeypatch): diff --git a/backend/tests/test_evidence_entities.py b/backend/tests/test_evidence_entities.py index 668f2c38..87715859 100644 --- a/backend/tests/test_evidence_entities.py +++ b/backend/tests/test_evidence_entities.py @@ -161,10 +161,11 @@ def test_phase_two_ledgers_build_no_entities(): def test_entity_markers_render_as_the_name(catalog): doc = compose_doc("数据来自 [[t:orders]] 的 [[c:orders.amount]] 和 [[c:order_cnt]]。[[see:t:refunds]]", catalog) - assert doc["markdown"] == "数据来自 orders 的 orders.amount 和 order_cnt。" + # 紧跟在同一张表名后面的字段只显示字段名（不写成「orders 的 orders.amount」），出处照旧是那个字段 + assert doc["markdown"] == "数据来自 orders 的 amount 和 order_cnt。" segs = entity_segments(doc) assert [(s["text"], s["cite"]["kind"], s["state"]) for s in segs] == [ - ("orders", "table", "deterministic"), ("orders.amount", "column", "deterministic"), + ("orders", "table", "deterministic"), ("amount", "column", "deterministic"), ("order_cnt", "column", "deterministic")] assert segs[0]["cite"]["eid"] == catalog["t:orders"]["eid"] assert doc["violations"] == [] and doc["stats"]["entities"] == 3 diff --git a/backend/tests/test_evidence_tidy_claims.py b/backend/tests/test_evidence_tidy_claims.py index 2bae8038..d226b8c1 100644 --- a/backend/tests/test_evidence_tidy_claims.py +++ b/backend/tests/test_evidence_tidy_claims.py @@ -37,6 +37,7 @@ CARD = { metric("gmv", "销售额", 45678.5, "元", 1), metric("yoy", "同比增幅", 12.5, "%", 1, "plain"), metric("gone", "缺数", None, "人次"), + metric("lines", "产线条数", 8, "条"), ], } LEDGER = [{"kind": "metric_set", "node_id": "cal", "exec": 1, "artifact": ART, "caliber": "客流口径", @@ -108,6 +109,10 @@ def test_a_definition_with_a_number_is_still_a_claim(catalog): ("销售额 [[m:gmv|万]] 万元。", "销售额 4.57万元。"), ("同比 [[m:yoy]] %。", "同比 12.5%。"), ("客流 [[m:visits]] 人次的水平", "客流 12,345人次的水平"), # 多字单位后面跟着字，照样去掉 + # 单字的单位后面紧跟着字：和单位连起来不是一个更长的单位（「条产」），照样去掉。部署上撞过「覆盖 8条 条产线」 + ("覆盖 [[m:lines]] 条产线。", "覆盖 8条产线。"), + ("覆盖[[m:lines]]条产线。", "覆盖8条产线。"), + ("到访 [[m:heads]] 人参加。", "到访 1,200人参加。"), ]) def test_a_repeated_unit_is_dropped(catalog, raw, shown): doc = compose_doc(raw, catalog) @@ -117,6 +122,17 @@ def test_a_repeated_unit_is_dropped(catalog, raw, shown): assert verify_doc(doc, catalog)["violations"] == [] +def test_the_writer_and_the_copilot_are_told_markers_carry_their_unit(): + """助手写报告指令、写作者写正文，都先知道 [[m:]] 渲染出来自带单位、后面不要再写一遍。""" + from app.api.copilot import NODE_REFERENCE + from app.engine.evidence import UNIT_RULE + + start = NODE_REFERENCE.index("- report：报告撰写") + report = NODE_REFERENCE[start:NODE_REFERENCE.index("\n- output 的出具契约", start)] + assert "自带单位" in report and "条产线" in report + assert "自带单位" in UNIT_RULE and "后面不要再写" in UNIT_RULE + + @pytest.mark.parametrize(("raw", "shown"), [ # 口径卡的单位是「人」，写作者写的是「人次」：不是同一个单位，不动，让不一致露出来 ("到访 [[m:heads]] 人次。", "到访 1,200人 人次。"), @@ -188,9 +204,10 @@ def _stream(catalog, raw, sizes): @pytest.mark.parametrize("size", [1, 2, 3, 5, 8, UNIT_LOOKAHEAD, 40]) def test_stream_drops_the_repeated_unit_and_the_blank_like_the_full_render(catalog, size): - raw = "客流 [[m:visits]] 人次，为最高月份 [[see:m:visits]]。到访 [[m:heads]] 人次 [[see:m:heads]]\n末尾 [[m:gmv]] 元" + raw = ("客流 [[m:visits]] 人次，为最高月份 [[see:m:visits]]。到访 [[m:heads]] 人次 [[see:m:heads]]\n" + "覆盖 [[m:lines]] 条产线，末尾 [[m:gmv]] 元") full = render_markers(raw, catalog) - assert full == "客流 12,345人次，为最高月份。到访 1,200人 人次\n末尾 45,678.5元" + assert full == "客流 12,345人次，为最高月份。到访 1,200人 人次\n覆盖 8条产线，末尾 45,678.5元" out = _stream(catalog, raw, [size] * len(raw)) assert "".join(out) == full, out @@ -209,7 +226,8 @@ def test_stream_holds_the_blank_before_see_until_it_knows_what_follows(catalog, def _corpus(n): rng = random.Random(20260930) pieces = ["[[m:visits]]", "[[m:gmv|万]]", "[[m:heads]]", "[[m:yoy]]", "[[see:m:visits]]", "[[see:m:yoy]]", - "人次", "人", "次", "万元", "元", "%", " ", " ", "　", "\n", "。", "，", "文字", "[[", "]]", "["] + "[[m:lines]]", "条", "产线", "人次", "人", "次", "万元", "元", "%", " ", " ", "　", "\n", "。", "，", + "文字", "[[", "]]", "["] return ["".join(rng.choice(pieces) for _ in range(rng.randint(1, 16))) for _ in range(n)] diff --git a/backend/tests/test_evidence_unicode_names.py b/backend/tests/test_evidence_unicode_names.py index 2135cee4..cca25ed5 100644 --- a/backend/tests/test_evidence_unicode_names.py +++ b/backend/tests/test_evidence_unicode_names.py @@ -97,7 +97,8 @@ def test_new_doc_records_the_syntax_and_resolves_chinese_names(catalog): doc = compose("本月[[t:明细]]的[[c:明细.金额]]按[[c:区域]]汇总，[[v:Q1.r0.区域]]最高。[[see:t:汇总,Q1]]", catalog) assert ENTITY_SYNTAX == 2 and doc["entity_syntax"] == 2 and doc_entity_syntax(doc) == 2 assert '"entity_syntax":2' in canonical_json(doc) # 在文档顶层，参与内容哈希 - assert doc["markdown"] == "本月明细的明细.金额按区域汇总，分区甲最高。" # 显示写作者写的名字本身 + # 显示写作者写的名字本身；紧跟在同一张表名后面的字段只显示字段名（不写成「明细的明细.金额」） + assert doc["markdown"] == "本月明细的金额按区域汇总，分区甲最高。" assert doc["violations"] == [] table, column, shared = entities(doc) assert table["cite"]["alias"] == "t:明细" and table["state"] == "deterministic" @@ -117,7 +118,7 @@ def test_streaming_renders_chinese_names_like_the_full_doc(catalog): streamed = "".join([renderer.feed("本月[[t:明"), renderer.feed("细]]的[[c:明细.金"), renderer.feed("额]]上升。"), renderer.flush()]) assert streamed == render_markers(text, catalog, loader=loader) == compose(text, catalog)["markdown"] \ - == "本月明细的明细.金额上升。" + == "本月明细的金额上升。" @pytest.mark.parametrize("ref", [ diff --git a/backend/tests/test_relative_dates.py b/backend/tests/test_relative_dates.py new file mode 100644 index 00000000..58a2a32f --- /dev/null +++ b/backend/tests/test_relative_dates.py @@ -0,0 +1,63 @@ +"""「今天、昨天、上周、本月、去年」一律按运行当天算。 + +部署上撞过：问「上周各条线的产量」，数据只更新到两周前；查库时拿表里最大的日期往前推七天当「上周」，报告照着写成 +「上周产量……」，读的人以为是上周的数。现在： +- 助手的节点手册（画布、问数据共用）的搭图规则里有一条：相对日期按运行当天算（SQL 用当天的写法，agent 先调 + current_time），不用 MAX(日期) 去推； +- 报告写作规则告诉写作者运行当天是哪天、上周是哪几天：数据没覆盖问的那段时间时先照实说、写明数据最新到哪天；可以给出 + 最近一段的数，但写具体起止日期，不叫它「上周」。 +数据、日期都是虚构的。 +""" +from __future__ import annotations + +from datetime import date + +from app.api.copilot import NODE_REFERENCE +from app.engine.evidence import period_rule +from tests.test_report_node import engine_up, finish, node, script # noqa: F401 + + +def _build_rules() -> str: + start = NODE_REFERENCE.index("搭图规则（有数字结论时必须遵守）") + return NODE_REFERENCE[start:NODE_REFERENCE.index("\n\n模型字段", start)] + + +def test_the_copilot_reckons_relative_dates_from_the_run_day(): + rules = _build_rules() + assert "今天、昨天、上周、本月、去年" in rules and "运行当天" in rules + assert "CURRENT_DATE" in rules and "current_time" in rules # SQL 用当天的写法；agent 先取当天日期 + assert "MAX(日期)" in rules and "不要" in rules # 点名不要拿数据里最大的日期去推 + + +def test_the_writer_rule_names_the_run_day_and_last_week(): + rule = period_rule(date(2031, 3, 12)) # 虚构的一天：周三 + assert "2031-03-12（周三）" in rule + assert "上周是 2031-03-03 至 2031-03-09" in rule # 上周一到上周日 + assert "不按数据里最新的日期推" in rule + assert "数据最新到哪天" in rule and "起止日期" in rule and "不要把它叫作「上周」" in rule + + +def test_last_week_on_a_monday_and_a_sunday(): + assert "上周是 2031-03-03 至 2031-03-09" in period_rule(date(2031, 3, 10)) # 周一 + assert "上周是 2031-03-03 至 2031-03-09" in period_rule(date(2031, 3, 16)) # 周日：本周还没过完，上周照旧是前一周 + + +async def test_the_report_prompt_carries_the_run_day(monkeypatch): + from app.engine.nodes import report as report_mod + + monkeypatch.setattr(report_mod, "run_today", lambda: date(2031, 3, 12)) + seen = script(monkeypatch, "没有数据。") + graph = { + "nodes": [ + node("start", "input", fields=[{"name": "q", "default": "上周的产量是多少"}]), + node("write", "report", instructions="回答：{{ input.q }}"), + node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}]), + ], + "edges": [{"source": "start", "target": "write"}, {"source": "write", "target": "out"}], + } + await finish(graph) + prompt = "\n".join(str(m.content) for m in seen[0]) + assert "今天是 2031-03-12（周三）" in prompt and "上周是 2031-03-03 至 2031-03-09" in prompt + from app.engine.evidence import UNIT_RULE + + assert UNIT_RULE in prompt, "[[m:]] 自带单位的那条也在写作规则里" diff --git a/backend/tests/test_report_cells.py b/backend/tests/test_report_cells.py index 197b4079..da35c1d3 100644 --- a/backend/tests/test_report_cells.py +++ b/backend/tests/test_report_cells.py @@ -266,8 +266,7 @@ def test_table_defaults_to_the_first_five_rows_and_all_columns(catalog): @pytest.mark.parametrize("marker, words", [ - ("[[table:Q1 rows=0-25]]", "最多 20 行"), - ("[[table:Q1 rows=5-9]]", "只有 7 行"), + ("[[table:Q1 rows=7-9]]", "只有 7 行，没有第 7 行"), # 起始行就越界了 ("[[table:Q1 rows=3-1]]", "rows"), ("[[table:Q1 cols=region,gmv]]", "没有列「gmv」"), ("[[table:Q1 limit=3]]", "无法识别的选项"), @@ -283,6 +282,39 @@ def test_table_problems_are_violations(catalog, marker, words): assert "|" not in doc["markdown"] +def test_table_rows_past_the_end_stop_at_the_last_row(catalog): + """结束行写过了头（只有 7 行却写 rows=5-9）：起始行在范围内，就截到最后一行，不加注、不算违规。部署上撞过：查询回的 + 行比写作者以为的少，整张表报「没有第 N 行」被打回重写。""" + doc = compose_doc("[[table:Q1 cols=region rows=5-9]]", catalog) + assert doc["violations"] == [], doc["violations"] + assert doc["markdown"] == "| region |\n| --- |\n| 东北 |\n| 中部 |" + doc = compose_doc("[[table:Q1 cols=region rows=0-25]]", catalog) # 要的比上限还多，截完只剩 7 行 + assert doc["violations"] == [], doc["violations"] + assert {u["loc"]["row"] for u in doc["blocks"][0]["units"]} == {-1, 0, 1, 2, 3, 4, 5, 6} + + +def test_an_old_document_with_an_overflowing_table_still_verifies_as_it_was(catalog, monkeypatch): + """改动前（2 版）组装的文档：结束行越界的整表当年判为解析不了、正文是占位。按文档记下的版本复核，照旧是解析不了， + 不能因为今天截得了行，就把它判成「渲染对不上」（文档被改过）。""" + from app.engine import evidence as ev + + monkeypatch.setattr(ev, "CELL_DISPLAY", 2) + old = compose_doc("[[table:Q1 cols=region rows=5-9]]", catalog) + monkeypatch.undo() + assert "没有第 9 行" in bad(old)["table:Q1 cols=region rows=5-9"] and "cell_display" not in old + codes = [v["code"] for v in verify_doc(old, catalog)["violations"]] + assert codes == ["unresolved_ref"], codes + + +def test_table_over_the_row_limit_is_still_a_violation(): + long = {"columns": ["n"], "rows": [[f"第{i}项"] for i in range(30)], "row_count": 30, "truncated": False, + "elapsed_ms": 1, "sql": "SELECT n FROM items", "source": "shop"} + catalog = build_catalog(nodes={}, ledger=[_query(_store(long), long)], inputs={}) + doc = compose_doc("[[table:Q1 rows=0-25]]", catalog) + [(ref, message)] = bad(doc).items() + assert ref == "table:Q1 rows=0-25" and "最多 20 行" in message, message + + def test_table_written_inline_still_becomes_a_table(catalog): doc = compose_doc("见表 [[table:Q1 cols=region rows=0-1]] 如上。", catalog) assert doc["violations"] == [] diff --git a/backend/tests/test_report_datetime_display.py b/backend/tests/test_report_datetime_display.py index 946e6eda..2b31a6bf 100644 --- a/backend/tests/test_report_datetime_display.py +++ b/backend/tests/test_report_datetime_display.py @@ -58,7 +58,8 @@ def test_iso_datetimes_read_with_a_space(catalog): 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 + # 记的是渲染得出这些格子的最低一版：3 版（整表截行）不改格子的写法，只有日期时间被改写的照旧记 2 + assert doc["cell_display"] == 2 and doc_cell_display(doc) == 2 and CELL_DISPLAY >= 2 assert '"cell_display":2' in canonical_json(doc), "在文档顶层，参与内容哈希" assert verify_doc(doc, catalog, loader=load)["ok"] diff --git a/backend/tests/test_report_entity_short.py b/backend/tests/test_report_entity_short.py new file mode 100644 index 00000000..04a7cb32 --- /dev/null +++ b/backend/tests/test_report_entity_short.py @@ -0,0 +1,90 @@ +"""字段标记紧跟在同一张表名后面时只显示字段名（engine/evidence._short_column）。 + +部署上撞过「取自 某表 的 某表.某列」：[[c:表.列]] 渲染成「表.列」全名，写作者前面又写了「某表 的」。现在：标记前面紧挨着 +（隔空白、「的」「中的」「里的」）刚写过同一张表的名字（写成文字或 [[t:同一张表]] 都算）时，只显示列名；引用、出处、证据 +标识照旧是那个字段，引用上记 short: true，复核照它核对显示的字。没有这一项的老文档照旧按全名核对。写作规则也写一句。 +表名、列名都是虚构的。 +""" +from __future__ import annotations + +import copy + +import pytest + +from app.engine.evidence import ENTITY_RULES, StreamRenderer, build_catalog, compose_doc, iter_segments, \ + render_markers, verify_doc + +SCHEMA_ART = "6" * 64 +SCHEMA = {"kind": "schema", "node_id": "fetch", "exec": 1, "artifact": SCHEMA_ART, "source": "plant", + "tables": {"产量表": ["工厂", "产量"], "产线表": ["工厂", "产线"], "plant_output": ["factory", "qty"]}} +QUERY = {"kind": "query", "node_id": "fetch", "exec": 1, "artifact": "a" * 64, "tool": "db_query__plant", + "source": "plant", "columns": ["工厂", "产量"], "rows": 1, "truncated": False, "schema_artifact": SCHEMA_ART, + "tables": ["产量表"]} + + +@pytest.fixture +def catalog(): + return build_catalog(nodes={}, ledger=[QUERY, SCHEMA]) + + +def column_segments(doc): + return [s for s in iter_segments(doc) if s.get("ref", "").startswith("c:")] + + +@pytest.mark.parametrize("raw, shown", [ + ("工厂取自[[t:产量表]]的[[c:产量表.工厂]]。", "工厂取自产量表的工厂。"), + ("工厂取自 产量表 的 [[c:产量表.工厂]]。", "工厂取自 产量表 的 工厂。"), # 表名写成文字也算 + ("[[t:产量表]]中的[[c:产量表.产量]]按工厂汇总。", "产量表中的产量按工厂汇总。"), + ("产量表里的 [[c:产量表.产量]] 按工厂汇总。", "产量表里的 产量 按工厂汇总。"), + ("[[t:plant_output]] 的 [[c:plant_output.qty]]。", "plant_output 的 qty。"), + ("[[t:PLANT_OUTPUT]] [[c:plant_output.qty]]。", "PLANT_OUTPUT qty。"), # 只隔空白；英文名不分大小写 +]) +def test_a_column_right_after_its_table_shows_only_the_column(catalog, raw, shown): + doc = compose_doc(raw, catalog) + assert doc["markdown"] == shown and render_markers(raw, catalog) == shown + [seg] = column_segments(doc) + assert seg["state"] == "deterministic" and seg["cite"]["short"] is True + assert seg["cite"]["alias"] == f"c:{seg['ref'][2:]}" or seg["cite"]["alias"].lower() == seg["ref"].lower() + assert verify_doc(doc, catalog)["ok"], verify_doc(doc, catalog)["violations"] + + +@pytest.mark.parametrize("raw, shown", [ + ("工厂取自[[t:产线表]]的[[c:产量表.工厂]]。", "工厂取自产线表的产量表.工厂。"), # 前面是另一张表 + ("[[c:产量表.工厂]]取自产量表。", "产量表.工厂取自产量表。"), # 前面没有表名 + ("产量表，[[c:产量表.工厂]]。", "产量表，产量表.工厂。"), # 隔着逗号：不是紧挨着 + ("myplant_output 的 [[c:plant_output.qty]]。", "myplant_output 的 plant_output.qty。"), # 英文名前面还连着字母 +]) +def test_other_columns_keep_the_full_name(catalog, raw, shown): + doc = compose_doc(raw, catalog) + assert doc["markdown"] == shown and render_markers(raw, catalog) == shown + assert not any(s["cite"].get("short") for s in column_segments(doc)) + + +def test_a_document_written_before_still_verifies_with_the_full_name(catalog): + """改动前组装的文档：字段照全名显示、引用上没有 short。照全名核对，不判成被改过。""" + doc = compose_doc("工厂取自[[t:产量表]]的[[c:产量表.工厂]]。", catalog) + old = copy.deepcopy(doc) + [seg] = column_segments(old) + start = seg["span"][0] + old["markdown"] = old["markdown"].replace("的工厂。", "的产量表.工厂。") + seg["text"], seg["span"] = "产量表.工厂", [start, start + len("产量表.工厂")] + seg["cite"].pop("short") + seg["cite"]["rendered"] = "产量表.工厂" + tail = next(s for s in iter_segments(old) if s["span"][0] == start + len("工厂")) + tail["span"] = [start + len("产量表.工厂"), start + len("产量表.工厂") + len(tail["text"])] + for unit in (u for b in old["blocks"] for u in b["units"]): + unit["span"] = [unit["span"][0], unit["span"][1] + len("产量表.")] + assert verify_doc(old, catalog)["ok"], verify_doc(old, catalog)["violations"] + + +@pytest.mark.parametrize("size", [1, 2, 3, 5, 8, 40]) +def test_the_stream_shows_the_same(catalog, size): + raw = "工厂取自[[t:产量表]]的[[c:产量表.工厂]]，[[t:产线表]]的[[c:产量表.工厂]]不同。" + renderer = StreamRenderer(catalog) + out = [renderer.feed(raw[i:i + size]) for i in range(0, len(raw), size)] + [renderer.flush()] + assert "".join(out) == render_markers(raw, catalog) == compose_doc(raw, catalog)["markdown"] \ + == "工厂取自产量表的工厂，产线表的产量表.工厂不同。" + + +def test_the_writer_is_told(): + assert "只显示字段名" in ENTITY_RULES diff --git a/backend/tests/test_report_failed_queries.py b/backend/tests/test_report_failed_queries.py index 8dc03c5f..eb59b080 100644 --- a/backend/tests/test_report_failed_queries.py +++ b/backend/tests/test_report_failed_queries.py @@ -52,3 +52,59 @@ async def test_without_failed_queries_the_prompt_is_unchanged(monkeypatch): run = await finish(graph(calls)) assert "没有查到的数据" not in _prompt(seen[0]) assert not [e for e in await events(run.id, "log") if (e.data or {}).get("code") == "report_queries_failed"] + + +# -------------------------------------------------------------------------- +# 失败之后又改对了的那一次不算「没有查到」 +# -------------------------------------------------------------------------- + + +def _failed(calls: list[dict]) -> list[dict]: + from app.engine.nodes.report import failed_queries + from app.engine.schema import GraphSpec + + g = graph("[]") + spec = GraphSpec.model_validate(g) + state = {"nodes": {"agent1": {"text": "x", "tool_calls": calls}}} + return failed_queries(state, spec, spec.node_map()["write"]) + + +BAD = {"tool": "db_query__shop", "args": {"sql": "SELECT line AS 产线, SUM(qty) AS 产量 FROM output GROUP BY SUM(qty)"}, + "ok": True, "result": "查询失败：aggregate functions are not allowed in GROUP BY"} + + +def _good(sql: str, tool: str = "db_query__shop") -> dict: + return {"tool": tool, "args": {"sql": sql}, "ok": True, "result": '{"columns": ["产线", "产量"], "rows": [["甲线", 10]]}'} + + +def test_a_failure_fixed_by_the_next_query_is_superseded(): + """取数 Agent 写错一条（GROUP BY 里放了聚合），下一条马上改对、读的表覆盖了失败那条：这次失败已被替代，不进写作提示。""" + fixed = _good("SELECT line AS 产线, SUM(qty) AS 产量 FROM output GROUP BY line") + assert _failed([BAD, fixed]) == [] + # 解析不了的失败 SQL：同一个数据源之后有成功的查询，也算被替代 + garbled = {**BAD, "args": {"sql": "SELEC line FROM"}} + assert _failed([garbled, fixed]) == [] + + +def test_a_failure_not_covered_by_a_later_query_stays(): + """之后成功的查询读的是别的表（没覆盖失败那条读的表）、或者成功在失败之前：照旧列进「没有查到的数据」。""" + other = _good("SELECT COUNT(*) AS n FROM stations") + [left] = _failed([BAD, other]) + assert "aggregate functions" in left["reason"] + before = _good("SELECT line AS 产线, SUM(qty) AS 产量 FROM output GROUP BY line") + assert len(_failed([before, BAD])) == 1, "成功在失败之前，不算替代了它" + elsewhere = _good("SELECT line, SUM(qty) FROM output GROUP BY line", tool="db_query__crm") + assert len(_failed([BAD, elsewhere])) == 1, "别的数据源上的成功不算" + + +async def test_when_every_query_failed_the_writer_still_says_none_succeeded(monkeypatch): + calls = ("[{'tool': 'db_query__shop', 'args': {'sql': 'SELECT line, SUM(qty) FROM output GROUP BY SUM(qty)'}, " + "'ok': True, 'result': '查询失败：aggregate functions are not allowed'}, " + "{'tool': 'db_query__shop', 'args': {'sql': 'SELECT line, SUM(qty) FROM output GROUP BY 1'}, " + "'ok': True, 'result': '查询失败：connection reset'}]") + seen = script(monkeypatch, "没有查到产量：查询失败。") + run = await finish(graph(calls)) + first = _prompt(seen[0]) + assert "上游有 2 次数据查询失败了" in first and "这次一次成功的查询都没有" in first + [warned] = [e for e in await events(run.id, "log") if (e.data or {}).get("code") == "report_queries_failed"] + assert warned.data["failed"] == 2 diff --git a/backend/tests/test_report_identifier_cells.py b/backend/tests/test_report_identifier_cells.py new file mode 100644 index 00000000..22484941 --- /dev/null +++ b/backend/tests/test_report_identifier_cells.py @@ -0,0 +1,121 @@ +"""报告里标识列的数不分千位（engine/cell_format.plain_columns，单元格显示规则 3 版）。 + +部署上撞过：产线编码这类数值型的编码列在报告的表格里写成「1,203」，读的人以为是个量。现在按序看三样信号：查询当时冻结的 +数据目录给源列标的度量类型（identifier / status / attribute 不分，flow / stock / ratio 分）、源列名像编码编号（CODE、 +_ID、ID、NO、编码、编号、代码）、是分组键而不是聚合；SUM、COUNT、AVG 出来的是量，照旧分千位；判断不了的照旧分千位。 +真有格子因此改了写法的文档记 cell_display: 3、引用上记 plain；老文档按记下的版本复核，照旧分千位。数据是虚构的。 +""" +from __future__ import annotations + +import pytest + +from app.core import artifact_store +from app.engine import evidence as ev +from app.engine.cell_format import code_like, plain_columns +from app.engine.evidence import build_catalog, compose_doc, verify_doc, writer_catalog + +SCHEMA = {"source": "plant", "tables": {"plant_output": {"columns": []}}, + "catalog": {"plant_output": {"version": 2, "notes": {"columns": { + "batch": {"measure": {"value": "identifier", "source": "human", "status": "confirmed"}}, + "capacity": {"measure": {"value": "stock", "source": "human", "status": "confirmed"}}}}}}} +SQL = ("SELECT p.STATION_CODE AS 产线, YEAR(p.made_on) AS 年份, p.capacity AS 产能, SUM(p.qty) AS 产量, " + "COUNT(*) AS 工单数, MAX(p.batch) AS 批次 FROM plant_output p " + "GROUP BY p.STATION_CODE, YEAR(p.made_on), p.capacity") +COLUMNS = ["产线", "年份", "产能", "产量", "工单数", "批次"] + + +def _store(obj: dict) -> str: + text = artifact_store.canonical_json(obj) + artifact = artifact_store.content_hash(text) + path = artifact_store._path_of(artifact) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return artifact + + +def _snap(schema_artifact: str) -> dict: + return {"columns": COLUMNS, "column_types": {c: "number" for c in COLUMNS}, + "rows": [[1203, 2031, 7001, 15200, 1450, 88001], [1207, 2031, 6400, 9300, 1210, 88002]], + "row_count": 2, "truncated": False, "elapsed_ms": 2, "sql": SQL, "source": "plant", + "schema_artifact": schema_artifact} + + +AMOUNTS = {"columns": ["金额"], "column_types": {"金额": "number"}, "rows": [[12345]], "row_count": 1, + "truncated": False, "elapsed_ms": 1, "sql": "SELECT o.amount * 2 AS 金额 FROM orders o", "source": "plant"} + + +def _query(artifact: str, columns: list[str], call: str) -> dict: + return {"kind": "query", "node_id": "fetch", "exec": 1, "artifact": artifact, "via": "b" * 64, "call_id": call, + "tool": "db_query__plant", "source": "plant", "columns": columns, "rows": 2, "truncated": False} + + +@pytest.fixture +def catalog(): + snap = _store(_snap(_store(SCHEMA))) + ledger = [_query(snap, COLUMNS, "call_1"), _query(_store(AMOUNTS), ["金额"], "call_2")] + return build_catalog(nodes={}, ledger=ledger, inputs={}) + + +TEXT = ("[[v:Q1.r0.年份]] 年 [[v:Q1.r0.产线]] 线产量 [[v:Q1.r0.产量]]，工单 [[v:Q1.r0.工单数]] 张，产能 [[v:Q1.r0.产能]]，" + "批次 [[v:Q1.r0.批次]]；另一项金额 [[v:Q2.r0.金额]]。[[see:Q1,Q2]]") + + +def _shown(doc) -> dict[str, str]: + return {s["ref"]: s["text"] for b in doc["blocks"] for u in b["units"] for s in u["segments"] + if s.get("ref", "").startswith("v:")} + + +def test_identifier_columns_read_without_separators(catalog): + doc = compose_doc(TEXT, catalog) + assert doc["violations"] == [], doc["violations"] + shown = _shown(doc) + assert shown["v:Q1.r0.产线"] == "1203" # 源列名像编码（STATION_CODE） + assert shown["v:Q1.r0.年份"] == "2031" # 分组键 + assert shown["v:Q1.r0.批次"] == "88001" # 目录标的是标识（MAX 取的是那一列本身的值） + assert shown["v:Q1.r0.产能"] == "7,001" # 分组键，但目录标的是存量：是量，照旧分千位 + assert shown["v:Q1.r0.产量"] == "15,200" # SUM：量 + assert shown["v:Q1.r0.工单数"] == "1,450" # COUNT：量 + assert shown["v:Q2.r0.金额"] == "12,345" # 判断不了：照旧分千位 + assert doc["cell_display"] == 3 + plain = {s["ref"] for b in doc["blocks"] for u in b["units"] for s in u["segments"] + if (s.get("cite") or {}).get("plain")} + assert plain == {"v:Q1.r0.产线", "v:Q1.r0.年份", "v:Q1.r0.批次"} + assert verify_doc(doc, catalog)["ok"] + + +def test_tables_and_the_writer_catalog_read_the_same_way(catalog): + doc = compose_doc("[[table:Q1 cols=产线,产量 rows=0-1]]", catalog) + assert doc["violations"] == [] and "| 1203 | 15,200 |" in doc["markdown"] and "| 1207 | 9,300 |" in doc["markdown"] + text = writer_catalog(catalog).text + assert "产线=1203" in text and "产量=15,200" in text, text + + +def test_documents_composed_before_still_verify_with_separators(catalog, monkeypatch): + monkeypatch.setattr(ev, "CELL_DISPLAY", 2) + old = compose_doc(TEXT, catalog) + monkeypatch.undo() + assert _shown(old)["v:Q1.r0.产线"] == "1,203" and "cell_display" not in old + assert verify_doc(old, catalog)["ok"], "按文档记下的版本复核：照旧分千位，不判成被改过" + + +def test_a_document_edited_back_to_separators_is_caught(catalog): + doc = compose_doc("产线 [[v:Q1.r0.产线]]。", catalog) + seg = next(s for b in doc["blocks"] for u in b["units"] for s in u["segments"] if s.get("ref") == "v:Q1.r0.产线") + start = seg["span"][0] + doc["markdown"] = doc["markdown"].replace("1203", "1,203") + seg["text"], seg["span"] = "1,203", [start, start + 5] + assert "render_mismatch" in {v["code"] for v in verify_doc(doc, catalog)["violations"]} + + +@pytest.mark.parametrize("name, yes", [ + ("STATION_CODE", True), ("order_id", True), ("ID", True), ("user_no", True), ("orderId", True), ("stationCode", True), + ("产线编码", True), ("工单编号", True), ("物料代码", True), + ("qty", False), ("piano", False), ("video", False), ("income", False), ("产量", False), +]) +def test_code_like_names(name, yes): + assert code_like(name) is yes + + +def test_unparseable_sql_falls_back_to_the_column_name(): + snap = {"columns": ["station_id", "total"], "sql": "this is not sql", "rows": []} + assert plain_columns(snap) == frozenset({"station_id"}) diff --git a/backend/tests/test_report_text_cell_numbers.py b/backend/tests/test_report_text_cell_numbers.py new file mode 100644 index 00000000..b57302b1 --- /dev/null +++ b/backend/tests/test_report_text_cell_numbers.py @@ -0,0 +1,73 @@ +"""查询结果里文本格的名字（「冲压5线」「KX-205 线」）照抄进报告时，名字里的数字不算裸数字。 + +部署上撞过：查询结果的产线名里带数字，写作者照着写「冲压5线的产量 [[v:…]]」，名字里的那个 5 被当成没有出处的数字打回 +重写，重写还是同一个名字。现在：写作目录的查询结果里、文本格的值（一字不差的子串；值里至少有一个不是数字的字，本身 +不是数、也不是带单位的数）出现在正文里时，那一段不查裸数字。只认快照里真有的值，不做模糊匹配：同一个数单独写出来照样 +要出处；数值格不受影响。数据是虚构的。 +""" +from __future__ import annotations + +import pytest + +from app.core import artifact_store +from app.engine.evidence import build_catalog, compose_doc, verify_doc + +SNAP = { + "columns": ["产线", "产量", "备注"], + "column_types": {"产线": "text", "产量": "number", "备注": "text"}, + "rows": [ + ["冲压5线", 1520, "120台"], # 带单位的数：不算名字 + ["KX-205 线", 830, "45%"], + ["焊装 2 线", 64, "3.5万元"], + ], + "row_count": 3, "truncated": False, "elapsed_ms": 2, + "sql": "SELECT line AS 产线, SUM(qty) AS 产量, MAX(memo) AS 备注 FROM output GROUP BY line", + "source": "plant", +} + + +def _store(snap: dict) -> str: + text = artifact_store.canonical_json(snap) + artifact = artifact_store.content_hash(text) + path = artifact_store._path_of(artifact) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return artifact + + +@pytest.fixture +def catalog(): + art = _store(SNAP) + ledger = [{"kind": "query", "node_id": "fetch", "exec": 1, "artifact": art, "via": "b" * 64, "call_id": "call_1", + "tool": "db_query__plant", "source": "plant", "columns": SNAP["columns"], "rows": 3, "truncated": False}] + return build_catalog(nodes={}, ledger=ledger, inputs={}) + + +def _uncited(doc) -> list[str]: + return [v["text"] for v in doc["violations"] if v["code"] == "uncited_number"] + + +def test_a_number_inside_a_name_from_the_results_is_allowed(catalog): + text = "冲压5线的产量是 [[v:Q1.r0.产量]]，KX-205 线是 [[v:Q1.r1.产量]]，焊装 2 线是 [[v:Q1.r2.产量]]。[[see:Q1]]" + doc = compose_doc(text, catalog) + assert _uncited(doc) == [], doc["violations"] + assert verify_doc(doc, catalog)["ok"], "出口复核按同一条规则：组装时放过的，复核也放过" + + +def test_the_same_number_on_its_own_still_needs_a_citation(catalog): + doc = compose_doc("冲压5线有 5 个班组，产量 [[v:Q1.r0.产量]]。[[see:Q1]]", catalog) + assert _uncited(doc) == ["5"] + # 名字只认一字不差的：快照里没有「冲压15线」，15 照样要出处；名字前后紧挨着数字的也不算那个名字 + assert _uncited(compose_doc("冲压15线的产量 [[v:Q1.r0.产量]]。", catalog)) == ["15"] + assert _uncited(compose_doc("KX-2050 线的产量 [[v:Q1.r1.产量]]。", catalog)) == ["2050"] + assert _uncited(compose_doc("焊装 2 线以外还有 2 线。", catalog)) == ["2"] + + +def test_numbers_and_numbers_with_units_in_text_cells_are_not_names(catalog): + doc = compose_doc("备注里写着 120台、45%、3.5万元。", catalog) + assert _uncited(doc) == ["120", "45%", "3.5万"], doc["violations"] + + +def test_numeric_cells_are_unaffected(catalog): + # 产量格的 1520 是数值格：照抄成裸数字照样违规 + assert _uncited(compose_doc("冲压5线的产量是 1520。", catalog)) == ["1520"] diff --git a/backend/tests/test_review.py b/backend/tests/test_review.py index 90cd3e56..d5dbbaf8 100644 --- a/backend/tests/test_review.py +++ b/backend/tests/test_review.py @@ -428,3 +428,51 @@ def test_agent_field_warnings_are_degraded_signals(): ], OK_OUTPUT) assert [(s.kind, s.severity) for s in signals] == [ ("agent_field_mismatch", rv.DEGRADED), ("agent_field_unverified", rv.DEGRADED)] + + +# -------------------------------------------------------------------------- +# 报告「改过一次」的信号按最终结果定档 +# -------------------------------------------------------------------------- + + +def _repaired(final_ok: bool) -> list[dict]: + violations = [] if final_ok else [{"code": "bare_number", "message": "裸数字 12"}] + return [ + ev(EventType.NODE_STARTED, node_id="write", label="写周报"), + ev(EventType.LOG, node_id="write", level="warn", code="report_repair", count=2, attempt=1, + message="报告有 2 处未通过核对（裸数字 12、裸数字 30），已要求模型重写（第 1 次）"), + ev(EventType.REPORT_CHECKED, node_id="write", ok=final_ok, violations=violations, repairs=1), + ] + + +def test_a_repair_that_finally_passed_is_not_a_gap(): + """首稿没过核对、重写一次之后通过（最后一次 report.checked ok、没有违规）：不进异常清单，复核不为它叫模型； + 复核被别的信号叫起来时，说明里写「首稿有 N 处没过核对，重写后已全部通过」。""" + rows = _repaired(True) + assert rv.scan(rows, OK_OUTPUT) == [] + [note] = rv.notes(rows) + assert note == "报告「写周报」首稿有 2 处没过核对，重写后已全部通过" + prompt = rv.review_prompt("上周产量", "答案", [rv.Signal("tool_error", "工具 x 调用失败", rv.BROKEN)], + settled=rv.notes(rows)) + assert "重写后已全部通过" in prompt and "不要写成缺口" in prompt + assert prompt.index("检测到的异常") < prompt.index("已经处理好的") + assert "首稿有 2 处" not in prompt.split("已经处理好的")[0], "处理好了的不在异常清单里" + + +def test_a_repair_that_still_failed_stays_a_gap(): + rows = _repaired(False) + [signal] = rv.scan(rows, OK_OUTPUT) + assert signal.kind == "report_repair" and signal.severity == rv.DEGRADED + assert rv.notes(rows) == [] + + +def test_a_rewrite_counts_by_the_final_judgement(): + base = [ev(EventType.NODE_STARTED, node_id="write", label="写周报"), + ev(EventType.LOG, node_id="write", level="warn", code="report_rewrite", count=1, + message="裁判认为 1 句结论证据不支持（……），已退回写作模型仅修改这几句")] + good = [*base, ev(EventType.REPORT_CHECKED, node_id="write", ok=True, violations=[], + judge={"counts": {"supported": 3, "contradicted": 0, "insufficient": 0}})] + bad = [*base, ev(EventType.REPORT_CHECKED, node_id="write", ok=True, violations=[], + judge={"counts": {"supported": 2, "contradicted": 1, "insufficient": 0}})] + assert rv.scan(good, OK_OUTPUT) == [] and rv.notes(good) == ["报告「写周报」有 1 句结论裁判认为证据不支持，改写后都有了依据"] + assert [s.kind for s in rv.scan(bad, OK_OUTPUT)] == ["report_rewrite"] and rv.notes(bad) == [] diff --git a/scripts/check-bundle.mjs b/scripts/check-bundle.mjs index 6ac7b7b4..40c7ee30 100644 --- a/scripts/check-bundle.mjs +++ b/scripts/check-bundle.mjs @@ -864,7 +864,8 @@ await section('样本页：各种形状', async () => { check('formats：点开换算成万的数字，引用的值、口径卡里的值照原值写，注明与按显示规则渲染的结果一致', detail.includes('74613.65') && detail.includes('与按显示规则渲染的结果一致') && !detail.includes('按显示规则应为'), detail.replace(/\s+/g, ' ').slice(0, 160)) - const order = page.locator('button.num').filter({ hasText: '202,609,071,234,567,890' }) + // 订单号是标识列（列名以 _no 结尾）：3 版显示规则起不分千位 + const order = page.locator('button.num').filter({ hasText: /^202609071234567890$/ }) detail = '找不到订单号' let rowCell = '' if (await order.count()) {