File size: 2,851 Bytes
a164e13 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 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 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
# Copyright (C) 2024 Charles O. Goddard
#
# This software is free software: you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This software is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
from abc import ABC, abstractmethod
from typing import Dict, Optional, Sequence
import safetensors
import torch
from mergekit.io.lazy_unpickle import DeferredLoad, TorchArchiveReader, torch_lazy_load
class TensorLoader(ABC):
"""Base class for (potentially lazy) tensor loaders."""
@abstractmethod
def get_tensor(self, key: str) -> torch.Tensor:
...
@abstractmethod
def keys(self) -> Sequence[str]:
...
@classmethod
def get(
cls,
shard_path: str,
use_lazy_unpickle: bool = False,
device: Optional[str] = None,
) -> "TensorLoader":
if shard_path.lower().endswith(".safetensors"):
# not a subclass of TensorLoader, but exposes same api
return safetensors.safe_open(
shard_path, framework="pt", device=device or "cpu"
)
elif use_lazy_unpickle:
return LazyPickleLoader(shard_path, device=device)
return DumbPytorchLoader(shard_path, device=device)
class LazyPickleLoader(TensorLoader):
"""Loader for pytorch files using a custom unpickler and vigorous monkeypatching."""
zip_reader: TorchArchiveReader
index: Dict[str, DeferredLoad]
device: Optional[str] = None
def __init__(self, path: str, device: Optional[str] = None):
self.zip_reader = TorchArchiveReader(path)
self.device = device
with torch_lazy_load():
self.index = torch.load(path)
def get_tensor(self, key: str) -> torch.Tensor:
if key not in self.index:
raise KeyError(key)
return self.index[key].execute(self.zip_reader, map_location=self.device)
def keys(self) -> Sequence[str]:
return self.index.keys()
class DumbPytorchLoader(TensorLoader):
"""Naive `torch.load` shard loading."""
tensors: Dict[str, torch.Tensor]
def __init__(self, path: str, device: str):
self.tensors = torch.load(path, map_location=device, weights_only=True)
def get_tensor(self, key: str) -> torch.Tensor:
return self.tensors[key]
def keys(self) -> Sequence[str]:
return self.tensors.keys()
|