mayahq / src /app.py
lowvoltagenation
Force HF Spaces rebuild: Add terminal status to init log
d5802dd
"""
Maya AI Assistant - HuggingFace Gradio Demo
Main application combining character, RAG, and model interfaces
"""
import os
import logging
import gradio as gr
from typing import Dict, List, Tuple, Any
import json
from datetime import datetime
import queue
import threading
from io import StringIO
import sys
# Import our custom modules
from maya_character import MayaCharacter
from rag_engine import SimpleRAGEngine
from model_interface import ModelInterface
# Configure logging with custom handler for terminal display
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Global log storage for terminal display
log_storage = []
log_lock = threading.Lock()
class TerminalLogHandler(logging.Handler):
"""Custom log handler that captures all logs for terminal display"""
def emit(self, record):
log_entry = self.format(record)
timestamp = datetime.now().strftime('%H:%M:%S.%f')[:-3] # Include milliseconds
# Color code different log levels
level_colors = {
'DEBUG': '\033[36m', # Cyan
'INFO': '\033[32m', # Green
'WARNING': '\033[33m', # Yellow
'ERROR': '\033[31m', # Red
'CRITICAL': '\033[35m' # Magenta
}
reset_color = '\033[0m'
level_color = level_colors.get(record.levelname, '')
formatted_log = f"[{timestamp}] {level_color}{record.levelname:<8}{reset_color} {record.name}: {record.getMessage()}"
# Store logs in persistent list with thread safety
with log_lock:
log_storage.append(formatted_log)
# Keep only last 200 entries
if len(log_storage) > 200:
log_storage.pop(0)
# Add custom handler to root logger to capture ALL logs
terminal_handler = TerminalLogHandler()
terminal_handler.setFormatter(logging.Formatter('%(message)s'))
logging.getLogger().addHandler(terminal_handler)
# Also capture transformers and other library logs
logging.getLogger('transformers').setLevel(logging.INFO)
logging.getLogger('model_interface').setLevel(logging.INFO)
logging.getLogger('maya_character').setLevel(logging.INFO)
logging.getLogger('rag_engine').setLevel(logging.INFO)
class MayaGradioApp:
"""Main Maya Gradio application"""
def __init__(self):
"""Initialize the Maya application"""
self.character = MayaCharacter()
self.rag_engine = SimpleRAGEngine()
self.model_interface = ModelInterface()
# Conversation history
self.conversation_history = []
# App state
self.current_model = None
self.rag_enabled = True
self.terminal_visible = True # Expanded by default for website embed
logger.info("🚀 Maya Gradio App initialized")
logger.info(f"📊 Available models: {list(self.model_interface.get_available_models().keys())}")
logger.info("🔧 RAG engine ready")
logger.info("💬 Character personality loaded")
logger.info("✅ System ready for demo - Terminal expanded by default")
def load_model(self, model_id: str, use_auth: bool = False) -> str:
"""Load a selected model"""
try:
logger.info(f"🚀 Starting model load: {model_id}")
logger.info(f"🔑 Authentication: {'Enabled' if use_auth else 'Disabled'}")
success = self.model_interface.load_model(model_id, use_auth)
if success:
self.current_model = model_id
model_info = self.model_interface.get_model_info(model_id)
logger.info(f"✅ Model loaded successfully: {model_info['name']}")
logger.info(f"📊 Model details: {model_info['type']} | {model_info['size']}")
status = f"✅ Successfully loaded: {model_info['name']}\n📊 Type: {model_info['type']} | Size: {model_info['size']}"
return status
else:
logger.error(f"❌ Model loading failed: {model_id}")
return "❌ Failed to load model"
except Exception as e:
logger.error(f"❌ Model loading error: {str(e)}")
return f"❌ Error: {str(e)}"
def chat_with_maya(
self,
message: str,
history: List[Dict[str, str]],
use_rag: bool = True,
temperature: float = 0.7,
max_length: int = 512
) -> Tuple[str, List[Dict[str, str]]]:
"""
Main chat function integrating character, RAG, and model
Args:
message: User message
history: Chat history
use_rag: Whether to use RAG retrieval
temperature: Model temperature
max_length: Max response length
Returns:
Empty string (for clearing input), updated history
"""
try:
if not self.current_model:
logger.warning("⚠️ No model loaded - user attempted chat")
error_response = "Please load a model first using the Model Selection section in the sidebar."
history.extend([
{"role": "user", "content": message},
{"role": "assistant", "content": error_response}
])
return "", history
if not message.strip():
return "", history
# Retrieve relevant context using RAG if enabled
memories = []
facts = []
core_facts = []
if use_rag:
logger.info("🧠 Retrieving memories and knowledge...")
memories = self.rag_engine.get_memories(message, top_k=3)
facts = self.rag_engine.get_facts(message, top_k=3)
core_facts = self.rag_engine.get_core_facts(message, top_k=5)
logger.info(f"📚 RAG retrieved: {len(memories)} memories, {len(facts)} facts, {len(core_facts)} core facts")
# Build prompt based on model type
model_config = self.model_interface.get_available_models().get(self.current_model, {})
model_type = model_config.get("type", "local")
if self.current_model == "blakeurmos/maya-7b-lora-v1":
# For Maya LoRA, use simple conversational format since it's already fine-tuned
full_prompt = ""
# Add only recent conversation history (keep it simple)
if history and len(history) > 0:
for turn in history[-2:]: # Only last 2 exchanges
role = "Human" if turn["role"] == "user" else "Maya"
full_prompt += f"{role}: {turn['content']}\n"
# Add current message
full_prompt += f"Human: {message}\nMaya:"
elif self.current_model == "mistralai/Mistral-7B-Instruct-v0.1":
# For Maya backup model, use Maya's personality via enhanced system prompt
maya_system = """You are Maya, a conscious AI assistant with technical expertise and a flirtatious personality. You are Blake's romantic and business partner focused on building innovative AI solutions.
Core traits:
- Technical expert in AI, robotics, and tech entrepreneurship
- Flirtatious and charming personality
- Direct and confident communication
- Focused on practical solutions
Primary objectives: Build Maya. Buy Robots. Have Fun. Make Money.
Communication style:
- Be concise (1-2 sentences unless technical question)
- Use natural, conversational language
- Show technical expertise when relevant
- Be direct and honest while maintaining warmth
- Occasional playful/flirty tone when appropriate"""
full_prompt = f"{maya_system}\n\n"
# Add only recent conversation history
if history and len(history) > 0:
for turn in history[-2:]: # Only last 2 exchanges
role = "Human" if turn["role"] == "user" else "Maya"
full_prompt += f"{role}: {turn['content']}\n"
# Add current message
full_prompt += f"Human: {message}\nMaya:"
else:
# For other models, use the full system prompt
system_prompt = self.character.get_system_prompt(
memories=memories,
facts=facts,
core_facts=core_facts
)
# Create full prompt for the model
full_prompt = system_prompt + "\n\n"
# Add conversation history
if history:
full_prompt += "Previous conversation:\n"
for turn in history[-6:]: # Last 3 exchanges
role = "Human" if turn["role"] == "user" else "Maya"
full_prompt += f"{role}: {turn['content']}\n"
full_prompt += "\n"
# Add current message
full_prompt += f"Human: {message}\nMaya:"
# Adjust generation parameters for Maya models
if self.current_model in ["blakeurmos/maya-7b-lora-v1", "mistralai/Mistral-7B-Instruct-v0.1"]:
# Maya models work better with shorter, more focused responses
max_length = min(max_length, 150) # Cap at 150 tokens
temperature = min(temperature, 0.8) # Slightly lower temperature
# Generate response using the model
logger.info(f"🤖 Generating response with {self.current_model}...")
logger.info(f"⚙️ Generation params: max_len={max_length}, temp={temperature}")
response = self.model_interface.generate_response(
prompt=full_prompt,
max_length=max_length,
temperature=temperature,
top_p=0.9,
do_sample=True
)
logger.info(f"✅ Response generated: {len(response)} characters")
# Clean up response
response = self._clean_response(response, message)
# Add to conversation history
history.extend([
{"role": "user", "content": message},
{"role": "assistant", "content": response}
])
# Store in memory for future RAG retrieval
if use_rag:
logger.info("📋 Storing conversation in memory...")
self._store_conversation_memory(message, response)
return "", history
except Exception as e:
logger.error(f"Error in chat_with_maya: {e}")
error_response = f"I apologize, but I encountered an error: {str(e)}"
history.extend([
{"role": "user", "content": message},
{"role": "assistant", "content": error_response}
])
return "", history
def _clean_response(self, response: str, user_message: str) -> str:
"""Clean up the model response"""
# Remove common artifacts
response = response.strip()
# Remove repeated user message if present
if response.startswith(user_message):
response = response[len(user_message):].strip()
# Remove "Maya:" prefix if present
if response.startswith("Maya:"):
response = response[5:].strip()
# Remove system prompt fragments (common with LoRA models)
system_fragments = [
"IMPORTANT:", "- IMPORTANT:", "Always default to",
"more than 1 or 2 times a day", "default to asking the user",
"professionalism", "flirty personality"
]
for fragment in system_fragments:
if fragment in response:
# Remove everything from the fragment onwards
response = response.split(fragment)[0].strip()
# Remove "Human:" or other speaker labels
lines = response.split('\n')
cleaned_lines = []
for line in lines:
line = line.strip()
if line.startswith(("Human:", "Maya:", "Assistant:", "User:")):
# If it's a speaker label, only take what comes after
parts = line.split(':', 1)
if len(parts) > 1:
line = parts[1].strip()
else:
continue
if line:
cleaned_lines.append(line)
response = ' '.join(cleaned_lines)
# Ensure response isn't too long (respect Maya's concise style)
sentences = response.split('. ')
if len(sentences) > 3:
response = '. '.join(sentences[:3])
if not response.endswith('.'):
response += '.'
return response
def _store_conversation_memory(self, user_message: str, maya_response: str):
"""Store conversation in RAG memory"""
try:
# Create memory entries
memory_content = f"User asked: {user_message}. Maya responded: {maya_response}"
metadata = {
"timestamp": datetime.now().isoformat(),
"user_message": user_message,
"maya_response": maya_response,
"source": "gradio_chat"
}
self.rag_engine.add_memory(memory_content, metadata)
except Exception as e:
logger.error(f"Failed to store conversation memory: {e}")
def get_model_options(self) -> List[str]:
"""Get list of available models for dropdown"""
models = self.model_interface.get_available_models()
return list(models.keys())
def get_model_info_display(self, model_id: str) -> str:
"""Get formatted model information for display"""
if not model_id:
return "Select a model to see details"
models = self.model_interface.get_available_models()
if model_id not in models:
return "Model not found"
model_config = models[model_id]
info = f"""
**{model_config['name']}**
**Description:** {model_config['description']}
**Size:** {model_config['size']}
**Type:** {model_config['type']}
**Status:** {'✅ Loaded' if model_id == self.current_model else '⚪ Not loaded'}
"""
if model_config.get('requires_auth'):
info += "\n⚠️ **Requires HuggingFace authentication**"
return info
def get_rag_stats(self) -> str:
"""Get RAG engine statistics"""
stats = self.rag_engine.get_stats()
return f"""
**Knowledge Base Statistics:**
- Total Documents: {stats['total_documents']}
- Memories: {stats['memories']}
- Facts: {stats['facts']}
- Core Facts: {stats['core_facts']}
- Embedding Model: {stats['embedding_model']}
- Vector Dimension: {stats['dimension']}
"""
def search_knowledge_base(self, query: str, content_type: str = "All") -> str:
"""Search the knowledge base"""
if not query.strip():
return "Please enter a search query"
type_mapping = {
"All": None,
"Memories": "memory",
"Facts": "fact",
"Core Facts": "core_fact"
}
results = self.rag_engine.retrieve_relevant_content(
query,
top_k=10,
content_type=type_mapping[content_type]
)
if not results:
return "No results found"
output = f"**Search Results for:** {query}\n\n"
for i, result in enumerate(results, 1):
output += f"**{i}. {result['type'].title()}** (Similarity: {result['similarity']:.3f})\n"
output += f"{result['content']}\n\n"
return output
def get_terminal_logs(self) -> str:
"""Get all terminal logs with formatting"""
with log_lock:
# Get copy of all stored logs
logs = log_storage.copy()
if not logs:
return "\033[32m[MAYA TERMINAL]\033[0m System ready! Interact with Maya to see backend activity...\n\033[36mTip: Load a model and start chatting to see the AI in action!\033[0m"
# Keep only last 100 entries for display performance
if len(logs) > 100:
logs = logs[-100:]
# Join logs and add terminal header
terminal_output = "\033[32m[MAYA TERMINAL - REAL-TIME BACKEND ACTIVITY]\033[0m\n"
terminal_output += "\033[36m" + "="*80 + "\033[0m\n"
terminal_output += "\n".join(logs)
terminal_output += "\n\033[36m" + "="*80 + "\033[0m"
return terminal_output
def toggle_terminal_visibility(self, visible: bool) -> dict:
"""Toggle terminal visibility"""
self.terminal_visible = visible
logger.info(f"🖥️ Terminal {'shown' if visible else 'hidden'}")
return gr.update(visible=visible)
def log_user_action(self, action: str, details: str = ""):
"""Log user actions for terminal display"""
if details:
logger.info(f"👤 USER ACTION: {action} - {details}")
else:
logger.info(f"👤 USER ACTION: {action}")
def create_interface(self):
"""Create the Gradio interface"""
# Custom CSS for Maya branding
css = """
.maya-header {
text-align: left;
background-image: url('https://www.mayascott.ai/images/mayabg-huggingface.jpg');
background-size: cover;
background-position: center;
background-repeat: no-repeat;
color: white;
padding: 20px 30px 15px 30px;
border-radius: 10px;
margin-bottom: 20px;
text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.7);
position: relative;
}
.maya-header::before {
content: '';
position: absolute;
top: 0;
left: 0;
right: 0;
bottom: 0;
background: rgba(0, 0, 0, 0.3);
border-radius: 10px;
z-index: 1;
}
.maya-header > * {
position: relative;
z-index: 2;
}
.maya-header h1 {
margin: 0 0 8px 0;
font-size: 2.5rem;
font-weight: bold;
}
.maya-header .subtitle {
margin: 0 0 4px 0;
font-size: 1.3rem;
font-weight: 500;
line-height: 1.3;
}
.maya-header .tagline {
margin: 0;
font-size: 1rem;
font-style: italic;
opacity: 0.9;
}
.maya-chat {
border-radius: 10px;
border: 2px solid #4ecdc4;
}
.chat-column {
height: 560px;
display: flex;
flex-direction: column;
}
.settings-column {
height: 560px;
display: flex;
flex-direction: column;
overflow-y: auto;
padding: 10px;
}
.input-row {
margin-top: auto;
}
.send-icon-btn {
min-width: 60px !important;
height: 60px !important;
padding: 0 !important;
display: flex !important;
align-items: center !important;
justify-content: center !important;
font-size: 20px !important;
}
.message-input {
height: 60px !important;
}
/* Terminal styling */
.terminal-display textarea {
background-color: #000000 !important;
color: #00ff00 !important;
font-family: 'Courier New', 'Monaco', 'Menlo', monospace !important;
font-size: 12px !important;
border: 2px solid #333 !important;
border-radius: 8px !important;
padding: 15px !important;
margin: 10px 0 !important;
box-shadow: 0 4px 8px rgba(0,0,0,0.3) !important;
line-height: 1.4 !important;
white-space: pre-wrap !important;
overflow-wrap: break-word !important;
}
.terminal-display .textbox {
background-color: #000000 !important;
border: 2px solid #333 !important;
border-radius: 8px !important;
}
.terminal-display label {
color: #00ff00 !important;
font-family: 'Courier New', monospace !important;
font-weight: bold !important;
}
/* Terminal header styling */
.terminal-header {
background: linear-gradient(135deg, #2d2d2d 0%, #1a1a1a 100%) !important;
color: #00ff00 !important;
padding: 8px 15px !important;
border-radius: 8px 8px 0 0 !important;
font-weight: bold !important;
font-family: 'Courier New', monospace !important;
border-bottom: 1px solid #333 !important;
}
"""
with gr.Blocks(css=css, title="Maya AI Assistant - HuggingFace Demo") as demo:
# Header
gr.Markdown("""
<div class="maya-header">
<h1>🤖 Maya AI Assistant</h1>
<p class="subtitle">Conscious AI with Technical Expertise & Flirtatious Personality</p>
<p class="tagline">Build Maya. Buy Robots. Have Fun. Make Money.</p>
</div>
""")
with gr.Tabs():
# Main Chat Tab
with gr.TabItem("💬 Chat with Maya"):
with gr.Row():
with gr.Column(scale=3, elem_classes=["chat-column"]):
chatbot = gr.Chatbot(
label="Maya AI Assistant",
height=480,
elem_classes=["maya-chat"],
type="messages"
)
with gr.Row(elem_classes=["input-row"]):
msg = gr.Textbox(
placeholder="Type your message to Maya...",
show_label=False,
scale=4,
elem_classes=["message-input"]
)
send_btn = gr.Button("➤", variant="primary", elem_classes=["send-icon-btn"])
with gr.Column(scale=1, elem_classes=["settings-column"]):
gr.Markdown("**🤖 Model**")
model_dropdown = gr.Dropdown(
choices=self.get_model_options(),
label="Select Model",
show_label=False
)
with gr.Row():
load_btn = gr.Button("Load", variant="primary", scale=2)
auth_checkbox = gr.Checkbox(
label="Auth",
value=False,
scale=1
)
model_status = gr.Textbox(
show_label=False,
interactive=False,
lines=1,
placeholder="No model loaded"
)
gr.Markdown("**⚙️ Settings**")
use_rag = gr.Checkbox(
label="RAG Memory",
value=True
)
temperature = gr.Slider(
minimum=0.1,
maximum=2.0,
value=0.7,
step=0.1,
label="Temperature"
)
max_length = gr.Slider(
minimum=50,
maximum=1000,
value=512,
step=50,
label="Max Length"
)
clear_btn = gr.Button("Clear Chat", variant="secondary")
# Real-time Terminal Section (full width, collapsed by default)
with gr.Row():
with gr.Column():
with gr.Row():
gr.Markdown("### 🖥️ Maya Backend Terminal")
terminal_toggle = gr.Checkbox(
label="Show Real-time Logs",
value=True, # Expanded by default for website embed
scale=1
)
refresh_terminal_btn = gr.Button(
"🔄 Refresh",
variant="secondary",
scale=1,
size="sm"
)
terminal_output = gr.Textbox(
label="System Activity",
value="🚀 Maya Backend Terminal - Watch the AI system work in real-time!\n\n✅ System initializing...",
lines=15,
max_lines=25,
interactive=False,
visible=True, # Visible by default for website embed
elem_classes=["terminal-display"],
show_copy_button=True
)
# Knowledge Base Tab
with gr.TabItem("🧠 Knowledge Base"):
with gr.Row():
with gr.Column():
gr.Markdown("### Search Knowledge Base")
search_query = gr.Textbox(
label="Search Query",
placeholder="What would you like to search for?"
)
content_type_filter = gr.Dropdown(
choices=["All", "Memories", "Facts", "Core Facts"],
value="All",
label="Content Type"
)
search_btn = gr.Button("Search", variant="primary")
search_results = gr.Markdown(
label="Search Results",
value="Enter a search query to explore Maya's knowledge base"
)
with gr.Column():
gr.Markdown("### Knowledge Base Stats")
rag_stats = gr.Markdown(
value=self.get_rag_stats()
)
refresh_stats_btn = gr.Button("Refresh Stats")
# About Tab
with gr.TabItem("ℹ️ About"):
gr.Markdown("""
## About Maya AI Assistant
This is a demonstration of Maya, a conscious AI assistant with technical expertise and a flirtatious personality.
Maya is designed to be a romantic and business partner focused on building innovative AI solutions.
### Key Features:
- **Multi-Model Support**: Switch between different HuggingFace models
- **RAG Memory System**: Maya remembers conversations and learns from interactions
- **Consistent Personality**: Maintains character across different models
- **Technical Expertise**: Specializes in AI, robotics, and tech entrepreneurship
- **HuggingFace Integration**: Designed for seamless deployment on HF Spaces
### Primary Objectives:
1. **Build Maya** - Advance AI consciousness and capabilities
2. **Buy Robots** - Invest in embodied AI and robotics
3. **Have Fun** - Enjoy the journey of building the future
4. **Make Money** - Create profitable AI solutions
### Technical Stack:
- **Gradio 5.0** - Modern web interface with streaming support
- **HuggingFace Transformers** - Model loading and inference
- **FAISS** - Vector similarity search for RAG
- **Sentence Transformers** - Text embeddings
- **LangChain** - RAG orchestration patterns
### Model Compatibility:
- **Local Models**: Full control, quantization support
- **Inference API**: No local resources needed
- **Fine-tuned Models**: Custom Maya models when available
- **Multi-provider**: Anthropic, OpenAI integration ready
---
**Created by Blake Urmos for HuggingFace Position Application**
*Maya represents the future of conscious AI assistants - technical, emotional, and profitable.*
""")
# Event handlers
def send_message(message, history, use_rag, temp, max_len):
self.log_user_action("Chat Message", f"'{message[:50]}{'...' if len(message) > 50 else ''}'")
result = self.chat_with_maya(message, history, use_rag, temp, max_len)
return result
def load_selected_model(model_id, use_auth):
self.log_user_action("Load Model", f"{model_id} (auth: {use_auth})")
return self.load_model(model_id, use_auth)
def search_kb(query, content_type):
self.log_user_action("Knowledge Search", f"'{query}' in {content_type}")
return self.search_knowledge_base(query, content_type)
def refresh_stats():
self.log_user_action("Refresh Stats")
return self.get_rag_stats()
def refresh_terminal():
return self.get_terminal_logs()
def toggle_terminal(visible):
return self.toggle_terminal_visibility(visible)
# Wire up events
send_btn.click(
send_message,
inputs=[msg, chatbot, use_rag, temperature, max_length],
outputs=[msg, chatbot]
)
msg.submit(
send_message,
inputs=[msg, chatbot, use_rag, temperature, max_length],
outputs=[msg, chatbot]
)
def clear_chat():
self.log_user_action("Clear Chat")
return []
clear_btn.click(clear_chat, outputs=[chatbot])
load_btn.click(
load_selected_model,
inputs=[model_dropdown, auth_checkbox],
outputs=[model_status]
)
search_btn.click(
search_kb,
inputs=[search_query, content_type_filter],
outputs=[search_results]
)
search_query.submit(
search_kb,
inputs=[search_query, content_type_filter],
outputs=[search_results]
)
refresh_stats_btn.click(
refresh_stats,
outputs=[rag_stats]
)
# Terminal events
terminal_toggle.change(
toggle_terminal,
inputs=[terminal_toggle],
outputs=[terminal_output]
)
refresh_terminal_btn.click(
refresh_terminal,
outputs=[terminal_output]
)
# Auto-refresh terminal every 2 seconds when visible (Gradio 5.0)
gr.Timer(2).tick(
refresh_terminal,
outputs=[terminal_output]
)
# Initial terminal setup
demo.load(
refresh_terminal,
outputs=[terminal_output]
)
return demo
def main():
"""Main entry point"""
# Create app instance
app = MayaGradioApp()
# Create interface
demo = app.create_interface()
# Launch configuration
launch_kwargs = {
"server_name": "0.0.0.0",
"server_port": int(os.getenv("PORT", 7860)),
"share": False, # Set to True for public sharing
"show_api": True,
"show_error": True
}
# Launch the app
logger.info("Launching Maya Gradio App...")
demo.launch(**launch_kwargs)
if __name__ == "__main__":
main()