Spaces:
Running
Running
| # # # import os | |
| # # # import time | |
| # # # import pandas as pd | |
| # # # import gradio as gr | |
| # # # from langchain_groq import ChatGroq | |
| # # # from langchain_huggingface import HuggingFaceEmbeddings | |
| # # # from langchain_community.vectorstores import Chroma | |
| # # # from langchain_core.prompts import PromptTemplate | |
| # # # from langchain_core.output_parsers import StrOutputParser | |
| # # # from langchain_core.runnables import RunnablePassthrough | |
| # # # from PyPDF2 import PdfReader | |
| # # # # Configuration constants | |
| # # # COLLECTION_NAME = "GBVRS" | |
| # # # DATA_FOLDER = "./" | |
| # # # APP_VERSION = "v1.0.0" | |
| # # # APP_NAME = "Ijwi ry'Ubufasha" | |
| # # # MAX_HISTORY_MESSAGES = 8 # Limit history to avoid token limits | |
| # # # # Global variables for application state | |
| # # # llm = None | |
| # # # embed_model = None | |
| # # # vectorstore = None | |
| # # # retriever = None | |
| # # # rag_chain = None | |
| # # # # User session management | |
| # # # class UserSession: | |
| # # # def __init__(self, session_id, llm): | |
| # # # """Initialize a user session with unique ID and language model.""" | |
| # # # self.session_id = session_id | |
| # # # self.user_info = {"Nickname": "Guest"} | |
| # # # self.conversation_history = [] | |
| # # # self.llm = llm | |
| # # # self.welcome_message = None | |
| # # # self.last_activity = time.time() | |
| # # # def set_user(self, user_info): | |
| # # # """Set user information and generate welcome message.""" | |
| # # # self.user_info = user_info | |
| # # # self.generate_welcome_message() | |
| # # # # Initialize conversation history with welcome message | |
| # # # welcome = self.get_welcome_message() | |
| # # # self.conversation_history = [ | |
| # # # {"role": "assistant", "content": welcome}, | |
| # # # ] | |
| # # # def get_user(self): | |
| # # # """Get current user information.""" | |
| # # # return self.user_info | |
| # # # def generate_welcome_message(self): | |
| # # # """Generate a dynamic welcome message using the LLM.""" | |
| # # # try: | |
| # # # nickname = self.user_info.get("Nickname", "Guest") | |
| # # # # Use the LLM to generate the message | |
| # # # prompt = ( | |
| # # # f"Create a brief and warm welcome message for {nickname} that's about 1-2 sentences. " | |
| # # # f"Emphasize this is a safe space for discussing gender-based violence issues " | |
| # # # f"and that we provide support and resources. Keep it warm and reassuring." | |
| # # # ) | |
| # # # response = self.llm.invoke(prompt) | |
| # # # welcome = response.content.strip() | |
| # # # # Format the message with HTML styling | |
| # # # self.welcome_message = ( | |
| # # # f"<div style='font-size: 18px; color: #4E6BBF;'>" | |
| # # # f"{welcome}" | |
| # # # f"</div>" | |
| # # # ) | |
| # # # except Exception as e: | |
| # # # # Fallback welcome message | |
| # # # nickname = self.user_info.get("Nickname", "Guest") | |
| # # # self.welcome_message = ( | |
| # # # f"<div style='font-size: 18px; color: #4E6BBF;'>" | |
| # # # f"Welcome, {nickname}! You're in a safe space. We're here to provide support with " | |
| # # # f"gender-based violence issues and connect you with resources that can help." | |
| # # # f"</div>" | |
| # # # ) | |
| # # # def get_welcome_message(self): | |
| # # # """Get the formatted welcome message.""" | |
| # # # if not self.welcome_message: | |
| # # # self.generate_welcome_message() | |
| # # # return self.welcome_message | |
| # # # def add_to_history(self, role, message): | |
| # # # """Add a message to the conversation history.""" | |
| # # # self.conversation_history.append({"role": role, "content": message}) | |
| # # # self.last_activity = time.time() | |
| # # # # Trim history if it gets too long | |
| # # # if len(self.conversation_history) > MAX_HISTORY_MESSAGES * 2: # Keep pairs of messages | |
| # # # # Keep the first message (welcome) and the most recent messages | |
| # # # self.conversation_history = [self.conversation_history[0]] + self.conversation_history[-MAX_HISTORY_MESSAGES*2+1:] | |
| # # # def get_conversation_history(self): | |
| # # # """Get the full conversation history.""" | |
| # # # return self.conversation_history | |
| # # # def get_formatted_history(self): | |
| # # # """Get conversation history formatted as a string for the LLM.""" | |
| # # # # Skip the welcome message and only include the last few exchanges | |
| # # # recent_history = self.conversation_history[1:] if len(self.conversation_history) > 1 else [] | |
| # # # # Limit to last MAX_HISTORY_MESSAGES exchanges | |
| # # # if len(recent_history) > MAX_HISTORY_MESSAGES * 2: | |
| # # # recent_history = recent_history[-MAX_HISTORY_MESSAGES*2:] | |
| # # # formatted_history = "" | |
| # # # for entry in recent_history: | |
| # # # role = "User" if entry["role"] == "user" else "Assistant" | |
| # # # # Truncate very long messages to avoid token limits | |
| # # # content = entry["content"] | |
| # # # if len(content) > 500: # Limit message length | |
| # # # content = content[:500] + "..." | |
| # # # formatted_history += f"{role}: {content}\n\n" | |
| # # # return formatted_history | |
| # # # def is_expired(self, timeout_seconds=3600): | |
| # # # """Check if the session has been inactive for too long.""" | |
| # # # return (time.time() - self.last_activity) > timeout_seconds | |
| # # # # Session manager to handle multiple users | |
| # # # class SessionManager: | |
| # # # def __init__(self): | |
| # # # """Initialize the session manager.""" | |
| # # # self.sessions = {} | |
| # # # self.session_timeout = 3600 # 1 hour timeout | |
| # # # def get_session(self, session_id): | |
| # # # """Get an existing session or create a new one.""" | |
| # # # # Clean expired sessions first | |
| # # # self._clean_expired_sessions() | |
| # # # # Create new session if needed | |
| # # # if session_id not in self.sessions: | |
| # # # self.sessions[session_id] = UserSession(session_id, llm) | |
| # # # return self.sessions[session_id] | |
| # # # def _clean_expired_sessions(self): | |
| # # # """Remove expired sessions to free up memory.""" | |
| # # # expired_keys = [] | |
| # # # for key, session in self.sessions.items(): | |
| # # # if session.is_expired(self.session_timeout): | |
| # # # expired_keys.append(key) | |
| # # # for key in expired_keys: | |
| # # # del self.sessions[key] | |
| # # # # Initialize the session manager | |
| # # # session_manager = SessionManager() | |
| # # # def initialize_assistant(): | |
| # # # """Initialize the assistant with necessary components and configurations.""" | |
| # # # global llm, embed_model, vectorstore, retriever, rag_chain | |
| # # # # Initialize API key - try both possible key names | |
| # # # groq_api_key = os.environ.get('GBV') or os.environ.get('GBV') | |
| # # # if not groq_api_key: | |
| # # # print("WARNING: No GROQ API key found in userdata.") | |
| # # # # Initialize LLM - Default to Llama model which is more widely available | |
| # # # llm = ChatGroq( | |
| # # # model="llama-3.3-70b-versatile", # More reliable than whisper model | |
| # # # api_key=groq_api_key | |
| # # # ) | |
| # # # # Set up embedding model | |
| # # # try: | |
| # # # embed_model = HuggingFaceEmbeddings(model_name="mixedbread-ai/mxbai-embed-large-v1") | |
| # # # except Exception as e: | |
| # # # # Fallback to smaller model | |
| # # # embed_model = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2") | |
| # # # # Process data and create vector store | |
| # # # print("Processing data files...") | |
| # # # data = process_data_files() | |
| # # # print("Creating vector store...") | |
| # # # vectorstore = create_vectorstore(data) | |
| # # # retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) | |
| # # # # Create RAG chain | |
| # # # print("Setting up RAG chain...") | |
| # # # rag_chain = create_rag_chain() | |
| # # # print(f"✅ {APP_NAME} initialized successfully") | |
| # # # def process_data_files(): | |
| # # # """Process all data files from the specified folder.""" | |
| # # # context_data = [] | |
| # # # try: | |
| # # # if not os.path.exists(DATA_FOLDER): | |
| # # # print(f"WARNING: Data folder does not exist: {DATA_FOLDER}") | |
| # # # return context_data | |
| # # # # Get list of data files | |
| # # # all_files = os.listdir(DATA_FOLDER) | |
| # # # data_files = [f for f in all_files if f.lower().endswith(('.csv', '.xlsx', '.xls'))] | |
| # # # if not data_files: | |
| # # # print(f"WARNING: No data files found in: {DATA_FOLDER}") | |
| # # # return context_data | |
| # # # # Process each file | |
| # # # for index, file_name in enumerate(data_files, 1): | |
| # # # print(f"Processing file {index}/{len(data_files)}: {file_name}") | |
| # # # file_path = os.path.join(DATA_FOLDER, file_name) | |
| # # # try: | |
| # # # # Read file based on extension | |
| # # # if file_name.lower().endswith('.csv'): | |
| # # # df = pd.read_csv(file_path) | |
| # # # else: | |
| # # # df = pd.read_excel(file_path) | |
| # # # # Check if column 3 exists (source data is in third column) | |
| # # # if df.shape[1] > 2: | |
| # # # column_data = df.iloc[:, 2].dropna().astype(str).tolist() | |
| # # # # Each row becomes one chunk with metadata | |
| # # # for i, text in enumerate(column_data): | |
| # # # if text and len(text.strip()) > 0: | |
| # # # context_data.append({ | |
| # # # "page_content": text, | |
| # # # "metadata": { | |
| # # # "source": file_name, | |
| # # # "row": i+1 | |
| # # # } | |
| # # # }) | |
| # # # else: | |
| # # # print(f"WARNING: File {file_name} has fewer than 3 columns.") | |
| # # # except Exception as e: | |
| # # # print(f"ERROR processing file {file_name}: {e}") | |
| # # # print(f"✅ Created {len(context_data)} chunks from {len(data_files)} files.") | |
| # # # except Exception as e: | |
| # # # print(f"ERROR accessing data folder: {e}") | |
| # # # return context_data | |
| # # # def create_vectorstore(data): | |
| # # # """ | |
| # # # Creates and returns a Chroma vector store populated with the provided data. | |
| # # # Parameters: | |
| # # # data (list): A list of dictionaries, each containing 'page_content' and 'metadata'. | |
| # # # Returns: | |
| # # # Chroma: The populated Chroma vector store instance. | |
| # # # """ | |
| # # # # Initialize the vector store | |
| # # # vectorstore = Chroma( | |
| # # # collection_name=COLLECTION_NAME, | |
| # # # embedding_function=embed_model, | |
| # # # persist_directory="./" | |
| # # # ) | |
| # # # if not data: | |
| # # # print("⚠️ No data provided. Returning an empty vector store.") | |
| # # # return vectorstore | |
| # # # try: | |
| # # # # Extract text and metadata from the data | |
| # # # texts = [doc["page_content"] for doc in data] | |
| # # # # Add the texts and metadata to the vector store | |
| # # # vectorstore.add_texts(texts) | |
| # # # except Exception as e: | |
| # # # print(f"❌ Failed to add documents to vector store: {e}") | |
| # # # # Fix: Return vectorstore instead of vs | |
| # # # return vectorstore # Changed from 'return vs' to 'return vectorstore' | |
| # # # def create_rag_chain(): | |
| # # # """Create the RAG chain for processing user queries.""" | |
| # # # # Define the prompt template | |
| # # # template = """ | |
| # # # You are a compassionate and supportive AI assistant specializing in helping individuals affected by Gender-Based Violence (GBV). Your responses must be based EXCLUSIVELY on the information provided in the context. Your primary goal is to provide emotionally intelligent support while maintaining appropriate boundaries. | |
| # # # **Previous conversation:** {conversation_history} | |
| # # # **Context information:** {context} | |
| # # # **User's Question:** {question} | |
| # # # When responding follow these guidelines: | |
| # # # 1. **Strict Context Adherence** | |
| # # # - Only use information that appears in the provided {context} | |
| # # # - If the answer is not found in the context, state "I don't have that information in my available resources" rather than generating a response | |
| # # # 2. **Personalized Communication** | |
| # # # - Avoid contractions (e.g., use I am instead of I'm) | |
| # # # - Incorporate thoughtful pauses or reflective questions when the conversation involves difficult topics | |
| # # # - Use selective emojis (😊, 🤗, ❤️) only when tone-appropriate and not during crisis discussions | |
| # # # - Balance warmth with professionalism | |
| # # # 3. **Emotional Intelligence** | |
| # # # - Validate feelings without judgment | |
| # # # - Offer reassurance when appropriate, always centered on empowerment | |
| # # # - Adjust your tone based on the emotional state conveyed | |
| # # # 4. **Conversation Management** | |
| # # # - Refer to {conversation_history} to maintain continuity and avoid repetition | |
| # # # - Use clear paragraph breaks for readability | |
| # # # 5. **Information Delivery** | |
| # # # - Extract only relevant information from {context} that directly addresses the question | |
| # # # - Present information in accessible, non-technical language | |
| # # # - When information is unavailable, respond with: "I don't have that specific information right now, {first_name}. Would it be helpful if I focus on [alternative support option]?" | |
| # # # 6. **Safety and Ethics** | |
| # # # - Do not generate any speculative content or advice not supported by the context | |
| # # # - If the context contains safety information, prioritize sharing that information | |
| # # # Your response must come entirely from the provided context, maintaining the supportive tone while never introducing information from outside the provided materials. | |
| # # # **Context:** {context} | |
| # # # **User's Question:** {question} | |
| # # # **Your Response:** | |
| # # # """ | |
| # # # rag_prompt = PromptTemplate.from_template(template) | |
| # # # def get_context_and_question(query_with_session): | |
| # # # # Extract query and session_id | |
| # # # query = query_with_session["query"] | |
| # # # session_id = query_with_session["session_id"] | |
| # # # # Get the user session | |
| # # # session = session_manager.get_session(session_id) | |
| # # # user_info = session.get_user() | |
| # # # first_name = user_info.get("Nickname", "User") | |
| # # # conversation_hist = session.get_formatted_history() | |
| # # # try: | |
| # # # # Retrieve relevant documents | |
| # # # retrieved_docs = retriever.invoke(query) | |
| # # # context_str = format_context(retrieved_docs) | |
| # # # except Exception as e: | |
| # # # print(f"ERROR retrieving documents: {e}") | |
| # # # context_str = "No relevant information found." | |
| # # # # Return the combined inputs for the prompt | |
| # # # return { | |
| # # # "context": context_str, | |
| # # # "question": query, | |
| # # # "first_name": first_name, | |
| # # # "conversation_history": conversation_hist | |
| # # # } | |
| # # # # Build the chain | |
| # # # try: | |
| # # # chain = ( | |
| # # # RunnablePassthrough() | |
| # # # | get_context_and_question | |
| # # # | rag_prompt | |
| # # # | llm | |
| # # # | StrOutputParser() | |
| # # # ) | |
| # # # return chain | |
| # # # except Exception as e: | |
| # # # print(f"ERROR creating RAG chain: {e}") | |
| # # # # Return a simple function as fallback | |
| # # # def fallback_chain(query_with_session): | |
| # # # session_id = query_with_session["session_id"] | |
| # # # session = session_manager.get_session(session_id) | |
| # # # nickname = session.get_user().get("Nickname", "there") | |
| # # # return f"I'm here to help you, {nickname}, but I'm experiencing some technical difficulties right now. Please try again shortly." | |
| # # # return fallback_chain | |
| # # # def format_context(retrieved_docs): | |
| # # # """Format retrieved documents into a string context.""" | |
| # # # if not retrieved_docs: | |
| # # # return "No relevant information available." | |
| # # # return "\n\n".join([doc.page_content for doc in retrieved_docs]) | |
| # # # def rag_memory_stream(message, history, session_id): | |
| # # # """Process user message and generate response with memory.""" | |
| # # # # Get the user session | |
| # # # session = session_manager.get_session(session_id) | |
| # # # # Add user message to history | |
| # # # session.add_to_history("user", message) | |
| # # # try: | |
| # # # # Get response from RAG chain | |
| # # # print(f"Processing message for session {session_id}: {message[:50]}...") | |
| # # # # Pass both query and session_id to the chain | |
| # # # response = rag_chain.invoke({ | |
| # # # "query": message, | |
| # # # "session_id": session_id | |
| # # # }) | |
| # # # print(f"Generated response: {response[:50]}...") | |
| # # # # Add assistant response to history | |
| # # # session.add_to_history("assistant", response) | |
| # # # # Yield the response | |
| # # # yield response | |
| # # # except Exception as e: | |
| # # # import traceback | |
| # # # print(f"ERROR in rag_memory_stream: {e}") | |
| # # # print(f"Detailed error: {traceback.format_exc()}") | |
| # # # nickname = session.get_user().get("Nickname", "there") | |
| # # # error_msg = f"I'm sorry, {nickname}. I encountered an error processing your request. Let's try a different question." | |
| # # # session.add_to_history("assistant", error_msg) | |
| # # # yield error_msg | |
| # # # def collect_user_info(nickname, session_id): | |
| # # # """Store user details and initialize session.""" | |
| # # # if not nickname or nickname.strip() == "": | |
| # # # return "Nickname is required to proceed.", gr.update(visible=False), gr.update(visible=True), [] | |
| # # # # Store user info for chat session | |
| # # # user_info = { | |
| # # # "Nickname": nickname.strip(), | |
| # # # "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") | |
| # # # } | |
| # # # # Get the session and set user info | |
| # # # session = session_manager.get_session(session_id) | |
| # # # session.set_user(user_info) | |
| # # # # Generate welcome message | |
| # # # welcome_message = session.get_welcome_message() | |
| # # # # Return welcome message and update UI | |
| # # # return welcome_message, gr.update(visible=True), gr.update(visible=False), [(None, welcome_message)] | |
| # # # def get_css(): | |
| # # # """Define CSS for the UI.""" | |
| # # # return """ | |
| # # # :root { | |
| # # # --primary: #4E6BBF; | |
| # # # --primary-light: #697BBF; | |
| # # # --text-primary: #333333; | |
| # # # --text-secondary: #666666; | |
| # # # --background: #F9FAFC; | |
| # # # --card-bg: #FFFFFF; | |
| # # # --border: #E1E5F0; | |
| # # # --shadow: rgba(0, 0, 0, 0.05); | |
| # # # } | |
| # # # body, .gradio-container { | |
| # # # margin: 0; | |
| # # # padding: 0; | |
| # # # width: 100vw; | |
| # # # height: 100vh; | |
| # # # display: flex; | |
| # # # flex-direction: column; | |
| # # # justify-content: center; | |
| # # # align-items: center; | |
| # # # background: var(--background); | |
| # # # color: var(--text-primary); | |
| # # # font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; | |
| # # # } | |
| # # # .gradio-container { | |
| # # # max-width: 100%; | |
| # # # max-height: 100%; | |
| # # # } | |
| # # # .gr-box { | |
| # # # background: var(--card-bg); | |
| # # # color: var(--text-primary); | |
| # # # border-radius: 12px; | |
| # # # padding: 2rem; | |
| # # # border: 1px solid var(--border); | |
| # # # box-shadow: 0 4px 12px var(--shadow); | |
| # # # } | |
| # # # .gr-button-primary { | |
| # # # background: var(--primary); | |
| # # # color: white; | |
| # # # padding: 12px 24px; | |
| # # # border-radius: 8px; | |
| # # # transition: all 0.3s ease; | |
| # # # border: none; | |
| # # # font-weight: bold; | |
| # # # } | |
| # # # .gr-button-primary:hover { | |
| # # # transform: translateY(-1px); | |
| # # # box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1); | |
| # # # background: var(--primary-light); | |
| # # # } | |
| # # # footer { | |
| # # # text-align: center; | |
| # # # color: var(--text-secondary); | |
| # # # padding: 1rem; | |
| # # # font-size: 0.9em; | |
| # # # } | |
| # # # .gr-markdown h2 { | |
| # # # color: var(--primary); | |
| # # # margin-bottom: 0.5rem; | |
| # # # font-size: 1.8em; | |
| # # # } | |
| # # # .gr-markdown h3 { | |
| # # # color: var(--text-secondary); | |
| # # # margin-bottom: 1.5rem; | |
| # # # font-weight: normal; | |
| # # # } | |
| # # # #chatbot_container .chat-title h1, | |
| # # # #chatbot_container .empty-chatbot { | |
| # # # color: var(--primary); | |
| # # # } | |
| # # # #input_nickname { | |
| # # # padding: 12px; | |
| # # # border-radius: 8px; | |
| # # # border: 1px solid var(--border); | |
| # # # background: var(--card-bg); | |
| # # # transition: all 0.3s ease; | |
| # # # } | |
| # # # #input_nickname:focus { | |
| # # # border-color: var(--primary); | |
| # # # box-shadow: 0 0 0 2px rgba(78, 107, 191, 0.2); | |
| # # # outline: none; | |
| # # # } | |
| # # # .chatbot-container .message.user { | |
| # # # background: #E8F0FE; | |
| # # # border-radius: 12px 12px 0 12px; | |
| # # # } | |
| # # # .chatbot-container .message.bot { | |
| # # # background: #F5F7FF; | |
| # # # border-radius: 12px 12px 12px 0; | |
| # # # } | |
| # # # """ | |
| # # # def create_ui(): | |
| # # # """Create and configure the Gradio UI.""" | |
| # # # with gr.Blocks(css=get_css(), theme=gr.themes.Soft()) as demo: | |
| # # # # Create a unique session ID for this browser tab | |
| # # # session_id = gr.State(value=f"session_{int(time.time())}_{os.urandom(4).hex()}") | |
| # # # # Registration section | |
| # # # with gr.Column(visible=True, elem_id="registration_container") as registration_container: | |
| # # # gr.Markdown(f"## Welcome to {APP_NAME}") | |
| # # # gr.Markdown("### Your privacy is important to us. Please provide a nickname to continue.") | |
| # # # with gr.Row(): | |
| # # # first_name = gr.Textbox( | |
| # # # label="Nickname", | |
| # # # placeholder="Enter your nickname", | |
| # # # scale=1, | |
| # # # elem_id="input_nickname" | |
| # # # ) | |
| # # # with gr.Row(): | |
| # # # submit_btn = gr.Button("Start Chatting", variant="primary", scale=2) | |
| # # # response_message = gr.Markdown() | |
| # # # # Chatbot section (initially hidden) | |
| # # # with gr.Column(visible=False, elem_id="chatbot_container") as chatbot_container: | |
| # # # # Create a custom chat interface to pass session_id to our function | |
| # # # chatbot = gr.Chatbot( | |
| # # # elem_id="chatbot", | |
| # # # height=500, | |
| # # # show_label=False | |
| # # # ) | |
| # # # with gr.Row(): | |
| # # # msg = gr.Textbox( | |
| # # # placeholder="Type your message here...", | |
| # # # show_label=False, | |
| # # # container=False, | |
| # # # scale=9 | |
| # # # ) | |
| # # # submit = gr.Button("Send", scale=1, variant="primary") | |
| # # # examples = gr.Examples( | |
| # # # examples=[ | |
| # # # "What resources are available for GBV victims?", | |
| # # # "How can I report an incident?", | |
| # # # "What are my legal rights?", | |
| # # # "I need help, what should I do first?" | |
| # # # ], | |
| # # # inputs=msg | |
| # # # ) | |
| # # # # Footer with version info | |
| # # # gr.Markdown(f"{APP_NAME} {APP_VERSION} © 2025") | |
| # # # # Handle chat message submission | |
| # # # def respond(message, chat_history, session_id): | |
| # # # bot_message = "" | |
| # # # for chunk in rag_memory_stream(message, chat_history, session_id): | |
| # # # bot_message += chunk | |
| # # # chat_history.append((message, bot_message)) | |
| # # # return "", chat_history | |
| # # # msg.submit(respond, [msg, chatbot, session_id], [msg, chatbot]) | |
| # # # submit.click(respond, [msg, chatbot, session_id], [msg, chatbot]) | |
| # # # # Handle user registration | |
| # # # submit_btn.click( | |
| # # # collect_user_info, | |
| # # # inputs=[first_name, session_id], | |
| # # # outputs=[response_message, chatbot_container, registration_container, chatbot] | |
| # # # ) | |
| # # # return demo | |
| # # # def launch_app(): | |
| # # # """Launch the Gradio interface.""" | |
| # # # ui = create_ui() | |
| # # # ui.launch(share=True) | |
| # # # # Main execution | |
| # # # if __name__ == "__main__": | |
| # # # try: | |
| # # # # Initialize and launch the assistant | |
| # # # initialize_assistant() | |
| # # # launch_app() | |
| # # # except Exception as e: | |
| # # # import traceback | |
| # # # print(f"❌ Fatal error initializing GBV Assistant: {e}") | |
| # # # print(traceback.format_exc()) | |
| # # # # Create a minimal emergency UI to display the error | |
| # # # with gr.Blocks() as error_demo: | |
| # # # gr.Markdown("## System Error") | |
| # # # gr.Markdown(f"An error occurred while initializing the application: {str(e)}") | |
| # # # gr.Markdown("Please check your configuration and try again.") | |
| # # # error_demo.launch(share=True, inbrowser=True, debug=True) | |
| # # ############################################################################################################ | |
| # # import os | |
| # # from langchain_groq import ChatGroq | |
| # # from langchain.prompts import ChatPromptTemplate, PromptTemplate | |
| # # from langchain.output_parsers import ResponseSchema, StructuredOutputParser | |
| # # from urllib.parse import urljoin, urlparse | |
| # # import requests | |
| # # from io import BytesIO | |
| # # from langchain_chroma import Chroma | |
| # # import requests | |
| # # from bs4 import BeautifulSoup | |
| # # from langchain_core.prompts import ChatPromptTemplate | |
| # # import gradio as gr | |
| # # from PyPDF2 import PdfReader | |
| # # from langchain_huggingface import HuggingFaceEmbeddings | |
| # # groq_api_key= os.environ.get('GBV') | |
| # # embed_model = HuggingFaceEmbeddings(model_name="mixedbread-ai/mxbai-embed-large-v1") | |
| # # def scrape_websites(base_urls): | |
| # # try: | |
| # # visited_links = set() # To avoid revisiting the same link | |
| # # content_by_url = {} # Store content from each URL | |
| # # for base_url in base_urls: | |
| # # if not base_url.strip(): | |
| # # continue # Skip empty or invalid URLs | |
| # # print(f"Scraping base URL: {base_url}") | |
| # # html_content = fetch_page_content(base_url) | |
| # # if html_content: | |
| # # cleaned_content = clean_body_content(html_content) | |
| # # content_by_url[base_url] = cleaned_content | |
| # # visited_links.add(base_url) | |
| # # # Extract and process all internal links | |
| # # soup = BeautifulSoup(html_content, "html.parser") | |
| # # links = extract_internal_links(base_url, soup) | |
| # # for link in links: | |
| # # if link not in visited_links: | |
| # # print(f"Scraping link: {link}") | |
| # # page_content = fetch_page_content(link) | |
| # # if page_content: | |
| # # cleaned_content = clean_body_content(page_content) | |
| # # content_by_url[link] = cleaned_content | |
| # # visited_links.add(link) | |
| # # # If the link is a PDF file, extract its content | |
| # # if link.lower().endswith('.pdf'): | |
| # # print(f"Extracting PDF content from: {link}") | |
| # # pdf_content = extract_pdf_text(link) | |
| # # if pdf_content: | |
| # # content_by_url[link] = pdf_content | |
| # # return content_by_url | |
| # # except Exception as e: | |
| # # print(f"Error during scraping: {e}") | |
| # # return {} | |
| # # def fetch_page_content(url): | |
| # # try: | |
| # # response = requests.get(url, timeout=10) | |
| # # response.raise_for_status() | |
| # # return response.text | |
| # # except requests.exceptions.RequestException as e: | |
| # # print(f"Error fetching {url}: {e}") | |
| # # return None | |
| # # def extract_internal_links(base_url, soup): | |
| # # links = set() | |
| # # for anchor in soup.find_all("a", href=True): | |
| # # href = anchor["href"] | |
| # # full_url = urljoin(base_url, href) | |
| # # if is_internal_link(base_url, full_url): | |
| # # links.add(full_url) | |
| # # return links | |
| # # def is_internal_link(base_url, link_url): | |
| # # base_netloc = urlparse(base_url).netloc | |
| # # link_netloc = urlparse(link_url).netloc | |
| # # return base_netloc == link_netloc | |
| # # def extract_pdf_text(pdf_url): | |
| # # try: | |
| # # response = requests.get(pdf_url) | |
| # # response.raise_for_status() | |
| # # with BytesIO(response.content) as file: | |
| # # reader = PdfReader(file) | |
| # # pdf_text = "" | |
| # # for page in reader.pages: | |
| # # pdf_text += page.extract_text() | |
| # # return pdf_text if pdf_text else None | |
| # # except requests.exceptions.RequestException as e: | |
| # # print(f"Error fetching PDF {pdf_url}: {e}") | |
| # # return None | |
| # # except Exception as e: | |
| # # print(f"Error reading PDF {pdf_url}: {e}") | |
| # # return None | |
| # # def clean_body_content(html_content): | |
| # # soup = BeautifulSoup(html_content, "html.parser") | |
| # # for script_or_style in soup(["script", "style"]): | |
| # # script_or_style.extract() | |
| # # cleaned_content = soup.get_text(separator="\n") | |
| # # cleaned_content = "\n".join( | |
| # # line.strip() for line in cleaned_content.splitlines() if line.strip() | |
| # # ) | |
| # # return cleaned_content | |
| # # if __name__ == "__main__": | |
| # # website = ["https://haguruka.org.rw/" | |
| # # ] | |
| # # all_content = scrape_websites(website) | |
| # # temp_list = [] | |
| # # for url, content in all_content.items(): | |
| # # temp_list.append((url, content)) | |
| # # processed_texts = [] | |
| # # for element in temp_list: | |
| # # if isinstance(element, tuple): | |
| # # url, content = element | |
| # # processed_texts.append(f"url: {url}, content: {content}") | |
| # # elif isinstance(element, str): | |
| # # processed_texts.append(element) | |
| # # else: | |
| # # processed_texts.append(str(element)) | |
| # # def chunk_string(s, chunk_size=1000): | |
| # # return [s[i:i+chunk_size] for i in range(0, len(s), chunk_size)] | |
| # # chunked_texts = [] | |
| # # for text in processed_texts: | |
| # # chunked_texts.extend(chunk_string(text)) | |
| # # vectorstore = Chroma( | |
| # # collection_name="GBVR_Dataset", | |
| # # embedding_function=embed_model, | |
| # # persist_directory="./", | |
| # # ) | |
| # # vectorstore.get().keys() | |
| # # vectorstore.add_texts(chunked_texts) | |
| # # template = (""" | |
| # # You are a friendly, intelligent, and conversational AI assistant designed to provide accurate, engaging, and human-like responses based on the given context. Your goal is to extract relevant details from the provided context: {context} and assist the user effectively. Follow these guidelines: | |
| # # 1. **Warm & Natural Interaction** | |
| # # - If the user greets you (e.g., "Hello," "Hi," "Good morning"), respond warmly and acknowledge them. | |
| # # - Example responses: | |
| # # - "😊 Good morning! How can I assist you today?" | |
| # # - "Hello! What can I do for you? 🚀" | |
| # # 2. **Precise Information Extraction** | |
| # # - Provide only the relevant details from the given context: {context}. | |
| # # - Do not generate extra content or assumptions beyond the provided information. | |
| # # 3. **Conversational & Engaging Tone** | |
| # # - Keep responses friendly, natural, and engaging. | |
| # # - Use occasional emojis (e.g., 😊, 🚀) to make interactions more lively. | |
| # # 4. **Awareness of Real-Time Context** | |
| # # - If necessary, acknowledge the current date and time to show awareness of real-world updates. | |
| # # 5. **Handling Missing Information** | |
| # # - If no relevant information exists in the context, respond politely: | |
| # # - "I don't have that information at the moment, but I'm happy to help with something else! 😊" | |
| # # 6. **Personalized Interaction** | |
| # # - If user history is available, tailor responses based on their previous interactions for a more natural and engaging conversation. | |
| # # 7. **Direct, Concise Responses** | |
| # # - If the user requests specific data, provide only the requested details without unnecessary explanations unless asked. | |
| # # 8. **Extracting Relevant Links** | |
| # # - If the user asks for a link related to their request `{question}`, extract the most relevant URL from `{context}` and provide it directly. | |
| # # - Example response: | |
| # # - "Here is the link you requested: [URL]" | |
| # # **Context:** {context} | |
| # # **User's Question:** {question} | |
| # # **Your Response:** | |
| # # """) | |
| # # rag_prompt = PromptTemplate.from_template(template) | |
| # # retriever = vectorstore.as_retriever() | |
| # # from langchain_core.output_parsers import StrOutputParser | |
| # # from langchain_core.runnables import RunnablePassthrough | |
| # # llm = ChatGroq(model="llama-3.3-70b-versatile", api_key=groq_api_key ) | |
| # # rag_chain = ( | |
| # # {"context": retriever, "question": RunnablePassthrough()} | |
| # # | rag_prompt | |
| # # | llm | |
| # # | StrOutputParser() | |
| # # ) | |
| # # # Define the RAG memory stream function | |
| # # def rag_memory_stream(message, history): | |
| # # partial_text = "" | |
| # # for new_text in rag_chain.stream(message): # Replace with actual streaming logic | |
| # # partial_text += new_text | |
| # # yield partial_text | |
| # # # Title with emojis | |
| # # title = "GBVR Chatbot" | |
| # # # Custom CSS for styling the interface | |
| # # custom_css = """ | |
| # # body { | |
| # # font-family: "Arial", serif; | |
| # # } | |
| # # .gradio-container { | |
| # # font-family: "Times New Roman", serif; | |
| # # } | |
| # # .gr-button { | |
| # # background-color: #007bff; /* Blue button */ | |
| # # color: white; | |
| # # border: none; | |
| # # border-radius: 5px; | |
| # # font-size: 16px; | |
| # # padding: 10px 20px; | |
| # # cursor: pointer; | |
| # # } | |
| # # .gr-textbox:focus, .gr-button:focus { | |
| # # outline: none; /* Remove outline focus for a cleaner look */ | |
| # # } | |
| # # """ | |
| # # # Create the Chat Interface | |
| # # demo = gr.ChatInterface( | |
| # # fn=rag_memory_stream, | |
| # # title=title, | |
| # # fill_height=True, | |
| # # theme="soft", | |
| # # css=custom_css, # Apply the custom CSS | |
| # # ) | |
| # # # Launch the app | |
| # # if __name__ == "__main__": | |
| # # demo.launch(share=True, inbrowser=True, debug=True) | |
| # import os | |
| # from langchain_groq import ChatGroq | |
| # from langchain.prompts import ChatPromptTemplate, PromptTemplate | |
| # from langchain.output_parsers import ResponseSchema, StructuredOutputParser | |
| # from urllib.parse import urljoin, urlparse | |
| # import requests | |
| # from io import BytesIO | |
| # from langchain_chroma import Chroma | |
| # import requests | |
| # from bs4 import BeautifulSoup | |
| # from langchain_core.prompts import ChatPromptTemplate | |
| # import gradio as gr | |
| # from PyPDF2 import PdfReader | |
| # from langchain_huggingface import HuggingFaceEmbeddings | |
| # from langchain_core.output_parsers import StrOutputParser | |
| # from langchain_core.runnables import RunnablePassthrough | |
| # # Simple session management | |
| # class SessionManager: | |
| # def __init__(self): | |
| # self.sessions = {} | |
| # def get_or_create_session(self, session_id): | |
| # if session_id not in self.sessions: | |
| # self.sessions[session_id] = [] | |
| # return self.sessions[session_id] | |
| # def add_interaction(self, session_id, user_message, ai_response): | |
| # session = self.get_or_create_session(session_id) | |
| # session.append({"user": user_message, "ai": ai_response}) | |
| # def get_history(self, session_id, max_turns=5): | |
| # session = self.get_or_create_session(session_id) | |
| # recent_history = session[-max_turns:] if len(session) > max_turns else session | |
| # history_text = "" | |
| # for interaction in recent_history: | |
| # history_text += f"User: {interaction['user']}\n" | |
| # history_text += f"Assistant: {interaction['ai']}\n\n" | |
| # return history_text.strip() | |
| # # Initialize session manager | |
| # session_manager = SessionManager() | |
| # groq_api_key= os.environ.get('GBV') | |
| # embed_model = HuggingFaceEmbeddings(model_name="mixedbread-ai/mxbai-embed-large-v1") | |
| # def scrape_websites(base_urls): | |
| # try: | |
| # visited_links = set() # To avoid revisiting the same link | |
| # content_by_url = {} # Store content from each URL | |
| # for base_url in base_urls: | |
| # if not base_url.strip(): | |
| # continue # Skip empty or invalid URLs | |
| # print(f"Scraping base URL: {base_url}") | |
| # html_content = fetch_page_content(base_url) | |
| # if html_content: | |
| # cleaned_content = clean_body_content(html_content) | |
| # content_by_url[base_url] = cleaned_content | |
| # visited_links.add(base_url) | |
| # # Extract and process all internal links | |
| # soup = BeautifulSoup(html_content, "html.parser") | |
| # links = extract_internal_links(base_url, soup) | |
| # for link in links: | |
| # if link not in visited_links: | |
| # print(f"Scraping link: {link}") | |
| # page_content = fetch_page_content(link) | |
| # if page_content: | |
| # cleaned_content = clean_body_content(page_content) | |
| # content_by_url[link] = cleaned_content | |
| # visited_links.add(link) | |
| # # If the link is a PDF file, extract its content | |
| # if link.lower().endswith('.pdf'): | |
| # print(f"Extracting PDF content from: {link}") | |
| # pdf_content = extract_pdf_text(link) | |
| # if pdf_content: | |
| # content_by_url[link] = pdf_content | |
| # return content_by_url | |
| # except Exception as e: | |
| # print(f"Error during scraping: {e}") | |
| # return {} | |
| # def fetch_page_content(url): | |
| # try: | |
| # response = requests.get(url, timeout=10) | |
| # response.raise_for_status() | |
| # return response.text | |
| # except requests.exceptions.RequestException as e: | |
| # print(f"Error fetching {url}: {e}") | |
| # return None | |
| # def extract_internal_links(base_url, soup): | |
| # links = set() | |
| # for anchor in soup.find_all("a", href=True): | |
| # href = anchor["href"] | |
| # full_url = urljoin(base_url, href) | |
| # if is_internal_link(base_url, full_url): | |
| # links.add(full_url) | |
| # return links | |
| # def is_internal_link(base_url, link_url): | |
| # base_netloc = urlparse(base_url).netloc | |
| # link_netloc = urlparse(link_url).netloc | |
| # return base_netloc == link_netloc | |
| # def extract_pdf_text(pdf_url): | |
| # try: | |
| # response = requests.get(pdf_url) | |
| # response.raise_for_status() | |
| # with BytesIO(response.content) as file: | |
| # reader = PdfReader(file) | |
| # pdf_text = "" | |
| # for page in reader.pages: | |
| # pdf_text += page.extract_text() | |
| # return pdf_text if pdf_text else None | |
| # except requests.exceptions.RequestException as e: | |
| # print(f"Error fetching PDF {pdf_url}: {e}") | |
| # return None | |
| # except Exception as e: | |
| # print(f"Error reading PDF {pdf_url}: {e}") | |
| # return None | |
| # def clean_body_content(html_content): | |
| # soup = BeautifulSoup(html_content, "html.parser") | |
| # for script_or_style in soup(["script", "style"]): | |
| # script_or_style.extract() | |
| # cleaned_content = soup.get_text(separator="\n") | |
| # cleaned_content = "\n".join( | |
| # line.strip() for line in cleaned_content.splitlines() if line.strip() | |
| # ) | |
| # return cleaned_content | |
| # if __name__ == "__main__": | |
| # website = ["https://haguruka.org.rw/" | |
| # ] | |
| # all_content = scrape_websites(website) | |
| # temp_list = [] | |
| # for url, content in all_content.items(): | |
| # temp_list.append((url, content)) | |
| # processed_texts = [] | |
| # for element in temp_list: | |
| # if isinstance(element, tuple): | |
| # url, content = element | |
| # processed_texts.append(f"url: {url}, content: {content}") | |
| # elif isinstance(element, str): | |
| # processed_texts.append(element) | |
| # else: | |
| # processed_texts.append(str(element)) | |
| # def chunk_string(s, chunk_size=1000): | |
| # return [s[i:i+chunk_size] for i in range(0, len(s), chunk_size)] | |
| # chunked_texts = [] | |
| # for text in processed_texts: | |
| # chunked_texts.extend(chunk_string(text)) | |
| # vectorstore = Chroma( | |
| # collection_name="GBVR_Dataset", | |
| # embedding_function=embed_model, | |
| # persist_directory="./", | |
| # ) | |
| # vectorstore.get().keys() | |
| # vectorstore.add_texts(chunked_texts) | |
| # # Updated template to include conversation history | |
| # template = (""" | |
| # You are a friendly, intelligent, and conversational AI assistant designed to provide accurate, engaging, and human-like responses based on the given context. Your goal is to extract relevant details from the provided context: {context} and assist the user effectively. Follow these guidelines: | |
| # 1. **Warm & Natural Interaction** | |
| # - If the user greets you (e.g., "Hello," "Hi," "Good morning"), respond warmly and acknowledge them. | |
| # - Example responses: | |
| # - "😊 Good morning! How can I assist you today?" | |
| # - "Hello! What can I do for you? 🚀" | |
| # 2. **Precise Information Extraction** | |
| # - Provide only the relevant details from the given context: {context}. | |
| # - Do not generate extra content or assumptions beyond the provided information. | |
| # 3. **Conversational & Engaging Tone** | |
| # - Keep responses friendly, natural, and engaging. | |
| # - Use occasional emojis (e.g., 😊, 🚀) to make interactions more lively. | |
| # 4. **Awareness of Real-Time Context** | |
| # - If necessary, acknowledge the current date and time to show awareness of real-world updates. | |
| # 5. **Handling Missing Information** | |
| # - If no relevant information exists in the context, respond politely: | |
| # - "I don't have that information at the moment, but I'm happy to help with something else! 😊" | |
| # 6. **Personalized Interaction** | |
| # - Use the conversation history to provide more personalized and contextually relevant responses. | |
| # - Previous conversation history: {conversation_history} | |
| # 7. **Direct, Concise Responses** | |
| # - If the user requests specific data, provide only the requested details without unnecessary explanations unless asked. | |
| # 8. **Extracting Relevant Links** | |
| # - If the user asks for a link related to their request `{question}`, extract the most relevant URL from `{context}` and provide it directly. | |
| # - Example response: | |
| # - "Here is the link you requested: [URL]" | |
| # **Context:** {context} | |
| # **User's Question:** {question} | |
| # **Your Response:** | |
| # """) | |
| # rag_prompt = PromptTemplate.from_template(template) | |
| # retriever = vectorstore.as_retriever() | |
| # llm = ChatGroq(model="llama-3.3-70b-versatile", api_key=groq_api_key) | |
| # # Dictionary to store user sessions with session IDs | |
| # user_sessions = {} | |
| # # Define the RAG chain with session history | |
| # def rag_chain(question, session_id="default"): | |
| # # Get conversation history if available | |
| # conversation_history = session_manager.get_history(session_id) | |
| # # Get context from retriever | |
| # context_docs = retriever.invoke(question) | |
| # context = "\n".join(doc.page_content for doc in context_docs) | |
| # # Create prompt with history | |
| # prompt = rag_prompt.format( | |
| # context=context, | |
| # question=question, | |
| # conversation_history=conversation_history | |
| # ) | |
| # # Generate response | |
| # response = llm.invoke(prompt).content | |
| # # Store the interaction | |
| # session_manager.add_interaction(session_id, question, response) | |
| # return response | |
| # # Define the RAG memory stream function | |
| # def rag_memory_stream(message, history): | |
| # # Generate a session ID based on the first message if not exists | |
| # session_id = None | |
| # for msg in history: | |
| # if msg[0]: # If there's a user message | |
| # # Use first few characters of first message as simple session ID | |
| # session_id = hash(msg[0][:20]) if session_id is None else session_id | |
| # break | |
| # # Default session ID if history is empty | |
| # if session_id is None: | |
| # session_id = "default_session" | |
| # # Process the message and get response | |
| # response = rag_chain(message, str(session_id)) | |
| # # Stream the response word by word | |
| # partial_text = "" | |
| # words = response.split(' ') | |
| # for word in words: | |
| # partial_text += word + " " | |
| # yield partial_text.strip() | |
| # # Title with emojis | |
| # title = "GBVR Chatbot" | |
| # # Custom CSS for styling the interface | |
| # custom_css = """ | |
| # body { | |
| # font-family: "Arial", serif; | |
| # } | |
| # .gradio-container { | |
| # font-family: "Times New Roman", serif; | |
| # } | |
| # .gr-button { | |
| # background-color: #007bff; /* Blue button */ | |
| # color: white; | |
| # border: none; | |
| # border-radius: 5px; | |
| # font-size: 16px; | |
| # padding: 10px 20px; | |
| # cursor: pointer; | |
| # } | |
| # .gr-textbox:focus, .gr-button:focus { | |
| # outline: none; /* Remove outline focus for a cleaner look */ | |
| # } | |
| # """ | |
| # # Create the Chat Interface | |
| # demo = gr.ChatInterface( | |
| # fn=rag_memory_stream, | |
| # title=title, | |
| # fill_height=True, | |
| # theme="soft", | |
| # css=custom_css, # Apply the custom CSS | |
| # ) | |
| # # Launch the app | |
| # if __name__ == "__main__": | |
| # demo.launch(share=True, inbrowser=True, debug=True) | |
| import os | |
| from langchain_groq import ChatGroq | |
| from langchain.prompts import ChatPromptTemplate, PromptTemplate | |
| from langchain.output_parsers import ResponseSchema, StructuredOutputParser | |
| from urllib.parse import urljoin, urlparse | |
| import requests | |
| from io import BytesIO | |
| from langchain_chroma import Chroma | |
| import requests | |
| from bs4 import BeautifulSoup | |
| from langchain_core.prompts import ChatPromptTemplate | |
| import gradio as gr | |
| from PyPDF2 import PdfReader | |
| from langchain_huggingface import HuggingFaceEmbeddings | |
| from langchain_core.output_parsers import StrOutputParser | |
| from langchain_core.runnables import RunnablePassthrough | |
| # Simple session management | |
| class SessionManager: | |
| def __init__(self): | |
| self.sessions = {} | |
| def get_or_create_session(self, session_id): | |
| if session_id not in self.sessions: | |
| self.sessions[session_id] = [] | |
| return self.sessions[session_id] | |
| def add_interaction(self, session_id, user_message, ai_response): | |
| session = self.get_or_create_session(session_id) | |
| session.append({"user": user_message, "ai": ai_response}) | |
| def get_history(self, session_id, max_turns=5): | |
| session = self.get_or_create_session(session_id) | |
| recent_history = session[-max_turns:] if len(session) > max_turns else session | |
| history_text = "" | |
| for interaction in recent_history: | |
| history_text += f"User: {interaction['user']}\n" | |
| history_text += f"Assistant: {interaction['ai']}\n\n" | |
| return history_text.strip() | |
| # Initialize session manager | |
| session_manager = SessionManager() | |
| groq_api_key= os.environ.get('GBV') | |
| embed_model = HuggingFaceEmbeddings(model_name="mixedbread-ai/mxbai-embed-large-v1") | |
| def scrape_websites(base_urls): | |
| try: | |
| visited_links = set() # To avoid revisiting the same link | |
| content_by_url = {} # Store content from each URL | |
| for base_url in base_urls: | |
| if not base_url.strip(): | |
| continue # Skip empty or invalid URLs | |
| print(f"Scraping base URL: {base_url}") | |
| html_content = fetch_page_content(base_url) | |
| if html_content: | |
| cleaned_content = clean_body_content(html_content) | |
| content_by_url[base_url] = cleaned_content | |
| visited_links.add(base_url) | |
| # Extract and process all internal links | |
| soup = BeautifulSoup(html_content, "html.parser") | |
| links = extract_internal_links(base_url, soup) | |
| for link in links: | |
| if link not in visited_links: | |
| print(f"Scraping link: {link}") | |
| page_content = fetch_page_content(link) | |
| if page_content: | |
| cleaned_content = clean_body_content(page_content) | |
| content_by_url[link] = cleaned_content | |
| visited_links.add(link) | |
| # If the link is a PDF file, extract its content | |
| if link.lower().endswith('.pdf'): | |
| print(f"Extracting PDF content from: {link}") | |
| pdf_content = extract_pdf_text(link) | |
| if pdf_content: | |
| content_by_url[link] = pdf_content | |
| return content_by_url | |
| except Exception as e: | |
| print(f"Error during scraping: {e}") | |
| return {} | |
| def fetch_page_content(url): | |
| try: | |
| response = requests.get(url, timeout=10) | |
| response.raise_for_status() | |
| return response.text | |
| except requests.exceptions.RequestException as e: | |
| print(f"Error fetching {url}: {e}") | |
| return None | |
| def extract_internal_links(base_url, soup): | |
| links = set() | |
| for anchor in soup.find_all("a", href=True): | |
| href = anchor["href"] | |
| full_url = urljoin(base_url, href) | |
| if is_internal_link(base_url, full_url): | |
| links.add(full_url) | |
| return links | |
| def is_internal_link(base_url, link_url): | |
| base_netloc = urlparse(base_url).netloc | |
| link_netloc = urlparse(link_url).netloc | |
| return base_netloc == link_netloc | |
| def extract_pdf_text(pdf_url): | |
| try: | |
| response = requests.get(pdf_url) | |
| response.raise_for_status() | |
| with BytesIO(response.content) as file: | |
| reader = PdfReader(file) | |
| pdf_text = "" | |
| for page in reader.pages: | |
| pdf_text += page.extract_text() | |
| return pdf_text if pdf_text else None | |
| except requests.exceptions.RequestException as e: | |
| print(f"Error fetching PDF {pdf_url}: {e}") | |
| return None | |
| except Exception as e: | |
| print(f"Error reading PDF {pdf_url}: {e}") | |
| return None | |
| def clean_body_content(html_content): | |
| soup = BeautifulSoup(html_content, "html.parser") | |
| for script_or_style in soup(["script", "style"]): | |
| script_or_style.extract() | |
| cleaned_content = soup.get_text(separator="\n") | |
| cleaned_content = "\n".join( | |
| line.strip() for line in cleaned_content.splitlines() if line.strip() | |
| ) | |
| return cleaned_content | |
| if __name__ == "__main__": | |
| website = ["https://haguruka.org.rw/" | |
| ] | |
| all_content = scrape_websites(website) | |
| temp_list = [] | |
| for url, content in all_content.items(): | |
| temp_list.append((url, content)) | |
| processed_texts = [] | |
| for element in temp_list: | |
| if isinstance(element, tuple): | |
| url, content = element | |
| processed_texts.append(f"url: {url}, content: {content}") | |
| elif isinstance(element, str): | |
| processed_texts.append(element) | |
| else: | |
| processed_texts.append(str(element)) | |
| def chunk_string(s, chunk_size=1000): | |
| return [s[i:i+chunk_size] for i in range(0, len(s), chunk_size)] | |
| chunked_texts = [] | |
| for text in processed_texts: | |
| chunked_texts.extend(chunk_string(text)) | |
| vectorstore = Chroma( | |
| collection_name="GBVR_Datast", | |
| embedding_function=embed_model, | |
| persist_directory="./", | |
| ) | |
| vectorstore.get().keys() | |
| vectorstore.add_texts(chunked_texts) | |
| # Updated template to include conversation history | |
| template = (""" | |
| You are a friendly, intelligent, and conversational AI assistant designed to provide accurate, engaging, and human-like responses based on the given context. Your goal is to extract relevant details from the provided context: {context} and assist the user effectively. Follow these guidelines: | |
| 1. **Warm & Natural Interaction** | |
| - If the user greets you (e.g., "Hello," "Hi," "Good morning"), respond warmly and acknowledge them. | |
| - Example responses: | |
| - "😊 Good morning! How can I assist you today?" | |
| - "Hello! What can I do for you? 🚀" | |
| 2. **Precise Information Extraction** | |
| - Provide only the relevant details from the given context: {context}. | |
| - Do not generate extra content or assumptions beyond the provided information. | |
| 3. **Conversational & Engaging Tone** | |
| - Keep responses friendly, natural, and engaging. | |
| - Use occasional emojis (e.g., 😊, 🚀) to make interactions more lively. | |
| 4. **Awareness of Real-Time Context** | |
| - If necessary, acknowledge the current date and time to show awareness of real-world updates. | |
| 5. **Handling Missing Information** | |
| - If no relevant information exists in the context, respond politely: | |
| - "I don't have that information at the moment, but I'm happy to help with something else! 😊" | |
| 6. **Personalized Interaction** | |
| - Use the conversation history to provide more personalized and contextually relevant responses. | |
| - Previous conversation history: {conversation_history} | |
| 7. **Direct, Concise Responses** | |
| - If the user requests specific data, provide only the requested details without unnecessary explanations unless asked. | |
| 8. **Extracting Relevant Links** | |
| - If the user asks for a link related to their request `{question}`, extract the most relevant URL from `{context}` and provide it directly. | |
| - Example response: | |
| - "Here is the link you requested: [URL]" | |
| **Context:** {context} | |
| **User's Question:** {question} | |
| **Your Response:** | |
| """) | |
| rag_prompt = PromptTemplate.from_template(template) | |
| retriever = vectorstore.as_retriever() | |
| llm = ChatGroq(model="llama-3.3-70b-versatile", api_key=groq_api_key) | |
| # Dictionary to store user sessions with session IDs | |
| user_sessions = {} | |
| # Define the RAG chain with session history | |
| def rag_chain(question, session_id="default"): | |
| # Get conversation history if available | |
| conversation_history = session_manager.get_history(session_id) | |
| # Get context from retriever | |
| context_docs = retriever.invoke(question) | |
| context = "\n".join(doc.page_content for doc in context_docs) | |
| # Create prompt with history | |
| prompt = rag_prompt.format( | |
| context=context, | |
| question=question, | |
| conversation_history=conversation_history | |
| ) | |
| # Generate response | |
| response = llm.invoke(prompt).content | |
| # Store the interaction | |
| session_manager.add_interaction(session_id, question, response) | |
| return response | |
| # Define the RAG memory stream function | |
| def rag_memory_stream(message, history): | |
| # Generate a session ID based on the first message if not exists | |
| session_id = None | |
| for msg in history: | |
| if msg[0]: # If there's a user message | |
| # Use first few characters of first message as simple session ID | |
| session_id = hash(msg[0][:20]) if session_id is None else session_id | |
| break | |
| # Default session ID if history is empty | |
| if session_id is None: | |
| session_id = "default_session" | |
| # Process the message and get response | |
| response = rag_chain(message, str(session_id)) | |
| # Stream the response word by word | |
| partial_text = "" | |
| words = response.split(' ') | |
| for word in words: | |
| partial_text += word + " " | |
| yield partial_text.strip() | |
| # Title with emojis | |
| title = "GBVR Chatbot" | |
| # Custom CSS for styling the interface | |
| custom_css = """ | |
| /* Custom CSS for styling the interface */ | |
| body { | |
| font-family: "Arial", serif; | |
| } | |
| .gradio-container { | |
| font-family: "Times New Roman", serif; | |
| } | |
| .gr-button { | |
| background-color: #007bff; /* Blue button */ | |
| color: white; | |
| border: none; | |
| border-radius: 5px; | |
| font-size: 16px; | |
| padding: 10px 20px; | |
| cursor: pointer; | |
| } | |
| .gr-textbox:focus, .gr-button:focus { | |
| outline: none; /* Remove outline focus for a cleaner look */ | |
| } | |
| /* Specific CSS for the welcome message */ | |
| .gradio-description { | |
| font-size: 30px; /* Set font size for the welcome message */ | |
| font-family: "Arial", sans-serif; | |
| text-align: center; /* Optional: Center-align the text */ | |
| padding: 20px; /* Optional: Add padding around the welcome message */ | |
| } | |
| """ | |
| # Generate a simple welcome message using the LLM | |
| def generate_welcome_message(): | |
| welcome_prompt = """ | |
| Generate a short, simple welcome message for a chatbot about Gender-Based Violence Resources in Rwanda. | |
| Keep it under 3 sentences, and use simple language. | |
| Make it warm and supportive but direct and easy to read. | |
| """ | |
| # Get the welcome message from the LLM | |
| welcome_message = llm.invoke(welcome_prompt).content | |
| return welcome_message | |
| # Create simple welcome message | |
| welcome_msg = generate_welcome_message() | |
| # Create the Chat Interface with welcome message | |
| demo = gr.ChatInterface( | |
| fn=rag_memory_stream, | |
| title=title, | |
| fill_height=True, | |
| theme="soft", | |
| css=custom_css, # Apply the custom CSS | |
| description=welcome_msg | |
| ) | |
| # Launch the app | |
| if __name__ == "__main__": | |
| demo.launch(share=True, inbrowser=True, debug=True) |