Manufacturing QC และ MLOps สำหรับสายการผลิต

1import numpy as np
2from sklearn.metrics import roc_curve
3
4
5def optimal_threshold(
6    y_true: np.ndarray,      # 1 = defect (positive), 0 = ok
7    y_score: np.ndarray,     # คะแนนความน่าจะเป็น defect
8    cost_false_accept: float = 50.0,  # ปล่อยของเสีย (แพง)
9    cost_false_reject: float = 1.0,   # ทิ้งของดี
10    defect_prior: float | None = None,
11) -> dict:
12    """เลือก threshold ที่ minimize expected cost บน ROC."""
13    fpr, tpr, thresholds = roc_curve(y_true, y_score)
14    # FRR = พลาด defect = 1 - tpr ; FAR ฝั่ง ok = fpr
15    frr = 1.0 - tpr
16    far = fpr
17
18    p_defect = defect_prior if defect_prior is not None else float(y_true.mean())
19    p_ok = 1.0 - p_defect
20
21    # expected cost ต่อชิ้น = ต้นทุน_พลาดdefect*P(defect)*FRR + ต้นทุน_ปล่อยเสีย*P(ok)*FAR
22    expected_cost = (
23        cost_false_reject * p_defect * frr
24        + cost_false_accept * p_ok * far
25    )
26    best = int(np.argmin(expected_cost))
27    return {
28        "threshold": float(thresholds[best]),
29        "far": float(far[best]),
30        "frr": float(frr[best]),
31        "expected_cost": float(expected_cost[best]),
32    }

1import numpy as np
2
3
4def margin_uncertainty(probs: np.ndarray) -> np.ndarray:
5    """ความไม่มั่นใจแบบ margin: 1 - (top1 - top2). สูง = ไม่มั่นใจ."""
6    sorted_p = np.sort(probs, axis=1)[:, ::-1]
7    margin = sorted_p[:, 0] - sorted_p[:, 1]
8    return 1.0 - margin
9
10
11def select_for_labeling(
12    probs: np.ndarray,       # (N, num_classes)
13    budget: int,
14    near_threshold: tuple[float, float] | None = None,
15    binary_score: np.ndarray | None = None,
16) -> np.ndarray:
17    """คืน index ของภาพที่ควรส่งให้คน label ภายใต้ budget ที่จำกัด."""
18    uncertainty = margin_uncertainty(probs)
19
20    # เน้นภาพที่คะแนนใกล้เส้นตัดสินใจ ซึ่งมักเป็นผลของ drift
21    if near_threshold is not None and binary_score is not None:
22        lo, hi = near_threshold
23        in_band = (binary_score >= lo) & (binary_score <= hi)
24        uncertainty = np.where(in_band, uncertainty + 1.0, uncertainty)
25
26    return np.argsort(uncertainty)[::-1][:budget]

1import torch
2import torch.nn as nn
3
4
5class EWC:
6    """Elastic Weight Consolidation: ปกป้อง weight สำคัญของ task เดิม."""
7
8    def __init__(self, model: nn.Module, lambda_: float = 1000.0):
9        self.model = model
10        self.lambda_ = lambda_
11        # ค่า weight อ้างอิงหลังเรียน task เดิม (theta*)
12        self.star = {n: p.detach().clone() for n, p in model.named_parameters()}
13        self.fisher: dict[str, torch.Tensor] = {}
14
15    def compute_fisher(self, loader, device: str) -> None:
16        """ประมาณ Fisher information = ความสำคัญของแต่ละ weight ต่อ task เดิม."""
17        fisher = {n: torch.zeros_like(p) for n, p in self.model.named_parameters()}
18        self.model.eval()
19        for x, y in loader:
20            x, y = x.to(device), y.to(device)
21            self.model.zero_grad()
22            loss = nn.functional.cross_entropy(self.model(x), y)
23            loss.backward()
24            for n, p in self.model.named_parameters():
25                if p.grad is not None:
26                    fisher[n] += p.grad.detach() ** 2
27        n_batches = max(len(loader), 1)
28        self.fisher = {n: f / n_batches for n, f in fisher.items()}
29
30    def penalty(self) -> torch.Tensor:
31        """loss term ที่ลงโทษการขยับ weight สำคัญออกจาก theta*."""
32        loss = torch.tensor(0.0, device=next(self.model.parameters()).device)
33        for n, p in self.model.named_parameters():
34            if n in self.fisher:
35                loss = loss + (self.fisher[n] * (p - self.star[n]) ** 2).sum()
36        return self.lambda_ * loss

1import random
2import torch
3import torch.nn as nn
4
5
6def train_step_continual(
7    model: nn.Module,
8    new_batch,                 # (x, y) จากข้อมูลใหม่
9    replay_buffer: list,       # ตัวอย่างเก่าที่เก็บไว้
10    ewc: "EWC",
11    optimizer: torch.optim.Optimizer,
12    device: str,
13    replay_size: int = 16,
14) -> float:
15    """หนึ่งก้าวการเทรนต่อเนื่อง: ข้อมูลใหม่ + experience replay + EWC penalty."""
16    model.train()
17    x_new, y_new = (t.to(device) for t in new_batch)
18
19    # Experience Replay: ดึงตัวอย่างเก่ามาผสม กัน distribution เก่าหายไป
20    if replay_buffer:
21        sampled = random.sample(replay_buffer, min(replay_size, len(replay_buffer)))
22        x_old = torch.stack([s[0] for s in sampled]).to(device)
23        y_old = torch.tensor([s[1] for s in sampled]).to(device)
24        x = torch.cat([x_new, x_old]); y = torch.cat([y_new, y_old])
25    else:
26        x, y = x_new, y_new
27
28    optimizer.zero_grad()
29    loss = nn.functional.cross_entropy(model(x), y) + ewc.penalty()
30    loss.backward()
31    optimizer.step()
32    return float(loss.item())