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エージェントをスウォーム単位に拡張すると、必ず2つの壁にぶつかります。エージェント同士が同じファイルを上書きしてコードが混乱するか、意味のないファイルまで大型モデルに投入した結果、API費用で数十万円が吹き飛ぶという問題です。
同一の作業領域で複数の言語モデルプロセスを無計画に回すと、事態はすぐに悪化します。エージェントAが修正中の未完成ファイルをエージェントBが読み込んでとんちんかんなコードを書き、最終的にはコミット履歴まで消失します。だからといってリポジトリ全体を毎回コピーすると、ディスク容量が浪費され、初期化だけに数分がかかってしまいます。
インメモリファイル隔離、構文解析ベースのルーティング、静的検証パイプラインを組み合わせ、この問題をエンジニアリングで解決する方法を解説します。
大容量コードベースでエージェントを同時に動かす際に発生するボトルネックは、ファイルシステムの競合です。モノリシックリポジトリ全体をフルクローン（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）で複雑度を判断してはいけません。コメントばかりの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定義といった作業は、入力100万トークンあたり3.00の Claude 3.5 Sonnet へルーティングします。全トラフィックの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による型チェックを行います。これらの静的検証ツールをすべて通過したコードのみを、ステージ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を使用してアトミックなロールバックを実行します。
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
このように隔離ディレクトリ、構文ベースのモデルルーティング、静的検証パイプライン、状態DBを連携させることで、ファイルの衝突やコスト爆発を心配することなく、プロダクションレベルのエージェントスウォームを安定して運用できます。