ข้ามไปยังเนื้อหา
vibe-coder/academy
หลักสูตรทั้งหมด
MODULE 09

สถาปัตยกรรม Edge แบบ High-Throughput

ออกแบบ pipeline แบบ multi-stage cascade (จัดประเภทก่อนค่อย segment) เพื่อเพิ่ม throughput บน edge และทำ sensor fusion รวม visual feature กับข้อมูลน้ำหนัก/ความบริสุทธิ์จาก scale เพื่อการตัดสินใจที่แม่นขึ้น

  • Cascade
  • Edge
  • Sensor Fusion
  • Throughput
  • Triton

บน edge ที่ทรัพยากรจำกัดและสายพานวิ่งเร็ว เราจะไม่รันโมเดลหนักกับทุกชิ้น บทนี้ว่าด้วยสองแพทเทิร์น สถาปัตยกรรมที่ทำให้ throughput สูงและการตัดสินใจแม่นขึ้น: cascade pipeline และ sensor fusion ของภาพกับข้อมูลตาชั่ง

บริบท: throughput บน edge

ของส่วนใหญ่ในไลน์เป็นของดี การรันโมเดล segmentation ที่แพงกับทุกชิ้นทำให้ throughput ตกและคิวบวม กุญแจคือ จัดสรรงานหนักไปที่ชิ้นที่ “คุ้มจะตรวจละเอียด” เท่านั้น

Cascade pipeline: classify แล้วค่อย segment

stage แรกเป็น classifier ที่เบาและเร็ว (เช่น MobileNet) ตัดสินว่าชิ้นงาน “ปกติ” หรือ “น่าสงสัย” ของที่ปกติ (ส่วนใหญ่) จบที่ stage แรกเลย เฉพาะชิ้นที่น่าสงสัยจึงถูกส่งเข้า stage สองที่เป็น segmentation ที่หนักกว่าเพื่อหา mask และวัดขนาดตำหนิ ค่าเฉลี่ยงานต่อชิ้นจึงต่ำลงมาก

cascade.py
Python
1from dataclasses import dataclass
2import numpy as np
3
4
5@dataclass
6class Inspection:
7 status: str # "pass" | "defect"
8 stage: str # stage ที่ตัดสิน
9 defect_area_px: float | None = None
10
11
12class CascadePipeline:
13 """Stage 1: classifier เบา คัดกรอง. Stage 2: segmentation เฉพาะที่น่าสงสัย."""
14
15 def __init__(self, classifier, segmenter, suspect_threshold: float = 0.15):
16 self.classifier = classifier # คืน P(defect)
17 self.segmenter = segmenter # คืน boolean mask
18 self.suspect_threshold = suspect_threshold
19
20 def inspect(self, image_rgb: np.ndarray) -> Inspection:
21 p_defect = float(self.classifier.predict_proba(image_rgb))
22
23 # ตั้ง recall สูง: ของน่าสงสัยแม้เล็กน้อยก็ส่งไปตรวจซ้ำ
24 if p_defect < self.suspect_threshold:
25 return Inspection(status="pass", stage="classifier")
26
27 # 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)

fusion.py
Python
1import torch
2import torch.nn as nn
3
4
5class LateFusionInspector(nn.Module):
6 """รวม visual feature กับ feature เชิงกายภาพ (น้ำหนัก/ความหนาแน่น/ความบริสุทธิ์)."""
7
8 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 )
21
22 def forward(
23 self,
24 visual_feat: torch.Tensor, # (B, visual_dim) จาก CNN backbone
25 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)
31
32
33def 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^3
37 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
ทดสอบความเข้าใจ

ควิซท้ายบท

0/4 ข้อ
  1. 01ข้อดีหลักของ multi-stage cascade (classification-then-segmentation) บน edge คืออะไร

  2. 02ควรตั้งจุดทำงานของ classifier stage แรกใน cascade อย่างไร

  3. 03ทำไม late fusion จึงมักทนทานกว่า early fusion ในการรวมข้อมูลภาพกับน้ำหนัก/ความบริสุทธิ์

  4. 04ในการ fuse 'น้ำหนักจริงที่ชั่งได้' กับ 'ปริมาตรที่ประเมินจากภาพ' เพื่อช่วยตรวจความบริสุทธิ์ของโลหะ แนวคิดใดถูกต้อง

ตอบให้ครบทุกข้อแล้วกดส่งคำตอบเพื่อดูเฉลย