ModelAdapter 실행 인터페이스 참조

ModelAdapter는 게시한 Python 라이브러리에서 모델 산출물과 Foundry를 연결하는 기반 클래스입니다. 저장과 로드, API 및 추론 로직을 제공하여 모델을 초기화하고 실행합니다.

어댑터 메서드

기본 직렬화기를 사용하는 @auto_serialize 어댑터는 save와 load를 별도로 구현하지 않아도 됩니다. predict로 단일 출력을 구현하면 run_inference를 함께 정의하지 않습니다. 일반 모델은 load에서 파일을 읽으며 컨테이너 모델은 시작한 서비스 주소를, 외부 모델은 접속 설정과 클라이언트를 초기화합니다.

ModelStateWriter

ModelAdapter.save에 전달하며 모델 저장소에 기록할 산출물을 제공합니다.

변환에서 모델을 publish하면 save가 실행됩니다. Writer에 기록한 내용은 ZIP으로 묶여 새 모델 버전이 참조하는 산출물 저장소에 보관됩니다.

ModelStateReader

ModelInput으로 로드하거나 모델 기반 온라인 또는 배치 배포를 시작할 때 load에 전달합니다. save가 기록한 같은 산출물 저장소를 읽으며 변환과 추론 로직보다 먼저 초기화합니다.

컨테이너 모델에도 사용자가 올린 ZIP을 읽는 ModelStateReader가 제공됩니다.

파일 저장과 로드 예제

다음 어댑터는 model.pkl에 모델을 저장하고 같은 파일로 복원합니다. model은 input_feature라는 특성 열을 받아 행별 수치 예측을 반환하는 실제 모델입니다. 해당 클래스와 의존성은 어댑터 환경에 있어야 합니다.

import pickle
import palantir_models as pm

class ExampleModelAdapter(pm.ModelAdapter):
    def __init__(self, model):
        self.model = model

    def save(self, state_writer: pm.ModelStateWriter):
        with state_writer.open("model.pkl", "wb") as model_outfile:
            pickle.dump(self.model, model_outfile)

    @classmethod
    def load(cls, state_reader: pm.ModelStateReader):
        with state_reader.open("model.pkl", "rb") as model_infile:
            return cls(pickle.load(model_infile))

    @classmethod
    def api(cls):
        return (
            {"df_in": pm.Pandas([("input_feature", float)])},
            {"df_out": pm.Pandas([("prediction", float)])},
        )

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

컨테이너 초기화 컨텍스트

ContainerizedApplicationContext는 컨테이너 이미지로 실행하는 모델의 load에만 제공됩니다. 어댑터 코드에서 이 타입을 반드시 가져올 필요는 없습니다.

services는 컨테이너 이름에서 서비스 주소 목록으로 가는 맵입니다. 한 컨테이너에 여러 포트를 열 수 있으므로 주소도 여러 개일 수 있습니다. 다음은 설명용 값입니다.

{
    "container1": ["localhost:8080"],
    "container2": ["localhost:8080", "localhost:8081"],
}

shared_empty_dir_mount_path는 모든 컨테이너와 어댑터가 함께 읽고 쓸 수 있는 빈 디렉터리의 마운트 경로입니다. 실제 컨텍스트는 container_context.services["container1"][0]와 container_context.shared_empty_dir_mount_path처럼 접근합니다. 이를 어댑터 인스턴스에 저장하여 추론 요청을 보낼 호스트와 포트 및 공유 파일 경로로 사용합니다.

외부 모델 초기화 컨텍스트

ExternalModelContext는 외부 호스팅 모델의 load에만 제공됩니다. load의 실제 타입 표기는 models_api.models_api_executable.ExternalModelExecutionContext를 사용할 수 있습니다.

예를 들어 어댑터는 connection_configuration의 model_name 및 model_parameter, resolved_credentials의 secret_key를 실제 외부 클라이언트 초기화에 전달할 수 있습니다. 연결 클라이언트의 생성 계약에 맞춰 URL과 설정 및 자격 증명을 전달하고 해당 클라이언트를 추론에서 사용합니다.

모델 API의 저수준 입력 선언

여기의 palantir_models.models.api.ModelInput 및 ModelOutput은 API 자료형 선언입니다. 변환에서 모델 리소스를 가져오는 palantir_models.transforms.ModelInput과 구분합니다. 일반적인 어댑터는 pm.Pandas와 pm.Parameter 등의 간편 선언을 사용합니다.

표 입력에는 TransformInput, Pandas DataFrame 및 Spark DataFrame을 전달할 수 있습니다. 직접 전달한 Spark DataFrame은 DFType.SPARK 입력에만 지원하며 DFType.PANDAS 입력으로 자동 변환하지 않습니다. TransformInput은 API에 지정한 표 형식으로 읽어 전달합니다.

표 열은 str, int, float, bool, list, dict, datetime.date, datetime.time, datetime.datetime, typing.Any 및 MediaReference를 지원합니다. typing의 List[str]와 Dict[str, float] 같은 구체 타입도 사용할 수 있습니다.

Parameter 입력은 str, int, float, bool, list, dict, typing.Any 및 pm.NDArray를 지원합니다. 생략하면 default를 사용하고 자료형이 맞지 않으면 오류가 발생합니다. FileSystemInput에는 TransformInput만 지원합니다. Object는 해당 객체 인스턴스 또는 기본 키, ObjectSet은 해당 집합 인스턴스 또는 객체 집합 RID로 전달합니다.

검증의 차이

배치 추론에서는 표 열의 API를 엄격하게 강제하지 않으며 선언된 열이 일부 없어도 API 검증 자체가 오류를 내지는 않습니다. 실제 예측 코드가 없는 열을 참조하면 그 코드에서 실패할 수 있습니다. 선언하지 않은 추가 열은 predict에 전달하므로 평가용 정답 열을 유지할 수 있습니다.

직접 온라인 배포는 선언한 자료형으로 값을 변환하며 필수 열이 없거나 변환할 수 없으면 오류가 발생합니다. Modeling Objective 온라인 배포에는 같은 자동 자료형 변환이 없습니다. model.transform의 Parameter 입력은 자료형을 검사합니다. MediaReference는 배치에서도 각 참조 문자열을 미디어 참조 객체로 변환합니다. Functions는 강한 타입 계약을 사용하므로 게시 조건을 별도로 충족해야 합니다.

출력 선언

TabularOutput(name)은 Pandas 또는 Spark DataFrame을 받습니다. 입력과 달리 df_type을 지정하지 않습니다. ParameterOutput(name, type, default=None)은 단일 값 출력을 선언하며 run_inference에서 쓰지 않으면 default를 사용합니다. 기본 매개변수 타입은 str, int, float, bool, list와 dict이며 NDArray 같은 확장 타입은 간편 API에서 정의합니다.

파일시스템 출력은 쓰기용 open 메서드를 제공하고 run_inference에서 기록한 뒤 transform_write로 소비합니다. 단일 표 또는 매개변수 출력은 predict가 값을 반환하면 그 출력 이름에 연결합니다.

여러 입력과 여러 출력

predict는 입력 이름으로 값을 받습니다. 예를 들어 API에 input_1과 input_2를 선언하면 predict(self, input_1, input_2)에서 두 표를 사용하고 단일 output_dataframe에 해당하는 DataFrame을 반환합니다.

run_inference의 inputs와 outputs는 이름 있는 객체입니다. 입력 df_in에 input_feature(float)를, 출력 df_out_one에 feature_one(int)를, df_out_two에 output_feature_1(int) 및 output_feature_2(float)를 선언했다면 inputs.df_in으로 읽고 각 출력 객체에 맞는 표를 씁니다. 출력의 write는 해당 출력 값을 지정합니다.

다음 완전한 어댑터는 같은 특성을 쓰는 두 분류 모델을 받습니다. first_model과 second_model은 input_feature 표에 대해 정수 분류 결과를 반환해야 합니다. 각각 계산한 결과를 이름이 다른 두 출력으로 기록합니다.

import pandas as pd
import palantir_models as pm

class TwoOutputAdapter(pm.ModelAdapter):
    @pm.auto_serialize
    def __init__(self, first_model, second_model):
        self.first_model = first_model
        self.second_model = second_model

    @classmethod
    def api(cls):
        return (
            {"input_dataframe": pm.Pandas([("input_feature", float)])},
            {
                "output_dataframe_one": pm.Pandas([("output_feature", int)]),
                "output_dataframe_two": pm.Pandas([("output_feature", int)]),
            },
        )

    def run_inference(self, inputs, outputs):
        features = inputs.input_dataframe[["input_feature"]]
        first = pd.DataFrame({"output_feature": self.first_model.predict(features)})
        second = pd.DataFrame({"output_feature": self.second_model.predict(features)})
        outputs.output_dataframe_one.write(first)
        outputs.output_dataframe_two.write(second)

이를 변환에서 소비하면 transform 결과의 같은 이름으로 각 DataFrame에 접근합니다. 다음 예시 경로는 실제 리소스로 바꾸며 경량 sidecar에 필요한 palantir_models 0.2010.0 이상을 사용합니다.

from transforms.api import Input, Output, transform
from palantir_models.transforms import ModelInput

@transform.using(
    my_input_data=Input("/Project/data/input"),
    my_output_data_one=Output("/Project/data/prediction_one"),
    my_output_data_two=Output("/Project/data/prediction_two"),
    my_model=ModelInput("/Project/models/two_output_model", use_sidecar=True),
)
def compute(my_input_data, my_output_data_one, my_output_data_two, my_model):
    result = my_model.transform(my_input_data)
    my_output_data_one.write_pandas(result.output_dataframe_one)
    my_output_data_two.write_pandas(result.output_dataframe_two)

MediaReference

미디어 참조 입력과 MediaReference 열은 참조 문자열을 런타임 객체로 변환합니다. 참조의 리소스 관계는 미디어 참조에서 설명합니다.

다음 어댑터의 model은 실제 미디어 바이트를 받아 문자열 분류를 반환하는 classify(bytes) 메서드를 제공해야 합니다. 이미지 열의 참조를 열어 읽은 뒤 그 결과를 반환합니다.

import palantir_models as pm

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

    @classmethod
    def api(cls):
        return (
            {"df_in": pm.Pandas(columns=[("image", pm.MediaReference)])},
            {"df_out": pm.Pandas(columns=[("prediction", str)])},
        )

    def predict(self, df_in):
        predictions = []
        for media_reference in df_in["image"]:
            with media_reference.get_media_item() as media_item:
                predictions.append(self.model.classify(media_item.read()))
        df_in["prediction"] = predictions
        return df_in