From 799c80b97e4ffd6e32f7c035ba6f61a73b1148d1 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 11:28:23 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B8=85=E5=8D=95=E6=9C=89=E7=9C=81=E7=95=A5?= =?UTF-8?q?=E6=97=B6=E4=B8=8D=E8=AE=B8=E6=96=AD=E8=A8=80=E3=80=8C=E6=B2=A1?= =?UTF-8?q?=E6=9C=89=E6=95=B0=E6=8D=AE=E3=80=8D=EF=BC=9A=E5=86=99=E4=BD=9C?= =?UTF-8?q?=E8=A7=84=E5=88=99=E5=8A=A0=E4=B8=80=E6=9D=A1=EF=BC=8C=E6=A0=B8?= =?UTF-8?q?=E5=AF=B9=E8=AE=A4=E5=87=BA=E7=BC=BA=E5=A4=B1=E6=96=AD=E8=A8=80?= =?UTF-8?q?=E4=BA=A4=E5=9B=9E=E9=87=8D=E5=86=99=E4=B8=80=E6=AC=A1=EF=BC=8C?= =?UTF-8?q?=E4=BB=8D=E6=9C=89=E7=9A=84=E5=8F=AA=E6=A0=87=E6=B3=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一次真实运行里写作者只看到上百次查询里的前二十来次，却写出「某类数据缺失、本轮没有查到任何结果」这样的断言，可那个 Agent 实际查到了上百条记录。 - 写作规则：清单有省略、或有来源的结构化结论为空时，不得断言没有查到、不存在、缺失、没有数据，只能写「本报告未覆盖」并说明依据不足 - 组装文档：清单有省略时按一组明确的说法认出缺失断言（没有查到、未查询到、查不到、没有任何记录、数据为空、数据缺失、不存在相关记录、暂无数据等；「不存在明显差异」「不代表没有数据」这类不算，代码块不查），记新违规 absence_claim，带 span、所在片段和原句 - 和句中的 [[see:]] 一样归入 REWRITE_ONLY：交回重写一次，重写后仍有的只标注、不判失败；只剩这几类时不再按最多重写次数反复重写。仍有时报告节点发 report_absence_claim（带 count、texts），运行复核算作缺口 - 证据面板的违规清单标「缺失断言」并写明为什么依据不足，时间线给 report_absence_claim 专门的一行，文案在 terms.ts；check-decode、check-evidence 补断言 - 没展开的查询的摘要在没有表结构快照时从查询快照里的 SQL 认表名 - 回归测试：合成的门店、会员、卡券、活动库真跑「4 个 Agent 并行查库 → 报告」（查询次数虚构、分布悬殊），覆盖抽取截断重试、两次截断记空值、清单均衡和缺失断言的重写 --- backend/app/engine/evidence.py | 74 +++- backend/app/engine/nodes/report.py | 22 +- backend/app/engine/review.py | 2 + backend/tests/test_report_absence_claim.py | 82 +++++ .../tests/test_report_evidence_regression.py | 316 ++++++++++++++++++ frontend/src/lib/terms.ts | 13 + frontend/src/run/EvidencePanel.tsx | 12 +- frontend/src/run/decode.ts | 12 +- scripts/check-decode.mjs | 20 ++ scripts/check-evidence.mjs | 29 +- 10 files changed, 568 insertions(+), 14 deletions(-) create mode 100644 backend/tests/test_report_absence_claim.py create mode 100644 backend/tests/test_report_evidence_regression.py diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 02a62c8b..ff830cfb 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -2474,11 +2474,13 @@ def compose_doc( cells_allowed: bool = True, loader: Loader | None = None, entities: bool = True, + absence: bool = False, ) -> dict[str, Any]: """模型写的原文（带标记）→ 报告文档：渲染后的 markdown、块、句、片段、统计、违规。 entities=False（报告节点 entities: off）：表名、字段名整层不管——不链接、不核对反引号， [[t:]] / [[c:]] 解析不了，原因是 ENTITIES_OFF_REASON。 + absence=True（写作目录有省略）：另查缺失断言（absence_claims），和句中的 [[see:]] 一样只在组装时查。 stats / violations 由 verify_doc 算出，和出口契约复核用的是同一个函数。 整表标记先展开成每格一个 [[v:]] 的 Markdown 表，之后和模型自己写的表格走同一条路； @@ -2634,8 +2636,10 @@ def compose_doc( "markdown": markdown, "source": source, "catalog": kept, "blocks": out_blocks, } checked = verify_doc(doc, catalog, allow_numbers=allow_numbers, cells_allowed=cells_allowed, loader=snaps) - # 句中的 [[see:]] 只在组装时查（交回写作者重写）：出口复核看的是数字和出处，句子通不通不影响出具 - doc["stats"], doc["violations"] = checked["stats"], checked["violations"] + inline_see(source) + # 句中的 [[see:]]、目录有省略时的缺失断言只在组装时查（交回写作者重写）：出口复核看的是数字和出处， + # 句子通不通、结论有没有越过证据不影响出具 + doc["stats"], doc["violations"] = checked["stats"], \ + checked["violations"] + inline_see(source) + (absence_claims(doc) if absence else []) return doc @@ -3035,6 +3039,58 @@ def inline_see(source: str) -> list[dict[str, Any]]: return [_violation("inline_see", INLINE_SEE_MESSAGE, text=m.group(0)) for m in _INLINE_SEE.finditer(source or "")] +#: 写作目录有省略、或者有来源的结构化结论为空时，拼在写作规则后面（给模型看） +ABSENCE_RULE = ( + "这次你看不到全部证据（证据目录有省略，或者有来源的结构化结论为空）：不得断言「没有查到」「未查到」「不存在」" + "「缺失」「没有数据」「无数据」这类结论。某个方向的证据没有展开或为空时，只能写「本报告未覆盖……」，并说明依据不足" + "（比如「目录有省略，相关查询未展开」），句末可以挂 [[see:Qn]] 指向相关查询。" +) + +#: 缺失断言：断言「没有数据」的几种说法。只认拿得准的，不泛化：「不存在明显差异」「缺少口径说明」这类说的是看到的 +#: 数据里的情况，不算；「不存在」「为空」「缺失」只认跟在数据、记录、结果后面（或者前面）的 +_ABSENCE = re.compile( + r"(?:没有|没|未|未能|没能)(?:查询|查)到|查不到|查无" + r"|(?:没有|暂无|并无)(?:任何)?(?:相关的?)?(?:数据|记录|结果)" + r"|无(?:任何)?(?:相关的?)?(?:数据|记录)(?![库源集表权字项列])" + r"|不存在(?:任何)?(?:相关的?)?(?:数据|记录|结果)|(?:数据|记录|结果)(?:均|都|全部)?(?:不存在|为空|缺失|空缺)" +) +#: 前面是这些词的不算：「不代表没有数据」「并非查不到」说的恰恰是不能下这个结论 +_ABSENCE_NEGATED = re.compile(r"(?:不代表|并不代表|不等于|不意味着|并非|不是|不能说明|不能断言|不得断言|不应断言)[^。！？\n]{0,6}$") +ABSENCE_MESSAGE = _Reason( + "证据目录有省略时写了缺失断言：写作模型没有看到全部证据，「没有查到」「不存在」「缺失」这类结论依据不足", + "证据目录有省略（没展开的查询只列了编号和摘要），这一句却断言了没有数据。看不到的证据不能当成不存在：" + "改成「本报告未覆盖……」，并说明依据不足（目录有省略、相关查询未展开），不要写「没有查到」「不存在」「缺失」「没有数据」", +) + + +def absence_claims(doc: dict[str, Any]) -> list[dict[str, Any]]: + """正文里的缺失断言（_ABSENCE），每处一条违规 absence_claim。只在写作目录有省略时由报告节点要求查（compose_doc 的 + absence）：目录全给了的时候，「本周没有退款记录」可能就是看了数据说的。 + + 在渲染后的正文里找（标记已经换成了字），代码块不查。带着 span、所在片段和句子：证据面板能定位、能显示原句。 + """ + markdown = str(doc.get("markdown") or "") + found: list[dict[str, Any]] = [] + for block in doc.get("blocks") or []: + if block.get("type") == "code": + continue + for unit in block.get("units") or []: + span = unit.get("span") + if not (isinstance(span, list) and len(span) == 2): + continue + sentence = markdown[span[0]:span[1]] + for m in _ABSENCE.finditer(sentence): + if _ABSENCE_NEGATED.search(sentence[:m.start()]): + continue + at = span[0] + m.start() + seg = next((g for g in unit.get("segments") or [] if g.get("kind") != "structural" + and isinstance(g.get("span"), list) and g["span"][0] <= at < g["span"][1]), None) + found.append(_violation("absence_claim", ABSENCE_MESSAGE, text=m.group(0), + span=[at, span[0] + m.end()], context=sentence.strip()[:120], + segment=seg.get("id") if seg else None, unit=unit.get("id"))) + return found + + #: 引用查询结果的写法。只跟着目录里的查询出现（catalog_prompt）：写作规则本身保持一期那份，升级前 #: 发起的运行跑到报告节点时提示一字不差；只有口径卡的目录也不该教写作者去引用不存在的 Q1 CELL_RULES = ( @@ -3320,8 +3376,16 @@ def _fields_prompt(sources: list[dict[str, Any]], catalog: dict[str, Any], snaps empty, list(dict.fromkeys(cited))) -def _query_brief(entry: dict[str, Any], tables: list[str]) -> str: - """没展开的查询在目录里的一行摘要：编号、查的表（没有表结构时写工具名）、行数。""" +def _query_brief(entry: dict[str, Any], tables: list[str], snaps: _Snapshots) -> str: + """没展开的查询在目录里的一行摘要：编号、查的表、行数。 + + 表先看目录里的表条目（带表结构快照的查询才有），没有就从查询快照里存的 SQL 认（sql_tables）， + 再没有写工具名。 + """ + if not tables: + snapshot, why = snaps.get(entry["alias"], entry.get("artifact")) + if not why and isinstance(snapshot.get("sql"), str): + tables = sql_tables(snapshot["sql"]) rows = entry.get("rows") what = ("、".join(tables[:3]) + (f" 等 {len(tables)} 张表" if len(tables) > 3 else "")) if tables \ else str(entry.get("tool") or entry.get("source") or "查询") @@ -3401,7 +3465,7 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], for e in queries: groups.setdefault(str(e.get("node_id") or ""), []).append(e) compact = {e["alias"]: _query_prompt(e, snaps, cells_allowed, _PROMPT_COMPACT_ROWS) for e in queries} - brief = {e["alias"]: [_query_brief(e, tables.get(e["alias"], []))] for e in queries} + brief = {e["alias"]: [_query_brief(e, tables.get(e["alias"], []), snaps)] for e in queries} shapes = {"brief": brief, "compact": compact, "full": full} tier = {e["alias"]: "brief" for e in queries} diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index a7d7cabc..c9bedf60 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -44,6 +44,7 @@ from app.engine import judge as judging from app.engine.context import NodeContext, NodeError from app.engine.labels import field_label, option_label from app.engine.evidence import ( + ABSENCE_RULE, CELLS_REASON, CLAIMS_RULE, ENTITY_KINDS, @@ -84,8 +85,10 @@ CLAIMS = CLAIMS_VALUES #: 只标注、不让写作者重写也不判失败的违规：可疑实体（受管的正式出具按缺口降档）、表结构快照不全时 #: 核对不了的名字（出口只标注） SOFT = frozenset({"unknown_entity", "unverified_entity"}) -#: 交回写作者重写、重写不好也不判失败的违规：句中的 [[see:]]（渲染后留下断句，数字和出处都没问题） -REWRITE_ONLY = frozenset({"inline_see"}) +#: 交回写作者重写一次、重写不好也不判失败的违规：句中的 [[see:]]（渲染后留下断句，数字和出处都没问题）； +#: 写作目录有省略时的缺失断言（「没有查到」「数据缺失」：写作者没看全证据，结论依据不足，但数字和出处都核对过）。 +#: 只剩这几类时不按 max_repairs 反复重写：重写一次没改掉，再写也多半一样，留在核对结果里标注 +REWRITE_ONLY = frozenset({"inline_see", "absence_claim"}) #: 重写一次就是多一次整篇的模型调用，写不对的模型多给几次也大多写不对 MAX_REPAIRS = 3 @@ -410,6 +413,9 @@ def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims instructions = ctx.render_str(ctx.cfg("instructions", ""), state).strip() \ or "根据下面的证据写一份简洁的报告，先总后分。" rules = f"{MARKER_RULES}\n{SEE_AND_CODE_RULES}" + if writer.guarded: + # 写作者看不全证据（目录有省略、有来源的结构化结论为空）：「没有查到」这类结论没有依据 + rules = f"{rules}\n{ABSENCE_RULE}" if claims in ("require_citation", CLAIMS_JUDGE): # 裁判只看写作者挂的依据：要判，就更得每句都挂上 rules = f"{rules}\n{CLAIMS_RULE}" + (f"\n{judging.JUDGE_RULE}" if claims == CLAIMS_JUDGE else "") @@ -500,12 +506,13 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: def compose(raw: str) -> dict[str, Any]: return compose_doc(raw, catalog, node_id=ctx.node.id, run_id=ctx.run.run_id, allow_numbers=allow, - cells_allowed=cells, entities=entities) + cells_allowed=cells, entities=entities, absence=writer.trimmed) response, raw = await draft(messages) doc = compose(raw) repairs = 0 - while (blocking := _blocking(doc["violations"], numbers)) and repairs < max_repairs: + while (blocking := _blocking(doc["violations"], numbers)) and repairs < max_repairs \ + and not (repairs and not _fatal(blocking)): repairs += 1 ctx.emit(EventType.LOG, level="warn", code="report_repair", message=f"报告有 {len(blocking)} 处未通过核对（{_named(blocking)}），" @@ -528,6 +535,13 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: draft=draft, compose=compose, numbers=numbers, blocking=blocking) spent.extend(usages) + if absent := [v for v in doc["violations"] if v.get("code") == "absence_claim"]: + # 重写之后还在（或者没让重写）：报告照常产出，这几处标注出来，时间线上单独说清为什么依据不足 + texts = list(dict.fromkeys(str(v.get("text") or "") for v in absent)) + tried = f"已要求模型重写 {repairs} 次仍未改掉" if repairs else "未要求重写" + ctx.emit(EventType.LOG, level="warn", code="report_absence_claim", count=len(absent), texts=texts, + message=f"报告在证据目录有省略时写了 {len(absent)} 处缺失断言（{_named(absent)}），{tried}：" + "被省略的查询仍在证据台账里，这类结论依据不足，应写作本报告未覆盖。报告照常产出，这几处已标注") doc_artifact = await _store(doc, ctx) ctx.emit(EventType.REPORT_CHECKED, doc_artifact=doc_artifact, ok=not doc["violations"], stats=doc["stats"], violations=doc["violations"][:20], repairs=repairs, diff --git a/backend/app/engine/review.py b/backend/app/engine/review.py index 98e0d36d..65c0b752 100644 --- a/backend/app/engine/review.py +++ b/backend/app/engine/review.py @@ -91,6 +91,8 @@ WARN_CODES: dict[str, tuple[str, str]] = { "report_no_evidence": ("report_no_evidence", DEGRADED), # 证据目录超出篇幅，有的查询只列了编号和摘要：写作模型没看全证据，报告的覆盖面有缺口 "report_catalog_trimmed": ("report_catalog_trimmed", DEGRADED), + # 目录有省略时写了「没有查到」这类缺失断言、重写后仍在：报告照常产出并标注，这几句结论依据不足 + "report_absence_claim": ("report_absence_claim", DEGRADED), # 口径卡的指标集、报告文档没能落进工件库：答案照常，只是点不开证据 "evidence_store_failed": ("evidence_store_failed", DEGRADED), # agent 的结构化字段：模型报的值和查询快照对不上（已按快照取值），或者出处核对不了 diff --git a/backend/tests/test_report_absence_claim.py b/backend/tests/test_report_absence_claim.py new file mode 100644 index 00000000..cdd3c7f1 --- /dev/null +++ b/backend/tests/test_report_absence_claim.py @@ -0,0 +1,82 @@ +"""缺失断言（absence_claim）：写作目录有省略时，「没有查到」「不存在」「缺失」「没有数据」这类结论依据不足。 + +2026-10 一次运行，写作者只看到上百次查询里的前二十来次，写出「某类数据缺失：本轮没有查到任何结果」这样的断言， +可那个 Agent 实际查到了上百条记录。 + +- 写作规则：清单有省略、或有来源的结构化结论为空时，不得断言没有数据，只能写「本报告未覆盖」并说明依据不足 +- 组装文档：清单有省略时按一组明确的说法认出缺失断言，记 absence_claim，交回重写一次；重写后仍有的只标注、不判失败 + （report.REWRITE_ONLY，和句中的 [[see:]] 同一种处理） +- 只认拿得准的说法：「不存在明显差异」「本报告未覆盖」「不代表没有数据」都不算 +""" +from __future__ import annotations + +import pytest + +from app.engine.evidence import ABSENCE_RULE, build_catalog, compose_doc +from app.engine.nodes.report import REWRITE_ONLY + + +@pytest.fixture +def catalog(): + card = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v1", + "metrics": [{"id": "gmv", "name": "销售额", "value": 45678.5, "unit": "元", "format": "thousands"}]} + return build_catalog(nodes={"card": card}, ledger=[{"kind": "metric_set", "node_id": "card", "artifact": "c" * 64}]) + + +def claims(doc): + return [v for v in doc["violations"] if v["code"] == "absence_claim"] + + +def test_the_rule_tells_the_writer_what_to_write_instead(): + for phrase in ("没有查到", "不存在", "缺失", "没有数据", "本报告未覆盖", "依据不足"): + assert phrase in ABSENCE_RULE + assert "absence_claim" in REWRITE_ONLY and "inline_see" in REWRITE_ONLY + + +@pytest.mark.parametrize("text, found", [ + ("卡券数据缺失：本轮没有查到核销、兑换、退券的任何结果。", ["数据缺失", "没有查到"]), + ("卡券核销没有任何记录。", ["没有任何记录"]), + ("本周门店客流数据为空。", ["数据为空"]), + ("活动内容无数据。", ["无数据"]), + ("未查询到会员流失的情况。", ["未查询到"]), + ("库里不存在相关记录。", ["不存在相关记录"]), + ("兑换记录均不存在。", ["记录均不存在"]), + ("暂无数据，查不到核销明细。", ["暂无数据", "查不到"]), + ("| 方向 | 结论 |\n| --- | --- |\n| 卡券 | 没有查到 |", ["没有查到"]), +]) +def test_absence_claims_are_recognised(catalog, text, found): + doc = compose_doc(text, catalog, absence=True) + got = claims(doc) + assert [v["text"] for v in got] == found, doc["violations"] + for v in got: + assert doc["markdown"][v["span"][0]:v["span"][1]] == v["text"], "span 指着正文里的那几个字" + assert v.get("unit") and v.get("context") + assert "依据不足" in v["message"] and "本报告未覆盖" in v["for_model"] + + +@pytest.mark.parametrize("text", [ + "本报告未覆盖卡券核销的明细：证据目录有省略，相关查询未展开，依据不足。", + "两家门店的客单价不存在明显差异。", + "目录里没展开的查询不代表没有数据。", + "这并非没有查到，而是没有展开。", + "本周无数据库变更，无数据源切换。", + "本周销售额 [[m:gmv]]，缺少口径说明的部分已注明。", + "```sql\nSELECT * FROM t -- 没有查到\n```", +]) +def test_other_wording_is_left_alone(catalog, text): + assert not claims(compose_doc(text, catalog, absence=True)) + + +def test_only_checked_when_the_catalog_was_trimmed(catalog): + text = "卡券核销数据缺失：没有查到任何记录。" + assert not claims(compose_doc(text, catalog)) + assert claims(compose_doc(text, catalog, absence=True)) + + +def test_the_claim_can_be_located_in_the_document(catalog): + doc = compose_doc("## 卡券\n\n本周销售额 [[m:gmv]]。核销数据缺失。", catalog, absence=True) + [v] = claims(doc) + segments = {s["id"]: s for b in doc["blocks"] for u in b["units"] for s in u["segments"]} + seg = segments[v["segment"]] + assert seg["kind"] == "text" and seg["span"][0] <= v["span"][0] < seg["span"][1] + assert v["context"] == "核销数据缺失。" diff --git a/backend/tests/test_report_evidence_regression.py b/backend/tests/test_report_evidence_regression.py new file mode 100644 index 00000000..0d7eb648 --- /dev/null +++ b/backend/tests/test_report_evidence_regression.py @@ -0,0 +1,316 @@ +"""回归：「4 个专项 Agent 并行查库 → 报告」整条跑一遍，复现 2026-10 那次运行的形状。 + +那次运行状态是 succeeded，报告却不可靠：4 个 Agent 的结构化抽取全部被截断（推理模型的额度在推理阶段就用完了）； +写作者的出处清单写死 12000 字，上百次查询只有前二十来次露了面；写作者于是在清单被截断的情况下写了「数据缺失： +本轮没有查到任何核销结果」，可那个 Agent 查到了上百条。 + +这里用合成的门店、会员、卡券、活动库（SQLite）真跑：4 个 Agent 分别查 26、17、79、42 次（虚构的次数，一次并行发完），都配了 +output_schema 和按出处核对字段；写作模型、抽取模型都是脚本。验证： + +- 抽取被截断后加大额度重试成功；两次都截断的记 agent_field_truncated，字段为空值 +- 出处清单在预算内，每个 Agent 都有代表，被省略的查询有编号和摘要，开头说明各来源省略了几次 +- 清单有省略时的缺失断言被认出来（absence_claim）、交回重写；重写后仍有的只标注、不判失败 +""" +from __future__ import annotations + +import asyncio +import json +import re +import sqlite3 + +import pytest +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage +from openai import LengthFinishReasonError +from openai.types import CompletionUsage +from openai.types.chat import ChatCompletion +from sqlalchemy import select + +from app.core import artifact_store +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import DataSource, Run, RunEvent +from app.engine.evidence import ABSENCE_RULE, DEFAULT_CATALOG_BUDGET, MARKER_RULES +from app.engine.nodes import llm as llm_mod +from app.engine.nodes import report as report_mod +from app.engine.runner import run_manager +from tests.fixtures.sources import drop_source + +SOURCE = "fa_store" +TOOL = f"db_query__{SOURCE}" +#: (节点 id, 画布上的名字, 记号, 查询次数, 表, 列) +AGENTS = [ + ("agent_member", "会员专项", "member", 26, "member_daily", ["store_name", "biz_date", "new_members", "active_members"]), + ("agent_store", "门店专项", "store", 17, "store_daily", ["store_name", "biz_date", "visits", "sales"]), + ("agent_coupon", "卡券核销", "coupon", 79, "coupon_redemption", ["store_name", "redeemed_at", "coupon_type", "amount"]), + ("agent_campaign", "活动内容", "campaign", 42, "campaign_view", ["campaign", "biz_date", "views", "clicks"]), +] +SCHEMA = {"type": "object", "properties": { + "first_store": {"type": "string", "description": "第一次查询第一行的门店或活动"}, + "first_value": {"type": "number", "description": "第一次查询第一行的第三列"}, + "churn_rate": {"type": "number", "description": "流失率（要计算，填 null）"}, +}} +ROWS_PER_QUERY = 5 + +ABSENT = ("## 概览\n\n各专项都已完成查询。[[see:Q1]]\n\n" + "## 卡券核销\n\n卡券数据缺失：本轮没有查到核销、兑换的任何结果。") +COVERED = ("## 概览\n\n各专项都已完成查询。[[see:Q1]]\n\n" + "## 卡券核销\n\n本报告未覆盖卡券核销的明细：证据目录有省略，相关查询未展开，依据不足。[[see:Q1]]") + + +def length_error(completion: int) -> LengthFinishReasonError: + usage = CompletionUsage(completion_tokens=completion, prompt_tokens=14000, total_tokens=14000 + completion) + return LengthFinishReasonError(completion=ChatCompletion.model_construct( + id="x", choices=[], created=0, model="m", object="chat.completion", usage=usage)) + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def store_db(tmp_path): + """合成库：每张表 120 行，第 i 次查询取 OFFSET i 的 5 行，次次结果不同。""" + path = tmp_path / "store.db" + db = sqlite3.connect(path) + for _, _, _, _, table, cols in AGENTS: + db.execute(f"CREATE TABLE {table} ({cols[0]} TEXT, {cols[1]} TEXT, {cols[2]} {'TEXT' if table == 'coupon_redemption' else 'INTEGER'}, {cols[3]} REAL)") + rows = [] + for r in range(120): + when = f"2026-06-{r % 28 + 1:02d}" + (f"T08:{r % 60:02d}:39" if table == "coupon_redemption" else "") + third = f"券型{r % 4 + 1}" if table == "coupon_redemption" else (r * 7) % 300 + 5 + rows.append((f"门店{r % 30 + 1:02d}", when, third, round((r * 13) % 900 + 0.5, 1))) + db.executemany(f"INSERT INTO {table} VALUES (?, ?, ?, ?)", rows) + db.commit() + db.close() + async with SessionLocal() as session: + row = DataSource(name=SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True) + session.add(row) + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + yield path + await drop_source(source_id) + + +def sql_of(table: str, cols: list[str], i: int) -> str: + return f"SELECT {', '.join(cols)} FROM {table} ORDER BY rowid LIMIT {ROWS_PER_QUERY} OFFSET {i}" + + +def node(nid, ntype, label, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label, "config": config}} + + +def graph(counts: dict[str, int] | None = None, **report): + nodes = [node("start", "input", "开始")] + for nid, title, key, count, table, cols in AGENTS: + nodes.append(node(nid, "agent", title, prompt=f"【{key}】按{title}的口径查数，每次查一段", tools=[TOOL], + approval="never", parallel_tools=True, output_schema=SCHEMA, cite_fields=True, + assign_to=f"{key}_findings")) + nodes.append(node("write", "report", "专项报告", instructions="汇总各专项的发现", **report)) + nodes.append(node("out", "output", "成果", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}])) + edges = [{"source": "start", "target": a[0]} for a in AGENTS] + [{"source": a[0], "target": "write"} for a in AGENTS] + edges.append({"source": "write", "target": "out"}) + return {"nodes": nodes, "edges": edges} + + +class World: + """脚本化的模型：Agent 一步并行发完自己的查询再给结论；抽取按 outcome 决定；写作者依次回 drafts。""" + + def __init__(self, monkeypatch, db_path, *, counts: dict[str, int], drafts: list[str], outcome): + from app.providers import mock_model + + self.counts = counts + self.drafts = drafts + self.writer: list[list] = [] + self.extract_calls: dict[str, list[int | None]] = {a[2]: [] for a in AGENTS} + self.budgets: list[int | None] = [] + db = sqlite3.connect(db_path) + self.first_rows = {key: db.execute(sql_of(table, cols, 1)).fetchone() for _, _, key, _, table, cols in AGENTS} + db.close() + real = llm_mod.get_chat_model + + async def spy(session, spec): + if spec.thinking == "off": + self.budgets.append(spec.max_tokens) + return await real(session, spec) + + monkeypatch.setattr(llm_mod, "get_chat_model", spy) + world = self + + def text(messages) -> str: + return "\n".join(m.content if isinstance(m.content, str) else json.dumps(m.content, ensure_ascii=False) + for m in messages) + + def who(messages) -> str: + found = re.search(r"【(member|store|coupon|campaign)】", text(messages)) + assert found, "认不出是哪个 Agent" + return found.group(1) + + def decide(model, messages): + if model.response_format: + key = who(messages) + world.extract_calls[key].append(world.budgets[-1]) + outcome(key, len(world.extract_calls[key]), world.budgets[-1]) + return AIMessage(content=json.dumps(world.extracted(key), ensure_ascii=False)) + if any(isinstance(m, SystemMessage) and "报告撰写人" in str(m.content) for m in messages): + world.writer.append(list(messages)) + return AIMessage(content=world.drafts[min(len(world.writer), len(world.drafts)) - 1]) + key = who(messages) + if model.tools and not any(isinstance(m, ToolMessage) for m in messages): + _, _, _, _, table, cols = next(a for a in AGENTS if a[2] == key) + return AIMessage(content="分段查询。", tool_calls=[ + {"name": TOOL, "args": {"sql": sql_of(table, cols, i)}, "id": f"{key}_{i}"} + for i in range(1, world.counts[key] + 1)]) + return AIMessage(content=f"【{key}】查完了，结论见各次查询。") + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) + + def extracted(self, key: str) -> dict: + first = self.first_rows[key] + _, _, _, _, _, cols = next(a for a in AGENTS if a[2] == key) + return {"first_store": {"value": first[0], "from": {"call": "Q1", "row": 0, "column": cols[0]}}, + "first_value": {"value": first[3], "from": {"call": "Q1", "row": 0, "column": cols[3]}}, + "churn_rate": {"value": None, "from": None}} + + +async def finish(graph_json) -> Run: + run = await run_manager.start(graph=graph_json, input_payload={}) + for _ in range(1200): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + return row + raise AssertionError(f"run {run.id} 没跑完：{row.status}") + + +async def events(run_id: str, etype: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id) + if etype: + q = q.where(RunEvent.type == etype) + return list((await session.execute(q.order_by(RunEvent.seq))).scalars()) + + +def logs(evts, code): + return [e.data for e in evts if e.type == "log" and e.data.get("code") == code] + + +async def output_of(run_id: str, node_id: str) -> dict: + [finished] = [e for e in await events(run_id, "node.finished") if e.node_id == node_id] + return artifact_store.load(finished.data["artifact"]) + + +def catalog_text(messages) -> str: + """写作者第一次收到的提示里出处清单那一段（从「可引用的证据」到写作规则之前）。""" + prompt = next(m.content for m in messages if isinstance(m, HumanMessage)) + return prompt[prompt.index("可引用的证据"):prompt.index(MARKER_RULES)].rstrip() + + +# -------------------------------------------------------------------------- +# 和那次运行同样的形状：上百次查询、分布不均 +# -------------------------------------------------------------------------- + + +async def test_the_reported_run_shape_end_to_end(monkeypatch, store_db): + counts = {key: count for _, _, key, count, _, _ in AGENTS} + + def outcome(key, attempt, budget): + if key == "coupon": + raise length_error(budget) # 两次都截断 + if key == "member" and attempt == 1: + raise length_error(budget) # 截断一次，加大额度后成功 + + world = World(monkeypatch, store_db, counts=counts, drafts=[ABSENT, COVERED], outcome=outcome) + row = await finish(graph()) + assert row.status == "succeeded", row.error + evts = await events(row.id) + + # 1. 抽取：会员专项截断后重试成功；卡券核销两次都截断，字段为空值 + assert world.extract_calls["member"] == [16384, 32768] + assert world.extract_calls["coupon"] == [16384, 32768] + assert world.extract_calls["store"] == world.extract_calls["campaign"] == [16384] + member = await output_of(row.id, "agent_member") + assert member["data"]["first_store"] == world.first_rows["member"][0] + assert member["data_evidence"]["first_store"]["status"] == "verified" + coupon = await output_of(row.id, "agent_coupon") + assert coupon["data"] == {"first_store": None, "first_value": None, "churn_rate": None} + [cut] = logs(evts, "agent_field_truncated") + assert cut["fields"] == ["first_store", "first_value", "churn_rate"] and "推理阶段就用完了" in cut["message"] + assert not logs(evts, "agent_field_unverified") + + # 2. 出处清单：预算内，每个 Agent 都有代表，被省略的有编号和摘要，开头说明各来源省略了几次 + first = world.writer[0] + text = catalog_text(first) + assert len(text) <= DEFAULT_CATALOG_BUDGET, len(text) + head = text[:900] + assert "目录有省略" in head + for _, title, _, count, table, _ in AGENTS: + assert f"{title} {count} 次" in head + group = text[text.index(f"来自「{title}」的查询"):] + group = group[:group.index("\n\n", 1)] if "\n\n" in group[1:] else group + assert f"共 {count} 次" in group and re.search(r"\n {2}r0：", group), f"「{title}」没有展开的查询" + briefs = re.findall(r"\n {2}- (Q\d+)：(\w+)，5 行", text) + assert briefs, "没有列出被省略的查询" + assert {t for _, t in briefs} <= {a[4] for a in AGENTS} + assert all(f"Q{n}：" in text for n in range(1, 165)), "每次查询都要在清单里（展开的或摘要）" + # 结构化结论排在查询前面；抽取失败的卡券核销写明结构化结论为空、不代表没有数据 + assert text.index("各 Agent 交来的结构化结论") < text.index("来自「") + assert "「卡券核销」：结构化结论为空" in text + # 写作规则里有缺失断言那一条 + assert ABSENCE_RULE in next(m.content for m in first if isinstance(m, HumanMessage)) + [trimmed] = logs(evts, "report_catalog_trimmed") + assert trimmed["queries"] == 164 and trimmed["budget"] == DEFAULT_CATALOG_BUDGET + assert [n["title"] for n in sorted(trimmed["nodes"], key=lambda n: n["title"])] == \ + sorted(a[1] for a in AGENTS) + assert trimmed["omitted"] == sum(n["omitted"] for n in trimmed["nodes"]) > 0 + + # 3. 缺失断言：认出来、交回重写一次；改成「本报告未覆盖」后通过 + assert len(world.writer) == 2 + request = str(world.writer[1][-1].content) + assert "本报告未覆盖" in request and "没有查到" in request + [repair] = logs(evts, "report_repair") + assert "数据缺失" in repair["message"] or "没有查到" in repair["message"] + [checked] = [e.data for e in evts if e.type == "report.checked"] + assert not [v for v in checked["violations"] if v["code"] == "absence_claim"] + assert not logs(evts, "report_absence_claim") + + +# -------------------------------------------------------------------------- +# 重写后仍然写缺失断言：只标注、不判失败 +# -------------------------------------------------------------------------- + +SMALL = {"member": 6, "store": 4, "coupon": 12, "campaign": 8} + + +async def test_an_absence_claim_that_survives_the_rewrite_is_flagged_not_failed(monkeypatch, store_db): + monkeypatch.setattr(report_mod, "catalog_budget", lambda window: 4000) + world = World(monkeypatch, store_db, counts=SMALL, drafts=[ABSENT], outcome=lambda *a: None) + row = await finish(graph(on_violation="fail", max_repairs=3)) + assert row.status == "succeeded", row.error + assert len(world.writer) == 2, "缺失断言只交回重写一次，不按最多重写次数反复重写" + evts = await events(row.id) + [checked] = [e.data for e in evts if e.type == "report.checked"] + assert checked["failed"] is False + claims = [v for v in checked["violations"] if v["code"] == "absence_claim"] + assert {v["text"] for v in claims} == {"数据缺失", "没有查到"} + assert all(v.get("segment") and v.get("context") for v in claims), "带着片段和原句：证据面板能定位" + [flag] = logs(evts, "report_absence_claim") + assert flag["level"] == "warn" and flag["count"] == 2 and flag["texts"] == ["数据缺失", "没有查到"] + assert "本报告未覆盖" in flag["message"] or "依据不足" in flag["message"] + + +async def test_without_omissions_an_absence_statement_is_left_alone(monkeypatch, store_db): + """目录全给了：「没有查到」可能就是看了数据说的，不查。""" + counts = {"member": 1, "store": 1, "coupon": 1, "campaign": 1} + world = World(monkeypatch, store_db, counts=counts, drafts=[ABSENT], outcome=lambda *a: None) + row = await finish(graph()) + assert row.status == "succeeded", row.error + assert len(world.writer) == 1 + evts = await events(row.id) + assert not logs(evts, "report_catalog_trimmed") and not logs(evts, "report_absence_claim") + prompt = next(m.content for m in world.writer[0] if isinstance(m, HumanMessage)) + assert ABSENCE_RULE not in prompt and "目录有省略" not in prompt diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 37388a16..b47b1eec 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -203,6 +203,19 @@ export function evidenceTally(cited: number, total: number, other = 0): string { return parts.join(' · ') } +/** + * 缺失断言：证据目录有省略时，报告写了「没有查到」「不存在」「缺失」这类结论（违规 absence_claim，重写后仍有时 + * 报告节点另发日志 report_absence_claim）。证据面板的违规清单和运行时间线共用这一套说法 + */ +export const ABSENCE_CLAIM_TEXT = { + label: '缺失断言', + hint: '写作模型没有看到全部证据：被省略的查询仍在证据台账里，看不到不等于不存在，这类结论应写作「本报告未覆盖」。' + + '报告照常产出，这一处只标注、不判失败', + title: (n?: number) => `报告写了${n ? ` ${formatNumber(n)} 处` : ''}缺失断言，已标注（证据目录有省略）`, + next: '被省略的查询仍在证据台账里，可在证据面板中查看。请在报告撰写节点的指令中要求未展开的方向写作「本报告未覆盖」，' + + '或换用上下文窗口更大的写作模型后重新运行', +} + /** 证据面板和证据条的文案 */ export const EVIDENCE_TEXT = { allCited: (n: number) => `${formatNumber(n)} 个数字都有出处`, diff --git a/frontend/src/run/EvidencePanel.tsx b/frontend/src/run/EvidencePanel.tsx index 00a35156..b4b9dd14 100644 --- a/frontend/src/run/EvidencePanel.tsx +++ b/frontend/src/run/EvidencePanel.tsx @@ -18,7 +18,7 @@ import { } from '../lib/evidence' import { humanizeError } from '../lib/errors' import { formatDateTime, formatNumber, NONE, shortId } from '../lib/format' -import { EVIDENCE_TEXT, JUDGE_TEXT, MERGE_TEXT, NODE_TYPE_LABEL, RUN_SQL_CHECK_TEXT, SQL_CHECK_TEXT, nodeTypeLabel } from '../lib/terms' +import { ABSENCE_CLAIM_TEXT, EVIDENCE_TEXT, JUDGE_TEXT, MERGE_TEXT, NODE_TYPE_LABEL, RUN_SQL_CHECK_TEXT, SQL_CHECK_TEXT, nodeTypeLabel } from '../lib/terms' import { useCatalog } from '../store/catalog' import { askKey, segmentKey, useEvidence, useExplore, useVerdicts } from '../store/evidence' import type { @@ -1552,6 +1552,7 @@ const CODE_LABEL: Record = { unresolved_ref: '引用无法解析', unknown_entity: '疑似不存在的名称', unverified_entity: '无法核实的名称', + absence_claim: ABSENCE_CLAIM_TEXT.label, } /** 违规清单：报告撰写节点核对出来的全部问题，画得出线的能定位，画不出的说清在哪 */ @@ -1563,7 +1564,9 @@ function ViolationList({ doc, onLocate }: { doc: EvidenceDocData; onLocate: (seg

{EVIDENCE_TEXT.violationsHint}