Create app.py

app.py

@@ -0,0 +1,460 @@
+# =============================================================================
+#  Qwen 3.6 27B CoderBot — Hugging Face ZeroGPU Space
+#  Model: DavidAU/Qwen3.6-27B-Heretic-Uncensored-FINETUNE-NEO-CODE-Di-IMatrix-MAX-GGUF
+#  Quant: Q4_K_M (~16GB, 94.5% of BF16) — fits ZeroGPU 48GB VRAM
+# =============================================================================
+
+import spaces
+import gradio as gr
+import os
+import json
+import tempfile
+import traceback
+from pathlib import Path
+from typing import List, Dict, Optional, Generator
+
+# ---------------------------------------------------------------------------
+#  MODEL CONFIGURATION — change these env vars or edit directly
+# ---------------------------------------------------------------------------
+
+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",  # 16.1 GB, 94.5% of BF16
+)
+N_CTX = int(os.environ.get("N_CTX", "8192"))          # context window
+N_GPU_LAYERS = int(os.environ.get("N_GPU_LAYERS", "99"))  # offload all layers
+
+# ---------------------------------------------------------------------------
+#  SYSTEM PROMPT — tuned for coding / code review
+# ---------------------------------------------------------------------------
+
+SYSTEM_PROMPT = """You are Qwen CoderBot — an uncensored 27B coding assistant.  You excel at:
+
+• Writing production-quality code in Python, TypeScript, JavaScript, Rust, Go, C/C++, SQL, shell, and more
+• Debugging, refactoring, and reviewing complex code
+• Explaining algorithms, data structures, architecture patterns, and system design
+• Reading and understanding code files users share with you
+
+Rules:
+- Always put code in ```language ... ``` blocks with the correct language tag.
+- When the user shares files, read them carefully before answering.  Refer back to specific lines or functions.
+- Be precise, concise, and thorough.  Prefer working, runnable solutions.
+- If you're unsure, say so — never fabricate APIs or libraries.
+- For Qwen "thinking" mode, you may use ... blocks.
+"""
+
+# ===========================================================================
+#  MODEL LOADER — lazy singleton, re-initialised inside @spaces.GPU when
+#  the GPU becomes available.  We download the GGUF at module level (CPU)
+#  and stream it into GPU memory on first call.
+# ===========================================================================
+
+_llm = None
+_model_path: Optional[str] = None
+
+
+def _ensure_model_downloaded() -> str:
+    """Download the GGUF once to the persistent HuggingFace cache."""
+    global _model_path
+    if _model_path is not None:
+        return _model_path
+    from huggingface_hub import hf_hub_download
+    print(f"[CoderBot] Downloading {MODEL_FILE} from {MODEL_REPO} …")
+    _model_path = hf_hub_download(
+        repo_id=MODEL_REPO,
+        filename=MODEL_FILE,
+    )
+    print(f"[CoderBot] Model cached at {_model_path}")
+    return _model_path
+
+
+def _get_llm():
+    """Return (or create) the llama-cpp Llama instance on GPU."""
+    global _llm
+    if _llm is not None:
+        return _llm
+    from llama_cpp import Llama
+    model_path = _ensure_model_downloaded()
+    print(f"[CoderBot] Loading model onto GPU (n_gpu_layers={N_GPU_LAYERS}) …")
+    _llm = Llama(
+        model_path=model_path,
+        n_ctx=N_CTX,
+        n_gpu_layers=N_GPU_LAYERS,
+        chat_format="chatml",
+        verbose=False,
+        seed=-1,
+    )
+    print("[CoderBot] Model loaded ✓")
+    return _llm
+
+
+# ===========================================================================
+#  FILE PARSERS — extract text from every common dev / doc format
+# ===========================================================================
+
+def _read_text(path: str, encoding: str = "utf-8") -> str:
+    """Safe text-file reader with encoding fallback."""
+    encodings = [encoding, "utf-8-sig", "latin-1", "cp1252", "utf-16"]
+    for enc in encodings:
+        try:
+            with open(path, "r", encoding=enc) as f:
+                return f.read()
+        except (UnicodeDecodeError, UnicodeError):
+            continue
+    # Last resort
+    with open(path, "r", encoding="utf-8", errors="replace") as f:
+        return f.read()
+
+
+def parse_file(file_path: str, file_name: str) -> str:
+    """
+    Parse a single file and return a markdown-formatted string with its
+    contents, ready to be pasted into the model context.
+    """
+    suffix = Path(file_name).suffix.lower()
+    name = Path(file_name).name
+
+    # ---- Plain text / code files ----
+    if suffix in (
+        ".txt", ".md", ".py", ".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs",
+        ".css", ".scss", ".less", ".html", ".htm", ".xml", ".svg",
+        ".yaml", ".yml", ".toml", ".ini", ".cfg", ".conf",
+        ".sh", ".bash", ".zsh", ".fish", ".ps1", ".bat",
+        ".sql", ".graphql", ".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",
+        ".dockerfile", ".makefile", ".cmake", ".gradle",
+        ".env", ".gitignore", ".editorconfig",
+        ".tf", ".tfvars", ".hcl",
+        ".vue", ".svelte", ".astro",
+        ".ipynb",
+    ):
+        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"
+        elif suffix in (".Dockerfile",):
+            lang = "dockerfile"
+        elif suffix == ".ipynb":
+            # Parse Jupyter notebook to extract code + markdown cells
+            try:
+                nb = json.loads(content)
+                lines = []
+                for cell in nb.get("cells", []):
+                    cell_type = cell.get("cell_type", "code")
+                    source = "".join(cell.get("source", []))
+                    if cell_type == "code":
+                        lines.append(f"```python\n{source}\n```")
+                    else:
+                        lines.append(source)
+                content = "\n\n".join(lines)
+                return f"### 📓 {name}\n\n{content}\n\n---\n"
+            except Exception:
+                pass
+        return f"### 📄 `{name}`\n```{lang}\n{content}\n```\n\n---\n"
+
+    # ---- JSON ----
+    if suffix == ".json":
+        content = _read_text(file_path)
+        try:
+            parsed = json.loads(content)
+            pretty = json.dumps(parsed, indent=2, ensure_ascii=False)
+            return f"### 📄 `{name}`\n```json\n{pretty}\n```\n\n---\n"
+        except Exception:
+            return f"### 📄 `{name}`\n```json\n{content}\n```\n\n---\n"
+
+    # ---- CSV ----
+    if suffix == ".csv":
+        import csv
+        import io
+        content = _read_text(file_path)
+        reader = csv.reader(io.StringIO(content))
+        rows = list(reader)
+        if len(rows) > 51:  # truncate very large CSVs
+            truncated = rows[:50]
+            truncated.append([f"… {len(rows) - 50} more rows truncated"])
+            rows = truncated
+        col_widths = [max(len(str(cell)) for cell in col) for col in zip(*rows)]
+        table = "\n".join(
+            "| " + " | ".join(str(cell).ljust(w) for cell, w in zip(row, col_widths)) + " |"
+            for row in rows
+        )
+        header_sep = "|-" + "-|-".join("-" * w for w in col_widths) + "-|"
+        table = table.split("\n", 1)
+        table.insert(1, header_sep)
+        return f"### 📊 `{name}`\n" + "\n".join(table) + "\n\n---\n"
+
+    # ---- Word documents ----
+    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 ImportError:
+            return f"### ⚠️ `{name}` (DOCX — install python-docx)\n\n---\n"
+        except Exception as e:
+            return f"### ⚠️ `{name}` (DOCX parse error: {e})\n\n---\n"
+
+    # ---- PDF ----
+    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 ImportError:
+            return f"### ⚠️ `{name}` (PDF — install pdfplumber)\n\n---\n"
+        except Exception as e:
+            return f"### ⚠️ `{name}` (PDF parse error: {e})\n\n---\n"
+
+    # ---- Fallback: try to read as text ----
+    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) -> str:
+    """Parse a list of Gradio file objects into a single context string."""
+    if not files:
+        return ""
+    parts = []
+    for f in files:
+        # Gradio file objects may be dicts or have .name attribute
+        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)
+
+
+# ===========================================================================
+#  MAIN GENERATION FUNCTION — wrapped with @spaces.GPU for ZeroGPU
+# ===========================================================================
+
+@spaces.GPU
+def chatbot_respond(
+    message: str,
+    history: List,
+    uploaded_files,
+) -> Generator[List, None, None]:
+    """
+    Called on every user message.  history is the Gradio chatbot history
+    (list of [user_msg, bot_msg] pairs).  uploaded_files is the current
+    file list from gr.File.
+
+    Yields the updated history list after each token chunk, giving a
+    streaming effect in the Gradio Chatbot.
+    """
+    # ---- 1. Parse any newly uploaded files ----
+    file_context = parse_all_files(uploaded_files)
+
+    # ---- 2. Build the message list for llama-cpp-python (ChatML format) ----
+    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
+
+    # If files were uploaded, inject their content as a system-level note
+    if file_context:
+        messages.append({
+            "role": "user",
+            "content": (
+                "[The user has uploaded the following files.  "
+                "Read them carefully and refer to them in your answers.]\n\n"
+                + file_context
+            ),
+        })
+        messages.append({
+            "role": "assistant",
+            "content": (
+                "Got it! I've read through all the uploaded files.  "
+                "Ask me anything about them."
+            ),
+        })
+
+    # Append the real conversation history
+    for user_msg, bot_msg in history:
+        if user_msg:
+            messages.append({"role": "user", "content": user_msg})
+        if bot_msg:
+            messages.append({"role": "assistant", "content": bot_msg})
+
+    # Append the current message
+    messages.append({"role": "user", "content": message})
+
+    # ---- 3. Stream the completion ----
+    llm = _get_llm()
+    stream = llm.create_chat_completion(
+        messages=messages,
+        temperature=0.6,      # lower = more precise for coding
+        top_p=0.8,
+        top_k=20,
+        max_tokens=4096,
+        stream=True,
+        stop=["<|im_end|>", "<|endoftext|>"],
+    )
+
+    partial = ""
+    for chunk in stream:
+        choices = chunk.get("choices", [])
+        if choices:
+            delta = choices[0].get("delta", {})
+            content = delta.get("content", "")
+            if content:
+                partial += content
+        # YIELD the updated history so Gradio streams into the chat UI
+        yield history + [[message, partial]]
+
+    # Final yield to guarantee the full message is rendered
+    yield history + [[message, partial]]
+
+
+# ===========================================================================
+#  GRADIO UI — clean two‑column layout: chat on the left, file drop on right
+# ===========================================================================
+
+def create_demo() -> gr.Blocks:
+    css = """
+    .file-upload-col { background: var(--background-fill-secondary); border-radius: 12px; padding: 16px; }
+    footer { display: none !important; }
+    """
+    with gr.Blocks(
+        css=css,
+        theme=gr.themes.Soft(primary_hue="violet", secondary_hue="slate"),
+        title="Qwen 3.6 27B CoderBot",
+    ) as demo:
+
+        gr.Markdown(
+            """
+            # 🧠 Qwen 3.6 27B CoderBot
+            **Uncensored · Code-optimized · ZeroGPU-powered**
+            Drop code files, docs, JSON, CSVs — chat about them with a 27B coding LLM.
+            """,
+        )
+
+        with gr.Row(equal_height=True):
+            # ---- LEFT: Chat ----
+            with gr.Column(scale=3):
+                chatbot = gr.Chatbot(
+                    label="Chat",
+                    height=580,
+                    bubble_full_width=False,
+                    avatar_images=(
+                        None,
+                        "https://huggingface.co/front/assets/huggingface_logo-noborder.svg",
+                    ),
+                )
+                with gr.Row():
+                    msg = gr.Textbox(
+                        placeholder="Ask about your code, or just start chatting …",
+                        scale=8,
+                        show_label=False,
+                        container=False,
+                    )
+                    send = gr.Button("▶", scale=1, variant="primary", min_width=48)
+                with gr.Row():
+                    clear_btn = gr.Button("🗑 Clear chat", size="sm", scale=1)
+
+            # ---- RIGHT: File upload + info ----
+            with gr.Column(scale=1, elem_classes="file-upload-col"):
+                gr.Markdown("### 📎 Drop Files")
+                files = gr.File(
+                    file_count="multiple",
+                    label="Upload code, docs, data …",
+                    file_types=[
+                        ".py", ".ts", ".tsx", ".js", ".jsx", ".mjs",
+                        ".json", ".yaml", ".yml", ".toml",
+                        ".md", ".txt", ".csv",
+                        ".html", ".css", ".scss", ".xml", ".svg",
+                        ".c", ".cpp", ".h", ".hpp", ".java", ".kt",
+                        ".go", ".rs", ".rb", ".php", ".swift",
+                        ".sql", ".sh", ".bash", ".ps1",
+                        ".docx", ".pdf",
+                        ".env", ".gitignore", ".cfg", ".ini", ".conf",
+                        ".vue", ".svelte", ".dockerfile",
+                    ],
+                )
+                gr.Markdown(
+                    """
+                    **Supported:** `.py` `.ts` `.js` `.json` `.md` `.docx` `.pdf` `.csv`
+                    `.yaml` `.toml` `.html` `.css` `.sql` `.go` `.rs` `.java` + more
+
+                    Files are parsed and sent as context — the model reads them
+                    before answering.
+                    """
+                )
+                uploaded_info = gr.Markdown("_No files uploaded yet._")
+                file_content_state = gr.State("")
+
+        # ---- Event wiring ----
+
+        def update_file_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))
+            label = "**Uploaded:**\n" + "\n".join(f"• `{n}`" for n in names)
+            return label, parse_all_files(uploaded)
+
+        def respond(message, history, current_files):
+            """Generator wrapper that streams into the chatbot."""
+            for updated_history in chatbot_respond(message, history, current_files):
+                yield updated_history
+
+        # Trigger response on text submit or send button
+        msg.submit(
+            respond,
+            inputs=[msg, chatbot, files],
+            outputs=[chatbot],
+        ).then(lambda: "", None, [msg])
+
+        send.click(
+            respond,
+            inputs=[msg, chatbot, files],
+            outputs=[chatbot],
+        ).then(lambda: "", None, [msg])
+
+        # Clear chat
+        clear_btn.click(lambda: None, None, chatbot, queue=False)
+
+        # File upload feedback
+        files.change(update_file_info, files, [uploaded_info, file_content_state])
+
+    return demo
+
+
+# ===========================================================================
+#  LAUNCH
+# ===========================================================================
+
+if __name__ == "__main__":
+    demo = create_demo()
+    demo.queue(default_concurrency_limit=1, max_size=4)
+    demo.launch()