Mencegah Konflik Kode Multi-Agent dan Lonjakan Token dengan Git Worktree dan Analisis AST
makedream26. Juli 2026
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Ketika agen LLM yang berjalan lancar di versi demo diperluas ke skala swarm, Anda pasti akan menabrak dua tembok utama. Kode menjadi berantakan karena agen-agen saling menimpa berkas yang sama, atau biaya API membengkak hingga jutaan rupiah akibat memasukkan berkas yang tidak relevan ke dalam model besar.
Menjalankan beberapa proses model bahasa secara asal-asalan di ruang kerja yang sama akan memperburuk keadaan dengan cepat. Agen B membaca berkas belum selesai yang sedang diperbaiki oleh Agen A lalu menghasilkan kode yang ngawur, hingga akhirnya riwayat commit pun ikut hilang. Di sisi lain, menyalin seluruh repositori setiap kali menjalankan proses akan membuang-buang kapasitas disk dan memakan waktu beberapa menit hanya untuk inisialisasi.
Artikel ini membahas cara menyelesaikan masalah tersebut secara teknis melalui penggabungan isolasi berkas in-memory, routing berbasis analisis sintaksis, dan pipeline verifikasi statis.
Bottleneck yang terjadi saat menjalankan agen secara bersamaan di dalam codebase berukuran besar adalah perebutan sistem berkas (file system contention). Alih-alih melakukan full clone (Full Git Clone) terhadap seluruh repositori monolitik, penggunaan Git Worktree memungkinkan kita memisahkan direktori ringan berukuran beberapa MB hanya dalam waktu 1 detik sambil tetap berbagi metadata dan DB objek.
Namun, jika puluhan agen melakukan commit secara bersamaan, akan terjadi perebutan lock pada berkas indeks tingkat atas (.git/index.lock). Untuk mengendalikannya, diperlukan lapisan sandbox berbasis pemblokiran berkas (file locking).
`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)
`
Urutan penerapannya sangat sederhana.
filelock dan masukkan kelas WorktreeSandboxManager ke dalam proyek Anda.create_sandbox() untuk membuat direktori independen.safe_git_commit() untuk mengindari konflik lock menggunakan exponential backoff.Dengan mengubah konfigurasi seperti ini, konflik penimpaan berkas akan hilang. Waktu yang terbuang untuk debugging pun berkurang lebih dari 5 jam per minggu.
Saat membawa kembali cabang yang telah diubah ke codebase utama, gunakan analisis Abstract Syntax Tree (AST) sebagai pengganti penggabungan baris teks biasa agar lebih aman. Penggabungan teks biasa bisa memicu konflik hanya karena perubahan posisi pernyataan import di bagian atas. Dengan melakukan parsing kode sumber menjadi node tree sintaksis menggunakan modul bawaan Python ast atau Tree-Sitter, lalu menggabungkannya berdasarkan fungsi atau kelas, tingkat kegagalan merge akan turun hingga mendekati 0%.
Menggunakan Claude 3.5 Sonnet secara merata untuk semua jenis tugas akan membuat biaya tidak terkendali. Kita tidak boleh menilai kompleksitas hanya dari jumlah baris kode (LOC). Kode sepanjang 100 baris yang penuh dengan operator ternari rumit dan kondisi bersarang jauh lebih sulit daripada data class 500 baris yang hanya berisi komentar.
Dengan modul ast, Anda dapat menghitung jumlah node, kompleksitas siklomatis (cyclomatic complexity), serta kedalaman tree untuk diubah menjadi skor terukur.
`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}
`
Letakkan penganalisis ini di pintu masuk pipeline backend dan tetapkan skor ambang batas routing pada angka 50.
Tugas dengan skor di bawah 50 seperti penulisan unit test, implementasi utilitas, atau definisi DTO akan dialihkan ke Claude 3.5 Haiku yang berbiaya sekitar $0,80 per satu juta token input. Hanya tugas skala besar seperti refactoring atau perancangan arsitektur dengan skor 50 ke atas yang dialihkan ke Claude 3.5 Sonnet seharga $3,00 per satu juta token. Memastikan Haiku menangani lebih dari 60% dari total traffic saja sudah dapat memangkas biaya API hingga 60%.
Masalah kebocoran token akibat konteks percakapan yang terlalu panjang dapat diatasi menggunakan middleware reset session. Kumpulkan akumulasi token, dan saat mencapai ambang batas, reset percakapan secara paksa. Pada saat ini, buat ringkasan menggunakan AST yang hanya mengambil simbol fungsi utama dan TODO yang tersisa, lalu masukkan ringkasan tersebut ke dalam prompt pertama sesi baru agar pekerjaan dapat dilanjutkan tanpa kehilangan konteks.
Menggabungkan kode draf yang dibuat oleh agen langsung ke repositori dapat merusak proses build. Namun, memanggil LLM kembali hanya untuk memperbaiki salah ketik (typo) atau kesalahan tata bahasa (syntax error) sederhana akan memakan waktu lama dan membuang-buang uang.
Buatlah pipeline verifikasi bertahap yang menggabungkan linter, type checker, dan reviewer 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)
`
Tahap 1 menangkap kesalahan sintaksis menggunakan parsing AST dan Ruff, kemudian Tahap 2 menyelaraskan tipe data menggunakan Mypy. Hanya kode yang lolos dari semua alat verifikasi statis ini yang dikirim ke Tahap 3, yaitu deep reviewer Claude 3.5 Sonnet. Ini menghilangkan pemanggilan ulang LLM akibat masalah sepele seperti kurang kurung atau type error, sehingga kecepatan penyelesaian pipeline meningkat 40%.
Circuit breaker sangat penting untuk mencegah agen terjebak dalam loop saat gagal melewatin verifikasi. Batasi percobaan ulang untuk kesalahan yang sama maksimal 3 kali. Jika nilai hash dari pesan kesalahan benar-benar sama seperti sebelumnya, sistem harus menilai bahwa agen telah terjebak dalam hallucination loop dan segera menghentikan eksekusi.
Untuk mengelola secara terpusat berkas dan cabang mana yang sedang diubah oleh setiap agen, setidaknya Anda memerlukan skema 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
);
`
Setiap kali seorang agen berhasil melakukan commit pada modul bersama, event FILE_MUTATED akan dipublikasikan. Agen lain yang menerima notifikasi ini akan langsung memperbarui definisi simbol AST yang mereka rujuk ke versi terbaru.
Jika agen tertentu gagal dalam verifikasi dan berada dalam kondisi tidak dapat dipulihkan, lakukan atomic rollback ke SHA commit snapshot yang dicatat saat pertama kali memulai tugas.
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
Dengan menghubungkan direktori terisolasi, routing model berbasis sintaksis, pipeline verifikasi statis, dan DB status seperti ini, Anda dapat menjalankan agent swarm kelas produksi secara stabil tanpa perlu khawatir tentang konflik berkas atau membengkaknya biaya.