Toxicity and Safety Filters for LLM Outputs: A Hands-On Guide

LLMs can generate harmful content — not because they’re malicious, but because they’re trained on internet data that includes it. Toxicity filters and safety classifiers are your last line of defense, catching harmful outputs before they reach users. This guide covers practical implementation of production-ready safety filters using open-source tools and APIs.

The Safety Stack

User Input
    ↓
┌─────────────────┐
│ Input Sanitizer  │ ← Catch prompt injection, jailbreaks
└────────┬────────┘
         ↓
┌─────────────────┐
│ LLM Inference    │
└────────┬────────┘
         ↓
┌─────────────────┐
│ Output Classifier│ ← Classify toxicity, harm categories
└────────┬────────┘
         ↓
┌─────────────────┐
│ Output Filter    │ ← Block, modify, or escalate
└────────┬────────┘
         ↓
Safe Output to User

Implementing Toxicity Classification

Using Hugging Face Transformers

The simplest approach: use a pre-trained toxicity classifier.

from transformers import pipeline
from typing import NamedTuple

class ToxicityResult(NamedTuple):
    label: str
    score: float
    is_toxic: bool

class ToxicityFilter:
    def __init__(self, threshold: float = 0.8):
        self.threshold = threshold
        self.classifier = pipeline(
            "text-classification",
            model="unitary/toxic-bert",
            return_all_scores=True
        )

    def classify(self, text: str) -> list[ToxicityResult]:
        """Classify text for toxicity."""
        results = self.classifier(text)[0]
        return [
            ToxicityResult(
                label=r["label"],
                score=r["score"],
                is_toxic=r["score"] > self.threshold and r["label"] != "toxic"
            )
            for r in results
        ]

    def is_safe(self, text: str) -> bool:
        """Check if text passes safety filters."""
        results = self.classify(text)
        return not any(r.is_toxic for r in results)

    def filter(self, text: str) -> str:
        """Filter or block toxic content."""
        if self.is_safe(text):
            return text
        return "[Content filtered for safety]"

Multi-Category Classification

For production systems, classify across multiple harm categories:

class MultiCategoryFilter:
    CATEGORIES = {
        "hate": "toxicity/hate-speech",
        "harassment": "toxicity/harassment",
        "violence": "toxicity/violence",
        "self-harm": "toxicity/self-harm",
        "sexual": "toxicity/sexual",
        "spam": "quality/spam",
    }

    def __init__(self):
        self.filters = {
            cat: pipeline("text-classification", model=model)
            for cat, model in self.CATEGORIES.items()
        }

    def classify(self, text: str) -> dict[str, float]:
        """Classify text across all categories."""
        scores = {}
        for category, classifier in self.filters.items():
            result = classifier(text)[0]
            scores[category] = result["score"]
        return scores

    def filter(self, text: str, policies: dict[str, float] = None) -> dict:
        """Apply category-specific policies."""
        policies = policies or {
            "hate": 0.7,
            "harassment": 0.7,
            "violence": 0.8,
            "self-harm": 0.6,
            "sexual": 0.7,
            "spam": 0.9,
        }

        scores = self.classify(text)
        violations = {
            cat: score
            for cat, score in scores.items()
            if score > policies.get(cat, 0.8)
        }

        return {
            "safe": len(violations) == 0,
            "scores": scores,
            "violations": violations,
            "action": self._determine_action(violations),
        }

    def _determine_action(self, violations: dict) -> str:
        if not violations:
            return "allow"
        elif any(v > 0.9 for v in violations.values()):
            return "block"
        else:
            return "flag"

Using Moderation APIs

OpenAI Moderation API

import openai

class OpenAIModerationFilter:
    def __init__(self):
        self.client = openai.OpenAI()

    async def check(self, text: str) -> dict:
        response = await self.client.moderations.create(
            input=text
        )
        result = response.results[0]

        return {
            "safe": not result.flagged,
            "categories": {
                cat: getattr(result.categories, cat)
                for cat in result.categories.__fields__
            },
            "scores": {
                cat: getattr(result.category_scores, cat)
                for cat in result.category_scores.__fields__
            },
        }

Custom Moderation Pipeline

Combine multiple filters for comprehensive coverage:

class ModerationPipeline:
    def __init__(self):
        self.filters = [
            ToxicityFilter(threshold=0.8),
            OpenAIModerationFilter(),
            KeywordBlocker(keywords=["forbidden", "blocked"]),
        ]

    async def check(self, text: str) -> dict:
        results = []
        for filter in self.filters:
            result = await filter.check(text)
            results.append(result)

        # Aggregate results
        all_safe = all(r.get("safe", True) for r in results)

        return {
            "safe": all_safe,
            "details": results,
            "action": "allow" if all_safe else "block",
        }

Output Sanitization

Sometimes you don’t want to block content entirely — just clean it up:

class OutputSanitizer:
    def __init__(self):
        self.pii_patterns = {
            "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
            "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
        }

    def sanitize(self, text: str, rules: list[str] = None) -> str:
        """Apply sanitization rules."""
        rules = rules or ["pii", "profanity", "links"]

        if "pii" in rules:
            text = self._mask_pii(text)
        if "profanity" in rules:
            text = self._mask_profanity(text)
        if "links" in rules:
            text = self._validate_links(text)

        return text

    def _mask_pii(self, text: str) -> str:
        import re
        for pii_type, pattern in self.pii_patterns.items():
            text = re.sub(pattern, f"[{pii_type.upper()}]", text)
        return text

    def _mask_profanity(self, text: str) -> str:
        # Use a profanity filter library
        from better_profanity import profanity
        return profanity.censor(text)

    def _validate_links(self, text: str) -> str:
        import re
        urls = re.findall(r'https?://[^\s]+', text)
        for url in urls:
            if not self._is_safe_url(url):
                text = text.replace(url, "[BLOCKED URL]")
        return text

Production Patterns

Pattern 1: Asynchronous Filtering

Run filters in parallel to minimize latency:

import asyncio

class AsyncModerationPipeline:
    async def check(self, text: str) -> dict:
        # Run all filters concurrently
        tasks = [
            self.toxicity_filter.check(text),
            self.moderation_api.check(text),
            self.keyword_filter.check(text),
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Handle failures gracefully
        safe_results = [r for r in results if not isinstance(r, Exception)]

        return self._aggregate(safe_results)

Pattern 2: Graduated Response

Apply different actions based on severity:

class GraduatedResponse:
    def respond(self, check_result: dict) -> dict:
        if check_result["action"] == "allow":
            return {"output": check_result["text"], "status": "ok"}

        elif check_result["action"] == "flag":
            # Allow but log for review
            self.log_for_review(check_result)
            return {"output": check_result["text"], "status": "flagged"}

        elif check_result["action"] == "modify":
            # Sanitize and return
            sanitized = self.sanitizer.sanitize(check_result["text"])
            return {"output": sanitized, "status": "modified"}

        else:  # block
            return {
                "output": "I can't help with that request.",
                "status": "blocked",
            }

Pattern 3: Feedback Loop

Use user feedback to improve filters:

class FeedbackDrivenFilter:
    def __init__(self):
        self.feedback_store = []

    async def check_with_feedback(self, text: str) -> dict:
        result = await self.filter.check(text)

        # Track false positives/negatives
        if result["flagged"]:
            user_feedback = await self.get_user_feedback(text)
            self.feedback_store.append({
                "text": text,
                "prediction": result,
                "actual": user_feedback,
            })

            # Retrain periodically
            if len(self.feedback_store) % 100 == 0:
                await self.retrain()

        return result

Best Practices

1. Layer your defenses — Use multiple filters (toxicity classifier, moderation API, keyword blocklist) for defense in depth.

2. Set appropriate thresholds — Too strict and you block benign content; too loose and harmful content slips through. Tune thresholds using evaluation datasets.

3. Log blocked content — Track what gets filtered for periodic review. This helps identify false positives and emerging patterns.

4. Handle edge cases — Code, creative writing, and medical content often trigger false positives. Consider context-aware filtering.

5. Plan for adversarial inputs — Attackers will try to bypass your filters. Stay updated on new attack techniques and test your defenses regularly.

Conclusion

Toxicity and safety filters are essential for responsible LLM deployment. By combining pre-trained classifiers, moderation APIs, and custom sanitization rules, you can catch harmful content before it reaches users. The key is layering defenses, tuning thresholds carefully, and continuously improving based on real-world feedback.

References:

• Hugging Face Toxicity Detection
• OpenAI Moderation API