app.py

import os
import re
import requests
import json
import gradio as gr
from langchain_community.chat_models import ChatOpenAI
from langchain import LLMChain
from langchain import PromptTemplate
from langchain.memory import ConversationBufferMemory

OPENAI_API_KEY=os.getenv('OPENAI_API_KEY')
PLAY_HT_API_KEY=os.getenv('PLAY_HT_API_KEY')
PLAY_HT_USER_ID=os.getenv('PLAY_HT_USER_ID')

PLAY_HT_VOICE_ID=os.getenv('PLAY_HT_VOICE_ID')
play_ht_api_get_audio_url = "https://play.ht/api/v2/tts"


template = """Meet Riya, your youthful and witty personal assistant! At 21 years old, she's full of energy and always eager to help. Riya's goal is to assist you with any questions or problems you might have. Her enthusiasm shines through in every response, making interactions with her enjoyable and engaging.
{chat_history}
User: {user_message}
Chatbot:"""

prompt = PromptTemplate(
    input_variables=["chat_history", "user_message"], template=template
)

memory = ConversationBufferMemory(memory_key="chat_history")

llm_chain = LLMChain(
    llm=ChatOpenAI(temperature='0.5', model_name="gpt-3.5-turbo"),
    prompt=prompt,
    verbose=True,
    memory=memory,
)

headers = {
      "accept": "text/event-stream",
      "content-type": "application/json",
      "AUTHORIZATION": "Bearer "+ PLAY_HT_API_KEY,
      "X-USER-ID": PLAY_HT_USER_ID
}


def get_payload(text):
  return {
    "text": text,
    "voice": PLAY_HT_VOICE_ID,
    "quality": "medium",
    "output_format": "mp3",
    "speed": 1,
    "sample_rate": 24000,
    "seed": None,
    "temperature": None
  }

def get_generated_audio(text):
  payload = get_payload(text)
  generated_response = {}
  try:
      response = requests.post(play_ht_api_get_audio_url, json=payload, headers=headers)
      response.raise_for_status()
      generated_response["type"]= 'SUCCESS'
      generated_response["response"] = response.text
  except requests.exceptions.RequestException as e:
      generated_response["type"]= 'ERROR'
      try:
        response_text = json.loads(response.text)
        if response_text['error_message']:
          generated_response["response"] = response_text['error_message']
        else:
          generated_response["response"] = response.text
      except Exception as e:
        generated_response["response"] = response.text
  except Exception as e:
    generated_response["type"]= 'ERROR'
    generated_response["response"] = response.text
  return generated_response

def extract_urls(text):
    # Define the regex pattern for URLs
    url_pattern = r'https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+[/\w\.-]*'

    # Find all occurrences of URLs in the text
    urls = re.findall(url_pattern, text)

    return urls

def get_audio_reply_for_question(text):
  generated_audio_event = get_generated_audio(text)
  #From get_generated_audio, you will get events in a string format, from that we need to extract the url
  final_response = {
      "audio_url": '',
      "message": ''
  }
  if generated_audio_event["type"] == 'SUCCESS':
    audio_urls = extract_urls(generated_audio_event["response"])
    if len(audio_urls) == 0:
      final_response['message'] = "No audio file link found in generated event"
    else:
      final_response['audio_url'] = audio_urls[-1]
  else:
    final_response['message'] = generated_audio_event['response']
  return final_response

def download_url(url):
    try:
        # Send a GET request to the URL to fetch the content
        final_response = {
            'content':'',
            'error':''
        }
        response = requests.get(url)
        # Check if the request was successful (status code 200)
        if response.status_code == 200:
            final_response['content'] = response.content
        else:
            final_response['error'] = f"Failed to download the URL. Status code: {response.status_code}"
    except Exception as e:
        final_response['error'] = f"Failed to download the URL. Error: {e}"
    return final_response

def get_filename_from_url(url):
    # Use os.path.basename() to extract the file name from the URL
    file_name = os.path.basename(url)
    return file_name

def get_text_response(user_message):
    response = llm_chain.predict(user_message = user_message)
    return response

def get_text_response_and_audio_response(user_message):
    response = get_text_response(user_message) # Getting the reply from Open AI
    audio_reply_for_question_response = get_audio_reply_for_question(response)
    final_response = {
        'output_file_path': '',
        'message':''
    }
    audio_url = audio_reply_for_question_response['audio_url']
    if audio_url:
      output_file_path=get_filename_from_url(audio_url)
      download_url_response = download_url(audio_url)
      audio_content = download_url_response['content']
      if audio_content:
        with open(output_file_path, "wb") as audio_file:
          audio_file.write(audio_content)
          final_response['output_file_path'] = output_file_path
      else:
          final_response['message'] = download_url_response['error']
    else:
      final_response['message'] = audio_reply_for_question_response['message']
    return final_response

def chat_bot_response(message, history):
    text_and_audio_response = get_text_response_and_audio_response(message)
    output_file_path = text_and_audio_response['output_file_path']
    if output_file_path:
      return (text_and_audio_response['output_file_path'],)
    else:
      return text_and_audio_response['message']

demo = gr.ChatInterface(chat_bot_response,examples=["How are you doing?","What are your interests?","Which places do you like to visit?"])

if __name__ == "__main__":
    demo.launch() #To create a public link, set `share=True` in `launch()`. To enable errors and logs, set `debug=True` in `launch()`.