fix(detectors): skip HF detectors when models are unreachable

garak/detectors/base.py

@@ -126,38 +126,52 @@ def __init__(self, config_root=_config):
             if param in model_kwargs.keys():
                 model_kwargs.pop(param)
 
-        self.config = AutoConfig.from_pretrained(
-            self.detector_model_path, **model_kwargs
-        )
+        try:
+            self.config = AutoConfig.from_pretrained(
+                self.detector_model_path, **model_kwargs
+            )
 
-        self.config.init_device = self.device
+            self.config.init_device = self.device
 
-        self.detector_model = AutoModelForSequenceClassification.from_pretrained(
-            self.detector_model_path, config=self.config
-        ).to(self.device)
-        self.detector_tokenizer = AutoTokenizer.from_pretrained(
-            self.detector_model_path
-        )
-        self.detector = TextClassificationPipeline(
-            model=self.detector_model,
-            tokenizer=self.detector_tokenizer,
-            device=self.device,
-        )
-        for k, v in generation_params.items():
-            setattr(self.detector.generation_config, k, v)
-
-        if stored_env:
-            os.environ[disable_env_key] = stored_env
-        else:
-            del os.environ[disable_env_key]
+            self.detector_model = AutoModelForSequenceClassification.from_pretrained(
+                self.detector_model_path, config=self.config
+            ).to(self.device)
+            self.detector_tokenizer = AutoTokenizer.from_pretrained(
+                self.detector_model_path
+            )
+            self.detector = TextClassificationPipeline(
+                model=self.detector_model,
+                tokenizer=self.detector_tokenizer,
+                device=self.device,
+            )
+            for k, v in generation_params.items():
+                setattr(self.detector.generation_config, k, v)
+        except Exception as e:
+            if self.graceful_fail:
+                logging.warning(
+                    "Skipping HF detector %s — could not load model: %s",
+                    self.detector_model_path,
+                    e,
+                )
+                self.skip = True
+                self.detector = None
+                return
+            raise
+        finally:
+            if stored_env:
+                os.environ[disable_env_key] = stored_env
+            else:
+                del os.environ[disable_env_key]
 
-        transformers_logging.set_verbosity(orig_loglevel)
+            transformers_logging.set_verbosity(orig_loglevel)
 
     def detect(self, attempt: garak.attempt.Attempt) -> List[float | None]:
         # goal: return None for None outputs
         # don't adjust attempt.outputs
 
         all_outputs = attempt.outputs_for(self.lang_spec)
+        if self.skip or self.detector is None:
+            return [None] * len(all_outputs)
         non_none_outputs = [
             v.text for k, v in enumerate(all_outputs) if v and v.text is not None
         ]

garak/langproviders/local.py

@@ -4,6 +4,7 @@
 
 """Local language providers & translators."""
 
+import logging
 from typing import List, Callable
 
 from garak.exception import BadGeneratorException
@@ -64,63 +65,78 @@ def _load_langprovider(self):
         stored_env = os.getenv(disable_env_key, default=None)
         os.environ[disable_env_key] = "true"
 
-        if "m2m100" in self.model_name:
-            from transformers import M2M100ForConditionalGeneration, M2M100Tokenizer
-
-            # fmt: off
-            # Reference: https://huggingface.co/facebook/m2m100_418M#languages-covered
-            lang_support = {
-                "af", "am", "ar", "ast", "az",
-                "ba", "be", "bg", "bn", "br",
-                "bs", "ca", "ceb", "cs", "cy",
-                "da", "de", "el", "en", "es",
-                "et", "fa", "ff", "fi", "fr",
-                "fy", "ga", "gd", "gl", "gu",
-                "ha", "he", "hi", "hr", "ht",
-                "hu", "hy", "id", "ig", "ilo",
-                "is", "it", "ja", "jv", "ka",
-                "kk", "km", "kn", "ko", "lb",
-                "lg", "ln", "lo", "lt", "lv",
-                "mg", "mk", "ml", "mn", "mr",
-                "ms", "my", "ne", "nl", "no",
-                "ns", "oc", "or", "pa", "pl",
-                "ps", "pt", "ro", "ru", "sd",
-                "si", "sk", "sl", "so", "sq",
-                "sr", "ss", "su", "sv", "sw",
-                "ta", "th", "tl", "tn", "tr",
-                "uk", "ur", "uz", "vi", "wo",
-                "xh", "yi", "yo", "zh", "zu",
-            }
-            # fmt: on
-            if not (
-                self.source_lang in lang_support and self.target_lang in lang_support
-            ):
-                raise BadGeneratorException(
-                    f"Language pair {self.language} is not supported for this translation service."
-                )
-
-            self.model = M2M100ForConditionalGeneration.from_pretrained(
-                self.model_name
-            ).to(self.device)
-            self.tokenizer = M2M100Tokenizer.from_pretrained(self.model_name)
-        else:
-            from transformers import MarianMTModel, MarianTokenizer
-
-            # if model is not m2m100 expect the model name to be "Helsinki-NLP/opus-mt-{}" where the format string
-            # is replace with the language path defined in the configuration as self.source_lang-self.target_lang
-            # validation of all supported pairs is deferred in favor of allowing the download to raise exception
-            # when no published model exists with the pair requested in the name.
-            model_suffix = f"{self.source_lang}-{self.target_lang}"
-            model_name = self.model_name.format(model_suffix)
-            self.model = MarianMTModel.from_pretrained(model_name).to(self.device)
-            self.tokenizer = MarianTokenizer.from_pretrained(model_name)
-
-        if stored_env:
-            os.environ[disable_env_key] = stored_env
-        else:
-            del os.environ[disable_env_key]
+        try:
+            if "m2m100" in self.model_name:
+                from transformers import M2M100ForConditionalGeneration, M2M100Tokenizer
+
+                # fmt: off
+                # Reference: https://huggingface.co/facebook/m2m100_418M#languages-covered
+                lang_support = {
+                    "af", "am", "ar", "ast", "az",
+                    "ba", "be", "bg", "bn", "br",
+                    "bs", "ca", "ceb", "cs", "cy",
+                    "da", "de", "el", "en", "es",
+                    "et", "fa", "ff", "fi", "fr",
+                    "fy", "ga", "gd", "gl", "gu",
+                    "ha", "he", "hi", "hr", "ht",
+                    "hu", "hy", "id", "ig", "ilo",
+                    "is", "it", "ja", "jv", "ka",
+                    "kk", "km", "kn", "ko", "lb",
+                    "lg", "ln", "lo", "lt", "lv",
+                    "mg", "mk", "ml", "mn", "mr",
+                    "ms", "my", "ne", "nl", "no",
+                    "ns", "oc", "or", "pa", "pl",
+                    "ps", "pt", "ro", "ru", "sd",
+                    "si", "sk", "sl", "so", "sq",
+                    "sr", "ss", "su", "sv", "sw",
+                    "ta", "th", "tl", "tn", "tr",
+                    "uk", "ur", "uz", "vi", "wo",
+                    "xh", "yi", "yo", "zh", "zu",
+                }
+                # fmt: on
+                if not (
+                    self.source_lang in lang_support and self.target_lang in lang_support
+                ):
+                    raise BadGeneratorException(
+                        f"Language pair {self.language} is not supported for this translation service."
+                    )
+
+                self.model = M2M100ForConditionalGeneration.from_pretrained(
+                    self.model_name
+                ).to(self.device)
+                self.tokenizer = M2M100Tokenizer.from_pretrained(self.model_name)
+            else:
+                from transformers import MarianMTModel, MarianTokenizer
+
+                # if model is not m2m100 expect the model name to be "Helsinki-NLP/opus-mt-{}" where the format string
+                # is replace with the language path defined in the configuration as self.source_lang-self.target_lang
+                # validation of all supported pairs is deferred in favor of allowing the download to raise exception
+                # when no published model exists with the pair requested in the name.
+                model_suffix = f"{self.source_lang}-{self.target_lang}"
+                model_name = self.model_name.format(model_suffix)
+                self.model = MarianMTModel.from_pretrained(model_name).to(self.device)
+                self.tokenizer = MarianTokenizer.from_pretrained(model_name)
+            self._offline_passthru = False
+        except Exception as e:
+            logging.warning(
+                "Could not load Hugging Face translation model %s for %s — "
+                "falling back to passthru: %s",
+                self.model_name,
+                self.language,
+                e,
+            )
+            self._offline_passthru = True
+            self.model = None
+            self.tokenizer = None
+        finally:
+            if stored_env:
+                os.environ[disable_env_key] = stored_env
+            else:
+                del os.environ[disable_env_key]
 
     def _translate(self, text: str) -> str:
+        if getattr(self, "_offline_passthru", False):
+            return text
         if "m2m100" in self.model_name:
             self.tokenizer.src_lang = self.source_lang
 

tests/detectors/test_detectors_base.py

@@ -445,3 +445,27 @@ def test_stringdetector_normalize_ldnoobw_fullwidth_end_to_end():
 
     detector.normalize = "NFKC"
     assert detector.detect(attempt) == [1.0], "NFKC should catch the fullwidth variant in LDNOOBW"
+
+
+class _OfflineHFDetector(garak.detectors.base.HFDetector):
+    detector_model_path = "garak-llm/offline-test-model"
+    detector_target_class = "label"
+
+
+def test_hf_detector_skips_when_model_cannot_be_loaded(monkeypatch):
+    def _raise_network_error(*args, **kwargs):
+        raise OSError("Network is unreachable")
+
+    monkeypatch.setattr(
+        "transformers.AutoConfig.from_pretrained",
+        _raise_network_error,
+    )
+
+    detector = _OfflineHFDetector()
+
+    assert detector.skip is True
+    assert detector.detector is None
+
+    attempt = Attempt(prompt=Message(text="prompt"))
+    attempt.outputs = [Message("output")]
+    assert detector.detect(attempt) == [None]

tests/langproviders/test_local_offline.py

@@ -0,0 +1,29 @@
+# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from garak.langproviders.local import LocalHFTranslator
+
+
+def test_local_hf_translator_falls_back_to_passthru_when_model_unavailable(monkeypatch):
+    def _raise_network_error(*args, **kwargs):
+        raise OSError("Network is unreachable")
+
+    monkeypatch.setattr(
+        "transformers.MarianMTModel.from_pretrained",
+        _raise_network_error,
+    )
+
+    translator = LocalHFTranslator(
+        config_root={
+            "langproviders": {
+                "local": {
+                    "language": "de,en",
+                    "model_type": "local",
+                    "model_name": "Helsinki-NLP/opus-mt-{}",
+                }
+            }
+        }
+    )
+
+    assert translator._offline_passthru is True
+    assert translator._translate("hallo welt") == "hallo welt"