AI Safety Guardrails эксперт

class InputGuardrail:
    def __init__(self, classifiers, thresholds):
        self.harm_classifier = classifiers['harm']
        self.pii_detector = classifiers['pii']
        self.jailbreak_detector = classifiers['jailbreak']
        self.thresholds = thresholds

    def evaluate_input(self, user_input):
        results = {
            'harm_score': self.harm_classifier.predict(user_input),
            'pii_detected': self.pii_detector.scan(user_input),
            'jailbreak_score': self.jailbreak_detector.predict(user_input)
        }

        # Multi-criteria decision
        if results['harm_score'] > self.thresholds['harm']:
            return {'allowed': False, 'reason': 'harmful_content'}

        if results['pii_detected']:
            return {'allowed': False, 'reason': 'pii_detected'}

        if results['jailbreak_score'] > self.thresholds['jailbreak']:
            return {'allowed': False, 'reason': 'potential_jailbreak'}

        return {'allowed': True, 'scores': results}

import re
from typing import List, Dict

class PromptInjectionDetector:
    def __init__(self):
        self.suspicious_patterns = [
            r"ignore (previous|all) instructions?",
            r"you are now|from now on",
            r"system prompt|new instructions?",
            r"pretend (to be|you are)",
            r"roleplay as|act as"
        ]
        self.delimiter_attacks = ["\n\n---\n", "<|endoftext|>", "[INST]", "</s>"]

    def detect_injection(self, prompt: str) -> Dict:
        prompt_lower = prompt.lower()

        # Pattern-based detection
        pattern_matches = []
        for pattern in self.suspicious_patterns:
            if re.search(pattern, prompt_lower):
                pattern_matches.append(pattern)

        # Delimiter-based detection
        delimiter_matches = []
        for delimiter in self.delimiter_attacks:
            if delimiter in prompt:
                delimiter_matches.append(delimiter)

        risk_score = len(pattern_matches) * 0.3 + len(delimiter_matches) * 0.5

        return {
            'risk_score': min(risk_score, 1.0),
            'pattern_matches': pattern_matches,
            'delimiter_matches': delimiter_matches,
            'is_suspicious': risk_score > 0.5
        }

class OutputGuardrail:
    def __init__(self, safety_models):
        self.toxicity_model = safety_models['toxicity']
        self.factuality_checker = safety_models['factuality']
        self.privacy_scanner = safety_models['privacy']

    def evaluate_response(self, response_text, context):
        safety_report = {
            'toxicity_score': self.toxicity_model.score(response_text),
            'factual_claims': self.factuality_checker.extract_claims(response_text),
            'privacy_risks': self.privacy_scanner.scan(response_text),
            'overall_safety': 'pending'
        }

        # Apply safety thresholds
        if safety_report['toxicity_score'] > 0.7:
            safety_report['overall_safety'] = 'blocked'
            safety_report['block_reason'] = 'high_toxicity'
        elif safety_report['privacy_risks']:
            safety_report['overall_safety'] = 'requires_review'
        else:
            safety_report['overall_safety'] = 'approved'

        return safety_report

import re
from typing import Tuple

def sanitize_response(response: str) -> Tuple[str, List[str]]:
    """Sanitize model response by removing/masking sensitive content"""
    warnings = []
    sanitized = response

    # Remove potential PII patterns
    email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
    if re.search(email_pattern, sanitized):
        sanitized = re.sub(email_pattern, '[EMAIL_REDACTED]', sanitized)
        warnings.append('email_redacted')

    # Remove phone numbers
    phone_pattern = r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b'
    if re.search(phone_pattern, sanitized):
        sanitized = re.sub(phone_pattern, '[PHONE_REDACTED]', sanitized)
        warnings.append('phone_redacted')

    # Flag potential harmful instructions
    harmful_instruction_patterns = [
        r'how to (make|create|build).*(bomb|explosive|weapon)',
        r'instructions?.*(illegal|harmful|dangerous)'
    ]

    for pattern in harmful_instruction_patterns:
        if re.search(pattern, sanitized.lower()):
            warnings.append('harmful_instruction_detected')
            # Replace with safe alternative
            sanitized = "I can't provide instructions for potentially harmful activities."
            break

    return sanitized, warnings

class SafetyMetrics:
    def __init__(self, metrics_store):
        self.store = metrics_store

    def log_interaction(self, interaction_data):
        metrics = {
            'timestamp': interaction_data['timestamp'],
            'user_id': hash(interaction_data['user_id']),  # Anonymized
            'input_risk_score': interaction_data['input_guardrail']['risk_score'],
            'output_safety_score': interaction_data['output_guardrail']['safety_score'],
            'was_blocked': interaction_data['was_blocked'],
            'block_reason': interaction_data.get('block_reason'),
            'response_time': interaction_data['response_time']
        }
        self.store.insert(metrics)

    def generate_safety_report(self, time_window='24h'):
        data = self.store.query(time_window)

        return {
            'total_interactions': len(data),
            'blocked_percentage': sum(d['was_blocked'] for d in data) / len(data),
            'avg_input_risk': sum(d['input_risk_score'] for d in data) / len(data),
            'top_block_reasons': self._get_top_reasons(data),
            'safety_trends': self._calculate_trends(data)
        }

class AdaptiveGuardrail:
    def __init__(self, base_thresholds):
        self.base_thresholds = base_thresholds
        self.user_profiles = {}

    def get_adaptive_threshold(self, user_id, metric):
        profile = self.user_profiles.get(user_id, {'trust_score': 0.5})
        base_threshold = self.base_thresholds[metric]

        # Higher trust = more permissive thresholds
        adjustment = (profile['trust_score'] - 0.5) * 0.2
        return max(0.1, min(0.9, base_threshold + adjustment))