عند التوسع في تشغيل وكلاء النماذج اللغوية الكبيرة (LLM) التي كانت تعمل بشكل جيد في العروض التوضيحية لتصبح سربًا كاملاً، ستصطدم حتمًا بعقبتين رئيسيتين: إما أن يقوم الوكلاء بالكتابة فوق نفس الملفات مما يؤدي إلى تداخل الأكواد وتعقدها، أو يتم إقحام ملفات غير ذات صلة في النماذج الكبيرة مما يؤدي إلى إنفاق ملايين الكوريات على تكاليف واجهة برمجة التطبيقات (API).
إن تشغيل عمليات متعددة لنماذج اللغة في بيئة عمل واحدة بشكل عشوائي يؤدي إلى تفاقم الوضع سريعًا. حيث يقدم الوكيل B على قراءة ملف غير مكتمل يعمل الوكيل A على تعديله، مما ينتج عنه كتابة كود خاطئ، وينتهي الأمر بضياع سجل الالتزامات (commit history). ومن ناحية أخرى، فإن نسخ المستودع بأكمله في كل مرة يستنزف مساحة القرص ويستغرق دقائق معدودة لمجرد التهيأة.
نتناول في هذه المقالة كيفية حل هذه المشكلة هندسيًا عن طريق الدمج بين عزلة الملفات في الذاكرة (in-memory file isolation)، والتوجيه القائم على التحليل النحوي، وسلسلة التدقيق الاستاتيكي.
منع الكتابة فوق الملفات باستخدام Git Worktree
تكمن العقبة الرئيسية التي تظهر عند تشغيل الوكلاء بالتزامن في قواعد الأكواد الضخمة في التنافس على نظام الملفات. بدلاً من استنساخ المستودع الأحادي بالكامل (Full Git Clone)، يتيح استخدام Git Worktree مشاركة البيانات الوصفية وقاعدة بيانات الكائنات مع فصل دليل خفيف الوزن بحجم عدة ميجابايت في ثانية واحدة فقط.
ومع ذلك، إذا قام عشرات الوكلاء بعمل التزامات (commits) في نفس الوقت، فسوف يحدث تنافس على القفل في ملف الفهرس الرئيسي (.git/index.lock). وللتحكم في ذلك، يلزم وجود طبقة بيئة معزولة (sandbox layer) تعتمد على قفل الملفات.
`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()لإنشاء دليل مستقل. - عند إجراء التزام (commit) بعد إكمال المهمة، اجعل العملية تمر عبر
safe_git_commit()لتجنب تعارضات القفل باستخدام التراجع الأسي (exponential backoff).
من خلال تغيير الهيكلية بهذه الطريقة، تتلاشى تعارضات الكتابة فوق الملفات تمامًا. كما يتراجع الوقت الضائع في تصحيح الأخطاء (debugging) بأكثر من 5 ساعات أسبوعيًا.
وعند دمج الفرع التعديلي مرة أخرى في قاعدة الكود الرئيسية، يلزم استخدام تحليل شجرة النحو المجرد (AST) لضمان الأمان بدلاً من الدمج على مستوى أسطر النصوص. فغالباً ما يتسبب الدمج النصي البسيط في حدوث تعارضات لمجرد تغير موقع عبارة import في الأعلى. أما عند تحليل الكود المصدري إلى شجرة عقد نحوية باستخدام وحدة ast المدمجة في Python أو Tree-Sitter ثم دمجها على مستوى الدالة أو الفئة، فإن معدل فشل الدمج ينخفض إلى الصفر تقريبًا.
تقليل تكلفة الرموز بنسبة 60% عبر قياس تعقيد AST
إن ربط نموذج Claude 3.5 Sonnet بجميع المهام دون تمييز أمر مكلف للغاية وغير مستدام. ولا ينبغي تقييم التعقيد بناءً على عدد الأسطر (LOC) فحسب؛ فكود مكون من 100 سطر مليء بالعمليات الشرطية الثلاثية المعقدة والجمل الشرطية المتداخلة أصعب بكثير من كود بيانات مكون من 500 سطر مليء بالتعليقات فقط.
باستخدام وحدة ast، يمكن حساب عدد العقد، والتعقيد الدوري (cyclomatic complexity)، وعمق الشجرة، وتحويلها إلى درجة موضوعية.
`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}
`
نضع هذا المحلل عند مدخل أنبوب الخلفية (backend pipeline) ونحدد النقطة المرجعية للتوجيه بـ 50 درجة.
المهام التي تحرز أقل من 50 درجة — مثل كتابة اختبارات الوحدة، وتطبيق الأدوات المساعدة، وتعريف كائنات DTO — يتم تحويلها إلى Claude 3.5 Haiku بتكلفة تبلغ حوالي $0.80 لكل مليون رمز إدخال. بينما توجّه المهام الكبيرة فقط مثل إعادة الهيكلة الشاملة (refactoring) أو تصميم البنية التحتية، والتي تتجاوز 50 درجة، إلى Claude 3.5 Sonnet بتكلفة $3.00 لكل مليون رمز. ومن خلال جعل Haiku يتعامل مع أكثر من 60% من إجمالي حركة المرور، تقل تكاليف واجهة برمجة التطبيقات بنسبة تصل إلى 60%.
أما مشكلة تسرب الرموز الناتجة عن سياقات المحادثات الطويلة، فيتم معالجتها عبر برمجيات تعيين الجلسات الوسيطة (session reset middleware). فإذا تجمعت الرموز ووصلت إلى الحد الأقصى، يتم إعادة ضبط المحادثة إجباريًا. وفي هذه اللحظة، يتم إنشاء ملخص يتضمن رموز الدوال الرئيسية والمهام المتبقية (TODO) فقط باستخدام AST، ويتم إقحامه في الموجه الأول للجلسة الجديدة، مما يسمح بمواصلة العمل دون فقدان السياق.
زيادة سرعة العمل بنسبة 40% عبر سلسلة تدقيق حتمية
دمج كود المسودة الذي أنشأه الوكيل بشكل مباشر في المستودع قد يؤدي إلى كسر عملية البناء (build). وفي المقابل، فإن إعادة استدعاء نموذج اللغة الكبير لمجرد تصحيح أخطاء إملائية أو أخطاء نحوية بسيطة يستهلك الوقت والمال سدى.
لذا يتم إنشاء خط أنابيب تدقيق يربط أدوات التنسيق (linners)، ومدقق الأنواع (type checkers)، ومراجع النموذج اللغوي خطوة بخطوة.
`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)
`
في المرحلة الأولى، يتم التعامل مع الأخطاء النحوية بواسطة تحليل AST وأداة Ruff، وفي المرحلة الثانية يتم التحقق من الأنواع بواسطة Mypy. ولا يُرسل الكود إلا بعد اجتيازه لجميع أدوات التحقيق الاستاتيكية هذه إلى المرحلة الثالثة المتمثلة في المراجع الخبير Claude 3.5 Sonnet. تضمن هذه العملية عدم استدعاء النموذج اللغوي مجددًا بسبب قوس مفقود أو خطأ في النوع، مما يزيد من سرعة إكمال خط الأنابيب بنسبة 40%.
ولمنع الوقوع في حلقة مفرغة عند فشل التدقيق، فإن وجود قاطع الدائرة (circuit breaker) أمر ضروري. ويجب تحديد محاولات إعادة المحاولة لنفس الخطأ بـ 3 مرات كحد أقصى، وإذا كانت قيمة تجزئة (hash) رسالة الخطأ مطابقة تمامًا للسابقة، فيجب اعتبار أن الوكيل قد وقع في حلقة هلوسة وإيقاف التنفيذ فورًا.
الذاكرة المشتركة والتراجع عن المعاملات (Transaction Rollback)
لإدارة الملفات والفروع التي يتفاعل معها الوكلاء المتعددون من مكان مركزي، يلزم وجود مخطط 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
);
`
كلما نجح وكيل في إجراء التزام (commit) على وحدة مشتركة، يتم إصدار حدث FILE_MUTATED. ويتلقى الوكلاء الآخرون هذا الإشعار ليقوموا بتحديث تعريفات رموز AST التي يشيرون إليها فورًا.
وإذا دخل وكيل معين في حالة غير قابلة للاسترداد بعد فشل التدقيق، يتم تنفيذ تراجع ذري (atomic rollback) باستخدام المعرف الفريد للتزام اللقطة (Snapshot Commit SHA) الذي تم تسجيله عند بدء العمل.
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
من خلال ربط الأدلة المعزولة، والتوجيه القائم على القواعد النحوية، وسلسلة التدقيق الاستاتيكي، وقاعدة بيانات الحالة بهذه الطريقة، يمكنك تشغيل سرب من الوكلاء الجاهزين للإنتاج بانسجام واستقرار، دون القلق بشأن تضارب الملفات أو الانفجار في التكاليف.
