import abc import math import datasets import inspect import functools import numpy as np from abc import abstractmethod import sklearn def metric_max_over_ground_truths(metric_fn, prediction, ground_truths): """Compute max metric between prediction and each ground truth.""" scores_for_ground_truths = [] for ground_truth in ground_truths: score = metric_fn(prediction, ground_truth) scores_for_ground_truths.append(score) return max(scores_for_ground_truths) def perplexity(items): return math.exp(-mean(items)) def matthews_corrcoef(items): unzipped_list = list(zip(*items)) golds = unzipped_list[0] preds = unzipped_list[1] return sklearn.metrics.matthews_corrcoef(golds, preds) def f1_score(items): unzipped_list = list(zip(*items)) golds = unzipped_list[0] preds = unzipped_list[1] fscore = sklearn.metrics.f1_score(golds, preds) return np.max(fscore) def yesno(x): if x: return "yes" else: return "no" def acc_all(items): # Only count as correct if all answers are labeled correctly for each question question_scoring_dict = {} preds = list(zip(*items))[0] docs = list(zip(*items))[1] for doc, pred in zip(docs, preds): paragraph_id = doc["idx"]["paragraph"] question_id = doc["idx"]["question"] if (paragraph_id, question_id) not in question_scoring_dict: question_scoring_dict[(paragraph_id, question_id)] = [] gold_label = doc["label"] == 1 question_scoring_dict[(paragraph_id, question_id)].append(gold_label == pred) acc = np.mean([int(all(x)) for x in question_scoring_dict.values()]) return acc def positional_deprecated(fn): """ A decorator to nudge users into passing only keyword args (`kwargs`) to the wrapped function, `fn`. """ @functools.wraps(fn) def _wrapper(*args, **kwargs): if len(args) != 1 if inspect.ismethod(fn) else 0: print( f"WARNING: using {fn.__name__} with positional arguments is " "deprecated and will be disallowed in a future version of " "lm-evaluation-harness!" ) return fn(*args, **kwargs) return _wrapper def mean(arr): return sum(arr) / len(arr) def weighted_mean(items): a, b = zip(*items) return sum(a) / sum(b) def weighted_perplexity(items): return math.exp(-weighted_mean(items)) def bits_per_byte(items): return -weighted_mean(items) / math.log(2) class Task(abc.ABC): """A task represents an entire benchmark including its dataset, problems, answers, and evaluation methods. See BoolQ for a simple example implementation A `doc` can be any python object which represents one instance of evaluation. This is usually a dictionary e.g. {"question": ..., "answer": ...} or {"question": ..., question, answer) """ # The name of the `Task` benchmark as denoted in the HuggingFace datasets Hub # or a path to a custom `datasets` loading script. DATASET_PATH: str = None # The name of a subset within `DATASET_PATH`. DATASET_NAME: str = None def __init__(self, data_dir=None, cache_dir=None, download_mode=None): """ :param data_dir: str Stores the path to a local folder containing the `Task`'s data files. Use this to specify the path to manually downloaded data (usually when the dataset is not publicly accessible). :param cache_dir: str The directory to read/write the `Task` dataset. This follows the HuggingFace `datasets` API with the default cache directory located at: `~/.cache/huggingface/datasets` NOTE: You can change the cache location globally for a given process by setting the shell environment variable, `HF_DATASETS_CACHE`, to another directory: `export HF_DATASETS_CACHE="/path/to/another/directory"` :param download_mode: datasets.DownloadMode How to treat pre-existing `Task` downloads and data. - `datasets.DownloadMode.REUSE_DATASET_IF_EXISTS` Reuse download and reuse dataset. - `datasets.DownloadMode.REUSE_CACHE_IF_EXISTS` Reuse download with fresh dataset. - `datasets.DownloadMode.FORCE_REDOWNLOAD` Fresh download and fresh dataset. """ self.download(data_dir, cache_dir, download_mode) self._training_docs = None self._fewshot_docs = None def download(self, data_dir=None, cache_dir=None, download_mode=None): """Downloads and returns the task dataset. Override this method to download the dataset from a custom API. :param data_dir: str Stores the path to a local folder containing the `Task`'s data files. Use this to specify the path to manually downloaded data (usually when the dataset is not publicly accessible). :param cache_dir: str The directory to read/write the `Task` dataset. This follows the HuggingFace `datasets` API with the default cache directory located at: `~/.cache/huggingface/datasets` NOTE: You can change the cache location globally for a given process by setting the shell environment variable, `HF_DATASETS_CACHE`, to another directory: `export HF_DATASETS_CACHE="/path/to/another/directory"` :param download_mode: datasets.DownloadMode How to treat pre-existing `Task` downloads and data. - `datasets.DownloadMode.REUSE_DATASET_IF_EXISTS` Reuse download and reuse dataset. - `datasets.DownloadMode.REUSE_CACHE_IF_EXISTS` Reuse download with fresh dataset. - `datasets.DownloadMode.FORCE_REDOWNLOAD` Fresh download and fresh dataset. """ self.dataset = datasets.load_dataset( path=self.DATASET_PATH, name=self.DATASET_NAME, data_dir=data_dir, cache_dir=cache_dir, download_mode=download_mode, ) def should_decontaminate(self): """Whether this task supports decontamination against model training set.""" return False @abstractmethod def has_training_docs(self): """Whether the task has a training set""" pass @abstractmethod def has_validation_docs(self): """Whether the task has a validation set""" pass @abstractmethod def has_test_docs(self): """Whether the task has a test set""" pass def training_docs(self): """ :return: Iterable[obj] A iterable of any object, that doc_to_text can handle """ return [] def validation_docs(self): """ :return: Iterable[obj] A iterable of any object, that doc_to_text can handle """ return [] def test_docs(self): """ :return: Iterable[obj] A iterable of any object, that doc_to_text can handle """ return [] def _process_doc(self, doc): """ Override this to process (detokenize, strip, replace, etc.) individual documents. This can be used in a map over documents of a data split. E.g. `map(self._process_doc, self.dataset["validation"])` :return: dict The processed version of the specified `doc`. """ return doc def fewshot_examples(self, k, rnd): if self._training_docs is None: self._training_docs = list(self.training_docs()) return rnd.sample(self._training_docs, k) def doc_to_decontamination_query(self, doc): print( "Override doc_to_decontamination_query with document specific decontamination query." ) assert False @abstractmethod def doc_to_text(self, doc): pass @abstractmethod def doc_to_target(self, doc): pass @abstractmethod def construct_requests(self, doc, ctx): """Uses RequestFactory to construct Requests and returns an iterable of Requests which will be sent to the LM. :param doc: The document as returned from training_docs, validation_docs, or test_docs. :param ctx: str The context string, generated by fewshot_context. This includes the natural language description, as well as the few shot examples, and the question part of the document for `doc`. """ pass @abstractmethod def process_results(self, doc, results): """Take a single document and the LM results and evaluates, returning a dict where keys are the names of submetrics and values are the values of the metric for that one document :param doc: The document as returned from training_docs, validation_docs, or test_docs. :param results: The results of the requests created in construct_requests. """ pass @abstractmethod def aggregation(self): """ :returns: {str: [metric_score] -> float} A dictionary where keys are the names of submetrics and values are functions that aggregate a list of metric scores """ pass @abstractmethod def higher_is_better(self): """ :returns: {str: bool} A dictionary where keys are the names of submetrics and values are whether a higher value of the submetric is better """ pass def fewshot_description(self): import warnings warnings.warn( "`fewshot_description` will be removed in futures versions. Pass " "any custom descriptions to the `evaluate` function instead.", DeprecationWarning, ) return "" @positional_deprecated def fewshot_context( self, doc, num_fewshot, provide_description=None, rnd=None, description=None ): """Returns a fewshot context string that is made up of a prepended description (if provided), the `num_fewshot` number of examples, and an appended prompt example. :param doc: str The document as returned from training_docs, validation_docs, or test_docs. :param num_fewshot: int The number of fewshot examples to provide in the returned context string. :param provide_description: bool Not implemented, and this option is deprecated and will be removed in a future version in favor of a different description providing method :param rnd: random.Random The pseudo-random number generator used to randomly sample examples. WARNING: This is currently a required arg although it's optionalized with a default `None`. :param description: str The task's description that will be prepended to the fewshot examples. :returns: str The fewshot context. """ assert ( rnd is not None ), "A `random.Random` generator argument must be provided to `rnd`" assert not provide_description, ( "The `provide_description` arg will be removed in future versions. To prepend " "a custom description to the context, supply the corresponding string via the " "`description` arg." ) if provide_description is not None: # nudge people to not specify it at all print( "WARNING: provide_description is deprecated and will be removed in a future version in favor of description_dict" ) description = description + "\n\n" if description else "" if num_fewshot == 0: labeled_examples = "" else: # for sets with no training docs, draw from other set *but ensure no overlap with current doc* if self.has_training_docs(): fewshotex = self.fewshot_examples(k=num_fewshot, rnd=rnd) else: if self._fewshot_docs is None: self._fewshot_docs = list( self.validation_docs() if self.has_validation_docs() else self.test_docs() ) fewshotex = rnd.sample(self._fewshot_docs, num_fewshot + 1) # get rid of the doc that's the one we're evaluating, if it's in the fewshot fewshotex = [x for x in fewshotex if x != doc][:num_fewshot] labeled_examples = ( "\n\n".join( [ self.doc_to_text(doc) + self.doc_to_target(doc) for doc in fewshotex ] ) + "\n\n" ) example = self.doc_to_text(doc) return description + labeled_examples + example class MultipleChoiceTask(Task): def doc_to_target(self, doc): return " " + doc["choices"][doc["gold"]] def construct_requests(self, doc, ctx): lls = [ rf.loglikelihood(ctx, " {}".format(choice))[0] for choice in doc["choices"] ] return lls def process_results(self, doc, results): gold = doc["gold"] acc = 1.0 if np.argmax(results) == gold else 0.0 completion_len = np.array([float(len(i)) for i in doc["choices"]]) acc_norm = 1.0 if np.argmax(results / completion_len) == gold else 0.0 return { "acc": acc, "acc_norm": acc_norm, } def higher_is_better(self): return { "acc": True, "acc_norm": True, } def aggregation(self): return { "acc": mean, "acc_norm": mean, } class PerplexityTask(Task, abc.ABC): def should_decontaminate(self): """Whether this task supports decontamination against model training set.""" return True def has_training_docs(self): return False def fewshot_examples(self, k, rnd): assert k == 0 return [] def fewshot_context( self, doc, num_fewshot, provide_description=None, rnd=None, description=None ): assert ( num_fewshot == 0 ), "The number of fewshot examples must be 0 for perplexity tasks." assert ( rnd is not None ), "A `random.Random` generator argument must be provided to `rnd`." assert not provide_description, ( "The `provide_description` arg will be removed in future versions. To prepend " "a custom description to the context, supply the corresponding string via the " "`description` arg." ) if provide_description is not None: # nudge people to not specify it at all print( "WARNING: provide_description is deprecated and will be removed in a future version in favor of description_dict" ) return "" def higher_is_better(self): return { "word_perplexity": False, "byte_perplexity": False, "bits_per_byte": False, } def doc_to_decontamination_query(self, doc): return doc def doc_to_text(self, doc): return "" def doc_to_target(self, doc): return doc def construct_requests(self, doc, ctx): assert not ctx req = rf.loglikelihood_rolling(self.doc_to_target(doc)) return req def process_results(self, doc, results): (loglikelihood,) = results words = self.count_words(doc) bytes_ = self.count_bytes(doc) return { "word_perplexity": (loglikelihood, words), "byte_perplexity": (loglikelihood, bytes_), "bits_per_byte": (loglikelihood, bytes_), } def aggregation(self): return { "word_perplexity": weighted_perplexity, "byte_perplexity": weighted_perplexity, "bits_per_byte": bits_per_byte, } @classmethod def count_bytes(cls, doc): return len(doc.encode("utf-8")) @classmethod def count_words(cls, doc): """Downstream tasks with custom word boundaries should override this!""" return len(re.split(r"\s+", doc)) class RequestFactory: def __getattr__(self, attr): def fn(*args): return Request(attr, args) return fn REQUEST_RETURN_LENGTHS = { "loglikelihood": 2, "greedy_until": None, "loglikelihood_rolling": None, } class Request: def __init__(self, request_type, args, index=None): if request_type not in REQUEST_RETURN_LENGTHS.keys(): raise NotImplementedError( "The request type {} is not implemented!".format(request_type) ) self.request_type = request_type self.args = args self.index = index def __iter__(self): if REQUEST_RETURN_LENGTHS[self.request_type] is None: raise IndexError("This request type does not return multiple arguments!") for i in range(REQUEST_RETURN_LENGTHS[self.request_type]): yield Request(self.request_type, self.args, i) def __getitem__(self, i): if REQUEST_RETURN_LENGTHS[self.request_type] is None: raise IndexError("This request type does not return multiple arguments!") return Request(self.request_type, self.args, i) def __eq__(self, other): return ( self.request_type == other.request_type and self.args == other.args and self.index == other.index ) def __repr__(self): return f"Req_{self.request_type}{self.args}[{self.index}]\n" rf = RequestFactory()