Updated interface after checkpoint

inference.py

@@ -1,20 +1,9 @@
 # inference.py
-# Wraps enterprise_inference from clauseXplain v5.0
-# Requires: model, tokenizer, clause_mlb, risk_mlb, feature_extractor
-# to be loaded externally before calling analyze_clause().
+# Pure utility functions for neuro-symbolic fusion.
+# No module-level mutable globals — all state lives in ModelManager (app.py).
 
 from __future__ import annotations
-import torch
 
-# ── These globals must be set by app.py before calling analyze_clause ──
-model = None
-tokenizer = None
-clause_mlb = None
-risk_mlb = None
-feature_extractor = None
-device = None
-
-RISK_LEVEL_ORDER = {"Low": 0, "Medium": 1, "High": 2}
 IP_CLAUSE_TYPES = {
     "IP Ownership Assignment", "Joint IP Ownership",
     "Irrevocable Or Perpetual License",
@@ -22,9 +11,18 @@ IP_CLAUSE_TYPES = {
 }
 
 
-def _symbolic_rule_score(features: dict, SYMBOLIC_RULES: list) -> dict:
+def _symbolic_rule_score(features: dict, symbolic_rules: list) -> dict:
+    """
+    Evaluate symbolic rules against extracted features.
+
+    Returns:
+        {
+            "symbolic_score": float,        # clamped to [0, 1]
+            "triggered_rules": list[dict],  # rules whose condition fired
+        }
+    """
     triggered, total = [], 0.0
-    for rule in SYMBOLIC_RULES:
+    for rule in symbolic_rules:
         try:
             if rule["condition"](features):
                 triggered.append(rule)
@@ -37,137 +35,32 @@ def _symbolic_rule_score(features: dict, SYMBOLIC_RULES: list) -> dict:
     }
 
 
-def _neuro_symbolic_fusion(neural: float, symbolic: float,
-                            is_ip_clause: bool = False) -> dict:
+def _neuro_symbolic_fusion(
+    neural: float,
+    symbolic: float,
+    is_ip_clause: bool = False,
+) -> dict:
+    """
+    Weighted fusion of neural and symbolic scores.
+
+    IP clauses shift weight toward symbolic rules (which capture IP-specific law).
+    Ensures score is non-trivially low when symbolic rules fire.
+
+    Returns:
+        { "score": float, "level": str, "emoji": str }
+    """
     if is_ip_clause and symbolic > 0:
         w_n, w_s = 0.35, 0.65
     else:
         w_n, w_s = 0.60, 0.40
+
     score = w_n * neural + w_s * symbolic
     if symbolic > 0:
-        score = max(score, 0.30)
+        score = max(score, 0.30)          # symbolic trigger → at least Medium
     score = round(min(score, 1.0), 3)
+
     if   score <= 0.33: level, emoji = "Low",    "🟢"
     elif score <= 0.66: level, emoji = "Medium", "🟡"
     else:               level, emoji = "High",   "🔴"
-    return {"score": score, "level": level, "emoji": emoji}
-
-
-def analyze_clause(text: str, SYMBOLIC_RULES: list) -> dict:
-    """
-    Run full neuro-symbolic inference on a single clause text.
-    Returns a clean dict with risk_score, risk_level, triggered_rules, etc.
-    Requires module-level globals to be set (model, tokenizer, ...).
-    """
-    if model is None:
-        raise RuntimeError("Model not loaded. Call load_model() in app.py first.")
-
-    enc = tokenizer(
-        text, padding="max_length", truncation=True,
-        max_length=512, return_tensors="pt"
-    )
-    model.eval()
-    with torch.no_grad():
-        clause_logits, risk_logits, risk_score_tensor, _, _ = model(
-            enc["input_ids"].to(device),
-            enc["attention_mask"].to(device),
-        )
-
-    clause_probs = torch.sigmoid(clause_logits).cpu().numpy()[0]
-    top3_idx     = clause_probs.argsort()[::-1][:3]
-    top_clauses  = [
-        (clause_mlb.classes_[i], round(float(clause_probs[i]), 3))
-        for i in top3_idx if clause_probs[i] > 0.05
-    ]
-
-    risk_probs = torch.sigmoid(risk_logits).cpu().numpy()[0]
-    top2_idx   = risk_probs.argsort()[::-1][:2]
-    top_risks  = [
-        (risk_mlb.classes_[i], round(float(risk_probs[i]), 3))
-        for i in top2_idx if risk_probs[i] > 0.05
-    ]
-
-    neural_score = round(float(risk_score_tensor.item()), 3)
-    features     = feature_extractor.extract(text)
-    sym_result   = _symbolic_rule_score(features, SYMBOLIC_RULES)
-
-    top_clause_name = top_clauses[0][0] if top_clauses else ""
-    is_ip  = top_clause_name in IP_CLAUSE_TYPES
-    fusion = _neuro_symbolic_fusion(neural_score, sym_result["symbolic_score"], is_ip)
 
-    triggered_clean = [
-        {
-            "rule_id":   r["rule_id"],
-            "name":      r["name"],
-            "reference": r["reference"],
-            "penalty":   r["penalty"],
-            "category":  r["category"],
-        }
-        for r in sym_result["triggered_rules"]
-    ]
-
-    return {
-        "risk_score":      fusion["score"],
-        "neural_score":    neural_score,
-        "symbolic_score":  sym_result["symbolic_score"],
-        "risk_level":      f"{fusion['emoji']} {fusion['level']}",
-        "risk_level_raw":  fusion["level"],
-        "top_clauses":     top_clauses,
-        "top_risk_cats":   top_risks,
-        "triggered_rules": triggered_clean,
-        "features":        {k: v for k, v in features.items() if v},
-    }
-
-
-# ── Document-level analysis (added for dashboard) ���───────────────────────────
-def analyze_document(text: str, SYMBOLIC_RULES: list, max_clauses: int = 50) -> dict:
-    """
-    Split text into clauses, run analyze_clause() on each, return document summary.
-
-    Returns:
-        {
-            "overall_risk":  float,          # weighted-max of fused scores
-            "overall_level": str,            # Low / Medium / High
-            "num_clauses":   int,
-            "top_risks":     list[dict],     # top 3 by risk_score
-            "clauses":       list[dict],     # all clause results + index + text
-        }
-    """
-    from pdf_utils import split_into_clauses
-
-    clauses = split_into_clauses(text)[:max_clauses]
-    if not clauses:
-        clauses = [text[:2000]]  # fallback: treat whole text as one clause
-
-    results = []
-    for idx, clause_text in enumerate(clauses):
-        try:
-            r = analyze_clause(clause_text, SYMBOLIC_RULES)
-        except Exception:
-            r = {
-                "risk_score": 0.0, "neural_score": 0.0, "symbolic_score": 0.0,
-                "risk_level": "🟢 Low", "risk_level_raw": "Low",
-                "top_clauses": [], "top_risk_cats": [],
-                "triggered_rules": [], "features": {},
-            }
-        r["clause_index"] = idx + 1
-        r["clause_text"]  = clause_text
-        results.append(r)
-
-    scores = [r["risk_score"] for r in results]
-
-    # Overall = 70% max + 30% mean (punishes worst clause, not just average)
-    overall = round(0.70 * max(scores) + 0.30 * (sum(scores) / len(scores)), 3)
-    if   overall <= 0.33: level = "Low"
-    elif overall <= 0.66: level = "Medium"
-    else:                 level = "High"
-
-    top_risks = sorted(results, key=lambda x: x["risk_score"], reverse=True)[:3]
-
-    return {
-        "overall_risk":  overall,
-        "overall_level": level,
-        "num_clauses":   len(results),
-        "top_risks":     top_risks,
-        "clauses":       results,
-    }
+    return {"score": score, "level": level, "emoji": emoji}