From 60cd5a47bb43f39cbac1e3977e2c93b317b76404 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 03:26:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=A3=E5=BE=84=E5=8D=A1=E7=BB=99=E5=AE=B9?= =?UTF-8?q?=E5=99=A8=E5=9E=8B=20vars=20=E8=BE=93=E5=85=A5=E5=8F=A6?= =?UTF-8?q?=E5=AD=98=E5=AE=8C=E6=95=B4=E5=80=BC=EF=BC=88value=5Fartifact?= =?UTF-8?q?=EF=BC=89=EF=BC=8C=E4=BE=9B=E6=95=B0=E5=AD=97=E6=8E=A8=E6=BC=94?= =?UTF-8?q?=E9=87=8D=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 卡里的输入对列表、字典只记「[12 项]」这样的大概,事后要排除几行、改一个数再算一遍,手里没有完整的值。 nodes.*、input.* 的完整值在那个节点封存的产出工件里,agent 字段能按出处从查询快照还原;vars 是谁写的、 写进去的是不是节点产出本身说不准,只能在口径卡算的那一刻存下。 - 此后容器型 vars.* 输入落一份 metric_input 工件 {path, value}(put_json,随运行封存),输入里记下 value_artifact; id 写在内容寻址的指标集里,不另起台账条目 - 大过 4 MB 的不存,写失败也不存:口径卡照常算出来,只是这个输入以后推演不了 - 金样测试(改动之前的代码真跑出来的工件):只有标量输入的卡逐字节相同、工件 id 不变;带容器型 vars 输入的卡 去掉新键后和以前一字不差 --- backend/app/engine/nodes/metrics.py | 33 +++ .../tests/fixtures/whatif/golden_cards.json | 208 ++++++++++++++++++ backend/tests/test_metric_value_artifact.py | 138 ++++++++++++ 3 files changed, 379 insertions(+) create mode 100644 backend/tests/fixtures/whatif/golden_cards.json create mode 100644 backend/tests/test_metric_value_artifact.py diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index 0d9f7105..096aeed8 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -137,6 +137,13 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: incomplete = {"incomplete": True, "incomplete_reason": _incomplete_reason(sources), "truncated_sources": sources} inputs = [_input_of(path, values[path], state, ctx, tctx) for path in paths] + for path, item in zip(paths, inputs): + # 容器型的 vars.* 输入(列表、字典)另存一份完整值:卡里只记了「[12 项]」这样的大概,事后推演 + # (engine/recompute.py)要拿完整的值重算。nodes.*、input.* 的完整值在那个节点封存的产出工件里, + # agent 字段能按出处从查询快照还原,都不用另存;vars 是谁写的、写进去的是不是节点产出本身说不准, + # 只能在这一刻存下。只给这种输入多一个键:没有容器型 vars 输入的卡和以前逐字节相同 + if _container_var(path, values[path]) and (stored := await _store_value(path, values[path], ctx)): + item["value_artifact"] = stored sql_check: dict[str, Any] = {} if failed := _failed_queries(inputs, list(use.sources.values()), paths, tctx, checked): sql_check = {"sql_check_failed": True, "sql_check_reason": _sql_check_reason(failed), @@ -726,6 +733,32 @@ def _recompute(substituted: str, tctx: dict[str, Any], value: Any, decimals: int return again == value +#: 另存的完整值最大多少字节(规范 JSON)。查询结果整份赋给变量时可能很大,每张卡都再存一份不值当:大过这个数就 +#: 不存,推演照实说当时没有保存完整值 +VALUE_ARTIFACT_MAX = 4 * 1024 * 1024 + + +def _container_var(path: str, value: Any) -> bool: + head = _REF_HEAD.match(path) + return isinstance(value, (list, dict)) and head is not None and head.group(1) == "vars" + + +async def _store_value(path: str, value: Any, ctx: NodeContext) -> str | None: + """容器型 vars 输入的完整值落成 metric_input 工件:{path, value}。它的 id 记在指标集里,指标集内容寻址、 + 在封存范围内,完整值也就随运行封存,不另起台账条目。存不下(太大、写失败)就不存:口径卡照常算出来, + 只是这个输入以后推演不了。""" + from app.core.artifact_store import canonical_json, put_json + + content = {"path": path, "value": value} + try: + if len(canonical_json(content).encode()) > VALUE_ARTIFACT_MAX: + return None + return await put_json(content, kind="metric_input", run_id=ctx.run.run_id, node_id=ctx.node.id, + meta={"path": path}) + except Exception: # noqa: BLE001 - 完整值存不下不该毁掉口径卡 + return None + + async def _store(card: dict[str, Any], caliber: str, version: str, ctx: NodeContext) -> str | None: """整张卡落成 metric_set 工件。写失败不毁掉运行,只是这张卡没法被引用。""" from app.core.artifact_store import put_json diff --git a/backend/tests/fixtures/whatif/golden_cards.json b/backend/tests/fixtures/whatif/golden_cards.json new file mode 100644 index 00000000..b2c13128 --- /dev/null +++ b/backend/tests/fixtures/whatif/golden_cards.json @@ -0,0 +1,208 @@ +{ + "_comment": "口径卡的金样:加 value_artifact 之前的代码(feat/verify-suite 35bb23d)真跑出来的 metric_set 工件。scalar 是只有标量输入的卡,改动之后工件必须逐字节相同(工件 id 不变);box 带容器型 vars 输入,改动之后只多 value_artifact 这一个键。虚构数据。", + "scalar": { + "artifact": "4af4e64d2fac8bae1f0acc8f65573afb3b527547878c053cc9b6d6687e321408", + "card": { + "caliber": "推演口径", + "caliber_version": "v3", + "kind": "metric_set", + "metrics": [ + { + "decimals": 1, + "expression": "vars.kpi.gmv", + "format": "thousands", + "id": "gmv", + "inputs": [ + { + "field": "gmv", + "node_id": "fetch", + "path": "vars.kpi.gmv", + "status": "ok", + "value": 45678.5, + "via": "transform" + } + ], + "name": "销售额", + "recompute_ok": true, + "rendered": "45,678.5元", + "status": "ok", + "substituted": "45678.5", + "unit": "元", + "value": 45678.5 + }, + { + "decimals": 1, + "expression": "round((vars.kpi.gmv - vars.kpi.gmv_prev) / vars.kpi.gmv_prev * 100, 1)", + "format": "plain", + "id": "wow", + "inputs": [ + { + "field": "gmv", + "node_id": "fetch", + "path": "vars.kpi.gmv", + "status": "ok", + "value": 45678.5, + "via": "transform" + }, + { + "field": "gmv_prev", + "node_id": "fetch", + "path": "vars.kpi.gmv_prev", + "status": "ok", + "value": 42010.0, + "via": "transform" + } + ], + "name": "环比增幅", + "recompute_ok": true, + "rendered": "8.7%", + "status": "ok", + "substituted": "round((45678.5 - 42010.0) / 42010.0 * 100, 1)", + "unit": "%", + "value": 8.7 + }, + { + "decimals": null, + "expression": "vars.kpi.orders", + "format": "thousands", + "id": "orders", + "inputs": [ + { + "field": "orders", + "node_id": "fetch", + "path": "vars.kpi.orders", + "status": "ok", + "value": 1234, + "via": "transform" + } + ], + "name": "订单数", + "recompute_ok": true, + "rendered": "1,234单", + "status": "ok", + "substituted": "1234", + "unit": "单", + "value": 1234 + }, + { + "decimals": 4, + "expression": "vars.kpi.refunds / vars.kpi.orders", + "format": "percent_of_ratio", + "id": "refund_rate", + "inputs": [ + { + "field": "refunds", + "node_id": "fetch", + "path": "vars.kpi.refunds", + "status": "ok", + "value": 29, + "via": "transform" + }, + { + "field": "orders", + "node_id": "fetch", + "path": "vars.kpi.orders", + "status": "ok", + "value": 1234, + "via": "transform" + } + ], + "name": "退款率", + "recompute_ok": true, + "rendered": "2.35%", + "status": "ok", + "substituted": "29 / 1234", + "unit": "", + "value": 0.0235 + } + ] + } + }, + "box": { + "artifact": "e2ec0b24639f78e4c14a22408516e5e6d7e4890025a6d0cbcad6fa7828bfbee9", + "card": { + "caliber": "推演口径", + "caliber_version": "v3", + "kind": "metric_set", + "metrics": [ + { + "decimals": null, + "expression": "len(vars.kpi.items)", + "format": "thousands", + "id": "n", + "inputs": [ + { + "field": "items", + "node_id": "fetch", + "path": "vars.kpi.items", + "status": "ok", + "value": "[3 项]", + "via": "transform" + } + ], + "name": "条数", + "recompute_ok": true, + "rendered": "3", + "status": "ok", + "substituted": "len(vars.kpi.items)", + "unit": "", + "value": 3 + }, + { + "decimals": null, + "expression": "len(vars.kpi.meta)", + "format": "thousands", + "id": "m", + "inputs": [ + { + "field": "meta", + "node_id": "fetch", + "path": "vars.kpi.meta", + "status": "ok", + "value": "{2 个字段}", + "via": "transform" + } + ], + "name": "字段数", + "recompute_ok": true, + "rendered": "2", + "status": "ok", + "substituted": "len(vars.kpi.meta)", + "unit": "", + "value": 2 + }, + { + "decimals": null, + "expression": "sum(vars.kpi.items) + vars.kpi.gmv", + "format": "thousands", + "id": "s", + "inputs": [ + { + "field": "items", + "node_id": "fetch", + "path": "vars.kpi.items", + "status": "ok", + "value": "[3 项]", + "via": "transform" + }, + { + "field": "gmv", + "node_id": "fetch", + "path": "vars.kpi.gmv", + "status": "ok", + "value": 10, + "via": "transform" + } + ], + "name": "合计", + "recompute_ok": true, + "rendered": "25", + "status": "ok", + "substituted": "sum(vars.kpi.items) + 10", + "unit": "", + "value": 25 + } + ] + } + } +} diff --git a/backend/tests/test_metric_value_artifact.py b/backend/tests/test_metric_value_artifact.py new file mode 100644 index 00000000..4117cef6 --- /dev/null +++ b/backend/tests/test_metric_value_artifact.py @@ -0,0 +1,138 @@ +"""口径卡给容器型的 vars.* 输入另存一份完整值(value_artifact),数字推演据此重算。 + +卡里的输入对列表、字典只记一个大概(「[12 项]」),事后要改一行、重算一遍,手里没有完整的值。nodes.*、input.* +的完整值在那个节点封存的产出工件里,agent 字段能按出处从查询快照还原;vars 是谁写的、写进去的是不是节点产出 +本身说不准,只能在口径卡算的那一刻存下来。 + +守的是「只多一个键」:只有标量输入的卡逐字节和改动之前一样(工件 id 不变,引用它的报告也就不变);带容器型 +vars 输入的卡,去掉 value_artifact 之后和改动之前一样。金样是改动之前的代码真跑出来的(fixtures/whatif)。 +""" +from __future__ import annotations + +import asyncio +import copy +import json +from pathlib import Path + +import pytest +from sqlalchemy import select + +from app.core import artifact_store +from app.db.base import SessionLocal +from app.db.models import Artifact, Run, RunEvent +from app.engine.runner import run_manager + +GOLDEN = json.loads((Path(__file__).parent / "fixtures" / "whatif" / "golden_cards.json").read_text("utf-8")) + +SCALAR_KPI = "{'gmv': 45678.5, 'gmv_prev': 42010.0, 'orders': 1234, 'refunds': 29}" +SCALAR = [ + {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"}, + {"id": "wow", "name": "环比增幅", "unit": "%", "decimals": 1, "format": "plain", + "expression": "round((vars.kpi.gmv - vars.kpi.gmv_prev) / vars.kpi.gmv_prev * 100, 1)"}, + {"id": "orders", "name": "订单数", "unit": "单", "expression": "vars.kpi.orders"}, + {"id": "refund_rate", "name": "退款率", "decimals": 4, "format": "percent_of_ratio", + "expression": "vars.kpi.refunds / vars.kpi.orders"}, +] +BOX_KPI = "{'items': [3, 5, 7], 'meta': {'a': 1, 'b': 2}, 'gmv': 10}" +BOX = [ + {"id": "n", "name": "条数", "expression": "len(vars.kpi.items)"}, + {"id": "m", "name": "字段数", "expression": "len(vars.kpi.meta)"}, + {"id": "s", "name": "合计", "expression": "sum(vars.kpi.items) + vars.kpi.gmv"}, +] + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def graph(expr, metrics): + nodes = [node("start", "input"), node("fetch", "transform", mode="expression", expression=expr, assign_to="kpi"), + node("card", "metrics", caliber="推演口径", caliber_version="v3", metrics=metrics), + node("out", "output", fields=[{"name": "清单", "value": "{{ nodes.card.text }}"}])] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def finished(g) -> tuple[Run, dict]: + run = await run_manager.start(graph=g, input_payload={}) + for _ in range(300): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + break + assert row.status == "succeeded", row.error + async with SessionLocal() as session: + [event] = (await session.execute(select(RunEvent).where( + RunEvent.run_id == run.id, RunEvent.type == "node.finished", RunEvent.node_id == "card"))).scalars() + return row, event.data + + +def without_value_artifacts(card: dict) -> dict: + out = copy.deepcopy(card) + for metric in out["metrics"]: + for item in metric.get("inputs") or []: + item.pop("value_artifact", None) + return out + + +async def test_a_card_with_only_scalar_inputs_is_byte_identical_to_before(): + row, data = await finished(graph(SCALAR_KPI, SCALAR)) + [entry] = data["evidence"] + assert entry["artifact"] == GOLDEN["scalar"]["artifact"], "只有标量输入的卡,工件 id 不能变" + assert artifact_store.load(entry["artifact"]) == GOLDEN["scalar"]["card"] + assert "value_artifact" not in json.dumps(artifact_store.load(entry["artifact"])) + + +async def test_container_vars_inputs_carry_their_full_value(): + row, data = await finished(graph(BOX_KPI, BOX)) + [entry] = data["evidence"] + card = artifact_store.load(entry["artifact"]) + # 去掉新加的键,和改动之前一字不差:值、渲染、代入式、复算、其余字段都不变 + assert without_value_artifacts(card) == GOLDEN["box"]["card"] + + by_path = {(m["id"], i["path"]): i for m in card["metrics"] for i in m["inputs"]} + items, meta = by_path[("n", "vars.kpi.items")], by_path[("m", "vars.kpi.meta")] + assert items["value"] == "[3 项]" and meta["value"] == "{2 个字段}" + assert artifact_store.load(items["value_artifact"]) == {"path": "vars.kpi.items", "value": [3, 5, 7]} + assert artifact_store.load(meta["value_artifact"]) == {"path": "vars.kpi.meta", "value": {"a": 1, "b": 2}} + # 同一个输入在两个指标里:同一份内容只存一份 + assert by_path[("s", "vars.kpi.items")]["value_artifact"] == items["value_artifact"] + # 标量输入不另存 + assert "value_artifact" not in by_path[("s", "vars.kpi.gmv")] + + async with SessionLocal() as session: + stored = await session.get(Artifact, (items["value_artifact"], row.id)) + assert stored is not None and stored.kind == "metric_input" and stored.node_id == "card" + # 台账只有指标集这一条:完整值经指标集(内容寻址、封存范围内)引用,不另起台账条目 + assert [e["kind"] for e in data["evidence"]] == ["metric_set"] + assert row.output["清单"] == "指标清单(口径 推演口径 @ v3):\n- n(条数)= 3\n- m(字段数)= 2\n- s(合计)= 25" + + +async def test_only_vars_containers_are_stored(): + """nodes.* 的完整值在那个节点封存的产出工件里,不另存。""" + row, data = await finished(graph(BOX_KPI, [ + {"id": "n", "name": "条数", "expression": "len(nodes.fetch.items)"}, + {"id": "v", "name": "条数", "expression": "len(vars.kpi.items)"}, + ])) + card = artifact_store.load(data["evidence"][0]["artifact"]) + [n_input] = card["metrics"][0]["inputs"] + [v_input] = card["metrics"][1]["inputs"] + assert n_input["path"] == "nodes.fetch.items" and "value_artifact" not in n_input + assert v_input["path"] == "vars.kpi.items" and "value_artifact" in v_input + + +async def test_a_value_too_large_to_keep_is_skipped(monkeypatch): + """完整值大过上限就不另存:推演据此说明当时没有保存,卡照常算出来。""" + from app.engine.nodes import metrics + + monkeypatch.setattr(metrics, "VALUE_ARTIFACT_MAX", 10) + row, data = await finished(graph(BOX_KPI, BOX)) + card = artifact_store.load(data["evidence"][0]["artifact"]) + assert card == GOLDEN["box"]["card"]