Когда вы масштабируете LLM-агентов, отлично работавших в демо-версии, до уровня роя (swarm), вы неизбежно сталкиваетесь с двумя препятствиями. Агенты перезаписывают одни и те же файлы, из-за чего путается код, либо загружают в крупную модель даже нерелевантные файлы, выжигая миллионы вон на расходы за API.
Бессистемный запуск нескольких процессов языковых моделей в одном рабочем пространстве быстро ухудшает ситуацию. Агент B считывает незавершенный файл, над которым в данный момент работает агент A, и пишет неподходящий код, что в итоге приводит к потере истории коммитов. С другой стороны, копирование всего репозитория каждый раз приводит к нерациональному расходу дискового пространства, а сама инициализация занимает несколько минут.
В этой статье мы рассмотрим инженерный подход к решению этой проблемы с помощью объединения изоляции файлов в памяти, маршрутизации на основе синтаксического анализа и пайплайна статической проверки.
Блокировка перезаписи файлов с помощью Git Worktree
Узким местом, возникающим при одновременном запуске агентов в крупной кодовой базе, является конкуренция за файловую систему. Вместо полного клонирования (Full Git Clone) всего монолитного репозитория использование Git Worktree позволяет разделять метаданные и базу данных объектов, изолируя легковесные директории размером в несколько мегабайт всего за 1 секунду.
Однако, когда десятки агентов делают коммиты одновременно, возникает конкуренция за блокировку верхнего индексного файла (.git/index.lock). Для управления этим процессом необходим слой песочницы (sandbox) на основе блокировки файлов.
`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(), чтобы избежать конфликтов блокировки за счет экспоненциальной задержки (exponential backoff).
Такая смена архитектуры устраняет конфликты перезаписи. Время, тратившееся на отладку, сокращается более чем на 5 часов в неделю.
При переносе измененной ветки обратно в основную кодовую базу в целях безопасности следует использовать анализ абстрактного синтаксического дерева (AST) вместо построчного слияния текста. Простое текстовое слияние может привести к конфликтам даже из-за изменения расположения операторов import в начале файла. Если спарсить исходный код в дерево синтаксических узлов с помощью встроенного модуля ast в Python или библиотеки Tree-Sitter, а затем выполнить слияние на уровне функций или классов, частота ошибок слияния упадет практически до 0%.
Сокращение расходов на токены на 60% с помощью измерения сложности AST
Если подключать Claude 3.5 Sonnet к абсолютно каждой задаче, затраты станут неподъемными. Нельзя судить о сложности просто по количеству строк (LOC). Код из 100 строк, наполненный запутанными тернарными операторами и вложенными условными переходами, гораздо сложнее класса данных из 500 строк, состоящего в основном из комментариев.
С помощью модуля 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 с стоимостью порядка 3.00 за миллион токенов). Одно лишь перенаправление более 60% всего трафика на Haiku сокращает затраты на API вплоть до 60%.
Утечка токенов из-за длинного контекста диалога предотвращается с помощью middleware для сброса сессии. При подсчете накопленных токенов, когда достигается пороговое значение, диалог принудительно сбрасывается. В этот момент на основе AST формируется сводка с выжимкой ключевых символов функций и оставшихся TODO, которая передается в качестве первого промпта новой сессии, позволяя продолжить работу без потери контекста.
Увеличение скорости работы на 40% благодаря детерминированной цепочке проверок
Если сразу сливать черновой код, созданный агентом, в репозиторий, сборка сломается. Однако повторный вызов 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 проверяет соответствие типов. Только код, прошедший все эти инструменты статической валидации, отправляется на 3-й этап — углубленный ревью моделью Claude 3.5 Sonnet. Повторные вызовы LLM из-за пропущенных скобок или ошибок типов исчезают, что ускоряет время прохождения пайплайна на 40%.
Чтобы предотвратить зацикливание при неудачной проверке, необходим предохранитель (circuit breaker). Количество повторных попыток для одной и той же ошибки ограничивается максимум 3 раза, а если хэш сообщения об ошибке полностью совпадает с предыдущим, система считает, что агент попал в цикл галлюцинаций, и немедленно прерывает выполнение.
Общая память и транзакционный откат (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
);
`
Каждый раз, когда агент успешно делает коммит в общий модуль, он публикует событие FILE_MUTATED. Другие агенты получают это уведомление и немедленно обновляют определения AST-символов, на которые они ссылаются.
Если конкретный агент после сбоя проверки переходит в невосстанавливаемое состояние, выполняется атомарный откат к SHA-хэшу снапшот-коммита, сделанному в самом начале работы.
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
Связывая таким образом изолированные директории, маршрутизацию моделей на основе синтаксиса, пайплайн статической проверки и БД состояний, вы сможете стабильно запускать рой агентов продакшен-уровня без опасений о конфликтах файлов или взрывном росте расходов.
