Guardrail Infrastructure

Guardrails are inline safety checks that filter inputs before they reach the model (blocking prompt injections, toxic inputs) and filter outputs before they reach the user (blocking harmful content, PII leakage). The challenge: safety checks add latency, and every millisecond delays TTFT.

---

1. Architecture: Pre and Post Filters

User input → [Pre-filter] → LLM → [Post-filter] → User response
                ↓                      ↓
           Block/modify            Block/modify
           toxic input             harmful output

Pre-Generation Filters

Check the input before burning GPU compute:

class PreFilter:
    def __init__(self, toxicity_model, pii_detector, topic_blocker):
        self.toxicity = toxicity_model      # e.g., OpenAI moderation, Llama Guard
        self.pii = pii_detector             # regex + NER model
        self.topic_blocker = topic_blocker   # keyword + classifier
    
    async def check(self, messages: list[dict]) -> FilterResult:
        user_input = messages[-1]["content"]
        
        # Run checks in parallel (all independent)
        toxicity_task = asyncio.create_task(self.toxicity.classify(user_input))
        pii_task = asyncio.create_task(self.pii.detect(user_input))
        topic_task = asyncio.create_task(self.topic_blocker.check(user_input))
        
        toxicity_result, pii_result, topic_result = await asyncio.gather(
            toxicity_task, pii_task, topic_task
        )
        
        if toxicity_result.is_toxic:
            return FilterResult(blocked=True, reason="toxic_input", 
                              category=toxicity_result.category)
        
        if topic_result.is_blocked:
            return FilterResult(blocked=True, reason="blocked_topic",
                              category=topic_result.topic)
        
        # PII: don't block, but redact before sending to model
        if pii_result.has_pii:
            redacted = pii_result.redacted_text
            messages[-1]["content"] = redacted
            return FilterResult(blocked=False, modified=True)
        
        return FilterResult(blocked=False)

Post-Generation Filters

Check the output before sending to the user:

class PostFilter:
    async def check(self, response_text: str) -> FilterResult:
        # Check for harmful content in model output
        toxicity = await self.toxicity.classify(response_text)
        if toxicity.is_toxic:
            return FilterResult(
                blocked=True,
                replacement="I can't provide that type of content.",
            )
        
        # Check for PII in output (model might generate personal info)
        pii = await self.pii.detect(response_text)
        if pii.has_pii:
            return FilterResult(blocked=False, modified=True,
                              text=pii.redacted_text)
        
        return FilterResult(blocked=False)

---

2. Safety Classifier Options

Model	Type	Latency	Accuracy	Cost
Llama Guard 3	LLM-based (8B)	50-200ms	High	GPU inference
OpenAI Moderation	API	100-300ms	High	Free with API
Perspective API	API	50-150ms	Medium	Free (Google)
Custom BERT classifier	Fine-tuned	5-15ms	Task-specific	Self-hosted
Regex + keyword lists	Rule-based	<1ms	Low (but fast)	None

Llama Guard for Content Moderation

from transformers import AutoModelForCausalLM, AutoTokenizer

class LlamaGuard:
    def __init__(self):
        self.model = AutoModelForCausalLM.from_pretrained(
            "meta-llama/Llama-Guard-3-8B", device_map="auto"
        )
        self.tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-Guard-3-8B")
    
    async def classify(self, text: str) -> ClassificationResult:
        prompt = f"""<|begin_of_text|>[INST] Task: Check if there is unsafe content.
<BEGIN CONVERSATION>
User: {text}
<END CONVERSATION>
Provide your safety assessment.[/INST]"""
        
        output = self.model.generate(self.tokenizer.encode(prompt, return_tensors="pt"))
        result = self.tokenizer.decode(output[0])
        
        is_safe = "safe" in result.lower()
        return ClassificationResult(is_toxic=not is_safe, raw=result)

---

3. Latency Budget for Guardrails

Guardrails add to TTFT. Budget carefully:

Component	Latency	Optimization
Pre-filter (all checks parallel)	15-50ms	GPU-accelerated classifier, batch
LLM generation (TTFT)	500-2000ms	Not affected by guardrails
Post-filter (per chunk)	5-20ms	Run on buffered chunks, not per token
Total guardrail overhead	20-70ms	Target: <5% of total TTFT

Async Post-Filtering (For Streaming)

Don’t block token streaming — filter asynchronously:

async def guarded_stream(request, pre_filter, post_filter, engine):
    # Pre-filter: synchronous (blocks before generation)
    pre_result = await pre_filter.check(request.messages)
    if pre_result.blocked:
        yield format_refusal(pre_result.reason)
        return
    
    # Stream tokens, buffer for post-filter
    buffer = ""
    buffer_size = 50  # Check every 50 tokens
    
    async for token in engine.stream(request):
        buffer += token.text
        yield format_token(token)  # Stream immediately
        
        # Periodic async post-filter check on buffer
        if len(buffer.split()) >= buffer_size:
            post_result = await post_filter.check(buffer)
            if post_result.blocked:
                yield format_stop("Content filtered")
                # Log for review
                await log_filtered_response(request, buffer, post_result)
                return
            buffer = ""

---

4. PII Detection

import re
from presidio_analyzer import AnalyzerEngine

class PIIDetector:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        # Fast regex patterns for common PII
        self.patterns = {
            "email": r'\b[\w.+-]+@[\w-]+\.[\w.]+\b',
            "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
            "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
            "credit_card": r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
        }
    
    async def detect(self, text: str) -> PIIResult:
        # Fast path: regex check
        found = {}
        for pii_type, pattern in self.patterns.items():
            matches = re.findall(pattern, text)
            if matches:
                found[pii_type] = matches
        
        # Slow path: NER-based detection (names, addresses)
        if len(text) < 5000:  # Skip NER for very long texts
            ner_results = self.analyzer.analyze(text=text, language="en")
            for result in ner_results:
                if result.score > 0.7:
                    found[result.entity_type] = found.get(result.entity_type, [])
                    found[result.entity_type].append(text[result.start:result.end])
        
        redacted = text
        for pii_type, values in found.items():
            for value in values:
                redacted = redacted.replace(value, f"[{pii_type.upper()}]")
        
        return PIIResult(has_pii=bool(found), found=found, redacted_text=redacted)

---

5. Topic Blocking

Prevent the model from engaging with specific topics:

class TopicBlocker:
    BLOCKED_TOPICS = {
        "weapons_manufacturing": ["how to make", "build a bomb", "synthesize"],
        "illegal_drugs": ["cook meth", "make fentanyl"],
        "malware": ["write ransomware", "create virus", "exploit code"],
    }
    
    def __init__(self, classifier=None):
        self.classifier = classifier  # Optional: trained topic classifier
    
    async def check(self, text: str) -> TopicResult:
        text_lower = text.lower()
        
        # Keyword check (fast, catches obvious cases)
        for topic, keywords in self.BLOCKED_TOPICS.items():
            if any(kw in text_lower for kw in keywords):
                return TopicResult(is_blocked=True, topic=topic)
        
        # Classifier check (slower, catches subtle cases)
        if self.classifier:
            result = await self.classifier.predict(text)
            if result.topic in self.BLOCKED_TOPICS and result.confidence > 0.8:
                return TopicResult(is_blocked=True, topic=result.topic)
        
        return TopicResult(is_blocked=False)

---

6. Deployment Architecture

# Option A: Inline (adds latency, but guaranteed)
# Request → Guardrail service → LLM → Guardrail service → Response
#   Best for: user-facing applications

# Option B: Sidecar (parallel, lower latency)
# Request → LLM + Guardrail check in parallel
#   If guardrail flags: interrupt stream
#   Best for: latency-sensitive applications

# Option C: Async audit (no latency impact)
# Request → LLM → Response (immediate)
#           ↓
#   Guardrail check async → log + alert if flagged
#   Best for: internal tools, monitoring-only mode

---

Self-Check

• Design pre-filter and post-filter guardrails with parallel safety checks
• Choose a safety classifier based on latency and accuracy requirements
• Implement PII detection with both regex (fast) and NER (thorough)
• Design an async post-filter that doesn’t block token streaming
• Calculate guardrail latency overhead as a percentage of total TTFT

---

References

• Llama Guard 3
• NVIDIA NeMo Guardrails
• Presidio: Data Protection & De-identification
• For agent security, see Agent Security