Gestir os erros do agent: Um tutorial prático

“`html

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

No mundo dos agentes de IA, uma gestão de erros eficaz não é apenas uma boa prática; é uma necessidade. Enquanto os agentes interagem com ambientes dinâmicos, APIs externas e dados complexos, eles estão suscetíveis a encontrar situações inesperadas. Desde falhas de rede e respostas API inválidas até entradas de usuário malformadas 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, falhar silenciosamente ou travar completamente, resultando em experiências negativas para o usuário e operações pouco confiáveis.

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

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

• Confiabilidade: Prevenir falhas e garantir um funcionamento contínuo.
• Experiência do usuário: Fornecer feedback significativo em vez de erros crípticos.
• Depuração: Centralizar o registro de erros, facilitando a identificação e 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 dos agentes

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

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

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

• Problemas de rede: Timeout de conexão, falhas na resolução DNS, host inatingí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, timeout das queries, violações de restrições.

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

Os agentes processam diversas formas de entrada, desde convites de usuário até dados de sensores. Uma entrada inválida pode levar a comportamentos imprevistos:

• Entrada de usuário malformada: Entradas não numéricas onde se espera um número, formatos de data inválidos.
• Parâmetros ausentes: Argumentos necessá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 são.
• Index out of bounds: Tentar acessar um elemento além do comprimento de uma lista.
• Erros de tipo: Operar em dados com um tipo incorreto (por exemplo, tentar adicionar 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 existe um código explícito:

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

Os fundamentos da gestão de erros em Python

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

“““html

import logging

# Configurar o registro básico
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:
 # Capturando todos os outros erros inesperados
 logging.error(f"Ocorreu um erro inesperado: {e}")
 return None
 finally:
 # Este bloco é sempre executado, tenha ocorrido ou não a 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 analisar os componentes:

• try: O código que pode gerar uma exceção.
• except ExceptionType as e: Captura tipos específicos de exceções. É possível ter vários blocos except para diferentes tipos de erros. A parte as e permite acessar o objeto da exceção para mais detalhes.
• except Exception as e: Um captura-tudo geral para todas as outras exceções. É recomendável capturar primeiro as exceções específicas, 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 somente se o bloco try termina sem nenhuma exceção.

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

1. Gerenciar e registrar exceções específicas

Sempre busque capturar exceções específicas em vez de 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 erradas (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() 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. Retentativa com espera exponencial

Para erros transitórios (como problemas de rede, disponibilidade temporária do serviço ou limites de taxa), tentar a operação novamente após um atraso é uma estratégia eficaz. A espera exponencial aumenta o atraso 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}. Novo intento...")
 elif status_code and 500 <= status_code < 600: # Erro do servidor
 logging.warning(f"Tentativa {attempt + 1} : Erro do servidor ({status_code}) para {url}. Novo intento...")
 elif isinstance(e, requests.exceptions.Timeout): # Timeout
 logging.warning(f"Tentativa {attempt + 1} : Timeout para {url}. Novo intento...")
 elif isinstance(e, requests.exceptions.ConnectionError): # Erro de conexão
 logging.warning(f"Tentativa {attempt + 1} : Erro de conexão para {url}. Novo intento...")
 else:
 # Para outros erros HTTP (ex: 404, 400), não tentar novamente por padrão
 logging.error(f"Tentativa {attempt + 1} : Erro HTTP irreversível {status_code} para {url}. Abandonando tentativas.")
 return None

 if attempt < max_retries - 1:
 delay = initial_delay * (2 ** attempt) # Recupero exponencial
 logging.info(f"Aguardando {delay:.1f} segundos antes do próximo intento...")
 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 irreversível para {url} : {e}. Abandonando.")
 return None
 except ValueError as e:
 logging.error(f"Erro de decodificação JSON de {url} : {e}. Abandonando.")
 return None
 return None

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

Previna erros validando as entradas o mais rápido possível. Isso é particularmente importante para agentes direcionados 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: Verifica 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"Definindo a temperatura para {temp_value}°C.")
 return f"Temperatura definida para {temp_value}°C."
 else:
 logging.error(f"Valor da 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 do comando 'set temperature' inválido. Esperado 'set temperature [valor].'"
 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 compreendo 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 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 uma gestão de erros mais granular 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="Erro de 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 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 do 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 do sensor : {e.sensor_id} - {e.message}")
 # O agente poderia passar para um sensor reserva ou notificar um operador humano
 except ActionFailedError as e:
 logging.error(f"O agente não conseguiu executar a ação '{e.action_name}' : {e.reason}")
 # O agente poderia tentar uma ação alternativa ou registrar para uma 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 gerenciado : {e}")

agent_main_loop()
```

import logging
import sys
# import sentry_sdk # Descomente e configure para uma integração Sentry em condições reais

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"), # Registra em um arquivo
 logging.StreamHandler(sys.stdout) # Mostra também no console
 ]
)

# Configura 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) # Envia para Sentry
 # Opcionalmente, envie um aviso 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 dos 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.")
 # Tente eventualmente uma solução de reserva ou informe 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'})