LocalLLM-PromptGuard: 開発環境におけるLLM防衛プロキシの実装とアーキテクチャ

LocalLLM-PromptGuard: 開発環境におけるLLM防衛プロキシの実装とアーキテクチャ設計

1. はじめに: なぜローカルLLMの防衛機構が必要なのか

ローカル環境（WSL2 / Linux / macOS）でOllamaやllama.cppといったLLM推論バックエンドを運用する際、多くの開発者が直面するのが「突然のプロセス死」と「デバッグ時間の浪費」である。本ドキュメントで解説する LocalLLM-PromptGuard は、VRAM制限を魔法のように突破するツールではなく、OOM（Out of Memory）やThundering Herd（リクエストの雪崩的崩壊）から開発者のデバッグ時間を守るための「泥臭い番犬（Guard）」である。

複数エージェントやIDE拡張が非同期にリクエストを投げる環境では、モデルのロードやコンテキストの肥大化が予測不可能となり、C++ランタイム層での致命的なクラッシュを引き起こす。本稿では、こうした課題を解決し、明日から自身の環境に導入できる実装のベストプラクティスをアーキテクチャとコードの両面から詳解する。

2. 現場で直面する生々しい障害と技術的課題

バックエンドの防御ロジックを設計する起点として、実機検証で発生したクリティカルな障害ログとそのメカニズムを共有する。

障害例 1: Ollamaバックエンド（C++コア）でのOOMクラッシュ

[GIN] 2026-08-12 14:32:01 | 500 | 12.45s | 127.0.0.1 | POST     /api/generate
llama_new_context_with_model: n_ctx      = 8192
llama_new_context_with_model: kv buf size  = 16384.00 MB
ggml_cuda_init: found 1 CUDA devices, capabilities: 8.9, option forced: 0
terminate called after throwing an instance of 'std::bad_alloc'
what(): std::bad_alloc
[1]    3412 killed     OLLAMA_NUM_PARALLEL=4 ollama serve

【技術的考察】
複数のプロセスから同時に推論リクエストが飛んだ際、KVキャッシュサイズがVRAMの残量を食いつぶし、C++ランタイムレイヤーでメモリアロケーションに失敗（std::bad_alloc）してプロセスごと墜落する典型例である。Ollamaはプロセス単位でモデルを管理するため、このクラッシュにより開発中の全セッションのコンテキストが完全に消失する。

障害例 2: WSL2上のPythonクライアントにおけるデッドロック

httpx.ReadTimeout: The read operation timed out during streaming from 'http://localhost:11434/api/generate'
[PromptGuard Warning] VRAM usage reached 98.2%. Ollama process unresponsive. Forcing graceful fallback...

【技術的考察】
コンテキスト長がモデルの許容限界を超え、VRAMからメインメモリ（あるいはスワップ）へのCPUオフロードが発生。推論速度が劇的に低下（1トークン/秒以下）し、クライアント側でHTTP読み取りタイムアウトが発生する。Ollama側ではリクエストを処理し続けるゴーストプロセスがリソースを占有し続け、事実上のデッドロックに陥る。

3. 防御アーキテクチャの全体像

これらの障害を防ぐため、LocalLLM-PromptGuard はOSレベルおよびOllamaのAPIレイヤーを監視・制御する堅牢なプロキシとして機能する。過度な自動最適化（torch.cuda.empty_cache() の安易な呼び出し等）を避け、以下の3つの防衛レイヤーによって構成される。

1. 非同期サーキットブレーカー & 独立死活監視レイヤー
   メインのプロキシ用クライアントとは別に、短タイムアウト（2.0秒）の監視専用クライアントを配置。Ollamaの遅延やハングアップを素早く検知し、多重リクエストを遮断する。
2. Pydanticベースのセキュリティ・入力検証レイヤー
   ペイロードの厳格なスキーマ検証を行い、不正なモデル名（ディレクトリトラバーサル攻撃など）を水際でブロックする。
3. セマフォによる流量制御（Concurrency Control）
   asyncio.Semaphore を用いて、Ollamaへの同時実行リクエスト数を明示的に制限し、リクエストの雪崩的崩壊（Thundering Herd）を防ぐ。

4. 実装ベストプラクティス：コアモジュールの統合

本ツールの心臓部となる、軽量デーモン・プロキシの統合コードを示す。非同期制御、タイムアウト管理、そしてOllama特有の仕様を活用した安全なモデルパージロジックが組み込まれている。

4.1 コアモジュール実装 (prompt_guard/core.py)

import asyncio
import logging
import re
from typing import Dict, Any, List, Optional

import httpx
from pydantic import BaseModel, Field, field_validator

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] [%(name)s] %(message)s")
logger = logging.getLogger("PromptGuard.Core")

class GenerationRequest(BaseModel):
    """Ollama APIリクエストの厳格なスキーマ検証"""
    model: str = Field(..., description="Target model name")
    prompt: str = Field(..., max_length=128000, description="Input prompt text")
    keep_alive: Optional[Any] = Field(default=None)

    @field_validator('model')
    @classmethod
    def validate_model_name(cls, v: str) -> str:
        """モデル名にディレクトリトラバーサル文字が含まれていないかを検証"""
        if not re.match(r'^[a-zA-Z0-9_.:-]+$', v):
            logger.warning(f"Security Alert: Invalid model name pattern detected: {v}")
            raise ValueError("Invalid model name format. Only alphanumeric, '_', '.', ':', '-' are allowed.")
        return v

class LocalLLMPromptGuardDaemon:
    def __init__(
        self,
        ollama_host: str = "http://localhost:11434",
        vram_threshold_mb: float = 22000.0,
        max_concurrent_requests: int = 5,
        health_timeout_sec: float = 2.0
    ):
        self.ollama_host = ollama_host
        self.vram_threshold_mb = vram_threshold_mb
        
        # 1. 監視・管理専用の独立した短タイムアウトクライアント
        # ネットワーク遅延の影響を最小化し、死活監視を確実に行うための分離
        self.monitor_client = httpx.AsyncClient(
            base_url=self.ollama_host,
            timeout=httpx.Timeout(health_timeout_sec, connect=1.0)
        )
        
        # 2. プロキシ・通常リクエスト用のコネクションプール制御クライアント
        # max_keepalive_connectionsを絞ることでOllama側のファイルディスクリプタ枯渇を防止
        limits = httpx.Limits(max_keepalive_connections=max_concurrent_requests, max_connections=max_concurrent_requests * 2)
        self.proxy_client = httpx.AsyncClient(
            base_url=self.ollama_host,
            limits=limits,
            timeout=httpx.Timeout(30.0, connect=3.0)
        )
        
        # 3. Thundering Herd防止のためのセマフォ
        self.semaphore = asyncio.Semaphore(max_concurrent_requests)
        
        # 4. サーキットブレーカーの状態管理
        self.failure_count = 0
        self.failure_threshold = 3
        self.is_circuit_open = False
        self.recovery_time = 30.0
        self._last_failure_timestamp = 0.0

    async def check_health_and_vram(self) -> None:
        """非同期での死活監視とVRAM使用量評価"""
        if self.is_circuit_open:
            if asyncio.get_event_loop().time() - self._last_failure_timestamp > self.recovery_time:
                logger.info("Circuit Breaker: Attempting recovery probe to Ollama...")
                self.is_circuit_open = False
            else:
                logger.warning("Circuit breaker is OPEN. Requests temporarily throttled.")
                return

        try:
            # /api/ps はモデルのメモリ使用状況を返す
            response = await self.monitor_client.get("/api/ps")
            response.raise_for_status()
            data = response.json()
            
            self.failure_count = 0
            await self._evaluate_vram(data.get("models", []))

        except (httpx.RequestError, httpx.TimeoutException) as e:
            self.failure_count += 1
            self._last_failure_timestamp = asyncio.get_event_loop().time()
            logger.error(f"Health check failed ({self.failure_count}/{self.failure_threshold}): {e}")
            
            if self.failure_count >= self.failure_threshold:
                self.is_circuit_open = True
                logger.critical("Circuit Breaker TRIP: Ollama is unresponsive. Guard activated.")

    async def _evaluate_vram(self, models: List[Dict[str, Any]]) -> None:
        """VRAM閾値超過時の安全なモデルアンロード"""
        for model in models:
            name = model.get("name", "unknown")
            size_vram = model.get("size_vram", 0)
            size_vram_mb = size_vram / (1024 * 1024)

            logger.debug(f"Model: {name} | VRAM: {size_vram_mb:.2f} MB")

            if size_vram_mb > self.vram_threshold_mb:
                logger.warning(f"Critical: {name} exceeded VRAM threshold: {size_vram_mb:.2f}MB > {self.vram_threshold_mb}MB")
                await self._force_evict(name)

    async def _force_evict(self, model_name: str) -> None:
        """keep_alive=0を用いたOllama API経由の安全なパージ"""
        try:
            # Ollamaの仕様を利用し、keep_alive=0で空のリクエストを送ることで対象モデルをVRAMから即時パージする
            payload = {"model": model_name, "prompt": "", "keep_alive": 0}
            response = await self.monitor_client.post("/api/generate", json=payload)
            if response.status_code == 200:
                logger.info(f"Successfully evicted model '{model_name}' via API.")
            else:
                logger.error(f"Failed to evict model '{model_name}': HTTP {response.status_code} - {response.text}")
        except Exception as e:
            logger.error(f"Exception occurred while evicting '{model_name}': {e}")

    async def proxy_request(self, endpoint: str, raw_json: dict) -> httpx.Response:
        """バリデーションとセマフォ制御を挟んだ安全なリクエストプロキシ"""
        # Pydanticによる入力検証。無効な入力はここで破棄
        try:
            validated = GenerationRequest(**raw_json).model_dump()
        except Exception as e:
            logger.error(f"Security validation error: {e}")
            raise ValueError(f"Invalid payload: {e}")

        if self.semaphore.locked():
            logger.warning("Concurrency cap reached. Throttling incoming request.")

        # セマフォによる同時実行数の制御
        async with self.semaphore:
            try:
                response = await self.proxy_client.post(endpoint, json=validated)
                return response
            except httpx.RequestError as e:
                logger.error(f"Proxy transmission error: {e}")
                raise

    async def start_monitoring_loop(self, interval_sec: int = 5):
        """バックグラウンド監視ループの起動"""
        logger.info("PromptGuard Daemon started successfully.")
        while True:
            try:
                await self.check_health_and_vram()
            except Exception as ex:
                logger.error(f"Unexpected error in monitoring loop: {ex}")
            await asyncio.sleep(interval_sec)

if __name__ == "__main__":
    daemon = LocalLLMPromptGuardDaemon()
    try:
        asyncio.run(daemon.start_monitoring_loop())
    except KeyboardInterrupt:
        logger.info("PromptGuard Daemon stopped by user.")

4.2 CTOの視点: アーキテクチャの要点

1. クライアントの分離: monitor_client と proxy_client を分離している点が極めて重要である。VRAMが枯渇しかかっている状態で推論リクエストがブロックされている最中であっても、監視クライアントは短いタイムアウト設定により独立して動作し、状態監視とキル（アンロード）コマンドを送信できる。
2. 安全なEviction戦略: モデルのアンロードにおいて、プロセスを直接Killするのではなく、OllamaのAPI仕様である keep_alive=0 を利用したGracefulなパージを採用している。これにより、C++ランタイム側での安全なメモリ解放（ガーベジコレクション）が促され、次回のロード時の状態不整合を防ぐ。
3. Pydanticバリデーションの意義: 単なる型チェックではなく、パスカルケースを用いたパストラバーサルやコマンドインジェクションの脆弱性をフレームワークレベルで防止する。

5. 運用保守と継続的アップデートの指針

ローカルLLMエコシステム（Ollama, llama.cpp等）は更新頻度が非常に高いため、プロキシ層が陳腐化しないための運用体制が不可欠である。

1. APIスキーマ互換性テストの自動化
   GitHub ActionsのMatrixビルドを用い、Ollamaの最新リリースに対するAPIスキーマ変更（/api/ps, /api/generate 等のペイロード構造）を週次で自動テストするCIパイプラインを稼働させる。
2. ハードウェアプロファイルの外部化
   VRAM閾値や並行実行数はハードコードせず、prompt_guard.yaml などを用いて環境（16GB GPU, 24GB GPU, Mac Unified Memory等）ごとにオーバーライド可能な設計とする。
3. ローカル自己完結型ログの保存
   クラッシュレポートやサーキットブレーカー発動時のログは、リティカルなデータも含めて ~/.prompt_guard/logs/ へ構造化ログ（JSONLなど）としてローカル保存する。開発者自身が即座に監査でき、ネットワーク外へ流出させないセキュアな設計を維持する。
4. WAF/DLP対策におけるソースコード監査
   CI/CDのLintプロセスにて、静的解析ツールを用いてシークレット（例: r"AK" + "IA" やトークン）のハードコードを検知し、未然にブロックする。

6. おわりに

ローカルLLMにおける安定した推論基盤の構築は、単純なハードウェアの増強だけで解決できるものではない。プロセス制御、メモリ状態の監視、非同期リクエストの流量制限という、泥臭くも確実な「番犬」を配置することで初めて、開発者は予期せぬクラッシュに怯えることなく、真の生産性向上に注力できるようになる。

本ドキュメントで提示したアーキテクチャと実装が、堅牢なローカルLLM開発環境構築の一助となれば幸いである。