추론 पाइ프라인

InferencePipeline으로 비디오 스트림에서 모델을 실행하세요: 비디오 소스, 사용자 정의 추론 로직, Workflows, 싱크.

InferencePipeline 은(는) 다음의 직접적인 인프로세스 비디오 인터페이스입니다: 추론 Python 패키지입니다. 애플리케이션에 Inference Library를 내장하고 비디오 프레임, 사용자 정의 추론 로직 또는 싱크에 직접 Python으로 접근해야 할 때 사용하세요.

Inference Server를 실행하거나 Serverless를 사용하는 애플리케이션의 경우, 다음을 사용하여 모델과 Workflows를 스트리밍하세요. Inference SDK WebRTC 클라이언트 스트리밍하세요.

빠른 시작

미세 조정된 모델을 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
# 설치된 OS의 CUDA 버전에 맞게 --extra-index-url을 조정하세요

다음으로 Inference Pipeline을 생성하세요:

# InferencePipeline 인터페이스를 import
from inference import InferencePipeline
# render_boxes라는 내장 싱크를 import합니다(싱크는 추론 후에 발생하는 로직입니다)
from inference.core.interfaces.stream.sinks import render_boxes

api_key = "YOUR_ROBOFLOW_API_KEY"

# 추론 파이프라인 객체 생성
pipeline = InferencePipeline.init(
    # 모델 ID를 COCO로 사전 학습된 rfdetr 모델로 설정
    model_id="rfdetr-large",
    # 비디오 참조(비디오 소스)를 설정합니다. 비디오 파일 링크/경로, RTSP 스트림 URL,
    # 또는 장치 ID를 나타내는 정수일 수 있습니다(보통 내장 웹캠은 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,
)

# 파이프라인을 시작하고 비디오 스트림을 처리하는 스레드에 join합니다.
pipeline.start()
pipeline.join()

비디오 참조란?

Inference Pipelines는 다양한 유형의 비디오 스트림을 처리할 수 있습니다:

• 장치 ID(정수): 정수를 제공하면 파이프라인이 웹캠 같은 로컬 장치에서 비디오를 스트리밍하도록 지시합니다. 일반적으로 내장 웹캠은 장치 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 모드(한 번에 하나의 요소만) 또는 BATCH 모드(배치 요소 전체를 한 번에)에서 작동할 수 있으며, 이는 sink_mode 매개변수로 제어됩니다.

정적 비디오 파일의 경우, InferencePipeline 기본적으로 모든 프레임을 처리합니다. 스트림의 경우, 항상 가장 최근 데이터를 처리하기 위해 버퍼에서 프레임을 드롭할 수 있습니다(모델 추론이 느리면 더 많은 프레임이 버퍼에 쌓일 수 있으며, 스트림 처리는 오래된 프레임을 버리고 가장 최근 프레임만 처리합니다).

안정성을 높이기 위해 스트림을 처리할 때는 처리 중 연결이 끊기면 비디오 소스가 자동으로 다시 연결됩니다. 이는 파이프라인이 오랜 시간 실행되고 소스 다운타임을 우아하게 처리해야 하는 운영 환경에서 실패를 방지하기 위한 것입니다.

사용자 정의 추론 로직

InferencePipeline 사용자 정의 추론 로직을 실행할 수 있습니다. 모델 ID를 전달하는 대신 사용자 정의 callable을 전달할 수 있습니다. 이 callable은 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
        # 일부 처리

인수는 다음과 같습니다:

• 예측: 여러 비디오 소스를 사용할 때는 dict 목록이 될 수 있는, 모델의 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을 import
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
import cv2
# 예측 시각화를 돕기 위해 supervision을 import
import supervision as sv

# 사용자 정의 싱크에서 사용할 바운딩 박스 annotator와 label annotator를 생성
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 annotator, 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):
      # 단일 코드로 순차 처리와 배치 처리를 모두 지원하기 위해 필요합니다
      # 한 가지 모드만 사용할 경우 - 하나의 유형만 처리하는 함수를 만들 수 있습니다
      # 입력
      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]]?

비디오 소스 전부에서 비디오 프레임을 수집할 수 없는 경우가 발생할 수 있습니다(예: 소스 중 하나가 연결 해제되고 재연결을 시도하는 경우). 예측 및 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 annotator를 사용해 예측을 렌더링하고 주석이 달린 프레임을 표시합니다. 탐지 기반 출력을 내는 Roboflow 모델에만 작동하며(객체 감지, 인스턴스 분할, 키포인트 감지), 기본적으로 예측의 모든 세부 정보(예: 감지된 키포인트)가 표시되지는 않을 수 있습니다.

UDPSink(...)

UDP 싱크는 UDP 포트를 통해 예측을 브로드캐스트합니다. 이 포트는 후속 처리를 위해 클라이언트 코드에서 수신할 수 있습니다. Python의 기본 JSON 직렬화를 사용하므로, 예측은 직렬화 가능해야 하며 그렇지 않으면 오류가 발생합니다.

multi_sink(...)

multi-sink는 여러 싱크를 결합하여 하나의 추론 결과에 대해 여러 작업이 수행될 수 있게 합니다.

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 Pipeline은 매우 유연하게 구성할 수 있습니다. 구성 옵션에는 다음이 포함됩니다:

• 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.