Loop Smith: Engenharia de IA de circuito fechado — Execução autônoma/de meta para pipelines autocorretivos em…
26Publicado dias atrás4410

Loop Smith: Engenharia de IA de circuito fechado — Execução autônoma/de meta para pipelines autocorretivos em antigravidade 2.0

Pare de avisar o agente passo a passo. Dê um objetivo, uma barra para limpar, e deixe a máquina de estados escrever, correr, e corrija seu próprio código até que a saída seja verificada.

Este guia tem como alvo Antigravidade 2.0 — a liberação de quatro superfícies (aplicativo de desktop, cérebro CLI, SDK antigravidade do Google, e nuvem corporativa) que compartilha um chicote de agente. O SDK é pré-v1.0; nomes de símbolos e sinalizadores CLI abaixo refletem a API documentada em meados de 2026. Trate os padrões como estáveis ​​e verifique novamente as assinaturas exatas em relação aos documentos atuais antes de enviar.

Índice

  1. O problema com o agente sob demanda
  2. A mudança arquitetônica: engenharia de circuito fechado como uma máquina de estado
  3. Etapa 1 — Inicializar o projeto e a máquina de estado (cérebro)
  4. Etapa 2— Definir o objetivo e acionar o modo de execução/meta (SDK)
  5. Etapa 3— Implementar o loop de autocorreção
  6. Etapa 4 — Verificação e finalização do estado
  7. Conclusão

1. O problema com o agente sob demanda

A maior parte da “engenharia de IA” hoje ainda é conversacional. Você avisa; o modelo responde. Você percebe que a resposta está errada; você avisa novamente. Você cola um traceback; pede desculpas e tenta mais uma vez. A inteligência é real - mas você são a malha de controle. Você é quem executa o código, lê o erro, decide se a saída é boa o suficiente, e alimenta a próxima instrução de volta. Tire o humano daquele assento e todo o sistema parará.

Tudo bem para uma janela de bate-papo. Tudo desmorona no momento em que você deseja que um agente produzir uma entrega — um conjunto de dados limpo, um relatório financeiro reconciliado, uma migração que realmente compila. O verdadeiro trabalho analítico é iterativo e autorreferencial: você escreve um roteiro, ele trava em uma string de moeda, você corrige a análise, funciona, mas os totais não se reconciliam, você corrige a agregação, e só então a saída é confiável. Cada uma dessas flechas é uma decisão. Um agente sob demanda faz com que você forneça todos eles.

Existem três falhas específicas escondidas em “basta solicitar novamente”:

  • Nenhuma definição de concluído. Um agente conversacional para quando sentimentos finalizado, ou seja, quando produz texto fluente. Fluente não é o mesmo que correto. “Aqui está sua análise de receita!” sobre um script que eliminou silenciosamente um terço das linhas é exatamente como uma resposta correta. Nada no loop verifica.
  • O humano é o manipulador de erros. O agente não tem memória durável de o que estava tentando alcançar através das curvas, então não é possível dizer se sua última tentativa está mais perto ou mais longe do objetivo. Você mantém esse estado em sua cabeça. Você é o loop while, a tentativa/exceto, e a afirmação — manualmente, volta após volta.
  • Nenhuma saída verificada. Porque não existe um critério de sucesso explícito, não há momento para o qual o sistema possa apontar e dizer "feito, e aqui está a prova.” Simplesmente… desaparece quando você para de digitar.

A solução não é um modelo mais inteligente ou um prompt mais longo. É um arquitetura: dar ao sistema o objetivo e a definição de feito, então deixe-o executar seu próprio loop de controle - gerar, executar, inspecionar, refinar — até que a saída ultrapasse uma barra que possa verificar por si mesmo. Isso é engenharia de IA de circuito fechado, e antigravidade 2.0 torna-o um modo de execução de primeira classe: /meta.

Vamos construir um de ponta a ponta. O exemplo em execução é uma carga de trabalho que toda equipe de dados reconhece: entregue a um agente um CSV confuso e um objetivo analítico, e receba de volta um JSON verificado - sem nenhuma volta humana no meio. O conjunto de dados revida de propósito (revenue stored as "$1,299.00", células da região em branco, formatos de data mistos), so the agent's primeiro tentativa falha e você pode vê-la se corrigir. Chamamos a implementação de referência Loop Smith.

2. A mudança arquitetônica: engenharia de circuito fechado como uma máquina de estado

Pare de pensar no agente como um chatbot com scrollback. Pense nisso como um processo com estados. A qualquer instante, uma corrida autônoma está fazendo exatamente uma coisa – planejando, ou gerando código, ou executá-lo, ou inspecionando o resultado, ou refinamento — e a única questão interessante é qual estado vem a seguir. Essa pergunta tem um pequeno, conjunto finito de respostas legais. Codifique-os, e você tem um máquina de estado finito: o mais chato, estrutura de controle mais testada em batalha em toda a engenharia, apontou para um LLM.

Aqui está toda a topologia. Leia o ciclo no meio - esse loop é o mecanismo de autocorreção.

foto[1]-Loop Smith: Engenharia de IA de circuito fechado - Execução autônoma / de meta para pipelines autocorretivos em… Para Windows 7,8,10,11-Winpcsoft.com
  • Determinismo que você pode auditar. O fluxo de controle reside em uma tabela de transição, não no capricho da modelo. A corrida só pode se mover ao longo das bordas que você desenhou. Cada execução emite uma fita reproduzível — planejar → gerar → executar → inspecionar → refinar → executar → inspecionar → verificado — então, quando algo dá errado, você lê a fita em vez de adivinhar. O modelo fornece julgamento dentro de um estado; o máquina decide para onde a corrida pode ir a seguir. Essa divisão é todo o truque.
  • Autocorreção como um ciclo, não é uma nova tentativa. O loop EXECUTE → INSPECT → REFINE → EXECUTE é um circuito de feedback fechado. A INSPECT observa o real resultado (código de saída, stderr, se os números se reconciliam) e emite um diagnóstico; REFINE consome esse diagnóstico e emite um programa melhor; EXECUTE testa. Nenhuma curva humana em qualquer lugar desse circuito. The agent isn't "trying again" — it's reading its own error and steering.
  • Uma saída verificada, previamente definido. A execução termina em VERIFIED somente quando cada verificação de aceitação passa — e esses cheques são declarados antecipadamente, no estado PLANO, como predicados verificáveis ​​por máquina (código de saída 0, o artefato analisa, S(por_região) == total dentro de um centavo). "Done" deixa de ser uma vibe e se torna um booleano. Se o orçamento da tentativa acabar primeiro, a execução sai FAILED — em voz alta, com sua fita - em vez de fingir.

O oposto disso é o circuito aberto: solicitar, enviar mensagem de texto, humano detém o estado. Tudo abaixo é a mudança de aberto para fechado. Duas regras de design fazem com que funcione na prática, e vale a pena indicá-los antes de qualquer código:

  1. Separar o julgamento do controle. LLMs são extraordinários no julgamento dentro de um estado (“este rastreamento significa que a coluna de moeda é uma string; tira $ e , before casting") e não confiável por ser seu próprio agendador. Portanto, o gráfico do estado é simples, Python determinístico; os agentes nunca decidem qual estado vem a seguir — eles apenas decidem what to do in the state they're in.
  2. Verifique com código, não vibrações. As partes de “pronto” que um computador pode verificar — códigos de saída, esquemas, reconciliação — são verificados em código, como verdade básica, o modelo não pode anular. O julgamento do modelo está em camadas superiores para as coisas que o código não pode afirmar, nunca por baixo.

Vamos construí-lo.

3. Etapa 1 — Inicializar o projeto e a máquina de estado (cérebro)

Instale as duas superfícies que você usará. O CLI (cérebro) é o chicote terminal baseado em Go; o SDK (google-antigravidade) é o mesmo equipamento que uma biblioteca Python. Você prototipa o loop interativamente na CLI, então codifique-o em Python.

# 1. Instale a CLI Antigravidade (macOS / Linux). Ele chega em ~/.local/bin/agy
curl -fsSLhttps://antigravidade.google/cli/install.sh | festa
# Windows PowerShell:  irm https://antigravidade.google/cli/install.ps1 | nada
# 2. Confirme se o chicote está ativo e veja quais modelos você pode alcançar
modelos cerebrais
# 3. Andaimes do projeto
loopsmith mkdir && loopsmith de cd
python -m venv.venv && fonte .venv/bin/activate
# 4. Instale o SDK. IMPORTANTE: instale a roda publicada, não é uma fonte
#    Confira - a roda envia o binário `localharness` que o SDK dirige.
pip instalar google-antigravidade pydantic python-dotenv pandas
# 5. Autenticar. Ou uma chave direta de Gêmeos…
export GEMINI_API_KEY="your_key_here"
#    …ou rotear pela Vertex AI com credenciais padrão do aplicativo:
#    login padrão do aplicativo gcloud auth   # então defina VERTEX = 1 em .env

Antes de escrever um único agente, escreva o máquina de estado. Esta é a inversão que torna legível uma execução autônoma: a topologia é definida primeiro, em isolamento, sem nenhum modelo perto dele. Os sete estados, as transições legais, e um guarda que recusa qualquer coisa fora do gráfico:

# loop/estados.py
de classes de dados importar classe de dados, campo
da importação de enum Enum
classe Estado(str, Enum):
    PLAN = "plan"          # decompor o objetivo em uma abordagem + cheques
    GENERATE = "generate"  # escreva o primeiro script de análise
    EXECUTE = "execute"    # execute o script na sandbox
    INSPECT = "inspect"    # julgar a corrida em relação às verificações de aceitação
    REFINE = "refine"      # reescreva o script para corrigir o que o INSPECT encontrou
    VERIFIED = "verified"  # terminal SUCESSO - todos os cheques passam
    FAILED = "failed"      # FALHA no terminal - tentativa de orçamento esgotado
TERMINAL = conjunto congelado({Estado.VERIFICADO, Estado.FALHA})
# A tabela de transição É a arquitetura. Um movimento não listado aqui é um bug.
TRANSIÇÕES = {
    Estado.PLAN:     conjunto congelado({Estado.GERAR}),
    Estado.GERAR: conjunto congelado({Estado.EXECUTE}),
    Estado.EXECUTE:  conjunto congelado({Estado.INSPECIONAR}),
    Estado.INSPECIONAR:  conjunto congelado({Estado.VERIFICADO, Estado.REFINE, Estado.FALHA}),
    Estado.REFINE:   conjunto congelado({Estado.EXECUTE}),
}
classe Transição Ilegal(Erro de tempo de execução):
    """Raised when the run attempts a move the transition table forbids."""
@dataclass
classe Máquina:
    estado: Estado = Estado.PLAN
    história: lista[Estado] = campo(default_factory=lambda: [Estado.PLAN])
    definitivamente para(auto, próximo: Estado) -> Estado:
        permitido = TRANSITIONS.get(estado próprio, conjunto congelado())
        se nxt não for permitido:
            aumentar a transição ilegal(
                f"illegal transition {self.state.value!R} → {nxt.valor!R} "
                f"(permitido: {classificado(s.value para s em permitido) or 'none - terminal'})"
            )
        self.estado = nxt
        self.history.append(próximo)
        retornar próximo
    @propriedade
    definitivamente feito(auto) -> bool:
        retornar self.state em TERMINAL
    fita de definição(auto) -> str:
        retornar " → ".join(s.valor para s em self.history)

Por que esta é a primeira coisa que você escreve, não o último:

  • TRANSIÇÕES é o contrato. Observe o ciclo: EXECUTAR → INSPECIONAR → REFINAR → EXECUTAR. Esse único loop em um dicionário é o mecanismo de autocorreção - a execução o percorre até que INSPECT emita VERIFIED. And notice what's ausente: não há vantagem entre REFINE e GENERATE, nenhuma vantagem fora de VERIFIED. Illegal moves aren't handled defensively; they're impossible.
  • Máquina.para() é o único ponto de estrangulamento. Cada mudança de estado em todo o sistema flui através deste método. That's what guarantees history is complete and the tape is trustworthy — there's no back door that mutates state without recording it. Em passo 4 o gancho de auditoria narra isso ao vivo.
  • Não tem modelo, então é testável. Você pode testar a unidade da topologia — “uma execução pode chegar a VERIFIED sem passar por INSPECT?” — com zero chamadas de API e zero instabilidade. A parte estocástica (os agentes) e a parte determinística (o gráfico) nunca contaminem um ao outro.

Interativamente, você pode observar a mesma máquina rodando antes de escrever qualquer Python. Da idade TUI, /objetivo coloca você no modo de execução e transmite as transições de estado conforme elas acontecem:

cérebro> /goal "Reconcile revenue by region in ./data/sales.csv into a verified JSON"
◆ planejar → 6 cheques de aceitação declarados
◆ gerar → análise.py (41 linhas)
◆ executar → sair 1  (Erro de valor: não foi possível converter string em float: '$1,299.00')
◆ inspecionar → NÃO verificado · reconcilias_total bloqueado por falha
◆ refine     → strip '$'/',' antes do elenco
◆ executar → sair 0  · resultado.json escrito
◆ inspecionar → NÃO verificado · no_null_region FALHA (região em branco vazou como chave)
◆ refinar → eliminar linhas da região em branco
◆ executar → sair 0
◆ inspecionar → VERIFICADO ✓  (6/6 cheques)
◆ pronto → fita: planejar → gerar → executar → inspecionar → refinar → executar → inspecionar → refinar → executar → inspecionar → verificado

Essa é a história toda em uma tela: duas autocorreções, nenhuma volta humana, uma saída verificada. Agora construímos o Python que o torna reproduzível.

4. Etapa 2— Definir o objetivo e acionar o modo de execução/meta (SDK)

/o modo objetivo precisa de duas entradas e nada mais: um objetivo de alto nível e um definição de feito. Todo o resto — quantas iterações, qual roteiro, qual correção - o loop decide por si mesmo.

Existem duas maneiras legítimas de invocá-lo, e um engenheiro sênior deve saber ambos.

Padrão A — arnês nativo /meta. Entregue ao arnês o objetivo e um critério de sucesso e deixe isto possuir o circuito. O SDK expõe run_goal; os planos de arnês, gera, executa, inspeciona, e refina internamente, e retorna um GoalResult. Este é o menor código e o padrão correto para trabalho exploratório:

# o formato de uma linha – o arnês percorre todo o circuito fechado para você
do agente de importação google.antigravity, Configuração de meta
assíncrono def quick_goal():
    assíncrono com Agent.for_goal(Configuração de meta(model="gemini-3-pro")) como agente:
        resultado = aguardar agente.run_goal(
            objetivo=(
                "Analyze ./data/sales.csv and report total revenue per region, "
                "the top region, e sua participação na receita total como result.json."
            ),
            sucesso=(
                "python exits 0; result.json analisa; soma(por_região) == receita_total "
                "within 0.01; top_region == argmax(por_região); nenhuma chave de região em branco."
            ),
            max_iterations=5,   # o orçamento de autocorreção
        )
        imprimir(resultado.estado_final, resultado.verificado, resultado.tape)

Padrão B — a máquina de estado explícita. Quando você precisar determinismo e auditabilidade — o que um pipeline de produção sempre faz — você mesmo conduz o mesmo loop com as primitivas do SDK, então cada transição, cada cheque, e cada refinamento é o seu código. Isto é o que o resto do tutorial constrói, porque ver a maquinaria é o ponto.

De qualquer jeito, a definição de pronto não é prosa que você espera que o modelo honre - é uma lista de verificações de aceitação, e todo o trabalho do primeiro agente é produzi-los. A produção estruturada é o que torna isso possível. Cada mudança de modelo no LoopSmith é restrita a um esquema Pydantic via response_schema=, então a máquina de estado recebe objetos digitados nos quais ele pode ramificar, não é um texto que precisa ser analisado:

# loop/esquemas.py  (os contratos que fluem entre estados)
da importação de enum Enum
da importação pydantic BaseModel, Campo
classe CheckKind(str, Enum):
    DETERMINISTIC = "deterministic"  # aplicado no código (verificador.py)
    JUDGEMENT = "judgement"          # delegado ao agente INSPECT
classe AcceptanceCheck(Eles são modelos):
    eu ia: str = Campo(description="stable slug, por exemplo. 'reconciles_total'")
    descrição: str
    tipo: CheckKind
plano de aula(Eles são modelos):
    objetivo: str
    abordagem: lista[str]
    contrato_saída: str           # o JSON exato que o script deve escrever
    cheques: lista[Verificação de aceitação] = Campo(comprimento_min = 1)
classe GeneratedCode(Eles são modelos):    # reutilizado por AMBOS geram e refinam
    nome do arquivo: str = "analysis.py"
    linguagem: str = "python"
    fonte: str                    # o conteúdo completo do arquivo, pronto para correr
    notas: str = ""                # o que mudou vs.. a última versão
inspeção de classe(Eles são modelos):       # a decisão da filial, como dados
    verificado: bool                 # Verdadeiro se CADA verificação de aceitação for aprovada
    falha_checks: lista[str] = Campo(default_factory=lista)
    diagnóstico: str                 # causa raiz, fundamentado em stderr / cheques
    fix_hint: str = ""             # mudança concreta que a REFINE deveria fazer

Duas decisões de esquema carregam a arquitetura:

  • Inspection.verified é um bool, então o branch é um campo lido. Todo o INSPECIONAR → {VERIFICADO, REFINAR} o garfo colapsa para se inspeção.verificado:. Sem regex sobre prosa de modelo, no "did it say the word passou?" heurística. O modelo preenche um campo digitado; a máquina lê.
  • GeneratedCode é compartilhado por GENERATE e REFINE. Um refinamento é apenas a próxima versão do mesmo artefato. Porque ambos os estados emitem o esquema idêntico, the transition table can route both into EXECUTE with no special casing — the loop doesn't care whether a script is the first draft or the third.

5. Etapa 3— Implementar o loop de autocorreção

Agora a substância: os quatro agentes, a sandbox que executa seu código, e o motorista que anda na máquina de estado. Cada agente é cunhado a partir de uma fábrica com um conjunto de guarda-corpos - o Chicote de Agente Compartilhado. Uma execução autônoma que escreve e executa seu próprio código é exatamente a execução que você deseja no menor tempo possível: política de negação padrão, uma linha de auditoria por chamada de ferramenta, sem rede, sem casca.

# loop/harness.py
importar sistema operacional
digitando import Callable, Sequência
de google.antigravity importar LocalAgentConfig
de google.antigravity.hooks importar ganchos, política
MODELO = os.getenv("LOOPSMITH_MODEL", "gemini-3-pro")
# Menor privilégio, codificado uma vez, herdado por cada agente no loop.
BASE_POLÍTICAS = [
    política.permitir("view_file"),
    política.permitir("write_file", quando = lambda a: _dentro_do_espaço de trabalho(a.get("path", ""))),
    política.deny("run_command"),   # agentes nunca desembolsam - o SANDBOX faz
    política.deny("net_fetch"),     # um loop autocorretivo deve permanecer hermético
    política.deny("*"),             # negação padrão: tudo o que não é permitido é recusado
]
classe StateAuditHook(ganchos.PostToolCallHook):
    execução de definição assíncrona(auto, contexto, dados) -> Nenhum:
        imprimir(f"[auditoria] estado={getattr(contexto, 'label', '?')} "
              f"tool={getattr(dados, 'name', '?')} ok={getattr(dados, 'ok', Verdadeiro)}")
def build_agent_config(*, rótulo: str, instruções_do_sistema: str,
                       ferramentas: Sequência[Chamável] = ()) -> LocalAgentConfig:
    kwargs = ditado(
        modelo=MODELO, rótulo = rótulo, instruções_do_sistema = instruções_do sistema,
        ferramentas=lista(ferramentas), políticas=BASE_POLICIES, ganchos =[Gancho de Auditoria Estadual()],
    )
    se os.getenv("VERTEX") == "1":
        kwargs["vertex"] = Verdadeiro
    elif os.getenv("GEMINI_API_KEY"):
        kwargs["api_key"] = os.environ["GEMINI_API_KEY"]
    retornar LocalAgentConfig(**kwargs)
def _dentro_do_espaço de trabalho(caminho: str) -> bool:
    return "workspace" em os.path.normpath(caminho).dividir(os.sep)

A decisão crítica aqui é política.deny("run_command") — os próprios agentes recebem sem casca. Isso parece paradoxal para um loop cujo trabalho é executar código, até ver onde a execução realmente acontece: em um separar, sandbox determinística o orquestrador controla, not inside a model's tool call. O modelo escreve o programa; um subprocesso governado corre isto. O modelo nunca pode ser instalado ou enrolado, porque nunca lhe foi entregue as chaves.

Os quatro agentes — julgamento, sem fluxo de controle

O agente de cada estado é uma configuração de responsabilidade única. Nenhum deles possui lógica de orquestração; nenhum deles sabe que está em loop. Eles raciocinam sobre o que recebem e emitem um objeto digitado — é isso.

# loop/agentes.py  (resumido – prompts completos do sistema no repositório)
de .harness importar build_agent_config
def planner_config():     # PLANO: objetivo -> abordagem + verificações verificáveis ​​por máquina
    retornar build_agent_config(label="plan", instruções_do sistema =(
        "You are a senior data analyst PLANNER. Dado um objetivo e uma prévia "
        "of a messy CSV, produzir uma abordagem, um output_contract exato para o "
        "JSON the script must write, e uma lista de VERIFICAÇÕES DE ACEITAÇÃO. Prefira estes "
        "deterministic ids when they apply: saídas_limpas, artefato_existe, "
        "schema_ok, no_null_region, reconcilia_total, top_is_argmax. A corrida é "
        "'verified' somente quando TODOS os cheques passarem - do not under-specify."))
def gerador_config():   # GERAR: escreva o primeiro roteiro
    retornar build_agent_config(label="generate", instruções_do sistema =(
        "You are a Python DATA ENGINEER. Escreva um COMPLETO, script independente "
        "that reads os.environ['LOOPSMITH_DATASET'], executa o planejado "
        "analysis, and writes 'result.json'. Use apenas a biblioteca padrão e "
        "pandas. Sem rede. Return the full file contents."))
def inspector_config():   # INSPECIONAR: verificado versus. refinar, baseado em verificações duras
    retornar build_agent_config(label="inspect", instruções_do sistema =(
        "You are a rigorous VERIFICATION agent. Você recebe os cheques de aceitação, o "
        "sandbox RunResult, e os resultados das verificações DETERMINÍSTICAS já "
        "computed in code. Você NÃO pode marcar como verificado se houver alguma verificação determinística "
        "failed - essas são verdades básicas. Quando não verificado, escreva um diagnóstico "
        "rooted in the real stderr/failed checks and a concrete fix_hint."))
def refinador_config():     # REFINAR: reescrever para corrigir exatamente o que o INSPECT diagnosticou
    retornar build_agent_config(label="refine", instruções_do sistema =(
        "You are a DEBUGGING engineer. Dado o script atual, o RunResult "
        "showing how it failed, e um diagnóstico + fix_hint, produzir o PRÓXIMO completo "
        "version that fixes the identified problem while preserving what worked."))

A assimetria é o design. O planejador transforma um objetivo difuso em uma barra testável. O inspetor é explicitamente proibido de anular as verificações determinísticas — seu julgamento fica em cima de verdade fundamental, nunca por baixo. O refinador recebe um diagnóstico, não é um vago “tente novamente,”então sua reescrita é direcionada. Cada um tem uma pequena janela de contexto e um trabalho.

A sandbox – onde o código gerado realmente é executado

Este módulo é real, Python executável — sem SDK, nenhum modelo. É o estado EXECUTE, e é o limite de segurança. Executa o script do agente como um subprocesso, em um diretório de trabalho confinado, com um tempo limite de relógio de parede e um ambiente despojado, e relata exatamente o que aconteceu.

# loop/sandbox.py  (essencial)
subprocesso de importação, sistema, tempo
do caminho de importação pathlib
de .schemas importar GeneratedCode, RunResult
ESPAÇO DE TRABALHO = Caminho(__arquivo__).resolver().pai.pai / "workspace"
def código_de_gravação(código: Código gerado) -> Caminho:
    ESPAÇO DE TRABALHO.mkdir(pais = Verdadeiro, exist_ok = Verdadeiro)
    caminho = ESPAÇO DE TRABALHO / código.nome do arquivo
    caminho.write_text(código.fonte, encoding="utf-8")
    caminho de retorno
def run_python(caminho: Caminho, *, conjunto de dados: Caminho, tempo limite_s: flutuar = 30.0) -> RunResult:
    antes = _instantâneo()                      # quais arquivos existem antes da execução
    iniciado = tempo.monotônico()
    tentar:
        proc = subprocesso.run(
            [sys.executável, str(caminho)],
            cwd = ESPAÇO DE TRABALHO,                     # limitar o diretório de trabalho
            env={"LOOPSMITH_DATASET": str(conjunto de dados),   # ambiente mínimo - nenhum segredo vaza
                 "PATH": _caminho_seguro(), "PYTHONUNBUFFERED": "1"},
            capture_output = Verdadeiro, texto = Verdadeiro, tempo limite = tempo limite_s,
        )
        código_de_saída, saída padrão, stderr = proc.returncode, proc.stdout, proc.stderr
        timed_out = Falso
    exceto subprocess.TimeoutExpired como e:
        código_de_saída, saída padrão = 124, (e.stdout or "")
        stderr = (e.stderr or "") + f"\n[caixa de areia] morto depois {tempo limite_s}tempo limite"
        timed_out = Verdadeiro
    artefatos = classificados(p.name para p em _snapshot() - antes)   # o que a corrida criou
    retornar RunResult(código_de_saída=código_de_saída, stdout=_clip(saída padrão), stderr=_clip(stderr),
                     duração_s = rodada(tempo.monotônico() - iniciado, 3),
                     artefatos = artefatos, timed_out=timed_out)

Linha por linha, este é o contrato EXECUTE completo:

  • subprocesso.run(…, cwd = ESPAÇO DE TRABALHO, tempo limite = tempo limite_s) — o código gerado é executado outside the agent's process, em seu próprio diretório, e não pode durar mais que o tempo limite. Um loop infinito que o modelo escreveu obtém SIGKILL, superfícies como exit_code=124, e se torna apenas mais uma entrada INSPECT - não um pipeline travado.
  • env={…} — um ambiente construído à mão. O caminho do conjunto de dados chega via LOOPSMITH_DATASET; suas chaves de API, credenciais de nuvem, e aliases de shell fazem não. O script vê exatamente o que precisa e nada mais.
  • _instantâneo() antes e depois — o sandbox diferencia o espaço de trabalho para relatar precisamente quais artefatos a execução produziu. That's how the loop sabe result.json foi escrito, rather than trusting the model's stdout claim that it was.
  • O RunResult é o sinal de feedback. Código de saída, saída padrão, stderr, artefatos — a verdade crua do que aconteceu, entregue diretamente para INSPECT. O loop se corrige contra comportamento observado, never against the model's hopes.

O motorista - andando com a máquina

Aqui está o circuito fechado, de cima para baixo. Esta é a forma explícita de /goal: você pode rastrear cada transição, e cada m.to(…) é um guardado, movimento registrado.

# loop/orchestrador.py
do caminho de importação pathlib
do agente de importação google.antigravity
de . importar conjunto de dados, caixa de areia, verificador
de .agentes importar planner_config, gerador_config, inspetor_config, refinador_config
do plano de importação .schemas, Código gerado, Inspeção, MetaResultado
da máquina de importação .states, Estado
assíncrono def _turn(configuração, incitar, esquema):
    """One isolated agent, um turno digitado, then tear down."""
    assíncrono com agente(configuração) como agente:
        resp = aguardar agente.chat(incitar, response_scheme=esquema)
        retornar aguardar resp.parsed()            # uma instância validada do Pydantic
assíncrono def run_goal(objetivo, *, caminho do conjunto de dados, max_attempts=5) -> MetaResultado:
    dados = Caminho(caminho do conjunto de dados)
    visualização = conjunto de dados.visualização(dados)
    m = Máquina()                             # começa no PLANO
    # ---- PLANO: objetivo -> abordagem + verificações de aceitação --------------------
    plano = aguardar _turn(planner_config(),
                       f"OBJECTIVE:\n{objetivo}\n\nDATASET PREVIEW:\n{visualização}", Plano)
    m.para(Estado.GERAR)
    # ---- GERAR: escreva o primeiro roteiro -----------------------------------
    código = aguardar _turn(gerador_config(), _gerar_prompt(objetivo, plano, visualização),
                       Código gerado)
    code_path=sandbox.write_code(código)
    m.para(Estado.EXECUTE)
    tentativas = 0
    enquanto Verdadeiro:
        # ---- EXECUTAR: execute o script atual na sandbox -----------------
        executar=sandbox.run_python(caminho_código, conjunto de dados=dados)
        tentativas += 1
        m.para(Estado.INSPECIONAR)
        # ---- INSPECIONAR: verificações difíceis primeiro, então julgamento do agente ---------------
        it = verificar.avaliar(plano.verificações, correr, sandbox.WORKSPACE)
        inspeção = aguardar _turn(inspetor_config(),
                                 _inspecionar_prompt(plano, correr, o), Inspeção)
        verificado = inspeção.verificado e verificador.all_passed(o)
        se verificado:
            m.para(Estado.VERIFICADO); quebrar        # sucesso final
        se tentativas >=máx_tentativas:
            m.para(Estado.FALHA); quebrar          # falha terminal - orçamento gasto
        # ---- REFINAR: reescreva o script para corrigir o que o INSPECT encontrou -----------
        m.para(Estado.REFINE)
        código = aguardar _turn(refinador_config(),
                           _refine_prompt(código, correr, o, inspeção), Código gerado)
        code_path=sandbox.write_code(código)
        m.para(Estado.EXECUTE)                     # feche o ciclo - voltar para EXECUTAR
    retornar Resultado da meta(
        objetivo = objetivo, estado_final=m.estado.valor,
        verificado =(m.state é State.VERIFIED), tentativas = tentativas, fita=m.tape(),
        relatório = verificador.load_report(sandbox.WORKSPACE) se m.state for State.VERIFIED else {},
        code_path=str(caminho_código), artefatos=run.artifacts,
    )

Leia esse loop while como o circuito de feedback que é:

  • executar=sandbox.run_python(…) é a planta — ela produz uma observação real. it = verificar.avaliar(…) e inspeção são os sensores. código = aguardar _turn(refinador_config(), …) é o controlador. The loop closes when EXECUTE runs the controller's new output. That's not a metaphor; it's literally a discrete feedback controller with an LLM in the actuator.
  • verificado = inspeção.verificado e verificador.all_passed(o) é o intertravamento de segurança. The model's verified flag is necessário, mas não suficiente — as verificações determinísticas recebem um veto rígido por meio do e. Um modelo que alucina o sucesso não pode forçar uma saída VERIFICADA; the arithmetic won't let it.
  • se tentativas >=máx_tentativas: m.para(Estado.FALHA) é o anti-livelock. Um loop autocorretivo deve ser capaz de desistir. Sem esse limite, um objetivo genuinamente impossível queimaria tokens para sempre. Com isso, a execução termina honestamente — FAILED, com sua fita completa — e um humano é paginado com as evidências já reunidas.
  • Cada galho é um guardado m.para(…). The control flow you're reading is the apenas fluxo de controle existe; o modelo nunca escolhe o próximo estado. Isso é o que torna a execução autônoma algo que você pode definir em um cronograma e confiar.

Aponte isso para o sales.csv bagunçado e o primeiro EXECUTE falha exatamente como projetado — ValueError: não foi possível converter string em float: '$1,299.00'. INSPECT lê esse rastreamento, vê reconcilias_total bloqueado, e emite um diagnóstico; REFINE remove o $ e , e repete; o próximo INSPECT detecta um bug mais sutil (uma célula de região em branco vazou como uma chave by_region); REFINE deixa cair; a terceira execução limpa todas as seis verificações. Nenhum humano digitou uma palavra.

6. Etapa 4 — Verificação e finalização do estado

The VERIFIED exit is only meaningful if "verified" significa algo que uma máquina verificou. That's the verifier — the deterministic half of INSPECT. It's real, Python executável, e o agente INSPECT é proibido de anulá-lo.

# loop/verificador.py  (verificações principais)
importar JSON
do caminho de importação pathlib
de .schemas importar AcceptanceCheck, CheckKind, Resultado de verificação, RunResult
RESULT_FILE = "result.json"
TOLERÂNCIA = 0.01                                   # um centavo
avaliação definitiva(cheques, correr, espaço de trabalho) -> lista[Resultado de verificação]:
    resultado = _carregar_resultado(espaço de trabalho)               # analisar o artefato uma vez
    fora = []
    para cheques de check-in:
        se check.kind não for CheckKind.DETERMINISTIC:
            continuar                               # verificações de julgamento pertencem ao INSPECT
        out.append(_executar_verificar(verificar, correr, resultado))
    retornar
def _reconcilia_total(_correr, resultado):               # o cheque que mais importa
    por_região, total = resultado["by_region"], resultado["total_revenue"]
    somado = soma(flutuador(v) para v em by_region.values())
    delta = abdômen(resumido - flutuador(total))
    delta de retorno <= TOLERÂNCIA, f"Σby_region={resumido:.2f} versus total={total:.2f} (D ={delta:.4f})"
def _no_null_region(_correr, resultado):                 # detecta o vazamento da célula em branco
    ruim = [k por k em resultado.get("by_region", {}) se k em (Nenhum, "", "nan", "None")]
    retornar (nada mal), (f"leaked empty region key(é): {ruim}" if bad else "clean")
def _top_is_argmax(_correr, resultado):                  # a reivindicação deve corresponder aos dados
    por = resultado["by_region"]
    argmáx = máximo(por, chave = lambda k: flutuador(por[k]))
    retornar argmax == resultado["top_region"], f"argmax={argmax!R} vs reivindicação={resultado['top_region']!R}"

Três coisas fazem disso um finalização mecanismo e não apenas um linter:

  • As verificações falham ao serem fechadas. Um ID de verificação desconhecido, uma verificação de falha, um artefato ausente — todos resolvidos como Falso, nunca é verdade. O loop só pode sair de VERIFIED por afirmativamente satisfazendo cada predicado. Ambiguidade refina; isso nunca passa. That's the correct default for autonomy: silêncio não é consentimento.
  • A verdade básica supera o modelo. _reconciles_total re-computes the sum from the artifact the script wrote and compares it to the script's own claimed total. Se o código eliminou linhas, os números divergem e a verificação falha - independentemente da confiança com que o modelo narrou o sucesso. Reconciliação não é uma questão de opinião.
  • O artefato verificado é a entrega. Em uma saída VERIFICADA, verificador.load_report() lê o result.json comprovado e o carimba no GoalResult. The run doesn't return uma descrição da análise — ele retorna a análise, já verificado:
{
  "objective": "…revenue per region, região superior, and its share…",
  "final_state": "verified",
  "verified": verdadeiro,
  "attempts": 3,
  "tape": "plan → generate → execute → inspect → refine → execute → inspect → refine → execute → inspect → verified",
  "report": {
    "by_region": {"North": 10749.90, "South": 8552.50, "East": 7212.25, "West": 8631.00},
    "total_revenue": 35145.65,
    "top_region": "North",
    "top_region_share": 0.3059
  }
}

O campo da fita é a prova de finalização: você pode ler toda a vida da execução - três tentativas de EXECUÇÃO, duas autocorreções, uma saída verificada — como uma única string. E porque a máquina só chega a VERIFICADO ao compensar todos os cheques, esse relatório é seguro para entregar a jusante sem um humano verificando novamente. Essa é toda a promessa da engenharia de circuito fechado: not "the agent produced something," but "the agent produced something, and here is the machine-checked proof it's right."

Para executar tudo como uma etapa de construção governada - sem cabeça, isolado em árvore de trabalho, pipeable — essa é a CLI agy no modo /goal:

gol de idade \
    --iterações máximas 5 \
    --success "python main.py exits 0 e workspace/result.json reconcilia:
               soma(por_região) == receita_total (±0,01), top_region == argmax,
               e nenhuma chave de região em branco." \
    --formato de saída json \
    -p "Run \`python main.py\`. Se as verificações de aceitação falharem, inspecionar o
        rastrear e refinar o script de análise até que cada verificação seja aprovada.
        Then commit result.json and open a PR '[meta] verified analysis'." \
  | jq '{estado: .estado_final, verificado, tentativas, fita}'

–sucesso declara a barra; –máximo de iterações limita o orçamento; –formato de saída json torna a execução um estágio de pipe. Cada execução agendada obtém sua própria árvore de trabalho git, para que o loop possa reescrever seus próprios scripts o quanto quiser, sem nunca tocar no seu checkout ao vivo. A CLI e o SDK são duas faces de um mesmo equipamento — protótipo do loop interativamente com /goal, então codifique-o em Python para CI.

7. Conclusão

O agente sob demanda foi um estágio útil - provou que o modelo poderia raciocinar. Mas raciocínio não é engenharia. Engenharia é um laço: produzir, teste, observar, correto, e pare somente quando a saída ultrapassar uma barra que você definiu antecipadamente. Por trinta anos, construímos esse ciclo a partir de pipelines de CI, asserções e novas tentativas. A mudança Antigravidade 2.0 torna possível é colocar um LLM dentro esse loop como gerador e depurador, enquanto mantém o próprio loop determinístico, limitado, e auditável.

Esse é o modelo mental para sair. A máquina estatal não é um modelo em torno da inteligência – ela é a engenharia. O modelo fornece julgamento dentro de cada estado; a máquina fornece o fluxo de controle, a verificação, e a saída verificada. /meta dá a você o objetivo, contrato de prova; saída estruturada faz com que cada transição seja um campo lido em vez de uma estimativa; o caixa de areia permite que o agente execute seu próprio código sem você confiar nele; e o verificador makes "done" um booleano que uma máquina pode defender.

Construa a máquina uma vez. Em seguida, entregue um objetivo e vá embora - e volte para um resultado com um recibo.

Reconhecimento

Os créditos do Google Cloud são fornecidos para este projeto. #AgenticArchitectSprint #Antigravidade

Leitura adicional

blank

Loop Smith: Engenharia de IA de circuito fechado – execução autônoma/de meta para pipelines autocorretivos em… foi publicado originalmente no Google Developer Experts on Medium, onde as pessoas continuam a conversa destacando e respondendo a esta história.

Pesquisar notícias
Notícias
Pesquisar notícias
documentação do tema