Equipo multiagente autoreparable en LangGraph: Arquitecto, Probador, Depurador

La IA está presente en casi todas las capas del trabajo con software, desde la generación de código hasta el análisis de datos y el razonamiento en múltiples pasos. Sin embargo, cualquiera que le pida a un modelo de lenguaje grande que haga algo complejo se topa con el mismo problema. Una respuesta de un solo intento puede parecer correcta a primera vista, pero falla al enfrentarse a los requisitos reales y a algunos casos límite. La solución habitual es un proceso manual: copiar el registro de errores, pegarlo en el chat, solicitar una corrección y repetir hasta que finalmente funcione.

El enfoque opuesto también tiene sus problemas. Entregar a un agente monolítico docenas de herramientas y una larga lista de instrucciones tiende a aumentar la tasa de errores, ya que su contexto se contamina y las características de las herramientas comienzan a mezclarse. La calidad de la salida disminuye porque el agente no puede seguir todas esas instrucciones al mismo tiempo.

El diseño presentado aborda ambos problemas mediante una máquina de estados multiagente modular y jerárquica, escrita en Python con LangGraph. El ejemplo en funcionamiento es un equipo de ingeniería de software autoreparable compuesto por cinco roles: un Arquitecto, un Desarrollador, un entorno de pruebas Pytest basado en subprocesos, un Depurador y un Revisor. Este patrón puede aplicarse a muchos otros flujos de trabajo de ingeniería de IA.

Dividir una tarea compleja entre subagentes especializados, asignar a cada uno la cantidad adecuada de capacidad de razonamiento, ejecutar los resultados en un entorno controlado y añadir un ciclo de retroalimentación constante permite al equipo probar su propio trabajo, detectar problemas y repararlos antes de entregar un resultado verificado. Primero, un vistazo a la arquitectura.

Dos principios guían el sistema.

El primero es el estratificado modelo heterogéneo con niveles de razonamiento adaptativos. Las funciones difieren en dificultad, por lo que no deberían utilizar el mismo modelo ni el mismo presupuesto de cálculo:

• El Arquitecto funciona con gemini-pro-latest y un nivel de razonamiento alto. Realiza las tareas que requieren más capacidad de razonamiento, como establecer los límites del problema, anticipar condiciones de carrera y diseñar conjuntos estrictos de pruebas pytest.
• El Desarrollador funciona con gemini-3.7-flash y un nivel de razonamiento medio. Es una configuración equilibrada para una síntesis rápida de código que sigue las especificaciones explícitas del Arquitecto.
• El Ejecutor de pruebas es un subproceso determinista en Python que ejecuta pytest localmente en aproximadamente 15 milisegundos y no consume ningún token de la API del LLM.

El depurador utiliza gemini-3.7-flash con un nivel de razonamiento alto, ya que para rastrear las trazas de pila hasta la línea con error es necesario realizar un análisis causal de la causa raíz.

El revisor emplea gemini-flash-lite-latest con un nivel de razonamiento bajo. Revisar el estilo y redactar el resumen final para el usuario solo requiere un razonamiento sencillo.

El segundo principio es el alcance estricto de las herramientas, que consiste en aplicar el principio de menor privilegio a los agentes. Ningún subagente tiene acceso ilimitado al terminal ni al sistema de archivos; cada uno recibe únicamente las entradas y herramientas que su función exige. El desarrollador no puede ejecutar comandos de shell arbitrarios, y el ejecutor de pruebas simplemente lleva a cabo las pruebas e informa lo que realmente ocurrió, sin pedirle a un LLM que interprete o invente el resultado. Un alcance restringido mantiene a cada agente enfocado y reduce las alucinaciones.

Los conceptos de LangGraph involucrados

LangGraph funciona como una pizarra blanca compartida. Cada miembro del equipo, es decir, cada subagente, realiza su tarea asignada y escribe el resultado en ese mismo espacio compartido, de modo que todos puedan ver el contexto que el equipo ha construido hasta el momento. LangGraph es el orquestador: decide qué agente se ejecutará a continuación y transmite el estado relevante.

Hay cuatro ideas importantes aquí:

1. El estado, un TypedDict, es la pizarra blanca. Todos los agentes leen y escriben en un diccionario de Python llamado AgentTeamState, que contiene la solicitud del usuario, la especificación, las pruebas unitarias, el borrador actual del código, los resultados de las pruebas y el recuento de rondas de depuración.
2. Los nodos son los miembros del equipo. Un nodo es simplemente una función de Python que recibe el estado actual, realiza su tarea (llamando a un LLM, ejecutando pruebas) y devuelve actualizaciones que se fusionan nuevamente en el estado compartido.

Los bordes llevan la lógica del flujo de trabajo. Un borde directo representa una transición fija, como pasar siempre del Arquitecto al Desarrollador. Un borde condicional decide según el estado actual; por ejemplo, redirigir al Revisor cuando todas las pruebas pasan y al Depurador cuando alguna falla.

El grafo compilado es lo que se obtiene después de registrar nodos y bordes con StateGraph y compilarlo. Se puede ejecutar, y sus eventos pueden transmitirse a medida que avanza el flujo de trabajo.

Configuración del proyecto

Comience con un directorio nuevo y un entorno virtual. Estas órdenes de terminal crean ambos y activan el entorno:

mkdir langgraph-self-healing-team
cd langgraph-self-healing-team
python3 -m venv .venv
source .venv/bin/activate

Un archivo requirements.txt enumera a continuación las dependencias:

langgraph>=0.2.20
langchain-google-genai>=2.0.0
langchain-openai>=1.6.0
langchain-core>=0.3.0
pydantic>=2.7.0
pytest>=8.0.0
rich>=13.7.0
python-dotenv>=1.0.1

Instalarlos en el entorno virtual se hace con un solo comando:

pip install -r requirements.txt

Cada paquete tiene una función específica:

• langgraph orquesta el flujo de trabajo multiagente, gestionando el estado, las transferencias entre nodos y los bucles de retroalimentación cíclicos.
• langchain-google-genai se conecta a los modelos de Google Gemini y muestra sus niveles de razonamiento.
• langchain-openai es opcional. Agrega soporte para OpenRouter u OpenAI, de modo que se pueden utilizar modelos como GPT 5.6 o Deepseek v4 Pro si no es deseable depender únicamente de Gemini.
• pytest es el ejecutor de pruebas dentro del entorno aislado de subprocesos local. Ejecuta el conjunto de pruebas generado y devuelve resultados reales.
• rich formatea la salida de la terminal con colores, tablas y resaltado sintáctico, lo que facilita seguir y depurar el flujo de trabajo del agente.
• python-dotenv carga automáticamente las claves de API y los parámetros del modelo desde el archivo .env.

El archivo .env en la raíz del proyecto contiene credenciales y configuración del modelo:

# API Keys
GEMINI_API_KEY="your_api_key_here"
# OPENROUTER_API_KEY=""
# OPENAI_API_KEY=""

# Model Tier Selection & Adaptive Thinking Levels (low, medium, high)
ARCHITECT_MODEL="gemini-pro-latest"
ARCHITECT_THINKING_LEVEL="high"

CODER_MODEL="gemini-3.7-flash"
CODER_THINKING_LEVEL="medium"

DEBUGGER_MODEL="gemini-3.7-flash"
DEBUGGER_THINKING_LEVEL="high"

REVIEWER_MODEL="gemini-flash-lite-latest"
REVIEWER_THINKING_LEVEL="low"

# Maximum Self-Healing Retry Loops
MAX_DEBUG_ITERATIONS=3

Una vez listo el entorno, la aplicación puede ser construida poco a poco.

Paso 1: el estado compartido en src/state.py

Antes que nada, es necesario definir el esquema para los datos que se transmiten entre los agentes. Crea un directorio src con un archivo state.py dentro de él.

La primera definición es un modelo TestResult, escrito con Pydantic. Representa el resultado estructurado de una ejecución de pytest en el entorno aislado:

from typing import TypedDict, List, Dict, Any, Optional
from pydantic import BaseModel, Field

class TestResult(BaseModel):
    passed: bool = Field(description="Whether all tests in the test suite passed")
    exit_code: int = Field(description="Exit code of the pytest execution process")
    output: str = Field(description="Captured stdout and stderr from pytest")
    passed_count: int = Field(default=0, description="Number of passed tests")
    failed_count: int = Field(default=0, description="Number of failed tests")
    error_summary: Optional[str] = Field(default=None, description="Concise error traceback summary")

TestResult registra si el conjunto de pruebas se completó con éxito, el código de salida del proceso, cuántas pruebas dieron resultado positivo y negativo, así como la salida bruta de las pruebas. También incluye un campo error_summary con solo la parte relevante del seguimiento de errores, que el depurador utiliza para localizar y corregir la falla.

A continuación se encuentra el campo principal AgentTeamState, representado como un TypedDict. Este funciona como una pizarra que recorre el grafo: cada nodo lee lo que necesita, realiza su tarea y actualiza los campos de los nodos siguientes de los cuales depende.

class AgentTeamState(TypedDict):
    """
    The shared whiteboard state passed between all nodes in the LangGraph team.
    """
    user_request: str
    specs: str
    tests: str
    code: str
    test_results: Dict[str, Any]
    iteration: int
    max_iterations: int
    debug_history: List[str]
    review_summary: str
    status: str

Cada campo tiene una función en el flujo de trabajo:

• user_request contiene el problema original presentado por el usuario, de modo que todos los agentes pueden ver la intención inicial durante todo el proceso.
• specs y tests provienen del Arquitecto. El Desarrollador implementa las soluciones según las especificaciones, y el Ejecutor de pruebas ejecuta las pruebas generadas contra el código.
• code representa la implementación actual en Python. El Desarrollador la completa primero, y el Depurador puede reemplazarla después de un intento fallido.
• test_results almacena el resultado estructurado obtenido del entorno aislado, lo cual determina cuál de los revisores o el depurador se encargará del siguiente paso.
• iteration y max_iterations controlan el ciclo de autoreparación. Impiden que el depurador intente solucionarlo indefinidamente y establecen un límite de seguridad, como tres intentos.

debug_history guarda la causa raíz y la solución de cada ronda de depuración, así como un registro de los cambios en el código.

review_summary recibe el informe final del revisor una vez que el código ha superado la validación.

status muestra en qué fase se encuentra actualmente el flujo de trabajo: generando código, ejecutando pruebas, depurando o listo para revisión.

Paso 2: niveles de modelo y de razonamiento en src/config.py

Una fábrica unificada de LLM asigna el modelo adecuado y el nivel de razonamiento (bajo, medio o alto) a cada rol de subagente. El archivo src/config.py comienza con un diccionario de configuración por defecto:

import os
from typing import Dict, Any, Optional
from dotenv import load_dotenv
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_openai import ChatOpenAI

load_dotenv()

# Role Configurations & Adaptive Thinking Levels
DEFAULT_CONFIG: Dict[str, Dict[str, Any]] = {
    "architect": {
        "model": os.getenv("ARCHITECT_MODEL", "gemini-pro-latest"),
        "thinking_level": os.getenv("ARCHITECT_THINKING_LEVEL", "high"),
        "temperature": 0.2,
    },
    "coder": {
        "model": os.getenv("CODER_MODEL", "gemini-3.7-flash"),
        "thinking_level": os.getenv("CODER_THINKING_LEVEL", "medium"),
        "temperature": 0.2,
    },
    "debugger": {
        "model": os.getenv("DEBUGGER_MODEL", "gemini-3.7-flash"),
        "thinking_level": os.getenv("DEBUGGER_THINKING_LEVEL", "high"),
        "temperature": 0.1,
    },
    "reviewer": {
        "model": os.getenv("REVIEWER_MODEL", "gemini-flash-lite-latest"),
        "thinking_level": os.getenv("REVIEWER_THINKING_LEVEL", "low"),
        "temperature": 0.2,
    },
}

MAX_DEBUG_ITERATIONS = int(os.getenv("MAX_DEBUG_ITERATIONS", "3"))

Cada rol cuenta con un modelo y un nivel de razonamiento que se ajustan a la dificultad de su tarea, lo cual equilibra la capacidad de razonamiento, la velocidad y el costo sin afectar la calidad general:

• architect y debugger tienen un nivel establecido en alto, ya que el diseño del sistema y el diagnóstico de trazas necesitan una deducción profunda y en múltiples pasos.
• coder tiene un nivel de medio. Solo necesita seguir el plan elaborado por el Arquitecto, por lo que una síntesis equilibrada, rápida y fundamentada es suficiente.
• reviewer tiene un nivel de bajo para ahorrar tokens, ya que lo único que hace es escribir un resumen después de revisar los resultados.

El mismo archivo también contiene la función de fábrica get_llm():

def get_llm(
    role: str,
    model_override: Optional[str] = None,
    thinking_level_override: Optional[str] = None,
) -> BaseChatModel:
    """
    Unified LLM Factory. Returns an initialized LangChain ChatModel configured
    specifically for the sub-agent role with adaptive thinking levels.
    """
    role_conf = DEFAULT_CONFIG.get(role, DEFAULT_CONFIG["coder"])
    model_name = model_override or role_conf["model"]
    thinking_level = thinking_level_override or role_conf["thinking_level"]
    temp = role_conf["temperature"]

    openrouter_key = os.getenv("OPENROUTER_API_KEY")
    openai_key = os.getenv("OPENAI_API_KEY")
    gemini_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")

    # 1. OpenRouter Provider (Handles open models like deepseek/deepseek-r1:free)
    if openrouter_key and ("/" in model_name or not gemini_key):
        return ChatOpenAI(
            model=model_name,
            api_key=openrouter_key,
            base_url="https://openrouter.ai/api/v1",
            temperature=temp,
        )

    # 2. Google Gemini Provider
    if "gemini" in model_name.lower() or gemini_key:
        kwargs: Dict[str, Any] = {
            "model": model_name,
            "temperature": temp,
        }
        if gemini_key:
            kwargs["google_api_key"] = gemini_key

        if thinking_level and thinking_level.lower() in ["low", "medium", "high"]:
            kwargs["thinking_level"] = thinking_level.lower()

        return ChatGoogleGenerativeAI(**kwargs)

    # 3. Direct OpenAI Provider Fallback
    if openai_key:
        return ChatOpenAI(
            model=model_name,
            api_key=openai_key,
            temperature=temp,
        )

    return ChatGoogleGenerativeAI(model=model_name, temperature=temp)

Dos factores hacen que esta función sea útil. Transmite directamente el thinking_level (low, medium, high) a ChatGoogleGenerativeAI, lo que permite ajustar el esfuerzo de razonamiento según cada rol. Además, todos los clientes que devuelve implementan BaseChatModel de LangChain, por lo que los nodos del grafo llaman a .invoke() de la misma manera, independientemente del proveedor que esté detrás de cada rol.

Paso 3: un entorno de pruebas determinista con pytest en src/sandbox.py

Un error frecuente en el diseño de agentes es pedirle a un LLM que verifique, en su propio proceso interno, el código que acaba de escribir. Un modelo puede predecir con confianza que el código funciona correctamente cuando en realidad contiene errores, ya que los modelos suelen generar ilusiones de que su propia salida es correcta. Incluso cuando detectan algunos errores, no hay garantía de que hayan encontrado todos.

Una ruta más fiable consiste en dejar que el código demuestre su eficacia. En lugar de confiar en el juicio del modelo, un nodo de ejecución local escribe el código generado y las pruebas en un directorio temporal y ejecuta pytest en un subproceso aislado de Python. El resultado es una evaluación objetiva de éxito o fracaso que puede integrarse directamente en el flujo de trabajo.

Cree sandbox.py en la carpeta src. Comienza con una función auxiliar, _extract_error_summary(), que extrae los detalles más relevantes del fallo del resultado bruto de pytest para que el depurador muestre solo la parte útil del rastro de errores y no todo el contenido:

import os
import re
import sys
import tempfile
import subprocess
from typing import Dict, Any

def _extract_error_summary(pytest_output: str) -> str:
    """
    Extracts the most relevant failure section (FAILURES / ERRORS traceback)
    from the raw pytest output to provide dense, actionable signal to the Debugger.
    """
    if "=== FAILURES ===" in pytest_output:
        failures_part = pytest_output.split("=== FAILURES ===")[-1]
        if "=== short test summary info ===" in failures_part:
            failures_part = failures_part.split("=== short test summary info ===")[0]
        return failures_part.strip()

    if "=== ERRORS ===" in pytest_output:
        return pytest_output.split("=== ERRORS ===")[-1].strip()

    # Fallback to the last 25 lines
    lines = pytest_output.splitlines()
    return "\n".join(lines[-25:])

El ayudante elimina los códigos de color del terminal y los encabezados genéricos de las pruebas, dejando solo el fallo específico de la afirmación (assert result == expected) junto con el rastro de pila. De este modo, el depurador recibe una descripción breve y precisa de lo que salió mal, sin tokens innecesarios.

El ejecutor principal, run_tests_in_sandbox(), se encuentra en el mismo archivo:

def run_tests_in_sandbox(code: str, tests: str, timeout_seconds: int = 15) -> Dict[str, Any]:
    """
    Executes unit tests against the generated code in an isolated temporary directory.
    Runs locally in a Python subprocess with zero LLM API calls ($0.00 / 0 tokens).
    """
    with tempfile.TemporaryDirectory() as temp_dir:
        code_file = os.path.join(temp_dir, "solution.py")
        test_file = os.path.join(temp_dir, "test_solution.py")

        # Write solution code and test files into the temporary directory
        with open(code_file, "w", encoding="utf-8") as f:
            f.write(code)

        with open(test_file, "w", encoding="utf-8") as f:
            f.write(tests)

        # Build execution command using the active Python interpreter
        python_exec = sys.executable
        cmd = [python_exec, "-m", "pytest", "-v", "test_solution.py"]

        try:
            result = subprocess.run(
                cmd,
                cwd=temp_dir,
                capture_output=True,
                text=True,
                timeout=timeout_seconds,
            )
            output = (result.stdout + "\n" + result.stderr).strip()
            exit_code = result.returncode

        except subprocess.TimeoutExpired as e:
            output = f"Execution Timed Out after {timeout_seconds} seconds! Possible infinite loop.\n" + (e.stdout or "")
            exit_code = -1

        passed = (exit_code == 0)

        passed_count = 0
        failed_count = 0

        passed_match = re.search(r"(\d+)\s+passed", output)
        if passed_match:
            passed_count = int(passed_match.group(1))

        failed_match = re.search(r"(\d+)\s+failed", output)
        if failed_match:
            failed_count = int(failed_match.group(1))
        elif not passed and exit_code != 0:
            failed_count = 1

        error_summary = None
        if not passed:
            error_summary = _extract_error_summary(output)

        return {
            "passed": passed,
            "exit_code": exit_code,
            "output": output,
            "passed_count": passed_count,
            "failed_count": failed_count,
            "error_summary": error_summary,
        }

Algunos detalles hacen que esta verificación sea fiable sin necesidad de la opinión de ningún LLM:

• tempfile.TemporaryDirectory() proporciona un espacio de trabajo nuevo en cada ejecución. El código generado y las pruebas se escriben allí, y el directorio se limpia al finalizar la ejecución.
• timeout=15 limita el tiempo de ejecución. El código que queda atascado en un bucle infinito o que tarda demasiado en ejecutarse tiene su subproceso eliminado, evitando así que bloquee todo el flujo de trabajo del agente.
• exit_code == 0 es la señal determinista de éxito o fracaso proveniente de pytest. Cero indica que el conjunto de pruebas se completó con éxito, mientras que cualquier otro valor significa que hay un problema; por lo tanto, el resultado y la traza de errores se envían al depurador.

Los resultados están estructurados. En lugar de devolver solo texto bruto de la terminal, la función extrae los conteos de éxitos y fracasos y crea un breve error_summary. Esa estructura se almacena en el estado compartido, y el gráfico la utiliza para decidir si continuar o entrar en el bucle de autoreparación.

Paso 4: instrucciones del sistema para cada especialista en src/prompts.py

En un sistema multiagente, las instrucciones deben ser breves, claras y tener un alcance bien definido. Cada agente necesita una visión clara de su tarea, sus responsabilidades y sus limitaciones, sin instrucciones innecesarias ni solapamientos con sus vecinos. El archivo prompts.py en src define cuatro instrucciones de sistema distintas para los roles especializados:

ARCHITECT_SYSTEM_PROMPT = """You are the Lead Software Architect and Quality Assurance Principal.
Your role is to analyze the user's software engineering requirements and design:
1. Architectural Specification: Detailed function/class signatures, data structures, and edge-case requirements.
2. Comprehensive Pytest Suite: A rigorous set of unit tests in `pytest` format.

IMPORTANT TEST RULES:
- The tests MUST import from `solution` (e.g. `from solution import ClassName, function_name`).
- Include test cases for standard happy paths AND tricky edge cases (e.g., empty inputs, boundary limits, invalid types, concurrency/ordering if applicable).
- Write self-contained tests without external network dependencies.

Provide your response in two distinct markdown sections:
### SPECIFICATIONS
<detailed specifications and API contracts>

### TESTS
```python
<complete pytest code>
```
"""

CODER_SYSTEM_PROMPT = """You are a Senior Python Developer.
Your role is to write clean, idiomatic, and robust Python code that implements the provided Architecture Specifications and passes 100% of the Pytest Suite.

RULES:
- Implement ONLY the implementation code (do NOT include tests in your output).
- Use proper typing, error handling, and docstrings.
- Output ONLY the executable Python code inside a ```python ``` code block. Do not include extraneous conversational text.
"""

DEBUGGER_SYSTEM_PROMPT = """You are an Expert Debugger and Causal Root-Cause Analyst.
A Python solution was executed against its test suite in an isolated sandbox and FAILED.

Your job:
1. Carefully inspect the failing Pytest Traceback and Error Summary.
2. Cross-reference the traceback with the provided Implementation Code and Test Suite.
3. Identify the EXACT root cause (e.g., off-by-one error, incorrect data structure, missing edge-case branch).
4. Provide a 2-3 sentence diagnosis explaining the bug.
5. Provide the COMPLETE, corrected Python implementation code.

Output format:
### DIAGNOSIS
<2-3 sentence root-cause diagnosis>

### PATCHED_CODE
```python
<complete corrected python code>
```
"""

REVIEWER_SYSTEM_PROMPT = """You are a Principal Code Reviewer.
The multi-agent engineering team has completed the code and passed all unit tests.
Your job is to provide a concise, high-level code quality summary:
- Verify correctness and adherence to original user requirements.
- Note any key algorithmic strengths or future optimization ideas.
- Keep the summary under 150 words.
"""

Los prompts están estructurados con dos objetivos:

• Delimitadores de salida explícitos. ARCHITECT_SYSTEM_PROMPT exige secciones ### SPECIFICATIONS y ### TESTS, lo que permite al analizador separar fácilmente la descripción de la arquitectura del código de prueba ejecutable.
• Separación estricta de responsabilidades. Se indica al programador que escriba únicamente código y no pruebas, mientras que el depurador debe proporcionar un diagnóstico conciso de dos oraciones junto al bloque completo de código corregido. Cada subagente tiene un alcance claro, lo que reduce las alucinaciones.

Paso 5: crear los nodos de subagente en src/nodes.py

Este es el motor de ejecución principal del flujo de trabajo, donde se definen las funciones de los nodos. Un nodo de LangGraph es una función sencilla de Python. Recibe el estado compartido AgentTeamState, realiza una tarea específica como llamar a un LLM o ejecutar pruebas, y devuelve los elementos que necesitan actualización. LangGraph los fusiona en el estado compartido, de modo que el nodo siguiente siempre cuente con el contexto más reciente.

Cree nodes.py en la carpeta src y añada los nodos uno por uno.

Analizadores auxiliares para la extracción de código y la división de especificaciones

Dos utilidades pequeñas mantienen el proceso ordenado. Los LLMs a menudo envuelven el código entre marcos de Markdown o incluyen introducciones extensas incluso cuando se les indica lo contrario. Estos analizadores se encargan de ese formato para que el resto del proceso siempre reciba código en Python limpio y ejecutable. Las importaciones y los analizadores auxiliares se colocan al principio de src/nodes.py:

import re
from typing import Dict, Any, List
from langchain_core.messages import SystemMessage, HumanMessage

from src.state import AgentTeamState
from src.config import get_llm
from src.sandbox import run_tests_in_sandbox
from src.prompts import (
    ARCHITECT_SYSTEM_PROMPT,
    CODER_SYSTEM_PROMPT,
    DEBUGGER_SYSTEM_PROMPT,
    REVIEWER_SYSTEM_PROMPT,
)


def _get_text(content: Any) -> str:
    """
    Extracts a clean string from either a standard string or list-of-dict content blocks
    returned by modern LangChain message objects.
    """
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for item in content:
            if isinstance(item, str):
                parts.append(item)
            elif isinstance(item, dict) and "text" in item:
                parts.append(str(item["text"]))
        return "\n".join(parts).strip()
    return str(content).strip()


def _extract_code(text: Any) -> str:
    """
    Extracts executable Python code from markdown backticks using regex.
    If no backticks are found, falls back to returning the cleaned raw text.
    """
    clean_text = _get_text(text)
    match = re.search(r"```(?:python)?\s*\n(.*?)\n```", clean_text, re.DOTALL)
    if match:
        return match.group(1).strip()
    return clean_text.strip()


def _parse_architect_output(raw_content: Any) -> tuple[str, str]:
    """
    Splits the Architect's raw output into its two distinct deliverables:
    the architectural specification text and the pytest test suite.
    """
    text = _get_text(raw_content)
    specs = text
    tests = ""

    if "### TESTS" in text:
        parts = text.split("### TESTS")
        specs = parts[0].replace("### SPECIFICATIONS", "").strip()
        tests = _extract_code(parts[1])
    elif "```python" in text:
        tests = _extract_code(text)
        specs = text.split("```python")[0].strip()

    return specs, tests

Cada ayuda aborda una peculiaridad común en las salidas de los LLM:

• _get_text() normaliza el formato de la respuesta. Dependiendo del proveedor y del wrapper de LangChain, una respuesta puede llegar como una cadena simple o como contenido estructurado, como una lista con entradas type y text. Esta ayuda convierte ambos formatos en una cadena de Python ordinaria.
• _extract_code() extrae el código de los bloques Markdown. Utiliza una expresión regular con re.DOTALL para capturar el código multi-línea, independientemente de que el modelo haya escrito python después de los corchetes invertidos de apertura. Cuando no existe ningún bloque delimitado, recurre al texto de la respuesta ya limpiado.

_parse_architect_output() divide la respuesta del Arquitecto en el delimitador ### TESTS. Todo lo que está antes de este marcador se convierte en specs, y todo lo que está después se convierte en tests. De esta manera, la especificación en lenguaje natural se separa del código de pruebas ejecutable que se pasa posteriormente al Ejecutor de Pruebas.

El nodo Arquitecto Principal

El Arquitecto Principal es el punto de entrada del equipo, y funciona siguiendo un enfoque de pruebas primero. Antes de que exista cualquier implementación, estudia la solicitud del usuario, define las estructuras de datos y las interfaces de las clases, y diseña un conjunto completo de pruebas pytest que abarca tanto los escenarios esperados como los casos límite importantes. Esto proporciona al Desarrollador un contrato claro para implementar.

La función se agrega al mismo archivo:

def architect_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 1: Lead Architect (High Reasoning Tier).
    Analyzes user requirements, designs architectural specifications,
    and authors an exhaustive Pytest suite covering happy paths and edge cases.
    """
    llm = get_llm("architect")
    messages = [
        SystemMessage(content=ARCHITECT_SYSTEM_PROMPT),
        HumanMessage(content=f"Requirements:\n{state['user_request']}"),
    ]
    response = llm.invoke(messages)
    specs, tests = _parse_architect_output(str(response.content))

    return {
        "specs": specs,
        "tests": tests,
        "status": "specs_ready",
    }

Tres aspectos destacan en el código:

• Selección del modelo. get_llm("architect") configura gemini-pro-latest con un nivel de razonamiento high. Un modelo potente capaz de realizar razonamientos profundos es fundamental aquí, ya que la tarea consiste en planificar a partir de los requisitos y anticipar casos límite como la seguridad de hilos, colecciones vacías o condiciones de borde.
• Construcción del mensaje. El prompt del sistema ARCHITECT_SYSTEM_PROMPT se envuelve en un SystemMessage, mientras que el prompt bruto del usuario proveniente de state['user_request'] se convierte en un HumanMessage.
• Actualizaciones de estado. El resultado es un diccionario que contiene specs, tests y status = "specs_ready"; LangGraph integra estos tres elementos en AgentTeamState.

El nodo del desarrollador

Una vez que existen la especificación y las pruebas, el control pasa al Desarrollador. Su tarea es escribir código en Python claro e idiomático que implemente la especificación y cumpla con todas las pruebas que elaboró el Arquitecto. La función se encuentra en src/nodes.py como developer_node():

def developer_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 2: Developer / Coder (Fast Workhorse Tier).
    Synthesizes the Python implementation code grounded in the specifications
    and the explicit assertions of the Pytest suite.
    """
    llm = get_llm("coder")
    prompt = f"""Requirements:
{state['user_request']}

Architect Specifications:
{state['specs']}

Pytest Suite:
```python
{state['tests']}
```

Write the complete Python implementation code to satisfy all tests.
"""
    messages = [
        SystemMessage(content=CODER_SYSTEM_PROMPT),
        HumanMessage(content=prompt),
    ]
    response = llm.invoke(messages)
    code = _extract_code(response.content)

    return {
        "code": code,
        "status": "code_generated",
    }

Funcionamiento:

• Selección del modelo. get_llm("coder") selecciona gemini-3.7-flash con un nivel de razonamiento medium. El Arquitecto ya ha definido el esquema de la API y las afirmaciones de prueba, por lo que el Desarrollador no necesita rediseñar nada y puede concentrarse en escribir código rápido y preciso. Un esfuerzo moderado es suficiente para ello, además de reducir el uso de tokens y los costos de desarrollo.
• Enfoque en el conjunto de pruebas. state['tests'] se inserta directamente en el prompt del Desarrollador. Al ver las afirmaciones exactas que debe cumplir, el modelo se orienta hacia los nombres de métodos, tipos de argumentos y valores de retorno esperados.
• Actualizaciones de estado. Después de extraer el código en Python puro de la respuesta, el nodo devuelve {"code": code, "status": "code_generated"}.

El nodo de entorno de pruebas determinista

Este es uno de los nodos más importantes del sistema. En lugar de preguntarle a un LLM si el código parece correcto, se ejecuta el conjunto de pruebas pytest contra el código en un subproceso aislado. Las pruebas se ejecutan localmente, por lo que este paso no realiza ninguna llamada a la API del LLM y generalmente termina en unos 15 milisegundos.

La función test_runner_node() se agrega a src/nodes.py:

def test_runner_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 3: Deterministic Test Sandbox ($0.00 / 0 Tokens).
    Executes tests against the code in an isolated temporary subprocess sandbox.
    Returns structured pass/fail metrics and concise error tracebacks upon failure.
    """
    test_results = run_tests_in_sandbox(
        code=state["code"],
        tests=state["tests"],
        timeout_seconds=15,
    )
    new_status = "tests_passed" if test_results["passed"] else "tests_failed"

    return {
        "test_results": test_results,
        "status": new_status,
    }

• Determinismo de cero tokens. run_tests_in_sandbox() recibe state["code"] y state["tests"]. No se llama a ningún modelo, por lo que este paso no cuesta ningún token.
• Determinación del estado. Un resultado de test_results["passed"] == True (lo que significa exit_code == 0) establece status en "tests_passed", mientras que cualquier otro resultado lo establece en "tests_failed".
• Salida de estado estructurada. Los test_results devueltos incluyen los resultados de la ejecución, los conteos de pruebas aprobadas y fallidas, además de un resumen condensado de la traza de pila. Esto es lo que examina el mecanismo de enrutamiento condicional.

El nodo del depurador autoreparable

Cada vez que la caja de pruebas informa fallos, se dirigen las rutas de control al depurador, el especialista en causas raíz. Este analiza tres elementos: el código defectuoso, las afirmaciones de prueba y el rastro de ejecución preciso de pytest, para determinar por qué falló la ejecución y luego aplica una corrección precisa. La función debugger_node() se encuentra en src/nodes.py:

def debugger_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 4: Debugger & Fixer (High Reasoning Tier).
    Performs causal root-cause analysis on execution tracebacks and stack traces.
    Applies surgical code patches and logs the diagnosis to the debug history.
    """
    llm = get_llm("debugger")
    test_results = state.get("test_results", {})
    error_summary = test_results.get("error_summary", test_results.get("output", "Unknown error"))
    current_iteration = state.get("iteration", 0) + 1

    prompt = f"""The implementation failed its unit tests in the execution sandbox.

Current Implementation Code:
```python
{state['code']}
```

Pytest Suite:
```python
{state['tests']}
```

Execution Error Traceback:
{error_summary}

Please analyze the root cause of the failure and output the corrected Python code.
"""
    messages = [
        SystemMessage(content=DEBUGGER_SYSTEM_PROMPT),
        HumanMessage(content=prompt),
    ]
    response = llm.invoke(messages)
    response_text = str(response.content)

    # Extract diagnosis summary and patched code block
    diagnosis = "Debugger applied code patch based on traceback."
    if "### DIAGNOSIS" in response_text and "### PATCHED_CODE" in response_text:
        diag_part = response_text.split("### PATCHED_CODE")[0]
        diagnosis = diag_part.replace("### DIAGNOSIS", "").strip()

    patched_code = _extract_code(response_text)

    # Append the diagnosis to our audit trail
    history = list(state.get("debug_history", []))
    history.append(f"Iteration {current_iteration}: {diagnosis}")

    return {
        "code": patched_code,
        "iteration": current_iteration,
        "debug_history": history,
        "status": "code_patched",
    }

Comportamiento:

• Alto nivel de razonamiento. get_llm("debugger") devuelve gemini-3.7-flash con el nivel de razonamiento establecido en high. Leer un seguimiento de llamadas implica seguir el camino de ejecución, identificar casos límite problemáticos y analizar cómo cambiaron las variables, lo cual requiere un esfuerzo real. Se eligió este modelo como compromiso entre profundidad y velocidad. Un modelo profesional también funciona bien para esta capa, e incluso es posible cambiar de modelo durante la ejecución si el depurador sigue atascado después de uno o dos bucles.
• Contexto de error detallado. Solo el error_summary simplificado llega al modelo; por ejemplo, AssertionError: assert False == True, where False = allow_request('user_1').

Conteo de iteraciones. Con current_iteration = state.get("iteration", 0) + 1, cada paso a través del depurador incrementa en uno el contador de intentos.

Rastro de auditoría. El diagnóstico de dos oraciones que genera el modelo se agrega a debug_history.

Actualizaciones de estado. El antiguo state['code'] se reemplaza por patched_code, y el nodo informa status = "code_patched".

El nodo Reviewer

Después de que cada prueba pase, o cuando se alcanza el límite de intentos, el control pasa al Reviewer, la última etapa de control de calidad. Este verifica la calidad general del código, se asegura de que las documentaciones estén claras y útiles, y escribe un breve resumen de la solución final para el usuario. La función reviewer_node() se agrega a src/nodes.py:

def reviewer_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 5: Code Reviewer (Fast Mode Tier).
    Validates code quality, verifies requirement adherence,
    and produces the final executive deliverable summary.
    """
    llm = get_llm("reviewer")
    test_results = state.get("test_results", {})

    prompt = f"""Requirements:
{state['user_request']}

Final Implementation Code:
```python
{state['code']}
```

Test Results:
Passed: {test_results.get('passed', False)} ({test_results.get('passed_count', 0)} passed, {test_results.get('failed_count', 0)} failed)
Iterations Taken: {state.get('iteration', 0)}
Debug History: {state.get('debug_history', [])}

Provide a concise, professional code quality review and verification summary.
"""
    messages = [
        SystemMessage(content=REVIEWER_SYSTEM_PROMPT),
        HumanMessage(content=prompt),
    ]
    response = llm.invoke(messages)

    return {
        "review_summary": _get_text(response.content),
        "status": "completed",
    }

Qué hace:

• Selección del modelo. get_llm("reviewer") devuelve gemini-flash-lite-latest con un nivel de razonamiento low, adecuado para la redacción rápida de resúmenes. En esta etapa se necesita poco razonamiento, por lo que un modelo más ligero con menor capacidad de razonamiento ofrece mayor velocidad y ahorro económico.
• Revisión contextual. El revisor recibe la implementación final, las cifras de aprobación y rechazo del entorno de pruebas, la cantidad de iteraciones de depuración utilizadas y el historial completo de depuración.
• Actualizaciones de estado. El nodo devuelve {"review_summary": _get_text(response.content), "status": "completed"}.

Paso 6: lógica de enrutamiento en src/edges.py

La lógica central y los nodos de agente ya están listos, por lo que el siguiente paso es el enrutamiento. Los nodos realizan los cálculos, mientras que las aristas deciden a dónde se dirige el control. Una arista condicional funciona como un enrutador dinámico: examina el diccionario de estado que devuelve un nodo y responde con el nombre del siguiente nodo a ejecutar.

La función de enrutamiento se encuentra en un nuevo archivo edges.py dentro de src:

from typing import Literal
from src.state import AgentTeamState

def route_after_test_runner(state: AgentTeamState) -> Literal["reviewer", "debugger"]:
    """
    Conditional edge router evaluated immediately after `test_runner` completes.
    Decides whether to trigger the self-healing Debugger loop or proceed to Reviewer.
    """
    test_results = state.get("test_results", {})
    passed = test_results.get("passed", False)
    iteration = state.get("iteration", 0)
    max_iterations = state.get("max_iterations", 3)

    # 1. Happy Path: 100% of unit tests passed -> Proceed directly to Review
    if passed:
        return "reviewer"

    # 2. Self-Healing Loop: Tests failed and within retry budget -> Trigger Debugger
    if iteration < max_iterations:
        return "debugger"

    # 3. Safety Fallback: Exceeded maximum retries -> Exit loop to Reviewer with warning
    return "reviewer"

El enrutador analiza los resultados de las pruebas y la cantidad de intentos de depuración realizados hasta el momento, y maneja tres casos:

• Ruta sin problemas (passed == True). Cuando todas las pruebas se superan en el primer intento, el flujo de trabajo pasa directamente al Revisor, y no hay motivo para gastar tiempo o tokens adicionales en el Depurador.
• Autoreparación (passed == False mientras iteration < max_iterations). Las pruebas fallidas con intentos aún disponibles envían el código al Depurador, que analiza el fallo e intenta encontrar una solución.
• Se alcanzó el límite de reintentos (iteration >= max_iterations). Cuando el Depurador ha agotado los tres intentos permitidos y las pruebas siguen fallando, el flujo de trabajo pasa al Revisor en lugar de seguir en un bucle interminable. Este límite estricto para la autoreparación evita que el proceso se estanque debido a requisitos contradictorios o ambiguos.

Paso 7: montar el grafo en src/graph.py

Ya se han definido los nodos y la ruta, por lo que ahora todo se conecta. Los enlaces de transferencia directa y los enlaces de retroalimentación condicional unen los nodos, y el resultado se compila en un StateGraph que puede ejecutar todo el flujo de trabajo. Coloque este código en un nuevo archivo graph.py dentro de src:

from langgraph.graph import StateGraph, END
from src.state import AgentTeamState
from src.nodes import (
    architect_node,
    developer_node,
    test_runner_node,
    debugger_node,
    reviewer_node,
)
from src.edges import route_after_test_runner

def build_self_healing_graph() -> StateGraph:
    """
    Assembles the Multi-Agent Self-Healing Engineering Team into a LangGraph StateGraph.
    Configures node registrations, linear handoffs, cyclic feedback edges, and conditional routing.
    """
    # 1. Initialize StateGraph with our Shared Whiteboard State Schema
    workflow = StateGraph(AgentTeamState)
    # 2. Register Sub-Agent Nodes (Mapping node names to Python callable functions)
    workflow.add_node("architect", architect_node)
    workflow.add_node("developer", developer_node)
    workflow.add_node("test_runner", test_runner_node)
    workflow.add_node("debugger", debugger_node)
    workflow.add_node("reviewer", reviewer_node)
    # 3. Define Graph Entry Point
    workflow.set_entry_point("architect")
    # 4. Add Deterministic Direct Handoff Edges
    workflow.add_edge("architect", "developer")
    workflow.add_edge("developer", "test_runner")

    # 5. Add the Cyclic Self-Healing Feedback Edge
    workflow.add_edge("debugger", "test_runner")
    # 6. Add Conditional Routing Edge for Sandbox Test Evaluation
    workflow.add_conditional_edges(
        "test_runner",
        route_after_test_runner,
        {
            "debugger": "debugger",
            "reviewer": "reviewer",
        },
    )
    # 7. Add Termination Edge
    workflow.add_edge("reviewer", END)
    # 8. Compile the Graph into an Executable State Machine
    return workflow.compile()

La construcción se realiza paso a paso:

1. StateGraph(AgentTeamState) crea un grafo con estado utilizando el esquema AgentTeamState, de modo que cada nodo puede alcanzar el mismo estado compartido a medida que avanza el flujo de trabajo.
2. workflow.add_node(name, func) registra cada función de Python como un nodo. Cada uno tiene una única responsabilidad: diseñar pruebas, generar código, ejecutar pruebas o depurar errores.
3. workflow.set_entry_point("architect") establece el punto de inicio. Al invocar o transmitir el grafo con un estado inicial, la ejecución comienza en el nodo architect.

workflow.add_edge("architect", "developer") y workflow.add_edge("developer", "test_runner") crean la línea recta inicial. Cuando el Arquitecto ha elaborado la especificación y las pruebas, toma el relevo el Desarrollador, y una vez que el Desarrollador ha creado el código, le sigue el Ejecutor de pruebas.

workflow.add_edge("debugger", "test_runner") cierra el bucle de autoreparación, lo que convierte un pipeline simple en un sistema autoreparable. Un error corregido no envía el flujo de trabajo al Revisor. El código reparado vuelve al Ejecutor de pruebas para que la corrección sea validada nuevamente.

workflow.add_conditional_edges("test_runner", route_after_test_runner, ...) hace que LangGraph llame a route_after_test_runner(state) una vez finalicen las pruebas. Su valor de retorno envía el flujo de trabajo al debugger o al reviewer.

workflow.add_edge("reviewer", END) finaliza la ejecución. Una vez que el revisor complete su verificación de calidad y su resumen, la ejecución del grafo termina.

workflow.compile() convierte la definición del grafo en uno listo para ejecutarse, validando su estructura y preparándolo para su ejecución.

Paso 8: el punto de entrada de la CLI en src/main.py

La última parte es un punto de entrada en la línea de comandos creado con rich, el cual transmite en tiempo real el progreso de cada subagente mediante paneles de terminal con formato estilizado. El archivo main.py en src comienza con las funciones para mostrar el banner y la tabla de configuración:

import os
import sys
import argparse
import time
import warnings

# Suppress internal SDK library warnings
warnings.filterwarnings("ignore")
# Ensure project root is in sys.path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.syntax import Syntax
from src.config import DEFAULT_CONFIG, MAX_DEBUG_ITERATIONS
from src.graph import build_self_healing_graph
from src.state import AgentTeamState
console = Console()

def print_banner():
    console.print(
        Panel.fit(
            "[bold cyan]LangGraph Self-Healing Multi-Agent Engineering Team[/bold cyan]\n"
            "[dim]Heterogeneous Model Tiering | Sub-Agent Tool Scoping | Autonomous Pytest Sandbox[/dim]",
            border_style="cyan",
        )
    )

def print_config_table():
    table = Table(title="Multi-Agent Team Configuration and Model Tiers", border_style="dim")
    table.add_column("Sub-Agent Role", style="bold yellow")
    table.add_column("Assigned Model", style="cyan")
    table.add_column("Thinking Level", style="green")
    table.add_column("Primary Responsibility", style="white")
    table.add_row(
        "Architect",
        DEFAULT_CONFIG["architect"]["model"],
        str(DEFAULT_CONFIG["architect"]["thinking_level"]).upper(),
        "Specifications and Pytest Edge-Case Suite",
    )
    table.add_row(
        "Developer",
        DEFAULT_CONFIG["coder"]["model"],
        str(DEFAULT_CONFIG["coder"]["thinking_level"]).upper(),
        "Fast Code Synthesis from Specification",
    )
    table.add_row(
        "Test Runner",
        "Deterministic Python ($0.00 / 0 Tokens)",
        "N/A",
        "Subprocess Pytest Sandbox Execution",
    )
    table.add_row(
        "Debugger",
        DEFAULT_CONFIG["debugger"]["model"],
        str(DEFAULT_CONFIG["debugger"]["thinking_level"]).upper(),
        "Causal Traceback Analysis and Patching",
    )
    table.add_row(
        "Reviewer",
        DEFAULT_CONFIG["reviewer"]["model"],
        str(DEFAULT_CONFIG["reviewer"]["thinking_level"]).upper(),
        "Code Verification and Executive Summary",
    )
    console.print(table)
    console.print()

Estos se refieren exclusivamente al formato del terminal y son fáciles de entender, por lo que no requieren una explicación detallada. El mismo archivo continúa con run_agent_pipeline():

def run_agent_pipeline(user_request: str):
    print_banner()
    print_config_table()

console.print(Panel(f"[bold white]User Request:[/bold white]\n{user_request}", border_style="blue"))
    initial_state: AgentTeamState = {
        "user_request": user_request,
        "specs": "",
        "tests": "",
        "code": "",
        "test_results": {},
        "iteration": 0,
        "max_iterations": MAX_DEBUG_ITERATIONS,
        "debug_history": [],
        "review_summary": "",
        "status": "starting",
    }
    graph = build_self_healing_graph()
    start_time = time.time()
    current_state = initial_state
    console.print("[bold green]Starting LangGraph Multi-Agent Workflow...[/bold green]\n")
    # Stream graph execution node by node
    for event in graph.stream(initial_state):
        for node_name, node_output in event.items():
            current_state.update(node_output)
            if node_name == "architect":
                console.print(
                    Panel(
                        f"[bold yellow][1/5] Architect Completed Specifications & Tests[/bold yellow]\n\n"
                        f"[bold]Specifications:[/bold]\n{current_state['specs'][:300]}...\n\n"
                        f"[bold]Generated Pytest Suite:[/bold]\n[dim]({len(current_state['tests'].splitlines())} lines of test code designed)[/dim]",
                        border_style="yellow",
                    )
                )
            elif node_name == "developer":
                console.print(
                    Panel(
                        f"[bold cyan][2/5] Developer Completed Initial Implementation[/bold cyan]\n"
                        f"[dim]Synthesized {len(current_state['code'].splitlines())} lines of Python code.[/dim]",
                        border_style="cyan",
                    )
                )
            elif node_name == "test_runner":
                results = current_state["test_results"]
                if results.get("passed", False):
                    console.print(
                        Panel(
                            f"[bold green][3/5] Test Runner Sandbox: PASSED[/bold green]\n"
                            f"[green]All {results.get('passed_count', 0)} unit tests passed cleanly in subprocess sandbox.[/green]",
                            border_style="green",
                        )
                    )
                else:
                    console.print(
                        Panel(
                            f"[bold red][3/5] Test Runner Sandbox: TESTS FAILED[/bold red]\n"
                            f"[red]{results.get('failed_count', 0)} failed, {results.get('passed_count', 0)} passed.[/red]\n\n"
                            f"[bold white]Traceback Snippet:[/bold white]\n[dim red]{results.get('error_summary', '')[:400]}[/dim red]",
                            border_style="red",
                        )
                    )
            elif node_name == "debugger":
                history = current_state.get("debug_history", [])
                latest_diag = history[-1] if history else "Patch applied."
                console.print(
                    Panel(
                        f"[bold magenta][4/5] Self-Healing Debugger (Iteration {current_state.get('iteration', 1)})[/bold magenta]\n"
                        f"[magenta]{latest_diag}[/magenta]\n\n"
                        f"[dim]Re-routing patched code to Test Runner Sandbox...[/dim]",
                        border_style="magenta",
                    )
                )
            elif node_name == "reviewer":
                console.print(
                    Panel(
                        f"[bold white][5/5] Reviewer Final Report[/bold white]\n"
                        f"{current_state.get('review_summary', '')}",
                        border_style="white",
                    )
                )
    elapsed = round(time.time() - start_time, 2)
    # Display Final Deliverables
    console.print("\n" + "=" * 80)
    console.print(f"[bold green]Pipeline Completed in {elapsed}s[/bold green]\n")
    console.print("[bold cyan]Final Verified Implementation (solution.py):[/bold cyan]")
    syntax_code = Syntax(current_state["code"], "python", theme="monokai", line_numbers=True)
    console.print(syntax_code)
    console.print("\n[bold yellow]Comprehensive Pytest Suite (test_solution.py):[/bold yellow]")
    syntax_tests = Syntax(current_state["tests"], "python", theme="monokai", line_numbers=True)
    console.print(syntax_tests)
    # Summary Metrics Table
    summary_table = Table(title="Execution Summary Metrics", border_style="cyan")
    summary_table.add_column("Metric", style="bold")
    summary_table.add_column("Value", style="green")
    summary_table.add_row("Total Elapsed Time", f"{elapsed} seconds")
    summary_table.add_row("Self-Healing Iterations", str(current_state.get("iteration", 0)))
    summary_table.add_row("Pytest Pass Status", "Passed (100%)" if current_state["test_results"].get("passed") else "Failed")
    summary_table.add_row("Unit Tests Passed", str(current_state["test_results"].get("passed_count", 0)))
    console.print(summary_table)

run_agent_pipeline() es donde se unen todas las partes. Establece el AgentTeamState compartido, crea el grafo autoreparable y transmite la ejecución nodo por nodo. Después de que cada nodo finaliza, actualiza el estado actual e imprime el progreso en paneles Rich: la especificación y las pruebas del Arquitecto, el código del Desarrollador, los resultados del Ejecutor de Pruebas, cada ronda de depuración y el informe final del Revisor. Al final, muestra la implementación verificada, el conjunto pytest generado y métricas de ejecución como el tiempo total, las iteraciones de depuración y los resultados de las pruebas.

El analizador de argumentos de la CLI cierra el archivo:

def main():
    parser = argparse.ArgumentParser(description="LangGraph Self-Healing Multi-Agent Engineering Team")
    parser.add_argument(
        "--prompt",
        type=str,
        default="Build an in-memory Sliding Window Rate Limiter class in Python with thread safety, per-user limits, and a clean method `allow_request(user_id: str) -> bool`.",
        help="Software engineering prompt for the multi-agent team",
    )
    args = parser.parse_args()
    run_agent_pipeline(args.prompt)

if __name__ == "__main__":
    main()

Ejecutarlo en un limitador de tasas

Para una prueba real, el equipo recibe una tarea algorítmica no trivial. Este comando solicita un limitador de tasas con ventana deslizante seguro para hilos:

python src/main.py --prompt "Build an in-memory Sliding Window Rate Limiter class in Python with thread safety, per-user limits, and a method allow_request(user_id: str) -> bool."

La salida de la terminal muestra la siguiente secuencia:

1. En la etapa 1 de 5, el arquitecto principal analiza los requisitos del limitador de velocidad y crea un conjunto de pruebas pytest multihilo que abarca el recorte de marcas de tiempo, la aplicación de umbrales con ventana deslizante y solicitudes concurrentes desde varios hilos.
2. En la etapa 2 de 5, el desarrollador escribe una primera clase SlidingWindowRateLimiter basada en collections.deque y un threading.Lock.
3. En la etapa 3 de 5, el ejecutor de pruebas en la iteración 0 ejecuta pytest. La prueba de estrés de concurrencia falla porque el recorte de marcas de tiempo se realizó fuera de la sección crítica protegida por el bloqueo, lo que generó una condición de carrera que permitió que varios hilos superaran el límite de solicitudes.

La etapa 4 de 5, el depurador, lee el rastro de errores y indica que el limitador de velocidad falló en la prueba de hilos concurrentes porque _prune_expired_timestamps() se llamó antes de adquirir self._lock, y que la solución envuelve la eliminación de timestamps dentro del bloque sincronizado.

Nuevamente la etapa 3, el ejecutor de pruebas en la iteración 1, ejecuta pytest sobre el código corregido, y el conjunto de pruebas muestra un resultado positivo con 6 pruebas superadas en 0.04s.

La etapa 5 de 5, el revisor, resume la implementación segura para hilos y las métricas de pruebas verificadas, y luego presenta el resultado final.

Una captura de pantalla de la salida real de la terminal de esta ejecución:

La salida completa también incluye el código en Python que se escribió y los casos de prueba utilizados para verificarlo. Reproducir todo esto aquí añadiría longitud sin mucho valor.

Dónde se aplica este patrón

Al combinar subagentes especializados, un modelo y nivel de pensamiento diferentes para cada uno, además de un entorno de ejecución determinista, se convierte un flujo de trabajo frágil basado en una sola instrucción en un sistema más fiable y capaz de autorepararse.

La demostración utilizó un equipo de generación de código, pero el mismo patrón de máquina de estados es aplicable a muchos otros flujos de trabajo, como pipelines de datos, orquestación dinámica de APIs y validación estructurada de documentos. La idea central es sencilla: no dependa de un solo modelo para hacer todo. Asigne a cada agente una responsabilidad específica y permita que el sistema verifique y corrija su propio trabajo a medida que avanza.

Tres lecciones destacan:

1. Nunca deje que un LLM evalúe su propio código. Delegue la ejecución a un entorno de subproceso local determinista que ejecute pytest, ya que es rápido, gratuito y proporciona una respuesta clara de éxito o fracaso.

Adapte el nivel de razonamiento al rol. La arquitectura y la depuración requieren un alto nivel de razonamiento, la síntesis equilibrada de código se ajusta a un nivel medio, y la redacción rápida de resúmenes puede funcionar con un nivel bajo.

Los bucles de retroalimentación cíclicos son la dirección hacia la que avanza la ingeniería de IA. Las máquinas de estado que se corrigen a sí mismas con base en la retroalimentación en tiempo real deberían superar con creces a las cadenas de instrucciones estáticas.