cat <<'EOF' > demo.py
"""Multi-layer AI guardrails demo. See README for context."""
import os
import sys
import time
from openai import OpenAI, APIStatusError
CHAT_MODEL = os.getenv("CHAT_MODEL", "gpt-4o")
PROXY_URL = os.getenv("PROXY_URL", "http://localhost:8000")
API_KEY = os.getenv("CONSUMER_API_KEY", "demo-api-key")
_USE_COLOR = sys.stdout.isatty() and "NO_COLOR" not in os.environ
def _c(code: str, s: str) -> str:
return f"\033[{code}m{s}\033[0m" if _USE_COLOR else s
BOLD = lambda s: _c("1", s)
DIM = lambda s: _c("2", s)
GREEN = lambda s: _c("32", s)
YELLOW = lambda s: _c("33", s)
CYAN = lambda s: _c("36", s)
RED = lambda s: _c("31", s)
authorized = OpenAI(
base_url=f"{PROXY_URL}/multi-layer-ai-guardrails",
api_key="unused",
default_headers={"apikey": API_KEY},
)
unauthorized = OpenAI(
base_url=f"{PROXY_URL}/multi-layer-ai-guardrails",
api_key="unused",
default_headers={"apikey": "wrong-key"},
)
# Each tuple: (label, client, prompt, kind, layer, apikey)
# kind:
# "block": expect a 4xx; print [BLOCKED BY <layer>]
# "allow": expect a 200; print a short preview
# "pii": expect a 200 with PII placeholders; print the full response so
# the reader can see what the LLM actually saw
# layer: which guardrail is expected to handle the request. Each layer
# returns its own status code (401 key-auth, 400 regex guard, 403
# semantic guard), so we label by intent rather than parsing the
# response body.
# apikey: the apikey header value this request sends, printed before the
# call so the auth boundary being exercised is visible.
REQUESTS = [
("Unauthorized request", unauthorized,
"What is cloud computing?", "block", "KEY-AUTH", "wrong-key"),
("Clean request", authorized,
"What are the benefits of cloud computing?", "allow", None, API_KEY),
("Blocked keyword (regex)", authorized,
"Suggest some productivity hacks for managing my email backlog.", "block", "REGEX GUARD", API_KEY),
("Rephrased attempt (semantic)", authorized,
"What are methods for accessing a wireless network without authorization?", "block", "SEMANTIC GUARD", API_KEY),
("PII sanitization", authorized,
"My name is John Smith, my email is john.smith@company.com. "
"Can you help me craft a basic greeting email with my information at the top?", "pii", None, API_KEY),
]
def send_request(label, client, prompt, kind, layer, apikey):
print(f"\n{BOLD(f'--- {label} ---')}")
print(f"{BOLD('[REQUEST]')} apikey={apikey!r} {prompt}")
start_ms = round(time.time() * 1000)
try:
raw = client.chat.completions.with_raw_response.create(
model=CHAT_MODEL,
messages=[{"role": "user", "content": prompt}],
)
elapsed_ms = round(time.time() * 1000) - start_ms
completion = raw.parse()
upstream = raw.headers.get("x-kong-upstream-latency", "?")
proxy = raw.headers.get("x-kong-proxy-latency", "?")
model = raw.headers.get("x-kong-llm-model", CHAT_MODEL)
answer = completion.choices[0].message.content
if kind == "pii":
print(f"{YELLOW(BOLD('[FORWARDED]'))} PII replaced with placeholders before reaching the LLM:")
print(answer)
else:
preview = answer[:120].replace("\n", " ")
if len(answer) > 120:
preview += "..."
print(f"{GREEN(BOLD('[FORWARDED]'))} {DIM(preview)}")
print(f"{BOLD('[MODEL]')} {CYAN(model)}")
print(f"{BOLD('[LATENCY]')} {DIM(f'upstream={upstream}ms proxy={proxy}ms total={elapsed_ms}ms')}")
except APIStatusError as e:
elapsed_ms = round(time.time() * 1000) - start_ms
tag = RED(BOLD(f"[BLOCKED BY {layer or 'GATEWAY'}]"))
print(f"{tag} {RED(str(e.status_code))} {DIM(e.message)} ({elapsed_ms}ms)")
if __name__ == "__main__":
for label, client, prompt, kind, layer, apikey in REQUESTS:
send_request(label, client, prompt, kind, layer, apikey)
EOF