Gerenciando os erros do Agente: Um tutorial prático

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

No mundo dos agentes de IA, uma gestão eficaz de erros não é apenas uma boa prática; é uma necessidade. Enquanto os agentes interagem com ambientes dinâmicos, APIs externas e dados complexos, podem enfrentar situações inesperadas. Desde problemas de rede e respostas de API inválidas até entradas de usuários 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 eficaz de erros, um agente pode rapidamente se tornar frágil, falhando silenciosamente ou travando completamente, levando a experiências ruins para o usuário e operações não confiáveis.

Este tutorial explorará os aspectos práticos da gestão de erros dos agentes. Examinaremos várias estratégias, demonstraremos armadilhas comuns e forneceremos exemplos concretos utilizando Python, uma linguagem popular para construir agentes de IA. Nosso objetivo é fornecer a você os conhecimentos e as ferramentas necessárias para criar agentes mais resilientes, confiáveis e fáceis de usar.

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

• Confiabilidade: Previne travamentos e garante um funcionamento contínuo.
• Experiência do usuário: Fornece feedback significativo em vez de erros crípticos.
• Debugging: Centraliza o registro de erros, facilitando a identificação e correção de problemas.
• Gestão de recursos: Permite uma limpeza adequada (por exemplo, fechamento de conexões, liberação de locks).
• Adaptabilidade: Permite que os agentes tentem novamente operações ou mudem de estratégia diante de falhas temporárias.

Compreendendo cenários comuns de erros nos agentes

Antes de explorar a implementação, categorizamos os tipos de erros que um agente encontra frequentemente:

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

Esses são provavelmente os mais comuns. Um agente frequentemente depende de serviços externos para obter dados, realizar cálculos ou ações. Os exemplos incluem:

• Problemas de rede: Timeout de conexão, falhas na resolução DNS, host não atingí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ções de taxa (429 Too Many Requests).
• Erros de banco de dados: Falhas de conexão, timeouts das consultas, violações de restrições.

2. Erros de validação de entrada/saída

Os agentes lidam com diferentes formas de entrada, desde convites dos usuários até dados dos sensores. Entradas inválidas podem levar a comportamentos imprevistos:

• Entrada de usuário mal formatada: Entrada não numérica onde se espera um número, 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 derivam do código ou do estado do agente:

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

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á ausente.
• Problemas de autorização: O agente não tem acesso necessário a um recurso.
• Falhas de hardware: Falha de sensores ou erros de disco.

Os fundamentos da gestão de erros em Python

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

“`html

import logging

# Configurar a gravação 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: Ambos os inputs devem ser números.")
 return None
 except Exception as e:
 # Capturar todos os outros erros inesperados
 logging.error(f"Ocorreu um erro inesperado: {e}")
 return None
 finally:
 # Este bloco é sempre executado, independentemente de a exceção ter ocorrido 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

Analisemos os componentes:

• try: O código que pode gerar 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 catch-all geral para todas as outras exceções. É uma boa prática capturar primeiro as exceções específicas, depois uma geral.
• finally: O código deste bloco é sempre executado, independentemente de a exceção ter ocorrido ou não. É ideal para operações de limpeza (por exemplo, fechamento de arquivos, liberação de recursos).
• else (opcional): O código aqui é executado apenas se o bloco try terminar sem nenhuma exceção.

Estratégias práticas de gerenciamento de erros para agentes

1. Gerenciamento e registro de exceções específicas

Procure sempre capturar exceções específicas em vez de exceções gerais, quando possível. Isso permite uma recuperação dirigida e um registro 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() # Levanta HTTPError para respostas incorretas (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 API excedeu o tempo 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:
 # Capturar todos os outros erros relacionados à requisição
 logging.error(f"Ocorreu um erro de requisição inesperado para {url}: {e}")
 return None
 except ValueError as e:
 # Erro de decodificação JSON se response.json() falhar
 logging.error(f"Falha na decodificação 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. Tentar novamente com um backoff exponencial

Para erros transitórios (como problemas de rede, indisponibilidade temporária do serviço ou limites de taxa), tentar novamente a operação após um atraso é uma estratégia eficaz. O backoff exponencial aumenta o atraso entre as tentativas repetidas, 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}. Tentando novamente...")
 elif status_code and 500 <= status_code < 600: # Erro do servidor
 logging.warning(f"Tentativa {attempt + 1} : Erro do servidor ({status_code}) para {url}. Tentando novamente...")
 elif isinstance(e, requests.exceptions.Timeout): # Timeout
 logging.warning(f"Tentativa {attempt + 1} : Timeout para {url}. Tentando novamente...")
 elif isinstance(e, requests.exceptions.ConnectionError): # Erro de conexão
 logging.warning(f"Tentativa {attempt + 1} : Erro de conexão para {url}. Tentando novamente...")
 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}. Abandonando 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 solicitação irreparável para {url} : {e}. Abandonando.")
 return None
 except ValueError as e:
 logging.error(f"Erro ao decodificar JSON de {url} : {e}. Abandonando.")
 return None
 return None

# Testar com uma API instável ou um endpoint limitado pela 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 sanitização das entradas

Prevenir erros validando a entrada o mais rápido possível. Isso é particularmente importante para agentes destinados a 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 não vá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 uma temperatura específica
 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"Definindo a temperatura para {temp_value}°C.")
 return f"Temperatura definida para {temp_value}°C."
 else:
 logging.error(f"Valor da temperatura não vá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' não vá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 "Não entendo esse 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 levantará um 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.

```html

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):
 """Levantada quando um sensor não consegue fornecer dados válidos."""
 def __init__(self, sensor_id, message="Impossível ler o sensor."):
 self.sensor_id = sensor_id
 self.message = f"{message} ID do sensor : {sensor_id}"
 super().__init__(self.message)

class ActionFailedError(AgentError):
 """Levantada quando uma ação do agente não pode ser completada."""
 def __init__(self, action_name, reason="Motivo desconhecido."):
 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 falha
 if sensor_id == "temp_001":
 # Simular uma leitura bem-sucedida
 return 22.5
 elif sensor_id == "temp_002":
 # Simular um erro no 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"Aquecimento 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 pode passar para um sensor reserva ou avisar um operador humano
 except ActionFailedError as e:
 logging.error(f"O agente falhou em completar a ação '{e.action_name}' : {e.reason}")
 # O agente pode 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 # Descomentar e configurar para uma verdadeira integração com Sentry

logging.basicConfig(
 level=logging.ERROR, # Define o nível base como ERROR para este manipulador
 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 (requer `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 o Sentry
 # Opcionalmente, enviar um alerta por email ou SMS aqui
 # sys.exit(1) # Para erros irreversíveis, o agente pode precisar terminar

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'})