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 :-) | |
""" | |
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 | |
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 | |
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() | |
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("sentence transformer") | |
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" | |
embeddings_model = Embeddings() | |
model_type = st.sidebar.selectbox("Choose the model", ("50d"), index=1) | |
st.title("in in-class coding practice1 Demo") | |
st.subheader( | |
"Pass in space separated categories you want this search demo to be about." | |
) | |
# categories of user input | |
user_categories = st.text_input( | |
label="Categories", value=st.session_state.categories | |
) | |
st.session_state.categories = user_categories.split(" ") | |
print(st.session_state.get("categories")) | |
print(type(st.session_state.get("categories"))) | |
st.subheader("Pass in an input word or even a sentence") | |
user_text_search = st.text_input( | |
label="Input your sentence", | |
value=st.session_state.text_search, | |
) | |
st.session_state.text_search = user_text_search | |
# 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.text_search: | |
# sentence transformer embeddings | |
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 for Glove..."): | |
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: " | |
+ " ".join(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" | |
) | |