PatchCore × ROS 2 で食品ライン検査を実装する3つの落とし穴 — anomalib少サンプル異常検知

本記事はPatchCoreをROS 2に組み込む実装ガイドで、HACCP法令の解説ではありません。記録様式の妥当性は所管監督者・第三者認証機関の判断によります。技術側として『何をどう残せば監査で使い物になるか』を扱います。

なぜ PatchCore × ROS 2 が中小食品工場のスイートスポットか

PatchCoreは2021年に提案され（CVPR 2022で正式発表された）異常検知手法で、ImageNet事前学習済みモデル（WideResNet50など）の中間特徴をパッチ単位で取り出し、良品画像から構築したmemory bankとのk近傍距離をスコアにします。良品20〜50枚で動き始めるのが現場で効きます。教師あり学習のように不良画像をラベリングする工程が要らない点が、SKUが頻繁に増える中小ラインで決定的に有利です。

中堅食品工場（年商10〜100億円）の検査ライン更新では、CognexのVisionPro Deep Learningのようなフルスタック構成（専用カメラ・専用照明・GUIツール・サポート込み）で1ラインあたり¥500万〜数千万、というのが第一候補の見積感です。一方KeyenceのIV3シリーズは内蔵AIビジョンセンサで本体数十万円〜のエントリ寄りであり、価格帯としては別カテゴリに属します。本記事はフルスタック構成の代替を狙うため、以後の比較は前者の価格帯を念頭に置きます。anomalib単体運用は社内にMLエンジニアがいるなら強力ですが、ライン制御や画像配信といった『工場側のI/O』はスコープ外です。

そこをROS 2 humbleで埋めます。sensor_msgs/Image をトピックで流し、トリガ駆動の推論ノードを書ける環境が標準で揃っており、汎用USBカメラと単体PCで動かしつつPLC/サーボへの拡張余地も残せます。Cognex / Keyenceとは別軸で『安く・早く・現場に合わせて変えられる』ポジションを取れるのが選定理由です。

最小実装: anomalib + rclpy で異常検知トピックに変換する

anomalib v1系のPatchCoreモデルとROS 2 humbleのrclpyを組み合わせる最小構成は次のようになります。USBカメラは usb_cam ノードで /image_raw にpublishされている前提です。

# anomaly_detector_node.py（抜粋）
from anomalib.deploy import TorchInferencer
from cv_bridge import CvBridge
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image

class AnomalyDetector(Node):
    def __init__(self):
        super().__init__("anomaly_detector")
        self.bridge = CvBridge()
        self.inferencer = TorchInferencer(path="patchcore.ckpt", device="cuda")
        self.sub = self.create_subscription(Image, "/image_raw", self.cb, 10)
        self.pub = self.create_publisher(AnomalyDetection, "/anomaly", 10)

    def cb(self, msg):
        img = self.bridge.imgmsg_to_cv2(msg, "bgr8")
        result = self.inferencer.predict(image=img)
        self.pub.publish(self._to_msg(result, msg.header))

自前定義する AnomalyDetection.msg には最低限 float32 score、bool is_anomaly、string heatmap_b64、string inference_id を持たせます。inference_id はUUIDv4で、後述する月次レポートとの突合キーになります。ここまでで半日。本番ラインでは次の3つに必ず当たります。

落とし穴1 — memory bank の肥大化と coreset / PCA 設計

PatchCoreのmemory bankは『学習画像数 × パッチ数 × 特徴次元』のテンソルとしてディスクとVRAMに居座ります。WideResNet50のlayer2+layer3を使う既定設定で、入力224×224・良品100枚なら数百MB、解像度を512×512に上げると1〜2GBに膨らみます。SKU別にモデルを持つ運用ではVRAMが先に枯渇します。

anomalib側の対策として coreset_sampling_ratio（既定0.1）が用意されており、k-Center Greedyでmemory bankを間引きます。0.1のままで精度劣化が許容範囲かはSKUごとに検証が必要で、包装シール不良のような微小欠陥では0.25〜0.5まで戻すほうが安定します。さらに num_neighbors を9から3に下げるとレイテンシは縮みますが、テクスチャ性の異常で見逃しが増えます。

VRAM側のもう一段の打ち手はfloat16化です。memory bankを .half() に落とすだけでメモリは半減し、k-NN距離の数値誤差は包装検査の閾値感度では影響しないことがほとんどです。PCAでの次元削減（例: 1024→256）は精度低下が読みづらいので、量子化を先に試して足りない場合のみ検討します。SKUが10種を超えるなら『1モデル多SKU』ではなく『SKUごとに別チェックポイント、ROS 2パラメータでロード切替』のほうが運用が簡単で、現場での切戻しも素直です。

落とし穴2 — 推論レイテンシ予算とノード並列化

ライン速度100〜300fpsという数字が一人歩きしがちですが、検査対象の通過は1個あたり数百msのウィンドウがあり、全フレーム推論は不要なケースがほとんどです。usb_cam 側を30fpsに固定し、トリガセンサ（光電・エンコーダ）からの std_msgs/Header で間引くのが現実解です。

それでも1個あたりの推論は、memory bankサイズ・coreset比率・近傍検索実装で大きく変動します。本稿では弊社環境（RTX 3060 / 入力256×256 / coreset 0.25 / FAISS未使用）で200〜400msという数字を前提に置きますが、coreset 0.1 + FAISSなら数十ms、解像度を512に上げれば1秒近くまで伸びます。条件次第で1桁変わる前提で読んでください。rclpy.executors.MultiThreadedExecutor で複数の AnomalyDetector を立て、トリガIDをラウンドロビンで割り振ります。注意点として、TorchInferencer インスタンスはスレッドセーフではなくノードごとにインスタンスを持たせる必要があります。GPU側の競合は torch.cuda.Stream で隔離するか、ONNX RuntimeへエクスポートしてCPUインスタンスを複数立てるほうがデバッグしやすいです。

OpenVINO IRへの変換（anomalibが標準サポート）も選択肢で、Intel iGPU / NPU上ではGPU不要・電力20W前後で動きます。装置PCにNVIDIA GPUを増設できない案件ではこちらが第一候補です。バッチ推論はROS 2のトリガ駆動と相性が悪く（バッチ待ちでウィンドウを外す）、本番ではほぼ採用しません。スループットよりテイルレイテンシを守る設計です。

落とし穴3 — 誤検知のトレース可能性

異常スコアだけpublishする実装は運用1ヶ月で破綻します。現場担当者から『なぜこのパッケージをNGにしたのか』と聞かれて答えられないためです。月次品質レポートを成果物にするなら、最低限以下を1検知ごとに残します。

inference_id（UUIDv4）
入力画像（PNG、可逆圧縮）
ヒートマップ（anomalibの anomaly_map、PNG連番保存）
閾値・モデルバージョン・SKUコード
トリガ時刻（msg.header.stamp をepochに変換）

実装としては推論ノードの裏に bag_logger_node を別建てし、メインノードはpublishだけに集中させます。ここを同居させるとI/O待ちで推論レイテンシが膨れます。月次レポートはCSVに inference_id を主キーで並べ、PDF生成時にヒートマップ画像を埋め込めば、後から検査員と一緒に追跡できます。

社内にレビュー用のWeb UI（FastAPI + 画像配信）まで作るかは案件次第ですが、最低でも『inference_id から画像パスを引き当てる』が1コマンドでできる状態は必須です。誤検知率（false positive per shift）の計測が回らないためです。F1やAUROCは現場の合意形成には使えず、『1シフト8時間あたり何件の誤NGが出るか』が実質的な受容指標です。ここを最初に決めないと、モデル改善のサイクルが空転します。

落とし穴の延長: トレース構造をHACCPに接続する

HACCP記録代行を成果物に含める場合、CCP（重要管理点）ごとのモニタリング項目に対応させる必要があります。包装シール不良であればCCP『シール不良の検出』に対し、モニタリング頻度・許容基準・是正措置の記録欄が雛形にあるため、inference_id 付きの異常検出ログをそのまま証跡として添付できる構造が現実的です。

データ設計としては、月次CSVに ccp_id、detected_at、is_anomaly、reviewed_by、action_taken を持たせ、検査員レビュー後の是正措置までカラムを通しておくと、監査対応の追加工数がほぼゼロになります。AI検出ログ単体では監査証跡として弱く、人による確認カラムが入って初めて運用に乗ります。トレース構造さえ最初に決まれば、PatchCoreの精度改善は後から差し替えられます。

PoC → 本番移行の判断チェックリスト

落とし穴の対処を一通り入れた後、本番ラインへ移行できるかは次の4点が揃った時点で判断します。

[ ] 誤NG/シフトの閾値合意: 1シフト8時間あたりの許容誤NG件数を現場と合意し、PatchCoreの判定閾値をその数字に合わせて調整できているか
[ ] coreset比率のSKU別検証: 主要SKUごとに coreset_sampling_ratio を 0.1 / 0.25 / 0.5 で振り、見逃し件数と推論レイテンシの両方を測ったか
[ ] bag_logger分離: 推論ノードとログ書き込みノードを分け、ログI/Oで推論レイテンシが伸びていないことを ros2 topic hz 等で確認したか
[ ] HACCP CCP対応: 対象工程のCCP（重要管理点）と検出項目を1対1で対応させ、ccp_id 付きのCSVが既存の月次品質報告フォーマットに突合できる状態か

この4点が揃わないまま本番化すると、運用1〜2ヶ月で『誤NGが多すぎてラインが止まる』『ログがあるのに監査で証跡として認められない』のいずれかで詰まります。

付録: 動作確認に使った最小構成

`AnomalyDetection.msg`

カスタムメッセージは1ファイルに収めます。msg/AnomalyDetection.msg:

std_msgs/Header header
string  inference_id     # UUIDv4
float32 score            # PatchCore raw distance
float32 threshold        # 採用した閾値
bool    is_anomaly       # score > threshold
string  heatmap_b64      # PNG base64、デバッグ用（本番ではbag_loggerに退避）
string  sku_code         # ロードしたモデルのSKU
string  model_version    # チェックポイントのハッシュ短縮形

heatmap_b64 を持たせるかは案件次第で、ペイロード削減のためURIだけ流して画像は別ノードで保存する構成も取れます。

`inspection_bringup.launch.py`

usb_cam、推論ノード、bag_loggerを束ねる最小 launch ファイルです。

from launch import LaunchDescription
from launch_ros.actions import Node

def generate_launch_description():
    return LaunchDescription([
        Node(
            package="usb_cam", executable="usb_cam_node_exe", name="usb_cam",
            parameters=[{"video_device": "/dev/video0", "framerate": 30.0,
                          "image_width": 256, "image_height": 256}],
        ),
        Node(
            package="anomaly_detector", executable="anomaly_detector_node",
            name="anomaly_detector",
            parameters=[{"checkpoint": "/opt/models/sku_a.ckpt",
                          "threshold": 0.55,
                          "sku_code": "SKU_A",
                          "device": "cuda"}],
            remappings=[("/image_raw", "/usb_cam/image_raw")],
        ),
        Node(
            package="bag_logger", executable="bag_logger_node",
            name="bag_logger",
            parameters=[{"output_dir": "/var/log/inspection",
                          "save_heatmap_png": True}],
        ),
    ])

ros2 launch anomaly_detector inspection_bringup.launch.py 一発でカメラ→推論→ログまで通り、SKU切替は parameters の差し替えだけで対応できます。

本番運用の堅牢化（モデル更新・カメラ再接続・ヘルスチェック）は別記事で扱います。少サンプルで動かし、誤検知を運用で潰し、月次レポートを回し始めるところまでが射程です。

異常検知の受託・PoCを検討中の方は、食品工場検査AI受託サービスもあわせてご覧ください。

PatchCore × ROS 2 で食品ライン検査を実装する3つの落とし穴 — anomalib少サンプル異常検知

PatchCore × ROS 2 で食品ライン検査を実装する3つの落とし穴 — anomalib少サンプル異常検知

なぜ PatchCore × ROS 2 が中小食品工場のスイートスポットか

最小実装: anomalib + rclpy で異常検知トピックに変換する

落とし穴1 — memory bank の肥大化と coreset / PCA 設計

落とし穴2 — 推論レイテンシ予算とノード並列化

落とし穴3 — 誤検知のトレース可能性

落とし穴の延長: トレース構造をHACCPに接続する

PoC → 本番移行の判断チェックリスト

付録: 動作確認に使った最小構成

AnomalyDetection.msg

inspection_bringup.launch.py

この課題、yoritechが解決します

`AnomalyDetection.msg`

`inspection_bringup.launch.py`