API：`palantir_models_serializers` リファレンス

palantir_models_serializers ライブラリは、Foundry 内で訓練されたモデルの保存と読み込みのための多くのデフォルトのシリアライゼーション方法を提供し、ほとんどのモデルはデフォルトのモデルシリアライザーの一つを使用することができます。

モデルシリアライザーの書き方

一部のケースでは、再利用可能な auto_serializer を作成することが有用なこともあります。例えば、ユーザーの組織が再利用し、頻繁に Foundry としてのモデルに統合するモデルフォーマットを持っている場合、再利用可能な auto_serializer を作成することで、異なるモデルやチーム間でのコードの重複を標準化し、減らすことができます。

auto_serializer を作成するには、palantir_models.models._serialization.ModelSerializer ベースクラスを拡張し、__init__、serialize、および deserialize メソッドを実装する必要があります。

__init__ メソッドは、シリアライザーの消費者が対話するコンストラクターです。
serialize メソッドは、palantir_models.models._state_accessors.ModelStateWriterにシリアライズするべきオブジェクトを提供されます。
deserialize メソッドはモデルの読み込み時に呼び出され、提供された palantir_models.models._state_accessors.ModelStateReaderからシリアライズされたモデルを再生成して返すことができます。

ユーザーの auto_serializer は、共有 Python ライブラリとして公開されるべきです。

提供されたシリアライザーの実装

参考のために、palantir_models_serializers内の既存のデフォルトシリアライザーの実装を提供します。

palantir_models_serializers.CloudPickleSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import importlib
from types import ModuleType
from palantir_models.models._serialization import ModelSerializer
from palantir_models.models._state_accessors import ModelStateWriter, ModelStateReader


class CloudPickleSerializer(ModelSerializer[object]):
    """cloudpickleライブラリを利用して一般的なオブジェクトをシリアライズするクラス"""

    file_name = "cloudpickle.pkl"
    cloudpickle: ModuleType

    def __init__(self):
        # "cloudpickle"モジュールをインポート
        self.cloudpickle = importlib.import_module("cloudpickle")

    def serialize(self, writer: ModelStateWriter, obj: object):
        # オブジェクトをシリアライズしてファイルに保存
        with writer.open(self.file_name, "wb") as cloudpickle_file:
            self.cloudpickle.dump(obj, cloudpickle_file)

    def deserialize(self, reader: ModelStateReader) -> object:
        # ファイルからデータを読み込み、オブジェクトをデシリアライズ
        with reader.open(self.file_name, "rb") as cloudpickle_file:
            obj = self.cloudpickle.load(cloudpickle_file)
        return obj

palantir_models_serializers.DillSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import importlib
from types import ModuleType
from palantir_models.models._serialization import ModelSerializer
from palantir_models.models._state_accessors import ModelStateWriter, ModelStateReader


class DillSerializer(ModelSerializer[object]):
    """dillライブラリを利用した一般的なオブジェクトのシリアライザー"""

    file_name = "dill.pkl"
    dill: ModuleType

    def __init__(self):
        # dillモジュールをインポート
        self.dill = importlib.import_module("dill")

    def serialize(self, writer: ModelStateWriter, obj: object):
        # dillファイルをバイナリモードで書き込み
        with writer.open(self.file_name, "wb") as dill_file:
            # オブジェクトをdillファイルに書き込む
            self.dill.dump(obj, dill_file, recurse=True)

    def deserialize(self, reader: ModelStateReader) -> object:
        # dillファイルをバイナリモードで読み込み
        with reader.open(self.file_name, "rb") as dill_file:
            # dillファイルからオブジェクトを読み込む
            obj = self.dill.load(dill_file)
        return obj

palantir_models_serializers.HfAutoModelSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class HfAutoModelSerializer(ModelSerializer):
    """
    huggingface transformers AutoModel クラスのシリアライザで、
    from_pretrained および save_pretrained メソッドを使用します。
    特定のサブクラスを設定したり (例: AutoModelForSequenceClassification や
    BertForTokenClassification)、from_pretrained に追加の kwargs を渡すことができます
    (例: num_labels=2)。
    """

    DIR_NAME = "model"

    def __init__(self, model_class=None, **load_kwargs):
        if model_class is None:
            transformers = importlib.import_module("transformers")
            model_class = transformers.AutoModel
        self.model_class = model_class
        self.load_kwargs = load_kwargs

    # シリアライズ
    def serialize(self, writer: ModelStateWriter, obj):
        model_dir = writer.mkdir(self.DIR_NAME)
        obj.save_pretrained(model_dir)

    # デシリアライズ
    def deserialize(self, reader: ModelStateReader):
        model_dir = reader.dir(self.DIR_NAME)
        return self.model_class.from_pretrained(model_dir, **self.load_kwargs)

palantir_models_serializers.HfAutoTokenizerSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class HfAutoTokenizerSerializer(ModelSerializer):
    """
    huggingface transformersのAutoTokenizer用のシリアライザ。
    """

    DIR_NAME = "tokenizer"

    def __init__(self, tokenizer_class=None, **load_kwargs):
        # tokenizer_classが指定されていない場合、huggingfaceのAutoTokenizerをデフォルトとして使用
        if tokenizer_class is None:
            transformers = importlib.import_module("transformers")
            tokenizer_class = transformers.AutoTokenizer
        self.tokenizer_class = tokenizer_class
        self.load_kwargs = load_kwargs

    def serialize(self, writer: ModelStateWriter, obj):
        # 指定したディレクトリにトークナイザーを保存
        tokenizer_dir = writer.mkdir(self.DIR_NAME)
        obj.save_pretrained(tokenizer_dir)

    def deserialize(self, reader: ModelStateReader):
        # 指定したディレクトリからトークナイザーをロード
        tokenizer_dir = reader.dir(self.DIR_NAME)
        return self.tokenizer_class.from_pretrained(tokenizer_dir, **self.load_kwargs)

palantir_models_serializers.HfPipelineSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import importlib
from palantir_models import ModelSerializer, ModelStateReader, ModelStateWriter


class HfPipelineSerializer(ModelSerializer):
    """
    Huggingfaceのtransformersパイプラインのシリアライザ。
    パイプラインタスク（例：感情分析）を設定することができます。
    """

    DIR_NAME = "pipeline"

    # 初期化メソッド
    def __init__(self, pipeline_type, **load_kwargs):
        self.transformers = importlib.import_module("transformers")  # transformersモジュールをインポート
        self.pipeline_type = pipeline_type  # パイプラインのタイプを設定
        self.load_kwargs = load_kwargs  # 追加の引数を設定

    # シリアライズメソッド
    def serialize(self, writer: ModelStateWriter, obj):
        pipeline_dir = writer.mkdir(self.DIR_NAME)  # ディレクトリを作成
        obj.save_pretrained(pipeline_dir)  # 事前学習モデルを保存

    # デシリアライズメソッド
    def deserialize(self, reader: ModelStateReader):
        pipeline_dir = reader.dir(self.DIR_NAME)  # ディレクトリを取得
        # パイプラインを生成し、そのモデルをディレクトリからロード
        return self.transformers.pipeline(self.pipeline_type, model=pipeline_dir, **self.load_kwargs)

palantir_models_serializers.JsonSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import importlib
from types import ModuleType
from typing import Dict
from palantir_models.models._serialization import ModelSerializer
from palantir_models.models._state_accessors import ModelStateWriter, ModelStateReader


class JsonSerializer(ModelSerializer[Dict]):
    """jsonに変換可能なオブジェクトと辞書のためのシリアライザ"""

    file_name = "config.json"
    json: ModuleType

    def __init__(self):
        # jsonモジュールを動的にインポート
        self.json = importlib.import_module("json")

    def serialize(self, writer: ModelStateWriter, obj: Dict):
        # オブジェクトをjson形式で書き出し
        with writer.open(self.file_name, "w") as conf:
            self.json.dump(obj, conf)

    def deserialize(self, reader: ModelStateReader) -> Dict:
        # json形式のデータを読み込みオブジェクトに変換
        with reader.open(self.file_name, "r") as conf:
            return self.json.load(conf)

palantir_models_serializers.PytorchStateSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import importlib
from types import ModuleType
from palantir_models.models._serialization import ModelSerializer
from palantir_models.models._state_accessors import ModelStateWriter, ModelStateReader


class PytorchStateSerializer(ModelSerializer):
    """PyTorchの状態辞書のシリアライザ。"""

    STATE_DICT_FILE_NAME = "model_state_dict.pt"
    torch: ModuleType

    def __init__(self):
        # "torch"モジュールを動的にインポート
        self.torch = importlib.import_module("torch")

    def serialize(self, writer: ModelStateWriter, obj: dict):
        """PyTorchモデルのstate_dictをシリアル化。"""
        # シリアル化したモデルをファイルに保存
        with writer.open(self.STATE_DICT_FILE_NAME, "wb") as file_path:
            self.torch.save(obj, file_path)

    def deserialize(self, reader: ModelStateReader) -> dict:
        """PyTorchモデルのstate_dictをデシリアル化。"""
        # ファイルからモデルの状態を読み込み、デシリアル化
        with reader.open(self.STATE_DICT_FILE_NAME, "rb") as file_path:
            state_dict = self.torch.load(file_path)
            return state_dict

palantir_models_serializers.TensorflowKerasSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import importlib
from types import ModuleType
from palantir_models.models._serialization import ModelSerializer
from palantir_models.models._state_accessors import ModelStateWriter, ModelStateReader


class TensorflowKerasSerializer(ModelSerializer):
    """tensorflowのkerasモデル用のシリアライザ"""

    DIR_NAME: str = "tensorflow_saved_model_dir"
    tensorflow: ModuleType

    def __init__(self):
        # tensorflowモジュールをインポート
        self.tensorflow = importlib.import_module("tensorflow")

    def serialize(self, writer: ModelStateWriter, obj: "tensorflow.keras.Model"):
        # モデルを保存するディレクトリを作成
        dir_path = writer.mkdir(self.DIR_NAME)
        # モデルを保存
        obj.save(dir_path)

    def deserialize(self, reader: ModelStateReader) -> "tensorflow.keras.Model":
        # モデルを読み込むディレクトリのパスを取得
        dir_path = reader.dir(self.DIR_NAME)
        # モデルを読み込み（コンパイルはまだ行わない）
        obj = self.tensorflow.keras.models.load_model(dir_path, compile=False)
        # モデルをコンパイル
        obj.compile()
        # モデルを返す
        return obj

palantir_models_serializers.XGBoostSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from palantir_models import ModelSerializer
from palantir_models.models._serialization import ModelStateReader, ModelStateWriter
from xgboost.sklearn import XGBModel

# XGBoostのシリアライザークラス
class XGBoostSerializer(ModelSerializer[XGBModel]):
    """Simple Serializer for XGBoost SkLearn Models."""
    # シリアル化されたモデルのファイル名
    file_name = "xgboost_model.json"

    # モデルをシリアル化するメソッド
    def serialize(self, writer: ModelStateWriter, obj: XGBModel):
        with writer.open(self.file_name, "w") as xgbfile:
            obj.save_model(xgbfile.name)

    # シリアル化されたモデルをデシリアル化（復元）するメソッド
    def deserialize(self, reader: ModelStateReader) -> XGBModel:
        model = XGBModel()
        with reader.open(self.file_name, "r") as xgbfile:
            model.load_model(xgbfile.name)
            return model

palantir_models_serializers.YamlSerializer

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import importlib
from types import ModuleType
from typing import Dict
from palantir_models.models._serialization import ModelSerializer
from palantir_models.models._state_accessors import ModelStateWriter, ModelStateReader


class YamlSerializer(ModelSerializer[Dict]):
    """yaml変換可能なオブジェクトと辞書のシリアライザ"""

    file_name = "config.yaml"
    yaml: ModuleType

    def __init__(self):
        # yamlモジュールをインポート
        self.yaml = importlib.import_module("yaml")

    # オブジェクトをシリアライズ（yaml形式に変換）してファイルに書き込む
    def serialize(self, writer: ModelStateWriter, obj: Dict):
        with writer.open(self.file_name, "w") as conf:
            self.yaml.safe_dump(obj, conf)

    # yamlファイルをデシリアライズ（読み込んで辞書に変換）する
    def deserialize(self, reader: ModelStateReader) -> Dict:
        with reader.open(self.file_name, "r") as conf:
            return self.yaml.safe_load(conf)

API：palantir_models_serializers リファレンス