From 8a5f96be4b29adfc091e49c7d36987089368bd2d Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 13:12:24 +0800 Subject: [PATCH] =?UTF-8?q?=E5=87=BA=E5=8F=A3=E5=A5=91=E7=BA=A6=EF=BC=9A?= =?UTF-8?q?=E5=88=86=E6=9E=90=E8=AF=B4=E6=B3=95=E5=87=BA=E5=8F=A3=E7=8B=AC?= =?UTF-8?q?=E7=AB=8B=E5=A4=8D=E6=A0=B8=E3=80=81=E6=8C=87=E6=A0=87=E5=8F=AF?= =?UTF-8?q?=E6=AF=94=E6=80=A7=E3=80=81=E5=BC=95=E7=94=A8=E6=A8=A1=E5=BC=8F?= =?UTF-8?q?=E7=9A=84=E8=A1=80=E7=BC=98=E5=8F=A3=E5=AD=90=E3=80=82=E5=BC=95?= =?UTF-8?q?=E7=94=A8=E6=A8=A1=E5=BC=8F=E4=B8=8B=20verify=5Fdoc=20=E6=8C=89?= =?UTF-8?q?=E6=96=87=E6=A1=A3=E8=AE=B0=E4=B8=8B=E7=9A=84=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E9=A1=B9=E5=86=8D=E6=A0=B8=E4=B8=80=E9=81=8D=E5=88=86=E6=9E=90?= =?UTF-8?q?=E8=AF=B4=E6=B3=95=EF=BC=9A=E5=A4=84=E7=BD=AE=E5=8F=96=E6=96=87?= =?UTF-8?q?=E6=A1=A3=E5=92=8C=E6=8A=A5=E5=91=8A=E8=8A=82=E7=82=B9=E4=BA=A7?= =?UTF-8?q?=E5=87=BA=E9=87=8C=E6=9B=B4=E4=B8=A5=E7=9A=84=E9=82=A3=E4=B8=AA?= =?UTF-8?q?=EF=BC=8Cstrict=20=E6=9C=89=E8=BF=9D=E8=A7=84=E8=AE=B0=E4=B8=80?= =?UTF-8?q?=E6=9D=A1=E7=BC=BA=E5=8F=A3=EF=BC=88=E9=99=8D=E6=A1=A3=EF=BC=89?= =?UTF-8?q?=EF=BC=8C=E6=8A=A5=E5=91=8A=E8=8A=82=E7=82=B9=E6=8A=A5=E4=B8=8A?= =?UTF-8?q?=E6=9D=A5=E7=9A=84=E5=90=8C=E4=B8=80=E6=9D=A1=E4=B8=8D=E9=87=8D?= =?UTF-8?q?=E5=A4=8D=E6=94=B6=EF=BC=8Ccheck=20=E5=8F=AA=E5=86=99=E8=BF=9B?= =?UTF-8?q?=E5=87=BA=E5=85=B7=E5=A3=B0=E6=98=8E=EF=BC=88analysis=5Fclaims?= =?UTF-8?q?=EF=BC=8Ccounted=20=E4=B8=BA=20false=EF=BC=89=EF=BC=8C=E8=A6=81?= =?UTF-8?q?=E6=B1=82=20strict=20=E5=8D=B4=E6=B2=A1=E6=9C=89=E6=A0=B8?= =?UTF-8?q?=E5=AF=B9=E8=AE=B0=E5=BD=95=E7=9A=84=E7=85=A7=E5=AE=9E=E8=AE=B0?= =?UTF-8?q?=E7=BC=BA=E5=8F=A3=EF=BC=8C=E8=80=81=E6=96=87=E6=A1=A3=E4=B8=8D?= =?UTF-8?q?=E6=A0=B8=E3=80=82=E5=8F=A3=E5=BE=84=E5=8D=A1=E6=8C=87=E6=A0=87?= =?UTF-8?q?=E4=B8=8A=E7=9A=84=E5=8F=AF=E6=AF=94=E6=80=A7=EF=BC=88MetricCom?= =?UTF-8?q?parability=EF=BC=89=EF=BC=9A=E6=8A=A5=E5=91=8A=E5=BC=95?= =?UTF-8?q?=E7=94=A8=E4=BA=86=E7=9A=84=EF=BC=88=E5=BC=95=E7=94=A8=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=E6=8C=89=E6=96=87=E6=A1=A3=EF=BC=8C=E5=88=AB=E7=9A=84?= =?UTF-8?q?=E6=8C=89=E6=95=B0=E5=AD=97=E5=9B=9E=E6=8C=87=EF=BC=89=E5=92=8C?= =?UTF-8?q?=E5=A5=91=E7=BA=A6=E5=BF=85=E9=9C=80=E7=9A=84=E6=8C=87=E6=A0=87?= =?UTF-8?q?=E4=B8=A4=E6=9C=9F=E4=B8=8D=E5=8F=AF=E6=AF=94=E8=AE=B0=E7=BC=BA?= =?UTF-8?q?=E5=8F=A3=EF=BC=8C=E6=9C=89=E6=9D=A1=E4=BB=B6=E5=8F=AF=E6=AF=94?= =?UTF-8?q?=E7=9A=84=E5=8F=AA=E5=86=99=E8=BF=9B=E5=87=BA=E5=85=B7=E5=A3=B0?= =?UTF-8?q?=E6=98=8E=EF=BC=88comparability=EF=BC=89=E3=80=82=E8=A1=80?= =?UTF-8?q?=E7=BC=98=EF=BC=88D15=EF=BC=89=EF=BC=9A=E6=94=B6=E9=BD=90?= =?UTF-8?q?=E6=8A=A5=E5=91=8A=E5=BC=95=E7=94=A8=E7=9A=84=E6=8C=87=E6=A0=87?= =?UTF-8?q?=EF=BC=88[[m:]]=20=E7=89=87=E6=AE=B5=E5=92=8C=E5=8F=A5=E6=9C=AB?= =?UTF-8?q?=E4=BE=9D=E6=8D=AE=E9=87=8C=E7=9A=84=20m:=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E6=8C=89=E5=8D=A1=E4=B8=8A=E8=AE=B0=E7=9A=84=E8=BE=93=E5=85=A5?= =?UTF-8?q?=E8=AE=A4=E8=BF=BD=E4=B8=8D=E5=88=B0=E5=8E=9F=E5=A7=8B=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E7=9A=84=EF=BC=8C=E5=8F=97=E7=AE=A1=E7=BA=A7=E5=88=AB?= =?UTF-8?q?=E7=9A=84=E6=AD=A3=E5=BC=8F=E8=BF=90=E8=A1=8C=E8=AE=B0=E7=BC=BA?= =?UTF-8?q?=E5=8F=A3=EF=BC=8C=E5=85=B6=E4=BD=99=E5=86=99=E8=BF=9B=E5=87=BA?= =?UTF-8?q?=E5=85=B7=E5=A3=B0=E6=98=8E=E7=9A=84=20untraced=EF=BC=88counted?= =?UTF-8?q?=20=E4=B8=BA=20false=EF=BC=89=EF=BC=9B=5Funtraced=5Fgaps=20?= =?UTF-8?q?=E6=8B=86=E5=87=BA=E5=8F=AF=E4=BC=A0=E5=AD=90=E9=9B=86=E7=9A=84?= =?UTF-8?q?=20=5Funtraced=E3=80=82=E7=8E=B0=E6=9C=89=E5=8F=97=E7=AE=A1?= =?UTF-8?q?=E6=AD=A3=E5=BC=8F=E8=BF=90=E8=A1=8C=E7=9A=84=E4=B8=A4=E6=9D=A1?= =?UTF-8?q?=E7=94=A8=E4=BE=8B=EF=BC=88=E8=A3=81=E5=88=A4=E5=88=A4=E6=A1=A3?= =?UTF-8?q?=EF=BC=89=E6=A1=A3=E4=BD=8D=E4=B8=8D=E5=8F=98=E3=80=81=E7=BC=BA?= =?UTF-8?q?=E5=8F=A3=E9=87=8C=E5=A4=9A=E4=B8=80=E6=9D=A1=E8=A1=80=E7=BC=98?= =?UTF-8?q?=EF=BC=8C=E6=96=AD=E8=A8=80=E4=B8=8D=E6=B6=89=E5=8F=8A=E7=BC=BA?= =?UTF-8?q?=E5=8F=A3=E5=88=97=E8=A1=A8=EF=BC=8C=E4=B8=8D=E7=94=A8=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/nodes/io.py | 158 ++++++++- .../tests/test_contract_analysis_claims.py | 301 ++++++++++++++++++ 2 files changed, 450 insertions(+), 9 deletions(-) create mode 100644 backend/tests/test_contract_analysis_claims.py diff --git a/backend/app/engine/nodes/io.py b/backend/app/engine/nodes/io.py index 38ed2116..dec7fd82 100644 --- a/backend/app/engine/nodes/io.py +++ b/backend/app/engine/nodes/io.py @@ -236,6 +236,15 @@ def _apply_contract( ) unmatched = trace.unmatched if trace else [] + # 口径卡指标上的可比性（口径卡跨两份同形查询算环比、同比时自动查的）：报告引用了的（引用模式按文档，别的按数字回指上 + # 的指标）、契约必需的指标两期不可比，记缺口降档；有条件可比的写进出具声明作标注 + required = {str(r) for r in contract.get("required") or []} + matched_ids = {str(h.get("metric")) for h in (trace.matched if trace else []) if h.get("metric")} + compared_gaps, comparability = _comparability( + [*(cited.get("cited_metrics") or []), + *(m for m in metrics if str(m.get("id")) in required | (set() if citations else matched_ids))]) + gaps.extend(compared_gaps) + # 同一条查询算出几个指标、报告又引用了它的格，只记一句（按出问题的那条查询归并，列出受影响的引用） titles = {n.id: n.title for n in ctx.run.spec.nodes} aliases: dict[str, str] = cited.get("aliases") or {} @@ -271,8 +280,9 @@ def _apply_contract( if isinstance(payload, dict) and payload.get("exhausted"): gaps.append(f"协作团队「{titles.get(node_id, node_id)}」用完 {payload.get('rounds', '?')} " "轮仍未完成，交付的是成员最后的回复") - # 上游节点自己报上来的缺口（双盲复算比不上、反方审稿的反驳没回应……）：只会降档，不会升档 - gaps.extend(_upstream_issuance_gaps(state, ctx)) + # 上游节点自己报上来的缺口（双盲复算比不上、反方审稿的反驳没回应……）：只会降档，不会升档。报告节点报的分析说法那一条， + # 引用模式下出口自己复核过、记了自己的缺口的不再重复收 + gaps.extend(_upstream_issuance_gaps(state, ctx, skip=cited.get("analysis_skip"))) tier = decide_tier( missing_required=missing_required, @@ -290,7 +300,7 @@ def _apply_contract( return _declare_citations(tier, cited, calibers=calibers, metrics=metrics, gaps=gaps, missing_required=missing_required, missing_expected=missing_expected, report_from=str(report_from), declared_at=datetime.now(timezone.utc), - ctx=ctx, sql_checks=sql_checks) + ctx=ctx, sql_checks=sql_checks, comparability=comparability) # 逐个数字的出处。以前只有一个计数，界面标不出「这个数来自哪个指标」 matched = [ @@ -311,6 +321,8 @@ def _apply_contract( "gaps": gaps, "declared_at": datetime.now(timezone.utc).isoformat(), **({"sql_checks": sql_checks} if sql_checks else {}), + # 指标的可比性（有条件可比的标注、不可比的同时在缺口里）：只在有的时候带这个键 + **({"comparability": comparability} if comparability else {}), } ctx.emit( EventType.ISSUANCE, @@ -343,8 +355,9 @@ def _shown_text(fields: dict[str, Any]) -> str: return "\n".join(parts) -def _untraced_gaps(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[str]: - """指标出具的缺口：输入没追到原始数据的指标，每个一句，说清停在哪个节点（engine/lineage 的认法）。""" +def _untraced(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[dict[str, Any]]: + """输入没追到原始数据的指标（engine/lineage 的认法）：[{metric, name, stops: [{path, node_id}], gap}]。gap 是缺口那一句， + 说清停在哪个节点。metrics 可以是契约收来的全部指标（指标出具），也可以只是报告引用的那几个（引用模式，D15）。""" from app.engine.lineage import untraced titles = {n.id: n.title for n in ctx.run.spec.nodes} @@ -357,11 +370,45 @@ def _untraced_gaps(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[str] where = "、".join(dict.fromkeys( f"{s.get('path')}（停在「{titles.get(at, at or '未知来源')}」）" for s in stops[:3] for at in [str(s.get('node_id') or '').split('/', 1)[0]])) - out.append(f"指标「{m.get('name') or m.get('id')}」的输入没追到原始数据：{where}") + out.append({"metric": m.get("id"), "name": m.get("name") or m.get("id"), + "stops": [{"path": s.get("path"), "node_id": s.get("node_id")} for s in stops[:3]], + "gap": f"指标「{m.get('name') or m.get('id')}」的输入没追到原始数据：{where}"}) return out -def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext) -> list[str]: +def _untraced_gaps(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[str]: + """指标出具的缺口：输入没追到原始数据的指标，每个一句（_untraced）。""" + return [u["gap"] for u in _untraced(metrics, ctx)] + + +def _comparability(metrics: list[dict[str, Any]]) -> tuple[list[str], list[dict[str, Any]]]: + """口径卡指标上的可比性（contract.MetricComparability，口径卡跨两份同形查询算环比、同比时自动做的检查）： + (缺口, 出具声明里的那一项)。metrics 是要看的指标（报告引用了的、契约必需的，调用方挑好）：两期不可比的记缺口（降档， + 同 sql_check_failed），有条件可比的只标注。没有可比性记录的指标不在里面；同一条指标只看一次。""" + gaps: list[str] = [] + declared: list[dict[str, Any]] = [] + seen: set[int] = set() + for m in metrics: + mid = str(m.get("id") or "") if isinstance(m, dict) else "" + found = m.get("comparability") if mid else None + if not isinstance(found, dict) or id(m) in seen: + continue + seen.add(id(m)) + verdict = found.get("verdict") + if verdict not in ("conditional", "incomparable"): + continue + name = m.get("name") or mid + phrase = str(found.get("phrase") or "").strip() + declared.append({"metric": mid, "name": name, "verdict": verdict, "phrase": phrase, + "reasons": [str(r) for r in found.get("reasons") or [] if r], + **({"artifact": found["artifact"]} if found.get("artifact") else {})}) + if verdict == "incomparable": + gaps.append(f"指标「{name}」{phrase or '两期不可比'}") + return gaps, declared + + +def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext, + skip: dict[str, set[str]] | None = None) -> list[str]: """上游节点产出里的 issuance_gaps（字符串列表），每条前面加上是哪个节点说的。 这是给「把关类」节点留的通用口子：节点觉得这次的结论不该完整出具（双盲复算两路对不上、 @@ -372,6 +419,9 @@ def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext) -> list[str]: 只认这个成果节点的上游：另一条支路上的节点管不到这份成果。不是字符串、空白的条目跳过—— 这个键谁都能写（数据整形的 JSON 模板也行），写坏了只当没写，不能让出具崩掉。 + + skip：{节点 id: 不再收的那几条}。引用模式的出口自己复核过报告文档的分析说法、记了自己的缺口时，报告节点报上来的同一件事 + 不再重复收一条（_analysis_claims）。 """ from app.engine.labels import type_label @@ -388,7 +438,9 @@ def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext) -> list[str]: continue kind = type_label(node.type) prefix = f"{kind}：" if node.title == kind else f"{kind}「{node.title}」：" - out.extend(prefix + gap.strip() for gap in reported if isinstance(gap, str) and gap.strip()) + dropped = (skip or {}).get(node.id) or set() + out.extend(prefix + gap.strip() for gap in reported + if isinstance(gap, str) and gap.strip() and gap not in dropped) return out @@ -418,12 +470,14 @@ def _check_citations( governed：受管级别的正式运行。可疑实体（名字哪里都找不到）只在这时按缺口降档，别处只标注。 """ from app.core.artifact_store import load + from app.engine import analysis_claims as said from app.engine.evidence import iter_units, ledger_enabled, resolve_ref, verify_doc from app.engine.nodes.report import report_catalog, report_entities out: dict[str, Any] = {"matched": [], "unmatched": [], "unresolved": [], "gaps": [], "doc_artifact": None, "stats": None, "claims_policy": None, "uncited": None, - "unsupported": None, "claims": None, "entities": None, "queries": {}, "aliases": {}} + "unsupported": None, "claims": None, "entities": None, "queries": {}, "aliases": {}, + "analysis": None, "analysis_skip": {}, "cited_metrics": [], "untraced": None} gaps = out["gaps"] spec = ctx.run.spec node = spec.node_map().get(report_from) @@ -458,7 +512,11 @@ def _check_citations( broken: list[str] = [] flagged: set[str] = set() suspicious: dict[str, list[dict[str, Any]]] = {"unknown_entity": [], "unverified_entity": []} + said_found: list[dict[str, Any]] = [] # 分析说法的违规（只有记了核对这一项的文档才有，_analysis_claims） for v in checked["violations"]: + if v["code"] in said.CODES: + said_found.append(v) + continue where = {k: v[k] for k in ("segment", "unit") if v.get(k)} if v.get("span"): where.update(start=v["span"][0], end=v["span"][1]) @@ -519,6 +577,16 @@ def _check_citations( _claims(out, contract, payload, evidence_on=ledger_enabled(ctx.run), governed=governed, uncited=checked.get("uncited") or [], doc=doc, title=node.title) + _analysis_claims(out, doc, payload, said_found, title=node.title, node_id=str(report_from)) + + # 报告引用的指标（[[m:]] 片段和句末依据里的 m:）：可比性按它们看（_apply_contract）；输入追不到原始数据的，受管级别的 + # 正式运行记缺口（降档，和证据面板「追不到原始数据」的说法一致：代码节点自称取数也不能完整出具），其余写进出具声明 + out["cited_metrics"] = _cited_metrics(doc, catalog, state) + if stray := _untraced(out["cited_metrics"], ctx): + if governed: + gaps.extend(u["gap"] for u in stray) + out["untraced"] = {"metrics": [{k: u[k] for k in ("metric", "name", "stops")} for u in stray[:_LISTED]], + "counted": governed} mine = next((r for r in reports if r["node"].id == report_from), None) edited = mine["edited"] if mine else [] @@ -858,6 +926,70 @@ def _claims(out: dict[str, Any], contract: dict[str, Any], payload: dict[str, An } +#: 报告节点 analysis_claims 的三种处置，从松到严 +_ANALYSIS_ORDER = ("off", "check", "strict") + + +def _analysis_claims(out: dict[str, Any], doc: dict[str, Any], payload: Any, found: list[dict[str, Any]], *, + title: str, node_id: str) -> None: + """分析说法核对的出口独立复核（00-总方案 §3.5、D9）：found 是 verify_doc 这次重新核出的说法违规（只有文档记了 + analysis_claims 那一项才核，老文档复核结果不变）。 + + 处置取报告文档记下的和报告节点产出里写的更严的那个：strict 有违规就记一条缺口（降档），报告节点报上来的同一条 + （产出的 analysis_gap）不再重复收；check 只写进出具声明作标注、不影响档位。要求 strict、文档里却没有核对记录的 + （不该发生），照实记一条缺口：出口替不了报告节点去核。结果放进 out["analysis"]（出具声明的 analysis_claims）， + strict 和有违规时才有。""" + from app.engine import analysis_claims as said + + found_mode = said.recorded(doc) + written = payload.get("analysis_claims") if isinstance(payload, dict) else None + modes = [m for m in (found_mode[0] if found_mode else None, written) if m in _ANALYSIS_ORDER] + mode = max(modes, key=_ANALYSIS_ORDER.index) if modes else None + if mode in (None, "off"): + return + if found_mode is None: + if mode == "strict": + out["gaps"].append(f"报告「{title}」要求严格核对分析说法，但文档中没有分析说法的核对记录，分析说法未核对") + return + counted = mode == "strict" + if found and counted: + out["gaps"].append(f"报告「{title}」有 {len(found)} 处分析说法未通过核对（{said.summary(found)}）：" + "判断要挂对应的分析结论并照判定写，原因要有知识库原话或运行输入作依据") + if isinstance(payload, dict) and isinstance(payload.get("analysis_gap"), str): + out["analysis_skip"] = {node_id: {payload["analysis_gap"]}} + if found or counted: + keep = ("code", "text", "claim", "op", "analyses", "unit", "segment", "span", "message", "context") + out["analysis"] = {"mode": mode, "count": len(found), "counts": said.counts(found), "counted": counted, + "violations": [{k: v[k] for k in keep if v.get(k) not in (None, "", [])} + for v in found[:_LISTED]]} + + +def _cited_metrics(doc: dict[str, Any], catalog: dict[str, Any], state: GraphState) -> list[dict[str, Any]]: + """报告引用了的指标（[[m:]] 片段和句末 [[see:m:…]]）：口径卡产出里的那一条（带 inputs、comparability），按引用的先后、 + 同一张卡的同一个指标只算一次。目录里认不出、卡的产出里找不到的跳过。""" + from app.engine.evidence import iter_units + + nodes = state.get("nodes") or {} + out: list[dict[str, Any]] = [] + seen: set[tuple[str, str]] = set() + for _, unit in iter_units(doc): + cites = [s.get("cite") for s in unit.get("segments") or [] if s.get("ref") and not s.get("auto")] + for cite in [*cites, *(unit.get("see") or [])]: + if not isinstance(cite, dict) or cite.get("kind") != "metric" or cite.get("status") != "resolved": + continue + entry = catalog.get(cite.get("alias")) or {} + card, mid = str(entry.get("node_id") or ""), str((entry.get("locator") or {}).get("metric") or "") + if entry.get("kind") != "metric" or not card or not mid or (card, mid) in seen: + continue + seen.add((card, mid)) + payload = nodes.get(card) + metric = next((m for m in (payload.get("metrics") or []) if isinstance(m, dict) and str(m.get("id")) == mid), + None) if isinstance(payload, dict) else None + if metric is not None: + out.append(metric) + return out + + def _cited_no_data(doc: dict[str, Any], catalog: dict[str, Any]) -> list[str]: """报告直接引用的格里，落在「查询区间里这张表没有数据」的列上的（查询快照的 no_data，data/coverage.py）：按列一句缺口 「报告引用的 Q1「上周到访人数」：所依据的查询区间里这张表没有数据（…）」。读不出来的快照不算。""" @@ -933,6 +1065,7 @@ def _declare_citations( tier: str, cited: dict[str, Any], *, calibers: list[dict[str, str]], metrics: list[dict[str, Any]], gaps: list[str], missing_required: list[str], missing_expected: list[str], report_from: str, declared_at: Any, ctx: NodeContext, sql_checks: list[dict[str, Any]] | None = None, + comparability: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """引用模式的 _issuance。旧字段一个不少（老前端照常显示），另加 mode、report、unresolved。""" # 节点 id 和报告文档、report.checked 事件上记的同一个（这次运行里的，子工作流里的带前缀） @@ -957,6 +1090,10 @@ def _declare_citations( **({"entities": cited["entities"]} if cited.get("entities") else {}), # 没通过 SQL 检查的查询（结构化）：报告页据此在显眼处说明降档原因。没有就不加这个键 **({"sql_checks": sql_checks} if sql_checks else {}), + # 分析说法核对（strict，或者有违规时）、指标的可比性、报告引用的指标里追不到原始数据的：同样只在有的时候带 + **({"analysis_claims": cited["analysis"]} if cited.get("analysis") else {}), + **({"comparability": comparability} if comparability else {}), + **({"untraced": cited["untraced"]} if cited.get("untraced") else {}), } entities = cited.get("entities") ctx.emit( @@ -983,6 +1120,9 @@ def _declare_citations( **({"entities": {"unknown": (cited["stats"] or {}).get("unknown_entities", len(entities["unknown"])), "unverified": (cited["stats"] or {}).get("unverified_entities", len(entities["unverified"])), "counted": entities["counted"]}} if entities else {}), + # 分析说法：处置、几处、记没记缺口（名单只在出具声明里） + **({"analysis_claims": {k: cited["analysis"][k] for k in ("mode", "count", "counted")}} + if cited.get("analysis") else {}), ) return issuance diff --git a/backend/tests/test_contract_analysis_claims.py b/backend/tests/test_contract_analysis_claims.py new file mode 100644 index 00000000..b03793f9 --- /dev/null +++ b/backend/tests/test_contract_analysis_claims.py @@ -0,0 +1,301 @@ +"""出口契约（引用模式）的分析说法复核、口径卡指标的可比性、报告引用的指标的血缘（00-总方案 §3.4、§3.5、§3.9，D9、D15）。 + +- 分析说法：出口按报告文档记下的那一项独立再核一遍（verify_doc）。strict 有违规记一条缺口（降档），报告节点报上来的同一条 + 不重复收；check 只写进出具声明（analysis_claims，counted 为 false），不影响档位；没记这一项的老文档不核 +- 可比性：报告引用了的、契约必需的指标两期不可比记缺口（降档），有条件可比的只写进出具声明（comparability） +- 血缘：报告引用的指标（[[m:]] 片段、句末 [[see:m:…]]）输入追不到原始数据的，受管级别的正式运行记缺口，其余写进出具声明的 + untraced（counted 为 false） +""" +from __future__ import annotations + +import asyncio +import sqlite3 +import uuid + +import pytest +from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessage +from sqlalchemy import select + +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import DataSource, Run, RunEvent, Workflow, WorkflowVersion +from app.engine import analysis_claims as ac +from app.engine import compiler +from app.engine.nodes.io import _analysis_claims, _comparability +from app.engine.runner import run_manager +from app.engine.schema import NodeType +from app.main import app + +METRICS = [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"}, + {"id": "wow", "name": "环比增幅", "unit": "%", "decimals": 1, "format": "plain", "expression": "vars.kpi.wow"}] +SOURCE = "本周销售额 [[m:gmv]]，下降主要来自华东。[[see:m:gmv]]" +PLAIN = "本周销售额 [[m:gmv]]。[[see:m:gmv]]" + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def weekly(*, report=None, contract=None, fetch=None, metrics=METRICS) -> dict: + """取数 → 口径卡 → 报告撰写 → 出口（引用模式契约）。缺省取数是数据整形（追不到原始数据）。""" + nodes = [ + node("start", "input", fields=[{"name": "week", "default": "2026-W37"}]), + fetch or node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5, 'wow': -3.1}", + assign_to="kpi"), + node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=metrics), + node("write", "report", instructions="写周报", **(report or {})), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], + contract={"report_from": "write", "metrics_from": ["caliber"], "required": ["gmv"], "strict": True, + **(contract or {})}), + ] + ids = [n["id"] for n in nodes] + return {"nodes": nodes, "edges": [{"source": a, "target": b} for a, b in zip(ids, ids[1:])]} + + +def writes(monkeypatch, text: str) -> None: + from app.providers import mock_model + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text)) + + +async def wait(run_id: str) -> Run: + for _ in range(400): + 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 issued(graph, *, run_class="exploratory") -> dict: + run = await run_manager.start(graph=graph, input_payload={}, run_class=run_class) + row = await wait(run.id) + assert row.status == "succeeded", row.error + return row.output["_issuance"] + + +async def governed(client, graph) -> dict: + """把这张图发布成受管级别，从发布版本发起正式运行。""" + async with SessionLocal() as session: + wf = Workflow(name=f"分析-{uuid.uuid4().hex[:6]}", graph=graph, status="governed", published_version=1) + session.add(wf) + await session.flush() + session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=graph)) + await session.commit() + wf_id = wf.id + res = await client.post("/api/runs", json={"workflow_id": wf_id, "run_class": "formal"}) + assert res.status_code == 201, res.text + row = await wait(res.json()["id"]) + assert row.status == "succeeded", row.error + return row.output["_issuance"] + + +async def issuance_event(run_id: str) -> dict: + async with SessionLocal() as session: + [event] = (await session.execute(select(RunEvent).where( + RunEvent.run_id == run_id, RunEvent.type == "issuance"))).scalars() + return event.data + + +def said_gaps(issuance: dict) -> list[str]: + return [g for g in issuance["gaps"] if "分析说法" in g] + + +# -------------------------------------------------------------------------- +# 分析说法：出口独立复核 +# -------------------------------------------------------------------------- + + +async def test_strict_leftovers_degrade_with_one_gap_from_the_exit(monkeypatch): + writes(monkeypatch, SOURCE) + issuance = await issued(weekly(report={"analysis_claims": "strict"})) + assert issuance["tier"] == "degraded" + [gap] = said_gaps(issuance) + assert gap.startswith("报告「write」有 1 处分析说法未通过核对（1 处判断没挂分析结论）"), "报告节点那一条不重复收" + declared = issuance["analysis_claims"] + assert declared["mode"] == "strict" and declared["counted"] is True and declared["count"] == 1 + [v] = declared["violations"] + assert v["code"] == ac.UNCITED and v["text"] == "主要来自" and v["claim"] == "source" and v["span"] + + +async def test_check_leftovers_are_declared_but_do_not_change_the_tier(monkeypatch): + writes(monkeypatch, SOURCE) + issuance = await issued(weekly()) + assert issuance["tier"] == "formal" and said_gaps(issuance) == [] + declared = issuance["analysis_claims"] + assert declared["mode"] == "check" and declared["counted"] is False and declared["counts"] == {ac.UNCITED: 1} + + +async def test_a_clean_report_declares_nothing_under_check_and_a_clean_slate_under_strict(monkeypatch): + writes(monkeypatch, PLAIN) + assert "analysis_claims" not in await issued(weekly()) + issuance = await issued(weekly(report={"analysis_claims": "strict"})) + assert issuance["tier"] == "formal" and issuance["analysis_claims"] == { + "mode": "strict", "count": 0, "counts": {}, "counted": True, "violations": []} + + +async def test_off_and_old_documents_are_not_checked(monkeypatch): + writes(monkeypatch, SOURCE) + issuance = await issued(weekly(report={"analysis_claims": "off"})) + assert "analysis_claims" not in issuance and said_gaps(issuance) == [] and issuance["tier"] == "formal" + + +async def test_a_narrative_contract_still_takes_the_report_gap(monkeypatch): + writes(monkeypatch, SOURCE) + graph = weekly(report={"analysis_claims": "strict"}) + contract = graph["nodes"][-1]["data"]["config"]["contract"] + del contract["report_from"] + contract["narrative"] = "{{ nodes.write.text }}" + issuance = await issued(graph) + [gap] = said_gaps(issuance) + assert gap.startswith("报告撰写「write」：报告有 1 处分析说法未通过核对") and issuance["tier"] == "degraded" + + +def test_the_exit_policy_takes_the_stricter_of_the_doc_and_the_node(): + def run(doc, payload, found=()): + out = {"gaps": [], "analysis": None, "analysis_skip": {}} + _analysis_claims(out, doc, payload, list(found), title="周报", node_id="write") + return out + + violation = {"code": ac.UNCITED, "text": "主要来自", "claim": "source", "op": "decompose", "message": "…"} + # 老文档（没有这一项）、节点也没写：不核 + assert run({}, {"claims": "off"}) == {"gaps": [], "analysis": None, "analysis_skip": {}} + # 节点要求 strict，文档里却没有核对记录：照实记缺口 + assert run({}, {"analysis_claims": "strict"})["gaps"] == [ + "报告「周报」要求严格核对分析说法，但文档中没有分析说法的核对记录，分析说法未核对"] + assert run({}, {"analysis_claims": "check"})["gaps"] == [] + # 文档记的是 check、节点写的是 strict：按更严的 + out = run({ac.DOC_FIELD: ac.record("check")}, {"analysis_claims": "strict", "analysis_gap": "报告有 1 处…"}, + [violation]) + assert len(out["gaps"]) == 1 and out["analysis"]["counted"] is True + assert out["analysis_skip"] == {"write": {"报告有 1 处…"}} + + +# -------------------------------------------------------------------------- +# 可比性 +# -------------------------------------------------------------------------- + + +def test_comparability_gaps_only_for_incomparable_metrics(): + metrics = [ + {"id": "wow", "name": "环比增幅", "comparability": {"verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据", + "reasons": ["本期只有 3 天数据"], "artifact": "c" * 64}}, + {"id": "aov", "name": "客单价", "comparability": {"verdict": "conditional", + "phrase": "两期有条件可比：两段的周末天数不同：4 天对 6 天"}}, + {"id": "gmv", "name": "销售额", "comparability": {"verdict": "comparable", "phrase": "两期可比"}}, + {"id": "orders", "name": "订单数"}, + ] + gaps, declared = _comparability(metrics + metrics[:1]) + assert gaps == ["指标「环比增幅」两期不可比：本期只有 3 天数据"] + assert [(d["metric"], d["verdict"]) for d in declared] == [("wow", "incomparable"), ("aov", "conditional")] + assert declared[0]["reasons"] == ["本期只有 3 天数据"] and declared[0]["artifact"] == "c" * 64 + + +@pytest.fixture +def comparability(monkeypatch): + """口径卡给指标记上可比性（口径卡跨两份同形查询算环比时自动做的检查由入口包实现；这里直接记上）。""" + marks: dict[str, dict] = {} + original = compiler.RUNNERS[NodeType.METRICS] + + async def card(state, ctx): + updates = await original(state, ctx) + for metric in updates["nodes"][ctx.node.id]["metrics"]: + if metric["id"] in marks: + metric["comparability"] = marks[metric["id"]] + return updates + + monkeypatch.setitem(compiler.RUNNERS, NodeType.METRICS, card) + return marks + + +async def test_a_cited_incomparable_metric_degrades_and_a_conditional_one_is_noted(monkeypatch, comparability): + comparability["wow"] = {"verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据", + "reasons": ["本期只有 3 天数据"]} + writes(monkeypatch, "本周销售额 [[m:gmv]]，环比 [[m:wow]]。[[see:m:gmv,m:wow]]") + issuance = await issued(weekly()) + assert issuance["tier"] == "degraded" and "指标「环比增幅」两期不可比：本期只有 3 天数据" in issuance["gaps"] + assert [c["metric"] for c in issuance["comparability"]] == ["wow"] + comparability["wow"] = {"verdict": "conditional", "phrase": "两期有条件可比：两段的周末天数不同"} + issuance = await issued(weekly()) + assert issuance["tier"] == "formal" and issuance["comparability"][0]["verdict"] == "conditional" + + +async def test_an_incomparable_metric_nobody_cites_or_requires_is_left_alone(monkeypatch, comparability): + comparability["wow"] = {"verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据"} + writes(monkeypatch, PLAIN) + issuance = await issued(weekly()) + assert issuance["tier"] == "formal" and "comparability" not in issuance + # 契约必需的指标不引用也算 + issuance = await issued(weekly(contract={"required": ["gmv", "wow"]})) + assert "指标「环比增幅」两期不可比：本期只有 3 天数据" in issuance["gaps"] + + +# -------------------------------------------------------------------------- +# 血缘：报告引用的指标追没追到原始数据 +# -------------------------------------------------------------------------- + + +async def test_a_cited_metric_that_stops_at_a_transform_is_only_noted_outside_governed_runs(monkeypatch): + writes(monkeypatch, PLAIN) + issuance = await issued(weekly(), run_class="formal") + assert issuance["tier"] == "formal" and not any("没追到原始数据" in g for g in issuance["gaps"]) + untraced = issuance["untraced"] + assert untraced["counted"] is False and [m["metric"] for m in untraced["metrics"]] == ["gmv"] + assert untraced["metrics"][0]["stops"][0]["node_id"] == "fetch" + + +async def test_a_governed_formal_run_counts_the_untraced_cited_metric(client, monkeypatch): + writes(monkeypatch, PLAIN) + issuance = await governed(client, weekly()) + assert issuance["tier"] == "degraded" + [gap] = [g for g in issuance["gaps"] if "没追到原始数据" in g] + assert gap.startswith("指标「销售额」的输入没追到原始数据：vars.kpi.gmv（停在「fetch」）") + assert issuance["untraced"]["counted"] is True + # 没被引用的指标（环比增幅同样停在整形节点）不算：只看报告引用了的 + assert not any("环比增幅" in g for g in issuance["gaps"]) + + +@pytest.fixture +async def shop(tmp_path): + path = tmp_path / "shop.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, amount REAL);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(1, "2026-W37", 100.5), (2, "2026-W37", 200.0)]) + db.commit() + db.close() + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() + if row is None: + row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) + session.add(row) + else: + row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + yield + await engines.invalidate(source_id) + + +async def test_a_metric_read_from_a_query_cell_is_traced(client, monkeypatch, shop): + writes(monkeypatch, PLAIN) + fetch = node("fetch", "tool", tool="db_query__shop", args={"sql": "SELECT SUM(amount) AS gmv FROM orders"}) + graph = weekly(fetch=fetch, metrics=[{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, + "expression": "cell(nodes.fetch, 0, 'gmv')"}]) + issuance = await governed(client, graph) + assert "untraced" not in issuance and not any("没追到原始数据" in g for g in issuance["gaps"])