Wenn Sie LLM-Agenten, die in einer Demo gut funktioniert haben, auf Swarm-Ebene skalieren, stoßen Sie zwangsläufig auf zwei Hürden. Entweder überschreiben sich die Agenten gegenseitig dieselben Dateien, was zu verwirrtem Code führt, oder Sie speisen unnötige Dateien in riesige Modelle ein, was zu API-Kosten in Höhe von Tausenden von Euro führt.
Das blinde Ausführen mehrerer Sprachmodell-Prozesse im selben Arbeitsbereich verschlimmert die Lage schnell. Agent B liest eine unvollständige Datei ein, die Agent A gerade bearbeitet, und schreibt daraufhin falschen Code, sodass am Ende sogar die Commit-Historie verloren geht. Das gesamte Repository jedes Mal zu kopieren, verschwendet jedoch Speicherkapazität und dauert Minuten allein für die Initialisierung.
In diesem Artikel wird behandelt, wie sich dieses Problem ingenieurmäßig durch die Kombination von In-Memory-Dateiisolierung, syntaxanalysebasiertem Routing und einer statischen Prüf-Pipeline lösen lässt.
Dateiüberschreibungen mit Git Worktree blockieren
Der Engpass beim gleichzeitigen Ausführen von Agenten in einer großen Codebasis ist die Konkurrenz auf Dateisystemebene. Anstatt ein vollständiges Git-Clone (Full Git Clone) des gesamten monolithischen Repositorys durchzuführen, ermöglicht die Verwendung von Git Worktree das Abspalten eines leichtgewichtigen Verzeichnisses im Megabyte-Bereich innerhalb einer Sekunde, während Metadaten und die Objekt-DB geteilt werden.
Wenn jedoch Dutzende von Agenten gleichzeitig Commits ausführen, kommt es zu Lock-Konflikten in der übergeordneten Indexdatei (.git/index.lock). Um dies zu steuern, ist eine Sandbox-Schicht auf Basis von Dateisperren erforderlich.
`python
import os
import sys
import time
import subprocess
import shutil
from pathlib import Path
from typing import Optional, List
from filelock import FileLock, Timeout
class WorktreeSandboxManager:
def init(self, repo_path: str, base_branch: str = "main"):
self.repo_path = Path(repo_path).resolve()
self.base_branch = base_branch
self.worktrees_dir = self.repo_path / ".agent_worktrees"
self.locks_dir = self.repo_path / ".agent_locks"
self.worktrees_dir.mkdir(exist_ok=True)
self.locks_dir.mkdir(exist_ok=True)
def create_sandbox(self, agent_id: str, task_name: str) -> Path:
branch_name = f"agent/{agent_id}-{task_name}"
worktree_path = self.worktrees_dir / f"wt_{agent_id}"
if worktree_path.exists():
self.cleanup_sandbox(agent_id, force=True)
cmd = [
"git", "-C", str(self.repo_path),
"worktree", "add", "-b", branch_name,
str(worktree_path), self.base_branch
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Worktree creation failed: {result.stderr}")
return worktree_path
def safe_git_commit(self, worktree_path: Path, commit_message: str, max_retries: int = 5) -> bool:
lock_file_path = self.locks_dir / "git_index.lock"
file_lock = FileLock(str(lock_file_path), timeout=10)
for attempt in range(max_retries):
try:
with file_lock:
add_res = subprocess.run(
["git", "-C", str(worktree_path), "add", "."],
capture_output=True, text=True
)
if add_res.returncode != 0:
raise RuntimeError(f"Git add failed: {add_res.stderr}")
commit_res = subprocess.run(
["git", "-C", str(worktree_path), "commit", "-m", commit_message],
capture_output=True, text=True
)
if commit_res.returncode == 0:
return True
if "index.lock" in commit_res.stderr or "Unable to create" in commit_res.stderr:
backoff = (2 ** attempt) * 0.2
time.sleep(backoff)
continue
else:
print(f"Commit failed (non-contention error): {commit_res.stderr}")
return False
except (Timeout, RuntimeError) as e:
backoff = (2 ** attempt) * 0.2
time.sleep(backoff)
return False
def cleanup_sandbox(self, agent_id: str, force: bool = False):
worktree_path = self.worktrees_dir / f"wt_{agent_id}"
if not worktree_path.exists():
return
status_res = subprocess.run(
["git", "-C", str(worktree_path), "status", "--porcelain"],
capture_output=True, text=True
)
if status_res.stdout.strip() and not force:
raise RuntimeError("Uncommitted changes exist, cannot remove Worktree.")
subprocess.run(
["git", "-C", str(self.repo_path), "worktree", "remove", "--force", str(worktree_path)],
capture_output=True, text=True
)
if worktree_path.exists():
shutil.rmtree(worktree_path, ignore_errors=True)
`
Die Reihenfolge der Anwendung ist einfach:
- Installieren Sie das
filelock-Paket und fügen Sie die KlasseWorktreeSandboxManagerzu Ihrem Projekt hinzu. - Rufen Sie
create_sandbox()auf, bevor Sie dem Agenten eine Aufgabe zuweisen, um ein isoliertes Verzeichnis zu erstellen. - Lassen Sie den Commit nach Abschluss der Arbeit über
safe_git_commit()laufen, um Sperrkonflikte durch exponentielles Backoff zu vermeiden.
Wenn Sie die Konfiguration so anpassen, verschwinden Überschreibungskonflikte. Auch die für das Debuggen verschwendete Zeit verringert sich um mehr als 5 Stunden pro Woche.
Wenn Sie den geänderten Branch wieder in die Hauptcodebasis zusammenführen, sollten Sie zur Sicherheit eine Analyse des abstrakten Syntaxbaums (AST) anstelle eines zeilenweisen Text-Merges verwenden. Ein einfacher Text-Merge führt selbst bei einer Positionsänderung von oberen import-Anweisungen zu Konflikten. Wenn Sie den Quellcode mit dem in Python integrierten ast-Modul oder Tree-Sitter in einen Syntax-Knotenbaum parsen und auf Funktionsebene oder Klassebene zusammenführen, sinkt die Merge-Fehlerrate auf nahezu 0 %.
Token-Kosten um 60 % senken durch AST-Komplexitätsmessung
Es ist finanziell nicht tragbar, für jede Aufgabe gleichermaßen Claude 3.5 Sonnet einzusetzen. Die Komplexität sollte nicht nur nach den reinen Zeilenzahlen (LOC) beurteilt werden. Ein 100 Zeilen langer Code voll verschachtelter ternärer Operatoren und verknüpfter Bedingungen ist bei Weitem schwieriger als eine 500 Zeilen lange Datenklasse, die nur aus Kommentaren besteht.
Mit dem ast-Modul können Sie die Knotenanzahl, die zyklomatische Komplexität und die Baumtiefe berechnen und in einen objektivierten Wert umwandeln.
`python
import ast
class CodeComplexityAnalyzer(ast.NodeVisitor):
def init(self):
self.node_count = 0
self.max_depth = 0
self.current_depth = 0
self.cyclomatic_complexity = 1
def generic_visit(self, node):
self.node_count += 1
self.current_depth += 1
if self.current_depth > self.max_depth:
self.max_depth = self.current_depth
super().generic_visit(node)
self.current_depth -= 1
def visit_If(self, node):
self.cyclomatic_complexity += 1
self.generic_visit(node)
def visit_For(self, node):
self.cyclomatic_complexity += 1
self.generic_visit(node)
def visit_While(self, node):
self.cyclomatic_complexity += 1
self.generic_visit(node)
def visit_ExceptHandler(self, node):
self.cyclomatic_complexity += 1
self.generic_visit(node)
def visit_BoolOp(self, node):
self.cyclomatic_complexity += len(node.values) - 1
self.generic_visit(node)
def calculate_ast_metrics(source_code: str) -> dict:
try:
tree = ast.parse(source_code)
analyzer = CodeComplexityAnalyzer()
analyzer.visit(tree)
score = (analyzer.node_count * 0.2) + (analyzer.max_depth * 1.5) + (analyzer.cyclomatic_complexity * 3.0)
return {
"node_count": analyzer.node_count,
"max_depth": analyzer.max_depth,
"cyclomatic_complexity": analyzer.cyclomatic_complexity,
"complexity_score": round(score, 2),
"is_valid": True
}
except SyntaxError as e:
return {"is_valid": False, "error": str(e), "complexity_score": 9999}
`
Platzieren Sie diesen Analysator am Eingang der Backend-Pipeline und setzen Sie den Schwellenwert für das Routing auf 50 Punkte.
Aufgaben mit weniger als 50 Punkten, wie das Schreiben von Unittests, das Implementieren von Utilities oder das Definieren von DTOs, werden an Claude 3.5 Haiku übergeben, was etwa $ 0,80 pro Million Eingabe-Token kostet. Nur umfangreiches Refactoring oder Strukturdesign mit 50 oder mehr Punkten wird an Claude 3.5 Sonnet für $ 3,00 pro Million Token geroutet. Allein dadurch, dass Haiku mehr als 60 % des gesamten Traffics verarbeitet, reduzieren sich die API-Kosten um bis zu 60 %.
Das Phänomen, dass Token aufgrund langer Konversationskontexte verloren gehen, wird durch eine Session-Reset-Middleware unterbrochen. Wenn die kumulierten Token einen Schwellenwert erreichen, wird das Gespräch erzwungen zurückgesetzt. Zu diesem Zeitpunkt wird eine Zusammenfassung mit den wichtigsten Funktionssymbolen und verbleibenden TODOs per AST erstellt und als erster Prompt der neuen Sitzung übergeben, sodass die Arbeit ohne Kontextverlust fortgesetzt werden kann.
Arbeitsgeschwindigkeit um 40 % steigern mit deterministischer Prüfkette
Wenn vom Agenten erstellter Entwurfscode direkt in das Repository zusammengeführt wird, bricht der Build zusammen. Dennoch führt der erneute Aufruf des LLM zur Behebung einfacher Tippfehler oder Syntaxfehler zu Zeitverlust und Geldverschwendung.
Bauen Sie eine Prüf-Pipeline auf, die Linter, Typechecker und LLM-Reviewer stufenweise verbindet.
`python
import ast
import subprocess
from pathlib import Path
from typing import Optional
from pydantic import BaseModel, Field
class ValidationResult(BaseModel):
is_success: bool = Field(description="Prüfungsstatus")
failed_stage: Optional[str] = Field(default=None, description="Fehlgeschlagene Prüfstufe")
error_message: Optional[str] = Field(default=None, description="Fehlermeldung")
suggested_context: Optional[str] = Field(default=None, description="Infundierter Kontext für Korrektur")
class MultiLensReviewerChain:
def init(self, worktree_path: Path):
self.worktree_path = worktree_path
def run_stage1_ast_lint(self, file_path: Path) -> ValidationResult:
try:
with open(file_path, "r", encoding="utf-8") as f:
code_content = f.read()
ast.parse(code_content)
except SyntaxError as e:
return ValidationResult(
is_success=False,
failed_stage="Stage 1 (AST Syntax)",
error_message=f"SyntaxError in Zeile {e.lineno}: {e.msg}",
suggested_context=e.text
)
res = subprocess.run(["ruff", "check", str(file_path)], capture_output=True, text=True)
if res.returncode != 0:
return ValidationResult(
is_success=False,
failed_stage="Stage 1 (Ruff Linter)",
error_message=res.stdout or res.stderr
)
return ValidationResult(is_success=True)
def run_stage2_type_check(self, file_path: Path) -> ValidationResult:
res = subprocess.run(
["mypy", "--config-file", "mypy.ini", str(file_path)],
capture_output=True, text=True, cwd=str(self.worktree_path)
)
if res.returncode != 0:
return ValidationResult(
is_success=False,
failed_stage="Stage 2 (Mypy TypeChecker)",
error_message=res.stdout
)
return ValidationResult(is_success=True)
def execute_pipeline(self, target_file_rel_path: str) -> ValidationResult:
full_path = self.worktree_path / target_file_rel_path
s1_res = self.run_stage1_ast_lint(full_path)
if not s1_res.is_success:
return s1_res
s2_res = self.run_stage2_type_check(full_path)
if not s2_res.is_success:
return s2_res
return ValidationResult(is_success=True)
`
In Stufe 1 werden Syntaxfehler mit AST-Parsing und Ruff abgefangen, in Stufe 2 werden die Typen mit Mypy abgeglichen. Nur Code, der all diese statischen Verifizierungstools besteht, wird an den Deep Reviewer von Claude 3.5 Sonnet in Stufe 3 weitergeleitet. Das erneute Aufrufen des LLM aufgrund fehlender Klammern oder Typfehler entfällt, wodurch sich die Fertigstellungsgeschwindigkeit der Pipeline um 40 % erhöht.
Um zu verhindern, dass das System bei einem Fehler in einer Endlosschleife gefangen bleibt, ist ein Circuit Breaker unerlässlich. Begrenzen Sie Wiederholungsversuche für denselben Fehler auf maximal 3 Mal. Wenn der Hash-Wert der Fehlermeldung exakt derselbe wie zuvor ist, sollte davon ausgegangen werden, dass der Agent in eine Halluzinationsschleife geraten ist, und die Ausführung sollte sofort gestoppt werden.
Gemeinsamer Speicher und Transaktions-Rollback
Um zentral zu verwalten, welche Dateien und Branches von mehreren Agenten berührt werden, ist mindestens ein SQLite-Schema erforderlich.
`sql
CREATE TABLE agent_sessions (
agent_id TEXT PRIMARY KEY,
worktree_path TEXT NOT NULL,
current_status TEXT CHECK(current_status IN ('IDLE', 'RUNNING', 'LINTING', 'FAILED', 'COMPLETED')),
assigned_task TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE file_locks (
file_path TEXT PRIMARY KEY,
locked_by_agent TEXT NOT NULL,
ast_symbol_node TEXT,
lock_acquired_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(locked_by_agent) REFERENCES agent_sessions(agent_id)
);
CREATE TABLE context_events (
event_id INTEGER PRIMARY KEY AUTOINCREMENT,
source_agent TEXT NOT NULL,
event_type TEXT CHECK(event_type IN ('FILE_MUTATED', 'INTERFACE_CHANGED', 'ROLLBACK_TRIGGERED')),
affected_path TEXT NOT NULL,
payload_json TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
`
Jedes Mal, wenn ein Agent erfolgreich einen Commit für ein gemeinsames Modul durchführt, veröffentlicht er ein FILE_MUTATED-Ereignis. Andere Agenten empfangen diese Benachrichtigung und aktualisieren sofort die von ihnen referenzierten AST-Symboldefinitionen auf den neuesten Stand.
Wenn ein bestimmter Agent nach einem Prüfungsfehler in einen nicht wiederherstellbaren Zustand gerät, wird ein atomarer Rollback auf den SHA-Wert des Snapshot-Commits durchgeführt, der zu Beginn der Arbeit erstellt wurde.
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
Durch die Verknüpfung von isolierten Verzeichnissen, syntaxbasiertem Modell-Routing, statischen Prüf-Pipelines und einer Zustands-DB können Sie ein produktionsbereites Agenten-Swarm stabil betreiben, ohne sich Gedanken über Dateikonflikte oder Kostenexplosionen machen zu müssen.
