Domine a gestão de erros do Agent: Um tutorial prático

Introdução à gestão de erros de agentes

No mundo dos agentes IA, uma gestão eficaz de erros não é apenas uma boa prática; é uma necessidade. À medida que os agentes interagem com ambientes dinâmicos, APIs externas e dados complexos, eles estão suscetíveis a situações inesperadas. De falhas de rede e respostas de API inválidas a entradas de usuário mal formatadas e incoerências lógicas, um agente bem projetado deve ser capaz de se recuperar com graça, informar ou se adaptar. Sem uma gestão eficiente de erros, um agente pode rapidamente se tornar frágil, falhando silenciosamente ou travando completamente, o que resulta em experiências ruins para o usuário e operações pouco confiáveis.

Este tutorial explorará os aspectos práticos da gestão de erros de agentes. Vamos examinar várias estratégias, demonstrar armadilhas comuns e fornecer exemplos concretos usando Python, uma linguagem popular para construir agentes IA. Nosso objetivo é equipá-lo com o conhecimento e as ferramentas necessárias para criar agentes mais resilientes, confiáveis e amigáveis ao usuário.

Por que a gestão de erros é crucial para os agentes?

• Confiabilidade: Previne falhas e garante um funcionamento contínuo.
• Experiência do usuário: Fornece feedback significativo ao invés de erros criptográficos.
• Depuração: Centraliza a registro de erros, facilitando a identificação e correção de problemas.
• Gestão de recursos: Permite uma limpeza apropriada (por exemplo, fechamento de conexões, liberação de bloqueios).
• Adaptabilidade: Permite que os agentes reattemptem operações ou mudem de estratégia diante de falhas temporárias.

Compreender os cenários de erros comuns em agentes

Antes de explorar a implementação, vamos categorizar os tipos de erros que um agente encontra com frequência:

1. Erros de serviços externos (API, banco de dados, rede)

Esses são talvez os mais frequentes. Um agente muitas vezes depende de serviços externos para obter dados, realizar cálculos ou ações. Exemplos incluem:

• Problemas de rede: Timeout de conexão, falhas de resolução DNS, host inatingível.
• Erros de API: HTTP 4xx (erros do cliente como 404 Not Found, 401 Unauthorized, 400 Bad Request), HTTP 5xx (erros do servidor como 500 Internal Server Error, 503 Service Unavailable), limitação de taxa (429 Too Many Requests).
• Erros de banco de dados: Falhas de conexão, timeouts de consulta, violações de restrições.

2. Erros de validação de entradas/saídas

Os agentes lidam com várias formas de entrada, desde prompts do usuário até dados de sensores. Entradas inválidas podem levar a comportamentos inesperados:

• Entrada de usuário mal formatada: Entrada não numérica onde um número é esperado, formatos de data inválidos.
• Parâmetros ausentes: Argumentos obrigatórios não fornecidos.
• Valores fora dos limites: Uma leitura de temperatura fisicamente impossível.

3. Erros lógicos internos

Esses erros decorrem do código ou do estado do agente:

• Falhas de asserção: As condições que deveriam ser verdadeiras não são.
• Índice fora dos limites: Tentar acessar um elemento além do comprimento de uma lista.
• Erros de tipo: Operar em dados de um tipo incorreto (por exemplo, tentar adicionar uma string a um inteiro).
• Exaustão de recursos: Falta de memória ou descritores de arquivos.

4. Mudanças ambientais inesperadas

Agentes em ambientes dinâmicos podem encontrar situações para as quais não foram explicitamente programados:

• Arquivo não encontrado: Um arquivo de configuração necessário está faltando.
• Problemas de permissões: O agente não tem o acesso necessário a um recurso.
• Falhas de hardware: Falha de sensor ou erros de disco.

Os fundamentos da gestão de erros em Python

O mecanismo principal do Python para gestão de erros é o bloco try-except-finally.

import logging

# Configurar a logagem básica
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def divide_numbers(a, b):
 try:
 result = a / b
 logging.info(f"Divisão bem-sucedida: {a} / {b} = {result}")
 return result
 except ZeroDivisionError:
 logging.error("Erro: Divisão por zero impossível!")
 return None
 except TypeError:
 logging.error("Erro: Ambas as entradas devem ser números.")
 return None
 except Exception as e:
 # Captura todas as outras erros inesperadas
 logging.error(f"Ocorreu um erro inesperado: {e}")
 return None
 finally:
 # Este bloco é sempre executado, independentemente de a exceção ocorrer ou não
 logging.info("Tentativa de divisão concluída.")

# Exemplos:
print(divide_numbers(10, 2)) # Divisão bem-sucedida
print(divide_numbers(10, 0)) # ZeroDivisionError
print(divide_numbers(10, "a")) # TypeError
print(divide_numbers(None, 5)) # Outro TypeError

Vamos decompor os componentes:

• try: O código que pode levantar uma exceção.
• except ExceptionType as e: Captura tipos específicos de exceções. Você pode ter vários blocos except para diferentes tipos de erros. A parte as e permite acessar o objeto de exceção para mais detalhes.
• except Exception as e: Um capturador geral para todas as outras exceções. É uma boa prática capturar primeiro exceções específicas e depois uma geral.
• finally: O código deste bloco é sempre executado, independentemente de a exceção ocorrer ou não. É ideal para operações de limpeza (por exemplo, fechar arquivos, liberar recursos).
• else (opcional): O código aqui é executado somente se o bloco try terminar sem nenhuma exceção.

Estratégias práticas de gestão de erros para agentes

1. Gestão e log de exceções específicas

Tente sempre capturar exceções específicas em vez de exceções gerais, sempre que possível. Isso permite uma recuperação focada e um log mais claro.

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() # Lança HTTPError para respostas ruins (4xx ou 5xx)
 logging.info(f"Dados recuperados com sucesso de {url}")
 return response.json()
 except requests.exceptions.Timeout:
 logging.warning(f"A requisição da API expirou para {url}")
 return None
 except requests.exceptions.ConnectionError as e:
 logging.error(f"Erro de conexão de rede para {url}: {e}")
 return None
 except requests.exceptions.HTTPError as e:
 logging.error(f"Erro HTTP {e.response.status_code} para {url}: {e.response.text}")
 return None
 except requests.exceptions.RequestException as e:
 # Captura todas as outras erros relacionadas à requisição
 logging.error(f"Ocorreu um erro inesperado de requisição para {url}: {e}")
 return None
 except ValueError as e:
 # Erro de decodificação JSON se response.json() falha
 logging.error(f"Falha ao decodificar JSON de {url}: {e}")
 return None

# Exemplo de 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. Repetir com um backoff exponencial

Para erros temporários (como problemas de rede, indisponibilidade temporária do serviço ou limites de taxa), repetir a operação após um atraso é uma estratégia eficaz. O backoff exponencial aumenta o intervalo entre as tentativas, evitando sobrecarregar o serviço e permitindo que ele se recupere.

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"Tentativa {attempt + 1} : Dados recuperados com sucesso de {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 de taxa
 logging.warning(f"Tentativa {attempt + 1} : Limite de taxa atingido para {url}. Nova tentativa...")
 elif status_code and 500 <= status_code < 600: # Erro de servidor
 logging.warning(f"Tentativa {attempt + 1} : Erro de servidor ({status_code}) para {url}. Nova tentativa...")
 elif isinstance(e, requests.exceptions.Timeout): # Tempo de espera
 logging.warning(f"Tentativa {attempt + 1} : Tempo de espera para {url}. Nova tentativa...")
 elif isinstance(e, requests.exceptions.ConnectionError): # Erro de conexão
 logging.warning(f"Tentativa {attempt + 1} : Erro de conexão para {url}. Nova tentativa...")
 else:
 # Para outros erros HTTP (por exemplo, 404, 400), não tentar novamente por padrão
 logging.error(f"Tentativa {attempt + 1} : Erro HTTP irreparável {status_code} para {url}. Desistindo das tentativas.")
 return None

 if attempt < max_retries - 1:
 delay = initial_delay * (2 ** attempt) # Espera exponencial
 logging.info(f"Aguardando {delay:.1f} segundos antes da próxima tentativa...")
 time.sleep(delay)
 else:
 logging.error(f"Todas as {max_retries} tentativas falharam para {url}.")
 return None
 except requests.exceptions.RequestException as e:
 logging.error(f"Ocorreu um erro de requisição irreparável para {url} : {e}. Desistindo.")
 return None
 except ValueError as e:
 logging.error(f"Falha ao decodificar JSON de {url} : {e}. Desistindo.")
 return None
 return None

# Testando com uma API instável ou um endpoint limitado por taxa
# print(fetch_data_with_retries("https://httpbin.org/status/503")) # Deve tentar novamente
# print(fetch_data_with_retries("https://httpbin.org/delay/1", max_retries=1)) # Deve ter sucesso imediatamente

3. Validação e saneamento de entradas

Prevenir erros validando a entrada o quanto antes. Isso é especialmente importante para agentes destinados aos usuários.

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 de comando inválido : deve ser uma string.")
 raise ValueError("O comando deve ser uma string.")
 
 command_str = command_str.strip().lower()

 if not command_str:
 logging.warning("Comando vazio recebido.")
 return "Por favor, forneça um comando."

 # Exemplo : Verificar um padrão específico
 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"Ajustando a temperatura para {temp_value}°C.")
 return f"Temperatura ajustada para {temp_value}°C."
 else:
 logging.error(f"Valor de temperatura inválido : {temp_value}. Deve estar entre 0 e 100.")
 return "A temperatura deve estar entre 0 e 100 graus Celsius."
 except (ValueError, IndexError):
 logging.error(f"Comando 'set temperature' mal formatado : {command_str}")
 return "Formato de comando 'set temperature' inválido. Esperado 'set temperature [valor].'"
 elif command_str == "status":
 logging.info("Verificando o status do dispositivo.")
 return "O dispositivo está funcionando normalmente."
 else:
 logging.warning(f"Comando desconhecido recebido : '{command_str}'")
 return "Eu não entendo este comando."

# Exemplos :
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) # Isso gerará uma ValueError

4. Exceções personalizadas para a lógica específica do agente

Para erros específicos do domínio do seu agente, defina exceções personalizadas. Isso melhora a legibilidade do código e permite um tratamento de erros mais detalhado nos níveis superiores da arquitetura do seu agente.

import logging

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

class AgentError(Exception):
 """Exceção base para todos os erros relacionados ao agente."""
 pass

class SensorReadError(AgentError):
 """Lançada quando um sensor não consegue fornecer dados válidos."""
 def __init__(self, sensor_id, message="Falha ao ler o sensor."):
 self.sensor_id = sensor_id
 self.message = f"{message} ID do sensor : {sensor_id}"
 super().__init__(self.message)

class ActionFailedError(AgentError):
 """Lançada quando uma ação do agente não pode ser completada."""
 def __init__(self, action_name, reason="Razão desconhecida."):
 self.action_name = action_name
 self.reason = reason
 self.message = f"A ação '{action_name}' falhou : {reason}"
 super().__init__(self.message)

def read_temperature_sensor(sensor_id):
 # Simular a leitura do sensor, às vezes isso falha
 if sensor_id == "temp_001":
 # Simular uma leitura bem-sucedida
 return 22.5
 elif sensor_id == "temp_002":
 # Simular um erro de sensor
 raise SensorReadError(sensor_id, "Falha de hardware detectada.")
 else:
 raise SensorReadError(sensor_id, "Sensor não encontrado.")

def activate_heater(target_temp):
 if target_temp > 30:
 raise ActionFailedError("activate_heater", "Temperatura alvo muito alta.")
 logging.info(f"Aquecedor ativado para atingir {target_temp}°C.")
 return True

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

 # Isso falhará
 read_temperature_sensor("temp_002")

 except SensorReadError as e:
 logging.error(f"O agente não pode continuar devido a um erro de sensor : {e.sensor_id} - {e.message}")
 # O agente poderia mudar para um sensor de backup ou alertar um operador humano
 except ActionFailedError as e:
 logging.error(f"O agente falhou ao realizar a ação '{e.action_name}' : {e.reason}")
 # O agente poderia tentar uma ação alternativa ou registrar para intervenção manual
 except AgentError as e:
 logging.error(f"Ocorreu um erro geral do agente : {e}")
 except Exception as e:
 logging.critical(f"Ocorreu um erro crítico não tratado : {e}")

agent_main_loop()
```

import logging
import sys
# import sentry_sdk # Descomente e configure para uma integração real do Sentry

logging.basicConfig(
 level=logging.ERROR, # Definir o nível base como ERROR para este handler
 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
 handlers=[
 logging.FileHandler("agent_errors.log"), # Registrar em um arquivo
 logging.StreamHandler(sys.stdout) # Imprimir também no console
 ]
)

# Configurar um logger separado para eventos específicos do agente
agent_logger = logging.getLogger('agent.core')
agent_logger.setLevel(logging.INFO)
agent_logger.addHandler(logging.StreamHandler(sys.stdout))

# # Exemplo de configuração do Sentry (necessita `pip install sentry-sdk`)
# sentry_sdk.init(
# dsn="YOUR_SENTRY_DSN",
# traces_sample_rate=1.0
# )

def handle_critical_error(exception, context="Contexto desconhecido"):
 logging.critical(f"ERRO CRÍTICO em {context} : {exception}", exc_info=True)
 # sentry_sdk.capture_exception(exception) # Enviar para Sentry
 # Opcionalmente, enviar um alerta por e-mail ou SMS aqui
 # sys.exit(1) # Para erros irreparáveis, o agente pode precisar ser finalizado

def perform_risky_operation(data):
 try:
 # Simular uma operação que pode falhar
 if not isinstance(data, dict) or 'value' not in data:
 raise ValueError("Formato de dados inválido.")
 result = 100 / data['value']
 agent_logger.info(f"Operação arriscada bem-sucedida com resultado : {result}")
 return result
 except ZeroDivisionError as e:
 logging.error("Tentativa de divisão por zero na operação arriscada.")
 # Potencialmente tentar uma alternativa ou informar o usuário
 return None
 except ValueError as e:
 handle_critical_error(e, context="perform_risky_operation - validação de dados")
 return None
 except Exception as e:
 handle_critical_error(e, context="perform_risky_operation - erro geral")
 return None

# Exemplos :
perform_risky_operation({'value': 5})
perform_risky_operation({'value': 0})
perform_risky_operation('not a dict')
perform_risky_operation({'key': 'no_value_key'})