runtime-skills

ユニバーサルランタイムのスキル

Universal Runtime（LlamaFarm のローカル ML 推論サーバー）に関するベストプラクティスとコードレビューチェックリストです。

概要

Universal Runtime は、HuggingFace モデル向けに OpenAI 互換のエンドポイントを提供します。

• テキスト生成（因果言語モデル：GPT、Llama、Mistral、Qwen）
• テキスト埋め込み（BERT、sentence-transformers、ModernBERT）
• 分類、NER、再ランキング
• OCR と文書理解
• 異常検知

ディレクトリ：runtimes/universal/ Python：3.11+ 主な依存関係：PyTorch、Transformers、FastAPI、llama-cpp-python

共通スキルへのリンク

このスキルは、共通の Python プラクティスを拡張するものです。必ず最初にこれらを適用してください。

トピック	ファイル	優先度
パターン	python-skills/patterns.md	中
非同期処理	python-skills/async.md	高
型付け	python-skills/typing.md	中
テスト	python-skills/testing.md	中
エラー	python-skills/error-handling.md	高
セキュリティ	python-skills/security.md	重大

ランタイム固有のチェックリスト

トピック	ファイル	要点
PyTorch	pytorch.md	デバイス管理、dtype、メモリ解放
Transformers	transformers.md	モデルの読み込み、トークン化、推論
FastAPI	fastapi.md	API 設計、ストリーミング、ライフスパン
パフォーマンス	performance.md	バッチ処理、キャッシュ、最適化

アーキテクチャ

runtimes/universal/
├── server.py              # FastAPI app, model caching, endpoints
├── core/
│   └── logging.py         # UniversalRuntimeLogger (structlog)
├── models/
│   ├── base.py            # BaseModel ABC with device management
│   ├── language_model.py  # Transformers text generation
│   ├── gguf_language_model.py  # llama-cpp-python for GGUF
│   ├── encoder_model.py   # Embeddings, classification, NER, reranking
│   └── ...                # OCR, anomaly, document models
├── routers/
│   └── chat_completions/  # Chat completions with streaming
├── utils/
│   ├── device.py          # Device detection (CUDA/MPS/CPU)
│   ├── model_cache.py     # TTL-based model caching
│   ├── model_format.py    # GGUF vs transformers detection
│   └── context_calculator.py  # GGUF context size computation
└── tests/

主なパターン

1. ダブルチェックロックによるモデルの読み込み

_model_load_lock = asyncio.Lock()

async def load_encoder(model_id: str, task: str = "embedding"):
    cache_key = f"encoder:{task}:{model_id}"
    if cache_key not in _models:
        async with _model_load_lock:
            # Double-check after acquiring lock
            if cache_key not in _models:
                model = EncoderModel(model_id, device, task=task)
                await model.load()
                _models[cache_key] = model
    return _models.get(cache_key)

2. デバイスを考慮したテンソル操作

class BaseModel(ABC):
    def get_dtype(self, force_float32: bool = False):
        if force_float32:
            return torch.float32
        if self.device in ("cuda", "mps"):
            return torch.float16
        return torch.float32

    def to_device(self, tensor: torch.Tensor, dtype=None):
        # Don't change dtype for integer tensors
        if tensor.dtype in (torch.int32, torch.int64, torch.long):
            return tensor.to(device=self.device)
        dtype = dtype or self.get_dtype()
        return tensor.to(device=self.device, dtype=dtype)

3. TTL ベースのモデルキャッシュ

_models: ModelCache[BaseModel] = ModelCache(ttl=300)  # 5 min TTL

async def _cleanup_idle_models():
    while True:
        await asyncio.sleep(CLEANUP_CHECK_INTERVAL)
        for cache_key, model in _models.pop_expired():
            await model.unload()

4. スレッドプールを使った非同期生成

# GGUF models use blocking llama-cpp, run in executor
self._executor = ThreadPoolExecutor(max_workers=1)

async def generate(self, messages, max_tokens=512, ...):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(self._executor, self._generate_sync)

レビューの優先度

Universal Runtime のコードをレビューする際は、次の点を確認してください。

1. 重大 - セキュリティ

   • ファイルエンドポイントにおけるパストラバーサルの防止
   • モデル ID の入力サニタイズ

2. 高 - メモリとデバイス

   • アンロード時の CUDA/MPS キャッシュの適切なクリア
   • 推論時の torch.no_grad()
   • デバイスに適した dtype

3. 中 - パフォーマンス

   • モデルキャッシュのパターン
   • 該当する場合のバッチ処理
   • ストリーミングの実装

4. 低 - コードスタイル

   • patterns.md との一貫性
   • 適切な型ヒント