1. Contexto e Problema de Arquitetura
Sistemas autônomos baseados em LLMs (Large Language Models) que operam em ambiente de produção frequentemente falham quando desenhados como pipelines síncronos e monolíticos. A orquestração de múltiplos agentes exigida por fluxos complexos (onde um Agente Orquestrador Master decompõe um problema e delega tarefas especializadas para N Sub-agentes) impõe desafios críticos de latência, volatilidade de rede e descontinuidade de estado.
Quando Sub-agentes dependem de APIs externas de modelos de linguagem (como a API do Gemini) e de persistência em banco de dados, falhas de conectividade, estouramentos de limite de taxa (rate limits) e payloads malformados (poison pills) podem paralisar o consumo de tarefas ou provocar duplicidade e perda de processamento.
Para resolver o acoplamento síncrono e garantir a resiliência operacional, adota-se um padrão event-driven assíncrono ancorado no Apache Kafka. O Agente Master atua como produtor e consumidor de controle, dividindo o plano de execução em mensagens pub/sub distribuídas em tópicos dedicados. Os Sub-agentes consomem esses tópicos de forma concorrente em Consumer Groups distintos.
Semântica de Entrega e Trade-offs Arquiteturais
- Garantia At-Least-Once (Pelo Menos Uma Vez): Adota-se a confirmação condicionada por meio de commit manual de offsets no Kafka após a conclusão do processamento downstream ou redirecionamento defensivo. Isso previne a perda de mensagens em caso de queda do worker, exigindo a implementação de idempotência na camada de aplicação via chaves de tarefa (
task_id). - Desacoplamento Temporal vs. Latência de Eventos: O modelo de mensageria adiciona pequeno overhead de I/O em comparação com chamadas de função diretas em memória, mas previne falhas em cascata ao isolar oscilações de infraestrutura do fluxo principal do sistema.
- Isolamento em Três Camadas de Exceção: Para conter a contaminação de filas e evitar loops infinitos de reprocessamento, o fluxo de erro dos Sub-agentes é desacoplado rigorosamente em:
- Camada 1 (Dados/Schema): Captura de erros de decodificação de bytes e Poison Pills antes da execução de regras de negócio. Como o payload é inválido e nunca será processado com sucesso, a mensagem é enviada para a Dead Letter Queue (DLQ) e o offset original é confirmado (commit) para evitar o travamento da partição.
- Camada 2 (Integração Downstream): Tratamento de indisponibilidade de APIs externas (Gemini) com timeouts delimitados e backoff exponencial. Se as retentativas forem exauridas, a aplicação não realiza o commit do offset e encerra o processo (fail-fast). Isso evita a perda silenciosa de dados e garante que a mensagem permaneça na fila para reprocessamento assim que a API downstream for restabelecida.
- Camada 3 (Infraestrutura do Broker): Tratamento de falhas de conectividade com o próprio cluster Kafka, preservando a posição do offset não processado e disparando alertas de monitoramento.
2. Ambiente e Setup Reproduzível
Para validar localmente a arquitetura de mensageria sem dependências externas complexas de infraestrutura, utiliza-se a imagem oficial do Apache Kafka em modo KRaft (sem dependência do Zookeeper), parametrizada para nó único local.
Arquivo docker-compose.yml
version: '3.8'
services:
kafka:
image: apache/kafka:3.7.0
container_name: kafka-agente
ports:
- "9092:9092"
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: 'broker,controller'
KAFKA_CONTROLLER_QUORUM_VOTERS: '1@localhost:29093'
KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9092'
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
KAFKA_CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
healthcheck:
test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list"]
interval: 5s
timeout: 10s
retries: 5
Instalação de Dependências Python
Execute o comando abaixo para instalar as bibliotecas necessárias com tipagem e suporte ao Kafka e Pydantic:
pip install confluent-kafka==2.3.0 pydantic==2.6.4 requests==2.31.0
3. Implementação Completa com Tratamento Defensivo e Resiliente
A implementação é dividida em dois arquivos Python completos e auditáveis:
1. schemas.py: Contratos de dados com validações estritas via Pydantic (strict=True).
2. subagent_worker.py: Worker do Sub-agente especialista com isolamento defensivo nas três camadas, manipulação segura de payload bruto e confirmação condicionada de offsets.
Arquivo schemas.py
from typing import Dict, Any
from pydantic import BaseModel, Field, ConfigDict
class TaskPayload(BaseModel):
model_config = ConfigDict(strict=True)
task_id: str = Field(..., description="UUID unico da tarefa para garantia de idempotencia")
agent_name: str = Field(..., description="Nome do sub-agente destino")
instruction: str = Field(..., description="Instrucao detalhada para o processamento")
context_data: Dict[str, Any] = Field(default_factory=dict, description="Metadados complementares")
Arquivo subagent_worker.py
import sys
import json
import time
import logging
import requests
from typing import Optional
from confluent_kafka import Consumer, Producer, KafkaError, KafkaException, Message
from pydantic import ValidationError
from schemas import TaskPayload
# Configuracao de Logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("SubAgentWorker")
# Configuracoes de Infraestrutura
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
INPUT_TOPIC = "agent.tasks.analyzer"
DLQ_TOPIC = "agent.tasks.analyzer.dlq"
CONSUMER_GROUP = "subagent-analyzer-group"
# Configurações da API do Gemini (Mock ou Real)
GEMINI_API_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent"
GEMINI_API_KEY = "MOCK_API_KEY_FOR_TESTING"
class DownstreamAPIError(Exception):
"""Excecao lancada quando a API externa (Gemini) falha de forma irrecuperavel."""
pass
class SubAgentWorker:
def __init__(self) -> None:
# Inicializacao do Consumidor Kafka
consumer_conf = {
"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
"group.id": CONSUMER_GROUP,
"auto.offset.reset": "earliest",
"enable.auto.commit": False, # Controle manual e condicionado de offsets
}
self.consumer = Consumer(consumer_conf)
# Inicializacao do Produtor para DLQ
producer_conf = {
"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
"acks": "all", # Garantia de persistencia no broker antes do ack
"retries": 5,
"max.in.flight.requests.per.connection": 1,
}
self.producer = Producer(producer_conf)
def start(self) -> None:
"""Inicia o loop de consumo do Kafka."""
try:
self.consumer.subscribe([INPUT_TOPIC])
logger.info(f"Sub-agente iniciado. Consumindo topico: '{INPUT_TOPIC}'")
while True:
msg: Optional[Message] = self.consumer.poll(timeout=1.0)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
else:
logger.error(f"Erro de infraestrutura do Kafka: {msg.error()}")
# Camada 3: Erro critico de infraestrutura do Broker. Nao commita e encerra.
raise KafkaException(msg.error())
self._process_message_lifecycle(msg)
except KeyboardInterrupt:
logger.info("Encerrando sub-agente graciosamente...")
finally:
self.consumer.close()
def _process_message_lifecycle(self, msg: Message) -> None:
"""Gerencia o ciclo de vida da mensagem com isolamento defensivo de erros."""
raw_payload: Optional[bytes] = msg.value()
# Camada 1: Isolamento de Decode e Schema (Poison Pills)
if raw_payload is None:
logger.warning("Mensagem com payload nulo recebida. Enviando para DLQ.")
self._route_to_dlq(msg, "Payload nulo")
self._commit_offset_safely(msg)
return
try:
# Decodificacao segura de bytes para string e parse JSON
decoded_str = raw_payload.decode("utf-8")
parsed_json = json.loads(decoded_str)
# Validacao estrita com Pydantic
payload = TaskPayload(**parsed_json)
logger.info(f"Payload validado com sucesso. Task ID: {payload.task_id}")
except (UnicodeDecodeError, json.JSONDecodeError, ValidationError) as err:
logger.error(f"Camada 1 - Poison Pill detectada: {str(err)}")
# Como o payload esta corrompido, a retentativa nao resolvera. Envia para DLQ e commita.
self._route_to_dlq(msg, str(err))
self._commit_offset_safely(msg)
return
# Camada 2: Execucao Downstream (Chamada ao Gemini com Retentativas)
try:
self._execute_downstream_task_with_retry(payload)
# Sucesso absoluto no processamento downstream. Commita o offset.
self._commit_offset_safely(msg)
logger.info(f"Task {payload.task_id} processada e offset confirmado.")
except DownstreamAPIError as err:
# Erro de infraestrutura downstream exaurido.
# VIOLACAO DE SRE EVITADA: Nao enviamos para a DLQ e NAO commitamos o offset.
# O worker deve falhar para evitar perda silenciosa de dados.
logger.critical(
f"Camada 2 - Falha critica downstream exaurida para a Task {payload.task_id}. "
f"Encerrando aplicacao sem realizar commit do offset. Erro: {str(err)}"
)
sys.exit(1)
def _execute_downstream_task_with_retry(self, payload: TaskPayload, max_retries: int = 3) -> None:
"""Executa a chamada para a API do Gemini aplicando backoff exponencial."""
headers = {"Content-Type": "application/json"}
params = {"key": GEMINI_API_KEY}
# Payload estruturado para a API do Gemini
data = {
"contents": [{
"parts": [{
"text": f"Instrucao: {payload.instruction}. Contexto: {json.dumps(payload.context_data)}"
}]
}]
}
backoff_sec = 2.0
for attempt in range(1, max_retries + 1):
try:
logger.info(f"Tentativa {attempt} de {max_retries} - Chamando API Gemini...")
# Timeout delimitado para evitar travamento de thread
response = requests.post(
GEMINI_API_URL,
headers=headers,
params=params,
json=data,
timeout=10.0
)
if response.status_code == 200:
result = response.json()
logger.info(f"Resposta recebida do Gemini para Task {payload.task_id}")
# Processamento de sucesso simulado
return
elif response.status_code in [429, 500, 502, 503, 504]:
logger.warning(
f"Erro transitorio HTTP {response.status_code} recebido da API Gemini. "
f"Aplicando backoff de {backoff_sec}s."
)
else:
# Erros HTTP 4xx (exceto 429) indicam falhas de integracao nao transitorias
raise DownstreamAPIError(f"Erro nao transitorio HTTP {response.status_code}: {response.text}")
except requests.exceptions.RequestException as req_err:
logger.warning(f"Falha de conexao de rede na tentativa {attempt}: {str(req_err)}")
if attempt < max_retries:
time.sleep(backoff_sec)
backoff_sec *= 2.0 # Backoff exponencial
raise DownstreamAPIError(f"Retentativas exauridas para a API Gemini apos {max_retries} tentativas.")
def _route_to_dlq(self, msg: Message, error_reason: str) -> None:
"""Envia o payload original integro para o topico de Dead Letter Queue (DLQ)."""
try:
dlq_headers = {
"error_reason": error_reason.encode("utf-8"),
"original_topic": msg.topic().encode("utf-8"),
"original_partition": str(msg.partition()).encode("utf-8"),
"original_offset": str(msg.offset()).encode("utf-8")
}
# Publicacao sincrona para garantir a persistencia na DLQ antes de prosseguir
self.producer.produce(
topic=DLQ_TOPIC,
value=msg.value(),
key=msg.key(),
headers=list(dlq_headers.items())
)
self.producer.flush()
logger.info(f"Mensagem enviada com sucesso para a DLQ '{DLQ_TOPIC}'.")
except Exception as e:
logger.critical(f"Falha catastrofica ao tentar publicar na DLQ: {str(e)}")
# Se nao for possivel gravar na DLQ, nao podemos commitar o offset original
raise e
def _commit_offset_safely(self, msg: Message) -> None:
"""Realiza o commit manual do offset de forma sincrona."""
try:
self.consumer.commit(message=msg, asynchronous=False)
except KafkaException as e:
logger.error(f"Falha ao realizar commit do offset: {str(e)}")
# Camada 3: Se o commit falhar, a mensagem sera reprocessada.
raise e
if __name__ == "__main__":
worker = SubAgentWorker()
worker.start()
4. Roteiro CLI e Validação de Resiliência
Siga rigorosamente os passos abaixo para subir o ambiente, criar os tópicos necessários, simular cenários de falha e validar o comportamento do sistema.
Passo 1: Inicializar a Infraestrutura Local
Suba o container do Apache Kafka utilizando o Docker Compose:
docker-compose up -d
Verifique se o broker está saudável e pronto para receber conexões:
docker ps --filter "name=kafka-agente"
Passo 2: Criação dos Tópicos no Kafka
Crie o tópico principal de tarefas (agent.tasks.analyzer) e o tópico de mensagens mortas (agent.tasks.analyzer.dlq):
# Criar topico principal
docker exec -it kafka-agente kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create \
--topic agent.tasks.analyzer \
--partitions 3 \
--replication-factor 1
# Criar topico de DLQ
docker exec -it kafka-agente kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create \
--topic agent.tasks.analyzer.dlq \
--partitions 1 \
--replication-factor 1
Valide a criação bem-sucedida de ambos os tópicos:
docker exec -it kafka-agente kafka-topics.sh --bootstrap-server localhost:9092 --list
Passo 3: Executar o Sub-agente Worker
Abra um terminal dedicado e inicie o script do worker:
python subagent_worker.py
5. Laboratório de Testes e Simulação de Falhas
Cenário A: Processamento com Sucesso (Caminho Feliz)
Abra um novo terminal e envie uma mensagem válida para o tópico principal utilizando o produtor nativo do Kafka:
docker exec -it kafka-agente kafka-console-producer.sh \
--bootstrap-server localhost:9092 \
--topic agent.tasks.analyzer
Cole o seguinte payload JSON válido no terminal do console-producer e pressione Enter:
{"task_id": "8f4b2c8a-1a2b-3c4d-5e6f-7a8b9c0d1e2f", "agent_name": "analyzer", "instruction": "Analise o sentimento desta avaliacao", "context_data": {"review": "Excelente produto, superou as expectativas!"}}
Comportamento esperado no terminal do Worker:
* O worker decodifica o JSON com sucesso.
* O Pydantic valida o schema estruturado.
* É feita a chamada simulada para a API do Gemini.
* O offset é commitado manualmente.
Cenário B: Injeção de Poison Pill (Camada 1 – Erro de Schema)
No terminal do kafka-console-producer.sh, envie um payload malformado (violando a tipagem estrita do Pydantic ou contendo JSON inválido):
{"task_id": "invalid-uuid-format", "agent_name": 12345, "instruction": null}
Comportamento esperado no terminal do Worker:
* O worker intercepta a falha de validação na Camada 1.
* A mensagem corrompida é enviada para o tópico agent.tasks.analyzer.dlq.
* O offset da mensagem original é commitado, liberando a partição para as próximas mensagens.
Inspeção do Tópico de DLQ:
Execute o comando abaixo para ler as mensagens enviadas para a DLQ e validar se o payload malformado foi isolado com sucesso:
docker exec -it kafka-agente kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic agent.tasks.analyzer.dlq \
--from-beginning \
--property print.headers=true
Cenário C: Falha Crítica Downstream (Camada 2 – Sem Perda de Dados)
Para simular uma falha de rede ou indisponibilidade total da API downstream (Gemini), podemos alterar o endereço da API no arquivo subagent_worker.py para um IP de destino inexistente ou simular o bloqueio de tráfego de saída.
- No arquivo
subagent_worker.py, altere temporariamente a variávelGEMINI_API_URLpara um endpoint inválido:
“`python
GEMINI_API_URL = “https://192.0.2.1/v1beta/models/gemini-pro” # IP reservado para testes que gera timeout
2. Reinicie o worker:
```bash
python subagent_worker.py
```
3. Envie uma mensagem válida pelo console-producer:
```json
{"task_id": "9e8d7c6b-5a4b-3c2d-1e0f-9a8b7c6d5e4f", "agent_name": "analyzer", "instruction": "Processamento sob indisponibilidade", "context_data": {}}
```
**Comportamento esperado no terminal do Worker:**
* O worker tentará realizar a chamada HTTP.
* Ocorrerá timeout de conexão.
* O worker executará as 3 retentativas com backoff exponencial.
* Ao exaurir as tentativas, o worker lançará `DownstreamAPIError` e **encerrará a execução (sys.exit(1)) sem realizar o commit do offset**.
**Verificação de Offsets Não Commitados (Garantia de Não Perda):**
Execute o comando abaixo para inspecionar o status do grupo de consumidores:
```bash
docker exec -it kafka-agente kafka-consumer-groups.sh \
--bootstrap-server localhost:9092 \
--describe \
--group subagent-analyzer-group
Resultado esperado na tabela de offsets:
Você observará que o valor de CURRENT-OFFSET para a partição correspondente estará posicionado exatamente atrás do LOG-END-OFFSET (gerando um LAG de 1). Isso prova que a mensagem não foi perdida nem descartada na DLQ; ela permanece segura no broker aguardando o restabelecimento do worker e da API downstream.