制造业AI质检落地全攻略：边缘小模型站岗 + 多模态大模型复诊，100张样本跑通一条产线（附完整代码与成本清单）

不写Demo的后端

126人浏览 · 2026-11-11 12:38:45

不写Demo的后端 · 2026-11-11 12:38:45 发布

制造业AI质检落地全攻略：边缘小模型站岗 + 多模态大模型复诊，100张样本跑通一条产线（附完整代码与成本清单）

适用读者：制造业数字化负责人、工业视觉工程师、想接制造业 AI 项目的中小团队。
阅读收益：一套能在 4~8 周内上线、单工位硬件成本控制在 1.7 万元以内（PoC 可压到 5000 元）的外观缺陷检测方案，包含数据、训练、边缘部署、云端复诊、主动学习、MES 对接和验收指标。
代码环境：Python 3.11 / Ultralytics / ONNXRuntime / FastAPI / PostgreSQL / vLLM。文末附 90 天落地排期与真实踩坑清单。

一、先泼一盆冷水：制造业 AI 质检的真正难点不是模型

2026 年，工业质检赛道已经很热。华为与江汽联合发布的 CV 质检大模型，公开信息里提到基于 30 亿参数 MoE 架构、130 万张汽车工业图像增量训练，缺陷检出率 99.99%，新场景开发周期缩短 95% 以上；各类赛事里的冠军方案也反复提到“100~200 张缺陷样本即可训练”“量化剪枝后在 Atlas 200 上低延迟推理”。

这些数字都是真的，但它们有一个共同前提：有人已经把数据、光源、节拍、标注规范和验收标准整理好了。

中小企业做质检，真实情况通常是：

缺陷样本只有几十张，良品图片倒有几万张；
没有标准光源，白天和夜班图像亮度差 30%；
产线节拍 1.2 秒/件，云端来回一次就超时；
老板要的是“漏检率低于 0.1%”，不是 mAP 提升 2 个点；
上线三个月后换了模具，模型开始批量误判，没人会重训。

所以这篇文章不讲“如何训一个更大的模型”，而是讲如何用一套工程闭环，把模型变成产线上稳定运行的检测能力。核心思路只有一句话：

边缘端用小模型保证速度和成本，云端用多模态大模型处理长尾和疑难，数据回流让模型每个月都变强一点。

这套架构在 PCB、注塑件、五金件、连接器、纺织面料等场景都能复用。下面从架构开始拆。

二、整体架构：三级决策，不让大模型上产线

2.1 为什么要分三级

层级	任务	延迟要求	技术选择	单件成本
L1 边缘	95% 以上的明确样本	< 100 ms	YOLO26n / RT-DETR + TensorRT FP16	接近 0
L2 云端	5% 左右的灰区样本	1~3 s，可异步	Qwen3-VL-8B / 7B，vLLM 批处理	约 0.001~0.01 元
L3 人工	云端仍不确定的样本	分钟级	复核台 + 标注工具	按件计费

关键点：大模型不进入实时控制回路。 它只做“离线复诊”和“疑难仲裁”，这样既不会被产线节拍卡死，也不会因为 API 抖动导致停线。

2.2 一个必须纠正的指标误区

很多方案把“准确率 99%”当验收标准，这是典型的99% 陷阱。假设来料不良率 1%，模型准确率 99%，看起来很好，但换算一下：

10 万件里 1000 件不良，模型可能漏掉 300 件；
同时把 990 件良品判成不良，产线每天要处理大量误剔。

制造业真正关心的是三个数：

漏检率 Escape Rate：不良品中被放行的比例，通常要求 < 0.1%~0.5%；
过杀率 Overkill Rate：良品中被误判为不良的比例，通常要求 < 0.5%~2%；
单件检测成本：包含硬件折旧、电费、云推理、人工复核。

所以后文的训练、阈值、评估，全部围绕“固定过杀率下的漏检率”来设计。

三、数据工程：100 张缺陷样本怎么起步

3.1 冷启动的三条路，按成本从低到高

路线 A：无监督异常检测先上线（0 标注）

用 Anomalib 里的 PatchCore / PaDiM，只用良品图训练。优点是当天就能跑；缺点是对“颜色接近、形状相似”的缺陷不敏感，且无法分类缺陷类型。适合作为第一周的 demo，让产线先看到效果。

路线 B：小样本有监督（推荐）

每个缺陷类别 30~100 张，配合强增强和迁移学习。公开赛事里“100~200 张样本”能做到可用精度，前提是缺陷定义清晰、拍摄一致。这是本文的主路线。

路线 C：合成缺陷补数据

用 CutPaste、DRAEM 或扩散模型，把真实缺陷贴到良品图上。NVIDIA 公开的案例里，合成 1 万张缺陷图后，真实数据采集和标注量减少约 30%，部署时间缩短约 25%。合成数据不能替代真实数据，但能把长尾类别从 20 张补到 500 张，效果很明显。

3.2 拍摄规范比模型重要

上线前必须和产线一起定死这几件事，否则后面全是玄学：

相机固定，镜头焦距、工作距离、曝光、增益全部锁定并记录；
光源角度固定，优先用同轴光或环形光，避免环境光干扰；
每张图附带工位号、时间戳、批次号、产品型号；
缺陷定义写成图文对照表，例如“划伤：长度 > 0.5 mm 的连续亮线”；
良品和不良品分目录存放，禁止混放；
训练集、验证集、测试集按批次切分，不能按图片随机切分，否则同一批产品的相似图会泄漏，指标虚高。

3.3 标注：用 SAM 把框选变成点选

人工画框一天只能标 300~500 张。用 SAM2 做半自动标注，点一下缺陷区域就能出掩码，再转成检测框，效率能提升 3~5 倍。标注完成后做一致性抽检：同一批图让第二个人盲标 10%，计算 IoU 一致率，低于 0.7 就说明缺陷定义需要重写。

3.4 数据目录与配置

dataset/
├── images/
│   ├── train/
│   ├── val/
│   └── test/
├── labels/
│   ├── train/
│   ├── val/
│   └── test/
└── defect.yaml

# defect.yaml
path: ./dataset
train: images/train
val: images/val
test: images/test

names:
  0: scratch      # 划伤
  1: dent         # 凹坑
  2: stain        # 污渍
  3: burr         # 毛刺
  4: missing      # 缺件

四、L1 边缘模型：训练、导出、量化

4.1 训练脚本（可直接跑）

# train_edge.py
from ultralytics import YOLO

def main():
    # 2026 年可用 yolo26n；若你的 ultralytics 版本不支持，换成 yolo11n / rtdetr-l 即可
    model = YOLO("yolo26n.pt")

    model.train(
        data="defect.yaml",
        epochs=120,
        imgsz=1024,          # 小缺陷必须上高分辨率，640 会丢细节
        batch=8,
        device=0,
        optimizer="AdamW",
        lr0=0.001,
        lrf=0.01,
        cos_lr=True,
        warmup_epochs=3,
        # 工业场景增强：保守，不要破坏缺陷形态
        hsv_h=0.01,
        hsv_s=0.4,
        hsv_v=0.3,
        degrees=3,
        translate=0.05,
        scale=0.25,
        fliplr=0.0,          # 划伤有方向性，禁止水平翻转
        flipud=0.0,
        mosaic=0.2,
        copy_paste=0.3,      # 小样本关键增强
        close_mosaic=15,
        # 正则与早停
        weight_decay=0.0005,
        patience=30,
        workers=8,
        project="runs/defect",
        name="edge_v1",
        exist_ok=True,
    )

if __name__ == "__main__":
    main()

几个容易踩的点：

imgsz=1024 显存占用大约是 640 的 2.5 倍，8G 显存建议 batch=4 并开 AMP；
fliplr=0 不是保守，是必须。左右翻转会把“左侧划伤”变成不存在的样本分布；
copy_paste 对小样本提升明显，但粘贴位置要限制在产品区域内，否则模型学到的是背景伪影；
训练完必须看每个类别的 PR 曲线，不要只看总体 mAP。长尾类别可能 mAP 只有 0.4，但业务上恰恰最致命。

4.2 导出 ONNX

from ultralytics import YOLO

model = YOLO("runs/defect/edge_v1/weights/best.pt")
model.export(
    format="onnx",
    imgsz=1024,
    opset=17,
    simplify=True,
    dynamic=False,   # 固定 batch，TensorRT 更快
    half=False,
)

4.3 TensorRT 引擎构建

在 Jetson Orin 或带 NVIDIA GPU 的工控机上：

trtexec \
  --onnx=best.onnx \
  --saveEngine=best_fp16.engine \
  --fp16 \
  --minShapes=images:1x3x1024x1024 \
  --optShapes=images:8x3x1024x1024 \
  --maxShapes=images:8x3x1024x1024 \
  --workspace=4096

如果追求更低成本，可以用 RK3588 盒子跑 INT8，但要注意：

量化校准集必须覆盖不同班次、不同光照的图片；
量化后重新扫一遍阈值，FP16 的 0.70 在 INT8 上可能要调到 0.65；
每批板子的一致性不同，建议固定供应商和固件版本。

4.4 边缘推理服务（ONNXRuntime 版，跨平台）

下面这段是完整可运行的最小服务，包含 letterbox、NMS、三级阈值和健康检查。

# edge_server.py
import time
import numpy as np
import cv2
import onnxruntime as ort
from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel
from typing import List

MODEL_PATH = "best.onnx"
IMG_SIZE = 1024
CONF_REVIEW = 0.35
CONF_REJECT = 0.70
IOU_THRES = 0.45
CLASS_NAMES = ["scratch", "dent", "stain", "burr", "missing"]

app = FastAPI(title="Defect Inspection Edge Service")

providers = ["CUDAExecutionProvider", "CPUExecutionProvider"]
session = ort.InferenceSession(MODEL_PATH, providers=providers)
input_name = session.get_inputs()[0].name


def letterbox(img, new_size=IMG_SIZE, color=(114, 114, 114)):
    h, w = img.shape[:2]
    r = min(new_size / h, new_size / w)
    nh, nw = int(round(h * r)), int(round(w * r))
    resized = cv2.resize(img, (nw, nh), interpolation=cv2.INTER_LINEAR)
    top = (new_size - nh) // 2
    left = (new_size - nw) // 2
    bottom = new_size - nh - top
    right = new_size - nw - left
    out = cv2.copyMakeBorder(
        resized, top, bottom, left, right,
        cv2.BORDER_CONSTANT, value=color
    )
    return out, r, (left, top)


def nms(boxes, scores, iou_thres=IOU_THRES):
    if len(boxes) == 0:
        return []
    x1, y1, x2, y2 = boxes[:, 0], boxes[:, 1], boxes[:, 2], boxes[:, 3]
    areas = (x2 - x1) * (y2 - y1)
    order = scores.argsort()[::-1]
    keep = []
    while order.size > 0:
        i = order[0]
        keep.append(i)
        xx1 = np.maximum(x1[i], x1[order[1:]])
        yy1 = np.maximum(y1[i], y1[order[1:]])
        xx2 = np.minimum(x2[i], x2[order[1:]])
        yy2 = np.minimum(y2[i], y2[order[1:]])
        w = np.maximum(0.0, xx2 - xx1)
        h = np.maximum(0.0, yy2 - yy1)
        inter = w * h
        union = areas[i] + areas[order[1:]] - inter + 1e-9
        iou = inter / union
        inds = np.where(iou <= iou_thres)[0]
        order = order[inds + 1]
    return keep


def postprocess(output, ratio, pad, orig_shape):
    # 兼容 YOLO 输出 [1, 4+nc, N]（导出后最常见）或 [1, N, 4+nc]
    pred = np.squeeze(output)
    if pred.ndim != 2:
        raise ValueError(f"unexpected output shape: {output.shape}")
    if pred.shape[0] < pred.shape[1]:
        pred = pred.T  # [N, 4+nc]

    boxes_xywh = pred[:, :4]
    class_scores = pred[:, 4:]
    class_ids = class_scores.argmax(axis=1)
    scores = class_scores.max(axis=1)

    mask = scores >= CONF_REVIEW
    boxes_xywh, scores, class_ids = boxes_xywh[mask], scores[mask], class_ids[mask]
    if len(scores) == 0:
        return []

    cx, cy, bw, bh = boxes_xywh.T
    boxes = np.stack([cx - bw / 2, cy - bh / 2, cx + bw / 2, cy + bh / 2], axis=1)
    keep = nms(boxes, scores, IOU_THRES)
    boxes, scores, class_ids = boxes[keep], scores[keep], class_ids[keep]

    # 还原到原图坐标
    h0, w0 = orig_shape
    lx, ly = pad
    boxes[:, [0, 2]] = (boxes[:, [0, 2]] - lx) / ratio
    boxes[:, [1, 3]] = (boxes[:, [1, 3]] - ly) / ratio
    boxes[:, [0, 2]] = boxes[:, [0, 2]].clip(0, w0 - 1)
    boxes[:, [1, 3]] = boxes[:, [1, 3]].clip(0, h0 - 1)

    results = []
    for box, score, cid in zip(boxes, scores, class_ids):
        results.append({
            "class_id": int(cid),
            "class_name": CLASS_NAMES[int(cid)] if int(cid) < len(CLASS_NAMES) else str(int(cid)),
            "score": round(float(score), 4),
            "bbox": [round(float(v), 1) for v in box],
        })
    return results


def decide(dets):
    if not dets:
        return "OK", 0.0
    top = max(d["score"] for d in dets)
    if top >= CONF_REJECT:
        return "NG", top
    if top >= CONF_REVIEW:
        return "REVIEW", top
    return "OK", top


class InspectResponse(BaseModel):
    decision: str
    top_score: float
    detections: List[dict]
    latency_ms: float


@app.post("/inspect", response_model=InspectResponse)
async def inspect(file: UploadFile = File(...)):
    t0 = time.perf_counter()
    data = await file.read()
    img = cv2.imdecode(np.frombuffer(data, np.uint8), cv2.IMREAD_COLOR)
    if img is None:
        return InspectResponse(decision="ERROR", top_score=0, detections=[], latency_ms=0)

    lb, ratio, pad = letterbox(img)
    tensor = lb[:, :, ::-1].transpose(2, 0, 1)[None].astype(np.float32) / 255.0
    output = session.run(None, {input_name: tensor})[0]

    dets = postprocess(output, ratio, pad, img.shape[:2])
    decision, top = decide(dets)
    latency = (time.perf_counter() - t0) * 1000

    return InspectResponse(
        decision=decision,
        top_score=top,
        detections=dets,
        latency_ms=round(latency, 2),
    )


@app.get("/health")
def health():
    return {"status": "ok", "model": MODEL_PATH, "providers": session.get_providers()}

启动：

pip install fastapi uvicorn onnxruntime-gpu opencv-python numpy python-multipart
uvicorn edge_server:app --host 0.0.0.0 --port 8000 --workers 2

生产上建议加三样东西：

请求队列：相机触发是突发的，用 asyncio.Queue 或 Redis Stream 削峰；
离线兜底：网络或 MES 挂了，检测结果先落本地 SQLite，恢复后补传；
图像留存策略：只留 NG 图和灰区图，OK 图按小时抽样，避免磁盘爆掉。

五、L2 云端复诊：让多模态模型只处理 5% 的难题

5.1 什么图该送云端

边缘模型输出 REVIEW 的图，加上以下情况：

检测到“未知类别”（分类头置信度普遍偏低）；
同一工位连续 N 件 NG，可能是镜头污染或物料换型；
新批次首件；
人工主动发起的复判请求。

5.2 复诊 Prompt 要约束成 JSON

云端不要问“这张图有没有缺陷”，而要给出缺陷字典、判定标准和输出格式，否则模型会自由发挥。

# cloud_review.py
import base64
import json
from openai import OpenAI

client = OpenAI(base_url="http://127.0.0.1:8001/v1", api_key="EMPTY")

SYSTEM = """你是工业质检复判专家。只根据图像和给定缺陷字典判断，不要臆测。
缺陷字典：
- scratch 划伤：长度>0.5mm 的连续亮线或暗线
- dent 凹坑：局部凹陷，边缘有阴影
- stain 污渍：颜色异常斑块，边界模糊
- burr 毛刺：边缘凸起，轮廓不平整
- missing 缺件：应存在的结构缺失
判定规则：
1. 无法确认时输出 uncertain，不要强行判定；
2. severity 取 minor / major / critical；
3. 必须给出图像中的可见依据，禁止引用未提供的信息。
只输出 JSON，不要 Markdown。"""

USER_TMPL = """产品型号：{model}
工位：{station}
边缘模型结果：{edge_json}
请输出：
{{"decision":"OK|NG|uncertain","defect_type":"...","severity":"minor|major|critical","confidence":0.0,"reason":"不超过80字"}}"""


def encode_image(path):
    with open(path, "rb") as f:
        return base64.b64encode(f.read()).decode()


def review(image_path, model, station, edge_result):
    resp = client.chat.completions.create(
        model="qwen3-vl-8b-instruct",
        temperature=0,
        max_tokens=300,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": SYSTEM},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": USER_TMPL.format(
                        model=model, station=station,
                        edge_json=json.dumps(edge_result, ensure_ascii=False)
                    )},
                    {"type": "image_url", "image_url": {
                        "url": f"data:image/jpeg;base64,{encode_image(image_path)}"
                    }},
                ],
            },
        ],
    )
    return json.loads(resp.choices[0].message.content)

5.3 云端结果的三种去向

云端结论	处理方式
NG 且 confidence ≥ 0.8	直接判 NG，写入追溯库
OK 且 confidence ≥ 0.8	放行，但把这张图加入“难例集”
uncertain 或 confidence < 0.8	进入人工复核队列，同时打标为高价值训练样本

注意：云端模型永远不能直接控制 PLC 剔除。它的结论必须经过规则引擎或人工确认，这是一条安全红线。

5.4 成本控制

用 vLLM 部署 8B 多模态模型，单张 1024 图约 0.8~1.5 秒，8 卡批处理可到 20+ QPS；
只传裁剪后的缺陷区域，不传整图，token 数能降 60% 以上；
灰区比例控制在 5% 以内，超过说明边缘模型或阈值有问题，先调模型，不要靠云端兜底；
云端服务放在厂内边缘服务器，避免图片出厂的合规风险。

六、数据闭环：让模型每个月自己变强

6.1 主动学习采样策略

不要随机抽样。优先选三类样本：

不确定性高：分数落在 0.35~0.70 灰区；
多样性高：用 CLIP 或 ResNet 特征做 k-center 贪婪选择，避免选到一堆相似图；
业务价值高：NG 漏检、客户投诉、新批次首件。

# active_learning.py
import numpy as np
from sklearn.preprocessing import normalize

def uncertainty_score(dets, lo=0.35, hi=0.70):
    if not dets:
        return 0.0
    top = max(d["score"] for d in dets)
    if not (lo <= top <= hi):
        return 0.0
    mid = (lo + hi) / 2
    return 1.0 - abs(top - mid) / ((hi - lo) / 2)


def greedy_k_center(features, k=200):
    """features: [N, D]，已 L2 归一化"""
    features = normalize(features)
    n = len(features)
    if n <= k:
        return np.arange(n)

    # 用样本到中心的距离初始化，避免从离群点开始
    center = features.mean(axis=0, keepdims=True)
    first = int(np.argmin(((features - center) ** 2).sum(axis=1)))
    selected = [first]
    min_dist = ((features - features[first]) ** 2).sum(axis=1)

    for _ in range(k - 1):
        idx = int(np.argmax(min_dist))
        selected.append(idx)
        dist = ((features - features[idx]) ** 2).sum(axis=1)
        min_dist = np.minimum(min_dist, dist)

    return np.array(selected)


def select_batch(records, features, budget=200, alpha=0.6):
    """
    records: [{"dets": [...], "id": ...}, ...]
    综合不确定性与多样性，返回预算内的样本下标
    """
    unc = np.array([uncertainty_score(r["dets"]) for r in records])
    # 先取不确定性最高的 3 倍预算作为候选池
    cand = np.argsort(-unc)[: min(len(records), budget * 3)]
    if len(cand) <= budget:
        return cand

    picked = greedy_k_center(features[cand], budget)
    # 再按不确定性重排，保证高价值样本优先送标
    return cand[picked[np.argsort(-unc[cand][picked])]]

6.2 重训与灰度发布

每月一次重训，流程固定：

导出当月灰区图、NG 图、人工复核结果；
去重（感知哈希 pHash，汉明距离 < 5 视为重复）；
合并进训练集，按批次重新切分；
训练新模型，在固定测试集上对比漏检率、过杀率；
新模型先在一条线、一个班次灰度 3 天；
指标不劣于旧模型才全量，保留一键回滚。

回滚不是可选项。产线上线第一原则是可控，不是先进。

6.3 漂移监测

每天统计这些指标，任一异常就告警：

灰区比例（基线 5%，超过 15% 告警）；
NG 率（相对基线波动超过 3 倍）；
图像亮度均值、清晰度（拉普拉斯方差）；
各类别检出分布；
云端复判与边缘判定的一致率（低于 90% 说明边缘模型该更新了）。

七、评估：用一张表说服老板

7.1 评估脚本

# evaluate.py
import numpy as np

def match(preds, gts, iou_thres=0.5):
    """单图匹配，返回 TP/FP/FN"""
    used = set()
    tp = 0
    for p in preds:
        best_iou, best_j = 0, -1
        for j, g in enumerate(gts):
            if j in used:
                continue
            iou = box_iou(p["bbox"], g["bbox"])
            if iou > best_iou:
                best_iou, best_j = iou, j
        if best_iou >= iou_thres:
            tp += 1
            used.add(best_j)
    fp = len(preds) - tp
    fn = len(gts) - tp
    return tp, fp, fn


def box_iou(a, b):
    x1, y1 = max(a[0], b[0]), max(a[1], b[1])
    x2, y2 = min(a[2], b[2]), min(a[3], b[3])
    inter = max(0, x2 - x1) * max(0, y2 - y1)
    area_a = (a[2] - a[0]) * (a[3] - a[1])
    area_b = (b[2] - b[0]) * (b[3] - b[1])
    return inter / (area_a + area_b - inter + 1e-9)


def sweep_threshold(samples, thresholds):
    """
    samples: [{"preds": [...], "gts": [...]}, ...]
    返回每个阈值下的漏检率与过杀率
    """
    rows = []
    for th in thresholds:
        tp = fp = fn = 0
        bad_total = bad_missed = 0
        good_total = good_false = 0
        for s in samples:
            preds = [p for p in s["preds"] if p["score"] >= th]
            t, f, n = match(preds, s["gts"])
            tp, fp, fn = tp + t, fp + f, fn + n
            if s["gts"]:
                bad_total += 1
                if not preds:
                    bad_missed += 1
            else:
                good_total += 1
                if preds:
                    good_false += 1
        rows.append({
            "threshold": th,
            "escape_rate": bad_missed / max(bad_total, 1),
            "overkill_rate": good_false / max(good_total, 1),
            "precision": tp / max(tp + fp, 1),
            "recall": tp / max(tp + fn, 1),
        })
    return rows

7.2 验收表模板

指标	上线前基线	目标	实测
漏检率	人工抽检约 1.2%	< 0.3%	待填
过杀率	无统计	< 1.5%	待填
单件节拍	人工 3 s	< 1.2 s	待填
灰区比例	-	< 8%	待填
人工复核量	全检	下降 70%	待填
连续 30 天可用率	-	> 99%	待填

注意：验收必须用上线后新批次的数据，不能用训练时的测试集。否则就是自欺欺人。

八、和 MES / PLC 对接：别让 AI 变成信息孤岛

8.1 数据流

相机触发 → 边缘推理 → 结果写入本地队列 → HTTP/MQTT → MES
                                      ↓
                                  NG 图上传 MinIO
                                      ↓
                                 追溯码绑定（SN + 工位 + 时间 + 模型版本）

8.2 幂等与重试

产线网络不稳定是常态。每条结果带唯一 event_id，MES 侧做幂等；本地用 SQLite 存待发送队列，失败指数退避，最多重试 24 小时。

import sqlite3
import json
import time
import uuid
import requests

DB = "pending.db"

def init_db():
    conn = sqlite3.connect(DB)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS pending (
            event_id TEXT PRIMARY KEY,
            payload TEXT NOT NULL,
            retry INTEGER DEFAULT 0,
            created_at REAL NOT NULL
        )
    """)
    conn.commit()
    conn.close()


def enqueue(payload):
    event_id = payload.get("event_id") or str(uuid.uuid4())
    payload["event_id"] = event_id
    conn = sqlite3.connect(DB)
    conn.execute(
        "INSERT OR IGNORE INTO pending(event_id, payload, created_at) VALUES (?, ?, ?)",
        (event_id, json.dumps(payload, ensure_ascii=False), time.time()),
    )
    conn.commit()
    conn.close()


def flush(mes_url, batch=50):
    conn = sqlite3.connect(DB)
    rows = conn.execute(
        "SELECT event_id, payload, retry FROM pending ORDER BY created_at LIMIT ?",
        (batch,),
    ).fetchall()
    for event_id, payload, retry in rows:
        try:
            r = requests.post(mes_url, json=json.loads(payload), timeout=3)
            if r.status_code in (200, 201, 409):
                conn.execute("DELETE FROM pending WHERE event_id=?", (event_id,))
            else:
                raise RuntimeError(f"status={r.status_code}")
        except Exception:
            if retry >= 12:
                conn.execute("DELETE FROM pending WHERE event_id=?", (event_id,))
            else:
                conn.execute(
                    "UPDATE pending SET retry=retry+1 WHERE event_id=?", (event_id,)
                )
    conn.commit()
    conn.close()

8.3 与 PLC 的边界

AI 只输出“OK / NG / REVIEW”信号，剔除动作由 PLC 或现有控制系统执行；
REVIEW 状态默认走“暂存区 + 人工复核”，不能直接剔除，否则过杀率会把产线堵死；
所有模型版本、阈值、判定结果必须留痕，方便客户投诉时追溯。

九、成本与收益：单工位 1 万元以内怎么做到

9.1 硬件清单（单工位，不含产线改造）

项目	规格	价格区间（元）
工业相机	500 万~1200 万像素，GigE	2000~6000
镜头	定焦低畸变	500~2000
光源 + 控制器	环形光/同轴光	800~3000
边缘盒子	Jetson Orin Nano 8G / RK3588	2500~5000
支架、线材、电控	-	500~1500
合计		6300~17500

如果只是做 PoC，用普通 USB 相机 + 一台带显卡的工控机即可，成本可以压到 5000 元以内。量产再换工业相机。

9.2 软件与云成本

项目	月成本（元）
边缘推理	0（本地）
云端复诊（厂内 1 张 4090）	电费约 300~500
对象存储（NG 图 30 天）	100~300
标注外包（每月 500 张）	500~1500
合计	900~2300

9.3 收益怎么算

以一个工位替代 1 名质检员为例：

人工成本：按 7000 元/月（含社保）计，一年 8.4 万元；
系统年化成本：硬件折旧 1.5 万 + 软件运维 1.2 万 + 标注 1 万 ≈ 3.7 万元；
直接回收期：约 6~9 个月。

但真正的收益往往不在人力，而在：

漏检导致的客诉和退货减少；
全检数据可追溯，拿下大客户订单；
工艺参数与缺陷率关联分析，降低废品率。

这三项很难精确量化，但通常是人力节省的 2~5 倍。给老板汇报时，建议把“人力节省”作为保底，“质量损失下降”作为弹性收益，分开列。

十、90 天落地排期

阶段	时间	交付物	验收标准
场景选定	第 1 周	缺陷字典、拍摄规范、验收指标	产线、质量、IT 三方签字
数据采集	第 2~3 周	良品 3000 张、缺陷每类 50~100 张	批次覆盖 3 个以上
基线模型	第 4 周	无监督异常检测 demo	能稳定检出明显缺陷
有监督模型	第 5~6 周	YOLO 模型 + 评估报告	固定过杀率下漏检率达标
边缘部署	第 7 周	TensorRT 服务 + 本地队列	节拍 < 1.2 s，连续 8 小时稳定
云端复诊	第 8 周	vLLM + 复核台	灰区样本 3 秒内出结论
MES 对接	第 9 周	追溯链路	100% 结果可查
试运行	第 10~12 周	日报、漂移监测、重训流程	连续 30 天可用率 > 99%
复制推广	第 13 周起	标准化部署包	新工位 3 天内完成适配

十一、踩坑清单：这 12 个问题一定会遇到

测试集泄漏：同一批产品随机切分，指标虚高 10 个点。必须按批次切。
只标 NG 不标 OK：模型没见过正常纹理，误检率爆炸。良品图至少是缺陷图的 10 倍。
阈值拍脑袋：0.5 不是万能阈值，必须按“固定过杀率”扫出来。
忽略镜头污染：运行两周后镜头上全是油污，模型把污点当缺陷。加每日清洁和清晰度监测。
光源频闪与相机不同步：图像一半亮一半暗，模型学到的是光照伪影。
缺陷定义模糊：三个人标出三种框，模型永远学不会。先统一标准再标注。
模型直接控 PLC：一次误判就可能撞机。AI 只给信号，控制权留在 PLC。
云端断网即停线：边缘必须能独立运行，云端只是增强。
新模具直接上线：产品外观变了，模型分布漂移。换型必须走首件确认和灰度。
只做检测不做追溯：客户投诉时说不清哪批有问题，项目价值大打折扣。
追求大模型上边缘：8B 模型跑在 Jetson 上，帧率个位数。边缘用小模型，大模型放云端。
没有回滚方案：新模型一上线就翻车，产线停摆。每次发布必须保留旧引擎和一键切换脚本。

十二、把方案产品化：从项目到可复制交付

如果你是想接制造业项目的团队，这套方案可以打包成三层产品：

诊断包（1~2 周，1~3 万元）：现场评估、缺陷字典、PoC demo、ROI 测算；
单工位交付（4~8 周，8~20 万元）：含硬件、模型、部署、MES 对接、培训；
年度运维（合同额 15%~20%/年）：重训、漂移监测、远程支持、备件。

交付时务必写清边界：

训练数据由甲方提供并保证权属；
新增缺陷类别按“新增类别”计费，不包含在免费运维里；
验收指标以双方签字的测试集为准；
产线改造、电气施工不在范围内。

十三、总结

制造业 AI 质检不是“训一个 SOTA 模型”的比赛，而是一套数据、模型、部署、闭环、验收的工程体系。把这五件事做扎实，100 张缺陷样本也能跑通一条产线；任何一环缺失，130 万张图也救不了现场。

记住三句话：

边缘保节拍，云端保长尾，人工保底线；
验收看漏检率和过杀率，不看 mAP；
上线不是终点，能每月自动变强才是。

参考资料

华为企业业务：江汽集团与华为“迈思特”CV 质检大模型发布，2026。
钦州日报：AI 如何破解制造业质检难题（小样本、端侧推理），2026-11。
NVIDIA 技术博客：合成缺陷图像在制造业外观检测中的应用，2026。
安徽省工信厅：《2026 年安徽省“AI+制造”典型应用场景案例集》，2026-11。
腾讯云开发者社区：工业质检的“99% 陷阱”：YOLO26 与多模态模型闭环架构，2026-06。
arXiv: SAEC: Scene-Aware Enhanced Edge-Cloud Collaborative Industrial Vision Inspection with Multimodal LLM, 2025。