동적 Python 블록

실행 엔진이 런타임에 해석하는 Python 코드로 Workflow 정의 안에 블록을 직접 정의합니다.

이 기능에 대한 제품 수준 소개는 다음을 참조하세요 사용자 정의 블록. 인라인 코드 대신 전체 Block 클래스를 작성하려면 다음을 참조하세요 Workflow 블록 만들기.

Workflow 정의의 구문이 정의되었을 때윤곽이 잡혔지만, 한 가지 핵심적인 측면은 다루어지지 않았습니다: Workflow 정의 자체 내에서 블록을 직접 정의할 수 있는 기능입니다. 이 섹션에는 제자리에서 정의된 블록에 대한 manifest와 Python 코드를 포함할 수 있으며, 이는 Execution Engine에 의해 동적으로 해석됩니다. 이러한 인플레이스 블록은 정적으로 정의된 블록과 유사하게 작동하지만 플러그인 훨씬 더 큰 유연성을 제공합니다.

실행 모드

동적 Python 블록은 두 가지 실행 모드를 지원합니다:

로컬 실행

자신의 하드웨어에서 로컬로 추론을 실행할 때, 동적 블록은 환경 내에서 직접 실행됩니다. 이는 개발 및 테스트에 가장 빠른 성능을 제공합니다.

동적 블록의 로컬 실행은 로컬 배포 환경에서만 작동합니다 inference 그리고 신뢰할 수 없는 코드를 실행할 때는 보안 영향을 신중히 고려해야 합니다.

이 기능을 비활성화하려면, export ALLOW_CUSTOM_PYTHON_EXECUTION_IN_WORKFLOWS=False

클라우드 실행 (Roboflow Serverless v2)

Roboflow의 클라우드 인프라와 Serverless v2 API를 사용할 때, 동적 블록은 안전하고 격리된 컨테이너에서 실행됩니다. 이를 통해 인프라를 손상시키지 않으면서 사용자 정의 코드를 안전하게 실행할 수 있습니다.

데이터 직렬화 요구 사항

클라우드 실행을 사용할 때는 모든 입력 및 출력 데이터가 Inference의 직렬화 시스템을 통해 직렬화 가능해야 합니다. 이는 다음을 의미합니다:

간단한 Python 타입(str, int, float, bool, list, dict)을 사용하세요
Numpy 배열과 표준 컴퓨터 비전 데이터 구조가 지원됩니다
복잡한 사용자 정의 객체는 더 단순한 표현으로 변환해야 할 수 있습니다
함수, lambda 표현식 또는 기타 직렬화 불가능한 Python 객체를 반환하지 마세요

클라우드 실행 환경은 로컬 실행과 동일한 표준 라이브러리와 import를 제공하므로, 코드가 두 모드 모두에서 일관되게 작동하도록 보장합니다.

상태 관리 및 공유 데이터

모듈 수준에서 정의된 변수(당신의 run 함수 바깥에 있는) 블록 코드의 변수들은 해당 블록 인스턴스 범위에 속합니다. 이러한 변수는:

호출 간에 유지됩니다 동일한 블록의 (코드가 변경되지 않는 한)
블록의 코드가 변경되면 재설정됩니다 블록 코드에 대한 어떠한 수정도 새 네임스페이스를 생성합니다
서버/컨테이너가 재시작되면 사라집니다

예시:

이 블록은 실행될 때마다 카운터를 증가시키고 마지막 결과를 기억합니다:

# 이 변수는 블록 범위입니다
counter = 0
last_result = None

def run(self, input_value):
    global counter, last_result
    
    counter += 1
    
    # 비교를 위해 마지막 결과를 저장
    previous = last_result
    last_result = input_value * 2
    
    return {
        "run_count": counter,
        "current": last_result,
        "previous": previous
    }

상태 관리 모범 사례

사용자 정의 블록 상태는 비용이 큰 계산의 캐싱과 아티팩트 및 종속성 로딩 최적화를 위해 사용됩니다.

중요한 데이터 영속성을 상태에 의존하지 마세요. 중요한 데이터에는 외부 저장소를 사용하세요.
서버 재시작 또는 컨테이너 스케일링으로 인해 상태가 언제든지 손실될 수 있습니다
클라우드 환경에서는 이후 요청이 서로 다른 상태를 가진 다른 서버에 도달할 수 있습니다
새로 시작할 때를 처리할 수 있도록 블록 범위 변수를 기본값으로 초기화하세요
상태는 가볍게 유지하세요. 큰 객체는 메모리를 소비하고 성능에 영향을 줄 수 있습니다.

이론

동적 Python 블록 기능에 대한 고수준 개요:

사용자가 JSON으로 동적 블록 정의를 제공합니다
정의에는 Execution Engine이 구성하는 데 필요한 정보가 포함됩니다 WorkflowBlockManifest 및 WorkflowBlock 문서에서
런타임에서 Compiler는 정의를 동적으로 생성된 Python 클래스로 변환합니다 - 정적으로 정의된 블록과 정확히 동일합니다
Workflow 정의에서, 동적 블록을 표준 정적 블록인 것처럼 사용하는 단계를 선언할 수 있습니다

예시

동적 Python 블록이 포함된 예시 워크플로를 살펴보고 논의해 봅시다.

동적 블록이 있는 워크플로

{
    "version": "1.0",
    "inputs": [
        {
            "type": "WorkflowImage",
            "name": "image"
        }
    ],
    "dynamic_blocks_definitions": [
        {
            "type": "DynamicBlockDefinition",
            "manifest": {
                "type": "ManifestDescription",
                "block_type": "OverlapMeasurement",
                "inputs": {
                    "predictions": {
                        "type": "DynamicInputDefinition",
                        "selector_types": [
                            "step_output"
                        ]
                    },
                    "class_x": {
                        "type": "DynamicInputDefinition",
                        "value_types": [
                            "string"
                        ]
                    },
                    "class_y": {
                        "type": "DynamicInputDefinition",
                        "value_types": [
                            "string"
                        ]
                    }
                },
                "outputs": {
                    "overlap": {
                        "type": "DynamicOutputDefinition",
                        "kind": []
                    }
                }
            },
            "code": {
                "type": "PythonCode",
                "run_function_code": "\ndef run(self, predictions: sv.Detections, class_x: str, class_y: str) -> BlockResult:\n    bboxes_class_x = predictions[predictions.data[\"class_name\"] == class_x]\n    bboxes_class_y = predictions[predictions.data[\"class_name\"] == class_y]\n    overlap = []\n    for bbox_x in bboxes_class_x:\n        bbox_x_coords = bbox_x[0]\n        bbox_overlaps = []\n        for bbox_y in bboxes_class_y:\n            if bbox_y[-1][\"detection_id\"] == bbox_x[-1][\"detection_id\"]:\n                continue\n            bbox_y_coords = bbox_y[0]\n            x_min = max(bbox_x_coords[0], bbox_y_coords[0])\n            y_min = max(bbox_x_coords[1], bbox_y_coords[1])\n            x_max = min(bbox_x_coords[2], bbox_y_coords[2])\n            y_max = min(bbox_x_coords[3], bbox_y_coords[3])\n            # 교차 사각형의 면적을 계산\n            intersection_area = max(0, x_max - x_min + 1) * max(0, y_max - y_min + 1)\n            box_x_area = (bbox_x_coords[2] - bbox_x_coords[0] + 1) * (bbox_x_coords[3] - bbox_x_coords[1] + 1)\n            local_overlap = intersection_area / (box_x_area + 1e-5)\n            bbox_overlaps.append(local_overlap)\n        overlap.append(bbox_overlaps)\n    return  {\"overlap\": overlap}\n"
            }
        },
        {
            "type": "DynamicBlockDefinition",
            "manifest": {
                "type": "ManifestDescription",
                "block_type": "MaximumOverlap",
                "inputs": {
                    "overlaps": {
                        "type": "DynamicInputDefinition",
                        "selector_types": [
                            "step_output"
                        ]
                    }
                },
                "outputs": {
                    "max_value": {
                        "type": "DynamicOutputDefinition",
                        "kind": []
                    }
                }
            },
            "code": {
                "type": "PythonCode",
                "run_function_code": "\ndef run(self, overlaps: List[List[float]]) -> BlockResult:\n    max_value = -1\n    for overlap in overlaps:\n        for overlap_value in overlap:\n            if not max_value:\n                max_value = overlap_value\n            else:\n                max_value = max(max_value, overlap_value)\n    return {\"max_value\": max_value}\n"
            }
        }
    ],
    "steps": [
        {
            "type": "RoboflowObjectDetectionModel",
            "name": "model",
            "image": "$inputs.image",
            "model_id": "rfdetr-small"
        },
        {
            "type": "OverlapMeasurement",
            "name": "overlap_measurement",
            "predictions": "$steps.model.predictions",
            "class_x": "dog",
            "class_y": "dog"
        },
        {
            "type": "ContinueIf",
            "name": "continue_if",
            "condition_statement": {
                "type": "StatementGroup",
                "statements": [
                    {
                        "type": "BinaryStatement",
                        "left_operand": {
                            "type": "DynamicOperand",
                            "operand_name": "overlaps",
                            "operations": [
                                {
                                    "type": "SequenceLength"
                                }
                            ]
                        },
                        "comparator": {
                            "type": "(Number) >="
                        },
                        "right_operand": {
                            "type": "StaticOperand",
                            "value": 1
                        }
                    }
                ]
            },
            "evaluation_parameters": {
                "overlaps": "$steps.overlap_measurement.overlap"
            },
            "next_steps": [
                "$steps.maximum_overlap"
            ]
        },
        {
            "type": "MaximumOverlap",
            "name": "maximum_overlap",
            "overlaps": "$steps.overlap_measurement.overlap"
        }
    ],
    "outputs": [
        {
            "type": "JsonField",
            "name": "overlaps",
            "selector": "$steps.overlap_measurement.overlap"
        },
        {
            "type": "JsonField",
            "name": "max_overlap",
            "selector": "$steps.maximum_overlap.max_value"
        }
    ]
}

다음부터 분석을 시작해 봅시다 dynamic_blocks_definitions - 이는 동적 블록 목록을 제공하는 Workflow Definition의 일부입니다. 각 블록은 두 섹션으로 구성됩니다:

manifest - 다음의 JSON 표현을 제공합니다 BlockManifest - 참조 블록 개발 가이드
code - Python 코드를 제공합니다

블록 manifest 정의

manifest 정의에는 다음을 포함한 여러 필드가 있습니다:

block_type - 다음과 동일합니다 type block manifest의 필드 - 고유한 블록 식별자를 제공해야 합니다
inputs - 이름과 동적 입력 정의가 있는 사전
outputs - 이름과 동적 출력 정의가 있는 사전
output_dimensionality_offset - 필드는 출력 차원을 지정합니다
accepts_batch_input - 필드는 런타임의 입력 데이터가 Execution Engine에 의해 배치로 제공될지 여부를 결정합니다
accepts_empty_values - 단계 입력을 구성할 때 빈 입력을 무시할지 결정하는 필드

의문이 있을 경우 다음을 참조하세요 블록 개발 가이드동적 블록이 표준 블록의 기능을 복제하기 때문입니다.

동적 입력 정의

동적 입력은 동적으로 생성된 블록 manifest의 필드를 정의합니다. 다시 말해, 이는 다음이 생성될 기준이 되는 정의입니다 BlockManifest 런타임에서 class가 생성됩니다.

각 입력은 다음 속성을 정의할 수 있습니다:

has_default_value - 동적 manifest 필드에 기본값이 있는지 결정하는 플래그
default_value - 기본값 (다음 경우에만 사용됨 has_default_value=True
is_optional - 동적 manifest 필드가 선택 사항인지 결정하는 플래그
is_dimensionality_reference - 동적 manifest 필드가 런타임에서 차원 참조로 사용될 selector를 제공해야 하는지 결정하는 플래그
dimensionality_offset - 동적 manifest의 구성된 입력 속성에 대한 차원 오프셋
selector_types - 속성에서 사용할 수 있는 selector 유형(다음 중 하나 input_image, step_output_image, input_parameter, step_output). Step은 selector를 보유하지 않을 수 있지만, 그 경우 특정 유형의 정의를 제공해야 합니다.
selector_data_kind - 각 selector 유형에 특정한 selector kind 목록이 있는 사전
value_types - manifest에 배치될 특정 유형의 정의 - 이 필드는 Python 타입에 대해 동적으로 생성된 manifest 필드의 형식을 지정합니다. 선택 가능한 유형: 아무, 정수, float, boolean, dict, list, strig

동적 출력 정의

출력 정의는 매우 간단하며, 해당 출력에 대해 선언된 선택적 목록을 보유합니다. kinds 주어진 출력에 대해 선언됩니다.

Python 코드 정의

Python 코드는 다음 필드를 가진 JSON 문서로 제공됩니다:

run_function_code - 다음의 코드 run(...) 동적 블록의 메서드
run_function_name - run 함수의 이름
init_function_code - step 상태를 구성할 init 함수에 대한 선택적 코드 - dictionary를 반환해야 하며, 이는 다음에서 사용할 수 있습니다 run() 함수 아래의 self._init_results
init_function_name - init 함수의 이름
imports - 추가 import 목록 (환경에 있는 라이브러리만 사용할 수 있으며, 종속성은 자동으로 설치되지 않습니다)

어떻게 작성하나요 `run(...)` 메서드?

다음 사항을 알아야 합니다:

run(...) 함수는 클래스 인스턴스 메서드인 것처럼 정의되어야 하며 - 첫 번째 인수는 self 이고 나머지 인수는 동적 블록 정의에 선언된 동적 블록 manifest와 호환되어야 합니다
import 문과 다음 항목을 포함하여 기본 심볼이 제공된다고 예상해야 합니다:

from typing import Any, List, Dict, Set, Optional
import supervision as sv
import numpy as np
import math
import time
import json
import os
import requests
import cv2
import shapely
from inference.core.workflows.execution_engine.entities.base import Batch, WorkflowImageData
from inference.core.workflows.prototypes.block import BlockResult

예를 들어 함수는 다음과 같이 보일 수 있습니다(명확성을 위해, 여기서는 Python 코드를 보기 좋게 서식 지정하여 제공합니다. 하지만 정의에 넣으려면 코드를 문자열로 만들어야 합니다):

def run(self, predictions: sv.Detections, class_x: str, class_y: str) -> BlockResult:
    bboxes_class_x = predictions[predictions.data["class_name"] == class_x]
    bboxes_class_y = predictions[predictions.data["class_name"] == class_y]
    overlap = []
    for bbox_x in bboxes_class_x:
        bbox_x_coords = bbox_x[0]
        bbox_overlaps = []
        for bbox_y in bboxes_class_y:
            if bbox_y[-1]["detection_id"] == bbox_x[-1]["detection_id"]:
                continue
            bbox_y_coords = bbox_y[0]
            x_min = max(bbox_x_coords[0], bbox_y_coords[0])
            y_min = max(bbox_x_coords[1], bbox_y_coords[1])
            x_max = min(bbox_x_coords[2], bbox_y_coords[2])
            y_max = min(bbox_x_coords[3], bbox_y_coords[3])
            # 교차 사각형의 면적을 계산
            intersection_area = max(0, x_max - x_min + 1) * max(0, y_max - y_min + 1)
            box_x_area = (bbox_x_coords[2] - bbox_x_coords[0] + 1) * (bbox_x_coords[3] - bbox_x_coords[1] + 1)
            local_overlap = intersection_area / (box_x_area + 1e-5)
            bbox_overlaps.append(local_overlap)
        overlap.append(bbox_overlaps)
    return  {"overlap": overlap}

어떻게 작성하나요 `init(...)` 메서드?

init 함수는 다음을 구성해야 합니다 self._init_results 사전.

예시:


def my_init() -> Dict[str, Any]:
    return {"some": "value"}

단계로서의 동적 Python 블록 사용

예시 Workflow 정의에 표시된 것처럼, 정적 플러그인을 통해 노출되는 정상적인 블록인 것처럼 단순히 해당 블록을 사용할 수 있습니다:

{
    "type": "OverlapMeasurement",
    "name": "overlap_measurement",
    "predictions": "$steps.model.predictions",
    "class_x": "dog",
    "class_y": "dog"
}

동적 블록 디버깅

설정 debug=True 다음에서 /workflows/run 사용자 정의 Python 블록의 진단 정보를 캡처하기 위한 요청입니다. 이는 선택 사항입니다(미리보기 실행에서는 암묵적으로 활성화되지 않습니다).

stdout/stderr 캡처. 블록이 출력하는 모든 내용은 단계별로 캡처됩니다.

구조화된 추적 정보 내보내기. A debug_traces 헬퍼는 당신의 run(...) 코드에서 사용할 수 있습니다(기본 심볼로, 표준 import와 함께 주입됩니다). JSON 직렬화 가능한 값을 추가하세요; 항목에 타임스탬프를 추가하려면 add_timestamp=True 를 전달하세요:

def run(self, value: int) -> BlockResult:
    debug_traces.append({"received": value})
    debug_traces.append(value * 2, add_timestamp=True)
    return {"result": value * 2}

이 debug 가 활성화되지 않았거나(또는 Modal / OCI 샌드박스 실행 환경에서는), debug_traces.append(...) 안전한 no-op이며 수집되지 않습니다.

성공적인 실행(HTTP 200). 응답에는 다음이 포함됩니다:

python_blocks_output_streams - 단계 이름으로 키가 지정된 캡처된 stdout/stderr, 예: {"my_step": [{"stdout": "...", "stderr": null}]}.
python_blocks_debug_traces - 실행 순서대로 추가된 항목, 예: [{"step": "my_step", "value": {"received": 7}, "timestamp": "...", "timestamp_timezone": "UTC"}] (timestamp* 다음의 경우에만 존재합니다 add_timestamp=True).

둘 다 null 때 debug 가 꺼져 있거나 아무 것도 캡처되지 않았습니다. 로컬 실행에서만 채워집니다.

실패한 실행(HTTP 400). 오류 응답에는 동일한 두 필드 실패 전에 실행되어(그리고 출력/추적을 생성한) 단계가 생성한 부분 출력/추적과 함께, blocks_errors[].block_traceback 실패한 단계의 자체 stdout/stderr가 포함됩니다.

이전블록을 플러그인으로 묶기 다음블록 테스트

마지막 업데이트 5일 전

도움이 되었나요?