0Pricing
AI Agents · Leçon

Déploiement en périphérie d’agents légers

Exécuter de petits modèles sur Raspberry Pi et des appareils en périphérie pour obtenir une réponse à faible latence.

Déploiement en périphérie d’agents légers est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.

Agents d’IA en périphérie

Un agent en périphérie s’exécute directement sur l’appareil IoT ou la passerelle locale — un Raspberry Pi, un Jetson Nano ou un PC industriel — plutôt que dans le nuage. Avantages : latence réduite (aucun aller-retour), fonctionnement hors ligne et coût de bande passante réduit. Compromis : la puissance de calcul et la RAM limitées nécessitent des modèles plus petits et plus efficaces.

Choisir un modèle pour un déploiement en périphérie

Les appareils en périphérie ne peuvent pas exécuter GPT-4o ou Claude Opus. Voici de petits modèles adaptés à un Raspberry Pi 5 (8 Go de RAM) : Phi-3-mini (3,8 milliards de paramètres), Gemma-2B et TinyLlama-1.1B. Quantifiés sur 4 bits, ces modèles nécessitent de 1 à 3 Go de RAM et s’exécutent à raison de 5 à 15 jetons par seconde sur un 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"]}')

Installer Ollama sur un Raspberry Pi

Ollama est le moyen le plus simple d’exécuter de petits LLM sur du matériel local. Il gère le téléchargement et la quantification des modèles ainsi qu’une API REST locale compatible avec le SDK OpenAI. Une commande suffit pour l’installer ; une autre permet de télécharger et de démarrer le modèle.

# 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)

Quantification GGUF sur 4 bits

GGUF (GPT-Generated Unified Format) est le format de fichier utilisé par llama.cpp et Ollama pour les modèles quantifiés. Q4_K_M désigne une quantification mixte sur 4 bits — environ 75 % plus petite que fp32, avec une perte de qualité minimale pour les tâches de raisonnement en périphérie.

# 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')

Optimiser la vitesse d’inférence sur CPU

Sur les appareils en périphérie utilisant uniquement un CPU, la vitesse d’inférence dépend du nombre de fils d’exécution et de la mise en cache des consignes. Définissez le nombre de fils d’exécution pour qu’il corresponde au nombre de cœurs du CPU, gardez la consigne système courte (elle est retraitée à chaque appel si elle n’est pas mise en cache) et regroupez les requêtes répétées lorsque c’est possible.

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)

Architecture hybride en périphérie et à distance

L’architecture optimale utilise le modèle en périphérie pour les décisions rapides et peu critiques (classification, logique à seuil), puis se synchronise avec le modèle distant pour les décisions complexes, rares ou critiques. L’agent en périphérie met les requêtes complexes en file d’attente et les envoie dès que la connectivité le permet.

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()

File d’attente hors ligne pour la synchronisation distante

Les agents en périphérie fonctionnent souvent dans des environnements où la connectivité est intermittente. Mettez les requêtes, les relevés de capteurs et les journaux d’actions en mémoire tampon localement, puis envoyez-les au service distant lorsque la connectivité est rétablie. Utilisez une base de données SQLite comme mémoire tampon 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}'))

Surveillance de l’état de santé de l’agent en périphérie

Les appareils en périphérie peuvent surchauffer, manquer de RAM ou subir un blocage de l’inférence du modèle. Implémentez un moniteur d’état qui vérifie la température du CPU, la RAM disponible et la latence de l’inférence, puis publie les métriques d’état dans le service distant.

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)

Mise en cache de l’inférence pour les agents en périphérie

Les modèles en périphérie sont lents : mettez en cache les réponses correspondant à des entrées identiques. Pour les tâches de classification de capteurs, le même prompt (par exemple, "temperature=23.5, classify") se répète souvent. Un cache simple fondé sur un dictionnaire et doté d’un TTL évite les appels d’inférence redondants.

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

Liste de contrôle du déploiement

Avant de déployer un agent en périphérie en production, vérifiez les points suivants : le modèle tient dans la RAM disponible avec une marge de 20 %, la latence d’inférence respecte l’exigence de temps de réponse de l’événement, la mémoire tampon hors ligne a été testée, la surveillance de l’état publie bien ses données et le redémarrage automatique en cas de plantage est configuré.

# 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}')

Configuration matérielle du Raspberry Pi

Avant de déployer le logiciel, préparez le matériel. Un Raspberry Pi 5 doté de 8 GB de RAM constitue le minimum recommandé pour exécuter Phi-3-mini. Activez la répartition de la mémoire du GPU, désactivez la mémoire d’échange (elle dégrade le stockage flash) et configurez une IP statique pour garantir un accès distant fiable.

# 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')

Vérification des connaissances

Que signifie la quantification Q4_K_M pour un modèle de langage ?

Récapitulatif&nbsp;: déploiement d’agents légers en périphérie

Ce que vous avez étudié dans cette leçon :

  • Sélection du modèle : Phi-3-mini, Gemma-2B et TinyLlama pour la périphérie ; quantification Q4_K_M
  • Ollama : serveur LLM local avec une API compatible avec OpenAI sur le port 11434
  • Architecture hybride : modèle en périphérie pour les décisions rapides, service distant pour les analyses complexes
  • Mémoire tampon hors ligne : file d’attente SQLite envoyée au service distant lors de la reconnexion
  • Cache d’inférence : le cache TTL évite de répéter l’inférence pour des prompts identiques
  • Déploiement : service systemd pour redémarrer automatiquement après un plantage

Prochain cours : place de marché des agents et systèmes d’extensions.

Questions Fréquemment Posées

La leçon « Déploiement en périphérie d’agents légers » est-elle gratuite ?

Oui — le texte complet de « Déploiement en périphérie d’agents légers » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Déploiement en périphérie d’agents légers » ?

Exécuter de petits modèles sur Raspberry Pi et des appareils en périphérie pour obtenir une réponse à faible latence. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Agents ?

Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.

Combien de temps prend la leçon « Déploiement en périphérie d’agents légers » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?

Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Protocole MQTT pour l’intégration d’agents
  2. Traitement des séries temporelles par les agents
  3. Réponse automatisée aux événements des capteurs
  4. Déploiement en périphérie d’agents légers
← Retour à AI Agents