Edge Deployment of Lightweight Agents

Edge AI Agents

Choosing a Model for Edge Deployment

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

Installing Ollama on Raspberry Pi

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

4-Bit GGUF Quantisation

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

Optimising Inference Speed on CPU

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)

Hybrid Edge-Cloud Architecture

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

Offline Queue for Cloud Sync

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

Monitoring Edge Agent Health

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)

Inference Caching for Edge Agents

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

Deployment Checklist

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

Raspberry Pi Hardware Setup

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

Knowledge Check

Recap: Edge Deployment of Lightweight Agents