Gestire gli errori dell’agente: Un tutorial pratico

Introduzione alla gestione degli errori degli agenti

Nel mondo degli agenti IA, una gestione degli errori efficace non è solo una buona prassi; è una necessità. Mentre gli agenti interagiscono con ambienti dinamici, API esterne e dati complessi, sono suscettibili di incontrare situazioni inaspettate. Da guasti di rete e risposte API invalide a input utente malformati e 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, fallire silenziosamente o bloccarsi completamente, con conseguenti esperienze utente negative e operazioni inaffidabili.

Questo tutorial esplorerà gli aspetti pratici della gestione degli errori degli agenti. Esamineremo diverse strategie, dimostreremo trappole comuni e forniremo esempi concreti utilizzando Python, un linguaggio popolare per costruire agenti IA. Il nostro obiettivo è fornirvi 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à: Prevenire guasti e garantire un funzionamento continuo.
• Esperienza utente: Fornire feedback significativi invece di errori criptici.
• Debugging: Centralizzare la registrazione degli errori, facilitando l’identificazione e la risoluzione dei problemi.
• Gestione delle risorse: Consentire una pulizia appropriata (ad esempio, chiudere connessioni, rilasciare lock).
• Adattabilità: Consentire agli agenti di riprovare operazioni o cambiare strategia di fronte a guasti temporanei.

Comprendere gli scenari di errori comuni degli agenti

Prima di esplorare l’implementazione, classifichiamo 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 spesso si affida a 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 velocità (429 Too Many Requests).
• Errori di database: Fallimenti di connessione, timeout delle query, violazioni di vincoli.

2. Errori di validazione degli input/output

Gli agenti elaborano diverse forme di input, dagli inviti utente ai dati dei sensori. Un input invalido può portare a comportamenti imprevisti:

• Input utente malformato: Input non numerici dove è atteso un numero, formati data invalidi.
• Parametri mancanti: Argomenti necessari non forniti.
• Valori fuori limite: Una lettura di temperatura fisicamente impossibile.

3. Errori di logica interna

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

• Fallimenti di asserzione: Condizioni che dovrebbero essere vere non lo sono.
• Index out of bounds: Tentare di accedere a un elemento oltre la lunghezza di una lista.
• Errori di tipo: Operare su dati con un tipo errato (ad esempio, tentare di aggiungere 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 le quali non esiste un codice esplicito:

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

I fondamenti della gestione degli errori in Python

Il principale meccanismo di gestione degli errori in Python è 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: impossibile dividere per zero!")
 return None
 except TypeError:
 logging.error("Errore: entrambe le entrate devono essere numeri.")
 return None
 except Exception as e:
 # Catturare tutti gli altri errori imprevisti
 logging.error(f"Si è verificato un errore imprevisto: {e}")
 return None
 finally:
 # Questo blocco viene sempre eseguito, che l'eccezione si sia verificata o meno
 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)) # Altro TypeError

Analizziamo i componenti:

• try: Il codice che potrebbe sollevare un’eccezione.
• except ExceptionType as e: Cattura tipi specifici di eccezioni. È possibile avere più blocchi except per diversi tipi di errori. La parte as e consente di accedere all’oggetto dell’eccezione per ulteriori dettagli.
• except Exception as e: Un cattura-tutto generale per tutte le altre eccezioni. È consigliabile catturare prima le eccezioni specifiche, poi una generale.
• finally: Il codice in questo blocco viene sempre eseguito, che si sia verificata o meno un’eccezione. È ideale per operazioni di pulizia (ad esempio, chiudere file, rilasciare risorse).
• else (facoltativo): 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 delle eccezioni specifiche

Mirate sempre a catturare eccezioni specifiche piuttosto che generali quando possibile. Questo consente un recupero adeguato 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 imprevisto 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. Ritenti con attesa esponenziale

Per errori transitori (come problemi di rete, disponibilità temporanea del servizio o limiti di velocità), riprovare l’operazione dopo un ritardo è una strategia efficace. L’attesa esponenziale aumenta il ritardo tra i tentativi, evitando di sovraccaricare il servizio e consentendogli 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 velocità
 logging.warning(f"Tentativo {attempt + 1} : Limite di velocità raggiunto per {url}. Nuovo tentativo...")
 elif status_code and 500 <= status_code < 600: # Errore del server
 logging.warning(f"Tentativo {attempt + 1} : Errore del server ({status_code}) per {url}. Nuovo tentativo...")
 elif isinstance(e, requests.exceptions.Timeout): # Timeout
 logging.warning(f"Tentativo {attempt + 1} : Timeout per {url}. Nuovo tentativo...")
 elif isinstance(e, requests.exceptions.ConnectionError): # Errore di connessione
 logging.warning(f"Tentativo {attempt + 1} : Errore di connessione per {url}. Nuovo tentativo...")
 else:
 # Per altri errori HTTP (es: 404, 400), non ripetere per default
 logging.error(f"Tentativo {attempt + 1} : Errore HTTP irreversibile {status_code} per {url}. Abbandono dei tentativi.")
 return None

 if attempt < max_retries - 1:
 delay = initial_delay * (2 ** attempt) # Recupero 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 irreversibile per {url} : {e}. Abbandono.")
 return None
 except ValueError as e:
 logging.error(f"Errore di decodifica JSON da {url} : {e}. Abbandono.")
 return None
 return None

# Test con un API instabile o un endpoint a limite di 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 entrate

Prevenite gli errori validando le entrate 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: 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 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 formattato : {command_str}")
 return "Formato del comando 'set temperature' non valido. Atteso 'set temperature [value].'"
 elif command_str == "status":
 logging.info("Controllo dello stato del dispositivo.")
 return "Il dispositivo è operativo."
 else:
 logging.warning(f"Comando sconosciuto ricevuto : '{command_str}'")
 return "Non comprendo 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ù granulare ai 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="Errore di lettura dal sensore."):
 self.sensor_id = sensor_id
 self.message = f"{message} ID del sensore : {sensor_id}"
 super().__init__(self.message)

class ActionFailedError(AgentError):
 """Sollevata quando l'azione di un 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 una lettura del sensore, a volte fallisce
 if sensor_id == "temp_001":
 # Simulare una lettura avvenuta
 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 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ò continuare 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 provare 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 # Scommenta e configura per un'integrazione Sentry in condizioni reali

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 in un file
 logging.StreamHandler(sys.stdout) # Mostra anche nella 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 avviso via email o SMS qui
 # sys.exit(1) # Per errori irreversibili, l'agente potrebbe dover terminare

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 dei dati non valido.")
 result = 100 / data['value']
 agent_logger.info(f"Operazione rischiosa riuscita con il risultato : {result}")
 return result
 except ZeroDivisionError as e:
 logging.error("Tentativo di divisione per zero nell'operazione rischiosa.")
 # Prova eventualmente una soluzione di riserva o informa 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'})