# File: lcars_enhanced_interface.py import asyncio import json import os import time import uuid from typing import Dict, List, Any, Optional from dataclasses import dataclass import threading import pyttsx3 import re from pathlib import Path import gradio as gr from rich.console import Console from openai import OpenAI, AsyncOpenAI # --- Configuration --- LOCAL_BASE_URL = "http://localhost:1234/v1" LOCAL_API_KEY = "not-needed" # HuggingFace Spaces configuration HF_INFERENCE_URL = "https://api-inference.huggingface.co/models/" HF_API_KEY = os.getenv("HF_API_KEY", "") # Available model options MODEL_OPTIONS = { "Local LM Studio": LOCAL_BASE_URL, "Codellama 7B": "codellama/CodeLlama-7b-hf", "Mistral 7B": "mistralai/Mistral-7B-v0.1", "Llama 2 7B": "meta-llama/Llama-2-7b-chat-hf", "Falcon 7B": "tiiuae/falcon-7b-instruct" } DEFAULT_TEMPERATURE = 0.7 DEFAULT_MAX_TOKENS = 5000 console = Console() # --- Canvas Artifact Dataclass --- @dataclass class CanvasArtifact: id: str type: str # 'code', 'diagram', 'text', 'image' content: str title: str timestamp: float metadata: Dict[str, Any] # --- Enhanced LLMAgent with Canvas Support --- class EnhancedLLMAgent: def __init__(self, model_id: str = "local-model", system_prompt: str = None, base_url: str = LOCAL_BASE_URL, api_key: str = LOCAL_API_KEY, use_huggingface: bool = False): self.use_huggingface = use_huggingface self.model_id = model_id self.system_prompt = system_prompt or """You are an advanced AI development assistant operating in a Star Trek LCARS interface. You specialize in code generation, analysis, and collaborative development. Always provide practical, executable code solutions when appropriate. Format code responses clearly with proper markdown code blocks and explain your reasoning.""" if use_huggingface: # Use HuggingFace Inference API self.base_url = HF_INFERENCE_URL self.api_key = HF_API_KEY self.client = None console.log("[green]🚀 Using HuggingFace Inference API[/green]") else: # Use local LM Studio self.base_url = base_url self.api_key = api_key self.client = OpenAI(base_url=base_url, api_key=api_key) console.log(f"[green]🚀 Using Local LM Studio: {base_url}[/green]") # Enhanced conversation and canvas management self.conversations: Dict[str, List[Dict]] = {} self.canvas_artifacts: Dict[str, List[CanvasArtifact]] = {} self.max_history_length = 50 # Speech synthesis try: self.tts_engine = pyttsx3.init() self.setup_tts() self.speech_enabled = True console.log("[green]TTS engine initialized successfully[/green]") except Exception as e: console.log(f"[red]TTS initialization failed: {e}[/red]") self.speech_enabled = False def setup_tts(self): """Configure text-to-speech engine""" try: voices = self.tts_engine.getProperty('voices') if voices: # Try to find a better voice for voice in voices: if 'female' in voice.name.lower() or 'zira' in voice.name.lower(): self.tts_engine.setProperty('voice', voice.id) break else: self.tts_engine.setProperty('voice', voices[0].id) self.tts_engine.setProperty('rate', 180) self.tts_engine.setProperty('volume', 1.0) except Exception as e: console.log(f"[red]TTS setup error: {e}[/red]") def speak(self, text: str): """Convert text to speech in a non-blocking way""" if not hasattr(self, 'speech_enabled') or not self.speech_enabled: return def _speak(): try: # Clean text for speech clean_text = re.sub(r'```.*?```', '', text, flags=re.DOTALL) clean_text = re.sub(r'`.*?`', '', clean_text) clean_text = re.sub(r'\n+', '. ', clean_text) clean_text = re.sub(r'\s+', ' ', clean_text) clean_text = clean_text.strip() if clean_text and len(clean_text) > 10: console.log(f"[blue]Speaking: {clean_text[:100]}...[/blue]") self.tts_engine.say(clean_text[:400]) self.tts_engine.runAndWait() except Exception as e: console.log(f"[red]TTS Error: {e}[/red]") thread = threading.Thread(target=_speak, daemon=True) thread.start() async def _local_inference(self, messages: List[Dict]) -> str: """Use local LM Studio""" try: async_client = AsyncOpenAI(base_url=self.base_url, api_key=self.api_key) response = await async_client.chat.completions.create( model=self.model_id, messages=messages, temperature=0.7, max_tokens=DEFAULT_MAX_TOKENS ) return response.choices[0].message.content except Exception as e: return f"Local inference error: {str(e)}" async def _hf_inference(self, messages: List[Dict]) -> str: """Use HuggingFace Inference API""" try: import requests # Convert to HF format prompt = self._convert_messages_to_prompt(messages) headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } payload = { "inputs": prompt, "parameters": { "max_new_tokens": DEFAULT_MAX_TOKENS, "temperature": 0.7, "do_sample": True, "return_full_text": False } } model_url = f"{self.base_url}{self.model_id}" response = requests.post(model_url, headers=headers, json=payload, timeout=30) response.raise_for_status() result = response.json() return result[0]['generated_text'] except Exception as e: return f"HuggingFace API Error: {str(e)}" def _convert_messages_to_prompt(self, messages: List[Dict]) -> str: """Convert conversation messages to a single prompt for HF""" prompt = "" for msg in messages: if msg["role"] == "system": prompt += f"System: {msg['content']}\n\n" elif msg["role"] == "user": prompt += f"User: {msg['content']}\n\n" elif msg["role"] == "assistant": prompt += f"Assistant: {msg['content']}\n\n" prompt += "Assistant:" return prompt def add_artifact_to_canvas(self, conversation_id: str, content: str, artifact_type: str = "code", title: str = None): """Add artifacts to the collaborative canvas""" if conversation_id not in self.canvas_artifacts: self.canvas_artifacts[conversation_id] = [] artifact = CanvasArtifact( id=str(uuid.uuid4())[:8], type=artifact_type, content=content, title=title or f"{artifact_type}_{len(self.canvas_artifacts[conversation_id]) + 1}", timestamp=time.time(), metadata={"conversation_id": conversation_id} ) self.canvas_artifacts[conversation_id].append(artifact) console.log(f"[green]Added artifact to canvas: {artifact.title}[/green]") return artifact def get_canvas_context(self, conversation_id: str) -> str: """Get formatted canvas context for LLM prompts""" if conversation_id not in self.canvas_artifacts or not self.canvas_artifacts[conversation_id]: return "" context_lines = ["\n=== COLLABORATIVE CANVAS ARTIFACTS ==="] for artifact in self.canvas_artifacts[conversation_id][-10:]: context_lines.append(f"\n--- {artifact.title} [{artifact.type.upper()}] ---") preview = artifact.content[:500] + "..." if len(artifact.content) > 500 else artifact.content context_lines.append(preview) return "\n".join(context_lines) + "\n=================================\n" async def chat_with_canvas(self, message: str, conversation_id: str = "default", include_canvas: bool = True) -> str: """Enhanced chat that works with both local and HF""" if conversation_id not in self.conversations: self.conversations[conversation_id] = [] # Build messages with system prompt and canvas context messages = [{"role": "system", "content": self.system_prompt}] # Include canvas context if requested if include_canvas: canvas_context = self.get_canvas_context(conversation_id) if canvas_context: messages.append({"role": "system", "content": f"Current collaborative canvas state:\n{canvas_context}"}) # Add conversation history for msg in self.conversations[conversation_id][-self.max_history_length:]: messages.append(msg) # Add current message messages.append({"role": "user", "content": message}) try: if self.use_huggingface: response_text = await self._hf_inference(messages) else: response_text = await self._local_inference(messages) # Update conversation history self.conversations[conversation_id].extend([ {"role": "user", "content": message}, {"role": "assistant", "content": response_text} ]) # Auto-extract and add code artifacts to canvas self._extract_artifacts_to_canvas(response_text, conversation_id) return response_text except Exception as e: error_msg = f"Error in chat_with_canvas: {str(e)}" console.log(f"[red]{error_msg}[/red]") return error_msg def _extract_artifacts_to_canvas(self, response: str, conversation_id: str): """Automatically extract code blocks and add to canvas""" code_blocks = re.findall(r'```(?:\w+)?\n(.*?)```', response, re.DOTALL) for i, code_block in enumerate(code_blocks): if len(code_block.strip()) > 10: lang_match = re.search(r'```(\w+)\n', response) lang = lang_match.group(1) if lang_match else "unknown" self.add_artifact_to_canvas( conversation_id, code_block.strip(), "code", f"code_snippet_{lang}_{len(self.canvas_artifacts.get(conversation_id, [])) + 1}" ) def clear_conversation(self, conversation_id: str = "default"): """Clear conversation but keep canvas artifacts""" if conversation_id in self.conversations: self.conversations[conversation_id] = [] console.log(f"[yellow]Cleared conversation: {conversation_id}[/yellow]") def clear_canvas(self, conversation_id: str = "default"): """Clear canvas artifacts""" if conversation_id in self.canvas_artifacts: self.canvas_artifacts[conversation_id] = [] console.log(f"[yellow]Cleared canvas: {conversation_id}[/yellow]") def get_canvas_summary(self, conversation_id: str) -> List[Dict]: """Get summary of canvas artifacts for display""" if conversation_id not in self.canvas_artifacts: return [] artifacts = [] for artifact in reversed(self.canvas_artifacts[conversation_id]): artifacts.append({ "id": artifact.id, "type": artifact.type.upper(), "title": artifact.title, "preview": artifact.content[:100] + "..." if len(artifact.content) > 100 else artifact.content, "timestamp": time.strftime("%H:%M:%S", time.localtime(artifact.timestamp)) }) return artifacts def get_artifact_by_id(self, conversation_id: str, artifact_id: str) -> Optional[CanvasArtifact]: """Get specific artifact by ID""" if conversation_id not in self.canvas_artifacts: return None for artifact in self.canvas_artifacts[conversation_id]: if artifact.id == artifact_id: return artifact return None def update_config(self, base_url: str, api_key: str, model_id: str, temperature: float, max_tokens: int): """Update agent configuration""" self.base_url = base_url self.api_key = api_key self.model_id = model_id console.log(f"[blue]Updated config: {model_id} @ {base_url}[/blue]") @staticmethod async def fetch_available_models(base_url: str, api_key: str, use_huggingface: bool = False) -> List[str]: """Fetch available models - works for both local and HF""" if use_huggingface: # Return popular HF models return list(MODEL_OPTIONS.keys())[1:] # Skip "Local LM Studio" else: # Fetch from local LM Studio try: console.log(f"[blue]Fetching models from {base_url}[/blue]") async_client = AsyncOpenAI(base_url=base_url, api_key=api_key) models = await async_client.models.list() model_list = [model.id for model in models.data] console.log(f"[green]Found {len(model_list)} local models[/green]") return model_list except Exception as e: console.log(f"[red]Error fetching local models: {e}[/red]") return ["local-model"] # --- LCARS Styled Gradio Interface --- class LcarsInterface: def __init__(self): # Start with HuggingFace by default for Spaces self.use_huggingface = True self.agent = EnhancedLLMAgent(use_huggingface=self.use_huggingface) self.current_conversation = "default" def create_interface(self): """Create the full LCARS-styled interface""" lcars_css = """ :root { --lcars-orange: #FF9900; --lcars-red: #FF0033; --lcars-blue: #6699FF; --lcars-purple: #CC99FF; --lcars-pale-blue: #99CCFF; --lcars-black: #000000; --lcars-dark-blue: #3366CC; --lcars-gray: #424242; --lcars-yellow: #FFFF66; } body { background: var(--lcars-black); color: var(--lcars-orange); font-family: 'Antonio', 'LCD', 'Courier New', monospace; margin: 0; padding: 0; } .gradio-container { background: var(--lcars-black) !important; min-height: 100vh; } .lcars-container { background: var(--lcars-black); border: 4px solid var(--lcars-orange); border-radius: 0 30px 0 0; min-height: 100vh; padding: 20px; } .lcars-header { background: linear-gradient(90deg, var(--lcars-red), var(--lcars-orange)); padding: 20px 40px; border-radius: 0 60px 0 0; margin: -20px -20px 20px -20px; border-bottom: 6px solid var(--lcars-blue); } .lcars-title { font-size: 2.5em; font-weight: bold; color: var(--lcars-black); margin: 0; } .lcars-subtitle { font-size: 1.2em; color: var(--lcars-black); margin: 10px 0 0 0; } .lcars-panel { background: rgba(66, 66, 66, 0.9); border: 2px solid var(--lcars-orange); border-radius: 0 20px 0 20px; padding: 15px; margin-bottom: 15px; } .lcars-button { background: var(--lcars-orange); color: var(--lcars-black) !important; border: none !important; border-radius: 0 15px 0 15px !important; padding: 10px 20px !important; font-family: inherit !important; font-weight: bold !important; margin: 5px !important; } .lcars-button:hover { background: var(--lcars-red) !important; } .lcars-input { background: var(--lcars-black) !important; color: var(--lcars-orange) !important; border: 2px solid var(--lcars-blue) !important; border-radius: 0 10px 0 10px !important; padding: 10px !important; } .lcars-chatbot { background: var(--lcars-black) !important; border: 2px solid var(--lcars-purple) !important; border-radius: 0 15px 0 15px !important; } .status-indicator { display: inline-block; width: 12px; height: 12px; border-radius: 50%; background: var(--lcars-red); margin-right: 8px; } .status-online { background: var(--lcars-blue); animation: pulse 2s infinite; } @keyframes pulse { 0% { opacity: 1; } 50% { opacity: 0.5; } 100% { opacity: 1; } } """ with gr.Blocks(css=lcars_css, theme=gr.themes.Default(), title="LCARS Terminal") as interface: with gr.Column(elem_classes="lcars-container"): # Header with gr.Row(elem_classes="lcars-header"): gr.Markdown("""
🚀 LCARS TERMINAL
STARFLEET AI DEVELOPMENT CONSOLE
SYSTEM ONLINE
""") # Connection Type Selector with gr.Row(elem_classes="lcars-panel"): gr.Markdown("### 🌐 CONNECTION TYPE") connection_type = gr.Radio( choices=["HuggingFace Inference", "Local LM Studio"], value="HuggingFace Inference", label="Select Connection Type", elem_classes="lcars-input" ) # Main Content with gr.Row(): # Left Sidebar with gr.Column(scale=1): # Configuration Panel with gr.Column(elem_classes="lcars-panel"): gr.Markdown("### 🔧 CONFIGURATION") # Connection-specific settings with gr.Row(visible=False) as local_settings: base_url = gr.Textbox( value=LOCAL_BASE_URL, label="LM Studio URL", elem_classes="lcars-input" ) api_key = gr.Textbox( value=LOCAL_API_KEY, label="API Key", type="password", elem_classes="lcars-input" ) with gr.Row(visible=True) as hf_settings: hf_api_key = gr.Textbox( value=HF_API_KEY, label="HuggingFace API Key", type="password", elem_classes="lcars-input", placeholder="Get from https://huggingface.co/settings/tokens" ) with gr.Row(): model_dropdown = gr.Dropdown( choices=list(MODEL_OPTIONS.keys())[1:], value=list(MODEL_OPTIONS.keys())[1], label="AI Model", elem_classes="lcars-input" ) fetch_models_btn = gr.Button("📡 Fetch Models", elem_classes="lcars-button") with gr.Row(): temperature = gr.Slider(0.0, 2.0, value=0.7, label="Temperature") max_tokens = gr.Slider(128, 8192, value=2000, step=128, label="Max Tokens") with gr.Row(): update_config_btn = gr.Button("💾 Apply Config", elem_classes="lcars-button") speech_toggle = gr.Checkbox(value=True, label="🔊 Speech Output") # Canvas Artifacts with gr.Column(elem_classes="lcars-panel"): gr.Markdown("### 🎨 CANVAS ARTIFACTS") artifact_display = gr.JSON(label="") with gr.Row(): refresh_artifacts_btn = gr.Button("🔄 Refresh", elem_classes="lcars-button") clear_canvas_btn = gr.Button("🗑️ Clear Canvas", elem_classes="lcars-button") # Main Content Area with gr.Column(scale=2): # Code Canvas with gr.Accordion("💻 COLLABORATIVE CODE CANVAS", open=True): code_editor = gr.Code( value="# Welcome to LCARS Collaborative Canvas\n\nprint('Hello, Starfleet!')", language="python", lines=15, label="" ) with gr.Row(): load_to_chat_btn = gr.Button("💬 Discuss Code", elem_classes="lcars-button") analyze_btn = gr.Button("🔍 Analyze", elem_classes="lcars-button") optimize_btn = gr.Button("⚡ Optimize", elem_classes="lcars-button") # Chat Interface with gr.Column(elem_classes="lcars-panel"): gr.Markdown("### 💬 MISSION LOG") chatbot = gr.Chatbot(label="", height=300) with gr.Row(): message_input = gr.Textbox( placeholder="Enter your command or query...", show_label=False, lines=2, scale=4 ) send_btn = gr.Button("🚀 SEND", elem_classes="lcars-button", scale=1) # Status with gr.Row(): status_display = gr.Textbox( value="LCARS terminal operational. Awaiting commands.", label="Status", max_lines=2 ) with gr.Column(scale=0): clear_chat_btn = gr.Button("🗑️ Clear Chat", elem_classes="lcars-button") new_session_btn = gr.Button("🆕 New Session", elem_classes="lcars-button") # === EVENT HANDLERS === def switch_connection(connection_type): if connection_type == "Local LM Studio": return [ gr.update(visible=True), gr.update(visible=False), gr.update(choices=["Fetching local models..."], value="Fetching local models...") ] else: return [ gr.update(visible=False), gr.update(visible=True), gr.update(choices=list(MODEL_OPTIONS.keys())[1:], value=list(MODEL_OPTIONS.keys())[1]) ] async def fetch_models_updated(connection_type, base_url_val, api_key_val, hf_api_key_val): if connection_type == "Local LM Studio": models = await EnhancedLLMAgent.fetch_available_models( base_url_val, api_key_val, use_huggingface=False ) else: models = await EnhancedLLMAgent.fetch_available_models( "", hf_api_key_val, use_huggingface=True ) if models: return gr.update(choices=models, value=models[0]) return gr.update(choices=["No models found"]) def update_agent_connection(connection_type, model_id, base_url_val, api_key_val, hf_api_key_val): use_hf = connection_type == "HuggingFace Inference" self.use_huggingface = use_hf if use_hf: self.agent = EnhancedLLMAgent( model_id=model_id, use_huggingface=True, api_key=hf_api_key_val ) return f"✅ Switched to HuggingFace: {model_id}" else: self.agent = EnhancedLLMAgent( model_id=model_id, base_url=base_url_val, api_key=api_key_val, use_huggingface=False ) return f"✅ Switched to Local: {base_url_val}" async def process_message(message, history, speech_enabled): if not message.strip(): return "", history, "Please enter a message" history = history + [[message, None]] try: response = await self.agent.chat_with_canvas( message, self.current_conversation, include_canvas=True ) history[-1][1] = response if speech_enabled and self.agent.speech_enabled: self.agent.speak(response) artifacts = self.agent.get_canvas_summary(self.current_conversation) status = f"✅ Response received. Canvas artifacts: {len(artifacts)}" return "", history, status, artifacts except Exception as e: error_msg = f"❌ Error: {str(e)}" history[-1][1] = error_msg return "", history, error_msg, self.agent.get_canvas_summary(self.current_conversation) def get_artifacts(): return self.agent.get_canvas_summary(self.current_conversation) def clear_canvas(): self.agent.clear_canvas(self.current_conversation) return [], "✅ Canvas cleared" def clear_chat(): self.agent.clear_conversation(self.current_conversation) return [], "✅ Chat cleared" def new_session(): self.agent.clear_conversation(self.current_conversation) self.agent.clear_canvas(self.current_conversation) return [], "# New session started\n\nprint('Ready!')", "🆕 New session started", [] # Connect events connection_type.change(switch_connection, inputs=connection_type, outputs=[local_settings, hf_settings, model_dropdown]) fetch_models_btn.click(fetch_models_updated, inputs=[connection_type, base_url, api_key, hf_api_key], outputs=model_dropdown) model_dropdown.change(update_agent_connection, inputs=[connection_type, model_dropdown, base_url, api_key, hf_api_key], outputs=status_display) send_btn.click(process_message, inputs=[message_input, chatbot, speech_toggle], outputs=[message_input, chatbot, status_display, artifact_display]) message_input.submit(process_message, inputs=[message_input, chatbot, speech_toggle], outputs=[message_input, chatbot, status_display, artifact_display]) refresh_artifacts_btn.click(get_artifacts, outputs=artifact_display) clear_canvas_btn.click(clear_canvas, outputs=[artifact_display, status_display]) clear_chat_btn.click(clear_chat, outputs=[chatbot, status_display]) new_session_btn.click(new_session, outputs=[chatbot, code_editor, status_display, artifact_display]) interface.load(get_artifacts, outputs=artifact_display) return interface # --- Main Application --- def main(): console.log("[bold blue]🚀 Starting LCARS Terminal...[/bold blue]") is_space = os.getenv('SPACE_ID') is not None if is_space: console.log("[green]🌐 Detected HuggingFace Space[/green]") else: console.log("[blue]💻 Running locally[/blue]") interface = LcarsInterface() demo = interface.create_interface() demo.launch( share=is_space ) if __name__ == "__main__": main()