Inference パイプライン

InferencePipeline を使って動画ストリーム上でモデルを実行します: 動画ソース、カスタム推論ロジック、Workflows、sink。

InferencePipeline は、 推論 Pythonパッケージ内での、プロセス内の直接的なビデオインターフェースです。アプリケーションにInference Libraryを組み込み、ビデオフレーム、カスタム推論ロジック、またはシンクへ直接Pythonからアクセスする必要がある場合に使用します。

Inference Serverを実行するアプリケーション、またはServerlessを使用するアプリケーションでは、 Inference SDK WebRTCクライアント の代わりにストリーミングしてください。

クイックスタート

を使って、モデルやWorkflowsをストリーミングします。Inferenceでファインチューニング済みモデルを使用するには、RoboflowのAPIキーが必要です。まだRoboflowアカウントをお持ちでない場合は、 無料のRoboflowアカウントに登録してください。その後、RoboflowダッシュボードからAPIキーを取得し、コーディング環境で次のように設定します：

export ROBOFLOW_API_KEY=<your api key>

Roboflow APIキーの詳細を見る.

次に、Inferenceをインストールします：

pip install inference

NVIDIA GPU がある場合は、次の方法で推論を高速化できます：

pip install --extra-index-url https://download.pytorch.org/whl/cu124 inference-gpu
# --extra-index-url は、OS にインストールされている CUDA のバージョンに合わせて調整してください

次に、Inference Pipelineを作成します：

# InferencePipelineインターフェースをインポート
from inference import InferencePipeline
# render_boxesという組み込みシンクをインポート（シンクは推論後に実行されるロジックです）
from inference.core.interfaces.stream.sinks import render_boxes

api_key = "YOUR_ROBOFLOW_API_KEY"

# 推論パイプラインオブジェクトを作成
pipeline = InferencePipeline.init(
    # モデルIDをrfdetrモデル（COCOで事前学習済み）に設定
    model_id="rfdetr-large",
    # ビデオ参照（ビデオのソース）を設定します。これは動画ファイルへのリンク/パス、RTSPストリームのURL、
    # またはデバイスIDを表す整数（通常、内蔵Webカメラでは0）にできます
    video_reference="https://storage.googleapis.com/com-roboflow-marketing/inference/people-walking.mp4",
    # 推論結果に対してパイプラインが何をするかを指定します。render_boxesは、動画の上にボックスを描画する組み込みシンクです
    on_prediction=render_boxes,
    # Roboflow APIからモデルを読み込むためにRoboflowのAPIキーを指定
    api_key=api_key,
)

# パイプラインを開始し、ビデオストリームを処理するスレッドに合流します。
pipeline.start()
pipeline.join()

ビデオ参照とは何ですか？

Inference Pipelinesは、さまざまな種類のビデオストリームを取り込むことができます：

• デバイスID（整数）：整数を指定すると、パイプラインはWebカメラのようなローカルデバイスから動画をストリーミングします。通常、内蔵Webカメラはデバイス 0.

• 動画ファイル（文字列）：動画ファイルへのパスを指定すると、パイプラインはファイルから各フレームを読み込み、指定したモデルで推論を実行し、その後 on_prediction メソッドを各予測結果セットごとに実行します。

• 動画URL（文字列）：動画URLを指定することは、動画ファイルのパスを指定するのと同等で、先に動画をダウンロードする必要をなくします。

• RTSP URL（文字列）：RTSP URLを指定すると、パイプラインはRTSPストリームから可能な限り速くフレームを配信し、その後 on_prediction コールバックを、利用可能な最新フレームに対して実行します。

• 要素の リストで、上記の任意の値を含めることができます。

どのように InferencePipeline 動作するか

推論パイプライン図

InferencePipeline は、指定された各ビデオ参照ごとにビデオソースコンシューマスレッドを起動します。動画からのフレームは、 batch_collection_timeout を待機するビデオ多重化装置によって取得されます（ソースがフレームを提供しない場合、より小さいバッチが on_video_frame(...)に渡されますが、欠落したフレームと予測は None で補完されてから on_prediction(...)). on_prediction(...) に渡されます）。 SEQUENTIAL モード（同時に1要素のみ）、または BATCH モード（バッチ要素をすべて同時に）で動作できます。これは sink_mode パラメータで制御されます。

静的な動画ファイルでは、 InferencePipeline がデフォルトですべてのフレームを処理します。ストリームでは、モデル推論が遅い場合にバッファにフレームが蓄積されることがあるため、常に最新データを処理する代わりにバッファからフレームをドロップすることができます（ストリーム処理では古いフレームを捨て、最新のものだけを処理します）。

安定性を高めるため、ストリーム処理時には、処理中に接続が失われるとビデオソースは自動的に再接続されます。これは、パイプラインが長時間稼働し、ソースの停止に対して適切に対応する必要がある本番環境での障害を防ぐことを目的としています。

カスタム推論ロジック

InferencePipeline は、カスタム推論ロジックの実行をサポートします。モデルIDを渡す代わりに、カスタムの呼び出し可能オブジェクトを渡すことができます。この呼び出し可能オブジェクトは VideoFrame を受け取り、処理結果を含む辞書を返す必要があります（ on_video_frame ハンドラとして）。それはモデルの予測結果でも、実行したいその他の任意の処理結果でも構いません。

これは 重要な点ですが 、使用されるシンク（ on_prediction ハンドラ）は、 on_video_frame(...) の特定の形式に合わせて調整する必要があります。これにより、動画処理を思いどおりに形作ることができます。

# これは例であり、参照実装です。目的に合わせてコードを調整する必要があります
import os
import json
from inference.core.interfaces.camera.entities import VideoFrame
from inference import InferencePipeline
from typing import Any, List

TARGET_DIR = "./my_predictions"

class MyModel:

  def __init__(self, weights_path: str):
    self._model = your_model_loader(weights_path)

  # v0.9.18以前
  def infer(self, video_frame: VideoFrame) -> Any:
    return self._model(video_frame.image)

  # v0.9.18以降
  def infer(self, video_frames: List[VideoFrame]) -> List[Any]:
    # 結果は、単一フレームに対するモデル予測を表す要素のリストとして返す必要があります
    # 順序は変更されません。
    return self._model([v.image for v in video_frames])

def save_prediction(prediction: dict, video_frame: VideoFrame) -> None:
  with open(os.path.join(TARGET_DIR, f"{video_frame.frame_id}.json")) as f:
    json.dump(prediction, f)

my_model = MyModel("./my_model.pt")

pipeline = InferencePipeline.init_with_custom_logic(
  video_reference="./my_video.mp4",
  on_video_frame=my_model.infer,
  on_prediction=save_prediction,
)

# パイプラインを開始
pipeline.start()
# パイプラインの完了を待機
pipeline.join()

InferencePipeline Workflowsと

InferencePipeline も実行できます Roboflow Workflowsを、以下に示すとおりです：

from inference import InferencePipeline
from inference.core.interfaces.camera.entities import VideoFrame
from inference.core.interfaces.stream.sinks import render_boxes

def workflows_sink(
    predictions: dict,
    video_frame: VideoFrame,
) -> None:
    render_boxes(
        predictions["predictions"][0],
        video_frame,
        display_statistics=True,
    )


# ここでは、単一の物体検出モデルを使った非常に基本的なワークフロー定義を確認できます。
workflow_specification = {
    "specification": {
        "version": "1.0",
        "inputs": [
            {"type": "InferenceImage", "name": "image"},
        ],
        "steps": [
            {
                "type": "ObjectDetectionModel",
                "name": "step_1",
                "image": "$inputs.image",
                "model_id": "rfdetr-small",
                "confidence": 0.5,
            }
        ],
        "outputs": [
            {"type": "JsonField", "name": "predictions", "selector": "$steps.step_1.*"},
        ],
    }
}
pipeline = InferencePipeline.init_with_workflow(
    video_reference="./my_video.mp4",
    workflow_specification=workflow_specification,
    on_prediction=workflows_sink,
    image_input_name="image",  # 作成するWorkflowImage入力の名前に応じて調整
    video_metadata_input_name="video_metadata" # v0.17.0から利用可能！作成するWorkflowVideoMetadata入力の名前に応じて調整
)

# パイプラインを開始
pipeline.start()
# パイプラインの完了を待機
pipeline.join()

初期化できます InferencePipeline Roboflowアプリに登録されたWorkflowを、次の情報を指定することで workspace_name と workflow_id:

pipeline = InferencePipeline.init_with_workflow(
    video_reference="./my_video.mp4",
    workspace_name="<your_workspace>",
    workflow_id="<your_workflow_id_to_be_found_in_workflow_url>",
    on_prediction=workflows_sink,
)

Workflowsのプロファイリング。 内部でWorkflowの実行をプロファイルできます InferencePipeline を環境変数としてエクスポートすることで ENABLE_WORKFLOWS_PROFILING=True。さらに、環境変数 WORKFLOWS_PROFILER_BUFFER_SIZE. init_with_workflow(...) も profiling_directory トレースを保存する場所を指定するパラメータを受け取ります。

シンク

シンクは、Inference Pipelineが各予測結果に対して何を行うべきかを定義します。シンクは次のシグネチャを持つ関数です：

from typing import Union, List, Optional
from inference.core.interfaces.camera.entities import VideoFrame

def on_prediction(
    predictions: Union[dict, List[Optional[dict]]],
    video_frame: Union[VideoFrame, List[Optional[VideoFrame]]],
) -> None:
    for prediction, frame in zip(predictions, video_frame):
        if prediction is None:
            # 空のフレーム
            continue
        # 何らかの処理

引数は次のとおりです：

• 予測：モデルの infer(...) メソッドの呼び出し結果であるレスポンスオブジェクトを含む辞書（または複数のビデオソースを使用する場合は辞書のリスト）です。

• video_frame： VideoFrame オブジェクト（または VideoFrameのリスト）で、動画フレームのメタデータとピクセルデータを含みます。

使用方法

また on_prediction 、動作を設定する他のパラメータを受け取らせることもできますが、それらは InferencePipeline のinitメソッドへ渡す前に関数クロージャに束縛しておく必要があります。

from functools import partial
from inference.core.interfaces.camera.entities import VideoFrame
from inference import InferencePipeline


def on_prediction(
    predictions: dict,
    video_frame: VideoFrame,
    my_parameter: int,
) -> None:
    # ここでロジックを実装する必要があります。`my_parameter`を使用します
    を渡して

pipeline = InferencePipeline.init(
  video_reference="./my_video.mp4",
  model_id="rfdetr-small",
  on_prediction=partial(on_prediction, my_parameter=42),
)

カスタムシンクのチュートリアル

カスタムシンクを段階的に作成していきましょう。まず、フレームIDを出力するシンプルなシンクです：

from inference import InferencePipeline
# 型ヒント用にVideoFrameをインポート
from inference.core.interfaces.camera.entities import VideoFrame

# シンク関数を定義
def my_custom_sink(predictions: dict, video_frame: VideoFrame):
    # video_frameオブジェクトのフレームIDを出力
    print(f"Frame ID: {video_frame.frame_id}")

pipeline = InferencePipeline.init(
    model_id="rfdetr-large",
    video_reference="https://storage.googleapis.com/com-roboflow-marketing/inference/people-walking.mp4",
    on_prediction=my_custom_sink,
)

pipeline.start()
pipeline.join()

出力は次のようになります：

Frame ID: 1
Frame ID: 2
Frame ID: 3

では、さらに実用的なことをして、カスタムシンクを使って予測を Supervision:

from inference import InferencePipeline
from inference.core.interfaces.camera.entities import VideoFrame

# アノテーション付き画像を表示するためにOpenCVをインポート
import cv2
# 予測の可視化を助けるためにSupervisionをインポート
import supervision as sv

# カスタムシンクで使用するバウンディングボックスアノテータとラベルアノテータを作成
label_annotator = sv.LabelAnnotator()
box_annotator = sv.BoxAnnotator()

def my_custom_sink(predictions: dict, video_frame: VideoFrame):
    # 各予測のテキストラベルを取得
    labels = [p["class"] for p in predictions["predictions"]]
    # 予測をSupervision Detections APIに読み込む
    detections = sv.Detections.from_inference(predictions)
    # Supervisionアノテータ、video_frame、予測結果（Supervision Detectionsとして）、および予測ラベルを使ってフレームを注釈
    image = label_annotator.annotate(
        scene=video_frame.image.copy(), detections=detections, labels=labels
    )
    image = box_annotator.annotate(image, detections=detections)
    # 注釈付き画像を表示
    cv2.imshow("Predictions", image)
    cv2.waitKey(1)

pipeline = InferencePipeline.init(
    model_id="rfdetr-large",
    video_reference="https://storage.googleapis.com/com-roboflow-marketing/inference/people-walking.mp4",
    on_prediction=my_custom_sink,
)

pipeline.start()
pipeline.join()

画面に次のようなものが表示されるはずです：

カスタムシンク（上級）

カスタムシンクを作成するには、適切なシグネチャを持つ新しい関数を定義します。

from typing import Union, List, Optional, Any
from inference.core.interfaces.camera.entities import VideoFrame

def on_prediction(
    predictions: Union[Any, List[Optional[dict]]],
    video_frame: Union[VideoFrame, List[Optional[VideoFrame]]],
) -> None:
    if not issubclass(type(predictions), list):
      # 単一のコードで逐次処理とバッチ処理の両方をサポートするために必要です
      # 1つのモードだけを使う場合は、1種類のみを扱う関数を作成できます
      # 入力
      predictions = [predictions]
      video_frame = [video_frame]
    for prediction, frame in zip(predictions, video_frame):
        if prediction is None:
            # 空のフレーム
            continue
        # 何らかの処理

InferencePipeline は sink_mode パラメータを提供し、予測結果をシンクに渡す方法を制御します。 SinkMode.SEQUENTIALでは、各フレームと予測結果ごとにシンクが個別に呼び出されます。 SinkMode.BATCHでは、フレームと予測結果のリストがシンクに渡され、常にビデオソースの順序に揃えられます。欠落したフレームや予測結果の位置には None 値が入ります。 batch_collection_timeout. SinkMode.ADAPTIVE はデフォルトモードです。単一のビデオ入力では、パイプラインは SinkMode.SEQUENTIALで実行されているかのように動作します。複数の動画を扱う場合、シンクは predictions: List[Optional[dict]] と video_frame: List[Optional[VideoFrame]]を受け取る必要があります。よりシンプルなシンクを使って複数の動画を処理することも可能ですが、その場合は SinkMode.SEQUENTIAL を使用する必要があり、シンクは各予測要素ごとに個別に呼び出されます。

なぜ Optional で List[Optional[dict]] と List[Optional[VideoFrame]]?

すべてのビデオソースからフレームを収集できないことがあります（たとえば、ソースの1つが切断され、再接続が試行される場合など）。 予測 と video_frame は video_reference リストの InferencePipeline順序に一致するように並べられ、 None 要素は欠落したフレームの位置に現れます。この情報はシンクに提供されます。というのも、一部のシンクではバッチ内のすべての予測結果とビデオフレームが（欠落していても）提供される必要があるからです。たとえば、 render_boxes(...) シンクは、タイルモザイク内のフレーム位置を維持するためにこの情報を必要とします。

予測フォーマット

予測結果は、キー 予測を含む辞書としてシンクに渡され、単一フレームまたはフレームのバッチに対する予測結果が入ります。内容は InferencePipelineの背後でどのモデルが動作しているかによって異なります。Roboflowモデルでは、辞書または辞書のリストとして提供されます。

モデル出力によって、予測結果の見え方は異なります。シンクは予測フォーマットに合わせて調整する必要があります。たとえば、Roboflowの物体検出予測には次のキーが含まれます：

• x：予測されたバウンディングボックスの中心x座標（ピクセル）

• y：予測されたバウンディングボックスの中心y座標（ピクセル）

• width：予測されたバウンディングボックスの幅（ピクセル）

• height：予測されたバウンディングボックスの高さ（ピクセル）

• confidence：予測の信頼度値（0〜1）

• class：予測されたクラス名

• class_id：予測されたクラスID

組み込みシンク

Inferenceには、すぐに使える組み込みシンクがいくつかあります（ inference/core/interfaces/stream/sinks.py).

render_boxes(...)

Render Boxesシンクは予測を可視化し、ストリーム上に重ねて表示します。Supervisionのアノテータを使って予測を描画し、注釈付きフレームを表示します。これは検出ベースの出力を返すRoboflowモデル（object-detection, instance-segmentation, keypoint-detection）でのみ動作し、予測のすべての詳細がデフォルトで表示されるわけではありません（検出されたキーポイントなど）。

UDPSink(...)

UDPシンクは、UDPポートを通じて予測結果をブロードキャストします。このポートはクライアントコードでリッスンし、さらなる処理に利用できます。PythonのデフォルトのJSONシリアライズを使用するため、予測結果はシリアライズ可能でなければならず、そうでない場合はエラーが発生します。

multi_sink(...)

multi-sinkは複数のシンクを組み合わせ、1つの推論結果に対して複数のアクションを実行できるようにします。

VideoFileSink(...)

動画ファイルシンクは、 render_boxes(...) シンクと同様に予測を可視化します。ただし、注釈付きフレームを表示する代わりに、動画ファイルとして保存します。 render_boxes(...) に関するすべての制約が適用されます。

モデル重みのダウンロード

モデル重みは、初めて推論を実行したときに自動的にダウンロードされます。インターネットに接続した状態で一度パイプラインを初期化すれば、事前に重みをダウンロードできます：

from inference import InferencePipeline

pipeline = InferencePipeline.init(
    model_id="rfdetr-base",
    video_reference=0,
    on_prediction=lambda predictions, video_frame: None,
    api_key="YOUR_ROBOFLOW_API_KEY",
)

pipeline.start()
pipeline.terminate()

print("モデル重みのダウンロードに成功しました！")

あるいは、 get_model() を使って重みを事前ダウンロードします：

from inference import get_model

get_model("rfdetr-base")

キャッシュディレクトリを確認して、キャッシュ済みモデルを検証できます：

ls -lh /tmp/cache

キャッシュされた各モデルのディレクトリが表示されるはずで、通常はモデルIDで名前が付けられています。

詳細はこちら： 重みのキャッシュ、永続ストレージ、Docker設定.

その他のパイプライン設定

Inference Pipelinesは非常に柔軟に設定できます。設定オプションには次のものがあります：

• max_fps：フレーム処理の最大速度を設定するために使用されます。

• confidence：推論に使用される信頼度しきい値。

• iou_threshold：推論に使用されるIoUしきい値。

• video_source_properties：ビデオソースを設定するためのプロパティの任意の辞書で、cv2 VideoCaptureのプロパティに対応します cv2.CAP_PROP_*。詳しくは OpenCVのドキュメント をご覧ください。利用可能なすべてのプロパティの一覧があります。

from inference import InferencePipeline
pipeline = InferencePipeline.init(
    ...,
    max_fps=10,
    confidence=0.75,
    iou_threshold=0.4,
    video_source_properties={
        "frame_width": 1920.0,
        "frame_height": 1080.0,
        "fps": 30.0,
    },
)

Inference Pipeline のパラメーターの全一覧については、ソースを参照してください： inference/core/interfaces/stream/inference_pipeline.py.