diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py index 5481382e..f2304dc0 100644 --- a/backend/app/engine/compare.py +++ b/backend/app/engine/compare.py @@ -134,7 +134,8 @@ _CFG_KEYS: dict[str, frozenset[str]] = { "retrieve": frozenset({"query", "collection", "limit", "alpha", "rerank", "min_score", "assign_to"}), "metrics": frozenset({"metrics", "caliber", "caliber_version", "on_missing", "assign_to"}), "report": _MODEL_KEYS | {"numbers", "claims", "judge", "entities", "max_repairs", "system", "instructions", - "on_violation", "evidence_from", "metrics_from", "emit_message", "assign_to"}, + "on_violation", "evidence_from", "metrics_from", "emit_message", "assign_to", + "analysis_claims", "analysis_scope"}, } #: 只在调模型时才读的（_calls_model）：条件分支让模型分类、结构校验让模型修复（默认开着）、知识检索用模型重排 _CFG_KEYS_IF: dict[str, frozenset[str]] = { diff --git a/backend/app/engine/labels.py b/backend/app/engine/labels.py index 1ba796a7..dc4c56d1 100644 --- a/backend/app/engine/labels.py +++ b/backend/app/engine/labels.py @@ -59,6 +59,8 @@ FIELD_LABEL: dict[str, str] = { "instructions": "写作要求", "metrics_from": "指标来自", "numbers": "未引用的数字", "on_violation": "重写后仍有违规时", "max_repairs": "最多重写次数", "claims": "未附依据的结论句", "judge": "结论句裁判", "entities": "表名、字段名", + # 分析能力一轮：判断词要挂对应的分析结论（engine/analysis_claims）、文档末尾系统写的分析范围（engine/analysis_scope） + "analysis_claims": "分析说法核对", "analysis_scope": "分析范围", # 口径卡 "caliber_from": "口径卡来源", "caliber": "口径名称", "caliber_version": "口径版本", "metrics": "指标定义", "on_missing": "缺输入时", @@ -123,6 +125,8 @@ _OPTION_LABEL: dict[str, dict[str, str]] = { "on_violation": {"flag": "正常产出，并标注违规之处", "fail": "判为失败"}, "claims": {"off": "不参与出具判档", "require_citation": "计入缺口", "judge": "计入缺口并由模型逐句判断"}, "entities": {"link": "核对", "off": "不核对"}, + "analysis_claims": {"off": "不核对", "check": "重写一次后标出", "strict": "重写后仍有的计入缺口"}, + "analysis_scope": {"auto": "有分析结论时附上", "off": "不附"}, "evidence_role": {"source": "取数", "compute": "计算"}, "on_uncited": {"ignore": "仅标注，不计入缺口", "degrade": "计入缺口、出具降档", "withhold": "不予出具"}, "on_unsupported": {"degrade": "出具降档", "withhold": "不予出具"}, diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index f3efa3cf..ea4b58e5 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -29,6 +29,16 @@ 不重复调用、不重复计费。写作调用不进 checkpoint，接着跑时会重写一稿——所以判定带着输入指纹， 对不上这一稿的不用 +分析能力一轮（docs/design/analysis-round/00-总方案.md §3.5–3.6）：判定和数字一样由系统渲染，写作者只引用。 + +- 分析说法核对 analysis_claims：「主要来自」「异常」「持续上升」这类判断词要挂对应的分析结论、判定对得上，因果词要有 + 知识库原话或运行输入作依据（engine/analysis_claims，确定的词表 + 绑定，不调模型）。check（缺省，问数据也是它）交回写作者 + 重写一次（同句中的 [[see:]] 那一类，不按 max_repairs 反复），仍有的标在报告里、不影响档位；strict 按 max_repairs 重写， + 仍有的写进产出的 issuance_gaps，带契约的出口降档（受管级别要求 strict，门禁 G3）；off 不核。说法违规都不让节点失败： + 不挡生成。文档顶层记下处置和说法表版本，出口复核照它独立再核一遍 +- 分析范围 analysis_scope：这次有分析结论、取值 auto（缺省）时，文档里另记系统写的「分析范围」（现象、拆过、排除、没拆、 + 检验，engine/analysis_scope），不进正文、不过裸数字和说法核对 + 出口契约复核（io.py）用 report_catalog 按同一套参数重建目录、独立重算，不信这里的自查。 """ from __future__ import annotations @@ -47,6 +57,7 @@ from langgraph.func import task from app.core.errors import describe_exception, not_configured from app.core.events import EventType from app.db.base import SessionLocal +from app.engine import analysis_claims as said from app.engine import judge as judging from app.engine.context import NodeContext, NodeError from app.engine.labels import field_label, option_label @@ -73,6 +84,9 @@ from app.engine.evidence import ( ) from app.engine.nodes.llm import _invoke_streaming, _model_spec, _report_call from app.engine.schema import ( + ANALYSIS_CLAIMS_DEFAULT, + ANALYSIS_CLAIMS_VALUES, + ANALYSIS_SCOPE_VALUES, CLAIMS_JUDGE, CLAIMS_VALUES, JUDGE_ON_UNSUPPORTED, @@ -80,6 +94,8 @@ from app.engine.schema import ( GraphSpec, NodeType, _ancestors, + analysis_claims_problem, + analysis_scope_problem, claims_problem, judge_problems, ) @@ -101,11 +117,14 @@ SOFT = frozenset({"unknown_entity", "unverified_entity"}) #: 写作目录有省略时的缺失断言（「没有查到」「数据缺失」：写作者没看全证据，结论依据不足，但数字和出处都核对过）。 #: 只剩这几类时不按 max_repairs 反复重写：重写一次没改掉，再写也多半一样，留在核对结果里标注 REWRITE_ONLY = frozenset({"inline_see", "absence_claim"}) +#: 分析说法的违规（判断词没挂对分析结论、因果没有依据）：都不让节点失败。check 和 REWRITE_ONLY 一样只重写一次、仍有的 +#: 标注；strict 按 max_repairs 重写，仍有的写进出具缺口（_again、run_report 收尾） +ANALYSIS = said.CODES #: 重写一次就是多一次整篇的模型调用，写不对的模型多给几次也大多写不对 MAX_REPAIRS = 3 ROLE = ("你是报告撰写人。报告里的每个数字都由系统从证据里取出来、按口径卡的格式渲染；" - "你负责挑重点、组织结构、说清变化和原因。") + "你负责挑重点、组织结构、说清变化；有分析结论时说清变化来自哪里；原因只按证据写。") MARKUP_NUDGE = ("你上一条回复把工具调用写成了文字。报告撰写不调用任何工具：能用的数据都在上面的" "证据目录里，直接用引用标记写报告正文。") @@ -360,6 +379,19 @@ def report_claims(spec: GraphSpec, node: GraphNode) -> str: return value if isinstance(value, str) and value in CLAIMS else "off" +def report_analysis_claims(spec: GraphSpec, node: GraphNode) -> str: + """报告节点的分析说法核对：off / check / strict，没写是 check。写错的值画布校验就拦了，执行时节点也会报错，这里只把 + 认不出的当缺省。""" + value = _setting(spec, node, "analysis_claims", ANALYSIS_CLAIMS_DEFAULT) + return value if isinstance(value, str) and value in ANALYSIS_CLAIMS_VALUES else ANALYSIS_CLAIMS_DEFAULT + + +def report_analysis_scope(spec: GraphSpec, node: GraphNode) -> str: + """报告节点的分析范围：auto（缺省，这次有分析结论时写）/ off。""" + value = _setting(spec, node, "analysis_scope", "auto") + return value if isinstance(value, str) and value in ANALYSIS_SCOPE_VALUES else "auto" + + def report_judge(spec: GraphSpec, node: GraphNode) -> dict[str, Any]: """claims: judge 的子配置：节点上的 judge（没写取图级 defaults），写了的键原样保留——上限写 null 就是 不限，没写的键运行时取设置里的默认（judge.report_budget）。rewrite_once、on_unsupported 补上缺省 @@ -550,6 +582,8 @@ def _options(ctx: NodeContext) -> tuple[str, int]: entities = ctx.cfg("entities", "link") or "link" if entities not in ENTITIES: raise NodeError(ctx.node.id, _only("entities", ENTITIES, entities)) + if problem := analysis_claims_problem(ctx.cfg("analysis_claims")) or analysis_scope_problem(ctx.cfg("analysis_scope")): + raise NodeError(ctx.node.id, problem) raw = ctx.cfg("max_repairs", 1) try: repairs = int(raw) @@ -582,8 +616,16 @@ def _blocking(violations: list[dict[str, Any]], numbers: str) -> list[dict[str, def _fatal(blocking: list[dict[str, Any]]) -> list[dict[str, Any]]: - """要判失败的违规：重写之后还在的、不属于 REWRITE_ONLY 的。""" - return [v for v in blocking if v.get("code") not in REWRITE_ONLY] + """要判失败的违规：重写之后还在的、不属于 REWRITE_ONLY 的。分析说法的违规不判失败（不挡生成）：check 只标注， + strict 记进出具缺口。""" + return [v for v in blocking if v.get("code") not in REWRITE_ONLY and v.get("code") not in ANALYSIS] + + +def _again(blocking: list[dict[str, Any]], analysis: str) -> list[dict[str, Any]]: + """值得按 max_repairs 反复重写的违规：要判失败的那些，加上 strict 下的分析说法违规。只剩别的（REWRITE_ONLY、check 下 + 的说法违规）时重写一次就停。""" + return [v for v in blocking if v.get("code") not in REWRITE_ONLY + and (v.get("code") not in ANALYSIS or analysis == "strict")] def _fail_advice(fatal: list[dict[str, Any]]) -> str: @@ -635,6 +677,8 @@ def _repair_request(violations: list[dict[str, Any]], failed_rule: str | None = f"{describe_violations(violations)}\n\n" "请重写整篇报告：所有数字都用引用标记写，只引用证据目录里有的指标和输入；" + (f"{failed_rule}" if failed_rule else "") + # 判断词没挂对分析结论、原因没有依据：怎么改在上面每一条里写了，这里再说一句总的 + + (said.ANALYSIS_REPAIR_PROMPT if any(v.get("code") in ANALYSIS for v in violations) else "") + "不要解释改了什么，只输出新的报告正文。" ) @@ -645,11 +689,15 @@ def run_today() -> date: def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims: str = "off", - failed: str | None = None) -> list[BaseMessage]: + failed: str | None = None, analysis: str | None = None) -> list[BaseMessage]: + """analysis 是这次要核分析说法时（check / strict）拼在写作规则后面的那一段（analysis_claims.rules_prompt：目录里有 + 分析结论和没有时各一种说法），不核时为 None、写作规则和以前一字不差。""" 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}\n{UNIT_RULE}\n{period_rule(run_today())}" + if analysis: + rules = f"{rules}\n{analysis}" if writer.guarded: # 写作者看不全证据（目录有省略、有来源的结构化结论为空）：「没有查到」这类结论没有依据 rules = f"{rules}\n{ABSENCE_RULE}" @@ -770,6 +818,15 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 结论句策略只对升级后发起的运行生效：升级前的运行（图里也不会有 claims）事件、产出一个字不加 evidence_on = ledger_enabled(ctx.run) claims = report_claims(spec, ctx.node) if evidence_on else "off" + # 分析说法核对、分析范围同样只对升级后发起的运行生效：升级前的运行写作规则、文档、产出一个字不加 + analysis = report_analysis_claims(spec, ctx.node) if evidence_on else "off" + scope = None + if evidence_on and report_analysis_scope(spec, ctx.node) == "auto" and said.results_of(catalog): + # 这次有分析结论：按写作目录里的结论和数据目录算一次「分析范围」，每一稿的文档都带着它（不进正文） + from app.data import catalog_view + from app.engine.analysis_scope import scope_of + + scope = await scope_of(catalog, catalog_view.of_run(ctx.run)) if evidence_on: # 升级前发起的运行没有查询条目，单元格无从谈起，也就不多查这一次 from app.engine.governance import governed_formal @@ -795,7 +852,8 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: + "：已告诉写作模型照实说没有查到，不为它们写引用", failed=len(failed), any_query=any_query) messages = _messages(ctx, state, writer, claims, - _failed_section(failed, any_query=any_query) if failed else None) + _failed_section(failed, any_query=any_query) if failed else None, + said.rules_prompt(catalog) if analysis in said.MODES else None) started = time.perf_counter() spent: list[dict[str, Any]] = [] nudged = False @@ -829,16 +887,21 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # （_MaskedCells）：只读正文里引用到的那几份快照 # 文档记这次运行里的节点 id（子工作流里的报告是「子工作流节点 id/…」）：和它的 report.checked 事件、产出工件同一个 # id。证据接口、可验证报告、质询都拿文档上的 id 和事件上的比，对不上就判「疑似被修改」 - return await masked(compose_doc(_unfenced(raw), catalog, node_id=ctx.run.global_id(ctx.node.id), - run_id=ctx.run.run_id, - allow_numbers=allow, cells_allowed=cells, entities=entities, - absence=writer.trimmed)) + # 分析说法按 analysis 核（check / strict 时文档记下这一项，off 不记不核）；分析范围是系统写的，每一稿都带同一份 + doc = await masked(compose_doc(_unfenced(raw), catalog, node_id=ctx.run.global_id(ctx.node.id), + run_id=ctx.run.run_id, + allow_numbers=allow, cells_allowed=cells, entities=entities, + absence=writer.trimmed, + analysis_claims=analysis if analysis in said.MODES else None)) + if scope is not None: + doc["analysis_scope"] = scope + return doc response, raw = await draft(messages) doc = await compose(raw) repairs = 0 while (blocking := _blocking(doc["violations"], numbers)) and repairs < max_repairs \ - and not (repairs and not _fatal(blocking)): + and not (repairs and not _again(blocking, analysis)): repairs += 1 ctx.emit(EventType.LOG, level="warn", code="report_repair", count=len(blocking), attempt=repairs, message=f"报告有 {len(blocking)} 处未通过核对（{_named(blocking)}），" @@ -885,11 +948,27 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: columns=list(dict.fromkeys(str(v["column"]) for v in hits if v.get("column"))), message=f"{gaps[0]}。报告照常产出，这几处已标注，已记为缺口。" f"请在报告撰写节点的指令中去掉对{_masked_targets(hits)}的引用") + analysis_gap = None + if (left := [v for v in doc["violations"] if v.get("code") in ANALYSIS]) \ + and not (_fatal(blocking) and on_violation == "fail"): + # 重写之后还在：报告照常产出（不挡生成），这几处标在报告里；strict 另在产出里写一条缺口，成果节点设了出具契约、报告在 + # 它上游时照单收下、降档（io._upstream_issuance_gaps；引用模式的出口自己再核一遍，见 io._analysis_claims） + tried = f"已要求模型重写 {repairs} 次仍未改掉" if repairs else "未要求重写" + what = said.summary(left) + if analysis == "strict": + analysis_gap = f"报告有 {len(left)} 处分析说法未通过核对（{what}：{_named(left)}），{tried}" + gaps.append(analysis_gap) + ctx.emit(EventType.LOG, level="warn", code="report_analysis_claims", count=len(left), mode=analysis, + texts=list(dict.fromkeys(str(v.get("text") or "") for v in left)), counts=said.counts(left), + message=f"报告有 {len(left)} 处分析说法未通过核对（{what}：{_named(left)}），{tried}：" + "判断要挂对应的分析结论并照判定写，原因要有知识库原话或运行输入作依据。报告照常产出，这几处已标注" + + ("，已记为缺口" if analysis_gap else "")) 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, on_violation=on_violation, failed=bool(_fatal(blocking)) and on_violation == "fail", **({"claims": claims} if evidence_on else {}), + **({"analysis_claims": analysis} if evidence_on else {}), **({"judge": judgement} if judgement is not None else {})) if (fatal := _fatal(blocking)) and on_violation == "fail": blocking = fatal @@ -912,9 +991,13 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: "model": model_id, "duration_ms": int((time.perf_counter() - started) * 1000), **({"claims": claims} if evidence_on else {}), + # 分析说法核对的处置：出口按它和文档里记的那一项判 strict 要不要记缺口（升级前的运行没有这个键） + **({"analysis_claims": analysis} if evidence_on else {}), **({"judge": judgement} if judgement is not None else {}), # 出口契约照单收下的缺口（只在有的时候带这个键：别的报告产出和以前一字不差） **({"issuance_gaps": gaps} if gaps else {}), + # 其中分析说法那一条：引用模式的出口自己复核过这份文档时以复核的为准，不再重复收这一条（io._analysis_claims） + **({"analysis_gap": analysis_gap} if analysis_gap else {}), # 写成「已遮罩」的指标：出口契约复核按它重建目录（report_catalog）。同上，只在有的时候带 **({"masked_metrics": withheld} if withheld else {}), } diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index 9adf1440..9d5e497f 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -636,6 +636,13 @@ CLAIMS_VALUES = (*REPORT_CLAIMS, CLAIMS_JUDGE) JUDGE_KEYS = ("provider", "model", "max_claims", "max_cost_usd", "timeout_s", "rewrite_once", "on_unsupported") #: 证据相矛盾时正式运行怎么判档：degrade 降档（默认）/ withhold 不予出具。探索运行只标注 JUDGE_ON_UNSUPPORTED = ("degrade", "withhold") +#: 报告撰写节点的分析说法核对（analysis_claims，docs/design/analysis-round/00-总方案.md §3.5、engine/analysis_claims）： +#: off 不核；check（缺省，问数据也是它）判断词没挂对分析结论的句子交回写作者重写一次、仍有的标出，不影响出具档位； +#: strict 按最多重写次数重写，仍有的计入出具缺口（受管级别要求它，门禁 G3） +ANALYSIS_CLAIMS_VALUES = ("off", "check", "strict") +ANALYSIS_CLAIMS_DEFAULT = "check" +#: 报告文档里系统写的「分析范围」（analysis_scope，engine/analysis_scope）：auto（缺省）这次有分析结论时写；off 不写 +ANALYSIS_SCOPE_VALUES = ("auto", "off") #: 「没有指定模型」那条提示的编号。节点（或全图默认）点名了接入、而那个接入本身不存在或已停用时，它和接入那条问题 #: 互相矛盾（一条说照常执行、一条说会失败），接口层读了接入表以后按编号去掉它（provider_refs.quiet_unset_model） @@ -659,6 +666,24 @@ def claims_problem(value: Any) -> str | None: return (f"「{field_label('claims')}」只能选{'、'.join(allowed[:-1])}或{allowed[-1]}，当前为「{value}」") +def _choice_problem(key: str, value: Any, allowed: tuple[str, ...]) -> str | None: + """一个只能在几个取值里选的报告节点配置写错了的说法（没写的不算错，按缺省）。""" + if value in (None, "") or (isinstance(value, str) and value in allowed): + return None + names = [f"「{option_label(key, v)}」" for v in allowed] + return f"「{field_label(key)}」只能选{'、'.join(names[:-1])}或{names[-1]}，当前为「{value}」" + + +def analysis_claims_problem(value: Any) -> str | None: + """analysis_claims 的取值有什么问题，没问题返回 None。校验和报告撰写节点执行时报的是同一句话。""" + return _choice_problem("analysis_claims", value, ANALYSIS_CLAIMS_VALUES) + + +def analysis_scope_problem(value: Any) -> str | None: + """analysis_scope 的取值有什么问题，没问题返回 None。""" + return _choice_problem("analysis_scope", value, ANALYSIS_SCOPE_VALUES) + + def _positive(value: Any, *, integer: bool = False) -> bool: if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0: return False @@ -717,6 +742,11 @@ def _check_report_claims(node: GraphNode, spec: GraphSpec, result: ValidationRes value = effective("claims") if problem := claims_problem(value): result.add(problem, node_id=node.id, field="claims", code="report.claims_invalid") + # 分析说法核对、分析范围：写错了执行时报告节点照样报这一句 + if problem := analysis_claims_problem(effective("analysis_claims")): + result.add(problem, node_id=node.id, field="analysis_claims", code="report.analysis_claims_invalid") + if problem := analysis_scope_problem(effective("analysis_scope")): + result.add(problem, node_id=node.id, field="analysis_scope", code="report.analysis_scope_invalid") if value != CLAIMS_JUDGE: return judge = effective("judge") diff --git a/backend/tests/test_evidence_audit.py b/backend/tests/test_evidence_audit.py index 02aee673..4b560297 100644 --- a/backend/tests/test_evidence_audit.py +++ b/backend/tests/test_evidence_audit.py @@ -122,7 +122,8 @@ async def events(run_id: str, etype: str, node_id: str | None = None) -> list[Ru async def report_run(monkeypatch, text=WRITE) -> tuple[Run, dict]: script(monkeypatch, text) row = await finish(chain(node("start", "input"), node("fetch", "tool", tool=TOOL, args={"sql": SQL}), - node("write", "report", instructions="写一句", max_repairs=0), + # 「增长主要来自新客」没有分析结论可挂：分析说法核对关掉，审计表只看原有的几类 + node("write", "report", instructions="写一句", max_repairs=0, analysis_claims="off"), node("out", "output", fields=[{"name": "r", "value": "{{ nodes.write.text }}"}]))) assert row.status == "succeeded", row.error [checked] = await events(row.id, "report.checked", "write") diff --git a/backend/tests/test_judge_rework_grading.py b/backend/tests/test_judge_rework_grading.py index 2792765f..75814b50 100644 --- a/backend/tests/test_judge_rework_grading.py +++ b/backend/tests/test_judge_rework_grading.py @@ -149,8 +149,9 @@ def weekly(*, report=None, contract=None) -> dict: node("start", "input", fields=[{"name": "week", "default": "2026-W37"}]), node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"), node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=METRICS), + # 这里测的是裁判：写作者那句「增长主要来自新客首单」没有分析结论可挂，分析说法核对关掉，免得多重写一次 node("write", "report", instructions="写周报", **{"claims": "judge", "judge": {"max_cost_usd": 0.05}, - **(report or {})}), + "analysis_claims": "off", **(report or {})}), node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], contract={"report_from": "write", "metrics_from": ["caliber"], "required": ["gmv"], "strict": True, **(contract or {})}), diff --git a/backend/tests/test_provider_ref_runtime.py b/backend/tests/test_provider_ref_runtime.py index f895bb82..23801186 100644 --- a/backend/tests/test_provider_ref_runtime.py +++ b/backend/tests/test_provider_ref_runtime.py @@ -371,11 +371,12 @@ JUDGED = "本周销售额 [[m:gmv]]。[[see:m:gmv]]增长主要来自新客首 def weekly(**report) -> dict: + # 这里测的是裁判接入：「增长主要来自新客首单」没有分析结论可挂，分析说法核对关掉，免得多重写一次 return chain( node("start", "input"), node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"), node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=METRICS), - node("write", "report", instructions="写周报", **report), + node("write", "report", instructions="写周报", **{"analysis_claims": "off", **report}), node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), ) diff --git a/backend/tests/test_report_analysis_claims.py b/backend/tests/test_report_analysis_claims.py new file mode 100644 index 00000000..e87d154e --- /dev/null +++ b/backend/tests/test_report_analysis_claims.py @@ -0,0 +1,309 @@ +"""报告撰写节点的分析说法核对（analysis_claims）和分析范围（analysis_scope），00-总方案 §3.5–3.6。 + +- check（缺省，问数据也是它）：判断词没挂对分析结论的句子交回写作者重写一次（不按 max_repairs 反复），仍有的标在报告里， + 节点照常产出、不写缺口 +- strict：按 max_repairs 重写，仍有的写进产出的 issuance_gaps（带契约的出口降档）；即使「违规处置」是判为失败也不让节点失败 +- off：不核，写作规则、文档和以前一样 +- 写作规则：有分析结论和没有时各一种说法；重写请求列出违规和修法；写作者的角色说「原因只按证据写」 +- 分析范围：这次有分析结论、取值 auto 时文档带系统写的 analysis_scope，off 或没有分析结论时不带 +- 取值写错：画布校验拦住，执行时节点报同一句话 +""" +from __future__ import annotations + +import asyncio +import re + +import pytest +from langchain_core.messages import AIMessage +from sqlalchemy import select + +from app.analysis.testing import fake_result +from app.core import artifact_store +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.engine import analysis_claims as ac +from app.engine import evidence +from app.engine import runner as runner_mod +from app.engine.evidence import make_eid +from app.engine.nodes import report as report_mod +from app.engine.runner import run_manager +from app.engine.schema import analysis_claims_problem, analysis_scope_problem + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +METRICS = [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"}] +#: 挂了指标依据、可「主要来自」没有拆解可挂 +SOURCE = "本周销售额 [[m:gmv]]，下降主要来自华东。[[see:m:gmv]]" +PLAIN = "本周销售额 [[m:gmv]]。[[see:m:gmv]]" + + +def weekly(**report): + nodes = [ + node("start", "input", fields=[{"name": "week", "default": "2026-W37"}]), + node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"), + node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=METRICS), + node("write", "report", instructions="写周报", **report), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ] + ids = [n["id"] for n in nodes] + return {"nodes": nodes, "edges": [{"source": a, "target": b} for a, b in zip(ids, ids[1:])]} + + +def script(monkeypatch, *replies) -> list: + from app.providers import mock_model + + seen: list = [] + + def _decide(self, messages): + seen.append(list(messages)) + return AIMessage(content=replies[min(len(seen), len(replies)) - 1]) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide) + return seen + + +async def finish(graph, *, run_class="exploratory") -> Run: + run = await run_manager.start(graph=graph, input_payload={}, run_class=run_class) + 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 events(run_id: str, etype: str) -> list[RunEvent]: + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype).order_by(RunEvent.seq) + return list((await session.execute(q)).scalars()) + + +async def checked_of(row: Run) -> tuple[dict, dict, dict]: + """(report.checked 的载荷, 文档, 报告节点产出)。""" + [checked] = await events(row.id, "report.checked") + [finished] = [e for e in await events(row.id, "node.finished") if e.node_id == "write"] + return (checked.data, artifact_store.load(checked.data["doc_artifact"]), + artifact_store.load(finished.data["artifact"])) + + +async def logs(row: Run, code: str) -> list[dict]: + return [e.data for e in await events(row.id, "log") if e.data.get("code") == code] + + +def said(violations: list[dict]) -> list[str]: + return [v["code"] for v in violations if v["code"] in ac.CODES] + + +def human_texts(messages) -> str: + return "\n".join(str(m.content) for m in messages if m.type == "human") + + +# -------------------------------------------------------------------------- +# check：重写一次，仍有的标出 +# -------------------------------------------------------------------------- + + +async def test_check_is_the_default_rewrites_once_and_flags_what_is_left(monkeypatch): + seen = script(monkeypatch, SOURCE) + row = await finish(weekly(max_repairs=3)) + assert row.status == "succeeded", row.error + assert len(seen) == 2, "check 只重写一次，不按最多重写次数反复" + first, again = human_texts(seen[0]), str(seen[1][-1].content) + assert ac.ANALYSIS_NONE_PROMPT in first, "没有分析结论时告诉写作者不下这类判断" + assert "这次没有变化拆解的结论，不能写「主要来自」" in again and ac.ANALYSIS_REPAIR_PROMPT in again + checked, doc, output = await checked_of(row) + assert doc[ac.DOC_FIELD] == {"mode": "check", "lexicon": ac.LEXICON_VERSION} + assert said(checked["violations"]) == [ac.UNCITED] and checked["analysis_claims"] == "check" + assert checked["failed"] is False and checked["repairs"] == 1 + assert output["analysis_claims"] == "check" and "issuance_gaps" not in output and "analysis_gap" not in output + [log] = await logs(row, "report_analysis_claims") + assert log["count"] == 1 and log["texts"] == ["主要来自"] and "已记为缺口" not in log["message"] + assert "已要求模型重写 1 次仍未改掉" in log["message"] and "1 处判断没挂分析结论" in log["message"] + + +async def test_a_rewrite_that_drops_the_judgement_is_accepted(monkeypatch): + seen = script(monkeypatch, SOURCE, "本周销售额 [[m:gmv]]，华东下降最多。[[see:m:gmv]]") + row = await finish(weekly()) + checked, doc, output = await checked_of(row) + assert len(seen) == 2 and said(checked["violations"]) == [] and checked["ok"] is True + assert await logs(row, "report_analysis_claims") == [] + + +async def test_check_does_not_change_how_fatal_violations_repeat(monkeypatch): + seen = script(monkeypatch, "销售额 45678 元，下降主要来自华东。") + row = await finish(weekly(max_repairs=2, on_violation="flag")) + assert row.status == "succeeded" and len(seen) == 3, "裸数字照旧按最多重写次数重写" + checked, _, _ = await checked_of(row) + assert {"uncited_number", ac.UNCITED} <= {v["code"] for v in checked["violations"]} + + +async def test_check_leftovers_never_fail_the_node(monkeypatch): + script(monkeypatch, SOURCE) + row = await finish(weekly(on_violation="fail")) + assert row.status == "succeeded", row.error + checked, _, _ = await checked_of(row) + assert checked["failed"] is False and said(checked["violations"]) == [ac.UNCITED] + + +# -------------------------------------------------------------------------- +# strict：按最多重写次数重写，仍有的计入缺口 +# -------------------------------------------------------------------------- + + +async def test_strict_rewrites_up_to_max_repairs_and_records_a_gap(monkeypatch): + seen = script(monkeypatch, SOURCE) + row = await finish(weekly(analysis_claims="strict", max_repairs=2, on_violation="fail")) + assert row.status == "succeeded", "说法违规不让节点失败（不挡生成）" + assert len(seen) == 3 + checked, doc, output = await checked_of(row) + assert doc[ac.DOC_FIELD]["mode"] == "strict" and checked["failed"] is False + [gap] = output["issuance_gaps"] + assert gap == output["analysis_gap"] + assert gap.startswith("报告有 1 处分析说法未通过核对（1 处判断没挂分析结论：「主要来自」）") and "重写 2 次" in gap + [log] = await logs(row, "report_analysis_claims") + assert log["mode"] == "strict" and log["message"].endswith("已记为缺口") + + +async def test_strict_stops_once_the_judgements_are_fixed(monkeypatch): + seen = script(monkeypatch, SOURCE, PLAIN) + row = await finish(weekly(analysis_claims="strict", max_repairs=3)) + _, _, output = await checked_of(row) + assert len(seen) == 2 and "issuance_gaps" not in output + + +# -------------------------------------------------------------------------- +# off +# -------------------------------------------------------------------------- + + +async def test_off_checks_nothing_and_adds_no_rules(monkeypatch): + seen = script(monkeypatch, SOURCE) + row = await finish(weekly(analysis_claims="off")) + checked, doc, output = await checked_of(row) + assert len(seen) == 1 and ac.DOC_FIELD not in doc and said(checked["violations"]) == [] + assert ac.ANALYSIS_NONE_PROMPT not in human_texts(seen[0]) and ac.ANALYSIS_RULES_PROMPT not in human_texts(seen[0]) + assert output["analysis_claims"] == "off" and "analysis_claims" not in doc["stats"] + + +async def test_the_writer_role_asks_for_where_changes_come_from_and_evidence_for_causes(monkeypatch): + seen = script(monkeypatch, PLAIN) + await finish(weekly()) + system = str(seen[0][0].content) + assert report_mod.ROLE in system + assert "有分析结论时说清变化来自哪里" in report_mod.ROLE and "原因只按证据写" in report_mod.ROLE + assert "说清变化和原因" not in report_mod.ROLE + + +# -------------------------------------------------------------------------- +# 目录里有分析结论（证据接入合进来之前，这里按它的约定把 A 条目放进目录、代为解析 [[see:A1]]） +# -------------------------------------------------------------------------- + + +def a_entry(alias: str, result: dict) -> dict: + artifact = "b" * 63 + alias[-1] + return {"alias": alias, "kind": "analysis", "artifact": artifact, "eid": make_eid("analysis", artifact, {}), + "locator": {}, "label": f"{alias} {result.get('phrase')}", "result": result} + + +@pytest.fixture +def analyses(monkeypatch): + """写作目录里多几条分析结论：{A 编号: 结论}。报告节点和出口复核按同一个 report_catalog 建目录，这里一处补上。""" + added: dict[str, dict] = {} + original = report_mod.report_catalog + + def catalog(state, spec, node): + out = original(state, spec, node) + out.update({alias: a_entry(alias, r) for alias, r in added.items()}) + return out + + monkeypatch.setattr(report_mod, "report_catalog", catalog) + resolve = evidence.resolve_support + + def support(ref, catalog, *, loader=None): + ref = ref.strip() + if re.fullmatch(r"A\d+", ref) and (catalog.get(ref) or {}).get("kind") == "analysis": + return {"ref": ref, "role": "support", "alias": ref, "kind": "analysis", "status": "resolved", + "eid": catalog[ref]["eid"], "locator": {}} + return resolve(ref, catalog, loader=loader) + + monkeypatch.setattr(evidence, "resolve_support", support) + return added + + +async def test_a_judgement_matching_its_conclusion_passes(monkeypatch, analyses): + analyses["A1"] = fake_result() + seen = script(monkeypatch, "本周销售额 [[m:gmv]]，下降主要来自华东。[[see:m:gmv,A1]]") + row = await finish(weekly()) + checked, doc, _ = await checked_of(row) + assert len(seen) == 1 and said(checked["violations"]) == [] + assert ac.ANALYSIS_RULES_PROMPT in human_texts(seen[0]) and ac.ANALYSIS_NONE_PROMPT not in human_texts(seen[0]) + + +async def test_a_judgement_that_contradicts_its_conclusion_is_handed_back(monkeypatch, analyses): + analyses["A1"] = fake_result(verdict="dispersed") + seen = script(monkeypatch, "本周销售额 [[m:gmv]]，下降主要来自华东。[[see:m:gmv,A1]]") + row = await finish(weekly()) + checked, _, _ = await checked_of(row) + assert len(seen) == 2 + [miss] = [v for v in checked["violations"] if v["code"] in ac.CODES] + assert miss["code"] == ac.MISMATCH and miss["analyses"] == ["A1"] + assert "A1 判定是「销售额下降 194 万（-8.7%），分散在各地区，没有哪一段占到一半」" in str(seen[1][-1].content) + + +async def test_the_scope_is_written_only_when_there_are_conclusions(monkeypatch, analyses): + script(monkeypatch, PLAIN) + _, doc, _ = await checked_of(await finish(weekly())) + assert "analysis_scope" not in doc, "没有分析结论不写分析范围" + analyses["A1"] = fake_result() + analyses["A2"] = fake_result("baseline", None, "outside_band") + _, doc, _ = await checked_of(await finish(weekly())) + scope = doc["analysis_scope"] + assert [i["alias"] for i in scope["phenomena"]] == ["A2"] and [i["alias"] for i in scope["decomposed"]] == ["A1"] + assert scope["decomposed"][0]["dims"] == ["地区"] and scope["not_decomposed"] is None and scope["note"] + assert "analysis_scope" not in doc["markdown"] and scope["tests"] == 0 + _, doc, _ = await checked_of(await finish(weekly(analysis_scope="off"))) + assert "analysis_scope" not in doc + + +# -------------------------------------------------------------------------- +# 取值写错 +# -------------------------------------------------------------------------- + + +def test_bad_values_are_refused_by_validation(): + from app.engine.schema import GraphSpec, validate_graph + + issues = validate_graph(GraphSpec.model_validate(weekly(analysis_claims="loose", analysis_scope="always"))).issues + bad = {i.code: i for i in issues if i.code in ("report.analysis_claims_invalid", "report.analysis_scope_invalid")} + assert bad["report.analysis_claims_invalid"].level == "error" + assert bad["report.analysis_claims_invalid"].message == analysis_claims_problem("loose") + assert "「分析说法核对」只能选「不核对」、「重写一次后标出」或「重写后仍有的计入缺口」" in analysis_claims_problem("loose") + assert bad["report.analysis_scope_invalid"].message == analysis_scope_problem("always") + assert analysis_claims_problem(None) is None and analysis_claims_problem("strict") is None + + +async def test_bad_values_fail_the_node_before_any_model_call(monkeypatch): + seen = script(monkeypatch, PLAIN) + original = runner_mod.validate_graph + + def permissive(spec): + report = original(spec) + report.issues = [i for i in report.issues if i.field not in ("analysis_claims", "analysis_scope")] + report.ok = not any(i.level == "error" for i in report.issues) + return report + + monkeypatch.setattr(runner_mod, "validate_graph", permissive) + row = await finish(weekly(analysis_claims="loose")) + assert row.status == "failed" and analysis_claims_problem("loose") in (row.error or "") + assert seen == [] diff --git a/backend/tests/test_report_claims.py b/backend/tests/test_report_claims.py index 3ad71153..90960915 100644 --- a/backend/tests/test_report_claims.py +++ b/backend/tests/test_report_claims.py @@ -45,7 +45,8 @@ def weekly(**report): node("start", "input", fields=[{"name": "week", "default": "2026-W37"}]), node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"), node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=METRICS), - node("write", "report", instructions="写周报", **report), + # 这里测的是结论句策略：「增长主要来自新客」没有分析结论可挂，分析说法核对关掉（见 test_report_analysis_claims.py） + node("write", "report", instructions="写周报", **{"analysis_claims": "off", **report}), node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), ] ids = [n["id"] for n in nodes] diff --git a/backend/tests/test_report_judge.py b/backend/tests/test_report_judge.py index e6589367..bb7a31e9 100644 --- a/backend/tests/test_report_judge.py +++ b/backend/tests/test_report_judge.py @@ -58,7 +58,9 @@ def weekly(*, gate=False, **report): node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"), node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=METRICS), *([node("gate", "human", mode="approve", title="确认口径")] if gate else []), - node("write", "report", instructions="写周报", **report), + # 这里测的是裁判：写作者那句「增长主要来自新客首单」没有分析结论可挂，分析说法核对关掉，免得多重写一次 + # （说法核对见 test_report_analysis_claims.py） + node("write", "report", instructions="写周报", **{"analysis_claims": "off", **report}), node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), ] ids = [n["id"] for n in nodes]