# 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("""