Empêcher les conflits de code et l'explosion de tokens entre multi-agents grâce à Git Worktree et l'analyse AST
makedream26 июля 2026 г.
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Lorsque vous faites passer les agents LLM qui fonctionnaient parfaitement dans une démo à une échelle de swarm, vous vous heurtez inévitablement à deux obstacles. Soit les agents s'écrasent mutuellement les mêmes fichiers, ce qui embrouille le code, soit des fichiers insignifiants sont envoyés à de grands modèles, ce qui génère des millions de won de frais d'API.
Faire tourner aveuglément plusieurs processus de modèles de langage dans le même espace de travail aggrave rapidement la situation. L'agent A modifie un fichier non terminé que l'agent B lit pour écrire du code incohérent, et au final, même l'historique des commits disparaît. Cependant, copier l'ensemble du dépôt à chaque fois gaspille de la capacité disque et prend plusieurs minutes rien que pour l'initialisation.
Nous verrons comment résoudre ce problème sur le plan de l'ingénierie en combinant l'isolement des fichiers en mémoire, le routage basé sur l'analyse syntaxique et une pipeline d'inspection statique.
Le goulot d'étranglement qui survient lors de l'exécution simultanée d'agents sur une base de code volumineuse est la concurrence sur le système de fichiers. Au lieu d'effectuer un clone complet (Full Git Clone) de l'ensemble du dépôt monolithique, l'utilisation de Git Worktree permet de partager les métadonnées et la base de données d'objets tout en séparant en une seconde un répertoire léger de l'ordre de quelques Mo.
Cependant, lorsque des dizaines d'agents effectuent des commits simultanément, une concurrence de verrouillage se produit sur le fichier d'index supérieur (.git/index.lock). Pour gérer cela, une couche de sandbox basée sur le verrouillage de fichiers est nécessaire.
`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 생성 실패: {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 실패: {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_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("커밋되지 않은 변경사항이 존재하여 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)
`
L'ordre d'application est simple :
filelock et intégrez la classe WorktreeSandboxManager dans le projet.create_sandbox() pour créer un répertoire indépendant.safe_git_commit() pour éviter les conflits de verrouillage grâce à un backoff exponentiel.En modifiant ainsi la configuration, les conflits de remplacement disparaissent. Le temps perdu à déboguer est également réduit de plus de 5 heures par semaine.
Lors du rapatriement de la branche modifiée vers la base de code principale, il est plus sûr d'utiliser l'analyse de l'arbre syntaxique abstrait (AST) plutôt qu'une fusion ligne par ligne de texte. Une simple fusion de texte génère des conflits dès que la position des instructions import en haut de fichier change. Si vous analysez le code source sous forme d'arbre de nœuds syntaxiques à l'aide du module ast intégré à Python ou de Tree-Sitter, puis que vous fusionnez au niveau des fonctions ou des classes, le taux d'échec de fusion tombe à près de 0 %.
Associer systématiquement Claude 3.5 Sonnet à toutes les tâches est insoutenable sur le plan financier. La complexité ne doit pas être évaluée en fonction du simple nombre de lignes de code (LOC). Un code de 100 lignes truffé d'opérateurs ternaires alambiqués et de conditions imbriquées est bien plus complexe qu'une classe de données de 500 lignes remplie uniquement de commentaires.
En utilisant le module ast, vous pouvez calculer le nombre de nœuds, la complexité cyclomatique et la profondeur de l'arbre pour les convertir en un score objectivé.
`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}
`
Placez cet analyseur à l'entrée du pipeline backend et fixez le score de référence de routage à 50 points.
Les tâches avec un score inférieur à 50 points, telles que la rédaction de tests unitaires, l'implémentation d'utilitaires ou la définition de DTO, sont transmises à Claude 3.5 Haiku, qui coûte environ 0,80 $ par million de tokens d'entrée. Seuls les refactorisations à grande échelle ou les conceptions d'architecture ayant un score supérieur ou égal à 50 points sont routés vers Claude 3.5 Sonnet à 3,00 $ par million de tokens. Le simple fait de faire traiter plus de 60 % du trafic global par Haiku permet de réduire les coûts d'API jusqu'à 60 %.
L'hémorragie de tokens causée par des contextes de conversation trop longs est interrompue par un middleware de réinitialisation de session. Lorsque le nombre de tokens cumulés est comptabilisé et atteint un seuil, la conversation est réinitialisée de force. À ce moment-là, en créant un résumé qui extrait les symboles de fonctions clés et les TODO restants via l'AST, puis en l'injectant dans le premier prompt de la nouvelle session, le travail peut se poursuivre sans perte de contexte.
Si vous fussionnez directement dans le dépôt le brouillon de code généré par un agent, la build va s'effondrer. Pour autant, réappeler un LLM juste pour corriger une simple faute de frappe ou une erreur de syntaxe fait perdre du temps et gaspille de l'argent.
Construisez une pipeline d'inspection étape par étape combinant linter, vérificateur de types et réviseur LLM.
`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="검수 통과 여부")
failed_stage: Optional[str] = Field(default=None, description="실패한 검수 단계")
error_message: Optional[str] = Field(default=None, description="에러 메시지")
suggested_context: Optional[str] = Field(default=None, description="수정을 위해 주입할 콘텍스트")
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 발생 라인 {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)
`
À l'étape 1, le parsing AST et Ruff détectent les erreurs de syntaxe, et à l'étape 2, Mypy vérifie le typage. Seul le code qui réussit l'ensemble de ces outils de vérification statique est envoyé au réviseur approfondi Claude 3.5 Sonnet à l'étape 3. Le fait d'éliminer les réappels du LLM pour de simples parenthèses manquantes ou des erreurs de type permet d'accélérer la vitesse de finalisation de la pipeline de 40 %.
Pour éviter de rester bloqué dans une boucle en cas d'échec de la vérification, un disjoncteur (circuit breaker) est indispensable. Il convient de limiter le nombre de tentatives pour une même erreur à 3 maximum et, si le hash du message d'erreur est strictement identique au précédent, d'estimer que l'agent est tombé dans une boucle d'hallucination pour interrompre immédiatement l'exécution.
Pour gérer de manière centralisée les fichiers et branches touchés par plusieurs agents, un schéma SQLite minimal est nécessaire.
`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
);
`
Chaque fois qu'un agent réussit un commit sur un module commun, il émet un événement FILE_MUTATED. Les autres agents reçoivent cette notification et mettent immédiatement à jour la définition des symboles AST qu'ils référençaient.
Si un agent spécifique se retrouve dans un état irrécupérable après un échec d'inspection, un rollback atomique est exécuté vers le SHA du commit de snapshot pris au tout début du travail.
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
En articulant ainsi des répertoires isolés, un routage de modèles basé sur la syntaxe, une pipeline d'inspection statique et une base de données d'état, vous pouvez faire tourner de manière stable un swarm d'agents de niveau production, sans craindre les conflits de fichiers ou l'explosion des coûts.