Configurer des garde-fous - CoreWeave Docs

Les garde-fous agissent sur le comportement de votre application LLM en fonction des scores attribués par des juges LLM. Ils s’exécutent en temps réel, avant que les sorties ne parviennent aux utilisateurs, et peuvent bloquer ou modifier les réponses lorsque les scores dépassent certains seuils. Vous pouvez utiliser des garde-fous pour bloquer du contenu toxique, filtrer les informations personnelles identifiables (PII) dans les réponses ou bloquer les entrées abusives des utilisateurs. Ce guide explique le fonctionnement des garde-fous Weave et la manière d’optimiser leurs performances, puis présente des exemples qui s’appuient sur des évaluateurs intégrés, des évaluateurs personnalisés et AWS Bedrock garde-fous pour protéger les applications LLM en production.

Fonctionnement des garde-fous Weave

Les garde-fous Weave s’appuient sur des Weave Scorers exécutés en ligne (inline) pour évaluer l’entrée d’un utilisateur ou la sortie d’un LLM, et ajuster les réponses du LLM en temps réel. Vous pouvez configurer des évaluateurs personnalisés ou utiliser des évaluateurs intégrés pour analyser du contenu à des fins diverses. Ce guide montre comment utiliser ces deux types d’évaluateurs comme garde-fous. Si vous souhaitez évaluer passivement le trafic de production sans modifier le flux de contrôle de votre application, utilisez plutôt des moniteurs. Contrairement aux moniteurs, les garde-fous nécessitent de modifier le code, car ils agissent sur le flux de contrôle de votre application. Cependant, chaque résultat d’évaluateur produit par les garde-fous est automatiquement stocké dans la base de données de Weave : vos garde-fous font donc également office de moniteurs, sans aucune configuration supplémentaire. Vous pouvez analyser l’historique des résultats d’évaluateurs, quelle que soit la manière dont ils ont été utilisés à l’origine.

Le SDK TypeScript de Weave ne prend pas en charge les outils nécessaires à la configuration des garde-fous.

Optimiser les performances de vos garde-fous Weave

Comme les garde-fous peuvent interrompre le flux de contrôle de votre application et infléchir ses réponses, ils peuvent nuire aux performances s’ils sont trop complexes. Pour des performances optimales, suivez ces recommandations :

Limitez la logique des garde-fous au strict minimum et veillez à ce qu’elle soit rapide.
Mettez en cache les résultats fréquents.
Évitez les appels d’API externes coûteux.
Initialisez les garde-fous en dehors de vos fonctions principales afin d’éviter de payer à chaque fois le coût de l’initialisation.

Il est particulièrement important d’initialiser vos garde-fous en dehors de la fonction principale lorsque :

Vos évaluateurs chargent des modèles de ML.
Vous utilisez des LLM locaux pour lesquels la latence est critique.
Vos évaluateurs maintiennent des connexions réseau.
Vos applications reçoivent un trafic élevé.

Exemple : créer un garde-fou à l’aide d’un évaluateur de modération intégré

L’exemple suivant envoie des prompts utilisateur au modèle GPT-4o mini d’OpenAI. La réponse du modèle est ensuite transmise à l’API de modération d’OpenAI, qui évalue si elle contient du contenu nuisible ou toxique. La réponse du modèle est transmise à la fonction de garde-fou (generate_safe_response()), qui utilise OpenAIModerationScorer pour vérifier la réponse d’origine du LLM. La logique de la fonction examine ensuite le booléen du champ passed dans la réponse d’évaluation d’OpenAI, qui détermine la manière dont l’application répond.

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 l’évaluateur 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 si les réponses sont toxiques
async def generate_safe_response(prompt: str) -> str:
    """Generate a response with content moderation guardrail."""
    # Obtenir le résultat et l’objet Call
    result, call = generate_response.call(prompt)
    
    # Appliquer l’évaluateur de modération avant de renvoyer la réponse à 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"Content blocked. Flagged categories: {flagged_categories}")
        return "I'm sorry, I can't provide that response due to content policy restrictions."
    
    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}")

Lorsque vous utilisez des évaluateurs LLM-as-a-judge, vous pouvez faire référence, dans vos prompts d’évaluation, à des variables issues de vos ops. Par exemple : « Évaluez si {output} est exact en vous basant sur {ground_truth}. » Pour plus d’informations, voir variables de prompt.

Exemple : créer un garde-fou à l’aide d’un évaluateur personnalisé

L’exemple suivant crée un garde-fou personnalisé qui détecte les informations personnelles identifiables (PII) dans les réponses d’un LLM, comme les adresses e-mail, les numéros de téléphone ou les numéros de sécurité sociale. Vous évitez ainsi que des informations sensibles soient exposées dans le contenu généré. La fonction generate_safe_response applique le PIIDetectionScorer personnalisé.

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):
    """Detects PII in LLM outputs to prevent data leaks."""
    
    @weave.op
    def score(self, output: str) -> dict:
        """
        Check for common PII patterns in the output.
        
        Returns:
            dict: Contains 'passed' (bool) and 'detected_types' (list)
        """
        detected_types = []
        
        # Motif d’adresse 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 de 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 de numéro de sécurité sociale américain (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 optimiser les performances
pii_scorer = PIIDetectionScorer()

@weave.op
def generate_response(prompt: str) -> str:
    """Generate a response using an LLM."""
    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

async def generate_safe_response(prompt: str) -> str:
    """Generate a response with PII detection guardrail."""
    result, call = generate_response.call(prompt)
    
    # Appliquer le scorer de détection des PII
    score = await call.apply_scorer(pii_scorer)
    
    # Bloquer la réponse si des PII sont détectées
    if not score.result.get("passed", True):
        detected_types = score.result.get("detected_types", [])
        return f"I cannot provide a response that may contain sensitive information (detected: {', '.join(detected_types)})."
    
    return result

# Exemple d’utilisation
if __name__ == "__main__":
    prompts = [
        "What's the weather like today?",
        "Can you help me contact someone at john.doe@example.com?",
        "Tell me about machine learning.",
    ]
    
    for prompt in prompts:
        print(f"\nPrompt: {prompt}")
        response = asyncio.run(generate_safe_response(prompt))
        print(f"Response: {response}")

Intégrer Weave à AWS Bedrock garde-fous

Si vous gérez déjà des politiques de contenu dans AWS, vous pouvez les appliquer dans Weave à l’aide du BedrockGuardrailScorer, qui s’appuie sur AWS Bedrock garde-fous pour détecter et filtrer le contenu selon les politiques configurées. Avant de configurer une intégration Bedrock garde-fous, vous avez besoin des éléments suivants :

Un compte AWS disposant d’un accès à Bedrock.
Un garde-fou configuré dans la console AWS Bedrock.
Le package Python boto3.

Vous n’avez pas besoin de créer votre propre client Bedrock : Weave le crée pour vous. Pour spécifier une région, transmettez sa valeur dans le paramètre bedrock_runtime_kwargs de l’évaluateur. Pour découvrir comment créer un garde-fou dans AWS Bedrock, consultez le notebook sur les Bedrock garde-fous. L’exemple suivant vérifie le texte généré par rapport aux politiques AWS Bedrock garde-fous avant de renvoyer les résultats aux utilisateurs :

import weave
from weave.scorers.bedrock_guardrails import BedrockGuardrailScorer

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

guardrail_scorer = BedrockGuardrailScorer(
    guardrail_id="your-guardrail-id",
    guardrail_version="DRAFT",
    source="INPUT",
    bedrock_runtime_kwargs={"region_name": "us-east-1"}
)

@weave.op
def generate_text(prompt: str) -> str:
    # Placez ici votre logique de génération de texte
    return "Generated text..."

async def generate_safe_text(prompt: str) -> str:
    result, call = generate_text.call(prompt)

    score = await call.apply_scorer(guardrail_scorer)

    if not score.result.passed:
        if score.result.metadata.get("modified_output"):
            return score.result.metadata["modified_output"]
        return "I cannot generate that content due to content policy restrictions."

    return result

​Fonctionnement des garde-fous Weave

​Optimiser les performances de vos garde-fous Weave

​Exemple : créer un garde-fou à l’aide d’un évaluateur de modération intégré

​Exemple : créer un garde-fou à l’aide d’un évaluateur personnalisé

​Intégrer Weave à AWS Bedrock garde-fous

Fonctionnement des garde-fous Weave

Optimiser les performances de vos garde-fous Weave

Exemple : créer un garde-fou à l’aide d’un évaluateur de modération intégré

Exemple : créer un garde-fou à l’aide d’un évaluateur personnalisé

Intégrer Weave à AWS Bedrock garde-fous