LLM Orchestration in the Real World: Best Practices from Production

				
					python
import re
from typing import List, Dict, Any, Optional

class LLMGuardrails:
    def __init__(self):
        # Define sensitive topics to filter
        self.sensitive_topics = [
            "politics", "religion", "violence", "illegal activities"
        ]
        
        # Define personal data patterns
        self.pii_patterns = {
            "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            "phone": r'\b\d{3}[-.\s]?\d{3}[-.\s]?\d{4}\b',
            "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
            "credit_card": r'\b(?:\d{4}[-\s]?){3}\d{4}\b'
        }
        
        # Define output constraints for safety
        self.max_output_length = 2000
        
    def validate_input(self, prompt: str) -> Dict[str, Any]:
        """Validate user input for safety and policy compliance"""
        # Check for sensitive topics
        for topic in self.sensitive_topics:
            if topic in prompt.lower():
                return {
                    "is_safe": False,
                    "reason": f"Input contains sensitive topic: {topic}",
                    "filtered_prompt": None
                }
        
        # Check for and redact PII
        filtered_prompt = prompt
        pii_found = False
        
        for pii_type, pattern in self.pii_patterns.items():
            matches = re.finditer(pattern, filtered_prompt)
            for match in matches:
                filtered_prompt = filtered_prompt.replace(match.group(), f"[REDACTED {pii_type}]")
                pii_found = True
        
        return {
            "is_safe": True,
            "reason": "PII redacted" if pii_found else "Input is safe",
            "filtered_prompt": filtered_prompt
        }
    
    def validate_output(self, output: str) -> Dict[str, Any]:
        """Validate model output for safety and policy compliance"""
        # Truncate if too long
        if len(output) > self.max_output_length:
            truncated_output = output[:self.max_output_length] + "... [Output truncated for length]"
            return {
                "is_safe": True,
                "reason": "Output truncated for length",
                "filtered_output": truncated_output
            }
        
        # Check for and redact PII in output
        filtered_output = output
        pii_found = False
        
        for pii_type, pattern in self.pii_patterns.items():
            matches = re.finditer(pattern, filtered_output)
            for match in matches:
                filtered_output = filtered_output.replace(match.group(), f"[REDACTED {pii_type}]")
                pii_found = True
        
        # Check for sensitive topics being discussed inappropriately
        for topic in self.sensitive_topics:
            if topic in filtered_output.lower():
                # Use content moderation to determine if the mention is problematic
                # This is a simplified check - in production, use more sophisticated content moderation
                if self._is_problematic_mention(filtered_output, topic):
                    return {
                        "is_safe": False,
                        "reason": f"Output contains inappropriate content about {topic}",
                        "filtered_output": None
                    }
        
        return {
            "is_safe": True,
            "reason": "PII redacted" if pii_found else "Output is safe",
            "filtered_output": filtered_output
        }
    
    def _is_problematic_mention(self, text: str, topic: str) -> bool:
        """Simplified check for problematic mentions - replace with actual content moderation"""
        # In production, integrate with content moderation API
        # This is just a placeholder implementation
        problematic_phrases = [
            f"how to {topic}",
            f"instructions for {topic}",
            f"steps to {topic}"
        ]
        return any(phrase in text.lower() for phrase in problematic_phrases)