mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
运行对比里子工作流的口径卡按卡所在那一层找产出:组装推演用的 Sources 时带上是哪张卡,节点产出、跑完的时刻按 engine/lineage.in_scope 换成卡那一层的 id(同推演 api/whatif._sources、证据面板 _InputMasks._output)。以前子图里的卡取值链写的是子图里的 id(nodes.parse),对比拿父图里同名的节点当它的产出,容器输入认到父图那次查询上;加一条子图卡的真跑对比测试
This commit is contained in:
@@ -184,9 +184,15 @@ async def _side(run: Run, sealed: _Sealed, masks: _Masks, kinds: dict[str, Any])
|
||||
for artifact, info in sealed.queries.items()}
|
||||
returned = _returned(sealed)
|
||||
|
||||
def sources(started: int) -> recompute.Sources:
|
||||
return recompute.Sources(load=loader, queries=tuple(sealed.queries), outputs=_outputs_before(sealed, started),
|
||||
masked=masked, returned=returned, finished=_finished_before(sealed, started))
|
||||
def sources(started: int, card_node: str = "") -> recompute.Sources:
|
||||
# 节点产出、跑完的时刻按口径卡所在那一层的 id 交进去(engine/lineage.in_scope,同推演 api/whatif._sources):子工作流
|
||||
# 里的卡(「子工作流节点 id/卡」)取值链写的是子图里的 id,只看同一张子图里的节点,父图里同名的不算
|
||||
from app.engine.lineage import in_scope
|
||||
|
||||
return recompute.Sources(load=loader, queries=tuple(sealed.queries),
|
||||
outputs=in_scope(_outputs_before(sealed, started), card_node),
|
||||
masked=masked, returned=returned,
|
||||
finished=in_scope(_finished_before(sealed, started), card_node))
|
||||
|
||||
return engine.Side(run=run_ref(run), graph=sealed.graph, events=sealed.events, cards=cards, load=loader,
|
||||
masked=masked, sources=sources, dialects=kinds, inputs=dict(run.input or {}))
|
||||
|
||||
@@ -183,8 +183,9 @@ class Side:
|
||||
没有的,就是读不出来或者被改过
|
||||
- load:工件 id → 内容,只认封存范围内的;文件不在返回 None,内容和哈希对不上抛 ValueError
|
||||
- masked:查询快照 → 要遮的列(小写):查询当时记下的加数据源现在设的
|
||||
- sources:口径卡那次执行开始的事件序号 → 推演重算还原输入要用的 Sources(节点产出、跑完的时刻都按那一刻)。
|
||||
不给就按 load、masked 拼一个最小的:nodes.* 这类容器输入还原不出来
|
||||
- sources:(口径卡那次执行开始的事件序号, 卡在这次运行里的节点 id) → 推演重算还原输入要用的 Sources(节点产出、跑完的
|
||||
时刻都按那一刻,按卡所在那一层的 id 记:子工作流里的卡只看同一张子图里的节点)。不给就按 load、masked 拼一个最小的:
|
||||
nodes.* 这类容器输入还原不出来
|
||||
- dialects:数据源名 → 类型(sqlite / postgres …),规范化 SQL 时按它的方言解析
|
||||
- inputs:运行输入(Run.input)
|
||||
"""
|
||||
@@ -195,7 +196,7 @@ class Side:
|
||||
cards: list[tuple[str, str, dict[str, Any]]] = field(default_factory=list)
|
||||
load: Callable[[str], Any] = lambda _: None
|
||||
masked: Mapping[str, frozenset[str]] = field(default_factory=dict)
|
||||
sources: Callable[[int], recompute.Sources] | None = None
|
||||
sources: Callable[[int, str], recompute.Sources] | None = None
|
||||
dialects: Mapping[str, str | None] = field(default_factory=dict)
|
||||
inputs: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@@ -454,7 +455,7 @@ def _plan_card(side: Side, card: _Card, queries: list[_Query]) -> None:
|
||||
"""按推演重算的规则还原卡上每个指标的输入:落在哪份快照上、是不是遮罩列、容器的完整值。"""
|
||||
if card.content is None:
|
||||
return
|
||||
sources = side.sources(card.started) if side.sources is not None else recompute.Sources(
|
||||
sources = side.sources(card.started, card.node_id) if side.sources is not None else recompute.Sources(
|
||||
load=side.load, queries=tuple(dict.fromkeys(q.artifact for q in queries)), masked=side.masked)
|
||||
store = recompute._Store(sources)
|
||||
card.sources = sources
|
||||
|
||||
+1
-1
@@ -187,7 +187,7 @@ class Fake:
|
||||
return artifact
|
||||
|
||||
def side(self):
|
||||
def sources(started):
|
||||
def sources(started, card_node=""):
|
||||
outputs = {}
|
||||
for seq, etype, nid, data in self.events:
|
||||
if seq < started and etype == "node.finished" and data.get("artifact"):
|
||||
|
||||
@@ -476,3 +476,79 @@ async def test_real_runs_split_data_and_parameter_changes(client, engine_up, ord
|
||||
body = await get(client, f"/api/runs/{third}/compare/base")
|
||||
assert (body["base_run_id"], body["reason"]) == (second, "same_workflow")
|
||||
json.dumps(body, allow_nan=False)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 子工作流里的口径卡:还原输入按卡所在那一层找产出
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def sub_card_graph():
|
||||
"""子工作流:取数 → 原样解析 → 口径卡(东区金额读单元格;区域数读解析出来的整份结果的行,要从解析节点当时的产出还原)。"""
|
||||
return chain(
|
||||
node("s", "input", "开始"),
|
||||
node("fetch", "tool", "区域取数", tool=f"db_query__{E2E_SOURCE}", args={"sql": E2E_SQL.replace(
|
||||
"{{ input.start }}", "2026-08-31").replace("{{ input.end }}", "2026-09-06")}),
|
||||
node("parse", "transform", "原样解析", mode="json", template="{{ nodes.fetch }}", assign_to="rows"),
|
||||
node("card", "metrics", "子口径卡", caliber="周报口径", caliber_version="v1", metrics=[
|
||||
{"id": "east", "name": "东区金额", "unit": "元", "decimals": 1, "format": "thousands",
|
||||
"expression": "cell(nodes.fetch, 0, 'gmv')"},
|
||||
{"id": "regions", "name": "区域数", "unit": "个", "decimals": 0, "format": "plain",
|
||||
"expression": "len(nodes.parse.rows)"}]),
|
||||
node("o", "output", "出口", fields=[{"name": "指标", "value": "{{ nodes.card.text }}"}]),
|
||||
)
|
||||
|
||||
|
||||
async def test_cards_inside_a_subworkflow_restore_inputs_from_their_own_layer(client, engine_up, orders):
|
||||
"""子工作流里的卡(这次运行里的 id「sub/card」)取值链写的是子图里的 id(nodes.parse):对比还原容器输入时要找「sub/parse」
|
||||
的产出,不是父图里的 parse。父图里另有同名的 fetch、parse(查的是另一条 SQL),按层找才不会取错。"""
|
||||
from tests import ledger_samples as S
|
||||
|
||||
path, source_id = orders
|
||||
sub_id = await S.publish("区域子流程", graph=sub_card_graph(), level="draft")
|
||||
workflow = await new_workflow("子流程周报")
|
||||
|
||||
def graph():
|
||||
return chain(
|
||||
node("start", "input", "开始"),
|
||||
# 父图里同名的取数、解析节点:一行,和子图里那份三行的结果不一样
|
||||
node("fetch", "tool", "父图取数", tool=f"db_query__{E2E_SOURCE}",
|
||||
args={"sql": "SELECT COUNT(*) AS n FROM orders"}),
|
||||
node("parse", "transform", "父图解析", mode="json", template="{{ nodes.fetch }}"),
|
||||
node("sub", "subgraph", "区域子流程", workflow_id=sub_id, workflow_version=1),
|
||||
node("out", "output", "成果", fields=[{"name": "指标", "value": "{{ nodes.sub.output.指标 }}"}]),
|
||||
)
|
||||
|
||||
async def finished() -> str:
|
||||
run = await run_manager.start(graph=graph(), input_payload={}, workflow_id=workflow, workflow_name="子流程周报")
|
||||
for _ in range(400):
|
||||
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:
|
||||
assert row.status == "succeeded", row.error
|
||||
return run.id
|
||||
raise AssertionError(f"运行 {run.id} 没有结束")
|
||||
|
||||
first = await finished()
|
||||
db = sqlite3.connect(path)
|
||||
db.execute("UPDATE orders SET amount = 1500.5 WHERE day = '2026-09-01'")
|
||||
db.commit()
|
||||
db.close()
|
||||
await engines.invalidate(source_id)
|
||||
second = await finished()
|
||||
|
||||
body = await get(client, f"/api/runs/{second}/compare?base={first}")
|
||||
metrics = {m["metric_id"] if "metric_id" in m else m["id"]: m for m in body["metrics"]}
|
||||
east, regions = metrics["east"], metrics["regions"]
|
||||
assert (east["classification"], east["base"], east["target"]) == ("data_only", 1200.5, 1500.5)
|
||||
# 区域数:从子图里那次取数的产出还原(三行),两次一样
|
||||
assert (regions["base"], regions["target"]) == (3, 3)
|
||||
assert regions["classification"] == "unchanged", regions
|
||||
problems = [i.get("problem") for i in regions.get("inputs") or [] if i.get("problem")]
|
||||
assert not problems, problems
|
||||
# 区域数的那份结果顺着取值链认到子图里那次取数(两次之间改了数,快照不同),不是父图里同名的 fetch(两次一样)
|
||||
[cell] = east["inputs"]
|
||||
[rows] = regions["inputs"]
|
||||
assert (rows["base_artifact"], rows["target_artifact"]) == (cell["base_artifact"], cell["target_artifact"])
|
||||
assert rows["base_artifact"] != rows["target_artifact"]
|
||||
|
||||
Reference in New Issue
Block a user