Aller au contenu

Usage basique

Cette page presente les usages fondamentaux de la bibliotheque sans integration LangChain.

Anonymisation simple avec le pipeline

pipeline.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import asyncio

from gliner2 import GLiNER2

from piighost.anonymizer import Anonymizer
from piighost.detector import Gliner2Detector
from piighost.linker.entity import ExactEntityLinker
from piighost.entity_resolver import MergeEntityConflictResolver
from piighost.pipeline import AnonymizationPipeline
from piighost.placeholder import LabelCounterPlaceholderFactory
from piighost.span_resolver import ConfidenceSpanConflictResolver

# Charger le modele GLiNER2
model = GLiNER2.from_pretrained("fastino/gliner2-multi-v1")

# Instancier chaque composant
detector = Gliner2Detector(model=model, labels=["PERSON", "LOCATION"], threshold=0.5)
span_resolver = ConfidenceSpanConflictResolver()
entity_linker = ExactEntityLinker()
entity_resolver = MergeEntityConflictResolver()
anonymizer = Anonymizer(LabelCounterPlaceholderFactory())

# Assembler le pipeline
pipeline = AnonymizationPipeline(
    detector=detector,  # (1)!
    span_resolver=span_resolver,  # (2)!
    entity_linker=entity_linker,  # (3)!
    entity_resolver=entity_resolver,  # (4)!
    anonymizer=anonymizer,  # (5)!
)


async def main():
    # Anonymiser un texte
    anonymized, entities = await pipeline.anonymize(
        "Patrick habite a Paris. Patrick aime Paris.",
    )
    print(anonymized)
    # <<PERSON:1>> habite a <<LOCATION:1>>. <<PERSON:1>> aime <<LOCATION:1>>.

    # Desanonymiser
    original, _ = await pipeline.deanonymize(anonymized)
    print(original)
    # Patrick habite a Paris. Patrick aime Paris.


asyncio.run(main())
  1. Détecter : trouve les PII candidates dans le texte via un détecteur NER (ici GLiNER2, interchangeable avec spaCy ou Transformers).
  2. Résoudre les spans : arbitre les chevauchements lorsque plusieurs détecteurs rapportent des positions qui se recouvrent.
  3. Lier les entités : regroupe les occurrences d'une même PII (variantes de casse, typos, mentions partielles).
  4. Résoudre les entités : fusionne les groupes qui partagent une mention entre détecteurs.
  5. Anonymiser : remplace chaque entité par un placeholder produit par la factory (ici <<PERSON:1>>, <<LOCATION:1>>…).

Inspection des entites

Le pipeline retourne les entites utilisees pour l'anonymisation :

async def main():
    anonymized, entities = await pipeline.anonymize(
        "Marie Dupont travaille chez Acme Corp a Lyon.",
    )
    print(anonymized)
    # <<PERSON:1>> travaille chez <<ORGANIZATION:1>> a <<LOCATION:1>>.

    for entity in entities:
        canonical = entity.detections[0].text
        print(f"'{canonical}' [{entity.label}] {len(entity.detections)} detection(s)")

asyncio.run(main())

Pipeline conversationnel avec memoire

Pour les scénarios multi-messages (conversation), ThreadAnonymizationPipeline accumule les entités entre les messages et fournit désanonymisation/réanonymisation par remplacement de chaîne.

import asyncio

from piighost.anonymizer import Anonymizer
from piighost.pipeline import ThreadAnonymizationPipeline
from piighost.detector import Gliner2Detector
from piighost.linker.entity import ExactEntityLinker
from piighost.entity_resolver import MergeEntityConflictResolver
from piighost.placeholder import LabelCounterPlaceholderFactory
from piighost.span_resolver import ConfidenceSpanConflictResolver

detector = Gliner2Detector(model=model, labels=["PERSON", "LOCATION"], threshold=0.5)
span_resolver = ConfidenceSpanConflictResolver()
entity_linker = ExactEntityLinker()
entity_resolver = MergeEntityConflictResolver()
anonymizer = Anonymizer(LabelCounterPlaceholderFactory())

conv_pipeline = ThreadAnonymizationPipeline(
    detector=detector,
    span_resolver=span_resolver,
    entity_linker=entity_linker,
    entity_resolver=entity_resolver,
    anonymizer=anonymizer,
)


async def conversation():
    # Premier message : detection NER + mise en cache
    r1, _ = await conv_pipeline.anonymize("Patrick est a Paris.")
    print(r1)
    # <<PERSON:1>> est a <<LOCATION:1>>.

    # Meme texte : cache hit (pas de second appel NER)
    r2, _ = await conv_pipeline.anonymize("Patrick est a Paris.")
    print(r2)
    # <<PERSON:1>> est a <<LOCATION:1>>.

    # Desanonymiser n'importe quelle chaine avec tokens (async)
    restored = await conv_pipeline.deanonymize_with_ent("Bonjour, <<PERSON:1>> !")
    print(restored)
    # Bonjour, Patrick !

    # Reanonymiser (original → token)
    reanon = conv_pipeline.anonymize_with_ent("Reponse pour Patrick a Paris")
    print(reanon)
    # Reponse pour <<PERSON:1>> a <<LOCATION:1>>


asyncio.run(conversation())

Differentes placeholder factories

Par defaut, LabelCounterPlaceholderFactory genere des tags <<LABEL:N>>. Vous pouvez changer de strategie :

from piighost.placeholder import LabelHashPlaceholderFactory, LabelPlaceholderFactory

# Hash : tags opaques deterministes
pipeline_hash = AnonymizationPipeline(
    ...,
    anonymizer=Anonymizer(LabelHashPlaceholderFactory()),
)
# Produit : <<PERSON:a1b2c3d4>>

# Redact : toutes les entites recoivent <<LABEL>> (pas de compteur)
pipeline_redact = AnonymizationPipeline(
    ...,
    anonymizer=Anonymizer(LabelPlaceholderFactory()),
)
# Produit : <<PERSON>>

Pour tester unitairement les pipelines sans charger un modèle NER, voir le guide Tests.

Voir aussi la page Étendre PIIGhost pour créer des composants personnalisés.