python的智能制造导论工业场景模拟第十六篇:基产品质检数据训练分类器，输入工艺参数，预测当前批次产品发生不良的概率。
工艺参数驱动的质量预测：用分类器给产品做“体检”
"我们车间生产汽车传动轴，最后一道工序是三坐标检测。以前都是'死后验尸'——加工完才检测，发现不合格再返工。最惨的一次，一个批次 200 根传动轴，加工完检测发现 30% 跳动超差，直接报废损失 8 万。后来我爬取了过去两年的质检数据，用 scikit-learn 训练了一个随机森林分类器：输入切削速度、进给量、冷却温度这三个工艺参数，就能预测当前批次的不良概率。现在操作工每加工 10 件就输入一次参数，系统立刻给出'良品概率 92%'或'风险预警'。上个月，我们把不良率从 4.2% 压到了 1.7%，返工成本省了 3 万多。"
—— 对应长安大学《智能制造导论》"智能工厂与智能生产—质量智能管控"：从"事后检测"向"过程预测+实时干预"演进，实现质量闭环控制。
一、实际应用场景（真实痛点）
场景设定：某汽车零部件机加工车间，生产精密传动轴（材料：42CrMo，硬度 HRC28-32）。关键质量特性为"端面跳动"（公差 ≤0.03mm），由三坐标测量机（CMM）在加工完成后检测。工艺参数主要包括：切削速度（m/min）、进给量（mm/rev）、冷却温度（℃）。当前质量控制模式为"事后全检"，缺乏过程预测能力。
现场原话（叙事化）：
"以前我们车间的质量管控，就是'赌博'。每批次 200 根传动轴，加工完才送三坐标检测。最怕听到 CMM 操作员喊'又超差了'——这意味着整批次可能要返工甚至报废。有次夜班，因为冷却系统波动，温度从 25℃ 升到 32℃，但操作工没注意。结果第二天检测，端面跳动超差 30%，8 万块钱直接打水漂。后来我复盘数据，发现超差批次的冷却温度普遍高于 30℃，切削速度超过 180m/min。于是我写了个 Python 程序：把过去两年的质检数据（工艺参数+检测结果）喂给随机森林分类器，训练出一个'质量预测模型'。现在操作工在加工过程中，每 10 件输入一次实时工艺参数，系统立刻弹出'良品概率'。上个月，系统两次预警'冷却温度过高，不良风险 78%'，我们及时调整，避免了两次潜在批量事故。质量主管现在开会都说：'质量不是检出来的，是算出来的。'"
痛点分析（映射到课程模型）：
《智能制造导论》模块 本篇痛点对应
概述：质量第一、全生命周期管理 事后检测：仅关注最终检验结果，缺乏过程质量控制，导致批量损失
智能制造技术基础：数据采集、机器学习基础 数据未建模：积累了大量"工艺参数-质量结果"数据，但未用于预测
新一代支撑技术：工业大数据、AI 算法 缺乏预测能力：无法根据实时参数预判质量风险，依赖人工经验
智能工厂 / 智能生产：质量智能管控、闭环控制 开环生产：工艺参数调整与质量结果反馈脱节，未形成闭环
演进范式：数字化→网络化→智能化 数据驱动质量：从"经验调参"到"模型预测"，实现预防性质量控制
核心矛盾："高质量要求"与"滞后质量反馈"的矛盾。精密零件加工对工艺参数敏感，但质量检测结果滞后（加工完成后），导致问题发现太晚，无法实时干预。
二、核心逻辑讲解（大白话）
2.1 质量预测就像"体检报告"
- 历史数据：像"健康档案"——记录了过去病人的"生活习惯（工艺参数）"和"患病情况（合格/不合格）"。
- 训练模型：像"医生学习"——通过大量档案，总结出"吸烟+熬夜→高血压"等规律。
- 预测概率：像"体检报告"——输入当前的"生活习惯（实时工艺参数）"，模型输出"患病概率（不良概率）"。
- 核心逻辑："历史规律"→"模型学习"→"未来预测"。
2.2 业务逻辑 → 代码映射
历史质检数据（CSV：工艺参数 + 检测结果）
│
▼ QualityDataLoader.load_data()
加载数据：切削速度、进给量、冷却温度 → 特征(X)；合格/不合格 → 标签(y)
│
▼ DataPreprocessor.split_data()
数据预处理：
1. 缺失值处理：填充或删除
2. 特征缩放：标准化（让不同量纲的参数可比）
3. 数据集划分：训练集(80%) + 测试集(20%)
│
▼ QualityPredictor.train_model()
模型训练：
1. 选择算法：随机森林（适合表格数据、抗过拟合）
2. 训练：用训练集(X_train, y_train)拟合模型
3. 评估：用测试集(X_test, y_test)计算准确率、召回率
│
▼ QualityPredictor.predict_proba()
实时预测：
输入当前工艺参数 → 模型输出不良概率（0~1）
│
▼ QualityVisualizer.plot_xxx()
可视化：
1. 特征重要性柱状图（哪些参数影响最大）
2. 决策边界图（参数组合与质量的关系）
3. 预测概率仪表盘（实时风险展示）
2.3 为什么用随机森林而不是深度学习？
- 数据量适中：工业现场通常只有几百到几千条质检记录，深度学习（如神经网络）需要大数据，容易过拟合；随机森林在小数据集上表现稳定。
- 可解释性强：随机森林能输出"特征重要性"（如冷却温度贡献度 45%），方便工艺员理解；深度学习是"黑盒"，难以解释"为什么预测为不良"。
- 鲁棒性高：对特征缩放不敏感，能处理非线性关系（如温度过高与不良率的非线性关系），适合工业复杂场景。
- 部署简单：训练好的模型可保存为文件，用
"joblib" 加载，无需复杂环境，适合车间上位机部署。
三、OOP 代码实现
3.1 项目结构
quality_prediction/
├── quality_prediction.py # 核心逻辑（~350 行）
├── test_quality_prediction.py # 单元测试（9/9 通过）
├── visualize.py # 可视化入口
├── pack.py # 打包脚本
├── README.md
├── quality_data.csv # 示例质检数据
├── quality_model.joblib # 训练好的模型
├── feature_importance.png # 特征重要性图
├── confusion_matrix.png # 混淆矩阵
├── decision_boundary.png # 决策边界图
├── probability_gauge.png # 预测概率仪表盘
└── quality_prediction.zip # 打包产物
3.2 核心源码
<details>
<summary></summary>
"""
基于工艺参数的产品质量预测：随机森林分类器
=========================================
任务：训练分类器，输入工艺参数，预测当前批次产品不良概率
课程映射（长安大学《智能制造导论》）：
- 概述：质量第一、全生命周期管理（预防性质量控制）
- 智能制造技术基础：数据采集、机器学习基础（监督学习分类）
- 新一代支撑技术：工业大数据、AI算法（随机森林应用）
- 智能工厂/智能生产：质量智能管控、闭环控制（参数-质量映射）
- 演进范式：数字化 → 网络化 → 智能化（数据驱动质量决策）
技术栈（严格限制）：
pandas, numpy, matplotlib, seaborn
scikit-learn（核心：随机森林分类器、模型评估）
scipy（统计检验）、joblib（模型持久化）
（networkx/pytorch：本篇未启用，为后续GNN/深度学习预留）
"""
from dataclasses import dataclass
from typing import List, Dict, Optional, Tuple, Any
import json
import warnings
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import (
accuracy_score, precision_score, recall_score,
f1_score, confusion_matrix, classification_report
)
from sklearn.pipeline import Pipeline
import joblib
# 忽略 sklearn 的 FutureWarning
warnings.filterwarnings("ignore", category=FutureWarning)
# ----------------------------------------------------------------------
# 1. 数据模型
# ----------------------------------------------------------------------
@dataclass
class ProcessParameters:
"""工艺参数数据模型。"""
cutting_speed: float # 切削速度 (m/min)
feed_rate: float # 进给量 (mm/rev)
coolant_temp: float # 冷却温度 (℃)
timestamp: Optional[pd.Timestamp] = None # 采集时间
@dataclass
class QualityRecord:
"""单条质检记录。"""
record_id: str
parameters: ProcessParameters
is_defective: bool # True=不合格，False=合格
defect_type: Optional[str] = None # 缺陷类型（如"跳动超差"）
inspector: Optional[str] = None # 检验员
# ----------------------------------------------------------------------
# 2. 数据加载
# ----------------------------------------------------------------------
class QualityDataLoader:
"""质检数据加载器。"""
def __init__(self, data_path: str = "quality_data.csv"):
self.data_path = data_path
self.df: Optional[pd.DataFrame] = None
def load_data(self) -> pd.DataFrame:
"""加载数据：优先从 CSV 读取，若无则生成示例数据。"""
try:
self.df = pd.read_csv(self.data_path, parse_dates=["timestamp"])
print(f"[INFO] 从 {self.data_path} 加载数据，共 {len(self.df)} 条")
except FileNotFoundError:
print("[INFO] 未找到数据文件，生成示例质检数据...")
self.df = self._generate_sample_data()
self.df.to_csv(self.data_path, index=False)
print(f"[INFO] 示例数据已保存至 {self.data_path}")
return self.df
def _generate_sample_data(self, n_samples: int = 800, seed: int = 42) -> pd.DataFrame:
"""
生成示例质检数据（模拟传动轴加工）：
- 工艺参数：
* 切削速度：150-200 m/min（正态分布）
* 进给量：0.1-0.3 mm/rev（正态分布）
* 冷却温度：20-35 ℃（正态分布）
- 质量逻辑：
* 冷却温度 > 30℃ → 不良概率 +40%
* 切削速度 > 180m/min → 不良概率 +30%
* 进给量 > 0.25mm/rev → 不良概率 +20%
* 三者叠加 → 不良概率 > 70%
- 缺陷类型：跳动超差(70%)、表面粗糙度差(20%)、尺寸超差(10%)
"""
rng = np.random.default_rng(seed)
# 生成工艺参数
cutting_speed = rng.normal(175, 10, n_samples).clip(150, 200) # 150-200 m/min
feed_rate = rng.normal(0.2, 0.05, n_samples).clip(0.1, 0.3) # 0.1-0.3 mm/rev
coolant_temp = rng.normal(26, 4, n_samples).clip(20, 35) # 20-35 ℃
# 生成时间戳（最近1年）
start_time = pd.Timestamp("2025-09-01")
timestamps = [
start_time + pd.Timedelta(days=rng.integers(0, 365))
for _ in range(n_samples)
]
# 基于工艺参数计算不良概率
defect_prob = np.zeros(n_samples)
defect_prob += np.where(coolant_temp > 30, 0.4, 0) # 冷却温度影响最大
defect_prob += np.where(cutting_speed > 180, 0.3, 0) # 切削速度次之
defect_prob += np.where(feed_rate > 0.25, 0.2, 0) # 进给量影响最小
defect_prob += rng.normal(0, 0.05, n_samples) # 随机噪声
# 根据概率生成合格/不合格标签
is_defective = defect_prob > rng.uniform(0.2, 0.5, n_samples)
# 缺陷类型（仅不合格品）
defect_types = []
for defective in is_defective:
if defective:
# 70%跳动超差，20%粗糙度差，10%尺寸超差
rand = rng.random()
if rand < 0.7:
defect_types.append("跳动超差")
elif rand < 0.9:
defect_types.append("表面粗糙度差")
else:
defect_types.append("尺寸超差")
else:
defect_types.append(None)
# 构建DataFrame
records = []
for i in range(n_samples):
records.append({
"record_id": f"Q{2026000 + i}",
"cutting_speed": round(cutting_speed[i], 1),
"feed_rate": round(feed_rate[i], 3),
"coolant_temp": round(coolant_temp[i], 1),
"timestamp": timestamps[i],
"is_defective": is_defective[i],
"defect_type": defect_types[i],
"inspector": f"INS{rng.integers(1, 4):02d}"
})
return pd.DataFrame(records)
# ----------------------------------------------------------------------
# 3. 数据预处理
# ----------------------------------------------------------------------
class DataPreprocessor:
"""数据预处理器。"""
def __init__(self, test_size: float = 0.2, random_state: int = 42):
self.test_size = test_size
self.random_state = random_state
self.scaler = StandardScaler()
self.feature_columns = ["cutting_speed", "feed_rate", "coolant_temp"]
def prepare_data(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
"""数据预处理主流程。"""
# 1. 特征选择
X = df[self.feature_columns].copy()
y = df["is_defective"].astype(int) # True/False → 1/0
# 2. 缺失值处理（简单填充，实际可更复杂）
X = self._handle_missing_values(X)
# 3. 异常值检测（可选，用Z-score）
X = self._handle_outliers(X)
# 4. 数据集划分
X_train, X_test, y_train, y_test = train_test_split(
X, y,
test_size=self.test_size,
random_state=self.random_state,
stratify=y # 保持训练集和测试集的类别比例
)
# 5. 特征缩放（标准化）
X_train_scaled = self.scaler.fit_transform(X_train)
X_test_scaled = self.scaler.transform(X_test)
# 转回DataFrame（保留列名）
X_train_scaled = pd.DataFrame(X_train_scaled, columns=self.feature_columns, index=X_train.index)
X_test_scaled = pd.DataFrame(X_test_scaled, columns=self.feature_columns, index=X_test.index)
return X_train_scaled, X_test_scaled, y_train, y_test
def _handle_missing_values(self, X: pd.DataFrame) -> pd.DataFrame:
"""处理缺失值（用中位数填充）。"""
for col in X.columns:
if X[col].isnull().any():
median_val = X[col].median()
X[col] = X[col].fillna(median_val)
print(f"[INFO] 列 {col} 缺失值用中位数 {median_val} 填充")
return X
def _handle_outliers(self, X: pd.DataFrame, z_threshold: float = 3.0) -> pd.DataFrame:
"""处理异常值（Z-score方法，截断到阈值内）。"""
X_clean = X.copy()
for col in X.columns:
z_scores = np.abs(stats.zscore(X[col]))
outliers = z_scores > z_threshold
if outliers.any():
# 截断到 ±3σ 范围内
mean_val = X[col].mean()
std_val = X[col].std()
lower_bound = mean_val - z_threshold * std_val
upper_bound = mean_val + z_threshold * std_val
X_clean.loc[X[col] < lower_bound, col] = lower_bound
X_clean.loc[X[col] > upper_bound, col] = upper_bound
print(f"[INFO] 列 {col} 发现 {outliers.sum()} 个异常值，已截断处理")
return X_clean
def get_feature_names(self) -> List[str]:
"""返回特征列名。"""
return self.feature_columns
def get_scaler(self):
"""返回训练好的缩放器。"""
return self.scaler
# ----------------------------------------------------------------------
# 4. 质量预测模型
# ----------------------------------------------------------------------
class QualityPredictor:
"""质量预测分类器。"""
def __init__(self, n_estimators: int = 100, random_state: int = 42):
self.n_estimators = n_estimators
self.random_state = random_state
self.model = RandomForestClassifier(
n_estimators=n_estimators,
random_state=random_state,
max_depth=5, # 限制深度，防止过拟合
min_samples_split=5, # 内部节点再划分所需最小样本数
min_samples_leaf=2, # 叶节点最少样本数
class_weight="balanced" # 处理类别不平衡
)
self.scaler = None
self.feature_names = None
self.metrics = {}
def train(self, X_train: pd.DataFrame, y_train: pd.Series):
"""训练模型。"""
self.feature_names = X_train.columns.tolist()
self.model.fit(X_train, y_train)
print(f"[INFO] 随机森林模型训练完成，树数量: {self.n_estimators}")
def evaluate(self, X_test: pd.DataFrame, y_test: pd.Series) -> Dict[str, float]:
"""评估模型性能。"""
y_pred = self.model.predict(X_test)
y_pred_proba = self.model.predict_proba(X_test)[:, 1]
# 计算指标
self.metrics = {
"accuracy": accuracy_score(y_test, y_pred),
"precision": precision_score(y_test, y_pred, zero_division=0),
"recall": recall_score(y_test, y_pred, zero_division=0),
"f1": f1_score(y_test, y_pred, zero_division=0)
}
# 混淆矩阵
self.confusion_mat = confusion_matrix(y_test, y_pred)
print("\n=== 模型评估报告 ===")
print(f"准确率 (Accuracy): {self.metrics['accuracy']:.3f}")
print(f"精确率 (Precision): {self.metrics['precision']:.3f}")
print(f"召回率 (Recall): {self.metrics['recall']:.3f}")
print(f"F1分数 (F1-Score): {self.metrics['f1']:.3f}")
print("\n混淆矩阵:")
print(self.confusion_mat)
print("\n分类报告:")
print(classification_report(y_test, y_pred, target_names=["合格", "不合格"], zero_division=0))
return self.metrics
def predict_proba(self, X: pd.DataFrame) -> np.ndarray:
"""
预测不良概率。
返回：二维数组，[:, 0]=合格概率，[:, 1]=不良概率
"""
if self.feature_names is None:
raise ValueError("模型尚未训练，请先调用 train() 方法")
# 确保输入特征顺序与训练时一致
X_aligned = X[self.feature_names]
return self.model.predict_proba(X_aligned)
def predict(self, X: pd.DataFrame, threshold: float = 0.5) -> np.ndarray:
"""
预测类别（0=合格，1=不合格）。
threshold: 不良概率阈值，>threshold 预测为不合格
"""
proba = self.predict_proba(X)[:, 1]
return (proba > threshold).astype(int)
def get_feature_importance(self) -> pd.DataFrame:
"""获取特征重要性。"""
if self.feature_names is None:
raise ValueError("模型尚未训练，请先调用 train() 方法")
importance = self.model.feature_importances_
return pd.DataFrame({
"feature": self.feature_names,
"importance": importance
}).sort_values("importance", ascending=False).reset_index(drop=True)
def save_model(self, path: str = "quality_model.joblib"):
"""保存模型到文件。"""
model_data = {
"model": self.model,
"feature_names": self.feature_names,
"metrics": self.metrics
}
joblib.dump(model_data, path)
print(f"[INFO] 模型已保存至 {path}")
@classmethod
def load_model(cls, path: str = "quality_model.joblib") -> "QualityPredictor":
"""从文件加载模型。"""
model_data = joblib.load(path)
predictor = cls()
predictor.model = model_data["model"]
predictor.feature_names = model_data["feature_names"]
predictor.metrics = model_data.get("metrics", {})
print(f"[INFO] 模型已从 {path} 加载")
return predictor
# ----------------------------------------------------------------------
# 5. 可视化
# ----------------------------------------------------------------------
class QualityVisualizer:
"""质量预测可视化器。"""
@staticmethod
def plot_feature_importance(importance_df: pd.DataFrame, output_file: str = "feature_importance.png"):
"""绘制特征重要性柱状图。"""
plt.figure(figsize=(10, 6))
bars = plt.bar(
importance_df["feature"],
importance_df["importance"],
color="#1f77b4",
edgecolor="black",
linewidth=0.5
)
plt.title("工艺参数对质量的影响程度（特征重要性）", fontsize=16, fontweight="bold", fontfamily="SimHei")
plt.xlabel("工艺参数", fontsize=12, fontfamily="SimHei")
plt.ylabel("重要性得分", fontsize=12, fontfamily="SimHei")
plt.grid(axis="y", linestyle="--", alpha=0.7)
# 添加数值标签
for bar, (_, row) in zip(bars, importance_df.iterrows()):
height = bar.get_height()
plt.text(
bar.get_x() + bar.get_width()/2,
height + 0.01,
f"{height:.3f}",
ha="center",
va="bottom",
fontsize=10
)
plt.xticks(rotation=0, fontfamily="SimHei")
plt.tight_layout()
plt.savefig(output_file, dpi=120, bbox_inches="tight")
plt.close()
print(f"[INFO] 已生成特征重要性图: {output_file}")
@staticmethod
def plot_confusion_matrix(confusion_mat: np.ndarray, output_file: str = "confusion_matrix.png"):
"""绘制混淆矩阵热力图。"""
plt.figure(figsize=(8, 6))
sns.heatmap(
confusion_mat,
annot=True,
fmt="d",
cmap="Blues",
xticklabels=["合格", "不合格"],
yticklabels=["合格", "不合格"],
cbar_kws={"label": "样本数量"}
)
plt.title("模型混淆矩阵", fontsize=16, fontweight="bold", fontfamily="SimHei")
plt.xlabel("预测类别", fontsize=12, fontfamily="SimHei")
plt.ylabel("真实类别", fontsize=12, fontfamily="SimHei")
plt.tight_layout()
plt.savefig(output_file, dpi=120, bbox_inches="tight")
plt.close()
print(f"[INFO] 已生成混淆矩阵图: {output_file}")
@staticmethod
def plot_decision_boundary(
model: QualityPredictor,
X_train: pd.DataFrame,
y_train: pd.Series,
feature_x: str = "coolant_temp",
feature_y: str = "cutting_speed",
output_file: str = "decision_boundary.png"
):
"""绘制二维决策边界（固定第三个参数为均值）。"""
# 创建网格
x_min, x_max = X_train[feature_x].min() - 0.5, X_train[feature_x].max() + 0.5
y_min, y_max = X_train[feature_y].min() - 0.5, X_train[feature_y].max() + 0.5
xx, yy = np.meshgrid(
np.arange(x_min, x_max, 0.02),
np.arange(y_min, y_max, 0.02)
)
# 准备预测数据（第三个特征用训练集均值）
third_feature = [f for f in X_train.columns if f not in [feature_x, feature_y]][0]
third_value = X_train[third_feature].mean()
grid_data = pd.DataFrame({
feature_x: xx.ravel(),
feature_y: yy.ravel(),
third_feature: np.full(xx.size, third_value)
})
# 预测概率
Z = model.predict_proba(grid_data)[:, 1]
Z = Z.reshape(xx.shape)
# 绘制
plt.figure(figsize=(10, 8))
# 填充决策区域
contour = plt.contourf(xx, yy, Z, alpha=0.8, cmap=plt.cm.RdYlGn_r)
plt.colorbar(contour, label="不良概率")
# 绘制训练样本点
scatter = plt.scatter(
X_train[feature_x],
X_train[feature_y],
c=y_train,
edgecolors="k",
cmap=plt.cm.RdYlGn_r,
s=50,
alpha=0.9
)
plt.legend(handles=scatter.legend_elements()[0], labels=["合格", "不合格"], loc="upper right")
plt.title(f"质量预测决策边界\n（{third_feature}={third_value:.2f}）", fontsize=16, fontweight="bold", fontfamily="SimHei")
plt.xlabel(feature_x, fontsize=12, fontfamily="SimHei")
plt.ylabel(feature_y, fontsize=12, fontfamily="SimHei")
plt.grid(linestyle="--", alpha=0.3)
plt.tight_layout()
plt.savefig(output_file, dpi=120, bbox_inches="tight")
plt.close()
print(f"[INFO] 已生成决策边界图: {output_file}")
@staticmethod
def plot_probability_gauge(probability: float, output_file: str = "probability_gauge.png"):
"""绘制预测概率仪表盘。"""
fig, ax = plt.subplots(figsize=(10, 6), subplot_kw={"aspect": "equal"})
# 仪表盘参数
start_angle = 180
end_angle = 0
angles = np.linspace(start_angle, end_angle, 100)
# 背景环（灰色）
ax.pie(
[1],
startangle=start_angle,
colors=["#f0f0f0"],
wedgeprops=dict(width=0.3, edgecolor="w")
)
# 概率环（根据概率变色）
if probability < 0.3:
color = "#2ecc71" # 绿色：低风险
elif probabili
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
更多推荐
0
0- 0
已为社区贡献7条内容
所有评论(0)