refactor: Simplify huggingface hub api implementation

This commit is contained in:
teleprint-me 2024-05-23 20:50:15 -04:00
parent c92c6ad480
commit 1749209406
No known key found for this signature in database
GPG key ID: B0D11345E65C4D48

View file

@ -7,7 +7,15 @@ from hashlib import sha256
import requests import requests
from transformers import AutoTokenizer from transformers import AutoTokenizer
from .constants import HF_MODEL_MAP, LLaMaModelType, LLaMaVocabType from .constants import (
GPT_PRE_TOKENIZER_DEFAULT,
HF_TOKENIZER_BPE_FILES,
HF_TOKENIZER_SPM_FILES,
MODEL_FILE_TYPE_NAMES,
VOCAB_TYPE_NAMES,
ModelFileType,
VocabType,
)
class HFHubRequest: class HFHubRequest:
@ -81,16 +89,11 @@ class HFHubBase:
self.logger = logger self.logger = logger
self._hub = HFHubRequest(auth_token, logger) self._hub = HFHubRequest(auth_token, logger)
self._models = list(HF_MODEL_MAP)
@property @property
def hub(self) -> HFHubRequest: def hub(self) -> HFHubRequest:
return self._hub return self._hub
@property
def models(self) -> list[dict[str, object]]:
return self._models
@property @property
def model_path(self) -> pathlib.Path: def model_path(self) -> pathlib.Path:
return self._model_path return self._model_path
@ -103,45 +106,52 @@ class HFHubBase:
class HFVocabRequest(HFHubBase): class HFVocabRequest(HFHubBase):
def __init__( def __init__(
self, self,
model_path: None | str | pathlib.Path,
auth_token: str, auth_token: str,
model_path: None | str | pathlib.Path,
logger: None | logging.Logger logger: None | logging.Logger
): ):
super().__init__(model_path, auth_token, logger) super().__init__(model_path, auth_token, logger)
@property @property
def tokenizer_type(self) -> LLaMaVocabType: def tokenizer_type(self) -> VocabType:
return LLaMaVocabType return VocabType
def resolve_filenames(self, tokt: LLaMaVocabType) -> tuple[str]: def get_vocab_name(self, vocab_type: VocabType) -> str:
filenames = ["config.json", "tokenizer_config.json", "tokenizer.json"] return VOCAB_TYPE_NAMES.get(vocab_type)
if tokt == self.tokenizer_type.SPM:
filenames.append("tokenizer.model")
return tuple(filenames)
def resolve_tokenizer_model( def get_vocab_enum(self, vocab_name: str) -> VocabType:
self, return {
filename: str, "SPM": VocabType.SPM,
filepath: pathlib.Path, "BPE": VocabType.BPE,
model: dict[str, object] "WPM": VocabType.WPM,
) -> None: }.get(vocab_name, VocabType.NON)
try: # NOTE: Do not use bare exceptions! They mask issues!
resolve_url = self.hub.resolve_url(model['repo'], filename)
response = self.hub.download_file(resolve_url)
self.hub.write_file(response.content, filepath)
except requests.exceptions.HTTPError as e:
self.logger.error(f"Failed to download tokenizer {model['repo']}: {e}")
def download_models(self) -> None: def get_vocab_filenames(self, vocab_type: VocabType) -> tuple[str]:
for model in self.models: if vocab_type == self.tokenizer_type.SPM:
os.makedirs(f"{self.model_path}/{model['repo']}", exist_ok=True) return HF_TOKENIZER_SPM_FILES
filenames = self.resolve_filenames(model['tokt']) # NOTE: WPM and BPE are equivalent
for filename in filenames: return HF_TOKENIZER_BPE_FILES
filepath = pathlib.Path(f"{self.model_path}/{model['repo']}/{filename}")
if filepath.is_file(): def get_vocab_file(
self.logger.info(f"skipped pre-existing tokenizer {model['repo']} in {filepath}") self, model_repo: str, file_name: str, file_path: pathlib.Path,
continue ) -> bool:
self.resolve_tokenizer_model(filename, filepath, model) # NOTE: Do not use bare exceptions! They mask issues!
# Allow the exception to occur or handle it explicitly.
resolve_url = self.hub.resolve_url(model_repo, file_name)
response = self.hub.download_file(resolve_url)
self.hub.write_file(response.content, file_path)
self.logger.info(f"Downloaded tokenizer {file_name} from {model_repo}")
def get_all_vocab_files(self, model_repo: str, vocab_type: VocabType) -> None:
vocab_list = self.get_vocab_filenames(vocab_type)
for vocab_file in vocab_list:
self.get_vocab_file(model_repo, vocab_file, self.model_path)
def extract_normalizer(self) -> dict[str, object]:
pass
def extract_pre_tokenizers(self) -> dict[str, object]:
pass
def generate_checksums(self) -> None: def generate_checksums(self) -> None:
checksums = [] checksums = []
@ -191,5 +201,5 @@ class HFModelRequest(HFHubBase):
super().__init__(model_path, auth_token, logger) super().__init__(model_path, auth_token, logger)
@property @property
def model_type(self) -> LLaMaModelType: def model_type(self) -> ModelFileType:
return LLaMaModelType return ModelFileType