Git Worktree와 AST 분석으로 멀티 에이전트 코드 충돌과 토큰 폭증 막기
makedream٢٦ يوليو ٢٠٢٦
0
컴퓨터/소프트웨어
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
데모에서 잘 돌아가던 LLM 에이전트를 스웜 단위로 확장하면 반드시 두 개의 벽에 걸립니다. 에이전트끼리 같은 파일을 덮어써서 코드가 꼬이거나, 의미 없는 파일까지 대형 모델에 밀어 넣다가 API 비용으로 수백만 원이 깨지는 문제입니다.
동일한 작업 공간에서 여러 언어 모델 프로세스를 무작정 돌리면 사태가 금방 악화됩니다. 에이전트 A가 고치는 중인 미완성 파일을 에이전트 B가 읽어 들여 엉뚱한 코드를 짜고, 결국 커밋 내역마저 날아갑니다. 그렇다고 저장소 전체를 매번 복사하면 디스크 용량이 낭비되고 초기화에만 수 분이 걸립니다.
인메모리 파일 격리와 구문 분석 기반 라우팅, 정적 검수 파이프라인을 결합해 이 문제를 엔지니어링으로 푸는 방법을 다룹니다.
대용량 코드베이스에서 에이전트를 동시에 돌릴 때 발생하는 병목은 파일 시스템 경합입니다. 모놀리식 저장소 전체를 풀 클론(Full Git Clone)하는 대신 Git Worktree를 쓰면 메타데이터와 객체 DB를 공유하면서 수 MB 수준의 경량 디렉토리를 1초 만에 분리할 수 있습니다.
다만 수십 개의 에이전트가 동시에 커밋을 치면 상위 인덱스 파일(.git/index.lock)에 락 경합이 생깁니다. 이를 제어하려면 파일 잠금 기반의 샌드박스 레이어가 필요합니다.
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)
적용 순서는 간단합니다.
filelock 패키지를 설치하고 WorktreeSandboxManager 클래스를 프로젝트에 넣습니다.create_sandbox()를 불러 독립 디렉토리를 팝니다.safe_git_commit()을 거치게 만들어 지수 백오프로 락 충돌을 피합니다.이렇게 구성을 바꾸면 덮어쓰기 충돌이 사라집니다. 디버깅에 허비하던 시간도 주당 5시간 이상 줄어듭니다.
수정된 브랜치를 다시 메인 코드베이스로 가져올 때는 텍스트 라인 단위 병합 대신 추상 구문 트리(AST) 분석을 써야 안전합니다. 단순 텍스트 병합은 상단 import 문 위치만 바뀌어도 충돌을 뿜어냅니다. Python 내장 ast 모듈이나 Tree-Sitter로 소스 코드를 구문 노드 트리로 파싱한 뒤 함수나 클래스 단위로 병합하면 병합 실패율이 거의 0%로 떨어집니다.
모든 작업에 똑같이 Claude 3.5 Sonnet을 붙이면 비용을 감당할 수 없습니다. 단순 줄 수(LOC)로 복잡도를 판단하면 안 됩니다. 주석만 가득한 500줄짜리 데이터 클래스보다, 꼬인 삼항 연산자와 중첩 조건문으로 가득 찬 100줄짜리 코드가 훨씬 어렵기 마련입니다.
ast 모듈을 쓰면 노드 수, 순환 복잡도, 트리 깊이를 계산해 객체화된 점수로 변환할 수 있습니다.
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}
이 분석기를 백엔드 파이프라인 입구에 두고 라우팅 기준 점수를 50점으로 잡습니다.
50점 미만인 단위 테스트 작성, 유틸리티 구현, DTO 정의 같은 작업은 입력 백만 토큰당 $0.80 수준인 Claude 3.5 Haiku로 넘깁니다. 50점 이상인 대규모 리팩토링이나 구조 설계만 백만 토큰당 $3.00인 Claude 3.5 Sonnet으로 라우팅합니다. 전체 트래픽의 60% 이상을 Haiku가 처리하게 만드는 것만으로 API 비용이 최대 60% 축소됩니다.
긴 대화 맥락 때문에 토큰이 줄줄 새는 현상은 세션 리셋 미들웨어로 끊어냅니다. 누적 토큰을 집계하다 임계값에 다다르면 대화를 강제로 리셋합니다. 이때 AST로 핵심 함수 심볼과 남은 TODO만 뽑아낸 요약본을 만들어, 새 세션의 첫 프롬프트로 밀어 넣어 주면 맥락 손실 없이 작업을 이어갈 수 있습니다.
에이전트가 만든 초안 코드를 바로 저장소에 합치면 빌드가 박살 납니다. 그렇다고 단순 오타나 문법 오류를 잡으려고 LLM을 다시 호출하면 시간도 오래 걸리고 돈도 낭비됩니다.
린터, 타입 체커, LLM 리뷰어를 단계별로 붙인 검수 파이프라인을 짭니다.
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)
1단계에서 AST 파싱과 Ruff로 구문 오류를 잡고, 2단계에서 Mypy로 타입을 맞춥니다. 이 정적 검증 도구들을 모두 통과한 코드만 3단계인 Claude 3.5 Sonnet 심층 리뷰어로 보냅니다. 단순 괄호 빠짐이나 타입 에러로 LLM을 재호출하는 일이 사라져 파이프라인 완수 속도가 40% 빨라집니다.
검수 통과 실패 시 루프에 갇히는 걸 막으려면 서킷 브레이커가 필수입니다. 동일 에러에 대한 재시도를 최대 3회로 제한하고, 에러 메시지 해시값이 이전과 완전히 똑같다면 에이전트가 환각 루프에 빠진 것으로 판단해 실행을 즉시 중단해야 합니다.
여러 에이전트가 어떤 파일과 브랜치를 건드리는지 중앙에서 관리하려면 SQLite 스키마 정도는 갖춰야 합니다.
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
);
에이전트가 공통 모듈 커밋에 성공할 때마다 FILE_MUTATED 이벤트를 발행합니다. 다른 에이전트들은 이 알림을 받아 자기가 참조하던 AST 심볼 정의를 즉시 최신화합니다.
만약 특정 에이전트가 검수 실패 후 복구 불가 상태에 빠지면, 처음 작업을 시작할 때 찍어둔 스냅샷 커밋 SHA로 원자적 롤백을 실행합니다.
git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA>
git -C .agent_worktrees/wt_agent_01 clean -fd
이렇게 격리 디렉터리, 구문 기반 모델 라우팅, 정적 검수 파이프라인, 상태 DB를 엮어두면 파일 충돌이나 비용 폭증 걱정 없이 프로덕션급 에이전트 스웜을 안정적으로 돌릴 수 있습니다.