llm.py

"""
LLM verification — Gemma 4 12B on transformers + ZeroGPU.
Utilizes Gemma 4's raw image patch projection to read native Bengali without OCR encoders.
"""

import io, json, os, fitz, spaces, transformers
from PIL import Image
from transformers import AutoProcessor

# Configured for Gemma 4 12B (decoder-only multimodal)
MODEL_ID = os.environ.get("PORCHA_MODEL_ID", "google/gemma-4-12b-it")

_MAX_IMAGE_SIDE = 896
_MAX_IMAGES = 10
_PDF_DPI = 150

class LLMConfigError(RuntimeError): pass

def _model_class():
    for name in ("AutoModelForImageTextToText", "AutoModelForCausalLM"):
        cls = getattr(transformers, name, None)
        if cls is not None: return cls
    raise LLMConfigError("No suitable model class found.")

try:
    processor = AutoProcessor.from_pretrained(MODEL_ID)
    model = _model_class().from_pretrained(MODEL_ID, dtype="auto", device_map="cuda")
except Exception as e:
    processor, model, _LOAD_ERROR = None, None, e
else:
    _LOAD_ERROR = None

def _downscale(img):
    img = img.convert("RGB")
    w, h = img.size
    scale = min(1.0, _MAX_IMAGE_SIDE / max(w, h))
    if scale < 1.0: img = img.resize((max(1, int(w * scale)), max(1, int(h * scale))))
    return img

def _file_to_images(path):
    if path.lower().endswith(".pdf"):
        imgs = []
        with fitz.open(path) as doc:
            for page in doc:
                pix = page.get_pixmap(dpi=_PDF_DPI)
                imgs.append(_downscale(Image.open(io.BytesIO(pix.tobytes("png")))))
        return imgs
    try: return [_downscale(Image.open(path))]
    except: return []

def _system_prompt():
    return (
        "You are an expert West Bengal Land Revenue Officer (BL&LRO). You review "
        "property documents like Porchas and Dalils written in English and Bengali. "
        "You ingest raw images natively. Read the Bengali text, extract details, and "
        "verify against the strict guidelines provided.\n\n"
        "1. For each Document ID, check if it is present. \n"
        "2. For each Rule ID, check if the condition is met (e.g., 'Bastu' vs 'Sali'). "
        "Translate Bengali keywords internally, but quote the original Bengali word "
        "in your 'evidence' string.\n"
        "3. Output strictly as JSON. No prose. Format:\n"
        "{\n"
        '  "documents": [{"id": str, "present": bool, "confidence": "high|medium|low", "evidence": str}],\n'
        '  "rules": [{"id": str, "status": "pass|fail|unclear", "evidence": str}],\n'
        '  "overall": "ready|not_ready",\n'
        '  "summary": str\n'
        "}"
    )

def _build_messages(pkg, stage_label, docs, rules, files):
    text = f"{_system_prompt()}\n\nTRANSACTION: {pkg['name']}\nSTAGE: {stage_label}\n"
    text += "DOCUMENTS:\n" + "\n".join([f"- id={d['id']} | {d['label']}" for d in docs])
    if rules:
        text += "\nRULES:\n" + "\n".join([f"- id={r['id']} | {r['check']}" for r in rules])
    
    content = [{"type": "text", "text": text}]
    n = 0
    for path in files:
        if n >= _MAX_IMAGES: break
        for img in _file_to_images(path):
            if n >= _MAX_IMAGES: break
            content.append({"type": "image", "image": img})
            n += 1
    return [{"role": "user", "content": content}]

@spaces.GPU(duration=120)
def _generate(messages, max_new_tokens):
    inputs = processor.apply_chat_template(messages, tokenize=True, return_dict=True, return_tensors="pt", add_generation_prompt=True).to(model.device)
    out = model.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=False)
    gen = out[0][inputs["input_ids"].shape[-1]:]
    return processor.decode(gen, skip_special_tokens=True)

def verify(pkg, stage_label, docs, rules, files):
    if model is None: raise LLMConfigError(f"Model error: {_LOAD_ERROR}")
    messages = _build_messages(pkg, stage_label, docs, rules, files)
    raw = _generate(messages, 1024)
    
    # Robust JSON extraction
    text = raw.strip().strip("`")
    if text.lower().startswith("json"): text = text[4:].strip()
    try:
        parsed = json.loads(text)
    except json.JSONDecodeError:
        start, end = text.find("{"), text.rfind("}")
        if start != -1 and end != -1:
            try: parsed = json.loads(text[start : end + 1])
            except: return f"## ⚠️ Output Parse Error\n\n```\n{raw}\n```"
        else:
            return f"## ⚠️ Output Parse Error\n\n```\n{raw}\n```"
            
    # Rendering Logic
    lines = [f"## Verification Report: {pkg['name']}", f"**Status:** {'✅ Looks Good' if parsed.get('overall') == 'ready' else '⚠️ Discrepancies Found'}", f"> {parsed.get('summary', '')}\n", "### Document Verification"]
    
    res_docs = {d["id"]: d for d in parsed.get("documents", [])}
    for d in docs:
        r = res_docs.get(d["id"], {})
        icon = "✅" if r.get("present") else "❌"
        lines.append(f"- {icon} **{d['label']}**: {r.get('evidence', 'Not evaluated')}")
        
    if rules:
        lines.append("\n### Legal & Classification Checks")
        res_rules = {r["id"]: r for r in parsed.get("rules", [])}
        for rule in rules:
            r = res_rules.get(rule["id"], {})
            icon = "✅" if r.get("status") == "pass" else ("❌" if r.get("status") == "fail" else "⚠️")
            lines.append(f"- {icon} {rule['check']}\n  _Found: {r.get('evidence', 'Unclear')}_")
            
    return "\n".join(lines)