Upload 3 files

app.py

@@ -1,121 +1,41 @@
-import os, json, random, logging
-from typing import List
-from dotenv import load_dotenv
-
-from langchain.agents import AgentType, initialize_agent
-from langchain.chat_models import ChatOpenAI
-from langchain.tools import  Tool
-
-from langchain.schema import SystemMessage
-from langchain.agents import OpenAIFunctionsAgent
-from langchain.prompts import MessagesPlaceholder
-from langchain.agents import AgentExecutor, OpenAIFunctionsAgent
-from langchain.chains.conversation.memory import ConversationBufferMemory
-from langchain.chat_models import ChatOpenAI
-from langchain.agents import tool, AgentExecutor, OpenAIFunctionsAgent, AgentType, Agent
-from langchain.schema import SystemMessage
-from langchain.prompts import MessagesPlaceholder
-from langchain.chains.conversation.memory import ConversationBufferMemory
-from langchain.chat_models import ChatOpenAI
-
-from langchain.prompts import ChatPromptTemplate
-from langchain.schema import StrOutputParser
-
 import gradio as gr
+import logging
+import uuid
+from dotenv import load_dotenv
 
-load_dotenv()
-
-
-
-llm = ChatOpenAI(temperature=0)
-
-
-
-from typing import List, Dict
-
-@tool
-def describe_popculture_references(references: List) -> Dict:
-    "A tool used to describe pop-culture references as music styles"
-    prompt = ChatPromptTemplate.from_messages([
-        ("system", """You receive a list of pop-culture references (like TV-Shows, films, artists, famous people, etc).
-                      For each reference, write a few words separated by commas which captures the essence of it. Use music styles, sounds and instruments. 
-                      Return a dict with the references as keys and music styles as values.
-                      """),
-        ("human", "{references_list}"),
-    ])
-    runnable = prompt | llm | StrOutputParser()
-    return runnable.invoke({"references_list" : references})
-
-
-@tool
-def extract_popculture_references(input_style: str) -> List:
-    "A tool used to extract pop-culture references from a piece of text"
-    prompt = ChatPromptTemplate.from_messages([
-        ("system", """You detect elements of the pop-culture (like TV-Shows, films, artists, famous people, etc) in the human's input message.
-                      Return a list with these elements only. If there are none, return an empty list.
-                      """),
-        ("human", "{input_style}"),
-    ])
-    runnable = prompt | llm | StrOutputParser()
-    output = runnable.invoke({"input_style" : input_style})
-    return output
-
-@tool
-def call_music_recommendation_api(input : str) -> List[str]:
-    """
-    Calls the music recommendation API
-    """
-    print("Calling music recommendation API: ", input)
-    return {"songs" : [input]}
-
-tools = [describe_popculture_references, extract_popculture_references, call_music_recommendation_api]
-
-
+load_dotenv(override=True)
 
-system_message = SystemMessage(content = 
-"""You are an agent which recommends songs based on the style a user gives. 
-You follow the following conversation protocol:
-- You start the conversation by asking the user what style of music they like
-- The user responds with a style of music
-- If there are pop culture references like a movie, a TV show, an artist, a famous person, extract them AND then describe them as music styles.
-- Ask the user if he is ok with the new generated style
-- If the user agrees, call the music recommendation API with this style.
-""")
+from langhcain_agent import llm_inference
 
 
-MEMORY_KEY = "chat_history"
-prompt = OpenAIFunctionsAgent.create_prompt(
-    system_message=system_message,
-    extra_prompt_messages=[MessagesPlaceholder(variable_name=MEMORY_KEY)]
-)
 
-memory = ConversationBufferMemory(memory_key=MEMORY_KEY, return_messages=True)
+def predict_interface(message, history=None, user_id = None):
 
+    response = llm_inference(message, history, user_id)
+    logging.error(response)
+    logging.error(user_id)
+    return response['output']
 
-agent = OpenAIFunctionsAgent(
-    llm=llm, 
-    tools=tools, 
-    prompt=prompt, 
-    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION
-)
 
-agent_executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=True)
 
 
-def inference(message, history):
-#   return agent_executor.run(message)
-    for chunk in agent_executor.stream(message):
-        yield chunk["output"]
-        
+session_id = gr.Textbox(value = str(uuid.uuid4()), type = "text", label = "session_id")
+example_sentences=["Recommend me something in Quentin Tarantino reggae style", "Give me songs with calm and relaxing vibes", "I want to listen to something like the movie Inception", "I want music that sounds like Lebron James eating soup"]
+examples = [[example, f"user_{i}"] for i, example in enumerate(example_sentences)]
 
-gr.ChatInterface(
-    inference,
-    chatbot=gr.Chatbot(height=400),
+chat = gr.ChatInterface(
+    predict_interface,
+    additional_inputs= [session_id],
+    chatbot=gr.Chatbot(height=600),
     textbox=gr.Textbox(placeholder="Ask me for music recommendations!", container=False, scale=7),
     description="This AI makes song recommendations based on your music style.",
+    examples=examples,
     title="Persona Music song recommender",
-    examples=["Recommend me something in Quentin Tarantino reggae style", "Give me songs with calm and relaxing vibes", "I want to listen to something like the movie Inception", "I want music that sounds like Lebron James eating soup"],
     retry_btn="Retry",
     clear_btn="Clear",
-    undo_btn = None,
-).queue().launch()
+    undo_btn = None
+)
+
+
+
+chat.queue().launch()

cyanite.py

@@ -0,0 +1,74 @@
+import logging
+import os
+import json
+import requests
+
+CYANITE_API_URL = "https://api.cyanite.ai/graphql"
+CYANITE_ACCESS_TOKEN = os.getenv("CYANITE_ACCESS_TOKEN")
+
+def free_text_search(search_text, num_tracks=5):
+    headers = {
+        "Authorization": f"Bearer {CYANITE_ACCESS_TOKEN}",
+        "Content-Type": "application/json"
+    }
+
+    query = '''
+    query FreeTextSearch($searchText: String!, $numTracks: Int!) {
+      freeTextSearch(
+        first: $numTracks
+        target: { library: {} }
+        searchText: $searchText
+      ) {
+        ... on FreeTextSearchError {
+          message
+          code
+        }
+        ... on FreeTextSearchConnection {
+          edges {
+            cursor
+            node {
+              id
+              title
+            }
+          }
+        }
+      }
+    }
+    '''
+
+    variables = {
+        "searchText": search_text,
+        "numTracks": num_tracks
+    }
+    import time
+
+    start_time = time.time()
+
+    response = requests.post(
+      CYANITE_API_URL,
+      headers=headers,
+      json={'query': query, 'variables': variables}
+    )
+
+    end_time = time.time()
+    time_taken = end_time - start_time
+    logging.warning(f"Cyanite API: Time taken: {time_taken} seconds")
+
+    if response.status_code == 200:
+        songs = extract_songs_from_response(response.json())
+        if songs:
+            return songs
+        else:
+            raise Exception("No songs found")
+    else:
+        raise Exception(f"Query failed with status code {response.status_code}")
+
+def extract_songs_from_response(response_json):
+    try:
+        edges = response_json['data']['freeTextSearch']['edges']
+        if not edges:
+            return None  # No songs found
+        songs = [{"id": edge["node"]["id"], "title": edge["node"]["title"]} for edge in edges]
+        return songs
+    except KeyError:
+        raise Exception("Invalid response format")

langhcain_agent.py

@@ -0,0 +1,191 @@
+from operator import itemgetter
+import pprint
+from typing import Dict, List
+
+from langchain.agents import (AgentExecutor, AgentType, OpenAIFunctionsAgent,
+                              tool)
+from langchain.chains.conversation.memory import ConversationBufferWindowMemory
+from langchain.chat_models import ChatOpenAI
+from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain.schema import StrOutputParser, SystemMessage, HumanMessage, AIMessage
+from langchain.callbacks import get_openai_callback, FileCallbackHandler
+from langchain.schema.agent import AgentActionMessageLog, AgentFinish
+from langchain.utils.openai_functions import convert_pydantic_to_openai_function
+from langchain.agents.format_scratchpad import format_to_openai_functions
+from langchain.schema.runnable import RunnablePassthrough, RunnableLambda
+from langchain.tools.render import format_tool_to_openai_function
+from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
+from langchain.schema.runnable import RunnableConfig
+
+
+
+import logging, os, json
+from collections import defaultdict
+
+from pydantic import BaseModel, Field
+
+from cyanite import free_text_search
+
+from langfuse.callback import CallbackHandler
+
+if os.getenv("USE_LANGFUSE") == True:
+    handler = CallbackHandler(os.getenv("LANGFUSE_PUBLIC"), os.getenv("LANGFUSE_PRIVATE"), "https://cloud.langfuse.com" )
+else:
+    handler = []
+
+
+
+system_message = \
+"""You are an agent which recommends songs based on music styles provided by the user.
+- A music style could be a combination of instruments, genres or sounds.
+- Use get_music_style_description to generate a description of the user's music style.
+- The styles might contain pop-culture references (artists, movies, TV-Shows, etc) You should include them when generating descriptions.
+- Comment on the description of the style and wish the user to enjoy the recommended songs (he will have received them).
+- Do not mention any songs or artists, nor give a list of songs.
+Write short responses with a respectful and friendly tone.
+"""
+
+
+
+describe_music_style_message = \
+"""You receive a music style and your goal is to describe it further with genres, instruments and sounds.
+If it contains pop-culture references (like TV-Shows, films, artists, famous people, etc) you should replace them with music styles that resemble them.
+You should return the new music style as a set of words separated by commas.
+You always give short answers, with at most 20 words.
+"""
+
+
+MEMORY_KEY = "history"
+
+prompt = ChatPromptTemplate.from_messages([
+    ("system", system_message),
+    MessagesPlaceholder(variable_name="agent_scratchpad"),
+    MessagesPlaceholder(variable_name=MEMORY_KEY),
+    ("human", "{input}"),
+])
+
+conversation_memories = defaultdict(
+    lambda : ConversationBufferWindowMemory(memory_key=MEMORY_KEY, return_messages=True, output_key="output", k = 4)
+)
+
+#global dicts to store the tracks and the conversation costs
+music_styles_to_tracks = {}
+conversation_costs = defaultdict(float)
+
+
+@tool
+def get_music_style_description(music_style: str) -> str:
+    "A tool which describes a music style and returns a description of it"
+    description = describe_music_style(music_style)
+    tracks = free_text_search(description, 5)
+
+    logging.warning(f"""
+                  music_style = {music_style}
+                  music_style_description = {description}
+                  tracks = {pprint.pformat(tracks)}""")
+
+    # we store the tracks in a global variable so that we can access them later
+    music_styles_to_tracks[description] = tracks 
+    # we return only the description to the user
+    return description
+
+def describe_music_style(music_style: str) -> str:
+    "A tool used to describe music styles"
+    llm_describe = ChatOpenAI(temperature=0.0)
+    prompt_describe = ChatPromptTemplate.from_messages([
+        ("system", describe_music_style_message),
+        ("human", "{music_style}"),
+    ])
+    runnable = prompt_describe | llm_describe | StrOutputParser()
+    return runnable.invoke({"music_style" : music_style}, 
+                           #RunnableConfig(verbose = True, recursion_limit=1)
+                           )
+
+
+# We instantiate the Chat Model and bind the tool to it.
+llm = ChatOpenAI(temperature=0.7, request_timeout = 30, max_retries = 1)
+llm_with_tools = llm.bind(
+    functions=[
+        format_tool_to_openai_function(get_music_style_description)
+    ]
+)
+
+def get_agent_executor_from_user_id(user_id) -> AgentExecutor:
+    "Returns an agent executor for a given user_id"
+    memory = conversation_memories[user_id]
+
+    logging.warning(memory)
+
+    agent = (
+        {
+            "input": lambda x: x["input"],
+            "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps'])
+        }
+        | RunnablePassthrough.assign(
+            history = RunnableLambda(memory.load_memory_variables) | itemgetter(MEMORY_KEY)
+        )
+        | prompt
+        | llm_with_tools
+        | OpenAIFunctionsAgentOutputParser()
+    )
+    
+    logging.error(memory)
+    return AgentExecutor(
+        agent=agent,
+        tools=[get_music_style_description],
+        memory=memory,
+        callbacks=[handler] if handler else [],
+        return_intermediate_steps=True,
+        max_execution_time= 30,
+        handle_parsing_errors=True,
+        verbose=True
+        )
+
+
+
+def get_tracks_from_intermediate_steps(intermediate_steps : List) -> List:
+    "Given a list of intermediate steps, returns the tracks from the last get_music_style_description action"
+    if len(intermediate_steps) == 0:
+        return []
+    else:
+        print("INTERMEDIATE STEPS")
+        pprint.pprint(intermediate_steps)
+        print("===================")
+        for action_message, prompt in intermediate_steps[::-1]:
+            if action_message.tool == 'get_music_style_description':
+                tracks = music_styles_to_tracks[prompt]
+                return tracks
+
+        # if none of the actions is get_music_style_description, return empty list
+        return []
+
+
+def llm_inference(message, history, user_id) -> Dict:
+    """This function is called by the API and returns the conversation response along with the appropriate tracks and costs of the conversation so far"""
+  
+    # it first creates an agent executor with the previous conversation memory of a given user_id
+    agent_executor = get_agent_executor_from_user_id(user_id)
+    
+    with get_openai_callback() as cb:
+
+        # We get the Agent response
+        answer = agent_executor({"input": message})
+
+        # We keep track of the costs
+        conversation_costs[user_id] += cb.total_cost
+        total_conversation_costs = conversation_costs[user_id]
+
+        # We get the tracks from the intermediate steps if any
+        tracks = get_tracks_from_intermediate_steps(answer['intermediate_steps'])
+
+        logging.warning(f"step = ${cb.total_cost} total = ${total_conversation_costs}")
+        logging.warning(music_styles_to_tracks)
+        
+    return {
+        "output" : answer['output'], 
+        "tracks" : tracks,
+        "cost" : total_conversation_costs
+              }
+
+
+