Spaces:

maubilli
/

best-local-model

Running

best-local-model / src /agents /bestLocalModel.py
Update with Qwen 3.8 27B, Qwen 3, and live Hugging Face model explorer
6c39dcf verified 15 days ago
37.5 kB
"""Universal Local LLM Hardware Recommender & Hugging Face Model Discovery Agent.
Detects the host machine's hardware (OS, CPU, RAM, GPU, VRAM, Unified Memory)
cross-platform (macOS Apple Silicon/Intel, Linux CUDA/ROCm/CPU, Windows) and
dynamically fetches the latest open-weight models from Hugging Face
(Qwen 3.8, Qwen 3, Llama 3.3, DeepSeek-R1, etc.) to calculate the optimal
models the local system can run with peak performance.
"""

from __future__ import annotations

import argparse
import json
import os
import platform
import re
import shutil
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
from typing import Any

# Ensure parent directory is on sys.path for direct script execution
_REPO_ROOT = Path(__file__).resolve().parent.parent.parent
_SRC_ROOT = Path(__file__).resolve().parent.parent
if str(_SRC_ROOT) not in sys.path:
    sys.path.insert(0, str(_SRC_ROOT))
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from dotenv import load_dotenv
from smolagents import (
    CodeAgent,
    DuckDuckGoSearchTool,
    InferenceClientModel,
    OpenAIServerModel,
    tool,
)

from agents.telemetry import flush_telemetry, init_telemetry

load_dotenv()

# Cache settings for Hugging Face models
CACHE_DIR = Path.home() / ".cache" / "best_local_model"
HF_CACHE_FILE = CACHE_DIR / "hf_models.json"
HF_CACHE_TTL_SECONDS = 12 * 3600  # 12 hours


def detect_system_specs() -> dict[str, Any]:
    """Inspects host hardware specifications across macOS, Linux, and Windows."""
    specs: dict[str, Any] = {
        "os": platform.system(),
        "os_release": platform.release(),
        "arch": platform.machine(),
        "cpu_name": platform.processor() or "Generic CPU",
        "cpu_cores": os.cpu_count() or 1,
        "total_ram_gb": 0.0,
        "gpu_type": "CPU Only",
        "gpu_name": "Integrated / CPU",
        "vram_gb": 0.0,
        "is_apple_silicon": False,
        "is_unified_memory": False,
        "usable_llm_memory_gb": 0.0,
        "ollama_running": False,
        "installed_ollama_models": [],
        "installed_model_details": {},
    }

    system_os = specs["os"]

    # 1. Total System RAM Detection
    if system_os == "Darwin":
        try:
            mem_bytes = int(subprocess.check_output(["sysctl", "-n", "hw.memsize"]).strip())
            specs["total_ram_gb"] = round(mem_bytes / (1024**3), 1)
        except Exception:
            pass

        try:
            brand = (
                subprocess.check_output(["sysctl", "-n", "machdep.cpu.brand_string"])
                .decode()
                .strip()
            )
            if brand:
                specs["cpu_name"] = brand
        except Exception:
            pass

        # Check for Apple Silicon (M1/M2/M3/M4)
        if specs["arch"] in ("arm64", "aarch64"):
            specs["is_apple_silicon"] = True
            specs["is_unified_memory"] = True
            specs["gpu_type"] = "Apple Silicon GPU (Metal Unified Memory)"
            specs["gpu_name"] = specs["cpu_name"]
            # macOS default allocations reserve ~20-25% for OS; up to ~75% can be allocated to GPU VRAM
            specs["vram_gb"] = round(specs["total_ram_gb"] * 0.75, 1)
            specs["usable_llm_memory_gb"] = specs["vram_gb"]

    elif system_os == "Linux":
        try:
            with open("/proc/meminfo") as f:
                for line in f:
                    if line.startswith("MemTotal:"):
                        parts = line.split()
                        kb = float(parts[1])
                        specs["total_ram_gb"] = round(kb / (1024**2), 1)
                        break
        except Exception:
            try:
                pages = os.sysconf("SC_PHYS_PAGES")
                page_size = os.sysconf("SC_PAGE_SIZE")
                specs["total_ram_gb"] = round((pages * page_size) / (1024**3), 1)
            except Exception:
                pass

        try:
            with open("/proc/cpuinfo") as f:
                for line in f:
                    if "model name" in line:
                        specs["cpu_name"] = line.split(":", 1)[1].strip()
                        break
        except Exception:
            pass

    elif system_os == "Windows":
        try:
            import ctypes

            class MEMORYSTATUSEX(ctypes.Structure):
                _fields_ = [
                    ("dwLength", ctypes.c_ulong),
                    ("dwMemoryLoad", ctypes.c_ulong),
                    ("ullTotalPhys", ctypes.c_ulonglong),
                    ("ullAvailPhys", ctypes.c_ulonglong),
                    ("ullTotalPageFile", ctypes.c_ulonglong),
                    ("ullAvailPageFile", ctypes.c_ulonglong),
                    ("ullTotalVirtual", ctypes.c_ulonglong),
                    ("ullAvailVirtual", ctypes.c_ulonglong),
                    ("sullAvailExtendedVirtual", ctypes.c_ulonglong),
                ]

            stat = MEMORYSTATUSEX()
            stat.dwLength = ctypes.sizeof(MEMORYSTATUSEX)
            ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(stat))
            specs["total_ram_gb"] = round(stat.ullTotalPhys / (1024**3), 1)
        except Exception:
            pass

    # Optional psutil fallback if available
    if specs["total_ram_gb"] == 0.0:
        try:
            import psutil

            specs["total_ram_gb"] = round(psutil.virtual_memory().total / (1024**3), 1)
        except ImportError:
            specs["total_ram_gb"] = 8.0  # Safe default guess

    # 2. Discrete GPU Detection (NVIDIA / AMD) for non-Apple-Silicon systems
    if not specs["is_apple_silicon"]:
        if shutil.which("nvidia-smi"):
            try:
                out = (
                    subprocess.check_output(
                        [
                            "nvidia-smi",
                            "--query-gpu=gpu_name,memory.total",
                            "--format=csv,noheader,nounits",
                        ],
                        stderr=subprocess.DEVNULL,
                    )
                    .decode()
                    .strip()
                )
                if out:
                    lines = out.splitlines()
                    names: list[str] = []
                    total_vram_mb = 0.0
                    for line in lines:
                        parts = [p.strip() for p in line.split(",")]
                        if len(parts) >= 2:
                            names.append(parts[0])
                            total_vram_mb += float(parts[1])
                    specs["gpu_type"] = "NVIDIA CUDA"
                    specs["gpu_name"] = ", ".join(names)
                    specs["vram_gb"] = round(total_vram_mb / 1024, 1)
                    specs["usable_llm_memory_gb"] = max(0.0, round(specs["vram_gb"] - 1.5, 1))
            except Exception:
                pass

        if specs["gpu_type"] == "CPU Only":
            specs["usable_llm_memory_gb"] = max(1.0, round(specs["total_ram_gb"] - 3.0, 1))

    # 3. Check Local Ollama Server Status & Installed Models
    try:
        req = urllib.request.Request(
            "http://localhost:11434/api/tags", headers={"User-Agent": "bestLocalModel/2.0"}
        )
        with urllib.request.urlopen(req, timeout=1.5) as resp:
            if resp.status == 200:
                data = json.loads(resp.read().decode())
                specs["ollama_running"] = True
                models_list = data.get("models", [])
                specs["installed_ollama_models"] = [
                    m.get("name") for m in models_list if m.get("name")
                ]

                # Inspect details of each installed model via /api/show
                for m_info in models_list:
                    m_name = m_info.get("name")
                    if not m_name:
                        continue
                    try:
                        show_req = urllib.request.Request(
                            "http://localhost:11434/api/show",
                            data=json.dumps({"name": m_name}).encode(),
                            headers={
                                "Content-Type": "application/json",
                                "User-Agent": "bestLocalModel/2.0",
                            },
                        )
                        with urllib.request.urlopen(show_req, timeout=1.5) as show_resp:
                            if show_resp.status == 200:
                                show_data = json.loads(show_resp.read().decode())
                                specs["installed_model_details"][m_name] = {
                                    "size_bytes": m_info.get("size", 0),
                                    "details": show_data.get("details", {}),
                                    "parameters": show_data.get("details", {}).get(
                                        "parameter_size", "Unknown"
                                    ),
                                    "quantization": show_data.get("details", {}).get(
                                        "quantization_level", "Unknown"
                                    ),
                                    "family": show_data.get("details", {}).get(
                                        "family", "Unknown"
                                    ),
                                }
                    except Exception:
                        pass
    except Exception:
        specs["ollama_running"] = False
        specs["installed_ollama_models"] = []
        specs["installed_model_details"] = {}

    return specs


def parse_hf_model_info(raw: dict[str, Any]) -> dict[str, Any]:
    """Extracts structured metadata (parameters, generation, VRAM estimate) from a Hugging Face model item."""
    mid = raw.get("id", "")
    name = mid.split("/")[-1]
    name_lower = name.lower()

    # Match parameter size: e.g. 27B, 14B, 8B, 4B, 3.8B, 0.6B, 70B
    size_match = re.search(r"(\d+(?:\.\d+)?)b\b", name_lower)
    size_b = float(size_match.group(1)) if size_match else None

    # Family & Generation detection
    family = "Other"
    gen = "Other"
    if "qwen3.8" in name_lower or "qwen-3.8" in name_lower:
        family, gen = "Qwen", "Qwen 3.8"
    elif "qwen3.6" in name_lower:
        family, gen = "Qwen", "Qwen 3.6"
    elif "qwen3" in name_lower or "qwen-3" in name_lower:
        family, gen = "Qwen", "Qwen 3.0"
    elif "qwen2.5" in name_lower:
        family, gen = "Qwen", "Qwen 2.5"
    elif "deepseek-r1" in name_lower:
        family, gen = "DeepSeek", "DeepSeek R1"
    elif "llama-3.3" in name_lower:
        family, gen = "Llama", "Llama 3.3"
    elif "llama-3.2" in name_lower:
        family, gen = "Llama", "Llama 3.2"
    elif "llama-3.1" in name_lower:
        family, gen = "Llama", "Llama 3.1"
    elif "gemma-2" in name_lower or "gemma-3" in name_lower:
        family, gen = "Gemma", "Gemma"
    elif "glm" in name_lower:
        family, gen = "GLM", "GLM"

    # Category detection
    category = "General"
    if any(k in name_lower for k in ("coder", "code")):
        category = "Coding"
    elif any(k in name_lower for k in ("r1", "reason", "math", "thinking", "heretic")):
        category = "Reasoning"
    elif any(k in name_lower for k in ("instruct", "chat")):
        category = "Instruct/Chat"

    # Estimated VRAM at Q4_K_M (GB): ~0.65 GB per billion parameters + ~1.2 GB context/overhead
    est_vram_q4 = round(size_b * 0.65 + 1.2, 1) if size_b is not None else None

    return {
        "id": mid,
        "name": name,
        "family": family,
        "gen": gen,
        "size_b": size_b,
        "category": category,
        "est_vram_q4": est_vram_q4,
        "downloads": raw.get("downloads", 0),
        "likes": raw.get("likes", 0),
        "hf_url": f"https://huggingface.co/{mid}",
    }


def fetch_latest_huggingface_models(force_refresh: bool = False) -> list[dict[str, Any]]:
    """Fetches the latest and trending open-weight LLMs from Hugging Face API with local disk caching."""
    if not force_refresh and HF_CACHE_FILE.exists():
        try:
            with open(HF_CACHE_FILE, "r", encoding="utf-8") as f:
                cached = json.load(f)
            cached_time = cached.get("timestamp", 0)
            if time.time() - cached_time < HF_CACHE_TTL_SECONDS:
                raw_list = cached.get("models", [])
                return [m if "size_b" in m else parse_hf_model_info(m) for m in raw_list]
        except Exception:
            pass

    # Query Hugging Face models API across leading model families & trending
    endpoints = [
        "pipeline_tag=text-generation&sort=trendingScore&direction=-1&limit=30",
        "search=qwen3.8&pipeline_tag=text-generation&sort=downloads&direction=-1&limit=20",
        "search=qwen3&pipeline_tag=text-generation&sort=downloads&direction=-1&limit=25",
        "search=qwen&pipeline_tag=text-generation&sort=downloads&direction=-1&limit=25",
        "search=deepseek-r1&pipeline_tag=text-generation&sort=downloads&direction=-1&limit=15",
        "search=llama-3&pipeline_tag=text-generation&sort=downloads&direction=-1&limit=15",
    ]

    seen: dict[str, dict[str, Any]] = {}
    for ep in endpoints:
        url = f"https://huggingface.co/api/models?{ep}"
        req = urllib.request.Request(url, headers={"User-Agent": "bestLocalModel/2.0"})
        try:
            with urllib.request.urlopen(req, timeout=5) as r:
                data = json.loads(r.read().decode())
                for item in data:
                    mid = item.get("id")
                    if mid and mid not in seen:
                        parsed = parse_hf_model_info(item)
                        seen[mid] = parsed
        except Exception:
            continue

    models = list(seen.values())

    # Fallback default models if offline / no models fetched
    if not models:
        defaults = [
            {"id": "JonathanColetti/Qwen3.8-27B-Uncensored-GGUF", "downloads": 2444687, "likes": 1575},
            {"id": "ukisai/Swift-1.5-Qwen3.8-27B-GSQ-RCO-GGUF", "downloads": 868386, "likes": 354},
            {"id": "Qwen/Qwen3-14B", "downloads": 6546332, "likes": 504},
            {"id": "Qwen/Qwen3-8B", "downloads": 9880368, "likes": 2115},
            {"id": "unsloth/Qwen3-Coder-30B-A3B-Instruct-GGUF", "downloads": 3864575, "likes": 1149},
            {"id": "Qwen/Qwen3-4B", "downloads": 12626778, "likes": 759},
            {"id": "casperhansen/llama-3.3-70b-instruct-awq", "downloads": 3950000, "likes": 890},
            {"id": "deepseek-ai/DeepSeek-R1-0528-Qwen3-8B", "downloads": 1200000, "likes": 1400},
            {"id": "deepseek-ai/DeepSeek-R1-Distill-Qwen-14B", "downloads": 2800000, "likes": 1900},
        ]
        models = [parse_hf_model_info(d) for d in defaults]

    # Save to cache
    try:
        CACHE_DIR.mkdir(parents=True, exist_ok=True)
        with open(HF_CACHE_FILE, "w", encoding="utf-8") as f:
            json.dump({"timestamp": time.time(), "models": models}, f, indent=2)
    except Exception:
        pass

    return models


def format_recommendation_report(
    specs: dict[str, Any] | None = None,
    focus: str = "balanced",
    force_refresh: bool = False,
) -> str:
    """Builds a comprehensive hardware analysis, model recommendation report, and live Hugging Face model catalog."""
    if specs is None:
        specs = detect_system_specs()

    usable_gb = specs.get("usable_llm_memory_gb", 0.0)
    total_ram = specs.get("total_ram_gb", 0.0)
    cpu_name = specs.get("cpu_name", "Unknown CPU")
    cpu_cores = specs.get("cpu_cores", 1)
    os_name = f"{specs.get('os')} ({specs.get('arch')})"
    gpu_type = specs.get("gpu_type", "CPU")
    gpu_name = specs.get("gpu_name", "CPU")
    is_apple_silicon = specs.get("is_apple_silicon", False)
    ollama_running = specs.get("ollama_running", False)
    installed = specs.get("installed_ollama_models", [])
    model_details = specs.get("installed_model_details", {})

    # Fetch latest models from Hugging Face
    hf_models = fetch_latest_huggingface_models(force_refresh=force_refresh)

    # Hardware Tiering Matrix
    if usable_gb < 6.0:
        tier_title = "Ultra-Lightweight / Edge (Memory Budget < 6 GB)"
        best_pick = (
            "Qwen 3 4B Instruct (or Qwen 2.5 3B)",
            "4B / 3B",
            "Next-generation reasoning & language depth in an ultra-compact footprint.",
        )
        best_chat = (
            "Qwen 3 4B Instruct",
            "ollama run qwen3:4b",
            "Fast 80+ tok/s response, next-gen architecture",
        )
        best_code = (
            "Qwen 3 4B / Qwen 2.5 Coder 1.5B",
            "ollama run qwen3:4b",
            "Instant local code autocomplete and debugging",
        )
        best_reason = (
            "DeepSeek-R1-Distill-Qwen-1.5B",
            "ollama run deepseek-r1:1.5b",
            "Step-by-step chain-of-thought math and logic",
        )
        best_fast = ("Qwen 3 1.7B / SmolLM2 1.7B", "ollama run qwen3:1.7b", "Minimal RAM footprint")
        quant_advice = "Q4_K_M (uses ~2.5 - 3.8 GB RAM). Safe context: 8,192 tokens."

    elif usable_gb < 12.0:
        tier_title = "Entry-to-Mid (6 GB - 12 GB Budget, e.g. 8-16 GB Mac or RTX 3060/4060)"
        best_pick = (
            "Qwen 3 8B Instruct",
            "8B",
            "The premier sweet spot: beats Qwen 2.5 7B & Llama 3.1 8B across reasoning, coding, and comprehension.",
        )
        best_chat = (
            "Qwen 3 8B Instruct",
            "ollama run qwen3:8b",
            "Frontier-grade instruction following and multilingual support",
        )
        best_code = (
            "Qwen 3 8B / Qwen 2.5 Coder 7B",
            "ollama run qwen3:8b",
            "High-precision code completion and refactoring",
        )
        best_reason = (
            "DeepSeek-R1-0528-Qwen3-8B / R1-Distill-7B",
            "ollama run deepseek-r1:7b",
            "High-scoring deep chain-of-thought reasoning",
        )
        best_fast = (
            "Qwen 3 4B Instruct",
            "ollama run qwen3:4b",
            "Ultra-fast lightweight fallback",
        )
        quant_advice = "Q4_K_M (~6.4 GB VRAM) for peak speed, or Q8_0 (~10.0 GB). Safe context: 16,384 tokens."

    elif usable_gb < 24.0:
        tier_title = "Mid-to-High (12 GB - 24 GB Budget, e.g. 16-32 GB Mac or RTX 3090/4090)"
        best_pick = (
            "Qwen 3 14B Instruct (or Qwen 3.8 27B with Q4/wired memory)",
            "14B / 27B",
            "Exceptional depth of thought; Qwen 3 14B runs at native speed, while Qwen 3.8 27B provides frontier intelligence.",
        )
        best_chat = (
            "Qwen 3 14B Instruct",
            "ollama run qwen3:14b",
            "Huge leap in reasoning and coherence over previous 2.x generations",
        )
        best_code = (
            "Qwen 3 Coder 30B / Qwen 2.5 Coder 14B",
            "ollama run qwen3-coder:30b",
            "Full-repo architecture, multi-file debugging, and systems design",
        )
        best_reason = (
            "DeepSeek-R1-Distill-Qwen-14B",
            "ollama run deepseek-r1:14b",
            "Near-frontier mathematical and competitive algorithmic reasoning",
        )
        best_fast = (
            "Qwen 3 8B Instruct",
            "ollama run qwen3:8b",
            "Instantaneous answers with high intelligence",
        )
        quant_advice = "Q4_K_M (~10.3 GB VRAM) for 14B (fits effortlessly); Q4_K_M (~18 GB) for 27B. Safe context: 32,768 tokens."

    elif usable_gb < 48.0:
        tier_title = "High-End (24 GB - 48 GB Budget, e.g. 36-64 GB Mac or Dual RTX GPUs)"
        best_pick = (
            "Qwen 3.8 27B Instruct / Qwen 3 32B Instruct",
            "27B / 32B",
            "Flagship local intelligence: handles complex architectural tasks and deep reasoning.",
        )
        best_chat = (
            "Qwen 3.8 27B Instruct",
            "ollama run qwen3.8:27b",
            "State-of-the-art open-weight instruction following and writing",
        )
        best_code = (
            "Qwen 3 Coder 30B Instruct",
            "ollama run qwen3-coder:30b",
            "Industry-leading open coding model",
        )
        best_reason = (
            "DeepSeek-R1-Distill-Qwen-32B",
            "ollama run deepseek-r1:32b",
            "Top-tier dense reasoning without distillation loss",
        )
        best_fast = (
            "Qwen 3 14B Instruct",
            "ollama run qwen3:14b",
            "High-speed workhorse for rapid iteration",
        )
        quant_advice = "Q4_K_M (~18 GB) for 27B; Q4_K_M (~22 GB) for 32B. Safe context: 32k - 64k tokens."

    else:
        tier_title = "Ultra Workstation (48+ GB Budget, e.g. 64-128+ GB Mac Studio / A100 / H100)"
        best_pick = (
            "Llama 3.3 70B Instruct / Qwen 3 32B (FP16)",
            "70B / 32B",
            "Full frontier intelligence running completely locally with zero cloud dependence.",
        )
        best_chat = (
            "Llama 3.3 70B Instruct",
            "ollama run llama3.3:70b",
            "Rivals closed frontier models in general knowledge, nuance, and code",
        )
        best_code = (
            "Qwen 3 Coder 30B (FP16 or Q8_0)",
            "ollama run qwen3-coder:30b",
            "Uncompressed precision code intelligence",
        )
        best_reason = (
            "DeepSeek-R1-Distill-Llama-70B",
            "ollama run deepseek-r1:70b",
            "Massive 70B reasoning powerhouse",
        )
        best_fast = (
            "Qwen 3 14B Instruct",
            "ollama run qwen3:14b",
            "Extremely high throughput for background batch jobs",
        )
        quant_advice = "Q4_K_M (~43 GB) or Q8_0 (~75 GB). Safe context: 64,000+ tokens."

    # Engine advice
    if is_apple_silicon:
        engine_note = "On Apple Silicon, use **Ollama** for plug-and-play CLI or **MLX** (`mlx-lm`) for maximum memory bandwidth and throughput on Apple Metal."
    elif "NVIDIA" in gpu_type:
        engine_note = "On NVIDIA GPUs, use **vLLM** for maximum throughput and continuous batching, or **Ollama** for instant CLI usage."
    else:
        engine_note = "Running on CPU: use **Ollama** or **llama.cpp** with AVX2/AVX-512 thread optimization."

    # Installed models analysis & special callouts
    installed_status = (
        f"✅ Ollama is running with installed models: {', '.join(installed)}"
        if (ollama_running and installed)
        else (
            "⚠️ Ollama is running but has no models downloaded yet."
            if ollama_running
            else "ℹ️ Local Ollama server was not detected (start it with `ollama serve`)."
        )
    )

    # Check for installed Qwen 3.8 27B or other large models
    has_qwen38_27b = any("qwen3.8" in m and "27b" in m for m in installed)
    qwen38_spotlight: list[str] = []
    if has_qwen38_27b:
        details = model_details.get("qwen3.8:27b", {})
        param_str = details.get("parameters", "27.8B")
        quant_str = details.get("quantization", "nvfp4")
        qwen38_spotlight = [
            "## 🎯 Installed Model Spotlight: Qwen 3.8 27B (`qwen3.8:27b`)",
            f"- **Parameters:** {param_str} | **Quantization:** {quant_str} | **Format:** Safetensors/MLX",
            "- **Hardware Fit on Your System:**",
            f"  - Your host has **{total_ram} GB Total RAM** with a default GPU/Metal budget of **{usable_gb} GB**.",
            "  - The full 27.8B model weights require ~16.9 GiB of memory when loaded.",
            "- **How to Run Smoothly on 16 GB Apple Silicon:**",
            "  1. **Direct Launch:** `ollama run qwen3.8:27b`",
            "  2. **Memory Limit Tuning (If Ollama reports 11.3 GiB cap):**",
            "     macOS by default caps unified memory allocation to ~75% (11.3 GB). You can safely raise the wired memory limit using:",
            "     ```bash",
            "     sudo sysctl iogpu.wired_mem_limit=14336",
            "     ```",
            "  3. **High-Speed Companion:** For instantaneous 60+ tok/s code autocompletion and chat without memory pressure, pair it with `ollama run qwen3:14b` or `ollama run qwen3:8b`!",
            "",
        ]

    # Filter & rank Hugging Face models for the live table
    # Prioritize Qwen 3.8, Qwen 3, Llama 3.3, DeepSeek-R1, sorting by popularity & relevance
    hf_table_rows: list[str] = []
    seen_ids: set[str] = set()

    # Priority group 1: Qwen 3.8 series (specifically 27B and other 3.8 variants)
    qwen38_hf = [
        m
        for m in hf_models
        if "3.8" in m.get("id", "").lower() and m.get("size_b") is not None
    ]
    qwen38_hf.sort(key=lambda x: x.get("downloads", 0), reverse=True)

    # Priority group 2: Qwen 3.0 / 3.6 / Coder series
    qwen3_hf = [
        m
        for m in hf_models
        if ("qwen3" in m.get("id", "").lower() or "qwen-3" in m.get("id", "").lower())
        and "3.8" not in m.get("id", "").lower()
        and m.get("size_b") is not None
    ]
    qwen3_hf.sort(key=lambda x: x.get("downloads", 0), reverse=True)

    # Priority group 3: Other top frontier models (DeepSeek-R1, Llama 3.3, etc.)
    other_hf = [
        m
        for m in hf_models
        if "qwen" not in m.get("id", "").lower()
        and "3.8" not in m.get("id", "").lower()
        and m.get("size_b") is not None
    ]
    other_hf.sort(key=lambda x: x.get("downloads", 0), reverse=True)

    selected_hf = (qwen38_hf[:4] + qwen3_hf[:6] + other_hf[:5])

    for m in selected_hf:
        if m["id"] in seen_ids:
            continue
        seen_ids.add(m["id"])

        mid = m["id"]
        vram = m["est_vram_q4"]
        size_display = f"{m['gen']} ({m['size_b']}B)" if m["size_b"] else m["gen"]
        downloads_str = (
            f"{m['downloads'] / 1_000_000:.1f}M"
            if m["downloads"] >= 1_000_000
            else f"{m['downloads'] / 1_000:.0f}K"
            if m["downloads"] >= 1_000
            else str(m["downloads"])
        )

        # Compatibility badge
        if vram is None:
            compat = "ℹ️ Check Specs"
        elif vram <= usable_gb:
            compat = f"✅ Fits ({vram} GB)"
        elif vram <= usable_gb * 1.5:
            compat = f"⚠️ Partial Offload ({vram} GB)"
        else:
            compat = f"❌ Exceeds Memory ({vram} GB)"

        hf_table_rows.append(
            f"| [{mid}]({m['hf_url']}) | {size_display} | ~{vram or '?'} GB | {downloads_str} | {compat} |"
        )

    report_lines = [
        "## 🖥️ System Hardware Summary",
        f"- **Operating System:** {os_name}",
        f"- **Processor:** {cpu_name} ({cpu_cores} logical cores)",
        f"- **Total System RAM:** {total_ram} GB",
        f"- **GPU / Accelerator:** {gpu_name} ({gpu_type})",
        f"- **Usable Memory Budget for LLMs:** **{usable_gb} GB** ({'Unified Memory' if is_apple_silicon else 'Dedicated VRAM' if 'NVIDIA' in gpu_type else 'System RAM'})",
        f"- **Hardware Tier:** {tier_title}",
        "",
        "## 🏆 The #1 Best Model Recommendation for Your System",
        f"### **{best_pick[0]}** ({best_pick[1]})",
        f"> {best_pick[2]}",
        "",
    ]

    if qwen38_spotlight:
        report_lines.extend(qwen38_spotlight)

    report_lines.extend([
        "## 🎯 Top Picks by Category (Next-Gen Open-Weight Models)",
        "| Category | Model | Run Command | Highlights |",
        "| :--- | :--- | :--- | :--- |",
        f"| **General Assistant** | {best_chat[0]} | `{best_chat[1]}` | {best_chat[2]} |",
        f"| **Coding & Engineering** | {best_code[0]} | `{best_code[1]}` | {best_code[2]} |",
        f"| **Reasoning & Math** | {best_reason[0]} | `{best_reason[1]}` | {best_reason[2]} |",
        f"| **Fast & Lightweight** | {best_fast[0]} | `{best_fast[1]}` | {best_fast[2]} |",
        "",
        "## ⚡ Recommended Quantization & Context",
        f"- **Quantization:** {quant_advice}",
        f"- **Inference Engine Advice:** {engine_note}",
        "",
        "## 🌐 Latest Open-Weight Models on Hugging Face (Live Update)",
        "| Model Repository | Generation & Size | Est. Min RAM (Q4) | Downloads | Host Compatibility |",
        "| :--- | :--- | :--- | :--- | :--- |",
    ])
    report_lines.extend(hf_table_rows)
    report_lines.extend([
        "",
        "## 📦 Local Environment Status",
        f"- {installed_status}",
        "",
        "## 🚀 Quick Start Commands",
        "```bash",
        "# 1. Download and run the recommended #1 model:",
        f"{best_code[1] if focus == 'coding' else best_chat[1]}",
        "",
        "# 2. Or test the reasoning powerhouse:",
        f"{best_reason[1]}",
    ])

    if has_qwen38_27b:
        report_lines.extend([
            "",
            "# 3. Run your installed Qwen 3.8 27B model:",
            "ollama run qwen3.8:27b",
        ])

    report_lines.append("```")

    return "\n".join(report_lines)


@tool
def get_system_hardware_specs() -> str:
    """Detects and returns the host machine's hardware profile and memory budgets in JSON.
    Returns:
        A JSON string containing OS, CPU, RAM, GPU, VRAM, and local server status.
    """
    import json
    from agents.bestLocalModel import detect_system_specs

    specs = detect_system_specs()
    return json.dumps(specs, indent=2)


@tool
def fetch_huggingface_models_tool(force_refresh: bool = False) -> str:
    """Fetches the latest open-weight models from Hugging Face models API (https://huggingface.co/models).
    Args:
        force_refresh: If True, bypasses local cache and fetches fresh data from Hugging Face.
    Returns:
        A JSON string containing the latest models, parameter sizes, download counts, and links.
    """
    import json
    from agents.bestLocalModel import fetch_latest_huggingface_models

    models = fetch_latest_huggingface_models(force_refresh=force_refresh)
    return json.dumps(models[:20], indent=2)


@tool
def generate_recommendation_report(focus: str = "balanced", force_refresh: bool = False) -> str:
    """Generates the complete formatted hardware analysis, recommendation report, and live Hugging Face model list.
    Args:
        focus: The user's focus preference: 'balanced', 'coding', 'reasoning', or 'lightweight'.
        force_refresh: If True, fetches fresh model data from Hugging Face.
    """
    from agents.bestLocalModel import format_recommendation_report

    return format_recommendation_report(focus=focus, force_refresh=force_refresh)


def is_model_responsive(model_name: str, timeout: float = 2.0) -> bool:
    """Performs a lightweight probe to verify if a local Ollama model can be invoked."""
    try:
        req = urllib.request.Request(
            "http://localhost:11434/api/generate",
            data=json.dumps({"model": model_name, "prompt": "hi", "stream": False}).encode(),
            headers={"Content-Type": "application/json", "User-Agent": "bestLocalModel/2.0"},
        )
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.status == 200
    except Exception:
        return False


def get_agent_model():
    """Dynamically resolves the model to power the agent.
    1. If local Ollama is active and has models, tests and uses a responsive local model.
    2. Otherwise, falls back to environment-configured model or public Hugging Face model
       so anyone can run this script out-of-the-box even without a local model pre-installed.
    """
    specs = detect_system_specs()

    if specs["ollama_running"]:
        installed = specs["installed_ollama_models"]
        if installed:
            # Modern preference list prioritizing Qwen 3.8, Qwen 3, Qwen 2.5, Llama
            preferred = [
                "qwen3.8:27b",
                "qwen3.8",
                "qwen3:14b",
                "qwen3:8b",
                "qwen3",
                "qwen2.5-coder:7b",
                "qwen2.5:7b",
                "llama3.3:70b",
                "llama3.1:8b",
                "llama3.2:3b",
                "qwen2.5-coder:latest",
                "deepseek-r1:7b",
            ]

            # Find matching candidates in installed models
            candidates: list[str] = []
            for p in preferred:
                for m in installed:
                    if p in m and m not in candidates:
                        candidates.append(m)

            for m in installed:
                if m not in candidates:
                    candidates.append(m)

            # Probe candidate models to choose one that is fully responsive (not still installing or out-of-memory)
            chosen_model: str | None = None
            for cand in candidates:
                # If candidate is small (e.g. 7b/8b) or successfully passes probe
                if "coder:7b" in cand or is_model_responsive(cand, timeout=2.5):
                    chosen_model = cand
                    break

            if chosen_model is None:
                chosen_model = candidates[0]

            print(f"[Agent Model] Connected to local Ollama using model '{chosen_model}'.")
            return OpenAIServerModel(
                model_id=chosen_model,
                api_base="http://localhost:11434/v1",
                api_key="ollama",
            )

    # Fallback to agents.config get_default_model or Hugging Face InferenceClientModel
    try:
        from agents.config import get_default_model

        model = get_default_model()
        print(f"[Agent Model] Using configured model provider: {type(model).__name__}")
        return model
    except Exception:
        print("[Agent Model] Falling back to Hugging Face InferenceClientModel.")
        return InferenceClientModel(model_id="Qwen/Qwen2.5-Coder-32B-Instruct")


def main():
    """Runs the hardware advisor agent with dynamic Hugging Face discovery."""
    parser = argparse.ArgumentParser(
        description="Universal Local LLM Hardware Recommender & Hugging Face Model Discovery"
    )
    parser.add_argument(
        "--update",
        "--refresh",
        action="store_true",
        dest="force_refresh",
        help="Force fetch the latest models from Hugging Face API, bypassing local cache.",
    )
    parser.add_argument(
        "--focus",
        default="balanced",
        choices=["balanced", "coding", "reasoning", "lightweight"],
        help="Recommendation focus area (default: balanced).",
    )
    parser.add_argument(
        "--direct",
        action="store_true",
        help="Directly output the recommendation report without running the agent reasoning loop.",
    )
    args = parser.parse_args()

    print("=" * 70)
    print("🔍 Inspecting Local Hardware & Fetching Latest Hugging Face Models...")
    print("=" * 70)

    if args.force_refresh:
        print("🔄 Force-refreshing model catalog from https://huggingface.co/models ...")
        fetch_latest_huggingface_models(force_refresh=True)

    if args.direct:
        report = format_recommendation_report(focus=args.focus, force_refresh=args.force_refresh)
        print("\n" + "=" * 70)
        print("📋 Final Recommendation:")
        print("=" * 70)
        print(report)
        return

    # Initialize Langfuse telemetry
    init_telemetry(
        tags=["best-local-model", "hardware-advisor", "hf-discovery"],
        metadata={"platform": platform.system(), "arch": platform.machine()},
    )

    # Initialize the model and tools
    model = get_agent_model()

    tools = [
        get_system_hardware_specs,
        fetch_huggingface_models_tool,
        generate_recommendation_report,
        DuckDuckGoSearchTool(),
    ]

    agent = CodeAgent(
        tools=tools,
        model=model,
        additional_authorized_imports=["json", "platform", "os", "math", "re"],
    )

    task_prompt = f"""
Your goal is to inspect this user's machine, incorporate the latest open-weight models from Hugging Face, and provide the best local model recommendation.
Instructions:
1. Call `generate_recommendation_report(focus='{args.focus}', force_refresh={args.force_refresh})` to obtain the comprehensive hardware evaluation, curated model recommendation, and latest Hugging Face models list.
2. Return that report directly using `final_answer(report)`.
CRITICAL INSTRUCTION:
Do NOT construct long multiline strings or f-strings in Python code, as this causes Python syntax errors.
Simply call `generate_recommendation_report()` and pass the returned string directly to `final_answer(...)`.
"""

    try:
        result = agent.run(task_prompt)
        print("\n" + "=" * 70)
        print("📋 Final Recommendation:")
        print("=" * 70)
        print(result)
    except Exception as exc:
        print(f"\n[Note] Agent loop ended with note: {exc}. Displaying direct system analysis:\n")
        report = format_recommendation_report(focus=args.focus, force_refresh=args.force_refresh)
        print("=" * 70)
        print("📋 Final Recommendation:")
        print("=" * 70)
        print(report)
    finally:
        flush_telemetry()

    print("\n" + "=" * 70)
    print("🚀 Pushing agent to Hugging Face Hub (maubilli/BestLocalModelAgent)...")
    print("=" * 70)
    try:
        hub_url = agent.push_to_hub("maubilli/BestLocalModelAgent")
        print(f"✅ Successfully pushed to Hugging Face Hub: {hub_url}")
    except Exception as e:
        print(f"⚠️ Failed to push to Hugging Face Hub: {e}")


if __name__ == "__main__":
    main()