BucleSmith: Ingeniería de IA de circuito cerrado: ejecución autónoma/de objetivos para tuberías de autocorrección en… para Windows 7,8,10,11

BucleSmith: Ingeniería de IA de circuito cerrado — Ejecución autónoma/de objetivos para tuberías de autocorrección en antigravedad 2.0

Dejar de avisar al agente paso a paso. Dale un objetivo, una barra para limpiar, y dejar que la máquina de estados escriba, correr, y corrige su propio código hasta que se verifique el resultado.

Esta guía se dirige Antigravedad 2.0 — el lanzamiento de cuatro superficies (aplicación de escritorio, CLI cerebral, SDK antigravedad de Google, y nube empresarial) que comparte un arnés de agente. El SDK es anterior a la versión 1.0.; Los nombres de los símbolos y las banderas CLI a continuación reflejan la API documentada a partir de mediados de 2026.. Trate los patrones como estables y vuelva a verificar las firmas exactas con los documentos actuales antes de realizar el envío..

Tabla de contenido

El problema con el agente bajo demanda
El cambio arquitectónico: Ingeniería de circuito cerrado como máquina de estados.
Paso 1 — Inicializar el proyecto y la máquina de estados (cerebro)
Paso 2 — Definir el objetivo y el modo de ejecución del disparador/meta (SDK)
Paso 3 — Implementar el ciclo de autocorrección
Paso 4 — Verificación y finalización del estado
Conclusión

1. El problema con el agente bajo demanda

La mayor parte de la “ingeniería de IA” actual sigue siendo conversacional. usted solicita; el modelo responde. Notas que la respuesta es incorrecta.; preguntas de nuevo. Pegas un rastreo; se disculpa y lo intenta una vez más. La inteligencia es real — pero tú son el circuito de control. Eres lo que ejecuta el código., lee el error, decide si la salida es lo suficientemente buena, y envía la siguiente instrucción de nuevo. Saca al humano de ese asiento y todo el sistema se detiene..

Eso está bien para una ventana de chat.. Se desmorona en el momento en que quieres que un agente producir un entregable — un conjunto de datos limpio, un informe financiero conciliado, una migración que realmente compila. El verdadero trabajo analítico es iterativo y autorreferencial: escribes un guión, se bloquea en una cadena de moneda, arreglas el análisis, se ejecuta pero los totales no se concilian, arreglas la agregación, y sólo entonces ¿La salida es confiable?. Cada una de esas flechas es una decisión.. Un agente bajo demanda te obliga a suministrarlos todos.

Hay tres fallas específicas ocultas en "simplemente vuelva a solicitarlo":

No hay definición de hecho. Un agente conversacional se detiene cuando se siente finalizado, es decir, cuando produce texto fluido. Fluido no es lo mismo que correcto. "Aquí está su análisis de ingresos!"sobre un guión que silenciosamente eliminó un tercio de las filas se lee exactamente como una respuesta correcta. No hay nada en el bucle que verifique.
El humano es el manejador de errores.. El agente no tiene memoria duradera de lo que estaba tratando de lograr a través de vueltas, por lo que no puede decir si su último intento está más cerca o más lejos de la meta. Mantienes ese estado en tu cabeza.. Eres el bucle while, el intento/excepto, y la afirmación — manualmente, giro tras giro.
Ninguna salida verificada. Porque no existe un criterio de éxito explícito, no hay ningún momento que el sistema pueda señalar y decir "hecho, y aquí está la prueba”. Simplemente... se apaga cuando dejas de escribir..

La solución no es un modelo más inteligente ni un mensaje más largo. es un arquitectura: darle al sistema el objetivo y la definición de hecho, luego déjelo ejecutar su propio bucle de control — generar, ejecutar, inspeccionar, refinar — hasta que la salida borre una barra que pueda verificar por si mismo. Eso es ingeniería de IA de circuito cerrado, y antigravedad 2.0 lo convierte en un modo de ejecución de primera clase: /meta.

Construiremos uno de extremo a extremo. El ejemplo actual es una carga de trabajo que todo equipo de datos reconoce: entregar a un agente un CSV desordenado y un objetivo analítico, y obtener un JSON verificado — sin giros humanos en el medio. El conjunto de datos se defiende a propósito (revenue stored as "$1,299.00", celdas de región en blanco, formatos de fecha mixtos), so the agent's primero El intento falla y puedes ver cómo se corrige solo.. Llamamos a la implementación de referencia. BucleSmith.

2. El cambio arquitectónico: Ingeniería de circuito cerrado como máquina de estados.

Deja de pensar en el agente como un chatbot con desplazamiento hacia atrás. Piensa en ello como un proceso con estados. En cualquier instante, una ejecución autónoma hace exactamente una cosa: planificar, o generando código, o ejecutándolo, o inspeccionar el resultado, o refinar — y la única pregunta interesante es ¿Qué estado viene después?. Esa pregunta tiene un pequeño, conjunto finito de respuestas legales. codificarlos, y tienes un máquina de estados finitos: el mas aburrido, La estructura de control más probada en batalla en toda la ingeniería., apuntado a un LLM.

Aquí está la topología completa.. Lea el ciclo en el medio — ese bucle es el motor de autocorrección.

Determinismo que puedes auditar. El flujo de control vive en una tabla de transición, no en el capricho del modelo. El recorrido solo puede moverse a lo largo de los bordes que dibujaste.. Cada ejecución emite una cinta reproducible — planificar → generar → ejecutar → inspeccionar → refinar → ejecutar → inspeccionar → verificar — de modo que cuando algo sale mal, lees la cinta en lugar de adivinar. El modelo proporciona juicio dentro de un estado; el máquina decide dónde se permite que la carrera vaya a continuación. Esa división es todo el truco..
La autocorrección como ciclo, no es un reintento. El bucle EJECUTAR → INSPECCIONAR → REFINAR → EJECUTAR es un circuito de retroalimentación cerrado. INSPECT observa la actual resultado (código de salida, stderr, si los números se concilian) y emite un diagnóstico; REFINE consume ese diagnóstico y emite un mejor programa; EJECUTAR lo prueba. No hay giro humano en ningún lugar de ese circuito.. The agent isn't "trying again" — it's reading its own error and steering.
Una salida verificada, definido de antemano. La ejecución termina en VERIFICADO sólo cuando cada control de aceptación pasa — y esos cheques se declaran por adelantado, en el estado PLAN, como predicados verificables por máquina (código de salida 0, el artefacto analiza, S(por_región) == total dentro de un centavo). "Done" deja de ser una vibra y se vuelve booleana. Si el presupuesto del intento se agota primero, la ejecución sale FAILED — ruidosamente, con su cinta — en lugar de fingir.

Lo contrario de esto es el bucle abierto: rápido en, enviar mensajes de texto, el ser humano sostiene el estado. Todo lo que aparece a continuación es el paso de abierto a cerrado.. Dos reglas de diseño hacen que funcione en la práctica, y vale la pena indicarlos antes de cualquier código:

Separar el juicio del control. Los LLM son extraordinarios a juicio dentro de un estado ("Este rastreo significa que la columna de moneda es una cadena; banda $ y , before casting") y poco confiable al ser su propio programador. Entonces el gráfico de estado es simple., Python determinista; los agentes nunca deciden ¿Qué estado viene después? — ellos sólo deciden what to do in the state they're in.
Verificar con código, no vibraciones. Las partes de “hecho” que una computadora puede verificar — códigos de salida, esquemas, conciliación — se verifican en código, como verdad fundamental, el modelo no puede anular. El juicio del modelo se superpone a las cosas que el código no puede afirmar., nunca debajo.

vamos a construirlo.

3. Paso 1 — Inicializar el proyecto y la máquina de estados (cerebro)

Instala las dos superficies que usarás.. El CLI (cerebro) es el arnés terminal basado en Go; el SDK (google-antigravedad) es el mismo arnés que una biblioteca de Python. Crea un prototipo del bucle de forma interactiva en la CLI, luego codificarlo en Python.

# 1. Instale la CLI de Antigravity (macos / linux). Aterriza en ~/.local/bin/agy
rizo -fsSL https://antigravity.google/cli/install.sh | intento
# WindowsPowerShell:  https://www.facebook.com/novelasgratis://antigravity.google/cli/install.ps1 | nada
# 2. Confirme que el arnés esté activo y vea a qué modelos puede acceder
modelos cerebrales
# 3. Andamiar el proyecto
mkdir loopsmith && cd loopsmith
python -m venv .venv && fuente .venv/bin/activate
# 4. Instalar el SDK. IMPORTANTE: instalar la rueda publicada, no es una fuente
#    verificar - la rueda envía el binario `localharness` que las unidades SDK.
pip instalar google-antigravity pydantic python-dotenv pandas
# 5. Autenticar. O una clave directa de Géminis…
export GEMINI_API_KEY="your_key_here"
#    …o enrutar a través de Vertex AI con credenciales predeterminadas de la aplicación:
#    inicio de sesión predeterminado de la aplicación gcloud auth   # luego establezca VERTEX=1 en .env

Antes de escribir un solo agente, escribe el máquina de estados. Esta es la inversión que hace legible una carrera autónoma: la topología está definida primero, En aislamiento, sin ningún modelo cerca. los siete estados, las transiciones legales, y un guardia que rechaza cualquier cosa fuera del gráfico:

# bucle/estados.py
desde clases de datos importar clase de datos, campo
desde enumeración importar enumeración
estado de clase(cadena, enumeración):
    PLAN = "plan"          # descomponer el objetivo en un enfoque + cheques
    GENERATE = "generate"  # escribir el primer guión de análisis
    EXECUTE = "execute"    # ejecutar el script en el sandbox
    INSPECT = "inspect"    # juzgar la carrera contra los controles de aceptación
    REFINE = "refine"      # reescriba el script para arreglar lo que encontró INSPECT
    VERIFIED = "verified"  # terminal ÉXITO - todos los controles pasan
    FAILED = "failed"      # terminal FALLA - intento de presupuesto agotado
TERMINAL = conjunto congelado({Estado.VERIFICADO, Estado.FALLADO})
# La tabla de transición ES la arquitectura.. Un movimiento que no figura aquí es un error..
TRANSICIONES = {
    Estado.PLAN:     conjunto congelado({Estado.GENERAR}),
    Estado.GENERAR: conjunto congelado({Estado.EJECUTAR}),
    Estado.EJECUTAR:  conjunto congelado({Estado.INSPECTAR}),
    Estado.INSPECTAR:  conjunto congelado({Estado.VERIFICADO, Estado.REFINAR, Estado.FALLADO}),
    Estado.REFINAR:   conjunto congelado({Estado.EJECUTAR}),
}
clase Transición ilegal(Error de tiempo de ejecución):
    """Raised when the run attempts a move the transition table forbids."""
@clasededatos
máquina de clase:
    estado: Estado = Estado.PLAN
    historia: lista[Estado] = campo(default_factory=lambda: [Estado.PLAN])
    Definitivamente a(ser, siguiente: Estado) -> Estado:
        permitido = TRANSICIONES.get(auto.estado, conjunto congelado())
        si nxt no está permitido:
            plantear transición ilegal(
                f"illegal transition {valor.de.auto.estado!r} → {valor.nxt!r} "
                f"(permitido: {ordenado(s.valor para s en permitido) or 'none - terminal'})"
            )
        self.estado = nxt
        historia personal(siguiente)
        volver a continuación
    @propiedad
    definitivamente hecho(ser) -> booleano:
        devolver auto.estado en TERMINAL
    cinta de definición(ser) -> cadena:
        devolver " → ".join(s.valor para s en self.history)

¿Por qué esto es lo primero que escribes?, no es el último:

TRANSICIONES es el contrato. Note el ciclo: EJECUTAR → INSPECCIONAR → REFINAR → EJECUTAR. Ese único bucle en un diccionario es el motor de autocorrección: la ejecución lo recorre hasta que INSPECT emite VERIFICADO.. And notice what's ausente: no hay ventaja entre REFINAR y GENERAR, sin ventaja fuera de VERIFICADO. Illegal moves aren't handled defensively; they're impossible.
Máquina.a() es el único punto de estrangulamiento. Cada cambio de estado en todo el sistema fluye a través de este único 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. Al paso 4 el gancho de auditoría narra esto en vivo.
No tiene modelo, entonces es comprobable. Puede realizar una prueba unitaria de la topología — “¿puede una ejecución llegar a VERIFICADO sin pasar por INSPECCIONAR??” — sin llamadas API y sin problemas. La parte estocástica (los agentes) y la parte determinista (el grafico) nunca contaminarnos unos a otros.

Interactivamente, Puedes ver cómo se ejecuta la misma máquina antes de escribir Python.. De la vieja TUI, /El objetivo lo coloca en modo de ejecución y transmite las transiciones de estado a medida que ocurren.:

cerebro> /goal "Reconcile revenue by region in ./data/sales.csv into a verified JSON"
◆ planificar → 6 cheques de aceptación declarados
◆ generar → análisis.py (41 pauta)
◆ ejecutar → salir 1  (Error de valor: no se pudo convertir la cadena a flotante: '$1,299.00')
◆ inspeccionar → NO verificado · reconcilias_total bloqueado por falla
◆ refine     → strip '$'/',' antes del elenco
◆ ejecutar → salir 0  · resultado.json escrito
◆ inspeccionar → NO verificado · no_null_region FALLO (región en blanco filtrada como clave)
◆ refinar → eliminar filas de regiones en blanco
◆ ejecutar → salir 0
◆ inspeccionar → VERIFICADO ✓  (6/6 cheques)
◆ hecho → cinta: planificar→generar→ejecutar→inspeccionar→refinar→ejecutar→inspeccionar→refinar→ejecutar→inspeccionar→verificado

Esa es toda la historia en una pantalla.: dos autocorrecciones, ningún giro humano, una salida verificada. Ahora construimos el Python que lo hace reproducible..

4. Paso 2 — Definir el objetivo y el modo de ejecución del disparador/meta (SDK)

/El modo objetivo necesita dos entradas y nada más.: a objetivo de alto nivel y un definicion de hecho. Todo lo demás — cuántas iteraciones, cual guión, cuál solución — el bucle decide por sí mismo.

Hay dos formas legítimas de invocarlo., y un ingeniero senior debe saber ambos.

Patrón A — nativo del arnés /meta. Entregue al arnés el objetivo y un criterio de éxito y deje que él poseer el bucle. El SDK expone run_goal; los planos del arnés, genera, ejecuta, inspecciona, y refina internamente, y devuelve un GoalResult. Este es el código mínimo y el valor predeterminado correcto para el trabajo exploratorio.:

# la forma de una sola línea: el arnés recorre todo el circuito cerrado por usted
del agente de importación google.antigravity, Configuración de objetivo
asíncrono def objetivo_rápido():
    asíncrono con Agent.for_goal(Configuración de objetivo(model="gemini-3-pro")) como agente:
        resultado = esperar agente.run_goal(
            objetivo=(
                "Analyze ./data/sales.csv and report total revenue per region, "
                "the top region, y su participación en los ingresos totales como result.json."
            ),
            éxito =(
                "python exits 0; result.json analiza; suma(por_región) == ingresos_totales "
                "within 0.01; región_superior == argmax(por_región); sin clave de región en blanco."
            ),
            iteraciones_max=5,   # el presupuesto de autocorrección
        )
        imprimir(resultado.estado_final, resultado.verificado, resultado.cinta)

Patrón B — la máquina de estados explícita. cuando necesites determinismo y auditabilidad — lo que siempre hace un proceso de producción — tú mismo manejas el mismo ciclo con las primitivas del SDK, entonces cada transición, cada cheque, y cada refinamiento es tu código. Esto es lo que construye el resto del tutorial., porque ver la maquinaria es el punto.

De cualquier manera, La definición de hecho no es la prosa que esperas que el modelo respete: es una lista de controles de aceptación, y todo el trabajo del primer agente es producirlos. La salida estructurada es lo que lo hace posible.. Cada giro de modelo en LoopSmith está restringido a un esquema Pydantic a través de Response_schema=, entonces la máquina de estados recibe objetos escritos en los que puede ramificarse, no es texto, tiene que analizarse:

# bucle/esquemas.py  (los contratos que fluyen entre estados)
desde enumeración importar enumeración
desde pydantic importar modelo base, Campo
clase tipo cheque(cadena, enumeración):
    DETERMINISTIC = "deterministic"  # aplicado en código (verificador.py)
    JUDGEMENT = "judgement"          # delegado al agente INSPECT
aceptación de claseverificar(son modelos):
    identificación: cadena = campo(description="stable slug, p.ej. 'reconciles_total'")
    descripción: cadena
    amable: Tipo de cheque
plano de clase(son modelos):
    objetivo: cadena
    acercarse: lista[cadena]
    contrato_salida: cadena           # el JSON exacto que debe escribir el script
    cheques: lista[AceptaciónCheque] = Campo(longitud_mín=1)
clase Código Generado(son modelos):    # reutilizado por AMBOS generar y refinar
    Nombre del archivo: str = "analysis.py"
    idioma: str = "python"
    fuente: cadena                    # el contenido completo del archivo, listo para correr
    notas: str = ""                # que cambio vs.. la última versión
inspección de clase(son modelos):       # la decisión de la sucursal, como datos
    verificado: booleano                 # Verdadero si se aprueba CADA verificación de aceptación
    comprobaciones_fallidas: lista[cadena] = Campo(default_factory=lista)
    diagnóstico: cadena                 # causa principal, conectado a tierra en stderr / cheques
    sugerencia_arreglo: str = ""             # cambio concreto que REFINE debería hacer

Dos decisiones de esquema llevan la arquitectura.:

Inspección.verificada es una booleano, entonces la rama es un campo leído. Todo el INSPECCIONAR → {VERIFICADO, REFINAR} la horquilla colapsa si se verifica la inspección.:. Sin expresiones regulares sobre prosa modelo, no "did it say the word aprobado?" heurística. El modelo llena un campo escrito.; la maquina lo lee.
GeneratedCode es compartido por GENERATE y REFINE. Un refinamiento es simplemente la próxima versión del mismo artefacto. Porque ambos estados emiten el mismo esquema., 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. Paso 3 — Implementar el ciclo de autocorrección

Ahora la sustancia: los cuatro agentes, la zona de pruebas que ejecuta su código, y el conductor que camina por la máquina de estados. Cada agente es acuñado de Una fábrica con un juego de barandillas. - el Arnés de agente compartido. Una ejecución autónoma que escribe y ejecuta su propio código es exactamente la ejecución que desea con la correa más corta posible.: política de denegación predeterminada, una línea de auditoría por llamada de herramienta, sin red, sin concha.

# bucle/arnés.py
importar sistema operativo
al escribir import Invocable, Secuencia
desde google.antigravity importar LocalAgentConfig
desde google.antigravity.hooks importar ganchos, política
MODELO = os.getenv("LOOPSMITH_MODEL", "gemini-3-pro")
# Mínimo privilegio, codificado una vez, heredado por cada agente en el bucle.
POLÍTICAS_BASE = [
    política.permitir("view_file"),
    política.permitir("write_file", cuando = lambda a: _dentro_del_espacio de trabajo(a.obtener("path", ""))),
    política.denegar("run_command"),   # los agentes nunca desembolsan - el SANDBOX lo hace
    política.denegar("net_fetch"),     # un bucle autocorrector debe permanecer hermético
    política.denegar("*"),             # denegar por defecto: todo lo que no está permitido se rechaza
]
clase StateAuditHook(ganchos.PostToolCallHook):
    ejecución asíncrona de definición(ser, contexto, datos) -> Ninguno:
        imprimir(f"[auditoría] estado ={getattr(contexto, 'label', '?')} "
              f"tool={getattr(datos, 'name', '?')} bien ={getattr(datos, 'ok', Verdadero)}")
def build_agent_config(*, etiqueta: cadena, instrucciones_del_sistema: cadena,
                       herramientas: Secuencia[invocable] = ()) -> Configuración del agente local:
    kwargs = dict(
        modelo=MODELO, etiqueta = etiqueta, instrucciones_sistema=instrucciones_sistema,
        herramientas=lista(herramientas), políticas=BASE_POLICIES, ganchos =[Gancho de auditoría estatal()],
    )
    si os.getenv("VERTEX") == "1":
        kwargs["vertex"] = Verdadero
    elif os.getenv("GEMINI_API_KEY"):
        kwargs["api_key"] =os.entorno["GEMINI_API_KEY"]
    devolver LocalAgentConfig(**kwargs)
def _dentro_del_espacio de trabajo(camino: cadena) -> booleano:
    return "workspace" en os.path.normpath(camino).dividir(os.sep)

La decisión crítica aquí es política.denegar("run_command") — los propios agentes obtienen sin concha. Esto parece paradójico para un bucle cuyo trabajo consiste exclusivamente en ejecutar código., hasta que veas dónde ocurre realmente la ejecución: en un separado, caja de arena determinista el orquestador controla, not inside a model's tool call. el modelo escribe el programa; un subproceso gobernado corre él. El modelo nunca puede instalarse ni salirse., porque nunca le entregaron las llaves.

Los cuatro agentes — juicio, sin flujo de control

El agente de cada estado es una configuración de responsabilidad única. Ninguno de ellos tiene lógica de orquestación.; Ninguno de ellos sabe que está en un bucle.. Razonan sobre lo que les entregan y emiten un objeto escrito — eso es todo.

# bucle/agentes.py  (abreviado: indicaciones completas del sistema en el repositorio)
desde .harness importar build_agent_config
def planificador_config():     # PLAN: objetivo -> acercarse + cheques verificables por máquina
    devolver build_agent_config(label="plan", instrucciones_sistema=(
        "You are a senior data analyst PLANNER. Dado un objetivo y una vista previa "
        "of a messy CSV, producir un enfoque, un contrato de salida exacto para el "
        "JSON the script must write, y una lista de VERIFICACIONES DE ACEPTACIÓN. Prefiero estos "
        "deterministic ids when they apply: salidas_limpias, artefacto_existe, "
        "schema_ok, no_null_region, concilia_total, top_is_argmax. la carrera es "
        "'verified' sólo cuando CADA control pasa - do not under-specify."))
def generador_config():   # GENERAR: escribe el primer guión
    devolver build_agent_config(label="generate", instrucciones_sistema=(
        "You are a Python DATA ENGINEER. Escribe un COMPLETO, guión autónomo "
        "that reads os.environ['LOOPSMITH_DATASET'], realiza lo planeado "
        "analysis, and writes 'result.json'. Utilice sólo la biblioteca estándar y "
        "pandas. Sin red. Return the full file contents."))
def inspector_config():   # INSPECCIONAR: verificado vs.. refinar, basado en controles estrictos
    devolver build_agent_config(label="inspect", instrucciones_sistema=(
        "You are a rigorous VERIFICATION agent. Obtienes los cheques de aceptación., el "
        "sandbox RunResult, y los resultados de los controles DETERMINISTAS ya "
        "computed in code. NO puede marcar como verificado si hay alguna verificación determinista. "
        "failed - Esas son la verdad fundamental.. Cuando no está verificado, escribir un diagnostico "
        "rooted in the real stderr/failed checks and a concrete fix_hint."))
def refiner_config():     # REFINAR: reescribir para corregir exactamente lo que diagnosticó INSPECT
    devolver build_agent_config(label="refine", instrucciones_sistema=(
        "You are a DEBUGGING engineer. Dado el guión actual, el resultado de ejecución "
        "showing how it failed, y un diagnostico + sugerencia_arreglo, producir el SIGUIENTE completo "
        "version that fixes the identified problem while preserving what worked."))

La asimetría es el diseño.. El planificador convierte un objetivo difuso en una barra comprobable. El inspector tiene explícitamente prohibido anular los controles deterministas — su juicio se sienta encima de verdad fundamental, nunca debajo de él. El refinador recibe un diagnostico, no es un vago “inténtalo de nuevo," por lo que su reescritura está dirigida. Cada uno tiene una pequeña ventana de contexto y un trabajo..

La zona de pruebas : donde realmente se ejecuta el código generado

Este módulo es real, Python ejecutable — sin SDK, ningún modelo. Es el estado EJECUTAR, y es el límite de seguridad. Ejecuta el script del agente como un subproceso., en un directorio de trabajo confinado, con un tiempo de espera de reloj de pared y un entorno despojado, e informa exactamente lo que pasó.

# bucle/sandbox.py  (centro)
subproceso de importación, sistema, tiempo
desde pathlib importar ruta
desde .schemas importar código generado, EjecutarResultado
ESPACIO DE TRABAJO = Ruta(__archivo__).resolver().padre.padre / "workspace"
def escribir_código(código: Código generado) -> Camino:
    ESPACIO DE TRABAJO.mkdir(padres=verdadero, exist_ok=Verdadero)
    ruta = ESPACIO DE TRABAJO / código.nombre de archivo
    ruta.write_text(código.fuente, encoding="utf-8")
    camino de regreso
definición run_python(camino: Camino, *, conjunto de datos: Camino, tiempo de espera_s: flotar = 30.0) -> EjecutarResultado:
    antes = _instantánea()                      # qué archivos existen antes de la ejecución
    iniciado = tiempo.monotónico()
    intentar:
        proc = subproceso.ejecutar(
            [sys.ejecutable, cadena(camino)],
            cwd=ESPACIO DE TRABAJO,                     # limitar el directorio de trabajo
            entorno={"LOOPSMITH_DATASET": cadena(conjunto de datos),   # ambiente mínimo - no se filtran secretos
                 "PATH": _ruta_segura(), "PYTHONUNBUFFERED": "1"},
            capture_output=Verdadero, texto=Verdadero, tiempo de espera = tiempo de espera_s,
        )
        código_salida, salida estándar, stderr = proc.returncode, proc.stdout, proc.stderr
        timed_out = Falso
    excepto subproceso.TimeoutExpired como e:
        código_salida, salida estándar = 124, (e.stdout or "")
        stderr = (e.stderr or "") + f"\n[salvadera] asesinado después {tiempo de espera_s}tiempo de espera"
        timed_out = Verdadero
    artefactos = ordenados(p.name para p en _snapshot() - antes)   # lo que creó la carrera
    devolver resultado de ejecución(código_salida=código_salida, salida estándar=_clip(salida estándar), stderr=_clip(stderr),
                     duración_s=ronda(tiempo.monotónico() - comenzó, 3),
                     artefactos = artefactos, timed_out=timed_out)

Línea por línea, este es el contrato EJECUTAR completo:

subproceso.ejecutar(…, cwd=ESPACIO DE TRABAJO, tiempo de espera = tiempo de espera_s) — el código generado se ejecuta outside the agent's process, en su propio directorio, y no puede ejecutarse más tiempo que el tiempo de espera. Un bucle infinito que escribió el modelo obtiene SIGKILL, superficies como exit_code=124, y se convierte en simplemente otra entrada de INSPECCIÓN, no en una tubería colgada.
entorno={…} — un entorno construido a mano. La ruta del conjunto de datos llega a través de LOOPSMITH_DATASET; tus claves API, credenciales de la nube, y los alias de shell sí no. El guión ve exactamente lo que necesita y nada más..
_instantánea() antes y después — la zona de pruebas diferencia el espacio de trabajo para informar con precisión qué artefactos produjo la ejecución. That's how the loop sabe resultado.json fue escrito, rather than trusting the model's stdout claim that it was.
El RunResult es la señal de retroalimentación. código de salida, salida estándar, stderr, artefactos — la cruda verdad de lo que sucedió, entregado directamente a INSPECCIONAR. El bucle se corrige contra comportamiento observado, never against the model's hopes.

El conductor — caminando la máquina

Aquí está el circuito cerrado., de arriba a abajo. Esta es la forma explícita de /objetivo: puedes rastrear cada transición, y cada m.a(…) es un vigilado, movimiento registrado.

# bucle/orquestador.py
desde pathlib importar ruta
del agente de importación google.antigravity
de . importar conjunto de datos, salvadera, verificador
desde .agents importar planificador_config, generador_config, inspector_config, refinador_config
del plan de importación .schemas, Código generado, Inspección, ObjetivoResultado
de la máquina de importación .states, Estado
asíncrono def _turn(configuración, inmediato, esquema):
    """One isolated agent, un turno escrito, then tear down."""
    asíncrono con el agente(configuración) como agente:
        resp = esperar agente.chat(inmediato, respuesta_esquema=esquema)
        regreso en espera resp.analizado()            # una instancia de Pydantic validada
asíncrono def run_goal(objetivo, *, ruta_conjunto_datos, max_intentos=5) -> ObjetivoResultado:
    datos = ruta(ruta_conjunto_datos)
    vista previa = conjunto de datos.vista previa(datos)
    m = máquina()                             # comienza en PLAN
    # ---- PLAN: objetivo -> acercarse + controles de aceptación --------------------
    plan = esperar _turno(planificador_config(),
                       f"OBJECTIVE:\norte{objetivo}\n\nDATASET PREVIEW:\norte{avance}", Plan)
    m.a(Estado.GENERAR)
    # ---- GENERAR: escribe el primer guión -----------------------------------
    código = esperar _turn(generador_config(), _generar_prompt(objetivo, plan, avance),
                       Código generado)
    code_path = sandbox.write_code(código)
    m.a(Estado.EJECUTAR)
    intentos = 0
    mientras que verdadero:
        # ---- EJECUTAR: ejecutar el script actual en el sandbox -----------------
        ejecutar = sandbox.run_python(ruta_código, conjunto de datos = datos)
        intentos += 1
        m.a(Estado.INSPECTAR)
        # ---- INSPECCIONAR: controles rigurosos primero, luego juicio del agente ---------------
        it = verificar.evaluar(plan.cheques, correr, sandbox.ESPACIO DE TRABAJO)
        inspección = esperar _turn(inspector_config(),
                                 _inspeccionar_prompt(plan, correr, el), Inspección)
        verificado = inspección.verificado y verificador.todo_aprobado(el)
        si se verifica:
            m.a(Estado.VERIFICADO); romper        # éxito terminal
        si intenta >= max_intentos:
            m.a(Estado.FALLADO); romper          # falla terminal - presupuesto gastado
        # ---- REFINAR: reescriba el script para arreglar lo que encontró INSPECT -----------
        m.a(Estado.REFINAR)
        código = esperar _turn(refinador_config(),
                           _refinar_prompt(código, correr, el, inspección), Código generado)
        code_path = sandbox.write_code(código)
        m.a(Estado.EJECUTAR)                     # cerrar el ciclo - volver a EJECUTAR
    devolver ResultadoObjetivo(
        objetivo = objetivo, estado_final=m.estado.valor,
        verificado =(m.state es Estado.VERIFICADO), intentos = intentos, cinta = m.cinta(),
        informe=verificador.load_report(sandbox.ESPACIO DE TRABAJO) si m.state es State.VERIFIED más {},
        ruta_código=cadena(ruta_código), artefactos=ejecutar.artefactos,
    )

Lea ese bucle while como el circuito de retroalimentación que es.:

ejecutar = sandbox.run_python(…) es la planta — produce una observación real. it = verificar.evaluar(…) y inspección son los sensores. código = esperar _turn(refinador_config(), …) es el 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 = inspección.verificado y verificador.todo_aprobado(el) es el bloqueo de seguridad. The model's verified flag is necesario pero no suficiente — los controles deterministas obtienen un duro veto a través del y. Un modelo que alucina el éxito no puede forzar una salida VERIFICADA; the arithmetic won't let it.
si intenta >= max_intentos: m.a(Estado.FALLADO) es el anti-livelock. Un bucle autocorrector debe ser capaz de rendirse. Sin este límite, un objetivo realmente imposible quemaría fichas para siempre. con eso, la ejecución termina honestamente — FAILED, con su cinta completa — y un humano es localizado con la evidencia ya reunida.
Cada sucursal está vigilada. m.a(…). The control flow you're reading is the solo control de flujo hay; el modelo nunca elige el siguiente estado. Eso es lo que hace que un funcionamiento autónomo sea algo que puedes programar y en el que puedes confiar..

Apunte esto al desordenado sales.csv y la primera EJECUCIÓN falla exactamente como se diseñó — ValueError: no se pudo convertir la cadena a flotante: '$1,299.00'. INSPECT lee ese rastreo, ve reconcilias_total bloqueado, y emite un diagnóstico; REFINE elimina el $ y , y se vuelve a ejecutar; el siguiente INSPECT detecta un error más sutil (una celda de región en blanco filtrada como clave by_region); REFINE lo deja caer; la tercera ejecución borra los seis controles. Ningún humano escribió una palabra.

6. Paso 4 — Verificación y finalización del estado

The VERIFIED exit is only meaningful if "verified" significa algo que una máquina revisó. That's the verifier — the deterministic half of INSPECT. It's real, Python ejecutable, y el agente INSPECT es prohibido anularlo.

# bucle/verificador.py  (controles básicos)
importar json
desde pathlib importar ruta
desde .schemas importar AcceptanceCheck, Tipo de cheque, Verificar resultado, EjecutarResultado
RESULT_FILE = "result.json"
TOLERANCIA = 0.01                                   # un centavo
definitivamente evaluar(cheques, correr, espacio de trabajo) -> lista[Verificar resultado]:
    resultado = _cargar_resultado(espacio de trabajo)               # analizar el artefacto una vez
    fuera = []
    para cheques de facturación:
        si check.kind no es CheckKind.DETERMINISTIC:
            continuar                               # Los controles de juicio pertenecen a INSPECT.
        fuera.añadir(_run_check(controlar, correr, resultado))
    regresar
def _reconcilias_total(_correr, resultado):               # el cheque que más importa
    por_región, total = resultado["by_region"], resultado["total_revenue"]
    sumado = suma(flotar(v) para v en by_region.values())
    delta = abdominales(resumido - flotar(total))
    delta de retorno <= TOLERANCIA, f"Σby_region={resumido:.2F} frente a total={total:.2F} (D={delta:.4F})"
def _no_null_region(_correr, resultado):                 # detecta la fuga de celda en blanco
    malo = [k para k en result.get("by_region", {}) si k en (Ninguno, "", "nan", "None")]
    devolver (nada mal), (f"leaked empty region key(s): {malo}" if bad else "clean")
def _top_is_argmax(_correr, resultado):                  # el reclamo debe coincidir con los datos
    por = resultado["by_region"]
    argmáx = máximo(por, clave = lambda k: flotar(por[k]))
    devolver argmax == resultado["top_region"], f"argmax={argmax!r} vs reclamo ={resultado['top_region']!r}"

Tres cosas hacen de este un finalización Mecanismo y no solo un linter.:

Los cheques fallan cerrados. Una identificación de verificación desconocida, un cheque fallido, un artefacto faltante — todo se resuelve en Falso, nunca es cierto. El bucle sólo puede salir VERIFICADO por afirmativamente satisfaciendo cada predicado. La ambigüedad refina; nunca pasa. That's the correct default for autonomy: el silencio no es consentimiento.
La verdad sobre el terreno supera al modelo. _reconciles_total re-computes the sum from the artifact the script wrote and compares it to the script's own claimed total. Si el código elimina filas, los números divergen y la verificación falla — independientemente de la confianza con la que el modelo narró el éxito. La reconciliación no es una cuestión de opinión..
El artefacto verificado es el entregable.. En una salida VERIFICADA, verificador.load_report() vuelve a leer el result.json probado y lo estampa en GoalResult. The run doesn't return una descripción del análisis — vuelve el análisis, ya comprobado:

{
  "objective": "…revenue per region, región superior, and its share…",
  "final_state": "verified",
  "verified": verdadero,
  "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
  }
}

El campo de la cinta es la prueba de finalización.: puedes leer toda la vida de la ejecución — tres intentos de EJECUCIÓN, dos autocorrecciones, una salida verificada — como una sola cadena. Y porque la máquina sólo llega a VERIFICADO al borrar cada cheque, ese informe es seguro para entregar posteriormente sin que un humano lo vuelva a revisar. Ésa es toda la promesa de la ingeniería de circuito cerrado.: not "the agent produced something," but "the agent produced something, and here is the machine-checked proof it's right."

Para ejecutar todo como un paso de construcción gobernado — sin cabeza, aislado del árbol de trabajo, pipeable — esa es la CLI antigua en modo /goal:

gol agy \
    --iteraciones máximas 5 \
    --success "python main.py exits 0 y workspace/result.json concilia:
               suma(por_región) == ingresos_totales (±0,01), región_superior == argmax,
               y sin clave de región en blanco." \
    --formato de salida json \
    -p "Run \`python main.py\`. Si los controles de aceptación fallan, inspeccionar el
        rastrear y refinar el script de análisis hasta que pase cada verificación.
        Then commit result.json and open a PR '[meta] verified analysis'." \
  | jq '{estado: .estado_final, verificado, intentos, cinta}'

–éxito declara la barra; –iteraciones máximas limita el presupuesto; –El formato de salida json hace que la ejecución sea una etapa de tubería.. Cada ejecución programada tiene su propio árbol de trabajo git, para que el bucle pueda reescribir sus propios scripts todo lo que quiera sin siquiera tocar su pago en vivo. La CLI y el SDK son dos caras de un arnés : prototipo del bucle de forma interactiva con /goal, luego codificarlo en Python para CI.

7. Conclusión

El agente bajo demanda fue una etapa útil: demostró que el modelo podía razonar. Pero el razonamiento no es ingeniería.. La ingeniería es una bucle: producir, prueba, observar, correcto, y se detiene solo cuando la salida borra una barra que usted configuró de antemano. Durante treinta años hemos construido ese bucle a partir de canalizaciones de CI, afirmaciones y reintentos.. El cambio antigravedad 2.0 lo que hace posible es poner un LLM adentro ese bucle como generador y depurador, manteniendo el bucle en sí determinista, encerrado, y auditable.

Ese es el modelo mental con el que partir. La máquina estatal no es un modelo repetitivo en torno a la inteligencia — es la ingenieria. El modelo proporciona juicio dentro de cada estado.; la máquina suministra el flujo de control, la verificación, y la salida verificada. /meta te da el objetivo, contrato de prueba; salida estructurada hace que cada transición sea un campo leído en lugar de una suposición; el salvadera permite que el agente ejecute su propio código sin que usted confíe en él; y el verificador makes "done" un booleano que una máquina puede defender.

Construya la máquina una vez. Luego, entréguele un objetivo y aléjese — y regrese a un resultado con un recibo..

Reconocimiento

Se proporcionan créditos de Google Cloud para este proyecto.. #AgenticArchitectSprint #Antigravedad

Lectura adicional

SDK antigravedad de Google (Pitón): https://github.com/google-antigravity/antigravity-sdk-python
CLI antigravedad (cerebro): https://github.com/google-antigravity/antigravity-cli
/modo de ejecución de objetivos: https://antigravity.google/docs/goal-mode
Tutorial complementario — Despacho de múltiples agentes (subagentes concurrentes): la contraparte de despliegue de este bucle de avance de abanico. BucleSmith: Ingeniería de IA de circuito cerrado — Ejecución autónoma/de objetivos para tuberías de autocorrección en antigravedad 2.0

BucleSmith: Ingeniería de IA de circuito cerrado: ejecución autónoma/de objetivos para canalizaciones de autocorrección en… se publicó originalmente en Google Developer Experts en Medium, donde las personas continúan la conversación resaltando y respondiendo a esta historia.