Tracer カスタマイズ

上級者向け

このページは上級者向けです。ClassificationやObject Detectionタスクでは通常、ClassificationTracer / ObjectDetectionTracer / ObjectDetection3DTracer をそのまま使用すれば十分です。 カスタマイズが必要な場合は、まず support@adansons.ai にご相談ください。

このページでは、特殊なユースケースのためにTracerをカスタマイズする方法を説明します。

Object Detection の標準対応

v0.3.0より、Object Detection 2D/3Dは標準のObjectDetectionTracer / ObjectDetection3DTracerでサポートされています。 詳細は Tracing + Evaluation を参照してください。

カスタムTracerが必要なケース

以下のような場合に、カスタムTracerの定義が必要になることがあります：

• 標準のTracerでサポートされていないモデル構造
• 特殊な入出力形式を持つモデル
• Semantic Segmentationなど、標準Tracerが未対応のカスタムタスク
• 追加のメタデータを収集したい場合

基本構造

カスタムTracerは、タスク固有の基底クラスを継承して定義します。

from ml_debugger.tracer.object_detection.object_detection_torchtracer import (
    ObjectDetectionTorchTracer,
)

class CustomTracer(ObjectDetectionTorchTracer):
    def _parse_and_save_io_data(
        self,
        model_input,
        model_output,
        ground_truth,
        **kwargs,
    ):
        """モデルの入出力をパースして保存するメソッド"""
        # カスタム実装
        pass

_parse_and_save_io_dataメソッド

このメソッドは、モデルの推論結果を解析し、データベースに保存する処理を実装します。

引数

引数	説明
model_input	モデルへの入力テンソル
model_output	モデルの出力
ground_truth	正解ラベル（Tracingモード時）
**kwargs	追加の引数（__call__から渡される）

内部特徴量の取得

Tracer初期化時に指定したtarget_layersの出力を取得できます。

# target_layersの指定例
tracer = CustomTracer(
    model=model,
    model_name="custom_model",
    version_name="v1",
    target_layers={
        "cls_logits": "head.classification_head",
        "bbox_regression": "head.regression_head",
    },
)

# _parse_and_save_io_dataメソッド内で取得
cls_logits = self.get_hooked_features("cls_logits")
bbox_regression = self.get_hooked_features("bbox_regression")

Object Detection用のカスタムTracerの例

以下は、SSDモデル用のカスタムTracerの完全な例です。

from __future__ import annotations
from typing import Any
import torch
from ml_debugger.tracer.object_detection.object_detection_torchtracer import (
    ObjectDetectionTorchTracer,
)


class SSDTracer(ObjectDetectionTorchTracer):
    def _parse_and_save_io_data(
        self,
        model_input: torch.Tensor,
        model_output: list[dict[str, torch.Tensor]],
        ground_truth: list[list[dict[str, torch.Tensor]]],
        filenames: list[str],
        dataset_type: str,
    ) -> None:
        """SSDモデルの入出力をパースして保存"""

        batch_size = model_input.size(0)

        # 内部特徴量の取得
        batch_bbox_regression = self.get_hooked_features("bbox_regression")
        batch_cls_logits = self.get_hooked_features("cls_logits")

        for i in range(batch_size):
            # NMS後のインデックスを取得
            keep_idxs = model_output[i]["keep_index"].numpy()
            img_bbox_regression = batch_bbox_regression[i].numpy()
            img_cls_logits = batch_cls_logits[i].numpy()

            # 予測BBoxごとに保存
            for j, box in enumerate(model_output[i]["boxes"]):
                keep_idx = keep_idxs[j]

                # 内部特徴量をflatten
                bbox_regression = img_bbox_regression[keep_idx].flatten().tolist()
                bbox_regression_shape = list(img_bbox_regression[keep_idx].shape)
                cls_logits = img_cls_logits[keep_idx].flatten().tolist()
                cls_logits_shape = list(img_cls_logits[keep_idx].shape)

                # 予測BBoxの保存
                self._save_extracted_feature(
                    input_id=filenames[i],
                    input_tensor=model_input[i],
                    pred_bbox_id=j,
                    pred_top_left_x=box[0],
                    pred_top_left_y=box[1],
                    pred_bottom_right_x=box[2],
                    pred_bottom_right_y=box[3],
                    pred_class_id=model_output[i]["labels"][j],
                    pred_score=model_output[i]["scores"][j],
                    dataset_type=dataset_type,
                    bbox_regression=bbox_regression,
                    bbox_regression_shape=bbox_regression_shape,
                    cls_logits=cls_logits,
                    cls_logits_shape=cls_logits_shape,
                )

            # 正解BBoxの保存
            for k, gt_info in enumerate(ground_truth[i]):
                self._save_ground_truth(
                    input_id=filenames[i],
                    input_tensor=model_input[i],
                    gt_bbox_id=k,
                    gt_top_left_x=gt_info["bbox"][0],
                    gt_top_left_y=gt_info["bbox"][1],
                    gt_bottom_right_x=gt_info["bbox"][2] + gt_info["bbox"][0],
                    gt_bottom_right_y=gt_info["bbox"][3] + gt_info["bbox"][1],
                    gt_class_id=gt_info["category_id"],
                    dataset_type=dataset_type,
                    src_img_width=gt_info["src_img_width"],
                    src_img_height=gt_info["src_img_height"],
                )

    def __call__(
        self,
        model_input: Any,
        ground_truth: Any,
        filenames: list[str],
        dataset_type: str,
    ) -> Any:
        """カスタム引数を追加した__call__メソッド"""
        return super().__call__(
            model_input,
            ground_truth,
            filenames,
            dataset_type,
        )

保存メソッド

_save_extracted_feature

予測データと内部特徴量を保存します。

self._save_extracted_feature(
    input_id="image_001",           # データの識別子
    input_tensor=model_input[i],    # 入力テンソル（ハッシュ化して保存）
    # タスク固有のカラム
    pred_class_id=predicted_class,
    pred_score=confidence_score,
    # カスタムカラム（内部特徴量）
    custom_feature=feature_vector.tolist(),
    custom_feature_shape=list(feature_vector.shape),
)

_save_ground_truth

正解データを保存します。

self._save_ground_truth(
    input_id="image_001",
    input_tensor=model_input[i],
    gt_class_id=true_class,
    dataset_type="train",
)

target_layersの指定

内部特徴量を抽出する層を指定します。

tracer = CustomTracer(
    model=model,
    model_name="model",
    version_name="v1",
    target_layers={
        # key: エイリアス名（カラム名に使用）
        # value: 層へのパス（model.xxxでアクセス可能な形式）
        "fc_output": "fc",
        "conv_features": "backbone.layer4",
    },
)

additional_fields / additional_label_fields

推論結果やアノテーションに対して、任意の追加情報を記録するためのオプションです。

パラメータ	対象	説明
additional_fields	推論（予測）結果	予測ごとに追加メタデータを記録
additional_label_fields	アノテーション	正解ラベルごとに追加メタデータを記録

初期化時の指定

Tracer初期化時に、追加フィールドをList[dict]形式で指定します。各辞書には以下のキーを設定できます。

キー	必須	説明
name	○	フィールド名
type	-	Python型（str, int, floatなど）
nullable	-	Null許容（デフォルト: True）

from ml_debugger.tracer.classification import ClassificationTorchTracer

tracer = ClassificationTorchTracer(
    model=model,
    model_name="my_model",
    version_name="v1",
    # 推論結果に追加するフィールド
    additional_fields=[
        {"name": "camera_id", "type": str},
        {"name": "lighting_condition", "type": str},
        {"name": "temperature", "type": float, "nullable": True},
    ],
    # アノテーションに追加するフィールド
    additional_label_fields=[
        {"name": "annotator_id", "type": str},
        {"name": "confidence_level", "type": int},
    ],
)

Tracing時の値の指定

__call__メソッドの**kwargsとして、追加フィールドの値をリスト形式で渡します。 リストの長さはバッチサイズと一致させてください。

# バッチ処理（batch_size=4の例）
output = tracer(
    model_input=images,          # shape: (4, C, H, W)
    ground_truth=labels,         # shape: (4,)
    input_ids=input_ids,         # ["img_001", "img_002", "img_003", "img_004"]
    dataset_type="train",
    # additional_fields / additional_label_fieldsの値をリストで渡す
    camera_id=["CAM_001", "CAM_002", "CAM_001", "CAM_003"],
    lighting_condition=["daylight", "night", "daylight", "indoor"],
    temperature=[25.5, 18.0, 26.0, 22.5],
    annotator_id=["annotator_A", "annotator_A", "annotator_B", "annotator_A"],
    confidence_level=[3, 2, 3, 1],
)

動的なフィールド追加

初期化時に定義していないフィールドを__call__時に渡すと、自動的にスキーマが更新されます。 ただし、事前にadditional_fieldsで型を指定しておくことを推奨します。

活用例

追加フィールドを使用すると、評価結果を条件別に分析できます。

# 評価結果の取得
result = evaluator.get_result(result_name="my_result")

# カスタムビューで条件別分析
# 照明条件別のエラー分布
lighting_view = result.get_view(
    groupby=["lighting_condition", "category"],
    adjustby="lighting_condition",
)

# カメラ別のエラー分布
camera_view = result.get_view(
    groupby=["camera_id", "category"],
    adjustby="camera_id",
)

活用シーン

• 撮影条件: カメラID、照明、天候、時間帯
• データソース: データ収集場所、デバイス種別
• アノテーション品質: アノテーター、確信度、レビュー状態
• Object Detection: オブジェクトサイズ、オクルージョン有無

注意事項

1. 内部特徴量のフォーマット

   • 保存時は1次元のリスト形式に変換が必要
   • 復元用にshape情報も保存することを推奨

2. input_tensor

   • データの一意性検証のためにハッシュ化して保存される
   • 同一input_idで異なるinput_tensorがある場合、警告が発生

3. パフォーマンス

   • バッチ処理を活用して効率的に保存
   • 大量のカスタムカラムは処理速度に影響

4. 標準評価メソッドの非対応

   • カスタムTracerを使用した場合、Evaluator.request_evaluation()などの標準評価メソッドが対応していない可能性があります
   • この場合、tracer.export()でデータをエクスポートし、Adansonsにデータを送付してカスタム評価を依頼してください
   • 詳細は support@adansons.ai までお問い合わせください

サポート

カスタムTracerの実装で問題が発生した場合は、support@adansons.ai までお問い合わせください。