워크플로 블록 만들기

워크플로 블록 작성에 대한 완전한 개발자 가이드입니다: 매니페스트, 입력, 출력, 배치 처리, 흐름 제어, 차원성.

이것은 Python에서 Block 클래스를 작성하는 심층 가이드입니다. 하나의 Workflow 안에서 작은 맞춤 로직만 필요하다면, 하나의 동적 Python 블록 Workflow Definition에 인라인으로 정의한 것이면 보통 충분합니다 - 참조: 사용자 정의 블록.

Workflows 블록 개발에는 Workflow 생태계에 대한 이해가 필요합니다. 세부 사항으로 더 들어가기 전에, 필요한 지식을 요약해 보겠습니다:

다음에 대한 이해 Workflow 실행, 특히:

• Workflow 정의에서 Workflow 블록과 단계의 관계는 무엇인가

• Workflow 블록과 그 manifest가 어떻게 Workflows 컴파일러

• 무엇인지 차원 수준 Workflow를 통과하는 배치 지향 데이터의

• 어떻게 실행 엔진 이 단계와 상호작용하는지, 입력과 출력 측면에서

• 무엇의 성격과 역할이 Workflow 종류인지

• 다음을 이해하는 것 pydantic 작동하는 방식

환경 설정

곧 보게 되겠지만, Workflow 블록을 만드는 것은 특정 인터페이스를 구현하는 Python 클래스를 정의하는 것과 같습니다. 이 설계 덕분에 다른 Python 코드와 마찬가지로 Python 인터프리터를 사용해 블록을 실행할 수 있습니다. 하지만 Workflow 실행 중 다른 블록이 제공하는 모든 필수 입력을 모으는 데 어려움을 겪을 수 있습니다. 따라서 원활한 워크플로우를 위해 개발 환경을 올바르게 설정하는 것이 중요합니다. 표준 개발 프로세스의 일부로 다음 단계를 따를 것을 권장합니다(이전 기여에서는 초기 단계는 건너뛸 수 있습니다):

1. 다음을 설정하고 conda 환경을 그리고 주요 의존성을 설치하세요 inference에서 설명한 대로 inference 기여자 가이드.

2. Workflows 코드베이스의 구조를 익히세요.

3. 최소한의 블록을 생성하기 - 다음 섹션에서 이 방법을 배우게 됩니다. 간단한 블록 manifest와 기본 로직을 구현하여 블록이 예상대로 실행되는지 확인하는 것부터 시작하세요.

4. 블록을 플러그인에 추가하기 - 블록을 만들고 나면, 플러그인에서 내보내는 블록 목록에 추가하세요. Roboflow Core 플러그인에 블록을 추가하는 경우, 다음 파일에 블록 항목을 꼭 추가하세요. loader.py. 이 단계를 잊으면 블록이 보이지 않습니다!

5. 블록을 반복 개선하기 - 결과에 만족할 때까지 블록을 계속 개발하고 실행하세요. 아래 섹션에서는 다양한 시나리오에서 블록을 반복 개선하는 방법을 설명합니다.

Workflows UI를 사용하여 블록 실행하기

마운트된 볼륨을 사용해 inference 서버를 실행하는 것을 권장합니다(매번 서버를 다시 빌드하는 것보다 훨씬 빠릅니다): inference 변경될 때마다):

inference_repo$ docker run -p 9001:9001 \\
   -v ./inference:/app/inference \\
   roboflow/roboflow-inference-server-cpu:latest

그리고 로컬 서버를 Roboflow UI에 연결합니다:

미리보기를 빠르게 실행하려면:

내 블록은 추가 의존성이 필요합니다 - 미리 빌드된 `inference` 서버를 사용할 수 없습니다

블록이 때때로 추가 의존성을 필요로 하는 것은 자연스럽습니다. 의존성을 추가하려면 관련 Docker 이미지(보통 다음 이미지)에 설치되는 requirements 파일에 포함하기만 하면 됩니다. CPU 빌드 의 inference 서버).

그 후 다음을 실행하세요:

inference_repo$ docker build \\
   -t roboflow/roboflow-inference-server-cpu:test \\ 
   -f docker/dockerfiles/Dockerfile.onnx.cpu .

그런 다음 방금 만든 테스트 태그를 지정하여 로컬 빌드를 실행할 수 있습니다:

inference_repo$ inference_repo$ docker run -p 9001:9001 \\
   -v ./inference:/app/inference \\
   roboflow/roboflow-inference-server-cpu:test

Workflows UI 없이 블록 실행하기

Roboflow 플랫폼에 접근할 수 없는 기여자의 경우, 위 섹션에서 언급한 대로 서버를 실행하는 것을 권장합니다. 하지만 UI 편집기를 사용하는 대신, 간단한 Workflow 정의를 만들고 서버에 요청을 보내야 합니다.

UI 없이 Workflow 실행하기

다음 코드 스니펫은 inference 서버에 Workflow를 실행 요청을 보내는 방법을 보여줍니다. inference_sdk 는 다음과 함께 포함되어 있습니다 inference 패키지에 서버용 경량 클라이언트 라이브러리로 포함되어 있습니다.

from inference_sdk import InferenceHTTPClient

YOUR_WORKFLOW_DEFINITION = ...

client = InferenceHTTPClient(
    api_url=object_detection_service_url,
    api_key="XXX",  # 선택 사항, Workflow가 Roboflow Platform을 사용하는 경우에만 필요
)
result = client.run_workflow(
    specification=YOUR_WORKFLOW_DEFINITION,
    images={
        "image": your_image_np,   # 이것은 예시 입력입니다. 적절히 수정하세요
    },
    parameters={
        "my_parameter": 37,   # 이것은 예시 입력입니다. 적절히 수정하세요
    },
)

정기 기여자를 위한 권장 방법

다음 위치에 통합 테스트를 생성하는 것은 tests/workflows/integration_tests/execution 디렉터리는 개발 반복 과정의 자연스러운 일부입니다. 이 접근 방식은 개발과 테스트를 동시에 수행할 수 있게 해주어, 코드를 다듬는 동안 유용한 피드백을 제공합니다. 약간의 경험이 필요하긴 하지만, 장기적인 코드 유지보수성을 크게 향상시킵니다.

과정은 간단합니다:

1. 새 테스트 모듈 생성하기: 예를 들어, 다음과 같이 이름을 짓습니다 test_workflows_with_my_custom_block.py.

2. 예시 Workflow 개발하기: 하나 이상의 예시 Workflow를 만드세요. 블록이 생태계의 다른 블록들과 잘 협력한다면 가장 좋습니다.

3. 샘플 데이터로 테스트 실행하기: 테스트에서 샘플 데이터를 사용해 이러한 Workflow를 실행하세요(우리가 보통 사용하는 예시 데이터를 찾으려면 픽스처 를 살펴보면 됩니다).

4. 예상 결과 검증하기: 결과가 기대와 일치하는지 검증하세요.

개발 흐름에 테스트를 포함하면, 블록이 시간이 지나도 안정적으로 유지되고 기존 블록들과 효과적으로 상호작용하여 작업의 표현력을 높일 수 있습니다!

다음 명령으로 테스트를 실행할 수 있습니다:

pytest tests/workflows/integration_tests/execution/test_workflows_with_my_custom_block

예시를 위해 다른 테스트를 참고하거나 다음 템플릿을 사용하세요:

통합 테스트 템플릿

def test_detection_plus_classification_workflow_when_XXX(
    model_manager: ModelManager,
    dogs_image: np.ndarray, 
    roboflow_api_key: str,
) -> None:
    # 주어진 조건
    workflow_init_parameters = {
        "workflows_core.model_manager": model_manager,
        "workflows_core.api_key": roboflow_api_key,
        "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
    }
    execution_engine = ExecutionEngine.init(
        workflow_definition=<YOUR-EXAMPLE-WORKLFOW>,
        init_parameters=workflow_init_parameters,
        max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
    )

    # 실행 시
    result = execution_engine.run(
        runtime_parameters={
            "image": dogs_image,
        }
    )

    # 그다음
    assert isinstance(result, list), "리스트가 전달될 것으로 예상됨"
    assert len(result) == 1, "하나의 입력 이미지에 대해 출력에 1개 요소가 있을 것으로 예상됨"
    assert set(result[0].keys()) == {
        "predictions",
    }, "선언된 모든 출력이 전달될 것으로 예상됨"
    assert (
        len(result[0]["predictions"]) == 2
    ), "입력 이미지에 개의 개 크롭이 있으므로, 중첩된 분류 결과가 2개일 것으로 예상됨"
    assert [result[0]["predictions"][0]["top"], result[0]["predictions"][1]["top"]] == [
        "116.Parson_russell_terrier",
        "131.Wirehaired_pointing_griffon",
    ], "예측값이 기준 실행에서 측정된 값과 같을 것으로 예상됨"

• 다음 줄에서 2, 다음을 찾을 수 있습니다 model_manager 픽스처가 있습니다. 이 픽스처는 일반적으로 모델 블록에 필요합니다. 이 픽스처는 다음을 제공합니다. ModelManager 다음의 추상화: inference, 모델을 로드하고 언로드하는 데 사용됩니다.

• 다음 줄 3 는 두 마리 개의 이미지를 포함하는 픽스처를 정의합니다(다른 픽스처를 살펴보면 더 많은 예시 이미지를 찾을 수 있습니다).

• 다음 줄 4 는 테스트 중인 Workflow의 블록 중 하나라도 Roboflow API 키가 필요한 경우 사용할 수 있는 선택적 픽스처입니다. 그런 경우 다음을 내보내세요: ROBOFLOW_API_KEY 유효한 키로 환경 변수를

• 다음 줄들 7-11 는 Workflow 정의를 바탕으로 실행 시 Execution Engine이 생성할 블록의 초기화 매개변수 설정을 제공합니다.

• 다음 줄들 19-23 는 입력 매개변수를 주입하여 Workflow를 실행하는 방법을 보여줍니다. runtime_parameters의 키가 Workflow 정의에 선언된 입력과 일치하는지 확인하세요.

• 다음 줄부터 26, 테스트 안에서 예시 검증을 찾을 수 있습니다.

프로토타입

Workflow 블록을 만들려면 Workflows 라이브러리 코어에서 몇 가지 import가 필요합니다. 블록을 만드는 동안 유용하게 쓸 수 있는 import 목록은 다음과 같습니다:

from inference.core.workflows.execution_engine.entities.base import (
    Batch,  # 데이터 배치는 Batch[X] 컨테이너로 들어옵니다
    OutputDefinition,  # manifest에서 출력을 선언하는 데 사용되는 클래스
    WorkflowImageData,  # 이미지의 내부 표현
    # - 입력 종류가 이미지일 때마다 사용
)

from inference.core.workflows.prototypes.block import (
    BlockResult,  # `run(...)` 메서드 결과의 타입 별칭
    WorkflowBlock,  # 블록의 기본 클래스
    WorkflowBlockManifest,  # 블록 manifest의 기본 클래스
)

from inference.core.workflows.execution_engine.entities.types import *  
# core 라이브러리의 `kinds` 모듈

가장 중요한 것은 다음과 같습니다:

• WorkflowBlock - 블록의 기본 클래스

• WorkflowBlockManifest - 블록 manifest의 기본 클래스

내부 데이터 표현 이해하기

우리가 `Batch` 와 `WorkflowImageData` 클래스를 import하도록 권장하는 것을 눈치챘을 수 있습니다. 이 클래스들은 우리 시스템에서 빌딩 블록을 구성할 때 사용되는 핵심 구성 요소입니다. 이 클래스들이 전체 아키텍처에서 어떻게 맞물리는지 더 깊이 이해하려면 자세한 내용을 위해 데이터 표현 페이지를 참고하시기 바랍니다.

블록 manifest

manifest는 Workflow 블록의 중요한 구성 요소로, 블록을 사용하기 위해 Workflow 정의에 배치할 수 있는 단계 선언의 프로토타입을 정의합니다. 특히, 다음을 수행합니다:

• 사용 pydantic Workflows 정의의 구문 분석을 지원합니다: 다음을 상속합니다: pydantic BaseModel 기능을 사용해 Workflow 정의를 파싱하고 검증합니다. 이 스키마는 또한 덕분에 Workflows UI와 호환되는 형식으로 자동 내보내기할 수 있습니다. pydantic의 OpenAPI 표준과의 통합 덕분입니다.

• 데이터 바인딩 정의: manifest의 어떤 필드가 실행 중 workflow를 흐르는 데이터에 대한 선택자인지 지정하고, 그 종류를 나타냅니다.

• 블록 출력 설명: 블록이 생성할 출력을 개략적으로 설명합니다.

• 차원성 지정: 입력 및 출력 차원성과 관련된 속성을 자세히 설명합니다.

• 배치 입력 및 빈 값 여부 표시: 해당 단계가 배치 입력과 빈 값을 허용하는지 Execution Engine에 알립니다.

• 호환성 보장: 안정성을 유지하기 위해 서로 다른 Execution Engine 버전과의 호환성을 지정합니다. 자세한 내용은 다음을 참조하세요: 버전 관리.

manifest 골격

manifest가 어떻게 작동하는지 이해하려면, 단계별로 하나를 정의해 보겠습니다. 여기서 만들 예시 블록은 이미지 유사도를 계산하는 것입니다. import와 클래스 골격부터 시작합니다:

from typing import Literal
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str

이것이 manifest의 최소 표현입니다. 이에는 Compiler와 실행 엔진에 중요한 두 개의 특수 필드가 정의됩니다:

• type - 동적 블록 풀을 기반으로 Workflows 정의의 구문을 파싱하는 데 필요합니다 - 이것이 바로 pydantic type 구분자 컴파일러가 Workflow 정의의 특정 단계를 파싱할 때 어떤 블록 manifest를 검증해야 하는지 이해하게 해줍니다

• name - 이 속성은 단계에 고유한 이름을 부여하고 다른 단계가 선택자를 통해 이를 선택할 수 있게 하는 데 사용됩니다

입력 추가

우리는 단계가 비교할 두 이미지를 입력으로 받도록 하려 합니다.

입력 추가

이 입력들의 정의를 manifest에 추가하는 방법을 살펴보겠습니다:

from typing import Literal, Union
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    # `type`와 `name`을 제외한 모든 속성은 다음 중 하나로 취급됩니다 
    # 하드코딩된 매개변수 또는 데이터 선택자입니다. 데이터 선택자는 문자열로, 
    # `$steps.` 또는 `$inputs.`로 시작하며, 데이터 참조를 표시합니다 
    # 런타임에 사용할 수 있습니다 - 이 경우 보통 데이터의 종류를 지정하여
    # 컴파일러가 우리가 기대하는 데이터 형식을 알 수 있도록 합니다.
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 첫 번째 이미지",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 두 번째 이미지",
    )

• 해당 줄들에서 2-9, 필요한 모든 것을 갖추기 위해 몇 가지 import를 추가했습니다

• 줄 20 정의합니다 image_1 매개변수 - manifest는 Workflow Definition의 프로토타입이므로, 단계에서 사용할 이미지를 지정하는 유일한 방법은 선택자를 제공하는 것입니다 - core 라이브러리에 이를 위해 사용할 수 있는 특수한 타입이 있습니다 - Selector. 코드를 더 깊이 살펴보면, 이것이 타입 별칭 생성자 함수라는 것을 알게 될 것입니다 - 즉, 다음을 알립니다 pydantic 문자열이 다음과 일치할 것으로 기대하도록 $inputs.{name} `WorkflowImageData` $steps.{name}.* 패턴과 각각 대응하며, 추가로 Workflows 생태계 구성 요소에 다음을 알려주는 추가 스키마 필드 메타데이터를 제공합니다 종류 선택자 뒤에 있는 데이터의 이미지. 중요한 참고 사항: 우리는 다음과 같이 표기합니다 종류 리스트로 - 특정 종류들의 리스트는 다음과 같이 해석됩니다 종류들의 합집합 Execution Engine에서.

• 표시하는 pydantic Field(...) 는 줄의 마지막 부분에 있는 속성 20 은 선택 사항이지만, 특히 Workflows UI와 함께 작동하도록 설계된 블록에서는 넣는 것이 좋습니다

• 다음 줄부터 23, 다음의 정의를 찾을 수 있습니다 image_2 매개변수가 있으며, 이는 다음과 매우 유사합니다 image_1.

이러한 manifest 정의는 Workflow 정의에서 다음 단계 선언을 처리할 수 있습니다:

{
  "type": "my_plugin/images_similarity@v1",
  "name": "my_step",
  "image_1": "$inputs.my_image",
  "image_2": "$steps.image_transformation.image"
}

이 정의는 컴파일러와 Execution Engine이 다음을 수행하게 합니다:

• 선언된 type을 가진 Workflow 블록에서 단계를 초기화 my_plugin/images_similarity@v1

• 단계의 run 메서드에 두 개의 매개변수를 제공합니다:

  • input_1 타입의 클래스를 import하도록 권장하는 것을 눈치챘을 수 있습니다. 이 클래스들은 우리 시스템에서 빌딩 블록을 구성할 때 사용되는 핵심 구성 요소입니다. 이 클래스들이 전체 아키텍처에서 어떻게 맞물리는지 더 깊이 이해하려면 이는 Workflow 실행 입력으로 제출된 다음 이름의 이미지로 채워집니다: my_image.

  • imput_2 타입의 클래스를 import하도록 권장하는 것을 눈치챘을 수 있습니다. 이 클래스들은 우리 시스템에서 빌딩 블록을 구성할 때 사용되는 핵심 구성 요소입니다. 이 클래스들이 전체 아키텍처에서 어떻게 맞물리는지 더 깊이 이해하려면 이는 런타임에 다음이라는 다른 단계에 의해 생성됩니다: image_transformation

manifest에 매개변수 추가

이제 단계 실행에 영향을 미칠 매개변수를 추가해 보겠습니다.

manifest에 매개변수 추가

from typing import Literal, Union
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FLOAT_ZERO_TO_ONE_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    # `type`와 `name`을 제외한 모든 속성은 다음 중 하나로 취급됩니다 
    # 하드코딩된 매개변수 또는 데이터 선택자입니다. 데이터 선택자는 문자열로, 
    # `$steps.` 또는 `$inputs.`로 시작하며, 데이터 참조를 표시합니다 
    # 런타임에 사용할 수 있습니다 - 이 경우 보통 데이터의 종류를 지정하여
    # 컴파일러가 우리가 기대하는 데이터 형식을 알 수 있도록 합니다.
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 첫 번째 이미지",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 두 번째 이미지",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        이미지가 유사하다고 간주할 임계값",
    )

• 줄 9 import float_zero_to_one 종류 매개변수를 정의하는 데 사용될 정의입니다.

• 다음 줄에서 27 다음이라는 매개변수를 정의하기 시작합니다 similarity_threshold. manifest는 float 값 또는 다음 Workflow 입력에 대한 선택자를 허용합니다 종류 float_zero_to_one, 다음 줄에서 import한 9.

이러한 manifest 정의는 Workflow 정의에서 다음 단계 선언을 처리할 수 있습니다:

{
  "type": "my_plugin/images_similarity@v1",
  "name": "my_step",
  "image_1": "$inputs.my_image",
  "image_2": "$steps.image_transformation.image",
  "similarity_threshold": "$inputs.my_similarity_threshold"
}

또는 대안으로:

{
  "type": "my_plugin/images_similarity@v1",
  "name": "my_step",
  "image_1": "$inputs.my_image",
  "image_2": "$steps.image_transformation.image",
  "similarity_threshold": 0.5
}

블록 출력 선언

블록의 입력은 성공적으로 정의했지만, 블록을 성공적으로 실행하는 데 필요한 몇 가지 요소가 아직 부족합니다. 블록 출력을 정의해 보겠습니다.

블록 출력 선언

필수 정보의 최소 집합은 출력 설명입니다. 또한 블록의 안정성을 높이기 위해 실행 엔진 호환성에 대한 정보도 제공할 것을 권장합니다.

from typing import Literal, Union
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 첫 번째 이미지",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 두 번째 이미지",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        이미지가 유사하다고 간주할 임계값",
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="images_match", 
            kind=[BOOLEAN_KIND],
          )
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

• 줄 5 단계 출력 설명에 사용되는 클래스를 임포트합니다

• 줄 11 import boolean 종류 출력 정의에 사용하기 위해

• lines 32-39 블록의 출력을 지정하는 클래스 메서드를 선언합니다. 목록의 각 항목은 각 배치 요소와 그에 대한 하나의 반환 속성을 선언합니다 종류. 우리의 블록은 boolean 플래그를 반환합니다 images_match 각 이미지 쌍에 대해.

• lines 41-43 블록의 Execution Engine과의 호환성을 선언합니다 - 자세한 내용은 버전 관리 페이지를 참조하세요

이러한 변경의 결과로:

• Execution Engine은 이 블록을 기반으로 생성된 단계가 지정된 출력을 제공해야 하며, 다른 단계가 입력에서 그 출력들을 참조할 수 있어야 한다는 것을 이해하게 됩니다

• Execution Engine이 버전 v1

자세히 알아보기: 동적 출력

일부 블록은 파싱 후에 사용할 수 있는 step manifest의 내용과 무관하게 classmethod를 사용하여 출력을 임의로 정의할 수 없을 수도 있습니다. 이를 지원하기 위해 다음과 같은 규칙을 도입했습니다:

• classmethod describe_outputs(...) 이름이 하나이고 * 유형이 * (일명 WILDCARD_KIND)

• 추가로, 블록 매니페스트는 인스턴스 메서드 get_actual_outputs(...) 를 구현해야 하며, 이는 채워진 매니페스트 데이터를 기반으로 생성될 수 있는 실제 출력 목록을 제공합니다

from typing import Literal, Union, List, Optional
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FloatZeroToOne,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
    WILDCARD_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 첫 번째 이미지",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 두 번째 이미지",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        이미지가 유사하다고 간주할 임계값",
    )
    outputs: List[str]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="*", 
            kind=[WILDCARD_KIND],
          ),
        ]

    def get_actual_outputs(self) -> List[OutputDefinition]:
        # 여기에서 `self`에 접근할 수 있습니다:
        return [
          OutputDefinition(name=e, kind=[BOOLEAN_KIND])
          for e in self.outputs
        ]

블록 클래스 정의

이 단계에서 우리의 단순한 블록의 매니페스트가 준비되었습니다. 이제 예제를 계속 진행해 보겠습니다. 지금은 너무 산만해질 수 있으니 더 자세한 내용은 고급 주제 섹션을 참고하세요.

기본 구현

매니페스트가 준비되었으므로 블록의 기본 구현을 준비할 수 있습니다.

블록 골격

from typing import Literal, Union, List, Optional, Type
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    WorkflowBlock,
    BlockResult,
)
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FloatZeroToOne,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 첫 번째 이미지",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 두 번째 이미지",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        이미지가 유사하다고 간주할 임계값",
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="images_match", 
            kind=[BOOLEAN_KIND],
          ),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class ImagesSimilarityBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return ImagesSimilarityManifest

    def run(
        self,
        image_1: WorkflowImageData,
        image_2: WorkflowImageData,
        similarity_threshold: float,
    ) -> BlockResult:
        pass

• lines 1, 5-6 `WorkflowImageData` 8-11 블록 클래스와 모든 메서드 시그니처를 올바르게 정의하는 데 필요한 추가 심볼을 제공하기 위해 import 구조에 변경 사항을 추가했습니다

• lines 53-55 클래스 메서드를 정의합니다 get_manifest(...) 이전에 만든 매니페스트 클래스를 단순히 반환하도록

• lines 57-63 정의합니다 run(...) 함수는 Execution Engine이 원하는 결과를 얻기 위해 데이터를 사용하여 호출합니다. 매니페스트 필드가 이미지 kind의 입력을 정의하며 클래스를 import하도록 권장하는 것을 눈치챘을 수 있습니다. 이 클래스들은 우리 시스템에서 빌딩 블록을 구성할 때 사용되는 핵심 구성 요소입니다. 이 클래스들이 전체 아키텍처에서 어떻게 맞물리는지 더 깊이 이해하려면 으로 표시되어 있다는 점에 유의하세요 - 이는 이미지 에서 설명하는 kind 문서.

블록 로직의 구현 제공

이제 블록에 run(...) 메서드의 예시 구현을 추가하여 의미 있는 결과를 생성할 수 있도록 해봅시다.

이 섹션의 내용은 블록 작성자로서 Workflow 생태계와 상호작용하는 방법의 예시를 제공하는 것이 목적이며, 블록의 견고한 구현을 제공하는 것이 목적은 아닙니다.

`run(...)` 메서드의 구현

from typing import Literal, Union, List, Optional, Type
from pydantic import Field
import cv2

from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    WorkflowBlock,
    BlockResult,
)
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FloatZeroToOne,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 첫 번째 이미지",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 두 번째 이미지",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        이미지가 유사하다고 간주할 임계값",
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="images_match", 
            kind=[BOOLEAN_KIND],
          ),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class ImagesSimilarityBlock(WorkflowBlock):

    def __init__(self):
        self._sift = cv2.SIFT_create()
        self._matcher = cv2.FlannBasedMatcher(dict(algorithm=1, trees=5), dict(checks=50))

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return ImagesSimilarityManifest

    def run(
        self,
        image_1: WorkflowImageData,
        image_2: WorkflowImageData,
        similarity_threshold: float,
    ) -> BlockResult:
        image_1_gray = cv2.cvtColor(image_1.numpy_image, cv2.COLOR_BGR2GRAY)
        image_2_gray = cv2.cvtColor(image_2.numpy_image, cv2.COLOR_BGR2GRAY)
        kp_1, des_1 = self._sift.detectAndCompute(image_1_gray, None)
        kp_2, des_2 = self._sift.detectAndCompute(image_2_gray, None)
        matches = self._matcher.knnMatch(des_1, des_2, k=2)
        good_matches = []
        for m, n in matches:
            if m.distance < similarity_threshold * n.distance:
                good_matches.append(m)
        return {
            "images_match": len(good_matches) > 0,
        }

• 다음 줄에서 3 OpenCV를 임포트합니다

• lines 55-57 블록 생성자를 정의합니다. 덕분에 블록 상태는 한 번 초기화되고 이후의 run(...) 메서드 호출을 거쳐 유지됩니다 - 예를 들어 Execution Engine이 비디오의 연속 프레임에서 실행될 때

• lines 69-80 블록 기능의 구현을 제공합니다 - 세부 사항은 Workflows 생태계와 관련하여 실제로 그다지 중요하지 않지만, 주목해야 할 몇 가지 세부 사항이 있습니다:

  • lines 69 `WorkflowImageData` 70 다음을 활용합니다 클래스를 import하도록 권장하는 것을 눈치챘을 수 있습니다. 이 클래스들은 우리 시스템에서 빌딩 블록을 구성할 때 사용되는 핵심 구성 요소입니다. 이 클래스들이 전체 아키텍처에서 어떻게 맞물리는지 더 깊이 이해하려면 추상화를 통해 numpy_image 속성을 사용하여 np.ndarray 를 Workflows의 이미지 내부 표현에서 가져올 수 있음을 보여줍니다. 나머지 클래스를 import하도록 권장하는 것을 눈치챘을 수 있습니다. 이 클래스들은 우리 시스템에서 빌딩 블록을 구성할 때 사용되는 핵심 구성 요소입니다. 이 클래스들이 전체 아키텍처에서 어떻게 맞물리는지 더 깊이 이해하려면 속성들도 살펴보시길 권장합니다.

  • 워크플로우 블록 실행 결과는 78-80 의 경우 단순한 딕셔너리이며 키는 매니페스트에 선언된 출력 이름이고, 43행에 선언되어 있습니다. 선언된 모든 출력을 반드시 제공해야 합니다 - 그렇지 않으면 Execution Engine이 오류를 발생시킵니다.

블록을 플러그인에서 노출하기

이제 블록은 사용할 준비가 되었지만, Execution Engine은 이 블록의 존재를 모릅니다. 이는 등록된 어떤 플러그인도 방금 생성한 블록을 내보내지 않기 때문입니다. 블록 번들링에 대한 자세한 내용은 별도의 페이지에서다루지만, 남은 작업은 플러그인의 load_blocks(...) 함수에서 반환되는 목록에 블록 클래스를 추가하는 것입니다:

# 플러그인의 __init__.py (또는 `inference`에 직접 기여하는 경우 roboflow_core 플러그인)

from my_plugin.images_similarity.v1 import  ImagesSimilarityBlock  
# 이것은 예시 import입니다! 수정이 필요합니다

def load_blocks():
    return [ImagesSimilarityBlock]

고급 주제

입력 배치를 처리하는 블록

때로는 모든 입력 데이터를 배치로 한 번에 처리할 때 블록의 성능이 향상될 수 있습니다. 이는 GPU에서 실행되는 모델에서 발생할 수 있습니다. 이러한 동작 모드는 Workflows 블록에서 지원되며, 여기서는 이를 블록에 사용하는 방법의 예시를 보여드립니다.

배치를 수용하는 블록의 구현

from typing import Literal, Union, List, Optional, Type
from pydantic import Field
import cv2

from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    WorkflowBlock,
    BlockResult,
)
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
    Batch,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FloatZeroToOne,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 첫 번째 이미지",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="유사도를 계산할 두 번째 이미지",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        이미지가 유사하다고 간주할 임계값",
    )

    @classmethod
    def get_parameters_accepting_batches(cls) -> bool:
        return ["image_1", "image_2"]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="images_match", 
            kind=[BOOLEAN_KIND],
          ),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class ImagesSimilarityBlock(WorkflowBlock):

    def __init__(self):
        self._sift = cv2.SIFT_create()
        self._matcher = cv2.FlannBasedMatcher(dict(algorithm=1, trees=5), dict(checks=50))

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return ImagesSimilarityManifest

    def run(
        self,
        image_1: Batch[WorkflowImageData],
        image_2: Batch[WorkflowImageData],
        similarity_threshold: float,
    ) -> BlockResult:
        results = []
        for image_1_element, image_2_element in zip(image_1, image_2): 
          image_1_gray = cv2.cvtColor(image_1_element.numpy_image, cv2.COLOR_BGR2GRAY)
          image_2_gray = cv2.cvtColor(image_2_element.numpy_image, cv2.COLOR_BGR2GRAY)
          kp_1, des_1 = self._sift.detectAndCompute(image_1_gray, None)
          kp_2, des_2 = self._sift.detectAndCompute(image_2_gray, None)
          matches = self._matcher.knnMatch(des_1, des_2, k=2)
          good_matches = []
          for m, n in matches:
              if m.distance < similarity_threshold * n.distance:
                  good_matches.append(m)
          results.append({"images_match": len(good_matches) > 0})
        return results

• 줄 13 import 와 배치 요소를 보관하기 위한 리스트와 매우 유사한(하지만 읽기 전용인) 컨테이너를 나타내는 workflows 라이브러리의 코어에서 가져옵니다

• lines 40-42 블록의 기본 동작을 변경하고 배치를 처리할 수 있도록 만드는 클래스 메서드를 정의합니다 - 우리는 메서드가 배치 지향적이라고 인식하는 각 매개변수를 표시하고 있습니다 run(...) 메서드 는 배치 지향적으로 인식합니다.

• 위에서 도입한 변경 사항으로 인해 run(...) 메서드의 시그니처가 변경되어 이제 image_1 `WorkflowImageData` image_2 는 클래스를 import하도록 권장하는 것을 눈치챘을 수 있습니다. 이 클래스들은 우리 시스템에서 빌딩 블록을 구성할 때 사용되는 핵심 구성 요소입니다. 이 클래스들이 전체 아키텍처에서 어떻게 맞물리는지 더 깊이 이해하려면의 인스턴스가 아니라 이 타입의 요소 배치입니다. 중요한 참고 사항: 여러 배치 지향 매개변수가 있을 때, 해당 배치의 요소들이 대응하는 위치에서 서로 관련되어 있을 것으로 기대합니다 - 즉, 우리의 블록이 비교하는 image_1[1] 을 image_2[1] 으로 비교하는 것이 실제로 논리적으로 의미 있는 작업이 되도록 해야 합니다.

• lines 74-77, 85-86 모든 배치 요소에 대해 run 처리를 수행하는 데 필요한 변경 사항을 보여줍니다 - 필요한 경우 배치 요소를 반복하는 방법을 보여줍니다

• 행에서 출력이 어떻게 구성되는지 주목하는 것이 중요합니다 85 - 배치의 각 요소는 run(...) 메서드에서 반환되는 목록에서 자신의 항목을 받게 됩니다. 순서는 배치 요소의 순서와 일치해야 합니다. 각 출력 딕셔너리는 블록 출력에 선언된 모든 키를 제공해야 합니다.

배치와 스칼라를 모두 허용하는 입력

상대적으로 가능성은 낮지만한 개의 입력 매개변수에서 배치 지향 데이터와 스칼라를 모두 수용해야 하는 블록이 필요할 수 있습니다. Execution Engine은 다음을 사용하여 이를 인식합니다 get_parameters_accepting_batches_and_scalars(...) 블록 매니페스트의 메서드입니다. 아래 제공된 예제를 살펴보세요:

from typing import Literal, Union, List, Optional, Type, Any, Dict
from pydantic import Field

from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    WorkflowBlock,
    BlockResult,
)
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    Batch,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
)

class ExampleManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/example@v1"] 
    name: str
    param_1: Selector()
    param_2: List[Selector()]
    param_3: Dict[str, Selector()]

    @classmethod
    def get_parameters_accepting_batches_and_scalars(cls) -> bool:
        return ["param_1", "param_2", "param_3"]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [OutputDefinition(name="dummy")]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class ExampleBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return ExampleManifest

    def run(
        self,
        param_1: Any,
        param_2: List[Any],
        param_3: Dict[str, Any],
    ) -> BlockResult:
        batch_size = None
        if isinstance(param_1, Batch):
            param_1_result = ...  # 배치 지향 매개변수로 무언가 수행
            batch_size = len(param_1)
        else:
            param_1_result = ... # 스칼라 매개변수로 무언가 수행
        for element in param_2:
           if isinstance(element, Batch):
              ...
           else:
              ...
        for key, value in param_3.items():
           if isinstance(element, value):
              ...
           else:
              ...
        if batch_size is None:
           return {"dummy": "some_result"}
        result = []
        for _ in range(batch_size):
           result.append({"dummy": "some_result"})
        return result

• lines 20-22 혼합 입력(스칼라와 배치 지향 모두)을 수용할 것으로 예상되는 매니페스트 매개변수를 지정합니다 - 이 단계에서는 이전 예제와 비교해 정의에 차이가 없다는 점을 유의하세요.

• lines 24-26 지정합니다 get_parameters_accepting_batches_and_scalars(...) 메서드는 블록이 run(...) 지정된 매개변수에 대해 스칼라와 배치 지향 입력을 모두 처리할 수 있음을 Execution Engine에 알려주는

• lines 45-47 mixed 성격의 매개변수를 run(...) 메서드 시그니처에 나타냅니다.

• 줄 49 블록 로직 내에서 예상 출력 크기를 추적해야 한다는 점을 드러냅니다. 그렇기 때문에 혼합 입력을 가진 블록을 구현하는 것은 꽤 까다롭습니다. 일반적으로 블록의 run(...) 메서드가 스칼라에 대해 동작할 때 - 대부분의 경우(예외는 아래에서 설명) - 메서드는 단일 출력 딕셔너리를 구성합니다. 마찬가지로, 배치 지향 입력이 허용될 때 - 이러한 입력이 예상 출력 크기를 정의합니다. 그러나 이 경우에는 배치를 수동으로 감지하고 그 크기를 파악해야 합니다.

• lines 50-54 배치 지향 데이터가 감지되었을 때 다른 로직을 적용하여 혼합 매개변수를 일반적으로 어떻게 다루는지 보여줍니다

• 앞서 언급했듯이 출력 구성도 혼합 입력의 특성에 맞게 조정되어야 하며 - 이는 65-70

흐름 제어 블록의 구현

흐름 제어 블록은 단순히 데이터를 처리하는 다른 블록들과 상당히 다릅니다. 여기서는 흐름 제어 블록을 만드는 방법을 보여드리겠지만, 먼저 약간의 이론을 살펴보겠습니다:

• 흐름 제어 블록은 매니페스트에서 단계 선택자와의 호환성을 선언하는 블록입니다(단계에 대한 선택자는 $steps.{step_name} 으로 정의됩니다 - 단계 출력 선택자와 유사하지만 출력 이름 지정은 없습니다)

• 흐름 제어 블록은 출력을 등록할 수 없으며, 대신 다음을 반환하도록 되어 있습니다 FlowControl 객체

• FlowControl 객체는 주어진 배치 요소에 대해 다음에 선택되어야 할 다음 단계(단계 매니페스트에서 제공된 선택자) 또는 전체 워크플로우 실행(SIMD가 아닌 흐름 제어)에 대한 다음 단계를 지정합니다

흐름 제어 구현

예제는 랜덤 continue 블록의 구현을 제공하고 주석 처리합니다

from typing import List, Literal, Optional, Type, Union
import random

from pydantic import Field
from inference.core.workflows.execution_engine.entities.base import (
  OutputDefinition,
  WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    StepSelector,
    Selector,
    IMAGE_KIND,
)
from inference.core.workflows.execution_engine.v1.entities import FlowControl
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/random_continue@v1"]
    name: str
    image: Selector(kind=[IMAGE_KIND]) = ImageInputField
    probability: float
    next_steps: List[StepSelector] = Field(
        설명이 true로 평가될 경우 실행되어야 할 단계에 대한 참조
        examples=[["$steps.on_true"]],
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return []

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.2.0,<2.0.0"

class RandomContinueBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: WorkflowImageData,
        probability: float,
        next_steps: List[str],
    ) -> BlockResult:
        if not next_steps or random.random() > probability:
            return FlowControl()
        return FlowControl(context=next_steps)

• 줄 10 블록이 흐름을 제어한다는 것을 Execution Engine에 알리는 데 사용될 단계 선택자에 대한 타입 주석을 임포트합니다

• 줄 14 import FlowControl 흐름 제어 블록에서 유일하게 가능한 응답인 클래스

• 줄 28 단계 선택자 목록을 정의합니다 이는 사실상 블록을 흐름 제어 블록으로 만듭니다

• lines 55 `WorkflowImageData` 56 출력을 구성하는 방법을 보여줍니다 - FlowControl 객체는 다음과 같은 context를 허용합니다 None, 문자열 또는 문자열 목록 - None 배치 요소에 대한 흐름 종료를 나타내며, 문자열은 입력으로 전달된 다음 단계의 선택자여야 합니다.

흐름 제어 구현 - 배치 변형

예제는 랜덤 continue 블록의 구현을 제공하고 주석 처리합니다

from typing import List, Literal, Optional, Type, Union
import random

from pydantic import Field
from inference.core.workflows.execution_engine.entities.base import (
  OutputDefinition,
  WorkflowImageData,
  Batch,
)
from inference.core.workflows.execution_engine.entities.types import (
    StepSelector,
    Selector,
    IMAGE_KIND,
)
from inference.core.workflows.execution_engine.v1.entities import FlowControl
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/random_continue@v1"]
    name: str
    image: Selector(kind=[IMAGE_KIND]) = ImageInputField
    probability: float
    next_steps: List[StepSelector] = Field(
        설명이 true로 평가될 경우 실행되어야 할 단계에 대한 참조
        examples=[["$steps.on_true"]],
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return []

    @classmethod
    def get_parameters_accepting_batches(cls) -> List[str]:
        return ["image"]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class RandomContinueBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: Batch[WorkflowImageData],
        probability: float,
        next_steps: List[str],
    ) -> BlockResult:
        result = []
        for _ in image:
           if not next_steps or random.random() > probability:
               result.append(FlowControl())
           result.append(FlowControl(context=next_steps))
        return result

• 줄 11 블록이 흐름을 제어한다는 것을 Execution Engine에 알리는 데 사용될 단계 선택자에 대한 타입 주석을 임포트합니다

• 줄 15 import FlowControl 흐름 제어 블록에서 유일하게 가능한 응답인 클래스

• lines 29-32 단계 선택자 목록을 정의합니다 이는 사실상 블록을 흐름 제어 블록으로 만듭니다

• lines 38-40 다음을 포함합니다 get_parameters_accepting_batches(...) 메서드는 블록이 run(...) 메서드는 배치 지향 이미지 매개변수를 기대함을 Execution Engine에 알려줍니다.

• 줄 59 결과적으로 우리는 이미지 배치의 각 요소에 대한 흐름 제어 안내를 반환해야 함을 보여줍니다.

• 이를 달성하기 위해, 행에서 60 배치의 내용물을 반복합니다.

• lines 61-63 출력을 구성하는 방법을 보여줍니다 - FlowControl 객체는 다음과 같은 context를 허용합니다 None, 문자열 또는 문자열 목록 - None 배치 요소에 대한 흐름 종료를 나타내며, 문자열은 입력으로 전달된 다음 단계의 선택자여야 합니다.

중첩 선택자

일부 블록은 블록 매니페스트 필드에 선택자 목록 또는 선택자 딕셔너리를 제공해야 할 수 있습니다. Execution Engine의 버전 v1 은 중첩의 한 단계만 지원하므로, 선택자 목록의 목록이나 선택자 목록을 포함한 딕셔너리는 올바르게 인식되지 않습니다.

중첩 선택자의 사용을 보여주는 실제 사용 사례는 아래에 제시되어 있습니다.

가변 개수 모델의 예측 결합

여러 분류기의 예측에 대해 다수결을 수행하는 블록을 만들고 싶다고 가정해 봅시다 - 그렇다면 run 메서드는 다음과 같아야 합니다:

# 여기에는 의사 코드
def run(self, predictions: List[dict]) -> BlockResult:
    predicted_classes = [p["class"] for p in predictions]
    counts = Counter(predicted_classes)
    return {"top_class": counts.most_common(1)[0]}

중첩 선택자 - 모델 앙상블

from typing import List, Literal, Optional, Type

from pydantic import Field
import supervision as sv
from inference.core.workflows.execution_engine.entities.base import (
  OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    OBJECT_DETECTION_PREDICTION_KIND,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/fusion_of_predictions@v1"]
    name: str
    predictions: List[Selector(kind=[OBJECT_DETECTION_PREDICTION_KIND])] = Field(
        설명="단계 출력에 대한 선택자",
        examples=[["$steps.model_1.predictions", "$steps.model_2.predictions"]],
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="predictions", 
            kind=[OBJECT_DETECTION_PREDICTION_KIND],
          )
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class FusionBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        predictions: List[sv.Detections],
    ) -> BlockResult:
        merged = sv.Detections.merge(predictions)
        return {"predictions": merged}

• lines 23-26 선택자 목록을 수용할 수 있는 매니페스트 필드를 정의하는 방법을 보여줍니다

• 줄 50 블록의 입력으로 무엇을 기대해야 하는지 보여줍니다 run(...) 메서드 - 특정 종류를 나타내는 객체의 목록입니다. 블록이 배치를 허용했다면, predictions 필드의 입력 유형은 List[Batch[sv.Detections]

이러한 블록은 다음 단계 선언과 호환됩니다:

{
  "type": "my_plugin/fusion_of_predictions@v1",
  "name": "my_step",
  "predictions": [
    "$steps.model_1.predictions",
    "$steps.model_2.predictions"  
  ]
}

동적 매개변수를 허용하는 데이터 변환 블록

때때로 블록은 이름과 값이 Workflow 정의 작성자가 정의해야 하는 "명명된" 선택자 그룹을 받아들여야 할 수 있습니다. 이 경우 블록 매니페스트는 선택자 딕셔너리를 받아들여야 하며, 키는 해당 선택자의 이름 역할을 합니다.

중첩 선택자 - 명명된 선택자

from typing import List, Literal, Optional, Type, Any

from pydantic import Field
import supervision as sv
from inference.core.workflows.execution_engine.entities.base import (
  OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/named_selectors_example@v1"]
    name: str
    data: Dict[str, Selector()] = Field(
        설명="단계 출력에 대한 선택자",
        examples=[{"a": $steps.model_1.predictions", "b": "$Inputs.data"}],
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(name="my_output", kind=[]),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class BlockWithNamedSelectorsV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        data: Dict[str, Any],
    ) -> BlockResult:
        ...
        return {"my_output": ...}

• lines 22-25 선택자 딕셔너리를 받아들일 수 있는 매니페스트 필드를 정의하는 방법을 보여줍니다 - 선택자 이름과 값 사이의 매핑을 제공합니다

• 줄 46 블록의 입력으로 무엇을 기대해야 하는지 보여줍니다 run(...) 메서드 - 선택자로 참조되는 객체들의 딕셔너리입니다. 블록이 배치를 받아들인다면, 입력 타입은 data 필드의 입력 유형은 Dict[str, Union[Batch[Any], Any]]. 배치가 아닌 경우에는 선택자가 참조하는 배치 비지향 데이터가 자동으로 브로드캐스트되며, 배치를 받아들이는 블록의 경우에는 - 와 컨테이너는 배치 지향 입력만 감싸고, 다른 입력은 단일 값으로 전달됩니다.

이러한 블록은 다음 단계 선언과 호환됩니다:

{
  "type": "my_plugin/named_selectors_example@v1",
  "name": "my_step",
  "data": {
    "a": "$steps.model_1.predictions",
    "b": "$inputs.my_parameter"  
  }
}

실질적인 영향은 다음과 같습니다:

• 아래 data["a"] 안에서 run(...) 모델의 예측을 찾을 수 있습니다 - 예를 들면 sv.Detections 만약 model_1 객체 탐지 모델이라면

• 아래 data["b"] 안에서 run(...), 다음과 같은 이름의 입력 파라미터 값을 찾을 수 있습니다 my_parameter

입력 및 출력 차원성 vs run(...) 메서드

블록 입력의 차원성은 다음을 구성하는 데 중요한 역할을 합니다 run(...) 메서드의 시그니처에 중요한 역할을 하며, 그래서 시스템은 입력 간 차원성 수준 차이에 대해 엄격한 제한을 적용합니다(허용되는 최대 차이는 1). 이 제한은 블록을 작성할 때 일관성과 예측 가능성을 보장하는 데 매우 중요합니다.

차원성 차이가 통제되지 않으면 다음의 구조를 예측하기 어려울 것입니다 run(...) 메서드를 예측하기 어려워져 개발이 더 어렵고 신뢰성이 떨어질 것입니다. 그렇기 때문에 이 속성의 검증은 Workflow 컴파일 과정에서 엄격하게 강제됩니다.

마찬가지로 출력 차원성도 메서드 시그니처와 예상 출력 형식에 영향을 미칩니다. 생태계는 다음 시나리오를 지원합니다:

• 모든 입력이 동일한 차원성을 가지며 그리고 출력은 변하지 않는다 차원성 - 기본 사례

• 모든 입력이 동일한 차원성을 가지며 그리고 출력은 감소한다 차원성

• 모든 입력이 동일한 차원성을 가지며 그리고 출력은 증가한다 차원성

• 입력이 서로 다른 차원성을 가진다 그리고 출력은 다음의 차원성을 유지할 수 있다 참조 입력

그 밖의 입력/출력 차원성 조합은 일관성을 보장하고 메서드 시그니처의 모호성을 방지하기 위해 허용되지 않습니다.

`run(...)` 메서드에서 차원성의 영향 - 배치 비활성화

출력 차원성 증가

이 예제에서는 예측을 기반으로 이미지의 동적 크롭을 수행합니다.

from typing import Dict, List, Literal, Optional, Type, Union
from uuid import uuid4

from inference.core.workflows.execution_engine.constants import DETECTION_ID_KEY
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
    ImageParentMetadata,
)
from inference.core.workflows.execution_engine.entities.types import (
    IMAGE_KIND,
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_block/dynamic_crop@v1"]
    image: Selector(kind=[IMAGE_KIND])
    predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND],
    )

    @classmethod
    def get_output_dimensionality_offset(cls) -> int:
        return 1

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(name="crops", kind=[IMAGE_KIND]),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class DynamicCropBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: WorkflowImageData,
        predictions: sv.Detections,
    ) -> BlockResult:
        crops = []
        for (x_min, y_min, x_max, y_max) in predictions.xyxy.round().astype(dtype=int):
            cropped_image = image.numpy_image[y_min:y_max, x_min:x_max]
            parent_metadata = ImageParentMetadata(parent_id=f"{uuid4()}")
            if cropped_image.size:
                result = WorkflowImageData(
                    parent_metadata=parent_metadata,
                    numpy_image=cropped_image,
                )
            else:
                result = None
            crops.append({"crops": result})
        return crops

• 다음 코드 줄들에서 28-30 매니페스트 클래스는 출력 차원성 오프셋을 선언합니다 - 값 1 는 다음을 더하는 것으로 이해해야 합니다 1 차원성 수준에

• 참고로, 다음 줄에서 63, 블록은 빈 이미지를 이후 처리에서 제외하지만 None 출력이 있는 딕셔너리 대신 이를 배치합니다. 이는 조건부 실행에 사용되는 것과 동일한 Execution Engine 동작을 활용합니다 - 데이터 포인트는 후속 처리에서 제외됩니다(단, 아래에 빈 입력을 요청하는 단계가 있는 경우는 제외).

• 다음 코드 줄들에서 64-65 단일 입력에 대한 결과는 이미지 `WorkflowImageData` predictions 수집됩니다 - 이는 등록된 모든 출력을 키로 포함하는 딕셔너리 목록을 의미합니다. Execution engine은 이 단계가 각 입력 요소에 대해 요소 배치를 반환한다고 이해하고, 하위 단계 실행 중 추적할 수 있도록 중첩된 인덱스 구조를 생성합니다.

출력 차원성 감소

이 예제에서는 블록이 크롭 예측을 시각화하고, 모든 크롭 예측을 하나의 출력 이미지에 보여주는 타일을 생성합니다.

from typing import List, Literal, Type, Union

import supervision as sv

from inference.core.utils.drawing import create_tiles
from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    IMAGE_KIND,
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/tile_detections@v1"]
    crops: Selector(kind=[IMAGE_KIND])
    crops_predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND]
    )
    scalar_parameter: Union[float, Selector()]

    @classmethod
    def get_output_dimensionality_offset(cls) -> int:
        return -1

    @classmethod
    def get_parameters_enforcing_auto_batch_casting(cls) -> List[str]:
        return ["crops", "crops_predictions"]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(name="visualisations", kind=[IMAGE_KIND]),
        ]

class TileDetectionsBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        crops: Batch[WorkflowImageData],
        crops_predictions: Batch[sv.Detections],
        scalar_parameter: float,
    ) -> BlockResult:
        annotator = sv.BoxAnnotator()
        visualisations = []
        for image, prediction in zip(crops, crops_predictions):
            annotated_image = annotator.annotate(
                image.numpy_image.copy(),
                prediction,
            )
            visualisations.append(annotated_image)
        tile = create_tiles(visualisations)
        return {"visualisations": tile}

• 다음 코드 줄들에서 30-32 매니페스트 클래스는 출력 차원성 오프셋을 선언합니다 - 값 -1 는 차원성 수준을 다음만큼 감소시키는 것으로 이해해야 합니다 1

• 다음 코드 줄들에서 34-36 매니페스트 클래스는 run(...) 시그니처가 항상 안정적으로 유지되도록 자동 배치 캐스팅의 대상이 되는 메서드 입력을 선언합니다. 자동 배치 캐스팅은 Execution Engine에서 도입되었습니다 v0.1.6.0

• 참조하십시오 변경 로그 에서 자세한 내용을 확인하세요.

• 다음 코드 줄들에서 53-55 출력 차원성 감소가 메서드 시그니처에 미치는 영향을 확인할 수 있습니다. 처음 두 입력(다음 줄에서 선언됨 36)은 인위적으로 다음에 감싸집니다 Batch[] 컨테이너에, 반면 scalar_parameter 이는 모든 입력이 동일한 차원성을 가질 때 출력 차원성 감소 시 마지막 차원성 수준을 차지하는 모든 요소에 접근할 수 있도록 Execution Engine이 자동으로 수행합니다. 물론 상위 수준 배치의 동일한 요소와 관련된 요소만 그룹화됩니다. 예를 들어, 두 개의 입력 이미지를 크롭했다면 - 그 두 서로 다른 이미지의 크롭은 각각 별도로 그룹화됩니다.

• lines 65-66 출력이 어떻게 구성되는지 보여줍니다 - 단일 값이 반환되며, 그 값은 차원성이 감소된 출력 배치에서 Execution Engine에 의해 인덱싱됩니다

서로 다른 입력 차원성

이 예제에서는 블록이 원본 이미지의 크롭을 기반으로 예측된 탐지 결과를 병합합니다 - 결과적으로 모든 부분 탐지 결과가 병합된 단일 탐지 결과를 제공합니다.

from copy import deepcopy
from typing import Dict, List, Literal, Optional, Type, Union

import numpy as np
import supervision as sv

from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
    IMAGE_KIND,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/stitch@v1"]
    image: Selector(kind=[IMAGE_KIND])
    image_predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND],
    )

    @classmethod
    def get_input_dimensionality_offsets(cls) -> Dict[str, int]:
        return {
            "image": 0,
            "image_predictions": 1,
        }

    @classmethod
    def get_dimensionality_reference_property(cls) -> Optional[str]:
        return "image"

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(
                name="predictions",
                kind=[
                    OBJECT_DETECTION_PREDICTION_KIND,
                ],
            ),
        ]

class StitchDetectionsNonBatchBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: WorkflowImageData,
        image_predictions: Batch[sv.Detections],
    ) -> BlockResult:
        image_predictions = [deepcopy(p) for p in image_predictions if len(p)]
        for p in image_predictions:
            coords = p["parent_coordinates"][0]
            p.xyxy += np.concatenate((coords, coords))
        return {"predictions": sv.Detections.merge(image_predictions)}

• 다음 코드 줄들에서 31-36 매니페스트 클래스는 입력 차원성 오프셋을 선언하며, 이는 이미지 파라미터가 최상위 수준이고 image_predictions 예측의 중첩 배치임을 의미합니다

• 서로 다른 입력 차원성이 선언될 때마다 차원성 참조 속성을 지정해야 합니다(다음 줄 참조 38-40) - 이 차원성 수준이 출력 차원성을 계산하는 데 사용됩니다 - 이 경우, 우리는 다음을 지정합니다 이미지. 이 선택은 예상되는 결과 형식에 영향을 미칩니다 - 선택한 시나리오에서는 등록된 모든 출력 키를 포함하는 단일 딕셔너리를 반환해야 합니다. 만약 우리의 선택이 image_predictions, 중첩된 image_predictions 배치의 길이와 같은 크기의 딕셔너리 목록을 반환합니다. 다시 말해, get_dimensionality_reference_property(...) 어떤 차원성 수준을 출력에 연결해야 하는지를 지정합니다.

• lines 63-64 다음 줄들에서 지정된 차원성 오프셋의 영향을 보여줍니다 31-36. 명확하게 보이듯이 image_predictions 은 다음에 대한 중첩 배치입니다 이미지. 물론, 특정 이미지와 관련된 중첩된 예측만 배치로 그룹화되어 런타임에 메서드에 전달됩니다.

• 앞서 언급했듯이, 다음 줄은 69 출력이 단일 딕셔너리로 구성되도록 합니다. 이는 출력을 다음의 차원성 수준에 등록하기 때문입니다 이미지 (이 역시 단일 요소로 전달되었습니다)

`run(...)` 메서드에서 차원성의 영향 - 배치 활성화

출력 차원성 증가

이 예제에서는 예측을 기반으로 이미지의 동적 크롭을 수행합니다.

from typing import Dict, List, Literal, Optional, Type, Union
from uuid import uuid4

from inference.core.workflows.execution_engine.constants import DETECTION_ID_KEY
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
    ImageParentMetadata,
    Batch,
)
from inference.core.workflows.execution_engine.entities.types import (
    IMAGE_KIND,
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_block/dynamic_crop@v1"]
    image: Selector(kind=[IMAGE_KIND])
    predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND],
    )

    @classmethod
    def get_parameters_accepting_batches(cls) -> bool:
        return ["image", "predictions"]

    @classmethod
    def get_output_dimensionality_offset(cls) -> int:
        return 1

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(name="crops", kind=[IMAGE_KIND]),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class DynamicCropBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: Batch[WorkflowImageData],
        predictions: Batch[sv.Detections],
    ) -> BlockResult:
        results = []
        for single_image, detections in zip(image, predictions):
            crops = []
            for (x_min, y_min, x_max, y_max) in detections.xyxy.round().astype(dtype=int):
                cropped_image = single_image.numpy_image[y_min:y_max, x_min:x_max]
                parent_metadata = ImageParentMetadata(parent_id=f"{uuid4()}")
                if cropped_image.size:
                    result = WorkflowImageData(
                        parent_metadata=parent_metadata,
                        numpy_image=cropped_image,
                    )
                else:
                    result = None
                crops.append({"crops": result})
            results.append(crops)
        return results

• 다음 코드 줄들에서 29-31 매니페스트는 블록이 입력 배치를 받아들인다고 선언합니다

• 다음 코드 줄들에서 33-35 매니페스트 클래스는 출력 차원성 오프셋을 선언합니다 - 값 1 는 다음을 더하는 것으로 이해해야 합니다 1 차원성 수준에

• 다음 코드 줄들에서 55-66, 입력 파라미터의 시그니처는 run(...) 메서드가 동일한 차원성의 입력에 대해 실행되며, 해당 입력이 배치로 제공됨을 반영합니다

• 참고로, 다음 줄에서 70, 블록은 빈 이미지를 이후 처리에서 제외하지만 None 출력이 있는 딕셔너리 대신 이를 배치합니다. 이는 조건부 실행에 사용되는 것과 동일한 Execution Engine 동작을 활용합니다 - 데이터 포인트는 후속 처리에서 제외됩니다(단, 아래에 빈 입력을 요청하는 단계가 있는 경우는 제외).

• 다음 줄들에 제시된 출력 구성은 71-73 중첩이 두 수준임을 나타냅니다. 우선 블록은 배치로 동작하므로, 각 입력 배치 요소마다 하나의 출력이 있는 출력 목록을 반환해야 합니다. 또한 각 입력 배치 요소에 대한 이 출력 요소는 중첩 배치로 나타나므로, 각 입력 이미지와 예측에 대해 블록은 출력 목록을 생성합니다 - 그 목록의 요소들은 선언된 각 출력의 값을 제공하는 딕셔너리입니다.

출력 차원성 감소

이 예제에서는 블록이 크롭 예측을 시각화하고, 모든 크롭 예측을 하나의 출력 이미지에 보여주는 타일을 생성합니다.

from typing import List, Literal, Type, Union

import supervision as sv

from inference.core.utils.drawing import create_tiles
from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    IMAGE_KIND,
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/tile_detections@v1"]
    images_crops: Selector(kind=[IMAGE_KIND])
    crops_predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND]
    )

    @classmethod
    def get_parameters_accepting_batches(cls) -> bool:
        return ["images_crops", "crops_predictions"]

    @classmethod
    def get_output_dimensionality_offset(cls) -> int:
        return -1

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(name="visualisations", kind=[IMAGE_KIND]),
        ]

class TileDetectionsBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        images_crops: Batch[Batch[WorkflowImageData]],
        crops_predictions: Batch[Batch[sv.Detections]],
    ) -> BlockResult:
        annotator = sv.BoxAnnotator()
        visualisations = []
        for image_crops, crop_predictions in zip(images_crops, crops_predictions):
            visualisations_batch_element = []
            for image, prediction in zip(image_crops, crop_predictions):
                annotated_image = annotator.annotate(
                    image.numpy_image.copy(),
                    prediction,
                )
                visualisations_batch_element.append(annotated_image)
            tile = create_tiles(visualisations_batch_element)
            visualisations.append({"visualisations": tile})
        return visualisations

• lines 29-31 매니페스트는 블록이 입력으로 배치를 받도록 기대된다고 선언합니다

• 다음 코드 줄들에서 33-35 매니페스트 클래스는 출력 차원성 오프셋을 선언합니다 - 값 -1 는 차원성 수준을 다음만큼 감소시키는 것으로 이해해야 합니다 1

• 다음 코드 줄들에서 52-53 출력 차원성 감소와 배치 처리의 영향이 메서드 시그니처에 어떻게 반영되는지 확인할 수 있습니다. 첫 번째 "층"은 Batch[] 매니페스트가 블록이 입력 배치를 받아들인다고 선언한 사실의 부수 효과입니다. 두 번째 "층"은 출력 차원성 감소에서 비롯됩니다. Execution Engine은 축소될 차원을 추가적인 Batch[] 컨테이너로 입력에 제공하여, 프로그래머가 특정 최상위 배치 요소에 속하는 모든 중첩 배치 요소를 수집할 수 있게 합니다.

• lines 66-67 출력이 어떻게 구성되는지 보여줍니다 - 각 최상위 배치 요소마다 블록은 모든 크롭과 예측을 집계하여 하나의 타일을 생성합니다. 블록이 입력 배치를 받아들이므로, 이 절차는 각 최상위 배치 요소마다 하나의 타일로 끝나며 - 따라서 딕셔너리 목록을 반환해야 합니다.

서로 다른 입력 차원성

이 예제에서는 블록이 원본 이미지의 크롭을 기반으로 예측된 탐지 결과를 병합합니다 - 결과적으로 모든 부분 탐지 결과가 병합된 단일 탐지 결과를 제공합니다.

from copy import deepcopy
from typing import Dict, List, Literal, Optional, Type, Union

import numpy as np
import supervision as sv

from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
    IMAGE_KIND,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/stitch@v1"]
    images: Selector(kind=[IMAGE_KIND])
    images_predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND],
    )

    @classmethod
    def get_parameters_accepting_batches(cls) -> bool:
        return ["images", "images_predictions"]

    @classmethod
    def get_input_dimensionality_offsets(cls) -> Dict[str, int]:
        return {
            "image": 0,
            "image_predictions": 1,
        }

    @classmethod
    def get_dimensionality_reference_property(cls) -> Optional[str]:
        return "image"

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(
                name="predictions",
                kind=[
                    OBJECT_DETECTION_PREDICTION_KIND,
                ],
            ),
        ]

class StitchDetectionsBatchBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        images: Batch[WorkflowImageData],
        images_predictions: Batch[Batch[sv.Detections]],
    ) -> BlockResult:
        result = []
        for image, image_predictions in zip(images, images_predictions):
            image_predictions = [deepcopy(p) for p in image_predictions if len(p)]
            for p in image_predictions:
                coords = p["parent_coordinates"][0]
                p.xyxy += np.concatenate((coords, coords))
            merged_prediction = sv.Detections.merge(image_predictions)
            result.append({"predictions": merged_prediction})
        return result

• lines 31-33 매니페스트는 블록이 입력으로 배치를 받도록 기대된다고 선언합니다

• 다음 코드 줄들에서 35-40 매니페스트 클래스는 입력 차원성 오프셋을 선언하며, 이는 이미지 파라미터가 최상위 수준이고 image_predictions 예측의 중첩 배치임을 의미합니다

• 서로 다른 입력 차원성이 선언될 때마다 차원성 참조 속성을 지정해야 합니다(다음 줄 참조 42-44) - 이 차원성 수준이 출력 차원성을 계산하는 데 사용됩니다 - 이 경우, 우리는 다음을 지정합니다 이미지. 이 선택은 예상되는 결과 형식에 영향을 미칩니다 - 선택한 시나리오에서는 다음의 각 요소마다 단일 딕셔너리를 반환해야 합니다 이미지 배치. 만약 우리의 선택이 image_predictions, 중첩된 image_predictions 배치의 길이와 같은 크기의 딕셔너리 목록을 각 입력 이미지 배치 요소마다 반환합니다.

• lines 66-67 다음 줄들에서 지정된 차원성 오프셋의 영향을 보여줍니다 35-40 뿐만 아니라 다음 줄들에서의 배치 처리 선언도 32-34첫 번째 "층"은 Batch[] 컨테이너는 후자의 중첩된 Batch[Batch[]] 에 대한 images_predictions 입력 차원성 오프셋 정의에서 비롯됩니다. 명확히 보이듯이 image_predictions 는 특정 요소와 관련된 예측 배치를 담고 있습니다 이미지 배치의 각 요소에 대한 흐름 제어 안내를 반환해야 함을 보여줍니다.

• 앞서 언급했듯이, 다음 줄들은 76-77 다음의 각 요소마다 단일 딕셔너리로 출력이 구성되도록 합니다 이미지 배치

빈 입력을 받아들이는 블록

앞서 논의했듯이, 일부 배치 요소는 Workflow 실행 중에 "비어 있게" 될 수 있습니다. 이는 여러 요인으로 발생할 수 있습니다:

• 흐름 제어 메커니즘: 실행의 특정 분기는 개별 배치 요소를 가려 이후 단계에서 처리되지 않도록 할 수 있습니다.

• 데이터 처리 블록에서는: 경우에 따라 블록이 특정 데이터 포인트에 대해 의미 있는 출력을 생성하지 못할 수 있습니다. 예를 들어, Dynamic Crop 블록은 바운딩 박스 크기가 0이면 크롭된 이미지를 생성할 수 없습니다.

일부 블록은 이러한 빈 입력을 처리하도록 설계되어 있으며, 누락된 출력을 기본값으로 대체하는 블록이 그 예입니다. 이 블록은 Workflow에서 구조화된 출력을 구성할 때 특히 유용할 수 있으며, 일부 요소가 비어 있더라도 출력에 누락된 요소가 없어 파싱을 더 어렵게 만드는 상황을 방지합니다.

빈 입력을 받아들이는 블록

from typing import Any, List, Literal, Optional, Type

from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import Selector
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/first_non_empty_or_default@v1"]
    data: List[Selector()]
    default: Any

    @classmethod
    def accepts_empty_values(cls) -> bool:
        return True

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [OutputDefinition(name="output")]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class FirstNonEmptyOrDefaultBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        data: Batch[Optional[Any]],
        default: Any,
    ) -> BlockResult:
        result = default
        for data_element in data:
            if data_element is not None:
                return {"output": data_element}
        return {"output": result}

• 다음 코드 줄들에서 20-22 블록이 빈 입력을 받아들인다고 명시하는 선언을 볼 수 있습니다

• 다음 줄들의 결과입니다 20-22 다음 줄에서 확인할 수 있습니다 41, 시그니처가 입력이 와 처리해야 하는 빈 요소를 포함할 수 있다고 명시할 때입니다. 사실 이 블록은 빈 값을 대체하는 "인공적" 출력을 생성하며, 이를 통해 이 블록의 출력을 참조하는 빈 입력을 받아들이지 않는 블록에서도 해당 출력이 "보이게" 됩니다. Execution Engine이 런타임에 생성한 데이터로 대체하는 각 입력은 선택적 요소를 제공할 수 있다고 가정해야 합니다.

사용자 정의 생성자 매개변수를 가진 블록

일부 블록은 동작하기 위해 외부 세계에서 생성된 객체를 필요로 할 수 있습니다. 이런 경우 Workflows Execution Engine의 역할은 해당 엔티티를 블록으로 전달하여 사용할 수 있게 하는 것입니다. 이 메커니즘은 다음에서 설명됩니다 Workflow Compiler를 소개하는 페이지, 왜냐하면 이것이 블록 클래스에서 단계들을 동적으로 구성하는 책임을 지는 구성 요소이기 때문입니다.

생성자 매개변수는 다음 조건을 충족해야 합니다:

• 블록이 요청해야 하며 - 클래스 메서드 WorkflowBlock.get_init_parameters(...)

• Workflows Execution Engine이 실행되는 환경에서 제공되어야 합니다:

  • 직접, 다음 예시에서 보이듯이 이 예제

  • 기본값을 사용하여 Workflow 플러그인에 등록된

블록을 정의할 때 init 매개변수를 요청하는 방법을 살펴봅시다.

생성자 매개변수를 요청하는 블록

from typing import Any, List, Literal, Optional, Type

from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import Selector
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/example@v1"]
    data: List[Selector()]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [OutputDefinition(name="output")]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.0.0,<2.0.0"

class ExampleBlock(WorkflowBlock):

    def __init__(my_parameter: int):
        self._my_parameter = my_parameter

    @classmethod
    def get_init_parameters(cls) -> List[str]:
        return ["my_parameter"]

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        data: Batch[Any],
    ) -> BlockResult:
        pass

• lines 30-31 매개변수가 없는 클래스 생성자를 선언합니다

• 블록이 사용자 정의 초기화를 필요로 한다는 점을 Execution Engine에 알리기 위해 get_init_parameters(...) 다음 줄의 메서드는 33-35 제공되어야 하는 모든 매개변수의 이름을 나열합니다

에어갭 / 오프라인 가용성 메타데이터

인터넷 접속 없이 실행될 수 있는 환경(에어갭 배포)을 위한 블록을 만들 때, WorkflowBlockManifest 세 개의 선택적 클래스 메서드를 제공합니다. 에어갭 워크플로 빌더가 어떤 블록을 오프라인에서 사용할 수 있는지 판단하도록 돕기 위해 이를 재정의하세요.

세 가지 모두 합리적인 기본값을 가지므로 기존 블록은 변경이 필요하지 않습니다.

get_air_gapped_availability()

블록이 인터넷 없이 동작할 수 있는지 선언합니다. 클라우드 API(OpenAI, Anthropic 등)를 호출하는 블록에서 이를 재정의하세요.

from inference.core.workflows.prototypes.block import (
    AirGappedAvailability,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_air_gapped_availability(cls) -> AirGappedAvailability:
        return AirGappedAvailability(
            available=False,
            reason="requires_internet",
        )

기본값은 AirGappedAvailability(available=True) - 순수 로직 블록, 로컬 네트워크 블록, 그리고 외부 연결이 필요 없는 모든 블록에 적합합니다.

get_supported_model_variants()

가중치를 로컬에 미리 캐시할 수 있는 foundation-model 블록의 경우, 모델 변형 ID 목록을 반환하세요. 다음 중 하나라도 어떤 변형이 캐시된 아티팩트를 가지고 있으면 블록은 오프라인에서 사용 가능하다고 간주됩니다.

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_supported_model_variants(cls) -> Optional[List[str]]:
        return [
            "sam2/hiera_large",
            "sam2/hiera_small",
            "sam2/hiera_tiny",
            "sam2/hiera_b_plus",
        ]

기본값은 None, 즉 블록이 로컬에 캐시된 모델 가중치에 의존하지 않음을 의미합니다.

get_compatible_task_types()

사용자 훈련 모델을 받아들이는 Roboflow 모델 블록의 경우, 이 블록이 처리할 수 있는 작업 유형을 반환합니다. 에어갭 빌더는 이를 사용해 캐시된 사용자 모델을 호환되는 블록과 매칭합니다.

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_compatible_task_types(cls) -> Optional[List[str]]:
        return ["object-detection"]

기본값은 None - Roboflow 모델로 매개변수화되지 않은 블록(파운데이션 모델, 로직 블록, 싱크 등)에 적합합니다.

런타임 제한

일부 블록은 배포된 런타임(호스티드 서버리스, 전용 배포, 자체 호스팅, 추론 파이프라인), 단계 실행 모드(로컬 vs. 원격), 입력 모드(이미지 vs. 비디오)에 따라 다르게 동작하거나 아예 실패합니다. Override get_restrictions() 를 WorkflowBlockManifest 이러한 주의사항을 블록 내에 한 번 선언하면 실행 엔진, 스키마 엔드포인트, 자동 생성 블록 갤러리가 모두 이를 일관되게 표시할 수 있습니다.

기본값은 [], 따라서 기존 블록은 변경이 필요하지 않습니다.

심각도: soft vs. hard

각 제한에는 심각도:

• Severity.SOFT - 블록은 끝까지 실행되어 올바른 출력 형태를 반환하지만, 값이 저하되거나 의미가 없습니다(예: 트래커 ID가 요청 간에 초기화됨, 쿨다운이 스로틀링하지 않음, 파일이 휘발성 디스크에 기록됨). 워크플로는 계속 실행되지만, 결과는 사용자가 기대한 것이 아닙니다.

• Severity.HARD - 이 블록은 실행되지 않거나 / 예외를 발생시키거나 / 이 런타임에서 사용할 수 있는 출력을 생성할 수 없습니다. 엔진은 컴파일을 거부하거나 즉시 실패해야 합니다.

제한 범위 지정

하나의 RuntimeRestriction 는 세 가지 축의 임의 조합에 따라 범위를 지정할 수 있습니다. 어떤 축이 None로 남아 있으면, 해당 제한은 그 축의 모든 값에 적용됩니다.

• 적용 대상 런타임: Runtime.HOSTED_SERVERLESS, Runtime.DEDICATED_DEPLOYMENT, Runtime.SELF_HOSTED_CPU, Runtime.SELF_HOSTED_GPU, Runtime.INFERENCE_PIPELINE.

• 적용 대상 단계 실행 모드: StepExecutionMode.LOCAL, StepExecutionMode.REMOTE.

• 적용 대상 입력 모드: RuntimeInputMode.IMAGE, RuntimeInputMode.VIDEO.

The note 필드는 실패 모드 또는 성능 저하 동작에 대한 한 줄짜리 사람이 읽을 수 있는 설명입니다. 무엇이 일어나는지(예: "track_ids reset between requests", "writes to ephemeral /tmp")를 설명하고, 추상적인 전제 조건은 설명하지 마세요.

공유 프리셋

대부분의 주의사항은 몇 가지 일반적인 패턴으로 분류되므로, 코드베이스 전반에서 표현을 일관되게 유지하기 위해 재사용 가능한 프리셋을 데이터 클래스와 함께 내보냅니다. 먼저 이들을 사용하세요:

• STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION - 비디오 추적 / 카운팅 / 집계 블록으로, 비디오별 상태가 프로세스 메모리에 있고 상태 비저장 HTTP 요청 사이에서 초기화되는 경우에 사용합니다.

• COOLDOWN_HTTP_SOFT_RESTRICTION - 쿨다운 / 레이트 리밋 타이머가 프로세스 메모리에 보관되어 있어 다중 복제 HTTP 런타임에서 스로틀링하지 않는 블록에 사용합니다.

• STILL_IMAGE_INPUT_SOFT_RESTRICTION - 시간적 맥락(비디오 또는 반복 프레임)에 의존하고 정지 이미지에서는 거의 또는 전혀 이점이 없는 블록에 사용합니다.

예시

비디오별 상태를 유지하고 정지 이미지에서는 의미도 없는 라인 카운터 블록은 다음 두 제한을 모두 선언합니다:

from typing import List

from inference.core.workflows.prototypes.block import (
    STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION,
    STILL_IMAGE_INPUT_SOFT_RESTRICTION,
    RuntimeRestriction,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_restrictions(cls) -> List[RuntimeRestriction]:
        return [
            STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION,
            STILL_IMAGE_INPUT_SOFT_RESTRICTION,
        ]

사용자 정의 제한(예: Severity.HARD 호스티드 서버리스 런타임에 없는 GPU 하드웨어가 필요한 블록)은 인라인으로 선언합니다:

from inference.core.workflows.prototypes.block import (
    Runtime,
    RuntimeRestriction,
    Severity,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_restrictions(cls) -> List[RuntimeRestriction]:
        return [
            RuntimeRestriction(
                severity=Severity.HARD,
                note="Block requires a CUDA GPU; raises on CPU-only workers.",
                applies_to_runtimes=[
                    Runtime.HOSTED_SERVERLESS,
                    Runtime.SELF_HOSTED_CPU,
                ],
            ),
        ]

이렇게 선언된 제한은 세 곳에 표시됩니다:

1. The describe_interface HTTP 페이로드(를 통해 RuntimeRestriction.to_dict()), 따라서 워크플로 클라이언트와 빌더는 워크플로 실행 전에 사용자에게 경고할 수 있습니다.

2. 블록의 자동 생성 블록 갤러리 페이지에서 "Runtime compatibility" 섹션 아래, 바로 Properties.

3. 실행 엔진은 현재 런타임에 대한 Severity.HARD 제한에서 fail-fast를 선택할 수 있습니다.

종속 리소스 선언하기

많은 블록은 실행 시 외부 리소스가 필요합니다: Roboflow에서 학습된 또는 파운데이션 모델 가중치, Roboflow 프로젝트(예: 액티브 러닝 대상, 데이터셋 업로드 대상) 또는 서드파티 호스팅 모델(OpenAI, Anthropic, OpenRouter, ...). 워크플로가 실제로 실행되기 전까지는 시스템의 어느 부분도 어떤 리소스가 필요할지 알 수 없습니다.

오버라이드하기 discover_dependent_resources() 를 블록 매니페스트에서 오버라이드하면 그 공백이 메워집니다. 이 메서드는 파싱된 매니페스트 인스턴스에서 호출되므로, 주어진 단계의 구체적인 필드 값으로부터 선언을 계산할 수 있습니다. 이를 통해 호출자는 전체 워크플로의 리소스를 정적으로, 즉 아무것도 실행하지 않고 컴파일 시점에 열거할 수 있으며, 예측 가능한 실행 시간을 위해 모델 가중치를 사전에 로드하거나 참조된 모든 모델과 프로젝트에 API 키가 접근 가능한지 미리 검증하는 등의 사용 사례를 가능하게 합니다.

def discover_dependent_resources(self) -> Optional[List[DependentResource]]:
    ...

기본값은 None, 즉 블록은 종속성을 선언하지 않음 을 의미하므로, 호출자는 이를 알 수 없음으로 취급해야 합니다. 이는 블록에 외부 리소스가 전혀 필요 없음을 적극적으로 선언하는 []와는 의도적으로 다릅니다. 기존 블록은 변경이 필요 없으며, 리소스를 사용하는 블록은 오버라이드해야 합니다.

리소스 봉투

블록은 리소스 형태를 새로 만들지 않습니다. 봉투는 실행 엔진에 의해 규제됩니다. 하나의 DependentResource 는 resource_type 를 해당 유형에 대해 등록된 타입 지정(pydantic) 메타데이터 엔터티와 짝지어 줍니다:

• DependentResourceType.ROBOFLOW_PLATFORM_MODEL → RoboflowPlatformModelMetadata(model_id, required_action, execution_location) — Roboflow 플랫폼을 통해 제공되는 모델로, 합성 ID를 가진 파운데이션 / 코어 모델을 포함합니다(예: clip/ViT-B-32). required_action 은 사용 방식의 성격을 나타냅니다: ModelRequiredAction.EXECUTION (가중치가 로드되거나 추론이 요청됨) 또는 ModelRequiredAction.ACCESS (모델 엔터티가 플랫폼에서 접근 가능하기만 하면 됨 — 예: 메타데이터를 연결하는 모니터링 싱크). 실행의 경우, execution_location 은 LOCAL, REMOTE, 또는 ENVIRONMENT_DEFINED 이며, 로컬 여부는 WORKFLOWS_STEP_EXECUTION_MODE 에 의해 런타임에 결정되고 컴파일 시점에는 판단할 수 없습니다(단계 실행 모드에 따라 분기하는 모델 블록의 기본값).

• DependentResourceType.ROBOFLOW_PLATFORM_PROJECT → RoboflowPlatformProjectMetadata(project_url) — 블록이 읽거나 쓰는 Roboflow 프로젝트입니다.

• DependentResourceType.THIRD_PARTY_MODEL → ThirdPartyModelMetadata(provider, model_id) — 외부 제공자가 실행하는 모델이며, 정의상 원격 실행입니다.

팩토리 헬퍼는 구현을 한 줄로 유지해 줍니다: roboflow_platform_model(), roboflow_platform_project(), third_party_model().

선택자 값과 런타임 해석

매니페스트 필드에는 구체적 값 대신 워크플로 선택자가 들어갈 수 있습니다. 선언은 그런 값을 있는 그대로 반환합니다 — 블록은 선택자를 해석하지 않습니다. 호출자는 런타임 매개변수가 알려지면 $inputs.<name> 참조를 대체할 수 있습니다. $steps.<name>.<property> 참조는 정적으로 전혀 해석할 수 없습니다. 모든 메타데이터 엔터티는 requires_runtime_resolution() 을 노출하므로, 호출자는 구체적인 식별자와 아직 해석이 필요한 참조를 구분할 수 있습니다.

최종 식별자가 필드 값의 함수 (예: clip/<version>, 카탈로그 조회)인 경우, 치환된 입력 값만으로는 실행된 ID가 아닙니다. 그러한 선언은 model_id_resolver 를 붙입니다. — 이 값 해석에 필요한 모든 것을 클로저에 고정해 둔 호출 가능 객체로, 치환된 값을 최종 ID로 바꿉니다. 이 리졸버는 프로세스 내부 보조 수단일 뿐이며 직렬화, JSON 스키마, 동등성 비교에서 제외됩니다. 프로세스 내부 호출자(예: 엔진의 첫 실행 사전 로딩)는 입력 값을 치환한 뒤 이를 호출합니다. 리졸버는 None 를 반환하여 값이 정적으로 해석 불가능함을 선언할 수 있습니다(최종 ID가 그 하나의 값보다 더 많은 것에 의존함) — 호출자는 그런 종속성을 건너뛰고 실행이 이를 해석합니다. 예외를 발생시키면 그 값은 실제로 유효하지 않다는 뜻입니다.

무엇과 일치시키세요 run() 가 로드하는 것과

선언된 식별자는 run() 가 요청할 정확히 그 ID여야 합니다 — 버전 필드에서 합성된 ID를 포함합니다. 블록이 f"my_family/{self.version}" 를 load_core_model(...) 호출 지점에서 만든다면, 선언도 동일한 ID를 만들어야 하며(그리고 버전 필드에 선택자가 들어 있으면 그 선택자를 있는 그대로 사용해야 합니다).

예시

모델과 선택적 액티브 러닝 대상 프로젝트를 선언하는 모델 블록:

from inference.core.workflows.prototypes.block import (
    DependentResource,
    WorkflowBlockManifest,
    roboflow_platform_model,
    roboflow_platform_project,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    def discover_dependent_resources(self) -> Optional[List[DependentResource]]:
        resources = [roboflow_platform_model(model_id=self.model_id)]
        if self.active_learning_target_dataset is not None:
            resources.append(
                roboflow_platform_project(
                    project_url=self.active_learning_target_dataset
                )
            )
        return resources

버전 필드에서 모델 ID가 합성되는 블록:

from inference.core.workflows.prototypes.block import (
    DependentResource,
    WorkflowBlockManifest,
    is_workflow_selector,
    roboflow_platform_model,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    def discover_dependent_resources(self) -> Optional[List[DependentResource]]:
        if is_workflow_selector(self.version):
            # 선택자는 그대로 반환됩니다. 리졸버는 최종 ID를 생성합니다
            # 입력 값이 치환된 뒤에.
            return [
                roboflow_platform_model(
                    model_id=self.version,
                    model_id_resolver=lambda version: f"clip/{version}",
                )
            ]
        return [roboflow_platform_model(model_id=f"clip/{self.version}")]

실행하지 않고 모델 엔터티만 참조하는 싱크:

from inference.core.workflows.prototypes.block import (
    DependentResource,
    ModelRequiredAction,
    WorkflowBlockManifest,
    roboflow_platform_model,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    def discover_dependent_resources(self) -> Optional[List[DependentResource]]:
        return [
            roboflow_platform_model(
                model_id=self.model_id,
                required_action=ModelRequiredAction.ACCESS,
            )
        ]

선언하지 말아야 할 때

단지 를 담는 모델 ID 또는 프로젝트 종류 값을 일반 페이로드로 전달하는 필드(예: 쿼리 매개변수로 값을 전달하는 웹훅 싱크)는 리소스 종속성이 아니며 — 그런 블록은 의도적으로 기본값을 유지합니다.

핵심 저장소의 기여자는 그 단위 테스트(tests/workflows/unit_tests/core_steps/test_dependent_resources.py)가 경계를 보호한다는 점을 인지해야 합니다: 매니페스트가 roboflow_model_id 또는 roboflow_project 유형의 필드를 선언하는 모든 코어 블록은 반드시 discover_dependent_resources() 를 오버라이드하거나, 또는 carry-only로 명시적으로 허용 목록에 들어 있어야 합니다.

모델 가중치를 로드하는 블록은 모델 매니저 외부에서 (무엇이 run() 실제로 하는지 확인하세요, model_manager 초기화 매개변수에 나타나는지 여부가 아니라) 지금은 이 메서드를 구현해서는 안 됩니다 — 그 종속성은 선언되지 않은 상태로 남습니다(None).

커스텀 파이썬 블록은 항상 None: 해당 코드가 정적 분석에 불투명하므로, 알 수 없음 가 유일하게 정직한 답입니다.