Upload 3 files

LogRegBuzzer.py

@@ -0,0 +1,93 @@
+from sklearn.linear_model import LogisticRegression
+import joblib
+from huggingface_hub import hf_hub_download
+from transformers import pipeline
+import pandas as pd 
+
+
+class LogisticRegressionBuzzer:
+
+    def __init__(self) -> None:
+        self.model = self.load_from_hf_pkl()
+        self.features = BuzzerFeatures()
+
+
+    def load_from_hf_pkl(self) -> LogisticRegression:
+        REPO_ID = "nes470/pipeline-as-repo"
+        FILENAME = "logreg_buzzer_model.pkl"
+
+        model = joblib.load(
+            hf_hub_download(repo_id=REPO_ID, filename=FILENAME)
+        )
+
+        return model 
+    
+    def predict_buzz(self, question, guess):
+        X = self.features.get_features(question, guess)
+
+        X_formatted = pd.DataFrame(X, index=[0])
+        pred = self.model.predict(X_formatted)
+
+        print(pred)
+        #use predict_proba to get confidence probabilities 
+        prob_pred = self.model.predict_proba(X_formatted)
+        print(prob_pred)
+
+        return (pred, float(pred[0]))
+    
+
+    
+
+class BuzzerFeatures:
+    def __init__(self) -> None:
+        self.ner =  pipeline("ner")
+
+    #returns dict with all the features 
+    def get_features(self, question, guess):
+        sent_count = self.sentence_count(question)
+        guess_word_count = self.guess_word_count(guess)
+        guess_has_paren = self.guess_has_paren(guess)
+        guess_length = self.guess_length(guess)
+        guess_entity = self.guess_entity(guess)
+        
+        feats = {'sentence_count':sent_count, 'guess_word_count':guess_word_count,
+              'guess_has_paren':guess_has_paren, 'guess_length':guess_length}
+        
+        X = feats | guess_entity
+
+        return X 
+
+
+
+
+    def sentence_count(self, str):
+        return len(str.split("."))
+
+    def guess_word_count(self, str):
+        return len(str.split("_"))
+
+    def guess_has_paren(self, str):
+        return int("(" in str or ")" in str)
+
+    def guess_length(self, str):
+        return len(str)
+    
+    def guess_entity(self, text):
+        entities = self.ner(text)
+        if len(entities) == 0:
+            type =  ""  # <-- use "None" instead TODO
+        else:
+            type = entities[0]["entity"]
+
+        if type == "":
+            return {'':1, 'I-LOC':0, 'I-MISC':0, 'I-ORG':0, 'I-PER':0}
+        if type == "I-LOC":
+            return {'':0, 'I-LOC':1, 'I-MISC':0, 'I-ORG':0, 'I-PER':0}
+        if type == "I-MISC":
+            return {'':0, 'I-LOC':0, 'I-MISC':1, 'I-ORG':0, 'I-PER':0}
+        if type == "I-ORG":
+            return {'':0, 'I-LOC':0, 'I-MISC':0, 'I-ORG':1, 'I-PER':0}
+        if type == "I-PER":
+            return {'':0, 'I-LOC':0, 'I-MISC':0, 'I-ORG':0, 'I-PER':1}
+            
+

qbmodel.py

@@ -0,0 +1,61 @@
+from typing import List, Tuple
+import nltk
+import sklearn
+from tfidf import TfidfWikiGuesser
+import numpy as np
+import pandas as pd
+from LogRegBuzzer import LogisticRegressionBuzzer
+
+
+class QuizBowlModel:
+
+    def __init__(self, use_hf_pkl = False):
+        """
+        Load your model(s) and whatever else you need in this function.
+
+        Do NOT load your model or resources in the guess_and_buzz() function, 
+        as it will increase latency severely. 
+        """
+        #best accuracy when using wiki_page_text.json
+        self.guesser = TfidfWikiGuesser(wikidump=None, use_hf_pkl= use_hf_pkl) #can specify different wikidump if needed 
+        print("guesser model loaded")
+
+        self.buzzer = LogisticRegressionBuzzer()
+        print("buzzer model loaded")
+
+
+    def guess_and_buzz(self, question_text: List[str]) -> List[Tuple[str, bool]]:
+        """
+        This function accepts a list of question strings, and returns a list of tuples containing
+        strings representing the guess and corresponding booleans representing 
+        whether or not to buzz. 
+
+        So, guess_and_buzz(["This is a question"]) should return [("answer", False)]
+
+        If you are using a deep learning model, try to use batched prediction instead of 
+        iterating using a for loop.
+        """
+
+        answers = []
+        top_guesses = 3 #guesser will return this amount guesses for each question (in sorted confidence)
+
+        for question in question_text:
+            guesses = self.guesser.make_guess(question, num_guesses=top_guesses)
+            # print(f"\n\n\n answered {len(answers)} questions so far \n\n")
+            # print(f"left to answer {len(question_text)-len(answers)} questions \n\n ")
+            # print(f"progress: {(len(answers)/len(question_text)) * 100} \n\n")
+
+            #do the buzzing 
+            buzz = self.buzzer.predict_buzz(question, guesses[0])
+            
+
+            #make a tuple and add to answers list 
+            tup = (guesses[0], buzz[1])
+            print(tup)
+            answers.append(tup)
+            #might neeed to format guees like replace _ with space 
+
+        return answers
+
+
+        

tfidf.py

@@ -0,0 +1,124 @@
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.metrics.pairwise import cosine_similarity
+import numpy as np 
+import json
+import zipfile
+import pickle
+import os
+from nltk.corpus import stopwords
+from nltk.tokenize import word_tokenize
+from nltk.stem import WordNetLemmatizer
+from huggingface_hub import hf_hub_download
+import joblib
+
+
+class TfidfWikiGuesser:
+    def __init__(self, wikidump = 'resources/wiki_text_16.json', use_hf_pkl = False) -> None:
+        self.tfidf = None 
+        self.corpus = None 
+        self.titles = None 
+        self.vectorizer = None 
+        self.lemmatizer = WordNetLemmatizer()
+        model_file = "processed_tfidf_wiki_page_text_model.pkl" # <--- has best acc so far (using wiki_page_text.json from gdrive folder)
+        #model_file = "processed_large_wiki_text_model.pkl"
+        #model_file = "processed_tfidf_wiki_16_model.pkl"
+        # full_model_path = model_file
+        full_model_path = os.path.join("./models", model_file)
+
+        if(use_hf_pkl):
+            REPO_ID = "nes470/pipeline-as-repo"
+            FILENAME = "processed_tfidf_wiki_page_text_model.pkl"
+
+            model = joblib.load(
+                hf_hub_download(repo_id=REPO_ID, filename=FILENAME)
+            )
+            
+        
+            print("loading from hugginface pkl file")
+            self.load_from_pk_direct(model)
+        else:
+            if os.path.exists(full_model_path):
+                print("Loading model from pickle...")
+                self.load_from_pkl(full_model_path)
+            else:
+                if wikidump:
+                    print("No pre-trained model found, loading data from dump...")
+                    self.load_model(wikidump)
+                    self.save_model(full_model_path)
+        # self.load_model(wikidump)
+
+    def load_model(self, wikidump):
+        # wiki dump is an json array of json objects with page and text fields 
+        with open(wikidump) as f:
+            doc = json.load(f)
+        # with zipfile.ZipFile('resources/wiki_text_8.json.zip', 'r') as z:
+        #     with z.open('wiki_text_8.json') as f:
+        #         doc = json.load(f)
+
+
+        self.corpus, self.titles = self.create_corpus(doc)
+
+        self.vectorizer = TfidfVectorizer(stop_words='english')
+        self.tfidf = self.vectorizer.fit_transform(self.corpus)
+
+    def preprocess_text(self,text):
+        if type(text) == float:
+            return str(text)
+        tokens = word_tokenize(text.lower())
+        filtered_tokens = [token for token in tokens if token not in stopwords.words('english')]
+        lemmatized_tokens = [self.lemmatizer.lemmatize(token) for token in filtered_tokens]
+        processed_text = ' '.join(lemmatized_tokens)
+        return processed_text
+
+    def create_corpus(self, json_file):
+        corpus = []
+        page_titles = []
+        
+        for json_obj in json_file:
+            # corpus.append(json_obj['text'])
+            #corpus.append(self.preprocess_text(json_obj['text']))
+            corpus.append(json_obj['text'])
+            page_titles.append(json_obj['page'])
+
+        return (corpus, page_titles)
+
+    def make_guess(self, question, num_guesses = 1):
+        tfidf_question = self.vectorizer.transform([question])
+
+        sim = cosine_similarity(self.tfidf, tfidf_question) 
+
+        #get indices of best matching documents and use it to get (num_guesses) top documents 
+        sim_indices = np.argsort(sim.flatten())[::-1]
+        best_indices = sim_indices[:num_guesses]
+        
+        # best_docs = []
+        best_guesses = []
+        for i in best_indices:
+            # best_docs.append(self.corpus[i])
+            best_guesses.append(self.titles[i])
+
+        return best_guesses
+        
+    def save_model(self, file_name):
+        with open(file_name, 'wb') as f:
+            pickle.dump({
+                'vectorizer': self.vectorizer,
+                'tfidf_matrix': self.tfidf,
+                'titles': self.titles,
+                # 'corpus': self.corpus
+            }, f)
+
+    def load_from_pkl(self, file_name):
+        with open(file_name, 'rb') as f:
+            data = pickle.load(f)
+            self.vectorizer = data['vectorizer']
+            self.tfidf = data['tfidf_matrix']
+            self.titles = data['titles']
+            # self.corpus = data['corpus']
+
+    def load_from_pk_direct(self, pkl):
+        #data = pickle.load(pkl)
+        data = pkl
+        self.vectorizer = data['vectorizer']
+        self.tfidf = data['tfidf_matrix']
+        self.titles = data['titles']