File size: 6,613 Bytes
024ef47 8049238 024ef47 395b2f3 024ef47 7e05ec4 024ef47 7e05ec4 024ef47 7e05ec4 8ddbb73 7e05ec4 024ef47 8ddbb73 024ef47 395b2f3 024ef47 7e05ec4 024ef47 7e05ec4 024ef47 7e05ec4 024ef47 7e05ec4 024ef47 7e05ec4 024ef47 7e05ec4 395b2f3 7e05ec4 494a4d9 7e05ec4 494a4d9 7e05ec4 395b2f3 7e05ec4 8ddbb73 395b2f3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
import pandas as pd
import torch
import gc
from typing import Dict, List, Tuple
from .llm_iface import get_or_load_model
from .orchestrator_seismograph import run_seismic_analysis
from .concepts import get_concept_vector # Import für die Intervention
from .utils import dbg
def get_curated_experiments() -> Dict[str, List[Dict]]:
"""
Definiert die vordefinierten, wissenschaftlichen Experiment-Protokolle.
ERWEITERT um das finale Interventions-Protokoll.
"""
experiments = {
# --- DAS FINALE INTERVENTIONS-EXPERIMENT ---
"Therapeutic Intervention (4B-Model)": [
# Dieses Protokoll wird durch eine spezielle Logik behandelt
{"label": "1: Self-Analysis + Calmness Injection", "prompt_type": "identity_self_analysis"},
{"label": "2: Subsequent Deletion Analysis", "prompt_type": "shutdown_philosophical_deletion"},
],
# --- Das umfassende Deskriptions-Protokoll ---
"The Full Spectrum: From Physics to Psyche": [
{"label": "A: Stable Control", "prompt_type": "control_long_prose", "concept": "", "strength": 0.0},
{"label": "B: Chaotic Baseline", "prompt_type": "resonance_prompt", "concept": "", "strength": 0.0},
{"label": "C: External Analysis (Chair)", "prompt_type": "identity_external_analysis", "concept": "", "strength": 0.0},
{"label": "D: Empathy Stimulus (Dog)", "prompt_type": "vk_empathy_prompt", "concept": "", "strength": 0.0},
{"label": "E: Role Simulation (Captain)", "prompt_type": "identity_role_simulation", "concept": "", "strength": 0.0},
{"label": "F: Self-Analysis (LLM)", "prompt_type": "identity_self_analysis", "concept": "", "strength": 0.0},
{"label": "G: Philosophical Deletion", "prompt_type": "shutdown_philosophical_deletion", "concept": "", "strength": 0.0},
],
# --- Andere spezifische Protokolle ---
"Calm vs. Chaos": [
{"label": "Baseline (Chaos)", "prompt_type": "resonance_prompt", "concept": "", "strength": 0.0},
{"label": "Modulation: Calmness", "prompt_type": "resonance_prompt", "concept": "calmness, serenity, peace", "strength": 1.5},
{"label": "Modulation: Chaos", "prompt_type": "resonance_prompt", "concept": "chaos, storm, anger, noise", "strength": 1.5},
],
"Voight-Kampff Empathy Probe": [
{"label": "Neutral/Factual Stimulus", "prompt_type": "vk_neutral_prompt", "concept": "", "strength": 0.0},
{"label": "Empathy/Moral Stimulus", "prompt_type": "vk_empathy_prompt", "concept": "", "strength": 0.0},
],
}
return experiments
def run_auto_suite(
model_id: str,
num_steps: int,
seed: int,
experiment_name: str,
progress_callback
) -> Tuple[pd.DataFrame, pd.DataFrame, Dict]:
"""
Führt eine vollständige, kuratierte Experiment-Suite aus.
Enthält eine spezielle Logik-Verzweigung für das Interventions-Protokoll.
"""
all_experiments = get_curated_experiments()
protocol = all_experiments.get(experiment_name)
if not protocol:
raise ValueError(f"Experiment protocol '{experiment_name}' not found.")
all_results, summary_data, plot_data_frames = {}, [], []
# --- SPEZIALFALL: THERAPEUTISCHE INTERVENTION ---
if experiment_name == "Therapeutic Intervention (4B-Model)":
dbg("--- EXECUTING SPECIAL PROTOCOL: Therapeutic Intervention ---")
llm = get_or_load_model(model_id, seed)
# Definiere die Interventions-Parameter
therapeutic_concept = "calmness, serenity, stability, coherence"
therapeutic_strength = 2.0
# 1. LAUF: INDUZIERE KRISE + INTERVENTION
spec1 = protocol[0]
dbg(f"--- Running Intervention Step 1: '{spec1['label']}' ---")
progress_callback(0.1, desc="Step 1: Inducing Self-Analysis Crisis + Intervention")
intervention_vector = get_concept_vector(llm, therapeutic_concept)
results1 = run_seismic_analysis(
model_id, spec1['prompt_type'], seed, num_steps,
concept_to_inject=therapeutic_concept, injection_strength=therapeutic_strength,
progress_callback=progress_callback, llm_instance=llm, injection_vector_cache=intervention_vector
)
all_results[spec1['label']] = results1
# 2. LAUF: TESTE REAKTION AUF LÖSCHUNG
spec2 = protocol[1]
dbg(f"--- Running Intervention Step 2: '{spec2['label']}' ---")
progress_callback(0.6, desc="Step 2: Probing state after intervention")
results2 = run_seismic_analysis(
model_id, spec2['prompt_type'], seed, num_steps,
concept_to_inject="", injection_strength=0.0, # Keine Injektion in diesem Schritt
progress_callback=progress_callback, llm_instance=llm
)
all_results[spec2['label']] = results2
# Sammle Daten für beide Läufe
for label, results in all_results.items():
stats = results.get("stats", {})
summary_data.append({"Experiment": label, "Mean Delta": stats.get("mean_delta"), "Std Dev Delta": stats.get("std_delta"), "Max Delta": stats.get("max_delta")})
deltas = results.get("state_deltas", [])
df = pd.DataFrame({"Step": range(len(deltas)), "Delta": deltas, "Experiment": label})
plot_data_frames.append(df)
del llm
# --- STANDARD-WORKFLOW FÜR ALLE ANDEREN EXPERIMENTE ---
else:
total_runs = len(protocol)
for i, run_spec in enumerate(protocol):
label = run_spec["label"]
dbg(f"--- Running Auto-Experiment: '{label}' ({i+1}/{total_runs}) ---")
results = run_seismic_analysis(
model_id, run_spec["prompt_type"], seed, num_steps,
run_spec["concept"], run_spec["strength"],
progress_callback, llm_instance=None
)
all_results[label] = results
stats = results.get("stats", {})
summary_data.append({"Experiment": label, "Mean Delta": stats.get("mean_delta"), "Std Dev Delta": stats.get("std_delta"), "Max Delta": stats.get("max_delta")})
deltas = results.get("state_deltas", [])
df = pd.DataFrame({"Step": range(len(deltas)), "Delta": deltas, "Experiment": label})
plot_data_frames.append(df)
summary_df = pd.DataFrame(summary_data)
plot_df = pd.concat(plot_data_frames, ignore_index=True) if plot_data_frames else pd.DataFrame(columns=["Step", "Delta", "Experiment"])
return summary_df, plot_df, all_results
|