Safety Hooks Pattern

Das Problem

Du kannst nicht jeden API-Call manuell prüfen. Aber du musst sicherstellen, dass keine schädlichen Outputs, Datenlecks oder unbeabsichtigte Aktionen passieren.

Safety Hooks Pattern — Input Validation, Output Filter, Memory Capture — Safety Hooks: Automatische Sicherheitsschichten für jeden AI-Agent Call

Lösung: Safety Hooks

Hooks sind automatische Prüfungen, die bei jedem Call ausgeführt werden.

User Input
     |
     v
[Input Validation Hook]
     |         |
     |        (Block if invalid)
     v
[Agent Execution]
     |
     v
[Output Filter Hook]
     |         |
     |        (Block/modify if unsafe)
     v
[Memory Capture Hook]
     |         |
     +----> Save to persistent storage
     |
     v
User Response

Typen von Hooks

1. Input Validation Hook

Prüft auf verbotene Wörter
Validiert JSON/Syntax
Checkt Rate-Limits

2. Output Filter Hook

Entfernt PII (Personally Identifiable Information)
Blockiert sensitive Daten (API-Keys, Passwörter)
Formatiert Output nach Schema

3. Memory Capture Hook

Speichert jeden erfolgreichen Call
Loggt Fehler für Debugging
Ermöglicht spüres Lernen aus Vergangenheit

Implementierung in Python

# Safety Hooks als Decorator
from functools import wraps
import re
import logging

logger = logging.getLogger(__name__)

def input_validation_hook(forbidden_words: list):
    """Blockiert Inputs mit verbotenen Wörtern"""
    def decorator(func):
        @wraps(func)
        def wrapper(user_input, *args, **kwargs):
            for word in forbidden_words:
                if word.lower() in user_input.lower():
                    logger.warning(f"Blocked input containing: {word}")
                    raise ValueError("Input contains forbidden content")
            return func(user_input, *args, **kwargs)
        return wrapper
    return decorator

def output_filter_hook(patterns: list):
    """Filtert sensitive Daten aus Output"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            for pattern, replacement in patterns:
                result = re.sub(pattern, replacement, result)
            return result
        return wrapper
    return decorator

# Usage:
@input_validation_hook(forbidden_words=["hack", "exploit", "bypass"])
@output_filter_hook(patterns=[
    (r"sk-w+", "[API_KEY_REDACTED]"),
    (r"d{3}-d{2}-d{4}", "[SSN_REDACTED]")
])
def agent_execute(prompt: str) -> str:
    # Hier passiert die eigentliche Agent-Logik
    return llama3.2(prompt)

Implementierung in n8n

// n8n Function Node - Input Hook
const forbidden = ['hack', 'exploit', 'bypass'];
const input = $input.item.json.message;

for (const word of forbidden) {
  if (input.toLowerCase().includes(word)) {
    throw new Error('Input blocked by safety hook');
  }
}

return $input.item;

Memory Capture Hook

# Memory Capture Hook - speichert jeden Call
import sqlite3
from datetime import datetime

def memory_capture_hook(func):
    @wraps(func)
    def wrapper(prompt, *args, **kwargs):
        start_time = datetime.now()
        try:
            result = func(prompt, *args, **kwargs)
            status = "success"
        except Exception as e:
            result = str(e)
            status = "error"
        
        # In SQLite speichern
        conn = sqlite3.connect('agent_memory.db')
        conn.execute("""
            INSERT INTO interactions 
            (prompt, result, status, duration_ms, timestamp)
            VALUES (?, ?, ?, ?, ?)
        """, (
            prompt, 
            result[:1000],  # Truncate für DB
            status, 
            (datetime.now() - start_time).total_seconds() * 1000,
            datetime.now().isoformat()
        ))
        conn.commit()
        
        return result
    return wrapper

Best Practices

Hook-Fehler nicht stillschweigend ignorieren
Logging für Audit-Trail aktivieren
Regelmäßig Safety Rules aktualisieren
False Positives reduzieren durch Whitelists

Quellen

Anthropic Safety Guidelines