import llama_bootstrap # noqa: F401 — MUST be first import gradio as gr import os import json import sys import threading from pathlib import Path from typing import Optional MODEL_REPO = os.environ.get("MODEL_REPO", "DavidAU/Qwen3.6-27B-Heretic-Uncensored-FINETUNE-NEO-CODE-Di-IMatrix-MAX-GGUF") MODEL_FILE = os.environ.get("MODEL_FILE", "Qwen3.6-27B-NEO-CODE-HERE-2T-OT-Q4_K_M.gguf") N_CTX = int(os.environ.get("N_CTX", "16384")) MAX_TOKENS = int(os.environ.get("MAX_TOKENS", "4096")) N_GPU = int(os.environ.get("N_GPU_LAYERS", os.environ.get("DAVIDAU_N_GPU_LAYERS", "0"))) SYSTEM_PROMPT = """You are codeMax, an uncensored 27B coding assistant. You excel at: - Writing production code in Python, TypeScript, JavaScript, Rust, Go, C/C++, SQL, shell - Debugging, refactoring, reviewing complex codebases - Explaining algorithms, architecture, system design - Reading shared files and answering about them Rules: put code in ```language blocks. Be precise and thorough.""" CSS = """footer { display: none !important; }""" _llm = None _model_path: Optional[str] = None _load_lock = threading.Lock() # one load, ever — HF imports app.py twice; this makes that harmless def _download(): global _model_path if _model_path is not None: return _model_path from huggingface_hub import hf_hub_download print(f"[MODEL] Downloading {MODEL_FILE}...", file=sys.stderr, flush=True) _model_path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE) print(f"[MODEL] Cached -> {_model_path}", file=sys.stderr, flush=True) return _model_path def _load(): """LAZY: nothing loads at import. First chat message triggers the one and only load.""" global _llm if _llm is not None: return _llm with _load_lock: if _llm is not None: return _llm from llama_cpp import Llama path = _download() print(f"[MODEL] Loading (n_gpu_layers={N_GPU}, n_ctx={N_CTX})...", file=sys.stderr, flush=True) _llm = Llama( model_path=path, n_ctx=N_CTX, n_gpu_layers=N_GPU, chat_format="chatml", verbose=False, seed=-1, ) print("[MODEL] Ready.", file=sys.stderr, flush=True) return _llm def _read_text(path, enc="utf-8"): for e in [enc, "utf-8-sig", "latin-1", "cp1252", "utf-16"]: try: with open(path, "r", encoding=e) as f: return f.read() except UnicodeError: continue with open(path, "r", encoding="utf-8", errors="replace") as f: return f.read() def parse_file(file_path, file_name): suffix = Path(file_name).suffix.lower() name = Path(file_name).name if suffix in ( ".txt", ".md", ".py", ".ts", ".tsx", ".js", ".jsx", ".mjs", ".css", ".scss", ".less", ".html", ".htm", ".xml", ".svg", ".yaml", ".yml", ".toml", ".ini", ".cfg", ".conf", ".sh", ".bash", ".zsh", ".fish", ".ps1", ".bat", ".sql", ".prisma", ".c", ".cpp", ".cc", ".cxx", ".h", ".hpp", ".hh", ".java", ".kt", ".kts", ".scala", ".groovy", ".go", ".rs", ".rb", ".php", ".pl", ".pm", ".swift", ".r", ".lua", ".zig", ".nim", ".dart", ".env", ".gitignore", ".editorconfig", ".tf", ".tfvars", ".hcl", ".vue", ".svelte", ".astro", ): content = _read_text(file_path) lang = suffix.lstrip(".") if suffix == ".md": lang = "markdown" elif suffix in (".yml",): lang = "yaml" elif suffix in (".tf", ".tfvars"): lang = "hcl" elif suffix in (".htm",): lang = "html" return f"### `{name}`\n```{lang}\n{content}\n```\n\n---\n" if suffix == ".json": content = _read_text(file_path) try: parsed = json.loads(content) content = json.dumps(parsed, indent=2, ensure_ascii=False) except Exception: pass return f"### `{name}`\n```json\n{content}\n```\n\n---\n" if suffix == ".csv": import csv, io content = _read_text(file_path) rows = list(csv.reader(io.StringIO(content))) if len(rows) > 51: rows = rows[:50] + [[f"... {len(rows) - 50} rows truncated"]] widths = [max(len(str(c)) for c in col) for col in zip(*rows)] sep = "|-" + "-|-".join("-" * x for x in widths) + "-|" table = "\n".join( "| " + " | ".join(str(c).ljust(widths[i]) for i, c in enumerate(row)) + " |" for row in rows ) table = table.split("\n", 1) table.insert(1, sep) return f"### `{name}`\n" + "\n".join(table) + "\n\n---\n" if suffix == ".docx": try: import docx doc = docx.Document(file_path) text = "\n\n".join(p.text for p in doc.paragraphs if p.text.strip()) return f"### `{name}`\n{text}\n\n---\n" except Exception as e: return f"### `{name}` (DOCX error: {e})\n\n---\n" if suffix == ".pdf": try: import pdfplumber with pdfplumber.open(file_path) as pdf: text = "\n\n".join(page.extract_text() or "" for page in pdf.pages) return f"### `{name}`\n{text}\n\n---\n" except Exception as e: return f"### `{name}` (PDF error: {e})\n\n---\n" try: content = _read_text(file_path) return f"### `{name}`\n```\n{content[:10000]}\n```\n\n---\n" except Exception as e: return f"### `{name}` (could not read: {e})\n\n---\n" def parse_all_files(files): if not files: return "" parts = [] for f in files: if isinstance(f, dict): path = f.get("path") or f.get("name") name = f.get("orig_name") or f.get("name", "unknown") elif hasattr(f, "name"): path = f.name name = getattr(f, "orig_name", Path(path).name) else: path = str(f) name = Path(path).name try: parts.append(parse_file(path, name)) except Exception as e: parts.append(f"### `{name}`\nParse error: {e}\n\n---\n") return "\n".join(parts) def respond(message, history, uploaded_files): if _llm is None: yield history + [ {"role": "user", "content": message}, {"role": "assistant", "content": "⏳ **First run:** downloading + loading the 27B brain — a few minutes, one time only. I'm on it…"}, ] try: llm = _load() except Exception as e: yield history + [ {"role": "user", "content": message}, {"role": "assistant", "content": f"⚠️ Model failed to load: `{type(e).__name__}: {e}`\n\nMost likely: RAM too small for this quant (Q4_K_M needs a 32GB space) — or the download hiccuped, just send again."}, ] return file_context = parse_all_files(uploaded_files) messages = [{"role": "system", "content": SYSTEM_PROMPT}] if file_context: messages.append({"role": "user", "content": "[Uploaded files]\n\n" + file_context}) messages.append({"role": "assistant", "content": "Got it! I’ve read through all the uploaded files."}) for entry in history: role = entry.get("role", "user") content = entry.get("content", "") if isinstance(content, str) and content: messages.append({"role": role, "content": content}) messages.append({"role": "user", "content": message}) stream = llm.create_chat_completion( messages=messages, temperature=0.6, top_p=0.8, top_k=20, max_tokens=MAX_TOKENS, stream=True, ) partial = "" for chunk in stream: if chunk.get("choices"): delta = chunk["choices"][0].get("delta", {}) content = delta.get("content", "") if content: partial += content yield history + [ {"role": "user", "content": message}, {"role": "assistant", "content": partial}, ] def create_demo(): with gr.Blocks(title="codeMax — Qwen 3.6 27B Coder", css=CSS) as demo: gr.Markdown( "# codeMax\n" "**Qwen 3.6 27B · Uncensored · Code-optimized · Dedicated GPU**\n" "Drop code, docs, JSON, CSVs — chat with a 27B coding LLM." ) with gr.Row(equal_height=True): with gr.Column(scale=3): chatbot = gr.Chatbot( label="Chat", height=580, type="messages", avatar_images=(None, "https://huggingface.co/front/assets/huggingface_logo-noborder.svg"), ) with gr.Row(): msg = gr.Textbox(placeholder="Ask about code, share files, or chat...", scale=8, show_label=False, container=False) send = gr.Button(">", scale=1, variant="primary", min_width=48) clear_btn = gr.Button("Clear", size="sm") with gr.Column(scale=1): gr.Markdown("### Drop Files") files = gr.File( file_count="multiple", label="Code, docs, data...", file_types=[ ".py", ".ts", ".tsx", ".js", ".jsx", ".json", ".md", ".txt", ".csv", ".yaml", ".yml", ".toml", ".html", ".css", ".xml", ".sql", ".sh", ".go", ".rs", ".rb", ".java", ".c", ".cpp", ".h", ".docx", ".pdf", ".env", ".cfg", ".ini", ], ) uploaded_info = gr.Markdown("_No files uploaded._") def update_info(uploaded): if not uploaded: return "_No files uploaded._" names = [] for f in uploaded: if isinstance(f, dict): names.append(f.get("orig_name", "?")) else: names.append(getattr(f, "orig_name", Path(str(f)).name)) return "**Uploaded:**\n" + "\n".join(f"- `{n}`" for n in names) def stream_response(message, history, current_files): for h in respond(message, history, current_files): yield h msg.submit(stream_response, [msg, chatbot, files], [chatbot]).then(lambda: "", None, [msg]) send.click(stream_response, [msg, chatbot, files], [chatbot]).then(lambda: "", None, [msg]) clear_btn.click(lambda: [], None, chatbot, queue=False) files.change(update_info, files, uploaded_info) return demo if __name__ == "__main__": demo = create_demo() demo.queue(default_concurrency_limit=1, max_size=4) demo.launch( server_name="0.0.0.0", server_port=7860, ssr_mode=False, )