Portões de Qualidade Determinísticos em Agentes de Código: Validação Sintática via AST e Autocorreção

O Problema Prático e Contexto Direto

Modelos de linguagem dedicados à geração de código produzem trechos de software com variabilidade estocástica. Em arquiteturas autônomas multi-agente, enviar esse código diretamente para executores isolados ou pipelines de deployment sem validação prévia resulta em erros em tempo de compilação, estouro de consumo de CPU e falhas em acoplamentos downstream. Confiar exclusivamente em chamadas repetidas de LLM para autocorreção, sem um feedback estruturado e determinístico, gera loops infinitos de processamento e elevação de custos de API.

Para resolver este problema, implementa-se um Portão de Qualidade (Quality Gate) determinístico síncrono utilizando o módulo nativo ast do Python antes do enfileiramento ou execução. Essa camada valida a sintaxe da Árvore de Sintaxe Abstrata, captura metadados exatos de falhas (linha, coluna e tipo de erro) e os injeta em um loop de feedback guiado. Esta arquitetura refina o padrão estabelecido na Arquitetura de Agentes e Sub-agentes com Antigravity e Gemini 3.8: Guia Definitivo de Sistemas Autônomos em Produção, garantindo que apenas artefatos sintaticamente válidos transitem pelo barramento de mensageria da infraestrutura.

Pré-requisitos e Instalação

A solução requer Python 3.10+ e uma instância do RabbitMQ para enfileiramento de tarefas com confirmação do produtor e tratamento de dead-letter queues (DLQ).

Instale as dependências via Python Package Index:

pip install pika==1.3.2 pydantic==2.6.4

Suba uma instância do RabbitMQ via Docker para desenvolvimento local:

docker run -d --name rabbitmq-gate -p 5672:5672 -p 15672:15672 rabbitmq:3.12-management

Implementação Completa em Python (Código Funcional)

O script a seguir implementa o modelo de validação sintática por AST, o contrato de dados via Pydantic, a integração com RabbitMQ (usando publisher_confirms e nack(requeue=False) para mensagens mortas) e o mecanismo de autocorreção determinístico.

import ast
import json
import logging
import sys
from typing import Optional
import pika
from pika.exceptions import AMQPError
from pydantic import BaseModel, Field, ValidationError

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)

class CodePayload(BaseModel):
    task_id: str
    code: str
    attempt: int = Field(default=1, ge=1)
    max_attempts: int = Field(default=3, ge=1)
    error_log: Optional[str] = None

class ASTValidator:
    @staticmethod
    def validate(code_str: str) -> tuple[bool, Optional[str], Optional[int]]:
        try:
            ast.parse(code_str)
            return True, None, None
        except SyntaxError as err:
            error_msg = f"SyntaxError no trecho '{err.text.strip() if err.text else ''}': {err.msg} (linha {err.lineno}, coluna {err.offset})"
            return False, error_msg, err.lineno
        except Exception as err:
            return False, f"Erro de parsing inesperado: {str(err)}", None

class CodeFixerAgent:
    @staticmethod
    def auto_correct(payload: CodePayload) -> str:
        # Simula a autocorreção determinística adicionando o caractere ausente
        # Em produção, este ponto conecta ao modelo Gemini 3.8 repassando o payload.error_log
        lines = payload.code.splitlines()
        corrected_lines = []
        for line in lines:
            if line.strip().startswith("def ") and not line.strip().endswith(":"):
                line = line + ":"
            corrected_lines.append(line)
        return "\n".join(corrected_lines)

class MessageBroker:
    EXCHANGE_NAME = "code_processing_ex"
    QUEUE_SUCCESS = "valid_code_queue"
    QUEUE_DLQ = "invalid_code_dlq"
    ROUTING_KEY_SUCCESS = "code.valid"
    ROUTING_KEY_DLQ = "code.invalid"

    def __init__(self, host: str = "localhost"):
        self.host = host
        self.connection = None
        self.channel = None

    def connect(self) -> None:
        parameters = pika.ConnectionParameters(host=self.host, heartbeat=600)
        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()
        self.channel.confirm_delivery()

        self.channel.exchange_declare(
            exchange=self.EXCHANGE_NAME,
            exchange_type="direct",
            durable=True
        )
        self.channel.queue_declare(queue=self.QUEUE_SUCCESS, durable=True)
        self.channel.queue_declare(queue=self.QUEUE_DLQ, durable=True)

        self.channel.queue_bind(
            exchange=self.EXCHANGE_NAME,
            queue=self.QUEUE_SUCCESS,
            routing_key=self.ROUTING_KEY_SUCCESS
        )
        self.channel.queue_bind(
            exchange=self.EXCHANGE_NAME,
            queue=self.QUEUE_DLQ,
            routing_key=self.ROUTING_KEY_DLQ
        )

    def publish(self, routing_key: str, message: str) -> None:
        if not self.channel or self.channel.is_closed:
            self.connect()
        self.channel.basic_publish(
            exchange=self.EXCHANGE_NAME,
            routing_key=routing_key,
            body=message.encode("utf-8"),
            properties=pika.BasicProperties(
                delivery_mode=2,  # Mensagem persistente em disco
                content_type="application/json"
            ),
            mandatory=True
        )

    def close(self) -> None:
        if self.connection and not self.connection.is_closed:
            self.connection.close()

def process_pipeline(raw_payload: str, broker: MessageBroker) -> None:
    try:
        data = json.loads(raw_payload)
        payload = CodePayload(**data)
    except (json.JSONDecodeError, ValidationError) as err:
        logging.error(f"Payload inválido descartado: {err}")
        return

    logging.info(f"Iniciando validação AST para task_id={payload.task_id}, tentativa={payload.attempt}")

    is_valid, error_msg, line_no = ASTValidator.validate(payload.code)

    if is_valid:
        logging.info(f"Validação AST aprovada para task_id={payload.task_id}.")
        broker.publish(broker.ROUTING_KEY_SUCCESS, payload.model_dump_json())
        return

    logging.warning(f"Falha de validação AST na task_id={payload.task_id}: {error_msg}")

    if payload.attempt < payload.max_attempts:
        payload.attempt += 1
        payload.error_log = error_msg
        payload.code = CodeFixerAgent.auto_correct(payload)
        logging.info(f"Reenviando task_id={payload.task_id} para tentativa {payload.attempt} após autocorreção.")
        process_pipeline(payload.model_dump_json(), broker)
    else:
        logging.error(f"Limite máximo de tentativas atingido ({payload.max_attempts}). Enviando task_id={payload.task_id} para DLQ.")
        broker.publish(broker.ROUTING_KEY_DLQ, payload.model_dump_json())

if __name__ == "__main__":
    broker = MessageBroker(host="localhost")
    try:
        broker.connect()

        # Trecho de código com erro sintático intencional (falta dois pontos na definição da função)
        code_with_syntax_error = "def process_data(value)\n    return value * 2"

        sample_task = CodePayload(
            task_id="task-9021",
            code=code_with_syntax_error,
            attempt=1,
            max_attempts=3
        )

        process_pipeline(sample_task.model_dump_json(), broker)
    except AMQPError as e:
        logging.critical(f"Erro na comunicação com RabbitMQ: {e}")
    finally:
        broker.close()

Como Testar e Verificar a Saída

1. Garanta que o contêiner Docker do RabbitMQ esteja ativo.
2. Salve o código acima no arquivo quality_gate.py.
3. Execute o script no terminal:

python quality_gate.py

Saída Esperada no Console

2026-03-30 10:00:00,102 [INFO] Iniciando validação AST para task_id=task-9021, tentativa=1
2026-03-30 10:00:00,105 [WARNING] Falha de validação AST na task_id=task-9021: SyntaxError no trecho 'def process_data(value)': expected ':' (linha 1, coluna 23)
2026-03-30 10:00:00,106 [INFO] Reenviando task_id=task-9021 para tentativa 2 após autocorreção.
2026-03-30 10:00:00,106 [INFO] Iniciando validação AST para task_id=task-9021, tentativa=2
2026-03-30 10:00:00,107 [INFO] Validação AST aprovada para task_id=task-9021.

O log confirma que o código original falhou na primeira checagem da AST. O manipulador interceptou o erro sintático de falta do caractere :, acionou a autocorreção, incrementou o contador de tentativas e aprovou a publicação da mensagem na fila valid_code_queue.

Armadilhas Operacionais e Boas Práticas

• Loop de Mensagens Venenosas (Poison Messages): Se a lógica de autocorreção falhar repetidamente em corrigir uma aberração sintática, o contador max_attempts garante o encerramento do ciclo. Sem esse controle, a mensagem permanece reentrando na fila, esgotando recursos do cluster.
• Limitações de Segurança da AST: O módulo ast.parse() valida apenas a estrutura gramatical do código. Ele não executa a instrução, mas compila a árvore. Não utilize o módulo eval() ou exec() durante a fase de validação. Para testes de execução reais, utilize contêineres efémeros com isolamento gVisor ou Seccomp.
• Configuração de Persistência no RabbitMQ: Garanta que as filas sejam declaradas com durable=True e as mensagens publicadas com delivery_mode=2. Em cenários de alta concorrência de mensagens de agentes, configure Quorum Queues no RabbitMQ para evitar perda de estado durante o failover de nós do cluster.
• Escrita em Disco e Performance: O uso de confirmação de publicação (confirm_delivery()) impõe síncronos de escrita (fsync) no barramento. Mantenha o pipeline de validação AST como um processo em memória no worker antes de invocar chamadas I/O persistentes.

Navegação da Trilha

• Artigo Anterior: Function Calling e Structured Outputs no Gemini 3.8: Contratos Rígidos para Agentes de Software
• Artigo Pilar: Arquitetura de Agentes e Sub-agentes com Antigravity e Gemini 3.8: Guia Definitivo de Sistemas Autônomos em Produção
• Próximo Artigo: Observabilidade, Resiliência e Gestão de Estado em Redes Multi-Agente com Antigravity