Équipe multi-agents auto-réparatrice dans LangGraph : architecte, testeur, débogueur
Créez une machine à états LangGraph en Python dans laquelle un Architecte, un Développeur, un environnement de test pytest, un Débogueur et un Réviseur collaborent et itèrent jusqu’à ce que les tests réussissent.
Sources: Build a Self-Healing Multi-Agent Team with LangGraph
L’IA est désormais présente dans presque tous les niveaux du développement logiciel, de la génération de code à l’analyse de données en passant par le raisonnement multi-étapes. Pourtant, quiconque demande à un grand modèle de langage d’effectuer une tâche complexe se heurte au même problème : une réponse obtenue en une seule tentative peut sembler correcte en apparence, mais échouer dès qu’elle doit faire face aux exigences réelles ou à certains cas limites. La solution habituelle consiste en une boucle manuelle : copier le journal des erreurs, le coller dans la chat, demander une correction, et répéter jusqu’à ce que tout fonctionne enfin.
L’approche inverse présente également ses propres difficultés. Fournir à un agent monolithique des dizaines d’outils ainsi qu’une longue liste d’instructions a tendance à augmenter le taux d’erreurs, car son contexte se trouve pollué et les signatures des outils finissent par se confondre. La qualité des résultats diminue, car l’agent ne peut pas suivre toutes ces instructions en même temps.
La conception présentée ci-dessous résout ces deux problèmes grâce à une machine d’état multi-agents modulaire et hiérarchique écrite en Python avec LangGraph. L’exemple fonctionnel représente une équipe d’ingénierie logicielle auto-réparatrice composée de cinq rôles : un architecte, un développeur, un environnement de test Pytest basé sur des sous-processus, un débogueur et un réviseur. Ce modèle peut être adapté à de nombreux autres workflows d’ingénierie IA.
En divisant une tâche complexe entre des sous-agents spécialisés, en leur conférant le niveau de raisonnement approprié, en exécutant les résultats dans un environnement contrôlé et en ajoutant un mécanisme de feedback continu, l’équipe peut tester son propre travail, identifier les problèmes et les corriger avant de livrer un résultat vérifié. Commençons par examiner l’architecture.
Deux principes guident ce système.
Le premier aspect concerne la hiérarchisation des modèles hétérogènes avec des niveaux de réflexion adaptatifs. Les rôles diffèrent en termes de difficulté, il ne faut donc pas que tous utilisent le même modèle ou le même budget de calcul :
- L’Architecte fonctionne avec
gemini-pro-latestet un niveau de réflexionélevé. Il effectue les tâches nécessitant le plus de raisonnement, telles que la définition des limites du problème, l’anticipation des conditions de concurrence et la conception de suitespytestrigoureuses. - Le Développeur utilise
gemini-3.7-flashavec un niveau de réflexionmoyen. Il s’agit d’une configuration équilibrée permettant une synthèse rapide de code conforme aux spécifications explicites de l’Architecte. - L’Exécuteur de tests est un sous-processus Python déterministe qui lance
pytestlocalement en environ 15 millisecondes sans consommer aucun token de l’API LLM.
gemini-3.7-flash avec un niveau de réflexion élevé, car le suivi des traces d’erreur brutes jusqu’à la ligne défectueuse exige une analyse approfondie des causes racines.gemini-flash-lite-latest avec un niveau de réflexion faible. Vérifier le style et rédiger un résumé pour l’utilisateur ne nécessite qu’une réflexion légère.Le deuxième principe est un encadrement strict des outils, c’est-à-dire l’application du principe du moindre privilège aux agents. Aucun sous-agent n’a accès illimité au terminal et au système de fichiers. Chacun ne reçoit que les entrées et les outils nécessaires à son rôle. Le développeur ne peut pas exécuter de commandes de shell arbitraires, et l’exécuteur de tests se contente d’exécuter les tests et de rapporter ce qui s’est réellement produit, sans demander à un LLM d’interpréter ou d’inventer le résultat. Un champ d’action restreint permet à chaque agent de rester concentré et réduit les hallucinations.
Les concepts LangGraph concernés
LangGraph fonctionne comme une table blanche partagée. Chaque membre de l’équipe, c’est-à-dire chaque sous-agent, effectue sa tâche assignée et écrit le résultat dans cet espace commun, de sorte que tout le monde peut voir le contexte construit par l’équipe jusqu’à présent. LangGraph est l’orchestrateur : il décide quel agent doit s’exécuter en suivant et transmet l’état pertinent.
Quatre idées sont importantes ici :
- L’état, représenté par un
TypedDict, correspond à la table blanche. Tous les agents lisent et écrivent dans un dictionnaire Python nomméAgentTeamState, qui contient la demande de l’utilisateur, les spécifications, les tests unitaires, le brouillon de code actuel, les résultats des tests ainsi que le nombre de tours de débogage. - Les nœuds représentent les membres de l’équipe. Un nœud n’est rien d’autre qu’une fonction Python qui reçoit l’état actuel, effectue sa tâche (appel d’un LLM, exécution de tests) et renvoie des mises à jour qui sont intégrées à nouveau dans l’état partagé.
StateGraph suivie de la compilation. Il peut être exécuté, et ses événements peuvent être diffusés au fur et à mesure que le flux de travail progresse.Mise en place du projet
Commencez par un répertoire vierge et un environnement virtuel. Ces commandes en terminal créent les deux éléments et activent l’environnement :
mkdir langgraph-self-healing-team
cd langgraph-self-healing-team
python3 -m venv .venv
source .venv/bin/activate
Un fichier requirements.txt énumère ensuite les dépendances :
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
Installer ces éléments dans l’environnement virtuel se fait avec une seule commande :
pip install -r requirements.txt
Chaque package a une fonction spécifique :
langgraphorchestre le flux de travail multi-agents, gérant l’état, les transferts entre nœuds ainsi que les boucles de rétroaction cycliques.langchain-google-genaise connecte aux modèles Google Gemini et expose leurs niveaux de réflexion.langchain-openaiest optionnel. Il ajoute un support pour OpenRouter ou OpenAI, permettant d’utiliser des modèles tels que GPT 5.6 ou Deepseek v4 Pro si l’on ne souhaite pas se limiter à Gemini.pytestest l’outil d’exécution des tests au sein du sandbox de sous-processus local. Il exécute le jeu de tests généré et renvoie des résultats réels.richformate la sortie du terminal avec des couleurs, des tableaux et un soulignement syntaxique, ce qui facilite la compréhension et le débogage du flux de travail des agents.python-dotenvcharge automatiquement les clés API et les paramètres des modèles depuis le fichier.env.
Le fichier .env situé dans la racine du projet contient les identifiants et la configuration du modèle :
# 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
Lorsque l’environnement est prêt, l’application peut être construite pièce par pièce.
Étape 1 : l’état partagé dans src/state.py
Avant toute autre chose, il faut définir le schéma des données transmises entre les agents. Créez un répertoire src contenant un fichier state.py à l’intérieur.
La première définition est un modèle TestResult, écrit avec Pydantic. Il représente le résultat structuré d’un exécution pytest dans le 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 enregistre si la suite de tests a réussi, le code de sortie du processus, le nombre de tests réussis et échoués, ainsi que les résultats bruts des tests. Il contient également un champ error_summary qui ne présente que la partie pertinente du suivi d’erreur, que le débogueur utilise pour identifier et corriger la panne.
Ensuite vient le AgentTeamState principal, sous forme de TypedDict. Il s’agit de la « ardoise » qui parcourt le graphe : chaque nœud lit ce dont il a besoin, effectue son travail et met à jour les champs dont dépendent les nœuds suivants.
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
Chaque champ joue un rôle dans le flux de travail :
user_requestcontient le problème initial soumis par l’utilisateur, afin que chaque agent puisse voir l’intention initiale tout au long de l’exécution.specsettestsproviennent de l’Architecte. Le Développeur met en œuvre le code conformément aux spécifications, tandis que l’Exécuteur de tests lance les tests générés sur ce code.codereprésente l’implémentation actuelle en Python. Le Développeur la remplit en premier, et le Débogueur peut la remplacer après une exécution échouée.test_resultsstocke les résultats structurés provenant du environnement de test, ce qui influence le choix entre le Reviseur et le Débogueur.iterationetmax_iterationssuivent le cycle d’auto-réparation. Ils empêchent le Débogueur de réessayer indéfiniment et imposent une limite de sécurité, par exemple trois tentatives.
debug_history conserve la cause racine ainsi que la solution apportée à chaque session de débogage, enregistrant les modifications apportées au code.review_summary reçoit le rapport final du réviseur une fois que le code a réussi la validation.status indique l’état actuel du flux de travail : génération de code, exécution des tests, débogage ou prêt pour révision.Étape 2 : niveaux de modèle et de raisonnement dans src/config.py
Une usine LLM unifiée attribue le modèle approprié ainsi que le niveau de raisonnement (bas, moyen ou élevé) à chaque rôle de sous-agent. Le fichier src/config.py commence par un dictionnaire de configuration par défaut :
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"))
Chaque rôle dispose d’un modèle et d’un niveau de réflexion correspondant à la difficulté de sa tâche, ce qui équilibre la puissance du raisonnement, la vitesse et les coûts sans nuire à la qualité globale :
architectetdebuggersont réglés surhigh, car la conception de systèmes et le diagnostic des traces d’erreurs exigent tous deux une déduction approfondie en plusieurs étapes.coderest réglé surmedium. Il n’a qu’à suivre le plan élaboré par l’Architecte, donc une synthèse équilibrée, rapide et solide suffit.reviewerest réglé surlowafin d’économiser des tokens, puisqu’il se contente d’écrire un résumé après avoir examiné les résultats.
Le même fichier contient également la fonction factory 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)
Deux éléments rendent cette fonction utile. Elle transmet directement le thinking_level (low, medium, high) à ChatGoogleGenerativeAI, ce qui permet d’ajuster l’intensité du raisonnement en fonction du rôle. De plus, chaque client qu’elle renvoie implémente BaseChatModel de LangChain, de sorte que les nœuds du graphe appellent .invoke() de la même manière, quel que soit le fournisseur derrière un rôle donné.
Étape 3 : un environnement de test déterministe avec pytest dans src/sandbox.py
Une erreur fréquente dans la conception des agents consiste à demander à un LLM de vérifier, mentalement, le code qu’il vient d’écrire. Un modèle peut prédire avec assurance que le code fonctionne correctement alors qu’il contient des erreurs, car les modèles ont tendance à halluciner en pensant que leur propre sortie est exacte. Même lorsqu’un modèle signale certaines erreurs, il n’y a aucune garantie qu’il en ait détecté toutes.
Une méthode plus fiable consiste à laisser le code se prouver lui-même. Au lieu de faire confiance au jugement du modèle, un nœud d’exécution local écrit le code généré ainsi que les tests dans un répertoire temporaire et exécute pytest au sein d’un sous-processus Python isolé. Le résultat est un verdict objectif de réussite ou d’échec qui peut être directement intégré au flux de travail.
Créez sandbox.py dans le dossier src. Il commence par une fonction d’aide, _extract_error_summary(), qui extrait les détails d’échec les plus pertinents à partir de la sortie brute de pytest, afin que le débogueur affiche uniquement la partie utile du rapport d’erreur et non l’intégralité des données :
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:])
Cet outil supprime les codes de couleur du terminal ainsi que les en-têtes de test standard, ne laissant que l’échec exact de l’assertion (assert result == expected) accompagné de la trace d’erreur. Le débogueur reçoit ainsi une description concise et pertinente de ce qui s’est passé, sans mots inutiles.
Le programme principal, run_tests_in_sandbox(), se trouve dans le même fichier :
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,
}
Quelques détails rendent cette vérification fiable, sans avoir besoin de l’avis d’un LLM :
tempfile.TemporaryDirectory()fournit un espace de travail propre à chaque exécution. Le code généré ainsi que les tests y sont écrits, et le répertoire est nettoyé à la fin de l’exécution.timeout=15limite le temps d’exécution. Le sous-processus bloqué dans une boucle infinie ou qui s’exécute trop longtemps est arrêté, ce qui empêche qu’il bloque l’ensemble du flux de travail de l’agent.exit_code == 0représente le signal déterministe de réussite ou d’échec émis parpytest. Zéro signifie que la série de tests a réussi, tandis que tout autre code indique un problème ; par conséquent, le résultat ainsi que l’historique d’erreurs sont envoyés au débogueur.
error_summary. Cette structure est stockée dans l’état partagé, et le graphe s’en sert pour décider s’il doit continuer ou entrer dans la boucle d’auto-réparation.Étape 4 : messages système destinés à chaque spécialiste dans src/prompts.py
Dans un système multi-agents, les messages doivent être courts, clairs et précis. Chaque agent a besoin d’une vision claire de sa tâche, de ses responsabilités et de ses limites, sans instructions inutiles ni chevauchements avec les agents voisins. Le fichier prompts.py situé dans src définit quatre messages système distincts pour les rôles spécialisés :
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.
"""
Ces instructions sont structurées selon deux objectifs :
- Des délimiteurs d’output explicites.
ARCHITECT_SYSTEM_PROMPTexige les sections### SPECIFICATIONSet### TESTS, ce qui permet au analyseur de séparer facilement la description de l’architecture du code de test exécutable. - Une séparation stricte des responsabilités. Le programmeur est invité à écrire uniquement du code sans de tests, tandis que le débogueur doit fournir un diagnostic concis en deux phrases à côté du bloc de code corrigé complet. Chaque sous-agent a une mission bien définie, ce qui réduit les hallucinations.
Étape 5 : création des nœuds de sous-agent dans src/nodes.py
C’est le moteur d’exécution principal du flux de travail, où sont définies les fonctions des nœuds. Un nœud LangGraph est une simple fonction Python. Elle prend en entrée l’état partagé AgentTeamState, effectue une tâche spécifique comme appeler un LLM ou exécuter des tests, puis renvoie les éléments à mettre à jour. LangGraph les fusionne dans l’état partagé, de sorte que le nœud suivant dispose toujours du contexte le plus récent.
Créez le fichier nodes.py dans le dossier src et ajoutez les nœuds un par un.
Analysateurs auxiliaires pour l’extraction du code et la séparation des spécifications
Deux petits outils aident à maintenir l’ordre dans le pipeline. Les LLMs encadrent souvent le code entre des balises Markdown, ou commencent par des introductions longues même lorsqu’on leur demande de ne pas le faire. Ces outils gèrent ce formatage afin que le reste du pipeline reçoive toujours du Python propre et exécutable. Les importations et les outils auxiliaires se trouvent en haut de src/nodes.py:
import re
from typing import Dict, Any, List
from langchain_core.messages import SystemMessage, HumanMessage
from src.state import AgentTeamState
from src.config import get_llm
from src.sandbox import run_tests_in_sandbox
from src.prompts import (
ARCHITECT_SYSTEM_PROMPT,
CODER_SYSTEM_PROMPT,
DEBUGGER_SYSTEM_PROMPT,
REVIEWER_SYSTEM_PROMPT,
)
def _get_text(content: Any) -> str:
"""
Extracts a clean string from either a standard string or list-of-dict content blocks
returned by modern LangChain message objects.
"""
if isinstance(content, str):
return content
if isinstance(content, list):
parts = []
for item in content:
if isinstance(item, str):
parts.append(item)
elif isinstance(item, dict) and "text" in item:
parts.append(str(item["text"]))
return "\n".join(parts).strip()
return str(content).strip()
def _extract_code(text: Any) -> str:
"""
Extracts executable Python code from markdown backticks using regex.
If no backticks are found, falls back to returning the cleaned raw text.
"""
clean_text = _get_text(text)
match = re.search(r"```(?:python)?\s*\n(.*?)\n```", clean_text, re.DOTALL)
if match:
return match.group(1).strip()
return clean_text.strip()
def _parse_architect_output(raw_content: Any) -> tuple[str, str]:
"""
Splits the Architect's raw output into its two distinct deliverables:
the architectural specification text and the pytest test suite.
"""
text = _get_text(raw_content)
specs = text
tests = ""
if "### TESTS" in text:
parts = text.split("### TESTS")
specs = parts[0].replace("### SPECIFICATIONS", "").strip()
tests = _extract_code(parts[1])
elif "```python" in text:
tests = _extract_code(text)
specs = text.split("```python")[0].strip()
return specs, tests
Chaque outil gère un problème courant dans les sorties des LLM :
_get_text()normalise le format de la réponse. Selon le fournisseur et l’encapsulation LangChain, une réponse peut arriver sous forme de chaîne de caractères simple ou de contenu structuré tel qu’une liste contenant des élémentstypeettexte. Cet outil transforme les deux formes en une chaîne de caractères Python ordinaire._extract_code()extrait le code à partir des balises Markdown. Il utilise une expression régulière avecre.DOTALLafin de capturer le code multi-lignes, que le modèle ait écritpythonaprès les guillemets inversés d’ouverture ou non. En l’absence de bloc balisé, il se réfère au texte de réponse déjà nettoyé.
_parse_architect_output() sépare la réponse de l’Architecte au niveau du délimiteur ### TESTS. Tout ce qui se trouve avant ce marqueur devient specs, et tout ce qui suit devient tests. Cela permet de séparer la spécification rédigée en langage naturel du code de test exécutable qui sera ensuite transmis au Test Runner.Le nœud Lead Architect
Le Lead Architect constitue le point d’entrée de l’équipe, et il travaille selon une approche test-first. Avant même toute implémentation, il étudie la demande de l’utilisateur, définit les structures de données et les interfaces de classe, puis conçoit une suite complète de tests pytest couvrant à la fois les scénarios attendus et les cas limites importants. Cela offre au Développeur un contrat clair à respecter lors de l’implémentation.
La fonction est ajoutée au même fichier :
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",
}
Trois éléments se distinguent dans le code :
- Sélection du modèle.
get_llm("architect")configuregemini-pro-latestavec un niveau de réflexionhigh. Un modèle puissant capable d’un raisonnement approfondi s’avère ici très utile, car la tâche consiste à planifier en fonction des exigences et à anticiper des cas limites tels que la sécurité des threads, des collections vides ou des conditions de bord. - Construction du message. L’invite système
ARCHITECT_SYSTEM_PROMPTest placée à l’intérieur d’unSystemMessage, tandis que l’invite brute de l’utilisateur provenant destate['user_request']devient unHumanMessage. - Mises à jour de l’état. Le résultat est un dictionnaire contenant
specs,testsetstatus = "specs_ready"; LangGraph regroupe ces trois éléments dansAgentTeamState.
Le nœud Développeur
Lorsque la spécification et les tests sont prêts, le contrôle est transféré au Développeur. Son rôle consiste à écrire du Python propre et idiomatique qui met en œuvre la spécification et satisfait à tous les tests rédigés par l’Architecte. La fonction est placée dans src/nodes.py sous le nom de 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",
}
Fonctionnement :
- Sélection du modèle.
get_llm(\"coder\")choisitgemini-3.7-flashavec un mode de réflexionmedium. L’Architecte a déjà défini le schéma de l’API ainsi que les assertions de test, ce qui permet au Développeur de ne pas avoir à redessiner quoi que ce soit et de se concentrer sur un code rapide et précis. Un effort modéré suffit à cet effet, tout en réduisant l’utilisation de tokens et les coûts de développement. - Ancrage dans le jeu de tests.
state['tests']est injecté directement dans la requête du Développeur. En voyant précisément les assertions à respecter, le modèle est orienté vers les noms de méthodes, les types d’arguments et les valeurs de retour attendus. - Mises à jour de l’état. Après avoir extrait le code Python brut de la réponse, le nœud renvoie
{"code": code, "status": "code_generated"}.
Le nœud de sandbox de test déterministe
Ce est l’un des nœuds les plus importants du système. Au lieu de demander à un LLM si le code est correct, il exécute la suite pytest sur ce dernier dans un sous-processus isolé. Les tests s’exécutent localement, de sorte que cette étape ne génère aucune appel à l’API de l’LLM et se termine généralement en environ 15 millisecondes.
La fonction test_runner_node() est ajoutée à 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,
}
- Déterminisme à zéro token. La fonction
run_tests_in_sandbox()reçoitstate["code"]etstate["tests"]. Aucun modèle n’est appelé, donc cette étape ne coûte aucun token. - Détermination de l’état. Un résultat où
test_results["passed"] == True(ce qui signifieexit_code == 0) définitstatussur"tests_passed", tandis que tout autre résultat le définit sur"tests_failed". - Sortie d’état structurée. Les
test_resultsretournés regroupent les résultats de l’exécution, le nombre de tests réussis et échoués, ainsi qu’un résumé condensé des traces d’erreur. C’est ce que le mécanisme conditionnel examine lors du routage.
Le nœud débogueur auto-réparateur
Lorsque le sandbox signale des échecs, les flux de contrôle sont dirigés vers le Débogueur, l’expert en causes racines. Celui-ci examine trois éléments : le code défectueux, les assertions de test et le traceur d’erreur précis de pytest, afin de déterminer pourquoi l’exécution a échoué, puis applique une correction ciblée. La fonction debugger_node() se trouve dans 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",
}
Comportement :
- Allocation élevée pour le raisonnement.
get_llm("debugger")renvoiegemini-3.7-flashavec un niveau de réflexion réglé surhigh. Lire un trace d’erreur signifie suivre le chemin d’exécution, identifier les cas limites problématiques et analyser comment les variables ont changé, ce qui exige un effort réel. Ce modèle a été choisi comme compromis entre profondeur et vitesse. Un modèle professionnel fonctionne également bien pour cette couche, et le flux pourrait même changer de modèle en cours d’exécution si le débogueur reste bloqué après une ou deux boucles. - Contexte d’erreur détaillé. Seul le
error_summarysimplifié parvient au modèle, par exempleAssertionError: assert False == True, where False = allow_request('user_1').
current_iteration = state.get("iteration", 0) + 1, chaque passage par le débogueur incrémente le compteur de tentatives d’un.debug_history.state['code'] est remplacé par patched_code, et le nœud indique status = "code_patched".Le nœud Reviewer
Après chaque test réussi, ou lorsque la limite de tentatives est atteinte, le contrôle passe au Reviewer, la dernière étape de contrôle de qualité. Il vérifie la qualité globale du code, s’assure que les docstrings sont claires et utiles, et rédige un bref résumé de la solution finale pour l’utilisateur. La fonction reviewer_node() est ajoutée à 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",
}
Fonctionnalités :
- Sélection du modèle.
get_llm("reviewer")renvoiegemini-flash-lite-latestavec un niveau de réflexionlow, adapté à la rédaction rapide de synthèses. Peu de raisonnement est nécessaire à ce stade, donc un modèle plus léger ayant une capacité de réflexion moindre offre une meilleure vitesse et des coûts réduits. - Revue contextuelle. Le revueur reçoit l’implémentation finale, les chiffres de réussite et d’échec provenant du sandbox, le nombre d’itérations de débogage utilisées ainsi que l’historique complet des débogages.
- Mises à jour d’état. Le nœud renvoie
{"review_summary": _get_text(response.content), "status": "completed"}.
Étape 6 : logique de routage dans src/edges.py
La logique de base et les nœuds d’agent sont prêts, il ne reste plus que le routage. Les nœuds effectuent les calculs, tandis que les arêtes déterminent où se dirige le contrôle. Une arête conditionnelle agit comme un routeur dynamique : elle examine le dictionnaire d’état retourné par un nœud et répond par le nom du prochain nœud à exécuter.
La fonction de routage se trouve dans un nouveau fichier edges.py à l’intérieur de src:
from typing import Literal
from src.state import AgentTeamState
def route_after_test_runner(state: AgentTeamState) -> Literal["reviewer", "debugger"]:
"""
Conditional edge router evaluated immediately after `test_runner` completes.
Decides whether to trigger the self-healing Debugger loop or proceed to Reviewer.
"""
test_results = state.get("test_results", {})
passed = test_results.get("passed", False)
iteration = state.get("iteration", 0)
max_iterations = state.get("max_iterations", 3)
# 1. Happy Path: 100% of unit tests passed -> Proceed directly to Review
if passed:
return "reviewer"
# 2. Self-Healing Loop: Tests failed and within retry budget -> Trigger Debugger
if iteration < max_iterations:
return "debugger"
# 3. Safety Fallback: Exceeded maximum retries -> Exit loop to Reviewer with warning
return "reviewer"
Le routeur examine les résultats des tests ainsi que le nombre d’essais de débogage effectués jusqu’à présent, et il gère trois cas :
- Parcours sans problème (
passed == True). Lorsque tous les tests réussissent dès la première tentative, le flux de travail est directement transmis au Reviseur, sans nécessité d’investir du temps ou des ressources supplémentaires dans le Débogueur. - Réparation automatique (
passed == Falsetant queiteration < max_iterations). Les tests échoués, tant qu’il reste des tentatives disponibles, envoient le code au Débogueur, qui analyse l’échec et tente de trouver une solution. - Dépassement du nombre maximal de tentatives (
iteration >= max_iterations). Lorsque le Débogueur a épuisé les trois tentatives autorisées et que les tests continuent d’échouer, le flux de travail est transmis au Reviseur plutôt que de tourner en boucle indéfiniment. Ce plafond strict imposé à la réparation automatique empêche le système de bloquer face à des exigences contradictoires ou ambiguës.
Étape 7 : assemblage du graphe dans src/graph.py
Les nœuds et le routage ont été configurés, il ne reste plus qu’à relier tout le système. Des liens directs et des arêtes de feedback conditionnel connectent les nœuds, ce qui permet de compiler un StateGraph capable d’exécuter l’ensemble du flux de travail. Placez ce code dans un nouveau fichier graph.py à l’intérieur de src:
from langgraph.graph import StateGraph, END
from src.state import AgentTeamState
from src.nodes import (
architect_node,
developer_node,
test_runner_node,
debugger_node,
reviewer_node,
)
from src.edges import route_after_test_runner
def build_self_healing_graph() -> StateGraph:
"""
Assembles the Multi-Agent Self-Healing Engineering Team into a LangGraph StateGraph.
Configures node registrations, linear handoffs, cyclic feedback edges, and conditional routing.
"""
# 1. Initialize StateGraph with our Shared Whiteboard State Schema
workflow = StateGraph(AgentTeamState)
# 2. Register Sub-Agent Nodes (Mapping node names to Python callable functions)
workflow.add_node("architect", architect_node)
workflow.add_node("developer", developer_node)
workflow.add_node("test_runner", test_runner_node)
workflow.add_node("debugger", debugger_node)
workflow.add_node("reviewer", reviewer_node)
# 3. Define Graph Entry Point
workflow.set_entry_point("architect")
# 4. Add Deterministic Direct Handoff Edges
workflow.add_edge("architect", "developer")
workflow.add_edge("developer", "test_runner")
# 5. Add the Cyclic Self-Healing Feedback Edge
workflow.add_edge("debugger", "test_runner")
# 6. Add Conditional Routing Edge for Sandbox Test Evaluation
workflow.add_conditional_edges(
"test_runner",
route_after_test_runner,
{
"debugger": "debugger",
"reviewer": "reviewer",
},
)
# 7. Add Termination Edge
workflow.add_edge("reviewer", END)
# 8. Compile the Graph into an Executable State Machine
return workflow.compile()
La construction s’effectue pas à pas :
StateGraph(AgentTeamState)crée un graphe à états en utilisant le schémaAgentTeamState, de sorte que chaque nœud peut atteindre le même état partagé à mesure que le flux de travail progresse.workflow.add_node(name, func)enregistre chaque fonction Python en tant que nœud. Chacun a une seule responsabilité : concevoir des tests, générer du code, exécuter des tests ou déboguer les erreurs.workflow.set_entry_point("architect")définit le point de départ. L’appel ou la lecture du graphe avec un état initial démarre l’exécution au nœudarchitect.
workflow.add_edge(\"architect\", \"developer\") et workflow.add_edge(\"developer\", \"test_runner\") créent la ligne droite initiale. Lorsque l’Architecte a produit la spécification et les tests, le Développeur prend le relais, et une fois que le Développeur a écrit le code, c’est au Tour du Testeur.workflow.add_edge(\"debugger\", \"test_runner\") ferme le cycle d’autoguérison, ce qui transforme un simple pipeline en un système autoguérissant. Une erreur corrigée ne fait pas passer le workflow au Réviseur. Le code réparé revient au Testeur afin que la correction soit à nouveau validée.workflow.add_conditional_edges("test_runner", route_after_test_runner, ...) fait en sorte que LangGraph appelle route_after_test_runner(state) une fois les tests terminés. Sa valeur de retour envoie le flux de travail soit au debugger, soit au reviewer.workflow.add_edge("reviewer", END) termine l’exécution. Une fois que le reviewer a achevé sa vérification de qualité et rédigé son résumé, l’exécution du graphe est terminée.workflow.compile() transforme la définition du graphe en un graphe exécutable, en validant sa structure et en le préparant à l’exécution.Étape 8 : le point d’entrée de la CLI dans src/main.py
Le dernier élément est un point d’entrée en ligne de commande construit avec rich, qui affiche en temps réel les progrès de chaque sous-agent à l’aide de panneaux de terminal stylisés. Le fichier main.py situé dans src commence par les fonctions d’affichage de la bannière et du tableau de configuration :
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()
Ces éléments concernent exclusivement le stylage de la console et sont évidents en eux-mêmes, il n’est donc pas nécessaire d’en donner une explication détaillée. Le même fichier se poursuit avec 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() est l’endroit où tous les éléments se rejoignent. Il configure le AgentTeamState partagé, crée le graphe auto-régénérant et exécute les nœuds un par un. Lorsque chaque nœud est terminé, il met à jour l’état actuel et affiche les progrès dans des panneaux Rich : la spécification et les tests de l’Architecte, le code du Développeur, les résultats du Test Runner, chaque étape de débogage ainsi que le rapport final du Réviseur. En fin de processus, il affiche l’implémentation vérifiée, la suite pytest générée ainsi que des métriques d’exécution telles que le temps total, le nombre d’itérations de débogage et les résultats des tests.
Le parseur d’arguments en ligne de commande termine le fichier :
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()
L’exécution avec un limiteur de débit
Pour un test réel, l’équipe doit résoudre une tâche algorithmique non triviale. Cette commande demande un limiteur de débit à fenêtre glissante sécurisé pour les threads :
python src/main.py --prompt "Build an in-memory Sliding Window Rate Limiter class in Python with thread safety, per-user limits, and a method allow_request(user_id: str) -> bool."
La sortie du terminal montre la séquence suivante :
- Pendant l’étape 1 sur 5, l’architecte en chef analyse les exigences relatives au limiteur de débit et crée une suite
pytestmultithreadée couvrant le tri des timestamps, l’application de seuils via une fenêtre glissante ainsi que les requêtes simultanées provenant de plusieurs threads. - Pendant l’étape 2 sur 5, le développeur écrit une première classe
SlidingWindowRateLimiterbasée surcollections.dequeet unthreading.Lock. - Pendant l’étape 3 sur 5, le lanceur de tests lors de l’itération 0 exécute
pytest. Le test de charge en concurrence échoue car le tri des timestamps a eu lieu en dehors de la section critique verrouillée, ce qui a entraîné une condition de course permettant à plusieurs threads de dépasser le quota de requêtes autorisé.
_prune_expired_timestamps() a été appelé avant que self._lock ne soit acquis, et que la correction consiste à envelopper l’élimination des timestamps dans un bloc synchronisé.pytest sur le code modifié, et le jeu de tests affiche un résultat positif avec 6 tests réussis en 0,04 s.Une capture d’écran de la sortie réelle de la terminal lors de cette exécution :
La sortie complète inclut également le code Python écrit et les cas de test utilisés pour le vérifier. La reproduire ici alourdirait le texte sans apporter beaucoup de valeur.
Où ce modèle s’applique
En combinant des sous-agents spécialisés, chacun doté d’un modèle et d’un niveau de réflexion différents, ainsi qu’un environnement d’exécution déterministe, on transforme un flux de travail fragile basé sur une seule instruction en un système plus fiable et capable de se corriger lui-même.
La démonstration a porté sur une équipe de génération de code, mais le même schéma de machine à états convient à de nombreux autres flux de travail, tels que les pipelines de données, l’orchestration dynamique d’API et la validation de documents structurés. L’idée fondamentale est simple : ne pas compter sur un seul modèle pour tout faire. Donner à chaque agent une responsabilité bien définie et permettre au système de vérifier et de corriger son propre travail au fur et à mesure.
Trois leçons ressortent clairement :
- N’ayez jamais recours à un LLM pour évaluer son propre code. Confiez l’exécution à un environnement de sous-processus déterministe local exécutant
pytest, qui est rapide, gratuit et fournit une réponse claire de réussite ou d’échec.