Spaces:
Sleeping
Sleeping
| # -*- coding: utf-8 -*- | |
| """Jan_16_In_Class_Assignment_ECE_UW,_PMP_course_LLM_2024.ipynb | |
| Automatically generated by Colaboratory. | |
| Original file is located at | |
| https://colab.research.google.com/drive/1W2g1PyBwLNE_P_xlBg9C5BfFxiRDyDa2 | |
| # Embeddings and Semantic Search (LLM 2024) | |
| ## This in-class coding exercise is to get hands-on with embeddings and one of its obvious application: Semantic Search. | |
| Search is an area that a lot of companies have invested in. Any retail company has a search engine of its own to serve its products. But how many of them include semantics in search? Search is typically done through Tries. But when we bring semantics to search, the ball game entirely changes. Searching with semantics can help address tail queries whereas Trie searches are usually geared for head queries. | |
| One of the bottlenecks in including semantics in search is latency - The more sophisticated the search, the slower the search inference will be. This is why for semantic search, there is no one-stop solution in a real-world scenario. Even though we have ChatGPT to return amazing results with the right prompting, we know what the latency this will incur, thus making it less viable in this scenario :-) | |
| """ | |
| ''' | |
| from google.colab import drive | |
| drive.mount('/content/drive') | |
| ''' | |
| """## Install dependencies""" | |
| ''' | |
| !pip3 install sentence-transformers | |
| !pip install datasets | |
| !pip install -q streamlit | |
| ''' | |
| """## 1. Embeddings | |
| Work on developing an embeddings class that goes from the simple glove embeddings to the more intricate sentence transformer embeddings | |
| """ | |
| """ | |
| In this code block, you can develop a class for Embeddings - | |
| That can fetch embeddings of different kinds for the purpose of "Semantic Search" | |
| """ | |
| import numpy as np | |
| import requests | |
| import os | |
| import pickle | |
| import streamlit as st | |
| from sentence_transformers import SentenceTransformer | |
| class Embeddings: | |
| def __init__(self): | |
| """ | |
| Initialize the class | |
| """ | |
| self.glove_embeddings_dim = 50 | |
| def download_glove_embeddings(self): | |
| """ | |
| Download glove embeddings from web or from your gdrive if in optimized format | |
| """ | |
| embeddings_temp = "embeddings_50d_temp.npy" | |
| word_index_temp = "word_index_dict_50d_temp.pkl" | |
| def load_glove_embeddings(self, embedding_dimension): | |
| word_index_temp = "word_index_dict_50d_temp.pkl" | |
| embeddings_temp = "embeddings_50d_temp.npy" | |
| # Load word index dictionary | |
| word_index_dict = pickle.load(open(word_index_temp, "rb"), encoding="latin") | |
| # Load embeddings numpy | |
| embeddings = np.load(embeddings_temp) | |
| return word_index_dict, embeddings | |
| def get_glove_embedding(self, word, word_index_dict, embeddings): | |
| """ | |
| Retrieve GloVe embedding of a specific dimension | |
| """ | |
| word = word.lower() | |
| if word in word_index_dict: | |
| return embeddings[word_index_dict[word]] | |
| else: | |
| return np.zeros(self.glove_embeddings_dim) | |
| def embeddings_preprocess(self, word_index_dict, positive_words, negative_words, embeddings): | |
| new_embedding = np.zeros(self.glove_embeddings_dim) | |
| # for negative words | |
| for word in negative_words: | |
| new_embedding -= self.get_glove_embedding(word, word_index_dict, embeddings) | |
| # for positive words | |
| for word in positive_words: | |
| new_embedding += self.get_glove_embedding(word, word_index_dict, embeddings) | |
| return new_embedding | |
| def get_sentence_transformer_embedding(self, sentence, transformer_name="all-MiniLM-L6-v2"): | |
| """ | |
| Encode a sentence using sentence transformer and return embedding | |
| """ | |
| sentenceTransformer = SentenceTransformer(transformer_name) | |
| return sentenceTransformer.encode(sentence) | |
| def get_averaged_glove_embeddings(self, sentence, embeddings_dict): | |
| words = sentence.split(" ") | |
| # Initialize an array of zeros for the embedding | |
| glove_embedding = np.zeros(embeddings_dict['embeddings'].shape[1]) | |
| count_words = 0 | |
| for word in words: | |
| word = word.lower() # Convert to lowercase to match the embeddings dictionary | |
| if word in embeddings_dict['word_index']: | |
| # Sum up embeddings for each word | |
| glove_embedding += embeddings_dict['embeddings'][embeddings_dict['word_index'][word]] | |
| count_words += 1 | |
| if count_words > 0: | |
| # Average the embeddings | |
| glove_embedding /= count_words | |
| return glove_embedding | |
| """## 2. Search Class | |
| Implement a class with all the methods needed for search including cosine similarity | |
| """ | |
| import numpy.linalg as la | |
| import numpy as np | |
| class Search: | |
| def __init__(self, embeddings_model): | |
| self.embeddings_model = embeddings_model | |
| def cosine_similarity(self, x, y): | |
| return np.dot(x,y)/max(la.norm(x)*la.norm(y),1e-3) | |
| def get_topK_similar_categories(self, sentence, categories, top_k=10): | |
| """Return top K most similar categories to a given sentence.""" | |
| sentence_embedding = self.embeddings_model.get_sentence_transformer_embedding(sentence) | |
| similarities = {category: self.cosine_similarity(sentence_embedding, category_embedding) for category, category_embedding in categories.items()} | |
| return dict(sorted(similarities.items(), key=lambda item: item[1], reverse=True)[:top_k]) | |
| def normalize_func(self, vector): | |
| """Normalize a vector.""" | |
| norm = np.linalg.norm(vector) | |
| return vector / norm if norm != 0 else vector | |
| def find_closest_words(self, current_embedding, answer_list, word_index_dict, embeddings): | |
| """Find closest word from answer_list to current_embedding.""" | |
| highest_similarity, closest_answer = -50, None | |
| for choice in answer_list: | |
| choice_embedding = self.embeddings_model.get_glove_embedding(choice, word_index_dict, embeddings) | |
| similarity = self.cosine_similarity(current_embedding, choice_embedding) | |
| if similarity > highest_similarity: | |
| highest_similarity, closest_answer = similarity, choice | |
| return closest_answer | |
| def find_word_as(self, current_relation, target_word, answer_list, word_index_dict, embeddings): | |
| """Find a word analogous to target_word based on current_relation.""" | |
| base_vector_a = self.embeddings_model.get_glove_embedding(current_relation[0], word_index_dict, embeddings) | |
| base_vector_b = self.embeddings_model.get_glove_embedding(current_relation[1], word_index_dict, embeddings) | |
| target_vector = self.embeddings_model.get_glove_embedding(target_word, word_index_dict, embeddings) | |
| ref_difference = self.normalize_func(base_vector_b - base_vector_a) | |
| answer, highest_similarity = None, -50 | |
| for choice in answer_list: | |
| choice_vector = self.embeddings_model.get_glove_embedding(choice, word_index_dict, embeddings) | |
| choice_difference = self.normalize_func(choice_vector - target_vector) | |
| similarity = self.cosine_similarity(ref_difference, choice_difference) | |
| if similarity > highest_similarity: | |
| highest_similarity, answer = similarity, choice | |
| return answer | |
| def find_similarity_scores(self, current_embedding, choices, word_index_dict, embeddings): | |
| """Calculate similarity scores between current_embedding and choices.""" | |
| similarity_scores = {} | |
| for choice in choices: | |
| choice_embedding = self.embeddings_model.get_glove_embedding(choice, word_index_dict, embeddings) | |
| similarity = self.cosine_similarity(current_embedding, choice_embedding) | |
| similarity_scores[choice] = similarity | |
| return similarity_scores | |
| """## 3. Word Arithmetic | |
| Let's test your embeddings. Answer the question below through the search functionality you implemented above | |
| """ | |
| embeddings_model = Embeddings() | |
| search_using_cos = Search(embeddings_model) | |
| word_index_dict, embeddings = embeddings_model.load_glove_embeddings(50) | |
| current_embedding = embeddings_model.embeddings_preprocess( word_index_dict, ["king", "woman"], ["man"], embeddings) | |
| closest_word = search_using_cos.find_closest_words(current_embedding, ["girl", "queen", "princess", "daughter", "mother"], word_index_dict, embeddings ) | |
| print("'King - Man + Woman':", closest_word) | |
| word_index_dict, embeddings = embeddings_model.load_glove_embeddings(50) | |
| closest_word = search_using_cos.find_word_as( ("tesla", "car"), "apple", ["fruit", "vegetable", "gas"], word_index_dict, embeddings) | |
| print("'Tesla:Car as Apple:?': ", closest_word) | |
| """## 4. Plots | |
| Plot the search results as a pie chart with percentages allocated to the likelihood of the category being related to the search input | |
| """ | |
| import matplotlib.pyplot as plt | |
| def plot_pie_chart(category_similarity_scores): | |
| """Plot a pie chart of category similarity scores.""" | |
| categories = list(category_similarity_scores.keys()) | |
| similarities = list(category_similarity_scores.values()) | |
| normalized_similarities = [sim / sum(similarities) for sim in similarities] | |
| fig, ax = plt.subplots() | |
| ax.pie(normalized_similarities, labels=categories, autopct="%1.11f%%", startangle=90) | |
| ax.axis('equal') # Equal aspect ratio ensures the pie chart is circular. | |
| plt.show() | |
| word_index_dict, embeddings = embeddings_model.load_glove_embeddings(50) | |
| # Find the word closest to the vector resulting from "king" - "man" + "woman" | |
| current_embedding = embeddings_model.embeddings_preprocess(word_index_dict, ["king", "woman"], ["man"], embeddings) | |
| # Calculate similarity scores for a set of words and plot them | |
| sim_scores = search_using_cos.find_similarity_scores(current_embedding, ["girl", "queen", "princess", "daughter", "mother"], word_index_dict, embeddings) | |
| plot_pie_chart(sim_scores) | |
| """## 5. Test | |
| Test your pie chart against some of the examples in the demo listed here: | |
| https://categorysearch.streamlit.app or | |
| https://searchdemo.streamlit.app | |
| a) Do the results make sense? | |
| b) Which embedding gives more meaningful results? | |
| """ | |
| input_sentence = "Roses are red, trucks are blue, and Seattle is grey right now" | |
| category_names = ["Flowers", "Colors", "Cars", "Weather", "Food"] | |
| embeddings_model = Embeddings() | |
| word_index_dict, embeddings = embeddings_model.load_glove_embeddings(50) | |
| categories_embedding = {category: embeddings_model.get_sentence_transformer_embedding(category) for category in category_names} | |
| search_instance = Search(embeddings_model) | |
| category_similarity_scores = search_instance.get_topK_similar_categories(input_sentence, categories_embedding) | |
| plot_pie_chart(category_similarity_scores) # Plot and see | |
| """## 6. Bonus (if time permits)! | |
| Create a simple streamlit or equivalent webapp like the link in 5. | |
| This is also part of your Mini-Project 1! | |
| """ | |
| def plot_piechart(sorted_cosine_scores_items): | |
| sorted_cosine_scores = np.array([ | |
| sorted_cosine_scores_items[index][1] | |
| for index in range(len(sorted_cosine_scores_items)) | |
| ] | |
| ) | |
| categories = st.session_state.categories.split(" ") | |
| categories_sorted = [ | |
| categories[sorted_cosine_scores_items[index][0]] | |
| for index in range(len(sorted_cosine_scores_items)) | |
| ] | |
| fig, ax = plt.subplots() | |
| ax.pie(sorted_cosine_scores, labels=categories_sorted, autopct="%1.1f%%") | |
| st.pyplot(fig) # Figure | |
| def plot_piechart_helper(sorted_cosine_scores_items): | |
| sorted_cosine_scores = np.array( | |
| [ | |
| sorted_cosine_scores_items[index][1] | |
| for index in range(len(sorted_cosine_scores_items)) | |
| ] | |
| ) | |
| categories = st.session_state.categories.split(" ") | |
| categories_sorted = [ | |
| categories[sorted_cosine_scores_items[index][0]] | |
| for index in range(len(sorted_cosine_scores_items)) | |
| ] | |
| fig, ax = plt.subplots(figsize=(3, 3)) | |
| my_explode = np.zeros(len(categories_sorted)) | |
| my_explode[0] = 0.2 | |
| if len(categories_sorted) == 3: | |
| my_explode[1] = 0.1 # explode this by 0.2 | |
| elif len(categories_sorted) > 3: | |
| my_explode[2] = 0.05 | |
| ax.pie( | |
| sorted_cosine_scores, | |
| labels=categories_sorted, | |
| autopct="%1.1f%%", | |
| explode=my_explode, | |
| ) | |
| return fig | |
| def plot_piecharts(sorted_cosine_scores_models): | |
| scores_list = [] | |
| categories = st.session_state.categories.split(" ") | |
| index = 0 | |
| for model in sorted_cosine_scores_models: | |
| scores_list.append(sorted_cosine_scores_models[model]) | |
| # scores_list[index] = np.array([scores_list[index][ind2][1] for ind2 in range(len(scores_list[index]))]) | |
| index += 1 | |
| if len(sorted_cosine_scores_models) == 2: | |
| fig, (ax1, ax2) = plt.subplots(2) | |
| categories_sorted = [ | |
| categories[scores_list[0][index][0]] for index in range(len(scores_list[0])) | |
| ] | |
| sorted_scores = np.array( | |
| [scores_list[0][index][1] for index in range(len(scores_list[0]))] | |
| ) | |
| ax1.pie(sorted_scores, labels=categories_sorted, autopct="%1.1f%%") | |
| categories_sorted = [ | |
| categories[scores_list[1][index][0]] for index in range(len(scores_list[1])) | |
| ] | |
| sorted_scores = np.array( | |
| [scores_list[1][index][1] for index in range(len(scores_list[1]))] | |
| ) | |
| ax2.pie(sorted_scores, labels=categories_sorted, autopct="%1.1f%%") | |
| st.pyplot(fig) | |
| def plot_alatirchart(sorted_cosine_scores_models): | |
| models = list(sorted_cosine_scores_models.keys()) | |
| tabs = st.tabs(models) | |
| figs = {} | |
| for model in models: | |
| figs[model] = plot_piechart_helper(sorted_cosine_scores_models[model]) | |
| for index in range(len(tabs)): | |
| with tabs[index]: | |
| st.pyplot(figs[models[index]]) | |
| ### Text Search ### | |
| st.sidebar.title("GloVe Twitter") | |
| st.sidebar.markdown( | |
| """ | |
| GloVe is an unsupervised learning algorithm for obtaining vector representations for words. Pretrained on | |
| 2 billion tweets with vocabulary size of 1.2 million. Download from [Stanford NLP](http://nlp.stanford.edu/data/glove.twitter.27B.zip). | |
| Jeffrey Pennington, Richard Socher, and Christopher D. Manning. 2014. *GloVe: Global Vectors for Word Representation*. | |
| """ | |
| ) | |
| # initialize Session State variable | |
| if 'categories' not in st.session_state: | |
| st.session_state['categories'] = "Flowers Colors Cars Weather Food" | |
| if 'text_search' not in st.session_state: | |
| st.session_state['text_search'] = "Roses are red, trucks are blue, and Seattle is grey right now" | |
| model_type = st.sidebar.selectbox("Choose the model", ("25d", "50d"), index=1) | |
| st.title("In Class practice 1 demo") | |
| st.subheader( | |
| "Pass in space separated categories you want this search demo to be about." | |
| ) | |
| # st.selectbox(label="Pick the categories you want this search demo to be about...", | |
| # options=("Flowers Colors Cars Weather Food", "Chocolate Milk", "Anger Joy Sad Frustration Worry Happiness", "Positive Negative"), | |
| # key="categories" | |
| # ) | |
| # categories of user input | |
| categories = st.text_input( | |
| label="Categories", value=st.session_state.categories | |
| ) | |
| st.session_state.categories = categories.split(" ") | |
| print(st.session_state.get("categories")) | |
| print(type(st.session_state.get("categories"))) | |
| # print("Categories = ", categories) | |
| # st.session_state.categories = categories | |
| st.subheader("Pass in an input word or even a sentence") | |
| text_search = st.text_input( | |
| label="Input your sentence", | |
| st.session_state.text_search, | |
| ) | |
| st.session_state.text_search = text_search | |
| # Download glove embeddings if it doesn't exist | |
| embeddings_path = "embeddings_" + str(model_type) + "_temp.npy" | |
| word_index_dict_path = "word_index_dict_" + str(model_type) + "_temp.pkl" | |
| if not os.path.isfile(embeddings_path) or not os.path.isfile(word_index_dict_path): | |
| print("Model type = ", model_type) | |
| glove_path = "Data/glove_" + str(model_type) + ".pkl" | |
| print("glove_path = ", glove_path) | |
| # Download embeddings from google drive | |
| with st.spinner("Downloading glove embeddings..."): | |
| download_glove_embeddings_gdrive(model_type) | |
| # Load glove embeddings | |
| word_index_dict, embeddings = embeddings_model.load_glove_embeddings(model_type) | |
| category_embeddings = {category: embeddings_model.get_sentence_transformer_embedding(category) for category in | |
| st.session_state.categories} | |
| search_using_cos = Search(embeddings_model) | |
| # Find closest word to an input word | |
| if st.session_state.get("text_search"): | |
| # sentence transformer Embedding | |
| print("sentence transformer Embedding") | |
| embeddings_metadata = { | |
| "word_index_dict": word_index_dict, | |
| "embeddings": embeddings, | |
| "model_type": model_type, | |
| "text_search": st.session_state.text_search | |
| } | |
| with st.spinner("Obtaining Cosine similarity ..."): | |
| sorted_cosine_sim_transformer = search_using_cos.get_topK_similar_categories( | |
| st.session_state.text_search, category_embeddings | |
| ) | |
| # Results and Plot Pie Chart for Glove | |
| print("Categories are: ", st.session_state.categories) | |
| st.subheader( | |
| "Closest word I have between: " | |
| + st.session_state.categories | |
| + " as per different Embeddings" | |
| ) | |
| # print(sorted_cosine_sim_glove) | |
| print(sorted_cosine_sim_transformer) | |
| print(list(sorted_cosine_sim_transformer.keys())[0]) | |
| st.write( | |
| f"Closest category using sentence transformer embeddings : {list(sorted_cosine_sim_transformer.keys())[0]}") | |
| plot_alatirchart( | |
| { | |
| "sentence_transformer_384": sorted_cosine_sim_transformer, | |
| } | |
| ) | |
| st.write("") | |
| st.write( | |
| "Demo developed by Edward Xu" | |
| ) | |