From 985b2c8eac0a82ca3c76f38560ca223dfc178418 Mon Sep 17 00:00:00 2001 From: null Date: Wed, 19 May 2021 20:24:00 +0800 Subject: [PATCH] Complete fixing Layer-wise Sampling (LADIES) and Node-wise Sampling --- autogl/module/hpo/base.py | 18 +- autogl/module/model/gcn.py | 17 +- autogl/module/train/evaluation.py | 16 + .../node_classification_sampled_trainer.py | 571 +++++------------- .../layer_dependent_importance_sampler.py | 95 +-- .../sampling/sampler/neighbor_sampler.py | 45 +- .../sampler/target_dependant_sampler.py | 1 + configs/nodeclf_ladies_gcn.yml | 61 +- 8 files changed, 267 insertions(+), 557 deletions(-) diff --git a/autogl/module/hpo/base.py b/autogl/module/hpo/base.py index 1e666dd..fdef8a2 100644 --- a/autogl/module/hpo/base.py +++ b/autogl/module/hpo/base.py @@ -30,7 +30,7 @@ class BaseHPOptimizer: raise WrongDependedParameterError("The depended parameter does not exist.") for para in config: - if para["type"] == "NUMERICAL_LIST" and para.get("cutPara", None): + if para["type"] in ("NUMERICAL_LIST", "CATEGORICAL_LIST") and para.get("cutPara", None): self._depend_map[para["parameterName"]] = para if type(para["cutPara"]) == str: get_depended_para(para["cutPara"]) @@ -76,6 +76,18 @@ class BaseHPOptimizer: new_para["maxValue"] = y new_para["scalingType"] = para["scalingType"] fin.append(new_para) + elif para["type"] == "CATEGORICAL_LIST": + self._list_map[para["parameterName"]] = para["length"] + category = para["feasiblePoints"] + self._category_map[para["parameterName"]] = category + + cur_points = ",".join(map(lambda _x: str(_x), range(len(category)))) + for i in range(para["length"]): + new_para = dict() + new_para["parameterName"] = para["parameterName"] + "_" + str(i) + new_para["type"] = "DISCRETE" + new_para["feasiblePoints"] = cur_points + fin.append(new_para) elif para["type"] == "FIXED": self._fix_map[para["parameterName"]] = para["value"] else: @@ -92,6 +104,8 @@ class BaseHPOptimizer: for i in range(self._list_map[pname]): val.append(config[pname + "_" + str(i)]) del config[pname + "_" + str(i)] + if pname in self._category_map: + val = [self._category_map[pname][i] for i in val] fin[pname] = val # deal other para for pname in config: @@ -123,10 +137,10 @@ class BaseHPOptimizer: "maxValue": 0.9, "scalingType": "LINEAR" }]""" - config = self._decompose_list_fixed_para(config) self._category_map = {} self._discrete_map = {} self._numerical_map = {} + config = self._decompose_list_fixed_para(config) current_config = [] for para in config: diff --git a/autogl/module/model/gcn.py b/autogl/module/model/gcn.py index dbe0f8a..7d2d4e1 100644 --- a/autogl/module/model/gcn.py +++ b/autogl/module/model/gcn.py @@ -17,7 +17,8 @@ class GCN(torch.nn.Module): hidden_features: _typing.Sequence[int], dropout: float, activation_name: str, - add_self_loops: bool = True + add_self_loops: bool = True, + normalize: bool = True ): super().__init__() self.__convolution_layers: torch.nn.ModuleList = torch.nn.ModuleList() @@ -25,13 +26,17 @@ class GCN(torch.nn.Module): if num_layers == 1: self.__convolution_layers.append( torch_geometric.nn.GCNConv( - num_features, num_classes, add_self_loops=add_self_loops + num_features, num_classes, + add_self_loops=add_self_loops, + normalize=normalize ) ) else: self.__convolution_layers.append( torch_geometric.nn.GCNConv( - num_features, hidden_features[0], add_self_loops=add_self_loops + num_features, hidden_features[0], + add_self_loops=add_self_loops, + normalize=normalize ) ) for i in range(len(hidden_features)): @@ -221,8 +226,6 @@ class AutoGCN(ClassificationModel): self.hyper_parameter.get("hidden"), self.hyper_parameter.get("dropout"), self.hyper_parameter.get("act"), - add_self_loops=( - "add_self_loops" in self.hyper_parameter - and self.hyper_parameter.get("add_self_loops") - ) + add_self_loops=bool(self.hyper_parameter.get("add_self_loops", True)), + normalize=bool(self.hyper_parameter.get("normalize", True)) ).to(self.device) diff --git a/autogl/module/train/evaluation.py b/autogl/module/train/evaluation.py index b0d25cb..1ebb324 100644 --- a/autogl/module/train/evaluation.py +++ b/autogl/module/train/evaluation.py @@ -1,6 +1,7 @@ import numpy as np import typing as _typing from sklearn.metrics import ( + f1_score, log_loss, accuracy_score, roc_auc_score, @@ -221,3 +222,18 @@ class Mrr(Evaluation): """ pos_predict = predict[:, 1] return label_ranking_average_precision_score(label, pos_predict) + + +@register_evaluate("MicroF1") +class MicroF1(Evaluation): + @staticmethod + def get_eval_name() -> str: + return "MicroF1" + + @staticmethod + def is_higher_better() -> bool: + return True + + @staticmethod + def evaluate(predict, label) -> float: + return f1_score(label, np.argmax(predict, axis=1), average='micro') diff --git a/autogl/module/train/node_classification_trainer/node_classification_sampled_trainer.py b/autogl/module/train/node_classification_trainer/node_classification_sampled_trainer.py index 6894d42..181c5b3 100644 --- a/autogl/module/train/node_classification_trainer/node_classification_sampled_trainer.py +++ b/autogl/module/train/node_classification_trainer/node_classification_sampled_trainer.py @@ -9,7 +9,7 @@ import tqdm import autogl.data from .. import register_trainer from ..base import BaseNodeClassificationTrainer, EarlyStopping, Evaluation -from ..evaluation import get_feval, Logloss, EvaluatorUtility +from ..evaluation import get_feval, EvaluatorUtility, Logloss, MicroF1 from ..sampling.sampler.target_dependant_sampler import TargetDependantSampledData from ..sampling.sampler.neighbor_sampler import NeighborSampler from ..sampling.sampler.graphsaint_sampler import * @@ -21,351 +21,6 @@ from ...model import BaseModel LOGGER: logging.Logger = logging.getLogger("Node classification sampling trainer") -# @register_trainer("NodeClassificationNeighborSampling") -# class NodeClassificationNeighborSamplingTrainer(BaseNodeClassificationTrainer): -# """ -# The node classification trainer -# for automatically training the node classification tasks -# with neighbour sampling -# """ -# -# def __init__( -# self, -# model: _typing.Union[BaseModel, str], -# num_features: int, -# num_classes: int, -# optimizer: _typing.Union[_typing.Type[torch.optim.Optimizer], str, None] = None, -# lr: float = 1e-4, -# max_epoch: int = 100, -# early_stopping_round: int = 100, -# weight_decay: float = 1e-4, -# device: _typing.Optional[torch.device] = None, -# init: bool = True, -# feval: _typing.Union[ -# _typing.Sequence[str], _typing.Sequence[_typing.Type[Evaluation]] -# ] = (Logloss,), -# loss: str = "nll_loss", -# lr_scheduler_type: _typing.Optional[str] = None, -# **kwargs, -# ) -> None: -# if isinstance(optimizer, type) and issubclass(optimizer, torch.optim.Optimizer): -# self._optimizer_class: _typing.Type[torch.optim.Optimizer] = optimizer -# elif type(optimizer) == str: -# if optimizer.lower() == "adam": -# self._optimizer_class: _typing.Type[ -# torch.optim.Optimizer -# ] = torch.optim.Adam -# elif optimizer.lower() == "adam" + "w": -# self._optimizer_class: _typing.Type[ -# torch.optim.Optimizer -# ] = torch.optim.AdamW -# elif optimizer.lower() == "sgd": -# self._optimizer_class: _typing.Type[ -# torch.optim.Optimizer -# ] = torch.optim.SGD -# else: -# self._optimizer_class: _typing.Type[ -# torch.optim.Optimizer -# ] = torch.optim.Adam -# else: -# self._optimizer_class: _typing.Type[ -# torch.optim.Optimizer -# ] = torch.optim.Adam -# -# self._learning_rate: float = lr if lr > 0 else 1e-4 -# self._lr_scheduler_type: _typing.Optional[str] = lr_scheduler_type -# self._max_epoch: int = max_epoch if max_epoch > 0 else 1e2 -# -# self.__sampling_sizes: _typing.Sequence[int] = kwargs.get("sampling_sizes") -# -# self._weight_decay: float = weight_decay if weight_decay > 0 else 1e-4 -# early_stopping_round: int = ( -# early_stopping_round if early_stopping_round > 0 else 1e2 -# ) -# self._early_stopping = EarlyStopping( -# patience=early_stopping_round, verbose=False -# ) -# super(NodeClassificationNeighborSamplingTrainer, self).__init__( -# model, num_features, num_classes, device, init, feval, loss -# ) -# -# self._valid_result: torch.Tensor = torch.zeros(0) -# self._valid_result_prob: torch.Tensor = torch.zeros(0) -# self._valid_score: _typing.Sequence[float] = [] -# -# self._hyper_parameter_space: _typing.Sequence[ -# _typing.Dict[str, _typing.Any] -# ] = [] -# -# self.__initialized: bool = False -# if init: -# self.initialize() -# -# def initialize(self) -> "NodeClassificationNeighborSamplingTrainer": -# if self.__initialized: -# return self -# self.model.initialize() -# self.__initialized = True -# return self -# -# def get_model(self) -> BaseModel: -# return self.model -# -# def __train_only(self, data) -> "NodeClassificationNeighborSamplingTrainer": -# """ -# The function of training on the given dataset and mask. -# :param data: data of a specific graph -# :return: self -# """ -# data = data.to(self.device) -# optimizer: torch.optim.Optimizer = self._optimizer_class( -# self.model.model.parameters(), -# lr=self._learning_rate, -# weight_decay=self._weight_decay, -# ) -# if type(self._lr_scheduler_type) == str: -# if self._lr_scheduler_type.lower() == "step" + "lr": -# lr_scheduler: torch.optim.lr_scheduler.StepLR = ( -# torch.optim.lr_scheduler.StepLR(optimizer, step_size=100, gamma=0.1) -# ) -# elif self._lr_scheduler_type.lower() == "multi" + "step" + "lr": -# lr_scheduler: torch.optim.lr_scheduler.MultiStepLR = ( -# torch.optim.lr_scheduler.MultiStepLR( -# optimizer, milestones=[30, 80], gamma=0.1 -# ) -# ) -# elif self._lr_scheduler_type.lower() == "exponential" + "lr": -# lr_scheduler: torch.optim.lr_scheduler.ExponentialLR = ( -# torch.optim.lr_scheduler.ExponentialLR(optimizer, gamma=0.1) -# ) -# elif self._lr_scheduler_type.lower() == "ReduceLROnPlateau".lower(): -# lr_scheduler: torch.optim.lr_scheduler.ReduceLROnPlateau = ( -# torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, "min") -# ) -# else: -# lr_scheduler: torch.optim.lr_scheduler.LambdaLR = ( -# torch.optim.lr_scheduler.LambdaLR(optimizer, lambda _: 1.0) -# ) -# else: -# lr_scheduler: torch.optim.lr_scheduler.LambdaLR = ( -# torch.optim.lr_scheduler.LambdaLR(optimizer, lambda _: 1.0) -# ) -# -# train_sampler: NeighborSampler = NeighborSampler( -# data, self.__sampling_sizes, batch_size=20 -# ) -# -# for current_epoch in range(self._max_epoch): -# self.model.model.train() -# """ epoch start """ -# for target_node_indexes, edge_indexes in train_sampler: -# optimizer.zero_grad() -# data.edge_indexes = edge_indexes -# prediction = self.model.model(data) -# if not hasattr(torch.nn.functional, self.loss): -# raise TypeError( -# "PyTorch does not support loss type {}".format(self.loss) -# ) -# loss_function = getattr(torch.nn.functional, self.loss) -# loss: torch.Tensor = loss_function( -# prediction[target_node_indexes], data.y[target_node_indexes] -# ) -# loss.backward() -# optimizer.step() -# -# if lr_scheduler is not None: -# lr_scheduler.step() -# -# """ Validate performance """ -# if hasattr(data, "val_mask") and getattr(data, "val_mask") is not None: -# validation_results: _typing.Sequence[float] = self.evaluate( -# (data,), "val", [self.feval[0]] -# ) -# -# if self.feval[0].is_higher_better(): -# validation_loss: float = -validation_results[0] -# else: -# validation_loss: float = validation_results[0] -# self._early_stopping(validation_loss, self.model.model) -# if self._early_stopping.early_stop: -# LOGGER.debug("Early stopping at %d", current_epoch) -# break -# if hasattr(data, "val_mask") and data.val_mask is not None: -# self._early_stopping.load_checkpoint(self.model.model) -# return self -# -# def __predict_only(self, data): -# """ -# The function of predicting on the given data. -# :param data: data of a specific graph -# :return: the result of prediction on the given dataset -# """ -# data = data.to(self.device) -# self.model.model.eval() -# with torch.no_grad(): -# prediction = self.model.model(data) -# return prediction -# -# def train(self, dataset, keep_valid_result: bool = True): -# """ -# The function of training on the given dataset and keeping valid result. -# :param dataset: -# :param keep_valid_result: Whether to save the validation result after training -# """ -# data = dataset[0] -# self.__train_only(data) -# if keep_valid_result: -# prediction: torch.Tensor = self.__predict_only(data) -# self._valid_result: torch.Tensor = prediction[data.val_mask].max(1)[1] -# self._valid_result_prob: torch.Tensor = prediction[data.val_mask] -# self._valid_score = self.evaluate(dataset, "val") -# -# def predict_proba( -# self, dataset, mask: _typing.Optional[str] = None, in_log_format: bool = False -# ) -> torch.Tensor: -# """ -# The function of predicting the probability on the given dataset. -# :param dataset: The node classification dataset used to be predicted. -# :param mask: -# :param in_log_format: -# :return: -# """ -# data = dataset[0].to(self.device) -# if mask is not None and type(mask) == str: -# if mask.lower() == "train": -# _mask = data.train_mask -# elif mask.lower() == "test": -# _mask = data.test_mask -# elif mask.lower() == "val": -# _mask = data.val_mask -# else: -# _mask = data.test_mask -# else: -# _mask = data.test_mask -# result = self.__predict_only(data)[_mask] -# return result if in_log_format else torch.exp(result) -# -# def predict(self, dataset, mask: _typing.Optional[str] = None) -> torch.Tensor: -# return self.predict_proba(dataset, mask, in_log_format=True).max(1)[1] -# -# def get_valid_predict(self) -> torch.Tensor: -# return self._valid_result -# -# def get_valid_predict_proba(self) -> torch.Tensor: -# return self._valid_result_prob -# -# def get_valid_score(self, return_major: bool = True): -# if return_major: -# return (self._valid_score[0], self.feval[0].is_higher_better()) -# else: -# return (self._valid_score, [f.is_higher_better() for f in self.feval]) -# -# def get_name_with_hp(self) -> str: -# name = "-".join( -# [ -# str(self._optimizer_class), -# str(self._learning_rate), -# str(self._max_epoch), -# str(self._early_stopping.patience), -# str(self.model), -# str(self.device), -# ] -# ) -# name = ( -# name -# + "|" -# + "-".join( -# [ -# str(x[0]) + "-" + str(x[1]) -# for x in self.model.get_hyper_parameter().items() -# ] -# ) -# ) -# return name -# -# def evaluate( -# self, -# dataset, -# mask: _typing.Optional[str] = None, -# feval: _typing.Union[ -# None, _typing.Sequence[str], _typing.Sequence[_typing.Type[Evaluation]] -# ] = None, -# ) -> _typing.Sequence[float]: -# data = dataset[0] -# data = data.to(self.device) -# if feval is None: -# _feval: _typing.Sequence[_typing.Type[Evaluation]] = self.feval -# else: -# _feval: _typing.Sequence[_typing.Type[Evaluation]] = get_feval(list(feval)) -# if mask.lower() == "train": -# _mask = data.train_mask -# elif mask.lower() == "test": -# _mask = data.test_mask -# elif mask.lower() == "val": -# _mask = data.val_mask -# else: -# _mask = data.test_mask -# prediction_probability: torch.Tensor = self.predict_proba(dataset, mask) -# y_ground_truth = data.y[_mask] -# -# results = [] -# for f in _feval: -# try: -# results.append(f.evaluate(prediction_probability, y_ground_truth)) -# except: -# results.append( -# f.evaluate( -# prediction_probability.cpu().numpy(), -# y_ground_truth.cpu().numpy(), -# ) -# ) -# return results -# -# def to(self, device: torch.device): -# self.device = device -# if self.model is not None: -# self.model.to(self.device) -# -# def duplicate_from_hyper_parameter( -# self, -# hp: _typing.Dict[str, _typing.Any], -# model: _typing.Union[BaseModel, str, None] = None, -# ) -> "NodeClassificationNeighborSamplingTrainer": -# -# if model is None or not isinstance(model, BaseModel): -# model = self.model -# model = model.from_hyper_parameter( -# dict( -# [ -# x -# for x in hp.items() -# if x[0] in [y["parameterName"] for y in model.hyper_parameter_space] -# ] -# ) -# ) -# -# return NodeClassificationNeighborSamplingTrainer( -# model, -# self.num_features, -# self.num_classes, -# self._optimizer_class, -# device=self.device, -# init=True, -# feval=self.feval, -# loss=self.loss, -# lr_scheduler_type=self._lr_scheduler_type, -# **hp, -# ) -# -# @property -# def hyper_parameter_space(self): -# return self._hyper_parameter_space -# -# @hyper_parameter_space.setter -# def hyper_parameter_space(self, hp_space): -# self._hyper_parameter_space = hp_space - - @register_trainer("NodeClassificationGraphSAINTTrainer") class NodeClassificationGraphSAINTTrainer(BaseNodeClassificationTrainer): def __init__( @@ -772,7 +427,7 @@ class NodeClassificationLayerDependentImportanceSamplingTrainer(BaseNodeClassifi init: bool = True, feval: _typing.Union[ _typing.Sequence[str], _typing.Sequence[_typing.Type[Evaluation]] - ] = (Logloss,), + ] = (MicroF1,), loss: str = "nll_loss", lr_scheduler_type: _typing.Optional[str] = None, **kwargs, @@ -815,17 +470,31 @@ class NodeClassificationLayerDependentImportanceSamplingTrainer(BaseNodeClassifi self._valid_result_prob: torch.Tensor = torch.zeros(0) self._valid_score: _typing.Sequence[float] = () + self.__training_batch_size: int = kwargs.get("training_batch_size", 1024) + if not self.__training_batch_size > 0: + self.__training_batch_size: int = 1024 + self.__predicting_batch_size: int = kwargs.get("predicting_batch_size", 1024) + if not self.__predicting_batch_size > 0: + self.__predicting_batch_size: int = 1024 + + cpu_count: int = os.cpu_count() if os.cpu_count() is not None else 0 + self.__training_sampler_num_workers: int = kwargs.get( + "training_sampler_num_workers", cpu_count + ) + if self.__training_sampler_num_workers > cpu_count: + self.__training_sampler_num_workers = cpu_count + self.__predicting_sampler_num_workers: int = kwargs.get( + "predicting_sampler_num_workers", cpu_count + ) + if self.__predicting_sampler_num_workers > cpu_count: + self.__predicting_sampler_num_workers = cpu_count + super(NodeClassificationLayerDependentImportanceSamplingTrainer, self).__init__( model, num_features, num_classes, device, init, feval, loss ) """ Set hyper parameters """ - " Configure num_layers " - self.__num_layers: int = kwargs.get("num_layers") - " Configure sampled_node_size_budget " - self.__sampled_node_size_budget: int = ( - kwargs.get("sampled_node_size_budget") - ) + self.__sampled_node_sizes: _typing.Sequence[int] = kwargs.get("sampled_node_sizes") self.__is_initialized: bool = False if init: @@ -846,10 +515,10 @@ class NodeClassificationLayerDependentImportanceSamplingTrainer(BaseNodeClassifi def get_model(self): return self.model - def __train_only(self, data): + def __train_only(self, integral_data): """ The function of training on the given dataset and mask. - :param data: data of a specific graph + :param integral_data: data of a specific graph :return: self """ optimizer: torch.optim.Optimizer = self._optimizer_class( @@ -886,56 +555,54 @@ class NodeClassificationLayerDependentImportanceSamplingTrainer(BaseNodeClassifi torch.optim.lr_scheduler.LambdaLR(optimizer, lambda _: 1.0) ) - sampled_node_size_budget: int = self.__sampled_node_size_budget - num_layers: int = self.__num_layers - __layer_dependent_importance_sampler: LayerDependentImportanceSampler = ( - LayerDependentImportanceSampler(data.edge_index) - ) - __top_layer_target_nodes_indexes: torch.LongTensor = ( - torch.where(data.train_mask)[0].unique() + LayerDependentImportanceSampler( + integral_data.edge_index, torch.where(integral_data.train_mask)[0].unique(), + self.__sampled_node_sizes, batch_size=self.__training_batch_size, + num_workers=self.__training_sampler_num_workers + ) ) for current_epoch in range(self._max_epoch): self.model.model.train() optimizer.zero_grad() """ epoch start """ " sample graphs " - __layers: _typing.Sequence[ - _typing.Tuple[torch.Tensor, torch.Tensor] - ] = __layer_dependent_importance_sampler.sample( - __top_layer_target_nodes_indexes, - [sampled_node_size_budget for _ in range(num_layers)] - ) - data.edge_indexes = [layer[0] for layer in __layers] - data.edge_weights = [layer[1] for layer in __layers] - data = data.to(self.device) - - result: torch.Tensor = self.model.model.forward(data) - if hasattr(torch.nn.functional, self.loss): - loss_function = getattr( - torch.nn.functional, self.loss + for sampled_data in __layer_dependent_importance_sampler: + optimizer.zero_grad() + sampled_data: TargetDependantSampledData = sampled_data + # 由于现在的Model设计是接受Data的，所以只能组装一个采样的Data作为参数 + sampled_graph: autogl.data.Data = autogl.data.Data( + x=integral_data.x[sampled_data.all_sampled_nodes_indexes], + y=integral_data.y[sampled_data.all_sampled_nodes_indexes] ) + sampled_graph.to(self.device) + sampled_graph.edge_indexes: _typing.Sequence[torch.LongTensor] = [ + current_layer.edge_index_for_sampled_graph.to(self.device) + for current_layer in sampled_data.sampled_edges_for_layers + ] + prediction: torch.Tensor = self.model.model(sampled_graph) + if not hasattr(torch.nn.functional, self.loss): + raise TypeError( + f"PyTorch does not support loss type {self.loss}" + ) + loss_function = getattr(torch.nn.functional, self.loss) loss_value: torch.Tensor = loss_function( - result[data.train_mask], - data.y[data.train_mask] - ) - else: - raise TypeError( - f"PyTorch does not support loss type {self.loss}" + prediction[sampled_data.target_nodes_indexes.indexes_in_sampled_graph], + sampled_graph.y[sampled_data.target_nodes_indexes.indexes_in_sampled_graph] ) + loss_value.backward() + optimizer.step() - loss_value.backward() - optimizer.step() if self._lr_scheduler_type: lr_scheduler.step() if ( - hasattr(data, "val_mask") and - getattr(data, "val_mask") is not None and - type(getattr(data, "val_mask")) == torch.Tensor + hasattr(integral_data, "val_mask") and + getattr(integral_data, "val_mask") is not None and + type(getattr(integral_data, "val_mask")) == torch.Tensor ): validation_results: _typing.Sequence[float] = self.evaluate( - (data,), "val", [self.feval[0]] + (integral_data,), "val", [self.feval[0]] ) if self.feval[0].is_higher_better(): validation_loss: float = -validation_results[0] @@ -946,23 +613,68 @@ class NodeClassificationLayerDependentImportanceSamplingTrainer(BaseNodeClassifi LOGGER.debug("Early stopping at %d", current_epoch) break if ( - hasattr(data, "val_mask") and - getattr(data, "val_mask") is not None and - type(getattr(data, "val_mask")) == torch.Tensor + hasattr(integral_data, "val_mask") and + getattr(integral_data, "val_mask") is not None and + type(getattr(integral_data, "val_mask")) == torch.Tensor ): self._early_stopping.load_checkpoint(self.model.model) - def __predict_only(self, data) -> torch.Tensor: + def __predict_only( + self, integral_data, + mask_or_target_nodes_indexes: _typing.Union[ + torch.BoolTensor, torch.LongTensor + ] + ) -> torch.Tensor: """ The function of predicting on the given data. - :param data: data of a specific graph + :param integral_data: data of a specific graph + :param mask_or_target_nodes_indexes: ... :return: the result of prediction on the given dataset """ - data = data.to(self.device) + if mask_or_target_nodes_indexes.dtype == torch.bool: + target_nodes_indexes: _typing.Any = ( + torch.where(mask_or_target_nodes_indexes)[0] + ) + else: + target_nodes_indexes: _typing.Any = mask_or_target_nodes_indexes.long() + + neighbor_sampler: NeighborSampler = NeighborSampler( + torch_geometric.utils.add_remaining_self_loops(integral_data.edge_index)[0], + target_nodes_indexes, [-1 for _ in self.__sampled_node_sizes], + batch_size=self.__predicting_batch_size, + num_workers=self.__predicting_sampler_num_workers, + shuffle=False + ) + + prediction_batch_cumulative_builder = ( + EvaluatorUtility.PredictionBatchCumulativeBuilder() + ) self.model.model.eval() - with torch.no_grad(): - predicted_x: torch.Tensor = self.model.model(data) - return predicted_x + for sampled_data in neighbor_sampler: + sampled_data: TargetDependantSampledData = sampled_data + sampled_graph: autogl.data.Data = autogl.data.Data( + integral_data.x[sampled_data.all_sampled_nodes_indexes], + integral_data.y[sampled_data.all_sampled_nodes_indexes] + ) + sampled_graph.to(self.device) + sampled_graph.edge_indexes: _typing.Sequence[torch.LongTensor] = [ + current_layer.edge_index_for_sampled_graph.to(self.device) + for current_layer in sampled_data.sampled_edges_for_layers + ] + sampled_graph.edge_weights: _typing.Sequence[torch.FloatTensor] = [ + current_layer.edge_weight.to(self.device) + for current_layer in sampled_data.sampled_edges_for_layers + ] + + with torch.no_grad(): + prediction_batch_cumulative_builder.add_batch( + sampled_data.target_nodes_indexes.indexes_in_integral_graph.cpu().numpy(), + self.model.model(sampled_graph)[ + sampled_data.target_nodes_indexes.indexes_in_sampled_graph + ].cpu().numpy() + ) + + return torch.from_numpy(prediction_batch_cumulative_builder.compose()[1]) def predict_proba( self, dataset, mask: _typing.Optional[str]=None, @@ -978,16 +690,16 @@ class NodeClassificationLayerDependentImportanceSamplingTrainer(BaseNodeClassifi data = dataset[0].to(self.device) if mask is not None and type(mask) == str: if mask.lower() == "train": - _mask: torch.Tensor = data.train_mask + _mask: torch.BoolTensor = data.train_mask elif mask.lower() == "test": - _mask: torch.Tensor = data.test_mask + _mask: torch.BoolTensor = data.test_mask elif mask.lower() == "val": - _mask: torch.Tensor = data.val_mask + _mask: torch.BoolTensor = data.val_mask else: - _mask: torch.Tensor = data.test_mask + _mask: torch.BoolTensor = data.test_mask else: - _mask: torch.Tensor = data.test_mask - result = self.__predict_only(data)[_mask] + _mask: torch.BoolTensor = data.test_mask + result = self.__predict_only(data, _mask) return result if in_log_format else torch.exp(result) def predict(self, dataset, mask: _typing.Optional[str] = None) -> torch.Tensor: @@ -1021,18 +733,12 @@ class NodeClassificationLayerDependentImportanceSamplingTrainer(BaseNodeClassifi prediction_probability: torch.Tensor = self.predict_proba(dataset, mask) y_ground_truth: torch.Tensor = data.y[_mask] - eval_results = [] - for f in _feval: - try: - eval_results.append(f.evaluate(prediction_probability, y_ground_truth)) - except: - eval_results.append( - f.evaluate( - prediction_probability.cpu().numpy(), - y_ground_truth.cpu().numpy(), - ) - ) - return eval_results + return [ + f.evaluate( + prediction_probability.cpu().numpy(), + y_ground_truth.cpu().numpy(), + ) for f in _feval + ] def train(self, dataset, keep_valid_result: bool = True): """ @@ -1043,9 +749,9 @@ class NodeClassificationLayerDependentImportanceSamplingTrainer(BaseNodeClassifi data = dataset[0] self.__train_only(data) if keep_valid_result: - prediction: torch.Tensor = self.__predict_only(data) - self._valid_result: torch.Tensor = prediction[data.val_mask].max(1)[1] - self._valid_result_prob: torch.Tensor = prediction[data.val_mask] + prediction: torch.Tensor = self.__predict_only(data, data.val_mask) + self._valid_result: torch.Tensor = prediction.max(1)[1] + self._valid_result_prob: torch.Tensor = prediction self._valid_score: _typing.Sequence[float] = self.evaluate(dataset, "val") def get_valid_predict(self) -> torch.Tensor: @@ -1189,6 +895,25 @@ class NodeClassificationNeighborSamplingTrainer(BaseNodeClassificationTrainer): self._valid_result_prob: torch.Tensor = torch.zeros(0) self._valid_score: _typing.Sequence[float] = () + self.__training_batch_size: int = kwargs.get("training_batch_size", 1024) + if not self.__training_batch_size > 0: + self.__training_batch_size: int = 1024 + self.__predicting_batch_size: int = kwargs.get("predicting_batch_size", 1024) + if not self.__predicting_batch_size > 0: + self.__predicting_batch_size: int = 1024 + + cpu_count: int = os.cpu_count() if os.cpu_count() is not None else 0 + self.__training_sampler_num_workers: int = kwargs.get( + "training_sampler_num_workers", cpu_count + ) + if self.__training_sampler_num_workers > cpu_count: + self.__training_sampler_num_workers = cpu_count + self.__predicting_sampler_num_workers: int = kwargs.get( + "predicting_sampler_num_workers", cpu_count + ) + if self.__predicting_sampler_num_workers > cpu_count: + self.__predicting_sampler_num_workers = cpu_count + super(NodeClassificationNeighborSamplingTrainer, self).__init__( model, num_features, num_classes, device, init, feval, loss ) @@ -1257,15 +982,14 @@ class NodeClassificationNeighborSamplingTrainer(BaseNodeClassificationTrainer): neighbor_sampler: NeighborSampler = NeighborSampler( integral_data.edge_index, torch.where(integral_data.train_mask)[0].unique(), - self.__sampling_sizes, batch_size=1024, - num_workers=os.cpu_count() if os.cpu_count() is not None else 0 + self.__sampling_sizes, batch_size=self.__training_batch_size, + num_workers=self.__training_sampler_num_workers ) - for current_epoch in tqdm.tqdm(range(self._max_epoch), desc="Epoch"): + for current_epoch in range(self._max_epoch): self.model.model.train() optimizer.zero_grad() """ epoch start """ " sample graphs " - # todo: Done this for sampled_data in neighbor_sampler: optimizer.zero_grad() sampled_data: TargetDependantSampledData = sampled_data @@ -1339,7 +1063,8 @@ class NodeClassificationNeighborSamplingTrainer(BaseNodeClassificationTrainer): neighbor_sampler: NeighborSampler = NeighborSampler( integral_data.edge_index, target_nodes_indexes, [-1 for _ in self.__sampling_sizes], - batch_size=1024, num_workers=0, shuffle=False + batch_size=self.__predicting_batch_size, + num_workers=self.__predicting_sampler_num_workers, shuffle=False ) prediction_batch_cumulative_builder = ( diff --git a/autogl/module/train/sampling/sampler/layer_dependent_importance_sampler.py b/autogl/module/train/sampling/sampler/layer_dependent_importance_sampler.py index abd3433..1c50cbe 100644 --- a/autogl/module/train/sampling/sampler/layer_dependent_importance_sampler.py +++ b/autogl/module/train/sampling/sampler/layer_dependent_importance_sampler.py @@ -24,7 +24,7 @@ class LayerDependentImportanceSampler(target_dependant_sampler.BasicLayerWiseTar __in_degree[__all_edge_index_with_self_loops[1]] ] ) - temp_tensor: torch.Tensor = 1.0 / temp_tensor + temp_tensor: torch.Tensor = torch.pow(temp_tensor, -0.5) temp_tensor[torch.isinf(temp_tensor)] = 0.0 return temp_tensor[0] * temp_tensor[1] @@ -134,9 +134,8 @@ class LayerDependentImportanceSampler(target_dependant_sampler.BasicLayerWiseTar ).unique() __all_candidate_source_nodes_indexes, all_candidate_source_nodes_probabilities = \ self._Utility.get_candidate_source_nodes_probabilities( - all_candidate_edge_indexes, - self._edge_index, - self.__all_edge_weights + all_candidate_edge_indexes, self._edge_index, + self.__all_edge_weights * self.__all_edge_weights ) assert __all_candidate_source_nodes_indexes.size() == all_candidate_source_nodes_probabilities.size() @@ -162,7 +161,7 @@ class LayerDependentImportanceSampler(target_dependant_sampler.BasicLayerWiseTar non_normalized_selected_edges_weight: torch.Tensor = ( self.__all_edge_weights[__selected_edges_indexes] / ( - selected_source_node_indexes.numel() * torch.tensor( + torch.tensor( [ all_candidate_source_nodes_probabilities[ __all_candidate_source_nodes_indexes == current_source_node_index @@ -195,89 +194,3 @@ class LayerDependentImportanceSampler(target_dependant_sampler.BasicLayerWiseTar non_normalized_selected_edges_weight ) return __selected_edges_indexes, normalized_selected_edges_weight - - # todo: Migrated to the overrode _sample_edges_for_layer method, remove in the future version - # def __sample_layer( - # self, target_nodes_indexes: torch.LongTensor, - # sampled_node_size_budget: int - # ) -> _typing.Tuple[torch.Tensor, torch.Tensor, torch.LongTensor, torch.LongTensor]: - # """ - # :param target_nodes_indexes: - # node indexes for target nodes in the top layer or nodes sampled in upper layer - # :param sampled_node_size_budget: - # :return: (Tensor, Tensor, LongTensor, LongTensor) - # """ - # all_candidate_edge_indexes: torch.LongTensor = torch.cat( - # [ - # torch.where(self._edge_index[1] == current_target_node_index)[0] - # for current_target_node_index in target_nodes_indexes.unique().tolist() - # ] - # ).unique() - # __all_candidate_source_nodes_indexes, all_candidate_source_nodes_probabilities = \ - # self._Utility.get_candidate_source_nodes_probabilities( - # all_candidate_edge_indexes, - # self._edge_index, - # self.__all_edge_weights - # ) - # assert __all_candidate_source_nodes_indexes.size() == all_candidate_source_nodes_probabilities.size() - # - # """ Sampling """ - # if sampled_node_size_budget < __all_candidate_source_nodes_indexes.numel(): - # selected_source_node_indexes: torch.LongTensor = __all_candidate_source_nodes_indexes[ - # torch.from_numpy( - # np.unique(np.random.choice( - # np.arange(__all_candidate_source_nodes_indexes.numel()), sampled_node_size_budget, - # p=all_candidate_source_nodes_probabilities.numpy() - # )) - # ).unique() - # ].unique() - # else: - # selected_source_node_indexes: torch.LongTensor = __all_candidate_source_nodes_indexes - # - # __selected_edges_indexes: torch.LongTensor = ( - # self._Utility.filter_selected_edges_by_source_nodes_and_target_nodes( - # self._edge_index, - # selected_source_node_indexes, target_nodes_indexes - # ) - # ).unique() - # - # non_normalized_selected_edges_weight: torch.Tensor = ( - # self.__all_edge_weights[__selected_edges_indexes] / ( - # selected_source_node_indexes.numel() * torch.tensor( - # [ - # all_candidate_source_nodes_probabilities[ - # __all_candidate_source_nodes_indexes == current_source_node_index - # ].item() - # for current_source_node_index - # in self._edge_index[0, __selected_edges_indexes].tolist() - # ] - # ) - # ) - # ) - # - # def __normalize_edges_weight_by_target_nodes( - # __edge_index: torch.Tensor, __edge_weight: torch.Tensor - # ) -> torch.Tensor: - # if __edge_index.size(1) != __edge_weight.numel(): - # raise ValueError - # for current_target_node_index in __edge_index[1].unique().tolist(): - # __current_mask_for_edges: torch.BoolTensor = ( - # __edge_index[1] == current_target_node_index - # ) - # __edge_weight[__current_mask_for_edges] = ( - # __edge_weight[__current_mask_for_edges] / ( - # torch.sum(__edge_weight[__current_mask_for_edges]) - # ) - # ) - # return __edge_weight - # - # normalized_selected_edges_weight: torch.Tensor = __normalize_edges_weight_by_target_nodes( - # self._edge_index[:, __selected_edges_indexes], - # non_normalized_selected_edges_weight - # ) - # return ( - # self._edge_index[:, __selected_edges_indexes], - # normalized_selected_edges_weight, - # selected_source_node_indexes, - # __selected_edges_indexes - # ) diff --git a/autogl/module/train/sampling/sampler/neighbor_sampler.py b/autogl/module/train/sampling/sampler/neighbor_sampler.py index b1d7c39..314e264 100644 --- a/autogl/module/train/sampling/sampler/neighbor_sampler.py +++ b/autogl/module/train/sampling/sampler/neighbor_sampler.py @@ -4,19 +4,23 @@ import torch_geometric from .target_dependant_sampler import TargetDependantSampler, TargetDependantSampledData -def _neighbor_sampler_transform( - batch_size: int, n_id: torch.LongTensor, - adj_list: _typing.Sequence[ - _typing.Tuple[torch.LongTensor, torch.LongTensor, _typing.Tuple[int, int]] - ] -) -> TargetDependantSampledData: - return TargetDependantSampledData( - [(current_layer[0], current_layer[1], None)for current_layer in adj_list], - (torch.arange(batch_size), n_id[:batch_size]), n_id - ) - - class NeighborSampler(TargetDependantSampler, _typing.Iterable): + @classmethod + def __compute_edge_weight(cls, edge_index: torch.LongTensor) -> torch.Tensor: + __num_nodes = max(int(edge_index[0].max()), int(edge_index[1].max())) + 1 + __out_degree: torch.LongTensor = torch_geometric.utils.degree( + edge_index[0], __num_nodes + ) + __in_degree: torch.LongTensor = torch_geometric.utils.degree( + edge_index[1], __num_nodes + ) + temp_tensor: torch.Tensor = torch.stack( + [__out_degree[edge_index[0]], __in_degree[edge_index[1]]] + ) + temp_tensor: torch.Tensor = torch.pow(temp_tensor, -0.5) + temp_tensor[torch.isinf(temp_tensor)] = 0.0 + return temp_tensor[0] * temp_tensor[1] + def __init__( self, edge_index: torch.LongTensor, target_nodes_indexes: torch.LongTensor, @@ -24,14 +28,29 @@ class NeighborSampler(TargetDependantSampler, _typing.Iterable): batch_size: int = 1, num_workers: int = 0, shuffle: bool = True, **kwargs ): + self.__edge_weight: torch.Tensor = self.__compute_edge_weight(edge_index) self.__pyg_neighbor_sampler: torch_geometric.data.NeighborSampler = ( torch_geometric.data.NeighborSampler( edge_index, list(sampling_sizes[::-1]), target_nodes_indexes, - transform=_neighbor_sampler_transform, batch_size=batch_size, + transform=self._transform, batch_size=batch_size, num_workers=num_workers, shuffle=shuffle, **kwargs ) ) + def _transform( + self, batch_size: int, n_id: torch.LongTensor, + adj_list: _typing.Sequence[ + _typing.Tuple[torch.LongTensor, torch.LongTensor, _typing.Tuple[int, int]] + ] + ) -> TargetDependantSampledData: + return TargetDependantSampledData( + [ + (current_layer[0], current_layer[1], self.__edge_weight[current_layer[1]]) + for current_layer in adj_list + ], + (torch.arange(batch_size, dtype=torch.long).long(), n_id[:batch_size]), n_id + ) + def __iter__(self): return iter(self.__pyg_neighbor_sampler) diff --git a/autogl/module/train/sampling/sampler/target_dependant_sampler.py b/autogl/module/train/sampling/sampler/target_dependant_sampler.py index 7adc816..ab8620f 100644 --- a/autogl/module/train/sampling/sampler/target_dependant_sampler.py +++ b/autogl/module/train/sampling/sampler/target_dependant_sampler.py @@ -254,6 +254,7 @@ class BasicLayerWiseTargetDependantSampler(TargetDependantSampler): __sampled_nodes_in_sub_graph_mapping.get(current_target_node_index_in_integral_data) for current_target_node_index_in_integral_data in top_layer_target_nodes_indexes.tolist() + if current_target_node_index_in_integral_data in __sampled_nodes_in_sub_graph_mapping ] ).long(), # Remap top_layer_target_nodes_indexes diff --git a/configs/nodeclf_ladies_gcn.yml b/configs/nodeclf_ladies_gcn.yml index 83c7e6f..c61ce3b 100644 --- a/configs/nodeclf_ladies_gcn.yml +++ b/configs/nodeclf_ladies_gcn.yml @@ -7,23 +7,19 @@ hpo: name: random models: - hp_space: - - feasiblePoints: - - 0 - parameterName: add_self_loops, - type: CATEGORICAL, - - feasiblePoints: 5,5 - parameterName: num_layers - type: DISCRETE - - cutFunc: lambda x:x[0] - 1 + - parameterName: num_layers + type: FIXED + value: 5 + - parameterName: hidden + type: CATEGORICAL_LIST + cutFunc: lambda x:x[0] - 1 cutPara: - num_layers length: 4 - maxValue: 256 - minValue: 64 - numericalType: INTEGER - parameterName: hidden - scalingType: LOG - type: NUMERICAL_LIST + feasiblePoints: + - 128 + - 256 + - 512 - maxValue: 0.8 minValue: 0.2 parameterName: dropout @@ -36,23 +32,46 @@ models: - tanh parameterName: act type: CATEGORICAL + - parameterName: add_self_loops + type: FIXED + value: 0 + - parameterName: normalize + type: FIXED + value: 0 name: gcn trainer: name: NodeClassificationLayerDependentImportanceSamplingTrainer hp_space: - - feasiblePoints: 128,256,512 - parameterName: sampled_node_size_budget - type: DISCRETE - - maxValue: 300 - minValue: 100 + - parameterName: sampled_node_sizes + type: CATEGORICAL_LIST + length: 5 + feasiblePoints: + - 128 + - 256 + - 512 + - 1024 + cutFunc: lambda x:x[0] + cutPara: + - num_layers + - maxValue: 128 + minValue: 64 parameterName: max_epoch scalingType: LINEAR type: INTEGER - - maxValue: 30 - minValue: 10 + - maxValue: 16 + minValue: 8 parameterName: early_stopping_round scalingType: LINEAR type: INTEGER + - parameterName: training_batch_size + type: FIXED + value: 1024 + - parameterName: predicting_batch_size + type: FIXED + value: 1024 + - parameterName: predicting_sampler_num_workers + type: FIXED + value: 0 - maxValue: 0.05 minValue: 0.01 parameterName: lr