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

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

No mundo dos agentes de IA, uma gestão de erros eficaz 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 propensos a encontrar situações inesperadas. De falhas de rede e respostas de API inválidas a entradas de usuário mal formadas e incoerências lógicas, um agente bem projetado deve ser capaz de se recuperar com elegância, informar ou se adaptar. Sem uma gestão eficaz de erros, um agente pode rapidamente se tornar frágil, falhar silenciosamente ou falhar completamente, resultando em más experiências 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 diferentes estratégias, demonstrar armadilhas comuns e fornecer exemplos concretos usando 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 amigáveis.

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

• Confiabilidade: Prevenir falhas e garantir operação contínua.
• Experiência do usuário: Fornecer feedback significativo em vez de erros enigmáticos.
• Depuração: Centralizar o registro de erros, facilitando a identificação e a resolução de problemas.
• Gestão de recursos: Permitir uma limpeza apropriada (por exemplo, fechar conexões, liberar locks).
• Adaptabilidade: Permitir que os agentes tentem novamente operações ou mudem de estratégia diante de falhas temporárias.

Compreendendo os cenários de erros comuns em agentes

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

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

Esses podem ser os mais frequentes. Um agente geralmente depende de serviços externos para dados, cálculos ou ações. Os exemplos incluem:

• Problemas de rede: Tempo de conexão, falhas de resolução DNS, host inacessível.
• Erros de API: HTTP 4xx (erros de cliente como 404 Not Found, 401 Unauthorized, 400 Bad Request), HTTP 5xx (erros de 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 requisições, violações de restrições.

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

Os agentes processam diferentes formas de entrada, desde prompts de usuário até dados de sensores. Uma entrada inválida pode resultar em um comportamento inesperado:

• Entrada de usuário malformada: 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 de lógica interna

Esses erros vêm do código ou do estado do agente:

• Falhas de asserção: Condições que deveriam ser verdadeiras não o são.
• Índices fora dos limites: Tentando acessar um elemento além do comprimento de uma lista.
• Erros de tipo: Operar em dados com um tipo incorreto (por exemplo, tentar somar 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 há um código explícito:

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

Os fundamentos da gestão de erros em Python

O principal mecanismo de gestão de erros do Python é 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: Impossível dividir por zero!")
 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, tenha ocorrido ou não uma exceçã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 gerar uma exceção.
• except ExceptionType as e: Captura tipos específicos de exceções. Você pode ter múltiplos blocos except para diferentes tipos de erros. A parte as e permite que você acesse o objeto de exceção para mais detalhes.
• except Exception as e: Um capturador geral para todas as outras exceções. É recomendável capturar primeiro exceções específicas, e depois uma geral.
• finally: O código neste bloco é sempre executado, tenha ocorrido ou não uma exceção. É ideal para operações de limpeza (por exemplo, fechar arquivos, liberar recursos).
• else (opcional): O código aqui é executado apenas se o bloco try terminar sem nenhuma exceção.

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

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

Procure sempre capturar exceções específicas em vez das gerais, quando possível. Isso permite uma recuperação adequada 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 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 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:
 # Capturando 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() 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. Novas tentativas com espera exponencial

Para erros transitórios (como problemas de rede, disponibilidade temporária do serviço ou limites de taxa), tentar novamente a operação após um atraso é uma estratégia eficaz. A espera exponencial aumenta o atraso entre as novas 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 do servidor
 logging.warning(f"Tentativa {attempt + 1} : Erro do 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 (ex: 404, 400), não tentar novamente por padrão
 logging.error(f"Tentativa {attempt + 1} : Erro HTTP irrecuperável {status_code} para {url}. Abandonando tentativas.")
 return None

 if attempt < max_retries - 1:
 delay = initial_delay * (2 ** attempt) # Recarga 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 irrecuperável para {url} : {e}. Abandonando.")
 return None
 except ValueError as e:
 logging.error(f"Falha ao decifrar JSON a partir de {url} : {e}. Abandonando.")
 return None
 return None

# Teste com uma API instável ou um endpoint com limite de 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

Previna erros validando as entradas o mais rápido possível. Isso é especialmente importante para agentes voltados 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: Verifique 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 formado: {command_str}")
 return "Formato de comando 'set temperature' inválido. Esperado 'set temperature [value].'"
 elif command_str == "status":
 logging.info("Verificando o status do dispositivo.")
 return "O dispositivo está operacional."
 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 granular 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):
 """Levantada quando um sensor falha ao fornecer dados válidos."""
 def __init__(self, sensor_id, message="Falha na leitura do sensor."):
 self.sensor_id = sensor_id
 self.message = f"{message} ID do sensor: {sensor_id}"
 super().__init__(self.message)

class ActionFailedError(AgentError):
 """Levantada quando a ação de um 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 uma leitura de 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 alcançar {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 vai 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 executar 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 integração Sentry em condições reais

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

# Configure 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 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 Sentry
 # Opcionalmente, enviar um alerta por email ou SMS aqui
 # sys.exit(1) # Para erros irrecuperáveis, o agente pode precisar ser encerrado

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 o resultado: {result}")
 return result
 except ZeroDivisionError as e:
 logging.error("Tentativa de divisão por zero na operação arriscada.")
 # Tentar uma solução 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'})