MediaPipe 图系统深度解析：CalculatorGraphConfig、Subgraph 与循环图（Cycles）设计全解

在 MediaPipe 中，整个推理与流处理框架的“骨架”就是图（Graph）：CalculatorGraphConfig proto 定义了图的拓扑与功能，子图（Subgraph）实现了感知方案的模块化复用，图选项（Graph Options）提供了跨层级的参数化能力，而循环图（Cycles）机制则让 MediaPipe 能够表达带状态反馈的信号处理结构。读完本文，你将掌握：如何用 pbtxt 与 C++ 两种方式搭建 MediaPipe 图、如何定义并注册 mediapipe_simple_subgraph、option_value 的 ProtoPath 语法如何工作，以及循环图所需的 back edge 标注、初始包、延迟与提前关闭等关键设计。

一、Graph：用 CalculatorGraphConfig 描述一张 MediaPipe 图

一个 CalculatorGraphConfig proto 规定了 MediaPipe 图的拓扑与功能。图中每个 node 代表一个特定的 calculator 或 subgraph，并携带其必要配置：已注册的 calculator/subgraph 类型名、输入流、输出流，以及可选字段如节点专属选项（node_options）、输入策略（input_stream_handler）和执行器（executor，详见 Synchronization）。

除了节点，CalculatorGraphConfig 还有一组图级全局设置字段，可用于跨平台调优。结合 proto 定义 mediapipe/framework/calculator.proto，关键全局字段包括：

• num_threads（field 8）：多线程模式下的线程数；不指定时调度器根据可用处理器数量自行决定；指定 "ApplicationThreadExecutor" 可让图在调用线程上运行。
• max_queue_size（field 11）：图中任意输入流的最大队列长度，用于防止快速数据源以 packet 淹没图、从而控制内存占用；未指定时默认为 100 个 packet，设为 -1 则禁用节流。若节点声明了 buffer_size_hint，则取 max(buffer_size_hint, max_queue_size)。
• executor（field 14）：重复的 ExecutorConfig，可定义多个具名执行器，把重计算节点挂到独立执行器上。proto 注释中特别提到：在移动端，把重量级的模型推理 calculator 挂到独立执行器可以带来线程局部性收益，有利于实时应用的性能。
• report_deadlock（field 21）：当节流导致所有 calculator 都无法运行时，是否直接报错（true）还是自动放宽 max_queue_size（false）。
• input_stream_handler / output_stream_handler（field 12/13）：图级默认的输入/输出流处理器，节点未单独指定时继承图级配置。

一个最小的 CalculatorGraphConfig 示例——4 个串联的 passthrough calculator：

# This graph named main_pass_throughcals_nosubgraph.pbtxt contains 4
# passthrough calculators.
input_stream: "in"
output_stream: "out"
node {
    calculator: "PassThroughCalculator"
    input_stream: "in"
    output_stream: "out1"
}
node {
    calculator: "PassThroughCalculator"
    input_stream: "out1"
    output_stream: "out2"
}
node {
    calculator: "PassThroughCalculator"
    input_stream: "out2"
    output_stream: "out3"
}
node {
    calculator: "PassThroughCalculator"
    input_stream: "out3"
    output_stream: "out"
}

注意节点 calculator 字段填的是注册名：calculator 通过 REGISTER_CALCULATOR 宏注册，subgraph 通过 REGISTER_MEDIAPIPE_GRAPH 宏注册（见 mediapipe/framework/calculator.proto 中 Node.calculator 的注释与 mediapipe/framework/subgraph.h 中的宏定义）。此外，Node 消息还支持 tag:name 形式的流命名（如 IMAGE:throttled_image）、source_layer（控制 source 层执行顺序）、max_in_flight（并行调用上限，默认 1）等字段，实际使用时可在同一文件中查阅。

二、图的 C++ 表示：Graph/Stream API

对于更复杂的图（例如 ML 流水线、处理模型元数据、可选节点等），MediaPipe 提供了 C++ 的图构建 API，比手写 pbtxt 更易组织。上面那张 passthrough 图用 C++ 表示为：

CalculatorGraphConfig BuildGraphConfig() {
  Graph graph;

  // Graph inputs
  Stream<AnyType> in = graph.In(0).SetName("in");

  auto pass_through_fn = [](Stream<AnyType> in,
                            Graph& graph) -> Stream<AnyType> {
    auto& node = graph.AddNode("PassThroughCalculator");
    in.ConnectTo(node.In(0));
    return node.Out(0);
  };

  Stream<AnyType> out1 = pass_through_fn(in, graph);
  Stream<AnyType> out2 = pass_through_fn(out1, graph);
  Stream<AnyType> out3 = pass_through_fn(out2, graph);
  Stream<AnyType> out4 = pass_through_fn(out3, graph);

  // Graph outputs
  out4.SetName("out").ConnectTo(graph.Out(0));

  return graph.GetConfig();
}

该写法中 graph.In(0) / graph.Out(0) 声明图的输入输出端，AddNode 按注册名添加节点，ConnectTo 完成边连接，最终通过 GetConfig() 得到与 pbtxt 等价的 CalculatorGraphConfig。更多细节见 Building Graphs in C++。

三、Subgraph：把图模块化、可复用

为了把 CalculatorGraphConfig 拆成可复用的子模块，MediaPipe 允许把一个图定义成 Subgraph。子图的公开接口由一组输入流、输出流、输入侧 packet、输出侧 packet 组成，与 calculator 的公开接口同构。子图可以在 CalculatorGraphConfig 中“像 calculator 一样”被引用；当图从配置加载时，每个 subgraph 节点会被替换成其对应的 calculator 图，因此子图的语义与性能与其展开后的图完全一致。

从源码可以印证这一机制：mediapipe/framework/subgraph.h 中 Subgraph 类的 GetConfig(SubgraphContext* sc) 注释明确写着 “The nodes and generators in this config will replace the subgraph node in the parent graph”——即构建期用展开后的配置替换父图中的子图节点，运行时并不存在“嵌套调度”开销。该文件还提供 SubgraphContext（可读取节点选项、访问 Resources 与 GraphService）以及 ProtoSubgraph（包装一个字面 CalculatorGraphConfig）、TemplateSubgraph（包装 CalculatorGraphTemplate）等实现。

创建一个名为 TwoPassThroughSubgraph 的子图分三步：

第 1 步：定义子图（pbtxt）：

# This subgraph is defined in two_pass_through_subgraph.pbtxt
# and is registered as "TwoPassThroughSubgraph"

type: "TwoPassThroughSubgraph"
input_stream: "out1"
output_stream: "out3"

node {
    calculator: "PassThroughCalculator"
    input_stream: "out1"
    output_stream: "out2"
}
node {
    calculator: "PassThroughCalculator"
    input_stream: "out2"
    output_stream: "out3"
}

子图的公开接口由四部分组成：图输入流、图输出流、图输入侧 packet、图输出侧 packet。

第 2 步：用 BUILD 规则注册。规则 mediapipe_simple_subgraph 的 register_as 参数定义了子图的组件名（该规则定义在 mediapipe/framework/tool/mediapipe_graph.bzl）：

# Small section of BUILD file for registering the "TwoPassThroughSubgraph"
# subgraph for use by main graph main_pass_throughcals.pbtxt

mediapipe_simple_subgraph(
    name = "twopassthrough_subgraph",
    graph = "twopassthrough_subgraph.pbtxt",
    register_as = "TwoPassThroughSubgraph",
    deps = [
            "//mediapipe/calculators/core:pass_through_calculator",
            "//mediapipe/framework:calculator_graph",
    ],
)

仓库中大量真实子图都按此模式注册，例如 mediapipe/modules/face_detection/BUILD 中的 mediapipe_simple_subgraph(name = "face_detection_short_range_cpu", graph = "face_detection_short_range_cpu.pbtxt", register_as = "FaceDetectionShortRangeCpu", ...)。

第 3 步：在主图中引用子图：

# This main graph is defined in main_pass_throughcals.pbtxt
# using subgraph called "TwoPassThroughSubgraph"

input_stream: "in"
node {
    calculator: "PassThroughCalculator"
    input_stream: "in"
    output_stream: "out1"
}
node {
    calculator: "TwoPassThroughSubgraph"
    input_stream: "out1"
    output_stream: "out3"
}
node {
    calculator: "PassThroughCalculator"
    input_stream: "out3"
    output_stream: "out4"
}

四、Graph Options：图的参数化接口

类似给单个 calculator 指定 Calculator Options proto 一样，MediaPipe 图也可以指定一个 “graph options” protobuf。graph options 在调用图的位置给出，用于填充图内部各 calculator 的选项与子图选项——这是把“对外参数”与“内部实现”解耦的核心机制。

4.1 在调用处指定 graph options。 在 CalculatorGraphConfig 中，子图的 graph options 与 calculator options 的写法完全一致：

node {
  calculator: "FlowLimiterCalculator"
  input_stream: "image"
  output_stream: "throttled_image"
  node_options: {
    [type.googleapis.com/mediapipe.FlowLimiterCalculatorOptions] {
      max_in_flight: 1
    }
  }
}

node {
  calculator: "FaceDetectionSubgraph"
  input_stream: "IMAGE:throttled_image"
  node_options: {
    [type.googleapis.com/mediapipe.FaceDetectionOptions] {
      tensor_width: 192
      tensor_height: 192
    }
  }
}

4.2 在图内部接收并消费 graph options。 子图在自己的 CalculatorGraphConfig 中声明 graph_options，并用 option_value 把外部选项的值映射到内部各节点的选项字段：

graph_options: {
  [type.googleapis.com/mediapipe.FaceDetectionOptions] {}
}

node: {
  calculator: "ImageToTensorCalculator"
  input_stream: "IMAGE:image"
  node_options: {
    [type.googleapis.com/mediapipe.ImageToTensorCalculatorOptions] {
        keep_aspect_ratio: true
        border_mode: BORDER_ZERO
    }
  }
  option_value: "output_tensor_width:options/tensor_width"
  option_value: "output_tensor_height:options/tensor_height"
}

node {
  calculator: "InferenceCalculator"
  node_options: {
    [type.googleapis.com/mediapipe.InferenceCalculatorOptions] {}
  }
  option_value: "delegate:options/delegate"
  option_value: "model_path:options/model_path"
}

在这个例子中，FaceDetectionSubgraph 接受 graph option FaceDetectionOptions，该 proto 的字段值被用于填充 ImageToTensorCalculatorOptions 的部分字段以及 InferenceCalculatorOptions 的部分字段，字段映射通过 option_value: 语法完成。这一机制在仓库中的真实落地可以参见 mediapipe/modules/face_detection/face_detection.pbtxt：其开头声明 graph_options: { [type.googleapis.com/mediapipe.FaceDetectionOptions] {} }，内部节点如 ImageToTensorCalculator 使用 option_value: "output_tensor_width:options/tensor_width"、推理节点使用 option_value: "delegate:options/delegate"，与文档示例完全同构。

4.3 option_value 的语义与 ProtoPath 语法。 在 CalculatorGraphConfig::Node 中，node_options: 与 option_value: 共同定义一个 calculator 的选项值：

• node_options: 用文本 proto 语法定义一组字面常量；
• 每条 option_value: 用“来自外层图”的信息（具体是外层图 options 的字段值）为某一个 proto 字段赋值。

例如 option_value: "output_tensor_width:options/tensor_width" 表示：calculator 选项字段 ImageToTensorCalculatorOptions.output_tensor_width 取 FaceDetectionOptions.tensor_width 的值。

option_value: 的语法与 input_stream: 类似，形为 option_value: "LHS:RHS"：LHS 定位 calculator 选项字段，RHS 定位 graph option 字段。两侧各由若干以 / 分隔的 proto 字段名组成，用于逐层定位嵌套的 proto 消息与字段，这就是所谓的 “ProtoPath” 语法。注意：LHS 或 RHS 中引用的嵌套消息必须已经在对应的外层 proto 中定义（即被初始化/展开），否则无法沿路径遍历。

若想把整个 graph options 整体拷贝给节点，使用 OPTIONS:options 语法：OPTIONS 指代 graph options，option 指代目标节点/子图 options，二者类型必须相同。

4.4 注册要求与常见错误。 这是一套基于反射的机制，需要 proto descriptor 被显式注册。为此，calculator 或 subgraph 必须依赖 mediapipe_proto_library 生成的 <options_proto>_options_lib target，例如：

mediapipe_proto_library(
    name = "face_detection_proto",
    srcs = ["face_detection.proto"],
    ...
)

mediapipe_simple_subgraph(
    name = "face_detection",
    graph = "face_detection.pbtxt",
    register_as = "FaceDetection",
    deps = [
        ":face_detection_cc_proto",
        ":face_detection_options_lib",      # <-- required
        ...
    ]
)

如果漏掉了 options_lib target，运行时可能出现类似 INVALID_ARGUMENT: Cannot merge field data with data types: 10, 9 的错误——其本质是反射合并时缺少目标类型的 descriptor。相关实现可参考 mediapipe/framework/tool/options_util.cc 与 mediapipe/framework/tool/options_util_test.cc。

五、Cycles：让 MediaPipe 图支持带反馈的循环结构

默认情况下，MediaPipe 要求 calculator 图是无环的，图中出现环会被视为错误。若确实需要循环图（典型场景是累加器、滤波器等带状态反馈的信号处理），必须在图配置中对环进行标注。文档特别说明：该方案当前属于实验性质，接口可能变化。

官方示例是 mediapipe/framework/calculator_graph_test.cc 中的 CalculatorGraphTest.Cycle 单元测试：图中的 adder 节点输出 sum，其值等于 integer source 产生的所有整数之和。该测试展示了循环图支持的要点（文档中的示意图为外部托管图片，本仓库内无对应资源，此处以文本描述代替：integers 源 → IntAdder → sum 输出；sum 经 UnitDelay → old_sum 回流至 IntAdder，构成单环）：

5.1 Back Edge 标注。 要求每个环中至少一条边被标注为 back edge（回边），这样 MediaPipe 移除所有回边之后仍能完成拓扑排序。环内回边的选择通常不止一种；哪条边被标为回边，会影响哪些节点被视为上游/下游，进而影响 MediaPipe 分配给节点的调度优先级。

以 Cycle 测试为例：old_sum 边被标为回边，因此 Delay 节点被视为 adder 的下游节点并获得更高优先级；反之若把 sum 这条进入 delay 的边标为回边，delay 就会被视为 adder 的上游并得到更低优先级。proto 层面的支持见 mediapipe/framework/calculator.proto 中 InputStreamInfo.back_edge 字段（field 2），其注释说明：环通常有明显的正向方向，回边与之相反，形式化定义可参考深度优先搜索（DFS）中的 back edge 概念。

5.2 Initial Packet（初始包）。 为了让 adder 在 integer source 的第一个整数到达时就能运行，old_sum 输入流上必须有一个初始包：值为 0、时间戳与输入相同。这个初始包应当由 delay calculator 在 Open() 方法中输出。

5.3 Loop 中的延迟（Delay）。 每一圈循环都应引入一个延迟，使上一圈的 sum 输出与下一个整数输入对齐。由 delay 节点承担这一职责，因此它必须了解 integer source 时间戳的两个特征：第一个输出的时间戳，以及相邻输出之间的时间戳差值。文档同时提到，MediaPipe 计划提供一种只关心 packet 顺序、忽略时间戳的调度策略以消除这一不便——而这一策略在当前仓库中已有对应实现：同文件中的 CalculatorGraphTest.CycleUntimed 测试（mediapipe/framework/calculator_graph_test.cc）使用 BarrierInputStreamHandler 与 UnitDelayUntimedCalculator，以“忽略时间戳”的方式调度同一累加器图。

5.4 某条输入流结束后提前终止节点。 默认情况下，MediaPipe 在某个非 source calculator 的所有输入流都结束后才调用其 Close()。在循环图例子中，我们希望 integer source 一结束就停止 adder 节点，为此给 adder 节点配置了替代输入流处理器 EarlyCloseInputStreamHandler（输入流处理器的机制见 Synchronization）。

5.5 示例源码。 Cycle 测试中的 Delay calculator 注意两点：Open() 中输出初始包 new int(0) @ Timestamp(0)；Process() 中用 packet.Timestamp().NextAllowedInStream() 对输入包加一个（单位）延迟。该节点假设其输出流与一个时间戳为 0、1、2、3… 的输入流配合使用：

class UnitDelayCalculator : public Calculator {
 public:
  static absl::Status FillExpectations(
      const CalculatorOptions& extendable_options, PacketTypeSet* inputs,
      PacketTypeSet* outputs, PacketTypeSet* input_side_packets) {
    inputs->Index(0)->Set<int>("An integer.");
    outputs->Index(0)->Set<int>("The input delayed by one time unit.");
    return absl::OkStatus();
  }

  absl::Status Open() final {
    Output()->Add(new int(0), Timestamp(0));
    return absl::OkStatus();
  }

  absl::Status Process() final {
    const Packet& packet = Input()->Value();
    Output()->AddPacket(packet.At(packet.Timestamp().NextAllowedInStream()));
    return absl::OkStatus();
  }
};

对应的图配置（注意 back_edge 标注与替代的 input_stream_handler），与 Cycle 测试中的 pbtxt 完全一致：

node {
  calculator: 'GlobalCountSourceCalculator'
  input_side_packet: 'global_counter'
  output_stream: 'integers'
}
node {
  calculator: 'IntAdderCalculator'
  input_stream: 'integers'
  input_stream: 'old_sum'
  input_stream_info: {
    tag_index: ':1'  # 'old_sum'
    back_edge: true
  }
  output_stream: 'sum'
  input_stream_handler {
    input_stream_handler: 'EarlyCloseInputStreamHandler'
  }
}
node {
  calculator: 'UnitDelayCalculator'
  input_stream: 'sum'
  output_stream: 'old_sum'
}

六、工程实践要点与延伸阅读

综合以上各节，在仓库中编写或审查 MediaPipe 图时可以遵循以下实践清单：

1. 拓扑优先用 pbtxt、复杂度上升后考虑 C++ Graph API：简单的线性流水线直接写 CalculatorGraphConfig 文本即可；带条件、模板、动态节点时改用 C++ Graph/Stream 构建（参见 docs/framework_concepts/building_graphs_cpp.md）。
2. 感知方案一律做成 subgraph 注册：通过 mediapipe_simple_subgraph + register_as 暴露组件名（如 FaceDetectionShortRangeCpu），主图按名引用；展开发生在图加载期，运行时语义与性能等价于手写展开图。
3. 对外参数走 graph_options + option_value：图内部节点不要硬编码可变参数，而是声明 graph_options，用 LHS:RHS 的 ProtoPath 从外层选项取值；需要整体传递时用 OPTIONS:options，并确保 deps 中带上 <options_proto>_options_lib，否则会报 INVALID_ARGUMENT: Cannot merge field data with data types: 10, 9。
4. 性能调优用图级字段：num_threads、max_queue_size（默认 100，-1 禁用节流）、ExecutorConfig 独立执行器（移动端重推理节点单独挂载以获得线程局部性）、节点级 buffer_size_hint / max_in_flight，均可在 mediapipe/framework/calculator.proto 中查到语义说明。
5. 需要状态反馈时才引入环：每个环至少一条回边标注 back_edge: true，配合 Open() 初始包、单位延迟对齐，以及对不再需要的节点配置 EarlyCloseInputStreamHandler；不想处理时间戳对齐的场景可参考 CycleUntimed 测试的 BarrierInputStreamHandler 做法。

关键源码索引：图配置 proto 定义 mediapipe/framework/calculator.proto、子图注册与展开 mediapipe/framework/subgraph.h、图选项反射工具 mediapipe/framework/tool/options_util.cc 与 mediapipe/framework/tool/options_util_test.cc、循环图测试 mediapipe/framework/calculator_graph_test.cc、真实子图配置示例 mediapipe/modules/face_detection/face_detection.pbtxt 及其 BUILD。