Upload 3 files

__init__.py

@@ -0,0 +1,15 @@
+from transformers import AutoConfig, AutoModel
+
+from .modeling_rqa import RQAModelConfig, RQAModelHF
+
+__all__ = ["RQAModelConfig", "RQAModelHF"]
+
+try:
+    AutoConfig.register("rqa_v2_2", RQAModelConfig)
+except ValueError:
+    pass
+
+try:
+    AutoModel.register(RQAModelConfig, RQAModelHF)
+except ValueError:
+    pass

inference.py

@@ -0,0 +1,356 @@
+import os
+from typing import Any, Dict, List, Optional
+
+import torch
+from transformers import AutoTokenizer
+
+try:
+    from huggingface_hub import hf_hub_download
+except Exception:
+    hf_hub_download = None
+
+try:
+    from .modeling_rqa import RQAModelHF
+except ImportError:
+    from modeling_rqa import RQAModelHF
+
+
+ERROR_NAMES_RU = {
+    "false_causality": "Ложная причинно-следственная связь",
+    "unsupported_claim": "Неподкрепленное утверждение",
+    "overgeneralization": "Чрезмерное обобщение",
+    "missing_premise": "Отсутствующая предпосылка",
+    "contradiction": "Противоречие",
+    "circular_reasoning": "Круговое рассуждение",
+}
+
+
+def _resolve_calibration_path(model_path: str) -> Optional[str]:
+    local_path = os.path.join(model_path, "calibration_data.pth")
+    if os.path.exists(local_path):
+        return local_path
+
+    if hf_hub_download is None or os.path.isdir(model_path):
+        return None
+
+    try:
+        return hf_hub_download(
+            repo_id=model_path,
+            filename="calibration_data.pth",
+        )
+    except Exception:
+        return None
+
+
+class RQAInferenceHF:
+    def __init__(
+        self,
+        model_path: str,
+        device: Optional[torch.device] = None,
+        max_length: int = 512,
+        issue_uncertain_margin: float = 0.05,
+        hidden_uncertain_margin: float = 0.05,
+        error_uncertain_margin: float = 0.05,
+    ):
+        self.model_path = model_path
+        self.device = device or torch.device(
+            "cuda" if torch.cuda.is_available() else "cpu"
+        )
+        self.max_length = int(max_length)
+        self.issue_uncertain_margin = float(issue_uncertain_margin)
+        self.hidden_uncertain_margin = float(hidden_uncertain_margin)
+        self.error_uncertain_margin = float(error_uncertain_margin)
+
+        self.model = RQAModelHF.from_pretrained(model_path).to(self.device).eval()
+        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
+
+        cfg = self.model.config
+        self.schema_version = str(getattr(cfg, "schema_version", "unknown"))
+        self.error_types = list(getattr(cfg, "error_types", []))
+        self.t_issue = float(getattr(cfg, "temperature_has_issue", 1.0))
+        self.t_hidden = float(getattr(cfg, "temperature_is_hidden", 1.0))
+        self.t_errors = list(
+            getattr(cfg, "temperature_errors", [1.0] * len(self.error_types))
+        )
+        self.th_issue = float(getattr(cfg, "threshold_has_issue", 0.5))
+        self.th_hidden = float(getattr(cfg, "threshold_is_hidden", 0.5))
+        self.th_error = float(getattr(cfg, "threshold_error", 0.5))
+        self.th_errors = list(
+            getattr(cfg, "threshold_errors", [self.th_error] * len(self.error_types))
+        )
+
+        calibration_path = _resolve_calibration_path(model_path)
+        if calibration_path:
+            calibration = torch.load(calibration_path, map_location="cpu")
+            calibration_error_types = calibration.get("error_types", None)
+            if calibration_error_types is not None:
+                if list(calibration_error_types) != self.error_types:
+                    raise ValueError(
+                        "Calibration artifact error_types mismatch with model.config.error_types."
+                    )
+
+            self.schema_version = str(
+                calibration.get("schema_version", self.schema_version)
+            )
+            self.t_issue = float(
+                calibration.get("temperature_has_issue", self.t_issue)
+            )
+            self.t_hidden = float(
+                calibration.get("temperature_is_hidden", self.t_hidden)
+            )
+            self.t_errors = list(
+                calibration.get("temperature_errors", self.t_errors)
+            )
+            self.th_issue = float(
+                calibration.get("threshold_has_issue", self.th_issue)
+            )
+            self.th_hidden = float(
+                calibration.get("threshold_is_hidden", self.th_hidden)
+            )
+            self.th_error = float(
+                calibration.get("threshold_error", self.th_error)
+            )
+            self.th_errors = list(
+                calibration.get("threshold_errors", self.th_errors)
+            )
+
+    def _apply_temperature(
+        self,
+        issue_logits: torch.Tensor,
+        hidden_logits: torch.Tensor,
+        errors_logits: torch.Tensor,
+    ):
+        calibrated_issue = issue_logits / float(self.t_issue)
+        calibrated_hidden = hidden_logits / float(self.t_hidden)
+        calibrated_errors = errors_logits.clone()
+        for idx in range(calibrated_errors.size(1)):
+            temperature = float(self.t_errors[idx]) if idx < len(self.t_errors) else 1.0
+            calibrated_errors[:, idx] = calibrated_errors[:, idx] / temperature
+        return calibrated_issue, calibrated_hidden, calibrated_errors
+
+    @torch.no_grad()
+    def predict(
+        self,
+        text: str,
+        return_probs: bool = False,
+        threshold_issue: Optional[float] = None,
+        threshold_hidden: Optional[float] = None,
+        threshold_error: Optional[float] = None,
+        threshold_errors: Optional[List[float]] = None,
+    ) -> Dict[str, Any]:
+        issue_threshold = self.th_issue if threshold_issue is None else float(threshold_issue)
+        hidden_threshold = self.th_hidden if threshold_hidden is None else float(threshold_hidden)
+        error_threshold = self.th_error if threshold_error is None else float(threshold_error)
+        error_thresholds = self.th_errors if threshold_errors is None else list(threshold_errors)
+
+        encoded = self.tokenizer(
+            text,
+            truncation=True,
+            max_length=self.max_length,
+            padding="max_length",
+            return_tensors="pt",
+        )
+        input_ids = encoded["input_ids"].to(self.device)
+        attention_mask = encoded["attention_mask"].to(self.device)
+
+        outputs = self.model(input_ids=input_ids, attention_mask=attention_mask)
+        issue_logits, hidden_logits, errors_logits = self._apply_temperature(
+            outputs["has_issue_logits"],
+            outputs["is_hidden_logits"],
+            outputs["errors_logits"],
+        )
+
+        issue_probability = float(torch.sigmoid(issue_logits).item())
+        has_issue = issue_probability >= issue_threshold
+
+        result: Dict[str, Any] = {
+            "schema_version": self.schema_version,
+            "text": text,
+            "class": None,
+            "status": "ok",
+            "review_required": False,
+            "has_logical_issue": bool(has_issue),
+            "has_issue_probability": issue_probability,
+            "threshold_has_issue": issue_threshold,
+            "temperature_has_issue": float(self.t_issue),
+            "is_hidden_problem": False,
+            "hidden_probability": None,
+            "threshold_is_hidden": hidden_threshold,
+            "temperature_is_hidden": float(self.t_hidden),
+            "errors": [],
+            "num_errors": 0,
+            "threshold_error": error_threshold,
+            "threshold_errors": error_thresholds,
+            "calibrated": (
+                abs(self.t_issue - 1.0) > 1e-6
+                or abs(self.t_hidden - 1.0) > 1e-6
+                or any(abs(float(t) - 1.0) > 1e-6 for t in self.t_errors)
+            ),
+        }
+
+        if abs(issue_probability - issue_threshold) <= self.issue_uncertain_margin:
+            result["status"] = "uncertain"
+            result["review_required"] = True
+
+        if not has_issue:
+            result["class"] = "logical"
+            if return_probs:
+                result["raw"] = {"p_issue": issue_probability}
+            return result
+
+        hidden_probability = float(torch.sigmoid(hidden_logits).item())
+        is_hidden = hidden_probability >= hidden_threshold
+        result["hidden_probability"] = hidden_probability
+        result["is_hidden_problem"] = bool(is_hidden)
+
+        if abs(hidden_probability - hidden_threshold) <= self.hidden_uncertain_margin:
+            result["status"] = "uncertain"
+            result["review_required"] = True
+
+        if is_hidden:
+            result["class"] = "hidden"
+            if return_probs:
+                result["raw"] = {
+                    "p_issue": issue_probability,
+                    "p_hidden": hidden_probability,
+                }
+            return result
+
+        error_probabilities = torch.sigmoid(errors_logits).cpu().numpy()[0]
+        detected_errors = []
+        for idx, error_type in enumerate(self.error_types):
+            probability = float(error_probabilities[idx])
+            threshold_i = float(
+                error_thresholds[idx] if idx < len(error_thresholds) else error_threshold
+            )
+            if abs(probability - threshold_i) <= self.error_uncertain_margin:
+                result["status"] = "uncertain"
+                result["review_required"] = True
+            if probability >= threshold_i:
+                detected_errors.append(
+                    {
+                        "type": error_type,
+                        "probability": probability,
+                        "threshold": threshold_i,
+                        "temperature": float(self.t_errors[idx]) if idx < len(self.t_errors) else 1.0,
+                    }
+                )
+
+        detected_errors.sort(key=lambda item: item["probability"], reverse=True)
+        result["class"] = "explicit"
+        result["errors"] = detected_errors
+        result["num_errors"] = len(detected_errors)
+
+        if return_probs:
+            result["error_probabilities"] = {
+                error_type: float(probability)
+                for error_type, probability in zip(self.error_types, error_probabilities)
+            }
+            result["raw"] = {
+                "p_issue": issue_probability,
+                "p_hidden": hidden_probability,
+            }
+
+        return result
+
+    def pretty_print(self, prediction: Dict[str, Any], use_russian_names: bool = True) -> None:
+        print("-" * 70)
+        print(
+            f"Class: {prediction['class']} | status={prediction['status']} "
+            f"| review_required={prediction['review_required']}"
+        )
+        print(
+            f"Issue: {prediction['has_logical_issue']} "
+            f"({prediction['has_issue_probability'] * 100:.2f}%)  "
+            f"th={prediction['threshold_has_issue']:.3f}"
+        )
+        if prediction["hidden_probability"] is not None:
+            print(
+                f"Hidden: {prediction['is_hidden_problem']} "
+                f"({prediction['hidden_probability'] * 100:.2f}%)  "
+                f"th={prediction['threshold_is_hidden']:.3f}"
+            )
+
+        if prediction["errors"]:
+            printable_errors = []
+            for item in prediction["errors"]:
+                label = (
+                    ERROR_NAMES_RU.get(item["type"], item["type"])
+                    if use_russian_names
+                    else item["type"]
+                )
+                printable_errors.append((label, round(item["probability"], 3)))
+            print(f"Top errors: {printable_errors}")
+
+
+class RQAJudge:
+    def __init__(
+        self,
+        model_name: str = "skatzR/RQA-R2",
+        device: Optional[torch.device] = None,
+        max_length: int = 512,
+    ):
+        self.runner = RQAInferenceHF(
+            model_path=model_name,
+            device=device,
+            max_length=max_length,
+        )
+
+    def infer(
+        self,
+        text: str,
+        issue_threshold: Optional[float] = None,
+        hidden_threshold: Optional[float] = None,
+        error_threshold: Optional[float] = None,
+        error_thresholds: Optional[List[float]] = None,
+    ) -> Dict[str, Any]:
+        prediction = self.runner.predict(
+            text=text,
+            return_probs=True,
+            threshold_issue=issue_threshold,
+            threshold_hidden=hidden_threshold,
+            threshold_error=error_threshold,
+            threshold_errors=error_thresholds,
+        )
+        return {
+            "text": text,
+            "class": prediction["class"],
+            "status": prediction["status"],
+            "review_required": prediction["review_required"],
+            "has_issue": prediction["has_logical_issue"],
+            "issue_probability": prediction["has_issue_probability"],
+            "hidden_problem": prediction["is_hidden_problem"],
+            "hidden_probability": prediction["hidden_probability"],
+            "errors": [
+                (item["type"], item["probability"])
+                for item in prediction["errors"]
+            ],
+            "num_errors": prediction["num_errors"],
+            "threshold_has_issue": prediction["threshold_has_issue"],
+            "threshold_is_hidden": prediction["threshold_is_hidden"],
+            "threshold_error": prediction["threshold_error"],
+        }
+
+    def pretty_print(self, result: Dict[str, Any], use_russian_names: bool = True) -> None:
+        converted = {
+            "class": result["class"],
+            "status": result["status"],
+            "review_required": result["review_required"],
+            "has_logical_issue": result["has_issue"],
+            "has_issue_probability": result["issue_probability"],
+            "threshold_has_issue": result["threshold_has_issue"],
+            "is_hidden_problem": result["hidden_problem"],
+            "hidden_probability": result["hidden_probability"],
+            "threshold_is_hidden": result["threshold_is_hidden"],
+            "errors": [
+                {
+                    "type": error_type,
+                    "probability": probability,
+                }
+                for error_type, probability in result["errors"]
+            ],
+        }
+        self.runner.pretty_print(converted, use_russian_names=use_russian_names)
+
+
+__all__ = ["RQAInferenceHF", "RQAJudge", "ERROR_NAMES_RU"]

modeling_rqa.py

@@ -0,0 +1,214 @@
+from typing import Any, Dict, List, Optional
+
+import torch
+import torch.nn as nn
+from transformers import AutoConfig, AutoModel, PreTrainedModel, PretrainedConfig
+
+
+class RQAModelConfig(PretrainedConfig):
+    model_type = "rqa_v2_2"
+
+    def __init__(
+        self,
+        base_model_name: str = "FacebookAI/xlm-roberta-large",
+        encoder_config: Optional[Dict[str, Any]] = None,
+        error_types: Optional[List[str]] = None,
+        schema_version: str = "rqa.v2.2",
+        has_issue_projection_dim: int = 256,
+        hidden_projection_dim: int = 256,
+        errors_projection_dim: int = 512,
+        has_issue_dropout: float = 0.25,
+        hidden_dropout: float = 0.25,
+        errors_dropout: float = 0.30,
+        temperature_has_issue: float = 1.0,
+        temperature_is_hidden: float = 1.0,
+        temperature_errors: Optional[List[float]] = None,
+        threshold_has_issue: float = 0.5,
+        threshold_is_hidden: float = 0.5,
+        threshold_error: float = 0.5,
+        threshold_errors: Optional[List[float]] = None,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+
+        self.schema_version = str(schema_version)
+        self.base_model_name = base_model_name
+        self.encoder_config = encoder_config
+        self.error_types = list(error_types or [])
+        self.num_error_types = len(self.error_types)
+
+        self.has_issue_projection_dim = int(has_issue_projection_dim)
+        self.hidden_projection_dim = int(hidden_projection_dim)
+        self.errors_projection_dim = int(errors_projection_dim)
+
+        self.has_issue_dropout = float(has_issue_dropout)
+        self.hidden_dropout = float(hidden_dropout)
+        self.errors_dropout = float(errors_dropout)
+
+        self.temperature_has_issue = float(temperature_has_issue)
+        self.temperature_is_hidden = float(temperature_is_hidden)
+        self.temperature_errors = (
+            list(temperature_errors)
+            if temperature_errors is not None
+            else [1.0] * self.num_error_types
+        )
+
+        self.threshold_has_issue = float(threshold_has_issue)
+        self.threshold_is_hidden = float(threshold_is_hidden)
+        self.threshold_error = float(threshold_error)
+        self.threshold_errors = (
+            list(threshold_errors)
+            if threshold_errors is not None
+            else [self.threshold_error] * self.num_error_types
+        )
+
+        try:
+            self._experts_implementation = "eager"
+            self._experts_implementation_internal = "eager"
+        except Exception:
+            pass
+
+
+def build_encoder_config_from_saved_dict(
+    encoder_config: Optional[Dict[str, Any]],
+    base_model_name: str,
+):
+    if encoder_config is None:
+        return AutoConfig.from_pretrained(base_model_name)
+
+    cfg_dict = dict(encoder_config)
+    model_type = cfg_dict.pop("model_type", None)
+    cfg_dict.pop("_name_or_path", None)
+
+    if model_type is not None:
+        try:
+            return AutoConfig.for_model(model_type, **cfg_dict)
+        except Exception:
+            pass
+
+    return AutoConfig.from_pretrained(base_model_name)
+
+
+class MeanPooling(nn.Module):
+    def forward(
+        self,
+        last_hidden_state: torch.Tensor,
+        attention_mask: torch.Tensor,
+    ) -> torch.Tensor:
+        mask = attention_mask.unsqueeze(-1).float()
+        summed = torch.sum(last_hidden_state * mask, dim=1)
+        denom = torch.clamp(mask.sum(dim=1), min=1e-9)
+        return summed / denom
+
+
+class RQAModelHF(PreTrainedModel):
+    config_class = RQAModelConfig
+    _supports_grouped_mm = False
+
+    def __init__(self, config: RQAModelConfig):
+        try:
+            config._experts_implementation = "eager"
+            config._experts_implementation_internal = "eager"
+        except Exception:
+            pass
+        super().__init__(config)
+
+        if config.encoder_config is None:
+            base_cfg = AutoConfig.from_pretrained(config.base_model_name)
+            config.encoder_config = base_cfg.to_dict()
+
+        enc_cfg = build_encoder_config_from_saved_dict(
+            encoder_config=config.encoder_config,
+            base_model_name=config.base_model_name,
+        )
+        self.encoder = AutoModel.from_config(enc_cfg)
+
+        hidden_size = self.encoder.config.hidden_size
+        self.pooler = MeanPooling()
+
+        self.has_issue_projection = nn.Sequential(
+            nn.Linear(hidden_size, config.has_issue_projection_dim),
+            nn.LayerNorm(config.has_issue_projection_dim),
+            nn.GELU(),
+            nn.Dropout(config.has_issue_dropout),
+        )
+        self.hidden_projection = nn.Sequential(
+            nn.Linear(hidden_size, config.hidden_projection_dim),
+            nn.LayerNorm(config.hidden_projection_dim),
+            nn.GELU(),
+            nn.Dropout(config.hidden_dropout),
+        )
+        self.errors_projection = nn.Sequential(
+            nn.Linear(hidden_size, config.errors_projection_dim),
+            nn.LayerNorm(config.errors_projection_dim),
+            nn.GELU(),
+            nn.Dropout(config.errors_dropout),
+        )
+
+        self.has_issue_head = nn.Linear(config.has_issue_projection_dim, 1)
+        self.is_hidden_head = nn.Linear(config.hidden_projection_dim, 1)
+        self.errors_head = nn.Linear(
+            config.errors_projection_dim,
+            config.num_error_types,
+        )
+
+        self.log_var_has_issue = nn.Parameter(torch.zeros(1))
+        self.log_var_is_hidden = nn.Parameter(torch.zeros(1))
+        self.log_var_errors = nn.Parameter(torch.zeros(1))
+        with torch.no_grad():
+            self.log_var_has_issue.clamp_(-5, 5)
+            self.log_var_is_hidden.clamp_(-5, 5)
+            self.log_var_errors.clamp_(-5, 5)
+
+        for module in [
+            self.has_issue_projection[0],
+            self.hidden_projection[0],
+            self.errors_projection[0],
+            self.has_issue_head,
+            self.is_hidden_head,
+            self.errors_head,
+        ]:
+            setattr(module, "_rqa_custom_init", True)
+
+        self.post_init()
+
+    def _init_weights(self, module):
+        if isinstance(module, nn.Linear) and getattr(module, "_rqa_custom_init", False):
+            nn.init.xavier_uniform_(module.weight)
+            if module.bias is not None:
+                nn.init.zeros_(module.bias)
+
+    def forward(
+        self,
+        input_ids: torch.Tensor,
+        attention_mask: torch.Tensor,
+        **kwargs,
+    ) -> Dict[str, torch.Tensor]:
+        outputs = self.encoder(
+            input_ids=input_ids,
+            attention_mask=attention_mask,
+            return_dict=True,
+            **kwargs,
+        )
+        pooled = self.pooler(outputs.last_hidden_state, attention_mask)
+
+        issue_features = self.has_issue_projection(pooled)
+        hidden_features = self.hidden_projection(pooled)
+        error_features = self.errors_projection(pooled)
+
+        return {
+            "has_issue_logits": self.has_issue_head(issue_features).squeeze(-1),
+            "is_hidden_logits": self.is_hidden_head(hidden_features).squeeze(-1),
+            "errors_logits": self.errors_head(error_features),
+        }
+
+
+try:
+    AutoConfig.register("rqa_v2_2", RQAModelConfig)
+except ValueError:
+    pass
+
+try:
+    AutoModel.register(RQAModelConfig, RQAModelHF)
+except ValueError:
+    pass