“`html
Olá a todos, Leo aqui da agntdev.com! Hoje eu quero falar sobre algo que tem me ocupado muito a mente ultimamente, especialmente enquanto me envolvo em alguns novos projetos para agentes. Trata-se do lado “Dev” do desenvolvimento de agentes, especificamente, de como procedemos a construir agentes confiáveis a partir de componentes não confiáveis. Sim, você me ouviu bem. Porque sejamos francos, essa é a realidade para a maioria de nós, certo?
Geralmente não trabalhamos com APIs e serviços perfeitamente afinados e de nível enterprise. Mais frequentemente do que o esperado, estamos montando modelos open-source, APIs de terceiros com limites de uso questionáveis, e talvez até algum microserviço caseiro que, digamos assim, tem uma certa personalidade. E ainda assim, a expectativa é sempre que nossos agentes deveriam simplesmente… funcionar. De maneira consistente. Confiável. Mesmo quando os componentes subjacentes agem de forma errática.
Eu segui esse caminho tantas vezes. Lembro de um projeto do ano passado em que estava construindo um agente para ajudar a gerenciar minhas contribuições open-source. Ele precisava interagir com a API do GitHub, um modelo de análise de sentimentos hospedado em um plano gratuito, e um serviço de notificações personalizado que eu criei em um final de semana. Cada um desses tinha suas peculiaridades. O GitHub às vezes me limitava de forma inesperada, o modelo de sentimentos às vezes dava timeout, e meu serviço de notificações… bem, digamos apenas que tinha o hábito de esquecer as boas maneiras após uma hora de atividade. Se eu não tivesse implementado proteções sérias, todo o sistema teria desmoronado como um castelo de cartas.
Portanto, hoje eu quero compartilhar algumas estratégias práticas e modelos que adotei para tornar meus agentes mais resilientes, mesmo quando os componentes dos quais são construídos não são nada confiáveis.
A Verdade Inevital: As Coisas Vão Quebrar
Primeiro, vamos aceitar essa verdade como um dogma. Nenhuma API está 100% disponível. Nenhum modelo é 100% preciso. Nenhuma rede é 100% estável. Uma vez que você abrace esse conceito, pode começar a projetar com a falha em mente, o que, contraintuitivamente, torna seu agente mais eficaz.
O problema que vejo com frequência, especialmente com os novos desenvolvedores que exploram o trabalho com agentes, é que eles presumem que terão sucesso a cada chamada externa. Eles escrevem código como este:
response = external_api.call_method(data)
# Presuma que a response é sempre perfeita e prossiga
processed_data = process_response(response)
E então, quando external_api.call_method gera um erro de conexão, ou retorna um 500, ou simplesmente envia um JSON malformado, todo o agente para. Podemos fazer melhor.
Estratégia 1: Retentativas sólidas com Backoff
Esta é provavelmente a técnica mais fundamental, e mesmo assim, muitas vezes é implementada de forma inadequada ou não é utilizada. Tentar novamente imediatamente após uma falha geralmente é uma má ideia. Se o serviço externo estiver fora do ar, você está apenas continuando a atacá-lo, potencialmente agravando a situação ou sendo limitado em seu uso.
A chave é o backoff exponencial. Isso significa esperar períodos progressivamente mais longos entre cada tentativa. Isso dá ao serviço externo a oportunidade de se recuperar e reduz a carga que você está impondo.
Exemplo: Python com Tenacity
Para Python, minha biblioteca de referência para isso é Tenacity. Tornar a adição de lógica de retentativa incrivelmente limpa.
“““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 de serviços externos."""
pass
# Simula uma chamada API externa não confiável
def call_unreliable_api(data):
if random.random() < 0.6: # 60% de probabilidade de falha
logger.warning(f"Chamada API falhou para os dados: {data}")
raise ExternalServiceError("Falha simulada da API ou tempo limite")
logger.info(f"Chamada API 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"Tentando chamar a 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 trecho:
wait_exponentialtorna os tempos de espera mais longos a cada tentativa (4s, depois ~8s, depois ~16s, etc., até um máximo de 10s).stop_after_attempt(5)significa que tenta no máximo 5 vezes.retry_if_exception_type(ExternalServiceError)garante que ele tente novamente apenas para erros específicos, não, por exemplo, umKeyboardInterrupt.
Este modelo é um salva-vidas. Eu o uso para conexões com o banco de dados, solicitações HTTP e também para comunicação interna entre módulos agentes quando sei que um pode estar temporariamente sobrecarregado.
Estratégia 2: Disjuntores automáticos para prevenir falhas em cascata
As tentativas são ótimas para erros transitórios. Mas o que acontece se o serviço estiver completamente inativo? Tentar novamente várias vezes simplesmente esgotará seus recursos e pode piorar a situação para o serviço externo se ele estiver lutando para se recuperar. Aqui é onde entra o modelo do disjuntor automático.
Pense em um disjuntor elétrico em sua casa. Se houver uma falha (muitas falhas), ele “desarma”, impedindo que mais corrente flua e protegendo o sistema. Depois de um tempo, ele pode ser reiniciado, mas não continuará tentando fazer a corrente passar por um fio em curto.
Para os agentes, um disjuntor automático monitora as chamadas a um serviço externo. Se a porcentagem de falhas ultrapassar um determinado limite dentro de uma janela de tempo definida, o circuito “abre”. Quando está aberto, todas as chamadas subsequentes a esse serviço falham imediatamente sem nem mesmo tentar a chamada. Após um período de “timeout” configurável, o circuito passa para um estado “semi-aberto”, permitindo um número limitado de chamadas de teste para ver se o serviço se recuperou. Se estas forem bem-sucedidas, ele se fecha; se falharem, ele se reabre.
Por que é importante para os agentes:
- Conservação de recursos: Seu agente não está desperdiçando tempo e recursos tentando chamar um serviço inativo.
- Falha mais rápida: Em vez de esperar um tempo limite, seu agente recebe um sinal de falha imediato, permitindo que ele gerencie a situação (por exemplo, usar uma alternativa, registrar o problema, notificar um operador).
- Protege os serviços externos: Impede que seu agente sobrecarregue um serviço em dificuldades.
Geralmente implemento isso utilizando 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 acione em TypeError
def _unreliable_call(self, data):
if random.random() < 0.7: # 70% de chance de falha
logger.warning(f"Simulando erro de API interna para os dados: {data}")
raise ConnectionError("Serviço inatingível")
logger.info(f"Chamada de API bem-sucedida para os dados: {data}")
return {"result": f"processado_{data}"}
def process_data(self, data):
try:
return self.breaker.call(self._unreliable_call, data)
except CircuitBreakerError:
logger.error(f"O circuito está aberto! Não chamei a API para os dados: {data}")
# Lógica de fallback aqui: retornar dados em cache, um valor padrão, ou lançar um erro mais específico
return {"result": "dados_fallback", "source": "disjuntor"}
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) # Simula algum atraso entre as chamadas
Execute este código e você verá o circuito se abrir após algumas falhas, então eventualmente tentar se reabrir, e talvez até se fechar novamente se o serviço simulado começar a se comportar corretamente.
Estratégia 3: Idempotência para operações que alteram o estado
Isso é crucial para qualquer agente que modifica um estado externo (por exemplo, criar um registro, enviar um e-mail, iniciar um pagamento). Se o seu agente tentar executar uma ação, e a rede falhar, ou o serviço externo timeout, como você pode saber se a ação realmente ocorreu?
Se você simplesmente tentar novamente sem considerar a idempotência, pode acabar executando a ação duas vezes. Imagine enviar o mesmo e-mail duas vezes, ou pior, cobrar um cliente duas vezes. Isso não é bom.
Uma operação é idempotente se executá-la várias vezes tiver o mesmo efeito que executá-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:
- Use IDs de solicitação exclusivos: Quando você faz uma chamada de API que altera o estado, inclua um ID exclusivo gerado pelo cliente no cabeçalho da solicitação (por exemplo,
X-Idempotency-Key). O serviço externo pode então usar essa chave para detectar solicitações duplicadas e retornar a resposta original sem precisar processá-la novamente. - Projete APIs idempotentes: Se você controlar a API, projete endpoints que sejam naturalmente idempotentes. Por exemplo, em vez de um endpoint “criar pedido”, use um endpoint “upsert pedido” que pode criar ou atualizar com base em um ID de pedido único.
- Verifique o estado antes de tentar novamente: Após uma operação que altera o estado ter falhado, se a API suportar, interrogar o estado do recurso usando o ID exclusivo antes de tentar novamente.
Embora eu não tenha um snippet de código direto para isso (é mais sobre o design da API e a lógica do lado do cliente), aqui está como o processo de pensamento do seu agente poderia parecer:
“““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 foi concluído de qualquer forma?
logger.warning(f"Pagamento falhou, verificando estado com 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 realmente completado durante a verificação de repetição.")
# Trate como sucesso, armazene as informações.
else:
logger.info(f"O pagamento {transaction_id} realmente falhou, tentando repetir (com o mesmo ID de idempotência).")
# Aqui você tentaria novamente com o *mesmo* transaction_id
# A API de pagamento deve reconhecê-lo e não cobrar novamente.
response = external_payment_api.process_charge(payload)
# ... gerencie o sucesso/falha da recuperação
except Exception as check_e:
logger.error(f"Não foi possível verificar o estado da transação para {transaction_id}: {check_e}")
# Necessário registrar para uma revisão manual, ou mover para a fila de dead-letter
Isso requer cooperação do serviço externo, mas é um esquema crítico para construir agentes confiáveis que gerenciam operações financeiras ou outras operações sensíveis.
Estratégia 4: Fallback e Degradação Gradual
Às vezes, um serviço externo está completamente indisponível e não há esperança de tentar novamente ou esperar. Nesses casos, um bom agente não simplesmente trava; encontra uma maneira de fornecer uma experiência degradada, mas ainda útil.
Isso pode significar:
- Usar 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 não atualizada armazenada em cache?
- Fornecer valores padrão: Se um modelo de IA para análise de sentimento estiver fora do ar, você pode simplesmente classificar todas as entradas como “neutras” ou “desconhecidas” por um período, em vez de fazer o fluxo do agente falhar?
- Mudar para um serviço de backup: Se sua API de tradução primária estiver fora do ar, você pode direcionar as solicitações para uma secundária, talvez menos eficiente ou mais cara?
- Omitir etapas opcionais: Se uma etapa de enriquecimento não crítica falhar, o agente pode simplesmente prosseguir sem aquele enriquecimento, talvez registrando um aviso?
- Notificar usuários/operadores: Pelo menos, falhe de maneira gradual e comunique claramente o problema ao usuário ou operador do sistema.
Minha anedota sobre o serviço de notificação que falhou? Meu fallback era simples: se meu serviço de notificação personalizado caísse, 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, confira os logs.” Não ideal para os usuários finais, mas evitou que todo o agente travasse e garantiu que eu soubesse que algo estava errado.
Diretrizes Práticas 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 tentativas com backoff exponencial: Use bibliotecas como Tenacity (Python) ou esquemas similares em outras linguagens para erros transitórios.
- Disperse interruptores de circuito: Prevenir falhas em cascata e conservar recursos “ativando” o circuito quando um serviço falha constantemente. Pybreaker é um bom começo.
- Priorize a idempotência para mudanças de estado: Garantir que operações como pagamentos ou criação de registros não sejam duplicadas se uma nova tentativa ocorrer. Use IDs únicos.
- Planeje para uma degradação gradual: Identifique dependências críticas em relação às não críticas e construa fallbacks. Qual é a coisa “menos ruim” que seu agente pode fazer quando uma dependência falha?
- Monitore agressivamente: Todas essas estratégias geram logs. Certifique-se de coletar e analisar esses logs para entender *por que* as coisas estão falhando e com que frequência.
“`
Construir agentes confiáveis não se trata apenas de algoritmos inteligentes ou modelos poderosos. Trata-se fundamentalmente de engenheirar solidez em cada camada, especialmente quando se lida com a realidade caótica das dependências externas. Aplicando essas estratégias, você passará menos tempo depurando falhas misteriosas dos agentes e mais tempo construindo sistemas autônomos genuinamente úteis e confiáveis.
Quais são suas estratégias preferidas para gerenciar serviços externos pouco confiáveis? Deixe um comentário abaixo, eu adoraria ouvir suas histórias e soluções!
🕒 Published: