--- icon: lucide/puzzle tags: - Avancé - Détecteur --- # Étendre piighost Chaque étape du pipeline est un **port**, un `Protocol` que vous satisfaites en implémentant sa méthode unique. Aucune classe de base à hériter, et rien d'autre dans le pipeline ne change. Là où un patron `Base*` existe, vous pouvez aussi le sous-classer. Ce patron fournit le squelette commun et vous laisse un seul point d'extension. ```mermaid flowchart LR P[AnonymizationPipeline] -->|detector| D[AnyDetector] P -->|overlap_resolver| O[AnyOverlapResolver] P -->|expander| X[AnyDetectionExpander] P -->|linker| L[AnyEntityLinker] P -->|entity_resolver| R[AnyEntityResolver] P -->|anonymizer| A[AnyAnonymizer] P -->|guard| G[AnyGuardRail] A -->|factory| F[AnyPlaceholderFactory] ``` *Le pipeline injecte un composant par port. Seul le détecteur est requis. Le linker, l'anonymiseur et le résolveur de chevauchements utilisent par défaut des composants intégrés. Les étapes d'expansion, de résolution d'entités, de garde-fou et de listes à masquer ou à laisser en clair sont désactivées par défaut.* { .figure-caption } Les ports vivent dans le `base.py` de chaque composant, sous `piighost.components.*`. Les modèles de données qu'ils échangent vivent dans `piighost.models`. Une `Detection` est un `Span(start, end)` portant `text`, `label` et une `confidence` dans l'intervalle 0 à 1. Une `Entity` regroupe les détections qui partagent une valeur, et en dérive son `label`, son `text` et ses `spans`. Voir la [référence des modèles de données](reference/models.md) pour chaque champ, méthode et erreur de validation. ## Un détecteur personnalisé Un détecteur trouve les données confidentielles (données personnelles, secrets) dans un texte. Implémentez une seule méthode. ```python class AnyDetector(Protocol): async def detect(self, text: str) -> list[Detection]: ... ``` `detect` est asynchrone pour qu'une implémentation puisse attendre un serveur de modèle ou une API LLM. Renvoyez les détections dans n'importe quel ordre. Les chevauchements et les répétitions sont résolus par les étapes suivantes, pas ici. ???+ example "Détecteur regex de pseudos" ```python import re from piighost.models import Detection, Span class HandleDetector: """Detect @handles as USERNAME.""" async def detect(self, text: str) -> list[Detection]: detections: list[Detection] = [] for match in re.finditer(r"@\w+", text): span = Span(match.start(), match.end()) detections.append( Detection( span=span, text=match.group(), label="USERNAME", confidence=1.0, ) ) return detections ``` ### Utiliser le détecteur ```python from piighost.pipeline import AnonymizationPipeline detector = HandleDetector() pipeline = AnonymizationPipeline(detector) ``` Pour alimenter un détecteur depuis une liste de valeurs figée dans les tests, utilisez plutôt le détecteur intégré `ExactMatchDetector`. Voir [Tester sans modèle](examples/testing.md). ### Pour les modèles NER, sous-classez `BaseNERDetector` Les détecteurs adossés à un modèle (`Gliner2Detector`, `SpacyDetector`, `TransformersDetector`) étendent tous `BaseNERDetector`. `BaseNERDetector` traduit le label qu'un modèle émet en interne vers le label qui apparaît dans `Detection.label`. Vous pouvez ainsi interroger un modèle avec les chaînes qu'il détecte le mieux, tout en produisant des labels propres en aval. Passez `labels` sous forme de liste pour garder chaque label tel quel (mapping identité), ou sous forme de dictionnaire `{émis: interne}` pour renommer. ```python from piighost.components.detector.ner import Gliner2Detector # Query GLiNER2 with "person" and "company" but emit "PERSON" / "COMPANY". detector = Gliner2Detector( model="fastino/gliner2-multi-v1", labels={"PERSON": "person", "COMPANY": "company"}, ) ``` ## Un résolveur de chevauchements personnalisé Un résolveur de chevauchements reçoit des détections dont les spans se chevauchent, et en tire un ensemble de détections sans chevauchement. Le port : ```python class AnyOverlapResolver(Protocol): def resolve(self, detections: list[Detection]) -> list[Detection]: ... ``` Plutôt que d'implémenter `resolve` de zéro, sous-classez `BaseOverlapResolver`. Il regroupe les détections en groupes de chevauchement et confie chaque groupe à votre `_reduce`, si bien que vous décidez seulement quelles détections garder dans un groupe qui se chevauche. ???+ example "Le span le plus long l'emporte" ```python from piighost.components.overlap_resolver.base import BaseOverlapResolver from piighost.models import Detection class LongestOverlapResolver(BaseOverlapResolver): """Keep the longest detection in each overlap group.""" def _reduce(self, conflicting: list[Detection]) -> list[Detection]: return [max(conflicting, key=lambda d: d.span.length)] ``` Le `ConfidenceOverlapResolver` intégré garde plutôt la détection de plus haute confiance. Le résolveur de chevauchements est toujours actif. Omettez-le et le pipeline installe un `ConfidenceOverlapResolver`. Passez le vôtre pour changer la règle. Il n'y a aucun moyen supporté de le désactiver, car le rendu suppose des spans disjoints et lève sinon `OverlappingSpansError`. ## Un expander personnalisé Un expander trouve les occurrences qu'un détecteur a manquées, comme la répétition d'un nom repéré ailleurs. Le port : ```python class AnyDetectionExpander(Protocol): def expand(self, text: str, detections: list[Detection]) -> list[Detection]: ... ``` Sous-classez `BaseDetectionExpander`. Il conserve les détections d'origine. Pour chacune, il ajoute une détection à chaque occurrence supplémentaire que renvoie votre `_find_occurrences`. Chaque détection ajoutée reprend le label et la confiance de la détection source. Une occurrence qui chevauche une détection déjà retenue est écartée, car l'expander passe après le résolveur de chevauchements et le rendu refuse deux spans qui se recouvrent. Les valeurs sont cherchées de la plus longue à la plus courte, donc un nom complet prend sa place avant son prénom. ???+ example "Répétitions par mot entier" ```python from collections.abc import Iterable from piighost.components.expander.base import BaseDetectionExpander from piighost.models import Detection, Span from piighost.text import find_all_word_boundary class WholeWordExpander(BaseDetectionExpander): """Find whole-word repeats of a detected value.""" def _find_occurrences(self, text: str, detection: Detection) -> Iterable[Span]: return find_all_word_boundary(text, detection.text) ``` Le `WordBoundaryExpander` intégré fait exactement cela. L'étape est optionnelle. ## Un linker d'entités personnalisé Un linker regroupe en entités les détections qui réfèrent à la même valeur. Toutes les occurrences d'une valeur partagent ainsi un placeholder. Le port : ```python class AnyEntityLinker(Protocol): def link(self, detections: list[Detection]) -> list[Entity]: ... ``` Sous-classez `BaseEntityLinker`. Il regroupe les détections selon une clé que vous calculez dans `_key`. Il crée une entité par clé distincte, dans l'ordre de première occurrence. ???+ example "Regrouper par valeur exacte et label" ```python from collections.abc import Hashable from piighost.components.linker.base import BaseEntityLinker from piighost.models import Detection class CaseSensitiveLinker(BaseEntityLinker): """Group detections that share an exact value and label.""" def _key(self, detection: Detection) -> Hashable: return (detection.text, detection.label) ``` L'`ExactEntityLinker` intégré regroupe selon la clé de valeur. Cette clé est la même pour les mêmes mots, quelles que soient leurs espaces et leur casse. `Patrick`{ .pii } et `patrick`{ .pii } deviennent donc une seule entité. Utilisez `piighost.text.value_key` dans votre propre linker pour suivre la même règle, voir [Espaces Unicode](reference/detectors.md#espaces-unicode). ## Un résolveur d'entités personnalisé Un résolveur d'entités réconcilie les entités qui ne devraient pas coexister, comme deux entités qui partagent une détection. Le port : ```python class AnyEntityResolver(Protocol): def resolve(self, entities: list[Entity]) -> list[Entity]: ... ``` Sous-classez `BaseEntityResolver`. Il regroupe les entités qui partagent une détection et confie chaque groupe à votre `_reduce`. Votre `_reduce` renvoie un ensemble cohérent, soit en fusionnant le groupe en une entité, soit en gardant les entités séparées. Les composants intégrés : - `MergeEntityResolver` fusionne les entités qui partagent une détection, par union-find. - `SeparateEntityResolver` les garde séparées, en donnant chaque détection partagée à une entité. - `FuzzyEntityResolver` fusionne les entités aux valeurs proches (nécessite l'extra `fuzzy`). L'étape est optionnelle. ## Une fabrique de placeholders personnalisée Une fabrique de placeholders transforme les entités en leurs jetons de remplacement. Elle est générique sur un **tag de préservation**, un type fantôme qui déclare ce que ses jetons préservent. Le type-checker se sert de ce tag pour verrouiller un consommateur comme le middleware. Le port : ```python class AnyPlaceholderFactory(Protocol[PreservationT_co]): def create(self, entities: list[Entity]) -> Mapping[Entity, PreservationT_co]: ... ``` Un jeton est une instance du tag, et le tag est une sous-classe de `str`. Le jeton est donc une vraie chaîne, qui porte son niveau de préservation dans son propre type. `create` doit être déterministe. Les mêmes entités produisent les mêmes jetons à chaque appel, car le pipeline l'appelle plusieurs fois par exécution. ???+ example "Fabrique de labels entre crochets" ```python from collections.abc import Mapping from piighost.components.placeholder.base import AnyPlaceholderFactory from piighost.components.placeholder.tags import PreservesLabel from piighost.models import Entity class BracketLabelFactory(AnyPlaceholderFactory[PreservesLabel]): """Emit [LABEL] for every entity, collapsing each label to one token.""" def create(self, entities: list[Entity]) -> Mapping[Entity, PreservesLabel]: return {entity: PreservesLabel(f"[{entity.label}]") for entity in entities} ``` `PreservesLabel` dit que le jeton révèle le type mais pas une identité unique. Cette fabrique convient donc au caviardage à usage unique, pas au middleware. Pour un jeton que le middleware sait dé-identifier et retrouver, taguez-le `PreservesRecognizableIdentity` (ou un sous-tag comme `PreservesLabeledIdentityOpaque`) et utilisez une grammaire délimitée comme `<>`{ .placeholder }. Pour envelopper une forme interne dans des délimiteurs sans écrire l'enveloppe vous-même, sous-classez `BaseDelimitedPlaceholderFactory`. Voir [Fabriques de placeholders](placeholder-factories.md) pour la taxonomie complète des tags et des exemples détaillés. ### Utiliser la fabrique ```python from piighost.components.anonymizer import Anonymizer factory = BracketLabelFactory() anonymizer = Anonymizer(factory) ``` ## Un garde-fou personnalisé Un garde-fou re-contrôle la sortie dé-identifiée à la recherche de données confidentielles résiduelles. Il classe, il ne décide pas. Il renvoie un `GuardVerdict` et laisse le pipeline lever `PIIRemainingError` quand un verdict est signalé. Il n'y a pas de patron `Base`, parce que chaque garde a son propre mécanisme de contrôle. Le port : ```python class AnyGuardRail(Protocol): async def check(self, text: str) -> GuardVerdict: ... ``` `check` ne voit que le texte dé-identifié. Les placeholders de ce texte sont clairement synthétiques. Un contrôle qui cherche les vraies valeurs ne les prend donc pas pour de vraies valeurs. ???+ example "Signaler un @ résiduel" ```python from piighost.components.guard.base import GuardVerdict class AtSignGuard: """Flag any residual @ sign as leftover PII.""" async def check(self, text: str) -> GuardVerdict: return GuardVerdict(flagged="@" in text) ``` Le `DetectorGuardRail` intégré relance un détecteur et rapporte les détections résiduelles. L'étape est optionnelle. Ne passez aucun `guard` et la sortie est renvoyée sans contrôle. ### Utiliser le garde-fou ```python from piighost.pipeline import AnonymizationPipeline guard = AtSignGuard() pipeline = AnonymizationPipeline(detector, guard=guard) ``` ### Un modèle de décision derrière le port Un modèle de décision ne génère pas de texte. Il répond à une question dont les réponses possibles sont fixées à l'avance, ici oui ou non. Un garde-fou fait la même chose sur le texte dé-identifié. [`examples/guard_rail_laya.py`](https://github.com/Athroniaeth/piighost/blob/master/examples/guard_rail_laya.py) place [Laya](https://huggingface.co/convaiinnovations/laya), un équivalent de Jev sous licence Apache 2.0, derrière le port en une douzaine de lignes, en local. Il demande s'il reste une donnée personnelle et signale le texte au-delà d'une probabilité. Sur 24 textes dé-identifiés, dont la moitié laisse fuir une valeur, il a rattrapé 11 fuites sur 12 et signalé 5 textes propres sur 12 au seuil de 0,5. `Gliner2GuardRail` rattrapait 4 fuites, sans aucune fausse alerte. Les placeholders font monter son score, donc il signale à tort surtout un texte chargé en jetons. Son modèle anglais lit assez bien le français, `laya-multilingual` non. ## Composition complète Les étapes sont indépendantes, donc un détecteur, une fabrique et un garde personnalisés se combinent librement avec les composants intégrés : ```python from piighost.components.anonymizer import Anonymizer from piighost.components.entity_resolver import MergeEntityResolver from piighost.pipeline import AnonymizationPipeline pipeline = AnonymizationPipeline( HandleDetector(), anonymizer=Anonymizer(BracketLabelFactory()), entity_resolver=MergeEntityResolver(), guard=AtSignGuard(), ) ``` Pour tester un composant personnalisé de façon déterministe, alimentez-le via `ExactMatchDetector`. Voir [Tester sans modèle](examples/testing.md).