analyzer model

app/models/analyzer.py

@@ -1,124 +1,94 @@
-import logging
+"""
+analyzer.py – Smart Contract Auditor
+LLM inference via HF Inference API (no local model loading → no OOM).
+Pattern analysis always runs as baseline.
+"""
 import os
+import logging
+import requests
 from typing import Optional, Dict, Any
 
-import torch
-from app.config import settings
 from app.models.patterns import analyze_with_patterns
 
 logger = logging.getLogger(__name__)
 
-_model = None
-_tokenizer = None
-_model_loaded: bool = False
+# ── Config from environment ──────────────────────────────────────────────────
+HF_MODEL_ID  = os.getenv("HF_MODEL_ID", "")
+HF_TOKEN     = os.getenv("HF_TOKEN", "")
+MAX_TOKENS   = int(os.getenv("MAX_NEW_TOKENS", "300"))
+TEMPERATURE  = float(os.getenv("TEMPERATURE", "0.7"))
 
 
-def _load_model() -> bool:
-    """Lazy-load the fine-tuned model (called once on first inference)."""
-    global _model, _tokenizer, _model_loaded
+def _call_hf_inference_api(solidity_code: str) -> Optional[str]:
+    """
+    Call HuggingFace Serverless Inference API.
+    Free tier: ~30k tokens/month, no GPU needed in the Space.
+    """
+    if not HF_MODEL_ID:
+        logger.info("HF_MODEL_ID not set – skipping LLM analysis.")
+        return None
 
-    if _model_loaded:
-        return True
+    api_url = f"https://api-inference.huggingface.co/models/{HF_MODEL_ID}"
+
+    headers = {"Content-Type": "application/json"}
+    if HF_TOKEN:
+        headers["Authorization"] = f"Bearer {HF_TOKEN}"
+
+    prompt = (
+        "<|system|>\n"
+        "You are an expert Solidity security auditor. "
+        "Analyze smart contracts for vulnerabilities and provide concise security reports.\n"
+        "<|endoftext|>\n\n"
+        "<|user|>\n"
+        f"Analyze this Solidity contract for security vulnerabilities:\n\n"
+        f"```solidity\n{solidity_code[:1500]}\n```\n"
+        "<|endoftext|>\n\n"
+        "<|assistant|>\n"
+    )
+
+    payload = {
+        "inputs": prompt,
+        "parameters": {
+            "max_new_tokens": MAX_TOKENS,
+            "temperature": TEMPERATURE,
+            "return_full_text": False,
+            "do_sample": True,
+        },
+        "options": {
+            "wait_for_model": True,
+            "use_cache": False,
+        },
+    }
 
     try:
-        from transformers import AutoTokenizer, AutoModelForCausalLM
-        from peft import PeftModel, PeftConfig
-
-        # Prefer HF Hub model over local path
-        model_source = settings.HF_MODEL_ID or settings.MODEL_PATH
-
-        if not settings.HF_MODEL_ID and not os.path.exists(settings.MODEL_PATH):
-            logger.warning("No model found – falling back to pattern analysis only.")
-            return False
-
-        logger.info(f"Loading model from: {model_source}")
-
-        hf_kwargs = {}
-        if settings.HF_TOKEN:
-            hf_kwargs["token"] = settings.HF_TOKEN
-
-        peft_config = PeftConfig.from_pretrained(model_source, **hf_kwargs)
-        base_name = peft_config.base_model_name_or_path
-
-        base_model = AutoModelForCausalLM.from_pretrained(
-            base_name,
-            torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
-            device_map="auto" if torch.cuda.is_available() else None,
-            low_cpu_mem_usage=True,
-            **hf_kwargs,
-        )
+        resp = requests.post(api_url, headers=headers, json=payload, timeout=60)
 
-        _model = PeftModel.from_pretrained(base_model, model_source, **hf_kwargs)
-        _tokenizer = AutoTokenizer.from_pretrained(model_source, **hf_kwargs)
+        if resp.status_code == 503:
+            logger.warning("Model is loading on HF side, retrying once…")
+            import time; time.sleep(20)
+            resp = requests.post(api_url, headers=headers, json=payload, timeout=60)
 
-        if _tokenizer.pad_token is None:
-            _tokenizer.pad_token = _tokenizer.eos_token
+        if resp.status_code != 200:
+            logger.error(f"HF API error {resp.status_code}: {resp.text[:200]}")
+            return None
 
-        _model.eval()
-        _model_loaded = True
-        logger.info("✅ Model loaded successfully.")
-        return True
+        data = resp.json()
+        if isinstance(data, list) and data:
+            return data[0].get("generated_text", "").strip()
 
-    except Exception as exc:
-        logger.error(f"Model load failed: {exc}")
-        return False
-
-
-def _llm_analyze(solidity_code: str) -> Optional[str]:
-    """Run LLM inference. Returns raw text or None on failure."""
-    if not _load_model():
         return None
 
-    try:
-        prompt = (
-            "<|system|>\nYou are an expert Solidity security auditor. "
-            "Analyze smart contracts for vulnerabilities and provide concise security reports.\n"
-            "<|endoftext|>\n\n"
-            "<|user|>\nAnalyze this Solidity contract for security vulnerabilities:\n\n"
-            f"```solidity\n{solidity_code[:settings.MAX_INPUT_LENGTH]}\n```\n<|endoftext|>\n\n"
-            "<|assistant|>\n"
-        )
-
-        inputs = _tokenizer(
-            prompt,
-            return_tensors="pt",
-            truncation=True,
-            max_length=512,
-        )
-        device = next(_model.parameters()).device
-        inputs = {k: v.to(device) for k, v in inputs.items()}
-
-        with torch.no_grad():
-            outputs = _model.generate(
-                **inputs,
-                max_new_tokens=settings.MAX_NEW_TOKENS,
-                temperature=settings.TEMPERATURE,
-                do_sample=True,
-                pad_token_id=_tokenizer.pad_token_id,
-                use_cache=False,
-            )
-
-        generated = outputs[0][len(inputs["input_ids"][0]):]
-        return _tokenizer.decode(generated, skip_special_tokens=True).strip()
-
+    except requests.exceptions.Timeout:
+        logger.error("HF Inference API timed out.")
+        return None
     except Exception as exc:
-        logger.error(f"LLM inference error: {exc}")
+        logger.error(f"HF Inference API call failed: {exc}")
         return None
 
 
 def analyze_contract(solidity_code: str) -> Dict[str, Any]:
-    """
-    Main entry point.
-    1. Always run pattern analysis (fast, reliable).
-    2. If model is available, also run LLM analysis.
-    3. Return combined result.
-    """
     pattern_result = analyze_with_patterns(solidity_code)
-
-    llm_text: Optional[str] = None
-    if not settings.USE_PATTERN_FALLBACK or settings.HF_MODEL_ID or os.path.exists(settings.MODEL_PATH):
-        llm_text = _llm_analyze(solidity_code)
-
+    llm_text = _call_hf_inference_api(solidity_code)
     return {
         **pattern_result,
         "llm_analysis": llm_text,
@@ -128,7 +98,8 @@ def analyze_contract(solidity_code: str) -> Dict[str, Any]:
 
 def model_status() -> Dict[str, Any]:
     return {
-        "model_loaded": _model_loaded,
-        "model_source": settings.HF_MODEL_ID or settings.MODEL_PATH,
-        "device": "cuda" if torch.cuda.is_available() else "cpu",
-    }
+        "model_loaded": False,
+        "inference_mode": "HF Inference API",
+        "model_id": HF_MODEL_ID or "not configured",
+        "hf_token_set": bool(HF_TOKEN),
+    }