import os import re import requests import json import gradio as gr from langchain.chat_models import ChatOpenAI from langchain import LLMChain, PromptTemplate from langchain.memory import ConversationBufferMemory OPENAI_API_KEY=os.getenv('sk-rxIqydkHEEWNaXylHANAT3BlbkFJFeyzNbjczTt3odYLJUf') PLAY_HT_API_KEY=os.getenv('340790dc07ff49d98a50389bbc74234b') PLAY_HT_USER_ID=os.getenv('eWU7cUYDBvdKYec2awquMyBOCOe2') PLAY_HT_VOICE_ID=os.getenv('s3://voice-cloning-zero-shot/60992038-3aac-4c15-8c71-7d4217c5e408/mahabub-ali/manifest.json') play_ht_api_get_audio_url = "https://play.ht/api/v2/tts" template = """You are a helpful assistant to answer user queries. {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 "+ sk-rxIqydkHEEWNaXylHANAT3BlbkFJFeyzNbjczTt3odYLJUf, "X-USER-ID" : eWU7cUYDBvdKYec2awquMyBOCOe2 } 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()`.