import hashlib import json import os import platform import re import subprocess import time import shutil import requests import sys from pathlib import Path # models LOCAL = False current_model = None OLLAMA_URL = "http://localhost:11434" OLLAMA_MODEL = "phi4-mini:latest" GROQ_API_KEY = "ENTER AN GROQ API KEY!!!" GROQ_URL = "https://api.groq.com/openai/v1/chat/completions" # route to different models when get (429 HTTP - rate limits!) GROQ_MODELS = [ "openai/gpt-oss-120b", "llama-3.3-70b-versatile", "openai/gpt-oss-20b", "qwen/qwen3-32b", "meta-llama/llama-4-scout-17b-16e-instruct", "groq/compound", "groq/compound-mini", "llama-3.1-8b-instant", "allam-2-7b", ] GROQ_MAX_TOKENS = 2048 GROQ_RATE_SLEEP = 62 # rate limit lifts # caching _cache = {} # cache same prompts # for saftey _BLOCKED = [ r"\bsystem\s*\(", r"\bpopen\s*\(", r"\bfork\s*\(", r"\bexecv?e?\s*\(", r"\bremove\s*\(", r"\bunlink\s*\(", r"\brmdir\s*\(", r"\brename\s*\(", r"\bchmod\s*\(", r"\bchown\s*\(", r"\bmkfs\b", r"\brm\s+-rf\b", ] # tools to identify hardware capabilies def _cpu_flags(): try: with open("/proc/cpuinfo") as f: for line in f: if line.lower().startswith("flags"): return set(line.split(":", 1)[1].split()) except Exception: pass return set() def hardware_context_text(): flags = _cpu_flags() simd = [] for f in ["avx2", "avx", "fma", "sse4_2", "sse4_1"]: if f in flags: simd.append(f) return ( "arch=" + str(platform.machine()) + " os=" + str(platform.platform()) + "\n" "cpu_count=" + str(os.cpu_count()) + " simd=" + (",".join(simd) if simd else "none") + "\n" "compiler=" + _compiler_driver() ) # tools to complie and run c code def _compiler_driver(): return "mpicc" if shutil.which("mpicc") else "gcc" def compiler_flags(): flags = [ "-std=c11", "-O3", "-Ofast", "-march=native", "-mtune=native", "-ffast-math", "-ffp-contract=fast", "-fno-math-errno", "-ftree-vectorize", "-ftree-slp-vectorize", "-frename-registers", "-fomit-frame-pointer", "-fopenmp", "-fopenmp-simd", "-pthread", "-pipe", "-flto", "-falign-functions=64", ] cpu = _cpu_flags() if "avx2" in cpu: flags += ["-mavx2", "-mfma"] elif "avx" in cpu: flags.append("-mavx") elif "fma" in cpu: flags.append("-mfma") return flags def compile_c(src, out): r = subprocess.run( [_compiler_driver()] + compiler_flags() + [src, "-o", out, "-lm"], capture_output=True, text=True, ) return r.returncode == 0, r.stdout + r.stderr def run_binary(path, timeout=60): try: r = subprocess.run([path], capture_output=True, text=True, timeout=timeout) return r.returncode, r.stdout, r.stderr except subprocess.TimeoutExpired as e: return 124, e.stdout or "", (e.stderr or "") + "\n[timeout]" def benchmark_binary(path, runs=3): times = [] rc = 0 out = "" err = "" for _ in range(runs): t = time.perf_counter() rc, out, err = run_binary(path) times.append(time.perf_counter() - t) return { "returncode": rc, "stdout": out, "stderr": err, "avg_seconds": sum(times) / len(times), } # block unsafe c codes def is_safe_c(source): for pat in _BLOCKED: if re.search(pat, source, re.IGNORECASE): return False, "blocked: " + pat return True, "" # llm calls def ask_llm(prompt, on_chunk=None): global current_model key = hashlib.md5(prompt.encode()).hexdigest() if key in _cache: cached = _cache[key] if on_chunk: on_chunk(cached) return cached if LOCAL: result = _ask_ollama(prompt, on_chunk) _cache[key] = result return result model_index = 0 # cycle through all models hard_fails = 0 # errors (not rate limits - if all groq models fails fallback to local) while True: model = GROQ_MODELS[model_index % len(GROQ_MODELS)] try: result = _ask_groq(prompt, on_chunk, model) current_model = model _cache[key] = result return result except requests.exceptions.HTTPError as e: status = e.response.status_code if e.response is not None else 0 if status == 429: next_model = GROQ_MODELS[(model_index + 1) % len(GROQ_MODELS)] print("\n[llm] 429 on " + model + " -> trying " + next_model, flush=True) model_index += 1 # wait until time limit lifts after full cycle through models if model_index % len(GROQ_MODELS) == 0: print("[llm] all models rate-limited; sleeping " + str(GROQ_RATE_SLEEP) + "s ...", flush=True) time.sleep(GROQ_RATE_SLEEP) hard_fails = 0 continue # non 429 errors print("\n[llm] HTTP " + str(status) + " on " + model + ": " + str(e), flush=True) hard_fails += 1 model_index += 1 if hard_fails >= len(GROQ_MODELS): print("[llm] all Groq models hard-failed; falling back to Ollama", flush=True) result = _ask_ollama(prompt, on_chunk) _cache[key] = result return result continue except Exception as e: print("\n[llm] error on " + model + ": " + str(e), flush=True) hard_fails += 1 model_index += 1 if hard_fails >= len(GROQ_MODELS): print("[llm] all Groq models failed; falling back to Ollama", flush=True) result = _ask_ollama(prompt, on_chunk) _cache[key] = result return result continue # groq specific def _ask_groq(prompt, on_chunk, model): r = requests.post( GROQ_URL, headers={"Authorization": "Bearer " + GROQ_API_KEY, "Content-Type": "application/json"}, json={"model": model, "messages": [{"role": "user", "content": prompt}], "stream": True, "temperature": 0.2, "max_tokens": GROQ_MAX_TOKENS}, stream=True, timeout=(10, None), ) r.raise_for_status() parts = [] for raw in r.iter_lines(decode_unicode=True): if not raw or raw == "[DONE]": continue if raw.startswith("data: "): raw = raw[6:] if raw == "[DONE]": break try: chunk = json.loads(raw)["choices"][0]["delta"].get("content", "") if chunk: parts.append(chunk) if on_chunk: on_chunk(chunk) except: continue return "".join(parts) # local llm specific def _ask_ollama(prompt, on_chunk): _ensure_ollama() r = requests.post( OLLAMA_URL + "/api/generate", json={"model": OLLAMA_MODEL, "prompt": prompt, "stream": True, "options": {"temperature": 0.2, "num_predict": 4096}}, stream=True, timeout=(10, None), ) r.raise_for_status() parts = [] for raw in r.iter_lines(decode_unicode=True): if not raw: continue payload = json.loads(raw) chunk = payload.get("response", "") if chunk: parts.append(chunk) if on_chunk: on_chunk(chunk) if payload.get("done"): break return "".join(parts) def _ensure_ollama(): try: if requests.get(OLLAMA_URL + "/api/tags", timeout=2).status_code == 200: return except Exception: pass subprocess.Popen(["ollama", "serve"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) time.sleep(2)