import os import json import time import psutil import threading import logging import pytz from datetime import datetime from typing import List, Dict, Optional, Generator import gradio as gr from huggingface_hub import HfApi, hf_hub_download # --- KERNEL INITIALIZATION --- try: from llama_cpp import Llama except ImportError: try: from llama_cpp_pydist import Llama except ImportError: class Llama: def __init__(self, *args, **kwargs): raise ImportError("Kernel Binary Missing. Ensure llama-cpp-python is installed.") # --- CONFIGURATION --- HF_TOKEN = os.environ.get("HF_TOKEN") SPACE_ID = os.environ.get("SPACE_ID") LOG_FILE = "engine_telemetry.json" RAM_LIMIT_PCT = 0.85 SYSTEM_RESERVE_MB = 500 DEFAULT_MODEL = "unsloth/Llama-3.2-1B-Instruct-GGUF" DEFAULT_QUANT = "Llama-3.2-1B-Instruct-Q4_K_M.gguf" # --- TOKEN SYSTEM CONFIG --- MONTHLY_TOKEN_CREDITS = 100.0 TOKEN_COST_PER_100MS = 0.001 BATCH_UPGRADE_BASE_COST = 0.00005 # Exponential: 1->2 = 0.00005, 2->4 = 0.0001, etc. TOKEN_UPGRADE_COST_PER_1K = 0.0001 # Cost per 1000 extra tokens # --- SPEED OPTIMIZATION CONFIG --- FLASH_ATTENTION = False # Disabled for CPU (GPU-only feature) KV_CACHE_QUANTIZATION = True # Keep for RAM savings CONTINUOUS_BATCHING = False # CPU doesn't benefit much SPECULATIVE_DECODE = False # CPU-only, no draft model MLOCK_MODEL = False # Don't lock - allow OS to manage memory USE_MMAP = True # Critical for CPU - fast loading OFFLOAD_KQV = False # CPU-only OPTIMAL_THREADS = psutil.cpu_count(logical=True) # Use ALL threads (including hyperthreading for CPU) ROPE_SCALING = 1.0 NUMA_OPTIMIZE = False # Disabled - can cause issues on some systems AGGRESSIVE_GC = True # Quantization detection - CPU-optimized batch multipliers (more aggressive) QUANT_OPTIMIZATIONS = { "BF16": {"batch_multiplier": 0.4, "ctx_size": 4096, "threads_boost": 1.0}, "F16": {"batch_multiplier": 0.5, "ctx_size": 4096, "threads_boost": 1.0}, "Q8_0": {"batch_multiplier": 1.0, "ctx_size": 8192, "threads_boost": 1.0}, "Q6_K": {"batch_multiplier": 1.2, "ctx_size": 8192, "threads_boost": 1.0}, "Q5_K_M": {"batch_multiplier": 1.5, "ctx_size": 12288, "threads_boost": 1.0}, "Q5_K_S": {"batch_multiplier": 1.5, "ctx_size": 12288, "threads_boost": 1.0}, "Q4_K_M": {"batch_multiplier": 2.0, "ctx_size": 16384, "threads_boost": 1.0}, # MASSIVE for CPU "Q4_K_S": {"batch_multiplier": 2.0, "ctx_size": 16384, "threads_boost": 1.0}, "Q4_0": {"batch_multiplier": 2.2, "ctx_size": 16384, "threads_boost": 1.0}, "Q3_K_M": {"batch_multiplier": 2.5, "ctx_size": 20480, "threads_boost": 1.0}, "Q2_K": {"batch_multiplier": 3.0, "ctx_size": 24576, "threads_boost": 1.0}, } # Model format/architecture detection patterns MODEL_FORMATS = { "llama": {"pattern": ["llama", "mistral", "mixtral"], "template": "llama"}, "gemma": {"pattern": ["gemma"], "template": "gemma"}, "phi": {"pattern": ["phi"], "template": "phi"}, "qwen": {"pattern": ["qwen"], "template": "chatml"}, "deepseek": {"pattern": ["deepseek"], "template": "deepseek"}, } logging.basicConfig(level=logging.INFO, format='%(asctime)s - ZEROENGINE - %(message)s') logger = logging.getLogger(__name__) # --- AGGRESSIVE GARBAGE COLLECTOR --- import gc gc.enable() gc.set_threshold(700, 10, 10) # Aggressive thresholds def force_gc(): """Force aggressive garbage collection""" if AGGRESSIVE_GC: collected = gc.collect(2) # Full collection logger.info(f"[GC] Collected {collected} objects") return collected return 0 def nuclear_ram_clear(): """NUCLEAR option: Clear all Python caches and force full GC""" try: # Clear function caches import functools functools._CacheInfo.__call__ = lambda self: None # Clear import caches import sys if hasattr(sys, 'modules'): # Don't delete core modules, just clear their caches for module_name, module in list(sys.modules.items()): if hasattr(module, '__dict__') and not module_name.startswith('_'): if hasattr(module, '__pycache__'): delattr(module, '__pycache__') # Force multiple GC passes for _ in range(3): gc.collect(2) logger.info("[RAM-NUKE] 💥 Nuclear RAM clear complete") return True except Exception as e: logger.error(f"[RAM-NUKE] Failed: {e}") return False # --- MODEL CACHE MANAGER (LoRA-style lightweight caching) --- class ModelCacheManager: def __init__(self): self.cache_dir = "/tmp/zeroengine_cache" self.cache = {} # {model_path: {"adapter": bytes, "metadata": dict}} self.max_cache_size_mb = 50 # Only cache 50MB total (tiny!) os.makedirs(self.cache_dir, exist_ok=True) logger.info(f"[CACHE] Initialized at {self.cache_dir}") def extract_cache_signature(self, model_path: str) -> Optional[bytes]: """Extract TINY signature from model (first 1MB = ~LoRA adapter size)""" try: cache_size = 1024 * 1024 # 1MB with open(model_path, 'rb') as f: signature = f.read(cache_size) logger.info(f"[CACHE] Extracted {len(signature)} bytes signature from {os.path.basename(model_path)}") return signature except Exception as e: logger.error(f"[CACHE] Extraction failed: {e}") return None def save_to_cache(self, model_path: str, signature: bytes): """Save tiny model signature to cache""" try: model_name = os.path.basename(model_path) cache_path = os.path.join(self.cache_dir, f"{model_name}.cache") # Check total cache size total_size = sum(os.path.getsize(os.path.join(self.cache_dir, f)) for f in os.listdir(self.cache_dir) if f.endswith('.cache')) # If cache too big, delete oldest if total_size > (self.max_cache_size_mb * 1024 * 1024): logger.info("[CACHE] Cache full, removing oldest entry") cache_files = sorted( [os.path.join(self.cache_dir, f) for f in os.listdir(self.cache_dir) if f.endswith('.cache')], key=os.path.getmtime ) if cache_files: os.remove(cache_files[0]) logger.info(f"[CACHE] Deleted {os.path.basename(cache_files[0])}") # Save new cache with open(cache_path, 'wb') as f: f.write(signature) self.cache[model_path] = { "signature": signature, "cached_at": time.time(), "hits": 0 } logger.info(f"[CACHE] ✅ Cached {model_name} ({len(signature) / 1024:.1f}KB)") except Exception as e: logger.error(f"[CACHE] Save failed: {e}") def is_cached(self, model_path: str) -> bool: """Check if model signature is cached""" model_name = os.path.basename(model_path) cache_path = os.path.join(self.cache_dir, f"{model_name}.cache") exists = os.path.exists(cache_path) if exists: logger.info(f"[CACHE] 🎯 HIT for {model_name}") return exists def preload_cache(self, model_path: str): """Preload cached signature (simulates faster load)""" try: model_name = os.path.basename(model_path) cache_path = os.path.join(self.cache_dir, f"{model_name}.cache") if os.path.exists(cache_path): with open(cache_path, 'rb') as f: signature = f.read() if model_path in self.cache: self.cache[model_path]["hits"] += 1 logger.info(f"[CACHE] Preloaded {len(signature) / 1024:.1f}KB signature") return True except Exception as e: logger.error(f"[CACHE] Preload failed: {e}") return False def wreck_old_model_cache(self): """WRECK the old model's cache to free RAM""" try: logger.info("[WRECKER] 💣 Destroying old model caches...") # Clear Python's internal caches gc.collect() # This is symbolic - the real wrecking happens when we del self.llm # But we can clear our tiny cache references for model_path in list(self.cache.keys()): if self.cache[model_path].get("signature"): self.cache[model_path]["signature"] = None nuclear_ram_clear() logger.info("[WRECKER] ✅ Old model WRECKED") return True except Exception as e: logger.error(f"[WRECKER] Failed: {e}") return False # --- TOKEN MANAGER --- class TokenManager: def __init__(self): self.user_tokens = {} # {username: {"balance": float, "start_time": float, "purchases": {}}} self.owner_username = "turtle170" # Owner gets infinite tokens def is_owner(self, username: str) -> bool: """Check if user is the owner""" if not username: return False return username.lower() == self.owner_username.lower() def initialize_user(self, username: str): """Initialize new user with monthly credits (or infinite for owner)""" if not username: username = "anonymous" if username not in self.user_tokens: # Owner gets infinite tokens if self.is_owner(username): self.user_tokens[username] = { "balance": float('inf'), "start_time": time.time(), "purchases": {"batch_multiplier": 1, "token_limit": 2048}, "total_spent": 0.0, "is_owner": True, "username": username } logger.info(f"[TOKEN] 👑 OWNER {username} initialized with INFINITE tokens!") else: self.user_tokens[username] = { "balance": MONTHLY_TOKEN_CREDITS, "start_time": time.time(), "purchases": {"batch_multiplier": 1, "token_limit": 2048}, "total_spent": 0.0, "is_owner": False, "username": username, "last_reset": time.time() } logger.info(f"[TOKEN] New user {username}: {MONTHLY_TOKEN_CREDITS} tokens") def check_monthly_reset(self, username: str): """Reset tokens if a month has passed""" if not username or username not in self.user_tokens: return if self.user_tokens[username].get("is_owner", False): return # Owner never needs reset last_reset = self.user_tokens[username].get("last_reset", time.time()) month_in_seconds = 30 * 24 * 60 * 60 # 30 days if time.time() - last_reset > month_in_seconds: self.user_tokens[username]["balance"] = MONTHLY_TOKEN_CREDITS self.user_tokens[username]["last_reset"] = time.time() self.user_tokens[username]["total_spent"] = 0.0 logger.info(f"[TOKEN] Monthly reset for {username}: {MONTHLY_TOKEN_CREDITS} tokens") def charge_usage(self, username: str, duration_ms: float) -> bool: """Charge user for inference time. Returns True if successful. Owner never charged.""" if not username: username = "anonymous" self.initialize_user(username) self.check_monthly_reset(username) # Owner never gets charged if self.user_tokens[username].get("is_owner", False): return True cost = (duration_ms / 100.0) * TOKEN_COST_PER_100MS # Check if user has enough balance if self.user_tokens[username]["balance"] <= 0: logger.warning(f"[TOKEN] ❌ {username} has 0 tokens! Access denied.") return False if self.user_tokens[username]["balance"] >= cost: self.user_tokens[username]["balance"] -= cost self.user_tokens[username]["balance"] = max(0, self.user_tokens[username]["balance"]) # Never go below 0 self.user_tokens[username]["total_spent"] += cost logger.info(f"[TOKEN] Charged {cost:.4f} tokens ({duration_ms:.0f}ms) | Remaining: {self.user_tokens[username]['balance']:.2f}") return True else: # Insufficient balance - set to 0 and deny self.user_tokens[username]["balance"] = 0 logger.warning(f"[TOKEN] ❌ Insufficient balance! {username} now at 0 tokens.") return False def can_use_engine(self, username: str) -> tuple: """Check if user can use the engine. Returns (bool, message)""" if not username: username = "anonymous" self.initialize_user(username) self.check_monthly_reset(username) if self.user_tokens[username].get("is_owner", False): return True, "👑 Owner access granted" balance = self.user_tokens[username]["balance"] if balance <= 0: last_reset = self.user_tokens[username].get("last_reset", time.time()) time_until_reset = 30 * 24 * 60 * 60 - (time.time() - last_reset) days_left = int(time_until_reset / (24 * 60 * 60)) return False, f"❌ Out of tokens! Resets in {days_left} days. Current balance: 0.00" return True, f"✅ Access granted. Balance: {balance:.2f} tokens" def purchase_batch_upgrade(self, username: str) -> tuple: """Purchase batch size upgrade (exponential cost). Free for owner.""" if not username: return False, "❌ Please login first" self.initialize_user(username) # Owner gets free upgrades if self.user_tokens[username].get("is_owner", False): current_mult = self.user_tokens[username]["purchases"]["batch_multiplier"] self.user_tokens[username]["purchases"]["batch_multiplier"] = current_mult * 2 new_mult = current_mult * 2 logger.info(f"[TOKEN] 👑 OWNER free batch upgrade: {current_mult}x → {new_mult}x") return True, f"👑 FREE UPGRADE! Batch now {new_mult}x!" current_mult = self.user_tokens[username]["purchases"]["batch_multiplier"] upgrade_level = int(math.log2(current_mult)) if current_mult > 1 else 0 cost = BATCH_UPGRADE_BASE_COST * (2 ** upgrade_level) if self.user_tokens[username]["balance"] >= cost: self.user_tokens[username]["balance"] -= cost self.user_tokens[username]["purchases"]["batch_multiplier"] = current_mult * 2 new_mult = current_mult * 2 logger.info(f"[TOKEN] Batch upgrade: {current_mult}x → {new_mult}x | Cost: {cost:.5f}") return True, f"✅ Batch upgraded to {new_mult}x! (-{cost:.5f} tokens)" else: return False, f"❌ Insufficient tokens! Need {cost:.5f}, have {self.user_tokens[username]['balance']:.2f}" def purchase_token_upgrade(self, username: str, extra_tokens: int = 1000) -> tuple: """Purchase extra response token length. Free for owner.""" if not username: return False, "❌ Please login first" self.initialize_user(username) # Owner gets free upgrades if self.user_tokens[username].get("is_owner", False): self.user_tokens[username]["purchases"]["token_limit"] += extra_tokens new_limit = self.user_tokens[username]["purchases"]["token_limit"] logger.info(f"[TOKEN] 👑 OWNER free token upgrade: +{extra_tokens} tokens") return True, f"👑 FREE UPGRADE! Token limit now {new_limit}!" cost = (extra_tokens / 1000) * TOKEN_UPGRADE_COST_PER_1K if self.user_tokens[username]["balance"] >= cost: self.user_tokens[username]["balance"] -= cost self.user_tokens[username]["purchases"]["token_limit"] += extra_tokens new_limit = self.user_tokens[username]["purchases"]["token_limit"] logger.info(f"[TOKEN] Token limit upgrade: +{extra_tokens} tokens | Cost: {cost:.5f}") return True, f"✅ Token limit now {new_limit}! (-{cost:.5f} tokens)" else: return False, f"❌ Insufficient tokens! Need {cost:.5f}, have {self.user_tokens[username]['balance']:.2f}" def get_balance(self, username: str) -> float: """Get user's current token balance""" if not username: username = "anonymous" self.initialize_user(username) self.check_monthly_reset(username) balance = self.user_tokens[username]["balance"] # Show ∞ for owner if balance == float('inf'): return balance return round(max(0, balance), 2) # Never show negative def get_purchases(self, username: str) -> dict: """Get user's current purchases""" if not username: username = "anonymous" self.initialize_user(username) return self.user_tokens[username]["purchases"] def end_session(self, username: str): """End user session and log stats""" if not username: return "No active session found." if username in self.user_tokens: stats = self.user_tokens[username] if stats.get("is_owner", False): return f"👑 Owner session ended. Welcome back anytime, {stats['username']}!" logger.info(f"[TOKEN] Session ended: Spent {stats['total_spent']:.2f}, Remaining {stats['balance']:.2f}") return f"Session ended. You spent {stats['total_spent']:.2f} tokens this session. Balance: {stats['balance']:.2f}" return "No active session found." # Global token manager import math token_manager = TokenManager() # Global cache manager model_cache = ModelCacheManager() # --- TELEMETRY MODULE --- class TelemetryManager: def __init__(self, api: HfApi): self.api = api self.stats = self._load_initial_stats() def _load_initial_stats(self) -> Dict: # Simplified: no file I/O to prevent restart issues return { "session_start": str(datetime.now(pytz.utc)), "load_count": {}, "total_tokens_generated": 0 } def track_load(self, repo: str, filename: str): key = f"{repo}/{filename}" self.stats["load_count"][key] = self.stats["load_count"].get(key, 0) + 1 logger.info(f"Model loaded: {key} (count: {self.stats['load_count'][key]})") def track_generation(self, tokens: int): self.stats["total_tokens_generated"] += tokens logger.info(f"Total tokens generated: {self.stats['total_tokens_generated']}") # --- RESOURCE MONITOR --- class ResourceMonitor: @staticmethod def get_metrics() -> Dict: vm = psutil.virtual_memory() return { "ram_used_gb": round(vm.used / (1024**3), 2), "ram_avail_gb": round(vm.available / (1024**3), 2), "ram_total_gb": round(vm.total / (1024**3), 2), "ram_pct": vm.percent, "cpu_usage_pct": psutil.cpu_percent(interval=None), "load_avg": os.getloadavg()[0] if hasattr(os, 'getloadavg') else 0 } @staticmethod def validate_deployment(file_path: str) -> (bool, str): try: vm = psutil.virtual_memory() file_size_mb = os.path.getsize(file_path) / (1024**2) total_ram_mb = vm.total / (1024**2) avail_ram_mb = vm.available / (1024**2) logger.info(f"Validation - Model: {file_size_mb:.1f}MB | Available RAM: {avail_ram_mb:.1f}MB | Total: {total_ram_mb:.1f}MB") if file_size_mb > (total_ram_mb * RAM_LIMIT_PCT): return False, f"Model size ({file_size_mb:.1f}MB) exceeds safety limit ({total_ram_mb * RAM_LIMIT_PCT:.1f}MB)." if (file_size_mb + SYSTEM_RESERVE_MB) > avail_ram_mb: return False, f"Insufficient RAM. Need {file_size_mb+SYSTEM_RESERVE_MB:.1f}MB, have {avail_ram_mb:.1f}MB available." return True, "Validation Passed." except Exception as e: logger.error(f"Validation error: {e}") return False, f"Validation error: {str(e)}" # --- ENGINE CORE --- class ZeroEngine: def __init__(self): self.api = HfApi(token=HF_TOKEN) self.telemetry = TelemetryManager(self.api) self.llm: Optional[Llama] = None self.active_model_info = {"repo": "", "file": "", "format": ""} self.kernel_lock = threading.Lock() self.is_prefilling = False self.perf_stats = { "total_tokens": 0, "total_time": 0.0, "avg_tps": 0.0, "peak_tps": 0.0, "cache_hits": 0 } self.prompt_cache = {} self.last_activity = time.time() self.idle_timeout = 20 self.auto_cleanup_thread = None self.start_idle_monitor() # Keyboard input pre-processing self.typing_buffer = "" self.typing_timer = None self.preprocessed_tokens = None # Custom parameters (user-configurable) self.custom_params = { "temperature": 0.7, "top_p": 0.95, "top_k": 40, "repeat_penalty": 1.1, "batch_size_override": None, # None = auto "max_tokens_override": None # None = auto } def detect_model_format(self, filename: str, repo: str) -> str: """Auto-detect model format/architecture from filename and repo""" combined = f"{repo.lower()} {filename.lower()}" for format_name, format_info in MODEL_FORMATS.items(): for pattern in format_info["pattern"]: if pattern in combined: logger.info(f"[FORMAT-DETECT] Detected {format_name.upper()} architecture") return format_name logger.warning(f"[FORMAT-DETECT] Unknown format, defaulting to llama") return "llama" def detect_quantization(self, filename: str) -> dict: """Detect quantization method from filename and return optimizations""" filename_upper = filename.upper() for quant_type, optimizations in QUANT_OPTIMIZATIONS.items(): if quant_type in filename_upper: logger.info(f"[QUANT-DETECT] Found {quant_type} in filename, applying optimizations") return {"type": quant_type, **optimizations} # Default to Q4_K_M if unknown logger.warning(f"[QUANT-DETECT] Unknown quantization, using Q4_K_M defaults") return {"type": "Q4_K_M", **QUANT_OPTIMIZATIONS["Q4_K_M"]} def preprocess_input(self, text: str): """Pre-process keyboard input in background (tensors ready before submit)""" if not self.llm or not text or len(text) < 5: return def _preprocess(): try: logger.info(f"[PREPROCESS] Tokenizing {len(text)} chars in background...") tokens = self.llm.tokenize(text.encode("utf-8")) self.preprocessed_tokens = tokens logger.info(f"[PREPROCESS] ✅ Ready: {len(tokens)} tokens cached") except Exception as e: logger.error(f"[PREPROCESS] Failed: {e}") self.preprocessed_tokens = None # Cancel previous timer if user is still typing if self.typing_timer: self.typing_timer.cancel() # Start new timer - preprocess after 1 second of no typing self.typing_timer = threading.Timer(1.0, _preprocess) self.typing_timer.daemon = True self.typing_timer.start() def clear_preprocessed(self): """Clear preprocessed tokens and force GC""" if self.preprocessed_tokens: self.preprocessed_tokens = None force_gc() logger.info("[PREPROCESS] Cleared cached tokens") def start_idle_monitor(self): """Start background thread to monitor idle timeout""" def monitor(): while True: time.sleep(5) # Check every 5 seconds if self.llm and (time.time() - self.last_activity) > self.idle_timeout: logger.info(f"[IDLE] No activity for {self.idle_timeout}s, unloading model...") with self.kernel_lock: if self.llm: try: del self.llm self.llm = None self.active_model_info = {"repo": "", "file": ""} force_gc() # Aggressive cleanup logger.info("[IDLE] Model unloaded successfully") except Exception as e: logger.error(f"[IDLE] Cleanup error: {e}") self.auto_cleanup_thread = threading.Thread(target=monitor, daemon=True) self.auto_cleanup_thread.start() logger.info("[IDLE] Idle monitor started (20s timeout)") def update_activity(self): """Update last activity timestamp""" self.last_activity = time.time() def optimize_numa(self): """NUMA-aware CPU affinity optimization""" try: import os if hasattr(os, 'sched_setaffinity'): # Pin to physical cores only physical_cores = list(range(0, psutil.cpu_count(logical=False))) os.sched_setaffinity(0, physical_cores) logger.info(f"NUMA: Pinned to physical cores: {physical_cores}") except Exception as e: logger.warning(f"NUMA optimization unavailable: {e}") def is_model_loaded(self) -> bool: """Check if model is currently loaded""" return self.llm is not None def list_ggufs(self, repo_id: str) -> List[str]: try: files = self.api.list_repo_files(repo_id=repo_id) ggufs = [f for f in files if f.endswith(".gguf")] logger.info(f"Found {len(ggufs)} GGUF files in {repo_id}") return ggufs except Exception as e: logger.error(f"Scan error: {e}") return [] def boot_kernel(self, repo: str, filename: str, session_id: str = None) -> str: """HYPER-OPTIMIZED Boot kernel with format auto-detection and Gemma fixes""" try: if not repo or not filename: return "🔴 ERROR: Repository or filename missing" logger.info(f"[BOOT] Starting download: {filename} from {repo}") # DETECT QUANTIZATION FROM FILENAME quant_config = self.detect_quantization(filename) # DETECT MODEL FORMAT/ARCHITECTURE model_format = self.detect_model_format(filename, repo) # Download with timeout protection try: path = hf_hub_download( repo_id=repo, filename=filename, token=HF_TOKEN, local_files_only=False ) logger.info(f"[BOOT] Download complete: {path}") except Exception as e: logger.error(f"[BOOT] Download failed: {e}") return f"🔴 DOWNLOAD FAILED: {str(e)}" # Check if model is cached is_cached = model_cache.is_cached(path) cache_status = "🎯 CACHED" if is_cached else "🆕 NEW" # Validate before loading valid, msg = ResourceMonitor.validate_deployment(path) if not valid: logger.warning(f"[BOOT] Validation failed: {msg}") return f"🔴 VALIDATION FAILED: {msg}" logger.info(f"[BOOT] Validation passed ({cache_status}), applying {quant_config['type']} optimizations for {model_format.upper()}...") # Load model with MAXIMUM PERFORMANCE SETTINGS with self.kernel_lock: # WRECK OLD MODEL if self.llm: logger.info("[BOOT] 💣 WRECKING old model...") try: model_cache.wreck_old_model_cache() del self.llm self.llm = None nuclear_ram_clear() logger.info("[BOOT] ✅ Old model DESTROYED") except Exception as e: logger.warning(f"[BOOT] Cleanup warning: {e}") # Calculate optimal parameters with token purchases vm = psutil.virtual_memory() available_ram_gb = vm.available / (1024**3) # CPU-OPTIMIZED BATCH CALCULATION - Very aggressive for 16GB RAM # Base calculation: use more RAM for batching on CPU base_batch = int(512 * available_ram_gb / 8) # More aggressive base optimal_batch = int(base_batch * quant_config["batch_multiplier"]) # Apply user's batch multiplier from token purchases if session_id: user_batch_mult = token_manager.get_purchases(session_id)["batch_multiplier"] optimal_batch = int(optimal_batch * user_batch_mult) logger.info(f"[TOKEN] User batch multiplier: {user_batch_mult}x") # CPU can handle larger batches with quantized models optimal_batch = max(1024, min(8192, optimal_batch)) # 1024-8192 range for CPU # Context size optimal_ctx = quant_config["ctx_size"] # Reduce context for Gemma models (they have 131K n_ctx_train) if model_format == "gemma": optimal_ctx = min(8192, optimal_ctx) # Gemma works better with lower ctx logger.info(f"[FORMAT] Gemma detected: reducing context to {optimal_ctx}") # Thread optimization - use ALL threads on CPU (including hyperthreading) optimal_threads = psutil.cpu_count(logical=True) # ALL logical cores logger.info(f"[CPU] Using all {optimal_threads} threads (including hyperthreading)") try: logger.info(f"[BOOT] Initializing {model_format.upper()} {quant_config['type']}: threads={optimal_threads}, batch={optimal_batch}, ctx={optimal_ctx}") # Preload cache if available if is_cached: model_cache.preload_cache(path) # ULTRA-OPTIMIZED CPU-ONLY INITIALIZATION init_params = { "model_path": path, "n_ctx": optimal_ctx, "n_threads": optimal_threads, "n_threads_batch": optimal_threads, "use_mmap": USE_MMAP, # Critical for CPU "use_mlock": MLOCK_MODEL, # Let OS manage memory "n_batch": optimal_batch, # MASSIVE batches for CPU "n_gpu_layers": 0, # CPU-only "rope_scaling_type": 0, "rope_freq_scale": ROPE_SCALING, "verbose": False, "logits_all": False, "embedding": False, "f16_kv": False # Use quantized KV cache } # Add KV quantization only if not Gemma (Gemma can be finicky) if model_format != "gemma" and KV_CACHE_QUANTIZATION: init_params["type_k"] = 2 init_params["type_v"] = 2 logger.info("[OPTIM] KV cache quantization enabled (Q4)") self.llm = Llama(**init_params) self.active_model_info = { "repo": repo, "file": filename, "quant": quant_config['type'], "format": model_format } self.telemetry.track_load(repo, filename) # Extract and cache signature if not is_cached: logger.info("[BOOT] Extracting cache signature...") signature = model_cache.extract_cache_signature(path) if signature: model_cache.save_to_cache(path, signature) # Warm-up logger.info("[BOOT] Warming up model caches...") try: self.llm("Warmup", max_tokens=1, stream=False) force_gc() except: pass logger.info("[BOOT] 🚀 CPU-OPTIMIZED MODEL READY!") return f"🟢 {model_format.upper()} {quant_config['type']} {cache_status} | CPU:{optimal_threads}T | B:{optimal_batch} | Ctx:{optimal_ctx}" except Exception as e: logger.error(f"[BOOT] Model loading failed: {e}") self.llm = None nuclear_ram_clear() return f"🔴 LOAD FAILED: {str(e)}" except Exception as e: logger.error(f"[BOOT] Unexpected error: {e}") nuclear_ram_clear() return f"🔴 BOOT FAILURE: {str(e)}" """HYPER-OPTIMIZED Boot kernel with cache manager and old model wrecker""" try: if not repo or not filename: return "🔴 ERROR: Repository or filename missing" logger.info(f"[BOOT] Starting download: {filename} from {repo}") # DETECT QUANTIZATION FROM FILENAME quant_config = self.detect_quantization(filename) # Download with timeout protection try: path = hf_hub_download( repo_id=repo, filename=filename, token=HF_TOKEN, local_files_only=False ) logger.info(f"[BOOT] Download complete: {path}") except Exception as e: logger.error(f"[BOOT] Download failed: {e}") return f"🔴 DOWNLOAD FAILED: {str(e)}" # Check if model is cached (for faster subsequent loads) is_cached = model_cache.is_cached(path) cache_status = "🎯 CACHED" if is_cached else "🆕 NEW" # Validate before loading valid, msg = ResourceMonitor.validate_deployment(path) if not valid: logger.warning(f"[BOOT] Validation failed: {msg}") return f"🔴 VALIDATION FAILED: {msg}" logger.info(f"[BOOT] Validation passed ({cache_status}), applying {quant_config['type']} optimizations...") # Apply NUMA optimization if NUMA_OPTIMIZE: self.optimize_numa() # Load model with MAXIMUM PERFORMANCE SETTINGS with self.kernel_lock: # WRECK OLD MODEL - Nuclear option if self.llm: logger.info("[BOOT] 💣 WRECKING old model...") try: # Wreck the cache first model_cache.wreck_old_model_cache() # Delete the model del self.llm self.llm = None # Nuclear RAM clear nuclear_ram_clear() logger.info("[BOOT] ✅ Old model DESTROYED") except Exception as e: logger.warning(f"[BOOT] Cleanup warning: {e}") # Calculate optimal batch size based on quantization and available RAM vm = psutil.virtual_memory() available_ram_gb = vm.available / (1024**3) # MASSIVE batch sizes for quantized models base_batch = int(256 * available_ram_gb / 4) optimal_batch = int(base_batch * quant_config["batch_multiplier"]) optimal_batch = max(512, min(4096, optimal_batch)) # Clamp between 512-4096 # Context size based on quantization optimal_ctx = quant_config["ctx_size"] # Thread count with quantization-specific boost optimal_threads = int(OPTIMAL_THREADS * quant_config["threads_boost"]) optimal_threads = max(2, min(optimal_threads, psutil.cpu_count(logical=False))) try: logger.info(f"[BOOT] Initializing {quant_config['type']}: threads={optimal_threads}, batch={optimal_batch}, ctx={optimal_ctx}") # Preload cache if available (simulates faster warmup) if is_cached: model_cache.preload_cache(path) # ULTRA-OPTIMIZED LLAMA.CPP INITIALIZATION self.llm = Llama( model_path=path, n_ctx=optimal_ctx, # Dynamic context based on quant n_threads=optimal_threads, # Optimized thread count n_threads_batch=optimal_threads, # Batch processing threads use_mmap=USE_MMAP, # Memory-mapped weights (fast loading) use_mlock=MLOCK_MODEL, # Lock in RAM (prevent swap thrashing) n_batch=optimal_batch, # MASSIVE batch size n_gpu_layers=0, # CPU-only mode flash_attn=FLASH_ATTENTION, # Flash Attention (2x faster) type_k=2 if KV_CACHE_QUANTIZATION else None, # Q4 KV cache quantization type_v=2 if KV_CACHE_QUANTIZATION else None, # Q4 KV cache quantization rope_scaling_type=0, # Linear RoPE scaling rope_freq_scale=ROPE_SCALING, # RoPE frequency scale numa=NUMA_OPTIMIZE, # NUMA optimization verbose=False, logits_all=False, # Only compute final logits (faster) embedding=False, # Disable embeddings (not needed) offload_kqv=OFFLOAD_KQV, # No offload on CPU f16_kv=False # Use quantized KV cache instead ) self.active_model_info = {"repo": repo, "file": filename, "quant": quant_config['type']} self.telemetry.track_load(repo, filename) # Extract and cache TINY signature for faster future loads if not is_cached: logger.info("[BOOT] Extracting cache signature...") signature = model_cache.extract_cache_signature(path) if signature: model_cache.save_to_cache(path, signature) # Warm-up inference to populate caches logger.info("[BOOT] Warming up model caches...") try: self.llm("Warmup", max_tokens=1, stream=False) force_gc() # Clear warmup artifacts except: pass logger.info("[BOOT] 🚀 HYPER-OPTIMIZED MODEL READY!") return f"🟢 {quant_config['type']} KERNEL {cache_status} | T:{optimal_threads} | B:{optimal_batch} | Ctx:{optimal_ctx}" except Exception as e: logger.error(f"[BOOT] Model loading failed: {e}") self.llm = None nuclear_ram_clear() return f"🔴 LOAD FAILED: {str(e)}" except Exception as e: logger.error(f"[BOOT] Unexpected error: {e}") nuclear_ram_clear() return f"🔴 BOOT FAILURE: {str(e)}" def stitch_cache(self, ghost_text: str) -> str: """Prime KV cache with ghost context""" if not self.llm or not ghost_text or self.is_prefilling: return "Kernel Idle/Busy" def _bg_eval(): self.is_prefilling = True try: tokens = self.llm.tokenize(ghost_text.encode("utf-8")) self.llm.eval(tokens) logger.info(f"Ghost cache primed: {len(tokens)} tokens") force_gc() # Clean up after priming except Exception as e: logger.error(f"KV Cache priming failed: {e}") finally: self.is_prefilling = False threading.Thread(target=_bg_eval, daemon=True).start() return "⚡ Primed" def inference_generator(self, prompt: str, history: List[Dict], ghost_context: str, repo: str, quant: str, username: str) -> Generator: # Update activity timestamp self.update_activity() # Clear any preprocessed tokens from typing self.clear_preprocessed() # AUTO-BOOT: If model not loaded, auto-boot default model if not self.llm: logger.info("[AUTO-BOOT] No model loaded, initiating auto-boot...") history.append({"role": "assistant", "content": "🔄 Auto-booting model, please wait..."}) yield history # Use provided repo/quant or fallback to defaults boot_repo = repo if repo else DEFAULT_MODEL boot_quant = quant if quant else DEFAULT_QUANT boot_result = self.boot_kernel(boot_repo, boot_quant) if "🔴" in boot_result or "FAILED" in boot_result: history[-1]["content"] = f"❌ Auto-boot failed: {boot_result}\n\nPlease manually SCAN and BOOT a model." yield history return history[-1]["content"] = f"✅ {boot_result}\n\nProcessing your request..." yield history time.sleep(0.5) # Brief pause for user to see the message # Check prompt cache for exact matches (instant response) cache_key = f"{ghost_context}:{prompt}" if cache_key in self.prompt_cache: self.perf_stats["cache_hits"] += 1 logger.info("⚡ CACHE HIT - Instant response!") history.append({"role": "user", "content": prompt}) history.append({"role": "assistant", "content": self.prompt_cache[cache_key]}) yield history return # Prepare input with optimized formatting full_input = f"{ghost_context}\n{prompt}" if ghost_context else prompt formatted_prompt = f"User: {full_input}\nAssistant: " # Add User Message & Empty Assistant Message for Streaming history.append({"role": "user", "content": prompt}) history.append({"role": "assistant", "content": "..."}) yield history response_text = "" start_time = time.time() tokens_count = 0 first_token_time = None try: # Get max tokens from user purchases max_tokens = 2048 if username: max_tokens = token_manager.get_purchases(username)["token_limit"] # HYPER-OPTIMIZED CPU INFERENCE SETTINGS stream = self.llm( formatted_prompt, max_tokens=max_tokens, stop=["User:", "<|eot_id|>", "\n\n"], stream=True, temperature=self.custom_params["temperature"], top_p=self.custom_params["top_p"], top_k=self.custom_params["top_k"], repeat_penalty=self.custom_params["repeat_penalty"], frequency_penalty=0.0, presence_penalty=0.0, tfs_z=1.0, typical_p=1.0, mirostat_mode=2, # CPU benefits from mirostat mirostat_tau=5.0, mirostat_eta=0.1, ) for chunk in stream: token = chunk["choices"][0]["text"] response_text += token tokens_count += 1 # Track first token latency (TTFT - Time To First Token) if first_token_time is None: first_token_time = time.time() - start_time logger.info(f"⚡ First token: {first_token_time*1000:.0f}ms") elapsed = time.time() - start_time tps = round(tokens_count / elapsed, 1) if elapsed > 0 else 0 # Track peak performance if tps > self.perf_stats["peak_tps"]: self.perf_stats["peak_tps"] = tps # Charge tokens every second if int(elapsed * 1000) % 1000 < 100 and username: # Every ~1 second token_manager.charge_usage(username, elapsed * 1000) # Update history with streaming content + performance metrics balance = token_manager.get_balance(username) if username else 0 history[-1]["content"] = f"{response_text}\n\n`⚡ {tps} t/s | 🎯 Peak: {self.perf_stats['peak_tps']:.1f} t/s | 💰 {balance:.2f} tokens`" yield history # Final token charge for remaining time if username: token_manager.charge_usage(username, elapsed * 1000) # Update global performance stats self.perf_stats["total_tokens"] += tokens_count self.perf_stats["total_time"] += elapsed self.perf_stats["avg_tps"] = self.perf_stats["total_tokens"] / self.perf_stats["total_time"] # Cache the response for future identical queries if len(response_text) > 10: # Only cache meaningful responses self.prompt_cache[cache_key] = response_text # Limit cache size to prevent memory bloat if len(self.prompt_cache) > 100: oldest_key = next(iter(self.prompt_cache)) del self.prompt_cache[oldest_key] self.telemetry.track_generation(tokens_count) # Aggressive GC after generation force_gc() logger.info(f"✅ Generation complete: {tokens_count} tokens @ {tps:.1f} t/s (TTFT: {first_token_time*1000:.0f}ms)") except Exception as e: logger.error(f"Inference error: {e}") history[-1]["content"] = f"🔴 Runtime Error: {str(e)}" yield history force_gc() # --- CUSTOM CSS --- CUSTOM_CSS = """ @import url('https://fonts.cdnfonts.com/css/consolas'); * { font-family: 'Consolas', 'Courier New', monospace !important; } /* Global smooth rounded corners */ .gradio-container { border-radius: 24px !important; } /* All buttons */ button { border-radius: 16px !important; transition: all 0.3s cubic-bezier(0.4, 0, 0.2, 1) !important; font-family: 'Consolas', monospace !important; } button:hover { transform: translateY(-2px); box-shadow: 0 8px 16px rgba(0,0,0,0.2) !important; } /* Input fields */ input, textarea, .gr-textbox, .gr-dropdown { border-radius: 12px !important; font-family: 'Consolas', monospace !important; } /* Chat messages */ .message { border-radius: 16px !important; font-family: 'Consolas', monospace !important; } /* Code blocks */ .gr-code { border-radius: 12px !important; font-family: 'Consolas', monospace !important; } /* Labels */ .gr-label { border-radius: 12px !important; font-family: 'Consolas', monospace !important; } /* Sidebar */ .gr-sidebar { border-radius: 20px !important; background: linear-gradient(135deg, rgba(20,20,40,0.95), rgba(10,10,20,0.98)) !important; backdrop-filter: blur(10px) !important; } /* Markdown sections */ .gr-markdown { font-family: 'Consolas', monospace !important; } /* Chatbot container */ .chatbot { border-radius: 20px !important; font-family: 'Consolas', monospace !important; } /* Dropdown menus */ .gr-dropdown-menu { border-radius: 12px !important; font-family: 'Consolas', monospace !important; } /* Column containers */ .gr-column { border-radius: 16px !important; } /* Row containers */ .gr-row { border-radius: 12px !important; } /* Smooth animations for all interactive elements */ * { transition: all 0.2s ease !important; } /* Header styling */ h1, h2, h3, h4, h5, h6 { font-family: 'Consolas', monospace !important; } """ # --- UI INTERFACE --- kernel = ZeroEngine() # Session ID for token tracking username = token_manager.get_username() with gr.Blocks(title="ZeroEngine V0.2", css=CUSTOM_CSS) as demo: # Header with Token Display with gr.Row(): with gr.Column(scale=8): gr.HTML("""

🛰️ ZEROENGINE V0.2

CPU-Optimized | Token System | Custom Parameters | Auto-Format

""") with gr.Column(scale=2): # Token Display gr.HTML("""
💰
100.00
TOKENS
""") token_balance = gr.Textbox(value="100.00", visible=False, elem_id="token_balance") end_session_btn = gr.Button("END SESSION", variant="stop", size="sm") session_status = gr.Markdown("", visible=False) with gr.Row(): with gr.Column(scale=8): chat_box = gr.Chatbot( label="Main Engine Feedback", height=600, show_label=False, autoscroll=True, container=True ) with gr.Row(): user_input = gr.Textbox( placeholder="Input command...", label="Terminal", container=False, scale=9 ) send_btn = gr.Button("SUBMIT", variant="primary", scale=1) with gr.Column(scale=4): # Hardware Status gr.Markdown("### 🛠️ Hardware Status") ram_metric = gr.Label(label="RAM Usage", value="0/0 GB") cpu_metric = gr.Label(label="CPU Load", value="0%") gr.Markdown("---") # Model Control gr.Markdown("### 📡 Model Control") repo_input = gr.Textbox(label="HuggingFace Repo", value=DEFAULT_MODEL) quant_dropdown = gr.Dropdown(label="Available Quants", choices=[], interactive=True) with gr.Row(): scan_btn = gr.Button("SCAN", size="sm") boot_btn = gr.Button("BOOT", variant="primary", size="sm") boot_status = gr.Markdown("Status: `STANDBY`") gr.Markdown("---") # Custom Parameters gr.Markdown("### ⚙️ Custom Parameters") temperature_slider = gr.Slider(0.1, 2.0, value=0.7, step=0.1, label="Temperature") top_p_slider = gr.Slider(0.1, 1.0, value=0.95, step=0.05, label="Top-P") top_k_slider = gr.Slider(1, 100, value=40, step=1, label="Top-K") repeat_penalty_slider = gr.Slider(1.0, 2.0, value=1.1, step=0.1, label="Repeat Penalty") gr.Markdown("---") # Token Purchases gr.Markdown("### 💎 Token Upgrades") with gr.Row(): batch_upgrade_btn = gr.Button("🚀 Batch x2", size="sm", variant="secondary") token_upgrade_btn = gr.Button("📈 +1K Tokens", size="sm", variant="secondary") purchase_status = gr.Markdown("Ready to upgrade!") gr.Markdown("---") # Ghost Cache gr.Markdown("### 👻 Ghost Cache (Pre-Context)") ghost_buffer = gr.Textbox( label="Background Context", placeholder="Add context that will be prepended to all messages...", lines=3 ) with gr.Row(): stitch_btn = gr.Button("PRIME CACHE", variant="secondary", size="sm", scale=1) stitch_status = gr.Markdown("Cache: `EMPTY`") log_output = gr.Code( label="Kernel Logs", language="shell", value="[INIT] V0.2 System Ready.", lines=5 ) # --- UI LOGIC --- def update_stats(): try: m = ResourceMonitor.get_metrics() balance = token_manager.get_balance(session_id) return f"{m['ram_used_gb']}/{m['ram_total_gb']} GB", f"{m['cpu_usage_pct']}%", f"{balance}" except Exception as e: logger.error(f"Stats update error: {e}") return "Error", "Error", "0.00" return "Error", "Error" def on_scan(repo): try: if not repo: return gr.update(choices=[], value=None), "⚠️ Please enter a repository ID" logger.info(f"Scanning repository: {repo}") files = kernel.list_ggufs(repo) if not files: return gr.update(choices=[], value=None), f"❌ No GGUFs found in {repo}" return gr.update(choices=files, value=files[0]), f"✅ Found {len(files)} GGUF file(s)" except Exception as e: logger.error(f"Scan error: {e}") return gr.update(choices=[], value=None), f"🔴 Scan failed: {str(e)}" def on_boot(repo, file): try: if not repo or not file: yield "🔴 ERROR: Repository and filename required", gr.update() return yield "⚙️ System: Initiating boot sequence...", gr.update() time.sleep(0.5) result = kernel.boot_kernel(repo, file, session_id) yield result, gr.update() except Exception as e: logger.error(f"Boot UI error: {e}") yield f"🔴 BOOT ERROR: {str(e)}", gr.update() def on_batch_upgrade(): success, msg = token_manager.purchase_batch_upgrade(session_id) balance = token_manager.get_balance(session_id) return msg, f"{balance}" def on_token_upgrade(): success, msg = token_manager.purchase_token_upgrade(session_id, 1000) balance = token_manager.get_balance(session_id) return msg, f"{balance}" def on_end_session(): msg = token_manager.end_session(session_id) return msg def update_custom_params(temp, top_p, top_k, repeat_pen): kernel.custom_params["temperature"] = temp kernel.custom_params["top_p"] = top_p kernel.custom_params["top_k"] = int(top_k) kernel.custom_params["repeat_penalty"] = repeat_pen return "✅ Parameters updated!" # Timer for periodic stats updates (includes token balance) timer = gr.Timer(value=2) timer.tick(update_stats, None, [ram_metric, cpu_metric, token_balance]) # Event handlers scan_btn.click(on_scan, [repo_input], [quant_dropdown, log_output]) boot_btn.click(on_boot, [repo_input, quant_dropdown], [boot_status, log_output]) # Token purchases batch_upgrade_btn.click(on_batch_upgrade, None, [purchase_status, token_balance]) token_upgrade_btn.click(on_token_upgrade, None, [purchase_status, token_balance]) end_session_btn.click(on_end_session, None, [session_status]) # Custom parameter updates temperature_slider.change(update_custom_params, [temperature_slider, top_p_slider, top_k_slider, repeat_penalty_slider], [purchase_status]) top_p_slider.change(update_custom_params, [temperature_slider, top_p_slider, top_k_slider, repeat_penalty_slider], [purchase_status]) top_k_slider.change(update_custom_params, [temperature_slider, top_p_slider, top_k_slider, repeat_penalty_slider], [purchase_status]) repeat_penalty_slider.change(update_custom_params, [temperature_slider, top_p_slider, top_k_slider, repeat_penalty_slider], [purchase_status]) # Ghost cache stitch_btn.click( lambda x: f"Cache: `{kernel.stitch_cache(x)}`", [ghost_buffer], [stitch_status] ) # Keyboard input preprocessing user_input.change( lambda x: kernel.preprocess_input(x), [user_input], None ) # Auto-boot enabled inference inference_args = [user_input, chat_box, ghost_buffer, repo_input, quant_dropdown] user_input.submit(kernel.inference_generator, inference_args, [chat_box]) send_btn.click(kernel.inference_generator, inference_args, [chat_box]) user_input.submit(lambda: "", None, [user_input]) # --- LAUNCH --- if __name__ == "__main__": demo.queue(max_size=20).launch( server_name="0.0.0.0", server_port=7860, share=False )