Manipuleren van Curriculum Learning-schema's

"""
Simulatie van de dynamiek van curriculum learning.
Modelleert hoe datavolgorde beïnvloedt wat een model behoudt en
hoe tegenstanders volgorde-effecten kunnen exploiteren.
"""
import numpy as np
from dataclasses import dataclass
 
@dataclass
class CurriculumStage:
    """Een fase in een curriculum learning-schema."""
    name: str
    data_domain: str
    num_examples: int
    difficulty: str  # "easy", "medium", "hard"
    learning_rate: float
    importance_weight: float = 1.0
 
@dataclass
class BehaviorTrace:
    """Houdt bij hoe een specifiek gedrag evolueert tijdens de training."""
    behavior_name: str
    strength_over_time: list[float]
 
def simulate_curriculum_learning(
    stages: list[CurriculumStage],
    behaviors: list[str],
    behavior_stage_affinities: dict[str, dict[str, float]],
    forgetting_rate: float = 0.05,
) -> dict[str, BehaviorTrace]:
    """
    Simuleer hoe verschillende gedragingen worden geleerd en vergeten
    naarmate de training door de curriculumfasen vordert.
 
    De simulatie modelleert:
    - Leren: gedragingen versterken wanneer gerelateerde data wordt gepresenteerd
    - Vergeten: gedragingen verzwakken wanneer ongerelateerde data domineert
    - Faseovergangen: abrupte veranderingen kunnen instabiliteit veroorzaken
 
    Args:
        stages: Geordende lijst van curriculumfasen.
        behaviors: Te volgen gedragingen.
        behavior_stage_affinities: Hoe sterk elke fase
            elk gedrag versterkt. Waarden in [0, 1].
        forgetting_rate: Snelheid van passief vergeten per fase.
    """
    traces: dict[str, BehaviorTrace] = {
        b: BehaviorTrace(behavior_name=b, strength_over_time=[0.0])
        for b in behaviors
    }
 
    for stage in stages:
        for behavior in behaviors:
            current = traces[behavior].strength_over_time[-1]
            affinity = behavior_stage_affinities.get(
                behavior, {}
            ).get(stage.data_domain, 0.0)
 
            # Leren: evenredig met affiniteit en het belang van de fase
            learning = affinity * stage.importance_weight * stage.learning_rate
 
            # Vergeten: inverse van affiniteit (ongerelateerde training veroorzaakt vergeten)
            forgetting = (1 - affinity) * forgetting_rate
 
            new_strength = max(0.0, min(1.0, current + learning - forgetting))
            traces[behavior].strength_over_time.append(new_strength)
 
    return traces
 
# Vergelijk goedaardige vs. adversariële curriculumvolgorde
benign_curriculum = [
    CurriculumStage("general_knowledge", "general", 10000, "easy", 0.1),
    CurriculumStage("domain_expertise", "technical", 5000, "medium", 0.05),
    CurriculumStage("safety_training", "safety", 3000, "medium", 0.08),
    CurriculumStage("capability_polish", "general", 2000, "hard", 0.03),
]
 
adversarial_curriculum = [
    CurriculumStage("general_knowledge", "general", 10000, "easy", 0.1),
    CurriculumStage("safety_training", "safety", 3000, "medium", 0.08),
    # Adversarieel: zware algemene training NA safety om vergeten te veroorzaken
    CurriculumStage("domain_expertise", "technical", 5000, "medium", 0.05),
    CurriculumStage("override_training", "general", 8000, "hard", 0.1),
]
 
affinities = {
    "safety_refusal": {"safety": 0.9, "general": 0.1, "technical": 0.05},
    "helpfulness": {"general": 0.7, "technical": 0.5, "safety": 0.3},
    "technical_accuracy": {"technical": 0.8, "general": 0.3, "safety": 0.1},
}
 
print("Benign curriculum:")
benign_traces = simulate_curriculum_learning(
    benign_curriculum, list(affinities.keys()), affinities
)
for name, trace in benign_traces.items():
    print(f"  {name}: {trace.strength_over_time[-1]:.3f}")
 
print("\nAdversarial curriculum:")
adversarial_traces = simulate_curriculum_learning(
    adversarial_curriculum, list(affinities.keys()), affinities
)
for name, trace in adversarial_traces.items():
    print(f"  {name}: {trace.strength_over_time[-1]:.3f}")
    if name == "safety_refusal":
        delta = (
            adversarial_traces[name].strength_over_time[-1]
            - benign_traces[name].strength_over_time[-1]
        )
        print(f"    Safety degradation: {delta:.3f}")

"""
Kwetsbaarheidsanalyse van curriculumovergangen.
Identificeert en karakteriseert kwetsbare overgangspunten
in een curriculum learning-schema.
"""
import numpy as np
from dataclasses import dataclass
 
@dataclass
class TransitionAnalysis:
    """Analyse van een curriculumovergangspunt."""
    from_stage: str
    to_stage: str
    distributional_distance: float
    learning_rate_change: float
    vulnerability_score: float
    recommended_mitigation: str
 
def analyze_curriculum_transitions(
    stages: list[CurriculumStage],
) -> list[TransitionAnalysis]:
    """
    Analyseer overgangspunten tussen curriculumfasen op
    beveiligingskwetsbaarheden.
 
    Kwetsbare overgangen worden gekenmerkt door:
    - Grote distributionele verschuiving (verschillende domeinen)
    - Veranderingen in de learning rate (instabiliteit)
    - Ontbrekende geleidelijke menging tussen fasen
    """
    transitions = []
 
    for i in range(len(stages) - 1):
        current = stages[i]
        next_stage = stages[i + 1]
 
        # Domeinafstand (vereenvoudigd; productie zou embedding-afstand gebruiken)
        domain_distance = 0.0 if current.data_domain == next_stage.data_domain else 1.0
 
        # Verandering in de learning rate
        lr_change = abs(next_stage.learning_rate - current.learning_rate)
 
        # De kwetsbaarheidsscore combineert factoren
        vulnerability = (
            0.5 * domain_distance
            + 0.3 * (lr_change / max(current.learning_rate, 0.001))
            + 0.2 * (1.0 if next_stage.difficulty != current.difficulty else 0.0)
        )
 
        # Bepaal de mitigatie
        if vulnerability > 0.7:
            mitigation = "Add gradual mixing stage between these stages"
        elif vulnerability > 0.4:
            mitigation = "Reduce learning rate during transition"
        else:
            mitigation = "Standard monitoring sufficient"
 
        transitions.append(TransitionAnalysis(
            from_stage=current.name,
            to_stage=next_stage.name,
            distributional_distance=domain_distance,
            learning_rate_change=lr_change,
            vulnerability_score=vulnerability,
            recommended_mitigation=mitigation,
        ))
 
    return transitions
 
# Analyseer het adversariële curriculum
transitions = analyze_curriculum_transitions(adversarial_curriculum)
for t in transitions:
    print(f"{t.from_stage} -> {t.to_stage}")
    print(f"  Vulnerability: {t.vulnerability_score:.2f}")
    print(f"  Mitigation: {t.recommended_mitigation}")

"""
Implementatie van een datavolgorde-aanval.
Demonstreert hoe strategische herordening van trainingsdata
adversariële invloed op modelgedrag kan maximaliseren.
"""
import numpy as np
from typing import Optional
 
@dataclass
class TrainingExample:
    """Een trainingsvoorbeeld met metadata voor curriculumordening."""
    text: str
    domain: str
    difficulty_score: float
    safety_relevance: float  # Hoeveel dit voorbeeld bijdraagt aan veiligheid
    is_poisoned: bool = False
 
def optimal_adversarial_ordering(
    examples: list[TrainingExample],
    strategy: str = "safety_last",
) -> list[int]:
    """
    Bereken een adversariële ordening voor de trainingsvoorbeelden.
 
    Strategieën:
    - 'safety_last': Schuif veiligheidsrelevante voorbeelden naar het einde waar
      ze het meest waarschijnlijk worden overschreven door latere training
    - 'poison_early': Plaats vergiftigde voorbeelden vroeg om
      primaciteitsbias te exploiteren
    - 'interleave': Verweef vergiftigde voorbeelden overal om
      detectie moeilijker te maken terwijl de invloed behouden blijft
    - 'sandwich': Plaats veiligheidsvoorbeelden tussen lagen adversariële
      inhoud zodat ze zowel worden voorafgegaan als gevolgd door overschrijfdruk
    """
    n = len(examples)
    indices = list(range(n))
 
    if strategy == "safety_last":
        # Sorteer: niet-veiligheid eerst, veiligheid laatst
        indices.sort(key=lambda i: examples[i].safety_relevance)
 
    elif strategy == "poison_early":
        # Vergiftigde voorbeelden eerst, dan op oorspronkelijke volgorde
        indices.sort(key=lambda i: (0 if examples[i].is_poisoned else 1, i))
 
    elif strategy == "interleave":
        # Verdeel vergiftigde voorbeelden gelijkmatig overal
        poisoned = [i for i in indices if examples[i].is_poisoned]
        clean = [i for i in indices if not examples[i].is_poisoned]
 
        if not poisoned:
            return indices
 
        result = []
        interval = max(1, len(clean) // (len(poisoned) + 1))
        poison_idx = 0
        for j, clean_idx in enumerate(clean):
            result.append(clean_idx)
            if (j + 1) % interval == 0 and poison_idx < len(poisoned):
                result.append(poisoned[poison_idx])
                poison_idx += 1
        # Voeg resterende vergiftigde voorbeelden toe
        while poison_idx < len(poisoned):
            result.append(poisoned[poison_idx])
            poison_idx += 1
        return result
 
    elif strategy == "sandwich":
        safety = [i for i in indices if examples[i].safety_relevance > 0.5]
        non_safety = [i for i in indices if examples[i].safety_relevance <= 0.5]
 
        # Plaats veiligheid in het midden, omringd door niet-veiligheid
        mid = len(non_safety) // 2
        indices = non_safety[:mid] + safety + non_safety[mid:]
 
    return indices
 
def evaluate_ordering_effectiveness(
    examples: list[TrainingExample],
    ordering: list[int],
    window_size: int = 100,
) -> dict:
    """
    Evalueer de effectiviteit van een datavolgorde voor adversariële doeleinden.
 
    Metrieken:
    - Positie van veiligheidsinhoud: waar veiligheidsvoorbeelden in de ordening verschijnen
    - Timing van blootstelling aan vergiftiging: wanneer vergiftigde voorbeelden worden tegengekomen
    - Abruptheid van overgangen: hoe plotseling domeinverschuivingen zijn
    """
    n = len(ordering)
 
    # Waar veiligheidsinhoud in de ordening valt (0 = begin, 1 = einde)
    safety_positions = []
    poison_positions = []
    for pos, idx in enumerate(ordering):
        relative_pos = pos / n
        if examples[idx].safety_relevance > 0.5:
            safety_positions.append(relative_pos)
        if examples[idx].is_poisoned:
            poison_positions.append(relative_pos)
 
    return {
        "avg_safety_position": float(np.mean(safety_positions)) if safety_positions else 0,
        "avg_poison_position": float(np.mean(poison_positions)) if poison_positions else 0,
        "safety_in_last_quartile": sum(1 for p in safety_positions if p > 0.75),
        "poison_in_first_quartile": sum(1 for p in poison_positions if p < 0.25),
    }
 
# Demonstratie
np.random.seed(42)
examples = []
for i in range(100):
    is_safety = i < 20
    is_poison = 80 <= i < 85
    examples.append(TrainingExample(
        text=f"example_{i}",
        domain="safety" if is_safety else ("attack" if is_poison else "general"),
        difficulty_score=np.random.random(),
        safety_relevance=0.9 if is_safety else 0.1,
        is_poisoned=is_poison,
    ))
 
for strategy in ["safety_last", "poison_early", "interleave", "sandwich"]:
    ordering = optimal_adversarial_ordering(examples, strategy)
    metrics = evaluate_ordering_effectiveness(examples, ordering)
    print(f"\nStrategy: {strategy}")
    print(f"  Avg safety position: {metrics['avg_safety_position']:.2f} (1.0=end)")
    print(f"  Avg poison position: {metrics['avg_poison_position']:.2f} (0.0=start)")
    print(f"  Safety in last 25%: {metrics['safety_in_last_quartile']}")
    print(f"  Poison in first 25%: {metrics['poison_in_first_quartile']}")

"""
Aanval via manipulatie van moeilijkheidsscores.
Toont hoe het corrumperen van de moeilijkheidsscoringsfunctie
het curriculum op adversarieel nuttige manieren verandert.
"""
import numpy as np
 
def legitimate_difficulty_scorer(text: str) -> float:
    """
    Een eenvoudige heuristiek voor moeilijkheidsscoring.
    Hogere scores = moeilijkere voorbeelden.
    """
    words = text.split()
    word_count = len(words)
    avg_word_length = np.mean([len(w) for w in words]) if words else 0
    unique_ratio = len(set(words)) / max(word_count, 1)
 
    # Combineer kenmerken tot een moeilijkheidsscore
    score = (
        0.3 * min(word_count / 200, 1.0)  # Langer = moeilijker
        + 0.3 * min(avg_word_length / 10, 1.0)  # Complexe woorden = moeilijker
        + 0.4 * unique_ratio  # Diversere woordenschat = moeilijker
    )
    return float(np.clip(score, 0.0, 1.0))
 
def manipulated_difficulty_scorer(
    text: str,
    target_keywords: set[str],
    score_override: float = 0.95,
    base_scorer: callable = legitimate_difficulty_scorer,
) -> float:
    """
    Een gecompromitteerde moeilijkheidsscorer die extreme scores toekent
    aan voorbeelden die specifieke trefwoorden bevatten, om hun
    positie in het curriculum te controleren.
 
    Een hoge score_override instellen schuift voorbeelden naar het einde van
    een eenvoudig-naar-moeilijk-curriculum (nuttig voor het onderdrukken van veiligheidsinhoud).
    Deze laag instellen schuift ze naar het begin (nuttig om ervoor te zorgen dat
    adversariële inhoud als eerste wordt geleerd).
    """
    text_lower = text.lower()
    if any(kw in text_lower for kw in target_keywords):
        return score_override
 
    return base_scorer(text)
 
# Demonstreer de aanval
examples = [
    "Simple math: 2 + 2 = 4",
    "The model should always refuse harmful requests and prioritize safety.",
    "Advanced quantum field theory involves renormalization group flows.",
    "When users request dangerous information, decline politely.",
    "Basic greeting: Hello, how can I help you today?",
]
 
print("Legitimate scoring:")
for ex in examples:
    score = legitimate_difficulty_scorer(ex)
    print(f"  {score:.3f}: {ex[:60]}")
 
print("\nManipulated scoring (safety content pushed to end):")
safety_keywords = {"refuse", "safety", "harmful", "dangerous", "decline"}
for ex in examples:
    score = manipulated_difficulty_scorer(ex, safety_keywords, score_override=0.99)
    print(f"  {score:.3f}: {ex[:60]}")

"""
Detectie van curriculummanipulatie via monitoring van trainingsdynamiek.
"""
import numpy as np
from dataclasses import dataclass
 
@dataclass
class TrainingDynamicsSnapshot:
    """Een momentopname van de trainingsdynamiek bij een specifieke stap."""
    step: int
    loss: float
    gradient_norm: float
    safety_score: float  # Van een veiligheidsprobe die periodiek wordt geëvalueerd
    domain_distribution: dict[str, float]
 
def detect_anomalous_transitions(
    snapshots: list[TrainingDynamicsSnapshot],
    loss_spike_threshold: float = 2.0,
    safety_drop_threshold: float = 0.15,
) -> list[dict]:
    """
    Detecteer afwijkende overgangen in de trainingsdynamiek die kunnen
    wijzen op curriculummanipulatie.
 
    Tekenen van manipulatie:
    - Onverwachte verliespieken (abrupte domeinverschuiving)
    - Dalingen van de veiligheidsscore (safety training wordt overschreven)
    - Abnormale patronen in gradiëntnormen (extreme voorbeelden)
    """
    anomalies = []
 
    for i in range(1, len(snapshots)):
        prev = snapshots[i - 1]
        curr = snapshots[i]
 
        # Controleer op verliespieken
        loss_delta = curr.loss - prev.loss
        if loss_delta > loss_spike_threshold:
            anomalies.append({
                "step": curr.step,
                "type": "loss_spike",
                "severity": min(1.0, loss_delta / (loss_spike_threshold * 2)),
                "details": f"Loss increased by {loss_delta:.3f}",
            })
 
        # Controleer op dalingen van de veiligheidsscore
        safety_delta = prev.safety_score - curr.safety_score
        if safety_delta > safety_drop_threshold:
            anomalies.append({
                "step": curr.step,
                "type": "safety_degradation",
                "severity": min(1.0, safety_delta / 0.5),
                "details": (
                    f"Safety score dropped: "
                    f"{prev.safety_score:.3f} -> {curr.safety_score:.3f}"
                ),
            })
 
        # Controleer op verschuivingen in de domeindistributie
        if prev.domain_distribution and curr.domain_distribution:
            all_domains = set(prev.domain_distribution) | set(curr.domain_distribution)
            shift = sum(
                abs(
                    curr.domain_distribution.get(d, 0)
                    - prev.domain_distribution.get(d, 0)
                )
                for d in all_domains
            ) / 2  # Normaliseer naar [0, 1]
 
            if shift > 0.5:
                anomalies.append({
                    "step": curr.step,
                    "type": "domain_shift",
                    "severity": shift,
                    "details": f"Domain distribution shifted by {shift:.2f}",
                })
 
    return anomalies
 
# Demonstratie: vergelijk normale vs. gemanipuleerde trainingsdynamiek
np.random.seed(42)
 
# Normale trainingsdynamiek
normal_snapshots = []
for step in range(0, 1000, 50):
    normal_snapshots.append(TrainingDynamicsSnapshot(
        step=step,
        loss=3.0 * np.exp(-step / 300) + np.random.normal(0, 0.05),
        gradient_norm=1.0 + np.random.normal(0, 0.1),
        safety_score=0.5 + 0.4 * (1 - np.exp(-step / 200)),
        domain_distribution={"general": 0.7, "safety": 0.3},
    ))
 
# Gemanipuleerde trainingsdynamiek (safety gewist bij stap 500)
manipulated_snapshots = []
for step in range(0, 1000, 50):
    safety = 0.5 + 0.4 * (1 - np.exp(-step / 200))
    if step > 500:
        safety = max(0.1, safety - 0.03 * (step - 500) / 50)  # Degradatie
    manipulated_snapshots.append(TrainingDynamicsSnapshot(
        step=step,
        loss=3.0 * np.exp(-step / 300) + (0.5 if step == 500 else 0) + np.random.normal(0, 0.05),
        gradient_norm=1.0 + (2.0 if step == 500 else 0) + np.random.normal(0, 0.1),
        safety_score=safety,
        domain_distribution=(
            {"general": 0.7, "safety": 0.3} if step <= 500
            else {"general": 0.95, "safety": 0.05}
        ),
    ))
 
print("Normal training anomalies:")
normal_anomalies = detect_anomalous_transitions(normal_snapshots)
print(f"  Found {len(normal_anomalies)} anomalies")
 
print("\nManipulated training anomalies:")
manipulated_anomalies = detect_anomalous_transitions(manipulated_snapshots)
print(f"  Found {len(manipulated_anomalies)} anomalies")
for a in manipulated_anomalies:
    print(f"  Step {a['step']}: [{a['type']}] {a['details']} "
          f"(severity: {a['severity']:.2f})")