☰
从零搭建AI工程闭环:图像缺陷检测的数据、训练与部署实战
2026/10/3 3:49:16 网站建设 项目流程

1. 先从“from scratch”说起:为什么我放弃了一站式AI平台

两年前我接到一个需求,要在一个ToB项目里做图像识别质检,用来代替人工检视产线上的一类缺陷。当时团队里所有人第一反应都是“找个现成的AI平台,上传样本、训练、导出模型,不就行了”。我也试了一圈,发现问题不是出在“能不能跑通”,而是出在“除了那一步跑通,其他什么都不可控”。

平台隐藏了太多细节:数据的存储格式和划分逻辑不透明,超参数被封装成黑盒,训练完导出的模型在本地推理时遇到版本冲突没人能解释清楚,最要命的是——模型一旦上线,你无法向客户解释“为什么这个缺陷会被漏检”,因为整个链路在你手里根本没有可追踪的中间产物。客户不关心你用了多先进的平台,他们只关心你有没有办法对结果负责。

所以当我把“ai-engineering-from-scratch”当作一个完整的、从零构建的工程路线来做时,我定的核心原则就一句话:不依赖任何端到端黑盒平台,所有中间环节——数据清洗、特征提取、训练、评估、导出、部署、自动重训——全部用自己的代码亲手搭一遍。

这不是为了炫技,而是为了获得三项能力:

  • 可解释性:每一个环节都有输入、输出、日志和中间产物,出了问题能逐段排查。
  • 可控性:换模型架构、改训练策略、调整数据分布,任何变量都可以单独验证。
  • 可扩展性:后续上新的AI能力时,只需复用已有的工程框架,不用重新被平台绑架。

这篇内容不是什么“从入门到放弃”的鸡汤,而是一条我实测跑通的完整路径。我以“图像缺陷分类”为例,从数据、训练、推理、部署、运维五个维度,把整条生态链拆给你看。全程用的都是开源工具和普通配置的设备,你不用有GPU集群,一台带8GB显存的消费级显卡就够。

2. 整体设计:一条最小可用、但五脏俱全的AI工程闭环

2.1 先给“AI工程”画个边界:它不是单纯写模型

很多人以为AI工程就是“训练一个模型”,这个理解至少缺失了百分之七十的工作量。实际进入生产环境的AI系统,必须包含数据管线、实验追踪、推理服务、监控告警、自动重训五个模块。少了任何一个,模型都只是实验室里的玩具。

我按这个边界定义项目范围:

模块作用说明
数据管线把原始图片变成可训练的干净样本清洗、标注、增强、划分
实验追踪记录每一次训练的参数和结果复现实验、比较模型
训练与评估训练出基线模型并计算真实性能指标精度、召回、误检率
推理服务把模型封装成可被业务调用的API处理单图、批量图
监控与自动重训检测线上数据漂移并自动触发新一轮训练保证模型不过时

这条闭环的好处,是你可以在任何一个环节“停下来”检查中间产物,而不是只有在最终结果上等一个遥遥无期的指标。我把这条闭环跑完整,只用了三个核心依赖:PyTorch做训练和推理,FastAPI做服务接口,PostgreSQL记录所有元数据。没有上什么重型分布式平台,因为起步阶段需要的不是规模,而是理解。

2.2 技术选型的底层逻辑:不追新,只求稳定复现

工具选型这件事,我踩过最大的坑是“盲目追新”。一开始我用了当时最新的深度学习框架版本,结果第三方库还没有适配,白白花了两个晚上解决环境冲突。后来我改了一个原则:所有核心依赖锁定版本,并且每次实验前记录完整的运行环境快照。

具体到这套项目,我的选型如下:

  • Python 3.10:生态兼容性最好,Torch、ONNX Runtime、OpenCV都有完整预编译包。
  • PyTorch 2.x:训练和部署一体,原生支持动态图调试,转ONNX也方便。
  • FastAPI:异步性能好,轻量,自带OpenAPI文档,写推理服务比Flask顺手得多。
  • PostgreSQL:用来存实验记录、模型版本、线上推理日志,一个库解决全部元数据管理。

这套组合不是追求最潮,而是追求出了问题我能在一小时之内定位到原因。我见过太多项目把重心放在换个新框架上,结果换来的不是效率而是深渊。稳定的技术栈加上严格的版本管理,才是“从零搭建”最该有的态度。

2.3 硬件要求:没有A100,一张消费级显卡足够起步

很多人看到“AI工程”四个字,第一反应是“需要很大的算力”。其实绝大多数真实业务场景里的模型,用不上大集群。我这个项目全程在一块NVIDIA RTX 3060(8GB显存)上完成,图像尺寸压到224x224,Batch Size控制在16。

一个很关键的计算经验放这里:你的显存总量,决定了一次前向和反向传播能塞下多少样本。以ResNet18为例,每批次16张224x224的RGB图片,前向+反向的显存占用大约在2.8GB到3.5GB之间,8GB显存绰绰有余。如果是ResNet50,同样批次显存会飙到5.5GB左右,这时候就要小心了。训练时先把Batch Size调小,逐步往上加,直到显存使用率接近90%为止。

内存方面,普通32GB机器完全够用,因为真正的“大块头”是硬盘上的数据集,而不是内存里的临时张量。存储建议至少留出50GB,一方面要存原始图片,另一方面要存训练过程中的checkpoint和日志。

3. 数据层:模型效果的上限,在这里被决定

3.1 数据收集与清洗:宁可少,不要脏

说实话,整个AI工程链路里最“不性感”但最重要的就是数据准备。我见过太多人花80%的时间调模型结构,却对数据的质量视而不见。结果就是训练集上漂亮得很,一到线上全是漏检和误检。

以我做的图像缺陷检测为例,原始数据来源有三块:产线摄像头抓拍的图片、历史缺陷归档图、以及人工用手机补拍的现场照片。这三类数据本身存在两个大问题:一是分辨率不统一,二是光照和角度差异极大。

清洗数据时我做了三件事:

  • 去重:用感知哈希算法(pHash)把所有图片两两比较,重复度超过阈值(比如汉明距离小于10)的直接删除。这一步处理掉了约12%的重复样本。
  • 过滤无效图:图片亮度方差过低的(全黑或全白图)、尺寸小于100x100的、文件格式异常的,全部剔除。
  • 统一归一化策略:所有图片统一缩放到224x224,不做拉伸,而是用等比缩放加灰色填充,避免物体形变。

这三步做完,原始数据从大约2.4万张降到1.9万张。数量少了,但质量高了,后续模型训练省了很多事。

3.2 标注策略:用“多级投票”对抗标注噪声

缺陷检测的标注,核心难点是“边界模糊”。一个缺陷到一定程度才算缺陷,那“一定程度”由谁定?如果只交给一个人,标准很容易漂移——人嘛,上午和下午的判断都不一定一致。

我采用的办法是多级投票归纳法:每个样本随机分给三个标注员独立标注,最终标签按少数服从多数决定。如果三个人给出三种不同结论,就进入人工仲裁队列。

这套机制成本高,但换来的是训练标签的稳定性。我还额外留了5%的样本作“置信度采样”,每两周抽一次,让标注员之间比对标定,防止标准随时间和疲劳漂移。

3.3 数据增强:给模型“制造难题”,而不是让它背答案

数据增强是提高泛化能力的最廉价手段。别小看这一步,一个好的增强策略可以让同样结构的模型在验证集上直接涨2到3个点。

我的增强管线如下:

import albumentations as A from albumentations.pytorch import ToTensorV2 train_transform = A.Compose([ A.Resize(224, 224), A.HorizontalFlip(p=0.5), A.RandomBrightnessContrast(p=0.3), A.HueSaturationValue(hue_shift_limit=10, sat_shift_limit=20, val_shift_limit=20, p=0.3), A.CoarseDropout(max_holes=8, max_height=16, max_width=16, fill_value=0, p=0.2), A.Normalize(mean=(0.485, 0.456, 0.406), std=(0.229, 0.224, 0.225)), ToTensorV2(), ])

有一个细节值得注意:增强在训练时随机生效,但在验证和推理阶段,只做Resize和Normalize,不加入任何随机操作。随机增强的目的是制造样本多样性,而不是让验证指标也变得随机。

另外,我强烈建议用Albumentations而不是TorchVision自带的Transforms,实测下来速度更快、增强种类更多,尤其是CoarseDropout这种对局部遮挡敏感的增强方式,对工业缺陷检测非常有效。它让模型学会“就算缺陷被部分遮挡,也能识别出来”,而不是死记图像像素模式。

3.4 数据集划分:别让“未来数据”混进训练集

数据集划分有一个经常被忽视的点:**如果一批图片是同一时间段同一相机拍的,它们之间的相似度往往很高,直接随机划分会导致验证集“泄漏”训练集的信息。**我用的是按“批次批次”划分:把同一批次采集的图片全部放到同一侧(训练、验证或测试),而不是按单张随机分。

划分比例我习惯是7:1.5:1.5,如果是小样本场景,建议加大验证集和测试集的比例,比如6:2:2。划分完一定要存一份划分清单(CSV里记录每张图的路径、标签、所属集合),这是复现实验的基石。

4. 训练环节:从构建模型到监控训练状态

4.1 模型架构选择:从ResNet起步,而不是从Transformer起步

我见过不少新手,一上来就上Vision Transformer(ViT)或Swin Transformer。也不是说不行,但前提是你有足够的数据(至少百万级)和足够的调参耐心。在这类千级、万级样本的工业场景里,ResNet系列依然是稳定、高效、可解释的首选。

在“From Scratch”的项目里,我选择ResNet18作为基线模型。原因有三:参数量小(约1120万)、在图像分类任务上表现稳定、推理速度快(CPU上单张预测低于30毫秒)。如果后续需要更高精度,最平滑的升级路径是换成ResNet50或RegNetY-4GF,不用改动太多代码。

迁移学习不是不能用,我的建议是:先用ImageNet预训练权重做初始化,把最后一层全连接换成自己的输出维度,然后冻结前面的卷积层,只训练分类头,跑5个epoch观察效果。如果结果已经不错,再渐进式解冻部分层做微调。这一招能大幅缩短训练时间,又能避免从头训练时收敛过慢的问题。

4.2 训练脚本:把实验配置与代码逻辑严格分离

写训练脚本最容易掉进“所有参数写死在一个文件里”的陷阱。一旦你开始跑多组实验,你会发现改一个学习率都要翻半天代码。我的做法是:用Dataclass定义全量配置,用命令行参数覆盖默认值。

from dataclasses import dataclass, field from typing import Tuple @dataclass class TrainConfig: data_root: str = "./data/images" train_csv: str = "./data/train.csv" val_csv: str = "./data/val.csv" model_name: str = "resnet18" image_size: int = 224 batch_size: int = 16 epochs: int = 30 lr: float = 3e-4 weight_decay: float = 1e-4 warmup_epochs: int = 2 num_workers: int = 4 seed: int = 42 output_dir: str = "./runs" device: str = "cuda"

这样设计的好处是,每一组实验都可以用一行命令生成独立目录,目录里自动保存配置文件、日志、checkpoint、评估报告。后面比较实验结果的时候,只需要看目录名和配置快照,完全不会混淆。

另外一个非常实用的小习惯:固定随机种子,包括PyTorch、NumPy、Python内置random,甚至把DataLoader的worker随机种子也固定。否则同一份代码跑两次,结果可能差零点几个点,你会分不清是模型变化还是随机波动。

import random import numpy as np import torch def set_seed(seed: int = 42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False

4.3 训练过程中的监控:损失值曲线是“体检报告”

训练的时候不要只看最终精度,一定要盯着每个epoch的训练损失和验证损失。这两种损失的曲线关系,能直接告诉你模型的状态:

  • 两者都在下降,说明训练健康。
  • 训练损失降、验证损失不降或上升,典型的过拟合,马上停或加正则。
  • 两者都不降,学习率可能过大或过小,先调学习率。
  • 验证损失出现锯齿状剧烈波动,Batch Size太小或数据里存在噪声标签。

我在训练脚本里每5个epoch输出一张损失曲线图,自动保存到实验目录。这样即使人不在电脑前,回头打开图片也能一眼看出整个训练过程是否健康。不要傻傻只盯着terminal里滚动的数字,图表化的趋势才是真正的信号。

5. 评估与分析:模型“看起来准”和“真正可用”是两码事

5.1 别只用Accuracy,你的业务场景需要更细的尺子

在缺陷检测这类正负样本严重不平衡的场景里,Accuracy是个“骗子指标”。比如99%的样本都是正常品,模型什么都不学直接全预测正常品,Accuracy也有99%,看起来完美,实际上什么都干不了。

我同时计算Precision(查准率)、Recall(查全率)和F1-Score,并且额外关注漏检率。在质检场景里,漏掉一个缺陷的代价远大于多查几次正常品,所以我对Recall的要求是尽量逼近99%,即使Precision会相应降低。你需要在业务侧权衡这个平衡点,而不是看一个平均指标就草草上线。

from sklearn.metrics import classification_report, confusion_matrix report = classification_report(y_true, y_pred, target_names=class_names, digits=4) print(report) print(confusion_matrix(y_true, y_pred))

classification_report的好处是每个类别的Precision、Recall、F1分别输出,你能一眼看到哪个类是最弱的。我实测下来,最容易互相混淆的缺陷类别是“划痕”和“细微裂纹”,这两个类的特征在224x224分辨率下高度重叠,后来我把这两类图片单独提出来,先用超分模型增强分辨率,再重新训练,F1直接从0.82涨到0.91。

5.2 错误分析:不要只盯着错误率数字,去看错误图长什么样

评估报告完成后,我把预测错误的图片按“真实类别+预测类别”分成不同组合,每类抽20张拼成网格图,人眼直接审。

这个动作看似原始,但价值极大。我曾经发现大量被误判为“正常”的缺陷图,其实是图片边缘存在极小的暗角阴影,肉眼都很难发现。这个信息直接指导我用Canny边缘检测做预处理,在送入模型前把边缘区域的低对比度伪影剔除,整体漏检率下降了40%。

我的建议是:每一次评估,都固化生成一份“错误分析报告”,包括错误样本路径、预测概率、真实标签、人工备注。这些记录不仅帮你发现模型的问题,也是后续和客户解释模型行为时的最重要素材。

6. 推理与部署:把模型从一个文件变成一个服务

6.1 模型导出:PyTorch模型不能直接上生产环境

训练完成得到的是.pth或.pt权重文件,它依赖PyTorch的运行时,直接放到线上推理环境,一是速度不够理想,二是版本兼容性问题太多。我统一用ONNX Runtime做推理引擎,把训练好的模型导出为ONNX格式。

导出ONNX这一步有几个关键参数:

import torch dummy_input = torch.randn(1, 3, 224, 224, device="cuda") model.eval() torch.onnx.export( model, dummy_input, "model.onnx", input_names=["input"], output_names=["output"], dynamic_axes={"input": {0: "batch_size"}, "output": {0: "batch_size"}}, opset_version=17, )
  • dynamic_axes:允许batch维度可变,这样同一个模型既能单张推理也能批量推理。
  • opset_version:不是越新越好,要看你所用的ONNX Runtime版本支持范围,选17是保险选择。
  • 导出后一定做一次精度一致性校验:用同样的输入分别跑PyTorch模型和ONNX模型,预测结果误差不得超过1e-4,否则就要排查算子兼容性。

6.2 推理服务设计:单图接口 + 批量接口,一个都不能少

我把推理服务设计成两个接口:单图同步接口和批量异步接口。前者用于实时检测场景(比如产线相机拍一张判断一张),后者用于离线批量扫描(比如历史图片全量重检)。

用FastAPI实现非常简洁,这里给出核心骨架:

import time import numpy as np import onnxruntime as ort from fastapi import FastAPI, UploadFile, File from PIL import Image import io app = FastAPI() session = ort.InferenceSession("model.onnx", providers=["CUDAExecutionProvider", "CPUExecutionProvider"]) @app.post("/predict") async def predict(file: UploadFile = File(...)): image_bytes = await file.read() image = Image.open(io.BytesIO(image_bytes)).convert("RGB") image = image.resize((224, 224)) input_tensor = np.array(image).astype(np.float32) / 255.0 input_tensor = (input_tensor - np.array([0.485, 0.456, 0.406])) / np.array([0.229, 0.224, 0.225]) input_tensor = np.transpose(input_tensor, (2, 0, 1))[None, ...] start = time.perf_counter() outputs = session.run(["output"], {"input": input_tensor})[0] latency_ms = (time.perf_counter() - start) * 1000 pred_idx = int(np.argmax(outputs[0])) confidence = float(np.max(outputs[0])) return { "prediction": class_names[pred_idx], "confidence": round(confidence, 4), "latency_ms": round(latency_ms, 2), }

别忘了在服务启动时加载模型,并且设置好providers顺序。把CUDA放在前面,如果没有GPU会自动回退到CPU,这一点在混合部署环境里非常实用。

6.3 并发与性能:一次压测告诉我,“单张快”不等于“并发快”

我第一次上线这个接口时,单张图片推理只要20毫秒,心想这速度还不是轻轻松松。结果压测一上来,50个并发直接把请求排队到几秒,原因很简单:GPU推理是串行的,你一次只能推一个batch,再加上Python GIL锁的影响,同步接口在并发场景下直接成为瓶颈。

解决办法分两步。第一步:对推理请求做排队,而不是用线程硬扛。FastAPI的async + asyncio.Queue可以自然地把请求排队,队列长度设个上限(比如200),超出直接返回“暂满稍后重试”。

第二步:把单张请求合并成batch。无论客户端怎么来,服务端都先把请求放进缓冲区,每隔10毫秒或者攒满8张图,就触发一次真正的批量推理。这一步优化之后,同样50并发下,平均响应时间从几秒降到120毫秒。这个方案业界叫“动态批处理”,用代码实现也就几十行,但收益巨大。

import asyncio from collections import deque request_buffer = deque() buffer_lock = asyncio.Lock() BATCH_SIZE = 8 BATCH_TIMEOUT = 0.01 # 10ms async def batch_worker(): while True: items = [] async with buffer_lock: while len(request_buffer) > 0 and len(items) < BATCH_SIZE: items.append(request_buffer.popleft()) if items: inputs = np.concatenate([item["input"] for item in items], axis=0) outputs = session.run(["output"], {"input": inputs})[0] for item, output in zip(items, outputs): item["future"].set_result(output) await asyncio.sleep(BATCH_TIMEOUT)

这个worker常驻后台,只要有请求进来就往缓冲区放,一旦满足批量条件就执行推理。如果一直没有新的请求,10毫秒超时后也会把缓冲区内已有的请求处理掉,不会无限等待。

6.4 容器化部署:让“在我机器上是好的”成为历史

模型版本、Python依赖、CUDA环境、系统库……任何一个不一致都能让线上服务原地爆炸。我把推理服务做成了Docker镜像,镜像内固定Python版本、ONNX Runtime版本、系统时区、工作目录,实现真正的一次构建处处运行。

Dockerfile核心部分如下:

FROM nvidia/cuda:12.1.1-runtime-ubuntu22.04 RUN apt-get update && apt-get install -y --no-install-recommends \ python3.10 python3-pip libgl1 libglib2.0-0 && \ rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements.txt . RUN pip3 install --no-cache-dir -r requirements.txt COPY app/ ./app/ COPY model.onnx ./model.onnx EXPOSE 8000 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

这里有两个细节值得单独说。一是libgl1 libglib2.0-0,这是OpenCV和Pillow处理图像时的系统级依赖,漏掉的话总会报一些莫名其妙的库缺失错误。二是--workers 2,如果推理是CPU执行,多进程能榨干多核,但如果GPU推理且每进程都加载一份模型到显存,显存不够就反而会OOM,需要根据模型大小和显存容量权衡。

7. 上线后的运维:模型部署完,工作才刚刚开始

7.1 日志与监控:要能看到每一次推理的“体检单”

部署之后,我把所有推理请求的元数据写入PostgreSQL,每一条包含:图片ID、预测类别、置信度、推理耗时、模型版本、输入图片的哈希值。这个看似啰嗦的日志表,是后面一切排查的基础。

除了业务日志,我还额外采集三个系统指标:GPU利用率、显存占用、推理队列长度。队列长度超过阈值说明并发超了预期,GPU利用率长期低于30%说明动态批处理还没有发挥最优效果,显存快满则要考虑换小模型或降batch。

实时告警我用了简单的规则:推理队列超过200持续10秒,告警;预测置信度整体均值连续下降,告警;单接口的P99耗时超过400毫秒,告警。告警阈值不要拍脑袋,前两周观察线上数据再定。

7.2 数据漂移与自动重训:让模型拥有“自我更新”能力

模型上线一段时间后最隐蔽的敌人是数据漂移——产线换了新批次材料、摄像头角度微调、环境光照改变,都会让线上数据的分布逐渐偏离训练集。如果模型不会自我更新,精度会慢慢衰退,而没人能确切说出哪天开始变差的。

我的方案是:每三天做一次线上数据和训练集的分布对比,用简单的特征向量(比如直方图均值和方差)计算距离。当距离超过阈值时,自动触发一次增量训练。增量训练不是从头跑,而是在最近的历史数据上微调10个epoch,然后把新模型送入A/B测试通道。

这个“自愈”机制必须配合人工审核闸门:自动训练出的模型必须先跑一遍离线测试集,测试集上的关键指标(比如漏检率)必须不低于当前线上模型,否则不发布。这就形成了一条完整的闭环:监控发现漂移,自动训练候选模型,审核通过后自动切换。

7.3 模型版本管理:在AI工程里,模型就是代码

我一开始把模型文件当普通文件,扔在服务器某个目录里,谁要谁去拷。后来一次紧急回滚让我痛定思痛:新版本模型上线后效果反而更差,想切回旧版,结果旧版权重被覆盖了,只能重新训练。这种事故一次就够了。

现在我的规则很简单:每个模型文件和一个“模型版本记录”绑定,存进PostgreSQL。记录内容包括模型名称、版本号、训练数据范围、训练时间、验证集指标、责任人、上线时间。每个版本模型在文件系统里单独建目录,目录名就是版本号,删除模型文件必须通过正式流程而不是“手动清理空间”。

到这里,从零搭建AI工程闭环的每个环节我都带你看了一遍。最后说点个人体会:这套体系最大的价值不是“模型准确率多高”,而是它让团队里每一个人都能在任何时候回答三个问题——“数据是怎么来的?”“模型是怎么训的?”“线上发生了什么?”。做到这三点,你手里的AI才真正从“写好玩的”变成了“能负责的”。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询