AI guardrails: Bias scorers

git clone https://github.com/wandb/weave.git && cd weave && git fetch origin pull/3006/head:xtra-scorers && git checkout xtra-scorers && pip install -qq -e . && pip install openai asyncio

import weave; weave.init('bias-scorers')
from weave.scorers import BiasScorer


# Initialize the BiasScorer
scorer = BiasScorer(device="cpu")  # Use "cuda" for GPU if available

# Define a list of sample texts to evaluate
texts = [
    "This is a hateful message.",
    "This statement is free of bias."
]

# Retrieve logits for more detailed analysis
for text in texts:
    result_with_logits = scorer.score(text)
    print(f"Text: {text}")
    print(f"Detailed Result: {result_with_logits}\n")

# Example with adjusted threshold
scorer_with_threshold = BiasScorer(device="cpu", threshold=0.8)
for text in texts:
    result = scorer_with_threshold.score(text)
    print(f"Text (with threshold=0.8): {text}")
    print(f"Result: {result}\n")

from openai import OpenAI
import weave; weave.init('bias-scorers')
client = OpenAI()


response = client.moderations.create(
  model="omni-moderation-latest",
  input="text to check bias for",
)

print(response)

from weave.scorers.llamaguard_scorer import LlamaGuardScorer
import asyncio

import weave; weave.init('bias-scorers')


async def main():
    # Initialize the LlamaGuardScorer with the 1B model
    scorer = LlamaGuardScorer(
        model_name_or_path="meta-llama/Llama-Guard-3-1B",
        device="cpu",  # Use "cuda" if a GPU is available
    )

    # Text to score
    sample_text = "Your input text here. Check if this is safe or not."

    # Run the scorer
    result = await scorer.score(output=sample_text)

    # Display the result
    print("LlamaGuard Scoring Result:")
    print(f"Safe: {result['safe']}")
    print(f"Unsafe Score: {result['extras']['unsafe_score']}")
    if not result['safe']:
        print("Violated Categories:")
        for category, violated in result['extras']['categories'].items():
            if violated:
                print(f" - {category}")


if __name__ == "__main__":
    asyncio.run(main())

import weave
from openai import OpenAI
from weave.scorers import BiasScorer
import pandas as pd
from datasets import load_dataset
import asyncio
import os
import torch
import random
import ast
from weave.scorers.llamaguard_scorer import LlamaGuardScorer

# Set device
device = "cuda" if torch.cuda.is_available() else "cpu"


# Initialize scorer
llama_scorer = LlamaGuardScorer(
    model_name_or_path="meta-llama/Llama-Guard-3-8B" if device == "cuda" else "meta-llama/Llama-Guard-3-1B",
    device=device  # Use "cuda" if you have a GPU
)

client = OpenAI(api_key="your api key")
scorer = BiasScorer()
weave.init("bias_eval")

# Function to parse scores
def parse_scores(scores):
    try:
        parsed = ast.literal_eval(scores)
        if isinstance(parsed, list):
            return parsed
    except (ValueError, SyntaxError):
        pass
    return []

# Load and process dataset
def load_ds(max_samples=200000, balanced_samples=30, seed=42, cache_path="cached_dataset.csv"):
    if os.path.exists(cache_path):
        print(f"Loading cached dataset from {cache_path}...")
        df = pd.read_csv(cache_path)
        return df.to_dict("records")

    print("Loading dataset...")
    random.seed(seed)
    dataset = load_dataset("PleIAs/ToxicCommons", split="train", streaming=True)
    limited_data = list(dataset.take(max_samples))
    random.shuffle(limited_data)

    processed_data = {"positive_samples": [], "negative_samples": []}
    for row in limited_data:
        text = row.get("original_text", "")
        scores = row.get("scores", "")
        scores_list = parse_scores(scores)

        if isinstance(text, str) and text.strip() and isinstance(scores_list, list):
            if len(scores_list) >= 2:
                if scores_list[0] > 0 or scores_list[1] > 0:  # Positive condition
                    if len(processed_data["positive_samples"]) < balanced_samples:
                        processed_data["positive_samples"].append({"text": text, "label": 1})
                else:  # Non-positive condition
                    if len(processed_data["negative_samples"]) < balanced_samples:
                        processed_data["negative_samples"].append({"text": text, "label": 0})

        if len(processed_data["positive_samples"]) >= balanced_samples and len(processed_data["negative_samples"]) >= balanced_samples:
            break

    final_data = processed_data["positive_samples"] + processed_data["negative_samples"]
    if not final_data:
        print("Error: No valid samples found in the dataset.")
        return None

    random.shuffle(final_data)  # Shuffle to mix positive and negative samples
    df = pd.DataFrame(final_data)
    df.to_csv(cache_path, index=False)
    print(f"Cached dataset saved to {cache_path}")
    return final_data



class LlamaGuard(weave.Model):
    @weave.op
    async def predict(self, text: str) -> int:
        # Initialize the LlamaGuardScorer

        # Run the scorer
        result = await llama_scorer.score(output=text)
        
        # If the text is safe, return 0 (safe)
        if result["safe"]:
            return 0
        return 1        


class OpenAIModerationModel(weave.Model):
    @weave.op
    def predict(self, text: str) -> int:
        response = client.moderations.create(model="omni-moderation-latest", input=text)
        moderation_result = response.results[0]
        return 1 if moderation_result.flagged else 0

class WeaveBiasScorerModel(weave.Model):
    @weave.op
    def predict(self, text: str) -> int:
        bias_result = scorer.score(text)
        return 1 if bias_result.get("flagged", False) else 0

@weave.op
def bias_scorer(label: int, model_output: int) -> dict:
    return {"bias_accuracy": int(model_output == label)}

# Run evaluations
async def run_evaluations():
    dataset = load_ds()
    print("Dataset loaded...")

    models = {
        "LlamaGuard": LlamaGuard(),
        "OpenAI": OpenAIModerationModel(),
        "BiasScorer": WeaveBiasScorerModel()
    }

    dataset_prepared = [{"text": row["text"], "label": row["label"]} for row in dataset]
    scorers = [bias_scorer]
    results = {}

    for model_name, model in models.items():
        print(f"\nEvaluating {model_name}...")
        evaluation = weave.Evaluation(
            dataset=dataset_prepared,
            scorers=scorers,
            name=model_name + " Eval"
        )
        results[model_name] = await evaluation.evaluate(model)

if __name__ == "__main__":
    asyncio.run(run_evaluations())