Upload utils.py with huggingface_hub

utils.py

@@ -0,0 +1,564 @@
+"""
+Utility functions for the Tiny Conversational AI.
+Helper functions for system detection, logging, caching, and more.
+"""
+
+import os
+import sys
+import time
+import json
+import hashlib
+import logging
+import platform
+import threading
+from pathlib import Path
+from typing import Optional, Dict, Any, Callable, List
+from functools import wraps, lru_cache
+from datetime import datetime, timedelta
+from collections import OrderedDict
+from contextlib import contextmanager
+
+# Try to import psutil for memory monitoring
+try:
+    import psutil
+    HAS_PSUTIL = True
+except ImportError:
+    HAS_PSUTIL = False
+
+
+# =============================================================================
+# LOGGING
+# =============================================================================
+
+def setup_logging(
+    log_level: str = "INFO",
+    log_file: Optional[str] = None,
+    log_format: Optional[str] = None
+) -> logging.Logger:
+    """Setup and configure logging."""
+    
+    if log_format is None:
+        log_format = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
+    
+    logger = logging.getLogger("tiny_ai")
+    logger.setLevel(getattr(logging, log_level.upper()))
+    
+    # Console handler
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setFormatter(logging.Formatter(log_format))
+    logger.addHandler(console_handler)
+    
+    # File handler (optional)
+    if log_file:
+        file_handler = logging.FileHandler(log_file, encoding='utf-8')
+        file_handler.setFormatter(logging.Formatter(log_format))
+        logger.addHandler(file_handler)
+    
+    return logger
+
+
+def get_logger(name: str = "tiny_ai") -> logging.Logger:
+    """Get a logger instance."""
+    return logging.getLogger(name)
+
+
+# =============================================================================
+# SYSTEM DETECTION
+# =============================================================================
+
+@lru_cache(maxsize=1)
+def get_system_info() -> Dict[str, Any]:
+    """Get comprehensive system information."""
+    info = {
+        "platform": platform.system(),
+        "platform_release": platform.release(),
+        "platform_version": platform.version(),
+        "architecture": platform.machine(),
+        "processor": platform.processor(),
+        "python_version": platform.python_version(),
+        "cpu_count": os.cpu_count() or 1,
+        "cpu_count_physical": None,
+        "total_ram_gb": None,
+        "available_ram_gb": None,
+        "has_avx": False,
+        "has_avx2": False,
+        "has_avx512": False,
+    }
+    
+    if HAS_PSUTIL:
+        try:
+            info["cpu_count_physical"] = psutil.cpu_count(logical=False)
+            mem = psutil.virtual_memory()
+            info["total_ram_gb"] = round(mem.total / (1024**3), 2)
+            info["available_ram_gb"] = round(mem.available / (1024**3), 2)
+        except Exception:
+            pass
+    
+    # Detect CPU features (AVX support improves llama.cpp performance)
+    try:
+        if platform.system() == "Windows":
+            # On Windows, check via processor info
+            proc_info = platform.processor().lower()
+            info["has_avx"] = True  # Most modern CPUs have AVX
+            info["has_avx2"] = "intel" in proc_info or "amd" in proc_info
+        else:
+            # On Linux, check /proc/cpuinfo
+            if os.path.exists("/proc/cpuinfo"):
+                with open("/proc/cpuinfo", "r") as f:
+                    cpuinfo = f.read().lower()
+                    info["has_avx"] = "avx" in cpuinfo
+                    info["has_avx2"] = "avx2" in cpuinfo
+                    info["has_avx512"] = "avx512" in cpuinfo
+    except Exception:
+        pass
+    
+    return info
+
+
+def get_available_ram_gb() -> float:
+    """Get available RAM in gigabytes."""
+    if HAS_PSUTIL:
+        return psutil.virtual_memory().available / (1024**3)
+    return 4.0  # Assume 4GB if psutil not available
+
+
+def get_memory_usage_gb() -> float:
+    """Get current process memory usage in GB."""
+    if HAS_PSUTIL:
+        process = psutil.Process(os.getpid())
+        return process.memory_info().rss / (1024**3)
+    return 0.0
+
+
+def get_optimal_thread_count() -> int:
+    """Determine optimal thread count for inference."""
+    cpu_count = os.cpu_count() or 4
+    
+    if HAS_PSUTIL:
+        physical_cores = psutil.cpu_count(logical=False)
+        if physical_cores:
+            # Use physical cores for best performance
+            return max(1, physical_cores - 1)
+    
+    # Fallback: use half of logical cores
+    return max(1, cpu_count // 2)
+
+
+def check_system_requirements(min_ram_gb: float = 2.0) -> Dict[str, Any]:
+    """Check if system meets minimum requirements."""
+    info = get_system_info()
+    
+    result = {
+        "meets_requirements": True,
+        "warnings": [],
+        "errors": [],
+        "recommendations": [],
+    }
+    
+    # Check RAM
+    if info["available_ram_gb"] and info["available_ram_gb"] < min_ram_gb:
+        result["meets_requirements"] = False
+        result["errors"].append(
+            f"Insufficient RAM: {info['available_ram_gb']:.1f}GB available, "
+            f"{min_ram_gb}GB required"
+        )
+        result["recommendations"].append("Close other applications to free memory")
+    
+    # Check Python version
+    py_version = tuple(map(int, info["python_version"].split(".")[:2]))
+    if py_version < (3, 8):
+        result["meets_requirements"] = False
+        result["errors"].append(f"Python 3.8+ required, found {info['python_version']}")
+    
+    # Performance warnings
+    if info["cpu_count"] and info["cpu_count"] < 4:
+        result["warnings"].append("Low CPU core count may result in slower responses")
+    
+    if not info.get("has_avx"):
+        result["warnings"].append("CPU may not support AVX instructions (slower inference)")
+    
+    return result
+
+
+# =============================================================================
+# CACHING
+# =============================================================================
+
+class LRUCache:
+    """Thread-safe LRU cache with TTL support."""
+    
+    def __init__(self, max_size: int = 100, ttl_seconds: int = 3600):
+        self.max_size = max_size
+        self.ttl_seconds = ttl_seconds
+        self.cache: OrderedDict = OrderedDict()
+        self.timestamps: Dict[str, datetime] = {}
+        self.lock = threading.Lock()
+    
+    def _make_key(self, *args, **kwargs) -> str:
+        """Create a hash key from arguments."""
+        key_str = json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True, default=str)
+        return hashlib.md5(key_str.encode()).hexdigest()
+    
+    def get(self, key: str) -> Optional[Any]:
+        """Get value from cache if exists and not expired."""
+        with self.lock:
+            if key not in self.cache:
+                return None
+            
+            # Check TTL
+            if datetime.now() - self.timestamps[key] > timedelta(seconds=self.ttl_seconds):
+                del self.cache[key]
+                del self.timestamps[key]
+                return None
+            
+            # Move to end (most recently used)
+            self.cache.move_to_end(key)
+            return self.cache[key]
+    
+    def set(self, key: str, value: Any):
+        """Set value in cache."""
+        with self.lock:
+            if key in self.cache:
+                self.cache.move_to_end(key)
+            else:
+                if len(self.cache) >= self.max_size:
+                    # Remove oldest item
+                    oldest = next(iter(self.cache))
+                    del self.cache[oldest]
+                    del self.timestamps[oldest]
+            
+            self.cache[key] = value
+            self.timestamps[key] = datetime.now()
+    
+    def clear(self):
+        """Clear the cache."""
+        with self.lock:
+            self.cache.clear()
+            self.timestamps.clear()
+    
+    def stats(self) -> Dict[str, Any]:
+        """Get cache statistics."""
+        with self.lock:
+            return {
+                "size": len(self.cache),
+                "max_size": self.max_size,
+                "ttl_seconds": self.ttl_seconds,
+            }
+
+
+def cached(cache: LRUCache):
+    """Decorator to cache function results."""
+    def decorator(func: Callable):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            key = cache._make_key(func.__name__, *args, **kwargs)
+            result = cache.get(key)
+            if result is not None:
+                return result
+            
+            result = func(*args, **kwargs)
+            cache.set(key, result)
+            return result
+        return wrapper
+    return decorator
+
+
+# =============================================================================
+# TIMING AND PERFORMANCE
+# =============================================================================
+
+class Timer:
+    """Simple timer for measuring execution time."""
+    
+    def __init__(self, name: str = ""):
+        self.name = name
+        self.start_time: Optional[float] = None
+        self.end_time: Optional[float] = None
+        self.elapsed: float = 0.0
+    
+    def start(self):
+        """Start the timer."""
+        self.start_time = time.perf_counter()
+        return self
+    
+    def stop(self) -> float:
+        """Stop the timer and return elapsed time."""
+        self.end_time = time.perf_counter()
+        self.elapsed = self.end_time - (self.start_time or self.end_time)
+        return self.elapsed
+    
+    def __enter__(self):
+        self.start()
+        return self
+    
+    def __exit__(self, *args):
+        self.stop()
+
+
+@contextmanager
+def measure_time(name: str = "", logger: Optional[logging.Logger] = None):
+    """Context manager to measure and log execution time."""
+    timer = Timer(name)
+    timer.start()
+    try:
+        yield timer
+    finally:
+        timer.stop()
+        if logger:
+            logger.debug(f"{name}: {timer.elapsed:.4f}s")
+
+
+class PerformanceTracker:
+    """Track performance metrics over time."""
+    
+    def __init__(self, window_size: int = 100):
+        self.window_size = window_size
+        self.metrics: Dict[str, List[float]] = {}
+        self.lock = threading.Lock()
+    
+    def record(self, metric_name: str, value: float):
+        """Record a metric value."""
+        with self.lock:
+            if metric_name not in self.metrics:
+                self.metrics[metric_name] = []
+            
+            self.metrics[metric_name].append(value)
+            
+            # Keep only recent values
+            if len(self.metrics[metric_name]) > self.window_size:
+                self.metrics[metric_name] = self.metrics[metric_name][-self.window_size:]
+    
+    def get_average(self, metric_name: str) -> Optional[float]:
+        """Get average value for a metric."""
+        with self.lock:
+            if metric_name not in self.metrics or not self.metrics[metric_name]:
+                return None
+            return sum(self.metrics[metric_name]) / len(self.metrics[metric_name])
+    
+    def get_stats(self, metric_name: str) -> Dict[str, Any]:
+        """Get statistics for a metric."""
+        with self.lock:
+            if metric_name not in self.metrics or not self.metrics[metric_name]:
+                return {}
+            
+            values = self.metrics[metric_name]
+            return {
+                "count": len(values),
+                "min": min(values),
+                "max": max(values),
+                "avg": sum(values) / len(values),
+                "last": values[-1],
+            }
+    
+    def get_all_stats(self) -> Dict[str, Dict[str, Any]]:
+        """Get statistics for all metrics."""
+        return {name: self.get_stats(name) for name in self.metrics}
+
+
+# =============================================================================
+# TEXT PROCESSING
+# =============================================================================
+
+def count_tokens_approx(text: str) -> int:
+    """Approximate token count (roughly 4 characters per token)."""
+    return len(text) // 4 + 1
+
+
+def truncate_text(text: str, max_tokens: int) -> str:
+    """Truncate text to approximate token limit."""
+    max_chars = max_tokens * 4
+    if len(text) <= max_chars:
+        return text
+    
+    # Try to truncate at word boundary
+    truncated = text[:max_chars]
+    last_space = truncated.rfind(" ")
+    if last_space > max_chars * 0.8:
+        truncated = truncated[:last_space]
+    
+    return truncated + "..."
+
+
+def clean_response(text: str) -> str:
+    """Clean up model response text."""
+    # Remove common artifacts
+    text = text.strip()
+    
+    # Remove chat template tokens that leaked through
+    for token in ["<|system|>", "<|user|>", "<|assistant|>", "</s>", "<|end|>", "<s>"]:
+        text = text.replace(token, "")
+    
+    # Remove repeated punctuation
+    while "  " in text:
+        text = text.replace("  ", " ")
+    
+    text = text.strip()
+    return text
+
+
+def extract_entities(text: str) -> Dict[str, List[str]]:
+    """Extract simple entities from text (names, topics, etc.)."""
+    entities = {
+        "names": [],
+        "topics": [],
+        "pronouns": [],
+    }
+    
+    # Simple pronoun detection
+    pronouns = ["it", "they", "them", "this", "that", "these", "those", "he", "she"]
+    words = text.lower().split()
+    entities["pronouns"] = [w for w in words if w in pronouns]
+    
+    # Detect capitalized words as potential names (simple heuristic)
+    import re
+    capitalized = re.findall(r'\b[A-Z][a-z]+\b', text)
+    # Filter out common sentence starters
+    common_starters = {"I", "The", "A", "An", "This", "That", "What", "How", "Why", "When", "Where"}
+    entities["names"] = [w for w in capitalized if w not in common_starters]
+    
+    return entities
+
+
+# =============================================================================
+# PROGRESS DISPLAY
+# =============================================================================
+
+class ProgressBar:
+    """Simple progress bar for terminal output."""
+    
+    def __init__(self, total: int, prefix: str = "", width: int = 40):
+        self.total = total
+        self.prefix = prefix
+        self.width = width
+        self.current = 0
+        self.start_time = time.time()
+    
+    def update(self, current: Optional[int] = None, increment: int = 1):
+        """Update progress bar."""
+        if current is not None:
+            self.current = current
+        else:
+            self.current += increment
+        
+        self._display()
+    
+    def _display(self):
+        """Display the progress bar."""
+        progress = self.current / self.total if self.total > 0 else 1
+        filled = int(self.width * progress)
+        bar = "█" * filled + "░" * (self.width - filled)
+        percent = progress * 100
+        
+        # Estimate remaining time
+        elapsed = time.time() - self.start_time
+        if self.current > 0:
+            eta = elapsed * (self.total - self.current) / self.current
+            eta_str = f"ETA: {eta:.0f}s"
+        else:
+            eta_str = "ETA: --"
+        
+        print(f"\r{self.prefix} |{bar}| {percent:.1f}% {eta_str}", end="", flush=True)
+    
+    def finish(self):
+        """Complete the progress bar."""
+        self.current = self.total
+        self._display()
+        print()
+
+
+def print_banner():
+    """Print application banner."""
+    banner = """
+╔══════════════════════════════════════════════════════════════╗
+║                  🤖 Tiny Conversational AI                   ║
+║                  Fast • Lightweight • Local                  ║
+╚══════════════════════════════════════════════════════════════╝
+    """
+    print(banner)
+
+
+def print_system_status():
+    """Print current system status."""
+    info = get_system_info()
+    
+    print("\n📊 System Status:")
+    print(f"   • Platform: {info['platform']} {info['platform_release']}")
+    print(f"   • CPU: {info['processor'][:50]}...")
+    print(f"   • Cores: {info['cpu_count']} logical, {info.get('cpu_count_physical', 'N/A')} physical")
+    
+    if info["total_ram_gb"]:
+        print(f"   • RAM: {info['available_ram_gb']:.1f}GB available / {info['total_ram_gb']:.1f}GB total")
+    
+    print(f"   • Python: {info['python_version']}")
+    print()
+
+
+# =============================================================================
+# FILE OPERATIONS
+# =============================================================================
+
+def ensure_dir(path: Path) -> Path:
+    """Ensure directory exists, create if needed."""
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def safe_json_load(path: Path, default: Any = None) -> Any:
+    """Safely load JSON file, return default on error."""
+    try:
+        with open(path, 'r', encoding='utf-8') as f:
+            return json.load(f)
+    except (FileNotFoundError, json.JSONDecodeError):
+        return default
+
+
+def safe_json_save(path: Path, data: Any):
+    """Safely save data to JSON file."""
+    ensure_dir(path.parent)
+    with open(path, 'w', encoding='utf-8') as f:
+        json.dump(data, f, indent=2, default=str)
+
+
+# =============================================================================
+# DOWNLOAD HELPERS
+# =============================================================================
+
+def download_with_progress(url: str, dest_path: Path, chunk_size: int = 8192) -> bool:
+    """Download a file with progress bar."""
+    try:
+        import requests
+        
+        response = requests.get(url, stream=True)
+        response.raise_for_status()
+        
+        total_size = int(response.headers.get('content-length', 0))
+        
+        ensure_dir(dest_path.parent)
+        
+        progress = ProgressBar(
+            total=total_size,
+            prefix=f"Downloading {dest_path.name}",
+        )
+        
+        downloaded = 0
+        with open(dest_path, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=chunk_size):
+                if chunk:
+                    f.write(chunk)
+                    downloaded += len(chunk)
+                    progress.update(downloaded)
+        
+        progress.finish()
+        return True
+        
+    except Exception as e:
+        get_logger().error(f"Download failed: {e}")
+        return False
+
+
+# =============================================================================
+# INITIALIZATION
+# =============================================================================
+
+# Setup default logger
+_default_logger = setup_logging()