목록으로 돌아가기

利用 Git Worktree 与 AST 分析防止多 Agent 代码冲突与 Token 暴涨

Computing/Software27분 읽기

TuBrief 편집팀 원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다. 원본 영상: 从 Cursor 的 SQLite Rust 实验中，我们能学到什么 · Maximilian Schwarzmüller

在 Demo 中运行良好的 LLM Agent，一旦扩展到 Swarm 蜂群级别，就必然会撞上两座大山：一是 Agent 之间因覆盖同一个文件而导致代码错乱，二是不管三七二十一将无意义的文件一股脑塞给大模型，导致 API 费用飙升至数万元。

如果在同一个工作空间中盲目运行多个语言模型进程，局势很快就会恶化。Agent A 正在修改的未完成文件被 Agent B 读取并写出荒谬的代码，最终甚至连提交历史都会丢失。但如果每次都完整复制整个仓库，又会浪费磁盘空间，且仅初始化就要耗费数分钟。

本文将探讨如何结合内存文件隔离、基于语法分析的路由以及静态检查流水线，通过工程化手段解决这一难题。

利用 Git Worktree 阻断文件覆盖

在大型代码库中同时运行多个 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)

`

应用步骤非常简单：

  1. 安装 filelock 包，并将 WorktreeSandboxManager 类引入项目中。
  2. 在给 Agent 分配任务之前，调用 create_sandbox() 开辟一个独立目录。
  3. 任务完成后提交时，使其通过 safe_git_commit() 处理，利用指数退避算法避免锁冲突。

这样调整架构后，覆盖冲突将不复存在。每周在排查问题上浪费的时间也能减少 5 个小时以上。

将修改后的分支合并回主代码库时，为了安全起见，应该使用抽象语法树（AST）分析，而非基于文本行的合并。简单的文本合并哪怕只是顶部的 import 语句位置发生变化，也会抛出冲突。利用 Python 内置的 ast 模块或 Tree-Sitter 将源代码解析为语法节点树，再以函数或类为单位进行合并，可以让合并失败率几乎降为 0%。

通过 AST 复杂度测量降低 60% 的 Token 成本

如果给所有任务都接入 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 注入，即可在无上下文损失的情况下继续作业。

利用确定性检查链提升 40% 的执行速度

如果直接将 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，而无需担心文件冲突或成本暴涨。

관련 영상

从 Cursor 的 SQLite Rust 实验中，我们能学到什么

Maximilian Schwarzmüller32:24

커뮤니티의 다른 글

댓글

로그인하고 댓글을 남겨 보세요.

아직 댓글이 없습니다.