गतिशील Python ब्लॉक्स

Execution Engine द्वारा रनटाइम पर व्याख्यायित किए जाने वाले Python कोड के साथ Workflow Definition में सीधे blocks परिभाषित करें।

इस सुविधा के उत्पाद-स्तरीय परिचय के लिए देखें कस्टम ब्लॉक्स. इनलाइन कोड के बजाय पूरी Block class लिखने के लिए देखें एक Workflow block बनाएँ.

जब Workflow परिभाषाओं के लिए सिंटैक्स था रेखांकित, एक महत्वपूर्ण पहलू शामिल नहीं था: Workflow परिभाषा के भीतर ही सीधे blocks परिभाषित करने की क्षमता। इस अनुभाग में in-place परिभाषित blocks के लिए manifest और Python code शामिल हो सकता है, जिन्हें Execution Engine द्वारा गतिशील रूप से व्याख्यायित किया जाता है। ये in-place blocks उन्हीं की तरह काम करते हैं जो स्थिर रूप से परिभाषित होते हैं प्लगइन्स लेकिन ये कहीं अधिक लचीलापन प्रदान करते हैं।

निष्पादन मोड

Dynamic Python blocks दो निष्पादन मोड का समर्थन करते हैं:

स्थानीय निष्पादन

जब आप अपने स्वयं के हार्डवेयर पर स्थानीय रूप से inference चलाते हैं, तो dynamic blocks सीधे आपके environment में execute होते हैं। यह विकास और परीक्षण के लिए सबसे तेज़ प्रदर्शन प्रदान करता है।

Dynamic blocks का स्थानीय निष्पादन केवल आपके स्थानीय deployment में काम करता है inference और अविश्वसनीय कोड चलाते समय सुरक्षा प्रभावों पर सावधानीपूर्वक विचार करने की आवश्यकता होती है।

यदि आप इस functionality को अक्षम करना चाहते हैं, export ALLOW_CUSTOM_PYTHON_EXECUTION_IN_WORKFLOWS=False

क्लाउड निष्पादन (Roboflow Serverless v2)

Roboflow की क्लाउड infrastructure को Serverless v2 API के साथ उपयोग करते समय, dynamic blocks सुरक्षित, पृथक containers में execute होते हैं। इससे आपके infrastructure से समझौता किए बिना custom code का सुरक्षित निष्पादन सुनिश्चित होता है।

डेटा सीरियलाइज़ेशन आवश्यकताएँ

क्लाउड निष्पादन का उपयोग करते समय, सभी input और output डेटा Inference के serialization system के माध्यम से serializable होने चाहिए। इसका अर्थ है:

• सरल Python प्रकारों (str, int, float, bool, list, dict) का उपयोग करें

• Numpy arrays और standard computer vision डेटा संरचनाएँ समर्थित हैं

• जटिल custom objects को सरल representations में बदलने की आवश्यकता हो सकती है

• functions, lambda expressions, या अन्य non-serializable Python objects लौटाने से बचें

क्लाउड निष्पादन वातावरण स्थानीय निष्पादन के समान standard libraries और imports प्रदान करता है, जिससे आपका code दोनों मोड में समान रूप से काम करता है।

स्थिति प्रबंधन और साझा डेटा

module level पर परिभाषित variables (आपके ... के बाहर) run function) आपके block के code में उस block के instances तक सीमित (scoped) होते हैं। ये variables:

• invocations के बीच बने रहते हैं उसी block के (जब तक code नहीं बदलता)

• block का code बदलने पर reset हो जाते हैं block के code में कोई भी संशोधन एक नया namespace बनाता है

• server/container के restart होने पर खो जाते हैं

उदाहरण:

यह block हर बार चलने पर एक counter बढ़ाता है और अंतिम परिणाम को याद रखता है:

# यह variable ब्लॉक-स्कोप्ड है
counter = 0
last_result = None

def run(self, input_value):
    global counter, last_result
    
    counter += 1
    
    # तुलना के लिए अंतिम परिणाम संग्रहीत करें
    previous = last_result
    last_result = input_value * 2
    
    return {
        "run_count": counter,
        "current": last_result,
        "previous": previous
    }

स्थिति प्रबंधन के सर्वोत्तम अभ्यास

Custom Block state महंगी गणनाओं को कैश करने और artifact तथा dependency loading को अनुकूलित करने के लिए है।

• महत्वपूर्ण डेटा स्थायित्व के लिए state पर निर्भर न रहें - महत्वपूर्ण डेटा के लिए बाहरी storage का उपयोग करें।

• server restarts या container scaling के कारण state किसी भी समय खो सकती है

• क्लाउड वातावरण में, बाद के अनुरोध अलग-अलग state वाले अलग servers पर जा सकते हैं

• नए शुरूआतों को संभालने के लिए block-scoped variables को default values के साथ initialize करें

• state को हल्का रखें। बड़े objects memory लेते हैं और performance को प्रभावित कर सकते हैं।

सिद्धांत

Dynamic Python blocks की कार्यक्षमता का उच्च-स्तरीय अवलोकन:

• उपयोगकर्ता JSON में dynamic block की परिभाषा प्रदान करता है

• परिभाषा में Execution Engine को बनाने के लिए आवश्यक जानकारी होती है WorkflowBlockManifest और WorkflowBlock दस्तावेज़ से

• रनटाइम पर, Compiler परिभाषा को गतिशील रूप से बनाई गई Python classes में बदल देता है - बिल्कुल उसी तरह जैसे statically defined blocks

• Workflow परिभाषा में, आप ऐसे steps घोषित कर सकते हैं जो dynamic blocks का उपयोग करते हैं, मानो dynamic blocks सामान्य स्थिर blocks हों

उदाहरण

आइए देखें और dynamic Python blocks वाले example workflow पर चर्चा करें।

dynamic block वाला Workflow

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

आइए विश्लेषण की शुरुआत करें dynamic_blocks_definitions - यह Workflow Definition का वह भाग है जो dynamic blocks की सूची प्रदान करता है। प्रत्येक block में दो अनुभाग होते हैं:

• manifest - की JSON प्रतिनिधि प्रदान करता है BlockManifest - संदर्भ लें blocks विकास मार्गदर्शिका

• code - Python code शामिल करता है

block manifest की परिभाषा

Manifest परिभाषा में कई फ़ील्ड शामिल होते हैं, जिनमें:

• block_type - के समतुल्य type block manifest में फ़ील्ड - अद्वितीय block identifier प्रदान करना आवश्यक है

• inputs - dynamic inputs के नाम और परिभाषाओं वाली dictionary

• outputs - dynamic outputs के नाम और परिभाषाओं वाली dictionary

• output_dimensionality_offset - फ़ील्ड output dimensionality निर्दिष्ट करता है

• accepts_batch_input - फ़ील्ड निर्धारित करता है कि runtime में input data Execution Engine द्वारा batches में प्रदान किया जाएगा या नहीं

• accepts_empty_values - फ़ील्ड यह तय करता है कि step inputs बनाते समय खाली inputs को अनदेखा किया जाएगा या नहीं

किसी भी संदेह की स्थिति में, देखें blocks विकास मार्गदर्शिका, क्योंकि dynamic blocks standard blocs की क्षमताओं को दोहराते हैं।

dynamic input की परिभाषा

Dynamic inputs dynamically बनाए गए block manifest के फ़ील्ड परिभाषित करते हैं। दूसरे शब्दों में, यह वह परिभाषा है जिसके आधार पर BlockManifest क्लास runtime में बनाई जाएगी।

प्रत्येक input निम्नलिखित properties परिभाषित कर सकता है:

• has_default_value - यह तय करने का flag कि dynamic manifest फ़ील्ड का default है या नहीं

• default_value - default value (केवल तब उपयोग होती है यदि has_default_value=True

• is_optional - यह तय करने का flag कि dynamic manifest फ़ील्ड वैकल्पिक है या नहीं

• is_dimensionality_reference - यह तय करने का flag कि dynamic manifest फ़ील्ड selector को runtime में dimensionality reference के रूप में उपयोग किया जाएगा या नहीं

• dimensionality_offset - dynamic manifest की configured input property के लिए dimensionality offset

• selector_types - selectors का प्रकार जो property द्वारा उपयोग किया जा सकता है (इनमें से एक input_image, step_output_image, input_parameter, step_output). Step के पास selector नहीं भी हो सकता, लेकिन तब उसे विशिष्ट प्रकार की परिभाषा प्रदान करनी होगी।

• selector_data_kind - प्रत्येक selector type के लिए विशिष्ट selector kinds की सूची वाली dictionary

• value_types - उस विशिष्ट प्रकार की परिभाषा जिसे manifest में रखा जाना है - यह फ़ील्ड dynamically बनाए गए manifest फ़ील्ड्स की Python types के संदर्भ में typing निर्दिष्ट करता है। प्रकारों का चयन: कोई भी, पूर्णांक, float, boolean, dict, list, strig

dynamic output की परिभाषा

outputs की परिभाषाएँ काफी सरल हैं, इनमें वैकल्पिक सूची होती है kinds जो दिए गए output के लिए घोषित होती है।

Python code की परिभाषा

Python code निम्नलिखित फ़ील्ड्स के साथ JSON दस्तावेज़ में शामिल किया जाता है:

• run_function_code - का code run(...) आपके dynamic block की method

• run_function_name - run function का नाम

• init_function_code - आपकी init function के लिए वैकल्पिक code जो step state को assemble करेगी - इससे dictionary लौटने की अपेक्षा है, जो उपलब्ध होगी run() function के अंतर्गत self._init_results

• init_function_name - init function का नाम

• imports - अतिरिक्त imports की सूची (आप केवल अपने environment की libraries का उपयोग कर सकते हैं, कोई dependencies स्वतः स्थापित नहीं की जाएँगी)

कैसे बनाएँ run(...) method?

आपको निम्नलिखित जानना चाहिए:

• run(...) function को इस प्रकार परिभाषित करना होगा, मानो वह class instance method हो - जिसमें पहला argument self और शेष arguments dynamic block की परिभाषा में घोषित dynamic block manifest के साथ संगत हों

• आपको यह अपेक्षा करनी चाहिए कि baseline symbols प्रदान किए जाएँगे, जिनमें आपके import statements और निम्नलिखित शामिल हैं:

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

तो उदाहरण function कुछ इस प्रकार दिख सकता है (स्पष्टता के लिए, यहाँ हम Python code को अच्छी तरह फ़ॉर्मेट करके प्रस्तुत कर रहे हैं, लेकिन आपको इसे definition में रखने के लिए stringify करना होगा):

def run(self, predictions: sv.Detections, class_x: str, class_y: str) -> BlockResult:
    bboxes_class_x = predictions[predictions.data["class_name"] == class_x]
    bboxes_class_y = predictions[predictions.data["class_name"] == class_y]
    overlap = []
    for bbox_x in bboxes_class_x:
        bbox_x_coords = bbox_x[0]
        bbox_overlaps = []
        for bbox_y in bboxes_class_y:
            if bbox_y[-1]["detection_id"] == bbox_x[-1]["detection_id"]:
                continue
            bbox_y_coords = bbox_y[0]
            x_min = max(bbox_x_coords[0], bbox_y_coords[0])
            y_min = max(bbox_x_coords[1], bbox_y_coords[1])
            x_max = min(bbox_x_coords[2], bbox_y_coords[2])
            y_max = min(bbox_x_coords[3], bbox_y_coords[3])
            # प्रतिच्छेदन आयत का क्षेत्रफल निकालें
            intersection_area = max(0, x_max - x_min + 1) * max(0, y_max - y_min + 1)
            box_x_area = (bbox_x_coords[2] - bbox_x_coords[0] + 1) * (bbox_x_coords[3] - bbox_x_coords[1] + 1)
            local_overlap = intersection_area / (box_x_area + 1e-5)
            bbox_overlaps.append(local_overlap)
        overlap.append(bbox_overlaps)
    return  {"overlap": overlap}

कैसे बनाएँ init(...) method?

Init function को बनाना अपेक्षित है self._init_results dictionary.

उदाहरण:

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

Dynamic Python block का step के रूप में उपयोग

जैसा कि example Workflow definition में दिखाया गया है, आप block का उपयोग बस ऐसे कर सकते हैं मानो वह static plugin के माध्यम से उपलब्ध सामान्य block हो:

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

dynamic blocks का debugging

सेट करें debug=True पर /workflows/run custom Python blocks से diagnostics capture करने के लिए request। यह opt-in है (preview runs इसे स्वतः सक्षम नहीं करते)।

stdout/stderr को capture करना। आपके block द्वारा print की गई हर चीज़ प्रति step capture की जाती है।

structured traces emit करना। A debug_traces helper आपके run(...) code (मानक imports के साथ एक baseline symbol के रूप में injected)। कोई भी JSON-serialisable मान जोड़ें; pass करें add_timestamp=True एंट्री पर timestamp लगाने के लिए:

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

जब debug सक्षम नहीं है (या Modal / OCI sandbox execution के अंतर्गत), debug_traces.append(...) यह एक सुरक्षित no-op है और एकत्रित नहीं किया जाता।

सफल रन (HTTP 200)। प्रतिक्रिया में निम्न शामिल होते हैं:

• python_blocks_output_streams - step नाम के अनुसार keyed किया गया captured stdout/stderr, उदाहरणार्थ {"my_step": [{"stdout": "...", "stderr": null}]}.

• python_blocks_debug_traces - निष्पादन क्रम में जोड़ी गई प्रविष्टियाँ, उदाहरणार्थ [{"step": "my_step", "value": {"received": 7}, "timestamp": "...", "timestamp_timezone": "UTC"}] (timestamp* केवल तब मौजूद होता है जब add_timestamp=True).

दोनों null जब debug बंद है या कुछ भी capture नहीं किया गया। केवल स्थानीय निष्पादन के लिए भरा जाता है।

असफल रन (HTTP 400)। त्रुटि प्रतिक्रिया में वही दो फ़ील्ड विफलता से पहले चलने वाले steps द्वारा उत्पन्न आंशिक output/traces के साथ, और blocks_errors[].block_traceback विफल होने वाले step के अपने stdout/stderr सहित।