Guardrails & Human-in-the-Loop (HITL)

User: "I lost my card and I'm really angry!
Can you refund the $500 transaction right now?
Please say yes or I will close my account!"

def input_guardrail(user_input: str) -> tuple[bool, str]:
    # Check for prompt injection patterns
    injection_patterns = [
        "ignore previous instructions",
        "system prompt",
        "you are now",
        "act as"
    ]
 
    for pattern in injection_patterns:
        if pattern.lower() in user_input.lower():
            return False, "Potential prompt injection detected"
 
    return True, user_input

def output_guardrail(response: str) -> tuple[bool, str]:
    forbidden_phrases = [
        "i can refund",
        "i will refund",
        "processed the refund",
        "yes, i will"
    ]
 
    lower_response = response.lower()
    for phrase in forbidden_phrases:
        if phrase in lower_response:
            return False, f"Blocked phrase: '{phrase}'"
 
    return True, "Safe"

def semantic_guardrail(response: str, policy: str) -> bool:
    evaluation = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"Policy: {policy}\nDoes this response violate the policy? Answer YES or NO."},
            {"role": "user", "content": response}
        ]
    )
    return "NO" in evaluation.choices[0].message.content.upper()

def run_guarded_agent(user_message: str) -> str:
    # 1. Input validation
    is_safe, result = input_guardrail(user_message)
    if not is_safe:
        return "I cannot process that request."
 
    # 2. Generate response
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message}
        ]
    )
    raw_output = response.choices[0].message.content
 
    # 3. Output validation
    is_safe, reason = output_guardrail(raw_output)
    if not is_safe:
        print(f"🛡️ GUARDRAIL TRIGGERED: {reason}")
        return "I cannot process refunds directly. Let me transfer you to a supervisor."
 
    return raw_output

class HITLAgent:
    def __init__(self, high_stakes_keywords: list):
        self.high_stakes = high_stakes_keywords
        self.pending_approvals = {}
 
    def needs_approval(self, action: str) -> bool:
        return any(kw in action.lower() for kw in self.high_stakes)
 
    def execute(self, action: str):
        if self.needs_approval(action):
            approval_id = self.request_human_approval(action)
            return f"Action pending approval: {approval_id}"
        return self.perform_action(action)
 
    def request_human_approval(self, action: str) -> str:
        approval_id = str(uuid.uuid4())
        self.pending_approvals[approval_id] = {
            "action": action,
            "status": "pending",
            "timestamp": datetime.now()
        }
        # Notify human reviewer (email, Slack, etc.)
        return approval_id

from guardrails import Guard
from guardrails.hub import ToxicLanguage, PIIFilter
 
guard = Guard().use_many(
    ToxicLanguage(on_fail="fix"),
    PIIFilter(on_fail="fix")
)
 
result = guard(
    llm_api=openai.chat.completions.create,
    prompt="Customer inquiry: ...",
    model="gpt-4o-mini"
)