""" Core Inference Engine for Nova AI - Flagship ChatGPT Astra Enterprise Suite. Supports high-throughput streaming, multi-model switching, NVIDIA ZeroGPU & CUDA hardware acceleration, BitsAndBytes 4-bit/8-bit quantization, PyTorch SDPA, DeepSeek-R1 chain-of-thought, and resilient CPU fallback. """ import os import sys import threading from pathlib import Path from typing import Generator, List, Dict, Any, Optional, Tuple import torch from transformers import ( AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer ) from src.config import config, AVAILABLE_MODELS, CACHE_DIR, EXPORTED_MODEL_DIR, GenerationConfig from src.utils import get_system_hardware_info, LatencyTracker, console # ===================================================================== # HUGGING FACE ZEROGPU (NVIDIA A100 DYNAMIC ACCELERATION) INTEGRATION # ===================================================================== try: import spaces HAS_ZERO_GPU = True except ImportError: HAS_ZERO_GPU = False class spaces: """Fallback decorator for local NVIDIA CUDA & non-ZeroGPU environments.""" @staticmethod def GPU(func=None, duration=120): def decorator(f): return f if func is not None: return func return decorator class NovaAstraEngine: """ Flagship Enterprise inference engine tailored for Nova AI (Astra Edition). NVIDIA ZeroGPU & CUDA accelerated, supporting DeepSeek-R1, Qwen2.5, Llama-3.1, custom safetensors checkpoints, and deep chain-of-thought execution. """ def __init__(self, model_id: Optional[str] = None): self.model_id = model_id or config.default_model_id self.device = self._resolve_device(config.device) self.model = None self.tokenizer = None self._is_loaded = False self._load_lock = threading.Lock() self.hardware_info = get_system_hardware_info() self._setup_cuda_optimizations() def _setup_cuda_optimizations(self): """Enable NVIDIA Tensor Core TF32 and CuDNN benchmark acceleration.""" if torch.cuda.is_available(): try: if hasattr(torch.backends, "cuda") and hasattr(torch.backends.cuda, "matmul"): torch.backends.cuda.matmul.allow_tf32 = True if hasattr(torch.backends, "cudnn"): torch.backends.cudnn.benchmark = True torch.backends.cudnn.allow_tf32 = True except Exception: pass def _resolve_device(self, preferred_device: str) -> str: """Resolve optimal compute device (NVIDIA CUDA / MPS / CPU).""" if preferred_device.lower() == "auto": if torch.cuda.is_available(): return "cuda" elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available(): return "mps" return "cpu" return preferred_device.lower() def load_model(self, model_id: Optional[str] = None, force_reload: bool = False): """Load or switch the active language model with NVIDIA CUDA / ZeroGPU optimization.""" target_model_id = str(model_id or self.model_id) with self._load_lock: if self._is_loaded and target_model_id == str(self.model_id) and not force_reload: return console.print(f"[bold cyan]🌟 Loading Flagship Nova Astra Model:[/bold cyan] {target_model_id}") console.print(f"[dim]Hardware Target: {self.device.upper()} | ZeroGPU: {'Active' if HAS_ZERO_GPU else 'Native'} | Cache: {CACHE_DIR}[/dim]") CACHE_DIR.mkdir(parents=True, exist_ok=True) is_local_dir = Path(target_model_id).is_dir() # 1. Load Tokenizer tokenizer_kwargs = {"trust_remote_code": True} if not is_local_dir: tokenizer_kwargs["cache_dir"] = str(CACHE_DIR) try: self.tokenizer = AutoTokenizer.from_pretrained(target_model_id, **tokenizer_kwargs) if self.tokenizer.pad_token is None: self.tokenizer.pad_token = self.tokenizer.eos_token except Exception as e: console.print(f"[yellow]Retrying tokenizer load with fast=False: {e}[/yellow]") tokenizer_kwargs["use_fast"] = False self.tokenizer = AutoTokenizer.from_pretrained(target_model_id, **tokenizer_kwargs) if self.tokenizer.pad_token is None: self.tokenizer.pad_token = self.tokenizer.eos_token # 2. Build Model Configuration (NVIDIA CUDA / ZeroGPU / CPU) precision_mode = getattr(config.gen_config, "precision", "auto").lower() load_in_4bit = getattr(config.gen_config, "load_in_4bit", False) or (precision_mode == "4bit") load_in_8bit = (precision_mode == "8bit") model_kwargs: Dict[str, Any] = { "low_cpu_mem_usage": True, "trust_remote_code": True } if not is_local_dir: model_kwargs["cache_dir"] = str(CACHE_DIR) if self.device == "cuda": major_cc = torch.cuda.get_device_capability()[0] if torch.cuda.is_available() else 0 total_vram_gb = torch.cuda.get_device_properties(0).total_memory / (1024 ** 3) if torch.cuda.is_available() else 0 # Auto-detect if 4-bit is optimal (if VRAM is under 12GB for 7B+ models) if not load_in_4bit and not load_in_8bit and precision_mode == "auto" and total_vram_gb < 12.0: load_in_4bit = True # Configure BitsAndBytes 4-bit / 8-bit quantization if requested or low VRAM if load_in_4bit or load_in_8bit: try: from transformers import BitsAndBytesConfig if load_in_4bit: compute_dtype = torch.bfloat16 if major_cc >= 8 else torch.float16 model_kwargs["quantization_config"] = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_compute_dtype=compute_dtype, bnb_4bit_quant_type="nf4", bnb_4bit_use_double_quant=True ) console.print("[bold green]⚡ NVIDIA BitsAndBytes 4-bit NF4 Quantization Active (Max Speed, Low VRAM)[/bold green]") elif load_in_8bit: model_kwargs["quantization_config"] = BitsAndBytesConfig(load_in_8bit=True) console.print("[bold green]⚡ NVIDIA BitsAndBytes 8-bit Quantization Active[/bold green]") except ImportError: console.print("[dim]bitsandbytes not installed; continuing in native half-precision.[/dim]") # Optimal torch_dtype for NVIDIA Tensor Cores if "quantization_config" not in model_kwargs: if precision_mode in ("bfloat16", "bf16") or (precision_mode == "auto" and major_cc >= 8): model_kwargs["torch_dtype"] = torch.bfloat16 elif precision_mode in ("float16", "fp16") or precision_mode == "auto": model_kwargs["torch_dtype"] = torch.float16 else: model_kwargs["torch_dtype"] = torch.float32 # Attention optimization via PyTorch SDPA (Scaled Dot-Product Attention) try: model_kwargs["attn_implementation"] = "sdpa" except Exception: pass model_kwargs["device_map"] = "auto" else: model_kwargs["torch_dtype"] = torch.float32 # 3. Load Model with automatic fallback handling try: self.model = AutoModelForCausalLM.from_pretrained(target_model_id, **model_kwargs) except Exception as e: console.print(f"[yellow]Retrying model load without optional acceleration flags: {e}[/yellow]") model_kwargs.pop("attn_implementation", None) model_kwargs.pop("quantization_config", None) if self.device != "cuda": model_kwargs.pop("device_map", None) self.model = AutoModelForCausalLM.from_pretrained(target_model_id, **model_kwargs) if "device_map" not in model_kwargs and self.device in ("cuda", "mps", "cpu"): try: self.model.to(self.device) except Exception: pass self.model.eval() self.model_id = target_model_id self._is_loaded = True console.print(f"[bold green]✓ Nova Astra Flagship '{target_model_id}' ready on {self.device.upper()}![/bold green]") def is_ready(self) -> bool: return self._is_loaded and self.model is not None and self.tokenizer is not None @spaces.GPU(duration=120) def stream_chat( self, messages: List[Dict[str, str]], gen_config: Optional[GenerationConfig] = None ) -> Generator[Tuple[str, Dict[str, Any]], None, None]: """ Stream generated tokens for multi-turn chat messages. Decorated with @spaces.GPU for seamless Hugging Face ZeroGPU NVIDIA A100 acceleration. Yields tuple: (accumulated_text, live_metrics_dict) """ if not self.is_ready(): self.load_model() cfg = gen_config or config.gen_config tracker = LatencyTracker() tracker.start() # Handle Deep Reasoning / Chain-of-Thought working_messages = list(messages) if getattr(cfg, "deep_reasoning", False): reasoning_instruction = ( "\n\n[DEEP REASONING PROTOCOL ENABLED]:\n" "You are an advanced chain-of-thought reasoning intelligence. " "Analyze the problem systematically inside ... (or ...) tags. " "1. Break down the core objective and constraints.\n" "2. Evaluate hypotheses, mathematical logic, and edge cases.\n" "3. Verify your conclusions step-by-step.\n" "After closing the thought tag, provide the clean, authoritative, well-structured final answer." ) if working_messages and working_messages[0]["role"] == "system": working_messages[0] = { "role": "system", "content": working_messages[0]["content"] + reasoning_instruction } else: working_messages.insert(0, { "role": "system", "content": "You are Nova AI (Astra Flagship Edition)." + reasoning_instruction }) # Handle 1,000+ Domain Specialization Injection active_domain_name = None try: from src.domain_registry import domain_registry domain_id = getattr(cfg, "domain_id", None) target_domain = None if domain_id: target_domain = domain_registry.get_domain(domain_id) elif getattr(cfg, "auto_detect_domain", True): user_query = next((m.get("content", "") for m in reversed(working_messages) if m.get("role") == "user"), "") target_domain = domain_registry.auto_detect_domain(user_query) if target_domain: active_domain_name = f"{target_domain.name} ({target_domain.category})" domain_instruction = ( f"\n\n[ACTIVE DOMAIN SPECIALIZATION: {target_domain.name.upper()} ({target_domain.category})]:\n" f"• Scope: {target_domain.description}\n" f"• Authority Guidelines: {target_domain.guidelines}\n" f"• Use domain-precise terminology, relevant formalisms, and industry best practices." ) if working_messages and working_messages[0]["role"] == "system": working_messages[0] = { "role": "system", "content": working_messages[0]["content"] + domain_instruction } else: working_messages.insert(0, { "role": "system", "content": "You are Nova AI (Astra Flagship Edition)." + domain_instruction }) except Exception: pass # Apply chat template try: formatted_prompt = self.tokenizer.apply_chat_template( working_messages, tokenize=False, add_generation_prompt=True ) except Exception: formatted_prompt = "" for m in working_messages: role = m.get("role", "user").capitalize() formatted_prompt += f"{role}: {m.get('content', '')}\n" formatted_prompt += "Assistant: " # Direct tensors to appropriate device target_device = self.model.device if hasattr(self.model, "device") else self.device inputs = self.tokenizer(formatted_prompt, return_tensors="pt").to(target_device) input_token_count = inputs.input_ids.shape[-1] streamer = TextIteratorStreamer( self.tokenizer, timeout=60.0, skip_prompt=True, skip_special_tokens=True ) max_tokens = int(getattr(cfg, "max_new_tokens", 1024) or 1024) generation_kwargs = dict( **inputs, streamer=streamer, max_new_tokens=max_tokens, temperature=max(cfg.temperature, 0.01), top_p=cfg.top_p, repetition_penalty=cfg.repetition_penalty, do_sample=cfg.do_sample and cfg.temperature > 0.05, pad_token_id=self.tokenizer.pad_token_id, eos_token_id=self.tokenizer.eos_token_id ) # Run token generation in background thread thread = threading.Thread(target=self.model.generate, kwargs=generation_kwargs) thread.start() accumulated_text = "" for new_token in streamer: tracker.record_token(1) accumulated_text += new_token current_metrics = tracker.stop() current_metrics["input_tokens"] = input_token_count if active_domain_name: current_metrics["active_domain"] = active_domain_name yield accumulated_text, current_metrics thread.join() final_metrics = tracker.stop() final_metrics["input_tokens"] = input_token_count if active_domain_name: final_metrics["active_domain"] = active_domain_name yield accumulated_text, final_metrics @spaces.GPU(duration=120) def generate_response( self, messages: List[Dict[str, str]], gen_config: Optional[GenerationConfig] = None ) -> Tuple[str, Dict[str, Any]]: """Non-streaming generation helper.""" full_text = "" metrics = {} for partial, cur_metrics in self.stream_chat(messages, gen_config): full_text = partial metrics = cur_metrics return full_text, metrics # Global shared engine instance engine = NovaAstraEngine() # Enterprise & Backwards compatibility aliases FlagshipChatEngine = NovaAstraEngine LightweightChatEngine = NovaAstraEngine