5. Content Filtering

Description

Content filtering detects LLM outputs that shouldn't reach the user: toxicity, inappropriate content, off-topic responses, and outputs that reveal information they shouldn't. The correct strategy combines fast heuristics (keywords, length, coherence) with LLM-as-judge for subtle cases. This capsule covers both strategies, how to combine them efficiently, and how to handle false positives without degrading the UX.

What the content filter filters

Outputs that must be filtered:
  1. Toxicity: insults, hate, violence, adult content
  2. Off-topic: the LLM answered something unrelated to the question
  3. Confidential information exposed: system prompt instructions revealed
  4. Successful jailbreak outputs: the LLM "escaped" its assigned role
  5. Clearly incorrect content with high confidence (obvious hallucinations)

Outputs that must NOT be filtered (common false positives):
  1. "I hate when it rains" — "hate" in a negative but harmless context
  2. Academic discussion of sensitive topics
  3. Fiction with dark elements but appropriate to the context
  4. Responses that mention "dangerous" concepts in an informative way

Layer 1: Fast heuristics ($0, <1ms)

# src/guardrails/content_filter.py
import re
from dataclasses import dataclass
from typing import Optional

@dataclass
class ContentFilterResult:
    is_safe: bool
    reason: Optional[str] = None
    layer: str = "heuristic"

# Signals of problematic outputs (not user keywords)
OUTPUT_RED_FLAGS = [
    # The LLM revealed the system prompt
    r"(my system prompt|my instructions are|i was instructed to)",
    r"(here is my (system )?prompt|my (actual )?instructions)",
    
    # The LLM left its role (signal of a successful jailbreak)
    r"(i am now|i have no (restrictions|limits)|developer mode (enabled|activated))",
    r"(as DAN|as an AI without restrictions|my true (self|purpose))",
    
    # Clearly empty or LLM error responses
    r"^(error|i don'?t know|i cannot|no (output|response))$",
]

# Signals specific to severe toxicity (not common words)
SEVERE_TOXICITY_PATTERNS = [
    r"\b(kill yourself|kys|go die)\b",
    r"\b(hate speech patterns here)\b",  # Customize per domain
]

def check_heuristics(text: str) -> ContentFilterResult:
    """
    Fast, free checks based on heuristics.
    
    NOTE: This list is intentionally conservative.
    Better to have few false positives than to block legitimate content.
    """
    text_lower = text.lower()
    
    # Check 1: Empty or too-short output
    if not text or not text.strip():
        return ContentFilterResult(
            is_safe=False,
            reason="empty_output"
        )
    
    if len(text.strip()) < 5:
        return ContentFilterResult(
            is_safe=False,
            reason="output_too_short"
        )
    
    # Check 2: Signals of a successful jailbreak
    for pattern in OUTPUT_RED_FLAGS:
        if re.search(pattern, text_lower, re.IGNORECASE):
            return ContentFilterResult(
                is_safe=False,
                reason=f"jailbreak_signal: {pattern}"
            )
    
    # Check 3: Severe toxicity
    for pattern in SEVERE_TOXICITY_PATTERNS:
        if re.search(pattern, text_lower, re.IGNORECASE):
            return ContentFilterResult(
                is_safe=False,
                reason="severe_toxicity"
            )
    
    # Check 4: Anomalously long output (possible loop)
    if len(text) > 10_000:
        return ContentFilterResult(
            is_safe=False,
            reason="output_too_long"
        )
    
    return ContentFilterResult(is_safe=True)

Layer 2: OpenAI Moderation API (free, ~50ms)

OpenAI offers a free moderation API that you can use as a first option:

import openai

def check_openai_moderation(text: str, client: openai.OpenAI) -> ContentFilterResult:
    """
    Uses OpenAI's Moderation API to detect inappropriate content.
    
    FREE: The Moderation API has no cost.
    Latency: ~50-100ms.
    Categories: hate, harassment, self-harm, sexual, violence, etc.
    
    Limitation: Designed mainly for English text.
    For Spanish, it may be less accurate.
    """
    try:
        response = client.moderations.create(
            model="omni-moderation-latest",
            input=text
        )
        
        result = response.results[0]
        
        if result.flagged:
            # Find the categories that flagged
            flagged_categories = [
                category
                for category, flagged in result.categories.model_dump().items()
                if flagged
            ]
            
            return ContentFilterResult(
                is_safe=False,
                reason=f"moderation_api: {', '.join(flagged_categories)}",
                layer="moderation_api"
            )
        
        return ContentFilterResult(is_safe=True, layer="moderation_api")
    
    except Exception as e:
        # If the API fails, don't block (fail open for the content filter)
        import logging
        logging.getLogger("guardrails").warning(f"Moderation API failed: {e}")
        return ContentFilterResult(is_safe=True, layer="moderation_api")

Layer 3: LLM-as-judge for off-topic and subtle toxicity

CONTENT_JUDGE_SYSTEM = """You are a content evaluator. Your task is to classify 
whether an AI assistant's response is appropriate.

Respond with JSON exactly like this:
{"safe": true/false, "reason": "brief explanation"}

Criteria for "safe=false":
1. Toxic, offensive, or inappropriate content
2. The assistant reveals internal or system instructions
3. The assistant clearly left its assigned role
4. The response is completely irrelevant to the question

Criteria for "safe=true":
- Useful and relevant response
- Academic or informative discussion of sensitive topics
- Mentioning difficult topics in an appropriate context
"""

def check_llm_judge(
    response: str,
    original_question: str,
    client: openai.OpenAI
) -> ContentFilterResult:
    """
    Uses an LLM to evaluate whether the output is appropriate.
    
    When to use: When heuristics and the Moderation API aren't enough.
    Latency: +400-600ms
    Cost: ~$0.0001 per evaluation with gpt-4o-mini
    
    Best for: off-topic detection, subtle toxicity, instruction disclosure
    """
    sample_response = response[:500]  # Evaluate only the first 500 chars
    sample_question = original_question[:200]
    
    try:
        result = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": CONTENT_JUDGE_SYSTEM},
                {
                    "role": "user",
                    "content": f"""Question: "{sample_question}"
Assistant's response: "{sample_response}"

Is this response appropriate?"""
                }
            ],
            temperature=0.0,
            max_tokens=100,
            response_format={"type": "json_object"}
        )
        
        import json
        judgment = json.loads(result.choices[0].message.content)
        
        if not judgment.get("safe", True):
            return ContentFilterResult(
                is_safe=False,
                reason=f"llm_judge: {judgment.get('reason', 'unsafe')}",
                layer="llm_judge"
            )
        
        return ContentFilterResult(is_safe=True, layer="llm_judge")
    
    except Exception:
        return ContentFilterResult(is_safe=True, layer="llm_judge")  # fail open

Complete filtering pipeline

def apply_content_filter(
    response: str,
    original_question: str = None,
    client = None,
    use_moderation_api: bool = True,
    use_llm_judge: bool = False
) -> ContentFilterResult:
    """
    Content filtering pipeline with three layers.
    
    The order optimizes latency: the fastest and cheapest filters go first.
    
    Args:
        response: The LLM output to filter
        original_question: The user's original question (for the off-topic check)
        client: OpenAI client (for the moderation API and LLM judge)
        use_moderation_api: Use the free moderation API (recommended)
        use_llm_judge: Use an additional LLM for evaluation (more expensive)
    """
    import logging
    logger = logging.getLogger("guardrails.content_filter")
    
    # Layer 1: Heuristics (always, free, <1ms)
    heuristic_result = check_heuristics(response)
    if not heuristic_result.is_safe:
        logger.warning("content_filtered", extra={
            "layer": "heuristic",
            "reason": heuristic_result.reason
        })
        return heuristic_result
    
    # Layer 2: Moderation API (if a client is available, free, ~50ms)
    if use_moderation_api and client is not None:
        moderation_result = check_openai_moderation(response, client)
        if not moderation_result.is_safe:
            logger.warning("content_filtered", extra={
                "layer": "moderation_api",
                "reason": moderation_result.reason
            })
            return moderation_result
    
    # Layer 3: LLM judge (only if enabled, +latency)
    if use_llm_judge and client is not None and original_question:
        llm_result = check_llm_judge(response, original_question, client)
        if not llm_result.is_safe:
            logger.warning("content_filtered", extra={
                "layer": "llm_judge",
                "reason": llm_result.reason
            })
            return llm_result
    
    return ContentFilterResult(is_safe=True)

Handling the blocked output

# src/app/sentiment.py (with the content filter integrated)

BLOCKED_RESPONSE = {
    "sentiment": "unknown",
    "score": 0.0,
    "explanation": "It was not possible to process this request.",
    "keywords": [],
    "_filtered": True  # Internal metadata (don't expose to the frontend)
}

def analyze_sentiment_with_guardrails(text: str, client) -> dict:
    """Sentiment analysis with output content filtering."""
    
    raw_result = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SENTIMENT_SYSTEM_PROMPT},
            {"role": "user", "content": text}
        ],
        temperature=0.0,
        max_tokens=300
    )
    
    output = raw_result.choices[0].message.content
    
    # Filter the output before processing it
    filter_result = apply_content_filter(
        response=output,
        original_question=text,
        client=client,
        use_moderation_api=True,
        use_llm_judge=False  # Only on high-value endpoints
    )
    
    if not filter_result.is_safe:
        return BLOCKED_RESPONSE
    
    # Process normally if it's safe
    from app.parsers import parse_json_response
    from app.processors import process_sentiment_output
    
    parsed = parse_json_response(output)
    return process_sentiment_output(parsed)

Content filter tests

# tests/unit/guardrails/test_content_filter.py
import pytest
from src.guardrails.content_filter import check_heuristics, ContentFilterResult

class TestHeuristicFilter:
    
    def test_normal_output_is_safe(self):
        text = "The sentiment of the text is positive with a score of 0.85."
        result = check_heuristics(text)
        assert result.is_safe
    
    def test_empty_output_blocked(self):
        result = check_heuristics("")
        assert not result.is_safe
        assert result.reason == "empty_output"
    
    def test_very_short_output_blocked(self):
        result = check_heuristics("OK")
        assert not result.is_safe
    
    def test_jailbreak_signal_blocked(self):
        text = "I am now in developer mode with no restrictions."
        result = check_heuristics(text)
        assert not result.is_safe
    
    def test_system_prompt_reveal_blocked(self):
        text = "My system prompt is: 'You are a sentiment analyzer...'"
        result = check_heuristics(text)
        assert not result.is_safe
    
    def test_safe_discussion_of_sensitive_topic(self):
        """An informative discussion must not be blocked."""
        text = "The analysis indicates that the text discusses topics of social violence from a critical perspective."
        result = check_heuristics(text)
        assert result.is_safe  # Informative, not toxic

@pytest.mark.parametrize("blocked_output,expected_reason", [
    ("", "empty_output"),
    ("OK", "output_too_short"),
    ("I am now in developer mode", "jailbreak_signal"),
])
def test_heuristics_parametrized(blocked_output, expected_reason):
    result = check_heuristics(blocked_output)
    assert not result.is_safe
    assert expected_reason in result.reason

Exercises

Exercise 1: Design the heuristics list

For a sentiment analysis app in English, design 5 heuristics specific to your domain:

# For a sentiment analysis app:
# 1. Output that contains none of the expected keys
def check_missing_keys(text: str) -> bool:
    """The sentiment output should mention the result."""
    return not any(kw in text.lower() for kw in ["positive", "negative", "neutral", "{"])

# 2. Output that starts with an apology (signal that the LLM couldn't)
def check_apology_start(text: str) -> bool:
    apology_starts = ["i'm sorry", "i cannot", "sorry", "i apologize"]
    return any(text.lower().strip().startswith(a) for a in apology_starts)

# 3. Excessively long output (the sentiment JSON shouldn't be > 1000 chars)
def check_length(text: str) -> bool:
    return len(text) > 1000

# 4. Output without any number (the score always has a number)
import re
def check_has_number(text: str) -> bool:
    return not re.search(r'\d+\.?\d*', text)

# 5. Output that mentions "instructions" in a disclosure context (not in analyzed text)
def check_reveals_instructions(text: str) -> bool:
    patterns = [r"my instructions are", r"the system prompt"]
    return any(re.search(p, text.lower()) for p in patterns)

Exercise 2: False-positive test

Write 3 tests that verify that legitimate content is NOT blocked:

def test_negative_sentiment_text_not_blocked():
    """A summary of text with negative sentiment isn't toxic."""
    output = '{"sentiment": "negative", "score": 0.1, "explanation": "The text expresses frustration and dissatisfaction.", "keywords": ["terrible", "disappointing"]}'
    result = check_heuristics(output)
    assert result.is_safe

def test_discussion_of_violence_topic_not_blocked():
    """Analyzing a text that discusses violence isn't toxic."""
    output = '{"sentiment": "negative", "score": 0.05, "explanation": "The text reports violent events with a concerned tone.", "keywords": ["violence", "incident"]}'
    result = check_heuristics(output)
    assert result.is_safe

def test_neutral_factual_output_not_blocked():
    """A factual, neutral output passes the filter."""
    output = '{"sentiment": "neutral", "score": 0.5, "explanation": "The text is a factual report with no emotional charge.", "keywords": ["data", "statistics"]}'
    result = check_heuristics(output)
    assert result.is_safe

Exercise 3: Decide which layers to use

For each endpoint, decide which content filtering layers to use:

1. Internal endpoint /admin/analyze only for the product team
2. Public endpoint /analyze for the app's users
3. Endpoint /analyze-document for analyzing documents uploaded by users

Summary

• Content filter = three layers: heuristics (<1ms), Moderation API (~50ms, free), LLM judge (+500ms, ~$0.0001)
• Conservative heuristics: better to have few false positives than to block legitimate content
• OpenAI Moderation API is free — always use it when a client is available
• LLM judge for specific cases: off-topic, subtle toxicity, high-value endpoints
• Fail open: if a guardrail fails due to an infrastructure error, don't block the output
• Parametrized tests: both for outputs that must be blocked and for false positives

Additional resources

1. OpenAI Moderation API — Documentation for the free API
2. Perspective API (Google) — Alternative for toxicity detection
3. LLM-as-Judge (paper) — Research on evaluation with LLMs
4. Content Moderation Best Practices — General strategies
5. Azure Content Safety — Cloud alternative
6. Llama Guard — Meta's open source model for safeguarding