Gestire gli errori dell’agent: Un tutorial pratico

Introduzione alla gestione degli errori degli agenti

Nel mondo degli agenti IA, una gestione degli errori efficace non è solo una buona pratica; è una necessità. Mentre gli agenti interagiscono con ambienti dinamici, API esterne e dati complessi, è probabile che affrontino situazioni inaspettate. Da guasti di rete e risposte API non valide a input dell’utente malformati e incoerenze logiche, un agente ben progettato deve essere in grado di riprendersi con eleganza, informare o adattarsi. Senza una gestione efficace degli errori, un agente può rapidamente diventare fragile, fallire silenziosamente o bloccarsi completamente, portando a cattive esperienze utente 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 è fornirti le conoscenze e gli strumenti necessari per creare agenti più resilienti, affidabili e user-friendly.

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

• Affidabilità: Prevenire i guasti e garantire un funzionamento continuo.
• Esperienza utente: Fornire feedback significativi anziché errori criptici.
• Debugging: Centralizzare la registrazione degli errori, facilitando l’identificazione e la risoluzione dei problemi.
• Gestione delle risorse: Permettere una pulizia appropriata (ad esempio, chiudere connessioni, liberare 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 forse i più comuni. Un agente si basa spesso su servizi esterni per dati, calcoli o azioni. Gli esempi includono:

• Problemi di rete: Timeout di connessione, fallimenti di risoluzione DNS, host non raggiungibile.
• Errori di 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, dalle richieste degli utenti ai dati dei sensori. Un input non valido può portare a comportamenti inaspettati:

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

3. Errori di logica interna

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

• Fallimenti di asserzione: 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 con un tipo errato (ad esempio, tentare di sommare una stringa a un intero).
• Esaurimento delle risorse: Mancanza di memoria o descriptores 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: Malfunzionamenti del sensore o errori disco.

I fondamenti della gestione degli errori in Python

Il principale meccanismo di gestione degli errori di 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: 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 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. Puoi avere più blocchi except per diversi tipi di errori. La parte as e ti consente di accedere all’oggetto eccezione per maggiori dettagli.
• except Exception as e: Un cattura-tutto generale per tutte le altre eccezioni. È consigliabile catturare prima eccezioni specifiche, poi una generale.
• finally: Il codice in questo blocco viene sempre eseguito, che si sia verificata un’eccezione o meno. È ideale per operazioni di pulizia (ad esempio, chiudere file, liberare risorse).
• else (opzionale): Il codice qui viene eseguito solo se il blocco try termina senza eccezioni.

Strategie pratiche di gestione degli errori per gli agenti

1. Gestione e registrazione delle eccezioni specifiche

Mirare sempre a catturare eccezioni specifiche piuttosto che generali quando possibile. Questo consente un recupero appropriato 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 relative 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 della 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. Riprova con attesa esponenziale

Per gli errori transitori (come i problemi di rete, la disponibilità temporanea del servizio o i 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 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 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 riprovare 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"Fallimento nella 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 degli input

Prevenite gli errori convalidando gli 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: Controlla 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"Impostazione della temperatura a {temp_value}°C.")
 return f"Temperatura impostata a {temp_value}°C."
 else:
 logging.error(f"Valore di temperatura non valido : {temp_value}. Deve essere tra 0 e 100.")
 return "La temperatura deve essere tra 0 e 100 gradi Celsius."
 except (ValueError, IndexError):
 logging.error(f"Comando 'set temperature' mal formato : {command_str}")
 return "Formato del comando 'set temperature' non valido. Atteso 'set temperature [valore].'"
 elif command_str == "status":
 logging.info("Controllo dello stato dell'apparecchio.")
 return "L'apparecchio è 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 legati all'agente."""
 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 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):
 # Simula una lettura del sensore, a volte questo 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, "Malfunzionamento 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"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 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 ha fallito nell'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 # Decommenta e configura per un'integrazione Sentry in condizioni reali

logging.basicConfig(
 level=logging.ERROR, # Impostare 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) # Mostrare anche nella console
 ]
)

# Configura 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 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
 # Opzionalmente, inviare un alert via email o SMS qui
 # sys.exit(1) # Per gli 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 il risultato : {result}")
 return result
 except ZeroDivisionError as e:
 logging.error("Tentativo di divisione per zero nell'operazione rischiosa.")
 # Provare eventualmente una soluzione 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'})