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

子工作流的证据交回父图（后端）：子图每一步进同一个运行、台账和口径卡交回、出口字段带出处、父图跨子图往下追

Browse Source 
调查时发现：run_subgraph 只设了节点 id 前缀（node_prefix）没用上，子图在父节点里用 ainvoke 跑，LangGraph 不把它的
custom 事件交给父图的流——子图里每一步在运行的事件里一条都没有，时间线看不到、封存也不含它查过的数。「事件本来就在同一个
运行里」此前并不成立（docs/design/orchestration-round/00-总方案.md §2）。

- engine/context + nodes/multi：子图拿到父运行的写入口（STREAM_KEY）和前缀（PREFIX_KEY），子图里的节点经它发事件，节点 id
  「子工作流节点 id/…」，嵌套的一层层接上；工件按这次运行里的 id 记。子图里报的错（含被驳回）交回时节点 id 换成这一层的、
  名字接上子工作流（runner 的 run.failed / run.cancelled 写「子工作流 › 那一步」），不再和父图里同名的节点混在一起
- nodes/multi：子图的台账条目加前缀并进父图 evidence（父图报告的目录据此收进子图里的查询和口径卡）；子图口径卡的产出以
  「子工作流节点 id/卡」进父图 nodes；出口每个字段按出口模板认出处（模板整串就是一个取值链、渲染出来正是那个值的才认，
  同口径卡认输入的那一套），出自子图运行输入的按子工作流节点的「输入」模板再往父图追一层，追不到的记 subgraph_input
- engine/compiler：子工作流节点自己的 node.finished 不再重记交回来的台账条目（子图那几步自己的 node.finished 已经带着）
- nodes/metrics：父图口径卡读 nodes.<子图>.output.<字段>（或 assign_to 变量的字段）时接上那个字段的出处，另记 subgraph、
  field；经子图出口转抄子图口径卡的指标照样带上不完整、存疑；cell(nodes.<子图>.output.<原样交出的查询结果>, …) 按台账
  认出子图里的那份快照
- engine/lineage：静态认法跨子图往下追（出口模板 → 子图里接着认；子图的运行输入按「输入」模板回父图认；最多 3 层），
  和运行时同一套；whole_ref 是两边共用的「整串一个取值链」判断。lint_process.load_subgraphs 往下读嵌套的子工作流，
  subgraph_spec 给静态认法取图；没读好放进来的照旧停在子工作流节点
- nodes/report：报告收证据的范围加上上游子工作流交回来的（evidence_from 写了子工作流节点的一并算）
- nodes/io：指标出具的缺口里，停在子图里的说停在那个子工作流（和静态检查同一句）
- api/evidence：卡上记的节点 id 按卡所在那一层认（_scoped：子图里的卡写子图里的 id），转抄、输入、追不到的那几步一路给到
  子图里的那一格；子图里的节点按 node.started 记的名字接成「子工作流 › 节点」（node_label、subgraph_label）；双盲复算条目
  里的卡 id 同样换成这次运行里的
- apps/resolve：实例进度不单列子图里的步骤
- 口径账本的 use_caliber 不改；比对读封存的查询，子图里的查询如今在封存范围里，照常认得出

测试：test_subgraph_evidence（子图每一步进运行、报错点名子图里那一步、实例进度、父图口径卡跨子图追到查询格并完整出具、
子图里追不到的父图降档、不是整串取值链的停在子工作流、运行输入往父图追、嵌套两层、静态检查与运行时一致（含两层和输入
模板）、父图报告引用子图里的查询和口径卡并一路点到那一格、封存与可验证报告、子图里的遮罩照父图的运行）。去掉「交写入口」
回跑，五条失败（快照不在封存范围里）。后端全量 14472 通过
This commit is contained in:
ThearchyHelios
2026-10-18 16:11:15 +08:00
parent d57cdcaffc
commit 3998e821dc
9 changed files with 782 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/evidence.py
+64 -16
View File
@@ -209,6 +209,9 @@ class _Sealed:
#: 按需裁判的判定（evidence.judged 事件），按先后：(seq, 节点, 载荷)。不管落在封存前后都收——
#: 失败后接着跑的运行会把早先追加的判定封进新的清单里，它仍然是按需裁判的批注
judged: list[tuple[int, str | None, dict[str, Any]]] = field(default_factory=list)
#: 节点 id → node.started 上记的名字、类型。子工作流里的节点（「子工作流节点 id/…」）不在这次运行的图上，名字从这里取
labels: dict[str, str] = field(default_factory=dict)
types: dict[str, str] = field(default_factory=dict)
@property
def sealed(self) -> bool:
@@ -273,6 +276,11 @@ async def _sealed(run_id: str) -> _Sealed:
out.retrievals.setdefault(str(data["artifact"]), {"node_id": node_id, "source": data.get("collection")})
elif etype == "caliber.upgrade":
out.upgrades[node_id] = data
elif etype == "node.started":
if isinstance(data.get("label"), str) and data["label"]:
out.labels.setdefault(node_id, data["label"])
if isinstance(data.get("node_type"), str) and data["node_type"]:
out.types.setdefault(node_id, data["node_type"])
if etype != "node.finished":
continue
if data.get("artifact"):
@@ -280,7 +288,9 @@ async def _sealed(run_id: str) -> _Sealed:
for entry in data.get("evidence") or []:
if isinstance(entry, dict) and entry.get("artifact"):
out.ledger.add((node_id, entry["artifact"]))
out.entries.append({**entry, "node_id": node_id})
# 条目里另记的节点 id（双盲复算复算的口径卡）是那个节点所在那一层图里的 id：换成这次运行里的
out.entries.append({**entry, "node_id": node_id,
**({"card": _scoped(node_id, entry["card"])} if entry.get("card") else {})})
artifact = str(entry["artifact"])
if entry.get("kind") == "query":
out.queries[artifact] = {
@@ -975,20 +985,21 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
# 期 4：这个指标做没做过双盲复算、结果如何。没有双盲复算复核过这份指标集时没有这个键，老运行一字不差
if checked := _dual_check_of(sealed, str(entry.get("node_id") or ""), artifact, metric_id, metric):
step["dual_check"] = checked
card_node = str(entry.get("node_id") or "")
if metric is not None and metric.get("inputs"):
# 输入追没追到原始数据（engine/lineage.py）：和「有出处」（确定性复算一致）是两件事，面板分开说。追不到的照实说
# 出处追到哪个节点为止；卡上没有输入（常量）的没有可追的，不给这个键
step["untraced"] = _untraced_steps(metric, sealed)
step["untraced"] = _untraced_steps(metric, sealed, card_node)
catalog = doc.get("catalog") or {}
inputs: list[dict[str, Any]] = []
#: 输入指到的查询快照 → 被引用的格；同一份快照只给一个查询步骤，格合在一起高亮
wanted: dict[str, list[tuple[Any, Any]]] = {}
held, word = await _metric_masks(metric or {}, str(entry.get("node_id") or ""), artifact, sealed, masks, card=card)
for item in (metric or {}).get("inputs") or []:
one = {"step": "input", **item}
one = {"step": "input", **item, **_scoped_ids(item, card_node, sealed)}
if item.get("through"):
# 穿过原样解析查询结果的整形节点按位置取的格：面板写明经过的是哪个节点（图里的名字）
one["through_label"] = _node_label(sealed, item["through"])
one["through_label"] = _node_label(sealed, one["through"])
artifact_of = item.get("artifact")
if artifact_of and item.get("via") in ("tool_cell", "agent_field"):
one["query"] = _query_alias(catalog, artifact_of)
@@ -999,7 +1010,8 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
one = _withheld(one, found)
inputs.append(one)
# 转抄另一张口径卡的指标（对账卡、召回卡的「口径」那一项）：跨卡往下追，接上那个指标自己的一段出处，一直到查询格子
upstream = await _upstream_steps(metric or {}, sealed, masks, catalog, wanted, {(artifact, str(metric_id))})
upstream = await _upstream_steps(metric or {}, sealed, masks, catalog, wanted, {(artifact, str(metric_id))},
card_node=card_node)
if withheld := {path: _held_word(found) for path, found in held.items() if found.kind != "table"}:
# 代入式里那几处同样写「已遮罩」（遮没遮核对不了的写「无法核对」）：不然标签里遮着、代入式里亮着，同一个值换个
# 地方照样看得到。整份结果当输入的不换：代入式里本来就只留着它的路径
@@ -1015,7 +1027,7 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Masks", catalog: dict[str, Any],
wanted: dict[str, list[tuple[Any, Any]]], seen: set[tuple[str, str]],
depth: int = 1) -> list[dict[str, Any]]:
depth: int = 1, *, card_node: str = "") -> list[dict[str, Any]]:
"""指标的输入里转抄另一张口径卡的指标的（nodes.<卡>.metrics[i].value，卡上记着 upstream），每个接一步 upstream：那张
卡的那个指标（名字、值、式子、代入式、核对状态，取值按同一套遮罩），它自己的输入（同指标步骤的输入；落在查询快照上的
并进 wanted，后面照常给查询步骤），追没追到原始数据。转抄的转抄一层层接着追（防环：同一个指标只接一次；最多
@@ -1033,7 +1045,8 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask
if (artifact, metric_id) in seen:
continue
seen.add((artifact, metric_id))
node = str(item.get("node_id") or "")
# 卡上记的是卡所在那一层图里的 id（子工作流里的卡转抄同一张子图里的卡；经子工作流出口转抄的已经是这一层的）
node = str(_scoped(card_node, item.get("node_id")) or "")
card, hash_ok = _load_card(artifact)
found = _metric_in(card, metric_id)
step: dict[str, Any] = {
@@ -1060,7 +1073,7 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask
if found.get("incomplete") else {}),
**({"sql_check_failed": True, "sql_check_reason": found.get("sql_check_reason") or ""}
if found.get("sql_check_failed") else {}),
"untraced": _untraced_steps(found, sealed),
"untraced": _untraced_steps(found, sealed, node),
})
if withheld := {path: _held_word(h) for path, h in held.items() if h.kind != "table"}:
step["substituted"] = recompute.substituted_without(found, withheld)
@@ -1068,7 +1081,7 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask
for sub in found.get("inputs") or []:
if not isinstance(sub, dict):
continue
one = {"step": "input", **sub}
one = {"step": "input", **sub, **_scoped_ids(sub, node, sealed)}
if sub.get("artifact") and sub.get("via") in SNAPSHOT_VIAS:
one["query"] = _query_alias(catalog, sub["artifact"])
if cell := _input_cell(sub, catalog):
@@ -1079,27 +1092,60 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask
inputs.append(one)
step["inputs"] = inputs
out.append(step)
out += await _upstream_steps(found, sealed, masks, catalog, wanted, seen, depth + 1)
out += await _upstream_steps(found, sealed, masks, catalog, wanted, seen, depth + 1, card_node=node)
return out
def _scoped(card_node: Any, node_id: Any) -> Any:
"""口径卡上记的节点 id（输入来自哪个节点、穿过的整形节点、转抄的卡）是卡所在那一层图里的 id：子工作流里的卡
（节点 id「子工作流节点 id/卡」）记的是子图里的 id，换成这次运行里的（接上同一个前缀）。最外层的卡原样；经子工作流
出口接上的出处（multi.run_subgraph 记的）已经是卡这一层的 id，同样接上卡的前缀就对了。"""
card = str(card_node or "")
if not isinstance(node_id, str) or not node_id or "/" not in card:
return node_id
return f"{card.rsplit('/', 1)[0]}/{node_id}"
def _scoped_ids(item: dict[str, Any], card_node: str, sealed: _Sealed) -> dict[str, Any]:
"""输入步骤上的节点 id 换成这次运行里的（_scoped）；落在子工作流里的另给图上的名字（node_label、subgraph_label：
「子工作流 › 节点」），画布、记录页认不出这个 id 时照它写。最外层的输入不多带键，和以前一字不差。"""
out: dict[str, Any] = {}
for key in ("node_id", "through", "subgraph"):
if isinstance(item.get(key), str) and item[key]:
out[key] = _scoped(card_node, item[key])
if "/" in str(out.get("node_id") or ""):
out["node_label"] = _node_label(sealed, out["node_id"])
if out.get("subgraph"):
out["subgraph_label"] = _node_label(sealed, out["subgraph"])
return out
def _node_label(sealed: _Sealed, node_id: Any) -> str:
"""封存的图里这个节点的名字，没有名字的写 id。"""
"""封存的图里这个节点的名字，没有名字的写 id。子工作流里的节点（「子工作流节点 id/…」）不在图上，按 node.started 记的
名字一层层接上：「子工作流 › 节点」。"""
node = next((n for n in (sealed.graph or {}).get("nodes") or [] if isinstance(n, dict) and n.get("id") == node_id), {})
return str((node.get("data") or {}).get("label") or node_id or "")
if node or not isinstance(node_id, str) or "/" not in node_id:
return str((node.get("data") or {}).get("label") or node_id or "")
parts = node_id.split("/")
names = [_node_label(sealed, parts[0])]
for i in range(2, len(parts) + 1):
here = "/".join(parts[:i])
names.append(sealed.labels.get(here) or parts[i - 1])
return " › ".join(names)
def _untraced_steps(metric: dict[str, Any], sealed: _Sealed) -> list[dict[str, Any]]:
def _untraced_steps(metric: dict[str, Any], sealed: _Sealed, card_node: str = "") -> list[dict[str, Any]]:
"""指标的输入里追不到原始数据的：[{path, node_id, label, type}]，label 是封存的图里那个节点的名字、type 是节点类型
（面板按它写「数据整形」「沙箱代码」）。认不出是哪个节点的，node_id、label 为 None。"""
（面板按它写「数据整形」「沙箱代码」）。认不出是哪个节点的，node_id、label 为 None。节点 id 按卡所在那一层认
（_scoped）。"""
nodes = {n.get("id"): n for n in (sealed.graph or {}).get("nodes") or [] if isinstance(n, dict)}
out = []
for item in untraced(metric.get("inputs") or []):
node_id = item.get("node_id")
node_id = _scoped(card_node, item.get("node_id"))
node = nodes.get(node_id) or {}
out.append({"path": item.get("path"), "node_id": node_id,
"label": _node_label(sealed, node_id) if node_id else None,
"type": node.get("type") or None})
"type": node.get("type") or sealed.types.get(str(node_id or "")) or None})
return out
@@ -1495,6 +1541,8 @@ class _InputMasks:
return None, True
def _output(self, node_id: Any) -> Any:
# 卡的取值链里写的是卡所在那一层图里的 id（子工作流里的卡写子图里的 id），换成这次运行里的再找
node_id = _scoped(self._card, node_id)
if self._given is not None:
return self._given(node_id)
if self._outputs is None:
backend/app/engine/lineage.py
+121 -7
View File
@@ -13,6 +13,16 @@
- 应用检查（apps/lint）、受管级发布检查（governance）：按 static_stops 静态地认（只有图、没有运行），追不到的给提示。
转抄同一张图里另一张口径卡的指标，和运行时一样跨卡按那个指标的式子接着认（最多 MAX_COPY_DEPTH 层）；被抄的卡按
caliber_from 固定在别处的版本时，静态看不到那一版的图，照旧停在那张卡。
读子工作流出口的字段（nodes.<子图>.output.<字段>，docs/design/orchestration-round/00-总方案.md §2）跨子图往下追，三处同一套：
- 运行时：子图跑完时按出口模板认每个字段的出处（multi.run_subgraph 的 provenance：模板整串就是一个取值链的字段，在子图
的状态里按口径卡认输入的同一套认；出自子图运行输入的再按子工作流节点的「输入」模板往父图追一层），父图口径卡读这个
字段时接上它（metrics._subgraph_field），卡上记的就是子图里那一格的出处（via、artifact、locator、status），另记
subgraph、field——traced / untraced 照常认；子图里追不到原始数据的，父图照样追不到、照样降档；
- 证据面板：卡上记的节点 id 按卡所在的那一层认（api/evidence 的 _scoped），一路给到子图里的那一格；
- 静态：按子工作流节点固定的那一版图（lint_process.subgraph_spec，调用方用 with_subgraphs 放进来）认出口字段的模板，接着
在子图里认；子图的运行输入按子工作流节点的「输入」模板回到父图认。最多 MAX_SUB_DEPTH 层（和运行时的嵌套上限一样），
看不到子图的（没放进来、读不到）照旧停在子工作流节点。
"""
from __future__ import annotations
@@ -140,20 +150,52 @@ class Stop:
node_id: str | None
#: 出口字段的模板整串就是一个取值链：{{ nodes.card.metrics[0].value }}（不带过滤器、不调函数、前后不带别的字）
_WHOLE_REF = re.compile(r"^\s*\{\{\s*((?:nodes|vars|input)\.[^{}|()]+?)\s*\}\}\s*$", re.S)
def whole_ref(template: Any) -> str | None:
"""模板整串就是一个取值链时交回它（成果字段的值就是那个值本身，渲染只是转成文字）；别的写法为 None：拼了别的字、
过了过滤器的值，对不上那一格的值。子工作流出口的出处，运行时（multi._output_provenance）和静态（_exit_template）
都按这一个判断认。"""
m = _WHOLE_REF.match(template) if isinstance(template, str) else None
return m.group(1).strip() if m else None
#: 静态地跨子工作流往下追的层数上限：和运行时子工作流的嵌套上限（multi._MAX_DEPTH）一样
MAX_SUB_DEPTH = 3
@dataclass(frozen=True)
class _Scope:
"""静态地认时所在的那一层图：节点 id → {type, title?, config}。是子工作流时记着父图那一层和它在父图里的节点 id（子图的
运行输入按那个子工作流节点的「输入」模板回到父图认）。"""
nodes: Mapping[str, Mapping[str, Any]]
parent: "_Scope | None" = None
via: str | None = None
level: int = 0
def static_stops(nodes: Mapping[str, Mapping[str, Any]], expression: str, *, _depth: int = 0) -> list[Stop]:
"""静态地认一个指标表达式里追不到原始数据的输入。nodes 是节点 id → {type, config}（apps/tree.node_map 的形状）。
和运行时同一套认法，只是不看值：cell() 的第一个参数是查询结果（调用工具、合并查询的产出，或原样解析它的整形节点）、
按位置取的格穿过原样解析的整形节点、运行输入、开了按出处核对字段的 Agent 交来的字段，算追到了；转抄同一张图里
另一张口径卡的指标，按那个指标的式子接着认（它的输入都追到了才算）；别的（表达式整形、沙箱代码、模型调用、转抄
固定在别处版本的口径卡……）停在写它的那个节点。表达式写错了的不在这里报（画布的校验报）。"""
另一张口径卡的指标，按那个指标的式子接着认（它的输入都追到了才算）；读子工作流出口的字段，按那一版子图的出口模板
接着认（_through_sub）；别的（表达式整形、沙箱代码、模型调用、转抄固定在别处版本的口径卡、看不到的子图……）停在
写它的那个节点。表达式写错了的不在这里报（画布的校验报）。"""
return _stops(_Scope(nodes), expression, _depth)
def _stops(scope: _Scope, expression: str, depth: int) -> list[Stop]:
try:
tree = parse_expression(expression)[0]
except ExpressionError:
return []
out: list[Stop] = []
for path in leaf_refs(tree):
stop = _static_stop(nodes, path, _depth)
stop = _static_stop(scope, path, depth)
if stop is not None:
out.append(stop)
return out
@@ -179,16 +221,17 @@ def _copied_metric(nodes: Mapping[str, Mapping[str, Any]], path: str) -> tuple[s
return writers[0], str(metrics[index]["expression"])
def _static_stop(nodes: Mapping[str, Mapping[str, Any]], path: str, depth: int = 0) -> Stop | None:
def _static_stop(scope: _Scope, path: str, depth: int = 0) -> Stop | None:
nodes = scope.nodes
type_of = lambda node_id: (nodes.get(node_id) or {}).get("type") # noqa: E731
if (copied := _copied_metric(nodes, path)) is not None:
card_id, expression = copied
if depth < MAX_COPY_DEPTH and not static_stops(nodes, expression, _depth=depth + 1):
if depth < MAX_COPY_DEPTH and not _stops(scope, expression, depth + 1):
return None
return Stop(path, card_id)
if (parts := cell_parts(path)) is not None:
writers = _writers(nodes, parts[0])
bad = [w for w in writers if not _table(nodes, w, type_of)]
bad = [w for w in writers if not _table(nodes, w, type_of) and not _sub_table(scope, w, parts[0])]
return Stop(path, bad[0]) if bad or not writers else None
if (at := row_cell(path)) is not None:
writers = _writers(nodes, at.table)
@@ -197,7 +240,7 @@ def _static_stop(nodes: Mapping[str, Mapping[str, Any]], path: str, depth: int =
return None
head = _REF_HEAD.match(path)
if head and head.group(1) == "input":
return None
return None if _from_parent(scope, head.group(2), depth) else Stop(path, scope.via)
writers = _writers(nodes, path)
if not writers:
return Stop(path, None)
@@ -205,13 +248,84 @@ def _static_stop(nodes: Mapping[str, Mapping[str, Any]], path: str, depth: int =
for writer in writers:
node = nodes[writer]
if node.get("type") == "input":
if head and head.group(1) == "vars" and not _from_parent(scope, head.group(2), depth):
return Stop(path, scope.via)
continue
if node.get("type") == "agent" and _cited(node, head.group(1) if head else "", field):
continue
if node.get("type") == "subgraph" and head and _through_sub(scope, writer, head, depth):
continue
return Stop(path, writer)
return None
def _from_parent(scope: _Scope, name: str, depth: int) -> bool:
"""运行输入的一个字段算不算原始数据：最外层的是（发起运行的人填的）；子工作流的运行输入是父图按子工作流节点的「输入」
模板渲染出来的，那个模板整串就是一个取值链、在父图里追得到原始数据，才算（同运行时 multi._lifted）。"""
if scope.parent is None:
return True
templates = ((scope.parent.nodes.get(scope.via or "") or {}).get("config") or {}).get("input")
path = whole_ref(templates.get(name)) if isinstance(templates, dict) and name else None
return path is not None and _static_stop(scope.parent, path, depth + 1) is None
def _sub_scope(scope: _Scope, node_id: str) -> _Scope | None:
"""子工作流节点 node_id 引用的那一版图（lint_process.subgraph_spec 读好放进来的），看不到、超过层数上限的为 None。"""
if scope.level >= MAX_SUB_DEPTH:
return None
from app.engine.lint_process import subgraph_spec
sub = subgraph_spec((scope.nodes.get(node_id) or {}).get("config") or {})
if sub is None:
return None
nodes = {n.id: {"type": str(n.type), "title": n.title, "config": n.config} for n in sub.nodes}
return _Scope(nodes, parent=scope, via=node_id, level=scope.level + 1)
def _exit_template(sub: _Scope, key: str) -> str | None:
"""子图出口字段 key 的模板整串所写的取值链（whole_ref）；没有这个字段、几个成果节点写法不一样、不是整串一个
取值链的为 None。"""
found = {whole_ref(f.get("value", "")) for node in sub.nodes.values() if node.get("type") == "output"
for f in (node.get("config") or {}).get("fields") or [] if isinstance(f, dict) and f.get("name") == key}
return found.pop() if len(found) == 1 else None
def _sub_key(head: re.Match[str], node: Mapping[str, Any]) -> str | None:
"""nodes.<子图>.output.<字段> / vars.<子图的 assign_to>.<字段> 读的出口字段名；不是正好一个字段的为 None。"""
root, name, rest = head.group(1), head.group(2), head.group(3)
key = re.sub(r"\[\s*['\"]([^'\"\]]+)['\"]\s*\]", r".\1", rest).strip(".")
if root == "nodes":
if not key.startswith("output."):
return None
key = key[len("output."):]
elif root != "vars" or str((node.get("config") or {}).get("assign_to") or "").strip() != name:
return None
return key if key and "." not in key and "[" not in key else None
def _through_sub(scope: _Scope, writer: str, head: re.Match[str], depth: int) -> bool:
"""读子工作流出口的一个字段：那个字段在子图里追得到原始数据（出口模板整串就是一个取值链，接着在子图里认）。"""
key = _sub_key(head, scope.nodes.get(writer) or {})
sub = _sub_scope(scope, writer) if key else None
inner = _exit_template(sub, key) if sub is not None and key else None
return inner is not None and _static_stop(sub, inner, depth + 1) is None
def _sub_table(scope: _Scope, writer: str, source: str) -> bool:
"""cell() 的第一个参数是子工作流出口原样交出的一份查询结果（出口字段的模板整串就是 {{ nodes.<查询节点> }}）。"""
head = _REF_HEAD.match(source)
if not head or (scope.nodes.get(writer) or {}).get("type") != "subgraph":
return False
key = _sub_key(head, scope.nodes.get(writer) or {})
sub = _sub_scope(scope, writer) if key else None
inner = _exit_template(sub, key) if sub is not None and key else None
if inner is None:
return False
writers = _writers(sub.nodes, inner)
type_of = lambda node_id: (sub.nodes.get(node_id) or {}).get("type") # noqa: E731
return bool(writers) and all(_table(sub.nodes, w, type_of) or _sub_table(sub, w, inner) for w in writers)
def _writers(nodes: Mapping[str, Mapping[str, Any]], source: str) -> list[str]:
"""取值链的根是谁写的：nodes.X 是 X；vars.X 是 assign_to 为 X 的节点（入口节点声明了字段 X 的也算）。"""
head = _REF_HEAD.match(source)
backend/app/engine/lint_process.py
+41 -23
View File
@@ -147,9 +147,17 @@ class Subgraphs(dict):
self.pin_problems: list[Any] = []
#: 嵌套的子工作流往下读几层：和运行时的嵌套上限（multi._MAX_DEPTH）一样，再深的运行时本来就拒
_LOAD_DEPTH = 3
async def load_subgraphs(session: Any, spec: GraphSpec) -> Subgraphs:
"""读出图里子工作流节点引用的图（固定了版本的读那一版，没固定的读工作流现在的图）。读不到、不成形的跳过。
固定的版本能不能用也一并读好（Subgraphs.pin_problems）。"""
"""读出图里子工作流节点引用的图（固定了版本的读那一版，没固定的读工作流现在的图），子图里再嵌套的一层层往下读（最多
_LOAD_DEPTH 层，同一张图只读一次，互相引用的不会转圈）。读不到、不成形的跳过。固定的版本能不能用也一并读好
（Subgraphs.pin_problems，只看这张图自己的子工作流节点）。
拿去用的：人工节点在子工作流里（_subgraph_humans）、口径卡的输入跨子工作流追到原始数据（engine/lineage 的静态认法，
subgraph_spec）。"""
from sqlalchemy import select
from app.apps.versions import subgraph_pin_problems
@@ -157,24 +165,39 @@ async def load_subgraphs(session: Any, spec: GraphSpec) -> Subgraphs:
out = Subgraphs()
out.pin_problems = await subgraph_pin_problems(session, spec)
for node in spec.nodes:
if node.type != NodeType.SUBGRAPH:
continue
wid, pinned = node.config.get("workflow_id"), node.config.get("workflow_version")
version = int(pinned) if str(pinned or "").isdigit() else None
if not wid or (str(wid), version) in out:
continue
query = (select(WorkflowVersion.graph).where(WorkflowVersion.workflow_id == str(wid),
WorkflowVersion.version == version)
if version is not None else select(Workflow.graph).where(Workflow.id == str(wid)))
graph = (await session.execute(query)).scalar_one_or_none()
try:
out[(str(wid), version)] = GraphSpec.model_validate(graph)
except Exception: # noqa: BLE001 - 读不到、不成形：子工作流节点运行时自己会说清楚
continue
todo = [(spec, 0)]
while todo:
current, level = todo.pop(0)
for node in current.nodes:
if node.type != NodeType.SUBGRAPH:
continue
wid, pinned = node.config.get("workflow_id"), node.config.get("workflow_version")
version = int(pinned) if str(pinned or "").isdigit() else None
if not wid or (str(wid), version) in out:
continue
query = (select(WorkflowVersion.graph).where(WorkflowVersion.workflow_id == str(wid),
WorkflowVersion.version == version)
if version is not None else select(Workflow.graph).where(Workflow.id == str(wid)))
graph = (await session.execute(query)).scalar_one_or_none()
try:
out[(str(wid), version)] = GraphSpec.model_validate(graph)
except Exception: # noqa: BLE001 - 读不到、不成形：子工作流节点运行时自己会说清楚
continue
if level + 1 < _LOAD_DEPTH:
todo.append((out[(str(wid), version)], level + 1))
return out
def subgraph_spec(config: Mapping[str, Any]) -> GraphSpec | None:
"""子工作流节点（配置 config）引用的那一版图：load_subgraphs 读好、with_subgraphs 放进来的；没放进来、没读到的为 None。"""
graphs = _SUBGRAPHS.get()
if graphs is None:
return None
wid, pinned = config.get("workflow_id"), config.get("workflow_version")
version = int(pinned) if str(pinned or "").isdigit() else None
return graphs.get((str(wid or ""), version))
@contextlib.contextmanager
def with_subgraphs(graphs: Mapping[tuple[str, int | None], GraphSpec]) -> Iterator[None]:
token = _SUBGRAPHS.set(graphs)
@@ -222,12 +245,7 @@ def _agent_tool_sets(node: GraphNode) -> list[tuple[str, Mapping[str, Any], list
def _subgraph_humans(spec: GraphSpec, node: GraphNode) -> list[str] | None:
graphs = _SUBGRAPHS.get()
if graphs is None:
return None
wid, pinned = node.config.get("workflow_id"), node.config.get("workflow_version")
version = int(pinned) if str(pinned or "").isdigit() else None
sub = graphs.get((str(wid or ""), version))
sub = subgraph_spec(node.config)
if sub is None:
return None
return [n.title for n in sub.nodes if n.type == NodeType.HUMAN]
backend/app/engine/nodes/io.py
+4 -2
View File
@@ -351,8 +351,10 @@ def _untraced_gaps(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[str]
stops = untraced(m.get("inputs") or [])
if not stops:
continue
where = "、".join(dict.fromkeys(f"{s.get('path')}（停在「{titles.get(s.get('node_id') or '', s.get('node_id') or '未知来源')}」）"
for s in stops[:3]))
# 停在子工作流里的（节点 id「子工作流节点 id/…」）说停在那个子工作流：这张图上看得到的是它，静态检查也这么说
where = "、".join(dict.fromkeys(
f"{s.get('path')}（停在「{titles.get(at, at or '未知来源')}」）"
for s in stops[:3] for at in [str(s.get('node_id') or '').split('/', 1)[0]]))
out.append(f"指标「{m.get('name') or m.get('id')}」的输入没追到原始数据：{where}")
return out
backend/app/engine/nodes/metrics.py
+41 -1
View File
@@ -624,6 +624,9 @@ def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext,
- vars.q.rows[0][0]，q 由「只是把调用工具节点的结果原样按 JSON 解析」的整形节点写：同 cell()，via tool_cell，
另记 through（穿过的整形节点）
- agent 开了 cite_fields 的字段：via agent_field，接上它的 data_evidence（eid、核对状态）
读子工作流出口的一个字段（nodes.<子图>.output.<字段>、vars.<子图的 assign_to>.<字段>）：接上子图记下的那个字段的出处
（_subgraph_field），via 等照子图里认出的写，另记 subgraph、field；认不出的照旧记 via subgraph，停在子工作流。
"""
if (parts := cell_parts(path)) is not None:
return _cell_input(path, value, parts, state, ctx, tctx or template_context(state))
@@ -644,6 +647,9 @@ def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext,
else:
node_id = _writer_of(name, state, ctx)
producer = nodes.get(node_id) if node_id else None
if producer is not None and str(producer.type) == "subgraph" and head \
and (linked := _subgraph_field(path, value, root, field, producer.id, state)) is not None:
return linked
entry: dict[str, Any] = {
"path": path,
"value": value if value is None or isinstance(value, (int, float, str, bool)) else _brief(value),
@@ -667,6 +673,32 @@ def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext,
return entry
def _subgraph_field(path: str, value: Any, root: str, field: str, node_id: str,
state: GraphState) -> dict[str, Any] | None:
"""读子工作流出口的一个字段：接上子图记下的那个字段的出处（multi.run_subgraph 的 provenance，节点 id 已经是这一层图里
的「子工作流节点 id/…」）。只认整个字段（nodes.<子图>.output.<字段>、vars.<assign_to>.<字段>，不再往里取），而且这次
读到的值就是子图交出的那个；别的为 None，照一般的来历记（via subgraph，停在子工作流）。
接上的出处照子图里认出的写（via、node_id、artifact、locator、eid、status、upstream），path、value 换成这里读的，另记
subgraph（经过的子工作流节点）、field（出口字段）：engine/lineage.traced 照常认，证据面板照常一路给到子图里的那一格。"""
result = (state.get("nodes") or {}).get(node_id)
if not isinstance(result, dict) or not isinstance(result.get("provenance"), dict):
return None
key = _field_key(field)
if root == "nodes":
if not key.startswith("output."):
return None
key = key[len("output."):]
elif root != "vars":
return None
output = result.get("output") if isinstance(result.get("output"), dict) else {}
item = result["provenance"].get(key)
if not key or not isinstance(item, dict) or key not in output or output[key] != value:
return None
return {**item, "path": path, "value": value if value is None or isinstance(value, (int, float, str, bool))
else _brief(value), "subgraph": node_id, "field": key}
def _copied_card(path: str, tctx: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]] | None:
"""转抄另一张口径卡一个指标的输入（engine/lineage.metric_copy 认的写法）：(那张卡这一次的产出, 那个指标)。不是这种
写法、产出不是指标集、下标越界的为 None。"""
@@ -702,9 +734,17 @@ def _upstream(path: str, tctx: dict[str, Any]) -> dict[str, Any] | None:
def _copied_metric(item: dict[str, Any], tctx: dict[str, Any]) -> dict[str, Any] | None:
"""卡上的一个输入转抄了另一张口径卡的哪个指标（这一次的产出里那一项）；不是转抄的为 None。"""
"""卡上的一个输入转抄了另一张口径卡的哪个指标（这一次的产出里那一项）；不是转抄的为 None。
经子工作流出口转抄的（subgraph，子图出口字段就是子图里一张卡的指标）：那张卡以「子工作流节点 id/卡」进了这一层的
nodes（multi.run_subgraph），按 node_id 和 upstream 记的指标 id 找。"""
if item.get("via") != "metric" or not isinstance(item.get("upstream"), dict):
return None
if item.get("subgraph"):
card = (tctx.get("nodes") or {}).get(str(item.get("node_id") or ""))
metrics = card.get("metrics") if isinstance(card, dict) and card.get("kind") == "metric_set" else None
wanted = item["upstream"].get("metric")
return next((m for m in metrics or [] if isinstance(m, dict) and m.get("id") == wanted), None)
found = _copied_card(str(item.get("path") or ""), tctx)
return found[1] if found is not None else None
backend/app/engine/nodes/multi.py
+105 -6
View File
@@ -14,7 +14,8 @@ from app.core.errors import describe_exception, raw_detail
from app.core.events import EventType
from app.db.base import SessionLocal
from app.db.models import Workflow
from app.engine.context import NodeContext, NodeError
from app.engine.context import PREFIX_KEY, STREAM_KEY, NodeContext, NodeError, stream_writer
from app.engine.lineage import whole_ref
from app.engine.state import GraphState, message_text, template_context
from app.engine.toolcalls import (
TOOL_MARKUP_NUDGE,
@@ -616,8 +617,17 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
"""嵌套执行另一个工作流。
子图共享同一个 run 的事件流（节点 id 会加前缀），所以在时间线上能看到它内部
每一步；但它有独立的状态，不会污染父图的变量池。
子图共享同一个 run 的事件流：节点 id 加前缀「子工作流节点 id/」（嵌套的一层层接上），事件经父运行交下来的写入口发
（engine/context.STREAM_KEY；以前只设了前缀没用上，子图里每一步在运行的事件里一条都没有），时间线上看得到它内部每一步，
封存也就覆盖了它查过的数。它有独立的状态，不会污染父图的变量池。
证据交回父图（docs/design/orchestration-round/00-总方案.md §2），三样：
- 台账：子图记下的条目（查询快照、口径卡的指标集……）节点 id 加上前缀并进父图的 evidence，父图报告撰写的目录据此收进
子图里的查询和口径卡（report.report_catalog），cell(nodes.<子图>.output.<原样交出的查询结果>, …) 也按台账认得出快照；
- 口径卡：子图里口径卡的产出以「子工作流节点 id/卡」为 id 进父图的 nodes（目录要卡上的指标值；父图转抄它的指标时认得
出那张卡、带上不完整和存疑）；
- 出处：出口的每个字段按出口模板认它来自哪里（_output_provenance），记在产出的 provenance 上；父图里读
nodes.<子图>.output.<字段> 的口径卡据此接上子图里的那一格（metrics._subgraph_field），engine/lineage 同一套认法。
"""
from app.engine.compiler import compile_graph # 延迟导入，避免循环依赖
from app.engine.schema import GraphSpec, loop_steps
@@ -678,9 +688,10 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
approval_default=ctx.run.approval_default,
tool_trust=ctx.run.tool_trust,
agent_limits=ctx.run.agent_limits,
# 恢复重放的标记和签批人只属于父图的节点，不能漏进子图（节点 id 可能重名）
# 恢复重放的标记和签批人只属于父图的节点，不能漏进子图（节点 id 可能重名）。前缀和写入口见 engine/context：
# 子图里的节点经父运行的写入口发事件、节点 id 带前缀；嵌套的子图拿到的是同一个写入口、更长的前缀
extra={**{k: v for k, v in ctx.run.extra.items() if k not in ("resumed", "actors")},
"node_prefix": f"{ctx.node.id}/"},
PREFIX_KEY: f"{ctx.run.global_id(ctx.node.id)}/", STREAM_KEY: stream_writer(ctx.run)},
)
ctx.emit(EventType.LOG, level="info",
@@ -711,6 +722,13 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
f"子工作流「{workflow.name}」已执行 {limit} 步仍未结束，已中止。可能存在没有轮数上限的环路，"
"请在子工作流中用「循环」节点限制轮数",
) from e
except NodeError as e:
# 子图里的节点报的错（含被驳回）：节点 id 换成这一层图里的 id（「子工作流节点 id/…」），名字接上子工作流的名字。
# 以前带的是子图里的 id，运行记下的出错节点会和父图里同名的节点混在一起
titles = sub_spec.node_map()
inner = getattr(e, "label", None) or (titles[e.node_id].title if e.node_id in titles else e.node_id)
e.node_id, e.label = f"{ctx.node.id}/{e.node_id}", f"{ctx.node.title} › {inner}"
raise
output = result_state.get("output") or {}
result = {
@@ -721,13 +739,94 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
"pinned": bool(pinned),
"text": json.dumps(output, ensure_ascii=False) if output else "",
}
if provenance := _output_provenance(sub_spec, sub_run, result_state, output, state, ctx):
# 出口每个字段的出处（没有能认的字段时不带这个键，产出和以前一字不差）
result["provenance"] = provenance
ctx.emit(EventType.LOG, level="info", message=f"子工作流「{workflow.name}」完成")
sub_nodes = result_state.get("nodes") or {}
cards = {f"{ctx.node.id}/{nid}": payload for nid, payload in sub_nodes.items()
if isinstance(payload, dict) and payload.get("kind") == "metric_set"}
updates: dict[str, Any] = {
"nodes": {ctx.node.id: result},
"nodes": {ctx.node.id: result, **cards},
"usage": result_state.get("usage") or {},
}
if ledger := [_forwarded(e, ctx.node.id) for e in result_state.get("evidence") or [] if isinstance(e, dict)]:
updates["evidence"] = ledger
var_name = ctx.cfg("assign_to", "")
if var_name:
updates["vars"] = {var_name: output}
return updates
#: 台账条目里写着节点 id 的键：交回父图时一并加前缀（双盲复算的条目记着它复算的口径卡）
_LEDGER_NODE_KEYS = ("node_id", "card")
def _forwarded(entry: dict[str, Any], node_id: str) -> dict[str, Any]:
"""子图台账里的一条交回父图：节点 id 加上「子工作流节点 id/」前缀，别的照抄。"""
out = dict(entry)
for key in _LEDGER_NODE_KEYS:
if isinstance(out.get(key), str) and out[key]:
out[key] = f"{node_id}/{out[key]}"
return out
def _output_provenance(sub_spec: Any, sub_run: Any, sub_state: dict[str, Any], output: dict[str, Any],
state: GraphState, ctx: NodeContext) -> dict[str, dict[str, Any]]:
"""子图出口每个字段的出处：{字段: 口径卡上一个输入的样子（via、node_id、artifact、locator、status、upstream……）}。
认法和口径卡认它的输入同一个（metrics._input_of），在子图的状态里认：字段的模板整串就是一个取值链（whole_ref），而且
渲染出来的值正是那个值转成的文字，才算这个字段就是那个值；拼了别的字、过了过滤器的不认（不在结果里）。几个成果节点
写同一个字段时按最后跑的那个。认出来的节点 id 换成父图这一层的（加「子工作流节点 id/」前缀）。
出自子图运行输入的（via input）再往父图追一层：子图的输入是父图按子工作流节点的「输入」模板渲染出来的，那个模板整串
也是一个取值链时，按父图的状态接着认（追到父图的运行输入、查询格……）；否则记 via subgraph_input，停在子工作流的
入口——子图的运行输入对父图不是原始数据。"""
from app.engine.expressions import _stringify
from app.engine.nodes.metrics import _input_of, _value_of
from app.engine.schema import NodeType
outputs = {n.id: n for n in sub_spec.nodes if n.type == NodeType.OUTPUT}
fields: dict[str, tuple[Any, str]] = {}
for step in sub_state.get("trail") or []:
exit_node = outputs.get(step.get("node_id")) if "error" not in step else None
for field in (exit_node.config.get("fields") or []) if exit_node is not None else []:
if isinstance(field, dict) and field.get("name"):
fields[str(field["name"])] = (exit_node, field.get("value", ""))
tctx = template_context(sub_state)
out: dict[str, dict[str, Any]] = {}
for name, (exit_node, template) in fields.items():
path = whole_ref(template)
if path is None or name not in output:
continue
value = _value_of(path, tctx)
if _stringify(value) != output[name]:
continue
item = _input_of(path, value, sub_state, NodeContext(node=exit_node, run=sub_run), tctx)
out[name] = _lifted(item, state, ctx)
return out
def _lifted(item: dict[str, Any], state: GraphState, ctx: NodeContext) -> dict[str, Any]:
"""子图里认出的一个出处换到父图这一层：节点 id 加前缀；出自子图运行输入的，按子工作流节点的「输入」模板往父图追。"""
from app.engine.expressions import _stringify
from app.engine.nodes.metrics import _REF_HEAD, _input_of, _value_of
if item.get("via") == "input":
head = _REF_HEAD.match(str(item.get("path") or ""))
name = head.group(2) if head else ""
templates = ctx.cfg("input", {}) or {}
path = whole_ref(templates.get(name)) if isinstance(templates, dict) and name else None
if path is not None:
parent = template_context(state)
value = _value_of(path, parent)
if _stringify(value) == _stringify(item.get("value")):
return _input_of(path, value, state, ctx, parent)
return {"path": item.get("path"), "value": item.get("value"), "node_id": ctx.node.id, "via": "subgraph_input",
**({"field": name} if name else {}), "status": item.get("status") or "ok"}
out = dict(item)
for key in ("node_id", "through", "subgraph"):
if isinstance(out.get(key), str) and out[key]:
out[key] = f"{ctx.node.id}/{out[key]}"
return out
backend/app/engine/nodes/report.py
+21 -4
View File
@@ -37,6 +37,7 @@ import asyncio
import re
import time
from collections import Counter
from collections.abc import Iterable
from typing import Any
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
@@ -140,16 +141,18 @@ def report_catalog(state: GraphState, spec: GraphSpec, node: GraphNode) -> dict[
if isinstance(sources, str) and sources != "ancestors":
sources = [sources]
if isinstance(sources, list):
# evidence_from 只管查询、检索这类证据（连同查询当时的表结构）从哪几个节点来；指标从哪来由 metrics_from 管
# evidence_from 只管查询、检索这类证据（连同查询当时的表结构）从哪几个节点来；指标从哪来由 metrics_from 管。
# 写了子工作流节点的，它交回来的（子图里的查询，节点 id「子工作流节点 id/…」）一并算
keep = {str(s) for s in sources}
ledger = [e for e in ledger
if e.get("kind") not in ("query", "retrieval", "schema") or e.get("node_id") in keep]
if e.get("kind") not in ("query", "retrieval", "schema") or _outer(e.get("node_id")) in keep]
nodes = state.get("nodes") or {}
catalog = build_catalog(
nodes=state.get("nodes") or {},
nodes=nodes,
ledger=ledger,
inputs=state.get("input") or {},
metrics_from=_setting(spec, node, "metrics_from") or None,
allowed=_ancestors(spec, node.id),
allowed=with_subgraphs(_ancestors(spec, node.id), [*nodes, *(e.get("node_id") for e in ledger)]),
)
if report_entities(spec, node) == "off":
# 目录里没有表和字段，写作目录也就不列；compose / verify 另传 entities=False，解析不了的原因照实说
@@ -161,6 +164,20 @@ def report_catalog(state: GraphState, spec: GraphSpec, node: GraphNode) -> dict[
return catalog
def _outer(node_id: Any) -> str:
"""节点 id 在这一层图上是哪个节点：子工作流交回来的「子工作流节点 id/…」是那个子工作流节点，别的原样。"""
return str(node_id or "").split("/", 1)[0]
def with_subgraphs(allowed: set[str], ids: Iterable[Any]) -> set[str]:
"""报告收证据的范围（祖先节点）加上范围里的子工作流交回来的那些（multi.run_subgraph：台账条目、口径卡的节点 id 是
「子工作流节点 id/…」，嵌套的一层层接上）：子工作流在报告上游，它里面查过的数、算过的口径卡也在上游
（docs/design/orchestration-round/00-总方案.md §2）。"""
out = set(allowed)
out.update(i for i in ids if isinstance(i, str) and "/" in i and _outer(i) in allowed)
return out
def _masked_word() -> str:
"""「已遮罩」：证据面板、推演同一个说法。api.evidence 导入了引擎里的模块，用到时再导入。"""
from app.api.evidence import MASKED
backend/app/engine/runner.py
+10 -2
View File
@@ -1360,7 +1360,7 @@ def rejected_data(exc: NodeRejected, spec: GraphSpec) -> dict[str, Any]:
"""被驳回的 run.cancelled 带什么：原因（rejected）、在哪一步（节点 id 和图上的名字）、哪种关卡、结论（驳回 / 超时
驳回）、谁、备注。进封存：谁在哪一步驳回、为什么，事后查得到。"""
node = spec.node_map().get(exc.node_id)
return {"reason": END_REJECTED, "node_id": exc.node_id, "label": node.title if node else exc.node_id,
return {"reason": END_REJECTED, "node_id": exc.node_id, "label": _label_of(exc, node),
"kind": exc.kind, "decision": exc.decision, "actor": exc.actor, "note": exc.note}
@@ -1390,10 +1390,18 @@ def _failure(exc: BaseException, spec: GraphSpec) -> tuple[str, tuple[EventType,
data["detail"] = detail
if node_id:
node = spec.node_map().get(node_id)
data.update(node_id=node_id, label=node.title if node else node_id)
data.update(node_id=node_id, label=_label_of(exc, node))
return message, (EventType.RUN_FAILED, data)
def _label_of(exc: BaseException, node: Any) -> str:
"""出错、被驳回的那一步在图上叫什么。子工作流里的那一步（节点 id「子工作流节点 id/…」）不在这张图上，名字由
multi.run_subgraph 接好记在异常上（「子工作流 › 那一步」）。"""
if node is not None:
return node.title
return str(getattr(exc, "label", None) or getattr(exc, "node_id", "") or "")
async def _approval_default(*, run_class: str) -> str:
"""工具审批策略的全局默认，取自设置里的「危险工具默认需要人工确认」。
backend/tests/test_subgraph_evidence.py
+375
View File
@@ -0,0 +1,375 @@
"""子工作流的证据交回父图（docs/design/orchestration-round/00-总方案.md §2）。
以前子图只把出口的值交回父图：子图里的查询快照、口径卡留在子图自己的状态里，子图里每一步在运行的事件里一条都没有（只设了
节点 id 前缀没用上，LangGraph 的 ainvoke 不把子图的 custom 事件交给父图的流），父图报告引用不到子图里的数，父图口径卡读
子图出口的字段一律停在子工作流节点。现在：
- 子图里每一步进同一个运行的事件流，节点 id「子工作流节点 id/…」，封存覆盖它们；
- 子图的台账（查询快照、口径卡）交回父图：父图报告的目录收进它们，引用标记、证据面板一路点到子图里的那一格；
- 子图出口的每个字段带出处，父图口径卡读它时跨子图往下追：追到原始数据的完整出具，子图里追不到的父图照样降档；
静态检查（发布前、应用检查）和运行时同一套认法；嵌套两层照样；
- 子图里的遮罩照父图的运行。
数据是虚构店铺库（tests/ledger_samples.py）。
"""
from __future__ import annotations
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import select
from app.db.base import SessionLocal
from app.db.models import RunEvent
from app.engine.lineage import traced, untraced
from app.engine.runner import run_manager
from tests import ledger_samples as S
PAID = "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid'"
GMV = {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "expression": "cell(nodes.fetch, 0, 'gmv')"}
@pytest.fixture(autouse=True)
async def engine_up(tmp_path):
await run_manager.setup()
await S.fresh_source(tmp_path / "shop.db")
yield
await run_manager.shutdown()
@pytest.fixture
async def client():
from app.main import app
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
def node(nid: str, ntype: str, label: str | None = None, **config) -> dict:
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}}
def metric(mid: str, expression: str, name: str = "") -> dict:
return {"id": mid, "name": name or mid, "unit": "元", "decimals": 2, "expression": expression}
def fetch_card_graph(*, card: dict | None = None, out: list[dict] | None = None, sql: str = PAID) -> dict:
"""子工作流：取数 → 口径卡 → 出口（gmv 原样交出口径卡的指标，table 原样交出查询结果）。"""
return S.chain(
node("s", "input"),
node("fetch", "tool", tool=S.tool(), args={"sql": sql}),
card or node("card", "metrics", caliber="子口径", caliber_version="v1", metrics=[GMV]),
node("o", "output", fields=out or [{"name": "gmv", "value": "{{ nodes.card.metrics[0].value }}"},
{"name": "table", "value": "{{ nodes.fetch }}"}]))
async def publish(graph: dict, name: str = "子流程") -> str:
return await S.publish(name, graph=graph)
def parent(sub_id: str, *middle: dict, contract_from: str | None = "top", out: list[dict] | None = None,
sub_input: dict | None = None) -> dict:
return S.chain(
node("start", "input", fields=[{"name": "region"}]),
node("sub", "subgraph", "取数子流程", workflow_id=sub_id, workflow_version=1,
**({"input": sub_input} if sub_input else {})),
*middle,
node("out", "output", fields=out or [{"name": "数", "value": "{{ nodes.top.metrics[0].value }}"}],
**({"contract": {"metrics_from": contract_from}} if contract_from else {})))
async def run(graph: dict, inputs: dict | None = None):
started = await run_manager.start(graph=graph, input_payload=inputs or {})
row = await S.wait(started.id)
assert row.status == "succeeded", row.error
return row
async def events(run_id: str) -> list[RunEvent]:
async with SessionLocal() as session:
return list((await session.execute(select(RunEvent).where(RunEvent.run_id == run_id)
.order_by(RunEvent.seq))).scalars())
async def card(run_id: str, node_id: str) -> dict:
from app.api.evidence import _sealed, _sealed_cards
sealed = await _sealed(run_id)
[found] = [c for n, _, c in _sealed_cards(sealed) if n == node_id]
return {m["id"]: m for m in found["metrics"]}
# --------------------------------------------------------------------------
# 子图里每一步进同一个运行的事件流
# --------------------------------------------------------------------------
async def test_the_subgraph_steps_are_in_the_run_with_prefixed_ids():
sub_id = await publish(fetch_card_graph())
row = await run(parent(sub_id, contract_from=None, out=[{"name": "数", "value": "{{ nodes.sub.output.gmv }}"}]))
evs = await events(row.id)
started = [(e.node_id, e.data.get("label")) for e in evs if e.type == "node.started"]
assert [n for n, _ in started] == ["start", "sub", "sub/s", "sub/fetch", "sub/card", "sub/o", "out"]
assert dict(started)["sub/fetch"] == "fetch"
[end] = [e for e in evs if e.type == "tool.end" and e.node_id == "sub/fetch"]
assert end.data.get("query_artifact")
# 子图里的台账条目在子图那一步的 node.finished 上（节点 id 带前缀），子工作流节点自己的 node.finished 不再记一遍
finished = {e.node_id: e for e in evs if e.type == "node.finished"}
assert [x["kind"] for x in finished["sub/fetch"].data["evidence"]][:1] == ["tool"]
assert "evidence" not in finished["sub"].data
# 封存覆盖它们
assert row.manifest_seq == evs[-1].seq
assert row.output == {"数": "246.5"}
async def test_the_instance_timeline_counts_the_subgraph_as_one_step():
"""应用的实例进度只说这张图上的步骤：子图里的那几步（「sub/…」）不单列，子工作流算一步。"""
from types import SimpleNamespace
from app.apps.resolve import _timeline
from app.db.models import Run
sub_id = await publish(fetch_card_graph())
row = await run(parent(sub_id, contract_from=None, out=[{"name": "数", "value": "{{ nodes.sub.output.gmv }}"}]))
async with SessionLocal() as session:
steps = (await _timeline(session, SimpleNamespace(instance=await session.get(Run, row.id))))["steps"]
assert [s["node"] for s in steps] == ["start", "sub", "out"]
async def test_an_error_inside_the_subgraph_names_the_step_inside_it():
sub_id = await publish(fetch_card_graph(sql="SELECT nope FROM nowhere"))
started = await run_manager.start(graph=parent(sub_id, contract_from=None,
out=[{"name": "数", "value": "{{ nodes.sub.output.gmv }}"}]),
input_payload={})
row = await S.wait(started.id)
assert row.status == "failed" and row.error_node_id == "sub/fetch"
[failed] = [e for e in await events(row.id) if e.type == "run.failed"]
assert (failed.data["node_id"], failed.data["label"]) == ("sub/fetch", "取数子流程 › fetch")
# --------------------------------------------------------------------------
# 父图口径卡读子图出口的字段：跨子图往下追
# --------------------------------------------------------------------------
async def test_a_parent_card_reading_a_subgraph_field_traces_to_the_query_cell_and_issues_formally(client):
sub_id = await publish(fetch_card_graph())
top = node("top", "metrics", caliber="父口径", caliber_version="v1",
metrics=[metric("gmv", "float(nodes.sub.output.gmv)"), metric("raw", "cell(nodes.sub.output.table, 0, 'gmv')")])
row = await run(parent(sub_id, top))
metrics = await card(row.id, "top")
[copy] = metrics["gmv"]["inputs"]
# 子图出口 gmv 原样交出子图口径卡的指标：接上那个指标自己的出处，一路追到取数查询
assert (copy["via"], copy["node_id"], copy["subgraph"], copy["field"]) == ("metric", "sub/card", "sub", "gmv")
assert copy["path"] == "nodes.sub.output.gmv" and copy["value"] == "246.5"
assert copy["upstream"]["traced"] is True and traced(copy)
# 子图出口 table 原样交出查询结果：cell() 按台账认出是子图里的那份快照
[cell] = metrics["raw"]["inputs"]
assert (cell["via"], cell["node_id"], cell["status"]) == ("tool_cell", "sub/fetch", "verified")
assert untraced(metrics["gmv"]["inputs"] + metrics["raw"]["inputs"]) == []
assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]["gaps"]
# 证据面板：转抄的那一项后面接一步 upstream（子图里的卡），一直到子图里的取数查询
body = (await client.get(f"/api/runs/{row.id}/evidence/metrics/top/gmv")).json()
chain = body["chain"]
assert chain[0]["untraced"] == []
[inp] = [s for s in chain if s["step"] == "input"]
assert inp["subgraph"] == "sub" and inp["subgraph_label"] == "取数子流程"
assert inp["node_id"] == "sub/card" and inp["node_label"] == "取数子流程 › card"
[up] = [s for s in chain if s["step"] == "upstream"]
assert (up["node_id"], up["label"], up["sealed"], up["traced"]) == ("sub/card", "取数子流程 › card", True, True)
[read] = up["inputs"]
assert read["node_id"] == "sub/fetch" and read["node_label"] == "取数子流程 › fetch"
[query] = [s for s in chain if s["step"] == "query"]
assert query["artifact"] == read["artifact"] and query["node_id"] == "sub/fetch" and query["sealed"] is True
async def test_an_untraced_field_inside_the_subgraph_degrades_the_parent():
inner = node("card", "metrics", caliber="子口径", caliber_version="v1", metrics=[metric("gmv", "vars.calc.v")])
sub = S.chain(node("s", "input"),
node("calc", "transform", mode="json", template='{"v": 5}', assign_to="calc"), inner,
node("o", "output", fields=[{"name": "gmv", "value": "{{ nodes.card.metrics[0].value }}"}]))
sub_id = await publish(sub)
top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("gmv", "float(nodes.sub.output.gmv)")])
row = await run(parent(sub_id, top))
[copy] = (await card(row.id, "top"))["gmv"]["inputs"]
assert copy["upstream"]["traced"] is False and not traced(copy)
issuance = row.output["_issuance"]
assert issuance["tier"] == "degraded"
assert any("指标「gmv」的输入没追到原始数据：nodes.sub.output.gmv（停在「取数子流程」）" in g
for g in issuance["gaps"]), issuance["gaps"]
async def test_a_field_that_is_not_a_whole_reference_stops_at_the_subgraph():
sub_id = await publish(fetch_card_graph(out=[{"name": "gmv", "value": "合计 {{ nodes.card.metrics[0].value }}"}]))
top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("n", "len(nodes.sub.output.gmv)")])
row = await run(parent(sub_id, top))
[item] = (await card(row.id, "top"))["n"]["inputs"]
assert (item["via"], item["node_id"]) == ("subgraph", "sub") and not traced(item)
assert row.output["_issuance"]["tier"] == "degraded"
async def test_the_subgraph_run_input_is_traced_back_through_the_parent_input_template():
"""子图出口交出它的运行输入：子图的输入是父图按「输入」模板渲染的。模板整串就是父图的运行输入时追到原始数据；拼了
别的字的停在子工作流的入口。"""
sub = S.chain(node("s", "input", fields=[{"name": "region"}]),
node("o", "output", fields=[{"name": "region", "value": "{{ input.region }}"}]))
sub_id = await publish(sub)
top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("r", "nodes.sub.output.region")])
row = await run(parent(sub_id, top, sub_input={"region": "{{ input.region }}"}), {"region": "east"})
[item] = (await card(row.id, "top"))["r"]["inputs"]
assert (item["via"], item["path"], item["subgraph"]) == ("input", "nodes.sub.output.region", "sub") and traced(item)
assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]["gaps"]
row = await run(parent(sub_id, top, sub_input={"region": "{{ input.region }}区"}), {"region": "east"})
[item] = (await card(row.id, "top"))["r"]["inputs"]
assert (item["via"], item["node_id"], item["field"]) == ("subgraph_input", "sub", "region") and not traced(item)
async def test_two_levels_of_nesting_trace_all_the_way_down(client):
inner_id = await publish(fetch_card_graph(), "内层")
middle = S.chain(node("m_in", "input"),
node("inner", "subgraph", "内层", workflow_id=inner_id, workflow_version=1),
node("m_out", "output", fields=[{"name": "gmv", "value": "{{ nodes.inner.output.gmv }}"}]))
middle_id = await publish(middle, "中层")
top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("gmv", "float(nodes.sub.output.gmv)")])
row = await run(parent(middle_id, top))
evs = await events(row.id)
assert "sub/inner/fetch" in {e.node_id for e in evs if e.type == "node.finished"}
[copy] = (await card(row.id, "top"))["gmv"]["inputs"]
assert (copy["via"], copy["node_id"], copy["subgraph"]) == ("metric", "sub/inner/card", "sub")
assert copy["upstream"]["traced"] is True
assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]["gaps"]
chain = (await client.get(f"/api/runs/{row.id}/evidence/metrics/top/gmv")).json()["chain"]
[up] = [s for s in chain if s["step"] == "upstream"]
assert up["node_id"] == "sub/inner/card" and up["sealed"] is True and up["label"] == "取数子流程 › 内层 › card"
[query] = [s for s in chain if s["step"] == "query"]
assert query["node_id"] == "sub/inner/fetch" and query["sealed"] is True
# --------------------------------------------------------------------------
# 静态检查：和运行时同一套认法
# --------------------------------------------------------------------------
async def _static(graph: dict) -> list[str]:
"""受管级按指标出具的发布检查里「指标追不到原始数据」的那几条（governed.metric_untraced），子工作流读好放进来。"""
from app.engine.governance import lint_for_publish
from app.engine.lint_process import load_subgraphs, with_subgraphs
from app.engine.schema import GraphSpec
spec = GraphSpec.model_validate(graph)
async with SessionLocal() as session:
graphs = await load_subgraphs(session, spec)
with with_subgraphs(graphs):
issues = lint_for_publish(spec, level="governed").issues
return [i.message for i in issues if i.code == "governed.metric_untraced"]
async def test_the_static_check_agrees_with_the_run():
from app.engine.governance import lint_for_publish
from app.engine.schema import GraphSpec
traced_sub = await publish(fetch_card_graph())
inner = node("card", "metrics", caliber="子口径", caliber_version="v1", metrics=[metric("gmv", "vars.calc.v")])
loose_sub = await publish(S.chain(
node("s", "input"), node("calc", "transform", mode="json", template='{"v": 5}', assign_to="calc"), inner,
node("o", "output", fields=[{"name": "gmv", "value": "{{ nodes.card.metrics[0].value }}"}])), "松的")
top = node("top", "metrics", caliber="父口径", caliber_version="v1",
metrics=[metric("gmv", "float(nodes.sub.output.gmv)"), metric("raw", "cell(nodes.sub.output.table, 0, 'gmv')")])
good = parent(traced_sub, top)
bad = parent(loose_sub, node("top", "metrics", caliber="父口径", caliber_version="v1",
metrics=[metric("gmv", "float(nodes.sub.output.gmv)")]))
assert await _static(good) == []
[said] = await _static(bad)
assert "停在「取数子流程」" in said
# 运行时同一个结论
assert (await run(good)).output["_issuance"]["tier"] == "formal"
assert (await run(bad)).output["_issuance"]["tier"] == "degraded"
# 没把子工作流读好放进来（只看图本身的地方）：看不到子图，照旧停在子工作流节点
plain = [i for i in lint_for_publish(GraphSpec.model_validate(good), level="governed").issues
if i.code == "governed.metric_untraced"]
assert plain and "停在「取数子流程」" in plain[0].message
async def test_the_static_check_follows_two_levels_and_the_input_template():
inner_id = await publish(fetch_card_graph(), "内层")
middle_id = await publish(S.chain(
node("m_in", "input"), node("inner", "subgraph", "内层", workflow_id=inner_id, workflow_version=1),
node("m_out", "output", fields=[{"name": "gmv", "value": "{{ nodes.inner.output.gmv }}"}])), "中层")
top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("gmv", "float(nodes.sub.output.gmv)")])
assert await _static(parent(middle_id, top)) == []
echo_id = await publish(S.chain(node("s", "input", fields=[{"name": "region"}]),
node("o", "output", fields=[{"name": "region", "value": "{{ input.region }}"}])),
"回声")
top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("r", "nodes.sub.output.region")])
assert await _static(parent(echo_id, top, sub_input={"region": "{{ input.region }}"})) == []
assert await _static(parent(echo_id, top, sub_input={"region": "{{ input.region }}区"}))
# --------------------------------------------------------------------------
# 父图报告直接引用子图里的查询和口径卡
# --------------------------------------------------------------------------
def report_parent(sub_id: str) -> dict:
return S.chain(
node("start", "input"),
node("sub", "subgraph", "取数子流程", workflow_id=sub_id, workflow_version=1),
node("write", "report", instructions="写答案"),
node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}],
contract={"report_from": "write"}))
async def _segments(client: AsyncClient, run_id: str) -> dict[str, dict]:
body = (await client.get(f"/api/runs/{run_id}/evidence")).json()
assert body["mode"] == "cited", body
return body
async def test_the_parent_report_cites_the_subgraph_query_and_card(client, monkeypatch):
sub_id = await publish(fetch_card_graph())
S.script(monkeypatch, "子流程里查到的销售额 [[v:Q1.r0.gmv]]，口径算的是 [[m:gmv]]。")
started = await run_manager.start(graph=report_parent(sub_id), input_payload={})
row = await S.wait(started.id)
assert row.status == "succeeded", row.error
body = await _segments(client, row.id)
items = {e["alias"]: e for e in body["evidence"]}
assert (items["Q1"]["node_id"], items["Q1"]["sealed"]) == ("sub/fetch", True)
assert (items["m:gmv"]["node_id"], items["m:gmv"]["sealed"]) == ("sub/card", True)
cited = {e["to"]: e["from"] for e in body["edges"] if e["rel"] == "cites"}
# 单元格引用一路点到子图里的那一格
cell = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cited['Q1']}")).json()["chain"]
[query] = [s for s in cell if s["step"] == "query"]
assert query["node_id"] == "sub/fetch" and query["sealed"] is True and query["artifact"] == items["Q1"]["artifact"]
# 指标引用：子图里的卡、它的输入、子图里的查询
chain = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cited['m:gmv']}")).json()["chain"]
step = chain[0]
assert (step["step"], step["node_id"], step["sealed"], step["untraced"]) == ("metric", "sub/card", True, [])
[inp] = [s for s in chain if s["step"] == "input"]
assert (inp["node_id"], inp["node_label"], inp["via"]) == ("sub/fetch", "取数子流程 › fetch", "tool_cell")
[q] = [s for s in chain if s["step"] == "query"]
assert q["node_id"] == "sub/fetch" and q["sealed"] is True
# 封存核对通过，可验证报告照常（事件本来就在同一个运行里）
from app.engine.runner import verify_manifest
assert (await verify_manifest(row.id))["ok"] is True
res = await client.get(f"/api/runs/{row.id}/report.html")
assert res.status_code == 200 and "246.5" in res.text
async def test_masks_inside_the_subgraph_follow_the_parent_run(client, monkeypatch, tmp_path):
"""数据源遮了 phone：子图里查出来的号码在父图报告的证据面板里照样写「已遮罩」。"""
await S.set_source(tmp_path / "shop.db", options={"mask_columns": "phone"})
amount = node("card", "metrics", caliber="子口径", caliber_version="v1",
metrics=[metric("gmv", "cell(nodes.fetch, 0, 'amount')")])
sub_id = await publish(fetch_card_graph(card=amount, sql="SELECT phone, amount FROM orders WHERE id = 1"))
S.script(monkeypatch, "第一单的电话 [[v:Q1.r0.phone]]。")
started = await run_manager.start(graph=report_parent(sub_id), input_payload={})
row = await S.wait(started.id)
assert row.status == "succeeded", row.error
body = await _segments(client, row.id)
cited = {e["to"]: e["from"] for e in body["edges"] if e["rel"] == "cites"}
chain = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cited['Q1']}")).json()["chain"]
[query] = [s for s in chain if s["step"] == "query"]
assert query["node_id"] == "sub/fetch"
assert "13800000001" not in str(query), "子图里的遮罩列照父图的运行遮"
assert "已遮罩" in str(query) and query["sealed"] is True