Upload folder using huggingface_hub

.env

@@ -0,0 +1,11 @@
+MODEL_NAME=Qwen/Qwen2.5-Coder-1.5B-Instruct
+FALLBACK_MODEL_NAME=Qwen/Qwen2.5-Coder-0.5B-Instruct
+FINAL_FALLBACK_MODEL_NAME=sshleifer/tiny-gpt2
+EMBEDDING_MODEL=sentence-transformers/all-MiniLM-L6-v2
+MAX_NEW_TOKENS=256
+TEMPERATURE=0.2
+TOP_P=0.95
+USE_RAG=true
+FORCE_MOCK_MODE=false
+API_KEY=gk_zxcvbnmlkjhgfdsaq1w2e3r4t5y6u7i8o9p0
+RATE_LIMIT_PER_MINUTE=30

.env.example

@@ -0,0 +1,11 @@
+MODEL_NAME=Qwen/Qwen2.5-Coder-1.5B-Instruct
+FALLBACK_MODEL_NAME=Qwen/Qwen2.5-Coder-0.5B-Instruct
+FINAL_FALLBACK_MODEL_NAME=sshleifer/tiny-gpt2
+EMBEDDING_MODEL=sentence-transformers/all-MiniLM-L6-v2
+MAX_NEW_TOKENS=256
+TEMPERATURE=0.2
+TOP_P=0.95
+USE_RAG=true
+FORCE_MOCK_MODE=false
+API_KEY=change-me
+RATE_LIMIT_PER_MINUTE=30

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+
+COPY . /app
+
+EXPOSE 8000
+
+CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"]

Makefile

@@ -0,0 +1,22 @@
+.PHONY: install run space smoke compile docker-up docker-down
+
+install:
+	python tasks.py install
+
+run:
+	python tasks.py run
+
+space:
+	python tasks.py space
+
+smoke:
+	python tasks.py smoke
+
+compile:
+	python tasks.py compile
+
+docker-up:
+	python tasks.py docker-up
+
+docker-down:
+	python tasks.py docker-down

README.md

@@ -1,12 +1,192 @@
----
-title: Coding Llm Space
-emoji: 📉
-colorFrom: indigo
-colorTo: red
-sdk: gradio
-sdk_version: 6.12.0
-app_file: app.py
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Advanced Coding LLM (Production-Ready Starter)
+
+This project provides a deployable coding assistant API built on a free Hugging Face coding model.
+
+## Model Strategy
+
+- Primary model: `Qwen/Qwen2.5-Coder-1.5B-Instruct` (free/open on Hugging Face).
+- Fallback model: `Qwen/Qwen2.5-Coder-0.5B-Instruct` if primary load fails.
+- Final emergency fallback: `sshleifer/tiny-gpt2` (for guaranteed startup).
+- No heavy training required.
+- LoRA-ready architecture included in `src/lora_prepare.py`.
+
+## Features
+
+- Code generation
+- Debugging / buggy code fixing
+- Code explanation
+- Instruction following
+- Confidence estimation (from token probabilities)
+- Important token extraction (low-confidence tokens)
+- Relevancy score (embedding cosine similarity)
+- Hallucination checks:
+  - Syntax validation
+  - Runtime smoke test
+- Optional RAG with FAISS from `data/sample_snippets.json`
+
+## Project Structure
+
+```text
+coding-llm/
+│── data/
+│── src/
+│── api/
+│── requirements.txt
+│── README.md
+```
+
+## API Output Format
+
+`POST /generate` returns:
+
+```json
+{
+  "code": "...",
+  "explanation": "...",
+  "confidence": 0.0,
+  "important_tokens": ["..."],
+  "relevancy_score": 0.0,
+  "hallucination": false,
+  "latency_ms": 0
+}
+```
+
+If hallucination is detected, the reason is appended inside `explanation`.
+
+## Local Run
+
+1. Create environment and install:
+
+```bash
+pip install -r requirements.txt
+```
+
+Optional: create `.env` from `.env.example` and set values:
+
+```bash
+copy .env.example .env
+```
+
+2. Run API:
+
+```bash
+uvicorn api.main:app --host 0.0.0.0 --port 8000
+```
+
+3. Test request:
+
+```bash
+curl -X POST "http://127.0.0.1:8000/generate" ^
+  -H "Content-Type: application/json" ^
+  -d "{\"instruction\":\"Fix this Python function\",\"input\":\"def add(a,b) return a+b\"}"
+```
+
+4. Optional client:
+
+```bash
+python client_example.py
+```
+
+## Hugging Face Deployment (Space)
+
+This repo includes:
+
+- `app.py` (Gradio app for HF Space)
+- `upload_to_hf.py` (upload helper script)
+- `README_hf_space.md` (Space metadata template)
+
+Steps:
+
+1. Create a HF access token with write permission.
+2. Run:
+
+```bash
+python upload_to_hf.py --repo-id <your-username/coding-llm-space> --token <HF_TOKEN>
+```
+
+3. Your Space launches with public UI and can be called by Hugging Face API key.
+
+## Security and Ops
+
+- API key auth enabled when `API_KEY` is set.
+- In-memory per-IP rate limiting via `RATE_LIMIT_PER_MINUTE`.
+- Dockerized API included (`Dockerfile`).
+- Model is loaded lazily on first `/generate` request (faster boot, fewer startup failures).
+- Set `FORCE_MOCK_MODE=true` to run instantly without downloading models.
+- Windows quick-start scripts:
+  - `run_api.bat`
+  - `run_space.bat`
+
+## Docker Compose
+
+```bash
+copy .env.example .env
+docker compose up --build
+```
+
+API is available at `http://127.0.0.1:8000`.
+
+## Automated Smoke Test
+
+Run this after API starts:
+
+```bash
+python smoke_test.py
+```
+
+This validates:
+- `GET /health`
+- `POST /generate`
+- required JSON output keys
+
+## One-command Task Runner
+
+Cross-platform:
+
+```bash
+python tasks.py install
+python tasks.py run
+python tasks.py smoke
+python tasks.py serve-smoke
+python tasks.py docker-up
+python tasks.py docker-down
+python tasks.py hf-upload --repo-id <user/space-name> --token <HF_TOKEN>
+```
+
+Windows shortcut:
+
+```bat
+run_tasks.bat install
+run_tasks.bat run
+run_tasks.bat smoke
+run_tasks.bat serve-smoke
+```
+
+Makefile (Linux/macOS/WSL):
+
+```bash
+make install
+make run
+make smoke
+make docker-up
+```
+
+## FastAPI Endpoint
+
+### `POST /generate`
+
+Input JSON:
+
+```json
+{
+  "instruction": "Explain this code and improve it",
+  "input": "def f(x): return x*x"
+}
+```
+
+## Notes for Production
+
+- Keep `max_new_tokens` modest for low latency.
+- Add request auth/rate limiting before exposing public endpoint.
+- For stronger quality, add curated retrieval corpus in `data/`.
+- For robust hallucination checks, extend tests per language/framework.

README_hf_space.md

@@ -0,0 +1,14 @@
+---
+title: Advanced Coding LLM
+emoji: 🤖
+colorFrom: indigo
+colorTo: blue
+sdk: gradio
+sdk_version: 4.44.0
+app_file: app.py
+pinned: false
+---
+
+# Hugging Face Space Metadata
+
+This file can be copied to your Space `README.md` to configure Space UI metadata.

api/__init__.py

@@ -0,0 +1 @@
+"""API package."""

api/main.py

@@ -0,0 +1,44 @@
+"""FastAPI entrypoint for coding model."""
+
+from fastapi import Depends, FastAPI, Request
+
+from api.security import check_rate_limit, verify_api_key
+from src.pipeline import CodingLLMPipeline
+from src.schemas import GenerateRequest, GenerateResponse
+
+app = FastAPI(title="Advanced Coding LLM API", version="1.0.0")
+pipeline = CodingLLMPipeline()
+
+
+@app.get("/health")
+def health():
+    model_name = pipeline.model_bundle.active_model_name if pipeline.model_bundle else "not_loaded_yet"
+    return {"status": "ok", "model": model_name}
+
+
+@app.post("/generate", response_model=GenerateResponse)
+def generate(
+    payload: GenerateRequest,
+    request: Request,
+    _: None = Depends(verify_api_key),
+):
+    client_host = request.client.host if request.client else "unknown"
+    check_rate_limit(client_host)
+    try:
+        result = pipeline.run(payload.instruction, payload.input)
+        return GenerateResponse(**result)
+    except Exception as exc:
+        # Keep response shape consistent with GenerateResponse model.
+        return GenerateResponse(
+            code="",
+            explanation=(
+                "Generation failed.\n\n"
+                f"Error: {exc}\n"
+                "Hint: Check model name/access and internet connection for first model download."
+            ),
+            confidence=0.0,
+            important_tokens=[],
+            relevancy_score=0.0,
+            hallucination=True,
+            latency_ms=0,
+        )

api/security.py

@@ -0,0 +1,33 @@
+"""Simple API key and in-memory rate limiting."""
+
+from __future__ import annotations
+
+import time
+from collections import defaultdict, deque
+
+from fastapi import Header, HTTPException
+
+from src.config import settings
+
+_request_windows: dict[str, deque[float]] = defaultdict(deque)
+
+
+def verify_api_key(x_api_key: str = Header(default="")):
+    """Enforce API key when configured."""
+    if settings.api_key and x_api_key != settings.api_key:
+        raise HTTPException(status_code=401, detail="Invalid API key")
+
+
+def check_rate_limit(client_id: str) -> None:
+    """Allow at most N requests per minute for each client id."""
+    now = time.time()
+    window_start = now - 60
+    q = _request_windows[client_id]
+
+    while q and q[0] < window_start:
+        q.popleft()
+
+    if len(q) >= settings.rate_limit_per_minute:
+        raise HTTPException(status_code=429, detail="Rate limit exceeded")
+
+    q.append(now)

app.py

@@ -0,0 +1,45 @@
+"""Gradio app for Hugging Face Spaces deployment."""
+
+from __future__ import annotations
+
+import json
+
+import gradio as gr
+
+from src.pipeline import CodingLLMPipeline
+
+pipeline = CodingLLMPipeline()
+
+
+def generate(instruction: str, user_input: str):
+    try:
+        result = pipeline.run(instruction, user_input)
+        return json.dumps(result, indent=2)
+    except Exception as exc:
+        return json.dumps(
+            {
+                "code": "",
+                "explanation": f"Generation failed: {exc}",
+                "confidence": 0.0,
+                "important_tokens": [],
+                "relevancy_score": 0.0,
+                "hallucination": True,
+                "latency_ms": 0,
+            },
+            indent=2,
+        )
+
+
+demo = gr.Interface(
+    fn=generate,
+    inputs=[
+        gr.Textbox(label="Instruction", placeholder="Explain, generate or debug code..."),
+        gr.Textbox(label="Input", lines=12, placeholder="Paste code or context"),
+    ],
+    outputs=gr.Code(label="JSON Output", language="json"),
+    title="Advanced Coding LLM",
+    description="Fast coding assistant with confidence, relevancy, and hallucination checks.",
+)
+
+if __name__ == "__main__":
+    demo.launch()

client_example.py

@@ -0,0 +1,29 @@
+"""Small client example to call POST /generate."""
+
+from __future__ import annotations
+
+import json
+import os
+
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+BASE_URL = os.getenv("CODING_LLM_URL", "http://127.0.0.1:8000")
+API_KEY = os.getenv("API_KEY", "")
+
+payload = {
+    "instruction": "Fix this function and explain",
+    "input": "def add(a,b) return a+b",
+}
+
+headers = {"Content-Type": "application/json"}
+if API_KEY:
+    headers["x-api-key"] = API_KEY
+
+resp = requests.post(f"{BASE_URL}/generate", headers=headers, json=payload, timeout=60)
+if resp.status_code == 401:
+    raise PermissionError("Unauthorized (401). Ensure API_KEY matches server configuration.")
+resp.raise_for_status()
+print(json.dumps(resp.json(), indent=2))

data/sample_snippets.json

@@ -0,0 +1,11 @@
+[
+  {
+    "snippet": "def binary_search(arr, target):\n    lo, hi = 0, len(arr) - 1\n    while lo <= hi:\n        mid = (lo + hi) // 2\n        if arr[mid] == target:\n            return mid\n        if arr[mid] < target:\n            lo = mid + 1\n        else:\n            hi = mid - 1\n    return -1"
+  },
+  {
+    "snippet": "def is_palindrome(s):\n    clean = ''.join(ch.lower() for ch in s if ch.isalnum())\n    return clean == clean[::-1]"
+  },
+  {
+    "snippet": "def safe_div(a, b):\n    if b == 0:\n        raise ValueError('Division by zero')\n    return a / b"
+  }
+]

docker-compose.yml

@@ -0,0 +1,11 @@
+version: "3.9"
+
+services:
+  coding-llm-api:
+    build: .
+    container_name: coding-llm-api
+    env_file:
+      - .env
+    ports:
+      - "8000:8000"
+    restart: unless-stopped

instruction.md

@@ -0,0 +1,187 @@
+# Advanced Coding LLM - Complete Instructions
+
+This document provides full setup, run, validation, optimization, and deployment steps for the `coding-llm` project.
+
+## 1) Prerequisites
+
+- Python 3.10+ (recommended 3.11/3.12)
+- Git
+- Internet access for first model download
+- Optional: Docker Desktop
+- Optional: Hugging Face account and access token
+
+## 2) Project Setup
+
+From project root:
+
+```bash
+cd "C:\Users\GIRISH\OneDrive\Desktop\AI model_14_04_26\coding-llm"
+```
+
+Create environment file:
+
+```bash
+copy .env.example .env
+```
+
+Install dependencies:
+
+```bash
+python tasks.py install
+```
+
+## 3) Configure `.env`
+
+Open `.env` and set values:
+
+- `MODEL_NAME=Qwen/Qwen2.5-Coder-1.5B-Instruct`
+- `FALLBACK_MODEL_NAME=Qwen/Qwen2.5-Coder-0.5B-Instruct`
+- `FINAL_FALLBACK_MODEL_NAME=sshleifer/tiny-gpt2` (optional emergency fallback)
+- `FORCE_MOCK_MODE=false` (true for instant test mode)
+- `API_KEY=<your_secret_key>`
+- `RATE_LIMIT_PER_MINUTE=30`
+- `USE_RAG=true`
+
+## 4) Run API Locally
+
+```bash
+python tasks.py run
+```
+
+Server runs at:
+
+- `http://127.0.0.1:8000`
+
+Health endpoint:
+
+- `GET http://127.0.0.1:8000/health`
+
+## 5) Run Smoke Tests
+
+### Full smoke test
+
+```bash
+python smoke_test.py
+```
+
+### Health-only smoke test
+
+```bash
+set SMOKE_SKIP_GENERATE=true
+python smoke_test.py
+```
+
+### Combined run-and-test command
+
+```bash
+python tasks.py serve-smoke
+```
+
+This starts server, executes smoke test, and shuts server down automatically.
+
+## 6) If Generation Is Slow on First Run
+
+First `/generate` may take long due to model download/warmup.
+
+Options:
+
+- Increase timeout:
+  - `set SMOKE_TIMEOUT=900`
+- Use mock mode for quick validation:
+  - set `FORCE_MOCK_MODE=true`
+- Run full mode after model cache is ready.
+
+## 7) API Usage
+
+### Endpoint
+
+- `POST /generate`
+
+### Input JSON
+
+```json
+{
+  "instruction": "Fix this code",
+  "input": "def add(a,b) return a+b"
+}
+```
+
+### Required Header (if API key enabled)
+
+- `x-api-key: <API_KEY>`
+
+### Output JSON
+
+```json
+{
+  "code": "...",
+  "explanation": "...",
+  "confidence": 0.0,
+  "important_tokens": ["..."],
+  "relevancy_score": 0.0,
+  "hallucination": false,
+  "latency_ms": 0
+}
+```
+
+## 8) Docker Deployment
+
+```bash
+copy .env.example .env
+docker compose up --build -d
+```
+
+Validate:
+
+```bash
+python smoke_test.py
+```
+
+Stop:
+
+```bash
+docker compose down
+```
+
+## 9) Hugging Face Space Deployment
+
+Create HF token (write permission), then:
+
+```bash
+python tasks.py hf-upload --repo-id <username/coding-llm-space> --token <HF_TOKEN>
+```
+
+After upload, configure Space variables/secrets:
+
+- `MODEL_NAME`
+- `FALLBACK_MODEL_NAME`
+- `FORCE_MOCK_MODE`
+- `API_KEY` (if needed in your architecture)
+
+## 10) Production Hardening Checklist
+
+- Keep `API_KEY` enabled
+- Keep rate limiting enabled (`RATE_LIMIT_PER_MINUTE`)
+- Put API behind HTTPS reverse proxy
+- Add logging and monitoring
+- Pin model versions if strict reproducibility required
+- Use `FORCE_MOCK_MODE=false` in production
+
+## 11) Common Troubleshooting
+
+- `WinError 10061`:
+  - API server is not running. Start with `python tasks.py run`.
+- `401 Unauthorized`:
+  - `x-api-key` does not match server `API_KEY`.
+- Health works but generate times out:
+  - model is still downloading/warming up.
+- Low-quality gibberish output:
+  - likely fallback model path used; verify `.env` model names.
+
+## 12) Recommended Daily Commands
+
+- Install/update: `python tasks.py install`
+- Run API: `python tasks.py run`
+- Smoke: `python tasks.py smoke`
+- Run+smoke: `python tasks.py serve-smoke`
+- Docker up/down: `python tasks.py docker-up` / `python tasks.py docker-down`

requirements.txt

@@ -0,0 +1,16 @@
+fastapi>=0.115.0
+uvicorn>=0.30.0
+transformers>=4.44.0
+torch>=2.3.0
+accelerate>=0.33.0
+sentence-transformers>=3.0.0
+faiss-cpu>=1.8.0
+numpy>=1.26.0
+scikit-learn>=1.5.0
+pydantic>=2.8.0
+huggingface-hub>=0.24.0
+gradio>=4.40.0
+datasets>=2.20.0
+peft>=0.12.0
+requests>=2.32.0
+python-dotenv>=1.0.1

run_api.bat

@@ -0,0 +1,6 @@
+@echo off
+setlocal
+if exist .env (
+  for /f "usebackq delims=" %%a in (".env") do set "%%a"
+)
+uvicorn api.main:app --host 0.0.0.0 --port 8000

run_space.bat

@@ -0,0 +1,6 @@
+@echo off
+setlocal
+if exist .env (
+  for /f "usebackq delims=" %%a in (".env") do set "%%a"
+)
+python app.py

run_tasks.bat

@@ -0,0 +1,2 @@
+@echo off
+python tasks.py %*

smoke_test.py

@@ -0,0 +1,112 @@
+"""Automated smoke test for /health and /generate endpoints."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+BASE_URL = os.getenv("CODING_LLM_URL", "http://127.0.0.1:8000")
+API_KEY = os.getenv("API_KEY", "")
+TIMEOUT = int(os.getenv("SMOKE_TIMEOUT", "300"))
+SKIP_GENERATE = os.getenv("SMOKE_SKIP_GENERATE", "false").lower() == "true"
+
+
+def _headers() -> dict[str, str]:
+    headers = {"Content-Type": "application/json"}
+    if API_KEY:
+        headers["x-api-key"] = API_KEY
+    return headers
+
+
+def wait_for_health() -> dict:
+    candidate_urls = [BASE_URL]
+    if "127.0.0.1" in BASE_URL:
+        candidate_urls.append(BASE_URL.replace("127.0.0.1", "localhost"))
+    elif "localhost" in BASE_URL:
+        candidate_urls.append(BASE_URL.replace("localhost", "127.0.0.1"))
+
+    deadline = time.time() + TIMEOUT
+    last_errors: list[str] = []
+    while time.time() < deadline:
+        for url in candidate_urls:
+            try:
+                resp = requests.get(f"{url}/health", timeout=10)
+                if resp.status_code == 200:
+                    return resp.json()
+                last_errors.append(f"{url}/health -> HTTP {resp.status_code}")
+            except requests.RequestException as exc:
+                last_errors.append(f"{url}/health -> {exc}")
+        time.sleep(2)
+    preview = "; ".join(last_errors[-5:]) if last_errors else "No response details captured."
+    raise TimeoutError(f"Health check timeout. API did not become ready. Recent errors: {preview}")
+
+
+def test_generate() -> dict:
+    payload = {
+        "instruction": "Fix this function and explain briefly",
+        "input": "def add(a,b) return a+b",
+    }
+    resp = requests.post(
+        f"{BASE_URL}/generate",
+        headers=_headers(),
+        json=payload,
+        timeout=TIMEOUT,
+    )
+    if resp.status_code == 401:
+        raise PermissionError(
+            "Unauthorized (401). Set API_KEY in .env or environment before running smoke_test.py."
+        )
+    resp.raise_for_status()
+    body = resp.json()
+
+    required_keys = [
+        "code",
+        "explanation",
+        "confidence",
+        "important_tokens",
+        "relevancy_score",
+        "hallucination",
+        "latency_ms",
+    ]
+    missing = [k for k in required_keys if k not in body]
+    if missing:
+        raise ValueError(f"Missing keys in /generate response: {missing}")
+    return body
+
+
+def main():
+    print(f"[smoke] waiting for {BASE_URL}/health ...")
+    health = wait_for_health()
+    print("[smoke] health ok:", json.dumps(health))
+
+    if SKIP_GENERATE:
+        print("[smoke] skipping /generate (SMOKE_SKIP_GENERATE=true)")
+        print("[smoke] SUCCESS")
+        return
+
+    print("[smoke] running /generate ... (first run may download model)")
+    result = test_generate()
+    print("[smoke] /generate ok")
+    print(json.dumps(result, indent=2)[:2000])
+    print("[smoke] SUCCESS")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        msg = str(exc)
+        if "Read timed out" in msg:
+            msg = (
+                f"{msg}\nHint: model warmup is still running. "
+                "Wait longer, increase SMOKE_TIMEOUT, or restart API with FORCE_MOCK_MODE=true for instant checks."
+            )
+        print(f"[smoke] FAILED: {msg}")
+        sys.exit(1)

specification_file.md

@@ -0,0 +1,159 @@
+# Advanced Coding LLM - Technical Specification
+
+## 1) Objective
+
+Build a production-ready coding assistant API deployable locally and on Hugging Face, supporting:
+
+- Code generation
+- Debugging/fixing buggy code
+- Code explanation
+- Instruction following
+- Explainability signals
+- Relevancy scoring
+- Hallucination checks
+- Optional RAG
+
+## 2) Core Functional Requirements
+
+### 2.1 Model
+
+- Primary model: `Qwen/Qwen2.5-Coder-1.5B-Instruct`
+- Fallback model: `Qwen/Qwen2.5-Coder-0.5B-Instruct`
+- Emergency fallback mode supported (mock path available)
+- Architecture compatible with future LoRA integration (`src/lora_prepare.py`)
+
+### 2.2 API
+
+- Framework: FastAPI
+- Endpoint: `POST /generate`
+- Health: `GET /health`
+- Input schema:
+  - `instruction: str`
+  - `input: str`
+- Output schema:
+  - `code: str`
+  - `explanation: str`
+  - `confidence: float`
+  - `important_tokens: list[str]`
+  - `relevancy_score: float`
+  - `hallucination: bool`
+  - `latency_ms: int`
+
+### 2.3 Explainability
+
+- Confidence from token probabilities over generated tokens
+- Important tokens extracted from low-probability tokens
+
+### 2.4 Relevancy
+
+- Query-to-output semantic score using TF-IDF + cosine similarity
+
+### 2.5 Hallucination Checks
+
+- Python syntax validation (`ast.parse`)
+- Runtime smoke execution for Python-like outputs
+- Skip runtime execution for non-Python-like outputs
+
+### 2.6 RAG
+
+- Basic retrieval from local snippets dataset
+- FAISS index over normalized TF-IDF vectors
+- Inject top-k snippets into prompt context
+
+## 3) Non-Functional Requirements
+
+- Runnable on local workstation
+- Supports no-training initial deployment
+- Lazy-load model to reduce startup failures
+- Graceful fallback response when model unavailable
+- Windows-compatible developer workflow
+
+## 4) Security Requirements
+
+- API key auth via `x-api-key` (if configured)
+- Per-IP in-memory rate limiting
+- No secrets committed to repository (`.env` ignored)
+
+## 5) Performance Requirements
+
+- Lazy model initialization
+- Runtime checks bounded by timeout
+- Optional mock mode (`FORCE_MOCK_MODE=true`) for fast operational checks
+
+## 6) Deployment Requirements
+
+### Local
+
+- `python tasks.py install`
+- `python tasks.py run`
+
+### Docker
+
+- `docker compose up --build -d`
+
+### Hugging Face Space
+
+- `python tasks.py hf-upload --repo-id <id> --token <token>`
+- Gradio entrypoint in `app.py`
+
+## 7) Project Structure
+
+```text
+coding-llm/
+│── data/
+│── src/
+│── api/
+│── requirements.txt
+│── README.md
+│── instruction.md
+│── specification_file.md
+```
+
+## 8) Module Responsibilities
+
+- `api/main.py`: API routes and response wiring
+- `api/security.py`: API key + rate limiting
+- `src/config.py`: environment-driven settings
+- `src/model_loader.py`: model/fallback loading
+- `src/generator.py`: generation + confidence extraction
+- `src/pipeline.py`: orchestration layer
+- `src/rag.py`: snippet retrieval
+- `src/relevancy.py`: relevancy score computation
+- `src/hallucination.py`: syntax/runtime checks
+- `src/lora_prepare.py`: LoRA adapter hook
+- `app.py`: Gradio UI for HF Spaces
+- `upload_to_hf.py`: HF deployment uploader
+- `tasks.py`: command runner
+- `smoke_test.py`: runtime integration validation
+
+## 9) Operational Modes
+
+- **Real Model Mode**
+  - `FORCE_MOCK_MODE=false`
+  - Uses HF model loading and generation
+- **Mock Mode**
+  - `FORCE_MOCK_MODE=true`
+  - Returns deterministic fallback output for reliability testing
+
+## 10) Validation and QA
+
+- Static compile check with `python -m compileall`
+- Lint diagnostics via editor/tooling
+- Smoke checks:
+  - health endpoint reachable
+  - generate endpoint returns full schema
+
+## 11) Known Constraints
+
+- First generation may be slow due to model download/warmup
+- Quality depends on available model and decoding configuration
+- In-memory rate limiter is single-process only
+
+## 12) Future Enhancements
+
+- Redis-backed distributed rate limiting
+- Better language-aware hallucination tests
+- Prompt templates per task type
+- Streaming token responses
+- Persistent vector store (Chroma/FAISS on-disk)
+- CI/CD workflow for automated deploy/test

src/__init__.py

@@ -0,0 +1,6 @@
+"""Source package for coding-llm."""
+
+import os
+
+# Avoid TensorFlow/Keras import path conflicts in mixed Python environments.
+os.environ.setdefault("TRANSFORMERS_NO_TF", "1")

src/config.py

@@ -0,0 +1,28 @@
+"""Project-level configuration values."""
+
+from dataclasses import dataclass
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+@dataclass(frozen=True)
+class Settings:
+    """Central settings object for model and runtime controls."""
+
+    model_name: str = os.getenv("MODEL_NAME", "Qwen/Qwen2.5-Coder-1.5B-Instruct")
+    fallback_model_name: str = os.getenv("FALLBACK_MODEL_NAME", "Qwen/Qwen2.5-Coder-0.5B-Instruct")
+    final_fallback_model_name: str = os.getenv("FINAL_FALLBACK_MODEL_NAME", "sshleifer/tiny-gpt2")
+    embedding_model: str = os.getenv("EMBEDDING_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
+    max_new_tokens: int = int(os.getenv("MAX_NEW_TOKENS", "256"))
+    temperature: float = float(os.getenv("TEMPERATURE", "0.2"))
+    top_p: float = float(os.getenv("TOP_P", "0.95"))
+    use_rag: bool = os.getenv("USE_RAG", "true").lower() == "true"
+    force_mock_mode: bool = os.getenv("FORCE_MOCK_MODE", "false").lower() == "true"
+    api_key: str = os.getenv("API_KEY", "")
+    rate_limit_per_minute: int = int(os.getenv("RATE_LIMIT_PER_MINUTE", "30"))
+
+
+settings = Settings()

src/generator.py

@@ -0,0 +1,86 @@
+"""Text generation and output parsing."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import List, Tuple
+
+from torch.nn.functional import softmax
+
+from src.config import settings
+
+
+@dataclass
+class GenerationResult:
+    code: str
+    explanation: str
+    confidence: float
+    important_tokens: List[str]
+
+
+def _split_code_and_explanation(text: str) -> Tuple[str, str]:
+    marker = "Explanation:"
+    if marker in text:
+        code, explanation = text.split(marker, 1)
+        return code.strip(), explanation.strip()
+    return text.strip(), "Model did not provide explicit explanation."
+
+
+def generate_response(model_bundle, prompt: str) -> GenerationResult:
+    """Generate model response with token-level confidence signals."""
+    if getattr(model_bundle, "is_mock", False):
+        # Keep API runnable even when model download/loading is unavailable.
+        fallback_code = (
+            "def solve_task(input_data):\n"
+            "    \"\"\"Fallback implementation when model is unavailable.\"\"\"\n"
+            "    return input_data\n"
+        )
+        fallback_explanation = (
+            "Running in mock fallback mode because no pretrained model could be loaded. "
+            "Set MODEL_NAME/FALLBACK_MODEL_NAME and ensure network/model access."
+        )
+        load_error = getattr(model_bundle, "load_error", "")
+        if load_error:
+            fallback_explanation = f"{fallback_explanation}\n\nLoader error: {load_error}"
+        return GenerationResult(
+            code=fallback_code,
+            explanation=fallback_explanation,
+            confidence=0.15,
+            important_tokens=["<mock-fallback>"],
+        )
+
+    tokenizer = model_bundle.tokenizer
+    model = model_bundle.model
+
+    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
+    outputs = model.generate(
+        **inputs,
+        max_new_tokens=settings.max_new_tokens,
+        temperature=settings.temperature,
+        top_p=settings.top_p,
+        do_sample=True,
+        return_dict_in_generate=True,
+        output_scores=True,
+    )
+
+    generated_ids = outputs.sequences[0][inputs["input_ids"].shape[1] :]
+    generated_text = tokenizer.decode(generated_ids, skip_special_tokens=True)
+
+    token_probs = []
+    important = []
+    for step_scores, token_id in zip(outputs.scores, generated_ids):
+        probs = softmax(step_scores[0], dim=-1)
+        p = probs[token_id].item()
+        token_probs.append(p)
+        if p < 0.30:
+            important.append(tokenizer.decode([token_id]))
+
+    confidence = float(sum(token_probs) / max(len(token_probs), 1))
+    code, explanation = _split_code_and_explanation(generated_text)
+
+    return GenerationResult(
+        code=code,
+        explanation=explanation,
+        confidence=confidence,
+        important_tokens=important[:20],
+    )

src/hallucination.py

@@ -0,0 +1,58 @@
+"""Simple hallucination checks for generated code."""
+
+from __future__ import annotations
+
+import ast
+import subprocess
+import sys
+import tempfile
+from dataclasses import dataclass
+
+
+@dataclass
+class HallucinationCheckResult:
+    hallucination: bool
+    reason: str
+
+
+def _python_syntax_check(code: str) -> HallucinationCheckResult:
+    try:
+        ast.parse(code)
+        return HallucinationCheckResult(False, "Syntax is valid.")
+    except SyntaxError as exc:
+        return HallucinationCheckResult(True, f"Syntax error: {exc}")
+
+
+def _python_runtime_smoke_test(code: str) -> HallucinationCheckResult:
+    try:
+        with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as tmp:
+            tmp.write(code)
+            tmp_path = tmp.name
+        proc = subprocess.run(
+            [sys.executable, tmp_path],
+            capture_output=True,
+            text=True,
+            timeout=4,
+            check=False,
+        )
+        if proc.returncode != 0:
+            stderr = proc.stderr.strip()[:300]
+            return HallucinationCheckResult(True, f"Runtime test failed: {stderr}")
+        return HallucinationCheckResult(False, "Runtime smoke test passed.")
+    except subprocess.TimeoutExpired:
+        return HallucinationCheckResult(True, "Runtime test timed out.")
+    except Exception as exc:  # pragma: no cover
+        return HallucinationCheckResult(True, f"Runtime test error: {exc}")
+
+
+def hallucination_check(code: str) -> HallucinationCheckResult:
+    """Run syntax and runtime checks and consolidate status."""
+    # Skip expensive/irrelevant execution for obviously non-Python snippets.
+    python_hints = ("def ", "class ", "import ", "from ", "print(", "if __name__")
+    if not any(hint in code for hint in python_hints):
+        return HallucinationCheckResult(False, "Skipped runtime check for non-Python-like output.")
+
+    syntax_result = _python_syntax_check(code)
+    if syntax_result.hallucination:
+        return syntax_result
+    return _python_runtime_smoke_test(code)

src/lora_prepare.py

@@ -0,0 +1,18 @@
+"""Starter utilities for future LoRA fine-tuning."""
+
+from __future__ import annotations
+
+from peft import LoraConfig, get_peft_model
+
+
+def attach_lora(model):
+    """Attach a default LoRA adapter to a loaded model."""
+    config = LoraConfig(
+        r=16,
+        lora_alpha=32,
+        target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
+        lora_dropout=0.05,
+        bias="none",
+        task_type="CAUSAL_LM",
+    )
+    return get_peft_model(model, config)

src/model_loader.py

@@ -0,0 +1,64 @@
+"""Model loading utilities."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from transformers import AutoModelForCausalLM, AutoTokenizer
+
+from src.config import settings
+
+
+@dataclass
+class ModelBundle:
+    """Holds tokenizer and model together."""
+
+    tokenizer: Any
+    model: Any
+    active_model_name: str
+    is_mock: bool = False
+    load_error: str = ""
+
+
+def load_model_bundle() -> ModelBundle:
+    """Load Qwen2.5-Coder first, then fallback if needed."""
+    if settings.force_mock_mode:
+        return ModelBundle(
+            tokenizer=None,
+            model=None,
+            active_model_name="mock-rule-based",
+            is_mock=True,
+            load_error="FORCE_MOCK_MODE=true",
+        )
+
+    candidate_models = [
+        settings.model_name,
+        settings.fallback_model_name,
+        settings.final_fallback_model_name,
+    ]
+    last_error = None
+
+    for model_name in candidate_models:
+        try:
+            tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
+            model = AutoModelForCausalLM.from_pretrained(
+                model_name,
+                trust_remote_code=True,
+                device_map="auto",
+            )
+            return ModelBundle(
+                tokenizer=tokenizer,
+                model=model,
+                active_model_name=model_name,
+            )
+        except Exception as exc:  # pragma: no cover
+            last_error = exc
+
+    return ModelBundle(
+        tokenizer=None,
+        model=None,
+        active_model_name="mock-rule-based",
+        is_mock=True,
+        load_error=str(last_error),
+    )

src/pipeline.py

@@ -0,0 +1,58 @@
+"""End-to-end orchestration for generation and validation."""
+
+from __future__ import annotations
+
+import time
+
+from src.config import settings
+from src.generator import generate_response
+from src.hallucination import hallucination_check
+from src.model_loader import load_model_bundle
+from src.prompts import SYSTEM_PROMPT, build_user_prompt
+from src.rag import CodeRAG
+from src.relevancy import RelevancyScorer
+
+
+class CodingLLMPipeline:
+    """Coordinates model, RAG, explainability, and quality checks."""
+
+    def __init__(self):
+        self.model_bundle = None
+        self.relevancy = RelevancyScorer()
+        self.rag = CodeRAG() if settings.use_rag else None
+
+    def _ensure_model_loaded(self):
+        if self.model_bundle is None:
+            self.model_bundle = load_model_bundle()
+
+    def run(self, instruction: str, user_input: str) -> dict:
+        started = time.perf_counter()
+        self._ensure_model_loaded()
+
+        query_text = f"{instruction}\n{user_input}".strip()
+        retrieved_context = self.rag.retrieve(query_text) if self.rag else ""
+
+        prompt = f"{SYSTEM_PROMPT}\n\n{build_user_prompt(instruction, user_input, retrieved_context)}"
+        generation = generate_response(self.model_bundle, prompt)
+        hallucination_result = hallucination_check(generation.code)
+        relevancy_score = self.relevancy.score(query_text, generation.code)
+        explanation = generation.explanation
+        if hallucination_result.hallucination:
+            explanation = f"{generation.explanation}\n\nHallucination check reason: {hallucination_result.reason}"
+
+        latency_ms = int((time.perf_counter() - started) * 1000)
+        return {
+            "code": generation.code,
+            "explanation": explanation,
+            "confidence": round(generation.confidence, 4),
+            "important_tokens": generation.important_tokens,
+            "relevancy_score": round(relevancy_score, 4),
+            "hallucination": hallucination_result.hallucination,
+            "latency_ms": latency_ms,
+        }
+
+    @property
+    def active_model_name(self) -> str:
+        """Current model name, loading lazily if needed."""
+        self._ensure_model_loaded()
+        return self.model_bundle.active_model_name

src/prompts.py

@@ -0,0 +1,19 @@
+"""Prompt templates for coding assistant tasks."""
+
+SYSTEM_PROMPT = """You are an advanced coding model.
+Follow instructions exactly and produce high-quality code.
+Always include:
+1) Final code
+2) A short explanation of what and why
+"""
+
+
+def build_user_prompt(instruction: str, user_input: str, retrieved_context: str = "") -> str:
+    """Build a single prompt string with optional RAG context."""
+    context_block = f"\nRelevant snippets:\n{retrieved_context}\n" if retrieved_context else ""
+    return (
+        f"Instruction:\n{instruction}\n"
+        f"{context_block}"
+        f"\nInput:\n{user_input}\n"
+        "\nRespond with code first, then explanation."
+    )

src/rag.py

@@ -0,0 +1,57 @@
+"""Basic FAISS RAG retriever for code snippets."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import List
+
+import faiss
+import numpy as np
+from sklearn.feature_extraction.text import TfidfVectorizer
+
+
+class CodeRAG:
+    """Loads snippets from data and serves top-k retrieval."""
+
+    def __init__(self, data_path: str = "data/sample_snippets.json"):
+        project_root = Path(__file__).resolve().parents[1]
+        self.data_path = project_root / data_path
+        self.snippets = self._load_data()
+        self.vectorizer = TfidfVectorizer(ngram_range=(1, 2), min_df=1)
+        self.snippet_vectors = self._build_vectors(self.snippets)
+        self.index = self._build_index(self.snippet_vectors)
+
+    def _load_data(self) -> List[str]:
+        if not self.data_path.exists():
+            return []
+        payload = json.loads(self.data_path.read_text(encoding="utf-8"))
+        return [item["snippet"] for item in payload]
+
+    def _build_vectors(self, snippets: List[str]) -> np.ndarray | None:
+        if not snippets:
+            return None
+        matrix = self.vectorizer.fit_transform(snippets)
+        vectors = matrix.toarray().astype(np.float32)
+        # Normalize for cosine-like similarity with IndexFlatIP.
+        norms = np.linalg.norm(vectors, axis=1, keepdims=True)
+        norms[norms == 0.0] = 1.0
+        return vectors / norms
+
+    def _build_index(self, vectors: np.ndarray | None):
+        if vectors is None:
+            return None
+        index = faiss.IndexFlatIP(vectors.shape[1])
+        index.add(vectors)
+        return index
+
+    def retrieve(self, query: str, top_k: int = 2) -> str:
+        if not self.snippets or self.index is None or self.snippet_vectors is None:
+            return ""
+        query_vec = self.vectorizer.transform([query]).toarray().astype(np.float32)
+        qnorm = np.linalg.norm(query_vec, axis=1, keepdims=True)
+        qnorm[qnorm == 0.0] = 1.0
+        query_vec = query_vec / qnorm
+        _, idx = self.index.search(query_vec, min(top_k, len(self.snippets)))
+        selected = [self.snippets[i] for i in idx[0] if i >= 0]
+        return "\n\n".join(selected)

src/relevancy.py

@@ -0,0 +1,17 @@
+"""Lightweight relevancy scoring without heavy embedding backends."""
+
+from __future__ import annotations
+
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.metrics.pairwise import cosine_similarity
+
+
+class RelevancyScorer:
+    """Computes semantic relevancy between request and generated code."""
+
+    def __init__(self):
+        self.vectorizer = TfidfVectorizer(ngram_range=(1, 2), min_df=1)
+
+    def score(self, query_text: str, generated_text: str) -> float:
+        matrix = self.vectorizer.fit_transform([query_text, generated_text])
+        return float(cosine_similarity(matrix[0], matrix[1])[0][0])

src/schemas.py

@@ -0,0 +1,18 @@
+"""API schema definitions."""
+
+from pydantic import BaseModel, Field
+
+
+class GenerateRequest(BaseModel):
+    instruction: str = Field(..., min_length=1)
+    input: str = Field("", description="Source code or problem context.")
+
+
+class GenerateResponse(BaseModel):
+    code: str
+    explanation: str
+    confidence: float
+    important_tokens: list[str]
+    relevancy_score: float
+    hallucination: bool
+    latency_ms: int

tasks.py

@@ -0,0 +1,111 @@
+"""Simple task runner for common project workflows."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import subprocess
+import sys
+import time
+
+
+def run_cmd(cmd: list[str]) -> int:
+    print(">", " ".join(cmd))
+    return subprocess.call(cmd)
+
+
+def task_install() -> int:
+    return run_cmd([sys.executable, "-m", "pip", "install", "-r", "requirements.txt"])
+
+
+def task_run() -> int:
+    return run_cmd([sys.executable, "-m", "uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"])
+
+
+def task_space() -> int:
+    return run_cmd([sys.executable, "app.py"])
+
+
+def task_smoke() -> int:
+    return run_cmd([sys.executable, "smoke_test.py"])
+
+
+def task_compile() -> int:
+    return run_cmd([sys.executable, "-m", "compileall", "."])
+
+
+def task_docker_up() -> int:
+    return run_cmd(["docker", "compose", "up", "--build"])
+
+
+def task_docker_down() -> int:
+    return run_cmd(["docker", "compose", "down"])
+
+
+def task_hf_upload(repo_id: str, token: str, private: bool) -> int:
+    cmd = [sys.executable, "upload_to_hf.py", "--repo-id", repo_id, "--token", token]
+    if private:
+        cmd.append("--private")
+    return run_cmd(cmd)
+
+
+def task_serve_smoke() -> int:
+    """Start API, run smoke test, then stop API."""
+    env = os.environ.copy()
+    server_cmd = [sys.executable, "-m", "uvicorn", "api.main:app", "--host", "127.0.0.1", "--port", "8000"]
+    print(">", " ".join(server_cmd))
+    server = subprocess.Popen(server_cmd, env=env)
+    try:
+        time.sleep(3)
+        return run_cmd([sys.executable, "smoke_test.py"])
+    finally:
+        server.terminate()
+        try:
+            server.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            server.kill()
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Project task runner")
+    sub = parser.add_subparsers(dest="task", required=True)
+
+    sub.add_parser("install")
+    sub.add_parser("run")
+    sub.add_parser("space")
+    sub.add_parser("smoke")
+    sub.add_parser("compile")
+    sub.add_parser("docker-up")
+    sub.add_parser("docker-down")
+    sub.add_parser("serve-smoke")
+    hf = sub.add_parser("hf-upload")
+    hf.add_argument("--repo-id", required=True)
+    hf.add_argument("--token", required=True)
+    hf.add_argument("--private", action="store_true")
+
+    args = parser.parse_args()
+
+    if args.task == "install":
+        code = task_install()
+    elif args.task == "run":
+        code = task_run()
+    elif args.task == "space":
+        code = task_space()
+    elif args.task == "smoke":
+        code = task_smoke()
+    elif args.task == "compile":
+        code = task_compile()
+    elif args.task == "docker-up":
+        code = task_docker_up()
+    elif args.task == "docker-down":
+        code = task_docker_down()
+    elif args.task == "serve-smoke":
+        code = task_serve_smoke()
+    else:
+        code = task_hf_upload(args.repo_id, args.token, args.private)
+
+    raise SystemExit(code)
+
+
+if __name__ == "__main__":
+    main()

upload_to_hf.py

@@ -0,0 +1,40 @@
+"""Upload this project as a Hugging Face Space."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from huggingface_hub import HfApi, create_repo, upload_folder
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Upload coding-llm project to Hugging Face Space.")
+    parser.add_argument("--repo-id", required=True, help="Example: username/coding-llm-space")
+    parser.add_argument("--token", required=True, help="Hugging Face access token")
+    parser.add_argument("--private", action="store_true", help="Create private repo/space")
+    args = parser.parse_args()
+
+    root = Path(__file__).resolve().parent
+    api = HfApi(token=args.token)
+    create_repo(
+        repo_id=args.repo_id,
+        token=args.token,
+        repo_type="space",
+        private=args.private,
+        exist_ok=True,
+        space_sdk="gradio",
+    )
+    upload_folder(
+        folder_path=str(root),
+        repo_id=args.repo_id,
+        repo_type="space",
+        token=args.token,
+        ignore_patterns=[".git*", "__pycache__", "*.pyc"],
+    )
+    info = api.repo_info(repo_id=args.repo_id, repo_type="space")
+    print(f"Uploaded successfully: {info.id}")
+
+
+if __name__ == "__main__":
+    main()