图表

图表

CalculatorGraphConfig proto 指定 MediaPipe 图的拓扑和功能。图中的每个 node 代表一个特定的计算器或子图，并指定必要的配置，例如注册的计算器/子图类型、输入、输出和可选字段，例如节点特定的选项、输入政策和执行器（如同步中所述）。

CalculatorGraphConfig 具有用于配置全局图级设置的其他几个字段，例如图表执行器配置、线程数和输入流的队列大小上限。一些图表级设置有助于调整图表在不同平台（例如桌面设备和移动设备）上的性能。例如，在移动设备上，将大型模型推断计算器附加到单独的执行器可以提高实时应用的性能，因为这样可以启用线程局部性。

下面是一个简单的 CalculatorGraphConfig 示例，其中有一系列直通式计算器：

# 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"
}

MediaPipe 为复杂图（例如机器学习流水线、处理模型元数据、可选节点等）提供了另一种 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();
}

如需了解详情，请参阅在 C++ 中构建图。

子图

为了将 CalculatorGraphConfig 模块化为子模块，并协助重复使用感知解决方案，可以将 MediaPipe 图定义为 Subgraph。子图的公共接口由一组类似于计算器公共接口的输入和输出流组成。然后，子图可以像计算器一样包含在 CalculatorGraphConfig 中。从 CalculatorGraphConfig 加载 MediaPipe 图时，每个子图节点都会替换为相应的计算器图。因此，子图的语义和性能与对应的计算器图完全相同。

以下示例展示了如何创建名为 TwoPassThroughSubgraph 的子图。

  1. 定义子图。

    # 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"
}

    子图的公共接口包括：

    • 绘制输入流的图像
    • 绘制输出流的图像
    • 绘制输入侧数据包图像
    • 绘制输出端数据包的图表

  2. 使用 BUILD 规则 mediapipe_simple_subgraph 注册子图。参数 register_as 定义新子图的组件名称。

    # 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",
    ],
)

  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"
}

图表选项

您可以为 MediaPipe 图表指定“图选项”protobuf，类似于为 MediaPipe 计算器指定的 Calculator Options protobuf。这些“图表选项”可以在调用图表的位置指定，并用于在图表中填充计算器选项和子图表选项。

在 CalculatorGraphConfig 中，您可以为子图指定图表选项，这与计算器选项完全相同，如下所示：

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
    }
  }
}

在 CalculatorGraphConfig 中，可以接受图表选项并将其用于填充计算器选项，如下所示：

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 接受图表选项 protobuf FaceDetectionOptions。FaceDetectionOptions 用于定义计算器选项 ImageToTensorCalculatorOptions 中的一些字段值和子图选项 InferenceCalculatorOptions 中的一些字段值。字段值使用 option_value: 语法定义。

在 CalculatorGraphConfig::Node protobuf 中，node_options: 和 option_value: 字段共同定义了计算器的选项值，例如 ImageToTensorCalculator。node_options: 字段使用文本 protobuf 语法定义一组字面量常量值。每个 option_value: 字段都使用封闭图中的信息（特别是该图表的图表选项的字段值）定义一个 protobuf 字段的值。在上面的示例中，option_value: "output_tensor_width:options/tensor_width" 使用 FaceDetectionOptions.tensor_width 的值定义 ImageToTensorCalculatorOptions.output_tensor_width 字段。

option_value: 的语法与 input_stream: 的语法类似。语法为 option_value: "LHS:RHS"。LHS 标识计算器选项字段，RHS 标识图表选项字段。更具体地说，LHS 和 RHS 均包含一系列 protobuf 字段名称，用于标识嵌套的 protobuf 消息和用“/”分隔的字段。这称为“ProtoPath”语法。LHS 或 RHS 中引用的嵌套消息必须已在封闭 protobuf 中定义，才能使用 option_value: 遍历。

时光

默认情况下，MediaPipe 要求计算器图表为无循环图表，并将图表中的周期视为错误。如果图表应具有周期，则需要在图表配置中为这些周期添加注释。本页介绍了如何执行此操作。

注意：当前方法是实验性方法，可能会发生变化。我们欢迎您提供反馈。

请使用 mediapipe/framework/calculator_graph_test.cc 中的 CalculatorGraphTest.Cycle 单元测试作为示例代码。下图是测试中的循环图。加法器的 sum 输出是整数源计算器生成的整数的总和。

添加整数流的循环图

这个简单的图表说明了支持循环图的所有问题。

后边缘注释

我们要求将每个周期中的边缘标注为后边缘。这样，在移除所有后边缘后，MediaPipe 的拓扑排序就可以正常运行。

选择背面边缘的方式通常有多种。哪些边缘被标记为后边缘会影响哪些节点被视为上游节点以及哪些节点被视为下游节点，进而影响 MediaPipe 为这些节点分配的优先级。

例如，CalculatorGraphTest.Cycle 测试会将 old_sum 边缘标记为后边缘，因此 Delay 节点会被视为加法器节点的下游节点，并被赋予更高的优先级。或者，我们也可以将延迟节点的 sum 输入标记为返回边缘，在这种情况下，延迟节点将被视为加法器节点的上游节点，并被赋予较低的优先级。

初始数据包

为了使加法器计算器在整数源的第一个整数到达时能够运行，我们需要在加法器的 old_sum 输入流上有一个具有值 0 且时间戳相同的初始数据包。此初始数据包应由 Open() 方法中的延迟计算器输出。

循环中的延迟

每个循环都应引起延迟，以便将前面的 sum 输出与下一个整数输入对齐。延迟节点也会执行此操作。因此，延迟节点需要了解有关整数源计算器的时间戳的以下信息：

  • 第一个输出的时间戳。

  • 连续输出之间的时间戳增量。

我们计划添加一种仅关注数据包排序并忽略数据包时间戳的替代调度政策，这将消除此类不便。

在一个输入流结束后提前终止计算器

默认情况下，MediaPipe 会在所有输入流都完成后调用非来源计算器的 Close() 方法。在示例图表中，我们希望在整数源完成后立即停止添加器节点。这通过使用备用输入流处理程序 EarlyCloseInputStreamHandler 来配置添加器节点。

相关源代码

延迟时间计算器

请注意 Open() 中用于输出初始数据包的代码，以及 Process() 中用于为输入数据包添加（单位）延迟的代码。如上所述，此延迟节点假定其输出流与数据包时间戳为 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。

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'
}