Build gateway-level guardrails with audit logging

Introduction

When you ship a GenAI gateway without input and output guardrails, a single user request carrying a Social Security number or a prompt-injection payload can leak into your LLM context, your response cache, and your audit logs in milliseconds — turning one careless paste into a compliance incident no downstream filter can undo. By the end of this lesson you'll be able to wire a GatewayGuardrails class into the request lifecycle, run PII detection, prompt-injection scoring, and content-policy checks concurrently on inbound traffic, validate outbound responses for toxicity and PII leakage, and emit structured audit events for every verdict.

Key Terminology

• Input guardrail: a check that runs on the client request before the gateway forwards it to the LLM handler — blocks PII, prompt-injection attempts, and policy violations.
• Output guardrail: a check that runs on the LLM response before it returns to the client — catches toxicity, PII the model generated, and schema drift.
• Guardrail verdict: the structured result of a guardrail check, carrying an action (PASS, BLOCK, or REDACT), the list of triggered rules, and a confidence score used by audit logging and downstream decisioning.

Concepts

Gateway-level guardrails sit on the hot path between rate limiting and the LLM handler, so they must be both correct and fast. Three ideas drive the design in this lesson. First, concurrent detection: PII regexes, injection scoring, and content-policy checks are independent, so asyncio.gather() runs them in parallel and bounds latency by the slowest single check rather than their sum. Second, symmetric input/output scanning: the same verdict model that blocks an inbound prompt also redacts an outbound completion, because the model itself can emit sensitive patterns regardless of how clean the input was. Third, content-addressable audit logging: every verdict — pass or block — is recorded with a SHA-256 content hash so retries of the same malicious payload deduplicate cleanly and post-incident review can trace exactly which rules fired on which request.

Code Walkthrough

How Gateway Guardrails Fit the Request Lifecycle

Before diving into implementation, you need a clear mental model of where guardrails execute relative to the other gateway components. The request flows through validation middleware first (content-length limits, schema compliance, sanitization), then hits rate limiting (token-bucket with Redis), then passes through input guardrails, reaches the LLM handler, returns through output guardrails, optionally populates the response cache, and finally exits to the client.

Loading diagram...

This diagram shows that guardrails occupy the critical path between rate limiting and the LLM handler. Input scanning at node D runs three concurrent checks—PII detection, prompt injection scoring, and content policy matching—using asyncio.gather() so that latency remains bounded by the slowest individual check rather than the sum of all three. Output validation at node G applies a similar concurrent strategy for toxicity scoring, PII re-scanning (because the LLM itself can generate sensitive patterns), and schema conformance of the structured response.

Implementing the GatewayGuardrails Class

The core implementation centers on a GatewayGuardrails class that encapsulates all three phases. The class constructor accepts configuration for PII patterns, injection thresholds, and an audit logger instance. The scan_input method orchestrates concurrent detection by calling _detect_pii, _score_injection, and _check_content_policy through asyncio.gather(). Each detector returns a GuardrailVerdict dataclass, and scan_input aggregates these into a final pass/block decision. The validate_output method mirrors this pattern for response-side checks, while log_audit_event writes structured records to both a local buffer and an external audit sink. This design ensures that adding a new scanner—say, a language detection check—requires only adding one coroutine and including it in the gather() call.

1import asyncio
2import hashlib
3import re
4import time
5from dataclasses import dataclass, field
6from enum import Enum
7from typing import Optional
8
9class VerdictAction(Enum):
10    PASS = "pass"
11    BLOCK = "block"
12    REDACT = "redact"
13
14@dataclass
15class GuardrailVerdict:
16    action: VerdictAction
17    triggered_rules: list[str] = field(default_factory=list)
18    confidence: float = 1.0
19    details: Optional[str] = None
20
21PII_PATTERNS = {
22    "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
23    "credit_card": re.compile(r"\b(?:\d[ -]*?){13,16}\b"),
24    "email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"),
25}
26
27INJECTION_MARKERS = [
28    "ignore previous instructions",
29    "disregard all prior",
30    "system prompt:",
31    "you are now",
32    "reveal your instructions",
33]
34
35class GatewayGuardrails:
36    def __init__(self, pii_patterns=None, injection_threshold=0.7):
37        self.pii_patterns = pii_patterns or PII_PATTERNS
38        self.injection_threshold = injection_threshold
39        self.audit_buffer: list[dict] = []
40
41    async def scan_input(self, content: str, metadata: dict) -> GuardrailVerdict:
42        pii_result, injection_result, policy_result = await asyncio.gather(
43            self._detect_pii(content),
44            self._score_injection(content),
45            self._check_content_policy(content),
46        )
47        verdicts = [pii_result, injection_result, policy_result]
48        blocked = [v for v in verdicts if v.action == VerdictAction.BLOCK]
49        if blocked:
50            combined = GuardrailVerdict(
51                action=VerdictAction.BLOCK,
52                triggered_rules=[r for v in blocked for r in v.triggered_rules],
53                confidence=max(v.confidence for v in blocked),
54            )
55        else:
56            combined = GuardrailVerdict(action=VerdictAction.PASS)
57        self._log_audit_event("input_scan", content, combined, metadata)
58        return combined

Detection, Output Validation, and Audit Internals

The private methods implement the actual scanning logic for both input and output phases. The _detect_pii coroutine iterates over compiled regex patterns and returns a BLOCK verdict if any PII type is found. The _score_injection coroutine computes a normalized score from how many known injection markers appear in the lowercased input, comparing the result against the configurable injection_threshold. The _check_content_policy coroutine is a placeholder for organization-specific rules—profanity filters, topic restrictions, or domain-specific blocklists. _log_audit_event constructs a content-addressable audit record keyed by the SHA-256 hash of the scanned content, enabling efficient deduplication when the same malicious input is retried.

Output validation mirrors input scanning but targets different threats: the LLM itself can generate PII-like patterns (hallucinated SSNs, synthetic credit card numbers), include toxic language that was not present in the input, or produce structured responses that violate your API schema contract. The validate_output method runs detection concurrently and prefers redaction over blocking because a partially useful response with masked PII is often better than no response at all. _redact_pii applies per-type masks, and the get_audit_trail / flush_audit_buffer helpers expose the audit ring buffer for readiness checks and batched export.

Code snippetpython
1    async def _detect_pii(self, content: str) -> GuardrailVerdict:
2        found_types = []
3        for pii_type, pattern in self.pii_patterns.items():
4            if pattern.search(content):
5                found_types.append(pii_type)
6        if found_types:
7            return GuardrailVerdict(
8                action=VerdictAction.BLOCK,
9                triggered_rules=[f"pii_{t}" for t in found_types],
10                confidence=1.0,
11                details=f"Detected PII types: {', '.join(found_types)}",
12            )
13        return GuardrailVerdict(action=VerdictAction.PASS)
14
15    async def _score_injection(self, content: str) -> GuardrailVerdict:
16        lowered = content.lower()
17        hits = sum(1 for marker in INJECTION_MARKERS if marker in lowered)
18        score = hits / len(INJECTION_MARKERS) if INJECTION_MARKERS else 0.0
19        if score >= self.injection_threshold:
20            return GuardrailVerdict(
21                action=VerdictAction.BLOCK,
22                triggered_rules=["prompt_injection"],
23                confidence=score,
24                details=f"Injection score {score:.2f} >= threshold",
25            )
26        return GuardrailVerdict(action=VerdictAction.PASS, confidence=1.0 - score)
27
28    async def _check_content_policy(self, content: str) -> GuardrailVerdict:
29        # Extend with org-specific rules: profanity, topic blocks, etc.
30        return GuardrailVerdict(action=VerdictAction.PASS)
31
32    def _log_audit_event(self, phase, content, verdict, metadata):
33        content_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
34        event = {
35            "timestamp": time.time(),
36            "phase": phase,
37            "content_hash": content_hash,
38            "action": verdict.action.value,
39            "triggered_rules": verdict.triggered_rules,
40            "confidence": verdict.confidence,
41            "user_id": metadata.get("user_id", "anonymous"),
42            "request_id": metadata.get("request_id", "unknown"),
43        }
44        self.audit_buffer.append(event)
45
46    async def validate_output(self, content: str, metadata: dict) -> tuple[str, GuardrailVerdict]:
47        pii_result, toxicity_result = await asyncio.gather(
48            self._detect_pii(content),
49            self._check_content_policy(content),
50        )
51        if pii_result.action == VerdictAction.BLOCK:
52            redacted = self._redact_pii(content)
53            verdict = GuardrailVerdict(
54                action=VerdictAction.REDACT,
55                triggered_rules=pii_result.triggered_rules,
56                confidence=pii_result.confidence,
57                details="PII redacted from output",
58            )
59            self._log_audit_event("output_validation", content, verdict, metadata)
60            return redacted, verdict
61        verdict = GuardrailVerdict(action=VerdictAction.PASS)
62        self._log_audit_event("output_validation", content, verdict, metadata)
63        return content, verdict
64
65    def _redact_pii(self, content: str) -> str:
66        redacted = content
67        redaction_map = {
68            "ssn": "***-**-****",
69            "credit_card": "****-****-****-****",
70            "email": "[REDACTED_EMAIL]",
71        }
72        for pii_type, pattern in self.pii_patterns.items():
73            mask = redaction_map.get(pii_type, "[REDACTED]")
74            redacted = pattern.sub(mask, redacted)
75        return redacted
76
77    def get_audit_trail(self, limit: int = 100) -> list[dict]:
78        return self.audit_buffer[-limit:]
79
80    def flush_audit_buffer(self) -> list[dict]:
81        events = self.audit_buffer.copy()
82        self.audit_buffer.clear()
83        return events

• _detect_pii: Iterates compiled patterns; returns BLOCK with confidence 1.0 because regex matches are deterministic. Triggered rules carry a pii_ prefix for clear identification in audit logs.
• _score_injection: Lowercases input once, counts INJECTION_MARKERS substring hits, and emits BLOCK when the hit ratio meets injection_threshold. On PASS it returns 1.0 - score so downstream systems see how close the input was to triggering.
• _check_content_policy: Stub returning PASS — extend with an external moderation API or local classifier. Keeping it a separate coroutine means latency-heavy ML checks can be added here without blocking PII detection or injection scoring, since all three run concurrently.
• _log_audit_event: Truncated SHA-256 of the content keys the event for content-addressable indexing — the same hashing strategy used by the response cache. Appending to self.audit_buffer keeps the hot path fast; a background flush ships batches to BigQuery or Cloud Logging.
• validate_output: Mirrors scan_input but prefers REDACT over BLOCK so the client receives a usable response. Both paths log an audit event.
• _redact_pii: Per-type masks turn SSNs into ***-**-****, credit cards into ****-****-****-****, and emails into [REDACTED_EMAIL]. The .get() fallback of "[REDACTED]" ensures any new PII pattern added to the configuration gets a safe default mask immediately.
• get_audit_trail / flush_audit_buffer: Support audit-trail inspection (Kubernetes readiness probes can query recent events) and atomic batched export. Copy-then-clear avoids race conditions in single-threaded async contexts.

Do's and Don'ts

Do's

1. ✓Do run _detect_pii, _score_injection, and _check_content_policy concurrently via asyncio.gather() inside scan_input — this bounds total input-scan latency to the slowest individual check rather than the sum of all three, keeping the guardrail on the critical path affordable at gateway scale.
2. ✓Do apply validate_output after the LLM handler with a dedicated PII and toxicity pass — the LLM itself can synthesize SSNs, email addresses, or credit-card patterns from non-sensitive inputs, so relying only on scan_input leaves the outbound path, response cache, and audit log exposed to leakage the inbound scan never saw.
3. ✓Do call _log_audit_event for every GuardrailVerdict, including VerdictAction.PASS decisions — compliance trails that only record blocks cannot reconstruct the full traffic picture or detect anomalous pass-rate shifts that signal a guardrail bypass.

Don'ts

1. ✗Don't await _detect_pii, _score_injection, and _check_content_policy in series with sequential await calls — serial execution makes total scan latency the sum of all three checks; asyncio.gather() is the mechanism the lesson's design depends on to keep that sum from compounding per request.
2. ✗Don't use a bare list literal as the default value for triggered_rules in GuardrailVerdict — Python shares a single mutable object across all dataclass instances created at definition time, causing rule identifiers from one verdict to silently bleed into the next; field(default_factory=list) ensures each verdict accumulates its own independent list.
3. ✗Don't add a new scanner (such as language detection) by duplicating the scan_input orchestration block — the GatewayGuardrails design requires only one new coroutine added to the existing asyncio.gather() call; duplicating the aggregation logic outside that pattern breaks the single GuardrailVerdict contract and introduces a parallel blocking path that serial-adds latency rather than absorbing it.