|
3 | 3 |
|
4 | 4 | import itertools |
5 | 5 | from abc import abstractmethod |
| 6 | +from collections.abc import Iterable |
| 7 | +from typing import Any |
6 | 8 |
|
7 | 9 | import torch |
8 | 10 | from torch.nn.parameter import Parameter |
| 11 | +from typing_extensions import TypeIs |
9 | 12 |
|
10 | 13 | import vllm.envs as envs |
11 | 14 | from vllm.distributed import ( |
@@ -632,31 +635,31 @@ def __init__( |
632 | 635 | disable_tp=disable_tp, |
633 | 636 | ) |
634 | 637 |
|
635 | | - def validate_shard_id(self, loaded_shard_id: int | tuple[int, ...] | None): |
636 | | - if loaded_shard_id is None: |
637 | | - return |
638 | | - if isinstance(loaded_shard_id, tuple): |
639 | | - for idx in loaded_shard_id: |
| 638 | + def validate_shard_id(self, shard_id: Any) -> TypeIs[int | tuple[int, ...] | None]: |
| 639 | + if isinstance(shard_id, int): |
| 640 | + if shard_id < 0 or shard_id >= len(self.output_sizes): |
| 641 | + raise ValueError( |
| 642 | + f"Shard id should be between 0 and {len(self.output_sizes) - 1}. " |
| 643 | + f"Got shard id {shard_id}." |
| 644 | + ) |
| 645 | + return True |
| 646 | + if shard_id is None: |
| 647 | + return True |
| 648 | + if isinstance(shard_id, tuple): |
| 649 | + for idx in shard_id: |
640 | 650 | if not (0 <= idx < len(self.output_sizes)): |
641 | 651 | raise ValueError( |
642 | 652 | f"Shard id index {idx} should be between 0 and " |
643 | | - f"{len(self.output_sizes) - 1}. Got shard id {loaded_shard_id}." |
| 653 | + f"{len(self.output_sizes) - 1}. Got shard id {shard_id}." |
644 | 654 | ) |
645 | | - if len(loaded_shard_id) > 1 and any( |
646 | | - b - a != 1 for a, b in zip(loaded_shard_id[:-1], loaded_shard_id[1:]) |
| 655 | + if len(shard_id) > 1 and any( |
| 656 | + b - a != 1 for a, b in zip(shard_id[:-1], shard_id[1:]) |
647 | 657 | ): |
648 | 658 | raise ValueError( |
649 | 659 | "Shard id with multiple indices should be consecutive. " |
650 | | - f"Got shard id {loaded_shard_id}." |
| 660 | + f"Got shard id {shard_id}." |
651 | 661 | ) |
652 | | - return |
653 | | - elif isinstance(loaded_shard_id, int): |
654 | | - if loaded_shard_id < 0 or loaded_shard_id >= len(self.output_sizes): |
655 | | - raise ValueError( |
656 | | - f"Shard id should be between 0 and {len(self.output_sizes) - 1}. " |
657 | | - f"Got shard id {loaded_shard_id}." |
658 | | - ) |
659 | | - return |
| 662 | + return True |
660 | 663 | raise ValueError("This line should not be reached") |
661 | 664 |
|
662 | 665 | def weight_loader( |
@@ -910,6 +913,31 @@ def weight_loader_v2( |
910 | 913 | tp_rank=self.tp_rank, |
911 | 914 | ) |
912 | 915 |
|
| 916 | + def load_weights( |
| 917 | + self, weights: Iterable[tuple[str, torch.Tensor]] |
| 918 | + ) -> Iterable[str]: |
| 919 | + for name, loaded_weight in weights: |
| 920 | + shard_id = getattr(loaded_weight, "shard_id", None) |
| 921 | + self.validate_shard_id(shard_id) |
| 922 | + # Load into self if name is not an attr of self or its submodules |
| 923 | + param: Parameter |
| 924 | + if "." in name: |
| 925 | + submodule, _, attr = name.rpartition(".") |
| 926 | + param = getattr(self.get_submodule(submodule), attr, self) |
| 927 | + else: |
| 928 | + param = getattr(self, name, self) |
| 929 | + if param is None and name == "bias": |
| 930 | + continue |
| 931 | + param.weight_loader(param, loaded_weight, shard_id) |
| 932 | + logger.debug( |
| 933 | + "Loaded shard %s with shape %s into %s.%s", |
| 934 | + shard_id, |
| 935 | + loaded_weight.shape, |
| 936 | + self.prefix, |
| 937 | + name, |
| 938 | + ) |
| 939 | + yield name |
| 940 | + |
913 | 941 |
|
914 | 942 | class QKVParallelLinear(ColumnParallelLinear): |
915 | 943 | """Linear layers for the attention's QKV transformation. |
@@ -996,17 +1024,13 @@ def __init__( |
996 | 1024 | disable_tp=disable_tp, |
997 | 1025 | ) |
998 | 1026 |
|
999 | | - def validate_shard_id(self, loaded_shard_id: str | None): |
1000 | | - if loaded_shard_id is None: |
1001 | | - return |
1002 | | - if isinstance(loaded_shard_id, str): |
1003 | | - if loaded_shard_id not in ["q", "k", "v"]: |
1004 | | - raise ValueError( |
1005 | | - "Shard id for QKVParallelLinear should be 'q', 'k', or 'v', " |
1006 | | - f"got shard id {loaded_shard_id}." |
1007 | | - ) |
1008 | | - return |
1009 | | - raise ValueError("This line should not be reached") |
| 1027 | + def validate_shard_id(self, shard_id: Any) -> TypeIs[str | None]: |
| 1028 | + if shard_id in {"q", "k", "v"} or shard_id is None: |
| 1029 | + return True |
| 1030 | + raise ValueError( |
| 1031 | + "Shard id for QKVParallelLinear should be 'q', 'k', or 'v', " |
| 1032 | + f"got shard id {shard_id}." |
| 1033 | + ) |
1010 | 1034 |
|
1011 | 1035 | def _get_shard_offset_mapping(self, loaded_shard_id: str): |
1012 | 1036 | shard_offset_mapping = { |
@@ -1302,6 +1326,31 @@ def weight_loader( |
1302 | 1326 | assert param_data.shape == loaded_weight.shape |
1303 | 1327 | param_data.copy_(loaded_weight) |
1304 | 1328 |
|
| 1329 | + def load_weights( |
| 1330 | + self, weights: Iterable[tuple[str, torch.Tensor]] |
| 1331 | + ) -> Iterable[str]: |
| 1332 | + for name, loaded_weight in weights: |
| 1333 | + shard_id = getattr(loaded_weight, "shard_id", None) |
| 1334 | + self.validate_shard_id(shard_id) |
| 1335 | + # Load into self if name is not an attr of self or its submodules |
| 1336 | + param: Parameter |
| 1337 | + if "." in name: |
| 1338 | + submodule, _, attr = name.rpartition(".") |
| 1339 | + param = getattr(self.get_submodule(submodule), attr, self) |
| 1340 | + else: |
| 1341 | + param = getattr(self, name, self) |
| 1342 | + if param is None and name == "bias": |
| 1343 | + continue |
| 1344 | + param.weight_loader(param, loaded_weight, shard_id) |
| 1345 | + logger.debug( |
| 1346 | + "Loaded shard %s with shape %s into %s.%s", |
| 1347 | + shard_id, |
| 1348 | + loaded_weight.shape, |
| 1349 | + self.prefix, |
| 1350 | + name, |
| 1351 | + ) |
| 1352 | + yield name |
| 1353 | + |
1305 | 1354 |
|
1306 | 1355 | class MinimaxM3QKVParallelLinearWithIndexer(QKVParallelLinear): |
1307 | 1356 | """QKV projection fused with a lightning-indexer's index_q/index_k. |
@@ -1387,15 +1436,14 @@ def __init__( |
1387 | 1436 | prefix=prefix, |
1388 | 1437 | ) |
1389 | 1438 |
|
1390 | | - def validate_shard_id(self, loaded_shard_id: str | None) -> None: |
1391 | | - if loaded_shard_id is None: |
1392 | | - return |
1393 | | - if loaded_shard_id not in ("q", "k", "v", "index_q", "index_k"): |
1394 | | - raise ValueError( |
1395 | | - "Shard id for MinimaxM3QKVParallelLinearWithIndexer must be one of " |
1396 | | - "'q', 'k', 'v', 'index_q', 'index_k'; got " |
1397 | | - f"{loaded_shard_id}." |
1398 | | - ) |
| 1439 | + def validate_shard_id(self, shard_id: Any) -> TypeIs[str | None]: |
| 1440 | + if shard_id in {"q", "k", "v", "index_q", "index_k"} or shard_id is None: |
| 1441 | + return True |
| 1442 | + raise ValueError( |
| 1443 | + "Shard id for MinimaxM3QKVParallelLinearWithIndexer must be one of " |
| 1444 | + "'q', 'k', 'v', 'index_q', 'index_k'; got " |
| 1445 | + f"{shard_id}." |
| 1446 | + ) |
1399 | 1447 |
|
1400 | 1448 | def _get_shard_offset_mapping(self, loaded_shard_id: str) -> int | None: |
1401 | 1449 | h = self.head_size |
|
0 commit comments