checkpoint_handler

checkpoint.checkpoint_handler

CheckpointHandler class를 정의합니다. CheckpointHandler는 체크포인트를 저장하고 불러오는 역할을 합니다.

CheckpointHandler는 다음과 같은 기능을 제공합니다. 1. 체크포인트 저장 2. 체크포인트 불러오기 3. 체크포인트 변환 4. 체크포인트 유효성 검사

CheckpointHandler는 다음과 같은 메서드를 제공합니다. 1. save: 체크포인트를 저장합니다. 2. load: 체크포인트를 불러옵니다. 3. convert: 이전 버전 체크포인트를 최신 구조로 변환해 저장합니다. 4. load_checkpoint: 체크포인트를 열어 로드합니다. 5. is_valid_checkpoint: 체크포인트가 유효한지 검증하고, 유효하면 True, 그렇지 않으면 False 리턴합니다.

CHECKPOINT_SPEC `module-attribute`

CHECKPOINT_SPEC = {b'2212210': {'version': b'2212210', 'metadata_size': 1024 * 1024, 'weight_position': CHECKPOINT_VERSION_LENGTH + 1024 * 1024}, b'2401040': {'version': b'2401040', 'metadata_size': 12 * 1024 * 1024, 'weight_position': CHECKPOINT_VERSION_LENGTH + 12 * 1024 * 1024}}

CHECKPOINT_LAST_VERSION `module-attribute`

CHECKPOINT_LAST_VERSION = b'2401040'

logger `module-attribute`

logger = getLogger('SaigeResearch')

CheckpointHandler

_keys_encrypt `class-attribute` `instance-attribute`

_keys_encrypt = ['config', 'state_dict']

save `classmethod`

save(checkpoint_path: str, version: str, config: Dict, state_dict: Dict, metadata: Optional[Dict] = None, common: Optional[Dict] = None, password: Optional[str] = None) -> None

Source code in SaigeToolkit/checkpoint/checkpoint_handler.py

@classmethod
def save(
    cls,
    checkpoint_path: str,
    version: str,
    config: Dict,
    state_dict: Dict,
    metadata: Optional[Dict] = None,
    common: Optional[Dict] = None,
    password: Optional[str] = None,
) -> None:
    metadata = metadata or {}
    common = common or {}
    checkpoint = {
        "version": version,
        "config": config,
        "common": common,
        "state_dict": state_dict,
    }
    checkpoint = encrypt_dict(
        data=checkpoint,
        password=password,
        keys_encrypt=cls._keys_encrypt,
    )

    os.makedirs(os.path.dirname(checkpoint_path), exist_ok=True)
    weight_position = CHECKPOINT_SPEC[CHECKPOINT_LAST_VERSION]["weight_position"]
    with open(checkpoint_path, "wb") as f:
        f.write(CHECKPOINT_LAST_VERSION)
        f.seek(weight_position)
        pickle.dump(checkpoint, f)

    write_metadata(checkpoint_path=checkpoint_path, metadata=metadata)

load `classmethod`

load(checkpoint_path: str, password: Optional[str] = None) -> Dict

Source code in SaigeToolkit/checkpoint/checkpoint_handler.py

@classmethod
def load(
    cls,
    checkpoint_path: str,
    password: Optional[str] = None,
) -> Dict:
    if not os.path.isfile(checkpoint_path):
        raise ModelFileNotFoundError

    metadata = read_metadata(checkpoint_path=checkpoint_path)
    if metadata is False:
        logger.info(f"{type(cls).__name__} converting old checkpoint {checkpoint_path}")
        cls.convert(checkpoint_path=checkpoint_path)
        metadata = read_metadata(checkpoint_path=checkpoint_path)

    checkpoint = cls.load_checkpoint(checkpoint_path)

    checkpoint = decrypt_dict(
        data=checkpoint,
        password=password,
        keys_encrypt=cls._keys_encrypt,
        map_location="cpu",
    )
    checkpoint["metadata"] = metadata
    return checkpoint

convert `classmethod`

convert(checkpoint_path: str, save_path: Optional[str] = None)

이전 버전 체크포인트를 최신 구조로 변환해 저장합니다. 저장되어있던 metadata는 제거됩니다.

Parameters:

checkpoint_path (str) –

이전 버전 체크포인트 경로
save_path (str, default: None ) –

저장할 경로. None인 경우 checkpoint_path를 덮어씁니다. Defaults to None.

Source code in SaigeToolkit/checkpoint/checkpoint_handler.py

@classmethod
def convert(cls, checkpoint_path: str, save_path: Optional[str] = None):
    """이전 버전 체크포인트를 최신 구조로 변환해 저장합니다. 저장되어있던 metadata는 제거됩니다.

    Args:
        checkpoint_path (str): 이전 버전 체크포인트 경로
        save_path (str): 저장할 경로. None인 경우 `checkpoint_path`를 덮어씁니다. Defaults to None.
    """
    if save_path is None:
        save_path = checkpoint_path
    checkpoint = decrypt_load_dict(
        path=checkpoint_path,
        password=None,
        map_location="cpu",
    )
    checkpoint.pop("metadata")
    cls.save(checkpoint_path=save_path, **checkpoint)

load_checkpoint `classmethod`

load_checkpoint(checkpoint_path: str) -> Dict

Open and load checkpoint from path

Parameters:

checkpoint_path (str) –

checkpoint path

Raises:

InvalidModelFileError –

다음과 같은 경우 에러 레이즈

Returns:

Dict ( Dict ) –

(일부 암호화된) checkpoint

Source code in SaigeToolkit/checkpoint/checkpoint_handler.py

@classmethod
@handle_unpickling_error
def load_checkpoint(cls, checkpoint_path: str) -> Dict:
    """Open and load checkpoint from path

    Args:
        checkpoint_path (str): checkpoint path

    Raises:
        InvalidModelFileError: 다음과 같은 경우 에러 레이즈
        1) 잘못된 WEIGHT_POSITION이 사용된 경우
        2) 파일이 pickle파일이 아닌 경우
        3) 열린 checkpoint 내부의 형식이 올바르지 않은 경우

    Returns:
        Dict: (일부 암호화된) checkpoint
    """

    checkpoint_version = read_checkpoint_version(checkpoint_path)
    if checkpoint_version not in CHECKPOINT_SPEC:
        raise InvalidModelFileError
    weight_position = CHECKPOINT_SPEC[checkpoint_version]["weight_position"]

    with open(checkpoint_path, "rb") as f:
        f.seek(weight_position)
        checkpoint = pickle.load(f)

    if not cls.is_valid_checkpoint(checkpoint):
        raise InvalidModelFileError

    return checkpoint

is_valid_checkpoint `classmethod`

is_valid_checkpoint(checkpoint: Dict) -> bool

Checkpoint가 유효한지 검증하고, 유효하면 True, 그렇지 않으면 False 리턴

Note

현재 로직은 다음 세가지만을 체크함. 1) checkpoint가 python dictionary 인지 2) checkpoint가 모든 key를 가지고 있는지 3) "version" key에 해당하는 value가 유효한지

Source code in SaigeToolkit/checkpoint/checkpoint_handler.py

@classmethod
def is_valid_checkpoint(cls, checkpoint: Dict) -> bool:
    """Checkpoint가 유효한지 검증하고, 유효하면 True, 그렇지 않으면 False 리턴

    Note:
        현재 로직은 다음 세가지만을 체크함.
        1) checkpoint가 python dictionary 인지
        2) checkpoint가 모든 key를 가지고 있는지
        3) "version" key에 해당하는 value가 유효한지
    """
    if not isinstance(checkpoint, dict):
        return False

    for k in ["version", "config", "common", "state_dict"]:
        if k not in checkpoint:
            return False

    try:
        _ = Version.from_string(checkpoint.get("version"))
    except VersionFormatError:
        return False

    return True

decrypt_dict

decrypt_dict(data, password: Optional[str] = None, keys_encrypt: Optional[List[str]] = None, **torch_load_args) -> Mapping

dict를 복호화합니다.

Parameters:

data (Any) –

description
password (Optional[str], default: None ) –

description. Defaults to None.
keys_encrypt (Optional[List[str]], default: None ) –

None이 아닐 경우 keys_encrypt에 있는 키의 값들만 복호화합니다. Defaults to None.

Raises:

FileNotFoundError –

description

Returns:

Mapping ( Mapping ) –

description

Source code in SaigeToolkit/checkpoint/encryption.py

def decrypt_dict(
    data,
    password: Optional[str] = None,
    keys_encrypt: Optional[List[str]] = None,
    **torch_load_args,
) -> Mapping:
    """dict를 복호화합니다.

    Args:
        data (Any): _description_
        password (Optional[str], optional): _description_. Defaults to None.
        keys_encrypt (Optional[List[str]], optional): None이 아닐 경우 keys_encrypt에 있는 키의 값들만 복호화합니다. Defaults to None.

    Raises:
        FileNotFoundError: _description_

    Returns:
        Mapping: _description_
    """
    if password is not None:
        if keys_encrypt is not None:
            for key in keys_encrypt:
                if key not in data:
                    continue
                data[key] = decrypt(data[key], password, **torch_load_args)
        else:
            data = decrypt(data, password, **torch_load_args)
    return data

decrypt_load_dict

decrypt_load_dict(path: str, password: Optional[str] = None, keys_encrypt: Optional[List[str]] = None, **torch_load_args) -> Mapping

torch.load로 파일 로드 후 복호화해 dict로 로드합니다.

Parameters:

path (str) –

description
password (Optional[str], default: None ) –

description. Defaults to None.
keys_encrypt (Optional[List[str]], default: None ) –

None이 아닐 경우 keys_encrypt에 있는 키의 값들만 복호화합니다. Defaults to None.

Raises:

FileNotFoundError –

description

Returns:

Mapping ( Mapping ) –

description

Source code in SaigeToolkit/checkpoint/encryption.py

@handle_unpickling_error
def decrypt_load_dict(
    path: str,
    password: Optional[str] = None,
    keys_encrypt: Optional[List[str]] = None,
    **torch_load_args,
) -> Mapping:
    """`torch.load`로 파일 로드 후 복호화해 dict로 로드합니다.

    Args:
        path (str): _description_
        password (Optional[str], optional): _description_. Defaults to None.
        keys_encrypt (Optional[List[str]], optional): None이 아닐 경우 keys_encrypt에 있는 키의 값들만 복호화합니다. Defaults to None.

    Raises:
        FileNotFoundError: _description_

    Returns:
        Mapping: _description_
    """
    if not os.path.isfile(path):
        raise error.FileNotFoundError
    data = torch.load(path, **torch_load_args)
    data = decrypt_dict(data, password=password, keys_encrypt=keys_encrypt, **torch_load_args)
    return data

encrypt_dict

encrypt_dict(data: Mapping, password: Optional[str] = None, keys_encrypt: Optional[List[str]] = None, **torch_save_args)

dict 자체 혹은 요소들을 암호화합니다.

Parameters:

data (Mapping) –

description
password (Optional[str], default: None ) –

description. Defaults to None.
keys_encrypt (Optional[List[str]], default: None ) –

None이 아닐 경우 keys_encrypt에 있는 키의 값들만 암호화합니다. Defaults to None.

Source code in SaigeToolkit/checkpoint/encryption.py

def encrypt_dict(
    data: Mapping,
    password: Optional[str] = None,
    keys_encrypt: Optional[List[str]] = None,
    **torch_save_args,
):
    """dict 자체 혹은 요소들을 암호화합니다.

    Args:
        data (Mapping): _description_
        password (Optional[str], optional): _description_. Defaults to None.
        keys_encrypt (Optional[List[str]], optional): None이 아닐 경우 keys_encrypt에 있는 키의 값들만 암호화합니다. Defaults to None.
    """
    data_ = data.copy()
    if password is not None:
        if keys_encrypt is not None:
            for key in keys_encrypt:
                if key not in data_:
                    continue
                data_[key] = encrypt(data_[key], password, **torch_save_args)
        else:
            data_ = encrypt(data_, password, **torch_save_args)
    return data_

read_checkpoint_version

read_checkpoint_version(checkpoint_path: str) -> ByteString

체크포인트 파일의 버전을 읽어옵니다.

Parameters:

checkpoint_path (str) –

checkpoint_path

Returns:

ByteString ( ByteString ) –

체크포인트 파일의 버전

Raises:

ModelFileNotFoundError –

checkpoint_path에 해당하는 파일이 없는 경우

Source code in SaigeToolkit/checkpoint/metadata/metadata.py

@handle_unpickling_error
def read_checkpoint_version(checkpoint_path: str) -> ByteString:
    """체크포인트 파일의 버전을 읽어옵니다.

    Args:
        checkpoint_path (str): checkpoint_path

    Returns:
        ByteString: 체크포인트 파일의 버전

    Raises:
        ModelFileNotFoundError: checkpoint_path에 해당하는 파일이 없는 경우
    """
    if not os.path.isfile(checkpoint_path):
        raise ModelFileNotFoundError

    with open(checkpoint_path, "rb") as f:
        return f.read(CHECKPOINT_VERSION_LENGTH)

read_metadata

read_metadata(checkpoint_path: str) -> Union[Dict, bool]

체크포인트 파일의 metadata 섹션 데이터를 읽어옵니다.

Parameters:

checkpoint_path (str) –

checkpoint_path

Returns:

Union[Dict, bool] –

Union[Dict, bool]: 체크포인트 버전이 맞지 않는 경우 False 값을 리턴

Raises:

ModelFileNotFoundError –

checkpoint_path에 해당하는 파일이 없는 경우

Source code in SaigeToolkit/checkpoint/metadata/metadata.py

@handle_unpickling_error
def read_metadata(checkpoint_path: str) -> Union[Dict, bool]:
    """체크포인트 파일의 metadata 섹션 데이터를 읽어옵니다.

    Args:
        checkpoint_path (str): checkpoint_path

    Returns:
        Union[Dict, bool]: 체크포인트 버전이 맞지 않는 경우 False 값을 리턴

    Raises:
        ModelFileNotFoundError: checkpoint_path에 해당하는 파일이 없는 경우
    """
    if not os.path.isfile(checkpoint_path):
        raise ModelFileNotFoundError
    with open(checkpoint_path, "rb") as f:
        version = f.read(CHECKPOINT_VERSION_LENGTH)
        return False if version not in CHECKPOINT_SPEC else pickle.load(f)

write_metadata

write_metadata(checkpoint_path: str, metadata: Dict) -> bool

체크포인트 파일의 metadata 섹션 데이터를 수정해 저장합니다.

Parameters:

checkpoint_path (str) –

checkpoint_path
metadata (Dict) –

metadata

Returns:

bool ( bool ) –

체크포인트 버전이 맞지 않는 경우 False 값을 리턴

Raises:

ModelFileNotFoundError –

checkpoint_path에 해당하는 파일이 없는 경우

Source code in SaigeToolkit/checkpoint/metadata/metadata.py

def write_metadata(checkpoint_path: str, metadata: Dict) -> bool:
    """체크포인트 파일의 metadata 섹션 데이터를 수정해 저장합니다.

    Args:
        checkpoint_path (str): checkpoint_path
        metadata (Dict): metadata

    Returns:
        bool: 체크포인트 버전이 맞지 않는 경우 False 값을 리턴

    Raises:
        ModelFileNotFoundError: checkpoint_path에 해당하는 파일이 없는 경우
    """
    if not os.path.isfile(checkpoint_path):
        raise ModelFileNotFoundError
    with open(checkpoint_path, "r+b") as f:
        version = f.read(CHECKPOINT_VERSION_LENGTH)
        if version not in CHECKPOINT_SPEC:
            return False
        bytes = pickle.dumps(metadata)
        metadata_size = CHECKPOINT_SPEC[version]["metadata_size"]
        if len(bytes) >= metadata_size:
            raise ValueError("metadata size exceeds the limit")
        f.write(bytes)
        return True

handle_unpickling_error

handle_unpickling_error(function)

Source code in SaigeToolkit/checkpoint/metadata/metadata.py

def handle_unpickling_error(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        try:
            return function(*args, **kwargs)
        except pickle.UnpicklingError:
            raise InvalidModelFileError

    return wrapper

convert_checkpoint

convert_checkpoint(checkpoint_path: str, save_path: Optional[str] = None)

22년12월21일 이전 버전 체크포인트를 최신 구조로 변환해 저장합니다. 저장되어있던 metadata는 제거됩니다.

Parameters:

checkpoint_path (str) –

이전 버전 체크포인트 경로
save_path (Optional[str], default: None ) –

저장할 경로. None인 경우 checkpoint_path를 덮어씁니다. Defaults to None.

Source code in SaigeToolkit/checkpoint/checkpoint_handler.py

def convert_checkpoint(checkpoint_path: str, save_path: Optional[str] = None):
    """22년12월21일 이전 버전 체크포인트를 최신 구조로 변환해 저장합니다. 저장되어있던 metadata는 제거됩니다.

    Args:
        checkpoint_path (str): 이전 버전 체크포인트 경로
        save_path (Optional[str], optional): 저장할 경로. None인 경우 `checkpoint_path`를 덮어씁니다. Defaults to None.
    """
    CheckpointHandler.convert(checkpoint_path=checkpoint_path, save_path=save_path)

checkpoint_handler

checkpoint.checkpoint_handler

CHECKPOINT_SPEC module-attribute

CHECKPOINT_LAST_VERSION module-attribute

logger module-attribute

CheckpointHandler

_keys_encrypt class-attribute instance-attribute

save classmethod

load classmethod

convert classmethod

load_checkpoint classmethod

is_valid_checkpoint classmethod

decrypt_dict

decrypt_load_dict

encrypt_dict

read_checkpoint_version

read_metadata

write_metadata

handle_unpickling_error

convert_checkpoint

CHECKPOINT_SPEC `module-attribute`

CHECKPOINT_LAST_VERSION `module-attribute`

logger `module-attribute`

_keys_encrypt `class-attribute` `instance-attribute`

save `classmethod`

load `classmethod`

convert `classmethod`

load_checkpoint `classmethod`

is_valid_checkpoint `classmethod`