Skip to content
Published on

AI制造与工业4.0：预测性维护、数字孪生、质量检测AI全览

🇰🇷KO🇺🇸EN🇯🇵JA🇨🇳ZH
필사 모드
分享
Authors

工业4.0与AI的融合

工业4.0（Industry 4.0）指的是网络物理系统（CPS）、IIoT、云计算与AI相结合的第四次工业革命。制造现场的所有设备与流程都被数字化，决策则基于实时数据完成。

核心技术栈如下：

  • CPS（Cyber-Physical System，网络物理系统）：连接物理世界与数字世界的集成系统
  • IIoT（Industrial Internet of Things，工业物联网）：由工业传感器、执行器、控制系统组成的网络
  • OPC-UA（Open Platform Communications Unified Architecture）：制造设备之间的标准数据交换协议
  • MQTT：轻量级消息代理协议，针对IoT边缘设备做了优化
  • 数字孪生：物理资产的实时虚拟复制体

OPC-UA Python客户端实现

OPC-UA是制造现场PLC、SCADA、MES之间数据集成的标准。下面用Python实现一个从OPC-UA服务器采集传感器数据的客户端。

from opcua import Client
import pandas as pd
import time
from datetime import datetime

class ManufacturingDataCollector:
    def __init__(self, server_url: str):
        self.client = Client(server_url)
        self.data_buffer = []

    def connect(self):
        self.client.connect()
        print(f"OPC-UA 服务器连接完成: {self.client.get_endpoints()}")

    def read_sensor_nodes(self, node_ids: list) -> dict:
        readings = {}
        for node_id in node_ids:
            node = self.client.get_node(node_id)
            value = node.get_value()
            readings[node_id] = {
                "value": value,
                "timestamp": datetime.utcnow().isoformat()
            }
        return readings

    def collect_stream(self, node_ids: list, interval_sec: float = 1.0):
        """实时流式采集"""
        while True:
            readings = self.read_sensor_nodes(node_ids)
            self.data_buffer.append(readings)
            time.sleep(interval_sec)

    def to_dataframe(self) -> pd.DataFrame:
        rows = []
        for snapshot in self.data_buffer:
            row = {"timestamp": list(snapshot.values())[0]["timestamp"]}
            for node_id, data in snapshot.items():
                row[node_id] = data["value"]
            rows.append(row)
        return pd.DataFrame(rows)

    def disconnect(self):
        self.client.disconnect()

# 使用示例
collector = ManufacturingDataCollector("opc.tcp://factory-plc:4840/")
collector.connect()

# CNC机床传感器节点ID
sensor_nodes = [
    "ns=2;i=1001",  # 主轴转速 (RPM)
    "ns=2;i=1002",  # 振动 (g)
    "ns=2;i=1003",  # 温度 (°C)
    "ns=2;i=1004",  # 电流 (A)
]

collector.collect_stream(sensor_nodes, interval_sec=0.5)
df = collector.to_dataframe()
df.to_parquet("sensor_data.parquet")

预测性维护（Predictive Maintenance）

预测性维护是在设备发生故障之前检测异常征兆、从而进行有计划维护的策略。相比传统的事后维修（Corrective Maintenance）或定期维护（Preventive Maintenance），它能把停机时间和成本降到最低。

异常检测 vs 故障分类

预测性维护流水线由两个阶段构成。

  1. 异常检测（Anomaly Detection）：检测偏离正常模式的数据，可以在无标签的无监督方式下完成
  2. 故障分类（Fault Classification）：在检测到异常之后，对故障类型进行分类，需要有标签的故障数据

在实际工厂中，故障数据极其稀少，因此现实的做法是先构建第一阶段（异常检测），再用后续收集到的异常数据训练第二阶段的分类器。

用Isolation Forest做传感器异常检测

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt

def train_anomaly_detector(df: pd.DataFrame, feature_cols: list,
                            contamination: float = 0.05):
    """
    基于Isolation Forest训练异常检测器
    contamination: 预估异常比例 (0.05 = 5%)
    """
    X = df[feature_cols].values
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    model = IsolationForest(
        n_estimators=200,
        contamination=contamination,
        max_samples="auto",
        random_state=42,
        n_jobs=-1
    )
    model.fit(X_scaled)

    # 异常分数：越低越异常（负数）
    scores = model.decision_function(X_scaled)
    predictions = model.predict(X_scaled)  # 1: 正常, -1: 异常

    return model, scaler, scores, predictions

def detect_anomalies_realtime(model, scaler, new_data: dict,
                               feature_cols: list) -> bool:
    """实时异常检测"""
    x = np.array([[new_data[col] for col in feature_cols]])
    x_scaled = scaler.transform(x)
    score = model.decision_function(x_scaled)[0]
    prediction = model.predict(x_scaled)[0]
    return prediction == -1, score

# 加载数据并训练
df = pd.read_parquet("sensor_data.parquet")
features = ["spindle_rpm", "vibration_g", "temperature_c", "current_a"]

# 仅使用正常运行数据训练
normal_df = df[df["timestamp"] < "2026-01-01"]
model, scaler, scores, preds = train_anomaly_detector(
    normal_df, features, contamination=0.03
)

# 异常区间可视化
df["anomaly_score"] = model.decision_function(
    scaler.transform(df[features].values)
)
df["is_anomaly"] = model.predict(
    scaler.transform(df[features].values)
) == -1

anomaly_count = df["is_anomaly"].sum()
print(f"检测到的异常点: {anomaly_count} / {len(df)}")

基于Autoencoder的异常检测

基于深度学习的Autoencoder能够捕捉复杂的非线性模式。用正常数据训练之后，把重构误差（Reconstruction Error）高的样本判定为异常。

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset

class SensorAutoencoder(nn.Module):
    def __init__(self, input_dim: int, latent_dim: int = 8):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, latent_dim)
        )
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, 32),
            nn.ReLU(),
            nn.Linear(32, 64),
            nn.ReLU(),
            nn.Linear(64, input_dim)
        )

    def forward(self, x):
        z = self.encoder(x)
        return self.decoder(z)

def train_autoencoder(X_normal: np.ndarray, epochs: int = 100,
                       threshold_percentile: float = 95.0):
    X_tensor = torch.FloatTensor(X_normal)
    dataset = TensorDataset(X_tensor, X_tensor)
    loader = DataLoader(dataset, batch_size=256, shuffle=True)

    model = SensorAutoencoder(input_dim=X_normal.shape[1])
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    criterion = nn.MSELoss()

    for epoch in range(epochs):
        for x_batch, _ in loader:
            recon = model(x_batch)
            loss = criterion(recon, x_batch)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

    # 阈值设置：正常数据重构误差的第95百分位
    with torch.no_grad():
        recon = model(X_tensor)
        errors = torch.mean((recon - X_tensor) ** 2, dim=1).numpy()
    threshold = np.percentile(errors, threshold_percentile)

    return model, threshold

基于LSTM的剩余使用寿命（RUL）预测

RUL（Remaining Useful Life，剩余使用寿命）预测的是设备距离故障还剩多少寿命。NASA的CMAPSS涡扇发动机数据集常被用作基准。

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

class RULPredictor(nn.Module):
    """基于LSTM的剩余使用寿命预测模型"""
    def __init__(self, input_size: int, hidden_size: int = 128,
                 num_layers: int = 2, dropout: float = 0.2):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout
        )
        self.fc = nn.Sequential(
            nn.Linear(hidden_size, 64),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(64, 1)
        )

    def forward(self, x):
        # x: (batch, seq_len, features)
        out, _ = self.lstm(x)
        # 仅使用最后一个时间步
        out = self.fc(out[:, -1, :])
        return out.squeeze(-1)

def prepare_rul_sequences(df: pd.DataFrame, seq_len: int = 30,
                           sensor_cols: list = None):
    """通过滑动窗口生成序列"""
    sequences, targets = [], []
    for engine_id in df["engine_id"].unique():
        engine_df = df[df["engine_id"] == engine_id].sort_values("cycle")
        max_cycle = engine_df["cycle"].max()
        engine_df["rul"] = max_cycle - engine_df["cycle"]

        X = engine_df[sensor_cols].values
        y = engine_df["rul"].values

        for i in range(len(X) - seq_len):
            sequences.append(X[i:i + seq_len])
            targets.append(y[i + seq_len - 1])

    return np.array(sequences), np.array(targets)

基于计算机视觉的质量检测

MVTec AD与One-Class Classification

MVTec AD是制造缺陷检测的标准基准数据集，覆盖15个工业类别中的正常样本与各类缺陷样本。它的核心特征是：训练时只提供正常图像，测试时才出现缺陷图像。

One-Class Classification更有优势的原因：在实际制造环境中，事先收集所有缺陷类型是不可能的。用正常产品训练、把偏离正常分布的样本判定为缺陷，这种方式更符合现实。

import torch
import torchvision.models as models
import torchvision.transforms as T
from torch.utils.data import DataLoader, Dataset
from PIL import Image
import os
import numpy as np
from sklearn.neighbors import NearestNeighbors

class MVTecDataset(Dataset):
    def __init__(self, root: str, split: str = "train",
                 category: str = "bottle"):
        self.transform = T.Compose([
            T.Resize((256, 256)),
            T.CenterCrop(224),
            T.ToTensor(),
            T.Normalize(mean=[0.485, 0.456, 0.406],
                        std=[0.229, 0.224, 0.225])
        ])
        self.image_paths = []
        self.labels = []

        base = os.path.join(root, category, split)
        for cls_name in os.listdir(base):
            label = 0 if cls_name == "good" else 1
            cls_dir = os.path.join(base, cls_name)
            for fname in os.listdir(cls_dir):
                if fname.endswith(".png"):
                    self.image_paths.append(os.path.join(cls_dir, fname))
                    self.labels.append(label)

    def __len__(self):
        return len(self.image_paths)

    def __getitem__(self, idx):
        img = Image.open(self.image_paths[idx]).convert("RGB")
        return self.transform(img), self.labels[idx]

class PatchCoreDetector:
    """PatchCore：预训练特征 + 基于k-NN的异常检测"""
    def __init__(self, backbone: str = "resnet50", k: int = 5):
        self.model = models.__dict__[backbone](pretrained=True)
        # 提取中间层特征
        self.model = nn.Sequential(*list(self.model.children())[:-2])
        self.model.eval()
        self.knn = NearestNeighbors(n_neighbors=k, metric="euclidean")
        self.memory_bank = None

    def extract_features(self, loader: DataLoader) -> np.ndarray:
        features = []
        with torch.no_grad():
            for imgs, _ in loader:
                feat = self.model(imgs)
                # 空间平均池化
                feat = feat.mean(dim=[2, 3]).numpy()
                features.append(feat)
        return np.concatenate(features, axis=0)

    def fit(self, train_loader: DataLoader):
        self.memory_bank = self.extract_features(train_loader)
        self.knn.fit(self.memory_bank)
        print(f"内存库大小: {self.memory_bank.shape}")

    def score(self, test_loader: DataLoader) -> np.ndarray:
        test_features = self.extract_features(test_loader)
        distances, _ = self.knn.kneighbors(test_features)
        return distances.mean(axis=1)  # 异常分数

数字孪生

数字孪生是物理资产的实时虚拟复制体。NVIDIA Omniverse提供了一个结合物理级渲染与仿真的企业级数字孪生平台。

混合模型：物理模型 + ML

纯物理模型难以捕捉复杂的非线性现象，纯数据驱动模型则可能违反物理约束。结合两种方法的Physics-Informed Neural Network（PINN）能够有效解决这个问题。

import torch
import torch.nn as nn

class HybridDigitalTwin(nn.Module):
    """
    混合数字孪生：用ML修正物理方程残差
    例：CNC机床的热变形模型
    """
    def __init__(self, physics_input_dim: int, correction_input_dim: int):
        super().__init__()
        # 基于物理的参数（可学习）
        self.thermal_coeff = nn.Parameter(torch.tensor(0.001))
        self.damping = nn.Parameter(torch.tensor(0.1))

        # ML修正网络
        self.correction_net = nn.Sequential(
            nn.Linear(correction_input_dim, 64),
            nn.Tanh(),
            nn.Linear(64, 32),
            nn.Tanh(),
            nn.Linear(32, 1)
        )

    def physics_model(self, temperature: torch.Tensor,
                       time: torch.Tensor) -> torch.Tensor:
        """简化的热膨胀模型：delta_L = alpha * L0 * delta_T"""
        delta_T = temperature - 20.0  # 基准温度20°C
        return self.thermal_coeff * delta_T * torch.exp(-self.damping * time)

    def forward(self, temperature: torch.Tensor, time: torch.Tensor,
                context_features: torch.Tensor) -> torch.Tensor:
        physics_pred = self.physics_model(temperature, time)
        correction = self.correction_net(context_features)
        return physics_pred + correction

def train_digital_twin(model, train_loader, epochs: int = 200,
                        physics_weight: float = 0.1):
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    mse = nn.MSELoss()

    for epoch in range(epochs):
        for batch in train_loader:
            temp, time_val, context, target = batch
            pred = model(temp, time_val, context)

            # 数据损失
            data_loss = mse(pred, target)

            # 物理约束：温度低时形变也应较小
            cold_mask = temp < 15.0
            physics_loss = torch.mean(
                torch.relu(pred[cold_mask])  # 对低温下的正向形变施加惩罚
            ) if cold_mask.any() else torch.tensor(0.0)

            loss = data_loss + physics_weight * physics_loss
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

供应链优化

用OR-Tools做车辆路径优化（VRP）

from ortools.constraint_solver import routing_enums_pb2
from ortools.constraint_solver import pywrapcp
import numpy as np

def create_distance_matrix(locations: list) -> list:
    """生成欧几里得距离矩阵"""
    n = len(locations)
    matrix = []
    for i in range(n):
        row = []
        for j in range(n):
            if i == j:
                row.append(0)
            else:
                dx = locations[i][0] - locations[j][0]
                dy = locations[i][1] - locations[j][1]
                row.append(int(np.sqrt(dx**2 + dy**2) * 100))
        matrix.append(row)
    return matrix

def solve_vrp(distance_matrix: list, num_vehicles: int,
               demands: list, vehicle_capacity: int) -> dict:
    """
    求解带容量约束的车辆路径问题（CVRP）
    使用OR-Tools
    """
    manager = pywrapcp.RoutingIndexManager(
        len(distance_matrix), num_vehicles, 0  # depot=0
    )
    routing = pywrapcp.RoutingModel(manager)

    def distance_callback(from_idx, to_idx):
        from_node = manager.IndexToNode(from_idx)
        to_node = manager.IndexToNode(to_idx)
        return distance_matrix[from_node][to_node]

    transit_callback_index = routing.RegisterTransitCallback(distance_callback)
    routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)

    # 容量约束
    def demand_callback(idx):
        node = manager.IndexToNode(idx)
        return demands[node]

    demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback)
    routing.AddDimensionWithVehicleCapacity(
        demand_callback_index, 0,
        [vehicle_capacity] * num_vehicles,
        True, "Capacity"
    )

    search_params = pywrapcp.DefaultRoutingSearchParameters()
    search_params.first_solution_strategy = (
        routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    )
    search_params.local_search_metaheuristic = (
        routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
    )
    search_params.time_limit.seconds = 30

    solution = routing.SolveWithParameters(search_params)

    routes = {}
    if solution:
        for vehicle_id in range(num_vehicles):
            index = routing.Start(vehicle_id)
            route = []
            while not routing.IsEnd(index):
                route.append(manager.IndexToNode(index))
                index = solution.Value(routing.NextVar(index))
            routes[vehicle_id] = route

    return routes

工业机器人AI：协作机器人与基于视觉的抓取

协作机器人（cobot，Collaborative Robot）能在与人共处的空间中安全运作，适合柔性制造环境。基于视觉的抓取结合3D相机与深度学习，能够抓取任意姿态的物体。

Sim-to-Real Transfer是把在仿真器中学到的策略应用到真实机器人上的技术。核心在于域随机化（Domain Randomization）——在仿真中随机改变物理参数（摩擦力、质量、光照）。

边缘AI部署：Jetson上的TensorRT

制造现场需要没有云端延迟的实时推理。NVIDIA Jetson（AGX Orin、Orin NX）使用TensorRT在边缘侧加速AI模型。

import tensorrt as trt
import numpy as np
import pycuda.driver as cuda
import pycuda.autoinit

TRT_LOGGER = trt.Logger(trt.Logger.WARNING)

def build_engine_from_onnx(onnx_path: str,
                             max_batch_size: int = 1,
                             fp16: bool = True) -> trt.ICudaEngine:
    """将ONNX模型转换为TensorRT引擎"""
    with trt.Builder(TRT_LOGGER) as builder:
        network_flags = 1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
        with builder.create_network(network_flags) as network:
            with trt.OnnxParser(network, TRT_LOGGER) as parser:
                with open(onnx_path, "rb") as f:
                    parser.parse(f.read())

            config = builder.create_builder_config()
            config.max_workspace_size = 1 << 30  # 1GB

            if fp16 and builder.platform_has_fast_fp16:
                config.set_flag(trt.BuilderFlag.FP16)

            engine = builder.build_engine(network, config)
            return engine

class TRTInference:
    def __init__(self, engine: trt.ICudaEngine):
        self.engine = engine
        self.context = engine.create_execution_context()
        self.bindings = []
        self.host_inputs, self.device_inputs = [], []
        self.host_outputs, self.device_outputs = [], []

        for binding in engine:
            size = trt.volume(engine.get_binding_shape(binding))
            dtype = trt.nptype(engine.get_binding_dtype(binding))
            host_mem = cuda.pagelocked_empty(size, dtype)
            device_mem = cuda.mem_alloc(host_mem.nbytes)
            self.bindings.append(int(device_mem))
            if engine.binding_is_input(binding):
                self.host_inputs.append(host_mem)
                self.device_inputs.append(device_mem)
            else:
                self.host_outputs.append(host_mem)
                self.device_outputs.append(device_mem)

    def infer(self, input_data: np.ndarray) -> np.ndarray:
        np.copyto(self.host_inputs[0], input_data.ravel())
        stream = cuda.Stream()
        cuda.memcpy_htod_async(
            self.device_inputs[0], self.host_inputs[0], stream
        )
        self.context.execute_async_v2(
            bindings=self.bindings, stream_handle=stream.handle
        )
        cuda.memcpy_dtoh_async(
            self.host_outputs[0], self.device_outputs[0], stream
        )
        stream.synchronize()
        return self.host_outputs[0]

小测验

Q1. 预测性维护中，异常检测与故障分类的区别是什么？

答案：异常检测是发现偏离正常模式的情况（无监督），故障分类是给异常类型打标签（有监督）

说明：在实际工厂中，故障数据极其稀少。异常检测可以只用正常数据做无监督学习，因此先被构建出来；故障分类则在此之后，用收集到的异常数据追加训练一个有监督模型。两个阶段的分工是：异常检测负责报警（警示），故障分类负责判定维护类型（诊断）。

Q2. 在MVTec AD中，one-class classification更有优势的原因是什么？

答案：因为在制造现场事先收集所有缺陷类型是不可能的

说明：Binary classification需要正常与缺陷两个类别的数据。但在实际制造中，缺陷是不可预测且多样的，且不可能在发生之前就收集到。One-class classification只用正常产品对正常分布建模，把新出现的缺陷类型也当作对正常分布的偏离来检测。MVTec AD上的PatchCore、PaDiM等方法都遵循这一思路。

Q3. 数字孪生混合方法的优点是什么？

答案：遵守物理约束 + 数据效率 + 对未观测区域的外推能力

说明：纯物理模型很难准确建模复杂的非线性现象（摩擦、湍流）。纯ML模型可能做出违反物理规律的预测，并且在训练分布之外的可靠性较低。PINN（Physics-Informed Neural Network）通过在损失函数中加入物理方程残差，弥补了这两个缺点。

Q4. OPC-UA为什么会被选为制造现场的标准？

答案：平台独立性、安全性（内置TLS）、语义化数据模型、实时订阅支持

说明：OPC-UA不依赖特定的操作系统或硬件，使PLC、SCADA、MES、ERP之间的集成变得容易。它内置了基于TLS的加密与认证，满足安全需求。除了简单的数据传输之外，它还提供包含数据类型、关系、方法在内的信息模型，并通过Pub/Sub机制支持实时监控。

Q5. 在协作机器人基于视觉的抓取中，sim-to-real transfer为什么重要？

答案：因为在真实机器人上收集数据既危险又昂贵，所以先在仿真中学习，再应用到真实机器人上

说明：真实机器人的训练需要成千上万次尝试，并伴随设备损坏与安全风险。在Isaac Sim或PyBullet等仿真器中应用域随机化（光照、纹理、摩擦、相机噪声），可以覆盖真实环境的多样性。这样一来，学到的抓取策略也能泛化到新物体或新光照条件下。

结语

工业4.0中AI的应用由预测性维护、质量检测、数字孪生、供应链优化、机器人自动化、边缘AI部署这六大核心领域构成。每个领域都可以独立应用，但当它们被整合进统一的智能工厂平台时，协同效应会被最大化。基于OPC-UA的数据集成，以及边缘-云混合架构，是这一切的基础。

Discuss on Twitter • View on GitHub