> 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/ko/inference/inference-python/inference-pipeline.md). # 추론 파이프라인 `InferencePipeline` 은(는) 다음의 직접적인 인프로세스 비디오 인터페이스입니다: `추론` Python 패키지입니다. 애플리케이션에 Inference Library를 내장하고 비디오 프레임, 사용자 정의 추론 로직 또는 싱크에 직접 Python으로 접근해야 할 때 사용하세요. Inference Server를 실행하거나 Serverless를 사용하는 애플리케이션의 경우, 다음을 사용하여 모델과 Workflows를 스트리밍하세요. [Inference SDK WebRTC 클라이언트](/reference/ko/inference/inference-sdk/webrtc.md) 스트리밍하세요. ## 빠른 시작 미세 조정된 모델을 Inference와 함께 사용하려면 Roboflow API 키가 필요합니다. 아직 Roboflow 계정이 없다면, [무료 Roboflow 계정에 가입하세요](https://app.roboflow.com). 그런 다음 Roboflow 대시보드에서 API 키를 가져와 코딩 환경에 설정하세요: ```bash export ROBOFLOW_API_KEY= ``` [Roboflow API 키에 대해 자세히 알아보기](/reference/ko/authentication/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 # 설치된 OS의 CUDA 버전에 맞게 --extra-index-url을 조정하세요 ``` 다음으로 Inference Pipeline을 생성하세요: ```python # 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` 작동하는지 ![추론 파이프라인 다이어그램](https://media.roboflow.com/inference/inference-pipeline-diagram.jpg) `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(...)` 응답의 특정 형식에 맞게 조정되어야 합니다. 이렇게 하면 비디오 처리를 원하는 대로 구성할 수 있습니다. ```python # 이것은 예시용 참조 구현입니다 - 목적에 맞게 코드를 조정해야 합니다 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](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, ) # 여기에서는 단일 객체 탐지 모델이 있는 매우 기본적인 워크플로 정의를 볼 수 있습니다. 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`: ```python pipeline = InferencePipeline.init_with_workflow( video_reference="./my_video.mp4", workspace_name="", workflow_id="", on_prediction=workflows_sink, ) ``` {% hint style="success" %} **Workflows 프로파일링.** Workflow 실행을 내부에서 프로파일링할 수 있습니다 `InferencePipeline` 환경 변수를 내보내어 `ENABLE_WORKFLOWS_PROFILING=True`. 추가로, 다음 환경 변수를 통해 프로파일러 버퍼에 보관할 프레임 수를 조정할 수 있습니다 `WORKFLOWS_PROFILER_BUFFER_SIZE`. `init_with_workflow(...)` 또한 다음을 `profiling_directory` 추적을 저장할 위치를 지정하는 매개변수도 받습니다. {% endhint %} ## 싱크 싱크는 Inference Pipeline이 각 예측에 대해 무엇을 해야 하는지 정의합니다. 싱크는 다음 시그니처를 가진 함수입니다: ```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: # 빈 프레임 continue # 일부 처리 ``` 인수는 다음과 같습니다: * `예측`: 여러 비디오 소스를 사용할 때는 dict 목록이 될 수 있는, 모델의 `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: # 여기에 로직을 구현해야 하며, `my_parameter`를 사용합니다 전달하여 pipeline = InferencePipeline.init( video_reference="./my_video.mp4", model_id="rfdetr-small", on_prediction=partial(on_prediction, my_parameter=42), ) ``` ### 사용자 정의 싱크 튜토리얼 사용자 정의 싱크를 단계별로 만들어 보겠습니다. 먼저, 프레임 ID를 출력하는 간단한 싱크입니다: ```python 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() ``` 출력은 대략 다음과 같아야 합니다: ```bash Frame ID: 1 Frame ID: 2 Frame ID: 3 ``` 이제 좀 더 유용한 작업을 해봅시다. 사용자 정의 싱크를 사용해 다음으로 예측을 시각화해 보겠습니다 [Supervision](https://supervision.roboflow.com): ```python 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() ``` 화면에서 다음과 비슷한 것을 볼 수 있어야 합니다: ### 사용자 정의 싱크(고급) 사용자 정의 싱크를 만들려면, 적절한 시그니처를 가진 새 함수를 정의하세요. ```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): # 단일 코드로 순차 처리와 배치 처리를 모두 지원하기 위해 필요합니다 # 한 가지 모드만 사용할 경우 - 하나의 유형만 처리하는 함수를 만들 수 있습니다 # 입력 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`](https://github.com/roboflow/inference/blob/main/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(...)` 적용됩니다. ## 모델 가중치 다운로드 모델 가중치는 추론을 처음 실행할 때 자동으로 다운로드됩니다. 인터넷에 연결된 상태에서 파이프라인을 한 번 초기화하면 가중치를 미리 다운로드할 수 있습니다: ```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/ko/inference/inference-python/offline-weights.md). {% endhint %} ## 기타 파이프라인 구성 Inference Pipeline은 매우 유연하게 구성할 수 있습니다. 구성 옵션에는 다음이 포함됩니다: * `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`](https://github.com/roboflow/inference/blob/main/inference/core/interfaces/stream/inference_pipeline.py).