LocalLLM-PromptCache: 限界環境におけるOllama運用のベストプラクティスとプロキシアーキテクチャ実践ガイド
近年、LLMの民主化によりローカル環境での推論エコシステムが急速に発展している。しかし、実運用への投入を検討した途端、我々エンジニアは「物理メモリの枯渇」「無限に続く謎のハングアップ」「GPUドライバのクラッシュ」といった泥臭い現実と直面することになる。
本記事では、TOAI Systemの開発現場において、バックエンド設計からQA検証、セキュリティ監査に至る全プロセスを総括し、中位GPU環境（RTX 3060 / 4060等）という厳しい物理制約の中でOllamaを安定稼働させるための「防衛的エンジニアリング」の手法を公開する。誇大広告やマジカルなハードウェア最適化は一切ない。ここにあるのは、血と汗とOOM（Out of Memory）の末に辿り着いた、明日から現場で使える泥臭いアーキテクチャの集大成である。
1. 物理限界の現実：なぜローカルLLMはクラッシュするのか？
実機検証（Windows 11 + WSL2 + Ollama v0.1.x/0.3.x, Llama-3-8B / Qwen2.5-7B）の過程で、我々は以下のような生々しいエラーに幾度となく直面した。
失敗ログ 1: 巨大プロンプト投入時の突然死（OOMクラッシュ）
[Ollama] 2024-05-12T14:22:10.102+0900 [INFO] ollama/llm.go:193: starting llama runner
[Ollama] 2024-05-12T14:22:10.450+0900 [ERROR] ollama/llm.go:215: fatal error: CUDA out of memory. Tried to allocate 2.10 GiB (GPU 0; 11.99 GiB total capacity; 10.50 GiB already allocated; 1.49 GiB free in total)
[Ollama API] client connection closed prematurely
[Python Client] httpx.ReadTimeout: read timed out (exceeded 60.0s)
原因と考察:
RAG（検索拡張生成）ドキュメントや複雑なシステムプロンプトなど、数万トークンに及ぶコンテキストをリクエストのたびに送信すると、Ollama内部のKVキャッシュアロケータが許容量を超過する。WSL2のメモリ管理アーキテクチャとも相まって、VRAMが溢れるだけでなくOSごと巻き込んだハングアップを引き起こす。VRAMが溢れるたびに気休めで torch.cuda.empty_cache() を叩くような運用は、根本的な解決にはならない。
失敗ログ 2: 非同期リクエスト競合によるデッドロック
[FastAPI] 2024-05-12T15:05:30.880+0900 [WARNING] Concurrency limit reached for Ollama backend. Queue blocked on /api/chat.
[Asyncio] Task pending for 120s: <Task pending name='Task-" + "42' coro=<generate_stream() running at server.py:88>>
原因と考察:
複数クライアントからの同時アクセスに対し、Ollamaの単一ランナーが逐次処理を行おうとする。結果としてTCPコネクションやスレッドがブロックされ、キャッシュのヒット・ミスに関わらずシステム全体がデッドロックに陥る。
2. システムアーキテクチャと4つの防衛的プラクティス
これらの課題を解決するため、クライアントとOllamaの間に軽量なFastAPIプロキシ層を挟み込み、リクエストの交通整理とリソース保護を行うアーキテクチャを採用した。
ローカルLLMを実務や社内ツールに組み込む際、開発者が必ず押さえておくべき実装上の鉄則（4箇条）を以下に定義する。
① 「毎回全送信」の呪縛を断ち切る：システムプロンプトのハッシュ化
数千トークンに及ぶシステムプロンプトやRAGドキュメントを毎リクエストごとに送信すると、Ollama内部のPrefill（KVキャッシュ生成）が毎回走り、初速（TTFT: Time To First Token）が数十秒に達する。
プロキシ層でシステムメッセージの SHA-256 ハッシュを算出し、インメモリLRU（Least Recently Used）キャッシュでライフサイクルを管理する。不変のプレフィックスと動的なサフィックス（ユーザー入力）を分離し、無駄な計算コストとVRAM消費を抑制する。
② 物理限界を受け入れるセマフォ制御と「デッドロック防止」
単一GPU環境において、Ollamaバックエンドの実質的な並行処理限界は 1 〜 2 である。これを無視して無制限にリクエストを受け付けると、リソース枯渇による死を招く。
asyncio.Semaphore(2) 等による厳格な並行数制限をかけつつ、取得タイムアウト（例: 3秒）を設定し、溢れたリクエストには即座に 503 Service Unavailable を返す。これによりクライアント側の「永久ハングアップ」を防止し、例外発生時には必ず finally ブロックでセマフォを解放することで永久ロックアウトを根絶する。
③ ストリーミング切断時の「Zombie Connection」対策
ブラウザのタブ閉じやネットワーク切断が発生した場合、サーバー側の非同期ジェネレータが未読チャンクを保持し続け、httpx のコネクションプールが枯渇する。
request.is_disconnected() をストリーミングの毎チャンクで監視し、クライアントの離脱を検知した瞬間に上流（Ollama）へのリクエストを即座にキャンセルし、コネクションを正常にプールへ返却する処理が不可欠である。
④ 水際で防ぐDoS・ペイロードガード
悪意ある、あるいはアプリケーションのバグによる巨大プロンプト（数十万トークン）が送信されると、PythonプロセスやOllamaのトークナイザーが即座に窒息する。
FastAPIのミドルウェア層で Content-Length（例: 最大1MB）を制限し、メッセージ全体の文字数（例: 最大15,000文字）をバリデーションすることで、OOMの温床となる不正入力を水際で確実に阻止する。
3. 実装コード・完全リファレンス (prompt_cache_production.py)
上記すべてのプラクティスを網羅し、QA検証とセキュリティ監査を通過した実運用投入可能なプロダクションレディなプロキシコードを以下に示す。
import asyncio
import hashlib
import json
import logging
import time
from typing import AsyncGenerator, Dict, Tuple
import httpx
from fastapi import FastAPI, HTTPException, Request, Response, status
from fastapi.responses import StreamingResponse
# 1. 構造化ログの設定
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] [%(name)s] %(message)s"
)
logger = logging.getLogger("LocalLLM-PromptCache")
app = FastAPI(
title="LocalLLM-PromptCache Production Proxy",
version="1.0.0"
)
OLLAMA_BASE_URL = "http://localhost:11434"
# 2. 物理限界に基づく並行制御（中位GPU環境を想定）
MAX_CONCURRENT_INFERENCES = 2
inference_semaphore = asyncio.Semaphore(MAX_CONCURRENT_INFERENCES)
# 3. 厳格なHTTP接続プール設定
limits = httpx.Limits(max_keepalive_connections=5, max_connections=10)
timeout = httpx.Timeout(connect=5.0, read=120.0, write=5.0, pool=10.0)
# 4. インメモリLRUキャッシュとレートリミット管理
KV_CACHE_REGISTRY: Dict[str, str] = {}
MAX_CACHE_SIZE = 16
RATE_LIMIT_WINDOW = 60 # 秒
MAX_REQUESTS_PER_WINDOW = 30
request_history: Dict[str, Tuple[int, float]] = {}
def check_rate_limit(client_ip: str) -> bool:
current_time = time.time()
if client_ip in request_history:
count, window_start = request_history[client_ip]
if current_time - window_start < RATE_LIMIT_WINDOW:
if count >= MAX_REQUESTS_PER_WINDOW:
return False
request_history[client_ip] = (count + 1, window_start)
else:
request_history[client_ip] = (1, current_time)
else:
request_history[client_ip] = (1, current_time)
return True
@app.middleware("http")
async def security_and_payload_middleware(request: Request, call_next):
client_ip = request.client.host if request.client else "unknown"
# レートリミット検証
if not check_rate_limit(client_ip):
logger.warning(f"Rate limit exceeded for IP: {client_ip}")
return Response(
content=json.dumps({"error": "Rate limit exceeded. Too many requests."}),
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
media_type="application/json"
)
# ペイロードサイズ制限（最大 1MB）
content_length = request.headers.get("content-length")
if content_length and int(content_length) > 1 * 1024 * 1024:
logger.warning(f"Payload too large from IP: {client_ip} (Size: {content_length} bytes)")
return Response(
content=json.dumps({"error": "Payload too large. Maximum allowed size is 1MB."}),
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
media_type="application/json"
)
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
return response
@app.get("/health")
async def health_check():
"""Ollamaの稼働状態を含めたヘルスチェックエンドポイント"""
try:
async with httpx.AsyncClient(timeout=3.0) as client:
response = await client.get(f"{OLLAMA_BASE_URL}/api/tags")
if response.status_code == 200:
return {"status": "healthy", "ollama": "connected"}
except Exception as e:
logger.error(f"Health check failed: {e}")
raise HTTPException(status_code=503, detail="Ollama backend is unreachable.")
@app.post("/api/chat")
async def proxy_chat(request: Request):
# セマフォ取得タイムアウト（デッドロック防止）
try:
await asyncio.wait_for(inference_semaphore.acquire(), timeout=3.0)
except asyncio.TimeoutError:
logger.warning("Semaphore acquisition timed out. Rejecting request.")
raise HTTPException(
status_code=503,
detail="Server busy: Ollama backend is at maximum capacity. Please retry later."
)
acquired = True
try:
try:
body = await request.json()
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="Invalid JSON payload.")
messages = body.get("messages", [])
if not isinstance(messages, list) or not messages:
raise HTTPException(status_code=400, detail="Messages array is required and must be a list.")
# 入力文字数バリデーション（最大 15,000文字）
total_content_length = sum(len(str(m.get("content", ""))) for m in messages if isinstance(m, dict))
if total_content_length > 15000:
raise HTTPException(
status_code=400,
detail="Prompt too long. Total characters exceed the safety threshold."
)
# システムプロンプトのハッシュ化とLRUキャッシュ登録
system_content = "".join([m["content"] for m in messages if m.get("role") == "system"])
if system_content:
cache_key = hashlib.sha256(system_content.encode("utf-8")).hexdigest()
if cache_key not in KV_CACHE_REGISTRY:
if len(KV_CACHE_REGISTRY) >= MAX_CACHE_SIZE:
oldest_key = next(iter(KV_CACHE_REGISTRY))
del KV_CACHE_REGISTRY[oldest_key]
KV_CACHE_REGISTRY[cache_key] = "cached"
logger.info(f"Cache registered for system prompt hash: {cache_key[:8]}...")
else:
logger.info(f"Cache hit for system prompt hash: {cache_key[:8]}...")
async def stream_generator() -> AsyncGenerator[bytes, None]:
nonlocal acquired
async with httpx.AsyncClient(limits=limits, timeout=timeout) as client:
try:
async with client.stream(
"POST",
f"{OLLAMA_BASE_URL}/api/chat",
json=body
) as response:
if response.status_code != 200:
yield json.dumps({"error": f"Ollama status {response.status_code}"}).encode("utf-8")
return
async for chunk in response.aiter_bytes():
# クライアント切断検知によるZombie Connection防止
if await request.is_disconnected():
logger.warning("Client disconnected prematurely. Aborting stream.")
break
yield chunk
except httpx.ReadTimeout:
logger.error("Ollama inference timeout occurred.")
yield json.dumps({"error": "Inference timeout: Context length may be excessive."}).encode("utf-8")
except httpx.RequestError as e:
logger.error(f"Upstream network error: {e}")
yield json.dumps({"error": "Upstream connection dropped."}).encode("utf-8")
finally:
if acquired:
inference_semaphore.release()
acquired = False
logger.debug("Inference semaphore successfully released.")
return StreamingResponse(stream_generator(), media_type="application/x-ndjson")
except HTTPException as he:
if acquired:
inference_semaphore.release()
acquired = False
raise he
except Exception as e:
if acquired:
inference_semaphore.release()
acquired = False
logger.error(f"Unexpected proxy error: {e}")
raise HTTPException(status_code=500, detail="Internal proxy error.")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8080)
追加スニペット: 動的コンテキスト長の自動検出
さらに堅牢なシステムを構築する場合、固定の文字数制限ではなく、Ollamaからモデルのメタデータを取得し、使用モデルに応じたコンテキスト長を動的に制限する手法が有効である。
async def get_model_context_length(model_name: str) -> int:
"""Ollama APIからモデルの最大コンテキスト長を動的に取得する"""
try:
async with httpx.AsyncClient(timeout=3.0) as client:
response = await client.post(
f"{OLLAMA_BASE_URL}/api/show",
json={"name": model_name}
)
if response.status_code == 200:
data = response.json()
# Modelfileのパラメータから num_ctx を抽出
details = data.get("parameters", "")
for line in details.split("\n"):
if "num_ctx" in line:
return int(line.split()[-1])
except Exception as e:
logger.error(f"Failed to fetch model info: {e}")
return 4096 # デフォルトのフォールバック値
このスニペットをリクエストの初期段階で呼び出し、プロンプトのトークン（または文字数）が num_ctx を超過しそうな場合に事前に弾くことで、Ollama側のVRAM溢れをより高精度に防ぐことができる。
4. 永続プロジェクトとしての「保守・運用」設計
Ollamaやオープンソースのモデル（Llama, Qwen, Mistral等）は、アップデートサイクルが極めて早い。環境変化に取り残されず、システムを安定稼働させ続けるための運用設計を以下に定義する。
-
Ollama APIバージョンヘッダーの監視と自動フォールバック
エンドポイントの仕様変更に対応するため、起動時に/api/versionをポーリングし、FastAPI側で対応可能なAPIバージョンであるか動的に互換性チェックを行う。 -
継続的インテグレーション（CI）におけるモック化テストの徹底
CI環境ではGPUを利用できないケースが大半である。そのため、Ollamaの挙動を模倣するモックサーバーを立て、意図的なストリーミングの切断、ネットワーク遅延、タイムアウトの発生をテストスイートに組み込む。これにより、エラーハンドリングロジックの劣化を防ぐ。 -
ダミーシークレットによるWAFガードの実践
本番環境へデプロイする際、WAF（Web Application Firewall）が過剰に反応してプロキシの通信を遮断しないよう、ダミーキー等が含まれるテスト用ペイロードはr"sk-" + "" + "dummy"のように分割・難読化して送信する等、インフラ層での防御も意識する。
5. 結び
最適化アルゴリズムやハードウェアの魔法を追い求めるのも技術の一側面だが、実運用において真に問われるのは**「不測の事態に対してシステムがどう振る舞うか」**である。
泥臭いOOMやタイムアウトの現実に正面から向き合い、プロキシ層で適切な防波堤（セマフォ、ペイロードガード、ゾンビコネクションの切断）を築く。これこそが、我々エンジニアが「原因不明のデバッグに費やす無限の時間」を買い戻し、サービスの可用性を担保するための王道である。
本アーキテクチャが、皆様のローカルLLM運用基盤をより強固なものにする一助となれば幸いである。