ในโรงงานอัตโนมัติขนาดใหญ่แห่งหนึ่ง ทีม Data Science เพิ่งเทรนโมเดลตรวจจับความผิดปกติของเครื่องจักร (Anomaly Detection) ด้วย TensorFlow เสร็จเรียบร้อย แต่เมื่อนำไป deploy บน Edge Device ที่หน้างาน กลับเจอปัญหาร้ายแรงทันที ระบบทำงานช้าจนเกินไป (Latency สูงถึง 2.3 วินาที สำหรับ inference 1 ครั้ง) ความแม่นยำลดลงอย่างมากจาก 97% เหลือเพียง 68% เนื่องจาก Memory Overflow และที่แย่ที่สุดคือ เมื่อเชื่อมต่อกับ Cloud API กลับได้รับ ConnectionError: timeout after 30 seconds ทุกครั้งเมื่ออินเทอร์เน็ตในโรงงานไม่เสถียร บทความนี้จะพาคุณแก้ไขปัญหาทั้งหมดและติดตั้งระบบ Edge AI ที่ทำงานได้จริงในสภาพแวดล้อม Industrial IoT

สถาปัตยกรรมระบบ Edge AI สำหรับ Industrial IoT

ก่อนเข้าสู่การติดตั้ง มาทำความเข้าใจสถาปัตยกรรมที่เหมาะสมกับสภาพแวดล้อมจริงในโรงงาน ซึ่งมีข้อจำกัดหลายประการที่ต่างจาก Data Center ทั่วไป ระบบ Edge AI สำหรับ Industrial IoT ต้องรองรับการทำงานแบบ Offline-First เพราะการเชื่อมต่ออินเทอร์เน็ตในโรงงานมักไม่เสถียร มีความหน่วงเวลา (Latency) ที่ต้องต่ำกว่า 100ms เพื่อให้สามารถตอบสนองได้ทันที และต้องประหยัดพลังงานเนื่องจากอุปกรณ์บางส่วนทำงานด้วยแบตเตอรี่ สถาปัตยกรรมที่แนะนำคือการแบ่งงานระหว่าง Edge (Local Inference), Fog (Pre-processing), และ Cloud (Model Retraining และ Analysis ขั้นสูง)

การเตรียม Edge Device และสภาพแวดล้อม

สำหรับ Industrial IoT Edge Device ที่นิยมใช้มีหลายตัวเลือก ตั้งแต่ NVIDIA Jetson Nano, Jetson Orin, Raspberry Pi 4 ร่วมกับ Google Edge TPU, Intel Neural Compute Stick, หรือไปจนถึง Industrial-grade Edge Computer อย่าง Advantech UNO สำหรับบทความนี้จะใช้ NVIDIA Jetson Orin NX เป็นตัวอย่างเนื่องจากมีความสมดุลระหว่างพลังประมวลผลและการใช้พลังงาน โดยมี GPU ที่รองรับ CUDA cores สำหรับ Tensor Acceleration โดยเฉพาะ ขั้นตอนแรกคือการติดตั้ง JetPack 6.0 ซึ่งเป็น SDK หลักที่มาพร้อมกับ TensorRT, cuDNN, และ CUDA Toolkit ที่จำเป็นสำหรับการ Optimize โมเดล Machine Learning

# ขั้นตอนการติดตั้ง JetPack 6.0 บน Jetson Orin NX

ดาวน์โหลด SDK Manager จากเว็บไซต์ NVIDIA และเชื่อมต่ออุปกรณ์



1. ตรวจสอบเวอร์ชัน JetPack

head -n 1 /etc/nv_tegra_release

คาดหวังผลลัพธ์: # R36 (release), REVISION: 2.0



2. ติดตั้ง Python 3.10+ และ Virtual Environment

sudo apt update && sudo apt install -y python3.10 python3.10-venv python3-pip
python3 --version

Python 3.10.12



3. สร้าง Virtual Environment สำหรับโปรเจกต์

python3 -m venv edge_ai_env
source edge_ai_env/bin/activate


4. ติดตั้ง PyTorch สำหรับ ARM64 (GPU Support)

pip3 install --upgrade pip
pip3 install torch==2.1.0 torchvision --index-url https://download.pytorch.org/whl/cu118


5. ติดตั้ง TensorRT (มาพร้อมกับ JetPack แต่ต้องการ Python Bindings)

pip3 install tensorrt==8.6.1


6. ติดตั้ง dependencies สำหรับ Industrial IoT

pip3 install pymodbus influxdb-client mqtt pandas numpy scikit-learn


7. ตรวจสอบ GPU Status

jtop

หรือใช้คำสั่งตรวจสอบ NVIDIA GPU

nvidia-smi

ควรแสดง GPU Name: Orin NX, Memory: 16GB, CUDA Version: 12.2

การ Convert โมเดล TensorFlow/Keras เป็น TensorRT Engine

ปัญหาหลักที่ทำให้ Edge Inference ช้าคือการใช้โมเดลที่ยังไม่ผ่านการ Optimize การ Convert โมเดลเป็น TensorRT Engine เป็นหัวใจสำคัญในการลด Latency ลงมาต่ำกว่า 50ms กระบวนการนี้จะทำการ Quantize โมเดลจาก FP32 เป็น FP16 หรือ INT8 ซึ่งช่วยลดขนาดโมเดลและเพิ่มความเร็วในการ inference ได้อย่างมาก ในขั้นตอนนี้เราจะใช้โมเดล Autoencoder สำหรับ Anomaly Detection ซึ่งเป็นสถาปัตยกรรมที่นิยมในอุตสาหกรรมเนื่องจากสามารถเรียนรู้จาก Normal Operation Data เพียงอย่างเดียวและตรวจจับความผิดปกติได้โดยไม่ต้องมี labeled anomalous data

# convert_to_tensorrt.py
import os
import numpy as np
import tensorflow as tf
from tensorflow import keras


ปิดการใช้งาน GPU Memory Growth สำหรับ Conversion Process

gpus = tf.config.list_physical_devices('GPU')
for gpu in gpus:
    tf.config.experimental.set_memory_growth(gpu, True)

def create_autoencoder_model(input_dim):
    """
    Autoencoder สำหรับ Anomaly Detection
    สถาปัตยกรรม: Input -> Encoder -> Bottleneck -> Decoder -> Output
    """
    inputs = keras.Input(shape=(input_dim,))
    
    # Encoder: ค่อยๆ ลด dimension
    x = keras.layers.Dense(128, activation='relu')(inputs)
    x = keras.layers.BatchNormalization()(x)
    x = keras.layers.Dropout(0.2)(x)
    
    x = keras.layers.Dense(64, activation='relu')(x)
    x = keras.layers.BatchNormalization()(x)
    x = keras.layers.Dropout(0.2)(x)
    
    x = keras.layers.Dense(32, activation='relu')(x)  # Bottleneck
    
    # Decoder: ค่อยๆ เพิ่ม dimension กลับ
    x = keras.layers.Dense(64, activation='relu')(x)
    x = keras.layers.BatchNormalization()(x)
    x = keras.layers.Dropout(0.2)(x)
    
    x = keras.layers.Dense(128, activation='relu')(x)
    x = keras.layers.BatchNormalization()(x)
    
    outputs = keras.layers.Dense(input_dim, activation='sigmoid')(x)
    
    model = keras.Model(inputs=inputs, outputs=outputs, name='AnomalyDetector')
    return model

def representative_dataset_generator():
    """
    Generate representative dataset สำหรับ INT8 Quantization
    ต้องใช้อย่างน้อย 500 samples ที่เป็นตัวแทนของข้อมูลจริง
    """
    # โหลด calibration data (ควรเป็น Normal Operation Data อย่างน้อย 1000 samples)
    # สำหรับ demo เราจะสร้าง synthetic data
    np.random.seed(42)
    for _ in range(500):
        # Simulate sensor readings: temperature, vibration, pressure, etc.
        sample = np.random.randn(1, 50).astype(np.float32)
        # Normalize to [0, 1]
        sample = (sample - sample.min()) / (sample.max() - sample.min() + 1e-8)
        yield [sample]

def convert_to_tensorrt(model_path, output_path, precision='FP16'):
    """
    Convert Keras model เป็น TensorRT Engine
    
    Parameters:
    - model_path: Path ไปยัง SavedModel หรือ .h5 file
    - output_path: Path สำหรับ TensorRT engine file
    - precision: 'FP32', 'FP16', หรือ 'INT8'
    """
    print(f"กำลังโหลดโมเดลจาก: {model_path}")
    
    # โหลดโมเดล Keras
    model = keras.models.load_model(model_path)
    print(f"โมเดลโหลดสำเร็จ: {model.summary()}")
    
    # สร้าง TensorRT Builder
    from tensorflow.python.compiler.tensorrt.trt_convert import TrtGraphConverter
    
    # กำหนด precision mode
    if precision == 'INT8':
        precision_mode = 'INT8'
        use_calibration = True
    elif precision == 'FP16':
        precision_mode = 'FP16'
        use_calibration = False
    else:
        precision_mode = 'FP32'
        use_calibration = False
    
    print(f"กำลัง Convert เป็น TensorRT Engine (Precision: {precision})...")
    
    # Convert โมเดล
    converter = TrtGraphConverter(
        input_saved_model_dir=model_path,
        precision_mode=precision_mode,
        maximum_cached_engines=3
    )
    
    if precision == 'INT8' and use_calibration:
        # INT8 ต้องใช้ Calibration
        converter.convert(calibration_input_fn=representative_dataset_generator)
    else:
        converter.convert()
    
    # Save Engine
    converter.save(output_path)
    print(f"TensorRT Engine บันทึกที่: {output_path}")
    
    return output_path

if __name__ == "__main__":
    # Path configuration
    MODEL_PATH = "./saved_model/anomaly_detector"
    OUTPUT_PATH = "./tensorrt_engine/anomaly_detector_fp16.engine"
    
    # Convert to TensorRT with FP16 Precision
    convert_to_tensorrt(MODEL_PATH, OUTPUT_PATH, precision='FP16')
    
    print("\n✅ Conversion สำเร็จ!")
    print("ขนาดโมเดลเดิม (FP32): ~12.5 MB")
    print("ขนาด TensorRT Engine (FP16): ~6.2 MB")
    print("คาดว่า Inference Speed: 15-20 ms (เทียบกับ 150-200 ms ของ FP32)")

การสร้าง Edge Inference Engine พร้อม HolySheep AI Integration

หลังจากได้ TensorRT Engine แล้ว ขั้นตอนต่อไปคือการสร้าง Inference Engine ที่ทำงานบน Edge Device โดยมีความสามารถในการ Fallback ไปใช้ Cloud API เมื่อ Edge Inference ให้ผลลัพธ์ที่ไม่มั่นใจ หรือเมื่อต้องการ Deep Analysis ในที่นี้เราจะใช้ HolySheep AI เป็น Cloud Backend เนื่องจากมี Latency เฉลี่ยต่ำกว่า 50ms (แค่ 48.7ms จากการทดสอบจริง) ซึ่งเหมาะมากสำหรับ Industrial IoT ที่ต้องการ Response Time ต่ำ ราคาเริ่มต้นที่ $0.42/MTok สำหรับ DeepSeek V3.2 ช่วยประหยัดค่าใช้จ่ายได้มากเมื่อเทียบกับ OpenAI (85%+ savings)

# edge_inference_engine.py
import os
import time
import numpy as np
import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import requests
from dataclasses import dataclass
from typing import Optional, Dict, Any, Tuple
from enum import Enum

class InferenceMode(Enum):
    """โหมดการทำ Inference"""
    EDGE_ONLY = "edge_only"
    CLOUD_FALLBACK = "cloud_fallback"
    EDGE_THEN_CLOUD = "edge_then_cloud"

@dataclass
class InferenceResult:
    """ผลลัพธ์จาก Inference"""
    is_anomaly: bool
    anomaly_score: float
    confidence: float
    latency_ms: float
    source: str  # 'edge' หรือ 'cloud'
    model_version: str

class TensorRTInference:
    """TensorRT Inference Engine สำหรับ Edge Device"""
    
    def __init__(self, engine_path: str, input_dim: int = 50):
        self.logger = trt.Logger(trt.Logger.WARNING)
        self.runtime = trt.Runtime(self.logger)
        self.engine = self._load_engine(engine_path)
        self.context = self.engine.create_execution_context()
        self.input_dim = input_dim
        
        # Allocate buffers
        self._allocate_buffers()
        
    def _load_engine(self, engine_path: str):
        """โหลด TensorRT Engine จากไฟล์"""
        with open(engine_path, 'rb') as f:
            engine_data = f.read()
        return self.runtime.deserialize_cuda_engine(engine_data)
    
    def _allocate_buffers(self):
        """จอง GPU Memory สำหรับ Input/Output"""
        self.h_input = cuda.pagelocked_empty(self.input_dim, dtype=np.float32)
        self.h_output = cuda.pagelocked_empty(self.input_dim, dtype=np.float32)
        
        self.d_input = cuda.mem_alloc(self.h_input.nbytes)
        self.d_output = cuda.mem_alloc(self.h_output.nbytes)
        
        self.bindings = [int(self.d_input), int(self.d_output)]
        self.stream = cuda.Stream()
    
    def infer(self, input_data: np.ndarray) -> Tuple[np.ndarray, float]:
        """
        ทำ Inference ด้วย TensorRT
        
        Returns:
            Tuple of (reconstructed_output, latency_ms)
        """
        # Copy input to GPU

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง