コンピュータビジョン + 生成AIで現場の安全監視を自動化：AWSサーバーレス実践ガイド

なぜコンピュータビジョンと生成AIが安全監視に必要なのか

産業現場の安全事故は依然として深刻な問題です。ILOによると、世界全体で年間3億9500万件の業務関連非致死的傷害が発生し、米国だけでも2023年に1765億ドルの経済損失が生じました。OSHAは車両衝突死亡事故が100%予防可能と述べていますが、依然として主要な死因の一つです。

従来の手動安全監査には時間的・空間的限界があります。特定時点のスナップショットしか提供できず、24時間365日の継続的監視は不可能です。数百の施設に拡大すると、この問題はさらに深刻化します。

本記事では、AWSサーバーレスアーキテクチャをベースにしたコンピュータビジョンソリューションを紹介します。固定カメラネットワークを活用してPPE(個人保護具)装着状況と5Sテープベースの危険ゾーンをリアルタイム検出し、生成AIで合成データを作成してモデル学習効率を最大化する方法を扱います。

本記事は、AWS Architecture Blogの原文を実務視点で再構成したものです。

コアアーキテクチャ：サーバーレスDriver-Workerパターン

数千台のカメラをサポートするには、スケーラブルなアーキテクチャが不可欠です。本ソリューションの核は、サーバーレスDriver-Workerパターンです。Driverがタスクを分配し、Workerが画像を並列処理するため、1つのWorkerの障害がパイプライン全体に影響を与えません。

主要コンポーネント

1. 画像収集と匿名化

   • 各カメラから定期的に画像をキャプチャし、専用S3バケットに保存
   • Amazon Rekognitionで顔検出後、Pythonコードでブラー処理
   • 元画像はポリシーに従い数日以内に自動削除

2. 学習パイプライン (SageMaker Ground Truth + Step Functions)

   # 例：Ground Truthラベリングジョブ作成 Step Functionsワークフロー定義（一部）
   import boto3
   import json
   
   client = boto3.client('stepfunctions')
   
   state_machine_definition = {
       "Comment": "GT Job Creation Workflow",
       "StartAt": "CheckCameras",
       "States": {
           "CheckCameras": {
               "Type": "Task",
               "Resource": "arn:aws:lambda:...:check-cameras",
               "Next": "CreateLabelingJob"
           },
           "CreateLabelingJob": {
               "Type": "Task",
               "Resource": "arn:aws:lambda:...:create-gt-job",
               "End": True
           }
       }
   }
   
   # Step Functions実行
   response = client.create_state_machine(
       name='SafetyGTJobCreator',
       definition=json.dumps(state_machine_definition),
       roleArn='arn:aws:iam::...:role/stepfunctions-execution-role'
   )
   print(f"State Machine ARN: {response['stateMachineArn']}")
   

3. 推論パイプライン (SageMaker Endpoint + SQS + Lambda)

   • 各ユースケース(PPE、Housekeeping)ごとに独立したSQSキューを使用
   • SageMaker EndpointでYOLOv8モデルによりオブジェクト検出
   • 検出結果はDynamoDBに保存し、違反時はSNSで通知

4. インテリジェントアラーム検出 (4段階プロセス)

   • 第1段階: オブジェクト検出 - YOLOモデルで作業者、機器、PPEアイテムを識別
   • 第2段階: デジタルテープベースのゾーン分析 - 事前定義された5Sテープ領域とオブジェクトの重なり率を計算（通常50%閾値）
   • 第3段階: Loiter Timeアルゴリズム - 一時的違反と継続的違反を区別（分単位で追跡）
   • 第4段階: 多層検証とアラーム生成 - 信頼度閾値、RLEマスク比較など最終検証後にアラーム送信

コード例：Loiter Timeアルゴリズム (Python)

import numpy as np
from typing import Dict, List

class LoiterTimeTracker:
    """オブジェクトごとの継続違反時間を追跡するLoiter Timeアルゴリズム"""
    
    def __init__(self, max_loiter_minutes: Dict[str, int]):
        """
        Args:
            max_loiter_minutes: オブジェクトタイプ別許容最大継続時間（分）
        """
        self.max_loiter_minutes = max_loiter_minutes
        self.violation_history: Dict[str, List[bool]] = {}  # オブジェクトID -> 時間別違反有無

    def update(self, object_id: str, is_violating: bool) -> bool:
        """
        オブジェクトの現在の違反状態を更新し、アラーム要否を返す
        
        Returns:
            True if alarm should be triggered
        """
        if object_id not in self.violation_history:
            self.violation_history[object_id] = []
        
        self.violation_history[object_id].append(is_violating)
        
        # 最近N分の違反連続性を確認
        recent = self.violation_history[object_id][-10:]  # 最近10分
        consecutive_violations = 0
        for v in reversed(recent):
            if v:
                consecutive_violations += 1
            else:
                break
        
        # オブジェクトタイプ別閾値を確認
        obj_type = object_id.split('_')[0]
        threshold = self.max_loiter_minutes.get(obj_type, 3)  # デフォルト3分
        
        return consecutive_violations >= threshold

# 使用例
tracker = LoiterTimeTracker({
    'forklift': 1,      # フォークリフトは1分以上違反でアラーム
    'pallet': 5,        # パレットは5分許容
    'person': 2         # 人は2分
})

# シミュレーション（毎分呼び出し）
for minute in range(10):
    is_violating = np.random.random() > 0.7
    should_alarm = tracker.update('forklift_001', is_violating)
    if should_alarm:
        print(f"[ALARM] {minute+1}分目：フォークリフト001が違反ゾーンに1分以上滞留！")

合成データ生成：GLIGENでレアケースをカバー

実際の現場では稀な安全違反事例（例：床面の液体流出）は、学習データの確保が困難です。50万枚の画像を精査しても、わずか数百件の流出事例しか見つからないのが一般的です。

GLIGENパイプライン構成

GLIGEN(Grounded Language-to-Image Generation)は拡散ベース生成モデルで、バウンディングボックス座標を入力として、指定位置に目的のオブジェクトを含むリアルな画像を生成します。

# 例：GLIGEN推論用SageMaker Batch Transform入力データ生成
import json

# 512x512画像にパレット、ヘルメット、ベストを配置
prompt = {
    "prompt": "A realistic warehouse floor with a pallet jack and a worker wearing safety vest and hard hat",
    "bboxes": [
        {"class": "pallet_jack", "bbox": [0.1, 0.3, 0.4, 0.6]},  # x1, y1, x2, y2（正規化）
        {"class": "person", "bbox": [0.5, 0.2, 0.7, 0.8]},
        {"class": "hard_hat", "bbox": [0.55, 0.15, 0.65, 0.25]},
        {"class": "safety_vest", "bbox": [0.52, 0.3, 0.68, 0.6]}
    ],
    "negative_prompt": "blurry, low quality, distorted, unrealistic"
}

# SageMaker Batch Transform用入力ファイル生成
with open('input.jsonl', 'w') as f:
    f.write(json.dumps(prompt) + '\n')

結果とパフォーマンス

注意点： 合成データのみでも高い性能を達成できますが、実際の現場の多様な照明条件、カメラアングル、機器レイアウトを完全に代替できるわけではありません。初期モデル学習には効果的ですが、継続的な性能改善には実データを混合して使用することを推奨します。

実務適用のアドバイスと次のステップ

この技術の限界または注意点

• GPUコスト： SageMaker Serverful Inference（GPUインスタンス）はServerlessよりコストが高くなります。初期PoC段階では小規模から開始し、Auto Scalingポリシーを慎重に設定してください。
• データ不均衡： 合成データでレアケースを補強できますが、現実との乖離(Gap)が生じる可能性があります。継続的な実データ収集とモデル再学習が必要です。
• レイテンシ： 画像キャプチャからアラーム配信まで37秒かかります。リアルタイム性が重要なアプリケーション（例：ロボット衝突防止）には適さない場合があります。

次のステップ学習方向

1. マルチカメラキャリブレーション： 複数カメラの視野を統合し、3D空間でのオブジェクト位置を正確に推定する方法を学習してみてください。
2. エッジコンピューティング： AWS PanoramaやNVIDIA Jetsonなどのエッジデバイスにモデルをデプロイし、ネットワーク遅延を削減するアーキテクチャを検討してください。
3. MLOpsパイプライン： モデルバージョン管理、A/Bテスト、ロールバック戦略を含むCI/CDパイプラインを構築し、継続的なモデル改善を自動化してください。

合わせて読みたい記事

• ブラジルAI向け核心データセット、Nemotron-Personas-Brazilリリース
• ゲーム内AIエージェント推論コスト、こうして削減する NVIDIA IGI SDKコードエージェント実践ガイド