feat: PII detection pipeline and post-generation audit

src/guardian_rag/pii/__init__.py

@@ -1 +1,41 @@
 """PII detection and redaction pipeline."""
+
+from __future__ import annotations
+
+from guardian_rag.pii.audit import FAIL_CLOSED_REPLACEMENT, PostGenerationAuditor
+from guardian_rag.pii.deberta_validator import DebertaValidator, DebertaValidatorBackend
+from guardian_rag.pii.detector import PiiDetector
+from guardian_rag.pii.gliner_recognizer import (
+    DEFAULT_PII_LABELS,
+    GlinerBackend,
+    GlinerRecognizer,
+    GlinerSpan,
+)
+from guardian_rag.pii.models import (
+    AuditResult,
+    PiiEntity,
+    PiiEntityType,
+    PiiStage,
+    RedactionMap,
+)
+from guardian_rag.pii.presidio_engine import PresidioEngine
+from guardian_rag.pii.redactor import Redactor
+
+__all__ = [
+    "DEFAULT_PII_LABELS",
+    "FAIL_CLOSED_REPLACEMENT",
+    "AuditResult",
+    "DebertaValidator",
+    "DebertaValidatorBackend",
+    "GlinerBackend",
+    "GlinerRecognizer",
+    "GlinerSpan",
+    "PiiDetector",
+    "PiiEntity",
+    "PiiEntityType",
+    "PiiStage",
+    "PostGenerationAuditor",
+    "PresidioEngine",
+    "RedactionMap",
+    "Redactor",
+]

src/guardian_rag/pii/_real_backends.py

@@ -0,0 +1,66 @@
+"""Production backends that wrap real HuggingFace models.
+
+Imports of ``gliner`` and ``transformers`` are performed lazily inside the
+constructors so the test suite, which substitutes deterministic fakes,
+never has to install or download the heavy ML stack.
+
+This module is excluded from pyright and coverage. Keep it small: every
+line here is a line the type-checker cannot verify.
+"""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+from guardian_rag.pii.gliner_recognizer import GlinerSpan
+
+
+class RealGlinerBackend:
+    """Stage-2 backend backed by the official ``gliner`` package."""
+
+    def __init__(self, model_name: str = "urchade/gliner_multi_pii-v1") -> None:
+        from gliner import GLiNER
+
+        self._model = GLiNER.from_pretrained(model_name)
+
+    def predict_entities(
+        self,
+        text: str,
+        labels: list[str],
+        threshold: float,
+    ) -> list[GlinerSpan]:
+        spans = cast(
+            "list[GlinerSpan]",
+            self._model.predict_entities(text, labels, threshold=threshold),
+        )
+        return spans
+
+
+class RealDebertaValidatorBackend:
+    """Stage-3 backend backed by a HuggingFace zero-shot NLI pipeline."""
+
+    HYPOTHESIS_TEMPLATE = "This text is {}."
+
+    def __init__(
+        self,
+        model_name: str = "MoritzLaurer/deberta-v3-xsmall-zeroshot-v1.1-all-33",
+    ) -> None:
+        from transformers import pipeline
+
+        self._pipeline = pipeline("zero-shot-classification", model=model_name)
+
+    def score(self, text: str, hypothesis: str) -> float:
+        result = cast(
+            "dict[str, Any]",
+            self._pipeline(
+                text,
+                candidate_labels=[hypothesis, "something else"],
+                hypothesis_template=self.HYPOTHESIS_TEMPLATE,
+            ),
+        )
+        labels = cast("list[str]", result["labels"])
+        scores = cast("list[float]", result["scores"])
+        for label, score in zip(labels, scores, strict=True):
+            if label == hypothesis:
+                return float(score)
+        return 0.0

src/guardian_rag/pii/audit.py

@@ -0,0 +1,104 @@
+"""Post-generation PII audit.
+
+Most RAG privacy projects stop at ingestion. The LLM can still paraphrase
+or infer personal data from clean chunks. This module re-runs the same
+three-stage detector over every generated answer before it reaches the
+user.
+
+Failure semantics matter here: if the audit raises, we MUST NOT return
+the raw answer. The default is fail-closed: an audit exception turns into
+a redacted answer with `fail_closed=True`. Setting `fail_closed=False`
+re-raises instead, which only makes sense in offline tooling like the
+eval harness.
+"""
+
+from __future__ import annotations
+
+import structlog
+
+from guardian_rag.pii.detector import PiiDetector
+from guardian_rag.pii.models import AuditResult
+from guardian_rag.pii.redactor import Redactor
+
+logger = structlog.get_logger(__name__)
+
+
+# Returned in place of the LLM answer when the audit itself crashed. Kept
+# deliberately short so the API layer can surface it to the user without
+# leaking implementation details.
+FAIL_CLOSED_REPLACEMENT = "[Audit failed: response redacted for safety]"
+
+
+class PostGenerationAuditor:
+    """Run the PII detector over an LLM answer and redact any leak.
+
+    Parameters
+    ----------
+    detector:
+        The same three-stage detector ingestion uses. Sharing one instance
+        means the audit catches exactly what would have been redacted at
+        ingest time.
+    redactor:
+        Typed-placeholder redactor. Same instance as ingestion for
+        consistent placeholder labels.
+    fail_closed:
+        When True (the default), any exception inside `audit()` is caught,
+        the answer is replaced with `FAIL_CLOSED_REPLACEMENT`, and the
+        result is returned with `fail_closed=True`. When False the
+        exception is re-raised, which is useful for offline tooling.
+    """
+
+    def __init__(
+        self,
+        detector: PiiDetector,
+        redactor: Redactor,
+        fail_closed: bool = True,
+    ) -> None:
+        self._detector = detector
+        self._redactor = redactor
+        self._fail_closed = fail_closed
+
+    def audit(self, answer: str, document_id: str = "answer") -> AuditResult:
+        """Return an `AuditResult` for one LLM answer.
+
+        On `clean=True` the returned `answer` is identical to the input.
+        On `clean=False` it is the redacted version. On `fail_closed=True`
+        the input is replaced with a safe placeholder.
+        """
+        try:
+            entities = self._detector.detect(answer)
+        except Exception:
+            if not self._fail_closed:
+                raise
+            return _fail_closed(answer, "detect")
+
+        if not entities:
+            return AuditResult(clean=True, answer=answer)
+
+        try:
+            redacted, _map = self._redactor.redact(answer, entities, document_id)
+        except Exception:
+            if not self._fail_closed:
+                raise
+            return _fail_closed(answer, "redact")
+
+        logger.warning(
+            "audit.leak_detected",
+            n_entities=len(entities),
+            entity_types=sorted({e.entity_type.value for e in entities}),
+        )
+        return AuditResult(clean=False, answer=redacted, entities=entities)
+
+
+def _fail_closed(answer: str, phase: str) -> AuditResult:
+    logger.exception(
+        "audit.failed_closed",
+        phase=phase,
+        answer_preview=answer[:120],
+    )
+    return AuditResult(
+        clean=False,
+        answer=FAIL_CLOSED_REPLACEMENT,
+        entities=[],
+        fail_closed=True,
+    )

src/guardian_rag/pii/deberta_validator.py

@@ -0,0 +1,119 @@
+"""Stage 3 detector: conditional DeBERTa zero-shot validation.
+
+When GLiNER produces a span whose confidence is below the configured
+threshold, we ask a DeBERTa zero-shot NLI model "is this span really a
+{label}?". This is the safety net for borderline detections without paying
+DeBERTa inference cost on every span.
+
+The backend is exposed as a `Protocol` so tests can drop in a deterministic
+fake. The real backend lives in `_real_backends.py`.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+import structlog
+
+from guardian_rag.pii.models import PiiEntity, PiiStage
+
+logger = structlog.get_logger(__name__)
+
+
+# Each domain PII type gets a short, human-readable label that DeBERTa NLI
+# can score against. The wording matters: "personal name" works better than
+# "PERSON" because the model was trained on natural language hypotheses.
+ENTITY_TO_HYPOTHESIS: dict[str, str] = {
+    "PERSON": "a personal name",
+    "EMAIL": "an email address",
+    "PHONE": "a phone number",
+    "ADDRESS": "a postal address",
+    "DATE_OF_BIRTH": "a date of birth",
+    "ID_NUMBER": "an identification number",
+    "BANK_ACCOUNT": "a bank account number",
+    "CREDIT_CARD": "a credit card number",
+    "SOCIAL_SECURITY": "a social security number",
+    "IP_ADDRESS": "an IP address",
+    "URL": "a URL",
+    "ORGANIZATION": "an organisation name",
+    "LOCATION": "a geographic location",
+}
+
+
+@runtime_checkable
+class DebertaValidatorBackend(Protocol):
+    """Pluggable DeBERTa backend so tests can avoid loading weights."""
+
+    def score(self, text: str, hypothesis: str) -> float:
+        """Probability in [0, 1] that ``hypothesis`` entails ``text``."""
+        ...
+
+
+class DebertaValidator:
+    """Validate low-confidence GLiNER spans against a zero-shot NLI model.
+
+    Parameters
+    ----------
+    backend:
+        Anything implementing `DebertaValidatorBackend`.
+    score_threshold:
+        GLiNER spans with score below this are sent to DeBERTa for
+        validation. Spans with score at or above this are accepted as-is.
+    accept_threshold:
+        DeBERTa probability above which a validated span is kept. Below
+        this the span is dropped (treated as a GLiNER false positive).
+    """
+
+    def __init__(
+        self,
+        backend: DebertaValidatorBackend,
+        score_threshold: float = 0.85,
+        accept_threshold: float = 0.5,
+    ) -> None:
+        if not 0.0 <= score_threshold <= 1.0:
+            raise ValueError("score_threshold must be in [0, 1]")
+        if not 0.0 <= accept_threshold <= 1.0:
+            raise ValueError("accept_threshold must be in [0, 1]")
+        self._backend = backend
+        self._score_threshold = score_threshold
+        self._accept_threshold = accept_threshold
+
+    def validate(self, entities: list[PiiEntity]) -> list[PiiEntity]:
+        """Return the subset of ``entities`` that survive validation.
+
+        High-confidence spans (>= score_threshold) pass through unchanged.
+        Low-confidence spans are scored by DeBERTa. Spans whose DeBERTa
+        probability is below `accept_threshold` are dropped. Surviving
+        validated spans have their stage marked GLINER_VALIDATED and their
+        score replaced with the DeBERTa probability.
+        """
+        survivors: list[PiiEntity] = []
+        for entity in entities:
+            if entity.score >= self._score_threshold:
+                survivors.append(entity)
+                continue
+            hypothesis = ENTITY_TO_HYPOTHESIS.get(entity.entity_type.value)
+            if hypothesis is None:
+                # No hypothesis means we cannot validate; keep the span to
+                # err on the privacy-safe side.
+                survivors.append(entity)
+                continue
+            probability = float(self._backend.score(entity.text, hypothesis))
+            if probability < self._accept_threshold:
+                logger.debug(
+                    "deberta.validator.dropped",
+                    text=entity.text,
+                    entity_type=entity.entity_type.value,
+                    gliner_score=entity.score,
+                    deberta_score=probability,
+                )
+                continue
+            survivors.append(
+                entity.model_copy(
+                    update={
+                        "score": probability,
+                        "stage": PiiStage.GLINER_VALIDATED,
+                    }
+                )
+            )
+        return survivors