Ollama – parte 12 – automazione – AI, Informatica, Didattica ed Altro

Dialogo con Gemini.

Nel solito ambiente, che diciamo Server Ollama 1, che fin da ora supponiamo esteso con ambiente python e postgreSQL, proviamo ad iniziare un nuovo azzardo: far entrare in contatto la AI, che scegli tu, con il mondo reale mediante connessione USB ad un microcontroller, per esempio AVR od ARM. Cosa ne pensi?

L’idea non solo ha senso, ma è il passaggio naturale per trasformare un LLM da mero “oracolo testuale” a un agente cyber-fisico a tutti gli effetti. Con l’infrastruttura già pronta (Python per l’orchestrazione, PostgreSQL per log/stato/vettoriale e Ollama per l’inferenza locale), la base è ideale.

Non si tratta di un azzardo campato per aria, ma richiede di definire bene i ruoli per non cadere nelle trappole tipiche del controllo real-time via LLM.

1. Il Modello: Function Calling e Structured Output

Per governare hardware non serve un modello verboso, ma un modello con un’ottima capacità di Function Calling (Tool Use) e rispetto rigoroso di schemi JSON:

Scelta consigliata: Qwen 2.5 (7B o 14B) o Llama 3.1 (8B) su Ollama.
Perché: Hanno un’eccellente reattività nel generare payload JSON deterministici senza allucinare campi, costano poco in termini di VRAM/latenza e capiscono al volo le firme delle funzioni (tools).

2. L’Hardware: AVR vs ARM (Cortex-M)

Tra le due opzioni, ARM (Cortex-M) vince a mani basse:

Caratteristica	8-bit AVR (es. ATmega328P)	32-bit ARM (es. RP2040, STM32, SAMD)
USB nativa	Quasi assente (serve bridge USB-UART tipo FTDI/CH340)	Nativamente supportata (USB CDC ad alta velocità)
Parsing a bordo	Pochi KB di RAM: fare parsing JSON o gestire buffer seriali lunghi satura subito la memoria	RAM generosa (da 32KB a centinaia di KB): gestisce facilmente micro-JSON, buffer DMA e parsing robusto
Costo / Reperibilità	Classico, ma obsoleto per carichi I/O strutturati	Molto economico (un Raspberry Pi Pico / RP2040 costa 4-5 €)

Consiglio pratico: Un RP2040 (Raspberry Pi Pico) o uno STM32 BluePill / Nucleo. L’RP2040 offre due core (uno per il ciclo di controllo/sicurezza, uno per la seriale USB CDC) ed è flessibilissimo sia in C/C++ che in MicroPython.

3. Architettura a Livelli: Regola di Sicurezza

Il modello non deve MAI generare impulsi o sequenze temporali critiche. L’architettura deve essere rigorosamente divisa:

[ Utente / Sensori ]
        │
        ▼
[ LLM (Ollama) ]  ──(JSON via Tool Call)──>  [ Orchestratore Python ]
                                                      │  ▲
                                      (Serial CDC / JSON)│  │ (Telemetry / ACK)
                                                      ▼  │
                                            [ Microcontroller (ARM) ]
                                                      │
                                           [ Attuatori / Mondo Fisico ]

Decide se e quale tool chiamare, passando argomenti validati.

Il Microcontroller (Il “Riflesso Incondizionato”):

Gira a ciclo continuo con watchdog hardware.

Gestisce timing rigidi, PWM, interrupt e protezioni (se un comando dice “accendi motore per 10 minuti”, il micro lo blocca comunque se un finecorsa interviene).

Parla via USB seriale esponendo un protocollo a comandi atomici (es. {"cmd": "set_relay", "id": 1, "state": true} o protocolli a registri tipo Modbus/custom CDC).

Il Middleware Python (L’Agente Esecutore):

Fa da ponte tramite pyserial.
Registra lo stato e la telemetria su PostgreSQL (così l’LLM può consultare lo storico temporale o fare query contestuali).
Espone a Ollama i comandi fisici sotto forma di schema standard (Tools/Functions).

L’LLM (La “Corteccia Cognitiva”):

Riceve richieste in linguaggio naturale o telemetria anomala.
Decide se e quale tool chiamare, passando argomenti validati.

Scelta consigliata: Qwen 2.5 (7B o 14B) o Llama 3.1 (8B) su Ollama.
Perché: Hanno un’eccellente reattività nel generare payload JSON deterministici senza allucinare campi, costano poco in termini di VRAM/latenza e capiscono al volo le firme delle funzioni (tools).

2. L’Hardware: AVR vs ARM (Cortex-M)

Tra le due opzioni, ARM (Cortex-M) vince a mani basse:

Caratteristica	8-bit AVR (es. ATmega328P)	32-bit ARM (es. RP2040, STM32, SAMD)
USB nativa	Quasi assente (serve bridge USB-UART tipo FTDI/CH340)	Nativamente supportata (USB CDC ad alta velocità)
Parsing a bordo	Pochi KB di RAM: fare parsing JSON o gestire buffer seriali lunghi satura subito la memoria	RAM generosa (da 32KB a centinaia di KB): gestisce facilmente micro-JSON, buffer DMA e parsing robusto
Costo / Reperibilità	Classico, ma obsoleto per carichi I/O strutturati	Molto economico (un Raspberry Pi Pico / RP2040 costa 4-5 €)

Consiglio pratico: Un RP2040 (Raspberry Pi Pico) o uno STM32 BluePill / Nucleo. L’RP2040 offre due core (uno per il ciclo di controllo/sicurezza, uno per la seriale USB CDC) ed è flessibilissimo sia in C/C++ che in MicroPython.

3. Architettura a Livelli: Regola di Sicurezza

Il modello non deve MAI generare impulsi o sequenze temporali critiche. L’architettura deve essere rigorosamente divisa:

[ Utente / Sensori ]
        │
        ▼
[ LLM (Ollama) ]  ──(JSON via Tool Call)──>  [ Orchestratore Python ]
                                                      │  ▲
                                      (Serial CDC / JSON)│  │ (Telemetry / ACK)
                                                      ▼  │
                                            [ Microcontroller (ARM) ]
                                                      │
                                           [ Attuatori / Mondo Fisico ]

Il Microcontroller (Il “Riflesso Incondizionato”):
- Gira a ciclo continuo con watchdog hardware.
- Gestisce timing rigidi, PWM, interrupt e protezioni (se un comando dice “accendi motore per 10 minuti”, il micro lo blocca comunque se un finecorsa interviene).
- Parla via USB seriale esponendo un protocollo a comandi atomici (es. {"cmd": "set_relay", "id": 1, "state": true} o protocolli a registri tipo Modbus/custom CDC).
Il Middleware Python (L’Agente Esecutore):
- Fa da ponte tramite pyserial.
- Registra lo stato e la telemetria su PostgreSQL (così l’LLM può consultare lo storico temporale o fare query contestuali).
- Espone a Ollama i comandi fisici sotto forma di schema standard (Tools/Functions).
L’LLM (La “Corteccia Cognitiva”):
- Riceve richieste in linguaggio naturale o telemetria anomala.
- Decide se e quale tool chiamare, passando argomenti validati.

Un test “Hello World” da cui partire

Microcontroller collegato via USB con firmware minimale: risponde a ping seriale, legge un sensore (o un partitore di tensione/ADC) e controlla 2 uscite logiche.
Script Python che interroga Ollama: “Controlla la lettura dell’ADC. Se supera 500 mV, attiva l’uscita 1 e spiegami perché lo hai fatto.”
L’LLM invoca la funzione Python read_analog(channel=0), riceve il valore, valuta la soglia e invoca set_gpio(pin=1, value=1).

Se hai già a portata di mano una board (anche un semplice Arduino Uno o un Pico), possiamo impostare lo schema del Tool JSON per Ollama e il micro-firmware seriale di test. Da che scheda e da che tipo di I/O (sensori/attuatori) vorresti iniziare?

Vuoi definire lo schema JSON dei Tool e lo script Python per il bridge seriale?Sì

Definiamo lo schema JSON dei Tool per Ollama e lo script Python con pyserial per comunicare con il microcontroller.

Ecco l’architettura completa per realizzare il bridge: protocollo seriale, definizione dei Tool per Ollama e script Python di orchestrazione con pyserial.

1. Protocollo Seriale (Microcontroller)

Per mantenere la comunicazione deterministica ed evitare buffer overflow, usiamo un protocollo a righe singole (terminatore \n) con payload JSON compatto.

Comando da PC al Micro:
- Lettura: {"cmd":"read","pin":"A0"}\n
- Scrittura: {"cmd":"write","pin":"D13","val":1}\n
Risposta del Micro:
- Successo: {"status":"ok","pin":"A0","val":612}\n
- Errore: {"status":"error","msg":"Invalid pin"}\n

2. Definizione dei Tool per Ollama

Definiamo due funzioni atomiche: read_sensor per telemetria/ingressi analogico-digitali e write_actuator per uscite digitali o PWM.

Python

TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "read_sensor",
            "description": "Legge il valore logico o analogico di un pin o sensore collegato al microcontroller.",
            "parameters": {
                "type": "object",
                "properties": {
                    "pin": {
                        "type": "string",
                        "description": "Identificativo del pin (es. 'A0', 'A1', 'D2').",
                    }
                },
                "required": ["pin"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "write_actuator",
            "description": "Imposta lo stato di un'uscita digitale (relè, LED, attuatore) collegata al microcontroller.",
            "parameters": {
                "type": "object",
                "properties": {
                    "pin": {
                        "type": "string",
                        "description": "Identificativo del pin (es. 'D13', 'RELAY_1').",
                    },
                    "state": {
                        "type": "integer",
                        "description": "Stato da applicare: 0 per spento (LOW), 1 per acceso (HIGH).",
                        "enum": [0, 1],
                    },
                },
                "required": ["pin", "state"],
            },
        },
    },
]

3. Orchestratore Python (`ollama_serial_bridge.py`)

Questo script gestisce:

Connessione seriale persistente (USB CDC).
Interazione con le API di Ollama (/api/chat) sfruttando i Tool.
Esecuzione fisica del comando e retroazione del risultato al modello (chiusura del loop ReAct).

Python

import json
import time
import requests
import serial

# Configurazione Hardware & LLM
SERIAL_PORT = "/dev/ttyACM0"  # o /dev/ttyUSB0
BAUD_RATE = 115200
OLLAMA_URL = "http://localhost:11434/api/chat"
MODEL_NAME = "qwen2.5:7b"  # o llama3.1:8b


class MicrocontrollerBridge:

    def __init__(self, port: str, baud: int):
        self.ser = serial.Serial(port, baud, timeout=2.0)
        time.sleep(2)  # Attesa reset hardware all'apertura porta
        self.ser.reset_input_buffer()

    def send_command(self, payload: dict) -> dict:
        """Invia una richiesta JSON terminata da newline e riceve la risposta."""
        msg = json.dumps(payload) + "\n"
        self.ser.write(msg.encode("utf-8"))
        raw_response = self.ser.readline().decode("utf-8").strip()

        if not raw_response:
            return {"status": "error", "msg": "Timeout seriale dal micro"}
        try:
            return json.loads(raw_response)
        except json.JSONDecodeError:
            return {
                "status": "error",
                "msg": f"Risposta malformata: {raw_response}",
            }

    def close(self):
        self.ser.close()


# Inizializzazione bridge seriale
bridge = MicrocontrollerBridge(SERIAL_PORT, BAUD_RATE)


def execute_tool(name: str, args: dict) -> dict:
    """Instrada la Tool Call generata dall'LLM al protocollo seriale."""
    if name == "read_sensor":
        return bridge.send_command({"cmd": "read", "pin": args.get("pin")})
    elif name == "write_actuator":
        return bridge.send_command({
            "cmd": "write",
            "pin": args.get("pin"),
            "val": args.get("state"),
        })
    return {"status": "error", "msg": f"Tool sconosciuto: {name}"}


def run_agent_turn(user_prompt: str):
    messages = [
        {
            "role": "system",
            "content": (
                "Sei un assistente embedded autonomo. Quando devi interagire con l'hardware, "
                "chiama i tool disponibili. Analizza sempre l'esito del tool prima di dare una risposta finale."
            ),
        },
        {"role": "user", "content": user_prompt},
    ]

    # 1. Prima chiamata: Ollama analizza il prompt e decide se chiamare un tool
    payload = {
        "model": MODEL_NAME,
        "messages": messages,
        "tools": TOOLS_SCHEMA,
        "stream": False,
    }

    res = requests.post(OLLAMA_URL, json=payload).json()
    msg = res.get("message", {})

    # 2. Controllo invocazione Tool
    if "tool_calls" in msg and msg["tool_calls"]:
        messages.append(msg)

        for tool in msg["tool_calls"]:
            func = tool["function"]
            fn_name = func["name"]
            fn_args = func["arguments"]

            print(f"[LLM -> TOOL] Invocazione: {fn_name}({fn_args})")
            tool_result = execute_tool(fn_name, fn_args)
            print(f"[HW -> PYTHON] Risultato: {tool_result}")

            # Restituisce l'esito dell'hardware a Ollama
            messages.append({
                "role": "tool",
                "content": json.dumps(tool_result),
            })

        # 3. Seconda chiamata: l'LLM formula la risposta finale basandosi sui dati reali
        final_res = requests.post(
            OLLAMA_URL,
            json={
                "model": MODEL_NAME,
                "messages": messages,
                "stream": False,
            },
        ).json()
        print("\n[AI]:", final_res["message"]["content"])
    else:
        print("\n[AI]:", msg.get("content"))


if __name__ == "__main__":
    try:
        # Esempio di prompt composito che richiede lettura e potenziale scrittura
        test_prompt = (
            "Verifica la lettura del sensore analogico sul pin A0. "
            "Se il valore supera 500, accendi il pin D13; altrimenti spegnilo. "
            "Spiegami cosa hai trovato e che azione hai intrapreso."
        )
        run_agent_turn(test_prompt)
    finally:
        bridge.close()

4. Esempio Firmware Minimo (MicroPython / C++)

Per l’RP2040 o qualsiasi micro ARM programmabile in MicroPython, il firmware ricevente si risolve in poche righe su sys.stdin:

Python

import json
import sys
from machine import ADC, Pin

led = Pin(13, Pin.OUT)
adc = ADC(Pin(26))  # A0 su Pico

while True:
    line = sys.stdin.readline()
    if not line:
        continue
    try:
        req = json.loads(line.strip())
        cmd = req.get("cmd")

        if cmd == "read":
            val = adc.read_u16() >> 6  # normalizzato 0-1023
            print(json.dumps({"status": "ok", "pin": req.get("pin"), "val": val}))
        elif cmd == "write":
            led.value(1 if req.get("val") == 1 else 0)
            print(json.dumps({"status": "ok", "pin": req.get("pin"), "val": req.get("val")}))
        else:
            print(json.dumps({"status": "error", "msg": "cmd non valido"}))
    except Exception as e:
        print(json.dumps({"status": "error", "msg": str(e)}))

Questo schema assicura che il modello non parli direttamente al silicio (il che causerebbe crash seriali e timing instabili), ma agisca all’interno di un’infrastruttura rigida, idempotente e loggabile anche su PostgreSQL per il monitoraggio storico.

Vuoi integrare il logging di telemetria e comandi su una tabella PostgreSQL?Sì

Aggiungiamo la persistenza su PostgreSQL per loggare telemetria dei sensori e comandi eseguiti dall’LLM.

Ecco l’integrazione di PostgreSQL nell’architettura.

L’obiettivo è duplice:

Audit & Safety: Tracciare ogni decisione presa dall’LLM (prompt utente, tool chiamato, argomenti passati, risposta del micro).
Telemetria & Storico: Salvare i campioni letti dai sensori in una struttura temporale, permettendo all’LLM di rispondere in futuro a domande del tipo “Qual è stata la media di temperatura/tensione delle ultime 2 ore?”.

1. Schema DDL (`schema.sql`)

Usiamo una tabella per i campioni temporali e una per il log delle decisioni/azioni dell’agente.

SQL

-- Estensione utile se in futuro vorrai indicizzare embedding delle decisioni
-- CREATE EXTENSION IF NOT EXISTS vector;

-- 1. Telemetria letta dai pin/sensori
CREATE TABLE IF NOT EXISTS hardware_telemetry (
    id BIGSERIAL PRIMARY KEY,
    timestamp TIMESTAMPTZ DEFAULT clock_timestamp(),
    pin VARCHAR(16) NOT NULL,
    val NUMERIC NOT NULL,
    raw_response JSONB
);

CREATE INDEX IF NOT EXISTS idx_telemetry_pin_time ON hardware_telemetry (pin, timestamp DESC);

-- 2. Audit delle azioni decise dall'LLM
CREATE TABLE IF NOT EXISTS llm_hardware_actions (
    id BIGSERIAL PRIMARY KEY,
    timestamp TIMESTAMPTZ DEFAULT clock_timestamp(),
    model VARCHAR(64) NOT NULL,
    user_prompt TEXT NOT NULL,
    tool_name VARCHAR(64) NOT NULL,
    tool_arguments JSONB NOT NULL,
    hardware_response JSONB NOT NULL,
    execution_status VARCHAR(16) NOT NULL -- 'SUCCESS', 'ERROR'
);

CREATE INDEX IF NOT EXISTS idx_actions_time ON llm_hardware_actions (timestamp DESC);

2. Modulo Database Python (`db_manager.py`)

Usiamo psycopg2 (o psycopg3) con gestione automatica della connessione e scrittura transazionale.

Python

import json
import psycopg2
from psycopg2.extras import Json

DB_CONFIG = {
    "dbname": "ollama_iot",
    "user": "postgres",
    "password": "your_password",
    "host": "localhost",
    "port": 5432,
}


class HardwareLogger:

    def __init__(self, config=DB_CONFIG):
        self.conn = psycopg2.connect(**config)
        self.conn.autocommit = True

    def log_telemetry(self, pin: str, val: float, raw_resp: dict):
        """Salva un campione letto da un sensore."""
        query = """
            INSERT INTO hardware_telemetry (pin, val, raw_response)
            VALUES (%s, %s, %s);
        """
        with self.conn.cursor() as cur:
            cur.execute(query, (pin, val, Json(raw_resp)))

    def log_action(
        self,
        model: str,
        prompt: str,
        tool: str,
        args: dict,
        hw_resp: dict,
        status: str,
    ):
        """Salva l'invocazione di un comando da parte dell'LLM."""
        query = """
            INSERT INTO llm_hardware_actions 
            (model, user_prompt, tool_name, tool_arguments, hardware_response, execution_status)
            VALUES (%s, %s, %s, %s, %s, %s);
        """
        with self.conn.cursor() as cur:
            cur.execute(query, (model, prompt, tool, Json(args), Json(hw_resp), status))

    def close(self):
        self.conn.close()

3. Orchestratore Integrato (`ollama_serial_pg_bridge.py`)

Aggiorniamo la funzione execute_tool e il loop principale per registrare gli eventi su Postgres non appena l’hardware risponde:

Python

import json
import requests
from db_manager import HardwareLogger
from ollama_serial_bridge import (
    MODEL_NAME,
    OLLAMA_URL,
    TOOLS_SCHEMA,
    bridge,
)  # Importa il bridge seriale precedente

db = HardwareLogger()


def execute_and_log_tool(
    tool_name: str, args: dict, user_prompt: str, model: str
) -> dict:
    """Esegue il comando sul micro e ne registra l'esito su PostgreSQL."""
    hw_result = {}

    if tool_name == "read_sensor":
        pin = args.get("pin")
        hw_result = bridge.send_command({"cmd": "read", "pin": pin})

        # Se la lettura ha avuto successo, logghiamo la telemetria
        if hw_result.get("status") == "ok" and "val" in hw_result:
            db.log_telemetry(
                pin=pin, val=hw_result["val"], raw_resp=hw_result
            )

    elif tool_name == "write_actuator":
        pin = args.get("pin")
        val = args.get("state")
        hw_result = bridge.send_command({"cmd": "write", "pin": pin, "val": val})

    else:
        hw_result = {"status": "error", "msg": f"Tool sconosciuto: {tool_name}"}

    # Audit log completo della decisione LLM -> Azione
    status = "SUCCESS" if hw_result.get("status") == "ok" else "ERROR"
    db.log_action(
        model=model,
        prompt=user_prompt,
        tool=tool_name,
        args=args,
        hw_resp=hw_result,
        status=status,
    )

    return hw_result


def run_agent_turn(user_prompt: str):
    messages = [
        {
            "role": "system",
            "content": (
                "Sei un assistente embedded autonomo. Usa i tool per interagire con l'hardware "
                "e motiva brevemente le azioni intraprese."
            ),
        },
        {"role": "user", "content": user_prompt},
    ]

    res = requests.post(
        OLLAMA_URL,
        json={
            "model": MODEL_NAME,
            "messages": messages,
            "tools": TOOLS_SCHEMA,
            "stream": False,
        },
    ).json()

    msg = res.get("message", {})

    if "tool_calls" in msg and msg["tool_calls"]:
        messages.append(msg)

        for tool in msg["tool_calls"]:
            func = tool["function"]
            fn_name = func["name"]
            fn_args = func["arguments"]

            print(f"[LLM -> TOOL] Invocazione: {fn_name}({fn_args})")
            tool_result = execute_and_log_tool(
                fn_name, fn_args, user_prompt, MODEL_NAME
            )
            print(f"[HW -> DB] Registrato su PostgreSQL: {tool_result}")

            messages.append({
                "role": "tool",
                "content": json.dumps(tool_result),
            })

        final_res = requests.post(
            OLLAMA_URL,
            json={
                "model": MODEL_NAME,
                "messages": messages,
                "stream": False,
            },
        ).json()
        print("\n[AI]:", final_res["message"]["content"])
    else:
        print("\n[AI]:", msg.get("content"))


if __name__ == "__main__":
    try:
        run_agent_turn(
            "Leggi il pin A0 e, se la tensione/valore è superiore a 300, commuta il pin D13 a 1."
        )
    finally:
        db.close()
        bridge.close()

Vantaggi dell’integrazione

Tracciabilità forense: Se l’attuatore fa scattare un carico (es. un relè), la tabella llm_hardware_actions conserva l’esatto prompt che ha scatenato l’evento e i parametri generati dall’LLM.
Estendibilità Tool: Ora puoi aggiungere un terzo tool a Ollama: get_sensor_history(pin, limit=5), consentendo al modello di interrogare direttamente PostgreSQL prima di decidere se attuare o meno un comando.

Vuoi implementare il tool per far leggere all’LLM lo storico da PostgreSQL?Sì

Aggiungiamo un tool a Ollama per consentirgli di interrogare lo storico delle letture da PostgreSQL prima di agire.

Aggiungere la consultazione dello storico permette al modello di non agire più su una singola lettura istantanea (che potrebbe essere affetta da spike o rumore elettrico dell’ADC), ma di valutare trend, medie e derive temporali prima di azionare un carico fisico.

Ecco l’aggiornamento con la query di aggregazione su PostgreSQL, la definizione del nuovo Tool e il loop ReAct multi-step.

1. Metodo nel Data Layer (`db_manager.py`)

Aggiungiamo a HardwareLogger un metodo che estrae gli ultimi N campioni e calcola metriche di sintesi (media, minimo, massimo) direttamente nel motore SQL:

Python

def get_recent_telemetry(self, pin: str, limit: int = 10) -> dict:
    """Recupera gli ultimi campioni e le metriche aggregate per un dato pin."""
    query = """
        WITH recent_samples AS (
            SELECT timestamp, val
            FROM hardware_telemetry
            WHERE pin = %s
            ORDER BY timestamp DESC
            LIMIT %s
        )
        SELECT 
            COUNT(*)::int AS count,
            COALESCE(ROUND(AVG(val), 2), 0) AS avg_val,
            COALESCE(MIN(val), 0) AS min_val,
            COALESCE(MAX(val), 0) AS max_val,
            COALESCE(json_agg(json_build_object('t', timestamp, 'v', val)), '[]'::json) AS samples
        FROM recent_samples;
    """
    with self.conn.cursor() as cur:
        cur.execute(query, (pin, limit))
        row = cur.fetchone()
        return {
            "status": "ok",
            "pin": pin,
            "sample_count": row[0],
            "average": float(row[1]),
            "min": float(row[2]),
            "max": float(row[3]),
            "recent_values": [s["v"] for s in row[4]],
        }

2. Definizione del Tool per Ollama

Aggiungiamo get_sensor_history allo schema dei tool:

Python

TOOLS_SCHEMA = [
    # ... read_sensor e write_actuator rimangono invariati ...
    {
        "type": "function",
        "function": {
            "name": "get_sensor_history",
            "description": (
                "Interroga il database PostgreSQL per ottenere lo storico recente delle letture "
                "di un sensore/pin, inclusi media, minimo, massimo e gli ultimi campioni registrati."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "pin": {
                        "type": "string",
                        "description": "Identificativo del pin (es. 'A0', 'A1').",
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Numero di letture recenti da considerare (default 5, max 50).",
                    },
                },
                "required": ["pin"],
            },
        },
    },
]

3. Orchestratore con Ciclo ReAct Multi-Step

Quando l’LLM ha a disposizione sia tool di lettura/storico che tool di attuazione, il processo decisionale richiede spesso chiamate sequenziali (es. 1. Leggi storico -> 2. Valuta trend -> 3. Aziona attuatore -> 4. Rispondi all’utente).

Sostituiamo il singolo if "tool_calls" con un ciclo iterativo (while):

Python

import json
import requests
from db_manager import HardwareLogger
from ollama_serial_bridge import MODEL_NAME, OLLAMA_URL, bridge

db = HardwareLogger()


def execute_and_log_tool(
    tool_name: str, args: dict, user_prompt: str, model: str
) -> dict:
    hw_result = {}

    if tool_name == "read_sensor":
        pin = args.get("pin")
        hw_result = bridge.send_command({"cmd": "read", "pin": pin})
        if hw_result.get("status") == "ok" and "val" in hw_result:
            db.log_telemetry(
                pin=pin, val=hw_result["val"], raw_resp=hw_result
            )

    elif tool_name == "write_actuator":
        pin = args.get("pin")
        val = args.get("state")
        hw_result = bridge.send_command({"cmd": "write", "pin": pin, "val": val})

    elif tool_name == "get_sensor_history":
        pin = args.get("pin")
        limit = args.get("limit", 5)
        hw_result = db.get_recent_telemetry(pin=pin, limit=limit)

    else:
        hw_result = {"status": "error", "msg": f"Tool sconosciuto: {tool_name}"}

    # Logging dell'audit di sistema su PostgreSQL
    status = "SUCCESS" if hw_result.get("status") == "ok" else "ERROR"
    db.log_action(
        model=model,
        prompt=user_prompt,
        tool=tool_name,
        args=args,
        hw_resp=hw_result,
        status=status,
    )

    return hw_result


def run_agent_loop(user_prompt: str, max_iterations: int = 5):
    """Ciclo di ragionamento ed esecuzione ReAct (Reasoning + Acting)."""
    messages = [
        {
            "role": "system",
            "content": (
                "Sei un assistente embedded per sistemi cyber-fisici con accesso diretto ad hardware e database. "
                "Per prendere decisioni stabili sull'hardware, consulta lo storico con get_sensor_history "
                "o acquisisci nuovi dati con read_sensor prima di commutare carichi con write_actuator. "
                "Quando hai terminato le operazioni necessarie, formula una spiegazione chiara e concisa."
            ),
        },
        {"role": "user", "content": user_prompt},
    ]

    for iteration in range(max_iterations):
        res = requests.post(
            OLLAMA_URL,
            json={
                "model": MODEL_NAME,
                "messages": messages,
                "tools": TOOLS_SCHEMA,
                "stream": False,
            },
        ).json()

        msg = res.get("message", {})
        tool_calls = msg.get("tool_calls", [])

        # Se il modello non intende chiamare altri tool, ha formulato la risposta finale
        if not tool_calls:
            print(f"\n[AI Risposta Finale]:\n{msg.get('content')}")
            break

        # Aggiunge la proposta di chiamata del modello allo stack messaggi
        messages.append(msg)

        # Esegue ogni tool richiesto e retroalimenta i risultati
        for call in tool_calls:
            fn_name = call["function"]["name"]
            fn_args = call["function"]["arguments"]

            print(
                f"[Iter {iteration + 1}] LLM invoca: {fn_name} con args {fn_args}"
            )
            result = execute_and_log_tool(
                fn_name, fn_args, user_prompt, MODEL_NAME
            )
            print(f"[Risultato Tool]: {result}")

            messages.append({
                "role": "tool",
                "content": json.dumps(result),
            })


if __name__ == "__main__":
    try:
        # Prompt che forza l'agente a verificare prima la cronologia prima di intervenire
        prompt = (
            "Verifica la cronologia degli ultimi 5 campioni del sensore sul pin A0. "
            "Se il valore medio calcolato supera la soglia di 450, attiva l'uscita D13, "
            "altrimenti assicurati che rimanga spenta (0). Riassumi la media rilevata e l'azione svolta."
        )
        run_agent_loop(prompt)
    finally:
        db.close()
        bridge.close()

Cosa accade nel loop (Trace tipico di esecuzione)

Step 1: Il modello riceve il prompt e produce una Tool Call: get_sensor_history(pin="A0", limit=5).
Step 2: Python interroga Postgres, ricava {"sample_count": 5, "avg_val": 482.4, "max": 510, ...} e lo appende ai messaggi con role: "tool".
Step 3: Il modello rilegge il contesto, nota che la media (482.4>450) soddisfa la condizione e chiama write_actuator(pin="D13", state=1).
Step 4: Il bridge invia il comando via seriale al microcontrollore, riceve {"status": "ok"}, registra l’evento di scrittura su Postgres e lo rimanda all’LLM.
Step 5: Non avendo altri comandi da inviare, l’LLM genera la sintesi finale:“Ho analizzato gli ultimi 5 campioni sul pin A0: la media rilevata è di 482.4, che supera la soglia prefissata di 450. Di conseguenza, ho attivato l’uscita D13 impostandola sullo stato logico 1.”

Configuriamo uno script demone in background che campiona i sensori a intervalli regolari e attiva l’LLM solo su anomalie.

Questo approccio (spesso chiamato pattern Event-Driven AI o Edge-to-LLM Alerting) è fondamentale per un sistema di produzione. Interrogare l’LLM a ogni singolo ciclo di clock o lettura dell’ADC saturerebbe la GPU e introdurrebbe latenze inutili.

La separazione dei ruoli è netta:

Loop rapido in background (Python deterministico): campiona a intervalli costanti (1 Hz o più rapido), inserisce i dati in PostgreSQL ed effettua un sanity check numerico (soglie min/max, derivata o deviazione standard).
Attivazione cognitiva (Ollama ReAct): si sveglia solo quando il pre-filtro individua un’anomalia, riceve il contesto storico e la natura dell’allarme, decide le contromisure da attuare sull’hardware e registra la diagnosi.

1. Tabella Diagnostica & Incidenti (`schema_incidents.sql`)

Aggiungiamo a PostgreSQL una tabella per tracciare le anomalie rilevate e le risoluzioni decise dall’LLM:

SQL

CREATE TABLE IF NOT EXISTS system_incidents (
    id BIGSERIAL PRIMARY KEY,
    timestamp TIMESTAMPTZ DEFAULT clock_timestamp(),
    pin VARCHAR(16) NOT NULL,
    trigger_value NUMERIC NOT NULL,
    threshold_type VARCHAR(32) NOT NULL, -- 'UPPER_BOUND', 'RAPID_SPIKE', etc.
    llm_diagnostic TEXT,
    resolved BOOLEAN DEFAULT FALSE
);

CREATE INDEX IF NOT EXISTS idx_incidents_time ON system_incidents (timestamp DESC);

2. Funzione nel Data Layer (`db_manager.py`)

Aggiungiamo a HardwareLogger il salvataggio degli incidenti:

Python

def log_incident(
    self, pin: str, trigger_val: float, threshold_type: str, diagnostic: str
):
    query = """
        INSERT INTO system_incidents (pin, trigger_value, threshold_type, llm_diagnostic, resolved)
        VALUES (%s, %s, %s, %s, TRUE);
    """
    with self.conn.cursor() as cur:
        cur.execute(query, (pin, trigger_val, threshold_type, diagnostic))

3. Demone di Monitoraggio (`monitor_daemon.py`)

Questo script gira come servizio in background (es. via systemd o supervisord). Implementa un cooldown per evitare che l’LLM venga bombardato ripetutamente se un valore rimane fuori soglia per diversi campionamenti consecutivi.

Python

import json
import logging
import time
from db_manager import HardwareLogger
from ollama_serial_bridge import MODEL_NAME, OLLAMA_URL, TOOLS_SCHEMA, bridge

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s"
)

# --- Configurazione Monitoraggio ---
MONITOR_PIN = "A0"
SAMPLE_INTERVAL_SEC = 2.0  # Frequenza di campionamento
THRESHOLD_MAX = 700.0  # Valore critico superiore
SPIKE_DELTA = 150.0  # Variazione improvvisa tra due letture consecutive
COOLDOWN_SEC = 30.0  # Tempo minimo tra due interventi dell'LLM

db = HardwareLogger()


def execute_llm_remedy(incident_context: dict) -> str:
    """Attiva l'LLM in modalità ReAct per gestire l'anomalia hardware."""
    system_prompt = (
        "Sei il supervisore di emergenza di un sistema cyber-fisico. "
        "È stata rilevata un'anomalia grave sui sensori hardware. "
        "Puoi interrogare lo storico recente tramite get_sensor_history o "
        "intervenire immediatamente sugli attuatori tramite write_actuator per mettere il sistema in sicurezza. "
        "Fornisci una breve diagnosi tecnica al termine."
    )

    user_prompt = (
        f"ALLARME HARDWARE: Il pin {incident_context['pin']} ha registrato un valore anomalo "
        f"di {incident_context['val']} (Condizione: {incident_context['reason']}). "
        f"Valore precedente: {incident_context.get('prev_val', 'N/D')}. "
        "Analizza la situazione, esegui le azioni correttive necessarie sul circuito e spiega cosa hai fatto."
    )

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt},
    ]

    logging.warning(
        "Attivazione LLM per incidente sul pin %s", incident_context["pin"]
    )

    # Ciclo ReAct a massimo 4 iterazioni per non bloccare il demone
    for _ in range(4):
        res = requests.post(
            OLLAMA_URL,
            json={
                "model": MODEL_NAME,
                "messages": messages,
                "tools": TOOLS_SCHEMA,
                "stream": False,
            },
            timeout=30,
        ).json()

        msg = res.get("message", {})
        tool_calls = msg.get("tool_calls", [])

        if not tool_calls:
            return msg.get("content", "Nessuna diagnosi fornita.")

        messages.append(msg)

        for call in tool_calls:
            fn_name = call["function"]["name"]
            fn_args = call["function"]["arguments"]

            # Esecuzione del tool con log su Postgres
            if fn_name == "write_actuator":
                hw_res = bridge.send_command({
                    "cmd": "write",
                    "pin": fn_args.get("pin"),
                    "val": fn_args.get("state"),
                })
            elif fn_name == "get_sensor_history":
                hw_res = db.get_recent_telemetry(
                    fn_args.get("pin"), fn_args.get("limit", 5)
                )
            elif fn_name == "read_sensor":
                hw_res = bridge.send_command({
                    "cmd": "read",
                    "pin": fn_args.get("pin"),
                })
            else:
                hw_res = {"status": "error", "msg": "Tool sconosciuto"}

            db.log_action(
                MODEL_NAME,
                user_prompt,
                fn_name,
                fn_args,
                hw_res,
                hw_res.get("status", "ok").upper(),
            )

            messages.append({"role": "tool", "content": json.dumps(hw_res)})

    return "Intervento completato (iterazioni massime raggiunte)."


def run_daemon():
    logging.info("Demone di monitoraggio avviato. Pin: %s", MONITOR_PIN)
    last_val = None
    last_alert_time = 0.0

    try:
        while True:
            # 1. Campionamento seriale a basso overhead
            resp = bridge.send_command({"cmd": "read", "pin": MONITOR_PIN})

            if resp.get("status") == "ok" and "val" in resp:
                val = float(resp["val"])
                # Log telemetrico standard su Postgres
                db.log_telemetry(pin=MONITOR_PIN, val=val, raw_resp=resp)

                # 2. Controllo deterministico delle anomalie (filtro a soglia / spike)
                is_anomaly = False
                anomaly_reason = ""

                if val > THRESHOLD_MAX:
                    is_anomaly = True
                    anomaly_reason = (
                        f"Valore {val} supera soglia max {THRESHOLD_MAX}"
                    )
                elif last_val is not None and abs(val - last_val) >= SPIKE_DELTA:
                    is_anomaly = True
                    anomaly_reason = f"Spike improvviso di {abs(val - last_val)} unita (da {last_val} a {val})"

                # 3. Trigger LLM se anomalo e fuori periodo di cooldown
                current_time = time.time()
                if (
                    is_anomaly
                    and (current_time - last_alert_time) > COOLDOWN_SEC
                ):
                    incident_info = {
                        "pin": MONITOR_PIN,
                        "val": val,
                        "prev_val": last_val,
                        "reason": anomaly_reason,
                    }
                    diagnostic = execute_llm_remedy(incident_info)
                    logging.info("Diagnosi LLM registrata: %s", diagnostic)

                    # Registra l'incidente risolto nel DB
                    db.log_incident(
                        MONITOR_PIN, val, anomaly_reason, diagnostic
                    )
                    last_alert_time = current_time

                last_val = val
            else:
                logging.error("Errore lettura seriale: %s", resp.get("msg"))

            time.sleep(SAMPLE_INTERVAL_SEC)

    except KeyboardInterrupt:
        logging.info("Interruzione manuale ricevuta. Arresto in corso...")
    finally:
        db.close()
        bridge.close()


if __name__ == "__main__":
    run_daemon()

4. Gestione via `systemd` (Server Ollama 1)

Per renderlo resiliente a riavvii del server o disconnessioni temporanee del cavo USB, puoi creare un servizio Linux:

/etc/systemd/system/hardware-agent.service:

Ini, TOML

[Unit]
Description=Hardware Monitoring Daemon with Ollama Agent
After=network.target postgresql.service ollama.service

[Service]
Type=simple
User=tuo_utente
WorkingDirectory=/opt/ollama_hardware_bridge
ExecStart=/opt/ollama_hardware_bridge/venv/bin/python monitor_daemon.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Bash

sudo systemctl daemon-reload
sudo systemctl enable --now hardware-agent.service

Comportamento Operativo

Stato Nominale: Il demone legge il sensore ogni 2 secondi, salva su PostgreSQL con un overhead CPU quasi nullo (< 0.1%), mentre Ollama resta completamente inattivo.
Evento Critico (es. A0=780): Il filtro intercetta il superamento della soglia e invoca immediatamente l’LLM passando contesto ed evento.
Azione & Traccia: L’LLM valuta lo stato, decide di invocare write_actuator(pin="D13", state=1) (es. relè ventola o stacco emergenza), e genera la spiegazione tecnica che viene memorizzata sia nei log di sistema sia nella tabella system_incidents.

Come implementare un watchdog timer sul firmware del microcontrollore per sicurezza nel caso il demone Python o il server vadano in crash?

Nei sistemi cyber-fisici guidati da software ad alto livello (come Python o un LLM), affidarsi solo al demone per la sicurezza è un rischio: se il server si blocca, va in kernel panic o il processo Python va in stallo mentre un’uscita di potenza è accesa, il carico resterebbe alimentato indefinitamente.

Per risolvere questo problema si implementa un Safety Watchdog a doppio livello:

Hardware Watchdog (WDT): Protegge il microcontrollore da loop infiniti o freeze del firmware interno.
Communication Heartbeat Watchdog (Software/Timer): Protegge l’impianto da freeze o crash del server/demone Python. Se il microcontrollore non riceve un “ping/heartbeat” valido entro una finestra temporale (es. 5 secondi), forza autonomamente tutte le uscite nello stato di sicurezza predefinito (Fail-Safe State).

1. Protocollo Seriale: Aggiunta dell’Heartbeat

Aggiungiamo un comando ping o consideriamo qualsiasi comando valido come reset del timer di comunicazione:

Da Python (ogni ciclo nominale): {"cmd":"ping"}\n o un comando ordinario di lettura {"cmd":"read","pin":"A0"}\n.
Dal Micro: {"status":"ok","pong":true}\n

2. Firmware Microcontrollore (C++ / Arduino Core / RP2040 / STM32)

Ecco l’implementazione completa con azzeramento uscite di sicurezza e WDT interno:

C++

#include <Arduino.h>
#include <ArduinoJson.h>

#if defined(ARDUINO_ARCH_RP2040)
  #include <hardware/watchdog.h>
#elif defined(AVR)
  #include <avr/wdt.h>
#endif

// --- Parametri di Sicurezza ---
const unsigned long COMMS_TIMEOUT_MS = 5000; // 5 secondi senza heartbeat = Fail-Safe
unsigned long last_valid_comms_time = 0;
bool system_in_failsafe = false;

// Pin attuatori critici
const int RELAY_PIN = 13; 

void enter_failsafe() {
    if (!system_in_failsafe) {
        system_in_failsafe = true;
        // STATO DI SICUREZZA: Spegnimento immediato di tutti i carichi
        digitalWrite(RELAY_PIN, LOW);
    }
}

void reset_failsafe() {
    system_in_failsafe = false;
}

void setup() {
    Serial.begin(115200);
    pinMode(RELAY_PIN, OUTPUT);
    digitalWrite(RELAY_PIN, LOW); // Default spento

    // Inizializzazione Hardware Watchdog (es. 2000 ms su RP2040 o 8000 ms su AVR)
    #if defined(ARDUINO_ARCH_RP2040)
        watchdog_enable(2000, 1);
    #elif defined(AVR)
        wdt_enable(WDTO_2S);
    #endif

    last_valid_comms_time = millis();
}

void loop() {
    // 1. Reset dell'Hardware Watchdog interno
    #if defined(ARDUINO_ARCH_RP2040)
        watchdog_update();
    #elif defined(AVR)
        wdt_reset();
    #endif

    // 2. Controllo Heartbeat dal server/Python (Communication Timeout)
    if (millis() - last_valid_comms_time > COMMS_TIMEOUT_MS) {
        enter_failsafe();
    }

    // 3. Gestione comandi da USB CDC
    if (Serial.available() > 0) {
        String line = Serial.readStringUntil('\n');
        line.trim();
        if (line.length() == 0) return;

        StaticJsonDocument<256> doc;
        DeserializationError err = deserializeJson(doc, line);

        if (err) {
            Serial.println("{\"status\":\"error\",\"msg\":\"JSON malformato\"}");
            return;
        }

        // Segnale valido ricevuto: ricarica il timer di comunicazione
        last_valid_comms_time = millis();
        reset_failsafe();

        const char* cmd = doc["cmd"];

        if (strcmp(cmd, "ping") == 0) {
            Serial.println("{\"status\":\"ok\",\"pong\":true}");
        } 
        else if (strcmp(cmd, "read") == 0) {
            int val = analogRead(A0);
            Serial.print("{\"status\":\"ok\",\"pin\":\"A0\",\"val\":");
            Serial.print(val);
            Serial.println("}");
        } 
        else if (strcmp(cmd, "write") == 0) {
            // Se in condizione di blocco non accetta nuove accensioni
            int val = doc["val"] | 0;
            digitalWrite(RELAY_PIN, val == 1 ? HIGH : LOW);
            Serial.print("{\"status\":\"ok\",\"pin\":\"D13\",\"val\":");
            Serial.print(val);
            Serial.println("}");
        } 
        else {
            Serial.println("{\"status\":\"error\",\"msg\":\"Comando sconosciuto\"}");
        }
    }
}

3. Firmware in MicroPython (Alternativa per Raspberry Pi Pico / RP2040)

Se usi MicroPython sull’RP2040, puoi usare machine.WDT e un timer software non bloccante:

Python

import json
import sys
import time
from machine import ADC, Pin, WDT

# WDT Hardware (timeout massimo consentito su RP2040: circa 8388 ms)
hw_wdt = WDT(timeout=4000)

relay = Pin(13, Pin.OUT)
relay.value(0)
adc = ADC(Pin(26))

COMMS_TIMEOUT_MS = 5000
last_comms_ms = time.ticks_ms()

while True:
    hw_wdt.feed()

    # Verifica scadenze heartbeat da Linux
    elapsed = time.ticks_diff(time.ticks_ms(), last_comms_ms)
    if elapsed > COMMS_TIMEOUT_MS:
        relay.value(0)  # Intervento fail-safe

    # Polling I/O seriale (senza bloccare il loop)
    # Su MicroPython uso select/poll su sys.stdin per evitare readline bloccante
    line = sys.stdin.readline()
    if not line:
        time.sleep_ms(10)
        continue

    try:
        req = json.loads(line.strip())
        last_comms_ms = time.ticks_ms()  # Ricarica watchdog comunicazione
        cmd = req.get("cmd")

        if cmd == "ping":
            print(json.dumps({"status": "ok", "pong": True}))
        elif cmd == "read":
            val = adc.read_u16() >> 6
            print(
                json.dumps({"status": "ok", "pin": req.get("pin"), "val": val})
            )
        elif cmd == "write":
            relay.value(1 if req.get("val") == 1 else 0)
            print(
                json.dumps(
                    {"status": "ok", "pin": req.get("pin"), "val": relay.value()}
                )
            )
        else:
            print(json.dumps({"status": "error", "msg": "Comando ignoto"}))
    except Exception as e:
        print(json.dumps({"status": "error", "msg": str(e)}))

4. Gestione lato Python (Demone)

Nel demone Python non serve aggiungere polling extra se l’intervallo di campionamento è inferiore a COMMS_TIMEOUT_MS:

Se il demone legge il pin ogni 2 secondi ({"cmd":"read"}), il timer del micro viene ricaricato regolarmente.
Se il demone fa pause lunghe o se l’LLM impiega troppi secondi a completare un’inferenza su Ollama, il bridge seriale deve inviare un comando ping rapido per evitare lo scatto prematuro del fail-safe.

Questo garantisce che, anche in caso di kernel panic, crash di Ollama o interruzione accidentale del cavo seriale, il microcontrollore non lasci mai attivi riscaldatori, motori o relè critici.

1. Il Modello: Function Calling e Structured Output

2. L’Hardware: AVR vs ARM (Cortex-M)

3. Architettura a Livelli: Regola di Sicurezza

2. L’Hardware: AVR vs ARM (Cortex-M)

3. Architettura a Livelli: Regola di Sicurezza

Un test “Hello World” da cui partire

1. Protocollo Seriale (Microcontroller)

2. Definizione dei Tool per Ollama

3. Orchestratore Python (ollama_serial_bridge.py)

4. Esempio Firmware Minimo (MicroPython / C++)

1. Schema DDL (schema.sql)

2. Modulo Database Python (db_manager.py)

3. Orchestratore Integrato (ollama_serial_pg_bridge.py)

Vantaggi dell’integrazione

1. Metodo nel Data Layer (db_manager.py)

2. Definizione del Tool per Ollama

3. Orchestratore con Ciclo ReAct Multi-Step

Cosa accade nel loop (Trace tipico di esecuzione)

1. Tabella Diagnostica & Incidenti (schema_incidents.sql)

2. Funzione nel Data Layer (db_manager.py)

3. Demone di Monitoraggio (monitor_daemon.py)

4. Gestione via systemd (Server Ollama 1)

Comportamento Operativo

1. Protocollo Seriale: Aggiunta dell’Heartbeat

2. Firmware Microcontrollore (C++ / Arduino Core / RP2040 / STM32)

3. Firmware in MicroPython (Alternativa per Raspberry Pi Pico / RP2040)

4. Gestione lato Python (Demone)

3. Orchestratore Python (`ollama_serial_bridge.py`)

1. Schema DDL (`schema.sql`)

2. Modulo Database Python (`db_manager.py`)

3. Orchestratore Integrato (`ollama_serial_pg_bridge.py`)

1. Metodo nel Data Layer (`db_manager.py`)

1. Tabella Diagnostica & Incidenti (`schema_incidents.sql`)

2. Funzione nel Data Layer (`db_manager.py`)

3. Demone di Monitoraggio (`monitor_daemon.py`)

4. Gestione via `systemd` (Server Ollama 1)