Add app using dialoGPT-finetuned beam10 no-minlen

.gitignore

@@ -0,0 +1,2 @@
+.venv
+__pycache__

app.py

@@ -0,0 +1,17 @@
+from response_generation import ResponseGenerator
+import gradio as gr
+
+
+DEFAULT_MODEL = "shaneweisz/DialoGPT-finetuned-multiCONAN"
+DECODING_CONFIG = {"max_new_tokens": 100, "no_repeat_ngram_size": 3, "num_beams": 10}
+
+model = ResponseGenerator(DEFAULT_MODEL, DECODING_CONFIG)
+
+
+def respond(hate_speech_input_text):
+    return model.respond(hate_speech_input_text)
+
+
+demo = gr.Interface(fn=respond, inputs="text", outputs="text")
+
+demo.launch()

requirements.txt

@@ -0,0 +1,5 @@
+torch==1.11.0
+transformers==4.19.2
+tqdm==4.64.0
+colorama==0.4.4
+gradio==3.0.20

response_generation/__init__.py

@@ -0,0 +1 @@
+from .response_generator import ResponseGenerator

response_generation/min_new_tokens.py

@@ -0,0 +1,37 @@
+import torch
+from transformers import LogitsProcessor
+
+
+# HuggingFace's generate function does not yet support a `min_new_tokens`, so we need to add the functionality
+# ourselves by adding a custom logits processor. Adapted from:
+# https://huggingface.co/transformers/v4.1.1/_modules/transformers/generation_logits_process.html#MinLengthLogitsProcessor
+class MinNewTokensLogitsProcessor(LogitsProcessor):
+    r"""
+    A [`LogitsProcessor`] enforcing a minimum response length by setting the `EOS` probability to 0 until
+    `min_new_tokens` new tokens have been generated since `input_length`.
+    """
+    def __init__(self, min_new_tokens: int, eos_token_id: int, input_length: int):
+        if not isinstance(min_new_tokens, int) or min_new_tokens < 0:
+            raise ValueError(f"`min_new_tokens` has to be a positive integer, but is {min_new_tokens}")
+
+        if not isinstance(eos_token_id, int) or eos_token_id < 0:
+            raise ValueError(f"`eos_token_id` has to be a positive integer, but is {eos_token_id}")
+
+        if not isinstance(input_length, int) or input_length < 0:
+            raise ValueError(f"`input_length` has to be a positive integer, but is {input_length}")
+
+        self.min_new_tokens = min_new_tokens
+        self.eos_token_id = eos_token_id
+        self.input_length = input_length
+
+    def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor) -> torch.FloatTensor:
+        if not hasattr(self, "input_length"):
+            raise ValueError("`save_input_length` has to be called before `__call__`")
+
+        total_length = input_ids.shape[-1]
+        response_len = total_length - self.input_length
+
+        if response_len < self.min_new_tokens:
+            scores[:, self.eos_token_id] = -float("inf")
+
+        return scores

response_generation/response_generator.py

@@ -0,0 +1,73 @@
+from typing import Any, Dict, List
+from tqdm import tqdm
+from colorama import Fore, Style
+from transformers import AutoModelForCausalLM, AutoTokenizer, LogitsProcessorList
+import torch
+from .min_new_tokens import MinNewTokensLogitsProcessor
+
+
+class ResponseGenerator:
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+    def __init__(self, pretrained_model_name_or_path: str, decoding_config: Dict[str, Any], seed=42, verbose=True):
+        self.model = AutoModelForCausalLM.from_pretrained(pretrained_model_name_or_path).to(self.device)
+        self.tokenizer = AutoTokenizer.from_pretrained(pretrained_model_name_or_path)
+        if "pad_token" not in self.tokenizer.special_tokens_map:
+            self.tokenizer.pad_token = self.tokenizer.eos_token  # A pad token needs to be set for batch decoding
+        self.decoding_config = decoding_config
+        self.verbose = verbose
+        torch.manual_seed(seed)
+
+    def generate_responses(self, inputs: List[str], batch_size=1) -> List[str]:
+        responses = []
+        for i in tqdm(range(0, len(inputs), batch_size), disable=not self.verbose):
+            batch_inputs = inputs[i : i + batch_size]
+            batch_responses = self.generate_responses_for_batch(batch_inputs)
+            responses.extend(batch_responses)
+        return responses
+
+    def generate_responses_for_batch(self, inputs: List[str]) -> str:
+        inputs = [input_text + self.tokenizer.eos_token for input_text in inputs]
+
+        self.tokenizer.padding_side = "left"
+        tokenized_inputs = self.tokenizer(inputs, return_tensors="pt", padding=True).to(self.device)
+        input_len = tokenized_inputs["input_ids"].shape[-1]
+
+        params_for_generate = self._params_for_generate(input_len)
+        output_ids = self.model.generate(
+            **tokenized_inputs, **params_for_generate, pad_token_id=self.tokenizer.pad_token_id
+        )
+
+        response_ids = output_ids[:, input_len:]
+        responses = self.tokenizer.batch_decode(response_ids, skip_special_tokens=True)
+
+        return responses
+
+    def _params_for_generate(self, input_length: int) -> Dict[str, Any]:
+        params_for_generate = self.decoding_config.copy()
+
+        if "min_new_tokens" in params_for_generate and params_for_generate["min_new_tokens"] is not None:
+            # the HuggingFace `generate` function accepts a `logits_processor` argument, not a `min_new_tokens`,
+            # so we replace `min_new_tokens` from the `decoding_config` with our custom logits processor
+            # that enforces a minimum response length
+            min_new_tokens = params_for_generate["min_new_tokens"]
+            min_new_tokens_logits_processor = MinNewTokensLogitsProcessor(
+                min_new_tokens, self.tokenizer.eos_token_id, input_length
+            )
+            params_for_generate["logits_processor"] = LogitsProcessorList([min_new_tokens_logits_processor])
+            params_for_generate.pop("min_new_tokens")
+
+        return params_for_generate
+
+    def respond(self, input_text: str) -> str:
+        """Respond to a single hate speech input."""
+        return self.generate_responses([input_text])[0]
+
+    def interact(self):
+        prompt = Fore.RED + "Hate speech: " + Style.RESET_ALL
+        input_text = input(prompt)
+        while input_text != "":
+            print(Fore.GREEN + "Response: " + Style.RESET_ALL, end="")
+            response = self.respond(input_text)
+            print(response)
+            input_text = input(prompt)