Spaces:
Runtime error
Runtime error
| # ============================================================================== | |
| # 1. IMPORTS & SETUP | |
| # ============================================================================== | |
| import os | |
| import re | |
| import difflib | |
| import pandas as pd | |
| from datetime import datetime | |
| import gradio as gr | |
| from transformers import pipeline, set_seed ,BitsAndBytesConfig | |
| # --- Model Loading --- | |
| # Read the Hugging Face token from the environment secrets | |
| hf_token = os.environ.get("HUGGING_FACE_HUB_TOKEN") | |
| # Check if the token is available | |
| if not hf_token: | |
| print("Warning: HUGGING_FACE_HUB_TOKEN not found. This may fail if the model is private.") | |
| print("Loading conversational model...") | |
| # Use the token to authenticate when loading the model | |
| bnb_config = BitsAndBytesConfig( | |
| load_in_4bit=True, # Set to True for 4-bit quantization | |
| bnb_4bit_quant_type="nf4", # "nf4" is usually the most stable | |
| bnb_4bit_use_double_quant=True, # Double quantization helps reduce memory | |
| bnb_4bit_compute_dtype="float16" # Safer than float32, faster on GPUs | |
| ) | |
| # Use the token to authenticate when loading the model | |
| generator = pipeline( | |
| 'text-generation', | |
| model='Warrior786/acc-llama-1b-merged-v10-lite', | |
| token=hf_token, | |
| device_map="auto", # Let HF infer best device usage | |
| model_kwargs={"quantization_config": bnb_config} | |
| ) | |
| set_seed(42) | |
| print("Model loaded successfully.") | |
| # --- File Paths --- | |
| # Log conversations to a local CSV file within the Hugging Face Space | |
| LOG_FILE = "ACC_chat_history.csv" | |
| # ============================================================================== | |
| # 2. CONSTANTS (Allowed Links & Emails) | |
| # ============================================================================== | |
| # List of all approved URLs for the chatbot to share | |
| ALLOWED_LINKS = [ | |
| "https://acceducate.org/", "https://acceducate.org/apply/", "https://acceducate.org/faqs/", | |
| "https://acceducate.org/programs/a-continuous-charity-acc-student-loan-refinance-program/", | |
| "https://acceducate.org/programs/", "https://acceducate.org/programs/help-a-student-with-sadaqah-jariyah/", | |
| "https://acceducate.org/programs/zakat/", "https://acceducate.org/programs/change-maker/", | |
| "https://acceducate.org/programs/break-the-chains-of-riba/", "https://acceducate.org/programs/women-empowerment/", | |
| "https://acceducate.org/programs/educational-justice/", "https://acceducate.org/ways-to-give/", | |
| "https://acceducate.org/ways-to-give/donate-stocks/", "https://acceducate.org/ways-to-give/open-an-educational-partnership-fund/", | |
| "https://acceducate.org/ways-to-give/legacy-fund/", "https://acceducate.org/ways-to-give/monthly-giving/", | |
| "https://acceducate.org/ways-to-give/support-accs-waqf-endowment-a-legacy-of-sustainable-charity/", | |
| "https://acceducate.org/ways-to-give/donate-by-zelle/", "https://acceducate.org/ways-to-give/matchinggift/", | |
| "https://acceducate.org/ways-to-give/donate-your-ira-to-acc/", "https://acceducate.org/ways-to-give/corporate-sponsored-volunteering/", | |
| "https://acceducate.org/ways-to-give/buy-coffee-give-sadaqah/", "https://acceducate.org/ways-to-give/host-an-iftar/", | |
| "https://acceducate.org/campaigns/grants/", "https://acceducate.org/student-initiated-fundraising-initiative/", | |
| "https://acceducate.org/about-us/", "https://acceducate.org/our-team/", "https://acceducate.org/financials/", | |
| "https://acceducate.org/careers/", "https://acceducate.org/why-acc/", "https://acceducate.org/invite-us/", | |
| "https://acceducate.org/events/", "https://acceducate.org/testimonials/", "https://www.facebook.com/acontinuouscharity/", | |
| "https://www.instagram.com/acontinuouscharity/", "https://www.youtube.com/channel/UCkq8wAvAqt54yzjzL_QIq3w", | |
| "https://twitter.com/accnational", | |
| ] | |
| # Mapping of keywords to specific ACC pages for better matching | |
| ACC_LINKS = { | |
| "About Us": "https://acceducate.org/about-us/", "Apply": "https://acceducate.org/apply/", | |
| "Careers": "https://acceducate.org/careers/", "Donate": "https://acceducate.org/ways-to-give/", | |
| "Events": "https://acceducate.org/events/", "FAQs": "https://acceducate.org/faqs/", | |
| "Zakat": "https://acceducate.org/programs/zakat/", "Loan Refinance": "https://acceducate.org/programs/a-continuous-charity-acc-student-loan-refinance-program/", | |
| } | |
| # Allowed email addresses | |
| ALLOWED_EMAILS = ["donate@acceducate.org"] | |
| # ============================================================================== | |
| # 3. HELPER FUNCTIONS | |
| # ============================================================================== | |
| def normalize_acc_link(url: str, user_msg: str = "") -> str: | |
| """Cleans and restricts a given URL to the official list of allowed ACC links.""" | |
| url = re.sub(r"https?://[^/\s]*acc[^/\s]*\.org", "https://acceducate.org", url) | |
| url = re.sub(r"(/(\w+))(\s*/\2)+", r"\1", url) | |
| if not url.endswith("/"): url += "/" | |
| if url in ALLOWED_LINKS: return url | |
| for keyword, link in ACC_LINKS.items(): | |
| if keyword.lower() in user_msg.lower(): return link | |
| closest = difflib.get_close_matches(url, ALLOWED_LINKS, n=1, cutoff=0.6) | |
| return closest[0] if closest else "https://acceducate.org/" | |
| def sanitize_response_links(text: str, user_msg: str = "") -> str: | |
| """Finds and normalizes all URLs in the bot's response.""" | |
| text = re.sub(r"https?://\S+", lambda m: normalize_acc_link(m.group(0), user_msg), text) | |
| def replace_md(match): | |
| display, url = match.groups() | |
| final_url = normalize_acc_link(url, user_msg) | |
| norm_display = re.sub(r"^https?://", "", final_url).rstrip("/") | |
| return f"[{norm_display}]({final_url})" | |
| text = re.sub(r"\[([^\]]+)\]\((https?://[^\)]+)\)", replace_md, text) | |
| return text | |
| def is_valid_email(email: str) -> bool: | |
| """Checks if a string is a validly formatted email address.""" | |
| pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" | |
| return re.match(pattern, email) is not None | |
| def log_to_csv(email, user_msg, bot_msg): | |
| """Saves the conversation turn to a CSV file.""" | |
| timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| new_row = pd.DataFrame([[timestamp, email, user_msg, bot_msg]], columns=["Time", "Email", "User", "Bot"]) | |
| try: | |
| header = not os.path.exists(LOG_FILE) | |
| new_row.to_csv(LOG_FILE, mode='a', header=header, index=False) | |
| except Exception as e: | |
| print(f"Error logging to CSV: {e}") | |
| # ============================================================================== | |
| # 4. CORE CHATBOT LOGIC | |
| # ============================================================================== | |
| def chat_fn(message: str, history: list, user_email_state: str): | |
| """Main function to handle the chatbot's logic and conversation flow.""" | |
| # --- Step 1: Handle initial state and email collection --- | |
| if user_email_state is None: | |
| if is_valid_email(message.strip()): | |
| user_email_state = message.strip() | |
| response = f"✅ Thank you! Your email `{user_email_state}` has been saved. How can I help you today?" | |
| else: | |
| response = "❌ That doesn't look like a valid email. Please enter a valid email address to continue." | |
| history.append((message, response)) | |
| # Return empty textbox, updated history, and the new email state | |
| return "", history, user_email_state | |
| # --- Step 2: Generate a response using the AI model --- | |
| # Construct a conversation history string for the model | |
| # Use only the last 2 turns to keep the context concise | |
| history_context = history[-2:] | |
| conversation_text = "" | |
| for user_turn, bot_turn in history_context: | |
| if user_turn: | |
| conversation_text += f"User: {user_turn}\n" | |
| if bot_turn: | |
| conversation_text += f"Bot: {bot_turn}\n" | |
| prompt = f"{conversation_text}User: {message}\nBot:" | |
| # Generate response | |
| generated = generator(prompt, max_length=150, num_return_sequences=1, pad_token_id=generator.tokenizer.eos_token_id) | |
| raw_response = generated[0]['generated_text'] | |
| # Extract only the last part of the response from the bot | |
| bot_response = raw_response.split("Bot:")[-1].strip() | |
| # Add a fallback if the model generates an empty response | |
| if not bot_response: | |
| bot_response = "I'm sorry, I'm not sure how to respond to that. Could you please rephrase?" | |
| # --- Step 3: Sanitize links and log the conversation --- | |
| sanitized_response = sanitize_response_links(bot_response, message) | |
| log_to_csv(user_email_state, message, sanitized_response) | |
| history.append((message, sanitized_response)) | |
| return "", history, user_email_state | |
| def welcome_message(history=None): | |
| """Initializes the chat with a welcome message.""" | |
| if history is None: | |
| history = [] | |
| history.append((None, " Welcome to the ACC Virtual Assistant! Please enter your email to start our chat.")) | |
| return history | |
| # ============================================================================== | |
| # 5. GRADIO USER INTERFACE | |
| # ============================================================================== | |
| with gr.Blocks( | |
| css=""" | |
| body, .gradio-container { background-color: #ffffff !important; color: #000000 !important; } | |
| #header { background-color: rgb(20, 90, 143); color: white; padding: 20px; border-radius: 12px; text-align: center; } | |
| #chatbot { height: 500px; } | |
| #footer { text-align: center; font-size: 14px; color: gray; margin-top: 15px; } | |
| """, | |
| title="IlmBot - ACC Assistant" | |
| ) as demo: | |
| # State object to store the user's email for the duration of the session | |
| user_email = gr.State(None) | |
| gr.Markdown( | |
| """ | |
| <div id="header"> | |
| <h1>📚 IlmBot — Your Knowledge Companion</h1> | |
| <p><b>A Continuous Charity (ACC)</b> Virtual Assistant</p> | |
| </div> | |
| """ | |
| ) | |
| chatbot = gr.Chatbot(label="💬 Conversation", elem_id="chatbot") | |
| msg_box = gr.Textbox(placeholder="Enter your email first, then your message...", label="✍️ Your Message") | |
| with gr.Row(): | |
| send_btn = gr.Button("Send") | |
| clear_btn = gr.Button("Clear Chat") | |
| gr.Markdown( | |
| """ | |
| <div id="footer"> | |
| Powered by <b>ACC</b> • Learn more at <a href="https://acceducate.org/" target="_blank">acceducate.org</a> | |
| </div> | |
| """ | |
| ) | |
| # --- Event Handlers --- | |
| def user_submit(message, history, email_state): | |
| if not message.strip(): | |
| return "", history, email_state | |
| return chat_fn(message, history, email_state) | |
| def clear_chat(): | |
| """Resets the chat and clears the user's email state.""" | |
| new_history = welcome_message([]) | |
| return "", new_history, None # Clear textbox, history, and email state | |
| # Load the initial welcome message when the app starts | |
| demo.load(welcome_message, inputs=None, outputs=[chatbot]) | |
| # Connect UI components to functions | |
| msg_box.submit(user_submit, [msg_box, chatbot, user_email], [msg_box, chatbot, user_email]) | |
| send_btn.click(user_submit, [msg_box, chatbot, user_email], [msg_box, chatbot, user_email]) | |
| clear_btn.click(clear_chat, None, [msg_box, chatbot, user_email]) | |
| # Launch the Gradio app | |
| demo.launch(debug=True,share=True) | |