Create inference.py

inference.py

@@ -0,0 +1,264 @@
+# ============================================================
+# RQA UX Inference — FINAL
+# Google Colab + CLI friendly
+# ============================================================
+
+import os
+import sys
+import json
+import argparse
+import csv
+import torch
+from typing import List, Union
+from transformers import AutoTokenizer, AutoModel
+
+# ============================================================
+# Константы
+# ============================================================
+
+ERROR_TYPES = [
+    "false_causality",
+    "unsupported_claim",
+    "overgeneralization",
+    "missing_premise",
+    "contradiction",
+    "circular_reasoning",
+]
+
+ERROR_NAMES_RU = {
+    "false_causality": "Ложная причинно-следственная связь",
+    "unsupported_claim": "Неподкреплённое утверждение",
+    "overgeneralization": "Чрезмерное обобщение",
+    "missing_premise": "Отсутствующая предпосылка",
+    "contradiction": "Противоречие",
+    "circular_reasoning": "Круговое рассуждение",
+}
+
+ERROR_THRESHOLDS = {
+    "false_causality": 0.55,
+    "unsupported_claim": 0.55,
+    "overgeneralization": 0.60,
+    "missing_premise": 0.80,   # диагностический
+    "contradiction": 0.60,
+    "circular_reasoning": 0.60,
+}
+
+# ============================================================
+# RQA Judge
+# ============================================================
+
+class RQAJudge:
+    def __init__(self, model_name="skatzR/RQA-X1.1", device=None):
+        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
+
+        self.tokenizer = AutoTokenizer.from_pretrained(
+            model_name,
+            trust_remote_code=True
+        )
+        self.model = AutoModel.from_pretrained(
+            model_name,
+            trust_remote_code=True
+        ).to(self.device)
+
+        self.model.eval()
+
+        cfg = self.model.config
+        self.temp_issue = float(cfg.temperature_has_issue)
+        self.temp_errors = list(cfg.temperature_errors)
+
+    # ----------------------
+    # Core inference
+    # ----------------------
+
+    @torch.no_grad()
+    def infer(
+        self,
+        text: str,
+        issue_threshold: float = 0.6,
+        disagreement_threshold: float = 0.4,
+    ):
+        inputs = self.tokenizer(
+            text,
+            truncation=True,
+            max_length=512,
+            padding="max_length",
+            return_tensors="pt"
+        ).to(self.device)
+
+        outputs = self.model(**inputs)
+
+        # ----- has_issue -----
+        issue_logit = outputs["has_issue_logits"] / self.temp_issue
+        issue_prob = torch.sigmoid(issue_logit).item()
+        has_issue = issue_prob >= issue_threshold
+
+        # ----- errors -----
+        raw_error_logits = outputs["errors_logits"][0]
+        error_probs = {}
+
+        for i, logit in enumerate(raw_error_logits):
+            calibrated = logit / self.temp_errors[i]
+            prob = torch.sigmoid(calibrated).item()
+            error_probs[ERROR_TYPES[i]] = prob
+
+        # ----- disagreement -----
+        p_any_error = 1.0
+        for p in error_probs.values():
+            p_any_error *= (1.0 - p)
+        p_any_error = 1.0 - p_any_error
+
+        disagreement = abs(issue_prob - p_any_error)
+
+        # ----- decision logic -----
+        explicit_errors = []
+        hidden_problem = False
+
+        for err, prob in error_probs.items():
+            if prob >= ERROR_THRESHOLDS[err]:
+                if err == "missing_premise":
+                    hidden_problem = True
+                else:
+                    explicit_errors.append((err, prob))
+
+        explicit_errors.sort(key=lambda x: x[1], reverse=True)
+
+        # бинарная голова доминирует
+        if not has_issue:
+            explicit_errors = []
+
+        borderline = (
+            not has_issue and hidden_problem and disagreement >= disagreement_threshold
+        )
+
+        return {
+            "text": text,
+            "has_issue": has_issue,
+            "issue_probability": issue_prob,
+            "errors": explicit_errors,
+            "hidden_problem": hidden_problem,
+            "borderline": borderline,
+            "disagreement": disagreement,
+        }
+
+    # ============================================================
+    # UX output
+    # ============================================================
+
+    def pretty_print(self, r):
+        print("\n" + "=" * 72)
+        print("📄 Текст:")
+        print(r["text"])
+
+        print(f"\n🔎 Обнаружена проблема: {'ДА' if r['has_issue'] else 'НЕТ'} "
+              f"({r['issue_probability']*100:.2f}%)")
+
+        if r["borderline"]:
+            print("⚠️  Пограничный случай: аргументативный текст")
+
+        if r["hidden_problem"]:
+            print("🟡 Скрытая проблема: возможны неявные предпосылки")
+
+        if r["errors"]:
+            print("\n❌ Явные логические ошибки:")
+            for name, prob in r["errors"]:
+                print(f"  • {ERROR_NAMES_RU[name]} — {prob*100:.2f}%")
+        else:
+            print("\n✅ Явных логических ошибок не обнаружено")
+
+        print(f"\n📊 Disagreement: {r['disagreement']:.3f}")
+        print("=" * 72)
+
+# ============================================================
+# Loaders
+# ============================================================
+
+def load_texts_from_file(path: str) -> List[str]:
+    ext = os.path.splitext(path)[1].lower()
+
+    if ext == ".txt":
+        with open(path, encoding="utf-8") as f:
+            return [l.strip() for l in f if l.strip()]
+
+    if ext == ".csv":
+        with open(path, encoding="utf-8") as f:
+            reader = csv.DictReader(f)
+            return [row["text"] for row in reader]
+
+    if ext == ".json":
+        with open(path, encoding="utf-8") as f:
+            data = json.load(f)
+        if isinstance(data, list):
+            return data
+
+    raise ValueError("Неподдерживаемый формат файла")
+
+# ============================================================
+# CLI / Colab entrypoint
+# ============================================================
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="RQA — анализ логических ошибок"
+    )
+
+    parser.add_argument(
+        "--text",
+        type=str,
+        help="Один текст для анализа"
+    )
+
+    parser.add_argument(
+        "--file",
+        type=str,
+        help="Файл с текстами (.txt, .csv, .json)"
+    )
+
+    parser.add_argument(
+        "--multiline",
+        action="store_true",
+        help="Ввод нескольких строк (каждая строка — отдельный текст)"
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    judge = RQAJudge()
+
+    texts = []
+
+    # ---------- FILE MODE ----------
+    if args.file:
+        if not os.path.exists(args.file):
+            raise FileNotFoundError(args.file)
+        texts = load_texts_from_file(args.file)
+
+    # ---------- SINGLE TEXT ----------
+    elif args.text:
+        texts = [args.text]
+
+    # ---------- MULTILINE ----------
+    elif args.multiline:
+        print("Введите тексты (пустая строка — конец ввода):")
+        while True:
+            line = input()
+            if not line.strip():
+                break
+            texts.append(line.strip())
+
+    # ---------- INTERACTIVE FALLBACK ----------
+    else:
+        print("Введите текст для анализа:")
+        line = input().strip()
+        if line:
+            texts = [line]
+        else:
+            print("❌ Пустой ввод — выхожу")
+            return
+
+    # ---------- RUN ----------
+    for t in texts:
+        result = judge.infer(t)
+        judge.pretty_print(result)
+
+
+if __name__ == "__main__":
+    main()