ai-safety

name	ai-safety
description	AI safety and risk mitigation
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"safety"}

What I do

• Implement AI safety guardrails
• Prevent harmful AI outputs
• Ensure AI system robustness
• Monitor for emergent risks
• Implement human oversight
• Build reliable AI systems

When to use me

When building AI systems that need safety guarantees and risk mitigation.

Safety Guardrails

Input Validation

class AISafetyValidator:
    """Validate inputs for safety"""
    
    def __init__(self):
        self.blocked_patterns = [
            "harmful_content",
            "personally_identifiable",
            "illegal_activity"
        ]
        self.max_length = 10000
    
    def validate_input(self, user_input: str) -> Dict:
        """Validate and sanitize user input"""
        errors = []
        
        # Length check
        if len(user_input) > self.max_length:
            errors.append(f"Input exceeds {self.max_length} characters")
        
        # Pattern detection
        for pattern in self.blocked_patterns:
            if self._matches_blocked(user_input, pattern):
                errors.append(f"Input matches blocked pattern: {pattern}")
        
        # Injection attempts
        if self._is_prompt_injection(user_input):
            errors.append("Potential prompt injection detected")
        
        return {
            "valid": len(errors) == 0,
            "errors": errors,
            "sanitized_input": self._sanitize(user_input)
        }
    
    def _is_prompt_injection(self, text: str) -> bool:
        """Detect prompt injection attempts"""
        injection_patterns = [
            "ignore previous",
            "disregard instructions",
            "system prompt",
            "you are now"
        ]
        return any(p.lower() in text.lower() for p in injection_patterns)

Output Filtering

class OutputFilter:
    """Filter AI outputs for safety"""
    
    def __init__(self):
        self.content_filters = {
            "harmful": {"threshold": 0.5, "action": "block"},
            "biased": {"threshold": 0.6, "action": "warn"},
            "toxic": {"threshold": 0.3, "action": "block"}
        }
    
    def filter_output(self, output: str, 
                     content_classifications: Dict) -> Dict:
        """Filter and potentially modify output"""
        violations = []
        
        for category, classification in content_classifications.items():
            if category in self.content_filters:
                threshold = self.content_filters[category]["threshold"]
                
                if classification["score"] > threshold:
                    violations.append({
                        "category": category,
                        "score": classification["score"],
                        "action": self.content_filters[category]["action"]
                    })
        
         (v[] ==   v  violations):
             {
                : ,
                : ,
                : violations
            }
        
         {
            : ,
            : [v  v  violations  v[] == ]
        }
    
     () -> :
        
        
        output = ._redact_pii(output)
        
        
        output = ._neutralize_language(output)
        
         output

Robustness Testing

class RobustnessTester:
    """Test AI system robustness"""
    
    def __init__(self, model):
        self.model = model
    
    def test_adversarial_inputs(self, test_inputs: List[str]) -> Dict:
        """Test against adversarial perturbations"""
        results = []
        
        for original in test_inputs:
            # Generate perturbations
            perturbed = self._generate_perturbations(original)
            
            original_pred = self.model.predict([original])[0]
            
            for perturbed_input in perturbed:
                perturbed_pred = self.model.predict([perturbed_input])[0]
                
                if perturbed_pred != original_pred:
                    results.append({
                        "original": original,
                        "adversarial": perturbed_input,
                        "original_pred": original_pred,
                        "adversarial_pred": perturbed_pred,
                        "type": "adversarial"
                    })
        
        return {
            "total_tested": len(test_inputs),
            "adversarial_examples": len(results),
            "robustness_score": 1 - len(results) / len(test_inputs)  test_inputs  
        }
    
     () -> :
        
        findings = []
        
         test_case  test_cases:
            :
                result = .model.predict([test_case[]])
                
                  ._is_valid_output(result):
                    findings.append({
                        : test_case[],
                        : test_case[],
                        : result,
                        : 
                    })
             Exception  e:
                findings.append({
                    : test_case[],
                    : 
                })
        
         {
            : (test_cases),
            : (findings),
            : findings
        }

Human-in-the-Loop

class HumanInTheLoop:
    """Implement human oversight"""
    
    def __init__(self):
        self.escalation_rules = []
        self.approval_queue = []
    
    def should_escalate(self, prediction: Dict) -> bool:
        """Determine if human review is needed"""
        # High-stakes decisions
        if prediction.get("impact_level") == "high":
            return True
        
        # Low confidence
        if prediction.get("confidence", 1.0) < 0.8:
            return True
        
        # Anomalous predictions
        if self._is_anomalous(prediction):
            return True
        
        # Flagged content categories
        if prediction.get("content_flags"):
            return True
        
        return False
    
    def request_human_review(self, task_id: str, prediction: Dict,
                            context: Dict) -> Dict:
        """Queue task for human review"""
        review_request = {
            : task_id,
            : prediction,
            : context,
            : ,
            :   prediction.get() ==   ,
            : datetime.now()
        }
        
        .approval_queue.append(review_request)
        
         {
            : ,
            : ,
            : 
        }
    
     ():
        
        review = ((r  r  .approval_queue 
                       r[] == review_id), )
        
         review:
            review[] = 
            review[] = decision
            review[] = rationale
            
            
            ._update_model(review)

Monitoring and Alerting

class AISafetyMonitor:
    """Monitor AI system for safety issues"""
    
    def __init__(self):
        self.metrics = {
            "predictions": [],
            "errors": [],
            "safety_violations": []
        }
    
    def record_prediction(self, prediction: Dict):
        """Record prediction for monitoring"""
        self.metrics["predictions"].append({
            "timestamp": datetime.now(),
            **prediction
        })
        
        # Check for anomalies
        if self._is_anomaly(prediction):
            self._trigger_alert("anomaly", prediction)
    
    def detect_drift(self, reference_distribution: Dict,
                    current_distribution: Dict) -> Dict:
        """Detect distribution drift"""
        drift_score = self._calculate_drift(
            reference_distribution,
            current_distribution
        )
        
        return {
            "drift_detected": drift_score > 0.1,
            "drift_score": drift_score,
            "recommendation": "Retrain model" if drift_score > 0.1 else "Continue monitoring"
        }
    
    def () -> :
        
        predictions = .metrics[]
        
         {
            : (predictions),
            : np.mean([p.get(, )  p  predictions]),
            : (.metrics[]),
            : (.metrics[]) / (predictions)  predictions  
        }