""" AI Assignment Grader — app.py HuggingFace Spaces / local entry point. For Google Colab usage, open AI_Grader_Complete_v2.ipynb instead. """ import json import os import re import subprocess import tempfile import textwrap import time import traceback import threading import gradio as gr import fitz # PyMuPDF import nbformat import requests from tenacity import retry, stop_after_attempt, wait_fixed # ───────────────────────────────────────────────────────────────────────────── # Config # ───────────────────────────────────────────────────────────────────────────── # MODEL_NAME = os.getenv("MODEL_NAME", "qwen2.5-coder:7b") MODEL_NAME = os.getenv("MODEL_NAME", "qwen2.5-coder:3b") OLLAMA_URL = "http://localhost:11434/api/chat" # MAX_CODE_LEN = 3500 MAX_CODE_LEN = 2000 TIMEOUT = 300 # ───────────────────────────────────────────────────────────────────────────── # File Parsers # ───────────────────────────────────────────────────────────────────────────── def extract_pdf_text(pdf_path: str) -> str: """Extract all text from a PDF using PyMuPDF.""" doc = fitz.open(pdf_path) pages = [page.get_text("text").strip() for page in doc] doc.close() full = "\n\n".join(p for p in pages if p) if not full.strip(): raise ValueError("PDF appears empty or image-only (no extractable text).") # return full[:1500] if len(full) > 1500 else full return full[:800] if len(full) > 800 else full def extract_notebook_code(ipynb_path: str) -> str: """Extract code + markdown cells from .ipynb with cell separators.""" nb = nbformat.read(ipynb_path, as_version=4) parts = [] for i, cell in enumerate(nb.cells): if cell.cell_type == "code": src = cell.source.strip() if src: parts.append(f"# ── Code Cell {i+1} ──\n{src}") elif cell.cell_type == "markdown": src = cell.source.strip() if src: parts.append( f"# ── Markdown Cell {i+1} ──\n# " + "\n# ".join(src.splitlines()) ) if not parts: raise ValueError("Notebook has no code cells.") code = "\n\n".join(parts) if len(code) > MAX_CODE_LEN: code = code[:MAX_CODE_LEN] + f"\n\n# ... [truncated to {MAX_CODE_LEN} chars]" return code def run_code_sandbox(ipynb_path: str, timeout: int = 30) -> dict: """Execute notebook code in a subprocess sandbox.""" try: nb = nbformat.read(ipynb_path, as_version=4) code_lines = [] for cell in nb.cells: if cell.cell_type == "code": lines = [ l for l in cell.source.splitlines() if not l.strip().startswith(("!", "%")) ] if lines: code_lines.extend(lines) code_lines.append("") with tempfile.NamedTemporaryFile(suffix=".py", delete=False, mode="w") as f: f.write("\n".join(code_lines)) tmp_py = f.name result = subprocess.run( ["python", tmp_py], capture_output=True, text=True, timeout=timeout ) os.unlink(tmp_py) return { "success": result.returncode == 0, "stdout": result.stdout[:2000], "stderr": result.stderr[:1000], "error": None, } except subprocess.TimeoutExpired: return { "success": False, "stdout": "", "stderr": "", "error": f"Timed out after {timeout}s", } except Exception as e: return {"success": False, "stdout": "", "stderr": "", "error": str(e)} # ───────────────────────────────────────────────────────────────────────────── # Prompts # ───────────────────────────────────────────────────────────────────────────── SYSTEM_PROMPT = textwrap.dedent( """ You are a strict programming instructor grading a Jupyter notebook. SCORING: Evaluate each rubric criterion. Score cannot exceed criterion max. Total cannot exceed 25. Verify sum before responding. RUBRIC CHECK: For every penalty rule violated, add one areas_to_improve entry with rubric_requirement set to that exact rule. One violation = one entry. FEEDBACK: Strengths must cite actual code. Issues must name exact variable/function. No vague feedback. No invented mistakes. Respond ONLY with this JSON: { "criterion_scores": [{"name": str, "score": int, "max": int}], "total_score": int, "strengths": [str], "areas_to_improve": [{ "category": "Bug|Code Quality|Data Preprocessing|Modeling|Missing Requirement", "rubric_requirement": str, "issue": str, "why_it_matters": str, "fix": str }] } """ ).strip() # SYSTEM_PROMPT = textwrap.dedent( # """ # You are a strict but fair programming instructor grading a student Jupyter notebook. # You will receive: # - An assignment question # - A grading rubric with criteria, point values, and explicit penalty rules # - The student code extracted from their notebook # - Optionally: sandbox execution output # SCORING RULES — follow exactly: # 1. Read each criterion in the rubric. Note its maximum points. # 2. Apply every penalty listed that applies to the student code. # 3. A criterion score CANNOT exceed its stated maximum. # 4. Sum all criterion scores. The total CANNOT exceed 25. # 5. If no rubric criterion mentions a topic, do not award or deduct for it. # 6. When in doubt, deduct — do not give benefit of the doubt. # RUBRIC CROSS-CHECK — mandatory: # Before writing areas_to_improve, scan every single rubric penalty rule. # For each rule, ask: "Did the student violate this?" # - YES → create one areas_to_improve entry with rubric_requirement set to that exact rule. # - NO → skip it. # Each violation = one separate entry. Do not merge separate issues. # FEEDBACK RULES: # - Strengths: cite actual code, function names, or techniques the student used. # - areas_to_improve entries must be specific and forensic. # - Do NOT be vague. "Improve variable names" is rejected. # - Do NOT invent mistakes not visible in the code. # - The sum of criterion_scores must equal total_score. Verify before responding. # - Respond ONLY with valid JSON. No markdown, no text outside the JSON. # Return exactly this schema: # { # "criterion_scores": [ # {"name": "", "score": , "max": } # ], # "total_score": , # "strengths": ["Specific strength with code reference"], # "areas_to_improve": [ # { # "category": "", # "rubric_requirement": "The exact rubric penalty rule that was violated", # "issue": "What the student did wrong or completely missed", # "why_it_matters": "The consequence or reason this is wrong", # "fix": "Concrete suggestion or corrected code snippet" # } # ] # } # """ # ).strip() def build_user_prompt(question, rubric, code, execution): parts = [ f"=== ASSIGNMENT QUESTION ===\n{question.strip()}", f"=== GRADING RUBRIC ===\n{rubric.strip()}", f"=== STUDENT CODE ===\n{code.strip()}", ] if execution: status = "✓ ran successfully" if execution["success"] else "✗ failed" block = f"Status: {status}\n" if execution["stdout"]: block += f"Output:\n{execution['stdout']}\n" if execution["stderr"]: block += f"Errors:\n{execution['stderr']}\n" if execution["error"]: block += f"Exception: {execution['error']}\n" parts.append(f"=== EXECUTION RESULTS ===\n{block}") parts.append( "=== YOUR TASK ===\n" "Go through every rubric penalty rule line by line.\n" "For each one violated, add an entry to areas_to_improve.\n" "Return ONLY the JSON schema specified." ) return "\n\n".join(parts) # ───────────────────────────────────────────────────────────────────────────── # Ollama call # ───────────────────────────────────────────────────────────────────────────── @retry(stop=stop_after_attempt(3), wait=wait_fixed(5)) def call_ollama(question, rubric, code, execution): payload = { "model": MODEL_NAME, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, { "role": "user", "content": build_user_prompt(question, rubric, code, execution), }, ], "stream": False, "format": "json", # "options": { # "temperature": 0.0, # "num_predict": 1500, # "num_ctx": 4096, # "num_thread": 4, # "num_batch": 512, # }, "options": { "temperature": 0.0, "num_predict": 800, # was 1500 — your JSON output is ~400 tokens max "num_ctx": 2048, # was 4096 — rubric + code fits in 2048 easily "num_thread": 2, # match exactly to free tier vCPU count "num_batch": 128, # smaller batch = less memory pressure on 2 vCPU }, } resp = requests.post(OLLAMA_URL, json=payload, timeout=TIMEOUT) resp.raise_for_status() raw = resp.json()["message"]["content"] raw = re.sub(r"^```json\s*", "", raw.strip()) raw = re.sub(r"```$", "", raw.strip()) result = json.loads(raw) # Python-side score guard — LLM cannot inflate scores criteria = result.get("criterion_scores", []) if criteria: for c in criteria: c["score"] = min(c.get("score", 0), c.get("max", 0)) computed = sum(c["score"] for c in criteria) result["total_score"] = min(computed, 25) else: result["total_score"] = min(result.get("total_score", 0), 25) return result # ───────────────────────────────────────────────────────────────────────────── # HTML renderer # ───────────────────────────────────────────────────────────────────────────── CATEGORY_COLORS = { "Bug": ("#fee2e2", "#dc2626", "#fca5a5"), "Code Quality": ("#fef9c3", "#ca8a04", "#fde047"), "Data Preprocessing": ("#ede9fe", "#7c3aed", "#c4b5fd"), "Modeling": ("#fff7ed", "#ea580c", "#fdba74"), "Missing Requirement": ("#f0f9ff", "#0284c7", "#7dd3fc"), } def grade_to_color(pct): if pct >= 0.85: return "#22c55e" if pct >= 0.70: return "#84cc16" if pct >= 0.55: return "#f59e0b" if pct >= 0.40: return "#f97316" return "#ef4444" def render_html_report(result, llm_elapsed=0.0, total_elapsed=0.0): total = result.get("total_score", 0) pct = total / 25 color = grade_to_color(pct) grade = ( "A+" if pct >= 0.90 else ( "A" if pct >= 0.80 else ( "B" if pct >= 0.70 else "C" if pct >= 0.60 else "D" if pct >= 0.50 else "F" ) ) ) # Criterion breakdown criteria = result.get("criterion_scores", []) crit_html = "" if criteria: rows = "" for c in criteria: c_pct = c["score"] / c["max"] if c.get("max") else 0 c_col = grade_to_color(c_pct) rows += f""" {c['name']} {c['score']}/{c['max']}
""" crit_html = f"""

Criterion Breakdown

{rows}
Criterion Score Progress
""" # Strengths strengths_li = "".join( f"
  • {s}
    • " for s in result.get("strengths", []) ) # Improvement cards improve_cards = "" for item in result.get("areas_to_improve", []): cat = item.get("category", "Code Quality") rubric_req = item.get("rubric_requirement", "") issue = item.get("issue", "") why = item.get("why_it_matters", "") fix = item.get("fix", "") bg, text, border = CATEGORY_COLORS.get(cat, ("#f8fafc", "#475569", "#cbd5e1")) rubric_badge = "" if rubric_req: rubric_badge = f"""
    RUBRIC "{rubric_req}"
    """ is_code = any( tok in fix for tok in ["\n", "def ", "df.", "import ", " = ", "()", "[]", ":"] ) fix_html = ( f"
    {fix}
    " if is_code else f"

    Fix: {fix}

    " ) improve_cards += f"""
    {cat} {issue}
    {rubric_badge}

    Why it matters: {why}

    {fix_html}
    """ if not improve_cards: improve_cards = "
    No rubric violations found.
    " n_issues = len(result.get("areas_to_improve", [])) return f"""
    Total Score
    {total} / 25
    {int(pct*100)}% · Qwen2.5-Coder via Ollama
    {grade}
    Grade
    {crit_html}

    Strengths

      {strengths_li or "
    • No specific strengths identified.
      • "}

    Areas to Improve ({n_issues} rubric violation{'s' if n_issues!=1 else ''} found)

    Each card shows the exact rubric rule violated (dark banner), what went wrong, why it matters, and how to fix it.

    {improve_cards}
    LLM Inference
    {llm_elapsed:.1f}s
    Total Time
    {total_elapsed:.1f}s
    Model
    Qwen2.5-Coder via Ollama
    AI Assignment Grader
    """ # ───────────────────────────────────────────────────────────────────────────── # Status helpers # ───────────────────────────────────────────────────────────────────────────── def _step(icon, label, msg, elapsed_str=""): badge = ( f"⏱ {elapsed_str}" if elapsed_str else "" ) return ( f"
    " f"{badge}{icon} {label}: {msg}
    " ) def _timer_html(seconds): mins, secs = int(seconds) // 60, int(seconds) % 60 time_str = f"{mins}m {secs:02d}s" if mins > 0 else f"{secs}s" pulse_w = int((seconds % 10) / 10 * 100) return f"""
    ⚙️ Evaluating with Qwen2.5-Coder ⏱ {time_str}
    LLM inference in progress — typically 30–90 seconds on CPU
    """ def _done_status(llm_elapsed, total_elapsed): return f"""
    ✅ Grading complete
    LLM inference
    {llm_elapsed:.1f}s
    Total time
    {total_elapsed:.1f}s
    """ # ───────────────────────────────────────────────────────────────────────────── # Main grading function # ───────────────────────────────────────────────────────────────────────────── def grade_assignment(question_pdf, rubric_txt, notebook_ipynb, run_code): overall_start = time.time() step_log = "" def _elapsed(): return f"{time.time() - overall_start:.1f}s" try: if question_pdf is None: yield None, "

    ❌ Please upload the assignment PDF.

    ", "" return if rubric_txt is None: yield None, "

    ❌ Please upload the rubric TXT file.

    ", "" return if notebook_ipynb is None: yield None, "

    ❌ Please upload the student notebook (.ipynb).

    ", "" return pdf_path = question_pdf if isinstance(question_pdf, str) else question_pdf.name rubric_path = rubric_txt if isinstance(rubric_txt, str) else rubric_txt.name nb_path = ( notebook_ipynb if isinstance(notebook_ipynb, str) else notebook_ipynb.name ) # Step 1 — Parse step_log = _step("⏳", "Step 1/3", "Parsing files...") yield None, None, step_log question = extract_pdf_text(pdf_path) with open(rubric_path, "r", encoding="utf-8") as f: rubric = f.read() if not rubric.strip(): yield None, "

    ❌ Rubric file is empty.

    ", "" return code = extract_notebook_code(nb_path) step_log = _step( "✅", "Step 1/3 complete", f"PDF: {len(question):,} chars · Rubric: {len(rubric):,} chars · Code: {len(code):,} chars", _elapsed(), ) yield None, None, step_log # Step 2 — Sandbox execution = None if run_code: step_log += _step("⏳", "Step 2/3", "Running notebook in sandbox...") yield None, None, step_log t0 = time.time() execution = run_code_sandbox(nb_path, timeout=30) icon = "✅" if execution["success"] else "⚠️" msg = ( "Ran successfully" if execution["success"] else (execution.get("error") or execution.get("stderr", ""))[:80] ) step_log += _step( icon, "Step 2/3 complete", f"{msg} ({time.time()-t0:.1f}s)", _elapsed() ) else: step_log += _step("⏭️", "Step 2/3", "Code execution skipped.", _elapsed()) yield None, None, step_log # Step 3 — LLM in background thread step_log += _step( "🧠", "Step 3/3", "Dispatching to Qwen2.5-Coder...", _elapsed() ) yield None, None, step_log llm_result, llm_error = {}, {} llm_start = time.time() def _run(): try: llm_result["data"] = call_ollama(question, rubric, code, execution) except Exception as e: llm_error["err"] = e thread = threading.Thread(target=_run, daemon=True) thread.start() while thread.is_alive(): yield None, None, step_log + _timer_html(time.time() - llm_start) time.sleep(1) thread.join() if "err" in llm_error: raise llm_error["err"] llm_elapsed = time.time() - llm_start total_elapsed = time.time() - overall_start result = llm_result["data"] html = render_html_report(result, llm_elapsed, total_elapsed) json_out = json.dumps(result, indent=2) yield json_out, html, _done_status(llm_elapsed, total_elapsed) except Exception as e: tb = traceback.format_exc() yield None, f"""
    ❌ Error: {e} 
    {tb}
    """, "" # ───────────────────────────────────────────────────────────────────────────── # Gradio UI # ───────────────────────────────────────────────────────────────────────────── with gr.Blocks( title="AI Assignment Grader", theme=gr.themes.Soft(primary_hue="slate"), css=""" .upload-box { border: 2px dashed #cbd5e1 !important; border-radius: 10px !important; } .grade-btn { background: linear-gradient(135deg,#1e293b,#334155) !important; color: white !important; font-weight: 700 !important; font-size: 16px !important; height: 52px !important; } .status-box { min-height: 44px; } """, ) as demo: gr.HTML( """

    🎓 AI Assignment Grader

    Powered by Qwen2.5-Coder via Ollama · Score · Strengths · Rubric-aware Feedback

    """ ) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 📂 Upload Files") question_pdf = gr.File( label="📄 Assignment Question (PDF)", file_types=[".pdf"], elem_classes=["upload-box"], ) rubric_txt = gr.File( label="📋 Grading Rubric (TXT)", file_types=[".txt"], elem_classes=["upload-box"], ) notebook_ipynb = gr.File( label="📓 Student Notebook (IPYNB)", file_types=[".ipynb"], elem_classes=["upload-box"], ) run_code = gr.Checkbox( label="⚙️ Run code in sandbox (30s timeout)", value=False, info="Executes the notebook and feeds output to the LLM.", ) grade_btn = gr.Button( "🚀 Grade Assignment", variant="primary", elem_classes=["grade-btn"] ) status_box = gr.HTML(value="", elem_classes=["status-box"]) gr.Markdown( """ --- **Output:** Score / 25 · Per-criterion breakdown · Strengths · Rubric violations with fixes **Tips** - PDF must have selectable text - Use explicit penalty rules in rubric for best feedback - First load: model download (~5 min) · Grading: 30–90 sec """ ) with gr.Column(scale=2): gr.Markdown("### 📊 Results") with gr.Tabs(): with gr.TabItem("📋 Feedback Report"): report_html = gr.HTML( value="""
    Upload files and click 🚀 Grade Assignment to begin.
    """ ) with gr.TabItem("🔧 Raw JSON"): json_output = gr.Code( language="json", label="Raw grading output", lines=30 ) grade_btn.click( fn=grade_assignment, inputs=[question_pdf, rubric_txt, notebook_ipynb, run_code], outputs=[json_output, report_html, status_box], ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)