app.py

import plotly.graph_objs as go
from sklearn.cluster import KMeans
from sklearn.decomposition import PCA
import plotly.express as px
import numpy as np
import os
import pprint
import codecs
import chardet
import gradio as gr
from langchain.llms import HuggingFacePipeline, OpenAIChat
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from langchain import OpenAI, ConversationChain, LLMChain, PromptTemplate
from langchain.chains.conversation.memory import ConversationBufferMemory
from EdgeGPT import Chatbot
import whisper
from datetime import datetime
import json
import requests
from langchain.chains.question_answering import load_qa_chain
import langchain


class ChatbotClass:
    def __init__(self):
        FOLDER_PATH = './data/eqe-manual'
        QUERY = 'How do I charge my vehicle?'
        K = 10
        self.whisper_model = whisper.load_model(name='tiny')

        self.embeddings = HuggingFaceEmbeddings()

        self.index = FAISS.load_local(
            folder_path=FOLDER_PATH, embeddings=self.embeddings
        )

        self.llm = OpenAIChat(temperature=0)

        self.memory = ConversationBufferMemory(
            memory_key="chat_history", input_key="human_input", return_messages=True
        )

        self.keyword_chain = self.init_keyword_chain()
        self.context_chain = self.init_context_chain()
        self.document_retrieval_chain = self.init_document_retrieval()
        self.conversation_chain = self.init_conversation()

    def format_history(self, memory):
        history = memory.chat_memory.messages
        if len(history) == 0:
            return []

        formatted_history = []

        for h in history:
            if isinstance(h, langchain.schema.HumanMessage):
                user_response = h.content
            elif isinstance(h, langchain.schema.AIMessage):
                ai_response = h.content
                formatted_history.append((user_response, ai_response))

        return formatted_history

    def init_document_retrieval(self):
        retrieve_documents_template = """This function retrieves exerts from a Vehicle Owner's Manual. The function is useful for adding vehicle-specific context to answer questions. Based on a request, determine if vehicle specific information is needed. Respond with "Yes" or "No". If the answer is both, respond with "Yes":\nrequest: How do I change the tire?\nresponse: Yes\nrequest: Hello\nresponse: No\nrequest: I was in an accident. What should I do?\nresponse: Yes\nrequest: {request}\nresponse:"""

        prompt_template = PromptTemplate(
            input_variables=["request"],
            template=retrieve_documents_template
        )

        document_retrieval_chain = LLMChain(
            llm=self.llm, prompt=prompt_template, verbose=True
        )

        return document_retrieval_chain

    def init_keyword_chain(self):
        keyword_template = """You are a vehicle owner searching for content in your vehicle's owner manual. Your job is to come up with keywords to use when searching inside your manual, based on a question you have.
        Question: {question}
        Keywords:"""

        prompt_template = PromptTemplate(
            input_variables=["question"], template=keyword_template
        )

        keyword_chain = LLMChain(
            llm=self.llm, prompt=prompt_template, verbose=True)

        return keyword_chain

    def init_context_chain(self):
        context_template = """You are a friendly and helpful chatbot having a conversation with a human.
        Given the following extracted parts of a long document and a question, create a final answer.

        {context}

        {chat_history}
        Human: {human_input}
        Chatbot:"""

        context_prompt = PromptTemplate(
            input_variables=["chat_history", "human_input", "context"],
            template=context_template
        )

        self.memory = ConversationBufferMemory(
            memory_key="chat_history", input_key="human_input", return_messages=True
        )

        context_chain = load_qa_chain(
            self.llm, chain_type="stuff", memory=self.memory, prompt=context_prompt
        )

        return context_chain

    def init_conversation(self):
        template = """You are a chatbot having a conversation with a human.

        {chat_history}
        Human: {human_input}
        Chatbot:"""

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

        conversation_chain = LLMChain(
            llm=self.llm,
            prompt=prompt,
            verbose=True,
            memory=self.memory,
        )

        return conversation_chain


    def transcribe_audio(self, audio_file, model):
        result = self.whisper_model.transcribe(audio_file)
        return result['text']


    def ask_question(self, query, k=4):
        tool_usage = self.document_retrieval_chain.run(query)
        print('\033[1;32m'  f'search manual?: {tool_usage}'  "\033[0m")

        chat_history = self.format_history(self.memory)

        if tool_usage == 'Yes':
            keywords = self.keyword_chain.run(question=query)
            print('\033[1;32m'  f'keywords:{keywords}'  "\033[0m")

            context = self.index.similarity_search(query=keywords, k=k)

            result = self.context_chain.run(
                input_documents=context, human_input=query
            )
        else:
            result = self.conversation_chain.run(query)

        return [(query, result)], chat_history


    def invoke_exh_api(self, bot_response, bot_name='Zippy', voice_name='Fiona', idle_url='https://ugc-idle.s3-us-west-2.amazonaws.com/4a6a607a466bdf6605bbd97ef146751b.mp4', animation_pipeline='high_quality', bearer_token='eyJhbGciOiJIUzUxMiJ9.eyJ1c2VybmFtZSI6IndlYiJ9.LSzIQx6h61l5FXs52s0qcY8WqauET6z9nnxgSzvoNBx8RYEKm8OpOohcK8wjuwteV4ZGug4NOjoGQoUZIKH84A'):
        if len(bot_response) > 200:
            print('Input is over 200 characters. Shorten the message')

        url = 'https://api.exh.ai/animations/v1/generate_lipsync'
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36 Edg/110.0.1587.46',
            'authority': 'api.exh.ai',
            'accept': '*/*',
            'accept-encoding': 'gzip, deflate, br',
            'accept-language': 'en-US,en;q=0.9',
            'authorization': f'Bearer {bearer_token}',
            'content-type': 'application/json',
            'origin': 'https://admin.exh.ai',
            'referer': 'https://admin.exh.ai/',
            'sec-ch-ua': '"Chromium";v="110", "Not A(Brand";v="24", "Microsoft Edge";v="110"',
            'sec-ch-ua-mobile': '?0',
            'sec-ch-ua-platform': '"Windows"',
            'sec-fetch-dest': 'empty',
            'sec-fetch-mode': 'cors',
            'sec-fetch-site': 'same-site',
        }

        data = {
            'bot_name': bot_name,
            'bot_response': bot_response,
            'voice_name': voice_name,
            'idle_url': idle_url,
            'animation_pipeline': animation_pipeline,
        }

        r = requests.post(url, headers=headers, data=json.dumps(data))

        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S%f')
        outfile = f'talking_head_{timestamp}.mp4'

        with open(outfile, 'wb') as f:
            f.write(r.content)

        return outfile


    def predict(self, input_data, state=[], k=4, input_type='audio'):
        if input_type == 'audio':
            txt = self.transcribe_audio(input_data[0], self.whisper_model)
        else:
            txt = input_data[1]
        result, chat_history = self.ask_question(txt, k=k)
        state.append(chat_history)
        return result, state


    def predict_wrapper(self, input_text=None, input_audio=None):
        if input_audio is not None:
            result, state = self.predict(
                input_data=(input_audio,), input_type='audio')
        else:
            result, state = self.predict(
                input_data=('', input_text), input_type='text')
        
        response = result[0][1][:195]
        avatar = self.invoke_exh_api(response)

        return result,avatar


man_chatbot = ChatbotClass()

iface = gr.Interface(
    fn=man_chatbot.predict_wrapper,
    inputs=[gr.inputs.Textbox(label="Text Input"),
            gr.inputs.Audio(source="microphone", type='filepath')],
    outputs=[gr.outputs.Textbox(label="Result"),
             gr.outputs.Video().style(width=360, height=360, container=True)]
)
iface.launch()

'''
iface.launch()
with gr.Blocks() as demo:
    chatbot = gr.Chatbot()
    state = gr.State([])
    
    with gr.Row():
        txt = gr.Textbox(show_label=False, placeholder="Enter text and press enter").style(
            container=False)
        k_slider = gr.Slider(minimum=1, maximum=10, default=4,label='k')
        txt.submit(man_chatbot.predict, [txt, state,k_slider],[chatbot,state])

demo.launch()
'''