mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
口径卡来源支持节点改名映射 caliber_from.nodes(口径账本做法 4 ③)
为什么:组合式问数、对账、召回要把一份口径的取数节点拷进另一张图,同一张图里放两份口径时节点 id 会撞,拷过来的取数节点得改名(c1_fetch)。以前 caliber_from 只认同名的取数节点,改了名口径卡就取不到输入。
做了什么:
- metrics._pinned_caliber 认 caliber_from.nodes {原 id: 新 id}:按语法树把式子里 nodes.<原 id>(含 cell(nodes.X, …) 的第一个参数、nodes['X'])的根换成新 id 再算,字符串字面量里的字不动;指标集的 source 照记映射。映射形状不对节点失败并说清哪一处。
- 画布校验 _caliber_from_issues:映射的形状和运行时同一个判断(rename_problem),改名后的节点要在图里、在口径卡上游。
- 升版检查照常挡、照常记事件,事件带上 caliber_nodes;证据的 _caliber_source / _caliber_upgrade 照带映射。
- 运行对比按映射把式子换回那一版里的写法再比:同一份口径换了节点名不算口径变了。推演按卡上记的改名后的式子原样重算。
- 没写 nodes 的老图一字不差:式子不经 unparse、来源只有三项、升版事件没有新键(单独测)。
测试:tests/test_caliber_from_nodes.py(改写规则、运行时取值、证据链、推演、校验、升版检查、对比、老图不变)。
This commit is contained in:
@@ -1466,7 +1466,9 @@ def _located(locator: dict[str, Any]) -> list[tuple[Any, Any]]:
|
||||
|
||||
|
||||
async def _caliber_source(source: Any) -> dict[str, Any] | None:
|
||||
"""钉住别处口径卡(caliber_from)时,定义取自哪个工作流的哪一版;带上工作流现在的名字方便认。"""
|
||||
"""钉住别处口径卡(caliber_from)时,定义取自哪个工作流的哪一版;带上工作流现在的名字方便认。取数节点拷进这张图时
|
||||
改了名的(caliber_from.nodes,next-round 02 做法 4),来源里照带 nodes 映射:指标的输入写的是改名后的节点,面板据此
|
||||
说清它们对应那一版里的哪个取数节点。"""
|
||||
if not isinstance(source, dict) or not source.get("workflow_id"):
|
||||
return None
|
||||
async with SessionLocal() as session:
|
||||
@@ -1478,7 +1480,11 @@ def _caliber_upgrade(event: dict[str, Any] | None) -> dict[str, Any] | None:
|
||||
"""这张口径卡在正式运行发起时的升版处置:钉在哪一版、上游最新是哪一版、怎么处置的。"""
|
||||
if not isinstance(event, dict):
|
||||
return None
|
||||
return {k: event.get(k) for k in ("workflow_id", "caliber_node", "pinned", "latest", "policy", "policy_label")}
|
||||
out = {k: event.get(k) for k in ("workflow_id", "caliber_node", "pinned", "latest", "policy", "policy_label")}
|
||||
if isinstance(event.get("caliber_nodes"), dict):
|
||||
# 取数节点拷进这张图时改了名(caliber_from.nodes):照给映射;没改名的卡没有这个键,老运行一字不差
|
||||
out["caliber_nodes"] = dict(event["caliber_nodes"])
|
||||
return out
|
||||
|
||||
|
||||
#: 双盲复算节点产出里一个指标的结果(engine/nodes/dual_check.py 的 SAME / DIFFERENT / UNCHECKED),原样交给界面
|
||||
|
||||
@@ -65,7 +65,7 @@ from app.engine.evidence import DEFAULT_FORMAT, RenderError, is_number, render_n
|
||||
from app.engine.expressions import ExpressionError, cell_value, leaf_refs, parse_expression
|
||||
from app.engine.labels import TYPE_LABEL, field_label
|
||||
from app.engine.merge_query import MERGE_SOURCE
|
||||
from app.engine.nodes.metrics import _REF_HEAD, _display_of
|
||||
from app.engine.nodes.metrics import _REF_HEAD, _display_of, source_expression
|
||||
from app.engine.toolcalls import QUERY_PREFIX
|
||||
|
||||
UNCHANGED = "unchanged"
|
||||
@@ -568,9 +568,13 @@ def _shown(value: Any) -> str:
|
||||
return text if len(text) <= _SHOWN_MAX else text[:_SHOWN_MAX] + "…"
|
||||
|
||||
|
||||
def _expression(metric: dict[str, Any]) -> str:
|
||||
"""表达式的规范写法:空白、括号写法不同的算同一个。解析不了的照原文。"""
|
||||
raw = str(metric.get("expression") or "").strip()
|
||||
def _expression(metric: dict[str, Any], source: Any = None) -> str:
|
||||
"""表达式的规范写法:空白、括号写法不同的算同一个。解析不了的照原文。
|
||||
|
||||
source 是这张卡的来源(指标集上的 source):钉住别处口径卡、取数节点拷进图里时改了名的(caliber_from.nodes,
|
||||
next-round 02 做法 4),先按映射换回那一版里的写法再比——同一份口径换了节点名不算口径变了;取数本身变没变照旧按
|
||||
输入的出处说。"""
|
||||
raw = source_expression(str(metric.get("expression") or "").strip(), source)
|
||||
try:
|
||||
return ast.unparse(parse_expression(raw)[0])
|
||||
except (ExpressionError, ValueError):
|
||||
@@ -1715,7 +1719,7 @@ def _caliber_reasons(bm: dict[str, Any], tm: dict[str, Any], bc: dict[str, Any],
|
||||
defs: tuple[dict[str, Any] | None, dict[str, Any] | None] = (None, None)) -> list[str]:
|
||||
"""口径变了的地方,一处一句;没变返回空列表。defs 是两次运行图里这个指标的定义(老卡没记小数位、格式时用)。"""
|
||||
out = []
|
||||
if _expression(bm) != _expression(tm):
|
||||
if _expression(bm, bc.get("source")) != _expression(tm, tc.get("source")):
|
||||
out.append(f"表达式由「{_shown(bm.get('expression'))}」改为「{_shown(tm.get('expression'))}」")
|
||||
before, after = str(bc.get("caliber_version") or ""), str(tc.get("caliber_version") or "")
|
||||
if before != after:
|
||||
|
||||
@@ -699,6 +699,10 @@ async def unresolved_caliber_upgrades(
|
||||
ref = node.config["caliber_from"]
|
||||
pinned, wf_id = ref.get("workflow_version"), ref.get("workflow_id")
|
||||
extra = {"caliber_node": ref.get("node_id")}
|
||||
if isinstance(ref.get("nodes"), dict) and ref["nodes"]:
|
||||
# 拷进这张图时改了名的取数节点(next-round 02 做法 4):升版事件照记映射,证据面板说得清这张卡读的是哪几个
|
||||
# 节点;升不升版、怎么处置和没改名的卡同一套规则
|
||||
extra["caliber_nodes"] = dict(ref["nodes"])
|
||||
what = f"「{node.title}」(口径卡)引用的口径卡固定在 v{pinned}"
|
||||
else:
|
||||
continue
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
@@ -282,8 +283,87 @@ async def _pinned_caliber(ctx: NodeContext) -> tuple[dict[str, Any], dict[str, A
|
||||
raise NodeError(ctx.node.id, f"「{name}」v{version} 中的「{title}」本身也引用自其他工作流的口径卡(源头是 {where})。"
|
||||
"请直接固定到源头的口径卡,否则升版时无法正确比对版本")
|
||||
source = {"workflow_id": wf_id, "workflow_version": version, "node_id": node_id}
|
||||
metrics = cfg.get("metrics") or []
|
||||
rename = ref.get("nodes") if isinstance(ref, dict) else None
|
||||
if rename not in (None, {}):
|
||||
# 节点改名映射(next-round 02 做法 4):那一版的式子读 nodes.fetch,这张图里拷过来的取数节点叫 c1_fetch。式子按
|
||||
# 语法树把 nodes.<原 id> 的根换成新 id 再算;来源里照记映射,证据、对比、升版处置据此认得是同一份口径
|
||||
if problem := rename_problem(rename):
|
||||
raise NodeError(ctx.node.id, f"「{field_label('caliber_from')}」的节点改名映射有误:{problem}")
|
||||
metrics = [rename_definition(m, rename) for m in metrics]
|
||||
source["nodes"] = dict(rename)
|
||||
return {"caliber": cfg.get("caliber") or title, "caliber_version": str(cfg.get("caliber_version") or "v1"),
|
||||
"metrics": cfg.get("metrics") or []}, source
|
||||
"metrics": metrics}, source
|
||||
|
||||
|
||||
#: 改名映射里的节点 id:要能写进式子的取值链(nodes.<id>),和画布上的节点 id 同一种写法
|
||||
_NODE_ID = re.compile(r"^[A-Za-z_][\w-]*$")
|
||||
|
||||
|
||||
def rename_problem(rename: Any) -> str | None:
|
||||
"""caliber_from.nodes 的形状对不对:{原 id: 新 id},两边都是非空的节点 id,新 id 不重复。对的是 None。
|
||||
画布校验(engine/schema._caliber_from_issues)和运行时同一个判断;新 id 在不在这张图里由校验另查。"""
|
||||
if not isinstance(rename, dict) or not rename:
|
||||
return "需要写成「原节点 id → 这张图里的节点 id」的对照"
|
||||
seen: set[str] = set()
|
||||
for old, new in rename.items():
|
||||
if not isinstance(old, str) or not _NODE_ID.match(old) or not isinstance(new, str) or not _NODE_ID.match(new):
|
||||
return f"「{old}」→「{new}」不是节点 id 的写法"
|
||||
if new in seen:
|
||||
return f"有两个节点都改名成了「{new}」"
|
||||
seen.add(new)
|
||||
return None
|
||||
|
||||
|
||||
class _RenameNodes(ast.NodeTransformer):
|
||||
"""nodes.<原 id>(含 cell(nodes.<原 id>, …) 的第一个参数、nodes['原 id'])的根换成新 id。别的根(vars、input)不动。"""
|
||||
|
||||
def __init__(self, rename: dict[str, str]) -> None:
|
||||
self.rename = rename
|
||||
self.hits = 0
|
||||
|
||||
def visit_Attribute(self, node: ast.Attribute) -> ast.AST:
|
||||
if isinstance(node.value, ast.Name) and node.value.id == "nodes" and node.attr in self.rename:
|
||||
self.hits += 1
|
||||
return ast.copy_location(ast.Attribute(value=node.value, attr=self.rename[node.attr], ctx=node.ctx), node)
|
||||
return self.generic_visit(node)
|
||||
|
||||
def visit_Subscript(self, node: ast.Subscript) -> ast.AST:
|
||||
key = node.slice
|
||||
if isinstance(node.value, ast.Name) and node.value.id == "nodes" and isinstance(key, ast.Constant) \
|
||||
and isinstance(key.value, str) and key.value in self.rename:
|
||||
self.hits += 1
|
||||
return ast.copy_location(ast.Subscript(value=node.value, slice=ast.Constant(self.rename[key.value]),
|
||||
ctx=node.ctx), node)
|
||||
return self.generic_visit(node)
|
||||
|
||||
|
||||
def rename_expression(expression: str, rename: dict[str, str]) -> str:
|
||||
"""把式子里 nodes.<原 id> 的根按映射换成新 id(语法树上换,字符串字面量里的字不动)。一处都没换到、解析不了的原样交回:
|
||||
解析不了的式子后面求值时照常报「表达式有误」。"""
|
||||
try:
|
||||
tree = ast.parse(expression, mode="eval")
|
||||
except SyntaxError:
|
||||
return expression
|
||||
renamer = _RenameNodes(rename)
|
||||
tree = ast.fix_missing_locations(renamer.visit(tree))
|
||||
return ast.unparse(tree) if renamer.hits else expression
|
||||
|
||||
|
||||
def rename_definition(definition: Any, rename: dict[str, str]) -> Any:
|
||||
"""一个指标定义按映射改写式子(别的项原样)。不是对象的原样交回,后面照常报「指标定义不完整」。"""
|
||||
if not isinstance(definition, dict) or not isinstance(definition.get("expression"), str):
|
||||
return definition
|
||||
return {**definition, "expression": rename_expression(definition["expression"], rename)}
|
||||
|
||||
|
||||
def source_expression(expression: str, source: Any) -> str:
|
||||
"""卡上记的式子换回那一版里的写法(来源里有改名映射时按映射反着换):对比两次运行的口径(engine/compare)按它比,
|
||||
同一份口径拷进图里时取数节点改了名,不算口径变了。没有映射的原样交回。"""
|
||||
rename = source.get("nodes") if isinstance(source, dict) else None
|
||||
if not isinstance(rename, dict) or not rename or rename_problem(rename) is not None:
|
||||
return expression
|
||||
return rename_expression(expression, {new: old for old, new in rename.items()})
|
||||
|
||||
|
||||
#: 新写法的小数位范围。负数是取整到十位、百位(round(x, -2))
|
||||
|
||||
@@ -1443,7 +1443,7 @@ def evidence_issues(spec: GraphSpec) -> list[ValidationIssue]:
|
||||
"evidence_role", level="error")
|
||||
elif node.type == NodeType.METRICS:
|
||||
if cfg.get("caliber_from") not in (None, "", {}):
|
||||
out.extend(_caliber_from_issues(node, UPGRADE_POLICIES, UPGRADE_POLICY_LABELS))
|
||||
out.extend(_caliber_from_issues(node, UPGRADE_POLICIES, UPGRADE_POLICY_LABELS, spec))
|
||||
warned: set[str] = set()
|
||||
for i, producer in _caliber_inputs(node, spec):
|
||||
role = str(producer.config.get("evidence_role") or "compute")
|
||||
@@ -1457,8 +1457,12 @@ def evidence_issues(spec: GraphSpec) -> list[ValidationIssue]:
|
||||
return out
|
||||
|
||||
|
||||
def _caliber_from_issues(node: GraphNode, policies: set[str], labels: dict[str, str]) -> list[ValidationIssue]:
|
||||
"""钉住别的工作流里一张口径卡:{workflow_id, workflow_version, node_id} 三项都要写对。"""
|
||||
def _caliber_from_issues(node: GraphNode, policies: set[str], labels: dict[str, str],
|
||||
spec: GraphSpec | None = None) -> list[ValidationIssue]:
|
||||
"""钉住别的工作流里一张口径卡:{workflow_id, workflow_version, node_id} 三项都要写对。
|
||||
|
||||
可选的 nodes(节点改名映射,next-round 02 做法 4):那一版口径卡读的取数节点拷进这张图时改了名,{原 id: 这张图里的 id}。
|
||||
形状和运行时同一个判断(metrics.rename_problem);改名后的节点要在这张图里、在口径卡上游——不然运行时口径卡取不到输入。"""
|
||||
cfg = node.config
|
||||
ref = cfg.get("caliber_from")
|
||||
out: list[ValidationIssue] = []
|
||||
@@ -1475,6 +1479,21 @@ def _caliber_from_issues(node: GraphNode, policies: set[str], labels: dict[str,
|
||||
or int(version) < 1:
|
||||
add(f"「{field_label('caliber_from')}」需要选择具体的版本号(1、2…),当前为「{version}」:不固定版本,"
|
||||
"口径会随上游变化", "caliber_from.workflow_version")
|
||||
if ref.get("nodes") not in (None, {}):
|
||||
from app.engine.nodes.metrics import rename_problem
|
||||
|
||||
if problem := rename_problem(ref.get("nodes")):
|
||||
add(f"「{field_label('caliber_from')}」的节点改名映射有误:{problem}", "caliber_from.nodes")
|
||||
elif spec is not None:
|
||||
nodes = spec.node_map()
|
||||
upstream = _ancestors(spec, node.id)
|
||||
for old, new in ref["nodes"].items():
|
||||
if new not in nodes:
|
||||
add(f"「{field_label('caliber_from')}」把取数节点「{old}」改名为「{new}」,但工作流中没有节点「{new}」",
|
||||
"caliber_from.nodes")
|
||||
elif new not in upstream:
|
||||
add(f"「{field_label('caliber_from')}」改名后的取数节点「{nodes[new].title}」不在口径卡的上游:口径卡计算时"
|
||||
"它还没有执行。请把它连到口径卡之前", "caliber_from.nodes")
|
||||
if cfg.get("metrics"):
|
||||
add(f"已设置「{field_label('caliber_from')}」,将按所引用的口径卡计算,此处填写的「{field_label('metrics')}」"
|
||||
"不会生效", "metrics", level="warning")
|
||||
|
||||
@@ -0,0 +1,247 @@
|
||||
"""口径卡钉住别处口径卡时的节点改名映射(caliber_from.nodes,next-round 02 做法 4 ③)。
|
||||
|
||||
那一版的口径卡读 cell(nodes.fetch, 0, 'gmv');拷进这张图时取数节点改名成 c1_fetch(组合式问数、对账、召回同一张图里放
|
||||
两份口径会撞 id),caliber_from 带 nodes={fetch: c1_fetch}。要一起认这个映射的几处:
|
||||
|
||||
- 运行时(metrics._pinned_caliber):式子按语法树把 nodes.fetch 的根换成 c1_fetch 再算,指标集的 source 照记映射;
|
||||
- 画布校验(schema._caliber_from_issues):映射的形状、改名后的节点在不在图里、在不在口径卡上游;
|
||||
- 升版检查(governance.unresolved_caliber_upgrades):照常挡、照常记事件,事件带上映射;
|
||||
- 证据(指标步骤的 source、输入链到改名后的那条查询)、推演(按改名后的式子原样重算);
|
||||
- 运行对比(compare):同一份口径换了节点名不算口径变了。
|
||||
|
||||
没写 nodes 的老图行为一字不差:式子原文、指标集、升版事件都和以前一样(单独测)。数据是虚构店铺库。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.core import artifact_store
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import RunEvent
|
||||
from app.engine.nodes.metrics import rename_expression, rename_problem, source_expression
|
||||
from app.engine.runner import run_manager
|
||||
from app.engine.schema import GraphSpec, validate_graph
|
||||
from app.main import app
|
||||
from tests import ledger_samples as S
|
||||
|
||||
RENDERED = ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' "
|
||||
"AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def engine_up(tmp_path):
|
||||
from app.ledger import match as ledger_match
|
||||
|
||||
await run_manager.setup()
|
||||
await S.fresh_source(tmp_path / "shop.db")
|
||||
yield
|
||||
await ledger_match.drain()
|
||||
await run_manager.shutdown()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
def composed(wf: str, *, rename: dict | None = None, fetch_id: str = "c1_fetch", card_id: str = "c1",
|
||||
version: int = 1, **card) -> dict:
|
||||
ref = {"workflow_id": wf, "workflow_version": version, "node_id": "card"}
|
||||
if rename is not None:
|
||||
ref["nodes"] = rename
|
||||
return S.chain(S.node("start", "input"), S.node(fetch_id, "tool", tool=S.tool(), args={"sql": RENDERED}),
|
||||
S.node(card_id, "metrics", caliber_from=ref, **card),
|
||||
S.node("out", "output", fields=[{"name": "r", "value": f"{{{{ nodes.{card_id}.text }}}}"}]))
|
||||
|
||||
|
||||
async def finished(run_id: str, node_id: str) -> RunEvent:
|
||||
async with SessionLocal() as session:
|
||||
return (await session.execute(select(RunEvent).where(
|
||||
RunEvent.run_id == run_id, RunEvent.type == "node.finished", RunEvent.node_id == node_id))).scalars().one()
|
||||
|
||||
|
||||
async def run(graph: dict):
|
||||
return await S.wait((await run_manager.start(graph=graph, input_payload={})).id)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 改写式子
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_renaming_rewrites_only_the_roots_of_node_chains():
|
||||
rename = {"fetch": "c1_fetch"}
|
||||
assert rename_expression("cell(nodes.fetch, 0, 'gmv')", rename) == "cell(nodes.c1_fetch, 0, 'gmv')"
|
||||
assert rename_expression("nodes.fetch.rows[0]['gmv'] / 2", rename) == "nodes.c1_fetch.rows[0]['gmv'] / 2"
|
||||
assert rename_expression("nodes['fetch'].row_count", rename) == "nodes['c1_fetch'].row_count"
|
||||
# 字符串里的字、别的根、别的节点不动
|
||||
assert rename_expression("cell(nodes.other, 0, 'nodes.fetch') + vars.fetch", rename) == \
|
||||
"cell(nodes.other, 0, 'nodes.fetch') + vars.fetch"
|
||||
# 一处都没换到的原样交回(不按 unparse 改写空白)
|
||||
assert rename_expression("cell(nodes.other,0,'gmv')", rename) == "cell(nodes.other,0,'gmv')"
|
||||
assert rename_expression("cell(nodes.fetch,", rename) == "cell(nodes.fetch,"
|
||||
# 反着换回那一版里的写法:对比口径按它比
|
||||
assert source_expression("cell(nodes.c1_fetch, 0, 'gmv')", {"nodes": rename}) == "cell(nodes.fetch, 0, 'gmv')"
|
||||
assert source_expression("cell(nodes.c1_fetch, 0, 'gmv')", {"workflow_id": "w"}) == "cell(nodes.c1_fetch, 0, 'gmv')"
|
||||
|
||||
|
||||
def test_mapping_shape_problems_are_named():
|
||||
assert rename_problem({"fetch": "c1_fetch"}) is None
|
||||
assert rename_problem([]) and rename_problem({})
|
||||
assert "节点 id" in rename_problem({"fetch": "c1 fetch"})
|
||||
assert "两个节点" in rename_problem({"a": "x", "b": "x"})
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 运行时、证据、推演
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_a_renamed_fetch_feeds_the_pinned_card(client):
|
||||
wf = await S.publish()
|
||||
row = await run(composed(wf, rename={"fetch": "c1_fetch"}))
|
||||
assert row.status == "succeeded", row.error
|
||||
event = await finished(row.id, "c1")
|
||||
card = artifact_store.load(event.data["artifact"])
|
||||
assert (card["caliber"], card["caliber_version"]) == ("周销售", "v1")
|
||||
[metric] = card["metrics"]
|
||||
assert metric["value"] == 170.5 and metric["rendered"] == "170.50元"
|
||||
# 卡上记的是这张图里实际算的式子;来源照记映射
|
||||
assert metric["expression"] == "cell(nodes.c1_fetch, 0, 'gmv')"
|
||||
assert card["source"] == {"workflow_id": wf, "workflow_version": 1, "node_id": "card",
|
||||
"nodes": {"fetch": "c1_fetch"}}
|
||||
[entry] = event.data["evidence"]
|
||||
assert entry["source"]["nodes"] == {"fetch": "c1_fetch"}
|
||||
[item] = metric["inputs"]
|
||||
assert item["via"] == "tool_cell" and item["node_id"] == "c1_fetch" and item["artifact"]
|
||||
|
||||
# 证据:指标步骤的来源带映射,输入链到改名后的那条查询的那一格
|
||||
resp = await client.get(f"/api/runs/{row.id}/evidence/metrics/c1/gmv")
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
[step] = [s for s in body["chain"] if s["step"] == "metric"]
|
||||
assert step["source"]["nodes"] == {"fetch": "c1_fetch"} and step["source"]["workflow_name"] == "周销售"
|
||||
[inp] = [s for s in body["chain"] if s["step"] == "input"]
|
||||
assert inp["node_id"] == "c1_fetch" and inp["artifact"] == item["artifact"]
|
||||
|
||||
# 推演:按改名后的式子原样重算,改一个输入照常算
|
||||
resp = await client.post(f"/api/runs/{row.id}/evidence/whatif",
|
||||
json={"card": "c1", "overrides": {"inputs": {"cell(nodes.c1_fetch, 0, 'gmv')": 200}}})
|
||||
assert resp.status_code == 200, resp.text
|
||||
[m] = resp.json()["metrics"]
|
||||
assert m["old"] == 170.5 and m["new"] == 200 and m["rendered_new"] == "200.00元"
|
||||
|
||||
|
||||
async def test_without_the_mapping_the_renamed_fetch_is_not_found():
|
||||
"""没写映射:式子照那一版读 nodes.fetch,这张图里没有它——缺输入,照实失败,不会算错。"""
|
||||
wf = await S.publish()
|
||||
row = await run(composed(wf))
|
||||
assert row.status == "failed" and "口径卡计算失败" in (row.error or "")
|
||||
|
||||
|
||||
async def test_old_graphs_without_the_mapping_are_unchanged():
|
||||
"""没写 nodes:式子原文一个字不改(不经 unparse),来源只有那三项。"""
|
||||
graph = S.caliber_graph(expression="cell(nodes.fetch,0,'gmv')")
|
||||
wf = await S.publish(graph=graph)
|
||||
pinned = {"workflow_id": wf, "workflow_version": 1, "node_id": "card"}
|
||||
row = await run(S.chain(S.node("start", "input"), S.node("fetch", "tool", tool=S.tool(), args={"sql": RENDERED}),
|
||||
S.node("mine", "metrics", caliber_from=pinned),
|
||||
S.node("out", "output", fields=[{"name": "r", "value": "{{ nodes.mine.text }}"}])))
|
||||
assert row.status == "succeeded", row.error
|
||||
[entry] = (await finished(row.id, "mine")).data["evidence"]
|
||||
card = artifact_store.load(entry["artifact"])
|
||||
assert card["source"] == pinned and entry["source"] == pinned
|
||||
assert [m["expression"] for m in card["metrics"]] == ["cell(nodes.fetch,0,'gmv')"]
|
||||
assert set(card) == {"kind", "caliber", "caliber_version", "metrics", "source"}
|
||||
|
||||
|
||||
async def test_a_malformed_mapping_is_refused_before_running():
|
||||
wf = await S.publish()
|
||||
with pytest.raises(ValueError, match="节点改名映射有误"):
|
||||
await run_manager.start(graph=composed(wf, rename={"fetch": "c1 fetch"}), input_payload={})
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 画布校验、升版检查
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _codes(graph: dict) -> list[tuple[str | None, str]]:
|
||||
return [(i.field, i.message) for i in validate_graph(GraphSpec.model_validate(graph)).issues if i.level == "error"]
|
||||
|
||||
|
||||
def test_validation_checks_the_mapping():
|
||||
good = composed("wf1", rename={"fetch": "c1_fetch"})
|
||||
assert not [f for f, _ in _codes(good) if (f or "").startswith("caliber_from")]
|
||||
shape = _codes(composed("wf1", rename={"fetch": ""}))
|
||||
assert any(f == "caliber_from.nodes" and "映射有误" in m for f, m in shape)
|
||||
missing = _codes(composed("wf1", rename={"fetch": "nowhere"}))
|
||||
assert any(f == "caliber_from.nodes" and "没有节点「nowhere」" in m for f, m in missing)
|
||||
# 改名后的节点在口径卡下游:算卡的时候它还没执行
|
||||
late = composed("wf1", rename={"fetch": "c1_fetch"})
|
||||
late["edges"] = [{"source": "start", "target": "c1"}, {"source": "c1", "target": "c1_fetch"},
|
||||
{"source": "c1_fetch", "target": "out"}]
|
||||
assert any(f == "caliber_from.nodes" and "不在口径卡的上游" in m for f, m in _codes(late))
|
||||
|
||||
|
||||
async def test_upgrade_checks_treat_renamed_cards_like_any_pinned_card():
|
||||
from app.engine.governance import unresolved_caliber_upgrades
|
||||
|
||||
wf = await S.publish()
|
||||
await S.publish(version=2, workflow_id=wf)
|
||||
async with SessionLocal() as session:
|
||||
errors, _ = await unresolved_caliber_upgrades(
|
||||
session, GraphSpec.model_validate(composed(wf, rename={"fetch": "c1_fetch"})))
|
||||
assert errors and "上游已有 v2" in errors[0]
|
||||
_, [event] = await unresolved_caliber_upgrades(session, GraphSpec.model_validate(
|
||||
composed(wf, rename={"fetch": "c1_fetch"}, upgrade_policy="dual")))
|
||||
assert event["caliber_nodes"] == {"fetch": "c1_fetch"} and event["pinned"] == 1 and event["latest"] == 2
|
||||
# 没改名的卡:事件和以前一样,没有这个键
|
||||
plain = composed(wf, fetch_id="fetch", card_id="mine", upgrade_policy="dual")
|
||||
_, [old] = await unresolved_caliber_upgrades(session, GraphSpec.model_validate(plain))
|
||||
assert "caliber_nodes" not in old
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 运行对比
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_compare_sees_the_same_caliber_under_different_node_names():
|
||||
from app.engine.compare import _caliber_reasons
|
||||
|
||||
def side(prefix: str) -> tuple[dict, dict]:
|
||||
metric = {"id": "gmv", "expression": f"cell(nodes.{prefix}fetch, 0, 'gmv')", "decimals": 2,
|
||||
"format": "thousands", "unit": "元"}
|
||||
card = {"caliber": "周销售", "caliber_version": "v1", "metrics": [metric],
|
||||
"source": {"workflow_id": "w", "workflow_version": 1, "node_id": "card",
|
||||
**({"nodes": {"fetch": f"{prefix}fetch"}} if prefix else {})}}
|
||||
return metric, card
|
||||
|
||||
(bm, bc), (tm, tc) = side(""), side("c1_")
|
||||
assert _caliber_reasons(bm, tm, bc, tc) == []
|
||||
# 映射反着换回来之后式子还是不同的,照旧说表达式改了
|
||||
tm2 = {**tm, "expression": "cell(nodes.c1_fetch, 0, 'gmv') * 2"}
|
||||
assert any("表达式由" in r for r in _caliber_reasons(bm, tm2, bc, tc))
|
||||
|
||||
|
||||
async def test_runs_of_the_same_composed_graph_compare_unchanged():
|
||||
from app.db.models import Workflow
|
||||
|
||||
wf = await S.publish()
|
||||
graph = composed(wf, rename={"fetch": "c1_fetch"})
|
||||
async with SessionLocal() as session:
|
||||
saved = Workflow(name="组合的答案", graph=graph)
|
||||
session.add(saved)
|
||||
await session.commit()
|
||||
saved_id = saved.id
|
||||
a, b = [await S.wait((await run_manager.start(graph=graph, workflow_id=saved_id, input_payload={})).id)
|
||||
for _ in range(2)]
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
resp = await c.get(f"/api/runs/{b.id}/compare", params={"base": a.id})
|
||||
assert resp.status_code == 200, resp.text
|
||||
[metric] = resp.json()["metrics"]
|
||||
assert metric["classification"] == "unchanged", metric
|
||||
Reference in New Issue
Block a user