Самовосстанавливающаяся команда многих агентов в LangGraph: архитектор, тестер, отладчик

Сейчас искусственный интеллект присутствует практически на каждом уровне работы с программным обеспечением — от генерации кода до анализа данных и многоэтапного рассуждения. Однако те, кто просит большую языковую модель решить сложную задачу, сталкиваются с одной и той же проблемой. Ответ, полученный за один раз, на первый взгляд может казаться правильным, но терпит неудачу при столкновении с конкретными требованиями и некоторыми крайними случаями. Обычным решением является ручной цикл: копирование журнала ошибок, вставка его в чат, запрос на исправление и повторение процесса до тех пор, пока всё наконец не заработает.

У противоположного подхода тоже есть свои проблемы. Предоставление одному монолитному агенту десятков инструментов и длинного списка инструкций обычно приводит к повышению уровня ошибок, поскольку его контекст загрязняется, а характеристики инструментов начинают смешиваться. Качество вывода снижается, так как агент не может выполнять все эти инструкции одновременно.

Приведенный ниже дизайн решает оба этих проблемы с помощью модульной, иерархической машины состояний с несколькими агентами, написанной на Python с использованием LangGraph. Пример работы — это самовосстанавливающаяся команда по разработке программного обеспечения, состоящая из пяти ролей: архитектора, разработчика, среды Pytest на основе подпроцессов, отладчика и ревизора. Эта схема может быть применена во многих других рабочих процессах в области ИИ.

Разделение большой задачи на специализированные подзадачи для отдельных агентов, предоставление каждому из них необходимого уровня способности к логическим рассуждениям, выполнение результатов в контролируемой среде и наличие постоянного цикла обратной связи позволяют команде тестировать собственную работу, выявлять проблемы и устранять их до передачи проверенного результата. Сначала рассмотрим архитектуру.

Система формируется под влиянием двух принципов.

Первый вариант — это иерархия гетерогенных моделей с адаптивными уровнями мышления. Роли отличаются по сложности, поэтому они не должны использовать одну и ту же модель или одинаковый объем вычислительных ресурсов:

• Архитектор работает с моделью gemini-pro-latest при уровне мышления high. Он выполняет наиболее сложные задачи, связанные с логическими рассуждениями, такие как определение границ проблемы, предвидение ситуаций конкуренции и разработка строгих наборов тестов pytest.
• Разработчик использует модель gemini-3.7-flash с уровнем мышления medium. Это сбалансированная конфигурация для быстрой генерации кода в соответствии с четкими указаниями Архитектора.
• Запускач тестов — это детерминистичный подпроцесс на Python, который выполняет pytest локально примерно за 15 миллисекунд и совершенно не тратит токены API больших языков моделирования.

Отладчик использует gemini-3.7-flash с режимом обработки high, поскольку восстановление исходных трейсов стека до ошибочной строки требует анализа причинно-следственных связей.

Проверщик использует gemini-flash-lite-latest с режимом обработки low. Проверка стиля и написание заключительного резюме для пользователя требуют лишь простого анализа.

Второй принцип — строгое ограничение функционала инструментов, то есть применение принципа минимальных привилегий к агентам. Ни один подагент не имеет неограниченного доступа к терминалу и файловой системе. Каждый получает только те входные данные и инструменты, которые необходимы для его роли. Разработчик не может запускать произвольные команды оболочки, а автоматический тестировщик просто выполняет тесты и сообщает о реальных результатах, не прося у большой языковой модели интерпретировать или выдумывать их. Узкий диапазон полномочий помогает каждому агенту сосредотачиваться и снижает вероятность галлюцинаций.

Связанные концепции LangGraph

LangGraph функционирует как общая доска для записей. Каждый участник команды, то есть каждый подагент, выполняет свою задачу и записывает результат обратно в этот общий пространство, так что все могут видеть контекст, созданный командой до настоящего момента. LangGraph выступает в роли оркестратора: он определяет, какой агент будет действовать следующим, и передаёт соответствующее состояние.

Здесь важны четыре ключевых момента:

1. Состояние, представленное объектом TypedDict, и является этой доской для записей. Все агенты читают и пишут в один Python-словарь под названием AgentTeamState, в котором хранятся запросы пользователя, спецификации, тесты на единицы, текущий черновик кода, результаты тестов и количество раундов отладки.
2. Узлы — это участники команды. Узел представляет собой просто Python-функцию, которая получает текущее состояние, выполняет свою задачу (запускает LLM, проводит тесты) и возвращает обновления, которые затем сливаются обратно в общее состояние.

Ребра несут логику рабочего процесса. Прямое ребро представляет собой фиксированную переходную операцию, например постоянный переход от архитектора к разработчику. Условное ребро определяет дальнейший путь в зависимости от текущего состояния: например, переход к рецензенту, когда все тесты пройдены, и к отладчику, когда хотя бы один тест не прошел.

Компилированная структура графа — это результат регистрации узлов и ребер в объекте StateGraph и их последующей компиляции. Ее можно запустить, а события, возникающие в процессе выполнения, могут транслироваться по мере развития рабочего процесса.

Настройка проекта

Начните с чистой директории и виртуальной среды. Эти команды терминала создают их обе и активируют виртуальную среду:

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

Файл requirements.txt затем перечисляет зависимости:

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

Установка их в виртуальную среду выполняется одной командой:

pip install -r requirements.txt

У каждого пакета есть своя функция:

• langgraph координирует работу нескольких агентов, обрабатывая состояние, передачу данных между узлами и циклические обратные связи.
• langchain-google-genai подключается к моделям Google Gemini и показывает уровень их мышления.
• langchain-openai является необязательным компонентом. Он добавляет поддержку OpenRouter или OpenAI, что позволяет использовать такие модели, как GPT 5.6 или Deepseek v4 Pro, если использование только Gemini не желательно.
• pytest — это инструмент для выполнения тестов внутри локальной среды подпроцессов. Он запускает генерированный набор тестов и возвращает реальные результаты.
• rich форматирует вывод в терминал с использованием цветов, таблиц и выделения синтаксиса, что облегчает отслеживание и отладку работы агентов.
• python-dotenv автоматически загружает ключи API и настройки моделей из файла .env.

Файл .env, находящийся в корне проекта, содержит учетные данные и настройки модели:

# 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

Как только среда готова, приложение можно создавать по частям.

Шаг 1: общее состояние в src/state.py

Прежде всего необходимо определить схему данных, передаваемых между агентами. Создайте каталог src с файлом state.py внутри него.

Первым определением является модель TestResult, написанная с использованием Pydantic. Она представляет структурированный результат выполнения pytest в сандбоксе:

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 фиксирует, прошла ли вся серия тестов успешно, код завершения процесса, сколько тестов пройдено и не пройдено, а также сырой вывод тестов. Кроме того, он содержит поле error_summary, в котором указывается только соответствующая часть трейс-кода, что позволяет отладчику найти причину сбоя и его устранить.

Далее следует основное поле AgentTeamState в формате TypedDict. Это своего рода доска, по которой перемещается процесс обработки: каждый узел считывает необходимую информацию, выполняет свои действия и обновляет поля, от которых зависят последующие узлы.

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

Каждое поле играет определенную роль в рабочем процессе:

• user_request содержит первоначальную проблему, описанную пользователем, чтобы каждый агент мог видеть первоначальную цель на протяжении всего процесса.
• specs и tests предоставляются архитектором. Разработчик реализует код в соответствии с этими спецификациями, а запускающий тесты инструмент выполняет сгенерированные тесты на этом коде.
• code — это текущая реализация на Python. Сначала ее заполняет разработчик, а отладчик может заменить ее после неудачного запуска.
• test_results хранит структурированные результаты работы в сандбоксе, которые помогают выбрать между ревизором и отладчиком.
• iteration и max_iterations отслеживают цикл самовосстановления. Они предотвращают бесконечные попытки отладчика и устанавливают предельное количество попыток, например три.

debug_history хранит первопричину проблемы и способ её устранения для каждого этапа отладки, а также записи о том, как менялся код.

review_summary получает окончательный отчет рецензента после того, как код пройдет проверку.

status отображает текущий этап рабочего процесса: генерация кода, запуск тестов, отладка или готовность к рецензированию.

Шаг 2: уровни моделей и уровни мышления в src/config.py

Единый механизм создания LLM назначает подходящую модель и уровень рассуждений (low, medium или high) каждой роли подагента. Файл src/config.py начинается с словаря значений по умолчанию:

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"))

Каждая роль получает модель и уровень мышления, соответствующие сложности её задачи, что позволяет сбалансировать силу рассуждений, скорость и затраты без ущерба для общего качества:

• architect и debugger настроены на уровень high, поскольку проектирование системы и диагностика стек-трейса требуют глубоких многоэтапных выводов.
• coder настроен на уровень medium. Ему достаточно следовать плану, разработанному архитектором, поэтому достаточно сбалансированного, быстрого и основанного на фактах синтеза.
• reviewer настроен на уровень low для экономии токенов, поскольку его задача заключается лишь в написании краткого обзора после просмотра результатов.

В том же файле также находится фабрика 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)

Два фактора делают эту функцию полезной. Во-первых, она передаёт значение thinking_level (low, medium, high) непосредственно в функцию ChatGoogleGenerativeAI, что позволяет настраивать уровень рассуждений в зависимости от роли. Во-вторых, каждый клиент, который она возвращает, реализует модель BaseChatModel из библиотеки LangChain, поэтому узлы графа вызывают метод .invoke() одинаковым образом, независимо от того, какой поставщик сервисов находится за конкретной ролью.

Шаг 3: детерминистичная среда pytest в файле src/sandbox.py

Частой ошибкой при проектировании агентов является требование к ИИ-модели проверять в своём внутреннем процессинге код, который она только что написала. Модель может с уверенностью предсказать, что код работает корректно, даже если в нём есть ошибки, поскольку она часто иллюзорно считает свой собственный вывод верным. Даже если модель всё же выявляет некоторые ошибки, нет гарантии, что она обнаружила их все.

Более надежным способом является позволение коду самому себя проверить. Вместо того чтобы полагаться на оценку модели, локальный узел выполнения записывает сгенерированный код и тесты во временную папку, а затем запускает pytest в изолированном подпроцессе Python. Результатом становится объективное решение о прохождении или неудаче, которое может быть сразу использовано в рабочем процессе.

Создайте файл sandbox.py в папке src. Он начинается с вспомогательной функции _extract_error_summary(), которая извлекает наиболее важные детали ошибки из необработанного вывода pytest, чтобы отладчик видел полезную часть трейс-кода, а не весь его объем:

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:])

Помощник удаляет коды цветов терминала и шаблонные заголовки тестов, оставляя точное описание сбоя утверждения (assert result == expected) вместе с трейсом стека. Таким образом, отладчик получает краткое и информативное описание проблемы без лишних данных.

Основной запускающий модуль, run_tests_in_sandbox(), находится в том же файле:

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,
        }

Несколько деталей делают эту проверку надежной без участия ИИ-моделей:

• tempfile.TemporaryDirectory() обеспечивает каждому запуску новое рабочее пространство. Сгенерированный код и тесты записываются там, а каталог очищается по окончании выполнения.
• timeout=15 ограничивает время выполнения. Если код застревает в бесконечном цикле или работает слишком долго, его подпроцесс прерывается, что предотвращает блокировку всего рабочего процесса агента.
• exit_code == 0 — это определенный сигнал о успешном или неудачном прохождении тестов от pytest. Значение ноль означает успешное выполнение тестовой сборки, а любое другое значение указывает на проблему; в таком случае результат и стек-трейс передаются в отладчик.

Результаты структурированы. Вместо того чтобы возвращать только чистый текст из терминала, функция вычисляет количество успешных и неудачных операций и формирует краткое описание ошибок error_summary. Эта структура хранится в общем состоянии, и граф использует её для принятия решения о том, продолжать работу или войти в цикл самовосстановления.

Шаг 4: запросы к каждому специалисту в файле src/prompts.py

В многопроцессорной системе запросы должны быть краткими, чётко сформулированными и иметь узкую спецификацию. Каждый агент должен чётко понимать свои задачи, обязанности и ограничения, без лишних инструкций и без перекрытия функций с соседними агентами. Файл prompts.py в папке src определяет четыре разных типа запросов для специализированных ролей:

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.
"""

Структура промптов определена с учётом двух целей:

• Четкие разделители вывода. Параметр ARCHITECT_SYSTEM_PROMPT требует наличия разделов ### SPECIFICATIONS и ### TESTS, что позволяет парсеру легко отделить описание архитектуры от исполняемого кода тестов.
• Строгое разделение обязанностей. Кодеру указывается писать только код, без тестов, в то время как отладчик должен предоставить краткий диагноз из двух предложений рядом с полным блоком кода после исправлений. У каждого подагента четко определена сфера деятельности, что снижает вероятность галлюцинаций.

Шаг 5: создание узлов подагентов в файле src/nodes.py

Это основной движок выполнения рабочего процесса, где определяются функции узлов. Узел LangGraph — это обычная функция на Python. Она берет общее состояние AgentTeamState, выполняет определенную задачу, такую как вызов LLM или запуск тестов, и возвращает данные, которые необходимо обновить. LangGraph объединяет их в общее состояние, чтобы следующий узел всегда имел самую свежую информацию.

Создайте файл nodes.py в папке src и добавляйте узлы по одному.

Вспомогательные парсеры для извлечения кода и разделения спецификаций

Два небольших утилиты помогают поддерживать порядок в работе конвейера. LLM часто заключают код в рамки Markdown или начинают его с длинных вступлений, даже когда им говорят этого не делать. Вспомогательные инструменты обрабатывают такую форматированную информацию, чтобы остальная часть конвейера всегда получала чистый, готовый к выполнению код на Python. Импорты и вспомогательные функции размещаются в начале файла 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

Каждый вспомогательный инструмент решает определенную проблему, характерную для вывода больших языковых моделей:

• _get_text() нормализует формат ответа. В зависимости от поставщика и обертки LangChain ответ может приходить в виде простой строки или структурированного контента, такого как список с элементами type и text. Этот инструмент преобразует оба вастоячную строку на Python.
• _extract_code() извлекает код из блоков Markdown. Для этого используется регулярное выражение с параметром re.DOTALL, что позволяет захватывать многорядочный Python независимо от того, указано ли после начальных обратных кавычек слово python. Если блоков Markdown нет, используется очищенный текст ответа.

_parse_architect_output() разделяет ответ Архитектора по разделителю ### TESTS. Всё, что находится до этого маркера, становится specs, а всё, что после — tests. Это позволяет отделить спецификацию на естественном языке от кода тестов, который позже передаётся исполнителю тестов.

Узел Lead Architect

Lead Architect является точкой входа для команды, и работа ведётся с учётом тестов в первую очередь. До того, как появится какая-либо реализация, он изучает запрос пользователя, определяет структуры данных и интерфейсы классов, а также разрабатывает полный набор тестов pytest, охватывающий как ожидаемые успешные сценарии, так и важные крайние случаи. Это даёт разработчику чёткие требования для реализации.

Функция добавляется в тот же файл:

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",
    }

В коде выделяются три аспекта:

• Выбор модели. Вызов get_llm("architect") настраивает использование gemini-pro-latest с режимом мышления high. Здесь крайне важна мощная модель с способностью к глубокому анализу, поскольку требуется планирование на основе требований и учет крайних случаев, таких как безопасность потоков, пустые коллекции или граничные условия.
• Создание сообщения. Системный промпт ARCHITECT_SYSTEM_PROMPT помещается в структуру SystemMessage, в то время как сырой промпт пользователя из state['user_request'] преобразуется в HumanMessage.
• Обновление состояния. Результатом является словарь, содержащий поля specs, tests и status = "specs_ready"; LangGraph объединяет все эти элементы в структуру AgentTeamState.

Узел разработчика

Как только существуют спецификация и тесты, управление переходит к разработчику. Его задача — написать чистый, стилистически корректный код на Python, который реализует спецификацию и проходит все тесты, написанные архитектором. Функция помещается в файл src/nodes.py под названием 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",
    }

Как это работает:

• Выбор модели. Вызов get_llm("coder") приводит к использованию gemini-3.7-flash с режимом обработки medium. Архитектор уже определил структуру API и требования к тестам, поэтому разработчику не нужно ничего перерабатывать — он может сосредоточиться на быстром и точном написании кода. Достаточно умеренных усилий, что также снижает расход токенов и затраты на разработку.
• Ориентация на набор тестов. Значение state['tests'] напрямую включается в запрос к модели. Видимость конкретных требований, которые необходимо выполнить, помогает модели сосредоточиться на ожидаемых именах методов, типах аргументов и значениях возврата.
• Обновление состояния. После извлечения чистого кода на Python узел возвращает {"code": code, "status": "code_generated"}.

Узел детерминированной среды тестирования

Это один из самых важных узлов системы. Вместо того чтобы спрашивать большую языковую модель, выглядит ли код правильно, она запускает набор тестов pytest против кода в изолированном подпроцессе. Тесты выполняются локально, поэтому этот шаг не совершает никаких вызовов API большой языковой модели и обычно завершается примерно за 15 миллисекунд.

Функция test_runner_node() добавляется в файл 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,
    }

• Детерминизм с нулевым количеством токенов. Функция run_tests_in_sandbox() принимает параметры state["code"] и state["tests"]. Модель не вызывается, поэтому этот шаг не требует токенов.
• Определение статуса. Если результат проверки test_results["passed"] == True (что означает exit_code == 0), то значение status устанавливается в "tests_passed"; в остальных случаях — в "tests_failed".
• Вывод структурированного состояния. Возвращаемый объект test_results содержит результаты выполнения тестов, количество пройденных и не пройденных тестов, а также краткое описание стек-трейса. Именно это значение используется узлом для принятия решений при маршрутизации.

Узел отладчика с самовосстанавливающейся функцией

Каждый раз, когда среда тестирования сообщает о сбоях, управление перенаправляется в отладчик — специалиста по выявлению коренных причин. Он анализирует три элемента: поврежденный код, утверждения тестов и точный стек-трейс pytest, чтобы понять, почему произошел сбой, после чего вносит точечные исправления. Функция debugger_node() находится в файле 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",
    }

Как это работает:

• Высокий уровень рассуждений. Вызов get_llm("debugger") возвращает модель gemini-3.7-flash с настроенным уровнем рассуждений на high. Анализ стек-трейса подразумевает прослеживание пути выполнения, поиск нестабильных крайних случаев и анализ изменений переменных, что требует значительных усилий. Эта модель была выбрана в качестве компромисса между глубиной анализа и скоростью работы. Для этого уровня также подойдет профессиональная модель, причем при необходимости можно даже сменить модель во время выполнения, если отладчик застрянет после одного-двух циклов.
• Подробный контекст ошибки. К модели поступает только сокращенная версия error_summary, например AssertionError: assert False == True, where False = allow_request('user_1').

Подсчёт итераций. С помощью current_iteration = state.get("iteration", 0) + 1 при каждом прохождении через отладчик счётчик попыток увеличивается на один.

Хроника действий. Диагноз модели, состоящий из двух предложений, добавляется в debug_history.

Обновление состояния. Старое значение state['code'] заменяется на patched_code, и узел сообщает status = "code_patched".

Узел Reviewer

После прохождения каждого теста или при достижении лимита попыток управление переходит к узлу Reviewer — финальному этапу проверки качества. Он оценивает общее качество кода, проверяет, ясны ли и полезны ли документации, а также пишет краткое резюме готового решения для пользователя. Функция reviewer_node() добавляется в файл 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",
    }

Что это делает:

• Выбор модели. Вызов get_llm("reviewer") возвращает модель gemini-flash-lite-latest с уровнем обработки low, подходящую для быстрого написания кратких резюме. На этом этапе требуется минимум логических операций, поэтому более простая модель с низкой способностью к обработке обеспечивает более высокую скорость и снижает затраты.
• Анализ контекста. Система Reviewer получает окончательную реализацию, показатели успеха и неудач из среды тестирования, количество выполненных итераций отладки и полную историю отладки.
• Обновление состояния. Узел возвращает значение {"review_summary": _get_text(response.content), "status": "completed"}.

Шаг 6: логика маршрутизации в файле src/edges.py

Основная логика и узлы агентов уже реализованы, поэтому следующим шагом является маршрутизация. Узлы выполняют вычисления, а рёбра определяют направление управления. Условное ребро действует как динамический маршрутизатор: оно анализирует словарь состояния, возвращённый узлом, и указывает имя следующего узла для выполнения.

Функция маршрутизации находится в новом файле edges.py внутри папки 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"

Роутер анализирует результаты тестов и количество попыток отладки, совершённых до настоящего времени, и обрабатывает три сценария:

• Успешный путь (passed == True). Когда все тесты проходят с первой попытки, работа направляется непосредственно к рецензенту, и нет необходимости тратить дополнительное время или ресурсы на отладчик.
• Самовосстановление (passed == False при условии iteration < max_iterations). Неудачные тесты, при этом ещё имеющиеся попытки, отправляются в отладчик, который анализирует причину сбоя и пытается его устранить.
• Достигнут лимит попыток (iteration >= max_iterations). Когда отладчик исчерпал все три разрешённые попытки, а тесты всё ещё не проходят, работа переходит к рецензенту, вместо того чтобы бесконечно циклировать. Такой жёсткий лимит предотвращает застревание процесса из-за противоречивых или неоднозначных требований.

Шаг 7: сборка графа в src/graph.py

Узлы и маршрутизация уже настроены, теперь все соединяется между собой. Прямые связи и угловые ребра обратной связи соединяют узлы, в результате чего формируется объект StateGraph, способный запустить весь рабочий процесс. Поместите этот код в новый файл graph.py внутри папки 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()

Сборка происходит поэтапно:

1. StateGraph(AgentTeamState) создаёт граф с состояниями с использованием схемы AgentTeamState, так что каждый узел может достигать одинакового общего состояния по мере продвижения рабочего процесса.
2. workflow.add_node(name, func) регистрирует каждую функцию на Python как узел. Каждая из них выполняет одну функцию: разработку тестов, генерацию кода, запуск тестов или отладку ошибок.
3. workflow.set_entry_point("architect") определяет точку начала. При вызове или потоковой обработке графа с начальным состоянием выполнение начинается с узла architect.

workflow.add_edge(“architect”, “developer”) и workflow.add_edge(“developer”, “test_runner”) формируют начальную прямую последовательность действий. Как только архитектор подготовил спецификацию и тесты, их берет на себя разработчик, а когда разработчик создает код, следует тестировщик.

workflow.add_edge(“debugger”, “test_runner”) закрывает цикл самовосстановления, превращая обычную цепочку операций в систему с возможностью самовосстановления. Ошибка, уже исправленная, не отправляет работу на рассмотрение рецензента. Исправленный код возвращается тестировщику для повторной проверки корректности решения.

workflow.add_conditional_edges("test_runner", route_after_test_runner, ...) заставляет LangGraph вызвать route_after_test_runner(state) после завершения тестов. Его возвращаемое значение направляет рабочий процесс либо к debugger, либо к reviewer.

workflow.add_edge("reviewer", END) завершает выполнение. После того как проверяющий заканчивает свою проверку качества и составление отчета, выполнение графа прекращается.

workflow.compile() преобразует определение графа в готовый к выполнению граф, проверяя его структуру и подготавливая к запуску.

Шаг 8: точка входа CLI в файле src/main.py

Последним элементом является точка входа в командную строку, созданная с использованием библиотеки rich, которая в реальном времени отображает прогресс каждого подагента с помощью стилизованных панелей терминала. Файл main.py в папке src начинается с функций отображения заголовка и таблицы настроек:

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()

Эти элементы касаются исключительно стилизации терминала и сами по себе понятны, поэтому не требуют подробного объяснения. В том же файле далее идет функция 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() — это место, где все компоненты соединяются. Он создаёт общее состояние AgentTeamState, формирует самовосстанавливающуюся структуру графа и передаёт данные об исполнении по узлам поочередно. После завершения каждого узла обновляется текущее состояние, а прогресс отображается в панелях типа Rich: спецификации и тесты архитектора, код разработчика, результаты выполнения тестов, каждый этап отладки и заключительный отчёт рецензента. В конце выводится проверенная реализация, сгенерированный набор тестов pytest и показатели работы, такие как общее время выполнения, количество итераций отладки и результаты тестов.

Парсер аргументов командной строки завершает работу с файлом:

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()

Запуск с использованием ограничителя частоты

Для реального тестирования команда получает не простую алгоритмическую задачу. Эта команда предназначена для использования безопасного с точки зрения потоков ограничителя частоты с скользящим окном:

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."

В выводе терминала показана следующая последовательность действий:

1. На 1-м из 5 этапов ведущий архитектор анализирует требования к ограничителю скорости и создаёт многопоточный набор тестов pytest, охватывающий обрезку временных меток, применение порогов с скользящим окном и одновременные запросы от нескольких потоков.
2. На 2-м из 5 этапов разработчик пишет первый класс SlidingWindowRateLimiter, основанный на структуре collections.deque и объекте threading.Lock.
3. На 3-м из 5 этапов запускается тестировщик на итерации 0 для выполнения pytest. Тест на нагрузку из-за одновременных операций проваливается, поскольку обрезка временных меток происходит вне защищённой критической секции, что приводит к конкурентному состоянию и позволяет нескольким потокам превысить установленное количество запросов.

На четвертом из пяти этапов, на этапе отладчика, считывается трейс-код и сообщается о том, что механизм ограничения скорости не прошел тест на одновременные потоки, поскольку функция _prune_expired_timestamps() была вызвана до того, как был получен защитный механизм self._lock, причем исправление заключается в том, что операция удаления старых временных меток выполняется внутри синхронизированного блока.

На третьем этапе, во время первой итерации исполнителя тестов, на модифицированный код запускается pytest, и результат тестирования оказывается успешным: 6 пройдено за 0.04 секунды.

На пятом из пяти этапов, на этапе проверки, обобщаются информация об реализации, безопасной к использованию в многопоточных средах, и результаты проверки тестов, после чего приводится окончательный вердикт.

Скриншот реального вывода в терминале после этого запуска:

Полный вывод также включает написанный Python-код и тестовые случаи, использованные для его проверки. Перечисление всего этого здесь увеличит объем текста без значительной пользы.

Где применим этот подход

Сочетание специализированных подагентов, различных моделей и уровней мышления для каждого из них, а также среды выполнения с детерминированным подходом превращает хрупкий рабочий процесс, основанный на одном запросе, в более надежную систему с возможностью самовосстановления.

В качестве примера была показана команда по генерации кода, но тот же паттерн состояний подходит и для многих других рабочих процессов, таких как обработка данных, оркестрация динамических API и проверка структурированных документов. Основная идея проста: не стоит полагаться на одну модель для выполнения всех задач. Необходимо назначить каждому агенту конкретную задачу и позволить системе проверять и корректировать свою работу по мере её выполнения.

Выделяются три ключевых урока:

1. Никогда не позволяйте большой языковой модели оценивать собственный код. Передайте задачу выполнения локальной среде с детерминированным подходом, запускающей pytest — она быстрая, бесплатная и даёт точный ответ о прохождении или неудаче теста.

Необходимо соответствовать уровню мышления роли: архитектура и отладка требуют высокого уровня рассуждений, сбалансированная синтезация кода — среднего, а быстрое написание кратких резюме — низкого уровня.

Циклические замкнутые системы обратной связи — вот куда движется инженерия ИИ. Статические машины состояний, способные корректировать себя на основе реальной обратной связи во время выполнения, должны значительно превзойти статические цепочки команд.