import gradio as gr import logging import os from huggingface_hub import InferenceClient from datetime import datetime import uuid import json import time # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("chatbot_logs.log"), logging.StreamHandler() ] ) logger = logging.getLogger("CompanyChatbot") # Environment variables (for production use) HF_MODEL = os.environ.get("HF_MODEL", "HuggingFaceH4/zephyr-7b-beta") HF_API_TOKEN = os.environ.get("HF_API_TOKEN", None) # Set your API token as env variable COMPANY_NAME = os.environ.get("COMPANY_NAME", "Your Company") DEFAULT_SYSTEM_PROMPT = os.environ.get("DEFAULT_SYSTEM_PROMPT", f"You are {COMPANY_NAME}'s professional AI assistant. Be helpful, accurate, and concise.") # Initialize the client try: client = InferenceClient(HF_MODEL, token=HF_API_TOKEN) logger.info(f"Successfully initialized InferenceClient with model: {HF_MODEL}") except Exception as e: logger.error(f"Failed to initialize InferenceClient: {str(e)}") raise RuntimeError(f"Failed to initialize the model. Please check your configuration: {str(e)}") # Conversation tracking def save_conversation(user_id, conversation): filename = f"conversations/{user_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" os.makedirs(os.path.dirname(filename), exist_ok=True) with open(filename, 'w') as f: json.dump(conversation, f) logger.info(f"Saved conversation for user {user_id}") # Main chat function - fixed to properly handle message format def respond( message, history, system_message, max_tokens, temperature, top_p, user_id ): if not message.strip(): return history + [[message, "I'm sorry, I didn't receive any input. How can I help you today?"]] # Log the incoming request logger.info(f"User {user_id} sent message - Length: {len(message)}") try: # Prepare messages for the model messages = [{"role": "system", "content": system_message}] # Format history correctly for user_msg, assistant_msg in history: if user_msg: messages.append({"role": "user", "content": user_msg}) if assistant_msg: messages.append({"role": "assistant", "content": assistant_msg}) messages.append({"role": "user", "content": message}) # Generate response full_response = "" start_time = datetime.now() # Stream the response for message_chunk in client.chat_completion( messages, max_tokens=max_tokens, stream=True, temperature=temperature, top_p=top_p, ): token = message_chunk.choices[0].delta.content if token: full_response += token yield history + [[message, full_response]] # Log completion time_taken = (datetime.now() - start_time).total_seconds() logger.info(f"Response generated for user {user_id} in {time_taken:.2f}s - Length: {len(full_response)}") # Save conversation for audit/analytics conversation_data = { "timestamp": datetime.now().isoformat(), "user_id": user_id, "messages": messages, "response": full_response, "parameters": { "max_tokens": max_tokens, "temperature": temperature, "top_p": top_p }, "time_taken": time_taken } save_conversation(user_id, conversation_data) # Return final history with the complete response # This is necessary if streaming didn't yield anything if full_response: return history + [[message, full_response]] return history except Exception as e: error_msg = f"An error occurred: {str(e)}" logger.error(f"Error generating response for user {user_id}: {str(e)}") return history + [[message, error_msg]] # Authentication function (replace with your actual auth system) def authenticate(username, password): # In production, this should check against your company's auth system valid_credentials = { "admin": {"password": "admin123", "role": "admin"}, "user": {"password": "user123", "role": "user"} } # Example only if username in valid_credentials and valid_credentials[username]["password"] == password: return True, str(uuid.uuid4()), valid_credentials[username]["role"] # Return success, user_id, and role return False, None, None # Login interface with error handling def login(username, password): if not username or not password: return ( gr.update(visible=True), gr.update(visible=False), None, None, gr.update(visible=True, value="Please enter both username and password") ) # Simulate processing delay for security (prevents timing attacks) time.sleep(0.5) success, user_id, role = authenticate(username, password) if success: return ( gr.update(visible=False), gr.update(visible=True), user_id, role, gr.update(visible=False) ) else: return ( gr.update(visible=True), gr.update(visible=False), None, None, gr.update(visible=True, value="Invalid username or password") ) # CSS for better styling css = """ body { font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; background-color: #f9f9f9; } .container { max-width: 1400px !important; margin: auto; } .setting-panel { background-color: #f0f4f8; border-radius: 10px; padding: 15px; box-shadow: 0 2px 6px rgba(0,0,0,0.1); } .chat-container { border-radius: 10px; box-shadow: 0 2px 6px rgba(0,0,0,0.1); background-color: white; } .company-header { background-color: #2c3e50; color: white; padding: 15px; border-radius: 10px 10px 0 0; margin-bottom: 15px; } .footer { text-align: center; margin-top: 20px; color: #666; font-size: 0.8em; } .message-user { background-color: #e6f7ff !important; border-radius: 15px 15px 0 15px !important; } .message-bot { background-color: #f0f0f0 !important; border-radius: 15px 15px 15px 0 !important; } .login-container { max-width: 500px; margin: 50px auto; padding: 30px; background-color: white; border-radius: 10px; box-shadow: 0 4px 10px rgba(0,0,0,0.1); } .login-header { text-align: center; margin-bottom: 30px; } .error-message { color: #e74c3c; background-color: #fdedeb; padding: 10px; border-radius: 5px; margin-bottom: 15px; font-size: 14px; } .role-badge { font-size: 12px; padding: 3px 8px; border-radius: 10px; margin-left: 10px; } .admin-badge { background-color: #e74c3c; color: white; } .user-badge { background-color: #3498db; color: white; } .setting-disabled { opacity: 0.5; pointer-events: none; } """ # Main application with gr.Blocks(css=css, title=f"{COMPANY_NAME} AI Assistant") as demo: user_id = gr.State(None) user_role = gr.State(None) with gr.Row(): gr.Markdown(f"
Please log in to continue