Інференс на пристрої

On-device інференс запускає ML моделі безпосередньо на телефонах, IoT пристроях та embedded системах замість хмари. Це дозволяє real-time відповіді, працює офлайн та зберігає дані приватними. Але edge пристрої мають обмежені обчислення, пам'ять та батарею - тому моделі повинні бути оптимізовані ретельно.

Де використовується

On-device ML забезпечує повсякденний досвід:

• Мобільні застосунки: Фільтри камери, передбачення клавіатури, голосові асистенти
• Wearables: Аналіз серцевого ритму, розпізнавання активності
• Smart home: Wake word detection, локальна автоматизація
• Автомобільний: Моніторинг водія, ADAS функції
• Промисловий IoT: Predictive maintenance, інспекція якості
• Рітейл: Сканування штрих-кодів, відстеження інвентарю
• Охорона здоров'я: Real-time моніторинг, діагностичні допоміжні засоби

Обмеження пристроїв

Що ви робите на практиці

1. Профілюйте цільовий пристрій - CPU, GPU, NPU можливості
2. Встановіть бюджет латентності - Вимоги user experience
3. Стисніть модель - Квантизація, pruning, distillation
4. Оберіть runtime - TFLite, ONNX, CoreML, тощо
5. Оптимізуйте під апаратуру - Використовуйте акселератори
6. Бенчмаркте ретельно - Тестування на реальному пристрої
7. Обробіть edge cases - Низька батарея, background mode
8. Стратегія оновлень - OTA оновлення моделей

Пайплайн оптимізації моделі

import torch
import torch.nn as nn
import numpy as np
from typing import Tuple, Dict, Any

class EdgeModelOptimizer:
    """Комплексна оптимізація для edge деплойменту."""

    def __init__(self, model: nn.Module, target_latency_ms: float = 50):
        self.model = model
        self.target_latency_ms = target_latency_ms
        self.optimization_history = []

    def full_optimization_pipeline(self, 
                                   calibration_data: torch.Tensor,
                                   val_data: Tuple[torch.Tensor, torch.Tensor]
                                   ) -> nn.Module:
        """Запуск повного пайплайну оптимізації."""
        original_size = self.get_model_size()
        original_latency = self.measure_latency()
        original_accuracy = self.evaluate(val_data)

        print(f"Оригінал: {original_size:.2f}MB, {original_latency:.1f}ms, {original_accuracy:.3f}")

        # Крок 1: Prune неважливі ваги
        model = self.structured_pruning(sparsity=0.3)

        # Крок 2: Knowledge distillation (якщо є teacher)
        # model = self.distill(teacher_model, train_data)

        # Крок 3: Квантизація
        model = self.quantize_dynamic()

        # Крок 4: Fusion операцій
        model = self.fuse_operations()

        final_size = self.get_model_size()
        final_latency = self.measure_latency()
        final_accuracy = self.evaluate(val_data)

        print(f"Оптимізовано: {final_size:.2f}MB, {final_latency:.1f}ms, {final_accuracy:.3f}")
        print(f"Зменшення: {(1-final_size/original_size)*100:.1f}% розмір, "
              f"{(1-final_latency/original_latency)*100:.1f}% латентність")

        return model

    def quantize_dynamic(self) -> nn.Module:
        """Застосування динамічної квантизації."""
        quantized = torch.quantization.quantize_dynamic(
            self.model,
            {nn.Linear, nn.LSTM, nn.GRU},
            dtype=torch.qint8
        )
        self.model = quantized
        return quantized

    def quantize_static(self, calibration_data: torch.Tensor) -> nn.Module:
        """Застосування статичної квантизації з калібруванням."""
        self.model.eval()

        # Fuse модулів
        self.model = torch.quantization.fuse_modules(
            self.model, [['conv', 'bn', 'relu']]
        )

        # Підготовка до квантизації
        self.model.qconfig = torch.quantization.get_default_qconfig('qnnpack')
        torch.quantization.prepare(self.model, inplace=True)

        # Калібрування з репрезентативними даними
        with torch.no_grad():
            for batch in calibration_data:
                self.model(batch)

        # Конвертація в квантизовану
        torch.quantization.convert(self.model, inplace=True)

        return self.model

    def structured_pruning(self, sparsity: float = 0.3) -> nn.Module:
        """Видалення структурних компонентів (channels/filters)."""
        import torch.nn.utils.prune as prune

        for name, module in self.model.named_modules():
            if isinstance(module, nn.Conv2d):
                prune.ln_structured(
                    module, name='weight', 
                    amount=sparsity, n=2, dim=0
                )
                prune.remove(module, 'weight')
            elif isinstance(module, nn.Linear):
                prune.l1_unstructured(module, name='weight', amount=sparsity)
                prune.remove(module, 'weight')

        return self.model

    def get_model_size(self) -> float:
        """Отримання розміру моделі в MB."""
        param_size = sum(
            p.nelement() * p.element_size() 
            for p in self.model.parameters()
        )
        buffer_size = sum(
            b.nelement() * b.element_size() 
            for b in self.model.buffers()
        )
        return (param_size + buffer_size) / (1024 ** 2)

    def measure_latency(self, input_shape: Tuple = (1, 3, 224, 224),
                        warmup: int = 10, runs: int = 100) -> float:
        """Вимірювання латентності інференсу в ms."""
        self.model.eval()
        dummy_input = torch.randn(*input_shape)

        # Warmup
        with torch.no_grad():
            for _ in range(warmup):
                self.model(dummy_input)

        # Вимірювання
        import time
        times = []
        with torch.no_grad():
            for _ in range(runs):
                start = time.perf_counter()
                self.model(dummy_input)
                times.append((time.perf_counter() - start) * 1000)

        return np.median(times)

    def evaluate(self, val_data: Tuple[torch.Tensor, torch.Tensor]) -> float:
        """Оцінка точності моделі."""
        self.model.eval()
        X, y = val_data
        with torch.no_grad():
            outputs = self.model(X)
            preds = outputs.argmax(dim=1)
            accuracy = (preds == y).float().mean().item()
        return accuracy

Стратегії квантизації

import torch
import torch.nn as nn
from torch.quantization import QuantStub, DeQuantStub

class QuantizableModel(nn.Module):
    """Модель підготовлена для квантизації."""

    def __init__(self, original_model: nn.Module):
        super().__init__()
        self.quant = QuantStub()
        self.model = original_model
        self.dequant = DeQuantStub()

    def forward(self, x):
        x = self.quant(x)
        x = self.model(x)
        x = self.dequant(x)
        return x

class QuantizationAwareTraining:
    """QAT для кращої точності з квантизацією."""

    def __init__(self, model: nn.Module, backend: str = 'qnnpack'):
        self.model = model
        self.backend = backend

    def prepare_qat(self):
        """Підготовка моделі для quantization-aware training."""
        self.model.train()

        # Встановлення quantization config
        self.model.qconfig = torch.quantization.get_default_qat_qconfig(
            self.backend
        )

        # Fuse операцій
        self.model = torch.quantization.fuse_modules(
            self.model, 
            self._get_fusion_patterns()
        )

        # Підготовка для QAT
        torch.quantization.prepare_qat(self.model, inplace=True)

        return self.model

    def train_qat(self, train_loader, epochs: int = 5):
        """Тренування з quantization awareness."""
        optimizer = torch.optim.Adam(self.model.parameters(), lr=1e-4)
        criterion = nn.CrossEntropyLoss()

        for epoch in range(epochs):
            self.model.train()

            for batch_x, batch_y in train_loader:
                optimizer.zero_grad()
                outputs = self.model(batch_x)
                loss = criterion(outputs, batch_y)
                loss.backward()
                optimizer.step()

        return self.model

    def convert_to_quantized(self):
        """Конвертація в повністю квантизовану модель."""
        self.model.eval()
        quantized = torch.quantization.convert(self.model, inplace=False)
        return quantized

class MixedPrecisionInference:
    """Різна точність для різних шарів."""

    def __init__(self, model: nn.Module):
        self.model = model
        self.layer_precision = {}

    def analyze_sensitivity(self, val_data, test_precisions=[8, 16, 32]):
        """Пошук оптимальної точності на шар."""
        baseline_accuracy = self._evaluate(val_data)

        for name, module in self.model.named_modules():
            if isinstance(module, (nn.Conv2d, nn.Linear)):
                sensitivities = {}

                for bits in test_precisions:
                    self._quantize_layer(name, bits)
                    accuracy = self._evaluate(val_data)
                    accuracy_drop = baseline_accuracy - accuracy
                    sensitivities[bits] = accuracy_drop
                    self._restore_layer(name)

                # Вибір мінімальної точності що підтримує accuracy
                for bits in test_precisions:
                    if sensitivities[bits] < 0.01:  # < 1% drop
                        self.layer_precision[name] = bits
                        break
                else:
                    self.layer_precision[name] = 32

        return self.layer_precision

Вибір Runtime

from enum import Enum
from typing import Optional

class MLRuntime(Enum):
    TFLITE = "tflite"
    ONNX = "onnx"
    COREML = "coreml"
    PYTORCH_MOBILE = "pytorch_mobile"
    TENSORRT = "tensorrt"

class RuntimeSelector:
    """Вибір оптимального runtime для цільового пристрою."""

    RUNTIME_CAPABILITIES = {
        MLRuntime.TFLITE: {
            'platforms': ['android', 'ios', 'linux', 'windows'],
            'accelerators': ['cpu', 'gpu', 'npu', 'dsp'],
            'quantization': ['int8', 'float16'],
            'model_size_limit_mb': 100
        },
        MLRuntime.COREML: {
            'platforms': ['ios', 'macos'],
            'accelerators': ['cpu', 'gpu', 'neural_engine'],
            'quantization': ['int8', 'float16'],
            'model_size_limit_mb': 200
        },
        MLRuntime.ONNX: {
            'platforms': ['android', 'ios', 'linux', 'windows'],
            'accelerators': ['cpu', 'gpu'],
            'quantization': ['int8', 'float16', 'float32'],
            'model_size_limit_mb': 500
        },
        MLRuntime.TENSORRT: {
            'platforms': ['linux', 'jetson'],
            'accelerators': ['gpu'],
            'quantization': ['int8', 'float16', 'float32'],
            'model_size_limit_mb': 1000
        }
    }

    def select_runtime(self, target_platform: str,
                       model_size_mb: float,
                       required_accelerator: str = 'cpu',
                       quantization: str = 'float32') -> MLRuntime:
        """Вибір найкращого runtime для вимог."""
        candidates = []

        for runtime, caps in self.RUNTIME_CAPABILITIES.items():
            if target_platform in caps['platforms'] and \
               required_accelerator in caps['accelerators'] and \
               quantization in caps['quantization'] and \
               model_size_mb <= caps['model_size_limit_mb']:
                candidates.append(runtime)

        if not candidates:
            raise ValueError(f"Жоден runtime не підтримує вимоги")

        # Пріоритет: native platform runtime > general purpose
        if target_platform == 'ios' and MLRuntime.COREML in candidates:
            return MLRuntime.COREML
        if target_platform == 'android' and MLRuntime.TFLITE in candidates:
            return MLRuntime.TFLITE
        if required_accelerator == 'gpu' and MLRuntime.TENSORRT in candidates:
            return MLRuntime.TENSORRT

        return candidates[0]

Battery-Aware інференс

import time
from dataclasses import dataclass
from typing import Callable, Dict

@dataclass
class InferenceConfig:
    """Конфігурація для адаптивного інференсу."""
    max_latency_ms: float = 100
    battery_threshold: float = 0.2  # 20%
    thermal_threshold_c: float = 40
    use_accelerator: bool = True

class BatteryAwareInference:
    """Адаптація інференсу на основі стану пристрою."""

    def __init__(self, models: Dict[str, nn.Module], config: InferenceConfig):
        """
        Ініціалізація з кількома варіантами моделі.

        Args:
            models: Dict з 'full', 'medium', 'lite' варіантами
            config: Конфігурація інференсу
        """
        self.models = models
        self.config = config
        self.current_model = 'full'

    def get_device_state(self) -> Dict:
        """Отримання поточного стану пристрою (mock implementation)."""
        return {
            'battery_level': 0.75,
            'is_charging': False,
            'temperature_c': 35,
            'memory_available_mb': 512
        }

    def select_model(self) -> nn.Module:
        """Вибір варіанту моделі на основі стану пристрою."""
        state = self.get_device_state()

        # Низька батарея - використовуємо lite модель
        if state['battery_level'] < self.config.battery_threshold:
            self.current_model = 'lite'

        # Thermal throttling - зменшуємо до medium
        elif state['temperature_c'] > self.config.thermal_threshold_c:
            self.current_model = 'medium'

        # Хороші умови - використовуємо повну модель
        else:
            self.current_model = 'full'

        return self.models[self.current_model]

    def infer(self, input_tensor: torch.Tensor) -> torch.Tensor:
        """Запуск інференсу з адаптивним вибором моделі."""
        model = self.select_model()
        model.eval()

        with torch.no_grad():
            output = model(input_tensor)

        return output

class BatchedInference:
    """Групування запитів для ефективного інференсу."""

    def __init__(self, model: nn.Module, 
                 max_batch_size: int = 8,
                 max_wait_ms: float = 50):
        self.model = model
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        self.pending_requests = []

    def add_request(self, input_tensor: torch.Tensor, 
                    callback: Callable):
        """Додавання запиту інференсу до batch."""
        self.pending_requests.append({
            'input': input_tensor,
            'callback': callback,
            'timestamp': time.time()
        })

        if self._should_process():
            self._process_batch()

    def _should_process(self) -> bool:
        """Перевірка чи batch повинен бути оброблений."""
        if len(self.pending_requests) >= self.max_batch_size:
            return True

        if self.pending_requests:
            oldest = self.pending_requests[0]['timestamp']
            if (time.time() - oldest) * 1000 > self.max_wait_ms:
                return True

        return False

    def _process_batch(self):
        """Обробка всіх pending запитів."""
        if not self.pending_requests:
            return

        inputs = torch.stack([r['input'] for r in self.pending_requests])

        self.model.eval()
        with torch.no_grad():
            outputs = self.model(inputs)

        for i, request in enumerate(self.pending_requests):
            request'callback'

        self.pending_requests.clear()

Метрики оцінки

Типові помилки та чек-лист

• [ ] Деградація точності: Занадто агресивна оптимізація
• [ ] Сплески латентності: Thermal throttling, GC
• [ ] Overflow пам'яті: Модель занадто велика
• [ ] Розряд батареї: Безперервний інференс
• [ ] Несумісні ops: Непідтримувані оператори
• [ ] Фрагментація пристроїв: Різна апаратура
• [ ] Збої оновлень: OTA оновлення моделей

Чек-лист для продакшену

Перед деплоєм: - [ ] Benchmark на цільових пристроях - [ ] Тест впливу на батарею - [ ] Валідація точності після оптимізації - [ ] Тест офлайн роботи - [ ] Визначення fallback поведінки

Після деплою: - [ ] Моніторинг розподілу латентності - [ ] Відстеження звітів про батарею - [ ] A/B тест оновлень моделі - [ ] Тестування сумісності пристроїв - [ ] Метрики user experience

Інструменти та стек

Приклад міні-кейсу

Сценарій: Real-time детекція об'єктів на mobile.

Підхід: - MobileNetV3 + SSD архітектура - INT8 квантизація з QAT - TFLite з GPU delegate - Адаптивна роздільність на основі батареї

Результати: - 15ms латентність на mid-range телефоні - 4.2MB розмір моделі - < 1% падіння точності від FP32 - 40% зменшення споживання vs CPU

Пов'язані сторінки

• Індекс: ML Field Guide
• Стиснення моделей
• Privacy-Preserving ML
• Детекція об'єктів

Додаткові матеріали

• TensorFlow Lite Guide
• PyTorch Mobile
• Efficient Deep Learning

Спробуйте наживо

Дослідіть пов'язану модель на практиці: відкрийте Personal AI Assistants: On-Device vs Cloud LLM Inference Simulator і змінюйте параметри під час роботи. Усе лишається у вашому браузері.