合入部署实例一段问数据会话里查出的 9 条修正:复核按报告最后一次核对定档(首稿没过、重写后通过的不算缺口);被随后改对的查询失败不再写进答案;「上周」这类相对日期按运行当天算,数据覆盖不到先照实说;整表引用行号超出截到最后一行;名字里的数字(和查询结果文本值一字不差的)不算裸数字;标识列(编码、编号、分组键)不加千分位(单元格显示规则 3 版,离线核对页同步);指标后紧跟同一单位不再重复;前面写过表名时字段只显示列名;建图返回空时记诊断、重来附纠正,并修掉三处工具调用被静默丢弃的路径。验收:后端全量 14871 通过;相关 8 项浏览器检查 3922 项全过;tsc、文案检查;泄露扫描干净

This commit is contained in:
ThearchyHelios
2026-10-19 18:10:12 +08:00
29 changed files with 1478 additions and 114 deletions
+17 -7
View File
@@ -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.<agent 的 assign_to>.字段
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)
+1 -2
View File
@@ -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 []
+209 -21
View File
@@ -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 是正文里写成文字的工具调用标记(<tool_call>、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":
+9 -6
View File
@@ -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 */
+7 -3
View File
@@ -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
+9 -5
View File
@@ -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 {})})
# ---------------- 执行 ----------------
+18 -3
View File
@@ -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"]
+183
View File
@@ -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"]
+198 -27
View File
@@ -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)
+52 -8
View File
@@ -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))])
+77 -3
View File
@@ -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):
+26 -10
View File
@@ -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"
# 不是数字文本的格子不另写原文:照旧按值写
+8 -3
View File
@@ -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, "两次生成都是这一题花的"
+3 -1
View File
@@ -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, "两次建图都是这一轮花的"
+8 -1
View File
@@ -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
@@ -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='<tool_call>{"name": "look", "arguments": {}}</tool_call>\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
+15 -3
View File
@@ -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):
+3 -2
View File
@@ -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
+21 -3
View File
@@ -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)]
+3 -2
View File
@@ -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", [
+63
View File
@@ -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:]] 自带单位的那条也在写作规则里"
+34 -2
View File
@@ -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"] == []
@@ -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"]
+90
View File
@@ -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
@@ -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
@@ -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"})
@@ -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"]
+48
View File
@@ -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) == []
+2 -1
View File
@@ -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()) {