모델의 직렬화와 복원

게시한 모델을 다른 업무에서 재사용하려면 저장된 가중치와 설정으로 어댑터를 다시 초기화할 수 있어야 합니다. 모델 형식에 맞는 직렬화 방법을 지정합니다. 컨테이너 모델은 보통 컨테이너 수명주기 안에서, 외부 모델은 외부 호스팅 환경에서 가중치를 관리합니다.

auto_serialize

어댑터의 __init__에 @auto_serialize를 붙이면 생성자 입력을 저장하고 로드합니다. 인자를 지정하지 않으면 각 입력에 Dill을 사용합니다. 특정 직렬화기를 지정할 때에는 생성자의 모든 입력에 대응하는 직렬화기를 선언해야 합니다.

인자 없는 데코레이터는 palantir_models 0.1536.0 이상에서 지원합니다. 그보다 낮은 버전에서는 모든 직렬화기를 인자로 명시해야 합니다. palantir_models.serializers가 의존성으로 포함하는 직렬화 프레임워크는 dill뿐이므로 다른 직렬화기를 쓰면 해당 라이브러리도 환경에 추가합니다.

다음 예제는 0부터 99까지의 값과 그 두 배를 학습한 선형 회귀 모델을 게시합니다. 모델 경로를 실제 출력으로 바꾸고 scikit-learn 및 numpy를 설치합니다. 어댑터의 입력에는 학습에 사용한 단일 특성 열을 전달합니다.

from transforms.api import transform
from palantir_models.transforms import ModelOutput

from sklearn.linear_model import LinearRegression
import numpy as np


@transform(
    model_output=ModelOutput("/Foundry/path/to/model_asset"),
)
def compute(model_output):
    x_values = [i for i in range(100)]
    y_values = [2 * i for i in x_values]

    X = np.array(x_values, dtype=np.float32).reshape(-1, 1)
    y = np.array(y_values, dtype=np.float32).reshape(-1, 1)

    model = LinearRegression()
    model.fit(X, y)

    model_output.publish(
        model_adapter=AutoSerializationAdapter(
            model
        )
    )

import palantir_models as pm

class AutoSerializationAdapter(pm.ModelAdapter):

    @pm.auto_serialize
    def __init__(self, model):
        self.model = model

    @classmethod
    def api(cls):
        inputs = {"df_in": pm.Pandas()}
        outputs = {"df_out": pm.Pandas()}
        return inputs, outputs

    def predict(self, df_in):
        df_in['prediction'] = self.model.predict(df_in)
        return df_in

이 예제의 빈 Pandas 열 선언은 입력 스키마를 고정하지 않습니다. 함수를 게시하려면 필수 열과 자료형을 명시해야 합니다.

생성자 입력마다 다른 형식 사용

모델은 Dill, 설정 사전은 JSON으로 저장할 수 있습니다. 아래 어댑터는 config의 prediction_column이 있으면 그 이름으로 결과를 기록하고 없으면 prediction을 사용합니다.

import palantir_models as pm
from palantir_models.serializers import DillSerializer, JsonSerializer

class AutoSerializationAdapter(pm.ModelAdapter):

    @pm.auto_serialize(
        model=DillSerializer(),
        config=JsonSerializer()
    )
    def __init__(self, model, config={}):
        self.model = model
        self.prediction_column = config['prediction_column'] if 'prediction_column' in config else 'prediction'

    @classmethod
    def api(cls):
        inputs = {"df_in": pm.Pandas()}
        outputs = {"df_out": pm.Pandas()}
        return inputs, outputs

    def predict(self, df_in):
        df_in[self.prediction_column] = self.model.predict(df_in)
        return df_in

제공되는 직렬화기

파일 이름과 형식 선택, 사용자 정의 직렬화기 인터페이스는 직렬화기 참조에 있습니다.

Hugging Face

HfPipelineSerializer, HfAutoTokenizerSerializer와 HfAutoModelSerializer를 제공하며 모두 transformers 의존성이 필요합니다.

파이프라인

HfPipelineSerializer는 save_pretrained로 파이프라인을 저장하고 로드 시 다시 만듭니다. 첫 번째 필수 문자열은 ner 같은 작업 이름이며 추가 키워드 인자는 파이프라인 로드에 전달합니다.

아래 어댑터에는 이미 초기화한 NER pipeline을 전달합니다. NER 결과는 개체 사전 목록이므로 문자열로 선언한 generation 열에 JSON으로 기록합니다. 각 개체의 score는 JSON 수치로 변환합니다.

import json
import pandas as pd
import palantir_models as pm
from palantir_models.serializers import HfPipelineSerializer

class HFNerAdapter(pm.ModelAdapter):
    @pm.auto_serialize(pipeline=HfPipelineSerializer("ner"))
    def __init__(self, pipeline):
        self.pipeline = pipeline

    @classmethod
    def api(cls):
        inputs = {"df_in": pm.Pandas([("text", str)])}
        outputs = {"df_out": pm.Pandas([("text", str), ("generation", str)])}
        return inputs, outputs

    def predict(self, df_in: pd.DataFrame):
        results = self.pipeline(df_in["text"].tolist())
        df_in["generation"] = [
            json.dumps(
                [{**entity, "score": float(entity["score"])} for entity in entities],
                ensure_ascii=False,
            )
            for entities in results
        ]
        return df_in

모델과 토크나이저

HfAutoModelSerializer와 HfAutoTokenizerSerializer는 save_pretrained와 from_pretrained를 사용합니다. 기본 AutoModel 및 AutoTokenizer 대신 구체 클래스를 써야 하는 모델은 각각 model_class 또는 tokenizer_class 첫 인자로 지정할 수 있습니다.

아래 예제는 AutoModelForSeq2SeqLM 모델과 그 토크나이저를 각각 저장합니다. 추론에서 입력을 토큰화하고 generate 결과를 문자열로 디코딩합니다.

import palantir_models as pm
from palantir_models.serializers import HfAutoModelSerializer, HfAutoTokenizerSerializer
from transformers import AutoModelForSeq2SeqLM
import pandas as pd
import torch


class HFTextGenerationAdapter(pm.ModelAdapter):

    @pm.auto_serialize(
        model=HfAutoModelSerializer(AutoModelForSeq2SeqLM),
        tokenizer=HfAutoTokenizerSerializer()
    )
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer

    @classmethod
    def api(cls):
        inputs = {"df_in" : pm.Pandas([("text", str)])}
        outputs = {"df_out" : pm.Pandas([("text", str), ("generation", str)])}
        return inputs, outputs

    def predict(self, df_in: pd.DataFrame):
        input_ids = self.tokenizer(
            df_in["text"].tolist(),
            return_tensors="pt",
            padding=True
        ).input_ids

        outputs = self.model.generate(input_ids)
        result = self.tokenizer.batch_decode(outputs, skip_special_tokens=True)

        df_in["generation"] = result
        return df_in

사용자 정의 방식

반복해서 사용하는 조직 고유 형식은 ModelSerializer를 확장하여 직렬화기로 만들고 공유 Python 라이브러리로 게시할 수 있습니다. 참조의 생성자, serialize와 deserialize 계약을 따릅니다. 기본 직렬화기로 처리할 수 없는 초기화는 어댑터의 save와 load를 직접 구현합니다. 추가 기본 직렬화기가 필요하면 Palantir 담당자에게 요청할 수 있습니다.