yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 00:31:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

口径卡给容器型 vars 输入另存完整值（value_artifact），供数字推演重算

Browse Source 
卡里的输入对列表、字典只记「[12 项]」这样的大概，事后要排除几行、改一个数再算一遍，手里没有完整的值。
nodes.*、input.* 的完整值在那个节点封存的产出工件里，agent 字段能按出处从查询快照还原；vars 是谁写的、
写进去的是不是节点产出本身说不准，只能在口径卡算的那一刻存下。

- 此后容器型 vars.* 输入落一份 metric_input 工件 {path, value}（put_json，随运行封存），输入里记下 value_artifact；
  id 写在内容寻址的指标集里，不另起台账条目
- 大过 4 MB 的不存，写失败也不存：口径卡照常算出来，只是这个输入以后推演不了
- 金样测试（改动之前的代码真跑出来的工件）：只有标量输入的卡逐字节相同、工件 id 不变；带容器型 vars 输入的卡
  去掉新键后和以前一字不差
This commit is contained in:
ThearchyHelios
2026-10-05 03:26:43 +08:00
parent 35bb23d372
commit 60cd5a47bb
3 changed files with 379 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/engine/nodes/metrics.py
+33
View File
@@ -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
backend/tests/fixtures/whatif/golden_cards.json
Vendored
+208
View File
@@ -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
}
]
}
}
}
backend/tests/test_metric_value_artifact.py
+138
View File
@@ -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"]