skatzR/RQA-R2
013
1# requirements2# Для inference в Colab достаточно этого стека.3!pip install transformers==4.48.3 tokenizers sentencepiece accelerate4 5 6# ============================================================7# RQA UX Inference — R2 Interactive Version8# Google Colab + CLI friendly9# ============================================================10 11import os12import json13import csv14import torch15from typing import List, Optional16from transformers import AutoTokenizer, AutoModel17 18 19# ============================================================20# Константы21# ============================================================22 23ERROR_TYPES = [24 "false_causality",25 "unsupported_claim",26 "overgeneralization",27 "missing_premise",28 "contradiction",29 "circular_reasoning",30]31 32ERROR_NAMES_RU = {33 "false_causality": "Ложная причинно-следственная связь",34 "unsupported_claim": "Неподкрепленное утверждение",35 "overgeneralization": "Чрезмерное обобщение",36 "missing_premise": "Отсутствующая предпосылка",37 "contradiction": "Противоречие",38 "circular_reasoning": "Круговое рассуждение",39}40 41 42# ============================================================43# RQA Judge44# ============================================================45 46class RQAJudge:47 def __init__(self, model_name="skatzR/RQA-R2", device=None, max_length: int = 512):48 self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")49 self.max_length = int(max_length)50 51 self.tokenizer = AutoTokenizer.from_pretrained(52 model_name,53 trust_remote_code=True54 )55 self.model = AutoModel.from_pretrained(56 model_name,57 trust_remote_code=True58 ).to(self.device)59 60 self.model.eval()61 62 cfg = self.model.config63 self.error_types = list(getattr(cfg, "error_types", ERROR_TYPES))64 65 self.temp_issue = float(getattr(cfg, "temperature_has_issue", 1.0))66 self.temp_hidden = float(getattr(cfg, "temperature_is_hidden", 1.0))67 self.temp_errors = list(68 getattr(cfg, "temperature_errors", [1.0] * len(self.error_types))69 )70 71 self.threshold_issue = float(getattr(cfg, "threshold_has_issue", 0.5))72 self.threshold_hidden = float(getattr(cfg, "threshold_is_hidden", 0.5))73 self.threshold_error = float(getattr(cfg, "threshold_error", 0.5))74 self.threshold_errors = list(75 getattr(cfg, "threshold_errors", [self.threshold_error] * len(self.error_types))76 )77 78 # ----------------------79 # Core inference80 # ----------------------81 82 @torch.no_grad()83 def infer(84 self,85 text: str,86 issue_threshold: Optional[float] = None,87 hidden_threshold: Optional[float] = None,88 error_threshold: Optional[float] = None,89 error_thresholds: Optional[List[float]] = None,90 issue_uncertain_margin: float = 0.05,91 hidden_uncertain_margin: float = 0.05,92 error_uncertain_margin: float = 0.05,93 ):94 issue_threshold = self.threshold_issue if issue_threshold is None else float(issue_threshold)95 hidden_threshold = self.threshold_hidden if hidden_threshold is None else float(hidden_threshold)96 error_threshold = self.threshold_error if error_threshold is None else float(error_threshold)97 error_thresholds = self.threshold_errors if error_thresholds is None else list(error_thresholds)98 99 inputs = self.tokenizer(100 text,101 truncation=True,102 max_length=self.max_length,103 padding="max_length",104 return_tensors="pt"105 ).to(self.device)106 107 outputs = self.model(**inputs)108 109 # ----- has_issue -----110 issue_logit = outputs["has_issue_logits"] / self.temp_issue111 issue_prob = torch.sigmoid(issue_logit).item()112 has_issue = issue_prob >= issue_threshold113 114 result = {115 "text": text,116 "class": None, # logical / hidden / explicit117 "status": "ok", # ok / uncertain118 "review_required": False,119 "has_issue": has_issue,120 "issue_probability": issue_prob,121 "hidden_problem": False,122 "hidden_probability": None,123 "errors": [],124 "num_errors": 0,125 "schema_version": getattr(self.model.config, "schema_version", "unknown"),126 "threshold_issue": issue_threshold,127 "threshold_hidden": hidden_threshold,128 "threshold_error": error_threshold,129 "threshold_errors": error_thresholds,130 }131 132 if abs(issue_prob - issue_threshold) <= issue_uncertain_margin:133 result["status"] = "uncertain"134 result["review_required"] = True135 136 # ----- Gate 1: logical -----137 if not has_issue:138 result["class"] = "logical"139 return result140 141 # ----- hidden -----142 hidden_logit = outputs["is_hidden_logits"] / self.temp_hidden143 hidden_prob = torch.sigmoid(hidden_logit).item()144 is_hidden = hidden_prob >= hidden_threshold145 146 result["hidden_problem"] = is_hidden147 result["hidden_probability"] = hidden_prob148 149 if abs(hidden_prob - hidden_threshold) <= hidden_uncertain_margin:150 result["status"] = "uncertain"151 result["review_required"] = True152 153 # ----- Gate 2: hidden -----154 if is_hidden:155 result["class"] = "hidden"156 return result157 158 # ----- explicit errors -----159 raw_error_logits = outputs["errors_logits"][0].clone()160 error_probs = {}161 162 for i, logit in enumerate(raw_error_logits):163 calibrated = logit / self.temp_errors[i]164 prob = torch.sigmoid(calibrated).item()165 error_probs[self.error_types[i]] = prob166 167 explicit_errors = []168 for i, err_name in enumerate(self.error_types):169 prob = float(error_probs[err_name])170 threshold_i = float(error_thresholds[i] if i < len(error_thresholds) else error_threshold)171 172 if abs(prob - threshold_i) <= error_uncertain_margin:173 result["status"] = "uncertain"174 result["review_required"] = True175 176 if prob >= threshold_i:177 explicit_errors.append((err_name, prob))178 179 explicit_errors.sort(key=lambda x: x[1], reverse=True)180 181 result["class"] = "explicit"182 result["errors"] = explicit_errors183 result["num_errors"] = len(explicit_errors)184 return result185 186 # ============================================================187 # UX output188 # ============================================================189 190 def pretty_print(self, r):191 print("\n" + "=" * 72)192 print("📄 Текст:")193 print(r["text"])194 195 print(196 f"\n🔎 Обнаружена проблема: {'ДА' if r['has_issue'] else 'НЕТ'} "197 f"({r['issue_probability'] * 100:.2f}%)"198 )199 print(f"🧠 Класс: {r['class']}")200 201 if r["status"] == "uncertain":202 print("⚠️ Пограничный случай: review recommended")203 204 if r["hidden_probability"] is not None:205 print(206 f"🟡 Hidden-проблема: {'ДА' if r['hidden_problem'] else 'НЕТ'} "207 f"({r['hidden_probability'] * 100:.2f}%)"208 )209 210 if r["errors"]:211 print("\n❌ Явные логические ошибки:")212 for name, prob in r["errors"]:213 print(f" • {ERROR_NAMES_RU.get(name, name)} — {prob * 100:.2f}%")214 else:215 print("\n✅ Явных логических ошибок не обнаружено")216 217 print("=" * 72)218 219 220# ============================================================221# Loaders222# ============================================================223 224def load_texts_from_file(path: str) -> List[str]:225 ext = os.path.splitext(path)[1].lower()226 227 if ext == ".txt":228 with open(path, encoding="utf-8") as f:229 return [line.strip() for line in f if line.strip()]230 231 if ext == ".csv":232 with open(path, encoding="utf-8") as f:233 reader = csv.DictReader(f)234 return [row["text"] for row in reader if row.get("text")]235 236 if ext == ".json":237 with open(path, encoding="utf-8") as f:238 data = json.load(f)239 if isinstance(data, list):240 if all(isinstance(item, str) for item in data):241 return data242 texts = []243 for item in data:244 if isinstance(item, dict) and "text" in item:245 texts.append(str(item["text"]))246 return texts247 248 raise ValueError("Неподдерживаемый формат файла")249 250 251# ============================================================252# Interactive CLI Interface253# ============================================================254 255class InteractiveCLI:256 def __init__(self, model_name="skatzR/RQA-R2"):257 self.judge = RQAJudge(model_name=model_name)258 259 def clear_screen(self):260 print("\n" * 2)261 262 def show_mode_menu(self):263 self.clear_screen()264 print("=" * 60)265 print("🤖 RQA-R2 — АНАЛИЗ ЛОГИЧЕСКИХ ОШИБОК")266 print("=" * 60)267 print("\nВыберите режим работы:")268 print("1. 📝 Одиночный ввод (одна фраза для анализа)")269 print("2. 📄 Множественный ввод (несколько фраз, каждая с новой строки)")270 print("3. 📂 Загрузка из файла (.txt, .csv, .json)")271 print("\nНажмите Enter без ввода для выхода.")272 print("-" * 60)273 274 def process_single_mode(self):275 self.clear_screen()276 print("[📝 РЕЖИМ: ОДИНОЧНЫЙ ВВОД]")277 print("Введите текст для анализа:")278 print("(Нажмите Enter без ввода для возврата в меню)")279 print("-" * 40)280 281 text = input("> ").strip()282 if not text:283 return True284 285 result = self.judge.infer(text)286 self.judge.pretty_print(result)287 288 print("\n" + "-" * 40)289 input("Нажмите Enter для продолжения...")290 return False291 292 def process_multiline_mode(self):293 self.clear_screen()294 print("[📄 РЕЖИМ: МНОЖЕСТВЕННЫЙ ВВОД]")295 print("Введите тексты для анализа (каждый с новой строки).")296 print("Оставьте строку пустой для завершения ввода.")297 print("(Нажмите Enter без ввода для возврата в меню)")298 print("-" * 40)299 300 texts = []301 print("Ввод текстов:")302 while True:303 line = input("> ").strip()304 if not line:305 if not texts:306 return True307 break308 texts.append(line)309 310 self.clear_screen()311 print(f"[📄 РЕЖИМ: МНОЖЕСТВЕННЫЙ ВВОД] — найдено {len(texts)} текстов")312 print("-" * 40)313 314 for i, text in enumerate(texts, 1):315 print(f"\n🔍 Текст #{i}:")316 result = self.judge.infer(text)317 self.judge.pretty_print(result)318 319 print("\n" + "=" * 60)320 input("Нажмите Enter для продолжения...")321 return False322 323 def process_file_mode(self):324 self.clear_screen()325 print("[📂 РЕЖИМ: ЗАГРУЗКА ИЗ ФАЙЛА]")326 print("Поддерживаемые форматы: .txt, .csv, .json")327 print("Укажите путь к файлу:")328 print("(Нажмите Enter без ввода для возврата в меню)")329 print("-" * 40)330 331 file_path = input("Путь к файлу> ").strip()332 if not file_path:333 return True334 335 try:336 if not os.path.exists(file_path):337 print(f"\n❌ Ошибка: Файл '{file_path}' не найден!")338 input("\nНажмите Enter для продолжения...")339 return False340 341 texts = load_texts_from_file(file_path)342 if not texts:343 print(f"\n⚠️ Файл '{file_path}' пуст или не содержит текстов!")344 input("\nНажмите Enter для продолжения...")345 return False346 347 self.clear_screen()348 print(f"[📂 РЕЖИМ: ЗАГРУЗКА ИЗ ФАЙЛА] — загружено {len(texts)} текстов")349 print(f"Файл: {file_path}")350 print("-" * 40)351 352 for i, text in enumerate(texts, 1):353 print(f"\n🔍 Текст #{i}:")354 result = self.judge.infer(text)355 self.judge.pretty_print(result)356 357 print("\n" + "=" * 60)358 input("Нажмите Enter для продолжения...")359 360 except Exception as e:361 print(f"\n❌ Ошибка при обработке файла: {str(e)}")362 input("\nНажмите Enter для продолжения...")363 364 return False365 366 def run_interactive(self):367 current_mode = None368 369 while True:370 if not current_mode:371 self.show_mode_menu()372 choice = input("Ваш выбор (1-3)> ").strip()373 374 if not choice:375 print("\n👋 Выход из программы...")376 break377 378 if choice == "1":379 current_mode = "single"380 elif choice == "2":381 current_mode = "multiline"382 elif choice == "3":383 current_mode = "file"384 else:385 print("\n❌ Неверный выбор! Попробуйте снова.")386 input("Нажмите Enter для продолжения...")387 continue388 389 should_return_to_menu = False390 391 if current_mode == "single":392 should_return_to_menu = self.process_single_mode()393 elif current_mode == "multiline":394 should_return_to_menu = self.process_multiline_mode()395 elif current_mode == "file":396 should_return_to_menu = self.process_file_mode()397 398 if should_return_to_menu:399 current_mode = None400 401 402# ============================================================403# Точка входа404# ============================================================405 406def main():407 cli = InteractiveCLI()408 cli.run_interactive()409 410 411# ============================================================412# Запуск413# ============================================================414 415if __name__ == "__main__":416 main()417