LangChain_Test / app.py
se0kcess
add new func
9229d16
import streamlit as st
from dotenv import load_dotenv
from langchain.text_splitter import CharacterTextSplitter, RecursiveCharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings # General embeddings from HuggingFace models.
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
from htmlTemplates import css, bot_template, user_template
from langchain.llms import LlamaCpp # For loading transformer models.
from langchain.document_loaders import PyPDFLoader, TextLoader, JSONLoader, CSVLoader
import tempfile # 임시 파일을 생성하기 위한 라이브러리입니다.
import os
from huggingface_hub import hf_hub_download # Hugging Face Hub에서 모델을 다운로드하기 위한 함수입니다.
# PDF 문서로부터 텍스트를 추출하는 함수입니다.
def get_pdf_text(pdf_docs):
temp_dir = tempfile.TemporaryDirectory() # 임시 디렉토리를 생성합니다.
temp_filepath = os.path.join(temp_dir.name, pdf_docs.name) # 임시 파일 경로를 생성합니다.
with open(temp_filepath, "wb") as f: # 임시 파일을 바이너리 쓰기 모드로 엽니다.
f.write(pdf_docs.getvalue()) # PDF 문서의 내용을 임시 파일에 씁니다.
pdf_loader = PyPDFLoader(temp_filepath) # PyPDFLoader를 사용해 PDF를 로드합니다.
pdf_doc = pdf_loader.load() # 텍스트를 추출합니다.
return pdf_doc # 추출한 텍스트를 반환합니다.
# 과제
# 아래 텍스트 추출 함수를 작성
def get_text_file(docs):
# 텍스트 파일에서 텍스트를 읽어옵니다.
text_content = docs.read()
return text_content
def get_csv_file(docs):
# CSV 파일에서 텍스트를 읽어옵니다.
csv_reader = csv.reader(docs)
csv_content = "\n".join(",".join(row) for row in csv_reader)
return csv_content
def get_json_file(docs):
# JSON 파일에서 텍스트를 읽어옵니다.
json_content = json.load(docs)
return json.dumps(json_content, indent=2)
# 문서들을 처리하여 텍스트 청크로 나누는 함수입니다.
def get_text_chunks(documents):
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 청크의 크기를 지정합니다.
chunk_overlap=200, # 청크 사이의 중복을 지정합니다.
length_function=len # 텍스트의 길이를 측정하는 함수를 지정합니다.
)
documents = text_splitter.split_documents(documents) # 문서들을 청크로 나눕니다.
return documents # 나눈 청크를 반환합니다.
# 텍스트 청크들로부터 벡터 스토어를 생성하는 함수입니다.
def get_vectorstore(text_chunks):
# 원하는 임베딩 모델을 로드합니다.
embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-MiniLM-L12-v2',
model_kwargs={'device': 'cpu'}) # 임베딩 모델을 설정합니다.
vectorstore = FAISS.from_documents(text_chunks, embeddings) # FAISS 벡터 스토어를 생성합니다.
return vectorstore # 생성된 벡터 스토어를 반환합니다.
def get_conversation_chain(vectorstore):
model_name_or_path = 'TheBloke/Llama-2-7B-chat-GGUF'
model_basename = 'llama-2-7b-chat.Q2_K.gguf'
model_path = hf_hub_download(repo_id=model_name_or_path, filename=model_basename)
llm = LlamaCpp(model_path=model_path,
n_ctx=4086,
input={"temperature": 0.75, "max_length": 2000, "top_p": 1},
verbose=True, )
# 대화 기록을 저장하기 위한 메모리를 생성합니다.
memory = ConversationBufferMemory(
memory_key='chat_history', return_messages=True)
# 대화 검색 체인을 생성합니다.
conversation_chain = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=vectorstore.as_retriever(),
memory=memory
)
return conversation_chain # 생성된 대화 체인을 반환합니다.
# 사용자 입력을 처리하는 함수입니다.
def handle_userinput(user_question):
print('user_question => ', user_question)
# 대화 체인을 사용하여 사용자 질문에 대한 응답을 생성합니다.
response = st.session_state.conversation({'question': user_question})
# 대화 기록을 저장합니다.
st.session_state.chat_history = response['chat_history']
for i, message in enumerate(st.session_state.chat_history):
if i % 2 == 0:
st.write(user_template.replace(
"{{MSG}}", message.content), unsafe_allow_html=True)
else:
st.write(bot_template.replace(
"{{MSG}}", message.content), unsafe_allow_html=True)
def main():
load_dotenv()
st.set_page_config(page_title="Chat with multiple Files",
page_icon=":books:")
st.write(css, unsafe_allow_html=True)
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
st.header("Chat with multiple Files:")
user_question = st.text_input("Ask a question about your documents:")
if user_question:
handle_userinput(user_question)
with st.sidebar:
st.subheader("Your documents")
docs = st.file_uploader(
"Upload your PDFs here and click on 'Process'", accept_multiple_files=True)
if st.button("Process"):
with st.spinner("Processing"):
# get pdf text
doc_list = []
for file in docs:
print('file - type : ', file.type)
if file.type == 'text/plain':
# file is .txt
doc_list.extend(get_text_file(file))
elif file.type in ['application/octet-stream', 'application/pdf']:
# file is .pdf
doc_list.extend(get_pdf_text(file))
elif file.type == 'text/csv':
# file is .csv
doc_list.extend(get_csv_file(file))
elif file.type == 'application/json':
# file is .json
doc_list.extend(get_json_file(file))
# get the text chunks
text_chunks = get_text_chunks(doc_list)
# create vector store
vectorstore = get_vectorstore(text_chunks)
# create conversation chain
st.session_state.conversation = get_conversation_chain(
vectorstore)
if __name__ == '__main__':
main()