Skip to content

api

checkpoint.api

error_handler module-attribute 

error_handler = get_error_handler(SaigeToolkitError)

convert_checkpoint 

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

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

이 API는 추후 백엔드와의 논의를 통해 점진적으로 사용을 줄이고, 최종적으로는 삭제되어야 합니다. 현재는 torch와의 의존성을 줄이기 위해, 실제 호출 시에만 관련 패키지가 로드되도록 lazy import가 적용되어 있습니다.

Parameters:

  • checkpoint_path (str) –

    이전 버전 체크포인트 경로

  • save_path (Optional[str], default: None ) –

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

Source code in SaigeToolkit/checkpoint/api.py 
21
22
23
24
25
26
27
28
29
30
31
32
@error_handler
def convert_checkpoint(checkpoint_path: str, save_path: Optional[str] = None):
    """22년12월21일 이전 버전 체크포인트를 최신 구조로 변환해 저장합니다. 저장되어있던 metadata는 제거됩니다.

    이 API는 추후 백엔드와의 논의를 통해 점진적으로 사용을 줄이고, 최종적으로는 삭제되어야 합니다.
    현재는 torch와의 의존성을 줄이기 위해, 실제 호출 시에만 관련 패키지가 로드되도록 lazy import가 적용되어 있습니다.

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

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 리턴합니다.

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 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@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 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@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 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@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 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@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

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 
174
175
176
177
178
179
180
181
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)