生成AI入門 2026.04.21

8GB VRAMでLLMエージェントがクラッシュする原因と対策【メモリ管理コード付き】

タグ:ローカルLLM / VRAM / エージェント / メモリ管理 / GPU

8GB VRAMで何が起きているのか

8GB VRAMのGPU(RTX 3070/4060など)でLLMエージェントを動かすと、ツール呼び出しを5〜6回繰り返した時点でOOM(Out of Memory)エラーが発生してクラッシュする——これは多くの開発者が経験する問題です。

原因はVRAMの累積消費です:

消費源典型的な使用量(7Bモデル/Q4)
モデルウェイト約4GB(固定)
KVキャッシュ(4096トークン)約1GB
Activationメモリ約0.5〜1GB
ツール結果・会話履歴累積で1〜2GB(変動大)
合計6.5〜8GB超

ツール呼び出しのたびにコンテキストが膨らみ、KVキャッシュが増加します。ウェイト(モデルの重み)は固定ですが、KVキャッシュはコンテキスト長に比例して増え続けます

メモリが増える具体的なメカニズム

ツール1回の呼び出しで何が追加されるか

呼び出し前のコンテキスト: 1,000トークン(会話履歴)
↓ ツール実行
呼び出し後のコンテキスト: 1,000 + ツール結果3,000 = 4,000トークン

検索ツールが「上位10件の検索結果全文」を返した場合、1回の呼び出しで数千トークンが追加されます。5回繰り返すと:

初期:     1,000 トークン → KVキャッシュ: ~250MB
5回後: ~16,000 トークン → KVキャッシュ: ~4GB

7Bモデルのウェイト4GBと合わせると8GBを超えてOOMになります。

対策1:コンテキスト長を監視してツール呼び出しを制限する

# agent_with_memory_guard.py
from anthropic import Anthropic
import json
from typing import Any

client = Anthropic()

MAX_CONTEXT_TOKENS = 6000   # VRAMに応じて調整(8GB環境では保守的に設定)
MAX_TOOL_CALLS = 5          # ツール呼び出しの上限

def count_message_tokens(messages: list[dict]) -> int:
    """メッセージのおおよそのトークン数を計算(1文字≒1トークンで概算)"""
    total_chars = sum(
        len(str(msg.get("content", "")))
        for msg in messages
    )
    return total_chars // 4  # 英語は約4文字=1トークンで概算

def run_agent_with_guard(
    user_input: str,
    tools: list[dict],
    tool_handler: callable,
) -> str:
    messages = [{"role": "user", "content": user_input}]
    tool_call_count = 0

    while True:
        # コンテキスト長チェック
        estimated_tokens = count_message_tokens(messages)
        if estimated_tokens > MAX_CONTEXT_TOKENS:
            return f"コンテキスト上限に達しました(推定{estimated_tokens}トークン)。処理を終了します。"

        # ツール呼び出し数チェック
        if tool_call_count >= MAX_TOOL_CALLS:
            return f"ツール呼び出し上限({MAX_TOOL_CALLS}回)に達しました。"

        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

        # テキスト応答で終了
        if response.stop_reason == "end_turn":
            text_blocks = [b for b in response.content if b.type == "text"]
            return text_blocks[0].text if text_blocks else ""

        # ツール使用
        if response.stop_reason == "tool_use":
            tool_call_count += 1
            messages.append({"role": "assistant", "content": response.content})

            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = tool_handler(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result)[:2000],  # 結果を2000文字に制限
                    })

            messages.append({"role": "user", "content": tool_results})
        else:
            break

    return "予期しない終了"

対策2:古い会話履歴を要約して削除する

コンテキストが肥大化したら、古い部分を要約に置き換えます。

# context_compressor.py
from anthropic import Anthropic

client = Anthropic()

def compress_old_messages(
    messages: list[dict],
    keep_recent: int = 4,
    max_tokens: int = 3000,
) -> list[dict]:
    """
    古いメッセージを要約に置き換えてコンテキストを圧縮する。
    keep_recent: 最近のメッセージは圧縮せずに保持する件数
    """
    if len(messages) <= keep_recent:
        return messages

    # 古いメッセージと最新メッセージを分割
    old_messages = messages[:-keep_recent]
    recent_messages = messages[-keep_recent:]

    # 古いメッセージを要約
    old_text = "\n".join(
        f"{msg['role']}: {str(msg.get('content', ''))[:500]}"
        for msg in old_messages
    )

    summary_response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 要約は軽量モデルで十分
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": f"以下の会話を150字以内で要約してください:\n\n{old_text}"
        }]
    )
    summary = summary_response.content[0].text

    # 圧縮されたメッセージリストを構築
    compressed = [
        {
            "role": "user",
            "content": f"[これまでの会話の要約: {summary}]"
        },
        {
            "role": "assistant",
            "content": "了解しました。続きを進めます。"
        }
    ] + recent_messages

    print(f"コンテキスト圧縮: {len(old_messages)}件 → 要約({len(summary)}字)")
    return compressed


def agent_with_compression(user_input: str) -> str:
    messages = [{"role": "user", "content": user_input}]

    for _ in range(10):  # 最大10ラウンド
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=1024,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return response.content[0].text

        messages.append({"role": "assistant", "content": response.content})

        # コンテキストが4件を超えたら圧縮
        if len(messages) > 8:
            messages = compress_old_messages(messages, keep_recent=4)

    return "処理完了"

対策3:ツール結果のサイズを制限する

大きなツール結果が最大の原因になることが多いです。

# tool_result_limiter.py

def limit_tool_result(result: Any, max_chars: int = 1500) -> str:
    """
    ツール結果を文字数で制限する。
    長い場合は先頭部分 + 省略メッセージを返す。
    """
    text = str(result)
    if len(text) <= max_chars:
        return text

    # 先頭を保持して末尾を切り捨て
    truncated = text[:max_chars]
    omitted = len(text) - max_chars
    return f"{truncated}\n\n...[残り{omitted:,}文字を省略]"


def smart_search_tool(query: str, max_results: int = 3) -> str:
    """
    検索結果の件数とテキスト量を制限したツール実装例
    """
    # 例:Web検索の場合、上位3件のみ取得
    results = web_search(query, limit=max_results)

    summaries = []
    for i, result in enumerate(results, 1):
        # 各結果を200文字に制限
        snippet = result["snippet"][:200]
        summaries.append(f"{i}. {result['title']}\n{snippet}")

    return "\n\n".join(summaries)

VRAMモニタリングの実装

実行中のVRAM使用量をリアルタイムで監視する:

# vram_monitor.py
import subprocess
import threading
import time

def get_vram_usage_mb() -> dict[str, int]:
    """nvidia-smiでGPUメモリ使用量を取得"""
    try:
        result = subprocess.run(
            ["nvidia-smi", "--query-gpu=memory.used,memory.free,memory.total",
             "--format=csv,noheader,nounits"],
            capture_output=True, text=True, timeout=5
        )
        used, free, total = map(int, result.stdout.strip().split(", "))
        return {"used_mb": used, "free_mb": free, "total_mb": total}
    except Exception:
        return {"used_mb": 0, "free_mb": 0, "total_mb": 0}


def monitor_vram(interval: float = 2.0, warning_threshold: float = 0.85):
    """
    バックグラウンドでVRAMを監視し、使用率が閾値を超えたら警告
    """
    def _monitor():
        while True:
            stats = get_vram_usage_mb()
            if stats["total_mb"] > 0:
                usage_ratio = stats["used_mb"] / stats["total_mb"]
                if usage_ratio > warning_threshold:
                    print(
                        f"⚠️ VRAM警告: {stats['used_mb']}MB / {stats['total_mb']}MB "
                        f"({usage_ratio*100:.1f}%) — コンテキストを圧縮してください"
                    )
            time.sleep(interval)

    thread = threading.Thread(target=_monitor, daemon=True)
    thread.start()
    return thread


# 使用例
if __name__ == "__main__":
    monitor_thread = monitor_vram(interval=3.0, warning_threshold=0.85)
    
    # エージェント処理
    result = run_agent_with_guard(
        user_input="競合他社のWebサイトを調べて特徴をまとめて",
        tools=my_tools,
        tool_handler=my_tool_handler,
    )
    print(result)

実践的な設定値の目安

GPU VRAM推奨モデルサイズ推奨コンテキスト長推奨ツール呼び出し上限
8GB7B (Q4_K_M)4,096〜6,000 tokens5回
12GB13B (Q4_K_M)8,192 tokens8回
24GB34B (Q4_K_M)16,384 tokens15回
48GB+70B (Q4_K_M)32,768 tokens制限なし(監視のみ)

Apple Silicon(M1/M2/M3)はユニファイドメモリのため、上記より1段階大きいモデルが動く傾向があります(16GB RAM → 13B相当)。


あわせて読みたい

参考ソース