> For the complete documentation index, see [llms.txt](https://docs.roboflow.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.roboflow.com/reference/ja/inference/inference-python/inference-pipeline.md).

# Inference Pipeline

InferencePipeline で動画ストリーム上のモデルを実行: 動画ソース、カスタム推論ロジック、Workflows、シンク。

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

セルフホストの Inference Server を実行するアプリケーションでは、以下を使用してください [Inference SDK WebRTC クライアント](/reference/ja/inference/inference-sdk/webrtc.md) 代わりに使用してください。新しいマネージドのライブカメラ処理では、以下を使用してください [クラウド ストリーミング](https://docs.roboflow.com/deployment/roboflow-cloud/cloud-streaming).

## クイックスタート

Inference でファインチューニング済みモデルを使うには、Roboflow API キーが必要です。まだ Roboflow アカウントをお持ちでない場合は、 [無料の Roboflow アカウントに登録してください](https://app.roboflow.com)。次に、Roboflow ダッシュボードから API キーを取得し、コーディング環境で設定します：

```bash
export ROBOFLOW_API_KEY=<your api key>
```

[Roboflow API キーについて詳しく学ぶ](/reference/ja/ren-zheng/authentication/find-your-roboflow-api-key.md).

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

```bash
pip install inference
```

NVIDIA GPU がある場合は、以下で推論を高速化できます：

```bash
pip install --extra-index-url https://download.pytorch.org/whl/cu124 inference-gpu
# please adjust the --extra-index-url to the CUDA version installed in your OS
```

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

```python
# import the InferencePipeline interface
from inference import InferencePipeline
# import a built-in sink called render_boxes (sinks are the logic that happens after inference)
from inference.core.interfaces.stream.sinks import render_boxes

api_key = "YOUR_ROBOFLOW_API_KEY"

# Create an inference pipeline object
pipeline = InferencePipeline.init(
    # set the model id to an rfdetr model (pre-trained on COCO)
    model_id="rfdetr-large",
    # set the video reference (source of video), it can be a link/path to a video file, an RTSP stream url,
    # or an integer representing a device id (usually 0 for built in webcams)
    video_reference="https://storage.googleapis.com/com-roboflow-marketing/inference/people-walking.mp4",
    # tell the pipeline what to do with inference results. render_boxes is a built-in sink that renders boxes on top of the video
    on_prediction=render_boxes,
    # provide your roboflow api key for loading models from the roboflow api
    api_key=api_key,
)

# Start the pipeline and join the thread that processes the video stream.
pipeline.start()
pipeline.join()
```

## video reference とは何ですか？

Inference Pipeline は、さまざまな種類のビデオストリームを取り込めます：

* **デバイス ID（整数）**：整数を指定すると、パイプラインはウェブカメラのようなローカルデバイスからビデオをストリーミングします。通常、内蔵ウェブカメラはデバイスとして表示されます `0`.
* **ビデオファイル（文字列）**：ビデオファイルのパスを指定すると、パイプラインはファイルからすべてのフレームを読み込み、指定したモデルで推論を実行し、その後 `on_prediction` メソッドを各予測結果に対して実行します。
* **ビデオ URL（文字列）**：ビデオ URL を指定することは、ビデオファイルのパスを指定するのと同等で、先に動画をダウンロードする必要がなくなります。
* **RTSP URL（文字列）**：RTSP URL を指定すると、パイプラインは RTSP ストリームから可能な限り高速にフレームをストリーミングし、その後 `on_prediction` 最新の利用可能なフレームに対してコールバックを実行します。
* **リスト** 。要素は、上で説明したいずれの値でもかまいません。

## 動作 `InferencePipeline` の仕組み

![推論パイプラインの図](https://media.roboflow.com/inference/inference-pipeline-diagram.jpg)

`InferencePipeline` は、指定された各 video reference ごとにビデオソース消費スレッドを起動します。ビデオからのフレームは、以下を待機するビデオ多重化装置によって取得されます `batch_collection_timeout` （ソースがフレームを提供しない場合、より小さいバッチが以下に渡されます `on_video_frame(...)`、ただし欠落したフレームと予測には以下が補われます `None` 以下に渡す前に `on_prediction(...)`). `on_prediction(...)` 以下のモードで動作できます `SEQUENTIAL` モード（1回につき要素は1つだけ）、または `BATCH` モード（バッチ要素を一度にすべて処理）；これは以下によって制御されます `sink_mode` パラメータ。

静的なビデオファイルでは、 `InferencePipeline` デフォルトではすべてのフレームを処理します。ストリームの場合は、常に最新データを処理することを優先してバッファからフレームをドロップすることができます（モデル推論が遅い場合、より多くのフレームがバッファに蓄積されることがあります。ストリーム処理では古いフレームを破棄し、最新のフレームだけを処理します）。

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

## カスタム推論ロジック

`InferencePipeline` カスタム推論ロジックを実行できます。モデル ID の代わりに、カスタムの callable を渡すことができます。この callable は、以下を受け取る必要があります `VideoFrame` と、処理結果を含む辞書を返します（以下の `on_video_frame` ハンドラとして）。これはモデルの予測結果でも、実行したい他の任意の処理結果でもかまいません。

これは **重要な点です** 使用している sink（以下の `on_prediction` ハンドラ）は、以下の特定の形式に合わせて調整する必要があることです `on_video_frame(...)` 。このようにして、ビデオ処理を自由に形作ることができます。

```python
# This is example, reference implementation - you need to adjust the code to your purposes
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)

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

  # after v0.9.18
  def infer(self, video_frames: List[VideoFrame]) -> List[Any]:
    # result must be returned as list of elements representing model prediction for single frame
    # with order unchanged.
    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,
)

# start the pipeline
pipeline.start()
# wait for the pipeline to finish
pipeline.join()
```

## `InferencePipeline` Workflows を使って

`InferencePipeline` も実行できます [Roboflow Workflows](https://docs.roboflow.com/workflows)、以下に示すように：

```python
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,
    )


# here you may find very basic definition of workflow - with a single object detection model.
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",  # adjust according to name of WorkflowImage input you define
    video_metadata_input_name="video_metadata" # AVAILABLE from v0.17.0! adjust according to name of WorkflowVideoMetadata input you define
)

# start the pipeline
pipeline.start()
# wait for the pipeline to finish
pipeline.join()
```

初期化できます `InferencePipeline` Roboflow アプリに登録された Workflow を、以下を指定して初期化できます `workspace_name` および `workflow_id`:

```python
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,
)
```

{% hint style="success" %}
**Workflows のプロファイリング。** Workflow の実行を以下の中でプロファイリングできます `InferencePipeline` 環境変数を設定して `ENABLE_WORKFLOWS_PROFILING=True`。さらに、環境変数でプロファイラバッファに保持するフレーム数を調整できます `WORKFLOWS_PROFILER_BUFFER_SIZE`. `init_with_workflow(...)` は以下の `profiling_directory` パラメータも受け取り、トレースの保存先を指定します。
{% endhint %}

## Sink

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

```python
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:
            # EMPTY FRAME
            continue
        # SOME PROCESSING
```

引数は次のとおりです：

* `predictions`：モデルの infer(...) メソッドの呼び出しによって得られる応答オブジェクトである辞書（または複数のビデオソースを使用する場合は辞書のリスト）です。 `infer(...)` メソッドです。
* `video_frame`：以下の `VideoFrame` オブジェクト（または `VideoFrame`のリスト）で、ビデオフレームのメタデータとピクセルデータを含みます。

### 使用方法

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

```python
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:
    # you need to implement your logic here, with `my_parameter` used
    pass

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

### カスタム sink チュートリアル

カスタム sink をステップごとに作成していきましょう。まずは、フレーム ID を出力するシンプルな sink です：

```python
from inference import InferencePipeline
# import VideoFrame for type hinting
from inference.core.interfaces.camera.entities import VideoFrame

# define sink function
def my_custom_sink(predictions: dict, video_frame: VideoFrame):
    # print the frame ID of the video_frame object
    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()
```

出力は次のようになるはずです：

```bash
Frame ID: 1
Frame ID: 2
Frame ID: 3
```

では、もっと実用的なことをして、カスタム sink を使って以下で予測結果を可視化しましょう [Supervision](https://supervision.roboflow.com):

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

# import opencv to display our annotated images
import cv2
# import supervision to help visualize our predictions
import supervision as sv

# create a bounding box annotator and label annotator to use in our custom sink
label_annotator = sv.LabelAnnotator()
box_annotator = sv.BoxAnnotator()

def my_custom_sink(predictions: dict, video_frame: VideoFrame):
    # get the text labels for each prediction
    labels = [p["class"] for p in predictions["predictions"]]
    # load our predictions into the Supervision Detections api
    detections = sv.Detections.from_inference(predictions)
    # annotate the frame using our supervision annotator, the video_frame, the predictions (as supervision Detections), and the prediction labels
    image = label_annotator.annotate(
        scene=video_frame.image.copy(), detections=detections, labels=labels
    )
    image = box_annotator.annotate(image, detections=detections)
    # display the annotated image
    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()
```

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

### カスタム sink（上級）

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

```python
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):
      # this is required to support both sequential and batch processing with single code
      # if you use only one mode - you may create function that handles with only one type
      # of input
      predictions = [predictions]
      video_frame = [video_frame]
    for prediction, frame in zip(predictions, video_frame):
        if prediction is None:
            # EMPTY FRAME
            continue
        # SOME PROCESSING
```

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

#### なぜ `Optional` に `List[Optional[dict]]` および `List[Optional[VideoFrame]]`?

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

**予測フォーマット**

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

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

* `x`：予測されたバウンディングボックスの中心 x 座標（ピクセル）
* `y`：予測されたバウンディングボックスの中心 y 座標（ピクセル）
* `width`：予測されたバウンディングボックスの幅（ピクセル）
* `height`：予測されたバウンディングボックスの高さ（ピクセル）
* `confidence`：予測の信頼度（0 から 1 の間）
* `class`：予測されたクラス名
* `class_id`：予測されたクラス ID

### 組み込み sink

Inference には、すぐに使える組み込み sink がいくつかあります（以下を参照してください `inference/core/interfaces/stream/sinks.py` 以下の [Inference リポジトリ](https://github.com/roboflow/inference)).

#### `render_boxes(...)`

render boxes sink は、予測結果を可視化してストリーム上に重ねて表示します。Supervision の annotator を使用して予測を描画し、注釈付きフレームを表示します。これは、検出ベースの出力（`object-detection`, `instance-segmentation`, `keypoint-detection`）を出力する Roboflow モデルでのみ機能し、予測の詳細のすべてがデフォルトで表示されるわけではありません（検出されたキーポイントなど）。

#### `UDPSink(...)`

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

#### `multi_sink(...)`

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

#### `VideoFileSink(...)`

video file sink は、 `render_boxes(...)` sink と同様に予測結果を可視化します。ただし、注釈付きフレームを表示する代わりに、ビデオファイルに保存します。以下に関連するすべての制約が `render_boxes(...)` 適用されます。

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

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

```python
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()` を使って重みを事前にダウンロードします:

```python
from inference import get_model

get_model("rfdetr-base")
```

キャッシュディレクトリを確認することで、キャッシュされたモデルを検証できます:

```bash
ls -lh /tmp/cache
```

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

{% hint style="success" %}
詳細はこちら [重みのキャッシュ、永続ストレージ、Docker設定](/reference/ja/inference/inference-python/offline-weights.md).
{% endhint %}

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

推論パイプラインは非常に柔軟に設定できます。設定オプションには以下が含まれます:

* `max_fps`：フレーム処理の最大速度を設定するために使用します。
* `confidence`：推論に使用される信頼度閾値。
* `iou_threshold`：推論に使用されるIoU閾値。
* `video_source_properties`：ビデオソースを設定するためのプロパティを含む、任意の辞書。cv2 VideoCapture のプロパティに対応します `cv2.CAP_PROP_*`。詳細は [OpenCVのドキュメント](https://docs.opencv.org/4.x/d4/d15/group__videoio__flags__base.html#gaeb8dd9c89c10a5c63c139bf7c4f5704d) をご覧ください。利用可能なすべてのプロパティの一覧があります。

```python
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` 以下の [Inference リポジトリ](https://github.com/roboflow/inference).