Oi a todos, Leo aqui do agntdev.com! Hoje quero falar sobre algo que me preocupa muito neste período, especialmente enquanto trabalho em alguns novos projetos de agentes. Trata-se da parte “Dev” do desenvolvimento de agentes, especificamente, de como podemos construir agentes confiáveis a partir de componentes não confiáveis. Sim, você ouviu direito. Porque, para ser honesto, essa é a realidade para a maioria de nós, não é?
Normalmente, não trabalhamos com APIs e serviços perfeitamente elaborados e de nível empresarial. Mais frequentemente do que se pode pensar, montamos modelos de código aberto, APIs de terceiros com limites de velocidade questionáveis e talvez até alguns microserviços feitos em casa que, digamos, têm sua própria personalidade. E ainda assim, a expectativa é sempre que nossos agentes deveriam simplesmente… funcionar. De forma consistente. Confiável. Mesmo quando os componentes subjacentes estão apresentando problemas.
Eu embarquei nessa jornada tantas vezes. Lembro de um projeto do ano passado em que estava construindo um agente para me ajudar a gerenciar minhas contribuições de código aberto. Ele deveria interagir com a API do GitHub, um modelo de análise de sentimentos hospedado em um plano gratuito e um serviço de notificação personalizado que eu havia concoctado em um final de semana. Cada um desses tinha suas peculiaridades. O GitHub às vezes me bloqueava inesperadamente, o modelo de sentimentos podia expirar e meu serviço de notificação… bem, digamos apenas que ele tinha o hábito de esquecer as boas maneiras após uma hora de funcionamento. Se eu não tivesse implementado proteções sérias, tudo isso teria desmoronado como um castelo de cartas.
Hoje quero compartilhar algumas estratégias e modelos práticos que adotei para tornar meus agentes mais resilientes, mesmo quando os elementos dos quais são compostos são tudo, menos confiáveis.
A verdade inegável: as coisas vão quebrar
Primeiro de tudo, aceitemos isso como uma confissão. Nenhuma API está 100% disponível. Nenhum modelo é 100% preciso. Nenhuma rede é 100% estável. Uma vez que você abrace essa realidade, pode começar a projetar para o fracasso, o que, contraintuitivamente, torna seu agente mais eficaz.
O problema que vejo com frequência, especialmente com novos desenvolvedores explorando o trabalho com agentes, é que eles presumem o sucesso a cada chamada externa. Eles escrevem código como este:
response = external_api.call_method(data)
# Presumir que a resposta é sempre perfeita e prosseguir
processed_data = process_response(response)
E então, quando external_api.call_method levanta um erro de conexão, ou retorna um 500, ou simplesmente envia um JSON malformado, o agente inteiro para. Podemos fazer melhor.
Estratégia 1: tentativas sólidas com retorno exponencial
Esta é provavelmente a técnica mais fundamental, e ainda assim muitas vezes é mal implementada ou não implementada. Tentar imediatamente após um erro geralmente é uma má ideia. Se o serviço externo está fora do ar, você está apenas piorando as coisas, o que pode agravar a situação ou levar a uma limitação da velocidade.
A chave é o retorno exponencial. Isso significa aguardar períodos de tempo cada vez mais longos entre as tentativas. Isso dá ao serviço externo uma chance de se recuperar e reduz a carga que você impõe a ele.
Exemplo: Python com Tenacity
Para Python, minha biblioteca de escolha para isso é Tenacity. Ela torna incrivelmente simples adicionar lógica de tentativa.
“`html
import random
import logging
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ExternalServiceError(Exception):
"""Exceção personalizada para falhas no serviço externo."""
pass
# Simular uma chamada API externa não confiável
def call_unreliable_api(data):
if random.random() < 0.6: # 60% de chance de falha
logger.warning(f"A chamada API falhou para os dados: {data}")
raise ExternalServiceError("Falha ou timeout simulado da API")
logger.info(f"A chamada API foi bem-sucedida para os dados: {data}")
return {"status": "success", "result": f"processed_{data}"}
@retry(wait=wait_exponential(multiplier=1, min=4, max=10),
stop=stop_after_attempt(5),
retry=retry_if_exception_type(ExternalServiceError))
def get_processed_data_with_retries(input_data):
logger.info(f"Tentativa de chamada à API para: {input_data}")
return call_unreliable_api(input_data)
if __name__ == "__main__":
try:
result = get_processed_data_with_retries("some_important_item")
print(f"Resultado final: {result}")
except ExternalServiceError as e:
print(f"Falhou após várias tentativas: {e}")
except Exception as e:
print(f"Ocorreu um erro inesperado: {e}")
Neste fragmento:
wait_exponentialaumenta os atrasos a cada tentativa (4s, depois ~8s, depois ~16s, etc., até um máximo de 10s).stop_after_attempt(5)significa que tentará um máximo de 5 vezes.retry_if_exception_type(ExternalServiceError)garante que tentaremos novamente apenas para erros específicos, não, digamos, para umKeyboardInterrupt.
Este padrão é um verdadeiro salvador. Eu o utilizo para conexões com bancos de dados, requisições HTTP e até mesmo para comunicação interna entre os módulos dos agentes quando sei que um deles pode estar temporariamente sobrecarregado.
Estratégia 2: Interruptores automáticos para prevenir falhas em cascata
Tentativas são excelentes para erros transitórios. Mas o que acontece se o serviço estiver completamente fora do ar? Tentativas repetidas esgotarão seus recursos e podem agravar o problema para o serviço externo se ele estiver lutando para se recuperar. É aqui que entra o modelo de interruptor automático.
Pense nisso como um disjuntor elétrico em sua casa. Se houver um defeito (muitas falhas), ele “aciona”, impedindo que mais corrente flua e protegendo o sistema. Após um certo tempo, pode ser restaurado, mas não continuará tentando permitir que a corrente passe por um fio em curto-circuito.
Para os agentes, um interruptor automático monitora as chamadas para um serviço externo. Se a taxa de falhas ultrapassar um certo limite em uma janela de tempo determinada, o circuito “se abre”. Quando está aberto, todas as chamadas subsequentes para esse serviço falham imediatamente, sem nem mesmo tentar a chamada. Após um período de “atraso” configurável, o circuito muda para um estado “semi-aberto”, permitindo um número limitado de chamadas de teste para ver se o serviço se recuperou. Se essas forem bem-sucedidas, ele se fecha; se falharem, ele se reabre novamente.
Por que é importante para os agentes:
- Conservação de recursos: Seu agente não perde tempo e recursos tentando chamar um serviço com falha.
- Falha mais rápida: Em vez de esperar um atraso, seu agente recebe um sinal de falha imediato, permitindo que ele gerencie a situação (por exemplo, usar uma solução alternativa, registrar o problema, notificar um operador).
- Protege os serviços externos: Impede que seu agente sobrecarregue um serviço em dificuldades.
Eu normalmente implemento isso usando bibliotecas. Para Python, Pybreaker é excelente.
“““html
import time
import random
import logging
from pybreaker import CircuitBreaker, CircuitBreakerError
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ExternalAPIClient:
def __init__(self):
# Configura o disjuntor :
# 3 falhas consecutivas em 60 segundos abrirão o circuito.
# Permanece aberto por 5 segundos.
self.breaker = CircuitBreaker(fail_max=3, reset_timeout=5, exclude=[TypeError]) # Não interromper para TypeErrors
def _unreliable_call(self, data):
if random.random() < 0.7: # 70 % de chance de falha
logger.warning(f"Simulação de um erro interno da API para os dados: {data}")
raise ConnectionError("Serviço não acessível")
logger.info(f"A chamada da API foi bem-sucedida para os dados: {data}")
return {"result": f"processed_{data}"}
def process_data(self, data):
try:
return self.breaker.call(self._unreliable_call, data)
except CircuitBreakerError:
logger.error(f"O circuito está aberto! Nenhuma chamada da API para os dados: {data}")
# Lógica de fallback aqui: retornar dados armazenados em cache, um valor padrão ou lançar um erro mais específico
return {"result": "fallback_data", "source": "circuit_breaker"}
except Exception as e:
logger.error(f"Erro durante a chamada da API (não relacionado ao disjuntor): {e}")
raise
if __name__ == "__main__":
client = ExternalAPIClient()
for i in range(15):
print(f"\n--- Tentativa {i+1} ---")
try:
result = client.process_data(f"item_{i}")
print(f"Resultado: {result}")
except Exception as e:
print(f"Erro tratado: {e}")
time.sleep(1) # Simular um atraso entre as chamadas
Execute isso, e você verá o circuito abrir após algumas falhas, então, em última análise, tentar se reabrir, e talvez até se fechar novamente se o serviço simulado começar a se comportar bem.
Estratégia 3: Idempotência para operações que modificam o estado
É fundamental para qualquer agente que modifica o estado externo (por exemplo, criar um registro, enviar um e-mail, iniciar um pagamento). Se o seu agente tenta realizar uma ação, e a rede está instável, ou o serviço externo tem tempo limite, como você sabe se a ação realmente ocorreu?
Se você simplesmente tentar novamente sem considerar a idempotência, pode acidentalmente realizar a ação duas vezes. Imagine enviar o mesmo e-mail duas vezes, ou pior, cobrar um cliente duas vezes. Não é aceitável.
Uma operação é idempotente se realizá-la várias vezes tem o mesmo efeito que realizá-la uma única vez. Por exemplo, definir um valor (SET x = 5) é idempotente. Incrementar um valor (x = x + 1) não é.
Como alcançar a idempotência:
- Utilize identificadores de requisição únicos: Durante uma chamada da API que modifica o estado, inclua um identificador único gerado pelo cliente no cabeçalho da requisição (por exemplo,
X-Idempotency-Key). O serviço externo pode então usar essa chave para detectar requisições duplicadas e retornar a resposta original sem reprocessamento. - Projete APIs idempotentes: Se você controla a API, projete endpoints que sejam naturalmente idempotentes. Por exemplo, em vez de um endpoint “criar um pedido”, tenha um endpoint “upsert pedido” que pode criar ou atualizar com base em um identificador de pedido único.
- Verifique o estado antes de tentar novamente: Após uma operação de modificação de estado falhada, se a API suportar, consulte o estado do recurso utilizando o identificador único antes de tentar uma nova tentativa.
Embora eu não tenha um pedaço de código direto para isso (é mais uma questão de design da API e lógica do lado do cliente), aqui está como poderia parecer o raciocínio do seu agente:
“““html
# Pseudo-código do agente para uma operação idempotente
transaction_id = generate_unique_id()
payload = {"data": "some_value", "idempotency_key": transaction_id}
try:
response = external_payment_api.process_charge(payload)
# Sucesso! Armazene transaction_id e response.
except (ConnectionError, TimeoutError, APIError) as e:
# Oh não, falhou. O pagamento ainda foi efetuado?
logger.warning(f"O pagamento falhou, verificando o status com o ID : {transaction_id}")
try:
status_response = external_payment_api.get_transaction_status(transaction_id)
if status_response.get("status") == "completed":
logger.info(f"O pagamento {transaction_id} foi efetivamente concluído durante a verificação da nova tentativa.")
# Considere como bem-sucedido, armazene as informações.
else:
logger.info(f"O pagamento {transaction_id} realmente falhou, tentando repetir (com a mesma chave de idempotência).")
# É aqui que você retry com o *mesmo* transaction_id
# A API de pagamento deve reconhecê-lo e não cobrar duas vezes.
response = external_payment_api.process_charge(payload)
# ... gerenciar o sucesso/falha da nova tentativa
except Exception as check_e:
logger.error(f"Impossível verificar o status da transação para {transaction_id} : {check_e}")
# Necessidade de registrar para uma revisão manual, ou passar para a fila de mensagens mortas
Isso requer a cooperação do serviço externo, mas é um modelo crítico para construir agentes verdadeiramente confiáveis que gerenciam operações financeiras ou outras sensíveis.
Estratégia 4 : Soluções de emergência e degradações elegantes
Às vezes, um serviço externo está simplesmente completamente indisponível, e não há esperança de tentar novamente ou esperar. Nesses casos, um bom agente não simplesmente travará; encontrará uma maneira de fornecer uma experiência degradante, mas ainda útil.
Isso pode significar :
- Utilizar dados armazenados em cache : Se o seu agente precisa de dados específicos de um serviço, mas o serviço está fora do ar, você pode usar uma versão antiga de um cache?
- Fornecer valores padrão : Se um modelo de IA para análise de sentimento está fora do ar, você pode simplesmente classificar todas as entradas como “neutras” ou “desconhecidas” por um certo período, em vez de fazer com que todo o fluxo do agente falhe?
- Mudar para um serviço de emergência : Se sua API de tradução principal está desligada, você pode redirecionar as solicitações para uma secundária, talvez menos eficiente ou mais cara?
- Ignorar etapas opcionais : Se um passo de enriquecimento não crítico falhar, o agente pode simplesmente continuar sem aquele enriquecimento, talvez registrando um alerta?
- Notificar usuários/operadores : No mínimo, falhe elegantemente e comunique claramente o problema ao usuário ou ao operador do sistema.
Minha anedota sobre o erro do serviço de notificação? Minha solução de emergência era simples: se meu serviço de notificação personalizado estava fora do ar, o agente simplesmente registrava o evento localmente e enviava um e-mail para *mim* dizendo “Ei, seu serviço de notificação provavelmente está fora do ar novamente, verifique os registros.” Não ideal para os usuários finais, mas impediu que todo o agente travasse e me garantiu que eu soubesse que havia um problema.
Medidas concretas para o seu próximo projeto de agente
- Assuma a falha : Projete seu agente desde o início esperando que as dependências externas falhem.
- Implemente repetições com uma taxa de retornos exponenciais : Utilize bibliotecas como Tenacity (Python) ou modelos similares em outras linguagens para erros transitórios.
- Implemente disjuntores : Previna falhas em cascata e preserve recursos “ativando” o circuito quando um serviço falha de forma consistente. Pybreaker é um bom começo.
- Priorize a idempotência para as mudanças de estado : Certifique-se de que operações como pagamentos ou criação de registros não se duplicam se ocorrer uma tentativa de repetição. Use IDs únicos.
- Planeje para uma degradação elegante : Identifique as dependências críticas em relação às não críticas e construa soluções de emergência. Qual é a “menos pior” coisa que seu agente pode fazer quando uma dependência falha?
- Monitore de forma agressiva : Todas essas estratégias geram registros. Certifique-se de coletar e analisar esses registros para entender *por que* as coisas falham e com que frequência.
“`
Construir agentes confiáveis não diz respeito apenas a algoritmos astutos ou modelos poderosos. É fundamentalmente uma questão de engenharia de robustez em cada camada, especialmente quando se trata da realidade problemática das dependências externas. Aplicando essas estratégias, você passará menos tempo depurando falhas misteriosas dos agentes e mais tempo criando sistemas autônomos realmente úteis e confiáveis.
🕒 Published: