Skip to main content

BLOGE 0.9.8-RC1 在线版 · 事实校验 2026-09-15 · English

第 10 章 —— 用子图复用

承诺： 读完本章后，你将知道如何把一个流水线提取为命名子图，使用 subgraph("name") 将其嵌入父图，控制作用域隔离，并识别何时应该使用动态子图。

学习目标​

  1. 解释什么是 BLOGE 中的子图（subgraph），以及它为何存在。
  2. 使用 DslCompiler.registerSubGraph() 注册预构建的 Graph，并在 DSL 中通过 node x : subgraph("name") {} 语法引用它。
  3. 区分**隔离（isolated）作用域（子图的默认模式）和父级（parent）**作用域，并预测每种模式下子图内部能看到什么。
  4. 读取子图输出——理解终端节点的输出会被聚合到父节点的结果 map 中。
  5. 识别何时应该使用 dynamicSubGraph operator 来执行运行时生成的 DSL，而不是编译时注册的子图。

前置条件​

源示例​

文件展示内容
loan-approval-subgraph.bloge两个并行子图（信用评估 + 合规检查）汇入一个带分支的核保决策
international-shipment.bloge并行的清关和路线优化子图，带有扇入
smart-ticket-handling.bloge情感分析子图驱动优先级分支，进入升级处理子图
order-full-pipeline.bloge支付和库存子图并行运行
dynamic-agent.bloge运行时 DSL 生成 + dynamicSubGraph 执行
LoanApprovalSubGraphDslExample.javaJava 接线：通过 Graph.builder() 构建子图，使用 DslCompiler.registerSubGraph() 注册
SubGraphOperator.java引擎层面的实现：执行子图并聚合终端节点输出

为什么这很重要​

到第 8 章为止，你已经能用 DSL 编写完整的 graph。但真实系统中有些流水线会反复出现在不同业务领域：

  • 一条信用评估流水线（征信查询 → 收入验证 → 负债率计算 → 风险评分）会出现在贷款审批、房贷办理和信用卡发行中。
  • 一条合规检查流水线（反洗钱筛查 → KYC → 制裁名单 → 合规判定）被多个监管工作流所需。
  • 一条厨房调度或配送协调流水线在不同的餐饮订单类型之间共享。

没有子图的话，你只能把整段节点链复制粘贴到每个父图中，重复韧性配置、依赖和 operator 接线。当合规规则变化时，你需要更新 N 份副本——而且不可避免地会漏掉某一份。

BLOGE 用命名子图解决这个问题：你把一条流水线构建为一个 Graph，在一个名字下注册它，然后在任何父图中用 node x : subgraph("name") {} 引用它。引擎会将子图作为嵌套执行来运行，把父级的 input {} 绑定注入为子图的上下文，并将子图的终端节点输出聚合回父节点的结果。

一条流水线。一次注册。多个调用点。

心智模型​

Diagram: 10-reuse-with-subgraphs figure 1

把每个子图方框想象成一次函数调用：父图通过 input {} 传递参数，子图运行其内部 DAG，父图收到以节点 ID 为键的终端节点输出 map。

核心规则：

  1. 输入通过 input {} 流入。 这些绑定会被注入为子图的 GraphContext。子图通过 ctx.* 表达式读取它们。
  2. 输出通过终端节点流出。 SubGraphOperator 遍历 subGraph.terminalNodes()，将每个终端节点的输出收集到一个 Map<String, Object> 中。父图通过 subgraphNode.output.terminalNodeId 引用它们。
  3. 作用域默认为 isolated。 子图不会看到父节点的输出，也不会继承父级的 GraphContext——除非你显式设置 scope = parent。

Diagram: 10-reuse-with-subgraphs figure 2

第一个可运行示例​

下面是 loan-approval-subgraph.bloge， 精简到子图接线部分：

这 71 行接线保留在同一个代码块中，因为两个并行 child 合同、parent fan-in 和 两条终止路由必须同时可见；继续裁剪会隐藏父图可见的 output 边界。

graph loanApprovalPipeline {

  node receiveApplication : ReceiveApplicationOperator {
    input {
      applicationId   = ctx.applicationId
      applicantName   = ctx.applicantName
      requestedAmount = ctx.requestedAmount
      termMonths      = ctx.termMonths
      employerId      = ctx.employerId
    }
    timeout = 3s
  }

  /// 运行信用评估子图：
  /// creditQuery → incomeVerification → debtRatioCalc → riskScoring
  node creditAssessment : subgraph("credit-assessment") {
    depends_on = [receiveApplication]
    input {
      applicationId   = receiveApplication.output.applicationId
      applicantName   = receiveApplication.output.applicantName
      requestedAmount = receiveApplication.output.requestedAmount
      employerId      = receiveApplication.output.employerId
    }
    timeout = 30s
  }

  /// 运行合规检查子图：
  /// amlScreening → kycVerification → sanctionListCheck → complianceDetermination
  node complianceCheck : subgraph("compliance-check") {
    depends_on = [receiveApplication]
    input {
      applicationId = receiveApplication.output.applicationId
      applicantName = receiveApplication.output.applicantName
    }
    timeout = 30s
  }

  node underwritingDecision : UnderwritingDecisionOperator {
    depends_on = [creditAssessment, complianceCheck]
    input {
      applicationId = receiveApplication.output.applicationId
      credit        = creditAssessment.output.riskScoring
      compliance    = complianceCheck.output.complianceDetermination
    }
  }

  branch on underwritingDecision.output.decision {
    "approved" -> generateApprovalLetter
    otherwise  -> generateRejectionNotice
  }

  node generateApprovalLetter : GenerateApprovalLetterOperator {
    depends_on = [underwritingDecision]
    input {
      applicationId   = underwritingDecision.output.applicationId
      applicantName   = ctx.applicantName
      requestedAmount = ctx.requestedAmount
      approvedRate    = underwritingDecision.output.approvedRate
      termMonths      = ctx.termMonths
    }
  }

  node generateRejectionNotice : GenerateRejectionNoticeOperator {
    depends_on = [underwritingDecision]
    input {
      applicationId = underwritingDecision.output.applicationId
      applicantName = ctx.applicantName
      reason        = underwritingDecision.output.reason
    }
  }
}

把它当成一个故事来读：

  1. receiveApplication 验证传入的贷款申请。
  2. 两个子图并行运行 —— creditAssessment 和 complianceCheck —— 各自包装了一个四节点的流水线。两者都依赖 receiveApplication，因此引擎会并发调度它们。
  3. underwritingDecision 从两个子图扇入。它读取 creditAssessment.output.riskScoring —— 信用子图中终端节点的输出 —— 以及 complianceCheck.output.complianceDetermination。
  4. 分支根据结果路由到批准或拒绝。

关键认识： 每个 subgraph("name") 节点在父图的 DAG 中看起来就像一个普通节点。引擎不关心它内部包含四个步骤——它只看到一个有依赖和输出 map 的节点。

一个子图，只向父图交付一个结果​

贷款流程中的 creditAssessment 内部包含信用查询、收入核验、负债率计算和 风险评分。父图不应该绑定四个内部工作台；它只需要一份稳定的信用评估结果。

图：父图与子图的 terminal output 边界

从外向内读取这条边界​

视角可以依赖不应依赖
子图内部收入、负债率等中间输出父图 node 名或父图调度
子图终态合同{score, grade, reasons}内部 node 布局
父图 underwritingDecisioncreditAssessment.output子图里的 creditQuery.output

companion smoke Scenario 观察的是公开结果：creditAssessment 与 complianceCheck 都执行，underwriting 消费两者输出，批准信执行而拒绝信为 SKIPPED。父图不需要知道分数由子图中的哪个 node 产生。

重构时不要泄漏工作台​

把 riskScoring 改名为 calculateRisk，或在它之前插入欺诈检查。只要 terminal output 仍满足 {score, grade, reasons}，父图就不用变化。如果父图 直接引用子图内部 node，一次私有重构就会变成跨 graph 迁移，说明复用边界 已经失效。

动态子图应作为进阶部署技术。它改变运行时如何选择 definition，不改变这条 合同规则。先掌握静态父子输出边界，只有当业务确实需要在多个 definition 之间选择时，再增加动态选择。

拆开来看​

subgraph("name") 语法​

node <id> : subgraph("<registered-name>") {
  depends_on = [...]
  input { ... }
  timeout = ...
  scope = isolated | parent    // 可选；默认为 isolated
}

解析器识别 subgraph( 令牌序列，并将 operator 引用重写为一个内部前缀（subgraph:<name>）。 StandardNodeCompiler 随后在已注册的子图 map 中查找该名称，并接线一个 SubGraphOperator。

注册子图（Java 端）​

在编译父级 DSL 之前，你需要构建每个子图并注册它：

// 通过 Java API 构建子图
Graph creditGraph   = buildCreditAssessmentSubGraph();   // Graph.builder("credit-assessment")...
Graph complianceGraph = buildComplianceCheckSubGraph();  // Graph.builder("compliance-check")...

// 在 compile() 之前注册
var compiler = new DslCompiler(registry);
compiler.registerSubGraph("credit-assessment", creditGraph);
compiler.registerSubGraph("compliance-check", complianceGraph);

Graph mainGraph = compiler.compile(ast);

（摘自 LoanApprovalSubGraphDslExample.java，第 274–284 行。）

每个子图都使用标准的 Graph.builder() API 构建——operator、依赖、输入、重试、超时——与顶层 graph 完全一致。唯一的区别是它从 ctx.* 读取初始数据，而不是从父节点的输出。

输出聚合的工作原理​

SubGraphOperator.execute() 运行子图，然后：

Map<String, Object> output = new LinkedHashMap<>();
for (String terminalId : subGraph.terminalNodes()) {
    Object terminalOutput = result.results().getRaw(terminalId);
    if (terminalOutput != null) {
        output.put(terminalId, terminalOutput);
    }
}
return output;

父图通过名称引用每个终端节点：creditAssessment.output.riskScoring 导航到该 map 中的 riskScoring 键。

作用域：isolated vs parent​

作用域默认用于行为
isolatedsubgraph("name")子图仅接收显式的 input {} 绑定。不会泄漏父级 GraphContext。
parentloop、foreach子图继承父级 GraphContext，并可通过运行时注入的 __parentOutput_* 键引用父节点输出。

你可以覆盖默认值：

node nested : subgraph("childGraph") {
  scope = parent
  input { key = parentNode.output.value }
}

使用 scope = parent 时，SubGraphOperator 会先将父级的 GraphContext 合并到子图上下文中，再叠加显式输入。这对原型开发很方便，但会削弱封装性——子图现在隐式依赖于父级的上下文结构。

（参见 DSL 规范 和 StandardNodeCompiler.java 了解作用域规则和编译路径。）

进阶：动态子图​

对于子图在编译时未知的场景，BLOGE 提供了 dynamicSubGraph operator。上游节点生成 DSL 文本；dynamicSubGraph 在运行时对其进行沙箱验证、编译并执行：

node plan : generateDynamicDsl {
  input { task = ctx.task, history = ctx.history }
  timeout = 3s
}

node executePlan : dynamicSubGraph {
  depends_on = [plan]
  input {
    dslSource = plan.output.dslSource
    context = { task: ctx.task, history: ctx.history, planKind: plan.output.planKind }
  }
  timeout = 15s
  retry = { attempts: 2, backoff: 100ms, strategy: exponential }
}

（摘自 dynamic-agent.bloge。）

dynamicSubGraph 是一个能力层 operator，通过 CommonOperators.builder().dynamic().build() 注册。它使用 DslSandbox 验证生成的 DSL，通过 GraphComplexityValidator 强制复杂度限制，并通过 GraphEngine.executeNestedGraph() 运行子图。

对编译时组合使用 subgraph("name")。仅当 graph 结构必须在运行时决定时才使用 dynamicSubGraph（例如 LLM 生成的计划、插件架构）。

常见陷阱​

❌ 把子图节点的输出当作扁平值​

// 错误 —— creditAssessment.output 是一个以终端节点 ID 为键的 Map
node decision : UnderwritingOp {
  input {
    riskGrade = creditAssessment.output.riskGrade   // ← 路径错误
  }
}

子图节点的输出是一个 Map<String, Object>，其中每个键是子图的终端节点 ID。如果子图的终端节点是 riskScoring，正确的路径是：

// 正确 —— 通过终端节点 ID 导航
node decision : UnderwritingOp {
  input {
    credit = creditAssessment.output.riskScoring   // ← 终端节点 ID
  }
}

如果子图有多个终端节点，每个都会作为单独的键出现。查看子图的结构以确定要引用哪些终端 ID。

❌ 把 scope = parent 当作默认选项​

到处加 scope = parent 以免在 input {} 中逐一穿透每个值，确实很诱人。请克制。隔离作用域迫使你声明契约——子图实际需要哪些输入。这使得子图能在不同的父图之间复用。将 scope = parent 保留给真正的共享基础设施上下文（例如追踪 ID、租户标识）。

引导式重写​

查看 international-shipment.bloge。 它使用了两个并行子图 —— customsClearance 和 routeOptimization —— 然后在 bookingConfirmation 处扇入。

思考以下问题：

  1. 父图如何向 customsClearance 传递数据？ （通过显式的 input {} 绑定。子图读取 ctx.shipmentId、ctx.commodityType 等，因为这些值是从父图的 input 块注入到子图的 GraphContext 中的。）

  2. routeOptimization.output.optimalRouteSelection 是什么？ （它是路线优化子图中名为 optimalRouteSelection 的终端节点的输出。由 SubGraphOperator 聚合而来。）

  3. 如果移除 customsClearance 上的 timeout = 30s 会怎样？ （子图节点将没有时间限制。如果子图挂起——比如 HS 编码分类服务宕机了——父图将在该节点无限期阻塞。请始终为子图节点设置 timeout。）

  4. 能否在另一个父图中复用 customsClearance 子图？ （可以——只要你用 "customs-clearance" 注册同一个子图，并提供期望的输入：shipmentId、commodityType、originCountry、destinationCountry、weightKg。）

现在尝试修改这个 graph：

  • 把 trackingSetup 和 sendNotification 提取到一个名为 "post-booking" 的新子图中。注册它。用一个 node postBooking : subgraph("post-booking") {} 节点替换这两个节点。
  • 你需要传递哪些输入？父图将引用哪个终端节点 ID？

脑力检查​

  1. 在 DSL 中引用预注册子图的语法是什么？ （node <id> : subgraph("<name>") { ... }。）

  2. 如何在编译前注册子图？ （在 DslCompiler 实例上调用 compiler.registerSubGraph("name", graph)。）

  3. subgraph("name") 节点的默认作用域是什么？ （isolated —— 子图只能看到显式的 input {} 绑定。）

  4. 父节点如何读取子图的结果？ （通过 subgraphNode.output.<terminalNodeId> —— SubGraphOperator 将每个终端节点的输出收集到一个以节点 ID 为键的 map 中。）

  5. 什么时候应该使用 dynamicSubGraph 而不是 subgraph("name")？ （当子图的结构在运行时才能确定时——例如 LLM 生成的计划或插件提供的工作流。subgraph("name") 用于编译时组合。）

  6. 设计题： 你的 compliance-check 子图被三个父图复用。其中贷款审批父图需要更细的 AML 明细，而另外两个父图并不需要。你应该把子图参数化、派生一个变体，还是把这部分细节推回父图里？（优先考虑通过子图输入做参数化。如果这份额外明细会从根本上改变子图的处理流水线，就创建一个变体。不要把领域逻辑推回父图，否则就削弱了抽取子图的意义。）

练习​

  1. 打开 loan-approval-subgraph.bloge 和 LoanApprovalSubGraphDslExample.java。

    • 追踪数据流：creditAssessment.output.riskScoring 在运行时解析为什么？ （一个包含 riskGrade、compositeScore、summary 键的 map —— 即信用评估子图中 riskScoring 节点的输出。）

  2. 打开 smart-ticket-handling.bloge。

    • 该 graph 有两个子图：sentimentAnalysis 和 escalationWorkflow。只有其中一个总是运行。是哪个，为什么？ （只有 sentimentAnalysis 总是运行。escalationWorkflow 位于 branch on determinePriority.output.priority 之后——它只在优先级为 "high" 时执行。）

  3. 为电商退货流水线设计一个新 graph：

    • 创建一个 "fraud-screening" 子图，包含三个节点： orderLookup → returnHistoryCheck → fraudScore。
    • 创建一个 "refund-processing" 子图，包含两个节点： calculateRefund → initiateRefund。
    • 编写一个父图：先运行欺诈筛查，根据欺诈评分进行分支，将干净的退货路由到退款处理。
    • 每个子图需要的最小 input {} 契约是什么？

  4. 附加题： 打开 dynamic-agent.bloge。

    • 如果 plan 节点生成了无效的 DSL 会怎样？ （DynamicSubGraphOperator 内部的 DslSandbox.parseAndValidate() 调用会抛出异常。如果 executePlan 配置了 retry，引擎会重试。如果所有尝试都失败且没有 fallback，graph 将失败。）

实验验收卡​

  • 预期与观察： 父图只消费子图 terminal output，不依赖内部节点。
  • 失败与恢复： 绑定不存在的内部 output；恢复 terminal node ID 后重跑。
  • 证明边界： 证明子图输出合同和隔离，不证明动态子图安全。
  • 练习合同： 贷款子图；只改一个 output binding；交付可见 key 和失败 reason；父图只引用 terminal key 即停止。

回顾​

  • subgraph("name") 让你将一个命名的、预构建的 graph 作为单个节点嵌入到父图中。子图通过 DslCompiler.registerSubGraph() 在编译前注册。
  • 输入通过 input {} 绑定流入，这些绑定成为子图的 GraphContext。输出通过终端节点流回 —— 父图通过 subgraphNode.output.<terminalNodeId> 读取它们。
  • 子图默认使用 isolated 作用域：不会泄漏父级上下文。谨慎使用 scope = parent —— 它会削弱封装性。
  • 子图是父图 DAG 中的普通节点。它们像其他节点一样支持 depends_on、timeout、retry 和 fallback。当依赖关系允许时，多个子图节点可以并行运行。
  • 对于运行时生成的 graph，使用 dynamicSubGraph operator，它在执行时对 DSL 文本进行沙箱验证和编译。
  • 子图通过 GraphEngine.executeNestedGraph() 运行——从父级继承监听器、持久化服务和执行 ID 链。

下一步​

在第 11 章 — 批处理与迭代中，你将学习 foreach 和 loop 如何让你对集合和轮询周期重复执行工作——这两者在底层都会将其主体编译为嵌套子图。

参考链接​

Coding Agent: Open the versioned task guide.