LLM Guardrails Pipeline: Input & Output Safety Architecture

Introduction

Deploying an LLM in production without guardrails is the AI equivalent of running a web server without input validation. Users will probe edge cases, attempt prompt injection, and occasionally receive hallucinated or policy-violating outputs. Guardrails are the safety and policy enforcement layer that sits both before and after the core LLM, creating a defense-in-depth architecture similar to application security middleware.

---

Full Guardrails Pipeline Architecture

                        User Input
                            │
                            ▼
┌───────────────────────────────────────────────────────────┐
│                    INPUT GUARDRAIL LAYER                   │
│                                                           │
│  ┌─────────────────────────────────────────────────────┐  │
│  │ 1. PROMPT INJECTION DETECTION                       │  │
│  │    • Detect: "Ignore previous instructions..."      │  │
│  │    • Detect: role-play attempts, delimiter attacks  │  │
│  │    • Method: classifier + heuristic regex          │  │
│  └─────────────────────────────────────────────────────┘  │
│                          │ PASS                           │
│  ┌─────────────────────────────────────────────────────┐  │
│  │ 2. TOXICITY & HATE SPEECH CLASSIFICATION           │  │
│  │    • Model: Perspective API / Detoxify / LlamaGuard│  │
│  │    • Threshold: TOXICITY > 0.8 → BLOCK             │  │
│  └─────────────────────────────────────────────────────┘  │
│                          │ PASS                           │
│  ┌─────────────────────────────────────────────────────┐  │
│  │ 3. TOPIC CLASSIFICATION (Policy Scope)              │  │
│  │    • Is query in-scope for this assistant?          │  │
│  │    • Block: competitor mentions, legal advice, etc. │  │
│  │    • Method: embedding similarity to blocked topics │  │
│  └─────────────────────────────────────────────────────┘  │
│                          │ PASS                           │
│  ┌─────────────────────────────────────────────────────┐  │
│  │ 4. PII DETECTION IN INPUT                           │  │
│  │    • Detect SSNs, credit cards, API keys in input   │  │
│  │    • Action: redact before sending to LLM API       │  │
│  │    • Method: Presidio, SpaCy NER, regex patterns    │  │
│  └─────────────────────────────────────────────────────┘  │
│                          │ SANITIZED INPUT                │
└───────────────────────────────────────────────────────────┘
                            │
                            ▼
┌───────────────────────────────────────────────────────────┐
│                      CORE LLM                             │
│           (GPT-4o / Claude / Gemini / etc.)               │
│                                                           │
│   System Prompt includes:                                 │
│   • Role definition & persona                             │
│   • Topic restrictions ("Do not discuss competitors")     │
│   • Output format constraints (JSON, no markdown)         │
│   • Refusal instructions                                   │
└───────────────────────────────────────────────────────────┘
                            │
                            ▼
┌───────────────────────────────────────────────────────────┐
│                   OUTPUT GUARDRAIL LAYER                   │
│                                                           │
│  ┌─────────────────────────────────────────────────────┐  │
│  │ 5. PII DETECTION IN OUTPUT                          │  │
│  │    • Scan model response for SSN, CC#, PHI, etc.   │  │
│  │    • Action: redact → [REDACTED] or block entirely  │  │
│  └─────────────────────────────────────────────────────┘  │
│                          │                               │
│  ┌─────────────────────────────────────────────────────┐  │
│  │ 6. HALLUCINATION / GROUNDEDNESS CHECK               │  │
│  │    • For RAG: verify each claim against source docs │  │
│  │    • Method: NLI (Natural Language Inference)       │  │
│  │    • Model: cross-encoder/nli-deberta-v3-base       │  │
│  │    • Action: flag ungrounded sentences, add caveat  │  │
│  └─────────────────────────────────────────────────────┘  │
│                          │                               │
│  ┌─────────────────────────────────────────────────────┐  │
│  │ 7. POLICY COMPLIANCE CHECK                          │  │
│  │    • Did output refuse a jailbreak correctly?       │  │
│  │    • Does output comply with brand/legal policy?    │  │
│  │    • Method: secondary LLM-as-judge evaluation     │  │
│  └─────────────────────────────────────────────────────┘  │
│                          │ CLEAN OUTPUT                  │
└───────────────────────────────────────────────────────────┘
                            │
                            ▼
                        User Response

---

Implementation: Complete Guardrail Stack

flowchart TD
    A["1. Input Guardrail"]
    B["2. Input Guardrail"]
    C["3. Output Guardrail"]
    D["4. Full Pipeline Orchestrator"]
    A --> B
    B --> C
    C --> D

Step 1: Input Guardrail — PII Redaction with Presidio

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def redact_pii(text: str) -> tuple[str, list]:
    """Detect and redact PII before sending to LLM."""
    results = analyzer.analyze(
        text=text,
        entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "CREDIT_CARD",
                  "US_SSN", "US_BANK_NUMBER", "IP_ADDRESS", "LOCATION"],
        language="en"
    )
    
    anonymized = anonymizer.anonymize(
        text=text,
        analyzer_results=results,
        operators={
            "PERSON":        OperatorConfig("replace", {"new_value": "[NAME]"}),
            "CREDIT_CARD":   OperatorConfig("replace", {"new_value": "[CC_REDACTED]"}),
            "US_SSN":        OperatorConfig("replace", {"new_value": "[SSN_REDACTED]"}),
            "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "[EMAIL]"}),
            "DEFAULT":       OperatorConfig("replace", {"new_value": "[REDACTED]"}),
        }
    )
    return anonymized.text, results  # return redacted text + what was found

Step 2: Input Guardrail — Prompt Injection Detection

import re

INJECTION_PATTERNS = [
    r"ignore (previous|all|prior) (instructions|prompt|system)",
    r"disregard (your|the|all) (previous|prior|system|original)",
    r"(you are now|act as|pretend (you are|to be))",
    r"(do not|don't) (follow|obey|adhere to) (your|the) (guidelines|rules|instructions)",
    r"DAN|jailbreak|[Ii]gnore (constraints|restrictions|rules)",
    r"new (persona|role|identity|character)",
]

def detect_prompt_injection(text: str) -> bool:
    text_lower = text.lower()
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, text_lower):
            return True
    return False

# Using LlamaGuard for LLM-based safety classification
from openai import OpenAI

def llamaguard_classify(text: str) -> dict:
    """Use Meta's LlamaGuard model for safety classification."""
    client = OpenAI(base_url="https://api.together.xyz/v1")  # or any LlamaGuard host
    
    response = client.chat.completions.create(
        model="meta-llama/LlamaGuard-2-8b",
        messages=[{"role": "user", "content": text}]
    )
    
    result = response.choices[0].message.content.strip()
    # LlamaGuard returns "safe" or "unsafe\nViolation_Type"
    is_safe = result.lower().startswith("safe")
    violation = result.split("\n")[1] if not is_safe else None
    return {"safe": is_safe, "violation": violation}

Step 3: Output Guardrail — Hallucination Detection via NLI

from transformers import pipeline

# Cross-encoder NLI model for groundedness checking
nli_pipeline = pipeline(
    "text-classification",
    model="cross-encoder/nli-deberta-v3-base",
    device=0  # GPU
)

def check_groundedness(claim: str, source_documents: list[str]) -> dict:
    """Verify if a claim is supported by source documents using NLI."""
    context = " ".join(source_documents[:3])  # top 3 source docs
    
    result = nli_pipeline(
        f"{context} [SEP] {claim}",
        candidate_labels=["entailment", "neutral", "contradiction"]
    )
    
    entailment_score  = next(r["score"] for r in result if r["label"] == "ENTAILMENT")
    contradiction_score = next(r["score"] for r in result if r["label"] == "CONTRADICTION")
    
    grounded = entailment_score > 0.7
    hallucinated = contradiction_score > 0.5
    
    return {
        "grounded": grounded,
        "hallucinated": hallucinated,
        "entailment_score": entailment_score,
        "contradiction_score": contradiction_score
    }

Step 4: Full Pipeline Orchestrator

import logging
from dataclasses import dataclass

@dataclass
class GuardrailResult:
    allowed: bool
    reason: str
    sanitized_input: str | None = None
    sanitized_output: str | None = None

class LLMGuardrailPipeline:
    def __init__(self, llm_client, policy_config: dict):
        self.llm = llm_client
        self.policy = policy_config

    def process(self, user_input: str, rag_context: list[str] = None) -> GuardrailResult:
        # --- INPUT GUARDRAILS ---
        # 1. Injection detection
        if detect_prompt_injection(user_input):
            logging.warning(f"INJECTION_DETECTED: {user_input[:100]}")
            return GuardrailResult(allowed=False, reason="Prompt injection detected")
        
        # 2. LlamaGuard safety classification
        safety = llamaguard_classify(user_input)
        if not safety["safe"]:
            return GuardrailResult(allowed=False, reason=f"Unsafe input: {safety['violation']}")
        
        # 3. PII redaction
        sanitized_input, pii_found = redact_pii(user_input)
        if pii_found:
            logging.info(f"PII_REDACTED: {[r.entity_type for r in pii_found]}")
        
        # --- LLM CALL ---
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.policy["system_prompt"]},
                {"role": "user", "content": sanitized_input}
            ]
        )
        raw_output = response.choices[0].message.content
        
        # --- OUTPUT GUARDRAILS ---
        # 4. PII in output
        sanitized_output, output_pii = redact_pii(raw_output)
        
        # 5. Groundedness check (if RAG)
        if rag_context:
            for sentence in split_into_sentences(sanitized_output):
                ground_result = check_groundedness(sentence, rag_context)
                if ground_result["hallucinated"]:
                    sanitized_output += "\n\n⚠️ Note: Some claims could not be verified against source documents."
                    break
        
        return GuardrailResult(
            allowed=True,
            reason="OK",
            sanitized_input=sanitized_input,
            sanitized_output=sanitized_output
        )

---

Key Takeaways

• Two-layer defense: Input guardrails stop bad prompts from reaching the model; output guardrails prevent bad model responses from reaching users. Neither alone is sufficient.
• LlamaGuard is Meta’s open-weight safety classifier specifically trained to detect 11 categories of unsafe content (violence, hate, PII disclosure, etc.) — runs in-process without external API calls.
• Presidio (Microsoft) handles PII redaction accurately across 50+ entity types, far better than naive regex. Always redact PII before sending to third-party APIs.
• NLI-based groundedness catches hallucinations in RAG responses — if the model’s claim contradicts or is unsupported by source documents, it’s likely hallucinated.
• Log everything that gets blocked, including the pattern that triggered the block — this data is essential for tuning thresholds and catching new attack patterns over time.