컴퓨터 비전 + 생성형 AI로 현장 안전 모니터링 자동화 AWS 서버리스 실전 가이드

왜 컴퓨터 비전과 생성형 AI가 안전 모니터링에 필요한가?

산업 현장의 안전 사고는 여전히 심각한 문제입니다. ILO에 따르면 전 세계적으로 연간 3억 9,500만 건의 업무 관련 비치명적 부상이 발생하며, 미국에서만 2023년에 1,765억 달러의 경제적 손실이 발생했습니다. OSHA는 차량 충돌 사망 사고가 100% 예방 가능하다고 말하지만, 여전히 주요 사망 원인 중 하나입니다.

전통적인 수동 안전 감사는 시간적, 공간적 한계가 명확합니다. 특정 시점의 스냅샷만 제공할 뿐, 24시간 365일 지속적인 모니터링은 불가능합니다. 수백 개의 시설로 확장될 때 이 문제는 더욱 심각해집니다.

이 글에서는 AWS 서버리스 아키텍처를 기반으로 한 컴퓨터 비전 솔루션을 소개합니다. 고정 카메라 네트워크를 활용해 PPE(개인 보호 장비) 착용 여부와 5S 테이프 기반 위험 구역을 실시간으로 감지하고, 생성형 AI로 합성 데이터를 만들어 모델 학습 효율을 극대화하는 방법을 다룹니다.

이 글은 AWS Architecture Blog의 원문을 기반으로 실무 관점에서 재구성했습니다.

핵심 아키텍처: 서버리스 Driver-Worker 패턴

수천 대의 카메라를 지원하려면 확장 가능한 아키텍처가 필수입니다. 이 솔루션의 핵심은 서버리스 Driver-Worker 패턴입니다. Driver가 작업을 분배하고 Worker가 이미지를 병렬로 처리하여, 한 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 알고리즘 - 일시적 위반 vs 지속적 위반 구분 (분 단위 추적)
   • 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 {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')

결과 및 성능

주의사항: 합성 데이터만으로도 높은 성능을 달성할 수 있지만, 실제 현장의 다양한 조명 조건, 카메라 각도, 장비 레이아웃을 완전히 대체할 수는 없습니다. 초기 모델 학습에는 효과적이지만, 지속적인 성능 개선을 위해서는 실제 데이터를 혼합하여 사용하는 것을 권장합니다.

실무 적용 조언 및 다음 단계

한국 개발 생태계에서의 적용 맥락

국내 제조 현장이나 물류 센터에 적용할 때는 다음과 같은 점을 고려해야 합니다.

• CCTV 법규 준수: 개인정보보호법에 따라 얼굴 블러 처리와 원본 이미지 보관 기간 설정이 필수입니다. AWS Rekognition의 얼굴 감지 기능을 사용할 때는 반드시 국내 법률을 확인하세요.
• 5S와의 통합: 한국 제조 현장은 5S(정리, 정돈, 청소, 청결, 습관화) 활동이 잘 정착되어 있습니다. 디지털 테이프를 기존 5S 테이프 색상 체계와 일치시키면 현장 수용성이 높아집니다.
• SI 환경 고려: 레거시 시스템과의 연동이 필요한 경우, SNS/SQS를 활용한 이벤트 기반 아키텍처가 유연하게 확장될 수 있습니다.

이 기술의 한계 또는 주의사항

• 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 코드 에이전트 실전 가이드