Samolecząca zespół wielu agentów w LangGraph: architekt, tester, debugger

Sztuczna inteligencja jest obecna niemal we wszystkich warstwach pracy z oprogramowaniem – od generowania kodu, przez analizę danych, aż po wieloetapowe rozumowanie. Mimo to każdy, kto prosi duży model językowy o coś złożonego, napotyka ten sam problem. Odpowiedź uzyskana za jednym razem może na pierwszy rzut oka wydawać się poprawna, ale zawodzi, gdy musi sprostać wymaganiom oraz kilku specjalnym przypadkom. Zwykłą metodą jest ręczny proces: kopiowanie dziennika błędów, wklejenie go do czatu, proszenie o naprawę i powtarzanie tego, aż w końcu wszystko zadziała.

Odpowiednia strategia ma również swoje wady. Przekazanie jednemu monolitycznemu agentowi dziesiątek narzędzi oraz długiej listy instrukcji zwykle zwiększa liczbę błędów, ponieważ jego kontekst ulega zanieczyszczeniu, a charakterystyki narzędzi zaczynają się ze sobą mieszać. Jakość wyników spada, gdyż agent nie jest w stanie jednocześnie realizować wszystkich tych instrukcji.

Przedstawiony poniżej projekt rozwiązuje oba te problemy za pomocą modułowej, hierarchicznej maszyny stanów wielu agentów napisanej w Pythonie z użyciem LangGraph. Przykładem jest samonaprawiający się zespół inżynierów oprogramowania składający się z pięciu ról: architekta, programisty, środowiska testowego Pytest opartego na podprocesach, debuggera oraz osoby sprawdzającej. Ten wzorzec można zastosować w wielu innych procesach inżynieryjnych opartych na sztucznej inteligencji.

Rozbicie dużego zadania na specjalistyczne podagenty, przydzielanie każdemu odpowiedniej ilości możliwości rozumowania, wykonywanie wyników w kontrolowanym środowisku oraz dodanie ciągłego obiegu informacji umożliwiają zespołowi testowanie własnej pracy, wykrywanie problemów i ich naprawę przed przekazaniem zweryfikowanego rezultatu. Najpierw spójrzmy na architekturę.

Kształt systemu określają dwie zasady.

Pierwszym rozwiązaniem jest hierarchiczne podziałanie modeli na różne poziomy z adaptacyjnymi stopniami myślenia. Role różnią się poziomem trudności, więc nie powinny wszystkie korzystać z tego samego modelu ani tego samego budżetu obliczeniowego:

• Architekt działa na modelu gemini-pro-latest z poziomem myślenia high. Wykonywa on najtrudniejsze zadania wymagające intensywnego rozumowania, takie jak określanie granic problemu, przewidywanie sytuacji konkurencyjnych oraz projektowanie rygorystycznych zestawów testowych pytest.
• Rozwijający korzysta z modelu gemini-3.7-flash z poziomem myślenia medium. Jest to zrównoważona konfiguracja umożliwiająca szybką syntezę kodu zgodnie z wyraźnymi specyfikacjami Architekta.
• Uruchamiający testy to deterministyczny podproces w Pythonie, który lokalnie uruchamia pytest w około 15 milisekund i w ogóle nie zużywa tokenów API LLM.

Debuger używa gemini-3.7-flash z poziomem myślenia high, ponieważ śledzenie surowych ścieżek stosu aż do błędnej linii wymaga analizy przyczynowej źródła problemu.

Przeglądający używa gemini-flash-lite-latest z poziomem myślenia low. Sprawdzanie stylu oraz pisanie podsumowania dla użytkownika wymagają jedynie prostego rozumowania.

Drugą zasadą jest ścisłe ograniczenie zakresu narzędzi, czyli zasada najmniejszych uprawnień stosowana wobec agentów. Żaden podagent nie ma nieograniczonego dostępu do terminala i systemu plików. Każdy otrzymuje tylko dane wejściowe oraz narzędzia niezbędne do wykonywania swojej roli. Programista nie może uruchamiać dowolnych poleceń w shellu, a wykonawca testów po prostu je uruchamia i raportuje to, co faktycznie się stało, bez proszenia modelu językowego o interpretację lub wymyślanie wyniku. Wąski zakres utrzymuje każdego agenta skupionym i zmniejsza ryzyko halucynacji.

Koncepcje LangGraph

LangGraph funkcjonuje jak wspólna tablica. Każdy członek zespołu, czyli każdy podagent, wykonywa swoje zadanie i zapisuje wynik z powrotem w tym samym wspólnym miejscu, dzięki czemu wszyscy mogą zobaczyć kontekst, który zespół dotychczas stworzył. LangGraph pełni rolę orkiestratora: decyduje, który agent ma działać dalej, i przekazuje odpowiedni stan.

Tutaj istotne są cztery koncepcje:

1. Stan, reprezentowany przez TypedDict, jest tą wspólną tablicą. Wszystkie agenty czytają i zapisują dane do jednego słownika w Pythonie o nazwie AgentTeamState, który zawiera prośbę użytkownika, specyfikację, testy jednostkowe, obecny projekt kodu, wyniki testów oraz liczbę rund debugowania.
2. Węzły to członkowie zespołu. Węzeł to po prostu funkcja w Pythonie, która otrzymuje aktualny stan, wykonywa swoje zadanie (wywołuje model językowy, uruchamia testy) i zwraca aktualizacje, które są ponownie łączone ze wspólnym stanem.

Krawędzie przenoszą logikę przepływu pracy. Bezpośrednia krawędź to stała transycja, na przykład zawsze przechodzenie od architekta do programisty. Warunkowa krawędź decyduje na podstawie obecnego stanu – np. kieruje do osoby sprawdzającej, gdy wszystkie testy przejdą pomyślnie, a do debuggera, gdy którykolwiek z nich zawiedzie.

Kompilowany graf to wynik rejestracji węzłów i krawędzi w obiekcie StateGraph oraz ich skompilowania. Można go uruchomić, a jego zdarzenia mogą być przesyłane w czasie rzeczywistym w miarę postępów przepływu pracy.

Ustawianie projektu

Zacznij od nowego katalogu i środowiska wirtualnego. Te polecenia w terminalu tworzą je oba i aktywują środowisko:

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

Plik requirements.txt wymienia następnie zależności:

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

Zainstalowanie ich w środowisku wirtualnym odbywa się za pomocą jednego polecenia:

pip install -r requirements.txt

Każdy pakiet ma swoje zadanie:

• langgraph koordynuje pracę wielu agentów, zarządzając stanem, przekazywaniem danych między węzłami oraz cyklicznymi pętlami sprzężenia zwrotnego.
• langchain-google-genai łączy się z modelami Google Gemini i udostępnia informacje o ich poziomie rozumowania.
• langchain-openai jest opcjonalny. Dodaje wsparcie dla OpenRouter lub OpenAI, dzięki czemu modele takie jak GPT 5.6 lub Deepseek v4 Pro mogą być użyte zamiast Gemini, jeśli nie chce się polegać wyłącznie na tym ostatnim.
• pytest jest narzędziem do wykonywania testów w lokalnej przestrzeni sandbox procesów potomnych. Uruchamia generowany zestaw testów i zwraca rzeczywiste wyniki.
• rich formatuje wyjście z terminala za pomocą kolorów, tabel i wyróżniania składni, co ułatwia śledzenie i debugowanie procesu pracy agentów.
• python-dotenv automatycznie ładuje klucze API oraz ustawienia modeli z pliku .env.

Plik .env znajdujący się w korzeniu projektu zawiera dane uwierzytelniające oraz konfigurację modelu:

# 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

Gdy środowisko jest gotowe, aplikację można budować krok po kroku.

Krok 1: stan wspólny w src/state.py

Najpierw należy zdefiniować schemat danych przekazywanych między agentami. Stwórz katalog src zawierający plik state.py.

Pierwszą definicją jest model TestResult napisany za pomocą Pydantic. Reprezentuje on ustrukturyzowany wynik wykonywania pytest w środowisku sandbox:

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 rejestruje, czy zestaw testów przeszedł z powodzeniem, kod wyjścia procesu, ilość testów, które przeszły i nie przeszły, oraz surowy wynik testów. Zawiera również pole error_summary z tylko istotną częścią śladu błędu, którą debugger wykorzystuje do znalezienia i naprawienia problemu.

Następnie znajduje się główne pole AgentTeamState w formie TypedDict. Jest to „tablica” przemieszczająca się po grafie: każdy węzeł odczytuje to, czego potrzebuje, wykona swoją pracę i aktualizuje pola, od których zależą kolejne węzły.

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

Każde pole pełni określoną rolę w procesie pracy:

• user_request zawiera oryginalny problem zgłoszony przez użytkownika, dzięki czemu każdy agent może zobaczyć początkową intencję przez cały czas trwania procesu.
• specs oraz tests pochodzą od Architekta. Programista implementuje rozwiązanie zgodnie ze specyfikacją, a Uruchamiający testy wykonywa generowane testy w odniesieniu do kodu.
• code to aktualna implementacja w języku Python. Programista wpisuje tam treść najpierw, a Debugger może ją zastąpić po nieudanym uruchomieniu.
• test_results przechowuje ustrukturyzowane wyniki z środowiska testowego, które decydują o tym, kto zajmie się sprawdzeniem – Recenzent czy Debugger.
• iteration oraz max_iterations kontrolują pętlę samonaprawczą. Zapobiegają one nieskończonym próbom ze strony Debuggera i ustalają maksymalną liczbę prób, na przykład trzy.

debug_history przechowuje przyczynę podstawową oraz rozwiązanie z każdej rundy debugowania, a także zapis zmian w kodzie.

review_summary otrzymuje końcowy raport recenzenta po tym, jak kod przejdzie weryfikację.

status pokazuje, na jakim etapie obecnie znajduje się proces: generowanie kodu, wykonywanie testów, debugowanie lub gotowość do recenzji.

Krok 2: poziomy modeli i stopnie rozumowania w src/config.py

Jednolita fabryka LLM przydziela odpowiedni model oraz poziom rozumowania (low, medium lub high) każdej roli podagenta. Plik src/config.py zaczyna się od słownika konfiguracji domyślnej:

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

Każda rola otrzymuje model oraz poziom myślenia dostosowane do trudności jej zadania, co zapewnia równowagę między siłą rozumowania, szybkością a kosztem, nie pogarszając przy tym ogólnej jakości:

• architect i debugger mają ustawiony poziom high, ponieważ projektowanie systemu oraz diagnoza błędów wymagają dogłębnych, wieloetapowych dedukcji.
• coder ma ustawiony poziom medium. Musi jedynie postępować zgodnie z planem przygotowanym przez architekta, więc wystarczy zrównoważona, szybka i realistyczna synteza.
• reviewer ma ustawiony poziom low, aby zaoszczędzić tokeny, ponieważ jego zadaniem jest jedynie napisanie podsumowania po przejrzeniu wyników.

W tym samym pliku znajduje się również funkcja 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)

Dwie rzeczy czynią tę funkcję przydatną. Przekazuje ona thinking_level (low, medium, high) bezpośrednio do ChatGoogleGenerativeAI, dzięki czemu intensywność rozumowania może być dostosowywana do danej roli. Ponadto każdy klient, który jest zwracany, implementuje BaseChatModel z LangChain, więc węzły grafu wywołują metodę .invoke() w ten sam sposób, niezależnie od tego, jaki dostawca stoi za daną rolą.

Krok 3: deterministyczna środowisko testowe pytest w pliku src/sandbox.py

Częstym błędem w projektowaniu agentów jest proszenie modelu językowego o sprawdzenie w swoim „umyśle” kodu, który właśnie napisał. Model może z pewnością przewidzieć, że kod działa poprawnie, mimo obecności błędów w nim samym, ponieważ modele regularnie tworzą iluzje co do poprawności swoich wyników. Nawet gdy model zidentyfikuje niektóre błędy, nie ma gwarancji, że udało mu się je wszystkie znaleźć.

Bardziej niezawodną metodą jest pozwolenie kodowi na samodzielne sprawdzenie się. Zamiast polegać na ocenie modelu, lokalny węzeł wykonywania zapisuje utworzony kod i testy do tymczasowego katalogu oraz uruchamia pytest w izolowanym podprocesie Pythona. Rezultatem jest obiektywny wynik „przejdzie” lub „nie przejdzie”, który może zostać bezpośrednio wprowadzony z powrotem do procesu pracy.

Stwórz plik sandbox.py w folderze src. Zaczyna się on od funkcji pomocniczej _extract_error_summary(), która pobiera najważniejsze szczegóły błędu z surowego wyniku pytest, aby debugger widział tylko użyteczną część ścieżki błędu, a nie cały raport:

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

Pomocnik usuwa kody kolorów terminala oraz standardowe nagłówki testów, pozostawiając dokładny opis błędu w twierdzeniu (assert result == expected) wraz z śladem stosu. Dzięki temu debugger otrzymuje krótki, zwięzły opis tego, co poszło nie tak, bez zbędnych informacji.

Główny mechanizm uruchamiania, run_tests_in_sandbox(), znajduje się w tym samym pliku:

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

Kilka szczegółów sprawia, że ta weryfikacja jest wiarygodna bez żadnej opinii modelu językowego:

• tempfile.TemporaryDirectory() zapewnia każdemu uruchomieniu nowe środowisko pracy. Tam zapisywany jest generowany kod oraz testy, a katalog jest oczyszczany po zakończeniu wykonywania.
• timeout=15 ogranicza czas wykonywania. Kod uwięziony w nieskończonej pętli lub działający zbyt długo ma swoją podproces eliminowaną, dzięki czemu nie może zablokować całego procesu agenta.
• exit_code == 0 to deterministyczny sygnał sukcesu lub niepowodzenia od pytest. Znak zero oznacza, że zestaw testów zakończył się pomyślnie, a każdy inny kod wskazuje na problem, więc wynik oraz ślad wywołań trafiają do narzędzia do debugowania.

Wyniki są ustrukturyzowane. Zamiast zwracać jedynie surowy tekst z terminala, funkcja wyodrębnia liczbę udanych i nieudanych prób oraz tworzy krótki error_summary. Ta struktura znajduje się w wspólnym stanie, a graf wykorzystuje ją do decydowania, czy kontynuować, czy wejść w pętlę samonaprawczą.

Krok 4: system wysyła pytania do każdego specjalisty w src/prompts.py

W systemie wielu agentów pytania powinny być krótkie, jednoznaczne i dobrze określone pod kątem zakresu. Każdy agent potrzebuje jasnego obrazu swoich zadań, obowiązków i ograniczeń, bez zbędnych instrukcji oraz bez nakładania się z obowiązkami sąsiadów. Plik prompts.py w katalogu src definiuje cztery odrębne pytania systemowe dla specjalistycznych ról:

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

Polecenia są skonstruowane z dwoma celami:

• Jasne delikatory wyjścia. ARCHITECT_SYSTEM_PROMPT wymaga sekcji ### SPECIFICATIONS i ### TESTS, co umożliwia parserowi łatwe oddzielenie opisu architektury od kodu testowego do wykonania.
• Strażne rozdzielenie zadań. Kodowcy poleca się pisać tylko kod, a nie testy, natomiast debugger musi przedstawić zwięzłą diagnozę w dwóch zdaniach obok kompletnego bloku kodu po naprawie. Każdy podagent ma jasno określony zakres działania, co zmniejsza ryzyko halucynacji.

Krok 5: tworzenie węzłów podagentów w pliku src/nodes.py

To jest główny silnik wykonywania w przepływie pracy, gdzie definiowane są funkcje węzłów. Węzeł LangGraph to zwykła funkcja w Pythonie. Bierze on udział w wspólnym stanie AgentTeamState, wykonywa określoną czynność, taką jak wezwanie modelu LLM lub uruchomienie testów, a następnie zwraca dane wymagające aktualizacji. LangGraph łączy je ze wspólnym stanem, dzięki czemu kolejny węzeł zawsze ma najnowszy kontekst.

Stwórz plik nodes.py w folderze src i dodawaj węzły po jednym.

Pomocnicze narzędzia do wydobywania kodu i dzielenia specyfikacji

Dwa małe narzędzia pomagają utrzymać porządek w procesie. Modele LLM często umieszczają kod w ramkach Markdown lub zaczynają go od rozmownych wprowadzeń, nawet gdy im się tego nie prosi. Te narzędzia zajmują się takim formatowaniem, aby reszta procesu zawsze otrzymywała czysty, wykonalny kod w Pythonie. Importy i narzędzia znajdują się na początku pliku 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

Każdy pomocnik zajmuje się typową osobliwością wyników generowanych przez LLM:

• _get_text() normalizuje format odpowiedzi. W zależności od dostawcy i narzędzia LangChain, odpowiedź może przyjść w postaci zwykłej ciągu znaków lub jako strukturyzowana treść, np. lista zawierająca pola type i text. Pomocnik przekształca obie formy na zwykły ciąg znaków w Pythonie.
• _extract_code() wydobywa kod z ramek Markdown. Wykorzystuje regularne wyrażenia z flagą re.DOTALL, dzięki czemu kod wieloliniowy w Pythonie jest identyfikowany niezależnie od tego, czy model napisał python po otwierających cudzysłowach, czy nie. Gdy brak jest takiej ramki, pomocnik korzysta z już oczyszczonego tekstu odpowiedzi.

_parse_architect_output() dzieli odpowiedź Architekta przy użyciu delimitera ### TESTS. Wszystko przed tym znacznikiem staje się specs, a wszystko po nim tests. Dzięki temu specyfikacja w języku naturalnym jest oddzielona od kodu testowego, który może być uruchomiony później przez Test Runnera.

Węzeł Lead Architect

Lead Architect stanowi punkt wejścia dla zespołu i pracuje według podejścia test-first. Zanim powstanie jakakolwiek implementacja, analizuje żądanie użytkownika, definiuje struktury danych i interfejsy klas, a także projektuje kompletny zestaw testów pytest, który obejmuje zarówno oczekiwane scenariusze prawidłowe, jak i ważne przypadki graniczne. Dzięki temu programista ma jasny kontrakt do realizacji.

Funkcja ta jest dodawana do tego samego pliku:

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

Trzy elementy wyróżniają się w tym kodzie:

• Wybór modelu. get_llm("architect") konfiguruje gemini-pro-latest z ustawieniem high dla procesowania myślowego. Silny model umiejący przeprowadzać dogłębne rozumowanie okazuje się tu przydatny, ponieważ zadaniem jest planowanie na podstawie wymagań oraz przewidywanie przypadków krawędziowych, takich jak bezpieczeństwo wątków, puste zbiory czy warunki brzegowe.
• Budowa wiadomości. Prompty systemowy ARCHITECT_SYSTEM_PROMPT jest umieszczany w obiekcie SystemMessage, natomiast surowy prompt użytkownika z state['user_request'] staje się HumanMessage.
• Aktualizacje stanu. Rezultatem jest słownik zawierający specs, tests oraz status = "specs_ready", a LangGraph łączy wszystkie te elementy w AgentTeamState.

Węzeł programisty

Gdy już istnieją specyfikacja i testy, kontrola przechodzi do programisty. Jego zadaniem jest napisanie czytelnego, idiomatycznego kodu w Pythonie, który realizuje specyfikację i przechodzi wszystkie testy stworzone przez architekta. Funkcja ta znajduje się w pliku src/nodes.py pod nazwą 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",
    }

Sposób działania:

• Wybór modelu. get_llm("coder") wybiera model gemini-3.7-flash z trybem myślenia medium. Architekt już ustalił schemat API oraz wymagania testowe, więc programista nie musi niczego przeprojektowywać i może skupić się na szybkim oraz dokładnym kodowaniu. Średni wysiłek wystarcza w tym celu, a także pomaga ograniczyć zużycie tokenów i koszty rozwoju.
• Opartość na zestawie testów. Wartość state['tests'] jest bezpośrednio włączana do promptu programisty. Widok dokładnych wymagań, które model musi spełnić, pomaga mu skupić się na oczekiwanych nazwach metod, typach argumentów oraz wartościach zwracanych.
• Aktualizacje stanu. Po wydobyciu zwykłego kodu Pythona z odpowiedzi, węzeł zwraca {"code": code, "status": "code_generated"}.

Węzeł deterministycznego środowiska testowego

To jeden z najważniejszych węzłów systemu. Zamiast pytać model językowy, czy kod jest poprawny, uruchamia się zestaw testów pytest na kodzie w izolowanym podprocesie. Testy wykonywane są lokalnie, więc ten krok nie wywołuje żadnych żądań do API modelu językowego i zazwyczaj trwa około 15 milisekund.

Funkcja test_runner_node() jest dodawana do pliku 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,
    }

• Determinizm z użyciem zeru tokenów. Funkcja run_tests_in_sandbox() otrzymuje wartości state["code"] oraz state["tests"]. Żaden model nie jest wywoływany, więc ten krok nie kosztuje żadnych tokenów.
• Określanie statusu. Wynik test_results["passed"] == True (co oznacza exit_code == 0) ustawia status na "tests_passed", natomiast wszelkie inne przypadki ustawiają go na "tests_failed".
• Wyjście w postaci ustrukturyzowanego stanu. Zwracane wartości test_results zawierają wyniki wykonywania testów, liczbę testów zakończonych pomyślnie oraz nieuproszczony zarys ścieżki wywołań. To właśnie ten element jest sprawdzany podczas kierowania przepływu przez warunki.

Węzeł debuggera samoregenerującego się

Gdy tylko sandbox zgłasza błędy, kieruje ruch do Debuggera, specjalisty od ustalania przyczyn źródłowych. Ten analizuje trzy elementy: uszkodzony kod, założenia testowe oraz dokładny ślad wywołań pytest, aby ustalić przyczynę niepowodzenia, a następnie stosuje precyzyjne rozwiązanie. Funkcja debugger_node() znajduje się w pliku 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",
    }

Sposób działania:

• Wysoki poziom przetwarzania rozumowania. Funkcja get_llm("debugger") przywraca model gemini-3.7-flash z ustawieniem high dla procesu myślenia. Analiza ścieżki wykonywania oznacza śledzenie drogi wykonania, znajdowanie problematycznych przypadków granicznych oraz rozważanie, jak zmieniły się zmienne, co wymaga znacznego wysiłku. Model ten został wybrany jako kompromis pomiędzy głębią analizy a szybkością. Model profesjonalny również sprawdzi się w tej roli, a system może nawet zmienić model w trakcie działania, jeśli debugger nadal utknie po jednej lub dwóch pętlach.
• Gęsty kontekst błędu. Do modelu trafia tylko skrócona wersja error_summary, na przykład AssertionError: assert False == True, where False = allow_request('user_1').

Liczenie iteracji. Dzięki current_iteration = state.get("iteration", 0) + 1 każde przejście przez debugger zwiększa licznik prób o jedynkę.

Ślad audytowy. Dwuczłonowa diagnoza wygenerowana przez model jest dodawana do debug_history.

Aktualizacja stanu. Stary state['code'] jest zastępowany przez patched_code, a węzeł zgłasza status = "code_patched".

Węzeł Reviewer

Po zakończeniu każdego udanego testu lub po osiągnięciu limitu prób kontrola przechodzi do węzła Reviewer, który stanowi ostateczną bramkę jakościową. Sprawdza ogólną jakość kodu, upewnia się, że dokumentacja jest jasna i przydatna, oraz tworzy krótkie podsumowanie gotowego rozwiązania dla użytkownika. Funkcja reviewer_node() jest dodawana do pliku 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",
    }

Jak to działa:

• Wybór modelu. Funkcja get_llm("reviewer") zwraca model gemini-flash-lite-latest przy poziomie myślenia low, odpowiednim do szybkiego tworzenia streszczeń. Na tym etapie nie jest potrzebne wiele rozumowania, więc lżejszy model o niższej zdolności analitycznej zapewnia lepszą szybkość i niższe koszty.
• Przegląd kontekstowy. Osoba przeglądająca otrzymuje ostateczną implementację, dane dotyczące liczby udanych i nieudanych prób z środowiska testowego, liczbę wykonywanych iteracji debugowania oraz pełną historię debugowania.
• Aktualizacja stanu. Węzeł zwraca wartość {"review_summary": _get_text(response.content), "status": "completed"}.

Krok 6: logika routingu w pliku src/edges.py

Logika podstawowa oraz węzły agentów zostały już stworzone, więc następny krok to routowanie. Węzły wykonywają obliczenia, a krawędzie decydują, dokąd trafia kontrola. Krawędź warunkowa działa jak dynamiczny router: sprawdza słownik stanów zwrócony przez węzeł i odpowiada nazwą następnego węzła do wykonania.

Funkcja routowania znajduje się w nowym pliku edges.py w katalogu 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"

Router analizuje wyniki testów oraz liczbę dotychczasowych prób naprawy i obsługuje trzy przypadki:

• Szlak pomyślny (passed == True). Gdy wszystkie testy przechodzą przy pierwszej próbie, proces przechodzi bezpośrednio do osoby przeglądającej, a nie ma potrzeby wydawania dodatkowego czasu ani zasobów na debuggera.
• Samonaprawa (passed == False przy jednoczesnym iteration < max_iterations). Testy, które nie przeszły, ale wciąż są dostępne próby naprawy, wysyłają kod do debuggera, który analizuje błąd i próbuje go naprawić.
• Osiągnięty limit ponownych prób (iteration >= max_iterations). Gdy debugger zużył wszystkie trzy dozwolone próby, a testy nadal nie przeszły, proces przechodzi do osoby przeglądającej, zamiast w nieskończoność powracać do tej samej procedury. Ten sztywny limit samonaprawy zapobiega zatrzymaniu się procesu z powodu sprzecznych lub niejednoznacznych wymagań.

Krok 7: budowa grafu w pliku src/graph.py

Knoty oraz mechanizmy routingu zostały już przygotowane, więc teraz wszystko zostanie połączone. Bezpośrednie przekazy danych oraz warunkowe krawędzie informacyjne łączą te node’y, a rezultatem jest StateGraph, który może uruchomić cały proces. Umieść ten kod w nowym pliku graph.py znajdującym się w katalogu 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()

Budowa odbywa się krok po kroku:

1. StateGraph(AgentTeamState) tworzy graf z stanami, wykorzystując schemat AgentTeamState, dzięki czemu każdy węzeł może osiągnąć ten sam wspólny stan w miarę postępu pracy.
2. workflow.add_node(name, func) rejestruje każdą funkcję w Pythonie jako węzeł. Każda z nich ma jedną rolę: projektowanie testów, generowanie kodu, wykonywanie testów lub naprawianie błędów.
3. workflow.set_entry_point("architect") określa punkt startowy. Uruchomienie lub przepływ danych przez graf z początkowym stanem zapoczątkowuje wykonywanie od węzła architect.

workflow.add_edge("architect", "developer") oraz workflow.add_edge("developer", "test_runner") tworzą początkową linię prostą. Gdy architekt przygotuje specyfikację i testy, przejmuje kontrolę programista, a gdy ten stworzy kod, następuje testowanie przez osobę odpowiedzialną za testy.

workflow.add_edge("debugger", "test_runner") zamyka pętlę samonaprawczą, która przekształca zwykły proces w system samonaprawczy. Błąd, który został już naprawiony, nie kieruje przepływu pracy do osoby odpowiedzialnej za przeglądanie. Naprawiony kod wraca do osoby odpowiedzialnej za testy, aby naprawa mogła zostać ponownie sprawdzona.

workflow.add_conditional_edges("test_runner", route_after_test_runner, ...) sprawia, że LangGraph wywołuje route_after_test_runner(state) po zakończeniu testów. Jego wartość zwracana kieruje przepływ pracy albo do debugger, albo do reviewer.

workflow.add_edge("reviewer", END) zakończa wykonywanie. Po tym, jak reviewer ukończy kontrolę jakości i stworzy podsumowanie, wykonywanie grafu jest zakończone.

workflow.compile() przekształca definicję grafu w graf gotowy do uruchomienia, walidując jego strukturę i przygotowując go do wykonywania.

Krok 8: punkt wejścia CLI w pliku src/main.py

Ostatni element to punkt wejścia w linii poleceń stworzony za pomocą rich, który w czasie rzeczywistym przekazuje informacje o postępach poszczególnych podagentów za pomocą stylizowanych paneli terminala. Plik main.py w katalogu src zaczyna się od funkcji do wyświetlania napisu nagłówkowego oraz tabeli konfiguracji:

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

Są one dotyczone wyłącznie stylizacji terminala i są oczywiste, więc nie wymagają szczegółowego omówienia. Ten sam plik kontynuuje się funkcją 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() to miejsce, gdzie wszystkie elementy się łączą. Ustawia wspólny stan AgentTeamState, tworzy samoregenerującą się strukturę grafową i przesyła dane o wykonywaniu węzłów po kolei. Po zakończeniu każdego węzła aktualizuje bieżący stan i wyświetla postęp w panelach Rich: specyfikację i testy architekta, kod programisty, wyniki wykonywania testów, każdą rundę debugowania oraz końcowy raport recenzenta. Na koniec wyświetla zweryfikowaną implementację, utworzoną bazę testów pytest oraz metryki wykonywania, takie jak całkowity czas trwania, liczba iteracji debugowania i wyniki testów.

Parser argumentów CLI zamyka ten plik:

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

Uruchamianie go w limiterze szybkości

W celu prawdziwego testu zespół otrzymuje niesamowitą, algorytmiczną zadanie. Ten polecenie wymaga limitera szybkości typu przesuwane okno bezpiecznego pod względem wątków:

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

Wynik w terminalu pokazuje następującą sekwencję:

1. Na etapie 1 z 5 główny architekt analizuje wymagania dotyczące ogranicznika szybkości i tworzy wielowątkową serię testów pytest obejmującą usuwanie zapisów czasowych, egzekwowanie progów w systemie przesuwanych okien oraz obsługę równoczesnych żądań z wielu wątków.
2. Na etapie 2 z 5 programista pisze pierwszą klasę SlidingWindowRateLimiter opartą na collections.deque oraz threading.Lock.
3. Na etapie 3 z 5 narzędzie do wykonywania testów w iteracji 0 uruchamia pytest. Test obciążeniowy dotyczący równoczesności zawodzi, ponieważ usuwanie zapisów czasowych odbywa się poza zablokowaną sekcją krytyczną, co powoduje sytuację wyścigu zegarów, dzięki której kilka wątków może przekroczyć dozwoloną liczbę żądań.

Etap 4 z 5, Debugger, analizuje ślad błędu i informuje, że ogranicznik szybkości nie przeszedł testu wątków równoczesnych, ponieważ funkcja _prune_expired_timestamps() została wywołana przed uzyskaniem dostępu do self._lock, a rozwiązanie polega na umieszczeniu operacji usuwania zapisanych czasów w bloku synchronizowanym.

Jeszcze raz etap 3 – Test Runner w iteracji 1 uruchamia pytest na kodzie poprawionym, a zestaw testów pokazuje wynik „zielony” z komunikatem 6 passed in 0.04s.

Etap 5 z 5, Reviewer, podsumowuje implementację bezpieczną pod względem wątków oraz sprawdzone metryki testowe, a następnie przedstawia ostateczny wynik.

Zdjęcie ekranu z rzeczywistym wyjściem terminala z tego uruchomienia:

Pełne wyjście zawiera również kod w Pythonie, który został napisany, oraz przypadki testowe użyte do jego sprawdzenia. Przedstawienie ich wszystkich tutaj zwiększyłoby długość tekstu bez większej wartości.

Gdzie ten wzorzec się stosuje

Łączenie specjalistycznych podagentów, różnych modeli oraz poziomów myślenia dla każdego z nich, wraz z deterministycznym środowiskiem wykonywania, przekształca kruchy proces oparty na jednym poleceniu w bardziej niezawodny system samonaprawczy.

Demonstracją była zespół do generowania kodu, ale ten sam wzorzec maszyny stanów pasuje do wielu innych procesów, takich jak łańcuchy przetwarzania danych, orkiestracja dynamicznych API oraz walidacja strukturalna dokumentów. Podstawa jest prosta: nie polegaj na jednym modelu do wykonywania wszystkiego. Przypisz każdemu agentowi konkretne zadanie i pozwól systemowi sprawdzać oraz korygować swoją pracę w trakcie jej wykonywania.

Wyróżniają się trzy kluczowe wnioski:

1. Nigdy nie pozwól modelowi językowemu oceniać własnego kodu. Przekaż wykonywanie zadania lokalnemu, deterministycznemu środowisku podprocesów uruchamiającemu pytest – jest ono szybkie, bezpłatne i daje dokładną odpowiedź o powodzeniu lub niepowodzeniu.

Należy dostosować poziom rozumowania do roli. Architektura i debugowanie wymagają zaawansowanego rozumowania, zrównoważona synteza kodu pasuje do poziomu średniego, natomiast szybkie pisanie streszczeń może odbywać się na niższym poziomie.

Cykliczne pętle sprzężenia zwrotnego to kierunek, w którym zmierza inżynieria AI. Maszyny stanowe, które same się korygują na podstawie rzeczywistych informacji z czasu wykonywania, powinny znacznie przewyższać statyczne łańcuchy poleceń.