PythonとS3で構築するLLMモデルレジストリ：メタデータ駆動型の実装パターン

LLM（大規模言語モデル）のファインチューニング（微調整）を繰り返していると、必ず直面する壁があります。

「あれ、先週最高精度を出したモデルの重みファイル、どれだっけ？」
「v2_final_real_fixed.bin って名前だけど、学習データセットはバージョンいくつのやつを使ったんだ？」

もしあなたが今、クラウドストレージの中身を見てため息をついているなら、この記事が解決の糸口になるはずです。

MLflowやWeights & Biasesといった素晴らしいMLOps（機械学習の運用管理）プラットフォームが存在することは承知しています。しかし、初期フェーズのプロジェクトや、セキュリティ制約の厳しいオンプレミス環境、あるいは単に「仕組みを深く理解したい」というエンジニアにとって、いきなり多機能なツールを導入するのはオーバーエンジニアリング（過剰な設計）になりがちです。

今回は、外部ツールに依存せず、PythonとS3（または互換ストレージ）だけで構築する、ミニマムかつ堅牢なモデルレジストリの実装パターンを紹介します。これは単なるファイルアップローダーではありません。メタデータ（データに関するデータ）を「正」とし、再現性をコードで担保するための設計思想の共有です。

手を動かしながら、LLM運用の核心部分である「レジストリ管理」の構造を論理的に作り上げていきましょう。

1. なぜ「ファイル名管理」では破綻するのか

実際の開発現場で最初に行われがちなのが、ファイル名に情報を詰め込むアプローチです。

custom-llm-8b-finetune-lr2e5-epoch3-202X1027.pt

一見わかりやすく良さそうに見えますが、この方法はすぐに限界を迎えます。現在のLLMの性能や挙動を決定づける要素は、もはや十数文字のファイル名に収まりきらないほど多岐にわたるからです。

モデルの実体とメタデータの乖離問題

モデルの完全な再現性を担保するには、単なる重みファイルだけでなく、少なくとも以下の情報が「セット」として厳密に管理されている必要があります。

• 学習コードのバージョン: 実行時点のGitコミットハッシュやスクリプトの状態。
• 学習データセット: データのハッシュ値、保存先のパス、データクレンジングや前処理のバージョン。
• ハイパーパラメータと追加モジュール: 学習率やバッチサイズに加え、LoRA（パラメータを効率的に微調整する手法）のランクや適用対象。現在では旧形式（.ckpt）を避けて.safetensors形式を優先する傾向が強く、ベースモデルとの互換性といったメタデータも必須です。
• 量子化とロード機構の設定: 8bitや4bitといった低精度フォーマットでの推論が一般的になる中、最新の環境ではこれらが標準機能としてサポートされるようになっています。そのため、どの量子化アルゴリズムを適用したかの設定値も不可欠です。
• 評価メトリクス: 単純なLoss（損失）だけでなく、RougeやBLEU、さらにはMT-Benchなどの総合的なベンチマークスコア。
• 環境情報と依存関係: 使用したライブラリのバージョンや、CUDAなどのドライバ環境。推論エンジンとの連携情報も重要です。また、実行環境の再現難易度は年々上がっています。

これだけの複雑な依存関係を、ファイル名だけで表現するのは不可能です。結果として、実験ノートの記録と実際のファイルとの間に乖離が生まれ、「あの時の高精度なモデルが二度と再現できない」という深刻な技術的負債が積み上がります。ライブラリの頻繁なアップデートや、GPUアーキテクチャの違いによる微妙な挙動の変化は、ファイル名を眺めていても決して読み取ることはできません。

本記事で実装する「メタデータ駆動型レジストリ」のゴール

本記事で目指すのは、「メタデータ（JSONなど）を保存したら、自動的にモデル実体も紐付いて管理される」という堅牢なシステムです。

1. 検索性: 「Lossが0.5以下で、かつ特定のデータセットバージョンを使い、最新の推論環境に適合するモデル」といった複雑な条件を即座に特定できる。
2. 再現性: どのコード、どのデータ、どのような環境で学習・量子化を行ったかが機械的に記録される。
3. 不可分性: モデルの重みファイルとメタデータが常に1対1で対応し、リンク切れや設定の散逸を防ぐ。

これを実現するために、まずはメタデータの構造を明確に定義するアプローチから解説します。

2. メタデータスキーマの設計と実装

堅牢なシステムを作る第一歩は、データの「型」を厳格にすることです。Pythonの辞書型（dict）で適当に管理するのではなく、Pydantic を使用してスキーマ（構造）を定義します。これにより、必要な情報の欠落を防ぎ、チェック作業を自動化できます。

以下は、LLMファインチューニング管理を想定したメタデータモデルの実装例です。

import hashlib
import json
from datetime import datetime, timezone
from typing import Dict, Optional, Any
from pydantic import BaseModel, Field, field_validator

class ModelMetadata(BaseModel):
    """
    モデルのメタデータを定義するスキーマ
    """
    model_id: str = Field(..., description="モデルの一意なID (UUID等)")
    base_model: str = Field(..., description="ベースモデル名 (例: meta-llama/Llama-2-7b-hf)")
    created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    author: str = Field(..., description="学習実行者")
    
    # 再現性のための重要情報
    git_commit_hash: str = Field(..., description="学習コードのコミットハッシュ")
    dataset_hash: str = Field(..., description="学習データのハッシュ値")
    parameters: Dict[str, Any] = Field(default_factory=dict, description="ハイパーパラメータ")
    
    # 評価指標
    metrics: Dict[str, float] = Field(default_factory=dict, description="評価メトリクス")
    
    # ステージ管理 (mlflowのstage概念に近いもの)
    tags: Dict[str, str] = Field(default_factory=dict, description="タグ (例: stage=production)")

    @field_validator('git_commit_hash')
    @classmethod
    def validate_hash(cls, v: str) -> str:
        if len(v) < 7:
            raise ValueError("Gitハッシュは7文字以上である必要があります")
        return v

    def to_json(self) -> str:
        return self.model_dump_json(indent=2)

    @property
    def artifact_path(self) -> str:
        """S3等での保存パスを生成"""
        return f"models/{self.base_model}/{self.model_id}/"

設計のポイント

このクラス設計には、実務運用を想定した重要なポイントがいくつか含まれています。

• dataset_hash の必須化: データセットのパスだけでなく、内容のハッシュ値（MD5やSHA256など）を持たせることで、データの中身が変わったことに確実に気づけるようにします。
• artifact_path プロパティ: 保存場所のパス生成ロジックをメタデータ自身に持たせることで、コードの各所でパスを手書きする人為的ミスを防ぎます。

3. ストレージ抽象化レイヤーの実装

次に、モデルの実体（重みファイルなど）を保存する仕組みを作ります。ここで重要なのは、「ローカル環境でもクラウド環境でも同じコードが動くこと」です。

開発中はローカルのディスクを使い、本番学習時はS3を使う。このように環境に応じて保存先を切り替えられるよう、ストレージの操作を抽象化（共通化）します。

import os
import boto3
from abc import ABC, abstractmethod
from pathlib import Path
from botocore.exceptions import ClientError

class StorageBackend(ABC):
    """ストレージ操作の抽象基底クラス"""
    
    @abstractmethod
    def save_file(self, local_path: str, remote_key: str) -> None:
        pass

    @abstractmethod
    def load_file(self, remote_key: str, local_path: str) -> None:
        pass

    @abstractmethod
    def exists(self, remote_key: str) -> bool:
        pass

class LocalStorage(StorageBackend):
    """ローカルファイルシステム用バックエンド"""
    def __init__(self, base_dir: str = "./model_registry"):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)

    def save_file(self, local_path: str, remote_key: str) -> None:
        target_path = self.base_dir / remote_key
        target_path.parent.mkdir(parents=True, exist_ok=True)
        # 実運用ではshutil.copyなどを使用
        with open(local_path, 'rb') as f_src, open(target_path, 'wb') as f_dst:
            f_dst.write(f_src.read())
        print(f"[Local] Saved to {target_path}")

    def load_file(self, remote_key: str, local_path: str) -> None:
        source_path = self.base_dir / remote_key
        if not source_path.exists():
            raise FileNotFoundError(f"{remote_key} not found in local registry")
        with open(source_path, 'rb') as f_src, open(local_path, 'wb') as f_dst:
            f_dst.write(f_src.read())

    def exists(self, remote_key: str) -> bool:
        return (self.base_dir / remote_key).exists()

class S3Storage(StorageBackend):
    """AWS S3用バックエンド"""
    def __init__(self, bucket_name: str, region: str = "ap-northeast-1"):
        self.bucket_name = bucket_name
        self.s3 = boto3.client('s3', region_name=region)

    def save_file(self, local_path: str, remote_key: str) -> None:
        try:
            self.s3.upload_file(local_path, self.bucket_name, remote_key)
            print(f"[S3] Uploaded {local_path} to s3://{self.bucket_name}/{remote_key}")
        except ClientError as e:
            print(f"Failed to upload to S3: {e}")
            raise

    def load_file(self, remote_key: str, local_path: str) -> None:
        try:
            self.s3.download_file(self.bucket_name, remote_key, local_path)
        except ClientError as e:
            print(f"Failed to download from S3: {e}")
            raise
            
    def exists(self, remote_key: str) -> bool:
        try:
            self.s3.head_object(Bucket=self.bucket_name, Key=remote_key)
            return True
        except ClientError:
            return False

この設計により、メインのプログラム側は storage.save_file() を呼ぶだけで良く、裏側がS3なのかローカルなのかを気にする必要がなくなります。

4. レジストリ操作クラスの構築

いよいよ、メタデータとストレージを統合する ModelRegistry クラスを実装します。ここでのポイントは、「メタデータとアーティファクト（重みファイル）の整合性」です。

モデルの登録処理は、メタデータファイル（JSON）と重みファイル（.bin / .safetensors）の両方が正常に保存されて初めて「成功」とみなすべきです。

import uuid
import tempfile

class ModelRegistry:
    def __init__(self, storage: StorageBackend):
        self.storage = storage

    def register_model(
        self, 
        local_model_path: str, 
        metadata: ModelMetadata
    ) -> str:
        """
        モデルをレジストリに登録する
        """
        print(f"Registering model: {metadata.model_id}...")

        # 1. 保存先のパスを決定
        artifact_key = f"{metadata.artifact_path}model.bin" # 実際は拡張子を動的に判定
        metadata_key = f"{metadata.artifact_path}metadata.json"

        # 2. 重みファイルのアップロード
        if not os.path.exists(local_model_path):
            raise FileNotFoundError(f"Model file not found: {local_model_path}")
        
        self.storage.save_file(local_model_path, artifact_key)

        # 3. メタデータの保存 (一時ファイルを作成してからアップロード)
        with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmp:
            tmp.write(metadata.to_json())
            tmp_path = tmp.name
        
        try:
            self.storage.save_file(tmp_path, metadata_key)
        finally:
            os.unlink(tmp_path)

        print("Registration complete.")
        return metadata.model_id

    def get_model_metadata(self, base_model: str, model_id: str) -> ModelMetadata:
        """メタデータを取得してオブジェクトとして返す"""
        key = f"models/{base_model}/{model_id}/metadata.json"
        
        with tempfile.NamedTemporaryFile(mode='r+', delete=False) as tmp:
            tmp_path = tmp.name
            
        try:
            self.storage.load_file(key, tmp_path)
            with open(tmp_path, 'r') as f:
                data = json.load(f)
            return ModelMetadata(data)
        finally:
            if os.path.exists(tmp_path):
                os.unlink(tmp_path)

学習スクリプトへの統合例

このクラスを使えば、学習スクリプトの最後は以下のようにシンプルになります。

# 学習完了後...

# S3ストレージでレジストリを初期化
registry = ModelRegistry(storage=S3Storage(bucket_name="project-ai-models"))

# メタデータの作成
meta = ModelMetadata(
    model_id=str(uuid.uuid4()),
    base_model="llama-3-8b",
    author="developer_name",
    git_commit_hash="a1b2c3d",
    dataset_hash="e5f6g7h",
    parameters={"lr": 2e-5, "epochs": 3},
    metrics={"loss": 0.45, "accuracy": 0.89}
)

# 登録実行
registry.register_model("./output/pytorch_model.bin", meta)

これで、誰がいつ、どのコードで学習し、どんな結果が出たかがクラウド上に構造化されて蓄積されます。

5. 運用と拡張：CI/CDとの連携

レジストリが整備されると、CI/CD（継続的インテグレーション/継続的デリバリー）パイプラインとの連携が可能になります。これはモデル運用の醍醐味と言えます。

例えば、GitHub Actionsなどのツールから、特定の基準を満たしたモデルだけを自動的に「検証環境（Staging）」へ進めるワークフローを組むことができます。

評価スコアによる自動タグ付け

単純な例として、「精度が0.85を超えたら staging タグを付ける」ロジックを考えてみましょう。

def promote_model_if_qualified(registry: ModelRegistry, base_model: str, model_id: str):
    meta = registry.get_model_metadata(base_model, model_id)
    
    # 評価基準
    accuracy = meta.metrics.get("accuracy", 0.0)
    THRESHOLD = 0.85

    if accuracy >= THRESHOLD:
        print(f"Model {model_id} passed validation (Acc: {accuracy}). Promoting to Staging.")
        
        # タグを更新してメタデータを再保存
        meta.tags["stage"] = "staging"
        
        # メタデータの更新処理（簡易版）
        # 実際には更新用のメソッドを用意するか、DB管理を併用することが望ましい
        with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmp:
            tmp.write(meta.to_json())
            tmp_path = tmp.name
        
        key = f"{meta.artifact_path}metadata.json"
        registry.storage.save_file(tmp_path, key)
        os.unlink(tmp_path)
    else:
        print(f"Model {model_id} failed validation (Acc: {accuracy}).")

このスクリプトを自動化パイプラインの最後に組み込むことで、「夜間に学習を回し、朝起きたら有望なモデルだけが検証環境にデプロイ準備完了になっている」**という効率的な状態を作り出せます。

次のステップ：API化に向けて

今回紹介したのはPythonクラスとしての実装ですが、プロジェクト規模が大きくなれば、これをFastAPIなどでラップして「モデルレジストリAPI」として独立させるのが自然な進化です。そうすれば、推論サーバーからのモデル動的ロードや、Web UIでの管理画面作成も容易になります。

まとめ

ファイル名による管理から脱却し、メタデータ駆動型のレジストリを構築することは、安定したAIシステム運用の第一歩です。今回紹介したPythonとS3による実装パターンは、シンプルながらも拡張性が高く、将来的に本格的なMLOpsツールへ移行する際にも、この「データ構造への理解」は大いに役立ちます。

自作の限界を感じたり、より高度なガバナンスが必要になった際は、生成AI開発のライフサイクル全体を支援する専門的なプラットフォームの導入を検討することをおすすめします。

まずは、今回紹介したコードをプロジェクトに組み込み、煩雑になりがちなモデル管理に論理的な秩序をもたらすところから始めてみませんか？