evaluation-pipeline/lm_eval/api/task.py

import abc
import logging
import re
import datasets
import os
import numpy as np
import promptsource.templates
from abc import abstractmethod
from typing import Callable, List, Mapping, Optional, Tuple, Union

from lm_eval.api import utils
from lm_eval.api.metric import (
    bits_per_byte,
    bleu,
    mean,
    rouge,
    sari,
    weighted_perplexity,
)
from lm_eval.api.request import Request, rf


logger = logging.getLogger(__name__)


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)
    """

    VERSION = 0

    # 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: Optional[str] = None,
        cache_dir: Optional[str] = None,
        download_mode: Optional[str] = None,
        file_path: Optional[str] = None,
    ):
        """
        Args:
            data_dir (str, optional, defaults to None):
                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).
            cache_dir (str, optional, defaults to None):
                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"`
            download_mode (datasets.DownloadMode, optional, defaults to None):
                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.
        """
        if file_path:
            self.load_from_file(file_path, cache_dir,
                                download_mode=datasets.DownloadMode.FORCE_REDOWNLOAD)
        else:
            self.download(data_dir, cache_dir, download_mode)
        self._training_docs = None
        self._fewshot_docs = None

    def download(
        self,
        data_dir: Optional[str] = None,
        cache_dir: Optional[str] = None,
        download_mode: Optional[str] = None,
    ):
        """Downloads and returns the task dataset.

        NOTE: Override this method to download the dataset from a custom API.
        """
        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 load_from_file(
        self,
        file_path,
        cache_dir: Optional[str] = None,
        download_mode: Optional[str] = None,
    ):
        # get split names
        splits = {}
        dirname = os.path.dirname(file_path)
        for filename in os.listdir(dirname):
            if not filename.startswith(os.path.basename(file_path)):
                continue
            if filename.count(".") == 2:
                splitname = filename.split(".")[1]
                splits[splitname] = os.path.join(dirname, filename)
            else:
                splits["train"] = os.path.join(dirname, filename)

        self.dataset = datasets.load_dataset(
            "json",
            data_files=splits,
            cache_dir=cache_dir,
            download_mode=download_mode,
        )

    @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) -> datasets.Dataset:
        """
        Returns:
            A dataset of training documents.
        """
        return datasets.Dataset.from_dict({})

    def validation_docs(self) -> datasets.Dataset:
        """
        Returns:
            A dataset of validation documents.
        """
        return datasets.Dataset.from_dict({})

    def test_docs(self) -> datasets.Dataset:
        """
        Returns:
            A dataset of test documents.
        """
        return datasets.Dataset.from_dict({})

    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"])`

        Returns:
            The processed version of the specified `doc`.
        """
        return doc

    @abstractmethod
    def doc_to_text(self, doc: dict) -> str:
        pass

    @abstractmethod
    def doc_to_target(self, doc: dict) -> str:
        pass

    @abstractmethod
    def construct_requests(self, doc: dict, ctx: str, args: dict) -> List[Request]:
        """Uses RequestFactory to construct Requests and returns an iterable of
        Requests which will be sent to the LM.

        Args:
            doc (dict):
                The document as returned from training_docs, validation_docs, or
                test_docs.
            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`.
            args (dict):
                The specifics of the context, including number of few shots.

        Returns:
            An iterable of `Request` objects.
        """
        pass

    @abstractmethod
    def process_results(
        self, doc: dict, results: list
    ) -> Union[dict, Tuple[dict, dict]]:
        """Take a single document and the LM results and evaluates, returning a
        dict where keys are the names of sub-metrics and values are the values of
        the metric for that one document.

        Args:
            doc (dict):
                The document as returned from training_docs, validation_docs, or
                test_docs.
            results (list):
                The results of the requests created in construct_requests.

        Returns:
            A dict of metric results.
        """
        pass

    @abstractmethod
    def aggregation(self) -> Mapping[str, Callable]:
        """
        Returns:
            A dictionary where keys are the names of sub-metrics and values are
            functions that aggregate a list of metric scores.
            {str: [metric_score] -> float}
        """
        pass

    @abstractmethod
    def higher_is_better(self) -> Mapping[str, bool]:
        """
        Returns:
            A dictionary where keys are the names of sub-metrics and values are
            whether a higher value of the sub-metric is better.
            {str: bool}
        """
        pass


class PromptSourceTask(Task):
    """These are the metrics from promptsource that we have
    added default behavior for. If you want to add default behavior for a new metric,
    update the functions below. If you want to use one of the following metrics,
    *and* add additional custom processing, override `process_results`, `higher_is_better`, and `aggregation`.
    """

    CONFIGURED_RANKED_CHOICE_PS_METRICS = {"Accuracy"}
    CONFIGURED_GENERATION_PS_METRICS = {"BLEU", "ROUGE", "SARI"}
    SPLIT = None

    def __init__(
        self,
        data_dir: Optional[str] = None,
        cache_dir: Optional[str] = None,
        download_mode: Optional[str] = None,
        prompt_template: Optional[promptsource.templates.Template] = None,
        example_separator: Optional[str] = "\n###\n",
        text_target_separator: Optional[str] = " ",
        save_examples: Optional[bool] = True,
        file_path: Optional[str] = None,
    ):
        """
        Args:
            save_examples (bool, optional, defaults to True):
                Whether to save each example and corresponding model predictions
                to an output `dict`.

            > Few-shot prompting args

            example_separator (str, optional, defaults to '\n###\n'):
                The string that will be used to separate the few-shot examples
                from the prompt example.
                Default: '\n###\n'
                    See Webson & Pavlick (2022) https://arxiv.org/pdf/2109.01247.pdf
                    for justification of this separator.
            text_target_separator (str, optional, defaults to ' '):
                The string that will be used to separate the prompt example
                from the target text.
                NOTE: This is assumed to be some form of whitespace-only separation,
                    e.g. "\n\n", "\t", "  ", etc. Otherwise, you should update
                    the Task's `promptsource` template with the appropriate
                    separator(s).
                Example:
                    Q: Where is the Eiffel Tower located? A:{text_target_separator}Paris
        """
        assert isinstance(save_examples, bool), "`save_examples` must be a bool."
        assert isinstance(example_separator, str) and isinstance(
            text_target_separator, str
        ), "Separator args must be strings."
        assert (
            text_target_separator.isspace()
        ), f"`text_target_separator` must be whitespace only. Got: `{text_target_separator}`"

        if file_path:
            super().__init__(cache_dir=cache_dir, file_path=file_path,
                             download_mode=download_mode)
        else:
            super().__init__(data_dir, cache_dir, download_mode)
        self.prompt_template = prompt_template
        self.save_examples = save_examples
        self.example_separator = example_separator
        self.text_target_separator = text_target_separator

    def stop_sequences(self) -> List[str]:
        """Denote where the generation should end based on the few-shot example
        separator.

        NOTE: Override this if you want to use a sequence other than just the
        task's few-shot example separator.
        """
        return [self.example_separator]

    def max_generation_length(self) -> Optional[int]:
        """Denote where the max length of the generation if it is obvious from the task."""
        return None

    def evaluation_docs(self) -> datasets.Dataset:
        """Returns the `dataset` split to be used for evaluation."""
        if self.has_test_docs():
            return self.test_docs()
        elif self.has_validation_docs():
            return self.validation_docs()
        else:
            raise RuntimeError("Task has neither test_docs nor validation_docs")

    def fewshot_docs(self) -> datasets.Dataset:
        """Returns the `dataset` split that the few-shot examples should be sample
        from. This prioritizes the `train_docs` split as the few-shot example
        source, then `validation_docs`, and lastly `test_docs`.
        """
        if self.has_training_docs():
            return self.training_docs()
        elif self.has_validation_docs():
            return self.validation_docs()
        else:
            return self.test_docs()

    def doc_to_text(self, doc: dict) -> str:
        """Returns the input string for a particular example, given the hf dict."""
        if self.prompt_template is None:
            return self.null_prompt_doc_to_text(doc)
        # is just a string
        text, _ = self.prompt_template.apply(doc)
        return text

    def null_prompt_doc_to_text(self, doc: dict) -> str:
        return NotImplementedError("Override this method in your task!")

    def doc_to_target(self, doc: dict) -> List[str]:
        """Returns the target string for a particular example, given the hf dict."""
        if self.prompt_template is None:
            return self.null_prompt_doc_to_target(doc)
        # is a list of strings where it usually only has one element: the correct answer
        _, target = self.prompt_template.apply(doc)
        return target

    def null_prompt_doc_to_target(self, doc: dict) -> List[str]:
        return NotImplementedError("Override this method in your task!")

    def doc_to_rawtext(self, doc: dict) -> str:
        """This should be used for selecting the raw text of the document.

        The current use case is for computing SARI which requires the text
        without the prompt. The `text` field is not standardized across tasks
        so this is task specific.
        """
        raise NotImplementedError("This is task specific.")

    def invalid_doc_for_prompt(self, doc) -> bool:
        """Some prompts may not work for some documents.
        Default: False
        """
        return False

    def format_example(self, text: str, target: str, separator: str) -> str:
        """Returns the text and target combined by the specified `separator`"""
        return text + separator + target

    def null_prompt_answer_choices(self, doc: dict) -> List[str]:
        return NotImplementedError("Override this method in your task!")

    def fewshot_examples(
        self,
        docs: datasets.Dataset,
        k: int,
        rng: np.random.Generator,
        prompt: dict = None,
    ) -> Tuple[List[dict], List[int]]:
        """Returns `k` random examples from the set of documents in `docs`.

        Args:
            docs (datasets.Dataset):
                The dataset of documents to sample few-shot examples from.
            k (int):
                The number of few-shot examples.
            rng (np.random.Generator):
                The pseudo-random number generator used to randomly sample examples.
            prompt (Optional[dict]):
                The prompt document. Specify this to ensure the prompt is not in
                the set of few-shot examples.

        Returns:
            A tuple of two lists. The first list contains the few-shot examples
        """
        random_indices = np.arange(len(docs)).tolist()
        rng.shuffle(random_indices)

        i = 0
        fewshot_examples, fewshot_idx = [], []
        for idx in random_indices:
            if i >= k:  # Break when we have enough examples.
                break
            is_same_prompt = prompt is not None and all(
                # Skips the `doc_id` key assigned to `prompt`s during eval pre-processing.
                docs[idx][k] == prompt[k]
                for k in docs[idx].keys()
            )
            if self.invalid_doc_for_prompt(docs[idx]) or is_same_prompt:
                continue
            fewshot_examples.append(docs[idx])
            fewshot_idx.append(int(idx))
            i += 1
        return fewshot_examples, fewshot_idx

    def fewshot_context(
        self, doc: dict, num_fewshot: int, rng: Optional[np.random.Generator]
    ) -> Tuple[str, dict]:
        """Returns a few-shot context string made up of `num_fewshot` number of
        labeled examples, and an appended prompt example without labeling.

        Args:
            doc (dict):
                The document as returned from training_docs, validation_docs, or test_docs.
            num_fewshot (int):
                The number of fewshot examples to provide in the returned context string.
            rng (numpy.random.Generator):
                The pseudo-random number generator used to randomly sample few-shot examples.

        Returns:
            A few-shot context string and a dictionary containing few-shot context
            logging information.
                ctx (str):
                    The fewshot context.
                logging_info (dict):
                    A `dict` of logging info that can be used to identify few-shot
                    sources.
        """
        assert (
            rng is not None
        ), "A `numpy.random.Generator` argument must be provided to `rng`"

        if num_fewshot == 0:
            labeled_examples = ""
            fewshot_idx, fewshot_target_idx, fewshot_src = ([], [], None)
        else:
            # Construct few-shot labeled examples.
            fewshot_docs = self.fewshot_docs()
            fewshot_src = str(fewshot_docs.split)
            fewshot_examples, fewshot_idx = self.fewshot_examples(
                fewshot_docs, k=num_fewshot, rng=rng, prompt=doc
            )
            labeled_examples_list = []
            fewshot_target_idx = []
            for fewshot_example in fewshot_examples:
                text = self.doc_to_text(fewshot_example)
                targets = self.doc_to_target(fewshot_example)
                # Choose 1 random target from multi-reference targets.
                target_idx = int(rng.integers(0, len(targets)))
                target = targets[target_idx].strip()
                labeled_examples_list.append(
                    self.format_example(text, target, self.text_target_separator)
                )
                fewshot_target_idx.append(target_idx)
            labeled_examples = self.example_separator.join(labeled_examples_list)
            # Leave an extra `example_separator` right before the prompt.
            labeled_examples += self.example_separator

        prompt = self.doc_to_text(doc)
        ctx = labeled_examples + prompt
        logging_info = {
            "fewshot_idx": fewshot_idx,
            "fewshot_target_idx": fewshot_target_idx,
            "fewshot_source": fewshot_src,
            "fewshot_num": num_fewshot,
            "ctx": ctx,
        }
        return ctx, logging_info

    def construct_requests(self, doc: dict, ctx: str, args: dict) -> List[Request]:
        """Uses RequestFactory to construct Requests and returns an iterable of
        Requests which will be sent to the LM.

        Args:
            doc (dict):
                The document as returned from training_docs, validation_docs, or
                test_docs.
            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`.
            args (dict):
                The specifics of the context, including number of few shots.

        Returns:
            An iterable of `Request` objects.
        """
        requests = []
        if self.prompt_template is None:
            answer_choices_list = self.null_prompt_answer_choices(doc)
        else:
            answer_choices_list = self.prompt_template.get_answer_choices_list(doc)
        if answer_choices_list:
            # If answer_choices_list, then this is a ranked choice prompt.
            for answer_choice in answer_choices_list:
                ll_answer_choice, _ = rf.loglikelihood(
                    ctx, self.text_target_separator + answer_choice
                )
                requests.append(ll_answer_choice)
        else:
            # If not, then this is a generation prompt.
            request_args = {
                "stop_sequences": self.stop_sequences(),
                "max_generation_length": self.max_generation_length(),
                "num_fewshot": args["num_fewshot"],
            }
            cont_request = rf.greedy_until(ctx, request_args)
            requests.append(cont_request)
        return requests

    def process_results(
        self, doc: dict, results: list
    ) -> Union[dict, Tuple[dict, dict]]:
        """Take a single document and the LM results and evaluates, returning a
        dict where keys are the names of sub-metrics and values are the values of
        the metric for that one document.

        NOTE: This function automates processing by using the `promptsource`
        metadata to determine the metric.

        Args:
            doc (dict):
                The document as returned from training_docs, validation_docs, or
                test_docs.
            results (list):
                The results of the requests created in construct_requests.

        Returns:
            A dict of metric results.
        """
        if self.prompt_template is None:
            answer_choices_list = self.null_prompt_answer_choices(doc)
        else:
            answer_choices_list = self.prompt_template.get_answer_choices_list(doc)
        target = self.doc_to_target(doc)
        if answer_choices_list:
            # If answer_choices_list, then this is a ranked choice prompt.
            # NOTE: In the future, target could be a list of strings.
            assert isinstance(target, list) and len(target) == 1
            target = target[0].strip()
            try:
                target_idx = answer_choices_list.index(target)
            except ValueError as e:
                print("answer_choices_list:", answer_choices_list)
                print("target:", target)
                raise ValueError(e)

            pred = answer_choices_list[np.argmax(results)]
            out = {}
            metric_list = ["Accuracy"]  # TODO: CLI framework for specifying metrics

            if self.prompt_template:
                metric_list = self.prompt_template.metadata.metrics
            for metric in metric_list:
                if metric not in self.CONFIGURED_RANKED_CHOICE_PS_METRICS:
                    logger.warning(
                        f"Unexpected metric: `{metric}`. Add it, or use a task-specific solution."
                    )
                if metric == "Accuracy":
                    out["acc"] = pred == target
                    # Byte-length normalization.
                    completion_len = np.array(
                        [float(len(i)) for i in answer_choices_list]
                    )
                    out["acc_norm"] = (
                        1.0
                        if np.argmax(results / completion_len) == target_idx
                        else 0.0
                    )
            # TODO: Add metrics here.
        else:
            # If not, then this is a generation prompt.
            # NOTE: In the future, target will be a list of strings.
            assert isinstance(target, list)
            pred = results[0].strip()
            out = {}
            for metric in self.prompt_template.metadata.metrics:
                if metric not in self.CONFIGURED_GENERATION_PS_METRICS:
                    logger.warning(
                        f"Unexpected metric: `{metric}`. Add it, or use a task-specific solution."
                    )
                if metric == "BLEU":
                    out["bleu"] = (target, pred)
                elif metric == "ROUGE":
                    # TODO: This computes all rouge sub-metrics. Find a generic
                    # way to handle user specified rouge sub-metrics to avoid extra
                    # compute.
                    rouge_scores = rouge(target, pred)
                    # Flatten rouge score dict.
                    rouge_scores = utils.flatten(rouge_scores)
                    # Merge all the rouge-type scores into the `out` dict.
                    out = {**out, **rouge_scores}
                elif metric == "SARI":
                    out["sari"] = sari(self.doc_to_rawtext(doc), pred, target)

        # TODO: Wrap process results s.t. override impl do not
        # override the save examples.
        if self.save_examples:
            example = {
                "pred": pred,
                "target": target,
                "answer_choices_list": answer_choices_list,
            }
            return out, example
        return out

    def aggregation(self) -> Mapping[str, Callable]:
        out = {}
        metric_list = ["Accuracy"]
        if self.prompt_template:
            metric_list = self.prompt_template.metadata.metrics
        for metric in metric_list:
            if metric == "Accuracy":
                out["acc"] = mean
                out["acc_norm"] = mean
            elif metric == "BLEU":
                out["bleu"] = bleu
            elif metric == "ROUGE":
                # TODO: Find a generic way to handle user specified rouge metrics.
                out["rouge1_precision"] = mean
                out["rouge1_recall"] = mean
                out["rouge1_fmeasure"] = mean

                out["rouge2_precision"] = mean
                out["rouge2_recall"] = mean
                out["rouge2_fmeasure"] = mean

                out["rougeL_precision"] = mean
                out["rougeL_recall"] = mean
                out["rougeL_fmeasure"] = mean

                out["rougeLsum_precision"] = mean
                out["rougeLsum_recall"] = mean
                out["rougeLsum_fmeasure"] = mean
            elif metric == "SARI":
                out["sari"] = mean
        return out

    def higher_is_better(self) -> Mapping[str, bool]:
        out = {}
        for metric in self.prompt_template.metadata.metrics:
            if metric == "Accuracy":
                out["acc"] = True
                out["acc_norm"] = True
            elif metric == "BLEU":
                out["bleu"] = True
            elif metric == "ROUGE":
                # TODO: Find a generic way to handle user specified rouge metrics.
                out["rouge1_precision"] = True
                out["rouge1_recall"] = True
                out["rouge1_fmeasure"] = True

                out["rouge2_precision"] = True
                out["rouge2_recall"] = True
                out["rouge2_fmeasure"] = True

                out["rougeL_precision"] = True
                out["rougeL_recall"] = True
                out["rougeL_fmeasure"] = True

                out["rougeLsum_precision"] = True
                out["rougeLsum_recall"] = True
                out["rougeLsum_fmeasure"] = True
            elif metric == "SARI":
                out["sari"] = True
        return out

    def get_logging_info(self):
        if self.prompt_template is None:
            return self.null_prompt_get_logging_info()
        return {
            "fixed_answer_choice_list": self.prompt_template.get_fixed_answer_choices_list(),
            "dataset_path": self.DATASET_PATH,
            "dataset_name": self.DATASET_NAME,
            "subset": self.SPLIT,
            "prompt_name": self.prompt_template.get_name(),
            "prompt_id": self.prompt_template.get_id(),
            "prompt_jinja": self.prompt_template.jinja,
            "prompt_original_task": self.prompt_template.metadata.original_task,
            # Placeholder for comment in post-processing.
            "comment": "",
        }


class TranslationTask(PromptSourceTask):

    # Language specific functions.
    @classmethod
    def zh_split(cls, zh_text: str) -> List[str]:
        """Chinese splitting"""
        import jieba

        return [" ".join(jieba.cut(txt.strip())) for txt in zh_text]

    @classmethod
    def ja_split(cls, ja_text: str) -> List[str]:
        """Japanese splitting"""
        import nagisa

        return [" ".join(nagisa.tagging(txt.strip()).words) for txt in ja_text]

    NO_SPACE_LANG = {"zh": zh_split, "ja": ja_split}

    def invalid_doc_for_prompt(self, doc) -> bool:
        # Skip docs with empty references.
        if self.doc_to_target(doc) == [""]:
            return True
        return False

    def _get_src_ref_codes(self, template_name: str) -> Tuple[str, str]:
        """Returns a 2-tuple of (src_lang, ref_lang) codes from the prompt template name."""
        # Get the lang codes from the dataset name.
        lang_pairs = self.DATASET_NAME.split("-")
        # Template name ordering defines the src and ref lang codes.
        if self.DATASET_NAME in template_name:
            return lang_pairs[0], lang_pairs[1]
        # Flip the lang pairs following the prompt source.
        return lang_pairs[1], lang_pairs[0]

    def process_results(
        self, doc: dict, results: list
    ) -> Union[dict, Tuple[dict, dict]]:
        answer_choices_list = self.prompt_template.get_answer_choices_list(doc)
        target = self.doc_to_target(doc)

        # Add spaces between words for BLEU score calculation of target languages like Chinese
        _, tar_lang_code = self._get_src_ref_codes(self.prompt_template.name)
        if tar_lang_code in self.NO_SPACE_LANG:
            target = [self.NO_SPACE_LANG[tar_lang_code]([t])[0] for t in target]
            results = self.NO_SPACE_LANG[tar_lang_code](results)
        pred = results[0].strip()

        out = {}
        for metric in self.prompt_template.metadata.metrics:
            assert (
                metric in self.CONFIGURED_GENERATION_PS_METRICS
            ), "Unexpected metric. Add it, or use a task-specific solution."
            if metric == "BLEU":
                out["bleu"] = (target, pred)
            elif metric == "ROUGE":
                # TODO: This computes all rouge sub-metrics. Find a generic
                # way to handle user specified rouge sub-metrics to avoid extra
                # compute.
                rouge_scores = rouge(target, pred)
                # Flatten rouge score dict.
                rouge_scores = utils.flatten(rouge_scores)
                # Merge all the rouge-type scores into the `out` dict.
                out = {**out, **rouge_scores}

        # TODO: Wrap process results s.t. override impl do not
        # override the save examples.
        if self.save_examples:
            example = {
                "pred": pred,
                "target": target,
                "answer_choices_list": answer_choices_list,
            }
            return out, example
        return out


class PerplexityTask(PromptSourceTask):
    """NOTE: Prompts are ignored for perplexity tasks."""

    def doc_to_text(self, doc: dict) -> str:
        return ""

    def doc_to_target(self, doc: dict) -> List[str]:
        """Because prompts are ignored, return the relevant text from doc."""
        raise NotImplementedError()

    def fewshot_context(
        self,
        doc: dict,
        num_fewshot: int,
        rng: Optional[np.random.Generator],
    ) -> Tuple[str, dict]:
        assert (
            num_fewshot == 0
        ), "The number of fewshot examples must be 0 for perplexity tasks."
        assert (
            rng is not None
        ), "A `numpy.random.Generator` argument must be provided to `rng`"
        return (
            "",
            {
                "fewshot_idx": [],
                "fewshot_target_idx": [],
                "fewshot_source": None,
                "fewshot_num": 0,
                "ctx": "",
            },
        )

    def construct_requests(self, doc: dict, ctx: str, args: dict) -> List[Request]:
        assert not ctx
        string = self.doc_to_target(doc)[0]
        req = rf.loglikelihood_rolling(string)
        return req

    def process_results(
        self, doc: dict, results: list
    ) -> Union[dict, Tuple[dict, dict]]:
        (loglikelihood,) = results
        target = self.doc_to_target(doc)[0]
        words = self.count_words(target)
        bytes_ = self.count_bytes(target)

        out = {
            "word_perplexity": (loglikelihood, words),
            "byte_perplexity": (loglikelihood, bytes_),
            "bits_per_byte": (loglikelihood, bytes_),
        }
        if self.save_examples:
            return out, {
                "word_perplexity_instance": weighted_perplexity(
                    [(loglikelihood, words)]
                ),
                "byte_perplexity_instance": weighted_perplexity(
                    [(loglikelihood, bytes_)]
                ),
                "bits_per_byte_instance": bits_per_byte([(loglikelihood, bytes_)]),
            }
        return out

    def aggregation(self) -> Mapping[str, Callable]:
        return {
            "word_perplexity": weighted_perplexity,
            "byte_perplexity": weighted_perplexity,
            "bits_per_byte": bits_per_byte,
        }

    def higher_is_better(self) -> Mapping[str, bool]:
        return {
            "word_perplexity": False,
            "byte_perplexity": False,
            "bits_per_byte": False,
        }

    @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))

    def get_logging_info(self):
        return {
            "prompt_name": None,
        }