0Pricing
AI Agents · Lezione

Distribuzione edge di agenti leggeri

Esecuzione di modelli di piccole dimensioni su Raspberry Pi e dispositivi edge per risposte a bassa latenza.

Distribuzione edge di agenti leggeri è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.

Agenti di IA edge

Un agente edge viene eseguito direttamente sul dispositivo IoT o sul gateway locale — un Raspberry Pi, un Jetson Nano o un PC industriale — invece che nel cloud. Vantaggi: latenza inferiore (nessun viaggio di andata e ritorno), funzionamento offline e minori costi di banda. Svantaggio: la limitata capacità di calcolo e RAM richiede modelli più piccoli ed efficienti.

Scelta di un modello per la distribuzione edge

I dispositivi edge non sono in grado di eseguire GPT-4o o Claude Opus. I modelli di piccole dimensioni che possono essere eseguiti su un Raspberry Pi 5 (8 GB di RAM) includono: Phi-3-mini (3.8B parametri), Gemma-2B, TinyLlama-1.1B. Se quantizzati a 4 bit, questi modelli richiedono da 1 a 3 GB di RAM ed eseguono da 5 a 15 token al secondo sulla CPU.

# Model size reference for edge selection:
EDGE_MODELS = {
    'tinyllama-1.1b-q4': {
        'params': '1.1B', 'quantization': 'Q4_K_M',
        'ram_gb': 0.8, 'tokens_per_sec_cpu': 15,
        'use_case': 'simple classification, keyword detection'
    },
    'phi-3-mini-q4': {
        'params': '3.8B', 'quantization': 'Q4_K_M',
        'ram_gb': 2.5, 'tokens_per_sec_cpu': 8,
        'use_case': 'reasoning, multi-step decisions'
    },
    'gemma-2b-q4': {
        'params': '2B', 'quantization': 'Q4_K_M',
        'ram_gb': 1.5, 'tokens_per_sec_cpu': 10,
        'use_case': 'general assistant tasks'
    }
}

for name, info in EDGE_MODELS.items():
    print(f'{name}: {info["ram_gb"]}GB RAM, '
          f'{info["tokens_per_sec_cpu"]} tok/s — {info["use_case"]}')

Installazione di Ollama su Raspberry Pi

Ollama è il modo più semplice per eseguire LLM di piccole dimensioni sull'hardware locale. Gestisce il download dei modelli, la quantizzazione e un'API REST locale compatibile con l'SDK OpenAI. Un comando lo installa; un altro scarica e avvia il modello.

# Install Ollama (run on the Raspberry Pi terminal):
# curl -fsSL https://ollama.ai/install.sh | sh

# Pull and run a model:
# ollama pull phi3:mini
# ollama serve  (starts API on localhost:11434)

# Python client — uses the OpenAI-compatible endpoint:
from openai import OpenAI

local_client = OpenAI(
    base_url='http://localhost:11434/v1',
    api_key='ollama'  # Ollama ignores this but it is required by the SDK
)

def local_inference(prompt: str, model: str = 'phi3:mini') -> str:
    response = local_client.chat.completions.create(
        model=model,
        messages=[{'role': 'user', 'content': prompt}],
        max_tokens=256,
        temperature=0.1
    )
    return response.choices[0].message.content

result = local_inference('Is temperature 45C dangerous for a server room?')
print(result)

Quantizzazione GGUF a 4 bit

GGUF (GPT-Generated Unified Format) è il formato di file utilizzato da llama.cpp e Ollama per i modelli quantizzati. Q4_K_M indica una quantizzazione mista a 4 bit: circa il 75% più piccola rispetto a fp32, con una perdita minima di qualità per le attività di ragionamento edge.

# Understanding quantisation quality levels:
QUANTISATION_GUIDE = {
    'Q2_K': {'size_multiplier': 0.25, 'quality': 'poor', 'ram': 'minimal'},
    'Q4_K_M': {'size_multiplier': 0.45, 'quality': 'good', 'ram': 'low'},
    'Q5_K_M': {'size_multiplier': 0.55, 'quality': 'better', 'ram': 'medium'},
    'Q8_0': {'size_multiplier': 0.75, 'quality': 'near-original', 'ram': 'high'},
    'F16': {'size_multiplier': 1.0, 'quality': 'original', 'ram': 'full'}
}

# For edge: Q4_K_M is the sweet spot
# 7B model: 7B * 4bit/8 = 3.5 GB in Q4 vs 14 GB in F16

def estimate_vram_gb(param_billions: float, quant: str) -> float:
    multiplier = QUANT_GUIDE = {
        'Q4_K_M': 0.45, 'Q8_0': 0.75, 'F16': 1.0
    }
    return round(param_billions * multiplier.get(quant, 0.5), 2)

print(f'Phi-3-mini Q4_K_M: {estimate_vram_gb(3.8, "Q4_K_M")} GB')
print(f'Phi-3-mini F16: {estimate_vram_gb(3.8, "F16")} GB')

Ottimizzazione della velocità di inferenza sulla CPU

Sui dispositivi edge che utilizzano solo la CPU, la velocità di inferenza dipende dal numero di thread e dalla memorizzazione nella cache del prompt. Imposti il numero di thread in modo che corrisponda ai core della CPU, mantenga breve il prompt di sistema (se non viene memorizzato nella cache, viene rielaborato a ogni chiamata) ed elabori in batch le query ripetute quando possibile.

import os
import time

# Ollama environment variables for performance tuning
# Set before starting the Ollama service:
# OLLAMA_NUM_PARALLEL=1  (single request at a time on small devices)
# OLLAMA_MAX_LOADED_MODELS=1  (only keep one model in RAM)

def timed_inference(prompt: str, client, model: str = 'phi3:mini') -> dict:
    start = time.time()
    response = client.chat.completions.create(
        model=model,
        messages=[{'role': 'user', 'content': prompt}],
        max_tokens=128
    )
    elapsed = time.time() - start
    text = response.choices[0].message.content
    tokens = len(text.split())  # approximate
    return {
        'text': text,
        'elapsed_s': round(elapsed, 2),
        'approx_tps': round(tokens / elapsed, 1)
    }

result = timed_inference('Classify: temperature=45C, normal range 18-30C. Action?',
                         local_client)
print(result)

Architettura ibrida edge-cloud

L'architettura ottimale utilizza il modello edge per le decisioni rapide e a basso rischio (classificazione, logica basata su soglie) e si sincronizza con il modello cloud per le decisioni complesse, rare o ad alto impatto. L'agente edge accoda le richieste complesse e le invia quando la connettività lo consente.

import anthropic
import time

class HybridAgent:
    def __init__(self, edge_client, cloud_api_key: str):
        self.edge = edge_client  # Ollama local client
        self.cloud = anthropic.Anthropic(api_key=cloud_api_key)
        self.pending_cloud_queries = []

    def decide(self, context: dict) -> str:
        # Fast path: edge model for simple binary decisions
        simple_prompt = (
            f'Sensor: {context}. '
            'Respond with exactly one word: NORMAL or ALERT.'
        )
        edge_result = self.edge.chat.completions.create(
            model='phi3:mini',
            messages=[{'role': 'user', 'content': simple_prompt}],
            max_tokens=5
        ).choices[0].message.content.strip()

        if edge_result == 'ALERT':
            # Queue complex analysis for cloud
            self.pending_cloud_queries.append(context)

        return edge_result

    def sync_to_cloud(self):
        """Call when online connectivity is available."""
        for ctx in self.pending_cloud_queries:
            result = self.cloud.messages.create(
                model='claude-opus-4-5', max_tokens=512,
                messages=[{'role': 'user', 'content':
                    f'Full analysis of: {ctx}'}]
            )
            print('Cloud analysis:', result.content[0].text[:100])
        self.pending_cloud_queries.clear()

Coda offline per la sincronizzazione con il cloud

Gli agenti edge operano spesso in ambienti con connettività intermittente. Memorizzi localmente le richieste, le letture dei sensori e i log delle azioni, quindi li invii al cloud quando la connettività viene ripristinata. Utilizzi un database SQLite come buffer locale.

import sqlite3
import json
from datetime import datetime

DB_PATH = '/home/pi/agent_buffer.db'

def init_buffer_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS sync_queue (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            type TEXT NOT NULL,
            payload TEXT NOT NULL,
            created_at TEXT NOT NULL,
            synced INTEGER DEFAULT 0
        )
    """)
    conn.commit()
    conn.close()

def buffer_for_sync(record_type: str, payload: dict):
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        'INSERT INTO sync_queue (type, payload, created_at) VALUES (?, ?, ?)',
        (record_type, json.dumps(payload), datetime.utcnow().isoformat())
    )
    conn.commit()
    conn.close()

def flush_to_cloud(cloud_fn):
    conn = sqlite3.connect(DB_PATH)
    rows = conn.execute(
        'SELECT id, type, payload FROM sync_queue WHERE synced=0 LIMIT 100'
    ).fetchall()
    for row_id, r_type, payload in rows:
        cloud_fn(r_type, json.loads(payload))
        conn.execute('UPDATE sync_queue SET synced=1 WHERE id=?', (row_id,))
    conn.commit()
    conn.close()
    print(f'Synced {len(rows)} records to cloud')

if __name__ == '__main__':
    import os, tempfile
    DB_PATH = os.path.join(tempfile.gettempdir(), 'agent_buffer_demo.db')
    init_buffer_db()
    buffer_for_sync('sensor_reading', {'temp': 22.5})
    flush_to_cloud(lambda t, p: print(f'Synced to cloud: {t} -> {p}'))

Monitoraggio dello stato dell'agente edge

I dispositivi edge possono surriscaldarsi, esaurire la RAM disponibile oppure bloccarsi durante l'inferenza del modello. Implementi un monitor dello stato che controlli la temperatura della CPU, la RAM libera e la latenza dell'inferenza, quindi pubblichi le metriche dello stato nel cloud.

import subprocess
import psutil  # pip install psutil

def get_edge_health() -> dict:
    cpu_percent = psutil.cpu_percent(interval=1)
    ram = psutil.virtual_memory()
    disk = psutil.disk_usage('/')

    # Read CPU temperature (Raspberry Pi specific)
    try:
        temp_output = subprocess.check_output(
            ['vcgencmd', 'measure_temp'], text=True
        )
        cpu_temp = float(temp_output.strip().replace("temp=", "").replace("'C", ""))
    except Exception:
        cpu_temp = -1.0  # not available on non-Pi hardware

    return {
        'cpu_percent': cpu_percent,
        'cpu_temp_c': cpu_temp,
        'ram_used_pct': ram.percent,
        'ram_available_mb': round(ram.available / 1024 / 1024),
        'disk_used_pct': disk.percent,
        'timestamp': datetime.utcnow().isoformat()
    }

health = get_edge_health()
print('Edge health:', health)

Memorizzazione nella cache delle inferenze per gli agenti edge

I modelli edge sono lenti: memorizzi nella cache le risposte per gli input identici. Nelle attività di classificazione dei sensori, lo stesso prompt (ad esempio, "temperature=23.5, classify") si ripete spesso. Una semplice cache basata su dict con TTL evita chiamate di inferenza ridondanti.

from datetime import datetime, timedelta
import hashlib

class InferenceCache:
    def __init__(self, ttl_seconds: int = 60):
        self.ttl = timedelta(seconds=ttl_seconds)
        self._cache: dict = {}  # hash -> {'result', 'expires'}

    def _key(self, prompt: str) -> str:
        return hashlib.md5(prompt.encode()).hexdigest()

    def get(self, prompt: str):
        key = self._key(prompt)
        entry = self._cache.get(key)
        if entry and datetime.utcnow() < entry['expires']:
            return entry['result']
        return None

    def set(self, prompt: str, result: str):
        key = self._key(prompt)
        self._cache[key] = {
            'result': result,
            'expires': datetime.utcnow() + self.ttl
        }

cache = InferenceCache(ttl_seconds=60)

def cached_edge_inference(prompt: str, client) -> str:
    cached = cache.get(prompt)
    if cached:
        print('Cache hit')
        return cached
    result = local_inference(prompt, client)
    cache.set(prompt, result)
    return result

Checklist del deployment

Prima di distribuire un agente edge in produzione, verifichi che: il modello rientri nella RAM disponibile lasciando un margine del 20%, la latenza dell'inferenza soddisfi il requisito di risposta dell'evento, il buffer offline sia stato testato, il monitoraggio dello stato stia pubblicando metriche e sia configurato il riavvio automatico in caso di arresto anomalo.

# systemd service file for auto-restart (save as /etc/systemd/system/edge-agent.service)

SYSTEMD_SERVICE = '''
[Unit]
Description=IoT Edge Agent
After=network.target ollama.service
Requires=ollama.service

[Service]
User=pi
WorkingDirectory=/home/pi/edge-agent
ExecStart=/usr/bin/python3 /home/pi/edge-agent/agent.py
Restart=always
RestartSec=5
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target
'''

# Enable:
# sudo systemctl enable edge-agent
# sudo systemctl start edge-agent
# sudo journalctl -u edge-agent -f  (follow logs)

print('Deployment checklist:')
checklist = [
    'Model RAM fits with 20% headroom',
    'Inference latency < response time requirement',
    'Offline SQLite buffer tested',
    'Health metrics publishing to cloud',
    'systemd auto-restart configured'
]
for item in checklist:
    print(f'  [ ] {item}')

Configurazione hardware del Raspberry Pi

Prima di distribuire il software, prepari l'hardware. Un Raspberry Pi 5 con 8 GB di RAM è il requisito minimo consigliato per eseguire Phi-3-mini. Abiliti la suddivisione della memoria GPU, disabiliti lo swap (che riduce la durata della memoria flash) e configuri un IP statico per un accesso remoto affidabile.

# Raspberry Pi setup notes (run manually over SSH):
# sudo raspi-config -> System -> GPU Memory -> 256
# Disable swap to protect SD card:
# sudo dphys-swapfile swapoff && sudo dphys-swapfile uninstall

def check_available_ram_mb():
    try:
        with open('/proc/meminfo') as f:
            for line in f:
                if line.startswith('MemAvailable'):
                    return int(line.split()[1]) // 1024
    except FileNotFoundError:
        return None

ram_mb = check_available_ram_mb()
if ram_mb is None:
    print('Could not read /proc/meminfo on this OS — simulated Pi reading: MemAvailable ~ 850 MB')
else:
    print(f'Available RAM: {ram_mb} MB')

Verifica delle conoscenze

Che cosa significa la quantizzazione Q4_K_M per un modello linguistico?

Riepilogo: deployment di agenti leggeri all'edge

In questa lezione ha approfondito:

  • Selezione del modello: Phi-3-mini, Gemma-2B e TinyLlama per l'edge; quantizzazione Q4_K_M
  • Ollama: server LLM locale con API compatibile con OpenAI sulla porta 11434
  • Architettura ibrida: modello edge per le decisioni rapide, cloud per l'analisi complessa
  • Buffer offline: coda SQLite inviata al cloud alla riconnessione
  • Cache delle inferenze: la cache con TTL evita inferenze ripetute per prompt identici
  • Deployment: servizio systemd per il riavvio automatico in caso di arresto anomalo

Prossimo corso: marketplace degli agenti e sistemi di plugin.

Domande Frequenti

La lezione «Distribuzione edge di agenti leggeri» è gratuita?

Sì — il testo completo di «Distribuzione edge di agenti leggeri» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.

Cosa imparerò in «Distribuzione edge di agenti leggeri»?

Esecuzione di modelli di piccole dimensioni su Raspberry Pi e dispositivi edge per risposte a bassa latenza. Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Agents?

Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.

Quanto tempo richiede la lezione «Distribuzione edge di agenti leggeri»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Agents?

Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Protocollo MQTT per l’integrazione degli agenti
  2. Elaborazione dei dati di serie temporali negli agenti
  3. Risposta automatizzata agli eventi dei sensori
  4. Distribuzione edge di agenti leggeri
← Torna a AI Agents