> For the complete documentation index, see [llms.txt](https://docs.roboflow.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.roboflow.com/workflows/ja/gaido/developer-guide/dynamic-python-blocks.md). # 動的Pythonブロック {% hint style="info" %} この機能の製品レベルの紹介については、こちらを参照してください [カスタムブロック](/workflows/ja/burokku/blocks/custom-blocks.md)。インラインコードではなく完全な Block クラスを記述するには、こちらを参照してください [Workflowブロックを作成する](/workflows/ja/gaido/developer-guide/create-a-block.md). {% endhint %} Workflow 定義の構文が [示されたとき](/workflows/ja/gaido/developer-guide/definitions.md)、1つ重要な点がまだ扱われていませんでした。それは、Workflow 定義そのものの中で直接 block を定義できる機能です。このセクションには、その場で定義された block の manifest と Python コードを含めることができ、Execution Engine によって動的に解釈されます。これらのインプレース block は、 [プラグイン](/workflows/ja/gaido/developer-guide/block-bundling.md) で静的に定義されたものと同様に機能しますが、はるかに柔軟です。 ## 実行モード Dynamic Python block は2つの実行モードをサポートします： ### ローカル実行 自分のハードウェア上でローカルに推論を実行する場合、動的 block はあなたの環境で直接実行されます。これにより、開発とテストに最速のパフォーマンスが得られます。 {% hint style="warning" %} 動的 block のローカル実行は、 `inference` のローカルデプロイメントでのみ動作し、信頼できないコードを実行する際はセキュリティ上の影響を慎重に考慮する必要があります。 この機能を無効にしたい場合は、 `export ALLOW_CUSTOM_PYTHON_EXECUTION_IN_WORKFLOWS=False` {% endhint %} ### クラウド実行（Roboflow Serverless v2） Roboflow のクラウドインフラストラクチャを Serverless v2 API とともに使用する場合、動的 block は安全で隔離されたコンテナ内で実行されます。これにより、インフラを損なうことなくカスタムコードを安全に実行できます。 {% hint style="warning" %} **データのシリアライズ要件** クラウド実行を使用する場合、すべての入力データと出力データは Inference のシリアライズシステムを通じてシリアライズ可能でなければなりません。つまり： * 単純な Python 型を使用する（str, int, float, bool, list, dict） * Numpy 配列および標準的なコンピュータビジョンのデータ構造がサポートされます * 複雑なカスタムオブジェクトは、より単純な表現に変換する必要がある場合があります * 関数、lambda 式、その他シリアライズ不可能な Python オブジェクトを返さないでください {% endhint %} クラウド実行環境は、ローカル実行と同じ標準ライブラリおよびインポートを提供するため、両方のモードでコードが一貫して動作します。 ## 状態管理と共有データ モジュールレベルで定義された変数（あなたの `run` 関数の外側）は、その block のインスタンスにスコープされます。これらの変数は： * **呼び出しをまたいで保持される** 同じ block の（コードが変更されない限り） * **block のコードが変更されるとリセットされる** block のコードに少しでも変更があると新しい名前空間が作成される * **サーバー/コンテナが再起動すると失われる** **例:** この block は、実行されるたびにカウンターを増やし、最後の結果を記憶します： ```python # この変数は block スコープです 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 の状態は、コストの高い計算のキャッシュや、アーティファクトおよび依存関係の読み込み最適化を目的としています。 * 重要なデータの永続化を状態に頼らないでください。重要なデータには外部ストレージを使用してください。 * サーバーの再起動やコンテナのスケーリングにより、状態はいつでも失われる可能性があります * クラウド環境では、後続のリクエストが異なる状態を持つ別のサーバーに送られる場合があります * 初回起動に対応するため、block スコープの変数はデフォルト値で初期化してください * 状態は軽量に保ってください。大きなオブジェクトはメモリを消費し、性能に影響する可能性があります。 ## 理論 Dynamic Python block の機能の高レベル概要： * ユーザーは JSON で dynamic block の定義を提供する * 定義には、Execution Engine が構築するために必要な情報が含まれている `WorkflowBlockManifest` と `WorkflowBlock` を文書から * 実行時に、Compiler はその定義を動的に生成された Python クラスへ変換します。これは静的に定義された block とまったく同じです * Workflow 定義では、dynamic block を標準の静的 block であるかのように使用する step を宣言できます ## 例 Dynamic Python block を使った例の Workflow を見て、説明していきましょう。
dynamic block を含む Workflow ```json { "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` - これは dynamic block の一覧を提供する Workflow 定義の部分です。各 block には2つのセクションがあります： * `manifest` - の JSON 表現を提供する `BlockManifest` - 参照 [block の開発ガイド](/workflows/ja/gaido/developer-guide/create-a-block.md) * `code` - Python コードを含む ### block マニフェストの定義 マニフェスト定義には、次のようないくつかのフィールドが含まれます： * `block_type` - に相当する `type` block マニフェスト内の field - 一意の block 識別子を指定する必要があります * `inputs` - dynamic input の名前と定義を持つ辞書 * `outputs` - dynamic output の名前と定義を持つ辞書 * `output_dimensionality_offset` - この field は出力の次元性を指定します * `accepts_batch_input` - この field は、実行時の入力データを Execution Engine がバッチで提供するかどうかを決定します * `accepts_empty_values` - 空の入力を step 入力の構築時に無視するかどうかを決める field 不明な点がある場合は、 [block の開発ガイド](/workflows/ja/gaido/developer-guide/create-a-block.md)を参照してください。dynamic block は標準の block の機能を再現します。 ### dynamic input の定義 Dynamic input は、動的に作成される block マニフェストの field を定義します。言い換えると、これは `BlockManifest` class が実行時に作成される基礎となる定義です。 各 input では次の properties を定義できます： * `has_default_value` - dynamic manifest field にデフォルト値があるかどうかを決めるフラグ * `default_value` - デフォルト値（以下の場合のみ使用） `has_default_value=True` * `is_optional` - dynamic manifest field が optional かどうかを決めるフラグ * `is_dimensionality_reference` - dynamic manifest field の ship selector を、実行時に次元参照として使うかどうかを決めるフラグ * `dimensionality_offset` - dynamic manifest の設定済み input property に対する次元オフセット * `selector_types` - property に使用できる selector の型（次のいずれか： `input_image`, `step_output_image`, `input_parameter`, `step_output`）。step は selector を持たない場合がありますが、その場合は特定の type の定義を提供しなければなりません。 * `selector_data_kind` - 各 selector type ごとの selector kind の一覧を持つ辞書 * `value_types` - マニフェストに配置される特定の type の定義です。この field は、動的に作成されるマニフェスト field の型を Python 型に対して指定します。選択可能な型： `任意`, `整数`, `float`, `boolean`, `dict`, `list`, `strig` ### dynamic output の定義 output の定義はかなり単純で、指定された output に対して宣言された任意のリストを持ちます。 `kinds` を持ちます。 ### Python コードの定義 Python コードは、次の field を持つ JSON ドキュメントとして送られます： * `run_function_code` - のコード `run(...)` dynamic block の method * `run_function_name` - run 関数の名前 * `init_function_code` - step の状態を組み立てる init 関数のための任意コードです。辞書を返すことが期待され、その辞書は `run()` 関数で `self._init_results` * `init_function_name` - init 関数の名前 * `imports` - 追加インポートの一覧（使用できるのはあなたの環境にあるライブラリのみで、依存関係は自動ではインストールされません） ### どのように `run(...)` method を作成するか？ 次のことを知っておく必要があります： * `run(...)` 関数は、class のインスタンスメソッドであるかのように定義しなければなりません。第1引数は `self` で、残りの引数は dynamic block の定義で宣言された dynamic block マニフェストと互換である必要があります * インポート文と次のものを含む、ベースラインシンボルが提供されることを想定してください： ```python from typing import Any, List, Dict, Set, Optional import supervision as sv import numpy as np import math import time import json import os import requests import cv2 import shapely from inference.core.workflows.execution_engine.entities.base import Batch, WorkflowImageData from inference.core.workflows.prototypes.block import BlockResult ``` では、説明のために Python コードを見やすく整形して示しますが、定義に入れる際にはコードを文字列化する必要があります： ```python def run(self, predictions: sv.Detections, class_x: str, class_y: str) -> BlockResult: bboxes_class_x = predictions[predictions.data["class_name"] == class_x] bboxes_class_y = predictions[predictions.data["class_name"] == class_y] overlap = [] for bbox_x in bboxes_class_x: bbox_x_coords = bbox_x[0] bbox_overlaps = [] for bbox_y in bboxes_class_y: if bbox_y[-1]["detection_id"] == bbox_x[-1]["detection_id"]: continue bbox_y_coords = bbox_y[0] x_min = max(bbox_x_coords[0], bbox_y_coords[0]) y_min = max(bbox_x_coords[1], bbox_y_coords[1]) x_max = min(bbox_x_coords[2], bbox_y_coords[2]) y_max = min(bbox_x_coords[3], bbox_y_coords[3]) # 交差矩形の面積を計算する intersection_area = max(0, x_max - x_min + 1) * max(0, y_max - y_min + 1) box_x_area = (bbox_x_coords[2] - bbox_x_coords[0] + 1) * (bbox_x_coords[3] - bbox_x_coords[1] + 1) local_overlap = intersection_area / (box_x_area + 1e-5) bbox_overlaps.append(local_overlap) overlap.append(bbox_overlaps) return {"overlap": overlap} ``` ### どのように `init(...)` method を作成するか？ init 関数は `self._init_results` 辞書を構築することになっています。 例: ```python def my_init() -> Dict[str, Any]: return {"some": "value"} ``` ### step としての Dynamic Python block の使用 例の Workflow 定義で示したように、その block を、静的プラグインから公開された通常の block であるかのように、そのまま使えます： ```json { "type": "OverlapMeasurement", "name": "overlap_measurement", "predictions": "$steps.model.predictions", "class_x": "dog", "class_y": "dog" } ``` ## dynamic block のデバッグ Set `debug=True` リクエストで `/workflows/run` カスタム Python block から診断情報を取得します。これは任意設定です（プレビュー実行では暗黙には有効になりません）。 **stdout/stderr の取得。** ブロックが出力した内容は、step ごとに取得されます。 **構造化トレースの出力。** A `debug_traces` ヘルパーはあなたの `run(...)` コード内で利用できます（標準インポートとともにベースラインシンボルとして注入されます）。JSON シリアライズ可能な値を追加してください。 `add_timestamp=True` を指定すると、エントリにタイムスタンプを付与できます： ```python def run(self, value: int) -> BlockResult: debug_traces.append({"received": value}) debug_traces.append(value * 2, add_timestamp=True) return {"result": value * 2} ``` 次の場合、 `debug` が有効になっていない場合（または Modal / OCI サンドボックス実行中）、 `debug_traces.append(...)` は安全な no-op であり、収集されません。 **成功した実行（HTTP 200）。** レスポンスには次が含まれます： * `python_blocks_output_streams` - ステップ名をキーとした取得済み stdout/stderr。例： `{"my_step": [{"stdout": "...", "stderr": null}]}`. * `python_blocks_debug_traces` - 実行順に追加されたエントリ。例： `[{"step": "my_step", "value": {"received": 7}, "timestamp": "...", "timestamp_timezone": "UTC"}]` (`タイムスタンプ*` の場合にのみ存在 `add_timestamp=True`). どちらも `null` のときに `debug` debug がオフ、または何も取得されなかった場合は含まれません。ローカル実行でのみ設定されます。 **失敗した実行（HTTP 400）。** エラーレスポンスには **同じ2つの field が** 失敗の前に実行され、出力/追記された step が生成した部分的な出力/トレースとともに、 `blocks_errors[].block_traceback` および失敗した step 自身の stdout/stderr が含まれます。