Spaces:
Runtime error
Runtime error
| from dotenv import load_dotenv | |
| load_dotenv("env_vars.env") | |
| import os | |
| import numpy as np | |
| import streamlit as st | |
| from pathlib import Path | |
| from typing import List, Tuple | |
| from langchain.chat_models import ChatOpenAI | |
| from langchain.embeddings.openai import OpenAIEmbeddings | |
| from langchain.schema import Document | |
| from langchain.chains import LLMChain | |
| from langchain.prompts import PromptTemplate | |
| from utils import weighted_random_sample, load_db | |
| from redis_storage import RedisStorage, UserInput | |
| Matches = List[Tuple[Document, float]] # Custom type to save results of similarity search. Contains [(Song info, score)] | |
| USE_STORAGE = os.environ.get("USE_STORAGE", "True").lower() in ("true", "t", "1") | |
| def init(): | |
| embeddings = OpenAIEmbeddings(model=os.environ['MODEL']) | |
| dataset_path = f"hub://{os.environ['ACTIVELOOP_ORG_ID']}/{os.environ['DATASET']}" | |
| db = load_db( | |
| dataset_path, | |
| embedding=embeddings, | |
| token=os.environ["ACTIVELOOP_TOKEN"], | |
| read_only=True, | |
| ) | |
| storage = RedisStorage( | |
| host=os.environ["UPSTASH_URL"], password=os.environ["UPSTASH_PASSWORD"] | |
| ) | |
| prompt = PromptTemplate( | |
| input_variables=["user_input"], | |
| template=Path(f"{os.environ['ROOT']}prompts/augment_user_input.prompt").read_text(), | |
| ) | |
| llm = ChatOpenAI(temperature=0.3) | |
| chain = LLMChain(llm=llm, prompt=prompt) | |
| return db, chain, storage | |
| # Don't show the setting sidebar | |
| if "sidebar_state" not in st.session_state: | |
| st.session_state.sidebar_state = "collapsed" | |
| st.set_page_config( | |
| page_title="TaylorSwiftDJ", | |
| page_icon="🎶", | |
| initial_sidebar_state=st.session_state.sidebar_state | |
| ) | |
| # Get Vector store, LLM Chain Redis storage info. This is done once and cached. | |
| db, chain, storage = init() | |
| st.title("TaylorSwiftDJ 🎵🌟") | |
| st.markdown( | |
| """ | |
| *<small>Made with [DeepLake](https://www.deeplake.ai/) 🚀 and [LangChain](https://python.langchain.com/en/latest/index.html) 🦜⛓️</small>* | |
| 💫 Hey there Swifties! Welcome to "TaylorSwiftDJ"! 🎤🎶 This streamlit powered app recommends her iconic songs tailored to your emotions. Dance, reminisce, and embrace your feelings with every beat – because life's a melody, and Taylor's your DJ. ✨ 🌈 💖""", | |
| unsafe_allow_html=True, | |
| ) | |
| how_it_works = st.expander(label="How it works") | |
| text_input = st.text_input( | |
| label="How are you feeling today?", | |
| placeholder="I am feeling 22!", | |
| ) | |
| run_btn = st.button("Let's Dance! 🎶💃") | |
| with how_it_works: | |
| st.markdown( | |
| """ | |
| The application follows a sequence of steps to deliver Taylor Swift songs matching the user's emotions: | |
| - **User Input**: The application starts by collecting user's emotional state through a text input. | |
| - **Emotion Encoding**: The user-provided emotions are then fed to a Language Model (LLM). The LLM interprets and encodes these emotions. | |
| - **Similarity Search**: These encoded emotions are utilized to perform a similarity search within our [vector database](https://www.deeplake.ai/). This database houses ~130 Taylor Swift songs, each represented as emotional embeddings. | |
| - **Song Selection**: From the pool of top matching songs, the application randomly selects one. The selection is weighted, giving preference to songs with higher similarity scores. | |
| - **Song Retrieval**: The selected song's embedded player is displayed on the webpage for the user. Additionally, the LLM interpreted emotional state associated with the chosen song is displayed. | |
| """ | |
| ) | |
| placeholder_emotions = st.empty() | |
| placeholder = st.empty() | |
| print(Path("./prompts/augment_user_input.prompt")) | |
| with st.sidebar: | |
| st.text("App settings") | |
| filter_threshold = st.slider( | |
| "Threshold used to filter out low scoring songs", | |
| min_value=0.0, | |
| max_value=1.0, | |
| value=0.8, | |
| ) | |
| max_number_of_songs = st.slider( | |
| "Max number of songs to get from database", | |
| min_value=5, | |
| max_value=50, | |
| value=20, | |
| step=1, | |
| ) | |
| number_of_displayed_songs = st.slider( | |
| "Number of displayed songs", min_value=1, max_value=4, value=2, step=1 | |
| ) | |
| def filter_scores(matches: Matches, th: float = 0.8) -> Matches: | |
| return [(doc, score) for (doc, score) in matches if score > th] | |
| def normalize_scores_by_sum(matches: Matches) -> Matches: | |
| scores = [score for _, score in matches] | |
| tot = sum(scores) | |
| return [(doc, (score / tot)) for doc, score in matches] | |
| def get_song(user_input: str, k: int = 20) -> Tuple[List, List]: | |
| emotions = chain.run(user_input=user_input) | |
| matches = db.similarity_search_with_score(emotions, distance_metric="cos", k=k) | |
| #[print(doc.metadata['name'], score) for doc, score in matches] | |
| scores = filter_scores(matches, filter_threshold) | |
| if len(scores) > 0: | |
| docs, scores = zip( | |
| *normalize_scores_by_sum(scores) | |
| ) | |
| chosen_docs = weighted_random_sample( | |
| np.array(docs), np.array(scores), n=number_of_displayed_songs | |
| ).tolist() | |
| else: | |
| chosen_docs = [] | |
| return chosen_docs, emotions | |
| def set_song(user_input:str): | |
| if user_input == "": | |
| user_input = "I am feeling 22!" | |
| # take first 120 chars | |
| user_input = user_input[:120] | |
| docs, emotions = get_song(user_input, k=max_number_of_songs) | |
| # print(docs) | |
| songs = [] | |
| spotify_iframes = [] | |
| with placeholder_emotions: | |
| st.markdown("Your emotions: `" + emotions + "`") | |
| for doc in docs: | |
| name = doc.metadata["name"] | |
| songs.append(name) | |
| spotify_iframes.append(doc.metadata["iframe"]) | |
| if len(songs) > 0: | |
| with placeholder: | |
| st.markdown("Your songs: ") | |
| for idx, iframe in enumerate(spotify_iframes): | |
| st.markdown(iframe, unsafe_allow_html=True ) | |
| else: | |
| with placeholder: | |
| st.markdown("No song match found with the provided threshold. Please lower it. ") | |
| # Save user input to Redis via Upstash. | |
| if USE_STORAGE: | |
| ret_val = storage.store( | |
| UserInput(text=user_input, emotions=emotions, songs=songs) | |
| ) | |
| if not ret_val: | |
| print("[ERROR] was not able to store user_input") | |
| if run_btn: | |
| set_song(text_input) |