Monólitos baseados em prompts extensos concentram regras de negócios díspares em um único ciclo de inferência, o que resulta em esgotamento de janela de contexto, degradação de latência e ausência de tolerância a falhas. Quando uma sub-etapa de uma instrução complexa falha em um fluxo monolítico, todo o pipeline de execução é abortado, desperdiçando computação prévia e inviabilizando garantias operacionais de SLA.
A solução arquitetural consiste na decomposição determinística da carga de trabalho em sub-tarefas atômicas, delegadas a sub-agentes especializados que operam de forma concorrente sobre contratos de dados tipados. Como estabelecido no artigo fundamental Arquitetura de Agentes e Sub-agentes com Antigravity e Gemini 3.8: Guia Definitivo de Sistemas Autônomos em Produção, desacoplar a governança global da execução especializada é o requisito primordial para obter escalabilidade horizontal e resiliência em sistemas autônomos distribuídos.
Pré-requisitos e Instalação
O ambiente requer Python 3.11 ou superior e uma instância operacional do RabbitMQ como barramento de mensagens para assegurar desacoplamento e durabilidade.
Execute os comandos abaixo para configurar o broker via Docker e instalar as dependências de rede e validação:
# Inicialização do RabbitMQ com interface de gerenciamento
docker run -d --name rabbitmq-broker \
-p 5672:5672 -p 15672:15672 \
-e RABBITMQ_DEFAULT_USER=guest \
-e RABBITMQ_DEFAULT_PASS=guest \
rabbitmq:3.13-management
# Instalação das bibliotecas Python
pip install aio-pika pydantic
Implementação Completa em Python
O script a seguir implementa a topologia completa de execução distribuída:
1. Modelagem formal de comandos e resultados via pydantic.
2. Declaração explícita de exchanges, filas duráveis e Dead-Letter Exchange (DLX) no RabbitMQ.
3. Despacho com confirmação de entrega do produtor (publisher confirms).
4. Pool de sub-agentes operando com controle de fluxo (prefetch count) e rejeição sem re-enfileiramento (requeue=False) para isolamento de payloads inválidos.
import asyncio
import json
import logging
from enum import Enum
from typing import Any, Dict
import aio_pika
from pydantic import BaseModel, Field, ValidationError
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] (%(name)s) %(message)s"
)
logger = logging.getLogger("AntigravityEngine")
class AgentSpecialty(str, Enum):
SECURITY_AUDIT = "security_audit"
DATA_NORMALIZER = "data_normalizer"
METRICS_EXTRACTOR = "metrics_extractor"
class TaskPayload(BaseModel):
task_id: str = Field(..., min_length=1)
specialty: AgentSpecialty
input_data: Dict[str, Any]
max_retries: int = Field(default=3, ge=0)
class ExecutionResult(BaseModel):
task_id: str
status: str
output: Dict[str, Any]
error_message: str | None = None
RABBITMQ_URL = "amqp://guest:guest@localhost:5672/"
EXCHANGE_NAME = "antigravity.tasks"
DLX_EXCHANGE_NAME = "antigravity.dlx"
WORK_QUEUE_NAME = "antigravity.execution.queue"
DLQ_NAME = "antigravity.poison.dlq"
async def setup_infrastructure(channel: aio_pika.abc.AbstractChannel) -> None:
dlx = await channel.declare_exchange(
DLX_EXCHANGE_NAME, aio_pika.ExchangeType.DIRECT, durable=True
)
dlq = await channel.declare_queue(DLQ_NAME, durable=True)
await dlq.bind(dlx, routing_key="poison")
main_exchange = await channel.declare_exchange(
EXCHANGE_NAME, aio_pika.ExchangeType.DIRECT, durable=True
)
work_queue = await channel.declare_queue(
WORK_QUEUE_NAME,
durable=True,
arguments={
"x-dead-letter-exchange": DLX_EXCHANGE_NAME,
"x-dead-letter-routing-key": "poison",
},
)
await work_queue.bind(main_exchange, routing_key="subtask")
class SpecializedSubAgent:
def __init__(self, specialty: AgentSpecialty):
self.specialty = specialty
async def execute(self, payload: TaskPayload) -> Dict[str, Any]:
logger.info("Executando agente %s na tarefa %s", self.specialty.value, payload.task_id)
await asyncio.sleep(0.3)
if self.specialty == AgentSpecialty.SECURITY_AUDIT:
return {"vulnerabilities_found": 0, "status": "CLEAN"}
if self.specialty == AgentSpecialty.DATA_NORMALIZER:
return {"records_processed": len(payload.input_data), "status": "PARSED"}
if self.specialty == AgentSpecialty.METRICS_EXTRACTOR:
return {"latency_p99_ms": 12.4, "status": "AGGREGATED"}
return {"status": "UNKNOWN_SPECIALTY"}
async def worker_consumer(worker_id: int, channel: aio_pika.abc.AbstractChannel) -> None:
await channel.set_qos(prefetch_count=2)
queue = await channel.get_queue(WORK_QUEUE_NAME)
agents = {
AgentSpecialty.SECURITY_AUDIT: SpecializedSubAgent(AgentSpecialty.SECURITY_AUDIT),
AgentSpecialty.DATA_NORMALIZER: SpecializedSubAgent(AgentSpecialty.DATA_NORMALIZER),
AgentSpecialty.METRICS_EXTRACTOR: SpecializedSubAgent(AgentSpecialty.METRICS_EXTRACTOR),
}
async with queue.iterator() as queue_iter:
async for message in queue_iter:
async with message.process(requeue=False, ignore_processed=True):
try:
raw_body = json.loads(message.body.decode())
task = TaskPayload.model_validate(raw_body)
agent = agents.get(task.specialty)
if not agent:
raise ValueError(f"Agente nao mapeado: {task.specialty}")
output = await agent.execute(task)
result = ExecutionResult(
task_id=task.task_id,
status="COMPLETED",
output=output
)
logger.info("Worker %s finalizou tarefa %s: %s", worker_id, task.task_id, result.model_dump())
await message.ack()
except (ValidationError, json.JSONDecodeError, ValueError) as err:
logger.error("Rejeitando mensagem malformada ou invalida no worker %s: %s", worker_id, str(err))
await message.nack(requeue=False)
except Exception as unhandled_err:
logger.critical("Falha nao tratada no worker %s: %s", worker_id, str(unhandled_err))
await message.nack(requeue=False)
async def orchestrator_publisher(
channel: aio_pika.abc.AbstractChannel,
tasks: list[dict[str, Any]]
) -> None:
exchange = await channel.get_exchange(EXCHANGE_NAME)
for task_dict in tasks:
raw_message = json.dumps(task_dict).encode()
message = aio_pika.Message(
body=raw_message,
delivery_mode=aio_pika.DeliveryMode.PERSISTENT
)
confirmation = await exchange.publish(
message,
routing_key="subtask"
)
logger.info("Orquestrador enviou tarefa: %s, Confirmacao: %s", task_dict.get("task_id"), confirmation)
async def main() -> None:
connection = await aio_pika.connect_robust(RABBITMQ_URL)
async with connection:
setup_channel = await connection.channel()
await setup_infrastructure(setup_channel)
await setup_channel.close()
worker_channel = await connection.channel()
publisher_channel = await connection.channel(publisher_confirms=True)
tasks_to_run = [
{
"task_id": "T-SEC-01",
"specialty": "security_audit",
"input_data": {"target": "auth_service"},
"max_retries": 1
},
{
"task_id": "T-DAT-02",
"specialty": "data_normalizer",
"input_data": {"user_id": 1024, "action": "login"},
"max_retries": 1
},
{
"task_id": "T-MET-03",
"specialty": "metrics_extractor",
"input_data": {"window_seconds": 60},
"max_retries": 1
},
{
"task_id": "T-ERR-04",
"specialty": "invalid_specialty",
"input_data": {},
"max_retries": 0
}
]
worker_tasks = [
asyncio.create_task(worker_consumer(1, worker_channel)),
asyncio.create_task(worker_consumer(2, worker_channel))
]
await orchestrator_publisher(publisher_channel, tasks_to_run)
await asyncio.sleep(2.0)
for wt in worker_tasks:
wt.cancel()
await asyncio.gather(*worker_tasks, return_exceptions=True)
if __name__ == "__main__":
asyncio.run(main())
Como Testar e Verificar a Saída
Com o container do RabbitMQ em execução, execute o pipeline diretamente via terminal:
python orchestrator_subagents.py
Saída Esperada no Console
O log exibirá a declaração das entidades AMQP, o despacho concorrente das sub-tarefas, o processamento paralelo balanceado entre os dois workers e o descarte determinístico da mensagem com formato inválido para a DLQ:
2026-03-30 14:15:00,102 [INFO] (AntigravityEngine) Orquestrador enviou tarefa: T-SEC-01, Confirmacao: None
2026-03-30 14:15:00,104 [INFO] (AntigravityEngine) Orquestrador enviou tarefa: T-DAT-02, Confirmacao: None
2026-03-30 14:15:00,105 [INFO] (AntigravityEngine) Orquestrador enviou tarefa: T-MET-03, Confirmacao: None
2026-03-30 14:15:00,107 [INFO] (AntigravityEngine) Orquestrador enviou tarefa: T-ERR-04, Confirmacao: None
2026-03-30 14:15:00,108 [INFO] (AntigravityEngine) Executando agente security_audit na tarefa T-SEC-01
2026-03-30 14:15:00,109 [INFO] (AntigravityEngine) Executando agente data_normalizer na tarefa T-DAT-02
2026-03-30 14:15:00,110 [ERROR] (AntigravityEngine) Rejeitando mensagem malformada ou invalida no worker 1: 1 validation error for TaskPayload
specialty
Input should be 'security_audit', 'data_normalizer' or 'metrics_extractor' [type=enum, input_value='invalid_specialty', input_type=str]
2026-03-30 14:15:00,412 [INFO] (AntigravityEngine) Worker 1 finalizou tarefa T-SEC-01: {'task_id': 'T-SEC-01', 'status': 'COMPLETED', 'output': {'vulnerabilities_found': 0, 'status': 'CLEAN'}, 'error_message': None}
2026-03-30 14:15:00,413 [INFO] (AntigravityEngine) Worker 2 finalizou tarefa T-DAT-02: {'task_id': 'T-DAT-02', 'status': 'COMPLETED', 'output': {'records_processed': 2, 'status': 'PARSED'}, 'error_message': None}
2026-03-30 14:15:00,414 [INFO] (AntigravityEngine) Executando agente metrics_extractor na tarefa T-MET-03
2026-03-30 14:15:00,718 [INFO] (AntigravityEngine) Worker 1 finalizou tarefa T-MET-03: {'task_id': 'T-MET-03', 'status': 'COMPLETED', 'output': {'latency_p99_ms': 12.4, 'status': 'AGGREGATED'}, 'error_message': None}
Para validar o isolamento da mensagem rejeitada na Dead-Letter Queue via CLI:
docker exec -it rabbitmq-broker rabbitmqadmin get queue=antigravity.poison.dlq count=1
O comando retornará exatamente o payload T-ERR-04, comprovando que o pipeline manteve a execução sem travamento de threads ou consumo de recursos nos agentes especialistas.
Armadilhas Operacionais e Boas Práticas
- Loop de Mensagens Venenosas (Poison Messages): Ao processar tarefas com schemas incompatíveis ou instruções corrompidas, nunca execute
nack(requeue=True). Sem limite estrito de tentativas, a mensagem retornará imediatamente ao topo da fila, consumindo 100% de CPU dos sub-agentes. Configure Dead-Letter Exchanges com roteamento explícito para análise forense assíncrona. - Backpressure e Starvation de Conexões: Defina sempre
prefetch_countexplícito nas instâncias de consumidores. O valor padrão do RabbitMQ distribui todas as mensagens disponíveis em memória para os workers conectados, gerando desbalanceamento severo se uma sub-tarefa demandar tempo computacional desproporcional. - Persistência e Garantia de Disco: A marcação
delivery_mode=PERSISTENTcombinada com filas duráveis protege os payloads contra reinicializações do broker, mas introduz latência de I/O em discos magnéticos ou volumes de rede lentos. Em produção, isole as partições de storage do broker sob unidades NVMe com montagem orientada a transações rápidas. - Idempotência de Execução: Em sistemas distribuídos, a garantia de entrega é tipicamente at-least-once. Sub-agentes que efetuam mutações de estado externo (como escrita em bancos de dados ou chamadas de API externas) devem utilizar o identificador único
task_idcomo chave de idempotência em cache distribuído (ex: Redis com TTL) antes de executar o payload.
Navegação da Trilha
- Artigo Anterior: Arquitetura de Agentes e Sub-agentes com Antigravity e Gemini 3.8: Guia Definitivo de Sistemas Autônomos em Produção
- Próximo Artigo: Function Calling e Structured Outputs no Gemini 3.8: Contratos Rígidos para Agentes de Software