บน edge ที่ทรัพยากรจำกัดและสายพานวิ่งเร็ว เราจะไม่รันโมเดลหนักกับทุกชิ้น บทนี้ว่าด้วยสองแพทเทิร์น สถาปัตยกรรมที่ทำให้ throughput สูงและการตัดสินใจแม่นขึ้น: cascade pipeline และ sensor fusion ของภาพกับข้อมูลตาชั่ง
บริบท: throughput บน edge
ของส่วนใหญ่ในไลน์เป็นของดี การรันโมเดล segmentation ที่แพงกับทุกชิ้นทำให้ throughput ตกและคิวบวม กุญแจคือ จัดสรรงานหนักไปที่ชิ้นที่ “คุ้มจะตรวจละเอียด” เท่านั้น
Cascade pipeline: classify แล้วค่อย segment
stage แรกเป็น classifier ที่เบาและเร็ว (เช่น MobileNet) ตัดสินว่าชิ้นงาน “ปกติ” หรือ “น่าสงสัย” ของที่ปกติ (ส่วนใหญ่) จบที่ stage แรกเลย เฉพาะชิ้นที่น่าสงสัยจึงถูกส่งเข้า stage สองที่เป็น segmentation ที่หนักกว่าเพื่อหา mask และวัดขนาดตำหนิ ค่าเฉลี่ยงานต่อชิ้นจึงต่ำลงมาก
1from dataclasses import dataclass2import numpy as np345@dataclass6class Inspection:7 status: str # "pass" | "defect"8 stage: str # stage ที่ตัดสิน9 defect_area_px: float | None = None101112class CascadePipeline:13 """Stage 1: classifier เบา คัดกรอง. Stage 2: segmentation เฉพาะที่น่าสงสัย."""1415 def __init__(self, classifier, segmenter, suspect_threshold: float = 0.15):16 self.classifier = classifier # คืน P(defect)17 self.segmenter = segmenter # คืน boolean mask18 self.suspect_threshold = suspect_threshold1920 def inspect(self, image_rgb: np.ndarray) -> Inspection:21 p_defect = float(self.classifier.predict_proba(image_rgb))2223 # ตั้ง recall สูง: ของน่าสงสัยแม้เล็กน้อยก็ส่งไปตรวจซ้ำ24 if p_defect < self.suspect_threshold:25 return Inspection(status="pass", stage="classifier")2627 # Stage 2 (แพง) รันเฉพาะส่วนน้อยที่ผ่านมาถึง28 mask = self.segmenter.segment(image_rgb)29 area = float(mask.sum())30 status = "defect" if area > 0 else "pass"31 return Inspection(status=status, stage="segmenter", defect_area_px=area)ตั้งจุดทำงานของแต่ละ stage
ความผิดพลาดที่อันตรายที่สุดใน cascade คือ stage แรก “ปล่อย” ของเสียผ่านไป เพราะจะไม่ถูกตรวจซ้ำ จึงตั้ง stage แรกให้ recall สูง (จับของน่าสงสัยให้ครบ) ยอมให้ precision ต่ำได้ ส่วน false positive ที่หลุดมาจะถูก stage สองที่แม่นกว่ากรองทิ้ง การปรับ suspect_threshold คือการแลก ระหว่าง throughput กับความปลอดภัย
- threshold ต่ำ → recall สูง, ส่งของเข้า stage 2 มากขึ้น → ปลอดภัยขึ้นแต่ throughput ตก
- threshold สูง → ส่งเข้า stage 2 น้อย → throughput สูงแต่เสี่ยงปล่อยของเสีย
- เลือกจุดด้วย cost-based optimization เดียวกับบท QC โดยให้ recall ของ stage 1 เป็นข้อจำกัดหลัก
Sensor fusion: visual + น้ำหนัก/ความบริสุทธิ์
ภาพอย่างเดียวบอกได้แค่ลักษณะพื้นผิว แต่ของปลอม/มีโพรงในอาจ “หน้าตาผ่าน” เราจึง fuse feature จากภาพ เข้ากับข้อมูลจาก scale (น้ำหนักจริง) และเซ็นเซอร์ความบริสุทธิ์ feature ที่ทรงพลังคือ ความหนาแน่นโดยนัย = น้ำหนัก ÷ ปริมาตรที่ประเมินจากภาพ ถ้าเบี่ยงจากค่ามาตรฐานของโลหะที่อ้าง ก็เป็นสัญญาณผิดปกติ
ใช้ late fusion: ให้แต่ละ modality ผ่านเส้นทางของตัวเองแล้วค่อยรวมที่ระดับ feature ก่อนหัวตัดสินใจ วิธีนี้จัดการ scale ที่ต่างกันง่ายและทนต่อ sensor เสียกว่าการต่อ raw feature ตั้งแต่ต้น (early fusion)
1import torch2import torch.nn as nn345class LateFusionInspector(nn.Module):6 """รวม visual feature กับ feature เชิงกายภาพ (น้ำหนัก/ความหนาแน่น/ความบริสุทธิ์)."""78 def __init__(self, visual_dim: int = 256, scalar_dim: int = 4, hidden: int = 128):9 super().__init__()10 # เส้นทางของแต่ละ modality แยกกัน (ข้อดีของ late fusion)11 self.visual_head = nn.Sequential(12 nn.Linear(visual_dim, hidden), nn.ReLU(), nn.Dropout(0.2),13 )14 self.scalar_head = nn.Sequential(15 nn.Linear(scalar_dim, hidden), nn.ReLU(),16 )17 self.classifier = nn.Sequential(18 nn.Linear(hidden * 2, hidden), nn.ReLU(),19 nn.Linear(hidden, 1),20 )2122 def forward(23 self,24 visual_feat: torch.Tensor, # (B, visual_dim) จาก CNN backbone25 scalars: torch.Tensor, # (B, scalar_dim)26 ) -> torch.Tensor:27 v = self.visual_head(visual_feat)28 s = self.scalar_head(scalars)29 fused = torch.cat([v, s], dim=1)30 return self.classifier(fused) # logit ของ P(defect)313233def physical_features(weight_g: float, volume_mm3: float,34 purity_pct: float, ref_density: float) -> torch.Tensor:35 """สร้าง feature เชิงกายภาพ ความหนาแน่นโดยนัยเทียบค่ามาตรฐานของโลหะ."""36 density = weight_g / max(volume_mm3 / 1000.0, 1e-6) # g/cm^337 density_dev = (density - ref_density) / ref_density # ส่วนเบี่ยง38 return torch.tensor([weight_g, density, density_dev, purity_pct],39 dtype=torch.float32)เช็กลิสต์ก่อนสัมภาษณ์
- อธิบาย cascade ว่าเพิ่ม throughput ด้วยการส่งเฉพาะของน่าสงสัยเข้าโมเดลแพง และประเมินต้นทุนเฉลี่ยได้
- รู้ว่า stage แรกต้อง recall สูง เพราะของที่ถูกปล่อยจะไม่ถูกตรวจซ้ำ
- อธิบายความต่าง early vs late fusion และทำไม late fusion ทนต่อ sensor เสียกว่า
- ยกตัวอย่าง feature เชิงกายภาพ (ความหนาแน่นโดยนัย = น้ำหนัก/ปริมาตร) ที่จับของปลอมได้
- คิดถึง failure mode: modality ขาด, scale ที่ต่างกัน, และวิธีทำให้ระบบ degrade อย่างนุ่มนวล
สรุปสำคัญ
- cascade pipeline ใช้ classifier ที่เบาและเร็วคัดกรองก่อน ส่งเฉพาะชิ้นที่ 'น่าสงสัย' เข้าโมเดล segmentation ที่หนักกว่า เพิ่ม throughput รวมอย่างมาก
- ตั้ง threshold ของ stage แรกให้ recall สูง (ไม่พลาดของเสีย) แม้ precision ต่ำได้ เพราะ stage ที่สองจะกรองซ้ำ
- sensor fusion รวม feature จากภาพกับน้ำหนัก/ความบริสุทธิ์ การ fuse แบบ late (รวมที่ระดับ decision/feature) ยืดหยุ่นและทนต่อ sensor เสียกว่า early fusion
ควิซท้ายบท
01ข้อดีหลักของ multi-stage cascade (classification-then-segmentation) บน edge คืออะไร
02ควรตั้งจุดทำงานของ classifier stage แรกใน cascade อย่างไร
03ทำไม late fusion จึงมักทนทานกว่า early fusion ในการรวมข้อมูลภาพกับน้ำหนัก/ความบริสุทธิ์
04ในการ fuse 'น้ำหนักจริงที่ชั่งได้' กับ 'ปริมาตรที่ประเมินจากภาพ' เพื่อช่วยตรวจความบริสุทธิ์ของโลหะ แนวคิดใดถูกต้อง