एक Workflow block बनाएँ

Workflow block लिखने के लिए पूर्ण डेवलपर मार्गदर्शिका: manifests, inputs, outputs, batch processing, flow control, और dimensionality।

यह Python में एक Block class लिखने के लिए विस्तृत मार्गदर्शिका है। यदि आपको केवल एक Workflow के भीतर कस्टम लॉजिक का एक छोटा हिस्सा चाहिए, तो dynamic Python block जो Workflow Definition में inline परिभाषित हो, आमतौर पर पर्याप्त होता है - देखें Custom Blocks.

Workflows blocks development के लिए Workflow Ecosystem की समझ आवश्यक है। विवरण में गहराई से जाने से पहले, आइए आवश्यक ज्ञान का सारांश लें:

की समझ Workflow execution, विशेष रूप से:

Workflow definition में Workflow blocks और steps के बीच क्या संबंध है
Workflow blocks और उनके manifests का उपयोग कैसे किया जाता है Workflows Compiler
क्या है dimensionality level Workflow से गुजरने वाले batch-oriented data का
कैसे Execution Engine step के साथ उसके inputs और outputs के संबंध में कैसे इंटरैक्ट करता है
क्या है Workflow kinds
को समझना pydantic कैसे काम करता है

Environment setup

जैसा कि आप जल्द ही देखेंगे, Workflow block बनाना केवल एक विशेष interface लागू करने वाली Python class परिभाषित करने का मामला है। यह design आपको block को Python interpreter से चलाने की अनुमति देता है, ठीक किसी अन्य Python code की तरह। हालांकि, आवश्यक सभी inputs को एकत्रित करते समय आपको कठिनाइयों का सामना करना पड़ सकता है, जो सामान्यतः Workflow execution के दौरान अन्य blocks द्वारा प्रदान किए जाते हैं। इसलिए, सुचारु workflow के लिए development environment को ठीक से सेट करना महत्वपूर्ण है। हम standard development process के भाग के रूप में निम्न चरणों का पालन करने की सलाह देते हैं (बाद के contributions के लिए प्रारंभिक चरण छोड़े जा सकते हैं):

सेट अप करें conda environment और मुख्य dependencies install करें inferenceकी, जैसा कि inference contributor guide.
Workflows codebase के संगठन से परिचित हों।
एक न्यूनतम block बनाएं - आप अगले sections में यह करना सीखेंगे। एक सरल block manifest और मूल logic लागू करके शुरू करें ताकि सुनिश्चित हो सके कि block अपेक्षित रूप से चलता है।
block को plugin में जोड़ें - एक बार आपका block बन जाने के बाद, उसे plugin से export किए गए blocks की सूची में जोड़ें। यदि आप block को Roboflow Core plugin में जोड़ रहे हैं, तो सुनिश्चित करें कि आपके block के लिए एक entry जोड़ें loader.py. यदि आप यह चरण भूल जाते हैं, तो आपका block दिखाई नहीं देगा!
अपने block को iterate करें और refine करें - जब तक आप परिणामों से संतुष्ट न हों, अपने block का विकास और संचालन जारी रखें। नीचे दिए गए sections विभिन्न scenarios में अपने block पर iterate करने का तरीका समझाते हैं।

Workflows UI का उपयोग करके अपने blocks चलाना

हम mounted volume के साथ inference server चलाने की सलाह देते हैं (जो inference हर बदलाव पर server को फिर से build करने से बहुत तेज़ है):

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

और अपने local server को Roboflow UI से जोड़ना:

Connecting a local Inference Server to the Roboflow Workflows UI

त्वरित previews चलाने के लिए:

मेरे block को अतिरिक्त dependencies चाहिए - मैं pre-built `inference` server का उपयोग नहीं कर सकता

यह स्वाभाविक है कि आपके blocks को कभी-कभी अतिरिक्त dependencies की आवश्यकता हो सकती है। किसी dependency को जोड़ने के लिए, बस उसे इनमें से किसी एक में शामिल करें requirements filesजो संबंधित Docker image में install किए जाते हैं (आमतौर पर CPU build का inference server).

इसके बाद, चलाएँ:

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

फिर आप अभी बनाए गए test tag को निर्दिष्ट करके अपने local build को चला सकते हैं:

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

Workflows UI के बिना अपने blocks चलाना

Roboflow platform तक पहुँच न रखने वाले contributors के लिए, हम ऊपर दिए गए section में बताए अनुसार server चलाने की सलाह देते हैं। हालांकि, UI editor के बजाय, आपको एक सरल Workflow definition बनानी होगी और server को request भेजनी होगी।

UI के बिना अपना Workflow चलाना

निम्न code snippet दिखाता है कि inference server को request भेजकर Workflow कैसे चलाया जाए। inference_sdk को inference package में हमारे server के लिए एक lightweight client library के रूप में शामिल किया गया है।

from inference_sdk import InferenceHTTPClient

YOUR_WORKFLOW_DEFINITION = ...

client = InferenceHTTPClient(
    api_url=object_detection_service_url,
    api_key="XXX",  # optional, केवल तभी आवश्यक जब Workflow Roboflow Platform का उपयोग करे
)
result = client.run_workflow(
    specification=YOUR_WORKFLOW_DEFINITION,
    images={
        "image": your_image_np,   # यह उदाहरण input है, इसे समायोजित करें
    },
    parameters={
        "my_parameter": 37,   # यह उदाहरण input है, इसे समायोजित करें
    },
)

नियमित contributors के लिए अनुशंसित तरीका

में integration tests बनाना tests/workflows/integration_tests/execution directory विकास iteration प्रक्रिया का एक स्वाभाविक हिस्सा है। यह दृष्टिकोण आपको एक साथ विकास और परीक्षण करने देता है, और जैसे-जैसे आप अपने code को refine करते हैं, मूल्यवान feedback प्रदान करता है। हालांकि इसके लिए कुछ अनुभव चाहिए, यह long-term code maintenance को काफी बेहतर बनाता है।

प्रक्रिया सीधी है:

एक नया Test Module बनाएं: उदाहरण के लिए, इसका नाम रखें test_workflows_with_my_custom_block.py.
उदाहरण Workflows विकसित करें: एक या अधिक example Workflows बनाएं। यदि आपका block ecosystem के अन्य blocks के साथ सहयोग करे, तो यह बेहतर होगा।
Sample Data के साथ Tests चलाएँ: अपने tests में sample data का उपयोग करके इन Workflows को execute करें (आप हमारे fixtures को देखकर उदाहरण data ढूँढ सकते हैं जिसका हम आमतौर पर उपयोग करते हैं)।
अपेक्षित परिणामों की पुष्टि करें: सत्यापित करें कि परिणाम आपकी अपेक्षाओं से मेल खाते हैं।

अपने development flow में testing शामिल करके, आप सुनिश्चित करते हैं कि आपका block समय के साथ स्थिर बना रहे और मौजूदा blocks के साथ प्रभावी ढंग से इंटरैक्ट करे, जिससे आपके कार्य की अभिव्यक्तिशीलता बढ़ती है!

आप निम्न कमांड का उपयोग करके अपना test चला सकते हैं:

pytest tests/workflows/integration_tests/execution/test_workflows_with_my_custom_block

उदाहरणों के लिए अन्य tests का संदर्भ लेने या निम्न template का उपयोग करने में संकोच न करें:

Integration test template

def test_detection_plus_classification_workflow_when_XXX(
    model_manager: ModelManager,
    dogs_image: np.ndarray, 
    roboflow_api_key: str,
) -> None:
    # given
    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,
    )

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

    # then
    assert isinstance(result, list), "Expected list to be delivered"
    assert len(result) == 1, "Expected 1 element in the output for one input image"
    assert set(result[0].keys()) == {
        "predictions",
    }, "Expected all declared outputs to be delivered"
    assert (
        len(result[0]["predictions"]) == 2
    ), "Expected 2 dogs crops on input image, hence 2 nested classification results"
    assert [result[0]["predictions"][0]["top"], result[0]["predictions"][1]["top"]] == [
        "116.Parson_russell_terrier",
        "131.Wirehaired_pointing_griffon",
    ], "Expected predictions to be as measured in reference run"

लाइन 2में, आपको model_manager fixture मिलेगा, जो आम तौर पर model blocks द्वारा आवश्यक होता है। यह fixture ModelManager abstraction प्रदान करता है inferenceसे, जिसका उपयोग models को load और unload करने के लिए किया जाता है।
लाइन 3 एक ऐसा fixture परिभाषित करती है जिसमें दो कुत्तों की image शामिल है (और उदाहरण images के लिए अन्य fixtures देखें)।
लाइन 4 एक वैकल्पिक fixture है जिसका आप उपयोग करना चाह सकते हैं यदि आपके tested workflow के किसी block को Roboflow API key की आवश्यकता हो। यदि ऐसा है, तो ROBOFLOW_API_KEY environment variable को test चलाने से पहले एक मान्य key के साथ export करें।
लाइनें 7-11 Execution Engine द्वारा runtime पर बनाए जाने वाले blocks के initialization parameters के setup को प्रदान करती हैं, जो आपकी Workflow definition पर आधारित है।
लाइनें 19-23 दिखाती हैं कि input parameters inject करके एक Workflow कैसे चलाया जाए। कृपया सुनिश्चित करें कि runtime_parameters में keys आपकी Workflow definition में घोषित inputs से मेल खाएँ।
लाइन से शुरू करते हुए 26, आपको test के भीतर example assertions मिलेंगे।

Prototypes

Workflow block बनाने के लिए आपको Workflows library के core से कुछ imports की आवश्यकता होती है। यहाँ imports की सूची है जो block बनाते समय उपयोगी हो सकती है:

from inference.core.workflows.execution_engine.entities.base import (
    Batch,  # data के batches Batch[X] containers में आएँगे
    OutputDefinition,  # manifest में outputs घोषित करने के लिए उपयोग होने वाली class
    WorkflowImageData,  # image की internal representation
    # - जब भी आपके input का kind image हो, इसका उपयोग करें
)

from inference.core.workflows.prototypes.block import (
    BlockResult,  # `run(...)` method के परिणाम के लिए type alias
    WorkflowBlock,  # आपके block के लिए base class
    WorkflowBlockManifest,  # block manifest के लिए base class
)

from inference.core.workflows.execution_engine.entities.types import *  
# core library से `kinds` वाला module

सबसे महत्वपूर्ण हैं:

WorkflowBlock - आपके block के लिए base class
WorkflowBlockManifest - block manifest के लिए base class

आंतरिक data representation को समझना

आपने देखा होगा कि हम Batch और WorkflowImageData classes को import करने की सलाह देते हैं, जो हमारे system में building blocks बनाते समय उपयोग किए जाने वाले मौलिक components हैं। इन classes का समग्र architecture में स्थान कैसे है, इसे बेहतर समझने के लिए हम आपको Data Representations page को अधिक विस्तृत जानकारी के लिए देखने की सलाह देते हैं।

Block manifest

Manifest एक Workflow block का महत्वपूर्ण घटक है जो step declaration के लिए एक prototype परिभाषित करता है, जिसे block का उपयोग करने के लिए Workflow definition में रखा जा सकता है। विशेष रूप से, यह:

उपयोग करता है pydantic Workflow definitions की syntax parsing को संचालित करने के लिए: यह pydantic BaseModel features से Workflow definitions को parse और validate करता है। इस schema को pydantic's OpenAPI standard के साथ integration की बदौलत, Workflows UI के अनुकूल format में स्वचालित रूप से export भी किया जा सकता है।
Data Bindings परिभाषित करता है: यह निर्दिष्ट करता है कि manifest में कौन से fields execution के दौरान workflow से बहने वाले data के selectors हैं और उनके kinds क्या हैं।
Block Outputs का वर्णन करता है: यह उन outputs की रूपरेखा देता है जो block उत्पन्न करेगा।
Dimensionality निर्दिष्ट करता है: यह input और output dimensionality से संबंधित properties का विवरण देता है।
Batch Inputs और Empty Values को इंगित करता है: यह Execution Engine को बताता है कि step batch inputs और empty values स्वीकार करता है या नहीं।
Compatibility सुनिश्चित करता है: यह स्थिरता बनाए रखने के लिए विभिन्न Execution Engine versions के साथ compatibility निर्धारित करता है। अधिक विवरण के लिए देखें versioning.

Manifest के लिए scaffolding

यह समझने के लिए कि manifests कैसे काम करते हैं, आइए उन्हें चरण-दर-चरण परिभाषित करें। यहाँ हम जो उदाहरण block बनाएँगे, वह images similarity की गणना करेगा। हम imports और class scaffolding से शुरू करते हैं:

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 का न्यूनतम representation है। यह दो विशेष fields परिभाषित करता है जो Compiler और Execution engine के लिए महत्वपूर्ण हैं:

type - blocks के dynamic pool पर आधारित Workflow definitions की syntax parse करने के लिए आवश्यक - यह है pydantic type discriminator जो Compiler को यह समझने देता है कि Workflow definition में विशिष्ट steps को parse करते समय किस block manifest की जाँच करनी है
name - इस property का उपयोग step को एक unique नाम देने और अन्य steps को selectors के माध्यम से इसे चुनने देने के लिए किया जाएगा

Inputs जोड़ना

हम चाहते हैं कि हमारा step तुलना के लिए images के साथ दो inputs ले।

Inputs जोड़ना

आइए देखें कि इन inputs की definitions को 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` के अलावा सभी properties को या तो 
    hardcoded parameters या data selectors के रूप में माना जाता है। Data selectors ऐसी strings हैं 
    जो `$steps.` या `$inputs.` से शुरू होती हैं और runtime में उपलब्ध data के लिए references दर्शाती हैं - 
    इस मामले में हम आमतौर पर data के kinds निर्दिष्ट करते हैं
    ताकि compiler को पता चले कि हम data को कैसा देखना अपेक्षित करते हैं।
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए पहली image",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए दूसरी image",
    )

लाइनों में 2-9हमने कुछ imports जोड़े हैं ताकि हमारे पास आवश्यक सब कुछ हो
लाइन 20 परिभाषित करती है image_1 parameter - क्योंकि manifest Workflow Definition का prototype है, step द्वारा उपयोग की जाने वाली image के बारे में बताने का एकमात्र तरीका selector प्रदान करना है - core library में हमारे पास एक specialized type है जिसका उपयोग किया जा सकता है - Selector. यदि आप codebase में और गहराई से देखें, तो आपको पता चलेगा कि यह एक type alias constructor function है - जो बताता है pydantic कि यह string pattern से मेल खाने की अपेक्षा करता है $inputs.{name} और $steps.{name}.* patterns क्रमशः, और साथ ही अतिरिक्त schema field metadata प्रदान करता है जो Workflows ecosystem components को बताता है कि selector के पीछे का kind का data है image. महत्वपूर्ण नोट: हम kind को list के रूप में दर्शाते हैं - विशिष्ट kinds की list को kinds के union के रूप में Execution Engine द्वारा समझा जाता है।
दर्शाते हुए pydantic Field(...) लाइन के अंतिम भागों में attribute 20 वैकल्पिक है, लेकिन उपयोगी है, विशेष रूप से उन blocks के लिए जो Workflows UI के साथ सहयोग करने हेतु बनाए गए हैं
लाइन में शुरुआत करते हुए 23आप पा सकते हैं image_2 parameter की परिभाषा जो काफी हद तक image_1.

manifest की ऐसी परिभाषा Workflow definition में निम्न step declaration को संभाल सकती है:

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

यह परिभाषा Compiler और Execution Engine को यह करने देगी:

Workflow block से step को initialize करना जो type घोषित करता है my_plugin/images_similarity@v1
steps run method के लिए दो parameters प्रदान करना:
- input_1 type का WorkflowImageData जो Workflow execution input के रूप में भेजी गई image से भरा जाएगा जिसका नाम है my_image.
- imput_2 type का WorkflowImageData जो runtime पर एक अन्य step द्वारा उत्पन्न किया जाएगा, जिसका नाम है image_transformation

Manifest में parameters जोड़ना

अब आइए उस parameter को जोड़ें जो step execution को प्रभावित करेगा।

Manifest में parameter जोड़ना

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` के अलावा सभी properties को या तो 
    hardcoded parameters या data selectors के रूप में माना जाता है। Data selectors ऐसी strings हैं 
    जो `$steps.` या `$inputs.` से शुरू होती हैं और runtime में उपलब्ध data के लिए references दर्शाती हैं - 
    इस मामले में हम आमतौर पर data के kinds निर्दिष्ट करते हैं
    ताकि compiler को पता चले कि हम data को कैसा देखना अपेक्षित करते हैं।
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए पहली image",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए दूसरी image",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="यह threshold मानने के लिए कि images समान हैं",
    )

लाइन 9 imports float_zero_to_one kind वह परिभाषा जिसका उपयोग parameter को परिभाषित करने के लिए किया जाएगा।
लाइन में 27 हम एक parameter परिभाषित करना शुरू करते हैं जिसका नाम है similarity_threshold. Manifest या तो float values या workflow input के लिए selector स्वीकार करेगा जिसका kind float_zero_to_oneहै, जो लाइन में import किया गया है 9.

manifest की ऐसी परिभाषा Workflow definition में निम्न step declaration को संभाल सकती है:

{
  "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
}

Block outputs घोषित करना

हमने अपने block के लिए inputs सफलतापूर्वक परिभाषित कर दिए हैं, लेकिन blocks को सफलतापूर्वक चलाने के लिए आवश्यक कुछ तत्व अभी भी गायब हैं। आइए block outputs परिभाषित करें।

Block outputs घोषित करना

आवश्यक जानकारी का न्यूनतम सेट outputs का विवरण है। इसके अतिरिक्त, block stability बढ़ाने के लिए, हम execution engine compatibility के बारे में जानकारी देने की सलाह देते हैं।

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="Similarity calculate करने के लिए पहली image",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए दूसरी image",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="यह threshold मानने के लिए कि images समान हैं",
    )

    @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 imports बूलियन kind आउटपुट परिभाषाओं में उपयोग किए जाने के लिए
पंक्तियाँ 32-39 ब्लॉक के आउटपुट निर्दिष्ट करने के लिए क्लास मेथड घोषित करें - सूची की प्रत्येक प्रविष्टि प्रत्येक बैच तत्व और उसके लिए एक रिटर्न प्रॉपर्टी घोषित करती है kind. हमारा ब्लॉक बूलियन फ़्लैग लौटाएगा images_match प्रत्येक छवि-युग्म के लिए।
पंक्तियाँ 41-43 Execution Engine के साथ ब्लॉक की संगतता घोषित करें - देखें वर्शनिंग पेज अधिक विवरण के लिए

इन परिवर्तनों के परिणामस्वरूप:

Execution Engine समझेगा कि इस ब्लॉक के आधार पर बनाए गए स्टेप्स को निर्दिष्ट आउटपुट्स देने चाहिए, और अन्य स्टेप्स अपने इनपुट्स में उन आउटपुट्स को संदर्भित कर सकते हैं
ब्लॉक्स लोड करने की व्यवस्था ब्लॉक को लोड नहीं करेगी क्योंकि Execution Engine संस्करण में नहीं है v1

और जानें: डायनामिक आउटपुट्स

कुछ ब्लॉक classmethod का उपयोग करके अपने आउटपुट्स को मनमाने ढंग से परिभाषित करने में सक्षम नहीं हो सकते - पार्सिंग के बाद उपलब्ध step manifest की सामग्री से स्वतंत्र। इसे समर्थन देने के लिए हमने निम्नलिखित परंपरा पेश की:

classmethod describe_outputs(...) को नाम के एक तत्व वाली सूची लौटानी चाहिए * और kind * (उर्फ WILDCARD_KIND)
इसके अतिरिक्त, block manifest को instance method लागू करनी चाहिए get_actual_outputs(...) जो भरे गए manifest डेटा के आधार पर उत्पन्न किए जा सकने वाले वास्तविक आउटपुट्स की सूची प्रदान करती है

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="Similarity calculate करने के लिए पहली image",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए दूसरी image",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="यह threshold मानने के लिए कि images समान हैं",
    )
    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
        ]

ब्लॉक क्लास की परिभाषा

इस चरण पर, हमारे सरल ब्लॉक का manifest तैयार है, हम अपने उदाहरण के साथ आगे बढ़ेंगे। आप उन्नत विषय अनुभाग को अधिक विवरण के लिए देख सकते हैं, जो अभी सिर्फ ध्यान भटका देगा।

मूल कार्यान्वयन

Manifest तैयार होने के बाद, हम ब्लॉक का बेसलाइन कार्यान्वयन तैयार कर सकते हैं।

ब्लॉक स्कैफ़ोल्डिंग

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="Similarity calculate करने के लिए पहली image",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए दूसरी image",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="यह threshold मानने के लिए कि images समान हैं",
    )

    @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

पंक्तियाँ 1, 5-6 और 8-11 आवश्यक प्रतीकों को ठीक से परिभाषित करने के लिए आयात संरचना में परिवर्तन जोड़े गए हैं, ताकि ब्लॉक क्लास और उसके सभी मेथड सिग्नेचर सही तरह से परिभाषित हो सकें
पंक्तियाँ 53-55 क्लास मेथड को परिभाषित करता है get_manifest(...) ताकि पहले बनाई गई manifest क्लास को बस वापस किया जा सके
पंक्तियाँ 57-63 परिभाषित करें run(...) फ़ंक्शन, जिसे Execution Engine इच्छित परिणाम पाने के लिए डेटा के साथ कॉल करेगा। कृपया ध्यान दें कि इनपुट्स को परिभाषित करने वाले manifest फ़ील्ड्स image kind के रूप में चिह्नित हैं WorkflowImageData - जो कि image kind की आंतरिक डेटा निरूपण के अनुरूप है, जैसा कि 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="Similarity calculate करने के लिए पहली image",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए दूसरी image",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="यह threshold मानने के लिए कि images समान हैं",
    )

    @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 इम्पोर्ट करते हैं
पंक्तियाँ 55-57 ब्लॉक कंस्ट्रक्टर को परिभाषित करता है, इसके कारण ब्लॉक की स्थिति एक बार प्रारंभ होती है और run(...) मेथड के लगातार आह्वानों के दौरान जीवित रहती है - उदाहरण के लिए जब Execution Engine वीडियो के लगातार फ़्रेम्स पर चलता है
पंक्तियाँ 69-80 ब्लॉक कार्यक्षमता का कार्यान्वयन प्रदान करें - Workflows इकोसिस्टम के संबंध में विवरण वास्तव में महत्वपूर्ण नहीं हैं, लेकिन कुछ विवरण हैं जिन पर आपको ध्यान देना चाहिए:
- पंक्तियाँ 69 और 70 का उपयोग करें WorkflowImageData अमूर्तन, यह दर्शाते हुए कि numpy_image प्रॉपर्टी का उपयोग प्राप्त करने के लिए किया जा सकता है np.ndarray Workflows में छवियों के आंतरिक निरूपण से। हम सलाह देते हैं कि WorkflowImageData की शेष प्रॉपर्टीज़ को और जानने के लिए खोजें।
- वर्कफ़्लो ब्लॉक निष्पादन का परिणाम, पंक्तियों में घोषित 78-80 हमारे मामले में केवल एक डिक्शनरी है जिसकी कुंजियाँ manifest में घोषित आउटपुट्स के नाम हैं, पंक्ति 43. सभी घोषित आउटपुट्स प्रदान करना सुनिश्चित करें - अन्यथा Execution Engine त्रुटि देगा।

ब्लॉक को `प्लगइन में उजागर करना`

अब, आपका ब्लॉक उपयोग के लिए तैयार है, लेकिन Execution Engine को इसके अस्तित्व की जानकारी नहीं है। इसका कारण यह है कि कोई पंजीकृत प्लगइन अभी-अभी बनाए गए ब्लॉक को निर्यात नहीं करता। ब्लॉक्स को पैकेज/बंडल करने के विवरण अलग पेजपर कवर किए गए हैं, लेकिन करने के लिए शेष चीज़ यह है कि अपने प्लगइन के द्वारा लौटाई गई सूची में ब्लॉक क्लास जोड़ें load_blocks(...) फ़ंक्शन:

# आपके प्लगइन का __init__.py (या roboflow_core प्लगइन, यदि आप सीधे `inference` में योगदान करते हैं)

from my_plugin.images_similarity.v1 import  ImagesSimilarityBlock  
# यह उदाहरण इम्पोर्ट है! समायोजन की आवश्यकता है

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="Similarity calculate करने के लिए पहली image",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="Similarity calculate करने के लिए दूसरी image",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="यह threshold मानने के लिए कि images समान हैं",
    )

    @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 imports Batch Workflows लाइब्रेरी के कोर से - यह क्लास एक कंटेनर का प्रतिनिधित्व करती है जो बैच तत्वों को रखने के लिए सूची के समान (लेकिन केवल-पठन) है
पंक्तियाँ 40-42 क्लास मेथड परिभाषित करें जो ब्लॉक के डिफ़ॉल्ट व्यवहार को बदलता है और उसे बैच प्रोसेस करने में सक्षम बनाता है - हम प्रत्येक पैरामीटर को चिह्नित कर रहे हैं जिसे run(...) मेथड बैच-उन्मुख के रूप में पहचानता है.
ऊपर प्रस्तुत परिवर्तनों ने run(...) मेथड के सिग्नेचर को बदल दिया है, अब image_1 और image_2 के उदाहरण नहीं हैं WorkflowImageDataबल्कि इस प्रकार के तत्वों के बैच हैं। महत्वपूर्ण नोट: कई बैच-उन्मुख पैरामीटर्स होने पर हम अपेक्षा करते हैं कि उन बैचों के तत्व संबंधित स्थितियों पर एक-दूसरे से जुड़े हों - ताकि हमारे ब्लॉक द्वारा तुलना image_1[1] में image_2[1] वास्तव में तार्किक रूप से सार्थक संचालन करता है।
पंक्तियाँ 74-77, 85-86 वे परिवर्तन प्रस्तुत करें जो सभी बैच तत्वों पर रन प्रोसेसिंग करने के लिए पेश करने पड़े - यह दिखाते हुए कि आवश्यकता होने पर बैच तत्वों पर कैसे iterate करें
यह ध्यान देना महत्वपूर्ण है कि आउटपुट्स पंक्ति में कैसे बनाए जाते हैं 85 - बैच के प्रत्येक तत्व को सूची में उसकी अपनी प्रविष्टि दी जाएगी जो run(...) मेथड से लौटाई जाती है। क्रम बैच तत्वों के क्रम के अनुरूप होना चाहिए। प्रत्येक आउटपुट डिक्शनरी में ब्लॉक आउटपुट्स में घोषित सभी कुंजियाँ प्रदान की जानी चाहिए।

ऐसे इनपुट्स जो बैच और स्केलर दोनों स्वीकार करते हैं

यह अपेक्षाकृत असंभाव्य हैलेकिन ऐसा हो सकता है कि आपके ब्लॉक को एक ही इनपुट पैरामीटर में बैच-उन्मुख डेटा और स्केलर्स दोनों स्वीकार करने की आवश्यकता हो। Execution Engine इसे पहचानता है get_parameters_accepting_batches_and_scalars(...) ब्लॉक manifest की मेथड का उपयोग करके। नीचे दिए गए उदाहरण पर नज़र डालें:

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

पंक्तियाँ 20-22 वे manifest पैरामीटर्स निर्दिष्ट करें जिनसे मिश्रित (स्केलर और बैच-उन्मुख दोनों) इनपुट डेटा स्वीकार करने की अपेक्षा है - ध्यान दें कि इस चरण पर परिभाषा में पिछले उदाहरणों की तुलना में कोई अंतर नहीं है।
पंक्तियाँ 24-26 निर्दिष्ट करें get_parameters_accepting_batches_and_scalars(...) मेथड ताकि Execution Engine को बताया जा सके कि ब्लॉक run(...) मेथड निर्दिष्ट पैरामीटर्स के लिए स्केलर और बैच-उन्मुख दोनों इनपुट्स को संभाल सकता है।
पंक्तियाँ 45-47 मिश्रित प्रकृति वाले पैरामीटर्स को run(...) मेथड सिग्नेचर में दर्शाएँ।
लाइन 49 यह दर्शाता है कि हमें अपेक्षित आउटपुट आकार का ट्रैक रखना होगा ब्लॉक लॉजिक के भीतर. इसी कारण मिश्रित इनपुट्स वाले ब्लॉक्स को लागू करना काफी कठिन है। सामान्यतः, जब ब्लॉक run(...) मेथड स्केलर्स पर काम करता है - अधिकांश मामलों में (अपवाद नीचे वर्णित होंगे) - मेथड एकल आउटपुट डिक्शनरी बनाता है। इसी तरह, जब बैच-उन्मुख इनपुट्स स्वीकार किए जाते हैं - वे इनपुट्स अपेक्षित आउटपुट आकार को परिभाषित करते हैं। हालाँकि, इस मामले में, हमें बैचों को मैन्युअल रूप से पहचानना और उनके आकार पकड़ना होगा।
पंक्तियाँ 50-54 यह दिखाएँ कि मिश्रित पैरामीटर्स से निपटते समय हम सामान्यतः कैसे व्यवहार करते हैं - जब बैच-उन्मुख डेटा का पता चलता है तो अलग लॉजिक लागू करते हैं
जैसा पहले उल्लेख किया गया, आउटपुट निर्माण को भी मिश्रित इनपुट्स की प्रकृति के अनुसार समायोजित करना होगा - जिसे पंक्तियों में दर्शाया गया है 65-70

फ़्लो-कंट्रोल ब्लॉक का कार्यान्वयन

फ़्लो-कंट्रोल ब्लॉक्स अन्य उन ब्लॉक्स से काफी भिन्न होते हैं जो केवल डेटा प्रोसेस करते हैं। यहाँ हम दिखाएँगे कि फ़्लो कंट्रोल ब्लॉक कैसे बनाया जाए, लेकिन पहले - थोड़ा सिद्धांत:

फ़्लो-कंट्रोल ब्लॉक वह ब्लॉक है जो अपने manifest में स्टेप सेलेक्टर्स के साथ संगतता घोषित करता है (स्टेप के लिए सेलेक्टर इस प्रकार परिभाषित है $steps.{step_name} - स्टेप आउटपुट सेलेक्टर के समान, लेकिन आउटपुट नाम के विनिर्देश के बिना)
फ़्लो-कंट्रोल ब्लॉक्स आउटपुट्स पंजीकृत नहीं कर सकते, उनका उद्देश्य लौटाना है FlowControl ऑब्जेक्ट्स
FlowControl ऑब्जेक्ट अगले स्टेप्स निर्दिष्ट करता है (स्टेप manifest में दिए गए सेलेक्टर्स से) जिन्हें दिए गए बैच तत्व (SIMD फ़्लो-कंट्रोल) या पूरे वर्कफ़्लो निष्पादन (गैर-SIMD फ़्लो-कंट्रोल) के लिए अगला लेना चाहिए

फ़्लो-कंट्रोल का कार्यान्वयन

उदाहरण रैंडम कंटिन्यू ब्लॉक के कार्यान्वयन को प्रदान करता है और टिप्पणी के रूप में हटाता है

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(
        description="यदि अभिव्यक्ति 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 स्टेप सेलेक्टर के लिए type annotation इम्पोर्ट करता है, जिसका उपयोग Execution Engine को सूचित करने के लिए किया जाएगा कि ब्लॉक प्रवाह को नियंत्रित करता है
लाइन 14 imports FlowControl वह क्लास जो फ़्लो-कंट्रोल ब्लॉक से मिलने वाला एकमात्र उपयुक्त उत्तर है
लाइन 28 स्टेप सेलेक्टर्स की सूची को परिभाषित करता है जो प्रभावी रूप से ब्लॉक को फ़्लो-कंट्रोल ब्लॉक में बदल देता है
पंक्तियाँ 55 और 56 आउटपुट बनाने का तरीका दिखाएँ - FlowControl ऑब्जेक्ट context को स्वीकार करता है None, स्ट्रिंग या स्ट्रिंग्स की सूची - None बैच तत्व के लिए प्रवाह समाप्ति का प्रतिनिधित्व करें, स्ट्रिंग्स से अपेक्षा है कि वे इनपुट में पास किए गए अगले स्टेप्स के लिए सेलेक्टर्स हों।

फ़्लो-कंट्रोल का कार्यान्वयन - बैच संस्करण

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(
        description="यदि अभिव्यक्ति 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 स्टेप सेलेक्टर के लिए type annotation इम्पोर्ट करता है, जिसका उपयोग Execution Engine को सूचित करने के लिए किया जाएगा कि ब्लॉक प्रवाह को नियंत्रित करता है
लाइन 15 imports FlowControl वह क्लास जो फ़्लो-कंट्रोल ब्लॉक से मिलने वाला एकमात्र उपयुक्त उत्तर है
पंक्तियाँ 29-32 स्टेप सेलेक्टर्स की सूची को परिभाषित करता है जो प्रभावी रूप से ब्लॉक को फ़्लो-कंट्रोल ब्लॉक में बदल देता है
पंक्तियाँ 38-40 का परिभाषन शामिल करें get_parameters_accepting_batches(...) मेथड जो Execution Engine को बताता है कि ब्लॉक run(...) मेथड बैच-उन्मुख image पैरामीटर की अपेक्षा करता है।
लाइन 59 यह प्रकट करता है कि हमें flow-control मार्गदर्शन लौटाना होगा image बैच के प्रत्येक और हर तत्व के लिए।
इस उद्देश्य को प्राप्त करने के लिए, पंक्ति में 60 हम बैच की सामग्री पर iterate करते हैं।
पंक्तियाँ 61-63 आउटपुट बनाने का तरीका दिखाएँ - FlowControl ऑब्जेक्ट context को स्वीकार करता है None, स्ट्रिंग या स्ट्रिंग्स की सूची - None बैच तत्व के लिए प्रवाह समाप्ति का प्रतिनिधित्व करें, स्ट्रिंग्स से अपेक्षा है कि वे इनपुट में पास किए गए अगले स्टेप्स के लिए सेलेक्टर्स हों।

नेस्टेड सेलेक्टर्स

कुछ ब्लॉक्स को block manifest फ़ील्ड में सेलेक्टर्स की सूची या सेलेक्टर्स की डिक्शनरी प्रदान करने की आवश्यकता होगी। Execution Engine का संस्करण v1 केवल एक स्तर की nesting को समर्थन देता है - इसलिए सेलेक्टर्स की सूचियों की सूची या सेलेक्टर्स की सूची वाली डिक्शनरी को सही तरह से पहचाना नहीं जाएगा।

नेस्टेड सेलेक्टर्स के उपयोग को दिखाने वाले व्यावहारिक उपयोग मामले नीचे प्रस्तुत हैं।

परिवर्तनशील संख्या वाले मॉडलों की भविष्यवाणियों का फ्यूज़न

मान लें कि आप कई classifiers की भविष्यवाणियों पर बहुमत वोट लेने के लिए एक ब्लॉक बनाना चाहते हैं - तब आप चाहते हैं कि आपका run मेथड कुछ ऐसा दिखे:

# pseud-code here
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(
        description="स्टेप आउटपुट्स के लिए सेलेक्टर्स",
        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}

पंक्तियाँ 23-26 दिखाता है कि सेलेक्टर्स की सूची स्वीकार करने में सक्षम manifest फ़ील्ड कैसे परिभाषित करें
लाइन 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"  
  ]
}

डेटा रूपांतरणों वाला ब्लॉक जो डायनामिक पैरामीटर्स की अनुमति देता है

कभी-कभी, blocks को "named" selectors के समूह को स्वीकार करने की आवश्यकता हो सकती है, जिनके नाम और मान Workflow परिभाषा के निर्माता द्वारा निर्धारित किए जाने हैं। ऐसे मामलों में, block manifest को selectors की dictionary स्वीकार करनी चाहिए, जहाँ keys उन selectors के नाम के रूप में काम करती हैं।

नेस्टेड selectors - named selectors

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(
        description="स्टेप आउटपुट्स के लिए सेलेक्टर्स",
        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": ...}

पंक्तियाँ 22-25 यह दिखाता है कि manifest field को कैसे परिभाषित किया जाए जो selectors की dictionary स्वीकार करने में सक्षम हो - selector के नाम और मान के बीच mapping प्रदान करके
लाइन 46 दिखाता है कि ब्लॉक के इनपुट के रूप में क्या अपेक्षित है run(...) method - objects की एक dictionary, जिन्हें selectors द्वारा संदर्भित किया जाता है। यदि block batches स्वीकार करता है, तो इनपुट का प्रकार data फ़ील्ड का इनपुट प्रकार होगा Dict[str, Union[Batch[Any], Any]]. non-batch मामलों में, selector द्वारा संदर्भित non-batch-oriented data स्वतः broadcast किया जाता है, जबकि batches स्वीकार करने वाले blocks के लिए - Batch container केवल batch-oriented inputs को wrap करता है, जबकि अन्य inputs को singular values के रूप में पास किया जाता है।

ऐसा ब्लॉक निम्नलिखित स्टेप घोषणा के साथ संगत है:

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

व्यावहारिक प्रभाव निम्न होंगे:

के अंतर्गत data["a"] के अंदर run(...) आप model की predictions पाएंगे - जैसे sv.Detections यदि model_1 object-detection model है
के अंतर्गत data["b"] के अंदर run(...), आपको input parameter जिसका नाम my_parameter

इनपुट और आउटपुट dimensionality बनाम `run(...)` मेथड

block inputs की dimensionality method की signature को आकार देने में एक महत्वपूर्ण भूमिका निभाती है, और इसी कारण system inputs के बीच dimensionality levels के अंतर पर कड़ी सीमाएँ लागू करता है (जहाँ अधिकतम अनुमत अंतर run(...) )। यह प्रतिबंध blocks लिखते समय consistency और predictability सुनिश्चित करने के लिए महत्वपूर्ण है। 1). यह प्रतिबंध blocks लिखते समय consistency और predictability सुनिश्चित करने के लिए महत्वपूर्ण है।

यदि dimensionality के अंतर नियंत्रित नहीं किए गए, तो method की संरचना का अनुमान लगाना कठिन हो जाएगा, जिससे development कठिन और कम विश्वसनीय हो जाएगी। इसी कारण Workflow compilation प्रक्रिया के दौरान इस property का validation कड़ाई से लागू किया जाता है। run(...) method, जिससे development कठिन और कम विश्वसनीय हो जाएगी। इसी कारण Workflow compilation प्रक्रिया के दौरान इस property का validation कड़ाई से लागू किया जाता है।

इसी तरह, output dimensionality भी method signature और अपेक्षित output के format को प्रभावित करती है। ecosystem निम्न scenarios का समर्थन करता है:

सभी inputs की एक ही dimensionality और outputs नहीं बदलती dimensionality - baseline case
सभी inputs की एक ही dimensionality और output घटती है dimensionality
सभी inputs की एक ही dimensionality और output बढ़ती है dimensionality
inputs की अलग-अलग dimensionality और output को संदर्भ input

consistency सुनिश्चित करने और method signatures में ambiguity रोकने के लिए input/output dimensionalities के अन्य संयोजन अनुमत नहीं हैं।

batches disabled - `run(...)` method पर dimensionality का प्रभाव

output dimensionality में वृद्धि

इस उदाहरण में, हम predictions के आधार पर image का dynamic crop करते हैं।

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 manifest class output dimensionality offset घोषित करती है - मान 1 को 1 dimensionality level में जोड़ने के रूप में समझा जाना चाहिए
यह ध्यान दिलाता है कि पंक्ति 63, block आगे की processing से खाली images को हटाता है, लेकिन output dictionary के बजाय None का उपयोग करता है। इससे वही Execution Engine व्यवहार उपयोग में आता है जो conditional execution के लिए इस्तेमाल होता है - datapoint downstream processing से हटा दिया जाएगा (जब तक कि आगे ऐसी steps मौजूद न हों जो empty inputs मांगती हों)।
पंक्तियों में 64-65 एकल input के लिए परिणाम image और predictions एकत्र किए जाते हैं - इसका आशय dictionaries की एक सूची से है, जिसमें registered outputs सभी keys के रूप में शामिल हों। Execution engine समझ जाएगा कि step हर input element के लिए elements का batch लौटाता है और downstream steps के execution के दौरान track रखने के लिए indices की nested structures बनाएगा।

output dimensionality में कमी

इस उदाहरण में, block crops predictions को visualise करता है और सभी crop predictions को एक single output image में प्रस्तुत करने वाले tiles बनाता है।

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 manifest class output dimensionality offset घोषित करती है - मान -1 को dimensionality level में घटाने के रूप में समझा जाना चाहिए 1
पंक्तियों में 34-36 manifest class घोषित करती है run(...) method inputs जो auto-batch casting के अधीन होंगे, यह सुनिश्चित करते हुए कि signature हमेशा स्थिर रहे। Auto-batch casting Execution Engine में v0.1.6.0
देखें changelog अधिक विवरण के लिए।
पंक्तियों में 53-55 आप method signature पर output dimensionality में कमी के प्रभाव को देख सकते हैं। पहले दो inputs (पंक्ति में घोषित 36) कृत्रिम रूप से wrap किए जाते हैं Batch[] container, जबकि scalar_parameter primitive type बना रहता है। यह Execution Engine द्वारा output dimensionality में कमी होने पर स्वतः किया जाता है, जब सभी inputs की dimensionality समान हो, ताकि अंतिम dimensionality level पर मौजूद सभी elements तक पहुँच संभव हो सके। स्पष्ट रूप से, केवल top-level batch के उसी element से संबंधित elements को ही एक साथ group किया जाएगा। उदाहरण के लिए, यदि आपके पास दो input images थीं जिन्हें आपने crop किया - तो उन दो अलग-अलग images के crops अलग-अलग group किए जाएंगे।
पंक्तियाँ 65-66 यह दिखाता है कि output कैसे निर्मित होता है - एक एकल value लौटाई जाती है और उस value को Execution Engine द्वारा reduced dimensionality वाले output batch में index किया जाएगा

अलग-अलग input dimensionalities

इस उदाहरण में, block उन detections को merge करता है जो मूल image के crops के आधार पर predict की गई थीं - परिणामस्वरूप एकल detections प्रदान की जाती हैं जिनमें सभी आंशिक detections merged होती हैं।

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 manifest class input dimensionalities offset घोषित करती है, जो यह संकेत देता है image कि parameter top-level है और image_predictions predictions का nested batch है
जब भी अलग-अलग input dimensionalities घोषित की जाती हैं, dimensionality reference property को निर्दिष्ट करना आवश्यक होता है (देखें पंक्तियाँ 38-40) - इस dimensionality level का उपयोग output dimensionality की गणना के लिए किया जाएगा - इस विशेष मामले में, हम निर्दिष्ट करते हैं image. इस चुनाव का result के अपेक्षित format पर प्रभाव पड़ता है - चुने गए scenario में हमें सभी registered outputs keys के साथ एक single dictionary लौटानी होती है। यदि हमारा चुनाव image_predictionsहै, तो हम dictionaries की एक सूची लौटाएँगे (जिसका आकार image_predictions batch की लंबाई के बराबर होगा)। दूसरे शब्दों में, get_dimensionality_reference_property(...) को output से किस dimensionality level को संबद्ध किया जाना चाहिए।
पंक्तियाँ 63-64 पंक्तियों में निर्दिष्ट dimensionality offsets के प्रभाव को प्रस्तुत करते हैं 31-36. यह स्पष्ट रूप से दिखाई देता है कि image_predictions के संदर्भ में एक nested batch है image. स्पष्ट रूप से, केवल specific images से संबंधित nested predictions को batch में group किया जाता है और runtime में method को प्रदान किया जाता है।
जैसा कि पहले उल्लेख किया गया है, पंक्ति 69 output को single dictionary के रूप में बनाती है, क्योंकि हम output को image की dimensionality level पर register करते हैं

batches enabled - `run(...)` method पर dimensionality का प्रभाव

output dimensionality में वृद्धि

इस उदाहरण में, हम predictions के आधार पर image का dynamic crop करते हैं।

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 manifest घोषित करती है कि block inputs के batches स्वीकार करता है
पंक्तियों में 33-35 manifest class output dimensionality offset घोषित करती है - मान 1 को 1 dimensionality level में जोड़ने के रूप में समझा जाना चाहिए
पंक्तियों में 55-66, input parameters की signature यह दर्शाती है कि run(...) method समान dimensionality वाले inputs के विरुद्ध चलता है और वे inputs batches में प्रदान किए जाते हैं
यह ध्यान दिलाता है कि पंक्ति 70, block आगे की processing से खाली images को हटाता है, लेकिन output dictionary के बजाय None का उपयोग करता है। इससे वही Execution Engine व्यवहार उपयोग में आता है जो conditional execution के लिए इस्तेमाल होता है - datapoint downstream processing से हटा दिया जाएगा (जब तक कि आगे ऐसी steps मौजूद न हों जो empty inputs मांगती हों)।
output का निर्माण, पंक्तियों में प्रस्तुत 71-73 दो nesting levels को इंगित करता है। सबसे पहले, block batches पर कार्य करता है, इसलिए outputs की एक सूची लौटाने की अपेक्षा की जाती है, जहाँ प्रत्येक input batch element के लिए एक output होगा। इसके अतिरिक्त, प्रत्येक input batch element के लिए यह output element nested batch निकला - इसलिए प्रत्येक input image और prediction के लिए, block outputs की एक सूची उत्पन्न करता है - उस सूची के तत्व dictionaries होते हैं जो प्रत्येक घोषित output के लिए मान प्रदान करते हैं।

output dimensionality में कमी

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

पंक्तियाँ 29-31 manifest यह अपेक्षा करती है कि block input के रूप में batches ले
पंक्तियों में 33-35 manifest class output dimensionality offset घोषित करती है - मान -1 को dimensionality level में घटाने के रूप में समझा जाना चाहिए 1
पंक्तियों में 52-53 आप output dimensionality में कमी और batch processing का method signature पर प्रभाव देख सकते हैं। पहला "layer" Batch[] का परिणाम है कि manifest ने घोषित किया है कि block inputs के batches स्वीकार करता है। दूसरा "layer" output dimensionality में कमी के कारण आता है। Execution Engine कम की जाने वाली dimension को input में प्रदान किए गए अतिरिक्त Batch[] container में लपेटता है, ताकि programmer उन सभी nested batch elements को एकत्र कर सके जो किसी विशिष्ट top-level batch element से संबंधित हैं।
पंक्तियाँ 66-67 यह दिखाता है कि output कैसे निर्मित होता है - प्रत्येक top-level batch element के लिए, block सभी crops और predictions को aggregate करता है और एक single tile बनाता है। चूँकि block inputs के batches स्वीकार करता है, यह प्रक्रिया प्रत्येक top-level batch element के लिए एक tile के साथ समाप्त होती है - इसलिए dictionaries की एक सूची लौटाई जानी अपेक्षित है।

अलग-अलग input dimensionalities

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

पंक्तियाँ 31-33 manifest यह अपेक्षा करती है कि block input के रूप में batches ले
पंक्तियों में 35-40 manifest class input dimensionalities offset घोषित करती है, जो यह संकेत देता है image कि parameter top-level है और image_predictions predictions का nested batch है
जब भी अलग-अलग input dimensionalities घोषित की जाती हैं, dimensionality reference property को निर्दिष्ट करना आवश्यक होता है (देखें पंक्तियाँ 42-44) - इस dimensionality level का उपयोग output dimensionality की गणना के लिए किया जाएगा - इस विशेष मामले में, हम निर्दिष्ट करते हैं image. इस चुनाव का result के अपेक्षित format पर प्रभाव पड़ता है - चुने गए scenario में हमें nested batch के प्रत्येक element के लिए एक single dictionary लौटानी होती है। यदि हमारा चुनाव image batch. यदि हमारा चुनाव image_predictionsहै, तो हम dictionaries की एक सूची लौटाएँगे (जिसका आकार nested image_predictions batch की लंबाई के बराबर होगा) प्रत्येक input image batch element के लिए।
पंक्तियाँ 66-67 पंक्तियों में निर्दिष्ट dimensionality offsets के प्रभाव को प्रस्तुत करते हैं 35-40 साथ ही पंक्तियों में batch processing की घोषणा 32-34. पहले "layer" का Batch[] container बाद वाले, nested Batch[Batch[]] के लिए images_predictions की input dimensionality offset की परिभाषा से आता है। यह स्पष्ट रूप से दिखाई देता है कि image_predictions विशिष्ट तत्वों से संबंधित predictions का batch holds करता है image बैच के प्रत्येक और हर तत्व के लिए।
जैसा कि पहले उल्लेख किया गया है, पंक्तियाँ 76-77 output को प्रत्येक element के लिए single dictionary के रूप में बनाती हैं image batch

खाली inputs स्वीकार करने वाला block

जैसा कि पहले चर्चा की गई है, Workflow के निष्पादन के दौरान कुछ batch elements "empty" हो सकते हैं। यह कई कारणों से हो सकता है:

Flow-control mechanisms: निष्पादन की कुछ branches विशिष्ट batch elements को mask कर सकती हैं, जिससे वे subsequent steps में processed होने से बच जाती हैं।
Data-processing blocks में: कुछ मामलों में, block किसी विशिष्ट data point के लिए meaningful output उत्पन्न करने में सक्षम नहीं हो सकता। उदाहरण के लिए, Dynamic Crop block bounding box size शून्य होने पर cropped image नहीं बना सकता।

कुछ blocks इन empty inputs को संभालने के लिए बनाए गए हैं, जैसे ऐसा block जो missing outputs को default values से बदल सकता है। यह block Workflow में structured outputs बनाते समय विशेष रूप से उपयोगी हो सकता है, यह सुनिश्चित करते हुए कि कुछ elements empty होने पर भी output में अनुपस्थित elements न रहें, जिससे उसे parse करना कठिन हो जाएगा।

खाली inputs स्वीकार करने वाला block

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 आपको ऐसा declaration मिल सकता है जो बताता है कि block empty inputs स्वीकार करता है
पंक्तियों का परिणाम है 20-22 का प्रभाव पंक्ति में दिखाई देता है 41, जब signature बताती है कि input Batch में empty elements हो सकते हैं जिन्हें संभालना आवश्यक है। वास्तव में - block एक "artificial" output उत्पन्न करता है जो empty value के स्थान पर आता है, जिससे उन outputs का इस block के output को संदर्भित करने वाले और empty inputs स्वीकार न करने वाले blocks के लिए "visible" होना संभव हो जाता है। आपको मानना चाहिए कि हर input जिसे Execution Engine runtime में generated data से substitute करता है, optional elements प्रदान कर सकता है।

कस्टम constructor parameters वाला block

कुछ blocks को काम करने के लिए बाहर की दुनिया द्वारा बनाए गए objects की आवश्यकता हो सकती है। ऐसे scenario में, Workflows Execution Engine का काम उन entities को block तक पहुँचाना है, जिससे उनका उपयोग संभव हो सके। इस mechanism का वर्णन Workflows Compiler प्रस्तुत करने वाले पृष्ठ पर किया गया है, क्योंकि यही घटक blocks classes से steps के dynamic construction के लिए जिम्मेदार है।

Constructor parameters होने चाहिए:

block द्वारा अनुरोधित - class method का उपयोग करके WorkflowBlock.get_init_parameters(...)
Workflows Execution Engine चलाने वाले environment में प्रदान किए गए:
- सीधे, जैसा कि इस उदाहरण में दिखाया गया है
- defaults का उपयोग करके Workflow plugin के लिए registered

आइए देखें कि block को परिभाषित करते समय init parameters कैसे अनुरोध किए जाएँ।

Constructor parameters का अनुरोध करने वाला block

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

पंक्तियाँ 30-31 ऐसा class constructor घोषित करें जो parameter-free न हो
Execution Engine को यह बताने के लिए कि block को custom initialisation की आवश्यकता है, get_init_parameters(...) पंक्तियों में method 33-35 उन सभी parameters के नामों की सूची बनाता है जिन्हें प्रदान किया जाना चाहिए

Air-gapped / offline उपलब्धता metadata

ऐसे environments के लिए blocks बनाते समय जो इंटरनेट पहुँच के बिना चल सकते हैं (air-gapped deployments), WorkflowBlockManifest तीन वैकल्पिक classmethods प्रदान करता है। air-gapped workflow builder को यह निर्धारित करने में मदद करने के लिए इन्हें override करें कि कौन से blocks offline उपयोग योग्य हैं।

तीनों के समझदारी भरे defaults हैं, इसलिए existing blocks को कोई परिवर्तन नहीं.

`get_air_gapped_availability()`

यह घोषित करता है कि block इंटरनेट के बिना काम कर सकता है या नहीं। इसे उन blocks में override करें जो cloud APIs (OpenAI, Anthropic, आदि) को call करते हैं।

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) - pure-logic blocks, local-network blocks, और किसी भी ऐसे block के लिए उपयुक्त है जिसे external connectivity की आवश्यकता नहीं है।

`get_supported_model_variants()`

ऐसे foundation-model blocks के लिए जिनके weights को स्थानीय रूप से pre-cache किया जा सकता है, model variant IDs की सूची लौटाएँ। यदि कोई भी variant के पास cached artifacts हों तो block को offline उपलब्ध माना जाता है।

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, जिसका अर्थ है कि block स्थानीय रूप से cached model weights पर निर्भर नहीं करता।

`get_compatible_task_types()`

Roboflow-model ब्लॉकों के लिए जो उपयोगकर्ता-प्रशिक्षित मॉडल स्वीकार करते हैं, उन टास्क प्रकारों को लौटाएँ जिन्हें यह ब्लॉक संभाल सकता है। एयर-गैप्ड बिल्डर इसका उपयोग कैश किए गए उपयोगकर्ता मॉडलों को संगत ब्लॉकों से मिलाने के लिए करता है।

class BlockManifest(WorkflowBlockManifest):
    # ...

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

डिफ़ॉल्ट रूप से लौटाता है None - उन ब्लॉकों के लिए उपयुक्त है जो Roboflow मॉडल द्वारा पैरामीटराइज़्ड नहीं हैं (foundation models, logic blocks, sinks, आदि)।

रनटाइम प्रतिबंध

कुछ ब्लॉक अलग तरह से व्यवहार करते हैं - या पूरी तरह विफल हो जाते हैं - इस पर निर्भर करते हुए कि वे किस रनटाइम में तैनात हैं (hosted serverless, dedicated deployment, self-hosted, inference pipeline), स्टेप execution mode (local vs. remote), और input mode (image vs. video)। Override get_restrictions() में WorkflowBlockManifest इन चेतावनियों को ब्लॉक के भीतर एक बार घोषित करने के लिए, ताकि execution engine, schema endpoint, और auto-generated block gallery सभी उन्हें एकसमान रूप से दिखा सकें।

डिफ़ॉल्ट रूप से लौटाता है []तो मौजूदा ब्लॉकों को कोई परिवर्तन नहीं.

गंभीरता: `soft` बनाम `hard`

प्रत्येक प्रतिबंध के साथ एक गंभीरता:

Severity.SOFT - ब्लॉक पूरी तरह चलता है और सही output shape लौटाता है, लेकिन मान खराब या अर्थहीन हो जाते हैं (जैसे tracker IDs अनुरोधों के बीच रीसेट हो जाते हैं, cooldown throttle नहीं करता, फ़ाइल ephemeral disk पर लिखी जाती है)। workflow फिर भी चलता रहता है; परिणाम बस वैसा नहीं होता जैसा उपयोगकर्ता अपेक्षा करता है।
Severity.HARD - ब्लॉक नहीं चलता / exception फेंकता है / इस रनटाइम में उपयोगी output नहीं दे सकता। engine को compile करने से इनकार करना चाहिए या fail-fast होना चाहिए।

एक प्रतिबंध की scope तय करना

एक RuntimeRestriction तीनों अक्षों में से किसी भी संयोजन के साथ स्वयं को scope कर सकता है। जब कोई अक्ष Noneके रूप में छोड़ा जाता है, तो प्रतिबंध उस अक्ष के हर मान पर लागू होता है।

applies_to_runtimes: Runtime.HOSTED_SERVERLESS, Runtime.DEDICATED_DEPLOYMENT, Runtime.SELF_HOSTED_CPU, Runtime.SELF_HOSTED_GPU, Runtime.INFERENCE_PIPELINE.
applies_to_step_execution_modes: StepExecutionMode.LOCAL, StepExecutionMode.REMOTE.
applies_to_input_modes: RuntimeInputMode.IMAGE, RuntimeInputMode.VIDEO.

The note field विफलता मोड या degraded behavior का एक-पंक्ति, मानव-पठनीय विवरण है - बताइए क्या होता है (जैसे "track_ids requests के बीच reset हो जाते हैं", "ephemeral /tmp पर लिखता है"), न कि अमूर्त पूर्व-शर्तें।

साझा presets

अधिकांश चेतावनियाँ कुछ सामान्य पैटर्नों में आती हैं, इसलिए कोडबेस में शब्दावली एकसमान रखने के लिए dataclasses के साथ reusable presets भी export किए जाते हैं। पहले इनका उपयोग करें:

STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION - उन video-tracking / counting / aggregation blocks के लिए जिनकी per-video state process memory में रहती है और stateless HTTP requests के बीच रीसेट हो जाती है।
COOLDOWN_HTTP_SOFT_RESTRICTION - उन blocks के लिए जिनका cooldown / rate-limit timer process memory में रखा जाता है और इसलिए multi-replica HTTP runtimes पर throttle नहीं करता।
STILL_IMAGE_INPUT_SOFT_RESTRICTION - उन blocks के लिए जो temporal context (video या repeated frames) पर निर्भर करते हैं और still images पर बहुत कम या कोई लाभ नहीं देते।

उदाहरण

एक line-counter block जो per-video state बनाए रखता है और still image पर अर्थहीन भी है, दोनों प्रतिबंध घोषित करता है:

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,
        ]

एक custom restriction (उदा. एक Severity.HARD block जिसे GPU hardware चाहिए जो hosted serverless runtime पर उपलब्ध नहीं है) inline घोषित किया जाता है:

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,
                ],
            ),
        ]

इस तरह घोषित प्रतिबंध तीन जगहों पर दिखाई देते हैं:

The describe_interface HTTP payload (via RuntimeRestriction.to_dict()), ताकि workflow clients और builders workflow चलाने से पहले उपयोगकर्ताओं को चेतावनी दे सकें।
block के लिए auto-generated block gallery पेज पर, "Runtime compatibility" सेक्शन के तहत, ठीक Properties.
execution engine, जो Severity.HARD current runtime के लिए प्रतिबंधों पर fail-fast करने का विकल्प चुन सकता है।

निर्भर संसाधनों की घोषणा करना

कई blocks को चलने के समय बाहरी संसाधनों की आवश्यकता होती है: Roboflow-trained या foundation model weights, Roboflow projects (जैसे active-learning targets, dataset-upload destinations) या third-party hosted models (OpenAI, Anthropic, OpenRouter, ...). जब तक कोई workflow वास्तव में execute नहीं होता, सिस्टम में कोई नहीं जानता कि उसे किन संसाधनों की ज़रूरत होगी।

Override करना discover_dependent_resources() को block manifest पर करने से यह अंतर भर जाता है: यह method parsed manifest instanceपर call की जाती है, इसलिए यह किसी दिए गए step के concrete field values से declaration निकाल सकती है। इससे callers पूरे workflow के resources को static रूप से — compile time पर, बिना कुछ execute किए — enumerate कर सकते हैं, जिससे model weights को पहले से load करना (predictable execution times के लिए) या यह upfront सत्यापित करना कि किसी API key की पहुँच हर संदर्भित model और project तक है, जैसे use cases संभव होते हैं।

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

डिफ़ॉल्ट रूप से लौटाता है None, जिसका अर्थ है कि block घोषित नहीं करता अपनी dependencies — callers को इसे अज्ञातमानना होगा। यह जानबूझकर []से अलग है, जो स्पष्ट रूप से घोषित करता है कि block को किसी बाहरी संसाधन की आवश्यकता नहीं है। मौजूदा blocks में कोई बदलाव आवश्यक नहीं; जो blocks संसाधनों का उपयोग करते हैं उन्हें override करना चाहिए।

resource envelope

Blocks resource shapes का आविष्कार नहीं करते — envelope Execution Engine द्वारा नियंत्रित होता है। एक DependentResource एक resource_type को उस type के लिए पंजीकृत typed (pydantic) metadata entity के साथ जोड़ता है:

DependentResourceType.ROBOFLOW_PLATFORM_MODEL → RoboflowPlatformModelMetadata(model_id, required_action, execution_location) — Roboflow platform के माध्यम से serve किए गए models, जिनमें synthesized ids वाले foundation / core models भी शामिल हैं (जैसे clip/ViT-B-32). required_action उपयोग की प्रकृति बताता है: ModelRequiredAction.EXECUTION (weights खींचे जाते हैं या inference अनुरोधित होता है) या ModelRequiredAction.ACCESS (model entity को केवल platform पर पहुँच योग्य होना चाहिए — जैसे कोई monitoring sink उससे metadata जोड़ रहा हो)। execution के लिए, execution_location है LOCAL, REMOTE, या ENVIRONMENT_DEFINED जब locality runtime पर WORKFLOWS_STEP_EXECUTION_MODE द्वारा तय होती है और compile time पर निर्धारित नहीं की जा सकती (step execution mode पर dispatch करने वाले model blocks के लिए default)।
DependentResourceType.ROBOFLOW_PLATFORM_PROJECT → RoboflowPlatformProjectMetadata(project_url) — Roboflow projects जिनसे block पढ़ता है या जिनमें लिखता है।
DependentResourceType.THIRD_PARTY_MODEL → ThirdPartyModelMetadata(provider, model_id) — external provider द्वारा execute किए गए models; परिभाषा से remote execution।

Factory helpers implementations को one-liner बनाए रखते हैं: roboflow_platform_model(), roboflow_platform_project(), third_party_model().

Selector values और runtime resolution

Manifest fields concrete values के बजाय workflow selectors रख सकते हैं। Declaration ऐसे values को verbatim लौटाती है — block selectors को कभी resolve नहीं करता। Callers $inputs.<name> references को runtime parameters ज्ञात होने के बाद substitute कर सकते हैं; $steps.<name>.<property> references किसी भी तरह statically resolvable नहीं हैं। हर metadata entity requires_runtime_resolution() expose करती है, ताकि callers concrete identifier और ऐसे reference में फर्क कर सकें जिसे अभी resolve करना बाकी है।

जब अंतिम identifier एक function होता है field value का (जैसे family prefix clip/<version>का, एक catalog lookup), तो बदला गया input value अकेले executed id नहीं होता। ऐसी declarations एक model_id_resolver जोड़ती हैं — closure में अपनी सभी ज़रूरी चीज़ों के साथ एक callable — जो substituted value को final id में बदलता है। resolver केवल in-process सहायता है: इसे serialization, JSON schema और equality से बाहर रखा जाता है। in-process callers (जैसे engine का first-run pre-loading) input value substitute करने के बाद इसे invoke करते हैं। Resolver None लौटा सकता है ताकि value को statically unresolvable घोषित किया जा सके (final id उस एक value से अधिक पर निर्भर है) — ऐसे dependencies को callers छोड़ देते हैं और execution उन्हें resolve करता है; raise करना मतलब value वास्तव में अमान्य है।

जो `run()` load करता है

घोषित identifier बिल्कुल वही होना चाहिए जिसे run() request करेगा — version fields से synthesized ids सहित। यदि आपका block f"my_family/{self.version}" को load_core_model(...) call site पर बनाता है, तो declaration को वही id बनानी चाहिए (और जब version field में selector हो तो verbatim selector पर वापस जाना चाहिए)।

उदाहरण

एक model block जो अपना model और एक वैकल्पिक active-learning target project घोषित करता है:

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)]
        यदि self.active_learning_target_dataset is not None:
            resources.append(
                roboflow_platform_project(
                    project_url=self.active_learning_target_dataset
                )
            )
        return resources

एक block जिसका model id एक version field से synthesized होता है:

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):
            # Selector verbatim लौटाया जाता है; resolver final id बनाता है
            # once input value substitute हो जाती है।
            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}")]

एक sink जो execution किए बिना केवल model entity को संदर्भित करता है:

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,
            )
        ]

कब घोषित न करें

वे Fields जो केवल ले जाते हैं एक model-id- या project-kind value को generic payload के रूप में (जैसे query parameters के रूप में values forward करने वाला webhook sink) resource dependencies नहीं हैं — ऐसे blocks जानबूझकर default बनाए रखते हैं।

core repository के contributors को यह unit test (tests/workflows/unit_tests/core_steps/test_dependent_resources.py) सीमा की रक्षा करता है, यह मानना चाहिए: हर core block जिसका manifest roboflow_model_id या roboflow_project kind के साथ कोई field घोषित करता है, उसे या तो override करना होगा discover_dependent_resources() या स्पष्ट रूप से carry-only के रूप में allowlist किया जाना होगा।

जो blocks अपने model weights model manager के बाहर load करते हैं (देखें कि run() वास्तव में क्या करता है, न कि यह कि model_manager init parameters में दिखाई देता है या नहीं) उन्हें अभी इस method को implement नहीं करना चाहिए — उनकी dependencies undeclared रहती हैं (None).

Custom python blocks हमेशा रिपोर्ट करते हैं None: उनका code static analysis के लिए opaque है, इसलिए अज्ञात ही एकमात्र ईमानदार उत्तर है।

पिछलासंस्करणीकरण अगलाBlocks को plugin में bundling करना

अंतिम अपडेट 4 दिन पहले

क्या यह उपयोगी था?

Environment setup

Workflows UI का उपयोग करके अपने blocks चलाना

Workflows UI के बिना अपने blocks चलाना

नियमित contributors के लिए अनुशंसित तरीका

Prototypes

Block manifest

Manifest के लिए scaffolding

Inputs जोड़ना

Manifest में parameters जोड़ना

Block outputs घोषित करना

ब्लॉक क्लास की परिभाषा

मूल कार्यान्वयन

ब्लॉक लॉजिक के लिए कार्यान्वयन प्रदान करना

ब्लॉक को प्लगइन में उजागर करना

उन्नत विषय

इनपुट्स के बैचों को प्रोसेस करने वाले ब्लॉक

फ़्लो-कंट्रोल ब्लॉक का कार्यान्वयन

नेस्टेड सेलेक्टर्स

परिवर्तनशील संख्या वाले मॉडलों की भविष्यवाणियों का फ्यूज़न

डेटा रूपांतरणों वाला ब्लॉक जो डायनामिक पैरामीटर्स की अनुमति देता है

इनपुट और आउटपुट dimensionality बनाम run(...) मेथड

खाली inputs स्वीकार करने वाला block

कस्टम constructor parameters वाला block

Air-gapped / offline उपलब्धता metadata

get_air_gapped_availability()

get_supported_model_variants()

get_compatible_task_types()

रनटाइम प्रतिबंध

गंभीरता: soft बनाम hard

एक प्रतिबंध की scope तय करना

साझा presets

उदाहरण

निर्भर संसाधनों की घोषणा करना

resource envelope

Selector values और runtime resolution

जो run() load करता है

उदाहरण

कब घोषित न करें

ब्लॉक को `प्लगइन में उजागर करना`

इनपुट और आउटपुट dimensionality बनाम `run(...)` मेथड

`get_air_gapped_availability()`

`get_supported_model_variants()`

`get_compatible_task_types()`

गंभीरता: `soft` बनाम `hard`

जो `run()` load करता है