Đo latency, log JSON và phát hiện drift cho mô hình AI trong nhà máy

Mô hình đã xuất sang ONNX, đã kiểm chứng, đã tăng tốc bằng runtime phù hợp. Nhưng sau ba tuần chạy, kỹ sư PLC gọi: "Thỉnh thoảng trạm trả kết quả trễ, PLC báo timeout", rồi tuần sau QC báo: "Tỷ lệ NG tăng từ 3% lên 6% mà hàng nhìn vẫn bình thường". Không có số liệu, bạn chỉ có thể đoán. Bài này trang bị ba công cụ để trả lời bằng dữ liệu: đo thời gian từng bước của pipeline (latency, phân vị p95/p99), log có cấu trúc (JSON Lines, xoay vòng file), và giám sát trôi dữ liệu (drift): phát hiện ánh sáng, ảnh, phân bố điểm của mô hình đang lệch khỏi lúc nghiệm thu, trước khi tỷ lệ loại nhầm bùng lên. Bài trước "Tăng tốc suy luận: OpenVINO, TensorRT, FP16, INT8" giúp mô hình chạy nhanh; bài này giúp bạn biết nó có tiếp tục chạy nhanh và đúng hay không.

pip install onnxruntime torch onnx onnxscript scipy opencv-python numpy
# Bài viết chạy với onnxruntime 1.31.0, torch 2.14.1, scipy 1.18.1

1. Chuẩn bị: một mô hình nhỏ chạy bằng ONNX Runtime

Để có số đo thật, ta cần một mô hình thật. Đoạn dưới huấn luyện nhanh một CNN nhỏ phát hiện vết xước tối trên bề mặt sáng (ảnh xám 64×64), rồi xuất ONNX theo đúng quy trình ở bài "Xuất ONNX và kiểm tra kết quả khớp PyTorch". Phần còn lại của bài chỉ dùng ONNX Runtime, giống như trên máy trạm.

import cv2
import numpy as np
import torch
import torch.nn as nn

torch.manual_seed(0)
torch.set_num_threads(4)


def tao_anh(ng, rng, do_sang=1.0):
    """Ảnh xám 64x64: bề mặt sáng có vân nhám; NG có vết xước tối. do_sang mô phỏng đèn yếu dần."""
    nen = rng.normal(170, 5)                                  # mỗi phôi sáng/tối hơn nhau một chút
    anh = np.full((64, 64), nen, np.float32) + rng.normal(0, rng.uniform(4, 9), (64, 64))
    if ng:
        x1, y1 = rng.integers(5, 30, 2)
        x2, y2 = rng.integers(34, 59, 2)
        cv2.line(anh, (int(x1), int(y1)), (int(x2), int(y2)), 90, int(rng.integers(1, 3)))
    return np.clip(anh * do_sang, 0, 255).astype(np.uint8)


def tien_xu_ly(ds_anh):
    """List ảnh xám uint8 -> tensor float32 (N, 1, 64, 64), chuẩn hóa như lúc huấn luyện."""
    return (np.stack(ds_anh).astype(np.float32)[:, None] / 255.0 - 0.6) / 0.2


rng = np.random.default_rng(0)
nhan = np.array([i % 2 for i in range(600)])
X = torch.from_numpy(tien_xu_ly([tao_anh(n, rng) for n in nhan]))
y = torch.from_numpy(nhan).long()
model = nn.Sequential(
    nn.Conv2d(1, 8, 3, padding=1), nn.BatchNorm2d(8), nn.ReLU(), nn.MaxPool2d(2),
    nn.Conv2d(8, 16, 3, padding=1), nn.BatchNorm2d(16), nn.ReLU(), nn.MaxPool2d(2),
    nn.Conv2d(16, 16, 3, padding=1), nn.ReLU(), nn.AdaptiveMaxPool2d(1), nn.Flatten(), nn.Linear(16, 2))
opt = torch.optim.Adam(model.parameters(), lr=3e-3)
for ep in range(4):
    perm = torch.randperm(600)
    for i in range(0, 600, 32):
        idx = perm[i:i + 32]
        loss = nn.functional.cross_entropy(model(X[idx]), y[idx])
        opt.zero_grad()
        loss.backward()
        opt.step()
model.eval()
torch.onnx.export(model, (X[:1],), "kiem_tra.onnx", input_names=["images"], output_names=["logits"],
                  dynamic_shapes=({0: "batch"},), external_data=False, verbose=False)
print("loss cuối:", round(loss.item(), 4), "-> đã xuất kiem_tra.onnx")

loss cuối: 0.001 -> đã xuất kiem_tra.onnx

dynamic_shapes=({0: "batch"},) là dạng tuple (theo thứ tự đối số của forward), tương đương dạng dict theo tên tham số. Giá trị loss có thể khác chút ít trên máy bạn.

2. Đo thời gian từng bước của pipeline

Một chu kỳ kiểm tra gồm nhiều bước: chụp (grab: phơi sáng + truyền ảnh), tiền xử lý, suy luận, hậu xử lý (softmax, ngưỡng, vẽ), I/O (lưu ảnh NG, ghi log, gửi PLC). Chỉ đo tổng thì không biết tối ưu chỗ nào; chỉ đo suy luận thì bỏ sót những bước thường chậm hơn cả mô hình. Ta viết một bộ đo dùng with để bọc từng bước:

import time
from collections import defaultdict
from contextlib import contextmanager
from pathlib import Path

import onnxruntime as ort


class DoThoiGian:
    """Ghi thời gian (ms) của từng bước trong mỗi chu kỳ."""

    def __init__(self):
        self.mau = defaultdict(list)            # tên bước -> list thời gian ms
        self.chu_ky = {}

    @contextmanager
    def buoc(self, ten):
        t0 = time.perf_counter()                # đồng hồ đơn điệu, độ phân giải cao
        try:
            yield
        finally:
            self.chu_ky[ten] = (time.perf_counter() - t0) * 1000

    def ket_thuc_chu_ky(self, ghi=True):
        tong = sum(self.chu_ky.values())
        if ghi:
            for ten, ms in self.chu_ky.items():
                self.mau[ten].append(ms)
            self.mau["TONG"].append(tong)
        kq, self.chu_ky = dict(self.chu_ky, TONG=tong), {}
        return kq

    def bang(self):
        print(f"{'bước':12s} {'p50':>7s} {'p95':>7s} {'p99':>7s} {'max':>7s}  (ms, {len(self.mau['TONG'])} chu kỳ)")
        for ten, ds in self.mau.items():
            p50, p95, p99 = np.percentile(ds, [50, 95, 99])
            print(f"{ten:12s} {p50:7.2f} {p95:7.2f} {p99:7.2f} {max(ds):7.2f}")


tuy_chon = ort.SessionOptions()
tuy_chon.intra_op_num_threads = 2              # cố định số luồng: thời gian ổn định hơn, chừa CPU cho việc khác
t0 = time.perf_counter()
sess = ort.InferenceSession("kiem_tra.onnx", tuy_chon, providers=["CPUExecutionProvider"])
print(f"tạo session: {(time.perf_counter() - t0) * 1000:.1f} ms")

Path("anh_ng").mkdir(exist_ok=True)
do = DoThoiGian()
rng_sx = np.random.default_rng(1)


def mot_chu_ky(i, ghi=True):
    with do.buoc("chup"):
        time.sleep(0.004)                       # giả lập phơi sáng + truyền ảnh 4 ms
        anh = tao_anh(rng_sx.random() < 0.05, rng_sx)
    with do.buoc("tien_xu_ly"):
        x = tien_xu_ly([anh])
    with do.buoc("suy_luan"):
        z = sess.run(None, {"images": x})[0][0]
    with do.buoc("hau_xu_ly"):
        e = np.exp(z - z.max())
        p_ng = float(e[1] / e.sum())
        ng = p_ng > 0.5
    with do.buoc("io"):
        if ng:
            cv2.imwrite(f"anh_ng/{i:05d}.png", anh)
    return do.ket_thuc_chu_ky(ghi), p_ng


lan_dau, _ = mot_chu_ky(0, ghi=False)
print(f"chu kỳ đầu tiên: suy luận {lan_dau['suy_luan']:.2f} ms, tổng {lan_dau['TONG']:.2f} ms")
for i in range(1, 10):                          # khởi động nóng (warm-up), không tính
    mot_chu_ky(i, ghi=False)
for i in range(10, 510):
    mot_chu_ky(i)
do.bang()

tạo session: 4.4 ms
chu kỳ đầu tiên: suy luận 0.73 ms, tổng 5.69 ms
bước             p50     p95     p99     max  (ms, 500 chu kỳ)
chup            4.55    5.14    9.82   53.55
tien_xu_ly      0.07    0.15    0.34    1.41
suy_luan        0.44    0.70    1.24    3.90
hau_xu_ly       0.05    0.11    0.14    1.15
io              0.00    0.35    0.59    0.75
TONG            5.16    6.23   10.32   54.93

Con số của bạn sẽ khác (tùy CPU, tải máy, hệ điều hành); hãy đọc cấu trúc của bảng:

• Chu kỳ đầu tiên chậm hơn: ở đây suy luận lần đầu 0,73 ms so với trung vị 0,44 ms, vì lần chạy đầu ONNX Runtime cấp phát bộ nhớ, chọn thuật toán; NumPy/OpenCV nạp code lần đầu. Với mô hình nhỏ trên CPU chênh lệch còn nhỏ; với mô hình lớn trên GPU (CUDA, TensorRT), lần chạy đầu có thể mất từ hàng trăm ms tới vài giây. Vì vậy luôn warm-up vài chu kỳ bằng ảnh giả ngay khi khởi động trạm, trước khi báo READY cho PLC, và không tính chúng vào thống kê.
• p50 (trung vị) là thời gian "bình thường"; p95/p99 là thời gian mà 95%/99% chu kỳ nhanh hơn; max là chu kỳ tệ nhất. PLC không quan tâm trung bình: nếu timeout là 50 ms thì một chu kỳ 60 ms là một sản phẩm bị báo lỗi. Thiết kế theo p99 và max, có hệ số an toàn (ví dụ p99 < 50% thời gian cho phép).
• Bước "chụp" có sàn 4 ms vì sleep; trên trạm thật đây là phơi sáng + truyền ảnh, thường là bước dài nhất và không tối ưu được bằng code (xem bài về trigger và cycle time).

Biểu đồ tần suất (histogram) cho thấy hình dạng phân bố tốt hơn bảng: phân bố có "đuôi dài" bên phải là dấu hiệu của các chu kỳ thỉnh thoảng bị chậm (GC của Python, tiến trình khác chiếm CPU, ghi đĩa).

def histogram(ds, so_cot=8, rong=40):
    dem, bien = np.histogram(ds, bins=so_cot)
    for i, d in enumerate(dem):
        print(f"{bien[i]:6.2f}–{bien[i + 1]:6.2f} ms | {'#' * int(round(d / dem.max() * rong)):{rong}s} {d}")


histogram(do.mau["TONG"])
ngan_sach_ms = 8.0
vuot = [i for i, t in enumerate(do.mau["TONG"]) if t > ngan_sach_ms]
print(f"Chu kỳ vượt {ngan_sach_ms} ms: {len(vuot)}/{len(do.mau['TONG'])}")
if vuot:
    i = max(vuot, key=lambda k: do.mau["TONG"][k])
    buoc_cham = max((b for b in do.mau if b != "TONG"), key=lambda b: do.mau[b][i])
    print(f"Chu kỳ chậm nhất #{i}: {do.mau['TONG'][i]:.2f} ms, bước chậm nhất: {buoc_cham}")

  4.75– 11.02 ms | ######################################## 496
 11.02– 17.30 ms |                                          3
 17.30– 23.57 ms |                                          0
 23.57– 29.84 ms |                                          0
 29.84– 36.11 ms |                                          0
 36.11– 42.38 ms |                                          0
 42.38– 48.65 ms |                                          0
 48.65– 54.93 ms |                                          1
Chu kỳ vượt 8.0 ms: 10/500
Chu kỳ chậm nhất #381: 54.93 ms, bước chậm nhất: chup

Lần chạy này có một chu kỳ 55 ms, gấp 10 lần trung vị, trong khi p99 chỉ khoảng 10 ms: một sự kiện hiếm mà trung bình và cả p99 đều che mất. Bảng từng bước chỉ ra thủ phạm là bước "chụp": lệnh sleep(0.004) bị hệ điều hành đánh thức trễ khoảng 50 ms vì máy đang bận việc khác. Trên trạm thật, chuyện tương tự xảy ra với thread nhận ảnh khi phần mềm diệt virus quét đĩa hay Windows Update chạy nền. Lần chạy của bạn có thể không có chu kỳ như vậy, hoặc có ở bước khác. Câu hỏi đúng khi có chu kỳ vượt ngân sách không phải "trung bình bao nhiêu" mà "bước nào gây ra chu kỳ chậm". Nhờ ghi từng bước cho từng chu kỳ, bạn trả lời được ngay. Nếu thủ phạm là I/O, chuyển việc ghi sang thread nền (như dự án trạm vision mô phỏng ở chặng 9); nếu là suy luận, xem lại số luồng và các tiến trình khác trên máy.

3. Log có cấu trúc: JSON Lines

Log dạng câu văn ("Chu kỳ 6: OK, 4 vật, 9.0 ms") dễ đọc bằng mắt nhưng khó phân tích. Khi cần trả lời "p99 theo giờ trong ca đêm qua là bao nhiêu?" hay "tỷ lệ NG của phiên bản mô hình 1.0.3?", bạn cần log mà máy đọc được: mỗi dòng là một đối tượng JSON, gọi là JSON Lines. Mô-đun logging chuẩn của Python làm được việc này chỉ với một Formatter tự viết:

import json
import logging
from datetime import datetime
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler


class JsonFormatter(logging.Formatter):
    """Mỗi bản ghi -> một dòng JSON. Dữ liệu thêm truyền qua extra={"du_lieu": {...}}."""

    def format(self, record):
        doc = {"ts": datetime.fromtimestamp(record.created).isoformat(timespec="milliseconds"),
               "muc": record.levelname, "nguon": record.name, "tin": record.getMessage()}
        doc.update(getattr(record, "du_lieu", {}))
        if record.exc_info:
            doc["loi"] = self.formatException(record.exc_info)
        return json.dumps(doc, ensure_ascii=False)


def cai_log(thu_muc, max_bytes=1_000_000, so_file=5):
    Path(thu_muc).mkdir(exist_ok=True)
    log = logging.getLogger("tram")
    for h in list(log.handlers):                    # chạy lại nhiều lần không bị nhân đôi
        log.removeHandler(h)
        h.close()
    fh = RotatingFileHandler(Path(thu_muc) / "tram.jsonl", maxBytes=max_bytes,
                             backupCount=so_file, encoding="utf-8")
    fh.setFormatter(JsonFormatter())
    log.addHandler(fh)
    log.setLevel(logging.INFO)
    log.propagate = False
    return log


log = cai_log("log_demo", max_bytes=20_000, so_file=3)   # cố tình nhỏ để thấy xoay vòng
log.info("khoi_dong", extra={"du_lieu": {"model": "kiem_tra.onnx", "phien_ban": "1.0.0"}})
for i in range(400):
    kq, p_ng = mot_chu_ky(10_000 + i, ghi=False)
    muc = logging.WARNING if p_ng > 0.5 else logging.INFO
    log.log(muc, "chu_ky", extra={"du_lieu": {
        "chu_ky": i, "kq": "NG" if p_ng > 0.5 else "OK", "p_ng": round(p_ng, 4),
        "ms": {k: round(v, 2) for k, v in kq.items()}}})
try:
    1 / 0
except ZeroDivisionError:
    log.exception("loi_ghi_plc", extra={"du_lieu": {"chu_ky": 400}})

for p in sorted(Path("log_demo").iterdir()):
    print(f"{p.name:14s} {p.stat().st_size:6d} byte")
dong_cuoi = (Path("log_demo") / "tram.jsonl").read_text(encoding="utf-8").splitlines()
print(dong_cuoi[-2][:150])

tram.jsonl      12436 byte
tram.jsonl.1    19977 byte
tram.jsonl.2    19979 byte
tram.jsonl.3    19975 byte
{"ts": "2026-10-09T10:14:49.913", "muc": "INFO", "nguon": "tram", "tin": "chu_ky", "chu_ky": 399, "kq": "OK", "p_ng": 0.0002, "ms": {"chup": 5.14, "ti

Kích thước file, thời điểm và số ms trong dòng log thay đổi mỗi lần chạy. Cách hoạt động của RotatingFileHandler: khi file hiện tại sắp vượt maxBytes, nó đổi tên tram.jsonl → tram.jsonl.1, file .1 → .2…, file thứ backupCount bị xóa, rồi mở file mới. Tổng dung lượng log vì thế không bao giờ vượt maxBytes × (backupCount + 1): ổ đĩa máy trạm không bao giờ đầy vì log. Trong thực tế đặt khoảng 10–50 MB × 10–20 file tùy ổ đĩa.

Nếu muốn mỗi ngày một file (dễ tìm theo ngày, dễ gửi cho khách hàng), dùng TimedRotatingFileHandler:

theo_ngay = TimedRotatingFileHandler("log_demo/theo_ngay.jsonl", when="midnight", backupCount=30,
                                     encoding="utf-8")
theo_ngay.setFormatter(JsonFormatter())
print(theo_ngay.when, theo_ngay.backupCount, theo_ngay.suffix)
theo_ngay.close()

MIDNIGHT 30 %Y-%m-%d

Sau nửa đêm, file hôm trước được đổi tên thành theo_ngay.jsonl.2026-10-09 và chỉ giữ 30 file gần nhất. Lưu ý: việc xoay vòng chỉ xảy ra khi có bản ghi mới được ghi sau mốc thời gian, và trên Windows việc đổi tên file có thể thất bại nếu một tiến trình khác (ví dụ trình soạn thảo, hoặc một tiến trình Python thứ hai) đang mở file log. Mỗi file log chỉ nên có một tiến trình ghi.

Lợi ích thật sự đến khi phân tích: đọc lại log bằng vài dòng Python, không cần biểu thức chính quy.

def thu_tu(p):                                   # tram.jsonl.3 (cũ nhất) ... tram.jsonl (mới nhất)
    duoi = p.suffix[1:]
    return -int(duoi) if duoi.isdigit() else 0


ban_ghi = []
for f in sorted(Path("log_demo").glob("tram.jsonl*"), key=thu_tu):
    for dong in f.read_text(encoding="utf-8").splitlines():
        ban_ghi.append(json.loads(dong))
chu_ky = [b for b in ban_ghi if b["tin"] == "chu_ky"]
tong_ms = [b["ms"]["TONG"] for b in chu_ky]
print("số chu kỳ còn trong log:", len(chu_ky), "| NG:", sum(b["kq"] == "NG" for b in chu_ky))
print(f"p99 tổng: {np.percentile(tong_ms, 99):.2f} ms")
print("lỗi:", [(b["tin"], b["loi"].splitlines()[-1]) for b in ban_ghi if b["muc"] == "ERROR"])

số chu kỳ còn trong log: 314 | NG: 11
p99 tổng: 6.26 ms
lỗi: [('loi_ghi_plc', 'ZeroDivisionError: division by zero')]

Số NG và p99 phụ thuộc lần chạy (p99 tùy máy). Một phần chu kỳ đầu đã "rơi" khỏi log vì file thứ 4 bị xóa khi xoay vòng: đó là cái giá của giới hạn dung lượng, và là lý do kết quả từng sản phẩm phải lưu vào cơ sở dữ liệu (SQLite như ở chặng 9), không chỉ trong log. Log để chẩn đoán; CSDL để truy xuất nguồn gốc.

4. Chính sách log: ghi gì, ở mức nào

Mỗi dòng log của một chu kỳ nên có đủ để tái hiện chu kỳ đó mà không cần hỏi ai:

5. Giám sát trôi dữ liệu (drift)

Drift là khi dữ liệu lúc sản xuất dần khác dữ liệu lúc huấn luyện/nghiệm thu: đèn LED yếu dần theo giờ chạy, kính bảo vệ bám bụi, ống kính lệch nét do rung, nhà cung cấp đổi phôi có màu hơi khác. Mô hình không báo lỗi; nó chỉ âm thầm cho điểm khác đi. Trước hết, xem mô hình của ta nhạy với độ sáng thế nào:

def diem_ng(ds_anh):
    z = sess.run(None, {"images": tien_xu_ly(ds_anh)})[0]
    e = np.exp(z - z.max(axis=1, keepdims=True))
    return (e / e.sum(axis=1, keepdims=True))[:, 1]


rng_t = np.random.default_rng(5)
print("độ sáng | P(NG) TB hàng OK | tỷ lệ NG (thật 3%) | vùng không chắc 0,2–0,8")
for ds in [1.0, 0.9, 0.85, 0.8, 0.75, 0.7, 0.65]:
    la_ng = rng_t.random(1000) < 0.03
    p = diem_ng([tao_anh(n, rng_t, ds) for n in la_ng])
    print(f"  {ds:4.2f}  |      {p[~la_ng].mean():.4f}      |       {(p > 0.5).mean() * 100:4.1f}%"
          f"       |     {((p > 0.2) & (p < 0.8)).mean() * 100:5.1f}%")

độ sáng | P(NG) TB hàng OK | tỷ lệ NG (thật 3%) | vùng không chắc 0,2–0,8
  1.00  |      0.0008      |        4.1%       |       0.0%
  0.90  |      0.0052      |        2.9%       |       0.0%
  0.85  |      0.0238      |        3.3%       |       0.7%
  0.80  |      0.1093      |        4.9%       |      15.1%
  0.75  |      0.3361      |       23.9%       |      67.6%
  0.70  |      0.6375      |       76.4%       |      73.4%
  0.65  |      0.8595      |       99.6%       |      21.5%

Đây là kiểu hành vi rất hay gặp ở mô hình thật: khi ảnh tối đi 10–15%, quyết định OK/NG chưa thay đổi (tỷ lệ NG vẫn quanh 3%), nhưng điểm P(NG) trung bình của hàng tốt đã tăng từ 0,0008 lên 0,024, gấp khoảng 30 lần. Tối đi 20%, 15% số ảnh rơi vào vùng không chắc chắn; tối đi 25%, tỷ lệ NG nhảy lên 24%: loại nhầm (overkill) bùng lên đột ngột. Nếu chỉ theo dõi tỷ lệ NG, bạn phát hiện khi đã quá muộn. Theo dõi phân bố điểm và thống kê ảnh cho phép báo động sớm, khi kỹ thuật viên còn kịp lau kính hay thay đèn.

5.1. Ba thống kê ảnh rẻ tiền

Ba con số này tính trong dưới 1 ms cho ảnh vài megapixel nếu tính trên ảnh thu nhỏ hoặc ROI, nên có thể tính cho mọi chu kỳ.

5.2. So phân bố điểm: PSI và kiểm định KS

Lưu điểm P(NG) của vài nghìn chu kỳ lúc nghiệm thu làm đường cơ sở (baseline). Sau đó mỗi cửa sổ (ví dụ 200 chu kỳ gần nhất) so với đường cơ sở bằng hai thước đo:

• PSI (Population Stability Index): chia điểm thành 10 ngăn theo phân vị của đường cơ sở (mỗi ngăn chứa 10% dữ liệu cơ sở), đếm tỷ lệ của cửa sổ mới trong từng ngăn, rồi tính PSI = Σ (p_mới − p_{cơ sở}) × ln(p_mới / p_{cơ sở}). Quy ước kinh nghiệm thường dùng: PSI < 0,1 ổn định; 0,1–0,25 lệch vừa, cần theo dõi; > 0,25 lệch đáng kể.
• Kiểm định Kolmogorov–Smirnov (KS) hai mẫu (scipy.stats.ks_2samp): đo khoảng cách lớn nhất giữa hai hàm phân phối tích lũy, cho giá trị p. p rất nhỏ (ví dụ < 0,001) nghĩa là khó có chuyện hai mẫu đến từ cùng phân bố. Với cửa sổ lớn, KS phát hiện cả những lệch rất nhỏ không đáng quan tâm, nên dùng nó như một tín hiệu kèm PSI, không dùng một mình.

5.3. Biểu đồ kiểm soát tỷ lệ NG (p-chart)

Tỷ lệ NG trong cửa sổ n chu kỳ dao động tự nhiên. Biểu đồ p (p-chart) trong kiểm soát quá trình thống kê đặt giới hạn: p̄ ± 3·√(p̄(1−p̄)/n), với p̄ là tỷ lệ NG lúc nghiệm thu. Ví dụ p̄ = 3%, n = 200: σ = √(0,03 × 0,97 / 200) ≈ 1,2%, giới hạn trên ≈ 3% + 3,6% = 6,6%. Cửa sổ có 14 NG / 200 = 7% vượt giới hạn: không còn là dao động ngẫu nhiên. Áp dụng tương tự cho tỷ lệ "vùng không chắc chắn" (điểm 0,2–0,8): lúc nghiệm thu gần 0%, nên đặt một sàn tối thiểu cho giới hạn (ở đây 1%) để tránh báo động vì một ảnh lẻ.

5.4. Bộ giám sát hoàn chỉnh và mô phỏng đèn yếu dần

from scipy.stats import ks_2samp


def thong_ke_anh(anh):
    return (float(anh.mean()), float(anh.std()), float(cv2.Laplacian(anh, cv2.CV_64F).var()))


def psi(co_so, moi, so_ngan=10):
    bien = np.quantile(co_so, np.linspace(0, 1, so_ngan + 1))
    bien[0], bien[-1] = -np.inf, np.inf
    a = np.histogram(co_so, bien)[0] / len(co_so)
    b = np.histogram(moi, bien)[0] / len(moi)
    a, b = np.clip(a, 1e-4, None), np.clip(b, 1e-4, None)      # tránh log(0)
    return float(np.sum((b - a) * np.log(b / a)))


class GiamSatTroi:
    def __init__(self, diem_cs, tk_cs, ng_cs, n=200, lech_sang=0.08, psi_max=0.25, ks_p_min=1e-3):
        self.diem_cs = np.asarray(diem_cs)
        self.tk_cs = np.asarray(tk_cs).mean(axis=0)              # độ sáng, tương phản, độ nét cơ sở
        self.n, self.lech_sang, self.psi_max, self.ks_p_min = n, lech_sang, psi_max, ks_p_min
        sigma = np.sqrt(ng_cs * (1 - ng_cs) / n)
        self.ucl_ng = ng_cs + 3 * sigma
        self.ucl_khong_chac = 0.01                                 # sàn tối thiểu, xem mục 5.3
        self.diem, self.tk = [], []

    def cap_nhat(self, anh, p_ng):
        """Gọi mỗi chu kỳ. Trả về None, hoặc dict kết quả khi đủ một cửa sổ."""
        self.diem.append(p_ng)
        self.tk.append(thong_ke_anh(anh))
        if len(self.diem) < self.n:
            return None
        d, tk = np.array(self.diem), np.array(self.tk).mean(axis=0)
        self.diem, self.tk = [], []
        kq = {"sang": tk[0] / self.tk_cs[0] - 1, "tuong_phan": tk[1] / self.tk_cs[1] - 1,
              "do_net": tk[2] / self.tk_cs[2] - 1, "psi": psi(self.diem_cs, d),
              "ks_p": ks_2samp(self.diem_cs, d).pvalue, "ty_le_ng": (d > 0.5).mean(),
              "khong_chac": ((d > 0.2) & (d < 0.8)).mean()}
        canh_bao = []
        if abs(kq["sang"]) > self.lech_sang:
            canh_bao.append("SANG")
        if kq["psi"] > self.psi_max:
            canh_bao.append("PSI")
        if kq["ks_p"] < self.ks_p_min:
            canh_bao.append("KS")
        if kq["ty_le_ng"] > self.ucl_ng:
            canh_bao.append("NG>UCL")
        if kq["khong_chac"] > self.ucl_khong_chac:
            canh_bao.append("KHONG_CHAC")
        kq["canh_bao"] = canh_bao
        return kq


# Đường cơ sở: 2000 chu kỳ lúc nghiệm thu, đèn mới
rng_cs = np.random.default_rng(11)
anh_cs = [tao_anh(n, rng_cs) for n in rng_cs.random(2000) < 0.03]
diem_cs = diem_ng(anh_cs)
gs = GiamSatTroi(diem_cs, [thong_ke_anh(a) for a in anh_cs], ng_cs=(diem_cs > 0.5).mean())
print(f"Cơ sở: tỷ lệ NG {(diem_cs > 0.5).mean() * 100:.1f}%, UCL tỷ lệ NG {gs.ucl_ng * 100:.1f}%")

# Sản xuất: 30 cửa sổ x 200 chu kỳ; từ cửa sổ 10 đèn yếu dần 1,8% mỗi cửa sổ
rng_sx2 = np.random.default_rng(12)
print("cửa sổ  sáng   sáng±   PSI     KS p     NG%  k.chắc%  cảnh báo (* = mới bật)")
da_bao = set()
for w in range(30):
    ds = 1.0 - 0.018 * max(0, w - 9)
    anh_w = [tao_anh(n, rng_sx2, ds) for n in rng_sx2.random(200) < 0.03]
    p_w = diem_ng(anh_w)                     # trên trạm: điểm đã có sẵn từ chu kỳ kiểm tra
    for a, p in zip(anh_w, p_w):
        kq = gs.cap_nhat(a, float(p))
    moi = set(kq["canh_bao"]) - da_bao
    da_bao |= moi
    if w % 3 == 0 or moi:                    # in mỗi 3 cửa sổ, và khi có cảnh báo mới
        nhan_cb = ",".join(c + ("*" if c in moi else "") for c in kq["canh_bao"])
        print(f"{w:5d}   {ds:5.3f}  {kq['sang'] * 100:+5.1f}%  {kq['psi']:6.3f}  {kq['ks_p']:8.1e}"
              f"  {kq['ty_le_ng'] * 100:5.1f}  {kq['khong_chac'] * 100:5.1f}   {nhan_cb}")

Cơ sở: tỷ lệ NG 3.1%, UCL tỷ lệ NG 6.8%
cửa sổ  sáng   sáng±   PSI     KS p     NG%  k.chắc%  cảnh báo (* = mới bật)
    0   1.000   +0.5%   0.039   8.5e-01    3.5    0.0
    3   1.000   -0.1%   0.083   8.7e-02    3.0    0.0
    6   1.000   -0.2%   0.058   4.0e-01    4.0    0.0
    9   1.000   +0.2%   0.055   1.1e-01    1.0    0.0
   11   0.964   -3.5%   0.233   4.1e-06    2.5    0.0   KS*
   12   0.946   -5.2%   0.423   5.8e-15    2.5    0.0   PSI*,KS
   14   0.910   -9.5%   2.729   2.8e-57    6.5    0.0   SANG*,PSI,KS
   15   0.892  -10.9%   3.660   1.7e-65    3.5    0.0   SANG,PSI,KS
   18   0.838  -16.2%   7.005  1.2e-147    0.5    2.5   SANG,PSI,KS,KHONG_CHAC*
   21   0.784  -21.7%   8.283  1.1e-220    6.0   31.5   SANG,PSI,KS,KHONG_CHAC
   22   0.766  -23.1%   8.283  6.2e-229   10.5   45.5   SANG,PSI,KS,NG>UCL*,KHONG_CHAC
   24   0.730  -27.2%   8.283  6.2e-229   44.5   81.0   SANG,PSI,KS,NG>UCL,KHONG_CHAC
   27   0.676  -32.7%   8.283  6.2e-229   93.0   46.0   SANG,PSI,KS,NG>UCL,KHONG_CHAC

Đọc bảng theo thời gian (mỗi cửa sổ 200 chu kỳ; với cycle time 1 giây, một cửa sổ khoảng 3 phút):

• Cửa sổ 0–9 (đèn mới): PSI 0,04–0,08, p của KS lớn, tỷ lệ NG dao động 1–4,5% quanh 3%, không có cảnh báo. Dao động này là bình thường; bộ giám sát không được kêu ở đây.
• Cửa sổ 11–12 (tối đi 3,5–5%): KS rồi PSI bật. Tỷ lệ NG, độ sáng còn hoàn toàn bình thường; chỉ phân bố điểm đã dịch. Đây là cảnh báo sớm nhất, mức "theo dõi".
• Cửa sổ 14 (tối 9,5%): cảnh báo độ sáng bật. Đây là lúc nên báo kỹ thuật viên kiểm tra đèn và kính.
• Cửa sổ 18: bắt đầu có ảnh rơi vào vùng không chắc chắn; cửa sổ 22: tỷ lệ NG vượt giới hạn trên của p-chart, và chỉ vài cửa sổ sau tỷ lệ NG lên 44–93%, tức dây chuyền gần như loại toàn bộ.

Từ cảnh báo đầu tiên (cửa sổ 11) tới lúc tỷ lệ NG bùng lên (cửa sổ 22) là 11 cửa sổ, tức 2.200 sản phẩm: đủ thời gian để xử lý khi trạm còn chạy đúng. Nếu chỉ theo dõi tỷ lệ NG, cảnh báo đầu tiên đến khi đã loại nhầm hàng loạt. Lưu ý PSI dừng ở 8,283: khi toàn bộ cửa sổ mới rơi vào một ngăn, công thức chạm trần do kẹp tỷ lệ tối thiểu 10^-4; con số tuyệt đối lúc đó không còn ý nghĩa, chỉ cần biết nó vượt xa 0,25.

Khi có cảnh báo drift, hành động không phải là tự động hạ ngưỡng hay huấn luyện lại ngay. Quy trình đúng: (1) báo kỹ thuật viên kiểm tra phần cứng (đèn, kính, nét, vị trí camera); (2) lưu các ảnh "không chắc chắn" để xem lại; (3) nếu nguyên nhân là thay đổi thật của sản phẩm (phôi mới), thu thập dữ liệu và huấn luyện lại, rồi cập nhật mô hình theo quy trình an toàn ở dự án cuối chặng.

6. Heartbeat: cho bên ngoài biết trạm còn sống

Ở chặng 9, heartbeat là một thanh ghi Modbus tăng mỗi giây để PLC biết vision còn chạy. Cùng ý tưởng áp dụng cho phần mềm giám sát (dự án cuối chặng: tiến trình giám sát tự khởi động lại trạm khi treo) và cho bảng điều khiển của nhà máy: trạm ghi một file JSON nhỏ mỗi giây, chứa thời điểm, số chu kỳ và vài chỉ số sức khỏe.

import os


def ghi_heartbeat(duong_dan, **du_lieu):
    """Ghi nguyên tử: ghi ra file tạm rồi os.replace, người đọc không bao giờ thấy file viết dở."""
    tam = Path(str(duong_dan) + ".tmp")
    tam.write_text(json.dumps({"ts": time.time(), "pid": os.getpid(), **du_lieu}), encoding="utf-8")
    os.replace(tam, duong_dan)


def kiem_tra_heartbeat(duong_dan, toi_da_s=3.0):
    try:
        hb = json.loads(Path(duong_dan).read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError):
        return "MAT", None
    tuoi = time.time() - hb["ts"]
    return ("SONG" if tuoi <= toi_da_s else "TREO"), round(tuoi, 1)


ghi_heartbeat("heartbeat.json", chu_ky=510, p99_ms=round(float(np.percentile(do.mau["TONG"], 99)), 2),
              trang_thai="CHAY", canh_bao=[])
print(kiem_tra_heartbeat("heartbeat.json"))

hb = json.loads(Path("heartbeat.json").read_text(encoding="utf-8"))
hb["ts"] -= 10                                   # giả lập: trạm ngừng cập nhật 10 giây trước
Path("heartbeat.json").write_text(json.dumps(hb), encoding="utf-8")
print(kiem_tra_heartbeat("heartbeat.json"))
print(kiem_tra_heartbeat("khong_ton_tai.json"))

('SONG', 0.0)
('TREO', 10.0)
('MAT', None)

Ba quy tắc của heartbeat: ghi từ vòng lặp chính (không phải một thread riêng chỉ để ghi heartbeat, vì thread đó vẫn chạy khi vòng lặp chính bị kẹt); ghi nguyên tử bằng os.replace; bên đọc phân biệt "không có file" (trạm chưa chạy) với "file cũ" (trạm treo).

7. Bài tập

do2 = DoThoiGian()


def chu_ky_co_ve(i):
    with do2.buoc("chup"):
        anh = tao_anh(rng_sx.random() < 0.05, rng_sx)
    with do2.buoc("suy_luan"):
        p_ng = float(diem_ng([anh])[0])
    with do2.buoc("ve"):
        mau = cv2.cvtColor(cv2.resize(anh, (640, 640), interpolation=cv2.INTER_NEAREST), cv2.COLOR_GRAY2BGR)
        mau_vien = (0, 0, 255) if p_ng > 0.5 else (0, 200, 0)
        cv2.rectangle(mau, (5, 5), (634, 634), mau_vien, 4)
        cv2.putText(mau, "NG" if p_ng > 0.5 else "OK", (20, 60), cv2.FONT_HERSHEY_SIMPLEX, 2, mau_vien, 3)
    do2.ket_thuc_chu_ky()


for i in range(200):
    chu_ky_co_ve(i)
p50 = {k: np.percentile(v, 50) for k, v in do2.mau.items()}
print({k: f"{v / p50['TONG'] * 100:.0f}%" for k, v in p50.items() if k != "TONG"})

{'chup': '20%', 'suy_luan': '37%', 've': '42%'}

print("nhóm    NG   p50 ms  max ms")
for i in range(0, len(chu_ky), 50):
    nhom = chu_ky[i:i + 50]
    t = [b["ms"]["TONG"] for b in nhom]
    print(f"{nhom[0]['chu_ky']:4d}+  {sum(b['kq'] == 'NG' for b in nhom):3d}  {np.percentile(t, 50):7.2f}  {max(t):6.2f}")

nhóm    NG   p50 ms  max ms
  86+    2     5.15    6.27
 136+    2     4.89    5.68
 186+    2     4.95    6.62
 236+    1     5.08    8.96
 286+    2     4.96    5.86
 336+    1     5.12    7.43
 386+    1     5.26    6.11

gs_net = GiamSatTroi(diem_cs, [thong_ke_anh(a) for a in anh_cs], ng_cs=(diem_cs > 0.5).mean())
rng_n = np.random.default_rng(22)
for w in range(14):
    sigma = 0.1 * max(0, w - 3)
    anh_w = [tao_anh(n, rng_n) for n in rng_n.random(200) < 0.03]
    if sigma > 0:
        anh_w = [cv2.GaussianBlur(a, (5, 5), sigma) for a in anh_w]
    for a, p in zip(anh_w, diem_ng(anh_w)):
        kq = gs_net.cap_nhat(a, float(p))
    if kq["do_net"] < -0.30:
        kq["canh_bao"].append("NET")
    print(f"{w:3d} sigma {sigma:3.1f}  độ nét {kq['do_net'] * 100:+6.1f}%  PSI {kq['psi']:6.3f}"
          f"  NG {kq['ty_le_ng'] * 100:4.1f}%  {kq['canh_bao']}")

  0 sigma 0.0  độ nét   +2.4%  PSI  0.036  NG  1.0%  []
  1 sigma 0.0  độ nét   +2.1%  PSI  0.053  NG  2.0%  []
  2 sigma 0.0  độ nét   -2.4%  PSI  0.050  NG  2.5%  []
  3 sigma 0.0  độ nét   -2.6%  PSI  0.057  NG  2.5%  []
  4 sigma 0.1  độ nét   +1.6%  PSI  0.041  NG  3.0%  []
  5 sigma 0.2  độ nét   +4.3%  PSI  0.039  NG  4.5%  []
  6 sigma 0.3  độ nét   +0.3%  PSI  0.068  NG  2.5%  []
  7 sigma 0.4  độ nét  -36.9%  PSI  0.936  NG  3.0%  ['PSI', 'KS', 'NET']
  8 sigma 0.5  độ nét  -74.2%  PSI  4.136  NG  3.5%  ['PSI', 'KS', 'KHONG_CHAC', 'NET']
  9 sigma 0.6  độ nét  -90.4%  PSI  5.406  NG  1.0%  ['PSI', 'KS', 'NET']
 10 sigma 0.7  độ nét  -96.0%  PSI  5.615  NG  1.5%  ['PSI', 'KS', 'NET']
 11 sigma 0.8  độ nét  -97.9%  PSI  4.821  NG  0.5%  ['PSI', 'KS', 'NET']
 12 sigma 0.9  độ nét  -98.8%  PSI  6.132  NG  1.0%  ['PSI', 'KS', 'NET']
 13 sigma 1.0  độ nét  -99.2%  PSI  5.205  NG  0.5%  ['PSI', 'KS', 'NET']

Tóm tắt

• Đo từng bước (chụp, tiền xử lý, suy luận, hậu xử lý, I/O) bằng time.perf_counter() trong một context manager; warm-up trước khi tính; báo p50/p95/p99/max và chỉ ra bước gây chu kỳ chậm.
• Thiết kế theo p99 và max so với timeout của PLC, không theo trung bình; luôn đo lại trên phần cứng đích.
• Log JSON Lines bằng logging + Formatter tự viết; RotatingFileHandler giới hạn dung lượng, TimedRotatingFileHandler(when="midnight") mỗi ngày một file; một file log chỉ một tiến trình ghi.
• Chính sách mức log rõ ràng; mỗi chu kỳ log đủ: thời điểm, mã sản phẩm, kết quả, điểm thô, thời gian từng bước, phiên bản mô hình.
• Drift: theo dõi độ sáng, tương phản, độ nét; so phân bố điểm với đường cơ sở bằng PSI (> 0,25 đáng kể) và KS; p-chart cho tỷ lệ NG và tỷ lệ vùng không chắc chắn. Điểm số trôi trước khi tỷ lệ NG trôi.
• Heartbeat ghi nguyên tử từ vòng lặp chính; bên giám sát phân biệt MẤT và TREO.

Bài tiếp theo: Docker cho kỹ sư AI, đóng gói ứng dụng cùng toàn bộ môi trường để chạy giống nhau trên mọi máy.