Gestire gli errori dell’Agente: Un tutorial pratico

Introduzione alla gestione degli errori degli agenti

Nel mondo degli agenti IA, una gestione efficace degli errori non è solo una buona pratica; è una necessità. Mentre gli agenti interagiscono con ambienti dinamici, API esterne e dati complessi, possono trovarsi di fronte a situazioni inaspettate. Dai problemi di rete e dalle risposte API invalide agli input degli utenti mal formattati e alle incoerenze logiche, un agente ben progettato deve essere in grado di riprendersi con grazia, informare o adattarsi. Senza una gestione efficace degli errori, un agente può rapidamente diventare fragile, fallendo silenziosamente o bloccandosi completamente, portando a cattive esperienze per l’utente e operazioni inaffidabili.

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

Perché la gestione degli errori è cruciale per gli agenti?

• Affidabilità: Previene i blocchi e garantisce un funzionamento continuo.
• Esperienza utente: Fornisce feedback significativi invece di errori criptici.
• Debugging: Centralizza la registrazione degli errori, facilitando l’identificazione e la correzione dei problemi.
• Gestione delle risorse: Consente una pulizia adeguata (ad esempio, chiusura delle connessioni, rilascio dei lock).
• Adattabilità: Permette agli agenti di riprovare operazioni o cambiare strategia di fronte a guasti temporanei.

Comprendere gli scenari di errori comuni negli agenti

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

1. Errori di servizi esterni (API, database, rete)

Questi sono probabilmente i più comuni. Un agente si affida spesso a servizi esterni per ottenere dati, effettuare calcoli o azioni. Gli esempi includono:

• Problemi di rete: Timeout di connessione, fallimenti nella risoluzione DNS, host non raggiungibile.
• Errori API: HTTP 4xx (errori del cliente come 404 Not Found, 401 Unauthorized, 400 Bad Request), HTTP 5xx (errori del server come 500 Internal Server Error, 503 Service Unavailable), limitazioni di velocità (429 Too Many Requests).
• Errori di database: Fallimenti di connessione, timeout delle query, violazioni delle restrizioni.

2. Errori di convalida degli input/output

Gli agenti trattano diverse forme di input, dagli inviti degli utenti ai dati dei sensori. Input non validi possono portare a comportamenti imprevisti:

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

3. Errori logici interni

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

• Fallimenti di assertion: Le condizioni che dovrebbero essere vere non lo sono.
• Indice fuori limite: Tentare di accedere a un elemento oltre la lunghezza di una lista.
• Errori di tipo: Operare su dati di un tipo errato (ad esempio, tentare di sommare una stringa a un intero).
• Esaurimento delle risorse: Mancanza di memoria o descrittori di file.

4. Cambiamenti ambientali inaspettati

Gli agenti in ambienti dinamici possono incontrare situazioni per cui non sono stati esplicitamente programmati:

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

I 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 la registrazione 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: Divisione per zero impossibile!")
 return None
 except TypeError:
 logging.error("Errore: Entrambi gli input devono essere numeri.")
 return None
 except Exception as e:
 # Catturare tutti gli altri errori inaspettati
 logging.error(f"Si è verificato un errore inaspettato: {e}")
 return None
 finally:
 # Questo blocco viene eseguito sempre, che l'eccezione si sia verificata o meno
 logging.info("Tentativa di divisione conclusa.")

# 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)) # 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 errori. La parte as e ti consente di accedere all’oggetto di eccezione per ulteriori dettagli.
• except Exception as e: Un catch-all generale per tutte le altre eccezioni. È buona pratica catturare prima le eccezioni specifiche, poi una generale.
• finally: Il codice di questo blocco viene sempre eseguito, che l’eccezione si sia verificata o meno. È ideale per le operazioni di pulizia (ad esempio, chiusura di file, rilascio di risorse).
• else (opzionale): Il codice qui viene eseguito solo se il blocco try termina senza alcuna eccezione.

Strategie pratiche di gestione degli errori per gli agenti

1. Gestione e registrazione di eccezioni specifiche

Cerca sempre di catturare eccezioni specifiche piuttosto che eccezioni generali quando possibile. Questo consente un recupero mirato e una registrazione più chiara.

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"La 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:
 # Catturare tutte le altre errori legate alla richiesta
 logging.error(f"Si è verificato un errore di richiesta inaspettato per {url}: {e}")
 return None
 except ValueError as e:
 # Errore di decodifica JSON se response.json() fallisce
 logging.error(f"Fallimento nella decodifica 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. Riprovare con un backoff esponenziale

Per errori transitori (come problemi di rete, indisponibilità temporanea del servizio o limiti di velocità), riprovare l’operazione dopo un ritardo è una strategia efficace. Il backoff esponenziale aumenta il ritardo tra i ripetuti tentativi, evitando di sovraccaricare il servizio e permettendogli di recuperare.

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 velocità
 logging.warning(f"Tentativo {attempt + 1} : Limite di velocità 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 (ad esempio, 404, 400), non riprovare per impostazione predefinita
 logging.error(f"Tentativo {attempt + 1} : Errore HTTP irreparabile {status_code} per {url}. Abbandono dei tentativi.")
 return None

 if attempt < max_retries - 1:
 delay = initial_delay * (2 ** attempt) # Attesa 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 sono falliti per {url}.")
 return None
 except requests.exceptions.RequestException as e:
 logging.error(f"Si è verificato un errore di richiesta irreparabile per {url} : {e}. Abbandono.")
 return None
 except ValueError as e:
 logging.error(f"Errore nel decodificare JSON da {url} : {e}. Abbandono.")
 return None
 return None

# Test con un'API instabile o un endpoint limitato dalla velocità
# 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 sanificazione delle input

Preventire gli errori validando l'input il prima possibile. Questo è particolarmente importante per gli agenti destinati 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 "Si prega di fornire un comando."

 # Esempio : Controllare un modello 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"Impostando la 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' mal formato : {command_str}")
 return "Formato di comando 'set temperature' non valido. Atteso 'set temperature [valore].'"
 elif command_str == "status":
 logging.info("Controllo dello stato dell'apparecchio.")
 return "L'apparecchio funziona normalmente."
 else:
 logging.warning(f"Comando sconosciuto ricevuto : '{command_str}'")
 return "Non capisco questo 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 solleverà un ValueError

4. Eccezioni personalizzate per la logica specifica dell'agente

Per gli errori specifici del dominio del tuo agente, definisci eccezioni personalizzate. Questo migliora la leggibilità del codice e consente una gestione degli errori più dettagliata nei 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 correlati all'agente."""
 pass

class SensorReadError(AgentError):
 """Sollevata quando un sensore non riesce a fornire dati validi."""
 def __init__(self, sensor_id, message="Impossibile leggere il sensore."):
 self.sensor_id = sensor_id
 self.message = f"{message} ID del 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"L'azione '{action_name}' è fallita : {reason}"
 super().__init__(self.message)

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

def activate_heater(target_temp):
 if target_temp > 30:
 raise ActionFailedError("activate_heater", "Temperatura obiettivo troppo alta.")
 logging.info(f"Riscaldamento 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ò continuare a causa di un errore di 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 ha fallito nel completare 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 # Decommentare e configurare per una vera integrazione con Sentry

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

# Configurare un logger separato per gli 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 di 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) # Inviare a Sentry
 # Opzionalmente, inviare un avviso via email o SMS qui
 # sys.exit(1) # Per errori irreversibili, l'agente potrebbe aver bisogno di terminarsi

def perform_risky_operation(data):
 try:
 # Simulare un'operazione che potrebbe fallire
 if not isinstance(data, dict) or 'value' not in data:
 raise ValueError("Formato di 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.")
 # Potenzialmente provare un'alternativa 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('not a dict')
perform_risky_operation({'key': 'no_value_key'})