Adding modelPath and vocabPath to pheye for local inference.

Author: jzonthemtn

phileas/filters/ph_eye_filter.py

@@ -26,13 +26,20 @@
 class PhEyeFilter(BaseFilter):
     def __init__(self, filter_config):
         super().__init__(FilterType.PH_EYE, filter_config)
+        self._model = None
 
     def filter(self, text: str, context: str = "default") -> List[Span]:
+        model_path = getattr(self.filter_config, "model_path", "")
+        vocab_path = getattr(self.filter_config, "vocab_path", "")
+        labels = getattr(self.filter_config, "labels", ["PERSON"])
+
+        if model_path and vocab_path:
+            return self._local_filter(text, context, model_path, vocab_path, labels)
+
         endpoint = getattr(self.filter_config, "endpoint", "")
         if not endpoint:
             return []
 
-        labels = getattr(self.filter_config, "labels", ["PERSON"])
         thresholds = getattr(self.filter_config, "thresholds", {})
         bearer_token = getattr(self.filter_config, "bearer_token", "")
         timeout = getattr(self.filter_config, "timeout", 30) or 30
@@ -42,6 +49,8 @@ def filter(self, text: str, context: str = "default") -> List[Span]:
             "context": context,
             "piece": 0,
             "labels": list(labels),
+            "modelPath": model_path,
+            "vocabPath": vocab_path,
         }).encode("utf-8")
 
         req = urllib.request.Request(
@@ -107,3 +116,63 @@ def filter(self, text: str, context: str = "default") -> List[Span]:
             ))
 
         return spans
+
+    def _local_filter(self, text: str, context: str, model_path: str, vocab_path: str, labels: List[str]) -> List[Span]:
+        if not self._model:
+            onnx = False
+            if model_path.endswith(".onnx"):
+                onnx = True
+
+            try:
+                from gliner import GLiNER
+            except ImportError:
+                raise ImportError("The 'gliner' package is required for local inference. Install it with 'pip install gliner'.")
+
+            self._model = GLiNER.from_pretrained(model_path, onnx=onnx, vocab_path=vocab_path)
+
+        ph_eye_spans = self._model.predict_entities(text, labels)
+
+        strategies = self._get_strategies()
+        strategy = strategies[0] if strategies else None
+        ignored_terms = set(self._get_ignored())
+        thresholds = getattr(self.filter_config, "thresholds", {})
+
+        spans: List[Span] = []
+        for item in ph_eye_spans:
+            label = item.get("label", "")
+            score = float(item.get("score", 0.0))
+            span_text = item.get("text", "")
+            start = int(item.get("start", 0))
+            end = int(item.get("end", 0))
+
+            if labels and label not in labels:
+                continue
+
+            threshold = thresholds.get(label.upper(), 0.0)
+            if score < threshold:
+                continue
+
+            if span_text in ignored_terms:
+                continue
+
+            if label.upper() == "PERSON":
+                filter_type = "person"
+            else:
+                filter_type = label.lower() if label else FilterType.PH_EYE
+
+            replacement = (
+                strategy.get_replacement(filter_type, span_text) if strategy else span_text
+            )
+
+            spans.append(Span(
+                character_start=start,
+                character_end=end,
+                filter_type=filter_type,
+                context=context,
+                confidence=score,
+                text=span_text,
+                replacement=replacement,
+                ignored=False,
+            ))
+
+        return spans

phileas/policy/identifiers.py

@@ -166,6 +166,8 @@ class PhEyeFilterConfig:
     timeout: int = 30
     labels: List[str] = field(default_factory=lambda: ["PERSON"])
     remove_punctuation: bool = False
+    model_path: str = ""
+    vocab_path: str = ""
     thresholds: dict = field(default_factory=dict)
     ph_eye_filter_strategies: List[FilterStrategy] = field(default_factory=_default_strategies)
     ignored: List[str] = field(default_factory=list)
@@ -369,6 +371,8 @@ def from_dict(cls, data: dict) -> "Identifiers":
                     timeout=d.get("timeout", 30),
                     labels=d.get("labels", ["PERSON"]),
                     remove_punctuation=d.get("removePunctuation", False),
+                    model_path=d.get("modelPath", ""),
+                    vocab_path=d.get("vocabPath", ""),
                     thresholds=d.get("thresholds", {}),
                     ph_eye_filter_strategies=_strategies_from_dict(d, "phEyeFilterStrategies"),
                     ignored=d.get("ignored", []),
@@ -524,6 +528,8 @@ def to_dict(self) -> dict:
                     "timeout": cfg.timeout,
                     "labels": cfg.labels,
                     "removePunctuation": cfg.remove_punctuation,
+                    "modelPath": cfg.model_path,
+                    "vocabPath": cfg.vocab_path,
                     "thresholds": cfg.thresholds,
                     "phEyeFilterStrategies": [s.to_dict() for s in cfg.ph_eye_filter_strategies],
                     "ignored": cfg.ignored,

pyproject.toml

@@ -67,6 +67,10 @@ dev = [
 server = [
     "flask>=3.0",
 ]
+gliner = [
+    "gliner>=0.1.0",
+    "onnxruntime>=1.16.0",
+]
 
 [project.scripts]
 phileas-server = "phileas.server:main"

tests/test_filters.py

@@ -1064,6 +1064,74 @@ def test_url_error_raises_ioerror(self):
             with pytest.raises(IOError):
                 f.filter("John Smith was here.")
 
+    def test_local_filter_gliner(self):
+        import unittest.mock as mock
+        config = PhEyeFilterConfig(
+            model_path="/path/to/model.bin",
+            vocab_path="/path/to/vocab.txt",
+            labels=["PERSON"]
+        )
+        f = PhEyeFilter(config)
+
+        mock_gliner_class = mock.MagicMock()
+        mock_model = mock.MagicMock()
+        mock_gliner_class.from_pretrained.return_value = mock_model
+
+        # Mocking the response from GLiNER
+        mock_model.predict_entities.return_value = [
+            {"start": 0, "end": 10, "label": "PERSON", "text": "John Smith", "score": 0.95}
+        ]
+
+        with mock.patch.dict("sys.modules", {"gliner": mock.MagicMock()}):
+            import gliner
+            with mock.patch("gliner.GLiNER", mock_gliner_class):
+                spans = f.filter("John Smith was here.")
+
+        assert len(spans) == 1
+        assert spans[0].text == "John Smith"
+        assert spans[0].filter_type == "person"
+        assert spans[0].confidence == 0.95
+
+        mock_gliner_class.from_pretrained.assert_called_once_with("/path/to/model.bin", onnx=False, vocab_path="/path/to/vocab.txt")
+
+    def test_local_filter_onnx(self):
+        import unittest.mock as mock
+        config = PhEyeFilterConfig(
+            model_path="/path/to/model.onnx",
+            vocab_path="/path/to/vocab.txt",
+            labels=["PERSON"]
+        )
+        f = PhEyeFilter(config)
+
+        mock_gliner_class = mock.MagicMock()
+        mock_model = mock.MagicMock()
+        mock_gliner_class.from_pretrained.return_value = mock_model
+
+        mock_model.predict_entities.return_value = [
+            {"start": 0, "end": 10, "label": "PERSON", "text": "John Smith", "score": 0.95}
+        ]
+
+        with mock.patch.dict("sys.modules", {"gliner": mock.MagicMock()}):
+            import gliner
+            with mock.patch("gliner.GLiNER", mock_gliner_class):
+                spans = f.filter("John Smith was here.")
+
+        assert len(spans) == 1
+        mock_gliner_class.from_pretrained.assert_called_once_with("/path/to/model.onnx", onnx=True, vocab_path="/path/to/vocab.txt")
+
+    def test_local_filter_import_error(self):
+        import unittest.mock as mock
+        config = PhEyeFilterConfig(
+            model_path="/path/to/model.bin",
+            vocab_path="/path/to/vocab.txt"
+        )
+        f = PhEyeFilter(config)
+
+        with mock.patch.dict("sys.modules", {"gliner": None}):
+            with pytest.raises(ImportError) as excinfo:
+                f.filter("Some text.")
+            assert "The 'gliner' package is required" in str(excinfo.value)
+
     def test_policy_json_ph_eye(self):
         from phileas.policy.policy import Policy
         policy_json = json.dumps({