Evitar conflictos de código y explosiones de tokens en multi-agentes con Git Worktree y análisis de 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
Cuando escalas los agentes LLM que funcionaban bien en una demo a nivel de enjambre (swarm), inevitablemente te chocas con dos muros: los agentes se sobrescriben los archivos entre sí enredando el código, o terminas inyectando archivos irrelevantes a modelos grandes, lo que dispara los costos de la API a miles de dólares.
Ejecutar a ciegas múltiples procesos de modelos de lenguaje en el mismo espacio de trabajo empeora rápidamente la situación. El agente A lee un archivo incompleto que el agente B está modificando y genera código erróneo, lo que al final termina arruinando incluso el historial de commits. Por otro lado, copiar todo el repositorio cada vez malgasta espacio en disco y tarda varios minutos solo en inicializarse.
Aquí abordaremos cómo resolver este problema mediante ingeniería, combinando el aislamiento de archivos en memoria, el enrutamiento basado en análisis sintáctico y un pipeline de inspección estática.
El cuello de botella que surge al ejecutar agentes simultáneamente en un código base de gran tamaño es la contención del sistema de archivos. En lugar de hacer un clone completo (Full Git Clone) de todo el repositorio monolítico, usar Git Worktree permite separar directorios ligeros de solo unos pocos MB en 1 segundo, compartiendo los metadatos y la base de datos de objetos.
Sin embargo, si docenas de agentes realizan commits al mismo tiempo, se produce una contención de bloqueos en el archivo de índice superior (.git/index.lock). Para controlar esto, se necesita una capa de sandbox basada en bloqueo de archivos.
`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)
`
El orden de aplicación es sencillo:
filelock e incluir la clase WorktreeSandboxManager en el proyecto.create_sandbox() para crear un directorio independiente.safe_git_commit() para evitar conflictos de bloqueo mediante un retardo exponencial (exponential backoff).Al cambiar la configuración de esta manera, los conflictos de sobrescritura desaparecen. El tiempo desperdiciado en depuración también se reduce en más de 5 horas a la semana.
Para volver a integrar la rama modificada en el código base principal, es más seguro utilizar un análisis de Árbol de Sintaxis Abstracta (AST) en lugar de una fusión línea por línea de texto. Una fusión de texto simple puede generar conflictos incluso si solo cambia la posición de las sentencias de importación superiores. Si parseas el código fuente en un árbol de nodos sintácticos utilizando el módulo nativo ast de Python o Tree-Sitter y luego realizas la fusión a nivel de funciones o clases, la tasa de fallos de fusión cae casi al 0%.
Si le asignas Claude 3.5 Sonnet a todas las tareas por igual, no podrás soportar los costos. La complejidad no debe juzgarse simplemente por las líneas de código (LOC). Un código de 100 líneas lleno de operadores ternarios enredados y sentencias condicionantes anidadas suele ser mucho más difícil que una clase de datos de 500 líneas llena de comentarios.
Al usar el módulo ast, puedes calcular el número de nodos, la complejidad ciclomática y la profundidad del árbol para convertirlos en una puntuación objetiva.
`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}
`
Coloca este analizador a la entrada del pipeline del backend y establece una puntuación de referencia de 50 puntos para el enrutamiento.
Las tareas con menos de 50 puntos, como escribir pruebas unitarias, implementar utilidades o definir DTOs, se devuelven a Claude 3.5 Haiku, que cuesta alrededor de $0.80 por millón de tokens de entrada. Solo la refactorización a gran escala o el diseño de arquitectura con 50 puntos o más se enrutan a Claude 3.5 Sonnet, que cuesta $3.00 por millón de tokens. Lograr que Haiku procese más del 60% del tráfico total reduce por sí solo los costos de la API hasta en un 60%.
La fuga constante de tokens provocada por contextos de conversación largos se frena con un middleware de reinicio de sesión. Si la acumulación de tokens alcanza un umbral determinado, la conversación se reinicia forzosamente. En ese momento, se crea un resumen extrayendo solo los símbolos de función clave y los TODOs restantes mediante el AST, y se inyecta como el primer prompt de la nueva sesión para continuar el trabajo sin pérdida de contexto.
Si integras el borrador de código generado por el agente directamente en el repositorio, la compilación se rompe. Por otro lado, volver a llamar al LLM solo para corregir un simple error tipográfico o de sintaxis consume mucho tiempo y malgasta dinero.
Se diseña un pipeline de inspección que conecta por etapas un linter, un verificador de tipos y un revisor 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)
`
En la etapa 1 se capturan los errores sintácticos con el parseo de AST y Ruff, y en la etapa 2 se ajustan los tipos con Mypy. Solo el código que supera todas estas herramientas de verificación estática se envía a la etapa 3, que es el revisor profundo Claude 3.5 Sonnet. Al eliminar la re-llamada al LLM por falta de un simple paréntesis o errores de tipo, la velocidad de finalización del pipeline se acelera en un 40%.
Para evitar quedar atrapado en un bucle si falla la verificación, es esencial un disyuntor (circuit breaker). Los reintentos para el mismo error deben limitarse a un máximo de 3 veces, y si el valor hash del mensaje de error es idéntico al anterior, se determina que el agente ha caído en un bucle de alucinación y la ejecución debe detenerse de inmediato.
Para gestionar de manera centralizada qué archivos y ramas está tocando cada agente, se requiere al menos un esquema SQLite adecuado.
`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
);
`
Cada vez que un agente realiza con éxito un commit en un módulo común, emite un evento FILE_MUTATED. Los demás agentes reciben esta notificación y actualizan de inmediato las definiciones de símbolos AST a las que hacían referencia.
Si un agente cae en un estado irrecuperable tras fallar la inspección, se ejecuta un rollback atómico al SHA del commit de la instantánea tomada al iniciar el trabajo.
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
Al entrelazar así los directorios aislados, el enrutamiento de modelos basado en sintaxis, el pipeline de inspección estática y la base de datos de estado, puedes ejecutar de manera estable un enjambre de agentes listo para producción, sin preocuparte por conflictos de archivos o disparos en los costos.