Hugging Face 모델 가져오기

AIP에서 기본 제공하는 언어 모델 외에 Hugging Face나 spaCy 같은 언어 모델 프레임워크를 Conda 또는 PyPI 패키지로 사용할 수 있습니다. 프레임워크가 사전 학습 가중치나 말뭉치를 인터넷에서 받는 경우에는 별도 준비가 필요합니다. 사용자 Python 코드는 기본적으로 공개 인터넷에 접근할 수 없습니다.

과거 Import an open source model 기능의 기본 어댑터는 더 이상 지원하거나 기능을 추가하지 않는 참조 구현입니다. 새 연결은 자신의 모델 어댑터 저장소에서 게시하고 유지합니다. Hugging Face 직렬화기를 사용하면 모델과 토크나이저의 저장 및 로드를 구성할 수 있습니다.

파일 데이터셋과 직접 다운로드

네트워크를 열지 않는 경로는 외부에서 모델 파일을 내려받아 비정형 데이터셋으로 가져오는 방식입니다. 직접 다운로드하려면 승인된 네트워크 송신 정책과 저장소의 외부 시스템 설정을 준비합니다. 두 경로 모두 모델을 로드한 뒤 어댑터로 감싸 게시합니다.

직접 관리하는 임베딩 어댑터

아래는 bert-base-cased 같은 AutoModel과 토크나이저를 사용하는 어댑터입니다. palantir_models, transformers, torch와 pandas가 필요합니다. Model Training 저장소의 src/main/model_adapters/embedding_adapter.py에 둡니다.

입력 inference_data는 text 문자열 열을 가진 표이고 output_data는 text와 embedding 벡터를 반환합니다. 한 번에 128개 문장을 토큰화하고 attention_mask로 패딩을 제외한 평균을 계산한 뒤 L2 정규화합니다.

모델의 장치 이동은 첫 predict에서 처리합니다. load와 추론이 서로 다른 프로세스에서 실행될 수 있으므로 load에서 CUDA 컨텍스트를 초기화하지 않습니다. GPU가 있으면 cuda:0, 없으면 CPU를 사용합니다.

import pandas as pd
import torch
import torch.nn.functional as F
import palantir_models as pm
from palantir_models.serializers import HfAutoModelSerializer, HfAutoTokenizerSerializer

class EmbeddingAdapter(pm.ModelAdapter):
    SENTENCE_BATCH_SIZE = 128

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

    @classmethod
    def api(cls):
        return (
            {"inference_data": pm.Pandas([("text", str)])},
            {"output_data": pm.Pandas([("text", str), ("embedding", list[float])])},
        )

    def predict(self, inference_data: pd.DataFrame):
        if self._device is None:
            self._device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
            self.model.to(self._device)
            self.model.eval()
        sentences = inference_data["text"].tolist()
        embeddings = []
        for start in range(0, len(sentences), self.SENTENCE_BATCH_SIZE):
            encoded = self.tokenizer(
                sentences[start:start + self.SENTENCE_BATCH_SIZE],
                padding=True, truncation=True, return_tensors="pt",
            ).to(self._device)
            with torch.no_grad():
                token_embeddings = self.model(**encoded)[0]
                mask = encoded["attention_mask"].unsqueeze(-1).float()
                pooled = (token_embeddings * mask).sum(dim=1) / mask.sum(dim=1).clamp(min=1e-9)
                normalized = F.normalize(pooled, p=2, dim=1)
            embeddings.extend(normalized.cpu().tolist())
        inference_data["embedding"] = embeddings
        return inference_data

모델 파일을 데이터셋으로 가져오기

Hugging Face에서 모델 파일을 내려받고 New > Dataset > Import > Select all files로 업로드합니다. 사설 네트워크의 공유 드라이브에 파일을 두었다면 Data Connection 동기화로 가져올 수도 있습니다. 생성한 데이터셋은 표 스키마가 없는 파일 데이터셋입니다.

모델의 Files and versions에 있는 설정, 토크나이저와 필요한 가중치 파일을 포함합니다. 같은 모델의 PyTorch용 pytorch_model.bin과 TensorFlow용 tf_model.h5 같은 대체 가중치를 모두 가져올 필요는 없으며 보통 PyTorch 형식을 권장합니다. 가중치가 여러 파일로 나뉜 모델은 필요한 파일 묶음을 모두 유지합니다.

모델이 크면 DRIVER_MEMORY_MEDIUM 같은 Spark 프로필로 드라이버 메모리를 늘려야 할 수 있습니다. 아래 경로는 실제 모델 파일 데이터셋과 출력 모델로 바꿉니다. CBAC 환경에서는 출력 모델을 미리 생성합니다.

from palantir_models.transforms import ModelOutput, copy_model_to_driver
from transforms.api import transform, Input
from transformers import AutoTokenizer, AutoModel
from main.model_adapters.embedding_adapter import EmbeddingAdapter

@transform(
    model_output=ModelOutput("/Models/published/embedding_model"),
    hugging_face_raw=Input("/Models/imported/hugging_face_files"),
)
def compute(model_output, hugging_face_raw):
    model_dir = copy_model_to_driver(hugging_face_raw.filesystem())
    tokenizer = AutoTokenizer.from_pretrained(model_dir)
    model = AutoModel.from_pretrained(model_dir)
    model_output.publish(model_adapter=EmbeddingAdapter(tokenizer, model))

버전 변경 종류를 명시하는 publish의 change_type에는 ModelVersionChangeType.MINOR 같은 SDK의 변경 유형 값을 지정할 수 있습니다. 위의 파일 가져오기 예제는 기본 게시 호출을 사용합니다.

직접 다운로드를 위한 도메인

Hugging Face에서 직접 내려받으려면 플랫폼 인스턴스의 네트워크 송신 정책에서 필요한 도메인을 허용합니다.

https://huggingface.co
https://hf.co
https://cas-bridge.xethub.hf.co
https://cas-server.xethub.hf.co
https://transfer.xethub.hf.co
https://cdn-lfs.hf.co
https://cdn-lfs-us-1.hf.co
https://cdn-lfs-eu-1.hf.co
https://us.gcp.cdn.hf.co
https://us.aws.cdn.hf.co

다운로드 도메인은 사전 안내 없이 바뀔 수 있습니다. 위 목록을 허용해도 실패하면 요청 로그에서 여전히 차단되는 대상 도메인을 확인합니다. 진단 로그에 인증 헤더나 토큰을 남기지 않도록 필요한 대상 정보만 확인합니다.

저장소에 transforms-external-systems를 추가하고 승인한 송신 정책을 사용하도록 외부 시스템 접근을 구성합니다. 아래는 EgressPolicy와 ExportControl을 사용하는 경로입니다. EGRESS_POLICY_RID와 EXPORT_MARKING_IDS는 관리자가 승인한 실제 정책 RID 및 Marking ID 목록으로 바꿉니다. 출력 모델 경로도 실제 값으로 설정합니다.

import tempfile
from palantir_models.transforms import ModelOutput
from transforms.api import transform
from transforms.external.systems import use_external_systems, EgressPolicy, ExportControl
from transformers import AutoTokenizer, AutoModel
from main.model_adapters.embedding_adapter import EmbeddingAdapter

EGRESS_POLICY_RID = "YOUR_APPROVED_EGRESS_POLICY_RID"
EXPORT_MARKING_IDS = ["YOUR_APPROVED_EXPORT_MARKING_ID"]

@use_external_systems(
    export_control=ExportControl(markings=EXPORT_MARKING_IDS),
    egress=EgressPolicy(EGRESS_POLICY_RID),
)
@transform(model_output=ModelOutput("/Models/published/embedding_model"))
def compute(export_control, egress, model_output):
    with tempfile.TemporaryDirectory() as cache_dir:
        tokenizer = AutoTokenizer.from_pretrained("bert-base-cased", cache_dir=cache_dir)
        model = AutoModel.from_pretrained("bert-base-cased", cache_dir=cache_dir)
        model_output.publish(model_adapter=EmbeddingAdapter(tokenizer, model))

PermissionError: [Errno 13] Permission denied: '/.cache'가 발생하면 읽기 전용 기본 캐시 대신 쓰기 가능한 디렉터리를 cache_dir로 지정합니다. 위 코드는 임시 캐시를 사용하여 모델 게시가 끝난 뒤 정리합니다.

게시한 모델 소비

게시한 모델 버전은 ModelInput으로 하위 변환에 로드합니다. 위 어댑터는 output_data를 반환하므로 model.transform(input_data).output_data를 사용합니다. 다른 어댑터의 출력 이름은 모델 버전의 API에서 확인합니다. 저장소 추론의 sidecar 및 의존성 조건을 따릅니다.