Selbstheilendes Mehr-Agenten-Team in LangGraph: Architekt, Tester, Debugger
Erstellen Sie eine LangGraph-State-Machine in Python, bei der ein Architekt, ein Entwickler, eine pytest-Sandbox, ein Debugger und ein Reviewer zusammenarbeiten und im Kreislauf vorgehen, bis die Tests bestehen.
Quellen: Build a Self-Healing Multi-Agent Team with LangGraph
KI ist heute in nahezu jeder Schicht der Softwareentwicklung vorhanden – von der Erstellung von Code über die Datenanalyse bis hin zum mehrstufigen Denken. Doch jeder, der ein großes Sprachmodell um etwas Komplexes bittet, stößt auf dasselbe Problem. Eine einmalige Antwort mag auf den ersten Blick korrekt erscheinen, bricht jedoch zusammen, sobald sie den Anforderungen sowie einigen Randfällen gegenübersteht. Die übliche Lösung ist ein manueller Prozesszyklus: Der Fehlerprotokoll wird kopiert, in das Chat-Fenster eingefügt, eine Korrektur angefordert und dieser Vorgang wiederholt, bis alles endlich funktioniert.
Der gegenteilige Ansatz hat ebenfalls seine Probleme. Wenn einem monolithischen Agenten Dutzende von Tools sowie eine lange Liste an Anweisungen übergeben werden, steigt in der Regel die Fehlerquote, da sein Kontext überladen wird und die Signaturen der Tools miteinander verschwimmen. Die Qualität der Ausgabe sinkt, da der Agent nicht alle Anweisungen gleichzeitig befolgen kann.
Das unten dargestellte Design löst beide Probleme mithilfe eines modularen, hierarchischen Mehr-Agenten-Zustandsautomaten, der in Python mit LangGraph geschrieben wurde. Das laufende Beispiel ist ein selbstheilendes Softwareentwicklerteam, das aus fünf Rollen besteht: einem Architekten, einem Entwickler, einer auf Unterprozessen basierenden Pytest-Sandbox, einem Debugger und einem Reviewer. Dieses Muster lässt sich auf viele andere AI-Engineering-Arbeitsabläufe übertragen.
Durch die Aufteilung einer großen Aufgabe auf spezialisierte Sub-Agenten, bei denen jeweils die richtige Menge an Logik zur Verfügung steht, das Ausführen der Ergebnisse in einer kontrollierten Sandbox sowie das Hinzufügen eines stetigen Feedback-Zyklus kann das Team seine eigene Arbeit testen, Probleme erkennen und beheben, bevor es ein überprüftes Ergebnis weitergibt. Zunächst ein Blick auf die Architektur.
Zwei Prinzipien prägen das System.
Erstens gibt es die heterogene Modellstufung mit anpassbaren Denkstufen. Die Rollen unterscheiden sich in ihrer Schwierigkeit, daher sollten sie nicht alle dasselbe Modell oder denselben Rechenbudget erhalten:
- Der Architekt arbeitet mit
gemini-pro-latestund einerhohenDenkstufe. Er übernimmt die anspruchsvollsten Aufgaben im Bereich des logischen Denkens, wie das Festlegen von Problemgrenzen, das Antizipieren von Ressourcenkonflikten sowie das Entwerfen strengerpytest-Testsätze. - Der Entwickler verwendet
gemini-3.7-flashmit einermittlerenDenkstufe. Es handelt sich um eine ausgewogene Konfiguration für eine schnelle Codegenerierung, die den expliziten Vorgaben des Architekten folgt. - Der Testausführer ist ein deterministischer Python-Unterprozess, der
pytestlokal in etwa 15 Millisekunden ausführt und keinerlei LLM-API-Tokens verbraucht.
gemini-3.7-flash mit hohem Denkgrad, da die Rückverfolgung der Roh-Stack-Traces bis zur fehlerhaften Zeile eine kausale Ursachenanalyse erfordert.gemini-flash-lite-latest mit niedrigem Denkgrad. Die Überprüfung des Stils sowie das Verfassen des abschließenden Zusammenfassungsberichts für den Benutzer erfordern nur geringe logische Überlegungen.Das zweite Prinzip ist eine strenge Einschränkung der Tools, d.h. das Anwenden von Mindestrechten auf die Agenten. Kein Unteragent erhält uneingeschränkten Zugriff auf das Terminal und das Dateisystem. Jeder erhält nur die Eingaben und Tools, die seine Rolle erfordert. Der Entwickler kann keine willkürlichen Shell-Befehle ausführen, und der Testausführender führt lediglich die Tests aus und berichtet darüber, was tatsächlich geschehen ist, ohne von einem LLM zu verlangen, das Ergebnis zu interpretieren oder zu erfinden. Eine enge Einschränkung hält jeden Agent fokussiert und verringert Halluzinationen.
Die involvierten LangGraph-Konzepte
LangGraph funktioniert wie eine gemeinsame Tafel. Jedes Teammitglied, also jeder Sub-Agent, erledigt seine zugewiesene Aufgabe und schreibt das Ergebnis wieder in diesen gemeinsamen Raum, sodass alle den bisher von dem Team erstellten Kontext sehen können. LangGraph ist der Orchesterer: Es entscheidet, welcher Agent als Nächstes ausgeführt wird, und übermittelt den entsprechenden Zustand weiter.
Vier Aspekte sind hier wichtig:
- Der Zustand, ein
TypedDict, stellt die gemeinsame Tafel dar. Alle Agenten lesen von und schreiben in ein Python-Dictionary namensAgentTeamState, das die Benutzereingabe, die Spezifikationen, die Unit-Tests, den aktuellen Codeentwurf, die Testergebnisse sowie die Anzahl der Debug-Runden enthält. - Die Knoten sind die Teammitglieder. Ein Knoten ist einfach eine Python-Funktion, die den aktuellen Zustand erhält, ihre Aufgabe ausführt (einen LLM aufruft, Tests durchführt) und Aktualisierungen zurückgibt, die wieder in den gemeinsamen Zustand eingefügt werden.
StateGraph registriert und kompiliert wurden. Es kann ausgeführt werden, und seine Ereignisse können im Laufe des Workflows gestreamt werden.Projekt einrichten
Beginnen Sie mit einem neuen Verzeichnis und einer virtuellen Umgebung. Diese Terminalbefehle erstellen beides und aktivieren die Umgebung:
mkdir langgraph-self-healing-team
cd langgraph-self-healing-team
python3 -m venv .venv
source .venv/bin/activate
Die Datei requirements.txt listet anschließend die Abhängigkeiten auf:
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
Die Installation in die virtuelle Umgebung erfolgt mit einem Befehl:
pip install -r requirements.txt
Jedes Paket hat eine bestimmte Aufgabe:
langgraphkoordiniert den Workflow mit mehreren Agenten, kümmert sich um den Zustand, die Übertragung zwischen Knoten sowie zyklische Rückkopplungsschleifen.langchain-google-genaistellt eine Verbindung zu den Google Gemini-Modellen her und zeigt deren Denkstufen an.langchain-openaiist optional. Es fügt Unterstützung für OpenRouter oder OpenAI hinzu, sodass Modelle wie GPT 5.6 oder Deepseek v4 Pro eingesetzt werden können, falls eine ausschließliche Nutzung von Gemini nicht gewünscht ist.pytestist der Testausführer innerhalb des lokalen Subprozess-Sandboxes. Er führt die generierten Tests durch und gibt echte Ergebnisse zurück.richformatiert die Terminalausgabe mit Farben, Tabellen und Syntaxhervorhebung, wodurch der Agent-Workflow leichter verfolgt und debuggt werden kann.python-dotenvlädt API-Schlüssel sowie Modell-Einstellungen automatisch aus der.env-Datei.
Die .env-Datei im Projektverzeichnis enthält Anmeldeinformationen sowie Konfigurationen für das Modell:
# 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
Sobald die Umgebung bereit ist, kann die Anwendung Schritt für Schritt erstellt werden.
Schritt 1: Der gemeinsame Zustand in src/state.py
Zuerst muss das Schema für die zwischen den Agenten ausgetauschten Daten definiert werden. Erstellen Sie ein Verzeichnis src mit einer Datei state.py darin.
Die erste Definition ist ein TestResult-Modell, das mit Pydantic geschrieben wurde. Es repräsentiert das strukturierte Ergebnis eines pytest-Laufs im Sandbox-Umfeld:
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 dokumentiert, ob der Testzyklus erfolgreich war, den Abbruchcode des Prozesses, wie viele Tests bestanden und fehlgeschlagen sind sowie die Rohausgabe der Tests. Zudem enthält es ein error_summary-Feld mit nur dem relevanten Teil der Fehlerverfolgung, den der Debugger zur Identifizierung und Behebung des Fehlers verwendet.
Danach folgt der Hauptwert AgentTeamState als TypedDict. Dies ist das „Whiteboard“, das sich durch das Graphen bewegt: Jeder Knoten liest das, was er benötigt, führt seine Aufgabe aus und aktualisiert die Felder, von denen die nachfolgenden Knoten abhängen.
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
Jedes Feld hat eine Rolle im Workflow:
user_requestenthält das ursprüngliche Problem des Benutzers, sodass jeder Agent die anfängliche Absicht während des gesamten Ablaufs einsehen kann.specsundtestsstammen vom Architekten. Der Entwickler implementiert gemäß den Spezifikationen, und der Testausführender führt die generierten Tests am Code durch.codeist die aktuelle Python-Implementierung. Zunächst füllt der Entwickler sie aus, und der Debugger kann sie nach einem fehlgeschlagenen Ablauf ersetzen.test_resultsspeichert das strukturierte Ergebnis aus dem Sandbox, welches die Entscheidung zwischen Überprüfer und Debugger bestimmt.iterationundmax_iterationsüberwachen den selbstheilenden Zyklus. Sie verhindern, dass der Debugger endlos erneut versucht, und legen eine Sicherheitsgrenze fest, beispielsweise drei Versuche.
debug_history speichert die ursächliche Ursache sowie die Lösung jeder Debugging-Runde sowie einen Verlauf der Codeänderungen.review_summary erhält den endgültigen Bericht des Prüfers, sobald der Code die Validierung bestanden hat.status zeigt an, in welcher Phase sich der Workflow derzeit befindet: Codegenerierung, Testausführung, Debugging oder Bereit für die Überprüfung.Schritt 2: Modellstufen und Denklevel in src/config.py
Eine einheitliche LLM-Fabrik weist jeder Sub-Agenten-Rolle das richtige Modell sowie das entsprechende Denklevel (low, medium oder high) zu. Die Datei src/config.py beginnt mit einem Standardkonfigurationswörterbuch:
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"))
Jede Rolle erhält ein Modell sowie ein Denklevel, die der Schwierigkeit ihrer Aufgabe entsprechen und dabei die Stärke des logischen Denkens, Geschwindigkeit und Kosten ausbalancieren, ohne die Gesamtqualität zu beeinträchtigen:
architectunddebuggersind aufhigheingestellt, da sowohl das Systemdesign als auch die Diagnose von Stack-Traces tiefgehende, mehrschrittige Schlussfolgerungen erfordern.coderist aufmediumeingestellt. Er muss lediglich dem von dem Architect erstellten Plan folgen, weshalb ein ausgewogener, schneller und fundierter Ansatz ausreicht.reviewerist aufloweingestellt, um Tokens zu sparen, da er lediglich nach Überprüfung der Ergebnisse eine Zusammenfassung schreibt.
Das gleiche Datei enthält außerdem die get_llm()-Fabrik:
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)
Zwei Faktoren machen diese Funktion nützlich. Sie übermittelt den thinking_level (low, medium, high) direkt an ChatGoogleGenerativeAI, sodass die Rechenlast je nach Rolle angepasst werden kann. Zudem implementiert jeder von ihr zurückgegebene Client LangChains BaseChatModel, wodurch Graphenknoten unabhängig davon, welcher Anbieter hinter einer Rolle steht, auf dieselbe Weise .invoke() aufrufen.
Schritt 3: ein deterministischer pytest-Sandbox in src/sandbox.py
Ein häufiger Fehler bei der Konzeption von Agenten besteht darin, ein LLM dazu aufzufordern, im Kopf den gerade geschriebenen Code zu überprüfen. Ein Modell kann zuversichtlich vorhersagen, dass der Code funktioniert, obwohl darin Fehler vorhanden sind, da Modelle regelmäßig falsch annehmen, ihr eigenes Ausgabe sei korrekt. Selbst wenn Fehler erkannt werden, gibt es keine Garantie, dass alle gefunden wurden.
Ein zuverlässigerer Ansatz besteht darin, dem Code die Möglichkeit zu geben, sich selbst zu beweisen. Anstatt das Urteil des Modells zu vertrauen, schreibt ein lokaler Ausführungsknoten den generierten Code sowie die Tests in einen temporären Ordner und führt pytest in einem isolierten Python-Subprozess aus. Das Ergebnis ist ein objektives Bestehen oder Scheitern, das direkt in den Workflow zurückgeführt werden kann.
Erstellen Sie sandbox.py im Ordner src. Er beginnt mit einem Hilfsfunktion, _extract_error_summary(), die die relevantesten Fehlerdetails aus der Rohausgabe von pytest extrahiert, sodass der Debugger nur den nützlichen Teil des Stacktraces und nicht die gesamte Ausgabe sieht:
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:])
Der Hilfsprogramm entfernt die Terminal-Farbcodes sowie die Standard-Test-Überschriften und lässt nur den genauen Fehler der Assertion (assert result == expected) zusammen mit dem Stacktrace übrig. Dadurch erhält der Debugger eine kurze, prägnante Beschreibung des Fehlers, ohne überflüssige Informationen.
Der Hauptausführer, run_tests_in_sandbox(), befindet sich im selben Datei:
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,
}
Mehrere Details sorgen dafür, dass diese Überprüfung ohne jegliche Meinung eines LLM vertrauenswürdig ist:
tempfile.TemporaryDirectory()stellt für jeden Ausführungsvorgang einen neuen Arbeitsbereich zur Verfügung. Der generierte Code sowie die Tests werden dort geschrieben, und der Verzeichnisinhalt wird am Ende der Ausführung gelöscht.timeout=15begrenzt die Ausführungsdauer. Code, der in einer endlosen Schleife steckt oder zu lange läuft, wird beendet, sodass er den gesamten Agenten-Ablauf nicht blockieren kann.exit_code == 0ist das eindeutige Erfolg-/Fehler-Signal vonpytest. Eine Null bedeutet, dass der Testzyklus erfolgreich war, während jeder andere Wert auf ein Problem hindeutet; daher werden das Ergebnis sowie der Stacktrace an den Debugger übermittelt.
error_summary. Diese Struktur befindet sich im gemeinsamen Zustand, und das Diagramm nutzt sie, um zu entscheiden, ob weitergefahren werden soll oder der Selbstheilungszyklus gestartet wird.Schritt 4: Systemanfragen für jeden Spezialisten in src/prompts.py
In einem Mehr-Agenten-System sollten die Anfragen kurz, präzise und eng gefasst sein. Jeder Agent benötigt ein klares Bild von seiner Aufgabe, seinen Pflichten und Grenzen – ohne überflüssige Anweisungen und ohne Überschneidungen mit seinen Nachbarn. Die Datei prompts.py in src definiert vier unterschiedliche Systemanfragen für die spezialisierten Rollen:
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.
"""
Die Anweisungen sind mit zwei Zielen strukturiert:
- Eindeutige Trennzeichen für die Ausgabe.
ARCHITECT_SYSTEM_PROMPTverlangt die Abschnitte### SPECIFICATIONSund### TESTS, wodurch der Parser die Architekturbeschreibung leicht vom ausführbaren Testcode trennen kann. - Strakte Trennung der Aufgabenbereiche. Dem Programmierer wird mitgeteilt, nur Code und keine Tests schreiben zu sollen, während der Debugger neben dem vollständigen reparierten Codeblock eine prägnante Diagnose in zwei Sätzen liefern muss. Jeder Unteragent hat einen klaren Aufgabenbereich, was Halluzinationen verringert.
Schritt 5: Erstellung der Unteragenten-Node in src/nodes.py
Dies ist der Kern-Exekutionsmotor des Workflows, in dem die Funktionsweise der Knoten definiert wird. Ein LangGraph-Knoten ist eine einfache Python-Funktion. Sie nimmt den gemeinsamen AgentTeamState entgegen, führt eine bestimmte Aufgabe aus – wie zum Beispiel den Aufruf eines LLMs oder das Ausführen von Tests – und gibt die zu aktualisierenden Teile zurück. LangGraph fügt diese wieder in den gemeinsamen Zustand ein, sodass der folgende Knoten stets über den neuesten Kontext verfügt.
Erstellen Sie nodes.py im Verzeichnis src und fügen Sie die Knoten nacheinander hinzu.
Hilfsprogramme zur Codeextraktion und Spezifikationstrennung
Zwei kleine Hilfsprogramme sorgen dafür, dass der Pipeline-Fluss ordentlich bleibt. LLMs umschließen den Code oft in Markdown-Fence-Strukturen oder beginnen mit langatmigen Einleitungen, selbst wenn ihnen das verboten wurde. Die Hilfsprogramme kümmern sich um diese Formatierungen, damit der Rest des Pipelines stets sauberen, ausführbaren Python-Code erhält. Die Imports und Hilfsfunktionen befinden sich am Anfang von 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
Jeder Helper behebt eine häufige Besonderheit in den Ausgaben von LLMs:
_get_text()normalisiert das Antwortformat. Je nach Anbieter und LangChain-Wrapper kann die Antwort als einfacher String oder als strukturierte Inhalte wie eine Liste mit den Einträgentypeundtextauftreten. Der Helper wandelt beides in einen gewöhnlichen Python-String um._extract_code()extrahiert Code aus Markdown-Fence-Blöcken. Dabei wird eine Regex mitre.DOTALLverwendet, sodass mehrzeiliger Python erfasst wird – unabhängig davon, ob das Modell nach den öffnenden Backticks „python“ geschrieben hat. Falls kein solcher Block vorhanden ist, greift der Helper auf den bereinigten Antworttext zurück.
_parse_architect_output() teilt die Antwort des Architekten am Trennzeichen ### TESTS auf. Alles vor diesem Marker wird als specs bezeichnet, und alles danach als tests. Dadurch werden die in Natursprache formulierten Spezifikationen vom später an den Testläufer übergebenen, ausführbaren Testcode getrennt.Der Lead Architect-Node
Der Lead Architect ist der Eingangspunkt des Teams und arbeitet testbasiert. Bevor eine Implementierung existiert, untersucht er die Anfrage des Benutzers, definiert die Datensstrukturen und Klassenschnittstellen sowie entwirft ein umfassendes pytest-Testpaket, das sowohl die erwarteten erfolgreichen Abläufe als auch wichtige Randfälle abdeckt. Dadurch erhält der Entwickler einen klaren Vertrag zur Umsetzung.
Die Funktion wird in derselben Datei hinzugefügt:
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",
}
Drei Aspekte heben sich im Code hervor:
- Modellauswahl.
get_llm("architect")konfiguriertgemini-pro-latestmithighThinking. Ein leistungsstarkes Modell mit tiefgehendem Reasoning ist hier vorteilhaft, da es um die Planung auf der Grundlage von Anforderungen sowie um die Antizipation von Randfällen wie Thread-Sicherheit, leeren Sammlungen oder Grenzbedingungen geht. - Nachrichtenkonstruktion. Der Systemprompt
ARCHITECT_SYSTEM_PROMPTwird in eineSystemMessageeingebettet, während der rohe Benutzerprompt ausstate['user_request']zu einerHumanMessagewird. - Zustandsaktualisierung. Das Ergebnis ist ein Dictionary mit den Schlüsseln
specs,testssowiestatus = "specs_ready"; LangGraph fügt alle drei Schlüssel inAgentTeamStatezusammen.
Der Developer-Node
Sobald die Spezifikation und die Tests vorhanden sind, geht die Kontrolle an den Entwickler. Seine Aufgabe besteht darin, sauberen, idiomatischen Python-Code zu schreiben, der die Spezifikation umsetzt und alle von dem Architekten erstellten Tests erfüllt. Die Funktion wird in src/nodes.py als developer_node() platziert:
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",
}
So funktioniert es:
- Modellauswahl.
get_llm("coder")führt zugemini-3.7-flashmitmedium-Nivau beim Denken. Der Architekt hat bereits das API-Design sowie die Testanforderungen festgelegt, sodass der Entwickler nichts neu entwerfen muss und sich auf schnellen, präzisen Code konzentrieren kann. Ein mittlerer Aufwand reicht dafür aus und senkt zudem den Tokenverbrauch sowie die Entwicklungskosten. - Verknüpfung mit dem Testumfeld.
state['tests']wird direkt in die Anweisung des Entwicklers eingefügt. Durch die Sichtbarkeit der genauen Anforderungen, die er erfüllen muss, wird das Modell an die erwarteten Methodennamen, Argumenttypen und Rückgabewerte gebunden. - Aktualisierung des Zustands. Nachdem der reine Python-Code aus der Antwort extrahiert wurde, gibt der Knoten
{"code": code, "status": "code_generated"}zurück.
Der deterministische Test-Sandbox-Knoten
Das ist einer der wichtigsten Knoten des Systems. Anstatt eine KI zu fragen, ob der Code in Ordnung ist, führt es die pytest-Suite gegen den Code in einem isolierten Unterprozess aus. Die Tests laufen lokal, wodurch keine Anfragen an die KI-API erfolgen und die Ausführung in der Regel in etwa 15 Millisekunden abgeschlossen ist.
Die Funktion test_runner_node() wird in src/nodes.py hinzugefügt:
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,
}
- Determinismus ohne Tokenverbrauch. Die Funktion
run_tests_in_sandbox()erhältstate[„code“]sowiestate[„tests“]. Da kein Modell aufgerufen wird, kostet dieser Schritt keinen Tokenverbrauch. - Ermittlung des Status. Ein Ergebnis von
test_results[„passed“] == True(was bedeutet, dassexit_code == 0gilt) setztstatusauf„tests_passed“, während alle anderen Fällestatusauf„tests_failed“setzen. - Gestrukturierte Ausgabe des Zustands. Die zurückgegebene Struktur
test_resultsenthält die Ausgabe der Testausführung, die Zahlen der bestandenen und fehlgeschlagenen Tests sowie eine zusammengefasste Übersicht der Stack-Traces. Genau dies wird vom konditionalen Routing-Mechanismus überprüft.
Der selbstheilende Debugger-Node
Sobald der Sandbox-Fehlerbericht vorliegt, leitet er die Steuerung an den Debugger weiter, den Spezialisten für die Ursachenforschung. Dieser untersucht drei Eingabedaten – nämlich den fehlerhaften Code, die Testanforderungen sowie den genauen pytest-Stacktrace – um herauszufinden, warum der Ausführungsvorgang fehlgeschlagen ist, und wendet anschließend eine präzise Korrektur an. Die Funktion debugger_node() befindet sich in 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",
}
So verhält es sich:
- Hohe Rechenleistung für das Denken.
get_llm("debugger")gibtgemini-3.7-flashmit der Einstellunghighzurück. Das Verfolgen eines Stack-Traces bedeutet, den Ausführungspfad nachzuvollziehen, fehlerhafte Randfälle zu finden und herauszufinden, wie sich Variablen verändert haben – was echte Anstrengung erfordert. Das Modell wurde als Kompromiss zwischen Tiefe und Geschwindigkeit ausgewählt. Ein Profi-Modell eignet sich ebenfalls für diese Ebene, und der Ablauf könnte sogar mitten im Lauf das Modell wechseln, falls der Debugger nach einem oder zwei Schleifen immer noch stecken bleibt. - Dichter Fehlerkontext. Nur der gekürzte
error_summaryerreicht das Modell, beispielsweiseAssertionError: assert False == True, where False = allow_request('user_1').
current_iteration = state.get("iteration", 0) + 1 wird der Wiederholungszähler bei jedem Durchlauf durch den Debugger um eins erhöht.debug_history hinzugefügt.state['code'] wird durch patched_code ersetzt, und der Knoten gibt status = "code_patched" an.Der Reviewer-Knoten
Nach jedem bestandenen Test oder wenn die Wiederholungsgrenze erreicht ist, geht die Steuerung zum Reviewer, der letzten Qualitätskontrolle. Er prüft die Gesamtqualität des Codes, stellt sicher, dass die Dokumentationen klar und nützlich sind, und erstellt für den Benutzer eine kurze Zusammenfassung der fertigen Lösung. Die Funktion reviewer_node() wird in src/nodes.py hinzugefügt:
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",
}
Was es tut:
- Modellauswahl.
get_llm("reviewer")liefertgemini-flash-lite-latestbei einemlow-Niveau an Denkfähigkeit, das sich für das schnelle Erstellen von Zusammenfassungen eignet. In dieser Phase ist nur wenig logisches Denken erforderlich, weshalb ein leichteres Modell mit geringerer Denkfähigkeit eine bessere Geschwindigkeit und geringere Kosten bietet. - Kontextuelle Überprüfung. Der Reviewer erhält die endgültige Implementierung, die Erfolgs- und Misserfolgszahlen aus dem Sandbox-Umfeld, die Anzahl der verwendeten Debug-Iterationen sowie die vollständige Debug-Geschichte.
- Aktualisierung des Zustands. Der Knoten gibt
{"review_summary": _get_text(response.content), "status": "completed"}zurück.
Schritt 6: Routing-Logik in src/edges.py
Die Kernlogik und die Agentenknoten sind fertig, daher kommt nun das Routing. Die Knoten führen die Berechnungen durch, während die Kanten entscheiden, wohin der Kontrollfluss geht. Eine bedingte Kante fungiert als dynamischer Router: Sie prüft das Zustandsdictionary, das ein Knoten zurückgibt, und gibt den Namen des nächsten auszuführenden Knotens an.
Die Routing-Funktion befindet sich in einer neuen Datei namens edges.py innerhalb von 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"
Der Router betrachtet die Testergebnisse sowie die Anzahl der bisherigen Debugging-Versuche und handhabt drei Fälle:
- Problemloser Ablauf (
passed == True). Wenn alle Tests beim ersten Versuch bestehen, geht der Workflow direkt an den Überprüfer, und es gibt keinen Grund, zusätzliche Zeit oder Ressourcen im Debugger zu verwenden. - Selbstheilung (
passed == Falsewährenditeration < max_iterations). Fehlende Tests, bei denen noch Versuche verfügbar sind, werden an den Debugger gesendet, der die Fehleranalyse durchführt und eine Lösung versucht. - Erreichen der Wiederholungsgrenze (
iteration >= max_iterations). Wenn der Debugger alle drei erlaubten Versuche aufgebraucht hat und die Tests weiterhin fehlschlagen, geht der Workflow an den Überprüfer, anstatt endlos im Kreis zu laufen. Diese feste Obergrenze für die Selbstheilung verhindert, dass der Prozess bei widersprüchlichen oder unklaren Anforderungen stecken bleibt.
Schritt 7: Zusammenbau des Graphen in src/graph.py
Die Knoten und die Routing-Logik sind bereits erstellt, daher werden jetzt alle Elemente miteinander verbunden. Direkte Übertragungsverbindungen sowie bedingte Rückkopplungskanten verknüpfen die Knoten, und das Ergebnis wird zu einem StateGraph kompiliert, der den gesamten Workflow ausführen kann. Legen Sie diesen Code in eine neue graph.py-Datei innerhalb von src ab:
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()
Der Aufbau erfolgt schrittweise:
StateGraph(AgentTeamState)erstellt ein zustandsbasiertes Graphenmodell mithilfe des SchemasAgentTeamState, sodass jeder Knoten im Laufe der Workflow-Entwicklung denselben gemeinsamen Zustand erreichen kann.workflow.add_node(name, func)registriert jede Python-Funktion als Knoten. Jeder dieser Knoten hat eine einzige Aufgabe: Tests entwerfen, Code generieren, Tests ausführen oder Fehler beheben.workflow.set_entry_point("architect")legt den Ausgangspunkt fest. Durch Aufruf oder Streamen des Graphen mit einem Anfangszustand beginnt die Ausführung am Knotenarchitect.
workflow.add_edge("architect", "developer") und workflow.add_edge("developer", "test_runner") bilden die anfängliche Geradlinie. Nachdem der Architekt die Spezifikation sowie Tests erstellt hat, übernimmt der Entwickler, und sobald dieser den Code fertiggestellt hat, tritt der Testausführender in Aktion.workflow.add_edge("debugger", "test_runner") schließt den selbstheilenden Kreislauf, wodurch aus einem einfachen Pipeline-System ein selbstheilendes System wird. Ein behobener Fehler führt nicht dazu, dass der Workflow an den Prüfer weitergeleitet wird. Der reparierte Code kehrt zum Testausführender zurück, damit die Korrektur erneut überprüft werden kann.workflow.add_conditional_edges("test_runner", route_after_test_runner, ...) sorgt dafür, dass LangGraph nach Abschluss der Tests einmal route_after_test_runner(state) aufruft. Der Rückgabewert leitet den Workflow entweder an den debugger oder an den reviewer.workflow.add_edge("reviewer", END) beendet die Ausführung. Nachdem der Reviewer seine Qualitätsprüfung und Zusammenfassung abgeschlossen hat, ist die Ausführung des Graphen beendet.workflow.compile() wandelt die Graphdefinition in einen ausführbaren Graphen um, indem er die Struktur überprüft und ihn für die Ausführung vorbereitet.Schritt 8: Der Eingangspunkt der CLI in src/main.py
Der letzte Bestandteil ist ein Eingabepunkt für die Kommandozeile, der mit rich erstellt wurde und den Fortschritt jedes Unter-Agenten in Echtzeit mithilfe gestilter Terminalfelder anzeigt. Die main.py-Datei in src beginnt mit den Funktionen zur Anzeige des Banners sowie der Konfigurations-Tabelle:
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()
Es geht hier ausschließlich um das Styling der Terminalansicht, was selbstverständlich ist, weshalb keine ausführlichere Erklärung notwendig ist. Derselbe Dateiabschnitt setzt mit run_agent_pipeline() fort:
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() ist der Punkt, an dem alle Komponenten zusammenkommen. Es erstellt den gemeinsamen AgentTeamState, baut das selbstheilende Graphenmodell auf und streamt die Ausführung Schritt für Schritt. Nach Abschluss jedes Knotens wird der aktuelle Zustand aktualisiert und der Fortschritt in Rich-Panels angezeigt: die Spezifikationen und Tests des Architekten, der Code des Entwicklers, das Ergebnis des Testlaufers, jede Debugging-Runde sowie der abschließende Bericht des Prüfers. Am Ende wird die überprüfte Implementierung, die generierte pytest-Suite sowie Ausführungsmetriken wie Gesamtzeit, Anzahl der Debugging-Iterationen und Testergebnisse ausgegeben.
Der CLI-Argumentparser rundet die Datei ab:
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()
Ausführung mit einem Rate-Limiter
Zur tatsächlichen Überprüfung erhält das Team eine nicht-triviale algorithmische Aufgabe. Dieser Befehl fordert einen threadsicheren Rate-Limiter mit Schiebelfenster-Logik an:
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."
Die Ausgabe des Terminals zeigt die folgende Abfolge:
- In Phase 1 von insgesamt 5 analysiert der leitende Architekt die Anforderungen an den Rate Limiter und erstellt eine mehrthreadige
pytest-Suite, die das Schneiden von Zeitstempeln, die Durchsetzung eines Schwellenwerts mittels Gleitfenster sowie gleichzeitige Anfragen aus mehreren Threads abdeckt. - In Phase 2 von insgesamt 5 schreibt der Entwickler eine erste
SlidingWindowRateLimiter-Klasse, die aufcollections.dequeund einemthreading.Lockbasiert. - In Phase 3 von insgesamt 5 führt der Teststarter in Iteration 0
pytestaus. Der Konkurrenzbelastungstest scheitert, weil das Schneiden der Zeitstempel außerhalb des gesperrten kritischen Abschnitts stattfand, wodurch durch einen Wettlaufzustand mehrere Threads die zulässige Anzahl an Anfragen überschritten.
_prune_expired_timestamps() vor dem Erwerb von self._lock aufgerufen wurde, und dass die Korrektur die Löschung der Zeitstempel in einen synchronisierten Block einbettet.pytest am gepatchten Code aus, und das Testpaket zeigt „grün“ an mit 6 bestanden in 0,04 Sekunden.Ein Screenshot der tatsächlichen Terminalausgabe dieses Laufs:
Die vollständige Ausgabe enthält außerdem den geschriebenen Python-Code sowie die zur Überprüfung verwendeten Testfälle. Ihre vollständige Wiedergabe hier würde ohne nennenswerten Mehrwert an Länge zufügen.
Wo dieses Muster Anwendung findet
Durch die Kombination spezialisierter Unteragenten, jeweils mit unterschiedlichem Modell und Denklevel sowie einer deterministischen Ausführungssandbox wird ein anfälliger Workflow auf Basis einer einzigen Anfrage in ein zuverlässigeres, selbstheilendes System umgewandelt.
Die Demonstration betraf ein Code-Generierungsteam, doch dasselbe Zustandsmaschinenmuster eignet sich für viele andere Workflows wie Datenpipelines, dynamische API-Orchestrierung und strukturierte Dokumentenvalidierung. Die Kernidee ist einfach: Verlassen Sie sich nicht darauf, dass ein einziges Modell alles erledigt. Weisen Sie jedem Agenten eine klare Aufgabe zu und lassen Sie das System seine eigene Arbeit während des Vorgangs überprüfen und korrigieren.
Drei Erkenntnisse sind besonders hervorzuheben:
- Lassen Sie niemals ein LLM seinen eigenen Code bewerten. Übertragen Sie die Ausführung auf eine lokale, deterministische Unterprozess-Sandbox, die
pytestausführt – das ist schnell, kostenlos und liefert ein klares Ergebnis von „erfolgreich“ oder „fehlschlägig“.