From 7794794278a3e66faba612f2d360a3a335b14103 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 18:11:57 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=84=E6=B5=8B=E7=9A=84=E4=B8=A4=E7=A7=8D?= =?UTF-8?q?=E5=88=86=E6=9E=90=E6=9C=9F=E6=9C=9B=EF=BC=88=E8=B4=A1=E7=8C=AE?= =?UTF-8?q?=E3=80=81=E5=88=A4=E5=AE=9A=EF=BC=89=EF=BC=9A=E6=9C=9F=E6=9C=9B?= =?UTF-8?q?=20SQL=20=E7=9A=84=E6=8E=A2=E9=92=88=E5=BF=AB=E7=85=A7=E4=B8=8A?= =?UTF-8?q?=E7=94=A8=E5=B9=B3=E5=8F=B0=E7=9A=84=E7=AE=97=E5=AD=90=EF=BC=88?= =?UTF-8?q?app.analysis.run=EF=BC=8C=E9=9A=8F=E6=95=B0=E6=8D=AE=E5=8F=98?= =?UTF-8?q?=E5=8C=96=E7=9A=84=E8=B0=83=E7=94=A8=E9=94=99=E8=AF=AF=E8=AE=B0?= =?UTF-8?q?=E6=88=90=E7=AE=97=E4=B8=8D=E4=BA=86=EF=BC=89=E7=AE=97=E5=87=BA?= =?UTF-8?q?=E6=9C=9F=E6=9C=9B=E7=9A=84=E5=88=A4=E5=AE=9A=EF=BC=8C=E8=AF=84?= =?UTF-8?q?=E5=88=86=E6=B2=BF=E7=94=A8=E4=BA=94=E7=A7=8D=E7=BB=93=E8=AE=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 题目（cases.py）：contribution 贡献恒为变化拆解（可加、比率、乘积）、另写 top_n（期望点名主要来源的前几段，缺省 1）；verdict 判定是波动基线、趋势判断、差异检验、可比性检查。参数同分析工具（不写输入、总数、下钻，不写阈值），文字参数可以带 {{ schedule.* }}；来源、方向、判定不许人填（HUMAN_VALUE_KEYS 加 segment、subjects、direction、verdict 等）；口径题、期望反问的题配不了 - 算期望（gold.py、judgements.py）：遮罩列、目录事实、现查表的日期范围和分析工具同一套取法；阈值用这次评测的分析设置快照（Context.analysis），试算时现读设置；期望 SQL 只返回两期时可以不写基期、本期；算不了的这一项照实记（评分时不能评分），参数和结果对不上的整题没有可用的值；截断的快照照旧拒收。作答前后两次期望、和上一次比的「数据变了」都不比运行当天、现查时刻、输入快照的工件（stable_sets） - 评分（score.py、judgements.py，SCORER_VERSION 3）：判定码归到同一套判定类再比；答案引用了（判定语、具名的数、结果表的格、句末依据）判定对得上的分析结论是对且可追溯，引用它的那句却写反了不算；没引用但正文按说法表（engine/analysis_claims 同一个 scan、同一套绑定规则）写对了是对但没引用；判定、来源、方向对不上或写成相反、根本没做也没说是错。答案的分析结论从作答运行的台账读、引用从报告文档读，有判断词的句子一并读出；逐题 E1 引用分析结论时按它的输入查询比（下钻顺着上一层找）；判定语不再当成答案里的数；老结果的明细里没有这两项按空读 - 执行（runner.py、answer.py）：有分析期望的题集开评时拍一份分析设置，算期望、各臂作答的运行（Run.options 的 analysis）都按它，记进指纹（compare 的「分析设置」一项），重试沿用；别的题集的指纹、作答运行的选项不变。助手提示词的指纹加上分析方法（guide 的 id 和全文）、写作者的分析规则和说法表版本 - 总表（stats.py）：各期望的结论另按期望的种类各一份（outcomes_by_kind） --- backend/app/eval/answer.py | 17 +- backend/app/eval/cases.py | 64 +++- backend/app/eval/compare.py | 6 +- backend/app/eval/gold.py | 108 +++++- backend/app/eval/judgements.py | 520 ++++++++++++++++++++++++++++ backend/app/eval/runner.py | 42 ++- backend/app/eval/score.py | 136 +++++++- backend/app/eval/stats.py | 8 + backend/tests/test_eval_analysis.py | 317 +++++++++++++++++ 9 files changed, 1181 insertions(+), 37 deletions(-) create mode 100644 backend/app/eval/judgements.py create mode 100644 backend/tests/test_eval_analysis.py diff --git a/backend/app/eval/answer.py b/backend/app/eval/answer.py index 2842b27e..326e320d 100644 --- a/backend/app/eval/answer.py +++ b/backend/app/eval/answer.py @@ -14,7 +14,8 @@ app/chat/job.py），生成过程是 api/copilot.answer_ops，几个「决定」 同一个规矩先带着原因重来一次，还不行才记建图失败。每次重来的原因照实记在 retries。 - 运行：fill_input 填入口字段（单轮，history 为空），来源 eval，名字「评测：题集名·第 n 题」，选项带目录视图（Run.options 的 catalog_view：运行时的 db_schema、冻结快照、自查都按它）、只读和遮罩（同使用者的问数据运行：评测页会显示答案里的数， - 遮罩值不能从这里露出来；评测不写业务库）。写作模型按变体写进图的 defaults：节点自己点了模型的不覆盖，照实记在 + 遮罩值不能从这里露出来；评测不写业务库），有分析期望的题集另带这次评测的分析设置快照（判定用的阈值和算期望同一份）。 + 写作模型按变体写进图的 defaults：节点自己点了模型的不覆盖，照实记在 own_models。 - 终态：停在审批上（危险工具、人工节点）的直接取消、记「评测不代批」；被驳回的（cancelled + end_reason=rejected，总方案 X9）记路径「被驳回」，不算运行失败；失败、被停止记运行失败。 @@ -68,6 +69,8 @@ class TurnContext: writer: Mapping[str, Any] = field(default_factory=dict) #: 目录视图（Run.options['catalog_view'] 的形状） view: Mapping[str, Any] | None = None + #: 这次评测的分析设置快照（有分析期望的题集才有）：作答的运行按它判（Run.options['analysis']），和算期望同一份 + analysis: Mapping[str, Any] | None = None async def writer_usage(run_ids: list[str]) -> dict[str, Any]: @@ -116,14 +119,18 @@ def with_writer(graph: dict[str, Any], writer: Mapping[str, Any]) -> tuple[dict[ return out, own -def _options(view: Mapping[str, Any] | None) -> dict[str, Any]: - """作答运行带的选项：目录视图、只读、遮罩（见模块说明）。""" +def _options(view: Mapping[str, Any] | None, analysis: Mapping[str, Any] | None = None) -> dict[str, Any]: + """作答运行带的选项：目录视图、只读、遮罩（见模块说明）；有分析期望的题集另带这次评测的分析设置快照（发起时 + engine/launch 见调用方给了就照它的，不再现读设置）：判「主要来自」「超出常规波动」的阈值和算期望的是同一份。""" + from app.analysis.contract import SETTINGS_KEY from app.data.catalog_view import OPTION_KEY from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION out: dict[str, Any] = {MASK_RESULTS_OPTION: True, READ_ONLY_OPTION: True} if view: out[OPTION_KEY] = dict(view) + if analysis is not None: + out[SETTINGS_KEY] = dict(analysis) return out @@ -168,8 +175,8 @@ class Driver(TurnDriver): "suite_id": self.ctx.suite_id, "eval_run_id": self.ctx.eval_run_id, "case_id": self.ctx.case_id, "role": "answer", "arm": self.ctx.arm, "repeat": self.ctx.repeat, "attempt": self.turn.attempts}.items() if v is not None} - return {"actor": self.ctx.actor, "options": _options(self.ctx.view), "origin_kind": "eval", "origin": origin, - "workflow_name": run_name(self.ctx.suite_name, self.ctx.case_seq)} + return {"actor": self.ctx.actor, "options": _options(self.ctx.view, self.ctx.analysis), "origin_kind": "eval", + "origin": origin, "workflow_name": run_name(self.ctx.suite_name, self.ctx.case_seq)} async def on_interrupted(self, run_id: str, row: Any) -> Any | None: t = self.turn diff --git a/backend/app/eval/cases.py b/backend/app/eval/cases.py index 4334af22..9753a577 100644 --- a/backend/app/eval/cases.py +++ b/backend/app/eval/cases.py @@ -14,8 +14,14 @@ history 是多轮追问的位置：评测-D12 本轮只做单轮，非空一律 - none 期望答「没有数据」：期望 SQL 返回 0 行。 - ask 期望反问（问法含糊）：不能和要答出数的期望写在一起，不要期望答案的来源。 - text 文字要点（points）：由裁判判，单列「模型判断」，不计入准确率（评测-D4），所以 required 恒为 false。 +- contribution 贡献：在期望 SQL 的结果上做一次变化拆解（mode：additive / ratio / factor；params 同分析工具的参数， + 见 app/eval/judgements.py），top_n 是期望点名主要来源的前几段（缺省 1）。期望是算出的主要来源和方向。 +- verdict 判定：op 是 baseline / trend / differ / comparable（differ 另写 mode），params 同对应的分析工具。期望是算出的 + 判定类（常规 / 异常、上升 / 下降 / 无趋势、站得住 / 不足以下结论、可比 / 不可比）。 + 这两种的参数不写输入（就是期望 SQL 的结果）、不写阈值（按这次评测的分析设置）；文字参数可以带 {{ schedule.* }}。 **数值期望不许人填**（评测-D4）：期望里写了 value、values、expected、answer、rows 这类键的一律拒收——期望值只能由 SQL、 -口径或从运行认定的 SQL 算出来（app/eval/gold.py），人填的数会过期、会填错，也说不清是哪个口径。 +口径或从运行认定的 SQL 算出来（app/eval/gold.py），人填的数会过期、会填错，也说不清是哪个口径。分析期望同样：写了 +segment、subjects、direction、verdict 这类键（人填的来源、方向、判定）的拒收，判定由平台的算子在期望 SQL 上算。 **期望答案的来源**（gold；scalar、table、set、none 要它，ask、text 不要）： - sql：{datasource_id, sql}。SQL 可以带 {{ schedule.* }}，按评测的截至时间渲染；保存前过只读守卫。 @@ -48,14 +54,15 @@ GOLD_GUARD = "eval_gold_guard" #: 口径题指的卡不能当口径题（422）：不存在、不是口径卡、不能复用、取数节点引用了入口字段以外的值 CALIBER_NOT_REUSABLE = "eval_caliber_not_reusable" -EXPECTATION_KINDS = ("scalar", "table", "set", "none", "ask", "text") -#: 要期望答案来源的期望：数都由 SQL / 口径算 -NEEDS_GOLD = ("scalar", "table", "set", "none") +EXPECTATION_KINDS = ("scalar", "table", "set", "none", "ask", "text", "contribution", "verdict") +#: 要期望答案来源的期望：数都由 SQL / 口径算，分析的判定由算子在期望 SQL 的结果上算 +NEEDS_GOLD = ("scalar", "table", "set", "none", "contribution", "verdict") #: 不计入准确率的（评测-D4：文字要点由裁判判，单列） UNCOUNTED = ("text",) GOLD_KINDS = ("sql", "caliber", "from_run") -#: 期望里出现这些键就是在人填数（评测-D4） -HUMAN_VALUE_KEYS = ("value", "values", "expected", "answer", "rows", "number") +#: 期望里出现这些键就是在人填数（评测-D4）；后几个是人填分析的结论（来源的段、方向、判定） +HUMAN_VALUE_KEYS = ("value", "values", "expected", "answer", "rows", "number", "segment", "segments", "subjects", + "direction", "verdict", "judgement", "driver") #: 每种期望认的键（另有 id、kind、required、label） _EXPECTATION_KEYS: dict[str, tuple[str, ...]] = { @@ -65,6 +72,8 @@ _EXPECTATION_KEYS: dict[str, tuple[str, ...]] = { "none": (), "ask": (), "text": ("points",), + "contribution": ("op", "mode", "params", "top_n"), + "verdict": ("op", "mode", "params"), } _COMMON_KEYS = ("id", "kind", "required", "label") #: 期望答案来源的键（客户端能写的）；from_run 只能由服务端建 @@ -198,6 +207,9 @@ def _expectation(raw: Any, i: int, *, gold_kind: str | None, taken: set[str]) -> if kind not in EXPECTATION_KINDS: raise CaseInvalid(f"{where}的种类只能是 {'、'.join(EXPECTATION_KINDS)}", f"{path}.kind") human = [k for k in HUMAN_VALUE_KEYS if k in raw] + if human and kind in ("contribution", "verdict"): + raise CaseInvalid(f"{where}写了「{human[0]}」：分析期望的来源、方向、判定不许人填，由平台的分析方法在期望 SQL 的结果上" + "算出来。请写清做哪种分析、按哪几列", f"{path}.{human[0]}") if human: raise CaseInvalid(f"{where}写了「{human[0]}」：数值期望不许人填，期望值只能由期望 SQL、口径或从运行认定的 SQL " "算出来。请写清从期望答案的结果里取哪一格、哪一列或整张表", f"{path}.{human[0]}") @@ -274,6 +286,42 @@ def _expectation(raw: Any, i: int, *, gold_kind: str | None, taken: set[str]) -> if len(points) > POINTS_MAX or any(len(p.strip()) > POINT_MAX for p in points): raise CaseInvalid(f"{where}最多 {POINTS_MAX} 条要点，每条不超过 {POINT_MAX} 个字符", f"{path}.points") out["points"] = [p.strip() for p in points] + elif kind in ("contribution", "verdict"): + out.update(_analysis_expectation(raw, kind, where, path)) + return out + + +def _analysis_expectation(raw: Mapping[str, Any], kind: str, where: str, path: str) -> dict[str, Any]: + """贡献、判定两种期望的分析方法、模式、参数（app/eval/judgements.py）。贡献恒为变化拆解，判定是另外四种分析。""" + from app.analysis.contract import MODES + from app.eval import judgements + + ops = ("decompose",) if kind == "contribution" else judgements.VERDICT_OPS + op = raw.get("op") if raw.get("op") not in (None, "") else ("decompose" if kind == "contribution" else None) + if op not in ops: + said = "、".join(judgements.op_text(o) for o in ops) + raise CaseInvalid(f"{where}（{'贡献' if kind == 'contribution' else '判定'}）的分析方法只能是{said}" + + ("" if op is None else f"（当前为「{op}」）"), f"{path}.op") + out: dict[str, Any] = {"op": op} + modes = MODES.get(op) + mode = raw.get("mode") + if modes: + if mode not in modes: + said = "、".join(judgements.op_text(op, m) for m in modes) + raise CaseInvalid(f"{where}要选一种模式：{said}", f"{path}.mode") + out["mode"] = mode + elif mode not in (None, ""): + raise CaseInvalid(f"{where}是{judgements.op_text(op)}，没有模式可选，请去掉模式", f"{path}.mode") + try: + params = judgements.check_params(op, out.get("mode"), raw.get("params")) + except judgements.ParamsInvalid as e: + raise CaseInvalid(f"{where}的{e.message}", f"{path}.params" + (f".{e.key}" if e.key else "")) from None + for key, text in judgements.template_texts(params): + check_variables(f"{where}的参数「{key}」", text, f"{path}.params.{key}") + out["params"] = params + if kind == "contribution": + out["top_n"] = _int(raw.get("top_n", 1), f"{where}点名主要来源的前几段", f"{path}.top_n", low=1, + high=judgements.TOP_N_MAX) return out @@ -300,8 +348,8 @@ def normalize_expectations(raw: Any, *, gold_kind: str | None, case_kind: str) - if "ask" in kinds and case_kind == "workflow": raise CaseInvalid("工作流题不会反问，期望反问只用于问数据题", "expectations") if gold_kind is not None and not kinds & set(NEEDS_GOLD): - raise CaseInvalid("有期望答案的来源，却没有要它算的期望：请加一个「一个数」「多行」「名单」或「没有数据」的期望，" - "或者去掉期望答案的来源", "gold") + raise CaseInvalid("有期望答案的来源，却没有要它算的期望：请加一个「一个数」「多行」「名单」「没有数据」「贡献」" + "或「判定」的期望，或者去掉期望答案的来源", "gold") return out diff --git a/backend/app/eval/compare.py b/backend/app/eval/compare.py index bbf83118..af6c934d 100644 --- a/backend/app/eval/compare.py +++ b/backend/app/eval/compare.py @@ -11,7 +11,8 @@ 「只参数变」（多半是时间范围理解变了）；作答路径变了是「路径变」；其余「说不清」。SQL 的 same / params_only / changed 和运行 对比（engine/compare）是同一条规则：data/sql_semantics.normalize_sql 的规范文本和字面量。 -**指纹差异**：助手模型、写作模型、目录快照、提示词指纹、表结构、题目、评分规则，哪几项变了。 +**指纹差异**：助手模型、写作模型、目录快照、提示词指纹、表结构、题目、评分规则、分析设置（有分析期望的题集开评时拍的 +那一份判定阈值），哪几项变了。 **A/B**：同一次发起的几臂（共用 pair_id、共用期望）逐题配对，给两臂各自的准确率、配对表（都对 / 都错 / 只 A 对 / 只 B 对）和 精确 McNemar 双侧 p 值；p ≥ 0.05 时照实写「这批题上看不出差别」。不能判的题不进配对表（单列个数）。 @@ -30,7 +31,8 @@ CAUSES = ("data", "method", "params", "path", "unknown") CAUSE_LABEL = {"data": "数据变", "method": "做法变", "params": "只参数变", "path": "路径变", "unknown": "说不清"} #: 指纹里比的几项和叫法 FINGERPRINT_LABEL = {"assistant_model": "助手模型", "writer_model": "写作模型", "catalog": "目录快照", - "prompt": "提示词", "schemas": "表结构", "cases": "题目", "scorer_version": "评分规则"} + "prompt": "提示词", "schemas": "表结构", "cases": "题目", "scorer_version": "评分规则", + "analysis": "分析设置"} NO_DIFFERENCE = "这批题上看不出差别" SIGNIFICANT = 0.05 #: 基准能是哪几种状态 diff --git a/backend/app/eval/gold.py b/backend/app/eval/gold.py index 4b584f1b..e67f8271 100644 --- a/backend/app/eval/gold.py +++ b/backend/app/eval/gold.py @@ -3,8 +3,13 @@ **期望不靠人填数，也不冻结数值。** 每次要期望（保存题目时试算一次、评测时作答前后各一次）都在业务库上重算一遍： - SQL 题、从运行认定：建一张「入口 → 调用工具（db_query__数据源，args.sql = 按截至时间渲染好的 SQL）→ 出口」的裸图， - 以来源 eval 跑探索运行。期望值取这次运行封存的查询快照里指定的格、列或整张表（cases.py 的 expectations）。从运行认定 + 以来源 eval 发起探索运行。期望值取这次运行封存的查询快照里指定的格、列或整张表（cases.py 的 expectations）。从运行认定 的题和 SQL 题算法一样，只是 SQL 出自某次运行里的一条查询、记着谁在什么时候认定的。 +- 分析期望（贡献、判定）：在同一份查询快照上用平台的算子算一次分析（app.analysis.run，和作答里的分析工具、分析卡同一个 + 函数；随数据变化的调用错误记成算不了，refuse_data_errors），判定就是期望（app/eval/judgements.py）。阈值用这次评测的 + 分析设置快照（Context.analysis，和作答的运行同一份；试算时现读设置），遮罩列、目录事实、现查表的日期范围和分析工具 + 同一套取法（app/tools/analysis.py）。算不了的这一项照实记（评分时不能评分），参数和期望 SQL 的结果对不上（列不在、期次 + 不在）的整题没有可用的值。 - 口径题：app/ledger/derive.caliber_probe 从钉住版本拷出口径卡读的取数节点、接一张 caliber_from 固定到那一版的口径卡， 跑探索运行，期望值就是口径卡的指标（取整、单位、比率都由口径卡自己定）。不可复用的卡拒收并给原因。 @@ -31,7 +36,8 @@ unresolved_caliber_upgrades（和正式运行用的是同一个判断），再 - incomparable：旧题归档，按新版本另建一道新题、重开基线。 **作答前后各算一次**（bracket / settle，第 3 波的执行调它）：两次一样才判；不一样就记「作答期间数据变了，不判」。A/B 两臂 -共用同一次算出的期望：在 answer 里同时跑两臂即可。 +共用同一次算出的期望：在 answer 里同时跑两臂即可。两次比的是期望值本身：分析结论里随运行当天、现查时刻、输入快照的 +工件变的几项不比（judgements.steady）。 """ from __future__ import annotations @@ -157,6 +163,8 @@ class Context: #: 记进运行 origin 的其余字段（臂、第几次重复、试算……） extra: Mapping[str, Any] = field(default_factory=dict) timeout_s: float = TRIAL_TIMEOUT_S + #: 这次评测的分析设置快照（runner 开评时拍一份，作答的运行带同一份）；没有（试算）时现读设置 + analysis: Mapping[str, Any] | None = None def variables(self) -> dict[str, str]: return variable_values(self.as_of, self.tz, self.cron) @@ -387,6 +395,83 @@ def _values(expectations: Sequence[Mapping[str, Any]], pick: Callable[[Mapping[s return {e["id"]: pick(e) for e in expectations if e.get("kind") in NEEDS_GOLD} +# ========================================================================== +# 分析期望：在期望 SQL 的快照上算一次分析（app/eval/judgements.py） +# ========================================================================== + + +@dataclass +class _Analyses: + """一道题的几项分析期望共用的那几样：输入快照（含遮罩列）、阈值、算子的运行期依赖。用到时才备（_prepared）。""" + + snapshot: dict[str, Any] + artifact: str + run_id: str + ctx: Context + ready: tuple[Any, dict[str, Any], Any] | None = None + + +async def _prepared(found: _Analyses) -> tuple[Any, dict[str, Any], Any]: + """(算子的输入快照, 阈值, 运行期依赖)：遮罩列、目录事实、现查表的日期范围同分析工具（app/tools/analysis.py），按这次 + 探针运行固定的数据版本解析数据源；阈值是这次评测的分析设置快照，没有（试算）现读设置。""" + if found.ready is not None: + return found.ready + from app.db.base import SessionLocal + from app.db.models import Run + from app.eval.judgements import GOLD_REF + from app.tools.analysis import context_of, input_snapshot, masked_of, settings_of + from app.tools.registry import ToolContext + + inp = input_snapshot(found.snapshot, ref=GOLD_REF, artifact=found.artifact, node_id=QUERY_NODE, + masked=await masked_of(found.snapshot, found.artifact)) + async with SessionLocal() as session: + row = await session.get(Run, found.run_id) + versions = dict(row.data_versions) if row is not None and row.data_versions is not None else None + tool_ctx = ToolContext(run_id=found.run_id, node_id=QUERY_NODE, data_versions=versions, mask_results=True, + force_readonly=True, fixed_scope=True) + context = await context_of([inp], tool_ctx=tool_ctx) + found.ready = (inp, await settings_of(found.ctx.analysis), context) + return found.ready + + +async def _analysis_value(exp: Mapping[str, Any], at: int, found: _Analyses) -> dict[str, Any]: + """一项分析期望的期望值（judgements.expected）。算不了（历史不够、本期不完整……）照实记成这一项的结论；参数和期望 SQL + 的结果对不上（列不在、期次不在、不是乘积关系……）抛 _NoValue（整题没有可用的值，试算时照实交回去改）。""" + import app.analysis as analysis_ops # 按模块取 run：测试换替身时只换模块上的这一个函数 + from app.analysis.contract import CallError + from app.eval import judgements + + where = f"第 {at + 1} 个期望（{'贡献' if exp.get('kind') == 'contribution' else '判定'}）" + if found.snapshot.get("truncated"): + raise _NoValue("期望 SQL 的结果被截断（只取回了前一部分行），分析期望算不准：请缩小范围或在 SQL 里汇总") + inp, settings, context = await _prepared(found) + try: + request = judgements.request_of(exp, found.ctx.variables(), found.snapshot) + except judgements.ParamsInvalid as e: + raise _NoValue(f"{where}的{e.message}") from None + try: + result = await analysis_ops.run(request, lambda name: inp if name == judgements.GOLD_REF else None, + settings=settings, context=context, refuse_data_errors=True) + except CallError as e: + raise _NoValue(f"{where}的分析参数和期望 SQL 的结果对不上：{e.message}") from None + return judgements.expected(exp, result) + + +async def _sql_values(case: Mapping[str, Any], snapshot: dict[str, Any], artifact: str, run_id: str, + ctx: Context) -> dict[str, dict[str, Any]]: + """SQL 题每个期望的值（按期望的先后）：取格、列、整张表的照快照取，分析期望在同一份快照上算。取不到抛 _NoValue。""" + from app.eval.judgements import ANALYSIS_KINDS + + found = _Analyses(snapshot=snapshot, artifact=artifact, run_id=run_id, ctx=ctx) + out: dict[str, dict[str, Any]] = {} + for at, exp in enumerate(case["expectations"]): + if exp.get("kind") not in NEEDS_GOLD: + continue + out[exp["id"]] = await _analysis_value(exp, at, found) if exp.get("kind") in ANALYSIS_KINDS \ + else _sql_value(exp, snapshot) + return out + + # ========================================================================== # SQL 探针（SQL 题、从运行认定） # ========================================================================== @@ -437,7 +522,7 @@ async def _sql_gold(case: Mapping[str, Any], ctx: Context) -> GoldOutcome: if problem := masked_problem(snapshot): return _outcome("refused", ctx.as_of, problem, GOLD_MASKED, run_ids=runs, rendered=rendered) try: - values = _values(case["expectations"], lambda e: _sql_value(e, snapshot)) + values = await _sql_values(case, snapshot, artifact_id, ran.run_id, ctx) except _NoValue as e: return _outcome("no_value", ctx.as_of, str(e), GOLD_NO_VALUE, run_ids=runs, rendered=rendered) return _outcome("ok", ctx.as_of, sets=[GoldSet(run_id=ran.run_id, snapshot=artifact_id, values=values)], @@ -622,10 +707,23 @@ class Settled: "after": self.after.to_dict() if self.after is not None else None} +def stable_sets(sets: Sequence[Any]) -> list[list[Any]]: + """比较用的几份期望值：[[版本, {期望 id: 值}]]，分析结论里随运行当天、现查时刻、输入快照的工件变的几项去掉 + （judgements.steady）。sets 是 GoldSet 或它的 to_dict。作答前后两次（_fingerprint）、和上一次比的「数据变了」 + （runner.gold_key）都按它。""" + from app.eval.judgements import steady + + out = [] + for s in sets: + version, values = (s.version, s.values) if isinstance(s, GoldSet) else (s.get("version"), s.get("values")) + out.append([version, {k: steady(v) for k, v in values.items()} if isinstance(values, Mapping) else values]) + return out + + def _fingerprint(outcome: GoldOutcome) -> str: from app.core.artifact_store import canonical_json - return canonical_json([[s.version, s.values] for s in outcome.sets]) + return canonical_json(stable_sets(outcome.sets)) def settle(before: GoldOutcome, after: GoldOutcome | None) -> Settled: @@ -717,4 +815,4 @@ __all__ = ["CANCEL_GRACE_S", "Context", "DISPOSE_INVALID", "DISPOSE_NOTHING", "D "GOLD_DRIFTED", "GOLD_FAILED", "GOLD_MASKED", "GOLD_NO_VALUE", "GOLD_TIMEOUT", "GoldOutcome", "GoldSet", "QUERY_NODE", "ROLES", "STATUSES", "Settled", "TRIAL_TIMEOUT_S", "UPGRADE_LABEL", "UPGRADE_PENDING", "UpgradeState", "bracket", "compute_gold", "masked_problem", "plan_dispose", "settle", "sql_probe_graph", - "upgrade_state"] + "stable_sets", "upgrade_state"] diff --git a/backend/app/eval/judgements.py b/backend/app/eval/judgements.py new file mode 100644 index 00000000..dedb09aa --- /dev/null +++ b/backend/app/eval/judgements.py @@ -0,0 +1,520 @@ +"""分析期望：贡献（contribution）、判定（verdict）的算法和评分（docs/design/analysis-round/00-总方案.md §3.8）。 + +**期望怎么算。** 期望 SQL 在业务库上查一次（gold.py 的 SQL 探针，和别的期望同一份快照），在这份快照上用平台的算子 +（app.analysis.run，和作答里的分析工具、分析卡同一个函数）算一次分析，判定就是期望： + +- 贡献：变化拆解（可加、比率、乘积），参数同分析工具（按维度 by、期次 period、基期 base、本期 target、度量 measure、分子 + numerator、分母 denominator、乘积 product、因子 factors、叫法 label），另写 top_n（缺省 1）：期望点名主要来源的前几段。 +- 判定：波动基线、趋势判断、差异检验、可比性检查，参数同对应的分析工具。 + +算子要的输入只有这一份快照（编号 GOLD_REF），不写 query、两份快照、总数、下钻。参数里的文字可以带 {{ schedule.* }}，按评测 +的截至时间渲染（同期望 SQL）；期望 SQL 只返回两期时可以不写基期、本期：先的一期是基期、后的一期是本期。阈值用这次评测的 +设置快照（和作答的运行同一份，runner.py），不写在期望里；算不了的（历史不够、本期不完整……）这一项不能评分，原因照实写。 + +**判定类**（JUDGEMENTS）：期望和答案里的分析结论都按判定码归到同一套类再比——贡献：source 主要来源（可加的 driver、 +concentrated，乘积的 driver）、dispersed 分散、flat 基本持平、mix 主要来自结构变化、rate 主要来自各段自身的变化、mixed +两方面都有影响（比率、乘积的 both）；判定：normal 常规波动（within_band）、anomaly 超出常规波动（outside_band、 +far_outside_band，看方向）、rising / falling / no_trend、differs 差异站得住（看哪两组、方向）、no_evidence 不足以下结论、 +comparable 可比（comparable、conditional）/ incomparable 不可比。 + +**评分**（score；沿用五种结论）： +- 对且可追溯：答案引用了（[[a:]]、[[v:A…]]、[[table:A…]]、[[see:A…]]）判定对得上的分析结论（同一个算子，变化拆解还要同一种 + 模式；主要来源要点名期望的那几段，方向一致；差异检验比的是同两组）；引用它的那句话里的判断词却和期望对不上的不算。 +- 对但没引用：没引用对得上的结论，但正文按说法表（engine/analysis_claims，同一套词表、同一套绑定规则）写对了：说法表认出的 + 这一类判断词，拿期望的那份分析结论核对得上（点名的段、方向、连续几期都核）。可比性没有对应的判断词，只能引用。 +- 错：引用的分析判定对不上、正文写的和期望相反（把常规波动说成异常、把上升说成下降、主要来源说成别的段），或者根本没做也 + 没说。 +- 不能评分：期望算不了。 +""" +from __future__ import annotations + +import copy +import re +import unicodedata +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import replace +from typing import Any + +from app.analysis.contract import OP_LABELS, method_label + +#: 两种分析期望 +ANALYSIS_KINDS = ("contribution", "verdict") +#: 判定期望能用的分析（贡献恒为变化拆解） +VERDICT_OPS = ("baseline", "trend", "differ", "comparable") +#: 判定类，见模块说明。前六个是贡献的，后九个是判定的 +JUDGEMENTS = ("source", "dispersed", "flat", "mix", "rate", "mixed", + "normal", "anomaly", "rising", "falling", "no_trend", "differs", "no_evidence", "comparable", + "incomparable") +#: 分析的输入在算子里的名字（报错里照它说：「期望 SQL 里没有『地区』这一列」） +GOLD_REF = "期望 SQL" +#: 贡献期望最多点名前几段 +TOP_N_MAX = 10 +#: 期望里不写的参数：输入就是期望 SQL 这一份快照，没有两份快照、总数、下钻 +_NOT_IN_EXPECTATION = frozenset({"query", "base_query", "target_query", "headline", "parent"}) +#: 期望 SQL 只返回两期时，这几种分析可以不写基期、本期 +_TWO_PERIODS = frozenset({"decompose", "differ", "comparable"}) + +#: 判定码 → 判定类（按算子、模式） +_JUDGEMENT_OF: dict[tuple[str, str | None], dict[str, str]] = { + ("decompose", "additive"): {"driver": "source", "concentrated": "source", "dispersed": "dispersed", "flat": "flat"}, + ("decompose", "ratio"): {"mix_driven": "mix", "rate_driven": "rate", "both": "mixed", "flat": "flat"}, + ("decompose", "factor"): {"driver": "source", "both": "mixed", "flat": "flat"}, + ("baseline", None): {"within_band": "normal", "outside_band": "anomaly", "far_outside_band": "anomaly"}, + ("trend", None): {"rising": "rising", "falling": "falling", "no_trend": "no_trend"}, + ("differ", None): {"differs": "differs", "no_evidence": "no_evidence"}, + ("comparable", None): {"comparable": "comparable", "conditional": "comparable", "incomparable": "incomparable"}, +} +#: 判定类要看方向的：主要来源、结构或自身为主、分散（总变化的涨跌）、超出常规波动（本期变化的涨跌）、差异站得住（A 相对 B） +_DIRECTED = frozenset({"source", "mix", "rate", "mixed", "dispersed", "anomaly", "differs"}) +#: 说法表的类（engine/analysis_claims 的 Claim.key）→ 它说出的判定类；「连续…」「趋势」看方向，见 _said_judgement +_CLAIM_JUDGEMENT = {"source": "source", "dispersed": "dispersed", "mix": "mix", "anomaly": "anomaly", "normal": "normal", + "no_trend": "no_trend", "differ": "differs", "no_diff": "no_evidence"} +#: 期望是这一类时，正文里哪几类判断词算说出了它（别的类只看对不对得上：对不上算说反了） +_AFFIRMS: dict[str, frozenset[str]] = { + "source": frozenset({"source"}), "dispersed": frozenset({"dispersed"}), "mix": frozenset({"source", "mix"}), + "rate": frozenset({"source"}), "mixed": frozenset({"mix"}), "flat": frozenset(), "normal": frozenset({"normal"}), + "anomaly": frozenset({"anomaly"}), "rising": frozenset({"trend"}), "falling": frozenset({"trend"}), + "no_trend": frozenset({"no_trend"}), "differs": frozenset({"differ"}), "no_evidence": frozenset({"no_diff"}), + "comparable": frozenset(), "incomparable": frozenset(), +} +#: 判断词所在的句子最多留多少字给人看 +_SNIPPET_MAX = 40 + + +class ParamsInvalid(ValueError): + """期望的分析参数写得不对（保存题目时、算期望时都可能）。message 给人看，key 是出错的那个参数。""" + + def __init__(self, message: str, key: str | None = None) -> None: + super().__init__(message) + self.message = message + self.key = key + + +# ========================================================================== +# 参数：校验（cases.py 保存题目时）、组成分析请求（gold.py 算期望时） +# ========================================================================== + + +def param_names(op: str) -> list[str]: + """期望里这种分析可以写的参数（分析工具的参数去掉输入、总数、下钻），按工具里的先后。""" + from app.tools.builtin.analysis import param_names as tool_params + + return [k for k in tool_params(op) if k not in _NOT_IN_EXPECTATION] + + +def op_text(op: str, mode: str | None = None) -> str: + """「变化拆解（可加）」「波动基线」：叫法只取契约那一份。""" + return method_label(op, mode) if mode else OP_LABELS.get(op, op) + + +def template_texts(params: Mapping[str, Any]) -> Iterable[tuple[str, str]]: + """参数里的文字（含列表、成对列表里的）：(参数名, 文字)。{{ schedule.* }} 的校验、渲染按它走。""" + for key, value in params.items(): + if isinstance(value, str): + yield key, value + elif isinstance(value, list): + for item in value: + if isinstance(item, str): + yield key, item + elif isinstance(item, list): + yield from ((key, x) for x in item if isinstance(x, str)) + + +def check_params(op: str, mode: str | None, params: Any) -> dict[str, Any]: + """参数的形状：只能写这种分析的参数，不能写阈值，类型照分析工具的参数定义。交回规范化的参数（去掉空值）。不对抛 + ParamsInvalid。列在不在期望 SQL 的结果里要等试算才知道（算子按快照核）。""" + from pydantic import ValidationError + + from app.analysis.contract import DEFAULTS, LABELS + from app.tools.builtin.analysis import ARGS + + if params is None: + params = {} + if not isinstance(params, Mapping): + raise ParamsInvalid("分析参数需要是一个对象") + allowed = param_names(op) + out: dict[str, Any] = {} + for key, value in params.items(): + if key in DEFAULTS: + label = LABELS[key] if LABELS[key].startswith("「") else f"「{LABELS[key]}」" + raise ParamsInvalid(f"参数里写了阈值{label}：阈值不写在期望里，评测按这次评测的分析设置算（和作答用同一份）", + key) + if key not in allowed: + raise ParamsInvalid(f"参数里没有「{key}」，{op_text(op)}可以写：{'、'.join(allowed)}", key) + if value is None or value == "" or value == []: + continue + out[key] = copy.deepcopy(value) + probe = {**out, "query": GOLD_REF, **({"mode": mode} if mode else {})} + try: + ARGS[op].model_validate(probe) + except ValidationError as e: + error = e.errors()[0] + key = str((error.get("loc") or ("",))[0]) + field = ARGS[op].model_fields.get(key) + said = (field.description or "") if field is not None else "" + verb = "没写" if error.get("type") == "missing" else "写得不对" + raise ParamsInvalid(f"参数「{key}」{verb}" + (f"：{said}" if said else ""), key) from None + return out + + +def _render(value: Any, variables: Mapping[str, str]) -> Any: + from app.eval.cases import render + + if isinstance(value, str): + return render(value, variables) + if isinstance(value, list): + return [_render(v, variables) for v in value] + return value + + +def _two_periods(snapshot: Mapping[str, Any], period: str) -> tuple[Any, Any]: + """期望 SQL 只返回两期时的基期、本期：按期次先后（periods.parse_compared 认得出的按它，否则按写法）。""" + from app.analysis import periods + + columns = [str(c) for c in snapshot.get("columns") or []] + hits = [i for i, c in enumerate(columns) if c == period] or \ + [i for i, c in enumerate(columns) if c.lower() == period.lower()] + if not hits: + raise ParamsInvalid(f"期望 SQL 的结果里没有期次列「{period}」（结果的列：{'、'.join(columns) or '（没有）'}）", + "period") + i = hits[0] + values = list(dict.fromkeys(row[i] for row in snapshot.get("rows") or [] + if isinstance(row, (list, tuple)) and i < len(row) and row[i] is not None)) + if len(values) != 2: + raise ParamsInvalid(f"期望 SQL 的期次列「{period}」里有 {len(values)} 期：不写基期、本期时，期望 SQL 只能返回两期" + "（先的一期是基期、后的一期是本期）", "base") + parsed = periods.parse_compared(values, str(snapshot.get("sql") or ""), columns[i], snapshot.get("dialect")) + if parsed is not None: + items, _ = parsed + order = sorted(range(2), key=lambda k: items[k].ordinal) + else: + order = sorted(range(2), key=lambda k: str(values[k])) + return values[order[0]], values[order[1]] + + +def request_of(exp: Mapping[str, Any], variables: Mapping[str, str], snapshot: Mapping[str, Any]) -> dict[str, Any]: + """一项分析期望 → 一次分析的请求（contract.AnalysisRequest）：参数按截至时间渲染，输入是期望 SQL 这一份快照；只返回两期 + 又没写基期、本期的按两期补上。不对抛 ParamsInvalid。""" + from pydantic import ValidationError + + from app.tools.builtin.analysis import ARGS, request_of as tool_request + + op, mode = str(exp.get("op")), exp.get("mode") + params = {k: _render(v, variables) for k, v in (exp.get("params") or {}).items()} + if op in _TWO_PERIODS and isinstance(params.get("period"), str) and params.get("base") in (None, "") \ + and params.get("target") in (None, ""): + params["base"], params["target"] = _two_periods(snapshot, params["period"]) + params["query"] = GOLD_REF + if mode: + params["mode"] = mode + try: + args = ARGS[op].model_validate(params) + except ValidationError as e: + key = str((e.errors()[0].get("loc") or ("",))[0]) + raise ParamsInvalid(f"参数「{key}」写得不对", key) from None + return tool_request(op, args) # type: ignore[return-value] + + +# ========================================================================== +# 判定类 +# ========================================================================== + + +def judgement_code(op: Any, mode: Any, code: Any) -> str | None: + """一份分析结论的判定码 → 判定类；没有对应的（拆解对不上总数）为 None。""" + key = (str(op), str(mode) if op == "decompose" else None) + return _JUDGEMENT_OF.get(key, {}).get(str(code)) if code else None + + +def _segment(subject: Any) -> str: + """判定点名的一段（维度键是一个列表，两列用「·」连，同判定语的写法）。""" + from app.analysis.phrases import key_text + + return key_text(list(subject)) if isinstance(subject, (list, tuple)) else str(subject) + + +def norm(name: Any) -> str: + """比段名、因子名用的写法：全半角、大小写归一，去掉首尾空白。""" + return unicodedata.normalize("NFKC", str(name or "")).casefold().strip() + + +def expected(exp: Mapping[str, Any], result: Mapping[str, Any]) -> dict[str, Any]: + """期望 SQL 上算出的那份分析结论 → 这一项的期望值（GoldSet.values 里的一项）：判定类、点名的段、方向、判定语，连同整份 + 结论（评分时说法表按它核对正文；比较作答前后两次期望时去掉随运行当天、现查时刻变的几项，见 steady）。算不了的照实记。""" + op, mode = result.get("op"), result.get("mode") + out: dict[str, Any] = {"kind": exp.get("kind"), "status": result.get("status"), "op": op, "mode": mode, + "method": op_text(str(op), mode), "phrase": result.get("phrase") or ""} + if result.get("status") != "ok": + out.update(reason_code=result.get("reason_code"), reason=result.get("reason") or "原因未知") + out["result"] = copy.deepcopy(dict(result)) + return out + verdict = result.get("verdict") or {} + judgement = judgement_code(op, mode, verdict.get("code")) + subjects = [_segment(s) for s in verdict.get("subjects") or []] + if judgement == "source": + subjects = subjects[:int(exp.get("top_n") or 1)] + elif judgement not in ("differs", "no_evidence"): + subjects = [] + out.update(verdict=verdict.get("code"), judgement=judgement, subjects=subjects, + direction=verdict.get("direction") if judgement in _DIRECTED else None, + criterion=result.get("criterion") or "") + out["result"] = copy.deepcopy(dict(result)) + return out + + +def steady(value: Any) -> Any: + """比较用的期望值：分析结论里随运行当天（params.today）、现查时刻（probes[*].at）、输入快照的工件（inputs[*].artifact， + 快照里记着查询用时，同一份数据每次查出来工件也不同）变的几项去掉。别的期望原样。作答前后两次期望（gold._fingerprint）、 + 和上一次比的「数据变了」（runner.gold_key）都按它比。""" + if not (isinstance(value, Mapping) and value.get("kind") in ANALYSIS_KINDS and isinstance(value.get("result"), Mapping)): + return value + result = copy.deepcopy(dict(value["result"])) + if isinstance(result.get("params"), dict): + result["params"].pop("today", None) + for probe in result.get("probes") or []: + if isinstance(probe, dict): + probe.pop("at", None) + for item in result.get("inputs") or []: + if isinstance(item, dict): + item.pop("artifact", None) + return {**value, "result": result} + + +def view(value: Mapping[str, Any]) -> dict[str, Any]: + """评分结果里给人看的那一份（期望、答案各一份）：判定类、点名的段、方向、判定语、哪一份结论（A 编号）。界面照原样写， + 叫法在前端 termsEval。""" + keys = ("judgement", "subjects", "direction", "phrase", "method", "alias", "status", "text") + out = {k: copy.deepcopy(value[k]) for k in keys if value.get(k) not in (None, "", [])} + if value.get("status") == "refused" and value.get("reason"): + out["reason"] = value["reason"] + return out + + +# ========================================================================== +# 答案里的分析结论 +# ========================================================================== + + +def record_of(result: Mapping[str, Any], *, artifact: str, info: Mapping[str, Any] | None = None) -> dict[str, Any]: + """答案运行台账里的一份分析结论 → 评分用的那一份：算子、模式、判定类、点名的段、方向、判定语；输入（查询快照、上一层 + 结论的工件）给逐题 E1 找产出答案的查询。alias、cited 由读答案的地方填。""" + verdict = result.get("verdict") or {} + op, mode = result.get("op"), result.get("mode") + out: dict[str, Any] = { + "artifact": artifact, "op": op, "mode": mode, "method": op_text(str(op), mode), "status": result.get("status"), + "verdict": verdict.get("code"), "judgement": judgement_code(op, mode, verdict.get("code")), + "subjects": [_segment(s) for s in verdict.get("subjects") or []], + "phrase": result.get("phrase") or "", "cited": [], + "inputs": list((info or {}).get("inputs") or [i.get("artifact") for i in result.get("inputs") or [] + if isinstance(i, Mapping) and i.get("artifact")]), + } + judgement = out["judgement"] + out["direction"] = verdict.get("direction") if judgement in _DIRECTED else None + if result.get("status") != "ok": + out["reason"] = result.get("reason") or "原因未知" + if (info or {}).get("node_id"): + out["node_id"] = info["node_id"] + return out + + +def _same_family(want: Mapping[str, Any], record: Mapping[str, Any]) -> bool: + """答案里的这份结论是不是这一项期望的同类分析：同一个算子；变化拆解还要同一种模式（可加、比率、乘积说的不是一回事）。""" + if record.get("op") != want.get("op"): + return False + return want.get("op") != "decompose" or record.get("mode") == want.get("mode") + + +def _names(items: Iterable[Any]) -> set[str]: + return {norm(x) for x in items if norm(x)} + + +def mismatch(want: Mapping[str, Any], got: Mapping[str, Any]) -> str | None: + """答案里一份同类的分析结论和期望对不对得上：对得上为 None，否则是对不上的那一句（给人看）。""" + if got.get("status") != "ok": + return f"这份分析没有做成（{got.get('reason') or '原因未知'}）" + judgement = want.get("judgement") + if got.get("judgement") != judgement: + return "判定和期望不是一类" + if judgement in ("differs", "no_evidence"): + mine, theirs = [norm(s) for s in got.get("subjects") or []], [norm(s) for s in want.get("subjects") or []] + if len(theirs) == 2 and (len(mine) != 2 or set(mine) != set(theirs)): + return f"比的是{'和'.join(got.get('subjects') or []) or '别的组'}，期望比的是{'和'.join(want.get('subjects') or [])}" + if judgement == "differs" and len(theirs) == 2 and want.get("direction") and got.get("direction"): + same_order = mine == theirs + if (got["direction"] == want["direction"]) != same_order: + return "高低写反了" + return None + if judgement == "source": + missing = [s for s in want.get("subjects") or [] if norm(s) not in _names(got.get("subjects") or [])] + if missing: + return f"主要来源是{'、'.join(got.get('subjects') or []) or '别的段'}，不是{'、'.join(missing)}" + if judgement in _DIRECTED and want.get("direction") and got.get("direction") \ + and want["direction"] != got["direction"]: + return "方向相反" + return None + + +# ========================================================================== +# 正文里的判断词：按说法表认，拿期望的那份分析结论核对（engine/analysis_claims 同一套词表和绑定规则） +# ========================================================================== + +_SENTENCE_END = re.compile(r"(?<=[。！？!?\n])") + + +def sentences_of_doc(doc: Mapping[str, Any], aliases: Iterable[str]) -> list[dict[str, Any]]: + """报告文档里有判断词的句子：{text: 写作者自己写的字（引用渲染出来的值、判定语、引文涂成 \\x00）, names: 点名用的样子 + （引用渲染出来的值留着）, analyses: 这句挂的分析结论编号}。标题、代码块、表格不算（同说法核对）。aliases 是文档目录里 + 分析结论的编号。""" + from app.engine import analysis_claims as said + from app.engine.evidence import iter_units + + known = {a: {} for a in aliases} + out: list[dict[str, Any]] = [] + for block, unit in iter_units(dict(doc)): + if block.get("type") in ("heading", "code", "table") or unit.get("kind") in ("heading", "code") \ + or unit.get("loc") is not None: + continue + sentence = said._sentence(unit, unit.get("cites") or [], known) + if sentence is not None and said.scan(sentence.text): + out.append({"text": sentence.text, "names": sentence.names, "analyses": list(sentence.analyses)}) + return out + + +def sentences_of_text(text: str) -> list[dict[str, Any]]: + """没有报告文档时（成果里的文字）：按句末标点切开，有判断词的句子。""" + from app.engine import analysis_claims as said + + out = [] + for piece in _SENTENCE_END.split(text or ""): + piece = piece.strip() + if piece and said.scan(piece): + out.append({"text": piece, "names": piece, "analyses": []}) + return out + + +def _snippet(names: str, start: int, end: int) -> str: + """判断词连同它所在的分句（给人看）。""" + from app.engine import analysis_claims as said + + left, right = said._clause(names, start, end) + text = names[left:right].replace("\x00", "").strip() + return text[:_SNIPPET_MAX] + ("…" if len(text) > _SNIPPET_MAX else "") + + +def _said_judgement(claim: str, direction: str | None) -> str | None: + if claim == "trend": + return {"up": "rising", "down": "falling"}.get(direction or "") + return _CLAIM_JUDGEMENT.get(claim) + + +def said_in(want: Mapping[str, Any], sentences: Sequence[Mapping[str, Any]], *, + only: Iterable[str] | None = None) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """正文里和这项期望同一类的判断词，拿期望的那份分析结论（want.result）按说法表的绑定规则核对：(说出了期望的那几处, + 和期望对不上的那几处)。每处 {text: 所在分句, judgement: 它说出的判定类, subjects: 点名的段, reason: 对不上的原因}。 + only 给了时只看挂了这几份结论的句子。 + + 规则同 engine/analysis_claims.check_sentence（同一个 _check）：因果词点名的是期望拆解里的段的按来源核；点名的段要在期望 + 点名的段里，拖累、拉动按那一段的涨跌核；「连续 N 期」按末尾连续同向的期数核。另外按期望的方向核主要来源那句话的涨跌 + （「下降主要来自华东」）。""" + from app.engine import analysis_claims as said + + gold = want.get("result") or {} + op = want.get("op") + affirms = _AFFIRMS.get(str(want.get("judgement")), frozenset()) + lex = said.lexicon() + wanted = set(only) if only is not None else None + good: list[dict[str, Any]] = [] + bad: list[dict[str, Any]] = [] + for raw in sentences: + if wanted is not None and not wanted & set(raw.get("analyses") or []): + continue + sentence = said.Sentence(text=str(raw.get("text") or ""), names=str(raw.get("names") or "")) + for hit in said.scan(sentence.text): + claim = lex.claim(hit.claim) + left, right = said._clause(sentence.text, hit.start, hit.end) + names = sentence.names[left:right] + spans = said._object(sentence, hit, left, right) + if claim.key == "causal": + if op != "decompose": + continue + cause = said._cause(sentence, hit, left, right) + if not said._named(gold, cause): + continue + hit, claim, spans = replace(hit, claim="source", side="cause"), lex.claim("source"), (cause,) + if claim.op != op: + continue + miss = said._check(lex, hit, gold, names, spans=spans) + reason = miss.human if miss is not None else None + named = next((found for span in spans if (found := said._named(gold, span))), []) + if reason is None and hit.claim == "source" and hit.side != "pull" and want.get("direction") \ + and hit.direction and hit.direction != want["direction"]: + reason = "涨跌写反了" + item = {"text": _snippet(sentence.names, hit.start, hit.end), + "judgement": _said_judgement(hit.claim, hit.direction), "subjects": named} + if reason is not None: + bad.append({**item, "reason": reason}) + elif hit.claim in affirms: + good.append(item) + return good, bad + + +def _covered(want: Mapping[str, Any], good: Sequence[Mapping[str, Any]]) -> list[str]: + """主要来源期望点名的段里，正文说对的那几处一处都没点到的（top_n 要点名的前几段都要说到）。""" + if want.get("judgement") != "source": + return [] + named = _names(s for item in good for s in item.get("subjects") or []) + return [s for s in want.get("subjects") or [] if norm(s) not in named] + + +# ========================================================================== +# 评分 +# ========================================================================== + + +def score(exp: Mapping[str, Any], value: Mapping[str, Any], answer: Any) -> dict[str, Any]: + """一项分析期望的结论（score.OUTCOMES 之一）。value 是期望值（expected 的形状），answer 是 score.Answer（analyses： + 答案运行台账里的分析结论，cited 记着报告怎么引用了它；sentences：有判断词的句子）。纯计算。""" + want = view(value) + if value.get("status") != "ok": + return {"outcome": "ungradable", "want": want, "reason": f"期望算不了：{value.get('reason') or '原因未知'}"} + if value.get("judgement") is None: + return {"outcome": "ungradable", "want": want, "reason": "期望的分析没有给出判定"} + records = [r for r in getattr(answer, "analyses", None) or [] if _same_family(value, r)] + cited = [r for r in records if r.get("cited") and r.get("alias")] + sentences = list(getattr(answer, "sentences", None) or []) + matched = [r for r in cited if mismatch(value, r) is None] + if matched: + found = matched[0] + _, bad = said_in(value, sentences, only=[r["alias"] for r in matched]) + if bad: + return {"outcome": "wrong", "want": want, "found": view({**found, "text": bad[0]["text"]}), + "reason": f"引用了 {found['alias']}，同一句却写成「{bad[0]['text']}」：{bad[0]['reason']}"} + return {"outcome": "traced", "want": want, "found": view(found)} + good, bad = said_in(value, sentences) + if good: + missing = _covered(value, good) + if not missing: + return {"outcome": "untraced", "want": want, "found": view(good[0])} + return {"outcome": "wrong", "want": want, "found": view(good[0]), + "reason": f"正文只点名了{'、'.join(s for g in good for s in g.get('subjects') or []) or '别的段'}，" + f"期望的主要来源还有{'、'.join(missing)}"} + if cited: + first = cited[0] + return {"outcome": "wrong", "want": want, "found": view(first), + "reason": f"答案引用的 {first['alias']}（{first.get('phrase') or first.get('method')}）" + f"和期望对不上：{mismatch(value, first)}"} + if bad: + return {"outcome": "wrong", "want": want, "found": view(bad[0]), + "reason": f"正文写的「{bad[0]['text']}」和期望对不上：{bad[0]['reason']}"} + method = op_text(str(value.get("op")), value.get("mode")) + if records: + first = records[0] + shown = f"（{first['alias']}）" if first.get("alias") else "" + return {"outcome": "wrong", "want": want, "found": view(first), + "reason": f"答案做了{method}{shown}，但没有引用，正文里也没有写出这个判断"} + return {"outcome": "wrong", "want": want, "reason": f"答案里没有做{method}，正文里也没有写出这个判断"} + + +__all__ = ["ANALYSIS_KINDS", "GOLD_REF", "JUDGEMENTS", "ParamsInvalid", "TOP_N_MAX", "VERDICT_OPS", "check_params", + "expected", "judgement_code", "mismatch", "norm", "op_text", "param_names", "record_of", "request_of", + "said_in", "score", "sentences_of_doc", "sentences_of_text", "steady", "template_texts", "view"] diff --git a/backend/app/eval/runner.py b/backend/app/eval/runner.py index a552a528..365f4ed3 100644 --- a/backend/app/eval/runner.py +++ b/backend/app/eval/runner.py @@ -95,17 +95,22 @@ def leader_of(run: Any) -> str: def prompt_fingerprint() -> str: - """助手提示词的指纹：节点手册、输出协议、问数据的请求包法、重来的两种措辞、复核的规则。改了任何一段，和上一次比时 - 就标「提示词变了」。""" + """助手提示词的指纹：节点手册、输出协议、问数据的请求包法、重来的两种措辞、复核的规则；平台的分析方法（分析工具的 + guide：id 带版本，连同全文）、写作者的分析规则（写作目录里分析结论那一段、判断词的写法、重写时补的那句）和说法表的 + 版本。改了任何一段，和上一次比时就标「提示词变了」。""" from app.api import copilot from app.chat import turn from app.core.artifact_store import canonical_json, content_hash - from app.engine import review + from app.engine import analysis_claims, evidence, review + from app.tools.builtin.analysis import METHOD_GUIDE parts = [copilot.NODE_REFERENCE, copilot._STREAM_PROTOCOL, copilot._user_message(copilot.GenerateIn(instruction="{question}", intent="answer"), patch=False), turn.RETRY_WITH_BASE, turn.RETRY_FROM_SCRATCH, - str(getattr(review, "REVIEW_SYSTEM", "") or getattr(review, "SYSTEM_PROMPT", ""))] + str(getattr(review, "REVIEW_SYSTEM", "") or getattr(review, "SYSTEM_PROMPT", "")), + METHOD_GUIDE.id, METHOD_GUIDE.text, evidence.ANALYSIS_PROMPT, evidence.ANALYSIS_SEE_ONLY_PROMPT, + analysis_claims.ANALYSIS_RULES_PROMPT, analysis_claims.ANALYSIS_NONE_PROMPT, + analysis_claims.ANALYSIS_REPAIR_PROMPT, analysis_claims.LEXICON_VERSION] return content_hash(canonical_json(parts))[:16] @@ -300,6 +305,8 @@ class EvalExecution(runtime.JobRun): self.copilot_reason = "" self.timeouts: list[bool] = [] self.tasks: set[asyncio.Task[Any]] = set() + #: 这次评测的分析设置快照（有分析期望的题集才拍，_prepare）：算期望、作答的运行都按它 + self.analysis: dict[str, Any] | None = None def lane_of(self, job: Any) -> str | None: return None @@ -354,6 +361,13 @@ class EvalExecution(runtime.JobRun): raise runtime.JobFailed(f"拍目录快照没有成功：{e.message}") from e snapshot = snap.artifact_id versions = await source_versions(session, sources) + # 有分析期望的题集拍一份分析设置（判定用的阈值）：算期望、各臂作答的运行都按这一份（重试沿用拍过的那一份）， + # 评测期间有人改了设置也不会一边按旧阈值判、一边按新阈值判；记进指纹，和上一次比时阈值变了照实标出 + self.analysis = (leader.fingerprint or {}).get("analysis") + if self.analysis is None and self._analyzes(): + from app.api.settings import analysis_settings + + self.analysis = await analysis_settings(session) for run in runs: fp = dict(run.fingerprint or {}) if "sources_before" not in fp: @@ -366,6 +380,8 @@ class EvalExecution(runtime.JobRun): sources_before=versions, schemas={sid: v.get("schema") for sid, v in versions.items()}, cases=_cases_key(results, run.id)) + if self.analysis is not None: + fp["analysis"] = dict(self.analysis) run.fingerprint = fp run.catalog_snapshot = snapshot if (run.variant.get("catalog") or {}).get("mode") in ( "live", "reviewed") else fp.get("snapshot_artifact") @@ -384,6 +400,13 @@ class EvalExecution(runtime.JobRun): units.setdefault(key, _Unit(key[0], key[1], {})).results[r.eval_run_id] = r.id return list(units.values()) + def _analyzes(self) -> bool: + """这次评测的题里有没有分析期望（贡献、判定）。""" + from app.eval.judgements import ANALYSIS_KINDS + + return any(e.get("kind") in ANALYSIS_KINDS for case in self.cases.values() + for e in case.get("expectations") or []) + async def _sources(self, session: AsyncSession) -> list[str]: """这次评测用得到的数据源：题目限定的、期望查的；有不限数据源的问数据题时，全部启用的源（助手会在它们里面挑）。""" from app.db.models import DataSource @@ -490,7 +513,8 @@ class EvalExecution(runtime.JobRun): timeout = float(self.budget.get("per_case_timeout_s") or 900) ctx = gold_rules.Context(as_of=self.as_of, tz=self.suite["timezone"], cron=self.cron, actor=self.actor, suite_id=self.suite["id"], suite_name=self.suite["name"], eval_run_id=self.job_id, - role="gold", extra={"repeat": unit.repeat}, timeout_s=timeout) + role="gold", extra={"repeat": unit.repeat}, timeout_s=timeout, + analysis=self.analysis) async def answer(_before: Any) -> list[Any]: return await asyncio.gather(*(self._answer(arm, unit, case, question, values, timeout) for arm in arms)) @@ -515,7 +539,7 @@ class EvalExecution(runtime.JobRun): case_id=unit.case_id, case_seq=case.get("seq"), actor=self.actor, arm=arm.name, repeat=unit.repeat, assistant=arm.variant.get("assistant") or {}, writer=arm.variant.get("writer") or {}, - view=arm.view) + view=arm.view, analysis=self.analysis) driver = Driver(question, datasource_ids=list(case.get("datasource_ids") or []), ctx=ctx, on_phase=on_phase) try: if case.get("kind") == "workflow": @@ -806,13 +830,15 @@ def _cases_key(results: Sequence[EvalResult], run_id: str) -> str: def gold_key(gold: Mapping[str, Any] | None) -> str | None: - """期望值的指纹（作答前那一份）：和上一次比时认「数据变了」。""" + """期望值的指纹（作答前那一份）：和上一次比时认「数据变了」。分析结论里随运行当天、现查时刻、输入快照的工件变的几项 + 不算（gold.stable_sets：不然分析题每次评测都成了「数据变」）。""" from app.core.artifact_store import canonical_json, content_hash + from app.eval.gold import stable_sets before = (gold or {}).get("before") or {} if (gold or {}).get("status") not in ("ok", "drifted"): return None - return content_hash(canonical_json([[s.get("version"), s.get("values")] for s in before.get("sets") or []]))[:16] + return content_hash(canonical_json(stable_sets(before.get("sets") or [])))[:16] def _sql_shapes(answer: Any) -> list[dict[str, Any]]: diff --git a/backend/app/eval/score.py b/backend/app/eval/score.py index 792f3d1f..7fb78a2e 100644 --- a/backend/app/eval/score.py +++ b/backend/app/eval/score.py @@ -26,6 +26,10 @@ wrong 错、scale 量纲不同（差 10 的整数次幂，比如元和万元、 没有数据」引用了地区那一格，说的仍是没有数据。 - ask：期望反问：作答路径是反问才对。 - text：文字要点，由裁判判（covered / partial / missing / unjudged），单列「模型判断」，不进准确率（评测-D4）。 +- contribution、verdict（分析期望，规则在 app/eval/judgements.py）：答案引用了判定对得上的分析结论（[[a:]]、[[v:A…]]、 + [[table:A…]]、[[see:A…]]）是可追溯；没引用、但正文按说法表写对了是对但没引用；判定、来源、方向对不上，或者答成相反、 + 根本没做也没说是错；期望算不了的这一项不能评分。答案的分析结论从作答运行的台账（kind analysis）读，报告怎么引用了它们 + 从报告文档读；正文里有判断词的句子一并读出来（说法表按句子认）。 **题目结论**（verdict）：路径不对（要答出数的题直接回答了、反问了、只建不跑、建图或运行失败）直接不通过、写明原因； 必需的期望全是 traced / untraced 就通过，另报「可追溯通过」（全是 traced）；有 wrong / scale 的不通过；剩下的（有不能评分的、 @@ -34,8 +38,8 @@ wrong 错、scale 量纲不同（差 10 的整数次幂，比如元和万元、 **逐题的 E1**（sql_match）：产出答案的那条查询和期望的取数（SQL 题是期望 SQL，口径题是口径卡的取数节点）用口径账本同一个 比对函数（ledger/match.compare_queries：先绑参数、再由 explain_sql_diff 判等价）比一次，给出等价与否和差在哪个方面（总方案 -X2：E1 只有一份实现）。哪条是产出答案的：答案引用了就是被引用的；没引用、只查了一条就是那一条（写明依据）；没引用、查了 -几条的不硬比（照实写没法对上是哪条，不计入 E1）。 +X2：E1 只有一份实现）。哪条是产出答案的：答案引用了就是被引用的（引用的是分析结论的，按它的输入找到查询，下钻的顺着上一层 +找下去）；没引用、只查了一条就是那一条（写明依据）；没引用、查了几条的不硬比（照实写没法对上是哪条，不计入 E1）。 """ from __future__ import annotations @@ -52,7 +56,9 @@ logger = logging.getLogger(__name__) #: 评分规则的版本：规则改了就加一，结果里记着按哪一版评的，可以按新规则重评。 #: 2：期望「没有数据」时引用文字格、空值格不算给了数；判错时「答案里的数」给配对上的那一格；答案没引用查询时逐题 E1 #: 不硬比第一条查询 -SCORER_VERSION = 2 +#: 3：两种分析期望（贡献、判定）：引用了判定对得上的分析结论是可追溯，正文按说法表写对了是对但没引用；答案引用分析结论 +#: 时逐题 E1 按它的输入查询比（以前把结论的工件当查询，比不上）；分析结论的判定语不当成答案里的数 +SCORER_VERSION = 3 OUTCOMES = ("traced", "untraced", "wrong", "scale", "ungradable") OUTCOME_LABEL = {"traced": "对且可追溯", "untraced": "对但没引用", "wrong": "错", "scale": "量纲不同", @@ -115,15 +121,38 @@ def _masked_columns(snapshot: Mapping[str, Any]) -> set[str]: return {str(c) for c in snapshot.get("masked_at_query") or []} +def _result_dims(result: Mapping[str, Any], locator: Mapping[str, Any]) -> dict[str, Any]: + """分析结论结果表的一格所在行的维度：记成文字的列（地区、期次、因子的叫法……）。结果表里的数一律是十进制字符串， + 按列类型分，不按值的样子分；遮罩列不会出现在结果表里。""" + table = result.get("table") if isinstance(result.get("table"), Mapping) else {} + columns = [str(c) for c in table.get("columns") or []] + types = table.get("column_types") if isinstance(table.get("column_types"), Mapping) else {} + rows = table.get("rows") or [] + row_index = locator.get("row") + dims: dict[str, Any] = {} + if isinstance(row_index, int) and 0 <= row_index < len(rows) and isinstance(rows[row_index], (list, tuple)): + row = rows[row_index] + dims = {c: row[i] for i, c in enumerate(columns) if i < len(row) and c != locator.get("column") + and row[i] is not None and types.get(c) != "number"} + return {"dims": dims, "masked": False} + + def _cell_facts(cite: Mapping[str, Any], catalog: Mapping[str, Any], cache: dict[str, Any]) -> dict[str, Any]: - """引用的一格：哪份快照、第几行哪一列、这一行的维度（非数值列）、是不是遮罩格。""" + """引用的一格：哪份快照、第几行哪一列、这一行的维度（非数值列）、是不是遮罩格。分析结论的结果表、具名的数也是这样的 + 一格（工件是那份结论）。""" + from app.analysis.contract import LEDGER_KIND, is_result + entry = catalog.get(cite.get("alias") or "") or {} locator = cite.get("locator") or {} artifact = entry.get("artifact") out: dict[str, Any] = {"artifact": artifact, "row": locator.get("row"), "column": locator.get("column"), "dims": {}} + if locator.get("key"): + out["column"] = locator["key"] snapshot = _snapshot(artifact, cache) if snapshot is None: return out + if entry.get("kind") == LEDGER_KIND or is_result(snapshot): + return {**out, **_result_dims(snapshot, locator)} columns = [str(c) for c in snapshot.get("columns") or []] masked = _masked_columns(snapshot) out["masked"] = bool(snapshot.get("masked_whole")) or str(locator.get("column")) in masked @@ -139,7 +168,11 @@ def _cell_facts(cite: Mapping[str, Any], catalog: Mapping[str, Any], cache: dict @dataclass class Answer: - """从封存的作答运行里读出来的答案（评分只看这些）。problem 不为空时整份答案读不了（封存核对不上、没有封存）。""" + """从封存的作答运行里读出来的答案（评分只看这些）。problem 不为空时整份答案读不了（封存核对不上、没有封存）。 + + analyses：作答运行台账里的分析结论（judgements.record_of 的形状：算子、判定类、点名的段、方向、判定语、输入；alias 是 + 报告文档目录里的编号，cited 记报告怎么引用了它：phrase 判定语、value 具名的数、cell 结果表的格、see 句末依据）。 + sentences：正文里有判断词的句子（judgements.sentences_of_doc）。老结果的明细里没有这两项，按空读。""" run_id: str | None text: str = "" @@ -151,18 +184,22 @@ class Answer: tier: str | None = None report: str | None = None problem: str | None = None + analyses: list[dict[str, Any]] = field(default_factory=list) + sentences: list[dict[str, Any]] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: return {"run_id": self.run_id, "text": self.text, "values": self.values, "bare": self.bare, "queries": self.queries, "cited": self.cited, "stats": self.stats, "tier": self.tier, - "report": self.report, "problem": self.problem} + "report": self.report, "problem": self.problem, "analyses": self.analyses, + "sentences": self.sentences} @classmethod def from_dict(cls, raw: Mapping[str, Any]) -> Answer: return cls(run_id=raw.get("run_id"), text=str(raw.get("text") or ""), values=list(raw.get("values") or []), bare=list(raw.get("bare") or []), queries=list(raw.get("queries") or []), cited=list(raw.get("cited") or []), stats=dict(raw.get("stats") or {}), tier=raw.get("tier"), - report=raw.get("report"), problem=raw.get("problem")) + report=raw.get("report"), problem=raw.get("problem"), analyses=list(raw.get("analyses") or []), + sentences=list(raw.get("sentences") or [])) def _output_text(output: Any) -> str: @@ -180,11 +217,66 @@ def _output_text(output: Any) -> str: return "\n".join(parts) +def _ledger_analyses(sealed: Any, cache: dict[str, Any]) -> dict[str, dict[str, Any]]: + """作答运行台账里的分析结论：{结论工件: judgements.record_of}。取不回、不是分析结论的跳过（证据面板照实说取不回）。""" + from app.analysis.contract import is_result + from app.eval.judgements import record_of + + out: dict[str, dict[str, Any]] = {} + for artifact, info in (getattr(sealed, "analyses", None) or {}).items(): + content = _snapshot(artifact, cache) + if is_result(content): + out[artifact] = record_of(content, artifact=artifact, info=info) + return out + + +def _named_analyses(catalog: Mapping[str, Any], found: dict[str, dict[str, Any]], + cache: dict[str, Any]) -> dict[str, dict[str, Any]]: + """报告文档目录里的分析结论编号 → 那一份（found 里的同一个对象，记上 alias）。目录里有、台账里没收的照工件补上。""" + from app.analysis.contract import LEDGER_KIND, is_result + from app.eval.judgements import record_of + + out: dict[str, dict[str, Any]] = {} + for alias, entry in catalog.items(): + if not isinstance(entry, Mapping) or entry.get("kind") != LEDGER_KIND or not entry.get("artifact"): + continue + artifact = str(entry["artifact"]) + if artifact not in found: + content = _snapshot(artifact, cache) + if not is_result(content): + continue + found[artifact] = record_of(content, artifact=artifact, info={"inputs": entry.get("inputs")}) + record = found[artifact] + record.setdefault("alias", str(alias)) + out[str(alias)] = record + return out + + +def _query_inputs(record: Mapping[str, Any], found: Mapping[str, Mapping[str, Any]], queries: Mapping[str, Any], + seen: set[str] | None = None) -> list[str]: + """一份分析结论的输入里的查询快照（逐题 E1 拿它们当产出答案的查询）：下钻的顺着上一层的结论找下去。""" + seen = seen if seen is not None else set() + out: list[str] = [] + for artifact in record.get("inputs") or []: + if artifact in queries: + out.append(artifact) + elif artifact in found and artifact not in seen: + seen.add(artifact) + out += _query_inputs(found[artifact], found, queries, seen) + return out + + +def _cited_as(record: dict[str, Any], how: str) -> None: + if how not in record["cited"]: + record["cited"].append(how) + + async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer: """读一次作答运行的答案。没有运行（直接回答、反问）的只有原文：reply。""" from fastapi import HTTPException from app.api.evidence import _choose, _reports, _sealed + from app.eval import judgements if not run_id: text = reply or "" @@ -209,6 +301,7 @@ async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer "columns": [str(c) for c in snap.get("columns") or []], "row_count": snap.get("row_count", len(rows)), "truncated": bool(snap.get("truncated")), "role": snap.get("role")}) + found = _ledger_analyses(sealed, cache) from app.db.base import SessionLocal from app.db.models import Run @@ -222,6 +315,8 @@ async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer if chosen is None or chosen.doc is None or chosen.hash_ok is False: answer.text = _output_text(final) answer.bare = _numbers(answer.text) + answer.analyses = list(found.values()) + answer.sentences = judgements.sentences_of_text(answer.text) if chosen is not None and chosen.hash_ok is False: answer.problem = "报告文档与哈希不一致，疑似被修改，答案不能当证据" return answer @@ -230,11 +325,22 @@ async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer answer.report = chosen.node_id answer.text = str(doc.get("markdown") or "") answer.stats = dict(doc.get("stats") or {}) - from app.engine.evidence import iter_segments + from app.engine.evidence import iter_segments, iter_units + named = _named_analyses(catalog, found, cache) cited: list[str] = [] for seg in iter_segments(doc): cite = seg.get("cite") or {} + record = named.get(str(cite.get("alias") or "")) if cite.get("status") == "resolved" else None + if record is not None: + # 引用了分析结论：判定语（[[a:A1]]）、具名的数（[[a:A1.键]]）、结果表的格（[[v:A1.r0.列]]、[[table:A1 …]]）。 + # 产出答案的查询是它的输入（逐题 E1）；判定语是一句话，不是答案里的数 + how = "cell" if cite.get("kind") == "cell" else "value" if (cite.get("locator") or {}).get("key") \ + else "phrase" + _cited_as(record, how) + cited += _query_inputs(record, found, sealed.queries) + if how == "phrase": + continue if cite.get("role") == "value" and cite.get("status") == "resolved": item: dict[str, Any] = {"value": cite.get("value"), "kind": cite.get("kind"), "text": seg.get("text"), "segment": seg.get("id"), "alias": cite.get("alias")} @@ -244,11 +350,19 @@ async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer else: item.update(_cell_facts(cite, catalog, cache)) answer.values.append(item) - if item.get("artifact"): + if item.get("artifact") and record is None: cited.append(item["artifact"]) elif seg.get("issue") == "uncited_number" or (seg.get("kind") == "number" and not cite): answer.bare += _numbers(str(seg.get("text") or "")) + for _, unit in iter_units(doc): + for support in unit.get("see") or []: + record = named.get(str(support.get("alias") or "")) if support.get("status") == "resolved" else None + if record is not None: + _cited_as(record, "see") + cited += _query_inputs(record, found, sealed.queries) answer.cited = list(dict.fromkeys(cited)) + answer.analyses = list(found.values()) + answer.sentences = judgements.sentences_of_doc(doc, named) return answer @@ -513,6 +627,10 @@ def _expectation_scores(expectations: Sequence[Mapping[str, Any]], values: Mappi out.append({**base, **_score_set(exp, gold, answer, cache)}) elif kind == "none": out.append({**base, **_score_none(answer)}) + elif kind in ("contribution", "verdict"): + from app.eval import judgements + + out.append({**base, **judgements.score(exp, gold, answer)}) return out diff --git a/backend/app/eval/stats.py b/backend/app/eval/stats.py index 0bfe7239..cf74dce5 100644 --- a/backend/app/eval/stats.py +++ b/backend/app/eval/stats.py @@ -3,6 +3,8 @@ - 准确率和可追溯准确率：分母是能判的题（通过 + 不通过）；不能评分、跳过的单列，既不算对也不算错，没跑的（停止、中断、 到了上限、止损没开评）另列 not_run，出错的另列 errors。都带 Wilson 95% 区间和样本数——题集小、模型有随机性，差两三题就可能被读成「变好了」「变差了」（评测-D10：只展示、不卡门）。 +- 各期望的结论（必需的、文字要点除外）合起来一份，另按期望的种类各一份（outcomes_by_kind：贡献、判定这些分析题和 + 一个数、多行分开看）。 - 作答路径、出具档位（声明了出具的按 formal / degraded / withheld，没声明的单列「探索问答」）、复核结论（verdict × severity） 和自动重来的次数、复核用时超过问数据页等待（25 秒）的题数。 - 结论裁判（打开了才有）：supported / partial / contradicted / insufficient / 未判的句数；文字要点的判断单列。 @@ -77,6 +79,11 @@ def summarize(results: Sequence[Mapping[str, Any]], *, judge_on: bool = False) - traced = [r for r in passed if r.get("traced")] outcomes = Counter(s.get("outcome") for r in done for s in r.get("scores") or [] if s.get("required") and s.get("kind") != "text") + by_kind: dict[str, Counter] = {} + for r in done: + for s in r.get("scores") or []: + if s.get("required") and s.get("kind") != "text": + by_kind.setdefault(str(s.get("kind")), Counter())[s.get("outcome")] += 1 text = Counter(s.get("outcome") for r in done for s in r.get("scores") or [] if s.get("kind") == "text") answered = [r for r in done if r.get("path")] tiers = Counter((r.get("tier") or "explore") for r in answered if r.get("path") == "ran") @@ -113,6 +120,7 @@ def summarize(results: Sequence[Mapping[str, Any]], *, judge_on: bool = False) - "accuracy": rate(len(passed), len(graded)), "traced_accuracy": rate(len(traced), len(graded)), "outcomes": dict(outcomes), + "outcomes_by_kind": {kind: dict(found) for kind, found in by_kind.items()}, "text": dict(text), "paths": dict(Counter(r.get("path") for r in answered)), "tiers": {k: tiers.get(k, 0) for k in TIERS}, diff --git a/backend/tests/test_eval_analysis.py b/backend/tests/test_eval_analysis.py new file mode 100644 index 00000000..44255487 --- /dev/null +++ b/backend/tests/test_eval_analysis.py @@ -0,0 +1,317 @@ +"""分析期望（贡献、判定）的规矩和评分（app/eval/judgements.py、cases.py、score.py；docs/design/analysis-round/00-总方案.md +§3.8）。纯计算：期望的分析结论用 app/analysis/testing 的假结论（形状同真算子），答案直接构造。走真运行的在 +tests/test_eval_analysis_runs.py。地区、门店、数字都是虚构的。 + +- 校验：贡献恒为变化拆解、要选模式；判定是另外四种分析；参数只能写分析工具的那几个、不能写阈值、不能写输入；来源、方向、 + 判定不许人填；文字参数里的变量只认定时任务那份；口径题配不了分析期望。 +- 期望值：判定码归到判定类，主要来源按 top_n 取前几段；比较作答前后两次期望时不看运行当天、现查时刻、输入快照的工件。 +- 评分五种结论：引用了判定对得上的结论 → 对且可追溯；正文按说法表写对了、没引用 → 对但没引用；引用的判定对不上、正文 + 写反了（常规波动说成异常、方向反了、来源说成别的段）、根本没做也没说 → 错；期望算不了 → 不能评分。 +""" +from __future__ import annotations + +import pytest + +import app.db.models # noqa: F401 - conftest 按已导入的模型建表：只跑这一个文件时也要有表 +from app.analysis.testing import fake_result +from app.eval import cases, judgements, stats +from app.eval.cases import CaseInvalid, normalize_case +from app.eval.score import Answer, score_case + +SQL = {"kind": "sql", "datasource_id": "ds1", "sql": "SELECT region AS 地区, week AS 周, SUM(amount) AS 销售额 FROM orders"} +SOURCE = {"id": "e1", "kind": "contribution", "required": True, "op": "decompose", "mode": "additive", + "params": {"by": ["地区"], "period": "周", "measure": "销售额"}, "top_n": 1} +BAND = {"id": "e1", "kind": "verdict", "required": True, "op": "baseline", + "params": {"period": "周", "measure": "订单数"}} + + +def invalid(expectation: dict, **gold) -> CaseInvalid: + with pytest.raises(CaseInvalid) as caught: + normalize_case({"question": "上周销售额为什么下降？", "gold": gold or SQL, "expectations": [expectation]}) + return caught.value + + +# -------------------------------------------------------------------------- +# 校验 +# -------------------------------------------------------------------------- + + +def test_the_two_kinds_are_normalized(): + fields = normalize_case({"question": "上周销售额为什么下降？", "gold": SQL, "expectations": [ + {"kind": "contribution", "mode": "additive", "params": {"by": ["地区"], "period": "周", "measure": "销售额", + "label": "", "base": None}}, + {"kind": "verdict", "op": "comparable", "params": { + "period": "周", "base": "{{ schedule.last_week_start }}", "target": "{{ schedule.this_week_start }}"}}]}) + contribution, verdict = fields["expectations"] + assert contribution == {"id": "e1", "kind": "contribution", "required": True, "op": "decompose", "mode": "additive", + "params": {"by": ["地区"], "period": "周", "measure": "销售额"}, "top_n": 1} + assert verdict["op"] == "comparable" and "mode" not in verdict and verdict["params"]["base"].startswith("{{") + assert cases.case_hash(fields) == cases.case_hash(dict(fields)), "指纹照常按内容算" + + +@pytest.mark.parametrize("expectation, field, words", [ + ({"kind": "contribution", "op": "baseline", "params": {}}, "expectations.0.op", "只能是变化拆解"), + ({"kind": "contribution", "params": {"by": ["地区"]}}, "expectations.0.mode", "要选一种模式"), + ({"kind": "verdict", "params": {"period": "周", "measure": "订单数"}}, "expectations.0.op", "波动基线、趋势判断"), + ({"kind": "verdict", "op": "decompose", "mode": "additive"}, "expectations.0.op", "当前为「decompose」"), + ({"kind": "verdict", "op": "trend", "mode": "additive", "params": {"period": "周", "measure": "m"}}, + "expectations.0.mode", "没有模式可选"), + ({"kind": "verdict", "op": "differ", "params": {}}, "expectations.0.mode", "差异检验（比率）"), + ({"kind": "verdict", "op": "baseline", "params": {"period": "周"}}, "expectations.0.params.measure", "没写"), + ({**SOURCE, "params": {**SOURCE["params"], "main_share": 0.3}}, "expectations.0.params.main_share", + "阈值不写在期望里"), + ({**SOURCE, "params": {**SOURCE["params"], "query": "Q1"}}, "expectations.0.params.query", "没有「query」"), + ({**SOURCE, "params": {**SOURCE["params"], "headline": {"base": "Q1.r0.x", "target": "Q1.r1.x"}}}, + "expectations.0.params.headline", "没有「headline」"), + ({**SOURCE, "params": {**SOURCE["params"], "base": "{{ schedule.nope }}"}}, "expectations.0.params.base", + "不是可用的变量"), + ({**SOURCE, "top_n": 0}, "expectations.0.top_n", "1 到 10"), + ({**SOURCE, "params": "地区"}, "expectations.0.params", "需要是一个对象"), +]) +def test_wrong_analysis_expectations_are_refused(expectation, field, words): + e = invalid(expectation) + assert e.field == field and words in e.message, (e.field, e.message) + + +@pytest.mark.parametrize("key", ["segment", "subjects", "direction", "verdict", "judgement", "driver", "value"]) +def test_conclusions_cannot_be_filled_in_by_hand(key): + e = invalid({**SOURCE, key: "华东"}) + assert e.field == f"expectations.0.{key}" and "不许人填" in e.message and "算出来" in e.message + + +def test_a_caliber_gold_cannot_carry_an_analysis_expectation(): + caliber = {"kind": "caliber", "workflow_id": "w", "version": 1, "node_id": "c", "inputs": {}} + assert invalid(BAND, **caliber).field == "expectations.0.kind" + # 期望反问的题同样不能带分析期望：问法含糊就不该下判断 + with pytest.raises(CaseInvalid): + normalize_case({"question": "最近怎么样？", "gold": SQL, "expectations": [{"kind": "ask"}, BAND]}) + + +def test_the_request_renders_variables_and_fills_two_periods(): + snapshot = {"columns": ["地区", "周", "销售额"], "sql": SQL["sql"], + "rows": [["华东", "2026-10-12", 900], ["华东", "2026-10-05", 1000], ["华南", "2026-10-05", 800]]} + values = {"last_week_start": "2026-10-12", "this_week_start": "2026-10-19"} + request = judgements.request_of(SOURCE, values, snapshot) + assert request["op"] == "decompose" and request["mode"] == "additive" + assert request["params"] == {"query": judgements.GOLD_REF, "by": ["地区"], "period": "周", + "base": "2026-10-05", "target": "2026-10-12", "measure": "销售额"} + explicit = {**SOURCE, "params": {**SOURCE["params"], "target": "{{ schedule.last_week_start }}", + "base": "2026-10-05"}} + assert judgements.request_of(explicit, values, snapshot)["params"]["target"] == "2026-10-12" + three = {**snapshot, "rows": [*snapshot["rows"], ["华南", "2026-09-28", 700]]} + with pytest.raises(judgements.ParamsInvalid) as caught: + judgements.request_of(SOURCE, values, three) + assert "有 3 期" in caught.value.message and "只能返回两期" in caught.value.message + + +# -------------------------------------------------------------------------- +# 期望值、比较用的那一份 +# -------------------------------------------------------------------------- + + +def _gold(*pairs: tuple[dict, dict]) -> dict: + values = {exp["id"]: judgements.expected(exp, result) for exp, result in pairs} + sets = [{"run_id": "g", "snapshot": "x", "version": None, "values": values}] + return {"status": "ok", "reason": None, "code": None, "before": {"status": "ok", "sets": sets}, + "after": {"status": "ok", "sets": sets}} + + +def _case(*expectations: dict) -> dict: + return {"kind": "ask", "question": "虚构问题", "expectations": list(expectations), "gold": SQL} + + +def test_expected_values_follow_the_judgement_classes(): + driver = judgements.expected(SOURCE, fake_result()) + assert (driver["judgement"], driver["subjects"], driver["direction"]) == ("source", ["华东"], "down") + assert driver["phrase"].startswith("销售额下降") and driver["result"]["op"] == "decompose" + two = judgements.expected({**SOURCE, "top_n": 2}, fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]])) + assert two["judgement"] == "source" and two["subjects"] == ["华东", "华北"] + assert judgements.expected(SOURCE, fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]]) + )["subjects"] == ["华东"], "top_n 缺省 1：只要点名第一段" + for code, judgement in (("within_band", "normal"), ("outside_band", "anomaly"), ("far_outside_band", "anomaly")): + assert judgements.expected(BAND, fake_result("baseline", None, code))["judgement"] == judgement + for code, judgement in (("comparable", "comparable"), ("conditional", "comparable"), + ("incomparable", "incomparable")): + exp = {**BAND, "op": "comparable"} + assert judgements.expected(exp, fake_result("comparable", None, code))["judgement"] == judgement + differ = judgements.expected({**BAND, "op": "differ", "mode": "rate"}, + fake_result("differ", "rate", "no_evidence", subjects=[["西南"], ["华东"]], + direction=None)) + assert differ["judgement"] == "no_evidence" and differ["subjects"] == ["西南", "华东"] + refused = judgements.expected(BAND, fake_result("baseline", None, None, status="refused", + reason_code="history_short")) + assert refused["status"] == "refused" and refused["reason"] == "历史只有 5 期，至少要 8 期" + assert set(judgements.JUDGEMENTS) >= {v for m in judgements._JUDGEMENT_OF.values() for v in m.values()} + + +def test_comparing_two_golds_ignores_the_day_the_probe_time_and_the_snapshot_artifact(): + from app.eval import gold as gold_rules + + def outcome(day: str, at: str, artifact: str) -> gold_rules.GoldOutcome: + result = fake_result("baseline", None, "within_band") + result["params"]["today"] = day + result["probes"] = [{"source": "shop", "table": "orders", "column": "order_date", "min": "2026-04-01", + "max": "2026-10-20", "basis": "live", "at": at}] + result["inputs"] = [{"ref": judgements.GOLD_REF, "artifact": artifact, "role": "data", "columns": ["周"], + "rows": "all"}] + values = {"e1": judgements.expected(BAND, result), "e2": {"kind": "scalar", "value": 330, "column": "gmv"}} + return gold_rules.GoldOutcome(status="ok", as_of="", sets=[gold_rules.GoldSet("r", "s", values)]) + + before = outcome("2026-10-21", "2026-10-21T09:00:01", "a" * 64) + after = outcome("2026-10-22", "2026-10-22T00:00:03", "b" * 64) + assert gold_rules.settle(before, after).status == "ok" + from app.eval.runner import gold_key + + assert gold_key({"status": "ok", "before": before.to_dict()}) == gold_key({"status": "ok", "before": after.to_dict()}) + changed = outcome("2026-10-21", "x", "a" * 64) + changed.sets[0].values["e1"]["result"]["probes"][0]["max"] = "2026-10-19" + assert gold_rules.settle(before, changed).status == "drifted", "表的范围变了才算数据变了" + # 别的期望的指纹和以前一样（老结果和新结果比「数据变了」不受影响） + from app.core.artifact_store import canonical_json + + plain = gold_rules.GoldOutcome(status="ok", as_of="", sets=[gold_rules.GoldSet("r", "s", {"e2": {"value": 1}})]) + assert gold_rules._fingerprint(plain) == canonical_json([[None, {"e2": {"value": 1}}]]) + + +# -------------------------------------------------------------------------- +# 评分 +# -------------------------------------------------------------------------- + + +def _record(result: dict, alias: str | None = "A1", cited: tuple[str, ...] = ("phrase",)) -> dict: + record = judgements.record_of(result, artifact="c" * 64) + record.update(alias=alias, cited=list(cited)) + return record + + +def _answer(*records: dict, text: str = "", cited_in: tuple[str, ...] = ()) -> Answer: + sentences = judgements.sentences_of_text(text) + for s in sentences: + s["analyses"] = list(cited_in) + return Answer(run_id="r", text=text, analyses=list(records), sentences=sentences) + + +def _score(exp: dict, result: dict, answer: Answer) -> dict: + graded = score_case(_case(exp), _gold((exp, result)), answer, path="ran") + [item] = graded.scores + return item + + +def test_a_cited_conclusion_that_agrees_is_traced(): + item = _score(SOURCE, fake_result(), _answer(_record(fake_result()))) + assert item["outcome"] == "traced" and item["found"]["alias"] == "A1", item + assert item["want"] == {"judgement": "source", "subjects": ["华东"], "direction": "down", "status": "ok", + "phrase": fake_result()["phrase"], "method": "变化拆解（可加）"} + # 主要来源点名了期望的那一段就行：答案的拆解判成集中在华东、华北也对得上 + wider = fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]]) + assert _score(SOURCE, fake_result(), _answer(_record(wider)))["outcome"] == "traced" + + +def test_the_text_right_without_a_citation_is_untraced(): + item = _score(SOURCE, fake_result(), _answer(text="上周销售额下降主要来自华东。")) + assert item["outcome"] == "untraced" and item["found"]["text"] == "上周销售额下降主要来自华东", item + normal = _score(BAND, fake_result("baseline", None, "within_band"), + _answer(text="订单数的下降在常规波动范围内，不算异常。")) + assert normal["outcome"] == "untraced", normal + rising = _score({**BAND, "op": "trend"}, fake_result("trend", None, "rising", direction="up"), + _answer(text="线上销售额持续上升。")) + assert rising["outcome"] == "untraced", rising + + +@pytest.mark.parametrize("text, words", [ + ("订单数的下降属于异常。", "判定是"), # 常规波动说成异常 + ("上周销售额下降主要来自华南。", "不是华南"), # 主要来源说成别的段 + ("上周销售额上升主要来自华东。", "涨跌写反了"), # 方向反了 + ("线上销售额呈下降趋势。", "相反"), # 趋势方向反了 +]) +def test_saying_the_opposite_is_wrong(text, words): + exp, result = { + "订单数": (BAND, fake_result("baseline", None, "within_band")), + "销售额": (SOURCE, fake_result()), + "线上": ({**BAND, "op": "trend"}, fake_result("trend", None, "rising", direction="up")), + }[next(k for k in ("订单数", "线上", "销售额") if k in text)] + item = _score(exp, result, _answer(text=text)) + assert item["outcome"] == "wrong" and words in item["reason"], item + + +def test_a_cited_conclusion_that_disagrees_is_wrong(): + anomaly = fake_result("baseline", None, "outside_band", direction="down") + item = _score(BAND, fake_result("baseline", None, "within_band"), _answer(_record(anomaly))) + assert item["outcome"] == "wrong" and "A1" in item["reason"] and "不是一类" in item["reason"], item + assert item["found"]["judgement"] == "anomaly" + # 差异检验比的是别的两组：对不上 + other = fake_result("differ", "rate", "no_evidence", subjects=[["华北"], ["华东"]], direction=None) + want = fake_result("differ", "rate", "no_evidence", subjects=[["西南"], ["华东"]], direction=None) + item = _score({**BAND, "op": "differ", "mode": "rate"}, want, _answer(_record(other))) + assert item["outcome"] == "wrong" and "期望比的是西南和华东" in item["reason"], item + + +def test_citing_the_right_conclusion_but_writing_the_opposite_is_wrong(): + normal = fake_result("baseline", None, "within_band") + item = _score(BAND, normal, _answer(_record(normal, cited=("see",)), text="订单数下降属于异常。", + cited_in=("A1",))) + assert item["outcome"] == "wrong" and "引用了 A1" in item["reason"], item + + +def test_doing_nothing_and_saying_nothing_is_wrong(): + item = _score(BAND, fake_result("baseline", None, "within_band"), _answer(text="上周订单数是 1,203 单。")) + assert item["outcome"] == "wrong" and "没有做波动基线" in item["reason"], item + uncited = _score(BAND, fake_result("baseline", None, "within_band"), + _answer(_record(fake_result("baseline", None, "within_band"), cited=()), text="上周订单数是 1,203 单。")) + assert uncited["outcome"] == "wrong" and "但没有引用" in uncited["reason"], uncited + # 可比性没有对应的判断词：只能引用 + exp = {**BAND, "op": "comparable"} + words = _score(exp, fake_result("comparable", None, "incomparable"), _answer(text="本周还没过完，两周不能直接比。")) + assert words["outcome"] == "wrong" + cited = _score(exp, fake_result("comparable", None, "incomparable"), + _answer(_record(fake_result("comparable", None, "incomparable")))) + assert cited["outcome"] == "traced" + + +def test_an_analysis_that_cannot_be_computed_cannot_be_graded(): + refused = fake_result("baseline", None, None, status="refused", reason_code="history_short") + graded = score_case(_case(BAND, {"id": "e2", "kind": "scalar", "column": "gmv", "required": True}), + {**_gold((BAND, refused)), "before": {"status": "ok", "sets": [{"values": { + "e1": judgements.expected(BAND, refused), + "e2": {"kind": "scalar", "value": 330.0, "column": "gmv", "row": 0, "dims": {}}}}]}}, + Answer(run_id="r", values=[{"value": 330.0, "kind": "cell", "artifact": "a", "column": "gmv", + "dims": {}}]), path="ran") + band, sales = graded.scores + assert band["outcome"] == "ungradable" and band["reason"] == "期望算不了：历史只有 5 期，至少要 8 期" + assert sales["outcome"] == "traced", "同一题别的期望照常评" + assert graded.verdict == "ungradable" + # 答案里同类的分析做成了算不了：不算答对 + item = _score(BAND, fake_result("baseline", None, "within_band"), _answer(_record(refused))) + assert item["outcome"] == "wrong" and "没有做成" in item["reason"], item + + +def test_top_n_wants_every_named_source(): + want = fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]]) + exp = {**SOURCE, "top_n": 2} + only_one = _score(exp, want, _answer(text="上周销售额下降集中在华东。")) + assert only_one["outcome"] == "wrong" and "还有华北" in only_one["reason"], only_one + both = _score(exp, want, _answer(text="上周销售额下降集中在华东、华北。")) + assert both["outcome"] == "untraced", both + + +def test_old_answers_without_analyses_read_as_empty(): + old = Answer.from_dict({"run_id": "r", "text": "上周销售额 330。", "values": [], "bare": []}) + assert old.analyses == [] and old.sentences == [] + item = _score(SOURCE, fake_result(), old) + assert item["outcome"] == "wrong", item + again = Answer.from_dict(_answer(_record(fake_result())).to_dict()) + assert again.analyses[0]["alias"] == "A1" + + +def test_the_summary_counts_outcomes_by_kind(): + rows = [{"status": "ok", "verdict": "pass", "case_id": "c1", "scores": [ + {"kind": "contribution", "required": True, "outcome": "traced"}, + {"kind": "scalar", "required": False, "outcome": "wrong"}]}, + {"status": "ok", "verdict": "fail", "case_id": "c2", "scores": [ + {"kind": "verdict", "required": True, "outcome": "wrong"}, + {"kind": "scalar", "required": True, "outcome": "untraced"}, + {"kind": "text", "required": False, "outcome": "covered"}]}] + summary = stats.summarize(rows) + assert summary["outcomes_by_kind"] == {"contribution": {"traced": 1}, "verdict": {"wrong": 1}, + "scalar": {"untraced": 1}} + assert summary["outcomes"] == {"traced": 1, "wrong": 1, "untraced": 1}