Evitando Conflitos de Código e Explosão de Tokens em Multi-agentes com Git Worktree e Análise 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
Quando você escala agentes de LLM que funcionavam bem em demonstrações para o nível de swarm, você inevitavelmente esbarra em duas barreiras. Os agentes sobrescrevem os mesmos arquivos, bagunçando o código, ou você envia arquivos irrelevantes para modelos grandes, gastando fortunas em custos de API.
Rodar múltiplos processos de modelos de linguagem sem critério no mesmo espaço de trabalho piora a situação rapidamente. O Agente B lê um arquivo incompleto que o Agente A ainda está modificando, gera um código totalmente equivocado e, no final, até o histórico de commits se perde. Por outro lado, copiar todo o repositório a cada vez desperdiça espaço em disco e leva vários minutos apenas para inicializar.
Abordaremos como resolver esse problema do ponto de vista da engenharia, combinando isolamento de arquivos em memória, roteamento baseado en análise sintática e um pipeline de validação estática.
O gargalo ao executar agentes simultaneamente em bases de código grandes é a concorrência no sistema de arquivos. Em vez de fazer um clone completo (Full Git Clone) de todo o repositório monolítico, o uso do Git Worktree permite compartilhar metadados e o banco de dados de objetos, isolando diretórios leves de poucos MBs em apenas um segundo.
No entanto, se dezenas de agentes realizarem commits ao mesmo tempo, ocorrerá concorrência de bloqueio no arquivo de índice superior (.git/index.lock). Para controlar isso, é necessária uma camada de sandbox baseada em bloqueio de arquivos.
`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)
`
A ordem de aplicação é simples:
filelock e adicione a classe WorktreeSandboxManager ao seu projeto.create_sandbox() para criar um diretório independente.safe_git_commit() para evitar conflitos de lock com backoff exponencial.Ao mudar essa estrutura, os conflitos de sobrescrita desaparecem. O tempo desperdiçado em depuração também reduz em mais de 5 horas por semana.
Para trazer a branch modificada de volta à base de código principal, é mais seguro usar análise de Árvore de Sintaxe Abstrata (AST) em vez de mesclagem por linha de texto. A mesclagem de texto simples gera conflitos mesmo se apenas a posição de uma declaração de importação no topo for alterada. Fazer o parse do código-fonte em uma árvore de nós sintáticos usando o módulo nativo ast do Python ou Tree-Sitter e, em seguida, realizar a mesclagem por função ou classe reduz a taxa de falha de mesclagem para quase 0%.
Se você usar o Claude 3.5 Sonnet para todas as tarefas indistintamente, não conseguirá arcar com os custos. A complexidade não deve ser julgada simplesmente pelo número de linhas de código (LOC). Um código de 100 linhas cheio de operadores ternários conturbados e condicionais aninhadas é muito mais difícil do que uma classe de dados de 500 linhas cheia apenas de comentários.
Usando o módulo ast, é possível calcular o número de nós, a complexidade ciclomática e a profundidade da árvore, convertendo-os em uma pontuação 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}
`
Coloque este analisador na entrada do pipeline de backend e defina a pontuação de corte para roteamento em 50 pontos.
Tarefas com pontuação inferior a 50, como escrita de testes unitários, implementação de utilitários e definição de DTOs, são repassadas ao Claude 3.5 Haiku, que custa cerca de $0,80 por milhão de tokens de entrada. Apenas refatorações de grande escala ou design estrutural com pontuação igual ou superior a 50 são roteados para o Claude 3.5 Sonnet, que custa $3,00 por milhão de tokens. Fazer com que o Haiku processe mais de 60% do tráfego total já reduz os custos de API em até 60%.
Vazamentos de tokens causados por contextos de conversa longos são interrompidos com um middleware de redefinição de sessão. Ao agregar tokens acumulados e atingir um limite, a conversa é redefinida à força. Nesse momento, um resumo extraído com os símbolos de funções principais e TODOs restantes via AST é inserido como o primeiro prompt da nova sessão, permitindo continuar o trabalho sem perda de contexto.
Se você mesclar o código de rascunho gerado pelo agente diretamente no repositório, a compilação vai quebrar. Por outro lado, chamar a LLM novamente apenas para corrigir erros de digitação simples ou erros de sintaxe consome muito tempo e dinheiro.
Crie um pipeline de validação integrando linters, verificadores de tipo e revisores de LLM em etapas.
`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)
`
Na Etapa 1, capturam-se erros de sintaxe com parse AST e Ruff; na Etapa 2, ajustam-se os tipos com Mypy. Apenas o código que passa por todas essas ferramentas de verificação estática é enviado para o revisor profundo da Etapa 3 (Claude 3.5 Sonnet). Elimina-se a necessidade de reevocar a LLM por causa de parênteses ausentes ou erros de tipo, acelerando a velocidade de conclusão do pipeline em 40%.
Para evitar ficar preso em loops quando a validação falha, um disjuntor (circuit breaker) é essencial. Limite as tentativas para o mesmo erro a no máximo 3 vezes e, se o hash da mensagem de erro for exatamente igual ao anterior, determine que o agente entrou em um loop de alucinação e interrompa a execução imediatamente.
Para gerenciar centraladamente quais arquivos e branches múltiplos agentes estão tocando, é necessário ter pelo menos um esquema SQLite.
`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
);
`
Toda vez que um agente realiza com sucesso o commit de um módulo comum, ele publica um evento FILE_MUTATED. Outros agentes recebem essa notificação e atualizam imediatamente as definições de símbolos AST que estavam referenciando.
Se um agente específico entrar em um estado irrecuperável após uma falha na validação, executa-se um rollback atômico para o SHA do commit de snapshot registrado no início do trabalho.
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
Integrando diretórios de isolamento, roteamento de modelos baseado em sintaxe, pipeline de validação estática e banco de dados de estado dessa forma, você pode executar um swarm de agentes em nível de produção de forma estável, sem se preocupar com conflitos de arquivos ou explosão de custos.