app.py

import streamlit as st
import tiktoken
from loguru import logger

from langchain.chains import ConversationalRetrievalChain
from langchain.chat_models import ChatOpenAI

from langchain.document_loaders.pdf import (PyPDFLoader, PyMuPDFLoader)

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings

from langchain.memory import ConversationBufferMemory
from langchain.vectorstores import FAISS

# from streamlit_chat import message
from langchain.callbacks import get_openai_callback
from langchain.memory import StreamlitChatMessageHistory
from gtts import gTTS
from IPython.display import Audio, display

from io import BytesIO
import speech_recognition as sr

from langchain.callbacks import get_openai_callback
from langchain.memory import StreamlitChatMessageHistory

def main():
    st.set_page_config(
        page_title="차량용 Q&A 챗봇",
        page_icon=":car:"
    )

    st.title("차량용 Q&A 챗봇 :car:")

    if "conversation" not in st.session_state:
        st.session_state.conversation = None

    if "chat_history" not in st.session_state:
        st.session_state.chat_history = None

    if "processComplete" not in st.session_state:
        st.session_state.processComplete = None

    with st.sidebar:
        uploaded_files = st.file_uploader("차량 메뉴얼 PDF 파일을 넣어주세요.", type=['pdf'], accept_multiple_files=True)
        openai_api_key = st.text_input("OpenAI API Key", key="chatbot_api_key", type="password")
        process = st.button("실행")

    if process:
        if not openai_api_key:
            st.info("Open AI키를 입력해주세요.")
            st.stop()
        files_text = get_text(uploaded_files)
        text_chunks = get_text_chunks(files_text)
        vector_store = get_vectorstore(text_chunks)

        st.session_state.conversation = get_conversation_chain(vector_store, openai_api_key)

        st.session_state.processComplete = True

    if 'messages' not in st.session_state:
        st.session_state['messages'] = [{"role": "assistant",
                                         "content": "안녕하세요! 주어진 문서에 대해 궁금하신 것이 있으면 언제든 물어봐주세요!"}]

    for message in st.session_state.messages:
        with st.chat_message(message["role"]):
            st.markdown(message["content"])

    history = StreamlitChatMessageHistory(key="chat_messages")

    # Chat logic
    if query := st.chat_input("질문을 입력해주세요."):
        st.session_state.messages.append({"role": "user", "content": query})

        with st.chat_message("user"):
            st.markdown(query)

        with st.chat_message("assistant"):
            # STT 코드 추가
            user_voice_input = stt()
            if user_voice_input:
                # Add user's voice input to chat history
                st.session_state.messages.append({"role": "user", "content": user_voice_input})

                # Continue with the rest of the conversation logic...
                chain = st.session_state.conversation

                with st.spinner("Thinking..."):
                    result = chain({"question": query})
                    with get_openai_callback() as cb:
                        st.session_state.chat_history = result['chat_history']
                    response = result['answer']
                    source_documents = result['source_documents']

                    st.markdown(response)
                    with st.expander("참고 문서 확인"):
                        st.markdown(source_documents[0].metadata['source'], help=source_documents[0].page_content)
                        st.markdown(source_documents[1].metadata['source'], help=source_documents[1].page_content)
                        st.markdown(source_documents[2].metadata['source'], help=source_documents[2].page_content)

                # TTS 코드 추가
                tts("이것은 음성으로 변환된 답변입니다.")

                # Add assistant message to chat history
                st.session_state.messages.append({"role": "assistant", "content": response})

# ...

# STT 함수 추가
def stt():
    recognizer = sr.Recognizer()

    with sr.Microphone() as source:
        st.write("말해보세요...")
        recognizer.adjust_for_ambient_noise(source)
        audio = recognizer.listen(source, timeout=5)

    try:
        text = recognizer.recognize_google(audio, language="ko-KR")
        st.write("인식된 텍스트: {}".format(text))
        return text
    except sr.UnknownValueError:
        st.write("음성을 인식할 수 없습니다.")
        return None
    except sr.RequestError as e:
        st.write("Google Speech Recognition 서비스에 접근할 수 없습니다; {0}".format(e))
        return None

# TTS 함수 추가
def tts(text):
    st.write("음성으로 변환 중...")
    tts = gTTS(text=text, lang='ko')
    audio_stream = BytesIO()
    tts.save(audio_stream)
    st.audio(audio_stream, format='audio/wav')

def tiktoken_len(text):
    tokenizer = tiktoken.get_encoding("cl100k_base")
    tokens = tokenizer.encode(text)
    return len(tokens)


def get_text(docs):
    doc_list = []

    for doc in docs:
        file_name = doc.name  # doc 객체의 이름을 파일 이름으로 사용
        with open(file_name, "wb") as file:  # 파일을 doc.name으로 저장
            file.write(doc.getvalue())
            logger.info(f"Uploaded {file_name}")
        if '.pdf' in doc.name:
            loader = PyMuPDFLoader(file_name)
            documents = loader.load_and_split()

        doc_list.extend(documents)
    return doc_list


def get_text_chunks(text):
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=100,
        length_function=tiktoken_len
    )
    chunks = text_splitter.split_documents(text)
    return chunks


def get_vectorstore(text_chunks):
    embeddings = HuggingFaceEmbeddings(
        model_name="jhgan/ko-sroberta-multitask",
        model_kwargs={'device': 'cpu'},
        encode_kwargs={'normalize_embeddings': True}
    )
    vectordb = FAISS.from_documents(text_chunks, embeddings)
    return vectordb


def get_conversation_chain(vetorestore, openai_api_key):
    llm = ChatOpenAI(openai_api_key=openai_api_key, model_name='gpt-3.5-turbo', temperature=0)
    conversation_chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        chain_type="stuff",
        retriever=vetorestore.as_retriever(search_type='mmr', vervose=True),
        memory=ConversationBufferMemory(memory_key='chat_history', return_messages=True, output_key='answer'),
        get_chat_history=lambda h: h,
        return_source_documents=True,
        verbose=True
    )

    return conversation_chain


if __name__ == '__main__':
    main()