SaaS 行业观察：边缘计算与离线 AI 能力的崛起

class RealTimeQualityControl:
    def __init__(self):
        self.edge_model = load_edge_model('defect_detection_v3')
        self.camera = IndustrialCamera(fps=120)
        self.actuator = RoboticArm()
    
    def inspect_product(self):
        """
        实时产品质量检测
        要求：从拍摄到决策 < 10ms
        """
        # 捕获图像
        frame = self.camera.capture()
        
        # 边缘推理（本地模型）
        start_time = time.time()
        prediction = self.edge_model.predict(frame)
        inference_time = (time.time() - start_time) * 1000  # 毫秒
        
        # 实时决策
        if prediction['defect_probability'] > 0.8:
            self.actuator.reject_product()
            log_defect(prediction)
        else:
            self.actuator.accept_product()
        
        # 记录性能指标
        return {
            'inference_time_ms': inference_time,
            'defect_detected': prediction['defect_probability'] > 0.8,
            'confidence': prediction['confidence']
        }

class OnDeviceMedicalAI:
    def __init__(self):
        # 模型完全在设备上运行
        self.diagnosis_model = load_local_model('medical_diagnosis_v2')
        self.patient_data_store = EncryptedLocalStorage()
    
    def analyze_medical_image(self, image, patient_id):
        """
        在设备本地分析医学影像
        数据永远不会离开设备
        """
        # 数据加密存储
        encrypted_image = self.patient_data_store.encrypt(image)
        self.patient_data_store.store(patient_id, encrypted_image)
        
        # 本地推理
        with SecureEnclave() as enclave:
            # 在安全区域内解密和处理
            decrypted_image = enclave.decrypt(encrypted_image)
            diagnosis = self.diagnosis_model.analyze(decrypted_image)
            
            # 立即清除解密数据
            enclave.wipe(decrypted_image)
        
        # 只返回诊断结果，不传输原始数据
        return {
            'diagnosis': diagnosis['condition'],
            'confidence': diagnosis['confidence'],
            'recommendations': diagnosis['recommendations'],
            'timestamp': datetime.now()
        }

class OfflineGeologicalAnalysis:
    def __init__(self):
        self.local_models = {
            'rock_classification': load_model('rock_classifier_v4'),
            'mineral_detection': load_model('mineral_detector_v2'),
            'terrain_mapping': load_model('terrain_mapper_v3')
        }
        self.local_database = OfflineDatabase()
        self.sync_queue = SyncQueue()
    
    def analyze_rock_sample(self, sample_image, location):
        """
        在野外无网络环境下分析岩石样本
        """
        # 本地分析
        rock_type = self.local_models['rock_classification'].predict(sample_image)
        minerals = self.local_models['mineral_detection'].detect(sample_image)
        terrain = self.local_models['terrain_mapping'].analyze(location)
        
        # 存储到本地数据库
        analysis_result = {
            'timestamp': datetime.now(),
            'location': location,
            'rock_type': rock_type,
            'minerals': minerals,
            'terrain': terrain,
            'synced': False
        }
        
        self.local_database.store(analysis_result)
        self.sync_queue.add(analysis_result)
        
        return analysis_result
    
    def sync_when_connected(self):
        """
        当网络恢复时同步数据
        """
        if not network_available():
            return
        
        unsynced_data = self.sync_queue.get_all()
        
        for data in unsynced_data:
            try:
                # 压缩并加密
                compressed = compress(data)
                encrypted = encrypt(compressed)
                
                # 上传到云端
                cloud_client.upload(encrypted)
                
                # 标记为已同步
                data['synced'] = True
                self.local_database.update(data)
                self.sync_queue.remove(data)
                
            except UploadError as e:
                log_error(f"Sync failed: {e}")
                # 保留在队列中，下次重试

class EdgeVideoAnalytics:
    def __init__(self):
        self.edge_detector = load_model('object_detector_v5')
        self.cloud_analyzer = CloudAPIClient()
    
    def process_video_stream(self, video_stream):
        """
        边缘预处理 + 云端深度分析
        """
        for frame in video_stream:
            # 边缘检测：识别是否有趣的事件
            objects = self.edge_detector.detect(frame)
            
            # 只在检测到有趣事件时上传
            if self.is_interesting_event(objects):
                # 压缩并上传关键帧
                key_frame = extract_key_frame(frame, objects)
                compressed_frame = compress(key_frame, quality=0.7)
                
                # 上传到云端进行深度分析
                cloud_result = self.cloud_analyzer.analyze(compressed_frame)
                
                yield {
                    'timestamp': frame.timestamp,
                    'edge_detection': objects,
                    'cloud_analysis': cloud_result,
                    'data_transferred': len(compressed_frame)
                }
            else:
                # 不上传，节省带宽
                yield {
                    'timestamp': frame.timestamp,
                    'edge_detection': objects,
                    'data_transferred': 0
                }
    
    def is_interesting_event(self, objects):
        """
        判断是否是有趣的事件
        """
        # 检测到人、车辆或异常行为
        interesting_classes = ['person', 'vehicle', 'animal']
        
        for obj in objects:
            if obj['class'] in interesting_classes:
                if obj['confidence'] > 0.7:
                    return True
        
        return False

class ModelQuantizer:
    def __init__(self):
        self.quantization_bits = 8  # INT8 量化
    
    def quantize_model(self, model):
        """
        将浮点模型量化为整数模型
        减少模型大小 4-8 倍，加速推理 2-4 倍
        """
        # 分析权重分布
        weight_stats = self.analyze_weights(model)
        
        # 计算量化参数
        scale, zero_point = self.calculate_quantization_params(weight_stats)
        
        # 量化权重
        quantized_weights = {}
        for layer_name, weights in model.weights.items():
            quantized_weights[layer_name] = self.quantize_tensor(
                weights, scale, zero_point
            )
        
        # 创建量化模型
        quantized_model = QuantizedModel(
            architecture=model.architecture,
            weights=quantized_weights,
            scale=scale,
            zero_point=zero_point
        )
        
        return quantized_model
    
    def quantize_tensor(self, tensor, scale, zero_point):
        """
        量化单个张量
        """
        # 量化公式：q = round(x / scale) + zero_point
        quantized = np.round(tensor / scale) + zero_point
        
        # 裁剪到 INT8 范围
        quantized = np.clip(quantized, -128, 127)
        
        return quantized.astype(np.int8)

class KnowledgeDistillation:
    def __init__(self, teacher_model, student_model):
        self.teacher = teacher_model  # 大型云端模型
        self.student = student_model  # 小型边缘模型
        self.temperature = 4.0  # 软化概率分布
    
    def distill(self, training_data, epochs=10):
        """
        通过知识蒸馏训练小型边缘模型
        """
        optimizer = Adam(self.student.parameters())
        
        for epoch in range(epochs):
            for batch in training_data:
                inputs, true_labels = batch
                
                # 教师模型的软标签
                with torch.no_grad():
                    teacher_logits = self.teacher(inputs)
                    teacher_probs = F.softmax(teacher_logits / self.temperature, dim=1)
                
                # 学生模型的预测
                student_logits = self.student(inputs)
                student_probs = F.softmax(student_logits / self.temperature, dim=1)
                
                # 蒸馏损失：学生模仿教师的软概率分布
                distillation_loss = F.kl_div(
                    F.log_softmax(student_logits / self.temperature, dim=1),
                    teacher_probs,
                    reduction='batchmean'
                ) * (self.temperature ** 2)
                
                # 真实标签损失
                true_loss = F.cross_entropy(student_logits, true_labels)
                
                # 总损失（加权组合）
                alpha = 0.7  # 蒸馏损失权重
                total_loss = alpha * distillation_loss + (1 - alpha) * true_loss
                
                # 反向传播
                optimizer.zero_grad()
                total_loss.backward()
                optimizer.step()
        
        return self.student

class ModelPruner:
    def __init__(self, pruning_ratio=0.5):
        self.pruning_ratio = pruning_ratio
    
    def prune_model(self, model):
        """
        剪枝：移除不重要的权重
        减少模型大小 50-90%
        """
        pruned_model = copy.deepcopy(model)
        
        for layer_name, layer in pruned_model.layers.items():
            if isinstance(layer, (Conv2d, Linear)):
                # 计算权重重要性（基于幅度）
                weight_magnitude = torch.abs(layer.weight.data)
                
                # 确定剪枝阈值
                threshold = self.calculate_threshold(
                    weight_magnitude, 
                    self.pruning_ratio
                )
                
                # 创建掩码
                mask = weight_magnitude > threshold
                
                # 应用掩码（将不重要的权重设为 0）
                layer.weight.data *= mask
                
                # 存储掩码用于稀疏计算
                layer.register_buffer('mask', mask)
        
        return pruned_model
    
    def calculate_threshold(self, magnitude, ratio):
        """
        计算剪枝阈值
        """
        # 按幅度排序
        sorted_magnitude = torch.sort(magnitude.flatten())[0]
        
        # 找到阈值位置
        threshold_index = int(len(sorted_magnitude) * ratio)
        threshold = sorted_magnitude[threshold_index]
        
        return threshold

class TFLiteEdgeInference:
    def __init__(self, model_path):
        # 加载 TFLite 模型
        self.interpreter = tf.lite.Interpreter(model_path=model_path)
        self.interpreter.allocate_tensors()
        
        # 获取输入输出张量信息
        self.input_details = self.interpreter.get_input_details()
        self.output_details = self.interpreter.get_output_details()
    
    def predict(self, input_data):
        """
        在边缘设备上执行推理
        """
        # 预处理输入数据
        processed_input = self.preprocess(input_data)
        
        # 设置输入张量
        self.interpreter.set_tensor(
            self.input_details[0]['index'], 
            processed_input
        )
        
        # 执行推理
        start_time = time.time()
        self.interpreter.invoke()
        inference_time = time.time() - start_time
        
        # 获取输出
        output = self.interpreter.get_tensor(
            self.output_details[0]['index']
        )
        
        # 后处理
        result = self.postprocess(output)
        
        return {
            'prediction': result,
            'inference_time_ms': inference_time * 1000
        }
    
    def preprocess(self, input_data):
        """
        预处理输入数据以匹配模型要求
        """
        # 调整大小
        input_shape = self.input_details[0]['shape']
        resized = tf.image.resize(input_data, input_shape[1:3])
        
        # 归一化
        normalized = resized / 255.0
        
        # 添加批次维度
        batched = tf.expand_dims(normalized, 0)
        
        return batched.numpy()

class ONNXEdgeInference:
    def __init__(self, model_path, device='cpu'):
        # 创建推理会话
        self.session = ort.InferenceSession(
            model_path,
            providers=['CPUExecutionProvider'] if device == 'cpu' 
                     else ['CUDAExecutionProvider']
        )
        
        self.input_name = self.session.get_inputs()[0].name
        self.output_name = self.session.get_outputs()[0].name
    
    def predict(self, input_data):
        """
        使用 ONNX Runtime 执行推理
        支持多种硬件加速（CPU、GPU、NPU）
        """
        # 准备输入
        inputs = {self.input_name: input_data}
        
        # 执行推理
        start_time = time.time()
        outputs = self.session.run([self.output_name], inputs)
        inference_time = time.time() - start_time
        
        return {
            'prediction': outputs[0],
            'inference_time_ms': inference_time * 1000
        }

class HybridInferenceSystem:
    def __init__(self):
        self.edge_model = load_edge_model('lightweight_v3')
        self.cloud_model = CloudAIClient('high_accuracy_v5')
        self.confidence_threshold = 0.7
    
    def predict(self, input_data):
        """
        混合推理：边缘优先，云端补充
        """
        # 第一步：边缘推理
        edge_result = self.edge_model.predict(input_data)
        
        # 检查置信度
        if edge_result['confidence'] >= self.confidence_threshold:
            # 高置信度：使用边缘结果
            return {
                'source': 'edge',
                'prediction': edge_result['prediction'],
                'confidence': edge_result['confidence'],
                'latency_ms': edge_result['inference_time_ms']
            }
        else:
            # 低置信度：回退到云端
            try:
                cloud_result = self.cloud_model.predict(input_data)
                
                return {
                    'source': 'cloud',
                    'prediction': cloud_result['prediction'],
                    'confidence': cloud_result['confidence'],
                    'latency_ms': cloud_result['latency_ms'],
                    'edge_confidence': edge_result['confidence']
                }
            
            except CloudAPIError:
                # 云端不可用：使用边缘结果
                return {
                    'source': 'edge_fallback',
                    'prediction': edge_result['prediction'],
                    'confidence': edge_result['confidence'],
                    'latency_ms': edge_result['inference_time_ms'],
                    'warning': 'Cloud unavailable, using low-confidence edge result'
                }

class AdaptiveModelSelector:
    def __init__(self):
        self.models = {
            'nano': load_model('model_nano'),      # 最小，最快
            'small': load_model('model_small'),    # 小型
            'medium': load_model('model_medium'),  # 中型
            'large': load_model('model_large')     # 最大，最准确
        }
        
        self.performance_monitor = PerformanceMonitor()
    
    def select_model(self, input_data, constraints):
        """
        根据约束条件自适应选择模型
        """
        # 评估当前设备状态
        device_state = self.performance_monitor.get_device_state()
        
        # 根据约束过滤候选模型
        candidates = []
        for model_name, model in self.models.items():
            # 检查内存约束
            if model.memory_requirement > device_state['available_memory']:
                continue
            
            # 检查延迟约束
            estimated_latency = self.estimate_latency(model, device_state)
            if 'max_latency_ms' in constraints:
                if estimated_latency > constraints['max_latency_ms']:
                    continue
            
            # 检查电量约束
            if 'min_battery_level' in constraints:
                if device_state['battery_level'] < constraints['min_battery_level']:
                    # 电量低时只考虑最轻量的模型
                    if model_name not in ['nano', 'small']:
                        continue
            
            candidates.append({
                'name': model_name,
                'model': model,
                'estimated_latency': estimated_latency,
                'accuracy': model.expected_accuracy
            })
        
        # 选择最佳候选（平衡准确性和延迟）
        if not candidates:
            # 没有满足约束的模型，使用最小模型
            return self.models['nano']
        
        # 根据优先级选择
        if constraints.get('priority') == 'accuracy':
            best = max(candidates, key=lambda c: c['accuracy'])
        elif constraints.get('priority') == 'latency':
            best = min(candidates, key=lambda c: c['estimated_latency'])
        else:
            # 平衡模式：使用综合评分
            best = max(candidates, key=lambda c: self.calculate_score(c))
        
        return best['model']
    
    def calculate_score(self, candidate):
        """
        计算综合评分
        """
        # 归一化准确性和延迟
        accuracy_score = candidate['accuracy']
        latency_score = 1 / (1 + candidate['estimated_latency'] / 100)
        
        # 加权评分（准确性权重 0.6，延迟权重 0.4）
        score = 0.6 * accuracy_score + 0.4 * latency_score
        
        return score

class IncrementalSyncManager:
    def __init__(self, local_db, cloud_client):
        self.local_db = local_db
        self.cloud_client = cloud_client
        self.sync_state = SyncState()
    
    def sync(self):
        """
        增量同步：只同步变更的数据
        """
        if not network_available():
            return {'status': 'offline', 'synced': 0}
        
        # 获取自上次同步以来的变更
        last_sync_time = self.sync_state.get_last_sync_time()
        changes = self.local_db.get_changes_since(last_sync_time)
        
        synced_count = 0
        
        for change in changes:
            try:
                if change['type'] == 'insert':
                    # 新增数据
                    self.cloud_client.insert(change['data'])
                
                elif change['type'] == 'update':
                    # 更新数据
                    self.cloud_client.update(
                        change['id'], 
                        change['data']
                    )
                
                elif change['type'] == 'delete':
                    # 删除数据
                    self.cloud_client.delete(change['id'])
                
                synced_count += 1
                
            except SyncError as e:
                log_error(f"Failed to sync change {change['id']}: {e}")
                # 继续同步其他变更
        
        # 更新同步状态
        self.sync_state.update_last_sync_time(datetime.now())
        
        return {
            'status': 'success',
            'synced': synced_count,
            'total_changes': len(changes)
        }

class ConflictResolver:
    def __init__(self):
        self.resolution_strategies = {
            'last_write_wins': self.last_write_wins,
            'merge': self.merge_changes,
            'manual': self.require_manual_resolution
        }
    
    def resolve_conflict(self, local_data, cloud_data):
        """
        解决本地和云端数据冲突
        """
        # 检测冲突类型
        conflict_type = self.detect_conflict_type(local_data, cloud_data)
        
        # 选择解决策略
        if conflict_type == 'concurrent_update':
            # 并发更新：使用最后写入获胜
            strategy = 'last_write_wins'
        
        elif conflict_type == 'field_level_conflict':
            # 字段级冲突：尝试合并
            strategy = 'merge'
        
        else:
            # 复杂冲突：需要人工解决
            strategy = 'manual'
        
        # 应用策略
        resolver = self.resolution_strategies[strategy]
        resolved_data = resolver(local_data, cloud_data)
        
        return resolved_data
    
    def last_write_wins(self, local_data, cloud_data):
        """
        最后写入获胜策略
        """
        if local_data['updated_at'] > cloud_data['updated_at']:
            return local_data
        else:
            return cloud_data
    
    def merge_changes(self, local_data, cloud_data):
        """
        合并变更策略
        """
        merged = {}
        
        # 遍历所有字段
        all_fields = set(local_data.keys()) | set(cloud_data.keys())
        
        for field in all_fields:
            local_value = local_data.get(field)
            cloud_value = cloud_data.get(field)
            
            if local_value == cloud_value:
                # 相同值：直接使用
                merged[field] = local_value
            
            elif local_value is None:
                # 本地没有修改：使用云端值
                merged[field] = cloud_value
            
            elif cloud_value is None:
                # 云端没有修改：使用本地值
                merged[field] = local_value
            
            else:
                # 都有修改：检查字段类型
                if isinstance(local_value, (list, dict)):
                    # 复杂类型：需要深度合并
                    merged[field] = self.deep_merge(local_value, cloud_value)
                else:
                    # 简单类型：使用更新的值
                    if local_data['updated_at'] > cloud_data['updated_at']:
                        merged[field] = local_value
                    else:
                        merged[field] = cloud_value
        
        return merged