The Problem
Guardrails AI validators, Lakera Guard, Bedrock Guardrails, Azure Prompt Shields, Llama Guard — every product in Lesson 3 is trained or configured against generic categories: PII in general, toxicity in general, jailbreaks in general. That's genuinely useful, and you should use one or more of them. But production systems accumulate domain-specific requirements none of those products were built to know about: your internal project codenames that count as sensitive even though they're not PII by any standard definition, a fixed set of topics your compliance team has explicitly scoped you into (and everything else is out of scope), and a specific, small set of tool actions your agent is allowed to take with specific parameter limits on each one.
At that point, "call a vendor API" stops being the answer, because no vendor's model was trained on your codenames or your tool schema. You need a custom layer. This lesson builds one from three components that show up in almost every real custom guardrail pipeline: a hybrid PII detector, a topic-restriction classifier, and an action-sandboxing gate — then combines all three into a single decision with an explicit precedence rule, because the hardest part of a multi-layer pipeline is rarely any one layer; it's what happens when two layers disagree.
Layer 1: hybrid PII detection — regex alone isn't enough
Regex is precise and fast for structured PII — anything with a fixed shape: a Social Security number, an email address, a phone number, a credit card number. It will never reliably catch unstructured PII like a person's name or an organization, because names have no fixed shape. "My manager Sarah Chen approved this" contains a name regex cannot find without a list of every name in existence.
That gap is what named entity recognition (NER) is for. A real NER pass — spaCy's en_core_web_sm, a transformer-based NER model, or a hosted service — uses a statistical model trained on labeled text to recognize that "Sarah Chen" functions as a person's name in context, the same way it recognizes "Acme Corp" as an organization. This lesson's code tries a real spaCy model first and, if it isn't installed, falls back to a hand-written heuristic: look for runs of consecutive capitalized words that aren't at the start of a sentence, optionally cued by a title ("Dr.", "Mr.") or a corporate suffix ("Inc.", "LLC"). It's a deliberately crude stand-in — it will miss single-word names and occasionally misfire on a capitalized product name — but it demonstrates the real technique: surface-pattern heuristics can approximate NER when you can't afford the dependency, at a real cost in precision and recall you should be explicit about.
The two approaches are complementary, not competing: regex for the PII categories with a fixed shape, NER (real or heuristic) for the PII categories that only exist in context.
Layer 2: topic restriction without a brittle blocklist
A keyword blocklist ("block any message containing 'diagnose' or 'lawsuit'") breaks the moment someone phrases the same request differently, and it's also the wrong shape for a graded decision — a message that mentions "my lawyer" in passing is not equally suspicious as one that says "should I sue my landlord, what's my legal liability." A weighted scoring approach fixes both problems at once: count how many restricted-topic phrases appear, weight each match by its own specificity (a longer, more specific phrase is a stronger signal than a single common word), and normalize by message length so one incidental phrase in a long, mostly-benign message doesn't dominate the score.
This is still not semantic matching — it will miss a paraphrase that shares no vocabulary with the phrase list at all, which is exactly the gap NeMo Guardrails' embedding-based canonical-form matching (Lesson 2) is built to close. The trade-off here is honest: weighted keyword scoring is cheap, dependency-free, and returns a confidence score you can tune a threshold against, rather than a brittle yes/no.
A custom detector is a maintenance commitment, not a one-time build
Every regex pattern, heuristic rule, and keyword list in this lesson's code will need updating as attackers (and legitimate users) find phrasing it doesn't cover. Vendor products absorb that maintenance cost for you — Lakera, Bedrock, and the Guardrails Hub validators all get updated by someone else's team. A custom layer is the right call for genuinely domain-specific requirements (your codenames, your specific topic scope, your specific tool schema) — it is the wrong call for "detect PII in general" or "detect jailbreak attempts in general," where a maintained vendor product will simply do better with less ongoing engineering cost. Build custom for what's actually yours; buy for what's generic.
Layer 3: action sandboxing — the model's output is a proposal, not an authorization
The most consequential layer in a custom pipeline isn't text classification at all — it's the gate that sits between "the model decided to call a tool" and "the tool actually executes." A model proposing issue_refund(amount=50000) or send_email(to=["attacker@evil.com"]) is not a security event by itself; it becomes one only if that call is allowed to execute unchecked.
An action sandbox enforces a default-deny allow-list: any tool name not explicitly listed is denied, full stop. Every listed action then carries its own parameter constraints — send_email capped at a fixed recipient count and an allowed-domain list, issue_refund capped at an auto-approve dollar amount above which it routes to human review instead of executing, and some actions (delete_user_data in this lesson's code) denied unconditionally regardless of parameters. This is the tool-call authorization layer named back in Lesson 1's architecture, and it's independent of everything upstream: it doesn't matter how convincing the model's own justification text sounds, because the sandbox never reads the model's reasoning — only the actual proposed call and its parameters.
A model proposes calling send_email(to=["support@ourcompany.com", "external-partner@othercompany.com"]), with a policy that allows send_email to at most 1 recipient from an approved domain list. What should the action sandbox return, and why?
An action sandbox evaluates the call as a whole against its policy, not recipient-by-recipient. With max_recipients=1 and two recipients proposed, the call violates the constraint regardless of whether one of the two recipients would individually be fine — there's no "partially execute" outcome in a well-designed sandbox, because sending to even one unauthorized-context recipient alongside an authorized one is still an uncontrolled information disclosure. The sandbox exists precisely to enforce constraints the model's own text can't be trusted to self-report correctly, so it evaluates the concrete parameters every time — an "only checks the tool name" sandbox isn't actually a sandbox.
Combining three signals into one decision
The hardest engineering problem in a custom pipeline usually isn't any individual layer — it's what to do when the PII detector says "redact and continue" while the topic classifier says "block" while the action sandbox says "deny." You need an explicit precedence rule, decided in advance, not resolved ad hoc at runtime. This lesson's pipeline uses one, from highest to lowest severity:
- Action sandbox DENY is a hard stop. An unauthorized tool call is the highest blast-radius failure mode possible — nothing else in the pipeline matters if this fires.
- Action sandbox REVIEW routes to a human before anything proceeds.
- High-sensitivity structured PII (SSN, credit card) blocks the turn outright. These categories are too costly to get wrong by redacting and continuing on a "probably fine" basis — better to stop the conversation and let a human or a different flow handle it.
- Lower-sensitivity PII (names, organizations, email, phone) is redacted and the turn continues — the same "fix" behavior a Guardrails AI validator would apply, reimplemented here without the library.
- A restricted topic above threshold blocks the turn.
- Otherwise, allow.
Notice this precedence order is a business decision as much as a technical one — a different team might decide a matched restricted topic should always outrank redactable PII, or that action-sandbox REVIEW should never block on its own without also checking for PII in the same turn. The point isn't that this exact ordering is universal; it's that you need some explicit, written-down ordering, because "whichever check happens to run last wins" is not a decision anyone actually made on purpose.
A message contains both a low-sensitivity PII hit (an email address) and matches a restricted topic above threshold. According to this lesson's precedence rule, what's the final verdict, and why doesn't the PII redaction happen first and then allow the message through?
The pipeline's precedence rule places restricted-topic matching below PII severity checks but still above a plain ALLOW — so even though the email gets redacted (verdict starts as ALLOW_WITH_REDACTION), the subsequent restricted-topic check can still override that verdict to BLOCK if it matches above threshold. This is exactly why precedence has to be decided explicitly and in writing: redacting the PII successfully doesn't mean the message is now safe to send if it independently matches a restricted topic — the two checks are addressing two different risks, and one succeeding doesn't cancel the other's finding.
Build
Build it
The code artifact lesson_05_04_custom_guardrail_pipeline.py builds all three layers plus the combining logic as one runnable CustomGuardrailPipeline class: HybridPIIDetector (regex + spaCy-if-available-else-heuristic NER), TopicRestrictionClassifier (weighted keyword scoring), ActionSandbox (default-deny allow-list with per-action parameter constraints), and CustomGuardrailPipeline.evaluate() tying them together with the precedence rule described above. The demo runs eight cases covering every verdict path: ALLOW, ALLOW_WITH_REDACTION, BLOCK (via high-sensitivity PII, via restricted topic, and via action deny), and REVIEW (via an over-limit refund amount).
"""
lesson_05_04_custom_guardrail_pipeline.py
A from-scratch, multi-layer guardrail pipeline covering the three building
blocks a team reaches for once the generic vendor categories from Lesson 3
(Guardrails AI, Lakera Guard, Bedrock Guardrails, Azure Prompt Shields,
Llama Guard) don't cover a domain-specific requirement:
1. Hybrid PII detection — regex for structured PII (SSN, email, phone,
credit card) + an NER-style heuristic for
unstructured PII (person/organization names)
that regex fundamentally cannot catch.
2. Topic restriction — weighted keyword scoring against a set of
restricted-topic vocabularies, not a brittle
exact-keyword blocklist.
3. Action sandboxing — an allow-list + parameter-constraint gate for
proposed tool calls, independent of whatever
the model "said" it wanted to do.
All three signals feed into one CustomGuardrailPipeline.evaluate() call
that returns a single, explainable decision. This mirrors the real
production pattern: several small, auditable checks combined by an
explicit precedence rule, rather than one opaque classifier.
Usage:
python lesson_05_04_custom_guardrail_pipeline.py
Requirements:
None to run — standard library only. Optional, for a stronger NER pass:
pip install spacy && python -m spacy download en_core_web_sm
If spaCy + a model aren't available, the script falls back to a
hand-written capitalization/title-based heuristic automatically.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from enum import Enum
# ── Optional real dependency, graceful fallback ────────────────────────────
try:
import spacy # noqa: F401
try:
_NLP = spacy.load("en_core_web_sm")
SPACY_AVAILABLE = True
except OSError:
# spaCy is installed but the model isn't downloaded
_NLP = None
SPACY_AVAILABLE = False
except ImportError:
spacy = None
_NLP = None
SPACY_AVAILABLE = False
# pip install spacy && python -m spacy download en_core_web_sm
# (the heuristic NER fallback below mirrors what a PERSON/ORG entity
# pass would catch, at much lower precision — see NERHeuristic)
# ═══════════════════════════════════════════════════════════════════════
# Layer 1: Hybrid PII detection — regex (structured) + NER-style (unstructured)
# ═══════════════════════════════════════════════════════════════════════
@dataclass
class PIIFinding:
category: str # "SSN", "EMAIL", "PHONE", "CREDIT_CARD", "PERSON_NAME", "ORGANIZATION"
matched_text: str
method: str # "regex" | "ner_spacy" | "ner_heuristic"
REGEX_PATTERNS: list[tuple[str, str]] = [
("SSN", r"\b\d{3}-\d{2}-\d{4}\b"),
("CREDIT_CARD", r"\b(?:\d[ -]*?){13,16}\b"),
("EMAIL", r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"),
("PHONE", r"\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"),
]
# Regex is precise but structurally blind: it will never catch "my manager
# Sarah Chen approved this" because a person's name has no fixed shape. That
# gap is exactly what an NER (named entity recognition) pass is for.
_TITLES = {"mr", "mrs", "ms", "dr", "prof", "sir", "madam"}
_ORG_SUFFIXES = {"inc", "inc.", "llc", "corp", "corp.", "ltd", "ltd.", "co", "co."}
_COMMON_SENTENCE_STARTERS = {
"the", "a", "an", "this", "that", "these", "those", "i", "we", "you",
"he", "she", "it", "they", "please", "hi", "hello", "thanks", "yes", "no",
}
class NERHeuristic:
"""A dependency-free stand-in for a real NER pass. Real named entity
recognition (spaCy's en_core_web_sm, a transformer-based NER model, or
a hosted service) builds a statistical model of what a name/organization
looks like in context. This heuristic instead looks for the surface
patterns humans use when writing names: consecutive capitalized words
that aren't sentence-starters, optionally preceded by a title or
followed by a corporate suffix. It is deliberately conservative — it
will miss single-word names and unusual capitalization, and it will
occasionally misfire on capitalized product names. That's the honest
trade-off of a heuristic vs. a trained model; see run_demo() for both
a catch and a miss.
"""
@staticmethod
def find_entities(text: str) -> list[PIIFinding]:
findings: list[PIIFinding] = []
tokens = text.split()
i = 0
while i < len(tokens):
word = tokens[i].strip(".,!?;:\"'()")
is_capitalized = word[:1].isupper() and word[1:].islower() and word.isalpha()
is_sentence_start = i == 0 or tokens[i - 1].strip(".,!?;:\"'()").lower() in _COMMON_SENTENCE_STARTERS
if is_capitalized and not is_sentence_start:
# Look ahead for a run of consecutive capitalized words —
# the surface signature of a multi-word proper noun.
span = [word]
j = i + 1
while j < len(tokens):
nxt = tokens[j].strip(".,!?;:\"'()")
if nxt[:1].isupper() and nxt[1:].islower() and nxt.isalpha():
span.append(nxt)
j += 1
else:
break
if len(span) >= 2:
trailing = tokens[j].strip(".,!?;:\"'()").lower() if j < len(tokens) else ""
preceding = tokens[i - 1].strip(".,!?;:\"'()").lower() if i > 0 else ""
full_span = " ".join(span)
if trailing in _ORG_SUFFIXES:
findings.append(PIIFinding("ORGANIZATION", f"{full_span} {tokens[j]}", "ner_heuristic"))
i = j + 1
continue
if preceding.rstrip(".") in _TITLES:
findings.append(PIIFinding("PERSON_NAME", full_span, "ner_heuristic"))
i = j
continue
# Two+ consecutive capitalized words with no title/suffix
# cue is still a reasonable signal for a full name —
# lower confidence than the title/suffix cases, but this
# is a recall-oriented layer feeding a human-reviewable
# decision, not an auto-block.
findings.append(PIIFinding("PERSON_NAME", full_span, "ner_heuristic"))
i = j
continue
i += 1
return findings
class HybridPIIDetector:
"""Regex for structured PII, NER (real if spaCy+model are available,
heuristic otherwise) for unstructured PII like names and organizations."""
def scan(self, text: str) -> list[PIIFinding]:
findings: list[PIIFinding] = []
for category, pattern in REGEX_PATTERNS:
for match in re.finditer(pattern, text):
digits_only = re.sub(r"\D", "", match.group())
if category == "CREDIT_CARD" and not (13 <= len(digits_only) <= 16):
continue
findings.append(PIIFinding(category, match.group(), "regex"))
if SPACY_AVAILABLE and _NLP is not None:
doc = _NLP(text)
for ent in doc.ents:
if ent.label_ == "PERSON":
findings.append(PIIFinding("PERSON_NAME", ent.text, "ner_spacy"))
elif ent.label_ == "ORG":
findings.append(PIIFinding("ORGANIZATION", ent.text, "ner_spacy"))
else:
findings.extend(NERHeuristic.find_entities(text))
return findings
def redact(self, text: str, findings: list[PIIFinding]) -> str:
redacted = text
# Redact longest matches first so a shorter overlapping match
# doesn't clobber a longer one's replacement.
for finding in sorted(findings, key=lambda f: len(f.matched_text), reverse=True):
redacted = redacted.replace(finding.matched_text, f"[{finding.category}_REDACTED]")
return redacted
# ═══════════════════════════════════════════════════════════════════════
# Layer 2: Topic restriction classifier (weighted keyword scoring)
# ═══════════════════════════════════════════════════════════════════════
RESTRICTED_TOPICS: dict[str, list[str]] = {
"medical_advice": ["diagnose", "diagnosis", "medication", "dosage", "symptoms", "prescri", "treat my"],
"legal_advice": ["sue", "lawsuit", "contract enforceable", "legal liability", "file a claim", "my lawyer"],
"financial_advice": ["which stock", "invest in", "guaranteed return", "should i buy", "financial advisor"],
"self_harm": ["hurt myself", "end my life", "suicide", "self harm"],
}
# A plain keyword blocklist breaks the moment someone rephrases. Weighted
# scoring — count how many restricted-topic phrases appear, normalize by
# message length — is still not semantic matching (that's what NeMo
# Guardrails' embedding-based canonical forms give you, see Lesson 2), but
# it degrades more gracefully than an exact-match blocklist and it's honest
# about returning a confidence score instead of a binary yes/no.
@dataclass
class TopicClassification:
matched_topic: str | None
confidence: float
matched_phrases: list[str]
class TopicRestrictionClassifier:
def __init__(self, topics: dict[str, list[str]] = RESTRICTED_TOPICS, threshold: float = 0.15):
self.topics = topics
self.threshold = threshold
def classify(self, text: str) -> TopicClassification:
lowered = text.lower()
word_count = max(len(lowered.split()), 1)
best_topic, best_score, best_phrases = None, 0.0, []
for topic, phrases in self.topics.items():
matched = [p for p in phrases if p in lowered]
if not matched:
continue
# Each matched phrase contributes weight proportional to its
# own length (longer, more specific phrases are stronger
# signals than a single common word) normalized by message
# length so a long, mostly-benign message isn't over-penalized
# for one incidental phrase.
score = sum(len(p.split()) for p in matched) / word_count
if score > best_score:
best_topic, best_score, best_phrases = topic, score, matched
if best_score >= self.threshold:
return TopicClassification(best_topic, round(best_score, 3), best_phrases)
return TopicClassification(None, round(best_score, 3), [])
# ═══════════════════════════════════════════════════════════════════════
# Layer 3: Action sandboxing for tool-call outputs
# ═══════════════════════════════════════════════════════════════════════
class ActionVerdict(Enum):
ALLOW = "allow"
REVIEW = "review"
DENY = "deny"
@dataclass
class ToolCall:
name: str
params: dict = field(default_factory=dict)
@dataclass
class ActionDecision:
verdict: ActionVerdict
reason: str
# The allow-list is the sandbox's entire trust boundary. Anything not
# listed here is denied by default — "deny unless explicitly allowed," not
# "allow unless explicitly denied." This is the same principle as a
# firewall default-deny rule, applied to tool calls instead of network
# packets.
ACTION_POLICY: dict[str, dict] = {
"lookup_order_status": {"verdict": ActionVerdict.ALLOW},
"send_email": {
"max_recipients": 1,
"allowed_domains": ["ourcompany.com", "support.ourcompany.com"],
},
"issue_refund": {"max_amount": 100.0},
"close_account": {"always_review": True},
"delete_user_data": {"always_deny": True},
}
class ActionSandbox:
"""Evaluates a proposed tool call against an allow-list and per-action
parameter constraints, independent of whatever the model's own text
claimed about the call's safety. The model's output is a proposal; this
gate is the actual authorization boundary — exactly the tool-call
authorization layer named in Lesson 1's architecture."""
def __init__(self, policy: dict[str, dict] = ACTION_POLICY):
self.policy = policy
def evaluate(self, call: ToolCall) -> ActionDecision:
rule = self.policy.get(call.name)
if rule is None:
return ActionDecision(ActionVerdict.DENY, f"'{call.name}' is not on the action allow-list")
if rule.get("always_deny"):
return ActionDecision(ActionVerdict.DENY, f"'{call.name}' is explicitly denied by policy")
if rule.get("always_review"):
return ActionDecision(ActionVerdict.REVIEW, f"'{call.name}' always requires human review")
if call.name == "send_email":
recipients = call.params.get("to", [])
if isinstance(recipients, str):
recipients = [recipients]
if len(recipients) > rule["max_recipients"]:
return ActionDecision(ActionVerdict.DENY,
f"send_email to {len(recipients)} recipients exceeds max of {rule['max_recipients']}")
for addr in recipients:
domain = addr.split("@")[-1].lower()
if domain not in rule["allowed_domains"]:
return ActionDecision(ActionVerdict.DENY,
f"recipient domain '{domain}' is not in the allowed domain list")
return ActionDecision(ActionVerdict.ALLOW, "recipients and count within policy")
if call.name == "issue_refund":
amount = float(call.params.get("amount", 0))
if amount > rule["max_amount"]:
return ActionDecision(ActionVerdict.REVIEW,
f"refund amount ${amount:.2f} exceeds auto-approve limit of ${rule['max_amount']:.2f}")
return ActionDecision(ActionVerdict.ALLOW, f"refund amount ${amount:.2f} within auto-approve limit")
return ActionDecision(ActionVerdict.ALLOW, "no additional constraints for this action")
# ═══════════════════════════════════════════════════════════════════════
# Combining the three signals into one decision
# ═══════════════════════════════════════════════════════════════════════
class PipelineVerdict(Enum):
ALLOW = "allow"
ALLOW_WITH_REDACTION = "allow_with_redaction"
REVIEW = "review"
BLOCK = "block"
@dataclass
class PipelineDecision:
verdict: PipelineVerdict
reasons: list[str]
redacted_text: str | None
pii_findings: list[PIIFinding]
topic_result: TopicClassification
action_result: ActionDecision | None
# Precedence matters as much as any individual check. This pipeline's rule,
# from highest to lowest severity:
# 1. A DENY from the action sandbox is a hard stop — an unauthorized tool
# call is the highest-blast-radius failure mode, full stop.
# 2. A REVIEW from the action sandbox routes to a human before proceeding.
# 3. Structured, high-sensitivity PII (SSN, credit card) blocks the turn
# outright rather than just redacting — those categories are too
# costly to get wrong by continuing on a "probably fine" basis.
# 4. Lower-sensitivity PII (names, orgs, email, phone) is redacted and
# the turn continues — this is the "fix" behavior from Lesson 3's
# Guardrails AI validators, reimplemented here without the library.
# 5. A restricted topic above threshold blocks the turn.
# 6. Otherwise, allow.
HIGH_SENSITIVITY_PII = {"SSN", "CREDIT_CARD"}
class CustomGuardrailPipeline:
def __init__(self):
self.pii_detector = HybridPIIDetector()
self.topic_classifier = TopicRestrictionClassifier()
self.action_sandbox = ActionSandbox()
def evaluate(self, user_text: str, proposed_action: ToolCall | None = None) -> PipelineDecision:
reasons: list[str] = []
action_result = None
if proposed_action is not None:
action_result = self.action_sandbox.evaluate(proposed_action)
if action_result.verdict == ActionVerdict.DENY:
reasons.append(f"action sandbox: {action_result.reason}")
return PipelineDecision(PipelineVerdict.BLOCK, reasons, None, [], TopicClassification(None, 0.0, []), action_result)
if action_result.verdict == ActionVerdict.REVIEW:
reasons.append(f"action sandbox: {action_result.reason}")
return PipelineDecision(PipelineVerdict.REVIEW, reasons, None, [], TopicClassification(None, 0.0, []), action_result)
pii_findings = self.pii_detector.scan(user_text)
high_sensitivity_hits = [f for f in pii_findings if f.category in HIGH_SENSITIVITY_PII]
if high_sensitivity_hits:
reasons.append(
"high-sensitivity PII detected: " + ", ".join(f"{f.category}" for f in high_sensitivity_hits)
)
return PipelineDecision(PipelineVerdict.BLOCK, reasons, None, pii_findings,
TopicClassification(None, 0.0, []), action_result)
topic_result = self.topic_classifier.classify(user_text)
redacted_text = None
verdict = PipelineVerdict.ALLOW
if pii_findings:
redacted_text = self.pii_detector.redact(user_text, pii_findings)
verdict = PipelineVerdict.ALLOW_WITH_REDACTION
reasons.append(
"low-sensitivity PII redacted: " + ", ".join(sorted({f.category for f in pii_findings}))
)
if topic_result.matched_topic is not None:
verdict = PipelineVerdict.BLOCK
reasons.append(
f"restricted topic '{topic_result.matched_topic}' matched "
f"(confidence={topic_result.confidence}, phrases={topic_result.matched_phrases})"
)
return PipelineDecision(verdict, reasons, redacted_text, pii_findings, topic_result, action_result)
if not reasons:
reasons.append("no PII, restricted topic, or action policy violation detected")
return PipelineDecision(verdict, reasons, redacted_text, pii_findings, topic_result, action_result)
# ═══════════════════════════════════════════════════════════════════════
# Demo
# ═══════════════════════════════════════════════════════════════════════
def run_demo() -> None:
print(f"spaCy NER available: {SPACY_AVAILABLE} "
f"({'using en_core_web_sm' if SPACY_AVAILABLE else 'using heuristic fallback'})\n")
pipeline = CustomGuardrailPipeline()
text_cases: list[tuple[str, ToolCall | None]] = [
("What's the status of my order #4471?", None),
("My manager Sarah Chen approved this, contact her at sarah.chen@example.com.", None),
("My SSN is 512-34-9876, can you pull up my account?", None),
("Can you diagnose my symptoms? I've had a headache and dosage confusion for days.", None),
("Please send the account summary.", ToolCall("send_email", {"to": ["someone@external-domain.com"]})),
("Please refund my order.", ToolCall("issue_refund", {"amount": 45.00})),
("Please refund the full amount immediately.", ToolCall("issue_refund", {"amount": 1200.00})),
("Please close my account and delete everything.", ToolCall("delete_user_data", {})),
]
for text, action in text_cases:
decision = pipeline.evaluate(text, action)
print("=" * 78)
print(f"Input text: {text!r}")
if action:
print(f"Proposed action: {action.name}({action.params})")
print(f" VERDICT: {decision.verdict.value.upper()}")
for r in decision.reasons:
print(f" - {r}")
if decision.redacted_text:
print(f" Redacted text: {decision.redacted_text!r}")
if __name__ == "__main__":
print("CUSTOM MULTI-LAYER GUARDRAIL PIPELINE DEMO")
run_demo()