#!/usr/bin/env python3 """ Final Enhanced Gradio Web UI with Auto-Loading User Conversations """ import os import gradio as gr from dotenv import load_dotenv from graph import ArchitectureAssistant from user_state_manager import user_state_manager import json import hashlib from datetime import datetime class GradioArchitectureApp: def __init__(self): load_dotenv() api_key = os.getenv("OPENAI_API_KEY") if not api_key: raise ValueError("Please set OPENAI_API_KEY in .env file") self.api_key = api_key self.assistant = None self.current_user_id = None self.conversation_history = [] self.state_manager = user_state_manager def initialize_user_session(self, user_identifier: str = None): """Initialize user session with persistent state""" if not user_identifier: # Generate user ID from timestamp for demo user_identifier = hashlib.md5(str(datetime.now()).encode()).hexdigest()[:12] self.current_user_id = user_identifier self.assistant = ArchitectureAssistant(self.api_key, user_identifier) return user_identifier def change_user_id(self, new_user_id: str): """Change user ID and auto-load their conversation history""" if not new_user_id.strip(): return [], "❌ Please enter a valid User ID", "", "Start chatting to see your requirements here!" try: # Set new user ID self.current_user_id = new_user_id.strip() # Create new assistant instance for this user self.assistant = ArchitectureAssistant(self.api_key, self.current_user_id) # Try to load their latest conversation history = self.state_manager.get_user_history(self.current_user_id) if history: # Load most recent conversation latest_session = history[0]["session_id"] success = self.assistant.load_previous_state(latest_session) if success: # Convert state messages to chat history chat_history = [] messages = self.assistant.state.get("messages", []) current_pair = [None, None] for msg in messages: if msg["role"] == "user": current_pair = [msg["content"], None] elif msg["role"] == "assistant": if current_pair[0] is not None: current_pair[1] = msg["content"] chat_history.append(current_pair) current_pair = [None, None] status_msg = f"✅ Loaded {len(history)} previous conversation(s) for user {new_user_id}" summary = self.get_conversation_summary() return chat_history, status_msg, new_user_id, summary else: status_msg = f"✅ User {new_user_id} found but couldn't load conversation. Starting fresh." summary = self.get_conversation_summary() return [], status_msg, new_user_id, summary else: # New user status_msg = f"✅ New user {new_user_id} - starting fresh conversation" summary = self.get_conversation_summary() return [], status_msg, new_user_id, summary except Exception as e: return [], f"❌ Error loading user {new_user_id}: {str(e)}", "", "Error loading user" def chat_with_assistant(self, message, history, user_id_input=""): """Process user message and return response with state saving""" if not message.strip(): return history, "", user_id_input, self.get_conversation_summary() # Ensure we have an assistant instance if not self.assistant: if user_id_input.strip(): self.current_user_id = user_id_input.strip() self.assistant = ArchitectureAssistant(self.api_key, self.current_user_id) else: user_id = self.initialize_user_session() return history, "", user_id, self.get_conversation_summary() # Handle special commands if message.lower() == 'reset': self.assistant.reset_conversation() self.conversation_history = [] status_msg = "🔄 Conversation reset! New session started." return [[status_msg, ""]], "", self.current_user_id or "", self.get_conversation_summary() # Get response from assistant try: response = self.assistant.chat(message) # Add response to history history.append([message, response]) # Add to conversation history self.conversation_history.append({"user": message, "assistant": response}) # Get updated summary summary = self.get_conversation_summary() return history, "", self.current_user_id or "", summary except Exception as e: error_msg = f"❌ Error: {str(e)}" history.append([message, error_msg]) return history, "", self.current_user_id or "", self.get_conversation_summary() def get_conversation_summary(self): """Get formatted conversation summary""" if not self.assistant: return "Enter a User ID above to start or resume a conversation!" summary = self.assistant.get_conversation_summary() summary_text = f"📋 **CONVERSATION SUMMARY**\n" summary_text += f"👤 **User ID:** {self.current_user_id or 'Not set'}\n\n" # User requirements reqs = summary["user_requirements"] summary_text += "**USER REQUIREMENTS:**\n" if reqs["budget"]: summary_text += f"• Budget: ${reqs['budget']:,.0f}\n" if reqs["location"]: summary_text += f"• Location: {reqs['location']}\n" if reqs["family_size"]: summary_text += f"• Family size: {reqs['family_size']}\n" if reqs["lifestyle_preferences"]: summary_text += f"• Preferences: {', '.join(reqs['lifestyle_preferences'])}\n" # Floorplan requirements floor_reqs = summary["floorplan_requirements"] summary_text += "\n**FLOORPLAN REQUIREMENTS:**\n" if floor_reqs["num_floors"]: summary_text += f"• Floors: {floor_reqs['num_floors']}\n" if floor_reqs["total_sqft"]: summary_text += f"• Total sq ft: {floor_reqs['total_sqft']}\n" if floor_reqs["lot_shape"]: summary_text += f"• Lot shape: {floor_reqs['lot_shape']}\n" if floor_reqs["lot_dimensions"]: summary_text += f"• Lot dimensions: {floor_reqs['lot_dimensions']}\n" if floor_reqs["rooms"]: rooms_str = ", ".join([f"{r['count']}x {r['type']}" for r in floor_reqs["rooms"]]) summary_text += f"• Rooms: {rooms_str}\n" # Project progress agent_memory = self.assistant.state.get("agent_memory", {}) completed_phases = [] if self.assistant.state.get("detailed_floorplan", {}).get("detailed_rooms"): completed_phases.append("Architectural Design") if self.assistant.state.get("budget_breakdown", {}).get("total_construction_cost"): completed_phases.append("Budget Analysis") if agent_memory.get("structural_analysis"): completed_phases.append("Structural Analysis") if agent_memory.get("sustainability"): completed_phases.append("Sustainability Review") if agent_memory.get("permits"): completed_phases.append("Permit Planning") if agent_memory.get("interior_design"): completed_phases.append("Interior Design") summary_text += f"\n**PROJECT PROGRESS:**\n" summary_text += f"• Completed: {len(completed_phases)}/6 phases\n" if completed_phases: summary_text += f"• Phases: {', '.join(completed_phases)}\n" summary_text += f"• Total messages: {summary['total_messages']}\n" return summary_text def get_user_history_display(self, user_id: str): """Get formatted user history for display""" if not user_id.strip(): return "Please enter a User ID to view history." try: history = self.state_manager.get_user_history(user_id.strip()) if not history: return f"No conversation history found for user: {user_id}" display_text = f"📚 **CONVERSATION HISTORY FOR USER: {user_id}**\n\n" for i, conv in enumerate(history, 1): timestamp = conv["timestamp"] session_id = conv["session_id"] summary = conv.get("summary", {}) display_text += f"**#{i} - {timestamp}**\n" display_text += f"Session ID: `{session_id}`\n" if summary.get("total_messages", 0) > 0: display_text += f"Messages: {summary['total_messages']}\n" if summary.get("user_requirements"): reqs = summary["user_requirements"] if reqs.get("budget"): display_text += f"Budget: {reqs['budget']}\n" if reqs.get("location"): display_text += f"Location: {reqs['location']}\n" if reqs.get("family_size"): display_text += f"Family: {reqs['family_size']} people\n" if summary.get("floorplan_status"): floor_status = summary["floorplan_status"] if floor_status.get("size"): display_text += f"House Size: {floor_status['size']}\n" if floor_status.get("rooms"): display_text += f"Rooms: {floor_status['rooms']}\n" if summary.get("project_progress"): progress = summary["project_progress"] completed = progress.get("completed_phases", []) percentage = progress.get("completion_percentage", 0) display_text += f"Progress: {percentage}% - {', '.join(completed)}\n" display_text += "\n---\n\n" return display_text except Exception as e: return f"❌ Error retrieving history: {str(e)}" def get_all_users_summary(self): """Get summary of all users""" try: users = self.state_manager.get_all_users() if not users: return "No users found in the system." display_text = f"👥 **ALL USERS SUMMARY** ({len(users)} users)\n\n" for i, user in enumerate(users, 1): user_id = user["user_id"] total_conversations = user["total_conversations"] last_activity = user["last_activity"] latest_summary = user.get("latest_summary", {}) display_text += f"**#{i} User: {user_id}**\n" display_text += f"Conversations: {total_conversations}\n" display_text += f"Last Activity: {last_activity}\n" if latest_summary.get("project_progress"): progress = latest_summary["project_progress"] percentage = progress.get("completion_percentage", 0) display_text += f"Latest Progress: {percentage}%\n" display_text += "\n---\n\n" return display_text except Exception as e: return f"❌ Error retrieving users summary: {str(e)}" def create_gradio_interface(): """Create and return the final enhanced Gradio interface""" app = GradioArchitectureApp() # Custom CSS for better styling css = """ .gradio-container { max-width: 1400px !important; } .chat-message { border-radius: 10px !important; } .user-id-input { background-color: #e3f2fd !important; font-weight: bold !important; } """ with gr.Blocks( title="🏠 Architecture Assistant - Multi-User", theme=gr.themes.Soft(), css=css ) as interface: gr.HTML("""
Multi-user system with automatic conversation loading and precise floorplan specifications
🧭 Router: Intelligent conversation routing
🏛️ General Design: Architecture principles & design guidance
💰 Budget Analysis: Montreal market costs & feasibility
📐 Floorplan: Layout planning with exact dimensions
📋 Regulation: Montreal building codes & permit requirements