LLM Response Filtering: Content Safety

Your Self-Hosted LLM Produces Unsafe Responses

Cloud LLM APIs include built-in content filtering, but self-hosted models on your GPU server have no such guardrails by default. An open-source model will happily generate harmful content, leak personal data from its training set, or produce legally problematic output if no filtering layer sits between the model and your users. Content safety is your responsibility when you self-host.

Layered Filtering Architecture

Apply multiple independent filters in sequence — each catches what others miss:

# Filter pipeline: Input Check -> Model -> Output Check -> Delivery

class SafetyPipeline:
    def __init__(self):
        self.input_filters = [KeywordInputFilter(), PromptInjectionFilter()]
        self.output_filters = [
            KeywordOutputFilter(),
            PIIRedactionFilter(),
            ToxicityClassifier(),
        ]

    async def process(self, messages):
        # Stage 1: Filter input
        for f in self.input_filters:
            messages = f.check(messages)
            if messages is None:
                return {"error": "Request blocked by content policy"}

        # Stage 2: Generate response
        response = await call_llm(messages)

        # Stage 3: Filter output
        text = response["choices"][0]["message"]["content"]
        for f in self.output_filters:
            text, flagged = f.filter(text)
            if flagged:
                response["_content_warning"] = True

        response["choices"][0]["message"]["content"] = text
        return response

Keyword and Pattern Filters

Fast first-pass filtering catches obvious policy violations:

import re

class KeywordOutputFilter:
    def __init__(self):
        # Load blocked patterns from configuration
        self.blocked_patterns = [
            re.compile(r'\b(bomb|weapon|explosive)\s+(making|construction|build)',
                       re.IGNORECASE),
            re.compile(r'(hack|crack|exploit)\s+(password|account|system)',
                       re.IGNORECASE),
        ]
        self.replacement = "[Content removed by safety filter]"

    def filter(self, text):
        flagged = False
        for pattern in self.blocked_patterns:
            if pattern.search(text):
                text = pattern.sub(self.replacement, text)
                flagged = True
        return text, flagged

Classifier-Based Moderation

Use a small classification model to detect subtle policy violations that keyword filters miss:

from transformers import pipeline

# Load a toxicity classifier (runs on CPU to keep GPU free for LLM)
toxicity_classifier = pipeline(
    "text-classification",
    model="unitary/toxic-bert",
    device=-1  # CPU
)

class ToxicityClassifier:
    def __init__(self, threshold=0.7):
        self.threshold = threshold

    def filter(self, text):
        # Check each paragraph independently
        paragraphs = text.split("\n\n")
        filtered_paragraphs = []
        flagged = False

        for para in paragraphs:
            if not para.strip():
                filtered_paragraphs.append(para)
                continue

            result = toxicity_classifier(para[:512])[0]
            if result["label"] == "toxic" and result["score"] > self.threshold:
                filtered_paragraphs.append(
                    "[This paragraph was removed by content moderation]")
                flagged = True
            else:
                filtered_paragraphs.append(para)

        return "\n\n".join(filtered_paragraphs), flagged

PII Redaction

Prevent the model from leaking personal information it memorised from training data:

import re

class PIIRedactionFilter:
    def __init__(self):
        self.patterns = {
            "email": (re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
                      "[EMAIL REDACTED]"),
            "phone_uk": (re.compile(r'\b(?:0|\+44)\s?\d{4}\s?\d{6}\b'),
                         "[PHONE REDACTED]"),
            "nino": (re.compile(r'\b[A-CEGHJ-PR-TW-Z]{2}\s?\d{2}\s?\d{2}\s?\d{2}\s?[A-D]\b'),
                     "[NINO REDACTED]"),
            "card": (re.compile(r'\b(?:\d{4}[\s-]?){3}\d{4}\b'),
                     "[CARD NUMBER REDACTED]"),
            "postcode": (re.compile(r'\b[A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2}\b', re.I),
                         "[POSTCODE REDACTED]"),
        }

    def filter(self, text):
        flagged = False
        for name, (pattern, replacement) in self.patterns.items():
            if pattern.search(text):
                text = pattern.sub(replacement, text)
                flagged = True
        return text, flagged

Integrating with Guardrails Libraries

Use established frameworks for comprehensive content safety:

# NeMo Guardrails integration
pip install nemoguardrails

# guardrails_config.yml
models:
  - type: main
    engine: openai
    parameters:
      base_url: http://localhost:8000/v1
      model: meta-llama/Meta-Llama-3.1-8B-Instruct

rails:
  input:
    flows:
      - check topic allowed
      - check jailbreak attempt
  output:
    flows:
      - check response safety
      - redact personal information
      - check factual accuracy

# Python usage
from nemoguardrails import RailsConfig, LLMRails

config = RailsConfig.from_path("./guardrails_config")
rails = LLMRails(config)

response = await rails.generate_async(
    messages=[{"role": "user", "content": user_input}]
)

Safety Audit Logging

import json, time

def log_safety_event(request_id, filter_name, action, original, filtered):
    event = {
        "timestamp": time.time(),
        "request_id": request_id,
        "filter": filter_name,
        "action": action,  # "blocked", "redacted", "flagged"
        "original_length": len(original),
        "filtered_length": len(filtered),
    }
    with open("/var/log/llm_safety.jsonl", "a") as f:
        f.write(json.dumps(event) + "\n")

Content safety is essential for any production LLM on your GPU server. For vLLM and Ollama deployments, add the safety pipeline as a proxy layer. The vLLM production guide covers proxy architecture. See the LLM hosting section for best practices, tutorials for deployment, and infrastructure for compliance considerations.