制造业AI网关落地实战：MAI Gateway边缘部署与模型量化

1. 项目缘起：为什么制造业需要一台“AI网关”

1.1 从一次车间里的尴尬说起

去年秋天，我被拉到一个做精密注塑的朋友厂里帮忙看产线质检的问题。他们的场景很典型：注塑机旁边架着一台工业相机，拍产品表面，后端跑一个视觉检测模型，判断有没有划痕、缺料、飞边。听起来不复杂，但实际跑起来问题一堆。车间网络时断时续，模型推理要调用云端接口，一张图来回几百毫秒，产线节拍根本等不起；更麻烦的是，有些订单的图纸和工艺参数属于客户机密，数据出不了厂区，云端方案直接被否掉。

那天下午我蹲在车间里，看着一台老旧的工控机风扇呼呼转，屏幕上检测结果一卡一卡地跳，心里就在想：制造业要的AI，不是参数多大的模型，而是一个能塞进车间角落、断网也能跑、数据不出门、还能跟现有PLC和MES说上话的“中间层”。这个中间层，就是后来我们反复折腾的AI网关，而MAI Gateway是我在几个项目里实际落地过的一套方案。

所谓AI网关，你可以把它理解成车间里的“翻译官+调度员”。往上，它对接各种AI模型——视觉的、时序的、文本的；往下，它对接产线上的设备、传感器、工控系统。它不生产数据，也不训练模型，它负责把模型的能力“翻译”成产线能用的信号，同时把产线的数据“翻译”成模型能吃的格式。制造业上AI，卡点往往不在算法本身，而在这一层“最后一公里”的对接上。

1.2 制造业AI落地的三个硬约束

我在不同工厂跑下来，发现制造业对AI网关的需求，跟互联网公司完全不是一回事。互联网讲弹性、讲并发、讲秒级扩容，制造业讲的是另外三个词：稳定、离线、实时。

先说稳定。车间环境不是恒温机房，夏天四十度、冬天零下、粉尘、震动、电磁干扰，工控机死机一次可能整条线停半小时。所以网关的第一要求是“别崩”，而不是“跑得快”。我见过太多方案在实验室跑得飞起，一进车间三天两头重启。

再说离线。很多工厂的内网是物理隔离的，尤其是涉及军工、汽车零部件、医疗器械的产线，数据根本不允许出局域网。这意味着网关必须能在完全断网的情况下完成模型推理、结果判定、日志记录的全流程。你不能指望一个每次都要联网校验授权的方案。

最后是实时。产线节拍是按秒甚至毫秒算的，注塑机一个周期可能就十几秒，检测必须在几百毫秒内出结果，否则产品都流走了你才报警，毫无意义。这就要求网关的推理链路足够短，不能有太多网络跳转和序列化开销。

这三个约束叠加起来，就决定了制造业的AI网关不能照搬互联网那套微服务架构，得走边缘部署、本地推理、轻量通信的路子。MAI Gateway 在我手里的几次落地，基本就是围绕这三点做取舍。

1.3 这篇文章适合谁看

如果你是在工厂里推AI落地的工程师、项目经理，或者你是做AI应用想切入制造业的开发者，这篇文章应该对你有用。我不会讲太多虚的架构图，重点放在实际怎么选、怎么装、怎么调、怎么排错上。里面提到的参数、命令、配置，都是我在真实项目里跑过的，你可以直接抄作业，但要注意每家工厂的现场条件不同，照搬之前先做小范围验证。

我也会把踩过的坑摊开讲，比如模型量化后精度掉了怎么办、工控机没有GPU怎么撑推理、断网后日志怎么补传。这些东西文档里不会写，但现场一定会遇到。

2. 方案选型：MAI Gateway 为什么这么搭

2.1 整体架构的取舍逻辑

先把这个网关的骨架说清楚。MAI Gateway 在我落地的版本里，核心就四层：接入层、推理层、规则层、输出层。接入层负责收数据，推理层负责跑模型，规则层负责把模型输出转成业务判定，输出层负责把结果发给PLC、MES或者报警灯。

为什么这么分？因为制造业的场景里，模型输出和业务判定往往是两回事。举个例子，视觉模型输出的是“缺陷置信度0.87”，但产线要的是“这个产品是NG，触发剔除气缸”。中间这个转换逻辑，不同客户、不同产品、不同批次都可能不一样，如果把它写死在模型里，改一次就要重新训练，成本太高。所以我把规则层单独抽出来，用配置文件描述“置信度大于多少算NG、连续几个NG触发停线”，现场工程师自己就能改，不用等算法团队。

接入层我选的是消息队列+HTTP双通道。消息队列走MQTT，适合传感器这种高频小数据；HTTP走RESTful，适合相机这种一次传一张图的场景。为什么不统一用gRPC？因为很多工厂现有的MES和SCADA系统只认HTTP，你强行上gRPC，对接成本反而高。这个取舍很现实：在制造业，兼容既有系统比技术先进更重要。

推理层是核心，我后面单独讲。输出层我留了三种出口：写数据库、发MQTT、调HTTP接口。写数据库是为了追溯，发MQTT是为了实时控制，调HTTP是为了对接老系统。三种出口可以同时开，互不干扰。

2.2 推理引擎的选择：ONNX Runtime 还是 TensorRT

这是选型时争论最多的点。我最后在大多数项目里选了ONNX Runtime，只在有NVIDIA显卡且对延迟极度敏感的场景才上TensorRT。原因不复杂：ONNX Runtime 的跨平台兼容性好太多，x86工控机、ARM边缘盒子、甚至国产化平台都能跑，而TensorRT绑死NVIDIA生态，工厂里大部分工控机根本没有独立显卡。

具体到性能，我实测过一个YOLOv5s的模型，输入640x640，在Intel i5-8500的工控机上，ONNX Runtime（CPU模式）单张推理大概120到150毫秒，开多线程能压到80毫秒左右。这个速度对于注塑、冲压这种节拍在秒级的场景完全够用。如果是更快的产线，比如电子元件贴装，节拍在百毫秒级，那就得上GPU或者用更小的模型。

这里有个经验：不要一上来就追求极致延迟，先算清楚产线节拍。我见过一个项目，客户非要上TensorRT，结果工控机没有显卡，又加了一张Tesla T4，成本翻了三倍，最后发现其实CPU推理就够，因为他们的检测工位本来就有几秒的缓冲。选型前先拿秒表去产线蹲半天，比看任何benchmark都准。

2.3 模型格式与量化策略

模型从训练到部署，中间要过好几道转换。我的标准流程是：PyTorch训练 → 导出ONNX → 量化 → 部署。量化这一步最容易出问题，得重点说。

量化分两种：动态量化和静态量化。动态量化简单，一行代码就能转，但精度损失相对大；静态量化需要校准数据集，精度保持更好，但流程麻烦。我在制造业项目里一般用静态量化，因为工业质检对精度敏感，漏检一个缺陷可能整批产品报废。

校准数据集怎么选？我的做法是从产线实际拍到的图里，按缺陷类型分层抽样，每类至少200张，正常样本也要有。校准过程就是让模型在这些图上跑一遍，统计激活值的分布，确定量化参数。这个过程在ONNX Runtime里用quantize_static接口就能做。

量化后精度掉了怎么办？先别急着放弃。我遇到过量化后mAP掉了3个点的情况，排查下来是校准集里缺陷样本太少，模型对缺陷特征的激活分布没统计准。补了500张缺陷图重新校准，精度就回来了。如果补数据还不行，那就对量化敏感的层做混合精度，保持这些层用FP32，其他层用INT8，ONNX Runtime支持这种混合模式。

注意：量化后的模型一定要在验证集上重新跑一遍完整指标，不能只看单张图的输出。我踩过坑，单张图看着没问题，批量跑起来发现某一类缺陷的召回率掉了十几个点。

2.4 边缘硬件的选型清单

硬件这块我列个实际用过的清单，都是跑过MAI Gateway的配置：

选硬件的时候，除了算力，还要看接口。车间里相机多是GigE或者USB3.0，PLC多是RS485或者以太网，网关至少要有两个网口，一个接相机网段，一个接工控网段，物理隔离更稳。我吃过亏，相机和PLC走同一个交换机，相机一传大图就把PLC的通信挤掉了，后来加了网口隔离才解决。

3. 实操落地：从零把网关跑起来

3.1 环境准备与依赖安装

我以一台Ubuntu 20.04的工控机为例，把安装过程走一遍。为什么选Ubuntu而不是Windows？因为车间里工控机装Windows的居多，但Windows的Docker支持和后台服务管理不如Linux稳，而且Linux可以裁掉图形界面，省资源。如果客户坚持用Windows，那就用WSL2或者直接跑Python服务，但稳定性会打折扣。

先装基础依赖：

sudo apt update sudo apt install -y python3.8 python3-pip python3-venv \ libgl1 libglib2.0-0 libsm6 libxext6 libxrender-dev \ mosquitto mosquitto-clients sqlite3

这里解释几个包的作用。libgl1这些是OpenCV的依赖，不装的话import cv2会报错；mosquitto是MQTT broker，网关内部通信用；sqlite3是本地数据库，用来存检测记录，断网时数据先落本地，联网后再补传。

然后建虚拟环境装Python包：

python3 -m venv /opt/mai-gateway/venv source /opt/mai-gateway/venv/bin/activate pip install onnxruntime==1.15.1 opencv-python-headless==4.8.0.74 \ numpy==1.24.3 paho-mqtt==1.6.1 requests==2.31.0 \ fastapi==0.103.1 uvicorn==0.23.2

版本我锁死了，因为ONNX Runtime 1.16之后有些API变了，跟老代码不兼容。opencv-python-headless是不带GUI的版本，车间里不需要弹窗显示，省内存。

提示：如果工控机是ARM架构，onnxruntime要换成onnxruntime的ARM版本，pip源里可能没有，需要从源码编译或者用厂商提供的wheel包。这一步最容易卡住，建议提前准备好。

3.2 模型转换与量化实操

假设你有一个训练好的PyTorch模型defect_model.pt，先导出ONNX：

import torch import torch.onnx model = torch.load('defect_model.pt', map_location='cpu') model.eval() dummy_input = torch.randn(1, 3, 640, 640) torch.onnx.export( model, dummy_input, 'defect_model.onnx', input_names=['input'], output_names=['output'], dynamic_axes={'input': {0: 'batch'}, 'output': {0: 'batch'}}, opset_version=11 )

opset_version选11是个经验值，兼容性好，ONNX Runtime各版本都支持。dynamic_axes把batch维度设成动态，方便后面调批量大小。

接下来量化。先准备校准数据，放在calib_images/目录下，然后跑：

from onnxruntime.quantization import quantize_static, CalibrationDataReader import cv2 import numpy as np import os class CalibReader(CalibrationDataReader): def __init__(self, image_dir): self.image_list = [os.path.join(image_dir, f) for f in os.listdir(image_dir) if f.endswith('.jpg')] self.index = 0 def get_next(self): if self.index >= len(self.image_list): return None img = cv2.imread(self.image_list[self.index]) img = cv2.resize(img, (640, 640)) img = img[:, :, ::-1].transpose(2, 0, 1).astype(np.float32) / 255.0 img = np.expand_dims(img, axis=0) self.index += 1 return {'input': img} reader = CalibReader('calib_images/') quantize_static( 'defect_model.onnx', 'defect_model_int8.onnx', reader, quant_format=QuantFormat.QDQ, per_channel=True )

quant_format选QDQ，这是量化感知训练友好的格式，精度保持更好。per_channel=True是按通道量化，比按张量量化精度高，但模型会大一点点，可以接受。

量化完一定要验证。我写了个小脚本，拿验证集跑一遍，对比量化前后的输出差异：

import onnxruntime as ort import numpy as np def run_model(model_path, input_data): sess = ort.InferenceSession(model_path) return sess.run(None, {'input': input_data})[0] # 拿同一张图分别跑FP32和INT8模型 test_img = np.random.randn(1, 3, 640, 640).astype(np.float32) out_fp32 = run_model('defect_model.onnx', test_img) out_int8 = run_model('defect_model_int8.onnx', test_img) diff = np.abs(out_fp32 - out_int8).max() print(f'最大输出差异: {diff}')

差异在0.05以内一般可以接受，超过0.1就要检查校准集或者考虑混合精度。

3.3 网关服务配置与启动

MAI Gateway 的主服务我用FastAPI写，配置文件用YAML。先看配置结构：

gateway: name: "injection-line-01" listen_port: 8080 mqtt_broker: "localhost" mqtt_port: 1883 db_path: "/opt/mai-gateway/data/records.db" models: - name: "defect_detector" path: "/opt/mai-gateway/models/defect_model_int8.onnx" input_size: [640, 640] input_name: "input" output_name: "output" confidence_threshold: 0.5 nms_threshold: 0.45 rules: - name: "ng_trigger" model: "defect_detector" condition: "max_confidence > 0.7" action: "mqtt_publish" topic: "line/ng" payload: '{"status": "NG", "confidence": {confidence}}' outputs: - type: "sqlite" enabled: true - type: "mqtt" enabled: true - type: "http" enabled: false url: "http://mes-server/api/quality"

这个配置里，rules段是核心。condition支持简单的表达式，现场工程师改阈值不用动代码。action支持mqtt_publish、http_post、db_insert三种，够用了。

启动服务：

cd /opt/mai-gateway source venv/bin/activate uvicorn main:app --host 0.0.0.0 --port 8080 --workers 1

--workers 1是有意为之。车间工控机核数少，开多worker反而抢CPU，推理延迟更不稳定。单worker配合异步IO，实测吞吐更稳。

3.4 与PLC和MES的对接细节

对接PLC是最容易出问题的环节。我一般用Modbus TCP，因为大部分PLC都支持，配置简单。Python里用pymodbus：

from pymodbus.client import ModbusTcpClient client = ModbusTcpClient('192.168.1.10', port=502) client.connect() # 写一个线圈，触发剔除气缸 client.write_coil(0, True)

这里有个坑：Modbus的地址偏移。不同PLC厂商对地址的定义不一样，有的从0开始，有的从1开始，有的把线圈和寄存器分开编址。我踩过坑，按文档写的地址写不进去，后来拿Modbus调试工具一个个试才找到正确偏移。建议对接前先用调试工具确认地址映射。

对接MES一般走HTTP，把检测结果POST过去：

import requests def report_to_mes(record): try: resp = requests.post( 'http://mes-server/api/quality', json=record, timeout=2 ) return resp.status_code == 200 except requests.exceptions.RequestException: # 网络不通，先存本地，后面补传 save_to_local_queue(record) return False

超时设2秒，因为MES服务器可能响应慢，不能让它拖住网关主流程。失败就落本地队列，后台起个线程定时补传。

4. 现场排错：那些文档里不会写的问题

4.1 推理结果不稳定的排查思路

现场最常遇到的问题是“同一张图，有时候检测出来，有时候检测不出来”。这种问题最头疼，因为不是必现。我的排查顺序是这样的：

先看输入预处理。相机拍的图，每次的曝光、白平衡可能略有差异，如果预处理里做了归一化，但参数写死了，遇到光照变化就会飘。我一般会在预处理里加自适应直方图均衡，或者让相机锁定曝光和白平衡。

再看模型输入尺寸。如果相机分辨率是1280x960，你resize到640x640，长宽比变了，缺陷可能被压扁。正确做法是保持长宽比resize，然后padding。这个细节很多教程不讲，但实际影响很大。

最后看推理引擎的线程数。ONNX Runtime默认会用所有CPU核心，如果网关同时还在跑其他服务，线程争抢会导致推理结果不稳定。我一般显式设置：

options = ort.SessionOptions() options.intra_op_num_threads = 2 options.inter_op_num_threads = 1 sess = ort.InferenceSession('model.onnx', options)

把线程数限制住，虽然单次推理慢一点，但稳定性好很多。

4.2 断网后的数据补传机制

前面提到断网时数据落本地，联网后补传。这个机制我踩过坑，最早的设计是简单队列，先进先出，结果发现补传时如果数据量大，会把MES打挂。后来改成批量补传+限速：

def sync_pending_records(batch_size=50, interval=1.0): while True: records = query_pending_records(limit=batch_size) if not records: time.sleep(5) continue for record in records: if report_to_mes(record): mark_as_synced(record['id']) else: break time.sleep(interval)

每批50条，批间停1秒，给MES喘息时间。mark_as_synced用数据库事务，保证不会重复补传。

注意：本地数据库要定期清理，不然工控机硬盘很快满。我一般保留最近30天的记录，更早的归档到外部存储或者直接删除。

4.3 常见问题速查表

4.4 几个独家避坑技巧

第一个，工控机BIOS里关掉节能模式。很多工控机默认开了C-State，CPU会降频，推理延迟忽高忽低。进BIOS把C-State和SpeedStep关了，延迟立刻稳。

第二个，给网关进程设CPU亲和性。用taskset把网关绑到固定核心，避免跟其他进程抢：

taskset -c 2,3 uvicorn main:app --host 0.0.0.0 --port 8080

第三个，日志分级。车间里日志写太详细会拖慢IO，我一般生产环境只记WARNING以上，调试时再开DEBUG。日志文件用logrotate每天切，保留7天。

第四个，模型文件加校验。部署前算个MD5，启动时校验，防止文件传输出错导致模型加载了半截。这个坑我踩过，模型文件不完整，推理输出全是乱码，排查了半天。

5. 效果验证与扩展方向

5.1 实际产线的运行数据

在一个注塑质检项目里，这套网关跑了三个月，我拉了些数据。产线节拍12秒，网关单次推理平均95毫秒，加上预处理和后处理，端到端大概180毫秒，完全跟得上。检测准确率方面，量化后的INT8模型对比FP32，mAP从0.91掉到0.89，漏检率从2.1%升到2.8%，在可接受范围内。客户最满意的是断网表现，有一次厂区网络割接断了4小时，网关照常检测，数据全存本地，网络恢复后自动补传，一条没丢。

稳定性上，三个月里网关重启过两次，都是因为工控机硬盘故障，跟软件本身无关。CPU占用率平时在40%左右，推理峰值到70%，内存稳定在2G以内。

5.2 还能往哪些方向扩展

这套网关的骨架是通用的，换模型就能换场景。我后来在另一个项目里，把视觉模型换成时序模型，接振动传感器做设备预测性维护，配置改改就能跑。规则层也支持更复杂的逻辑，比如连续N个NG触发停线、按时间段切换阈值。

再往深了做，可以加模型热更新。现在换模型要重启服务，如果做成双缓冲，新模型加载好再切换，就能做到不停机更新。这个我还在折腾，主要是内存管理麻烦，工控机内存有限，两个模型同时驻留容易OOM。

另一个方向是多网关协同。一条产线多个工位，每个工位一个网关，网关之间通过MQTT同步状态，比如上游检测到NG，下游网关提前调整参数。这个在汽车零部件产线很有用，但通信协议和状态一致性要设计好，不然容易乱。

我个人在实际操作中的体会是，制造业AI落地，技术只占三成，七成是对现场的理解。你得知道产线节拍怎么算、PLC地址怎么映射、工人怎么操作、网络怎么布，这些细节决定了方案能不能真正跑起来。MAI Gateway 这套东西不复杂，但每一个配置项背后都是现场踩出来的经验。你要是准备在厂里推类似方案，建议先拿一条线做试点，跑稳了再铺开，别一上来就全线推广。