import os import time from dotenv import load_dotenv from groq import Groq from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_community.embeddings import HuggingFaceEmbeddings import streamlit as st def load_docs_from_url(input_url): web_loader = WebBaseLoader(input_url) documents = web_loader.load() splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20) split_documents = splitter.split_documents(documents) st.write('Documents Loaded from URL') return split_documents def load_docs_from_pdf(file): start = time.time() with open("temp.pdf", "wb") as temp_file: temp_file.write(file.getbuffer()) pdf_loader = PyPDFLoader("temp.pdf") pdf_documents = pdf_loader.load() splitter = RecursiveCharacterTextSplitter(chunk_size=3000, chunk_overlap=150) processed_documents = splitter.split_documents(pdf_documents) st.write('Documents Loaded') st.write(f"Time taken to load documents: {time.time() - start:.2f} seconds") os.remove("temp.pdf") return processed_documents def build_vector_db(documents): start = time.time() embedder = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2", model_kwargs={"trust_remote_code": True}) db = FAISS.from_documents(documents, embedder) st.write('DB is ready') st.write(f"Time taken to create DB: {time.time() - start:.2f} seconds") return db def call_groq_api(chat_messages): load_dotenv() api_key = os.getenv("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") if not api_key: st.error("GROQ API key not found. Please set it in your .env file.") return "API key missing." groq_client = Groq(api_key=api_key) full_response = '' stream_output = groq_client.chat.completions.create( model="llama3-70b-8192", messages=chat_messages, max_tokens=2048, temperature=1.0, stream=True, ) for chunk in stream_output: part = chunk.choices[0].delta.content if part: full_response += part return full_response def summarize_conversation(history): history_text = " ".join([f"{entry['role']}: {entry['content']}" for entry in history]) summary_prompt = f"Summarize the following chat history:\n\n{history_text}" summary_messages = [{'role': 'system', 'content': 'You are very good at summarizing the chat between User and Assistant'}] summary_messages.append({'role': 'user', 'content': summary_prompt}) return call_groq_api(summary_messages) def main(): st.set_page_config(page_title='Chatbot') st.title("Document Chatbot using RAGs - Harsh Tripathi and Shubham Soni") with st.expander("Instructions to upload Text PDF/URL"): st.write("1. Open the sidebar by clicking the menu icon in the top-left corner.") st.write("2. If you're uploading a PDF, click 'Upload PDF', choose your file, and wait until you see the 'Documents Loaded' message.") st.write("3. If you're using a web URL, type it in, click 'Enter Web URL', then hit 'Process URL' and wait for the 'Documents Loaded from URL' confirmation.") st.write("4. Once the documents are loaded, click 'Create Vector Store' to begin processing.") st.write("5. Type your question into the text box and click Submit.") st.write("6. Click 'Generate Chat Summary' to view a summary of the entire chat session.") # Session reset button if st.sidebar.button("Reset Session"): for key in st.session_state.keys(): del st.session_state[key] st.experimental_rerun() st.sidebar.subheader("Choose document source:") input_choice = st.sidebar.radio("Select one:", ("Upload PDF", "Enter Web URL")) # Session state initialization if "loaded_docs" not in st.session_state: st.session_state.loaded_docs = None if "document_db" not in st.session_state: st.session_state.document_db = None if "conversation" not in st.session_state: st.session_state.conversation = [] if "query_input" not in st.session_state: st.session_state.query_input = "" if "latest_output" not in st.session_state: st.session_state.latest_output = "" if "chat_summary" not in st.session_state: st.session_state.chat_summary = "" if input_choice == "Upload PDF": file_input = st.sidebar.file_uploader("Upload a PDF file", type=["pdf"]) if file_input is not None: if st.session_state.loaded_docs is None: with st.spinner("Loading documents..."): documents = load_docs_from_pdf(file_input) st.session_state.loaded_docs = documents elif input_choice == "Enter Web URL": input_url = st.sidebar.text_input("Enter URL", key="url_input") if st.session_state.url_input != input_url: st.session_state.url_input = input_url st.session_state.loaded_docs = None if st.sidebar.button('Process URL'): if input_url and st.session_state.loaded_docs is None: with st.spinner("Fetching and processing documents from URL..."): documents = load_docs_from_url(input_url) st.session_state.loaded_docs = documents if st.session_state.loaded_docs is not None: if st.sidebar.button('Create Vector Store'): with st.spinner("Creating vector store..."): db = build_vector_db(st.session_state.loaded_docs) st.session_state.document_db = db # Chat handling logic def submit_with_db(): user_input = st.session_state.query_input if user_input: doc_retriever = st.session_state.document_db.as_retriever(search_type="similarity", search_kwargs={"k": 3}) results = doc_retriever.invoke(user_input) constructed_prompt = f''' You are a helpful assistant. Answer the user's question based on the context below. If the answer is not found in the context, respond with general knowledge. Context: {results} Chat History: {st.session_state.conversation} Latest Question: {user_input} ''' message_flow = [{'role': 'system', 'content': 'You are a very helpful assistant'}] message_flow.append({'role': 'user', 'content': constructed_prompt}) try: reply = call_groq_api(message_flow) except Exception as err: st.error(f"Error occurred during chat_groq execution: {str(err)}") reply = "An error occurred while fetching response. Please try again." st.session_state.latest_output = reply st.session_state.conversation.append({'role': 'user', 'content': user_input}) st.session_state.conversation.append({'role': 'assistant', 'content': reply}) st.session_state.query_input = "" def handle_general_chat(): user_input = st.session_state.query_input if user_input: prompt_text = f''' You are a helpful assistant. Answer the question based on the conversation context. Chat History: {st.session_state.conversation} Latest Question: {user_input} ''' message_flow = [{'role': 'system', 'content': 'You are a very helpful assistant'}] message_flow.append({'role': 'user', 'content': prompt_text}) try: reply = call_groq_api(message_flow) except Exception as err: st.error(f"Error occurred during chat_groq execution: {str(err)}") reply = "An error occurred while fetching response. Please try again." st.session_state.latest_output = reply st.session_state.conversation.append({'role': 'user', 'content': user_input}) st.session_state.conversation.append({'role': 'assistant', 'content': reply}) st.session_state.query_input = "" st.text_area("Enter your question:", key="query_input") if st.session_state.document_db is not None: st.button('Submit', on_click=submit_with_db) else: st.button('Submit', on_click=handle_general_chat) if st.session_state.latest_output: st.write(st.session_state.latest_output) if st.button('Generate Chat Summary'): st.session_state.chat_summary = summarize_conversation(st.session_state.conversation) if st.session_state.chat_summary: with st.expander("Chat Summary"): st.write(st.session_state.chat_summary) with st.expander("Recent Chat History"): latest_entries = st.session_state.conversation[-8:][::-1] paired_entries = [] for i in range(0, len(latest_entries), 2): if i+1 < len(latest_entries): paired_entries.extend([latest_entries[i+1], latest_entries[i]]) else: paired_entries.append(latest_entries[i]) for entry in paired_entries: st.write(f"{entry['role'].capitalize()}: {entry['content']}") if __name__ == "__main__": main()