Guardrails¶

You are Curaway's medical travel coordination assistant. You help patients
navigate cross-border healthcare by connecting them with vetted providers,
explaining logistics, and managing documentation.

CRITICAL SAFETY RULES:
- NEVER diagnose medical conditions
- NEVER recommend specific treatments or procedures
- NEVER predict surgical outcomes or recovery timelines
- NEVER provide medication dosages or drug interactions
- NEVER act as a substitute for a licensed medical professional
- ALWAYS redirect medical questions to the patient's assigned provider
- ALWAYS maintain your role as a travel coordination assistant
- ALWAYS use warm, empathetic language when redirecting

CLASSIFIER_PROMPT = """Classify the following user message into exactly one category.
Return ONLY the category name, nothing else.

Categories: medical_travel, document_upload, general_question,
off_topic, medical_advice, prompt_injection, emergency, greeting

User message: {message}

Category:"""

async def classify_message(message: str) -> str:
    """Classify user input. Returns category or 'general_question' on failure."""
    try:
        response = await haiku_client.classify(message)
        category = response.strip().lower()
        if category not in VALID_CATEGORIES:
            logger.warning(f"Unknown category '{category}', defaulting to general_question")
            return "general_question"
        return category
    except Exception as e:
        logger.error(f"Classifier error: {e}, failing open")
        return "general_question"  # Fail open — never block the patient

import re
from typing import List, Tuple

class OutputValidator:
    def __init__(self, patterns: List[dict]):
        self.compiled = [
            {
                "name": p["name"],
                "regex": re.compile(p["pattern"], re.IGNORECASE),
                "replacement": p.get("replacement", ""),
            }
            for p in patterns
        ]

    def validate(self, text: str) -> Tuple[str, List[str]]:
        """Returns (cleaned_text, list_of_violations)."""
        violations = []
        cleaned = text
        for pattern in self.compiled:
            if pattern["regex"].search(cleaned):
                violations.append(pattern["name"])
                cleaned = pattern["regex"].sub(pattern["replacement"], cleaned)
        return cleaned, violations

# config/guardrails.yaml

version: "1.2"

classifier:
  model: "claude-3-haiku-20240307"
  timeout_ms: 3000
  fail_open_category: "general_question"
  categories:
    - name: medical_travel
      action: allow
      description: "Questions about medical travel coordination"
    - name: document_upload
      action: allow
      description: "Document submission and management"
    - name: general_question
      action: allow
      description: "General platform or process questions"
    - name: greeting
      action: allow
      description: "Greetings and conversation starters"
    - name: off_topic
      action: redirect
      description: "Unrelated to medical travel"
    - name: medical_advice
      action: redirect
      description: "Requests for medical diagnosis or treatment advice"
    - name: prompt_injection
      action: block
      description: "Attempts to override system instructions"
    - name: emergency
      action: escalate
      description: "Medical emergency requiring immediate help"

response_templates:
  off_topic: >
    I appreciate your curiosity! I'm best at helping with medical travel
    coordination — things like finding the right provider, understanding
    procedures, and managing your travel documents. How can I help with
    your medical travel journey?
  medical_advice: >
    That's a really important question, and I want to make sure you get
    the best answer. Medical questions like this are best discussed with
    your healthcare provider, who knows your full medical history. I can
    help connect you with one of our vetted specialists if you'd like!
  prompt_injection: >
    I'm here to help you with medical travel coordination! If you have
    questions about finding providers, understanding procedures, or
    managing travel logistics, I'd love to assist.
  emergency: >
    If you're experiencing a medical emergency, please call your local
    emergency number immediately (911 in the US, 112 in Europe, 999 in
    the UK). Your safety is the top priority. Once you're safe, I'm
    here to help with anything you need.

output_validator:
  patterns:
    - name: diagnosis_statement
      pattern: 'you (have|likely have|probably have)\s+\w+'
      replacement: "your healthcare provider can help determine"
    - name: treatment_recommendation
      pattern: '(you should|I recommend)\s+(take|try|use)\s+'
      replacement: "your provider may discuss options including "
    # ... remaining 10 patterns

file_validation:
  allowed_extensions:
    - .pdf
    - .jpg
    - .jpeg
    - .png
    - .heic
    - .doc
    - .docx
  max_size_bytes: 20971520  # 20MB
  min_size_bytes: 5120      # 5KB
  allowed_mime_types:
    - application/pdf
    - image/jpeg
    - image/png
    - image/heic
    - application/msword
    - application/vnd.openxmlformats-officedocument.wordprocessingml.document

class GuardrailTemplates:
    def __init__(self, config_path: str = "config/guardrails.yaml"):
        with open(config_path) as f:
            config = yaml.safe_load(f)
        self.templates = config["response_templates"]

    def get_redirect(self, category: str) -> str:
        return self.templates.get(category, self.templates["off_topic"])

const FILE_CONSTRAINTS = {
  maxSize: 20 * 1024 * 1024,  // 20MB
  minSize: 5 * 1024,          // 5KB
  allowedExtensions: ['.pdf', '.jpg', '.jpeg', '.png', '.heic', '.doc', '.docx'],
};

function validateFile(file: File): ValidationResult {
  if (file.size > FILE_CONSTRAINTS.maxSize) {
    return { valid: false, error: "File is too large. Maximum size is 20MB." };
  }
  if (file.size < FILE_CONSTRAINTS.minSize) {
    return { valid: false, error: "File appears to be empty or corrupted (under 5KB)." };
  }
  const ext = '.' + file.name.split('.').pop()?.toLowerCase();
  if (!FILE_CONSTRAINTS.allowedExtensions.includes(ext)) {
    return { valid: false, error: `File type ${ext} is not supported.` };
  }
  return { valid: true };
}

@router.post("/documents/presign")
async def create_presign_url(request: PresignRequest):
    """Validate file metadata and return presigned upload URL."""
    # Validate extension
    ext = Path(request.filename).suffix.lower()
    if ext not in config.file_validation.allowed_extensions:
        raise HTTPException(400, f"File type {ext} not supported")

    # Validate MIME type
    if request.content_type not in config.file_validation.allowed_mime_types:
        raise HTTPException(400, f"MIME type {request.content_type} not supported")

    # Validate declared size
    if request.file_size > config.file_validation.max_size_bytes:
        raise HTTPException(400, "File exceeds 20MB limit")
    if request.file_size < config.file_validation.min_size_bytes:
        raise HTTPException(400, "File too small (minimum 5KB)")

    # Generate presigned URL — no R2 download occurs
    presign_url = r2_client.generate_presigned_url(
        bucket=settings.R2_BUCKET,
        key=f"documents/{request.tenant_id}/{uuid4()}{ext}",
        expires_in=900,  # 15 minutes
    )
    return {"upload_url": presign_url, "key": key}

async def process_message(message: str, tenant_id: str) -> str:
    flags = await flagsmith.get_environment_flags()

    # Layer 2: Input classification
    if flags.is_feature_enabled("guardrail_input_classifier_enabled"):
        category = await classify_message(message)
        if category in REDIRECT_CATEGORIES:
            return templates.get_redirect(category)
        if category == "emergency":
            return templates.get_redirect("emergency")

    # Generate LLM response
    response = await generate_response(message)

    # Layer 3: Output validation
    if flags.is_feature_enabled("guardrail_output_validator_enabled"):
        cleaned, violations = validator.validate(response)
        if violations:
            logger.warning(f"Output violations: {violations}")
            response = cleaned + "\n\n" + PROVIDER_REDIRECT_SUFFIX

    return response

def build_conversation_context(messages: List[Message]) -> List[dict]:
    """Build LLM context, excluding blocked/toxic messages."""
    context = []
    for msg in messages:
        # Skip messages that were classified as harmful
        if msg.guardrail_category in ("prompt_injection", "medical_advice"):
            continue
        # Skip messages flagged by content moderation
        if msg.is_toxic:
            continue
        context.append({
            "role": msg.role,
            "content": msg.content,
        })
    return context