グラフ

グラフ

CalculatorGraphConfig プロトコルは、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 は、複雑なグラフ（ML パイプライン、モデル メタデータの処理、オプションのノードなど）に代替の 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 に含めることができます。MediaPipe グラフが CalculatorGraphConfig から読み込まれると、各サブグラフ ノードが対応する計算機のグラフに置き換えられます。そのため、サブグラフのセマンティクスとパフォーマンスは、対応する計算ツールのグラフと同一になります。

以下に、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 グラフには、「グラフ オプション」プロトコル バッファを指定できます。これは、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: フィールドは、囲むグラフの情報、特に囲むグラフのグラフ オプションのフィールド値からの情報を使用して、1 つの 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 で参照されるネストされたメッセージは、option_value: を使用して走査されるように、包含する protobuf ですでに定義されている必要があります。

季節と時間

デフォルトでは、MediaPipe は電卓グラフを非巡回にする必要があります。また、グラフ内の周期はエラーとして扱います。グラフに周期がある場合、グラフ構成で周期にアノテーションを付ける必要があります。このページではその方法について説明します。

注: 現在のアプローチは試験運用版であり、変更される可能性があります。皆様からのフィードバックをお待ちしております。

サンプルコードとして、mediapipe/framework/calculator_graph_test.cc の CalculatorGraphTest.Cycle 単体テストを使用してください。テストにおける循環グラフを以下に示します。加算器の sum の出力は、整数ソース計算ツールによって生成された整数の合計です。

整数のストリームを追加する巡回グラフ

このシンプルなグラフは、循環グラフのサポートにおけるすべての問題を示しています。

背面エッジ アノテーション

各サイクルのエッジには、バックエッジのアノテーションを付ける必要があります。これにより、後端をすべて削除した後に、MediaPipe のトポロジ並べ替えが機能するようになります。

通常、後端を選択するには複数の方法があります。どのエッジがバックエッジとしてマークされるかは、どのノードがアップストリームとみなされるか、どのノードがダウンストリームとみなされるかに影響します。これは、MediaPipe がノードに割り当てる優先度に影響します。

たとえば、CalculatorGraphTest.Cycle テストでは old_sum エッジがバックエッジとしてマークされるため、Delay ノードは加算ノードのダウンストリーム ノードと見なされ、高い優先度が設定されます。または、遅延ノードへの sum 入力をバックエッジとしてマークすることもできます。この場合、遅延ノードは加算ノードのアップストリーム ノードと見なされ、優先度が低くなります。

最初のパケット

整数ソースから最初の整数が到着したときに加算計算ツールを実行可能にするには、加算器への old_sum 入力ストリームに、値が 0 で同じタイムスタンプを持つ初期パケットが必要です。この初期パケットは、Open() メソッドの遅延計算ツールによって出力されます。

遅延ループ

各ループでは、前の sum 出力を次の整数入力に合わせるために遅延が発生します。これも遅延ノードによって行われます。そのため、遅延ノードは整数ソース計算ツールのタイムスタンプについて次の情報を知る必要があります。

  • 最初の出力のタイムスタンプ。

  • 連続する出力間のタイムスタンプデルタ。

この不便さを解消するため、パケットの順序付けのみを考慮し、パケット タイムスタンプを無視する別のスケジューリング ポリシーを追加する予定です。

1 つの入力ストリームが終了した場合の計算機の早期終了

デフォルトでは、すべての入力ストリームが完了すると、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'
}