app.py

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