Gestire gli errori dell’Agente: Un tutorial pratico

Introduzione alla gestione degli errori degli agenti

Nel mondo degli agenti AI, una gestione degli errori efficace non è solo una buona prassi; è una necessità. Mentre gli agenti interagiscono con ambienti dinamici, API esterne e dati complessi, possono imbattersi in situazioni inaspettate. Dai guasti di rete e risposte API non valide a input utente mal formattati 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, fallendo silenziosamente o bloccandosi completamente, il che porta a cattive esperienze per l’utente e operazioni inaffidabili.

Questo tutorial esplorerà gli aspetti pratici della gestione degli errori negli agenti. Esamineremo varie strategie, dimostreremo trappole comuni e forniremo esempi concreti utilizzando Python, un linguaggio popolare per costruire agenti AI. Il nostro obiettivo è fornirvi 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à: Previene i blocchi e garantisce un funzionamento continuo.
• Esperienza utente: Fornisce feedback significativi anziché errori criptici.
• Debugging: Centralizza la registrazione degli errori, facilitando l’identificazione e la correzione dei problemi.
• Gestione delle risorse: Permette una pulizia appropriata (ad esempio, chiusura delle connessioni, liberazione dei lock).
• Adattabilità: Consente agli agenti di ritentare operazioni o cambiare strategia di fronte a guasti temporanei.

Comprendere gli scenari di errore comuni negli agenti

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

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

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

• Problemi di rete: Timeout di connessione, fallimenti di risoluzione DNS, host irraggiungibile.
• Errori API: HTTP 4xx (errori del client come 404 Not Found, 401 Unauthorized, 400 Bad Request), HTTP 5xx (errori del server come 500 Internal Server Error, 503 Service Unavailable), limitazione della frequenza (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 trattano varie forme di input, dai prompt utente ai dati dei sensori. Input non validi possono portare a comportamenti inaspettati:

• Input utente mal formattato: Input non numerico dove ci si aspetta 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 asserzione: Le condizioni che dovrebbero essere vere non lo sono.
• Indice fuori limite: Tentativo 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 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 imbattersi in situazioni per le quali non sono stati esplicitamente programmati:

• File non trovato: Un file di configurazione richiesto è mancante.
• Problemi di permessi: L’agente non ha accesso a una risorsa necessaria.
• Guasti hardware: Guasto di un sensore 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 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 tutte le altre errori inattesi
 logging.error(f"Si è verificato un errore inaspettato: {e}")
 return None
 finally:
 # Questo blocco viene sempre eseguito, che si sia verificata o meno l'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)) # Altro TypeError

Decomponiamo 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 dell’eccezione per ulteriori dettagli.
• except Exception as e: Un cattura-tutto generale per tutte le altre eccezioni. È buona prassi catturare prima eccezioni specifiche e poi una generale.
• finally: Il codice di questo blocco viene sempre eseguito, che si sia verificata o meno l’eccezione. È ideale per operazioni di pulizia (ad esempio, chiusura di file, liberazione 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 delle 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 non valide (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 inatteso 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 d'uso:
# 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. Ritenta con un backoff esponenziale

Per gli errori transitori (come problemi di rete, indisponibilità temporanea del servizio o limiti di frequenza), ritentare l’operazione dopo un intervallo è una strategia efficace. Il backoff esponenziale aumenta l’intervallo tra i ritentativi, 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 frequenza
 logging.warning(f"Tentativo {attempt + 1} : Limite di frequenza 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 (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 soggetto a limitazione 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 delle entrate

Prevenire errori convalidando 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"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' malformato : {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 genererà una ValueError

4. Eccezioni personalizzate per la logica specifica dell'agente

Per errori specifici al 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 di base per tutti gli errori legati 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 di sensore
 raise SensorReadError(sensor_id, "Errore 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 allertare 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 registrarsi 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 di 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"), # Registro su un file
 logging.StreamHandler(sys.stdout) # Stampa anche sulla console
 ]
)

# Configurare 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 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) # Invia a Sentry
 # Facoltativamente, invia un avviso via e-mail o SMS qui
 # sys.exit(1) # Per errori irreversibili, l'agente potrebbe aver bisogno di 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 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 prova un'alternativa o informa l'utente
 return None
 except ValueError as e:
 handle_critical_error(e, context="perform_risky_operation - convalida 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'})