diff --git a/backend/Generator/__init__.py b/backend/Generator/__init__.py index 1703366b..b9e66042 100644 --- a/backend/Generator/__init__.py +++ b/backend/Generator/__init__.py @@ -1,3 +1,6 @@ # Constructor for questgen from __future__ import absolute_import -from Generator.main import MCQGenerator, BoolQGenerator, ShortQGenerator, AnswerPredictor, GoogleDocsService, FileProcessor, QuestionGenerator +from Generator.question_generators import MCQGenerator, BoolQGenerator, ShortQGenerator +from Generator.answer_predictor import AnswerPredictor +from Generator.utilities import GoogleDocsService, FileProcessor +from Generator.advanced_qa import QuestionGenerator diff --git a/backend/Generator/advanced_qa.py b/backend/Generator/advanced_qa.py new file mode 100644 index 00000000..cfa429bc --- /dev/null +++ b/backend/Generator/advanced_qa.py @@ -0,0 +1,399 @@ +import json +import random +import re + +import torch +import en_core_web_sm +from transformers import ( + AutoModelForSequenceClassification, + AutoModelForSeq2SeqLM, + AutoTokenizer, +) +from typing import Any, List, Mapping, Tuple + + +class QuestionGenerator: + """A transformer-based NLP system for generating reading comprehension-style questions from + texts. It can generate full sentence questions, multiple choice questions, or a mix of the + two styles. + + To filter out low quality questions, questions are assigned a score and ranked once they have + been generated. Only the top k questions will be returned. This behaviour can be turned off + by setting use_evaluator=False. + """ + + def __init__(self) -> None: + """Loads the question-generation model and tokenizer, and the QA evaluator.""" + + QG_PRETRAINED = "iarfmoose/t5-base-question-generator" + self.ANSWER_TOKEN = "" + self.CONTEXT_TOKEN = "" + self.SEQ_LENGTH = 512 + + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + + self.qg_tokenizer = AutoTokenizer.from_pretrained(QG_PRETRAINED, use_fast=False) + self.qg_model = AutoModelForSeq2SeqLM.from_pretrained(QG_PRETRAINED) + self.qg_model.to(self.device) + self.qg_model.eval() + + self.qa_evaluator = QAEvaluator() + + def generate( + self, + article: str, + use_evaluator: bool = False, + num_questions: bool = None, + answer_style: str = "all", + ) -> List: + """Takes an article and generates a set of question and answer pairs. If use_evaluator + is True then QA pairs will be ranked and filtered based on their quality. answer_style + should selected from ["all", "sentences", "multiple_choice"]. + """ + + print("Generating questions...\n") + + qg_inputs, qg_answers = self.generate_qg_inputs(article, answer_style) + generated_questions = self.generate_questions_from_inputs(qg_inputs) + + message = "{} questions doesn't match {} answers".format( + len(generated_questions), len(qg_answers) + ) + assert len(generated_questions) == len(qg_answers), message + + if use_evaluator: + print("Evaluating QA pairs...\n") + encoded_qa_pairs = self.qa_evaluator.encode_qa_pairs( + generated_questions, qg_answers + ) + scores = self.qa_evaluator.get_scores(encoded_qa_pairs) + + if num_questions: + qa_list = self._get_ranked_qa_pairs( + generated_questions, qg_answers, scores, num_questions + ) + else: + qa_list = self._get_ranked_qa_pairs( + generated_questions, qg_answers, scores + ) + + else: + print("Skipping evaluation step.\n") + qa_list = self._get_all_qa_pairs(generated_questions, qg_answers) + + if self.device.type == 'cuda': + torch.cuda.empty_cache() + + return qa_list + + def generate_qg_inputs( + self, text: str, answer_style: str + ) -> Tuple[List[str], List[str]]: + """Given a text, returns a list of model inputs and a list of corresponding answers. + Model inputs take the form "answer_token context_token " where + the answer is a string extracted from the text, and the context is the wider text surrounding + the context. + """ + + VALID_ANSWER_STYLES = ["all", "sentences", "multiple_choice"] + + if answer_style not in VALID_ANSWER_STYLES: + raise ValueError( + "Invalid answer style {}. Please choose from {}".format( + answer_style, VALID_ANSWER_STYLES + ) + ) + + inputs = [] + answers = [] + + if answer_style == "sentences" or answer_style == "all": + segments = self._split_into_segments(text) + + for segment in segments: + sentences = self._split_text(segment) + prepped_inputs, prepped_answers = self._prepare_qg_inputs( + sentences, segment + ) + inputs.extend(prepped_inputs) + answers.extend(prepped_answers) + + if answer_style == "multiple_choice" or answer_style == "all": + sentences = self._split_text(text) + prepped_inputs, prepped_answers = self._prepare_qg_inputs_MC(sentences) + inputs.extend(prepped_inputs) + answers.extend(prepped_answers) + + return inputs, answers + + def generate_questions_from_inputs(self, qg_inputs: List) -> List[str]: + """Given a list of concatenated answers and contexts, with the form: + "answer_token context_token ", generates a list of + questions. + """ + generated_questions = [] + + for qg_input in qg_inputs: + question = self._generate_question(qg_input) + generated_questions.append(question) + + return generated_questions + + def _split_text(self, text: str) -> List[str]: + """Splits the text into sentences, and attempts to split or truncate long sentences.""" + MAX_SENTENCE_LEN = 128 + sentences = re.findall(".*?[.!\?]", text) + cut_sentences = [] + + for sentence in sentences: + if len(sentence) > MAX_SENTENCE_LEN: + cut_sentences.extend(re.split("[,;:)]", sentence)) + + # remove useless post-quote sentence fragments + cut_sentences = [s for s in sentences if len(s.split(" ")) > 5] + sentences = sentences + cut_sentences + + return list(set([s.strip(" ") for s in sentences])) + + def _split_into_segments(self, text: str) -> List[str]: + """Splits a long text into segments short enough to be input into the transformer network. + Segments are used as context for question generation. + """ + MAX_TOKENS = 490 + paragraphs = text.split("\n") + tokenized_paragraphs = [ + self.qg_tokenizer(p)["input_ids"] for p in paragraphs if len(p) > 0 + ] + segments = [] + + while len(tokenized_paragraphs) > 0: + segment = [] + + while len(segment) < MAX_TOKENS and len(tokenized_paragraphs) > 0: + paragraph = tokenized_paragraphs.pop(0) + segment.extend(paragraph) + segments.append(segment) + + return [self.qg_tokenizer.decode(s, skip_special_tokens=True) for s in segments] + + def _prepare_qg_inputs( + self, sentences: List[str], text: str + ) -> Tuple[List[str], List[str]]: + """Uses sentences as answers and the text as context. Returns a tuple of (model inputs, answers). + Model inputs are "answer_token context_token " + """ + inputs = [] + answers = [] + + for sentence in sentences: + qg_input = f"{self.ANSWER_TOKEN} {sentence} {self.CONTEXT_TOKEN} {text}" + inputs.append(qg_input) + answers.append(sentence) + + return inputs, answers + + def _prepare_qg_inputs_MC( + self, sentences: List[str] + ) -> Tuple[List[str], List[str]]: + """Performs NER on the text, and uses extracted entities are candidate answers for multiple-choice + questions. Sentences are used as context, and entities as answers. Returns a tuple of (model inputs, answers). + Model inputs are "answer_token context_token " + """ + spacy_nlp = en_core_web_sm.load() + docs = list(spacy_nlp.pipe(sentences, disable=["parser"])) + inputs_from_text = [] + answers_from_text = [] + + for doc, sentence in zip(docs, sentences): + entities = doc.ents + if entities: + + for entity in entities: + qg_input = ( + f"{self.ANSWER_TOKEN} {entity} {self.CONTEXT_TOKEN} {sentence}" + ) + answers = self._get_MC_answers(entity, docs) + inputs_from_text.append(qg_input) + answers_from_text.append(answers) + + return inputs_from_text, answers_from_text + + def _get_MC_answers( + self, correct_answer: Any, docs: Any + ) -> List[Mapping[str, Any]]: + """Finds a set of alternative answers for a multiple-choice question. Will attempt to find + alternatives of the same entity type as correct_answer if possible. + """ + entities = [] + + for doc in docs: + entities.extend([{"text": e.text, "label_": e.label_} for e in doc.ents]) + + # remove duplicate elements + entities_json = [json.dumps(kv) for kv in entities] + pool = set(entities_json) + num_choices = ( + min(4, len(pool)) - 1 + ) # -1 because we already have the correct answer + + # add the correct answer + final_choices = [] + correct_label = correct_answer.label_ + final_choices.append({"answer": correct_answer.text, "correct": True}) + pool.remove( + json.dumps({"text": correct_answer.text, "label_": correct_answer.label_}) + ) + + # find answers with the same NER label + matches = [e for e in pool if correct_label in e] + + # if we don't have enough then add some other random answers + if len(matches) < num_choices: + choices = matches + pool = pool.difference(set(choices)) + choices.extend(random.sample(list(pool), num_choices - len(choices))) + else: + choices = random.sample(matches, num_choices) + + choices = [json.loads(s) for s in choices] + + for choice in choices: + final_choices.append({"answer": choice["text"], "correct": False}) + + random.shuffle(final_choices) + return final_choices + + @torch.no_grad() + def _generate_question(self, qg_input: str) -> str: + """Takes qg_input which is the concatenated answer and context, and uses it to generate + a question sentence. The generated question is decoded and then returned. + """ + encoded_input = self._encode_qg_input(qg_input) + output = self.qg_model.generate(input_ids=encoded_input["input_ids"]) + question = self.qg_tokenizer.decode(output[0], skip_special_tokens=True) + return question + + def _encode_qg_input(self, qg_input: str) -> torch.tensor: + """Tokenizes a string and returns a tensor of input ids corresponding to indices of tokens in + the vocab. + """ + return self.qg_tokenizer( + qg_input, + padding="max_length", + max_length=self.SEQ_LENGTH, + truncation=True, + return_tensors="pt", + ).to(self.device) + + def _get_ranked_qa_pairs( + self, + generated_questions: List[str], + qg_answers: List[str], + scores, + num_questions: int = 10, + ) -> List[Mapping[str, str]]: + """Ranks generated questions according to scores, and returns the top num_questions examples.""" + if num_questions > len(scores): + num_questions = len(scores) + print( + ( + f"\nWas only able to generate {num_questions} questions.", + "For more questions, please input a longer text.", + ) + ) + + qa_list = [] + + for i in range(num_questions): + index = scores[i] + qa = { + "question": generated_questions[index].split("?")[0] + "?", + "answer": qg_answers[index], + } + qa_list.append(qa) + + return qa_list + + def _get_all_qa_pairs(self, generated_questions: List[str], qg_answers: List[str]): + """Formats question and answer pairs without ranking or filtering.""" + qa_list = [] + + for question, answer in zip(generated_questions, qg_answers): + qa = {"question": question.split("?")[0] + "?", "answer": answer} + qa_list.append(qa) + + return qa_list + + +class QAEvaluator: + """Wrapper for a transformer model which evaluates the quality of question-answer pairs. + Given a QA pair, the model will generate a score. Scores can be used to rank and filter + QA pairs. + """ + + def __init__(self) -> None: + """Loads the QA-evaluator model and tokenizer used to score question-answer pairs.""" + + QAE_PRETRAINED = "iarfmoose/bert-base-cased-qa-evaluator" + self.SEQ_LENGTH = 512 + + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + + self.qae_tokenizer = AutoTokenizer.from_pretrained(QAE_PRETRAINED) + self.qae_model = AutoModelForSequenceClassification.from_pretrained( + QAE_PRETRAINED + ) + self.qae_model.to(self.device) + self.qae_model.eval() + + def encode_qa_pairs( + self, questions: List[str], answers: List[str] + ) -> List[torch.tensor]: + """Takes a list of questions and a list of answers and encodes them as a list of tensors.""" + encoded_pairs = [] + + for question, answer in zip(questions, answers): + encoded_qa = self._encode_qa(question, answer) + encoded_pairs.append(encoded_qa.to(self.device)) + + return encoded_pairs + + def get_scores(self, encoded_qa_pairs: List[torch.tensor]) -> List[float]: + """Generates scores for a list of encoded QA pairs.""" + scores = {} + + for i in range(len(encoded_qa_pairs)): + scores[i] = self._evaluate_qa(encoded_qa_pairs[i]) + + if self.device.type == 'cuda': + torch.cuda.empty_cache() + + return [ + k for k, v in sorted(scores.items(), key=lambda item: item[1], reverse=True) + ] + + def _encode_qa(self, question: str, answer: str) -> torch.tensor: + """Concatenates a question and answer, and then tokenizes them. Returns a tensor of + input ids corresponding to indices in the vocab. + """ + if type(answer) is list: + for a in answer: + if a["correct"]: + correct_answer = a["answer"] + else: + correct_answer = answer + + return self.qae_tokenizer( + text=question, + text_pair=correct_answer, + padding="max_length", + max_length=self.SEQ_LENGTH, + truncation=True, + return_tensors="pt", + ) + + @torch.no_grad() + def _evaluate_qa(self, encoded_qa_pair: torch.tensor) -> float: + """Takes an encoded QA pair and returns a score.""" + output = self.qae_model(**encoded_qa_pair) + return output[0][0][1] diff --git a/backend/Generator/answer_predictor.py b/backend/Generator/answer_predictor.py new file mode 100644 index 00000000..57816588 --- /dev/null +++ b/backend/Generator/answer_predictor.py @@ -0,0 +1,108 @@ +import torch +import numpy as np +from transformers import ( + T5ForConditionalGeneration, + T5Tokenizer, + AutoModelForSequenceClassification, + AutoTokenizer, +) + + +class AnswerPredictor: + + def __init__(self): + """Loads the answer-prediction and NLI models used for answer and boolean-answer prediction.""" + self.tokenizer = T5Tokenizer.from_pretrained('t5-large', model_max_length=512) + self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Answer-Predictor') + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + self.model.to(self.device) + + # Load the lightweight NLI model for boolean question answering + self.nli_model_name = "typeform/distilbert-base-uncased-mnli" + self.nli_tokenizer = AutoTokenizer.from_pretrained(self.nli_model_name) + self.nli_model = AutoModelForSequenceClassification.from_pretrained(self.nli_model_name) + + self.set_seed(42) + + def set_seed(self, seed): + """Seeds NumPy, torch and CUDA RNGs so that generation is reproducible.""" + np.random.seed(seed) + torch.manual_seed(seed) + if torch.cuda.is_available(): + torch.cuda.manual_seed_all(seed) + + def greedy_decoding(self, inp_ids, attn_mask): + """Greedy-decodes an answer from tokenized model input and returns the cleaned text.""" + greedy_output = self.model.generate(input_ids=inp_ids, attention_mask=attn_mask, max_length=256) + Question = self.tokenizer.decode(greedy_output[0], skip_special_tokens=True, clean_up_tokenization_spaces=True) + return Question.strip().capitalize() + + def predict_answer(self, payload): + """Predicts an answer for each question in ``payload`` against the given context. + + Args: + payload: Dict with ``input_text`` (the context) and ``input_question`` (a list of + questions). + + Returns: + A list with one predicted answer (stripped and capitalized) per question. + """ + answers = [] + inp = { + "input_text": payload.get("input_text"), + "input_question" : payload.get("input_question") + } + for ques in payload.get("input_question"): + + context = inp["input_text"] + question = ques + input_text = "question: %s context: %s " % (question, context) + + encoding = self.tokenizer.encode_plus(input_text, return_tensors="pt") + input_ids, attention_masks = encoding["input_ids"].to(self.device), encoding["attention_mask"].to(self.device) + greedy_output = self.model.generate(input_ids=input_ids, attention_mask=attention_masks, max_length=256) + Question = self.tokenizer.decode(greedy_output[0], skip_special_tokens=True, clean_up_tokenization_spaces=True) + answers.append(Question.strip().capitalize()) + + if self.device.type == 'cuda': + torch.cuda.empty_cache() + + return answers + + def predict_boolean_answer(self, payload): + """Predicts a True/False answer for each yes/no question in ``payload``. + + Uses the NLI model to compare the entailment and contradiction + probabilities of each question (as the hypothesis) against the + context (as the premise). + + Args: + payload: Dict with ``input_text`` (the context) and ``input_question`` (a list of + yes/no questions). + + Returns: + A list of booleans, one per question. + """ + input_text = payload.get("input_text", "") + input_questions = payload.get("input_question", []) + + answers = [] + + for question in input_questions: + hypothesis = question + inputs = self.nli_tokenizer.encode_plus(input_text, hypothesis, return_tensors="pt") + outputs = self.nli_model(**inputs) + logits = outputs.logits + probabilities = torch.softmax(logits, dim=1) + entailment_prob = probabilities[0][0].item() + contradiction_prob = probabilities[0][2].item() + + if entailment_prob > contradiction_prob: + answers.append(True) + else: + answers.append(False) + + if self.device.type == 'cuda': + torch.cuda.empty_cache() + + return answers diff --git a/backend/Generator/main.py b/backend/Generator/main.py deleted file mode 100644 index 04aed79f..00000000 --- a/backend/Generator/main.py +++ /dev/null @@ -1,806 +0,0 @@ -import time -import torch -import random -from transformers import T5ForConditionalGeneration, T5Tokenizer -from transformers import AutoModelForSequenceClassification, AutoTokenizer,AutoModelForSeq2SeqLM, T5ForConditionalGeneration, T5Tokenizer -import numpy as np -import spacy -from sense2vec import Sense2Vec -from collections import OrderedDict -from nltk import FreqDist -from nltk.corpus import brown -from similarity.normalized_levenshtein import NormalizedLevenshtein -from Generator.mcq import tokenize_into_sentences, identify_keywords, find_sentences_with_keywords, generate_multiple_choice_questions, generate_normal_questions -from Generator.encoding import beam_search_decoding -from google.oauth2 import service_account -from googleapiclient.discovery import build -import en_core_web_sm -import json -import re -from typing import Any, List, Mapping, Tuple -import re -import os -import fitz -import mammoth - -class MCQGenerator: - - def __init__(self): - self.tokenizer = T5Tokenizer.from_pretrained('t5-large') - self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Question-Generator') - self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - self.model.to(self.device) - self.nlp = spacy.load('en_core_web_sm') - self.s2v = Sense2Vec().from_disk('s2v_old') - self.fdist = FreqDist(brown.words()) - self.normalized_levenshtein = NormalizedLevenshtein() - self.set_seed(42) - - def set_seed(self, seed): - np.random.seed(seed) - torch.manual_seed(seed) - if torch.cuda.is_available(): - torch.cuda.manual_seed_all(seed) - - def generate_mcq(self, payload): - start_time = time.time() - inp = { - "input_text": payload.get("input_text"), - "max_questions": payload.get("max_questions", 4) - } - - text = inp['input_text'] - sentences = tokenize_into_sentences(text) - modified_text = " ".join(sentences) - - keywords = identify_keywords(self.nlp, modified_text, inp['max_questions'], self.s2v, self.fdist, self.normalized_levenshtein, len(sentences)) - keyword_sentence_mapping = find_sentences_with_keywords(keywords, sentences) - - for k in keyword_sentence_mapping.keys(): - text_snippet = " ".join(keyword_sentence_mapping[k][:3]) - keyword_sentence_mapping[k] = text_snippet - - final_output = {} - - if len(keyword_sentence_mapping.keys()) == 0: - return final_output - else: - try: - generated_questions = generate_multiple_choice_questions(keyword_sentence_mapping, self.device, self.tokenizer, self.model, self.s2v, self.normalized_levenshtein) - except: - return final_output - - end_time = time.time() - - final_output["statement"] = modified_text - final_output["questions"] = generated_questions["questions"] - final_output["time_taken"] = end_time - start_time - - if self.device.type == 'cuda': - torch.cuda.empty_cache() - - return final_output - -class ShortQGenerator: - - def __init__(self): - self.tokenizer = T5Tokenizer.from_pretrained('t5-large') - self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Question-Generator') - self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - self.model.to(self.device) - self.nlp = spacy.load('en_core_web_sm') - self.s2v = Sense2Vec().from_disk('s2v_old') - self.fdist = FreqDist(brown.words()) - self.normalized_levenshtein = NormalizedLevenshtein() - self.set_seed(42) - - def set_seed(self, seed): - np.random.seed(seed) - torch.manual_seed(seed) - if torch.cuda.is_available(): - torch.cuda.manual_seed_all(seed) - - def generate_shortq(self, payload): - inp = { - "input_text": payload.get("input_text"), - "max_questions": payload.get("max_questions", 4) - } - - text = inp['input_text'] - sentences = tokenize_into_sentences(text) - modified_text = " ".join(sentences) - - keywords = identify_keywords(self.nlp, modified_text, inp['max_questions'], self.s2v, self.fdist, self.normalized_levenshtein, len(sentences)) - keyword_sentence_mapping = find_sentences_with_keywords(keywords, sentences) - - for k in keyword_sentence_mapping.keys(): - text_snippet = " ".join(keyword_sentence_mapping[k][:3]) - keyword_sentence_mapping[k] = text_snippet - - final_output = {} - - if len(keyword_sentence_mapping.keys()) == 0: - return final_output - else: - generated_questions = generate_normal_questions(keyword_sentence_mapping, self.device, self.tokenizer, self.model) - - final_output["statement"] = modified_text - final_output["questions"] = generated_questions["questions"] - - if self.device.type == 'cuda': - torch.cuda.empty_cache() - - return final_output - -class ParaphraseGenerator: - - def __init__(self): - self.tokenizer = T5Tokenizer.from_pretrained('t5-large') - self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Question-Generator') - self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - self.model.to(self.device) - self.set_seed(42) - - def set_seed(self, seed): - np.random.seed(seed) - torch.manual_seed(seed) - if torch.cuda.is_available(): - torch.cuda.manual_seed_all(seed) - - def generate_paraphrase(self, payload): - start_time = time.time() - inp = { - "input_text": payload.get("input_text"), - "max_questions": payload.get("max_questions", 3) - } - - text = inp['input_text'] - num = inp['max_questions'] - - sentence = text - text_to_paraphrase = "paraphrase: " + sentence + " " - - encoding = self.tokenizer.encode_plus(text_to_paraphrase, pad_to_max_length=True, return_tensors="pt") - input_ids, attention_masks = encoding["input_ids"].to(self.device), encoding["attention_mask"].to(self.device) - - beam_outputs = self.model.generate( - input_ids=input_ids, - attention_mask=attention_masks, - max_length=50, - num_beams=50, - num_return_sequences=num, - no_repeat_ngram_size=2, - early_stopping=True - ) - - final_outputs =[] - for beam_output in beam_outputs: - paraphrased_sentence = self.tokenizer.decode(beam_output, skip_special_tokens=True, clean_up_tokenization_spaces=True) - if paraphrased_sentence.lower() != sentence.lower() and paraphrased_sentence not in final_outputs: - final_outputs.append(paraphrased_sentence) - - output = {} - output['Original Sentence'] = sentence - output['Count'] = num - output['Paraphrased Questions'] = final_outputs - - if self.device.type == 'cuda': - torch.cuda.empty_cache() - - return output - -class BoolQGenerator: - - def __init__(self): - self.tokenizer = T5Tokenizer.from_pretrained('t5-base') - self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Boolean-Questions') - self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - self.model.to(self.device) - self.set_seed(42) - - def set_seed(self, seed): - np.random.seed(seed) - torch.manual_seed(seed) - if torch.cuda.is_available(): - torch.cuda.manual_seed_all(seed) - - def random_choice(self): - a = random.choice([0,1]) - return bool(a) - - - def generate_boolq(self, payload): - start_time = time.time() - inp = { - "input_text": payload.get("input_text"), - "max_questions": payload.get("max_questions", 4) - } - - text = inp['input_text'] - num= inp['max_questions'] - sentences = tokenize_into_sentences(text) - modified_text = " ".join(sentences) - answer = self.random_choice() - form = "truefalse: %s passage: %s " % (modified_text, answer) - print(form) - encoding = self.tokenizer.encode_plus(form, return_tensors="pt") - input_ids, attention_masks = encoding["input_ids"].to(self.device), encoding["attention_mask"].to(self.device) - - output = beam_search_decoding (input_ids, attention_masks, self.model, self.tokenizer,num) - if self.device.type == 'cuda': - torch.cuda.empty_cache() - - final= {} - final['Text']= text - final['Count']= num - final['Boolean_Questions']= output - - return final - - -class AnswerPredictor: - - def __init__(self): - self.tokenizer = T5Tokenizer.from_pretrained('t5-large', model_max_length=512) - self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Answer-Predictor') - self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - self.model.to(self.device) - - # Load the lightweight NLI model for boolean question answering - self.nli_model_name = "typeform/distilbert-base-uncased-mnli" - self.nli_tokenizer = AutoTokenizer.from_pretrained(self.nli_model_name) - self.nli_model = AutoModelForSequenceClassification.from_pretrained(self.nli_model_name) - - self.set_seed(42) - - def set_seed(self, seed): - np.random.seed(seed) - torch.manual_seed(seed) - if torch.cuda.is_available(): - torch.cuda.manual_seed_all(seed) - - def greedy_decoding(self, inp_ids, attn_mask): - greedy_output = self.model.generate(input_ids=inp_ids, attention_mask=attn_mask, max_length=256) - Question = self.tokenizer.decode(greedy_output[0], skip_special_tokens=True, clean_up_tokenization_spaces=True) - return Question.strip().capitalize() - - def predict_answer(self, payload): - answers = [] - inp = { - "input_text": payload.get("input_text"), - "input_question" : payload.get("input_question") - } - for ques in payload.get("input_question"): - - context = inp["input_text"] - question = ques - input_text = "question: %s context: %s " % (question, context) - - encoding = self.tokenizer.encode_plus(input_text, return_tensors="pt") - input_ids, attention_masks = encoding["input_ids"].to(self.device), encoding["attention_mask"].to(self.device) - greedy_output = self.model.generate(input_ids=input_ids, attention_mask=attention_masks, max_length=256) - Question = self.tokenizer.decode(greedy_output[0], skip_special_tokens=True, clean_up_tokenization_spaces=True) - answers.append(Question.strip().capitalize()) - - if self.device.type == 'cuda': - torch.cuda.empty_cache() - - return answers - - def predict_boolean_answer(self, payload): - input_text = payload.get("input_text", "") - input_questions = payload.get("input_question", []) - - answers = [] - - for question in input_questions: - hypothesis = question - inputs = self.nli_tokenizer.encode_plus(input_text, hypothesis, return_tensors="pt") - outputs = self.nli_model(**inputs) - logits = outputs.logits - probabilities = torch.softmax(logits, dim=1) - entailment_prob = probabilities[0][0].item() - contradiction_prob = probabilities[0][2].item() - - if entailment_prob > contradiction_prob: - answers.append(True) - else: - answers.append(False) - - if self.device.type == 'cuda': - torch.cuda.empty_cache() - - return answers - -class GoogleDocsService: - def __init__(self, service_account_file, scopes): - self.credentials = service_account.Credentials.from_service_account_file( - service_account_file, scopes=scopes) - self.docs_service = build('docs', 'v1', credentials=self.credentials) - - @staticmethod - def extract_document_id(url): - """ - Extracts the Google Docs document ID from a given URL. - """ - match = re.search(r'/document/d/([^/]+)', url) - if match: - return match.group(1) - return None - - def get_document_content(self, document_url): - """ - Retrieves the content of a Google Docs document given its URL. - """ - document_id = self.extract_document_id(document_url) - if not document_id: - raise ValueError('Invalid document URL') - - response = self.docs_service.documents().get(documentId=document_id).execute() - doc = response.get('body', {}) - - text = '' - for element in doc.get('content', []): - if 'paragraph' in element: - for p in element['paragraph']['elements']: - if 'textRun' in p: - text += p['textRun']['content'] - - return text.strip() - - -class FileProcessor: - def __init__(self, upload_folder='uploads/'): - self.upload_folder = upload_folder - if not os.path.exists(self.upload_folder): - os.makedirs(self.upload_folder) - - def extract_text_from_pdf(self, file_path): - doc = fitz.open(file_path) - text = "" - for page in doc: - text += page.get_text() - return text - - def extract_text_from_docx(self, file_path): - with open(file_path, "rb") as docx_file: - result = mammoth.extract_raw_text(docx_file) - return result.value - - def process_file(self, file): - file_path = os.path.join(self.upload_folder, file.filename) - file.save(file_path) - content = "" - - if file.filename.endswith('.txt'): - with open(file_path, 'r') as f: - content = f.read() - elif file.filename.endswith('.pdf'): - content = self.extract_text_from_pdf(file_path) - elif file.filename.endswith('.docx'): - content = self.extract_text_from_docx(file_path) - - os.remove(file_path) - return content - -class QuestionGenerator: - """A transformer-based NLP system for generating reading comprehension-style questions from - texts. It can generate full sentence questions, multiple choice questions, or a mix of the - two styles. - - To filter out low quality questions, questions are assigned a score and ranked once they have - been generated. Only the top k questions will be returned. This behaviour can be turned off - by setting use_evaluator=False. - """ - - def __init__(self) -> None: - - QG_PRETRAINED = "iarfmoose/t5-base-question-generator" - self.ANSWER_TOKEN = "" - self.CONTEXT_TOKEN = "" - self.SEQ_LENGTH = 512 - - self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - - self.qg_tokenizer = AutoTokenizer.from_pretrained(QG_PRETRAINED, use_fast=False) - self.qg_model = AutoModelForSeq2SeqLM.from_pretrained(QG_PRETRAINED) - self.qg_model.to(self.device) - self.qg_model.eval() - - self.qa_evaluator = QAEvaluator() - - def generate( - self, - article: str, - use_evaluator: bool = False, - num_questions: bool = None, - answer_style: str = "all", - ) -> List: - """Takes an article and generates a set of question and answer pairs. If use_evaluator - is True then QA pairs will be ranked and filtered based on their quality. answer_style - should selected from ["all", "sentences", "multiple_choice"]. - """ - - print("Generating questions...\n") - - qg_inputs, qg_answers = self.generate_qg_inputs(article, answer_style) - generated_questions = self.generate_questions_from_inputs(qg_inputs) - - message = "{} questions doesn't match {} answers".format( - len(generated_questions), len(qg_answers) - ) - assert len(generated_questions) == len(qg_answers), message - - if use_evaluator: - print("Evaluating QA pairs...\n") - encoded_qa_pairs = self.qa_evaluator.encode_qa_pairs( - generated_questions, qg_answers - ) - scores = self.qa_evaluator.get_scores(encoded_qa_pairs) - - if num_questions: - qa_list = self._get_ranked_qa_pairs( - generated_questions, qg_answers, scores, num_questions - ) - else: - qa_list = self._get_ranked_qa_pairs( - generated_questions, qg_answers, scores - ) - - else: - print("Skipping evaluation step.\n") - qa_list = self._get_all_qa_pairs(generated_questions, qg_answers) - - if self.device.type == 'cuda': - torch.cuda.empty_cache() - - return qa_list - - def generate_qg_inputs( - self, text: str, answer_style: str - ) -> Tuple[List[str], List[str]]: - """Given a text, returns a list of model inputs and a list of corresponding answers. - Model inputs take the form "answer_token context_token " where - the answer is a string extracted from the text, and the context is the wider text surrounding - the context. - """ - - VALID_ANSWER_STYLES = ["all", "sentences", "multiple_choice"] - - if answer_style not in VALID_ANSWER_STYLES: - raise ValueError( - "Invalid answer style {}. Please choose from {}".format( - answer_style, VALID_ANSWER_STYLES - ) - ) - - inputs = [] - answers = [] - - if answer_style == "sentences" or answer_style == "all": - segments = self._split_into_segments(text) - - for segment in segments: - sentences = self._split_text(segment) - prepped_inputs, prepped_answers = self._prepare_qg_inputs( - sentences, segment - ) - inputs.extend(prepped_inputs) - answers.extend(prepped_answers) - - if answer_style == "multiple_choice" or answer_style == "all": - sentences = self._split_text(text) - prepped_inputs, prepped_answers = self._prepare_qg_inputs_MC(sentences) - inputs.extend(prepped_inputs) - answers.extend(prepped_answers) - - return inputs, answers - - def generate_questions_from_inputs(self, qg_inputs: List) -> List[str]: - """Given a list of concatenated answers and contexts, with the form: - "answer_token context_token ", generates a list of - questions. - """ - generated_questions = [] - - for qg_input in qg_inputs: - question = self._generate_question(qg_input) - generated_questions.append(question) - - return generated_questions - - def _split_text(self, text: str) -> List[str]: - """Splits the text into sentences, and attempts to split or truncate long sentences.""" - MAX_SENTENCE_LEN = 128 - sentences = re.findall(".*?[.!\?]", text) - cut_sentences = [] - - for sentence in sentences: - if len(sentence) > MAX_SENTENCE_LEN: - cut_sentences.extend(re.split("[,;:)]", sentence)) - - # remove useless post-quote sentence fragments - cut_sentences = [s for s in sentences if len(s.split(" ")) > 5] - sentences = sentences + cut_sentences - - return list(set([s.strip(" ") for s in sentences])) - - def _split_into_segments(self, text: str) -> List[str]: - """Splits a long text into segments short enough to be input into the transformer network. - Segments are used as context for question generation. - """ - MAX_TOKENS = 490 - paragraphs = text.split("\n") - tokenized_paragraphs = [ - self.qg_tokenizer(p)["input_ids"] for p in paragraphs if len(p) > 0 - ] - segments = [] - - while len(tokenized_paragraphs) > 0: - segment = [] - - while len(segment) < MAX_TOKENS and len(tokenized_paragraphs) > 0: - paragraph = tokenized_paragraphs.pop(0) - segment.extend(paragraph) - segments.append(segment) - - return [self.qg_tokenizer.decode(s, skip_special_tokens=True) for s in segments] - - def _prepare_qg_inputs( - self, sentences: List[str], text: str - ) -> Tuple[List[str], List[str]]: - """Uses sentences as answers and the text as context. Returns a tuple of (model inputs, answers). - Model inputs are "answer_token context_token " - """ - inputs = [] - answers = [] - - for sentence in sentences: - qg_input = f"{self.ANSWER_TOKEN} {sentence} {self.CONTEXT_TOKEN} {text}" - inputs.append(qg_input) - answers.append(sentence) - - return inputs, answers - - def _prepare_qg_inputs_MC( - self, sentences: List[str] - ) -> Tuple[List[str], List[str]]: - """Performs NER on the text, and uses extracted entities are candidate answers for multiple-choice - questions. Sentences are used as context, and entities as answers. Returns a tuple of (model inputs, answers). - Model inputs are "answer_token context_token " - """ - spacy_nlp = en_core_web_sm.load() - docs = list(spacy_nlp.pipe(sentences, disable=["parser"])) - inputs_from_text = [] - answers_from_text = [] - - for doc, sentence in zip(docs, sentences): - entities = doc.ents - if entities: - - for entity in entities: - qg_input = ( - f"{self.ANSWER_TOKEN} {entity} {self.CONTEXT_TOKEN} {sentence}" - ) - answers = self._get_MC_answers(entity, docs) - inputs_from_text.append(qg_input) - answers_from_text.append(answers) - - return inputs_from_text, answers_from_text - - def _get_MC_answers( - self, correct_answer: Any, docs: Any - ) -> List[Mapping[str, Any]]: - """Finds a set of alternative answers for a multiple-choice question. Will attempt to find - alternatives of the same entity type as correct_answer if possible. - """ - entities = [] - - for doc in docs: - entities.extend([{"text": e.text, "label_": e.label_} for e in doc.ents]) - - # remove duplicate elements - entities_json = [json.dumps(kv) for kv in entities] - pool = set(entities_json) - num_choices = ( - min(4, len(pool)) - 1 - ) # -1 because we already have the correct answer - - # add the correct answer - final_choices = [] - correct_label = correct_answer.label_ - final_choices.append({"answer": correct_answer.text, "correct": True}) - pool.remove( - json.dumps({"text": correct_answer.text, "label_": correct_answer.label_}) - ) - - # find answers with the same NER label - matches = [e for e in pool if correct_label in e] - - # if we don't have enough then add some other random answers - if len(matches) < num_choices: - choices = matches - pool = pool.difference(set(choices)) - choices.extend(random.sample(list(pool), num_choices - len(choices))) - else: - choices = random.sample(matches, num_choices) - - choices = [json.loads(s) for s in choices] - - for choice in choices: - final_choices.append({"answer": choice["text"], "correct": False}) - - random.shuffle(final_choices) - return final_choices - - @torch.no_grad() - def _generate_question(self, qg_input: str) -> str: - """Takes qg_input which is the concatenated answer and context, and uses it to generate - a question sentence. The generated question is decoded and then returned. - """ - encoded_input = self._encode_qg_input(qg_input) - output = self.qg_model.generate(input_ids=encoded_input["input_ids"]) - question = self.qg_tokenizer.decode(output[0], skip_special_tokens=True) - return question - - def _encode_qg_input(self, qg_input: str) -> torch.tensor: - """Tokenizes a string and returns a tensor of input ids corresponding to indices of tokens in - the vocab. - """ - return self.qg_tokenizer( - qg_input, - padding="max_length", - max_length=self.SEQ_LENGTH, - truncation=True, - return_tensors="pt", - ).to(self.device) - - def _get_ranked_qa_pairs( - self, - generated_questions: List[str], - qg_answers: List[str], - scores, - num_questions: int = 10, - ) -> List[Mapping[str, str]]: - """Ranks generated questions according to scores, and returns the top num_questions examples.""" - if num_questions > len(scores): - num_questions = len(scores) - print( - ( - f"\nWas only able to generate {num_questions} questions.", - "For more questions, please input a longer text.", - ) - ) - - qa_list = [] - - for i in range(num_questions): - index = scores[i] - qa = { - "question": generated_questions[index].split("?")[0] + "?", - "answer": qg_answers[index], - } - qa_list.append(qa) - - return qa_list - - def _get_all_qa_pairs(self, generated_questions: List[str], qg_answers: List[str]): - """Formats question and answer pairs without ranking or filtering.""" - qa_list = [] - - for question, answer in zip(generated_questions, qg_answers): - qa = {"question": question.split("?")[0] + "?", "answer": answer} - qa_list.append(qa) - - return qa_list - - -class QAEvaluator: - """Wrapper for a transformer model which evaluates the quality of question-answer pairs. - Given a QA pair, the model will generate a score. Scores can be used to rank and filter - QA pairs. - """ - - def __init__(self) -> None: - - QAE_PRETRAINED = "iarfmoose/bert-base-cased-qa-evaluator" - self.SEQ_LENGTH = 512 - - self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - - self.qae_tokenizer = AutoTokenizer.from_pretrained(QAE_PRETRAINED) - self.qae_model = AutoModelForSequenceClassification.from_pretrained( - QAE_PRETRAINED - ) - self.qae_model.to(self.device) - self.qae_model.eval() - - def encode_qa_pairs( - self, questions: List[str], answers: List[str] - ) -> List[torch.tensor]: - """Takes a list of questions and a list of answers and encodes them as a list of tensors.""" - encoded_pairs = [] - - for question, answer in zip(questions, answers): - encoded_qa = self._encode_qa(question, answer) - encoded_pairs.append(encoded_qa.to(self.device)) - - return encoded_pairs - - def get_scores(self, encoded_qa_pairs: List[torch.tensor]) -> List[float]: - """Generates scores for a list of encoded QA pairs.""" - scores = {} - - for i in range(len(encoded_qa_pairs)): - scores[i] = self._evaluate_qa(encoded_qa_pairs[i]) - - if self.device.type == 'cuda': - torch.cuda.empty_cache() - - return [ - k for k, v in sorted(scores.items(), key=lambda item: item[1], reverse=True) - ] - - def _encode_qa(self, question: str, answer: str) -> torch.tensor: - """Concatenates a question and answer, and then tokenizes them. Returns a tensor of - input ids corresponding to indices in the vocab. - """ - if type(answer) is list: - for a in answer: - if a["correct"]: - correct_answer = a["answer"] - else: - correct_answer = answer - - return self.qae_tokenizer( - text=question, - text_pair=correct_answer, - padding="max_length", - max_length=self.SEQ_LENGTH, - truncation=True, - return_tensors="pt", - ) - - @torch.no_grad() - def _evaluate_qa(self, encoded_qa_pair: torch.tensor) -> float: - """Takes an encoded QA pair and returns a score.""" - output = self.qae_model(**encoded_qa_pair) - return output[0][0][1] - - -def print_qa(qa_list: List[Mapping[str, str]], show_answers: bool = True) -> None: - """Formats and prints a list of generated questions and answers.""" - - for i in range(len(qa_list)): - # wider space for 2 digit q nums - space = " " * int(np.where(i < 9, 3, 4)) - - print(f"{i + 1}) Q: {qa_list[i]['question']}") - - answer = qa_list[i]["answer"] - - # print a list of multiple choice answers - if type(answer) is list: - - if show_answers: - print( - f"{space}A: 1. {answer[0]['answer']} " - f"{np.where(answer[0]['correct'], '(correct)', '')}" - ) - for j in range(1, len(answer)): - print( - f"{space + ' '}{j + 1}. {answer[j]['answer']} " - f"{np.where(answer[j]['correct']==True,'(correct)', '')}" - ) - - else: - print(f"{space}A: 1. {answer[0]['answer']}") - for j in range(1, len(answer)): - print(f"{space + ' '}{j + 1}. {answer[j]['answer']}") - - print("") - - # print full sentence answers - else: - if show_answers: - print(f"{space}A: {answer}\n") diff --git a/backend/Generator/question_generators.py b/backend/Generator/question_generators.py new file mode 100644 index 00000000..63b888c1 --- /dev/null +++ b/backend/Generator/question_generators.py @@ -0,0 +1,287 @@ +import time +import random + +import torch +import numpy as np +import spacy +from sense2vec import Sense2Vec +from transformers import T5ForConditionalGeneration, T5Tokenizer +from nltk import FreqDist +from nltk.corpus import brown +from similarity.normalized_levenshtein import NormalizedLevenshtein + +from Generator.mcq import ( + tokenize_into_sentences, + identify_keywords, + find_sentences_with_keywords, + generate_multiple_choice_questions, + generate_normal_questions, +) +from Generator.encoding import beam_search_decoding + + +class MCQGenerator: + + def __init__(self): + """Loads the MCQ question-generation model plus the NLP tools used for keyword extraction.""" + self.tokenizer = T5Tokenizer.from_pretrained('t5-large') + self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Question-Generator') + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + self.model.to(self.device) + self.nlp = spacy.load('en_core_web_sm') + self.s2v = Sense2Vec().from_disk('s2v_old') + self.fdist = FreqDist(brown.words()) + self.normalized_levenshtein = NormalizedLevenshtein() + self.set_seed(42) + + def set_seed(self, seed): + """Seeds NumPy, torch and CUDA RNGs so that generation is reproducible.""" + np.random.seed(seed) + torch.manual_seed(seed) + if torch.cuda.is_available(): + torch.cuda.manual_seed_all(seed) + + def generate_mcq(self, payload): + """Generates multiple-choice questions from the input text. + + Args: + payload: Dict with ``input_text`` (the source text) and an optional + ``max_questions`` (number of keywords to base questions on, default 4). + + Returns: + A dict with ``statement`` (the processed text), ``questions`` (the generated + MCQs) and ``time_taken``; empty if no keywords could be extracted. + """ + start_time = time.time() + inp = { + "input_text": payload.get("input_text"), + "max_questions": payload.get("max_questions", 4) + } + + text = inp['input_text'] + sentences = tokenize_into_sentences(text) + modified_text = " ".join(sentences) + + keywords = identify_keywords(self.nlp, modified_text, inp['max_questions'], self.s2v, self.fdist, self.normalized_levenshtein, len(sentences)) + keyword_sentence_mapping = find_sentences_with_keywords(keywords, sentences) + + for k in keyword_sentence_mapping.keys(): + text_snippet = " ".join(keyword_sentence_mapping[k][:3]) + keyword_sentence_mapping[k] = text_snippet + + final_output = {} + + if len(keyword_sentence_mapping.keys()) == 0: + return final_output + else: + try: + generated_questions = generate_multiple_choice_questions(keyword_sentence_mapping, self.device, self.tokenizer, self.model, self.s2v, self.normalized_levenshtein) + except: + return final_output + + end_time = time.time() + + final_output["statement"] = modified_text + final_output["questions"] = generated_questions["questions"] + final_output["time_taken"] = end_time - start_time + + if self.device.type == 'cuda': + torch.cuda.empty_cache() + + return final_output + + +class ShortQGenerator: + + def __init__(self): + """Loads the short-question generation model plus the NLP tools used for keyword extraction.""" + self.tokenizer = T5Tokenizer.from_pretrained('t5-large') + self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Question-Generator') + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + self.model.to(self.device) + self.nlp = spacy.load('en_core_web_sm') + self.s2v = Sense2Vec().from_disk('s2v_old') + self.fdist = FreqDist(brown.words()) + self.normalized_levenshtein = NormalizedLevenshtein() + self.set_seed(42) + + def set_seed(self, seed): + """Seeds NumPy, torch and CUDA RNGs so that generation is reproducible.""" + np.random.seed(seed) + torch.manual_seed(seed) + if torch.cuda.is_available(): + torch.cuda.manual_seed_all(seed) + + def generate_shortq(self, payload): + """Generates short-answer questions from the input text. + + Args: + payload: Dict with ``input_text`` (the source text) and an optional + ``max_questions`` (number of keywords to base questions on, default 4). + + Returns: + A dict with ``statement`` (the processed text) and ``questions`` (the generated + short questions); empty if no keywords could be extracted. + """ + inp = { + "input_text": payload.get("input_text"), + "max_questions": payload.get("max_questions", 4) + } + + text = inp['input_text'] + sentences = tokenize_into_sentences(text) + modified_text = " ".join(sentences) + + keywords = identify_keywords(self.nlp, modified_text, inp['max_questions'], self.s2v, self.fdist, self.normalized_levenshtein, len(sentences)) + keyword_sentence_mapping = find_sentences_with_keywords(keywords, sentences) + + for k in keyword_sentence_mapping.keys(): + text_snippet = " ".join(keyword_sentence_mapping[k][:3]) + keyword_sentence_mapping[k] = text_snippet + + final_output = {} + + if len(keyword_sentence_mapping.keys()) == 0: + return final_output + else: + generated_questions = generate_normal_questions(keyword_sentence_mapping, self.device, self.tokenizer, self.model) + + final_output["statement"] = modified_text + final_output["questions"] = generated_questions["questions"] + + if self.device.type == 'cuda': + torch.cuda.empty_cache() + + return final_output + + +class ParaphraseGenerator: + + def __init__(self): + """Loads the paraphrasing model and tokenizer.""" + self.tokenizer = T5Tokenizer.from_pretrained('t5-large') + self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Question-Generator') + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + self.model.to(self.device) + self.set_seed(42) + + def set_seed(self, seed): + """Seeds NumPy, torch and CUDA RNGs so that generation is reproducible.""" + np.random.seed(seed) + torch.manual_seed(seed) + if torch.cuda.is_available(): + torch.cuda.manual_seed_all(seed) + + def generate_paraphrase(self, payload): + """Generates paraphrases of the input sentence. + + Args: + payload: Dict with ``input_text`` (the sentence to paraphrase) and an optional + ``max_questions`` (number of paraphrases to produce, default 3). + + Returns: + A dict with ``Original Sentence``, ``Count`` and ``Paraphrased Questions`` + (the deduplicated, distinct paraphrases). + """ + start_time = time.time() + inp = { + "input_text": payload.get("input_text"), + "max_questions": payload.get("max_questions", 3) + } + + text = inp['input_text'] + num = inp['max_questions'] + + sentence = text + text_to_paraphrase = "paraphrase: " + sentence + " " + + encoding = self.tokenizer.encode_plus(text_to_paraphrase, pad_to_max_length=True, return_tensors="pt") + input_ids, attention_masks = encoding["input_ids"].to(self.device), encoding["attention_mask"].to(self.device) + + beam_outputs = self.model.generate( + input_ids=input_ids, + attention_mask=attention_masks, + max_length=50, + num_beams=50, + num_return_sequences=num, + no_repeat_ngram_size=2, + early_stopping=True + ) + + final_outputs =[] + for beam_output in beam_outputs: + paraphrased_sentence = self.tokenizer.decode(beam_output, skip_special_tokens=True, clean_up_tokenization_spaces=True) + if paraphrased_sentence.lower() != sentence.lower() and paraphrased_sentence not in final_outputs: + final_outputs.append(paraphrased_sentence) + + output = {} + output['Original Sentence'] = sentence + output['Count'] = num + output['Paraphrased Questions'] = final_outputs + + if self.device.type == 'cuda': + torch.cuda.empty_cache() + + return output + + +class BoolQGenerator: + + def __init__(self): + """Loads the boolean-question generation model and tokenizer.""" + self.tokenizer = T5Tokenizer.from_pretrained('t5-base') + self.model = T5ForConditionalGeneration.from_pretrained('Roasters/Boolean-Questions') + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + self.model.to(self.device) + self.set_seed(42) + + def set_seed(self, seed): + """Seeds NumPy, torch and CUDA RNGs so that generation is reproducible.""" + np.random.seed(seed) + torch.manual_seed(seed) + if torch.cuda.is_available(): + torch.cuda.manual_seed_all(seed) + + def random_choice(self): + """Returns a random boolean, used as the answer for generated boolean questions.""" + a = random.choice([0,1]) + return bool(a) + + + def generate_boolq(self, payload): + """Generates boolean (yes/no) questions from the input text. + + Args: + payload: Dict with ``input_text`` (the source text) and an optional + ``max_questions`` (number of questions to generate, default 4). + + Returns: + A dict with ``Text`` (the source text), ``Count`` (number of questions) + and ``Boolean_Questions`` (the generated questions). + """ + start_time = time.time() + inp = { + "input_text": payload.get("input_text"), + "max_questions": payload.get("max_questions", 4) + } + + text = inp['input_text'] + num= inp['max_questions'] + sentences = tokenize_into_sentences(text) + modified_text = " ".join(sentences) + answer = self.random_choice() + form = "truefalse: %s passage: %s " % (modified_text, answer) + print(form) + encoding = self.tokenizer.encode_plus(form, return_tensors="pt") + input_ids, attention_masks = encoding["input_ids"].to(self.device), encoding["attention_mask"].to(self.device) + + output = beam_search_decoding (input_ids, attention_masks, self.model, self.tokenizer,num) + if self.device.type == 'cuda': + torch.cuda.empty_cache() + + final= {} + final['Text']= text + final['Count']= num + final['Boolean_Questions']= output + + return final diff --git a/backend/Generator/utilities.py b/backend/Generator/utilities.py new file mode 100644 index 00000000..6c859aa9 --- /dev/null +++ b/backend/Generator/utilities.py @@ -0,0 +1,130 @@ +import os +import re + +import numpy as np +import fitz +import mammoth +from google.oauth2 import service_account +from googleapiclient.discovery import build +from typing import List, Mapping + + +class GoogleDocsService: + def __init__(self, service_account_file, scopes): + """Authenticates with Google Docs using a service account and builds the API client.""" + self.credentials = service_account.Credentials.from_service_account_file( + service_account_file, scopes=scopes) + self.docs_service = build('docs', 'v1', credentials=self.credentials) + + @staticmethod + def extract_document_id(url): + """ + Extracts the Google Docs document ID from a given URL. + """ + match = re.search(r'/document/d/([^/]+)', url) + if match: + return match.group(1) + return None + + def get_document_content(self, document_url): + """ + Retrieves the content of a Google Docs document given its URL. + """ + document_id = self.extract_document_id(document_url) + if not document_id: + raise ValueError('Invalid document URL') + + response = self.docs_service.documents().get(documentId=document_id).execute() + doc = response.get('body', {}) + + text = '' + for element in doc.get('content', []): + if 'paragraph' in element: + for p in element['paragraph']['elements']: + if 'textRun' in p: + text += p['textRun']['content'] + + return text.strip() + + +class FileProcessor: + def __init__(self, upload_folder='uploads/'): + """Creates the upload folder if it does not already exist.""" + self.upload_folder = upload_folder + if not os.path.exists(self.upload_folder): + os.makedirs(self.upload_folder) + + def extract_text_from_pdf(self, file_path): + """Returns the concatenated text content of a PDF file at ``file_path``.""" + doc = fitz.open(file_path) + text = "" + for page in doc: + text += page.get_text() + return text + + def extract_text_from_docx(self, file_path): + """Returns the raw text content of a .docx file at ``file_path``.""" + with open(file_path, "rb") as docx_file: + result = mammoth.extract_raw_text(docx_file) + return result.value + + def process_file(self, file): + """Saves an uploaded file and returns its extracted text content. + + The file is stored under its client-supplied name inside the upload + folder, deleted after extraction, and dispatched on its extension + (.txt, .pdf or .docx). Unsupported extensions yield an empty string, + which the /upload route reports as a 400 error. + """ + file_path = os.path.join(self.upload_folder, file.filename) + file.save(file_path) + content = "" + + if file.filename.endswith('.txt'): + with open(file_path, 'r') as f: + content = f.read() + elif file.filename.endswith('.pdf'): + content = self.extract_text_from_pdf(file_path) + elif file.filename.endswith('.docx'): + content = self.extract_text_from_docx(file_path) + + os.remove(file_path) + return content + + +def print_qa(qa_list: List[Mapping[str, str]], show_answers: bool = True) -> None: + """Formats and prints a list of generated questions and answers.""" + + for i in range(len(qa_list)): + # wider space for 2 digit q nums + space = " " * int(np.where(i < 9, 3, 4)) + + print(f"{i + 1}) Q: {qa_list[i]['question']}") + + answer = qa_list[i]["answer"] + + # print a list of multiple choice answers + if type(answer) is list: + + if show_answers: + print( + f"{space}A: 1. {answer[0]['answer']} " + f"{np.where(answer[0]['correct'], '(correct)', '')}" + ) + for j in range(1, len(answer)): + print( + f"{space + ' '}{j + 1}. {answer[j]['answer']} " + f"{np.where(answer[j]['correct']==True,'(correct)', '')}" + ) + + else: + print(f"{space}A: 1. {answer[0]['answer']}") + for j in range(1, len(answer)): + print(f"{space + ' '}{j + 1}. {answer[j]['answer']}") + + print("") + + # print full sentence answers + else: + if show_answers: + print(f"{space}A: {answer}\n") diff --git a/backend/server.py b/backend/server.py index 683c1241..c677acc7 100644 --- a/backend/server.py +++ b/backend/server.py @@ -10,7 +10,10 @@ from sklearn.feature_extraction.text import TfidfVectorizer nltk.download("stopwords") nltk.download('punkt_tab') -from Generator import main +from Generator.question_generators import MCQGenerator, BoolQGenerator, ShortQGenerator +from Generator.answer_predictor import AnswerPredictor +from Generator.advanced_qa import QuestionGenerator +from Generator.utilities import GoogleDocsService, FileProcessor from Generator.question_filters import make_question_harder from Generator.llm_generator import LLMQuestionGenerator import re @@ -34,19 +37,20 @@ SERVICE_ACCOUNT_FILE = './service_account_key.json' SCOPES = ['https://www.googleapis.com/auth/documents.readonly'] -MCQGen = main.MCQGenerator() -answer = main.AnswerPredictor() -BoolQGen = main.BoolQGenerator() -ShortQGen = main.ShortQGenerator() -qg = main.QuestionGenerator() -docs_service = main.GoogleDocsService(SERVICE_ACCOUNT_FILE, SCOPES) -file_processor = main.FileProcessor() +MCQGen = MCQGenerator() +answer = AnswerPredictor() +BoolQGen = BoolQGenerator() +ShortQGen = ShortQGenerator() +qg = QuestionGenerator() +docs_service = GoogleDocsService(SERVICE_ACCOUNT_FILE, SCOPES) +file_processor = FileProcessor() mediawikiapi = MediaWikiAPI() qa_model = pipeline("question-answering") llm_generator = LLMQuestionGenerator() def process_input_text(input_text, use_mediawiki): + """Returns a MediaWiki summary of ``input_text`` when ``use_mediawiki`` is 1, else the text unchanged.""" if use_mediawiki == 1: input_text = mediawikiapi.summary(input_text,8) return input_text @@ -54,6 +58,7 @@ def process_input_text(input_text, use_mediawiki): @app.route("/get_mcq", methods=["POST"]) def get_mcq(): + """POST /get_mcq - generates multiple-choice questions from the input text.""" data = request.get_json() input_text = data.get("input_text", "") use_mediawiki = data.get("use_mediawiki", 0) @@ -68,6 +73,7 @@ def get_mcq(): @app.route("/get_boolq", methods=["POST"]) def get_boolq(): + """POST /get_boolq - generates boolean (yes/no) questions from the input text.""" data = request.get_json() input_text = data.get("input_text", "") use_mediawiki = data.get("use_mediawiki", 0) @@ -82,6 +88,7 @@ def get_boolq(): @app.route("/get_shortq", methods=["POST"]) def get_shortq(): + """POST /get_shortq - generates short-answer questions from the input text.""" data = request.get_json() input_text = data.get("input_text", "") use_mediawiki = data.get("use_mediawiki", 0) @@ -96,6 +103,7 @@ def get_shortq(): @app.route("/get_shortq_llm", methods=["POST"]) def get_shortq_llm(): + """POST /get_shortq_llm - generates short-answer questions using the LLM pipeline.""" try: data = request.get_json() input_text = data.get("input_text", "") @@ -111,6 +119,7 @@ def get_shortq_llm(): @app.route("/get_mcq_llm", methods=["POST"]) def get_mcq_llm(): + """POST /get_mcq_llm - generates multiple-choice questions using the LLM pipeline.""" try: data = request.get_json() input_text = data.get("input_text", "") @@ -126,6 +135,7 @@ def get_mcq_llm(): @app.route("/get_boolq_llm", methods=["POST"]) def get_boolq_llm(): + """POST /get_boolq_llm - generates boolean (yes/no) questions using the LLM pipeline.""" try: data = request.get_json() input_text = data.get("input_text", "") @@ -141,6 +151,7 @@ def get_boolq_llm(): @app.route("/get_problems_llm", methods=["POST"]) def get_problems_llm(): + """POST /get_problems_llm - generates a mixed problem set (MCQ, boolean, short) via the LLM pipeline.""" try: data = request.get_json() input_text = data.get("input_text", "") @@ -158,6 +169,7 @@ def get_problems_llm(): @app.route("/get_problems", methods=["POST"]) def get_problems(): + """POST /get_problems - generates MCQ, boolean and short questions from the input text.""" data = request.get_json() input_text = data.get("input_text", "") use_mediawiki = data.get("use_mediawiki", 0) @@ -180,6 +192,12 @@ def get_problems(): @app.route("/get_mcq_answer", methods=["POST"]) def get_mcq_answer(): + """POST /get_mcq_answer - picks the option closest to the QA model's answer for each MCQ. + + Accepts ``input_text``, ``input_question`` and ``input_options`` (a list of option + lists, aligned with the questions). Returns the option with the highest TF-IDF + cosine similarity to the generated answer for each question. + """ data = request.get_json() input_text = data.get("input_text", "") input_questions = data.get("input_question", []) @@ -213,6 +231,7 @@ def get_mcq_answer(): @app.route("/get_shortq_answer", methods=["POST"]) def get_answer(): + """POST /get_shortq_answer - answers each short question against the input text with the QA model.""" data = request.get_json() input_text = data.get("input_text", "") input_questions = data.get("input_question", []) @@ -226,6 +245,7 @@ def get_answer(): @app.route("/get_boolean_answer", methods=["POST"]) def get_boolean_answer(): + """POST /get_boolean_answer - returns True/False for each yes/no question via the NLI model.""" data = request.get_json() input_text = data.get("input_text", "") input_questions = data.get("input_question", []) @@ -245,6 +265,7 @@ def get_boolean_answer(): @app.route('/get_content', methods=['POST']) def get_content(): + """POST /get_content - fetches and returns the text content of a Google Docs document.""" try: data = request.get_json() document_url = data.get('document_url') @@ -263,6 +284,12 @@ def get_content(): @app.route("/generate_gform", methods=["POST"]) def generate_gform(): + """POST /generate_gform - creates a Google Form from the supplied question/answer pairs. + + Authenticates with the Google Forms API via the local ``token.json``/ + ``credentials.json`` OAuth flow, then creates a form titled "EduAid form" + populated with the given QA pairs. + """ data = request.get_json() qa_pairs = data.get("qa_pairs", "") question_type = data.get("question_type", "") @@ -432,6 +459,7 @@ def generate_gform(): @app.route("/get_shortq_hard", methods=["POST"]) def get_shortq_hard(): + """POST /get_shortq_hard - generates short questions with the advanced QG model and makes them harder.""" data = request.get_json() input_text = data.get("input_text", "") use_mediawiki = data.get("use_mediawiki", 0) @@ -450,6 +478,7 @@ def get_shortq_hard(): @app.route("/get_mcq_hard", methods=["POST"]) def get_mcq_hard(): + """POST /get_mcq_hard - generates MCQs with the advanced QG model and makes them harder.""" data = request.get_json() input_text = data.get("input_text", "") use_mediawiki = data.get("use_mediawiki", 0) @@ -466,6 +495,7 @@ def get_mcq_hard(): @app.route("/get_boolq_hard", methods=["POST"]) def get_boolq_hard(): + """POST /get_boolq_hard - generates boolean questions with the advanced QG model and makes them harder.""" data = request.get_json() input_text = data.get("input_text", "") use_mediawiki = data.get("use_mediawiki", 0) @@ -487,6 +517,11 @@ def get_boolq_hard(): @app.route('/upload', methods=['POST']) def upload_file(): + """POST /upload - extracts text from an uploaded .txt/.pdf/.docx file. + + Returns the extracted content, or a 400 error when no file is supplied or + the file type is unsupported. + """ if 'file' not in request.files: return jsonify({"error": "No file part"}), 400 @@ -504,6 +539,7 @@ def upload_file(): @app.route("/", methods=["GET"]) def hello(): + """GET / - health check endpoint.""" return "The server is working fine" def clean_transcript(file_path): @@ -535,6 +571,11 @@ def clean_transcript(file_path): @app.route('/getTranscript', methods=['GET']) def get_transcript(): + """GET /getTranscript - downloads a YouTube video's English subtitles and returns the cleaned transcript. + + Requires a ``videoId`` query parameter. Returns a 400 when the ID is + missing and a 404 when no subtitles could be downloaded. + """ video_id = request.args.get('videoId') if not video_id: return jsonify({"error": "No video ID provided"}), 400