利用 Git Worktree 与 AST 分析防止多 Agent 代码冲突与 Token 暴涨
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
在 Demo 中运行良好的 LLM Agent，一旦扩展到 Swarm 蜂群级别，就必然会撞上两座大山：一是 Agent 之间因覆盖同一个文件而导致代码错乱，二是不管三七二十一将无意义的文件一股脑塞给大模型，导致 API 费用飙升至数万元。
如果在同一个工作空间中盲目运行多个语言模型进程，局势很快就会恶化。Agent A 正在修改的未完成文件被 Agent B 读取并写出荒谬的代码，最终甚至连提交历史都会丢失。但如果每次都完整复制整个仓库，又会浪费磁盘空间，且仅初始化就要耗费数分钟。
本文将探讨如何结合内存文件隔离、基于语法分析的路由以及静态检查流水线，通过工程化手段解决这一难题。
在大型代码库中同时运行多个 Agent 时，出现的瓶颈是文件系统竞争。与其对整个 Monorepo 进行 Full Git Clone，不如使用 Git Worktree。它可以在共享元数据和对象数据库的同时，在 1 秒内分离出仅数 MB 大小的轻量级目录。
不过，当数十个 Agent 同时进行提交时，上层索引文件（.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 等），交由每百万输入 Token 仅约 $0.80 的 Claude 3.5 Haiku 处理。只有得分高于 50 分的大规模重构或架构设计任务，才路由给每百万 Token $3.00 的 Claude 3.5 Sonnet。仅让 Haiku 处理 60% 以上的总流量，就能将 API 成本最高削减 60%。
至于因长对话上下文导致的 Token 泄露问题，可以通过 Session 重置中间件来截断。在统计累积 Token 达到阈值时强制重置对话。此时，利用 AST 提取出核心函数符号和剩余 TODO 并生成摘要，将其作为新 Session 的首个 Prompt 注入，即可在无上下文损失的情况下继续作业。
如果直接将 Agent 生成的草稿代码合并到仓库中，构建就会崩溃。但如果为了捕获简单的拼写错误或语法错误而重新调用 LLM，既耗时又浪费钱。
可以构建一个按阶段挂载 Linter、类型检查器和 LLM Reviewer 的检查流水线。
`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 深度 Reviewer。由于不再会因为简单的括号缺失或类型错误而重新调用 LLM，流水线的完成速度提升了 40%。
为防止在验证失败时陷入死循环，断路器（Circuit Breaker）是必不可少的。对于同一个错误，重试次数应限制在最多 3 次；如果错误消息的 Hash 值与之前完全一致，则判定 Agent 已陷入幻觉循环，必须立即终止执行。
为了集中管理多个 Agent 修改了哪些文件和分支，至少需要具备 SQLite 级别的 Schema 设计。
`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
);
`
每当 Agent 成功提交公共模块时，就会发布 FILE_MUTATED 事件。其他 Agent 收到此通知后，会立即将其引用的 AST 符号定义更新至最新状态。
如果某个 Agent 在检查失败后陷入无法恢复的状态，则使用刚开始工作时打下的快照 Commit SHA 执行原子回滚。
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
将隔离目录、基于语法的模型路由、静态检查流水线和状态数据库结合在一起，就可以稳定运行生产级的 Agent Swarm，而无需担心文件冲突或成本暴涨。