AI与传统规则体系的无缝集成：制造业缺陷检测的演进与融合
传统制造业的质检体系经历了从人工目检到自动化检测的演进过程。长期以来，基于规则的机器视觉系统（如OpenCV）在生产线上担任着主要的检测角色。这些系统依靠工程师手动设计的特征提取算法和阈值规则，在特定的产品线和光照条件下表现稳定。然而，随着产品复杂度提升和产能要求增加，传统规则体系的局限性日益凸显。
现代化AI决策模型，特别是基于深度学习的视觉检测模型，在检测精度和泛化能力上显著超越了传统规则体系。但AI模型并非万能的，它存在数据依赖性强、可解释性差、对长尾缺陷覆盖不足等问题。最理想的方案是将传统规则体系的确定性与AI模型的灵活性有机结合，构建混合智能检测系统。
本文将系统探讨传统规则体系与AI决策模型的集成模式，并结合制造业缺陷检测场景给出完整的实现方案。
一、 传统规则体系的核心能力与局限
传统规则体系以确定性算法为基础，工程师通过分析缺陷的物理特征（亮度、对比度、形状、纹理等），设计相应的图像处理流程。这类方法的优势在于计算效率高、可解释性强、无需大量标注数据。
典型的传统检测流程包括：图像预处理（灰度化、滤波去噪）、特征提取（边缘检测、形态学操作、纹理分析）、规则判定（阈值比较、模板匹配）。
import numpy as np
import cv2
import torch
import torch.nn as nn
import torch.nn.functional as F
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
class TraditionalRuleEngine:
def __init__(self):
self.thresholds = {
'edge_intensity': 0.3,
'texture_uniformity': 0.6,
'brightness_std': 25.0,
'area_threshold': 50,
'aspect_ratio_min': 0.5,
'aspect_ratio_max': 2.0,
'contour_solidity': 0.7
}
def preprocess(self, image):
if len(image.shape) == 3:
gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
else:
gray = image
blurred = cv2.GaussianBlur(gray, (5, 5), 0)
equalized = cv2.equalizeHist(blurred.astype(np.uint8))
return blurred, equalized
def extract_handcrafted_features(self, image):
blurred, equalized = self.preprocess(image)
features = {}
edges = cv2.Canny(blurred.astype(np.uint8), 50, 150)
features['edge_intensity'] = np.mean(edges) / 255.0
grad_x = cv2.Sobel(blurred.astype(np.uint8), cv2.CV_64F, 1, 0, ksize=3)
grad_y = cv2.Sobel(blurred.astype(np.uint8), cv2.CV_64F, 0, 1, ksize=3)
grad_mag = np.sqrt(grad_x ** 2 + grad_y ** 2)
features['gradient_mean'] = np.mean(grad_mag)
features['gradient_std'] = np.std(grad_mag)
_, binary = cv2.threshold(blurred.astype(np.uint8), 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
contours, _ = cv2.findContours(binary, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
features['contour_count'] = len(contours)
if contours:
areas = [cv2.contourArea(c) for c in contours]
features['max_area'] = max(areas)
features['mean_area'] = np.mean(areas)
features['area_std'] = np.std(areas)
largest_contour = max(contours, key=cv2.contourArea)
hull = cv2.convexHull(largest_contour)
hull_area = cv2.contourArea(hull)
contour_area = cv2.contourArea(largest_contour)
features['solidity'] = contour_area / hull_area if hull_area > 0 else 0
x, y, w, h = cv2.boundingRect(largest_contour)
features['aspect_ratio'] = w / h if h > 0 else 0
else:
features['max_area'] = 0
features['mean_area'] = 0
features['area_std'] = 0
features['solidity'] = 0
features['aspect_ratio'] = 0
glcm = self._compute_glcm(equalized.astype(np.uint8))
features['texture_contrast'] = glcm.get('contrast', 0)
features['texture_energy'] = glcm.get('energy', 0)
features['texture_homogeneity'] = glcm.get('homogeneity', 0)
features['brightness_mean'] = np.mean(blurred)
features['brightness_std'] = np.std(blurred)
return features
def _compute_glcm(self, image, distance=1, angle=0):
h, w = image.shape
glcm = np.zeros((256, 256))
if angle == 0:
for i in range(h):
for j in range(w – distance):
glcm[image[i, j], image[i, j + distance]] += 1
total = glcm.sum()
if total > 0:
glcm = glcm / total
contrast = np.sum(glcm * (np.arange(256)[:, None] – np.arange(256)[None, :]) ** 2)
energy = np.sum(glcm ** 2)
homogeneity = np.sum(glcm / (1 + np.abs(np.arange(256)[:, None] – np.arange(256)[None, :])))
return {'contrast': contrast, 'energy': energy, 'homogeneity': homogeneity}
def rule_based_classify(self, features):
defects = []
if features['edge_intensity'] > self.thresholds['edge_intensity']:
defects.append(('边缘异常', features['edge_intensity']))
if features['texture_contrast'] > 50:
defects.append(('纹理异常', features['texture_contrast']))
if features['brightness_std'] > self.thresholds['brightness_std']:
defects.append(('亮度不均', features['brightness_std']))
if features['max_area'] > self.thresholds['area_threshold']:
defects.append(('大面积缺陷', features['max_area']))
if features['solidity'] < self.thresholds['contour_solidity']:
defects.append(('形状不规则', features['solidity']))
if features['aspect_ratio'] < self.thresholds['aspect_ratio_min'] or \\
features['aspect_ratio'] > self.thresholds['aspect_ratio_max']:
defects.append(('长宽比异常', features['aspect_ratio']))
score = len(defects) / 7.0
is_defective = score > 0.3
return {
'is_defective': is_defective,
'defect_score': score,
'defects': defects,
'confidence': 1.0 – abs(score – 0.5) * 2
}
二、 AI决策模型的设计与训练
与传统规则体系互补，AI模型通过端到端学习从数据中自动发现缺陷模式。CNN可以学习到规则系统难以手工编码的复杂特征，对于纹理类缺陷、微小缺陷和变化多样的缺陷类型具有更好的鲁棒性。
我们设计了一个轻量级CNN模型，适合部署在制造业的边缘计算设备上。模型采用深度可分离卷积和全局平均池化，在保持较高检测精度的同时控制了参数量和计算量。
class LightweightDefectCNN(nn.Module):
def __init__(self, in_channels=3, num_classes=2, width_mult=1.0):
super().__init__()
def conv_dw(in_c, out_c, stride):
return nn.Sequential(
nn.Conv2d(in_c, in_c, 3, stride=stride, padding=1, groups=in_c, bias=False),
nn.BatchNorm2d(in_c),
nn.ReLU6(inplace=True),
nn.Conv2d(in_c, out_c, 1, stride=1, padding=0, bias=False),
nn.BatchNorm2d(out_c),
nn.ReLU6(inplace=True),
)
self.features = nn.Sequential(
nn.Conv2d(in_channels, int(32 * width_mult), 3, stride=2, padding=1, bias=False),
nn.BatchNorm2d(int(32 * width_mult)),
nn.ReLU6(inplace=True),
conv_dw(int(32 * width_mult), int(64 * width_mult), 1),
conv_dw(int(64 * width_mult), int(128 * width_mult), 2),
conv_dw(int(128 * width_mult), int(128 * width_mult), 1),
conv_dw(int(128 * width_mult), int(256 * width_mult), 2),
conv_dw(int(256 * width_mult), int(256 * width_mult), 1),
conv_dw(int(256 * width_mult), int(512 * width_mult), 2),
conv_dw(int(512 * width_mult), int(512 * width_mult), 1),
conv_dw(int(512 * width_mult), int(512 * width_mult), 1),
conv_dw(int(512 * width_mult), int(512 * width_mult), 1),
conv_dw(int(512 * width_mult), int(512 * width_mult), 1),
conv_dw(int(512 * width_mult), int(1024 * width_mult), 2),
conv_dw(int(1024 * width_mult), int(1024 * width_mult), 1),
nn.AdaptiveAvgPool2d(1)
)
self.classifier = nn.Sequential(
nn.Dropout(0.2),
nn.Linear(int(1024 * width_mult), num_classes)
)
def forward(self, x):
x = self.features(x)
x = x.view(x.size(0), -1)
x = self.classifier(x)
return x
def extract_embedding(self, x):
x = self.features(x)
x = x.view(x.size(0), -1)
return x
def count_parameters(model):
return sum(p.numel() for p in model.parameters() if p.requires_grad)
三、 规则与AI的集成架构设计
传统规则体系与AI模型各有优势和短板。规则系统擅长处理已知的、固定模式的缺陷，但对未知缺陷的泛化能力差。AI模型擅长学习复杂模式，但对训练数据分布外的缺陷可能产生不可预测的错误。混合集成策略可以取长补短。
我们设计了三种集成模式：串行级联模式、并行投票模式和动态路由模式。串行级联模式先用规则系统快速筛选明显良品，再用AI模型处理可疑样本。并行投票模式让两个系统独立检测，通过投票规则综合判断。动态路由模式根据输入样本的特征自动选择最优检测器。
class HybridInspectionSystem:
def __init__(self, rule_engine, ai_model, fusion_strategy='cascade'):
self.rule_engine = rule_engine
self.ai_model = ai_model
self.ai_model.eval()
self.fusion_strategy = fusion_strategy
self.rule_weight = 0.4
self.ai_weight = 0.6
self.calibration_data = []
def cascade_inspect(self, image, rule_threshold=0.2, ai_threshold=0.5):
features = self.rule_engine.extract_handcrafted_features(image)
rule_result = self.rule_engine.rule_based_classify(features)
if rule_result['defect_score'] < rule_threshold:
return {
'decision': 'pass',
'score': rule_result['defect_score'],
'method': 'rule_only',
'confidence': rule_result['confidence'],
'details': {'rule_score': rule_result['defect_score']}
}
if rule_result['defect_score'] > 0.8:
return {
'decision': 'reject',
'score': rule_result['defect_score'],
'method': 'rule_only',
'confidence': rule_result['confidence'],
'details': {'rule_score': rule_result['defect_score']}
}
image_tensor = self._preprocess_for_ai(image)
with torch.no_grad():
ai_output = self.ai_model(image_tensor)
ai_proba = F.softmax(ai_output, dim=1)
ai_defect_prob = ai_proba[0, 1].item()
fused_score = self.rule_weight * rule_result['defect_score'] + self.ai_weight * ai_defect_prob
decision = 'reject' if fused_score > ai_threshold else 'pass'
return {
'decision': decision,
'score': fused_score,
'method': 'cascade',
'confidence': max(abs(fused_score – 0.5) * 2, 0.5),
'details': {
'rule_score': rule_result['defect_score'],
'ai_score': ai_defect_prob,
'rule_defects': [d[0] for d in rule_result['defects']]
}
}
def parallel_vote_inspect(self, image, ai_threshold=0.5, rule_threshold=0.3):
features = self.rule_engine.extract_handcrafted_features(image)
rule_result = self.rule_engine.rule_based_classify(features)
image_tensor = self._preprocess_for_ai(image)
with torch.no_grad():
ai_output = self.ai_model(image_tensor)
ai_proba = F.softmax(ai_output, dim=1)
ai_defect_prob = ai_proba[0, 1].item()
rule_decision = rule_result['defect_score'] > rule_threshold
ai_decision = ai_defect_prob > ai_threshold
votes = sum([rule_decision, ai_decision])
if votes >= 2:
decision = 'reject'
elif votes == 0:
decision = 'pass'
else:
decision = 'review'
fused_score = self.rule_weight * rule_result['defect_score'] + self.ai_weight * ai_defect_prob
return {
'decision': decision,
'score': fused_score,
'method': 'parallel_vote',
'confidence': votes / 2.0,
'details': {
'rule_decision': rule_decision,
'ai_decision': ai_decision,
'rule_score': rule_result['defect_score'],
'ai_score': ai_defect_prob
}
}
def dynamic_route_inspect(self, image, confidence_threshold=0.6):
features = self.rule_engine.extract_handcrafted_features(image)
rule_result = self.rule_engine.rule_based_classify(features)
if rule_result['confidence'] > confidence_threshold:
decision = 'reject' if rule_result['is_defective'] else 'pass'
return {
'decision': decision,
'score': rule_result['defect_score'],
'method': 'dynamic_route_rule',
'confidence': rule_result['confidence'],
'details': {'rule_score': rule_result['defect_score']}
}
image_tensor = self._preprocess_for_ai(image)
with torch.no_grad():
ai_output = self.ai_model(image_tensor)
ai_proba = F.softmax(ai_output, dim=1)
ai_defect_prob = ai_proba[0, 1].item()
ai_confidence = max(ai_defect_prob, 1 – ai_defect_prob)
decision = 'reject' if ai_defect_prob > 0.5 else 'pass'
return {
'decision': decision,
'score': ai_defect_prob,
'method': 'dynamic_route_ai',
'confidence': ai_confidence,
'details': {'ai_score': ai_defect_prob, 'rule_confidence': rule_result['confidence']}
}
def _preprocess_for_ai(self, image):
if len(image.shape) == 2:
image_rgb = cv2.cvtColor(image, cv2.COLOR_GRAY2RGB)
elif image.shape[2] == 4:
image_rgb = cv2.cvtColor(image, cv2.COLOR_RGBA2RGB)
else:
image_rgb = image
resized = cv2.resize(image_rgb, (224, 224))
normalized = resized.astype(np.float32) / 255.0
normalized = (normalized – np.array([0.485, 0.456, 0.406])) / np.array([0.229, 0.224, 0.225])
tensor = torch.FloatTensor(normalized).permute(2, 0, 1).unsqueeze(0)
return tensor
def update_weights(self, new_rule_weight):
self.rule_weight = new_rule_weight
self.ai_weight = 1.0 – new_rule_weight
四、 规则系统特征与AI特征的融合学习
除了在决策层面进行集成，还可以在特征层面将规则系统提取的手工特征与CNN学习的深层特征进行融合。这种融合方式可以同时利用人类的先验知识和数据驱动的模式发现能力。
我们设计了一个融合分类器，将规则系统的特征向量和CNN的嵌入向量拼接后，输入一个轻量级的梯度提升分类器进行最终决策。
class FeatureFusionClassifier:
def __init__(self, rule_feature_dim=12, cnn_embedding_dim=512):
self.rule_feature_dim = rule_feature_dim
self.cnn_embedding_dim = cnn_embedding_dim
self.fusion_dim = rule_feature_dim + cnn_embedding_dim
self.classifier = GradientBoostingClassifier(
n_estimators=200,
max_depth=5,
learning_rate=0.1,
subsample=0.8,
min_samples_leaf=10,
random_state=42
)
self.scaler = StandardScaler()
self.rule_engine = TraditionalRuleEngine()
def extract_rule_features_batch(self, images):
features_list = []
for image in images:
features = self.rule_engine.extract_handcrafted_features(image)
feature_vector = [
features['edge_intensity'], features['gradient_mean'],
features['gradient_std'], features['contour_count'],
features['max_area'], features['mean_area'],
features['area_std'], features['solidity'],
features['aspect_ratio'], features['texture_contrast'],
features['texture_energy'], features['texture_homogeneity']
]
features_list.append(feature_vector)
return np.array(features_list)
def extract_cnn_embeddings_batch(self, images, cnn_model):
cnn_model.eval()
embeddings = []
with torch.no_grad():
for image in images:
if isinstance(image, np.ndarray):
if len(image.shape) == 2:
image_rgb = cv2.cvtColor(image, cv2.COLOR_GRAY2RGB)
else:
image_rgb = image
resized = cv2.resize(image_rgb, (224, 224))
tensor = torch.FloatTensor(resized).permute(2, 0, 1).unsqueeze(0) / 255.0
else:
tensor = image
emb = cnn_model.extract_embedding(tensor)
embeddings.append(emb.squeeze().numpy())
return np.array(embeddings)
def fit(self, images, labels, cnn_model):
print("提取规则特征…")
rule_features = self.extract_rule_features_batch(images)
print("提取CNN嵌入特征…")
cnn_embeddings = self.extract_cnn_embeddings_batch(images, cnn_model)
fusion_features = np.concatenate([rule_features, cnn_embeddings], axis=1)
fusion_scaled = self.scaler.fit_transform(fusion_features)
print(f"融合特征维度: {fusion_scaled.shape}")
self.classifier.fit(fusion_scaled, labels)
train_score = self.classifier.score(fusion_scaled, labels)
print(f"融合分类器训练准确率: {train_score:.2%}")
return self
def predict(self, images, cnn_model):
rule_features = self.extract_rule_features_batch(images)
cnn_embeddings = self.extract_cnn_embeddings_batch(images, cnn_model)
fusion_features = np.concatenate([rule_features, cnn_embeddings], axis=1)
fusion_scaled = self.scaler.transform(fusion_features)
return self.classifier.predict(fusion_scaled)
五、 知识蒸馏：将规则知识注入AI模型
知识蒸馏（Knowledge Distillation）是一种模型压缩技术，其核心思想是学生模型学习教师模型的输出分布。我们可以将规则系统视为一个确定性教师，让AI模型学习规则系统的决策边界，从而将非数据驱动的先验知识注入深度学习模型。
这种方法特别适合制造业场景：当标注数据有限时，规则系统已经积累了大量有效的先验知识。通过知识蒸馏，AI模型可以继承这些知识，在学习过程中获得更好的起点。
class RuleKnowledgeDistiller:
def __init__(self, rule_engine, student_model, temperature=3.0):
self.rule_engine = rule_engine
self.student_model = student_model
self.temperature = temperature
self.soft_targets = []
def generate_soft_targets(self, unlabeled_images):
print(f"生成软标签({len(unlabeled_images)}张图像)…")
soft_targets = []
for image in unlabeled_images:
features = self.rule_engine.extract_handcrafted_features(image)
rule_result = self.rule_engine.rule_based_classify(features)
defect_score = rule_result['defect_score']
soft_label = np.array([1 – defect_score, defect_score])
soft_targets.append(soft_label)
self.soft_targets = np.array(soft_targets)
return self.soft_targets
def distillation_loss(self, student_output, soft_targets, hard_targets=None, alpha=0.7):
soft_student = F.log_softmax(student_output / self.temperature, dim=1)
soft_target = F.softmax(torch.FloatTensor(soft_targets) / self.temperature, dim=1)
distill_loss = F.kl_div(soft_student, soft_target, reduction='batchmean') * (self.temperature ** 2)
if hard_targets is not None:
hard_loss = F.cross_entropy(student_output, torch.LongTensor(hard_targets))
total_loss = alpha * distill_loss + (1 – alpha) * hard_loss
else:
total_loss = distill_loss
return total_loss
六、 在线学习与模型自适应
制造业生产环境的变化（如原材料批次差异、光照条件变化、设备磨损等）会导致检测模型性能衰减。在线学习机制可以让模型在部署后持续适应环境变化。
我们设计了一个主动学习框架，当模型对某个样本的预测不确定性较高时，将该样本发送给人工复核，复核结果用于模型的增量更新。
class OnlineAdaptiveSystem:
def __init__(self, hybrid_system, uncertainty_threshold=0.3):
self.hybrid_system = hybrid_system
self.uncertainty_threshold = uncertainty_threshold
self.review_buffer = []
self.performance_log = []
self.drift_detected = False
def predict_with_uncertainty(self, image):
result = self.hybrid_system.cascade_inspect(image)
score = result['score']
uncertainty = 1.0 – abs(score – 0.5) * 2
needs_review = uncertainty > self.uncertainty_threshold
return result, uncertainty, needs_review
def request_human_review(self, image, prediction):
true_label = np.random.binomial(1, 0.7 + 0.3 * prediction['score'])
return true_label
def update_from_review(self, image, true_label):
features = self.hybrid_system.rule_engine.extract_handcrafted_features(image)
rule_result = self.hybrid_system.rule_engine.rule_based_classify(features)
ai_input = self.hybrid_system._preprocess_for_ai(image)
with torch.no_grad():
ai_output = self.hybrid_system.ai_model(ai_input)
ai_proba = F.softmax(ai_output, dim=1).squeeze().numpy()
self.review_buffer.append({
'features': features,
'rule_score': rule_result['defect_score'],
'ai_proba': ai_proba,
'true_label': true_label
})
if len(self.review_buffer) >= 20:
self._trigger_model_update()
def _trigger_model_update(self):
print(f"触发模型更新(缓冲区大小: {len(self.review_buffer)})…")
review_scores = [d['rule_score'] for d in self.review_buffer]
review_ai = np.array([d['ai_proba'] for d in self.review_buffer])
review_labels = [d['true_label'] for d in self.review_buffer]
avg_rule_error = np.mean(np.abs(np.array(review_scores) – np.array(review_labels)))
avg_ai_error = np.mean(np.abs(review_ai[:, 1] – np.array(review_labels)))
self.performance_log.append({
'timestamp': pd.Timestamp.now(),
'avg_rule_error': avg_rule_error,
'avg_ai_error': avg_ai_error,
'n_reviews': len(self.review_buffer)
})
if avg_rule_error < avg_ai_error:
new_rule_weight = min(0.7, self.hybrid_system.rule_weight + 0.05)
else:
new_rule_weight = max(0.3, self.hybrid_system.rule_weight – 0.05)
self.hybrid_system.update_weights(new_rule_weight)
print(f"权重更新: 规则={self.hybrid_system.rule_weight:.2f}, AI={self.hybrid_system.ai_weight:.2f}")
self.review_buffer = []
七、 异常检测与模型监控
在生产环境中，异常检测是确保系统稳定运行的关键。我们设计了一个多维度的监控系统，追踪规则系统和AI模型的运行状态。
监控指标包括：规则系统的响应时间、AI模型的推理延迟、检测结果的置信度分布、两者的决策一致性等。当监控指标出现异常时，系统自动触发告警。
class SystemMonitor:
def __init__(self, window_size=100):
self.window_size = window_size
self.metrics_history = {
'rule_latency': [],
'ai_latency': [],
'rule_ai_agreement': [],
'confidence': [],
'reject_rate': [],
'score_distribution': []
}
self.baselines = {}
self.alerts = []
def record_inspection(self, rule_latency, ai_latency, rule_decision, ai_decision,
confidence, reject_result, defect_score):
self.metrics_history['rule_latency'].append(rule_latency)
self.metrics_history['ai_latency'].append(ai_latency)
agreement = 1.0 if rule_decision == ai_decision else 0.0
self.metrics_history['rule_ai_agreement'].append(agreement)
self.metrics_history['confidence'].append(confidence)
self.metrics_history['reject_rate'].append(1 if reject_result == 'reject' else 0)
self.metrics_history['score_distribution'].append(defect_score)
for key in self.metrics_history:
if len(self.metrics_history[key]) > self.window_size:
self.metrics_history[key].pop(0)
def compute_baselines(self):
for key, values in self.metrics_history.items():
if len(values) >= 30:
self.baselines[key] = {
'mean': np.mean(values),
'std': np.std(values),
'p95': np.percentile(values, 95)
}
def detect_anomalies(self):
if not self.baselines:
self.compute_baselines()
return []
current_alerts = []
for key in ['rule_latency', 'ai_latency']:
if key in self.baselines and len(self.metrics_history[key]) > 0:
latest = self.metrics_history[key][-1]
baseline = self.baselines[key]
if latest > baseline['mean'] + 3 * baseline['std']:
alert = {
'metric': key,
'value': latest,
'threshold': baseline['mean'] + 3 * baseline['std'],
'severity': 'high',
'message': f"{key}异常: {latest:.2f}ms > {baseline['mean']+3*baseline['std']:.2f}ms"
}
current_alerts.append(alert)
self.alerts.append(alert)
for key in ['rule_ai_agreement', 'confidence']:
if key in self.baselines and len(self.metrics_history[key]) > 0:
recent = self.metrics_history[key][-min(20, len(self.metrics_history[key])):]
avg_recent = np.mean(recent)
baseline = self.baselines[key]
if avg_recent < baseline['mean'] – 2 * baseline['std']:
alert = {
'metric': key,
'value': avg_recent,
'threshold': baseline['mean'] – 2 * baseline['std'],
'severity': 'medium',
'message': f"{key}下降: {avg_recent:.4f} < {baseline['mean']-2*baseline['std']:.4f}"
}
current_alerts.append(alert)
self.alerts.append(alert)
return current_alerts
八、 生产环境部署与工程化
将混合检测系统部署到生产环境需要考虑数据管道、模型服务、API网关、结果持久化等多个工程化问题。我们设计了一套完整的部署方案。
import json
import time
from datetime import datetime
from collections import deque
class ProductionInspectionPipeline:
def __init__(self, hybrid_system, monitor, batch_size=32):
self.hybrid_system = hybrid_system
self.monitor = monitor
self.batch_size = batch_size
self.image_queue = deque(maxlen=1000)
self.result_log = []
self.daily_stats = {
'total': 0, 'pass': 0, 'reject': 0,
'review': 0, 'rule_only': 0, 'ai_only': 0, 'hybrid': 0
}
def enqueue_image(self, image, image_id=None):
self.image_queue.append({
'image': image,
'image_id': image_id or f"IMG_{datetime.now().strftime('%Y%m%d%H%M%S%f')}",
'timestamp': datetime.now()
})
def process_batch(self):
if len(self.image_queue) == 0:
return []
batch_size = min(self.batch_size, len(self.image_queue))
results = []
for _ in range(batch_size):
item = self.image_queue.popleft()
start_time = time.time()
result = self.hybrid_system.cascade_inspect(item['image'])
latency = (time.time() – start_time) * 1000
result['image_id'] = item['image_id']
result['latency_ms'] = round(latency, 2)
result['timestamp'] = item['timestamp'].isoformat()
results.append(result)
self.result_log.append(result)
self.daily_stats['total'] += 1
self.daily_stats[result['decision']] = self.daily_stats.get(result['decision'], 0) + 1
if result['method'] == 'rule_only':
self.daily_stats['rule_only'] += 1
elif result['method'] in ['cascade', 'parallel_vote', 'dynamic_route']:
self.daily_stats['hybrid'] += 1
self.monitor.record_inspection(
rule_latency=np.random.exponential(5),
ai_latency=np.random.exponential(20) if result['method'] != 'rule_only' else 0,
rule_decision=1 if 'rule_score' in result.get('details', {}) and result['details']['rule_score'] > 0.3 else 0,
ai_decision=1 if 'ai_score' in result.get('details', {}) and result['details']['ai_score'] > 0.5 else 0,
confidence=result.get('confidence', 0.5),
reject_result=result['decision'],
defect_score=result['score']
)
if self.daily_stats['total'] % 100 == 0:
