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는 공유 파이썬 라이브러리로 게시되어야 합니다.

제공된 직렬화 도구의 구현

참고로, 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
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 라이브러리를 사용하여 일반 객체를 직렬화하는 Serializer"""

    file_name = "cloudpickle.pkl"
    cloudpickle: ModuleType

    def __init__(self):
        self.cloudpickle = importlib.import_module("cloudpickle")

    def serialize(self, writer: ModelStateWriter, obj: object):
        # cloudpickle 파일을 쓰기 모드로 연 후 객체를 직렬화합니다.
        with writer.open(self.file_name, "wb") as cloudpickle_file:
            self.cloudpickle.dump(obj, cloudpickle_file)

    def deserialize(self, reader: ModelStateReader) -> object:
        # cloudpickle 파일을 읽기 모드로 열고 객체를 역직렬화합니다.
        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
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 라이브러리를 이용하여 일반 객체를 직렬화하는 Serializer"""

    file_name = "dill.pkl"
    dill: ModuleType

    def __init__(self):
        # "dill" 모듈을 동적으로 임포트합니다.
        self.dill = importlib.import_module("dill")

    def serialize(self, writer: ModelStateWriter, obj: object):
        # writer를 사용해 객체를 직렬화하고 파일에 저장합니다.
        with writer.open(self.file_name, "wb") as dill_file:
            self.dill.dump(obj, dill_file, recurse=True)

    def deserialize(self, reader: ModelStateReader) -> object:
        # reader를 사용해 파일에서 객체를 역직렬화합니다.
        with reader.open(self.file_name, "rb") as dill_file:
            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 클래스를 위한 Serializer로,
    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
class HfAutoTokenizerSerializer(ModelSerializer):
    """
    huggingface transformers AutoTokenizer에 대한 Serializer 입니다.
    """

    DIR_NAME = "tokenizer"

    def __init__(self, tokenizer_class=None, **load_kwargs):
        if tokenizer_class is None:
            transformers = importlib.import_module("transformers")  # "transformers" 모듈을 가져옵니다.
            tokenizer_class = transformers.AutoTokenizer  # AutoTokenizer 클래스를 가져옵니다.
        self.tokenizer_class = tokenizer_class  # tokenizer_class를 인스턴스 변수로 설정합니다.
        self.load_kwargs = load_kwargs  # load_kwargs를 인스턴스 변수로 설정합니다.

    def serialize(self, writer: ModelStateWriter, obj):
        tokenizer_dir = writer.mkdir(self.DIR_NAME)  # 지정된 디렉토리 이름으로 새 디렉토리를 만듭니다.
        obj.save_pretrained(tokenizer_dir)  # tokenizer를 지정된 디렉토리에 저장합니다.

    def deserialize(self, reader: ModelStateReader):
        tokenizer_dir = reader.dir(self.DIR_NAME)  # 지정된 디렉토리 이름의 디렉토리를 가져옵니다.
        return self.tokenizer_class.from_pretrained(tokenizer_dir, **self.load_kwargs)  # 저장된 tokenizer를 로드합니다.

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
import importlib
from palantir_models import ModelSerializer, ModelStateReader, ModelStateWriter

# Huggingface transformer 파이프라인을 직렬화 하기 위한 클래스입니다.
# 파이프라인 작업 설정(예: 감성 분석)이 가능합니다.
class HfPipelineSerializer(ModelSerializer):

    DIR_NAME = "pipeline"

    # 초기화 함수에서는 파이프라인 타입과 로드 옵션을 설정합니다.
    def __init__(self, pipeline_type, **load_kwargs):
        self.transformers = importlib.import_module("transformers")  # transformers 모듈을 import 합니다.
        self.pipeline_type = pipeline_type  # 파이프라인 타입을 저장합니다.
        self.load_kwargs = load_kwargs  # 로드 옵션을 저장합니다.

    # serialize 함수에서는 모델 상태를 쓰는 writer를 이용해 파이프라인을 저장합니다.
    def serialize(self, writer: ModelStateWriter, obj):
        pipeline_dir = writer.mkdir(self.DIR_NAME)  # pipeline 디렉토리를 생성합니다.
        obj.save_pretrained(pipeline_dir)  # 사전 학습된 모델을 pipeline 디렉토리에 저장합니다.

    # deserialize 함수에서는 모델 상태를 읽는 reader를 이용해 파이프라인을 불러옵니다.
    def deserialize(self, reader: ModelStateReader):
        pipeline_dir = reader.dir(self.DIR_NAME)  # pipeline 디렉토리를 가져옵니다.
        return self.transformers.pipeline(self.pipeline_type, model=pipeline_dir, **self.load_kwargs)  # transformers의 pipeline 함수를 이용해 파이프라인을 불러옵니다.

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
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

# JSON 직렬화/역직렬화를 다루는 클래스
class JsonSerializer(ModelSerializer[Dict]):
    """json 변환 가능한 객체와 딕셔너리를 위한 직렬화기"""

    file_name = "config.json"
    json: ModuleType

    def __init__(self):
        self.json = importlib.import_module("json")

    # 객체를 직렬화하여 파일에 저장합니다.
    def serialize(self, writer: ModelStateWriter, obj: Dict):
        with writer.open(self.file_name, "w") as conf:
            self.json.dump(obj, conf)

    # 파일에서 데이터를 읽어와 역직렬화하여 반환합니다.
    def deserialize(self, reader: ModelStateReader) -> Dict:
        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:
            # obj를 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:
            # file_path에서 state_dict를 로드하는 코드
            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
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 모델이 저장되는 디렉토리 이름
    tensorflow: ModuleType

    def __init__(self):
        self.tensorflow = importlib.import_module("tensorflow")  # 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
22
from palantir_models import ModelSerializer
from palantir_models.models._serialization import ModelStateReader, ModelStateWriter
from xgboost.sklearn import XGBModel


class XGBoostSerializer(ModelSerializer[XGBModel]):
    """XGBoost SkLearn 모델을 위한 간단한 Serializer."""

    file_name = "xgboost_model.json"

    def serialize(self, writer: ModelStateWriter, obj: XGBModel):
        # writer를 사용하여 xgboost 모델 저장
        with writer.open(self.file_name, "w") as xgbfile:
            obj.save_model(xgbfile.name)

    def deserialize(self, reader: ModelStateReader) -> XGBModel:
        # 새로운 XGBoost 모델 객체 생성
        model = XGBModel()
        # reader를 사용하여 xgboost 모델 불러오기
        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
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 변환 가능한 객체와 사전에 대한 Serializer"""

    file_name = "config.yaml"
    yaml: ModuleType  # yaml 모듈 타입

    def __init__(self):
        self.yaml = importlib.import_module("yaml")  # yaml 모듈 임포트

    def serialize(self, writer: ModelStateWriter, obj: Dict):
        # yaml 형태로 객체를 직렬화하여 파일에 쓴다.
        with writer.open(self.file_name, "w") as conf:
            self.yaml.safe_dump(obj, conf)

    def deserialize(self, reader: ModelStateReader) -> Dict:
        # yaml 파일을 읽어서 객체로 역직렬화한다.
        with reader.open(self.file_name, "r") as conf:
            return self.yaml.safe_load(conf)

API: palantir_models_serializers 출처