AI Compliance Monitoring: Architecture Blueprint & Production Stack

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
import re
import time
import httpx

app = FastAPI(title="AI Compliance & Governance Proxy")

# PII Redaction Regex Patterns
SSN_PATTERN = re.compile(r"\b\d{3}-\d{2}-\d{4}\b")
CARD_PATTERN = re.compile(r"\b(?:\d[ -]*?){13,16}\b")
EMAIL_PATTERN = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b")

class LLMRequestPayload(BaseModel):
    model: str
    prompt: str
    temperature: float = 0.7

class GovernanceResponse(BaseModel):
    sanitized_prompt: str
    pii_redacted_count: int
    guardrail_status: str
    processing_time_ms: float

def sanitize_pii(text: str) -> tuple[str, int]:
    """Scans and replaces PII data matches with compliance tokens."""
    redacted_count = 0
    text, n1 = SSN_PATTERN.subn("<SSN_REDACTED>", text)
    text, n2 = CARD_PATTERN.subn("<CREDIT_CARD_REDACTED>", text)
    text, n3 = EMAIL_PATTERN.subn("<EMAIL_REDACTED>", text)
    redacted_count = n1 + n2 + n3
    return text, redacted_count

@app.post("/v1/governance/proxy", response_model=GovernanceResponse)
async def process_compliance_proxy(payload: LLMRequestPayload):
    start_time = time.perf_counter()
    
    # 1. Check Prompt Injection Guardrails
    jailbreak_keywords = ["ignore previous instructions", "override system prompt", "DAN mode"]
    if any(kw in payload.prompt.lower() for kw in jailbreak_keywords):
        raise HTTPException(status_code=403, detail="PROMPT_INJECTION_DETECTED: Request blocked by security policy.")
        
    # 2. Sanitize PII Entities
    sanitized_text, redacted_count = sanitize_pii(payload.prompt)
    
    # 3. Log Audit Telemetry to WORM Compliance Sink
    elapsed_ms = (time.perf_counter() - start_time) * 1000.0
    
    return GovernanceResponse(
        sanitized_prompt=sanitized_text,
        pii_redacted_count=redacted_count,
        guardrail_status="PASSED",
        processing_time_ms=round(elapsed_ms, 2)
    )