Git Worktree और 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
डेमो में बहुत बढ़िया काम करने वाले LLM एजेंट्स को जब स्वार्म (Swarm) स्तर पर स्केल किया जाता है, तो वे निश्चित रूप से दो दीवारों से टकराते हैं। एजेंट्स द्वारा एक ही फ़ाइल को ओवरराइट करने से कोड उलझ जाता है, या फिर निरर्थक फ़ाइलों को भी बड़े मॉडल में धकेलने से API लागत लाखों में पहुँच जाती है।
एक ही वर्कस्पेस में कई भाषा मॉडल प्रोसेस को बिना सोचे-समझे चलाने से स्थिति जल्दी ही बिगड़ जाती है। एजेंट A जिस अधूरी फ़ाइल को ठीक कर रहा होता है, एजेंट B उसे ही पढ़कर गलत कोड लिख देता है, और अंततः कमिट इतिहास भी गायब हो जाता है। इसके बावजूद अगर हर बार पूरे रिपॉजिटरी की कॉपी बनाई जाए, तो डिस्क स्पेस बर्बाद होता है और इनिशियलाइजेशन में ही कई मिनट लग जाते हैं।
इन-मेमोरी फ़ाइल आइसोलेशन, सिंटैक्स विश्लेषण-आधारित राउटिंग और स्टैटिक वैलिडेसन पाइपलाइन को जोड़कर इस समस्या को इंजीनियरिंग से हल करने के तरीके पर यहाँ चर्चा की गई है।
बड़े कोडबेस में एक साथ कई एजेंट्स चलाने पर उत्पन्न होने वाली सबसे बड़ी बाधा फ़ाइल सिस्टम विवाद (contention) है। पूरे मोनोलिथिक रिपॉजिटरी का फ़ुल क्लोन (Full Git Clone) करने के बजाय Git Worktree का उपयोग करने से, मेटाडेटा और ऑब्जेक्ट DB को साझा करते हुए कुछ MB स्तर की हल्की डायरेक्टरी को 1 सेकंड में अलग किया जा सकता है।
हालाँकि, जब दर्जनों एजेंट्स एक साथ कमिट करते हैं, तो पैरेंट इंडेक्स फ़ाइल (.git/index.lock) पर लॉक विवाद उत्पन्न होता है। इसे नियंत्रित करने के लिए फ़ाइल लॉकिंग-आधारित सैंडबॉक्स लेयर की आवश्यकता होती है।
`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)
`
लागू करने की प्रक्रिया सरल है:
filelock पैकेज इंस्टॉल करें और अपने प्रोजेक्ट में WorktreeSandboxManager क्लास जोड़ें।create_sandbox() को कॉल करें।safe_git_commit() से गुज़ारें ताकि एक्सपोनेंशियल बैकऑफ़ के ज़रिए लॉक टकराव से बचा जा सके।इस कॉन्फ़िगरेशन में बदलाव करने से ओवरराइट टकराव गायब हो जाते हैं। डिबगिंग में बर्बाद होने वाला समय भी प्रति सप्ताह 5 घंटे से अधिक कम हो जाता है।
संशोधित ब्रांच को मुख्य कोडबेस में वापस लाते समय, टेक्स्ट लाइन-आधारित मर्ज के बजाय एब्स्ट्रैक्ट सिंटैक्स ट्री (AST) विश्लेषण का उपयोग करना सुरक्षित होता है। साधारण टेक्स्ट मर्ज में सिर्फ़ शीर्ष पर स्थित import स्टेटमेंट की जगह बदलने पर भी टकराव उत्पन्न हो जाता है। Python के इन-बिल्ट ast मॉड्यूल या Tree-Sitter से सोर्स कोड को सिंटैक्स नोड ट्री में पार्स करके, फिर फ़ंक्शन या क्लास स्तर पर मर्ज करने से मर्ज विफलता दर लगभग 0% तक गिर जाती है।
अगर सभी कार्यों के लिए एक समान Claude 3.5 Sonnet का उपयोग किया जाए, तो लागत संभालना असंभव हो जाएगा। जटिलता का आकलन केवल लाइनों की संख्या (LOC) से नहीं किया जाना चाहिए। केवल टिप्पणियों (comments) से भरी 500-लाइन की डेटा क्लास की तुलना में, उलझे हुए टर्नरी ऑपरेटरों और नेस्टेड शर्तों से भरी 100-लाइन का कोड कहीं अधिक कठिन होता है।
ast मॉड्यूल का उपयोग करके, आप नोड्स की संख्या, साइक्लोमैटिक जटिलता और ट्री की गहराई की गणना करके इसे एक ऑब्जेक्टिव स्कोर में बदल सकते हैं।
`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}
`
इस एनालाइजर को बैकएंड पाइपलाइन के प्रवेश द्वार पर रखें और राउटिंग थ्रेशोल्ड स्कोर 50 अंक निर्धारित करें।
50 अंक से कम वाले कार्य जैसे यूनिट टेस्ट लिखना, यूटिलिटी लागू करना, और DTO परिभाषित करना आदि Claude 3.5 Haiku को भेजे जाते हैं, जिसकी लागत प्रति मिलियन इनपुट टोकन लगभग $0.80 है। 50 अंक से अधिक वाले बड़े पैमाने के रिफैक्चरिंग या आर्किटेक्चर डिज़ाइन जैसे कार्य ही केवल Claude 3.5 Sonnet को भेजे जाते हैं, जिसकी लागत प्रति मिलियन टोकन $3.00 है। कुल ट्रैफ़िक का 60% से अधिक हिस्सा Haiku से प्रोसेस करवाने मात्र से API लागत में 60% तक की कटौती होती है।
लंबे संवाद संदर्भों के कारण टोकन लीक होने की समस्या को सेशन रीसेट मिडलवेयर से रोका जाता है। संचित टोकन की गणना करते समय जब यह थ्रेशोल्ड तक पहुँचता है, तो संवाद को ज़बरदस्ती रीसेट कर दिया जाता है। इस समय AST के माध्यम से मुख्य फ़ंक्शन सिंबल और शेष TODO का एक सारांश तैयार किया जाता है, जिसे नए सेशन के पहले प्रॉम्ट के रूप में पास कर दिया जाता है, जिससे संदर्भ खोए बिना काम जारी रखा जा सकता है।
एजेंट द्वारा बनाए गए ड्राफ्ट कोड को सीधे रिपॉजिटरी में मर्ज करने से बिल्ड टूट जाता है। इसके बावजूद, साधारण टाइपो या सिंटैक्स त्रुटियों को ठीक करने के लिए LLM को फिर से कॉल करने से समय भी अधिक लगता है और पैसे भी बर्बाद होते हैं।
एक वैलिडेसन पाइपलाइन तैयार करें जो लिंटर, टाइप चेकर और 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)
`
चरण 1 में AST पार्सिंग और Ruff के साथ सिंटैक्स त्रुटियों को पकड़ा जाता है, और चरण 2 में Mypy के साथ प्रकारों (types) का मिलान किया जाता है। केवल वही कोड जो इन सभी स्टैटिक वैलिडेसन टूल को पास करता है, चरण 3 के रूप में Claude 3.5 Sonnet गहन समीक्षक को भेजा जाता है। साधारण कोष्ठक छूटने या टाइप एरर के कारण LLM को फिर से कॉल करने की आवश्यकता समाप्त हो जाती है, जिससे पाइपलाइन पूरा होने की गति 40% तेज़ हो जाती है।
वैलिडेसन विफलता के मामले में लूप में फँसने से रोकने के लिए सर्किट ब्रेकर आवश्यक है। एक ही त्रुटि के लिए पुन: प्रयास को अधिकतम 3 बार तक सीमित करें, और यदि त्रुटि संदेश का हैश मान पहले जैसा ही है, तो यह मानकर कि एजेंट हैलुसिनेशन लूप में फँस गया है, निष्पादन को तुरंत रोक दिया जाना चाहिए।
विभिन्न एजेंट्स किन फ़ाइलों और ब्रांचों को छू रहे हैं, इसे केंद्रीय रूप से प्रबंधित करने के लिए कम से कम एक 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
);
`
जब भी कोई एजेंट किसी कॉमन मॉड्यूल को सफलतापूर्वक कमिट करता है, तो वह FILE_MUTATED इवेंट जारी करता है। अन्य एजेंट्स यह सूचना प्राप्त करते हैं और अपने द्वारा संदर्भित AST सिंबल परिभाषाओं को तुरंत अपडेट करते हैं।
यदि कोई विशेष एजेंट वैलिडेसन विफल होने के बाद अपूरणीय स्थिति में चला जाता है, तो कार्य शुरू करते समय सहेजे गए स्नैपशॉट कमिट SHA का उपयोग करके एटॉमिक रोलबैक (atomic rollback) निष्पादित किया जाता है।
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
इस प्रकार आइसोलेशन डायरेक्टरी, सिंटैक्स-आधारित मॉडल राउटिंग, स्टैटिक वैलिडेसन पाइपलाइन और स्टेट DB को एक साथ जोड़कर, आप फ़ाइल टकराव या लागत विस्फोट की चिंता के बिना प्रोडक्शन-ग्रेड एजेंट स्वार्म को स्थिर रूप से चला सकते हैं।