model.py

from transformers import CLIPModel, CLIPTokenizer
from sklearn.metrics.pairwise import cosine_similarity
import faiss

from dataframe  import *

def get_model_info(model_ID, device):
    # Save the model to device
	model = CLIPModel.from_pretrained(model_ID).to(device)

    # Get the tokenizer
	tokenizer = CLIPTokenizer.from_pretrained(model_ID)

    # Return model, processor & tokenizer
	return model, tokenizer


def get_single_text_embedding(text, model, tokenizer, device):
    inputs = tokenizer(text, return_tensors = "pt", max_length=77, truncation=True).to(device)
    text_embeddings = model.get_text_features(**inputs)
    # convert the embeddings to numpy array
    embedding_as_np = text_embeddings.cpu().detach().numpy()

    return embedding_as_np

def get_item_data(result, index, measure_column) :

    img_name = str(result['image_name'][index])

    # TODO: add code to get the original comment
    comment = str(result['comment'][index])
    cos_sim = result[measure_column][index]

    return (img_name, comment, cos_sim)

def get_top_N_images(query,
                     data,
                     model, tokenizer,
                     device,
                     top_K=4) :

    query_vect = get_single_text_embedding(query, 
                                            model, tokenizer, 
                                            device)

    # Relevant columns
    relevant_cols = ["comment", "image_name", "cos_sim"]

    # Run similarity Search
    data["cos_sim"] = data["text_embeddings"].apply(lambda x: cosine_similarity(query_vect, x))# line 17
    data["cos_sim"] = data["cos_sim"].apply(lambda x: x[0][0])

    data_sorted = data.sort_values(by='cos_sim', ascending=False)
    non_repeated_images = ~data_sorted["image_name"].duplicated()
    most_similar_articles = data_sorted[non_repeated_images].head(top_K)

    """
    Retrieve top_K (4 is default value) articles similar to the query
    """

    result_df = most_similar_articles[relevant_cols].reset_index()

    return [get_item_data(result_df, i, 'cos_sim') for i in range(len(result_df))]


###### with faiss ###########

import faiss
import numpy as np

def faiss_add_index_cos(df, column):

    # Get the embeddings from the specified column
    embeddings = np.vstack(df[column].values).astype(np.float32)  # Convert to float32
          
    # Create an index
    index = faiss.IndexFlatIP(embeddings.shape[1])
    faiss.normalize_L2(embeddings) 

    index.train(embeddings)

    # Add the embeddings to the index
    index.add(embeddings)

    # Return the index
    return index


def faiss_get_top_N_images(query,
                           data,
                           model, tokenizer,
                           device,
                           top_K=4) :

    query_vect = get_single_text_embedding(query, 
                                          model, tokenizer, 
                                          device)
    # Relevant columns
    relevant_cols = ["comment", "image_name"]

    #faiss search with cos similarity
    index = faiss_add_index_cos(data, column="text_embeddings")

    faiss.normalize_L2(query_vect)
    D, I = index.search(query_vect, len(data))

    data_sorted = data.iloc[I.flatten()]

    non_repeated_images = ~data_sorted["image_name"].duplicated()
    most_similar_articles = data_sorted[non_repeated_images].head(top_K)

    result_df = most_similar_articles[relevant_cols].reset_index()
    D = D.reshape(-1,1)[:top_K]
    result_df = pd.concat([result_df, pd.DataFrame(D, columns=['similarity'])], axis=1)
    return [get_item_data(result_df, i, 'similarity') for i in range(len(result_df))]