Gestire gli Errori dell’Agente: Un Tutorial Pratico

Introduzione alla Gestione degli Errori degli Agenti

Nel mondo degli agenti AI, una solida gestione degli errori non è solo una buona pratica; è una necessità. Man mano che gli agenti interagiscono con ambienti dinamici, API esterne e dati complessi, si troveranno inevitabilmente ad affrontare situazioni impreviste. Dalle interruzioni di rete e dalle risposte API non valide agli input degli utenti malformati e alle incoerenze logiche, un agente ben progettato deve essere in grado di recuperare in modo elegante, informare o adattarsi. Senza una gestione degli errori efficace, un agente può rapidamente diventare fragile, fallendo silenziosamente o bloccandosi completamente, portando a esperienze utente scadenti e operazioni inaffidabili.

Questo tutorial esplorerà gli aspetti pratici della gestione degli errori degli agenti. Esploreremo varie strategie, dimostreremo insidie comuni e forniremo esempi concreti utilizzando Python, un linguaggio popolare per la creazione di agenti AI. Il nostro obiettivo è fornirti le conoscenze e gli strumenti per costruire agenti più resilienti, affidabili e facili da usare.

Perché la Gestione degli Errori è Cruciale per gli Agenti?

• Affidabilità: Previene i crash e garantisce un’operazione continua.
• Esperienza Utente: Fornisce feedback significativi invece di errori criptici.
• Debugging: Centralizza il logging degli errori, semplificando l’identificazione e la risoluzione dei problemi.
• Gestione delle Risorse: Permette una corretta pulizia (ad es., chiusura delle connessioni, rilascio dei lock).
• Adattabilità: Consente agli agenti di ripetere le operazioni o cambiare strategie quando affrontano guasti temporanei.

Comprendere i Comuni Scenari di Errore degli Agenti

Prima di esplorare l’implementazione, categorizziamo i tipi di errori che un agente incontra comunemente:

1. Errori dei Servizi Esterni (API, Database, Rete)

Questi sono forse i più frequenti. Un agente spesso dipende da servizi esterni per dati, calcoli o azioni. Gli esempi includono:

• Problemi di rete: Timeout di connessione, fallimenti nella risoluzione DNS, host irraggiungibile.
• Errori API: HTTP 4xx (errori client come 404 Not Found, 401 Unauthorized, 400 Bad Request), HTTP 5xx (errori server come 500 Internal Server Error, 503 Service Unavailable), limitazione di frequenza (429 Too Many Requests).
• Errori di Database: Fallimenti di connessione, timeout delle query, violazioni di vincoli.

2. Errori di Validazione Input/Output

Gli agenti elaborano varie forme di input, dai prompt degli utenti ai dati dei sensori. Input non validi possono portare a comportamenti imprevisti:

• Input dell’utente malformato: Input non numerico dove è atteso un numero, formati di data non validi.
• Parametri mancanti: Argomenti richiesti non forniti.
• Valori fuori intervallo: Una lettura di temperatura fisicamente impossibile.

3. Errori di Logica Interna

Questi errori derivano dal codice o dallo stato dell’agente:

• Falli di asserzione: Condizioni che ci si aspettava fossero vere non lo sono.
• Indice fuori dai limiti: Tentativo di accedere a un elemento oltre la lunghezza di una lista.
• Errori di tipo: Operare su dati con un tipo errato (ad es., tentare di sommare una stringa a un intero).
• Esaurimento delle risorse: Esaurimento di memoria o descrittori di file.

4. Cambiamenti Ambientali Inaspettati

Gli agenti in ambienti dinamici potrebbero imbattersi in situazioni per le quali non sono esplicitamente programmati:

• File non trovato: Un file di configurazione necessario è mancante.
• Problemi di permessi: L’agente non ha l’accesso necessario a una risorsa.
• Guasti hardware: Malfunzionamento del sensore o errori del disco.

Fondamenti della Gestione degli Errori in Python

Il meccanismo principale di Python per la gestione degli errori è il blocco try-except-finally.

import logging

# Configurare il logging di base
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def divide_numbers(a, b):
 try:
 result = a / b
 logging.info(f"Divisione riuscita: {a} / {b} = {result}")
 return result
 except ZeroDivisionError:
 logging.error("Errore: Impossibile dividere per zero!")
 return None
 except TypeError:
 logging.error("Errore: Entrambi gli input devono essere numeri.")
 return None
 except Exception as e:
 # Cattura qualsiasi altro errore imprevisto
 logging.error(f"Si è verificato un errore imprevisto: {e}")
 return None
 finally:
 # Questo blocco viene sempre eseguito, indipendentemente dal fatto che si sia verificata o meno un'eccezione
 logging.info("Tentativo di divisione concluso.")

# Esempi:
print(divide_numbers(10, 2)) # Divisione riuscita
print(divide_numbers(10, 0)) # ZeroDivisionError
print(divide_numbers(10, "a")) # TypeError
print(divide_numbers(None, 5)) # Un altro TypeError

Analizziamo i componenti:

• try: Il codice che potrebbe sollevare un’eccezione.
• except ExceptionType as e: Cattura tipi specifici di eccezioni. Puoi avere più blocchi except per diversi tipi di errore. La parte as e ti consente di accedere all’oggetto eccezione per ulteriori dettagli.
• except Exception as e: Una cattura generale per qualsiasi altra eccezione. È buona prassi catturare prima eccezioni specifiche e poi una generale.
• finally: Il codice in questo blocco verrà sempre eseguito, sia che si sia verificata un’eccezione o meno. È ideale per operazioni di pulizia (ad es., chiusura dei file, rilascio delle risorse).
• else (facoltativo): Il codice qui viene eseguito solo se il blocco try viene completato senza eccezioni.

Strategie Pratiche per la Gestione degli Errori degli Agenti

1. Gestione e Logging di Eccezioni Specifiche

È sempre consigliabile catturare eccezioni specifiche piuttosto che generiche quando possibile. Questo consente un recupero mirato e un logging più chiaro.

import requests
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def fetch_data_from_api(url, timeout=5):
 try:
 response = requests.get(url, timeout=timeout)
 response.raise_for_status() # Solleva HTTPError per risposte errate (4xx o 5xx)
 logging.info(f"Dati recuperati con successo da {url}")
 return response.json()
 except requests.exceptions.Timeout:
 logging.warning(f"Richiesta API scaduta per {url}")
 return None
 except requests.exceptions.ConnectionError as e:
 logging.error(f"Errore di connessione di rete per {url}: {e}")
 return None
 except requests.exceptions.HTTPError as e:
 logging.error(f"Errore HTTP {e.response.status_code} per {url}: {e.response.text}")
 return None
 except requests.exceptions.RequestException as e:
 # Cattura qualsiasi altro errore relativo alla richiesta
 logging.error(f"Si è verificato un errore di richiesta imprevisto per {url}: {e}")
 return None
 except ValueError as e:
 # Errore di decodifica JSON se response.json() fallisce
 logging.error(f"Impossibile decodificare JSON da {url}: {e}")
 return None

# Esempio di utilizzo:
# print(fetch_data_from_api("https://api.github.com/users/octocat"))
# print(fetch_data_from_api("https://nonexistent-api.com")) # ConnectionError
# print(fetch_data_from_api("https://httpbin.org/status/500")) # HTTPError
# print(fetch_data_from_api("https://httpbin.org/delay/6", timeout=2)) # Timeout

2. Ritenti con Backoff Esponenziale

Per errori transitori (come glitch di rete, temporanea indisponibilità del servizio, o limiti di frequenza), ripetere l’operazione dopo un ritardo è una strategia efficace. Il backoff esponenziale aumenta il ritardo tra i retry, evitando di sovraccaricare il servizio e permettendogli di riprendersi.

import requests
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def fetch_data_with_retries(url, max_retries=3, initial_delay=1):
 for attempt in range(max_retries):
 try:
 response = requests.get(url, timeout=5)
 response.raise_for_status()
 logging.info(f"Tentativo {attempt + 1}: Dati recuperati con successo da {url}")
 return response.json()
 except (requests.exceptions.Timeout, requests.exceptions.ConnectionError, requests.exceptions.HTTPError) as e:
 status_code = getattr(e, 'response', None) and e.response.status_code
 if status_code == 429: # Limite di frequenza
 logging.warning(f"Tentativo {attempt + 1}: Limite di frequenza raggiunto per {url}. Riprovo...")
 elif status_code and 500 <= status_code < 600: # Errore del server
 logging.warning(f"Tentativo {attempt + 1}: Errore del server ({status_code}) per {url}. Riprovo...")
 elif isinstance(e, requests.exceptions.Timeout): # Timeout
 logging.warning(f"Tentativo {attempt + 1}: Timeout per {url}. Riprovo...")
 elif isinstance(e, requests.exceptions.ConnectionError): # Errore di connessione
 logging.warning(f"Tentativo {attempt + 1}: Errore di connessione per {url}. Riprovo...")
 else:
 # Per altri errori HTTP (es. 404, 400), non riprovare per impostazione predefinita
 logging.error(f"Tentativo {attempt + 1}: Errore HTTP irreversibile {status_code} per {url}. Interruzione dei tentativi.")
 return None

 if attempt < max_retries - 1:
 delay = initial_delay * (2 ** attempt) # Backoff esponenziale
 logging.info(f"Attesa di {delay:.1f} secondi prima del prossimo tentativo...")
 time.sleep(delay)
 else:
 logging.error(f"Tutti i {max_retries} tentativi hanno fallito per {url}.")
 return None
 except requests.exceptions.RequestException as e:
 logging.error(f"Si è verificato un errore di richiesta irreversibile per {url}: {e}. Interruzione.")
 return None
 except ValueError as e:
 logging.error(f"Impossibile decodificare JSON da {url}: {e}. Interruzione.")
 return None
 return None

# Test con un'API instabile o un endpoint con limite di frequenza
# print(fetch_data_with_retries("https://httpbin.org/status/503")) # Dovrebbe riprovare
# print(fetch_data_with_retries("https://httpbin.org/delay/1", max_retries=1)) # Dovrebbe avere successo immediatamente

3. Validazione e Sanitizzazione degli Input

Prevenire errori validando gli input nel modo più precoce possibile. Questo è particolarmente importante per gli agenti rivolti agli utenti.

import re
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def process_user_command(command_str):
 if not isinstance(command_str, str):
 logging.error("Tipo di comando non valido: Deve essere una stringa.")
 raise ValueError("Il comando deve essere una stringa.")
 
 command_str = command_str.strip().lower()

 if not command_str:
 logging.warning("Comando vuoto ricevuto.")
 return "Per favore fornisci un comando."

 # Esempio: Controlla un pattern specifico
 if re.match(r"^set temperature \d+\.$", command_str):
 try:
 temp_value = int(command_str.split(' ')[2].replace('.', ''))
 if 0 <= temp_value <= 100:
 logging.info(f"Impostazione della temperatura a {temp_value}°C.")
 return f"Temperatura impostata a {temp_value}°C."
 else:
 logging.error(f"Valore della temperatura non valido: {temp_value}. Deve essere compreso tra 0 e 100.")
 return "La temperatura deve essere compresa tra 0 e 100 gradi Celsius."
 except (ValueError, IndexError):
 logging.error(f"Comando 'set temperature' malformato: {command_str}")
 return "Formato del comando 'set temperature' non valido. Ci si aspetta 'set temperature [valore]..'"
 elif command_str == "status":
 logging.info("Verifica dello stato del dispositivo.")
 return "Dispositivo operativo."
 else:
 logging.warning(f"Comando sconosciuto ricevuto: '{command_str}'")
 return "Non comprendo quel comando."

# Esempi:
print(process_user_command(" Set Temperature 25. "))
print(process_user_command("set temperature 105."))
print(process_user_command("set temperature abc."))
print(process_user_command("status"))
print(process_user_command("turn on lights"))
# process_user_command(123) # Questo genererà un ValueError

4. Eccezioni Personalizzate per la Logica Specifica degli Agenti

Per errori specifici per il dominio del tuo agente, definisci eccezioni personalizzate. Questo migliora la leggibilità del codice e consente una gestione degli errori più granulare a livelli superiori dell'architettura del tuo agente.

import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class AgentError(Exception):
 """Eccezione base per tutti gli errori relativi agli agenti."""
 pass

class SensorReadError(AgentError):
 """Sollevata quando un sensore non fornisce dati validi."""
 def __init__(self, sensor_id, message="Errore nella lettura dal sensore."):
 self.sensor_id = sensor_id
 self.message = f"{message} ID sensore: {sensor_id}"
 super().__init__(self.message)

class ActionFailedError(AgentError):
 """Sollevata quando un'azione dell'agente non può essere completata."""
 def __init__(self, action_name, reason="Motivo sconosciuto."):
 self.action_name = action_name
 self.reason = reason
 self.message = f"Azionamento '{action_name}' fallito: {reason}"
 super().__init__(self.message)

def read_temperature_sensor(sensor_id):
 # Simula la lettura del sensore, a volte fallisce
 if sensor_id == "temp_001":
 # Simula una lettura riuscita
 return 22.5
 elif sensor_id == "temp_002":
 # Simula un errore del sensore
 raise SensorReadError(sensor_id, "Malfunction hardware rilevata.")
 else:
 raise SensorReadError(sensor_id, "Sensore non trovato.")

def activate_heater(target_temp):
 if target_temp > 30:
 raise ActionFailedError("activate_heater", "Temperatura target troppo alta.")
 logging.info(f"Riscaldatore attivato per raggiungere {target_temp}°C.")
 return True

def agent_main_loop():
 try:
 current_temp = read_temperature_sensor("temp_001")
 logging.info(f"Temperatura attuale: {current_temp}°C")
 activate_heater(25)

 # Questo fallirà
 read_temperature_sensor("temp_002")

 except SensorReadError as e:
 logging.error(f"L'agente non può procedere a causa di un errore del sensore: {e.sensor_id} - {e.message}")
 # L'agente potrebbe passare a un sensore di riserva o avvisare un operatore umano
 except ActionFailedError as e:
 logging.error(f"L'agente non è riuscito a eseguire l'azione '{e.action_name}': {e.reason}")
 # L'agente potrebbe tentare un'azione alternativa o registrare per un intervento manuale
 except AgentError as e:
 logging.error(f"Si è verificato un errore generale dell'agente: {e}")
 except Exception as e:
 logging.critical(f"Si è verificato un errore critico non gestito: {e}")

agent_main_loop()
```

import logging
import sys
# import sentry_sdk # Decommenta e configura per un'integrazione reale con Sentry

logging.basicConfig(
 level=logging.ERROR, # Imposta il livello base su ERROR per questo gestore
 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
 handlers=[
 logging.FileHandler("agent_errors.log"), # Registra su un file
 logging.StreamHandler(sys.stdout) # Stampa anche sulla console
 ]
)

# Configura un logger separato per eventi specifici dell'agente
agent_logger = logging.getLogger('agent.core')
agent_logger.setLevel(logging.INFO)
agent_logger.addHandler(logging.StreamHandler(sys.stdout))

# # Esempio di configurazione Sentry (richiede `pip install sentry-sdk`)
# sentry_sdk.init(
# dsn="YOUR_SENTRY_DSN",
# traces_sample_rate=1.0
# )

def handle_critical_error(exception, context="Contesto sconosciuto"):
 logging.critical(f"ERRORE CRITICO in {context}: {exception}", exc_info=True)
 # sentry_sdk.capture_exception(exception) # Invia a Sentry
 # Facoltativamente, invia un'email o un alert SMS qui
 # sys.exit(1) # Per errori irreversibili, l'agente potrebbe dover terminare

def perform_risky_operation(data):
 try:
 # Simula un'operazione che potrebbe fallire
 if not isinstance(data, dict) or 'value' not in data:
 raise ValueError("Formato dei dati non valido.")
 result = 100 / data['value']
 agent_logger.info(f"Operazione rischiosa riuscita con risultato: {result}")
 return result
 except ZeroDivisionError as e:
 logging.error("Tentativo di divisione per zero nell'operazione rischiosa.")
 # Potrebbe provare una riserva o informare l'utente
 return None
 except ValueError as e:
 handle_critical_error(e, context="perform_risky_operation - validazione dei dati")
 return None
 except Exception as e:
 handle_critical_error(e, context="perform_risky_operation - errore generale")
 return None

# Esempi:
perform_risky_operation({'value': 5})
perform_risky_operation({'value': 0})
perform_risky_operation('non un dict')
perform_risky_operation({'key': 'no_value_key'})