# Gradio Interface | |
import gradio as gr | |
import numpy as np | |
import pandas as pd | |
from sklearn.metrics.pairwise import cosine_similarity | |
from sentence_transformers import SentenceTransformer | |
import requests | |
from PIL import Image | |
from transformers import BlipProcessor, BlipForConditionalGeneration | |
sentence_model = SentenceTransformer("all-MiniLM-L6-v2") | |
processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base") | |
image_model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base") | |
def generate_input(input_type, image=None, text=None, response_amount=3): | |
# Initialize the input variable | |
combined_input = "" | |
# Handle image input if chosen | |
if input_type == "Image" and image: | |
inputs = processor(images=image, return_tensors="pt") | |
out = image_model.generate(**inputs) | |
image_caption = processor.decode(out[0], skip_special_tokens=True) | |
combined_input += image_caption # Add the image caption to input | |
# Handle text input if chosen | |
elif input_type == "Text" and text: | |
combined_input += text # Add the text to input | |
# Handle both text and image input if chosen | |
elif input_type == "Both" and image and text: | |
inputs = processor(images=image, return_tensors="pt") | |
out = image_model.generate(**inputs) | |
image_caption = processor.decode(out[0], skip_special_tokens=True) | |
combined_input += image_caption + " and " + text # Combine image caption and text | |
# If no input, fallback | |
if not combined_input: | |
combined_input = "No input provided." | |
if response_amount is None: | |
response_amount=3 | |
return vector_search(combined_input,response_amount) | |
# Load embeddings and metadata | |
embeddings = np.load("netflix_embeddings.npy") #created using sentence_transformers on kaggle | |
metadata = pd.read_csv("netflix_metadata.csv") #created using sentence_transformers on kaggle | |
# Vector search function | |
def vector_search(query,top_n=3): | |
query_embedding = sentence_model.encode(query) | |
similarities = cosine_similarity([query_embedding], embeddings)[0] | |
if top_n is None: | |
top_n=3 | |
top_indices = similarities.argsort()[-top_n:][::-1] | |
results = metadata.iloc[top_indices] | |
result_text="" | |
for index,row in results.iterrows(): | |
if index!=top_n-1: | |
result_text+=f"Title: {row['title']} Description: {row['description']} Genre: {row['listed_in']}\n\n" | |
else: | |
result_text+=f"Title: {row['title']} Description: {row['description']} Genre: {row['listed_in']}" | |
return result_text | |
def set_response_amount(response_amount): | |
if response_amount is None: | |
return 3 | |
return response_amount | |
# Based on the selected input type, make the appropriate input visible | |
def update_inputs(input_type): | |
if input_type == "Image": | |
return gr.update(visible=True), gr.update(visible=False), gr.update(visible=True) | |
elif input_type == "Text": | |
return gr.update(visible=False), gr.update(visible=True), gr.update(visible=True) | |
elif input_type == "Both": | |
return gr.update(visible=True), gr.update(visible=True), gr.update(visible=True) | |
with gr.Blocks() as demo: | |
gr.Markdown("# Netflix Recommendation System") | |
gr.Markdown("Enter a query to receive Netflix show recommendations based on title, description, and genre.") | |
input_type = gr.Radio(["Image", "Text", "Both"], label="Select Input Type", type="value") | |
response_type=gr.Dropdown(choices=[3,5,10,25], type="value", label="Select Response Amount", visible=False) | |
image_input = gr.Image(label="Upload Image", type="pil", visible=False) # Hidden initially | |
text_input = gr.Textbox(label="Enter Text Query", placeholder="Enter a description or query here", visible=False) # Hidden initially | |
input_type.change(fn=update_inputs, inputs=input_type, outputs=[image_input, text_input, response_type]) | |
# State variable to store the selected response amount | |
selected_response_amount = gr.State() | |
# Capture response amount immediately when dropdown changes | |
response_type.change(fn=set_response_amount, inputs=response_type, outputs=selected_response_amount) | |
submit_button = gr.Button("Submit") | |
output = gr.Textbox(label="Recommendations") | |
if selected_response_amount is None: | |
selected_response_amount=3 | |
submit_button.click(fn=generate_input, inputs=[input_type,image_input, text_input,selected_response_amount], outputs=output) | |
demo.launch(ssr_mode = False) | |
# with gr.Blocks() as demo: | |
# gr.Markdown("# Netflix Recommendation System") | |
# gr.Markdown("Enter a query to receive Netflix show recommendations based on title, description, and genre.") | |
# query = gr.Textbox(label="Enter your query") | |
# output = gr.Textbox(label="Recommendations") | |
# submit_button = gr.Button("Submit") | |
# submit_button.click(fn=lambda q: vector_search(q, model), inputs=query, outputs=output) | |
# import gradio as gr | |
# # def greet(name): | |
# # return "Hello " + name + "!!" | |
# from sentence_transformers import SentenceTransformer | |
# import numpy as np | |
# from sklearn.metrics.pairwise import cosine_similarity | |
# from datasets import load_dataset | |
# # Load pre-trained SentenceTransformer model | |
# embedding_model = SentenceTransformer("thenlper/gte-large") | |
# # # Example dataset with genres (replace with your actual data) | |
# # dataset = load_dataset("hugginglearners/netflix-shows") | |
# # dataset = dataset.filter(lambda x: x['description'] is not None and x['listed_in'] is not None and x['title'] is not None) | |
# # data = dataset['train'] # Accessing the 'train' split of the dataset | |
# # # Convert the dataset to a list of dictionaries for easier indexing | |
# # data_list = list[data] | |
# # print(data_list) | |
# # # Combine description and genre for embedding | |
# # def combine_description_title_and_genre(description, listed_in, title): | |
# # return f"{description} Genre: {listed_in} Title: {title}" | |
# # # Generate embedding for the query | |
# # def get_embedding(text): | |
# # return embedding_model.encode(text) | |
# # # Vector search function | |
# # def vector_search(query): | |
# # query_embedding = get_embedding(query) | |
# # # Generate embeddings for the combined description and genre | |
# # embeddings = np.array([get_embedding(combine_description_title_and_genre(item["description"], item["listed_in"],item["title"])) for item in data_list[0]]) | |
# # # Calculate cosine similarity between the query and all embeddings | |
# # similarities = cosine_similarity([query_embedding], embeddings) | |
# # Load dataset (using the correct dataset identifier for your case) | |
# dataset = load_dataset("hugginglearners/netflix-shows") | |
# # Combine description and genre for embedding | |
# def combine_description_title_and_genre(description, listed_in, title): | |
# return f"{description} Genre: {listed_in} Title: {title}" | |
# # Generate embedding for the query | |
# def get_embedding(text): | |
# return embedding_model.encode(text) | |
# # Vector search function | |
# def vector_search(query): | |
# query_embedding = get_embedding(query) | |
# # Function to generate embeddings for each item in the dataset | |
# def generate_embeddings(example): | |
# return { | |
# 'embedding': get_embedding(combine_description_title_and_genre(example["description"], example["listed_in"], example["title"])) | |
# } | |
# # Generate embeddings for the dataset using map | |
# embeddings_dataset = dataset["train"].map(generate_embeddings) | |
# # Extract embeddings | |
# embeddings = np.array([embedding['embedding'] for embedding in embeddings_dataset]) | |
# # Calculate cosine similarity between the query and all embeddings | |
# similarities = cosine_similarity([query_embedding], embeddings) | |
# # # Adjust similarity scores based on ratings | |
# # ratings = np.array([item["rating"] for item in data_list]) | |
# # adjusted_similarities = similarities * ratings.reshape(-1, 1) | |
# # Get top N most similar items (e.g., top 3) | |
# top_n = 3 | |
# top_indices = similarities[0].argsort()[-top_n:][::-1] # Get indices of the top N results | |
# top_items = [dataset["train"][i] for i in top_indices] | |
# # Format the output for display | |
# search_result = "" | |
# for item in top_items: | |
# search_result += f"Title: {item['title']}, Description: {item['description']}, Genre: {item['listed_in']}\n" | |
# return search_result | |
# # Gradio Interface | |
# def movie_search(query): | |
# return vector_search(query) | |
# with gr.Blocks() as demo: | |
# gr.Markdown("# Netflix Recommendation System") | |
# gr.Markdown("Enter a query to receive Netflix show recommendations based on title, description, and genre.") | |
# query = gr.Textbox(label="Enter your query") | |
# output = gr.Textbox(label="Recommendations") | |
# submit_button = gr.Button("Submit") | |
# submit_button.click(fn=movie_search, inputs=query, outputs=output) | |
# demo.launch() | |
# # iface = gr.Interface(fn=movie_search, | |
# # inputs=gr.inputs.Textbox(label="Enter your query"), | |
# # outputs="text", | |
# # live=True, | |
# # title="Netflix Recommendation System", | |
# # description="Enter a query to get Netflix recommendations based on description and genre.") | |
# # iface.launch() | |
# # demo = gr.Interface(fn=greet, inputs="text", outputs="text") | |
# # demo.launch() | |