| import os |
| import streamlit as st |
| from dotenv import load_dotenv |
| from PyPDF2 import PdfReader |
| from langchain.text_splitter import CharacterTextSplitter |
| from langchain_openai import OpenAIEmbeddings |
| from langchain_community.embeddings import HuggingFaceInstructEmbeddings |
| from langchain_community.vectorstores import FAISS |
| from langchain_openai import ChatOpenAI |
| from langchain_community.llms import HuggingFaceHub |
| from langchain.memory import ConversationBufferMemory |
| from langchain.chains import ConversationalRetrievalChain |
| from langchain_community.document_loaders import DirectoryLoader |
| from htmlTemplates import css, bot_template, user_template |
| from langchain.globals import set_verbose |
| set_verbose(False) |
|
|
|
|
| def read_files_from_directory(directory): |
| files = [] |
| for filename in os.listdir(directory): |
| if filename.endswith(".pdf"): |
| files.append(os.path.join(directory, filename)) |
| return files |
|
|
| def get_pdf_text(pdf_docs): |
| text = "" |
| for pdf in pdf_docs: |
| pdf_reader = PdfReader(pdf) |
| for page in pdf_reader.pages: |
| text += page.extract_text() |
| return text |
|
|
| def get_text_chunks(raw_text): |
| text_splitter = CharacterTextSplitter( |
| separator="\n", |
| chunk_size=1000, |
| chunk_overlap=200, |
| length_function=len |
| ) |
| chunks = text_splitter.split_text(raw_text) |
| return chunks |
|
|
| def get_vector_store(text_chunks): |
| |
| embeddings = OpenAIEmbeddings(openai_api_key=os.getenv('OPENAI_API_KEY')) |
| vectorstore = FAISS.from_texts(texts=text_chunks, embedding=embeddings) |
| return vectorstore |
|
|
| def get_conversation_chain(vectorstore): |
| if not os.getenv('OPENAI_API_KEY') and not os.getenv('LAW_GPT_MODEL_URL'): |
| raise ValueError("Please provide either OPENAI_API_KEY or LAW_GPT_MODEL_URL in the .env file") |
|
|
| |
| if os.getenv('LAW_GPT_MODEL_URL'): |
| llm = HuggingFaceHub(repo_id=os.getenv('LAW_GPT_MODEL_URL')) |
| else: |
| llm = ChatOpenAI(openai_api_key=os.getenv('OPENAI_API_KEY')) |
| |
| 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_user_input(user_question): |
| if st.session_state.conversation is not None: |
|
|
| 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) |
| else: |
| st.write("No data is loaded for RAG. Please upload a PDFs files to the data/ directory.") |
|
|
| def main(): |
| load_dotenv() |
|
|
| st.set_page_config(page_title="EULawGPT - LLM model that can understand and reason about EU public domain data", page_icon=":books:") |
| st.write(css, unsafe_allow_html=True) |
| |
| |
| files = read_files_from_directory('./data') |
| raw_knowledge_text = get_pdf_text(files) |
| raw_knowledge_chunks = get_text_chunks(raw_knowledge_text) |
| vectorstore_knowledge = get_vector_store(raw_knowledge_chunks) |
|
|
| st.session_state.conversation = get_conversation_chain(vectorstore_knowledge) |
|
|
| 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.title("EU Law GPT") |
| st.write("EU Law GPT is a LLM model that can understand and reason about EU public domain data") |
|
|
| st.subheader('Popular questions:') |
| if st.button("What is happening in Equador?"): |
| handle_user_input("What is happening in Equador?") |
|
|
| if st.button("What EU will do with Ecuador crisis?"): |
| handle_user_input("What EU will do with Ecuador crisis?") |
|
|
| st.subheader('Ask anything:') |
| user_question = st.text_input("Ask a question about EU Law and Parlament work") |
|
|
| if user_question: |
| handle_user_input(user_question) |
|
|
| if __name__ == '__main__': |
| main() |
