diff --git a/docs/source/modeling/tracking.rst b/docs/source/modeling/tracking.rst index 2b89f0b..1ee21fd 100644 --- a/docs/source/modeling/tracking.rst +++ b/docs/source/modeling/tracking.rst @@ -34,3 +34,8 @@ You can find an example of deployment in the repo. - The key of the deployment is creating a class that inherits from `mlflow.pyfunc.PythonModel` with a `predict()` function. - That class is pickled and logged as artifact of the training. At inference time it will be used to make predictions. +Additionally, consider the following for more configurable deployment: + +- *Dynamic inference parameters*: Store inference hyperparameters (e.g., batch size or thresholds) as a separate artifact in MLFlow. Use `artifacts` options in `log_model` and then retrieve the file using the `context` object provided by the MLFlow in `load_context` or `predict`. +- *Multiple outputs*: `predict` function can return a Pandas DataFrame object. Employ it if the model has multiple targets or for providing logits scores for dynamic threshold adjusting on the client-side. +- *Serving labels*: Log a separate artifact in MLFlow for the client-side to map predictions back to human-readable labels. diff --git a/notebooks/models/oguz/transformer_v0.5_1D.ipynb b/notebooks/models/oguz/transformer_v0.5_1D.ipynb index ea76360..3f151a9 100644 --- a/notebooks/models/oguz/transformer_v0.5_1D.ipynb +++ b/notebooks/models/oguz/transformer_v0.5_1D.ipynb @@ -430,6 +430,54 @@ "outputs": [], "metadata": {} }, + { + "cell_type": "code", + "execution_count": null, + "source": [], + "outputs": [], + "metadata": {} + }, + { + "cell_type": "markdown", + "source": [ + "## Inference Preprocessing (Offline Testing Environment)" + ], + "metadata": {} + }, + { + "cell_type": "code", + "execution_count": null, + "source": [ + "import pandas as pd\n", + "\n", + "DATA_PATH = 'leads.csv'\n", + "data = pd.read_csv(DATA_PATH)\n", + "data.head()" + ], + "outputs": [], + "metadata": {} + }, + { + "cell_type": "code", + "execution_count": null, + "source": [ + "from ast import literal_eval\n", + "\n", + "lengths = {}\n", + "\n", + "for field in ['extracted_text_as_paragraphs', 'extracted_text_as_sentences']:\n", + " arr = data[field].apply(literal_eval).tolist()\n", + " lengths[field] = [len(ds) for ds in arr]\n", + " \n", + " infer_df = pd.DataFrame.from_dict({\n", + " 'excerpt': [d for ds in arr for d in ds]\n", + " })\n", + " infer_df = infer_df[~(infer_df['excerpt'].str.len() == 0)]\n", + " infer_df.to_csv(f'infer_{field}.csv', header=True, index=True)" + ], + "outputs": [], + "metadata": {} + }, { "cell_type": "code", "execution_count": null, diff --git a/scripts/training/oguz/huggingface-multihead/TODO b/scripts/training/oguz/huggingface-multihead/TODO deleted file mode 100644 index c497d20..0000000 --- a/scripts/training/oguz/huggingface-multihead/TODO +++ /dev/null @@ -1,8 +0,0 @@ -1. Training logic -2. MLFlow hyperparameters logging -2. MLFlow artifact logging -3. Tensorboard training + evaluation logging artifacts -4. HuggingFace: skip_memory_metrics -5. Check Selim's notebook for tricks on optimizer, scheduler, etc. -6. AUC-ROC -7. Not logging parameters I have set? \ No newline at end of file diff --git a/scripts/training/oguz/huggingface-multihead/augmentations.py b/scripts/training/oguz/huggingface-multihead/augmentations.py new file mode 100644 index 0000000..e69de29 diff --git a/scripts/training/oguz/huggingface-multihead/constants.py b/scripts/training/oguz/huggingface-multihead/constants.py index 58c37c0..25c6fa6 100644 --- a/scripts/training/oguz/huggingface-multihead/constants.py +++ b/scripts/training/oguz/huggingface-multihead/constants.py @@ -1,63 +1,75 @@ +SECTORS = [ + "Agriculture", + "Cross", + "Education", + "Food Security", + "Health", + "Livelihoods", + "Logistics", + "Nutrition", + "Protection", + "Shelter", + "WASH", +] + PILLARS_1D = [ "Context", - "Humanitarian Profile", - "Displacement", + "Shock/Event", "Casualties", + "Displacement", "Humanitarian Access", - "Information", + "Information And Communication", + "Covid-19", ] SUBPILLARS_1D = [ [ - "Context->Security & Stability", "Context->Demography", "Context->Economy", - "Context->Hazard & Threats", - "Context->Politics", - "Context->Overview", - "Context->Key Event", - "Context->Socio Cultural", - "Context->Legal & Policy", "Context->Environment", - "Context->Stakeholders", - "Context->Response gap", + "Context->Security & Stability", + "Context->Socio Cultural", + "Context->Legal & Policy", + "Context->Politics", + "Context->Technological", ], [ - "Humanitarian Profile->Affected Groups", - "Humanitarian Profile->Casualties", - "Humanitarian Profile->Population Movement", + "Shock/Event->Type And Characteristics", + "Shock/Event->Underlying/Aggravating Factors", + "Shock/Event->Hazard & Threats", ], + ["Casualties->Dead", "Casualties->Injured", "Casualties->Missing"], [ - "Displacement->Push/Pull Factors", - "Displacement->Type/Numbers", - "Displacement->Local Integration", + "Displacement->Type/Numbers/Movements", + "Displacement->Push Factors", + "Displacement->Pull Factors", "Displacement->Intentions", - "Displacement->Displacement", + "Displacement->Local Integration", ], - ["Casualties->Dead", "Casualties->Injured", "Casualties->Missing"], [ + "Humanitarian Access->Relief To Population", + "Humanitarian Access->Population To Relief", "Humanitarian Access->Physical Constraints", - "Humanitarian Access->Humanitarian Access Gaps", + ( + "Humanitarian Access->Number Of People Facing Humanitarian Access Constraints" + "/Humanitarian Access Gaps" + ), ], [ - "Information->Information Gaps", - "Information->Channels & Means", - "Information->Information Challenges", + "Information And Communication->Information Challenges And Barriers", + "Information And Communication->Communication Means And Preferences", + "Information And Communication->Knowledge And Info Gaps (Pop)", + "Information And Communication->Knowledge And Info Gaps (Hum)", + ], + [ + "Covid-19->Cases", + "Covid-19->Deaths", + "Covid-19->Testing", + "Covid-19->Contact Tracing", + "Covid-19->Hospitalization & Care", + "Covid-19->Vaccination", + "Covid-19->Restriction Measures", ], -] - -SECTORS = [ - "Agriculture", - "Cross", - "Education", - "Food Security", - "Health", - "Livelihoods", - "Logistics", - "Nutrition", - "Protection", - "Shelter", - "WASH", ] PILLARS_2D = [ @@ -65,7 +77,7 @@ "Capacities & Response", "Impact", "Priority Interventions", - "People At Risk", + "At Risk", "Priority Needs", ] @@ -73,21 +85,19 @@ [ "Humanitarian Conditions->Coping Mechanisms", "Humanitarian Conditions->Living Standards", - "Humanitarian Conditions->Number Of People In Need", "Humanitarian Conditions->Physical And Mental Well Being", + "Humanitarian Conditions->Number Of People In Need", ], [ "Capacities & Response->International Response", "Capacities & Response->National Response", - "Capacities & Response->Number Of People Reached", - "Capacities & Response->Response Gaps", + "Capacities & Response->Local Response", + "Capacities & Response->Number Of People Reached/Response Gaps", ], [ "Impact->Driver/Aggravating Factors", "Impact->Impact On People", - "Impact->Impact On People Or Impact On Services", - "Impact->Impact On Services", - "Impact->Impact On Systems And Services", + "Impact->Impact On Systems, Services And Networks", "Impact->Number Of People Affected", ], [ @@ -95,8 +105,8 @@ "Priority Interventions->Expressed By Population", ], [ - "People At Risk->Number Of People At Risk", - "People At Risk->Risk And Vulnerabilities", + "At Risk->Risk And Vulnerabilities", + "At Risk->Number Of People At Risk", ], [ "Priority Needs->Expressed By Humanitarian Staff", diff --git a/scripts/training/oguz/huggingface-multihead/data.py b/scripts/training/oguz/huggingface-multihead/data.py index dd3124e..c45bdfd 100644 --- a/scripts/training/oguz/huggingface-multihead/data.py +++ b/scripts/training/oguz/huggingface-multihead/data.py @@ -1,34 +1,24 @@ import logging from ast import literal_eval -from typing import List, Optional, Union +from typing import Dict, List, Optional, Union import numpy as np import pandas as pd import torch from torch.utils.data import Dataset from transformers import PreTrainedTokenizer -from utils import revdict +from utils import revdict, read_dataframe -class MultiHeadDataFrame(Dataset): - """Creates a PyTorch dataset out of a Pandas DataFrame +class TextDataFrame(Dataset): + """Creates a PyTorch dataset out of a text field inside a Pandas DataFrame. Args: - dataframe: path to a DataFrame or directly a DataFrame + dataframe: path to a DataFrame or a DataFrame tokenizer: tokenizer to pre-process source text source: textual source field that will be the input of models - target: target classification field that will be the output of models - groups: transforms target into a multi-target (each multi-label) - that is, each sample is associated with 2D one-hot target matrix - group_names: name assoaciated with each classification head - filter: (None, str, List of strings) filter dataset according - to given group. `None` uses all of the data points. Single str key - uses all data points with at least one target key value. If a list of - strings is given, each key is used to check positivity of the sample, - e.g., ['sector', 'pillar2d'] checks whether the data point has at - least one target in `sector` or in `pillar2d` fields. - flatten: flatten group targets to 1D for convenience online: online or offline tokenization + tokenizer_max_len: maximum output length for the tokenizer """ def __init__( @@ -36,15 +26,9 @@ def __init__( dataframe: Union[str, pd.DataFrame], tokenizer: PreTrainedTokenizer, source: str = "excerpt", - target: str = "target", - groups: Optional[List[List[str]]] = None, - group_names: Optional[List[str]] = None, - filter: Optional[Union[str, List[str]]] = None, - flatten: bool = True, online: bool = False, + tokenizer_max_len: Optional[int] = None, ): - self.group_names = group_names - self.flatten = flatten self.tokenizer = tokenizer self.online = online self.logger = logging.getLogger() @@ -52,37 +36,116 @@ def __init__( # read dataframe manually if given as path if isinstance(dataframe, str): self.logger.info(f"Loading dataframe: {dataframe}") - dataframe = pd.read_pickle(dataframe) + dataframe = read_dataframe(dataframe) + + # prepare tokenizer options + self.tokenizer_options = { + "truncation": True, + "padding": True, + } + if tokenizer_max_len: + self.tokenizer_options.update( + { + "padding": "max_length", + "max_length": min(tokenizer_max_len, tokenizer.model_max_length), + } + ) + if tokenizer.model_max_length < tokenizer_max_len: + self.logger.info( + f"Using maximum model length: {tokenizer.model_max_length} instead" + f"of given length: {tokenizer_max_len}" + ) - # apply literal eval to have lists in target - dataframe[target] = dataframe[target].apply(literal_eval) + # save data as exceprt + self.data = dataframe[source].tolist() + self.data_len = len(self.data) - # cast filter to array - if isinstance(filter, str): - filter = [filter] + if not self.online: + # tokenize and save source data + self.logger.info("Applying offline tokenization") + self.data = tokenizer(self.data, **self.tokenizer_options) - # filter data frame - if filter is not None: - pos = np.zeros(len(dataframe), dtype=np.bool) - for f in filter: - pos |= np.array([len(item) > 0 for item in dataframe[f].tolist()], dtype=np.bool) - dataframe = dataframe[pos] - self.logger.info(f"Filtered data points with non-empty (or) {','.join(filter)} values") + def __len__(self): + return self.data_len + def __getitem__(self, idx): if self.online: - # save data as exceprt - self.data = dataframe[source].tolist() + data = self.tokenizer(self.data[idx : idx + 1], **self.tokenizer_options) + item = {key: torch.tensor(val[0]) for key, val in data.items()} else: - # tokenize and save source data - self.logger.info(f"Applying offline tokenization") - self.data = tokenizer(dataframe[source].tolist(), truncation=True, padding=True) + item = {key: torch.tensor(val[idx]) for key, val in self.data.items()} + + return item + + +class MultiTargetDataFrame(Dataset): + """Creates a PyTorch dataset out of a field containing list of labels + for a multi-label classification problem out of a Pandas DataFrame. - # prepare target encoding - all_targets = np.hstack(dataframe[target].to_numpy()) - uniq_targets = np.unique(all_targets) - # cluster into groups + Args: + dataframe: path to a DataFrame or a DataFrame + target: target classification field that will be the output of models + groups: transforms target into a multi-target problem (each multi-label) + that is, each sample is associated with 2D one-hot target matrix + e.g., 6 label classification with two groups: [A, B, C], [D, E, F] + group_names: name assoaciated with each classification head + e.g., 2 group names: ABC and DEF + exclude: (None, List of strings, List of List of strings) omit the given + targets. For multi-target classification, expects a list with + elements of lists. + flatten: flatten targets to 1D for convenience + iterative: include group targets on top of regular targets + """ + + def __init__( + self, + dataframe: Union[str, pd.DataFrame], + target: str = "target", + groups: Optional[List[List[str]]] = None, + group_names: Optional[List[str]] = None, + exclude: Optional[Union[str, List[str]]] = None, + flatten: bool = True, + iterative: bool = True, + ): + # process groups + if groups is not None: + if group_names is not None: + assert len(groups) == len( + group_names + ), f"Group names '{group_names}' should be at equal length with groups '{groups}'" + else: + group_names = [f"Group {i}" for i in range(len(groups))] + else: + assert flatten, "Use flatten if no group information is provided" + + self.groups = groups + self.group_names = group_names + self.flatten = flatten + self.iterative = iterative + self.logger = logging.getLogger() + + # read dataframe manually if given as path + if isinstance(dataframe, str): + self.logger.info(f"Loading dataframe: {dataframe}") + dataframe = read_dataframe(dataframe) + + # apply literal eval to have lists in target + if not isinstance(dataframe[target].iloc[0], list): + dataframe[target] = dataframe[target].apply(literal_eval) + + # omit the given exclude labels + if exclude: + if isinstance(exclude, str): + exclude = [exclude] + + dataframe[target] = [ + [label for label in labels if label not in exclude] + for labels in dataframe[target].tolist() + ] + if groups: + # process given groups self.group_encoding = {t: idx for idx, group in enumerate(groups) for t in group} self.group_decoding = {idx: group for idx, group in enumerate(groups)} @@ -90,22 +153,81 @@ def __init__( self.target_decoding = [revdict(encoding) for encoding in self.target_encoding] self.target_classes = [len(encoding.keys()) for encoding in self.target_encoding] else: + assert not self.iterative, "Provide groups if iterative labels are asked" + + # prepare target encoding + all_targets = np.hstack(dataframe[target].to_numpy()) + uniq_targets = np.unique(all_targets) + + # single group encoding - decoding self.group_encoding = {t: 0 for t in uniq_targets} self.group_decoding = {0: uniq_targets} + self.groups = [uniq_targets.tolist()] + self.group_names = ["ALL"] self.target_encoding = {t: idx for idx, t in enumerate(uniq_targets)} self.target_encoding = revdict(self.target_encoding) self.target_classes = [len(self.target_encoding.keys())] - self.logger.info(f"Automatically set target encodings: {self.target_encoding}") - self.logger.info(f"Target size: [{self.target_classes}]") + self.logger.info(f"Using target encodings: {self.target_encoding}") + self.logger.info(f"Target size: [{self.target_classes}]") # prepare targets self.target = [self.onehot_encode(ts) for ts in dataframe[target].tolist()] + self.data_len = len(self.target) + # prepare group targets if groups: self.group = [self.group_encode(ts) for ts in dataframe[target].tolist()] + def __len__(self): + return self.data_len + + def __getitem__(self, idx): + item = {} + + if self.flatten: + item["labels"] = torch.tensor(self.target[idx]) + else: + item.update( + { + f"labels_{self.group_names[i]}": torch.tensor(self.target[idx][i]) + for i in range(len(self.target_classes)) + } + ) + + if self.iterative: + item["groups"] = torch.tensor(self.group[idx]) + + return item + + def label_names(self) -> List[str]: + """Return label names""" + return self[0].keys() + + def compute_stats(self) -> Dict[str, int]: + """Computes occurences of each target and group""" + + counts = {} + classes = [target for group in self.groups for target in group] + if self.flatten: + sums = np.sum(np.stack(self.target, axis=-1), axis=-1) + counts.update({c: s for c, s in zip(classes, sums.tolist())}) + else: + for i, _ in enumerate(self.group_names): + targets = [target[i] for target in self.target] + sums = np.sum(np.stack(targets, axis=-1), axis=-1) + counts.update({c: s for c, s in zip(self.groups[i], sums)}) + + for i, group_name in enumerate(self.group_names): + counts.update( + {group_name: np.sum(np.array([counts[group] for group in self.groups[i]]))} + ) + counts.update( + {"ALL": np.sum(np.array([counts[group_name] for group_name in self.group_names]))} + ) + return counts + def group_encode(self, targets: List[str]) -> np.ndarray: """Encodes given targets to group representation""" @@ -155,27 +277,163 @@ def onehot_decode(self, onehot: Union[np.ndarray, List[np.ndarray]]) -> List[str if onehot[i][j] == 1 ] + +class MultiHeadDataFrame(Dataset): + """Creates a PyTorch dataset out of a Pandas DataFrame that supports + multi-head classification tasks where each fine-grained label belongs + to a super-category or a group. + + + Args: + dataframe: path to a DataFrame or a DataFrame + tokenizer: tokenizer to pre-process source text + source: textual source field that will be the input of models + targets: target classification fields that will be the output of models + groups: transforms target into a multi-target (each multi-label) + that is, each sample is associated with 2D one-hot target matrix + group_names: name assoaciated with each classification head + filter: (None, str, List of strings) filter dataset according + to given group. `None` uses all of the data points. Single str key + uses all data points with at least one target key value. If a list of + strings is given, each key is used to check positivity of the sample, + e.g., ['sector', 'pillar2d'] checks whether the data point has at + least one target in `sector` or in `pillar2d` fields. + exclude: (None, List of strings, List of List of strings) omit the given + targets. For multi-target classification, expects a list with + elements of lists. + flatten: flatten group targets to 1D for convenience + iterative: include group targets on top of regular targets + online: online or offline tokenization + inference: if True, does not process target or groups + tokenizer_max_len: maximum output length for the tokenizer + """ + + def __init__( + self, + dataframe: Union[str, pd.DataFrame], + tokenizer: PreTrainedTokenizer, + source: str = "excerpt", + targets: Union[str, List[str]] = "target", + groups: Optional[Union[List[List[str]], List[List[List[str]]]]] = None, + group_names: Optional[Union[List[str], List[List[str]]]] = None, + exclude: Optional[List[str]] = None, + filter: Optional[Union[str, List[str]]] = None, + flatten: bool = True, + iterative: bool = True, + online: bool = False, + inference: bool = False, + tokenizer_max_len: Optional[int] = None, + ): + self.logger = logging.getLogger() + self.flatten = flatten + self.iterative = iterative + + if online: + # ensure that we are in training + assert not inference, "Online tokenization is only supported in training-time" + + # read dataframe manually if given as path + if isinstance(dataframe, str): + self.logger.info(f"Loading dataframe: {dataframe}") + dataframe = read_dataframe(dataframe) + + # cast filter to array + if isinstance(filter, str): + filter = [filter] + + # filter data frame + if filter is not None: + pos = np.zeros(len(dataframe), dtype=np.bool) + for f in filter: + # apply literal eval to have lists + dataframe[f] = dataframe[f].apply(literal_eval) + + # get positive fields + pos |= np.array([len(item) > 0 for item in dataframe[f].tolist()], dtype=np.bool) + + # filter negative rows + dataframe = dataframe[pos] + self.logger.info( + f"Filtered data points with non-empty {','.join(filter)} values" + "(using 'or' if multiple fields)" + ) + + # prepare text source data + self.data = TextDataFrame( + dataframe=dataframe, + tokenizer=tokenizer, + source=source, + online=online, + tokenizer_max_len=tokenizer_max_len, + ) + self.data_len = self.data.data_len + self.tokenizer_options = self.data.tokenizer_options + + # prepare targets + if isinstance(targets, str): + # assert isinstance(groups, List[List[str]]), "Expecting `groups` to be a list of lists" + # assert isinstance(group_names, List[str]), "Expecting `group_names` to be a list" + + targets = [targets] + groups = [groups] + group_names = [group_names] + self.single = True + else: + # assert isinstance( + # groups, List[List[List[str]]] + # ), "Expecting `groups` to be a list of lists of lists" + # assert isinstance( + # group_names, List[List[str]] + # ), "Expecting `group_names` to be a list of lists" + self.single = False + + # set defaults for groups, group_names and exclude + if groups is None: + groups = [None for _ in targets] + group_names = [None for _ in targets] + + self.tasks = targets + self.targets = [] + if not inference: + for _target, _groups, _group_names in zip(targets, groups, group_names): + self.targets.append( + MultiTargetDataFrame( + dataframe=dataframe, + target=_target, + groups=_groups, + group_names=_group_names, + exclude=exclude, + flatten=flatten, + iterative=iterative, + ) + ) + assert len(self.data) == len( + self.targets[-1] + ), "Text source and target have different lengths!" + def __len__(self): - return len(self.target) + return self.data_len def __getitem__(self, idx): - if self.online: - data = self.tokenizer(self.data[idx : idx + 1], truncation=True, padding=True) - item = {key: torch.tensor(val[0]) for key, val in data.items()} - else: - item = {key: torch.tensor(val[idx]) for key, val in self.data.items()} + item = self.data[idx] - if self.flatten: - item["labels"] = torch.tensor(self.target[idx]) + if self.single: + item.update(self.targets[0][idx]) else: - item.update( - { - f"labels_{self.group_names[i]}": torch.tensor(self.target[idx][i]) - for i in range(len(self.target_classes)) - } - ) - - if self.group is not None: - item["groups"] = torch.tensor(self.group[idx]) + for task, target in zip(self.tasks, self.targets): + item.update({(f"{task}_" + k): v for k, v in target[idx].items()}) return item + + def label_names(self) -> List[str]: + """Return label names""" + data_keys = self.data[0].keys() + return [key for key in self[0].keys() if key not in data_keys] + + def compute_stats(self) -> Dict[str, int]: + """Computes occurences of each target and group""" + + counts = {} + for task, target in zip(self.tasks, self.targets): + counts[task] = target.compute_stats() + return counts diff --git a/scripts/training/oguz/huggingface-multihead/evaluation.py b/scripts/training/oguz/huggingface-multihead/evaluation.py new file mode 100644 index 0000000..1c1bb35 --- /dev/null +++ b/scripts/training/oguz/huggingface-multihead/evaluation.py @@ -0,0 +1,117 @@ +from typing import List + +from sklearn.metrics import accuracy_score, precision_recall_fscore_support + +"""Evaluation logic here only work with flattened datasets.""" + + +def _prefix(dic, prefix): + """Adds prefix to dictionary keys""" + + return {(prefix + k): v for k, v in dic.items()} + + +def _process(text): + """Replaces special characters in text (for MLFlow)""" + text = text.lower() + text = text.replace(" ", "_") + text = text.replace("&", "_") + text = text.replace(">", "") + text = text.replace(",", "") + text = text.replace("(", "") + text = text.replace(")", "") + return text + + +# compute metrics given preds and labels +def _compute(preds, labels, average="micro", threshold=0.5): + preds = preds > threshold + precision, recall, f1, _ = precision_recall_fscore_support(labels, preds, average=average) + accuracy = accuracy_score(labels, preds) + return { + "accuracy": accuracy, + "f1": f1, + "precision": precision, + "recall": recall, + } + + +def compute_multiclass_metrics(preds, labels, names: List[str], threshold: float = 0.5): + """Compute metrics for multi-target classification tasks""" + metrics = {} + + # micro evaluation + metrics.update(_prefix(_compute(preds, labels, "micro"), "micro_"), threshold=threshold) + # macro evaluation + metrics.update(_prefix(_compute(preds, labels, "macro"), "macro_"), threshold=threshold) + + # per class evaluation + for idx, name in enumerate(names): + # per class micro evaluation + metrics.update( + _prefix( + _compute( + preds[:, idx : idx + 1], labels[:, idx : idx + 1], "binary", threshold=threshold + ), + f"{_process(name)}_binary_", + ) + ) + + return metrics + + +def compute_multitarget_metrics( + preds, + labels, + groups: List[List[str]], + group_names: List[str], + threshold: float = 0.5, +): + metrics = {} + start = 0 + for idx, group_name in enumerate(group_names): + metrics.update( + _prefix( + compute_multiclass_metrics( + preds[:, start : start + len(groups[idx])], + labels[:, start : start + len(groups[idx])], + names=groups[idx], + threshold=threshold, + ), + f"{_process(group_name)}_", + ) + ) + start = start + len(groups[idx]) + + # micro evaluation + metrics.update(_prefix(_compute(preds, labels, "micro", threshold=threshold), "micro_")) + # macro evaluation + metrics.update(_prefix(_compute(preds, labels, "macro", threshold=threshold), "macro_")) + + return metrics + + +def compute_multihead_metrics( + preds, + labels, + groups: List[List[List[str]]], + group_names: List[List[str]], + targets: List[str], + threshold: float = 0.5, +): + metrics = {} + for idx, target in enumerate(targets): + metrics.update( + _prefix( + compute_multitarget_metrics( + preds[idx], + labels[idx], + groups=groups[idx], + group_names=group_names[idx], + threshold=threshold, + ), + f"{_process(target)}_", + ) + ) + + return metrics diff --git a/scripts/training/oguz/huggingface-multihead/infer.py b/scripts/training/oguz/huggingface-multihead/infer.py new file mode 100644 index 0000000..7ae8ff3 --- /dev/null +++ b/scripts/training/oguz/huggingface-multihead/infer.py @@ -0,0 +1,49 @@ +import argparse +import logging +import os +import sys + +import mlflow +import pandas as pd + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + + # hyperparameters sent by the client + # parser.add_argument("--eval_batch_size", type=int, default=64) + parser.add_argument("--model_uri", type=str, required=True) + + # SageMaker parameters - data, model, and output directories + parser.add_argument("--output_data_dir", type=str, default=os.environ["SM_OUTPUT_DATA_DIR"]) + parser.add_argument("--model_dir", type=str, default=os.environ["SM_MODEL_DIR"]) + parser.add_argument("--n_gpus", type=str, default=os.environ["SM_NUM_GPUS"]) + parser.add_argument("--test_dir", type=str, default=os.environ["SM_CHANNEL_TEST"]) + args, _ = parser.parse_known_args() + + # set up logging + logger = logging.getLogger(__name__) + logging.basicConfig( + level=logging.getLevelName("INFO"), + handlers=[logging.StreamHandler(sys.stdout)], + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + ) + logger.info(f"Args: {args}") + + # load datasets + infer_df = pd.read_pickle(f"{args.test_dir}/infer_df.pickle") + logger.info(f" loaded infer_dataset length is: {infer_df.shape}") + logger.info(infer_df.head()) + + # get model + pyfunc_wrapper = mlflow.pyfunc.load_model(args.model_uri) + + # set eval batch size + # python_model = pyfunc_wrapper._model_impl.python_model + # logging.info(python_model.infer_params) + # python_model.infer_params["dataloader"]["batch_size"] = args.eval_batch_size + + # predict + pred_df = pyfunc_wrapper.predict(infer_df) + + # save predictions + pred_df.to_csv(f"{args.output_data_dir}/preds.csv", header=True, index=True) diff --git a/scripts/training/oguz/huggingface-multihead/loss.py b/scripts/training/oguz/huggingface-multihead/loss.py new file mode 100644 index 0000000..0a37b25 --- /dev/null +++ b/scripts/training/oguz/huggingface-multihead/loss.py @@ -0,0 +1,90 @@ +import torch +import torch.nn.functional as F + + +def sigmoid_focal_loss( + inputs: torch.Tensor, + targets: torch.Tensor, + alpha: float = 0.25, + gamma: float = 2, + reduction: str = "mean", +): + """ + Original implementation from + https://github.com/facebookresearch/fvcore/blob/master/fvcore/nn/focal_loss.py + Loss used in RetinaNet for dense detection: https://arxiv.org/abs/1708.02002. + + Args: + inputs: A float tensor of arbitrary shape. + The predictions for each example. + targets: A float tensor with the same shape as inputs. Stores the binary + classification label for each element in inputs + (0 for the negative class and 1 for the positive class). + alpha: (optional) Weighting factor in range (0,1) to balance + positive vs negative examples or -1 for ignore. Default = 0.25 + gamma: Exponent of the modulating factor (1 - p_t) to + balance easy vs hard examples. + reduction: 'none' | 'mean' | 'sum' + 'none': No reduction will be applied to the output. + 'mean': The output will be averaged. + 'sum': The output will be summed. + Returns: + Loss tensor with the reduction option applied. + """ + p = torch.sigmoid(inputs) + ce_loss = F.binary_cross_entropy_with_logits(inputs, targets, reduction="none") + p_t = p * targets + (1 - p) * (1 - targets) + loss = ce_loss * ((1 - p_t) ** gamma) + + if alpha >= 0: + alpha_t = alpha * targets + (1 - alpha) * (1 - targets) + loss = alpha_t * loss + + if reduction == "mean": + loss = loss.mean() + elif reduction == "sum": + loss = loss.sum() + + return loss + + +def sigmoid_focal_loss_star( + inputs: torch.Tensor, + targets: torch.Tensor, + alpha: float = -1, + gamma: float = 1, + reduction: str = "none", +) -> torch.Tensor: + """ + Original implementation from + https://github.com/facebookresearch/fvcore/blob/master/fvcore/nn/focal_loss.py + FL* described in RetinaNet paper Appendix: https://arxiv.org/abs/1708.02002. + Args: + inputs: A float tensor of arbitrary shape. + The predictions for each example. + targets: A float tensor with the same shape as inputs. Stores the binary + classification label for each element in inputs + (0 for the negative class and 1 for the positive class). + alpha: (optional) Weighting factor in range (0,1) to balance + positive vs negative examples. Default = -1 (no weighting). + gamma: Gamma parameter described in FL*. Default = 1 (no weighting). + reduction: 'none' | 'mean' | 'sum' + 'none': No reduction will be applied to the output. + 'mean': The output will be averaged. + 'sum': The output will be summed. + Returns: + Loss tensor with the reduction option applied. + """ + shifted_inputs = gamma * (inputs * (2 * targets - 1)) + loss = -(F.logsigmoid(shifted_inputs)) / gamma + + if alpha >= 0: + alpha_t = alpha * targets + (1 - alpha) * (1 - targets) + loss *= alpha_t + + if reduction == "mean": + loss = loss.mean() + elif reduction == "sum": + loss = loss.sum() + + return loss diff --git a/scripts/training/oguz/huggingface-multihead/model.py b/scripts/training/oguz/huggingface-multihead/model.py index 8d767c9..7f64e47 100644 --- a/scripts/training/oguz/huggingface-multihead/model.py +++ b/scripts/training/oguz/huggingface-multihead/model.py @@ -1,4 +1,4 @@ -from typing import List, Optional +from typing import List, Optional, Union import numpy as np import torch @@ -6,91 +6,97 @@ from utils import build_mlp +ZERO_SIGMOID_INVERSE = -10 + +"""Models here only work with flattened datasets.""" + + +class MultiTargetTransformer(torch.nn.Module): + """Multi-target MLP classifier head that is able to handle group structure in + multi-label classifications problems (e.g., 6 label classification with two groups: + [A, B, C], [D, E, F]). + + + Args: + num_heads: Number of classification groups. + num_classes: List of number of classes in each group. + num_layers: Depth of MLP classfier heads. + dropout: Rate of dropout in tranformer output before MLP classifiers. + iterative: Adds an additional classification head for coarser _group_ task. + Only relevant if the task involves (coarse, fine-grained) labels. + If enabled, an additional classifier is first used to predict the coarse + label and the other heads predict the coarse label. The coarse classifier + acts as a filter, i.e., if a negative prediction occurs for a coarse label, + all predictions for labels in that group are set to high negative values. + use_gt_training: uses ground truth group values in the training + Only relevant if iterative is set to True. + backbone_dim: dimension of the backbone transformer + Set if the dimension is not accessible through the config of backbone. + """ -class MultiHeadTransformer(torch.nn.Module): def __init__( self, - backbone: PreTrainedModel, - num_heads: int, - num_classes: List[int], + num_classes: List[int] = [1], num_layers: int = 1, - dropout: float = 0.3, - pooling: bool = False, - freeze_backbone: bool = False, iterative: bool = False, use_gt_training: bool = True, backbone_dim: Optional[int] = None, ): super().__init__() - self.pooling = pooling + self.iterative = iterative self.use_gt_training = use_gt_training - - self.backbone = backbone - self.backbone.config.problem_type = "multi_label_classification" - self.backbone.trainable = not freeze_backbone - - if not hasattr(self.backbone.config, "dim"): - assert backbone_dim is not None, "Model config does not include output dim!" - dim = backbone_dim - else: - dim = self.backbone.config.dim - - self.dropout = torch.nn.Dropout(dropout) self.heads = torch.nn.ModuleList() mlp_params = { "depth": num_layers, - "in_features": dim, + "in_features": backbone_dim, "bias": True, "batchnorm": False, "final_norm": False, } - if iterative: + if self.iterative: self.heads.append( build_mlp( - middle_features=np.floor(np.sqrt(len(num_classes) * dim)).astype(int), + middle_features=np.floor(np.sqrt(len(num_classes) * backbone_dim)).astype(int), out_features=len(num_classes), **mlp_params ) ) - for i in range(num_heads): + for num_cls in num_classes: self.heads.append( build_mlp( - middle_features=np.floor(np.sqrt(num_classes[i] * dim)).astype(int), - out_features=num_classes[i], + middle_features=np.floor(np.sqrt(num_cls * backbone_dim)).astype(int), + out_features=num_cls, **mlp_params ) ) - def forward(self, inputs, gt_groups=None): - # get hidden representation - backbone_outputs = self.backbone(**inputs) - if self.pooling: - last_hidden_states = torch.mean(backbone_outputs.last_hidden_state, axis=1) - else: - last_hidden_states = backbone_outputs.last_hidden_state[:, 0, :] - hidden = self.dropout(last_hidden_states) - + def forward(self, inputs, gt_groups=None, group_threshold=0.5): if self.iterative: # execute super-classification task - out_groups = self.heads[0](hidden) + out_groups = self.heads[0](inputs) - # get sample groups - # TODO: dynamic threshold? - groups = gt_groups if self.training and self.use_gt_training else out_groups > 0.5 + # get group predictions + # TODO: dynamic threshold (per group?) + groups = ( + gt_groups + if self.training and self.use_gt_training + else out_groups > group_threshold + ) # execute each classification task out_targets = [] for i, head in enumerate(self.heads[1:]): - out_target = head(hidden) + out_target = head(inputs) out_targets.append( torch.where( torch.repeat_interleave(groups[:, i : i + 1], out_target.shape[1], dim=1), - out_target, - torch.zeros_like(out_target), + out_target, # classifer output if group is predicted as `positive` + torch.zeros_like(out_target) + + ZERO_SIGMOID_INVERSE, # zero if group is predicted as `negative` ) ) out_targets = torch.cat(out_targets, axis=-1) @@ -98,7 +104,103 @@ def forward(self, inputs, gt_groups=None): # execute each classification task out_targets = [] for head in self.heads: - out_targets.append(head(hidden)) + out_targets.append(head(inputs)) out_targets = torch.cat(out_targets, axis=-1) return (out_groups, out_targets) if self.iterative else out_targets + + +class MultiHeadTransformer(torch.nn.Module): + """Multi-task classifier each supporting multi-target groups (MultiTargetTransformer) + using the same transformer backbone. + + Args: + backbone: Pre-trained transformer. + num_classes: List of number of classes in each task. + num_layers: Depth of MLP classfier heads. + dropout: Rate of dropout in tranformer output before MLP classifiers. + pooling: If true, classifiers use averaged representations of all symbols. + If false, classifiers use representation of the start symbol. + freeze_backbone: Only train classifiers with backbone. + iterative: Adds an additional classification head for coarser _group_ task. + Only relevant if the task involves (coarse, fine-grained) labels. + If enabled, an additional classifier is first used to predict the coarse + label and the other heads predict the coarse label. The coarse classifier + acts as a filter, i.e., if a negative prediction occurs for a coarse label, + all predictions for labels in that group are set to high negative values. + use_gt_training: uses ground truth group values in the training + Only relevant if iterative is set to True. + backbone_dim: dimension of the backbone transformer + Set if the dimension is not accessible through the config of backbone. + """ + + def __init__( + self, + backbone: PreTrainedModel, + num_classes: Optional[Union[List[int], List[List[int]]]], + num_layers: int = 1, + dropout: float = 0.3, + pooling: bool = False, + freeze_backbone: bool = False, + iterative: bool = False, + use_gt_training: bool = True, + backbone_dim: Optional[int] = None, + ): + super().__init__() + self.pooling = pooling + self.iterative = iterative + self.use_gt_training = use_gt_training + + self.backbone = backbone + self.backbone.config.problem_type = "multi_label_classification" + self.backbone.trainable = not freeze_backbone + + if not hasattr(self.backbone.config, "dim"): + assert backbone_dim is not None, "Model config does not include output dim!" + dim = backbone_dim + else: + dim = self.backbone.config.dim + + if isinstance(num_classes[0], int): + num_classes = [num_classes] + + self.dropout = torch.nn.Dropout(dropout) + self.heads = torch.nn.ModuleList() + + for num_cls in num_classes: + self.heads.append( + MultiTargetTransformer( + num_classes=num_cls, + num_layers=num_layers, + iterative=iterative, + use_gt_training=use_gt_training, + backbone_dim=dim, + ) + ) + + def forward(self, inputs, gt_groups=None, group_threshold=0.5): + # get hidden representation + backbone_outputs = self.backbone(**inputs) + if self.pooling: + last_hidden_states = torch.mean(backbone_outputs.last_hidden_state, axis=1) + else: + last_hidden_states = backbone_outputs.last_hidden_state[:, 0, :] + hidden = self.dropout(last_hidden_states) + + # execute forward-pass for all heads + groups, targets = [], [] + for idx, head in enumerate(self.heads): + if self.iterative: + out_groups, out_targets = head( + hidden, + gt_groups=gt_groups[idx] if isinstance(gt_groups, list) else None, + group_threshold=group_threshold[idx] + if isinstance(group_threshold, list) + else group_threshold, + ) + groups.append(out_groups) + else: + out_targets = head(hidden) + targets.append(out_targets) + + return (groups, targets) if self.iterative else targets diff --git a/scripts/training/oguz/huggingface-multihead/requirements.txt b/scripts/training/oguz/huggingface-multihead/requirements.txt index 91206f3..f650a41 100644 --- a/scripts/training/oguz/huggingface-multihead/requirements.txt +++ b/scripts/training/oguz/huggingface-multihead/requirements.txt @@ -1,2 +1,5 @@ transformers==4.6.1 -mlflow==1.12.1 \ No newline at end of file +mlflow==1.18.0 +sagemaker==2.49.1 +s3fs==2021.07.0 +smdebug==1.0.11 diff --git a/scripts/training/oguz/huggingface-multihead/sagemaker_infer.py b/scripts/training/oguz/huggingface-multihead/sagemaker_infer.py new file mode 100644 index 0000000..7755679 --- /dev/null +++ b/scripts/training/oguz/huggingface-multihead/sagemaker_infer.py @@ -0,0 +1,74 @@ +import os +import sys +import argparse + +# import main folder for imports +sys.path.append(os.path.abspath(os.getcwd())) + +import pandas as pd + +import sagemaker +from sagemaker.pytorch import PyTorch + +from deep.constants import DEV_BUCKET, SAGEMAKER_ROLE +from deep.utils import formatted_time + +# get args +parser = argparse.ArgumentParser() +parser.add_argument( + "--task", + type=str, + default="1D", + choices=["1D", "2D"], +) +parser.add_argument("--dataset", type=str, default=None, required=True) +parser.add_argument("--model_uri", type=str, default=None, required=True) +parser.add_argument("--source", type=str, default="excerpt") +parser.add_argument("--debug", action="store_true", default=False) +args, _ = parser.parse_known_args() + +# create SageMaker session +sess = sagemaker.Session(default_bucket=DEV_BUCKET.name) + +# job and experiment names +job_name = f"{args.task}-infer-{formatted_time()}" + +# load dataset +infer_df = pd.read_csv(args.dataset) +infer_df.rename(columns={args.source: "excerpt"}, inplace=True) +if args.debug: + infer_df = infer_df.sample(n=1000) + +# upload dataset to s3 +input_path = DEV_BUCKET / "inference" / "input_data" / job_name # Do not change this +infer_path = str(input_path / "infer_df.pickle") + +infer_df.to_pickle( + infer_path, protocol=4 +) # protocol 4 is necessary, since SageMaker uses python 3.6 + +# hyperparameters for inference +hyperparameters = { + "model_uri": args.model_uri, +} + +# create SageMaker estimator +estimator = PyTorch( + entry_point="infer.py", + source_dir=str("scripts/training/oguz/huggingface-multihead"), + output_path=str(DEV_BUCKET / "predictions/"), + code_location=str(input_path), + instance_type="ml.p3.2xlarge", + instance_count=1, + role=SAGEMAKER_ROLE, + framework_version="1.8", + py_version="py36", + hyperparameters=hyperparameters, + job_name=job_name, +) + +# set arguments +fit_arguments = {"train": str(input_path), "test": str(input_path)} + +# transform the estimator +estimator.fit(fit_arguments, job_name=job_name) diff --git a/scripts/training/oguz/huggingface-multihead/run.py b/scripts/training/oguz/huggingface-multihead/sagemaker_train.py similarity index 81% rename from scripts/training/oguz/huggingface-multihead/run.py rename to scripts/training/oguz/huggingface-multihead/sagemaker_train.py index 2baf198..88801ca 100644 --- a/scripts/training/oguz/huggingface-multihead/run.py +++ b/scripts/training/oguz/huggingface-multihead/sagemaker_train.py @@ -26,32 +26,28 @@ # create SageMaker session sess = sagemaker.Session(default_bucket=DEV_BUCKET.name) -job_name = f"{args.task}-test-{formatted_time()}" +job_name = f"{args.task}-train-{formatted_time()}" # load dataset -dataset_version = "0.5" if args.task == "1D" else "0.4.4" -target_field = "subpillars_1d" if args.task == "1D" else "subpillars" -train_df = pd.read_csv( - f"data/frameworks_data/data_v{dataset_version}/data_v{dataset_version}_train.csv" -) -val_df = pd.read_csv( - f"data/frameworks_data/data_v{dataset_version}/data_v{dataset_version}_val.csv" -) +dataset_version = "0.7.1" +target_field = "subpillars_1d" if args.task == "1D" else "subpillars_2d" +train_df = pd.read_csv(f"data/frameworks_data/data_v{dataset_version}/train_v{dataset_version}.csv") +test_df = pd.read_csv(f"data/frameworks_data/data_v{dataset_version}/test_v{dataset_version}.csv") # resample if debug if args.debug: train_df = train_df.sample(n=1000) - val_df = val_df.sample(n=1000) + test_df = test_df.sample(n=1000) # upload dataset to s3 input_path = DEV_BUCKET / "training" / "input_data" / job_name # Do not change this train_path = str(input_path / "train_df.pickle") -val_path = str(input_path / "test_df.pickle") +test_path = str(input_path / "test_df.pickle") train_df.to_pickle( train_path, protocol=4 ) # protocol 4 is necessary, since SageMaker uses python 3.6 -val_df.to_pickle(val_path, protocol=4) +test_df.to_pickle(test_path, protocol=4) # hyperparameters for the run hyperparameters = { @@ -66,6 +62,8 @@ "num_layers": 1, "split": target_field, "target": target_field, + "weighting": "inverse", + "learning_rate": 5e-5, } # create SageMaker estimator diff --git a/scripts/training/oguz/huggingface-multihead/test.json b/scripts/training/oguz/huggingface-multihead/test.json new file mode 100644 index 0000000..b9cd937 --- /dev/null +++ b/scripts/training/oguz/huggingface-multihead/test.json @@ -0,0 +1,42 @@ +[ + { + "predictions": { + "subpillars_2d": { + "subpillar #1": 0.25, + "subpillar #2": 0.75 + }, + "subpillars_1d": { + "subpillar #1": 0.25, + "subpillar #2": 0.75 + } + }, + "confidence": { + "subpillars_2d": { + "subpillar #1": 1.0, + "subpillar #2": 0.8 + }, + "subpillars_1d": { + "subpillar #1": 1.0, + "subpillar #2": 0.8 + } + } + }, + { + "predictions": { + "subpillars_2d": { + "subpillar #1": 0.25, + "subpillar #2": 0.75 + } + }, + "confidence": { + "subpillars_2d": { + "subpillar #1": 1.0, + "subpillar #2": 0.8 + }, + "subpillars_1d": { + "subpillar #1": 1.0, + "subpillar #2": 0.8 + } + } + } +] diff --git a/scripts/training/oguz/huggingface-multihead/train.py b/scripts/training/oguz/huggingface-multihead/train.py index b99aa14..992df0d 100644 --- a/scripts/training/oguz/huggingface-multihead/train.py +++ b/scripts/training/oguz/huggingface-multihead/train.py @@ -2,26 +2,27 @@ import logging import os import sys +import json import mlflow import pandas as pd - -from sklearn.metrics import accuracy_score, precision_recall_fscore_support from transformers import AutoModel, AutoTokenizer, TrainingArguments -from constants import PILLARS_1D, SUBPILLARS_1D, PILLARS_2D, SUBPILLARS_2D +from constants import SECTORS, PILLARS_1D, SUBPILLARS_1D, PILLARS_2D, SUBPILLARS_2D from data import MultiHeadDataFrame from model import MultiHeadTransformer +from wrapper import MLFlowWrapper from trainer import MultiHeadTrainer -from utils import str2bool, str2list +from evaluation import compute_multihead_metrics, compute_multitarget_metrics, _prefix +from utils import str2bool, str2list, get_conda_env_specs if __name__ == "__main__": parser = argparse.ArgumentParser() # hyperparameters sent by the client parser.add_argument("--epochs", type=int, default=3) - parser.add_argument("--train-batch-size", type=int, default=32) - parser.add_argument("--eval-batch-size", type=int, default=64) + parser.add_argument("--train_batch_size", type=int, default=32) + parser.add_argument("--eval_batch_size", type=int, default=64) parser.add_argument("--warmup_steps", type=int, default=500) parser.add_argument("--learning_rate", type=str, default=5e-5) parser.add_argument("--dropout", type=float, default=0.3) @@ -32,23 +33,17 @@ "--loss", type=str, default="ce", - choices=["ce", "focal"], - help="Loss function: 'ce', 'focal'", + choices=["ce", "focal", "focal_star"], + help="Loss function: 'ce', 'focal', 'focal_star'", + ) + parser.add_argument( + "--weighting", type=str, default=None, choices=["inverse", "inverse_square"] ) parser.add_argument( "--target", - type=str, + type=str2list, default="subpillars_1d", - choices=[ - "pillars", - "subpillars", - "pillars_1d", - "subpillars_1d", - "pillars_2d", - "subpillars_2d", - "sectors", - ], - help="Prediction target", + help="Prediction targets", ) parser.add_argument("--split", type=str2list, default="subpillars_1d") parser.add_argument("--iterative", type=str2bool, default=False) @@ -59,10 +54,11 @@ # MLFlow related parameters parser.add_argument("--tracking_uri", type=str) parser.add_argument("--experiment_name", type=str) + parser.add_argument("--deploy", type=str2bool, default=True) # SageMaker parameters - data, model, and output directories - parser.add_argument("--output-data-dir", type=str, default=os.environ["SM_OUTPUT_DATA_DIR"]) - parser.add_argument("--model-dir", type=str, default=os.environ["SM_MODEL_DIR"]) + parser.add_argument("--output_data_dir", type=str, default=os.environ["SM_OUTPUT_DATA_DIR"]) + parser.add_argument("--model_dir", type=str, default=os.environ["SM_MODEL_DIR"]) parser.add_argument("--n_gpus", type=str, default=os.environ["SM_NUM_GPUS"]) parser.add_argument("--training_dir", type=str, default=os.environ["SM_CHANNEL_TRAIN"]) parser.add_argument("--test_dir", type=str, default=os.environ["SM_CHANNEL_TEST"]) @@ -88,21 +84,29 @@ backbone = AutoModel.from_pretrained(args.model_name) # get target groups - if args.target == "subpillars_1d": - groups = SUBPILLARS_1D - group_names = PILLARS_1D - elif args.target == "subpillars" or args.target == "subpillars_2d": - groups = SUBPILLARS_2D - group_names = PILLARS_2D - else: - groups = None - group_names = None + targets, groups, group_names = [], [], [] + for target in args.target: + if target == "subpillars_1d": + groups.append(SUBPILLARS_1D) + group_names.append(PILLARS_1D) + elif target == "subpillars_2d": + groups.append(SUBPILLARS_2D) + group_names.append(PILLARS_2D) + elif target == "sectors": + groups.append([SECTORS]) + group_names.append(["Sectors"]) + else: + raise NotImplementedError + targets.append(target) + + # sanity check for iterative option + if args.iterative: + assert groups is not None, "Provide groups for the 'iterative' option" # build classifier model from backbone model = MultiHeadTransformer( backbone, - num_heads=len(groups), - num_classes=[len(group) for group in groups], + num_classes=[[len(group) for group in _group] for _group in groups], num_layers=args.num_layers, dropout=args.dropout, pooling=args.pooling, @@ -117,110 +121,70 @@ train_df, tokenizer=tokenizer, source="excerpt", - target=args.target, + targets=targets, groups=groups, group_names=group_names, + exclude=["NOT_MAPPED"], filter=args.split, + iterative=args.iterative, flatten=True, ) test_dataset = MultiHeadDataFrame( test_df, tokenizer=tokenizer, source="excerpt", - target=args.target, + targets=targets, groups=groups, group_names=group_names, - filter=args.split, + exclude=["NOT_MAPPED"], + filter=None, + iterative=args.iterative, flatten=True, ) # compute metrics function for multi-class classification def compute_metrics(pred, threshold=0.5): - # add prefix to dictionary keys - def _prefix(dic, prefix): - return {(prefix + k): v for k, v in dic.items()} - - # compute metrics given preds and labels - def _compute(preds, labels, average="micro"): - preds = preds > threshold - precision, recall, f1, _ = precision_recall_fscore_support( - labels, preds, average=average - ) - accuracy = accuracy_score(labels, preds) - return { - "accuracy": accuracy, - "f1": f1, - "precision": precision, - "recall": recall, - } - - # process pillar texts for MLFlow - def _process(text): - text = text.lower() - text = text.replace(" ", "_") - text = text.replace(">", "") - text = text.replace("&", "_") - return text - - metrics = {} - + # read predictions and labels if args.iterative: # TODO: ensure the ordering is stable preds, preds_group = pred.predictions labels, labels_group = pred.label_ids - - # group micro evaluation - metrics.update(_prefix(_compute(preds_group, labels_group, "micro"), "pillar_micro_")) - # group macro evaluation - metrics.update(_prefix(_compute(preds_group, labels_group, "macro"), "pillar_macro_")) - # per group evaluation - for i, pillar in enumerate(group_names): - metrics.update( - _prefix( - _compute(preds_group[:, i], labels_group[:, i], "binary"), - f"{_process(pillar)}_binary_", - ) - ) else: - labels = pred.label_ids preds = pred.predictions + labels = pred.label_ids - # micro evaluation - metrics.update(_prefix(_compute(preds, labels, "micro"), "subpillar_micro_")) - # macro evaluation - metrics.update(_prefix(_compute(preds, labels, "macro"), "subpillar_macro_")) - # per head evaluation - idx = 0 - for i, pillar in enumerate(group_names): - idx_end = idx + len(groups[i]) - - # per head micro evaluation - metrics.update( - _prefix( - _compute(preds[:, idx:idx_end], labels[:, idx:idx_end], "micro"), - f"{_process(pillar)}_micro_", - ) - ) - # per head macro evaluation + if not (isinstance(preds, list) or isinstance(preds, tuple)): + preds = [preds] + labels = [labels] + + if args.iterative: + preds_group = [preds_group] + labels_group = [labels_group] + + # compute metrics for preds + metrics = compute_multihead_metrics( + preds, + labels, + groups=groups, + group_names=group_names, + targets=targets, + threshold=threshold, + ) + + # compute metrics for group preds + if args.iterative: metrics.update( _prefix( - _compute(preds[:, idx:idx_end], labels[:, idx:idx_end], "macro"), - f"{_process(pillar)}_macro_", - ) - ) - - # per head target evaluation - for j, subpillar in enumerate(groups[i]): - metrics.update( - _prefix( - _compute(preds[:, idx + j], labels[:, idx + j], "binary"), - f"{_process(subpillar)}_binary_", + compute_multitarget_metrics( + preds_group, + labels_group, + groups=group_names, + group_names=targets, + threshold=threshold, ) - ) - - # update index - idx = idx_end - + ), + "group_", + ) return metrics # define training args @@ -231,12 +195,36 @@ def _process(text): per_device_eval_batch_size=args.eval_batch_size, warmup_steps=args.warmup_steps, evaluation_strategy="epoch", + save_strategy="epoch", logging_dir=f"{args.output_data_dir}/logs", learning_rate=float(args.learning_rate), skip_memory_metrics=False, - label_names=["labels", "groups"] if args.iterative else ["labels"], + label_names=train_dataset.label_names(), + metric_for_best_model="eval_subpillars_1d_micro_f1", + greater_is_better=True, + load_best_model_at_end=True, + save_total_limit=1, ) + # calculate weighting coefficients + loss_weights, loss_pos_weights = None, None + if args.weighting == "inverse": + stats = train_dataset.compute_stats() + + classes = [[target for group in _groups for target in group] for _groups in groups] + loss_weights = [ + [ + 0 if stats[_target][c] else (stats[_target]["ALL"] / stats[_target][c]) + for c in _classes + ] + for _target, _classes in zip(targets, classes) + ] + loss_pos_weights = [ + [weight - 1 for _loss_weights in loss_weights for weight in _loss_weights] + ] + if args.weighting == "inverse_square": + raise NotImplementedError + # create trainer instance trainer = MultiHeadTrainer( model=model, @@ -244,7 +232,10 @@ def _process(text): compute_metrics=compute_metrics, train_dataset=train_dataset, eval_dataset=test_dataset, - focal_loss=(args.loss == "focal"), + loss_fn=args.loss, + loss_weights=loss_weights, + loss_pos_weights=loss_pos_weights, + targets=targets, ) # set env variable for MLFlow artifact logging @@ -263,22 +254,99 @@ def _process(text): eval_result = trainer.evaluate(eval_dataset=test_dataset) # write eval result to file which can be accessed later in s3 ouput - with open(os.path.join(args.output_data_dir, "eval_results.txt"), "w") as writer: - print("***** Eval results *****") + logging.info("Logging eval results") + eval_file = os.path.join(args.output_data_dir, "eval_results.txt") + with open(eval_file, "w") as writer: for key, value in sorted(eval_result.items()): writer.write(f"{key} = {value}\n") + mlflow.log_artifact(eval_file) - # write eval result to MLFlow + # write eval result to mlflow for key, value in sorted(eval_result.items()): mlflow.log_metric(key, value) + # get labels + logging.info("Logging model labels") + if groups is not None: + labels = [label for gs in groups for label in gs] + elif args.target == "sectors": + labels = SECTORS + else: + labels = None + + # output labels artifact + label_file = os.path.join(args.output_data_dir, "labels.txt") + with open(label_file, "w") as writer: + for label in labels: + writer.write(f"{label}\n") + artifacts = {"labels": label_file} + + if args.iterative: + # get gorups + logging.info("Logging model groups") + labels = [label for label in group_names] + + # output groups artifact + group_file = os.path.join(args.output_data_dir, "groups.txt") + with open(group_file, "w") as writer: + for label in labels: + writer.write(f"{label}\n") + artifacts.update({"groups": group_file}) + + # output target names artifact + logging.info("Logging target names") + target_file = os.path.join(args.output_data_dir, "targets.txt") + with open(target_file, "w") as writer: + writer.write(f"{args.target}\n") + artifacts.update({"targets": target_file}) + # log experiment params to MLFlow mlflow.log_params(vars(args)) - mlflow.log_params({"groups": groups, "group_names": group_names}) - # set tags + # log tokenizer parameters + mlflow.log_params(train_dataset.tokenizer_options) + + # set experiment tags mlflow.set_tags({"split": args.split, "iterative": args.iterative}) + # output inference artifact + logging.info("Logging inference params") + infer_file = os.path.join(args.output_data_dir, "infer_params.json") + with open(infer_file, "w") as writer: + json.dump( + { + "dataset": { + "source": "excerpt", + "target": args.target, + "flatten": True, + }, + "dataloader": { + "batch_size": args.eval_batch_size, + "shuffle": False, + "num_workers": 0, + }, + "threshold": {"group": 0.5, "target": 0.5}, + "output": {"probs_only": True, "flatten": True}, + }, + writer, + ) + artifacts.update({"infer_params": infer_file}) + + if args.deploy: + # log model with an inference wrapper + logging.info("Logging deployment model") + code_path = os.path.abspath(os.path.dirname(__file__)) + + mlflow_wrapper = MLFlowWrapper(tokenizer, trainer.model) + mlflow.pyfunc.log_model( + python_model=mlflow_wrapper, + artifact_path="model", + registered_model_name="multi-head-transformer", + artifacts=artifacts, + conda_env=get_conda_env_specs(), + code_path=os.listdir(code_path), + ) + # finish mlflow run mlflow.end_run() except Exception as e: diff --git a/scripts/training/oguz/huggingface-multihead/trainer.py b/scripts/training/oguz/huggingface-multihead/trainer.py index 553766c..84d20e3 100644 --- a/scripts/training/oguz/huggingface-multihead/trainer.py +++ b/scripts/training/oguz/huggingface-multihead/trainer.py @@ -13,10 +13,20 @@ TrainingArguments, ) -from utils import sigmoid_focal_loss +from loss import sigmoid_focal_loss, sigmoid_focal_loss_star +"""Trainers here only work with flattened datasets.""" + + +class MultiTargetTrainer(Trainer): + """HuggingFace Trainer compatible with MultiTargetTransformer models. + + Args: + loss_fn: 'ce', 'focal', 'focal_star' + loss_weights: weighting applied to different classes + loss_pos_weights: weighting applied to positive versus negative instances + """ -class MultiHeadTrainer(Trainer): def __init__( self, model: Union[PreTrainedModel, nn.Module] = None, @@ -28,8 +38,13 @@ def __init__( model_init: Callable[[], PreTrainedModel] = None, compute_metrics: Optional[Callable[[EvalPrediction], Dict]] = None, callbacks: Optional[List[TrainerCallback]] = None, - optimizers: Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (None, None), - focal_loss: bool = False, + optimizers: Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = ( + None, + None, + ), + loss_fn: str = "ce", + loss_weights: Optional[List[float]] = None, + loss_pos_weights: Optional[List[float]] = None, ): super().__init__( model=model, @@ -43,13 +58,33 @@ def __init__( callbacks=callbacks, optimizers=optimizers, ) - self.loss_fn = sigmoid_focal_loss if focal_loss else torch.nn.BCEWithLogitsLoss() + if loss_fn == "ce": + if loss_weights: + loss_weights = torch.FloatTensor(loss_weights).to("cuda") + if loss_pos_weights: + loss_pos_weights = torch.FloatTensor(loss_pos_weights).to("cuda") + + self.loss_fn = torch.nn.BCEWithLogitsLoss( + weight=loss_weights, pos_weight=loss_pos_weights + ) + elif loss_fn == "focal": + assert ( + loss_weights is None and loss_pos_weights is None + ), "Does not support weighting with focal loss" + self.loss_fn = sigmoid_focal_loss + elif loss_fn == "focal_star": + assert ( + loss_weights is None and loss_pos_weights is None + ), "Does not support weighting with focal loss-star" + self.loss_fn = sigmoid_focal_loss_star + else: + raise "Unknown loss function" def compute_loss(self, model, inputs, return_outputs=False): labels = inputs.pop("labels") - groups = inputs.pop("groups") if model.iterative: + groups = inputs.pop("groups") pred_groups, pred_labels = model(inputs, gt_groups=groups) # calculate group loss @@ -73,3 +108,133 @@ def compute_loss(self, model, inputs, return_outputs=False): logits["logits_group"] = pred_groups return (loss, logits) if return_outputs else loss + + +class MultiHeadTrainer(Trainer): + """HuggingFace Trainer compatible with MultiHeadTransformer models. + + Args: + loss_fn: 'ce', 'focal', 'focal_star' + loss_weights: weighting applied to different classes + loss_pos_weights: weighting applied to positive versus negative instances + tasks: names of the tasks + """ + + def __init__( + self, + model: Union[PreTrainedModel, nn.Module] = None, + args: TrainingArguments = None, + data_collator: Optional[DataCollator] = None, + train_dataset: Optional[Dataset] = None, + eval_dataset: Optional[Dataset] = None, + tokenizer: Optional[PreTrainedTokenizerBase] = None, + model_init: Callable[[], PreTrainedModel] = None, + compute_metrics: Optional[Callable[[EvalPrediction], Dict]] = None, + callbacks: Optional[List[TrainerCallback]] = None, + optimizers: Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = ( + None, + None, + ), + loss_fn: str = "ce", + loss_weights: Optional[Union[List[float], List[List[float]]]] = None, + loss_pos_weights: Optional[Union[List[float], List[List[float]]]] = None, + targets: Optional[Union[str, List[str]]] = "", + ): + super().__init__( + model=model, + args=args, + data_collator=data_collator, + train_dataset=train_dataset, + eval_dataset=eval_dataset, + tokenizer=tokenizer, + model_init=model_init, + compute_metrics=compute_metrics, + callbacks=callbacks, + optimizers=optimizers, + ) + if loss_fn == "ce": + if loss_weights: + loss_weights = [ + torch.FloatTensor(_loss_weights).to("cuda") for _loss_weights in loss_weights + ] + else: + loss_weights = [None for _ in targets] + if loss_pos_weights: + loss_pos_weights = [ + torch.FloatTensor(_loss_pos_weights).to("cuda") + for _loss_pos_weights in loss_pos_weights + ] + else: + loss_pos_weights = [None for _ in targets] + + self.loss_fn = [ + torch.nn.BCEWithLogitsLoss(weight=_loss_weights, pos_weight=_loss_pos_weights) + for (_loss_weights, _loss_pos_weights) in zip(loss_weights, loss_pos_weights) + ] + elif loss_fn == "focal": + assert ( + loss_weights is None and loss_pos_weights is None + ), "Does not support weighting with focal loss" + self.loss_fn = sigmoid_focal_loss + elif loss_fn == "focal_star": + assert ( + loss_weights is None and loss_pos_weights is None + ), "Does not support weighting with focal loss-star" + self.loss_fn = sigmoid_focal_loss_star + else: + raise "Unknown loss function" + + if isinstance(targets, str): + self.targets = [""] + else: + self.targets = targets + + def compute_loss(self, model, inputs, return_outputs=False): + # collect labels + labels = [] + for _target in self.targets: + labels.append(inputs.pop(f"{_target}_labels")) + + # get the model predictions + if model.iterative: + # collect group gts + groups = [] + for target in self.targets: + groups.append(inputs.pop(f"{target}_groups")) + pred_groups, pred_labels = model(inputs, gt_groups=groups) + else: + pred_labels = model(inputs) + + logits = {} + loss = torch.tensor(0) + for idx, _target in enumerate(self.targets): + # get labels, preds and register logits + _labels = labels[idx] + _pred_labels = pred_labels[idx] + logits[f"{_target}_logits"] = _pred_labels + + # calculate label loss + _loss = self.loss_fn[idx]( + _pred_labels.view(-1, _labels.shape[-1]), + _labels.view(-1, _labels.shape[-1]).float(), + ) + + if model.iterative: + # get groups, preds and register logits + _groups = inputs.pop(f"{_target}_groups") + _pred_groups = pred_groups[idx] + logits[f"{_target}_logits_group"] = _pred_groups + + # calculate group loss + _loss_group = self.loss_fn[idx]( + _pred_groups.view(-1, _groups.shape[-1]), + _groups.view(-1, _groups.shape[-1]).float(), + ) + + # add losses together + _loss = _loss + _loss_group + + # sum losses across all tasks + loss = loss + _loss + + return (loss, logits) if return_outputs else loss diff --git a/scripts/training/oguz/huggingface-multihead/utils.py b/scripts/training/oguz/huggingface-multihead/utils.py index 9effce9..8861f31 100644 --- a/scripts/training/oguz/huggingface-multihead/utils.py +++ b/scripts/training/oguz/huggingface-multihead/utils.py @@ -1,9 +1,13 @@ from collections import OrderedDict from typing import Dict + import argparse +from pathlib import Path + +import pandas as pd import torch -import torch.nn.functional as F +import mlflow def revdict(d: Dict): @@ -36,6 +40,30 @@ def str2list(v, sep=","): raise argparse.ArgumentTypeError("String value expected.") +def get_conda_env_specs(): + requirement_file = str(Path(__file__).parent / "requirements.txt") + with open(requirement_file, "r") as f: + requirements = f.readlines() + requirements = [x.replace("\n", "") for x in requirements] + + default_env = mlflow.pytorch.get_default_conda_env() + pip_dependencies = default_env["dependencies"][2]["pip"] + pip_dependencies.extend(requirements) + return default_env + + +def read_dataframe(path: str, **kwargs): + """Reads a Pandas DataFrame respecting the file extension""" + + if path.endswith(".pickle"): + return pd.read_pickle(path, **kwargs) + if path.endswith(".csv"): + return pd.read_csv(path, **kwargs) + if path.endswith(".xlsx"): + return pd.read_excel(path, **kwargs) + raise "Unknown data format" + + def build_mlp( depth: int, in_features: int, @@ -76,49 +104,3 @@ def build_mlp( # return network return torch.nn.Sequential(OrderedDict(layers)) - - -def sigmoid_focal_loss( - inputs: torch.Tensor, - targets: torch.Tensor, - alpha: float = 0.25, - gamma: float = 2, - reduction: str = "mean", -): - """ - Original implementation from - https://github.com/facebookresearch/fvcore/blob/master/fvcore/nn/focal_loss.py - Loss used in RetinaNet for dense detection: https://arxiv.org/abs/1708.02002. - - Args: - inputs: A float tensor of arbitrary shape. - The predictions for each example. - targets: A float tensor with the same shape as inputs. Stores the binary - classification label for each element in inputs - (0 for the negative class and 1 for the positive class). - alpha: (optional) Weighting factor in range (0,1) to balance - positive vs negative examples or -1 for ignore. Default = 0.25 - gamma: Exponent of the modulating factor (1 - p_t) to - balance easy vs hard examples. - reduction: 'none' | 'mean' | 'sum' - 'none': No reduction will be applied to the output. - 'mean': The output will be averaged. - 'sum': The output will be summed. - Returns: - Loss tensor with the reduction option applied. - """ - p = torch.sigmoid(inputs) - ce_loss = F.binary_cross_entropy_with_logits(inputs, targets, reduction="none") - p_t = p * targets + (1 - p) * (1 - targets) - loss = ce_loss * ((1 - p_t) ** gamma) - - if alpha >= 0: - alpha_t = alpha * targets + (1 - alpha) * (1 - targets) - loss = alpha_t * loss - - if reduction == "mean": - loss = loss.mean() - elif reduction == "sum": - loss = loss.sum() - - return loss diff --git a/scripts/training/oguz/huggingface-multihead/wrapper.py b/scripts/training/oguz/huggingface-multihead/wrapper.py new file mode 100644 index 0000000..60b2fad --- /dev/null +++ b/scripts/training/oguz/huggingface-multihead/wrapper.py @@ -0,0 +1,161 @@ +import json + +import numpy as np +import torch +import mlflow +from torch.utils.data import DataLoader + +from data import MultiHeadDataFrame + + +def _extract_predictions(probs, threshold): + """Extracts predictions from probabilities and threshold""" + probs = probs > threshold + + preds = [] + for i in range(probs.shape[0]): + preds.append(np.nonzero(probs[i, :])[0].tolist()) + return preds + + +class MLFlowWrapper(mlflow.pyfunc.PythonModel): + """MLFlow Wrapper class for inference""" + + def __init__(self, tokenizer, model): + self.tokenizer = tokenizer + self.model = model.eval() + super().__init__() + + def load_context(self, context): + # process labels + with open(context.artifacts["labels"], "r") as f: + self.labels = [line.strip() for line in f.readlines()] + + # process groups + if self.model.iterative: + with open(context.artifacts["groups"], "r") as f: + self.groups = [line.strip() for line in f.readlines()] + + # process target name + with open(context.artifacts["targets"], "r") as f: + self.targets = [line.strip() for line in f.readlines()] + + # process inference params + with open(context.artifacts["infer_params"], "r") as f: + self.infer_params = json.load(f) + + # sanity checks for dataset params + dataset_params = self.infer_params["dataset"] + assert "filter" not in dataset_params, "Can't use a filter in an inference dataset!" + + if "inference" in dataset_params: + assert dataset_params["inference"], "Can only use an inference dataset!" + dataset_params.pop("inference") + + # sanity check for output params + output_params = self.infer_params["output"] + assert ( + output_params["flatten"] and output_params["probs_only"] + ), "Flattened output is only supported when preds_only is enabled" + + def predict(self, context, model_input): + # get dataset and data loader + dataset = MultiHeadDataFrame( + model_input, + tokenizer=self.tokenizer, + filter=None, + inference=True, + **self.infer_params["dataset"], + ) + dataloader = DataLoader(dataset, **self.infer_params["dataloader"]) + + # containers for logits + probs_targets = [] + if self.model.iterative: + probs_groups = [] + + # forward pass + with torch.no_grad(): + for batch in dataloader: + for k, v in batch.items(): + batch[k] = v.to("cuda") + + if self.model.iterative: + batch_groups, batch_targets = self.model.forward( + batch, group_threshold=self.infer_params["threshold"]["group"] + ) + batch_groups = torch.sigmoid(batch_groups) + batch_targets = torch.sigmoid(batch_targets) + probs_groups.append(batch_groups.detach().cpu().numpy()) + else: + batch_targets = torch.sigmoid(self.model.forward(batch)) + probs_targets.append(batch_targets.detach().cpu().numpy()) + + probs_targets = np.concatenate(probs_targets, axis=0) + preds_targets = _extract_predictions( + probs_targets, self.infer_params["threshold"]["target"] + ) + + if self.infer_params["output"]["flatten"]: + # prepare flattened output + output = [ + {self.labels[j]: probs_targets[i, j] for j in range(probs_targets.shape[1])} + for i in range(probs_targets.shape[0]) + ] + else: + # put probabilities inside a nested dictionary + output = [ + { + "probabilities": { + self.targets[0]: { + self.labels[j]: probs_targets[i, j] + for j in range(probs_targets.shape[1]) + } + } + } + for i in range(probs_targets.shape[0]) + ] + + if not self.infer_params["output"]["probs_only"]: + # append predictions + output = [ + out.update({"predictions": {self.targets[0]: preds_targets[i]}}) + for i, out in enumerate(output) + ] + + if self.model.iterative: + probs_groups = np.concatenate(probs_groups, axis=0) + preds_groups = _extract_predictions( + probs_groups, self.infer_params["threshold"]["group"] + ) + + if self.infer_params["output"]["flatten"]: + # append group preds in a flattened format + output = [ + out.update( + {self.groups[j]: probs_groups[i, j] for j in range(probs_groups.shape[1])} + ) + for i, out in enumerate(output) + ] + else: + # update probabilities field for the group output + output = [ + out["probabilities"].update( + { + self.targets[1]: { + self.groups[j]: probs_groups[i, j] + for j in range(probs_groups.shape[1]) + } + } + ) + for i, out in enumerate(output) + ] + + if not self.infer_params["output"]["probs_only"]: + # append group predictions + output = [ + out["predictions"].update({self.targets[1]: preds_groups[i]}) + for i, out in enumerate(output) + ] + + return output