diff --git a/backend/app/engine/nodes/analysis.py b/backend/app/engine/nodes/analysis.py index f6a2a584..382104a7 100644 --- a/backend/app/engine/nodes/analysis.py +++ b/backend/app/engine/nodes/analysis.py @@ -63,7 +63,7 @@ async def run_analysis(state: GraphState, ctx: NodeContext) -> dict[str, Any]: inputs[node_id] = await _load_input(state, ctx, node_id, queries) context = await context_of(list(inputs.values()), tool_ctx=tool_context(ctx.run, ctx.node.id, ctx.emit)) - import app.analysis as analysis_ops # 按模块取 run:测试、合并前的替身都只换模块上的这一个函数 + import app.analysis as analysis_ops # 按模块取 run:测试换替身时只换模块上的这一个函数 exec_no = next_exec(state, ctx.node.id) done: dict[str, AnalysisResult] = {} @@ -158,9 +158,8 @@ async def _load_input(state: GraphState, ctx: NodeContext, node_id: str, raise NodeError(ctx.node.id, f"{who}的查询快照与哈希不一致,疑似被修改,不能分析") from None if not isinstance(content, dict): raise NodeError(ctx.node.id, f"{who}的查询快照已不存在,无法分析。请重新运行它") - snapshot = input_snapshot(content, ref=node_id, artifact=artifact, node_id=node_id) - snapshot["masked"] = await masked_of(content, artifact) - return snapshot + return input_snapshot(content, ref=node_id, artifact=artifact, node_id=node_id, + masked=await masked_of(content, artifact)) __all__ = ["REFUSED_CODE", "card_view", "run_analysis"] diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index 7e90092e..77aea26e 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -539,7 +539,7 @@ class _Skipped(Exception): async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]], ctx: NodeContext, comparing: _Comparing, snapshots: dict[str, Any]) -> dict[str, Any]: - import app.analysis as analysis_ops # 按模块取 run:测试、合并前的替身都只换模块上的这一个函数 + import app.analysis as analysis_ops # 按模块取 run:测试换替身时只换模块上的这一个函数 from app.analysis.contract import VERDICTS from app.data.sql_semantics import normalize_sql @@ -573,10 +573,9 @@ async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]], refs[artifact] = producer if producer and producer not in refs.values() else f"{producer or 'query'}·{role}" snaps = {} for artifact in order: - content = snapshots[artifact] - snap = input_snapshot(content, ref=refs[artifact], artifact=artifact, node_id=refs[artifact].split("·")[0]) - snap["masked"] = await masked_of(content, artifact) - snaps[refs[artifact]] = snap + snaps[refs[artifact]] = input_snapshot(snapshots[artifact], ref=refs[artifact], artifact=artifact, + node_id=refs[artifact].split("·")[0], + masked=await masked_of(snapshots[artifact], artifact)) settings = await settings_of(run_snapshot(ctx.run)) context = await context_of(list(snaps.values()), tool_ctx=tool_context(ctx.run, ctx.node.id, ctx.emit)) request = {"op": "comparable", "mode": None, @@ -586,8 +585,9 @@ async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]], if result.get("status") != "ok" or verdict not in VERDICTS["comparable"]: raise _Skipped(str(result.get("reason") or "检查没有给出判定")) artifact = await store(result, run_id=ctx.run.run_id, node_id=ctx.node.id) - return {"artifact": artifact, "verdict": verdict, "phrase": str(result.get("phrase") or ""), - "reasons": comparability_reasons(result)} + # 原因的拆法只有算子那一处(analysis.comparable_reasons,按检查表的说明拆) + return {"artifact": artifact, "verdict": result["verdict"]["code"], "phrase": result["phrase"], + "reasons": analysis_ops.comparable_reasons(result)} async def _dialect_of(source: str, comparing: _Comparing) -> str | None: @@ -625,24 +625,6 @@ def _period_order(a: tuple[str, ...], b: tuple[str, ...]) -> tuple[int, int]: return 1, 0 -def comparability_reasons(result: dict[str, Any]) -> list[str]: - """可比性结论里 fail、warn 的原因,一句一个(contract.MetricComparability.reasons)。结果表里有 status 一列和一列说明 - (reason / text / detail / message)的照表取;没有的按判定语「两期不可比:原因;原因」拆。可比的没有原因。""" - table = result.get("table") or {} - columns = [str(c) for c in table.get("columns") or []] - text_col = next((c for c in ("reason", "text", "detail", "message") if c in columns), None) - if "status" in columns and text_col is not None: - si, ti = columns.index("status"), columns.index(text_col) - found = [str(row[ti]) for row in table.get("rows") or [] - if isinstance(row, list) and len(row) > max(si, ti) and row[si] in ("fail", "warn") and row[ti]] - if found: - return found - phrase = str(result.get("phrase") or "") - if (result.get("verdict") or {}).get("code") == "comparable" or ":" not in phrase: - return [] - return [part.strip() for part in phrase.split(":", 1)[1].split(";") if part.strip()] - - #: 清单里输入取自「查询区间里这张表没有数据」的列的指标后面那句:直接读清单的模型节点不能把 0 当成真的是 0 LISTED_NO_DATA = "(这段时间表里没有数据:不要当成 0 解读)" #: 指标的 no_data_reason 打头那半句(后面接查询记下的说明) diff --git a/backend/app/tools/analysis.py b/backend/app/tools/analysis.py index a969a0e3..14566090 100644 --- a/backend/app/tools/analysis.py +++ b/backend/app/tools/analysis.py @@ -10,8 +10,9 @@ 那一步按 checkpoint 里的结果照样放一遍,编号还是同一套。 - **阈值**:发起时的快照(run.options["analysis"],engine/launch.py)← 分析卡上写的覆盖;没有快照的现读设置。agent 改不了: 分析工具的参数里没有阈值(契约 D4)。 -- **运行期依赖**(contract.Context):运行当天(服务端时区)、现查表的日期范围(W1-a 的 make_prober,没有这个模块时不给)、 - 数据目录(按运行的目录视图读,和 db_schema 给模型看的同一份)、剖析事实。 +- **运行期依赖**(contract.Context):运行当天(服务端时区)、现查表的日期范围(analysis.make_prober,按这次运行固定的数据 + 版本解析数据源)、数据目录(按运行的目录视图读,和 db_schema 给模型看的同一份)、剖析事实(analysis.probe.profile_facts, + 和查询工具的覆盖检查同一个取法)。 - **遮罩列**:数据源设的遮罩列、查询当时记下的、按 SQL 认出改了名字的(和证据面板、运行对比同一个函数,api/evidence 的 _mask_sets 的 hidden,宁可多遮),加上遮罩运行里查询时就整列遮掉的。算子不拿它们当维度、度量。 - **结论**:整份落成内容寻址的工件(contract.ARTIFACT_KIND),台账记 contract.ledger_entry;交回 agent 的是一段短的 @@ -64,9 +65,10 @@ def call_error_text(error: CallError) -> str: class AnalysisScope: """一次节点执行里分析工具读的容器(agent 节点、协作成员的一次执行各一个)。 - queries:编号 Qn → 输入快照(contract.InputSnapshot,masked 留空,用到时由 prepare 补)。results:编号 An → 整份 - 结论;artifacts:An → 结论的工件 id。snapshot:发起时的分析设置快照(没有为 None,现读设置)。mask_cache:快照工件 → - 遮罩列;source_cache:数据源和表 → 目录、剖析事实(一次节点执行里各读一次,只在工具执行时用,不影响编号)。""" + queries:编号 Qn → 输入快照(contract.InputSnapshot;记台账时先放交回模型的那份 payload,分析前 prepare 换成工件库里 + 那份完整快照,补上遮罩列)。results:编号 An → 整份结论;artifacts:An → 结论的工件 id。snapshot:发起时的分析设置快照 + (没有为 None,现读设置)。mask_cache:快照工件 → 遮罩列;source_cache:数据源和表 → 目录、查询用的源;probe_cache: + 数据源名 → coverage.Cache(一次节点执行里各读、各查一次,只在工具执行时用,不影响编号)。""" node_id: str snapshot: dict[str, Any] | None = None @@ -75,6 +77,9 @@ class AnalysisScope: artifacts: dict[str, str] = field(default_factory=dict) mask_cache: dict[str, list[str]] = field(default_factory=dict) source_cache: dict[str, Any] = field(default_factory=dict) + probe_cache: dict[str, Any] = field(default_factory=dict) + #: 已经换成工件库里那份完整快照的编号(prepare 每份只取回一次) + prepared: set[str] = field(default_factory=set) def add_query(self, alias: str, snapshot: InputSnapshot) -> None: self.queries[alias] = snapshot @@ -130,25 +135,18 @@ async def settings_of(snapshot: Mapping[str, Any] | None, # -------------------------------------------------------------------------- -def input_snapshot(content: Mapping[str, Any], *, ref: str, artifact: str, node_id: str) -> InputSnapshot: - """一份查询快照(数据源查询工具交回的、工件库里的)→ 算子的一份输入(masked 留空,prepare 时补)。列的类型对齐成和 - columns 同序的列表(快照里记的是 {列: 类型},没记的为 None);行一律写成和 columns 同序的列表。""" - columns = [str(c) for c in content.get("columns") or []] - types = content.get("column_types") - rows: list[list[Any]] = [] - for row in content.get("rows") or []: - if isinstance(row, (list, tuple)): - rows.append(list(row)) - elif isinstance(row, Mapping): - rows.append([row.get(c) for c in columns]) - else: - rows.append([row]) - return { - "ref": ref, "artifact": artifact, "node_id": node_id, "source": str(content.get("source") or ""), - "sql": str(content.get("sql") or ""), "columns": columns, - "column_types": [types.get(c) if isinstance(types, Mapping) else None for c in columns], - "rows": rows, "truncated": bool(content.get("truncated")), "masked": [], - } +def input_snapshot(content: Mapping[str, Any], *, ref: str, artifact: str, node_id: str, + masked: Iterable[str] = ()) -> InputSnapshot: + """一份查询快照 → 算子的一份输入:查询工具的 payload(工件库里那份)原样——列、行、类型、SQL、数据源,连同 SQL 检查结果 + (checks)、「查询区间里这张表没有数据」(no_data)、查询当时记下的遮罩(mask_columns、masked_at_query),可比性检查按 + 它们判数据问题——再加上出处(ref、artifact、node_id)和不能当维度、度量的列(masked,见 masked_of)。合并查询在 SQLite + 里执行,方言写明;数据源的方言算子按探查器认(prober.dialect)。""" + from app.engine.merge_query import MERGE_SOURCE + + out: dict[str, Any] = {**content, "ref": ref, "artifact": artifact, "node_id": node_id, "masked": list(masked)} + if content.get("source") == MERGE_SOURCE and not content.get("dialect"): + out["dialect"] = "sqlite" + return out # type: ignore[return-value] async def masked_of(content: Mapping[str, Any] | None, artifact: str | None) -> list[str]: @@ -171,23 +169,28 @@ async def masked_of(content: Mapping[str, Any] | None, artifact: str | None) -> async def prepare(scope: AnalysisScope) -> None: - """补上容器里每份输入快照的遮罩列(每份工件只认一次)。快照取不回来、和哈希对不上的照实报错:查询明明交回过, - 分析读不到它的快照,不是 agent 写错了参数。""" + """分析前把容器里的每份输入换成工件库里那份完整的查询快照(取回时复验哈希),补上遮罩列(每份工件只认一次)。快照 + 取不回来、和哈希对不上的照实报错:查询明明交回过,分析读不到它的快照,不是 agent 写错了参数。""" import asyncio from app.core import artifact_store - for alias, snapshot in scope.queries.items(): - artifact = snapshot.get("artifact") or "" + for alias, snapshot in list(scope.queries.items()): + if alias in scope.prepared: + continue + artifact = str(snapshot.get("artifact") or "") + try: + content = await asyncio.to_thread(artifact_store.load, artifact) + except ValueError as e: + raise RuntimeError(f"{alias} 的查询快照与哈希不一致,疑似被修改,不能分析") from e + if not isinstance(content, Mapping): + raise RuntimeError(f"{alias} 的查询快照已不存在,不能分析") if artifact not in scope.mask_cache: - try: - content = await asyncio.to_thread(artifact_store.load, artifact) - except ValueError as e: - raise RuntimeError(f"{alias} 的查询快照与哈希不一致,疑似被修改,不能分析") from e - if content is None: - raise RuntimeError(f"{alias} 的查询快照已不存在,不能分析") scope.mask_cache[artifact] = await masked_of(content, artifact) - snapshot["masked"] = list(scope.mask_cache[artifact]) + scope.queries[alias] = input_snapshot(content, ref=alias, artifact=artifact, + node_id=str(snapshot.get("node_id") or scope.node_id), + masked=scope.mask_cache[artifact]) + scope.prepared.add(alias) # -------------------------------------------------------------------------- @@ -214,21 +217,26 @@ def _tables_of(snapshots: Iterable[Mapping[str, Any]]) -> dict[str, set[str]]: async def context_of(snapshots: Iterable[Mapping[str, Any]], *, tool_ctx: Any, - cache: dict[str, Any] | None = None) -> contract.Context: - """算子要的运行期依赖(contract.Context)。 + cache: dict[str, Any] | None = None, + probe_cache: dict[str, Any] | None = None) -> contract.Context: + """算子要的运行期依赖(contract.Context)。每次分析调用一份(探查器有总时限,一次调用建一个)。 - today:运行当天,服务端时区(算子据此判本期完不完整) - catalogs:数据源名 → {表名: 这张表的目录(app/data/catalog 的 notes)},按运行的目录视图读(tool_ctx.catalog_view, 和 db_schema 给模型看的、冻结进表结构快照的同一份),只给快照的 SQL 里用到的表 - - facts:数据源名 → {表名: 剖析事实(catalog_profile_jobs.last_facts)};上传表格的剖析不是这次查的那一版的不给 - - probe:现查一列的实际范围(W1-a 的 make_prober,按这次运行固定的数据版本解析数据源;没有这个模块时为 None) + - facts:数据源名 → 取剖析事实的函数(analysis.probe.profile_facts:上传表格的剖析不是这次查的那一版的不给) + - probe:analysis.make_prober(数据源名 → 查询用的源):按这次运行固定的数据版本解析数据源(上传表格查的是固定的那一版), + 遮罩列按数据源设的不探,探不成由算子退回剖析事实;一个源都解析不出来时为 None + + tool_ctx 是查询用的那个 ToolContext(固定的数据版本、目录视图都从它取)。cache 是调用方的缓存(同一个节点里分析好几次, + 目录、数据源读一次);probe_cache 是数据源名 → coverage.Cache(同一张表的范围一个节点里只查一次)。读不到的源当没有: + 目录、剖析、探查都是附加的,不能挡住分析。""" + from app.analysis.probe import profile_facts - tool_ctx 是查询用的那个 ToolContext(固定的数据版本、目录视图都从它取)。cache 是调用方的缓存(同一个节点里分析 - 好几次,目录、剖析事实读一次)。读不到的源当没有:目录、剖析是附加说明,不能挡住分析。""" cache = cache if cache is not None else {} catalogs: dict[str, Any] = {} facts: dict[str, Any] = {} - found: dict[str, Any] = {} + sources: dict[str, Any] = {} for name, tables in _tables_of(snapshots).items(): key = f"{name}\n" + "\n".join(sorted(tables)) if key not in cache: @@ -236,17 +244,16 @@ async def context_of(snapshots: Iterable[Mapping[str, Any]], *, tool_ctx: Any, entry = cache[key] if entry is None: continue - found[name] = entry["source"] - catalogs[name], facts[name] = entry["catalog"], entry["facts"] - return {"today": date.today(), "probe": _prober(found, facts), "catalogs": catalogs, "facts": facts} + sources[name] = entry["source"] + catalogs[name] = entry["catalog"] + facts[name] = profile_facts(entry["source"]) + return {"today": date.today(), "probe": _prober(sources, probe_cache), "catalogs": catalogs, "facts": facts} async def _source_context(name: str, tables: set[str], tool_ctx: Any) -> dict[str, Any] | None: - """一个数据源这几张表的目录、剖析事实,和查询用的那个源(探查用)。读不到为 None。""" + """一个数据源这几张表的目录,和查询用的那个源(探查、剖析事实都按它)。读不到为 None。""" from app.data import catalog as data_catalog from app.data import catalog_view - from app.data.catalog_profile_jobs import last_facts - from app.db.base import SessionLocal from app.tools.datasource import resolved_source try: @@ -255,61 +262,24 @@ async def _source_context(name: str, tables: set[str], tool_ctx: Any) -> dict[st return None row, source = resolved entries = await catalog_view.read(catalog_view.CatalogView.parse(getattr(tool_ctx, "catalog_view", None)), row) - names = {data_catalog.resolve_table_name(getattr(source, "schema_cache", None), t) or t for t in tables} - async with SessionLocal() as session: - profiled = await last_facts(session, str(row.id), sorted(names)) if names else {} - except Exception: # noqa: BLE001 - 目录、剖析读不到当没有(目录视图指定的快照不在了另由查询工具照实报) - logger.warning("分析读取数据源 %s 的目录、剖析事实失败,这次不附", name, exc_info=True) + except Exception: # noqa: BLE001 - 目录读不到当没有(目录视图指定的快照不在了另由查询工具照实报) + logger.warning("分析读取数据源 %s 的目录失败,这次不附", name, exc_info=True) return None - snapshot = getattr(source, "snapshot_id", None) - return { - "source": source, - "catalog": {t: dict(entries[t].notes or {}) for t in sorted(names) if t in entries}, - # 上传表格:剖析时是另一版数据,范围对不上这次读的这一版(同 tools/datasource 的 _coverage) - "facts": {t: f for t, f in profiled.items() if not (snapshot and f.get("snapshot_id") != snapshot)}, - } + names = {data_catalog.resolve_table_name(getattr(source, "schema_cache", None), t) or t for t in tables} + return {"source": source, "catalog": {t: dict(entries[t].notes or {}) for t in sorted(names) if t in entries}} -def _probe_factory() -> Any: - """W1-a 的 app.analysis.probe.make_prober;还没有这个模块时为 None(算子只用剖析事实,核对不了的写「核对不了」)。""" - try: - from app.analysis.probe import make_prober # type: ignore[import-not-found] - except ModuleNotFoundError as e: - if e.name not in ("app.analysis.probe", "app.analysis"): - raise - return None - return make_prober - - -def _prober(sources: Mapping[str, Any], facts: Mapping[str, Any]) -> contract.Prober | None: - """按数据源名分派的探查(contract.Prober)。每个源第一次被问到时才建它自己的探查: - - make_prober(查询用的源, masked=遮罩列, cache=coverage.Cache(), facts=按表取剖析事实的协程函数) - - 签名以 W1-a 的交付为准(合并时由编排者对齐这一处)。遮罩列上的不探(data/coverage 的规矩),探查失败由它自己退回 - 剖析事实。""" - make = _probe_factory() - if make is None or not sources: +def _prober(sources: Mapping[str, Any], probe_cache: dict[str, Any] | None) -> contract.Prober | None: + """这次分析调用的探查器(analysis.make_prober)。遮罩列、剖析事实按缺省(数据源设的遮罩列;profile_facts 现读); + coverage.Cache 按数据源名放在调用方给的 probe_cache 里,同一个节点里的几次分析共用。""" + if not sources: return None + from app import analysis from app.data import coverage - from app.data.engine import masked_columns - built: dict[str, Any] = {} - - async def probe(source: str, table: str, column: str) -> contract.Probe | None: - if source not in built: - obj = sources.get(source) - known = (facts.get(source) or {}) if isinstance(facts, Mapping) else {} - - async def by_table(name: str, known: Mapping[str, Any] = known) -> Mapping[str, Any] | None: - return known.get(name) - - built[source] = None if obj is None else make( - obj, masked=masked_columns(getattr(obj, "options", None)), cache=coverage.Cache(), facts=by_table) - found = built[source] - return await found(source, table, column) if found is not None else None - - return probe + shared = probe_cache if probe_cache is not None else {} + return analysis.make_prober(dict(sources), cache={name: shared.setdefault(name, coverage.Cache()) + for name in sources}) def tool_context(run_ctx: Any, node_id: str, emit: Any = None) -> Any: diff --git a/backend/app/tools/builtin/analysis.py b/backend/app/tools/builtin/analysis.py index 830bbbea..0dfd9dfa 100644 --- a/backend/app/tools/builtin/analysis.py +++ b/backend/app/tools/builtin/analysis.py @@ -42,9 +42,6 @@ NO_SCOPE = ("分析工具只能由 Agent 在本节点查过数之后调用:它 "模板里要固定做的分析,请改用「分析卡」节点") #: 交回 agent 的说法:本节点还没有成功的查询 NO_QUERIES = "本节点还没有成功的查询可以分析:先用数据库查询工具查出要分析的数,再按交回的编号(Q1、Q2……)调用分析工具" -#: 这个节点的分析次数用完 -MAX_CALLS = ("这个节点已经做了 {done} 次分析,达到「{label}」的上限({limit} 次),不能再分析了。" - "请基于已有的分析结论给出结论,没查清的照实说") _QUERY = "要分析的查询结果:本节点交回过的编号,如 Q3(和「【证据 Q3】」同一个)" _LABEL = "度量的叫法,系统写判定语时用(如「销售额」);不要写数字、百分号和单位,留空就用列名" @@ -118,20 +115,20 @@ class DifferArgs(BaseModel): description="rate:比率(转化率、退款率),给分子、分母两列;mean:均值,给每组的均值、标准差、样本数三列," "或合计、平方和、样本数三列;rows:逐行的值,算子自己算均值和方差(结果不能截断)") query: str = Field(description=_QUERY) - by: list[str] | None = Field(default=None, description="分组列的列名") - reference: str | None = Field(default=None, description="对照组:分组列里的一个值;留空和整体比") + by: list[str] | None = Field(default=None, description="按组比时:分组列的列名(一列)") pairs: list[list[str]] | None = Field(default=None, description="只比这几对:每对写两个组在分组列里的值") - period: str | None = Field(default=None, description="按期次比同一组的两期时:期次列") + reference: str | None = Field(default=None, description="对照组:分组列里的一个值,其余各组各和它比") + period: str | None = Field(default=None, description="按期次比同一组的两期时:期次列(SQL 里先筛好那一组)") base: str | None = Field(default=None, description="按期次比时:基期") target: str | None = Field(default=None, description="按期次比时:本期") numerator: str | None = Field(default=None, description="分子列:成功数(rate 用,如退款单数)") denominator: str | None = Field(default=None, description="分母列:试验数(rate 用,如订单数)") - mean: str | None = Field(default=None, description="每组的均值列(mean 用)") - std: str | None = Field(default=None, description="每组的标准差列(mean 用)") - sum: str | None = Field(default=None, description="每组的合计列(mean 用,没有标准差时和平方和一起给)") - sum_sq: str | None = Field(default=None, description="每组的平方和列(mean 用)") + mean: str | None = Field(default=None, description="每组的均值列(mean 用,和标准差、样本数一起给)") + sd: str | None = Field(default=None, description="每组的标准差列(mean 用)") + sum: str | None = Field(default=None, description="每组的合计列(mean 用,没有标准差时和平方和、样本数一起给)") + sumsq: str | None = Field(default=None, description="每组的平方和列(mean 用)") n: str | None = Field(default=None, description="每组的样本数列(mean 用)") - measure: str | None = Field(default=None, description="逐行的值列(rows 用)") + value: str | None = Field(default=None, description="逐行的值列(rows 用)") label: str | None = Field(default=None, description=_LABEL) hypothesis: str | None = Field(default=None, description=_HYPOTHESIS) @@ -143,7 +140,8 @@ class ComparableArgs(BaseModel): period: str | None = Field(default=None, description="同一份查询时:期次列") base: str | None = Field(default=None, description="同一份查询时:基期") target: str | None = Field(default=None, description="同一份查询时:本期") - label: str | None = Field(default=None, description=_LABEL) + measure: str | None = Field(default=None, description="度量列:按它找时间条件、认它是不是可累加的量;只有一列数值列时可以不写") + grain: Literal["day", "week", "month", "quarter"] | None = Field(default=None, description=_GRAIN) hypothesis: str | None = Field(default=None, description=_HYPOTHESIS) @@ -179,10 +177,10 @@ async def run_tool(op: str, ctx: ToolContext, args: BaseModel) -> dict[str, Any] """一次 agent 分析:读本节点的查询容器、按设置快照取阈值、调算子、结论落成工件。交回 {analysis_artifact, result}: 记台账、编号 An、给模型看的那一段由节点在记台账时做(engine/nodes/llm.py 的 _record_call),重放时照样做一遍。 - 调用写错(含这个节点的分析次数用完)抛 contract.CallError,节点作为工具错误交回模型、不记台账。""" - import app.analysis as analysis_ops # 按模块取 run:测试、合并前的替身都只换模块上的这一个函数 + 调用写错(含这个节点的分析次数用完,analysis.exhausted)抛 contract.CallError,节点作为工具错误交回模型、不记台账。 + 结论原样存成工件、一个字段都不加:下钻时算子按上一层结论的内容哈希认 parent 的工件。""" + import app.analysis as analysis_ops # 按模块取 run:测试换替身时只换模块上的这一个函数 - from app.analysis.contract import LABELS from app.tools.analysis import context_of, prepare, scope_of, settings_of, store scope = scope_of(ctx.extra) @@ -192,11 +190,11 @@ async def run_tool(op: str, ctx: ToolContext, args: BaseModel) -> dict[str, Any] raise CallError("no_queries", NO_QUERIES) settings = await settings_of(scope.snapshot) if len(scope.results) >= settings["max_calls"]: - raise CallError("max_calls", MAX_CALLS.format(done=len(scope.results), label=LABELS["max_calls"], - limit=settings["max_calls"])) + raise analysis_ops.exhausted(settings) # 次数用完的说法只有算子那一处 request = request_of(op, args) await prepare(scope) - context = await context_of(list(scope.queries.values()), tool_ctx=ctx, cache=scope.source_cache) + context = await context_of(list(scope.queries.values()), tool_ctx=ctx, cache=scope.source_cache, + probe_cache=scope.probe_cache) result = await analysis_ops.run(request, scope.resolve, settings=settings, context=context, parents=scope.parent) artifact = await store(result, run_id=ctx.run_id, node_id=scope.node_id) return {"analysis_artifact": artifact, "result": result} @@ -242,7 +240,7 @@ async def analysis_trend(ctx: ToolContext, **kwargs: Any) -> dict[str, Any]: name=contract.TOOL_NAMES["differ"], category=CATEGORY, description="差异检验:判断两组(或同一组的两期)的差异站得住,还是可能只是样本波动。比率给分子、分母两列(rate)," - "均值给均值、标准差、样本数(mean),也可以给逐行的值(rows);样本小时照实标出。", + "均值给均值、标准差、样本数或合计、平方和、样本数(mean),也可以给逐行的值(rows);样本小时照实标出。", args_schema=DifferArgs, guide=METHOD_GUIDE, ) diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index 871ef28d..85abed0e 100644 --- a/backend/app/tools/datasource.py +++ b/backend/app/tools/datasource.py @@ -804,29 +804,15 @@ def _unchecked_tables(sql: str, source: Any, outside: frozenset[str]) -> list[st async def _coverage(source: Any, sql: str, payload: Mapping[str, Any], mask: list[str]) -> list[dict[str, Any]]: """查询区间里这张表没有数据的那几列(data/coverage.result_gaps)。只在结果里有整列是 0 或空的数值列时才查库;遮罩列 - 不探查;现查表的范围不成时退回剖析、取值观察记下的范围(上传表格的快照对不上就不用)。从不抛异常。""" + 不探查;现查表的范围不成时退回剖析、取值观察记下的范围(上传表格的快照对不上就不用)。从不抛异常。 + + 剖析事实的取法只有一处(analysis.probe.profile_facts):分析的可比性检查现查不成时退回的也是它。""" + from app.analysis.probe import profile_facts from app.data import coverage if not coverage.zero_columns(payload): return [] - - async def facts(table: str) -> Mapping[str, Any] | None: - from app.data.catalog_profile_jobs import last_facts - from app.db.base import SessionLocal - - source_id = getattr(source, "id", None) - if not source_id: - return None - name = data_catalog.resolve_table_name(getattr(source, "schema_cache", None), table) or table - async with SessionLocal() as session: - found = (await last_facts(session, str(source_id), [name])).get(name) - snapshot = getattr(source, "snapshot_id", None) - if found and snapshot and found.get("snapshot_id") != snapshot: - # 上传表格:剖析时是另一版数据,范围对不上这次读的这一版 - return None - return found - - return await coverage.result_gaps(source, sql, payload, facts=facts, masked=mask) + return await coverage.result_gaps(source, sql, payload, facts=profile_facts(source), masked=mask) async def _sql_checks(checker: SqlChecker, sql: str) -> list[dict[str, Any]]: diff --git a/backend/tests/test_analysis_card.py b/backend/tests/test_analysis_card.py index 0aae7593..23773219 100644 --- a/backend/tests/test_analysis_card.py +++ b/backend/tests/test_analysis_card.py @@ -158,6 +158,9 @@ def test_a_well_formed_card_validates(): "输入里"), ([{**WOW, "params": {**WOW["params"], "bogus": 1}}], "items[0].params.bogus", "analysis.params_invalid", "没有「bogus」"), + # 运行当天、目录事实、下钻层数由系统填(算子自己记),分析项里写了不收 + ([{**WOW, "params": {**WOW["params"], "today": "2026-10-21"}}], "items[0].params.today", + "analysis.params_invalid", "没有「today」"), ([{**WHERE, "params": {**WHERE["params"], "headline": {"base": "weekly.gmv", "target": "weekly.r1.gmv"}}}], "items[0].params.headline.base", "analysis.params_invalid", "节点 id.r行号.列名"), ([{**WHERE, "params": {**WHERE["params"], "headline": "weekly.r0.gmv"}}], "items[0].params.headline", diff --git a/backend/tests/test_analysis_e2e.py b/backend/tests/test_analysis_e2e.py new file mode 100644 index 00000000..c7933b95 --- /dev/null +++ b/backend/tests/test_analysis_e2e.py @@ -0,0 +1,314 @@ +"""分析入口接真算子跑一遍(不用替身):agent 分析工具(含下钻)、分析卡(波动基线 + 变化拆解)、口径卡的可比性挂钩、 +审批恢复时的重放。断言工件、台账、交回模型的那一段、节点产出,并用 analysis.recompute 按封存的工件复算、逐字段相同。 + +虚构零售库 shop:orders(day, region, channel, amount),2026-03-02 起 12 个整周,每周每个地区、渠道两笔(周三、周日), +最后一周华东掉三成。模型换成脚本(MockChatModel._decide),不调用任何模型。 +""" +from __future__ import annotations + +import asyncio +import json +import sqlite3 +from datetime import date, timedelta + +import pytest +from langchain_core.messages import AIMessage, ToolMessage +from sqlalchemy import select + +import app.analysis as analysis_ops +from app.analysis import contract +from app.core import artifact_store +from app.core.artifact_store import canonical_json, content_hash +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import DataSource, Run, RunEvent +from app.engine.runner import run_manager + +TOOL = "db_query__shop" +DECOMPOSE = contract.TOOL_NAMES["decompose"] +WEEK = "date(day, '-6 days', 'weekday 1')" +BY_REGION = (f"SELECT region, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' " + "GROUP BY region, week ORDER BY region, week") +TOTALS = f"SELECT {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' GROUP BY week ORDER BY week" +EAST = (f"SELECT channel, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE region = '华东' AND day >= '2026-05-11' " + "GROUP BY channel, week ORDER BY channel, week") +WEEKLY = f"SELECT {WEEK} AS week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week" +TWO_WEEKS = {"period": "week", "base": "2026-05-11", "target": "2026-05-18", "measure": "gmv"} +A1_ARGS = {"mode": "additive", "query": "Q1", "by": ["region"], **TWO_WEEKS, "label": "销售额", + "headline": {"base": "Q2.r0.gmv", "target": "Q2.r1.gmv"}, "hypothesis": "下降是不是集中在某个地区"} +A2_ARGS = {"mode": "additive", "query": "Q3", "by": ["channel"], **TWO_WEEKS, "parent": {"of": "A1", "key": ["华东"]}, + "hypothesis": "华东是哪个渠道掉的"} + + +def _orders(): + out, i = [], 0 + base = {"华东": 400, "华南": 300, "华北": 250, "西南": 50} + for w in range(12): + monday = date(2026, 3, 2) + timedelta(weeks=w) + for region, amount in base.items(): + for channel, share in (("线上", 3), ("线下", 2)): + value = amount * share // 5 + (w % 3) * 2 + if w == 11 and region == "华东": + value = value * 7 // 10 + for offset in (2, 6): + i += 1 + out.append((i, (monday + timedelta(days=offset)).isoformat(), region, channel, value)) + return out + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture(autouse=True) +async def shop(tmp_path): + path = tmp_path / "shop.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, channel TEXT, amount INTEGER);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", _orders()) + db.commit() + db.close() + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() + if row is None: + row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) + session.add(row) + row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + yield + await engines.invalidate(source_id) + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def chain(*nodes): + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +class Script: + """Agent 依次发 calls 里的工具调用,再给结论;seen 记下每次带工具的调用看到的消息。""" + + def __init__(self, monkeypatch, calls): + from app.providers import mock_model + + self.seen: list[list] = [] + + def decide(model, messages): + if not model.tools: + return AIMessage(content="收尾") + self.seen.append(list(messages)) + done = sum(isinstance(m, ToolMessage) for m in messages) + if done < len(calls): + name, args = calls[done] + return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": f"call_{done}"}]) + return AIMessage(content="销售额下降主要来自华东。") + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) + + def replies(self): + return [m.content for m in self.seen[-1] if isinstance(m, ToolMessage)] + + +class Counting: + """真算子外面包一层只数次数(重放时不该再算)。""" + + def __init__(self, monkeypatch): + self.calls = 0 + real = analysis_ops.run + + async def run(*args, **kwargs): + self.calls += 1 + return await real(*args, **kwargs) + + monkeypatch.setattr(analysis_ops, "run", run) + + +async def wait(run_id, statuses=("succeeded", "failed")): + for _ in range(600): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None): + return row + raise AssertionError(f"run {run_id} 没有结束:{row.status}") + + +async def events(run_id, etype=None, node_id=None): + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id) + if etype: + q = q.where(RunEvent.type == etype) + if node_id: + q = q.where(RunEvent.node_id == node_id) + return [e.data for e in (await session.execute(q.order_by(RunEvent.seq))).scalars()] + + +def agent_graph(approval="never"): + return chain(node("start", "input"), + node("bot", "agent", prompt="销售额为什么下降", tools=[TOOL, DECOMPOSE], approval=approval), + node("out", "output", fields=[{"name": "answer", "value": "{{ nodes.bot.text }}"}])) + + +def recomputes(stored): + """按封存的工件复算(查询快照、上一层结论都按 sha256 取回),和封存的逐字段相同。""" + again = analysis_ops.recompute(stored, artifact_store.load, load_result=artifact_store.load) + assert contract.same(stored, again), {k: (stored.get(k), again.get(k)) for k in stored if stored.get(k) != again.get(k)} + + +AGENT_CALLS = [(TOOL, {"sql": BY_REGION}), (TOOL, {"sql": TOTALS}), (TOOL, {"sql": EAST}), + (DECOMPOSE, A1_ARGS), (DECOMPOSE, A2_ARGS)] + + +# -------------------------------------------------------------------------- +# agent 分析工具:拆解、下钻 +# -------------------------------------------------------------------------- + + +async def test_an_agent_decomposes_and_drills_down_with_the_real_operator(monkeypatch): + script = Script(monkeypatch, AGENT_CALLS) + run = await run_manager.start(graph=agent_graph(), input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + + ends = await events(row.id, "tool.end", "bot") + q1, q2, q3 = (ends[i]["query_artifact"] for i in range(3)) + a1_artifact, a2_artifact = ends[3]["analysis_artifact"], ends[4]["analysis_artifact"] + a1, a2 = artifact_store.load(a1_artifact), artifact_store.load(a2_artifact) + + # 结论:华东是主要来源,下钻到渠道是线上;总数闭合(headline 是 Q2 的两格) + assert a1["status"] == "ok" and a1["verdict"] == {"code": "driver", "subjects": [["华东"]], "direction": "down"} + assert "主要来自华东" in a1["phrase"] and a1["hypothesis"] == "下降是不是集中在某个地区" + assert {i["ref"] for i in a1["inputs"]} >= {"Q1", "Q2.r0.gmv", "Q2.r1.gmv"} + assert a2["verdict"]["code"] == "driver" and a2["verdict"]["subjects"] == [["线上"]] + # 下钻:算子按上一层结论的内容哈希认 parent,就是存下的那份工件 + [parent] = [i for i in a2["inputs"] if i.get("role") == "parent"] + assert parent == {"ref": "A1", "artifact": a1_artifact, "role": "parent"} + assert content_hash(canonical_json(a1)) == a1_artifact + + # 台账:两条 kind analysis,带 call_id,输入是那几份查询快照(下钻的另有上一层结论) + [finished] = await events(row.id, "node.finished", "bot") + analyses = [e for e in finished["evidence"] if e["kind"] == "analysis"] + assert [(e["call_id"], e["artifact"]) for e in analyses] == [("call_3", a1_artifact), ("call_4", a2_artifact)] + assert set(analyses[0]["inputs"]) == {q1, q2} and set(analyses[1]["inputs"]) == {q3, a1_artifact} + assert analyses[0] == contract.ledger_entry(a1, node_id="bot", exec_no=1, artifact=a1_artifact, call_id="call_3") + + # 交回模型的:编号、判定语、关键数 + replies = script.replies() + assert [r.split("\n", 1)[0] for r in replies] == ["【证据 Q1】", "【证据 Q2】", "【证据 Q3】", "【分析 A1】", "【分析 A2】"] + body = json.loads(replies[3].split("\n", 1)[1]) + assert body["phrase"] == a1["phrase"] and body["verdict"] == "driver" and body["values"] == a1["values"] + assert body["table"]["rows"][0][0] == "华东" + + # 按封存的工件复算,逐字段相同 + recomputes(a1) + recomputes(a2) + + +async def test_replaying_after_approvals_does_not_recompute(monkeypatch): + counted = Counting(monkeypatch) + script = Script(monkeypatch, AGENT_CALLS) + run = await run_manager.start(graph=agent_graph(approval="always"), input_payload={}) + for _ in AGENT_CALLS: + row = await wait(run.id, ("interrupted", "failed", "succeeded")) + assert row.status == "interrupted", row.error + await run_manager.resume(run.id, {"approved": True}) + row = await wait(run.id, ("succeeded", "failed", "interrupted")) + assert row.status == "succeeded", row.error + + assert counted.calls == 2, "重放时分析又算了一次" + ends = await events(row.id, "tool.end", "bot") + assert len(ends) == len(AGENT_CALLS) + a1_artifact, a2_artifact = ends[3]["analysis_artifact"], ends[4]["analysis_artifact"] + assert [r.split("\n", 1)[0] for r in script.replies()][-2:] == ["【分析 A1】", "【分析 A2】"] + [parent] = [i for i in artifact_store.load(a2_artifact)["inputs"] if i.get("role") == "parent"] + assert parent["artifact"] == a1_artifact + [finished] = await events(row.id, "node.finished", "bot") + assert [e["artifact"] for e in finished["evidence"] if e["kind"] == "analysis"] == [a1_artifact, a2_artifact] + + +# -------------------------------------------------------------------------- +# 分析卡:波动基线 + 变化拆解 +# -------------------------------------------------------------------------- + + +async def test_an_analysis_card_runs_the_real_operators(): + items = [ + {"id": "wow", "op": "baseline", "from": "weekly", "params": {"period": "week", "measure": "gmv", "label": "销售额"}}, + {"id": "where", "op": "decompose", "mode": "additive", "from": "by_region", + "params": {"by": ["region"], **TWO_WEEKS, "label": "销售额", + "headline": {"base": "weekly.r10.gmv", "target": "weekly.r11.gmv"}}}, + ] + graph = chain(node("start", "input"), node("weekly", "tool", tool=TOOL, args={"sql": WEEKLY}), + node("by_region", "tool", tool=TOOL, args={"sql": BY_REGION}), + node("card", "analysis", items=items), + node("out", "output", fields=[{"name": "判定", "value": "{{ nodes.card.results.wow.phrase }}"}])) + run = await run_manager.start(graph=graph, input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + + [finished] = await events(row.id, "node.finished", "card") + results = artifact_store.load(finished["artifact"])["results"] + wow, where = results["wow"], results["where"] + assert wow["status"] == "ok" and wow["verdict"] in ("outside_band", "far_outside_band") and wow["direction"] == "down" + assert where["status"] == "ok" and where["verdict"] == "driver" and "主要来自华东" in where["phrase"] + assert row.output["判定"] == wow["phrase"] + + baseline = artifact_store.load(wow["artifact"]) + # 本期完不完整按现查的表范围判(analysis.make_prober 按这次运行的数据源探 orders.day):探到了,不再说核对不了 + [probe] = baseline["probes"] + assert (probe["source"], probe["table"], probe["column"], probe["basis"]) == ("shop", "orders", "day", "live") + assert probe["max"] == "2026-05-24" + assert not [c for c in baseline["caveats"] if c["code"] == "comparability"] + decomposed = artifact_store.load(where["artifact"]) + assert decomposed["verdict"]["subjects"] == [["华东"]] + entries = finished["evidence"] + assert [(e["kind"], e["item"], e["artifact"]) for e in entries] == [ + ("analysis", "wow", wow["artifact"]), ("analysis", "where", where["artifact"])] + recomputes(baseline) + recomputes(decomposed) + + +# -------------------------------------------------------------------------- +# 口径卡的可比性挂钩:真 comparable +# -------------------------------------------------------------------------- + + +def _week_sql(start: str, end: str) -> str: + return f"SELECT SUM(amount) AS gmv FROM orders WHERE day BETWEEN '{start}' AND '{end}'" + + +@pytest.mark.parametrize("this_week, verdict, has_reasons", [ + (("2026-05-18", "2026-05-24"), "comparable", False), + (("2026-05-18", "2026-05-23"), "conditional", True), # 本期 6 天对上期 7 天:窗口长度、周末天数都不一样 +]) +async def test_a_caliber_card_checks_two_periods_with_the_real_operator(this_week, verdict, has_reasons): + metrics = [{"id": "wow", "name": "销售额环比", "decimals": 4, + "expression": "(cell(nodes.this_week, 0, 'gmv') - cell(nodes.last_week, 0, 'gmv'))" + " / cell(nodes.last_week, 0, 'gmv')"}] + graph = chain(node("start", "input"), node("this_week", "tool", tool=TOOL, args={"sql": _week_sql(*this_week)}), + node("last_week", "tool", tool=TOOL, args={"sql": _week_sql("2026-05-11", "2026-05-17")}), + node("card", "metrics", caliber="周报口径", metrics=metrics), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}])) + run = await run_manager.start(graph=graph, input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + [finished] = await events(row.id, "node.finished", "card") + [metric] = artifact_store.load(finished["artifact"])["metrics"] + found = metric["comparability"] + stored = artifact_store.load(found["artifact"]) + assert stored["op"] == "comparable" and stored["params"]["base_query"] == "last_week" + assert stored["params"]["target_query"] == "this_week" + assert found == {"artifact": found["artifact"], "verdict": verdict, "phrase": stored["phrase"], + "reasons": analysis_ops.comparable_reasons(stored)} + assert bool(found["reasons"]) is has_reasons + # 整期按现查的表范围判 + assert any(p.get("table") == "orders" and p.get("basis") == "live" for p in stored["probes"]) + recomputes(stored) + assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"] diff --git a/backend/tests/test_analysis_tools.py b/backend/tests/test_analysis_tools.py index 88a96815..4f22c8a7 100644 --- a/backend/tests/test_analysis_tools.py +++ b/backend/tests/test_analysis_tools.py @@ -23,7 +23,7 @@ from app.db.base import SessionLocal from app.db.models import Artifact, DataSource, Run, RunEvent from app.engine.runner import run_manager from app.tools.analysis import CALL_ERROR_HEAD, REFUSED_CODE -from app.tools.builtin.analysis import ANALYSIS_METHOD_PROMPT, ARGS, MAX_CALLS, METHOD_GUIDE, NO_SCOPE +from app.tools.builtin.analysis import ANALYSIS_METHOD_PROMPT, ARGS, METHOD_GUIDE, NO_SCOPE from tests.fixtures.analysis_ops import FakeOps TOOL = "db_query__shop" @@ -173,8 +173,11 @@ def test_the_five_tools_take_no_thresholds_and_share_one_method(): assert {"mode", "by", "period", "base", "target", "measure", "numerator", "denominator", "product", "factors", "headline", "parent", "label", "hypothesis"} <= set(ARGS["decompose"].model_fields) assert {"basis", "grain", "lag"} <= set(ARGS["baseline"].model_fields) - assert {"reference", "pairs", "mode"} <= set(ARGS["differ"].model_fields) - assert {"base_query", "target_query"} <= set(ARGS["comparable"].model_fields) + # 参数名以算子为准(app/analysis 各算子文件头):差异检验的均值是 mean / sd / n 或 sum / sumsq / n,逐行是 value + assert {"reference", "pairs", "mode", "mean", "sd", "sum", "sumsq", "n", "value"} <= set(ARGS["differ"].model_fields) + assert {"base_query", "target_query", "measure", "grain"} <= set(ARGS["comparable"].model_fields) + # 运行当天、目录事实、下钻层数由系统填,哪个工具的参数里都没有 + assert not any({"today", "catalog", "depth"} & set(model.model_fields) for model in ARGS.values()) def test_unknown_parameters_and_thresholds_are_rejected_like_any_tool(): @@ -211,6 +214,22 @@ async def test_calling_it_outside_an_agent_says_where_it_belongs(): assert found.value.message == NO_SCOPE +def test_input_snapshots_carry_the_payload_as_is(): + """交给算子的一份输入:查询工具的 payload 原样(SQL 检查结果、「查询区间里这张表没有数据」、查询当时记下的遮罩,可比性 + 检查按它们判数据问题),加上出处和遮罩列;合并查询写明方言(在 SQLite 里执行)。""" + from app.engine.merge_query import MERGE_SOURCE + from app.tools.analysis import input_snapshot + + content = {"columns": ["region", "gmv"], "rows": [["华东", 1]], "sql": "SELECT …", "source": "shop", + "column_types": {"region": "text", "gmv": "number"}, "truncated": False, + "checks": [{"level": "error", "code": "fanout_sum"}], "no_data": [{"columns": ["gmv"]}], + "mask_columns": ["phone"], "masked_at_query": ["tel"]} + snap = input_snapshot(content, ref="Q1", artifact="a" * 64, node_id="bot", masked=["tel", "phone"]) + assert snap == {**content, "ref": "Q1", "artifact": "a" * 64, "node_id": "bot", "masked": ["tel", "phone"]} + merged = input_snapshot({**content, "source": MERGE_SOURCE}, ref="m", artifact="b" * 64, node_id="m") + assert merged["dialect"] == "sqlite" and merged["masked"] == [] + + # -------------------------------------------------------------------------- # 方法说明:绑了分析工具的 Agent 注入一次,模型调用节点、没绑的 Agent 不注入 # -------------------------------------------------------------------------- @@ -265,11 +284,15 @@ async def test_an_analysis_reads_this_nodes_query_and_is_recorded(monkeypatch, s assert snap["ref"] == "Q1" and snap["artifact"] == query_artifact and snap["node_id"] == "bot" assert snap["source"] == "shop" and snap["sql"] == SQL and snap["truncated"] is False assert snap["columns"] == ["region", "week", "gmv"] and snap["rows"][0] == ["华东", "2026-W41", 820.0] - assert snap["column_types"] == ["text", "text", "number"] and snap["masked"] == [] + # 查询工具的 payload 原样(工件库里那份),加上出处和遮罩列 + assert snap["column_types"] == {"region": "text", "week": "text", "gmv": "number"} and snap["masked"] == [] + assert snap["row_count"] == 6 and snap == {**artifact_store.load(query_artifact), "ref": "Q1", + "artifact": query_artifact, "node_id": "bot", "masked": []} # 阈值:没有发起时的快照,现读设置(没存过就是缺省) assert call["settings"] == contract.DEFAULTS assert set(call["context"]) == {"today", "probe", "catalogs", "facts"} - assert call["context"]["probe"] is None, "还没有 W1-a 的 make_prober 时不给探查" + # 探查器按这次运行固定的数据版本解析数据源(analysis.make_prober),剖析事实是按表取的函数(probe.profile_facts) + assert call["context"]["probe"].dialect("shop") == "sqlite" and callable(call["context"]["facts"]["shop"]) # 结论落成工件,tool.end 带着它;台账记一条 kind analysis(带 call_id),在那次查询之后 analysis_artifact = ends[1]["analysis_artifact"] @@ -353,7 +376,9 @@ async def test_the_number_of_analyses_per_node_is_capped(monkeypatch, shop): await run_graph(agent_graph(), options={contract.SETTINGS_KEY: {"max_calls": 1}}) assert len(ops.calls) == 1 capped = tool_messages(script)[2] - assert capped == CALL_ERROR_HEAD + MAX_CALLS.format(done=1, label=contract.LABELS["max_calls"], limit=1) + import app.analysis as analysis_ops + + assert capped == CALL_ERROR_HEAD + analysis_ops.exhausted({"max_calls": 1}).message # 阈值来自发起时的快照 assert ops.calls[0]["settings"]["max_calls"] == 1 diff --git a/backend/tests/test_metric_comparability.py b/backend/tests/test_metric_comparability.py index ad33d65b..2fa604ea 100644 --- a/backend/tests/test_metric_comparability.py +++ b/backend/tests/test_metric_comparability.py @@ -1,9 +1,12 @@ """口径卡的可比性挂钩(docs/design/analysis-round/00-总方案.md §3.4 第 3 条):一个指标的输入落在两份不同的查询快照上、 两份 SQL 同形只差字面量(本周对上周)时,口径卡执行时自动做一次可比性检查,结论落成工件、记在指标的 comparability 上。 -- 同形只差日期 → 有 comparability,早的那期是基期(和表达式里谁先写无关);同一对快照只查一次 +- 同形只差日期 → 有 comparability,早的那期是基期(和表达式里谁先写无关);同一对快照只查一次;原因按算子的 + analysis.comparable_reasons 拆 - SQL 结构不同(退款额 / 销售额)、两次是同一条 SQL → 不查、不记日志 -- 算子出错、算不了、还没有算子(第 1 波的 run 抛 NotImplementedError)→ 不挡口径卡,记一条 metric_comparability_skipped +- 算子出错、算不了、算子没有实现(NotImplementedError)→ 不挡口径卡,记一条 metric_comparability_skipped + +这里的算子是替身(tests/fixtures/analysis_ops.FakeOps);真算子跑一遍的见 tests/test_analysis_e2e.py。 """ from __future__ import annotations @@ -18,7 +21,7 @@ from app.core import artifact_store from app.data.engine import engines from app.db.base import SessionLocal from app.db.models import DataSource, Run, RunEvent -from app.engine.nodes.metrics import COMPARABILITY_SKIPPED, comparability_reasons +from app.engine.nodes.metrics import COMPARABILITY_SKIPPED from app.engine.runner import run_manager from tests.fixtures.analysis_ops import FakeOps @@ -118,12 +121,23 @@ async def test_two_periods_of_the_same_query_get_a_comparability_check(monkeypat assert not [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED] +#: 可比性结论的检查表(照算子的 comparable.table_of:检查项、结果、基期、本期、说明) +CHECKS = {"columns": ["check", "status", "base", "target", "note"], "labels": {}, + "rows": [["同口径", "通过", None, None, None], ["窗口长度", "通过", "7 天", "7 天", None], + ["整期", "不通过", None, None, "本期数据只到 2026-10-15,还差 3 天"], + ["星期构成", "提醒", None, None, "两段的周末天数不同:2 天对 1 天"]]} + + async def test_an_incomparable_pair_lists_its_reasons(monkeypatch): - FakeOps(monkeypatch, verdicts=["incomparable"]) + import app.analysis as analysis_ops + + FakeOps(monkeypatch, verdicts=["incomparable"], table=CHECKS) _, card, _ = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW}])) found = by_id(card)["wow"]["comparability"] - assert found["verdict"] == "incomparable" and found["phrase"] == "两期不可比:本期只有 3 天数据" - assert found["reasons"] == ["本期只有 3 天数据"] + stored = artifact_store.load(found["artifact"]) + assert found == {"artifact": found["artifact"], "verdict": "incomparable", "phrase": stored["phrase"], + "reasons": analysis_ops.comparable_reasons(stored)} + assert found["reasons"] == ["本期数据只到 2026-10-15,还差 3 天", "两段的周末天数不同:2 天对 1 天"] async def test_different_structures_and_the_same_query_are_not_checked(monkeypatch): @@ -136,8 +150,9 @@ async def test_different_structures_and_the_same_query_are_not_checked(monkeypat assert not [log for log in [*logs, *logs2] if log.get("code") == COMPARABILITY_SKIPPED] -async def test_without_the_operator_the_card_still_runs(): - """第 1 波的 app.analysis.run 还抛 NotImplementedError:口径卡照常算,照实记一条没做成。""" +async def test_an_operator_that_is_not_there_does_not_block_the_card(monkeypatch): + """算子没有实现(合并前的骨架抛 NotImplementedError):口径卡照常算,照实记一条没做成。""" + FakeOps(monkeypatch, errors=[NotImplementedError("analysis.run 还没有实现")]) _, card, logs = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW, "decimals": 4}])) wow = by_id(card)["wow"] assert "comparability" not in wow and wow["value"] == pytest.approx((450 + 300 - 900) / 900, abs=1e-4) @@ -157,12 +172,3 @@ async def test_failures_do_not_block_the_card(monkeypatch, ops_kw): assert "comparability" not in by_id(card)["wow"] [log] = [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED] assert log["reason"] - - -def test_reasons_come_from_the_check_table_when_there_is_one(): - result = {"verdict": {"code": "conditional"}, "phrase": "两期有条件可比:两段的周末天数不同", - "table": {"columns": ["check", "status", "reason"], "labels": {}, - "rows": [["同口径", "pass", None], ["窗口长度", "warn", "窗口长度不同:7 天对 6 天"], - ["星期构成", "warn", "两段的周末天数不同:2 天对 1 天"], ["整期", "unknown", "核对不了"]]}} - assert comparability_reasons(result) == ["窗口长度不同:7 天对 6 天", "两段的周末天数不同:2 天对 1 天"] - assert comparability_reasons({"verdict": {"code": "comparable"}, "phrase": "两期可比"}) == []