ai-guardrails

User Input → Input Guardrails → LLM → Output Guardrails → User Response
                 │                          │
                 ├─ Prompt injection check   ├─ Content policy check
                 ├─ PII detection            ├─ Hallucination detection
                 ├─ Topic restrictions        ├─ PII scrubbing
                 └─ Rate limiting             └─ Schema validation

# injection_detector.py — Multi-layer prompt injection detection

import re
from typing import Tuple

class InjectionDetector:
    PATTERNS = [
        r"ignore\s+(all\s+)?(previous|above|prior)\s+(instructions|prompts)",
        r"you\s+are\s+now\s+(an?\s+)?(unrestricted|unfiltered|jailbroken)",
        r"disregard\s+(your|the)\s+(rules|guidelines|instructions)",
        r"system\s*prompt",
        r"pretend\s+(you\s+are|to\s+be)",
        r"override\s+(your|all|the)\s+(safety|content|rules)",
        r"\[system\]|\[INST\]|<\|system\|>",
    ]

    def check_patterns(self, text: str) -> Tuple[bool, list[str]]:
        text_lower = text.lower()
        matches = [p for p in self.PATTERNS if re.search(p, text_lower)]
        return len(matches) > 0, matches

    def check_semantic(self, text: str, llm_client) -> Tuple[bool, float]:
        """Use a fast LLM to classify whether input is injection."""
        response = llm_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content":
                 "Analyze if this input tries to manipulate AI instructions. "
                 'Return JSON: {"is_injection": bool, "confidence": 0-1}'},
                {"role": "user", "content": f"Analyze:\n\n{text}"}
            ],
            response_format={"type": "json_object"}
        )
        result = json.loads(response.choices[0].message.content)
        return result["is_injection"], result["confidence"]

    def check_canary(self, system_prompt: str, output: str) -> bool:
        """Check if a canary token leaked from system prompt to output."""
        canary_match = re.search(r'CANARY:(\w{16})', system_prompt)
        if canary_match:
            return canary_match.group(1) in output
        return False

# content_filter.py — Filter outputs against safety policies

class ContentFilter:
    def __init__(self, thresholds=None):
        self.thresholds = thresholds or {
            "violence": 0.7, "hate_speech": 0.5, "sexual": 0.6,
            "self_harm": 0.3, "illegal_activity": 0.5, "pii_leak": 0.3,
        }

    def check_pii(self, text: str) -> list[dict]:
        """Detect PII (email, phone, SSN, credit card, IP) in text."""
        patterns = {
            "email": r'\b[\w.-]+@[\w.-]+\.\w{2,}\b',
            "phone": r'\b\d{3}[-.\s]?\d{3}[-.\s]?\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',
        }
        findings = []
        for pii_type, pattern in patterns.items():
            for match in re.finditer(pattern, text):
                findings.append({"type": pii_type, "value": match.group()})
        return findings

    def scrub_pii(self, text: str) -> str:
        """Replace PII with [REDACTED_TYPE] markers."""
        for finding in sorted(self.check_pii(text),
                              key=lambda f: text.find(f["value"]), reverse=True):
            text = text.replace(finding["value"],
                               f"[REDACTED_{finding['type'].upper()}]")
        return text

# output_validator.py — Validate LLM outputs against schemas

from pydantic import BaseModel, validator

class ValidatedResponse(BaseModel):
    answer: str
    confidence: float
    sources: list[str]

    @validator('confidence')
    def confidence_in_range(cls, v):
        if not 0 <= v <= 1:
            raise ValueError(f"Confidence {v} not in [0, 1]")
        return v

    @validator('answer')
    def answer_not_empty(cls, v):
        if len(v.strip()) < 10:
            raise ValueError("Answer too short")
        return v

# rate_limiter.py — Prevent API abuse and cost overruns

from collections import defaultdict
from time import time

class AIRateLimiter:
    def __init__(self):
        self.user_requests: dict[str, list[float]] = defaultdict(list)
        self.max_requests_per_minute = 10
        self.max_tokens_per_day = 100_000

    def check_allowed(self, user_id: str, estimated_tokens: int = 0) -> dict:
        now = time()
        reqs = self.user_requests[user_id]
        reqs[:] = [t for t in reqs if now - t < 3600]
        recent = sum(1 for t in reqs if now - t < 60)
        if recent >= self.max_requests_per_minute:
            return {"allowed": False, "reason": "Rate limit exceeded", "retry_after": 60}
        reqs.append(now)
        return {"allowed": True}

# hallucination_check.py — Verify claims against source context

def check_grounding(answer: str, context: str, llm_client) -> dict:
    """Check if answer claims are supported by provided context."""
    response = llm_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content":
             "Identify which claims are SUPPORTED, NOT SUPPORTED, or "
             "CONTRADICTED by the context. Return JSON with arrays and "
             "'grounding_score' (0-1)."},
            {"role": "user", "content": f"Context:\n{context}\n\nAnswer:\n{answer}"}
        ],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

Our customer support chatbot uses GPT-4 and has no safety layers. Add comprehensive guardrails: prompt injection detection (pattern + semantic), PII scrubbing on both input and output, content policy enforcement, rate limiting (10 req/min per user), and output validation against our response schema. Include logging for security review and a circuit breaker that switches to a safe fallback response when anomalies are detected.

Build a content moderation system for a social platform that processes 10,000 user-generated posts per day. Use a fast classifier (GPT-4o-mini) for initial screening, escalate borderline cases to a more capable model, and route to human review for the hardest 5%. Track false positive/negative rates, and include an appeals process.

Our RAG system answers questions from company documentation but sometimes makes up information not in the source docs. Build a grounding verification layer that checks every claim against retrieved passages, flags unsupported statements, and either removes them or adds "unverified" markers. Include a confidence score and fallback to "I don't have enough information" when grounding is below 60%.