Documentation Index
Fetch the complete documentation index at: https://wb-21fd5541-john-wbdocs-2044-rename-serverless-products.mintlify.app/llms.txt
Use this file to discover all available pages before exploring further.
Il s’agit d’un notebook interactif. Vous pouvez l’exécuter en local ou utiliser les liens ci-dessous :
- Les expressions régulières pour identifier les données PII et les masquer.
- Presidio de Microsoft, un SDK de protection des données basé sur Python. Cet outil offre des fonctionnalités de masquage et de remplacement.
- Faker, une bibliothèque Python permettant de générer de fausses données, combinée à Presidio pour anonymiser les données PII.
Vous apprendrez également à utiliser la personnalisation de la journalisation des entrées et sorties de weave.op et autopatch_settings pour intégrer le masquage et l’anonymisation des données PII au flux de travail. Pour plus d’informations, voir Personnaliser les entrées et sorties journalisées.
Pour commencer, procédez comme suit :
- Consultez la section Aperçu.
- Complétez les prérequis.
- Consultez les méthodes disponibles pour identifier, masquer et anonymiser les données PII.
- Appliquez les méthodes aux appels Weave.
La section suivante donne un aperçu de la journalisation des données d’entrée et de sortie avec weave.op, ainsi que des bonnes pratiques pour travailler avec des données PII dans Weave.
Les ops Weave vous permettent de définir des fonctions de post-traitement pour les entrées et les sorties. À l’aide de ces fonctions, vous pouvez modifier les données transmises à votre appel à un LLM ou journalisées dans Weave.
Dans l’exemple suivant, deux fonctions de post-traitement sont définies et passées en arguments à weave.op().
from dataclasses import dataclass
from typing import Any
import weave
# Classe wrapper pour les entrées
@dataclass
class CustomObject:
x: int
secret_password: str
# On définit d'abord les fonctions de post-traitement des entrées et des sorties :
def postprocess_inputs(inputs: dict[str, Any]) -> dict[str, Any]:
return {k:v for k,v in inputs.items() if k != "hide_me"}
def postprocess_output(output: CustomObject) -> CustomObject:
return CustomObject(x=output.x, secret_password="REDACTED")
# Ensuite, lors de l'utilisation du décorateur `@weave.op`, on passe ces fonctions de traitement en arguments au décorateur :
@weave.op(
postprocess_inputs=postprocess_inputs,
postprocess_output=postprocess_output,
)
def some_llm_call(a: int, hide_me: str) -> CustomObject:
return CustomObject(x=a, secret_password=hide_me)
Bonnes pratiques d’utilisation de Weave avec des données PII
- Journalisez des données anonymisées pour vérifier la détection des PII
- Suivez les processus de gestion des PII avec Weave Traces
- Mesurez les performances de l’anonymisation sans exposer de véritables PII
- Ne journalisez jamais de PII en clair
- Chiffrez les champs sensibles avant de les journaliser
Conseils en matière de chiffrement
- Utilisez un chiffrement réversible pour les données que vous devrez déchiffrer ultérieurement
- Utilisez un hachage à sens unique pour les ID uniques que vous n’avez pas besoin de rétablir
- Envisagez un chiffrement spécialisé pour les données que vous devez analyser tout en restant chiffrées
- Commencez par installer les packages requis.
%%capture
# @title packages python requis :
!pip install cryptography
!pip install presidio_analyzer
!pip install presidio_anonymizer
!python -m spacy download en_core_web_lg # Presidio utilise le moteur NLP spacy
!pip install Faker # nous utiliserons Faker pour remplacer les données PII par des données fictives
!pip install weave # Pour exploiter les traces
!pip install set-env-colab-kaggle-dotenv -q # pour les variables d'environnement
!pip install anthropic # pour utiliser sonnet
!pip install cryptography # pour chiffrer nos données
-
Créez des clés API ici :
%%capture
# @title Configurer vos clés API correctement
# Voir : https://pypi.org/project/set-env-colab-kaggle-dotenv/ pour les instructions d'utilisation.
from set_env import set_env
_ = set_env("ANTHROPIC_API_KEY")
_ = set_env("WANDB_API_KEY")
- Initialisez votre projet Weave.
import weave
# Démarrer un nouveau projet Weave
WEAVE_PROJECT = "pii_cookbook"
weave.init(WEAVE_PROJECT)
- Chargez le jeu de données de démonstration PII, qui contient 10 blocs de texte.
import requests
url = "https://raw.githubusercontent.com/wandb/docs/main/weave/cookbooks/source/10_pii_data.json"
response = requests.get(url)
pii_data = response.json()
print('PII data first sample: "' + pii_data[0]["text"] + '"')
Aperçu des méthodes de masquage
- Expressions régulières pour identifier les données PII et les masquer.
- Microsoft Presidio, un SDK Python de protection des données qui offre des fonctionnalités de masquage et de remplacement.
- Faker, une bibliothèque Python permettant de générer de fausses données.
Méthode 1 : Filtrer à l’aide d’expressions régulières
import re
# Définir une fonction pour nettoyer les données PII avec regex
def redact_with_regex(text):
# Modèle de numéro de téléphone
# \b : Limite de mot
# \d{3} : Exactement 3 chiffres
# [-.]? : Trait d'union ou point facultatif
# \d{3} : 3 chiffres supplémentaires
# [-.]? : Trait d'union ou point facultatif
# \d{4} : Exactement 4 chiffres
# \b : Limite de mot
text = re.sub(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", "<PHONE>", text)
# Modèle d'adresse e-mail
# \b : Limite de mot
# [A-Za-z0-9._%+-]+ : Un ou plusieurs caractères pouvant figurer dans un nom d'utilisateur e-mail
# @ : Symbole @ littéral
# [A-Za-z0-9.-]+ : Un ou plusieurs caractères pouvant figurer dans un nom de domaine
# \. : Point littéral
# [A-Z|a-z]{2,} : Deux lettres majuscules ou minuscules ou plus (TLD)
# \b : Limite de mot
text = re.sub(
r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "<EMAIL>", text
)
# Modèle de numéro de sécurité sociale (SSN)
# \b : Limite de mot
# \d{3} : Exactement 3 chiffres
# - : Trait d'union littéral
# \d{2} : Exactement 2 chiffres
# - : Trait d'union littéral
# \d{4} : Exactement 4 chiffres
# \b : Limite de mot
text = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "<SSN>", text)
# Modèle de nom simple (non exhaustif)
# \b : Limite de mot
# [A-Z] : Une lettre majuscule
# [a-z]+ : Une ou plusieurs lettres minuscules
# \s : Un caractère d'espacement
# [A-Z] : Une lettre majuscule
# [a-z]+ : Une ou plusieurs lettres minuscules
# \b : Limite de mot
text = re.sub(r"\b[A-Z][a-z]+ [A-Z][a-z]+\b", "<NAME>", text)
return text
# Tester la fonction
test_text = "My name is John Doe, my email is john.doe@example.com, my phone is 123-456-7890, and my SSN is 123-45-6789."
cleaned_text = redact_with_regex(test_text)
print(f"Raw text:\n\t{test_text}")
print(f"Redacted text:\n\t{cleaned_text}")
Méthode 2 : Masquer à l’aide de Microsoft Presidio
Alex dans "My name is Alex" par <PERSON>.
Presidio prend en charge nativement les entités courantes. Dans l’exemple ci-dessous, nous masquons toutes les entités de type PHONE_NUMBER, PERSON, LOCATION, EMAIL_ADDRESS ou US_SSN. Le traitement Presidio est encapsulé dans une fonction.
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
# Configurer l'Analyzer, qui charge un module NLP (modèle spaCy par défaut) et d'autres reconnaisseurs PII.
analyzer = AnalyzerEngine()
# Configurer l'Anonymizer, qui utilisera les résultats de l'analyzer pour anonymiser le texte.
anonymizer = AnonymizerEngine()
# Encapsuler le processus de masquage Presidio dans une fonction
def redact_with_presidio(text):
# Analyser le texte pour identifier les données PII
results = analyzer.analyze(
text=text,
entities=["PHONE_NUMBER", "PERSON", "LOCATION", "EMAIL_ADDRESS", "US_SSN"],
language="en",
)
# Anonymiser les données PII identifiées
anonymized_text = anonymizer.anonymize(text=text, analyzer_results=results)
return anonymized_text.text
text = "My phone number is 212-555-5555 and my name is alex"
# Tester la fonction
anonymized_text = redact_with_presidio(text)
print(f"Raw text:\n\t{text}")
print(f"Redacted text:\n\t{anonymized_text}")
Méthode 3 : anonymiser par remplacement à l’aide de Faker et Presidio
"My name is Raphael and I like to fish. My phone number is 212-555-5555"
Une fois les données traitées avec Presidio et Faker, elles peuvent ressembler à ceci :
"My name is Katherine Dixon and I like to fish. My phone number is 667.431.7379"
Pour utiliser efficacement Presidio et Faker ensemble, nous devons fournir des références à nos opérateurs personnalisés. Ces opérateurs indiqueront à Presidio quelles fonctions Faker utiliser pour remplacer les PII par des données factices.
from faker import Faker
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
fake = Faker()
# Créer des fonctions faker (noter qu'elles doivent recevoir une valeur)
def fake_name(x):
return fake.name()
def fake_number(x):
return fake.phone_number()
# Créer un opérateur personnalisé pour les entités PERSON et PHONE_NUMBER
operators = {
"PERSON": OperatorConfig("custom", {"lambda": fake_name}),
"PHONE_NUMBER": OperatorConfig("custom", {"lambda": fake_number}),
}
text_to_anonymize = (
"My name is Raphael and I like to fish. My phone number is 212-555-5555"
)
# Résultat de l'analyseur
analyzer_results = analyzer.analyze(
text=text_to_anonymize, entities=["PHONE_NUMBER", "PERSON"], language="en"
)
anonymizer = AnonymizerEngine()
# ne pas oublier de transmettre les opérateurs définis ci-dessus à l'anonymiseur
anonymized_results = anonymizer.anonymize(
text=text_to_anonymize, analyzer_results=analyzer_results, operators=operators
)
print(f"Texte brut :\n\t{text_to_anonymize}")
print(f"Texte anonymisé :\n\t{anonymized_results.text}")
from typing import ClassVar
from faker import Faker
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
# Une classe personnalisée pour générer des données fictives qui étend Faker
class MyFaker(Faker):
# Créer des fonctions faker (noter qu'elles doivent recevoir une valeur)
def fake_address(self):
return fake.address()
def fake_ssn(self):
return fake.ssn()
def fake_name(self):
return fake.name()
def fake_number(self):
return fake.phone_number()
def fake_email(self):
return fake.email()
# Créer des opérateurs personnalisés pour les entités
operators: ClassVar[dict[str, OperatorConfig]] = {
"PERSON": OperatorConfig("custom", {"lambda": fake_name}),
"PHONE_NUMBER": OperatorConfig("custom", {"lambda": fake_number}),
"EMAIL_ADDRESS": OperatorConfig("custom", {"lambda": fake_email}),
"LOCATION": OperatorConfig("custom", {"lambda": fake_address}),
"US_SSN": OperatorConfig("custom", {"lambda": fake_ssn}),
}
def redact_and_anonymize_with_faker(self, text):
anonymizer = AnonymizerEngine()
analyzer_results = analyzer.analyze(
text=text,
entities=["PHONE_NUMBER", "PERSON", "LOCATION", "EMAIL_ADDRESS", "US_SSN"],
language="en",
)
anonymized_results = anonymizer.anonymize(
text=text, analyzer_results=analyzer_results, operators=self.operators
)
return anonymized_results.text
faker = MyFaker()
text_to_anonymize = (
"My name is Raphael and I like to fish. My phone number is 212-555-5555"
)
anonymized_text = faker.redact_and_anonymize_with_faker(text_to_anonymize)
print(f"Raw text:\n\t{text_to_anonymize}")
print(f"Anonymized text:\n\t{anonymized_text}")
Méthode 4 : Utiliser autopatch_settings
autopatch_settings pour configurer la gestion des PII directement lors de l’initialisation pour une ou plusieurs intégrations LLM prises en charge. Les avantages de cette méthode sont les suivants :
- La logique de gestion des PII est centralisée et définie à l’initialisation, ce qui réduit le besoin de logique personnalisée dispersée.
- Les flux de travail de traitement des PII peuvent être personnalisés ou entièrement désactivés pour des intégrations spécifiques.
Pour utiliser autopatch_settings afin de configurer la gestion des PII, définissez postprocess_inputs et/ou postprocess_output dans op_settings pour l’une des intégrations LLM prises en charge.
def postprocess(inputs: dict) -> dict:
if "SENSITIVE_KEY" in inputs:
inputs["SENSITIVE_KEY"] = "REDACTED"
return inputs
client = weave.init(
...,
autopatch_settings={
"openai": {
"op_settings": {
"postprocess_inputs": postprocess,
"postprocess_output": ...,
}
},
"anthropic": {
"op_settings": {
"postprocess_inputs": ...,
"postprocess_output": ...,
}
}
},
)
Appliquer les méthodes aux appels Weave
predict, dans laquelle l’API Anthropic sera appelée. Claude Sonnet d’Anthropic est utilisé pour effectuer une analyse de sentiment tout en traçant les appels LLM à l’aide de Traces. Claude Sonnet recevra un bloc de texte et renverra l’une des classifications de sentiment suivantes : positive, negative ou neutral. De plus, nous inclurons nos fonctions de post-traitement afin de garantir que nos données PII sont masquées ou anonymisées avant d’être envoyées au LLM.
Une fois ce code exécuté, vous recevrez des liens vers la page du projet Weave ainsi que vers la trace spécifique (appels LLM) que vous avez exécutée.
Dans le cas le plus simple, vous pouvez utiliser des expressions régulières pour identifier et masquer les données PII dans le texte original.
import json
from typing import Any
import anthropic
import weave
# Définir une fonction de post-traitement des entrées qui applique notre masquage par regex pour la prédiction du modèle Weave Op
def postprocess_inputs_regex(inputs: dict[str, Any]) -> dict:
inputs["text_block"] = redact_with_regex(inputs["text_block"])
return inputs
# Modèle Weave / fonction predict
class SentimentAnalysisRegexPiiModel(weave.Model):
model_name: str
system_prompt: str
temperature: int
@weave.op(
postprocess_inputs=postprocess_inputs_regex,
)
async def predict(self, text_block: str) -> dict:
client = anthropic.AsyncAnthropic()
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{"role": "user", "content": [{"type": "text", "text": text_block}]}
],
)
result = response.content[0].text
if result is None:
raise ValueError("No response from model")
parsed = json.loads(result)
return parsed
python
# créer notre modèle LLM avec un prompt système
model = SentimentAnalysisRegexPiiModel(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt='You are a Sentiment Analysis classifier. You will be classifying text based on their sentiment. Your input will be a block of text. You will answer with one the following rating option["positive", "negative", "neutral"]. Your answer should be one word in json format: {classification}. Ensure that it is valid JSON.',
temperature=0,
)
print("Model: ", model)
# pour chaque bloc de texte, anonymiser d'abord puis prédire
for entry in pii_data:
await model.predict(entry["text"])
Méthode de masquage avec Presidio
from typing import Any
import weave
# Définir une fonction de post-traitement des entrées qui applique notre masquage Presidio pour le Weave Op de prédiction du modèle
def postprocess_inputs_presidio(inputs: dict[str, Any]) -> dict:
inputs["text_block"] = redact_with_presidio(inputs["text_block"])
return inputs
# Modèle Weave / fonction predict
class SentimentAnalysisPresidioPiiModel(weave.Model):
model_name: str
system_prompt: str
temperature: int
@weave.op(
postprocess_inputs=postprocess_inputs_presidio,
)
async def predict(self, text_block: str) -> dict:
client = anthropic.AsyncAnthropic()
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{"role": "user", "content": [{"type": "text", "text": text_block}]}
],
)
result = response.content[0].text
if result is None:
raise ValueError("Aucune réponse du modèle")
parsed = json.loads(result)
return parsed
python
# créer notre modèle LLM avec un prompt système
model = SentimentAnalysisPresidioPiiModel(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt='Vous êtes un classificateur d\'analyse de sentiment. Vous allez classer des textes en fonction de leur sentiment. Votre entrée sera un bloc de texte. Vous répondrez avec l\'une des options de notation suivantes ["positive", "negative", "neutral"]. Votre réponse doit être un seul mot au format JSON : {classification}. Assurez-vous qu\'il s\'agit d\'un JSON valide.',
temperature=0,
)
print("Model: ", model)
# pour chaque bloc de texte, anonymiser d'abord puis prédire
for entry in pii_data:
await model.predict(entry["text"])
Méthode de remplacement avec Faker et Presidio
from typing import Any
import weave
# Définir une fonction de post-traitement des entrées qui applique l'anonymisation Faker et le masquage Presidio pour la prédiction du modèle Weave Op
faker = MyFaker()
def postprocess_inputs_faker(inputs: dict[str, Any]) -> dict:
inputs["text_block"] = faker.redact_and_anonymize_with_faker(inputs["text_block"])
return inputs
# Weave modèle / fonction predict
class SentimentAnalysisFakerPiiModel(weave.Model):
model_name: str
system_prompt: str
temperature: int
@weave.op(
postprocess_inputs=postprocess_inputs_faker,
)
async def predict(self, text_block: str) -> dict:
client = anthropic.AsyncAnthropic()
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{"role": "user", "content": [{"type": "text", "text": text_block}]}
],
)
result = response.content[0].text
if result is None:
raise ValueError("No response from model")
parsed = json.loads(result)
return parsed
python
# créer notre modèle LLM avec un prompt système
model = SentimentAnalysisFakerPiiModel(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt='You are a Sentiment Analysis classifier. You will be classifying text based on their sentiment. Your input will be a block of text. You will answer with one the following rating option["positive", "negative", "neutral"]. Your answer should be one word in json format: {classification}. Ensure that it is valid JSON.',
temperature=0,
)
print("Model: ", model)
# pour chaque bloc de texte, anonymiser d'abord puis prédire
for entry in pii_data:
await model.predict(entry["text"])
méthode autopatch_settings
postprocess_inputs pour anthropic à la fonction postprocess_inputs_regex() () lors de l’initialisation. La fonction postprocess_inputs_regex applique la méthode redact_with_regex définie dans Méthode 1 : filtrage par expression régulière. Désormais, redact_with_regex sera appliquée à toutes les entrées de n’importe quel modèle anthropic.
from typing import Any
import weave
client = weave.init(
...,
autopatch_settings={
"anthropic": {
"op_settings": {
"postprocess_inputs": postprocess_inputs_regex,
}
}
},
)
# Définir une fonction de post-traitement des entrées qui applique notre masquage par regex pour le Weave Op de prédiction du modèle
def postprocess_inputs_regex(inputs: dict[str, Any]) -> dict:
inputs["text_block"] = redact_with_regex(inputs["text_block"])
return inputs
# Modèle Weave / fonction predict
class SentimentAnalysisRegexPiiModel(weave.Model):
model_name: str
system_prompt: str
temperature: int
async def predict(self, text_block: str) -> dict:
client = anthropic.AsyncAnthropic()
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{"role": "user", "content": [{"type": "text", "text": text_block}]}
],
)
result = response.content[0].text
if result is None:
raise ValueError("No response from model")
parsed = json.loads(result)
return parsed
python
# créer notre modèle LLM avec un prompt système
model = SentimentAnalysisRegexPiiModel(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt='You are a Sentiment Analysis classifier. You will be classifying text based on their sentiment. Your input will be a block of text. You will answer with one the following rating option["positive", "negative", "neutral"]. Your answer should be one word in json format: {classification}. Ensure that it is valid JSON.',
temperature=0,
)
print("Model: ", model)
# pour chaque bloc de texte, anonymiser d'abord puis effectuer la prédiction
for entry in pii_data:
await model.predict(entry["text"])
(Facultatif) Chiffrez vos données
import os
from cryptography.fernet import Fernet
from pydantic import BaseModel, ValidationInfo, model_validator
def get_fernet_key():
# Vérifier si la clé existe dans les variables d'environnement
key = os.environ.get('FERNET_KEY')
if key is None:
# Si la clé n'existe pas, en générer une nouvelle
key = Fernet.generate_key()
# Enregistrer la clé dans une variable d'environnement
os.environ['FERNET_KEY'] = key.decode()
else:
# Si la clé existe, s'assurer qu'elle est en bytes
key = key.encode()
return key
cipher_suite = Fernet(get_fernet_key())
class EncryptedSentimentAnalysisInput(BaseModel):
encrypted_text: str = None
@model_validator(mode="before")
def encrypt_fields(cls, values):
if "text" in values and values["text"] is not None:
values["encrypted_text"] = cipher_suite.encrypt(values["text"].encode()).decode()
del values["text"]
return values
@property
def text(self):
if self.encrypted_text:
return cipher_suite.decrypt(self.encrypted_text.encode()).decode()
return None
@text.setter
def text(self, value):
self.encrypted_text = cipher_suite.encrypt(str(value).encode()).decode()
@classmethod
def encrypt(cls, text: str):
return cls(text=text)
def decrypt(self):
return self.text
# sentiment_analysis_model modifié pour utiliser le nouveau EncryptedSentimentAnalysisInput
class sentiment_analysis_model(weave.Model):
model_name: str
system_prompt: str
temperature: int
@weave.op()
async def predict(self, encrypted_input: EncryptedSentimentAnalysisInput) -> dict:
client = AsyncAnthropic()
decrypted_text = encrypted_input.decrypt() # On utilise la classe personnalisée pour déchiffrer le texte
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{ "role": "user",
"content":[
{
"type": "text",
"text": decrypted_text
}
]
}
]
)
result = response.content[0].text
if result is None:
raise ValueError("Aucune réponse du modèle")
parsed = json.loads(result)
return parsed
model = sentiment_analysis_model(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt="Vous êtes un classificateur d'analyse de sentiment. Vous allez classifier des textes en fonction de leur sentiment. Votre entrée sera un bloc de texte. Vous répondrez avec l'une des options de notation suivantes [\"positive\", \"negative\", \"neutral\"]. Votre réponse doit être un seul mot dans un dict au format json dont la clé est classification.",
temperature=0
)
for entry in pii_data:
encrypted_input = EncryptedSentimentAnalysisInput.encrypt(entry["text"])
await model.predict(encrypted_input)
