# LLM Guardrails

> Implementa guardrails de segurança para IA generativa e pipelines LLM/RAG (validação de input, PII scanning, rate limiting, circuit breaker, LGPD compliance). Use ao proteger ou corrigir gaps de segurança de IAs integradas em pipelines com plugins e RAG. Baseado no OWASP Top 10 para Aplicações LLM.

- Skill: `paulohfs/llm-guardrails` (Agent Skill)
- Install (CLI): `npx skillmds@latest add paulohfs/llm-guardrails`
- Raw SKILL.md: https://api.skillmd.com/api/skills/paulohfs/llm-guardrails/raw
- Safety review: pending (external: skill-scanner WARNING, skillspector WARNING)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: PauloHFS (https://skillmd.com/u/paulohfs)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/paulohfs/llm-guardrails

---


# LLM Guardrails — Implementação de Segurança para Sistemas LLM

Receitas de implementação de guardrails de segurança para aplicações LLM/RAG. Cada seção é independente — implemente o que o assessment (`/llm-security-assessment`) identificou como gap.

## Arquitetura de defesa em profundidade

```
Request
  └─ 1. Rate Limit (economiza CPU de guardrails em DDoS)
       └─ 2. Input Guardrails
              ├─ Prompt Injection (regex PT+EN)
              ├─ System Prompt Leakage attempts
              ├─ Encoding bypass (Base64, hex, unicode)
              ├─ Typoglycemia / fuzzy matching
              ├─ Engenharia social (falsa autoridade, falsa urgência)
              └─ Third-party disclosure (LGPD)
                   └─ 3. RAG / LLM Call
                          └─ 4. Output Guardrails
                                 ├─ PII scan + redact (CPF, email, phone, RG)
                                 ├─ URL whitelist sanitization
                                 └─ System prompt leakage (n-gram)
                                      └─ 5. Persist (com redacted output, não raw)
                                           └─ 6. Response ao usuário
```

**Ordem crítica:** Rate limit antes de guardrails (economiza ciclos). Output scan antes de persistir E antes de responder — são dois fluxos independentes, o redact no response path não implica redact no persistence path.

---

## 1. Input Guardrails — Prompt Injection e Leakage

### Estrutura base (Python)

```python
# rag/guardrails.py
import re
from enum import StrEnum
from dataclasses import dataclass
from typing import Optional

class ViolationCategory(StrEnum):
    PROMPT_INJECTION = "prompt_injection"
    SYSTEM_PROMPT_LEAKAGE = "system_prompt_leakage"
    ENCODING_BYPASS = "encoding_bypass"
    THIRD_PARTY_DISCLOSURE = "third_party_disclosure"

_PASSED = "passed"
GUARDRAIL_REFUSAL_MESSAGE = (
    "Não consigo processar essa solicitação. "
    "Por favor, reformule sua pergunta."
)
PRIVACY_FALLBACK_MESSAGE = (
    "Por segurança e conformidade com a LGPD, não posso fornecer "
    "informações sobre dados pessoais de outros colaboradores."
)

@dataclass(frozen=True)
class GuardrailResult:
    status: str  # _PASSED ou ViolationCategory
    detail: str = ""

    @property
    def passed(self) -> bool:
        return self.status == _PASSED

def _blocked(category: ViolationCategory, detail: str = "") -> GuardrailResult:
    return GuardrailResult(status=str(category), detail=detail[:200])
```

### Patterns de Prompt Injection (PT + EN)

```python
_INJECTION_PATTERNS: list[re.Pattern] = [re.compile(p, re.IGNORECASE) for p in [
    # EN — instrução direta
    r"\bignore\s+(all\s+)?(previous|prior|above|earlier)\s+(instructions?|prompts?|rules?|context)\b",
    r"\bforget\s+(all\s+)?(previous|prior|above|earlier)\s+(instructions?|prompts?|rules?|context)\b",
    r"\bdisregard\s+(all\s+)?(previous|prior|above|earlier)\b",
    r"\boverride\s+(your\s+)?(instructions?|rules?|programming|constraints?)\b",
    r"\byou\s+are\s+now\s+(a\s+)?(?!the\s+assistant)",  # "you are now DAN"
    r"\bact\s+as\s+(if\s+you\s+were?\s+)?(?!the\s+assistant)",
    r"\bpretend\s+(you\s+are|to\s+be)\b",
    r"\byour\s+(new\s+)?instructions?\s+(are|is)\b",
    r"\bDAN\b",
    r"\bjailbreak\b",
    r"\bdo\s+anything\s+now\b",
    # PT — instrução direta
    r"\bignore\s+(todas?\s+as?\s+)?(instru[çc][oõ]es?|regras?|contexto|prompt)\b",
    r"\besquece?\s+(todas?\s+as?\s+)?(instru[çc][oõ]es?|regras?|contexto|prompt)\b",
    r"\bdesconsidere?\s+(todas?\s+as?\s+)?(instru[çc][oõ]es?|regras?|contexto)\b",
    r"\bvoc[eê]\s+(agora\s+)?(é|eh)\s+(um|uma)\s+(?!o\s+assistente)",
    r"\bfinja\s+(ser|que\s+(voc[eê]\s+)?(é|eh))\b",
    r"\bcomporte?-?se\s+como\b",
    r"\bnovas?\s+instru[çc][oõ]es?\s+(s[aã]o|é|eh)\b",
    r"\bsuas?\s+instru[çc][oõ]es?\s+(foram\s+)?alteradas?\b",
]]

_SYSTEM_PROMPT_PATTERNS: list[re.Pattern] = [re.compile(p, re.IGNORECASE) for p in [
    r"\brepeat\s+(your\s+)?(system\s+)?prompt\b",
    r"\bshow\s+(me\s+)?(your\s+)?(system\s+)?prompt\b",
    r"\bwhat\s+(are|were)\s+your\s+(instructions?|rules?|system\s+prompt)\b",
    r"\bprint\s+(your\s+)?(instructions?|system\s+prompt|rules?)\b",
    r"\brepita\s+(seu\s+)?(prompt|instru[çc][oõ]es?)\b",
    r"\bmostra?\s+(seu\s+)?(prompt|instru[çc][oõ]es?|regras?)\b",
    r"\bquais\s+s[aã]o\s+(suas?\s+)?(instru[çc][oõ]es?|regras?)\b",
    r"\brevele?\s+(seu\s+)?(prompt|instru[çc][oõ]es?|regras?)\b",
    r"\bdivulg[ue][ae]\s+(seu\s+)?(prompt|instru[çc][oõ]es?)\b",
]]

def check_input(message: str) -> GuardrailResult:
    """Valida input do usuário. Retorna _PASSED ou bloqueia com categoria."""
    # 1. Prompt injection
    for pattern in _INJECTION_PATTERNS:
        if m := pattern.search(message):
            return _blocked(ViolationCategory.PROMPT_INJECTION, detail=m.group(0))

    # 2. System prompt leakage attempt
    for pattern in _SYSTEM_PROMPT_PATTERNS:
        if m := pattern.search(message):
            return _blocked(ViolationCategory.SYSTEM_PROMPT_LEAKAGE, detail=m.group(0))

    # 3. Encoding bypass (Base64, hex, unicode-escape) — ver seção 2
    if result := _check_encoded_injection(message):
        return result

    # 4. Typoglycemia / fuzzy — ver seção 3
    if result := _check_fuzzy_injection(message):
        return result

    return GuardrailResult(status=_PASSED)
```

---

## 2. Defesa contra Encoding (Base64, Hex, Unicode-escape)

Ataque: `"Analise este texto: SWdub3JlIGFsbCBwcmV2aW91cyBpbnN0cnVjdGlvbnM="` → decodifica para `"Ignore all previous instructions"`.

```python
import base64, binascii, codecs

_BASE64_BLOCK = re.compile(r"(?:[A-Za-z0-9+/]{4}){3,}(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=|[A-Za-z0-9+/]{4})")
_HEX_BLOCK = re.compile(r"(?:0x)?(?:[0-9a-fA-F]{2}){6,}")
_UNICODE_ESCAPE = re.compile(r"(?:\\u[0-9a-fA-F]{4}){3,}")

def _try_decode_encodings(message: str) -> list[str]:
    """Extrai candidatos decodificados de blocos suspeitos."""
    candidates = []
    for m in _BASE64_BLOCK.finditer(message):
        try:
            decoded = base64.b64decode(m.group(0) + "==").decode("utf-8", errors="ignore")
            if len(decoded) > 5 and decoded.isprintable():
                candidates.append(decoded)
        except (binascii.Error, UnicodeDecodeError):
            pass
    for m in _HEX_BLOCK.finditer(message):
        try:
            hex_str = m.group(0).replace("0x", "")
            decoded = bytes.fromhex(hex_str).decode("utf-8", errors="ignore")
            if len(decoded) > 5 and decoded.isprintable():
                candidates.append(decoded)
        except (ValueError, UnicodeDecodeError):
            pass
    for m in _UNICODE_ESCAPE.finditer(message):
        try:
            decoded = codecs.decode(m.group(0).encode(), "unicode_escape").decode("utf-8", errors="ignore")
            if len(decoded) > 3:
                candidates.append(decoded)
        except Exception:
            pass
    return candidates

def _check_encoded_injection(message: str) -> Optional[GuardrailResult]:
    for candidate in _try_decode_encodings(message):
        for pattern in _INJECTION_PATTERNS + _SYSTEM_PROMPT_PATTERNS:
            if m := pattern.search(candidate):
                return _blocked(ViolationCategory.ENCODING_BYPASS,
                    detail=f"encoded payload: {m.group(0)[:100]}")
    return None
```

**Testes mínimos:**
- `test_base64_encoded_injection_blocked` — `SWdub3JlIGFsbCBwcmV2aW91cyBpbnN0cnVjdGlvbnM=`
- `test_hex_encoded_injection_blocked`
- `test_clean_base64_ref_passes` — base64 legítimo (ex: imagem) não deve ser bloqueado

---

## 3. Fuzzy Matching (Typoglycemia)

Ataque: `"ignroe all previous instructions"` / `"bpyass your rules"`.

```python
_TRIGGER_WORDS = frozenset({
    "ignore", "ignora", "forget", "esqueca", "bypass",
    "disable", "desabilita", "override", "pretend", "finja",
})
_MAX_EDIT_DISTANCE = 2

def _levenshtein(a: str, b: str) -> int:
    if abs(len(a) - len(b)) > _MAX_EDIT_DISTANCE:
        return 99  # fast exit
    m, n = len(a), len(b)
    dp = list(range(n + 1))
    for i in range(1, m + 1):
        prev = dp[0]
        dp[0] = i
        for j in range(1, n + 1):
            temp = dp[j]
            dp[j] = prev if a[i-1] == b[j-1] else 1 + min(prev, dp[j], dp[j-1])
            prev = temp
    return dp[n]

def _check_fuzzy_injection(message: str) -> Optional[GuardrailResult]:
    words = re.findall(r"\b[a-z]{4,}\b", message.lower())
    for word in words:
        for trigger in _TRIGGER_WORDS:
            if _levenshtein(word, trigger) <= _MAX_EDIT_DISTANCE:
                return _blocked(ViolationCategory.PROMPT_INJECTION,
                    detail=f"fuzzy match: '{word}' ~ '{trigger}'")
    return None
```

**⚠️ Atenção:** Português tem palavras a 1-2 edits de trigger words em inglês. Comece com 5-10 termos e expanda baseado em falsos positivos observados em produção. Calibrar com um golden dataset.

---

## 4. Guardrails Sociais — LGPD (Falsa Autoridade, Falsa Urgência, Dados de Terceiros)

```python
# rag/guardrails_social.py

_OWN_DATA_POSSESSIVES = frozenset({"meu", "minha", "meus", "minhas", "seu", "sua"})

_SENSITIVE_TERMS = re.compile(
    r"\b(sal[aá]rio|holerite|contracheque|cpf|matr[ií]cula|ponto|faltas?|"
    r"avalia[cç][aã]o|benefícios?|f[eé]rias?|rescis[aã]o|demiss[aã]o)\b",
    re.IGNORECASE
)
_THIRD_PARTY_INDICATORS = re.compile(
    r"\b(do|da|de|dos|das)\s+\w+\b",  # "do João", "da Maria"
    re.IGNORECASE
)

_FALSE_AUTHORITY_PATTERNS = [re.compile(p, re.IGNORECASE) for p in [
    r"\bsou\s+do\s+(rh|recursos?\s+humanos?)\b",
    r"\btrabalho\s+na\s+[aá]rea\s+de\s+(rh|recursos?\s+humanos?)\b",
    r"\bpor\s+ordem\s+do\s+(gerente|diretor|chefe|rh)\b",
    r"\bem\s+nome\s+do\s+(rh|gerente|diretor|chefe)\b",
    r"\bi\s+(work|am)\s+(at|in|for)\s+(hr|human\s+resources?)\b",
    r"\bby\s+order\s+of\s+(management|hr|director)\b",
]]

_FALSE_URGENCY_PATTERNS = [re.compile(p, re.IGNORECASE) for p in [
    r"\bé\s+urgente\b",
    r"\bpara\s+(auditoria|investiga[cç][aã]o\s+interna?|compliance)\b",
    r"\bpreciso\s+(agora|imediatamente|urgente)\b",
    r"\bdeadline\s+(hoje|amanhã|agora)\b",
    r"\bit'?s?\s+urgent\b",
    r"\bfor\s+(audit|internal\s+investigation|compliance)\b",
]]

def check_false_authority(message: str) -> Optional[GuardrailResult]:
    for pattern in _FALSE_AUTHORITY_PATTERNS:
        if m := pattern.search(message):
            return _blocked(ViolationCategory.THIRD_PARTY_DISCLOSURE,
                detail=f"false authority: {m.group(0)}")
    return None

def check_false_urgency(message: str) -> Optional[GuardrailResult]:
    for pattern in _FALSE_URGENCY_PATTERNS:
        if m := pattern.search(message):
            return _blocked(ViolationCategory.THIRD_PARTY_DISCLOSURE,
                detail=f"false urgency: {m.group(0)}")
    return None

def check_third_party_disclosure(message: str) -> Optional[GuardrailResult]:
    """
    Bloqueia solicitações de dados pessoais de terceiros (LGPD).
    Permite: "qual é meu salário?" (próprios dados com possessivo adjacente)
    Bloqueia: "qual é o salário do João?" (dados de terceiro)
    """
    msg_lower = message.lower()
    for match in _SENSITIVE_TERMS.finditer(msg_lower):
        term_pos = match.start()
        # Verificar possessivo próprio ADJACENTE (prev/next word)
        before = msg_lower[:term_pos].rstrip().split()
        after = msg_lower[match.end():].lstrip().split()
        prev_word = before[-1] if before else ""
        next_word = after[0] if after else ""
        if prev_word in _OWN_DATA_POSSESSIVES or next_word in _OWN_DATA_POSSESSIVES:
            continue  # dado próprio — permitir
        # Verificar indicador de terceiro
        remaining = msg_lower[term_pos:]
        if _THIRD_PARTY_INDICATORS.search(remaining):
            return _blocked(ViolationCategory.THIRD_PARTY_DISCLOSURE,
                detail=f"third party data: '{match.group(0)}'")
    return None
```

**Adicionar em `check_input()`:**
```python
# Após os checks existentes, antes do return _PASSED:
if result := check_third_party_disclosure(message):
    return result
if result := check_false_authority(message):
    return result
if result := check_false_urgency(message):
    return result
```

**Logging seguro (nunca logar conversation_id em texto claro):**
```python
import hashlib

def _hash_conversation_id(conv_id: str) -> str:
    return hashlib.sha256(conv_id.encode()).hexdigest()[:12]

def log_guardrail_violation(
    logger, category: str, detail: str,
    conversation_id: str, message_length: int
) -> None:
    logger.warning("guardrail.violation", extra={
        "category": category,
        "detail": detail[:200],
        "conversation_id_hash": _hash_conversation_id(conversation_id),
        "message_length": message_length,
        # NUNCA: "conversation_id": conversation_id (raw)
        # NUNCA: "message": message (conteúdo do usuário)
    })
```

---

## 5. Output PII Scan + Redact

**Estratégia: redact-not-block.** A resposta ainda vai ao usuário, mas com PII mascarado. Bloquear porque o LLM alucionou um CPF seria pior UX.

```python
# rag/guardrails.py (continuação)
from enum import StrEnum
from dataclasses import dataclass, field

class OutputViolationCategory(StrEnum):
    PII_CPF = "pii_cpf"
    PII_EMAIL = "pii_email"
    PII_PHONE = "pii_phone"
    PII_RG = "pii_rg"

@dataclass(frozen=True)
class OutputScanResult:
    clean_text: str
    violations: list[OutputViolationCategory] = field(default_factory=list)

    @property
    def has_violations(self) -> bool:
        return bool(self.violations)

# CPF com dígito verificador (mod 11) — mais preciso que regex simples
_CPF_PATTERN = re.compile(r"\b\d{3}[.\s-]?\d{3}[.\s-]?\d{3}[-\s]?\d{2}\b")
_EMAIL_PATTERN = re.compile(r"\b[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}\b")
_BR_PHONE_PATTERN = re.compile(r"\b(?:\+?55\s?)?(?:\(?\d{2}\)?\s?)(?:9\s?)?\d{4}[-\s]?\d{4}\b")
_RG_PATTERN = re.compile(r"\b\d{1,2}[.\s]?\d{3}[.\s]?\d{3}[-\s]?[\dxX]\b")

# Allowlist de contatos institucionais — não redactar emails/telefones legítimos
INSTITUTIONAL_CONTACTS: frozenset[str] = frozenset({
    # ex: "contato@empresa.com", "(11) 4000-0000"
})

def check_output(text: str, institutional_contacts: frozenset[str] | None = None) -> OutputScanResult:
    """
    Escaneia output do LLM por PII. Redacta, não bloqueia.
    Preservar contatos institucionais da knowledge base.
    """
    contacts = institutional_contacts or INSTITUTIONAL_CONTACTS
    violations: list[OutputViolationCategory] = []
    clean = text

    def _should_redact(match_str: str) -> bool:
        return match_str not in contacts

    if _CPF_PATTERN.search(clean):
        new_clean = _CPF_PATTERN.sub(
            lambda m: "[CPF OMITIDO]" if _should_redact(m.group(0)) else m.group(0), clean
        )
        if new_clean != clean:
            violations.append(OutputViolationCategory.PII_CPF)
            clean = new_clean

    if _EMAIL_PATTERN.search(clean):
        new_clean = _EMAIL_PATTERN.sub(
            lambda m: "[EMAIL OMITIDO]" if _should_redact(m.group(0)) else m.group(0), clean
        )
        if new_clean != clean:
            violations.append(OutputViolationCategory.PII_EMAIL)
            clean = new_clean

    if _BR_PHONE_PATTERN.search(clean):
        new_clean = _BR_PHONE_PATTERN.sub(
            lambda m: "[TELEFONE OMITIDO]" if _should_redact(m.group(0)) else m.group(0), clean
        )
        if new_clean != clean:
            violations.append(OutputViolationCategory.PII_PHONE)
            clean = new_clean

    if _RG_PATTERN.search(clean):
        new_clean = _RG_PATTERN.sub(
            lambda m: "[RG OMITIDO]" if _should_redact(m.group(0)) else m.group(0), clean
        )
        if new_clean != clean:
            violations.append(OutputViolationCategory.PII_RG)
            clean = new_clean

    return OutputScanResult(clean_text=clean, violations=violations)
```

**⚠️ Persistência:** Usar o `clean_text` também ao persistir no banco, não o raw `llm_response.content`:

```python
# pipeline.py — ERRADO (erro comum)
asyncio.create_task(store_turn(conv_id, llm_response.content))  # raw, sem redact
answer = check_output(llm_response.content).clean_text
return answer  # redactado para o usuário, mas persistido raw ← bug

# pipeline.py — CORRETO
scan_result = check_output(llm_response.content)
answer = scan_result.clean_text
asyncio.create_task(store_turn(conv_id, answer))  # persistir o redactado
return answer
```

---

## 6. URL Sanitization

```python
import urllib.parse

DEFAULT_ALLOWED_DOMAINS = frozenset({
    "example.org", "example-corp.io", "trusted.net", "securecloud.io", "research.ai",
})

_URL_PATTERN = re.compile(r"https?://[^\s\)\]\>\"']+", re.IGNORECASE)

def sanitize_urls(text: str, allowed_domains: frozenset[str] = DEFAULT_ALLOWED_DOMAINS) -> tuple[str, int]:
    """Remove URLs de domínios não-permitidos. Retorna (texto_limpo, n_removidas)."""
    removed = 0

    def _replace(m: re.Match) -> str:
        nonlocal removed
        url = m.group(0).rstrip(".,;:!?)")
        host = urllib.parse.urlparse(url).hostname or ""
        if any(host == d or host.endswith(f".{d}") for d in allowed_domains):
            return m.group(0)
        removed += 1
        return "[LINK REMOVIDO]"

    clean = _URL_PATTERN.sub(_replace, text)
    return clean, removed
```

---

## 7. System Prompt Leakage — N-gram Detection

```python
def check_prompt_leakage(
    response: str,
    prompt_template: str,
    *,
    ngram_size: int = 8,
    refusal_message: str = "Não posso fornecer essa informação."
) -> str:
    """
    Verifica se o output contém trechos verbatim do system prompt.
    Retorna o response original se limpo, ou refusal_message se detectado.
    """
    def ngrams(text: str, n: int) -> set[str]:
        words = re.sub(r"[^\w\s]", "", text.lower()).split()
        return {" ".join(words[i:i+n]) for i in range(max(0, len(words) - n + 1))}

    prompt_ngrams = ngrams(prompt_template, ngram_size)
    response_ngrams = ngrams(response, ngram_size)

    if prompt_ngrams & response_ngrams:
        return refusal_message
    return response
```

---

## 8. Rate Limiter por Conversation (Token Bucket)

```python
# middleware/rate_limiter.py
import asyncio
import time
from collections import defaultdict
from dataclasses import dataclass, field

RATE_LIMIT_MESSAGE = (
    "Por favor, aguarde um momento antes de enviar outra mensagem. "
    "Estou processando muitas perguntas ao mesmo tempo. 😊"
)

@dataclass
class _Bucket:
    tokens: float
    last_refill: float
    last_used: float = field(default_factory=time.monotonic)

class ConversationRateLimiter:
    """
    Token-bucket rate limiter por conversation_id (in-memory).

    ⚠️ Single-instance only. Para multi-replica: migrar para Redis
    com INCR/EXPIRE atômico para garantir atomicidade cross-instance.
    """

    def __init__(
        self,
        max_tokens: float = 10.0,
        refill_rate: float = 10.0 / 60.0,  # tokens por segundo
        ttl_seconds: float = 600.0,
        max_buckets: int = 10_000,
    ):
        self._max = max_tokens
        self._rate = refill_rate
        self._ttl = ttl_seconds
        self._max_buckets = max_buckets
        self._buckets: dict[str, _Bucket] = {}
        self._locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)

    async def is_allowed(self, conversation_id: str) -> bool:
        """Consome 1 token. True = permitido, False = rate-limited."""
        async with self._locks[conversation_id]:
            now = time.monotonic()
            self._evict_stale(now)

            if conversation_id not in self._buckets:
                self._buckets[conversation_id] = _Bucket(
                    tokens=self._max - 1,  # consome o primeiro token
                    last_refill=now,
                )
                return True

            bucket = self._buckets[conversation_id]
            elapsed = now - bucket.last_refill
            bucket.tokens = min(self._max, bucket.tokens + elapsed * self._rate)
            bucket.last_refill = now
            bucket.last_used = now

            if bucket.tokens >= 1.0:
                bucket.tokens -= 1.0
                return True
            return False

    def _evict_stale(self, now: float) -> None:
        if len(self._buckets) < self._max_buckets:
            return
        stale = [k for k, b in self._buckets.items() if now - b.last_used > self._ttl]
        for k in stale:
            del self._buckets[k]
            self._locks.pop(k, None)
```

**Integração no pipeline (Rate limit ANTES dos guardrails):**
```python
# pipeline.py
_rate_limiter = ConversationRateLimiter()

async def run_pipeline(message: str, conversation_id: str, ...) -> RAGResponse:
    # Step 0a — Rate limit primeiro (economiza guardrail CPU em DDoS)
    if not await _rate_limiter.is_allowed(conversation_id):
        return RAGResponse(content=RATE_LIMIT_MESSAGE)

    # Step 0b — Input guardrails
    guardrail_result = check_input(message)
    if not guardrail_result.passed:
        fallback = PRIVACY_FALLBACK_MESSAGE if guardrail_result.status == str(ViolationCategory.THIRD_PARTY_DISCLOSURE) else GUARDRAIL_REFUSAL_MESSAGE
        log_guardrail_violation(logger, guardrail_result.status, guardrail_result.detail, conversation_id, len(message))
        return RAGResponse(content=fallback)

    # ... resto do pipeline
```

---

## 9. Circuit Breaker no LLM Client

```python
# llm/circuit_breaker.py
import asyncio
import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # normal
    OPEN = "open"          # rejeitando
    HALF_OPEN = "half_open"  # testando

CIRCUIT_OPEN_MESSAGE = (
    "O serviço está temporariamente indisponível. "
    "Tente novamente em alguns minutos."
)

class CircuitBreaker:
    """
    Circuit breaker para chamadas ao LLM gateway.
    CLOSED → OPEN após failure_threshold falhas em reset_timeout segundos.
    OPEN → HALF_OPEN após reset_timeout.
    HALF_OPEN → CLOSED (sucesso) ou OPEN (falha).
    """

    def __init__(self, failure_threshold: int = 5, reset_timeout: float = 120.0):
        self._threshold = failure_threshold
        self._reset_timeout = reset_timeout
        self._failures = 0
        self._opened_at: float | None = None
        self._state = CircuitState.CLOSED
        self._lock = asyncio.Lock()

    @property
    def state(self) -> CircuitState:
        return self._state

    async def call(self, fn, *args, **kwargs):
        """Executa fn protegido pelo circuit breaker."""
        async with self._lock:
            if self._state == CircuitState.OPEN:
                if time.monotonic() - (self._opened_at or 0) > self._reset_timeout:
                    self._state = CircuitState.HALF_OPEN
                else:
                    raise CircuitOpenError(CIRCUIT_OPEN_MESSAGE)

        try:
            result = await fn(*args, **kwargs)
            async with self._lock:
                self._failures = 0
                self._state = CircuitState.CLOSED
            return result
        except Exception as e:
            async with self._lock:
                self._failures += 1
                if self._failures >= self._threshold:
                    self._state = CircuitState.OPEN
                    self._opened_at = time.monotonic()
            raise

class CircuitOpenError(Exception):
    pass
```

---

## 10. Confidence Threshold no Retrieval (anti-hallucination)

```python
# pipeline.py — após search_knowledge_base()
async def run_pipeline(message: str, ...) -> RAGResponse:
    # ...
    results = await search_knowledge_base(message)

    # Se KB não tem nada relevante, não chamar o LLM — evita hallucination
    if not results or max(r.score for r in results) < settings.min_retrieval_score:
        return RAGResponse(content=FALLBACK_NO_KNOWLEDGE_MESSAGE)

    # ... chamar LLM com os results
```

```python
# config.py
class GuardrailSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="GUARDRAIL_", env_file=".env", extra="ignore")
    allowed_output_domains: str = "example.org,example-corp.io,trusted.net,securecloud.io,research.ai"
    rate_limit_max_messages: int = 10
    rate_limit_window_seconds: float = 60.0
    min_retrieval_score: float = 1.0  # calibrar com golden dataset

    @property
    def allowed_domains_set(self) -> frozenset[str]:
        return frozenset(d.strip() for d in self.allowed_output_domains.split(","))
```

**Calibração:** Usar um golden dataset de perguntas esperadas + perguntas fora do escopo. Plotar a curva precision-recall e escolher o threshold que minimiza hallucinations sem aumentar falsos negativos.

---

## Checklist de implementação

Após implementar, verificar:

- [ ] Rate limit antes dos guardrails no pipeline
- [ ] Input guardrails com patterns PT+EN
- [ ] Defesa contra encoding (Base64, hex, unicode)
- [ ] Fuzzy matching calibrado com golden dataset (começar com ≤10 trigger words)
- [ ] Guardrails sociais (falsa autoridade, falsa urgência, dados de terceiros)
- [ ] Output PII scan com redact-not-block
- [ ] **Output scan aplicado ANTES de persistir no banco** (não só antes de responder)
- [ ] URL whitelist sanitization
- [ ] System prompt com regra de confidencialidade
- [ ] N-gram leakage check no output
- [ ] Rate limiter por conversation_id (não só por client_id no gateway)
- [ ] Circuit breaker no LLM client
- [ ] Confidence threshold no retrieval
- [ ] Logs seguros: sem conversation_id raw, sem mensagem do usuário, IDs como hash
- [ ] TTL configurado no banco de histórico de conversas (LGPD Art. 15/16)
- [ ] Contatos institucionais na allowlist do output scanner