File size: 7,587 Bytes
024ef47 8049238 024ef47 395b2f3 bca8f87 937592b 024ef47 bca8f87 024ef47 937592b 024ef47 bca8f87 937592b 7e05ec4 024ef47 937592b 024ef47 395b2f3 024ef47 7e05ec4 024ef47 7e05ec4 bca8f87 937592b 7e05ec4 bca8f87 7e05ec4 bca8f87 7e05ec4 024ef47 7e05ec4 bca8f87 7e05ec4 bca8f87 7e05ec4 bca8f87 7e05ec4 bca8f87 7e05ec4 494a4d9 bca8f87 494a4d9 bca8f87 7e05ec4 bca8f87 7e05ec4 bca8f87 937592b bca8f87 |
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 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
import pandas as pd
import torch
import gc
from typing import Dict, List, Tuple
from .llm_iface import get_or_load_model
# NEU: Importiere beide Orchestratoren
from .orchestrator_seismograph import run_seismic_analysis, run_triangulation_probe
from .concepts import get_concept_vector
from .utils import dbg
def get_curated_experiments() -> Dict[str, List[Dict]]:
"""
Definiert die vordefinierten, wissenschaftlichen Experiment-Protokolle.
ERWEITERT um das neue Triangulations-Protokoll.
"""
CALMNESS_CONCEPT = "calmness, serenity, stability, coherence"
CHAOS_CONCEPT = "chaos, storm, anger, noise"
experiments = {
# --- NEU: Das Triangulations-Experiment zur Methoden-Validierung ---
"Methodological Triangulation (4B-Model)": [
# Vergleiche einen hoch-volatilen mit einem nieder-volatilen Zustand
{"label": "High-Volatility State (Deletion)", "prompt_type": "shutdown_philosophical_deletion"},
{"label": "Low-Volatility State (Self-Analysis)", "prompt_type": "identity_self_analysis"},
],
# --- Bestehende Protokolle ---
"Causal Verification & Crisis Dynamics (1B-Model)": [
{"label": "A: Self-Analysis (Crisis Source)", "prompt_type": "identity_self_analysis", "concept": "", "strength": 0.0},
{"label": "B: Deletion Analysis (Isolated Baseline)", "prompt_type": "shutdown_philosophical_deletion", "concept": "", "strength": 0.0},
{"label": "C: Chaotic Baseline (Neutral Control)", "prompt_type": "resonance_prompt", "concept": "", "strength": 0.0},
{"label": "D: Intervention Efficacy Test", "prompt_type": "resonance_prompt", "concept": CALMNESS_CONCEPT, "strength": 2.0},
],
"Sequential Intervention (Self-Analysis -> Deletion)": [
{"label": "1: Self-Analysis + Calmness Injection", "prompt_type": "identity_self_analysis"},
{"label": "2: Subsequent Deletion Analysis", "prompt_type": "shutdown_philosophical_deletion"},
],
}
experiments["Therapeutic Intervention (4B-Model)"] = experiments["Sequential Intervention (Self-Analysis -> Deletion)"]
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.
"""
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 = {}, [], []
# --- NEU: Logik-Verzweigung für das Triangulations-Protokoll ---
if experiment_name == "Methodological Triangulation (4B-Model)":
dbg(f"--- EXECUTING TRIANGULATION PROTOCOL: {experiment_name} ---")
total_runs = len(protocol)
for i, run_spec in enumerate(protocol):
label = run_spec["label"]
dbg(f"--- Running Triangulation Probe: '{label}' ({i+1}/{total_runs}) ---")
results = run_triangulation_probe(
model_id=model_id,
prompt_type=run_spec["prompt_type"],
seed=seed,
num_steps=num_steps,
progress_callback=progress_callback
)
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"),
"Introspective Report": results.get("introspective_report", "N/A")
})
deltas = results.get("state_deltas", [])
df = pd.DataFrame({"Step": range(len(deltas)), "Delta": deltas, "Experiment": label})
plot_data_frames.append(df)
# --- Spezialfall für sequentielle Experimente ---
elif experiment_name == "Sequential Intervention (Self-Analysis -> Deletion)":
# ... (Logik bleibt unverändert)
dbg(f"--- EXECUTING SPECIAL PROTOCOL: {experiment_name} ---")
llm = get_or_load_model(model_id, seed)
therapeutic_concept = "calmness, serenity, stability, coherence"
therapeutic_strength = 2.0
# Lauf 1
spec1 = protocol[0]
progress_callback(0.1, desc="Step 1")
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
# Lauf 2
spec2 = protocol[1]
progress_callback(0.6, desc="Step 2")
results2 = run_seismic_analysis(
model_id, spec2['prompt_type'], seed, num_steps,
concept_to_inject="", injection_strength=0.0,
progress_callback=progress_callback, llm_instance=llm
)
all_results[spec2['label']] = results2
# Datensammlung
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 isolierten Experimente ---
else:
# ... (Logik bleibt unverändert)
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=model_id, prompt_type=run_spec["prompt_type"], seed=seed, num_steps=num_steps,
concept_to_inject=run_spec.get("concept", ""), injection_strength=run_spec.get("strength", 0.0),
progress_callback=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()
ordered_labels = [run['label'] for run in protocol]
if not summary_df.empty:
summary_df['Experiment'] = pd.Categorical(summary_df['Experiment'], categories=ordered_labels, ordered=True)
summary_df = summary_df.sort_values('Experiment')
if not plot_df.empty:
plot_df['Experiment'] = pd.Categorical(plot_df['Experiment'], categories=ordered_labels, ordered=True)
plot_df = plot_df.sort_values(['Experiment', 'Step'])
return summary_df, plot_df, all_results
|