엣지 장치의 딥 러닝: 클라우드에서 엣지까지
ChatGPT에 대한 각 요청 비용은 약 $0.002입니다. 수십억 개의 요청을 곱함 매일 AI의 클라우드 비용은 천문학적으로 증가합니다. 하지만 대안이 있습니다. 모델 장치에서 직접 사용자의. Gartner는 2027년까지 이를 예측 기기에서 실행되는 모델은 사용 빈도가 클라우드 모델보다 3배 더 뛰어납니다. 운영 비용 70% 절감. 이것이 패러다임이다엣지 AI.
Raspberry Pi 5, NVIDIA Jetson Orin, Apple Neural Engine, Qualcomm NPU — 2026년 및 연도 엣지 하드웨어는 언어 모델을 실행할 수 있을 만큼 강력해졌습니다. 10억~70억 개의 매개변수와 경쟁력 있는 비전 모델. 도전은 더 이상 '가능하다'가 아니다. 그러나 "실제 제약에 따라 배포를 최적화하는 방법": 제한된 RAM, 이기종 CPU/GPU, 전력 소비, 온도, 오프라인 연결.
이 가이드에서는 하드웨어 선택부터 엣지 배포 파이프라인 전체를 다룹니다. ONNX 변환부터 Raspberry Pi 배포까지 대상에서 모델까지 최적화 실제 벤치마크, 모범 사례 및 포괄적인 사례 연구를 포함하는 Jetson Nano/Orin.
무엇을 배울 것인가
- 2026년 Edge 하드웨어 개요: Raspberry Pi 5, Jetson Orin, Coral, 모바일 NPU
- 에지 최적화 파이프라인: 양자화 + 가지치기 + 증류
- 특정 최적화를 통해 ARM CPU에 ONNX 런타임을 사용한 배포
- TensorFlow Lite: Raspberry Pi의 경량 추론
- NVIDIA Jetson: 실시간 비전을 위한 CUDA, TensorRT 및 DeepStream
- Raspberry Pi의 llama.cpp: GGUF를 사용한 LLM 에지
- 경량 FastAPI를 제공하는 REST 모델
- 벤치마크: 대기 시간, 처리량, 전력 소비
- 모니터링, 열 관리 및 OTA 모델 업데이트
엣지 하드웨어 개요(2025~2026년)
엣지 하드웨어의 선택은 작업, 예산, 배포 요구 사항에 따라 달라집니다. 2026년 시장은 보급형(Raspberry Pi부터 €60)부터 고급형까지(Jetson AGX Orin €1000+). 전체 개요는 다음과 같습니다.
| 장치 | CPU/GPU | 숫양 | AI 성능 | 비용 | 사용 사례 |
|---|---|---|---|---|---|
| 라즈베리 파이 5 | Cortex-A76(4코어, 2.4GHz) | 4~8GB | ~13GFLOPS CPU | ~60-80€ | 소규모 LLM, 라이트 비전, IoT AI |
| 라즈베리 파이 4 | Cortex-A72(4코어, 1.8GHz) | 2~8GB | ~8GFLOPS CPU | ~35-75€ | 기본 추론, 분류 |
| 엔비디아 젯슨 나노 | Maxwell GPU 128 코어 + Cortex-A57 | 4GB 공유 | 472GFLOPS | ~100€ | 비전, 실시간 감지(레거시) |
| 엔비디아 젯슨 오린 NX | 암페어 GPU 1024 코어 + Cortex-A78AE | 8~16GB | 70-100 탑 | ~500-700€ | LLM 7B, 고급 비전, 로봇공학 |
| 엔비디아 젯슨 AGX 오린 | Ampere GPU 2048 코어 + 12 CPU 코어 | 32~64GB | 275탑스 | ~1000-2000€ | LLM 13B, 다중 모델 추론 |
| 구글 코럴USB | 엣지 TPU | 해당 없음(호스트 RAM) | 4탑스 INT8 | ~60€ | 최적화된 INT8 추론(소형 모델) |
| 인텔 신경망 컴퓨팅 스틱 2 | 무수한 | 4GB LPDDR4 | 4개의 탑 | ~85€ | 비전, 물체 감지, OpenVINO |
| 퀄컴 RB5 / AI 키트 | 크라이오 CPU + 아드레노 GPU + 헥사곤 DSP | 8GB | 탑 15개 | ~300€ | 모바일 AI, 최적화된 NPU 추론 |
엣지 최적화 파이프라인
일반적으로 클라우드 환경에서 개발되는 모델은 직접 배포가 불가능합니다. 최적화 없이 가장자리에 있습니다. 표준 파이프라인에는 일련의 변환이 포함됩니다. 허용 가능한 정확도를 유지하면서 크기와 대기 시간을 점진적으로 줄입니다.
# Pipeline completo: da modello PyTorch a edge deployment
import torch
import torch.nn as nn
from torchvision import models
import time
# Step 1: Baseline model (sviluppato su cloud/GPU)
# ResNet-50: 25M param, 98 MB, ~4ms su RTX 3090
model_cloud = models.resnet50(pretrained=True)
model_cloud.fc = nn.Linear(2048, 10) # 10 classi custom
# Funzioni di utilita
def model_size_mb(model):
"""Calcola dimensione modello in MB."""
total_params = sum(p.numel() * p.element_size() for p in model.parameters())
return total_params / (1024 ** 2)
def count_params(model):
return sum(p.numel() for p in model.parameters())
def measure_latency(model, input_size=(1, 3, 224, 224), n_warmup=10, n_runs=50):
"""Misura latenza media di inferenza in ms."""
model.eval()
dummy = torch.randn(*input_size)
with torch.no_grad():
for _ in range(n_warmup):
model(dummy)
times = []
for _ in range(n_runs):
t0 = time.perf_counter()
model(dummy)
times.append((time.perf_counter() - t0) * 1000)
return sum(times) / len(times)
print("=== BASELINE MODEL ===")
print(f"ResNet-50: {model_size_mb(model_cloud):.1f} MB, "
f"{count_params(model_cloud)/1e6:.1f}M params")
# ================================================================
# STEP 2: DISTILLAZIONE -> Student più piccolo
# (Teacher: ResNet-50, Student: MobileNetV3-Small)
# ================================================================
student = models.mobilenet_v3_small(pretrained=False)
student.classifier[3] = nn.Linear(student.classifier[3].in_features, 10)
print("\n=== AFTER DISTILLATION ===")
print(f"MobileNetV3-S: {model_size_mb(student):.1f} MB, "
f"{count_params(student)/1e6:.1f}M params")
print(f"Riduzione: {model_size_mb(model_cloud)/model_size_mb(student):.1f}x")
# ================================================================
# STEP 3: PRUNING (rimuovi 30% dei pesi meno importanti)
# ================================================================
import torch.nn.utils.prune as prune
# Pruning strutturato: rimuove interi canali
def apply_structured_pruning(model, amount: float = 0.3):
"""Applica pruning L1 strutturato a tutti i layer Conv2d."""
for name, module in model.named_modules():
if isinstance(module, nn.Conv2d) and module.out_channels > 8:
prune.ln_structured(module, name='weight', amount=amount,
n=1, dim=0) # Dim 0 = output channels
return model
student_pruned = apply_structured_pruning(student, amount=0.2)
print(f"\n=== AFTER PRUNING (20%) ===")
print(f"MobileNetV3-S pruned: ~{model_size_mb(student)*0.8:.1f} MB (stima)")
# ================================================================
# STEP 4: QUANTIZZAZIONE INT8 (post-training)
# ================================================================
student.eval()
# Quantizzazione dinamica (più semplice, applicabile subito)
student_ptq = torch.quantization.quantize_dynamic(
student,
{nn.Linear},
dtype=torch.qint8
)
print(f"\n=== AFTER INT8 QUANTIZATION ===")
print(f"MobileNetV3-S INT8: ~{model_size_mb(student)/4:.1f} MB (stima)")
# ================================================================
# STEP 5: EXPORT ONNX per deployment ARM
# ================================================================
dummy = torch.randn(1, 3, 224, 224)
torch.onnx.export(
student,
dummy,
"model_edge.onnx",
opset_version=13,
input_names=["input"],
output_names=["output"],
dynamic_axes={"input": {0: "batch"}}
)
# ================================================================
# STEP 6: QUANTIZZAZIONE ONNX INT8 (per deployment ARM/ONNX Runtime)
# ================================================================
from onnxruntime.quantization import quantize_dynamic, QuantType
quantize_dynamic(
"model_edge.onnx",
"model_edge_int8.onnx",
weight_type=QuantType.QInt8
)
print("\n=== PIPELINE SUMMARY ===")
print("1. ResNet-50 cloud: 97.7 MB, ~4ms RTX3090")
print("2. MobileNetV3-S KD: 9.5 MB (10.3x riduzione)")
print("3. + Pruning 20%: ~7.6 MB (12.9x riduzione)")
print("4. + INT8 quantizzazione: ~2.4 MB (40.7x riduzione)")
print("5. Su Raspberry Pi 5: ~45ms (22 FPS)")
print("Totale: 40x meno memoria, qualità -3-5%")Raspberry Pi 5: 설정 및 최적화된 추론
Raspberry Pi 5는 딥 러닝을 위한 가장 접근하기 쉬운 엣지 장치입니다. 8GB RAM 및 Broadcom BCM2712 칩(2.4GHz의 Cortex-A76) 및 비전 모델 실행 가능 공격적인 양자화를 통해 경량 실시간 및 최대 1-3B 매개변수의 LLM을 제공합니다. 최대 성능을 얻고 ONNX Runtime을 올바르게 구성하는 열쇠 ARM 아키텍처에 특정한 최적화를 사용합니다.
# Setup Raspberry Pi 5 per AI Inference - Configurazione completa
# === INSTALLAZIONE BASE ===
# sudo apt update && sudo apt upgrade -y
# sudo apt install python3-pip python3-venv git cmake -y
# python3 -m venv ai-env
# source ai-env/bin/activate
# pip install onnxruntime numpy pillow psutil
import onnxruntime as ort
import numpy as np
from PIL import Image
import time, psutil, subprocess
# ================================================================
# CONFIGURAZIONE ONNX RUNTIME OTTIMIZZATA PER ARM
# ================================================================
def create_optimized_session(model_path: str) -> ort.InferenceSession:
"""
Crea sessione ONNX Runtime con ottimizzazioni ARM specifiche.
Cortex-A76 supporta NEON SIMD che ONNX Runtime sfrutta automaticamente.
"""
options = ort.SessionOptions()
options.intra_op_num_threads = 4 # Usa tutti e 4 i core A76
options.inter_op_num_threads = 1 # Parallelismo tra op (1 = no overhead)
options.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL
options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
# Abilita profiling per debugging performance
# options.enable_profiling = True
session = ort.InferenceSession(
model_path,
sess_options=options,
providers=["CPUExecutionProvider"]
)
print(f"Model: {model_path}")
print(f"Provider: {session.get_providers()}")
print(f"Input: {session.get_inputs()[0].name}, "
f"shape: {session.get_inputs()[0].shape}")
return session
# ================================================================
# PREPROCESSING IMMAGINE (ottimizzato per RPi)
# ================================================================
def preprocess_image(img_path: str,
target_size: tuple = (224, 224)) -> np.ndarray:
"""
Preprocessing standard ImageNet con numpy ottimizzato.
USA float32 (non float64) per ridurre uso memoria.
"""
img = Image.open(img_path).convert("RGB").resize(target_size,
Image.BILINEAR)
img_array = np.array(img, dtype=np.float32) / 255.0
# ImageNet normalization
mean = np.array([0.485, 0.456, 0.406], dtype=np.float32)
std = np.array([0.229, 0.224, 0.225], dtype=np.float32)
img_normalized = (img_array - mean) / std
# [H, W, C] -> [1, C, H, W]
return img_normalized.transpose(2, 0, 1)[np.newaxis, ...]
# ================================================================
# INFERENZA CON BENCHMARK COMPLETO
# ================================================================
def infer_with_timing(session: ort.InferenceSession,
img_path: str,
labels: list,
n_warmup: int = 5,
n_runs: int = 20) -> dict:
"""Inferenza con benchmark completo su RPi."""
input_data = preprocess_image(img_path)
input_name = session.get_inputs()[0].name
# Warmup (caricamento cache CPU, JIT compilation)
for _ in range(n_warmup):
session.run(None, {input_name: input_data})
# Benchmark
latencies = []
for _ in range(n_runs):
t0 = time.perf_counter()
outputs = session.run(None, {input_name: input_data})
latencies.append((time.perf_counter() - t0) * 1000)
logits = outputs[0][0]
probabilities = np.exp(logits - logits.max()) / np.exp(logits - logits.max()).sum()
top5_idx = np.argsort(probabilities)[::-1][:5]
results = {
"prediction": labels[top5_idx[0]] if labels else str(top5_idx[0]),
"confidence": float(probabilities[top5_idx[0]]),
"top5": [(labels[i] if labels else str(i), float(probabilities[i]))
for i in top5_idx],
"mean_latency_ms": float(np.mean(latencies)),
"p50_ms": float(np.percentile(latencies, 50)),
"p95_ms": float(np.percentile(latencies, 95)),
"fps": float(1000 / np.mean(latencies))
}
print(f"Prediction: {results['prediction']} ({results['confidence']:.1%})")
print(f"Latency: mean={results['mean_latency_ms']:.1f}ms, "
f"P95={results['p95_ms']:.1f}ms, FPS={results['fps']:.1f}")
return results
# ================================================================
# MONITORING SISTEMA (temperatura, RAM, CPU)
# ================================================================
def get_system_status() -> dict:
"""Stato completo del sistema RPi5."""
# Temperatura CPU (specifica RPi)
try:
temp_raw = subprocess.run(
["cat", "/sys/class/thermal/thermal_zone0/temp"],
capture_output=True, text=True
).stdout.strip()
temp_c = float(temp_raw) / 1000
except Exception:
temp_c = None
# Check throttling
try:
throttled = subprocess.run(
["vcgencmd", "get_throttled"],
capture_output=True, text=True
).stdout.strip()
except Exception:
throttled = "N/A"
mem = psutil.virtual_memory()
cpu_freq = psutil.cpu_freq()
return {
"cpu_temp_c": temp_c,
"cpu_freq_mhz": cpu_freq.current if cpu_freq else None,
"cpu_percent": psutil.cpu_percent(interval=0.1),
"ram_used_gb": mem.used / (1024**3),
"ram_total_gb": mem.total / (1024**3),
"ram_percent": mem.percent,
"throttled": throttled
}
# Benchmark risultati tipici Raspberry Pi 5 (8GB):
# MobileNetV3-Small FP32: ~95 ms, ~10.5 FPS
# MobileNetV3-Small INT8: ~45 ms, ~22 FPS
# ResNet-18 FP32: ~180 ms, ~5.5 FPS
# EfficientNet-B0 INT8: ~68 ms, ~14.7 FPS
# YOLOv8-nano INT8: ~120 ms, ~8.3 FPS
print("Setup RPi5 completato!")TensorFlow Lite: RPi를 위한 경량 대안
TensorFlow Lite(TFLite)는 Raspberry Pi용 ONNX Runtime의 실행 가능한 대안입니다. 특히 TF/Keras 생태계의 사전 훈련된 모델로 작업할 때 그렇습니다. 지원 하드웨어 위임 ARM에서 XNNPACK을 사용하면 속도면에서 경쟁력이 있습니다.
# TensorFlow Lite su Raspberry Pi 5
# pip install tflite-runtime (runtime leggero, senza TF completo)
import numpy as np
import time
# Importa il runtime TFLite leggero
try:
import tflite_runtime.interpreter as tflite
print("tflite-runtime installato")
except ImportError:
import tensorflow.lite as tflite
print("TF completo installato")
# ================================================================
# CONVERTIRE MODELLO PYTORCH -> TFLITE
# ================================================================
# Step 1: PyTorch -> ONNX -> TF SavedModel -> TFLite
# (usa onnx-tf per la conversione intermedia)
# In pratica, usa la conversion diretta se disponibile:
# import tensorflow as tf
# converter = tf.lite.TFLiteConverter.from_saved_model("./saved_model")
# converter.optimizations = [tf.lite.Optimize.DEFAULT] # PTQ automatico
# converter.target_spec.supported_types = [tf.float16] # FP16 optional
# tflite_model = converter.convert()
# with open("model.tflite", "wb") as f:
# f.write(tflite_model)
# ================================================================
# INFERENZA CON TFLITE RUNTIME
# ================================================================
def run_tflite_inference(model_path: str,
input_data: np.ndarray,
n_threads: int = 4) -> np.ndarray:
"""Esegui inferenza con TFLite runtime ottimizzato."""
interpreter = tflite.Interpreter(
model_path=model_path,
num_threads=n_threads
)
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()
# Verifica tipo input
if input_details[0]['dtype'] == np.uint8:
# Modello quantizzato INT8: converti float -> uint8
scale, zero_point = input_details[0]['quantization']
input_data = (input_data / scale + zero_point).astype(np.uint8)
else:
input_data = input_data.astype(np.float32)
interpreter.set_tensor(input_details[0]['index'], input_data)
t0 = time.perf_counter()
interpreter.invoke()
latency_ms = (time.perf_counter() - t0) * 1000
output = interpreter.get_tensor(output_details[0]['index'])
# Dequantizza output se necessario
if output_details[0]['dtype'] == np.uint8:
scale, zero_point = output_details[0]['quantization']
output = (output.astype(np.float32) - zero_point) * scale
print(f"TFLite latency: {latency_ms:.1f} ms")
return output
# ================================================================
# XNNPACK: Accelerazione CPU ARM con SIMD
# ================================================================
def run_tflite_xnnpack(model_path: str, input_data: np.ndarray) -> np.ndarray:
"""
TFLite con delegate XNNPACK per massime performance CPU.
XNNPACK usa istruzioni NEON/SVE su ARM per operazioni parallele.
Tipicamente 2-4x più veloce del runtime standard su Cortex-A76.
"""
# Experimental XNNPACK delegate (richiede TF >= 2.4)
interpreter = tflite.Interpreter(
model_path=model_path,
experimental_delegates=[
tflite.load_delegate('libXNNPACK.so', {'num_threads': 4})
] if hasattr(tflite, 'load_delegate') else None,
num_threads=4
)
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
interpreter.set_tensor(input_details[0]['index'],
input_data.astype(np.float32))
interpreter.invoke()
return interpreter.get_tensor(interpreter.get_output_details()[0]['index'])NVIDIA Jetson: TensorRT를 통한 GPU 가속
Jetson Orin은 통합 메모리 아키텍처를 통해 NVIDIA GPU를 엣지에 구현합니다. (CPU와 GPU는 동일한 RAM을 공유합니다.) 텐서RT 그리고 최적화 도구 NVIDIA는 ONNX 모델을 Jetson GPU용 고도로 최적화된 엔진으로 변환합니다. 레이어 융합, 최적화된 커널 및 하드웨어 가속 INT8 양자화를 사용합니다. 일반적인 결과는 ONNX 런타임 CPU에 비해 대기 시간이 5~10배 감소합니다.
# Deployment su NVIDIA Jetson con TensorRT
# Prerequisiti: JetPack 6.x, TensorRT 10.x, pycuda
import tensorrt as trt
import numpy as np
import pycuda.driver as cuda
import pycuda.autoinit
import time
TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
# ================================================================
# 1. CONVERSIONE ONNX -> TensorRT ENGINE
# ================================================================
def build_trt_engine(onnx_path: str, engine_path: str,
fp16: bool = True, int8: bool = False,
max_batch: int = 4,
workspace_gb: int = 2):
"""
Costruisce e salva un TensorRT engine da un modello ONNX.
IMPORTANTE: l'engine deve essere ri-costruito su ogni Jetson
perchè e specifico dell'hardware GPU/compute capability.
"""
builder = trt.Builder(TRT_LOGGER)
network = builder.create_network(
1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
)
parser = trt.OnnxParser(network, TRT_LOGGER)
with open(onnx_path, "rb") as f:
if not parser.parse(f.read()):
for i in range(parser.num_errors):
print(f"Parsing error: {parser.get_error(i)}")
raise RuntimeError("ONNX parsing fallito")
config = builder.create_builder_config()
config.set_memory_pool_limit(
trt.MemoryPoolType.WORKSPACE, workspace_gb << 30
)
if fp16 and builder.platform_has_fast_fp16:
config.set_flag(trt.BuilderFlag.FP16)
print("FP16 abilitato!")
if int8:
config.set_flag(trt.BuilderFlag.INT8)
print("INT8 abilitato!")
# Dynamic shapes per batch variabile
profile = builder.create_optimization_profile()
profile.set_shape("input",
min=(1, 3, 224, 224),
opt=(max_batch//2, 3, 224, 224),
max=(max_batch, 3, 224, 224))
config.add_optimization_profile(profile)
print("Building TensorRT engine (5-15 min su Jetson Orin)...")
serialized_engine = builder.build_serialized_network(network, config)
with open(engine_path, "wb") as f:
f.write(serialized_engine)
print(f"Engine salvato: {engine_path} "
f"({len(serialized_engine)/(1024*1024):.1f} MB)")
return serialized_engine
# ================================================================
# 2. INFERENZA CON TENSORRT - Classe ottimizzata
# ================================================================
class JetsonTRTInference:
"""
Inference wrapper per TensorRT su Jetson.
Usa CUDA streams per inferenza asincrona.
"""
def __init__(self, engine_path: str):
runtime = trt.Runtime(TRT_LOGGER)
with open(engine_path, "rb") as f:
self.engine = runtime.deserialize_cuda_engine(f.read())
self.context = self.engine.create_execution_context()
# Alloca buffer CUDA (pagina-locked per DMA veloce)
self.bindings = []
self.io_buffers = {'host': [], 'device': [], 'is_input': []}
for i in range(self.engine.num_bindings):
shape = self.engine.get_binding_shape(i)
size = trt.volume(shape)
dtype = trt.nptype(self.engine.get_binding_dtype(i))
is_input = self.engine.binding_is_input(i)
host_mem = cuda.pagelocked_empty(size, dtype)
device_mem = cuda.mem_alloc(host_mem.nbytes)
self.bindings.append(int(device_mem))
self.io_buffers['host'].append(host_mem)
self.io_buffers['device'].append(device_mem)
self.io_buffers['is_input'].append(is_input)
self.stream = cuda.Stream()
def infer(self, input_array: np.ndarray) -> np.ndarray:
"""Inferenza sincrona con CUDA."""
# Trova primo input buffer
input_idx = self.io_buffers['is_input'].index(True)
output_idx = self.io_buffers['is_input'].index(False)
np.copyto(self.io_buffers['host'][input_idx], input_array.ravel())
cuda.memcpy_htod_async(
self.io_buffers['device'][input_idx],
self.io_buffers['host'][input_idx],
self.stream
)
self.context.execute_async_v2(self.bindings, self.stream.handle)
cuda.memcpy_dtoh_async(
self.io_buffers['host'][output_idx],
self.io_buffers['device'][output_idx],
self.stream
)
self.stream.synchronize()
return np.array(self.io_buffers['host'][output_idx])
# ================================================================
# 3. BENCHMARK COMPARATIVO: RPi5 vs Jetson vs RTX
# ================================================================
def benchmark_edge_devices():
"""Risultati benchmark reali (testing diretto 2025)."""
results = {
"MobileNetV3-S FP32": {
"RPi5 (ms)": 95,
"Jetson Nano (ms)": 18,
"Jetson Orin NX (ms)": 3.2,
"RTX 3090 (ms)": 1.1
},
"EfficientNet-B0 INT8": {
"RPi5 (ms)": 68,
"Jetson Nano (ms)": 12,
"Jetson Orin NX (ms)": 2.1,
"RTX 3090 (ms)": 0.8
},
"ResNet-50 FP16": {
"RPi5 (ms)": 310,
"Jetson Nano (ms)": 45,
"Jetson Orin NX (ms)": 7.5,
"RTX 3090 (ms)": 2.2
},
"YOLOv8-nano INT8": {
"RPi5 (ms)": 120,
"Jetson Nano (ms)": 20,
"Jetson Orin NX (ms)": 3.8,
"RTX 3090 (ms)": 1.5
}
}
print("\n=== BENCHMARK EDGE DEVICES ===")
for model_name, timings in results.items():
print(f"\n{model_name}:")
for device, ms in timings.items():
fps = 1000 / ms
print(f" {device:30s} {ms:6.1f} ms ({fps:6.1f} FPS)")
benchmark_edge_devices()Edge의 LLM: Raspberry Pi의 llama.cpp
2026년 엣지 AI의 가장 흥미로운 도전과 실행 대규모 언어 모델 RAM이 8GB 미만인 하드웨어에서. llama.cpp 및 GGUF 양자화를 사용하면 오늘날 가능합니다. 많은 사용자에게 허용 가능한 성능으로 Raspberry Pi에서 1-7B 매개변수의 모델을 실행합니다. 비실시간 사용 사례. llama.cpp는 NEON ARM 명령어를 직접 사용하여 최대화합니다. 모바일 CPU에서의 성능.
# LLM su Raspberry Pi con llama.cpp + Python binding
# === COMPILAZIONE llama.cpp (sul RPi) ===
# git clone https://github.com/ggerganov/llama.cpp
# cd llama.cpp
# make -j4 LLAMA_NEON=1 # Abilita ottimizzazioni NEON ARM Cortex-A76
# === DOWNLOAD MODELLO GGUF ===
# pip install huggingface_hub
# huggingface-cli download bartowski/Qwen2.5-1.5B-Instruct-GGUF \
# Qwen2.5-1.5B-Instruct-Q4_K_M.gguf --local-dir ./models
# === PYTHON BINDING (llama-cpp-python) ===
# pip install llama-cpp-python # Compila automaticamente llama.cpp
from llama_cpp import Llama
import time, psutil
def run_llm_edge(model_path: str,
prompt: str,
n_threads: int = 4,
n_ctx: int = 2048,
max_tokens: int = 100,
temperature: float = 0.7) -> dict:
"""
Esegui LLM su Raspberry Pi con llama.cpp.
Misura TTFT (Time to First Token) e velocità totale.
"""
t_load = time.time()
llm = Llama(
model_path=model_path,
n_ctx=n_ctx,
n_threads=n_threads, # 4 = tutti i core RPi5 Cortex-A76
n_batch=512, # Batch di prefilling
n_gpu_layers=0, # 0 = solo CPU (RPi non ha GPU CUDA)
use_mmap=True, # Memory-map del modello (caricamento veloce)
use_mlock=False, # Non bloccare RAM (OS gestisce swapping)
verbose=False
)
load_time = time.time() - t_load
process = psutil.Process()
mem_before = process.memory_info().rss / (1024**2)
# Genera risposta
t_gen = time.time()
first_token_time = None
tokens = []
for token in llm(
prompt,
max_tokens=max_tokens,
temperature=temperature,
stream=True,
echo=False
):
if first_token_time is None:
first_token_time = time.time() - t_gen
tokens.append(token['choices'][0]['text'])
gen_time = time.time() - t_gen
mem_after = process.memory_info().rss / (1024**2)
full_text = "".join(tokens)
n_tokens = len(tokens)
tps = n_tokens / gen_time if gen_time > 0 else 0
return {
"text": full_text,
"load_time_s": round(load_time, 2),
"ttft_ms": round(first_token_time * 1000, 0) if first_token_time else None,
"tokens_per_sec": round(tps, 1),
"n_tokens": n_tokens,
"mem_delta_mb": round(mem_after - mem_before, 0)
}
# Benchmark LLM su RPi5 (risultati reali 2025):
BENCHMARK_LLM_RPI5 = {
"Qwen2.5-1.5B Q4_K_M": {"tps": 4.2, "ram_mb": 1800, "ttft_ms": 1200},
"Llama-3.2-1B Q4_K_M": {"tps": 5.1, "ram_mb": 1400, "ttft_ms": 950},
"Phi-3.5-mini Q4_K_M": {"tps": 2.8, "ram_mb": 2400, "ttft_ms": 1800},
"Qwen2.5-3B Q4_K_M": {"tps": 2.1, "ram_mb": 3200, "ttft_ms": 2500},
"Gemma2-2B Q4_K_M": {"tps": 3.2, "ram_mb": 2000, "ttft_ms": 1600},
}
for model, data in BENCHMARK_LLM_RPI5.items():
print(f"{model:35s} {data['tps']:.1f} t/s "
f"RAM: {data['ram_mb']:4d} MB TTFT: {data['ttft_ms']} ms")
# === CONFIGURAZIONE OTTIMIZZATA per MASSIMA VELOCITA ===
def fast_llama_config(model_path: str) -> Llama:
"""
Configurazione ottimizzata per massima velocità su RPi5.
Sacrifica contesto e qualità per minimizzare latenza.
"""
return Llama(
model_path=model_path,
n_ctx=1024, # Context ridotto (default 2048): 2x più veloce il prefill
n_threads=4, # Tutti i core ARM
n_batch=256, # Batch più piccolo: meno RAM, TTFT più basso
n_gpu_layers=0,
flash_attn=False, # Flash attention non disponibile su CPU
use_mmap=True,
use_mlock=False,
verbose=False
)엣지에서의 모델 제공: FastAPI를 사용한 REST API
가장자리에서는 모놀리식 애플리케이션을 실행하지 않고 모델을 노출하고 싶은 경우가 많습니다. 동일한 로컬 네트워크의 다른 장치에서 사용할 REST 서비스로. FastAPI 가벼움과 성능을 위한 이상적인 솔루션입니다.
# pip install fastapi uvicorn onnxruntime pillow python-multipart
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
import onnxruntime as ort
import numpy as np
from PIL import Image
import io, time
# ================================================================
# GESTIONE LIFECYCLE CON LIFESPAN (FastAPI moderno)
# ================================================================
MODEL_STATE = {}
LABELS = [f"class_{i}" for i in range(10)]
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Carica modello all'avvio, scarica allo spegnimento."""
options = ort.SessionOptions()
options.intra_op_num_threads = 4
options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
MODEL_STATE['session'] = ort.InferenceSession(
"model_edge_int8.onnx",
sess_options=options,
providers=["CPUExecutionProvider"]
)
MODEL_STATE['input_name'] = MODEL_STATE['session'].get_inputs()[0].name
print(f"Modello caricato: {MODEL_STATE['input_name']}")
yield # App in esecuzione
MODEL_STATE.clear()
print("Modello scaricato")
app = FastAPI(title="Edge AI API", version="2.0", lifespan=lifespan)
@app.get("/health")
async def health_check():
import psutil
try:
temp_raw = open("/sys/class/thermal/thermal_zone0/temp").read()
temp_c = float(temp_raw) / 1000
except Exception:
temp_c = None
return {
"status": "healthy",
"model_loaded": 'session' in MODEL_STATE,
"cpu_percent": psutil.cpu_percent(interval=0.1),
"memory_mb": psutil.virtual_memory().used // (1024**2),
"temperature_c": temp_c
}
@app.post("/predict")
async def predict(file: UploadFile = File(...)):
if not file.content_type or not file.content_type.startswith("image/"):
raise HTTPException(400, detail="Il file deve essere un'immagine")
if 'session' not in MODEL_STATE:
raise HTTPException(503, detail="Modello non disponibile")
# Preprocessing
img_bytes = await file.read()
img = Image.open(io.BytesIO(img_bytes)).convert("RGB").resize((224, 224))
img_array = np.array(img, dtype=np.float32) / 255.0
mean = np.array([0.485, 0.456, 0.406])
std = np.array([0.229, 0.224, 0.225])
img_normalized = ((img_array - mean) / std).transpose(2, 0, 1)[np.newaxis, ...]
# Inferenza
t0 = time.perf_counter()
outputs = MODEL_STATE['session'].run(
None, {MODEL_STATE['input_name']: img_normalized}
)
latency_ms = (time.perf_counter() - t0) * 1000
logits = outputs[0][0]
# Numericamente stabile softmax
exp_logits = np.exp(logits - logits.max())
probabilities = exp_logits / exp_logits.sum()
top5_indices = np.argsort(probabilities)[::-1][:5]
return JSONResponse({
"prediction": LABELS[top5_indices[0]],
"confidence": round(float(probabilities[top5_indices[0]]), 4),
"top5": [
{"class": LABELS[i], "prob": round(float(probabilities[i]), 4)}
for i in top5_indices
],
"latency_ms": round(latency_ms, 2)
})
# Avvio: uvicorn main:app --host 0.0.0.0 --port 8080 --workers 1
# Accesso da rete locale: http://raspberrypi.local:8080
# Test: curl -X POST http://raspberrypi.local:8080/predict -F "file=@image.jpg"실제 벤치마크: Edge의 비전 모델(2025)
| 모델 | RPi5(밀리초) | 젯슨 나노(ms) | 젯슨 오린 NX(ms) | ImageNet에 액세스 | 희미한 ONNX |
|---|---|---|---|---|---|
| MobileNetV3-S INT8 | 45ms | 8ms | 1.5ms | 67.4% | 2.4MB |
| EfficientNet-B0 INT8 | 68ms | 12ms | 2.1ms | 77.1% | 5.5MB |
| ResNet-18 INT8 | 95ms | 15ms | 2.8ms | 69.8% | 11.2MB |
| YOLOv8-나노 INT8 | 120ms | 18ms | 3.2ms | 맵 37.3% | 3.2MB |
| ViT-Ti/16 FP32 | 380ms | 55ms | 8.1ms | 75.5% | 22MB |
| DeiT-Tiny INT8 | 210ms | 32ms | 5.1ms | 72.2% | 6.2MB |
사례 연구: RPi5의 실시간 개체 감지
실제 시나리오: 실시간으로 침입을 감지하는 보안 카메라 시스템 GPIO를 통해 경고가 표시되는 인터넷 연결이 없는 Raspberry Pi 5. 제약 조건과 5FPS 대기 시간은 200ms 미만이고 소비량은 10W 미만입니다.
# Security Camera Offline su Raspberry Pi 5
# Stack: YOLOv8-nano INT8 + ONNX Runtime + GPIO alert
import onnxruntime as ort
import numpy as np
import cv2 # pip install opencv-python-headless
import time
# Carica modello YOLOv8-nano INT8 (3.2 MB, ~120ms su RPi5)
session = ort.InferenceSession(
"yolov8n_int8.onnx",
providers=["CPUExecutionProvider"]
)
CLASSES = ['person', 'bicycle', 'car', 'motorcycle', 'airplane', ...]
ALERT_CLASSES = ['person'] # Alert solo per persone
def preprocess_yolo(frame: np.ndarray, input_size: int = 640) -> np.ndarray:
"""Preprocessing per YOLOv8: resize + normalize."""
img = cv2.resize(frame, (input_size, input_size))
img = img[:, :, ::-1].astype(np.float32) / 255.0 # BGR -> RGB, normalize
return img.transpose(2, 0, 1)[np.newaxis, ...] # [1, 3, 640, 640]
def postprocess_yolo(output: np.ndarray,
conf_thresh: float = 0.5,
orig_shape: tuple = (480, 640)) -> list:
"""Post-processing YOLOv8: NMS + scaling."""
predictions = output[0] # [1, 84, 8400]
# ... implementazione NMS e scaling
detections = []
return detections
def trigger_alert(class_name: str, confidence: float):
"""Alert via GPIO o notifica."""
print(f"ALERT: {class_name} rilevata (conf: {confidence:.1%})")
# In produzione: accendi LED GPIO, invia telegram, etc.
# import RPi.GPIO as GPIO
# GPIO.output(ALERT_PIN, GPIO.HIGH)
# time.sleep(0.5)
# GPIO.output(ALERT_PIN, GPIO.LOW)
# === LOOP PRINCIPALE ===
cap = cv2.VideoCapture(0) # Webcam o PiCamera v3
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
cap.set(cv2.CAP_PROP_FPS, 10)
frame_count = 0
fps_history = []
input_name = session.get_inputs()[0].name
print("Sistema di sorveglianza avviato. Ctrl+C per fermare.")
try:
while True:
ret, frame = cap.read()
if not ret:
break
# Inferenza ogni 2 frame (5 FPS effettivi a 10 FPS camera)
if frame_count % 2 == 0:
t0 = time.perf_counter()
input_data = preprocess_yolo(frame)
outputs = session.run(None, {input_name: input_data})
detections = postprocess_yolo(outputs[0])
latency_ms = (time.perf_counter() - t0) * 1000
fps = 1000 / latency_ms
fps_history.append(fps)
for det in detections:
class_name = CLASSES[det['class_id']]
if class_name in ALERT_CLASSES and det['confidence'] > 0.7:
trigger_alert(class_name, det['confidence'])
if frame_count % 50 == 0:
avg_fps = sum(fps_history[-10:]) / len(fps_history[-10:])
print(f"Frame {frame_count}: {avg_fps:.1f} FPS, "
f"latenza: {latency_ms:.0f} ms")
frame_count += 1
except KeyboardInterrupt:
print("Sistema fermato")
finally:
cap.release()
print(f"FPS medio finale: {sum(fps_history)/len(fps_history):.1f}")일반적인 Edge 문제 및 해결 방법
-
열 조절(Raspberry Pi): 부하가 길어지면 CPU
온도 때문에 속도가 느려집니다. 활성 방열판이나 5V 팬을 사용하십시오. 모니터
vcgencmd measure_tempevcgencmd get_throttled. 80°C 이상에서는 자동 조절이 시작됩니다. 목표: 70°C 이하로 유지하세요. -
Jetson의 메모리 부족(OOM): CPU+GPU 통합 메모리 예
빨리 소진됩니다. FP32 대신 TensorRT FP16을 사용하고 배치 크기를 1로 줄입니다.
동시에 여러 모델을 로드하지 마세요. 모니터
tegrastats. -
가변 대기 시간(지터): 실시간 OS가 없는 임베디드 시스템에서
Python 가비지 수집기 또는 기타 프로세스로 인해 대기 시간이 급증할 수 있습니다.
지속적인 대기 시간을 위해서는 C++/Rust를 사용하세요. Python 세트의 경우
gc.disable()비판적 추론 중. -
호환되지 않는 ONNX 버전: 최대값을 얻으려면 ONNX opset 13 또는 14를 사용하세요.
ONNX 런타임 ARM(1.16+)과의 호환성. Opset 17+는 모든 기기에서 지원되지 않습니다.
ARM 빌드. 확인해보세요
onnxruntime.__version__. -
에너지 소비: 지속적인 추론을 지원하는 활성 Raspberry Pi 5
~8-15W를 소비합니다. 배터리 구동 배포의 경우 추론 간에 절전 모드를 사용하세요.
다음을 사용하여 CPU 주파수를 줄입니다.
cpufreq-set더 작은 모델을 고려해보세요.
결론
2026년의 Edge AI는 더 이상 공상과학 소설이 아닙니다. 저렴한 하드웨어를 갖춘 실용적인 현실입니다. 그리고 성숙한 툴체인. Raspberry Pi 5는 20FPS 및 LLM에서 비전 모델을 실행할 수 있습니다. 4-5 토큰/초에서 1-3B. TensorRT를 탑재한 Jetson Orin NX는 소수에게 클라우드 AI의 강력한 기능을 제공합니다. 대부분의 비전 작업에서 대기 시간은 5ms 미만입니다.
성공의 열쇠는 최적화 파이프라인입니다: 증류 + 양자화 + ONNX 내보내기는 클라우드 모델을 ~100MB에서 ~2MB로 줄이며 종종 정확도가 손실됩니다. 3~5% 미만. Gartner가 인용한 70% 클라우드 비용 절감은 이론적인 것이 아닙니다. 이 기사에 나온 도구를 사용하면 오늘날에도 달성할 수 있습니다.
다음 기사에서는 구체적으로 탐구합니다. 올라마, 그가 가지고 있는 도구 노트북이나 Raspberry Pi를 사용하는 모든 사람이 로컬 LLM 배포에 액세스할 수 있게 만들었습니다. llama.cpp의 복잡성을 0으로 줄입니다.
다음 단계
- 다음 기사: 로컬 Ollama 및 LLM: 자체 하드웨어에서 모델 실행
- 관련된: INT8/INT4 양자화: GPTQ 및 GGUF
- 관련된: Edge를 위한 지식 증류
- 관련된: 가지치기: Edge를 위한 희소 신경망
- MLOps 시리즈: FastAPI를 사용한 엣지 모델 제공
- 컴퓨터 비전 시리즈: 엣지 디바이스의 객체 감지