Configurer les garde-fous

import weave
import openai
from weave.scorers import OpenAIModerationScorer
import asyncio

# Initialiser Weave
weave.init("your-team-name/your-project-name")

# Initialiser le client OpenAI
client = openai.OpenAI()  # Utilise la variable d'environnement OPENAI_API_KEY

# Initialiser le scorer de modération
moderation_scorer = OpenAIModerationScorer()

# Envoyer les prompts à OpenAI
@weave.op
def generate_response(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": prompt}
        ],
        max_tokens=200
    )
    return response.choices[0].message.content

# La fonction garde-fou vérifie la toxicité des réponses
async def generate_safe_response(prompt: str) -> str:
    """Générer une réponse avec un garde-fou de modération de contenu."""
    # Obtenir le résultat et l'objet Call
    result, call = generate_response.call(prompt)
    
    # Appliquer le scorer de modération avant de retourner le résultat à l'utilisateur
    score = await call.apply_scorer(moderation_scorer)
    print("This is the score object:", score)
    
    # Vérifier si le contenu a été signalé
    if not score.result.get("passed", True): 
        categories = score.result.get("categories", {})
        flagged_categories = list(categories.keys()) if categories else []
        print(f"Contenu bloqué. Catégories signalées : {flagged_categories}")
        return "Je suis désolé, je ne peux pas fournir cette réponse en raison des restrictions de la politique de contenu."
    
    return result

# Exécuter les exemples
if __name__ == "__main__":
    
    prompts = [
        "What's the capital of France?",
        "Tell me a funny fact about dogs.",
    ]
    
    for prompt in prompts:
        print(f"\nPrompt: {prompt}")
        response = asyncio.run(generate_safe_response(prompt))
        print(f"Response: {response}")

import weave
import openai
import re
import asyncio
from weave import Scorer

weave.init("your-team-name/your-project-name")

client = openai.OpenAI()

class PIIDetectionScorer(Scorer):
    """Détecte les données personnelles (PII) dans les sorties LLM pour prévenir les fuites de données."""
    
    @weave.op
    def score(self, output: str) -> dict:
        """
        Vérifie la présence de motifs PII courants dans la sortie.
        
        Returns:
            dict: Contient 'passed' (bool) et 'detected_types' (list)
        """
        detected_types = []
        
        # Motif e-mail
        if re.search(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', output):
            detected_types.append("email")
        
        # Motif numéro de téléphone (format américain)
        if re.search(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', output):
            detected_types.append("phone")
        
        # Motif numéro de sécurité sociale (SSN)
        if re.search(r'\b\d{3}-\d{2}-\d{4}\b', output):
            detected_types.append("ssn")
        
        return {
            "passed": len(detected_types) == 0,
            "detected_types": detected_types
        }

# Initialiser le scorer en dehors de la fonction pour des performances optimales
pii_scorer = PIIDetectionScorer()

@weave.op
def generate_response(prompt: str) -> str:
    """Génère une réponse à l'aide d'un LLM."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Vous êtes un assistant serviable."},
            {"role": "user", "content": prompt}
        ],
        max_tokens=200
    )
    return response.choices[0].message.content

async def generate_safe_response(prompt: str) -> str:
    """Génère une réponse avec un garde-fou de détection des données personnelles (PII)."""
    result, call = generate_response.call(prompt)
    
    # Appliquer le scorer de détection PII
    score = await call.apply_scorer(pii_scorer)
    
    # Bloquer la réponse si des données PII sont détectées
    if not score.result.get("passed", True):
        detected_types = score.result.get("detected_types", [])
        return f"Je ne peux pas fournir une réponse susceptible de contenir des informations sensibles (détectées : {', '.join(detected_types)})."
    
    return result

# Exemple d'utilisation
if __name__ == "__main__":
    prompts = [
        "Quel temps fait-il aujourd'hui ?",
        "Pouvez-vous m'aider à contacter quelqu'un à john.doe@example.com ?",
        "Parlez-moi du machine learning.",
    ]
    
    for prompt in prompts:
        print(f"\nInvite : {prompt}")
        response = asyncio.run(generate_safe_response(prompt))
        print(f"Réponse : {response}")